feat(write): route postpone batch writes to fixed buckets

[XiaoHongbo-Hope]

Purpose

Align Rust batch writes for postpone-bucket primary-key tables with the fixed-bucket behavior proposed in apache/paimon#8985. This extracts the core writer change from #658; Go binding and release changes remain in the draft PR.

Changes

• route batch writes to non-negative fixed buckets by default, while preserving legacy bucket -2 behavior when postpone.batch-write-fixed-bucket=false
• infer bucket counts for new partitions and reuse persisted per-partition counts on later writes
• stream writes for partitions whose bucket counts are already known, avoiding accumulation of every Arrow batch
• persist total_buckets in commit messages and reject concurrent conflicting bucket-count decisions
• reject the unsupported combination with deletion vectors, whose scans skip the level-0 files produced by batch writers
• cover visibility, repeated writes, partitioning, conflict detection, option validation, and DataFusion reads

Verification

• cargo fmt --all -- --check
• cargo check --locked --offline -p paimon
• cargo test --locked --offline -p paimon test_postpone_ --lib
• cargo test --locked --offline -p paimon-datafusion test_postpone_batch_write_uses_visible_fixed_bucket --test pk_tables -- --exact
• DLF regression REST catalog E2E: write, commit, and immediate read from real buckets

[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.