feat(table): unify index options and maintain PK/global indexes

bindings/python/src/vector_search.rs

@@ -662,6 +662,8 @@ for factory, expected in (
         raise AssertionError('Unconfigured query was accepted')
 
 single = table.new_vector_search_builder().with_vector_column('embedding').with_query_vector([1., 0.]).with_limit(1)
+for factory in (table.new_vector_search_builder, table.new_batch_vector_search_builder, table.new_hybrid_search_builder):
+    assert not hasattr(factory(), 'with_vector_index_type')
 assert isinstance(single.execute(), sdk.GlobalIndexResult)
 assert len(single.execute()) == 0
 batch = table.new_batch_vector_search_builder().with_vector_column('embedding').with_query_vectors([[1., 0.], [0., 1.]]).with_limit(1)

bindings/python/tests/test_vector_search.py

@@ -52,6 +52,113 @@ def _single(table):
     return table.new_vector_search_builder().with_vector_column('embedding').with_query_vector([1, 0]).with_limit(10)
 
 
+@pytest.mark.parametrize('mode', ['full', 'detail'])
+@pytest.mark.parametrize('batch', [False, True])
+@pytest.mark.parametrize('source', ['table', 'query', 'both'])
+def test_untyped_raw_field_json_metric_and_query_precedence(table, mode, batch, source):
+    import pickle
+    key = 'fields.embedding.index-options'
+    table_options = {'vector-index.search-mode': mode}
+    if source in ('table', 'both'):
+        table_options[key] = '{"metric":"cosine"}'
+    configured = table.copy(table_options)
+    builder = (configured.new_batch_vector_search_builder().with_query_vectors([[1, 0], [1, 0]])
+               if batch else configured.new_vector_search_builder().with_query_vector([1, 0]))
+    query_options = {'opaque.query.option': '{"values":[1,2]}'}
+    if source in ('query', 'both'):
+        query_options[key] = '{"metric":"l2"}' if source == 'both' else '{"metric":"cosine"}'
+    builder.with_options(query_options).with_vector_column('embedding').with_limit(3)
+    plan = pickle.loads(pickle.dumps(builder.new_vector_search_scan().scan()))
+    reader = builder.new_batch_vector_search_read() if batch else builder.new_vector_search_read()
+    results = reader.read_batch_plan(plan) if batch else [reader.read_plan(plan)]
+    expected = [1 / 3, 1, 1] if source == 'both' else [0, 1, 1]
+    for result in results:
+        assert sorted(result.row_ids().values()) == pytest.approx(expected)
+
+
+def test_vector_build_accepts_raw_schema_options_and_field_json(table):
+    options = {
+        'bucket': '-1', 'data-evolution.enabled': 'true', 'row-tracking.enabled': 'true',
+        'global-index.enabled': 'true', 'vector-index.search-mode': 'full',
+        'global-index.ivf-flat.index-columns': 'embedding',
+        'fields.embedding.pk-vector.index.options': 'invalid replaced JSON',
+        'fields.embedding.index-options': '{"dimension":2,"nlist":1,"metric":"cosine"}',
+        'ivf-flat.metric': 'l2', 'opaque.future.option': 'preserved',
+    }
+    configured = table.copy(options)
+    builder = (configured.new_global_index_build_builder().with_index_column('embedding')
+               .with_index_type('ivf-flat').with_options(options))
+    assert builder.plan().shards()
+    assert builder.execute() > 0
+    actual = _single(configured).execute().row_ids()
+    assert sorted(actual.values()) == pytest.approx([0, 1, 1])
+
+
+@pytest.mark.parametrize('kind', ['ivf-flat', 'diskann'])
+def test_vector_build_echoed_schema_retains_source_semantics(table, kind):
+    import pickle
+    configured = table.copy({
+        f'{kind}.dimension': '2', f'{kind}.metric': 'cosine',
+        'nlist': 'ignored table property', 'vindex.build.granule.enabled': 'true',
+    })
+    options = configured.schema().options()
+    builder = (configured.new_global_index_build_builder().with_index_column('embedding')
+               .with_index_type(kind))
+    assert builder.plan().shards()
+    builder.with_options(options)
+    shards = pickle.loads(pickle.dumps(builder.plan())).shards()
+    worker = configured.copy({f'{kind}.dimension': '999'})
+    for shard in shards:
+        assert worker.new_global_index_build_builder().build_shard(shard)
+    assert builder.execute() > 0
+    assert sorted(_single(configured).execute().row_ids().values()) == pytest.approx([0, 1, 1])
+
+
+@pytest.mark.parametrize('shared_json', [False, True, 'aliases'])
+@pytest.mark.parametrize('declared', [False, True])
+def test_full_text_json_encoded_lists_keep_raw_and_indexed_semantics(tmp_path, shared_json, declared):
+    import json
+    ctx = SQLContext()
+    ctx.register_catalog('paimon', {'warehouse': str(tmp_path)})
+    ctx.sql('CREATE SCHEMA paimon.db')
+    ctx.sql("CREATE TABLE paimon.db.texts (id INT, body STRING) WITH ("
+            "'bucket'='-1','row-tracking.enabled'='true','data-evolution.enabled'='true',"
+            "'global-index.enabled'='true','write-only'='true','full-text-index.search-mode'='full')")
+    table = PaimonCatalog({'warehouse': str(tmp_path)}).get_table('db.texts')
+    write = table.new_batch_write_builder()
+    writer, commit = write.new_write(), write.new_commit()
+    try:
+        writer.write_arrow(pa.record_batch([[1, 2], ['paimon data', 'needle']], schema=pa.schema([
+            ('id', pa.int32()), ('body', pa.string())])))
+        commit.commit(writer.prepare_commit())
+    finally:
+        writer.close()
+        commit.close()
+    options = {'global-index.full-text.index-columns': 'body'} if declared else {}
+    if shared_json == 'aliases':
+        options.update({'full-text.stop-words': json.dumps(['paimon']), 'full-text.columns': json.dumps(['body'])})
+    elif shared_json:
+        options['fields.body.index-options'] = json.dumps({'stop-words': ['paimon'], 'text-fields': ['body']})
+    else:
+        options.update({'full-text.stop-words': json.dumps(['paimon']),
+                        'full-text.text-fields': json.dumps(['body'])})
+    table = table.copy(options)
+    search_table = table.copy({'global-index.full-text.index-columns': 'body'})
+
+    def hits(query):
+        return (search_table.new_full_text_search_builder().with_query('body', json.dumps({'match': {'query': query}}))
+                .with_limit(3).execute().row_ids())
+
+    assert len(hits('paimon')) == 0
+    before = hits('data')
+    assert len(before) == 1
+    builder = (table.new_global_index_build_builder().with_index_column('body')
+               .with_index_type('full-text').with_options(table.schema().options()))
+    assert builder.execute() == 1
+    assert len(hits('paimon')) == 0
+    assert hits('data') == before
+
+
 def test_filters_accumulate_and_partition_uses_table_schema(table):
     result = (_single(table).with_filter({'method': 'greaterThan', 'field': 'id', 'literals': [1]})
               .with_filter({'method': 'lessThan', 'field': 'id', 'literals': [4]})
@@ -158,6 +265,68 @@ def test_single_vector_keyword_matches_stub(table):
     assert builder.execute().row_ids() == _single(table).with_limit(1).execute().row_ids()
 
 
+@pytest.mark.parametrize('batch', [False, True])
+def test_raw_options_survive_scan_read_and_use_raw_metric_options(table, batch):
+    import pickle
+    builder = (table.new_batch_vector_search_builder().with_query_vectors([[1, 0], [1, 0]])
+               if batch else table.new_vector_search_builder().with_query_vector([1, 0]))
+    builder.with_options({'global-index.ivf-flat.index-columns': 'embedding',
+                          'fields.embedding.index-options': '{"dimension":2,"nlist":1}',
+                          'opaque.future.option': '{"value":[1,2]}'})
+    builder.with_vector_column('embedding').with_limit(3)
+    plan = pickle.loads(pickle.dumps(builder.new_vector_search_scan().scan()))
+    reader = builder.new_batch_vector_search_read() if batch else builder.new_vector_search_read()
+    # Mutating the builder must not change captured reader configuration.
+    builder.with_options({'global-index.ivf-flat.index-columns': 'other',
+                          'global-index.diskann.index-columns': 'embedding', 'metric': 'l2'})
+    results = reader.read_batch_plan(plan) if batch else [reader.read_plan(plan)]
+    for result in results:
+        assert sorted(result.row_ids().values()) == pytest.approx([0, 1, 1])
+    results = builder.execute() if batch else [builder.execute()]
+    for result in results:
+        assert sorted(result.row_ids().values()) == pytest.approx([1 / 3, 1, 1])
+
+
+def test_options_keep_index_metadata_and_ignore_removed_option_hints(table):
+    expected = _single(table).execute().row_ids()
+    for key in ('index_type', 'index-type', 'vector.index-type', 'fields.embedding.index-type'):
+        assert _single(table.copy({key: 'diskann'})).with_option(key, 'ivf-flat').execute().row_ids() == expected
+    (table.new_global_index_build_builder().with_index_column('embedding')
+     .with_index_type('ivf-flat').with_options({'dimension': '2', 'nlist': '1', 'metric': 'l2'}).execute())
+    expected = _single(table).execute().row_ids()
+    assert (_single(table).with_options({'global-index.invalid-unused-type.index-columns': 'embedding'})
+            .execute().row_ids()) == expected
+    # Table defaults are not explicit query metrics. Both JSON namespaces obey
+    # that boundary when the persisted index uses a different metric.
+    for key, value in (
+        ('ivf-flat.metric', 'cosine'),
+        ('fields.embedding.index-options', '{"metric":"cosine"}'),
+        ('fields.embedding.pk-vector.index.options', '{"metric":"cosine"}'),
+    ):
+        assert (_single(table.copy({key: value, 'global-index.diskann.index-columns': 'embedding'}))
+                .execute().row_ids()) == expected
+        with pytest.raises(Exception, match='does not match index metric'):
+            _single(table).with_option(key, value).execute()
+
+
+def test_hybrid_options_are_captured_with_the_route_builder(table):
+    hybrid = table.new_hybrid_search_builder()
+    hybrid.add_vector_route('embedding', [1, 0], 3, options={
+        'global-index.ivf-flat.index-columns': 'embedding', 'ivf-flat.metric': 'l2'}).with_limit(3)
+    captured = hybrid.route_builders()[0].search_builder()
+    hybrid.add_vector_route('embedding', [1, 0], 3, options={
+        'global-index.diskann.index-columns': 'embedding', 'diskann.metric': 'inner_product'})
+    assert sorted(captured.execute().row_ids().values()) == pytest.approx([1 / 3, 1, 1])
+    updated = hybrid.route_builders()[1].search_builder()
+    assert sorted(updated.execute().row_ids().values()) == pytest.approx([0, 1, 1])
+
+
+def test_builders_have_no_separate_type_setter(table):
+    for builder in (table.new_vector_search_builder(), table.new_batch_vector_search_builder(),
+                    table.new_hybrid_search_builder()):
+        assert not hasattr(builder, 'with_vector_index_type')
+
+
 def test_external_result_uses_same_scan_type_without_search_context(table):
     found = _single(table).with_limit(1).execute()
     external = GlobalIndexResult.from_scored_row_ids(found.row_ids())

crates/paimon/Cargo.toml

@@ -110,7 +110,7 @@ opendal-service-oss = { version = "0.58.0", optional = true }
 opendal-service-s3 = { version = "0.58.0", optional = true }
 pretty_assertions = "1"
 apache-avro = { version = "0.21", features = ["snappy", "zstandard", "bzip", "xz"] }
-indexmap = "2.5.0"
+indexmap = { version = "2.5.0", features = ["serde"] }
 lru = "0.18.2"
 roaring = "0.11"
 datasketches = "0.2"