[core] Refactor scalar index eval and apply pre filter to vector search

Author: JingsongLi

paimon-common/src/main/java/org/apache/paimon/globalindex/UnionGlobalIndexReader.java

@@ -21,10 +21,19 @@
 import org.apache.paimon.predicate.FieldRef;
 import org.apache.paimon.predicate.VectorSearch;
 
+import javax.annotation.Nullable;
+
 import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Iterator;
 import java.util.List;
 import java.util.Optional;
+import java.util.concurrent.ExecutorService;
 import java.util.function.Function;
+import java.util.stream.Collectors;
+
+import static java.util.Collections.singletonList;
+import static org.apache.paimon.utils.ThreadPoolUtils.randomlyExecuteSequentialReturn;
 
 /**
  * A {@link GlobalIndexReader} that combines results from multiple readers by performing a union
@@ -33,9 +42,16 @@
 public class UnionGlobalIndexReader implements GlobalIndexReader {
 
     private final List<GlobalIndexReader> readers;
+    private final @Nullable ExecutorService executor;
 
     public UnionGlobalIndexReader(List<GlobalIndexReader> readers) {
+        this(readers, null);
+    }
+
+    public UnionGlobalIndexReader(
+            List<GlobalIndexReader> readers, @Nullable ExecutorService executor) {
         this.readers = readers;
+        this.executor = executor;
     }
 
     @Override
@@ -116,8 +132,9 @@ public Optional<GlobalIndexResult> visitBetween(FieldRef fieldRef, Object from,
     @Override
     public Optional<ScoredGlobalIndexResult> visitVectorSearch(VectorSearch vectorSearch) {
         Optional<ScoredGlobalIndexResult> result = Optional.empty();
-        for (GlobalIndexReader reader : readers) {
-            Optional<ScoredGlobalIndexResult> current = reader.visitVectorSearch(vectorSearch);
+        List<Optional<ScoredGlobalIndexResult>> results =
+                executeAllReaders(reader -> reader.visitVectorSearch(vectorSearch));
+        for (Optional<ScoredGlobalIndexResult> current : results) {
             if (!current.isPresent()) {
                 continue;
             }
@@ -132,8 +149,8 @@ public Optional<ScoredGlobalIndexResult> visitVectorSearch(VectorSearch vectorSe
     private Optional<GlobalIndexResult> union(
             Function<GlobalIndexReader, Optional<GlobalIndexResult>> visitor) {
         Optional<GlobalIndexResult> result = Optional.empty();
-        for (GlobalIndexReader reader : readers) {
-            Optional<GlobalIndexResult> current = visitor.apply(reader);
+        List<Optional<GlobalIndexResult>> results = executeAllReaders(visitor);
+        for (Optional<GlobalIndexResult> current : results) {
             if (!current.isPresent()) {
                 continue;
             }
@@ -145,6 +162,21 @@ private Optional<GlobalIndexResult> union(
         return result;
     }
 
+    private <R> List<R> executeAllReaders(Function<GlobalIndexReader, R> function) {
+        if (executor == null) {
+            return readers.stream().map(function).collect(Collectors.toList());
+        }
+
+        Iterator<R> iterator =
+                randomlyExecuteSequentialReturn(
+                        executor, reader -> singletonList(function.apply(reader)), readers);
+        List<R> result = new ArrayList<>();
+        while (iterator.hasNext()) {
+            result.add(iterator.next());
+        }
+        return result;
+    }
+
     @Override
     public void close() throws IOException {
         for (GlobalIndexReader reader : readers) {

paimon-core/src/main/java/org/apache/paimon/AbstractFileStore.java

@@ -26,8 +26,6 @@
 import org.apache.paimon.format.FileFormat;
 import org.apache.paimon.fs.FileIO;
 import org.apache.paimon.fs.Path;
-import org.apache.paimon.globalindex.GlobalIndexScanBuilder;
-import org.apache.paimon.globalindex.GlobalIndexScanBuilderImpl;
 import org.apache.paimon.iceberg.IcebergCommitCallback;
 import org.apache.paimon.iceberg.IcebergOptions;
 import org.apache.paimon.index.IndexFileHandler;
@@ -523,15 +521,4 @@ public void setManifestCache(SegmentsCache<Path> manifestCache) {
     public void setSnapshotCache(Cache<Path, Snapshot> cache) {
         this.snapshotCache = cache;
     }
-
-    @Override
-    public GlobalIndexScanBuilder newGlobalIndexScanBuilder() {
-        return new GlobalIndexScanBuilderImpl(
-                options.toConfiguration(),
-                schema.logicalRowType(),
-                fileIO,
-                pathFactory().globalIndexFileFactory(),
-                snapshotManager(),
-                newIndexFileHandler());
-    }
 }

paimon-core/src/main/java/org/apache/paimon/FileStore.java

@@ -19,7 +19,6 @@
 package org.apache.paimon;
 
 import org.apache.paimon.fs.Path;
-import org.apache.paimon.globalindex.GlobalIndexScanBuilder;
 import org.apache.paimon.index.IndexFileHandler;
 import org.apache.paimon.manifest.IndexManifestFile;
 import org.apache.paimon.manifest.ManifestFile;
@@ -125,6 +124,4 @@ PartitionExpire newPartitionExpire(
     void setManifestCache(SegmentsCache<Path> manifestCache);
 
     void setSnapshotCache(Cache<Path, Snapshot> cache);
-
-    GlobalIndexScanBuilder newGlobalIndexScanBuilder();
 }

paimon-core/src/main/java/org/apache/paimon/globalindex/DataEvolutionBatchScan.java

@@ -18,7 +18,7 @@
 
 package org.apache.paimon.globalindex;
 
-import org.apache.paimon.Snapshot;
+import org.apache.paimon.CoreOptions;
 import org.apache.paimon.annotation.VisibleForTesting;
 import org.apache.paimon.data.BinaryRow;
 import org.apache.paimon.io.DataFileMeta;
@@ -36,23 +36,21 @@
 import org.apache.paimon.table.source.DataTableScan;
 import org.apache.paimon.table.source.InnerTableScan;
 import org.apache.paimon.table.source.Split;
-import org.apache.paimon.table.source.snapshot.TimeTravelUtil;
 import org.apache.paimon.types.RowType;
 import org.apache.paimon.utils.Filter;
 import org.apache.paimon.utils.Range;
 import org.apache.paimon.utils.RowRangeIndex;
 
 import javax.annotation.Nullable;
 
+import java.io.IOException;
 import java.util.ArrayList;
 import java.util.Collections;
 import java.util.List;
 import java.util.Map;
-import java.util.Objects;
 import java.util.Optional;
 import java.util.function.Function;
 
-import static org.apache.paimon.globalindex.GlobalIndexScanBuilder.parallelScan;
 import static org.apache.paimon.table.SpecialFields.ROW_ID;
 import static org.apache.paimon.utils.ManifestReadThreadPool.randomlyExecuteSequentialReturn;
 
@@ -263,39 +261,18 @@ private Optional<GlobalIndexResult> evalGlobalIndex() {
         if (filter == null) {
             return Optional.empty();
         }
-        if (!table.coreOptions().globalIndexEnabled()) {
+        CoreOptions options = table.coreOptions();
+        if (!options.globalIndexEnabled()) {
             return Optional.empty();
         }
-        PartitionPredicate partitionPredicate =
+        PartitionPredicate partitionFilter =
                 batchScan.snapshotReader().manifestsReader().partitionFilter();
-        GlobalIndexScanBuilder indexScanBuilder = table.store().newGlobalIndexScanBuilder();
-        Snapshot snapshot = TimeTravelUtil.tryTravelOrLatest(table);
-        indexScanBuilder.withPartitionPredicate(partitionPredicate).withSnapshot(snapshot);
-        List<Range> indexedRowRanges = indexScanBuilder.shardList();
-        if (indexedRowRanges.isEmpty()) {
-            return Optional.empty();
-        }
-
-        Long nextRowId = Objects.requireNonNull(snapshot.nextRowId());
-        List<Range> nonIndexedRowRanges = new Range(0, nextRowId - 1).exclude(indexedRowRanges);
-        Optional<GlobalIndexResult> resultOptional =
-                parallelScan(
-                        indexedRowRanges,
-                        indexScanBuilder,
-                        filter,
-                        table.coreOptions().globalIndexThreadNum());
-        if (!resultOptional.isPresent()) {
-            return Optional.empty();
+        try (GlobalIndexScanner scanner =
+                GlobalIndexScanner.create(table, partitionFilter, filter)) {
+            return scanner.scan(filter);
+        } catch (IOException e) {
+            throw new RuntimeException(e);
         }
-
-        GlobalIndexResult result = resultOptional.get();
-        if (!nonIndexedRowRanges.isEmpty()) {
-            for (Range range : nonIndexedRowRanges) {
-                result.or(GlobalIndexResult.fromRange(range));
-            }
-        }
-
-        return Optional.of(result);
     }
 
     @VisibleForTesting