[core] Enable multiple vector index columns on primary-key tables

docs/docs/primary-key-table/global-index.mdx

@@ -104,7 +104,8 @@ Vector additionally requires:
 - `deletion-vectors.enabled = true`, except for `first-row`, where it must be `false`.
 - The configured ANN implementation on every writer and reader classpath.
 
-Exactly one vector column is currently supported per table.
+Multiple vector columns are supported. Each column owns an independent index with its own
+dimension, ANN implementation, distance metric, and build options.
 
 </TabItem>
 
@@ -265,11 +266,40 @@ schema validation.
 
 </details>
 
+### Index Multiple Vector Columns
+
+List the vector columns in `pk-vector.index.columns` and configure each column separately.
+For example, this Flink table indexes image and text embeddings independently:
+
+```sql
+CREATE TABLE multimodal_items (
+    id BIGINT,
+    image_embedding ARRAY<FLOAT> COMMENT '__VECTOR_FIELD;512',
+    text_embedding ARRAY<FLOAT> COMMENT '__VECTOR_FIELD;768',
+    PRIMARY KEY (id) NOT ENFORCED
+) WITH (
+    'bucket' = '8',
+    'deletion-vectors.enabled' = 'true',
+    'pk-vector.index.columns' = 'image_embedding,text_embedding',
+    'fields.image_embedding.pk-vector.index.type' = 'ivf-flat',
+    'fields.image_embedding.pk-vector.distance.metric' = 'cosine',
+    'fields.image_embedding.pk-vector.index.options' = '{"nlist":"256"}',
+    'fields.text_embedding.pk-vector.index.type' = 'ivf-flat',
+    'fields.text_embedding.pk-vector.distance.metric' = 'inner_product',
+    'fields.text_embedding.pk-vector.index.options' = '{"nlist":"512"}'
+);
+```
+
+Each vector search selects one column and uses that column's index, dimension, and distance
+metric. Listing multiple columns does not combine their vectors or define a fused ranking.
+Index coverage is tracked separately for each column, so an index on `image_embedding` does
+not cover a query on `text_embedding`.
+
 ### Options
 
 | Option | Default | Description |
 |---|---|---|
-| `pk-vector.index.columns` | Not set | Vector column to index. Exactly one vector column is currently supported. |
+| `pk-vector.index.columns` | Not set | Comma-separated vector columns which own independent vector indexes. |
 | `fields.<column>.pk-vector.index.type` | Required | ANN implementation, such as `ivf-flat`, `ivf-pq`, `ivf-sq`, `ivf-rq`, or `diskann`. |
 | `fields.<column>.pk-vector.distance.metric` | `inner_product` | Distance metric: `l2`, `cosine`, or `inner_product`. |
 | `fields.<column>.pk-vector.index.options` | Not set | JSON object containing build options for the selected ANN implementation. |
@@ -610,7 +640,7 @@ create a table with the desired definition and migrate the data.
 - Multivalue supports arrays of scalar element types accepted by the global-index key serializer.
   Null elements are not indexed. Higher array and element cardinality increases the expanded-sort
   I/O and index-file size.
-- Exactly one vector index column is currently supported per table.
+- Vector indexes are single-column indexes; each configured vector column has an independent index.
 - Exactly one full-text index column is currently supported per table.
 - Only `FLOAT` vectors are supported.
 - Indexes are built from eligible compact output, not directly from Level-0 appends.

docs/generated/core_configuration.html

@@ -1402,7 +1402,7 @@
             <td><h5>pk-vector.index.columns</h5></td>
             <td style="word-wrap: break-word;">(none)</td>
             <td>String</td>
-            <td>Comma-separated VECTOR columns indexed by primary-key vector indexes. Each column owns one index and must define fields.&lt;column&gt;.pk-vector.index.type. Index options and distance metric are also field-scoped. The first release supports exactly one column.</td>
+            <td>Comma-separated VECTOR columns indexed by primary-key vector indexes. Each column owns one index and must define fields.&lt;column&gt;.pk-vector.index.type. Index options and distance metric are also field-scoped.</td>
         </tr>
         <tr>
             <td><h5>postpone.batch-write-fixed-bucket</h5></td>

paimon-api/src/main/java/org/apache/paimon/CoreOptions.java

@@ -3171,8 +3171,7 @@ public String toString() {
                             "Comma-separated VECTOR columns indexed by primary-key vector indexes. "
                                     + "Each column owns one index and must define "
                                     + "fields.<column>.pk-vector.index.type. Index options and distance "
-                                    + "metric are also field-scoped. The first release supports exactly "
-                                    + "one column.");
+                                    + "metric are also field-scoped.");
 
     public static final ConfigOption<String> PK_BTREE_INDEX_COLUMNS =
             key("pk-btree.index.columns")
@@ -5122,15 +5121,6 @@ private Options primaryKeySortedIndexOptions(
         return resolved;
     }
 
-    public String primaryKeyVectorIndexColumn() {
-        List<String> columns = primaryKeyVectorIndexColumns();
-        checkArgument(
-                columns.size() == 1,
-                "pk-vector.index.columns must contain exactly one column in the first release, but is %s.",
-                columns);
-        return columns.get(0);
-    }
-
     @Nullable
     public String primaryKeyVectorIndexType(String column) {
         return options.get("fields." + column + ".pk-vector.index.type");

paimon-common/src/test/java/org/apache/paimon/globalindex/testvector/TestVectorGlobalIndexerFactory.java

@@ -40,6 +40,11 @@ public String identifier() {
 
     @Override
     public GlobalIndexer create(List<DataField> indexFields, Options options) {
+        String metric = options.getString(IDENTIFIER + ".metric", null);
+        if (metric != null) {
+            options = new Options(options.toMap());
+            options.setString(TestVectorGlobalIndexer.OPT_METRIC, metric);
+        }
         if (indexFields.size() == 1) {
             return new TestVectorGlobalIndexer(indexFields.get(0).type(), options);
         }

paimon-core/src/main/java/org/apache/paimon/index/pk/BucketedPrimaryKeyIndexMaintainer.java

@@ -40,6 +40,8 @@
 import org.apache.paimon.io.KeyValueFileReaderFactory;
 import org.apache.paimon.schema.TableSchema;
 import org.apache.paimon.types.DataField;
+import org.apache.paimon.utils.ExceptionUtils;
+import org.apache.paimon.utils.IOUtils;
 
 import javax.annotation.Nullable;
 
@@ -55,16 +57,16 @@
 /** Coordinates all source-backed primary-key indexes for one bucket. */
 public final class BucketedPrimaryKeyIndexMaintainer {
 
-    @Nullable private final BucketedVectorIndexMaintainer vectorMaintainer;
+    private final List<BucketedVectorIndexMaintainer> vectorMaintainers;
     @Nullable private final BucketedFullTextIndexMaintainer fullTextMaintainer;
     private final List<BucketedSortedIndexMaintainer> sortedMaintainers;
     private int nextSortedMaintainerIndex;
 
     private BucketedPrimaryKeyIndexMaintainer(
-            @Nullable BucketedVectorIndexMaintainer vectorMaintainer,
+            List<BucketedVectorIndexMaintainer> vectorMaintainers,
             @Nullable BucketedFullTextIndexMaintainer fullTextMaintainer,
             List<BucketedSortedIndexMaintainer> sortedMaintainers) {
-        this.vectorMaintainer = vectorMaintainer;
+        this.vectorMaintainers = Collections.unmodifiableList(new ArrayList<>(vectorMaintainers));
         this.fullTextMaintainer = fullTextMaintainer;
         List<BucketedSortedIndexMaintainer> sorted = new ArrayList<>(sortedMaintainers);
         sorted.sort(Comparator.comparingInt(BucketedSortedIndexMaintainer::fieldId));
@@ -74,47 +76,60 @@ private BucketedPrimaryKeyIndexMaintainer(
     public static BucketedPrimaryKeyIndexMaintainer ofVector(
             BucketedVectorIndexMaintainer vectorMaintainer) {
         return new BucketedPrimaryKeyIndexMaintainer(
-                vectorMaintainer, null, Collections.emptyList());
+                Collections.singletonList(vectorMaintainer), null, Collections.emptyList());
     }
 
     public static BucketedPrimaryKeyIndexMaintainer ofFullText(
             BucketedFullTextIndexMaintainer fullTextMaintainer) {
         return new BucketedPrimaryKeyIndexMaintainer(
-                null, fullTextMaintainer, Collections.emptyList());
+                Collections.emptyList(), fullTextMaintainer, Collections.emptyList());
     }
 
     public static BucketedPrimaryKeyIndexMaintainer of(
             BucketedVectorIndexMaintainer vectorMaintainer,
             List<BucketedSortedIndexMaintainer> sortedMaintainers) {
-        return new BucketedPrimaryKeyIndexMaintainer(vectorMaintainer, null, sortedMaintainers);
+        return of(vectorMaintainer, null, sortedMaintainers);
     }
 
     public static BucketedPrimaryKeyIndexMaintainer of(
             @Nullable BucketedVectorIndexMaintainer vectorMaintainer,
             @Nullable BucketedFullTextIndexMaintainer fullTextMaintainer,
             List<BucketedSortedIndexMaintainer> sortedMaintainers) {
         return new BucketedPrimaryKeyIndexMaintainer(
-                vectorMaintainer, fullTextMaintainer, sortedMaintainers);
+                vectorMaintainer == null
+                        ? Collections.emptyList()
+                        : Collections.singletonList(vectorMaintainer),
+                fullTextMaintainer,
+                sortedMaintainers);
+    }
+
+    public static BucketedPrimaryKeyIndexMaintainer ofVectors(
+            List<BucketedVectorIndexMaintainer> vectorMaintainers,
+            @Nullable BucketedFullTextIndexMaintainer fullTextMaintainer,
+            List<BucketedSortedIndexMaintainer> sortedMaintainers) {
+        return new BucketedPrimaryKeyIndexMaintainer(
+                vectorMaintainers, fullTextMaintainer, sortedMaintainers);
     }
 
     public static BucketedPrimaryKeyIndexMaintainer ofSorted(
             List<BucketedSortedIndexMaintainer> sortedMaintainers) {
-        return new BucketedPrimaryKeyIndexMaintainer(null, null, sortedMaintainers);
+        return new BucketedPrimaryKeyIndexMaintainer(
+                Collections.emptyList(), null, sortedMaintainers);
     }
 
     public synchronized void prepareCommit(
             DataIncrement appendIncrement,
             CompactIncrement compactIncrement,
             boolean waitCompaction)
             throws Exception {
-        BucketedVectorIndexMaintainer.VectorIndexCommit vectorCommit = null;
+        List<BucketedVectorIndexMaintainer.VectorIndexCommit> vectorCommits = new ArrayList<>();
         BucketedFullTextIndexMaintainer.FullTextIndexCommit fullTextCommit = null;
         List<BucketedSortedIndexMaintainer.SortedIndexCommit> sortedCommits = new ArrayList<>();
         try {
-            if (vectorMaintainer != null) {
-                vectorCommit =
-                        vectorMaintainer.prepareCommit(
-                                appendIncrement, compactIncrement, waitCompaction);
+            for (BucketedVectorIndexMaintainer maintainer : vectorMaintainers) {
+                vectorCommits.add(
+                        maintainer.prepareCommit(
+                                appendIncrement, compactIncrement, waitCompaction));
             }
             if (fullTextMaintainer != null) {
                 fullTextCommit =
@@ -128,8 +143,8 @@ public synchronized void prepareCommit(
                 prepareSortedNonBlocking(appendIncrement, compactIncrement, sortedCommits);
             }
 
-            if (vectorCommit != null) {
-                mergeVectorCommit(appendIncrement, compactIncrement, vectorCommit);
+            for (BucketedVectorIndexMaintainer.VectorIndexCommit commit : vectorCommits) {
+                mergeVectorCommit(appendIncrement, compactIncrement, commit);
             }
             if (fullTextCommit != null) {
                 mergeFullTextCommit(appendIncrement, compactIncrement, fullTextCommit);
@@ -144,8 +159,8 @@ public synchronized void prepareCommit(
             if (fullTextCommit != null) {
                 fullTextCommit.abort(failure);
             }
-            if (vectorCommit != null) {
-                vectorCommit.abort(failure);
+            for (int i = vectorCommits.size() - 1; i >= 0; i--) {
+                vectorCommits.get(i).abort(failure);
             }
             if (failure instanceof Exception) {
                 throw (Exception) failure;
@@ -287,8 +302,10 @@ private static void applyIndexIncrement(
     }
 
     public boolean buildNotCompleted() {
-        if (vectorMaintainer != null && vectorMaintainer.buildNotCompleted()) {
-            return true;
+        for (BucketedVectorIndexMaintainer maintainer : vectorMaintainers) {
+            if (maintainer.buildNotCompleted()) {
+                return true;
+            }
         }
         if (fullTextMaintainer != null && fullTextMaintainer.buildNotCompleted()) {
             return true;
@@ -297,8 +314,11 @@ public boolean buildNotCompleted() {
     }
 
     public void withExecutor(ExecutorService executor) {
-        if (vectorMaintainer != null) {
-            vectorMaintainer.withExecutor(executor);
+        checkArgument(
+                !buildNotCompleted(),
+                "Cannot replace executor during primary-key index maintenance.");
+        for (BucketedVectorIndexMaintainer maintainer : vectorMaintainers) {
+            maintainer.withExecutor(executor);
         }
         if (fullTextMaintainer != null) {
             fullTextMaintainer.withExecutor(executor);
@@ -309,39 +329,48 @@ public void withExecutor(ExecutorService executor) {
     }
 
     public void close() {
-        if (vectorMaintainer != null) {
-            vectorMaintainer.close();
+        List<AutoCloseable> closeables = new ArrayList<>();
+        for (BucketedVectorIndexMaintainer maintainer : vectorMaintainers) {
+            closeables.add(maintainer::close);
         }
         if (fullTextMaintainer != null) {
-            fullTextMaintainer.close();
+            closeables.add(fullTextMaintainer::close);
         }
         for (BucketedSortedIndexMaintainer maintainer : sortedMaintainers) {
-            maintainer.close();
+            closeables.add(maintainer::close);
+        }
+        try {
+            IOUtils.closeAll(closeables);
+        } catch (Exception e) {
+            ExceptionUtils.rethrow(e);
         }
     }
 
     /** Factory to restore all configured source-backed indexes for a bucket. */
     public static final class Factory {
 
         private final IndexFileHandler handler;
-        @Nullable private final BucketedVectorIndexMaintainer.Factory vectorFactory;
+        private final List<BucketedVectorIndexMaintainer.Factory> vectorFactories;
         @Nullable private final FullTextDefinitionFactory fullTextFactory;
         private final List<SortedDefinitionFactory> sortedFactories;
 
         private Factory(
                 IndexFileHandler handler,
-                @Nullable BucketedVectorIndexMaintainer.Factory vectorFactory,
+                List<BucketedVectorIndexMaintainer.Factory> vectorFactories,
                 @Nullable FullTextDefinitionFactory fullTextFactory,
                 List<SortedDefinitionFactory> sortedFactories) {
             this.handler = handler;
-            this.vectorFactory = vectorFactory;
+            this.vectorFactories = Collections.unmodifiableList(new ArrayList<>(vectorFactories));
             this.fullTextFactory = fullTextFactory;
             this.sortedFactories = Collections.unmodifiableList(new ArrayList<>(sortedFactories));
         }
 
         public static Factory ofVector(BucketedVectorIndexMaintainer.Factory vectorFactory) {
             return new Factory(
-                    vectorFactory.indexFileHandler(), vectorFactory, null, Collections.emptyList());
+                    vectorFactory.indexFileHandler(),
+                    Collections.singletonList(vectorFactory),
+                    null,
+                    Collections.emptyList());
         }
 
         public static Factory create(
@@ -350,7 +379,7 @@ public static Factory create(
                 TableSchema schema) {
             CoreOptions coreOptions = new CoreOptions(schema.options());
             Map<String, DataField> fields = schema.nameToFieldMap();
-            BucketedVectorIndexMaintainer.Factory vectorFactory = null;
+            List<BucketedVectorIndexMaintainer.Factory> vectorFactories = new ArrayList<>();
             FullTextDefinitionFactory fullTextFactory = null;
             List<SortedDefinitionFactory> sortedFactories = new ArrayList<>();
             for (PrimaryKeyIndexDefinition definition :
@@ -362,18 +391,15 @@ public static Factory create(
                         definition.column());
                 switch (definition.family()) {
                     case VECTOR:
-                        checkArgument(
-                                vectorFactory == null,
-                                "Only one primary-key vector index is supported.");
-                        vectorFactory =
+                        vectorFactories.add(
                                 new BucketedVectorIndexMaintainer.Factory(
                                                 handler,
                                                 readerFactoryBuilder,
                                                 field,
                                                 coreOptions.primaryKeyVectorDistanceMetric(
                                                         definition.column()),
                                                 definition.indexType())
-                                        .withIndexOptions(definition.options());
+                                        .withIndexOptions(definition.options()));
                         break;
                     case BTREE:
                     case BITMAP:
@@ -409,9 +435,11 @@ public static Factory create(
                 }
             }
             checkArgument(
-                    vectorFactory != null || fullTextFactory != null || !sortedFactories.isEmpty(),
+                    !vectorFactories.isEmpty()
+                            || fullTextFactory != null
+                            || !sortedFactories.isEmpty(),
                     "No primary-key index definition is configured.");
-            return new Factory(handler, vectorFactory, fullTextFactory, sortedFactories);
+            return new Factory(handler, vectorFactories, fullTextFactory, sortedFactories);
         }
 
         public IndexFileHandler indexFileHandler() {
@@ -456,16 +484,17 @@ public BucketedPrimaryKeyIndexMaintainer create(
                     restoredDataFiles == null ? Collections.emptyList() : restoredDataFiles;
             List<IndexFileMeta> payloads =
                     restoredPayloads == null ? Collections.emptyList() : restoredPayloads;
-            BucketedVectorIndexMaintainer vector =
-                    vectorFactory == null
-                            ? null
-                            : vectorFactory.create(
-                                    partition,
-                                    bucket,
-                                    dataFiles,
-                                    payloads,
-                                    executor,
-                                    deletionVectorFactory);
+            List<BucketedVectorIndexMaintainer> vectors = new ArrayList<>();
+            for (BucketedVectorIndexMaintainer.Factory factory : vectorFactories) {
+                vectors.add(
+                        factory.create(
+                                partition,
+                                bucket,
+                                dataFiles,
+                                payloads,
+                                executor,
+                                deletionVectorFactory));
+            }
             BucketedFullTextIndexMaintainer fullText =
                     fullTextFactory == null
                             ? null
@@ -478,7 +507,7 @@ public BucketedPrimaryKeyIndexMaintainer create(
                                 handler, partition, bucket, dataFiles, payloads, executor,
                                 ioManager));
             }
-            return new BucketedPrimaryKeyIndexMaintainer(vector, fullText, sorted);
+            return new BucketedPrimaryKeyIndexMaintainer(vectors, fullText, sorted);
         }
 
         private static final class FullTextDefinitionFactory {