You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@paimon.apache.org by "XiaoHongbo-Hope (via GitHub)" <gi...@apache.org> on 2026/08/01 14:11:08 UTC
[PR] [python] Write postpone batches to fixed buckets [paimon]
XiaoHongbo-Hope opened a new pull request, #8985:
URL: https://github.com/apache/paimon/pull/8985
### Purpose
PyPaimon batch writes to postpone-bucket tables currently always create `bucket-postpone` files. Unlike the Java batch integrations, those files are not visible to normal readers until compaction finishes.
This aligns PyPaimon batch writes with the default `postpone.batch-write-fixed-bucket=true` behavior so committed rows are immediately visible.
### Changes
- Route PyPaimon batch rows directly to real buckets for `bucket=-2` tables.
- Reuse the current bucket count for existing partitions and use `postpone.default-bucket-num` (default `1`) for new partitions.
- Carry the per-partition total bucket count through commit messages and manifest entries.
- Reject a concurrent non-overwrite commit that would mix bucket counts in one partition.
- Preserve the old `bucket-postpone` behavior when `postpone.batch-write-fixed-bucket=false`.
- Update the Ray write documentation and regression test.
Streaming writes are unchanged.
### Tests
- `table_write_test.py`, `file_store_commit_test.py`, `table_commit_test.py`, and `conflict_detection_test.py`: 76 passed, 3 subtests passed.
- Ray postpone-bucket round trip: 1 passed; all files landed in real bucket 0 and all 10 rows were readable.
- Python 3.6 `py_compile`: passed.
- `git diff --check`: passed.
--
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: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [python] Write postpone batches to fixed buckets [paimon]
Posted by "XiaoHongbo-Hope (via GitHub)" <gi...@apache.org>.
XiaoHongbo-Hope commented on PR #8985:
URL: https://github.com/apache/paimon/pull/8985#issuecomment-5164595074
> You can take a review to #8992
Thanks. I aligned the implementation with #8992. Could you take another look?
--
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: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [python] Write postpone batches to fixed buckets [paimon]
Posted by "leaves12138 (via GitHub)" <gi...@apache.org>.
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: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [python] Write postpone batches to fixed buckets [paimon]
Posted by "leaves12138 (via GitHub)" <gi...@apache.org>.
leaves12138 commented on PR #8985:
URL: https://github.com/apache/paimon/pull/8985#issuecomment-5154665092
Thanks for the update. I re-tested the two original issues: multiple `write_arrow_batch` calls now produce the same bucket plan as the concatenated input, and `postpone.target-row-num-per-bucket` now correctly takes precedence. The Daft legacy fallback also looks reasonable.
I found one remaining API consistency issue: `BatchTableWrite.write_row` still bypasses the new postpone planning path. It immediately calls `extract_partition_bucket_row`, and `PostponeFixedBucketRowKeyExtractor.num_buckets` defaults an unknown partition to one bucket.
With `postpone.target-row-num-per-bucket=1`, `postpone.batch-write-fixed-bucket.max-parallelism=8`, and 8 rows in one new partition, I get:
```text
write_arrow: {8}
write_row: {1}
```
This means the bucket count depends on which public write API is used. Calling `write_row` first also records the new partition as one bucket, so subsequent Arrow batches cannot re-plan it. Please either buffer and plan row writes together with Arrow inputs, or explicitly reject `write_row` for postpone fixed-bucket batch writes, and add a regression test.
One additional resource concern: `_postpone_batches` retains all unknown-partition `RecordBatch` objects until `prepare_commit`, so the chunked API now has memory usage proportional to the complete input and cannot benefit from normal incremental file rolling. Spark can persist/spill and Ray can use object-store spilling; the local Python path has no equivalent. Please at least document this memory bound, or consider temporary spilling or an explicit preplanned bucket map.
--
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: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [python] Write postpone batches to fixed buckets [paimon]
Posted by "JingsongLi (via GitHub)" <gi...@apache.org>.
JingsongLi commented on PR #8985:
URL: https://github.com/apache/paimon/pull/8985#issuecomment-5188035703
+1
--
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: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [python] Write postpone batches to fixed buckets [paimon]
Posted by "XiaoHongbo-Hope (via GitHub)" <gi...@apache.org>.
XiaoHongbo-Hope commented on code in PR #8985:
URL: https://github.com/apache/paimon/pull/8985#discussion_r3704109369
##########
paimon-python/pypaimon/table/file_store_table.py:
##########
@@ -426,8 +426,17 @@ def new_stream_read_builder(self) -> 'StreamReadBuilder':
return StreamReadBuilder(self)
def new_batch_write_builder(self) -> BatchWriteBuilder:
+ if (self.bucket_mode() == BucketMode.POSTPONE_MODE
Review Comment:
> Please not do this. By default, we should not use this mode.
Updated
--
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: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [python] Write postpone batches to fixed buckets [paimon]
Posted by "XiaoHongbo-Hope (via GitHub)" <gi...@apache.org>.
XiaoHongbo-Hope commented on PR #8985:
URL: https://github.com/apache/paimon/pull/8985#issuecomment-5154563426
> 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.
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: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [python] Write postpone batches to fixed buckets [paimon]
Posted by "JingsongLi (via GitHub)" <gi...@apache.org>.
JingsongLi commented on code in PR #8985:
URL: https://github.com/apache/paimon/pull/8985#discussion_r3718186856
##########
paimon-python/pypaimon/write/file_store_write.py:
##########
@@ -330,3 +377,10 @@ def _load_seq_number_stats(self, partition: Tuple) -> dict:
if current_seq_num > existing_max:
max_seq_numbers[split.bucket] = current_seq_num
return max_seq_numbers
+
+
+class PostponeFixedBucketFileStoreWrite(FileStoreWrite):
Review Comment:
Move _runtime_total_buckets into this class.
--
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: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [python] Write postpone batches to fixed buckets [paimon]
Posted by "JingsongLi (via GitHub)" <gi...@apache.org>.
JingsongLi commented on code in PR #8985:
URL: https://github.com/apache/paimon/pull/8985#discussion_r3703328043
##########
paimon-python/pypaimon/write/commit/overwrite_changes_provider.py:
##########
@@ -132,12 +132,17 @@ def _build_result(self, existing_entries: List[ManifestEntry]) -> List[ManifestE
# New files being written by this overwrite.
for msg in self.commit_messages:
partition = GenericRow(list(msg.partition), self.table.partition_keys_fields)
+ total_buckets = (
+ msg.total_buckets
+ if msg.total_buckets is not None
+ else self.table.total_buckets
+ )
for file in msg.new_files:
entries.append(ManifestEntry(
kind=0,
partition=partition,
bucket=msg.bucket,
- total_buckets=self.table.total_buckets,
+ total_buckets=total_buckets,
Review Comment:
[P2] Preserve the replacement bucket count in partition statistics. This list emits DELETE entries first with the old `total_buckets`, then ADD entries with the new value. `_generate_partition_statistics` records only the first value it sees for a partition, so an overwrite from postpone mode (`-2`) to `N` leaves catalog partition statistics at `-2` even though the new manifest entries use `N`. Java's `PartitionEntry.merge` takes the later entry's `totalBuckets`. Please make the statistics aggregation last-wins (and add a catalog-statistics assertion for `-2 -> N`, ideally also `N -> M`).
##########
paimon-python/pypaimon/write/row_key_extractor.py:
##########
@@ -590,3 +590,59 @@ def _extract_buckets_batch(self, data: pa.RecordBatch) -> List[int]:
def _extract_bucket_row(self, values_by_name: Dict[str, Any]) -> int:
return BucketMode.POSTPONE_BUCKET.value
+
+
+class PostponeFixedBucketRowKeyExtractor(RowKeyExtractor):
+ """Route postpone batches using a resolved bucket plan."""
+
+ def __init__(self, table, bucket_plan):
+ super().__init__(table.table_schema)
+ if table.options.bucket() != BucketMode.POSTPONE_BUCKET.value:
+ raise ValueError(
+ "Postpone fixed bucket writes require bucket = -2, got {}".format(
+ table.options.bucket()
+ )
+ )
+ self.bucket_keys = table.table_schema.bucket_keys
+ self.bucket_key_indices = self._get_field_indices(self.bucket_keys)
+ self._bucket_key_fields = table.table_schema.logical_bucket_key_fields
+ self._bucket_plan = bucket_plan
+
+ def with_bucket_plan(self, bucket_plan) -> None:
+ self._bucket_plan = bucket_plan
+
+ def num_buckets(self, partition: Tuple) -> int:
+ return self._bucket_plan.num_buckets(partition)
+
+ def extract_partition_bucket_batch(
+ self, data: pa.RecordBatch
+ ) -> Tuple[List[Tuple], List[int]]:
+ partitions = self._extract_partitions_batch(data)
+ columns = [data.column(i) for i in self.bucket_key_indices]
+ buckets = [
+ _bucket_from_hash(
Review Comment:
[P1] Respect the configured bucket function. This extractor always applies the default BinaryRow hash, while Java's `PostponeFixBucketProcessor` constructs `BucketFunction` from `CoreOptions` and therefore supports `default`, `mod`, and `hive`. For a valid `bucket-function.type=mod` or `hive` table, Python and Java can route the same primary key to different buckets; Java bucket pruning then follows the configured function and may miss rows written by Python. Please dispatch according to `bucket-function.type` (with Java-compatible semantics), or reject non-default functions until they are supported, and add cross-engine fixtures for all supported functions.
##########
paimon-python/pypaimon/write/postpone_bucket.py:
##########
@@ -0,0 +1,189 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+from typing import Dict, Tuple
+
+import pyarrow as pa
+
+from pypaimon.table.bucket_mode import BucketMode
+
+
+class PostponeBucketPlan:
+ """Resolved bucket counts by partition."""
+
+ def __init__(self, num_buckets: Dict[Tuple, int]):
+ self._num_buckets = dict(num_buckets)
+
+ def contains(self, partition: Tuple) -> bool:
+ return tuple(partition) in self._num_buckets
+
+ def num_buckets(self, partition: Tuple) -> int:
+ partition = tuple(partition)
+ if partition not in self._num_buckets:
+ raise ValueError("Missing bucket plan for partition {}".format(partition))
+ return self._num_buckets[partition]
+
+ def as_dict(self) -> Dict[Tuple, int]:
+ return dict(self._num_buckets)
+
+
+class PostponeBucketPlanner:
+ """Plans fixed bucket counts for postpone batch writes."""
+
+ def __init__(
+ self,
+ table,
+ known_num_buckets=None,
+ postpone_row_counts=None,
+ ):
+ options = table.options
+ if options.bucket() != BucketMode.POSTPONE_BUCKET.value:
+ raise ValueError(
+ "Postpone fixed bucket writes require bucket = -2, got {}"
+ .format(options.bucket())
+ )
+
+ self.max_num_buckets = (
+ options.postpone_batch_write_fixed_bucket_max_parallelism()
+ )
+ if self.max_num_buckets <= 0:
+ raise ValueError(
+ "postpone.batch-write-fixed-bucket.max-parallelism must be "
+ "positive, got {}".format(self.max_num_buckets)
+ )
+ self.target_row_num_per_bucket = (
+ options.postpone_target_row_num_per_bucket()
+ )
+ if self.target_row_num_per_bucket is not None:
+ if self.target_row_num_per_bucket <= 0:
+ raise ValueError(
+ "postpone.target-row-num-per-bucket must be positive, "
+ "got {}".format(self.target_row_num_per_bucket)
+ )
+ self.target_size_per_bucket = None
+ else:
+ self.target_size_per_bucket = (
+ options.postpone_target_size_per_bucket()
+ )
+ if self.target_size_per_bucket <= 0:
+ raise ValueError(
+ "postpone.target-size-per-bucket must be positive, got "
+ "{}".format(self.target_size_per_bucket)
+ )
+
+ self._partition_indices = [
+ table.field_names.index(key) for key in table.partition_keys
+ ]
+ if known_num_buckets is None:
+ known_num_buckets, loaded_postpone_counts = (
+ self._load_bucket_metadata(table)
+ )
+ if postpone_row_counts is None:
+ postpone_row_counts = loaded_postpone_counts
+ self._known_num_buckets = dict(known_num_buckets)
+ self._postpone_row_counts = dict(postpone_row_counts or {})
+
+ @staticmethod
+ def _load_bucket_metadata(table):
+ scan = table.new_read_builder().new_scan().file_scanner
+ manifest_files, _ = scan.manifest_scanner()
+ entries = scan.manifest_file_manager.read_entries_parallel(
+ manifest_files,
+ max_workers=table.options.scan_manifest_parallelism(),
+ )
+
+ known = {}
+ postpone_counts = {}
+ for entry in entries:
+ partition = tuple(entry.partition.values)
+ if entry.bucket == BucketMode.POSTPONE_BUCKET.value:
+ postpone_counts[partition] = (
+ postpone_counts.get(partition, 0) + entry.file.row_count
+ )
+ elif entry.bucket >= 0 and entry.total_buckets > 0:
+ previous = known.get(partition)
+ if previous is not None and previous != entry.total_buckets:
+ raise RuntimeError(
+ "Partition {} has different total buckets {} and {}"
+ .format(partition, previous, entry.total_buckets)
+ )
+ known[partition] = entry.total_buckets
+ return known, postpone_counts
+
+ def current_plan(self) -> PostponeBucketPlan:
+ return PostponeBucketPlan(self._known_num_buckets)
+
+ def input_partition_stats(self, data) -> Dict[Tuple, Tuple[int, int]]:
+ if self._partition_indices:
+ columns = [data.column(i) for i in self._partition_indices]
+ partitions = [
+ tuple(column[row].as_py() for column in columns)
+ for row in range(data.num_rows)
+ ]
+ else:
+ partitions = [()] * data.num_rows
+
+ row_indices = {}
+ for index, partition in enumerate(partitions):
+ row_indices.setdefault(partition, []).append(index)
+
+ stats = {}
+ for partition, indices in row_indices.items():
+ partition_data = (
+ data
+ if len(indices) == data.num_rows
+ else pa.compute.take(data, pa.array(indices, type=pa.int64()))
+ )
+ stats[partition] = (len(indices), partition_data.nbytes)
Review Comment:
[P2] Use the same size metric as Java when deriving the bucket count. `RecordBatch.nbytes` measures Arrow buffer memory, whereas Java sums `BinaryRow.getSizeInBytes()`. These values differ materially; for example, 1,000 `(INT, STRING, STRING)` rows occupy 14,000 bytes by this metric but 32,000 bytes as Java BinaryRows, so a 20 KiB target plans 1 bucket in Python and 2 in Java. Because the first writer persists the partition's bucket count, engine order changes the table layout. Please use a BinaryRow-compatible serialized-size calculation (or an exactly equivalent vectorized calculation) and verify it with shared fixtures.
##########
paimon-python/pypaimon/write/file_store_commit.py:
##########
@@ -211,6 +211,10 @@ def commit(self, commit_messages: List[CommitMessage], commit_identifier: int):
if self.conflict_detection.has_hash_index_changes(
index_adds + index_deletes):
detect_conflicts = True
+ if any(message.total_buckets is not None
+ for message in commit_messages):
+ # Detect concurrent bucket-count changes in postpone APPENDs.
+ detect_conflicts = True
Review Comment:
[P2] Keep deterministic CAS failures on the abortable conflict path. Enabling conflict detection here exposes a cleanup gap: when `atomic_commit` returns `False`, `_try_commit_once` creates `RetryResult(exception=None)` because the snapshot definitely was not committed. If the retry then detects a bucket-count conflict, `_try_commit` converts it to a generic exception solely because `retry_result` is non-null. `TableCommit` aborts files only for `CommitConflictError`, so the losing writer's files can be orphaned. Please distinguish `retry_result.exception is None` from an uncertain commit exception and retain `CommitConflictError` for the former; add a concurrent different-plan test that also asserts cleanup.
--
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: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [python] Write postpone batches to fixed buckets [paimon]
Posted by "XiaoHongbo-Hope (via GitHub)" <gi...@apache.org>.
XiaoHongbo-Hope commented on PR #8985:
URL: https://github.com/apache/paimon/pull/8985#issuecomment-5158109427
> You might need to introduce some abstraction—right now, a bunch of `Postpone` implementations are tightly coupled within the common classes `TableWrite` and `BatchTableWrite`.
Got it, updated
--
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: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [python] Write postpone batches to fixed buckets [paimon]
Posted by "JingsongLi (via GitHub)" <gi...@apache.org>.
JingsongLi commented on code in PR #8985:
URL: https://github.com/apache/paimon/pull/8985#discussion_r3712299780
##########
paimon-python/pypaimon/table/row/generic_row.py:
##########
@@ -447,6 +459,26 @@ def to_bytes(cls, row: Union[GenericRow, BinaryRow]) -> bytes:
absolute_offset = fixed_part_size + offset_in_variable_part
offset_and_nano = (absolute_offset << 32) | nano_of_millisecond
struct.pack_into('<q', fixed_part, field_fixed_offset, offset_and_nano)
+ elif is_variant_type:
Review Comment:
[P1] Complete the VARIANT read path before accepting it here. This branch now lets a VARIANT bucket/primary key serialize and commit, but `GenericRowDeserializer` has no VARIANT case and falls through to `_parse_string()` when manifest `_MIN_KEY`/`_MAX_KEY` values are loaded. I reproduced this with a valid `GenericVariant.from_python(-1)`: the fixed-bucket commit succeeds, but reopening the table and calling `new_scan().plan()` raises `UnicodeDecodeError` on the payload's `0xff` byte. Please add the matching VARIANT decoder (offset/size, followed by the 4-byte value length, value, and metadata) and an end-to-end commit/reopen/scan test using a real variant value with non-UTF-8 payload bytes.
##########
paimon-python/pypaimon/table/row/generic_row.py:
##########
@@ -420,13 +426,21 @@ def to_bytes(cls, row: Union[GenericRow, BinaryRow]) -> bytes:
'CHAR', 'VARCHAR', 'STRING', 'BINARY', 'VARBINARY', 'BYTES', 'BLOB'])
is_decimal_type = type_name.startswith('DECIMAL') or type_name.startswith('NUMERIC')
is_timestamp_type = type_name.startswith('TIMESTAMP')
+ is_ltz_type = type_name.startswith('TIMESTAMP_LTZ') or 'WITH LOCAL TIME ZONE' in type_name
+ is_variant_type = type_name == 'VARIANT'
if is_decimal_type or is_timestamp_type:
precision, scale = _parse_type_precision_scale(field.type)
else:
precision, scale = 0, 0
is_high_precision_decimal = is_decimal_type and precision > 18
is_non_compact_timestamp = is_timestamp_type and precision > 3
+ if is_timestamp_type:
+ if is_ltz_type:
+ value = _normalize_ltz(value)
Review Comment:
[P1] Preserve LTZ awareness when deserializing manifest statistics. Normalizing an aware LTZ value to naive UTC makes the BinaryRow hash match Java, but `_parse_timestamp()` still returns a naive `datetime` for both TIMESTAMP and TIMESTAMP_LTZ. I reproduced a valid LTZ primary-key fixed-bucket commit succeeding, followed by an equality-filtered scan failing while comparing the naive manifest min/max with the aware Arrow predicate literal: `TypeError: can't compare offset-naive and offset-aware datetimes`. Please return UTC-aware values when the declared type is TIMESTAMP_LTZ (including the WITH LOCAL TIME ZONE spelling) and add a commit/reopen/filter regression test, ideally covering both compact and non-compact precisions.
--
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: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [python] Write postpone batches to fixed buckets [paimon]
Posted by "JingsongLi (via GitHub)" <gi...@apache.org>.
JingsongLi commented on PR #8985:
URL: https://github.com/apache/paimon/pull/8985#issuecomment-5162677623
You can take a review to https://github.com/apache/paimon/pull/8992
--
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: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [python] Write postpone batches to fixed buckets [paimon]
Posted by "JingsongLi (via GitHub)" <gi...@apache.org>.
JingsongLi commented on code in PR #8985:
URL: https://github.com/apache/paimon/pull/8985#discussion_r3703133429
##########
paimon-python/pypaimon/table/file_store_table.py:
##########
@@ -426,8 +426,17 @@ def new_stream_read_builder(self) -> 'StreamReadBuilder':
return StreamReadBuilder(self)
def new_batch_write_builder(self) -> BatchWriteBuilder:
+ if (self.bucket_mode() == BucketMode.POSTPONE_MODE
Review Comment:
Please not do this. By default, we should not use this mode.
--
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: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [python] Write postpone batches to fixed buckets [paimon]
Posted by "XiaoHongbo-Hope (via GitHub)" <gi...@apache.org>.
XiaoHongbo-Hope commented on code in PR #8985:
URL: https://github.com/apache/paimon/pull/8985#discussion_r3704145933
##########
paimon-python/pypaimon/write/commit/overwrite_changes_provider.py:
##########
@@ -132,12 +132,17 @@ def _build_result(self, existing_entries: List[ManifestEntry]) -> List[ManifestE
# New files being written by this overwrite.
for msg in self.commit_messages:
partition = GenericRow(list(msg.partition), self.table.partition_keys_fields)
+ total_buckets = (
+ msg.total_buckets
+ if msg.total_buckets is not None
+ else self.table.total_buckets
+ )
for file in msg.new_files:
entries.append(ManifestEntry(
kind=0,
partition=partition,
bucket=msg.bucket,
- total_buckets=self.table.total_buckets,
+ total_buckets=total_buckets,
Review Comment:
> [P2] Preserve the replacement bucket count in partition statistics. This list emits DELETE entries first with the old `total_buckets`, then ADD entries with the new value. `_generate_partition_statistics` records only the first value it sees for a partition, so an overwrite from postpone mode (`-2`) to `N` leaves catalog partition statistics at `-2` even though the new manifest entries use `N`. Java's `PartitionEntry.merge` takes the later entry's `totalBuckets`. Please make the statistics aggregation last-wins (and add a catalog-statistics assertion for `-2 -> N`, ideally also `N -> M`).
Fixed
##########
paimon-python/pypaimon/write/file_store_commit.py:
##########
@@ -211,6 +211,10 @@ def commit(self, commit_messages: List[CommitMessage], commit_identifier: int):
if self.conflict_detection.has_hash_index_changes(
index_adds + index_deletes):
detect_conflicts = True
+ if any(message.total_buckets is not None
+ for message in commit_messages):
+ # Detect concurrent bucket-count changes in postpone APPENDs.
+ detect_conflicts = True
Review Comment:
> [P2] Keep deterministic CAS failures on the abortable conflict path. Enabling conflict detection here exposes a cleanup gap: when `atomic_commit` returns `False`, `_try_commit_once` creates `RetryResult(exception=None)` because the snapshot definitely was not committed. If the retry then detects a bucket-count conflict, `_try_commit` converts it to a generic exception solely because `retry_result` is non-null. `TableCommit` aborts files only for `CommitConflictError`, so the losing writer's files can be orphaned. Please distinguish `retry_result.exception is None` from an uncertain commit exception and retain `CommitConflictError` for the former; add a concurrent different-plan test that also asserts cleanup.
Fixed
--
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: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [python] Write postpone batches to fixed buckets [paimon]
Posted by "JingsongLi (via GitHub)" <gi...@apache.org>.
JingsongLi merged PR #8985:
URL: https://github.com/apache/paimon/pull/8985
--
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: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [python] Write postpone batches to fixed buckets [paimon]
Posted by "JingsongLi (via GitHub)" <gi...@apache.org>.
JingsongLi commented on code in PR #8985:
URL: https://github.com/apache/paimon/pull/8985#discussion_r3704289332
##########
paimon-python/pypaimon/write/postpone_bucket.py:
##########
@@ -0,0 +1,198 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+from typing import Dict, Tuple
+
+from pypaimon.table.bucket_mode import BucketMode
+from pypaimon.table.row.generic_row import GenericRow, GenericRowSerializer
+from pypaimon.table.row.internal_row import RowKind
+
+
+class PostponeBucketPlan:
+ """Resolved bucket counts by partition."""
+
+ def __init__(self, num_buckets: Dict[Tuple, int]):
+ self._num_buckets = dict(num_buckets)
+
+ def contains(self, partition: Tuple) -> bool:
+ return tuple(partition) in self._num_buckets
+
+ def num_buckets(self, partition: Tuple) -> int:
+ partition = tuple(partition)
+ if partition not in self._num_buckets:
+ raise ValueError("Missing bucket plan for partition {}".format(partition))
+ return self._num_buckets[partition]
+
+ def as_dict(self) -> Dict[Tuple, int]:
+ return dict(self._num_buckets)
+
+
+class PostponeBucketPlanner:
+ """Plans fixed bucket counts for postpone batch writes."""
+
+ def __init__(
+ self,
+ table,
+ known_num_buckets=None,
+ postpone_row_counts=None,
+ ):
+ options = table.options
+ if options.bucket() != BucketMode.POSTPONE_BUCKET.value:
+ raise ValueError(
+ "Postpone fixed bucket writes require bucket = -2, got {}"
+ .format(options.bucket())
+ )
+ bucket_function = str(
+ table.table_schema.options.get("bucket-function.type", "default")
+ ).strip().lower()
+ if bucket_function != "default":
+ raise ValueError(
+ "Postpone fixed bucket writes only support "
+ "bucket-function.type=default, got {}"
+ .format(bucket_function)
+ )
+
+ self.max_num_buckets = (
+ options.postpone_batch_write_fixed_bucket_max_parallelism()
+ )
+ if self.max_num_buckets <= 0:
+ raise ValueError(
+ "postpone.batch-write-fixed-bucket.max-parallelism must be "
+ "positive, got {}".format(self.max_num_buckets)
+ )
+ self.target_row_num_per_bucket = (
+ options.postpone_target_row_num_per_bucket()
+ )
+ if self.target_row_num_per_bucket is not None:
+ if self.target_row_num_per_bucket <= 0:
+ raise ValueError(
+ "postpone.target-row-num-per-bucket must be positive, "
+ "got {}".format(self.target_row_num_per_bucket)
+ )
+ self.target_size_per_bucket = None
+ else:
+ self.target_size_per_bucket = (
+ options.postpone_target_size_per_bucket()
+ )
+ if self.target_size_per_bucket <= 0:
+ raise ValueError(
+ "postpone.target-size-per-bucket must be positive, got "
+ "{}".format(self.target_size_per_bucket)
+ )
+
+ self._partition_keys = list(table.partition_keys)
+ self._field_dict = dict(table.field_dict)
+ if known_num_buckets is None:
+ known_num_buckets, loaded_postpone_counts = (
+ self._load_bucket_metadata(table)
+ )
+ if postpone_row_counts is None:
+ postpone_row_counts = loaded_postpone_counts
+ self._known_num_buckets = dict(known_num_buckets)
+ self._postpone_row_counts = dict(postpone_row_counts or {})
+
+ @staticmethod
+ def _load_bucket_metadata(table):
+ scan = table.new_read_builder().new_scan().file_scanner
+ manifest_files, _ = scan.manifest_scanner()
+ entries = scan.manifest_file_manager.read_entries_parallel(
+ manifest_files,
+ max_workers=table.options.scan_manifest_parallelism(),
+ )
+
+ known = {}
+ postpone_counts = {}
+ for entry in entries:
+ partition = tuple(entry.partition.values)
+ if entry.bucket == BucketMode.POSTPONE_BUCKET.value:
+ postpone_counts[partition] = (
+ postpone_counts.get(partition, 0) + entry.file.row_count
+ )
+ elif entry.bucket >= 0 and entry.total_buckets > 0:
+ previous = known.get(partition)
+ if previous is not None and previous != entry.total_buckets:
+ raise RuntimeError(
+ "Partition {} has different total buckets {} and {}"
+ .format(partition, previous, entry.total_buckets)
+ )
+ known[partition] = entry.total_buckets
+ return known, postpone_counts
+
+ def current_plan(self) -> PostponeBucketPlan:
+ return PostponeBucketPlan(self._known_num_buckets)
+
+ def input_partition_stats(self, data) -> Dict[Tuple, Tuple[int, int]]:
+ if self._partition_keys:
+ columns = [data.column(key) for key in self._partition_keys]
+ partitions = [
+ tuple(column[row].as_py() for column in columns)
+ for row in range(data.num_rows)
+ ]
+ else:
+ partitions = [()] * data.num_rows
+
+ stats = {}
+ fields = [self._field_dict[name] for name in data.schema.names]
+ columns = [data.column(i) for i in range(len(fields))]
+ collect_size = self.target_row_num_per_bucket is None
+ for row, partition in enumerate(partitions):
+ data_size = 0
+ if collect_size:
+ values = [column[row].as_py() for column in columns]
+ data_size = len(GenericRowSerializer.to_bytes(
Review Comment:
[P1] Do not use the atomic-only `GenericRowSerializer` for full-row size planning. This now serializes every input field, but that serializer rejects non-`AtomicType` values and timezone-aware timestamps; Java uses `InternalSerializers.create(rowType)`, which supports ARRAY, MAP, ROW, VECTOR, VARIANT, BLOB, and TIMESTAMP_LTZ. I reproduced an `ARRAY<INT>` value failing with `ValueError: BinaryRow only support AtomicType`, and an `id` primary-key table with a non-key `TIMESTAMP_LTZ` value succeeds through the legacy postpone writer but fails here with `datetime tzinfo not supported yet`. Ray enters this size-based planner by default, so valid existing postpone writes regress before routing any rows. Please use a Java-compatible full type-surface size calculator/serializer and add nested-value and non-key TIMESTAMP_LTZ fixtures in addition to the current all-atomic 32,000-byte fixture.
##########
paimon-python/pypaimon/write/file_store_commit.py:
##########
@@ -458,6 +462,8 @@ def _try_commit(self, commit_kind, commit_identifier, commit_entries_plan,
f"after {elapsed_ms} millis with {retry_count} retries, "
f"there maybe exist commit conflicts between multiple jobs."
)
+ if retry_result is not None and retry_result.exception is None:
Review Comment:
[P1] Preserve commit uncertainty across all retries before aborting files. `_try_commit` replaces `retry_result` after every attempt, so an earlier commit exception (which may have occurred after the snapshot was successfully committed) can be overwritten by a later CAS `False` result with `exception=None`. If duplicate detection cannot read the original snapshot because it has expired or is temporarily unavailable, this branch raises `CommitConflictError`, and `TableCommit` aborts the data files even though a later snapshot can still reference them through its base manifests. I reproduced: attempt 1 commits snapshot 1 then loses the response; another writer creates snapshot 2; snapshot 1 is unavailable to `_is_duplicate_commit`; attempt 2 returns `False`; the final `CommitConflictError` deletes snapshot 1's data file while snapshot 2 remains latest. Please keep a monotonic `commit_result_may_be_uncertain` state across the whole retry loop (cleared only after duplicate-success con
firmation) and add this multi-attempt regression test.
--
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: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [python] Write postpone batches to fixed buckets [paimon]
Posted by "JingsongLi (via GitHub)" <gi...@apache.org>.
JingsongLi commented on code in PR #8985:
URL: https://github.com/apache/paimon/pull/8985#discussion_r3709224648
##########
paimon-python/pypaimon/write/row_key_extractor.py:
##########
@@ -590,3 +590,68 @@ def _extract_buckets_batch(self, data: pa.RecordBatch) -> List[int]:
def _extract_bucket_row(self, values_by_name: Dict[str, Any]) -> int:
return BucketMode.POSTPONE_BUCKET.value
+
+
+class PostponeFixedBucketRowKeyExtractor(RowKeyExtractor):
+ """Route postpone batches using a resolved bucket plan."""
+
+ def __init__(self, table, bucket_plan):
+ super().__init__(table.table_schema)
+ if table.options.bucket() != BucketMode.POSTPONE_BUCKET.value:
+ raise ValueError(
+ "Postpone fixed bucket writes require bucket = -2, got {}".format(
+ table.options.bucket()
+ )
+ )
+ bucket_function = str(
+ table.table_schema.options.get("bucket-function.type", "default")
+ ).strip().lower()
+ if bucket_function != "default":
+ raise ValueError(
+ "Postpone fixed bucket writes only support "
+ "bucket-function.type=default, got {}"
+ .format(bucket_function)
+ )
+ self.bucket_keys = table.table_schema.bucket_keys
+ self.bucket_key_indices = self._get_field_indices(self.bucket_keys)
+ self._bucket_key_fields = table.table_schema.logical_bucket_key_fields
+ self._bucket_plan = bucket_plan
+
+ def with_bucket_plan(self, bucket_plan) -> None:
+ self._bucket_plan = bucket_plan
+
+ def num_buckets(self, partition: Tuple) -> int:
+ return self._bucket_plan.num_buckets(partition)
+
+ def extract_partition_bucket_batch(
+ self, data: pa.RecordBatch
+ ) -> Tuple[List[Tuple], List[int]]:
+ partitions = self._extract_partitions_batch(data)
+ columns = [data.column(i) for i in self.bucket_key_indices]
+ buckets = [
+ _bucket_from_hash(
Review Comment:
[P1] Do not route valid bucket-key types through the incomplete `GenericRowSerializer`. This new fixed-postpone path hashes bucket keys via `_binary_row_hash_code()`, whose serializer rejects timezone-aware `TIMESTAMP_LTZ` values (`datetime tzinfo not supported yet`) and has no `VARIANT` encoding (`Unsupported type for serialization: VARIANT`). Both are atomic key types accepted by Java schema validation, so the explicit builder—and Ray’s default fixed-bucket path—cannot route Java-compatible tables using those keys. Please use a Java-compatible BinaryRow encoder for the full supported atomic key surface (especially LTZ instant semantics and VARIANT layout), or explicitly reject these types until supported, and add Java hash fixtures for both.
##########
paimon-python/pypaimon/write/postpone_bucket.py:
##########
@@ -0,0 +1,345 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+from typing import Dict, Tuple
+
+from pypaimon.schema.data_types import (
+ ArrayType,
+ AtomicType,
+ MapType,
+ MultisetType,
+ RowType,
+ VectorType,
+)
+from pypaimon.table.bucket_mode import BucketMode
+from pypaimon.table.row.generic_row import _parse_type_precision_scale
+
+
+def _ceil_div(dividend, divisor):
+ return (dividend + divisor - 1) // divisor
+
+
+def _round_to_word(size):
+ return _ceil_div(size, 8) * 8
+
+
+class _BinaryRowSizeEstimator:
+ """Calculates the Java internal binary size without serializing rows."""
+
+ @classmethod
+ def row_size(cls, values, fields):
+ size = ((len(fields) + 71) // 64) * 8 + len(fields) * 8
+ for value, field in zip(values, fields):
+ size += cls._variable_size(value, field.type)
+ return size
+
+ @classmethod
+ def _variable_size(cls, value, data_type):
+ if isinstance(data_type, AtomicType):
+ return cls._atomic_variable_size(value, data_type)
+ if value is None:
+ return 0
+ if isinstance(data_type, ArrayType):
+ return _round_to_word(cls._array_size(value, data_type.element))
+ if isinstance(data_type, VectorType):
+ return _round_to_word(
+ 4 + data_type.length * cls._primitive_width(data_type.element)
+ )
+ if isinstance(data_type, MapType):
+ keys, values = cls._map_values(value)
+ return _round_to_word(
+ 4
+ + cls._array_size(keys, data_type.key)
+ + cls._array_size(values, data_type.value)
+ )
+ if isinstance(data_type, MultisetType):
+ keys, values = cls._map_values(value)
+ return _round_to_word(
+ 4
+ + cls._array_size(keys, data_type.element)
+ + cls._array_size(values, AtomicType("INT", False))
+ )
+ if isinstance(data_type, RowType):
+ return _round_to_word(
+ cls.row_size(cls._row_values(value, data_type), data_type.fields)
+ )
+ raise ValueError("Unsupported data type: {}".format(data_type))
+
+ @classmethod
+ def _atomic_variable_size(cls, value, data_type):
+ type_name = data_type.type.upper()
+ if type_name.startswith(("DECIMAL", "NUMERIC")):
+ precision, _ = _parse_type_precision_scale(data_type)
+ return 16 if precision > 18 else 0
+ if type_name.startswith("TIMESTAMP"):
+ precision, _ = _parse_type_precision_scale(data_type)
+ return 8 if precision > 3 else 0
+ if value is None:
+ return 0
+ if type_name == "VARIANT":
+ value_bytes, metadata = cls._variant_bytes(value)
+ return _round_to_word(4 + len(value_bytes) + len(metadata))
+ if type_name == "BLOB":
+ value = value if isinstance(value, (bytes, bytearray)) else value.to_data()
+ return cls._binary_size(value)
+ if type_name.startswith(("CHAR", "VARCHAR", "STRING")):
+ return cls._binary_size(str(value).encode("utf-8"))
+ if type_name.startswith(("BINARY", "VARBINARY", "BYTES")):
+ return cls._binary_size(value)
+ return 0
+
+ @classmethod
+ def _array_size(cls, values, element_type):
+ values = list(values)
+ header_size = 4 + ((len(values) + 31) // 32) * 4
+ size = _round_to_word(
+ header_size + len(values) * cls._fixed_width(element_type)
+ )
+ return size + sum(
+ cls._variable_size(value, element_type)
+ for value in values
+ if value is not None
+ )
+
+ @staticmethod
+ def _binary_size(value):
+ length = len(bytes(value))
+ return 0 if length <= 7 else _round_to_word(length)
+
+ @staticmethod
+ def _variant_bytes(value):
+ if isinstance(value, dict):
+ return bytes(value["value"]), bytes(value["metadata"])
+ return bytes(value.value()), bytes(value.metadata())
+
+ @staticmethod
+ def _map_values(value):
+ items = value.items() if isinstance(value, dict) else value
+ items = list(items)
+ return [item[0] for item in items], [item[1] for item in items]
+
+ @staticmethod
+ def _row_values(value, row_type):
+ if isinstance(value, dict):
+ return [value[field.name] for field in row_type.fields]
+ if hasattr(value, "values"):
+ return value.values
+ return list(value)
+
+ @classmethod
+ def _fixed_width(cls, data_type):
+ if isinstance(data_type, AtomicType):
+ type_name = data_type.type.upper()
+ if type_name in ("BOOLEAN", "BOOL", "TINYINT", "BYTE"):
+ return 1
+ if type_name in ("SMALLINT", "SHORT"):
+ return 2
+ if type_name in ("INT", "INTEGER", "FLOAT", "REAL", "DATE", "TIME") \
+ or type_name.startswith("TIME("):
+ return 4
+ return 8
+ if isinstance(data_type, (ArrayType, MapType, MultisetType, RowType)):
Review Comment:
[P2] Treat nested `VectorType` values as variable-width collection elements. `_fixed_width()` handles ARRAY/MAP/MULTISET/ROW slots but omits `VectorType`, even though `_variable_size()` already knows how to size a vector payload. I reproduced a valid postpone table with a non-key `ARRAY<VECTOR<FLOAT, 2>>`: the legacy postpone writer committed it successfully, while the new default size-based planner failed before writing with `Unsupported array element type: VECTOR<FLOAT, 2>`. Please include `VectorType` in the 8-byte offset/length case and add an ARRAY<VECTOR> regression fixture (the same gap affects MAP/MULTISET values containing VECTOR).
##########
paimon-python/pypaimon/write/commit/conflict_detection.py:
##########
@@ -265,6 +269,27 @@ def check_conflicts(
return self.check_row_id_from_snapshot(latest_snapshot, delta_entries)
+ @staticmethod
+ def check_bucket_num_conflicts(entries, commit_kind):
+ if commit_kind == "OVERWRITE":
Review Comment:
[P1] Validate bucket-count consistency in overwrite output. Returning early for every `OVERWRITE` also skips validation between the new files in the final merged state. I reproduced the existing multi-worker mismatch scenario with an overwrite commit: one writer produced `total_buckets=1`, another `total_buckets=3`, and the commit succeeded with both values in the same partition. The next `PostponeBucketPlanner` then failed with “different total buckets”, blocking subsequent fixed-bucket writes; partition statistics also retain whichever value is processed last. Since old files have already been canceled by their DELETE entries in `merged_entries`, please validate the remaining positive ADD counts for overwrite as well (or validate new overwrite additions separately) while still allowing a legitimate old→new rescale.
--
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: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [python] Write postpone batches to fixed buckets [paimon]
Posted by "XiaoHongbo-Hope (via GitHub)" <gi...@apache.org>.
XiaoHongbo-Hope commented on code in PR #8985:
URL: https://github.com/apache/paimon/pull/8985#discussion_r3704135964
##########
paimon-python/pypaimon/write/row_key_extractor.py:
##########
@@ -590,3 +590,59 @@ def _extract_buckets_batch(self, data: pa.RecordBatch) -> List[int]:
def _extract_bucket_row(self, values_by_name: Dict[str, Any]) -> int:
return BucketMode.POSTPONE_BUCKET.value
+
+
+class PostponeFixedBucketRowKeyExtractor(RowKeyExtractor):
+ """Route postpone batches using a resolved bucket plan."""
+
+ def __init__(self, table, bucket_plan):
+ super().__init__(table.table_schema)
+ if table.options.bucket() != BucketMode.POSTPONE_BUCKET.value:
+ raise ValueError(
+ "Postpone fixed bucket writes require bucket = -2, got {}".format(
+ table.options.bucket()
+ )
+ )
+ self.bucket_keys = table.table_schema.bucket_keys
+ self.bucket_key_indices = self._get_field_indices(self.bucket_keys)
+ self._bucket_key_fields = table.table_schema.logical_bucket_key_fields
+ self._bucket_plan = bucket_plan
+
+ def with_bucket_plan(self, bucket_plan) -> None:
+ self._bucket_plan = bucket_plan
+
+ def num_buckets(self, partition: Tuple) -> int:
+ return self._bucket_plan.num_buckets(partition)
+
+ def extract_partition_bucket_batch(
+ self, data: pa.RecordBatch
+ ) -> Tuple[List[Tuple], List[int]]:
+ partitions = self._extract_partitions_batch(data)
+ columns = [data.column(i) for i in self.bucket_key_indices]
+ buckets = [
+ _bucket_from_hash(
Review Comment:
> [P1] Respect the configured bucket function. This extractor always applies the default BinaryRow hash, while Java's `PostponeFixBucketProcessor` constructs `BucketFunction` from `CoreOptions` and therefore supports `default`, `mod`, and `hive`. For a valid `bucket-function.type=mod` or `hive` table, Python and Java can route the same primary key to different buckets; Java bucket pruning then follows the configured function and may miss rows written by Python. Please dispatch according to `bucket-function.type` (with Java-compatible semantics), or reject non-default functions until they are supported, and add cross-engine fixtures for all supported functions.
Fixed
--
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: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [python] Write postpone batches to fixed buckets [paimon]
Posted by "XiaoHongbo-Hope (via GitHub)" <gi...@apache.org>.
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: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [python] Write postpone batches to fixed buckets [paimon]
Posted by "XiaoHongbo-Hope (via GitHub)" <gi...@apache.org>.
XiaoHongbo-Hope commented on PR #8985:
URL: https://github.com/apache/paimon/pull/8985#issuecomment-5161995604
> Is it possible to provide a higher-level API to accomplish this?
Sure, done
--
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: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [python] Write postpone batches to fixed buckets [paimon]
Posted by "XiaoHongbo-Hope (via GitHub)" <gi...@apache.org>.
XiaoHongbo-Hope commented on PR #8985:
URL: https://github.com/apache/paimon/pull/8985#issuecomment-5154799933
> Thanks for the update. I re-tested the two original issues: multiple `write_arrow_batch` calls now produce the same bucket plan as the concatenated input, and `postpone.target-row-num-per-bucket` now correctly takes precedence. The Daft legacy fallback also looks reasonable.
>
> I found one remaining API consistency issue: `BatchTableWrite.write_row` still bypasses the new postpone planning path. It immediately calls `extract_partition_bucket_row`, and `PostponeFixedBucketRowKeyExtractor.num_buckets` defaults an unknown partition to one bucket.
>
> With `postpone.target-row-num-per-bucket=1`, `postpone.batch-write-fixed-bucket.max-parallelism=8`, and 8 rows in one new partition, I get:
>
> ```
> write_arrow: {8}
> write_row: {1}
> ```
>
> This means the bucket count depends on which public write API is used. Calling `write_row` first also records the new partition as one bucket, so subsequent Arrow batches cannot re-plan it. Please either buffer and plan row writes together with Arrow inputs, or explicitly reject `write_row` for postpone fixed-bucket batch writes, and add a regression test.
>
> One additional resource concern: `_postpone_batches` retains all unknown-partition `RecordBatch` objects until `prepare_commit`, so the chunked API now has memory usage proportional to the complete input and cannot benefit from normal incremental file rolling. Spark can persist/spill and Ray can use object-store spilling; the local Python path has no equivalent. Please at least document this memory bound, or consider temporary spilling or an explicit preplanned bucket map.
Thanks for the follow-up. write_row() now joins the deferred bucket planning with Arrow inputs. Added pure-row and mixed-input tests, and documented the local buffering memory bound. Existing-bucket partitions still write incrementally. 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: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org