[common][core][format] Support using file index result for RowGroup f…

…iltering.
apache · Nov 7, 2024 · c41fafa · c41fafa
1 parent 8e4de02
commit c41fafa
Show file tree

Hide file tree

Showing 14 changed files with 2,084 additions and 31 deletions.
diff --git a/paimon-common/src/main/java/org/apache/paimon/fileindex/FileIndexPredicate.java b/paimon-common/src/main/java/org/apache/paimon/fileindex/FileIndexPredicate.java
@@ -67,6 +67,16 @@ public FileIndexPredicate(SeekableInputStream inputStream, RowType fileRowType)
         this.reader = FileIndexFormat.createReader(inputStream, fileRowType);
     }
 
+    public FileIndexResult evaluate(@Nullable Predicate predicate) {
+        if (predicate == null) {
+            return REMAIN;
+        }
+        Set<String> requiredFieldNames = getRequiredNames(predicate);
+        Map<String, Collection<FileIndexReader>> indexReaders = new HashMap<>();
+        requiredFieldNames.forEach(name -> indexReaders.put(name, reader.readColumnIndex(name)));
+        return new FileIndexPredicateTest(indexReaders).test(predicate);
+    }
+
     public boolean testPredicate(@Nullable Predicate filePredicate) {
         if (filePredicate == null) {
             return true;

diff --git a/paimon-common/src/main/java/org/apache/paimon/format/FormatReaderContext.java b/paimon-common/src/main/java/org/apache/paimon/format/FormatReaderContext.java
@@ -18,6 +18,7 @@
 
 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;
@@ -28,13 +29,18 @@ public class FormatReaderContext implements FormatReaderFactory.Context {
     private final FileIO fileIO;
     private final Path file;
     private final long fileSize;
+    private FileIndexResult fileIndexResult;
 
     public FormatReaderContext(FileIO fileIO, Path file, long fileSize) {
         this.fileIO = fileIO;
         this.file = file;
         this.fileSize = fileSize;
     }
 
+    public void setFileIndexResult(FileIndexResult fileIndexResult) {
+        this.fileIndexResult = fileIndexResult;
+    }
+
     @Override
     public FileIO fileIO() {
         return fileIO;
@@ -49,4 +55,9 @@ public Path filePath() {
     public long fileSize() {
         return fileSize;
     }
+
+    @Override
+    public FileIndexResult fileIndex() {
+        return fileIndexResult;
+    }
 }
diff --git a/paimon-common/src/main/java/org/apache/paimon/format/FormatReaderFactory.java b/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();
     }
 }
diff --git a/paimon-common/src/main/java/org/apache/paimon/reader/RecordReader.java b/paimon-common/src/main/java/org/apache/paimon/reader/RecordReader.java
@@ -19,8 +19,11 @@
 package org.apache.paimon.reader;
 
 import org.apache.paimon.annotation.Public;
+import org.apache.paimon.fileindex.FileIndexResult;
+import org.apache.paimon.fileindex.bitmap.BitmapIndexResultLazy;
 import org.apache.paimon.utils.CloseableIterator;
 import org.apache.paimon.utils.Filter;
+import org.apache.paimon.utils.RoaringBitmap32;
 
 import javax.annotation.Nullable;
 
@@ -209,4 +212,13 @@ public void close() throws IOException {
     default CloseableIterator<T> toCloseableIterator() {
         return new RecordReaderIterator<>(this);
     }
+
+    default boolean checkIndexResultExist(
+            FileIndexResult fileIndexResult, long startRowNum, long endRowNum) {
+        if (fileIndexResult instanceof BitmapIndexResultLazy) {
+            RoaringBitmap32 bitmap = ((BitmapIndexResultLazy) fileIndexResult).get();
+            return bitmap.rangeCardinality(startRowNum, endRowNum) > 0;
+        }
+        return true;
+    }
 }
diff --git a/paimon-common/src/main/java/org/apache/paimon/utils/RoaringBitmap32.java b/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 void serialize(DataOutput out) throws IOException {
         roaringBitmap.runOptimize();
         roaringBitmap.serialize(out);

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
@@ -0,0 +1,66 @@
+/*
+ * 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.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;
+import org.apache.paimon.schema.TableSchema;
+
+import java.io.IOException;
+import java.util.List;
+import java.util.stream.Collectors;
+
+/** Evaluate file index result. */
+public class FileIndexEvaluator {
+
+    public static FileIndexResult evaluate(
+            FileIO fileIO,
+            TableSchema dataSchema,
+            List<Predicate> dataFilter,
+            DataFilePathFactory dataFilePathFactory,
+            DataFileMeta file)
+            throws IOException {
+        if (dataFilter != null && !dataFilter.isEmpty()) {
+            List<String> indexFiles =
+                    file.extraFiles().stream()
+                            .filter(name -> name.endsWith(DataFilePathFactory.INDEX_PATH_SUFFIX))
+                            .collect(Collectors.toList());
+            if (!indexFiles.isEmpty()) {
+                if (indexFiles.size() > 1) {
+                    throw new RuntimeException(
+                            "Found more than one index file for one data file: "
+                                    + String.join(" and ", indexFiles));
+                }
+                // go to file index check
+                try (FileIndexPredicate predicate =
+                        new FileIndexPredicate(
+                                dataFilePathFactory.toPath(indexFiles.get(0)),
+                                fileIO,
+                                dataSchema.logicalRowType())) {
+                    return predicate.evaluate(
+                            PredicateBuilder.and(dataFilter.toArray(new Predicate[0])));
+                }
+            }
+        }
+        return FileIndexResult.REMAIN;
+    }
+}
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
@@ -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,38 @@ private RecordReader<InternalRow> createFileReader(
             BulkFormatMapping bulkFormatMapping,
             IOExceptionSupplier<DeletionVector> dvFactory)
             throws IOException {
+        FileIndexResult fileIndexResult = null;
         if (fileIndexReadEnabled) {
-            boolean skip =
-                    FileIndexSkipper.skip(
+            //            boolean skip =
+            //                    FileIndexSkipper.skip(
+            //                            fileIO,
+            //                            bulkFormatMapping.getDataSchema(),
+            //                            bulkFormatMapping.getDataFilters(),
+            //                            dataFilePathFactory,
+            //                            file);
+            //            if (skip) {
+            //                return new EmptyRecordReader<>();
+            //            }
+            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());
+        formatReaderContext.setFileIndexResult(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));