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