Parquet: Use native getRowIndexOffset support instead of calculating it

arrow/src/main/java/org/apache/iceberg/arrow/vectorized/BaseBatchReader.java

@@ -42,9 +42,15 @@ protected BaseBatchReader(List<VectorizedReader<?>> readers) {
   @Override
   public void setRowGroupInfo(
       PageReadStore pageStore, Map<ColumnPath, ColumnChunkMetaData> metaData, long rowPosition) {
+    setRowGroupInfo(pageStore, metaData);
+  }
+
+  @Override
+  public void setRowGroupInfo(
+      PageReadStore pageStore, Map<ColumnPath, ColumnChunkMetaData> metaData) {
     for (VectorizedArrowReader reader : readers) {
       if (reader != null) {
-        reader.setRowGroupInfo(pageStore, metaData, rowPosition);
+        reader.setRowGroupInfo(pageStore, metaData);
       }
     }
   }

arrow/src/main/java/org/apache/iceberg/arrow/vectorized/VectorizedArrowReader.java

@@ -432,6 +432,11 @@ private void allocateVectorBasedOnTypeName(PrimitiveType primitive, Field arrowF
   @Override
   public void setRowGroupInfo(
       PageReadStore source, Map<ColumnPath, ColumnChunkMetaData> metadata, long rowPosition) {
+    setRowGroupInfo(source, metadata);
+  }
+
+  @Override
+  public void setRowGroupInfo(PageReadStore source, Map<ColumnPath, ColumnChunkMetaData> metadata) {
     ColumnChunkMetaData chunkMetaData = metadata.get(ColumnPath.get(columnDescriptor.getPath()));
     this.dictionary =
         vectorizedColumnIterator.setRowGroupInfo(
@@ -475,6 +480,10 @@ public VectorHolder read(VectorHolder reuse, int numValsToRead) {
     public void setRowGroupInfo(
         PageReadStore source, Map<ColumnPath, ColumnChunkMetaData> metadata, long rowPosition) {}
 
+    @Override
+    public void setRowGroupInfo(
+        PageReadStore source, Map<ColumnPath, ColumnChunkMetaData> metadata) {}
+
     @Override
     public String toString() {
       return "NullReader";
@@ -541,7 +550,13 @@ private static NullabilityHolder newNullabilityHolder(int size) {
     @Override
     public void setRowGroupInfo(
         PageReadStore source, Map<ColumnPath, ColumnChunkMetaData> metadata, long rowPosition) {
-      this.rowStart = rowPosition;
+      setRowGroupInfo(source, metadata);
+    }
+
+    @Override
+    public void setRowGroupInfo(
+        PageReadStore source, Map<ColumnPath, ColumnChunkMetaData> metadata) {
+      this.rowStart = source.getRowIndexOffset().orElse(0L);
     }
 
     @Override
@@ -586,6 +601,10 @@ public VectorHolder read(VectorHolder reuse, int numValsToRead) {
     public void setRowGroupInfo(
         PageReadStore source, Map<ColumnPath, ColumnChunkMetaData> metadata, long rowPosition) {}
 
+    @Override
+    public void setRowGroupInfo(
+        PageReadStore source, Map<ColumnPath, ColumnChunkMetaData> metadata) {}
+
     @Override
     public String toString() {
       return String.format("ConstantReader: %s", value);
@@ -613,6 +632,10 @@ public VectorHolder read(VectorHolder reuse, int numValsToRead) {
     public void setRowGroupInfo(
         PageReadStore source, Map<ColumnPath, ColumnChunkMetaData> metadata, long rowPosition) {}
 
+    @Override
+    public void setRowGroupInfo(
+        PageReadStore source, Map<ColumnPath, ColumnChunkMetaData> metadata) {}
+
     @Override
     public String toString() {
       return "DeletedVectorReader";

parquet/src/main/java/org/apache/iceberg/parquet/ParquetReader.java

@@ -99,7 +99,6 @@ private static class FileIterator<T> implements CloseableIterator<T> {
     private final ParquetValueReader<T> model;
     private final long totalValues;
     private final boolean reuseContainers;
-    private final long[] rowGroupsStartRowPos;
 
     private int nextRowGroup = 0;
     private long nextRowGroupStart = 0;
@@ -112,7 +111,6 @@ private static class FileIterator<T> implements CloseableIterator<T> {
       this.model = conf.model();
       this.totalValues = conf.totalValues();
       this.reuseContainers = conf.reuseContainers();
-      this.rowGroupsStartRowPos = conf.startRowPositions();
     }
 
     @Override
@@ -149,11 +147,10 @@ private void advance() {
         throw new RuntimeIOException(e);
       }
 
-      long rowPosition = rowGroupsStartRowPos[nextRowGroup];
       nextRowGroupStart += pages.getRowCount();
       nextRowGroup += 1;
 
-      model.setPageSource(pages, rowPosition);
+      model.setPageSource(pages);
     }
 
     @Override

parquet/src/main/java/org/apache/iceberg/parquet/ParquetValueReader.java

@@ -28,5 +28,14 @@ public interface ParquetValueReader<T> {
 
   List<TripleIterator<?>> columns();
 
+  /**
+   * @deprecated since 1.6.0, will be removed in 1.7.0; use setPageSource(PageReadStore) instead.
+   */
+  @Deprecated
   void setPageSource(PageReadStore pageStore, long rowPosition);
+
+  default void setPageSource(PageReadStore pageStore) {
+    throw new UnsupportedOperationException(
+        this.getClass().getName() + " doesn't implement setPageSource(PageReadStore)");
+  }
 }

parquet/src/main/java/org/apache/iceberg/parquet/ParquetValueReaders.java

@@ -113,6 +113,9 @@ public List<TripleIterator<?>> columns() {
 
     @Override
     public void setPageSource(PageReadStore pageStore, long rowPosition) {}
+
+    @Override
+    public void setPageSource(PageReadStore pageStore) {}
   }
 
   static class ConstantReader<C> implements ParquetValueReader<C> {
@@ -176,6 +179,9 @@ public List<TripleIterator<?>> columns() {
 
     @Override
     public void setPageSource(PageReadStore pageStore, long rowPosition) {}
+
+    @Override
+    public void setPageSource(PageReadStore pageStore) {}
   }
 
   static class PositionReader implements ParquetValueReader<Long> {
@@ -200,7 +206,12 @@ public List<TripleIterator<?>> columns() {
 
     @Override
     public void setPageSource(PageReadStore pageStore, long rowPosition) {
-      this.rowGroupStart = rowPosition;
+      setPageSource(pageStore);
+    }
+
+    @Override
+    public void setPageSource(PageReadStore pageStore) {
+      this.rowGroupStart = pageStore.getRowIndexOffset().orElse(0L);
       this.rowOffset = -1;
     }
   }
@@ -221,6 +232,11 @@ protected PrimitiveReader(ColumnDescriptor desc) {
 
     @Override
     public void setPageSource(PageReadStore pageStore, long rowPosition) {
+      setPageSource(pageStore);
+    }
+
+    @Override
+    public void setPageSource(PageReadStore pageStore) {
       column.setPageSource(pageStore.getPageReader(desc));
     }
 
@@ -405,7 +421,12 @@ private static class OptionReader<T> implements ParquetValueReader<T> {
 
     @Override
     public void setPageSource(PageReadStore pageStore, long rowPosition) {
-      reader.setPageSource(pageStore, rowPosition);
+      setPageSource(pageStore);
+    }
+
+    @Override
+    public void setPageSource(PageReadStore pageStore) {
+      reader.setPageSource(pageStore);
     }
 
     @Override
@@ -450,7 +471,12 @@ protected RepeatedReader(
 
     @Override
     public void setPageSource(PageReadStore pageStore, long rowPosition) {
-      reader.setPageSource(pageStore, rowPosition);
+      setPageSource(pageStore);
+    }
+
+    @Override
+    public void setPageSource(PageReadStore pageStore) {
+      reader.setPageSource(pageStore);
     }
 
     @Override
@@ -569,8 +595,13 @@ protected RepeatedKeyValueReader(
 
     @Override
     public void setPageSource(PageReadStore pageStore, long rowPosition) {
-      keyReader.setPageSource(pageStore, rowPosition);
-      valueReader.setPageSource(pageStore, rowPosition);
+      setPageSource(pageStore);
+    }
+
+    @Override
+    public void setPageSource(PageReadStore pageStore) {
+      keyReader.setPageSource(pageStore);
+      valueReader.setPageSource(pageStore);
     }
 
     @Override
@@ -720,8 +751,13 @@ protected StructReader(List<Type> types, List<ParquetValueReader<?>> readers) {
 
     @Override
     public final void setPageSource(PageReadStore pageStore, long rowPosition) {
+      setPageSource(pageStore);
+    }
+
+    @Override
+    public final void setPageSource(PageReadStore pageStore) {
       for (ParquetValueReader<?> reader : readers) {
-        reader.setPageSource(pageStore, rowPosition);
+        reader.setPageSource(pageStore);
       }
     }
 

parquet/src/main/java/org/apache/iceberg/parquet/ReadConf.java

@@ -19,23 +19,19 @@
 package org.apache.iceberg.parquet;
 
 import java.io.IOException;
-import java.io.UncheckedIOException;
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
 import java.util.function.Function;
 import java.util.stream.Collectors;
-import org.apache.iceberg.MetadataColumns;
 import org.apache.iceberg.Schema;
 import org.apache.iceberg.exceptions.RuntimeIOException;
 import org.apache.iceberg.expressions.Expression;
 import org.apache.iceberg.io.InputFile;
 import org.apache.iceberg.mapping.NameMapping;
 import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
 import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
-import org.apache.iceberg.relocated.com.google.common.collect.Maps;
 import org.apache.parquet.ParquetReadOptions;
-import org.apache.parquet.crypto.FileDecryptionProperties;
 import org.apache.parquet.hadoop.ParquetFileReader;
 import org.apache.parquet.hadoop.metadata.BlockMetaData;
 import org.apache.parquet.hadoop.metadata.ColumnChunkMetaData;
@@ -59,7 +55,6 @@ class ReadConf<T> {
   private final long totalValues;
   private final boolean reuseContainers;
   private final Integer batchSize;
-  private final long[] startRowPositions;
 
   // List of column chunk metadata for each row group
   private final List<Map<ColumnPath, ColumnChunkMetaData>> columnChunkMetaDataForRowGroups;
@@ -95,10 +90,6 @@ class ReadConf<T> {
 
     this.rowGroups = reader.getRowGroups();
     this.shouldSkip = new boolean[rowGroups.size()];
-    this.startRowPositions = new long[rowGroups.size()];
-
-    // Fetch all row groups starting positions to compute the row offsets of the filtered row groups
-    Map<Long, Long> offsetToStartPos = generateOffsetToStartPos(expectedSchema);
 
     ParquetMetricsRowGroupFilter statsFilter = null;
     ParquetDictionaryRowGroupFilter dictFilter = null;
@@ -112,8 +103,6 @@ class ReadConf<T> {
     long computedTotalValues = 0L;
     for (int i = 0; i < shouldSkip.length; i += 1) {
       BlockMetaData rowGroup = rowGroups.get(i);
-      startRowPositions[i] =
-          offsetToStartPos == null ? 0 : offsetToStartPos.get(rowGroup.getStartingPos());
       boolean shouldRead =
           filter == null
               || (statsFilter.shouldRead(typeWithIds, rowGroup)
@@ -155,7 +144,6 @@ private ReadConf(ReadConf<T> toCopy) {
     this.batchSize = toCopy.batchSize;
     this.vectorizedModel = toCopy.vectorizedModel;
     this.columnChunkMetaDataForRowGroups = toCopy.columnChunkMetaDataForRowGroups;
-    this.startRowPositions = toCopy.startRowPositions;
   }
 
   ParquetFileReader reader() {
@@ -181,38 +169,6 @@ boolean[] shouldSkip() {
     return shouldSkip;
   }
 
-  private Map<Long, Long> generateOffsetToStartPos(Schema schema) {
-    if (schema.findField(MetadataColumns.ROW_POSITION.fieldId()) == null) {
-      return null;
-    }
-
-    FileDecryptionProperties decryptionProperties =
-        (options == null) ? null : options.getDecryptionProperties();
-
-    ParquetReadOptions readOptions =
-        ParquetReadOptions.builder().withDecryption(decryptionProperties).build();
-
-    try (ParquetFileReader fileReader = newReader(file, readOptions)) {
-      Map<Long, Long> offsetToStartPos = Maps.newHashMap();
-
-      long curRowCount = 0;
-      for (int i = 0; i < fileReader.getRowGroups().size(); i += 1) {
-        BlockMetaData meta = fileReader.getRowGroups().get(i);
-        offsetToStartPos.put(meta.getStartingPos(), curRowCount);
-        curRowCount += meta.getRowCount();
-      }
-
-      return offsetToStartPos;
-
-    } catch (IOException e) {
-      throw new UncheckedIOException("Failed to create/close reader for file: " + file, e);
-    }
-  }
-
-  long[] startRowPositions() {
-    return startRowPositions;
-  }
-
   long totalValues() {
     return totalValues;
   }

parquet/src/main/java/org/apache/iceberg/parquet/VectorizedParquetReader.java

@@ -113,7 +113,6 @@ private static class FileIterator<T> implements CloseableIterator<T> {
     private long nextRowGroupStart = 0;
     private long valuesRead = 0;
     private T last = null;
-    private final long[] rowGroupsStartRowPos;
 
     FileIterator(ReadConf conf) {
       this.reader = conf.reader();
@@ -124,7 +123,6 @@ private static class FileIterator<T> implements CloseableIterator<T> {
       this.batchSize = conf.batchSize();
       this.model.setBatchSize(this.batchSize);
       this.columnChunkMetadata = conf.columnChunkMetadataForRowGroups();
-      this.rowGroupsStartRowPos = conf.startRowPositions();
     }
 
     @Override
@@ -165,8 +163,7 @@ private void advance() {
         throw new RuntimeIOException(e);
       }
 
-      long rowPosition = rowGroupsStartRowPos[nextRowGroup];
-      model.setRowGroupInfo(pages, columnChunkMetadata.get(nextRowGroup), rowPosition);
+      model.setRowGroupInfo(pages, columnChunkMetadata.get(nextRowGroup));
       nextRowGroupStart += pages.getRowCount();
       nextRowGroup += 1;
     }

parquet/src/main/java/org/apache/iceberg/parquet/VectorizedReader.java

@@ -43,10 +43,25 @@ public interface VectorizedReader<T> {
    * @param pages row group information for all the columns
    * @param metadata map of {@link ColumnPath} -&gt; {@link ColumnChunkMetaData} for the row group
    * @param rowPosition the row group's row offset in the parquet file
+   * @deprecated since 1.6.0, will be removed in 1.7.0; use setRowGroupInfo(PageReadStore,
+   *     Map&lt;ColumnPath, ColumnChunkMetaData&gt;) instead.
    */
+  @Deprecated
   void setRowGroupInfo(
       PageReadStore pages, Map<ColumnPath, ColumnChunkMetaData> metadata, long rowPosition);
 
+  /**
+   * Sets the row group information to be used with this reader
+   *
+   * @param pages row group information for all the columns
+   * @param metadata map of {@link ColumnPath} -&gt; {@link ColumnChunkMetaData} for the row group
+   */
+  default void setRowGroupInfo(PageReadStore pages, Map<ColumnPath, ColumnChunkMetaData> metadata) {
+    throw new UnsupportedOperationException(
+        this.getClass().getName()
+            + " doesn't implement setRowGroupInfo(PageReadStore, Map<ColumnPath, ColumnChunkMetaData>)");
+  }
+
   /** Release any resources allocated. */
   void close();
 }

spark/v3.3/spark/src/main/java/org/apache/iceberg/spark/data/vectorized/ColumnarBatchReader.java

@@ -55,8 +55,14 @@ public ColumnarBatchReader(List<VectorizedReader<?>> readers) {
   @Override
   public void setRowGroupInfo(
       PageReadStore pageStore, Map<ColumnPath, ColumnChunkMetaData> metaData, long rowPosition) {
-    super.setRowGroupInfo(pageStore, metaData, rowPosition);
-    this.rowStartPosInBatch = rowPosition;
+    setRowGroupInfo(pageStore, metaData);
+  }
+
+  @Override
+  public void setRowGroupInfo(
+      PageReadStore pageStore, Map<ColumnPath, ColumnChunkMetaData> metaData) {
+    super.setRowGroupInfo(pageStore, metaData);
+    this.rowStartPosInBatch = pageStore.getRowIndexOffset().orElse(0L);
   }
 
   public void setDeleteFilter(DeleteFilter<InternalRow> deleteFilter) {