So , I am using autoloader in a dlt pipeline for my data ingestion. I am using @Dlt.append_flow because I have data to load from multiple sources.

When I load a new file say x it has 3 rows my target gets 3 rows. But next even if I don't load any file and rerun the dlt pipeline 3 more rows are getting added. 
I have used the cloud files config of - includeexistingfiles = False. It is working only when I full refresh my pipelines. For examples now I have 1 file already in my directory and when I full refresh my dlt pipeline the target table get completely empty. bcz full refresh truncates the data and includeexisting files = False is not allowing me to upload new data. ( This is my logic for understand now sure whether correct or not)

But when I upload a new file say y with 4 rows. and run the dlt pipeline 4 new rows get added to my target table with total rows 4. But again when I rerun my pipeline this it again write the files y with 4 news rows in the target table so total 8 rows in my target table with duplicate records. I don't understand why this is happening.

I am using directory listing mode - my data is in my databricks volume which is also an s3 bucket path so I think it will work normally.

my file naming convention is like this - data file path/ xyz_15122024.csv 

data file path/ xyz_16122024.csv

this is my code 

# Load parameters
import dlt
target_catalog = spark.conf.get("catalog").lower()
target_schema = spark.conf.get("target_schema").lower()
target_table_name = spark.conf.get("target_table_name").lower()
data_file_path = spark.conf.get("data_file_path")
header = spark.conf.get("header")
file_format = spark.conf.get("file_format")
user = spark.conf.get("user")
evolve_schema = spark.conf.get("evolve_schema")
mode = spark.conf.get("mode")
import sys
sys.path.append(f"/Workspace/Users/{user}/.bundle/wtd_ingestion_framework/dev/files/src/")

# Import necessary modules
from utils.logging_handling import get_logger, log_event, log_error
from utils.dlt_utils import create_dlt_table
from pyspark.sql import SparkSession

def apply_model_extensions(input_df,target_catalog,target_schema,table_name):
    """
    Apply model extensions to the specified table by dynamically fetching the corresponding extensions 
    from the Delta table created by the load_model_extensions function.

    Parameters:
    table_name: str - The name of the table for which extensions are being applied.
    
    Returns:
    DataFrame: The updated DataFrame with applied model extensions.
    """
    try:
        # Log start of operation
        log_event(logger, f"Loading model extensions for table '{table_name}' from the Delta table.", "INFO")

        # Construct the model extensions table path dynamically
        model_extensions_table = f"{target_catalog}.metadata.model_extension_bb123_loans"

        # Read the model extensions for the specific table
        extension_df = spark.table(model_extensions_table)

        # Validate if there are extensions available for the table
        if extension_df.count() == 0:
            log_event(logger, f"No model extensions found for table '{table_name}'. Skipping.", "INFO")
            # return spark.readStream.table(table_name)
            return input_df

        # Load the target table
        # df = spark.readStream.table(f"live.{target_catalog}.{target_schema}.{table_name}")

        # Apply each extension to the target table
        for row in extension_df.collect():
            column_name = row['column_name']
            data_type = row['data_type']

            # Check if the column already exists to prevent overwriting
            if column_name not in input_df.columns:
                # Determine the default value based on the data type
                if data_type == "timestamp" and "UTC" in column_name:
                    default_value = from_utc_timestamp(current_timestamp(), "UTC")
                elif data_type == "timestamp":
                    default_value = current_timestamp()
                else:
                    default_value = expr("null")

                # Add the column with the default value
                input_df = input_df.withColumn(column_name, default_value.cast(data_type))
                log_event(logger, f"Added column '{column_name}' of type '{data_type}' to table '{table_name}'.", "INFO")
            else:
                log_event(logger, f"Column '{column_name}' already exists in table '{table_name}'. Skipping.", "INFO")

        log_event(logger, f"Successfully applied {extension_df.count()} model extensions to table '{table_name}'.", "INFO")
        return input_df

    except Exception as e:
        log_error(logger, f"Error applying model extensions to table '{table_name}': {str(e)}")
        raise

# Initialize logger and Spark session
logger = get_logger()


log_event(logger, "Starting file ingestion process.", "INFO")
table_path = f"{target_catalog}.{target_schema}.{target_table_name}"

filtered_elements = elements.filter(elements["Source_feed_name"] == "BB1123_loans").orderBy("Position")

    # Define schema using elements metadata
schema = StructType([
        StructField(
            row["Entity_Technical_Name"],  # Adjusted to match column key
            map_data_type(row["Datatype"], row["Length"] if "Length" in row else None),
            row["Nullable"] == "yes"
        )
        for row in filtered_elements.filter(elements["Source_feed_name"] == "BB1123_loans").collect()
])
log_event(logger, f"Schema: {schema}", "INFO")

header="true"
evolve_schema=True
schema_hints = None

if schema and evolve_schema:
    schema_hints = ",".join([f"{field.name} {field.dataType.simpleString()}" for field in schema.fields])
  
log_event(logger, f"Schema hints: {schema_hints}", "INFO")

# Determine schema evolution mode based on header and evolve_schema

if header == "true":
    infer_column_types = "true"  # Infer schema from file
    schema_evolution_mode = "addNewColumns" if evolve_schema else "rescue"
else:
    infer_column_types = "false"  # Do not infer schema
    schema_evolution_mode = "addNewColumns" if evolve_schema else "rescue"

# Define the cloud file configuration
cloudfile = {
    "cloudFiles.format": file_format,
    "cloudFiles.allowOverwrites": "false",
    "cloudFiles.inferColumnTypes": infer_column_types,
    "cloudFiles.schemaLocation": f"{data_file_path}/_schema/{table_name}/",
    "cloudFiles.schemaEvolutionMode": schema_evolution_mode,
    "cloudFiles.schemaHints": schema_hints,
    "cloudFiles.useNotifications": "false",
    "cloudFiles.includeExistingFiles": "false"
}
file_specific_config = {
    "header": header,  # Pass the header option directly
    "delimiter": "|",  # Set pipe as the delimiter
    "quote": "\"",     # Handle quoted fields
    "escape": "\\"     # Use backslash as escape character
}

table_type = "Streaming"
@dlt.append_flow(
    target = target_table_name,
    name = f"{target_table_name}_{file_format}_ingestion",
    # name = f"{target_table_name}",
    comment = f"{mode} Mode {table_type} DLT table created/updated from {data_file_path}."
)
def source_to_bronze():
    input_df = (spark
          .readStream
          .format("cloudfiles")
          .options(**cloudfile)
          .options(**file_specific_config)
          .option("checkpointLocation", f"{data_file_path}/_checkpoints/{target_table_name}")
          .load(data_file_path))
    
    # return df.writeStream.trigger(availableNow=True if mode == "batch" else False)
    input_df = apply_model_extensions(input_df,target_catalog,target_schema,target_table_name)
    log_event(logger, f"Inferred columns: {input_df.columns}", "INFO")

    # Write to Delta
    input_df.writeStream \
        .format("delta") \
        .option("mergeSchema", "true") \
        .outputMode("append") \
        .option("checkpointLocation", f"{data_file_path}/_checkpoints/{target_table_name}") \
        .start(table_path)

    return input_df


log_event(logger, f"DLT {table_type} table {table_name} created or updated from input path.", "INFO")

Most probably I am doing something wrong with checkpointing / file naming