XiaoHongbo-Hope commented on PR #8985:
URL: https://github.com/apache/paimon/pull/8985#issuecomment-5154567891
> > 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`:
> > ```
> > 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:
> > ```
> > 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:
> > ```
> > 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.
>
> Thanks for the review.
>
> 1. Multiple `write_arrow_batch()` inputs are now planned together before
writing. Ray keeps its driver-side plan; Daft temporarily retains the legacy
postpone path. Daft will be processed in following PR.
> 2. Added `postpone.target-row-num-per-bucket` with precedence over size,
including existing postpone rows for append but not overwrite.
> 3. Added coverage for multi-batch, multi-worker conflicts, row-count
planning, and Daft fallback.
> cc @JingsongLi
--
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]
