address review finding

Author: Peter Toth

parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ParquetFileReader.java

@@ -1492,7 +1492,8 @@ public ColumnIndexStore getColumnIndexStore(int blockIndex) {
   /**
    * Computes the {@link RowRanges} within the given row group that may pass the configured filter
    * (set via {@link ParquetReadOptions} or {@link ParquetInputFormat#setFilterPredicate}). If no
-   * filter is configured, returns a {@link RowRanges} covering all rows in the row group.
+   * filter is configured, returns a {@link RowRanges} covering all rows in the row group. If the
+   * row group has no rows, returns {@link RowRanges#EMPTY}.
    *
    * <p>This computation is metadata-only: it consults each filter-referenced column's column
    * index from the file footer; no column data is read from disk. The result can be passed to
@@ -1501,10 +1502,20 @@ public ColumnIndexStore getColumnIndexStore(int blockIndex) {
    *
    * @param blockIndex the row group (block) index
    * @return row ranges within the block that may pass the configured filter
+   * @throws IllegalArgumentException if {@code blockIndex} is out of range
    */
   public RowRanges getRowRanges(int blockIndex) {
+    if (blockIndex < 0 || blockIndex >= blocks.size()) {
+      throw new IllegalArgumentException(String.format(
+          "Invalid block index %s, the valid block index range are: [%s, %s]",
+          blockIndex, 0, blocks.size() - 1));
+    }
+    long rowCount = blocks.get(blockIndex).getRowCount();
+    if (rowCount == 0L) {
+      return RowRanges.EMPTY;
+    }
     if (!FilterCompat.isFilteringRequired(options.getRecordFilter())) {
-      return RowRanges.createSingle(blocks.get(blockIndex).getRowCount());
+      return RowRanges.createSingle(rowCount);
     }
     RowRanges rowRanges = blockRowRanges.get(blockIndex);
     if (rowRanges == null) {

parquet-hadoop/src/test/java/org/apache/parquet/hadoop/TestParquetFileReaderRowRanges.java

@@ -20,6 +20,7 @@
 
 import static org.apache.parquet.hadoop.ParquetFileWriter.Mode.OVERWRITE;
 import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertThrows;
 import static org.junit.Assert.assertTrue;
 
 import java.io.File;
@@ -93,4 +94,13 @@ public void getRowRangesWithoutFilterCoversAllRows() throws IOException {
       assertTrue(ranges.isOverlapping(0L, block.getRowCount() - 1));
     }
   }
+
+  @Test
+  public void getRowRangesRejectsOutOfRangeBlockIndex() throws IOException {
+    try (ParquetFileReader reader = openReader()) {
+      int blockCount = reader.getRowGroups().size();
+      assertThrows(IllegalArgumentException.class, () -> reader.getRowRanges(-1));
+      assertThrows(IllegalArgumentException.class, () -> reader.getRowRanges(blockCount));
+    }
+  }
 }