Skip to content

perf: cheaper cascading writes (ALP-RD, string min/max, float stats) #543

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 3 commits into from
Oct 17, 2026
Merged

perf: cheaper cascading writes (ALP-RD, string min/max, float stats) #543

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
ab84ebd
perf: ALP-RD encoder without boxed maps and lists
dfa1 Oct 16, 2026
ea8cfa8
perf: compare each dictionary code once per chunk for string min/max
dfa1 Oct 16, 2026
a48e11b
perf: count float stats once per run of equal values
dfa1 Oct 16, 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
3 changes: 3 additions & 0 deletions CHANGELOG.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -7,6 +7,9 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0

## [Unreleased]

### Changed
- Cascading writes allocate about 13% less and run faster, from cheaper ALP-RD encoding, dictionary string statistics and float statistics. ([#543](https://github.com/dfa1/vortex-java/pull/543))

## [0.17.1] — 2026-10-11

### Added
Expand Down
33 changes: 32 additions & 1 deletion 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,6 @@
package io.github.dfa1.vortex.writer;

import io.github.dfa1.vortex.core.compute.Utf8Order;
import io.github.dfa1.vortex.core.model.DType;
import io.github.dfa1.vortex.core.model.PType;
import io.github.dfa1.vortex.core.proto.ProtoScalarValue;
Expand All @@ -11,6 +12,7 @@

import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.HashSet;
import java.util.LinkedHashMap;
import java.util.List;
Expand Down Expand Up @@ -70,6 +72,9 @@ final class DictColumnState {
// One code array per ingested chunk (null slots hold nullCode).
private final List<short[]> chunkCodes = new ArrayList<>();
private final List<boolean[]> chunkValidity = new ArrayList<>();
// Per code, the stamp of the last chunk whose min/max already compared it; avoids clearing a flag array per chunk.
private int[] seenInChunk;
private int stamp;
private final List<Long> chunkRowCounts = new ArrayList<>();
private final List<Long> chunkNullCounts = new ArrayList<>();
private final List<ProtoScalarValue> chunkStatsMin = new ArrayList<>();
Expand Down Expand Up @@ -227,6 +232,13 @@ boolean ingestDictChunk(Object data) {
// earlier shape ran a non-mutating counting pass first to get that atomicity, which cost a
// second boxed map probe for every row in the file.
short[] codes = new short[len];
boolean statsStrings = strings != null && !binary;
String minString = null;
String maxString = null;
int chunkStamp = ++stamp;
if (statsStrings && seenInChunk == null) {
seenInChunk = new int[16];
}
if (strings != null) {
for (int i = 0; i < len; i++) {
// Nullable Utf8/Binary keeps a real null at invalid positions (ChunkImpl.adaptUtf8,
Expand All @@ -250,6 +262,25 @@ boolean ingestDictChunk(Object data) {
valueToCode.put(v, code);
}
codes[i] = code.shortValue();
// Min/max depend on the distinct values only, so compare each code once per
// chunk instead of twice per row (UTF-8 order compare was 4% of a cascading write).
if (statsStrings) {
if (code >= seenInChunk.length) {
// Grown with the dictionary, not sized to the cap: a low-cardinality column
// must not hold 128 KiB the retained-bytes budget cannot see.
seenInChunk = Arrays.copyOf(seenInChunk, Math.max(code + 1, seenInChunk.length * 2));
}
}
if (statsStrings && seenInChunk[code] != chunkStamp) {
seenInChunk[code] = chunkStamp;
String s = (String) strings[i];
if (minString == null || Utf8Order.compare(s, minString) < 0) {
minString = s;
}
if (maxString == null || Utf8Order.compare(s, maxString) > 0) {
maxString = s;
}
}
}
} else {
// Primitive path: probe the unboxed bits index and box only when inserting a value the
Expand Down Expand Up @@ -290,7 +321,7 @@ boolean ingestDictChunk(Object data) {
chunkStatsMax.add(null);
chunkStatsSum.add(null);
} else if (varBin) {
MinMax mm = VarBinEncodingEncoder.minMaxStats((String[]) values);
MinMax mm = VarBinEncodingEncoder.encodeMinMax(minString, maxString);
chunkStatsMin.add(mm != null ? mm.min() : null);
chunkStatsMax.add(mm != null ? mm.max() : null);
chunkStatsSum.add(null);
Expand Down
141 changes: 81 additions & 60 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 @@ -11,9 +11,8 @@

import java.lang.foreign.MemorySegment;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.Arrays;
import java.util.List;
import java.util.Map;

/// Write-only encoder for `vortex.alprd`.
public final class AlpRdEncodingEncoder implements EncodingEncoder {
Expand Down Expand Up @@ -61,13 +60,11 @@ private static EncodeResult encodeF64(double[] values, EncodeContext ctx) {
int sampleLen = sample.length;
Dictionary64 best = findBestDictionaryF64(sample, sampleLen);

Map<Short, Short> lookup = buildLookup(best.dict);
long rightMask = -1L >>> (64 - best.rightBitWidth);

short[] leftCodes = new short[n];
long[] rightParts = new long[n];
List<Long> excPos = new ArrayList<>();
List<Short> excVals = new ArrayList<>();
Exceptions exceptions = new Exceptions();

double min = Double.MAX_VALUE;
double max = -Double.MAX_VALUE;
Expand All @@ -76,13 +73,12 @@ private static EncodeResult encodeF64(double[] values, EncodeContext ctx) {
long bits = Double.doubleToRawLongBits(v);
short leftU16 = (short) (bits >>> best.rightBitWidth);
rightParts[i] = bits & rightMask;
Short code = lookup.get(leftU16);
if (code != null) {
leftCodes[i] = code;
int code = indexOf(best.dict, leftU16);
if (code >= 0) {
leftCodes[i] = (short) code;
} else {
leftCodes[i] = 0;
excPos.add((long) i);
excVals.add(leftU16);
exceptions.add(i, leftU16);
}
if (v < min) {
min = v;
Expand All @@ -94,7 +90,7 @@ private static EncodeResult encodeF64(double[] values, EncodeContext ctx) {

return buildEncodeResult(
best.dict, best.rightBitWidth, leftCodes, rightParts,
DType.U64, excPos, excVals, scalarF64(min), scalarF64(max), ctx);
DType.U64, exceptions, scalarF64(min), scalarF64(max), ctx);
}

/// Rust's `SamplePlan::subsample` (`alp` crate): the whole input when it has at most
Expand Down Expand Up @@ -141,16 +137,21 @@ private static Dictionary64 findBestDictionaryF64(double[] values, int sampleLen
double bestEstSize = Double.MAX_VALUE;
int bestRightBw = 48;
short[] bestDict = new short[]{0};
LongIntMap counts = new LongIntMap(sampleLen);
short[] distinct = new short[Math.max(sampleLen, 1)];

for (int p = 1; p <= MAX_CUT; p++) {
int rightBw = 64 - p;
Map<Short, Integer> counts = new HashMap<>();
counts.clear();
int distinctLen = 0;
for (int i = 0; i < sampleLen; i++) {
long bits = Double.doubleToRawLongBits(values[i]);
short leftU16 = (short) (bits >>> rightBw);
counts.merge(leftU16, 1, Integer::sum);
int leftU16 = (int) (bits >>> rightBw) & 0xFFFF;
if (counts.increment(leftU16) == 1) {
distinct[distinctLen++] = (short) leftU16;
}
}
short[] dict = topKByCount(counts);
short[] dict = topKByCount(counts, distinct, distinctLen);
int excCount = countExceptionsF64(values, sampleLen, dict, rightBw);
int maxCode = dict.length - 1;
int leftBw = maxCode == 0 ? 1 : (Integer.SIZE - Integer.numberOfLeadingZeros(maxCode));
Expand All @@ -165,15 +166,11 @@ private static Dictionary64 findBestDictionaryF64(double[] values, int sampleLen
}

private static int countExceptionsF64(double[] values, int sampleLen, short[] dict, int rightBw) {
Map<Short, Boolean> dictSet = new HashMap<>();
for (short d : dict) {
dictSet.put(d, Boolean.TRUE);
}
int count = 0;
for (int i = 0; i < sampleLen; i++) {
long bits = Double.doubleToRawLongBits(values[i]);
short leftU16 = (short) (bits >>> rightBw);
if (!dictSet.containsKey(leftU16)) {
if (indexOf(dict, leftU16) < 0) {
count++;
}
}
Expand All @@ -191,13 +188,11 @@ private static EncodeResult encodeF32(float[] values, EncodeContext ctx) {
int sampleLen = sample.length;
Dictionary32 best = findBestDictionaryF32(sample, sampleLen);

Map<Short, Short> lookup = buildLookup(best.dict);
int rightMask = -1 >>> (32 - best.rightBitWidth);

short[] leftCodes = new short[n];
int[] rightParts = new int[n];
List<Long> excPos = new ArrayList<>();
List<Short> excVals = new ArrayList<>();
Exceptions exceptions = new Exceptions();

float min = Float.MAX_VALUE;
float max = -Float.MAX_VALUE;
Expand All @@ -206,13 +201,12 @@ private static EncodeResult encodeF32(float[] values, EncodeContext ctx) {
int bits = Float.floatToRawIntBits(v);
short leftU16 = (short) (bits >>> best.rightBitWidth);
rightParts[i] = bits & rightMask;
Short code = lookup.get(leftU16);
if (code != null) {
leftCodes[i] = code;
int code = indexOf(best.dict, leftU16);
if (code >= 0) {
leftCodes[i] = (short) code;
} else {
leftCodes[i] = 0;
excPos.add((long) i);
excVals.add(leftU16);
exceptions.add(i, leftU16);
}
if (v < min) {
min = v;
Expand All @@ -224,23 +218,28 @@ private static EncodeResult encodeF32(float[] values, EncodeContext ctx) {

return buildEncodeResult(
best.dict, best.rightBitWidth, leftCodes, rightParts,
DType.U32, excPos, excVals, scalarF32(min), scalarF32(max), ctx);
DType.U32, exceptions, scalarF32(min), scalarF32(max), ctx);
}

private static Dictionary32 findBestDictionaryF32(float[] values, int sampleLen) {
double bestEstSize = Double.MAX_VALUE;
int bestRightBw = 16;
short[] bestDict = new short[]{0};
LongIntMap counts = new LongIntMap(sampleLen);
short[] distinct = new short[Math.max(sampleLen, 1)];

for (int p = 1; p <= MAX_CUT; p++) {
int rightBw = 32 - p;
Map<Short, Integer> counts = new HashMap<>();
counts.clear();
int distinctLen = 0;
for (int i = 0; i < sampleLen; i++) {
int bits = Float.floatToRawIntBits(values[i]);
short leftU16 = (short) (bits >>> rightBw);
counts.merge(leftU16, 1, Integer::sum);
int leftU16 = (bits >>> rightBw) & 0xFFFF;
if (counts.increment(leftU16) == 1) {
distinct[distinctLen++] = (short) leftU16;
}
}
short[] dict = topKByCount(counts);
short[] dict = topKByCount(counts, distinct, distinctLen);
int excCount = countExceptionsF32(values, sampleLen, dict, rightBw);
int maxCode = dict.length - 1;
int leftBw = maxCode == 0 ? 1 : (Integer.SIZE - Integer.numberOfLeadingZeros(maxCode));
Expand All @@ -255,47 +254,54 @@ private static Dictionary32 findBestDictionaryF32(float[] values, int sampleLen)
}

private static int countExceptionsF32(float[] values, int sampleLen, short[] dict, int rightBw) {
Map<Short, Boolean> dictSet = new HashMap<>();
for (short d : dict) {
dictSet.put(d, Boolean.TRUE);
}
int count = 0;
for (int i = 0; i < sampleLen; i++) {
int bits = Float.floatToRawIntBits(values[i]);
short leftU16 = (short) (bits >>> rightBw);
if (!dictSet.containsKey(leftU16)) {
if (indexOf(dict, leftU16) < 0) {
count++;
}
}
return count;
}

private static short[] topKByCount(Map<Short, Integer> counts) {
List<Map.Entry<Short, Integer>> sorted = new ArrayList<>(counts.entrySet());
// Rust's select_dictionary: most frequent first, ties by ascending (unsigned) pattern, so the
// dictionary is decided by the data and not by HashMap iteration order.
sorted.sort(Map.Entry.<Short, Integer>comparingByValue().reversed()
.thenComparing(e -> Short.toUnsignedInt(e.getKey())));
int dictSize = Math.min(sorted.size(), MAX_DICT_SIZE);
/// Rust's `select_dictionary`: most frequent first, ties by ascending (unsigned) pattern, so the
/// dictionary is decided by the data and not by hash order.
private static short[] topKByCount(LongIntMap counts, short[] distinct, int distinctLen) {
int dictSize = Math.min(distinctLen, MAX_DICT_SIZE);
short[] dict = new short[dictSize];
for (int i = 0; i < dictSize; i++) {
dict[i] = sorted.get(i).getKey();
for (int k = 0; k < dictSize; k++) {
int best = k;
int bestCount = counts.get(distinct[k] & 0xFFFF);
for (int i = k + 1; i < distinctLen; i++) {
int c = counts.get(distinct[i] & 0xFFFF);
if (c > bestCount || (c == bestCount && (distinct[i] & 0xFFFF) < (distinct[best] & 0xFFFF))) {
best = i;
bestCount = c;
}
}
short t = distinct[k];
distinct[k] = distinct[best];
distinct[best] = t;
dict[k] = distinct[k];
}
return dict;
}

private static Map<Short, Short> buildLookup(short[] dict) {
Map<Short, Short> lookup = new HashMap<>();
for (short i = 0; i < dict.length; i++) {
lookup.put(dict[i], i);
/// Linear scan: the dictionary holds at most [#MAX_DICT_SIZE] entries, so this beats a boxed map lookup per row.
private static int indexOf(short[] dict, short value) {
for (int i = 0; i < dict.length; i++) {
if (dict[i] == value) {
return i;
}
}
return lookup;
return -1;
}

private static EncodeResult buildEncodeResult(
short[] dict, int rightBitWidth,
short[] leftCodes, Object rightPartsData, DType rightDtype,
List<Long> excPos, List<Short> excVals, ProtoScalarValue statsMin, ProtoScalarValue statsMax, EncodeContext ctx) {
Exceptions exceptions, ProtoScalarValue statsMin, ProtoScalarValue statsMax, EncodeContext ctx) {

EncodingEncoder bp = ctx.lookupEncoder(EncodingId.FASTLANES_BITPACKED);
EncodeResult leftResult = bp.encode(DType.U16, leftCodes, ctx);
Expand All @@ -315,14 +321,11 @@ private static EncodeResult buildEncodeResult(

EncodeNode[] children;
ProtoPatchesMetadata patchesMeta = null;
if (excPos.isEmpty()) {
if (exceptions.size == 0) {
children = new EncodeNode[]{leftNode, rightNode};
} else {
long[] excPosArr = excPos.stream().mapToLong(Long::longValue).toArray();
short[] excValsArr = new short[excVals.size()];
for (int i = 0; i < excVals.size(); i++) {
excValsArr[i] = excVals.get(i);
}
long[] excPosArr = Arrays.copyOf(exceptions.positions, exceptions.size);
short[] excValsArr = Arrays.copyOf(exceptions.values, exceptions.size);

// Rust's compress_patches: indices narrowed to the smallest unsigned type that holds them,
// values left as they are (a constant array when they are all equal), and none of it
Expand All @@ -347,7 +350,7 @@ private static EncodeResult buildEncodeResult(
}

patchesMeta = new ProtoPatchesMetadata(
excPos.size(),
exceptions.size,
0L,
io.github.dfa1.vortex.core.proto.ProtoPType.fromValue(idxType.ordinal()),
null, null, null);
Expand All @@ -366,6 +369,24 @@ private static EncodeResult buildEncodeResult(
return new EncodeResult(root, List.copyOf(allBuffers), statsMin, statsMax);
}

/// The rows whose left bits missed the dictionary, kept in primitive arrays: boxing a `Long` and
/// a `Short` per exception showed up in the profile on noisy doubles, where most rows miss.
private static final class Exceptions {
long[] positions = new long[16];
short[] values = new short[16];
int size;

void add(long position, short value) {
if (size == positions.length) {
positions = Arrays.copyOf(positions, size * 2);
values = Arrays.copyOf(values, size * 2);
}
positions[size] = position;
values[size] = value;
size++;
}
}

private static ProtoScalarValue scalarF64(double v) {
return ProtoScalarValue.ofF64Value(v);
}
Expand Down
Loading
Loading