From 59283cca5797332282cd43bdc63844a0b770a747 Mon Sep 17 00:00:00 2001 From: JingsongLi Date: Sun, 1 Nov 2026 20:03:34 +0800 Subject: [PATCH] [python] Preserve native index build option sources and typed parallelism --- paimon-python/dev/native-index-options.md | 6 ++ .../native/write/native_index_build.py | 26 +++++++-- .../pypaimon/tests/global_index_build_test.py | 56 ++++++++++++++++--- .../tests/native_parallel_index_build_test.py | 15 +++-- .../pypaimon/tests/native_search_plan_test.py | 4 +- .../tests/native_vector_index_build_test.py | 6 +- 6 files changed, 93 insertions(+), 20 deletions(-) diff --git a/paimon-python/dev/native-index-options.md b/paimon-python/dev/native-index-options.md index d9d4808196db..7c9d440d6ccf 100644 --- a/paimon-python/dev/native-index-options.md +++ b/paimon-python/dev/native-index-options.md @@ -30,6 +30,12 @@ strings, and collections to JSON. A None table option represents an unset value. Python preserves effective authorization and write-policy values at the generic table boundary. +Global-index builds retain schema defaults separately from explicit build options, +so an alias in the build request can override a different alias in the schema. +A None build value removes that schema default. Typed build parallelism retains +Python's integer conversion before crossing the string-only ABI; other index +parameters are interpreted in Rust. + ## Resolution contract * The shared resolver returns the selected implementation together with its diff --git a/paimon-python/pypaimon/execution/native/write/native_index_build.py b/paimon-python/pypaimon/execution/native/write/native_index_build.py index d1738b6d03ba..6c8d7f99575a 100644 --- a/paimon-python/pypaimon/execution/native/write/native_index_build.py +++ b/paimon-python/pypaimon/execution/native/write/native_index_build.py @@ -21,6 +21,8 @@ from pypaimon.execution.native.read.builder import (_predicate_to_native) from pypaimon.execution.native.runtime import (native_runtime_available) from pypaimon.execution.native.write.native_commit import _rest_catalog_supported, from_native_commit_messages +from pypaimon.common.options.core_options import CoreOptions +from pypaimon.common.options.options import Options def build_native_global_index(builder, partition_filter, *, execution="local", concurrency=None, ray_remote_args=None): @@ -43,12 +45,19 @@ def build_native_global_index(builder, partition_filter, *, execution="local", c from pypaimon_rust.datafusion import GlobalIndexBuildBuilder if not GlobalIndexBuildBuilder.supports_index_type(builder._index_type): return None - options = dict(table.options.options.to_map()) - options.update(builder._user_options) - native_table = table.copy_without_time_travel(builder._user_options) + # Keep build parameters separate from schema defaults: alias precedence is + # resolved within each source, then explicit build parameters win. + options = _build_options_to_native(builder._user_options) + table_overrides = {key: None for key, value in builder._user_options.items() if value is None} + parallelism = CoreOptions.GLOBAL_INDEX_BUILD_PARALLELISM.key() + table_values = table.options.options.to_map() + if parallelism not in builder._user_options and isinstance(table_values.get(parallelism), (bool, float)): + table_overrides[parallelism] = _build_options_to_native( + {parallelism: table_values[parallelism]})[parallelism] + native_table = table.copy_without_time_travel(table_overrides) native = _native_table(native_table).new_global_index_build_builder() native.with_index_column(builder._index_columns[0]).with_index_type(builder._index_type) - native.with_options(_options_to_native(options)) + native.with_options(options) if partition_filter is not None: native.with_partition_filter(_predicate_to_native(partition_filter)) if execution == "ray": @@ -60,6 +69,15 @@ def build_native_global_index(builder, partition_filter, *, execution="local", c return from_native_commit_messages(table, native.build()) +def _build_options_to_native(options): + encoded = _options_to_native(options) + key = CoreOptions.GLOBAL_INDEX_BUILD_PARALLELISM.key() + if isinstance(options.get(key), (bool, float)): + # Retain Python's typed integer conversion before the string-only ABI. + encoded[key] = str(int(Options(options).get(CoreOptions.GLOBAL_INDEX_BUILD_PARALLELISM))) + return encoded + + def build_native_index_shard(table, shard): """Execute captured metadata in Core; workers never replan or publish.""" native = _native_table(table).new_global_index_build_builder() diff --git a/paimon-python/pypaimon/tests/global_index_build_test.py b/paimon-python/pypaimon/tests/global_index_build_test.py index ea910c402488..9008e03154b4 100644 --- a/paimon-python/pypaimon/tests/global_index_build_test.py +++ b/paimon-python/pypaimon/tests/global_index_build_test.py @@ -283,7 +283,7 @@ def test_native_build_keeps_table_and_user_options(self): for key, value in dict(values, **user).items(): self.assertEqual(value, actual[key]) - def test_native_build_preserves_captured_view_and_applies_overrides(self): + def test_native_build_preserves_captured_view_and_option_sources(self): from pypaimon.execution.native.write import native_index_build from pypaimon.execution.native.table import _resolved_schema_json from unittest.mock import Mock @@ -309,13 +309,49 @@ def test_native_build_preserves_captured_view_and_applies_overrides(self): self.assertTrue(hasattr(captured, '_read_snapshot')) self.assertIsNone(captured._read_snapshot) schema = json.loads(_resolved_schema_json(captured)) - for actual in (schema['options'], native.with_options.call_args[0][0]): - if metric is None: - self.assertNotIn('ivf-pq.metric', actual) - else: - self.assertEqual(metric, actual['ivf-pq.metric']) + actual = native.with_options.call_args[0][0] + if metric is None: + self.assertNotIn('ivf-pq.metric', schema['options']) + self.assertNotIn('ivf-pq.metric', actual) + else: + self.assertEqual('l2', schema['options']['ivf-pq.metric']) + self.assertEqual({'ivf-pq.metric': metric}, actual) self.assertEqual('l2', table.table_schema.options['ivf-pq.metric']) + def test_native_build_preserves_dimension_override_source(self): + schema = pa.schema([('id', pa.int32()), ('embedding', pa.list_(pa.float32()))]) + table = self._create_table(pa_schema=schema, options=dict(self.table_options, **{ + 'write.native.enabled': 'true', 'ivf-flat.dimension': '4'})) + self._native_build_options(GlobalIndexBuilder( + table, 'embedding', 'ivf-flat', options={'ivf-flat.index.dimension': '2'})) + self.assertEqual('4', self._captured_native_table.table_schema.options['ivf-flat.dimension']) + self.assertNotIn('ivf-flat.index.dimension', self._captured_native_table.table_schema.options) + + def test_native_build_preserves_typed_parallelism(self): + key = 'global-index.build.parallelism' + schema = pa.schema([('id', pa.int32()), ('embedding', pa.list_(pa.float32()))]) + for source in ('table', 'build'): + for value, expected in ((2.0, '2'), (True, '1'), (None, None)): + with self.subTest(source=source, value=value): + values = dict(self.table_options, **{'write.native.enabled': 'true', key: '3'}) + if source == 'table': + values[key] = value + table = self._create_table(pa_schema=schema, options=values) + actual = self._native_build_options(GlobalIndexBuilder( + table, 'embedding', 'ivf-flat', options={key: value} if source == 'build' else {})) + self.assertEqual(expected, actual.get(key)) + + def test_native_build_parallelism_override_ignores_invalid_typed_default(self): + key = 'global-index.build.parallelism' + schema = pa.schema([('id', pa.int32()), ('embedding', pa.list_(pa.float32()))]) + for value in (float('nan'), float('inf')): + with self.subTest(value=value): + table = self._create_table(pa_schema=schema, options=dict( + self.table_options, **{'write.native.enabled': 'true', key: value})) + actual = self._native_build_options(GlobalIndexBuilder( + table, 'embedding', 'ivf-flat', options={key: '2'})) + self.assertEqual('2', actual[key]) + def _native_build_options(self, request): from unittest.mock import Mock from pypaimon.execution.native.write import native_index_build @@ -328,11 +364,15 @@ def _native_build_options(self, request): with patch.dict(sys.modules, {'pypaimon_rust.datafusion': sdk}), \ patch.object(native_index_build, 'native_runtime_available', return_value=True), \ patch.object(native_index_build, '_rest_catalog_supported', return_value=True), \ - patch.object(native_index_build, '_native_table', return_value=proxy), \ + patch.object(native_index_build, '_native_table', return_value=proxy) as bridge, \ patch.object(native_index_build, 'from_native_commit_messages', side_effect=lambda table, messages: messages): self.assertEqual([], request.build()) - return native.with_options.call_args[0][0] + self._captured_native_table = bridge.call_args[0][0] + from pypaimon.execution.native.table import _resolved_schema_json + options = json.loads(_resolved_schema_json(self._captured_native_table))['options'] + options.update(native.with_options.call_args[0][0]) + return options def test_full_text_build_and_raw_fallback_share_field_options(self): from pypaimon.execution.python.read.search.full_text_read import DataEvolutionFullTextRead diff --git a/paimon-python/pypaimon/tests/native_parallel_index_build_test.py b/paimon-python/pypaimon/tests/native_parallel_index_build_test.py index b8dde48cc9de..f5a05cf973e2 100644 --- a/paimon-python/pypaimon/tests/native_parallel_index_build_test.py +++ b/paimon-python/pypaimon/tests/native_parallel_index_build_test.py @@ -133,16 +133,21 @@ def test_parallel_partition_selection_preserves_global_row_ids(rest_catalog, kin @pytest.mark.parametrize('kind', generic.KINDS) -@pytest.mark.parametrize('value,expected', [(2.0, '2'), (True, '1'), (None, '1')]) +@pytest.mark.parametrize('value,expected', [(2.0, '2'), (True, '1'), (None, None)]) def test_typed_build_parallelism_uses_existing_option_conversion(rest_catalog, kind, value, expected): catalog, _ = rest_catalog table = _table(catalog) # Observe the control passed over FFI while retaining the real Rust builder. - native = Mock(wraps=_native_table(table).new_global_index_build_builder()) - proxy = SimpleNamespace(new_global_index_build_builder=lambda: native) - with patch('pypaimon.execution.native.write.native_index_build._native_table', return_value=proxy): + captured = [] + + def native_table(source): + native = Mock(wraps=_native_table(source).new_global_index_build_builder()) + captured.append(native) + return SimpleNamespace(new_global_index_build_builder=lambda: native) + + with patch('pypaimon.execution.native.write.native_index_build._native_table', side_effect=native_table): messages = _build(table, kind, {PARALLELISM: value}) - assert native.with_options.call_args[0][0][PARALLELISM] == expected + assert captured[0].with_options.call_args[0][0].get(PARALLELISM) == expected assert len(indexes._message_files(messages)) == (5 if kind == 'full-text' else 4) diff --git a/paimon-python/pypaimon/tests/native_search_plan_test.py b/paimon-python/pypaimon/tests/native_search_plan_test.py index 0b8471d26401..b6544e6836f3 100644 --- a/paimon-python/pypaimon/tests/native_search_plan_test.py +++ b/paimon-python/pypaimon/tests/native_search_plan_test.py @@ -420,8 +420,8 @@ def test_pk_plan_retains_snapshot_and_sources_after_primary_key_update(source, p schema = pa.schema([pa.field('id', pa.int32(), False), pa.field('embedding', pa.list_(pa.float32(), 4))]) rows = [{'id': 1, 'embedding': [0., 0., 0., 1.]}, {'id': 100, 'embedding': [0., 0., 0., 1.]}] - write = table.copy({'write.native.enabled': 'true', 'pk-vector.index.columns': '', - 'pk-full-text.index.columns': ''}).new_batch_write_builder() + write = table.copy({'write.native.enabled': 'true', 'pk-vector.index.columns': None, + 'pk-full-text.index.columns': None}).new_batch_write_builder() writer, commit = write.new_write(), write.new_commit() try: writer.write_arrow(pa.Table.from_pylist(rows, schema=schema)) diff --git a/paimon-python/pypaimon/tests/native_vector_index_build_test.py b/paimon-python/pypaimon/tests/native_vector_index_build_test.py index 5392d27ad0dc..fcc1c8aa0942 100644 --- a/paimon-python/pypaimon/tests/native_vector_index_build_test.py +++ b/paimon-python/pypaimon/tests/native_vector_index_build_test.py @@ -303,9 +303,13 @@ def test_invalid_build_parallelism_is_rejected_before_writing_files(rest_catalog def test_all_null_build_does_not_instantiate_native_training(rest_catalog, kind): catalog, _ = rest_catalog table = _table(catalog, [None] * 4) - options = {kind + '.dimension': '2', kind + '.distance.metric': 'invalid'} + # Metric resolution precedes scanning; these invalid training parameters + # instead prove that an all-null shard never instantiates a trainer. + options = {kind + '.dimension': '2', kind + '.distance.metric': 'l2'} if kind != 'diskann': options[kind + '.nlist'] = '0' + else: + options['diskann.build-preset'] = 'invalid' assert _build(table, kind, options) == [] classic = table.copy({'write.native.enabled': 'false'}) assert GlobalIndexBuilder(classic, 'embedding', index_type=kind, options=options).build() == []