JingsongLi commented on code in PR #10570:
URL: https://github.com/apache/paimon/pull/10570#discussion_r4284919389
##########
paimon-python/pypaimon/write/writer/key_value_data_writer.py:
##########
@@ -116,6 +162,33 @@ def close(self):
finally:
self._buffer.reset()
+ def _write_data_to_file(self, data: pa.Table):
+ if self._blob_externalizer is not None:
+ data = self._blob_externalizer.externalize_table(data)
+ try:
+ super()._write_data_to_file(data)
+ except Exception:
+ if self._blob_externalizer is not None:
+ self._blob_externalizer.discard_pending_hand_off()
+ raise
+ if self._blob_externalizer is not None:
+ self._blob_externalizer.hand_off_pending()
+
+ def _additional_extra_files(self, file_path: str, data: pa.Table) ->
List[str]:
+ if not self._managed_blob_fields:
+ return []
+ collector = ManagedBlobReferenceCollector(
+ self.file_io, file_path, self.table.fields,
self._managed_blob_fields)
+ try:
+ collector.collect_table(data)
+ names = [collector.close()]
+ if self._blob_externalizer is not None:
+ names[0:0] =
self._blob_externalizer.take_referenced_pack_names()
Review Comment:
[P1] Keep shared packs out of exclusively owned extra_files
Registering the same pack on several DataFileMeta.extra_files makes it
subject to Java cleanup, which still treats every extra file as exclusively
owned. On this exact head I wrote three rolled Python PK data files sharing
three managed packs, committed the first two, serialized only the third
(uncommitted) CommitMessage with v14, and called the public Java
table.newCommit(...).abort([third]). FileStoreCommitImpl.abort ->
DataFileMeta.collectFiles deleted all three shared packs. Snapshot 1 and its
two committed files remained, but a subsequent public Python read failed with
missing-pack. The Python-only abort guard cannot protect this path. Please
either avoid publishing shared packs in generic extra_files (retain them
through .blobref ownership), or make every consuming engine's cleanup honor
shared references before introducing this metadata.
##########
paimon-python/pypaimon/read/table_read.py:
##########
@@ -886,6 +889,8 @@ def _arrow_batch_generator(self, splits: List[Split],
schema: pyarrow.Schema,
batch = batch.slice(0, remaining)
batch = self._serialize_blob_views_as_descriptors(
batch, reader)
+ batch = self._materialize_managed_blob_batch(batch)
+ batch = self._filter_resolved_managed_blob_batch(batch)
Review Comment:
[P2] Apply the row limit after resolved-payload filtering
The split reader receives remaining as its limit and the batch is also
sliced before this newly added post-filter. With a real committed PK table
containing (1, b"visible"), (2, b"other"), native write/read/scan disabled,
equal("payload", b"other").with_limit(1) returns zero rows through both
to_arrow and to_iterator; the expected row is id=2. The first nonmatching row
exhausts the inner limit before its descriptor is resolved and filtered. Please
disable the pre-filter limit when managed payload filtering is required, and
count only matching rows towards the requested limit in all read paths.
##########
paimon-python/pypaimon/read/table_read.py:
##########
@@ -1631,6 +1640,95 @@ def _build_split_read(self, split: Split, read_type=None,
limit=effective_limit,
)
+ def _managed_blob_specs(self):
+ cached = getattr(self, '_managed_blob_specs_cache', None)
+ if cached is not None:
+ return cached
+ specs = managed_blob_output_specs(
+ self.table, self._adapter_read_type, self._blob_as_descriptor)
+ self._managed_blob_specs_cache = specs
+ return specs
+
+ def _materialize_managed_blob_batch(self, batch: pyarrow.RecordBatch) ->
pyarrow.RecordBatch:
+ specs = self._managed_blob_specs()
+ if not specs:
+ return batch
+ file_io = self.table.file_io
+ table_path = self.table.table_path
+ for _index, name, kind in specs:
+ if name not in batch.schema.names:
+ continue
+ values = [
+ materialize_managed_blob(value, kind, file_io, table_path)
+ for value in batch.column(name).to_pylist()
+ ]
+ column_idx = batch.schema.names.index(name)
+ field = batch.schema.field(column_idx)
+ batch = batch.set_column(
+ column_idx, field, pyarrow.array(values, type=field.type))
+ return batch
+
+ def _materialize_managed_blob_values(self, values):
+ specs = self._managed_blob_specs()
+ if not specs:
+ return values
+ converted = list(values)
+ file_io = self.table.file_io
+ table_path = self.table.table_path
+ for index, _name, kind in specs:
+ if index >= len(converted):
+ continue
+ converted[index] = materialize_managed_blob(
+ converted[index], kind, file_io, table_path)
+ return tuple(converted)
+
+ def _managed_blob_post_filter_fields(self):
+ if self.predicate is None:
+ return set()
+ names = {name for _index, name, _kind in self._managed_blob_specs()}
+ if not names:
+ return set()
+ return predicate_field_names(self.predicate) & names
+
+ def _filter_resolved_managed_blob_batch(self, batch):
+ if batch is None or batch.num_rows == 0:
+ return batch
+ if not self._managed_blob_post_filter_fields():
+ return batch
+ expr = self.predicate.to_arrow(batch.schema)
+ if expr is None:
+ return batch
+ import pyarrow.dataset as ds
+ filtered = ds.InMemoryDataset(
+
pyarrow.Table.from_batches([batch])).scanner(filter=expr).to_table()
+ if filtered.num_rows == 0:
+ return batch.slice(0, 0)
+ return filtered.to_batches()[0]
Review Comment:
[P2] Preserve every batch produced by the payload filter
The dataset scanner can split one input batch into several output batches,
so selecting [0] silently discards matching rows. I reproduced this via public
APIs on a real native-disabled PK BLOB table: write 140,000 rows all carrying
b"x", set read.batch-size=200000, then read with equal("payload", b"x"). The
unfiltered read returns 140,000 rows, but the filtered read returns only
131,072 (ids 0..131071), losing 8,928 matches. Please return/iterate the entire
filtered result or combine its chunks into one RecordBatch rather than taking
the first.
--
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]
