Merge statement creates duplicates - Low shuffle merge as cause?

In our notebook pipeline we retrieve data from our bronze layer and upsert them to the silver layer through a MERGE statement. Every week the merge statement results in around 10 duplicates spread over different tables. See the merge statement below:

def merge_data(table_name, bronze_filtered, target_table, id_col='Id'):

    bronze_count = bronze_filtered.count()
    # Upsert new records to the silver table based on the id of the record if there are any records
    if bronze_count > 0:
        print(f"🔄️ Processing {bronze_count} records for {table_name}.")
        merge_condition = f"""
        target.{id_col} = source.{id_col}
        """

        target_table.alias("target").merge(
            bronze_filtered.alias("source"),
            merge_condition
        ).whenMatchedUpdateAll() \
         .whenNotMatchedInsertAll() \
         .execute()
        print(f"✅ Upsert for {table_name} complete.")
    else:
        print(f"❌ No records to upsert for {table_name}.")

Id's are always unique in the bronze layer and runs are scheduled far enough apart to exclude concurrency as the root cause. After discussing with Claude about the number of source row and the number of target rows that were inserted in duplicate runs it suggested to disable the Low Shuffle Merge Optimization (see below). To me this seems to be a too low level setting to play with for this issue. 

Are there any other people having similar issues, or that have tried similar solutions? Thanks in advance!  

===

The critical clue is numTargetRowsInserted: 5 with numSourceRows: 1. This should be impossible with standard merge semantics. This looks like it could be a bug specific to Microsoft Fabric's custom Low Shuffle Merge optimization. From the docs you found earlier:

The Microsoft Spark Delta team implemented a custom Low Shuffle Merge optimization where unmodified rows are excluded from an expensive shuffling operation. It is controlled by spark.microsoft.delta.merge.lowShuffle.enabled, enabled by default.

The numTargetRowsCopied: 0 is suspicious here — in normal Low Shuffle Merge, unmodified rows that need to be rewritten to new files are counted as "copied". Getting 0 copied but 5 inserted from 1 source row suggests the Low Shuffle implementation may be misclassifying existing target rows as unmatched inserts under certain conditions.

I'd try disabling it for your pipeline and re-running to see if the behaviour changes:

spark.conf.set("spark.microsoft.delta.merge.lowShuffle.enabled", "false")