mirror of
https://github.com/nosqlbench/nosqlbench.git
synced 2026-10-12 09:24:52 -05:00
feat(virtdata): prefetch progress, profile warm-up, and cycle ranges from exprs
A prefetch driven from a workload was silent and awkward to aim: - every expr prefetch now reports through the same meter the bindings use, so a download started from a workload announces its cost, emits progress while the bytes move, and confirms completion - prefetchProfile(spec) warms every facet the profile declares, each to the window it declares, in one call — the form that cannot accidentally pull a whole shared base file, since an empty window means the entire facet by design - prefetchCycles(spec, facet, start, end) takes the ordinal range directly, which is how an activity already describes its work The docs record the three things that bite: expressions evaluate at load time rather than when a tagged block runs, an empty window is a request for everything, and assignment sigils render their value where they are written.
This commit is contained in:
1 parent
20eb4f3c89
commit
9085bf06ad
4 files changed
+99
-15
No files matched your search
@@ -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`:
|
||||
|
||||
+8
-3
@@ -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.
|
||||
|
||||
|
||||
+48
-1
@@ -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)
|
||||
|
||||
+9
-8
@@ -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));
|
||||
|
||||
Reference in new issue
Block a user