Hi @ManojkMohan  ,

Thanks for your reply.

The checkpoint location is set to a new path. I tried multiple approaches and cleared the checkpoint location each time so it's considered as the first run. I followed all the steps you've mentioned above. Still processing all old files.

The only difference I see is that I'm using trigger while doing the write stream.

df_stream.writeStream \
.format("delta") \
.trigger(availableNow=True) \
.option("checkpointLocation", CHECKPOINT_PATH) \
.start(OUTPUT_PATH)

I did try multiple date formats supported in Azure Databricks. I also tried using 
.option('cloudFiles.includeExistingFiles', False). It still didn't work.