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 62cd57fcee [Python] Enable native row-tracking sequence reads and
predicates (#10408)
62cd57fcee is described below
commit 62cd57fceec2f9318f254051984487502caf5244
Author: Jingsong Lee <[email protected]>
AuthorDate: Wed Oct 7 09:33:56 2026 +0800
[Python] Enable native row-tracking sequence reads and predicates (#10408)
---
.../pypaimon/read/reader/data_file_batch_reader.py | 5 +-
paimon-python/pypaimon/read/split_read.py | 12 +-
paimon-python/pypaimon/read/table_read.py | 16 --
.../tests/native_data_evolution_commit_test.py | 5 +-
paimon-python/pypaimon/tests/native_read_test.py | 21 ---
.../pypaimon/tests/native_row_sidecar_test.py | 13 +-
.../pypaimon/tests/native_system_columns_test.py | 197 +++++++++++++++++++++
7 files changed, 216 insertions(+), 53 deletions(-)
diff --git a/paimon-python/pypaimon/read/reader/data_file_batch_reader.py
b/paimon-python/pypaimon/read/reader/data_file_batch_reader.py
index 4952692b2c..b50842299f 100644
--- a/paimon-python/pypaimon/read/reader/data_file_batch_reader.py
+++ b/paimon-python/pypaimon/read/reader/data_file_batch_reader.py
@@ -469,8 +469,9 @@ class DataFileBatchReader(RecordBatchReader):
# Handle _SEQUENCE_NUMBER field
if SpecialFields.SEQUENCE_NUMBER.name in self.system_fields.keys():
idx = self.system_fields[SpecialFields.SEQUENCE_NUMBER.name]
- # Create a new array that fills with max_sequence_number
- arrays[idx] = pa.repeat(self.max_sequence_number,
record_batch.num_rows)
+ # Java's tracking vector retains physical versions and only fills
+ # NULL entries from the manifest's max sequence number.
+ arrays[idx] = pc.fill_null(arrays[idx], self.max_sequence_number)
names = record_batch.schema.names
table = None
diff --git a/paimon-python/pypaimon/read/split_read.py
b/paimon-python/pypaimon/read/split_read.py
index 80aeb562f1..d0e875426d 100644
--- a/paimon-python/pypaimon/read/split_read.py
+++ b/paimon-python/pypaimon/read/split_read.py
@@ -159,7 +159,7 @@ class SplitRead(ABC):
self.predicate = predicate
self.push_down_predicate = self._push_down_predicate()
self._arrow_filter_pushdown_enabled = predicate_supports_arrow_filter(
- self.push_down_predicate)
+ self.push_down_predicate) and not
self._has_row_tracking_predicate()
self.split = split
self.row_tracking_enabled = row_tracking_enabled
self.value_arity = len(read_type)
@@ -254,6 +254,10 @@ class SplitRead(ABC):
def _push_down_predicate(self) -> Optional[Predicate]:
if self.predicate is None:
return None
+ elif self._has_row_tracking_predicate():
+ # Metadata is assigned after file decoding. Filtering here would
+ # see missing/NULL versions and renumber positional row IDs.
+ return None
elif self.table.is_primary_key_table:
pk_predicate = trim_predicate_by_fields(self.predicate,
self.table.primary_keys)
if not pk_predicate:
@@ -262,6 +266,12 @@ class SplitRead(ABC):
else:
return self.predicate
+ def _has_row_tracking_predicate(self) -> bool:
+ return (self.table.options.row_tracking_enabled()
+ and self.predicate is not None
+ and bool(predicate_field_names(self.predicate) & {
+ SpecialFields.ROW_ID.name,
SpecialFields.SEQUENCE_NUMBER.name}))
+
@abstractmethod
def create_reader(self) -> RecordReader:
"""Create a record reader for the given split."""
diff --git a/paimon-python/pypaimon/read/table_read.py
b/paimon-python/pypaimon/read/table_read.py
index 0c572ef921..551d0300be 100644
--- a/paimon-python/pypaimon/read/table_read.py
+++ b/paimon-python/pypaimon/read/table_read.py
@@ -519,10 +519,6 @@ class TableRead:
if any(isinstance(split, QueryAuthSplit) for split in splits):
return self._native_fallback(
"query authorization requires the Python reader")
- if not self._native_data_evolution_sequence_supported(splits):
- return self._native_fallback(
- "sequence-number projection over partial data-evolution files "
- "requires the Python reader")
try:
from pypaimon.read.native_plan import (
_prepare_native_read, _raise_if_native_fork_safety_error,
@@ -815,18 +811,6 @@ class TableRead:
return False
return True
- def _native_data_evolution_sequence_supported(self, splits):
- from pypaimon.table.special_fields import SpecialFields
-
- if (not self.table.options.data_evolution_enabled()
- or not any(field.id == SpecialFields.SEQUENCE_NUMBER.id for
field in self._scan_read_type)):
- return True
- # Current Rust provider selection cannot synthesize the sequence
- # column from metadata when partial files omit that physical column.
- return all(data_file.write_cols is None
- or SpecialFields.SEQUENCE_NUMBER.name in
data_file.write_cols
- for split in splits for data_file in split.files)
-
def _convert_native_batches(self, batches, schema):
remaining = self.limit
try:
diff --git a/paimon-python/pypaimon/tests/native_data_evolution_commit_test.py
b/paimon-python/pypaimon/tests/native_data_evolution_commit_test.py
index e789ff4ebe..344aa2ff83 100644
--- a/paimon-python/pypaimon/tests/native_data_evolution_commit_test.py
+++ b/paimon-python/pypaimon/tests/native_data_evolution_commit_test.py
@@ -75,10 +75,7 @@ def _read(table, native=False, metadata=False):
'scan.native-plan.enabled': str(native).lower()})
builder = read_table.new_read_builder()
if metadata:
- fields = ['id', 'v', 'w', 'p', '_ROW_ID']
- if not native:
- fields.append('_SEQUENCE_NUMBER')
- builder.with_projection(fields)
+ builder.with_projection(['id', 'v', 'w', 'p', '_ROW_ID',
'_SEQUENCE_NUMBER'])
read = builder.new_read()
with ExitStack() as stack:
if native:
diff --git a/paimon-python/pypaimon/tests/native_read_test.py
b/paimon-python/pypaimon/tests/native_read_test.py
index e01e95ede3..b50c9c15e4 100644
--- a/paimon-python/pypaimon/tests/native_read_test.py
+++ b/paimon-python/pypaimon/tests/native_read_test.py
@@ -89,27 +89,6 @@ def
test_native_read_falls_back_before_opening_primary_file_with_row_sidecar():
native.assert_not_called()
[email protected]('project_sequence', [False, True])
-def
test_native_partial_data_evolution_read_requires_a_sequence_provider(project_sequence):
- from pypaimon.table.special_fields import SpecialFields
-
- read = _table_read()
- read.table.options.data_evolution_enabled.return_value = True
- if project_sequence:
- read._scan_read_type += [SpecialFields.SEQUENCE_NUMBER]
- split = _Split()
- split._native_split = object()
- split.files[0].write_cols = ['id']
- with patch('pypaimon.read.native_plan.native_read',
return_value=[_id_batch([1])]) as native:
- batches = read._try_native_batches([split], pa.schema([('id',
pa.int32())]))
- if project_sequence:
- assert batches is None
- native.assert_not_called()
- else:
- assert [batch.to_pydict() for batch in batches] == [{'id': [1]}]
- native.assert_called_once()
-
-
def test_native_read_returns_named_variant_expression_columns():
read = _table_read()
variants = {
diff --git a/paimon-python/pypaimon/tests/native_row_sidecar_test.py
b/paimon-python/pypaimon/tests/native_row_sidecar_test.py
index 4241aad7e2..510bd6bd05 100644
--- a/paimon-python/pypaimon/tests/native_row_sidecar_test.py
+++ b/paimon-python/pypaimon/tests/native_row_sidecar_test.py
@@ -216,19 +216,14 @@ def
test_sidecars_store_logical_nested_map_and_variant_values(tmp_path, layout):
assert _read(table, projection, row_id).to_pydict() == expected
[email protected]('enabled', [None, 'false', 'true'])
-def test_missing_sidecar_can_fall_back_to_native_parquet(tmp_path, enabled):
- options = {} if enabled is None else {'scan.ignore-lost-files': enabled}
- table = _table(tmp_path, options=options)
+def test_missing_selected_sidecar_reports_its_path(tmp_path):
+ table = _table(tmp_path)
messages = _append(table)
for message in messages:
for file in message.new_files:
os.remove(next(path for path in file.collect_files() if
path.endswith('.row')))
- if enabled == 'true':
- assert _read(table, ['id', 'value'], 5).to_pydict() == {'id': [5],
'value': ['v5']}
- else:
- with pytest.raises(ValueError, match=r'parquet\.row'):
- _read(table, ['id', 'value'], 5)
+ with pytest.raises(ValueError, match=r'parquet\.row'):
+ _read(table, ['id', 'value'], 5)
@pytest.mark.parametrize('enabled', ['false', 'off', 'true', 'on'])
diff --git a/paimon-python/pypaimon/tests/native_system_columns_test.py
b/paimon-python/pypaimon/tests/native_system_columns_test.py
new file mode 100644
index 0000000000..8ba62eab7a
--- /dev/null
+++ b/paimon-python/pypaimon/tests/native_system_columns_test.py
@@ -0,0 +1,197 @@
+# 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.
+
+"""Row-tracking metadata and filters use Java's data-file versions."""
+
+from contextlib import ExitStack
+from unittest.mock import patch
+
+import pyarrow as pa
+import pytest
+
+from pypaimon import CatalogFactory, Schema
+
+pytestmark = pytest.mark.native_plan
+_SCHEMA = pa.schema([('id', pa.int32()), ('v', pa.int32()), ('w', pa.int32())])
+
+
+def _create_table(tmp_path, schema, options):
+ catalog = CatalogFactory.create({'warehouse': str(tmp_path)})
+ catalog.create_database('default', True)
+ catalog.create_table('default.t', Schema.from_pyarrow_schema(schema,
options=options), False)
+ return catalog.get_table('default.t')
+
+
+def _table(tmp_path, evolution, native_write=True, deletion_vectors=False):
+ options = {'row-tracking.enabled': 'true', 'data-evolution.enabled':
str(evolution).lower(),
+ 'write.native.enabled': str(native_write).lower(),
+ 'deletion-vectors.enabled': str(deletion_vectors).lower()}
+ table = _create_table(tmp_path, _SCHEMA, options)
+ for start in (100, 200):
+ builder = table.new_batch_write_builder()
+ writer, commit = builder.new_write(), builder.new_commit()
+ try:
+ writer.write_arrow(pa.table({'id': [start, start + 1], 'v': [10,
20],
+ 'w': [100, 200]}, schema=_SCHEMA))
+ commit.commit(writer.prepare_commit())
+ finally:
+ writer.close()
+ commit.close()
+ return table
+
+
+def _update(table, field, row_id, value):
+ builder = table.new_batch_write_builder()
+ messages = builder.new_update().update_by_arrow_with_row_id(pa.table({
+ '_ROW_ID': pa.array([row_id], pa.int64()), field: pa.array([value],
pa.int32())}))
+ commit = builder.new_commit()
+ try:
+ commit.commit(messages)
+ finally:
+ commit.close()
+
+
+def _read(table, native, projection, predicate=None, limit=None):
+ copy = table.copy({'read.native.enabled': str(native).lower(),
+ 'scan.native-plan.enabled': str(native).lower()})
+ builder = copy.new_read_builder().with_projection(projection)
+ if predicate is not None:
+ builder.with_filter(predicate)
+ if limit is not None:
+ builder.with_limit(limit)
+ reader = builder.new_read()
+ with ExitStack() as stack:
+ if native:
+ stack.enter_context(patch.object(reader, '_create_split_read',
+
side_effect=AssertionError('Python read fallback')))
+ return reader.to_arrow(builder.new_scan().plan().splits(),
parallelism=1)
+
+
[email protected]('native', [False, True])
[email protected]('native_write', [False, True])
[email protected]('deletion_vectors', [False, True])
+def test_partial_column_updates_keep_latest_metadata_provider(tmp_path,
native, native_write, deletion_vectors):
+ table = _table(tmp_path, True, native_write, deletion_vectors)
+ _update(table, 'v', 0, 77)
+ _update(table, 'w', 1, 88)
+ for projection in (['_SEQUENCE_NUMBER'], ['id', '_SEQUENCE_NUMBER'],
+ ['_SEQUENCE_NUMBER', 'w', '_ROW_ID', 'v', 'id']):
+ result = _read(table, native, projection)
+ assert result.column_names == projection
+ assert not result.schema.field('_SEQUENCE_NUMBER').nullable
+ assert result['_SEQUENCE_NUMBER'].null_count == 0
+ if 'id' in projection:
+ result = result.sort_by('id')
+ assert result['id'].to_pylist() == [100, 101, 200, 201]
+ assert result['_SEQUENCE_NUMBER'].to_pylist() == [4, 4, 2, 2]
+ if 'v' in projection:
+ assert result['v'].to_pylist() == [77, 20, 10, 20]
+ assert result['w'].to_pylist() == [100, 88, 100, 200]
+ assert result['_ROW_ID'].to_pylist() == [0, 1, 2, 3]
+ else:
+ assert sorted(result['_SEQUENCE_NUMBER'].to_pylist()) == [2, 2, 4,
4]
+
+
[email protected]('native', [False, True])
[email protected]('evolution', [False, True])
+def test_metadata_predicates_with_unprojected_columns(tmp_path, native,
evolution):
+ table = _table(tmp_path, evolution)
+ if evolution:
+ _update(table, 'v', 0, 77)
+ latest = 3 if evolution else 1
+ pb = table.new_read_builder().with_projection(['id', 'v', '_ROW_ID',
'_SEQUENCE_NUMBER']).new_predicate_builder()
+ cases = [
+ (pb.equal('_SEQUENCE_NUMBER', latest), [100, 101]),
+ (pb.is_null('_SEQUENCE_NUMBER'), []),
+ (pb.is_not_null('_SEQUENCE_NUMBER'), [100, 101, 200, 201]),
+ (pb.and_predicates([pb.equal('_SEQUENCE_NUMBER', latest),
pb.equal('id', 101)]), [101]),
+ (pb.or_predicates([pb.equal('_SEQUENCE_NUMBER', latest),
pb.equal('id', 200)]), [100, 101, 200]),
+ (pb.or_predicates([pb.equal('_SEQUENCE_NUMBER', 2),
pb.equal('_ROW_ID', 0)]), [100, 200, 201]),
+ ]
+ for predicate, expected in cases:
+ result = _read(table, native, ['id'], predicate).sort_by('id')
+ assert result.column_names == ['id']
+ assert result['id'].to_pylist() == expected
+ result = _read(table, native, ['id'], pb.equal('_SEQUENCE_NUMBER',
latest), limit=1)
+ assert result.num_rows == 1
+ assert result['id'][0].as_py() in [100, 101]
+
+
[email protected]('native', [False, True])
+def test_physical_versions_override_manifest_only_when_not_null(tmp_path,
native):
+ from dataclasses import replace
+ from pathlib import Path
+ import pyarrow.parquet as pq
+
+ table = _create_table(tmp_path, _SCHEMA, {
+ 'row-tracking.enabled': 'true', 'data-evolution.enabled': 'true',
+ 'write.native.enabled': 'true'})
+ builder = table.new_batch_write_builder()
+ writer, commit = builder.new_write(), builder.new_commit()
+ try:
+ data = pa.table({'id': [100, 101], 'v': [10, 20], 'w': [100, 200]},
schema=_SCHEMA)
+ writer.write_arrow(data)
+ messages = writer.prepare_commit()
+ file = messages[0].new_files[0]
+ data = data.append_column('_SEQUENCE_NUMBER', pa.array([7, None],
pa.int64()))
+ pq.write_table(data, file.file_path)
+ messages[0].new_files[0] = replace(file,
file_size=Path(file.file_path).stat().st_size)
+ commit.commit(messages)
+ finally:
+ writer.close()
+ commit.close()
+ result = _read(table, native, ['id', '_SEQUENCE_NUMBER']).sort_by('id')
+ assert result['_SEQUENCE_NUMBER'].to_pylist() == [7, 1]
+ pb = table.new_read_builder().with_projection(['id',
'_SEQUENCE_NUMBER']).new_predicate_builder()
+ assert _read(table, native, ['id'], pb.equal('_SEQUENCE_NUMBER',
7))['id'].to_pylist() == [100]
+ assert _read(table, native, ['id'], pb.equal('_SEQUENCE_NUMBER',
1))['id'].to_pylist() == [101]
+
+
[email protected]('native', [False, True])
+def test_case_sensitive_user_column_does_not_shadow_metadata(tmp_path, native):
+ schema = pa.schema([('id', pa.int32()), ('_sequence_number', pa.int32())])
+ table = _create_table(tmp_path, schema, {
+ 'row-tracking.enabled': 'true', 'data-evolution.enabled': 'true',
+ 'write.native.enabled': 'true'})
+ builder = table.new_batch_write_builder()
+ writer, commit = builder.new_write(), builder.new_commit()
+ try:
+ writer.write_arrow(pa.table({'id': [100, 101], '_sequence_number':
[40, 50]}, schema=schema))
+ commit.commit(writer.prepare_commit())
+ finally:
+ writer.close()
+ commit.close()
+ projection = ['id', '_sequence_number', '_SEQUENCE_NUMBER']
+ pb =
table.new_read_builder().with_projection(projection).new_predicate_builder()
+ predicate = pb.and_predicates([pb.equal('_SEQUENCE_NUMBER', 1),
pb.equal('_sequence_number', 50)])
+ assert _read(table, native, projection, predicate).to_pydict() == {
+ 'id': [101], '_sequence_number': [50], '_SEQUENCE_NUMBER': [1]}
+
+
[email protected]('user_field,metadata', [
+ ('_sequence_number', '_SEQUENCE_NUMBER'), ('_row_id', '_ROW_ID')])
+def test_insensitive_metadata_name_collision_is_ambiguous(tmp_path,
user_field, metadata):
+ from pypaimon.write.native_commit import create_native_write_table
+
+ table = _create_table(tmp_path, pa.schema([(user_field, pa.int32())]), {
+ 'row-tracking.enabled': 'true', 'write.native.enabled': 'true'})
+ reader =
create_native_write_table(table).new_read_builder().with_case_sensitive(False)
+ for name in (metadata, user_field):
+ with pytest.raises(ValueError, match='Ambiguous'):
+ reader.with_projection([name]).new_read().read_arrow([])
+ with pytest.raises(ValueError, match='Ambiguous'):
+ reader.with_filter({'method': 'equal', 'field': name, 'literals':
[1]})
