Skip to content

Add in predicate support for row group filter pushdown #172

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
ArnavBalyan wants to merge 3 commits into from
Closed

Add in predicate support for row group filter pushdown #172

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
bb25bd9
#177 Add In predicate support for row group and page level filter pus…
ArnavBalyan Mar 29, 2026
164dc6a
#177 Adding docs, validating parameters, DRY-ing up a bit
gunnarmorling Mar 29, 2026
b4bc199
Merge branch 'main' into arnavb/in-predicate
gunnarmorling Mar 29, 2026
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
2 changes: 1 addition & 1 deletion CLAUDE.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -19,7 +19,7 @@ Design documents describe the intended end state. Do not include references to t

# Coding

When adding or changing public APIs, update the usage documentation in the README.md accordingly.
When adding or changing public APIs, update the usage documentation under docs/ accordingly.
Minimize the surface of the public API, only make user-facing what needs to be user-facing. Keep anything else in an `internal` package.
Similarly, only expose configuration options truly needed in the CLI.
Never do unsafe downcasts with potential value loss. E.g. prefer Math::toIntExact() where applicable.
Expand Down
21 changes: 21 additions & 0 deletions core/src/main/java/dev/hardwood/internal/reader/PageFilterEvaluator.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -21,11 +21,14 @@
import dev.hardwood.reader.FilterPredicate;
import dev.hardwood.reader.FilterPredicate.And;
import dev.hardwood.reader.FilterPredicate.BinaryColumnPredicate;
import dev.hardwood.reader.FilterPredicate.BinaryInPredicate;
import dev.hardwood.reader.FilterPredicate.BooleanColumnPredicate;
import dev.hardwood.reader.FilterPredicate.DoubleColumnPredicate;
import dev.hardwood.reader.FilterPredicate.FloatColumnPredicate;
import dev.hardwood.reader.FilterPredicate.IntColumnPredicate;
import dev.hardwood.reader.FilterPredicate.IntInPredicate;
import dev.hardwood.reader.FilterPredicate.LongColumnPredicate;
import dev.hardwood.reader.FilterPredicate.LongInPredicate;
import dev.hardwood.reader.FilterPredicate.Not;
import dev.hardwood.reader.FilterPredicate.Or;
import dev.hardwood.schema.FileSchema;
Expand Down Expand Up @@ -98,6 +101,24 @@ private static RowRanges evaluate(FilterPredicate predicate, RowGroup rowGroup,
return RowGroupFilterEvaluator.canDropCompared(p.op(), cmpMin, cmpMax,
StatisticsDecoder.compareBinary(min, max));
});
case IntInPredicate p -> evaluateLeaf(p.column(), rowGroup, schema, indexBuffers, rowCount,
(ci, i) -> {
int min = StatisticsDecoder.decodeInt(ci.minValues().get(i));
int max = StatisticsDecoder.decodeInt(ci.maxValues().get(i));
return RowGroupFilterEvaluator.canDropIntIn(p.values(), min, max);
});
case LongInPredicate p -> evaluateLeaf(p.column(), rowGroup, schema, indexBuffers, rowCount,
(ci, i) -> {
long min = StatisticsDecoder.decodeLong(ci.minValues().get(i));
long max = StatisticsDecoder.decodeLong(ci.maxValues().get(i));
return RowGroupFilterEvaluator.canDropLongIn(p.values(), min, max);
});
case BinaryInPredicate p -> evaluateLeaf(p.column(), rowGroup, schema, indexBuffers, rowCount,
(ci, i) -> {
byte[] min = ci.minValues().get(i);
byte[] max = ci.maxValues().get(i);
return RowGroupFilterEvaluator.canDropBinaryIn(p.values(), min, max);
});
case And a -> {
RowRanges result = RowRanges.all(rowCount);
for (FilterPredicate child : a.filters()) {
Expand Down
64 changes: 64 additions & 0 deletions core/src/main/java/dev/hardwood/internal/reader/RowGroupFilterEvaluator.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -15,11 +15,14 @@
import dev.hardwood.reader.FilterPredicate;
import dev.hardwood.reader.FilterPredicate.And;
import dev.hardwood.reader.FilterPredicate.BinaryColumnPredicate;
import dev.hardwood.reader.FilterPredicate.BinaryInPredicate;
import dev.hardwood.reader.FilterPredicate.BooleanColumnPredicate;
import dev.hardwood.reader.FilterPredicate.DoubleColumnPredicate;
import dev.hardwood.reader.FilterPredicate.FloatColumnPredicate;
import dev.hardwood.reader.FilterPredicate.IntColumnPredicate;
import dev.hardwood.reader.FilterPredicate.IntInPredicate;
import dev.hardwood.reader.FilterPredicate.LongColumnPredicate;
import dev.hardwood.reader.FilterPredicate.LongInPredicate;
import dev.hardwood.reader.FilterPredicate.Not;
import dev.hardwood.reader.FilterPredicate.Or;
import dev.hardwood.schema.FileSchema;
Expand All @@ -46,6 +49,9 @@ public static boolean canDropRowGroup(FilterPredicate predicate, RowGroup rowGro
case DoubleColumnPredicate p -> evaluateDouble(p, rowGroup, schema);
case BooleanColumnPredicate p -> evaluateBoolean(p, rowGroup, schema);
case BinaryColumnPredicate p -> evaluateBinary(p, rowGroup, schema);
case IntInPredicate p -> evaluateIntIn(p, rowGroup, schema);
case LongInPredicate p -> evaluateLongIn(p, rowGroup, schema);
case BinaryInPredicate p -> evaluateBinaryIn(p, rowGroup, schema);
case And a -> {
for (FilterPredicate f : a.filters()) {
if (canDropRowGroup(f, rowGroup, schema)) {
Expand Down Expand Up @@ -202,6 +208,64 @@ private static boolean evaluateBinary(BinaryColumnPredicate p, RowGroup rowGroup
StatisticsDecoder.compareBinary(min, max));
}

private static boolean evaluateIntIn(IntInPredicate p, RowGroup rowGroup, FileSchema schema) {
Statistics stats = findStatistics(p.column(), rowGroup, schema);
if (stats == null || stats.minValue() == null || stats.maxValue() == null) {
return false;
}
int min = StatisticsDecoder.decodeInt(stats.minValue());
int max = StatisticsDecoder.decodeInt(stats.maxValue());
return canDropIntIn(p.values(), min, max);
}

private static boolean evaluateLongIn(LongInPredicate p, RowGroup rowGroup, FileSchema schema) {
Statistics stats = findStatistics(p.column(), rowGroup, schema);
if (stats == null || stats.minValue() == null || stats.maxValue() == null) {
return false;
}
long min = StatisticsDecoder.decodeLong(stats.minValue());
long max = StatisticsDecoder.decodeLong(stats.maxValue());
return canDropLongIn(p.values(), min, max);
}

private static boolean evaluateBinaryIn(BinaryInPredicate p, RowGroup rowGroup, FileSchema schema) {
Statistics stats = findStatistics(p.column(), rowGroup, schema);
if (stats == null || stats.minValue() == null || stats.maxValue() == null) {
return false;
}
byte[] min = stats.minValue();
byte[] max = stats.maxValue();
return canDropBinaryIn(p.values(), min, max);
}

static boolean canDropIntIn(int[] values, int min, int max) {
for (int value : values) {
if (value >= min && value <= max) {
return false;
}
}
return true;
}

static boolean canDropLongIn(long[] values, long min, long max) {
for (long value : values) {
if (value >= min && value <= max) {
return false;
}
}
return true;
}

static boolean canDropBinaryIn(byte[][] values, byte[] min, byte[] max) {
for (byte[] value : values) {
if (StatisticsDecoder.compareBinary(value, min) >= 0
&& StatisticsDecoder.compareBinary(value, max) <= 0) {
return false;
}
}
return true;
}

// ==================== Generic comparison logic ====================

/// Determines if a range can be dropped given integer-comparable min/max statistics.
Expand Down
88 changes: 88 additions & 0 deletions core/src/main/java/dev/hardwood/reader/FilterPredicate.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -40,6 +40,9 @@ public sealed interface FilterPredicate
FilterPredicate.DoubleColumnPredicate,
FilterPredicate.BooleanColumnPredicate,
FilterPredicate.BinaryColumnPredicate,
FilterPredicate.IntInPredicate,
FilterPredicate.LongInPredicate,
FilterPredicate.BinaryInPredicate,
FilterPredicate.And,
FilterPredicate.Or,
FilterPredicate.Not {
Expand Down Expand Up @@ -190,6 +193,31 @@ static FilterPredicate gtEq(String column, String value) {
return new BinaryColumnPredicate(column, Operator.GT_EQ, value.getBytes(StandardCharsets.UTF_8));
}

static FilterPredicate in(String column, int... values) {
if (values.length == 0) {
throw new IllegalArgumentException("IN predicate requires at least one value");
}
return new IntInPredicate(column, values);
}

static FilterPredicate in(String column, long... values) {
if (values.length == 0) {
throw new IllegalArgumentException("IN predicate requires at least one value");
}
return new LongInPredicate(column, values);
}

static FilterPredicate inStrings(String column, String... values) {
if (values.length == 0) {
throw new IllegalArgumentException("IN predicate requires at least one value");
}
byte[][] encoded = new byte[values.length][];
for (int i = 0; i < values.length; i++) {
encoded[i] = values[i].getBytes(StandardCharsets.UTF_8);
}
return new BinaryInPredicate(column, encoded);
}

// ==================== Logical Combinators ====================

static FilterPredicate and(FilterPredicate left, FilterPredicate right) {
Expand Down Expand Up @@ -247,6 +275,66 @@ public int hashCode() {
}
}

record IntInPredicate(String column, int[] values) implements FilterPredicate {

public IntInPredicate(String column, int[] values) {
this.column = column;
this.values = values.clone();
}

@Override
public boolean equals(Object o) {
if (this == o) return true;
if (!(o instanceof IntInPredicate that)) return false;
return column.equals(that.column) && Arrays.equals(values, that.values);
}

@Override
public int hashCode() {
return 31 * column.hashCode() + Arrays.hashCode(values);
}
}

record LongInPredicate(String column, long[] values) implements FilterPredicate {

public LongInPredicate(String column, long[] values) {
this.column = column;
this.values = values.clone();
}

@Override
public boolean equals(Object o) {
if (this == o) return true;
if (!(o instanceof LongInPredicate that)) return false;
return column.equals(that.column) && Arrays.equals(values, that.values);
}

@Override
public int hashCode() {
return 31 * column.hashCode() + Arrays.hashCode(values);
}
}

record BinaryInPredicate(String column, byte[][] values) implements FilterPredicate {

public BinaryInPredicate(String column, byte[][] values) {
this.column = column;
this.values = values.clone();
}

@Override
public boolean equals(Object o) {
if (this == o) return true;
if (!(o instanceof BinaryInPredicate that)) return false;
return column.equals(that.column) && Arrays.deepEquals(values, that.values);
}

@Override
public int hashCode() {
return 31 * column.hashCode() + Arrays.deepHashCode(values);
}
}

// ==================== Logical Combinator Records ====================

record And(List<FilterPredicate> filters) implements FilterPredicate {
Expand Down
71 changes: 71 additions & 0 deletions core/src/test/java/dev/hardwood/FilterPredicateTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -370,6 +370,77 @@ void testUnknownColumnNeverDrop() {
FilterPredicate.eq("nonexistent", 42), rg, schema)).isFalse();
}

@Test
void testIntInPredicateCreation() {
FilterPredicate p = FilterPredicate.in("id", 1, 5, 10);
assertThat(p).isInstanceOf(FilterPredicate.IntInPredicate.class);
FilterPredicate.IntInPredicate ip = (FilterPredicate.IntInPredicate) p;
assertThat(ip.column()).isEqualTo("id");
assertThat(ip.values()).containsExactly(1, 5, 10);
}

@Test
void testLongInPredicateCreation() {
FilterPredicate p = FilterPredicate.in("ts", 100L, 200L);
assertThat(p).isInstanceOf(FilterPredicate.LongInPredicate.class);
}

@Test
void testStringInPredicateCreation() {
FilterPredicate p = FilterPredicate.inStrings("city", "NYC", "LA");
assertThat(p).isInstanceOf(FilterPredicate.BinaryInPredicate.class);
}

@Test
void testCanDropWithIntIn() {
RowGroup rg = createIntRowGroup(10, 20);
FileSchema schema = createIntSchema();

assertThat(RowGroupFilterEvaluator.canDropRowGroup(
FilterPredicate.in("col", 1, 5, 8), rg, schema)).isTrue();

assertThat(RowGroupFilterEvaluator.canDropRowGroup(
FilterPredicate.in("col", 25, 30), rg, schema)).isTrue();

assertThat(RowGroupFilterEvaluator.canDropRowGroup(
FilterPredicate.in("col", 5, 15, 25), rg, schema)).isFalse();

assertThat(RowGroupFilterEvaluator.canDropRowGroup(
FilterPredicate.in("col", 1, 10), rg, schema)).isFalse();
}

@Test
void testCanDropWithLongIn() {
RowGroup rg = createLongRowGroup(100L, 200L);
FileSchema schema = createLongSchema();

assertThat(RowGroupFilterEvaluator.canDropRowGroup(
FilterPredicate.in("col", 50L, 80L), rg, schema)).isTrue();
assertThat(RowGroupFilterEvaluator.canDropRowGroup(
FilterPredicate.in("col", 50L, 150L), rg, schema)).isFalse();
}

@Test
void testCanDropWithStringIn() {
RowGroup rg = createBinaryRowGroup("banana".getBytes(), "date".getBytes());
FileSchema schema = createBinarySchema();

assertThat(RowGroupFilterEvaluator.canDropRowGroup(
FilterPredicate.inStrings("col", "apple", "elderberry"), rg, schema)).isTrue();

assertThat(RowGroupFilterEvaluator.canDropRowGroup(
FilterPredicate.inStrings("col", "apple", "cherry"), rg, schema)).isFalse();
}

@Test
void testCanDropWithInMissingStatistics() {
RowGroup rg = createRowGroupWithoutStatistics();
FileSchema schema = createIntSchema();

assertThat(RowGroupFilterEvaluator.canDropRowGroup(
FilterPredicate.in("col", 1, 2, 3), rg, schema)).isFalse();
}

// ==================== Helpers ====================

private static RowGroup createIntRowGroup(int min, int max) {
Expand Down
61 changes: 61 additions & 0 deletions core/src/test/java/dev/hardwood/internal/reader/PageFilterEvaluatorTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -479,6 +479,67 @@ private static ColumnIndex longColumnIndex(long[] mins, long[] maxs) {
ColumnIndex.BoundaryOrder.UNORDERED, null);
}

@Test
void testIntInPageFiltering() {
RowRanges ranges = PageFilterEvaluator.evaluatePages(INT_COLUMN_INDEX, THREE_PAGE_OFFSET_INDEX, THREE_PAGE_ROW_COUNT,
(ci, i) -> {
int min = StatisticsDecoder.decodeInt(ci.minValues().get(i));
int max = StatisticsDecoder.decodeInt(ci.maxValues().get(i));
return RowGroupFilterEvaluator.canDropIntIn(new int[]{ 5, 15 }, min, max);
});
assertTrue(ranges.overlapsPage(0, 30));
assertTrue(ranges.overlapsPage(30, 60));
assertFalse(ranges.overlapsPage(60, 90));
}

@Test
void testIntInPageFilteringAllOutside() {
RowRanges ranges = PageFilterEvaluator.evaluatePages(INT_COLUMN_INDEX, THREE_PAGE_OFFSET_INDEX, THREE_PAGE_ROW_COUNT,
(ci, i) -> {
int min = StatisticsDecoder.decodeInt(ci.minValues().get(i));
int max = StatisticsDecoder.decodeInt(ci.maxValues().get(i));
return RowGroupFilterEvaluator.canDropIntIn(new int[]{ 50, 60 }, min, max);
});
assertFalse(ranges.overlapsPage(0, 30));
assertFalse(ranges.overlapsPage(30, 60));
assertFalse(ranges.overlapsPage(60, 90));
}

@Test
void testLongInPageFiltering() {
ColumnIndex longIdx = longColumnIndex(new long[]{ 100, 200, 300 }, new long[]{ 199, 299, 399 });
OffsetIndex oi = offsetIndex(30, 30, 30);

RowRanges ranges = PageFilterEvaluator.evaluatePages(longIdx, oi, 90,
(ci, i) -> {
long min = StatisticsDecoder.decodeLong(ci.minValues().get(i));
long max = StatisticsDecoder.decodeLong(ci.maxValues().get(i));
return RowGroupFilterEvaluator.canDropLongIn(new long[]{ 150, 350 }, min, max);
});
assertTrue(ranges.overlapsPage(0, 30));
assertFalse(ranges.overlapsPage(30, 60));
assertTrue(ranges.overlapsPage(60, 90));
}

@Test
void testBinaryInPageFiltering() {
ColumnIndex binIdx = binaryColumnIndex(
List.of("apple".getBytes(StandardCharsets.UTF_8), "date".getBytes(StandardCharsets.UTF_8)),
List.of("cherry".getBytes(StandardCharsets.UTF_8), "fig".getBytes(StandardCharsets.UTF_8)));
OffsetIndex oi = offsetIndex(30, 30);

RowRanges ranges = PageFilterEvaluator.evaluatePages(binIdx, oi, 60,
(ci, i) -> {
byte[] min = ci.minValues().get(i);
byte[] max = ci.maxValues().get(i);
return RowGroupFilterEvaluator.canDropBinaryIn(
new byte[][]{ "banana".getBytes(StandardCharsets.UTF_8), "zebra".getBytes(StandardCharsets.UTF_8) },
min, max);
});
assertTrue(ranges.overlapsPage(0, 30));
assertFalse(ranges.overlapsPage(30, 60));
}

private static ColumnIndex floatColumnIndex(float[] mins, float[] maxs) {
List<byte[]> minValues = new ArrayList<>();
List<byte[]> maxValues = new ArrayList<>();
Expand Down
Loading