Skip to content

Commit e28ea62

Browse files
clairemcgintyclairemcginty
committed
PARQUET-34: Implement page-level Size filter
1 parent 2e8ba22 commit e28ea62

File tree

4 files changed

+212
-17
lines changed

4 files changed

+212
-17
lines changed

‎parquet-column/src/main/java/org/apache/parquet/internal/column/columnindex/ColumnIndexBuilder.java‎

Lines changed: 80 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -381,7 +381,69 @@ public <T extends Comparable<T>> PrimitiveIterator.OfInt visit(Contains<T> conta
381381

382382
@Override
383383
public PrimitiveIterator.OfInt visit(Size size) {
384-
return IndexIterator.all(getPageCount());
384+
if (repLevelHistogram == null || defLevelHistogram == null) {
385+
return IndexIterator.all(getPageCount());
386+
}
387+
388+
final int[] repLevelOffsets = calculateOffsetsForHistogram(repLevelHistogram, nullPages);
389+
final int[] defLevelOffsets = calculateOffsetsForHistogram(defLevelHistogram, nullPages);
390+
391+
return IndexIterator.filter(getPageCount(), pageIndex -> {
392+
final boolean isFinalPage = pageIndex + 1 == nullPages.length;
393+
final List<Long> pageRepLevelHistogram = getRepetitionLevelHistogram()
394+
.subList(
395+
repLevelOffsets[pageIndex],
396+
isFinalPage ? repLevelHistogram.length : repLevelOffsets[pageIndex + 1]);
397+
final List<Long> pageDefLevelHistogram = getDefinitionLevelHistogram()
398+
.subList(
399+
defLevelOffsets[pageIndex],
400+
isFinalPage ? defLevelHistogram.length : defLevelOffsets[pageIndex + 1]);
401+
402+
if (pageRepLevelHistogram.isEmpty() || pageDefLevelHistogram.isEmpty()) {
403+
// Page might match; cannot be filtered out
404+
return true;
405+
}
406+
407+
// If all values have repetition level 0, then no array has more than 1 element
408+
if (pageRepLevelHistogram.size() == 1
409+
|| pageRepLevelHistogram.subList(1, pageRepLevelHistogram.size()).stream()
410+
.allMatch(l -> l == 0)) {
411+
412+
if ((pageDefLevelHistogram.subList(1, pageDefLevelHistogram.size()).stream()
413+
.allMatch(l -> l == 0)) // all lists are null
414+
|| // all lists are size 0
415+
(pageDefLevelHistogram.stream().allMatch(l -> l == 0)) // all lists are empty
416+
) {
417+
return size.filter(
418+
(eq) -> eq <= 0, (lt) -> true, (lte) -> true, (gt) -> gt < 0, (gte) -> gte <= 0);
419+
}
420+
421+
int maxDefinitionLevel = pageDefLevelHistogram.size() - 1;
422+
423+
// If all repetition levels are zero and all definition levels are > MAX_DEFINITION_LEVEL - 1, all
424+
// lists are of size 1
425+
if (pageDefLevelHistogram.subList(0, maxDefinitionLevel - 1).stream()
426+
.allMatch(l -> l == 0)) {
427+
return size.filter(
428+
(eq) -> eq == 1, (lt) -> lt > 1, (lte) -> lte >= 1, (gt) -> gt < 1, (gte) -> gte <= 1);
429+
}
430+
}
431+
432+
final long nonNullElementCount =
433+
pageRepLevelHistogram.stream().mapToLong(l -> l).sum() - pageDefLevelHistogram.get(0);
434+
final long numNonNullRecords = pageRepLevelHistogram.get(0) - pageDefLevelHistogram.get(0);
435+
436+
// Given the total number of elements and non-null fields, we can compute the max size of any array
437+
// field
438+
final long maxArrayElementCount = 1 + (nonNullElementCount - numNonNullRecords);
439+
440+
return size.filter(
441+
(eq) -> eq <= maxArrayElementCount,
442+
(lt) -> true,
443+
(lte) -> true,
444+
(gt) -> gt < maxArrayElementCount,
445+
(gte) -> gte <= maxArrayElementCount);
446+
});
385447
}
386448

387449
@Override
@@ -444,6 +506,23 @@ public boolean test(int pageIndex) {
444506
}
445507
});
446508
}
509+
510+
// Calculates each page's starting offset in a concatenated histogram
511+
private static int[] calculateOffsetsForHistogram(long[] histogram, boolean[] nullPages) {
512+
final int numNullPages =
513+
(int) BooleanList.of(nullPages).stream().filter(p -> p).count();
514+
final int numNonNullPages = nullPages.length - numNullPages;
515+
final int numLevelsPerNonNullPage = (histogram.length - numNullPages) / numNonNullPages;
516+
517+
int[] offsets = new int[nullPages.length];
518+
int currOffset = 0;
519+
for (int i = 0; i < nullPages.length; ++i) {
520+
offsets[i] = currOffset;
521+
currOffset += (nullPages[i] ? 1 : numLevelsPerNonNullPage);
522+
}
523+
524+
return offsets;
525+
}
447526
}
448527

449528
private static final ColumnIndexBuilder NO_OP_BUILDER = new ColumnIndexBuilder() {

‎parquet-column/src/test/java/org/apache/parquet/internal/column/columnindex/TestColumnIndexBuilder.java‎

Lines changed: 97 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -36,6 +36,7 @@
3636
import static org.apache.parquet.filter2.predicate.FilterApi.notEq;
3737
import static org.apache.parquet.filter2.predicate.FilterApi.notIn;
3838
import static org.apache.parquet.filter2.predicate.FilterApi.or;
39+
import static org.apache.parquet.filter2.predicate.FilterApi.size;
3940
import static org.apache.parquet.filter2.predicate.FilterApi.userDefined;
4041
import static org.apache.parquet.filter2.predicate.LogicalInverter.invert;
4142
import static org.apache.parquet.schema.OriginalType.DECIMAL;
@@ -56,6 +57,7 @@
5657
import static org.junit.Assert.assertTrue;
5758
import static org.junit.Assert.fail;
5859

60+
import com.google.common.collect.ImmutableList;
5961
import java.math.BigDecimal;
6062
import java.nio.ByteBuffer;
6163
import java.util.ArrayList;
@@ -64,9 +66,11 @@
6466
import java.util.List;
6567
import java.util.Set;
6668
import org.apache.parquet.bytes.BytesUtils;
69+
import org.apache.parquet.column.statistics.SizeStatistics;
6770
import org.apache.parquet.column.statistics.Statistics;
6871
import org.apache.parquet.filter2.predicate.ContainsRewriter;
6972
import org.apache.parquet.filter2.predicate.FilterPredicate;
73+
import org.apache.parquet.filter2.predicate.Operators;
7074
import org.apache.parquet.filter2.predicate.Operators.BinaryColumn;
7175
import org.apache.parquet.filter2.predicate.Operators.BooleanColumn;
7276
import org.apache.parquet.filter2.predicate.Operators.DoubleColumn;
@@ -1627,6 +1631,99 @@ public void testNoOpBuilder() {
16271631
assertNull(builder.build());
16281632
}
16291633

1634+
@Test
1635+
public void testSizeRequiredElements() throws Exception {
1636+
final PrimitiveType type = Types.required(DOUBLE).named("element");
1637+
final DoubleColumn col = doubleColumn(type.getName());
1638+
1639+
final List<List<Double>> pageValueList = new ArrayList<>();
1640+
pageValueList.add(ImmutableList.of(1.0, 2.0, 3.0));
1641+
pageValueList.add(ImmutableList.of(1.0, 2.0, 3.0, 4.0, 5.0));
1642+
pageValueList.add(ImmutableList.of(-1.0));
1643+
pageValueList.add(ImmutableList.of());
1644+
pageValueList.add(null);
1645+
1646+
final ColumnIndex columnIndex = createArrayColumnIndex(type, pageValueList);
1647+
1648+
assertEquals(BoundaryOrder.UNORDERED, columnIndex.getBoundaryOrder());
1649+
assertCorrectNullCounts(columnIndex, 0, 0, 0, 0, 0);
1650+
assertCorrectNullPages(columnIndex, false, false, false, true, true);
1651+
assertCorrectValues(columnIndex.getMaxValues(), 3.0, 5.0, -1.0, null, null);
1652+
assertCorrectValues(columnIndex.getMinValues(), 1.0, 1.0, -1.0, null, null);
1653+
1654+
// we know max array size is 5; all elements of page 2 have size 1; and page 3 and 4 are null or empty
1655+
assertCorrectFiltering(columnIndex, size(col, Operators.Size.Operator.EQ, 0), 0, 1, 3, 4);
1656+
assertCorrectFiltering(columnIndex, size(col, Operators.Size.Operator.EQ, 4), 1);
1657+
assertCorrectFiltering(columnIndex, size(col, Operators.Size.Operator.EQ, 3), 0, 1);
1658+
assertCorrectFiltering(columnIndex, size(col, Operators.Size.Operator.LT, 2), 0, 1, 2, 3, 4);
1659+
assertCorrectFiltering(columnIndex, size(col, Operators.Size.Operator.LTE, 1), 0, 1, 2, 3, 4);
1660+
assertCorrectFiltering(columnIndex, size(col, Operators.Size.Operator.GT, 0), 0, 1, 2);
1661+
assertCorrectFiltering(columnIndex, size(col, Operators.Size.Operator.GTE, 0), 0, 1, 2, 3, 4);
1662+
}
1663+
1664+
@Test
1665+
public void testSizeOptionalElements() throws Exception {
1666+
final PrimitiveType type = Types.optional(DOUBLE).named("element");
1667+
final DoubleColumn col = doubleColumn(type.getName());
1668+
1669+
final List<Double> listWithNulls = new ArrayList<>();
1670+
listWithNulls.add(null);
1671+
listWithNulls.add(3.0);
1672+
listWithNulls.add(null);
1673+
1674+
final List<List<Double>> pageValueList = new ArrayList<>();
1675+
pageValueList.add(listWithNulls);
1676+
1677+
final ColumnIndex columnIndex = createArrayColumnIndex(type, pageValueList);
1678+
1679+
assertCorrectNullCounts(columnIndex, 2);
1680+
assertCorrectNullPages(columnIndex, false);
1681+
assertCorrectValues(columnIndex.getMaxValues(), 3.0);
1682+
assertCorrectValues(columnIndex.getMinValues(), 3.0);
1683+
1684+
// We know that the array values for the page have min size 0 and max size 3
1685+
assertCorrectFiltering(columnIndex, size(col, Operators.Size.Operator.EQ, 0), 0);
1686+
assertCorrectFiltering(columnIndex, size(col, Operators.Size.Operator.EQ, 5));
1687+
assertCorrectFiltering(columnIndex, size(col, Operators.Size.Operator.LT, 4), 0);
1688+
assertCorrectFiltering(columnIndex, size(col, Operators.Size.Operator.LTE, 3), 0);
1689+
assertCorrectFiltering(columnIndex, size(col, Operators.Size.Operator.GT, 0), 0);
1690+
assertCorrectFiltering(columnIndex, size(col, Operators.Size.Operator.GT, 3));
1691+
assertCorrectFiltering(columnIndex, size(col, Operators.Size.Operator.GTE, 3), 0);
1692+
assertCorrectFiltering(columnIndex, size(col, Operators.Size.Operator.GTE, 4));
1693+
}
1694+
1695+
private static ColumnIndex createArrayColumnIndex(PrimitiveType type, List<List<Double>> pageValueList) {
1696+
final ColumnIndexBuilder builder = ColumnIndexBuilder.getBuilder(type, Integer.MAX_VALUE);
1697+
1698+
for (List<Double> pageValues : pageValueList) {
1699+
final StatsBuilder sb = new StatsBuilder();
1700+
boolean isNullOrEmpty = pageValues == null || pageValues.isEmpty();
1701+
1702+
final SizeStatistics.Builder sizeStatistics =
1703+
SizeStatistics.newBuilder(type, isNullOrEmpty ? 0 : 1, isNullOrEmpty ? 0 : 1);
1704+
1705+
if (isNullOrEmpty) sizeStatistics.add(0, 0);
1706+
1707+
if (pageValues != null) {
1708+
for (int i = 0; i < pageValues.size(); i++) {
1709+
if (i == 0) {
1710+
sizeStatistics.add(0, 1);
1711+
} else {
1712+
sizeStatistics.add(1, 1);
1713+
}
1714+
}
1715+
}
1716+
1717+
if (pageValues == null) {
1718+
builder.add(sb.stats(type), sizeStatistics.build());
1719+
} else {
1720+
builder.add(sb.stats(type, pageValues.toArray(new Double[0])), sizeStatistics.build());
1721+
}
1722+
}
1723+
1724+
return builder.build();
1725+
}
1726+
16301727
private static List<ByteBuffer> toBBList(Binary... values) {
16311728
List<ByteBuffer> buffers = new ArrayList<>(values.length);
16321729
for (Binary value : values) {

‎parquet-hadoop/src/main/java/org/apache/parquet/filter2/statisticslevel/StatisticsFilter.java‎

Lines changed: 9 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -219,6 +219,10 @@ public <T extends Comparable<T>> Boolean visit(Contains<T> contains) {
219219
return contains.filter(this, (l, r) -> l || r, (l, r) -> l && r, v -> BLOCK_MIGHT_MATCH);
220220
}
221221

222+
/**
223+
* Logically equivalent to {@link org.apache.parquet.internal.column.columnindex.ColumnIndexBuilder},
224+
* but for block granularity
225+
**/
222226
@Override
223227
public Boolean visit(Size size) {
224228
final ColumnChunkMetaData metadata = getColumnChunk(size.getColumn().getColumnPath());
@@ -247,20 +251,19 @@ public Boolean visit(Size size) {
247251
definitionLevelHistogram.subList(1, definitionLevelHistogram.size()).stream()
248252
.allMatch(l -> l == 0))
249253
|| // all lists are size 0
250-
(definitionLevelHistogram.get(0) == 0
251-
&& definitionLevelHistogram.subList(2, definitionLevelHistogram.size()).stream()
252-
.allMatch(l -> l == 0))) {
254+
(definitionLevelHistogram.stream().allMatch(l -> l == 0))) {
253255

254256
final boolean blockCannotMatch =
255257
size.filter((eq) -> eq > 0, (lt) -> false, (lte) -> false, (gt) -> gt >= 0, (gte) -> gte > 0);
256258
return blockCannotMatch ? BLOCK_CANNOT_MATCH : BLOCK_MIGHT_MATCH;
257259
}
258260

259-
long maxDefinitionLevel = definitionLevelHistogram.get(definitionLevelHistogram.size() - 1);
261+
int maxDefinitionLevel = definitionLevelHistogram.size() - 1;
260262

261-
// If all repetition levels are zero and all definitions level are > MAX_DEFINITION_LEVEL - 1, all lists
263+
// If all repetition levels are zero and all definition levels are > MAX_DEFINITION_LEVEL - 1, all lists
262264
// are of size 1
263-
if (definitionLevelHistogram.stream().allMatch(l -> l > maxDefinitionLevel - 1)) {
265+
if (definitionLevelHistogram.subList(0, maxDefinitionLevel - 1).stream()
266+
.allMatch(l -> l == 0)) {
264267
final boolean blockCannotMatch = size.filter(
265268
(eq) -> eq != 1, (lt) -> lt <= 1, (lte) -> lte < 1, (gt) -> gt >= 1, (gte) -> gte > 1);
266269

‎parquet-hadoop/src/test/java/org/apache/parquet/filter2/statisticslevel/TestStatisticsFilter.java‎

Lines changed: 26 additions & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -513,6 +513,23 @@ public void testSizeFilterRequiredGroupRequiredElements() throws Exception {
513513
assertFalse(canDrop(size(nestedListColumn, Operators.Size.Operator.LTE, 1), columnMeta));
514514
assertFalse(canDrop(size(nestedListColumn, Operators.Size.Operator.LT, 1), columnMeta));
515515
assertFalse(canDrop(size(nestedListColumn, Operators.Size.Operator.EQ, 0), columnMeta));
516+
517+
// Case 3: all lists have size 1
518+
columnMeta = Collections.singletonList(getIntColumnMeta(
519+
nestedListColumn.getColumnPath(),
520+
minMaxStats,
521+
createSizeStatisticsForRepeatedField(true, ImmutableList.of(ImmutableList.of(1), ImmutableList.of(1))),
522+
2));
523+
524+
// We know that records have max array size 1
525+
assertTrue(canDrop(size(nestedListColumn, Operators.Size.Operator.GTE, 2), columnMeta));
526+
assertTrue(canDrop(size(nestedListColumn, Operators.Size.Operator.GT, 1), columnMeta));
527+
assertTrue(canDrop(size(nestedListColumn, Operators.Size.Operator.EQ, 2), columnMeta));
528+
529+
// These predicates should not be able to filter out the page
530+
assertFalse(canDrop(size(nestedListColumn, Operators.Size.Operator.LT, 2), columnMeta));
531+
assertFalse(canDrop(size(nestedListColumn, Operators.Size.Operator.LTE, 1), columnMeta));
532+
assertFalse(canDrop(size(nestedListColumn, Operators.Size.Operator.EQ, 1), columnMeta));
516533
}
517534

518535
@Test
@@ -549,13 +566,13 @@ public void testSizeFilterOptionalGroup() throws Exception {
549566
final IntStatistics minMaxStats = new IntStatistics();
550567

551568
// Case 1: List is non-null
552-
List<ColumnChunkMetaData> columnMeta = Collections.singletonList(getIntColumnMeta(
553-
nestedListColumn.getColumnPath(),
554-
minMaxStats,
555-
createSizeStatisticsForRepeatedField(
556-
false,
557-
ImmutableList.of(ImmutableList.of(1, 2, 3), ImmutableList.of(1), ImmutableList.of(1, 2, 3))),
558-
3));
569+
/* List<ColumnChunkMetaData> columnMeta = Collections.singletonList(getIntColumnMeta(
570+
nestedListColumn.getColumnPath(),
571+
minMaxStats,
572+
createSizeStatisticsForRepeatedField(
573+
false,
574+
ImmutableList.of(ImmutableList.of(1, 2, 3), ImmutableList.of(1), ImmutableList.of(1, 2, 3))),
575+
3));
559576
560577
// These predicates should be able to filter out the page
561578
assertTrue(canDrop(size(nestedListColumn, Operators.Size.Operator.GTE, 6), columnMeta));
@@ -566,11 +583,11 @@ public void testSizeFilterOptionalGroup() throws Exception {
566583
assertFalse(canDrop(size(nestedListColumn, Operators.Size.Operator.LT, 5), columnMeta));
567584
assertFalse(canDrop(size(nestedListColumn, Operators.Size.Operator.LTE, 3), columnMeta));
568585
assertFalse(canDrop(size(nestedListColumn, Operators.Size.Operator.EQ, 5), columnMeta));
569-
586+
*/
570587
// Case 2: List is null
571588
final List<List<Integer>> listWithNull = new ArrayList<>();
572589
listWithNull.add(null);
573-
columnMeta = Collections.singletonList(getIntColumnMeta(
590+
List<ColumnChunkMetaData> columnMeta = Collections.singletonList(getIntColumnMeta(
574591
nestedListColumn.getColumnPath(),
575592
minMaxStats,
576593
createSizeStatisticsForRepeatedField(true, listWithNull),
@@ -615,7 +632,6 @@ private static SizeStatistics createSizeStatisticsForRepeatedField(
615632
for (List<Integer> arrayValue : arrayValues) {
616633
final SimpleGroup record = new SimpleGroup(messageSchema);
617634
final Group nestedGroup = record.addGroup("nestedGroup");
618-
619635
if (arrayValue != null) {
620636
Group listField = nestedGroup.addGroup("listField");
621637
for (Integer value : arrayValue) {

0 commit comments

Comments
 (0)