From 62b49e8a524e2c56aaa7b30c1abc201d28295cc6 Mon Sep 17 00:00:00 2001 From: Jonathan Shook Date: Fri, 4 Sep 2026 23:34:00 +0000 Subject: [PATCH] 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. --- .../lib/vectors/vectordata/CoreRecords.java | 64 +++++ .../vectors/vectordata/PredicateParams.java | 122 +++++++++ .../lib/vectors/vectordata/RecordField.java | 62 +++++ .../lib/vectors/vectordata/RecordFields.java | 102 +++++++ .../lib/vectors/vectordata/RecordForm.java | 64 +++++ .../vectors/vectordata/RecordProjections.java | 165 ++++++++++++ .../RecordBindingFunctionsTest.java | 254 ++++++++++++++++++ 7 files changed, 833 insertions(+) create mode 100644 nb-virtdata/virtdata-lib-vectors/src/main/java/io/nosqlbench/virtdata/lib/vectors/vectordata/CoreRecords.java create mode 100644 nb-virtdata/virtdata-lib-vectors/src/main/java/io/nosqlbench/virtdata/lib/vectors/vectordata/PredicateParams.java create mode 100644 nb-virtdata/virtdata-lib-vectors/src/main/java/io/nosqlbench/virtdata/lib/vectors/vectordata/RecordField.java create mode 100644 nb-virtdata/virtdata-lib-vectors/src/main/java/io/nosqlbench/virtdata/lib/vectors/vectordata/RecordFields.java create mode 100644 nb-virtdata/virtdata-lib-vectors/src/main/java/io/nosqlbench/virtdata/lib/vectors/vectordata/RecordForm.java create mode 100644 nb-virtdata/virtdata-lib-vectors/src/main/java/io/nosqlbench/virtdata/lib/vectors/vectordata/RecordProjections.java create mode 100644 nb-virtdata/virtdata-lib-vectors/src/test/java/io/nosqlbench/virtdata/lib/vectors/vectordata/RecordBindingFunctionsTest.java diff --git a/nb-virtdata/virtdata-lib-vectors/src/main/java/io/nosqlbench/virtdata/lib/vectors/vectordata/CoreRecords.java b/nb-virtdata/virtdata-lib-vectors/src/main/java/io/nosqlbench/virtdata/lib/vectors/vectordata/CoreRecords.java new file mode 100644 index 000000000..76f3366e6 --- /dev/null +++ b/nb-virtdata/virtdata-lib-vectors/src/main/java/io/nosqlbench/virtdata/lib/vectors/vectordata/CoreRecords.java @@ -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 implements LongFunction { + + 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(); } +} diff --git a/nb-virtdata/virtdata-lib-vectors/src/main/java/io/nosqlbench/virtdata/lib/vectors/vectordata/PredicateParams.java b/nb-virtdata/virtdata-lib-vectors/src/main/java/io/nosqlbench/virtdata/lib/vectors/vectordata/PredicateParams.java new file mode 100644 index 000000000..c22222d36 --- /dev/null +++ b/nb-virtdata/virtdata-lib-vectors/src/main/java/io/nosqlbench/virtdata/lib/vectors/vectordata/PredicateParams.java @@ -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> { + + 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 conditions() { return binder.conditions(); } + + /// The parameter names, in condition order — the keys of every map + /// this function returns. + public List parameters() { + List names = new ArrayList<>(); + for (Condition c : binder.conditions()) names.add(c.parameter()); + return names; + } + + @Override + public Map apply(long ordinal) { + Map 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 comparands) { + if (condition.op() != OpType.IN && comparands.size() == 1) return RecordProjections.of(comparands.get(0), condition.bindType()); + List values = new ArrayList<>(comparands.size()); + for (Comparand c : comparands) values.add(RecordProjections.of(c, condition.bindType())); + return values; + } +} diff --git a/nb-virtdata/virtdata-lib-vectors/src/main/java/io/nosqlbench/virtdata/lib/vectors/vectordata/RecordField.java b/nb-virtdata/virtdata-lib-vectors/src/main/java/io/nosqlbench/virtdata/lib/vectors/vectordata/RecordField.java new file mode 100644 index 000000000..091f8aa83 --- /dev/null +++ b/nb-virtdata/virtdata-lib-vectors/src/main/java/io/nosqlbench/virtdata/lib/vectors/vectordata/RecordField.java @@ -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 { + + 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)); + } +} diff --git a/nb-virtdata/virtdata-lib-vectors/src/main/java/io/nosqlbench/virtdata/lib/vectors/vectordata/RecordFields.java b/nb-virtdata/virtdata-lib-vectors/src/main/java/io/nosqlbench/virtdata/lib/vectors/vectordata/RecordFields.java new file mode 100644 index 000000000..1b219c4e5 --- /dev/null +++ b/nb-virtdata/virtdata-lib-vectors/src/main/java/io/nosqlbench/virtdata/lib/vectors/vectordata/RecordFields.java @@ -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> { + + /// 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 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 split(String fields) { + List 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 parameters() { return binder.parameters(); } + + /// What each parameter binds as, in bind order. + public List types() { return binder.types(); } + + @Override + public Map apply(long ordinal) { + List parameters = binder.parameters(); + Object[] values = new Object[parameters.size()]; + binder.bindEach(facet.recordBytes(ordinal), (slot, field) -> values[slot] = RecordProjections.of(field)); + Map bound = new LinkedHashMap<>(values.length * 2); + for (int slot = 0; slot < values.length; slot++) bound.put(parameters.get(slot), values[slot]); + return bound; + } +} diff --git a/nb-virtdata/virtdata-lib-vectors/src/main/java/io/nosqlbench/virtdata/lib/vectors/vectordata/RecordForm.java b/nb-virtdata/virtdata-lib-vectors/src/main/java/io/nosqlbench/virtdata/lib/vectors/vectordata/RecordForm.java new file mode 100644 index 000000000..a7722c588 --- /dev/null +++ b/nb-virtdata/virtdata-lib-vectors/src/main/java/io/nosqlbench/virtdata/lib/vectors/vectordata/RecordForm.java @@ -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; } +} diff --git a/nb-virtdata/virtdata-lib-vectors/src/main/java/io/nosqlbench/virtdata/lib/vectors/vectordata/RecordProjections.java b/nb-virtdata/virtdata-lib-vectors/src/main/java/io/nosqlbench/virtdata/lib/vectors/vectordata/RecordProjections.java new file mode 100644 index 000000000..55ba69075 --- /dev/null +++ b/nb-virtdata/virtdata-lib-vectors/src/main/java/io/nosqlbench/virtdata/lib/vectors/vectordata/RecordProjections.java @@ -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 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 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 map(MNode node) { + Map 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 list(List items) { List 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()); } +} diff --git a/nb-virtdata/virtdata-lib-vectors/src/test/java/io/nosqlbench/virtdata/lib/vectors/vectordata/RecordBindingFunctionsTest.java b/nb-virtdata/virtdata-lib-vectors/src/test/java/io/nosqlbench/virtdata/lib/vectors/vectordata/RecordBindingFunctionsTest.java new file mode 100644 index 000000000..b630f8b7b --- /dev/null +++ b/nb-virtdata/virtdata-lib-vectors/src/test/java/io/nosqlbench/virtdata/lib/vectors/vectordata/RecordBindingFunctionsTest.java @@ -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 rows = new ArrayList<>(), predicates = new ArrayList<>(); + for (int i = 0; i < 20; i++) { rows.add(row(i)); predicates.add(predicate(i)); } + List 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 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 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 records, Map> namespaces, int perPage) throws IOException { + ByteArrayOutputStream out = new ByteArrayOutputStream(); + List index = new ArrayList<>(); + dataPages(out, index, records, perPage, 1); + if (namespaces.isEmpty()) out.write(page(index, 0, 1, 1)); + else { + List 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> namespace : namespaces.entrySet()) { + List 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 index, List 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 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(); + } +}