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]

Reply via email to