add extra file-index file check

Akwangg · Akwangg · commit 5cd9bceb7c13 · 2025-07-22T17:14:58.000+08:00
diff --git a/paimon-core/src/main/java/org/apache/paimon/AppendOnlyFileStore.java b/paimon-core/src/main/java/org/apache/paimon/AppendOnlyFileStore.java
@@ -155,6 +155,8 @@ public AppendOnlyFileStoreScan newScan() {
                 schemaManager,
                 schema,
                 manifestFileFactory(),
+                fileIO,
+                pathFactory(),
                 options.scanManifestParallelism(),
                 options.fileIndexReadEnabled());
     }
diff --git a/paimon-core/src/main/java/org/apache/paimon/io/FileIndexEvaluator.java b/paimon-core/src/main/java/org/apache/paimon/io/FileIndexEvaluator.java
@@ -22,7 +22,6 @@
 import org.apache.paimon.fileindex.FileIndexResult;
 import org.apache.paimon.fs.FileIO;
 import org.apache.paimon.predicate.Predicate;
-import org.apache.paimon.predicate.PredicateBuilder;
 import org.apache.paimon.schema.TableSchema;
 
 import java.io.IOException;
@@ -35,17 +34,16 @@ public class FileIndexEvaluator {
     public static FileIndexResult evaluate(
             FileIO fileIO,
             TableSchema dataSchema,
-            List<Predicate> dataFilter,
+            Predicate dataPredicate,
             DataFilePathFactory dataFilePathFactory,
             DataFileMeta file)
             throws IOException {
-        if (dataFilter != null && !dataFilter.isEmpty()) {
+        if (dataPredicate != null) {
             byte[] embeddedIndex = file.embeddedIndex();
             if (embeddedIndex != null) {
                 try (FileIndexPredicate predicate =
                         new FileIndexPredicate(embeddedIndex, dataSchema.logicalRowType())) {
-                    return predicate.evaluate(
-                            PredicateBuilder.and(dataFilter.toArray(new Predicate[0])));
+                    return predicate.evaluate(dataPredicate);
                 }
             }
 
@@ -65,8 +63,7 @@ public static FileIndexResult evaluate(
                                 dataFilePathFactory.toAlignedPath(indexFiles.get(0), file),
                                 fileIO,
                                 dataSchema.logicalRowType())) {
-                    return predicate.evaluate(
-                            PredicateBuilder.and(dataFilter.toArray(new Predicate[0])));
+                    return predicate.evaluate(dataPredicate);
                 }
             }
         }
diff --git a/paimon-core/src/main/java/org/apache/paimon/operation/AppendOnlyFileStoreScan.java b/paimon-core/src/main/java/org/apache/paimon/operation/AppendOnlyFileStoreScan.java
@@ -19,15 +19,17 @@
 package org.apache.paimon.operation;
 
 import org.apache.paimon.AppendOnlyFileStore;
-import org.apache.paimon.fileindex.FileIndexPredicate;
+import org.apache.paimon.fs.FileIO;
+import org.apache.paimon.io.DataFilePathFactory;
+import org.apache.paimon.io.FileIndexEvaluator;
 import org.apache.paimon.manifest.ManifestEntry;
 import org.apache.paimon.manifest.ManifestFile;
 import org.apache.paimon.predicate.Predicate;
 import org.apache.paimon.schema.SchemaManager;
 import org.apache.paimon.schema.TableSchema;
 import org.apache.paimon.stats.SimpleStatsEvolution;
 import org.apache.paimon.stats.SimpleStatsEvolutions;
-import org.apache.paimon.types.RowType;
+import org.apache.paimon.utils.FileStorePathFactory;
 import org.apache.paimon.utils.SnapshotManager;
 
 import javax.annotation.Nullable;
@@ -46,6 +48,9 @@ public class AppendOnlyFileStoreScan extends AbstractFileStoreScan {
 
     private Predicate filter;
 
+    private FileIO fileIO;
+    private FileStorePathFactory pathFactory;
+
     // just cache.
     private final Map<Long, Predicate> dataFilterMapping = new ConcurrentHashMap<>();
 
@@ -56,6 +61,8 @@ public AppendOnlyFileStoreScan(
             SchemaManager schemaManager,
             TableSchema schema,
             ManifestFile.Factory manifestFileFactory,
+            FileIO fileIO,
+            FileStorePathFactory pathFactory,
             Integer scanManifestParallelism,
             boolean fileIndexReadEnabled) {
         super(
@@ -69,6 +76,8 @@ public AppendOnlyFileStoreScan(
         this.simpleStatsEvolutions =
                 new SimpleStatsEvolutions(sid -> scanTableSchema(sid).fields(), schema.id());
         this.fileIndexReadEnabled = fileIndexReadEnabled;
+        this.fileIO = fileIO;
+        this.pathFactory = pathFactory;
     }
 
     public AppendOnlyFileStoreScan withFilter(Predicate predicate) {
@@ -102,24 +111,22 @@ protected boolean filterByStats(ManifestEntry entry) {
     }
 
     private boolean testFileIndex(@Nullable byte[] embeddedIndexBytes, ManifestEntry entry) {
-        if (embeddedIndexBytes == null) {
-            return true;
-        }
-
-        RowType dataRowType = scanTableSchema(entry.file().schemaId()).logicalRowType();
-
+        TableSchema tableSchema = scanTableSchema(entry.file().schemaId());
         Predicate dataPredicate =
                 dataFilterMapping.computeIfAbsent(
                         entry.file().schemaId(),
                         id ->
                                 simpleStatsEvolutions.toEvolutionSafeStatsFilter(
                                         entry.file().schemaId(), filter));
 
-        try (FileIndexPredicate predicate =
-                new FileIndexPredicate(embeddedIndexBytes, dataRowType)) {
-            return predicate.evaluate(dataPredicate).remain();
+        DataFilePathFactory dataFilePathFactory =
+                pathFactory.createDataFilePathFactory(entry.partition(), entry.bucket());
+        try {
+            return FileIndexEvaluator.evaluate(
+                            fileIO, tableSchema, dataPredicate, dataFilePathFactory, entry.file())
+                    .remain();
         } catch (IOException e) {
-            throw new RuntimeException("Exception happens while checking predicate.", e);
+            throw new RuntimeException("Exception happens while checking file index.", e);
         }
     }
 }
diff --git a/paimon-core/src/main/java/org/apache/paimon/operation/RawFileSplitRead.java b/paimon-core/src/main/java/org/apache/paimon/operation/RawFileSplitRead.java
@@ -38,6 +38,7 @@
 import org.apache.paimon.mergetree.compact.ConcatRecordReader;
 import org.apache.paimon.partition.PartitionUtils;
 import org.apache.paimon.predicate.Predicate;
+import org.apache.paimon.predicate.PredicateBuilder;
 import org.apache.paimon.reader.EmptyFileRecordReader;
 import org.apache.paimon.reader.FileRecordReader;
 import org.apache.paimon.reader.ReaderSupplier;
@@ -198,11 +199,14 @@ private FileRecordReader<InternalRow> createFileReader(
             throws IOException {
         FileIndexResult fileIndexResult = null;
         if (fileIndexReadEnabled) {
+            Predicate dataPredicate =
+                    PredicateBuilder.and(
+                            formatReaderMapping.getDataFilters().toArray(new Predicate[0]));
             fileIndexResult =
                     FileIndexEvaluator.evaluate(
                             fileIO,
                             formatReaderMapping.getDataSchema(),
-                            formatReaderMapping.getDataFilters(),
+                            dataPredicate,
                             dataFilePathFactory,
                             file);
             if (!fileIndexResult.remain()) {
diff --git a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/fileindex/FileIndexTest.scala b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/fileindex/FileIndexTest.scala
@@ -0,0 +1,83 @@
+/*
+ * 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.spark.fileindex
+
+import org.apache.paimon.operation.AppendOnlyFileStoreScan
+import org.apache.paimon.predicate.PredicateBuilder
+import org.apache.paimon.spark.PaimonSparkTestBase
+import org.apache.paimon.table.FileStoreTable
+
+import scala.jdk.CollectionConverters._
+
+class FileIndexTest extends PaimonSparkTestBase {
+
+  test("test file index in plan phase") {
+    Seq("bloom-filter", "bitmap", "bsi").foreach {
+      filter =>
+        Seq("1B", "1MB").foreach {
+          size =>
+            val fileIndexProp = if (filter == "bloom-filter") {
+              s" 'file-index.bloom-filter.columns' = 'id'"
+            } else if (filter.equals("bitmap")) {
+              s" 'file-index.bitmap.columns' = 'id'"
+            } else if (filter.equals("bsi")) {
+              s" 'file-index.bsi.columns' = 'id'"
+            }
+
+            withTable("T") {
+              spark.sql(s"""
+                           |create table T (
+                           |id int,
+                           |name string)
+                           |USING paimon
+                           |TBLPROPERTIES(
+                           |  $fileIndexProp,
+                           |  'file-index.in-manifest-threshold'= '$size')
+                           |""".stripMargin)
+
+              spark.sql("insert into T values(1,'a')")
+              spark.sql("insert into T values(2,'b')")
+              spark.sql("insert into T values(3,'c')")
+              assert(sql("select * from `T$files`").collect().length == 3)
+
+              val fileNames = spark.sql("select input_file_name() from T where id == 2 ").collect()
+              assert(fileNames.length == 1)
+              val location = fileNames(0).getString(0)
+              val fileName = location.substring(location.lastIndexOf('/') + 1)
+
+              val table: FileStoreTable = loadTable("T")
+              val predicateBuilder = new PredicateBuilder(table.rowType)
+              // predicate is 'id=2'
+              val predicate = predicateBuilder.equal(0, 2)
+              val files = table
+                .store()
+                .newScan()
+                .asInstanceOf[AppendOnlyFileStoreScan]
+                .withFilter(predicate)
+                .plan()
+                .files()
+                .asScala
+                .map(x => x.fileName())
+              assert(files.length == 1)
+              assert(fileName.equals(files.head))
+            }
+        }
+    }
+  }
+}

Original file line number	Diff line number	Diff line change
`@@ -155,6 +155,8 @@ public AppendOnlyFileStoreScan newScan() {`
`155`	`155`	`schemaManager,`
`156`	`156`	`schema,`
`157`	`157`	`manifestFileFactory(),`
	`158`	`+ fileIO,`
	`159`	`+ pathFactory(),`
`158`	`160`	`options.scanManifestParallelism(),`
`159`	`161`	`options.fileIndexReadEnabled());`
`160`	`162`	`}`