fix(calcite): aggregate push-down wrong results (DISTINCT, FILTER) and Latin-1 failure

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))

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);

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

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)

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 {