leaves12138 commented on PR #8985:
URL: https://github.com/apache/paimon/pull/8985#issuecomment-5154428998
Thanks for working on this. The overall direction is aligned with the Java
Spark fixed-bucket batch path (reuse existing bucket counts, include active
postpone rows for append estimation, carry `total_buckets`, and detect
bucket-count conflicts), but I found two correctness/parity issues that should
be addressed before merging.
### 1. Bucket planning is not global for chunked or distributed batch writes
`TableWrite.write_arrow_batch` calls `plan_input` for each incoming batch,
while `PostponeFixedBucketRowKeyExtractor.plan_num_buckets` skips a partition
as soon as it is present in `_known_num_buckets`. Therefore, the first batch
containing a new partition permanently determines its bucket count; later
batches are not included in the estimate.
I reproduced this with the same logical input and
`postpone.target-size-per-bucket=100 b`, `max-parallelism=3`:
```text
write_arrow_batch(small_batch), then write_arrow_batch(large_batch): {1}
write_arrow(concatenated_input): {3}
```
This differs from Java Spark's `preparePostponeBucketAssignment`, which
persists the complete DataFrame, collects all partition statistics, and
resolves bucket counts before routing any rows.
This also affects Daft. `PaimonDataSink.write` creates an independent
`BatchTableWrite` in each worker and feeds it record batches. Different workers
can choose different bucket counts for the same new partition. A two-writer
reproduction produced plans `{1}` and `{3}`, and the combined commit failed
with:
```text
CommitConflictError: Total buckets of partition ('p',) changed from 1 to 3
without overwrite.
```
Ray avoids this because this PR adds a driver-side plan, but the generic
batch API and Daft do not. Bucket planning needs to happen once over the
complete input before writing. Daft likely needs a coordinated driver plan
similar to Ray, or should retain the legacy postpone-bucket path until such
planning is available. Please also add coverage for multiple
`write_arrow_batch` calls and multiple worker message sets for the same
partition.
### 2. `postpone.target-row-num-per-bucket` is ignored
Java defines `postpone.target-row-num-per-bucket` and Spark gives it
precedence over `postpone.target-size-per-bucket`. PyPaimon only adds the size
option and always plans by Arrow size.
I reproduced this with `target-row-num-per-bucket=1`,
`target-size-per-bucket=1 gb`, `max-parallelism=8`, and 8 rows in one new
partition:
```text
PyPaimon: {1}
Java Spark: {8}
```
This is especially problematic for tables created/configured through Java:
PyPaimon silently ignores the persisted table option when it creates the first
real buckets for a new partition. Please add the option/accessor and choose the
row-count calculation when configured, including active postpone rows for
append and excluding them for overwrite, matching the Java Spark behavior.
For context, Java Flink's batch path uses capped sink parallelism for
unknown partitions, while this implementation is primarily adapting Java
Spark's data-statistics behavior, so it is not expected to be numerically
identical to every Java engine. However, the global planning and row-count
precedence above are correctness/configuration semantics that should remain
consistent.
--
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]
