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 778d640423 [python] Use Native planning for primary-key sorted indexes
(#10473)
778d640423 is described below
commit 778d640423a65cf27f8763950f8311ed6778b3d8
Author: Jingsong Lee <[email protected]>
AuthorDate: Fri Oct 9 19:55:28 2026 +0800
[python] Use Native planning for primary-key sorted indexes (#10473)
---
paimon-python/pypaimon/read/table_scan.py | 13 +-
.../pypaimon/tests/native_pk_sorted_index_test.py | 178 +++++++++++++++++++++
.../pypaimon/tests/native_plan_expanded_test.py | 9 +-
3 files changed, 185 insertions(+), 15 deletions(-)
diff --git a/paimon-python/pypaimon/read/table_scan.py
b/paimon-python/pypaimon/read/table_scan.py
index 879072a12f..eb9c06d7ef 100755
--- a/paimon-python/pypaimon/read/table_scan.py
+++ b/paimon-python/pypaimon/read/table_scan.py
@@ -210,11 +210,9 @@ class TableScan:
self.file_scanner.partition_key_predicate,
) if predicate is not None
])
- # Append positions and Python sorted-index refinement still precede
- # LIMIT. Ordinary PK shards are selected inside Rust before LIMIT.
- defer_limit = has_distribution and (
- not self.table.is_primary_key_table
- or self.table.options.global_index_enabled())
+ # Append positions still precede LIMIT. Rust owns PK shards and
+ # source-backed sorted-index refinement before applying LIMIT.
+ defer_limit = has_distribution and not
self.table.is_primary_key_table
plan = native_plan(
self.table,
predicate=native_predicate,
@@ -240,11 +238,6 @@ class TableScan:
splits = [s for s in splits
if getattr(s, 'partition', None) is None
or partition_predicate.test(s.partition)]
- if (self.table.is_primary_key_table and not fs.is_streaming and
self.predicate is not None
- and self.table.options.global_index_enabled()
- and plan.snapshot_id is not None):
- snapshot =
self.table.snapshot_manager().get_snapshot_by_id(plan.snapshot_id)
- splits = fs._apply_primary_key_sorted_indexes(splits, snapshot)
if chunk_shuffle is None and defer_limit:
# A partial IndexedSplit plus a file-wide DV cardinality cannot
# reveal how many deleted rows lie inside the selected range.
diff --git a/paimon-python/pypaimon/tests/native_pk_sorted_index_test.py
b/paimon-python/pypaimon/tests/native_pk_sorted_index_test.py
new file mode 100644
index 0000000000..9932dc20e6
--- /dev/null
+++ b/paimon-python/pypaimon/tests/native_pk_sorted_index_test.py
@@ -0,0 +1,178 @@
+# 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.
+
+"""Java source-backed scalar indexes through the REST Native planner."""
+
+import shutil
+from pathlib import Path
+from unittest.mock import patch
+
+import pyarrow as pa
+import pytest
+
+from pypaimon import Schema
+from pypaimon.globalindex.indexed_split import IndexedSplit
+from pypaimon.read.native_plan import native_plan, native_runtime_available
+from pypaimon.tests import native_plan_rest_test,
primary_key_global_index_golden_test
+
+rest_catalog = native_plan_rest_test.rest_catalog
+golden_catalog = primary_key_global_index_golden_test.catalog
+
+pytestmark = [pytest.mark.native_plan, pytest.mark.skipif(
+ not native_runtime_available(), reason='Rust main required')]
+
+
[email protected]
+def indexed_table(rest_catalog, golden_catalog):
+ catalog, server = rest_catalog
+ golden = golden_catalog.get_table('default.test_pk_global_index_golden')
+ schema = golden.table_schema
+ catalog.create_table('default.indexed', Schema(
+ schema.fields, schema.partition_keys, schema.primary_keys,
schema.options), False)
+ source = Path(golden.table_path)
+ destination = Path(server.data_path) / server.warehouse / 'default' /
'indexed'
+ for directory in ('bucket-0', 'index', 'manifest', 'snapshot'):
+ shutil.copytree(str(source / directory), str(destination / directory))
+ return catalog.get_table('default.indexed')
+
+
[email protected]('column,value,ids', [
+ ('name', 'name-4', [4]), ('category', 'odd', [1, 3, 5]),
+ ('name', 'absent', []),
+])
+def test_java_scalar_indexes_are_planned_in_rust(indexed_table, column, value,
ids):
+ table = indexed_table.copy({'scan.native-plan.enabled': 'true',
'read.native.enabled': 'true'})
+ predicate = table.new_read_builder().new_predicate_builder().equal(column,
value)
+ plan = native_plan(table, predicate=predicate)
+ assert plan.snapshot_id == 6
+ # Java's fixture also contains an uncovered APPEND file. Keep it as an
+ # ordinary split; only the compacted source can use index positions.
+ indexed = [split for split in plan.splits() if isinstance(split,
IndexedSplit)]
+ assert len(indexed) == (1 if ids else 0)
+ assert all(not split.raw_convertible for split in indexed)
+ assert all(file.file_source != 1 for split in plan.splits()
+ if not isinstance(split, IndexedSplit) for file in split.files)
+ with patch('pypaimon.index.index_file_handler.IndexFileHandler.scan',
+ side_effect=AssertionError('Python index planning must not
run')):
+ builder = table.new_read_builder().with_filter(predicate)
+ scan_plan = builder.new_scan().plan()
+ result = builder.new_read().to_arrow(scan_plan.splits())
+ assert sorted(result.column('id').to_pylist()) == ids
+
+
[email protected]('mode,ids', [('and', [3]), ('or', [1, 3, 4, 5]),
('unsupported-or', [1, 2])])
+def test_native_compound_predicates_keep_residuals(indexed_table, mode, ids):
+ from pypaimon.common.predicate_builder import PredicateBuilder
+ table = indexed_table.copy({'scan.native-plan.enabled': 'true',
'read.native.enabled': 'true'})
+ pb = table.new_read_builder().new_predicate_builder()
+ if mode == 'and':
+ predicate = PredicateBuilder.and_predicates([pb.equal('name',
'name-3'), pb.equal('category', 'odd')])
+ elif mode == 'or':
+ predicate = PredicateBuilder.or_predicates([pb.equal('name',
'name-4'), pb.equal('category', 'odd')])
+ else:
+ predicate = PredicateBuilder.or_predicates([pb.equal('name',
'name-1'), pb.equal('id', 2)])
+ with patch('pypaimon.index.index_file_handler.IndexFileHandler.scan',
+ side_effect=AssertionError('Python index planning must not
run')):
+ builder = table.new_read_builder().with_filter(predicate)
+ result =
builder.new_read().to_arrow(builder.new_scan().plan().splits())
+ assert sorted(result.column('id').to_pylist()) == ids
+
+
[email protected]('version', [2, 4, 6])
+def test_native_index_uses_selected_snapshot(indexed_table, version):
+ base = indexed_table.copy({'scan.version': str(version),
'read.native.enabled': 'true'})
+ predicate =
base.new_read_builder().new_predicate_builder().equal('category', 'odd')
+ classic = base.copy({'scan.native-plan.enabled': 'false',
'read.native.enabled': 'false'})
+ classic = classic.new_read_builder().with_filter(predicate)
+ expected =
classic.new_read().to_arrow(classic.new_scan().plan().splits()).to_pylist()
+ with patch('pypaimon.index.index_file_handler.IndexFileHandler.scan',
+ side_effect=AssertionError('Python index planning must not
run')):
+ builder = base.copy({'scan.native-plan.enabled':
'true'}).new_read_builder().with_filter(predicate)
+ plan = builder.new_scan().plan()
+ rows = builder.new_read().to_arrow(plan.splits()).to_pylist()
+ assert plan.snapshot_id == version
+ assert sorted(rows, key=lambda row: row['id']) == sorted(expected,
key=lambda row: row['id'])
+
+
[email protected]('index_type,column,value,ids', [
+ ('btree', 'name', 'name-4', [4]), ('bitmap', 'category', 'odd', [1, 3, 5]),
+])
+def test_native_index_read_failure_keeps_data(indexed_table, index_type,
column, value, ids):
+ from pypaimon.index.index_file_handler import IndexFileHandler
+ table = indexed_table.copy({'scan.native-plan.enabled': 'true',
'read.native.enabled': 'true'})
+ entries =
IndexFileHandler(table).scan(table.snapshot_manager().get_latest_snapshot())
+ entry = next(entry for entry in entries if entry.index_file.index_type ==
index_type)
+ table.file_io.delete(table.table_path + '/index/' +
entry.index_file.file_name)
+ predicate = table.new_read_builder().new_predicate_builder().equal(column,
value)
+ with patch('pypaimon.index.index_file_handler.IndexFileHandler.scan',
+ side_effect=AssertionError('Python index planning must not
run')):
+ builder = table.new_read_builder().with_filter(predicate)
+ plan = builder.new_scan().plan()
+ assert all(not isinstance(split, IndexedSplit) for split in
plan.splits())
+ rows = builder.new_read().to_arrow(plan.splits())
+ assert sorted(rows.column('id').to_pylist()) == ids
+
+
+def
test_native_pk_index_positions_are_file_local_with_row_tracking(indexed_table):
+ from copy import deepcopy
+ table = indexed_table.copy({'read.native.enabled': 'true'})
+ predicate = table.new_read_builder().new_predicate_builder().equal('name',
'name-4')
+ plan = native_plan(table, predicate=predicate)
+ splits = deepcopy(plan.splits())
+ for split in splits:
+ for file in split.files:
+ file.first_row_id = 100
+ builder = table.new_read_builder().with_filter(predicate)
+ assert builder.new_read().to_arrow(splits).column('id').to_pylist() == [4]
+
+
[email protected]('disabled', [False, True], ids=['indexed',
'index-disabled'])
+def test_native_sorted_index_limit_and_shards(indexed_table, disabled):
+ table = indexed_table.copy({'scan.native-plan.enabled': 'true',
'read.native.enabled': 'true',
+ 'global-index.enabled': str(not
disabled).lower()})
+ predicate =
table.new_read_builder().new_predicate_builder().equal('category', 'odd')
+ ids = []
+ with patch('pypaimon.index.index_file_handler.IndexFileHandler.scan',
+ side_effect=AssertionError('Python index planning must not
run')):
+ for shard in range(2):
+ builder =
table.new_read_builder().with_filter(predicate).with_limit(1)
+ plan = builder.new_scan().with_shard(shard, 2).plan()
+ if disabled:
+ assert all(not isinstance(split, IndexedSplit) for split in
plan.splits())
+ rows = builder.new_read().to_arrow(plan.splits())
+ assert rows.num_rows <= 1
+ ids.extend(rows.column('id').to_pylist())
+ assert len(ids) == 1
+ assert ids[0] in [1, 3, 5]
+
+
+def test_native_sorted_index_does_not_resurrect_an_old_version(indexed_table):
+ from pypaimon.tests.native_plan_expanded_test import write
+ table = indexed_table.copy({
+ 'deletion-vectors.enabled': 'false', 'write-only': 'true',
+ 'write.native.enabled': 'false', 'commit.native.enabled': 'false',
+ 'scan.native-plan.enabled': 'true', 'read.native.enabled': 'true',
+ })
+ write(table, [{'id': 4, 'name': 'changed', 'category': 'even'}],
+ pa.schema([('id', pa.int32()), ('name', pa.string()), ('category',
pa.string())]))
+ predicate = table.new_read_builder().new_predicate_builder().equal('name',
'name-4')
+ with patch('pypaimon.index.index_file_handler.IndexFileHandler.scan',
+ side_effect=AssertionError('Python index planning must not
run')):
+ builder = table.new_read_builder().with_filter(predicate)
+ plan = builder.new_scan().plan()
+ assert any(not split.raw_convertible and len(split.files) > 1 for
split in plan.splits())
+ assert builder.new_read().to_arrow(plan.splits()).num_rows == 0
diff --git a/paimon-python/pypaimon/tests/native_plan_expanded_test.py
b/paimon-python/pypaimon/tests/native_plan_expanded_test.py
index 244ab1cfa1..f2bd2b7b7f 100644
--- a/paimon-python/pypaimon/tests/native_plan_expanded_test.py
+++ b/paimon-python/pypaimon/tests/native_plan_expanded_test.py
@@ -199,7 +199,8 @@ def
test_scored_de_preserves_score_mapping_and_selection(catalog, native, dv):
@pytest.mark.parametrize('covered', [True, False])
-def test_pk_sorted_index_refines_native_splits_at_selected_snapshot(catalog,
native, covered):
[email protected]_plan
+def test_python_pk_sorted_index_refines_splits_at_selected_snapshot(catalog,
covered):
from pypaimon.globalindex.global_index_meta import GlobalIndexMeta
from pypaimon.globalindex.indexed_split import IndexedSplit
from pypaimon.index.index_file_meta import IndexFileMeta
@@ -234,19 +235,17 @@ def
test_pk_sorted_index_refines_native_splits_at_selected_snapshot(catalog, nat
with patch('pypaimon.index.index_file_handler.IndexFileHandler.scan',
return_value=[entry]) as scan_index, \
patch('pypaimon.table.source.primary_key_sorted_index_scan.reader_factory',
return_value=lambda *args: _Reader([Range(1, 1)])):
- plan, rows = read(table.copy({'scan.version': '1'}), native, predicate)
+ plan, rows = read(table.copy({'scan.version': '1'}), False, predicate)
assert rows == [{'id': 2, 'v': 20}]
assert plan.snapshot_id == 1
assert scan_index.call_args[0][0].id == 1
assert any(isinstance(split, IndexedSplit) for split in plan.splits()) ==
covered
- if native:
- assert all(split.snapshot_id == 1 for split in plan.splits())
# The latest snapshot includes a newer version. Index pruning must not
# resurrect the old matching version from a merge-required split.
with patch('pypaimon.index.index_file_handler.IndexFileHandler.scan',
return_value=[entry]), \
patch('pypaimon.table.source.primary_key_sorted_index_scan.reader_factory',
return_value=lambda *args: _Reader([Range(1, 1)])):
- assert read(table, native, predicate)[1] == []
+ assert read(table, False, predicate)[1] == []
def test_scored_result_requires_a_score_for_each_selected_row(catalog, native):
