feat(virtdata): bind vectordata records as projective functions

Four mappers over a facet of opaque records, each learning the layout
once and resolving no field name per cycle: RecordField projects one
field to the Java value its wire type says, RecordFields binds a record
to named values in bind order, RecordForm binds under a form the facet
declares, and PredicateParams binds a predicate's comparands as
parameters typed from the metadata fields they constrain. The shared
projection maps bind types to driver-ready Java values, reading varint
and decimal payloads as big-endian two's complement.
This commit is contained in:
Jonathan Shook committed 2026-09-04 23:34:00 +00:00
1 parent e0331a9e46
commit 62b49e8a52
7 files changed
+833

No files matched your search

@@ -0,0 +1,64 @@
package io.nosqlbench.virtdata.lib.vectors.vectordata;
/*
* Copyright (c) nosqlbench
*
* 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.
*/
import io.nosqlbench.vectordata.Catalog;
import io.nosqlbench.vectordata.CatalogSources;
import io.nosqlbench.vectordata.FacetNames;
import io.nosqlbench.vectordata.TestDataView;
import io.nosqlbench.vectordata.VectorDataSettings;
import io.nosqlbench.vectordata.binding.Layout;
import io.nosqlbench.vectordata.records.RecordFacet;
import java.util.function.LongFunction;
/// Base plumbing for record-binding functions: opens the named dataset
/// profile from the default catalogs, opens a facet of opaque records
/// by ordinal, and learns its field layout **once**, before the first
/// cycle. Everything a subclass compiles — a binder, a form, a
/// predicate template — is compiled against that layout, so the
/// per-cycle path resolves no field names.
///
/// A cycle value is the record ordinal, as it is for every other
/// function in this package; wrapping, hashing, or striding cycles
/// onto ordinals is the workload's policy. Reading a record costs the
/// page that holds it, so a workload touching a scattered fraction of
/// a remote facet fetches a fraction of it.
public abstract class CoreRecords<T> implements LongFunction<T> {
protected final TestDataView tdv;
protected final String facetName;
protected final RecordFacet facet;
protected final Layout layout;
protected CoreRecords(String datasetAndProfile, String facetName, VectorDataSettings settings) {
tdv = Catalog.of(CatalogSources.defaults(), settings).openProfile(datasetAndProfile);
this.facetName = FacetNames.canonical(facetName);
facet = tdv.openFacetRecords(this.facetName);
layout = Layout.discover(facet);
}
/// The facet's field layout: names in wire order, with what each
/// binds as.
public Layout layout() { return layout; }
/// The number of records the facet holds — exact and cheap, from
/// the container's index.
public long count() { return facet.count(); }
}
@@ -0,0 +1,122 @@
package io.nosqlbench.virtdata.lib.vectors.vectordata;
/*
* Copyright (c) nosqlbench
*
* 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.
*/
import io.nosqlbench.vectordata.Catalog;
import io.nosqlbench.vectordata.CatalogSources;
import io.nosqlbench.vectordata.FacetNames;
import io.nosqlbench.vectordata.TestDataView;
import io.nosqlbench.vectordata.VectorDataSettings;
import io.nosqlbench.vectordata.anode.ANode;
import io.nosqlbench.vectordata.anode.PNode;
import io.nosqlbench.vectordata.anode.PNode.Comparand;
import io.nosqlbench.vectordata.anode.PNode.OpType;
import io.nosqlbench.vectordata.anode.PNodeDisplay;
import io.nosqlbench.vectordata.binding.BindException;
import io.nosqlbench.vectordata.binding.Condition;
import io.nosqlbench.vectordata.binding.Layout;
import io.nosqlbench.vectordata.binding.PredicateBinder;
import io.nosqlbench.vectordata.records.RecordFacet;
import io.nosqlbench.virtdata.api.annotations.Categories;
import io.nosqlbench.virtdata.api.annotations.Category;
import io.nosqlbench.virtdata.api.annotations.Example;
import io.nosqlbench.virtdata.api.annotations.ThreadSafeMapper;
import java.util.ArrayList;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.function.LongFunction;
/*
* Binds each predicate record of a predicate facet to the parameters
* of a filter prepared once: a map of parameter name to comparand
* value, in condition order, each typed from the metadata field the
* condition constrains rather than from the comparand. The filter's
* shape comes from a template — the facet's first predicate, or one
* written in the display grammar, '(created >= 0 AND count > 0)' — and
* must be a flat conjunction over named fields; a record of another
* shape is refused rather than bound into the wrong conditions. A
* membership condition ('IN') binds a list. The cycle value is the
* predicate record's ordinal. */
@ThreadSafeMapper
@Categories(Category.vectors)
public class PredicateParams implements LongFunction<Map<String, Object>> {
private final TestDataView tdv;
private final RecordFacet predicates;
private final PredicateBinder binder;
@Example({"PredicateParams('exampledataset:exampleprofile','metadata_predicates','metadata_content')",
"Bind each predicate's comparands as parameters, in the shape of the facet's first predicate, typed from the metadata fields"})
public PredicateParams(String datasetAndProfile, String predicateFacet, String metadataFacet) {
this(datasetAndProfile, predicateFacet, metadataFacet, "", VectorDataSettings.defaults());
}
@Example({"PredicateParams('exampledataset:exampleprofile','metadata_predicates','metadata_content','(created >= 0 AND count > 0)')",
"Bind each predicate's comparands in the shape of a template written in the display grammar"})
public PredicateParams(String datasetAndProfile, String predicateFacet, String metadataFacet, String template) {
this(datasetAndProfile, predicateFacet, metadataFacet, template, VectorDataSettings.defaults());
}
/** Construct with explicit vectordata settings, including an isolated cache location. */
public PredicateParams(String datasetAndProfile, String predicateFacet, String metadataFacet, String template,
VectorDataSettings settings) {
tdv = Catalog.of(CatalogSources.defaults(), settings).openProfile(datasetAndProfile);
predicates = tdv.openFacetRecords(FacetNames.canonical(predicateFacet));
Layout layout = Layout.discover(tdv.openFacetRecords(FacetNames.canonical(metadataFacet)));
PNode sample = template == null || template.isBlank() ? firstPredicate() : PNodeDisplay.parse(template);
binder = PredicateBinder.compile(sample, layout);
}
private PNode firstPredicate() {
if (predicates.count() == 0) throw BindException.noLayout(predicates.name());
ANode first = ANode.decode(predicates.recordBytes(0));
if (!(first instanceof ANode.P p))
throw BindException.record("facet '" + predicates.name() + "': its first record is not a predicate — its dialect byte says MNode");
return p.node();
}
/// The conditions the filter binds, in order: field, parameter
/// name, operator, bind type, and arity. Read once, to build the
/// statement's WHERE clause with a placeholder per parameter.
public List<Condition> conditions() { return binder.conditions(); }
/// The parameter names, in condition order — the keys of every map
/// this function returns.
public List<String> parameters() {
List<String> names = new ArrayList<>();
for (Condition c : binder.conditions()) names.add(c.parameter());
return names;
}
@Override
public Map<String, Object> apply(long ordinal) {
Map<String, Object> bound = new LinkedHashMap<>();
binder.bindEach(predicates.recordBytes(ordinal), (condition, comparands) -> bound.put(condition.parameter(), value(condition, comparands)));
return bound;
}
private static Object value(Condition condition, List<Comparand> comparands) {
if (condition.op() != OpType.IN && comparands.size() == 1) return RecordProjections.of(comparands.get(0), condition.bindType());
List<Object> values = new ArrayList<>(comparands.size());
for (Comparand c : comparands) values.add(RecordProjections.of(c, condition.bindType()));
return values;
}
}
@@ -0,0 +1,62 @@
package io.nosqlbench.virtdata.lib.vectors.vectordata;
/*
* Copyright (c) nosqlbench
*
* 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.
*/
import io.nosqlbench.vectordata.VectorDataSettings;
import io.nosqlbench.vectordata.binding.BindType;
import io.nosqlbench.vectordata.binding.Binder;
import io.nosqlbench.virtdata.api.annotations.Categories;
import io.nosqlbench.virtdata.api.annotations.Category;
import io.nosqlbench.virtdata.api.annotations.Example;
import io.nosqlbench.virtdata.api.annotations.ThreadSafeMapper;
/*
* Projects one named field of each record in a facet of opaque records
* — a slab of metadata — to the Java value its wire type says it is:
* integers by width, floats by width, text as String, timestamps as
* Instant, UUIDs as UUID, blobs as byte[], containers as lists, sets,
* and maps. The field is resolved to its position once, when the
* function is built; a field the facet lacks is refused then, naming
* the fields it has. The cycle value is the record ordinal. */
@ThreadSafeMapper
@Categories(Category.vectors)
public class RecordField extends CoreRecords<Object> {
private final Binder binder;
@Example({"RecordField('exampledataset:exampleprofile','metadata_content','id')",
"Bind the 'id' field of each metadata record, typed as the record types it"})
public RecordField(String datasetAndProfile, String facetName, String field) {
this(datasetAndProfile, facetName, field, VectorDataSettings.defaults());
}
/** Construct with explicit vectordata settings, including an isolated cache location. */
public RecordField(String datasetAndProfile, String facetName, String field, VectorDataSettings settings) {
super(datasetAndProfile, facetName, settings);
binder = Binder.select(layout, field);
}
/// What the projected value binds as, from the field's tag.
public BindType bindType() { return binder.types().get(0); }
@Override
public Object apply(long ordinal) {
return RecordProjections.of(binder.bind(facet.recordBytes(ordinal)).get(0));
}
}
@@ -0,0 +1,102 @@
package io.nosqlbench.virtdata.lib.vectors.vectordata;
/*
* Copyright (c) nosqlbench
*
* 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.
*/
import io.nosqlbench.vectordata.VectorDataSettings;
import io.nosqlbench.vectordata.binding.BindType;
import io.nosqlbench.vectordata.binding.Binder;
import io.nosqlbench.vectordata.binding.Layout;
import io.nosqlbench.vectordata.records.RecordFacet;
import io.nosqlbench.virtdata.api.annotations.Categories;
import io.nosqlbench.virtdata.api.annotations.Category;
import io.nosqlbench.virtdata.api.annotations.Example;
import io.nosqlbench.virtdata.api.annotations.ThreadSafeMapper;
import java.util.ArrayList;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
/*
* Binds each record of a facet of opaque records — a slab of metadata —
* to a map of parameter name to Java value, in bind order: every field
* under its own name, or a comma-separated selection in the order
* given. Names are resolved to positions once, when the function is
* built, so the per-cycle path reads values in place and never looks
* at a field name. The cycle value is the record ordinal. */
@ThreadSafeMapper
@Categories(Category.vectors)
public class RecordFields extends CoreRecords<Map<String, Object>> {
/// Compiles the binder a function drives, against the facet and
/// the layout learned from it.
@FunctionalInterface
interface Compilation { Binder compile(RecordFacet facet, Layout layout); }
private final Binder binder;
@Example({"RecordFields('exampledataset:exampleprofile','metadata_content')",
"Bind every field of each metadata record, under its own name, in the record's order"})
public RecordFields(String datasetAndProfile, String facetName) {
this(datasetAndProfile, facetName, "", VectorDataSettings.defaults());
}
@Example({"RecordFields('exampledataset:exampleprofile','metadata_content','id,tag')",
"Bind the named fields, in the order given"})
public RecordFields(String datasetAndProfile, String facetName, String fields) {
this(datasetAndProfile, facetName, fields, VectorDataSettings.defaults());
}
/** Construct with explicit vectordata settings, including an isolated cache location. */
public RecordFields(String datasetAndProfile, String facetName, String fields, VectorDataSettings settings) {
this(datasetAndProfile, facetName, settings, (facet, layout) -> {
List<String> names = split(fields);
return names.isEmpty() ? Binder.all(layout) : Binder.select(layout, names);
});
}
RecordFields(String datasetAndProfile, String facetName, VectorDataSettings settings, Compilation compilation) {
super(datasetAndProfile, facetName, settings);
binder = compilation.compile(facet, layout);
}
static List<String> split(String fields) {
List<String> names = new ArrayList<>();
if (fields == null) return names;
for (String name : fields.split(",")) if (!name.isBlank()) names.add(name.trim());
return names;
}
/// Parameter names, in bind order — the keys of every map this
/// function returns. Read once, to prepare a statement.
public List<String> parameters() { return binder.parameters(); }
/// What each parameter binds as, in bind order.
public List<BindType> types() { return binder.types(); }
@Override
public Map<String, Object> apply(long ordinal) {
List<String> parameters = binder.parameters();
Object[] values = new Object[parameters.size()];
binder.bindEach(facet.recordBytes(ordinal), (slot, field) -> values[slot] = RecordProjections.of(field));
Map<String, Object> bound = new LinkedHashMap<>(values.length * 2);
for (int slot = 0; slot < values.length; slot++) bound.put(parameters.get(slot), values[slot]);
return bound;
}
}
@@ -0,0 +1,64 @@
package io.nosqlbench.virtdata.lib.vectors.vectordata;
/*
* Copyright (c) nosqlbench
*
* 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.
*/
import io.nosqlbench.vectordata.VectorDataSettings;
import io.nosqlbench.vectordata.binding.Form;
import io.nosqlbench.vectordata.binding.Forms;
import io.nosqlbench.virtdata.api.annotations.Categories;
import io.nosqlbench.virtdata.api.annotations.Category;
import io.nosqlbench.virtdata.api.annotations.Example;
import io.nosqlbench.virtdata.api.annotations.ThreadSafeMapper;
/*
* Binds each record of a facet of opaque records under a form the
* facet declares in its 'forms' namespace — the fields that form
* binds, in its order, under the parameter names it gives them. A
* facet declaring no forms offers one implicit form, 'default', which
* binds every field under its own name; a form the facet does not
* offer is refused when the function is built, naming the ones it
* does. The cycle value is the record ordinal. */
@ThreadSafeMapper
@Categories(Category.vectors)
public class RecordForm extends RecordFields {
private final Form form;
@Example({"RecordForm('exampledataset:exampleprofile','metadata_content','row')",
"Bind each metadata record as the facet's 'row' form declares"})
public RecordForm(String datasetAndProfile, String facetName, String formName) {
this(datasetAndProfile, facetName, formName, VectorDataSettings.defaults());
}
/** Construct with explicit vectordata settings, including an isolated cache location. */
public RecordForm(String datasetAndProfile, String facetName, String formName, VectorDataSettings settings) {
this(datasetAndProfile, facetName, settings, new Form[1], formName);
}
private RecordForm(String datasetAndProfile, String facetName, VectorDataSettings settings, Form[] selected, String formName) {
super(datasetAndProfile, facetName, settings, (facet, layout) -> {
selected[0] = Forms.byName(facet, formName);
return selected[0].binder(layout);
});
form = selected[0];
}
/// The form this function binds under, as the facet declares it.
public Form form() { return form; }
}
@@ -0,0 +1,165 @@
package io.nosqlbench.virtdata.lib.vectors.vectordata;
/*
* Copyright (c) nosqlbench
*
* 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.
*/
import io.nosqlbench.vectordata.anode.Field;
import io.nosqlbench.vectordata.anode.MNode;
import io.nosqlbench.vectordata.anode.MValue;
import io.nosqlbench.vectordata.anode.PNode.Comparand;
import io.nosqlbench.vectordata.anode.TypeTag;
import io.nosqlbench.vectordata.binding.BindType;
import java.math.BigDecimal;
import java.math.BigInteger;
import java.nio.ByteBuffer;
import java.time.Instant;
import java.util.ArrayList;
import java.util.LinkedHashMap;
import java.util.LinkedHashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.UUID;
/// Projects bound values to the Java types a driver binds: what a value
/// **is**, by its bind type, rather than how a vernacular would print
/// it. Integers by width (`Short`, `Integer`, `Long`), floats by width
/// (`Float` for binary16 and binary32, `Double`), text families as
/// `String`, the numeric timestamps as [Instant], the UUIDs as
/// [UUID], a ULID and a blob as `byte[]`, and containers as
/// insertion-ordered lists, sets, and maps of the same projections.
///
/// Varint and decimal payloads are projected as [BigInteger] and
/// [BigDecimal] on the assumption that they carry big-endian two's
/// complement digits — the encoding the CQL types they are named after
/// use. The reference reads both as opaque bytes and states no
/// convention; a dataset written otherwise would need its own
/// projection.
public final class RecordProjections {
private RecordProjections() { }
/// A bound field as a Java value, read in place where the family
/// allows and through the AST view for containers.
public static Object of(Field field) {
TypeTag tag = field.typeTag();
if (tag == null) { ByteBuffer raw = field.valueBuffer(); byte[] copy = new byte[raw.remaining()]; raw.get(copy); return copy; }
return switch (tag) {
case NULL -> null;
case SHORT -> (short) field.longValue();
case INT32, ENUM_ORD -> (int) field.longValue();
case INT -> field.longValue();
case HALF, FLOAT32 -> (float) field.doubleValue();
case FLOAT -> field.doubleValue();
case BOOL -> field.booleanValue();
case TEXT, TEXT_VALIDATED, ENUM_STR, ASCII, DATE, TIME, DATETIME -> field.stringValue();
case BYTES, ULID -> field.bytesValue();
case VARINT -> bigInteger(field.bytesValue());
case DECIMAL -> new BigDecimal(bigInteger(field.decimalDigits()), field.decimalScale());
case MILLIS -> Instant.ofEpochMilli(field.longValue());
case NANOS -> Instant.ofEpochSecond(field.nanosSeconds(), field.nanosAdjust());
case UUID_V1, UUID_V7 -> uuid(field.bytesValue());
case LIST, ARRAY, SET, MAP, TYPED_MAP -> of(field.value());
};
}
/// A node value as a Java value, recursively.
public static Object of(MValue value) {
return switch (value.tag()) {
case NULL -> null;
case TEXT -> ((MValue.Text) value).value();
case ENUM_STR -> ((MValue.EnumStr) value).value();
case ASCII -> ((MValue.Ascii) value).value();
case DATE -> ((MValue.Date) value).value();
case TIME -> ((MValue.Time) value).value();
case DATETIME -> ((MValue.DateTime) value).value();
case INT -> ((MValue.Int) value).value();
case INT32 -> ((MValue.Int32) value).value();
case ENUM_ORD -> ((MValue.EnumOrd) value).value();
case SHORT -> ((MValue.Short) value).value();
case FLOAT -> ((MValue.Float) value).value();
case FLOAT32 -> ((MValue.Float32) value).value();
case HALF -> Float.float16ToFloat((short) ((MValue.Half) value).bits());
case BOOL -> ((MValue.Bool) value).value();
case BYTES -> ((MValue.Bytes) value).value();
case MILLIS -> Instant.ofEpochMilli(((MValue.Millis) value).value());
case NANOS -> { MValue.Nanos n = (MValue.Nanos) value; yield Instant.ofEpochSecond(n.epochSeconds(), n.nanoAdjust()); }
case UUID_V1 -> uuid(((MValue.UuidV1) value).value());
case UUID_V7 -> uuid(((MValue.UuidV7) value).value());
case ULID -> ((MValue.Ulid) value).value();
case LIST -> list(((MValue.ListValue) value).items());
case ARRAY -> list(((MValue.ArrayValue) value).items());
case SET -> { Set<Object> out = new LinkedHashSet<>(); for (MValue item : ((MValue.SetValue) value).items()) out.add(of(item)); yield out; }
case MAP -> map(((MValue.MapValue) value).node());
case TYPED_MAP -> { Map<Object, Object> out = new LinkedHashMap<>(); for (MValue.TypedMap.Entry e : ((MValue.TypedMap) value).entries()) out.put(of(e.key()), of(e.value())); yield out; }
case TEXT_VALIDATED, DECIMAL, VARINT -> throw new IllegalStateException("no node value carries tag " + value.tag());
};
}
/// A record as an insertion-ordered map of projected values.
public static Map<String, Object> map(MNode node) {
Map<String, Object> out = new LinkedHashMap<>();
node.fields().forEach((name, value) -> out.put(name, of(value)));
return out;
}
/// A predicate comparand as a Java value, coerced toward the
/// **field's** bind type where its own variant allows: an `Int`
/// comparand against a `TIMESTAMP_MILLIS` field becomes an
/// [Instant], against an `INT32` field an `Integer`; a `Bytes`
/// comparand against a `UUID` field becomes a [UUID]. A comparand
/// the type cannot claim keeps its natural projection.
public static Object of(Comparand comparand, BindType type) {
BindType.Kind kind = type == null ? BindType.Kind.NULL : type.kind();
if (comparand instanceof Comparand.Null) return null;
if (comparand instanceof Comparand.Int i) {
return switch (kind) {
case INT16 -> (short) i.value();
case INT32 -> (int) i.value();
case FLOAT16, FLOAT32 -> (float) i.value();
case FLOAT64 -> (double) i.value();
case TIMESTAMP_MILLIS -> Instant.ofEpochMilli(i.value());
case TIMESTAMP_NANOS -> Instant.ofEpochSecond(i.value());
case VARINT -> BigInteger.valueOf(i.value());
case DECIMAL -> BigDecimal.valueOf(i.value());
default -> i.value();
};
}
if (comparand instanceof Comparand.Float f) {
return switch (kind) {
case FLOAT16, FLOAT32 -> (float) f.value();
case DECIMAL -> BigDecimal.valueOf(f.value());
default -> f.value();
};
}
if (comparand instanceof Comparand.Text t) return t.value();
if (comparand instanceof Comparand.Bool b) return b.value();
byte[] bytes = ((Comparand.Bytes) comparand).value();
return switch (kind) {
case UUID, TIME_UUID -> bytes.length == 16 ? uuid(bytes) : bytes;
case VARINT -> bigInteger(bytes);
default -> bytes;
};
}
private static List<Object> list(List<MValue> items) { List<Object> out = new ArrayList<>(items.size()); for (MValue item : items) out.add(of(item)); return out; }
private static BigInteger bigInteger(byte[] twosComplement) { return twosComplement.length == 0 ? BigInteger.ZERO : new BigInteger(twosComplement); }
private static UUID uuid(byte[] sixteen) { ByteBuffer b = ByteBuffer.wrap(sixteen); return new UUID(b.getLong(), b.getLong()); }
}
@@ -0,0 +1,254 @@
package io.nosqlbench.virtdata.lib.vectors.vectordata;
/*
* Copyright (c) nosqlbench
*
* 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.
*/
import io.nosqlbench.vectordata.VectorDataSettings;
import io.nosqlbench.vectordata.anode.MNode;
import io.nosqlbench.vectordata.anode.MValue;
import io.nosqlbench.vectordata.anode.PNode;
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 io.nosqlbench.vectordata.binding.BindException;
import io.nosqlbench.vectordata.binding.BindType;
import org.junit.jupiter.api.AfterAll;
import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.Tag;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.io.TempDir;
import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.nio.ByteOrder;
import java.nio.charset.StandardCharsets;
import java.nio.file.Files;
import java.nio.file.Path;
import java.time.Instant;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.UUID;
import static org.junit.jupiter.api.Assertions.*;
/// The record-binding functions: one field projected by its wire type,
/// a record as named values in bind order, a record under a declared
/// form, and a predicate's comparands as parameters typed from the
/// metadata fields they constrain. The fixture is a local dataset with
/// a metadata slab, its forms namespace, and a predicate slab.
@Tag("unit")
public class RecordBindingFunctionsTest {
@TempDir static Path temporary;
static VectorDataSettings settings;
static String priorCatalog;
static final byte[] OWNER = {0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15};
static final long EPOCH_MILLIS = 1_700_000_000_000L;
static byte[] row(int i) {
return new MNode()
.insert("id", new MValue.Int(i))
.insert("tag", new MValue.Text("t" + i % 3))
.insert("score", new MValue.Float(i * 0.5))
.insert("created", new MValue.Millis(EPOCH_MILLIS + i))
.insert("owner", new MValue.UuidV7(OWNER))
.insert("flags", new MValue.ListValue(List.of(new MValue.Int(i), new MValue.Text("x"))))
.insert("h", new MValue.Half(0x3C00))
.insert("n", new MValue.Nanos(1_700_000_000L + i, 500))
.insert("count", new MValue.Int32(i * 2))
.toBytes();
}
static byte[] predicate(int i) {
return new PNode.Conjugate(ConjugateType.AND, List.of(
new PNode.Predicate(new FieldRef.Named("created"), OpType.GE, List.of(new Comparand.Int(EPOCH_MILLIS + i))),
new PNode.Predicate(new FieldRef.Named("tag"), OpType.EQ, List.of(new Comparand.Text("t" + i % 3))))).toBytesNamed();
}
@BeforeAll
static void publishFixture() throws IOException {
Path dataset = Files.createDirectories(temporary.resolve("records"));
fvec(dataset.resolve("base.fvec"), new float[][] {{0f, 1f}});
List<byte[]> rows = new ArrayList<>(), predicates = new ArrayList<>();
for (int i = 0; i < 20; i++) { rows.add(row(i)); predicates.add(predicate(i)); }
List<byte[]> forms = List.of(
"{\"name\":\"row\",\"operation\":\"insert\",\"fields\":[\"id\",\"tag\"]}".getBytes(StandardCharsets.UTF_8),
"{\"name\":\"key\",\"operation\":\"get\",\"fields\":[\"id\"],\"parameters\":{\"id\":\"pk\"}}".getBytes(StandardCharsets.UTF_8));
slab(dataset.resolve("metadata_content.slab"), rows, Map.of("forms", forms), 4);
slab(dataset.resolve("metadata_predicates.slab"), predicates, Map.of(), 4);
Files.writeString(dataset.resolve("dataset.yaml"), """
name: records
profiles:
default:
base_vectors: base.fvec
metadata_content: metadata_content.slab
metadata_predicates: metadata_predicates.slab
""");
Files.writeString(temporary.resolve("catalog.yaml"), """
datasets:
- name: records
path: records/dataset.yaml
""");
settings = VectorDataSettings.builder().cacheDirectory(temporary.resolve("cache")).build();
priorCatalog = System.getProperty("vectordata.catalog");
System.setProperty("vectordata.catalog", temporary.resolve("catalog.yaml").toString());
}
@AfterAll
static void restoreCatalogProperty() {
if (priorCatalog == null) System.clearProperty("vectordata.catalog");
else System.setProperty("vectordata.catalog", priorCatalog);
}
@Test
void recordFieldProjectsByWireType() {
assertEquals(5L, new RecordField("records:default", "metadata_content", "id", settings).apply(5));
assertEquals("t2", new RecordField("records:default", "metadata_content", "tag", settings).apply(5));
assertEquals(2.5, new RecordField("records:default", "metadata_content", "score", settings).apply(5));
RecordField created = new RecordField("records:default", "metadata_content", "created", settings);
assertEquals(Instant.ofEpochMilli(EPOCH_MILLIS + 5), created.apply(5), "a Millis is an instant, not a long");
assertEquals(BindType.TIMESTAMP_MILLIS, created.bindType());
assertEquals(new UUID(0x0001020304050607L, 0x08090a0b0c0d0e0fL), new RecordField("records:default", "metadata_content", "owner", settings).apply(0));
assertEquals(List.of(5L, "x"), new RecordField("records:default", "metadata_content", "flags", settings).apply(5), "a container projects its elements");
assertEquals(1.0f, new RecordField("records:default", "metadata_content", "h", settings).apply(0), "a Half is a float, not its bit pattern");
assertEquals(Instant.ofEpochSecond(1_700_000_000L + 5, 500), new RecordField("records:default", "metadata_content", "n", settings).apply(5));
assertEquals(10, new RecordField("records:default", "metadata_content", "count", settings).apply(5), "an Int32 is an Integer");
}
@Test
void recordFieldsBindsEveryFieldOrASelection() {
RecordFields all = new RecordFields("records:default", "metadata_content", "", settings);
assertEquals(List.of("id", "tag", "score", "created", "owner", "flags", "h", "n", "count"), all.parameters());
Map<String, Object> bound = all.apply(3);
assertEquals(all.parameters(), new ArrayList<>(bound.keySet()), "keys in bind order");
assertEquals(3L, bound.get("id"));
assertEquals(20, all.count());
// A selection, in template order, through a facet alias and with
// whitespace around the names.
RecordFields some = new RecordFields("records:default", "content", "tag, id", settings);
assertEquals(List.of("tag", "id"), some.parameters());
assertEquals(List.of(BindType.TEXT, BindType.INT64), some.types());
assertEquals(Map.of("tag", "t1", "id", 7L), some.apply(7));
}
@Test
void recordFormBindsAsTheFacetDeclares() {
RecordForm key = new RecordForm("records:default", "metadata_content", "key", settings);
assertEquals(List.of("pk"), key.parameters(), "the form's parameter name, not the field's");
assertEquals(Map.of("pk", 7L), key.apply(7));
assertEquals("get", key.form().operation());
RecordForm row = new RecordForm("records:default", "metadata_content", "row", settings);
assertEquals(List.of("id", "tag"), row.parameters());
assertEquals(Map.of("id", 2L, "tag", "t2"), row.apply(2));
BindException e = assertThrows(BindException.class, () -> new RecordForm("records:default", "metadata_content", "document", settings));
assertEquals(BindException.Kind.NO_SUCH_FORM, e.kind());
assertEquals(List.of("row", "key"), e.available(), "refused naming what is offered");
}
@Test
void predicateParamsBindInTheShapeOfTheFirstPredicate() {
PredicateParams params = new PredicateParams("records:default", "metadata_predicates", "metadata_content", "", settings);
assertEquals(List.of("created", "tag"), params.parameters());
assertEquals(OpType.GE, params.conditions().get(0).op());
assertEquals(BindType.TIMESTAMP_MILLIS, params.conditions().get(0).bindType(), "typed from the field, not the Int comparand");
Map<String, Object> bound = params.apply(4);
assertEquals(List.of("created", "tag"), new ArrayList<>(bound.keySet()));
assertEquals(Instant.ofEpochMilli(EPOCH_MILLIS + 4), bound.get("created"), "the comparand is coerced to the field's type");
assertEquals("t1", bound.get("tag"));
}
@Test
void predicateParamsAcceptATemplateAndRefuseAnotherShape() {
PredicateParams params = new PredicateParams("records:default", "metadata_predicates", "metadata_content", "(created >= 0 AND tag = '')", settings);
assertEquals(List.of("created", "tag"), params.parameters());
assertEquals("t2", params.apply(2).get("tag"));
PredicateParams other = new PredicateParams("records:default", "metadata_predicates", "metadata_content", "id = 0", settings);
BindException e = assertThrows(BindException.class, () -> other.apply(0));
assertTrue(e.getMessage().contains("shape differs"), e.getMessage());
}
@Test
void anUnknownFieldIsRefusedWhenTheFunctionIsBuilt() {
BindException e = assertThrows(BindException.class, () -> new RecordField("records:default", "metadata_content", "nope", settings));
assertEquals(BindException.Kind.NO_SUCH_FIELD, e.kind());
assertEquals("nope", e.name());
assertTrue(e.available().contains("id"), "naming what the facet has");
}
// ── fixture writers ────────────────────────────────────────────
static void fvec(Path path, float[][] vectors) throws IOException {
ByteBuffer out = ByteBuffer.allocate(vectors.length * (4 + 4 * vectors[0].length)).order(ByteOrder.LITTLE_ENDIAN);
for (float[] v : vectors) { out.putInt(v.length); for (float f : v) out.putFloat(f); }
Files.write(path, out.array());
}
/// A slab in the reference layout: data pages of `perPage` records,
/// a pages page per namespace, and — when named namespaces exist —
/// a namespaces page locating each pages page.
static void slab(Path path, List<byte[]> records, Map<String, List<byte[]>> namespaces, int perPage) throws IOException {
ByteArrayOutputStream out = new ByteArrayOutputStream();
List<byte[]> index = new ArrayList<>();
dataPages(out, index, records, perPage, 1);
if (namespaces.isEmpty()) out.write(page(index, 0, 1, 1));
else {
List<byte[]> entries = new ArrayList<>();
long at = out.size(); out.write(page(index, 0, 1, 1)); entries.add(namespaceEntry(1, "", at));
int namespaceIndex = 2;
for (Map.Entry<String, List<byte[]>> namespace : namespaces.entrySet()) {
List<byte[]> pages = new ArrayList<>();
dataPages(out, pages, namespace.getValue(), perPage, namespaceIndex);
long pagesAt = out.size(); out.write(page(pages, 0, 1, namespaceIndex));
entries.add(namespaceEntry(namespaceIndex, namespace.getKey(), pagesAt));
namespaceIndex++;
}
out.write(page(entries, 0, 3, 1));
}
Files.write(path, out.toByteArray());
}
private static void dataPages(ByteArrayOutputStream out, List<byte[]> index, List<byte[]> records, int perPage, int namespaceIndex) throws IOException {
for (int start = 0; start < records.size(); start += perPage) {
ByteBuffer entry = ByteBuffer.allocate(16).order(ByteOrder.LITTLE_ENDIAN); entry.putLong(start).putLong(out.size()); index.add(entry.array());
out.write(page(records.subList(start, Math.min(records.size(), start + perPage)), start, 2, namespaceIndex));
}
}
private static byte[] namespaceEntry(int index, String name, long pagesPageOffset) {
byte[] utf8 = name.getBytes(StandardCharsets.UTF_8);
ByteBuffer entry = ByteBuffer.allocate(2 + utf8.length + 8).order(ByteOrder.LITTLE_ENDIAN);
entry.put((byte) index).put((byte) utf8.length).put(utf8).putLong(pagesPageOffset);
return entry.array();
}
private static byte[] page(List<byte[]> records, long startOrdinal, int pageType, int namespaceIndex) {
int data = 0; for (byte[] record : records) data += record.length;
int size = 8 + data + 4 * (records.size() + 1) + 16;
ByteBuffer page = ByteBuffer.allocate(size).order(ByteOrder.LITTLE_ENDIAN);
page.put(new byte[] {'S', 'L', 'A', 'B'}).putInt(size);
for (byte[] record : records) page.put(record);
int offset = 8; for (byte[] record : records) { page.putInt(offset); offset += record.length; } page.putInt(offset);
for (int i = 0; i < 5; i++) page.put((byte) (startOrdinal >>> (8 * i)));
int count = records.size(); page.put((byte) count).put((byte) (count >>> 8)).put((byte) (count >>> 16));
page.putInt(size).put((byte) pageType).put((byte) namespaceIndex).putShort((short) 16);
return page.array();
}
}