handle delta arrow load id partition column merge disposition

dlt/common/libs/deltalake.py

@@ -131,9 +131,10 @@ def merge_delta_table(
             primary_keys = get_columns_names_with_prop(schema, "primary_key")
             predicate = " AND ".join([f"target.{c} = source.{c}" for c in primary_keys])
 
+        partition_by = get_columns_names_with_prop(schema, "partition")
         qry = (
             table.merge(
-                source=ensure_delta_compatible_arrow_data(data),
+                source=ensure_delta_compatible_arrow_data(data, partition_by),
                 predicate=predicate,
                 source_alias="source",
                 target_alias="target",

tests/load/pipeline/test_filesystem_pipeline.py

@@ -610,6 +610,7 @@ def test_delta_table_partitioning_arrow_load_id(
 
     pipeline = destination_config.setup_pipeline("fs_pipe", dev_mode=True)
 
+    # append write disposition
     info = pipeline.run(
         pyarrow.table({"foo": [1]}),
         table_name="delta_table",
@@ -621,6 +622,20 @@ def test_delta_table_partitioning_arrow_load_id(
     assert dt.metadata().partition_columns == ["_dlt_load_id"]
     assert load_table_counts(pipeline, "delta_table")["delta_table"] == 1
 
+    # merge write disposition
+    info = pipeline.run(
+        pyarrow.table({"foo": [1, 2]}),
+        table_name="delta_table",
+        write_disposition={"disposition": "merge", "strategy": "upsert"},
+        columns={"_dlt_load_id": {"partition": True}},
+        primary_key="foo",
+        table_format="delta",
+    )
+    assert_load_info(info)
+    dt = get_delta_tables(pipeline, "delta_table")["delta_table"]
+    assert dt.metadata().partition_columns == ["_dlt_load_id"]
+    assert load_table_counts(pipeline, "delta_table")["delta_table"] == 2
+
 
 @pytest.mark.essential
 @pytest.mark.parametrize(