mirror of
https://github.com/nosqlbench/nosqlbench.git
synced 2026-10-13 09:52:01 -05:00
feat(vectordata): walk MNode records in place
Port the reference's raw scan: a record's fields reached as views over its own bytes, names skipped rather than copied, every tag advanced at its true width. A Field reads its payload through accessors typed by the wire form — a Half as a float, the temporal trio as text, Nanos as two numbers — or materializes the AST view on request. Schema discovery and predicate flattening ride on the same walk.
This commit is contained in:
1 parent
97fd3e2495
commit
0f1b0e8a30
4 files changed
+641
No files matched your search
@@ -0,0 +1,164 @@
|
||||
/*
|
||||
* Copyright (c) 2026 The NoSQLBench Authors.
|
||||
*
|
||||
* Licensed under the Apache License, Version 2.0 (the "License");
|
||||
* you may not use this file except in compliance with the License.
|
||||
* You may obtain a copy of the License at
|
||||
*
|
||||
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
*
|
||||
* Unless required by applicable law or agreed to in writing, software
|
||||
* distributed under the License is distributed on an "AS IS" BASIS,
|
||||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
* See the License for the specific language governing permissions and
|
||||
* limitations under the License.
|
||||
*/
|
||||
package io.nosqlbench.vectordata.anode;
|
||||
|
||||
import java.nio.ByteBuffer;
|
||||
import java.nio.ByteOrder;
|
||||
import java.nio.charset.StandardCharsets;
|
||||
import java.util.Arrays;
|
||||
|
||||
/// One field of a record, viewed in place.
|
||||
///
|
||||
/// Nothing is copied: the name and the value payload are positions
|
||||
/// into the buffer the record was read from, and the tag is the wire
|
||||
/// discriminant. The accessors read the payload directly — a caller on
|
||||
/// a hot path uses the primitive ones and never materializes a name or
|
||||
/// an [MValue]; a caller wanting the AST view converts once through
|
||||
/// [#value].
|
||||
///
|
||||
/// The accessors are typed by the tag's **wire** form, not its name:
|
||||
/// `Date`, `Time`, and `DateTime` are length-prefixed strings and read
|
||||
/// through [#stringValue]; `Half` holds binary16 bits and reads through
|
||||
/// [#doubleValue], never as an integer; `Nanos` is two numbers and
|
||||
/// reads through [#nanosSeconds] and [#nanosAdjust], never as one.
|
||||
public final class Field {
|
||||
private final byte[] data;
|
||||
private final int index;
|
||||
private final int nameStart;
|
||||
private final int nameLength;
|
||||
private final int tag;
|
||||
private final int valueStart;
|
||||
private final int valueEnd;
|
||||
|
||||
Field(byte[] data, int index, int nameStart, int nameLength, int tag, int valueStart, int valueEnd) {
|
||||
this.data = data; this.index = index; this.nameStart = nameStart; this.nameLength = nameLength;
|
||||
this.tag = tag; this.valueStart = valueStart; this.valueEnd = valueEnd;
|
||||
}
|
||||
|
||||
/// Position in the record's field order — the stable identity a
|
||||
/// compiled binder addresses fields by.
|
||||
public int index() { return index; }
|
||||
/// The wire type discriminant; see [TypeTag].
|
||||
public int tag() { return tag; }
|
||||
/// The tag as an enum, or `null` for a byte the format does not define.
|
||||
public TypeTag typeTag() { return tag >= 0 && tag <= 28 ? TypeTag.fromCode(tag) : null; }
|
||||
/// The field's name, decoded. Allocates; a compiled path resolves
|
||||
/// names to positions once and never calls this per record.
|
||||
public String name() { return new String(data, nameStart, nameLength, StandardCharsets.UTF_8); }
|
||||
/// Where the value payload starts in the record, after the tag byte.
|
||||
public int valueOffset() { return valueStart; }
|
||||
/// The payload's length in bytes.
|
||||
public int valueLength() { return valueEnd - valueStart; }
|
||||
/// The payload as a little-endian buffer over the record's bytes.
|
||||
public ByteBuffer valueBuffer() { return ByteBuffer.wrap(data, valueStart, valueEnd - valueStart).slice().order(ByteOrder.LITTLE_ENDIAN); }
|
||||
|
||||
/// Whether the payload **is** an integer: `Int` and `Millis` (i64),
|
||||
/// `EnumOrd` and `Int32` (i32), `Short` (i16).
|
||||
public boolean isIntegral() { return tag == 1 || tag == 18 || tag == 7 || tag == 12 || tag == 13; }
|
||||
/// The integer value; refused for any other family.
|
||||
public long longValue() {
|
||||
return switch (tag) {
|
||||
case 1, 18 -> i64(valueStart);
|
||||
case 7, 12 -> i32(valueStart);
|
||||
case 13 -> i16(valueStart);
|
||||
default -> throw wrongFamily("an integer");
|
||||
};
|
||||
}
|
||||
|
||||
/// Whether the payload is a float: `Float`, `Float32`, or `Half`.
|
||||
public boolean isFloating() { return tag == 2 || tag == 16 || tag == 17; }
|
||||
/// The float value, a `Half` widened from its binary16 bits.
|
||||
public double doubleValue() {
|
||||
return switch (tag) {
|
||||
case 2 -> Double.longBitsToDouble(i64(valueStart));
|
||||
case 16 -> Float.intBitsToFloat(i32(valueStart));
|
||||
case 17 -> halfToDouble(u16(valueStart));
|
||||
default -> throw wrongFamily("a float");
|
||||
};
|
||||
}
|
||||
|
||||
public boolean isBoolean() { return tag == 3; }
|
||||
public boolean booleanValue() { if (tag != 3) throw wrongFamily("a boolean"); return data[valueStart] != 0; }
|
||||
|
||||
/// Whether the payload is text: `Text`, `EnumStr`, `TextValidated`,
|
||||
/// `Ascii`, and the temporal trio `Date`, `Time`, `DateTime`.
|
||||
public boolean isText() { return tag == 0 || tag == 6 || tag == 10 || tag == 11 || tag == 20 || tag == 21 || tag == 22; }
|
||||
/// The text, without its length prefix.
|
||||
public String stringValue() { if (!isText()) throw wrongFamily("text"); return new String(data, valueStart + 4, (int) u32(valueStart), StandardCharsets.UTF_8); }
|
||||
|
||||
/// Whether the payload is bytes: `Bytes` and `Varint` (length
|
||||
/// prefixed) or the three sixteen-byte identifiers.
|
||||
public boolean isBinary() { return tag == 4 || tag == 15 || tag == 23 || tag == 24 || tag == 25; }
|
||||
/// A copy of the bytes, without any length prefix.
|
||||
public byte[] bytesValue() {
|
||||
if (tag == 4 || tag == 15) return Arrays.copyOfRange(data, valueStart + 4, valueStart + 4 + (int) u32(valueStart));
|
||||
if (isBinary()) return Arrays.copyOfRange(data, valueStart, valueEnd);
|
||||
throw wrongFamily("bytes");
|
||||
}
|
||||
|
||||
/// Whether the payload is a `Nanos` instant.
|
||||
public boolean isNanos() { return tag == 19; }
|
||||
/// The whole seconds of a `Nanos` instant.
|
||||
public long nanosSeconds() { if (tag != 19) throw wrongFamily("a nanos instant"); return i64(valueStart); }
|
||||
/// The sub-second adjustment of a `Nanos` instant. Two fields on
|
||||
/// the wire, read as two: the seconds alone would look right and
|
||||
/// lose this.
|
||||
public int nanosAdjust() { if (tag != 19) throw wrongFamily("a nanos instant"); return i32(valueStart + 8); }
|
||||
|
||||
/// Whether the payload is a `Decimal`.
|
||||
public boolean isDecimal() { return tag == 14; }
|
||||
/// A `Decimal`'s scale; the value is the digits scaled by `10^-scale`.
|
||||
public int decimalScale() { if (tag != 14) throw wrongFamily("a decimal"); return i32(valueStart); }
|
||||
/// A `Decimal`'s unscaled big-integer digits, handed over rather
|
||||
/// than converted: the caller chooses the arbitrary-precision type.
|
||||
public byte[] decimalDigits() { if (tag != 14) throw wrongFamily("a decimal"); return Arrays.copyOfRange(data, valueStart + 8, valueStart + 8 + (int) u32(valueStart + 4)); }
|
||||
|
||||
/// Whether this field is the null value.
|
||||
public boolean isNull() { return tag == 5; }
|
||||
|
||||
/// The AST view of this field: its payload materialized as an
|
||||
/// [MValue], read exactly as [MNode#fromBytes] would read it. The
|
||||
/// bridge from a bound field to the node types, for a caller that
|
||||
/// wants structure rather than a primitive.
|
||||
public MValue value() {
|
||||
TypeTag type = typeTag();
|
||||
if (type == null) throw ScanException.invalidTag(tag);
|
||||
return MNode.readValue(valueBuffer(), type);
|
||||
}
|
||||
|
||||
private IllegalStateException wrongFamily(String wanted) {
|
||||
return new IllegalStateException("field '" + name() + "' at " + index + " holds tag " + tag + ", not " + wanted);
|
||||
}
|
||||
|
||||
private int u16(int at) { return (data[at] & 0xff) | ((data[at + 1] & 0xff) << 8); }
|
||||
private short i16(int at) { return (short) u16(at); }
|
||||
private int i32(int at) { return (data[at] & 0xff) | ((data[at + 1] & 0xff) << 8) | ((data[at + 2] & 0xff) << 16) | ((data[at + 3] & 0xff) << 24); }
|
||||
private long u32(int at) { return i32(at) & 0xffffffffL; }
|
||||
private long i64(int at) { return (i32(at) & 0xffffffffL) | ((long) i32(at + 4) << 32); }
|
||||
|
||||
/// Widens an IEEE binary16 bit pattern. `Half` stores the bits, not
|
||||
/// a value, so this is the only correct way to read one as a number.
|
||||
static double halfToDouble(int bits) {
|
||||
double sign = (bits & 0x8000) != 0 ? -1.0 : 1.0;
|
||||
int exp = (bits >> 10) & 0x1F;
|
||||
double frac = bits & 0x03FF;
|
||||
if (exp == 0) return sign * frac * Math.pow(2, -24);
|
||||
if (exp == 31) return frac == 0 ? sign * Double.POSITIVE_INFINITY : Double.NaN;
|
||||
return sign * (1.0 + frac / 1024.0) * Math.pow(2, exp - 15);
|
||||
}
|
||||
|
||||
@Override public String toString() { return "Field[" + index + " " + name() + " tag=" + tag + " bytes=" + valueLength() + "]"; }
|
||||
}
|
||||
@@ -0,0 +1,210 @@
|
||||
/*
|
||||
* Copyright (c) 2026 The NoSQLBench Authors.
|
||||
*
|
||||
* Licensed under the Apache License, Version 2.0 (the "License");
|
||||
* you may not use this file except in compliance with the License.
|
||||
* You may obtain a copy of the License at
|
||||
*
|
||||
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
*
|
||||
* Unless required by applicable law or agreed to in writing, software
|
||||
* distributed under the License is distributed on an "AS IS" BASIS,
|
||||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
* See the License for the specific language governing permissions and
|
||||
* limitations under the License.
|
||||
*/
|
||||
package io.nosqlbench.vectordata.anode;
|
||||
|
||||
import io.nosqlbench.vectordata.anode.PNode.Comparand;
|
||||
import io.nosqlbench.vectordata.anode.PNode.FieldRef;
|
||||
import io.nosqlbench.vectordata.anode.PNode.OpType;
|
||||
|
||||
import java.util.ArrayList;
|
||||
import java.util.Iterator;
|
||||
import java.util.List;
|
||||
import java.util.NoSuchElementException;
|
||||
|
||||
/// The raw walk over MNode bytes — the reference's `mnode::scan`.
|
||||
///
|
||||
/// Walks a record's fields without materializing an [MNode] or an
|
||||
/// [MValue]: names are skipped over rather than copied, and each value
|
||||
/// is reached as a position and a tag. A caller addressing fields by
|
||||
/// position never looks at a name. Built for predicate scanning in the
|
||||
/// reference and reused there as the binding hot path; here it is the
|
||||
/// binding hot path.
|
||||
public final class Scan {
|
||||
private Scan() { }
|
||||
|
||||
/// Walks a record's fields in wire order. The leader byte must be
|
||||
/// the MNode dialect: a record says what it is, and a PNode brought
|
||||
/// here is refused by its own first byte.
|
||||
public static Fields fields(byte[] data) {
|
||||
if (data.length == 0) throw ScanException.unexpectedEof();
|
||||
if ((data[0] & 0xff) != MNode.DIALECT) throw ScanException.invalidDialect(data[0] & 0xff);
|
||||
if (data.length < 3) throw ScanException.unexpectedEof();
|
||||
return new Fields(data, 3, u16(data, 1));
|
||||
}
|
||||
|
||||
/// The fields of one record, walked lazily. One [Field] view per
|
||||
/// field, each a handful of offsets into the record; no copies.
|
||||
public static final class Fields implements Iterator<Field>, Iterable<Field> {
|
||||
private final byte[] data;
|
||||
private int pos;
|
||||
private int remaining;
|
||||
private int index;
|
||||
|
||||
Fields(byte[] data, int pos, int count) { this.data = data; this.pos = pos; this.remaining = count; }
|
||||
|
||||
@Override public Iterator<Field> iterator() { return this; }
|
||||
@Override public boolean hasNext() { return remaining > 0; }
|
||||
|
||||
@Override public Field next() {
|
||||
if (remaining == 0) throw new NoSuchElementException();
|
||||
remaining--;
|
||||
int at = index++;
|
||||
try {
|
||||
if (pos + 2 > data.length) throw ScanException.unexpectedEof();
|
||||
int nameLength = u16(data, pos);
|
||||
int nameStart = pos + 2;
|
||||
if (nameStart + nameLength > data.length) throw ScanException.unexpectedEof();
|
||||
int tagAt = nameStart + nameLength;
|
||||
if (tagAt >= data.length) throw ScanException.unexpectedEof();
|
||||
int tag = data[tagAt] & 0xff;
|
||||
int valueStart = tagAt + 1;
|
||||
int end = skipValue(data, valueStart, tag);
|
||||
pos = end;
|
||||
return new Field(data, at, nameStart, nameLength, tag, valueStart, end);
|
||||
} catch (ScanException fault) {
|
||||
// A malformed record cannot be walked past; stop rather
|
||||
// than report the same fault once per declared field.
|
||||
remaining = 0;
|
||||
throw fault;
|
||||
}
|
||||
}
|
||||
|
||||
/// The offset the walk has reached: the end of the last field
|
||||
/// returned, or the end of the record once exhausted.
|
||||
public int position() { return pos; }
|
||||
}
|
||||
|
||||
/// Advances past one typed value whose payload starts at `pos`,
|
||||
/// immediately after the tag byte, returning the offset of the
|
||||
/// first byte after it. Pure indexing, no buffers.
|
||||
public static int skipValue(byte[] data, int pos, int tag) {
|
||||
switch (tag) {
|
||||
case 5: return pos; // Null
|
||||
case 3: return bounded(data, pos, 1); // Bool
|
||||
case 13: case 17: return bounded(data, pos, 2); // Short, Half
|
||||
case 7: case 12: case 16: return bounded(data, pos, 4); // EnumOrd, Int32, Float32
|
||||
case 1: case 2: case 18: return bounded(data, pos, 8); // Int, Float, Millis
|
||||
case 19: return bounded(data, pos, 12); // Nanos
|
||||
case 23: case 24: case 25: return bounded(data, pos, 16); // UuidV1, UuidV7, Ulid
|
||||
// Length-prefixed: Text, Bytes, EnumStr, Map, TextValidated, Ascii, Date, Time, DateTime, Varint
|
||||
case 0: case 4: case 6: case 9: case 10: case 11: case 20: case 21: case 22: case 15: return skipLength32(data, pos);
|
||||
case 14: { // Decimal: i32 scale + u32 len + digits
|
||||
if (pos + 8 > data.length) throw ScanException.unexpectedEof();
|
||||
return bounded(data, pos, 8 + u32(data, pos + 4));
|
||||
}
|
||||
case 8: case 27: return skipTaggedList(data, pos); // List, Set
|
||||
case 26: { // Array: elem tag + u32 count + untagged elements
|
||||
if (pos + 5 > data.length) throw ScanException.unexpectedEof();
|
||||
int elementTag = data[pos] & 0xff;
|
||||
long count = u32(data, pos + 1);
|
||||
int p = pos + 5;
|
||||
for (long i = 0; i < count; i++) p = skipValue(data, p, elementTag);
|
||||
return p;
|
||||
}
|
||||
case 28: { // TypedMap: u32 count + tagged pairs
|
||||
if (pos + 4 > data.length) throw ScanException.unexpectedEof();
|
||||
long count = u32(data, pos);
|
||||
int p = pos + 4;
|
||||
for (long i = 0; i < count; i++) {
|
||||
if (p >= data.length) throw ScanException.unexpectedEof();
|
||||
p = skipValue(data, p + 1, data[p] & 0xff);
|
||||
if (p >= data.length) throw ScanException.unexpectedEof();
|
||||
p = skipValue(data, p + 1, data[p] & 0xff);
|
||||
}
|
||||
return p;
|
||||
}
|
||||
default: throw ScanException.invalidTag(tag);
|
||||
}
|
||||
}
|
||||
|
||||
private static int bounded(byte[] data, int pos, long size) {
|
||||
long end = pos + size;
|
||||
if (end > data.length) throw ScanException.unexpectedEof();
|
||||
return (int) end;
|
||||
}
|
||||
|
||||
private static int skipLength32(byte[] data, int pos) {
|
||||
if (pos + 4 > data.length) throw ScanException.unexpectedEof();
|
||||
return bounded(data, pos, 4 + u32(data, pos));
|
||||
}
|
||||
|
||||
private static int skipTaggedList(byte[] data, int pos) {
|
||||
if (pos + 4 > data.length) throw ScanException.unexpectedEof();
|
||||
long count = u32(data, pos);
|
||||
int p = pos + 4;
|
||||
for (long i = 0; i < count; i++) {
|
||||
if (p >= data.length) throw ScanException.unexpectedEof();
|
||||
p = skipValue(data, p + 1, data[p] & 0xff);
|
||||
}
|
||||
return p;
|
||||
}
|
||||
|
||||
static int u16(byte[] data, int at) { return (data[at] & 0xff) | ((data[at + 1] & 0xff) << 8); }
|
||||
static long u32(byte[] data, int at) { return ((data[at] & 0xff) | ((data[at + 1] & 0xff) << 8) | ((data[at + 2] & 0xff) << 16) | ((long) (data[at + 3] & 0xff) << 24)); }
|
||||
|
||||
/// The field layout a sample record carries: names in wire order
|
||||
/// and the tag each holds, captured once so a compiled path can
|
||||
/// address fields by position thereafter.
|
||||
public record RecordSchema(List<String> fieldNames, List<Integer> tags) {
|
||||
public RecordSchema { fieldNames = List.copyOf(fieldNames); tags = List.copyOf(tags); }
|
||||
public int fieldCount() { return fieldNames.size(); }
|
||||
}
|
||||
|
||||
/// Discovers the field layout from a raw MNode record, skipping
|
||||
/// values as it goes.
|
||||
public static RecordSchema discoverSchema(byte[] data) {
|
||||
List<String> names = new ArrayList<>();
|
||||
List<Integer> tags = new ArrayList<>();
|
||||
for (Field field : fields(data)) { names.add(field.name()); tags.add(field.tag()); }
|
||||
return new RecordSchema(names, tags);
|
||||
}
|
||||
|
||||
/// One condition of a flat predicate: `field op comparands`.
|
||||
public record FlatCondition(String field, OpType op, List<Comparand> comparands) {
|
||||
public FlatCondition { comparands = List.copyOf(comparands); }
|
||||
}
|
||||
|
||||
/// Flattens a predicate into `AND`-joined conditions over named
|
||||
/// fields, or `null` when it is not one: an `OR` anywhere, or an
|
||||
/// indexed field reference, leaves it without a flat parameter
|
||||
/// list.
|
||||
public static List<FlatCondition> flattenAnd(PNode node) {
|
||||
if (node instanceof PNode.Predicate p) {
|
||||
if (!(p.field() instanceof FieldRef.Named named)) return null;
|
||||
return List.of(new FlatCondition(named.name(), p.op(), p.comparands()));
|
||||
}
|
||||
PNode.Conjugate c = (PNode.Conjugate) node;
|
||||
if (c.type() != PNode.ConjugateType.AND) return null;
|
||||
List<FlatCondition> out = new ArrayList<>();
|
||||
for (PNode child : c.children()) {
|
||||
List<FlatCondition> flat = flattenAnd(child);
|
||||
if (flat == null) return null;
|
||||
out.addAll(flat);
|
||||
}
|
||||
return out;
|
||||
}
|
||||
|
||||
/// Whether a field absent from a record satisfies a condition — the
|
||||
/// evaluator's rule: a missing field matches `= NULL`, `IN (…NULL…)`,
|
||||
/// and `!= <anything but NULL>`.
|
||||
public static boolean missingFieldPasses(OpType op, List<Comparand> comparands) {
|
||||
return switch (op) {
|
||||
case EQ, IN -> comparands.stream().anyMatch(c -> c instanceof Comparand.Null);
|
||||
case NE -> comparands.stream().noneMatch(c -> c instanceof Comparand.Null);
|
||||
default -> false;
|
||||
};
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,42 @@
|
||||
/*
|
||||
* Copyright (c) 2026 The NoSQLBench Authors.
|
||||
*
|
||||
* Licensed under the Apache License, Version 2.0 (the "License");
|
||||
* you may not use this file except in compliance with the License.
|
||||
* You may obtain a copy of the License at
|
||||
*
|
||||
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
*
|
||||
* Unless required by applicable law or agreed to in writing, software
|
||||
* distributed under the License is distributed on an "AS IS" BASIS,
|
||||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
* See the License for the specific language governing permissions and
|
||||
* limitations under the License.
|
||||
*/
|
||||
package io.nosqlbench.vectordata.anode;
|
||||
|
||||
/// What went wrong walking raw MNode bytes. The [Kind] is matchable,
|
||||
/// not only readable.
|
||||
public class ScanException extends ANodeException {
|
||||
public enum Kind {
|
||||
/// The data ended before the structure it declares was complete.
|
||||
UNEXPECTED_EOF,
|
||||
/// A type tag byte the wire format does not define.
|
||||
INVALID_TAG,
|
||||
/// The record does not begin with the MNode dialect leader.
|
||||
INVALID_DIALECT
|
||||
}
|
||||
|
||||
private final Kind kind;
|
||||
private final int value;
|
||||
|
||||
private ScanException(Kind kind, String message, int value) { super(message); this.kind = kind; this.value = value; }
|
||||
|
||||
public Kind kind() { return kind; }
|
||||
/// The offending tag or dialect byte, for `INVALID_TAG` and `INVALID_DIALECT`.
|
||||
public int value() { return value; }
|
||||
|
||||
public static ScanException unexpectedEof() { return new ScanException(Kind.UNEXPECTED_EOF, "unexpected end of data", -1); }
|
||||
public static ScanException invalidTag(int tag) { return new ScanException(Kind.INVALID_TAG, "invalid type tag: " + tag, tag); }
|
||||
public static ScanException invalidDialect(int dialect) { return new ScanException(Kind.INVALID_DIALECT, String.format("invalid dialect: 0x%02x", dialect), dialect); }
|
||||
}
|
||||
@@ -0,0 +1,225 @@
|
||||
/*
|
||||
* Copyright (c) 2026 The NoSQLBench Authors.
|
||||
*
|
||||
* Licensed under the Apache License, Version 2.0 (the "License");
|
||||
* you may not use this file except in compliance with the License.
|
||||
* You may obtain a copy of the License at
|
||||
*
|
||||
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
*
|
||||
* Unless required by applicable law or agreed to in writing, software
|
||||
* distributed under the License is distributed on an "AS IS" BASIS,
|
||||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
* See the License for the specific language governing permissions and
|
||||
* limitations under the License.
|
||||
*/
|
||||
package io.nosqlbench.vectordata.anode;
|
||||
|
||||
import io.nosqlbench.vectordata.anode.PNode.Comparand;
|
||||
import io.nosqlbench.vectordata.anode.PNode.ConjugateType;
|
||||
import io.nosqlbench.vectordata.anode.PNode.FieldRef;
|
||||
import io.nosqlbench.vectordata.anode.PNode.OpType;
|
||||
import org.junit.jupiter.api.Tag;
|
||||
import org.junit.jupiter.api.Test;
|
||||
|
||||
import java.io.ByteArrayOutputStream;
|
||||
import java.util.ArrayList;
|
||||
import java.util.Arrays;
|
||||
import java.util.List;
|
||||
import java.util.Map;
|
||||
|
||||
import static org.junit.jupiter.api.Assertions.*;
|
||||
|
||||
/// The raw walk over MNode bytes: every tag skipped at its true width,
|
||||
/// every accessor typed by the wire form, the leader byte respected,
|
||||
/// and a malformed record stopped rather than misread.
|
||||
@Tag("unit")
|
||||
class ScanTest {
|
||||
static byte[] sixteen(int fill) { byte[] b = new byte[16]; Arrays.fill(b, (byte) fill); return b; }
|
||||
|
||||
/// One field of every value kind an MNode can carry, in a fixed order.
|
||||
static MNode everyKind() {
|
||||
return new MNode()
|
||||
.insert("text", new MValue.Text("héllo"))
|
||||
.insert("int", new MValue.Int(-7L))
|
||||
.insert("float", new MValue.Float(2.5))
|
||||
.insert("bool", new MValue.Bool(true))
|
||||
.insert("bytes", new MValue.Bytes(new byte[] {1, 2, 3}))
|
||||
.insert("null", MValue.NULL)
|
||||
.insert("enum_str", new MValue.EnumStr("red"))
|
||||
.insert("enum_ord", new MValue.EnumOrd(3))
|
||||
.insert("list", new MValue.ListValue(List.of(new MValue.Int(1), new MValue.Text("a"))))
|
||||
.insert("map", new MValue.MapValue(new MNode().insert("k", new MValue.Int(9))))
|
||||
.insert("ascii", new MValue.Ascii("abc"))
|
||||
.insert("int32", new MValue.Int32(65536))
|
||||
.insert("short", new MValue.Short((short) -2))
|
||||
.insert("float32", new MValue.Float32(1.5f))
|
||||
.insert("half", new MValue.Half(0x3C00))
|
||||
.insert("millis", new MValue.Millis(1_700_000_000_000L))
|
||||
.insert("nanos", new MValue.Nanos(1_700_000_000L, 123))
|
||||
.insert("date", new MValue.Date("2026-01-02"))
|
||||
.insert("time", new MValue.Time("03:04:05"))
|
||||
.insert("datetime", new MValue.DateTime("2026-01-02T03:04:05Z"))
|
||||
.insert("uuid1", new MValue.UuidV1(sixteen(1)))
|
||||
.insert("uuid7", new MValue.UuidV7(sixteen(7)))
|
||||
.insert("ulid", new MValue.Ulid(sixteen(9)))
|
||||
.insert("array", new MValue.ArrayValue(TypeTag.INT32, List.of(new MValue.Int32(1), new MValue.Int32(2))))
|
||||
.insert("set", new MValue.SetValue(List.of(new MValue.Text("x"))))
|
||||
.insert("typed_map", new MValue.TypedMap(List.of(new MValue.TypedMap.Entry(new MValue.Text("k"), new MValue.Int(1)))));
|
||||
}
|
||||
|
||||
static Map<String, Field> byName(byte[] record) {
|
||||
Map<String, Field> out = new java.util.LinkedHashMap<>();
|
||||
for (Field f : Scan.fields(record)) out.put(f.name(), f);
|
||||
return out;
|
||||
}
|
||||
|
||||
/// Every value kind is skipped at its true width — the walk lands
|
||||
/// exactly on the end of the record — and each field's in-place
|
||||
/// view materializes back to the value that was written.
|
||||
@Test void everyValueKindWalksInPlace() {
|
||||
MNode node = everyKind();
|
||||
byte[] bytes = node.toBytes();
|
||||
Scan.Fields fields = Scan.fields(bytes);
|
||||
List<String> names = new ArrayList<>();
|
||||
int index = 0;
|
||||
for (Field f : fields) {
|
||||
names.add(f.name());
|
||||
assertEquals(index++, f.index());
|
||||
MValue original = node.get(f.name());
|
||||
assertEquals(original.tag().code(), f.tag(), f.name());
|
||||
assertEquals(original.tag(), f.typeTag());
|
||||
assertEquals(original.display(), f.value().display(), "AST bridge for " + f.name());
|
||||
}
|
||||
assertEquals(new ArrayList<>(node.fields().keySet()), names, "wire order");
|
||||
assertEquals(bytes.length, fields.position(), "the walk consumed exactly the record");
|
||||
assertEquals(26, names.size());
|
||||
}
|
||||
|
||||
/// Accessors are typed by the wire form: integers, floats (a Half
|
||||
/// widened from its bits), text (the temporal trio included),
|
||||
/// bytes (identifiers unprefixed), a Nanos as two numbers.
|
||||
@Test void accessorsReadEachFamilyInPlace() {
|
||||
Map<String, Field> f = byName(everyKind().toBytes());
|
||||
assertTrue(f.get("int").isIntegral()); assertEquals(-7L, f.get("int").longValue());
|
||||
assertEquals(65536L, f.get("int32").longValue());
|
||||
assertEquals(-2L, f.get("short").longValue());
|
||||
assertEquals(3L, f.get("enum_ord").longValue());
|
||||
assertEquals(1_700_000_000_000L, f.get("millis").longValue());
|
||||
assertTrue(f.get("float").isFloating()); assertEquals(2.5, f.get("float").doubleValue());
|
||||
assertEquals(1.5, f.get("float32").doubleValue());
|
||||
assertEquals(1.0, f.get("half").doubleValue(), "0x3C00 is 1.0, not 15360");
|
||||
assertFalse(f.get("half").isIntegral(), "a half is a float on the wire");
|
||||
assertTrue(f.get("bool").booleanValue());
|
||||
assertEquals("héllo", f.get("text").stringValue());
|
||||
assertEquals("abc", f.get("ascii").stringValue());
|
||||
assertEquals("red", f.get("enum_str").stringValue());
|
||||
assertEquals("2026-01-02", f.get("date").stringValue());
|
||||
assertEquals("03:04:05", f.get("time").stringValue());
|
||||
assertEquals("2026-01-02T03:04:05Z", f.get("datetime").stringValue());
|
||||
assertFalse(f.get("datetime").isIntegral(), "a datetime is text on the wire");
|
||||
assertArrayEquals(new byte[] {1, 2, 3}, f.get("bytes").bytesValue());
|
||||
assertArrayEquals(sixteen(1), f.get("uuid1").bytesValue());
|
||||
assertArrayEquals(sixteen(7), f.get("uuid7").bytesValue());
|
||||
assertArrayEquals(sixteen(9), f.get("ulid").bytesValue());
|
||||
assertTrue(f.get("nanos").isNanos());
|
||||
assertEquals(1_700_000_000L, f.get("nanos").nanosSeconds());
|
||||
assertEquals(123, f.get("nanos").nanosAdjust());
|
||||
assertTrue(f.get("null").isNull());
|
||||
assertEquals(0, f.get("null").valueLength());
|
||||
assertThrows(IllegalStateException.class, () -> f.get("text").longValue(), "the wrong family is refused, not misread");
|
||||
assertThrows(IllegalStateException.class, () -> f.get("half").longValue());
|
||||
}
|
||||
|
||||
/// The three tags the node types read as something else — Decimal,
|
||||
/// Varint, TextValidated — are skipped at their own widths and read
|
||||
/// through their own accessors, from a hand-built record.
|
||||
@Test void decimalVarintAndValidatedTextAreWalkedFromTheWire() {
|
||||
ByteArrayOutputStream out = new ByteArrayOutputStream();
|
||||
out.write(MNode.DIALECT); Wire.u16(out, 3);
|
||||
Wire.u16(out, 1); out.write('d'); out.write(14); Wire.i32(out, 2); Wire.u32(out, 1); out.write(7);
|
||||
Wire.u16(out, 1); out.write('v'); out.write(15); Wire.u32(out, 2); out.write(1); out.write(2);
|
||||
Wire.u16(out, 1); out.write('t'); out.write(10); Wire.u32(out, 2); out.write('h'); out.write('i');
|
||||
byte[] record = out.toByteArray();
|
||||
Map<String, Field> f = byName(record);
|
||||
assertTrue(f.get("d").isDecimal());
|
||||
assertEquals(2, f.get("d").decimalScale());
|
||||
assertArrayEquals(new byte[] {7}, f.get("d").decimalDigits());
|
||||
assertEquals(TypeTag.DECIMAL, f.get("d").typeTag());
|
||||
assertTrue(f.get("v").isBinary());
|
||||
assertArrayEquals(new byte[] {1, 2}, f.get("v").bytesValue());
|
||||
assertTrue(f.get("t").isText());
|
||||
assertEquals("hi", f.get("t").stringValue());
|
||||
assertEquals(TypeTag.TEXT_VALIDATED, f.get("t").typeTag());
|
||||
Scan.Fields walk = Scan.fields(record);
|
||||
while (walk.hasNext()) walk.next();
|
||||
assertEquals(record.length, walk.position(), "each width was right");
|
||||
ScanException bad = assertThrows(ScanException.class, () -> Scan.skipValue(record, 5, 99));
|
||||
assertEquals(ScanException.Kind.INVALID_TAG, bad.kind());
|
||||
assertEquals(99, bad.value());
|
||||
}
|
||||
|
||||
/// A record says what it is: a PNode is refused by its own leader.
|
||||
@Test void aPredicateIsRefusedByItsLeaderByte() {
|
||||
PNode p = new PNode.Predicate(new FieldRef.Named("a"), OpType.EQ, List.of(new Comparand.Int(1)));
|
||||
ScanException e = assertThrows(ScanException.class, () -> Scan.fields(p.toBytesNamed()));
|
||||
assertEquals(ScanException.Kind.INVALID_DIALECT, e.kind());
|
||||
assertEquals(PNode.DIALECT, e.value());
|
||||
}
|
||||
|
||||
/// A malformed record cannot be walked past: the fault is reported
|
||||
/// once and the walk ends, rather than once per declared field.
|
||||
@Test void aTruncatedRecordStopsTheWalk() {
|
||||
assertEquals(ScanException.Kind.UNEXPECTED_EOF, assertThrows(ScanException.class, () -> Scan.fields(new byte[0])).kind());
|
||||
assertEquals(ScanException.Kind.UNEXPECTED_EOF, assertThrows(ScanException.class, () -> Scan.fields(new byte[] {1, 5})).kind());
|
||||
byte[] whole = everyKind().toBytes();
|
||||
byte[] cut = Arrays.copyOf(whole, whole.length - 2);
|
||||
Scan.Fields fields = Scan.fields(cut);
|
||||
int walked = 0;
|
||||
ScanException e = null;
|
||||
while (fields.hasNext()) {
|
||||
try { fields.next(); walked++; }
|
||||
catch (ScanException fault) { e = fault; }
|
||||
}
|
||||
assertNotNull(e, "the cut field faulted");
|
||||
assertEquals(ScanException.Kind.UNEXPECTED_EOF, e.kind());
|
||||
assertEquals(25, walked, "every whole field before it was read");
|
||||
assertFalse(fields.hasNext(), "and the walk ended there");
|
||||
}
|
||||
|
||||
/// The schema a sample record carries: names and tags in wire order.
|
||||
@Test void aSchemaIsDiscoveredFromOneRecord() {
|
||||
Scan.RecordSchema schema = Scan.discoverSchema(new MNode().insert("id", new MValue.Int(1)).insert("tag", new MValue.Text("a")).toBytes());
|
||||
assertEquals(List.of("id", "tag"), schema.fieldNames());
|
||||
assertEquals(List.of(TypeTag.INT.code(), TypeTag.TEXT.code()), schema.tags());
|
||||
assertEquals(2, schema.fieldCount());
|
||||
}
|
||||
|
||||
static PNode pred(String field, OpType op, Comparand c) { return new PNode.Predicate(new FieldRef.Named(field), op, List.of(c)); }
|
||||
static PNode and(PNode... children) { return new PNode.Conjugate(ConjugateType.AND, List.of(children)); }
|
||||
|
||||
/// A flat conjunction flattens, nested ANDs included; an OR anywhere
|
||||
/// or an indexed field leaves the predicate without a flat list.
|
||||
@Test void onlyAFlatConjunctionFlattens() {
|
||||
List<Scan.FlatCondition> one = Scan.flattenAnd(pred("a", OpType.GT, new Comparand.Int(1)));
|
||||
assertEquals(1, one.size());
|
||||
assertEquals("a", one.get(0).field());
|
||||
List<Scan.FlatCondition> nested = Scan.flattenAnd(and(pred("a", OpType.EQ, new Comparand.Int(1)), and(pred("b", OpType.LT, new Comparand.Float(2.0)), pred("c", OpType.IN, new Comparand.Text("x")))));
|
||||
assertEquals(List.of("a", "b", "c"), nested.stream().map(Scan.FlatCondition::field).toList());
|
||||
assertEquals(OpType.IN, nested.get(2).op());
|
||||
PNode or = new PNode.Conjugate(ConjugateType.OR, List.of(pred("a", OpType.EQ, new Comparand.Int(1)), pred("a", OpType.EQ, new Comparand.Int(2))));
|
||||
assertNull(Scan.flattenAnd(or));
|
||||
assertNull(Scan.flattenAnd(and(pred("a", OpType.EQ, new Comparand.Int(1)), or)), "an OR anywhere");
|
||||
assertNull(Scan.flattenAnd(new PNode.Predicate(new FieldRef.Index(0), OpType.EQ, List.of(new Comparand.Int(1)))), "an indexed field");
|
||||
}
|
||||
|
||||
/// The evaluator's rule for an absent field, stated once.
|
||||
@Test void aMissingFieldPassesOnlyNullComparisons() {
|
||||
assertTrue(Scan.missingFieldPasses(OpType.EQ, List.of(Comparand.NULL)));
|
||||
assertTrue(Scan.missingFieldPasses(OpType.IN, List.of(new Comparand.Int(1), Comparand.NULL)));
|
||||
assertTrue(Scan.missingFieldPasses(OpType.NE, List.of(new Comparand.Int(5))));
|
||||
assertFalse(Scan.missingFieldPasses(OpType.NE, List.of(Comparand.NULL)));
|
||||
assertFalse(Scan.missingFieldPasses(OpType.EQ, List.of(new Comparand.Int(5))));
|
||||
assertFalse(Scan.missingFieldPasses(OpType.GT, List.of(Comparand.NULL)));
|
||||
}
|
||||
}
|
||||
Reference in new issue
Block a user