Skip to content

Commit cd2572d

Browse files
peter-tothmbutrovichPeter Toth
authored
GH-3598: Expose getRowRanges(int) (#3599)
### Rationale for this change Opening up APIs needed by a later materialization feature in Spark. External readers (e.g. a Spark-side scanner) need (a) the column-index-derived row ranges that may pass the configured filter for a row group, and (b) a metadata-only estimate of the on-disk compressed bytes those ranges correspond to for the currently requested columns, so they can plan I/O without reading column data. ### What changes are included in this PR? - `getRowRanges(int blockIndex)`: made public; returns row ranges that may pass the configured filter. With no filter, shortcuts to all rows of the row group. - `getCompressedBytesForRowRanges(int blockIndex, RowRanges rowRanges)`: metadata-only sum of compressed page sizes for the reader's currently requested columns whose pages overlap the given row ranges. Dictionary pages are not represented in OffsetIndex and are therefore excluded. ### Are these changes tested? Yes. `TestParquetFileReaderRowRanges` covers: no-filter row ranges cover all rows, empty ranges short-circuit to 0, full ranges equal the per-page OffsetIndex sum and are strictly less than the column-chunk total (proving dictionary-page exclusion), and partial ranges fall between 0 and the full total. ### Are there any user-facing changes? No. Closes #3598 Co-authored-by: Matt Butrovich <mbutrovich@gmail.com> Co-authored-by: Peter Toth <p_toth@apple.com>
1 parent ec1b866 commit cd2572d

2 files changed

Lines changed: 134 additions & 3 deletions

File tree

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

Lines changed: 28 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -1489,9 +1489,34 @@ public ColumnIndexStore getColumnIndexStore(int blockIndex) {
14891489
return ciStore;
14901490
}
14911491

1492-
private RowRanges getRowRanges(int blockIndex) {
1493-
assert FilterCompat.isFilteringRequired(options.getRecordFilter())
1494-
: "Should not be invoked if filter is null or NOOP";
1492+
/**
1493+
* Computes the {@link RowRanges} within the given row group that may pass the configured filter
1494+
* (set via {@link ParquetReadOptions} or {@link ParquetInputFormat#setFilterPredicate}). If no
1495+
* filter is configured, returns a {@link RowRanges} covering all rows in the row group. If the
1496+
* row group has no rows, returns {@link RowRanges#EMPTY}.
1497+
*
1498+
* <p>This computation is metadata-only: it consults each filter-referenced column's column
1499+
* index from the file footer; no column data is read from disk. The result can be passed to
1500+
* {@link #readFilteredRowGroup(int, RowRanges)} (intersected with any caller-supplied row
1501+
* ranges if desired) to read only the matching pages.
1502+
*
1503+
* @param blockIndex the row group (block) index
1504+
* @return row ranges within the block that may pass the configured filter
1505+
* @throws IllegalArgumentException if {@code blockIndex} is out of range
1506+
*/
1507+
public RowRanges getRowRanges(int blockIndex) {
1508+
if (blockIndex < 0 || blockIndex >= blocks.size()) {
1509+
throw new IllegalArgumentException(String.format(
1510+
"Invalid block index %s, the valid block index range are: [%s, %s]",
1511+
blockIndex, 0, blocks.size() - 1));
1512+
}
1513+
long rowCount = blocks.get(blockIndex).getRowCount();
1514+
if (rowCount == 0L) {
1515+
return RowRanges.EMPTY;
1516+
}
1517+
if (!FilterCompat.isFilteringRequired(options.getRecordFilter())) {
1518+
return RowRanges.createSingle(rowCount);
1519+
}
14951520
RowRanges rowRanges = blockRowRanges.get(blockIndex);
14961521
if (rowRanges == null) {
14971522
rowRanges = ColumnIndexFilter.calculateRowRanges(
Lines changed: 106 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,106 @@
1+
/*
2+
* Licensed to the Apache Software Foundation (ASF) under one
3+
* or more contributor license agreements. See the NOTICE file
4+
* distributed with this work for additional information
5+
* regarding copyright ownership. The ASF licenses this file
6+
* to you under the Apache License, Version 2.0 (the
7+
* "License"); you may not use this file except in compliance
8+
* with the License. You may obtain a copy of the License at
9+
*
10+
* http://www.apache.org/licenses/LICENSE-2.0
11+
*
12+
* Unless required by applicable law or agreed to in writing,
13+
* software distributed under the License is distributed on an
14+
* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
15+
* KIND, either express or implied. See the License for the
16+
* specific language governing permissions and limitations
17+
* under the License.
18+
*/
19+
package org.apache.parquet.hadoop;
20+
21+
import static org.apache.parquet.hadoop.ParquetFileWriter.Mode.OVERWRITE;
22+
import static org.junit.Assert.assertEquals;
23+
import static org.junit.Assert.assertThrows;
24+
import static org.junit.Assert.assertTrue;
25+
26+
import java.io.File;
27+
import java.io.IOException;
28+
import org.apache.hadoop.conf.Configuration;
29+
import org.apache.hadoop.fs.Path;
30+
import org.apache.parquet.HadoopReadOptions;
31+
import org.apache.parquet.ParquetReadOptions;
32+
import org.apache.parquet.example.data.Group;
33+
import org.apache.parquet.example.data.simple.SimpleGroupFactory;
34+
import org.apache.parquet.hadoop.example.ExampleParquetWriter;
35+
import org.apache.parquet.hadoop.metadata.BlockMetaData;
36+
import org.apache.parquet.hadoop.util.HadoopInputFile;
37+
import org.apache.parquet.internal.filter2.columnindex.RowRanges;
38+
import org.apache.parquet.schema.MessageType;
39+
import org.apache.parquet.schema.MessageTypeParser;
40+
import org.junit.Before;
41+
import org.junit.Rule;
42+
import org.junit.Test;
43+
import org.junit.rules.TemporaryFolder;
44+
45+
/**
46+
* Tests {@link ParquetFileReader#getRowRanges(int)}.
47+
*/
48+
public class TestParquetFileReaderRowRanges {
49+
50+
private static final int ROW_COUNT = 10_000;
51+
private static final MessageType SCHEMA =
52+
MessageTypeParser.parseMessageType("message test { required int64 id; required int64 grp; }");
53+
54+
@Rule
55+
public final TemporaryFolder temp = new TemporaryFolder();
56+
57+
private Path file;
58+
59+
@Before
60+
public void writeFile() throws IOException {
61+
File f = temp.newFile();
62+
f.delete();
63+
file = new Path(f.toURI());
64+
65+
// Small page size produces many pages per column chunk.
66+
try (ParquetWriter<Group> writer = ExampleParquetWriter.builder(file)
67+
.withType(SCHEMA)
68+
.withWriteMode(OVERWRITE)
69+
.withRowGroupSize(64L * 1024 * 1024)
70+
.withPageSize(4 * 1024)
71+
.build()) {
72+
SimpleGroupFactory factory = new SimpleGroupFactory(SCHEMA);
73+
for (int i = 0; i < ROW_COUNT; i++) {
74+
writer.write(factory.newGroup().append("id", (long) i).append("grp", (long) (i % 8)));
75+
}
76+
}
77+
}
78+
79+
private ParquetFileReader openReader() throws IOException {
80+
Configuration conf = new Configuration();
81+
ParquetReadOptions options = HadoopReadOptions.builder(conf).build();
82+
return ParquetFileReader.open(HadoopInputFile.fromPath(file, conf), options);
83+
}
84+
85+
@Test
86+
public void getRowRangesWithoutFilterCoversAllRows() throws IOException {
87+
try (ParquetFileReader reader = openReader()) {
88+
assertEquals(1, reader.getRowGroups().size());
89+
BlockMetaData block = reader.getRowGroups().get(0);
90+
91+
RowRanges ranges = reader.getRowRanges(0);
92+
93+
assertEquals(block.getRowCount(), ranges.rowCount());
94+
assertTrue(ranges.isOverlapping(0L, block.getRowCount() - 1));
95+
}
96+
}
97+
98+
@Test
99+
public void getRowRangesRejectsOutOfRangeBlockIndex() throws IOException {
100+
try (ParquetFileReader reader = openReader()) {
101+
int blockCount = reader.getRowGroups().size();
102+
assertThrows(IllegalArgumentException.class, () -> reader.getRowRanges(-1));
103+
assertThrows(IllegalArgumentException.class, () -> reader.getRowRanges(blockCount));
104+
}
105+
}
106+
}

0 commit comments

Comments
 (0)