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 5d902e8e04 [Python] Use the common native global index builder (#10487)
5d902e8e04 is described below
commit 5d902e8e047f4e2e8fb1341647d52e21bb12761b
Author: Jingsong Lee <[email protected]>
AuthorDate: Sat Oct 10 16:12:54 2026 +0800
[Python] Use the common native global index builder (#10487)
---
.../pypaimon/globalindex/create_global_index.py | 6 +-
.../pypaimon/globalindex/native_index_build.py | 26 ++-
.../tests/native_generic_index_build_test.py | 207 +++++++++++++++++++++
.../tests/native_sorted_index_build_test.py | 8 +-
4 files changed, 232 insertions(+), 15 deletions(-)
diff --git a/paimon-python/pypaimon/globalindex/create_global_index.py
b/paimon-python/pypaimon/globalindex/create_global_index.py
index ec7fc24488..4ff50f8507 100644
--- a/paimon-python/pypaimon/globalindex/create_global_index.py
+++ b/paimon-python/pypaimon/globalindex/create_global_index.py
@@ -174,9 +174,9 @@ class GlobalIndexBuilder:
concurrency, ray_remote_args = validate_build_options(
self, concurrency, ray_remote_args)
partition_filter = self._resolve_partition_filter()
- if execution == "local" and self._index_type in
_SORTED_INDEX_IDENTIFIERS:
- from pypaimon.globalindex.native_index_build import
build_native_sorted_index
- messages = build_native_sorted_index(self, partition_filter)
+ if execution == "local":
+ from pypaimon.globalindex.native_index_build import
build_native_global_index
+ messages = build_native_global_index(self, partition_filter)
if messages is not None:
return messages
read_builder = self._table.new_read_builder()
diff --git a/paimon-python/pypaimon/globalindex/native_index_build.py
b/paimon-python/pypaimon/globalindex/native_index_build.py
index 266ebac6ee..48ecf421e3 100644
--- a/paimon-python/pypaimon/globalindex/native_index_build.py
+++ b/paimon-python/pypaimon/globalindex/native_index_build.py
@@ -15,21 +15,23 @@
# specific language governing permissions and limitations
# under the License.
-"""Local sorted-index builds delegated to Rust, retaining Python's commit
API."""
+"""Local global-index builds delegated to Rust, retaining Python's commit
API."""
from pypaimon.read.native_plan import (
_native_table, _option_value_to_string, _predicate_to_native,
native_method_available)
from pypaimon.write.native_commit import _rest_catalog_supported,
from_native_commit_messages
-def build_native_sorted_index(builder, partition_filter):
+def build_native_global_index(builder, partition_filter):
"""Return unpublished messages, or None before building for an ineligible
table.
- Rust owns snapshot selection, range planning, sorting and file generation.
+ Rust owns snapshot selection, range planning and index file generation.
Build errors propagate: a failed native build must not start a second
build.
"""
table = builder._table
- if (builder._index_type not in ('btree', 'bitmap')
+ # The Rust vector pipeline requires dense, non-null vectors. Keep Python
+ # vector construction until it supports Java's sparse relative row IDs.
+ if (builder._index_type not in ('btree', 'bitmap', 'full-text')
or not table.options.native_write_enabled()
or not table.options.data_evolution_enabled()
or not table.options.global_index_enabled()
@@ -37,13 +39,21 @@ def build_native_sorted_index(builder, partition_filter):
or table.options.query_auth_enabled
or table.is_primary_key_table
or table.options.file_format() != 'parquet'
- or not native_method_available('Table',
'new_sorted_global_index_build_builder')
+ or not native_method_available('Table',
'new_global_index_build_builder')
or not _rest_catalog_supported(table)):
return None
- native = _native_table(table).new_sorted_global_index_build_builder()
+ from pypaimon_rust.datafusion import GlobalIndexBuildBuilder
+ if not GlobalIndexBuildBuilder.supports_index_type(builder._index_type):
+ return None
+ native = _native_table(table).new_global_index_build_builder()
native.with_index_column(builder._index_columns[0]).with_index_type(builder._index_type)
- native.with_options({str(key): _option_value_to_string(value)
- for key, value in builder._user_options.items() if
value is not None})
+ options = {str(key): _option_value_to_string(value)
+ for key, value in builder._user_options.items() if value is not
None}
+ if builder._index_type == 'full-text':
+ from
pypaimon.globalindex.full_text.native_full_text_global_index_reader import
NativeFullTextIndexOptions
+ options.update({'full-text.' + key: value for key, value in
+
NativeFullTextIndexOptions.from_options(builder._options.to_map()).to_native_options().items()})
+ native.with_options(options)
if partition_filter is not None:
native.with_partition_filter(_predicate_to_native(partition_filter))
return from_native_commit_messages(table, native.build())
diff --git a/paimon-python/pypaimon/tests/native_generic_index_build_test.py
b/paimon-python/pypaimon/tests/native_generic_index_build_test.py
new file mode 100644
index 0000000000..b700ae53c5
--- /dev/null
+++ b/paimon-python/pypaimon/tests/native_generic_index_build_test.py
@@ -0,0 +1,207 @@
+# 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.
+
+"""The common Rust index builder prepares full-text and vector indexes over
REST."""
+
+import struct
+from unittest.mock import patch
+
+import pyarrow as pa
+import pytest
+
+from pypaimon.globalindex.create_global_index import GlobalIndexBuilder
+from pypaimon.read.native_plan import _native_table, _predicate_to_native,
native_method_available
+from pypaimon.tests import native_sorted_index_build_test as index_tests
+from pypaimon.write.native_commit import from_native_commit_messages
+
+rest_catalog = index_tests.rest_catalog
+pytestmark = [pytest.mark.native_plan, pytest.mark.skipif(
+ not native_method_available('Table', 'new_global_index_build_builder'),
+ reason='Common Rust global index builder required')]
+
+KINDS = ['full-text', 'ivf-flat', 'ivf-pq', 'ivf-sq', 'ivf-rq', 'diskann']
+SCHEMA = pa.schema([('id', pa.int32()), ('name', pa.string()),
+ ('embedding', pa.list_(pa.float32(), 2)), ('pt',
pa.int32())])
+ROWS = [{'id': i, 'name': 'paimon lake' if i % 2 == 0 else None,
+ 'embedding': [float(i % 4), float(i // 4)], 'pt': i // 16} for i in
range(32)]
+
+
+def _options(kind, rows_per_shard=100):
+ options = {'global-index.row-count-per-shard': str(rows_per_shard)}
+ if kind != 'full-text':
+ options[kind + '.dimension'] = '2'
+ if kind != 'diskann':
+ options[kind + '.nlist'] = '1'
+ if kind == 'ivf-pq':
+ options['ivf-pq.pq.m'] = '1'
+ if kind == 'diskann':
+ options['diskann.pq.bits'] = '4'
+ return options
+
+
+def _table(catalog, options=None):
+ settings = {'vector-index.search-mode': 'fast',
'full-text-index.search-mode': 'fast'}
+ settings.update(options or {})
+ table = index_tests._create(catalog, schema=SCHEMA, partitioned=True,
options=settings)
+ index_tests._append(table, ROWS, schema=SCHEMA)
+ return table
+
+
+def _builder(table, kind, rows_per_shard=100):
+ return (_native_table(table).new_global_index_build_builder()
+ .with_index_column('name' if kind == 'full-text' else 'embedding')
+ .with_index_type(' ' + kind.upper() + ' ')
+ .with_options(_options(kind, rows_per_shard)))
+
+
+def _build(table, builder):
+ return from_native_commit_messages(table, builder.build())
+
+
[email protected]('kind', KINDS)
[email protected]('native_commit', [False, True])
+def test_prepare_partition_then_commit_and_query(rest_catalog, kind,
native_commit):
+ catalog, _ = rest_catalog
+ table = _table(catalog, {'commit.native.enabled':
str(native_commit).lower()})
+ predicate = table.new_read_builder().new_predicate_builder()
+ builder = _builder(table, kind, rows_per_shard=4)
+
builder.with_partition_filter(_predicate_to_native(predicate.greater_or_equal('pt',
0)))
+ builder.with_partition_filter(_predicate_to_native(predicate.equal('pt',
1)))
+ before = table.snapshot_manager().get_latest_snapshot().id
+ messages = _build(table, builder)
+ files = index_tests._message_files(messages)
+ assert table.snapshot_manager().get_latest_snapshot().id == before
+ assert not index_tests._files(table)
+ assert len(files) == 4
+ assert {message.partition for message in messages} == {(1,)}
+ assert {file.index_type for file in files} == {kind}
+ assert sum(file.row_count for file in files) == 16
+ assert all(struct.unpack('>iiq', file.global_index_meta.source_meta)
+ == (0x44454958, 1, before) for file in files)
+ index_tests._commit(table, messages)
+ assert table.snapshot_manager().get_latest_snapshot().id == before + 1
+ assert _build(table, builder) == []
+ classic = table.copy({'read.native.enabled': 'false',
'scan.native-plan.enabled': 'false'})
+ if kind == 'full-text':
+ result = (classic.new_full_text_search_builder().with_query('name',
'{"match":{"query":"paimon"}}')
+ .with_partition_filter(predicate.equal('pt',
1)).with_limit(64).execute_local())
+ assert sorted(result.results()) == list(range(16, 32, 2))
+ else:
+ result =
(classic.new_vector_search_builder().with_vector_column('embedding')
+ .with_query_vector(ROWS[16]['embedding']).with_limit(1)
+ .with_partition_filter(predicate.equal('pt',
1)).execute_local())
+ assert len(list(result.results())) == 1
+ assert 16 <= list(result.results())[0] < 32
+
+
[email protected]('kind', KINDS)
+def test_all_families_use_external_index_placement(rest_catalog, kind,
tmp_path):
+ catalog, _ = rest_catalog
+ external = (tmp_path / 'indexes').as_uri()
+ table = _table(catalog, {'global-index.external-path': external})
+ messages = _build(table, _builder(table, kind))
+ assert messages
+ for file in index_tests._message_files(messages):
+ assert file.external_path.startswith(external + '/')
+ assert table.file_io.exists(file.external_path)
+ index_tests._commit(table, messages)
+ assert len(index_tests._files(table)) == 2
+
+
[email protected]('kind', ['full-text', 'ivf-flat'])
[email protected]('native_commit', [False, True])
+def
test_indexed_column_changes_reject_messages_and_preserve_files(rest_catalog,
kind, native_commit):
+ catalog, _ = rest_catalog
+ table = _table(catalog, {'commit.native.enabled':
str(native_commit).lower()})
+ messages = _build(table, _builder(table, kind))
+ column = 'name' if kind == 'full-text' else 'embedding'
+ changed = 'changed' if kind == 'full-text' else [99., 99.]
+ data = pa.table({'_ROW_ID': pa.array([0], pa.int64()),
+ column: pa.array([changed], SCHEMA.field(column).type)})
+ update =
table.new_batch_write_builder().new_update().update_by_arrow_with_row_id(data)
+ index_tests._commit(table, update)
+ before = table.snapshot_manager().get_latest_snapshot().id
+ with pytest.raises(Exception, match='Global index source conflict'):
+ index_tests._commit(table, messages)
+ assert table.snapshot_manager().get_latest_snapshot().id == before
+ assert all(table.file_io.exists(index_tests._path(table, file))
+ for file in index_tests._message_files(messages))
+
+
+def test_python_full_text_build_delegates_to_common_native_entry(rest_catalog):
+ catalog, _ = rest_catalog
+ table = _table(catalog)
+ predicate = table.new_read_builder().new_predicate_builder().equal('pt', 1)
+ with patch.object(GlobalIndexBuilder, '_build_generic_index',
+ side_effect=AssertionError('Python index build ran')):
+ messages = GlobalIndexBuilder(table, 'name', index_type='full-text',
+ partition_filter=predicate,
options=_options('full-text', 4)).build()
+ assert len(index_tests._message_files(messages)) == 4
+ assert {message.partition for message in messages} == {(1,)}
+ index_tests._commit(table, messages)
+
+
+def test_python_nullable_vector_build_keeps_sparse_row_ids(rest_catalog):
+ catalog, _ = rest_catalog
+ schema = pa.schema([('id', pa.int32()), ('name', pa.string()),
+ ('embedding', pa.list_(pa.float32())), ('pt',
pa.int32())])
+ table = index_tests._create(catalog, schema=schema)
+ rows = [dict(ROWS[0]), dict(ROWS[1]), dict(ROWS[2])]
+ rows[1]['embedding'] = None
+ index_tests._append(table, rows, schema=schema)
+ with patch.object(GlobalIndexBuilder, '_build_generic_index',
+ wraps=GlobalIndexBuilder(table, 'embedding',
index_type='ivf-flat',
+
options=_options('ivf-flat'))._build_generic_index) as build:
+ messages = GlobalIndexBuilder(table, 'embedding',
index_type='ivf-flat',
+ options=_options('ivf-flat')).build()
+ assert build.call_count == 1
+ assert sum(file.row_count for file in
index_tests._message_files(messages)) == 3
+ index_tests._commit(table, messages)
+
+
[email protected]('option_source', ['build', 'table'])
[email protected]('stop_words', [['paimon', 'apache'], 'paimon;apache'])
+def test_native_full_text_options_match_python_list_encoding(rest_catalog,
option_source, stop_words):
+ import json
+ from pypaimon.globalindex.full_text.native_full_text_global_index_reader
import NativeFullTextIndexOptions
+
+ catalog, _ = rest_catalog
+ options = {'full-text.stop-words': stop_words, 'full-text.with-position':
True}
+ table_options = {'full-text.' + key: value for key, value in
+
NativeFullTextIndexOptions.from_options(options).to_native_options().items()}
+ table = _table(catalog, table_options if option_source == 'table' else
None)
+ messages = GlobalIndexBuilder(table, 'name', index_type='full-text',
+ options=options if option_source == 'build'
else {}).build()
+ native_meta =
[json.loads(file.global_index_meta.index_meta.decode('utf-8'))
+ for file in index_tests._message_files(messages)]
+ assert all(meta['stop-words'] == 'paimon;apache' and meta['with-position']
== 'true'
+ for meta in native_meta)
+ index_tests._commit(table, messages)
+ query = '{"match":{"query":"paimon"}}'
+ result = table.new_full_text_search_builder().with_query('name',
query).with_limit(64).execute_local()
+ assert not list(result.results())
+ # Rebuild through Python's original writer and compare serialized options
and results.
+ classic = table.copy({'write.native.enabled': 'false'})
+ classic.drop_global_index('name', index_type='full-text')
+ messages = GlobalIndexBuilder(classic, 'name', index_type='full-text',
+ options=options if option_source == 'build'
else {}).build()
+ classic_meta =
[NativeFullTextIndexOptions.deserialize(file.global_index_meta.index_meta).to_native_options()
+ for file in index_tests._message_files(messages)]
+ assert classic_meta == native_meta
+ index_tests._commit(classic, messages)
+ result = classic.new_full_text_search_builder().with_query('name',
query).with_limit(64).execute_local()
+ assert not list(result.results())
diff --git a/paimon-python/pypaimon/tests/native_sorted_index_build_test.py
b/paimon-python/pypaimon/tests/native_sorted_index_build_test.py
index b56ec60992..b2849ee60e 100644
--- a/paimon-python/pypaimon/tests/native_sorted_index_build_test.py
+++ b/paimon-python/pypaimon/tests/native_sorted_index_build_test.py
@@ -38,7 +38,7 @@ from pypaimon.write.table_commit import BatchTableCommit
rest_catalog = native_plan_rest_test.rest_catalog
pytestmark = [pytest.mark.native_plan, pytest.mark.skipif(
- not native_method_available('Table',
'new_sorted_global_index_build_builder'),
+ not native_method_available('Table', 'new_global_index_build_builder'),
reason='Rust index builder required')]
SCHEMA = pa.schema([('id', pa.int32()), ('name', pa.string()), ('pt',
pa.int32())])
@@ -258,7 +258,7 @@ def
test_binding_partition_conjunction_and_rejection_are_core_semantics(rest_cat
table = _create(catalog, partitioned=True)
_append(table, ROWS)
predicates = table.new_read_builder().new_predicate_builder()
- native =
_native_table(table).new_sorted_global_index_build_builder().with_index_column('id').with_index_type(kind)
+ native =
_native_table(table).new_global_index_build_builder().with_index_column('id').with_index_type(kind)
# Python partition predicates carry partition-row indices; bindings
resolve by field name.
native.with_partition_filter(_predicate_to_native(predicates.equal('pt',
0)))
native.with_partition_filter(_predicate_to_native(predicates.equal('pt',
1)))
@@ -275,7 +275,7 @@ def test_empty_table_build_is_a_noop(rest_catalog, kind):
table = _create(catalog)
assert _build(table, kind) == []
assert table.create_global_index('name', index_type=kind) == 0
- assert
_native_table(table).new_sorted_global_index_build_builder().with_index_column('name').execute()
== 0
+ assert
_native_table(table).new_global_index_build_builder().with_index_column('name').execute()
== 0
assert table.snapshot_manager().get_latest_snapshot() is None
@@ -284,7 +284,7 @@ def
test_native_binding_can_build_and_execute_separate_indexes(rest_catalog):
table = _create(catalog)
_append(table, ROWS)
native = _native_table(table)
- builder =
native.new_sorted_global_index_build_builder().with_index_column('name')
+ builder = native.new_global_index_build_builder().with_index_column('name')
prepared = builder.build()
assert all(message.serialize() for message in prepared)
_commit(table, from_native_commit_messages(table, prepared))
