[PDGGK]

Purpose

Linked issue: close #1318

On a fixed-bucket append table ('bucket' = 'N', 'bucket-key' = '...'), UPDATE or a matched MERGE INTO update that changes a bucket-key column leaves the row in the bucket of its old value. Scans prune buckets by the bucket-key hash, so any query filtering on the new value skips the row, while a full scan still returns it.

Before this change, on main:

>>> UPDATE db.orders SET customer = 'erin' WHERE id = 1
>>> SELECT * FROM db.orders ORDER BY id
+----+----------+--------+
| id | customer | amount |
+----+----------+--------+
| 1  | erin     | 10     |
| 2  | bob      | 20     |
| 3  | carol    | 30     |
| 4  | dave     | 40     |
+----+----------+--------+
>>> SELECT * FROM db.orders WHERE customer = 'erin'
++
++

After:

>>> SELECT * FROM db.orders WHERE customer = 'erin'
+----+----------+--------+
| id | customer | amount |
+----+----------+--------+
| 1  | erin     | 10     |
+----+----------+--------+

The table is CREATE TABLE db.orders (id INT, customer STRING, amount INT) WITH ('bucket' = '4', 'bucket-key' = 'customer') with rows (1, 'alice', 10), (2, 'bob', 20), (3, 'carol', 30), (4, 'dave', 40). Before the fix, orders$files shows the rewritten file still in bucket-2 (where alice hashes), while a new INSERT of 'erin' goes to bucket-1.

Brief change log

• CopyOnWriteMergeWriter::prepare_commit: when the updated columns include a bucket-key column and the table has a fixed bucket count above 1, compute the bucket of each rewritten row with the table's bucket function and bucket count, as TableWrite does for inserts, and write each bucket's rows into that bucket. The deleted file stays in the commit message of its original bucket; rows that moved are added in the message of their new bucket.
• Updates that do not touch a bucket key, and tables without a fixed bucket count, keep the existing path unchanged.

This matches Java: Spark's UPDATE on an append table writes the updated rows through PaimonSparkWriter, which assigns HASH_FIXED buckets from the bucket keys of each row.

Assumptions worth checking:

• The new buckets use the table's current bucket option. TableWrite uses it too, and the commit records it as the files' total bucket count. A partition still holding files with an older bucket count (rescaled without overwrite) gets the rewritten rows hashed with the current count.
• When a bucket key is updated, every surviving row of a touched file is hashed again. Rows whose key did not change hash to the bucket they were already in.

Tests

• update_tests::test_update_bucket_key_moves_row_to_new_bucket: UPDATE of the bucket key, then a filter on the new value returns the row.
• append_merge_into::test_update_bucket_key_moves_row_to_new_bucket: the same through MERGE INTO, with one updated row moving out of a file that also keeps an unchanged row.
• Without the change in cow_writer.rs, both tests fail with the filtered query returning [] instead of the updated rows.
• RUST_MIN_STACK=16777216 cargo test -p paimon-datafusion --test update_tests --test append_merge_into --test delete_tests --test merge_into_tests (21, 35, 25 and 29 tests) pass.
• cargo fmt --all -- --check and cargo clippy --locked -p paimon -p paimon-datafusion --all-targets -- -D warnings pass.
• I also checked by hand a partitioned bucketed table, an update to NULL, 'bucket-function.type' = 'mod', and a MERGE mixing matched DELETE, matched UPDATE and NOT MATCHED INSERT.

The full paimon unit-test suite and the vortex feature are left to CI.

API and Format

No API or storage format change.

Documentation

No.

[JeonDaehong]

Checked a two-column bucket key with one of them updated, and the moved row lands in the same bucket as an INSERT of the new value. LGTM.