feat(vectordata): read a facet spread across several files

- a facet may be a series: the uniform form names `base__NNNN.fvec`
  with shard_stride, shard_count, and record_count; the explicit form
  lists entries in the source grammar, each bare, `=`-counted,
  windowed, or both, checked against each other
- ordinals map in O(1) whenever the lengths are uniform however the
  series was spelled, and by prefix-sum search otherwise; an entry
  window is in that file's ordinals, the facet window is in facet
  ordinals, and a suffix on a uniform pattern is the facet window
- storage, sidecars, and offset indexes are per file, so two shards of
  one file share one of each; the readers present a single surface,
  refuse shards that disagree on shape, and name a missing shard
  rather than reading a shorter facet
- prefetch plans carry ShardRange in place of ByteRange, decompose a
  window across the shards it spans, never merge across shards, and
  sum prerequisite bytes per file; a whole-facet request names every
  shard, and cache statistics report the weakest access mode
- format_version is a minimum reader requirement: a dataset above 2,
  or one stating less than its content needs, is refused naming both
  numbers; declarations are checked for self-consistency at load, and
  a directory falls back to knn_entries.yaml only when dataset.yaml is
  absent, so a refusal keeps its diagnosis
- a series holds open no more files than a quarter of the descriptor
  soft limit, floor 8, or VECTORDATA_SHARD_FD_CAP
This commit is contained in:
Jonathan Shook committed 2026-09-04 21:17:39 +00:00
1 parent f999cce819
commit 19f04125ee
20 files changed
+2301 -341

No files matched your search

@@ -15,5 +15,25 @@
*/
package io.nosqlbench.vectordata;
/** Backing mode selected for a source. */
public enum AccessMode { LOCAL, MERKLE_HASHED, MERKLE_CHUNKED, FULL_TRANSFER }
/// Backing mode selected for a source. Ranked by what a caller must plan
/// around — a full transfer constrains most, a local file least, and the
/// two chunked modes differ in trust rather than access — so that a facet
/// spanning several files can report the [#weakest] of them.
public enum AccessMode {
LOCAL(3), MERKLE_HASHED(2), MERKLE_CHUNKED(1), FULL_TRANSFER(0);
private final int strength;
AccessMode(int strength) { this.strength = strength; }
/// The weakest of several modes — the one true of every file. A
/// facet's mode is a promise about every read it will serve, and a
/// caller that plans against "supports range" and then reaches a
/// file that does not has been misled by an average. Understating
/// the good files costs efficiency; overstating costs correctness.
/// `null` for no modes: making no promise is not promising little.
public static AccessMode weakest(Iterable<AccessMode> modes) {
AccessMode result = null;
for (AccessMode mode : modes) if (result == null || mode.strength < result.strength) result = mode;
return result;
}
}
@@ -1,23 +0,0 @@
/*
* 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;
/// A half-open byte range `[start, end)` within a facet's payload.
public record ByteRange(long start, long end) {
/// Bytes spanned by this range.
public long length() { return Math.max(0, end - start); }
@Override public String toString() { return "[" + start + ".." + end + ")"; }
}
@@ -16,7 +16,35 @@
package io.nosqlbench.vectordata;
import java.net.URI;
import java.util.List;
import java.util.Map;
/** Resolved manifest facet, including custom application-defined fields. */
public record FacetDescriptor(String name, URI source, String window, Map<String, Object> attributes) { }
/// Resolved manifest facet, including custom application-defined
/// fields.
///
/// `source` is the facet's one file, resolved against its manifest, or
/// `null` for a multi-file [Series]: a caller written before sharding
/// must fail visibly rather than read a series as its first file — or,
/// for the uniform form, as the `NNNN` *pattern*, which names no file at
/// all. `window` is the **facet** window, in facet ordinals: a suffix on
/// a single source, or the `window:` field a series carries, having no
/// single source to put one on.
public record FacetDescriptor(String name, URI source, String window, Map<String, Object> attributes, Series series) {
public FacetDescriptor(String name, URI source, String window, Map<String, Object> attributes) {
this(name, source, window, attributes, null);
}
/// A facet declared as a series of files. `entries` are the source
/// strings in ordinal order, resolved against the manifest: one
/// `NNNN` pattern for the uniform form, one per shard for the
/// explicit form — each carrying its own window or `=count` suffix
/// exactly as written, because an entry window is in that *file's*
/// ordinals and must stay with the entry it bounds.
public record Series(List<String> entries, boolean declaredAsArray, Long shardStride, Integer shardCount, Long recordCount) {
public Series { entries = List.copyOf(entries); }
}
/// Whether this facet spans more than one file.
public boolean isSeries() { return series != null; }
}
@@ -0,0 +1,55 @@
/*
* 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;
/// The `dataset.yaml` format version: a **minimum reader requirement**,
/// not a timestamp. Absent means [#BASE], which is every dataset in
/// circulation and not a distinct unversioned state. A writer emits the
/// lowest version describing what it wrote, so only a change that older
/// readers would misread — multi-file facets were the first — bumps it.
public final class FormatVersion {
private FormatVersion() { }
/// Everything written before the field existed: single-file facets.
public static final int BASE = 1;
/// Multi-file facet series.
public static final int SHARDED = 2;
/// The highest version this implementation reads.
public static final int SUPPORTED = SHARDED;
/// Refuses a dataset this implementation cannot read, naming both
/// numbers, and returns the effective version otherwise. Shared by
/// every loader rather than mirrored, so a dataset accepted through
/// one route is never refused through another.
public static int checkSupported(Integer stated) {
int version = stated == null ? BASE : stated;
if (version > SUPPORTED)
throw new VectorDataException("dataset requires format_version " + version + "; this build supports up to "
+ SUPPORTED + ". Upgrade nb-vectordata to read it.");
return version;
}
/// Refuses a declaration that understates what it holds: a stated
/// version lower than the content requires is a declaration
/// contradicting itself, the same class of fault as a record count
/// that disagrees with its shards. An absent field is not a claim
/// and passes; a version higher than needed is merely generous.
public static void checkStatedAgainstContent(Integer stated, int required) {
if (stated != null && required > stated)
throw new VectorDataException("dataset declares format_version " + stated + " but its content requires "
+ required + " — a declaration cannot understate what it holds");
}
}
@@ -20,25 +20,30 @@ import java.util.List;
/// What a prefetch would fetch, before any of it moves. Returned by
/// [TestDataView#prefetchPlan].
///
/// - `requestedRanges` — the byte ranges the window resolved to, one
/// per interval, before merging: what the caller actually asked for.
/// - `requestedRanges` — the byte ranges the window resolved to, before
/// merging: what the caller actually asked for. Each is qualified by
/// the shard it lies in, because across a series the same byte offset
/// exists in every file; a single-file facet is shard `0`.
/// - `byteRanges` — the ranges that will be issued, after merging those
/// whose fetches would overlap: one request each.
/// whose fetches would overlap: one request each. Ranges in different
/// shards never merge — they are in different files.
/// - `fills` — chunk-level cost of each issued range. Empty when the
/// facet has no chunks (local storage, which is free by definition).
/// - `prerequisiteBytes` — bytes that had to be read before the window
/// could be resolved at all: the offset index, for variable-length
/// formats; zero for uniform-stride formats. Reported whether or not
/// it was paid this time — the index is cached on the view's facet
/// handle, so a plan and the fetch that follows it load it once.
/// formats; zero for uniform-stride formats. Across a series each
/// touched file's index is a separate read, so they sum. Reported
/// whether or not it was paid this time — the index is cached on the
/// view's facet handle, so a plan and the fetch that follows it load
/// it once.
/// - `degradesToFullDownload` — set when a window was asked for and
/// could not be resolved for this format or storage, so honouring the
/// request means fetching the whole facet. A prefetch with *no*
/// window is a request for the whole facet, not a fallback, and never
/// sets this.
/// - `facetBytes` — size of the facet, for reading the degrade case
/// against.
public record PrefetchPlan(List<ByteRange> requestedRanges, List<ByteRange> byteRanges, List<RangeFill> fills,
/// - `facetBytes` — size of the facet, every shard included, for reading
/// the degrade case against.
public record PrefetchPlan(List<ShardRange> requestedRanges, List<ShardRange> byteRanges, List<RangeFill> fills,
long prerequisiteBytes, boolean degradesToFullDownload, long facetBytes) {
public PrefetchPlan {
@@ -57,7 +62,7 @@ public record PrefetchPlan(List<ByteRange> requestedRanges, List<ByteRange> byte
public int chunksToFetch() { return fills.stream().mapToInt(RangeFill::chunksToFetch).sum(); }
/// Requests that will be issued. Lower than the interval count when
/// intervals were merged.
/// intervals were merged; higher when a window crossed a shard seam.
public int requests() { return byteRanges.size(); }
/// Bytes fetched beyond what was asked for. Two sources, counted
@@ -66,7 +71,7 @@ public record PrefetchPlan(List<ByteRange> requestedRanges, List<ByteRange> byte
/// for.
public long overfetchBytes() {
long spanned = fills.stream().mapToLong(fill -> Math.max(0, fill.alignedEnd() - fill.alignedStart())).sum();
long asked = requestedRanges.stream().mapToLong(ByteRange::length).sum();
long asked = requestedRanges.stream().mapToLong(ShardRange::length).sum();
return Math.max(0, spanned - asked);
}
@@ -0,0 +1,36 @@
/*
* 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;
/// A half-open byte range `[start, end)` within a named shard of a facet.
///
/// The shard index is not decoration: across a series the same byte
/// offset exists in every file, so a range without one names no bytes.
/// A single-file facet is shard `0`, which is exactly the old meaning
/// under a new name — see [#whole].
public record ShardRange(int shard, long start, long end) {
/// A range in the only shard of a single-file facet.
public static ShardRange whole(long start, long end) { return new ShardRange(0, start, end); }
/// Bytes spanned by this range.
public long length() { return Math.max(0, end - start); }
/// Whether the range names no bytes.
public boolean isEmpty() { return end <= start; }
@Override public String toString() { return "shard " + shard + " [" + start + ".." + end + ")"; }
}
@@ -17,6 +17,8 @@ package io.nosqlbench.vectordata;
import io.nosqlbench.vectordata.internal.HttpTransport;
import io.nosqlbench.vectordata.internal.ManifestView;
import io.nosqlbench.vectordata.internal.Shards;
import io.nosqlbench.vectordata.internal.SourceSpec;
import io.nosqlbench.vectordata.internal.YamlData;
import java.io.IOException;
@@ -54,13 +56,27 @@ public final class TestDataGroup {
}
public static TestDataGroup load(URI source, VectorDataSettings settings) {
if (!isYaml(source)) {
// A directory resolves to its dataset.yaml, then to a legacy
// knn_entries.yaml — but only when the canonical manifest is
// absent. One that exists and is refused stays refused: a
// fallback that swallowed the refusal would report "no such
// file" for a dataset whose fault has a name.
URI canonical = child(source, "dataset.yaml");
try { return load(canonical, settings); }
catch (VectorDataException ignored) { return load(child(source, "knn_entries.yaml"), settings); }
String text;
try { text = read(canonical, settings); }
catch (VectorDataException absent) { return load(child(source, "knn_entries.yaml"), settings); }
return parse(canonical, text, settings);
}
URI manifest = source;
Map<String, Object> root = YamlData.parse(read(manifest, settings), manifest.toString());
return parse(source, read(source, settings), settings);
}
private static TestDataGroup parse(URI manifest, String text, VectorDataSettings settings) {
Map<String, Object> root = YamlData.parse(text, manifest.toString());
if (!(root.get("profiles") instanceof Map<?, ?>)) return legacy(root, manifest, settings, directoryName(manifest));
// A dataset above this build's version is refused before
// anything else is read from it: the field exists to turn "no
// such file" into a diagnosis.
Integer stated = root.get("format_version") == null ? null : YamlData.integer(root.get("format_version"), "format_version");
FormatVersion.checkSupported(stated);
String name = YamlData.optionalString(root.get("name")); if (name == null) name = basename(manifest);
Object rawProfiles = root.get("profiles");
Map<String, Map<String, Object>> profiles = new LinkedHashMap<>();
@@ -72,7 +88,12 @@ public final class TestDataGroup {
if (profiles.isEmpty()) throw new VectorDataException("Manifest contains no profiles: " + manifest);
Map<String, Object> attributes = root.get("attributes") instanceof Map<?, ?> declared
? Map.copyOf(YamlData.map(declared, "attributes")) : Map.of();
return new TestDataGroup(name, manifest, profiles, settings, attributes);
TestDataGroup group = new TestDataGroup(name, manifest, profiles, settings, attributes);
// The version the content requires is derived by folding every
// declaration, never asserted — and every declaration is
// checked for self-consistency here, before a facet opens.
FormatVersion.checkStatedAgainstContent(stated, group.requiredVersion());
return group;
}
private static TestDataGroup legacy(Map<String, Object> root, URI manifest, VectorDataSettings settings, String preferred) {
String configuredBase = root.get("_defaults") instanceof Map<?, ?> map ? YamlData.optionalString(YamlData.map(map, "_defaults").get("base_url")) : null;
@@ -84,9 +105,9 @@ public final class TestDataGroup {
Map<String, Object> values = YamlData.map(item.getValue(), "legacy entry " + item.getKey()); Map<String, Object> resolved = new LinkedHashMap<>();
for (Map.Entry<String, Object> facet : values.entrySet()) {
if (facet.getValue() instanceof String path) {
String[] split = splitWindowSuffix(path);
resolved.put(facet.getKey(), split[1] == null ? base.resolve(split[0]).toString()
: Map.of("source", base.resolve(split[0]).toString(), "window", split[1]));
SourceSpec spec = SourceSpec.parse(path);
resolved.put(facet.getKey(), spec.windowText() == null ? base.resolve(spec.path()).toString()
: Map.of("source", base.resolve(spec.path()).toString(), "window", spec.windowText()));
} else resolved.put(facet.getKey(), facet.getValue());
}
grouped.computeIfAbsent(dataset, ignored -> new LinkedHashMap<>()).put(profile, resolved);
@@ -99,35 +120,93 @@ public final class TestDataGroup {
public Map<String, Map<String, Object>> profiles() { return Map.copyOf(profiles); }
public TestDataView profile(String profile) {
String selected = profile == null || profile.isBlank() ? (profiles.containsKey("default") ? "default" : profiles.keySet().iterator().next()) : profile;
return new ManifestView(name, selected, facets(selected, new LinkedHashMap<>()), settings, attributes);
return new ManifestView(name, selected, facets(selected), settings, attributes);
}
private Map<String, FacetDescriptor> facets(String profile, Map<String, FacetDescriptor> inherited) {
/// The format version this manifest's content requires: sharded if
/// any profile declares a multi-file facet, base otherwise.
private int requiredVersion() {
for (Map.Entry<String, Map<String, Object>> profile : profiles.entrySet())
for (FacetDescriptor facet : declared(profile.getKey(), profile.getValue()).values())
if (facet.isSeries()) return FormatVersion.SHARDED;
return FormatVersion.BASE;
}
private Map<String, FacetDescriptor> facets(String profile) {
Map<String, Object> definition = profiles.get(profile);
if (definition == null) throw new VectorDataException("Dataset " + name + " has no profile " + profile);
Map<String, FacetDescriptor> inherited = new LinkedHashMap<>();
// Profiles name only what differs: facets a profile does not
// declare resolve from `default`. A sized profile like `100k`
// carries a windowed base and its own neighbor facets and
// inherits the rest — query vectors included — without an
// explicit `extends`.
if (!"default".equals(profile) && profiles.containsKey("default")) inherited.putAll(facets("default", new LinkedHashMap<>()));
if (!"default".equals(profile) && profiles.containsKey("default")) inherited.putAll(facets("default"));
String parent = YamlData.optionalString(definition.get("extends"));
if (parent != null) {
if (parent.equals(profile)) throw new VectorDataException("Profile cannot extend itself: " + profile);
inherited.putAll(facets(parent, new LinkedHashMap<>()));
inherited.putAll(facets(parent));
}
inherited.putAll(declared(profile, definition));
return inherited;
}
/// The facets a profile declares itself, resolved against the
/// manifest. Declaration shape is resolved here, once: a string
/// source is one file; a `NNNN` pattern with `shard_stride` and
/// `shard_count`, or an array of sources, is a series. Every series
/// declaration is checked for self-consistency at load, so a
/// dataset that disagrees with itself is refused before a facet
/// opens.
private Map<String, FacetDescriptor> declared(String profile, Map<String, Object> definition) {
Map<String, FacetDescriptor> result = new LinkedHashMap<>();
for (Map.Entry<String, Object> entry : definition.entrySet()) {
if (NON_FACETS.contains(entry.getKey())) continue;
String facetName = canonical(entry.getKey());
if (entry.getValue() instanceof List<?>)
throw new VectorDataException("facet '" + facetName + "' in profile " + profile + ": a list of sources belongs under 'source:'");
Map<String, Object> values = entry.getValue() instanceof Map<?, ?> ? YamlData.map(entry.getValue(), "facet " + entry.getKey()) : Map.of("source", entry.getValue());
String source = YamlData.optionalString(values.get("source"));
if (source == null) source = YamlData.optionalString(values.get("path"));
if (source == null) continue;
Object rawSource = values.get("source");
if (rawSource == null) rawSource = values.get("path");
if (rawSource == null) continue;
String window = windowText(values.get("window"));
String[] split = splitWindowSuffix(source);
if (split[1] != null) window = split[1];
inherited.put(facetName, new FacetDescriptor(facetName, manifest.resolve(split[0]), window, Map.copyOf(values)));
Long stride = longOrNull(values.get("shard_stride"), "shard_stride");
Integer count = values.get("shard_count") == null ? null : YamlData.integer(values.get("shard_count"), "shard_count");
Long recordCount = longOrNull(values.get("record_count"), "record_count");
boolean isArray = rawSource instanceof List<?>;
List<String> raw = new ArrayList<>();
if (isArray) for (Object item : (List<?>) rawSource) raw.add(YamlData.string(item, "facet " + facetName + " source entry"));
else raw.add(String.valueOf(rawSource));
boolean layout = stride != null || count != null;
if (!isArray && !layout && !Shards.hasShardField(raw.get(0))) {
SourceSpec spec = SourceSpec.parse(raw.get(0));
if (spec.windowText() != null) window = spec.windowText();
result.put(facetName, new FacetDescriptor(facetName, manifest.resolve(spec.path()), window, Map.copyOf(values)));
continue;
}
List<String> entries = new ArrayList<>();
for (String item : raw) {
SourceSpec spec = SourceSpec.parse(item);
// A window suffix on a uniform pattern is the facet
// window: a pattern names no file, so there is no entry
// for it to bound. Given both ways, it is refused rather
// than guessed at.
if (layout && spec.windowText() != null) {
if (window != null)
throw new VectorDataException("facet '" + facetName + "': the facet window is given twice: on the source pattern and as 'window'");
window = spec.windowText();
spec = new SourceSpec(spec.path(), DSWindow.ALL, null, spec.declaredCount());
}
entries.add(spec.withPath(manifest.resolve(spec.path()).toString()).render());
}
Shards.validate(facetName, new Shards.Declaration(raw, isArray, stride, count, recordCount));
result.put(facetName, new FacetDescriptor(facetName, null, window, Map.copyOf(values),
new FacetDescriptor.Series(entries, isArray, stride, count, recordCount)));
}
return inherited;
return result;
}
private static Long longOrNull(Object value, String label) {
if (value == null) return null;
if (value instanceof Number number) return number.longValue();
try { return Long.parseLong(String.valueOf(value).trim()); }
catch (NumberFormatException e) { throw new VectorDataException("Invalid integer " + label + ": " + value, e); }
}
/// Normalizes the `window:` key to the canonical interval text.
/// Manifests carry it as the string grammar, but the serializer
@@ -157,28 +236,6 @@ public final class TestDataGroup {
}
throw new VectorDataException("Unrecognized window interval: " + item);
}
/// Splits the documented window-suffix sugar off a facet source
/// string — `base.fvec[0..1M)` names the file plus a record window,
/// with either bracket kind on either side. The outer delimiters
/// are structural: they separate the path from the window and do
/// not affect interval bound semantics. Returns
/// `{path, windowOrNull}`. A source string only fails when it
/// *looks* like it carries a window suffix and that suffix is
/// malformed — a plain path always passes through — so an error
/// here names the broken window rather than turning it into "no
/// such file".
private static String[] splitWindowSuffix(String source) {
if (!source.endsWith("]") && !source.endsWith(")")) return new String[] {source, null};
int bracket = source.indexOf('['); int paren = source.indexOf('(');
int open = bracket < 0 ? paren : paren < 0 ? bracket : Math.min(bracket, paren);
if (open <= 0) return new String[] {source, null};
String inner = source.substring(open + 1, source.length() - 1);
try { DSWindow.parse(inner); }
catch (VectorDataException malformed) {
throw new VectorDataException("source '" + source + "' has a malformed window: " + malformed.getMessage());
}
return new String[] {source.substring(0, open), inner};
}
private static boolean isYaml(URI source) { String path = source.getPath() == null ? "" : source.getPath().toLowerCase(); return path.endsWith(".yaml") || path.endsWith(".yml"); }
private static URI child(URI source, String name) { String text = source.toString(); return URI.create(text.endsWith("/") ? text + name : text + "/" + name); }
private static String directoryName(URI source) { String path = source.getPath(); int end = path.lastIndexOf('/'); String parent = path.substring(0, end); int start = parent.lastIndexOf('/'); return parent.substring(start + 1); }
@@ -29,6 +29,10 @@ public interface TestDataView {
VectorReader<int[]> neighborIndices();
VectorReader<float[]> neighborDistances();
VectorReader<int[]> metadataResults();
/// Opens a facet by name. A facet spanning several files presents
/// the same reader surface as a single-file one — `count()` is the
/// series total, `get(o)` resolves and reads from the owning shard —
/// and callers that never ask about layout never learn it exists.
VectorReader<?> openFacet(String name);
VvecReader<?> openVariableFacet(String name);
/// Dataset-level attributes from the manifest's `attributes:`
@@ -52,7 +56,8 @@ public interface TestDataView {
/// reporting per-range progress through `progress`. An empty window
/// requests the whole facet. A window that cannot be resolved for
/// the facet's format fails under [WholeFacetFallback#REFUSE]
/// rather than quietly fetching everything.
/// rather than quietly fetching everything. Across a series the
/// window costs the shards it spans, and only those.
PrefetchReport prefetch(String facet, DSWindow window, WholeFacetFallback fallback, PrebufferProgress progress);
/// Same as [#prefetch(String, DSWindow, WholeFacetFallback, PrebufferProgress)]
@@ -18,6 +18,7 @@ package io.nosqlbench.vectordata;
import io.nosqlbench.vectordata.internal.ByteStorage;
import io.nosqlbench.vectordata.internal.FixedVectorReader;
import io.nosqlbench.vectordata.internal.ScalarReader;
import io.nosqlbench.vectordata.internal.ShardedVectorReader;
import io.nosqlbench.vectordata.internal.StorageFactory;
import io.nosqlbench.vectordata.internal.VariableVectorReader;
@@ -30,9 +31,14 @@ public final class VectorReaders {
private VectorReaders() { }
public static VectorReader<?> open(String source) { return open(toUri(source), VectorDataSettings.defaults(), "direct"); }
public static VectorReader<?> open(URI source, VectorDataSettings settings, String identity) {
ElementType type = ElementType.forExtension(source.toString());
ByteStorage storage = new StorageFactory(settings).open(source, identity);
return isScalar(source.toString()) ? new ScalarReader(storage, type) : new FixedVectorReader<>(storage, type);
return over(new StorageFactory(settings).open(source, identity), source.toString());
}
/// A reader over storage already open, chosen by the source's
/// extension — the per-file reader a facet spread across several
/// files composes.
public static VectorReader<?> over(ByteStorage storage, String source) {
ElementType type = ElementType.forExtension(source);
return isScalar(source) ? new ScalarReader(storage, type) : new FixedVectorReader<>(storage, type);
}
public static VvecReader<?> openVvec(URI source, VectorDataSettings settings, String identity) {
String text = source.toString().toLowerCase(Locale.ROOT);
@@ -50,13 +56,23 @@ public final class VectorReaders {
return new VariableVectorReader<>(data, new StorageFactory(settings).open(other, identity), type, !expect64);
}
}
@SuppressWarnings("unchecked") public static VectorReader<float[]> f32(URI source, VectorDataSettings settings, String identity) {
VectorReader<?> reader = open(source, settings, identity); if (reader.elementType() != ElementType.F32 || !(reader instanceof FixedVectorReader<?>)) throw new VectorDataException("Expected f32 fixed vectors: " + source); return (VectorReader<float[]>) reader;
public static VectorReader<float[]> f32(URI source, VectorDataSettings settings, String identity) {
return expectFixed(open(source, settings, identity), ElementType.F32, source);
}
@SuppressWarnings("unchecked") public static VectorReader<int[]> i32(URI source, VectorDataSettings settings, String identity) {
VectorReader<?> reader = open(source, settings, identity); if (reader.elementType() != ElementType.I32 || !(reader instanceof FixedVectorReader<?>)) throw new VectorDataException("Expected i32 fixed vectors: " + source); return (VectorReader<int[]>) reader;
public static VectorReader<int[]> i32(URI source, VectorDataSettings settings, String identity) {
return expectFixed(open(source, settings, identity), ElementType.I32, source);
}
static boolean isScalar(String source) { String ext = extension(source); return ext.matches("u8|i8|u16|i16|u32|i32|u64|i64"); }
/// Narrows a reader to a fixed-vector element type, single-file and
/// sharded alike; a packed scalar or another element type is refused
/// naming what was expected.
@SuppressWarnings("unchecked")
public static <T> VectorReader<T> expectFixed(VectorReader<?> reader, ElementType type, Object source) {
boolean fixed = reader instanceof FixedVectorReader<?> || (reader instanceof ShardedVectorReader<?> sharded && !sharded.isScalar());
if (reader.elementType() != type || !fixed)
throw new VectorDataException("Expected " + type.name().toLowerCase(Locale.ROOT) + " fixed vectors: " + source);
return (VectorReader<T>) reader;
}
public static boolean isScalar(String source) { String ext = extension(source); return ext.matches("u8|i8|u16|i16|u32|i32|u64|i64"); }
static boolean isVector(String source) { return !isScalar(source); }
private static String extension(String value) { int dot = value.lastIndexOf('.'); int end = value.indexOf('?', dot); return value.substring(dot + 1, end < 0 ? value.length() : end).toLowerCase(Locale.ROOT); }
private static URI toUri(String source) { return source.contains("://") || source.startsWith("file:") ? URI.create(source) : Path.of(source).toAbsolutePath().toUri(); }
@@ -0,0 +1,263 @@
/*
* 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.internal;
import io.nosqlbench.vectordata.AccessMode;
import io.nosqlbench.vectordata.CacheStats;
import io.nosqlbench.vectordata.RangeFill;
import io.nosqlbench.vectordata.VectorDataException;
import io.nosqlbench.vectordata.VectorDataSettings;
import java.net.URI;
import java.util.ArrayDeque;
import java.util.ArrayList;
import java.util.Deque;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
/// The files behind a multi-file facet, and the ordinal model that maps
/// records onto them.
///
/// Ordinal concerns count per **shard** and storage concerns count per
/// **file**: two shards sliced from one file share one storage, one
/// sidecar, one offset index, and one fetch. Anything byte-shaped here
/// iterates files; anything ordinal-shaped iterates shards.
///
/// Files open lazily — reading ordinal 0 must not open shard 400 — and
/// the number held open at once is capped by a budget derived from the
/// process descriptor limit. Eviction releases this series' reference;
/// the storage registry holds files weakly, so a file nobody else holds
/// closes on its own.
public final class FacetSeries {
private static final System.Logger LOGGER = System.getLogger(FacetSeries.class.getName());
private static final long FD_FRACTION = 4;
private static final int FD_FLOOR = 8;
private final Shards shards;
private final List<String> filePaths;
private final List<URI> fileUris;
private final int[] fileOfShard;
private final VectorDataSettings settings;
private final String identity;
private final ByteStorage[] opened;
private final long[][] offsets;
/// Open files, most-recently-used last, bounded by `cap`.
private final Deque<Integer> lru = new ArrayDeque<>();
private final int cap;
private volatile long totalBytes = -1;
public FacetSeries(Shards shards, VectorDataSettings settings, String identity) {
this(shards, settings, identity, openFileCap());
}
/// As the public constructor, with the descriptor budget supplied —
/// what lets eviction be exercised at a size a test can build.
FacetSeries(Shards shards, VectorDataSettings settings, String identity, int budget) {
this.shards = shards; this.settings = settings; this.identity = identity;
Map<String, Integer> index = new LinkedHashMap<>();
List<String> paths = new ArrayList<>();
fileOfShard = new int[shards.entries().size()];
for (int shard = 0; shard < fileOfShard.length; shard++) {
String source = shards.entries().get(shard).source();
Integer at = index.get(source);
if (at == null) { at = paths.size(); index.put(source, at); paths.add(source); }
fileOfShard[shard] = at;
}
filePaths = List.copyOf(paths);
fileUris = paths.stream().map(URI::create).toList();
opened = new ByteStorage[paths.size()];
offsets = new long[paths.size()][];
if (paths.size() > budget)
LOGGER.log(System.Logger.Level.WARNING, "facet spans " + paths.size() + " files but only " + budget
+ " may stay open at once; readers crossing that many files will reopen them. Raise VECTORDATA_SHARD_FD_CAP or ulimit -n.");
// A series never needs more descriptors than it has files.
cap = Math.min(budget, Math.max(1, paths.size()));
}
/// How many files a series may hold open at once: a quarter of the
/// process descriptor soft limit, floor 8, or `VECTORDATA_SHARD_FD_CAP`
/// when set. Derived rather than fixed because a constant is wrong
/// in both directions — it strands descriptors on a generous host
/// and thrashes on a constrained one — and is invisible to the
/// operator who raised `ulimit -n` precisely to make this work.
public static int openFileCap() { return resolveOpenFileCap(System.getenv("VECTORDATA_SHARD_FD_CAP"), fdSoftLimit()); }
static int resolveOpenFileCap(String env, Long softLimit) {
if (env != null) {
try { int forced = Integer.parseInt(env.trim()); if (forced > 0) return forced; }
catch (NumberFormatException ignored) { }
}
long soft = softLimit == null ? 1024 : softLimit;
return (int) Math.max(FD_FLOOR, soft / FD_FRACTION);
}
private static Long fdSoftLimit() {
try {
if (java.lang.management.ManagementFactory.getOperatingSystemMXBean() instanceof com.sun.management.UnixOperatingSystemMXBean unix)
return unix.getMaxFileDescriptorCount();
} catch (Throwable unavailable) { /* not a Unix runtime, or the bean is inaccessible */ }
return null;
}
public Shards shards() { return shards; }
public int fileCount() { return filePaths.size(); }
public String filePath(int file) { return filePaths.get(file); }
public URI fileUri(int file) { return fileUris.get(file); }
public VectorDataSettings settings() { return settings; }
public String identity() { return identity; }
/// Which file a shard is drawn from.
public int fileIndexOfShard(int shard) {
if (shard < 0 || shard >= fileOfShard.length) throw new VectorDataException("shard " + shard + " out of range");
return fileOfShard[shard];
}
/// The storage of one file, opened on first ask. A file that will
/// not open is named: a declared shard that is absent is a broken
/// facet, never a shorter one.
public ByteStorage file(int file) {
synchronized (lru) {
if (opened[file] != null) { touch(file); return opened[file]; }
}
ByteStorage storage;
try { storage = new StorageFactory(settings).open(fileUris.get(file), identity); }
catch (VectorDataException failure) {
throw new VectorDataException("open shard file '" + filePaths.get(file) + "': " + failure.getMessage(), failure);
}
synchronized (lru) {
if (opened[file] == null) opened[file] = storage;
touch(file);
while (lru.size() > cap) {
Integer victim = lru.pollFirst();
if (victim == null) break;
if (victim == file) { lru.addLast(victim); break; }
opened[victim] = null;
}
return opened[file];
}
}
private void touch(int file) { lru.remove(file); lru.addLast(file); }
/// Files currently held open by this series.
public int openFileCount() {
synchronized (lru) { int count = 0; for (ByteStorage storage : opened) if (storage != null) count++; return count; }
}
/// The record-offset index of one **file**, loaded on first use and
/// reused; two shards sliced from one file load it once between
/// them. `null` when it cannot be loaded without fetching.
public long[] publishedOffsets(int file) {
synchronized (offsets) {
if (offsets[file] == null) offsets[file] = Prefetcher.publishedOffsets(file(file), fileUris.get(file), settings, identity);
return offsets[file];
}
}
/// The byte range a shard can address within its file — its whole
/// file for a whole-file shard, its window's bytes for a sliced one
/// — or `null` when the format cannot map it.
public long[] shardByteExtent(int shard) {
Shards.Entry entry = shards.entries().get(shard);
int file = fileIndexOfShard(shard);
Prefetcher.MappedRange mapped = Prefetcher.mapInFile(filePaths.get(file), entry.fileBase(), entry.fileBase() + entry.len(),
file(file), () -> publishedOffsets(file));
return mapped == null ? null : new long[] {mapped.byteStart(), mapped.byteEnd()};
}
/// Every file's bytes summed, opening each on the first ask. Fails
/// naming the file that will not open.
public long tryTotalSize() {
long cached = totalBytes;
if (cached >= 0) return cached;
long sum = 0;
for (int file = 0; file < filePaths.size(); file++) sum += file(file).size();
totalBytes = sum;
return sum;
}
/// As [#tryTotalSize], answering `0` for a series that cannot be
/// sized — the only answer this signature allows, which is why
/// every caller that can report a failure uses the other.
public long totalSize() {
try { return tryTotalSize(); }
catch (VectorDataException failure) {
LOGGER.log(System.Logger.Level.ERROR, "facet size unavailable: " + failure.getMessage());
return 0;
}
}
/// Bytes per shard, in shard order; `0` for a shard whose file will
/// not open.
public long[] shardSizes() {
long[] sizes = new long[fileOfShard.length];
for (int shard = 0; shard < sizes.length; shard++) {
try { sizes[shard] = file(fileOfShard[shard]).size(); }
catch (VectorDataException unopenable) { sizes[shard] = 0; }
}
return sizes;
}
/// Whether **every byte this facet can address** is resident — not
/// every byte of every file it draws from. A facet slicing a tenth
/// of a large file would otherwise report incomplete forever unless
/// the other nine tenths were downloaded: bytes it can never
/// address, fetched to satisfy a predicate about them.
public boolean isComplete() {
try {
for (int shard = 0; shard < fileOfShard.length; shard++) {
ByteStorage storage = file(fileOfShard[shard]);
long[] extent = shardByteExtent(shard);
if (extent == null) { if (!storage.isComplete()) return false; continue; }
RangeFill fill = storage.rangeFill(extent[0], extent[1]);
if (fill != null && !fill.isResident()) return false;
}
return true;
} catch (VectorDataException unopenable) { return false; }
}
/// Whether every file is local.
public boolean isLocal() {
for (int file = 0; file < filePaths.size(); file++) if (file(file).stats().accessMode() != AccessMode.LOCAL) return false;
return true;
}
/// Statistics folded over every file: bytes and chunk counts sum,
/// the chunk size is the largest in play, completeness is the
/// conjunction, and the access mode is the [AccessMode#weakest] —
/// the promise true of every read this facet will serve.
public CacheStats cacheStats() {
List<AccessMode> modes = new ArrayList<>();
long total = 0, cached = 0, chunk = 0, hits = 0, misses = 0;
boolean complete = true;
for (int file = 0; file < filePaths.size(); file++) {
CacheStats stats = file(file).stats();
modes.add(stats.accessMode());
total += stats.totalBytes(); cached += stats.cachedBytes(); chunk = Math.max(chunk, stats.chunkSize());
hits += stats.hits(); misses += stats.misses(); complete &= stats.complete();
}
return new CacheStats(AccessMode.weakest(modes), total, cached, chunk, hits, misses, complete);
}
/// The largest transfer-chunk size across the files; `0` when none
/// is chunked.
public long chunkSize() {
long chunk = 0;
for (int file = 0; file < filePaths.size(); file++) chunk = Math.max(chunk, file(file).stats().chunkSize());
return chunk;
}
}
@@ -15,14 +15,15 @@
*/
package io.nosqlbench.vectordata.internal;
import io.nosqlbench.vectordata.ByteRange;
import io.nosqlbench.vectordata.DSWindow;
import io.nosqlbench.vectordata.ElementType;
import io.nosqlbench.vectordata.FacetDescriptor;
import io.nosqlbench.vectordata.FacetNames;
import io.nosqlbench.vectordata.PrebufferProgress;
import io.nosqlbench.vectordata.PrefetchHandle;
import io.nosqlbench.vectordata.PrefetchPlan;
import io.nosqlbench.vectordata.PrefetchReport;
import io.nosqlbench.vectordata.ShardRange;
import io.nosqlbench.vectordata.TestDataView;
import io.nosqlbench.vectordata.VectorDataException;
import io.nosqlbench.vectordata.VectorDataSettings;
@@ -31,7 +32,6 @@ import io.nosqlbench.vectordata.VectorReaders;
import io.nosqlbench.vectordata.VvecReader;
import io.nosqlbench.vectordata.WholeFacetFallback;
import java.net.URI;
import java.util.LinkedHashMap;
import java.util.Map;
import java.util.Optional;
@@ -43,7 +43,8 @@ public final class ManifestView implements TestDataView {
private final Map<String, Object> attributes;
/// One handle per facet, so a plan and the fetch that follows it —
/// and every later ask through this view — share one loaded offset
/// index rather than loading it repeatedly.
/// index and one realized series rather than loading them
/// repeatedly.
private final Map<String, Prefetcher.FacetHandle> handles = new ConcurrentHashMap<>();
public ManifestView(String dataset, String profile, Map<String, FacetDescriptor> facets, VectorDataSettings settings) {
this(dataset, profile, facets, settings, Map.of());
@@ -57,13 +58,16 @@ public final class ManifestView implements TestDataView {
@Override public String profile() { return profile; }
@Override public Map<String, FacetDescriptor> facets() { return facets; }
@Override public Optional<FacetDescriptor> facet(String name) { return Optional.ofNullable(facets.get(canonical(name))); }
@Override public VectorReader<float[]> baseVectors() { return fixed("base_vectors", VectorReaders::f32); }
@Override public VectorReader<float[]> queryVectors() { return fixed("query_vectors", VectorReaders::f32); }
@Override public VectorReader<int[]> neighborIndices() { return fixed("neighbor_indices", VectorReaders::i32); }
@Override public VectorReader<float[]> neighborDistances() { return fixed("neighbor_distances", VectorReaders::f32); }
@Override public VectorReader<int[]> metadataResults() { return fixed("metadata_results", VectorReaders::i32); }
@Override public VectorReader<?> openFacet(String name) { FacetDescriptor facet = require(name); return window(VectorReaders.open(facet.source(), settings, dataset), facet); }
@Override public VvecReader<?> openVariableFacet(String name) { FacetDescriptor facet = require(name); return VectorReaders.openVvec(facet.source(), settings, dataset); }
@Override public VectorReader<float[]> baseVectors() { return fixed("base_vectors", ElementType.F32); }
@Override public VectorReader<float[]> queryVectors() { return fixed("query_vectors", ElementType.F32); }
@Override public VectorReader<int[]> neighborIndices() { return fixed("neighbor_indices", ElementType.I32); }
@Override public VectorReader<float[]> neighborDistances() { return fixed("neighbor_distances", ElementType.F32); }
@Override public VectorReader<int[]> metadataResults() { return fixed("metadata_results", ElementType.I32); }
@Override public VectorReader<?> openFacet(String name) { FacetDescriptor facet = require(name); return window(facet, open(facet)); }
@Override public VvecReader<?> openVariableFacet(String name) {
FacetDescriptor facet = require(name);
return facet.isSeries() ? new ShardedVvecReader<>(handle(name).series()) : VectorReaders.openVvec(facet.source(), settings, dataset);
}
@Override public Map<String, Object> attributes() { return attributes; }
@Override public void prebuffer(PrebufferProgress progress) { for (String name : facets.keySet()) openFacet(name).prebuffer(progress); }
@Override public PrefetchPlan prefetchPlan(String facet, DSWindow window) { return Prefetcher.plan(handle(facet), window); }
@@ -73,36 +77,49 @@ public final class ManifestView implements TestDataView {
Prefetcher.FacetHandle handle = handle(facet);
PrefetchPlan planned = Prefetcher.plan(handle, window);
Prefetcher.checkFallback(facet, planned, fallback);
ByteStorage storage = handle.data();
if (planned.degradesToFullDownload()) { storage.prebuffer(progress); return new PrefetchReport(planned, 1); }
int fetched = 0;
for (ByteRange range : planned.byteRanges()) { storage.prebufferRange(range.start(), range.end(), progress); fetched++; }
return new PrefetchReport(planned, fetched);
return new PrefetchReport(planned, fetch(handle, planned, progress));
}
@Override public PrefetchHandle prefetchInBackground(String facet, DSWindow window, WholeFacetFallback fallback) {
Prefetcher.FacetHandle handle = handle(facet);
PrefetchPlan planned = Prefetcher.plan(handle, window);
Prefetcher.checkFallback(facet, planned, fallback);
ByteStorage storage = handle.data();
return PrefetchHandle.launch(facet, planned, ticker -> {
if (planned.degradesToFullDownload()) { storage.prebuffer((cached, total) -> ticker.bytes(cached)); return; }
if (planned.degradesToFullDownload()) { handle.prebufferWhole((cached, total) -> ticker.bytes(cached)); return; }
long fetched = 0;
for (ByteRange range : planned.byteRanges()) {
for (ShardRange range : planned.byteRanges()) {
if (ticker.cancelled()) break;
long base = fetched;
long[] seen = {0};
storage.prebufferRange(range.start(), range.end(), (cached, total) -> { seen[0] = cached; ticker.bytes(base + cached); });
handle.prebufferShardRange(range.shard(), range.start(), range.end(), (cached, total) -> { seen[0] = cached; ticker.bytes(base + cached); });
fetched = base + seen[0];
ticker.rangeDone();
}
});
}
/// Fetches what a plan says, range by range within the shard each
/// names, and reports how many ranges were issued.
private static int fetch(Prefetcher.FacetHandle handle, PrefetchPlan plan, PrebufferProgress progress) {
if (plan.degradesToFullDownload()) { handle.prebufferWhole(progress); return 1; }
int fetched = 0;
for (ShardRange range : plan.byteRanges()) { handle.prebufferShardRange(range.shard(), range.start(), range.end(), progress); fetched++; }
return fetched;
}
private Prefetcher.FacetHandle handle(String name) {
return handles.computeIfAbsent(canonical(name), key -> new Prefetcher.FacetHandle(require(key), settings, dataset));
}
private FacetDescriptor require(String name) { return facet(name).orElseThrow(() -> new VectorDataException("Profile " + profile + " lacks facet " + name)); }
private <T> VectorReader<T> fixed(String name, FixedFactory<T> factory) { FacetDescriptor facet = require(name); return window(factory.open(facet.source(), settings, dataset), facet); }
private static <T> VectorReader<T> window(VectorReader<T> reader, FacetDescriptor facet) { return facet.window() == null || facet.window().isBlank() ? reader : new WindowedVectorReader<>(reader, facet.window()); }
private VectorReader<?> open(FacetDescriptor facet) {
return facet.isSeries() ? new ShardedVectorReader<>(handle(facet.name()).series()) : VectorReaders.open(facet.source(), settings, dataset);
}
private <T> VectorReader<T> fixed(String name, ElementType type) {
FacetDescriptor facet = require(name);
return window(facet, VectorReaders.expectFixed(open(facet), type, facet.isSeries() ? name : facet.source()));
}
/// Wraps a reader in the facet's declared window — in facet
/// ordinals, so it selects the same records over one file or over
/// five.
private static <T> VectorReader<T> window(FacetDescriptor facet, VectorReader<T> reader) {
return facet.window() == null || facet.window().isBlank() ? reader : new WindowedVectorReader<>(reader, facet.window());
}
private static String canonical(String name) { return FacetNames.canonical(name); }
@FunctionalInterface private interface FixedFactory<T> { VectorReader<T> open(URI source, VectorDataSettings settings, String identity); }
}
@@ -16,12 +16,14 @@
package io.nosqlbench.vectordata.internal;
import io.nosqlbench.vectordata.AccessMode;
import io.nosqlbench.vectordata.ByteRange;
import io.nosqlbench.vectordata.CacheStats;
import io.nosqlbench.vectordata.DSWindow;
import io.nosqlbench.vectordata.ElementType;
import io.nosqlbench.vectordata.FacetDescriptor;
import io.nosqlbench.vectordata.PrebufferProgress;
import io.nosqlbench.vectordata.PrefetchPlan;
import io.nosqlbench.vectordata.RangeFill;
import io.nosqlbench.vectordata.ShardRange;
import io.nosqlbench.vectordata.VectorDataException;
import io.nosqlbench.vectordata.VectorDataSettings;
import io.nosqlbench.vectordata.WholeFacetFallback;
@@ -38,54 +40,72 @@ import java.util.Arrays;
import java.util.Comparator;
import java.util.List;
import java.util.Locale;
import java.util.function.Supplier;
/// Resolves record-coordinate prefetch windows to byte ranges and
/// plans their chunk-level cost, mirroring the `vectordata-rs` prefetch
/// surface. Three per-format mappings:
/// surface. Three per-format mappings, each applied to a shard's
/// **file** ordinals:
///
/// - **xvec** (uniform stride): `4 + dim × elemSize`, with `dim` read
/// from the header at byte 0 — the same first-chunk read any reader
/// does on first access.
/// - **vvec** (variable length): the sibling `IDXFOR__` offset index,
/// loaded whole and cached on the facet handle; on local storage a
/// facet without a sidecar is walked once instead.
/// - **vvec** (variable length): the sibling `IDXFOR__` offset index of
/// that file, loaded whole and cached on the facet handle; on local
/// storage a file without a sidecar is walked once instead.
/// - **scalar** (raw packed values): `ordinal × elemSize` exactly,
/// with no header.
///
/// `null` from the mapping means "cannot window this" — an unsupported
/// A window over a facet spread across several files decomposes into
/// one sub-window per shard it touches, and each maps by the rule above
/// against its own file — so a window costs the shards it spans, and
/// ranges in different shards never merge.
///
/// `null` from a mapping means "cannot window this" — an unsupported
/// format, a corrupt header, an index that cannot be loaded, or an
/// empty range. That is the answer, not a failure, and
/// [WholeFacetFallback] decides what a caller does with it.
public final class Prefetcher {
private Prefetcher() { }
/// A record range resolved to bytes, and what resolving it cost.
/// A record range within one file resolved to bytes, and what
/// resolving it cost.
record MappedRange(long byteStart, long byteEnd, long prerequisiteBytes) { }
/// Record start offsets, one per record. Both published sidecar
/// layouts normalize to this: a trailing entry equal to the payload
/// size is an end-of-data sentinel — no record can start there —
/// and is dropped at parse, so `prerequisiteBytes` reports record
/// starts × 8 for either layout.
private record CachedOffsets(long[] starts) { }
/// A record range resolved to bytes across the shards it spans.
record MappedRanges(List<ShardRange> ranges, long prerequisiteBytes) { }
/// Per-facet handle holding the lazily opened data storage and the
/// Per-facet handle holding the lazily opened storage — one file,
/// or the [FacetSeries] a multi-file facet resolves to — and the
/// offset index once loaded. The index cache is scoped to the view
/// that owns this handle: holding a view is how a caller says it
/// will ask about its facets repeatedly, and dropping it is how
/// they say they are done.
///
/// Everything byte-shaped is asked per shard, a single file being
/// shard `0`, so the planner never branches on layout.
public static final class FacetHandle {
private final FacetDescriptor facet;
private final VectorDataSettings settings;
private final String identity;
private volatile ByteStorage data;
private volatile CachedOffsets offsets;
private volatile FacetSeries series;
private volatile long[] offsets;
public FacetHandle(FacetDescriptor facet, VectorDataSettings settings, String identity) {
this.facet = facet; this.settings = settings; this.identity = identity;
}
public FacetDescriptor facet() { return facet; }
/// Whether this facet spans more than one file.
public boolean isSeries() { return facet.isSeries(); }
/// The one storage behind a single-file facet.
public ByteStorage data() {
if (facet.isSeries())
throw new VectorDataException("facet '" + facet.name() + "' is a series of " + facet.series().entries().size()
+ " sources and has no single storage");
ByteStorage current = data;
if (current == null) {
synchronized (this) {
@@ -96,142 +116,219 @@ public final class Prefetcher {
return current;
}
/// The offset index for a variable-length facet, loaded on
/// first use and reused for the life of this handle. `null`
/// when it cannot be loaded; failures are retried on the next
/// ask rather than cached.
private synchronized CachedOffsets offsets() {
if (offsets == null) offsets = loadOffsets();
/// The series behind a multi-file facet, realized on first ask
/// from its declaration. A bare local entry is measured by
/// opening its file; a remote one was refused at load.
public FacetSeries series() {
if (!facet.isSeries()) throw new VectorDataException("facet '" + facet.name() + "' is a single file, not a series");
FacetSeries current = series;
if (current == null) {
synchronized (this) {
if (series == null) {
FacetDescriptor.Series declared = facet.series();
Shards.Declaration declaration = new Shards.Declaration(declared.entries(), declared.declaredAsArray(),
declared.shardStride(), declared.shardCount(), declared.recordCount());
Shards shards = Shards.realize(facet.name(), declaration, source -> {
ByteStorage storage = new StorageFactory(settings).open(URI.create(source.path()), identity);
Long records = recordsIn(source.path(), storage, settings, identity);
if (records == null) throw new VectorDataException("cannot derive a record count from '" + source.path() + "'");
return records;
});
series = new FacetSeries(shards, settings, identity);
}
current = series;
}
}
return current;
}
/// Size of the facet: every shard's bytes, not the first shard's.
public long totalSize() { return isSeries() ? series().totalSize() : data().size(); }
/// Bytes per shard, in shard order; `0` for a shard whose file
/// will not open.
public long[] shardSizes() { return isSeries() ? series().shardSizes() : new long[] {data().size()}; }
/// The storage a shard's bytes live in.
public ByteStorage shardStorage(int shard) {
if (!isSeries()) { if (shard != 0) throw new VectorDataException("shard " + shard + " out of range"); return data(); }
return series().file(series().fileIndexOfShard(shard));
}
/// Chunk residency of a byte range within one shard, or `null`
/// where the file has no chunks — or will not open, which a
/// plan reports through its sizes rather than by failing here.
public RangeFill shardRangeFill(int shard, long byteStart, long byteEnd) {
try { return shardStorage(shard).rangeFill(byteStart, byteEnd); }
catch (VectorDataException unopenable) { return null; }
}
/// Fetches a byte range within one shard.
public void prebufferShardRange(int shard, long byteStart, long byteEnd, PrebufferProgress progress) {
shardStorage(shard).prebufferRange(byteStart, byteEnd, progress);
}
/// Fetches every file whole — the degrade path, and the meaning
/// of a whole-facet request against a source with no chunks.
public void prebufferWhole(PrebufferProgress progress) {
if (!isSeries()) { data().prebuffer(progress); return; }
FacetSeries files = series();
for (int i = 0; i < files.fileCount(); i++) files.file(i).prebuffer(progress);
}
/// Whether every byte this facet can address is resident.
public boolean isComplete() { return isSeries() ? series().isComplete() : data().isComplete(); }
/// Cache statistics, folded over every file of a series.
public CacheStats cacheStats() { return isSeries() ? series().cacheStats() : data().stats(); }
/// The transfer-chunk size to plan against: the largest across
/// the files, `0` when nothing is chunked.
public long chunkSize() { return isSeries() ? series().chunkSize() : data().stats().chunkSize(); }
/// The offset index for a single-file variable-length facet,
/// loaded on first use and reused for the life of this handle.
/// `null` when it cannot be loaded; failures are retried on the
/// next ask rather than cached.
private synchronized long[] offsets() {
if (offsets == null) offsets = publishedOffsets(data(), facet.source(), settings, identity);
return offsets;
}
/// Planning-only offset loading, mirroring `load_offsets` with
/// `OffsetSource::Published`: a published sidecar, a persisted
/// rebuild, or a local mmap walk — never a remote walk, which
/// would transfer the facet in order to price transferring part
/// of it. Local walks persist their result beside the data,
/// best-effort, so the walk is paid once rather than per view.
private CachedOffsets loadOffsets() {
long payload = data().size();
String expected = indexExtFor(payload);
if (data().stats().accessMode() == AccessMode.LOCAL) return loadLocalOffsets(payload, expected);
URI source = facet.source();
String path = source.getPath() == null ? source.toString() : source.getPath();
String filename = path.substring(path.lastIndexOf('/') + 1);
// Probe the width the payload size calls for first — the
// same rule the local sidecar name uses — so the common
// case costs one request instead of a guaranteed miss.
for (String ext : new String[] {expected, "i32".equals(expected) ? "i64" : "i32"}) {
int width = "i32".equals(ext) ? 4 : 8;
URI sidecar = URI.create(source.toString().replaceAll("[^/]+$", java.util.regex.Matcher.quoteReplacement("IDXFOR__" + filename + "." + ext)));
try {
long[] entries = readEntries(new StorageFactory(settings).open(sidecar, identity), width);
if (entries == null || entries.length == 0) continue;
return new CachedOffsets(dropSentinel(entries, payload));
} catch (VectorDataException missing) { /* try the next sidecar width */ }
}
return null;
}
private String path() { return facet.source().toString(); }
}
/// The local arm of `load_or_build_local_offsets`: read the
/// size-appropriate sidecar beside the data when it is no older
/// than the data, else walk the dimension headers and persist
/// the rebuilt index atomically for the next open.
private CachedOffsets loadLocalOffsets(long payload, String expected) {
Path dataPath = Path.of(facet.source());
int width = "i32".equals(expected) ? 4 : 8;
Path indexPath = dataPath.resolveSibling("IDXFOR__" + dataPath.getFileName() + "." + expected);
long[] fromSidecar = loadLocalIndex(indexPath, dataPath, width);
if (fromSidecar != null) return new CachedOffsets(dropSentinel(fromSidecar, payload));
long[] walked = walkOffsets();
if (walked == null) return null;
persistIndex(indexPath, walked, width);
return new CachedOffsets(walked);
}
// -- Offset indexes --
/// Loads a sidecar file beside local data, or `null` when it is
/// absent, stale (older than the data it describes), or not a
/// whole number of offsets — all of which send the caller back
/// to the walk rather than yielding a reader that cannot see
/// the records the tail describes.
private static long[] loadLocalIndex(Path indexPath, Path dataPath, int width) {
/// Planning-only offset loading, mirroring `load_offsets` with
/// `OffsetSource::Published`: a published sidecar, a persisted
/// rebuild, or a local mmap walk — never a remote walk, which would
/// transfer the file in order to price transferring part of it.
/// Local walks persist their result beside the data, best-effort,
/// so the walk is paid once rather than per view. Both published
/// sidecar layouts normalize to record starts: a trailing entry
/// equal to the payload size is an end-of-data sentinel — no record
/// can start there — and is dropped, so `prerequisiteBytes` reports
/// record starts × 8 for either layout. Per **file**: a shard's
/// sidecar covers its file's records, in file ordinals.
static long[] publishedOffsets(ByteStorage data, URI source, VectorDataSettings settings, String identity) {
long payload = data.size();
String expected = indexExtFor(payload);
if (data.stats().accessMode() == AccessMode.LOCAL) return loadLocalOffsets(data, source, payload, expected);
String path = source.getPath() == null ? source.toString() : source.getPath();
String filename = path.substring(path.lastIndexOf('/') + 1);
// Probe the width the payload size calls for first — the same
// rule the local sidecar name uses — so the common case costs
// one request instead of a guaranteed miss.
for (String ext : new String[] {expected, "i32".equals(expected) ? "i64" : "i32"}) {
int width = "i32".equals(ext) ? 4 : 8;
URI sidecar = URI.create(source.toString().replaceAll("[^/]+$", java.util.regex.Matcher.quoteReplacement("IDXFOR__" + filename + "." + ext)));
try {
if (!Files.isRegularFile(indexPath)) return null;
if (Files.getLastModifiedTime(indexPath).compareTo(Files.getLastModifiedTime(dataPath)) < 0) return null;
byte[] raw = Files.readAllBytes(indexPath);
if (raw.length == 0 || raw.length % width != 0) return null;
ByteBuffer bytes = ByteBuffer.wrap(raw).order(ByteOrder.LITTLE_ENDIAN);
long[] entries = new long[raw.length / width];
for (int at = 0; at < entries.length; at++)
entries[at] = width == 4 ? Integer.toUnsignedLong(bytes.getInt()) : bytes.getLong();
return entries;
} catch (IOException unreadable) { return null; }
long[] entries = readEntries(new StorageFactory(settings).open(sidecar, identity), width);
if (entries == null || entries.length == 0) continue;
return dropSentinel(entries, payload);
} catch (VectorDataException missing) { /* try the next sidecar width */ }
}
return null;
}
/// Persists a rebuilt index **atomically** beside the data —
/// starts only, in the width the payload size calls for — via a
/// uniquely named temporary and a rename, so a concurrent
/// reader sees either the previous index or the whole new one.
/// Best-effort: a read-only directory only costs the rebuild.
private static void persistIndex(Path indexPath, long[] starts, int width) {
Path temporary = indexPath.resolveSibling("." + indexPath.getFileName() + "."
+ ProcessHandle.current().pid() + "." + PERSIST_SEQUENCE.getAndIncrement() + ".tmp");
try {
ByteBuffer bytes = ByteBuffer.allocate(starts.length * width).order(ByteOrder.LITTLE_ENDIAN);
for (long start : starts) { if (width == 4) bytes.putInt((int) start); else bytes.putLong(start); }
Files.write(temporary, bytes.array());
try { Files.move(temporary, indexPath, StandardCopyOption.ATOMIC_MOVE, StandardCopyOption.REPLACE_EXISTING); }
catch (IOException notAtomic) { Files.move(temporary, indexPath, StandardCopyOption.REPLACE_EXISTING); }
} catch (IOException ignored) {
try { Files.deleteIfExists(temporary); } catch (IOException alsoIgnored) { }
}
}
/// The local arm of `load_or_build_local_offsets`: read the
/// size-appropriate sidecar beside the data when it is no older than
/// the data, else walk the dimension headers and persist the rebuilt
/// index atomically for the next open.
private static long[] loadLocalOffsets(ByteStorage data, URI source, long payload, String expected) {
Path dataPath = Path.of(source);
int width = "i32".equals(expected) ? 4 : 8;
Path indexPath = dataPath.resolveSibling("IDXFOR__" + dataPath.getFileName() + "." + expected);
long[] fromSidecar = loadLocalIndex(indexPath, dataPath, width);
if (fromSidecar != null) return dropSentinel(fromSidecar, payload);
long[] walked = walkOffsets(data, source.toString());
if (walked == null) return null;
persistIndex(indexPath, walked, width);
return walked;
}
/// Normalizes a parsed sidecar to record starts: a trailing
/// entry equal to the payload size is unambiguously an
/// end-of-data sentinel — no record can start there — published
/// by the `N + 1`-entry layout, and is dropped.
private static long[] dropSentinel(long[] entries, long payload) {
return entries.length > 0 && entries[entries.length - 1] == payload
? Arrays.copyOf(entries, entries.length - 1) : entries;
}
/// Builds record starts by walking the dimension headers of a
/// local variable-length file that published no sidecar.
private long[] walkOffsets() {
int width;
try { width = ElementType.forExtension(facet.source().toString()).width(); }
catch (VectorDataException unsupported) { return null; }
ByteStorage storage = data();
long size = storage.size();
long offset = 0;
long[] starts = new long[16];
int found = 0;
while (offset + 4 <= size) {
if (found == starts.length) starts = Arrays.copyOf(starts, found * 2);
starts[found++] = offset;
int dim = storage.read(offset, 4).order(ByteOrder.LITTLE_ENDIAN).getInt();
if (dim < 0) return null;
offset += 4 + (long) dim * width;
}
return offset == size ? Arrays.copyOf(starts, found) : null;
}
private static long[] readEntries(ByteStorage index, int width) {
long size = index.size();
if (size == 0 || size % width != 0 || size / width > Integer.MAX_VALUE) return null;
long[] entries = new long[(int) (size / width)];
int at = 0;
int block = 1 << 20;
for (long offset = 0; offset < size; offset += block) {
ByteBuffer bytes = index.read(offset, (int) Math.min(block, size - offset)).order(ByteOrder.LITTLE_ENDIAN);
if (width == 4) while (bytes.remaining() >= 4) entries[at++] = Integer.toUnsignedLong(bytes.getInt());
else while (bytes.remaining() >= 8) entries[at++] = bytes.getLong();
}
/// Loads a sidecar file beside local data, or `null` when it is
/// absent, stale (older than the data it describes), or not a whole
/// number of offsets — all of which send the caller back to the walk
/// rather than yielding a reader that cannot see the records the
/// tail describes.
private static long[] loadLocalIndex(Path indexPath, Path dataPath, int width) {
try {
if (!Files.isRegularFile(indexPath)) return null;
if (Files.getLastModifiedTime(indexPath).compareTo(Files.getLastModifiedTime(dataPath)) < 0) return null;
byte[] raw = Files.readAllBytes(indexPath);
if (raw.length == 0 || raw.length % width != 0) return null;
ByteBuffer bytes = ByteBuffer.wrap(raw).order(ByteOrder.LITTLE_ENDIAN);
long[] entries = new long[raw.length / width];
for (int at = 0; at < entries.length; at++)
entries[at] = width == 4 ? Integer.toUnsignedLong(bytes.getInt()) : bytes.getLong();
return entries;
} catch (IOException unreadable) { return null; }
}
/// Persists a rebuilt index **atomically** beside the data — starts
/// only, in the width the payload size calls for — via a uniquely
/// named temporary and a rename, so a concurrent reader sees either
/// the previous index or the whole new one. Best-effort: a
/// read-only directory only costs the rebuild.
private static void persistIndex(Path indexPath, long[] starts, int width) {
Path temporary = indexPath.resolveSibling("." + indexPath.getFileName() + "."
+ ProcessHandle.current().pid() + "." + PERSIST_SEQUENCE.getAndIncrement() + ".tmp");
try {
ByteBuffer bytes = ByteBuffer.allocate(starts.length * width).order(ByteOrder.LITTLE_ENDIAN);
for (long start : starts) { if (width == 4) bytes.putInt((int) start); else bytes.putLong(start); }
Files.write(temporary, bytes.array());
try { Files.move(temporary, indexPath, StandardCopyOption.ATOMIC_MOVE, StandardCopyOption.REPLACE_EXISTING); }
catch (IOException notAtomic) { Files.move(temporary, indexPath, StandardCopyOption.REPLACE_EXISTING); }
} catch (IOException ignored) {
try { Files.deleteIfExists(temporary); } catch (IOException alsoIgnored) { }
}
}
/// Normalizes a parsed sidecar to record starts: a trailing entry
/// equal to the payload size is unambiguously an end-of-data
/// sentinel — no record can start there — published by the
/// `N + 1`-entry layout, and is dropped.
private static long[] dropSentinel(long[] entries, long payload) {
return entries.length > 0 && entries[entries.length - 1] == payload
? Arrays.copyOf(entries, entries.length - 1) : entries;
}
/// Builds record starts by walking the dimension headers of a local
/// variable-length file that published no sidecar.
private static long[] walkOffsets(ByteStorage storage, String source) {
int width;
try { width = ElementType.forExtension(source).width(); }
catch (VectorDataException unsupported) { return null; }
long size = storage.size();
long offset = 0;
long[] starts = new long[16];
int found = 0;
while (offset + 4 <= size) {
if (found == starts.length) starts = Arrays.copyOf(starts, found * 2);
starts[found++] = offset;
int dim = storage.read(offset, 4).order(ByteOrder.LITTLE_ENDIAN).getInt();
if (dim < 0) return null;
offset += 4 + (long) dim * width;
}
return offset == size ? Arrays.copyOf(starts, found) : null;
}
private static long[] readEntries(ByteStorage index, int width) {
long size = index.size();
if (size == 0 || size % width != 0 || size / width > Integer.MAX_VALUE) return null;
long[] entries = new long[(int) (size / width)];
int at = 0;
int block = 1 << 20;
for (long offset = 0; offset < size; offset += block) {
ByteBuffer bytes = index.read(offset, (int) Math.min(block, size - offset)).order(ByteOrder.LITTLE_ENDIAN);
if (width == 4) while (bytes.remaining() >= 4) entries[at++] = Integer.toUnsignedLong(bytes.getInt());
else while (bytes.remaining() >= 8) entries[at++] = bytes.getLong();
}
return entries;
}
/// The sidecar entry width a payload of `fileSize` bytes calls
/// for: `i32` while every offset still fits one, `i64` beyond.
/// Shared by the local sidecar name and the remote sidecar probe so
@@ -240,38 +337,70 @@ public final class Prefetcher {
private static final java.util.concurrent.atomic.AtomicLong PERSIST_SEQUENCE = new java.util.concurrent.atomic.AtomicLong();
/// Ports `prefetch_plan_on`: resolves the window per interval,
/// merges ranges whose fetches would overlap, and prices the
/// result at chunk granularity. An empty window is a request for
/// the whole facet — it resolves to the whole byte range whatever
/// the format, with no ordinal mapping and no degrade.
/// The number of records a file holds, by its format's rule — the
/// cardinality probe for a bare series entry. A variable-length file
/// counts through its offsets, rebuilt by a walk when local and
/// unpublished. `null` when the file cannot be measured.
static Long recordsIn(String path, ByteStorage storage, VectorDataSettings settings, String identity) {
String ext = extensionOf(path);
int width;
try { width = ElementType.forExtension(path).width(); }
catch (VectorDataException unsupported) { return null; }
long total = storage.size();
if (isVvecExt(ext)) {
long[] starts = publishedOffsets(storage, URI.create(path), settings, identity);
return starts == null ? null : (long) starts.length;
}
if (isScalarExt(ext)) return total / width;
if (total < 4) return null;
int dim = storage.read(0, 4).order(ByteOrder.LITTLE_ENDIAN).getInt();
if (dim <= 0 || dim > 1_000_000) return null;
return total / (4 + (long) dim * width);
}
// -- Planning --
/// Ports `prefetch_plan_on`: resolves the window per interval across
/// the shards it spans, merges ranges whose fetches would overlap,
/// and prices the result at chunk granularity. An empty window is a
/// request for the whole facet — it resolves to every shard's whole
/// byte range whatever the format, with no ordinal mapping and no
/// degrade.
public static PrefetchPlan plan(FacetHandle handle, DSWindow window) {
ByteStorage storage = handle.data();
long facetBytes = storage.size();
long facetBytes = handle.totalSize();
if (window == null || window.isEmpty()) {
List<ByteRange> whole = List.of(new ByteRange(0, facetBytes));
RangeFill fill = storage.rangeFill(0, facetBytes);
return new PrefetchPlan(whole, whole, fill == null ? List.of() : List.of(fill), 0, false, facetBytes);
List<ShardRange> whole = new ArrayList<>();
List<RangeFill> fills = new ArrayList<>();
long[] sizes = handle.shardSizes();
for (int shard = 0; shard < sizes.length; shard++) {
whole.add(new ShardRange(shard, 0, sizes[shard]));
RangeFill fill = handle.shardRangeFill(shard, 0, sizes[shard]);
if (fill != null) fills.add(fill);
}
return new PrefetchPlan(whole, whole, fills, 0, false, facetBytes);
}
long prerequisite = 0;
List<ByteRange> requested = new ArrayList<>();
List<ShardRange> requested = new ArrayList<>();
for (DSWindow.Interval interval : window.intervals()) {
MappedRange mapped = recordRangeToBytes(handle, interval.minIncl(), interval.maxExcl());
MappedRanges mapped = recordRangeToBytes(handle, interval.minIncl(), interval.maxExcl());
// One unmappable interval makes the whole request a full
// download; a partial plan beside it would understate what
// is about to happen.
if (mapped == null) return new PrefetchPlan(List.of(), List.of(), List.of(), prerequisite, true, facetBytes);
prerequisite = Math.max(prerequisite, mapped.prerequisiteBytes());
requested.add(new ByteRange(mapped.byteStart(), mapped.byteEnd()));
requested.addAll(mapped.ranges());
}
// A source that cannot service byte ranges has no partial fetch
// to offer: honouring a window against it means the whole file.
if (!storage.rangeCapable() && !storage.isComplete())
return new PrefetchPlan(List.of(), List.of(), List.of(), prerequisite, true, facetBytes);
List<ByteRange> issued = coalesce(requested, storage.stats().chunkSize());
// A file that cannot service byte ranges has no partial fetch to
// offer: honouring a window that touches it means the whole file.
for (ShardRange range : requested) {
ByteStorage storage = handle.shardStorage(range.shard());
if (!storage.rangeCapable() && !storage.isComplete())
return new PrefetchPlan(List.of(), List.of(), List.of(), prerequisite, true, facetBytes);
}
List<ShardRange> issued = coalesce(requested, handle.chunkSize());
List<RangeFill> fills = new ArrayList<>();
for (ByteRange range : issued) {
RangeFill fill = storage.rangeFill(range.start(), range.end());
for (ShardRange range : issued) {
RangeFill fill = handle.shardRangeFill(range.shard(), range.start(), range.end());
if (fill != null) fills.add(fill);
}
return new PrefetchPlan(requested, issued, fills, prerequisite, false, facetBytes);
@@ -290,19 +419,20 @@ public final class Prefetcher {
/// Merges byte ranges whose fetches would overlap: the same chunk,
/// or adjacent chunks, which are contiguous on the device. A whole
/// chunk of gap is not bridged, because bridging it fetches a chunk
/// nobody asked for. With `chunkSize` `0` — local storage, which
/// has no chunks — merging falls back to plain byte overlap or
/// adjacency.
static List<ByteRange> coalesce(List<ByteRange> ranges, long chunkSize) {
List<ByteRange> sorted = new ArrayList<>();
for (ByteRange range : ranges) if (range.end() > range.start()) sorted.add(range);
/// nobody asked for. Ranges in different shards never merge — they
/// are in different files. With `chunkSize` `0` — local storage,
/// which has no chunks — merging falls back to plain byte overlap
/// or adjacency.
static List<ShardRange> coalesce(List<ShardRange> ranges, long chunkSize) {
List<ShardRange> sorted = new ArrayList<>();
for (ShardRange range : ranges) if (!range.isEmpty()) sorted.add(range);
if (sorted.size() < 2) return sorted;
sorted.sort(Comparator.comparingLong(ByteRange::start).thenComparingLong(ByteRange::end));
List<ByteRange> merged = new ArrayList<>();
for (ByteRange next : sorted) {
if (!merged.isEmpty() && joins(merged.get(merged.size() - 1).end(), next.start(), chunkSize)) {
ByteRange current = merged.get(merged.size() - 1);
merged.set(merged.size() - 1, new ByteRange(current.start(), Math.max(current.end(), next.end())));
sorted.sort(Comparator.comparingInt(ShardRange::shard).thenComparingLong(ShardRange::start).thenComparingLong(ShardRange::end));
List<ShardRange> merged = new ArrayList<>();
for (ShardRange next : sorted) {
ShardRange current = merged.isEmpty() ? null : merged.get(merged.size() - 1);
if (current != null && current.shard() == next.shard() && joins(current.end(), next.start(), chunkSize)) {
merged.set(merged.size() - 1, new ShardRange(current.shard(), current.start(), Math.max(current.end(), next.end())));
} else merged.add(next);
}
return merged;
@@ -313,66 +443,75 @@ public final class Prefetcher {
return nextStart <= currentEnd;
}
/// Maps a record range to a byte range, for formats where that
/// mapping is computable from the file alone. See the class
/// documentation for the per-format policy.
static MappedRange recordRangeToBytes(FacetHandle handle, long winStart, long winEnd) {
/// Maps a record range to byte ranges: one range for a single file,
/// one per shard the range spans for a series, each mapped by the
/// format's rule against its own file. Prerequisite bytes are
/// summed, not maxed: each shard's index is a separate read, and a
/// window touching three shards pays for three.
static MappedRanges recordRangeToBytes(FacetHandle handle, long winStart, long winEnd) {
if (winEnd <= winStart) return null;
String ext = extensionOf(handle.facet.source().toString());
if (isVvecExt(ext)) return vvecRangeToBytes(handle, winStart, winEnd);
if (isScalarExt(ext)) {
int width;
try { width = ElementType.forExtension(handle.facet.source().toString()).width(); }
catch (VectorDataException unsupported) { return null; }
long total = handle.data().size();
long byteStart = Math.min(saturatedMultiply(winStart, width), total);
long byteEnd = Math.min(saturatedMultiply(winEnd, width), total);
return byteStart >= byteEnd ? null : new MappedRange(byteStart, byteEnd, 0);
if (!handle.isSeries()) {
MappedRange mapped = mapInFile(handle.path(), winStart, winEnd, handle.data(), handle::offsets);
return mapped == null ? null : new MappedRanges(List.of(ShardRange.whole(mapped.byteStart(), mapped.byteEnd())), mapped.prerequisiteBytes());
}
return uniformRangeToBytes(handle, winStart, winEnd);
FacetSeries series = handle.series();
List<ShardRange> ranges = new ArrayList<>();
long prerequisite = 0;
for (Shards.SubWindow part : series.shards().decompose(winStart, winEnd)) {
int file = series.fileIndexOfShard(part.shard());
MappedRange mapped = mapInFile(series.filePath(file), part.fileLo(), part.fileHi(), series.file(file), () -> series.publishedOffsets(file));
if (mapped == null) return null;
prerequisite += mapped.prerequisiteBytes();
ranges.add(new ShardRange(part.shard(), mapped.byteStart(), mapped.byteEnd()));
}
return ranges.isEmpty() ? null : new MappedRanges(ranges, prerequisite);
}
/// The uniform-stride mapping: `4 + dim × elemSize`, with `dim`
/// read from the header at byte 0 — the same first-chunk read any
/// reader does on first access, so nothing to report as a
/// prerequisite.
private static MappedRange uniformRangeToBytes(FacetHandle handle, long winStart, long winEnd) {
/// Maps a record range **within one file** to bytes, for formats
/// where that mapping is computable from the file alone. `offsets`
/// is consulted only for variable-length formats, and is a supplier
/// so the caller decides where the index comes from: a facet
/// handle's cache for a single file, the series' per-file cache for
/// a shard.
static MappedRange mapInFile(String path, long lo, long hi, ByteStorage storage, Supplier<long[]> offsets) {
if (hi <= lo) return null;
String ext = extensionOf(path);
int width;
try { width = ElementType.forExtension(handle.facet.source().toString()).width(); }
try { width = ElementType.forExtension(path).width(); }
catch (VectorDataException unsupported) { return null; }
ByteStorage storage = handle.data();
long total = storage.size();
if (isVvecExt(ext)) {
// Exact, not estimated, and a window past the last record
// ends at the file, not at a record that does not exist.
long[] starts = offsets.get();
if (starts == null || starts.length == 0 || lo >= starts.length) return null;
long byteStart = starts[(int) lo];
long byteEnd = hi >= starts.length ? total : starts[(int) hi];
// The index had to be read whole to answer this at all.
return byteEnd <= byteStart ? null : new MappedRange(byteStart, byteEnd, starts.length * 8L);
}
if (isScalarExt(ext)) {
long byteStart = Math.min(saturatedMultiply(lo, width), total);
long byteEnd = Math.min(saturatedMultiply(hi, width), total);
return byteStart >= byteEnd ? null : new MappedRange(byteStart, byteEnd, 0);
}
// The uniform-stride mapping: `dim` read from the header at byte
// 0 — the same first-chunk read any reader does on first access,
// so nothing to report as a prerequisite.
if (total < 4) return null;
int dim = storage.read(0, 4).order(ByteOrder.LITTLE_ENDIAN).getInt();
if (dim <= 0 || dim > 1_000_000) return null; // sanity vs corrupt header
long bytesPerRecord = 4 + (long) dim * width;
long byteStart = Math.min(saturatedMultiply(winStart, bytesPerRecord), total);
long byteEnd = Math.min(saturatedMultiply(winEnd, bytesPerRecord), total);
long byteStart = Math.min(saturatedMultiply(lo, bytesPerRecord), total);
long byteEnd = Math.min(saturatedMultiply(hi, bytesPerRecord), total);
return byteStart >= byteEnd ? null : new MappedRange(byteStart, byteEnd, 0);
}
/// Maps through the offset index: exact, not estimated, and a
/// window past the last record ends at the file, not at a record
/// that does not exist.
private static MappedRange vvecRangeToBytes(FacetHandle handle, long winStart, long winEnd) {
CachedOffsets cached = handle.offsets();
if (cached == null || cached.starts().length == 0) return null;
long[] starts = cached.starts();
long count = starts.length;
if (winStart >= count) return null;
long byteStart = starts[(int) winStart];
long byteEnd = winEnd >= count ? handle.data().size() : starts[(int) winEnd];
if (byteEnd <= byteStart) return null;
// The index had to be read whole to answer this at all. Record
// starts × 8, so both sidecar layouts report the same figure.
return new MappedRange(byteStart, byteEnd, starts.length * 8L);
}
private static long saturatedMultiply(long a, long b) {
try { return Math.multiplyExact(a, b); } catch (ArithmeticException overflow) { return Long.MAX_VALUE; }
}
private static String extensionOf(String source) {
static String extensionOf(String source) {
String value = source.toLowerCase(Locale.ROOT);
int slash = Math.max(value.lastIndexOf('/'), value.lastIndexOf('\\'));
if (slash >= 0) value = value.substring(slash + 1);
@@ -386,7 +525,7 @@ public final class Prefetcher {
/// this catalog format, `ivec`/`ivecs` records are length-qualified
/// but fixed throughout — ground-truth neighbor files — and map at
/// the uniform header stride like every other xvec.
private static boolean isVvecExt(String ext) { return ext.contains("vvec"); }
static boolean isVvecExt(String ext) { return ext.contains("vvec"); }
private static boolean isScalarExt(String ext) { return ext.matches("u8|i8|u16|i16|u32|i32|u64|i64"); }
static boolean isScalarExt(String ext) { return ext.matches("u8|i8|u16|i16|u32|i32|u64|i64"); }
}
@@ -0,0 +1,94 @@
/*
* 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.internal;
import io.nosqlbench.vectordata.CacheStats;
import io.nosqlbench.vectordata.ElementType;
import io.nosqlbench.vectordata.PrebufferProgress;
import io.nosqlbench.vectordata.VectorDataException;
import io.nosqlbench.vectordata.VectorReader;
import io.nosqlbench.vectordata.VectorReaders;
import java.util.concurrent.atomic.AtomicReferenceArray;
/// A fixed-record facet spread across several files, presented as one
/// reader: `count()` is the series total, `dimension()` the shared
/// dimension, and `get(o)` resolves the owning shard and reads from its
/// file. Callers that never ask about layout never learn it exists.
///
/// Every file is opened at construction to check that it agrees on
/// dimension and element width — a disagreement means these files are
/// not one facet, and is refused rather than tolerated.
public final class ShardedVectorReader<A> implements VectorReader<A> {
private final FacetSeries series;
private final AtomicReferenceArray<VectorReader<A>> readers;
private final ElementType type;
private final boolean scalar;
private final int dimension;
private final long count;
public ShardedVectorReader(FacetSeries series) {
this.series = series;
this.readers = new AtomicReferenceArray<>(series.fileCount());
this.scalar = VectorReaders.isScalar(series.filePath(0));
VectorReader<A> first = reader(0);
this.type = first.elementType();
this.dimension = first.dimension();
for (int file = 1; file < series.fileCount(); file++) {
VectorReader<A> other = reader(file);
if (other.dimension() != dimension || other.elementType() != type)
throw new VectorDataException("shard file " + file + " ('" + series.filePath(file) + "') has dimension " + other.dimension()
+ " of " + other.elementType() + ", but the series is dimension " + dimension + " of " + type
+ " — every shard of a facet must share one shape");
}
this.count = series.shards().count();
}
/// Whether the files are packed scalars rather than xvec records.
public boolean isScalar() { return scalar; }
@SuppressWarnings("unchecked")
private VectorReader<A> reader(int file) {
VectorReader<A> existing = readers.get(file);
if (existing != null) return existing;
VectorReader<A> opened = (VectorReader<A>) VectorReaders.over(series.file(file), series.filePath(file));
readers.compareAndSet(file, null, opened);
return readers.get(file);
}
private VectorReader<A> at(long index, long[] fileOrdinal) {
Shards.Located located = series.shards().locate(index);
if (located == null) throw new IndexOutOfBoundsException("Vector index " + index + " outside 0.." + (count - 1));
fileOrdinal[0] = located.fileOrdinal();
return reader(series.fileIndexOfShard(located.shard()));
}
@Override public long count() { return count; }
@Override public int dimension() { return dimension; }
@Override public A get(long index) { long[] at = new long[1]; return at(index, at).get(at[0]); }
@Override public void get(long index, A target) { long[] at = new long[1]; at(index, at).get(at[0], target); }
/// Fetches every file whole; the window-scoped fetch is the view's
/// prefetch, planned against the facet's declared window.
@Override public void prebuffer(PrebufferProgress progress) {
for (int file = 0; file < series.fileCount(); file++) series.file(file).prebuffer(progress);
}
@Override public boolean isComplete() {
for (int file = 0; file < series.fileCount(); file++) if (!series.file(file).isComplete()) return false;
return true;
}
@Override public CacheStats cacheStats() { return series.cacheStats(); }
@Override public ElementType elementType() { return type; }
}
@@ -0,0 +1,72 @@
/*
* 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.internal;
import io.nosqlbench.vectordata.CacheStats;
import io.nosqlbench.vectordata.ElementType;
import io.nosqlbench.vectordata.PrebufferProgress;
import io.nosqlbench.vectordata.VectorReaders;
import io.nosqlbench.vectordata.VvecReader;
import java.util.concurrent.atomic.AtomicReferenceArray;
/// A variable-length facet spread across several files, presented as
/// one reader. Each **file** has its own `IDXFOR__` sidecar whose
/// offsets are local to that file, so a window touching one shard loads
/// one file's index rather than the whole facet's — and two shards
/// slicing one file load that index once. Files open on first read.
public final class ShardedVvecReader<A> implements VvecReader<A> {
private final FacetSeries series;
private final AtomicReferenceArray<VvecReader<A>> readers;
private final ElementType type;
private final long count;
public ShardedVvecReader(FacetSeries series) {
this.series = series;
this.readers = new AtomicReferenceArray<>(series.fileCount());
this.type = ElementType.forExtension(series.filePath(0));
this.count = series.shards().count();
}
@SuppressWarnings("unchecked")
private VvecReader<A> reader(int file) {
VvecReader<A> existing = readers.get(file);
if (existing != null) return existing;
VvecReader<A> opened = (VvecReader<A>) VectorReaders.openVvec(series.fileUri(file), series.settings(), series.identity());
readers.compareAndSet(file, null, opened);
return readers.get(file);
}
private VvecReader<A> at(long index, long[] fileOrdinal) {
Shards.Located located = series.shards().locate(index);
if (located == null) throw new IndexOutOfBoundsException("vvec index " + index + " outside 0.." + (count - 1));
fileOrdinal[0] = located.fileOrdinal();
return reader(series.fileIndexOfShard(located.shard()));
}
@Override public long count() { return count; }
@Override public int dimensionAt(long index) { long[] at = new long[1]; return at(index, at).dimensionAt(at[0]); }
@Override public A get(long index) { long[] at = new long[1]; return at(index, at).get(at[0]); }
@Override public void prebuffer(PrebufferProgress progress) {
for (int file = 0; file < series.fileCount(); file++) reader(file).prebuffer(progress);
}
@Override public boolean isComplete() {
for (int file = 0; file < series.fileCount(); file++) if (!reader(file).isComplete()) return false;
return true;
}
@Override public CacheStats cacheStats() { return series.cacheStats(); }
@Override public ElementType elementType() { return type; }
}
@@ -0,0 +1,370 @@
/*
* 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.internal;
import io.nosqlbench.vectordata.DSWindow;
import io.nosqlbench.vectordata.VectorDataException;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.Locale;
/// Ordinal mapping for multi-file facets, mirroring the `vectordata-rs`
/// shard model.
///
/// A facet may be one file or a series of them. Either way its records
/// form one dense, gapless ordinal space, and this class turns a global
/// ordinal in that space into the file that holds it and the ordinal
/// within that file. Three coordinate levels, never conflated:
///
/// ```text
/// global ordinal o
/// ↓ locate — which shard, how far into it
/// local ordinal l within shard s
/// ↓ + entries[s].fileBase — the entry window's lower bound
/// file ordinal f within the file s is drawn from
/// ↓ the format's record→byte rule — see Prefetcher
/// byte offset
/// ```
///
/// **Shards and files are counted separately.** A shard is a contiguous
/// run of ordinals; a file is where bytes live. Two shards may be drawn
/// from one file at different windows, in which case they are two
/// shards and one file. Ordinal mapping is also separate from source
/// resolution: the map answers "which shard, and how far into it"; the
/// entries answer "which file, and how far into that".
public final class Shards {
/// The literal token marking the shard-index field in a uniform
/// source. Exactly four `N`s: the width is fixed, so there is no
/// `NNN` or `NNNNN` form to accept.
public static final String SHARD_FIELD = "NNNN";
/// How a global ordinal becomes a shard index and a local ordinal.
/// The two arms are resolved once, when the declaration is
/// realized, and are thereafter a single dispatch — never a general
/// path with a fast case tested for on every lookup.
sealed interface OrdinalMap permits Uniform, Explicit {
long total();
int shardCount();
long shardBase(int shard);
long shardLen(int shard);
/// `{shard, local}`, or `null` past the end — never a clamp,
/// because a silently clamped ordinal reads the wrong record.
long[] locate(long ordinal);
}
/// Every shard but the last holds exactly `stride` ordinals, so the
/// lookup is division and remainder: O(1), no allocation, no search.
record Uniform(long stride, int count, long total) implements OrdinalMap {
@Override public int shardCount() { return count; }
@Override public long shardBase(int shard) { return (long) shard * stride; }
@Override public long shardLen(int shard) { return Math.min(shardBase(shard) + stride, total) - shardBase(shard); }
@Override public long[] locate(long ordinal) {
return ordinal < 0 || ordinal >= total ? null : new long[] {ordinal / stride, ordinal % stride};
}
}
/// Shard lengths are uneven, so the lookup is a binary search over
/// prefix sums. `starts` has `count + 1` entries: it opens at `0`,
/// closes at `total`, and is strictly increasing.
record Explicit(long[] starts, long total) implements OrdinalMap {
@Override public int shardCount() { return starts.length - 1; }
@Override public long shardBase(int shard) { return starts[shard]; }
@Override public long shardLen(int shard) { return starts[shard + 1] - starts[shard]; }
@Override public long[] locate(long ordinal) {
if (ordinal < 0 || ordinal >= total) return null;
// The last index whose start is <= ordinal.
int found = Arrays.binarySearch(starts, ordinal);
int shard = found >= 0 ? found : -found - 2;
return new long[] {shard, ordinal - starts[shard]};
}
}
/// Builds the map from per-shard lengths, collapsing to [Uniform]
/// whenever the lengths permit it. Uniformity is a property of the
/// lengths, not of how the series was spelled: a list of
/// evenly-sized files — what an importer routinely produces — gets
/// the O(1) map. `null` if any length is zero: a zero-length shard
/// contributes no ordinals and would put two shards at one
/// prefix-sum boundary.
static OrdinalMap fromLengths(long[] lens) {
if (lens.length == 0) return null;
long total = 0;
for (long len : lens) { if (len <= 0) return null; total += len; }
long stride = lens[0];
boolean uniform = lens[lens.length - 1] <= stride;
for (int i = 0; i < lens.length - 1 && uniform; i++) uniform = lens[i] == stride;
if (uniform) return new Uniform(stride, lens.length, total);
long[] starts = new long[lens.length + 1];
for (int i = 0; i < lens.length; i++) starts[i + 1] = starts[i] + lens[i];
return new Explicit(starts, total);
}
/// One shard's binding to the file it is drawn from: the resolved
/// source path — the window has already been folded into `fileBase`
/// and `len` — the first **file** ordinal this shard reads (zero for
/// a whole-file shard, the entry window's lower bound for a sliced
/// one), and the ordinals it holds.
public record Entry(String source, long fileBase, long len) { }
/// Where a global ordinal lives: the shard, its offset within that
/// shard, and its ordinal within the file the shard is drawn from.
public record Located(int shard, long local, long fileOrdinal) { }
/// One shard's slice of a decomposed window: half-open bounds in the
/// shard's local ordinals, and the same bounds in the file's
/// ordinals, which is what a format's record→byte rule wants.
public record SubWindow(int shard, long localLo, long localHi, long fileLo, long fileHi) { }
private final OrdinalMap map;
private final List<Entry> entries;
private Shards(OrdinalMap map, List<Entry> entries) { this.map = map; this.entries = List.copyOf(entries); }
/// Builds from per-shard entries, deriving the map from their
/// lengths; `null` when the entries are empty or any is zero-length.
static Shards of(List<Entry> entries) {
long[] lens = new long[entries.size()];
for (int i = 0; i < lens.length; i++) lens[i] = entries.get(i).len();
OrdinalMap map = fromLengths(lens);
return map == null ? null : new Shards(map, entries);
}
/// The shard entries, indexed by shard number.
public List<Entry> entries() { return entries; }
/// Total records across the series.
public long count() { return map.total(); }
public int shardCount() { return map.shardCount(); }
/// First global ordinal of a shard.
public long shardBase(int shard) { return map.shardBase(shard); }
/// Ordinals held by a shard.
public long shardLen(int shard) { return map.shardLen(shard); }
/// Whether the uniform O(1) map is in force.
public boolean isUniform() { return map instanceof Uniform; }
/// Whether this facet is a single whole file — the canonical shape
/// for everything written before sharding existed.
public boolean isSingleFile() { return entries.size() == 1 && entries.get(0).fileBase() == 0; }
/// Resolves a global ordinal through all three coordinate levels, or
/// `null` past the end of the series.
public Located locate(long ordinal) {
long[] at = map.locate(ordinal);
if (at == null) return null;
int shard = (int) at[0];
return new Located(shard, at[1], entries.get(shard).fileBase() + at[1]);
}
/// Decomposes the window `[lo, hi)` into per-shard sub-windows.
/// Empty for an empty or out-of-range window; never emits an empty
/// sub-window, so a window ending exactly on a shard boundary does
/// not produce a trailing no-op.
public List<SubWindow> decompose(long lo, long hi) {
long end = Math.min(hi, map.total());
if (lo < 0 || lo >= end) return List.of();
int first = (int) map.locate(lo)[0];
int last = (int) map.locate(end - 1)[0];
List<SubWindow> parts = new ArrayList<>();
for (int shard = first; shard <= last; shard++) {
long base = map.shardBase(shard), len = map.shardLen(shard);
long localLo = Math.max(lo, base) - base, localHi = Math.min(end, base + len) - base;
long fileBase = entries.get(shard).fileBase();
parts.add(new SubWindow(shard, localLo, localHi, fileBase + localLo, fileBase + localHi));
}
return parts;
}
// -- Declarations --
/// A facet declaration in the form the shard model consumes: source
/// strings in ordinal order — one for a single file or a uniform
/// series, several for an explicit one — whether they came from an
/// array, and the uniform layout fields.
public record Declaration(List<String> sources, boolean isArray, Long shardStride, Integer shardCount, Long recordCount) {
public Declaration { sources = List.copyOf(sources); }
}
/// Answers how many records a file holds. Consulted only for an
/// entry that declares neither a window nor a count — the
/// local-convenience spelling; every other entry is self-describing
/// and never reaches this. Throws with a reason when it cannot.
@FunctionalInterface
public interface Cardinality { long of(SourceSpec source); }
/// Whether a source string declares a uniform series.
public static boolean hasShardField(String source) { return source.contains(SHARD_FIELD); }
/// Substitutes the shard index into a uniform source's `NNNN` field.
/// The four digits appear here and nowhere else, so a writer and a
/// reader cannot drift apart on the width.
public static String shardFilename(String pattern, int index) {
return pattern.replaceFirst(SHARD_FIELD, String.format(Locale.ROOT, "%04d", index));
}
/// Realizes a declaration into the ordinal model. This is where
/// declaration shape stops mattering: whatever form the facet was
/// written in — single file, uniform series, explicit series, pinned
/// or bare — the result is one [Shards], and no stage above this
/// branches on the spelling again. A non-canonical single-shard
/// declaration is accepted and realized as the single-file facet it
/// describes; reporting it is a validator's job, not a loader's.
public static Shards realize(String facet, Declaration decl, Cardinality cardinality) {
boolean uniform = !decl.sources().isEmpty() && hasShardField(decl.sources().get(0));
boolean layout = decl.shardStride() != null || decl.shardCount() != null;
if (decl.isArray() && layout)
throw new VectorDataException("facet '" + facet + "': an array source cannot also carry shard_stride or shard_count"
+ " — the array already states the layout");
if (uniform) return realizeUniform(facet, decl);
if (layout) throw incomplete(facet, "shard_stride/shard_count without a '" + SHARD_FIELD + "' field in the source");
return realizeEntries(facet, decl, cardinality);
}
/// Marker detail for a length that only the file can answer.
static final String UNPROBED = "deferred: needs the file";
/// Checks everything about a declaration that can be checked without
/// touching a file: mixed forms, half-stated uniform fields, a count
/// that contradicts its interval, a total that contradicts its
/// entries. A length that only the file can answer is not a fault
/// here — it is the local-convenience spelling, resolved at open.
/// Implemented by running [#realize] with a probe that declines,
/// rather than by restating its rules: two copies of these checks
/// would be two chances to disagree.
public static void validate(String facet, Declaration decl) {
try { realize(facet, decl, source -> { throw new VectorDataException(UNPROBED); }); }
catch (VectorDataException fault) { if (!fault.getMessage().endsWith(UNPROBED)) throw fault; }
}
private static VectorDataException incomplete(String facet, String detail) {
return new VectorDataException("facet '" + facet + "': incomplete shard declaration — " + detail);
}
private static VectorDataException at(String facet, int index, String detail) {
return new VectorDataException("facet '" + facet + "' shard " + index + ": " + detail);
}
/// The uniform form: filenames derived from `NNNN`, lengths from the
/// stride and the declared total.
private static Shards realizeUniform(String facet, Declaration decl) {
if (decl.isArray()) throw incomplete(facet, "a '" + SHARD_FIELD + "' field belongs to a single source string, not an array");
String pattern = decl.sources().get(0);
String ambiguous = ambiguousTokenBeforeShardField(pattern);
if (ambiguous != null)
throw incomplete(facet, "the token '" + ambiguous + "' before the '" + SHARD_FIELD + "' field is all digits, so the"
+ " derived filenames have two readings and neither is decidable; give it a non-numeric prefix");
if (decl.shardStride() == null) throw incomplete(facet, "'" + SHARD_FIELD + "' without shard_stride");
if (decl.shardCount() == null) throw incomplete(facet, "'" + SHARD_FIELD + "' without shard_count");
if (decl.recordCount() == null) throw incomplete(facet, "a sharded facet must declare record_count");
long stride = decl.shardStride(); int count = decl.shardCount(); long total = decl.recordCount();
if (stride <= 0) throw incomplete(facet, "shard_stride must be greater than zero");
if (count <= 0) throw incomplete(facet, "shard_count must be greater than zero");
// Lengths follow from the stride and the declared total: every
// shard but the last is `stride`, and the last is whatever
// remains. A remainder outside `1..=stride` means the
// declaration disagrees with itself.
long full = (count - 1L) * stride;
if (total <= full || total > full + stride)
throw new VectorDataException("facet '" + facet + "': declared record_count " + total + " does not match the "
+ (full + stride) + " records its shards hold");
List<Entry> entries = new ArrayList<>(count);
for (int i = 0; i < count; i++) {
SourceSpec source;
try { source = SourceSpec.parse(shardFilename(pattern, i)); }
catch (VectorDataException malformed) { throw at(facet, i, malformed.getMessage()); }
entries.add(new Entry(source.path(), 0, i < count - 1 ? stride : total - full));
}
Shards shards = of(entries);
if (shards == null) throw at(facet, 0, "resolves to zero records");
return shards;
}
/// The single-file and explicit forms: one entry per listed source.
private static Shards realizeEntries(String facet, Declaration decl, Cardinality cardinality) {
List<Entry> entries = new ArrayList<>(decl.sources().size());
for (int i = 0; i < decl.sources().size(); i++) {
SourceSpec parsed;
try { parsed = SourceSpec.parse(decl.sources().get(i)); }
catch (VectorDataException malformed) { throw at(facet, i, malformed.getMessage()); }
DSWindow window = parsed.window();
if (window.intervals().size() > 1)
throw at(facet, i, "an entry carries at most one interval — list the file once per interval instead");
DSWindow.Interval interval = window.isEmpty() ? null : window.intervals().get(0);
long fileBase, len;
if (interval != null && interval.maxExcl() != Long.MAX_VALUE) {
fileBase = interval.minIncl();
len = Math.max(0, interval.maxExcl() - interval.minIncl());
// The edifying count checks whatever it annotates: the
// interval's length when there is one.
if (parsed.declaredCount() != null && parsed.declaredCount() != len)
throw at(facet, i, "declared count " + parsed.declaredCount() + " does not match its length " + len);
} else if (interval == null && parsed.declaredCount() != null) {
fileBase = 0; len = parsed.declaredCount();
} else {
// Nothing self-describing — a bare name, or an open-ended
// window: the length can only come from the file.
// Refused for a remote *series*, where building the map
// would open every shard before a single record is read,
// which is the expense a declaration exists to avoid. A
// single remote file is not that case: its reader must
// open it to read anything, so its count is the same open
// rather than an extra one.
if (decl.isArray() && isRemote(parsed.path()))
throw at(facet, i, "remote entry '" + parsed.path() + "' states no window or count, and its length cannot be"
+ " learned without fetching");
long records;
try { records = cardinality.of(parsed); }
catch (VectorDataException unavailable) { throw at(facet, i, "cannot establish record count — " + unavailable.getMessage()); }
fileBase = interval == null ? 0 : interval.minIncl();
len = Math.max(0, records - fileBase);
}
if (len == 0) throw at(facet, i, "resolves to zero records");
entries.add(new Entry(parsed.path(), fileBase, len));
}
Shards shards = of(entries);
if (shards == null) throw at(facet, 0, "resolves to zero records");
// A declared total is checked, never preferred. Only a series is
// required to carry one; a plain single-file facet has always
// been spelled without.
if (decl.recordCount() != null && decl.recordCount() != shards.count())
throw new VectorDataException("facet '" + facet + "': declared record_count " + decl.recordCount()
+ " does not match the " + shards.count() + " records its shards hold");
if (decl.isArray() && decl.recordCount() == null) throw incomplete(facet, "a sharded facet must declare record_count");
return shards;
}
/// The all-digit token, if any, immediately before the `__NNNN`
/// field of a pattern's stem. Such a name has two readings — the
/// token could itself be a shard index — so it is refused rather
/// than guessed at.
static String ambiguousTokenBeforeShardField(String pattern) {
int colon = pattern.indexOf(':');
String path = colon < 0 ? pattern : pattern.substring(0, colon);
int dot = path.lastIndexOf('.');
String stem = dot < 0 ? path : path.substring(0, dot);
if (!stem.endsWith(SHARD_FIELD)) return null;
String before = stem.substring(0, stem.length() - SHARD_FIELD.length());
if (!before.endsWith("__")) return null;
before = before.substring(0, before.length() - 2);
int separator = before.lastIndexOf("__");
String token = separator < 0 ? before : before.substring(separator + 2);
return !token.isEmpty() && token.chars().allMatch(Character::isDigit) ? token : null;
}
static boolean isRemote(String path) {
String lower = path.toLowerCase(Locale.ROOT);
return lower.startsWith("http://") || lower.startsWith("https://");
}
}
@@ -0,0 +1,96 @@
/*
* 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.internal;
import io.nosqlbench.vectordata.DSWindow;
import io.nosqlbench.vectordata.VectorDataException;
/// A parsed facet source string, in the grammar `vectordata-rs` parses:
/// a path, an optional ordinal window, and an optional declared
/// cardinality, in increasing specificity —
///
/// | Spelling | Cardinality |
/// |------------------|----------------------------------|
/// | `a.u8` | discovered by opening the file |
/// | `a.u8=4194304` | declared |
/// | `a.u8[0..1M]` | implied by the interval |
/// | `a.u8[0..1M]=1M` | implied *and* declared, checked |
///
/// The `=<count>` suffix is taken from the end and accepts the same
/// suffixes intervals do, under two restrictions. A source containing
/// `?` is never split on `=`, since that is the key/value separator
/// inside a URL query string — such a source declares its cardinality
/// by window instead. And an `=` whose tail does not parse as a positive
/// count stays in the path: `weird=name.u8` is a filename, not a
/// malformed count, so recognition is positive rather than a rule the
/// path has to escape.
public record SourceSpec(String path, DSWindow window, String windowText, Long declaredCount) {
/// Parses a source string. Only a suffix that *looks* like a window
/// and is malformed fails — a plain path always passes through — so
/// an error here names the broken window rather than turning it
/// into "no such file".
public static SourceSpec parse(String raw) {
String text = raw.trim();
Long count = null;
if (text.indexOf('?') < 0) {
int eq = text.lastIndexOf('=');
if (eq > 0 && eq < text.length() - 1) {
Long parsed = tryCount(text.substring(eq + 1));
if (parsed != null) { count = parsed; text = text.substring(0, eq).trim(); }
}
}
String[] split = splitWindowSuffix(text);
DSWindow window = split[1] == null ? DSWindow.ALL : DSWindow.parse(split[1]);
return new SourceSpec(split[0], window, split[1], count);
}
/// Renders the spelling back: path, then `[window]`, then `=count`.
public String render() {
StringBuilder text = new StringBuilder(path);
if (windowText != null) text.append('[').append(windowText).append(']');
if (declaredCount != null) text.append('=').append(declaredCount);
return text.toString();
}
/// This spec with its path replaced — the window and count travel
/// with it, which is how a relative source becomes an absolute one
/// without losing what it said.
public SourceSpec withPath(String replacement) { return new SourceSpec(replacement, window, windowText, declaredCount); }
private static Long tryCount(String tail) {
try { long value = DSWindow.parseNumberWithSuffix(tail); return value > 0 ? value : null; }
catch (VectorDataException notACount) { return null; }
}
/// Splits the documented window-suffix sugar off a source string —
/// `base.fvec[0..1M)` names the file plus a record window, with
/// either bracket kind on either side. The outer delimiters are
/// structural: they separate the path from the window and do not
/// affect interval bound semantics. Returns `{path, windowOrNull}`.
static String[] splitWindowSuffix(String source) {
if (!source.endsWith("]") && !source.endsWith(")")) return new String[] {source, null};
int bracket = source.indexOf('['); int paren = source.indexOf('(');
int open = bracket < 0 ? paren : paren < 0 ? bracket : Math.min(bracket, paren);
if (open <= 0) return new String[] {source, null};
String inner = source.substring(open + 1, source.length() - 1);
try { DSWindow.parse(inner); }
catch (VectorDataException malformed) {
throw new VectorDataException("source '" + source + "' has a malformed window: " + malformed.getMessage());
}
return new String[] {source.substring(0, open), inner};
}
}
@@ -159,7 +159,7 @@ class PrefetchRemoteIntegrationTest {
assertFalse(plan.degradesToFullDownload(), "a published index makes a remote vvec windowable");
assertEquals(1, plan.requestedRanges().size());
assertTrue(plan.prerequisiteBytes() > 0, "the index had to be read, and the plan says so");
ByteRange requested = plan.requestedRanges().get(0);
ShardRange requested = plan.requestedRanges().get(0);
assertTrue(requested.end() > requested.start() && requested.end() <= plan.facetBytes());
assertTrue(requested.length() < plan.facetBytes(), "50 of 400 ragged records must be a fraction of the facet");
@@ -279,7 +279,7 @@ class PrefetchRemoteIntegrationTest {
PrefetchPlan plan = view.prefetchPlan("neighbor_indices", DSWindow.parse("[0..3)"));
assertFalse(plan.degradesToFullDownload(), "a uniform ivec is windowable by its header stride");
assertEquals(List.of(new ByteRange(0, 3L * recordBytes)), plan.requestedRanges());
assertEquals(List.of(ShardRange.whole(0, 3L * recordBytes)), plan.requestedRanges());
view.prefetch("neighbor_indices", DSWindow.parse("[0..3)"), WholeFacetFallback.REFUSE);
assertArrayEquals(new int[] {2, 3}, java.util.Arrays.copyOfRange((int[]) view.neighborIndices().get(2), 0, 2),
@@ -302,4 +302,125 @@ class PrefetchRemoteIntegrationTest {
assertTrue(view.prefetchPlan("metadata_predicates", DSWindow.ALL).isResident(),
"consenting to the whole facet fetches the whole facet");
}
// -- Facets spread across several remote files --
/// dim 8 → 36 bytes per record.
private static final int SERIES_BPR = 36;
private static byte[] seriesShard(int records, int first) {
ByteBuffer bytes = ByteBuffer.allocate(records * SERIES_BPR).order(ByteOrder.LITTLE_ENDIAN);
for (int r = 0; r < records; r++) { bytes.putInt(8); for (int d = 0; d < 8; d++) bytes.putFloat((first + r) * 100f + d); }
return bytes.array();
}
/// Publishes `big__0000..` shards of `records` each, `.mref`-published
/// so chunk residency is real, and returns the server's base URL.
private String publishSeries(int shards, int records, boolean mrefs) throws Exception {
Path published = Files.createDirectories(temporary.resolve("pub"));
for (int shard = 0; shard < shards; shard++) {
byte[] payload = seriesShard(records, shard * records);
Path file = published.resolve(String.format("big__%04d.fvec", shard));
Files.write(file, payload);
if (mrefs) Files.write(published.resolve(file.getFileName() + ".mref"), FixtureSupport.mref(payload, CHUNK));
}
HttpServer server = serveDirectory(published, new AtomicInteger(), true);
return "http://127.0.0.1:" + server.getAddress().getPort() + "/";
}
private TestDataView seriesView(String name, String yaml, String profile) throws Exception {
Path dataset = Files.createDirectories(temporary.resolve(name));
Files.writeString(dataset.resolve("dataset.yaml"), yaml);
VectorDataSettings settings = VectorDataSettings.builder().cacheDirectory(temporary.resolve("cache")).build();
return TestDataGroup.load(dataset.resolve("dataset.yaml").toUri(), settings).profile(profile);
}
private static String uniformYaml(String base, int shards, int records, String extra) {
return """
name: remote-series
profiles:
default:
base_vectors:
source: %sbig__NNNN.fvec
shard_stride: %d
shard_count: %d
record_count: %d
""".formatted(base, records, shards, shards * records) + extra;
}
@Test void aUniformSeriesReadsOverHttp() throws Exception {
String base = publishSeries(2, 25, false);
VectorReader<float[]> reader = seriesView("uniform-http", uniformYaml(base, 2, 25, ""), "default").baseVectors();
assertEquals(50, reader.count(), "the count spans the series");
assertEquals(8, reader.dimension());
for (int i = 0; i < 50; i++) { assertEquals(i * 100f, reader.get(i)[0], "record " + i + " over HTTP"); assertEquals(i * 100f + 7, reader.get(i)[7]); }
for (int i = 49; i >= 0; i--) assertEquals(i * 100f, reader.get(i)[0]);
}
@Test void anExplicitSeriesReadsOverHttp() throws Exception {
String base = publishSeries(2, 25, false);
VectorReader<float[]> reader = seriesView("explicit-http", """
name: explicit
profiles:
default:
base_vectors:
source:
- %sbig__0000.fvec=25
- %sbig__0001.fvec=25
record_count: 50
""".formatted(base, base), "default").baseVectors();
assertEquals(50, reader.count());
for (int i : new int[] {0, 24, 25, 49}) assertEquals(i * 100f, reader.get(i)[0], "record " + i);
}
@Test void aWindowInsideOneShardDoesNotDegradeToAFullDownload() throws Exception {
String base = publishSeries(2, 25, true);
TestDataView view = seriesView("one-shard", uniformYaml(base, 2, 25, ""), "default");
PrefetchPlan plan = view.prefetchPlan("base_vectors", DSWindow.parse("5..10"));
assertFalse(plan.degradesToFullDownload(), "a five-record window must not price as the whole facet");
assertEquals(1, plan.fills().size(), "a window inside shard 0 touches one shard");
assertEquals(0, plan.byteRanges().get(0).shard());
}
@Test void aWindowAcrossTheSeamPlansOneFillPerShard() throws Exception {
String base = publishSeries(2, 25, true);
TestDataView view = seriesView("seam", uniformYaml(base, 2, 25, ""), "default");
PrefetchPlan plan = view.prefetchPlan("base_vectors", DSWindow.parse("20..30"));
assertFalse(plan.degradesToFullDownload());
assertEquals(2, plan.fills().size(), "the window spans both shards");
assertEquals(2, plan.requests());
view.prefetch("base_vectors", DSWindow.parse("20..30"), WholeFacetFallback.REFUSE);
assertTrue(view.prefetchPlan("base_vectors", DSWindow.parse("20..30")).isResident());
}
@Test void aFacetByteRangePrebuffersTheShardItLivesIn() throws Exception {
String base = publishSeries(3, 400, true);
TestDataView view = seriesView("deep", uniformYaml(base, 3, 400, ""), "default");
VectorReader<float[]> reader = view.baseVectors();
long before = reader.cacheStats().cachedBytes();
view.prefetch("base_vectors", DSWindow.parse("1150..1160"), WholeFacetFallback.REFUSE);
CacheStats after = reader.cacheStats();
assertTrue(after.cachedBytes() > before, "a range in the last shard fetched nothing (" + before + " → " + after.cachedBytes() + ")");
assertFalse(reader.isComplete(), "only the range's chunks were fetched");
assertTrue(after.cachedBytes() - before < after.totalBytes() / 4,
"fetched " + (after.cachedBytes() - before) + " of " + after.totalBytes() + " bytes for ten records");
assertEquals(1155 * 100f, reader.get(1155)[0]);
}
@Test void cacheStatsForASeriesCoverEveryShard() throws Exception {
String base = publishSeries(3, 400, true);
TestDataView view = seriesView("stats", uniformYaml(base, 3, 400, ""), "default");
CacheStats stats = view.baseVectors().cacheStats();
assertEquals(3L * 400 * SERIES_BPR, stats.totalBytes(), "every shard's bytes");
assertEquals(CHUNK, stats.chunkSize());
assertEquals(AccessMode.MERKLE_HASHED, stats.accessMode());
assertFalse(stats.complete());
}
@Test void aMissingShardOverHttpIsNamed() throws Exception {
String base = publishSeries(2, 25, false);
TestDataView view = seriesView("missing", uniformYaml(base, 3, 25, ""), "default");
VectorDataException missing = assertThrows(VectorDataException.class, view::baseVectors);
assertTrue(missing.getMessage().contains("big__0002.fvec"), "the message must name the missing shard: " + missing.getMessage());
}
}
@@ -65,12 +65,12 @@ class PrefetchWindowsTest {
@Test void anArbitraryWindowResolvesWithoutAProfileDeclaringIt() throws Exception {
PrefetchPlan plan = view().prefetchPlan("base_vectors", DSWindow.parse("10..20"));
assertFalse(plan.degradesToFullDownload(), "a uniform-stride facet must be windowable on demand");
assertEquals(List.of(new ByteRange(10 * BPR, 20 * BPR)), plan.byteRanges(), "records map to bytes at 4 + dim*elem_size");
assertEquals(List.of(ShardRange.whole(10 * BPR, 20 * BPR)), plan.byteRanges(), "records map to bytes at 4 + dim*elem_size");
}
@Test void aMultiIntervalWindowResolvesEveryInterval() throws Exception {
PrefetchPlan plan = view().prefetchPlan("base_vectors", DSWindow.parse("[0..10, 50..60]"));
assertEquals(List.of(new ByteRange(0, 10 * BPR), new ByteRange(50 * BPR, 60 * BPR)), plan.byteRanges(),
assertEquals(List.of(ShardRange.whole(0, 10 * BPR), ShardRange.whole(50 * BPR, 60 * BPR)), plan.byteRanges(),
"both intervals must survive; the reader's single-window limit is a reader limit, not a fetch limit");
}
@@ -78,11 +78,11 @@ class PrefetchWindowsTest {
TestDataView view = view();
PrefetchPlan touching = view.prefetchPlan("base_vectors", DSWindow.parse("[0..10, 10..20]"));
assertEquals(2, touching.requestedRanges().size(), "two intervals were asked for");
assertEquals(List.of(new ByteRange(0, 20 * BPR)), touching.byteRanges(), "and they merge into one request");
assertEquals(List.of(ShardRange.whole(0, 20 * BPR)), touching.byteRanges(), "and they merge into one request");
assertEquals(1, touching.requests());
PrefetchPlan apart = view.prefetchPlan("base_vectors", DSWindow.parse("[0..10, 80..90]"));
assertEquals(2, apart.requestedRanges().size());
assertEquals(List.of(new ByteRange(0, 10 * BPR), new ByteRange(80 * BPR, 90 * BPR)), apart.byteRanges(),
assertEquals(List.of(ShardRange.whole(0, 10 * BPR), ShardRange.whole(80 * BPR, 90 * BPR)), apart.byteRanges(),
"nothing bridges a gap this size on chunkless local storage");
assertEquals(2, apart.requests());
}
@@ -90,12 +90,12 @@ class PrefetchWindowsTest {
@Test void overlappingIntervalsBecomeOneRequest() throws Exception {
PrefetchPlan plan = view().prefetchPlan("base_vectors", DSWindow.parse("[0..30, 20..40]"));
assertEquals(1, plan.requests());
assertEquals(List.of(new ByteRange(0, 40 * BPR)), plan.byteRanges());
assertEquals(List.of(ShardRange.whole(0, 40 * BPR)), plan.byteRanges());
}
@Test void anEmptyWindowCoversTheWholeFacet() throws Exception {
PrefetchPlan plan = view().prefetchPlan("base_vectors", DSWindow.ALL);
assertEquals(List.of(new ByteRange(0, 100 * BPR)), plan.byteRanges());
assertEquals(List.of(ShardRange.whole(0, 100 * BPR)), plan.byteRanges());
assertFalse(plan.degradesToFullDownload(), "asking for everything is a request, not a fallback");
}
@@ -112,7 +112,7 @@ class PrefetchWindowsTest {
@Test void aWindowPastTheEndClampsToTheFacet() throws Exception {
PrefetchPlan plan = view().prefetchPlan("base_vectors", DSWindow.parse("90..500"));
assertEquals(List.of(new ByteRange(90 * BPR, 100 * BPR)), plan.byteRanges());
assertEquals(List.of(ShardRange.whole(90 * BPR, 100 * BPR)), plan.byteRanges());
}
/// Ragged variable-length records with a published sentinel-form
@@ -148,7 +148,7 @@ class PrefetchWindowsTest {
VvecFixture fixture = vvecView(new int[] {1, 7, 3, 9, 2, 5, 4, 8});
PrefetchPlan plan = fixture.view().prefetchPlan("metadata_content", DSWindow.parse("2..5"));
assertFalse(plan.degradesToFullDownload(), "an indexed vvec facet is windowable");
assertEquals(List.of(new ByteRange(fixture.offsets()[2], fixture.offsets()[5])), plan.byteRanges(),
assertEquals(List.of(ShardRange.whole(fixture.offsets()[2], fixture.offsets()[5])), plan.byteRanges(),
"records 2..5 are exactly offsets[2]..offsets[5]");
}
@@ -165,7 +165,7 @@ class PrefetchWindowsTest {
@Test void aVvecWindowPastTheEndEndsAtTheFile() throws Exception {
VvecFixture fixture = vvecView(new int[] {1, 7, 3, 9});
PrefetchPlan plan = fixture.view().prefetchPlan("metadata_content", DSWindow.parse("2..99"));
assertEquals(List.of(new ByteRange(fixture.offsets()[2], fixture.fileLength())), plan.byteRanges());
assertEquals(List.of(ShardRange.whole(fixture.offsets()[2], fixture.fileLength())), plan.byteRanges());
}
/// A local vvec facet that published no sidecar is walked in place,
@@ -190,7 +190,7 @@ class PrefetchWindowsTest {
TestDataView view = TestDataGroup.load(dataset.toUri(), settings()).profile("default");
PrefetchPlan plan = view.prefetchPlan("metadata_content", DSWindow.parse("1..4"));
assertFalse(plan.degradesToFullDownload());
assertEquals(List.of(new ByteRange(offsets[1], offsets[4])), plan.byteRanges());
assertEquals(List.of(ShardRange.whole(offsets[1], offsets[4])), plan.byteRanges());
// The walk persists its result beside the data — starts only,
// in the width the payload size calls for — so it is paid once
// rather than per view.
@@ -220,7 +220,7 @@ class PrefetchWindowsTest {
int width = Integer.parseInt(scalar[1]);
PrefetchPlan plan = view.prefetchPlan("facet_" + scalar[0], DSWindow.parse("2..5"));
assertFalse(plan.degradesToFullDownload(), scalar[0] + " is trivially windowable");
assertEquals(List.of(new ByteRange(2L * width, 5L * width)), plan.byteRanges(), scalar[0]);
assertEquals(List.of(ShardRange.whole(2L * width, 5L * width)), plan.byteRanges(), scalar[0]);
assertEquals(0, plan.prerequisiteBytes(), "a fixed stride costs nothing to know");
}
}
@@ -244,7 +244,7 @@ class PrefetchWindowsTest {
TestDataView view = TestDataGroup.load(dataset.toUri(), settings()).profile("default");
PrefetchPlan plan = view.prefetchPlan("metadata_results", DSWindow.parse("2..5"));
assertFalse(plan.degradesToFullDownload());
assertEquals(List.of(new ByteRange(2 * 4, 5 * 4)), plan.byteRanges(), "scalars map at ordinal times width, no header");
assertEquals(List.of(ShardRange.whole(2 * 4, 5 * 4)), plan.byteRanges(), "scalars map at ordinal times width, no header");
}
private TestDataView parquetView() throws IOException {
@@ -287,7 +287,7 @@ class PrefetchWindowsTest {
TestDataView view = parquetView();
PrefetchPlan plan = view.prefetchPlan("metadata_content", DSWindow.ALL);
assertFalse(plan.degradesToFullDownload(), "asking for everything and getting everything is not a degrade");
assertEquals(List.of(new ByteRange(0, 64)), plan.byteRanges());
assertEquals(List.of(ShardRange.whole(0, 64)), plan.byteRanges());
view.prefetch("metadata_content", DSWindow.ALL, WholeFacetFallback.REFUSE);
}
@@ -304,9 +304,9 @@ class PrefetchWindowsTest {
@Test void aBackgroundPrefetchReportsItsPlanBeforeFinishing() throws Exception {
PrefetchHandle handle = view().prefetchInBackground("base_vectors", DSWindow.parse("10..20"), WholeFacetFallback.REFUSE);
assertEquals(List.of(new ByteRange(10 * BPR, 20 * BPR)), handle.plan().byteRanges(), "the plan is known before the fetch is");
assertEquals(List.of(ShardRange.whole(10 * BPR, 20 * BPR)), handle.plan().byteRanges(), "the plan is known before the fetch is");
PrefetchReport report = handle.join();
assertEquals(List.of(new ByteRange(10 * BPR, 20 * BPR)), report.planned().byteRanges());
assertEquals(List.of(ShardRange.whole(10 * BPR, 20 * BPR)), report.planned().byteRanges());
assertTrue(handle.isDone());
}
@@ -0,0 +1,431 @@
/*
* 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;
import org.junit.jupiter.api.Tag;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.io.TempDir;
import java.io.IOException;
import java.nio.file.Files;
import java.nio.file.Path;
import java.util.List;
import static org.junit.jupiter.api.Assertions.*;
/// A facet spread across several files reads as one dense ordinal
/// space, plans by the shards a window spans, and refuses a declaration
/// that disagrees with itself or with its files. Mirrors the
/// `sharded_facets` suite of `vectordata-rs`.
@Tag("unit")
class ShardedFacetsTest {
@TempDir Path temporary;
/// dim 4 → 4 + 4×4 bytes per record.
private static final long BPR = 20;
private VectorDataSettings settings() {
return VectorDataSettings.builder().cacheDirectory(temporary.resolve("cache")).build();
}
/// Records numbered from `first`: element `d` of record `r` is `r*100 + d`.
private static void fvec(Path dir, String name, int records, int first) throws IOException {
float[][] values = new float[records][4];
for (int r = 0; r < records; r++) for (int d = 0; d < 4; d++) values[r][d] = (first + r) * 100f + d;
FixtureSupport.fvec(dir, name, values);
}
private Path dataset(String name, String yaml) throws IOException {
Path dir = Files.createDirectories(temporary.resolve(name));
Files.writeString(dir.resolve("dataset.yaml"), yaml);
return dir;
}
/// Three shards of 100, 100, and 40 records.
private Path uniformSeries(String extra) throws IOException {
Path dir = Files.createDirectories(temporary.resolve("uniform"));
fvec(dir, "base__0000.fvec", 100, 0);
fvec(dir, "base__0001.fvec", 100, 100);
fvec(dir, "base__0002.fvec", 40, 200);
Files.writeString(dir.resolve("dataset.yaml"), """
name: sharded
profiles:
default:
base_vectors:
source: base__NNNN.fvec
shard_stride: 100
shard_count: 3
record_count: 240
""" + extra);
return dir;
}
private TestDataView view(Path dir) { return view(dir, "default"); }
private TestDataView view(Path dir, String profile) { return TestDataGroup.load(dir.toUri(), settings()).profile(profile); }
@Test void aUniformSeriesOpensAndReportsTheWholeFacet() throws Exception {
TestDataView view = view(uniformSeries(""));
VectorReader<float[]> base = view.baseVectors();
assertEquals(240, base.count(), "the count spans the series");
assertEquals(4, base.dimension());
assertEquals(0f, base.get(0)[0]);
assertEquals(99 * 100f, base.get(99)[0], "the last record of shard 0");
assertEquals(100 * 100f, base.get(100)[0], "the first record of shard 1");
assertEquals(239 * 100f + 3, base.get(239)[3], "the last record of the last, short shard");
assertThrows(IndexOutOfBoundsException.class, () -> base.get(240), "past the series end is out of bounds, never a clamp");
PrefetchPlan whole = view.prefetchPlan("base_vectors", DSWindow.ALL);
assertEquals(240 * BPR, whole.facetBytes(), "the facet is every shard's bytes, not the first shard's");
assertTrue(whole.isResident());
}
@Test void aSeriesReadsIdenticallyToTheSingleFileItWasSplitFrom() throws Exception {
TestDataView series = view(uniformSeries(""));
Path single = Files.createDirectories(temporary.resolve("single"));
fvec(single, "base.fvec", 240, 0);
TestDataView whole = view(dataset("single", "name: whole\nprofiles:\n default:\n base_vectors: base.fvec\n"));
VectorReader<float[]> a = series.baseVectors(), b = whole.baseVectors();
assertEquals(b.count(), a.count());
for (long i = 0; i < a.count(); i++) assertArrayEquals(b.get(i), a.get(i), "record " + i);
for (long i = a.count() - 1; i >= 0; i -= 7) assertArrayEquals(b.get(i), a.get(i), "record " + i + ", read backwards");
}
@Test void aSeriesWhoseFilesContradictItsDeclarationIsRefused() throws Exception {
Path dir = Files.createDirectories(temporary.resolve("bad"));
fvec(dir, "base__0000.fvec", 100, 0);
Files.writeString(dir.resolve("dataset.yaml"), """
name: bad
profiles:
default:
base_vectors:
source: base__NNNN.fvec
shard_stride: 100
shard_count: 3
record_count: 9999
""");
VectorDataException refused = assertThrows(VectorDataException.class, () -> TestDataGroup.load(dir.toUri(), settings()));
assertTrue(refused.getMessage().contains("record_count"), "the message must name the disagreement: " + refused.getMessage());
}
@Test void anExplicitSeriesOpensFromNamedFiles() throws Exception {
Path dir = uniformSeries("");
Files.writeString(dir.resolve("dataset.yaml"), """
name: explicit
profiles:
default:
base_vectors:
source:
- base__0000.fvec=100
- base__0001.fvec=100
- base__0002.fvec=40
record_count: 240
""");
VectorReader<float[]> base = view(dir).baseVectors();
assertEquals(240, base.count());
for (long i : new long[] {0, 99, 100, 199, 200, 239}) assertEquals(i * 100f, base.get(i)[0], "record " + i);
}
@Test void bareNamesResolveToTheSameFacetAsCountedOnes() throws Exception {
Path dir = uniformSeries("");
Files.writeString(dir.resolve("dataset.yaml"), """
name: bare
profiles:
default:
base_vectors:
source: [ base__0000.fvec, base__0001.fvec, base__0002.fvec ]
record_count: 240
""");
VectorReader<float[]> base = view(dir).baseVectors();
assertEquals(240, base.count(), "a local bare entry is measured by opening its file");
assertEquals(239 * 100f, base.get(239)[0]);
}
@Test void twoShardsOfOneFileReadTheirOwnWindows() throws Exception {
Path dir = Files.createDirectories(temporary.resolve("sliced"));
int[] values = new int[10_000];
for (int i = 0; i < values.length; i++) values[i] = i;
FixtureSupport.scalarI32(dir, "corpus.u32", values);
Files.writeString(dir.resolve("dataset.yaml"), """
name: sliced
profiles:
default:
metadata_layout:
source:
- corpus.u32[0..10]=10
- corpus.u32[9990..10000]=10
record_count: 20
""");
TestDataView view = view(dir);
VectorReader<?> reader = view.openFacet("metadata_layout");
assertEquals(20, reader.count(), "the global space is the entries' lengths laid end to end");
assertEquals(0L, ((Number) reader.get(0)).longValue());
assertEquals(9L, ((Number) reader.get(9)).longValue());
assertEquals(9990L, ((Number) reader.get(10)).longValue(), "the second entry starts at its window's lower bound");
assertEquals(9999L, ((Number) reader.get(19)).longValue());
assertTrue(reader.isComplete(), "a local sliced facet is complete");
view.prebuffer(PrebufferProgress.NONE);
PrefetchPlan plan = view.prefetchPlan("metadata_layout", DSWindow.ALL);
assertEquals(2, plan.byteRanges().size(), "two shards, one file: byte ranges are per shard");
assertEquals(new ShardRange(0, 0, 10_000 * 4), plan.byteRanges().get(0), "a whole-facet plan names each shard by its file's extent");
assertEquals(new ShardRange(1, 0, 10_000 * 4), plan.byteRanges().get(1));
PrefetchPlan windowed = view.prefetchPlan("metadata_layout", DSWindow.parse("5..15"));
assertEquals(List.of(new ShardRange(0, 5 * 4, 10 * 4), new ShardRange(1, 9990 * 4, 9995 * 4)), windowed.byteRanges(),
"a window maps each shard's local ordinals through its entry's file base");
}
@Test void aMissingShardIsNamedRatherThanReadAsEmptiness() throws Exception {
Path dir = Files.createDirectories(temporary.resolve("gap"));
fvec(dir, "base__0000.fvec", 100, 0);
Files.writeString(dir.resolve("dataset.yaml"), """
name: gap
profiles:
default:
base_vectors:
source: base__NNNN.fvec
shard_stride: 100
shard_count: 2
record_count: 150
""");
TestDataView view = view(dir);
VectorDataException missing = assertThrows(VectorDataException.class, view::baseVectors,
"a declared shard that is absent must be reported");
assertTrue(missing.getMessage().contains("base__0001.fvec"), "the message must name the missing shard: " + missing.getMessage());
assertEquals(0, view.prefetchPlan("base_vectors", DSWindow.ALL).facetBytes(),
"a facet that cannot be sized answers 0 through the infallible accessor, never a shorter facet");
}
@Test void aProfileWindowClipsTheSeriesNotAShard() throws Exception {
Path dir = Files.createDirectories(temporary.resolve("win"));
fvec(dir, "base__0000.fvec", 100, 0);
fvec(dir, "base__0001.fvec", 100, 100);
Files.writeString(dir.resolve("dataset.yaml"), """
name: win
profiles:
default:
base_vectors:
source: base__NNNN.fvec
shard_stride: 100
shard_count: 2
record_count: 200
window: 50..150
""");
VectorReader<float[]> windowed = view(dir).baseVectors();
assertEquals(100, windowed.count(), "the window spans a shard boundary");
assertEquals(50 * 100f, windowed.get(0)[0]);
assertEquals(99 * 100f, windowed.get(49)[0]);
assertEquals(100 * 100f, windowed.get(50)[0]);
assertEquals(149 * 100f, windowed.get(99)[0]);
assertThrows(IndexOutOfBoundsException.class, () -> windowed.get(100));
}
/// A window suffix on a uniform pattern is the facet window: a
/// pattern names no file, so there is no entry for it to bound.
@Test void aWindowOnAUniformPatternBoundsTheReader() throws Exception {
Path dir = uniformSeries("");
Files.writeString(dir.resolve("dataset.yaml"), """
name: suffix
profiles:
default:
base_vectors:
source: base__NNNN.fvec[0..150]
shard_stride: 100
shard_count: 3
record_count: 240
""");
VectorReader<float[]> base = view(dir).baseVectors();
assertEquals(150, base.count(), "the suffix bounds the facet, not shard 0");
assertEquals(149 * 100f, base.get(149)[0], "record 149 lives in shard 1");
assertEquals("0..150", view(dir).facet("base_vectors").orElseThrow().window(), "the suffix is reported as the facet window");
Files.writeString(dir.resolve("dataset.yaml"), """
name: twice
profiles:
default:
base_vectors:
source: base__NNNN.fvec[0..150]
shard_stride: 100
shard_count: 3
record_count: 240
window: 0..100
""");
VectorDataException twice = assertThrows(VectorDataException.class, () -> TestDataGroup.load(dir.toUri(), settings()));
assertTrue(twice.getMessage().contains("twice"), twice.getMessage());
}
@Test void aWindowInsideOneShardPlansOnlyThatShard() throws Exception {
PrefetchPlan plan = view(uniformSeries("")).prefetchPlan("base_vectors", DSWindow.parse("5..10"));
assertFalse(plan.degradesToFullDownload());
assertEquals(List.of(new ShardRange(0, 5 * BPR, 10 * BPR)), plan.byteRanges());
}
@Test void aWindowAcrossASeamPlansOneRangePerShard() throws Exception {
PrefetchPlan plan = view(uniformSeries("")).prefetchPlan("base_vectors", DSWindow.parse("80..220"));
assertEquals(List.of(
new ShardRange(0, 80 * BPR, 100 * BPR),
new ShardRange(1, 0, 100 * BPR),
new ShardRange(2, 0, 20 * BPR)), plan.byteRanges(), "the window touches three shards");
assertEquals(140 * BPR, plan.byteRanges().stream().mapToLong(ShardRange::length).sum());
}
@Test void rangesInDifferentShardsNeverMerge() throws Exception {
Path dir = Files.createDirectories(temporary.resolve("nm"));
fvec(dir, "base__0000.fvec", 10, 0);
fvec(dir, "base__0001.fvec", 10, 10);
Files.writeString(dir.resolve("dataset.yaml"), """
name: nm
profiles:
default:
base_vectors:
source: base__NNNN.fvec
shard_stride: 10
shard_count: 2
record_count: 20
""");
PrefetchPlan plan = view(dir).prefetchPlan("base_vectors", DSWindow.parse("9..11"));
assertEquals(2, plan.byteRanges().size(), "adjacent bytes in different files are two fetches: " + plan.byteRanges());
assertEquals(0, plan.byteRanges().get(0).shard());
assertEquals(1, plan.byteRanges().get(1).shard());
assertEquals(1, plan.byteRanges().get(0).length() / BPR);
assertEquals(1, plan.byteRanges().get(1).length() / BPR);
}
@Test void aWholeFacetPlanNamesEveryShard() throws Exception {
PrefetchPlan plan = view(uniformSeries("")).prefetchPlan("base_vectors", DSWindow.ALL);
assertEquals(List.of(new ShardRange(0, 0, 100 * BPR), new ShardRange(1, 0, 100 * BPR), new ShardRange(2, 0, 40 * BPR)),
plan.byteRanges());
assertFalse(plan.degradesToFullDownload(), "asking for everything is a request, not a fallback");
}
@Test void aVvecSeriesReadsThroughPerFileIndexes() throws Exception {
Path dir = Files.createDirectories(temporary.resolve("vvec"));
int[][] first = {{0}, {1, 1}, {2, 2, 2}, {3}, {4, 4}};
int[][] second = {{5, 5, 5}, {6}, {7, 7}};
FixtureSupport.vvec(dir, "meta__0000.ivvec", first);
FixtureSupport.vvec(dir, "meta__0001.ivvec", second);
fvec(dir, "base.fvec", 8, 0);
Files.writeString(dir.resolve("dataset.yaml"), """
name: vvec
profiles:
default:
base_vectors: base.fvec
metadata_content:
source: meta__NNNN.ivvec
shard_stride: 5
shard_count: 2
record_count: 8
""");
TestDataView view = view(dir);
VvecReader<?> reader = view.openVariableFacet("metadata_content");
assertEquals(8, reader.count());
assertEquals(3, reader.dimensionAt(2));
assertArrayEquals(new int[] {4, 4}, (int[]) reader.get(4), "the last record of shard 0");
assertArrayEquals(new int[] {5, 5, 5}, (int[]) reader.get(5), "the first record of shard 1, through its own index");
assertArrayEquals(new int[] {7, 7}, (int[]) reader.get(7));
PrefetchPlan plan = view.prefetchPlan("metadata_content", DSWindow.parse("3..7"));
assertFalse(plan.degradesToFullDownload(), "each file's sidecar makes its shard windowable");
assertEquals(2, plan.byteRanges().size(), "the window spans both shards");
// Record starts × 8 per touched file, summed: 5 + 3 records.
assertEquals((5 + 3) * 8L, plan.prerequisiteBytes(), "each shard's index is a separate read, and a window touching two pays for two");
long firstFile = Files.size(dir.resolve("meta__0000.ivvec"));
assertEquals(new ShardRange(0, 4 + 4 + (4 + 8) + (4 + 12), firstFile), plan.byteRanges().get(0), "records 3..5 of file 0 run to its end");
assertEquals(new ShardRange(1, 0, (4 + 12) + (4 + 4)), plan.byteRanges().get(1), "records 0..2 of file 1");
}
// -- Declarations that disagree with themselves are refused at load --
private VectorDataException refusedAtLoad(String name, String facet) throws IOException {
Path dir = Files.createDirectories(temporary.resolve(name));
Files.writeString(dir.resolve("dataset.yaml"), "name: " + name + "\nprofiles:\n default:\n base_vectors:\n" + facet);
return assertThrows(VectorDataException.class, () -> TestDataGroup.load(dir.toUri(), settings()));
}
@Test void anArraySourceCannotAlsoCarryAUniformLayout() throws Exception {
VectorDataException mixed = refusedAtLoad("mixed", " source: [ a.fvec=1, b.fvec=1 ]\n shard_stride: 1\n record_count: 2\n");
assertTrue(mixed.getMessage().contains("array"), mixed.getMessage());
}
@Test void aPatternWithoutItsLayoutIsIncomplete() throws Exception {
VectorDataException incomplete = refusedAtLoad("nolayout", " source: base__NNNN.fvec\n record_count: 2\n");
assertTrue(incomplete.getMessage().contains("shard_stride"), incomplete.getMessage());
VectorDataException noPattern = refusedAtLoad("nopattern", " source: base.fvec\n shard_stride: 1\n shard_count: 2\n record_count: 2\n");
assertTrue(noPattern.getMessage().contains("NNNN"), noPattern.getMessage());
VectorDataException noTotal = refusedAtLoad("nototal", " source: base__NNNN.fvec\n shard_stride: 1\n shard_count: 2\n");
assertTrue(noTotal.getMessage().contains("record_count"), noTotal.getMessage());
}
@Test void aSliceCountThatDisagreesWithItsIntervalIsRefused() throws Exception {
VectorDataException mismatch = refusedAtLoad("slice", " source:\n - a.fvec[0..10]=11\n - b.fvec=5\n record_count: 16\n");
assertTrue(mismatch.getMessage().contains("declared count 11"), mismatch.getMessage());
}
@Test void aRemoteBareEntryInASeriesIsRefusedWithoutFetching() throws Exception {
VectorDataException unbounded = refusedAtLoad("remote",
" source: [ http://127.0.0.1:9/a.fvec, http://127.0.0.1:9/b.fvec ]\n record_count: 2\n");
assertTrue(unbounded.getMessage().contains("without fetching"), unbounded.getMessage());
}
@Test void anAllDigitTokenBeforeTheShardFieldIsRefused() throws Exception {
VectorDataException ambiguous = refusedAtLoad("ambiguous",
" source: p__1234__NNNN.fvec\n shard_stride: 1\n shard_count: 2\n record_count: 2\n");
assertTrue(ambiguous.getMessage().contains("two readings"), ambiguous.getMessage());
}
@Test void aPlainFacetIsUntouchedByAnyOfThis() throws Exception {
Path dir = Files.createDirectories(temporary.resolve("plain"));
fvec(dir, "base.fvec", 10, 0);
TestDataView view = view(dataset("plain", "name: plain\nprofiles:\n default:\n base_vectors: base.fvec\n"));
FacetDescriptor facet = view.facet("base_vectors").orElseThrow();
assertFalse(facet.isSeries());
assertNotNull(facet.source(), "a single file keeps its source");
assertEquals(List.of(ShardRange.whole(0, 10 * BPR)), view.prefetchPlan("base_vectors", DSWindow.ALL).byteRanges(),
"a single file is shard 0");
}
// -- format_version --
@Test void aDatasetFromTheFutureIsRefusedWithBothNumbers() throws Exception {
Path dir = dataset("future", "name: future\nformat_version: 99\nprofiles:\n default:\n base_vectors: base.fvec\n");
VectorDataException refused = assertThrows(VectorDataException.class, () -> TestDataGroup.load(dir.toUri(), settings()),
"a dataset above this build's version must not load at all");
assertTrue(refused.getMessage().contains("99"), "names what the dataset needs: " + refused.getMessage());
assertTrue(refused.getMessage().contains(String.valueOf(FormatVersion.SUPPORTED)), "names what this build supports: " + refused.getMessage());
}
@Test void aStatedVersionBelowTheContentIsRefused() throws Exception {
Path dir = Files.createDirectories(temporary.resolve("understated"));
Files.writeString(dir.resolve("dataset.yaml"), """
name: understated
format_version: 1
profiles:
default:
base_vectors:
source: base__NNNN.fvec
shard_stride: 1
shard_count: 2
record_count: 2
""");
VectorDataException refused = assertThrows(VectorDataException.class, () -> TestDataGroup.load(dir.toUri(), settings()));
assertTrue(refused.getMessage().contains("understate"), refused.getMessage());
}
@Test void anAbsentVersionIsNotAnUnderstatement() throws Exception {
TestDataView view = view(uniformSeries(""));
assertEquals(240, view.baseVectors().count(), "an unannotated sharded dataset loads: absence is not a claim");
Path generous = uniformSeries("");
Files.writeString(generous.resolve("dataset.yaml"), "format_version: 2\n" + Files.readString(generous.resolve("dataset.yaml")));
assertEquals(240, view(generous).baseVectors().count(), "a version equal to the content's requirement is accepted");
Path plain = dataset("generous", "format_version: 2\nname: g\nprofiles:\n default:\n base_vectors: base.fvec\n");
assertNotNull(TestDataGroup.load(plain.toUri(), settings()), "a version higher than the content requires is merely generous");
}
}
@@ -0,0 +1,158 @@
/*
* 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.internal;
import io.nosqlbench.vectordata.DSWindow;
import io.nosqlbench.vectordata.VectorDataException;
import io.nosqlbench.vectordata.VectorDataSettings;
import io.nosqlbench.vectordata.VectorReader;
import org.junit.jupiter.api.Tag;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.io.TempDir;
import java.nio.ByteBuffer;
import java.nio.ByteOrder;
import java.nio.file.Files;
import java.nio.file.Path;
import java.util.List;
import static org.junit.jupiter.api.Assertions.*;
/// The ordinal model on its own: how lengths choose a map, how a global
/// ordinal becomes a shard and a file ordinal, how a window decomposes,
/// how a source string is read — and the descriptor budget a series
/// lives within.
@Tag("unit")
class ShardsTest {
@TempDir Path temporary;
private static Shards.Entry entry(long len) { return new Shards.Entry("f" + len + ".fvec", 0, len); }
@Test void uniformLengthsMapInConstantTimeHoweverTheyWereSpelled() {
Shards shards = Shards.of(List.of(entry(100), entry(100), entry(40)));
assertTrue(shards.isUniform(), "uniformity is a property of the lengths, not the declaration form");
assertEquals(240, shards.count());
assertEquals(new Shards.Located(0, 0, 0), shards.locate(0));
assertEquals(new Shards.Located(0, 99, 99), shards.locate(99));
assertEquals(new Shards.Located(1, 0, 0), shards.locate(100));
assertEquals(new Shards.Located(2, 39, 39), shards.locate(239));
assertNull(shards.locate(240), "past the end is null, never a clamp");
assertNull(shards.locate(-1));
}
@Test void unevenLengthsSearchPrefixSums() {
Shards shards = Shards.of(List.of(entry(100), entry(50), entry(100)));
assertFalse(shards.isUniform(), "a short interior shard is genuinely uneven");
assertEquals(new Shards.Located(1, 49, 49), shards.locate(149));
assertEquals(new Shards.Located(2, 0, 0), shards.locate(150));
assertEquals(new Shards.Located(2, 99, 99), shards.locate(249));
assertNull(shards.locate(250));
}
@Test void aSlicedEntryOffsetsFileOrdinalsByItsWindow() {
Shards shards = Shards.of(List.of(new Shards.Entry("corpus.u32", 0, 10), new Shards.Entry("corpus.u32", 9990, 10)));
assertEquals(new Shards.Located(1, 3, 9993), shards.locate(13), "local ordinal plus the entry's file base");
assertFalse(shards.isSingleFile());
}
@Test void aWindowDecomposesIntoPerShardSubWindows() {
Shards shards = Shards.of(List.of(entry(100), entry(100), entry(50)));
assertEquals(List.of(
new Shards.SubWindow(0, 80, 100, 80, 100),
new Shards.SubWindow(1, 0, 100, 0, 100),
new Shards.SubWindow(2, 0, 20, 0, 20)), shards.decompose(80, 220));
assertEquals(List.of(new Shards.SubWindow(0, 0, 100, 0, 100)), shards.decompose(0, 100),
"a window ending on a seam produces no trailing no-op");
assertEquals(List.of(new Shards.SubWindow(2, 40, 50, 40, 50)), shards.decompose(240, 999), "clamped to the series");
assertTrue(shards.decompose(250, 260).isEmpty());
assertTrue(shards.decompose(20, 10).isEmpty());
}
@Test void aZeroLengthEntryIsNotAShard() {
assertNull(Shards.of(List.of(entry(10), entry(0))));
assertNull(Shards.of(List.of()));
}
@Test void sourceStringsCarryWindowsAndCounts() {
SourceSpec bare = SourceSpec.parse("a.u8");
assertEquals("a.u8", bare.path()); assertTrue(bare.window().isEmpty()); assertNull(bare.declaredCount());
SourceSpec counted = SourceSpec.parse("corpus-part-a.u8=4194304");
assertEquals("corpus-part-a.u8", counted.path()); assertEquals(4_194_304L, counted.declaredCount());
SourceSpec sliced = SourceSpec.parse("a.u8[0..1M]=1M");
assertEquals("a.u8", sliced.path()); assertEquals(1_000_000L, sliced.declaredCount());
assertEquals(new DSWindow.Interval(0, 1_000_000), sliced.window().intervals().get(0));
SourceSpec query = SourceSpec.parse("https://h/f.fvec?token=12345");
assertEquals("https://h/f.fvec?token=12345", query.path(), "a query string is never split on '='");
assertNull(query.declaredCount());
SourceSpec weird = SourceSpec.parse("weird=name.u8");
assertEquals("weird=name.u8", weird.path(), "an '=' whose tail is not a count stays in the path");
for (String raw : new String[] {"a.u8", "a.u8[0..100]", "a.u8=500", "a.u8[0..1000000]=1000000", "https://h/f.fvec", "b__0000.fvec"})
assertEquals(raw, SourceSpec.parse(raw).render(), "round trip of " + raw);
}
@Test void shardNamesAreFourDigitsExactly() {
assertEquals("base__0007.fvec", Shards.shardFilename("base__NNNN.fvec", 7));
assertEquals("base__0000.fvec", Shards.shardFilename("base__NNNN.fvec", 0));
assertTrue(Shards.hasShardField("x/base__NNNN.fvec"));
assertFalse(Shards.hasShardField("base__0007.fvec"));
assertEquals("1234", Shards.ambiguousTokenBeforeShardField("p__1234__NNNN.fvec"));
assertNull(Shards.ambiguousTokenBeforeShardField("p__abc__NNNN.fvec"));
assertNull(Shards.ambiguousTokenBeforeShardField("base__NNNN.fvec"));
}
@Test void realizationChecksWhatItCanWithoutAFile() {
Shards.Declaration remoteBare = new Shards.Declaration(List.of("http://h/a.fvec", "http://h/b.fvec"), true, null, null, 2L);
assertThrows(VectorDataException.class, () -> Shards.validate("base", remoteBare));
Shards.Declaration localBare = new Shards.Declaration(List.of("a.fvec", "b.fvec"), true, null, null, 2L);
assertDoesNotThrow(() -> Shards.validate("base", localBare), "a local bare entry is deferred to open, not a fault");
Shards.Declaration single = new Shards.Declaration(List.of("base.fvec"), false, null, null, null);
assertDoesNotThrow(() -> Shards.validate("base", single));
Shards.Declaration uniform = new Shards.Declaration(List.of("b__NNNN.fvec"), false, 100L, 3, 250L);
Shards realized = Shards.realize("base", uniform, source -> { throw new AssertionError("a uniform series opens nothing"); });
assertEquals(List.of(new Shards.Entry("b__0000.fvec", 0, 100), new Shards.Entry("b__0001.fvec", 0, 100), new Shards.Entry("b__0002.fvec", 0, 50)),
realized.entries());
VectorDataException short_ = assertThrows(VectorDataException.class,
() -> Shards.realize("base", new Shards.Declaration(List.of("b__NNNN.fvec"), false, 100L, 3, 200L), s -> 0));
assertTrue(short_.getMessage().contains("record_count"), "a total that leaves the last shard empty disagrees with itself: " + short_.getMessage());
}
@Test void theOpenFileCapIsDerivedAndLeavesHeadroom() {
assertEquals(256, FacetSeries.resolveOpenFileCap(null, 1024L), "a quarter of the soft limit");
assertEquals(256, FacetSeries.resolveOpenFileCap(null, null), "1024 when the limit is unknown");
assertEquals(8, FacetSeries.resolveOpenFileCap(null, 8L), "never below the floor");
assertEquals(3, FacetSeries.resolveOpenFileCap("3", 1024L), "the environment overrides");
assertEquals(256, FacetSeries.resolveOpenFileCap("nonsense", 1024L));
assertTrue(FacetSeries.openFileCap() >= 8);
}
@Test void aSeriesWiderThanTheDescriptorBudgetStillReads() throws Exception {
Path dir = Files.createDirectories(temporary.resolve("wide"));
for (int shard = 0; shard < 5; shard++) {
ByteBuffer bytes = ByteBuffer.allocate(10 * (4 + 4)).order(ByteOrder.LITTLE_ENDIAN);
for (int r = 0; r < 10; r++) { bytes.putInt(1); bytes.putFloat(shard * 10 + r); }
Files.write(dir.resolve(String.format("w__%04d.fvec", shard)), bytes.array());
}
VectorDataSettings settings = VectorDataSettings.builder().cacheDirectory(temporary.resolve("cache")).build();
Shards shards = Shards.realize("w", new Shards.Declaration(List.of(dir.toUri() + "w__NNNN.fvec"), false, 10L, 5, 50L), s -> 0);
FacetSeries series = new FacetSeries(shards, settings, "wide", 2);
VectorReader<float[]> reader = new ShardedVectorReader<>(series);
assertEquals(50, reader.count());
for (int i = 0; i < 50; i++) assertEquals((float) i, reader.get(i)[0], "record " + i);
for (int i = 49; i >= 0; i--) assertEquals((float) i, reader.get(i)[0], "record " + i + " backwards");
assertTrue(series.openFileCount() <= 2, "the series holds no more files than its budget: " + series.openFileCount());
assertEquals(50 * 8, series.tryTotalSize());
}
}