Skip to content

PARQUET-9: Filtering records across multiple blocks #9

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
tomwhite wants to merge 2 commits into from
Closed

PARQUET-9: Filtering records across multiple blocks #9

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
9e723ee
Test for filtering records across multiple blocks
Jun 18, 2014
afb08a4
Minimal fix
tomwhite Jun 19, 2014
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
52 changes: 51 additions & 1 deletion parquet-avro/src/test/java/parquet/avro/TestSpecificReadWrite.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -76,6 +76,52 @@ public void testFilterMatchesMultiple() throws IOException {
assertNull(reader.read());
}

@Test
public void testFilterMatchesMultipleBlocks() throws IOException {
Path path = writeCarsToParquetFile(10000, CompressionCodecName.UNCOMPRESSED, false, DEFAULT_BLOCK_SIZE/64, DEFAULT_PAGE_SIZE/64);
ParquetReader<Car> reader = new AvroParquetReader<Car>(path, column("make", equalTo("Volkswagen")));
for (int i = 0; i < 10000; i++) {
assertEquals(getVwPolo().toString(), reader.read().toString());
assertEquals(getVwPassat().toString(), reader.read().toString());
}
assertNull(reader.read());
}

@Test
public void testFilterMatchesNoBlocks() throws IOException {
Path path = writeCarsToParquetFile(10000, CompressionCodecName.UNCOMPRESSED, false, DEFAULT_BLOCK_SIZE/64, DEFAULT_PAGE_SIZE/64);
ParquetReader<Car> reader = new AvroParquetReader<Car>(path, column("make", equalTo("Bogus")));
assertNull(reader.read());
}

@Test
public void testFilterMatchesFinalBlockOnly() throws IOException {
File tmp = File.createTempFile(getClass().getSimpleName(), ".tmp");
tmp.deleteOnExit();
tmp.delete();
Path path = new Path(tmp.getPath());

Car vwPolo = getVwPolo();
Car vwPassat = getVwPassat();
Car bmwMini = getBmwMini();

ParquetWriter<Car> writer = new AvroParquetWriter<Car>(path, Car.SCHEMA$,
CompressionCodecName.UNCOMPRESSED, DEFAULT_BLOCK_SIZE/128, DEFAULT_PAGE_SIZE/128,
false);
for (int i = 0; i < 10000; i++) {
writer.write(vwPolo);
writer.write(vwPassat);
writer.write(vwPolo);
}
writer.write(bmwMini); // only write BMW in last block
writer.close();

ParquetReader<Car> reader = new AvroParquetReader<Car>(path, column("make",
equalTo("BMW")));
assertEquals(getBmwMini().toString(), reader.read().toString());
assertNull(reader.read());
}

@Test
public void testFilterWithDictionary() throws IOException {
Path path = writeCarsToParquetFile(1,CompressionCodecName.UNCOMPRESSED,true);
Expand Down Expand Up @@ -159,6 +205,10 @@ public void testAvroReadSchema() throws IOException {
}

private Path writeCarsToParquetFile( int num, CompressionCodecName compression, boolean enableDictionary) throws IOException {
return writeCarsToParquetFile(num, compression, enableDictionary, DEFAULT_BLOCK_SIZE, DEFAULT_PAGE_SIZE);
}

private Path writeCarsToParquetFile( int num, CompressionCodecName compression, boolean enableDictionary, int blockSize, int pageSize) throws IOException {
File tmp = File.createTempFile(getClass().getSimpleName(), ".tmp");
tmp.deleteOnExit();
tmp.delete();
Expand All @@ -169,7 +219,7 @@ private Path writeCarsToParquetFile( int num, CompressionCodecName compression,
Car bmwMini = getBmwMini();

ParquetWriter<Car> writer = new AvroParquetWriter<Car>(path,Car.SCHEMA$, compression,
DEFAULT_BLOCK_SIZE, DEFAULT_PAGE_SIZE, enableDictionary);
blockSize, pageSize, enableDictionary);
for (int i = 0; i < num; i++) {
writer.write(vwPolo);
writer.write(vwPassat);
Expand Down
14 changes: 12 additions & 2 deletions parquet-hadoop/src/main/java/parquet/hadoop/InternalParquetRecordReader.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -53,7 +53,7 @@ class InternalParquetRecordReader<T> {

private T currentValue;
private long total;
private int current = 0;
private long current = 0;
private int currentBlock = -1;
private ParquetFileReader reader;
private parquet.io.RecordReader<T> recordReader;
Expand Down Expand Up @@ -173,8 +173,18 @@ public boolean nextKeyValue() throws IOException, InterruptedException {
try {
checkRead();
currentValue = recordReader.read();
if (DEBUG) LOG.debug("read value: " + currentValue);
current ++;
while (currentValue == null) { // only happens with FilteredRecordReader at end of block
current = totalCountLoadedSoFar;
if (current < total) {
checkRead();
currentValue = recordReader.read();
current ++;
continue;
}
return false;
}
if (DEBUG) LOG.debug("read value: " + currentValue);
} catch (RuntimeException e) {
throw new ParquetDecodingException(format("Can not read value at %d in block %d in file %s", current, currentBlock, file), e);
}
Expand Down