Skip to content

refactor: type writeChunk map keys as ColumnName #201

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
dfa1 merged 1 commit into from
Jul 6, 2026
Merged

refactor: type writeChunk map keys as ColumnName #201

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion calcite/src/test/java/io/github/dfa1/vortex/calcite/AggregateSumNullTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -52,7 +52,7 @@ private SchemaPlus tableOf(long[] values, boolean[] valid) throws IOException {
WriteOptions opts = new WriteOptions(1024, true, 0.90, 0, false, false);
try (var ch = FileChannel.open(file, StandardOpenOption.CREATE, StandardOpenOption.WRITE);
var writer = VortexWriter.create(ch, schema, opts)) {
writer.writeChunk(Map.of("v", new NullableData(values, valid)));
writer.writeChunk(Map.of(ColumnName.of("v"), new NullableData(values, valid)));
}
SchemaPlus root = Frameworks.createRootSchema(true);
return root.add("vtx", new VortexSchema(Map.of("t", file)));
Expand Down
34 changes: 17 additions & 17 deletions calcite/src/test/java/io/github/dfa1/vortex/calcite/AggregateWhereBoundaryTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -68,7 +68,7 @@ static void write() throws Exception {
id[i] = v;
val[i] = v;
}
writer.writeChunk(Map.of("id", id, "val", val));
writer.writeChunk(Map.of(ColumnName.of("id"), id, ColumnName.of("val"), val));
}
}
}
Expand Down Expand Up @@ -207,10 +207,10 @@ void nullableAggColumnAcrossBoundaryChunk() throws Exception {
false);
Path f = tmp.resolve("nullable-boundary.vortex");
writeChunks(f, schema,
Map.of("id", new long[]{0, 1, 2, 3},
"val", new NullableData(new long[]{10, 0, 30, 0}, new boolean[]{true, false, true, false})),
Map.of("id", new long[]{10, 11, 12, 13},
"val", new NullableData(new long[]{1, 2, 3, 4}, new boolean[]{true, true, true, true})));
Map.of(ColumnName.of("id"), new long[]{0, 1, 2, 3},
ColumnName.of("val"), new NullableData(new long[]{10, 0, 30, 0}, new boolean[]{true, false, true, false})),
Map.of(ColumnName.of("id"), new long[]{10, 11, 12, 13},
ColumnName.of("val"), new NullableData(new long[]{1, 2, 3, 4}, new boolean[]{true, true, true, true})));

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

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

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

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

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

private static void writeChunks(Path file, DType.Struct schema, Map<String, Object> chunk0,
Map<String, Object> chunk1) throws Exception {
private static void writeChunks(Path file, DType.Struct schema, Map<ColumnName, Object> chunk0,
Map<ColumnName, Object> chunk1) throws Exception {
// chunkSize large so each writeChunk is exactly one chunk (one zone).
WriteOptions opts = new WriteOptions(1024, true, 0.90, 0, false, false);
try (var ch = FileChannel.open(file, StandardOpenOption.CREATE, StandardOpenOption.WRITE);
Expand Down
38 changes: 19 additions & 19 deletions calcite/src/test/java/io/github/dfa1/vortex/calcite/AggregateWhereCleanPartitionTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -67,7 +67,7 @@ static void write() throws Exception {
id[i] = v;
val[i] = v;
}
writer.writeChunk(Map.of("id", id, "val", val));
writer.writeChunk(Map.of(ColumnName.of("id"), id, ColumnName.of("val"), val));
}
}
}
Expand Down Expand Up @@ -161,8 +161,8 @@ void narrowIntegerKeyFoldsFromStats() throws Exception {
false);
Path f = tmp.resolve("i32.vortex");
writeChunks(f, schema,
Map.of("id", new int[]{0, 1, 2, 3}, "val", new int[]{0, 1, 2, 3}),
Map.of("id", new int[]{10, 11, 12, 13}, "val", new int[]{10, 11, 12, 13}));
Map.of(ColumnName.of("id"), new int[]{0, 1, 2, 3}, ColumnName.of("val"), new int[]{0, 1, 2, 3}),
Map.of(ColumnName.of("id"), new int[]{10, 11, 12, 13}, ColumnName.of("val"), new int[]{10, 11, 12, 13}));

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

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

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

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

private static void writeChunks(Path file, DType.Struct schema, Map<String, Object> chunk0,
Map<String, Object> chunk1) throws Exception {
private static void writeChunks(Path file, DType.Struct schema, Map<ColumnName, Object> chunk0,
Map<ColumnName, Object> chunk1) throws Exception {
// chunkSize large so each writeChunk is exactly one chunk (one zone).
WriteOptions opts = new WriteOptions(1024, true, 0.90, 0, false, false);
try (var ch = FileChannel.open(file, StandardOpenOption.CREATE, StandardOpenOption.WRITE);
Expand Down
21 changes: 11 additions & 10 deletions calcite/src/test/java/io/github/dfa1/vortex/calcite/FilterPushDownTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,5 +1,6 @@
package io.github.dfa1.vortex.calcite;

import io.github.dfa1.vortex.core.model.ColumnName;
import io.github.dfa1.vortex.core.model.DType;
import io.github.dfa1.vortex.writer.VortexWriter;
import io.github.dfa1.vortex.writer.WriteOptions;
Expand Down Expand Up @@ -47,17 +48,17 @@ static void write() throws Exception {
try (var ch = FileChannel.open(file, StandardOpenOption.CREATE, StandardOpenOption.WRITE);
var w = VortexWriter.create(ch, SCHEMA, WriteOptions.defaults())) {
w.writeChunk(Map.of(
"i64", new long[]{1000L, 2000L, 3000L},
"i32", new int[]{100, 200, 300},
"f64", new double[]{1.0, 2.0, 3.0},
"s", new String[]{"a", "b", "c"},
"b", new boolean[]{true, false, true}));
ColumnName.of("i64"), new long[]{1000L, 2000L, 3000L},
ColumnName.of("i32"), new int[]{100, 200, 300},
ColumnName.of("f64"), new double[]{1.0, 2.0, 3.0},
ColumnName.of("s"), new String[]{"a", "b", "c"},
ColumnName.of("b"), new boolean[]{true, false, true}));
w.writeChunk(Map.of(
"i64", new long[]{4000L, 5000L, 6000L},
"i32", new int[]{400, 500, 600},
"f64", new double[]{4.0, 5.0, 6.0},
"s", new String[]{"d", "e", "f"},
"b", new boolean[]{false, true, false}));
ColumnName.of("i64"), new long[]{4000L, 5000L, 6000L},
ColumnName.of("i32"), new int[]{400, 500, 600},
ColumnName.of("f64"), new double[]{4.0, 5.0, 6.0},
ColumnName.of("s"), new String[]{"d", "e", "f"},
ColumnName.of("b"), new boolean[]{false, true, false}));
}
}

Expand Down
17 changes: 9 additions & 8 deletions calcite/src/test/java/io/github/dfa1/vortex/calcite/OhlcGenerator.java
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,5 +1,6 @@
package io.github.dfa1.vortex.calcite;

import io.github.dfa1.vortex.core.model.ColumnName;
import io.github.dfa1.vortex.core.testing.OhlcData;
import io.github.dfa1.vortex.writer.VortexWriter;
import io.github.dfa1.vortex.writer.WriteOptions;
Expand All @@ -23,14 +24,14 @@ static void write(Path file, int totalRows, int chunkSize) throws IOException {
try (var ch = FileChannel.open(file, StandardOpenOption.CREATE, StandardOpenOption.WRITE);
var writer = VortexWriter.create(ch, OhlcData.SCHEMA, opts)) {
for (OhlcData.Batch batch : OhlcData.generate(totalRows, chunkSize)) {
Map<String, Object> columns = new LinkedHashMap<>();
columns.put("date", batch.date());
columns.put("symbol", batch.symbol());
columns.put("open", batch.open());
columns.put("high", batch.high());
columns.put("low", batch.low());
columns.put("close", batch.close());
columns.put("volume", batch.volume());
Map<ColumnName, Object> columns = new LinkedHashMap<>();
columns.put(ColumnName.of("date"), batch.date());
columns.put(ColumnName.of("symbol"), batch.symbol());
columns.put(ColumnName.of("open"), batch.open());
columns.put(ColumnName.of("high"), batch.high());
columns.put(ColumnName.of("low"), batch.low());
columns.put(ColumnName.of("close"), batch.close());
columns.put(ColumnName.of("volume"), batch.volume());
writer.writeChunk(columns);
}
}
Expand Down
50 changes: 26 additions & 24 deletions calcite/src/test/java/io/github/dfa1/vortex/calcite/VortexAdapterCoverageTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,5 +1,7 @@
package io.github.dfa1.vortex.calcite;

import io.github.dfa1.vortex.core.model.ColumnName;

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

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

// When
Expand Down
Loading
Loading