[core] Search primary-key vector indexes

paimon-core/src/main/java/org/apache/paimon/table/source/BatchVectorSearchBuilderImpl.java

@@ -125,7 +125,7 @@ public BatchVectorSearchBuilder withOption(String key, String value) {
 
     @Override
     public VectorScan newVectorScan() {
-        return new VectorScanImpl(table, partitionFilter, filter, vectorColumn, options);
+        return new DataEvolutionVectorScan(table, partitionFilter, filter, vectorColumn, options);
     }
 
     @Override

paimon-core/src/main/java/org/apache/paimon/table/source/BucketVectorSearchSplit.java

@@ -0,0 +1,110 @@
+/*
+ * 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.
+ */
+
+package org.apache.paimon.table.source;
+
+import org.apache.paimon.index.IndexFileMeta;
+import org.apache.paimon.index.IndexFileMetaSerializer;
+import org.apache.paimon.io.DataInputViewStreamWrapper;
+import org.apache.paimon.io.DataOutputViewStreamWrapper;
+
+import java.io.IOException;
+import java.io.ObjectInputStream;
+import java.io.ObjectOutputStream;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Objects;
+
+import static org.apache.paimon.utils.Preconditions.checkArgument;
+
+/** All active data files and vector payloads for one snapshot bucket. */
+public class BucketVectorSearchSplit extends VectorSearchSplit {
+
+    private static final long serialVersionUID = 1L;
+    private static final int VERSION = 1;
+
+    private DataSplit dataSplit;
+    private transient List<IndexFileMeta> payloadFiles;
+
+    public BucketVectorSearchSplit(DataSplit dataSplit, List<IndexFileMeta> payloadFiles) {
+        this.dataSplit = dataSplit;
+        for (IndexFileMeta payload : payloadFiles) {
+            checkArgument(
+                    payload.globalIndexMeta() != null
+                            && payload.globalIndexMeta().sourceMeta() != null,
+                    "Primary-key vector payload %s has no source metadata.",
+                    payload.fileName());
+        }
+        this.payloadFiles = Collections.unmodifiableList(new ArrayList<>(payloadFiles));
+    }
+
+    public DataSplit dataSplit() {
+        return dataSplit;
+    }
+
+    public List<IndexFileMeta> payloadFiles() {
+        return payloadFiles;
+    }
+
+    private void writeObject(ObjectOutputStream out) throws IOException {
+        out.defaultWriteObject();
+        out.writeInt(VERSION);
+        out.writeInt(payloadFiles.size());
+        IndexFileMetaSerializer serializer = new IndexFileMetaSerializer();
+        for (IndexFileMeta payloadFile : payloadFiles) {
+            serializer.serialize(payloadFile, new DataOutputViewStreamWrapper(out));
+        }
+    }
+
+    private void readObject(ObjectInputStream in) throws IOException, ClassNotFoundException {
+        in.defaultReadObject();
+        int version = in.readInt();
+        if (version != VERSION) {
+            throw new IOException("Unsupported BucketVectorSearchSplit version: " + version);
+        }
+        int payloadFileCount = in.readInt();
+        if (payloadFileCount < 0) {
+            throw new IOException("Negative primary-key vector payload file count.");
+        }
+        List<IndexFileMeta> payloadFiles = new ArrayList<>(payloadFileCount);
+        IndexFileMetaSerializer serializer = new IndexFileMetaSerializer();
+        for (int i = 0; i < payloadFileCount; i++) {
+            payloadFiles.add(serializer.deserialize(new DataInputViewStreamWrapper(in)));
+        }
+        this.payloadFiles = Collections.unmodifiableList(payloadFiles);
+    }
+
+    @Override
+    public boolean equals(Object o) {
+        if (this == o) {
+            return true;
+        }
+        if (o == null || getClass() != o.getClass()) {
+            return false;
+        }
+        BucketVectorSearchSplit that = (BucketVectorSearchSplit) o;
+        return Objects.equals(dataSplit, that.dataSplit)
+                && Objects.equals(payloadFiles, that.payloadFiles);
+    }
+
+    @Override
+    public int hashCode() {
+        return Objects.hash(dataSplit, payloadFiles);
+    }
+}

paimon-core/src/main/java/org/apache/paimon/table/source/VectorReadImpl.java → paimon-core/src/main/java/org/apache/paimon/table/source/DataEvolutionVectorRead.java

@@ -40,14 +40,14 @@
 
 import static org.apache.paimon.CoreOptions.GLOBAL_INDEX_THREAD_NUM;
 
-/** Implementation for {@link VectorRead}. */
-public class VectorReadImpl extends AbstractVectorRead implements VectorRead {
+/** Data-evolution implementation for {@link VectorRead}. */
+public class DataEvolutionVectorRead extends AbstractVectorRead implements VectorRead {
 
     private static final long serialVersionUID = 1L;
 
     protected final float[] vector;
 
-    public VectorReadImpl(
+    public DataEvolutionVectorRead(
             FileStoreTable table,
             @Nullable PartitionPredicate partitionFilter,
             @Nullable Predicate filter,

paimon-core/src/main/java/org/apache/paimon/table/source/VectorScanImpl.java → paimon-core/src/main/java/org/apache/paimon/table/source/DataEvolutionVectorScan.java

@@ -46,16 +46,16 @@
 import static org.apache.paimon.predicate.PredicateVisitor.collectFieldIds;
 import static org.apache.paimon.utils.Preconditions.checkNotNull;
 
-/** Implementation for {@link VectorScan}. */
-public class VectorScanImpl implements VectorScan {
+/** Data-evolution implementation for {@link VectorScan}. */
+public class DataEvolutionVectorScan implements VectorScan {
 
     private final FileStoreTable table;
     @Nullable private final PartitionPredicate partitionFilter;
     @Nullable private final Predicate filter;
     private final DataField vectorColumn;
     private final Map<String, String> options;
 
-    public VectorScanImpl(
+    public DataEvolutionVectorScan(
             FileStoreTable table,
             @Nullable PartitionPredicate partitionFilter,
             @Nullable Predicate filter,

paimon-core/src/main/java/org/apache/paimon/table/source/DataTableBatchScan.java

@@ -19,6 +19,7 @@
 package org.apache.paimon.table.source;
 
 import org.apache.paimon.CoreOptions;
+import org.apache.paimon.globalindex.GlobalIndexResult;
 import org.apache.paimon.manifest.PartitionEntry;
 import org.apache.paimon.predicate.Predicate;
 import org.apache.paimon.predicate.SortValue;
@@ -66,6 +67,7 @@ public static void validateScanBucketOption(
 
     private final SchemaManager schemaManager;
     @Nullable private String readProtectionTagName;
+    @Nullable private GlobalIndexSplitResult globalIndexSplitResult;
 
     public DataTableBatchScan(
             TableSchema schema,
@@ -110,6 +112,17 @@ public InnerTableScan withTopN(TopN topN) {
 
     @Override
     protected TableScan.Plan planWithoutAuth() {
+        if (globalIndexSplitResult != null) {
+            if (!hasNext) {
+                throw new EndOfScanException();
+            }
+            hasNext = false;
+            if (globalIndexSplitResult.snapshotId() > 0) {
+                maybeCreateReadProtectionTag(globalIndexSplitResult.snapshotId());
+            }
+            List<Split> splits = new ArrayList<>(globalIndexSplitResult.splits());
+            return new PlanImpl(null, globalIndexSplitResult.snapshotId(), splits);
+        }
         if (startingScanner == null) {
             startingScanner = createStartingScanner(false);
         }
@@ -135,6 +148,14 @@ protected TableScan.Plan planWithoutAuth() {
         }
     }
 
+    @Override
+    public DataTableBatchScan withGlobalIndexResult(GlobalIndexResult globalIndexResult) {
+        if (globalIndexResult instanceof GlobalIndexSplitResult) {
+            this.globalIndexSplitResult = (GlobalIndexSplitResult) globalIndexResult;
+        }
+        return this;
+    }
+
     @Override
     public List<PartitionEntry> listPartitionEntries() {
         if (startingScanner == null) {

paimon-core/src/main/java/org/apache/paimon/table/source/GlobalIndexSplitResult.java

@@ -0,0 +1,31 @@
+/*
+ * 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.
+ */
+
+package org.apache.paimon.table.source;
+
+import org.apache.paimon.globalindex.GlobalIndexResult;
+
+import java.util.List;
+
+/** Global index result materialized as ready-to-read splits from one table snapshot. */
+public interface GlobalIndexSplitResult extends GlobalIndexResult {
+
+    long snapshotId();
+
+    List<? extends Split> splits();
+}