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 68914a6087 [python] Forward blob-as-descriptor to native readers
(#10112)
68914a6087 is described below
commit 68914a6087fbd7d1b8aba9bb1850ee2a75d5de23
Author: XiaoHongbo <[email protected]>
AuthorDate: Tue Sep 22 21:59:42 2026 +0800
[python] Forward blob-as-descriptor to native readers (#10112)
---
paimon-python/pypaimon/read/native_plan.py | 1 +
.../pypaimon/tests/native_plan_integration_test.py | 62 ++++++++++++++++++++++
paimon-python/pypaimon/tests/native_plan_test.py | 9 ++++
3 files changed, 72 insertions(+)
diff --git a/paimon-python/pypaimon/read/native_plan.py
b/paimon-python/pypaimon/read/native_plan.py
index 0908be7dc3..e457bccb28 100644
--- a/paimon-python/pypaimon/read/native_plan.py
+++ b/paimon-python/pypaimon/read/native_plan.py
@@ -179,6 +179,7 @@ def _read_options(table) -> dict:
}
table_options = table.options.options
for option in (
+ CoreOptions.BLOB_AS_DESCRIPTOR,
CoreOptions.SCAN_VERSION,
CoreOptions.SCAN_SNAPSHOT_ID,
CoreOptions.SCAN_TAG_NAME,
diff --git a/paimon-python/pypaimon/tests/native_plan_integration_test.py
b/paimon-python/pypaimon/tests/native_plan_integration_test.py
index a3caec88f8..5fb207ea4d 100644
--- a/paimon-python/pypaimon/tests/native_plan_integration_test.py
+++ b/paimon-python/pypaimon/tests/native_plan_integration_test.py
@@ -1128,6 +1128,68 @@ class NativePlanIntegrationTest(unittest.TestCase):
[{'id': 10, 'payload': b'from-source'}])
native.assert_not_called()
+ @unittest.skipUnless(native_reader_available(),
+ "pypaimon-rust native reader API not installed")
+ @patch('pypaimon.read.native_plan.native_method_available',
+ side_effect=lambda type_name, method: (
+ False if method in ('from_resolved_schema',
'copy_with_resolved_schema')
+ else native_method_available(type_name, method)))
+ def test_native_read_dynamic_blob_as_descriptor(self, capabilities):
+ # Exercise the catalog path used by Rust versions without resolved
schemas.
+ schema = pa.schema([('id', pa.int32()), ('payload',
pa.large_binary())])
+ self.cat.create_table(
+ 'default.native_dynamic_descriptor',
+ 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_dynamic_descriptor')
+ with tempfile.TemporaryDirectory() as payload_dir:
+ path = os.path.join(payload_dir, 'payload')
+ payload = b'blob content'
+ with open(path, 'wb') as output:
+ output.write(payload)
+ descriptor = BlobDescriptor('file://' + path, 0,
len(payload)).serialize()
+ wb = table.new_batch_write_builder()
+ write, commit = wb.new_write(), wb.new_commit()
+ try:
+ write.write_arrow(pa.Table.from_pydict({
+ 'id': [1], 'payload': [descriptor],
+ }, schema=schema))
+ commit.commit(write.prepare_commit())
+ finally:
+ write.close()
+ commit.close()
+
+ for value in ('true', 'false', True, False):
+ with self.subTest(value=value):
+ builder = table.copy({
+ 'read.native.enabled': 'true',
+ 'blob-as-descriptor': value,
+ }).new_read_builder()
+ splits = builder.new_scan().plan().splits()
+ with patch('pypaimon.read.native_plan.native_read',
+ 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(splits)
+ native.assert_called_once()
+ expected = descriptor if str(value).lower() == 'true' else
payload
+ self.assertEqual(result.to_pydict(), {'id': [1],
'payload': [expected]})
+
+ # Descriptor queries must still work when the payload is
unavailable.
+ os.remove(path)
+ builder = table.copy({
+ 'read.native.enabled': 'true',
+ 'blob-as-descriptor': 'true',
+ }).new_read_builder()
+ splits = builder.new_scan().plan().splits()
+ with patch('pypaimon.read.table_read.TableRead._create_split_read',
+ side_effect=AssertionError('Python reader used')):
+
self.assertEqual(builder.new_read().to_arrow(splits).to_pydict(),
+ {'id': [1], 'payload': [descriptor]})
+
@unittest.skipUnless(native_reader_available(),
"pypaimon-rust native reader API not installed")
def test_native_read_pruning_limit_defers_descriptor_blob_payload_io(self):
diff --git a/paimon-python/pypaimon/tests/native_plan_test.py
b/paimon-python/pypaimon/tests/native_plan_test.py
index 2687d55c99..124a64c4c2 100644
--- a/paimon-python/pypaimon/tests/native_plan_test.py
+++ b/paimon-python/pypaimon/tests/native_plan_test.py
@@ -624,6 +624,15 @@ class NativePlanTest(unittest.TestCase):
with self.assertRaisesRegex(ValueError, 'exact built-in catalog
loader'):
_catalog_options(table)
+ def test_blob_as_descriptor_is_forwarded_to_rust(self):
+ for value in ('true', 'false', True, False):
+ with self.subTest(value=value):
+ table = Mock()
+ table.options = CoreOptions(Options({'blob-as-descriptor':
value}))
+ self.assertEqual(
+ _read_options(table)['blob-as-descriptor'],
+ str(value).lower())
+
def test_predicate_and_time_travel_are_converted_for_rust(self):
predicate = PredicateBuilder.and_predicates([
Predicate('greaterOrEqual', 0, 'k', [10]),
