This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/paimon.git
The following commit(s) were added to refs/heads/master by this push:
new 1aa02a9294 [python] Use native plan and read for BLOB LIMIT scans
(#10047)
1aa02a9294 is described below
commit 1aa02a9294995fbc0039dab69a78d88e17923412
Author: Jingsong Lee <[email protected]>
AuthorDate: Mon Sep 21 16:11:36 2026 +0800
[python] Use native plan and read for BLOB LIMIT scans (#10047)
---
paimon-python/pypaimon/read/table_read.py | 26 ++++++-
.../pypaimon/tests/native_plan_integration_test.py | 85 ++++++++++++++++------
paimon-python/pypaimon/tests/native_read_test.py | 77 ++++++++++++++++++++
3 files changed, 164 insertions(+), 24 deletions(-)
diff --git a/paimon-python/pypaimon/read/table_read.py
b/paimon-python/pypaimon/read/table_read.py
index 0215d8eec5..b7b1181381 100644
--- a/paimon-python/pypaimon/read/table_read.py
+++ b/paimon-python/pypaimon/read/table_read.py
@@ -420,7 +420,8 @@ class TableRead:
return None
if not splits:
return []
- if self._deferred_blob_limit_may_prune(splits):
+ if (self._deferred_blob_limit_may_prune(splits)
+ and not self._native_pruning_blob_limit_supported()):
return None
# Query authorization has additional filtering, masking and projection
# semantics which are already implemented by the Python reader.
@@ -923,6 +924,29 @@ class TableRead:
or self._native_inline_blob_fields())
and not self._limit_covers_all_splits(splits))
+ def _native_pruning_blob_limit_supported(self) -> bool:
+ """Rust caps DE batches before payload resolution when no post-filter
is needed.
+
+ A predicate on a managed BLOB or BLOB view can require payload I/O
+ before the output quota is known. Inline descriptors can still use
+ native reads if the predicate only references ordinary columns.
+ """
+ if not self.table.options.data_evolution_enabled():
+ return False
+ read_names = {field.name for field in self._scan_read_type}
+ if self.table.options.blob_view_fields() & read_names:
+ return False
+ if self.predicate is not None:
+ # Managed BLOBs are decoded by the physical file reader, before a
+ # residual filter. Inline descriptors are resolved later, so a
+ # predicate on ordinary columns can safely select rows first.
+ if self._deferred_blob_fields:
+ return False
+ from pypaimon.read.push_down_utils import predicate_field_names
+ if predicate_field_names(self.predicate) &
self._native_inline_blob_fields():
+ return False
+ return True
+
def _native_inline_blob_fields(self) -> set:
"""Return configured BLOB fields that native reads resolve eagerly."""
options = self.table.options
diff --git a/paimon-python/pypaimon/tests/native_plan_integration_test.py
b/paimon-python/pypaimon/tests/native_plan_integration_test.py
index 9581cead18..a053ed5627 100644
--- a/paimon-python/pypaimon/tests/native_plan_integration_test.py
+++ b/paimon-python/pypaimon/tests/native_plan_integration_test.py
@@ -37,7 +37,7 @@ from pypaimon.read.native_plan import (
native_split_from_python,
)
from pypaimon.schema.schema_change import SchemaChange
-from pypaimon.table.row.blob import BlobDescriptor, BlobRef
+from pypaimon.table.row.blob import BlobDescriptor
from pypaimon.utils.range import Range
@@ -1009,25 +1009,23 @@ class NativePlanIntegrationTest(unittest.TestCase):
table.new_batch_write_builder().new_commit().commit(write.prepare_commit())
write.close()
- native_table = table.copy({'read.native.enabled': 'true'})
+ native_table = table.copy({
+ 'scan.native-plan.enabled': 'true', 'read.native.enabled': 'true'})
builder = native_table.new_read_builder().with_limit(1)
plan = builder.new_scan().plan()
- fetched = []
- original_to_data = BlobRef.to_data
-
- def tracked_to_data(blob):
- fetched.append(blob)
- return original_to_data(blob)
+ self.assertTrue(all(getattr(split, '_native_split', None) is not None
+ for split in plan.splits()))
- with patch.object(BlobRef, 'to_data', tracked_to_data), patch(
+ with patch(
'pypaimon.read.native_plan.native_read',
- return_value=[]) as native:
+ wraps=native_read) as native, patch(
+ 'pypaimon.read.table_read.TableRead._create_split_read',
+ side_effect=AssertionError('Python reader used')):
result = builder.new_read().to_arrow(
plan.splits(), parallelism=1)
- native.assert_not_called()
+ native.assert_called_once()
self.assertEqual(result.to_pydict(), {'id': [1], 'payload': [b'a']})
- self.assertEqual(len(fetched), 1)
@unittest.skipUnless(native_reader_available(),
"pypaimon-rust native reader API not installed")
@@ -1065,25 +1063,66 @@ class NativePlanIntegrationTest(unittest.TestCase):
write.prepare_commit())
write.close()
- native_table = table.copy({'read.native.enabled': 'true'})
+ native_table = table.copy({
+ 'scan.native-plan.enabled': 'true', 'read.native.enabled':
'true'})
builder = native_table.new_read_builder().with_limit(1)
plan = builder.new_scan().plan()
- fetched = []
- original_to_data = BlobRef.to_data
-
- def tracked_to_data(blob):
- fetched.append(blob)
- return original_to_data(blob)
+ self.assertTrue(all(getattr(split, '_native_split', None) is not
None
+ for split in plan.splits()))
- with patch.object(BlobRef, 'to_data', tracked_to_data), patch(
+ with patch(
'pypaimon.read.native_plan.native_read',
- return_value=[]) as native:
+ wraps=native_read) as native, patch(
+ 'pypaimon.read.table_read.TableRead._create_split_read',
+ side_effect=AssertionError('Python reader used')):
result = builder.new_read().to_arrow(plan.splits())
- native.assert_not_called()
+ native.assert_called_once()
self.assertEqual(
result.to_pydict(), {'id': [1], 'payload': [b'first']})
- self.assertEqual(len(fetched), 1)
+
+ @unittest.skipUnless(native_reader_available(),
+ "pypaimon-rust native reader API not installed")
+ def test_native_read_limit_filters_before_descriptor_payload_io(self):
+ schema = pa.schema([('id', pa.int32()), ('payload',
pa.large_binary())])
+ with tempfile.TemporaryDirectory() as payload_dir:
+ selected_path = os.path.join(payload_dir, 'selected')
+ with open(selected_path, 'wb') as output:
+ output.write(b'selected')
+ descriptors = [
+ BlobDescriptor('file://' + os.path.join(payload_dir,
'missing'), 0, 7).serialize(),
+ BlobDescriptor('file://' + selected_path, 0, 8).serialize(),
+ ]
+ self.cat.create_table('default.native_descriptor_filter_limit_t',
+ Schema.from_pyarrow_schema(schema, options={
+ 'row-tracking.enabled': 'true',
+ 'data-evolution.enabled': 'true',
+ 'blob-descriptor-field': 'payload',
+ }), False)
+ table =
self.cat.get_table('default.native_descriptor_filter_limit_t')
+ write = table.new_batch_write_builder().new_write()
+ write.write_arrow(pa.Table.from_pydict({
+ 'id': [1, 2], 'payload': descriptors,
+ }, schema=schema))
+
table.new_batch_write_builder().new_commit().commit(write.prepare_commit())
+ write.close()
+
+ native_table = table.copy({
+ 'scan.native-plan.enabled': 'true', 'read.native.enabled':
'true'})
+ builder = native_table.new_read_builder().with_limit(1)
+ builder.with_filter(builder.new_predicate_builder().equal('id', 2))
+ plan = builder.new_scan().plan()
+ self.assertTrue(all(getattr(split, '_native_split', None) is not
None
+ for split in plan.splits()))
+ with patch('pypaimon.read.native_plan.native_read',
+ wraps=native_read) as native:
+ with
patch('pypaimon.read.table_read.TableRead._create_split_read',
+ side_effect=AssertionError('Python reader used')):
+ result = builder.new_read().to_arrow(plan.splits())
+
+ native.assert_called_once()
+ self.assertEqual(result.to_pydict(),
+ {'id': [2], 'payload': [b'selected']})
@unittest.skipUnless(native_reader_available(),
"pypaimon-rust native reader API not installed")
diff --git a/paimon-python/pypaimon/tests/native_read_test.py
b/paimon-python/pypaimon/tests/native_read_test.py
index 841324086e..c9649e441f 100644
--- a/paimon-python/pypaimon/tests/native_read_test.py
+++ b/paimon-python/pypaimon/tests/native_read_test.py
@@ -39,6 +39,7 @@ def _table_read(limit=None):
read.table.options.blob_as_descriptor.return_value = False
read.table.options.blob_descriptor_fields.return_value = set()
read.table.options.blob_view_fields.return_value = set()
+ read.table.options.data_evolution_enabled.return_value = False
read.predicate = None
read.read_type = [DataField(0, 'id', AtomicType('INT'))]
read.include_row_kind = False
@@ -47,6 +48,7 @@ def _table_read(limit=None):
read._read_parallelism = 1
read._deferred_blob_fields = set()
read._predicate_extra_fields = []
+ read._scan_read_type = read.read_type
read._output_column_names = ['id']
return read
@@ -55,6 +57,7 @@ def _blob_table_read(limit=None):
read = _table_read(limit)
read.read_type = [DataField(0, 'payload', AtomicType('BLOB'))]
read._output_column_names = ['payload']
+ read._scan_read_type = read.read_type
return read
@@ -798,6 +801,80 @@ def
test_native_read_defers_to_python_for_pruning_descriptor_blob_limit():
native.assert_not_called()
[email protected]('descriptor', [False, True])
+def test_native_read_pruning_blob_limit_uses_data_evolution_reader(descriptor):
+ read = _blob_table_read(limit=1)
+ read.table.options.data_evolution_enabled.return_value = True
+ if descriptor:
+ read.table.options.blob_descriptor_fields.return_value = {'payload'}
+ else:
+ read._deferred_blob_fields = {'payload'}
+ split = _Split('payload.parquet' if descriptor else 'payload.blob')
+ split._native_split = object()
+ split.merged_row_count = Mock(return_value=2)
+ batch = pa.record_batch(
+ [pa.array([b'first'], type=pa.large_binary())], names=['payload'])
+
+ with patch('pypaimon.read.native_plan.native_read',
+ return_value=[batch]) as native:
+ actual = list(read._try_native_batches(
+ [split], pa.schema([('payload', pa.large_binary())]),
+ blob_parallelism=1))
+
+ assert actual == [batch]
+ native.assert_called_once()
+
+
+def test_native_read_pruning_blob_limit_keeps_predicate_and_view_fallback():
+ read = _blob_table_read(limit=1)
+ read.table.options.data_evolution_enabled.return_value = True
+ read._deferred_blob_fields = {'payload'}
+ split = _Split('payload.blob')
+ split._native_split = object()
+ split.merged_row_count = Mock(return_value=2)
+
+ with patch('pypaimon.read.native_plan.native_read') as native:
+ read.predicate = Mock()
+ assert read._try_native_batches(
+ [split], pa.schema([('payload', pa.large_binary())])) is None
+ read.predicate = None
+ read.table.options.blob_view_fields.return_value = {'payload'}
+ assert read._try_native_batches(
+ [split], pa.schema([('payload', pa.large_binary())])) is None
+ read.table.options.blob_view_fields.return_value = set()
+ read.table.options.blob_descriptor_fields.return_value = {'payload'}
+ read._deferred_blob_fields = set()
+ read.predicate = Mock()
+ with patch('pypaimon.read.push_down_utils.predicate_field_names',
+ return_value={'payload'}):
+ assert read._try_native_batches(
+ [split], pa.schema([('payload', pa.large_binary())])) is None
+
+ native.assert_not_called()
+
+
+def test_native_read_pruning_descriptor_limit_allows_non_blob_predicate():
+ read = _blob_table_read(limit=1)
+ read.table.options.data_evolution_enabled.return_value = True
+ read.table.options.blob_descriptor_fields.return_value = {'payload'}
+ read.predicate = Mock()
+ split = _Split('payload.parquet')
+ split._native_split = object()
+ split.merged_row_count = Mock(return_value=2)
+ batch = pa.record_batch(
+ [pa.array([b'selected'], type=pa.large_binary())], names=['payload'])
+
+ with patch('pypaimon.read.push_down_utils.predicate_field_names',
+ return_value={'id'}):
+ with patch('pypaimon.read.native_plan.native_read',
+ return_value=[batch]) as native:
+ actual = list(read._try_native_batches(
+ [split], pa.schema([('payload', pa.large_binary())])))
+
+ assert actual == [batch]
+ native.assert_called_once()
+
+
@pytest.mark.parametrize('data_type, values', [
(pa.timestamp('s'), [0, 1]),
(pa.timestamp('s', tz='UTC'), [0, 1]),
