[WIP][spark] Distribute lateral vector search across index splits

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

@@ -41,6 +41,7 @@
 import java.util.List;
 import java.util.Map;
 import java.util.Objects;
+import java.util.OptionalLong;
 import java.util.Set;
 import java.util.stream.Collectors;
 
@@ -186,6 +187,11 @@ public Plan scan() {
             public List<VectorSearchSplit> splits() {
                 return splits;
             }
+
+            @Override
+            public OptionalLong plannedSnapshotId() {
+                return snapshot == null ? OptionalLong.empty() : OptionalLong.of(snapshot.id());
+            }
         };
     }
 

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

@@ -19,6 +19,7 @@
 package org.apache.paimon.table.source;
 
 import java.util.List;
+import java.util.OptionalLong;
 
 /** Vector scan to pre-filter and scan index files. */
 public interface VectorScan {
@@ -28,5 +29,9 @@ public interface VectorScan {
     /** Plan of vector scan. */
     interface Plan {
         List<VectorSearchSplit> splits();
+
+        default OptionalLong plannedSnapshotId() {
+            return OptionalLong.empty();
+        }
     }
 }

paimon-core/src/test/java/org/apache/paimon/table/source/VectorSearchBuilderTest.java

@@ -464,6 +464,56 @@ public void testVectorSearchFastModeSkipsUnindexedDataByDefault() throws Excepti
         assertThat(result.results()).doesNotContain(2L, 3L);
     }
 
+    @Test
+    public void testVectorScanPlanCarriesLatestSnapshot() throws Exception {
+        createTableDefault();
+        FileStoreTable table = getTableDefault();
+
+        float[][] vectors = {{1.0f, 0.0f}, {0.0f, 1.0f}};
+        writeVectors(table, vectors);
+        buildAndCommitIndex(table, vectors);
+
+        long latestSnapshotId = table.snapshotManager().latestSnapshotId();
+        VectorScan.Plan plan =
+                table.newVectorSearchBuilder()
+                        .withVector(new float[] {1.0f, 0.0f})
+                        .withLimit(1)
+                        .withVectorColumn(VECTOR_FIELD_NAME)
+                        .newVectorScan()
+                        .scan();
+
+        assertThat(plan.plannedSnapshotId()).hasValue(latestSnapshotId);
+    }
+
+    @Test
+    public void testVectorScanPlanCarriesTimeTravelSnapshot() throws Exception {
+        createTableDefault();
+        FileStoreTable table = getTableDefault();
+
+        float[][] vectors = {{1.0f, 0.0f}, {0.0f, 1.0f}};
+        writeVectors(table, vectors);
+        buildAndCommitIndex(table, vectors);
+        long historicalSnapshotId = table.snapshotManager().latestSnapshotId();
+        writeVectors(table, new float[][] {{0.5f, 0.5f}});
+        assertThat(table.snapshotManager().latestSnapshotId()).isGreaterThan(historicalSnapshotId);
+
+        FileStoreTable timeTravelTable =
+                table.copy(
+                        Collections.singletonMap(
+                                CoreOptions.SCAN_SNAPSHOT_ID.key(),
+                                String.valueOf(historicalSnapshotId)));
+        VectorScan.Plan plan =
+                timeTravelTable
+                        .newVectorSearchBuilder()
+                        .withVector(new float[] {1.0f, 0.0f})
+                        .withLimit(1)
+                        .withVectorColumn(VECTOR_FIELD_NAME)
+                        .newVectorScan()
+                        .scan();
+
+        assertThat(plan.plannedSnapshotId()).hasValue(historicalSnapshotId);
+    }
+
     @Test
     public void testVectorSearchFullModeScansFilteredUnindexedData() throws Exception {
         catalog.createTable(

paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/SparkConnectorOptions.java

@@ -47,6 +47,24 @@ public class SparkConnectorOptions {
                             "Parallelism used to repartition a single-partition LIMIT input before "
                                     + "executing a lateral vector search.");
 
+    public static final ConfigOption<Boolean> VECTOR_SEARCH_LATERAL_JOIN_DISTRIBUTED_ENABLED =
+            key("vector-search.lateral-join.distributed.enabled")
+                    .booleanType()
+                    .defaultValue(false)
+                    .withDescription(
+                            "Whether to distribute lateral batch vector search across compatible "
+                                    + "vector index split groups. Unsupported plans fall back to "
+                                    + "the local batch search path.");
+
+    public static final ConfigOption<Integer>
+            VECTOR_SEARCH_LATERAL_JOIN_DISTRIBUTED_MAX_SPLIT_GROUPS =
+                    key("vector-search.lateral-join.distributed.max-split-groups")
+                            .intType()
+                            .defaultValue(16)
+                            .withDescription(
+                                    "Maximum number of independently searched vector index split "
+                                            + "groups for distributed lateral search.");
+
     public static final ConfigOption<Boolean> MERGE_SCHEMA =
             key("write.merge-schema")
                     .booleanType()