#177 Add In predicate support for row group and page level filter pushdown

ArnavBalyan · ArnavBalyan · commit bb25bd9bcdd6 · 2026-03-29T12:24:14.000+05:30
diff --git a/core/src/main/java/dev/hardwood/internal/reader/PageFilterEvaluator.java b/core/src/main/java/dev/hardwood/internal/reader/PageFilterEvaluator.java
@@ -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;
@@ -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()) {
diff --git a/core/src/main/java/dev/hardwood/internal/reader/RowGroupFilterEvaluator.java b/core/src/main/java/dev/hardwood/internal/reader/RowGroupFilterEvaluator.java
@@ -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;
@@ -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)) {
@@ -202,6 +208,80 @@ 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());
+        for (int value : p.values()) {
+            if (value >= min && value <= max) {
+                return false;
+            }
+        }
+        return true;
+    }
+
+    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());
+        for (long value : p.values()) {
+            if (value >= min && value <= max) {
+                return false;
+            }
+        }
+        return true;
+    }
+
+    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();
+        for (byte[] value : p.values()) {
+            if (StatisticsDecoder.compareBinary(value, min) >= 0
+                    && StatisticsDecoder.compareBinary(value, max) <= 0) {
+                return false;
+            }
+        }
+        return true;
+    }
+
+    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.
diff --git a/core/src/main/java/dev/hardwood/reader/FilterPredicate.java b/core/src/main/java/dev/hardwood/reader/FilterPredicate.java
@@ -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 {
@@ -190,6 +193,22 @@ 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) {
+        return new IntInPredicate(column, values);
+    }
+
+    static FilterPredicate in(String column, long... values) {
+        return new LongInPredicate(column, values);
+    }
+
+    static FilterPredicate inStrings(String column, String... values) {
+        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) {
@@ -247,6 +266,51 @@ public int hashCode() {
         }
     }
 
+    record IntInPredicate(String column, int[] values) implements FilterPredicate {
+
+        @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 {
+
+        @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 {
+
+        @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 {
diff --git a/core/src/test/java/dev/hardwood/FilterPredicateTest.java b/core/src/test/java/dev/hardwood/FilterPredicateTest.java
@@ -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) {
diff --git a/core/src/test/java/dev/hardwood/internal/reader/PageFilterEvaluatorTest.java b/core/src/test/java/dev/hardwood/internal/reader/PageFilterEvaluatorTest.java
@@ -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<>();
