Skip to content

Commit debf147

Browse files
dfa1claude
dfa1
and
claude
committed
refactor: type writeChunk map keys as ColumnName
VortexWriter.writeChunk(Map<String, Object>) became Map<ColumnName, Object>: the keys must equal schema.fieldNames(), so String lied about the domain. Now a wrong-order or wrong-name key is a compile error, and the map form matches the typed builder and the read path. Callers across csv/jdbc/parquet importers, calcite/writer tests, and performance benchmarks updated to key chunks by ColumnName. External boundaries (Calcite table names, JNI writer maps) stay String. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
1 parent aded670 commit debf147

37 files changed

Lines changed: 325 additions & 324 deletions

File tree

‎calcite/src/test/java/io/github/dfa1/vortex/calcite/AggregateSumNullTest.java‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -52,7 +52,7 @@ private SchemaPlus tableOf(long[] values, boolean[] valid) throws IOException {
5252
WriteOptions opts = new WriteOptions(1024, true, 0.90, 0, false, false);
5353
try (var ch = FileChannel.open(file, StandardOpenOption.CREATE, StandardOpenOption.WRITE);
5454
var writer = VortexWriter.create(ch, schema, opts)) {
55-
writer.writeChunk(Map.of("v", new NullableData(values, valid)));
55+
writer.writeChunk(Map.of(ColumnName.of("v"), new NullableData(values, valid)));
5656
}
5757
SchemaPlus root = Frameworks.createRootSchema(true);
5858
return root.add("vtx", new VortexSchema(Map.of("t", file)));

‎calcite/src/test/java/io/github/dfa1/vortex/calcite/AggregateWhereBoundaryTest.java‎

Lines changed: 17 additions & 17 deletions
Original file line numberDiff line numberDiff line change
@@ -68,7 +68,7 @@ static void write() throws Exception {
6868
id[i] = v;
6969
val[i] = v;
7070
}
71-
writer.writeChunk(Map.of("id", id, "val", val));
71+
writer.writeChunk(Map.of(ColumnName.of("id"), id, ColumnName.of("val"), val));
7272
}
7373
}
7474
}
@@ -207,10 +207,10 @@ void nullableAggColumnAcrossBoundaryChunk() throws Exception {
207207
false);
208208
Path f = tmp.resolve("nullable-boundary.vortex");
209209
writeChunks(f, schema,
210-
Map.of("id", new long[]{0, 1, 2, 3},
211-
"val", new NullableData(new long[]{10, 0, 30, 0}, new boolean[]{true, false, true, false})),
212-
Map.of("id", new long[]{10, 11, 12, 13},
213-
"val", new NullableData(new long[]{1, 2, 3, 4}, new boolean[]{true, true, true, true})));
210+
Map.of(ColumnName.of("id"), new long[]{0, 1, 2, 3},
211+
ColumnName.of("val"), new NullableData(new long[]{10, 0, 30, 0}, new boolean[]{true, false, true, false})),
212+
Map.of(ColumnName.of("id"), new long[]{10, 11, 12, 13},
213+
ColumnName.of("val"), new NullableData(new long[]{1, 2, 3, 4}, new boolean[]{true, true, true, true})));
214214

215215
try (Connection conn = connect(f)) {
216216
// When the aggregates are taken — the boundary chunk is decoded and folded, no full scan
@@ -244,8 +244,8 @@ void nonNumericMinAcrossBoundaryAbandonsButScanIsCorrect() throws Exception {
244244
false);
245245
Path f = tmp.resolve("strmin-boundary.vortex");
246246
writeChunks(f, schema,
247-
Map.of("id", new long[]{0, 1, 2, 3}, "name", new String[]{"a", "b", "c", "d"}),
248-
Map.of("id", new long[]{10, 11, 12, 13}, "name", new String[]{"e", "f", "g", "h"}));
247+
Map.of(ColumnName.of("id"), new long[]{0, 1, 2, 3}, ColumnName.of("name"), new String[]{"a", "b", "c", "d"}),
248+
Map.of(ColumnName.of("id"), new long[]{10, 11, 12, 13}, ColumnName.of("name"), new String[]{"e", "f", "g", "h"}));
249249

250250
try (Connection conn = connect(f)) {
251251
// When MIN over the string column is taken across the boundary — the rewrite abandons, so
@@ -273,8 +273,8 @@ void unsignedKeyAcrossBoundaryAbandonsButScanIsCorrect() throws Exception {
273273
false);
274274
Path f = tmp.resolve("u64-boundary.vortex");
275275
writeChunks(f, schema,
276-
Map.of("id", new long[]{0, 1, 2, 3}, "val", new long[]{0, 1, 2, 3}),
277-
Map.of("id", new long[]{10, 11, 12, 13}, "val", new long[]{10, 11, 12, 13}));
276+
Map.of(ColumnName.of("id"), new long[]{0, 1, 2, 3}, ColumnName.of("val"), new long[]{0, 1, 2, 3}),
277+
Map.of(ColumnName.of("id"), new long[]{10, 11, 12, 13}, ColumnName.of("val"), new long[]{10, 11, 12, 13}));
278278

279279
try (Connection conn = connect(f)) {
280280
// When the aggregate spans the unsigned key across a boundary — a scan remains in the plan
@@ -340,10 +340,10 @@ void neqWithNullInFilterColumnAcrossBoundaryExcludesNullAndValue() throws Except
340340
false);
341341
Path f = tmp.resolve("neq-null-boundary.vortex");
342342
writeChunks(f, schema,
343-
Map.of("id", new long[]{0, 1, 2, 3},
344-
"amt", new NullableData(new long[]{10, 0, 30, 40}, new boolean[]{true, false, true, true})),
345-
Map.of("id", new long[]{10, 11, 12, 13},
346-
"amt", new NullableData(new long[]{50, 60, 70, 80}, new boolean[]{true, true, true, true})));
343+
Map.of(ColumnName.of("id"), new long[]{0, 1, 2, 3},
344+
ColumnName.of("amt"), new NullableData(new long[]{10, 0, 30, 40}, new boolean[]{true, false, true, true})),
345+
Map.of(ColumnName.of("id"), new long[]{10, 11, 12, 13},
346+
ColumnName.of("amt"), new NullableData(new long[]{50, 60, 70, 80}, new boolean[]{true, true, true, true})));
347347

348348
// Ground truth: reduce the same data, dropping the null row (UNKNOWN) and the amt == 30 row.
349349
long[] allAmt = {10, 0, 30, 40, 50, 60, 70, 80};
@@ -391,8 +391,8 @@ void floatFilterColumnAcrossBoundaryAbandonsButScanIsNaNCorrect() throws Excepti
391391
false);
392392
Path f = tmp.resolve("float-nan-boundary.vortex");
393393
writeChunks(f, schema,
394-
Map.of("f", new double[]{1.0, Double.NaN, 2.0, 3.0}, "val", new long[]{10, 20, 30, 40}),
395-
Map.of("f", new double[]{4.0, 5.0, 6.0, 7.0}, "val", new long[]{50, 60, 70, 80}));
394+
Map.of(ColumnName.of("f"), new double[]{1.0, Double.NaN, 2.0, 3.0}, ColumnName.of("val"), new long[]{10, 20, 30, 40}),
395+
Map.of(ColumnName.of("f"), new double[]{4.0, 5.0, 6.0, 7.0}, ColumnName.of("val"), new long[]{50, 60, 70, 80}));
396396

397397
// Ground truth: reduce the same data with SQL-NaN-correct `f >= 1.5` (NaN row excluded).
398398
double[] allF = {1.0, Double.NaN, 2.0, 3.0, 4.0, 5.0, 6.0, 7.0};
@@ -453,8 +453,8 @@ private static Ground reduce(java.util.function.LongPredicate predicate) {
453453
return new Ground(sum, count, min, max);
454454
}
455455

456-
private static void writeChunks(Path file, DType.Struct schema, Map<String, Object> chunk0,
457-
Map<String, Object> chunk1) throws Exception {
456+
private static void writeChunks(Path file, DType.Struct schema, Map<ColumnName, Object> chunk0,
457+
Map<ColumnName, Object> chunk1) throws Exception {
458458
// chunkSize large so each writeChunk is exactly one chunk (one zone).
459459
WriteOptions opts = new WriteOptions(1024, true, 0.90, 0, false, false);
460460
try (var ch = FileChannel.open(file, StandardOpenOption.CREATE, StandardOpenOption.WRITE);

‎calcite/src/test/java/io/github/dfa1/vortex/calcite/AggregateWhereCleanPartitionTest.java‎

Lines changed: 19 additions & 19 deletions
Original file line numberDiff line numberDiff line change
@@ -67,7 +67,7 @@ static void write() throws Exception {
6767
id[i] = v;
6868
val[i] = v;
6969
}
70-
writer.writeChunk(Map.of("id", id, "val", val));
70+
writer.writeChunk(Map.of(ColumnName.of("id"), id, ColumnName.of("val"), val));
7171
}
7272
}
7373
}
@@ -161,8 +161,8 @@ void narrowIntegerKeyFoldsFromStats() throws Exception {
161161
false);
162162
Path f = tmp.resolve("i32.vortex");
163163
writeChunks(f, schema,
164-
Map.of("id", new int[]{0, 1, 2, 3}, "val", new int[]{0, 1, 2, 3}),
165-
Map.of("id", new int[]{10, 11, 12, 13}, "val", new int[]{10, 11, 12, 13}));
164+
Map.of(ColumnName.of("id"), new int[]{0, 1, 2, 3}, ColumnName.of("val"), new int[]{0, 1, 2, 3}),
165+
Map.of(ColumnName.of("id"), new int[]{10, 11, 12, 13}, ColumnName.of("val"), new int[]{10, 11, 12, 13}));
166166

167167
try (Connection conn = connect(f)) {
168168
// When the first chunk is selected (id < 10), the aggregates fold from stats — no scan
@@ -191,10 +191,10 @@ void nullAwareSumAndCountOverKeptZones() throws Exception {
191191
false);
192192
Path f = tmp.resolve("nullable.vortex");
193193
writeChunks(f, schema,
194-
Map.of("id", new long[]{0, 1, 2, 3},
195-
"val", new NullableData(new long[]{10, 0, 30, 0}, new boolean[]{true, false, true, false})),
196-
Map.of("id", new long[]{10, 11, 12, 13},
197-
"val", new NullableData(new long[]{1, 2, 3, 4}, new boolean[]{true, true, true, true})));
194+
Map.of(ColumnName.of("id"), new long[]{0, 1, 2, 3},
195+
ColumnName.of("val"), new NullableData(new long[]{10, 0, 30, 0}, new boolean[]{true, false, true, false})),
196+
Map.of(ColumnName.of("id"), new long[]{10, 11, 12, 13},
197+
ColumnName.of("val"), new NullableData(new long[]{1, 2, 3, 4}, new boolean[]{true, true, true, true})));
198198

199199
try (Connection conn = connect(f)) {
200200
// When the first chunk is selected (id < 10) — folded from stats, no scan
@@ -223,8 +223,8 @@ void unsignedKeyColumnAbandonsToScan() throws Exception {
223223
false);
224224
Path f = tmp.resolve("u64.vortex");
225225
writeChunks(f, schema,
226-
Map.of("id", new long[]{0, 1, 2, 3}, "val", new long[]{0, 1, 2, 3}),
227-
Map.of("id", new long[]{10, 11, 12, 13}, "val", new long[]{10, 11, 12, 13}));
226+
Map.of(ColumnName.of("id"), new long[]{0, 1, 2, 3}, ColumnName.of("val"), new long[]{0, 1, 2, 3}),
227+
Map.of(ColumnName.of("id"), new long[]{10, 11, 12, 13}, ColumnName.of("val"), new long[]{10, 11, 12, 13}));
228228

229229
try (Connection conn = connect(f)) {
230230
// When id < 10 over the unsigned key — a scan is present (no stats fold)
@@ -307,10 +307,10 @@ void isNotNullBoundaryChunkFoldsViaDecode() throws Exception {
307307
false);
308308
Path f = tmp.resolve("isnotnull-boundary.vortex");
309309
writeChunks(f, schema,
310-
Map.of("id", new long[]{0, 1, 2, 3},
311-
"val", new NullableData(new long[]{5, 0, 7, 0}, new boolean[]{true, false, true, false})),
312-
Map.of("id", new long[]{10, 11, 12, 13},
313-
"val", new NullableData(new long[]{10, 20, 30, 40}, new boolean[]{true, true, true, true})));
310+
Map.of(ColumnName.of("id"), new long[]{0, 1, 2, 3},
311+
ColumnName.of("val"), new NullableData(new long[]{5, 0, 7, 0}, new boolean[]{true, false, true, false})),
312+
Map.of(ColumnName.of("id"), new long[]{10, 11, 12, 13},
313+
ColumnName.of("val"), new NullableData(new long[]{10, 20, 30, 40}, new boolean[]{true, true, true, true})));
314314

315315
try (Connection conn = connect(f)) {
316316
// When EXPLAIN runs, no scan is present — the boundary chunk was decoded and folded
@@ -419,15 +419,15 @@ private static Path nullPartitionedFile(String name) throws Exception {
419419
false);
420420
Path f = tmp.resolve(name);
421421
writeChunks(f, schema,
422-
Map.of("id", new long[]{0, 1, 2, 3},
423-
"val", new NullableData(new long[]{0, 0, 0, 0}, new boolean[]{false, false, false, false})),
424-
Map.of("id", new long[]{10, 11, 12, 13},
425-
"val", new NullableData(new long[]{10, 20, 30, 40}, new boolean[]{true, true, true, true})));
422+
Map.of(ColumnName.of("id"), new long[]{0, 1, 2, 3},
423+
ColumnName.of("val"), new NullableData(new long[]{0, 0, 0, 0}, new boolean[]{false, false, false, false})),
424+
Map.of(ColumnName.of("id"), new long[]{10, 11, 12, 13},
425+
ColumnName.of("val"), new NullableData(new long[]{10, 20, 30, 40}, new boolean[]{true, true, true, true})));
426426
return f;
427427
}
428428

429-
private static void writeChunks(Path file, DType.Struct schema, Map<String, Object> chunk0,
430-
Map<String, Object> chunk1) throws Exception {
429+
private static void writeChunks(Path file, DType.Struct schema, Map<ColumnName, Object> chunk0,
430+
Map<ColumnName, Object> chunk1) throws Exception {
431431
// chunkSize large so each writeChunk is exactly one chunk (one zone).
432432
WriteOptions opts = new WriteOptions(1024, true, 0.90, 0, false, false);
433433
try (var ch = FileChannel.open(file, StandardOpenOption.CREATE, StandardOpenOption.WRITE);

‎calcite/src/test/java/io/github/dfa1/vortex/calcite/FilterPushDownTest.java‎

Lines changed: 11 additions & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,6 @@
11
package io.github.dfa1.vortex.calcite;
22

3+
import io.github.dfa1.vortex.core.model.ColumnName;
34
import io.github.dfa1.vortex.core.model.DType;
45
import io.github.dfa1.vortex.writer.VortexWriter;
56
import io.github.dfa1.vortex.writer.WriteOptions;
@@ -47,17 +48,17 @@ static void write() throws Exception {
4748
try (var ch = FileChannel.open(file, StandardOpenOption.CREATE, StandardOpenOption.WRITE);
4849
var w = VortexWriter.create(ch, SCHEMA, WriteOptions.defaults())) {
4950
w.writeChunk(Map.of(
50-
"i64", new long[]{1000L, 2000L, 3000L},
51-
"i32", new int[]{100, 200, 300},
52-
"f64", new double[]{1.0, 2.0, 3.0},
53-
"s", new String[]{"a", "b", "c"},
54-
"b", new boolean[]{true, false, true}));
51+
ColumnName.of("i64"), new long[]{1000L, 2000L, 3000L},
52+
ColumnName.of("i32"), new int[]{100, 200, 300},
53+
ColumnName.of("f64"), new double[]{1.0, 2.0, 3.0},
54+
ColumnName.of("s"), new String[]{"a", "b", "c"},
55+
ColumnName.of("b"), new boolean[]{true, false, true}));
5556
w.writeChunk(Map.of(
56-
"i64", new long[]{4000L, 5000L, 6000L},
57-
"i32", new int[]{400, 500, 600},
58-
"f64", new double[]{4.0, 5.0, 6.0},
59-
"s", new String[]{"d", "e", "f"},
60-
"b", new boolean[]{false, true, false}));
57+
ColumnName.of("i64"), new long[]{4000L, 5000L, 6000L},
58+
ColumnName.of("i32"), new int[]{400, 500, 600},
59+
ColumnName.of("f64"), new double[]{4.0, 5.0, 6.0},
60+
ColumnName.of("s"), new String[]{"d", "e", "f"},
61+
ColumnName.of("b"), new boolean[]{false, true, false}));
6162
}
6263
}
6364

‎calcite/src/test/java/io/github/dfa1/vortex/calcite/OhlcGenerator.java‎

Lines changed: 9 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,6 @@
11
package io.github.dfa1.vortex.calcite;
22

3+
import io.github.dfa1.vortex.core.model.ColumnName;
34
import io.github.dfa1.vortex.core.testing.OhlcData;
45
import io.github.dfa1.vortex.writer.VortexWriter;
56
import io.github.dfa1.vortex.writer.WriteOptions;
@@ -23,14 +24,14 @@ static void write(Path file, int totalRows, int chunkSize) throws IOException {
2324
try (var ch = FileChannel.open(file, StandardOpenOption.CREATE, StandardOpenOption.WRITE);
2425
var writer = VortexWriter.create(ch, OhlcData.SCHEMA, opts)) {
2526
for (OhlcData.Batch batch : OhlcData.generate(totalRows, chunkSize)) {
26-
Map<String, Object> columns = new LinkedHashMap<>();
27-
columns.put("date", batch.date());
28-
columns.put("symbol", batch.symbol());
29-
columns.put("open", batch.open());
30-
columns.put("high", batch.high());
31-
columns.put("low", batch.low());
32-
columns.put("close", batch.close());
33-
columns.put("volume", batch.volume());
27+
Map<ColumnName, Object> columns = new LinkedHashMap<>();
28+
columns.put(ColumnName.of("date"), batch.date());
29+
columns.put(ColumnName.of("symbol"), batch.symbol());
30+
columns.put(ColumnName.of("open"), batch.open());
31+
columns.put(ColumnName.of("high"), batch.high());
32+
columns.put(ColumnName.of("low"), batch.low());
33+
columns.put(ColumnName.of("close"), batch.close());
34+
columns.put(ColumnName.of("volume"), batch.volume());
3435
writer.writeChunk(columns);
3536
}
3637
}

‎calcite/src/test/java/io/github/dfa1/vortex/calcite/VortexAdapterCoverageTest.java‎

Lines changed: 26 additions & 24 deletions
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,7 @@
11
package io.github.dfa1.vortex.calcite;
22

3+
import io.github.dfa1.vortex.core.model.ColumnName;
4+
35
import io.github.dfa1.vortex.core.model.DType;
46
import io.github.dfa1.vortex.reader.ReadRegistry;
57
import io.github.dfa1.vortex.reader.VortexReader;
@@ -56,18 +58,18 @@ static void write() throws Exception {
5658
try (var ch = FileChannel.open(file, StandardOpenOption.CREATE, StandardOpenOption.WRITE);
5759
var w = VortexWriter.create(ch, SCHEMA, WriteOptions.defaults())) {
5860
w.writeChunk(Map.ofEntries(
59-
Map.entry("i8", new byte[]{1, 2, 3}),
60-
Map.entry("i16", new short[]{10, 20, 30}),
61-
Map.entry("i32", new int[]{100, 200, 300}),
62-
Map.entry("i64", new long[]{1000L, 2000L, 3000L}),
63-
Map.entry("u8", new byte[]{4, 5, 6}),
64-
Map.entry("u16", new short[]{40, 50, 60}),
65-
Map.entry("u32", new int[]{400, 500, 600}),
66-
Map.entry("u64", new long[]{4000L, 5000L, 6000L}),
67-
Map.entry("f32", new float[]{1.5f, 2.5f, 3.5f}),
68-
Map.entry("f64", new double[]{1.25, 2.25, 3.25}),
69-
Map.entry("s", new String[]{"a", "b", "c"}),
70-
Map.entry("b", new boolean[]{true, false, true})));
61+
Map.entry(ColumnName.of("i8"), new byte[]{1, 2, 3}),
62+
Map.entry(ColumnName.of("i16"), new short[]{10, 20, 30}),
63+
Map.entry(ColumnName.of("i32"), new int[]{100, 200, 300}),
64+
Map.entry(ColumnName.of("i64"), new long[]{1000L, 2000L, 3000L}),
65+
Map.entry(ColumnName.of("u8"), new byte[]{4, 5, 6}),
66+
Map.entry(ColumnName.of("u16"), new short[]{40, 50, 60}),
67+
Map.entry(ColumnName.of("u32"), new int[]{400, 500, 600}),
68+
Map.entry(ColumnName.of("u64"), new long[]{4000L, 5000L, 6000L}),
69+
Map.entry(ColumnName.of("f32"), new float[]{1.5f, 2.5f, 3.5f}),
70+
Map.entry(ColumnName.of("f64"), new double[]{1.25, 2.25, 3.25}),
71+
Map.entry(ColumnName.of("s"), new String[]{"a", "b", "c"}),
72+
Map.entry(ColumnName.of("b"), new boolean[]{true, false, true})));
7173
}
7274
}
7375

@@ -278,18 +280,18 @@ void noZoneMap_sumFallsBackToFullScan(@TempDir Path noStats) throws Exception {
278280
try (var ch = FileChannel.open(bare, StandardOpenOption.CREATE, StandardOpenOption.WRITE);
279281
var w = VortexWriter.create(ch, SCHEMA, noZoneMaps)) {
280282
w.writeChunk(Map.ofEntries(
281-
Map.entry("i8", new byte[]{1, 2, 3}),
282-
Map.entry("i16", new short[]{10, 20, 30}),
283-
Map.entry("i32", new int[]{100, 200, 300}),
284-
Map.entry("i64", new long[]{1000L, 2000L, 3000L}),
285-
Map.entry("u8", new byte[]{4, 5, 6}),
286-
Map.entry("u16", new short[]{40, 50, 60}),
287-
Map.entry("u32", new int[]{400, 500, 600}),
288-
Map.entry("u64", new long[]{4000L, 5000L, 6000L}),
289-
Map.entry("f32", new float[]{1.5f, 2.5f, 3.5f}),
290-
Map.entry("f64", new double[]{1.25, 2.25, 3.25}),
291-
Map.entry("s", new String[]{"a", "b", "c"}),
292-
Map.entry("b", new boolean[]{true, false, true})));
283+
Map.entry(ColumnName.of("i8"), new byte[]{1, 2, 3}),
284+
Map.entry(ColumnName.of("i16"), new short[]{10, 20, 30}),
285+
Map.entry(ColumnName.of("i32"), new int[]{100, 200, 300}),
286+
Map.entry(ColumnName.of("i64"), new long[]{1000L, 2000L, 3000L}),
287+
Map.entry(ColumnName.of("u8"), new byte[]{4, 5, 6}),
288+
Map.entry(ColumnName.of("u16"), new short[]{40, 50, 60}),
289+
Map.entry(ColumnName.of("u32"), new int[]{400, 500, 600}),
290+
Map.entry(ColumnName.of("u64"), new long[]{4000L, 5000L, 6000L}),
291+
Map.entry(ColumnName.of("f32"), new float[]{1.5f, 2.5f, 3.5f}),
292+
Map.entry(ColumnName.of("f64"), new double[]{1.25, 2.25, 3.25}),
293+
Map.entry(ColumnName.of("s"), new String[]{"a", "b", "c"}),
294+
Map.entry(ColumnName.of("b"), new boolean[]{true, false, true})));
293295
}
294296

295297
// When

0 commit comments

Comments
 (0)