diff --git a/devdocs/vectordata/vectordata_remote_access.md b/devdocs/vectordata/vectordata_remote_access.md index 9c478ae63..a422f769e 100644 --- a/devdocs/vectordata/vectordata_remote_access.md +++ b/devdocs/vectordata/vectordata_remote_access.md @@ -142,10 +142,15 @@ expr forms (`{{= expr}}`, `{{name = expr}}`, `{{@name}}`); anything else passes through untouched. The vectordata expr functions: ```yaml -# Precache the whole facet before the run (empty window = everything): -# {{= prefetch("example:default", "base_vectors", "").rangesFetched()}} +# One-call warm-up: every facet the profile declares, each to the +# window the profile declares for it. This is the form to reach for — +# it never fetches more than the profile describes. +# {{= prefetchProfile("example:default")}} -# Ordinal-range warm-up, priced first: +# A cycle range, in the coordinates an activity already speaks: +# {{= prefetchCycles("example:default", "base_vectors", 0, 100000).rangesFetched()}} + +# An explicit window, priced first: # {{= prefetchPlan("example:default", "base_vectors", "[0..100k)").bytesToFetch()}} # {{= prefetch("example:default", "base_vectors", "[0..100k)").rangesFetched()}} @@ -163,6 +168,32 @@ passes through untouched. The vectordata expr functions: Diagnose expr processing with `dryrun=exprs` on the activity to dump the expression-processed workload and context. +### Three things to know when prefetching from a workload + +**Expressions evaluate when the workload loads, not when a block runs.** +The whole document is expression-processed before ops are parsed, so a +prefetch written into an op body fires on every activity start — +including `run tags='block:drop'` — no matter what tags that op carries. +That is fine for an up-front warm-up, but it is not a way to tie +fetching to a phase. To warm only when a phase actually uses the data, +let the bindings do it: `BaseVectors('ds:profile','[0..1M)','eager')` +warms when that binding resolves, which happens only for blocks that use +it, and `'background'` lets the fetch overlap the run. + +**An empty window means the whole facet, not the profile's window.** +`prefetch(spec,"base_vectors","")` on a sized profile fetches the entire +shared base file — which may be terabytes — because an empty window is a +request for everything rather than a fallback. Use `prefetchProfile`, +which resolves each facet to its declared window, or pass the window +explicitly. + +**An assignment sigil renders its value.** `{{p = "ds:" + profile}}` +substitutes the assigned value into the document at that spot, so +writing assignments in an op body leaves stray text in the op. Put them +in `description:` (where they read as documentation) and reference them +with `{{@p}}` where the value belongs — the workload example below does +exactly this. + For a complete workload where every dataset-shaped parameter is derived this way, see `nb-adapters/adapter-cqld4/src/main/resources/activities/baselinesv2/cql_vector_dataset.yaml`: diff --git a/nb-adapters/adapter-cqld4/src/main/resources/activities/baselinesv2/cql_vector_dataset.yaml b/nb-adapters/adapter-cqld4/src/main/resources/activities/baselinesv2/cql_vector_dataset.yaml index 3293a139a..017df5db8 100644 --- a/nb-adapters/adapter-cqld4/src/main/resources/activities/baselinesv2/cql_vector_dataset.yaml +++ b/nb-adapters/adapter-cqld4/src/main/resources/activities/baselinesv2/cql_vector_dataset.yaml @@ -14,9 +14,14 @@ description: | maxk: {{maxk == neighborIndices(spec).dimension()}} similarity: {{similarity == dataset(spec).attributes().getOrDefault('distance_function','COSINE').toString().toLowerCase()}} base bytes to warm (profile window): {{warmbytes == prefetchPlan(spec, 'base_vectors', dataset(spec).facet('base_vectors').get().window() ?: '').bytesToFetch()}} - The rampup bindings warm the profile's base window eagerly before the - first cycle (with download progress on stderr); pass a window and - prefetch mode on the binding functions to shape that differently. + The bindings warm what they read, when they resolve — so a block that + does not use a facet never fetches it — with download progress on + stderr. Pass a window and prefetch mode to shape that: + BaseVectors('{{@spec}}','[0..1M)','background') warms a cycle range + while the run proceeds. To warm everything up front instead, run + prefetchProfile('{{@spec}}') from an expr, remembering that + expressions evaluate when the workload loads rather than when a block + runs. Operational knobs stay conventional: keyspace, table, rf, and cl are TEMPLATE parameters with the usual defaults. diff --git a/nb-virtdata/virtdata-lib-vectors/src/main/java/io/nosqlbench/exprs/lib/vectors/VectorDataExprs.java b/nb-virtdata/virtdata-lib-vectors/src/main/java/io/nosqlbench/exprs/lib/vectors/VectorDataExprs.java index 65cfce2e8..46dea909d 100644 --- a/nb-virtdata/virtdata-lib-vectors/src/main/java/io/nosqlbench/exprs/lib/vectors/VectorDataExprs.java +++ b/nb-virtdata/virtdata-lib-vectors/src/main/java/io/nosqlbench/exprs/lib/vectors/VectorDataExprs.java @@ -25,6 +25,7 @@ import io.nosqlbench.nb.api.expr.annotations.ExprFunctionSpec; import io.nosqlbench.vectordata.Catalog; import io.nosqlbench.vectordata.CatalogSources; import io.nosqlbench.vectordata.DSWindow; +import io.nosqlbench.vectordata.FacetDescriptor; import io.nosqlbench.vectordata.PrefetchHandle; import io.nosqlbench.vectordata.PrefetchPlan; import io.nosqlbench.vectordata.PrefetchReport; @@ -32,6 +33,7 @@ import io.nosqlbench.vectordata.TestDataView; import io.nosqlbench.vectordata.VectorReader; import io.nosqlbench.vectordata.VvecReader; import io.nosqlbench.vectordata.WholeFacetFallback; +import io.nosqlbench.virtdata.lib.vectors.vectordata.PrefetchMeter; import io.nosqlbench.virtdata.lib.vectors.vectordata.WindowedReader; import org.apache.logging.log4j.LogManager; import org.apache.logging.log4j.Logger; @@ -160,7 +162,52 @@ public class VectorDataExprs implements ExprFunctionProvider { description = "Fetch a record window of a facet and return when it is resident. An empty window requests the whole facet; an unresolvable window is refused rather than fetching everything." ) public PrefetchReport prefetch(String datasetNameAndProfile, String facetName, String window) { - return dataset(datasetNameAndProfile).prefetch(facetName, DSWindow.parse(window), WholeFacetFallback.REFUSE); + return warm(dataset(datasetNameAndProfile), datasetNameAndProfile, facetName, DSWindow.parse(window)); + } + + @ExprExample(args = {"\"airports:demo\"", "\"base_vectors\"", "0", "1000"}, expectNotNull = true) + @ExprExample(args = {"\"airports:demo\"", "\"query_vectors\"", "0", "10"}, expectNotNull = true) + @ExprFunctionSpec( + name = "prefetchCycles", + synopsis = "prefetchCycles(\"dataset:profile\", \"facet_name\", start, end)", + description = "Fetch the records a cycle range will read — the half-open ordinal interval [start,end) — and return when they are resident. The natural form when the range comes from an activity's cycles." + ) + public PrefetchReport prefetchCycles(String datasetNameAndProfile, String facetName, Number start, Number end) { + DSWindow window = new DSWindow(java.util.List.of(new DSWindow.Interval(start.longValue(), end.longValue()))); + return warm(dataset(datasetNameAndProfile), datasetNameAndProfile, facetName, window); + } + + @ExprExample(args = {"\"airports:demo\""}, expectNotNull = true) + @ExprExample(args = {"\"airports:demo\""}, matches = ".+") + @ExprFunctionSpec( + name = "prefetchProfile", + synopsis = "prefetchProfile(\"dataset:profile\")", + description = "Fetch every facet the profile declares, each to the window the profile declares for it (or whole when it declares none), and return a per-facet summary. The one-call warm-up for a workload." + ) + public String prefetchProfile(String datasetNameAndProfile) { + TestDataView view = dataset(datasetNameAndProfile); + StringBuilder summary = new StringBuilder(); + for (String facetName : view.facets().keySet()) { + String declared = view.facet(facetName).map(FacetDescriptor::window).orElse(null); + DSWindow window = DSWindow.parse(declared == null ? "" : declared); + PrefetchPlan plan = view.prefetchPlan(facetName, window); + warm(view, datasetNameAndProfile, facetName, window); + if (!summary.isEmpty()) summary.append("; "); + summary.append(facetName).append(' ').append(declared == null ? "ALL" : declared).append(' ') + .append(plan.isResident() ? "resident" : PrefetchMeter.bytes(plan.bytesToFetch())); + } + return summary.toString(); + } + + /// Plans, announces, fetches with progress, and confirms — the one + /// path every expr-driven warm-up takes, so a download started from + /// a workload reports itself the same way one started from a + /// binding does. + private static PrefetchReport warm(TestDataView view, String spec, String facetName, DSWindow window) { + PrefetchMeter meter = new PrefetchMeter(spec + ":" + facetName, view.prefetchPlan(facetName, window)); + PrefetchReport report = view.prefetch(facetName, window, WholeFacetFallback.REFUSE, meter); + meter.complete(); + return report; } @ExprExample(args = {"\"airports:demo\"", "\"base_vectors\"", "\"[0..100)\""}, expectNotNull = true) diff --git a/nb-virtdata/virtdata-lib-vectors/src/main/java/io/nosqlbench/virtdata/lib/vectors/vectordata/PrefetchMeter.java b/nb-virtdata/virtdata-lib-vectors/src/main/java/io/nosqlbench/virtdata/lib/vectors/vectordata/PrefetchMeter.java index e27debce5..ee30f3292 100644 --- a/nb-virtdata/virtdata-lib-vectors/src/main/java/io/nosqlbench/virtdata/lib/vectors/vectordata/PrefetchMeter.java +++ b/nb-virtdata/virtdata-lib-vectors/src/main/java/io/nosqlbench/virtdata/lib/vectors/vectordata/PrefetchMeter.java @@ -22,20 +22,21 @@ import io.nosqlbench.vectordata.PrebufferProgress; import io.nosqlbench.vectordata.PrefetchHandle; import io.nosqlbench.vectordata.PrefetchPlan; -/// Console meter for a binding-level prefetch: the plan's cost is -/// announced before any bytes move, progress is emitted during the -/// download (rate-limited), and completion is confirmed. Lines go to +/// Console meter for a prefetch driven from a binding or an expr: the +/// plan's cost is announced before any bytes move, progress is emitted +/// during the download (rate-limited), and completion is confirmed. +/// Lines go to /// stderr so they are visible regardless of logging configuration — /// this is a download meter, not a log event. Silent when the plan has /// nothing to fetch, so local and already-warm facets add no noise. -final class PrefetchMeter implements PrebufferProgress { +public final class PrefetchMeter implements PrebufferProgress { private static final long INTERVAL_NANOS = 2_000_000_000L; private final String label; private final long planned; private volatile long lastEmit; - PrefetchMeter(String label, PrefetchPlan plan) { + public PrefetchMeter(String label, PrefetchPlan plan) { this.label = label; this.planned = plan.bytesToFetch(); if (planned > 0) @@ -52,13 +53,13 @@ final class PrefetchMeter implements PrebufferProgress { } /// Confirms a completed blocking fetch. - void complete() { + public void complete() { if (planned > 0) System.err.printf("[vectordata] %s: fetch complete (%s)%n", label, bytes(planned)); } /// Follows a background prefetch on a daemon thread, emitting the /// handle's byte counter until the worker finishes or is cancelled. - void watch(PrefetchHandle handle) { + public void watch(PrefetchHandle handle) { if (planned == 0) return; Thread watcher = new Thread(() -> { while (!handle.isDone()) { @@ -72,7 +73,7 @@ final class PrefetchMeter implements PrebufferProgress { watcher.start(); } - static String bytes(long count) { + public static String bytes(long count) { if (count >= 1L << 30) return String.format("%.1f GiB", count / (double) (1L << 30)); if (count >= 1L << 20) return String.format("%.1f MiB", count / (double) (1L << 20)); if (count >= 1L << 10) return String.format("%.1f KiB", count / (double) (1L << 10));