[python][daft] Fix primary-key distributed write correctness

[kerwin-zk]

Purpose

Daft previously let independent sink tasks write the same primary-key bucket. For fixed-bucket tables this could produce overlapping sequence numbers, while dynamic-bucket tables used task-local bucket assignments that could place the same key in different buckets. Both cases could silently return stale or duplicate primary keys.

This change:

• writes each fixed (partition, bucket) group through one Paimon writer;
• coordinates dynamic bucket assignment through persistent, Java-compatible HASH indexes and restores mappings across commits;
• pins one base snapshot per Daft job and adds optimistic conflict checks for concurrent index changes and overwrite remapping;
• makes regular PyPaimon dynamic-bucket writes restore the same persistent mapping;
• fails fast for primary-key write modes that do not yet have safe global ownership, and preserves files when commit success is uncertain;
• adds Java/Python hash goldens and bidirectional HASH index interoperability coverage.

The group-based Daft path intentionally favors correctness: all rows for one (partition, bucket) are materialized by one worker. Fixed-bucket tables with very few or highly skewed buckets can therefore have a single-worker memory/CPU hotspot; dynamic buckets are bounded by their target row count.

Tests

CI

[Copilot]

Pull request overview

Improves correctness for distributed primary-key writes from Daft into Paimon by enforcing single-writer ownership per (partition, bucket) and by making HASH_DYNAMIC bucket assignment persistent and Java-interoperable via maintained HASH indexes with conflict detection.

Changes:

• Add coordinated Daft write paths for primary-key tables: group materialization and upstream bucket/key-hash routing to ensure a single writer owns each (partition, bucket) in a commit.
• Implement persistent HASH_DYNAMIC bucket mapping in PyPaimon via maintained HASH index files, plus optimistic conflict checks (including pinned base snapshot semantics).
• Add Python/Java interoperability + concurrency correctness tests and update mixed test runner wiring.

Reviewed changes

Copilot reviewed 16 out of 16 changed files in this pull request and generated 1 comment.

Show a summary per file

File	Description
paimon-python/pypaimon/write/table_write.py	Adds dynamic-bucket index maintenance hooks, precomputed-bucket group writes, and index lifecycle handling during commit/abort/close.
paimon-python/pypaimon/write/table_commit.py	Aborts prepared files on deterministic commit conflicts (CommitConflictError).
paimon-python/pypaimon/write/row_key_extractor.py	Refactors hashing, adds dynamic-bucket persistent assigner/index integration and precomputed-hash flows.
paimon-python/pypaimon/write/file_store_commit.py	Adds HASH index conflict detection triggers and base-snapshot pinning checks; enhances overwrite to delete HASH indexes for replaced partitions.
paimon-python/pypaimon/write/commit/conflict_detection.py	Introduces CommitConflictError and HASH index replacement conflict checks.
paimon-python/pypaimon/write/commit_message.py	Adds hash_index_base_snapshot to propagate pinned snapshot semantics through commits.
paimon-python/pypaimon/tests/write/dynamic_bucket_test.py	New unit tests covering hash goldens, assigner behavior, index lifecycle, concurrency conflicts, and failure modes.
paimon-python/pypaimon/tests/e2e/java_py_read_write_test.py	Adds E2E coverage for reading Java-written dynamic-bucket HASH indexes and writing Python dynamic-bucket HASH indexes.
paimon-python/pypaimon/tests/daft/daft_pk_distributed_write_test.py	New correctness tests for distributed Daft PK writes (fixed/dynamic/postpone modes, conflicts, and task ownership).
paimon-python/pypaimon/table/file_store_table.py	Updates row-key extractor creation to enable persistent dynamic-bucket behavior and fail fast for CROSS_PARTITION PK writes.
paimon-python/pypaimon/manifest/index_manifest_file.py	Enforces one-HASH-index-file-per-(partition,bucket) invariant in manifests.
paimon-python/pypaimon/index/dynamic_bucket.py	New module implementing persistent hash assignment, HASH index IO, and index maintenance for dynamic buckets.
paimon-python/pypaimon/daft/daft_paimon.py	Reworks Daft write flow for PK tables to coordinate routing, bucket assignment, and group-owned writers.
paimon-python/pypaimon/daft/daft_datasink.py	Adds worker-local table metadata cache, PK safety checks for direct sink writes, and group-write/commit sinks (including commit-message transport).
paimon-python/dev/run_mixed_tests.sh	Extends mixed Java/Python E2E test runner to include the new dynamic-bucket HASH-index tests.
paimon-core/src/test/java/org/apache/paimon/JavaPyE2ETest.java	Adds Java E2E helpers/tests for dynamic-bucket HASH index read/write interop with Python.

Comments suppressed due to low confidence (1)

paimon-python/pypaimon/daft/daft_datasink.py:764

• The group-write UDF currently emits a raw pickle blob. If you add framing/magic-header checks before unpickling in PaimonCommitDataSink, this producer needs to prefix the same header when serializing the commit messages.

        return Series.from_arrow(
            pa.array([pickle.dumps(commit_messages)], type=pa.binary())
        )

---

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

[YannByron]

@kerwin-zk Thanks for this. This closes a real correctness gap in the Daft write path — nice fix.