perf(calcite): answer temporal MIN/MAX from zone maps; scan them without java.time

CHANGELOG.md

@@ -17,6 +17,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 - Custom extension types: register an `ExtensionDecoder` on `ReadRegistry` and an `ExtensionEncoder` on `WriteRegistry`, with a `validate` hook on both sides, and `Chunk.as` decodes any registered type. ([#530](https://github.com/dfa1/vortex-java/issues/530))
 
 ### Changed
+- SQL `MIN`/`MAX` of a date, time or timestamp column is answered from zone maps, and scans read those columns without building a `java.time` object per row. ([#559](https://github.com/dfa1/vortex-java/pull/559))
+- The Calcite adapter declares `TIME` and `TIMESTAMP` columns with millisecond precision, so `getString` shows the milliseconds the values carry. ([#559](https://github.com/dfa1/vortex-java/pull/559))
 - SQL `LIKE 'prefix%'` on a string column is about 5 times faster: it is pushed into the scan as a string range that prunes chunks, instead of being matched by Calcite row by row. ([#558](https://github.com/dfa1/vortex-java/pull/558))
 - SQL `ORDER BY ... LIMIT n` is up to 9 times faster: the first rows are kept in a bounded heap during the scan instead of sorting the whole table. ([#557](https://github.com/dfa1/vortex-java/pull/557))
 - SQL `GROUP BY` on a dictionary-encoded string column is 10 to 30 times faster: rows are counted and summed per dictionary code instead of being built one by one for Calcite. ([#555](https://github.com/dfa1/vortex-java/pull/555))

calcite/src/main/java/io/github/dfa1/vortex/calcite/Temporal.java

@@ -0,0 +1,84 @@
+package io.github.dfa1.vortex.calcite;
+
+import io.github.dfa1.vortex.core.error.VortexException;
+import io.github.dfa1.vortex.core.model.DType;
+import io.github.dfa1.vortex.core.model.ExtensionId;
+import io.github.dfa1.vortex.core.model.TimeUnit;
+import io.github.dfa1.vortex.core.model.TimestampDtype;
+import io.github.dfa1.vortex.reader.array.ByteArray;
+import io.github.dfa1.vortex.reader.array.IntArray;
+import io.github.dfa1.vortex.reader.array.LongArray;
+import io.github.dfa1.vortex.reader.array.MaskedArray;
+import io.github.dfa1.vortex.reader.array.ShortArray;
+
+/// A `vortex.date`, `vortex.time` or `vortex.timestamp` column's conversion from its raw storage
+/// integer to Calcite's internal form: `DATE` as epoch days (`Integer`), `TIME` as milliseconds of
+/// the day (`Integer`), `TIMESTAMP` as epoch milliseconds (`Long`).
+///
+/// Resolved once per column, so the scan neither parses the extension metadata nor builds a
+/// `LocalDate`/`LocalTime`/`Instant` per row, and the aggregate rule converts a zone statistic,
+/// which is the raw storage value, the same way. Rounding matches the reader's decoders: a
+/// millisecond date floors to its day, a timestamp floors to its millisecond.
+///
+/// @param kind which temporal extension the column is
+/// @param unit the column's storage unit
+record Temporal(ExtensionId.WellKnown kind, TimeUnit unit) {
+
+    private static final long MILLIS_PER_DAY = 86_400_000L;
+
+    /// The conversion for a column of `type`, or `null` when it is not a temporal extension.
+    ///
+    /// @throws VortexException for a unit the extension does not allow, as the decoders do
+    static Temporal of(DType type) {
+        if (!(type instanceof DType.Extension ext)
+                || !(ExtensionId.parse(ext.extensionId()) instanceof ExtensionId.WellKnown kind)
+                || kind != ExtensionId.WellKnown.VORTEX_DATE && kind != ExtensionId.WellKnown.VORTEX_TIME
+                && kind != ExtensionId.WellKnown.VORTEX_TIMESTAMP) {
+            return null;
+        }
+        // the unit byte leads every temporal extension's metadata
+        TimeUnit unit = TimestampDtype.readUnit(ext);
+        boolean valid = switch (kind) {
+            case VORTEX_DATE -> unit == TimeUnit.Days || unit == TimeUnit.Milliseconds;
+            default -> unit != TimeUnit.Days;
+        };
+        if (!valid) {
+            throw new VortexException(ext.extensionId() + " does not support time unit " + unit);
+        }
+        return new Temporal(kind, unit);
+    }
+
+    /// Converts one raw storage value.
+    Object convert(long raw) {
+        return switch (kind) {
+            case VORTEX_DATE -> Math.toIntExact(unit == TimeUnit.Days ? raw : Math.floorDiv(raw, MILLIS_PER_DAY));
+            case VORTEX_TIME -> {
+                long millis = unit == TimeUnit.Seconds ? raw * 1000 : raw / (unit.divisor() / 1000);
+                if (raw < 0 || millis >= MILLIS_PER_DAY) {
+                    throw new VortexException("vortex.time value " + raw + " " + unit + " is outside a day");
+                }
+                yield (int) millis;
+            }
+            default -> unit == TimeUnit.Seconds ? Math.multiplyExact(raw, 1000)
+                    : Math.floorDiv(raw, unit.divisor() / 1000);
+        };
+    }
+
+    /// Reads row `r` of the column's storage array (possibly masked) and converts it.
+    ///
+    /// @return the converted value, or `null` for a null row
+    Object read(Object array, long r) {
+        if (array instanceof MaskedArray masked) {
+            return masked.isValid(r) ? read(masked.inner(), r) : null;
+        }
+        long raw = switch (array) {
+            case LongArray a -> a.getLong(r);
+            case IntArray a -> a.getInt(r);
+            case ShortArray a -> a.getShort(r);
+            case ByteArray a -> a.getByte(r);
+            default -> throw new VortexException("temporal storage must be a signed integer, got "
+                    + array.getClass().getSimpleName());
+        };
+        return convert(raw);
+    }
+}

calcite/src/main/java/io/github/dfa1/vortex/calcite/VortexAggregatePushDownRule.java

@@ -238,7 +238,7 @@ private static void rewriteGrouped(RelOptRuleCall call, Aggregate aggregate, Vor
             row.add(key);
             for (int i = 0; i < calls.size(); i++) {
                 RexLiteral literal = groupedLiteral(calls.get(i), callColumns.get(i), entry.getValue(),
-                        outTypes.get(i + 1), scanRowType, rexBuilder);
+                        outTypes.get(i + 1), scanRowType, rexBuilder, table);
                 if (literal == null) {
                     return;
                 }
@@ -252,7 +252,8 @@ private static void rewriteGrouped(RelOptRuleCall call, Aggregate aggregate, Vor
 
     /// One aggregate call's literal for one group, or `null` when it cannot be built.
     private static RexLiteral groupedLiteral(AggregateCall aggCall, ColumnName column, VortexTable.GroupFold fold,
-                                             RelDataType outType, RelDataType scanRowType, RexBuilder rexBuilder) {
+                                             RelDataType outType, RelDataType scanRowType, RexBuilder rexBuilder,
+                                             VortexTable table) {
         if (column == null) {
             return exact(rexBuilder, fold.rows(), outType);
         }
@@ -264,7 +265,8 @@ private static RexLiteral groupedLiteral(AggregateCall aggCall, ColumnName colum
                     isNullable(scanRowType, column.value()));
             case MIN, MAX -> {
                 Object value = aggCall.getAggregation().getKind() == SqlKind.MIN ? columnFold.min() : columnFold.max();
-                yield value == null ? rexBuilder.makeNullLiteral(outType) : minMaxLiteral(rexBuilder, value, outType);
+                yield value == null ? rexBuilder.makeNullLiteral(outType)
+                        : extremeLiteral(rexBuilder, value, outType, table.columnType(column));
             }
             default -> null;
         };
@@ -326,7 +328,7 @@ private static RexLiteral evaluate(AggregateCall agg, RelDataType outType, Vorte
                             || (fold.nullCount() != null && fold.nullCount() == fold.rows());
                     yield provablyNoValues ? rexBuilder.makeNullLiteral(outType) : null;
                 }
-                yield minMaxLiteral(rexBuilder, value, outType);
+                yield extremeLiteral(rexBuilder, value, outType, table.columnType(col));
             }
             case SUM -> {
                 if (agg.getArgList().size() != 1) {
@@ -430,6 +432,18 @@ private static RexLiteral exact(RexBuilder rexBuilder, long value, RelDataType t
         return rexBuilder.makeExactLiteral(BigDecimal.valueOf(value), type);
     }
 
+    /// The literal for a `MIN`/`MAX` value of a column of `columnType`. A temporal column's statistic
+    /// is its raw storage integer, converted to the `DATE`/`TIME`/`TIMESTAMP` form first; anything
+    /// else goes through [#minMaxLiteral(RexBuilder, Object, RelDataType)].
+    private static RexLiteral extremeLiteral(RexBuilder rexBuilder, Object value, RelDataType type,
+                                             io.github.dfa1.vortex.core.model.DType columnType) {
+        Temporal temporal = Temporal.of(columnType);
+        if (temporal != null && value instanceof Number raw) {
+            return VortexSortPushDownRule.literal(rexBuilder, temporal.convert(raw.longValue()), type);
+        }
+        return minMaxLiteral(rexBuilder, value, type);
+    }
+
     /// 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

calcite/src/main/java/io/github/dfa1/vortex/calcite/VortexEnumerator.java

@@ -45,6 +45,8 @@ final class VortexEnumerator implements Enumerator<Object[]> {
     private final AtomicLong chunksScannedLastQuery;
     private final ColumnName[] names;
     private final DType[] types;
+    /// Per output column, its temporal conversion or `null`: resolved once, not per row.
+    private final Temporal[] temporals;
     private final RowFilter exactFilter;
     private final VortexReader reader;
     private final ScanIterator scan;
@@ -73,6 +75,10 @@ final class VortexEnumerator implements Enumerator<Object[]> {
         this.chunksScannedLastQuery = chunksScannedLastQuery;
         this.names = names;
         this.types = types;
+        this.temporals = new Temporal[types.length];
+        for (int c = 0; c < types.length; c++) {
+            temporals[c] = Temporal.of(types[c]);
+        }
         this.exactFilter = exactFilter;
         chunksScannedLastQuery.set(0);
         VortexReader openedReader = null;
@@ -144,14 +150,15 @@ boolean advance() {
 
     /// Column `c` of the row [#advance()] moved to.
     Object column(int c) {
-        return value(columns[c], types[c], currentRow);
+        Temporal temporal = temporals[c];
+        return temporal == null ? value(columns[c], types[c], currentRow) : temporal.read(columns[c], currentRow);
     }
 
     /// Every column of the row [#advance()] moved to.
     Object[] row() {
         Object[] row = new Object[names.length];
         for (int c = 0; c < names.length; c++) {
-            row[c] = value(columns[c], types[c], currentRow);
+            row[c] = column(c);
         }
         return row;
     }

calcite/src/main/java/io/github/dfa1/vortex/calcite/VortexTable.java

@@ -432,6 +432,12 @@ public Optional<Map<String, GroupFold>> groupedFold(ColumnName group, RowFilter
         return Optional.of(java.util.Collections.unmodifiableMap(result));
     }
 
+    /// The dtype of `column`.
+    DType columnType(ColumnName column) {
+        DType.Struct struct = struct();
+        return struct.fieldTypes().get(struct.fieldNames().indexOf(column));
+    }
+
     /// One `ORDER BY` key of [#topN(int[], RowFilter, List, int)].
     ///
     /// @param field      the key's position in the projected row
@@ -1088,8 +1094,10 @@ private static RelDataType toSqlType(RelDataTypeFactory factory, DType type) {
             case DType.Decimal d -> factory.createSqlType(SqlTypeName.DECIMAL, d.precision(), d.scale());
             case DType.Extension ext -> switch (ExtensionId.parse(ext.extensionId())) {
                 case ExtensionId.WellKnown.VORTEX_DATE -> factory.createSqlType(SqlTypeName.DATE);
-                case ExtensionId.WellKnown.VORTEX_TIME -> factory.createSqlType(SqlTypeName.TIME);
-                case ExtensionId.WellKnown.VORTEX_TIMESTAMP -> factory.createSqlType(SqlTypeName.TIMESTAMP);
+                // millisecond precision, the precision the scan delivers: at Calcite's default of 0 a
+                // literal built for the column (a pushed-down MIN/MAX, a top-N row) lost its millis
+                case ExtensionId.WellKnown.VORTEX_TIME -> factory.createSqlType(SqlTypeName.TIME, 3);
+                case ExtensionId.WellKnown.VORTEX_TIMESTAMP -> factory.createSqlType(SqlTypeName.TIMESTAMP, 3);
                 case ExtensionId.WellKnown.VORTEX_UUID -> factory.createSqlType(SqlTypeName.UUID);
                 default -> toSqlType(factory, ext.storageDType());
             };

calcite/src/test/java/io/github/dfa1/vortex/calcite/ColumnTypesSqlTest.java

@@ -96,7 +96,8 @@ void everyScalarType_readsThroughSql() throws Exception {
             assertThat(result.getBytes("blob")).containsExactly(1, 2, 3);
             assertThat(result.getDate("day").toLocalDate()).isEqualTo(DATE);
             assertThat(result.getTime("at").toLocalTime()).isEqualTo(TIME.withNano(0));
-            assertThat(result.getString("at")).isEqualTo("13:45:30");
+            // TIME is declared with millisecond precision, the precision values carry
+            assertThat(result.getString("at")).isEqualTo("13:45:30.123");
             // TIMESTAMP is a wall clock: the stored value read as UTC, with no zone shift
             assertThat(result.getTimestamp("wall").toLocalDateTime())
                     .isEqualTo(LocalDateTime.ofInstant(INSTANT, ZoneOffset.UTC));

calcite/src/test/java/io/github/dfa1/vortex/calcite/TemporalPushDownTest.java

@@ -0,0 +1,124 @@
+package io.github.dfa1.vortex.calcite;
+
+import io.github.dfa1.vortex.core.error.VortexException;
+import io.github.dfa1.vortex.core.model.ColumnName;
+import io.github.dfa1.vortex.core.model.DType;
+import io.github.dfa1.vortex.core.model.ExtensionId;
+import io.github.dfa1.vortex.core.model.TimeDtype;
+import io.github.dfa1.vortex.core.model.TimeUnit;
+import io.github.dfa1.vortex.core.model.TimestampDtype;
+import io.github.dfa1.vortex.writer.VortexWriter;
+import io.github.dfa1.vortex.writer.WriteOptions;
+import io.github.dfa1.vortex.writer.encode.DateExtensionEncoder;
+
+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.CsvSource;
+
+import java.nio.channels.FileChannel;
+import java.nio.file.Path;
+import java.nio.file.StandardOpenOption;
+import java.sql.Connection;
+import java.sql.ResultSet;
+import java.sql.Statement;
+import java.time.Instant;
+import java.time.LocalDate;
+import java.time.LocalTime;
+import java.util.List;
+import java.util.Map;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/// `MIN`/`MAX` of a date, time or timestamp column is answered from zone statistics, which hold the
+/// raw storage integer: [Temporal] converts it as the scan converts each row. Every query must equal
+/// the same aggregate over the column plus a zero interval, an expression the rule leaves to the scan.
+class TemporalPushDownTest {
+
+    @TempDir
+    static Path tmp;
+    private static Path file;
+
+    /// Two chunks; the timestamps are microseconds, one before the epoch with a fractional
+    /// millisecond, so flooring (not truncating toward zero) is what makes the answers agree.
+    @BeforeAll
+    static void writeFile() throws Exception {
+        file = tmp.resolve("temporal.vortex");
+        DType.Struct schema = DType.structBuilder()
+                .field(ColumnName.of("d"), DateExtensionEncoder.INSTANCE.dtype(false))
+                .field(ColumnName.of("t"), TimeDtype.of(TimeUnit.Milliseconds, false))
+                .field(ColumnName.of("ts"), TimestampDtype.of(TimeUnit.Microseconds, null, false))
+                .build();
+        try (var ch = FileChannel.open(file, StandardOpenOption.CREATE, StandardOpenOption.WRITE);
+             var writer = VortexWriter.create(ch, schema, WriteOptions.defaults())) {
+            writer.writeChunk(Map.of(
+                    ColumnName.of("d"), List.of(LocalDate.of(2024, 2, 29), LocalDate.of(1969, 7, 20)),
+                    ColumnName.of("t"), List.of(LocalTime.of(13, 45, 30, 123_000_000), LocalTime.of(0, 0, 1)),
+                    ColumnName.of("ts"), List.of(Instant.parse("1969-12-31T23:59:59.250500Z"),
+                            Instant.parse("2020-01-01T00:00:00.335Z"))));
+            writer.writeChunk(Map.of(
+                    ColumnName.of("d"), List.of(LocalDate.of(2001, 9, 9)),
+                    ColumnName.of("t"), List.of(LocalTime.of(23, 59, 59, 999_000_000)),
+                    ColumnName.of("ts"), List.of(Instant.parse("2038-01-19T03:14:07.999Z"))));
+        }
+    }
+
+    @ParameterizedTest(name = "{0}")
+    @CsvSource(delimiter = '|', value = {
+            "min(d), max(d)   | min(d + interval '0' day), max(d + interval '0' day)",
+            "min(t), max(t)   | min(t + interval '0' second), max(t + interval '0' second)",
+            "min(ts), max(ts) | min(ts + interval '0' second), max(ts + interval '0' second)"})
+    void minMax_answeredFromStatsEqualsTheScan(String aggregates, String oracle) throws Exception {
+        // Given
+        String sql = "select " + aggregates + " from vtx.t";
+
+        try (Connection conn = VortexCalcite.connect("vtx", Map.of("t", file))) {
+            // When
+            String plan = single(conn, "explain plan for " + sql);
+            String result = row(conn, sql);
+
+            // Then
+            assertThat(plan).contains("Values").doesNotContain("TableScan");
+            assertThat(result).isEqualTo(row(conn, "select " + oracle + " from vtx.t"));
+        }
+    }
+
+    @Test
+    void convert_floorsToTheUnitCalciteUses() {
+        // Given
+        Temporal timestamp = new Temporal(ExtensionId.WellKnown.VORTEX_TIMESTAMP, TimeUnit.Microseconds);
+        Temporal date = new Temporal(ExtensionId.WellKnown.VORTEX_DATE, TimeUnit.Milliseconds);
+
+        // When / Then — before the epoch, a partial millisecond or day floors to the earlier one
+        assertThat(timestamp.convert(-1)).isEqualTo(-1L);
+        assertThat(timestamp.convert(1_999)).isEqualTo(1L);
+        assertThat(date.convert(-1)).isEqualTo(-1);
+        assertThat(date.convert(86_399_999)).isEqualTo(0);
+    }
+
+    @Test
+    void convert_timeOutsideADay_throwsVortexException() {
+        // Given — malformed input: a time of day is in [0, 24h)
+        Temporal time = new Temporal(ExtensionId.WellKnown.VORTEX_TIME, TimeUnit.Seconds);
+
+        // When / Then
+        assertThatThrownBy(() -> time.convert(86_400)).isInstanceOf(VortexException.class);
+        assertThatThrownBy(() -> time.convert(-1)).isInstanceOf(VortexException.class);
+    }
+
+    private static String single(Connection conn, String sql) throws Exception {
+        try (Statement st = conn.createStatement(); ResultSet rs = st.executeQuery(sql)) {
+            rs.next();
+            return rs.getString(1);
+        }
+    }
+
+    private static String row(Connection conn, String sql) throws Exception {
+        try (Statement st = conn.createStatement(); ResultSet rs = st.executeQuery(sql)) {
+            rs.next();
+            return rs.getString(1) + "|" + rs.getString(2);
+        }
+    }
+}