JaydenHD opened a new issue, #474: URL: https://github.com/apache/paimon-rust/issues/474
### Summary `MERGE INTO` on append-only partitioned tables can incorrectly prune the target table using partition values read from the source table, even when the `ON` condition does not equate the target partition column with the source partition column. This can make existing target rows invisible to the MERGE join and cause data correctness issues such as missed updates/deletes or duplicate inserts. ### Impact High data correctness risk for the CoW MERGE path on append-only partitioned tables. This is especially risky when: - the target table is partitioned, for example by `pt` - the source table also has a column with the same name, `pt` - the `ON` condition matches rows by another key, for example `t.a = s.a` - MERGE has a `WHEN NOT MATCHED THEN INSERT` branch In that case a row that should be matched and updated can be treated as not matched and inserted as a duplicate. ### Source evidence In the CoW MERGE path, the target scan is limited before evaluating the actual MERGE `ON` condition: - `execute_cow_merge_once` builds a partition set from the source table and passes it into the CoW writer: https://github.com/apache/paimon-rust/blob/94a556530e6a9fd30ce3ea7ea0ca51fec0f22fc4/crates/integrations/datafusion/src/merge_into.rs#L332-L336 - `build_source_partition_set` selects distinct values of the target partition key names from the source table: https://github.com/apache/paimon-rust/blob/94a556530e6a9fd30ce3ea7ea0ca51fec0f22fc4/crates/integrations/datafusion/src/merge_into.rs#L1365-L1393 For a target partition key `pt`, this becomes equivalent to: ```sql SELECT DISTINCT s."pt" FROM source AS s ``` There is no check that the MERGE condition contains a safe partition equivalence such as `t.pt = s.pt`. The resulting source-derived partition set is then used to filter the target table scan: - `CopyOnWriteMergeWriter::new` converts the set into a `PartitionFilter`: https://github.com/apache/paimon-rust/blob/94a556530e6a9fd30ce3ea7ea0ca51fec0f22fc4/crates/paimon/src/table/cow_writer.rs#L127-L137 - only files from that filtered scan are added to the CoW file index: https://github.com/apache/paimon-rust/blob/94a556530e6a9fd30ce3ea7ea0ca51fec0f22fc4/crates/paimon/src/table/cow_writer.rs#L140-L156 The subsequent matched and not-matched logic uses only the temporary target table created from this filtered file index: - matched UPDATE/DELETE joins against `cow_target_name`: https://github.com/apache/paimon-rust/blob/94a556530e6a9fd30ce3ea7ea0ca51fec0f22fc4/crates/integrations/datafusion/src/merge_into.rs#L487-L492 - not-matched INSERT uses a left join against the same filtered `cow_target_name`: https://github.com/apache/paimon-rust/blob/94a556530e6a9fd30ce3ea7ea0ca51fec0f22fc4/crates/integrations/datafusion/src/merge_into.rs#L575-L585 Therefore, if the real matching target row is in a partition not present in `source.pt`, the target row is excluded before the MERGE join is evaluated. ### End-to-end reproduction ```sql CREATE TABLE paimon.test_db.target (a INT, b INT, pt INT) PARTITIONED BY (pt); INSERT INTO paimon.test_db.target VALUES (1, 10, 2); CREATE TEMPORARY TABLE paimon.test_db.source (a INT, b INT, pt INT) AS SELECT * FROM (VALUES (1, 100, 1)) AS t(a, b, pt); MERGE INTO paimon.test_db.target t USING paimon.test_db.source s ON t.a = s.a WHEN MATCHED THEN UPDATE SET b = s.b WHEN NOT MATCHED THEN INSERT (a, b, pt) VALUES (s.a, s.b, s.pt); SELECT a, b, pt FROM paimon.test_db.target ORDER BY pt, a; ``` ### Expected result The `ON t.a = s.a` condition matches the existing target row `(1, 10, 2)`, so MERGE should update it: ```text (1, 100, 2) ``` ### Actual / risk from the current implementation The CoW path first prunes target partitions using `source.pt = 1`, so the existing target row in `pt = 2` is not present in `cow_target_name`. The source row is then classified as not matched and inserted: ```text (1, 100, 1) (1, 10, 2) ``` This is a silent data correctness issue: MERGE reports success but produces the wrong table contents. ### Suggested fix direction The source-derived target partition pruning should only be applied when it is proven safe from the MERGE `ON` condition, for example when every target partition key is constrained by an equality with the corresponding source expression/column. Otherwise, the CoW MERGE path should fall back to scanning all target partitions. ### Suggested regression test Add a test to `crates/integrations/datafusion/tests/append_merge_into.rs`, reusing the existing partitioned-table helpers, for example: ```rust #[tokio::test] async fn test_partitioned_merge_without_partition_equality_does_not_prune_by_source_partition() { let (_tmp, sql_context) = setup("CREATE TABLE paimon.test_db.target (a INT, b INT, pt INT) PARTITIONED BY (pt)").await; exec(&sql_context, "INSERT INTO paimon.test_db.target VALUES (1, 10, 2)").await; exec( &sql_context, "CREATE TEMPORARY TABLE paimon.test_db.source (a INT, b INT, pt INT) AS SELECT * FROM (VALUES (1, 100, 1)) AS t(a, b, pt)", ) .await; exec( &sql_context, "MERGE INTO paimon.test_db.target t \ USING paimon.test_db.source s ON t.a = s.a \ WHEN MATCHED THEN UPDATE SET b = s.b \ WHEN NOT MATCHED THEN INSERT (a, b, pt) VALUES (s.a, s.b, s.pt)", ) .await; assert_eq!(query_a_b_pt(&sql_context).await, vec![(1, 100, 2)]); } ``` ### Verification note I attempted to run a narrow local test command: ```text cargo test -p paimon-datafusion --test append_merge_into test_partitioned_update_and_insert -- --nocapture ``` but it did not complete within 4 minutes in my local environment. The issue above is based on the traced source path and should be reproducible with the SQL scenario described above. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
