From ec023de302e51da67777fc437736f7e8e2850347 Mon Sep 17 00:00:00 2001 From: Davide Angelocola Date: Sun, 18 Oct 2026 09:08:21 +0200 Subject: [PATCH 1/2] fix(calcite): aggregate push-down ignored DISTINCT and FILTER VortexAggregatePushDownRule answered every COUNT/MIN/MAX/SUM call from zone statistics without looking at the call's modifiers. Statistics describe every row of a zone, so they cannot answer an aggregate over a subset of its rows: count(distinct g) 3000 rows, 7 values -> 3000 count(*) filter (where id < 10) -> 3000, not 10 min(id) filter (where id > 2000) -> 0, not 2001 On Raincloud sf-salaries, count(distinct `Organization Group`) returned 1,180,053 instead of 7. The rule now leaves the plan to the scan when any call is DISTINCT or carries a FILTER clause. Co-Authored-By: Claude Sonnet 5 Claude-Session: https://claude.ai/code/session_01WtuVGQ5h5Lk6Zn4JnkxF1K --- CHANGELOG.md | 1 + .../calcite/VortexAggregatePushDownRule.java | 10 ++++- .../calcite/AggregateRuleBranchTest.java | 15 ++++++++ .../CalciteInteropIntegrationTest.java | 37 ++++++++++++++++++- 4 files changed, 61 insertions(+), 2 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index f7999c84f..6fca76518 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -29,6 +29,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 - **Breaking:** `ExtensionId` is a sealed `WellKnown`/`Custom` type like `EncodingId`, so third-party extension ids can be named, and `ExtensionId.parse` returns the id instead of an `Optional`. ([#530](https://github.com/dfa1/vortex-java/issues/530)) ### Fixed +- SQL `COUNT(DISTINCT col)` and aggregates with a `FILTER (WHERE ...)` clause return the right answer; the statistics push-down returned the row count and ignored the filter. ([#554](https://github.com/dfa1/vortex-java/pull/554)) - The Calcite adapter reads decimal, binary, date, time, timestamp and uuid columns, and a list, struct, map, union or variant column no longer stops the other columns of its table from being queried. ([#550](https://github.com/dfa1/vortex-java/pull/550)) - An extension dtype read from a file stays usable after its `VortexReader` is closed, instead of failing with "Already closed". ([#550](https://github.com/dfa1/vortex-java/pull/550)) - `vortex count`, `schema`, `stats` and `export` report a malformed file as an error instead of crashing with a stack trace. ([#547](https://github.com/dfa1/vortex-java/pull/547)) diff --git a/calcite/src/main/java/io/github/dfa1/vortex/calcite/VortexAggregatePushDownRule.java b/calcite/src/main/java/io/github/dfa1/vortex/calcite/VortexAggregatePushDownRule.java index 041145899..154a87e0b 100644 --- a/calcite/src/main/java/io/github/dfa1/vortex/calcite/VortexAggregatePushDownRule.java +++ b/calcite/src/main/java/io/github/dfa1/vortex/calcite/VortexAggregatePushDownRule.java @@ -36,7 +36,7 @@ /// /// Fires only when it can answer *every* aggregate from statistics: no `GROUP BY`, and each call /// is `COUNT(*)`, `COUNT(col)`, `MIN(col)`, `MAX(col)` (numeric or `VARCHAR`/`CHAR`), or `SUM(col)` -/// over a numeric column. `SUM` folds the per-zone `SUM` rows via [VortexTable#zoneSum(ColumnName)]; it +/// over a numeric column, with no `DISTINCT` and no `FILTER (WHERE ...)` clause. `SUM` folds the per-zone `SUM` rows via [VortexTable#zoneSum(ColumnName)]; it /// emits the SQL `NULL` of an all-null (or empty) column, and abandons (falling back to the scan) /// for a column whose zone-map table cannot answer it — no zone map, an overflowed zone, or a zero /// fold whose null count is unknown (a genuine zero is indistinguishable from all-null). Anything @@ -88,6 +88,14 @@ public void onMatch(RelOptRuleCall call) { if (aggregate.getGroupCount() != 0) { return; } + // Statistics describe every row of a zone, so they cannot answer an aggregate over a subset + // of its rows: COUNT(DISTINCT g) is not COUNT(g), and a FILTER (WHERE ...) clause narrows + // the rows one call sees. Leave either to the scan. + for (AggregateCall aggCall : aggregate.getAggCallList()) { + if (aggCall.isDistinct() || aggCall.hasFilter()) { + return; + } + } // Explicit operands give concrete rels under both Hep and Volcano. Walk the optional // Project and Filter nodes between the Aggregate and the TableScan; each rule variant fixes // the shape, so the levels present here are exactly those its operand declared. diff --git a/calcite/src/test/java/io/github/dfa1/vortex/calcite/AggregateRuleBranchTest.java b/calcite/src/test/java/io/github/dfa1/vortex/calcite/AggregateRuleBranchTest.java index c08e86c07..afdd1d758 100644 --- a/calcite/src/test/java/io/github/dfa1/vortex/calcite/AggregateRuleBranchTest.java +++ b/calcite/src/test/java/io/github/dfa1/vortex/calcite/AggregateRuleBranchTest.java @@ -21,6 +21,8 @@ import org.junit.jupiter.api.BeforeAll; import org.junit.jupiter.api.Test; import org.junit.jupiter.api.io.TempDir; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.ValueSource; import java.nio.channels.FileChannel; import java.nio.file.Path; @@ -128,6 +130,19 @@ void groupedAggregate_isLeftUntouched() { assertThat(optimize("select symbol, max(high) from ohlc group by symbol")).contains("Aggregate"); } + /// Zone statistics cover every row, so they cannot answer an aggregate over a subset of rows: + /// answering these from stats returned the row count for `COUNT(DISTINCT)` and ignored `FILTER`. + @ParameterizedTest + @ValueSource(strings = { + "select count(distinct volume) from ohlc", + "select count(*) filter (where volume > 0) from ohlc", + "select min(low) filter (where high > 0), count(*) from ohlc"}) + void distinctOrFilteredAggregate_isLeftUntouched(String sql) { + // Given an aggregate call the stats cannot answer + // When / Then — Aggregate stays + assertThat(optimize(sql)).contains("Aggregate"); + } + private static String optimize(String sql) { FrameworkConfig config = Frameworks.newConfigBuilder() .defaultSchema(schema) diff --git a/integration/src/test/java/io/github/dfa1/vortex/integration/CalciteInteropIntegrationTest.java b/integration/src/test/java/io/github/dfa1/vortex/integration/CalciteInteropIntegrationTest.java index 5f5f07326..0e228259e 100644 --- a/integration/src/test/java/io/github/dfa1/vortex/integration/CalciteInteropIntegrationTest.java +++ b/integration/src/test/java/io/github/dfa1/vortex/integration/CalciteInteropIntegrationTest.java @@ -194,21 +194,56 @@ private static void writeTypesWithJni(Path file) throws Exception { } } + @ParameterizedTest(name = "{0}") + @CsvSource(delimiter = '|', value = { + // g = id % 7 has 7 distinct values over 3000 rows + "select count(distinct g) from vtx.t | 7", + "select count(distinct g) from vtx.t where id < 3 | 3", + "select count(*) filter (where id < 10) from vtx.t | 10", + "select count(n) filter (where id < 1001) from vtx.t | 999", + "select min(id) filter (where id > 2000) from vtx.t | 2001", + "select max(id) filter (where id < 5) from vtx.t | 4", + "select sum(id) filter (where id < 4) from vtx.t | 6"}) + void jniWriter_distinctAndFilteredAggregates_answerOverTheSelectedRows(String sql, long expected, + @TempDir Path tmp) throws Exception { + // Given — the statistics push-down answered COUNT/MIN/MAX/SUM from whole zones without + // looking at DISTINCT or an aggregate's FILTER clause, so COUNT(DISTINCT g) returned the + // row count and a FILTERed aggregate ignored its filter + Path file = tmp.resolve("jni_distinct.vortex"); + writeIdsWithJni(file, BATCHES, BATCH_ROWS); + + // When + long result; + try (Connection conn = VortexCalcite.connect("vtx", Map.of("t", file)); + Statement st = conn.createStatement(); + ResultSet rs = st.executeQuery(sql)) { + rs.next(); + result = rs.getLong(1); + } + + // Then + assertThat(result).isEqualTo(expected); + } + private static void writeIdsWithJni(Path file, int batches, int batchRows) throws Exception { Schema schema = new Schema(List.of( Field.notNullable("id", new ArrowType.Int(64, true)), - Field.nullable("n", new ArrowType.Int(64, true)))); + Field.nullable("n", new ArrowType.Int(64, true)), + Field.notNullable("g", new ArrowType.Int(64, true)))); try (VortexWriter writer = VortexWriter.builder(SESSION, file.toAbsolutePath().toUri().toString(), schema, ALLOCATOR).build()) { for (int batch = 0; batch < batches; batch++) { try (VectorSchemaRoot root = VectorSchemaRoot.create(schema, ALLOCATOR)) { BigIntVector ids = (BigIntVector) root.getVector("id"); BigIntVector n = (BigIntVector) root.getVector("n"); + BigIntVector g = (BigIntVector) root.getVector("g"); ids.allocateNew(batchRows); n.allocateNew(batchRows); + g.allocateNew(batchRows); for (int i = 0; i < batchRows; i++) { long id = (long) batch * batchRows + i; ids.set(i, id); + g.set(i, id % 7); if (id % 1000 == 0) { n.setNull(i); } else { From a6f0b6da1afbc865d0c53f96ca33170c8f86207e Mon Sep 17 00:00:00 2001 From: Davide Angelocola Date: Sun, 18 Oct 2026 09:15:52 +0200 Subject: [PATCH 2/2] fix(calcite): string MIN/MAX push-down failed outside Latin-1 MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Calcite types VARCHAR as ISO-8859-1, and the statistics push-down built a literal from a string MIN/MAX stat without checking it fits. A column whose max holds a character Latin-1 cannot encode failed the whole query: select min(s), max(s) from t -- s holds "Ωmega" CalciteException: Failed to encode 'Ωmega' in character set 'ISO-8859-1' The scan returns the same strings without building a literal, so the rule now abandons when the type's charset cannot encode the value. Rust records string statistics as bounds the rule does not use, so this hit files with exact string stats, as the Java writer emits. Co-Authored-By: Claude Sonnet 5 Claude-Session: https://claude.ai/code/session_01WtuVGQ5h5Lk6Zn4JnkxF1K --- CHANGELOG.md | 1 + .../calcite/VortexAggregatePushDownRule.java | 10 ++++- .../vortex/calcite/AggregatePushDownTest.java | 30 ++++++++++++++ .../CalciteInteropIntegrationTest.java | 39 +++++++++++++++++++ 4 files changed, 79 insertions(+), 1 deletion(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 6fca76518..3d42e9f57 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -30,6 +30,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Fixed - SQL `COUNT(DISTINCT col)` and aggregates with a `FILTER (WHERE ...)` clause return the right answer; the statistics push-down returned the row count and ignored the filter. ([#554](https://github.com/dfa1/vortex-java/pull/554)) +- SQL `MIN`/`MAX` of a string column whose values fall outside Latin-1 answers instead of failing the query. ([#554](https://github.com/dfa1/vortex-java/pull/554)) - The Calcite adapter reads decimal, binary, date, time, timestamp and uuid columns, and a list, struct, map, union or variant column no longer stops the other columns of its table from being queried. ([#550](https://github.com/dfa1/vortex-java/pull/550)) - An extension dtype read from a file stays usable after its `VortexReader` is closed, instead of failing with "Already closed". ([#550](https://github.com/dfa1/vortex-java/pull/550)) - `vortex count`, `schema`, `stats` and `export` report a malformed file as an error instead of crashing with a stack trace. ([#547](https://github.com/dfa1/vortex-java/pull/547)) diff --git a/calcite/src/main/java/io/github/dfa1/vortex/calcite/VortexAggregatePushDownRule.java b/calcite/src/main/java/io/github/dfa1/vortex/calcite/VortexAggregatePushDownRule.java index 154a87e0b..c0c80f6e8 100644 --- a/calcite/src/main/java/io/github/dfa1/vortex/calcite/VortexAggregatePushDownRule.java +++ b/calcite/src/main/java/io/github/dfa1/vortex/calcite/VortexAggregatePushDownRule.java @@ -24,6 +24,7 @@ import org.apache.calcite.sql.SqlKind; import org.apache.calcite.sql.type.SqlTypeName; import org.apache.calcite.util.NlsString; +import org.apache.calcite.util.Util; import java.math.BigDecimal; import java.util.ArrayList; @@ -311,7 +312,8 @@ private static RexLiteral exact(RexBuilder rexBuilder, long value, RelDataType t return rexBuilder.makeExactLiteral(BigDecimal.valueOf(value), type); } - /// Builds a literal for a non-null `MIN`/`MAX` value: a [String] zone-map stat (a `Utf8` column, + /// Builds a literal for a non-null `MIN`/`MAX` value, or `null` (the rule abandons) for a string + /// the type's charset cannot encode: a [String] zone-map stat (a `Utf8` column, /// whose min/max the writer records as the full string, not a numeric summary) against a /// `CHAR`/`VARCHAR` output type, wrapped as an [NlsString] — [RexBuilder#makeLiteral(Object, /// RelDataType)] asserts a `CHAR`-family value already comes in that form, populating charset @@ -320,6 +322,12 @@ private static RexLiteral exact(RexBuilder rexBuilder, long value, RelDataType t private static RexLiteral minMaxLiteral(RexBuilder rexBuilder, Object value, RelDataType type) { if (value instanceof String s && (type.getSqlTypeName() == SqlTypeName.VARCHAR || type.getSqlTypeName() == SqlTypeName.CHAR)) { + // A literal must fit the type's charset, ISO-8859-1 unless declared otherwise; one that + // does not fails the whole query, so leave it to the scan, which builds no literal. + java.nio.charset.Charset charset = type.getCharset() != null ? type.getCharset() : Util.getDefaultCharset(); + if (!charset.newEncoder().canEncode(s)) { + return null; + } return rexBuilder.makeLiteral(new NlsString(s, null, null), type); } return numericLiteral(rexBuilder, value, type); diff --git a/calcite/src/test/java/io/github/dfa1/vortex/calcite/AggregatePushDownTest.java b/calcite/src/test/java/io/github/dfa1/vortex/calcite/AggregatePushDownTest.java index b64d659f4..722491016 100644 --- a/calcite/src/test/java/io/github/dfa1/vortex/calcite/AggregatePushDownTest.java +++ b/calcite/src/test/java/io/github/dfa1/vortex/calcite/AggregatePushDownTest.java @@ -146,6 +146,36 @@ void minMaxOnVarcharColumnRewritesToValuesFromStats(@TempDir Path localTmp) thro assertThat(symbolRow.get(1).getValueAs(String.class)).isEqualTo("TSLA"); } + @Test + void minMaxOfStringsOutsideLatin1_answerInsteadOfFailing(@TempDir Path localTmp) throws Exception { + // Given a VARCHAR column whose zone-map max holds a character ISO-8859-1 cannot encode. + // Calcite types VARCHAR as ISO-8859-1, so a literal built from that stat failed the whole + // query ("Failed to encode 'Ω'"); the scan returns the same string without a literal + Path stringsFile = localTmp.resolve("greek.vortex"); + DType.Struct stringsSchema = DType.structBuilder().field(ColumnName.of("s"), DType.UTF8).build(); + WriteOptions stringsOpts = WriteOptions.cascading(0).withZoneMaps(true).withGlobalDict(false) + .withEdition(Editions.CORE_2025_10_0); + try (var ch = FileChannel.open(stringsFile, StandardOpenOption.CREATE, StandardOpenOption.WRITE); + var writer = VortexWriter.create(ch, stringsSchema, stringsOpts)) { + writer.writeChunk(Map.of(ColumnName.of("s"), new String[]{"MSFT", "\u03A9mega", "AAPL"})); + } + + // When + String min; + String max; + try (Connection conn = VortexCalcite.connect("vtx", Map.of("t", stringsFile)); + Statement st = conn.createStatement(); + ResultSet rs = st.executeQuery("select min(s), max(s) from vtx.t")) { + rs.next(); + min = rs.getString(1); + max = rs.getString(2); + } + + // Then + assertThat(min).isEqualTo("AAPL"); + assertThat(max).isEqualTo("\u03A9mega"); + } + @Test void sumRewritesToValuesFromZoneStats() throws Exception { // Given a whole-table SUM over an integer column (volume, I64 → exact Long) and a floating diff --git a/integration/src/test/java/io/github/dfa1/vortex/integration/CalciteInteropIntegrationTest.java b/integration/src/test/java/io/github/dfa1/vortex/integration/CalciteInteropIntegrationTest.java index 0e228259e..4d4c316c8 100644 --- a/integration/src/test/java/io/github/dfa1/vortex/integration/CalciteInteropIntegrationTest.java +++ b/integration/src/test/java/io/github/dfa1/vortex/integration/CalciteInteropIntegrationTest.java @@ -225,6 +225,45 @@ void jniWriter_distinctAndFilteredAggregates_answerOverTheSelectedRows(String sq assertThat(result).isEqualTo(expected); } + @Test + void jniWriter_minMaxOfStringsOutsideLatin1_answer(@TempDir Path tmp) throws Exception { + // Given — Calcite types VARCHAR as ISO-8859-1, so a pushed-down literal holding 'Ω' failed + // the query; the answer must be the strings Rust wrote, however it is computed + Path file = tmp.resolve("jni_greek.vortex"); + Schema schema = new Schema(List.of(Field.notNullable("s", new ArrowType.Utf8()))); + String[] values = {"MSFT", "\u03A9mega", "AAPL"}; + try (VortexWriter writer = VortexWriter.builder(SESSION, file.toAbsolutePath().toUri().toString(), schema, + ALLOCATOR).build(); + VectorSchemaRoot root = VectorSchemaRoot.create(schema, ALLOCATOR)) { + org.apache.arrow.vector.VarCharVector s = (org.apache.arrow.vector.VarCharVector) root.getVector("s"); + s.allocateNew(values.length); + for (int i = 0; i < values.length; i++) { + s.setSafe(i, values[i].getBytes(java.nio.charset.StandardCharsets.UTF_8)); + } + root.setRowCount(values.length); + try (ArrowArray array = ArrowArray.allocateNew(ALLOCATOR); + ArrowSchema arrowSchema = ArrowSchema.allocateNew(ALLOCATOR)) { + Data.exportVectorSchemaRoot(ALLOCATOR, root, null, array, arrowSchema); + writer.writeBatch(array.memoryAddress(), arrowSchema.memoryAddress()); + } + } + + // When + String min; + String max; + try (Connection conn = VortexCalcite.connect("vtx", Map.of("t", file)); + Statement st = conn.createStatement(); + ResultSet rs = st.executeQuery("select min(s), max(s) from vtx.t")) { + rs.next(); + min = rs.getString(1); + max = rs.getString(2); + } + + // Then + assertThat(min).isEqualTo("AAPL"); + assertThat(max).isEqualTo("\u03A9mega"); + } + private static void writeIdsWithJni(Path file, int batches, int batchRows) throws Exception { Schema schema = new Schema(List.of( Field.notNullable("id", new ArrowType.Int(64, true)),