[common][core][format] Support using file index result for RowGroup filtering.

paimon-common/src/main/java/org/apache/paimon/fileindex/FileIndexPredicate.java

@@ -67,22 +67,20 @@ public FileIndexPredicate(SeekableInputStream inputStream, RowType fileRowType)
         this.reader = FileIndexFormat.createReader(inputStream, fileRowType);
     }
 
-    public boolean testPredicate(@Nullable Predicate filePredicate) {
-        if (filePredicate == null) {
-            return true;
+    public FileIndexResult evaluate(@Nullable Predicate predicate) {
+        if (predicate == null) {
+            return REMAIN;
         }
-
-        Set<String> requiredFieldNames = getRequiredNames(filePredicate);
-
+        Set<String> requiredFieldNames = getRequiredNames(predicate);
         Map<String, Collection<FileIndexReader>> indexReaders = new HashMap<>();
         requiredFieldNames.forEach(name -> indexReaders.put(name, reader.readColumnIndex(name)));
-        if (!new FileIndexPredicateTest(indexReaders).test(filePredicate).remain()) {
+        FileIndexResult result = new FileIndexPredicateTest(indexReaders).test(predicate);
+        if (!result.remain()) {
             LOG.debug(
                     "One file has been filtered: "
                             + (path == null ? "in scan stage" : path.toString()));
-            return false;
         }
-        return true;
+        return result;
     }
 
     private Set<String> getRequiredNames(Predicate filePredicate) {

paimon-common/src/main/java/org/apache/paimon/format/FormatReaderContext.java

@@ -18,21 +18,31 @@
 
 package org.apache.paimon.format;
 
+import org.apache.paimon.fileindex.FileIndexResult;
 import org.apache.paimon.fs.FileIO;
 import org.apache.paimon.fs.Path;
 import org.apache.paimon.reader.RecordReader;
 
+import javax.annotation.Nullable;
+
 /** the context for creating RecordReader {@link RecordReader}. */
 public class FormatReaderContext implements FormatReaderFactory.Context {
 
     private final FileIO fileIO;
     private final Path file;
     private final long fileSize;
+    @Nullable private final FileIndexResult fileIndexResult;
 
     public FormatReaderContext(FileIO fileIO, Path file, long fileSize) {
+        this(fileIO, file, fileSize, null);
+    }
+
+    public FormatReaderContext(
+            FileIO fileIO, Path file, long fileSize, @Nullable FileIndexResult fileIndexResult) {
         this.fileIO = fileIO;
         this.file = file;
         this.fileSize = fileSize;
+        this.fileIndexResult = fileIndexResult;
     }
 
     @Override
@@ -49,4 +59,9 @@ public Path filePath() {
     public long fileSize() {
         return fileSize;
     }
+
+    @Override
+    public FileIndexResult fileIndex() {
+        return fileIndexResult;
+    }
 }

paimon-common/src/main/java/org/apache/paimon/format/FormatReaderFactory.java

@@ -19,6 +19,7 @@
 package org.apache.paimon.format;
 
 import org.apache.paimon.data.InternalRow;
+import org.apache.paimon.fileindex.FileIndexResult;
 import org.apache.paimon.fs.FileIO;
 import org.apache.paimon.fs.Path;
 import org.apache.paimon.reader.RecordReader;
@@ -38,5 +39,7 @@ interface Context {
         Path filePath();
 
         long fileSize();
+
+        FileIndexResult fileIndex();
     }
 }

paimon-common/src/main/java/org/apache/paimon/utils/RoaringBitmap32.java

@@ -68,6 +68,10 @@ public long getCardinality() {
         return roaringBitmap.getLongCardinality();
     }
 
+    public long rangeCardinality(long start, long end) {
+        return roaringBitmap.rangeCardinality(start, end);
+    }
+
     public RoaringBitmap32 clone() {
         return new RoaringBitmap32(roaringBitmap.clone());
     }

paimon-core/src/main/java/org/apache/paimon/io/FileIndexSkipper.java → paimon-core/src/main/java/org/apache/paimon/io/FileIndexEvaluator.java

@@ -19,6 +19,7 @@
 package org.apache.paimon.io;
 
 import org.apache.paimon.fileindex.FileIndexPredicate;
+import org.apache.paimon.fileindex.FileIndexResult;
 import org.apache.paimon.fs.FileIO;
 import org.apache.paimon.predicate.Predicate;
 import org.apache.paimon.predicate.PredicateBuilder;
@@ -28,10 +29,10 @@
 import java.util.List;
 import java.util.stream.Collectors;
 
-/** File index reader, do the filter in the constructor. */
-public class FileIndexSkipper {
+/** Evaluate file index result. */
+public class FileIndexEvaluator {
 
-    public static boolean skip(
+    public static FileIndexResult evaluate(
             FileIO fileIO,
             TableSchema dataSchema,
             List<Predicate> dataFilter,
@@ -55,14 +56,11 @@ public static boolean skip(
                                 dataFilePathFactory.toPath(indexFiles.get(0)),
                                 fileIO,
                                 dataSchema.logicalRowType())) {
-                    if (!predicate.testPredicate(
-                            PredicateBuilder.and(dataFilter.toArray(new Predicate[0])))) {
-                        return true;
-                    }
+                    return predicate.evaluate(
+                            PredicateBuilder.and(dataFilter.toArray(new Predicate[0])));
                 }
             }
         }
-
-        return false;
+        return FileIndexResult.REMAIN;
     }
 }

paimon-core/src/main/java/org/apache/paimon/operation/AppendOnlyFileStoreScan.java

@@ -120,7 +120,7 @@ private boolean testFileIndex(@Nullable byte[] embeddedIndexBytes, ManifestEntry
 
         try (FileIndexPredicate predicate =
                 new FileIndexPredicate(embeddedIndexBytes, dataRowType)) {
-            return predicate.testPredicate(dataPredicate);
+            return predicate.evaluate(dataPredicate).remain();
         } catch (IOException e) {
             throw new RuntimeException("Exception happens while checking predicate.", e);
         }

paimon-core/src/main/java/org/apache/paimon/operation/KeyValueFileStoreScan.java

@@ -144,7 +144,7 @@ private boolean filterByFileIndex(@Nullable byte[] embeddedIndexBytes, ManifestE
                             id ->
                                     fieldValueStatsConverters.convertFilter(
                                             entry.file().schemaId(), valueFilter));
-            return predicate.testPredicate(dataPredicate);
+            return predicate.evaluate(dataPredicate).remain();
         } catch (IOException e) {
             throw new RuntimeException("Exception happens while checking fileIndex predicate.", e);
         }

paimon-core/src/main/java/org/apache/paimon/operation/RawFileSplitRead.java

@@ -23,13 +23,14 @@
 import org.apache.paimon.deletionvectors.ApplyDeletionVectorReader;
 import org.apache.paimon.deletionvectors.DeletionVector;
 import org.apache.paimon.disk.IOManager;
+import org.apache.paimon.fileindex.FileIndexResult;
 import org.apache.paimon.format.FileFormatDiscover;
 import org.apache.paimon.format.FormatKey;
 import org.apache.paimon.format.FormatReaderContext;
 import org.apache.paimon.fs.FileIO;
 import org.apache.paimon.io.DataFileMeta;
 import org.apache.paimon.io.DataFilePathFactory;
-import org.apache.paimon.io.FileIndexSkipper;
+import org.apache.paimon.io.FileIndexEvaluator;
 import org.apache.paimon.io.FileRecordReader;
 import org.apache.paimon.mergetree.compact.ConcatRecordReader;
 import org.apache.paimon.partition.PartitionUtils;
@@ -191,26 +192,30 @@ private RecordReader<InternalRow> createFileReader(
             BulkFormatMapping bulkFormatMapping,
             IOExceptionSupplier<DeletionVector> dvFactory)
             throws IOException {
+        FileIndexResult fileIndexResult = null;
         if (fileIndexReadEnabled) {
-            boolean skip =
-                    FileIndexSkipper.skip(
+            fileIndexResult =
+                    FileIndexEvaluator.evaluate(
                             fileIO,
                             bulkFormatMapping.getDataSchema(),
                             bulkFormatMapping.getDataFilters(),
                             dataFilePathFactory,
                             file);
-            if (skip) {
+            if (!fileIndexResult.remain()) {
                 return new EmptyRecordReader<>();
             }
         }
 
+        FormatReaderContext formatReaderContext =
+                new FormatReaderContext(
+                        fileIO,
+                        dataFilePathFactory.toPath(file.fileName()),
+                        file.fileSize(),
+                        fileIndexResult);
         FileRecordReader fileRecordReader =
                 new FileRecordReader(
                         bulkFormatMapping.getReaderFactory(),
-                        new FormatReaderContext(
-                                fileIO,
-                                dataFilePathFactory.toPath(file.fileName()),
-                                file.fileSize()),
+                        formatReaderContext,
                         bulkFormatMapping.getIndexMapping(),
                         bulkFormatMapping.getCastMapping(),
                         PartitionUtils.create(bulkFormatMapping.getPartitionPair(), partition));

paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/compact/changelog/format/CompactedChangelogFormatReaderFactory.java

@@ -20,6 +20,7 @@
 
 import org.apache.paimon.catalog.CatalogContext;
 import org.apache.paimon.data.InternalRow;
+import org.apache.paimon.fileindex.FileIndexResult;
 import org.apache.paimon.format.FormatReaderFactory;
 import org.apache.paimon.fs.FileIO;
 import org.apache.paimon.fs.FileStatus;
@@ -80,6 +81,11 @@ public Path filePath() {
                     public long fileSize() {
                         return length;
                     }
+
+                    @Override
+                    public FileIndexResult fileIndex() {
+                        return context.fileIndex();
+                    }
                 });
     }