[python] Preserve native index build option sources and typed parallelism

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

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()

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

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)
 
 

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))

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() == []