diff --git a/CHANGELOG.md b/CHANGELOG.md index b0364702..3d7d7d27 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -11,6 +11,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 - Per-zone stats from current Rust writers (`vortex.zoned`, vortex-jni 0.76.0) decode again. The 0.76 zoned layout replaced the legacy `vortex.stats` bit-set metadata with an aggregate-function spec list and dropped the per-stat truncation flags; the reader now reconstructs the stats table from that spec list (min/max/sum/null_count), so `columnZoneStats` and aggregate push-down work against those files instead of throwing `ClassCastException`. ([#197](https://github.com/dfa1/vortex-java/pull/197)) - Scans of files whose columns use different chunk grids no longer fail with `mixed per-column chunking beyond 1-vs-N is not supported`. The scan planner now splits at the merged boundary grid — the sorted union of every column's chunk boundaries, matching the Rust reference — and decodes each column's covering chunk once, slicing it zero-copy to each window. This handles both nested grids (Raincloud `emotions-dataset-for-nlp`, where `label`'s coarse chunks nest inside `text`'s finer grid) and disjoint grids (`uci-beijing-multi-site-air-quality`, where numeric and `station` boundaries do not nest). Aligned N-vs-N and 1-vs-N scans keep their existing slice-free fast path. ([#221](https://github.com/dfa1/vortex-java/issues/221)) +- Hardened the `vortex.zoned` metadata decoder against malformed input: an attacker-controlled length varint could overflow its `pos + len` bounds checks and crash with `NegativeArraySizeException`/`IndexOutOfBoundsException`; the checks are now overflow-safe and unparseable metadata falls back to per-chunk stats. Decimal columns — whose Rust zone table keeps a `sum` field this reader cannot map — now also fall back instead of decoding a misaligned table. ([#197](https://github.com/dfa1/vortex-java/pull/197)) - CSV export renders nested struct columns as JSON object cells (`{"field":value,...}`, strings JSON-escaped, null fields as JSON `null`, nested structs recursed) instead of throwing `unsupported array type: StructArray`. ([#217](https://github.com/dfa1/vortex-java/issues/217)) - Scanning a struct whose columns include a nested struct no longer fails chunk planning. A nested `vortex.struct` layout column (e.g. Raincloud `countries-of-the-world`'s `data` column) is now treated as a single full-range chunk source and decoded through the layout registry's new struct decoder into a `StructArray`, matching the Rust reference (a nested struct spans its parent's row range, not an independent chunking). `schema`/`count`/`inspect` already worked; plain scans and `select` of sibling columns now work too. CSV export of the struct column itself remains unsupported (a separate rendering limitation). ([#207](https://github.com/dfa1/vortex-java/issues/207)) - CSV export renders unsigned integer columns (U8–U64) with their unsigned values — high-half values previously printed as two's-complement negatives (uci-wine `magnesium` U8 132 exported as -124), silent corruption found by the Raincloud conformance suite. ([#208](https://github.com/dfa1/vortex-java/issues/208)) diff --git a/docs/explanation.md b/docs/explanation.md index 550ba08d..3869de36 100644 --- a/docs/explanation.md +++ b/docs/explanation.md @@ -222,7 +222,7 @@ file-level `SegmentSpec[]` table). Five node types exist today: | ID | Constant | Children | Role | |-----------------|-----------|-----------|------| | `vortex.struct` | `STRUCT` | N | Row type. One child per column. Root of every file. | -| `vortex.stats` | `ZONED` | 1 | Wraps a child layout and carries per-chunk min/max as zone maps. Pruned at scan time when filter predicate falls outside `[min, max]`. | +| `vortex.stats` | `ZONED` | 1 | Wraps a child layout and carries a per-zone stats table as a zone map. The legacy `vortex.stats` form declares its columns with a `Stat` bitset; the Rust >= 0.76 `vortex.zoned` form declares them with an aggregate-spec list, adding `sum`/`null_count` alongside `min`/`max`. Pruned at scan time when the filter predicate falls outside `[min, max]`. | | `vortex.chunked`| `CHUNKED` | M (+1) | Row-group sequence. Optional stats child at index 0 when `metadata[0] == 1` (per-chunk stats sidecar); remaining children are the data chunks. | | `vortex.dict` | `DICT` | 2 | Dictionary-encoded leaf. `children[0]` = values layout, `children[1]` = codes layout. `metadata` holds the codes `PType` (varint, proto field 1). Decoder gathers values by code. | | `vortex.flat` | `FLAT` | 0 | Leaf. References one `SegmentSpec` via `segments[0]`. Decoded by the encoding named in the segment's `arraySpec`, not by `encodingId` itself — see below. | diff --git a/reader/src/main/java/io/github/dfa1/vortex/reader/layout/ZonedStatsSchema.java b/reader/src/main/java/io/github/dfa1/vortex/reader/layout/ZonedStatsSchema.java index fd31c07d..267d7a0f 100644 --- a/reader/src/main/java/io/github/dfa1/vortex/reader/layout/ZonedStatsSchema.java +++ b/reader/src/main/java/io/github/dfa1/vortex/reader/layout/ZonedStatsSchema.java @@ -29,9 +29,12 @@ /// each get an extra trailing field `max_is_truncated` / /// `min_is_truncated` of type `Bool` (non-nullable). /// -/// `Sum` widening rules and Decimal handling are not yet implemented — -/// when the column dtype has no resolvable stat dtype the stat is skipped so -/// the inspector degrades to "no schema" rather than failing. +/// `Sum` widening for Decimal columns is not yet modeled. The legacy bitset path +/// ([#statsTableDtype(DType, List)]) skips any stat with no resolvable dtype so the inspector +/// degrades to "no schema" rather than failing. The aggregate-spec path +/// ([#aggregateStatsTableDtype(DType, MemorySegment)]) is positional, so it only skips fields Rust +/// also omits and otherwise returns `null` (fall back to per-chunk stats) rather than emit a +/// misaligned struct. public final class ZonedStatsSchema { /// Ordinal positions of [Stat] in the Rust enum — kept stable across @@ -171,9 +174,17 @@ public static DType.Struct aggregateStatsTableDtype(DType columnDtype, MemorySeg } DType stype = statDtype(stat, columnDtype); if (stype == null) { - // Rust's schema drops aggregates with no state dtype for this column (e.g. - // nan_count over a non-float); skip to stay aligned with the encoded table. - continue; + if (rustAlsoOmits(stat, columnDtype)) { + // Rust's aggregate_stats_table_dtype also drops this field (its aggregate's + // state_dtype is None for this column), so skipping stays aligned with the + // encoded table. + continue; + } + // Rust keeps a field here that we cannot map — notably Sum over a Decimal column, + // whose state widens to Decimal(precision + 10) (writer.rs default_zoned_aggregate_fns + // only emits Sum when Sum.return_dtype is Some, which it is for Decimal). Dropping it + // would misalign the positional decode, so bail to per-chunk stats. + return null; } names.add(ColumnName.of(stat.fieldName())); types.add(stype.withNullable(true)); @@ -181,6 +192,34 @@ public static DType.Struct aggregateStatsTableDtype(DType columnDtype, MemorySeg return new DType.Struct(List.copyOf(names), List.copyOf(types), false); } + /// Reports whether Rust's `aggregate_stats_table_dtype` also omits this aggregate for the given + /// column — i.e. the aggregate's `state_dtype` is `None` there, so dropping it keeps our + /// positional schema aligned with the encoded table. Any other null resolution means Rust keeps + /// a field we cannot describe, and the caller must bail instead. + /// + /// The two legitimate drops mirror the Rust reference + /// ([vortex-array/src/aggregate_fn/fns](https://github.com/spiraldb/vortex/tree/develop/vortex-array/src/aggregate_fn/fns)): + /// - `nan_count` returns `None` for non-float columns (`nan_count/mod.rs`); + /// - `min`/`max` return `None` when `minmax_supported_dtype` is false — for us that is only + /// `DType.Null`, the sole column for which the min/max resolver yields null. + private static boolean rustAlsoOmits(Stat stat, DType columnDtype) { + return switch (stat) { + case NAN_COUNT -> !isFloatingColumn(columnDtype); + case MAX, MIN -> true; + default -> false; + }; + } + + private static boolean isFloatingColumn(DType columnDtype) { + if (columnDtype instanceof DType.Primitive p) { + return p.ptype().isFloating(); + } + if (columnDtype instanceof DType.Extension ext) { + return isFloatingColumn(ext.storageDType()); + } + return false; + } + /// Reconstructs the per-zone stats-table dtype for the given column dtype /// and explicit stat list (already decoded from metadata). /// @@ -400,7 +439,10 @@ boolean skipField(int wireType) { /// or `null` if the message is malformed or carries no id. String readAggregateSpecId() { long len = readVarint(); - if (len < 0 || pos + len > end) { + // Overflow-safe bound: pos <= end holds throughout, so end - pos never overflows; a + // varint-derived len up to Long.MAX_VALUE would overflow pos + len to negative and slip + // past a pos + len > end guard, then advance pos past the segment. + if (len < 0 || len > end - pos) { return null; } long messageEnd = pos + len; @@ -426,7 +468,11 @@ String readAggregateSpecId() { private String readString(long limit) { long len = readVarint(); - if (len < 0 || pos + len > limit) { + // Overflow-safe bound (see readAggregateSpecId): compare against limit - pos, never + // pos + len. If a prior varint pushed pos past limit, limit - pos is negative and any + // non-negative len is rejected, so we never size new byte[(int) len] from an + // attacker-controlled, overflowed length. + if (len < 0 || len > limit - pos) { return null; } byte[] bytes = new byte[(int) len]; @@ -436,7 +482,10 @@ private String readString(long limit) { } private boolean advance(long count) { - if (count < 0 || pos + count > end) { + // Overflow-safe bound: pos <= end holds throughout, so end - pos never overflows, + // whereas pos + count could overflow to negative for a varint-derived count near + // Long.MAX_VALUE and slip past a pos + count > end guard. + if (count < 0 || count > end - pos) { return false; } pos += count; diff --git a/reader/src/test/java/io/github/dfa1/vortex/reader/layout/ZonedStatsSchemaTest.java b/reader/src/test/java/io/github/dfa1/vortex/reader/layout/ZonedStatsSchemaTest.java index 09f85c71..71780efa 100644 --- a/reader/src/test/java/io/github/dfa1/vortex/reader/layout/ZonedStatsSchemaTest.java +++ b/reader/src/test/java/io/github/dfa1/vortex/reader/layout/ZonedStatsSchemaTest.java @@ -256,17 +256,17 @@ void buildsNumericSchemaWithoutTruncationFlags() { "vortex.max", "vortex.min", "vortex.sum", "vortex.nan_count", "vortex.null_count"); // When - DType.Struct schema = ZonedStatsSchema.aggregateStatsTableDtype(DType.I64, meta); + DType.Struct result = ZonedStatsSchema.aggregateStatsTableDtype(DType.I64, meta); // Then — no `_is_truncated` fields (the new format has none) and nan_count is dropped - assertThat(schema.fieldNames().stream().map(ColumnName::value).toList()) + assertThat(result.fieldNames().stream().map(ColumnName::value).toList()) .containsExactly("max", "min", "sum", "null_count"); - assertThat(schema.fieldTypes()).containsExactly( + assertThat(result.fieldTypes()).containsExactly( new DType.Primitive(PType.I64, true), new DType.Primitive(PType.I64, true), new DType.Primitive(PType.I64, true), new DType.Primitive(PType.U64, true)); - assertThat(schema.nullable()).isFalse(); + assertThat(result.nullable()).isFalse(); } @Test @@ -275,12 +275,12 @@ void keepsNanCountForFloatColumn() { MemorySegment meta = aggregateMeta(1024, "vortex.max", "vortex.nan_count"); // When - DType.Struct schema = ZonedStatsSchema.aggregateStatsTableDtype(DType.F64, meta); + DType.Struct result = ZonedStatsSchema.aggregateStatsTableDtype(DType.F64, meta); // Then - assertThat(schema.fieldNames().stream().map(ColumnName::value).toList()) + assertThat(result.fieldNames().stream().map(ColumnName::value).toList()) .containsExactly("max", "nan_count"); - assertThat(schema.fieldTypes()).element(1).isEqualTo(new DType.Primitive(PType.U64, true)); + assertThat(result.fieldTypes()).element(1).isEqualTo(new DType.Primitive(PType.U64, true)); } @Test @@ -290,10 +290,27 @@ void bailsOnUnknownAggregate() { MemorySegment meta = aggregateMeta(4096, "vortex.bounded_max", "vortex.null_count"); // When - DType.Struct schema = ZonedStatsSchema.aggregateStatsTableDtype(DType.UTF8, meta); + DType.Struct result = ZonedStatsSchema.aggregateStatsTableDtype(DType.UTF8, meta); // Then - assertThat(schema).isNull(); + assertThat(result).isNull(); + } + + @Test + void bailsForDecimalSumColumn() { + // Given — Rust's default_zoned_aggregate_fns emits a `sum` column for a Decimal column + // (Sum.return_dtype widens to Decimal(precision + 10)), but this reader cannot map a + // decimal sum state. Reconstructing {max, min, null_count} while the encoded table is + // {max, min, sum, null_count} would read null_count out of the sum buffer, so the whole + // column must fall back to per-chunk (unpruned) stats rather than emit a mis-sized struct. + MemorySegment meta = aggregateMeta(8192, + "vortex.max", "vortex.min", "vortex.sum", "vortex.null_count"); + + // When + DType.Struct result = ZonedStatsSchema.aggregateStatsTableDtype(DType.decimal(20, 4), meta); + + // Then + assertThat(result).isNull(); } @Test @@ -303,19 +320,19 @@ void bailsOnUnsupportedVersion() { meta.set(java.lang.foreign.ValueLayout.JAVA_BYTE, 0, (byte) 2); // When - DType.Struct schema = ZonedStatsSchema.aggregateStatsTableDtype(DType.I64, meta); + DType.Struct result = ZonedStatsSchema.aggregateStatsTableDtype(DType.I64, meta); // Then - assertThat(schema).isNull(); + assertThat(result).isNull(); } @Test void bailsOnNullMetadata() { // Given / When — a zoned layout with no metadata cannot be reconstructed - DType.Struct schema = ZonedStatsSchema.aggregateStatsTableDtype(DType.I64, null); + DType.Struct result = ZonedStatsSchema.aggregateStatsTableDtype(DType.I64, null); // Then - assertThat(schema).isNull(); + assertThat(result).isNull(); } @Test @@ -324,11 +341,103 @@ void returnsEmptyStructWhenNoAggregatesPresent() { MemorySegment meta = aggregateMeta(8192); // When - DType.Struct schema = ZonedStatsSchema.aggregateStatsTableDtype(DType.I64, meta); + DType.Struct result = ZonedStatsSchema.aggregateStatsTableDtype(DType.I64, meta); // Then — an empty (non-null) struct; the caller falls back to per-chunk stats - assertThat(schema).isNotNull(); - assertThat(schema.fieldNames()).isEmpty(); + assertThat(result).isNotNull(); + assertThat(result.fieldNames()).isEmpty(); + } + } + + @Nested + class MalformedAggregateMetadata { + // These blobs are untrusted file metadata. Each malformed shape must degrade to the + // fallback (null) return — never a raw JDK exception (NegativeArraySizeException / + // IndexOutOfBoundsException), which the pre-fix overflow-unsafe bounds checks allowed. + + @Test + void bailsOnTruncatedVarint() { + // Given — the only field tag is an unterminated varint (continuation bit set, no next + // byte); the cursor must fail rather than read past the buffer. + java.io.ByteArrayOutputStream out = new java.io.ByteArrayOutputStream(); + out.write(1); // envelope version + out.write(0x80); // dangling varint continuation byte + + // When + DType.Struct result = ZonedStatsSchema.aggregateStatsTableDtype( + DType.I64, MemorySegment.ofArray(out.toByteArray())); + + // Then + assertThat(result).isNull(); + } + + @Test + void bailsOnOversizedStringLength() { + // Given — a framed AggregateSpecProto whose id-string length varint is Long.MAX_VALUE. + // The pre-fix `pos + len > limit` guard overflowed to negative and passed, then + // `new byte[(int) len]` threw NegativeArraySizeException. + java.io.ByteArrayOutputStream spec = new java.io.ByteArrayOutputStream(); + writeTag(spec, 1, 2); // id, wire LEN + writeVarintLong(spec, Long.MAX_VALUE); // absurd string length + byte[] specBytes = spec.toByteArray(); + java.io.ByteArrayOutputStream out = new java.io.ByteArrayOutputStream(); + out.write(1); // envelope version + writeTag(out, 2, 2); // aggregate_specs, wire LEN + writeVarint(out, specBytes.length); + out.writeBytes(specBytes); + + // When / Then — must not throw NegativeArraySizeException + DType.Struct result = ZonedStatsSchema.aggregateStatsTableDtype( + DType.I64, MemorySegment.ofArray(out.toByteArray())); + assertThat(result).isNull(); + } + + @Test + void bailsOnOversizedSkippedFieldLength() { + // Given — a non-target length-delimited field whose length is Long.MAX_VALUE. The + // pre-fix advance guard `pos + count > end` overflowed and advanced pos negative, + // throwing IndexOutOfBoundsException on the next read. + java.io.ByteArrayOutputStream out = new java.io.ByteArrayOutputStream(); + out.write(1); // envelope version + writeTag(out, 1, 2); // field 1 with wire LEN → skipped, not an aggregate spec + writeVarintLong(out, Long.MAX_VALUE); + + // When / Then — must not throw IndexOutOfBoundsException + DType.Struct result = ZonedStatsSchema.aggregateStatsTableDtype( + DType.I64, MemorySegment.ofArray(out.toByteArray())); + assertThat(result).isNull(); + } + + @Test + void bailsOnUnknownWireType() { + // Given — a field using wire type 3 (group start), which the cursor cannot skip. + java.io.ByteArrayOutputStream out = new java.io.ByteArrayOutputStream(); + out.write(1); // envelope version + writeTag(out, 3, 3); // field 3, wire type 3 (unsupported) + + // When + DType.Struct result = ZonedStatsSchema.aggregateStatsTableDtype( + DType.I64, MemorySegment.ofArray(out.toByteArray())); + + // Then + assertThat(result).isNull(); + } + + @Test + void bailsOnMessageLengthPastEnd() { + // Given — an aggregate_specs field whose declared length runs past the buffer end. + java.io.ByteArrayOutputStream out = new java.io.ByteArrayOutputStream(); + out.write(1); // envelope version + writeTag(out, 2, 2); // aggregate_specs, wire LEN + writeVarint(out, 64); // claims 64 bytes... + out.write(0x01); // ...but only one follows + + // When + DType.Struct result = ZonedStatsSchema.aggregateStatsTableDtype( + DType.I64, MemorySegment.ofArray(out.toByteArray())); + + // Then + assertThat(result).isNull(); } } @@ -372,4 +481,15 @@ private static void writeVarint(java.io.ByteArrayOutputStream out, int value) { } out.write(v); } + + /// Writes an unsigned base-128 varint of a `long` — used to inject absurd (overflow-inducing) + /// lengths like `Long.MAX_VALUE` that a plain `int` varint cannot express. + private static void writeVarintLong(java.io.ByteArrayOutputStream out, long value) { + long v = value; + while ((v & ~0x7fL) != 0) { + out.write((int) ((v & 0x7f) | 0x80)); + v >>>= 7; + } + out.write((int) v); + } }