bench: add pathBasedReader A/B comparison benchmark to IcebergReaderBenchmark

Author: Shekharrajak

benchmarks/pom.xml

@@ -240,6 +240,12 @@
       <version>${project.parent.version}</version>
       <scope>test</scope>
     </dependency>
+    <dependency>
+      <groupId>org.apache.druid.extensions</groupId>
+      <artifactId>druid-parquet-extensions</artifactId>
+      <version>${project.parent.version}</version>
+      <scope>test</scope>
+    </dependency>
     <dependency>
       <groupId>org.apache.iceberg</groupId>
       <artifactId>iceberg-arrow</artifactId>

benchmarks/src/test/java/org/apache/druid/benchmark/IcebergReaderBenchmark.java

@@ -28,10 +28,15 @@
 import org.apache.druid.data.input.impl.DoubleDimensionSchema;
 import org.apache.druid.data.input.impl.LongDimensionSchema;
 import org.apache.druid.data.input.impl.StringDimensionSchema;
+import org.apache.druid.data.input.impl.LocalInputSourceFactory;
 import org.apache.druid.data.input.impl.TimestampSpec;
+import org.apache.druid.data.input.InputSourceReader;
+import org.apache.druid.data.input.parquet.ParquetInputFormat;
 import org.apache.druid.iceberg.input.IcebergArrowInputSourceReader;
+import org.apache.druid.iceberg.input.IcebergInputSource;
 import org.apache.druid.iceberg.input.LocalCatalog;
 import org.apache.druid.java.util.common.parsers.CloseableIterator;
+import org.apache.hadoop.conf.Configuration;
 import org.apache.iceberg.DataFile;
 import org.apache.iceberg.PartitionSpec;
 import org.apache.iceberg.Schema;
@@ -199,6 +204,40 @@ public void arrowReaderLargeBatch(final Blackhole bh) throws IOException
     }
   }
 
+  /**
+   * Existing path-based reader (current production behaviour when useArrowReader=false):
+   * IcebergCatalog extracts data-file paths from the snapshot, then a LocalInputSource +
+   * ParquetInputFormat re-opens and re-parses each Parquet file generically.
+   * No delete-file awareness, no scan-level column projection, no schema evolution.
+   */
+  @Benchmark
+  public void pathBasedReader(final Blackhole bh) throws IOException
+  {
+    final IcebergInputSource source = new IcebergInputSource(
+        TABLE,
+        NAMESPACE,
+        null,
+        catalog,
+        new LocalInputSourceFactory(),
+        null,
+        null,
+        false,
+        0
+    );
+    final ParquetInputFormat parquetFormat = new ParquetInputFormat(null, null, new Configuration());
+    final InputSourceReader reader = source.reader(inputRowSchema, parquetFormat, warehouseDir);
+    int count = 0;
+    try (CloseableIterator<InputRow> it = reader.read(NoopStats.INSTANCE)) {
+      while (it.hasNext()) {
+        bh.consume(it.next());
+        count++;
+      }
+    }
+    if (count != numRows) {
+      throw new RuntimeException("Expected " + numRows + " rows but got " + count);
+    }
+  }
+
   // --- helpers ---
 
   private static Schema buildSchema(final int numColumns)