Skip to content

Commit 4566dca

Browse files
dfa1claude
dfa1
and
claude
committed
feat(variant): shredding — optional typed shredded child (Layer C)
VariantData can now carry a row-aligned shredded typed column (shreddedData + shreddedDtype). When present, the encoder emits it as the container's second child (encoded via the matching primitive/varbin/bool encoder) and records its dtype in VariantMetadata.shredded_dtype; otherwise the container stays single-child as before. The existing VariantEncodingDecoder already reads the shredded child, so this round-trips Java-side (decode exposes VariantArray.shredded()) and through the Rust reference reader (JNI). NOTE: with the current scalar-only variant model there are no object paths, so a shredded child can only re-express the values as a typed column. Real path-based shredding of nested objects needs vortex.parquet.variant (deferred, ADR 0014). Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
1 parent 76e4c74 commit 4566dca

4 files changed

Lines changed: 169 additions & 12 deletions

File tree

‎integration/src/test/java/io/github/dfa1/vortex/integration/VariantJavaWritesRustReadsIntegrationTest.java‎

Lines changed: 22 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -5,6 +5,7 @@
55
import dev.vortex.arrow.ArrowAllocation;
66
import dev.vortex.jni.NativeLoader;
77
import io.github.dfa1.vortex.core.DType;
8+
import io.github.dfa1.vortex.core.PType;
89
import io.github.dfa1.vortex.proto.Primitive;
910
import io.github.dfa1.vortex.proto.Scalar;
1011
import io.github.dfa1.vortex.proto.ScalarValue;
@@ -89,6 +90,27 @@ void javaWriter_jniReader_varyingVariantColumn(@TempDir Path tmp) throws IOExcep
8990
assertThat(ds.arrowSchema(ALLOCATOR).getFields()).extracting(f -> f.getName()).contains("v");
9091
}
9192

93+
@Test
94+
void javaWriter_jniReader_shreddedVariantColumn(@TempDir Path tmp) throws IOException {
95+
// Given — a variant column with a row-aligned shredded i32 projection. The container
96+
// gains a second (shredded) typed child plus shredded_dtype in its metadata.
97+
Path file = tmp.resolve("java_variant_shredded.vtx");
98+
List<Scalar> values = List.of(i32Variant(10L), i32Variant(20L), i32Variant(30L));
99+
VariantData data = VariantData.shredded(
100+
values, new int[]{10, 20, 30}, new DType.Primitive(PType.I32, false));
101+
102+
try (var ch = FileChannel.open(file, StandardOpenOption.CREATE, StandardOpenOption.WRITE);
103+
var sut = VortexWriter.create(ch, VARIANT_SCHEMA, WriteOptions.defaults())) {
104+
// When
105+
sut.writeChunk(Map.of("v", data));
106+
}
107+
108+
// Then — the Rust reader parses the shredded variant layout and agrees on row count + schema.
109+
DataSource ds = DataSource.open(SESSION, file.toAbsolutePath().toUri().toString());
110+
assertThat(ds.rowCount().asOptional()).hasValue(values.size());
111+
assertThat(ds.arrowSchema(ALLOCATOR).getFields()).extracting(f -> f.getName()).contains("v");
112+
}
113+
92114
private static Scalar i32Variant(long value) {
93115
// Inner typed scalar carrying its own i32 dtype, wrapped as a variant value
94116
// (mirrors Rust Scalar::variant(Scalar::primitive(value))).

‎writer/src/main/java/io/github/dfa1/vortex/writer/encode/VariantData.java‎

Lines changed: 41 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -1,28 +1,47 @@
11
package io.github.dfa1.vortex.writer.encode;
22

3+
import io.github.dfa1.vortex.core.DType;
34
import io.github.dfa1.vortex.proto.Scalar;
45

56
import java.util.Collections;
67
import java.util.List;
78

89
/// Input data for encoding a `vortex.variant` column.
910
///
10-
/// Holds one inner typed scalar per row, each wrapped as a variant value (mirroring
11-
/// Rust `Scalar::variant(inner)`). The encoder coalesces adjacent equal values into
12-
/// constant runs: an all-equal column becomes a single `vortex.constant` child, while
13-
/// a column with varying values becomes a `vortex.chunked` of per-run constants. There
14-
/// is no shredded child.
11+
/// `values` holds one inner typed scalar per row, each wrapped as a variant value
12+
/// (mirroring Rust `Scalar::variant(inner)`). The encoder coalesces adjacent equal
13+
/// values into constant runs: an all-equal column becomes a single `vortex.constant`
14+
/// child, while a varying column becomes a `vortex.chunked` of per-run constants.
1515
///
16-
/// @param values one inner scalar per row, in row order
17-
public record VariantData(List<Scalar> values) {
16+
/// Optionally, `shreddedData` carries a row-aligned typed column pulled out for fast
17+
/// typed access; when present it is encoded as the container's `shredded` child and its
18+
/// `shreddedDtype` is recorded in `VariantMetadata.shredded_dtype`. `shreddedData` uses
19+
/// the same encoder-input shape as the matching encoding (e.g. `int[]` for I32,
20+
/// `String[]` for Utf8). Both `shreddedData` and `shreddedDtype` must be set together or
21+
/// both left `null`.
22+
///
23+
/// @param values one inner scalar per row, in row order
24+
/// @param shreddedData optional typed column data for the shredded child, or `null`
25+
/// @param shreddedDtype dtype of `shreddedData`, or `null`
26+
public record VariantData(List<Scalar> values, Object shreddedData, DType shreddedDtype) {
1827

19-
/// Validates and defensively copies the per-row values. Rejects empty input and
20-
/// `null` elements.
28+
/// Validates and defensively copies the per-row values; rejects empty input and a
29+
/// half-specified shredded column.
2130
public VariantData {
2231
values = List.copyOf(values);
2332
if (values.isEmpty()) {
2433
throw new IllegalArgumentException("values must not be empty");
2534
}
35+
if ((shreddedData == null) != (shreddedDtype == null)) {
36+
throw new IllegalArgumentException("shreddedData and shreddedDtype must be both set or both null");
37+
}
38+
}
39+
40+
/// Creates unshredded input from per-row scalars.
41+
///
42+
/// @param values one inner scalar per row, in row order
43+
public VariantData(List<Scalar> values) {
44+
this(values, null, null);
2645
}
2746

2847
/// Creates input for a constant variant column: `length` rows all holding `value`.
@@ -40,6 +59,19 @@ public static VariantData constant(int length, Scalar value) {
4059
return new VariantData(Collections.nCopies(length, value));
4160
}
4261

62+
/// Creates input with a row-aligned shredded typed column.
63+
///
64+
/// @param values one inner scalar per row, in row order
65+
/// @param shreddedData typed column data (encoder-input shape for `shreddedDtype`)
66+
/// @param shreddedDtype dtype of `shreddedData`
67+
/// @return variant input carrying a shredded child
68+
public static VariantData shredded(List<Scalar> values, Object shreddedData, DType shreddedDtype) {
69+
if (shreddedData == null || shreddedDtype == null) {
70+
throw new IllegalArgumentException("shreddedData and shreddedDtype are required");
71+
}
72+
return new VariantData(values, shreddedData, shreddedDtype);
73+
}
74+
4375
/// Returns the number of rows in the column.
4476
///
4577
/// @return row count

‎writer/src/main/java/io/github/dfa1/vortex/writer/encode/VariantEncodingEncoder.java‎

Lines changed: 53 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -65,12 +65,62 @@ public EncodeResult encode(DType dtype, Object data, EncodeContext ctx) {
6565
? constantChild(runValues.get(0), buffers)
6666
: chunkedConstants(runValues, runLengths, ctx, buffers);
6767

68-
ByteBuffer containerMeta = ByteBuffer.wrap(new VariantMetadata(null).encode());
69-
EncodeNode root = new EncodeNode(EncodingId.VORTEX_VARIANT, containerMeta,
70-
new EncodeNode[]{coreStorage}, new int[0]);
68+
EncodeNode[] children;
69+
io.github.dfa1.vortex.proto.DType shreddedProto = null;
70+
if (variantData.shreddedData() != null) {
71+
children = new EncodeNode[]{coreStorage, encodeShredded(variantData, ctx, buffers)};
72+
shreddedProto = toProtoDtype(variantData.shreddedDtype());
73+
} else {
74+
children = new EncodeNode[]{coreStorage};
75+
}
76+
77+
ByteBuffer containerMeta = ByteBuffer.wrap(new VariantMetadata(shreddedProto).encode());
78+
EncodeNode root = new EncodeNode(EncodingId.VORTEX_VARIANT, containerMeta, children, new int[0]);
7179
return new EncodeResult(root, List.copyOf(buffers), null, null);
7280
}
7381

82+
/// Encoders eligible to back the shredded child, tried in order by dtype.
83+
private static final List<EncodingEncoder> SHREDDED_FALLBACK = List.of(
84+
new PrimitiveEncodingEncoder(), new VarBinEncodingEncoder(),
85+
new BoolEncodingEncoder(), new NullEncodingEncoder());
86+
87+
/// Encodes the shredded typed column as a child node, appending its buffers (remapped
88+
/// to follow the core-storage buffers already in `buffers`).
89+
private static EncodeNode encodeShredded(VariantData data, EncodeContext ctx, List<MemorySegment> buffers) {
90+
EncodingEncoder enc = null;
91+
for (EncodingEncoder e : SHREDDED_FALLBACK) {
92+
if (e.accepts(data.shreddedDtype())) {
93+
enc = e;
94+
break;
95+
}
96+
}
97+
if (enc == null) {
98+
throw new VortexException(EncodingId.VORTEX_VARIANT,
99+
"no encoder for shredded dtype: " + data.shreddedDtype());
100+
}
101+
EncodeResult shredded = enc.encode(data.shreddedDtype(), data.shreddedData(), ctx);
102+
EncodeNode child = EncodeNode.remapBufferIndices(shredded.rootNode(), buffers.size());
103+
buffers.addAll(shredded.buffers());
104+
return child;
105+
}
106+
107+
/// Converts a shreddable scalar dtype to its protobuf form for `VariantMetadata`.
108+
private static io.github.dfa1.vortex.proto.DType toProtoDtype(DType dtype) {
109+
return switch (dtype) {
110+
case DType.Primitive p -> io.github.dfa1.vortex.proto.DType.ofPrimitive(
111+
new io.github.dfa1.vortex.proto.Primitive(
112+
io.github.dfa1.vortex.proto.PType.fromValue(p.ptype().ordinal()), p.nullable()));
113+
case DType.Bool b -> io.github.dfa1.vortex.proto.DType.ofBool(
114+
new io.github.dfa1.vortex.proto.Bool(b.nullable()));
115+
case DType.Utf8 u -> io.github.dfa1.vortex.proto.DType.ofUtf8(
116+
new io.github.dfa1.vortex.proto.Utf8(u.nullable()));
117+
case DType.Binary bin -> io.github.dfa1.vortex.proto.DType.ofBinary(
118+
new io.github.dfa1.vortex.proto.Binary(bin.nullable()));
119+
default -> throw new VortexException(EncodingId.VORTEX_VARIANT,
120+
"shredded dtype not supported: " + dtype);
121+
};
122+
}
123+
74124
/// Groups adjacent equal scalars into runs, appending each run's value and length.
75125
private static void coalesceRuns(List<Scalar> values, List<Scalar> runValues, List<Long> runLengths) {
76126
Scalar prev = null;

‎writer/src/test/java/io/github/dfa1/vortex/writer/encode/VariantEncodingEncoderTest.java‎

Lines changed: 53 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -196,6 +196,59 @@ void varyingColumn_decodesPerRowValuesInOrder() {
196196
assertThat(core.getInt(1)).isEqualTo(20);
197197
assertThat(core.getInt(2)).isEqualTo(30);
198198
}
199+
200+
@Test
201+
void shreddedColumn_decodesShreddedTypedChild() {
202+
// Given/When a column with a shredded i32 projection is encoded then decoded
203+
DType i32 = new DType.Primitive(io.github.dfa1.vortex.core.PType.I32, false);
204+
var data = VariantData.shredded(
205+
List.of(i32Scalar(10L), i32Scalar(20L), i32Scalar(30L)), new int[]{10, 20, 30}, i32);
206+
var variant = decode(SUT.encode(VARIANT, data, EncodeTestHelper.testCtx()), 3);
207+
208+
// Then the shredded child decodes as the typed column
209+
assertThat(variant.shredded()).isNotNull();
210+
var shredded = (io.github.dfa1.vortex.reader.array.IntArray) variant.shredded();
211+
assertThat(shredded.dtype()).isEqualTo(i32);
212+
assertThat(shredded.getInt(0)).isEqualTo(10);
213+
assertThat(shredded.getInt(1)).isEqualTo(20);
214+
assertThat(shredded.getInt(2)).isEqualTo(30);
215+
}
216+
}
217+
218+
@Nested
219+
class Shredded {
220+
221+
@Test
222+
void emitsSecondChildAndRecordsShreddedDtype() throws Exception {
223+
// Given a column with a shredded i32 projection
224+
DType i32 = new DType.Primitive(io.github.dfa1.vortex.core.PType.I32, false);
225+
var data = VariantData.shredded(
226+
List.of(i32Scalar(10L), i32Scalar(20L), i32Scalar(30L)), new int[]{10, 20, 30}, i32);
227+
228+
// When
229+
EncodeResult result = SUT.encode(VARIANT, data, EncodeTestHelper.testCtx());
230+
231+
// Then the container has a second (shredded) child encoded as a primitive array...
232+
EncodeNode root = result.rootNode();
233+
assertThat(root.children()).hasSize(2);
234+
assertThat(root.children()[1].encodingId()).isEqualTo(EncodingId.VORTEX_PRIMITIVE);
235+
236+
// ...and the metadata records shredded_dtype = i32.
237+
MemorySegment meta = MemorySegment.ofBuffer(root.metadata().duplicate());
238+
VariantMetadata vm = VariantMetadata.decode(meta, 0, meta.byteSize());
239+
assertThat(vm.shredded_dtype()).isNotNull();
240+
assertThat(vm.shredded_dtype().primitive()).isNotNull();
241+
assertThat(vm.shredded_dtype().primitive().type()).isEqualTo(io.github.dfa1.vortex.proto.PType.I32);
242+
}
243+
244+
@Test
245+
void halfSpecifiedShredded_throws() {
246+
assertThatThrownBy(() -> VariantData.shredded(List.of(i32Scalar(1L)), null, null))
247+
.isInstanceOf(IllegalArgumentException.class);
248+
assertThatThrownBy(() -> new VariantData(List.of(i32Scalar(1L)), new int[]{1}, null))
249+
.isInstanceOf(IllegalArgumentException.class)
250+
.hasMessageContaining("both set or both null");
251+
}
199252
}
200253

201254
@Nested

0 commit comments

Comments
 (0)