Root cause:

includeExistingFiles is only evaluated the first time the stream is started with a fresh checkpoint. If the stream is restarted or the checkpoint folder is reused, changing this option will have no effect on subsequent runs—old files previously seen or untracked can be reprocessed

To reliably skip existing files, you must use a new (clean) checkpoint location when starting the stream. Otherwise, Auto Loader will refer to the checkpoint's metadata of already-processed files and the original setting at the stream's initialization.

Solution:

1. Identify Your Container Path and Set a Clean Checkpoint Location
Determine the correct storage path (e.g., for ADLS Gen2: abfss://<container>@<account>.dfs.core.windows.net/<path>).

Choose a brand new checkpoint location that Auto Loader has never used. Example: abfss://<container>@<account>.dfs.core.windows.net/<path>/checkpoints/autoloader_run1.

Old checkpoint locations must not be reused; they store processed file metadata.

2. Set the Key Options in the Stream Reader
Use .option("cloudFiles.includeExistingFiles", "false") to exclude files present before the stream starts.

(Optional) Add .option("cloudFiles.modifiedAfter", "<timestamp>") if you know new files will have a modified time after the specified value.

Specify the correct cloudFiles.format (e.g., "text" for txt files).

3. Start the Stream (Python Example)
Replace <LANDED_PATH> and <CHECKPOINT_PATH> with your paths.

python
df_stream = spark.readStream \
.format("cloudFiles") \
.option("cloudFiles.format", "text") \
.option("cloudFiles.includeExistingFiles", "false") \
.option("cloudFiles.modifiedAfter", "2025-09-09T00:00:00.000Z") \ # Optional for fine control
.load(LANDED_PATH)

query = df_stream.writeStream \
.format("delta") \
.option("checkpointLocation", CHECKPOINT_PATH) \
.start(OUTPUT_PATH)
Do not reuse any checkpoint directories from previous runs.

4. Validate File Processing
Upload a test file to your source folder. Only files that arrive after the streaming job starts will be ingested.

Old files will be skipped unless their lastModified attribute is after your specified timestamp in cloudFiles.modifiedAfter.

5. Forcing Exclusion of Old Files
If any old file does get processed, re-check your checkpoint is new and the file's lastModified attribute is correct.

If re-running or reconfiguring, always delete or provide a new checkpoint directory.

Let me know if it works