[core] Allow FileIndexer to seek and skip if they need to. (#3406)

apache · Jun 17, 2024 · 773b7b5 · 773b7b5
1 parent 61d1945
commit 773b7b5
Show file tree

Hide file tree

Showing 4 changed files with 28 additions and 6 deletions.
diff --git a/paimon-common/src/main/java/org/apache/paimon/fileindex/FileIndexFormat.java b/paimon-common/src/main/java/org/apache/paimon/fileindex/FileIndexFormat.java
@@ -309,7 +309,10 @@ private FileIndexReader getFileIndexReader(
                             indexType,
                             FileIndexCommon.getFieldType(fields, columnName),
                             new Options())
-                    .createReader(getBytesWithStartAndLength(startAndLength));
+                    .createReader(
+                            seekableInputStream,
+                            startAndLength.getLeft(),
+                            startAndLength.getRight());
         }
 
         private byte[] getBytesWithStartAndLength(Pair<Integer, Integer> startAndLength) {

diff --git a/paimon-common/src/main/java/org/apache/paimon/fileindex/FileIndexer.java b/paimon-common/src/main/java/org/apache/paimon/fileindex/FileIndexer.java
@@ -18,6 +18,7 @@
 
 package org.apache.paimon.fileindex;
 
+import org.apache.paimon.fs.SeekableInputStream;
 import org.apache.paimon.options.Options;
 import org.apache.paimon.types.DataType;
 
@@ -31,7 +32,7 @@ public interface FileIndexer {
 
     FileIndexWriter createWriter();
 
-    FileIndexReader createReader(byte[] serializedBytes);
+    FileIndexReader createReader(SeekableInputStream inputStream, int start, int length);
 
     static FileIndexer create(String type, DataType dataType, Options options) {
         FileIndexerFactory fileIndexerFactory = FileIndexerFactoryUtils.load(type);

diff --git a/...on-common/src/main/java/org/apache/paimon/fileindex/bloomfilter/BloomFilterFileIndex.java b/...on-common/src/main/java/org/apache/paimon/fileindex/bloomfilter/BloomFilterFileIndex.java
@@ -22,14 +22,18 @@
 import org.apache.paimon.fileindex.FileIndexResult;
 import org.apache.paimon.fileindex.FileIndexWriter;
 import org.apache.paimon.fileindex.FileIndexer;
+import org.apache.paimon.fs.SeekableInputStream;
 import org.apache.paimon.options.Options;
 import org.apache.paimon.predicate.FieldRef;
 import org.apache.paimon.types.DataType;
 import org.apache.paimon.utils.BloomFilter64;
 import org.apache.paimon.utils.BloomFilter64.BitSet;
+import org.apache.paimon.utils.IOUtils;
 
 import org.apache.hadoop.util.bloom.HashFunction;
 
+import java.io.IOException;
+
 import static org.apache.paimon.fileindex.FileIndexResult.REMAIN;
 import static org.apache.paimon.fileindex.FileIndexResult.SKIP;
 
@@ -65,8 +69,15 @@ public FileIndexWriter createWriter() {
     }
 
     @Override
-    public FileIndexReader createReader(byte[] serializedBytes) {
-        return new Reader(dataType, serializedBytes);
+    public FileIndexReader createReader(SeekableInputStream inputStream, int start, int length) {
+        try {
+            inputStream.seek(start);
+            byte[] serializedBytes = new byte[length];
+            IOUtils.readFully(inputStream, serializedBytes);
+            return new Reader(dataType, serializedBytes);
+        } catch (IOException e) {
+            throw new RuntimeException(e);
+        }
     }
 
     private static class Writer extends FileIndexWriter {

diff --git a/...ommon/src/test/java/org/apache/paimon/fileindex/bloomfilter/BloomFilterFileIndexTest.java b/...ommon/src/test/java/org/apache/paimon/fileindex/bloomfilter/BloomFilterFileIndexTest.java
@@ -20,6 +20,7 @@
 
 import org.apache.paimon.fileindex.FileIndexReader;
 import org.apache.paimon.fileindex.FileIndexWriter;
+import org.apache.paimon.fs.ByteArraySeekableStream;
 import org.apache.paimon.options.Options;
 import org.apache.paimon.types.DataTypes;
 
@@ -60,7 +61,10 @@ public void testAddFindByRandom() {
 
         testData.forEach(writer::write);
 
-        FileIndexReader reader = filter.createReader(writer.serializedBytes());
+        byte[] serializedBytes = writer.serializedBytes();
+        FileIndexReader reader =
+                filter.createReader(
+                        new ByteArraySeekableStream(serializedBytes), 0, serializedBytes.length);
 
         for (byte[] bytes : testData) {
             Assertions.assertThat(reader.visitEqual(null, bytes).remain()).isTrue();
@@ -100,7 +104,10 @@ public void testAddFindByRandomLong() {
 
         testData.forEach(writer::write);
 
-        FileIndexReader reader = filter.createReader(writer.serializedBytes());
+        byte[] serializedBytes = writer.serializedBytes();
+        FileIndexReader reader =
+                filter.createReader(
+                        new ByteArraySeekableStream(serializedBytes), 0, serializedBytes.length);
 
         for (Long value : testData) {
             Assertions.assertThat(reader.visitEqual(null, value).remain()).isTrue();