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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
