feat(write): add fixed-bucket write primitives for postpone tables

[XiaoHongbo-Hope]

Purpose

Add explicit low-level Rust/C primitives for writing fixed buckets in postpone-bucket primary-key tables.

Changes

• add an immutable bucket plan and use an internal pure router in the dedicated writer
• add a dedicated writer which requires a resolved plan
• expose explicit C builder and shared-plan APIs
• carry overwrite mode into the writer and committer
• validate overlapping (partition, bucket) ownership at commit time
• persist and validate total_buckets

Scope

This PR does not change normal WriteBuilder, DataFusion, or Fusion writes; they continue to use bucket = -2.

Callers must provide one shared plan and route each (partition, bucket) to a single writer. Commit-time ownership validation is the final backstop.

Bucket planning, global statistics, preclustering, and distributed shuffle belong to the calling integration and are outside this PR. Java Spark staged-file planning and static or partial-partition overwrite are also out of scope.

Deletion-vector tables are not supported by this fixed-bucket path and are rejected before writing.

Verification

• cargo fmt --all -- --check
• cargo clippy --locked -p paimon-c --all-targets -- -D warnings
• cargo test --locked -p paimon postpone --lib (13 passed)
• cargo test --locked -p paimon-c (58 passed)

API and Format

No storage-format change. Fixed-bucket behavior is available only through the explicit Rust/C APIs and requires a caller-provided bucket plan.

[JingsongLi]

TableWrite::prepare_commit declares that the writer is reusable, but the fixed-bucket state is cleared after each prepare. If the writer is reused before the first message has been committed, the second batch will re-derive the number of buckets based on the old snapshot, and subsequent commits may conflict due to inconsistencies in total_buckets for the same partition. The reference implementation restricts this pattern to a one-shot operation.

[XiaoHongbo-Hope]

Thanks @JingsongLi, fixed in a18eede.

[JingsongLi]

• write_arrow_batch appends the internal _VALUE_KIND column via enrich_rowkind_batch.
• prepare_commit calls binary_row_batch_size(batch, table.schema().fields()) when planning the bucket, which requires the number of batch columns to be strictly equal to the number of fields in the user table.
• I added a temporary regression test: bucket=-2, PK table, rowkind.field=op, the write succeeded but prepare_commit consistently failed: BinaryRow size planning expected 3 columns, got 4.
• It is recommended to exclude the internal _VALUE_KIND during sizing and add a fixed-bucket + rowkind regression test.

[JingsongLi]

Create a new rs file for postpone writer.

[JingsongLi]

I found three behavior gaps compared with the merged PyPaimon implementation: distributed writers do not share a bucket plan, overwrite rescaling is rejected by conflict detection, and the size target is validated even when the row-count target should take precedence. Details are inline.

[JingsongLi]

Each TableWrite derives total_buckets from only its local buffered rows. However, the C API supports merging messages from multiple fixed-bucket writers that share a commit_user, and there is no way to provide those writers with one precomputed bucket plan. Two workers writing the same new partition can therefore infer different counts and make the whole commit fail; even when they infer the same count, the plan is based on shard-local rather than global batch statistics and can under-bucket the partition. PyPaimon avoids this by aggregating partition statistics on the driver and injecting the same PostponeBucketPlan into every worker. Please expose and validate a precomputed partition -> total_buckets plan in the Rust builder/writer and C API, or explicitly reject this multi-writer mode.

[JingsongLi]

This bucket-count check runs unconditionally before considering commit_kind, and it compares raw base and delta entries. An overwrite contains DELETE entries for the old layout followed by ADD entries for the replacement layout, so a valid rescale such as 1 -> 3 buckets is rejected as a conflict. The upstream Java implementation skips the old-layout consistency check for OVERWRITE, and PyPaimon has test_postpone_overwrite_allows_bucket_rescale. Please continue checking that all new ADD entries in the delta agree, but compare them with the base layout only for non-overwrite commits (or apply ADD/DELETE changes before checking the final active entries).

[JingsongLi]

postpone.target-size-per-bucket is parsed and validated even when postpone.target-row-num-per-bucket is configured. The option contract says that the size target is ignored in that case, and the PyPaimon planner only reads it in the row target is None branch. With a valid row target plus an invalid or zero size target, Rust currently rejects writer creation even though the size value is unused. Please parse and validate the size target only when no row-count target is present.

[JingsongLi]

One new regression remains after the latest fixes: the core builder now auto-enables fixed mode without the global plan that PyPaimon obtains at the integration layer. Details inline. The previous three findings are fixed.

[JingsongLi]

Why enabling fixed_bucket write by default? Can you create a real PostponeFixedBucketWriteBuilder?

[JingsongLi]

Thanks for adding the fixed-bucket writer. The low-level writer is mostly consistent with PyPaimon, but the high-level distributed workflow is not aligned yet.

[P1] The integration does not select the fixed-bucket path

PyPaimon's Ray sink automatically selects fixed-bucket writing when postpone.batch-write-fixed-bucket=true, gathers global statistics, creates one shared plan, and passes it to every worker (source).

This PR only exposes an explicit Rust/C builder. The DataFusion integration is unchanged, and Rust does not define the corresponding boolean option. Therefore, the normal high-level Rust write path still uses postpone buckets rather than Python's fixed-bucket workflow.

Could we either wire this into the high-level batch integration or explicitly scope the PR to low-level primitives rather than PyPaimon parity?

[P1] Sharing the bucket-count plan is insufficient for safe distributed writes

PyPaimon preclusters data by partition, bucket, and primary key before dispatching it to writers (source). This is important because independent writers targeting the same bucket can restore the same sequence-number state; PyPaimon explicitly documents direct distributed primary-key writes as unsafe without this routing (source).

The new C API shares only partition -> total_buckets. Each writer can still receive arbitrary rows and independently write the same (partition, bucket). The distributed test uses distinct IDs and therefore does not exercise the conflicting-worker case.

Please add an ownership/routing contract—ideally one writer per (partition, bucket)—and a regression test where two workers would otherwise write the same bucket or primary key.

[P2] Rust exposes a resolved plan, but not the Python-equivalent planner

PyPaimon exposes PostponeBucketPlanner, including current metadata, input partition statistics, row/size target precedence, and active postpone-row handling (source).

Rust exposes only PostponeBucketPlan::from_arrow, so every distributed caller must independently reimplement the planning algorithm and compatible size estimation. Exposing a planner/statistics API would make the shared-plan path usable without duplicating internal rules.

One clarification: buffering in the low-level writer without a supplied plan is aligned with PyPaimon's direct writer. The missing parity is primarily the coordinator/integration layer that computes one global plan and safely partitions work across writers.

Until these gaps are addressed, I would describe this PR as providing the core Rust/C fixed-bucket primitives, rather than full alignment with PyPaimon's distributed batch-write workflow.

[XiaoHongbo-Hope]

Thanks for the review. Fixed the distributed ownership issue in afbb593:

• Rust commits and C message merging now reject multiple writers for the same (partition, bucket).
• Added Rust and C regressions with two writers targeting the same primary key/bucket.
• Kept the shared-plan success case on disjoint writer ownership.

I also clarified the PR scope: this PR provides the explicit low-level Rust/C primitives. High-level integration selection, global planning, and preclustering remain integration-layer work.

[JingsongLi]

Thanks for addressing the overlapping-writer correctness issue in afbb593. I rechecked the new Rust and C paths: the commit-time guard is applied to append, expected-snapshot commit, and overwrite, and the focused Rust/C regressions pass.

My remaining concern is architectural rather than another isolated correctness bug. I think the reason this implementation still feels quite different from both Java and PyPaimon is that it sits halfway between a low-level writer primitive and a complete batch-write pipeline.

1. PostponeFixedBucketWriter is doing much more than writing. It loads table metadata, derives bucket counts, buffers batches, routes rows, and manages the one-shot lifecycle through plan_provided, metadata_loaded, and prepare_started (code). PyPaimon separates these responsibilities into PostponeBucketPlanner, PostponeFixedBucketRowKeyExtractor, and PostponeFixedBucketBatchTableWrite. Java moves planning/routing even further toward the execution layer through PostponeFixedBucketChannelComputer and PostponeBatchWriteOperator.

2. The generic TableWrite is now coupled to this special mode. It contains an optional fixed-bucket sidecar, asks that sidecar to plan during prepare_commit, feeds the routed batches back through the generic write_bucket path, and later queries the sidecar again to populate CommitMessage.total_buckets (code). This makes the control flow and lifecycle harder to follow than a dedicated PostponeFixedBucketTableWrite which composes the normal file writers.

3. The stated low-level scope does not quite match the implementation boundary. The PR now says that global planning and shuffling belong to integrations, but the core writer still performs local metadata scans, buffering, planning, and routing. If this is truly a low-level primitive, I would expect it to consume a resolved plan (or preassigned bucket plus total_buckets) and leave planning/topology outside. If local single-process convenience is also required, it could be a separate wrapper around that lower-level primitive.

4. Ownership is detected after writing rather than established by topology. The new TableCommit validation is useful as a defensive backstop, but by that point conflicting workers have already produced files. The same invariant is also duplicated in the C message merge API (code). Java establishes ownership in the channel computer, while PyPaimon establishes it through Ray preclustering before worker writers run. The commit check should remain, but it should not be the primary ownership mechanism.

5. The builder does not describe one coherent operation. with_overwrite() changes writer-side planning, but new_commit() still returns an unconfigured TableCommit; the caller can accidentally pair an overwrite-planned writer with commit() instead of overwrite(). Both Java and PyPaimon carry the overwrite/static-partition mode into the writer and committer created by the builder.

I do not think Rust needs to copy either implementation's class hierarchy, but the responsibility boundaries should be similar. My preferred direction would be:

• a standalone planner producing an immutable resolved plan;
• a pure router mapping (partition, bucket key, plan) to a bucket;
• a dedicated fixed-bucket table writer consuming that plan/router;
• integration-owned selection and (partition, bucket) shuffling;
• commit ownership validation retained only as a final safety check.

At minimum, I think we should decide before merging whether this PR is a genuinely low-level primitive or a local end-to-end implementation. The current middle ground is what makes the code spread across TableWrite, TableCommit, CommitMessage, and the C merge layer, and makes it look structurally different from both reference implementations.

[XiaoHongbo-Hope]

Thanks for addressing the overlapping-writer correctness issue in afbb593. I rechecked the new Rust and C paths: the commit-time guard is applied to append, expected-snapshot commit, and overwrite, and the focused Rust/C regressions pass.

My remaining concern is architectural rather than another isolated correctness bug. I think the reason this implementation still feels quite different from both Java and PyPaimon is that it sits halfway between a low-level writer primitive and a complete batch-write pipeline.

1. PostponeFixedBucketWriter is doing much more than writing. It loads table metadata, derives bucket counts, buffers batches, routes rows, and manages the one-shot lifecycle through plan_provided, metadata_loaded, and prepare_started (code). PyPaimon separates these responsibilities into PostponeBucketPlanner, PostponeFixedBucketRowKeyExtractor, and PostponeFixedBucketBatchTableWrite. Java moves planning/routing even further toward the execution layer through PostponeFixedBucketChannelComputer and PostponeBatchWriteOperator.
2. The generic TableWrite is now coupled to this special mode. It contains an optional fixed-bucket sidecar, asks that sidecar to plan during prepare_commit, feeds the routed batches back through the generic write_bucket path, and later queries the sidecar again to populate CommitMessage.total_buckets (code). This makes the control flow and lifecycle harder to follow than a dedicated PostponeFixedBucketTableWrite which composes the normal file writers.
3. The stated low-level scope does not quite match the implementation boundary. The PR now says that global planning and shuffling belong to integrations, but the core writer still performs local metadata scans, buffering, planning, and routing. If this is truly a low-level primitive, I would expect it to consume a resolved plan (or preassigned bucket plus total_buckets) and leave planning/topology outside. If local single-process convenience is also required, it could be a separate wrapper around that lower-level primitive.
4. Ownership is detected after writing rather than established by topology. The new TableCommit validation is useful as a defensive backstop, but by that point conflicting workers have already produced files. The same invariant is also duplicated in the C message merge API (code). Java establishes ownership in the channel computer, while PyPaimon establishes it through Ray preclustering before worker writers run. The commit check should remain, but it should not be the primary ownership mechanism.
5. The builder does not describe one coherent operation. with_overwrite() changes writer-side planning, but new_commit() still returns an unconfigured TableCommit; the caller can accidentally pair an overwrite-planned writer with commit() instead of overwrite(). Both Java and PyPaimon carry the overwrite/static-partition mode into the writer and committer created by the builder.

I do not think Rust needs to copy either implementation's class hierarchy, but the responsibility boundaries should be similar. My preferred direction would be:

• a standalone planner producing an immutable resolved plan;
• a pure router mapping (partition, bucket key, plan) to a bucket;
• a dedicated fixed-bucket table writer consuming that plan/router;
• integration-owned selection and (partition, bucket) shuffling;
• commit ownership validation retained only as a final safety check.

At minimum, I think we should decide before merging whether this PR is a genuinely low-level primitive or a local end-to-end implementation. The current middle ground is what makes the code spread across TableWrite, TableCommit, CommitMessage, and the C merge layer, and makes it look structurally different from both reference implementations.

Got it

[XiaoHongbo-Hope]

@JingsongLi Thanks for your patient and detailed reviews throughout these iterations. The original intent was simply to allow Go users to write to bucket = -2 tables and make the written data visible immediately.

However, the user has now migrated the table to fixed-bucket mode, so I have lowered this PR to medium priority. I have done self-review tonight, removed the local bucket-planning wrapper, and refactored the implementation to make the change more clear.

This PR now focuses only on the low-level APIs. In a follow-up PR, I will implement bucket planning on the Go side and connect it to these APIs to complete the end-to-end Go write path. Given above information, could you help review the PR again when you are free.