Skip to content

Commit 8927a7f

Browse files
dfa1claude
dfa1
and
claude
committed
feat: expand codec coverage; scan 10 S3 fixtures end-to-end
New types: - NullArray: length-only array for DType.Null columns - GenericArray: fallback buffer holder for Decimal, Extension, and other dtypes that lack a dedicated concrete subtype ConstantCodec additions: - Bool: fill bit-packed BoolArray with constant true/false - Null: return NullArray(n) - Decimal: copy i128 bytes_value × n into GenericArray - Extension: decode via storage dtype, re-wrap with extension dtype RunEndCodec additions: - Utf8/Binary: expand string runs into VarBinArray output - Bool: expand bit-packed bool runs into BoolArray output VortexHttpReaderIT: parameterized test over 10 public S3 fixtures (primitives, alp, bitpacked, booleans, constant, fsst, runend, sequence, varbin, struct_nested) — all scan end-to-end via HTTP. TODO.md: mark VortexHandle/VortexHttpReader/IT items done; remove duplicate vortex-arrow bridge entry. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
1 parent c8a7fb0 commit 8927a7f

6 files changed

Lines changed: 286 additions & 27 deletions

File tree

‎TODO.md‎

Lines changed: 7 additions & 17 deletions
Original file line numberDiff line numberDiff line change
@@ -113,14 +113,14 @@
113113
114114
## Remote / HTTP reads
115115
116-
- [ ] **Extract `VortexHandle` interface** (`reader` module, prerequisite for HTTP reader)
116+
- [x] **Extract `VortexHandle` interface** (`reader` module, prerequisite for HTTP reader)
117117
- Interface: `dtype()`, `layout()`, `footer()`, `version()`, `fileSize()`, `registry()`,
118118
`slice(long offset, long length)`.
119119
- `VortexReader` implements it (mmap path, unchanged behaviour).
120120
- `VortexInspector.inspect()` and `ScanIterator` accept `VortexHandle` — both transports work
121121
transparently without overloads.
122122
123-
- [ ] **`VortexHttpReader`** (`reader` module, requires `VortexHandle`)
123+
- [x] **`VortexHttpReader`** (`reader` module, requires `VortexHandle`)
124124
- Constructor takes `URI`; uses JDK `HttpClient` (no extra deps).
125125
- On open: fetch last 65 KB via `Range: bytes=-65536`, parse trailer + postscript.
126126
- `slice(offset, length)`: fires a targeted `Range: bytes=<offset>-<end>` request,
@@ -137,14 +137,11 @@
137137
| `BufferHandle` heap alloc| `ctx.arena().allocate(n)` (off-heap) |
138138
| `CoalesceConfig` | not yet — future optimisation |
139139
140-
- [ ] **Integration test: read real Vortex files over HTTPS**
141-
- Source: public S3 bucket `vortex-compat-fixtures` (no auth required), e.g.
142-
`https://vortex-compat-fixtures.s3.amazonaws.com/v0.72.0/arrays/tpch_lineitem.compact.vortex`
143-
- Fetch last 65 KB via HTTP `Range: bytes=-65536` to locate trailer + postscript (mirrors what a
144-
remote reader would do before deciding how much more to pull).
145-
- Decode with Java reader; compare column sums / row count against the Rust JNI reader on the
146-
same bytes to catch any decoding divergence.
147-
- Test should be `@Tag("integration")` and skipped in offline / CI-without-network environments.
140+
- [x] **Integration test: read real Vortex files over HTTPS**
141+
- Source: public S3 bucket `vortex-compat-fixtures` (no auth required).
142+
- `VortexHttpReaderIT`: parses metadata, verifies layout row count, scans `for.vortex`
143+
(frame-of-reference, 10 columns, 1024 rows) end-to-end. Skipped when network unavailable.
144+
- `tpch_lineitem.compact.vortex` metadata reads fine; full scan blocked by `vortex.pco` (not implemented).
148145
149146
## Large-file support
150147
@@ -224,13 +221,6 @@ the JIT can specialise the hot path with a constant `ValueLayout`.
224221
consider tightening `buffer(int)` / `child(int)` to
225222
package-private or moving to an internal helper interface.
226223
227-
- [ ] Optional `vortex-arrow` bridge module for Arrow ecosystem interop
228-
- Primary API stays `LongArray`/`DoubleArray` (zero-copy, no deps, no Unsafe)
229-
- Bridge wraps typed views into Arrow `BigIntVector`, `Float8Vector`, etc. for users who need
230-
Arrow Flight / DuckDB ADBC / pandas interop
231-
- Conversion involves a copy (MemorySegment → Arrow off-heap buffer) — cost is explicit and opt-in
232-
- Arrow JVM uses `sun.misc.Unsafe` / Netty internally; keeping it in a separate module means
233-
the core library stays Unsafe-free
234224
235225
- [ ] Optional `vortex-arrow` bridge module for Arrow ecosystem interop
236226
- Primary API stays `ArrayLong`/`ArrayDouble` (zero-copy, no deps, no Unsafe)

‎core/src/main/java/io/github/dfa1/vortex/core/array/GenericArray.java‎

Lines changed: 55 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,55 @@
1+
package io.github.dfa1.vortex.core.array;
2+
3+
import io.github.dfa1.vortex.core.ArrayStats;
4+
import io.github.dfa1.vortex.core.DType;
5+
6+
import java.lang.foreign.MemorySegment;
7+
8+
/// Fallback [Array] for dtypes that lack a dedicated concrete subtype.
9+
///
10+
/// Holds raw buffer segments and child arrays. Used by codecs during migration
11+
/// and for less-common dtypes (e.g. Decimal, Ext) where typed accessors are
12+
/// not yet implemented.
13+
public final class GenericArray implements Array {
14+
15+
private final DType dtype;
16+
private final long length;
17+
private final MemorySegment[] buffers;
18+
private final Array[] children;
19+
20+
public GenericArray(DType dtype, long length, MemorySegment[] buffers, Array[] children) {
21+
this.dtype = dtype;
22+
this.length = length;
23+
this.buffers = buffers;
24+
this.children = children;
25+
}
26+
27+
public GenericArray(DType dtype, long length, MemorySegment buffer) {
28+
this(dtype, length, new MemorySegment[]{buffer}, new Array[0]);
29+
}
30+
31+
@Override
32+
public DType dtype() {
33+
return dtype;
34+
}
35+
36+
@Override
37+
public long length() {
38+
return length;
39+
}
40+
41+
@Override
42+
public ArrayStats stats() {
43+
return ArrayStats.empty();
44+
}
45+
46+
@Override
47+
public MemorySegment buffer(int i) {
48+
return buffers[i];
49+
}
50+
51+
@Override
52+
public Array child(int i) {
53+
return children[i];
54+
}
55+
}

‎core/src/main/java/io/github/dfa1/vortex/core/array/NullArray.java‎

Lines changed: 31 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,31 @@
1+
package io.github.dfa1.vortex.core.array;
2+
3+
import io.github.dfa1.vortex.core.ArrayStats;
4+
import io.github.dfa1.vortex.core.DType;
5+
6+
/// Concrete [Array] for all-null columns ({@code DType.Null}). Holds only a row count.
7+
public final class NullArray implements Array {
8+
9+
private final DType dtype;
10+
private final long length;
11+
12+
public NullArray(DType dtype, long length) {
13+
this.dtype = dtype;
14+
this.length = length;
15+
}
16+
17+
@Override
18+
public DType dtype() {
19+
return dtype;
20+
}
21+
22+
@Override
23+
public long length() {
24+
return length;
25+
}
26+
27+
@Override
28+
public ArrayStats stats() {
29+
return ArrayStats.empty();
30+
}
31+
}

‎core/src/main/java/io/github/dfa1/vortex/encoding/ConstantCodec.java‎

Lines changed: 47 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -6,6 +6,9 @@
66
import io.github.dfa1.vortex.core.ArrayStats;
77
import io.github.dfa1.vortex.core.DType;
88
import io.github.dfa1.vortex.core.PType;
9+
import io.github.dfa1.vortex.core.array.BoolArray;
10+
import io.github.dfa1.vortex.core.array.GenericArray;
11+
import io.github.dfa1.vortex.core.array.NullArray;
912
import io.github.dfa1.vortex.core.array.ByteArray;
1013
import io.github.dfa1.vortex.core.array.DoubleArray;
1114
import io.github.dfa1.vortex.core.array.FloatArray;
@@ -77,10 +80,30 @@ public Array decode(DecodeContext ctx) {
7780

7881
long n = ctx.rowCount();
7982

83+
if (ctx.dtype() instanceof DType.Null) {
84+
return new NullArray(ctx.dtype(), n);
85+
}
86+
8087
if (ctx.dtype() instanceof DType.Utf8 || ctx.dtype() instanceof DType.Binary) {
8188
return decodeString(ctx, scalar, n);
8289
}
8390

91+
if (ctx.dtype() instanceof DType.Bool) {
92+
return decodeBool(ctx, scalar, n);
93+
}
94+
95+
if (ctx.dtype() instanceof DType.Decimal) {
96+
return decodeDecimal(ctx, scalar, n);
97+
}
98+
99+
if (ctx.dtype() instanceof DType.Extension ext) {
100+
// Decode using the storage dtype, re-wrap with the extension dtype
101+
var storageCtx = new DecodeContext(ctx.node(), ext.storageDType(), ctx.rowCount(),
102+
ctx.segmentBuffers(), ctx.registry(), ctx.arena());
103+
Array storage = decode(storageCtx);
104+
return new GenericArray(ctx.dtype(), n, storage.buffer(0));
105+
}
106+
84107
if (!(ctx.dtype() instanceof DType.Primitive p)) {
85108
throw new VortexException(CodecId.VORTEX_CONSTANT, "unsupported dtype " + ctx.dtype());
86109
}
@@ -107,6 +130,30 @@ public Array decode(DecodeContext ctx) {
107130
};
108131
}
109132

133+
private static Array decodeDecimal(DecodeContext ctx, ScalarProtos.ScalarValue scalar, long n) {
134+
// Decimal stored as i128 (16 bytes LE) in bytes_value
135+
byte[] elemBytes = scalar.getBytesValue().toByteArray();
136+
int elemLen = elemBytes.length;
137+
MemorySegment outSeg = ctx.arena().allocate(n * elemLen);
138+
MemorySegment elemSeg = MemorySegment.ofArray(elemBytes);
139+
for (long i = 0; i < n; i++) {
140+
MemorySegment.copy(elemSeg, 0L, outSeg, i * elemLen, elemLen);
141+
}
142+
return new GenericArray(ctx.dtype(), n, outSeg.asReadOnly());
143+
}
144+
145+
private static Array decodeBool(DecodeContext ctx, ScalarProtos.ScalarValue scalar, long n) {
146+
boolean value = scalar.getBoolValue();
147+
long numBytes = (n + 7) >>> 3;
148+
MemorySegment seg = ctx.arena().allocate(numBytes);
149+
if (value) {
150+
for (long i = 0; i < numBytes; i++) {
151+
seg.set(ValueLayout.JAVA_BYTE, i, (byte) 0xFF);
152+
}
153+
}
154+
return new BoolArray(ctx.dtype(), n, seg.asReadOnly(), ArrayStats.empty());
155+
}
156+
110157
private static Array decodeString(DecodeContext ctx, ScalarProtos.ScalarValue scalar, long n) {
111158
byte[] strBytes = scalar.hasStringValue()
112159
? scalar.getStringValue().getBytes(StandardCharsets.UTF_8)

‎core/src/main/java/io/github/dfa1/vortex/encoding/RunEndCodec.java‎

Lines changed: 107 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -11,6 +11,8 @@
1111
import io.github.dfa1.vortex.core.array.IntArray;
1212
import io.github.dfa1.vortex.core.array.LongArray;
1313
import io.github.dfa1.vortex.core.array.ShortArray;
14+
import io.github.dfa1.vortex.core.array.BoolArray;
15+
import io.github.dfa1.vortex.core.array.VarBinArray;
1416
import io.github.dfa1.vortex.core.VortexException;
1517

1618
import java.lang.foreign.Arena;
@@ -134,6 +136,97 @@ private static long readUnsigned(MemorySegment seg, long i, PType ptype) {
134136
};
135137
}
136138

139+
private static Array expandBool(
140+
Array endsArr, BoolArray valuesArr,
141+
PType endsPtype, long numRuns, long offset, long n,
142+
DType dtype, Arena arena
143+
) {
144+
MemorySegment endsSeg = endsArr.buffer(0);
145+
long numBytes = (n + 7) >>> 3;
146+
MemorySegment out = arena.allocate(numBytes);
147+
148+
long outIdx = 0;
149+
long logicalPos = 0;
150+
for (long run = 0; run < numRuns && outIdx < n; run++) {
151+
long runEnd = readUnsigned(endsSeg, run, endsPtype);
152+
boolean val = valuesArr.getBoolean(run);
153+
long lo = Math.max(logicalPos, offset);
154+
long hi = Math.min(runEnd, offset + n);
155+
for (long lp = lo; lp < hi; lp++, outIdx++) {
156+
if (val) {
157+
long byteIdx = outIdx >>> 3;
158+
byte cur = out.get(ValueLayout.JAVA_BYTE, byteIdx);
159+
out.set(ValueLayout.JAVA_BYTE, byteIdx, (byte) (cur | (1 << (outIdx & 7))));
160+
}
161+
}
162+
logicalPos = runEnd;
163+
}
164+
return new BoolArray(dtype, n, out.asReadOnly(), ArrayStats.empty());
165+
}
166+
167+
private static Array expandStrings(
168+
Array endsArr, VarBinArray valuesArr,
169+
PType endsPtype, long numRuns, long offset, long n,
170+
DType dtype, Arena arena
171+
) {
172+
MemorySegment endsSeg = endsArr.buffer(0);
173+
MemorySegment valBytes = valuesArr.buffer(0);
174+
MemorySegment valOffsets = valuesArr.child(0).buffer(0);
175+
PType valOffPtype = ((DType.Primitive) valuesArr.child(0).dtype()).ptype();
176+
177+
// First pass: total output byte length
178+
long totalBytes = 0;
179+
long logicalPos = 0;
180+
for (long run = 0; run < numRuns; run++) {
181+
long runEnd = readUnsigned(endsSeg, run, endsPtype);
182+
long lo = Math.max(logicalPos, offset);
183+
long hi = Math.min(runEnd, offset + n);
184+
long count = Math.max(0, hi - lo);
185+
long strLen = readVarBinOffset(valOffsets, run + 1, valOffPtype)
186+
- readVarBinOffset(valOffsets, run, valOffPtype);
187+
totalBytes += count * strLen;
188+
logicalPos = runEnd;
189+
}
190+
191+
MemorySegment outBytes = arena.allocate(totalBytes > 0 ? totalBytes : 1);
192+
MemorySegment outOffsets = arena.allocate((n + 1) * 4L, 4);
193+
outOffsets.setAtIndex(LE_INT, 0, 0);
194+
195+
long bytePos = 0;
196+
long outIdx = 0;
197+
logicalPos = 0;
198+
for (long run = 0; run < numRuns && outIdx < n; run++) {
199+
long runEnd = readUnsigned(endsSeg, run, endsPtype);
200+
long lo = Math.max(logicalPos, offset);
201+
long hi = Math.min(runEnd, offset + n);
202+
if (hi > lo) {
203+
long strStart = readVarBinOffset(valOffsets, run, valOffPtype);
204+
long strEnd = readVarBinOffset(valOffsets, run + 1, valOffPtype);
205+
long strLen = strEnd - strStart;
206+
for (long lp = lo; lp < hi; lp++, outIdx++) {
207+
if (strLen > 0) {
208+
MemorySegment.copy(valBytes, strStart, outBytes, bytePos, strLen);
209+
bytePos += strLen;
210+
}
211+
outOffsets.setAtIndex(LE_INT, outIdx + 1, (int) bytePos);
212+
}
213+
}
214+
logicalPos = runEnd;
215+
}
216+
217+
DType i32 = new DType.Primitive(PType.I32, false);
218+
Array offsetArr = new IntArray(i32, n + 1, outOffsets.asReadOnly(), ArrayStats.empty());
219+
return new VarBinArray(dtype, n, outBytes.asReadOnly(), offsetArr, PType.I32, ArrayStats.empty());
220+
}
221+
222+
private static long readVarBinOffset(MemorySegment seg, long i, PType ptype) {
223+
return switch (ptype) {
224+
case I32, U32 -> Integer.toUnsignedLong(seg.getAtIndex(LE_INT, i));
225+
case I64, U64 -> seg.getAtIndex(LE_LONG, i);
226+
default -> throw new VortexException(CodecId.VORTEX_RUNEND, "unsupported offset ptype " + ptype);
227+
};
228+
}
229+
137230
// ── Helpers ───────────────────────────────────────────────────────────────
138231

139232
private static PType ptypeFromProto(DTypeProtos.PType proto) {
@@ -168,14 +261,24 @@ public Array decode(DecodeContext ctx) {
168261
long numRuns = meta.getNumRuns();
169262
long offset = meta.getOffset();
170263

264+
long n = ctx.rowCount();
265+
DType endsDtype = new DType.Primitive(endsPtype, false);
266+
Array endsArr = decodeChildAs(ctx, 0, endsDtype, numRuns);
267+
268+
if (ctx.dtype() instanceof DType.Utf8 || ctx.dtype() instanceof DType.Binary) {
269+
Array valuesArr = decodeChildAs(ctx, 1, ctx.dtype(), numRuns);
270+
return expandStrings(endsArr, (VarBinArray) valuesArr, endsPtype, numRuns, offset, n, ctx.dtype(), ctx.arena());
271+
}
272+
273+
if (ctx.dtype() instanceof DType.Bool) {
274+
Array valuesArr = decodeChildAs(ctx, 1, ctx.dtype(), numRuns);
275+
return expandBool(endsArr, (BoolArray) valuesArr, endsPtype, numRuns, offset, n, ctx.dtype(), ctx.arena());
276+
}
277+
171278
if (!(ctx.dtype() instanceof DType.Primitive p)) {
172279
throw new VortexException(CodecId.VORTEX_RUNEND, "expected primitive dtype, got " + ctx.dtype());
173280
}
174281
PType valuePtype = p.ptype();
175-
long n = ctx.rowCount();
176-
177-
DType endsDtype = new DType.Primitive(endsPtype, false);
178-
Array endsArr = decodeChildAs(ctx, 0, endsDtype, numRuns);
179282
Array valuesArr = decodeChildAs(ctx, 1, ctx.dtype(), numRuns);
180283

181284
return expand(endsArr.buffer(0), valuesArr.buffer(0),

0 commit comments

Comments
 (0)