apache · yuzelin · Nov 13, 2024 · Nov 13, 2024
diff --git a/...common/src/main/java/org/apache/paimon/fileindex/bitmap/ApplyBitmapIndexRecordReader.java b/...common/src/main/java/org/apache/paimon/fileindex/bitmap/ApplyBitmapIndexRecordReader.java
@@ -20,41 +20,35 @@
 
 import org.apache.paimon.data.InternalRow;
 import org.apache.paimon.reader.FileRecordIterator;
+import org.apache.paimon.reader.FileRecordReader;
 import org.apache.paimon.reader.RecordReader;
 
 import javax.annotation.Nullable;
 
 import java.io.IOException;
 
-import static org.apache.paimon.utils.Preconditions.checkArgument;
-
 /** A {@link RecordReader} which apply {@link BitmapIndexResult} to filter record. */
-public class ApplyBitmapIndexRecordReader implements RecordReader<InternalRow> {
+public class ApplyBitmapIndexRecordReader implements FileRecordReader<InternalRow> {
 
-    private final RecordReader<InternalRow> reader;
+    private final FileRecordReader<InternalRow> reader;
 
     private final BitmapIndexResult fileIndexResult;
 
     public ApplyBitmapIndexRecordReader(
-            RecordReader<InternalRow> reader, BitmapIndexResult fileIndexResult) {
+            FileRecordReader<InternalRow> reader, BitmapIndexResult fileIndexResult) {
         this.reader = reader;
         this.fileIndexResult = fileIndexResult;
     }
 
     @Nullable
     @Override
-    public RecordIterator<InternalRow> readBatch() throws IOException {
-        RecordIterator<InternalRow> batch = reader.readBatch();
+    public FileRecordIterator<InternalRow> readBatch() throws IOException {
+        FileRecordIterator<InternalRow> batch = reader.readBatch();
         if (batch == null) {
             return null;
         }
 
-        checkArgument(
-                batch instanceof FileRecordIterator,
-                "There is a bug, RecordIterator in ApplyBitmapIndexRecordReader must be FileRecordIterator");
-
-        return new ApplyBitmapIndexFileRecordIterator(
-                (FileRecordIterator<InternalRow>) batch, fileIndexResult);
+        return new ApplyBitmapIndexFileRecordIterator(batch, fileIndexResult);
     }
 
     @Override

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
@@ -22,14 +22,15 @@
 import org.apache.paimon.fileindex.FileIndexResult;
 import org.apache.paimon.fs.FileIO;
 import org.apache.paimon.fs.Path;
+import org.apache.paimon.reader.FileRecordReader;
 import org.apache.paimon.reader.RecordReader;
 
 import java.io.IOException;
 
 /** A factory to create {@link RecordReader} for file. */
 public interface FormatReaderFactory {
 
-    RecordReader<InternalRow> createReader(Context context) throws IOException;
+    FileRecordReader<InternalRow> createReader(Context context) throws IOException;
 
     /** Context for creating reader. */
     interface Context {

diff --git a/paimon-common/src/main/java/org/apache/paimon/reader/EmptyFileRecordReader.java b/paimon-common/src/main/java/org/apache/paimon/reader/EmptyFileRecordReader.java
@@ -0,0 +1,36 @@
+/*
+ * 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.reader;
+
+import javax.annotation.Nullable;
+
+import java.io.IOException;
+
+/** An empty {@link FileRecordReader}. */
+public class EmptyFileRecordReader<T> implements FileRecordReader<T> {
+
+    @Nullable
+    @Override
+    public FileRecordIterator<T> readBatch() throws IOException {
+        return null;
+    }
+
+    @Override
+    public void close() throws IOException {}
+}
diff --git a/paimon-common/src/main/java/org/apache/paimon/reader/FileRecordIterator.java b/paimon-common/src/main/java/org/apache/paimon/reader/FileRecordIterator.java
@@ -27,10 +27,8 @@
 import java.util.function.Function;
 
 /**
- * Wrap {@link RecordReader.RecordIterator} to support returning the record's row position and file
+ * A {@link RecordReader.RecordIterator} to support returning the record's row position and file
  * Path.
- *
- * @param <T> The type of the record.
  */
 public interface FileRecordIterator<T> extends RecordReader.RecordIterator<T> {
 

diff --git a/paimon-common/src/main/java/org/apache/paimon/reader/FileRecordReader.java b/paimon-common/src/main/java/org/apache/paimon/reader/FileRecordReader.java
@@ -0,0 +1,31 @@
+/*
+ * 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.reader;
+
+import javax.annotation.Nullable;
+
+import java.io.IOException;
+
+/** A {@link RecordReader} to support returning {@link FileRecordIterator}. */
+public interface FileRecordReader<T> extends RecordReader<T> {
+
+    @Override
+    @Nullable
+    FileRecordIterator<T> readBatch() throws IOException;
+}
diff --git a/paimon-core/src/main/java/org/apache/paimon/deletionvectors/ApplyDeletionVectorReader.java b/paimon-core/src/main/java/org/apache/paimon/deletionvectors/ApplyDeletionVectorReader.java
@@ -20,23 +20,22 @@
 
 import org.apache.paimon.data.InternalRow;
 import org.apache.paimon.reader.FileRecordIterator;
+import org.apache.paimon.reader.FileRecordReader;
 import org.apache.paimon.reader.RecordReader;
 
 import javax.annotation.Nullable;
 
 import java.io.IOException;
 
-import static org.apache.paimon.utils.Preconditions.checkArgument;
-
 /** A {@link RecordReader} which apply {@link DeletionVector} to filter record. */
-public class ApplyDeletionVectorReader implements RecordReader<InternalRow> {
+public class ApplyDeletionVectorReader implements FileRecordReader<InternalRow> {
 
-    private final RecordReader<InternalRow> reader;
+    private final FileRecordReader<InternalRow> reader;
 
     private final DeletionVector deletionVector;
 
     public ApplyDeletionVectorReader(
-            RecordReader<InternalRow> reader, DeletionVector deletionVector) {
+            FileRecordReader<InternalRow> reader, DeletionVector deletionVector) {
         this.reader = reader;
         this.deletionVector = deletionVector;
     }
@@ -51,19 +50,14 @@ public DeletionVector deletionVector() {
 
     @Nullable
     @Override
-    public RecordIterator<InternalRow> readBatch() throws IOException {
-        RecordIterator<InternalRow> batch = reader.readBatch();
+    public FileRecordIterator<InternalRow> readBatch() throws IOException {
+        FileRecordIterator<InternalRow> batch = reader.readBatch();
 
         if (batch == null) {
             return null;
         }
 
-        checkArgument(
-                batch instanceof FileRecordIterator,
-                "There is a bug, RecordIterator in ApplyDeletionVectorReader must be FileRecordIterator");
-
-        return new ApplyDeletionFileRecordIterator(
-                (FileRecordIterator<InternalRow>) batch, deletionVector);
+        return new ApplyDeletionFileRecordIterator(batch, deletionVector);
     }
 
     @Override

diff --git a/...rg/apache/paimon/io/FileRecordReader.java → ...pache/paimon/io/DataFileRecordReader.java b/...rg/apache/paimon/io/FileRecordReader.java → ...pache/paimon/io/DataFileRecordReader.java
@@ -25,7 +25,8 @@
 import org.apache.paimon.data.PartitionInfo;
 import org.apache.paimon.data.columnar.ColumnarRowIterator;
 import org.apache.paimon.format.FormatReaderFactory;
-import org.apache.paimon.reader.RecordReader;
+import org.apache.paimon.reader.FileRecordIterator;
+import org.apache.paimon.reader.FileRecordReader;
 import org.apache.paimon.utils.FileUtils;
 import org.apache.paimon.utils.ProjectedRow;
 
@@ -34,17 +35,35 @@
 import java.io.IOException;
 
 /** Reads {@link InternalRow} from data files. */
-public class FileRecordReader implements RecordReader<InternalRow> {
+public class DataFileRecordReader implements FileRecordReader<InternalRow> {
 
-    private final RecordReader<InternalRow> reader;
+    private final FileRecordReader<InternalRow> reader;
     @Nullable private final int[] indexMapping;
     @Nullable private final PartitionInfo partitionInfo;
     @Nullable private final CastFieldGetter[] castMapping;
 
+    public DataFileRecordReader(
+            FormatReaderFactory readerFactory,
+            FormatReaderFactory.Context context,
+            @Nullable int[] indexMapping,
+            @Nullable CastFieldGetter[] castMapping,
+            @Nullable PartitionInfo partitionInfo)
+            throws IOException {
+        try {
+            this.reader = readerFactory.createReader(context);
+        } catch (Exception e) {
+            FileUtils.checkExists(context.fileIO(), context.filePath());
+            throw e;
+        }
+        this.indexMapping = indexMapping;
+        this.partitionInfo = partitionInfo;
+        this.castMapping = castMapping;
+    }
+
     @Nullable
     @Override
-    public RecordReader.RecordIterator<InternalRow> readBatch() throws IOException {
-        RecordIterator<InternalRow> iterator = reader.readBatch();
+    public FileRecordIterator<InternalRow> readBatch() throws IOException {
+        FileRecordIterator<InternalRow> iterator = reader.readBatch();
         if (iterator == null) {
             return null;
         }
@@ -71,24 +90,6 @@ public RecordReader.RecordIterator<InternalRow> readBatch() throws IOException {
         return iterator;
     }
 
-    public FileRecordReader(
-            FormatReaderFactory readerFactory,
-            FormatReaderFactory.Context context,
-            @Nullable int[] indexMapping,
-            @Nullable CastFieldGetter[] castMapping,
-            @Nullable PartitionInfo partitionInfo)
-            throws IOException {
-        try {
-            this.reader = readerFactory.createReader(context);
-        } catch (Exception e) {
-            FileUtils.checkExists(context.fileIO(), context.filePath());
-            throw e;
-        }
-        this.indexMapping = indexMapping;
-        this.partitionInfo = partitionInfo;
-        this.castMapping = castMapping;
-    }
-
     @Override
     public void close() throws IOException {
         reader.close();

diff --git a/paimon-core/src/main/java/org/apache/paimon/io/KeyValueDataFileRecordReader.java b/paimon-core/src/main/java/org/apache/paimon/io/KeyValueDataFileRecordReader.java
@@ -21,6 +21,8 @@
 import org.apache.paimon.KeyValue;
 import org.apache.paimon.KeyValueSerializer;
 import org.apache.paimon.data.InternalRow;
+import org.apache.paimon.reader.FileRecordIterator;
+import org.apache.paimon.reader.FileRecordReader;
 import org.apache.paimon.reader.RecordReader;
 import org.apache.paimon.types.RowType;
 
@@ -29,23 +31,23 @@
 import java.io.IOException;
 
 /** {@link RecordReader} for reading {@link KeyValue} data files. */
-public class KeyValueDataFileRecordReader implements RecordReader<KeyValue> {
+public class KeyValueDataFileRecordReader implements FileRecordReader<KeyValue> {
 
-    private final RecordReader<InternalRow> reader;
+    private final FileRecordReader<InternalRow> reader;
     private final KeyValueSerializer serializer;
     private final int level;
 
     public KeyValueDataFileRecordReader(
-            RecordReader<InternalRow> reader, RowType keyType, RowType valueType, int level) {
+            FileRecordReader<InternalRow> reader, RowType keyType, RowType valueType, int level) {
         this.reader = reader;
         this.serializer = new KeyValueSerializer(keyType, valueType);
         this.level = level;
     }
 
     @Nullable
     @Override
-    public RecordIterator<KeyValue> readBatch() throws IOException {
-        RecordReader.RecordIterator<InternalRow> iterator = reader.readBatch();
+    public FileRecordIterator<KeyValue> readBatch() throws IOException {
+        FileRecordIterator<InternalRow> iterator = reader.readBatch();
         if (iterator == null) {
             return null;
         }

diff --git a/paimon-core/src/main/java/org/apache/paimon/io/KeyValueFileReaderFactory.java b/paimon-core/src/main/java/org/apache/paimon/io/KeyValueFileReaderFactory.java
@@ -32,6 +32,7 @@
 import org.apache.paimon.fs.Path;
 import org.apache.paimon.partition.PartitionUtils;
 import org.apache.paimon.predicate.Predicate;
+import org.apache.paimon.reader.FileRecordReader;
 import org.apache.paimon.reader.RecordReader;
 import org.apache.paimon.schema.KeyValueFieldsExtractor;
 import org.apache.paimon.schema.SchemaManager;
@@ -109,7 +110,7 @@ public RecordReader<KeyValue> createRecordReader(
         return createRecordReader(schemaId, fileName, level, true, null, fileSize);
     }
 
-    private RecordReader<KeyValue> createRecordReader(
+    private FileRecordReader<KeyValue> createRecordReader(
             long schemaId,
             String fileName,
             int level,
@@ -134,8 +135,8 @@ private RecordReader<KeyValue> createRecordReader(
                         : formatSupplier.get();
         Path filePath = pathFactory.toPath(fileName);
 
-        RecordReader<InternalRow> fileRecordReader =
-                new FileRecordReader(
+        FileRecordReader<InternalRow> fileRecordReader =
+                new DataFileRecordReader(
                         bulkFormatMapping.getReaderFactory(),
                         orcPoolSize == null
                                 ? new FormatReaderContext(fileIO, filePath, fileSize)

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
@@ -32,12 +32,13 @@
 import org.apache.paimon.fs.FileIO;
 import org.apache.paimon.io.DataFileMeta;
 import org.apache.paimon.io.DataFilePathFactory;
+import org.apache.paimon.io.DataFileRecordReader;
 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;
 import org.apache.paimon.predicate.Predicate;
-import org.apache.paimon.reader.EmptyRecordReader;
+import org.apache.paimon.reader.EmptyFileRecordReader;
+import org.apache.paimon.reader.FileRecordReader;
 import org.apache.paimon.reader.ReaderSupplier;
 import org.apache.paimon.reader.RecordReader;
 import org.apache.paimon.schema.SchemaManager;
@@ -187,7 +188,7 @@ public RecordReader<InternalRow> createReader(
         return ConcatRecordReader.create(suppliers);
     }
 
-    private RecordReader<InternalRow> createFileReader(
+    private FileRecordReader<InternalRow> createFileReader(
             BinaryRow partition,
             DataFileMeta file,
             DataFilePathFactory dataFilePathFactory,
@@ -204,7 +205,7 @@ private RecordReader<InternalRow> createFileReader(
                             dataFilePathFactory,
                             file);
             if (!fileIndexResult.remain()) {
-                return new EmptyRecordReader<>();
+                return new EmptyFileRecordReader<>();
             }
         }
 
@@ -214,8 +215,8 @@ private RecordReader<InternalRow> createFileReader(
                         dataFilePathFactory.toPath(file.fileName()),
                         file.fileSize(),
                         fileIndexResult);
-        RecordReader<InternalRow> fileRecordReader =
-                new FileRecordReader(
+        FileRecordReader<InternalRow> fileRecordReader =
+                new DataFileRecordReader(
                         bulkFormatMapping.getReaderFactory(),
                         formatReaderContext,
                         bulkFormatMapping.getIndexMapping(),