diff --git a/CHANGELOG.md b/CHANGELOG.md index ae9b68d05..b9f741296 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -8,11 +8,16 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] ### Added +- `Chunk.slice(offset, length)` gives a zero-copy view of a row range. ([#550](https://github.com/dfa1/vortex-java/pull/550)) +- The Calcite adapter tells the planner each table's row count, so join order and costing use the real table size. ([#550](https://github.com/dfa1/vortex-java/pull/550)) - `vortex.json` and the `vortex.st.*` geometry columns open without a decoder and read as their storage, as vortex-jni reads them. ([#547](https://github.com/dfa1/vortex-java/pull/547)) - The CLI takes `--allow-unknown` to open files with extension types it has no decoder for, and CSV `ExportOptions.withReadRegistry` picks the registry an export opens the file with. ([#547](https://github.com/dfa1/vortex-java/pull/547)) - 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 +- The Calcite adapter pushes `IN` lists, comparisons with the literal on the left and `IS NULL` into the scan, so they prune chunks instead of being re-checked row by row. ([#550](https://github.com/dfa1/vortex-java/pull/550)) +- The Calcite adapter answers a filtered `COUNT`, `MIN` or `MAX` from Rust's `vortex.zoned` zone maps, the default, decoding only the zones the filter cuts. ([#550](https://github.com/dfa1/vortex-java/pull/550)) +- The Calcite adapter reads a table's schema and row count once instead of reopening the file for every planning call. ([#550](https://github.com/dfa1/vortex-java/pull/550)) - **Breaking:** opening a file with an extension id that has no registered decoder fails unless `ReadRegistry.Builder#allowUnknown()` is set, as in Rust; the column then reads as its storage array. ([#530](https://github.com/dfa1/vortex-java/issues/530)) - **Breaking:** `vortex.date`, `vortex.time`, `vortex.timestamp` and `vortex.uuid` dtypes are checked as Rust checks them (time unit, storage type, metadata) when writing and reading, and a file whose extension dtype has no metadata field is refused, as Rust refuses it. ([#547](https://github.com/dfa1/vortex-java/pull/547)) - **Breaking:** `DateExtensionDecoder.decode` and `decodeAll` take the column's `DType.Extension`, which carries the time unit. ([#547](https://github.com/dfa1/vortex-java/pull/547)) @@ -20,6 +25,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 +- 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)) - `vortex.date` columns in milliseconds, as Rust writes an Arrow `Date64`, decode to the right dates instead of failing or reading the count as days. ([#547](https://github.com/dfa1/vortex-java/pull/547)) - An extension column with no metadata no longer writes a file Rust refuses to read. ([#547](https://github.com/dfa1/vortex-java/pull/547)) diff --git a/calcite/pom.xml b/calcite/pom.xml index 2f0eff1e5..c41131f9b 100644 --- a/calcite/pom.xml +++ b/calcite/pom.xml @@ -10,7 +10,7 @@ vortex-calcite vortex-calcite - Apache Calcite SQL adapter over the Vortex columnar file format (demo: filter/project/aggregate push-down). + Apache Calcite SQL adapter over the Vortex columnar file format with projection, filter and aggregate push-down. diff --git a/calcite/src/main/java/io/github/dfa1/vortex/calcite/RexFilterTranslator.java b/calcite/src/main/java/io/github/dfa1/vortex/calcite/RexFilterTranslator.java index 454f52f2f..ff83817f5 100644 --- a/calcite/src/main/java/io/github/dfa1/vortex/calcite/RexFilterTranslator.java +++ b/calcite/src/main/java/io/github/dfa1/vortex/calcite/RexFilterTranslator.java @@ -3,6 +3,7 @@ import io.github.dfa1.vortex.core.model.ColumnName; import io.github.dfa1.vortex.core.model.DType; import io.github.dfa1.vortex.reader.RowFilter; +import io.github.dfa1.vortex.reader.compute.Predicate; import org.apache.calcite.jdbc.JavaTypeFactoryImpl; import org.apache.calcite.rex.RexBuilder; @@ -32,9 +33,10 @@ final class RexFilterTranslator { private RexFilterTranslator() { } - /// Translates the Calcite predicates into a zone-map [RowFilter], keeping only the comparisons - /// we can prune on (`=`, `<>`, `<`, `<=`, `>`, `>=` between a column and a literal, plus `AND`). - /// Predicates we don't understand are simply not pushed — Calcite still applies them. + /// Translates the Calcite predicates into a zone-map [RowFilter], keeping only what we can prune + /// on: `=`, `<>`, `<`, `<=`, `>`, `>=` between a column and a literal, `IS [NOT] NULL`, a + /// single-column `OR` (such as an `IN` list), and `AND`. Predicates we don't understand are + /// simply not pushed — Calcite still applies them. /// /// @param filters the predicates Calcite pushed onto the scan /// @param names the table's column names, by field index @@ -108,9 +110,8 @@ private static Optional toComparison(RexNode node, List n return comparison(node, names, types, false); } - /// Strict counterpart of [#toComparison]: the same column-vs-literal / `AND` grammar, but a - /// single unrecognized node (or one `AND` conjunct) collapses the whole result to empty rather - /// than being dropped, and bare `IS NULL` / `IS NOT NULL` are also translatable. Used by + /// Strict counterpart of [#toComparison]: the same grammar, but a single unrecognized node (or + /// one `AND` conjunct) collapses the whole result to empty rather than being dropped. Used by /// [#translateStrict] so aggregate push-down answers from stats only when the [RowFilter] /// captures the predicate in full. Delegates to the shared [#comparison] dispatch with /// `strict = true`. @@ -119,14 +120,14 @@ private static Optional strictComparison(RexNode node, List`, `<`, `<=`, `>`, `>=`) translate identically - /// in both walks through [#binary]. + /// (strict). The two differ only in how an untranslatable `AND` conjunct is handled: a `strict` + /// walk abandons the whole predicate on the first one, while a lenient walk drops it and keeps + /// the rest. Comparisons (`=`, `<>`, `<`, `<=`, `>`, `>=`, either operand order) and the null + /// tests translate identically in both walks. + /// + /// An `OR` translates only when every disjunct does and all of them test one column, into one + /// column predicate (an `IN` list arrives this way once `SEARCH` is expanded). Dropping a + /// disjunct would narrow the filter and lose rows, so an `OR` is all-or-nothing in both walks. /// /// @param node the (already `SEARCH`-expanded) predicate node /// @param names the table's column names, by field index @@ -153,13 +154,50 @@ private static Optional comparison(RexNode node, List nam yield parts.isEmpty() ? Optional.empty() : Optional.of(RowFilter.and(parts.toArray(RowFilter[]::new))); } + case OR -> { + ColumnName column = null; + Predicate predicate = null; + for (RexNode operand : call.getOperands()) { + Optional part = comparison(operand, names, types, true) + .flatMap(RexFilterTranslator::singleColumn); + if (part.isEmpty() || column != null && !column.equals(part.get().column())) { + yield Optional.empty(); + } + column = part.get().column(); + predicate = predicate == null ? part.get().predicate() + : new Predicate.Or(predicate, part.get().predicate()); + } + yield Optional.of(new RowFilter.Column(column, predicate)); + } case EQUALS, NOT_EQUALS, LESS_THAN, LESS_THAN_OR_EQUAL, GREATER_THAN, GREATER_THAN_OR_EQUAL -> binary(call, names, types); - case IS_NULL, IS_NOT_NULL -> strict ? nullCheck(call, names) : Optional.empty(); + case IS_NULL, IS_NOT_NULL -> nullCheck(call, names); default -> Optional.empty(); }; } + /// The filter as one column's predicate: itself for a column filter, the conjunction of an `AND` + /// whose parts all test the same column, else empty. + private static Optional singleColumn(RowFilter filter) { + return switch (filter) { + case RowFilter.Column column -> Optional.of(column); + case RowFilter.And(var parts) -> { + ColumnName column = null; + Predicate predicate = null; + for (RowFilter part : parts) { + Optional one = singleColumn(part); + if (one.isEmpty() || column != null && !column.equals(one.get().column())) { + yield Optional.empty(); + } + column = one.get().column(); + predicate = predicate == null ? one.get().predicate() + : new Predicate.And(predicate, one.get().predicate()); + } + yield Optional.of(new RowFilter.Column(column, predicate)); + } + }; + } + /// Translates `col IS NULL` / `col IS NOT NULL` over a bare column reference into the matching /// [RowFilter], which the zone-map fold answers from each zone's null count. Anything other than /// a direct [RexInputRef] operand (e.g. `IS NULL` over an expression) abandons the translation — @@ -177,7 +215,21 @@ private static Optional nullCheck(RexCall call, List name private static Optional binary(RexCall call, List names, List types) { List ops = call.getOperands(); - if (ops.size() != 2 || !(ops.get(0) instanceof RexInputRef ref) || !(ops.get(1) instanceof RexLiteral lit)) { + if (ops.size() != 2) { + return Optional.empty(); + } + SqlKind kind = call.getKind(); + RexInputRef ref; + RexLiteral lit; + if (ops.get(0) instanceof RexInputRef r && ops.get(1) instanceof RexLiteral l) { + ref = r; + lit = l; + } else if (ops.get(0) instanceof RexLiteral l && ops.get(1) instanceof RexInputRef r) { + // `5 < x` is `x > 5` + ref = r; + lit = l; + kind = kind.reverse(); + } else { return Optional.empty(); } Object val = literalValue(lit, types.get(ref.getIndex())); @@ -186,14 +238,14 @@ private static Optional binary(RexCall call, List names, } ColumnName col = names.get(ref.getIndex()); Comparable cmp = (Comparable) val; - return Optional.of(switch (call.getKind()) { + return Optional.of(switch (kind) { case EQUALS -> RowFilter.eq(col, val); case NOT_EQUALS -> RowFilter.neq(col, val); case LESS_THAN -> RowFilter.lt(col, cmp); case LESS_THAN_OR_EQUAL -> RowFilter.lte(col, cmp); case GREATER_THAN -> RowFilter.gt(col, cmp); case GREATER_THAN_OR_EQUAL -> RowFilter.gte(col, cmp); - default -> throw new IllegalStateException("unreachable kind: " + call.getKind()); + default -> throw new IllegalStateException("unreachable kind: " + kind); }); } diff --git a/calcite/src/main/java/io/github/dfa1/vortex/calcite/VortexEnumerator.java b/calcite/src/main/java/io/github/dfa1/vortex/calcite/VortexEnumerator.java index f8436228f..cb42762bb 100644 --- a/calcite/src/main/java/io/github/dfa1/vortex/calcite/VortexEnumerator.java +++ b/calcite/src/main/java/io/github/dfa1/vortex/calcite/VortexEnumerator.java @@ -3,14 +3,17 @@ 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.reader.Chunk; import io.github.dfa1.vortex.reader.RowFilter; import io.github.dfa1.vortex.reader.ScanIterator; import io.github.dfa1.vortex.reader.ScanOptions; import io.github.dfa1.vortex.reader.VortexReader; import io.github.dfa1.vortex.reader.compute.Compute; +import io.github.dfa1.vortex.reader.array.Array; import io.github.dfa1.vortex.reader.array.BoolArray; import io.github.dfa1.vortex.reader.array.ByteArray; +import io.github.dfa1.vortex.reader.array.DecimalArray; import io.github.dfa1.vortex.reader.array.DoubleArray; import io.github.dfa1.vortex.reader.array.Float16Array; import io.github.dfa1.vortex.reader.array.FloatArray; @@ -19,7 +22,12 @@ import io.github.dfa1.vortex.reader.array.MaskedArray; import io.github.dfa1.vortex.reader.array.ShortArray; import io.github.dfa1.vortex.reader.array.VarBinArray; +import io.github.dfa1.vortex.reader.extension.DateExtensionDecoder; +import io.github.dfa1.vortex.reader.extension.TimeExtensionDecoder; +import io.github.dfa1.vortex.reader.extension.TimestampExtensionDecoder; +import io.github.dfa1.vortex.reader.extension.UuidExtensionDecoder; +import org.apache.calcite.avatica.util.ByteString; import org.apache.calcite.linq4j.Enumerator; import java.io.IOException; @@ -168,8 +176,24 @@ private static Object value(Object array, DType type, long r) { case I8, U8 -> ((ByteArray) array).getInt(r); }; case DType.Utf8 _ -> ((VarBinArray) array).getString(r); + case DType.Binary _ -> new ByteString(((VarBinArray) array).getBytes(r)); case DType.Bool _ -> ((BoolArray) array).getBoolean(r); - default -> throw new IllegalStateException("unsupported column dtype: " + type); + case DType.Null _ -> null; + case DType.Decimal _ -> ((DecimalArray) array).getDecimal(r); + // Calcite's internal forms: DATE is days since the epoch (int), TIME milliseconds of the + // day (int), TIMESTAMP milliseconds since the epoch (long) + case DType.Extension ext -> switch (ExtensionId.parse(ext.extensionId())) { + case ExtensionId.WellKnown.VORTEX_DATE -> + Math.toIntExact(DateExtensionDecoder.INSTANCE.decode(ext, (Array) array, r).toEpochDay()); + case ExtensionId.WellKnown.VORTEX_TIME -> + (int) (TimeExtensionDecoder.INSTANCE.decode(ext, (Array) array, r).toNanoOfDay() / 1_000_000); + case ExtensionId.WellKnown.VORTEX_TIMESTAMP -> + TimestampExtensionDecoder.INSTANCE.instant(ext, (Array) array, r).toEpochMilli(); + case ExtensionId.WellKnown.VORTEX_UUID -> UuidExtensionDecoder.INSTANCE.decode((Array) array, r); + default -> value(array, ext.storageDType(), r); + }; + case DType.List _, DType.FixedSizeList _, DType.Struct _, DType.Map _, DType.Union _, + DType.Variant _ -> throw new VortexException("SQL cannot read a " + type + " column yet"); }; } diff --git a/calcite/src/main/java/io/github/dfa1/vortex/calcite/VortexTable.java b/calcite/src/main/java/io/github/dfa1/vortex/calcite/VortexTable.java index 99b3afc41..b761c8471 100644 --- a/calcite/src/main/java/io/github/dfa1/vortex/calcite/VortexTable.java +++ b/calcite/src/main/java/io/github/dfa1/vortex/calcite/VortexTable.java @@ -1,7 +1,9 @@ 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.PType; import io.github.dfa1.vortex.reader.ArrayStats; import io.github.dfa1.vortex.reader.Chunk; @@ -9,6 +11,7 @@ import io.github.dfa1.vortex.reader.ScanIterator; import io.github.dfa1.vortex.reader.ScanOptions; import io.github.dfa1.vortex.reader.VortexReader; +import io.github.dfa1.vortex.reader.Zone; import io.github.dfa1.vortex.reader.compute.Compare; import io.github.dfa1.vortex.reader.compute.Compute; import io.github.dfa1.vortex.reader.compute.FilteredAggregate; @@ -25,6 +28,8 @@ import org.apache.calcite.rel.type.RelDataTypeFactory; import org.apache.calcite.rex.RexNode; import org.apache.calcite.schema.ProjectableFilterableTable; +import org.apache.calcite.schema.Statistic; +import org.apache.calcite.schema.Statistics; import org.apache.calcite.schema.TranslatableTable; import org.apache.calcite.schema.impl.AbstractTable; import org.apache.calcite.sql.type.SqlTypeName; @@ -50,10 +55,22 @@ /// only partially captures is left in `filters` for Calcite's own row check. The win is decoding /// far fewer chunks when the filter is selective on a clustered column, and — for a fully captured /// filter — not re-evaluating it a second time over the chunks that are decoded. +/// +/// The schema and row count are read from the footer on first use and cached: +/// a Vortex file is written once, so they cannot change under the table. public final class VortexTable extends AbstractTable implements ProjectableFilterableTable, TranslatableTable { private final Path file; private final AtomicLong chunksScannedLastQuery = new AtomicLong(); + // Read lazily so a missing file fails at query time, as before; a racing first read only + // reads the footer twice, so no lock. + private volatile Footer footer; + + /// The footer facts every planning call needs, read once instead of reopening the file for each + /// `getRowType`, `scan` and statistic. Column statistics stay out: folding them walks every + /// column's zones, which only the aggregate rule needs. + private record Footer(DType.Struct struct, long totalRows) { + } /// Creates a table backed by the Vortex file at `file`. /// @@ -88,9 +105,9 @@ public record ColumnStats(ArrayStats stats, long totalRows) { /// @param column the column name /// @return the column's aggregated statistics paired with the file's total row count public ColumnStats statsAndRows(ColumnName column) { + long totalRows = footer().totalRows(); try (VortexReader reader = VortexReader.open(file)) { - ArrayStats stats = reader.columnStats().getOrDefault(column, ArrayStats.empty()); - return new ColumnStats(stats, countRows(reader)); + return new ColumnStats(reader.columnStats().getOrDefault(column, ArrayStats.empty()), totalRows); } catch (IOException e) { throw new UncheckedIOException("cannot read stats of " + file, e); } @@ -120,11 +137,12 @@ public record ZoneSum(Number sum, Long nullCount, long totalRows) { /// @param column the numeric column name /// @return the folded sum with the null and row counts needed to interpret a zero public ZoneSum zoneSum(ColumnName column) { + long totalRows = footer().totalRows(); try (VortexReader reader = VortexReader.open(file)) { Number sum = new ZoneReducer(reader).sum(column); Long nullCount = reader.columnStats() .getOrDefault(column, ArrayStats.empty()).nullCount(); - return new ZoneSum(sum, nullCount, countRows(reader)); + return new ZoneSum(sum, nullCount, totalRows); } catch (IOException e) { throw new UncheckedIOException("cannot read zone sum of " + file, e); } @@ -134,15 +152,39 @@ public ZoneSum zoneSum(ColumnName column) { /// /// @return the number of rows in the file public long totalRows() { + return footer().totalRows(); + } + + /// Gives the planner the file's row count, so join ordering and costing see the real table + /// size instead of Calcite's default guess. + /// + /// @return a statistic carrying the row count, with no declared keys + @Override + public Statistic getStatistic() { + return Statistics.of(footer().totalRows(), List.of()); + } + + private Footer footer() { + Footer f = footer; + if (f == null) { + f = readFooter(); + footer = f; + } + return f; + } + + private Footer readFooter() { try (VortexReader reader = VortexReader.open(file)) { - return countRows(reader); + if (!(reader.dtype() instanceof DType.Struct s)) { + throw new IllegalStateException("top-level Vortex dtype is not a struct: " + reader.dtype()); + } + return new Footer(s, countRows(reader)); } catch (IOException e) { - throw new UncheckedIOException("cannot count rows of " + file, e); + throw new UncheckedIOException("cannot read the footer of " + file, e); } } - /// Sums the chunk row counts of `reader` from chunk metadata, without decoding data — shared by - /// [#totalRows()] and [#zoneSum(ColumnName)] so both read the count from a single open reader. + /// Sums the chunk row counts of `reader` from chunk metadata, without decoding data. private static long countRows(VortexReader reader) throws IOException { long total = 0; try (ScanIterator scan = reader.scan(ScanOptions.all())) { @@ -183,14 +225,18 @@ public record FilteredFold(Number sum, Object min, Object max, Long nullCount, l /// `aggColumn` is the column whose values are reduced over the selected rows (`null` for /// `COUNT(*)`, which needs only the selected row count). The result is empty — abandoning to a /// full scan, which is always correctness-safe — when the fold cannot be trusted: a `U64` - /// column involved (whose stat order [#isU64] cannot classify — see the guard below), a - /// zone-map zone count that does not align 1:1 with the chunks, or a boundary zone whose mask - /// cannot be built. + /// column involved (whose stat order [#isU64] cannot classify — see the guard below), columns + /// whose zones lie on different rows, zones that do not tile the chunks' rows, or a boundary + /// zone whose columns cannot be sliced. + /// + /// Zones need not line up with chunks: Rust's `vortex.zoned` writes fixed 8192-row zones while a + /// chunk often spans many of them, so a boundary zone is folded from just its rows of the chunk + /// (or two chunks) holding it. /// /// The fold also abandons for performance, not correctness, when there is at least one boundary /// zone but no fully-selected (IN) zone: the only win the fold has over a scan is folding IN zones /// from statistics without decoding them, so with zero IN zones it would decode exactly the chunks - /// a scan decodes and only add the stats-table classify and mask-building overhead. The + /// a scan decodes and only add the stats-table classify overhead. The /// zone-map-pruned scan is the cheaper answer. An all-OUT selection (no IN, no boundary) is not /// this case — it folds to a correct empty aggregate with no decode and is kept. /// @@ -224,30 +270,35 @@ public Optional filteredFold(RowFilter filter, ColumnName aggColum return Optional.empty(); } } - long[] rowCounts = scan.chunkRowCounts(); - int zones = rowCounts.length; + long[] chunkRows = scan.chunkRowCounts(); + // Every column must place its zones on the same rows, so one zone index classifies + // the whole filter. Rust's vortex.zoned uses fixed zones independent of the chunks; + // the legacy vortex.stats and the no-zone-map fallback have one zone per chunk. + List geometry = null; java.util.Map> zoneStats = new java.util.HashMap<>(); for (ColumnName column : columns) { - List perZone = scan.columnZoneStats(column); - if (perZone.size() != zones) { - return Optional.empty(); // zone-map zones don't align 1:1 with chunks — cannot classify + List zones = zonesOf(scan, column, chunkRows); + if (zones == null || geometry != null && !sameRows(geometry, zones)) { + return Optional.empty(); } - zoneStats.put(column, perZone); + geometry = zones; + zoneStats.put(column, zones.stream().map(Zone::stats).toList()); + } + if (!tiles(geometry, chunkRows)) { + return Optional.empty(); // zones and chunks disagree on the rows } // Classify every zone first, so the IN count is known before any boundary decode. The - // fold's only win over a scan is folding IN zones from statistics with no decode; a - // BOUNDARY zone is decoded exactly as a scan would decode it. So when there is at least - // one BOUNDARY zone but no IN zone, the fold decodes the same chunks a scan does and adds - // the stats-table classify + mask-building overhead on top — a net loss. Abandon to the - // (zone-map-pruned) scan in that case. The all-OUT case (no IN, no BOUNDARY) is NOT this - // case: it folds to a correct empty/zero aggregate with no decode at all, so it is kept. - Match[] matches = new Match[zones]; + // fold's only win over a scan is folding IN zones from statistics with no decode, so + // when there is at least one BOUNDARY zone but no IN zone, abandon to the + // (zone-map-pruned) scan. The all-OUT case (no IN, no BOUNDARY) folds to a correct + // empty/zero aggregate with no decode at all, so it is kept. + Match[] matches = new Match[geometry.size()]; int inZones = 0; int boundaryZones = 0; - for (int zone = 0; zone < zones; zone++) { - Match match = classify(filter, zone, zoneStats, rowCounts[zone], struct); - matches[zone] = match; + for (int z = 0; z < geometry.size(); z++) { + Match match = classify(filter, z, zoneStats, geometry.get(z).rowCount(), struct); + matches[z] = match; if (match == Match.IN) { inZones++; } else if (match == Match.BOUNDARY) { @@ -260,27 +311,32 @@ public Optional filteredFold(RowFilter filter, ColumnName aggColum List decodeColumns = new ArrayList<>(columns); Fold fold = new Fold(); - for (int zone = 0; zone < zones; zone++) { - Match match = matches[zone]; - if (match == Match.OUT) { - continue; - } - if (match == Match.BOUNDARY) { - // Tier 2: a partially-selected zone. Decode this chunk (and the filter columns), - // build a row mask from the filter, and reduce the masked rows. The chunk index - // equals the zone index — the perZone.size() == zones guard above proved the - // zone-map rows align 1:1 with the chunks. - foldBoundaryZone(reader, zone, decodeColumns, filter, aggColumn, fold); - continue; - } - // Tier 1 (Match.IN): every row matches — fold the whole zone from its statistics. - fold.keptRows += rowCounts[zone]; - if (aggColumn != null) { - ArrayStats stats = zoneStats.get(aggColumn).get(zone); - fold.addSum(stats.sum()); - fold.addMin(stats.min()); - fold.addMax(stats.max()); - fold.addNulls(stats.nullCount()); + try (BoundaryChunks chunks = new BoundaryChunks(reader, decodeColumns, chunkRows)) { + for (int z = 0; z < geometry.size(); z++) { + Zone zone = geometry.get(z); + if (matches[z] == Match.OUT) { + continue; + } + if (matches[z] == Match.BOUNDARY) { + // Tier 2: fold the rows the filter selects, from the zone's rows of the + // chunks holding it — a zone may straddle a chunk edge. + if (!chunks.fold(zone, filter, aggColumn, fold)) { + return Optional.empty(); + } + continue; + } + // Tier 1 (Match.IN): every row matches — fold the whole zone from its statistics. + fold.keptRows += zone.rowCount(); + if (aggColumn != null) { + ArrayStats stats = zoneStats.get(aggColumn).get(z); + // a zone with values but no min/max stat (a string zone holds only bounds) + // leaves the extreme unknown, rather than silently ignored + boolean hasValues = stats.nullCount() == null || stats.nullCount() < zone.rowCount(); + fold.addSum(stats.sum()); + fold.addMin(stats.min(), hasValues); + fold.addMax(stats.max(), hasValues); + fold.addNulls(stats.nullCount()); + } } } return Optional.of(fold.result()); @@ -289,30 +345,79 @@ public Optional filteredFold(RowFilter filter, ColumnName aggColum } } - /// Decodes one boundary chunk and folds the aggregate over the rows the filter selects in it, - /// contributing to the running [Fold]. The chunk is released per zone via try-with-resources, so - /// nothing outlives the reduction. - /// - /// The whole filter — an n-ary `AND` of column-bound [Predicate] leaves — and the aggregate fold - /// run in one fused pass over the chunk's rows via - /// [Compute#filteredAggregate(Chunk, RowFilter, ColumnName)], with no intermediate selection bitmap. - /// `SUM` is guarded: the fused kernel reports a `null` sum for a non-numeric aggregate column, - /// which disables the sum (mirroring a zone that records no usable sum) rather than failing the - /// whole fold — a `MIN`/`MAX` over the same column still folds. The aggregate column's null count - /// among the selected rows is the selected count minus the non-null count. - private static void foldBoundaryZone(VortexReader reader, int zone, List columns, RowFilter filter, ColumnName aggColumn, Fold fold) { - try (Chunk chunk = reader.decodeChunk(zone, columns)) { - FilteredAggregate aggregate = Compute.filteredAggregate(chunk, filter, aggColumn); - fold.keptRows += aggregate.selectedRows(); - if (aggColumn == null) { + /// Decodes the chunks that hold boundary zones, each once, in row order, and folds a zone's rows + /// through a [Chunk#slice(long, long)] view. Zones arrive in row order, so only the chunk under + /// the current zone is kept open. + private static final class BoundaryChunks implements AutoCloseable { + private final VortexReader reader; + private final List columns; + private final long[] chunkRows; + private int index = -1; + private long chunkStart; + private Chunk chunk; + + BoundaryChunks(VortexReader reader, List columns, long[] chunkRows) { + this.reader = reader; + this.columns = columns; + this.chunkRows = chunkRows; + } + + /// Folds the rows `filter` selects in `zone`, in one fused pass per chunk part via + /// [Compute#filteredAggregate(Chunk, RowFilter, ColumnName)]. `SUM` is guarded: the kernel + /// reports a `null` sum for a non-numeric aggregate column, which disables only the sum. + /// + /// @return `false` when a column cannot be sliced, so the caller abandons to the scan + boolean fold(Zone zone, RowFilter filter, ColumnName aggColumn, Fold fold) { + long row = zone.firstRow(); + long end = row + zone.rowCount(); + while (row < end) { + moveTo(row); + long offset = row - chunkStart; + long length = Math.min(end, chunkStart + chunkRows[index]) - row; + FilteredAggregate aggregate; + try (Chunk part = chunk.slice(offset, length)) { + aggregate = Compute.filteredAggregate(part, filter, aggColumn); + } catch (VortexException e) { + // ponytail: some array types (F16, variant) have no slice yet; the scan answers + return false; + } + fold.keptRows += aggregate.selectedRows(); + if (aggColumn != null) { + // the kernel's min/max are null only when no selected row has a value + fold.addSum(aggregate.sum()); + fold.addMin(aggregate.min(), false); + fold.addMax(aggregate.max(), false); + fold.addNulls(aggregate.selectedRows() - aggregate.aggNonNullCount()); + } + row += length; + } + return true; + } + + /// Opens the chunk holding `row`; `tiles` proved every row lies in some chunk. + private void moveTo(long row) { + if (chunk != null && row < chunkStart + chunkRows[index]) { return; } - // A null sum marks a non-numeric aggregate column — unanswerable, like a missing zone sum; - // addSum(null) disables only the sum, leaving MIN/MAX to fold. - fold.addSum(aggregate.sum()); - fold.addMin(aggregate.min()); - fold.addMax(aggregate.max()); - fold.addNulls(aggregate.selectedRows() - aggregate.aggNonNullCount()); + if (chunk != null) { + chunk.close(); + chunk = null; + } + if (index < 0) { + index = 0; + } + while (row >= chunkStart + chunkRows[index]) { + chunkStart += chunkRows[index]; + index++; + } + chunk = reader.decodeChunk(index, columns); + } + + @Override + public void close() { + if (chunk != null) { + chunk.close(); + } } } @@ -327,6 +432,7 @@ private static final class Fold { private boolean sumUsable = true; private Object min; private Object max; + private boolean extremesKnown = true; private long nullCount; private boolean nullCountKnown = true; private long keptRows; @@ -348,11 +454,18 @@ private void addSum(Object contribution) { } } - private void addMin(Object candidate) { + /// Adds a min candidate; a `null` one from rows that do hold values makes the min unknown. + private void addMin(Object candidate, boolean hasValues) { + if (candidate == null && hasValues) { + extremesKnown = false; + } min = pickExtreme(min, candidate, true); } - private void addMax(Object candidate) { + private void addMax(Object candidate, boolean hasValues) { + if (candidate == null && hasValues) { + extremesKnown = false; + } max = pickExtreme(max, candidate, false); } @@ -378,7 +491,10 @@ private FilteredFold result() { foldedSum = longSum; } Long foldedNulls = nullCountKnown ? nullCount : null; - return new FilteredFold(foldedSum, min, max, foldedNulls, keptRows); + // an unknown extreme is reported as absent, which the rule cannot mistake for an empty + // selection while rows were kept + return new FilteredFold(foldedSum, extremesKnown ? min : null, extremesKnown ? max : null, + foldedNulls, keptRows); } } @@ -399,6 +515,55 @@ public Optional translatePushedFilters(List filters) { return RexFilterTranslator.translateStrict(filters, names, types); } + /// A column's zones: its zone map when the zones can be placed on rows, else one zone per chunk + /// from the chunks' own statistics (the legacy `vortex.stats` table with batches of mixed + /// length, or no zone map at all). `null` when neither lines up with the chunks. + private static List zonesOf(ScanIterator scan, ColumnName column, long[] chunkRows) { + List zones = scan.columnZones(column); + if (!zones.isEmpty()) { + return zones; + } + List perChunk = scan.columnZoneStats(column); + if (perChunk.size() != chunkRows.length) { + return null; + } + List out = new ArrayList<>(chunkRows.length); + long first = 0; + for (int c = 0; c < chunkRows.length; c++) { + out.add(new Zone(first, chunkRows[c], perChunk.get(c))); + first += chunkRows[c]; + } + return out; + } + + private static boolean sameRows(List a, List b) { + if (a.size() != b.size()) { + return false; + } + for (int i = 0; i < a.size(); i++) { + if (a.get(i).firstRow() != b.get(i).firstRow() || a.get(i).rowCount() != b.get(i).rowCount()) { + return false; + } + } + return true; + } + + /// Whether `zones` cover the chunks' rows exactly, in order and with no gap or overlap. + private static boolean tiles(List zones, long[] chunkRows) { + long next = 0; + for (Zone zone : zones) { + if (zone.firstRow() != next) { + return false; + } + next += zone.rowCount(); + } + long total = 0; + for (long rows : chunkRows) { + total += rows; + } + return next == total; + } + /// How a zone relates to a filter: every row matches ([Match#IN]), no row matches /// ([Match#OUT]), or the zone straddles the predicate ([Match#BOUNDARY]) and can be folded only /// by decoding it. @@ -455,10 +620,9 @@ private static Match classifyColumn(Predicate predicate, ArrayStats s, long rowC /// [Predicate] unification: an unrecognized stat shape or a partially-overlapping zone is /// [Match#BOUNDARY], a zone provably outside the predicate is [Match#OUT], and a zone every row /// of which matches (which, for a value comparison, also requires the zone to carry no nulls) is - /// [Match#IN]. The composite and range predicates ([Predicate.Between] / [Predicate.And] / - /// [Predicate.Or]) cannot arise from the Calcite translation — `SEARCH` / `BETWEEN` expand to an - /// `AND` of single-leaf columns — so they fall through to [Match#BOUNDARY], which decodes and - /// applies the kernel mask and is therefore always correct. + /// [Match#IN]. A single-column [Predicate.And] / [Predicate.Or] (an `IN` list, a range union) + /// combines its sides' verdicts; [Predicate.Between] cannot arise from the Calcite translation + /// and stays [Match#BOUNDARY], which decodes and is therefore always correct. /// /// @param predicate the column-bound value-test /// @param s the zone's statistics for that column @@ -519,8 +683,22 @@ private static Match classifyColumnRaw(Predicate predicate, ArrayStats s, long r yield nulls == rowCount ? Match.OUT : Match.BOUNDARY; } case Predicate.Between _ -> Match.BOUNDARY; - case Predicate.And _ -> Match.BOUNDARY; - case Predicate.Or _ -> Match.BOUNDARY; + case Predicate.And(var left, var right) -> { + Match l = classifyColumnRaw(left, s, rowCount); + Match r = classifyColumnRaw(right, s, rowCount); + if (l == Match.OUT || r == Match.OUT) { + yield Match.OUT; + } + yield l == Match.IN && r == Match.IN ? Match.IN : Match.BOUNDARY; + } + case Predicate.Or(var left, var right) -> { + Match l = classifyColumnRaw(left, s, rowCount); + Match r = classifyColumnRaw(right, s, rowCount); + if (l == Match.IN || r == Match.IN) { + yield Match.IN; + } + yield l == Match.OUT && r == Match.OUT ? Match.OUT : Match.BOUNDARY; + } }; } @@ -699,14 +877,7 @@ public Enumerator enumerator() { } private DType.Struct struct() { - try (VortexReader reader = VortexReader.open(file)) { - if (!(reader.dtype() instanceof DType.Struct s)) { - throw new IllegalStateException("top-level Vortex dtype is not a struct: " + reader.dtype()); - } - return s; - } catch (IOException e) { - throw new UncheckedIOException("cannot read schema of " + file, e); - } + return footer().struct(); } private static int[] allColumns(int n) { @@ -722,6 +893,12 @@ private static int[] allColumns(int n) { /// `U32`->`BIGINT`); `U64` is the exception — no wider SQL integer exists, so it maps to /// `BIGINT` and [VortexEnumerator] guards the high-half values that cannot fit. /// + /// `vortex.timestamp` is a `TIMESTAMP` holding the stored value as a UTC wall clock, zone or + /// not: Avatica shifts `TIMESTAMP WITH LOCAL TIME ZONE` by the JVM's zone in its JDBC getters, + /// so a zoned instant would read back wrong. An extension with no SQL counterpart reads as its storage + /// type. Nested types (list, struct, map, union, variant) map to `ANY` so the rest of the table + /// stays queryable; selecting one fails in [VortexEnumerator]. + /// /// @param factory the Calcite type factory /// @param type the Vortex column dtype /// @return the nullable-adjusted SQL type @@ -737,8 +914,19 @@ private static RelDataType toSqlType(RelDataTypeFactory factory, DType type) { case I8 -> factory.createSqlType(SqlTypeName.TINYINT); }; case DType.Utf8 _ -> factory.createSqlType(SqlTypeName.VARCHAR); + case DType.Binary _ -> factory.createSqlType(SqlTypeName.VARBINARY); case DType.Bool _ -> factory.createSqlType(SqlTypeName.BOOLEAN); - default -> throw new IllegalStateException("unsupported column dtype: " + type); + case DType.Null _ -> factory.createSqlType(SqlTypeName.NULL); + 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); + case ExtensionId.WellKnown.VORTEX_UUID -> factory.createSqlType(SqlTypeName.UUID); + default -> toSqlType(factory, ext.storageDType()); + }; + case DType.List _, DType.FixedSizeList _, DType.Struct _, DType.Map _, DType.Union _, + DType.Variant _ -> factory.createSqlType(SqlTypeName.ANY); }; return factory.createTypeWithNullability(sql, type.nullable()); } diff --git a/calcite/src/test/java/io/github/dfa1/vortex/calcite/ColumnTypesSqlTest.java b/calcite/src/test/java/io/github/dfa1/vortex/calcite/ColumnTypesSqlTest.java new file mode 100644 index 000000000..2c21b1dc6 --- /dev/null +++ b/calcite/src/test/java/io/github/dfa1/vortex/calcite/ColumnTypesSqlTest.java @@ -0,0 +1,140 @@ +package io.github.dfa1.vortex.calcite; + +import io.github.dfa1.vortex.core.model.ColumnName; +import io.github.dfa1.vortex.core.model.DType; +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 io.github.dfa1.vortex.writer.encode.ListData; +import io.github.dfa1.vortex.writer.encode.UuidExtensionEncoder; + +import org.junit.jupiter.api.BeforeAll; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.io.TempDir; + +import java.math.BigDecimal; +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.SQLException; +import java.sql.Statement; +import java.time.Instant; +import java.time.LocalDate; +import java.time.LocalDateTime; +import java.time.LocalTime; +import java.time.ZoneId; +import java.time.ZoneOffset; +import java.util.List; +import java.util.Map; +import java.util.UUID; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; + +/// Column types beyond numbers, strings and booleans. Before, `getRowType` threw on the first +/// decimal, binary, temporal, uuid or nested column, so a file with a single timestamp could not be +/// queried at all, not even its other columns. +class ColumnTypesSqlTest { + + private static final LocalDate DATE = LocalDate.of(2024, 2, 29); + private static final LocalTime TIME = LocalTime.of(13, 45, 30, 123_000_000); + // pre-epoch on purpose: millisecond truncation must floor, not round toward zero + private static final Instant INSTANT = Instant.parse("1969-12-31T23:59:59.250Z"); + private static final UUID ID = UUID.fromString("123e4567-e89b-12d3-a456-426614174000"); + + @TempDir + static Path tmp; + private static Path file; + + @BeforeAll + static void write() throws Exception { + file = tmp.resolve("types.vortex"); + DType.Struct schema = DType.structBuilder() + .field(ColumnName.of("id"), DType.I32) + .field(ColumnName.of("amount"), DType.decimal(10, 2)) + .field(ColumnName.of("blob"), DType.BINARY) + .field(ColumnName.of("day"), DateExtensionEncoder.INSTANCE.dtype(false)) + .field(ColumnName.of("at"), TimeDtype.of(TimeUnit.Milliseconds, false)) + .field(ColumnName.of("wall"), TimestampDtype.of(TimeUnit.Microseconds, null, false)) + .field(ColumnName.of("instant"), TimestampDtype.of(TimeUnit.Microseconds, ZoneId.of("UTC"), false)) + .field(ColumnName.of("uuid"), UuidExtensionEncoder.INSTANCE.dtype(false)) + .field(ColumnName.of("tags"), new DType.List(DType.I64, false)) + .build(); + try (FileChannel ch = FileChannel.open(file, StandardOpenOption.CREATE, StandardOpenOption.WRITE); + VortexWriter writer = VortexWriter.create(ch, schema, WriteOptions.defaults())) { + writer.writeChunk(Map.of( + ColumnName.of("id"), new int[]{1}, + ColumnName.of("amount"), new BigDecimal[]{new BigDecimal("12.34")}, + ColumnName.of("blob"), new byte[][]{{1, 2, 3}}, + ColumnName.of("day"), List.of(DATE), + ColumnName.of("at"), List.of(TIME), + ColumnName.of("wall"), List.of(INSTANT), + ColumnName.of("instant"), List.of(INSTANT), + ColumnName.of("uuid"), List.of(ID), + ColumnName.of("tags"), new ListData(new long[]{7, 8}, new long[]{0, 2}, 1))); + } + } + + @Test + void everyScalarType_readsThroughSql() throws Exception { + // Given + String sql = "select amount, blob, `day`, at, wall, `instant`, `uuid` from vtx.t"; + + // When + try (Connection conn = VortexCalcite.connect("vtx", Map.of("t", file)); + Statement st = conn.createStatement(); + ResultSet result = st.executeQuery(sql)) { + + // Then + assertThat(result.next()).isTrue(); + assertThat(result.getBigDecimal("amount")).isEqualByComparingTo("12.34"); + 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"); + // 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)); + // a zoned timestamp too: the instant as a UTC wall clock, the same for every JVM zone + assertThat(result.getTimestamp("instant").toLocalDateTime()) + .isEqualTo(LocalDateTime.ofInstant(INSTANT, ZoneOffset.UTC)); + assertThat(result.getObject("uuid")).isEqualTo(ID); + assertThat(result.next()).isFalse(); + } + } + + @Test + void nestedColumn_doesNotBlockTheOtherColumns() throws Exception { + // Given a table with a list column the adapter cannot read yet + + // When + try (Connection conn = VortexCalcite.connect("vtx", Map.of("t", file)); + Statement st = conn.createStatement(); + ResultSet result = st.executeQuery("select id from vtx.t")) { + + // Then + assertThat(result.next()).isTrue(); + assertThat(result.getInt("id")).isEqualTo(1); + } + } + + @Test + void nestedColumn_failsLoudWhenSelected() throws Exception { + // Given + try (Connection conn = VortexCalcite.connect("vtx", Map.of("t", file)); + Statement st = conn.createStatement()) { + + // When / Then + assertThatThrownBy(() -> { + try (ResultSet rs = st.executeQuery("select tags from vtx.t")) { + rs.next(); + } + }).isInstanceOf(SQLException.class).hasStackTraceContaining("cannot read a List"); + } + } +} diff --git a/calcite/src/test/java/io/github/dfa1/vortex/calcite/FilterTranslationTest.java b/calcite/src/test/java/io/github/dfa1/vortex/calcite/FilterTranslationTest.java new file mode 100644 index 000000000..e367fab7e --- /dev/null +++ b/calcite/src/test/java/io/github/dfa1/vortex/calcite/FilterTranslationTest.java @@ -0,0 +1,132 @@ +package io.github.dfa1.vortex.calcite; + +import io.github.dfa1.vortex.core.model.ColumnName; +import io.github.dfa1.vortex.core.model.DType; +import io.github.dfa1.vortex.writer.VortexWriter; +import io.github.dfa1.vortex.writer.encode.NullableData; + +import org.apache.calcite.jdbc.JavaTypeFactoryImpl; +import org.apache.calcite.linq4j.Enumerator; +import org.apache.calcite.rel.type.RelDataType; +import org.apache.calcite.rex.RexBuilder; +import org.apache.calcite.rex.RexNode; +import org.apache.calcite.sql.fun.SqlStdOperatorTable; +import org.apache.calcite.sql.type.SqlTypeName; +import org.junit.jupiter.api.BeforeAll; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.io.TempDir; + +import java.math.BigDecimal; +import java.nio.channels.FileChannel; +import java.nio.file.Path; +import java.nio.file.StandardOpenOption; +import java.util.ArrayList; +import java.util.List; +import java.util.Map; + +import static org.assertj.core.api.Assertions.assertThat; + +/// Predicate shapes the translator used to leave to Calcite: an `IN` list (an `OR` of equalities +/// once Calcite's `SEARCH` is expanded), a literal on the left, and `IS NULL`. Left untranslated +/// they neither prune chunks nor are enforced by the scan, so Calcite re-checks every row. +class FilterTranslationTest { + + private static final RexBuilder REX = new RexBuilder(new JavaTypeFactoryImpl()); + private static final RelDataType BIGINT = REX.getTypeFactory().createSqlType(SqlTypeName.BIGINT); + private static final RelDataType NULLABLE_BIGINT = REX.getTypeFactory() + .createTypeWithNullability(BIGINT, true); + + @TempDir + static Path tmp; + private static Path file; + + // two chunks: id 1..3 with no nulls, id 4..6 where n is null on id 5 + @BeforeAll + static void write() throws Exception { + file = tmp.resolve("translate.vortex"); + DType.Struct schema = DType.structBuilder() + .field(ColumnName.of("id"), DType.I64) + .field(ColumnName.of("n"), DType.I64.asNullable()) + .build(); + try (FileChannel ch = FileChannel.open(file, StandardOpenOption.CREATE, StandardOpenOption.WRITE); + VortexWriter w = VortexWriter.create(ch, schema, LegacyZoneMaps.OPTIONS)) { + w.writeChunk(Map.of(ColumnName.of("id"), new long[]{1, 2, 3}, + ColumnName.of("n"), new NullableData(new long[]{10, 20, 30}, new boolean[]{true, true, true}))); + w.writeChunk(Map.of(ColumnName.of("id"), new long[]{4, 5, 6}, + ColumnName.of("n"), new NullableData(new long[]{40, 0, 60}, new boolean[]{true, false, true}))); + } + } + + @Test + void inList_isEnforcedAndPrunes() throws Exception { + // Given id IN (1, 3) + RexNode in = REX.makeIn(REX.makeInputRef(BIGINT, 0), List.of(literal(1), literal(3))); + + // When / Then — only the first chunk is decoded, and the scan enforces the list itself + assertScan(in, List.of(1L, 3L), 1); + } + + @Test + void inListOverTwoColumns_isLeftToCalcite() throws Exception { + // Given id = 1 OR n = 60 — no single-column predicate captures it + RexNode or = REX.makeCall(SqlStdOperatorTable.OR, + REX.makeCall(SqlStdOperatorTable.EQUALS, REX.makeInputRef(BIGINT, 0), literal(1)), + REX.makeCall(SqlStdOperatorTable.EQUALS, REX.makeInputRef(NULLABLE_BIGINT, 1), literal(60))); + List filters = new ArrayList<>(List.of(or)); + + // When + List result = ids(filters); + + // Then — nothing pushed, so Calcite keeps the predicate and the scan returns every row + assertThat(filters).hasSize(1); + assertThat(result).hasSize(6); + } + + @Test + void literalOnTheLeft_isEnforcedAndPrunes() throws Exception { + // Given 3 >= id + RexNode cmp = REX.makeCall(SqlStdOperatorTable.GREATER_THAN_OR_EQUAL, literal(3), REX.makeInputRef(BIGINT, 0)); + + // When / Then + assertScan(cmp, List.of(1L, 2L, 3L), 1); + } + + @Test + void isNull_prunesChunksWithoutNulls() throws Exception { + // Given n IS NULL — the first chunk has a zero null count + RexNode isNull = REX.makeCall(SqlStdOperatorTable.IS_NULL, REX.makeInputRef(NULLABLE_BIGINT, 1)); + + // When / Then + assertScan(isNull, List.of(5L), 1); + } + + private static void assertScan(RexNode predicate, List expectedIds, long expectedChunks) { + // When + List filters = new ArrayList<>(List.of(predicate)); + VortexTable table = new VortexTable(file); + List result = ids(table, filters); + + // Then + assertThat(filters).as("captured in full, so Calcite does not re-check").isEmpty(); + assertThat(result).containsExactlyElementsOf(expectedIds); + assertThat(table.chunksScannedLastQuery()).as("chunks decoded").isEqualTo(expectedChunks); + } + + private static List ids(List filters) { + return ids(new VortexTable(file), filters); + } + + private static List ids(VortexTable table, List filters) { + List ids = new ArrayList<>(); + try (Enumerator en = table.scan(null, filters, new int[]{0}).enumerator()) { + while (en.moveNext()) { + ids.add(en.current()[0]); + } + } + return ids; + } + + private static RexNode literal(long value) { + return REX.makeExactLiteral(BigDecimal.valueOf(value), BIGINT); + } +} diff --git a/calcite/src/test/java/io/github/dfa1/vortex/calcite/LegacyZoneMaps.java b/calcite/src/test/java/io/github/dfa1/vortex/calcite/LegacyZoneMaps.java index d28d9a320..be998f7f9 100644 --- a/calcite/src/test/java/io/github/dfa1/vortex/calcite/LegacyZoneMaps.java +++ b/calcite/src/test/java/io/github/dfa1/vortex/calcite/LegacyZoneMaps.java @@ -5,8 +5,8 @@ /// Editions targeting `core2025.10.0`, the last core edition before Rust's `vortex.zoned`: the /// writer then emits the legacy `vortex.stats` zone map, one zone per `writeChunk` batch with a -/// per-zone sum — the shape the zone fold folds. Default writes use 8192-row `vortex.zoned` zones -/// without sums, which the fold cannot use yet, so those aggregates fall back to a scan. +/// per-zone sum. Default writes use 8192-row `vortex.zoned` zones without sums, so a `SUM` over +/// them falls back to a scan. final class LegacyZoneMaps { static final WriteOptions OPTIONS = WriteOptions.defaults().withEdition(Editions.CORE_2025_10_0); diff --git a/calcite/src/test/java/io/github/dfa1/vortex/calcite/VortexAdapterCoverageTest.java b/calcite/src/test/java/io/github/dfa1/vortex/calcite/VortexAdapterCoverageTest.java index b8da92fbf..dc431fed5 100644 --- a/calcite/src/test/java/io/github/dfa1/vortex/calcite/VortexAdapterCoverageTest.java +++ b/calcite/src/test/java/io/github/dfa1/vortex/calcite/VortexAdapterCoverageTest.java @@ -137,6 +137,35 @@ void table_totalRowsAndStats() { assertThat(columnStats.totalRows()).isEqualTo(3); } + @Test + void getStatistic_reportsTheRowCount() { + // Given + VortexTable table = new VortexTable(file); + + // When + Double result = table.getStatistic().getRowCount(); + + // Then — Calcite's default (no statistic) is a null row count, which makes it guess the + // table size when ordering joins + assertThat(result).isEqualTo(3.0); + } + + @Test + void schemaAndRowCount_areReadFromTheFooterOnce(@TempDir Path dir) throws Exception { + // Given a table whose footer was read, then the file goes away + Path copy = java.nio.file.Files.copy(file, dir.resolve("copy.vortex")); + VortexTable table = new VortexTable(copy); + table.getRowType(new JavaTypeFactoryImpl()); + java.nio.file.Files.delete(copy); + + // When — planning calls that used to reopen the file each time + long result = table.totalRows(); + + // Then + assertThat(result).isEqualTo(3); + assertThat(table.getRowType(new JavaTypeFactoryImpl()).getFieldCount()).isEqualTo(12); + } + @Test void enumerator_resetUnsupportedAndCloseReleasesOpenChunk() { // Given a scan positioned inside the first chunk diff --git a/calcite/src/test/java/io/github/dfa1/vortex/calcite/ZonedAggregateFoldTest.java b/calcite/src/test/java/io/github/dfa1/vortex/calcite/ZonedAggregateFoldTest.java new file mode 100644 index 000000000..3e4bfd047 --- /dev/null +++ b/calcite/src/test/java/io/github/dfa1/vortex/calcite/ZonedAggregateFoldTest.java @@ -0,0 +1,132 @@ +package io.github.dfa1.vortex.calcite; + +import io.github.dfa1.vortex.core.model.ColumnName; +import io.github.dfa1.vortex.core.model.DType; +import io.github.dfa1.vortex.writer.VortexWriter; +import io.github.dfa1.vortex.writer.WriteOptions; + +import org.junit.jupiter.api.BeforeAll; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.io.TempDir; + +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.util.Map; + +import static org.assertj.core.api.Assertions.assertThat; + +/// Filtered aggregates over Rust's `vortex.zoned` zone map, the default since #447: fixed 8192-row +/// zones that do not line up with chunks. The fold required one zone per chunk, so on these files +/// every filtered aggregate fell back to a scan. +class ZonedAggregateFoldTest { + + private static final int ZONE = 8_192; + // batches that straddle zone edges, so zones and chunks disagree + private static final int BATCH = 5_000; + private static final int ROWS = 12 * ZONE; + + @TempDir + static Path tmp; + private static Path file; + + @BeforeAll + static void write() throws Exception { + file = tmp.resolve("zoned.vortex"); + DType.Struct schema = DType.structBuilder() + .field(ColumnName.of("id"), DType.I64) + .field(ColumnName.of("val"), DType.I64) + .field(ColumnName.of("s"), DType.UTF8) + .build(); + try (FileChannel ch = FileChannel.open(file, StandardOpenOption.CREATE, StandardOpenOption.WRITE); + VortexWriter writer = VortexWriter.create(ch, schema, WriteOptions.defaults())) { + for (int first = 0; first < ROWS; first += BATCH) { + int n = Math.min(BATCH, ROWS - first); + long[] id = new long[n]; + long[] val = new long[n]; + String[] s = new String[n]; + for (int i = 0; i < n; i++) { + id[i] = first + i; + val[i] = 3L * (first + i); + s[i] = String.format("k%06d", first + i); + } + writer.writeChunk(Map.of(ColumnName.of("id"), id, ColumnName.of("val"), val, ColumnName.of("s"), s)); + } + } + } + + @Test + void countMinMax_foldFromZonesAndBoundaryChunks() throws Exception { + // Given a range that covers several whole zones and cuts two of them + long lo = ZONE + 100; + long hi = 9L * ZONE + 300; + String sql = "select count(*) c, count(val) cv, min(val) mn, max(val) mx from vtx.t where id between " + + lo + " and " + hi; + + try (Connection conn = VortexCalcite.connect("vtx", Map.of("t", file))) { + // When + String result = explain(conn, sql); + + // Then — answered without a scan, and the values match the data + assertThat(result).containsIgnoringCase("Values").doesNotContain("TableScan"); + try (Statement st = conn.createStatement(); ResultSet rs = st.executeQuery(sql)) { + rs.next(); + assertThat(rs.getLong("c")).isEqualTo(hi - lo + 1); + assertThat(rs.getLong("cv")).isEqualTo(hi - lo + 1); + assertThat(rs.getLong("mn")).isEqualTo(3 * lo); + assertThat(rs.getLong("mx")).isEqualTo(3 * hi); + } + } + } + + @Test + void sum_hasNoZoneStatistic_soItScans() throws Exception { + // Given — Rust's zoned aggregate set records no sum, so a whole zone cannot be folded + long lo = ZONE + 100; + long hi = 9L * ZONE + 300; + String sql = "select sum(val) from vtx.t where id between " + lo + " and " + hi; + + try (Connection conn = VortexCalcite.connect("vtx", Map.of("t", file))) { + // When + String result = explain(conn, sql); + + // Then + assertThat(result).contains("TableScan"); + try (Statement st = conn.createStatement(); ResultSet rs = st.executeQuery(sql)) { + rs.next(); + assertThat(rs.getLong(1)).isEqualTo(3 * (hi * (hi + 1) / 2 - (lo - 1) * lo / 2)); + } + } + } + + @Test + void minOfAString_withoutAnExactZoneStatistic_isNotFoldedFromTheBoundaryAlone() throws Exception { + // Given — a zoned string column records only truncated bounds, so whole zones carry no min; + // folding only the boundary chunks would answer the min of the cut zone, not of the range + long lo = ZONE + 100; + long hi = 9L * ZONE + 300; + String sql = "select min(s), max(s) from vtx.t where id between " + lo + " and " + hi; + + try (Connection conn = VortexCalcite.connect("vtx", Map.of("t", file)); + Statement st = conn.createStatement(); + // When + ResultSet result = st.executeQuery(sql)) { + + // Then + result.next(); + assertThat(result.getString(1)).isEqualTo(String.format("k%06d", lo)); + assertThat(result.getString(2)).isEqualTo(String.format("k%06d", hi)); + } + } + + private static String explain(Connection conn, String sql) throws Exception { + try (Statement st = conn.createStatement(); + ResultSet rs = st.executeQuery("explain plan for " + sql)) { + rs.next(); + return rs.getString(1); + } + } +} diff --git a/core/src/main/java/io/github/dfa1/vortex/core/model/DType.java b/core/src/main/java/io/github/dfa1/vortex/core/model/DType.java index f807b8d1d..fe0e1d588 100644 --- a/core/src/main/java/io/github/dfa1/vortex/core/model/DType.java +++ b/core/src/main/java/io/github/dfa1/vortex/core/model/DType.java @@ -365,6 +365,13 @@ record Extension( throw new VortexException("extension metadata too large: " + metadata.byteSize() + " > " + MAX_METADATA_SIZE.bytes()); } + // A dtype outlives the reader that parsed it (callers keep reader.dtype()), but a slice + // of the file mapping fails once the reader closes: copy it onto the heap. Small, as + // the cap above bounds it. + if (metadata != null && metadata.isNative()) { + metadata = MemorySegment.ofArray(metadata.toArray(java.lang.foreign.ValueLayout.JAVA_BYTE)) + .asReadOnly(); + } } /// Value equality with content-based metadata comparison. diff --git a/core/src/test/java/io/github/dfa1/vortex/core/model/DTypeExtensionTest.java b/core/src/test/java/io/github/dfa1/vortex/core/model/DTypeExtensionTest.java index cd2ad1f7f..dcf51c9e5 100644 --- a/core/src/test/java/io/github/dfa1/vortex/core/model/DTypeExtensionTest.java +++ b/core/src/test/java/io/github/dfa1/vortex/core/model/DTypeExtensionTest.java @@ -36,6 +36,22 @@ void metadataOverCap_rejected() { .hasMessageContaining("metadata too large"); } + @Test + void nativeMetadata_outlivesItsArena() { + // Given metadata sliced from off-heap memory, as a reader parses it out of the file mapping + DType.Extension result; + try (java.lang.foreign.Arena arena = java.lang.foreign.Arena.ofConfined()) { + MemorySegment meta = arena.allocate(2); + meta.set(java.lang.foreign.ValueLayout.JAVA_BYTE, 0, (byte) 3); + + // When + result = new DType.Extension("vortex.timestamp", DType.I64, meta, false); + } + + // Then — the dtype is still readable after the arena (the reader) is closed + assertThat(result.metadata().get(java.lang.foreign.ValueLayout.JAVA_BYTE, 0)).isEqualTo((byte) 3); + } + @Test void nullMetadata_accepted() { // Given / When / Then — date/uuid extensions carry no metadata; null must remain valid diff --git a/docs/reference.md b/docs/reference.md index 849bac9b0..b1833afbc 100644 --- a/docs/reference.md +++ b/docs/reference.md @@ -269,6 +269,7 @@ never desync. | `columns()` | `SequencedMap`, schema order, unmodifiable | | ` column(ColumnName name)` | Typed column lookup; throws `VortexException` if absent | | `as(ColumnName name, Class domainType)` | Extension column → typed `List`, via the registered `ExtensionDecoder` | +| `slice(long offset, long length)` | Zero-copy view of a row range; valid while this chunk is open, closing it does nothing | | `isClosed()` | Whether `close()` has run | | `close()` | Releases the chunk's arena. Idempotent. | @@ -611,11 +612,31 @@ preserved) and the **Babel** SQL parser, so a column named after a reserved word keywords that open a typed literal (`date`, `time`, `timestamp`, `interval`), which stay reserved even under Babel and need back-ticks: `` select `date` from vtx.ohlc ``. -Whole-table aggregates (`min`/`max`/`sum`) are answered from zone-map statistics via -`VortexAggregatePushDownRule` where possible — no data segment is decoded. The fold needs one zone -per chunk and a per-zone sum, i.e. the legacy `vortex.stats` zone map (writes targeting an edition -before `core2026.08.0`); files with Rust's `vortex.zoned` — the default since #447, and every -Rust-written file — answer the same aggregates through a scan. `connect` is the +A `WHERE` of those predicates (an `OR` only within one column) prunes chunks from zone maps and is +enforced by the scan, so Calcite does not re-check it; anything else is left to Calcite. + +Column types map as follows. Unsigned integers widen to the next signed type (`U64` stays `BIGINT` +and fails on values past 2^63); `F16` reads as `REAL`. + +| Vortex | SQL | +|---------------------------------------------|---------------------------------------------------| +| `bool`, `utf8`, `binary` | `BOOLEAN`, `VARCHAR`, `VARBINARY` | +| `decimal(p, s)` | `DECIMAL(p, s)` | +| `vortex.date`, `vortex.time`, `vortex.uuid` | `DATE`, `TIME`, `UUID` | +| `vortex.timestamp`, with or without a zone | `TIMESTAMP`, the stored instant as a UTC wall clock | +| other extensions | their storage type | +| list, struct, map, union, variant | `ANY`; selecting the column fails | + +`DATE`, `TIME` and `TIMESTAMP` values carry millisecond precision. A zoned timestamp is not +`TIMESTAMP WITH LOCAL TIME ZONE` because Avatica's JDBC getters shift that type by the JVM's zone. + +`count`/`min`/`max`/`sum` (and `avg`, reduced to `sum`/`count`) are answered from zone-map +statistics via `VortexAggregatePushDownRule` where possible, with or without a `WHERE` of +column-vs-literal comparisons, `IN` lists and `IS [NOT] NULL` joined by `AND`: zones the filter selects entirely fold from their statistics, and only +the rows of zones it cuts are decoded. `sum` needs a per-zone sum, which only the legacy +`vortex.stats` zone map records (writes targeting an edition before `core2026.08.0`); with Rust's +`vortex.zoned` — the default since #447, and every Rust-written file — `sum` is answered by a scan. +`connect` is the supported entry point; `VortexSchema`/`VortexTable` (`io.github.dfa1.vortex.calcite`) are the underlying schema/table plumbing it assembles, exposed for callers building a Calcite schema tree by hand instead of going through `connect`. 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 new file mode 100644 index 000000000..5f5f07326 --- /dev/null +++ b/integration/src/test/java/io/github/dfa1/vortex/integration/CalciteInteropIntegrationTest.java @@ -0,0 +1,228 @@ +package io.github.dfa1.vortex.integration; + +import dev.vortex.api.Session; +import dev.vortex.api.VortexWriter; +import dev.vortex.arrow.ArrowAllocation; +import dev.vortex.jni.NativeLoader; +import io.github.dfa1.vortex.calcite.VortexCalcite; +import io.github.dfa1.vortex.calcite.VortexTable; +import org.apache.arrow.c.ArrowArray; +import org.apache.arrow.c.ArrowSchema; +import org.apache.arrow.c.Data; +import org.apache.arrow.memory.BufferAllocator; +import org.apache.arrow.vector.BigIntVector; +import org.apache.arrow.vector.DateDayVector; +import org.apache.arrow.vector.DecimalVector; +import org.apache.arrow.vector.TimeMilliVector; +import org.apache.arrow.vector.TimeStampMicroTZVector; +import org.apache.arrow.vector.TimeStampMicroVector; +import org.apache.arrow.vector.VarBinaryVector; +import org.apache.arrow.vector.VectorSchemaRoot; +import org.apache.arrow.vector.types.DateUnit; +import org.apache.arrow.vector.types.TimeUnit; +import org.apache.arrow.vector.types.pojo.ArrowType; +import org.apache.arrow.vector.types.pojo.Field; +import org.apache.arrow.vector.types.pojo.Schema; +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.math.BigDecimal; +import java.nio.file.Path; +import java.sql.Connection; +import java.sql.ResultSet; +import java.sql.Statement; +import java.time.Instant; +import java.time.LocalDate; +import java.time.LocalDateTime; +import java.time.LocalTime; +import java.time.ZoneOffset; +import java.util.List; +import java.util.Map; + +import static org.assertj.core.api.Assertions.assertThat; + +/// SQL over a vortex-jni-written file through the Calcite adapter: what the planner is told about +/// the table must match what Rust wrote. +class CalciteInteropIntegrationTest { + + private static final Session SESSION = Session.create(); + private static final BufferAllocator ALLOCATOR = ArrowAllocation.rootAllocator(); + // several batches, so the row count is a sum over chunks, not a single chunk's length + private static final int BATCHES = 3; + private static final int BATCH_ROWS = 1_000; + private static final LocalDate DATE = LocalDate.of(2024, 2, 29); + private static final LocalTime TIME = LocalTime.of(13, 45, 30); + // pre-epoch on purpose: millisecond truncation must floor, not round toward zero + private static final Instant INSTANT = Instant.parse("1969-12-31T23:59:59.250Z"); + + static { + NativeLoader.loadJni(); + } + + @Test + void jniWriter_tableStatistic_isTheRowCountRustWrote(@TempDir Path tmp) throws Exception { + // Given + Path file = tmp.resolve("jni.vortex"); + writeIdsWithJni(file, BATCHES, BATCH_ROWS); + VortexTable sut = new VortexTable(file); + + // When + Double result = sut.getStatistic().getRowCount(); + + // Then — and SQL agrees with the statistic + assertThat(result).isEqualTo((double) BATCHES * BATCH_ROWS); + try (Connection conn = VortexCalcite.connect("vtx", Map.of("t", file)); + Statement st = conn.createStatement(); + ResultSet rs = st.executeQuery("select count(*) from vtx.t")) { + assertThat(rs.next()).isTrue(); + assertThat(rs.getLong(1)).isEqualTo((long) BATCHES * BATCH_ROWS); + } + } + + @Test + void jniWriter_filteredCountMinMax_foldFromRustZones(@TempDir Path tmp) throws Exception { + // Given — Rust writes fixed 8192-row vortex.zoned zones over chunks of its own sizing; the + // fold needed one zone per chunk, so every filtered aggregate on such a file scanned + Path file = tmp.resolve("jni_zoned.vortex"); + writeIdsWithJni(file, 8, 25_000); + long lo = 10_000; + long hi = 170_000; + String sql = "select count(*) c, min(id) mn, max(id) mx from vtx.t where id between " + lo + " and " + hi; + + try (Connection conn = VortexCalcite.connect("vtx", Map.of("t", file))) { + // When + String plan; + try (Statement st = conn.createStatement(); ResultSet rs = st.executeQuery("explain plan for " + sql)) { + rs.next(); + plan = rs.getString(1); + } + + // Then — answered from zones and boundary rows, and equal to the data + assertThat(plan).containsIgnoringCase("Values").doesNotContain("TableScan"); + try (Statement st = conn.createStatement(); ResultSet rs = st.executeQuery(sql)) { + rs.next(); + assertThat(rs.getLong("c")).isEqualTo(hi - lo + 1); + assertThat(rs.getLong("mn")).isEqualTo(lo); + assertThat(rs.getLong("mx")).isEqualTo(hi); + } + } + } + + @ParameterizedTest(name = "where {0}") + @CsvSource(delimiter = '|', value = { + // an IN list arrives as an OR of equalities on one column + "id in (3, 1500, 2999) | 3, 1500, 2999", + // a literal on the left + "2996 < id | 2997, 2998, 2999", + // n is null on every 1000th row + "n is null | 0, 1000, 2000", + "n is null and id in (0, 1, 1000) | 0, 1000"}) + void jniWriter_translatedFilters_returnExactlyTheMatchingRows(String where, String expected, @TempDir Path tmp) + throws Exception { + // Given + Path file = tmp.resolve("jni_filters.vortex"); + writeIdsWithJni(file, BATCHES, BATCH_ROWS); + + // When + List result = new java.util.ArrayList<>(); + try (Connection conn = VortexCalcite.connect("vtx", Map.of("t", file)); + Statement st = conn.createStatement(); + ResultSet rs = st.executeQuery("select id from vtx.t where " + where + " order by id")) { + while (rs.next()) { + result.add(rs.getLong(1)); + } + } + + // Then + assertThat(result).containsExactlyElementsOf( + java.util.Arrays.stream(expected.split(",")).map(String::trim).map(Long::valueOf).toList()); + } + + @Test + void jniWriter_temporalDecimalAndBinaryColumns_readThroughSql(@TempDir Path tmp) throws Exception { + // Given — a file with a timestamp could not be queried at all: the row type threw + Path file = tmp.resolve("jni_types.vortex"); + writeTypesWithJni(file); + String sql = "select `day`, at, wall, `instant`, amount, blob from vtx.t"; + + // When + try (Connection conn = VortexCalcite.connect("vtx", Map.of("t", file)); + Statement st = conn.createStatement(); + ResultSet result = st.executeQuery(sql)) { + + // Then + assertThat(result.next()).isTrue(); + assertThat(result.getDate("day").toLocalDate()).isEqualTo(DATE); + assertThat(result.getTime("at").toLocalTime()).isEqualTo(TIME); + assertThat(result.getTimestamp("wall").toLocalDateTime()) + .isEqualTo(LocalDateTime.ofInstant(INSTANT, ZoneOffset.UTC)); + assertThat(result.getTimestamp("instant").toLocalDateTime()) + .isEqualTo(LocalDateTime.ofInstant(INSTANT, ZoneOffset.UTC)); + assertThat(result.getBigDecimal("amount")).isEqualByComparingTo("12.34"); + assertThat(result.getBytes("blob")).containsExactly(1, 2, 3); + assertThat(result.next()).isFalse(); + } + } + + private static void writeTypesWithJni(Path file) throws Exception { + Schema schema = new Schema(List.of( + Field.notNullable("day", new ArrowType.Date(DateUnit.DAY)), + Field.notNullable("at", new ArrowType.Time(TimeUnit.MILLISECOND, 32)), + Field.notNullable("wall", new ArrowType.Timestamp(TimeUnit.MICROSECOND, null)), + Field.notNullable("instant", new ArrowType.Timestamp(TimeUnit.MICROSECOND, "UTC")), + Field.notNullable("amount", new ArrowType.Decimal(10, 2, 128)), + Field.notNullable("blob", new ArrowType.Binary()))); + long micros = INSTANT.getEpochSecond() * 1_000_000 + INSTANT.getNano() / 1_000; + try (VortexWriter writer = VortexWriter.builder(SESSION, file.toAbsolutePath().toUri().toString(), schema, + ALLOCATOR).build(); + VectorSchemaRoot root = VectorSchemaRoot.create(schema, ALLOCATOR)) { + root.allocateNew(); + ((DateDayVector) root.getVector("day")).setSafe(0, Math.toIntExact(DATE.toEpochDay())); + ((TimeMilliVector) root.getVector("at")).setSafe(0, (int) (TIME.toNanoOfDay() / 1_000_000)); + ((TimeStampMicroVector) root.getVector("wall")).setSafe(0, micros); + ((TimeStampMicroTZVector) root.getVector("instant")).setSafe(0, micros); + ((DecimalVector) root.getVector("amount")).setSafe(0, new BigDecimal("12.34")); + ((VarBinaryVector) root.getVector("blob")).setSafe(0, new byte[]{1, 2, 3}); + root.setRowCount(1); + try (ArrowArray array = ArrowArray.allocateNew(ALLOCATOR); + ArrowSchema arrowSchema = ArrowSchema.allocateNew(ALLOCATOR)) { + Data.exportVectorSchemaRoot(ALLOCATOR, root, null, array, arrowSchema); + writer.writeBatch(array.memoryAddress(), arrowSchema.memoryAddress()); + } + } + } + + 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)))); + 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"); + ids.allocateNew(batchRows); + n.allocateNew(batchRows); + for (int i = 0; i < batchRows; i++) { + long id = (long) batch * batchRows + i; + ids.set(i, id); + if (id % 1000 == 0) { + n.setNull(i); + } else { + n.set(i, id); + } + } + root.setRowCount(batchRows); + try (ArrowArray array = ArrowArray.allocateNew(ALLOCATOR); + ArrowSchema arrowSchema = ArrowSchema.allocateNew(ALLOCATOR)) { + Data.exportVectorSchemaRoot(ALLOCATOR, root, null, array, arrowSchema); + writer.writeBatch(array.memoryAddress(), arrowSchema.memoryAddress()); + } + } + } + } + } +} diff --git a/integration/src/test/java/io/github/dfa1/vortex/integration/DateInteropIntegrationTest.java b/integration/src/test/java/io/github/dfa1/vortex/integration/DateInteropIntegrationTest.java index 5d22af281..8e471ea24 100644 --- a/integration/src/test/java/io/github/dfa1/vortex/integration/DateInteropIntegrationTest.java +++ b/integration/src/test/java/io/github/dfa1/vortex/integration/DateInteropIntegrationTest.java @@ -110,6 +110,24 @@ void rustWrites_dtypeRecordsTheUnit(DateUnit unit, @TempDir Path tmp) throws IOE } } + @ParameterizedTest + @EnumSource(DateUnit.class) + void rustWrites_dtypeOutlivesTheReader(DateUnit unit, @TempDir Path tmp) throws IOException { + // Given + Path file = tmp.resolve("dates.vortex"); + writeWithJni(file, unit); + DType.Extension result; + + // When — the dtype is kept past close, as a SQL schema caches it + try (VortexReader reader = VortexReader.open(file, ReadRegistry.loadAll())) { + result = (DType.Extension) ((DType.Struct) reader.dtype()).fieldTypes().get(1); + } + + // Then — the metadata was a slice of the closed file mapping, so this threw "Already closed" + TimeUnit expectedUnit = unit == DateUnit.DAY ? TimeUnit.Days : TimeUnit.Milliseconds; + assertThat(TimeUnit.fromTag(result.metadata().get(ValueLayout.JAVA_BYTE, 0))).isEqualTo(expectedUnit); + } + /// Every ninth row is null; the rest step 37 days from 1915, so a quarter of them precede the /// epoch, where flooring a millisecond count must not round toward zero. private static LocalDate expected(long row) { diff --git a/performance/src/main/java/io/github/dfa1/vortex/performance/CalciteBoundaryAggregateBenchmark.java b/performance/src/main/java/io/github/dfa1/vortex/performance/CalciteBoundaryAggregateBenchmark.java index f578177dd..e917fb1fe 100644 --- a/performance/src/main/java/io/github/dfa1/vortex/performance/CalciteBoundaryAggregateBenchmark.java +++ b/performance/src/main/java/io/github/dfa1/vortex/performance/CalciteBoundaryAggregateBenchmark.java @@ -3,6 +3,7 @@ import io.github.dfa1.vortex.core.model.ColumnName; import io.github.dfa1.vortex.calcite.VortexTable; import io.github.dfa1.vortex.core.model.DType; +import io.github.dfa1.vortex.core.model.Editions; import io.github.dfa1.vortex.reader.Chunk; import io.github.dfa1.vortex.reader.RowFilter; import io.github.dfa1.vortex.reader.ScanOptions; @@ -23,6 +24,7 @@ import org.openjdk.jmh.annotations.State; import org.openjdk.jmh.annotations.TearDown; import org.openjdk.jmh.annotations.Warmup; +import org.openjdk.jmh.infra.BenchmarkParams; import java.io.IOException; import java.nio.channels.FileChannel; @@ -70,8 +72,8 @@ /// [#fullScanBaseline] decodes one chunk per surviving zone, so the fold wins in proportion to the /// interior-chunk count — large on `wide`, smaller on `medium`, parity on `narrow`. The mode is /// [Mode#AverageTime] in milliseconds because the metric of interest is per-query latency. Both -/// methods return the sum so JMH cannot dead-code-eliminate the work; [#setup] asserts the two paths -/// agree once before timing so the A/B is honest. +/// methods return the sum so JMH cannot dead-code-eliminate the work; the fold's trial asserts in +/// [#setup] that the two paths agree before timing, so the A/B is honest. @State(Scope.Benchmark) @BenchmarkMode(Mode.AverageTime) @OutputTimeUnit(TimeUnit.MILLISECONDS) @@ -103,7 +105,7 @@ public class CalciteBoundaryAggregateBenchmark { private long hi; @Setup(Level.Trial) - public void setup() throws IOException { + public void setup(BenchmarkParams params) throws IOException { long[] bounds = bounds(range); lo = bounds[0]; hi = bounds[1]; @@ -116,6 +118,17 @@ public void setup() throws IOException { filter = RowFilter.and(RowFilter.gte(ColumnName.of("id"), lo), RowFilter.lte(ColumnName.of("id"), hi)); baselineOptions = ScanOptions.columns(ColumnName.of("id"), ColumnName.of(AGG_COLUMN)).withFilter(filter); + // Only the fold's trial runs the checks: decoding through the fold in the baseline's JVM + // pollutes the reader's type profiles and slowed the scan 2-3x, timing the setup, not the scan. + if (!params.getBenchmark().endsWith(".boundaryFold")) { + return; + } + // The fold must not abandon where interior zones exist: an abandoned fold falls back to the + // scan, and the A/B would silently time the baseline twice. + if (!"narrow".equals(range) && table.filteredFold(filter, ColumnName.of(AGG_COLUMN)) + .map(VortexTable.FilteredFold::sum).isEmpty()) { + throw new IllegalStateException("boundary fold cannot answer SUM for range=" + range); + } // Honesty check: the optimized fold and the full-scan baseline must agree before we time them. long folded = boundaryFold(); long scanned = fullScanBaseline(); @@ -193,8 +206,11 @@ private static void writeFixture(Path file) throws IOException { .field(ColumnName.of("id"), DType.I64) .field(ColumnName.of("val"), DType.I64) .build(); - // enableZoneMaps=true emits the per-chunk min/max/sum/null-count the interior-zone fold reads. - WriteOptions opts = WriteOptions.cascading(0).withZoneMaps(true).withGlobalDict(true).withoutEditions(); + // core2025.10.0 predates Rust's vortex.zoned, so the writer emits the legacy vortex.stats zone + // map: one zone per batch with the min/max/sum/null-count the interior-zone fold reads. The + // default vortex.zoned has 8192-row zones and no sum, so a filtered SUM always scans. + WriteOptions opts = WriteOptions.cascading(0).withZoneMaps(true).withGlobalDict(true) + .withEdition(Editions.CORE_2025_10_0); java.util.Random rng = new java.util.Random(SEED); try (FileChannel ch = FileChannel.open(file, StandardOpenOption.CREATE, StandardOpenOption.WRITE, StandardOpenOption.TRUNCATE_EXISTING); diff --git a/reader/src/main/java/io/github/dfa1/vortex/reader/Chunk.java b/reader/src/main/java/io/github/dfa1/vortex/reader/Chunk.java index 0470446ba..253c0bd6c 100644 --- a/reader/src/main/java/io/github/dfa1/vortex/reader/Chunk.java +++ b/reader/src/main/java/io/github/dfa1/vortex/reader/Chunk.java @@ -8,6 +8,8 @@ import io.github.dfa1.vortex.reader.extension.ExtensionDecoder; import java.lang.foreign.Arena; +import java.util.Collections; +import java.util.LinkedHashMap; import java.util.List; import java.util.Objects; import java.util.SequencedMap; @@ -44,15 +46,39 @@ public final class Chunk implements AutoCloseable { private final Arena arena; private final ReadRegistry registry; private final Consumer onClose; + // a slice shares its parent's arena, which only the parent releases + private final boolean ownsArena; private boolean closed; Chunk(long rowCount, SequencedMap columns, Arena arena, ReadRegistry registry, Consumer onClose) { + this(rowCount, columns, arena, registry, onClose, true); + } + + private Chunk(long rowCount, SequencedMap columns, + Arena arena, ReadRegistry registry, Consumer onClose, boolean ownsArena) { this.rowCount = rowCount; this.columns = Objects.requireNonNull(columns); this.arena = Objects.requireNonNull(arena); this.registry = Objects.requireNonNull(registry); this.onClose = Objects.requireNonNull(onClose); + this.ownsArena = ownsArena; + } + + /// A view of rows `[offset, offset + length)` of every column, sharing this chunk's memory + /// with no copy. The view is valid while this chunk is open; closing the view does nothing. + /// + /// @param offset first row of the view + /// @param length number of rows in the view + /// @return the row-range view + /// @throws IndexOutOfBoundsException if the range is not within `[0, rowCount())` + /// @throws VortexException if a column's array type cannot be sliced + public Chunk slice(long offset, long length) { + Objects.checkFromIndexSize(offset, length, rowCount); + SequencedMap sliced = new LinkedHashMap<>(); + columns.forEach((name, column) -> sliced.put(name, new Column( + ScanIterator.sliceArray(column.array(), offset, length, column.dtype()), column.dtype()))); + return new Chunk(length, Collections.unmodifiableSequencedMap(sliced), arena, registry, _ -> { }, false); } /// One decoded column: its zero-copy [Array] view and the [DType] it was declared with in the @@ -155,6 +181,9 @@ public void close() { return; } closed = true; + if (!ownsArena) { + return; + } try { onClose.accept(this); } finally { diff --git a/reader/src/test/java/io/github/dfa1/vortex/reader/ChunkSliceTest.java b/reader/src/test/java/io/github/dfa1/vortex/reader/ChunkSliceTest.java new file mode 100644 index 000000000..eef5ccde0 --- /dev/null +++ b/reader/src/test/java/io/github/dfa1/vortex/reader/ChunkSliceTest.java @@ -0,0 +1,75 @@ +package io.github.dfa1.vortex.reader; + +import io.github.dfa1.vortex.core.io.VortexFormat; +import io.github.dfa1.vortex.core.model.ColumnName; +import io.github.dfa1.vortex.core.model.DType; +import io.github.dfa1.vortex.reader.array.LongArray; +import io.github.dfa1.vortex.reader.array.MaterializedLongArray; + +import org.junit.jupiter.api.Test; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.CsvSource; + +import java.lang.foreign.Arena; +import java.lang.foreign.MemorySegment; +import java.util.LinkedHashMap; +import java.util.SequencedMap; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; + +class ChunkSliceTest { + + private static final ColumnName ID = ColumnName.of("id"); + private static final Arena ARENA = Arena.ofAuto(); + + @Test + void slice_viewsTheRowRange() { + // Given + Chunk sut = chunk(10); + + // When + Chunk result = sut.slice(3, 4); + + // Then + assertThat(result.rowCount()).isEqualTo(4); + LongArray ids = result.column(ID); + assertThat(ids.length()).isEqualTo(4); + assertThat(ids.getLong(0)).isEqualTo(3); + assertThat(ids.getLong(3)).isEqualTo(6); + } + + @Test + void closingTheSlice_leavesTheParentOpen() { + // Given + Chunk sut = chunk(10); + Chunk slice = sut.slice(0, 5); + + // When + slice.close(); + + // Then — the shared arena belongs to the parent + assertThat(sut.isClosed()).isFalse(); + assertThat(((LongArray) sut.column(ID)).getLong(9)).isEqualTo(9); + } + + @ParameterizedTest + @CsvSource({"-1, 1", "0, 11", "10, 1", "5, -1"}) + void slice_outsideTheChunk_throws(long offset, long length) { + // Given + Chunk sut = chunk(10); + + // When / Then + assertThatThrownBy(() -> sut.slice(offset, length)).isInstanceOf(IndexOutOfBoundsException.class); + } + + private static Chunk chunk(int rows) { + MemorySegment segment = ARENA.allocate(rows * 8L, 8); + for (int i = 0; i < rows; i++) { + segment.setAtIndex(VortexFormat.LE_LONG, i, i); + } + SequencedMap columns = new LinkedHashMap<>(); + columns.put(ID, new Chunk.Column(new MaterializedLongArray(DType.I64, rows, segment), DType.I64)); + return new Chunk(rows, columns, ARENA, ReadRegistry.loadAll(), _ -> { }); + } +}