Skip to content

refactor: typed ScalarValue in encoder stats (#531) #536

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

Open
dfa1 wants to merge 3 commits into
base: main
Choose a base branch
from
Open

refactor: typed ScalarValue in encoder stats (#531) #536

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
1f31641
refactor: carry typed ScalarValue in encoder stats instead of seriali…
dfa1 Oct 13, 2026
bf5d930
refactor: type the zone-map sum stat as ProtoScalarValue too (#531)
dfa1 Oct 13, 2026
194ffeb
refactor: MinMax record replaces the two-element stats array (#531)
dfa1 Oct 13, 2026
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
26 changes: 14 additions & 12 deletions writer/src/main/java/io/github/dfa1/vortex/writer/DictColumnState.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.writer;

import io.github.dfa1.vortex.writer.encode.MinMax;
import io.github.dfa1.vortex.core.proto.ProtoScalarValue;
import io.github.dfa1.vortex.core.model.DType;
import io.github.dfa1.vortex.core.model.PType;
import io.github.dfa1.vortex.writer.encode.LongIntMap;
Expand Down Expand Up @@ -70,9 +72,9 @@ final class DictColumnState {
private final List<boolean[]> chunkValidity = new ArrayList<>();
private final List<Long> chunkRowCounts = new ArrayList<>();
private final List<Long> chunkNullCounts = new ArrayList<>();
private final List<byte[]> chunkStatsMin = new ArrayList<>();
private final List<byte[]> chunkStatsMax = new ArrayList<>();
private final List<byte[]> chunkStatsSum = new ArrayList<>();
private final List<ProtoScalarValue> chunkStatsMin = new ArrayList<>();
private final List<ProtoScalarValue> chunkStatsMax = new ArrayList<>();
private final List<ProtoScalarValue> chunkStatsSum = new ArrayList<>();
private long codeArrayBytes;

DictColumnState(DType dtype) {
Expand Down Expand Up @@ -138,15 +140,15 @@ List<Long> chunkNullCounts() {
return chunkNullCounts;
}

List<byte[]> chunkStatsMin() {
List<ProtoScalarValue> chunkStatsMin() {
return chunkStatsMin;
}

List<byte[]> chunkStatsMax() {
List<ProtoScalarValue> chunkStatsMax() {
return chunkStatsMax;
}

List<byte[]> chunkStatsSum() {
List<ProtoScalarValue> chunkStatsSum() {
return chunkStatsSum;
}

Expand Down Expand Up @@ -288,14 +290,14 @@ boolean ingestDictChunk(Object data) {
chunkStatsMax.add(null);
chunkStatsSum.add(null);
} else if (varBin) {
byte[][] mm = VarBinEncodingEncoder.minMaxStats((String[]) values);
chunkStatsMin.add(mm != null ? mm[0] : null);
chunkStatsMax.add(mm != null ? mm[1] : null);
MinMax mm = VarBinEncodingEncoder.minMaxStats((String[]) values);
chunkStatsMin.add(mm != null ? mm.min() : null);
chunkStatsMax.add(mm != null ? mm.max() : null);
chunkStatsSum.add(null);
} else {
byte[][] mm = PrimitiveEncodingEncoder.minMaxStats(ptype, values);
chunkStatsMin.add(mm != null ? mm[0] : null);
chunkStatsMax.add(mm != null ? mm[1] : null);
MinMax mm = PrimitiveEncodingEncoder.minMaxStats(ptype, values);
chunkStatsMin.add(mm != null ? mm.min() : null);
chunkStatsMax.add(mm != null ? mm.max() : null);
chunkStatsSum.add(PrimitiveEncodingEncoder.sumStat(ptype, values));
}
codeArrayBytes += 2L * len;
Expand Down
25 changes: 13 additions & 12 deletions writer/src/main/java/io/github/dfa1/vortex/writer/VortexWriter.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.writer;

import io.github.dfa1.vortex.writer.encode.MinMax;
import io.github.dfa1.vortex.core.proto.ProtoScalarValue;
import io.github.dfa1.vortex.core.fbs.FbsBuilder;
import io.github.dfa1.vortex.writer.encode.DateTimePartsData;
import io.github.dfa1.vortex.writer.encode.FixedSizeListData;
Expand Down Expand Up @@ -853,15 +855,15 @@ private EncodedSegment encodeSegment(DType dtype, Object data, EncodingEncoder e
// fallback computes them from the untouched input, independent of which encoder ran.
// This is what makes stats coverage an encoder-independent guarantee rather than a
// per-encoder convention every new encoder has to remember (ADR 0025).
byte[] min;
byte[] max;
ProtoScalarValue min;
ProtoScalarValue max;
if (result.hasStats()) {
min = result.statsMin();
max = result.statsMax();
} else {
byte[][] fallback = ZoneMapStatCodec.columnMinMax(dtype, data);
min = fallback != null ? fallback[0] : null;
max = fallback != null ? fallback[1] : null;
MinMax fallback = ZoneMapStatCodec.columnMinMax(dtype, data);
min = fallback != null ? fallback.min() : null;
max = fallback != null ? fallback.max() : null;
}
SegmentStats stats = new SegmentStats(min, max, ZoneMapStatCodec.columnSum(dtype, data), nullCount);
return new EncodedSegment(result, arena, stats);
Expand Down Expand Up @@ -1005,9 +1007,9 @@ private ByteBuffer buildArrayFlatBuffer(EncodeResult result, int[] paddings, lon
// (Rust's flat writer doesn't either — flat/writer.rs retains only pre-computed stats); the
// per-zone sum lives in the vortex.stats zone-map table emitted by flushZoneMaps().
int minVec = result.hasStats()
? io.github.dfa1.vortex.core.fbs.FbsArrayStats.createMinVector(fbb, result.statsMin()) : 0;
? io.github.dfa1.vortex.core.fbs.FbsArrayStats.createMinVector(fbb, result.statsMin().encode()) : 0;
int maxVec = result.hasStats()
? io.github.dfa1.vortex.core.fbs.FbsArrayStats.createMaxVector(fbb, result.statsMax()) : 0;
? io.github.dfa1.vortex.core.fbs.FbsArrayStats.createMaxVector(fbb, result.statsMax().encode()) : 0;
// forceDefaults only while building ArrayStats, so null_count = 0 is serialized (flatbuffers
// omits a scalar equal to its default otherwise) — matching the Rust writer and letting the
// reader prune IS NULL on zero-null chunks. Reset immediately so the Array/ArrayNode tables
Expand Down Expand Up @@ -1142,8 +1144,8 @@ private DType columnDtype(ColumnName colName) {
/// min/max) rather than dropping the stat for the whole column — MIN/MAX/SUM are nullable per
/// zone, matching Rust. Field/bit order follows ZonedStatsSchema: MAX(3), MIN(4), SUM(5),
/// NULL_COUNT(6).
private void emitZoneMap(ColumnName colName, DType minMaxDtype, List<byte[]> minBytes, List<byte[]> maxBytes,
DType sumDtype, List<byte[]> sumBytes, long[] nullCounts,
private void emitZoneMap(ColumnName colName, DType minMaxDtype, List<ProtoScalarValue> minBytes, List<ProtoScalarValue> maxBytes,
DType sumDtype, List<ProtoScalarValue> sumBytes, long[] nullCounts,
long[] rowCounts) throws IOException {
int nZones = nullCounts.length;
boolean[] allValid = new boolean[nZones];
Expand Down Expand Up @@ -1548,8 +1550,7 @@ private record SegRef(long offset, long len, SegmentStats stats) {

// S6218: the byte[] stat components are never value-compared — SegmentStats instances are only
// read positionally, so the default identity equals is fine.
@SuppressWarnings("java:S6218")
private record SegmentStats(byte[] min, byte[] max, byte[] sum, long nullCount) {
private record SegmentStats(ProtoScalarValue min, ProtoScalarValue max, ProtoScalarValue sum, long nullCount) {
}

private record EncodedSegment(EncodeResult result, Arena arena, SegmentStats stats) {
Expand All @@ -1568,7 +1569,7 @@ private record ZoneMapRef(int zonesSegIdx, long nZones, byte[] metadata) {
/// @param nullableCodes whether the codes child carries validity (Rust's `is_nullable_codes`)
private record DictColRef(int valuesSegIdx, long valuesLen, List<Integer> codesSegIdxes,
List<Long> codesRowCounts, List<Long> chunkRowCounts, List<Long> chunkNullCounts,
List<byte[]> chunkStatsMin, List<byte[]> chunkStatsMax, List<byte[]> chunkStatsSum,
List<ProtoScalarValue> chunkStatsMin, List<ProtoScalarValue> chunkStatsMax, List<ProtoScalarValue> chunkStatsSum,
boolean nullableCodes) {
long totalRows() {
return chunkRowCounts.stream().mapToLong(Long::longValue).sum();
Expand Down
35 changes: 15 additions & 20 deletions writer/src/main/java/io/github/dfa1/vortex/writer/ZoneMapStatCodec.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.writer;

import io.github.dfa1.vortex.writer.encode.MinMax;
import io.github.dfa1.vortex.core.compute.PrimitiveArrays;
import io.github.dfa1.vortex.core.model.DType;
import io.github.dfa1.vortex.core.model.PType;
Expand All @@ -10,14 +11,13 @@
import io.github.dfa1.vortex.writer.encode.VarBinEncodingEncoder;

import java.io.IOException;
import java.lang.foreign.MemorySegment;
import java.nio.ByteBuffer;
import java.nio.ByteOrder;
import java.util.List;

/// Pure helpers for `vortex.stats` zone-map stats (ADR): deriving the per-zone min/max/sum dtype
/// for a column, encoding a zone-map layout's `vortex.stats` metadata bytes, and decoding each
/// zone's serialized [ProtoScalarValue] stat back into a typed array the zone-map's struct
/// zone's serialized [ProtoScalarValue] stat back into a typed array the zone-map's struc
/// encoder expects. Matches [io.github.dfa1.vortex.reader]'s `ZonedStatsSchema`.
final class ZoneMapStatCodec {

Expand All @@ -31,7 +31,7 @@ private ZoneMapStatCodec() {
}

/// The zone length to declare for a one-zone-per-chunk zone map over chunks of `rowCounts`
/// rows: the shared chunk length when every chunk but the last has it (and the last is not
/// rows: the shared chunk length when every chunk but the last has it (and the last is no
/// empty and no longer), else `0`.
///
/// Rust reads the declared length as a uniform stride - row `r` belongs to zone
Expand Down Expand Up @@ -109,7 +109,7 @@ static DType zoneSumDtype(DType dtype) {
/// The serialized per-chunk SUM scalar for `data` of logical type `dtype`, or `null` when the
/// column is not summable (non-primitive) or the sum overflowed. Validity placeholders are zero
/// and therefore sum-neutral, so a nullable carrier sums correctly via its values.
static byte[] columnSum(DType dtype, Object data) {
static ProtoScalarValue columnSum(DType dtype, Object data) {
if (!(dtype instanceof DType.Primitive p)) {
return null;
}
Expand All @@ -120,7 +120,7 @@ static byte[] columnSum(DType dtype, Object data) {
/// The serialized `{min, max}` pair for `data` of logical type `dtype`, or `null` when the
/// column has no recordable min/max ([#zoneMinMaxDtype] returns `null` for the same `dtype`
/// shapes). This is the generic fallback every column gets regardless of which
/// [io.github.dfa1.vortex.writer.encode.EncodingEncoder] wrote it — mirroring the Rust
/// [io.github.dfa1.vortex.writer.encode.EncodingEncoder] wrote it — mirroring the Rus
/// reference, which computes stats through one generic reduction for every encoding rather
/// than trusting each encoding to report its own (see [ComparableValues]'s Rust-parity note).
/// `VortexWriter#writeSegment` (private) only calls this when the winning encoder's own
Expand All @@ -138,7 +138,7 @@ static byte[] columnSum(DType dtype, Object data) {
/// @param dtype the segment's logical type
/// @param data the segment's input data, possibly [NullableData]- or [ComparableValues]-wrapped
/// @return a `{min, max}` pair, or `null` when `dtype` has no recordable min/max
static byte[][] columnMinMax(DType dtype, Object data) {
static MinMax columnMinMax(DType dtype, Object data) {
if (data instanceof ComparableValues cv) {
return PrimitiveEncodingEncoder.minMaxStats(cv.ptype(), cv.values());
}
Expand Down Expand Up @@ -167,7 +167,7 @@ private static Object compactIfNullable(PType ptype, Object data) {
/// or an encoder that does not compute one) marks that zone invalid in `valid` rather than
/// failing the whole column's zone-map — matching Rust, which stores min/max as a nullable
/// per-zone stat.
static Object zoneStatValues(DType minMaxDtype, List<byte[]> statBytes, boolean[] valid) throws IOException {
static Object zoneStatValues(DType minMaxDtype, List<ProtoScalarValue> statBytes, boolean[] valid) throws IOException {
return switch (minMaxDtype) {
case DType.Primitive p -> statColumn(p.ptype(), statBytes, valid);
case DType.Utf8 _ -> statStringColumn(statBytes, valid);
Expand All @@ -178,7 +178,7 @@ static Object zoneStatValues(DType minMaxDtype, List<byte[]> statBytes, boolean[
/// Builds the per-zone SUM array for `sumDtype` (i64/u64 → `long[]`, f64 → `double[]`), decoding
/// each zone's serialized scalar. Zones whose sum overflowed carry a `null` entry in `sumBytes`;
/// `valid[i]` is set accordingly so the stat field reports them as null.
static Object sumColumn(DType sumDtype, List<byte[]> sumBytes, boolean[] valid) throws IOException {
static Object sumColumn(DType sumDtype, List<ProtoScalarValue> sumBytes, boolean[] valid) throws IOException {
PType ptype = ((DType.Primitive) sumDtype).ptype();
int n = sumBytes.size();
if (ptype == PType.F64) {
Expand All @@ -202,19 +202,19 @@ static Object sumColumn(DType sumDtype, List<byte[]> sumBytes, boolean[] valid)
/// A `null` entry in `statBytes` sets `valid[i]` to `false` and fills the slot with `""` (a
/// sum-neutral-style placeholder never read back, since [NullableData] carries validity
/// separately from values).
private static String[] statStringColumn(List<byte[]> statBytes, boolean[] valid) throws IOException {
private static String[] statStringColumn(List<ProtoScalarValue> statBytes, boolean[] valid) throws IOException {
String[] out = new String[statBytes.size()];
for (int i = 0; i < out.length; i++) {
valid[i] = statBytes.get(i) != null;
out[i] = valid[i] ? decodeScalar(statBytes.get(i)).string_value() : "";
out[i] = valid[i] ? statBytes.get(i).string_value() : "";
}
return out;
}

/// Builds the per-zone values array in the storage shape the primitive encoder expects, decoding
/// each zone's serialized [ProtoScalarValue] stat. A `null` entry in `statBytes` sets `valid[i]`
/// to `false` and fills the slot with a zero placeholder, never read back.
private static Object statColumn(PType ptype, List<byte[]> statBytes, boolean[] valid) throws IOException {
private static Object statColumn(PType ptype, List<ProtoScalarValue> statBytes, boolean[] valid) throws IOException {
int n = statBytes.size();
return switch (ptype) {
case I8, U8 -> {
Expand Down Expand Up @@ -277,16 +277,16 @@ private static Object statColumn(PType ptype, List<byte[]> statBytes, boolean[]
};
}

private static long scalarLong(byte[] bytes) throws IOException {
private static long scalarLong(ProtoScalarValue sv) {
// Integer columns serialize min/max as int64 (signed) or uint64 (unsigned).
ProtoScalarValue sv = decodeScalar(bytes);

return sv.int64_value() != null ? sv.int64_value() : sv.uint64_value();
}

private static double scalarDouble(byte[] bytes) throws IOException {
private static double scalarDouble(ProtoScalarValue sv) {
// Float columns serialize min/max as f64 (F64), f32 (F32) or f16 (F16, the half bits).
// Branch rather than use a ternary so each path widens explicitly instead of mixing boxed types.
ProtoScalarValue sv = decodeScalar(bytes);

if (sv.f64_value() != null) {
return sv.f64_value();
}
Expand All @@ -295,9 +295,4 @@ private static double scalarDouble(byte[] bytes) throws IOException {
}
return sv.f32_value();
}

private static ProtoScalarValue decodeScalar(byte[] bytes) throws IOException {
MemorySegment seg = MemorySegment.ofArray(bytes);
return ProtoScalarValue.decode(seg, 0, seg.byteSize());
}
}
18 changes: 9 additions & 9 deletions writer/src/main/java/io/github/dfa1/vortex/writer/encode/AlpEncodingEncoder.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -216,8 +216,8 @@ private static AlpF64Data computeF64(double[] values) {
}
}

byte[] statsMin = n > 0 ? scalarF64(min) : null;
byte[] statsMax = n > 0 ? scalarF64(max) : null;
ProtoScalarValue statsMin = n > 0 ? scalarF64(min) : null;
ProtoScalarValue statsMax = n > 0 ? scalarF64(max) : null;
return new AlpF64Data(expE, expF, encodedArr, patchIndices, patchValues, statsMin, statsMax);
}

Expand Down Expand Up @@ -401,8 +401,8 @@ private static EncodeResult encodeF32(float[] values, EncodeContext ctx) {
}
}

byte[] statsMin = n > 0 ? scalarF32(min) : null;
byte[] statsMax = n > 0 ? scalarF32(max) : null;
ProtoScalarValue statsMin = n > 0 ? scalarF32(min) : null;
ProtoScalarValue statsMax = n > 0 ? scalarF32(max) : null;

MemorySegment encodedBuf = ctx.arena().allocate((long) n * 4, 4);
for (int i = 0; i < n; i++) {
Expand Down Expand Up @@ -440,17 +440,17 @@ private static EncodeResult encodeF32(float[] values, EncodeContext ctx) {
EncodedBuffer.of(patchBuffers.offsets(), patchBuffers.offsetType())), statsMin, statsMax);
}

private static byte[] scalarF64(double v) {
return ProtoScalarValue.ofF64Value(v).encode();
private static ProtoScalarValue scalarF64(double v) {
return ProtoScalarValue.ofF64Value(v);
}

private static byte[] scalarF32(float v) {
return ProtoScalarValue.ofF32Value(v).encode();
private static ProtoScalarValue scalarF32(float v) {
return ProtoScalarValue.ofF32Value(v);
}

@SuppressWarnings("java:S6218") // internal data carrier; record components are arrays of immutable primitives or refs that flow through pipelines without ever being compared.
private record AlpF64Data(int expE, int expF, long[] encodedArr,
List<Integer> patchIndices, List<Double> patchValues,
byte[] statsMin, byte[] statsMax) {
ProtoScalarValue statsMin, ProtoScalarValue statsMax) {
}
}
10 changes: 5 additions & 5 deletions writer/src/main/java/io/github/dfa1/vortex/writer/encode/AlpRdEncodingEncoder.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -295,7 +295,7 @@ private static Map<Short, Short> buildLookup(short[] dict) {
private static EncodeResult buildEncodeResult(
short[] dict, int rightBitWidth,
short[] leftCodes, Object rightPartsData, DType rightDtype,
List<Long> excPos, List<Short> excVals, byte[] statsMin, byte[] statsMax, EncodeContext ctx) {
List<Long> excPos, List<Short> excVals, ProtoScalarValue statsMin, ProtoScalarValue statsMax, EncodeContext ctx) {

EncodingEncoder bp = ctx.lookupEncoder(EncodingId.FASTLANES_BITPACKED);
EncodeResult leftResult = bp.encode(DType.U16, leftCodes, ctx);
Expand Down Expand Up @@ -366,12 +366,12 @@ private static EncodeResult buildEncodeResult(
return new EncodeResult(root, List.copyOf(allBuffers), statsMin, statsMax);
}

private static byte[] scalarF64(double v) {
return ProtoScalarValue.ofF64Value(v).encode();
private static ProtoScalarValue scalarF64(double v) {
return ProtoScalarValue.ofF64Value(v);
}

private static byte[] scalarF32(float v) {
return ProtoScalarValue.ofF32Value(v).encode();
private static ProtoScalarValue scalarF32(float v) {
return ProtoScalarValue.ofF32Value(v);
}

private static EncodeResult emptyResult(DType rightDtype, EncodeContext ctx) {
Expand Down
Loading
Loading