fix: fix spark/sql test failures in native_iceberg_compat (apache#1593)

Author: parthchandra

common/src/main/java/org/apache/comet/parquet/BatchReader.java

@@ -23,10 +23,7 @@
 import java.io.IOException;
 import java.net.URI;
 import java.net.URISyntaxException;
-import java.util.Arrays;
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
+import java.util.*;
 import java.util.concurrent.Callable;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Future;
@@ -285,8 +282,8 @@ public void init() throws URISyntaxException, IOException {
     // Initialize missing columns and use null vectors for them
     missingColumns = new boolean[columns.size()];
     List<String[]> paths = requestedSchema.getPaths();
-    StructField[] nonPartitionFields = sparkSchema.fields();
     ShimFileFormat.findRowIndexColumnIndexInSchema(sparkSchema);
+    StructField[] nonPartitionFields = sparkSchema.fields();
     for (int i = 0; i < requestedSchema.getFieldCount(); i++) {
       Type t = requestedSchema.getFields().get(i);
       Preconditions.checkState(

common/src/main/java/org/apache/comet/parquet/FileReader.java

@@ -80,6 +80,7 @@
 import org.apache.parquet.io.InputFile;
 import org.apache.parquet.io.ParquetDecodingException;
 import org.apache.parquet.io.SeekableInputStream;
+import org.apache.parquet.schema.MessageType;
 import org.apache.parquet.schema.PrimitiveType;
 import org.apache.spark.sql.execution.metric.SQLMetric;
 
@@ -578,6 +579,10 @@ private boolean advanceToNextBlock() {
   }
 
   public long[] getRowIndices() {
+    return getRowIndices(blocks);
+  }
+
+  public static long[] getRowIndices(List<BlockMetaData> blocks) {
     long[] rowIndices = new long[blocks.size() * 2];
     for (int i = 0, n = blocks.size(); i < n; i++) {
       BlockMetaData block = blocks.get(i);
@@ -591,7 +596,7 @@ public long[] getRowIndices() {
   //
   // The reason reflection is used here is that some Spark versions still depend on a
   // Parquet version where the method `getRowIndexOffset` is not public.
-  private long getRowIndexOffset(BlockMetaData metaData) {
+  public static long getRowIndexOffset(BlockMetaData metaData) {
     try {
       Method method = BlockMetaData.class.getMethod("getRowIndexOffset");
       method.setAccessible(true);
@@ -699,6 +704,35 @@ private static ParquetMetadata readFooter(
   }
 
   private List<BlockMetaData> filterRowGroups(List<BlockMetaData> blocks) {
+    return filterRowGroups(options, blocks, this);
+  }
+
+  public static List<BlockMetaData> filterRowGroups(
+      ParquetReadOptions options, List<BlockMetaData> blocks, FileReader fileReader) {
+    FilterCompat.Filter recordFilter = options.getRecordFilter();
+    if (FilterCompat.isFilteringRequired(recordFilter)) {
+      // set up data filters based on configured levels
+      List<RowGroupFilter.FilterLevel> levels = new ArrayList<>();
+
+      if (options.useStatsFilter()) {
+        levels.add(STATISTICS);
+      }
+
+      if (options.useDictionaryFilter()) {
+        levels.add(DICTIONARY);
+      }
+
+      if (options.useBloomFilter()) {
+        levels.add(BLOOMFILTER);
+      }
+      return RowGroupFilter.filterRowGroups(levels, recordFilter, blocks, fileReader);
+    }
+
+    return blocks;
+  }
+
+  public static List<BlockMetaData> filterRowGroups(
+      ParquetReadOptions options, List<BlockMetaData> blocks, MessageType schema) {
     FilterCompat.Filter recordFilter = options.getRecordFilter();
     if (FilterCompat.isFilteringRequired(recordFilter)) {
       // set up data filters based on configured levels
@@ -715,7 +749,7 @@ private List<BlockMetaData> filterRowGroups(List<BlockMetaData> blocks) {
       if (options.useBloomFilter()) {
         levels.add(BLOOMFILTER);
       }
-      return RowGroupFilter.filterRowGroups(levels, recordFilter, blocks, this);
+      return RowGroupFilter.filterRowGroups(levels, recordFilter, blocks, schema);
     }
 
     return blocks;

common/src/main/java/org/apache/comet/parquet/Native.java

@@ -244,15 +244,15 @@ public static native void setPageV2(
    * Initialize a record batch reader for a PartitionedFile
    *
    * @param filePath
-   * @param start
-   * @param length
+   * @param starts
+   * @param lengths
    * @return a handle to the record batch reader, used in subsequent calls.
    */
   public static native long initRecordBatchReader(
       String filePath,
       long fileSize,
-      long start,
-      long length,
+      long[] starts,
+      long[] lengths,
       byte[] filter,
       byte[] requiredSchema,
       byte[] dataSchema,