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