[core][python] Fix global index coverage for residual predicates

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

@@ -39,6 +39,7 @@
 import java.util.ArrayDeque;
 import java.util.ArrayList;
 import java.util.Collection;
+import java.util.Collections;
 import java.util.Deque;
 import java.util.HashSet;
 import java.util.List;
@@ -71,7 +72,15 @@ public Optional<GlobalIndexResult> evaluate(@Nullable Predicate predicate) {
         if (predicate == null) {
             return Optional.empty();
         }
-        return awaitGlobalIndexResult(visitAsync(predicate));
+        return await(visitAsync(predicate)).map(Evaluation::result);
+    }
+
+    /** Evaluate the predicate and return the fields whose supported indexes contributed. */
+    public Optional<Evaluation> evaluateWithContributingFields(@Nullable Predicate predicate) {
+        if (predicate == null) {
+            return Optional.empty();
+        }
+        return await(visitAsync(predicate));
     }
 
     public Optional<GlobalIndexResult> evaluateTopN(TopN topN) {
@@ -84,11 +93,10 @@ public Optional<GlobalIndexResult> evaluateTopN(TopN topN) {
             return Optional.empty();
         }
         checkArgument(readers.size() == 1, "TopN expects one aggregated global index reader.");
-        return awaitGlobalIndexResult(readers.iterator().next().visitTopN(topN));
+        return await(readers.iterator().next().visitTopN(topN));
     }
 
-    private Optional<GlobalIndexResult> awaitGlobalIndexResult(
-            CompletableFuture<Optional<GlobalIndexResult>> future) {
+    private <T> T await(CompletableFuture<T> future) {
         try {
             return future.get();
         } catch (InterruptedException e) {
@@ -105,14 +113,14 @@ private Optional<GlobalIndexResult> awaitGlobalIndexResult(
         }
     }
 
-    private CompletableFuture<Optional<GlobalIndexResult>> visitAsync(Predicate predicate) {
+    private CompletableFuture<Optional<Evaluation>> visitAsync(Predicate predicate) {
         if (predicate instanceof LeafPredicate) {
             return visitLeafAsync((LeafPredicate) predicate);
         }
         return visitCompoundAsync((CompoundPredicate) predicate);
     }
 
-    private CompletableFuture<Optional<GlobalIndexResult>> visitLeafAsync(LeafPredicate predicate) {
+    private CompletableFuture<Optional<Evaluation>> visitLeafAsync(LeafPredicate predicate) {
         Optional<FieldRef> fieldRefOptional = predicate.fieldRefOptional();
         if (!fieldRefOptional.isPresent()) {
             return CompletableFuture.completedFuture(Optional.empty());
@@ -145,18 +153,20 @@ private CompletableFuture<Optional<GlobalIndexResult>> visitLeafAsync(LeafPredic
                                     compoundResult = childResult;
                                 }
                                 if (compoundResult.get().results().isEmpty()) {
-                                    return compoundResult;
+                                    break;
                                 }
                             }
-                            return compoundResult;
+                            return compoundResult.map(
+                                    result ->
+                                            new Evaluation(result, Collections.singleton(fieldId)));
                         });
     }
 
-    private CompletableFuture<Optional<GlobalIndexResult>> visitCompoundAsync(
+    private CompletableFuture<Optional<Evaluation>> visitCompoundAsync(
             CompoundPredicate predicate) {
         List<Predicate> children =
                 pruneRedundantIsNotNullForAnd(flattenChildren(predicate), predicate);
-        List<CompletableFuture<Optional<GlobalIndexResult>>> childFutures =
+        List<CompletableFuture<Optional<Evaluation>>> childFutures =
                 new ArrayList<>(children.size());
         for (Predicate child : children) {
             childFutures.add(visitAsync(child));
@@ -165,40 +175,67 @@ private CompletableFuture<Optional<GlobalIndexResult>> visitCompoundAsync(
         return CompletableFuture.allOf(childFutures.toArray(new CompletableFuture[0]))
                 .thenApply(
                         v -> {
-                            List<Optional<GlobalIndexResult>> results = new ArrayList<>();
-                            for (CompletableFuture<Optional<GlobalIndexResult>> f : childFutures) {
+                            List<Optional<Evaluation>> results = new ArrayList<>();
+                            for (CompletableFuture<Optional<Evaluation>> f : childFutures) {
                                 results.add(f.join());
                             }
                             return combineResults(results, predicate);
                         });
     }
 
-    private Optional<GlobalIndexResult> combineResults(
-            List<Optional<GlobalIndexResult>> results, CompoundPredicate predicate) {
+    private Optional<Evaluation> combineResults(
+            List<Optional<Evaluation>> results, CompoundPredicate predicate) {
+        Set<Integer> contributingFieldIds = new HashSet<>();
         if (predicate.function() instanceof Or) {
             GlobalIndexResult compoundResult = GlobalIndexResult.createEmpty();
-            for (Optional<GlobalIndexResult> childResult : results) {
-                if (!childResult.isPresent()) {
+            for (Optional<Evaluation> child : results) {
+                if (!child.isPresent()) {
                     return Optional.empty();
                 }
-                compoundResult = compoundResult.or(childResult.get());
+                compoundResult = compoundResult.or(child.get().result());
+                contributingFieldIds.addAll(child.get().contributingFieldIds());
             }
-            return Optional.of(compoundResult);
+            return Optional.of(new Evaluation(compoundResult, contributingFieldIds));
         } else {
             Optional<GlobalIndexResult> compoundResult = Optional.empty();
-            for (Optional<GlobalIndexResult> childResult : results) {
-                if (childResult.isPresent()) {
+            for (Optional<Evaluation> child : results) {
+                if (child.isPresent()) {
                     if (compoundResult.isPresent()) {
-                        compoundResult = Optional.of(compoundResult.get().and(childResult.get()));
+                        compoundResult =
+                                Optional.of(compoundResult.get().and(child.get().result()));
                     } else {
-                        compoundResult = childResult;
+                        compoundResult = Optional.of(child.get().result());
                     }
+                    contributingFieldIds.addAll(child.get().contributingFieldIds());
                 }
                 if (compoundResult.isPresent() && compoundResult.get().results().isEmpty()) {
-                    return compoundResult;
+                    break;
                 }
             }
-            return compoundResult;
+            return compoundResult.map(result -> new Evaluation(result, contributingFieldIds));
+        }
+    }
+
+    /**
+     * Matches and fields whose supported index results contributed; discarded branches excluded.
+     */
+    public static final class Evaluation {
+
+        private final GlobalIndexResult result;
+        private final Set<Integer> contributingFieldIds;
+
+        private Evaluation(GlobalIndexResult result, Collection<Integer> contributingFieldIds) {
+            this.result = result;
+            this.contributingFieldIds =
+                    Collections.unmodifiableSet(new HashSet<>(contributingFieldIds));
+        }
+
+        public GlobalIndexResult result() {
+            return result;
+        }
+
+        public Set<Integer> contributingFieldIds() {
+            return contributingFieldIds;
         }
     }
 

paimon-common/src/test/java/org/apache/paimon/globalindex/GlobalIndexEvaluatorTest.java

@@ -225,6 +225,57 @@ void testOrReturnsEmptyWhenChildUnsupported() {
         evaluator.close();
     }
 
+    @Test
+    void testAndTracksOnlyEvaluatedFields() {
+        executor = Executors.newFixedThreadPool(2);
+        RowType rowType = rowType();
+
+        GlobalIndexEvaluator evaluator =
+                new GlobalIndexEvaluator(
+                        rowType,
+                        fieldId ->
+                                fieldId == 0
+                                        ? Collections.singletonList(readerReturning(resultOf(42)))
+                                        : Collections.emptyList());
+        PredicateBuilder builder = new PredicateBuilder(rowType);
+        Predicate predicate = PredicateBuilder.and(builder.equal(0, 42), builder.equal(1, 99));
+
+        Optional<GlobalIndexEvaluator.Evaluation> evaluation =
+                evaluator.evaluateWithContributingFields(predicate);
+
+        assertThat(evaluation).isPresent();
+        assertThat(evaluation.get().contributingFieldIds()).containsExactly(0);
+        assertBitmapContainsExactly(evaluation.get().result().results(), 42L);
+        evaluator.close();
+    }
+
+    @Test
+    void testDiscardedOrBranchDoesNotContributeFields() {
+        executor = Executors.newFixedThreadPool(2);
+        RowType rowType = rowType();
+
+        GlobalIndexEvaluator evaluator =
+                new GlobalIndexEvaluator(
+                        rowType,
+                        fieldId ->
+                                fieldId == 0 || fieldId == 2
+                                        ? Collections.singletonList(readerReturning(resultOf(42)))
+                                        : Collections.emptyList());
+        PredicateBuilder builder = new PredicateBuilder(rowType);
+        Predicate predicate =
+                PredicateBuilder.and(
+                        PredicateBuilder.or(builder.equal(0, 42), builder.equal(1, 99)),
+                        builder.equal(2, 42));
+
+        Optional<GlobalIndexEvaluator.Evaluation> evaluation =
+                evaluator.evaluateWithContributingFields(predicate);
+
+        assertThat(evaluation).isPresent();
+        assertThat(evaluation.get().contributingFieldIds()).containsExactly(2);
+        assertBitmapContainsExactly(evaluation.get().result().results(), 42L);
+        evaluator.close();
+    }
+
     @Test
     void testAndWithEmptyResultShortCircuits() {
         executor = Executors.newFixedThreadPool(2);

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

@@ -318,12 +318,17 @@ private Optional<GlobalIndexResult> evalGlobalIndex() {
 
         try (DataEvolutionGlobalIndexScanner scanner = optionalScanner.get()) {
             long lookupStart = System.nanoTime();
-            Optional<GlobalIndexResult> result = scanner.scan(globalIndexFilter);
+            Optional<GlobalIndexEvaluator.Evaluation> result =
+                    scanner.scanWithCoverage(globalIndexFilter);
             long lookupDuration = System.nanoTime() - lookupStart;
             if (result.isPresent()) {
                 long coverageStart = System.nanoTime();
                 GlobalIndexResult finalResult =
-                        result.get().or(scanner.unindexedRows(globalIndexFilter));
+                        result.get()
+                                .result()
+                                .or(
+                                        scanner.unindexedRowsForContributingFields(
+                                                result.get().contributingFieldIds()));
                 long coverageDuration = System.nanoTime() - coverageStart;
                 long totalDuration = System.nanoTime() - totalStart;
                 LOG.info(

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

@@ -372,6 +372,10 @@ public Optional<GlobalIndexResult> scan(Predicate predicate) {
         return globalIndexEvaluator.evaluate(predicate);
     }
 
+    public Optional<GlobalIndexEvaluator.Evaluation> scanWithCoverage(Predicate predicate) {
+        return globalIndexEvaluator.evaluateWithContributingFields(predicate);
+    }
+
     public Optional<GlobalIndexResult> scan(TopN topN) {
         if (!isSupportedTopN(topN)) {
             return Optional.empty();
@@ -394,6 +398,15 @@ public GlobalIndexResult unindexedRows(Predicate predicate) {
         return GlobalIndexResult.create(rows);
     }
 
+    public GlobalIndexResult unindexedRowsForContributingFields(
+            Collection<Integer> contributingFieldIds) {
+        RoaringNavigableMap64 rows = new RoaringNavigableMap64();
+        for (Range range : coverage.unindexedRanges(contributingFieldIds)) {
+            rows.addRange(range);
+        }
+        return GlobalIndexResult.create(rows);
+    }
+
     public GlobalIndexResult unindexedRows(TopN topN) {
         String fieldName = topN.orders().get(0).field().name();
         RoaringNavigableMap64 rows = new RoaringNavigableMap64();

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

@@ -25,6 +25,7 @@
 import org.apache.paimon.data.InternalVector;
 import org.apache.paimon.fs.FileIO;
 import org.apache.paimon.globalindex.DataEvolutionGlobalIndexScanner;
+import org.apache.paimon.globalindex.GlobalIndexEvaluator;
 import org.apache.paimon.globalindex.GlobalIndexIOMeta;
 import org.apache.paimon.globalindex.GlobalIndexReader;
 import org.apache.paimon.globalindex.GlobalIndexResult;
@@ -221,12 +222,14 @@ protected RoaringNavigableMap64 rawPreFilter(List<RawVectorSearchSplit> splits)
 
         RoaringNavigableMap64 include = new RoaringNavigableMap64();
         try (DataEvolutionGlobalIndexScanner scanner = optionalScanner.get()) {
-            Optional<GlobalIndexResult> result = scanner.scan(filter);
+            Optional<GlobalIndexEvaluator.Evaluation> result = scanner.scanWithCoverage(filter);
             if (!result.isPresent()) {
                 return null;
             }
-            include.or(result.get().results());
-            include.or(scanner.unindexedRows(filter).results());
+            include.or(result.get().result().results());
+            include.or(
+                    scanner.unindexedRowsForContributingFields(result.get().contributingFieldIds())
+                            .results());
         } catch (IOException e) {
             throw new RuntimeException(e);
         }

paimon-core/src/test/java/org/apache/paimon/table/BtreeGlobalIndexTableTest.java

@@ -151,6 +151,32 @@ public void testBTreeGlobalIndexWithCoreScan() throws Exception {
         assertThat(readF1).containsExactly("a200", "a300", "a400", "a56789");
     }
 
+    @Test
+    public void testFullSearchIgnoresUnindexedAndResidualForCoverage() throws Exception {
+        write(100L);
+        createIndex("f1");
+
+        FileStoreTable table =
+                tableWithSearchMode((FileStoreTable) catalog.getTable(identifier()), "full");
+        PredicateBuilder builder = new PredicateBuilder(table.rowType());
+        Predicate predicate =
+                PredicateBuilder.and(
+                        builder.equal(1, BinaryString.fromString("a42")),
+                        builder.equal(2, BinaryString.fromString("b42")));
+        ReadBuilder readBuilder = table.newReadBuilder().withFilter(predicate);
+
+        TableScan.Plan plan = readBuilder.newScan().plan();
+
+        assertThat(plan.splits()).allMatch(IndexedSplit.class::isInstance);
+        assertThat(
+                        plan.splits().stream()
+                                .map(IndexedSplit.class::cast)
+                                .flatMap(split -> split.rowRanges().stream())
+                                .collect(Collectors.toList()))
+                .containsExactly(new Range(42, 42));
+        assertThat(readF1(readBuilder, plan)).containsExactly("a42");
+    }
+
     @Test
     public void testBTreeGlobalIndexTopNCandidatesAcrossRanges() throws Exception {
         write(100L);