Stephen0421 commented on code in PR #10570:
URL: https://github.com/apache/paimon/pull/10570#discussion_r4290774724
##########
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:
Fixed. Shared managed packs are no longer registered in `extra_files`; only
the exclusively owned `.blobref` sidecar is registered. Python abort discovers
pack candidates through the sidecars and checks surviving references before
deletion.
I also verified the cross-engine path using a Python v14 CommitMessage and
the public Java abort API: aborting the uncommitted file preserves the shared
packs, and the committed rows remain readable.
##########
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:
Fixed. When managed-payload filtering is required, the limit is no longer
pushed into the split reader. Filtering now happens before outer limit
accounting, including the parallel read quota.
I reproduced the reported case: `equal("payload", b"other").with_limit(1)`
now returns id=2 through both Arrow and iterator APIs. Added coverage for
parallel reads and projected-out payload columns as well.
##########
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:
Fixed by reusing `FilterRecordBatchReader`, which preserves all batches
produced by the filter instead of taking only the first. The compatibility
fallback also combines chunks correctly when `pyarrow.concat_batches` is
unavailable.
The regression test uses the reported 140,000 matching rows with
`read.batch-size=200000` and verifies that all 140,000 rows and IDs are
returned.
--
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]
