diff --git a/CLAUDE.md b/CLAUDE.md index 5675fcc3..c327a29f 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -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. diff --git a/core/src/main/java/dev/hardwood/internal/reader/PageFilterEvaluator.java b/core/src/main/java/dev/hardwood/internal/reader/PageFilterEvaluator.java index 95f1ddf3..379c67ff 100644 --- 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 index cd7ab8c1..bf922541 100644 --- 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,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. diff --git a/core/src/main/java/dev/hardwood/reader/FilterPredicate.java b/core/src/main/java/dev/hardwood/reader/FilterPredicate.java index 01e9770b..17d22a10 100644 --- 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,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) { @@ -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 filters) implements FilterPredicate { diff --git a/core/src/test/java/dev/hardwood/FilterPredicateTest.java b/core/src/test/java/dev/hardwood/FilterPredicateTest.java index cfa02461..803d6edc 100644 --- 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 index 88a5fcb9..e2b5f37a 100644 --- 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 minValues = new ArrayList<>(); List maxValues = new ArrayList<>(); diff --git a/docs/content/usage.md b/docs/content/usage.md index ac18fe61..e9ef5f01 100644 --- a/docs/content/usage.md +++ b/docs/content/usage.md @@ -190,6 +190,10 @@ FilterPredicate filter = FilterPredicate.and( FilterPredicate.lt("age", 65) ); +// IN filter +FilterPredicate filter = FilterPredicate.in("department_id", 1, 3, 7); +FilterPredicate filter = FilterPredicate.inStrings("city", "NYC", "LA", "Chicago"); + try (ParquetFileReader fileReader = ParquetFileReader.open(InputFile.of(path)); RowReader rowReader = fileReader.createRowReader(filter)) { @@ -200,9 +204,9 @@ try (ParquetFileReader fileReader = ParquetFileReader.open(InputFile.of(path)); } ``` -Supported operators: `eq`, `notEq`, `lt`, `ltEq`, `gt`, `gtEq`. -Supported types: `int`, `long`, `float`, `double`, `boolean`, `String`. -Logical combinators: `and`, `or`, `not` — the `and` and `or` combinators also accept varargs for three or more conditions. +Supported operators: `eq`, `notEq`, `lt`, `ltEq`, `gt`, `gtEq`, `in`, `inStrings`. +Supported types: `int`, `long`, `float`, `double`, `boolean`, `String` (comparison operators); `int`, `long`, `String` (`in`/`inStrings`). +Logical combinators: `and`, `or`, `not`; the `and` and `or` combinators also accept varargs for three or more conditions. Filtering operates on the physical column type. Logical types like DATE (stored as INT32) and TIMESTAMP (stored as INT64) are filtered using their underlying physical type — for example, filter a DATE column with `FilterPredicate.gt("date_col", 19000)` (days since epoch).