mirror of
https://github.com/nosqlbench/nosqlbench.git
synced 2026-10-12 09:24:52 -05:00
Merge branch 'main' into dependabot/github_actions/actions/setup-java-6.0.0
This commit is contained in:
14 files changed
+392
-74
No files matched your search
@@ -153,7 +153,7 @@ jobs:
|
||||
|
||||
# https://github.com/softprops/action-gh-release
|
||||
- name: create github release
|
||||
uses: softprops/action-gh-release@v3.0.2
|
||||
uses: softprops/action-gh-release@v3.0.3
|
||||
if: startsWith(github.ref, 'refs/tags/')
|
||||
with:
|
||||
# body: ${{ steps.prepare_summary.outputs.release_summary }}
|
||||
|
||||
@@ -139,7 +139,7 @@ jobs:
|
||||
|
||||
# https://github.com/softprops/action-gh-release
|
||||
- name: create github release
|
||||
uses: softprops/action-gh-release@v3.0.2
|
||||
uses: softprops/action-gh-release@v3.0.3
|
||||
if: startsWith(github.ref, 'refs/tags/')
|
||||
with:
|
||||
# body: ${{ steps.prepare_summary.outputs.release_summary }}
|
||||
|
||||
@@ -26,6 +26,56 @@ e.g. `NB='java -jar nb5.jar'`). `VECTORDATA_CATALOG` outranks any
|
||||
`VECTORDATA_HOME` under `/tmp/vdtest` so it never touches
|
||||
`~/.config/vectordata` or your real cache.
|
||||
|
||||
## Watch a download happen
|
||||
|
||||
`vectordata_demo` is a workload bundled in nb5 — no checkout, no
|
||||
database. It uses the stdout driver, so the only thing it exercises is
|
||||
dataset access: it prints what it derived from the dataset, fetches a
|
||||
record range with a live meter, and reads records back.
|
||||
|
||||
```bash
|
||||
export VECTORDATA_CATALOG='https://your.host/path/catalog.yaml'
|
||||
export VECTORDATA_HOME=/tmp/vdtest # empty dir ⇒ nothing is cached yet
|
||||
nb5 vectordata_demo dataset=mydataset:myprofile
|
||||
```
|
||||
|
||||
That is the scenario form — the workload name stands alone, with no
|
||||
`run` in front of it (`run` is itself a command, so `nb5 run
|
||||
vectordata_demo` fails). The equivalent long form names the full path:
|
||||
|
||||
```bash
|
||||
nb5 run workload=activities/examples/vectordata_demo.yaml dataset=mydataset:myprofile
|
||||
```
|
||||
|
||||
`nb5 --cat vectordata_demo` prints it and `nb5 --copy vectordata_demo`
|
||||
drops an editable copy in the working directory.
|
||||
|
||||
The meter goes to stderr while the bytes move:
|
||||
|
||||
```
|
||||
[vectordata] mydataset:myprofile:base_vectors: fetching 29.0 MiB in 1 range(s)
|
||||
[vectordata] mydataset:myprofile:base_vectors: 12.0 MiB / 30.0 MiB
|
||||
[vectordata] mydataset:myprofile:base_vectors: fetch complete (29.0 MiB)
|
||||
```
|
||||
|
||||
To watch the fetch overlap the run instead of preceding it, ask for the
|
||||
background form — give it a range big enough and enough cycles to
|
||||
outlast the download:
|
||||
|
||||
```bash
|
||||
nb5 vectordata_demo dataset=mydataset:myprofile \
|
||||
prefetch=background records=100000 cycles=200
|
||||
```
|
||||
|
||||
`prefetch` selects which prefetch call does the work: `load` (the
|
||||
default) blocks on `prefetchCycles` before cycle 0, `background` hands
|
||||
back a handle from `prefetchCyclesBackground` and meters it while the
|
||||
run proceeds, and `none` leaves reads to demand-page. `records` sets how
|
||||
many ordinals the demo spans, `cycles` how long it runs. The meter is
|
||||
silent when there is nothing to fetch, so a second run against a warm
|
||||
cache prints nothing — point `VECTORDATA_HOME` at an empty directory to
|
||||
see it fetch again.
|
||||
|
||||
Beyond the script's checks:
|
||||
|
||||
```bash
|
||||
@@ -142,14 +192,22 @@ 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()}}
|
||||
|
||||
# Background warm-up captured in an expr variable, joined later:
|
||||
# The same, in the background: the handle comes back immediately and
|
||||
# the fetch overlaps whatever runs next. Both background forms meter
|
||||
# from the handle, so the download reports itself either way.
|
||||
# {{= prefetchCyclesBackground("example:default", "base_vectors", 0, 1000000).plan().requests()}}
|
||||
# {{warmup = prefetchBackground("example:default", "base_vectors", "[0..1M)")}}
|
||||
# {{= warmup.join().rangesFetched()}}
|
||||
|
||||
@@ -163,6 +221,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`:
|
||||
@@ -212,10 +296,17 @@ java -jar nb5.jar run driver=stdout cycles=10 \
|
||||
```
|
||||
|
||||
Prefetch modes on every mapper: `eager` (default; aliases `prebuffer`,
|
||||
`true`), `background`, `none` (aliases `demand`, `false`). A binding
|
||||
window is in the reader's own coordinates — after any profile window —
|
||||
and is shifted to absolute records for the fetch, so the warmed bytes
|
||||
are exactly the bytes the clipped reader exposes.
|
||||
`true`), `background`, `none` (aliases `demand`, `false`).
|
||||
|
||||
A binding window says **which records to warm**, and nothing else —
|
||||
indices stay absolute record ordinals.
|
||||
`BaseVectors('ds:profile','[50000..100000)')` warms records 50000
|
||||
through 99999, and a run of `cycles=50000..100000` addresses exactly
|
||||
those records. Reading outside the window is not an error; it just
|
||||
demand-pages, as it would with no window. Ordinals mean the same thing
|
||||
here, in `VariableFacet`, and in `prefetchCycles`. The window is
|
||||
translated into dataset coordinates for the fetch, so the warmed bytes
|
||||
are the bytes the reader will expose.
|
||||
|
||||
## Verifying that warming happened
|
||||
|
||||
|
||||
@@ -25,7 +25,7 @@
|
||||
<packaging>pom</packaging>
|
||||
|
||||
<properties>
|
||||
<revision>5.25.14-SNAPSHOT</revision>
|
||||
<revision>5.25.15-SNAPSHOT</revision>
|
||||
<!-- Set this level to override the logging level for tests during build -->
|
||||
<project.testlevel>INFO</project.testlevel>
|
||||
<!-- Set this level to override the logging level for tests logging configuration during build -->
|
||||
|
||||
+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.
|
||||
|
||||
|
||||
+77
@@ -0,0 +1,77 @@
|
||||
min_version: "5.25"
|
||||
|
||||
description: |
|
||||
Remote vector dataset access with a live download meter. No database
|
||||
required — this uses the stdout driver, so the only thing exercised is
|
||||
dataset access itself.
|
||||
|
||||
nb5 vectordata_demo dataset=<name>:<profile>
|
||||
|
||||
Point it at a catalog first, either of these:
|
||||
|
||||
export VECTORDATA_CATALOG='https://your.host/path/catalog.yaml'
|
||||
nb5 vectordata_demo dataset=<name>:<profile> -Dvectordata.catalog=<url>
|
||||
|
||||
Everything the workload needs is read from the dataset itself when the
|
||||
file loads — dimensions, record counts, k, and the similarity function
|
||||
come from the data rather than from parameters you have to restate.
|
||||
|
||||
dataset: {{spec == paramOr('dataset','MISSING-dataset-parameter')}}
|
||||
records: {{records == paramOr('records','20000') as Long}}
|
||||
prefetch: {{prefetchMode == paramOr('prefetch','load')}}
|
||||
{{= spec == 'MISSING-dataset-parameter' ? ' (pass dataset=<name>:<profile> to derive the rest from a dataset)' :
|
||||
' dimensions: ' + baseVectors(spec).dimension() + '\n' +
|
||||
' base records: ' + baseVectors(spec).count() + '\n' +
|
||||
' query records: ' + queryVectors(spec).count() + '\n' +
|
||||
' neighbors k: ' + neighborIndices(spec).dimension() + '\n' +
|
||||
' similarity: ' + dataset(spec).attributes().getOrDefault('distance_function','unspecified')}}
|
||||
|
||||
The demo reads the first `records` ordinals of the base facet, and
|
||||
`prefetch` chooses which prefetch call fetches them. Both fetching
|
||||
forms report the same way — the plan is announced, progress is metered
|
||||
while the bytes move, and completion is confirmed, all on stderr:
|
||||
|
||||
prefetch=load prefetchCycles(...) — fetches before the run
|
||||
starts, so cycle 0 reads warm data. Blocks
|
||||
until resident. This is the default.
|
||||
prefetch=background prefetchCyclesBackground(...) — returns a
|
||||
handle immediately and fetches while the cycles
|
||||
run, so reads and the download overlap.
|
||||
Progress is polled from the handle. Give it
|
||||
enough cycles to outlast the download, or the
|
||||
run ends first and the fetch stops partway.
|
||||
prefetch=none no warm-up; reads demand-page whatever they
|
||||
touch. Nothing to meter, so it stays quiet.
|
||||
|
||||
{{= spec == 'MISSING-dataset-parameter' ? ' warm-up: skipped (no dataset given)' :
|
||||
prefetchMode == 'load' ? ' warm-up: fetched ' + prefetchCycles(spec,'base_vectors',0,records).rangesFetched() + ' range(s) before the run' :
|
||||
prefetchMode == 'background' ? ' warm-up: started in the background, ' + prefetchCyclesBackground(spec,'base_vectors',0,records).plan().requests() + ' range(s) planned' :
|
||||
' warm-up: none, reads demand-page'}}
|
||||
|
||||
Two ways to watch it:
|
||||
|
||||
# fetch up front, then read — the meter runs before any cycles
|
||||
nb5 vectordata_demo dataset=<name>:<profile>
|
||||
|
||||
# fetch during the run — a large range and enough cycles to see it
|
||||
nb5 vectordata_demo dataset=<name>:<profile> \
|
||||
prefetch=background records=100000 cycles=200
|
||||
|
||||
Parameters: dataset (required), cycles (default 3), records (default
|
||||
20000), prefetch (load|background|none, default load). The meter only
|
||||
speaks when there are bytes to move, so a second run against a warm
|
||||
cache is silent by design — point VECTORDATA_HOME at an empty
|
||||
directory to watch it fetch again.
|
||||
|
||||
scenarios:
|
||||
default: >-
|
||||
run driver=stdout format=readout cycles=TEMPLATE(cycles,3) threads=1
|
||||
|
||||
bindings:
|
||||
cycle: Identity()
|
||||
# Reads are demand-paged here on purpose: the prefetch above is what
|
||||
# warms the data, so the meter has one owner and this window only
|
||||
# clips what the bindings expose.
|
||||
base_vector: BaseVectors('{{@spec}}','[0..{{@records}})','none'); Stringify()
|
||||
query_vector: QueryVectors('{{@spec}}',false); Stringify()
|
||||
neighbor_indices: NeighborIndices('{{@spec}}',false); Stringify()
|
||||
@@ -76,7 +76,10 @@ def currentDate() {
|
||||
)
|
||||
@ExprExample(expectNotNull = true)
|
||||
def currentTime() {
|
||||
LocalTime.now().truncatedTo(ChronoUnit.SECONDS).toString()
|
||||
// Explicit pattern, not toString(): LocalTime's ISO form omits the
|
||||
// seconds field entirely when the second is 00, so toString() emits
|
||||
// HH:mm once a minute and HH:mm:ss the rest of the time.
|
||||
LocalTime.now().format(DateTimeFormatter.ofPattern("HH:mm:ss"))
|
||||
}
|
||||
|
||||
@ExprFunctionSpec(
|
||||
@@ -85,7 +88,8 @@ def currentTime() {
|
||||
)
|
||||
@ExprExample(expectNotNull = true)
|
||||
def currentDateTime() {
|
||||
LocalDateTime.now().truncatedTo(ChronoUnit.SECONDS).toString()
|
||||
// See currentTime(): the ISO form drops seconds at second 00.
|
||||
LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyy-MM-dd'T'HH:mm:ss"))
|
||||
}
|
||||
|
||||
@ExprFunctionSpec(
|
||||
@@ -266,11 +270,12 @@ def millisToDate(millis) {
|
||||
)
|
||||
@ExprExample(expectNotNull = true)
|
||||
def millisToDateTime(millis) {
|
||||
// See currentTime(): the ISO form drops seconds at second 00, so a
|
||||
// timestamp landing on a whole minute rendered as yyyy-MM-ddTHH:mm.
|
||||
Instant.ofEpochMilli(millis as long)
|
||||
.atZone(ZoneId.systemDefault())
|
||||
.toLocalDateTime()
|
||||
.truncatedTo(ChronoUnit.SECONDS)
|
||||
.toString()
|
||||
.format(DateTimeFormatter.ofPattern("yyyy-MM-dd'T'HH:mm:ss"))
|
||||
}
|
||||
|
||||
@ExprFunctionSpec(
|
||||
@@ -300,5 +305,6 @@ def currentDateInZone(timezone) {
|
||||
)
|
||||
@ExprExample(expectNotNull = true)
|
||||
def currentTimeInZone(timezone) {
|
||||
LocalTime.now(ZoneId.of(timezone)).truncatedTo(ChronoUnit.SECONDS).toString()
|
||||
// See currentTime(): the ISO form drops seconds at second 00.
|
||||
LocalTime.now(ZoneId.of(timezone)).format(DateTimeFormatter.ofPattern("HH:mm:ss"))
|
||||
}
|
||||
@@ -19,6 +19,10 @@ package io.nosqlbench.nb.api.expr;
|
||||
|
||||
import org.junit.jupiter.api.BeforeEach;
|
||||
import org.junit.jupiter.api.Test;
|
||||
|
||||
import java.time.ZoneId;
|
||||
import java.time.ZonedDateTime;
|
||||
import java.time.temporal.ChronoUnit;
|
||||
import org.junit.jupiter.api.Tag;
|
||||
|
||||
import java.util.Map;
|
||||
@@ -122,6 +126,28 @@ class BundledLibrariesTest {
|
||||
assertTrue(result.matches("\\d{2}:\\d{2}:\\d{2}"));
|
||||
}
|
||||
|
||||
/// The seconds field is optional in the ISO forms of `LocalTime` and
|
||||
/// `LocalDateTime`: `toString()` omits it entirely when the second
|
||||
/// is `00`, so a shape assertion against the wall clock only catches
|
||||
/// the defect one second in sixty. `millisToDateTime` takes its
|
||||
/// instant from the caller and shares the same rendering, so the
|
||||
/// boundary can be pinned exactly rather than waited for.
|
||||
@Test
|
||||
void timesOnAWholeMinuteStillRenderTheirSeconds() {
|
||||
long onTheMinute = ZonedDateTime.now(ZoneId.systemDefault())
|
||||
.truncatedTo(ChronoUnit.MINUTES).toInstant().toEpochMilli();
|
||||
String atSecondZero = processor.process("{{= millisToDateTime(" + onTheMinute + ") }}", null, Map.of());
|
||||
assertTrue(atSecondZero.matches("\\d{4}-\\d{2}-\\d{2}T\\d{2}:\\d{2}:\\d{2}"),
|
||||
"a timestamp on a whole minute must still render seconds, got: " + atSecondZero);
|
||||
assertTrue(atSecondZero.endsWith(":00"), "and those seconds are 00, got: " + atSecondZero);
|
||||
|
||||
String atSecondThirty = processor.process(
|
||||
"{{= millisToDateTime(" + (onTheMinute + 30_000) + ") }}", null, Map.of());
|
||||
assertTrue(atSecondThirty.matches("\\d{4}-\\d{2}-\\d{2}T\\d{2}:\\d{2}:\\d{2}"),
|
||||
"and a non-zero second renders the same width, got: " + atSecondThirty);
|
||||
assertTrue(atSecondThirty.endsWith(":30"), atSecondThirty);
|
||||
}
|
||||
|
||||
@Test
|
||||
void testCurrentDateTime() {
|
||||
String result = processor.process("{{= currentDateTime() }}", null, Map.of());
|
||||
|
||||
+71
-3
@@ -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;
|
||||
@@ -124,7 +126,7 @@ public class VectorDataExprs implements ExprFunctionProvider {
|
||||
@ExprFunctionSpec(
|
||||
name = "windowedFacet",
|
||||
synopsis = "windowedFacet(\"dataset:profile\", \"facet_name\", \"window\")",
|
||||
description = "Return a reader for a named facet, clipped to a record window in the reader's own coordinates."
|
||||
description = "Return a reader for a named facet, restricted to a record window. Indices stay absolute — a window of [50000..100000) accepts 50000..99999 — so cycle values address the records they name."
|
||||
)
|
||||
public VectorReader<?> windowedFacet(String datasetNameAndProfile, String facetName, String window) {
|
||||
return WindowedReader.clip(dataset(datasetNameAndProfile).openFacet(facetName), window);
|
||||
@@ -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)
|
||||
@@ -171,7 +218,28 @@ public class VectorDataExprs implements ExprFunctionProvider {
|
||||
description = "Start fetching a record window of a facet on another thread and return the handle; join() waits, cancel() stops between ranges."
|
||||
)
|
||||
public PrefetchHandle prefetchBackground(String datasetNameAndProfile, String facetName, String window) {
|
||||
return dataset(datasetNameAndProfile).prefetchInBackground(facetName, DSWindow.parse(window), WholeFacetFallback.REFUSE);
|
||||
TestDataView view = dataset(datasetNameAndProfile);
|
||||
DSWindow parsed = DSWindow.parse(window);
|
||||
// Metered like every other prefetch form: the plan is announced
|
||||
// before the handle is returned, and a watcher reports the
|
||||
// handle's byte counter while the fetch runs alongside whatever
|
||||
// the caller does next.
|
||||
PrefetchMeter meter = new PrefetchMeter(datasetNameAndProfile + ":" + facetName, view.prefetchPlan(facetName, parsed));
|
||||
PrefetchHandle handle = view.prefetchInBackground(facetName, parsed, WholeFacetFallback.REFUSE);
|
||||
meter.watch(handle);
|
||||
return handle;
|
||||
}
|
||||
|
||||
@ExprExample(args = {"\"airports:demo\"", "\"base_vectors\"", "0", "1000"}, expectNotNull = true)
|
||||
@ExprExample(args = {"\"airports:demo\"", "\"query_vectors\"", "0", "10"}, expectNotNull = true)
|
||||
@ExprFunctionSpec(
|
||||
name = "prefetchCyclesBackground",
|
||||
synopsis = "prefetchCyclesBackground(\"dataset:profile\", \"facet_name\", start, end)",
|
||||
description = "Start fetching the records a cycle range will read, on another thread, and return the handle. The background counterpart of prefetchCycles, metered the same way."
|
||||
)
|
||||
public PrefetchHandle prefetchCyclesBackground(String datasetNameAndProfile, String facetName, Number start, Number end) {
|
||||
return prefetchBackground(datasetNameAndProfile, facetName,
|
||||
"[" + start.longValue() + ".." + end.longValue() + ")");
|
||||
}
|
||||
|
||||
}
|
||||
+22
-13
@@ -35,9 +35,8 @@ import java.util.Locale;
|
||||
import java.util.function.LongFunction;
|
||||
|
||||
/// Base plumbing for vectordata-backed binding functions: opens the
|
||||
/// named dataset profile from the default catalogs, clips the reader to
|
||||
/// an optional record window, and warms data according to a prefetch
|
||||
/// mode:
|
||||
/// named dataset profile from the default catalogs and warms an
|
||||
/// optional record window according to a prefetch mode:
|
||||
///
|
||||
/// - `eager` (aliases `prebuffer`, `true`) — fetch the effective window,
|
||||
/// or the whole facet when no window applies, before returning;
|
||||
@@ -45,11 +44,18 @@ import java.util.function.LongFunction;
|
||||
/// immediately, letting reads overlap the download;
|
||||
/// - `none` (aliases `demand`, `false`) — demand-paged access only.
|
||||
///
|
||||
/// A binding window is expressed in the reader's own index coordinates,
|
||||
/// after any profile-declared window. For warming, it is translated to
|
||||
/// absolute record coordinates by shifting against the profile window,
|
||||
/// so the bytes fetched are exactly the bytes the clipped reader
|
||||
/// exposes.
|
||||
/// A binding window selects **which records to warm**, and nothing
|
||||
/// else: indices stay absolute record ordinals, so a workload running
|
||||
/// `cycles=50000..100000` with a window of `[50000..100000)` addresses
|
||||
/// the records those cycles name. Reads outside the window are not an
|
||||
/// error — they simply demand-page, the way they would with no window
|
||||
/// at all — which also keeps the binding resolvable, since VirtData
|
||||
/// probes a newly resolved function with a small sample index before
|
||||
/// any cycle runs. Ordinals mean the same thing here, in
|
||||
/// [VariableFacet], and in the `prefetchCycles` expression function.
|
||||
/// The window is translated into dataset coordinates against any
|
||||
/// profile-declared window, so the bytes fetched are the bytes the
|
||||
/// reader will expose.
|
||||
public abstract class CoreVectors<T> implements LongFunction<T> {
|
||||
|
||||
protected final TestDataView tdv;
|
||||
@@ -66,7 +72,8 @@ public abstract class CoreVectors<T> implements LongFunction<T> {
|
||||
Catalog catalog = Catalog.of(CatalogSources.defaults(), settings);
|
||||
tdv = catalog.openProfile(datasetAndProfile);
|
||||
this.facetName = FacetNames.canonical(facetName);
|
||||
dataset = WindowedReader.clip(getRandomAccessData(), window);
|
||||
// Deliberately unwrapped: the window warms, it does not clip.
|
||||
dataset = getRandomAccessData();
|
||||
backgroundPrefetch = warm(window, Prefetch.parse(prefetchMode));
|
||||
}
|
||||
|
||||
@@ -83,7 +90,7 @@ public abstract class CoreVectors<T> implements LongFunction<T> {
|
||||
// The plan is announced before any bytes move and progress is
|
||||
// emitted during the download; the meter is silent when there
|
||||
// is nothing to fetch.
|
||||
PrefetchMeter meter = new PrefetchMeter(tdv.dataset() + ":" + facetName, tdv.prefetchPlan(facetName, effective));
|
||||
PrefetchMeter meter = new PrefetchMeter(tdv.dataset() + ":" + tdv.profile() + ":" + facetName, tdv.prefetchPlan(facetName, effective));
|
||||
if (mode == Prefetch.BACKGROUND) {
|
||||
PrefetchHandle handle = tdv.prefetchInBackground(facetName, effective, WholeFacetFallback.REFUSE);
|
||||
meter.watch(handle);
|
||||
@@ -94,9 +101,11 @@ public abstract class CoreVectors<T> implements LongFunction<T> {
|
||||
return null;
|
||||
}
|
||||
|
||||
/// Translates a reader-relative binding window to absolute record
|
||||
/// coordinates: shifted by the profile window's start and clamped
|
||||
/// to its end when the facet declares one. Returns the profile
|
||||
/// Translates a binding window into dataset record coordinates for
|
||||
/// the fetch. A profile-declared window re-bases the reader it
|
||||
/// hands back, so a binding index is offset by that window's start
|
||||
/// and clamped to its end; profiles in practice declare windows
|
||||
/// starting at 0, where this is the identity. Returns the profile
|
||||
/// window itself when no binding window is given, [DSWindow#ALL]
|
||||
/// when neither applies, and `null` when the binding window falls
|
||||
/// entirely outside the profile window — nothing to warm.
|
||||
|
||||
+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));
|
||||
|
||||
+1
-1
@@ -78,7 +78,7 @@ public class VariableFacet implements LongFunction<Object> {
|
||||
DSWindow parsed = DSWindow.parse(window == null ? "" : window);
|
||||
CoreVectors.Prefetch mode = CoreVectors.Prefetch.parse(prefetchMode);
|
||||
PrefetchMeter meter = mode == CoreVectors.Prefetch.NONE ? null
|
||||
: new PrefetchMeter(tdv.dataset() + ":" + this.facetName, tdv.prefetchPlan(this.facetName, parsed));
|
||||
: new PrefetchMeter(tdv.dataset() + ":" + tdv.profile() + ":" + this.facetName, tdv.prefetchPlan(this.facetName, parsed));
|
||||
backgroundPrefetch = switch (mode) {
|
||||
case NONE -> null;
|
||||
case BACKGROUND -> {
|
||||
|
||||
+35
-21
@@ -24,41 +24,55 @@ import io.nosqlbench.vectordata.ElementType;
|
||||
import io.nosqlbench.vectordata.PrebufferProgress;
|
||||
import io.nosqlbench.vectordata.VectorReader;
|
||||
|
||||
/// Binding-layer clip of a [VectorReader] to a record window, with the
|
||||
/// same semantics a profile-declared window gives a reader in
|
||||
/// nb-vectordata (and in `vectordata-rs`): only the first interval
|
||||
/// applies, and both bounds clamp to the underlying reader's count.
|
||||
/// This adapter lives here rather than in nb-vectordata because the
|
||||
/// vectordata crate exposes no caller-side reader windowing — windows
|
||||
/// there come from facet configuration — and the binding functions'
|
||||
/// window argument is a binding-layer convenience.
|
||||
/// Restricts a [VectorReader] to a record range **without renumbering
|
||||
/// it**: a window of `[50000..100000)` means indices 50000 through
|
||||
/// 99999 are readable and everything else is refused. Indices stay
|
||||
/// absolute, so the same ordinals mean the same records here as
|
||||
/// everywhere else in this package.
|
||||
///
|
||||
/// Used by the `windowedFacet` expression function, where a caller has
|
||||
/// explicitly asked for a restricted reader. The binding functions do
|
||||
/// **not** wrap their readers in this: a binding window only warms, and
|
||||
/// restricting it would break resolution, since VirtData probes a newly
|
||||
/// resolved function with a small sample index before any cycle runs.
|
||||
///
|
||||
/// Only the first interval of the window applies, and both bounds clamp
|
||||
/// to the underlying reader's count.
|
||||
public final class WindowedReader<A> implements VectorReader<A> {
|
||||
|
||||
private final VectorReader<A> inner;
|
||||
private final long begin, count;
|
||||
private final long begin, end;
|
||||
|
||||
/// Clips `reader` to the first interval of `window`, in the
|
||||
/// reader's own index coordinates. A blank window returns the
|
||||
/// reader unchanged.
|
||||
/// Restricts `reader` to the first interval of `window`, in the
|
||||
/// reader's own — absolute — index coordinates. A blank window
|
||||
/// returns the reader unchanged.
|
||||
public static <A> VectorReader<A> clip(VectorReader<A> reader, String window) {
|
||||
DSWindow parsed = DSWindow.parse(window == null ? "" : window);
|
||||
if (parsed.isEmpty()) return reader;
|
||||
DSWindow.Interval first = parsed.intervals().get(0);
|
||||
long total = reader.count();
|
||||
long begin = Math.min(first.minIncl(), total);
|
||||
long count = Math.max(0, Math.min(first.maxExcl(), total) - begin);
|
||||
return new WindowedReader<>(reader, begin, count);
|
||||
return new WindowedReader<>(reader, Math.min(first.minIncl(), total), Math.min(first.maxExcl(), total));
|
||||
}
|
||||
|
||||
private WindowedReader(VectorReader<A> inner, long begin, long count) {
|
||||
this.inner = inner; this.begin = begin; this.count = count;
|
||||
private WindowedReader(VectorReader<A> inner, long begin, long end) {
|
||||
this.inner = inner; this.begin = begin; this.end = end;
|
||||
}
|
||||
|
||||
@Override public long count() { return count; }
|
||||
/// The exclusive upper bound of the readable range — the highest
|
||||
/// index this reader accepts, plus one. Indices below the window's
|
||||
/// start are refused even though they are below this count, because
|
||||
/// the window is a restriction rather than a re-basing.
|
||||
@Override public long count() { return end; }
|
||||
/// First index this reader accepts.
|
||||
public long begin() { return begin; }
|
||||
@Override public int dimension() { return inner.dimension(); }
|
||||
@Override public A get(long index) { check(index); return inner.get(begin + index); }
|
||||
@Override public void get(long index, A target) { check(index); inner.get(begin + index, target); }
|
||||
private void check(long index) { if (index < 0 || index >= count) throw new IndexOutOfBoundsException("windowed index " + index + " outside 0.." + (count - 1)); }
|
||||
@Override public A get(long index) { check(index); return inner.get(index); }
|
||||
@Override public void get(long index, A target) { check(index); inner.get(index, target); }
|
||||
private void check(long index) {
|
||||
if (index < begin || index >= end)
|
||||
throw new IndexOutOfBoundsException("index " + index + " is outside the binding window ["
|
||||
+ begin + ".." + end + "); window indices are absolute record ordinals, so cycles must fall inside it");
|
||||
}
|
||||
@Override public void prebuffer(PrebufferProgress progress) { inner.prebuffer(progress); }
|
||||
@Override public boolean isComplete() { return inner.isComplete(); }
|
||||
@Override public CacheStats cacheStats() { return inner.cacheStats(); }
|
||||
|
||||
+30
-9
@@ -108,10 +108,31 @@ public class VectorDataBindingsTest {
|
||||
}
|
||||
|
||||
@Test
|
||||
void aWindowArgumentClipsTheReaderAndWarmsIt() {
|
||||
BaseVectors clipped = new BaseVectors("example:demo", "1..3", "eager", settings);
|
||||
assertArrayEquals(new float[] {1f, 2f}, clipped.apply(0), "windowed record 0 is underlying record 1");
|
||||
assertThrows(IndexOutOfBoundsException.class, () -> clipped.apply(2), "the window is two records long");
|
||||
void aWindowWarmsWithoutRenumbering() {
|
||||
// The window says which records to warm; it does not re-base
|
||||
// them. Index 1 is record 1, as it would be with no window.
|
||||
BaseVectors warmed = new BaseVectors("example:demo", "1..3", "eager", settings);
|
||||
assertArrayEquals(new float[] {1f, 2f}, warmed.apply(1), "index 1 is record 1, not the window's first record");
|
||||
assertArrayEquals(new float[] {2f, 3f}, warmed.apply(2));
|
||||
}
|
||||
|
||||
@Test
|
||||
void aWindowFarFromZeroReadsTheRecordsItsCyclesName() {
|
||||
// The field failure: a window starting well above zero used to
|
||||
// re-base to 0, so a run over those same cycles indexed past the
|
||||
// end and threw.
|
||||
BaseVectors slice = new BaseVectors("example:demo", "[7..10)", "none", settings);
|
||||
assertArrayEquals(new float[] {7f, 8f}, slice.apply(7));
|
||||
assertArrayEquals(new float[] {9f, 10f}, slice.apply(9));
|
||||
}
|
||||
|
||||
@Test
|
||||
void readingOutsideTheWindowStillWorks() {
|
||||
// Not an error, and load-bearing: VirtData probes a resolved
|
||||
// function with a small sample index before any cycle runs, so a
|
||||
// window starting above zero would otherwise fail to resolve.
|
||||
BaseVectors slice = new BaseVectors("example:demo", "[7..10)", "none", settings);
|
||||
assertArrayEquals(new float[] {1f, 2f}, slice.apply(1), "outside the warmed window, but readable");
|
||||
}
|
||||
|
||||
@Test
|
||||
@@ -119,16 +140,16 @@ public class VectorDataBindingsTest {
|
||||
BaseVectors vectors = new BaseVectors("example:demo", "1..3", "background", settings);
|
||||
assertNotNull(vectors.backgroundPrefetch(), "background mode hands back the running prefetch");
|
||||
vectors.backgroundPrefetch().join();
|
||||
assertArrayEquals(new float[] {2f, 3f}, vectors.apply(1));
|
||||
assertArrayEquals(new float[] {1f, 2f}, vectors.apply(1));
|
||||
}
|
||||
|
||||
@Test
|
||||
void aBindingWindowComposesWithAProfileWindow() {
|
||||
// The profile clips to records [2..8); the binding window is in
|
||||
// the clipped reader's coordinates, so 1..3 reads records 3..5.
|
||||
// The profile clips to records [2..8), which re-bases the reader
|
||||
// it hands back; the binding window then restricts that reader's
|
||||
// indices 1..2, and warming shifts them to dataset records 3..5.
|
||||
BaseVectors vectors = new BaseVectors("example:windowed", "1..3", "eager", settings);
|
||||
assertArrayEquals(new float[] {3f, 4f}, vectors.apply(0));
|
||||
assertThrows(IndexOutOfBoundsException.class, () -> vectors.apply(2));
|
||||
assertArrayEquals(new float[] {3f, 4f}, vectors.apply(1));
|
||||
// And warming shifts the same window to absolute records 3..5.
|
||||
DSWindow effective = CoreVectors.effectiveWindow(vectors.tdv, "base_vectors", "1..3");
|
||||
assertEquals(new DSWindow.Interval(3, 5), effective.intervals().get(0));
|
||||
|
||||
Reference in new issue
Block a user