Skip to content

Commit 9e20edc

Browse files
dfa1claude
andcommitted
fix(reader): decode vortex.zoned aggregate-spec zone maps (vortex-jni 0.76.0)
Bumping vortex-jni to 0.76.0 broke RustWritesJavaReadsIntegrationTest #jniWriter_perZoneSum_readFromZoneMapTable with a ClassCastException (DType$Bool -> DType$Primitive) while decoding the per-zone stats table. 0.76.0's zoned layout (spiraldb/vortex #7938) splits the format in two: the canonical `vortex.zoned` id now stores an ordered aggregate-function spec list in its metadata (proto: version byte + ZonedMetadataProto) and its stats table drops the legacy per-stat `_is_truncated` Bool flags, while the legacy `vortex.stats` alias keeps the Stat-bitset metadata. Our reader treated the two ids as pure aliases and always reconstructed the legacy schema, so the extra Bool fields no longer matched the encoded table and the positional struct decode mismatched. The reader now dispatches on the layout id: `vortex.zoned` goes through a new ZonedStatsSchema.aggregateStatsTableDtype that parses the aggregate specs and maps each known aggregate (max/min/sum/null_count/nan_count/...) to its stored stat dtype, with no truncation flags; `vortex.stats` keeps the bitset path. Unknown/nested-state aggregates (e.g. vortex.bounded_max) bail to the per-chunk fallback rather than decode a misaligned table. The CLI stats inspector uses the same dispatch. This is a genuine compatibility fix on top of the dependency bump, not a rebase-conflict resolution (the rebase onto main was clean). Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
1 parent ec4e9de commit 9e20edc

5 files changed

Lines changed: 355 additions & 7 deletions

File tree

CHANGELOG.md

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -9,6 +9,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
99

1010
### Fixed
1111

12+
- 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))
1213
- 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))
1314
- 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))
1415
- 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))

cli/src/main/java/io/github/dfa1/vortex/cli/tui/VortexInspectorTui.java

Lines changed: 7 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -4,6 +4,7 @@
44
import io.github.dfa1.vortex.reader.SegmentSpec;
55
import io.github.dfa1.vortex.core.model.DType;
66
import io.github.dfa1.vortex.core.model.ColumnName;
7+
import io.github.dfa1.vortex.core.model.LayoutId;
78
import io.github.dfa1.vortex.reader.array.Array;
89
import io.github.dfa1.vortex.cli.tui.term.Ansi;
910
import io.github.dfa1.vortex.cli.tui.term.Key;
@@ -673,8 +674,12 @@ private void runStatsLoad(InspectorTree.Node anchor) {
673674
statsCache.put(anchor, new DataState.Failed("no column dtype"));
674675
return;
675676
}
676-
DType.Struct statsDtype = ZonedStatsSchema.statsTableDtype(columnDtype, anchorLayout.metadata());
677-
if (statsDtype.fieldNames().isEmpty()) {
677+
// `vortex.zoned` (Rust >= 0.76) carries aggregate-spec metadata; the legacy
678+
// `vortex.stats` alias carries a Stat bitset — reconstruct the schema accordingly.
679+
DType.Struct statsDtype = anchorLayout.layoutId() == LayoutId.ZONED
680+
? ZonedStatsSchema.aggregateStatsTableDtype(columnDtype, anchorLayout.metadata())
681+
: ZonedStatsSchema.statsTableDtype(columnDtype, anchorLayout.metadata());
682+
if (statsDtype == null || statsDtype.fieldNames().isEmpty()) {
678683
statsCache.put(anchor, new DataState.Failed("no stats present in metadata"));
679684
return;
680685
}

reader/src/main/java/io/github/dfa1/vortex/reader/ScanIterator.java

Lines changed: 15 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -3,6 +3,7 @@
33
import static io.github.dfa1.vortex.core.io.VortexFormat.LE_INT;
44
import io.github.dfa1.vortex.core.model.ColumnName;
55
import io.github.dfa1.vortex.core.model.DType;
6+
import io.github.dfa1.vortex.core.model.LayoutId;
67
import io.github.dfa1.vortex.core.io.IoBounds;
78
import io.github.dfa1.vortex.core.error.VortexException;
89
import io.github.dfa1.vortex.reader.array.Array;
@@ -395,10 +396,11 @@ public List<ArrayStats> columnZoneStats(String column) {
395396
return out;
396397
}
397398

398-
/// Decodes the column's `vortex.stats` zone-map table into one [ArrayStats] per zone, or
399-
/// returns `null` when the column has no zone map (so the caller falls back to per-chunk
400-
/// node stats). The table is a single flat segment encoding a struct with a subset of the
401-
/// `min`/`max`/`sum`/`null_count` fields (see [ZonedStatsSchema]); it is decoded into a
399+
/// Decodes the column's zone-map table into one [ArrayStats] per zone, or returns `null` when
400+
/// the column has no zone map (so the caller falls back to per-chunk node stats). The table is
401+
/// a single flat segment encoding a struct with a subset of the `min`/`max`/`sum`/`null_count`
402+
/// fields; the schema is reconstructed from the layout metadata (legacy `vortex.stats` bitset
403+
/// or newer `vortex.zoned` aggregate specs — see [ZonedStatsSchema]). It is decoded into a
402404
/// short-lived confined arena and the scalar values are boxed out before the arena closes.
403405
private List<ArrayStats> decodeZoneTable(ColumnName column) {
404406
Layout zoned = findZonedLayout(file.layout(), column);
@@ -417,7 +419,15 @@ private List<ArrayStats> decodeZoneTable(ColumnName column) {
417419
if (segIdx < 0 || segIdx >= file.footer().segmentSpecs().size()) {
418420
return null;
419421
}
420-
DType.Struct statsDtype = ZonedStatsSchema.statsTableDtype(columnDtype, zoned.metadata());
422+
// The canonical `vortex.zoned` id (Rust >= 0.76) stores an aggregate-spec metadata blob;
423+
// the legacy `vortex.stats` alias (what vortex-java writes) stores a Stat bitset. They
424+
// reconstruct the table schema differently — dispatch on the layout id.
425+
DType.Struct statsDtype = zoned.layoutId() == LayoutId.ZONED
426+
? ZonedStatsSchema.aggregateStatsTableDtype(columnDtype, zoned.metadata())
427+
: ZonedStatsSchema.statsTableDtype(columnDtype, zoned.metadata());
428+
if (statsDtype == null || statsDtype.fieldNames().isEmpty()) {
429+
return null;
430+
}
421431
long nZones = statsFlat.rowCount();
422432
SegmentSpec spec = file.footer().segmentSpecs().get(segIdx);
423433
try (Arena tableArena = Arena.ofConfined()) {

reader/src/main/java/io/github/dfa1/vortex/reader/layout/ZonedStatsSchema.java

Lines changed: 212 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -135,6 +135,52 @@ public static DType.Struct statsTableDtype(DType columnDtype, MemorySegment meta
135135
return statsTableDtype(columnDtype, presentStats(metadata));
136136
}
137137

138+
/// Reconstructs the per-zone stats-table dtype for a newer `vortex.zoned` layout, whose
139+
/// metadata carries an ordered list of aggregate-function specs instead of the legacy
140+
/// [Stat] bitset.
141+
///
142+
/// The shape mirrors Rust's `aggregate_stats_table_dtype`
143+
/// ([vortex-layout/src/layouts/zoned/schema.rs](https://github.com/spiraldb/vortex/blob/develop/vortex-layout/src/layouts/zoned/schema.rs)):
144+
/// for every aggregate whose state dtype is defined for the column, append one
145+
/// `(name, nullable state-dtype)` field, in spec order — no `_is_truncated` flags. Field
146+
/// names are the aggregate's canonical [Stat#fieldName()] so the reader can look them up by
147+
/// stat; the struct decode itself is positional, so alignment with the encoded table is what
148+
/// matters.
149+
///
150+
/// Returns `null` when the metadata is not a decodable aggregate-spec blob or names an
151+
/// aggregate this reader cannot map to a [Stat] (e.g. `vortex.bounded_max`, whose state is a
152+
/// nested struct). Bailing keeps the positional schema faithful: the caller falls back to
153+
/// per-chunk stats rather than decoding a misaligned table.
154+
///
155+
/// @param columnDtype the column's logical dtype (the `data` child's dtype)
156+
/// @param metadata raw `vortex.zoned` layout metadata
157+
/// @return reconstructed non-nullable struct dtype, or `null` when it cannot be reconstructed
158+
public static DType.Struct aggregateStatsTableDtype(DType columnDtype, MemorySegment metadata) {
159+
List<String> aggregateIds = aggregateIds(metadata);
160+
if (aggregateIds == null) {
161+
return null;
162+
}
163+
List<ColumnName> names = new ArrayList<>(aggregateIds.size());
164+
List<DType> types = new ArrayList<>(aggregateIds.size());
165+
for (String aggregateId : aggregateIds) {
166+
Stat stat = statForAggregate(aggregateId);
167+
if (stat == null) {
168+
// Unknown aggregate — the encoded table has a field here we cannot describe, so
169+
// any reconstruction would be positionally misaligned. Bail to the fallback path.
170+
return null;
171+
}
172+
DType stype = statDtype(stat, columnDtype);
173+
if (stype == null) {
174+
// Rust's schema drops aggregates with no state dtype for this column (e.g.
175+
// nan_count over a non-float); skip to stay aligned with the encoded table.
176+
continue;
177+
}
178+
names.add(ColumnName.of(stat.fieldName()));
179+
types.add(stype.withNullable(true));
180+
}
181+
return new DType.Struct(List.copyOf(names), List.copyOf(types), false);
182+
}
183+
138184
/// Reconstructs the per-zone stats-table dtype for the given column dtype
139185
/// and explicit stat list (already decoded from metadata).
140186
///
@@ -231,4 +277,170 @@ private static DType sumDtype(DType columnDtype) {
231277
// so the stat is dropped from the schema rather than causing a decode mismatch.
232278
return null;
233279
}
280+
281+
/// First byte of `vortex.zoned` metadata: the protobuf envelope version Rust writes.
282+
private static final int AGGREGATE_METADATA_VERSION = 1;
283+
284+
/// Maps a well-known aggregate-function id (Rust `AggregateFnId`) to the [Stat] whose stored
285+
/// dtype and canonical field name it uses in the zone-map table, or `null` when the reader
286+
/// has no faithful mapping (nested-state aggregates like `vortex.bounded_max`, or functions
287+
/// not stored as a scalar stat).
288+
private static Stat statForAggregate(String aggregateId) {
289+
return switch (aggregateId) {
290+
case "vortex.max" -> Stat.MAX;
291+
case "vortex.min" -> Stat.MIN;
292+
case "vortex.sum" -> Stat.SUM;
293+
case "vortex.null_count" -> Stat.NULL_COUNT;
294+
case "vortex.nan_count" -> Stat.NAN_COUNT;
295+
case "vortex.is_constant" -> Stat.IS_CONSTANT;
296+
case "vortex.is_sorted" -> Stat.IS_SORTED;
297+
case "vortex.is_strict_sorted" -> Stat.IS_STRICT_SORTED;
298+
case "vortex.uncompressed_size_in_bytes" -> Stat.UNCOMPRESSED_SIZE_IN_BYTES;
299+
default -> null;
300+
};
301+
}
302+
303+
/// Extracts the ordered aggregate-function ids from `vortex.zoned` metadata.
304+
///
305+
/// The metadata is a version byte (value [#AGGREGATE_METADATA_VERSION]) followed by a
306+
/// protobuf `ZonedMetadataProto { uint32 zone_len = 1; repeated AggregateSpecProto
307+
/// aggregate_specs = 2; }`, where `AggregateSpecProto { string id = 1; bytes options = 2; }`.
308+
/// Only the `id` of each spec is needed to reconstruct the table schema.
309+
///
310+
/// Returns `null` when the blob is empty, carries an unsupported version, or is not
311+
/// well-formed protobuf — signalling the caller to fall back rather than trust a partial parse.
312+
private static List<String> aggregateIds(MemorySegment metadata) {
313+
if (metadata == null || metadata.byteSize() < 1) {
314+
return null;
315+
}
316+
if ((metadata.get(ValueLayout.JAVA_BYTE, 0) & 0xff) != AGGREGATE_METADATA_VERSION) {
317+
return null;
318+
}
319+
ProtoCursor cursor = new ProtoCursor(metadata, 1, metadata.byteSize());
320+
List<String> ids = new ArrayList<>();
321+
while (cursor.hasRemaining()) {
322+
long tag = cursor.readVarint();
323+
if (tag < 0) {
324+
return null;
325+
}
326+
int fieldNumber = (int) (tag >>> 3);
327+
int wireType = (int) (tag & 0x7);
328+
if (fieldNumber == 2 && wireType == ProtoCursor.WIRE_LEN) {
329+
String id = cursor.readAggregateSpecId();
330+
if (id == null) {
331+
return null;
332+
}
333+
ids.add(id);
334+
} else if (!cursor.skipField(wireType)) {
335+
return null;
336+
}
337+
}
338+
return List.copyOf(ids);
339+
}
340+
341+
/// Minimal forward cursor over a protobuf byte range within a [MemorySegment]. Reads only the
342+
/// varints and length-delimited fields the zoned metadata uses; any malformed or unexpected
343+
/// wire shape is reported as a failure so the caller can bail rather than misread.
344+
private static final class ProtoCursor {
345+
static final int WIRE_VARINT = 0;
346+
static final int WIRE_I64 = 1;
347+
static final int WIRE_LEN = 2;
348+
static final int WIRE_I32 = 5;
349+
350+
private final MemorySegment segment;
351+
private long pos;
352+
private final long end;
353+
354+
ProtoCursor(MemorySegment segment, long start, long end) {
355+
this.segment = segment;
356+
this.pos = start;
357+
this.end = end;
358+
}
359+
360+
boolean hasRemaining() {
361+
return pos < end;
362+
}
363+
364+
/// Reads a base-128 varint, or `-1` when it runs past the end or exceeds 64 bits.
365+
long readVarint() {
366+
long result = 0;
367+
int shift = 0;
368+
while (pos < end) {
369+
int b = segment.get(ValueLayout.JAVA_BYTE, pos) & 0xff;
370+
pos++;
371+
if (shift < 64) {
372+
result |= (long) (b & 0x7f) << shift;
373+
}
374+
if ((b & 0x80) == 0) {
375+
return result;
376+
}
377+
shift += 7;
378+
if (shift > 63) {
379+
return -1;
380+
}
381+
}
382+
return -1;
383+
}
384+
385+
/// Skips a field of the given wire type, returning `false` on an unknown type or overrun.
386+
boolean skipField(int wireType) {
387+
return switch (wireType) {
388+
case WIRE_VARINT -> readVarint() >= 0;
389+
case WIRE_I64 -> advance(8);
390+
case WIRE_I32 -> advance(4);
391+
case WIRE_LEN -> {
392+
long len = readVarint();
393+
yield len >= 0 && advance(len);
394+
}
395+
default -> false;
396+
};
397+
}
398+
399+
/// Reads an `AggregateSpecProto` length-delimited message and returns its `id` (field 1),
400+
/// or `null` if the message is malformed or carries no id.
401+
String readAggregateSpecId() {
402+
long len = readVarint();
403+
if (len < 0 || pos + len > end) {
404+
return null;
405+
}
406+
long messageEnd = pos + len;
407+
String id = null;
408+
while (pos < messageEnd) {
409+
long tag = readVarint();
410+
if (tag < 0) {
411+
return null;
412+
}
413+
int fieldNumber = (int) (tag >>> 3);
414+
int wireType = (int) (tag & 0x7);
415+
if (fieldNumber == 1 && wireType == WIRE_LEN) {
416+
id = readString(messageEnd);
417+
if (id == null) {
418+
return null;
419+
}
420+
} else if (!skipField(wireType)) {
421+
return null;
422+
}
423+
}
424+
return id;
425+
}
426+
427+
private String readString(long limit) {
428+
long len = readVarint();
429+
if (len < 0 || pos + len > limit) {
430+
return null;
431+
}
432+
byte[] bytes = new byte[(int) len];
433+
MemorySegment.copy(segment, ValueLayout.JAVA_BYTE, pos, bytes, 0, (int) len);
434+
pos += len;
435+
return new String(bytes, java.nio.charset.StandardCharsets.UTF_8);
436+
}
437+
438+
private boolean advance(long count) {
439+
if (count < 0 || pos + count > end) {
440+
return false;
441+
}
442+
pos += count;
443+
return true;
444+
}
445+
}
234446
}

0 commit comments

Comments
 (0)