Skip to content

Commit a63980f

Browse files
dfa1claude
dfa1
and
claude
committed
feat(compat): JNI↔Java cross-compatibility tests
- Add integration tests: RustWritesJavaReadsIT and JavaWritesRustReadsIT - Add SequenceCodec (vortex.sequence: A[i] = base + i * multiplier) - Fix writer: Buffer.alignment_exponent=6, SegmentSpec.alignment_exponent=6, and pre-segment 64-byte padding — Arrow rejects buffers whose logical alignment field is below 64 bytes, even if physically aligned - Wire vortex-jni 0.72.0 + Arrow 19.0.0 into integration module Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
1 parent 4abd197 commit a63980f

8 files changed

Lines changed: 402 additions & 100 deletions

File tree

‎TODO.md‎

Lines changed: 12 additions & 14 deletions
Original file line numberDiff line numberDiff line change
@@ -67,20 +67,18 @@
6767
- Step 4: apply patches for exceptions that don't fit the ALP transform
6868
- Reference: `encodings/alp/src/alp/decompress.rs`
6969

70-
## Cross-compatibility (blocked by: JNI bindings)
71-
72-
- [ ] **#8 Rust writes → Java reads**
73-
- Prerequisite: #7a (bitpacked JNI decode), #7b (for), #7c (sparse), #7d (alp)
74-
- Step 1: integer columns only — I64 round-trip through JNI writer → Java reader
75-
- Step 2: float columns — F64 (requires #7d ALP decoder)
76-
- Step 3: full OHLC file (all column types) — assert sum/close values match JNI reader
77-
- Tracked in `OhlcReadBenchmark`: `javaReadClose` and `javaReadSum` must equal `jniReadClose`
78-
79-
- [ ] **#9 Java writes → Rust reads**
80-
- Use `VortexWriter` to produce a `.vtx` file
81-
- Decode with JNI reader, assert decoded values match input
82-
83-
**Module:** `integration/` — activate with `-Pintegration`. Tests `@Disabled` until JNI artifact coordinates are known.
70+
## Cross-compatibility
71+
72+
- [x] **#8 Rust writes → Java reads** (`RustWritesJavaReadsIT`, `-Pintegration`)
73+
- JNI writes I64+F64 file; Java reader decodes via `DecoderRegistry.loadAll()`
74+
- Fixed: added `SequenceCodec` (`vortex.sequence` = `A[i] = base + i * multiplier`)
75+
- Covers: single chunk, multiple chunks (JNI may merge), column projection
76+
77+
- [x] **#9 Java writes → Rust reads** (`JavaWritesRustReadsIT`, `-Pintegration`)
78+
- Java writer produces file; JNI reader decodes via Arrow C Data Interface
79+
- Fixed writer: `Buffer.alignment_exponent = 6` + `SegmentSpec.alignment_exponent = 6` + pre-segment 64-byte padding
80+
- Root cause: Rust decoder tracks logical alignment from FlatBuffer field; Arrow rejects buffers with alignment < 64 bytes
81+
- Covers: single chunk, multiple chunks
8482

8583
## Performance (blocked by: JNI bindings for comparison baseline)
8684

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

Lines changed: 108 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,108 @@
1+
package io.github.dfa1.vortex.encoding;
2+
3+
import com.google.protobuf.InvalidProtocolBufferException;
4+
import dev.vortex.proto.EncodingProtos;
5+
import io.github.dfa1.vortex.core.DType;
6+
import io.github.dfa1.vortex.core.PType;
7+
8+
import java.lang.foreign.MemorySegment;
9+
import java.nio.ByteBuffer;
10+
import java.nio.ByteOrder;
11+
12+
/// Decoder for {@code vortex.sequence}: {@code A[i] = base + i * multiplier}.
13+
///
14+
/// No buffers, no children. Metadata is a protobuf {@code SequenceMetadata}
15+
/// with {@code base} (tag 1) and {@code multiplier} (tag 2) as {@code ScalarValue}.
16+
/// Output is allocated on the heap; not backed by the file's mapped region.
17+
public final class SequenceCodec implements Decoder {
18+
19+
@Override
20+
public String encodingId() {
21+
return "vortex.sequence";
22+
}
23+
24+
@Override
25+
public Array decode(DecodeContext ctx) {
26+
ByteBuffer metaBuf = ctx.metadata();
27+
if (metaBuf == null || !metaBuf.hasRemaining()) {
28+
throw new IllegalStateException("vortex.sequence: missing metadata");
29+
}
30+
byte[] metaBytes = new byte[metaBuf.remaining()];
31+
metaBuf.duplicate().get(metaBytes);
32+
33+
EncodingProtos.SequenceMetadata meta;
34+
try {
35+
meta = EncodingProtos.SequenceMetadata.parseFrom(metaBytes);
36+
} catch (InvalidProtocolBufferException e) {
37+
throw new IllegalStateException("vortex.sequence: invalid metadata", e);
38+
}
39+
40+
if (!(ctx.dtype() instanceof DType.Primitive p)) {
41+
throw new IllegalStateException("vortex.sequence: expected primitive dtype, got " + ctx.dtype());
42+
}
43+
44+
long n = ctx.rowCount();
45+
PType pt = p.ptype();
46+
return switch (pt) {
47+
case I8, I16, I32, I64, U8, U16, U32, U64 -> decodeInteger(meta, pt, n, ctx.dtype());
48+
case F32 -> decodeF32(meta, n, ctx.dtype());
49+
case F64 -> decodeF64(meta, n, ctx.dtype());
50+
case F16 -> throw new IllegalStateException("vortex.sequence: F16 not supported");
51+
};
52+
}
53+
54+
private static Array decodeInteger(
55+
EncodingProtos.SequenceMetadata meta, PType pt, long n, DType dtype
56+
) {
57+
long base = signedValue(meta.getBase());
58+
long mul = signedValue(meta.getMultiplier());
59+
int elemBytes = pt.byteSize();
60+
ByteBuffer buf = ByteBuffer.allocate((int) (n * elemBytes)).order(ByteOrder.LITTLE_ENDIAN);
61+
for (long i = 0; i < n; i++) {
62+
long v = base + i * mul;
63+
switch (pt) {
64+
case I8, U8 -> buf.put((byte) v);
65+
case I16, U16 -> buf.putShort((short) v);
66+
case I32, U32 -> buf.putInt((int) v);
67+
case I64, U64 -> buf.putLong(v);
68+
default -> throw new IllegalStateException("unreachable");
69+
}
70+
}
71+
buf.flip();
72+
return new Array(dtype, n, new MemorySegment[]{MemorySegment.ofBuffer(buf)},
73+
new Array[0], ArrayStats.empty());
74+
}
75+
76+
private static Array decodeF32(EncodingProtos.SequenceMetadata meta, long n, DType dtype) {
77+
float base = meta.getBase().getF32Value();
78+
float mul = meta.getMultiplier().getF32Value();
79+
ByteBuffer buf = ByteBuffer.allocate((int) (n * 4)).order(ByteOrder.LITTLE_ENDIAN);
80+
for (long i = 0; i < n; i++) {
81+
buf.putFloat(base + i * mul);
82+
}
83+
buf.flip();
84+
return new Array(dtype, n, new MemorySegment[]{MemorySegment.ofBuffer(buf)},
85+
new Array[0], ArrayStats.empty());
86+
}
87+
88+
private static Array decodeF64(EncodingProtos.SequenceMetadata meta, long n, DType dtype) {
89+
double base = meta.getBase().getF64Value();
90+
double mul = meta.getMultiplier().getF64Value();
91+
ByteBuffer buf = ByteBuffer.allocate((int) (n * 8)).order(ByteOrder.LITTLE_ENDIAN);
92+
for (long i = 0; i < n; i++) {
93+
buf.putDouble(base + i * mul);
94+
}
95+
buf.flip();
96+
return new Array(dtype, n, new MemorySegment[]{MemorySegment.ofBuffer(buf)},
97+
new Array[0], ArrayStats.empty());
98+
}
99+
100+
private static long signedValue(dev.vortex.proto.ScalarProtos.ScalarValue sv) {
101+
return switch (sv.getKindCase()) {
102+
case INT64_VALUE -> sv.getInt64Value();
103+
case UINT64_VALUE -> sv.getUint64Value();
104+
case KIND_NOT_SET -> 0L;
105+
default -> throw new IllegalStateException("vortex.sequence: unexpected scalar kind " + sv.getKindCase());
106+
};
107+
}
108+
}

‎core/src/main/proto/encodings.proto‎

Lines changed: 7 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -6,7 +6,14 @@ package vortex.encodings;
66
option java_package = "dev.vortex.proto";
77
option java_outer_classname = "EncodingProtos";
88

9+
import "scalar.proto";
10+
911
message BitPackedMetadata {
1012
uint32 bit_width = 1;
1113
uint32 offset = 2;
1214
}
15+
16+
message SequenceMetadata {
17+
vortex.scalar.ScalarValue base = 1;
18+
vortex.scalar.ScalarValue multiplier = 2;
19+
}

‎core/src/main/resources/META-INF/services/io.github.dfa1.vortex.encoding.Decoder‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -3,3 +3,4 @@ io.github.dfa1.vortex.encoding.BoolCodec
33
io.github.dfa1.vortex.encoding.DictCodec
44
io.github.dfa1.vortex.encoding.BitpackedCodec
55
io.github.dfa1.vortex.encoding.DeltaCodec
6+
io.github.dfa1.vortex.encoding.SequenceCodec

‎integration/pom.xml‎

Lines changed: 48 additions & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -22,17 +22,30 @@
2222
<artifactId>reader</artifactId>
2323
<scope>test</scope>
2424
</dependency>
25-
<!--
26-
JNI vortex bindings — not yet published.
27-
Uncomment and set coordinates when the artifact is available:
28-
29-
<dependency>
30-
<groupId>io.github.spiraldb</groupId>
31-
<artifactId>vortex-jni</artifactId>
32-
<version>TODO</version>
33-
<scope>test</scope>
34-
</dependency>
35-
-->
25+
<dependency>
26+
<groupId>dev.vortex</groupId>
27+
<artifactId>vortex-jni</artifactId>
28+
<version>0.72.0</version>
29+
<scope>test</scope>
30+
</dependency>
31+
<dependency>
32+
<groupId>org.apache.arrow</groupId>
33+
<artifactId>arrow-vector</artifactId>
34+
<version>19.0.0</version>
35+
<scope>test</scope>
36+
</dependency>
37+
<dependency>
38+
<groupId>org.apache.arrow</groupId>
39+
<artifactId>arrow-c-data</artifactId>
40+
<version>19.0.0</version>
41+
<scope>test</scope>
42+
</dependency>
43+
<dependency>
44+
<groupId>org.apache.arrow</groupId>
45+
<artifactId>arrow-memory-unsafe</artifactId>
46+
<version>19.0.0</version>
47+
<scope>test</scope>
48+
</dependency>
3649
<dependency>
3750
<groupId>org.junit.jupiter</groupId>
3851
<artifactId>junit-jupiter</artifactId>
@@ -42,4 +55,28 @@
4255
<artifactId>assertj-core</artifactId>
4356
</dependency>
4457
</dependencies>
58+
59+
<build>
60+
<plugins>
61+
<plugin>
62+
<groupId>org.apache.maven.plugins</groupId>
63+
<artifactId>maven-failsafe-plugin</artifactId>
64+
<version>3.5.2</version>
65+
<configuration>
66+
<argLine>
67+
--add-opens java.base/java.nio=ALL-UNNAMED
68+
--enable-native-access=ALL-UNNAMED
69+
</argLine>
70+
</configuration>
71+
<executions>
72+
<execution>
73+
<goals>
74+
<goal>integration-test</goal>
75+
<goal>verify</goal>
76+
</goals>
77+
</execution>
78+
</executions>
79+
</plugin>
80+
</plugins>
81+
</build>
4582
</project>

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

Lines changed: 86 additions & 24 deletions
Original file line numberDiff line numberDiff line change
@@ -1,71 +1,133 @@
11
package io.github.dfa1.vortex.integration;
22

3+
import dev.vortex.api.DataSource;
4+
import dev.vortex.api.Expression;
5+
import dev.vortex.api.Partition;
6+
import dev.vortex.api.Scan;
7+
import dev.vortex.api.ScanOptions;
8+
import dev.vortex.api.Session;
9+
import dev.vortex.arrow.ArrowAllocation;
10+
import dev.vortex.jni.NativeLoader;
311
import io.github.dfa1.vortex.core.DType;
412
import io.github.dfa1.vortex.core.PType;
513
import io.github.dfa1.vortex.writer.VortexWriter;
614
import io.github.dfa1.vortex.writer.WriteOptions;
7-
import org.junit.jupiter.api.Disabled;
15+
import org.apache.arrow.memory.BufferAllocator;
16+
import org.apache.arrow.vector.BigIntVector;
17+
import org.apache.arrow.vector.Float8Vector;
18+
import org.apache.arrow.vector.VectorSchemaRoot;
19+
import org.apache.arrow.vector.ipc.ArrowReader;
820
import org.junit.jupiter.api.Test;
921
import org.junit.jupiter.api.io.TempDir;
1022

1123
import java.io.IOException;
1224
import java.nio.channels.FileChannel;
1325
import java.nio.file.Path;
1426
import java.nio.file.StandardOpenOption;
27+
import java.util.ArrayList;
1528
import java.util.List;
1629
import java.util.Map;
1730

1831
import static org.assertj.core.api.Assertions.assertThat;
1932

2033
/// Cross-compatibility: Java writer → Rust (JNI) reader.
21-
///
22-
/// Tests are disabled until the JNI vortex bindings are available.
23-
/// See TODO.md #9 and the commented-out `vortex-jni` dep in `it/pom.xml`.
24-
@Disabled("blocked by JNI bindings — see TODO.md #9")
2534
class JavaWritesRustReadsIT {
2635

36+
static {
37+
NativeLoader.loadJni();
38+
}
39+
40+
private static final Session SESSION = Session.create();
41+
private static final BufferAllocator ALLOCATOR = ArrowAllocation.rootAllocator();
42+
2743
private static final DType.Struct SCHEMA = new DType.Struct(
2844
List.of("id", "value"),
2945
List.of(new DType.Primitive(PType.I64, false),
3046
new DType.Primitive(PType.F64, false)),
3147
false);
3248

3349
@Test
34-
void javaWriter_rustReader_singleChunk(@TempDir Path tmp) throws IOException {
50+
void javaWriter_jniReader_singleChunk(@TempDir Path tmp) throws IOException {
3551
// Given
36-
Path file = tmp.resolve("java_written.vtx");
52+
Path file = tmp.resolve("java_single.vtx");
3753
long[] ids = {1L, 2L, 3L};
3854
double[] vals = {1.1, 2.2, 3.3};
39-
40-
// When
4155
try (var ch = FileChannel.open(file, StandardOpenOption.CREATE, StandardOpenOption.WRITE);
4256
var sut = VortexWriter.create(ch, SCHEMA, WriteOptions.defaults())) {
4357
sut.writeChunk(Map.of("id", ids, "value", vals));
4458
}
4559

60+
// When
61+
long[] decodedIds = readLongColumn(file, "id");
62+
double[] decodedVals = readDoubleColumn(file, "value");
63+
4664
// Then
47-
// TODO: replace with JNI call once bindings are available
48-
// long[] decodedIds = VortexJni.readColumn(file, "id", long[].class);
49-
// double[] decodedVals = VortexJni.readColumn(file, "value", double[].class);
50-
// assertThat(decodedIds).containsExactly(1L, 2L, 3L);
51-
// assertThat(decodedVals).containsExactly(1.1, 2.2, 3.3);
52-
throw new UnsupportedOperationException("JNI reader not available");
65+
assertThat(decodedIds).containsExactly(1L, 2L, 3L);
66+
assertThat(decodedVals).containsExactly(1.1, 2.2, 3.3);
5367
}
5468

5569
@Test
56-
void javaWriter_rustReader_multipleChunks(@TempDir Path tmp) throws IOException {
70+
void javaWriter_jniReader_multipleChunks(@TempDir Path tmp) throws IOException {
5771
// Given
58-
Path file = tmp.resolve("java_written_multi.vtx");
59-
60-
// When
72+
Path file = tmp.resolve("java_multi.vtx");
6173
try (var ch = FileChannel.open(file, StandardOpenOption.CREATE, StandardOpenOption.WRITE);
6274
var sut = VortexWriter.create(ch, SCHEMA, WriteOptions.defaults())) {
63-
sut.writeChunk(Map.of("id", new long[]{1L, 2L}, "value", new double[]{1.1, 2.2}));
64-
sut.writeChunk(Map.of("id", new long[]{3L, 4L, 5L}, "value", new double[]{3.3, 4.4, 5.5}));
75+
sut.writeChunk(Map.of("id", new long[]{1L, 2L}, "value", new double[]{1.1, 2.2}));
76+
sut.writeChunk(Map.of("id", new long[]{3L, 4L, 5L}, "value", new double[]{3.3, 4.4, 5.5}));
6577
}
6678

67-
// Then
68-
// TODO: assert JNI reader sees 5 total rows across 2 chunks
69-
throw new UnsupportedOperationException("JNI reader not available");
79+
// When
80+
long[] decodedIds = readLongColumn(file, "id");
81+
82+
// Then — JNI may merge chunks; verify all values present regardless of partition count
83+
assertThat(decodedIds).containsExactly(1L, 2L, 3L, 4L, 5L);
84+
}
85+
86+
// ── JNI read helpers ──────────────────────────────────────────────────────
87+
88+
private static long[] readLongColumn(Path file, String column) throws IOException {
89+
String uri = file.toAbsolutePath().toUri().toString();
90+
ScanOptions opts = ScanOptions.builder()
91+
.projection(Expression.select(new String[]{column}, Expression.root()))
92+
.build();
93+
var longs = new ArrayList<Long>();
94+
DataSource ds = DataSource.open(SESSION, uri);
95+
Scan scan = ds.scan(opts);
96+
while (scan.hasNext()) {
97+
Partition partition = scan.next();
98+
try (ArrowReader reader = partition.scanArrow(ALLOCATOR)) {
99+
while (reader.loadNextBatch()) {
100+
VectorSchemaRoot root = reader.getVectorSchemaRoot();
101+
BigIntVector vec = (BigIntVector) root.getVector(column);
102+
for (int i = 0; i < root.getRowCount(); i++) {
103+
longs.add(vec.get(i));
104+
}
105+
}
106+
}
107+
}
108+
return longs.stream().mapToLong(Long::longValue).toArray();
109+
}
110+
111+
private static double[] readDoubleColumn(Path file, String column) throws IOException {
112+
String uri = file.toAbsolutePath().toUri().toString();
113+
ScanOptions opts = ScanOptions.builder()
114+
.projection(Expression.select(new String[]{column}, Expression.root()))
115+
.build();
116+
var doubles = new ArrayList<Double>();
117+
DataSource ds = DataSource.open(SESSION, uri);
118+
Scan scan = ds.scan(opts);
119+
while (scan.hasNext()) {
120+
Partition partition = scan.next();
121+
try (ArrowReader reader = partition.scanArrow(ALLOCATOR)) {
122+
while (reader.loadNextBatch()) {
123+
VectorSchemaRoot root = reader.getVectorSchemaRoot();
124+
Float8Vector vec = (Float8Vector) root.getVector(column);
125+
for (int i = 0; i < root.getRowCount(); i++) {
126+
doubles.add(vec.get(i));
127+
}
128+
}
129+
}
130+
}
131+
return doubles.stream().mapToDouble(Double::doubleValue).toArray();
70132
}
71133
}

0 commit comments

Comments
 (0)