Calcite adapter: close push-down and type-mapping gaps

CHANGELOG.md

@@ -8,18 +8,25 @@ 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))
 - **Breaking:** `ExtensionDecoder` is generic in its domain type and declares `domainType()` and `decodeAll(DType.Extension, Array)`. ([#530](https://github.com/dfa1/vortex-java/issues/530))
 - **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))

calcite/pom.xml

@@ -10,7 +10,7 @@
     <artifactId>vortex-calcite</artifactId>
 
     <name>vortex-calcite</name>
-    <description>Apache Calcite SQL adapter over the Vortex columnar file format (demo: filter/project/aggregate push-down).</description>
+    <description>Apache Calcite SQL adapter over the Vortex columnar file format with projection, filter and aggregate push-down.</description>
 
     <properties>
         <!-- Calcite pulls in protobuf-java, whose UnsafeUtil calls sun.misc.Unsafe (not ours: we never use Unsafe). -->

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<RowFilter> toComparison(RexNode node, List<ColumnName> 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<RowFilter> strictComparison(RexNode node, List<ColumnNam
     }
 
     /// Shared comparison-kind dispatch behind [#toComparison] (lenient) and [#strictComparison]
-    /// (strict). The two differ only in how an untranslatable node is handled and whether the null
-    /// tests are accepted, both keyed off `strict`:
-    /// - in an `AND`, a `strict` walk abandons the whole predicate on the first untranslatable
-    ///   conjunct, while a lenient walk drops it and keeps the rest;
-    /// - bare `IS NULL` / `IS NOT NULL` translate only in the `strict` walk (the lenient scan path has
-    ///   no zone-map use for them and drops them).
-    /// The column-vs-literal comparison kinds (`=`, `<>`, `<`, `<=`, `>`, `>=`) 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<RowFilter> comparison(RexNode node, List<ColumnName> 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<RowFilter.Column> 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<RowFilter.Column> 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<RowFilter.Column> 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<RowFilter> nullCheck(RexCall call, List<ColumnName> name
 
     private static Optional<RowFilter> binary(RexCall call, List<ColumnName> names, List<DType> types) {
         List<RexNode> 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<RowFilter> binary(RexCall call, List<ColumnName> 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);
         });
     }
 

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