diff --git a/CHANGELOG.md b/CHANGELOG.md index f7999c84f..3d42e9f57 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -29,6 +29,8 @@ 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)) +- 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 041145899..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; @@ -36,7 +37,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 +89,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. @@ -303,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 @@ -312,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/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..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 @@ -194,21 +194,95 @@ 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); + } + + @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)), - 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 {