JingsongLi commented on code in PR #8021:
URL: https://github.com/apache/paimon/pull/8021#discussion_r3339008582
##########
paimon-python/pypaimon/table/row/blob.py:
##########
@@ -382,3 +492,42 @@ def __eq__(self, other) -> bool:
def __hash__(self) -> int:
return hash(self._descriptor)
+
+
+class BlobView(Blob):
Review Comment:
This adds the `BlobView` abstraction, but generic deserialization is still
not wired up: Java `Blob.fromBytes` detects `BlobViewStruct` before
`BlobDescriptor` and returns an unresolved `BlobView`. Python
`Blob.from_bytes()` currently only checks descriptors, so serialized
`BlobViewStruct` bytes fall through to `BlobData`; `row.get_blob()` /
forwarding paths will treat view references as raw bytes instead of BlobView
values. Please add `BlobViewStruct` detection there and cover the
non-`to_arrow()` path in tests.
##########
paimon-python/pypaimon/utils/blob_view_lookup.py:
##########
@@ -0,0 +1,253 @@
+# 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 concurrent.futures import ThreadPoolExecutor, as_completed
+from typing import Dict, List, Tuple
+
+from pypaimon.common.identifier import Identifier
+from pypaimon.common.options.core_options import CoreOptions
+from pypaimon.table.row.blob import BlobDescriptor, BlobViewStruct
+from pypaimon.table.special_fields import SpecialFields
+from pypaimon.utils.range import Range
+
+_PRELOAD_THREAD_NUM = 100
+_MIN_ROWS_PER_TASK = 100
+
+
+class TableReferences:
+ """Groups BlobViewStruct references by upstream table."""
+
+ def __init__(self, identifier: Identifier):
+ self.identifier: Identifier = identifier
+ self.references_by_field: Dict[int, List[BlobViewStruct]] = {}
+ self.row_ids: List[int] = []
+
+ def add(self, view_struct: BlobViewStruct) -> None:
+ self.references_by_field.setdefault(view_struct.field_id,
[]).append(view_struct)
+ self.row_ids.append(int(view_struct.row_id))
+
+
+class TableReadPlan:
+ """A plan for reading blob descriptors from one upstream table."""
+
+ def __init__(self, identifier: Identifier, upstream_table,
+ read_fields: List, row_ranges: List[Range]):
+ self.identifier: Identifier = identifier
+ self.upstream_table = upstream_table
+ self.read_fields: List = read_fields
+ self.row_ranges: List[Range] = row_ranges
+
+
+class BlobViewLookup:
+ """Resolve BlobViewStruct references by reading upstream blob
descriptors."""
+
+ def __init__(self, table):
+ self._table = table
+ self._descriptor_cache: Dict[BlobViewStruct, BlobDescriptor] = {}
+
+ def preload(self, view_structs: List[BlobViewStruct]):
+ if not view_structs:
+ return
+
+ grouped: Dict[str, TableReferences] =
self._group_by_table(view_structs)
+ plans: List[TableReadPlan] = []
+ for table_refs in grouped.values():
+ plans.append(self._create_table_read_plan(table_refs))
+
+ target_rows: int = self._target_rows_per_task(plans)
+ tasks: List[Tuple[TableReadPlan, List[Range]]] = []
+ for plan in plans:
+ for range_chunk in self._split_row_ranges(plan.row_ranges,
target_rows):
+ tasks.append((plan, range_chunk))
+
+ if len(tasks) <= 1:
+ for plan, range_chunk in tasks:
+
self._descriptor_cache.update(self._load_descriptor_chunk(plan, range_chunk))
+ return
+
+ with ThreadPoolExecutor(max_workers=min(_PRELOAD_THREAD_NUM,
len(tasks))) as executor:
+ futures = {
+ executor.submit(self._load_descriptor_chunk, plan,
range_chunk): (plan, range_chunk)
+ for plan, range_chunk in tasks
+ }
+ for future in as_completed(futures):
+ try:
+ self._descriptor_cache.update(future.result())
+ except Exception as exc:
+ # Cancel remaining futures that have not started yet so a
single
+ # failure can abort the rest of the preload work as early
as possible.
+ for pending_future in futures:
+ pending_future.cancel()
+ raise RuntimeError("Failed to preload blob descriptors.")
from exc
+
+ def resolve_descriptor(self, view_struct: BlobViewStruct) ->
BlobDescriptor:
+ descriptor: BlobDescriptor = self._descriptor_cache.get(view_struct)
+ if descriptor is None:
+ raise ValueError(
+ "Cannot resolve BlobViewStruct {} because row id {} was not
found "
+ "in upstream table.".format(view_struct, view_struct.row_id)
+ )
+ return descriptor
+
+ def _group_by_table(
+ self, view_structs: List[BlobViewStruct]
+ ) -> Dict[str, TableReferences]:
+ grouped: Dict[str, TableReferences] = {}
+ for view_struct in view_structs:
+ key = view_struct.identifier.get_full_name()
+ if key not in grouped:
+ grouped[key] = TableReferences(view_struct.identifier)
+ grouped[key].add(view_struct)
+ return grouped
+
+ def _create_table_read_plan(self, table_refs: TableReferences) ->
TableReadPlan:
+ upstream_table = self._load_table(table_refs.identifier)
+
+ fields: List = []
+ for field_id in table_refs.references_by_field:
+ fields.append(self._field_by_id(upstream_table, field_id))
+
+ read_fields = SpecialFields.row_type_with_row_id(fields)
+ return TableReadPlan(
+ table_refs.identifier, upstream_table, read_fields,
+ Range.to_ranges(table_refs.row_ids))
+
+ def _load_descriptor_chunk(
+ self, plan: TableReadPlan, row_ranges: List[Range]
+ ) -> Dict[BlobViewStruct, BlobDescriptor]:
+ identifier: Identifier = plan.identifier
+ upstream_table = plan.upstream_table
+ read_fields = plan.read_fields
+
+ projection_field_names: List[str] = [f.name for f in read_fields]
+
+ descriptor_table =
upstream_table.copy({CoreOptions.BLOB_AS_DESCRIPTOR.key(): "true"})
+ read_builder =
descriptor_table.new_read_builder().with_projection(projection_field_names)
+
+ if SpecialFields.ROW_ID.name not in [
+ data_field.name for data_field in read_builder.read_type()
+ ]:
+ raise ValueError(
+ "Cannot resolve blob view for table {} because row tracking is
not readable."
+ .format(identifier.get_full_name())
+ )
+
+ predicate_builder = read_builder.new_predicate_builder()
+ range_predicates: List = []
+ for r in row_ranges:
+ if r.from_ == r.to:
+ range_predicates.append(
+ predicate_builder.equal(SpecialFields.ROW_ID.name,
r.from_))
+ else:
+ range_predicates.append(
+ predicate_builder.between(SpecialFields.ROW_ID.name,
r.from_, r.to))
+ if len(range_predicates) == 1:
+ predicate = range_predicates[0]
+ else:
+ predicate = predicate_builder.or_predicates(range_predicates)
+ read_builder.with_filter(predicate)
+ result =
read_builder.new_read().to_arrow(read_builder.new_scan().plan().splits())
+
+ if SpecialFields.ROW_ID.name not in result.schema.names:
+ raise ValueError(
+ "Cannot resolve blob view for table {} because row tracking is
not readable."
+ .format(identifier.get_full_name())
+ )
+
+ row_id_values: List =
result.column(SpecialFields.ROW_ID.name).to_pylist()
+ resolved: Dict[BlobViewStruct, BlobDescriptor] = {}
+ for field in read_fields:
+ if field.name == SpecialFields.ROW_ID.name:
+ continue
+ if field.name not in result.schema.names:
+ continue
+ values = result.column(field.name).to_pylist()
+ for row_id, value in zip(row_id_values, values):
+ if value is None:
Review Comment:
Java keeps null upstream blob values as a separate resolved state
(`PreloadedBlobViews.putNull(...)`), and `BlobViewResolvingRow` then returns
null for that view. Here nulls are skipped, so a valid view pointing to a null
upstream blob will later fail in `resolve_descriptor()` as if the row id was
missing. Please track null resolutions separately and make the read path return
null for those views.
##########
paimon-python/pypaimon/read/split_read.py:
##########
@@ -780,6 +780,19 @@ def _push_down_predicate(self) -> Optional[Predicate]:
return None
def create_reader(self) -> RecordReader:
+ reader = self._create_raw_reader()
+
+ if (CoreOptions.blob_view_fields(self.table.options)
Review Comment:
Java has `blob-view.resolve.enabled` (default true) so callers can set it to
false and preserve `BlobViewStruct` references when forwarding blob views to
another table. This Python path resolves whenever `blob-view-field` is
configured, so it loses that mode. Please add the option and gate the
`BlobInlineConvertReader` wrapping on it, matching Java semantics.
--
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]
