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