From eca197e2e1aaf46771b3d8f10772386623311797 Mon Sep 17 00:00:00 2001 From: tellet-q Date: Fri, 10 Jul 2026 12:56:55 +0200 Subject: [PATCH 1/4] feat: per-stage indexing timings from the optimizations endpoint Co-Authored-By: Claude Opus 4.8 (1M context) --- DEVELOPMENT.md | 1 + README.md | 36 ++ src/args/mod.rs | 4 + src/main.rs | 96 +++- src/optimizations.rs | 469 ++++++++++++++++++++ src/results.rs | 23 +- tests/fixtures/optimizations_completed.json | 235 ++++++++++ 7 files changed, 854 insertions(+), 10 deletions(-) create mode 100644 src/optimizations.rs create mode 100644 tests/fixtures/optimizations_completed.json diff --git a/DEVELOPMENT.md b/DEVELOPMENT.md index 511f21c..f488ad8 100644 --- a/DEVELOPMENT.md +++ b/DEVELOPMENT.md @@ -13,6 +13,7 @@ src/ ├── upsert.rs # upsert request execution + timing ├── scroll.rs # scroll benchmarking processor ├── query.rs # search/scroll entry points +├── optimizations.rs # per-stage indexing timings (REST /collections/{c}/optimizations) ├── processor.rs # Processor trait + Timing measurement type ├── results.rs # unified `{config, results}` document written by --json ├── stats.rs # benchmark run loops (parallel/RPS), stats output, throttling diff --git a/README.md b/README.md index c2dec65..9a5a958 100644 --- a/README.md +++ b/README.md @@ -152,6 +152,42 @@ and search configs. The CLI still controls *how* the benchmark runs (`-n`, `-p`, The flag-driven path (`bfb --scroll --keywords 100 …`) is still available when you prefer flat CLI flags over a YAML file. +### Per-stage indexing timings + +`bfb upload` and legacy runs already wait for a green index before exiting. That +wait now also reports the server's own breakdown of what the optimizer spent the +time on: + +```bash +bfb upload --file config.yaml -n 1M --json results.json +``` + +``` +Index ready in 20.255 seconds +--- Optimization stages (1 completed, 18.068 s total) --- +vector_index/main_graph: 16.663360 s total over 1 run(s), max 16.663360 s +copy_data: 0.336660 s total over 1 run(s), max 0.336660 s +payload_index/keyword:color: 0.133206 s total over 1 run(s), max 0.133206 s +``` + +Stages are keyed by their path in the optimizer's progress tree, so a name that +appears under two parents stays distinct: `payload_index/keyword:color` (building +the field index) is separate from `vector_index/additional_links/keyword:color` +(adding HNSW links for that field). Named vectors add a level of their own — +`vector_index/image/main_graph`. + +The server exposes a rolling window of the last 16 completed optimizations, not +the ones belonging to any particular run — so bfb snapshots which had already +finished *before* the phase and excludes them, rather than claiming earlier work +as its own. Only `done` optimizations are aggregated; cancelled or errored ones +stopped partway through their tree and are counted separately as `unfinished`. + +This data is only exposed over Qdrant's **REST** API, so bfb reads it on the REST +port, derived from `--uri` by mapping port 6334 → 6333. A failure here warns +rather than failing a run that already has its headline number; +`--skip-server-stats` turns it off. It lands under `results.index.optimizations` +in `--json`. + ### `schema` — print the upload-config file schema Print an annotated YAML reference enumerating every option accepted by an diff --git a/src/args/mod.rs b/src/args/mod.rs index eaac486..1fa58e4 100644 --- a/src/args/mod.rs +++ b/src/args/mod.rs @@ -43,6 +43,10 @@ pub struct Args { #[clap(long, default_value = "http://localhost:6334", global = true)] pub uri: Vec, + /// Skip the REST-only extras: per-stage optimization timings after indexing + #[clap(long, default_value_t = false, global = true)] + pub skip_server_stats: bool, + /// Source of data to upload - fbin file. Random if not specified #[clap(long)] pub fbin: Option, diff --git a/src/main.rs b/src/main.rs index 7d7d8ec..d3eda43 100644 --- a/src/main.rs +++ b/src/main.rs @@ -15,6 +15,7 @@ mod config; mod dataset; mod fbin_reader; mod generators; +mod optimizations; mod processor; mod query; mod results; @@ -25,12 +26,89 @@ mod stats; mod upload; mod upsert; -/// Wait for the index and record how long it took. -async fn run_wait_index(args: &Args, stopped: Arc) -> Result { +/// Wait for the index, then read the server's per-stage optimization timings. +/// +/// `baseline` must have been captured *before* the work being timed started, so +/// optimizations left over from an earlier phase are not attributed to this one. +async fn wait_and_report_index( + args: &Args, + stopped: Arc, + baseline: optimizations::Baseline, +) -> Result { println!("Waiting for index to be ready..."); let wait_secs = collection::wait_index(args, stopped).await?; println!("Index ready in {wait_secs} seconds"); - Ok(IndexPhase { wait_secs }) + + Ok(IndexPhase { + wait_secs, + optimizations: fetch_optimization_stages(args, baseline).await, + }) +} + +/// The REST base URL used for endpoints gRPC does not expose. Derived from +/// `--uri` by mapping Qdrant's gRPC port to its REST one. +fn rest_uri(args: &Args) -> String { + optimizations::rest_url_from_grpc(args.uri.first().expect("clap guarantees one --uri")) +} + +/// Snapshot the optimizations that have already completed, so the ones this +/// phase triggers can be told apart from them. +/// +/// Best-effort, like the fetch itself: on failure we fall back to an empty +/// baseline, which over-counts rather than failing the run. +async fn optimization_baseline(args: &Args) -> optimizations::Baseline { + if args.skip_server_stats { + return optimizations::Baseline::none(); + } + let (rest_uri, collection) = (rest_uri(args), args.collection_name.clone()); + let api_key = std::env::var("QDRANT_API_KEY").ok(); + + tokio::task::spawn_blocking(move || { + optimizations::baseline(&rest_uri, &collection, api_key.as_deref()) + }) + .await + .ok() + .and_then(Result::ok) + .unwrap_or_default() +} + +/// Read per-stage optimization timings from `GET /collections/{c}/optimizations`. +/// +/// Best-effort: a benchmark that reached a green index has its headline number +/// already, so a REST failure (wrong port, no permission, older server) warns +/// rather than failing the run. +async fn fetch_optimization_stages( + args: &Args, + baseline: optimizations::Baseline, +) -> Option { + if args.skip_server_stats { + return None; + } + + let (rest_uri, collection) = (rest_uri(args), args.collection_name.clone()); + let api_key = std::env::var("QDRANT_API_KEY").ok(); + + // `ureq` is blocking; keep it off the async worker threads. + let fetched = tokio::task::spawn_blocking(move || { + optimizations::fetch(&rest_uri, &collection, api_key.as_deref(), &baseline) + }) + .await; + + match fetched { + Ok(Ok(report)) => { + report.print(); + Some(report) + } + Ok(Err(err)) => { + eprintln!("Warning: could not read optimization stages: {err:#}"); + eprintln!(" (pass --skip-server-stats to silence)"); + None + } + Err(err) => { + eprintln!("Warning: optimization-stage task failed: {err}"); + None + } + } } /// `bfb scroll --file config.yaml`: YAML-config-driven scroll. @@ -102,13 +180,17 @@ async fn run_upload( collection::recreate_collection_from_config(&config, &args, stopped.clone()).await?; } + // Indexing can start during upload, so the baseline predates it. + let baseline = optimization_baseline(&args).await; + if !args.skip_upload && !args.skip_setup { results.results.upload = Some(upload::upload_with_config(&args, &config, stopped.clone()).await?); } if !args.skip_wait_index && !args.skip_setup { - results.results.index = Some(run_wait_index(&args, stopped.clone()).await?); + results.results.index = + Some(wait_and_report_index(&args, stopped.clone(), baseline).await?); } results.write_if_requested(&args) @@ -133,12 +215,16 @@ async fn run_benchmark(args: Args, stopped: Arc) -> Result<()> { collection::recreate_collection(&args, stopped.clone()).await?; } + // Indexing can start during upload, so the baseline predates it. + let baseline = optimization_baseline(&args).await; + if !args.skip_upload && !args.skip_setup { results.results.upload = Some(upload::upload_data(&args, stopped.clone()).await?); } if !args.skip_wait_index && !args.skip_setup { - results.results.index = Some(run_wait_index(&args, stopped.clone()).await?); + results.results.index = + Some(wait_and_report_index(&args, stopped.clone(), baseline).await?); } if args.search || args.search_quality { diff --git a/src/optimizations.rs b/src/optimizations.rs new file mode 100644 index 0000000..89bb14a --- /dev/null +++ b/src/optimizations.rs @@ -0,0 +1,469 @@ +//! Per-stage indexing timings from `GET /collections/{c}/optimizations`. +//! +//! This is the one thing BFB cannot get over gRPC — the public API exposes it +//! only over REST (the gRPC `GetShardOptimizations` is an internal, node-to-node +//! RPC that ships this same JSON as opaque bytes). So we speak HTTP here, on the +//! REST port, reusing the `ureq` client already pulled in for dataset downloads. +//! +//! The server keeps the last 16 completed optimizations in a ring buffer, so +//! this must be read *after* indexing finishes but before too many further +//! optimizations run. + +use std::collections::{BTreeMap, HashSet}; + +use anyhow::{Context, Result}; +use serde::{Deserialize, Serialize}; + +/// Server default for `?completed_limit=`; also the depth of the server's +/// completed-optimization ring buffer. +const DEFAULT_COMPLETED_LIMIT: usize = 16; + +// ------------------------- wire types (server shape) ---------------------- + +/// The REST envelope: `{result, status, time}`. +#[derive(Debug, Deserialize)] +struct RestEnvelope { + result: T, +} + +#[derive(Debug, Default, Deserialize)] +pub struct OptimizationsResponse { + #[serde(default)] + pub running: Vec, + #[serde(default)] + pub completed: Option>, +} + +#[derive(Debug, Deserialize)] +pub struct Optimization { + /// Identifies this optimization run. Used to tell ours apart from those + /// that had already completed before the phase started. + pub uuid: String, + /// Name of the optimizer that ran (e.g. `indexing`, `merge`). + pub optimizer: String, + /// `"done"`, or a single-key object for `cancelled` / `error` (each carrying + /// a reason string), or `"optimizing"` while still running. + pub status: serde_json::Value, + pub progress: ProgressTree, +} + +impl Optimization { + /// Only `done` optimizations have trustworthy stage timings; a cancelled or + /// errored one stopped partway through its tree. + fn is_done(&self) -> bool { + self.status.as_str() == Some("done") + } +} + +/// A recursive stage tree. Each node carries its own wall-clock duration. +#[derive(Debug, Deserialize)] +pub struct ProgressTree { + pub name: String, + #[serde(default)] + pub duration_sec: Option, + #[serde(default)] + pub children: Vec, +} + +// ------------------------- reported types --------------------------------- + +/// Per-stage aggregate across every completed optimization. +#[derive(Debug, Clone, Default, Serialize, Deserialize, PartialEq)] +pub struct StageAggregate { + /// How many optimizations reported this stage. + pub count: usize, + pub total_secs: f64, + pub max_secs: f64, +} + +/// Per-stage breakdown of the optimizations the server ran, attached to the +/// index phase. +#[derive(Debug, Clone, Default, Serialize, Deserialize, PartialEq)] +pub struct OptimizationsReport { + /// Number of `done` optimizations aggregated below. + pub completed: usize, + /// Optimizations that had already completed before this phase started, and + /// so were excluded. See [`Baseline`]. + pub pre_existing_skipped: usize, + /// The server returned a full page of completed optimizations, so older + /// ones may have been evicted from its ring buffer before we read it — + /// meaning `stages` can undercount. Run fewer optimizations per phase. + pub possibly_truncated: bool, + /// Completed-but-not-`done` entries (cancelled or errored), whose partial + /// stage timings are excluded from `stages`. + pub unfinished: usize, + /// Number of optimizations still running when we read the endpoint. + pub running: usize, + /// Wall-clock seconds summed over every completed optimization's root node. + pub total_secs: f64, + /// Which optimizers ran, e.g. `indexing` → 5, `merge` → 1. + pub optimizers: BTreeMap, + /// Stage path → aggregate, e.g. `vector_index/main_graph`, `quantization`, + /// `payload_index/keyword:color`. + pub stages: BTreeMap, +} + +impl OptimizationsReport { + fn from_response(response: &OptimizationsResponse, baseline: &Baseline) -> Self { + let completed = response.completed.as_deref().unwrap_or_default(); + + let mut report = OptimizationsReport { + running: response.running.len(), + // The server caps the page, and its own buffer, at this many. + possibly_truncated: completed.len() >= DEFAULT_COMPLETED_LIMIT, + ..Default::default() + }; + + for optimization in completed { + // Optimizations that had already finished before the phase began + // belong to an earlier phase, not this one. + if baseline.contains(&optimization.uuid) { + report.pre_existing_skipped += 1; + continue; + } + if !optimization.is_done() { + report.unfinished += 1; + continue; + } + report.completed += 1; + *report + .optimizers + .entry(optimization.optimizer.clone()) + .or_default() += 1; + report.total_secs += optimization.progress.duration_sec.unwrap_or(0.0); + // The root node ("Segment Optimizing") is the total, already counted + // above; only its descendants are stages. + for child in &optimization.progress.children { + report.collect_stage(child, ""); + } + } + + report + } + + /// Fold `node` and its descendants into `stages`, keyed by path. + /// + /// Keys are paths rather than bare names because a name can repeat under + /// different parents with entirely different meanings: a payload field shows + /// up once under `payload_index` (building the field index) and again under + /// `additional_links` (adding HNSW links for it). Collapsing those into one + /// `keyword:color` bucket would sum two unrelated costs. + /// + /// Under `vector_index` the next level is the *vector name*, which is empty + /// for the unnamed default vector. Such levels are skipped rather than + /// recorded, so a default vector yields `vector_index/main_graph` (not + /// `vector_index//main_graph`) while a named one yields + /// `vector_index/image/main_graph`. + fn collect_stage(&mut self, node: &ProgressTree, parent_path: &str) { + let path = if node.name.is_empty() { + parent_path.to_string() + } else if parent_path.is_empty() { + node.name.clone() + } else { + format!("{parent_path}/{}", node.name) + }; + + if !node.name.is_empty() { + let secs = node.duration_sec.unwrap_or(0.0); + let stage = self.stages.entry(path.clone()).or_default(); + stage.count += 1; + stage.total_secs += secs; + stage.max_secs = stage.max_secs.max(secs); + } + + for child in &node.children { + self.collect_stage(child, &path); + } + } + + /// Print the stages slowest-first. + pub fn print(&self) { + if self.completed == 0 { + println!("No completed optimizations reported by the server"); + return; + } + println!( + "--- Optimization stages ({} completed, {:.3} s total) ---", + self.completed, self.total_secs + ); + if self.unfinished > 0 { + println!( + "({} cancelled/errored optimization(s) excluded)", + self.unfinished + ); + } + if self.possibly_truncated { + println!( + "(warning: the server's completed-optimization buffer was full; \ + older stages may be missing)" + ); + } + let mut stages: Vec<_> = self.stages.iter().collect(); + stages.sort_unstable_by(|(_, a), (_, b)| b.total_secs.total_cmp(&a.total_secs)); + for (name, stage) in stages { + println!( + "{name}: {:.6} s total over {} run(s), max {:.6} s", + stage.total_secs, stage.count, stage.max_secs + ); + } + } +} + +// ------------------------- fetching --------------------------------------- + +/// Derive the REST base URL from a gRPC `--uri`. +/// +/// Qdrant serves gRPC on 6334 and REST on 6333, so this is a one-port shift. +/// A `--uri` on any other port is returned unchanged. +pub fn rest_url_from_grpc(uri: &str) -> String { + let trimmed = uri.trim_end_matches('/'); + match trimmed.rsplit_once(":6334") { + Some((head, "")) => format!("{head}:6333"), + _ => trimmed.to_string(), + } +} + +/// The optimizations that had already completed when a phase began. +/// +/// The server reports a rolling window of completed optimizations, not the ones +/// belonging to any particular phase. Without a baseline, `bfb wait-index` would +/// happily attribute a preceding `bfb create-index`'s work to indexing. +#[derive(Debug, Default, Clone)] +pub struct Baseline(HashSet); + +impl Baseline { + fn contains(&self, uuid: &str) -> bool { + self.0.contains(uuid) + } + + /// An empty baseline: every completed optimization counts. + pub fn none() -> Self { + Baseline::default() + } +} + +/// Record which optimizations have already completed, to exclude them later. +pub fn baseline(rest_url: &str, collection: &str, api_key: Option<&str>) -> Result { + let response = get(rest_url, collection, api_key)?; + Ok(Baseline( + response + .completed + .unwrap_or_default() + .into_iter() + .map(|optimization| optimization.uuid) + .collect(), + )) +} + +/// Read completed optimization stages for `collection`, excluding anything +/// already present in `baseline`. +pub fn fetch( + rest_url: &str, + collection: &str, + api_key: Option<&str>, + baseline: &Baseline, +) -> Result { + let response = get(rest_url, collection, api_key)?; + Ok(OptimizationsReport::from_response(&response, baseline)) +} + +fn get(rest_url: &str, collection: &str, api_key: Option<&str>) -> Result { + let url = format!( + "{}/collections/{collection}/optimizations?with=completed&completed_limit={DEFAULT_COMPLETED_LIMIT}", + rest_url.trim_end_matches('/') + ); + + let agent = ureq::Agent::new_with_defaults(); + let mut request = agent.get(&url); + if let Some(key) = api_key { + request = request.header("api-key", key); + } + + let body = request + .call() + .with_context(|| format!("failed to GET {url}"))? + .into_body() + .read_to_string() + .with_context(|| format!("failed to read response from {url}"))?; + + let envelope: RestEnvelope = serde_json::from_str(&body) + .with_context(|| format!("failed to parse optimizations response from {url}"))?; + Ok(envelope.result) +} + +#[cfg(test)] +mod tests { + use super::*; + + fn parse_fixture() -> OptimizationsResponse { + let text = include_str!("../tests/fixtures/optimizations_completed.json"); + let envelope: RestEnvelope = serde_json::from_str(text).unwrap(); + envelope.result + } + + #[test] + fn parses_real_server_response() { + let response = parse_fixture(); + assert_eq!(response.completed.as_ref().unwrap().len(), 2); + assert!(response.running.is_empty()); + assert_eq!( + response.completed.as_ref().unwrap()[0].optimizer, + "indexing" + ); + } + + #[test] + fn aggregates_the_stages_the_benchmarks_ask_for() { + let report = OptimizationsReport::from_response(&parse_fixture(), &Baseline::none()); + + assert_eq!(report.completed, 2); + assert_eq!(report.running, 0); + + // The stages the indexing benchmarks are written against. + for stage in ["vector_index/main_graph", "quantization", "payload_index"] { + assert!( + report.stages.contains_key(stage), + "missing stage {stage}; got {:?}", + report.stages.keys().collect::>() + ); + } + + // The fixture uses the unnamed default vector, so the vector-name level + // under `vector_index` is empty and must collapse out of the path. + let main_graph = &report.stages["vector_index/main_graph"]; + assert_eq!(main_graph.count, 2, "one per completed optimization"); + assert!(main_graph.total_secs > 0.0); + assert!(main_graph.max_secs <= main_graph.total_secs); + + // The unnamed wrapper node must not become a stage, nor leave an empty + // path segment behind. + assert!(!report.stages.contains_key("")); + assert!(report.stages.keys().all(|k| !k.contains("//"))); + + // The same field name appears under two different parents with two + // different meanings; they must not be summed into one bucket. + assert!(report.stages.contains_key("payload_index/keyword:a")); + assert!( + report + .stages + .contains_key("vector_index/additional_links/keyword:a") + ); + } + + #[test] + fn total_secs_sums_root_nodes_only() { + let report = OptimizationsReport::from_response(&parse_fixture(), &Baseline::none()); + // Root duration exceeds any single child stage, and is not the sum of + // all stages (which would double-count nested nodes). + assert!(report.total_secs > report.stages["vector_index/main_graph"].total_secs); + let stage_sum: f64 = report.stages.values().map(|s| s.total_secs).sum(); + assert!( + stage_sum > report.total_secs, + "nested stages overlap in time" + ); + } + + #[test] + fn records_which_optimizers_ran() { + let report = OptimizationsReport::from_response(&parse_fixture(), &Baseline::none()); + assert_eq!(report.optimizers.get("indexing"), Some(&2)); + assert_eq!(report.unfinished, 0); + } + + #[test] + fn cancelled_and_errored_optimizations_are_excluded() { + // A cancelled/errored entry stopped partway through its stage tree, so + // folding its durations in would understate the real stage cost. + let json = r#"{ + "running": [], + "completed": [ + {"uuid": "a", "optimizer": "indexing", "status": "done", + "progress": {"name": "Segment Optimizing", "duration_sec": 2.0, + "children": [{"name": "main_graph", "duration_sec": 1.5}]}}, + {"uuid": "b", "optimizer": "indexing", "status": {"cancelled": "shutdown"}, + "progress": {"name": "Segment Optimizing", "duration_sec": 0.3, + "children": [{"name": "main_graph", "duration_sec": 0.1}]}}, + {"uuid": "c", "optimizer": "merge", "status": {"error": "disk full"}, + "progress": {"name": "Segment Optimizing", "duration_sec": 0.2, "children": []}} + ] + }"#; + let response: OptimizationsResponse = serde_json::from_str(json).unwrap(); + let report = OptimizationsReport::from_response(&response, &Baseline::none()); + + assert_eq!(report.completed, 1); + assert_eq!(report.unfinished, 2); + assert_eq!(report.total_secs, 2.0, "only the `done` root counts"); + assert_eq!(report.stages["main_graph"].total_secs, 1.5); + assert_eq!(report.stages["main_graph"].count, 1); + assert_eq!(report.optimizers.get("merge"), None); + } + + #[test] + fn optimizations_completed_before_the_phase_are_excluded() { + // A `bfb create-index` run leaves completed optimizations behind; the + // `bfb wait-index` that follows must not claim them as indexing work. + let response = parse_fixture(); + let first = response.completed.as_ref().unwrap()[0].uuid.clone(); + let baseline = Baseline(HashSet::from([first])); + + let report = OptimizationsReport::from_response(&response, &baseline); + + assert_eq!(report.completed, 1, "only the new optimization counts"); + assert_eq!(report.pre_existing_skipped, 1); + assert_eq!(report.stages["vector_index/main_graph"].count, 1); + + // Without the baseline, both would be attributed to this phase. + let unfiltered = OptimizationsReport::from_response(&response, &Baseline::none()); + assert_eq!(unfiltered.completed, 2); + assert!(unfiltered.total_secs > report.total_secs); + } + + #[test] + fn a_full_page_of_completed_optimizations_flags_truncation() { + // The server keeps only the last 16; a full page means older stages may + // already be gone, so `stages` can undercount. + let one = r#"{"uuid": "%", "optimizer": "indexing", "status": "done", + "progress": {"name": "Segment Optimizing", "duration_sec": 1.0, "children": []}}"#; + let entries: Vec = (0..16).map(|i| one.replace('%', &i.to_string())).collect(); + let json = format!(r#"{{"running": [], "completed": [{}]}}"#, entries.join(",")); + + let response: OptimizationsResponse = serde_json::from_str(&json).unwrap(); + let report = OptimizationsReport::from_response(&response, &Baseline::none()); + + assert_eq!(report.completed, 16); + assert!(report.possibly_truncated); + } + + #[test] + fn a_partial_page_is_not_flagged_as_truncated() { + let report = OptimizationsReport::from_response(&parse_fixture(), &Baseline::none()); + assert!(!report.possibly_truncated); + } + + #[test] + fn empty_response_reports_nothing() { + let report = OptimizationsReport::from_response( + &OptimizationsResponse::default(), + &Baseline::none(), + ); + assert_eq!(report.completed, 0); + assert!(report.stages.is_empty()); + assert_eq!(report.total_secs, 0.0); + } + + #[test] + fn derives_the_rest_url_from_the_grpc_uri() { + assert_eq!( + rest_url_from_grpc("http://localhost:6334"), + "http://localhost:6334".replace("6334", "6333") + ); + assert_eq!( + rest_url_from_grpc("https://xyz.cloud.qdrant.io:6334/"), + "https://xyz.cloud.qdrant.io:6333" + ); + // A non-default port is left alone rather than silently rewritten. + assert_eq!(rest_url_from_grpc("http://host:7000"), "http://host:7000"); + // A port that merely *contains* 6334 must not match. + assert_eq!(rest_url_from_grpc("http://host:63340"), "http://host:63340"); + } +} diff --git a/src/results.rs b/src/results.rs index b64fff7..1b05519 100644 --- a/src/results.rs +++ b/src/results.rs @@ -11,6 +11,7 @@ use anyhow::{Context, Result}; use serde::{Deserialize, Serialize}; use crate::args::Args; +use crate::optimizations::OptimizationsReport; use crate::processor::Timing; /// Top-level document: `{ config: {...}, results: {...} }`. @@ -87,10 +88,13 @@ pub struct UploadPhase { pub points_per_sec: f64, } -/// M2: time spent waiting for the collection to report a green index. -#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)] +/// Time spent waiting for the collection to report a green index, plus the +/// server's own per-stage breakdown of the optimizations it ran. +#[derive(Debug, Clone, Default, Serialize, Deserialize, PartialEq)] pub struct IndexPhase { pub wait_secs: f64, + #[serde(skip_serializing_if = "Option::is_none")] + pub optimizations: Option, } /// A search or scroll phase: latency/throughput series plus their summaries. @@ -285,7 +289,10 @@ mod tests { #[test] fn skipped_phases_are_omitted_from_json() { let results = PhaseResults { - index: Some(IndexPhase { wait_secs: 1.5 }), + index: Some(IndexPhase { + wait_secs: 1.5, + ..Default::default() + }), ..Default::default() }; let json = serde_json::to_string(&results).unwrap(); @@ -331,7 +338,10 @@ mod tests { fn document_roundtrips() { let mut doc = doc(PhaseResults { upload: Some(UploadPhase::new(1.0, 100)), - index: Some(IndexPhase { wait_secs: 0.5 }), + index: Some(IndexPhase { + wait_secs: 0.5, + ..Default::default() + }), search: Some(phase(1.0)), scroll: None, }); @@ -387,7 +397,10 @@ mod tests { fn upload_only_run_emits_no_legacy_fields() { let mut doc = doc(PhaseResults { upload: Some(UploadPhase::new(1.0, 100)), - index: Some(IndexPhase { wait_secs: 0.5 }), + index: Some(IndexPhase { + wait_secs: 0.5, + ..Default::default() + }), ..Default::default() }); doc.populate_legacy_fields(); diff --git a/tests/fixtures/optimizations_completed.json b/tests/fixtures/optimizations_completed.json new file mode 100644 index 0000000..1b30a08 --- /dev/null +++ b/tests/fixtures/optimizations_completed.json @@ -0,0 +1,235 @@ +{ + "result": { + "summary": { + "queued_optimizations": 0, + "queued_segments": 0, + "queued_points": 0, + "idle_segments": 4 + }, + "running": [], + "completed": [ + { + "uuid": "21a7e712-dd1e-4069-9f06-b292e4997ef7", + "optimizer": "indexing", + "status": "done", + "segments": [ + { + "uuid": "1821de91-a124-4c57-bcd8-cdad6b535194", + "points_count": 15000 + }, + { + "uuid": "42488f00-c7f9-4e20-b59a-7273974397b8", + "points_count": 1012 + } + ], + "progress": { + "name": "Segment Optimizing", + "started_at": "2026-07-10T09:40:21.765178905Z", + "finished_at": "2026-07-10T09:40:23.145000753Z", + "duration_sec": 1.379821628, + "children": [ + { + "name": "copy_data", + "started_at": "2026-07-10T09:40:21.836839467Z", + "finished_at": "2026-07-10T09:40:21.895042118Z", + "duration_sec": 0.058201569 + }, + { + "name": "populate_vector_storages", + "started_at": "2026-07-10T09:40:21.895044172Z", + "finished_at": "2026-07-10T09:40:21.946601106Z", + "duration_sec": 0.051556734 + }, + { + "name": "wait_cpu_permit", + "started_at": "2026-07-10T09:40:21.946602378Z", + "finished_at": "2026-07-10T09:40:22.160921861Z", + "duration_sec": 0.214319082 + }, + { + "name": "quantization", + "started_at": "2026-07-10T09:40:22.192142510Z", + "finished_at": "2026-07-10T09:40:22.192145886Z", + "duration_sec": 2.795e-06 + }, + { + "name": "payload_index", + "started_at": "2026-07-10T09:40:22.231933459Z", + "finished_at": "2026-07-10T09:40:22.276830503Z", + "duration_sec": 0.044896443, + "children": [ + { + "name": "keyword:a", + "started_at": "2026-07-10T09:40:22.235260060Z", + "finished_at": "2026-07-10T09:40:22.276829041Z", + "duration_sec": 0.041560936 + } + ] + }, + { + "name": "vector_index", + "started_at": "2026-07-10T09:40:22.282012799Z", + "finished_at": "2026-07-10T09:40:23.121430494Z", + "duration_sec": 0.839417014, + "children": [ + { + "name": "", + "started_at": "2026-07-10T09:40:22.282019201Z", + "finished_at": "2026-07-10T09:40:23.121374811Z", + "duration_sec": 0.83935554, + "children": [ + { + "name": "migrate", + "started_at": "2026-07-10T09:40:22.284579186Z", + "finished_at": "2026-07-10T09:40:22.285133379Z", + "duration_sec": 0.000553562 + }, + { + "name": "main_graph", + "started_at": "2026-07-10T09:40:22.285133650Z", + "finished_at": "2026-07-10T09:40:23.082625933Z", + "duration_sec": 0.797491983, + "done": 15000, + "total": 15000 + }, + { + "name": "additional_links", + "started_at": "2026-07-10T09:40:23.082628588Z", + "finished_at": "2026-07-10T09:40:23.095550070Z", + "duration_sec": 0.012921402, + "children": [ + { + "name": "keyword:a", + "started_at": "2026-07-10T09:40:23.095520355Z", + "finished_at": "2026-07-10T09:40:23.095546554Z", + "duration_sec": 2.5608e-05, + "done": 0 + } + ] + } + ] + } + ] + }, + { + "name": "sparse_vector_index", + "started_at": "2026-07-10T09:40:23.121430985Z", + "finished_at": "2026-07-10T09:40:23.121431566Z", + "duration_sec": 5.01e-07 + } + ] + } + }, + { + "uuid": "93cc97c6-ace5-4d4d-ba32-c2144e2b5f7a", + "optimizer": "indexing", + "status": "done", + "segments": [ + { + "uuid": "c6d47929-9308-4a1a-b575-c7752c400630", + "points_count": 3500 + } + ], + "progress": { + "name": "Segment Optimizing", + "started_at": "2026-07-10T09:40:20.800774711Z", + "finished_at": "2026-07-10T09:40:22.123890285Z", + "duration_sec": 1.323115413, + "children": [ + { + "name": "copy_data", + "started_at": "2026-07-10T09:40:20.870915263Z", + "finished_at": "2026-07-10T09:40:20.888020114Z", + "duration_sec": 0.017103609 + }, + { + "name": "populate_vector_storages", + "started_at": "2026-07-10T09:40:20.888022348Z", + "finished_at": "2026-07-10T09:40:20.943550449Z", + "duration_sec": 0.055527811 + }, + { + "name": "wait_cpu_permit", + "started_at": "2026-07-10T09:40:20.943551912Z", + "finished_at": "2026-07-10T09:40:21.765102494Z", + "duration_sec": 0.821550302 + }, + { + "name": "quantization", + "started_at": "2026-07-10T09:40:21.793020004Z", + "finished_at": "2026-07-10T09:40:21.793023851Z", + "duration_sec": 2.895e-06 + }, + { + "name": "payload_index", + "started_at": "2026-07-10T09:40:21.815024641Z", + "finished_at": "2026-07-10T09:40:21.857925898Z", + "duration_sec": 0.042900546, + "children": [ + { + "name": "keyword:a", + "started_at": "2026-07-10T09:40:21.822396678Z", + "finished_at": "2026-07-10T09:40:21.857924255Z", + "duration_sec": 0.035526415 + } + ] + }, + { + "name": "vector_index", + "started_at": "2026-07-10T09:40:21.860372765Z", + "finished_at": "2026-07-10T09:40:22.099029295Z", + "duration_sec": 0.238655959, + "children": [ + { + "name": "", + "started_at": "2026-07-10T09:40:21.860377454Z", + "finished_at": "2026-07-10T09:40:22.098965377Z", + "duration_sec": 0.238587823, + "children": [ + { + "name": "migrate", + "started_at": "2026-07-10T09:40:21.861825747Z", + "finished_at": "2026-07-10T09:40:21.861825747Z", + "duration_sec": 0.0 + }, + { + "name": "main_graph", + "started_at": "2026-07-10T09:40:21.861826749Z", + "finished_at": "2026-07-10T09:40:22.080672230Z", + "duration_sec": 0.218845121, + "done": 5500, + "total": 5500 + }, + { + "name": "additional_links", + "started_at": "2026-07-10T09:40:22.080674735Z", + "finished_at": "2026-07-10T09:40:22.083122724Z", + "duration_sec": 0.002447788, + "children": [ + { + "name": "keyword:a", + "started_at": "2026-07-10T09:40:22.083107005Z", + "finished_at": "2026-07-10T09:40:22.083120840Z", + "duration_sec": 1.3505e-05, + "done": 0 + } + ] + } + ] + } + ] + }, + { + "name": "sparse_vector_index", + "started_at": "2026-07-10T09:40:22.099030157Z", + "finished_at": "2026-07-10T09:40:22.099030648Z", + "duration_sec": 3.81e-07 + } + ] + } + } + ] + }, + "status": "ok", + "time": 5.2477e-05 +} \ No newline at end of file From 0f28eb584c56cc5b6eafbc4cdd3eaee93fcd18ab Mon Sep 17 00:00:00 2001 From: tellet-q Date: Wed, 15 Jul 2026 08:20:53 +0200 Subject: [PATCH 2/4] Address review comments --- src/main.rs | 39 ++++++++++++++++++++++++++++++--------- src/optimizations.rs | 29 ++++++++++++++++++++++++----- 2 files changed, 54 insertions(+), 14 deletions(-) diff --git a/src/main.rs b/src/main.rs index d3eda43..4e1851e 100644 --- a/src/main.rs +++ b/src/main.rs @@ -54,22 +54,42 @@ fn rest_uri(args: &Args) -> String { /// Snapshot the optimizations that have already completed, so the ones this /// phase triggers can be told apart from them. /// -/// Best-effort, like the fetch itself: on failure we fall back to an empty -/// baseline, which over-counts rather than failing the run. +/// Best-effort: on failure we warn and fall back to an empty baseline, which +/// over-counts (may attribute pre-existing optimizations to this phase) rather +/// than failing the run. async fn optimization_baseline(args: &Args) -> optimizations::Baseline { if args.skip_server_stats { return optimizations::Baseline::none(); } let (rest_uri, collection) = (rest_uri(args), args.collection_name.clone()); let api_key = std::env::var("QDRANT_API_KEY").ok(); + let timeout = optimization_http_timeout(args); - tokio::task::spawn_blocking(move || { - optimizations::baseline(&rest_uri, &collection, api_key.as_deref()) + let captured = tokio::task::spawn_blocking(move || { + optimizations::baseline(&rest_uri, &collection, api_key.as_deref(), timeout) }) - .await - .ok() - .and_then(Result::ok) - .unwrap_or_default() + .await; + + match captured { + Ok(Ok(baseline)) => baseline, + Ok(Err(err)) => { + eprintln!("Warning: could not capture optimization baseline: {err:#}"); + eprintln!(" (stage attribution may include pre-existing optimizations)"); + optimizations::Baseline::none() + } + Err(err) => { + eprintln!("Warning: optimization-baseline task failed: {err}"); + eprintln!(" (stage attribution may include pre-existing optimizations)"); + optimizations::Baseline::none() + } + } +} + +/// The bound applied to the best-effort optimization REST calls: the CLI +/// `--timeout` if set, otherwise `optimizations`' own fallback. +fn optimization_http_timeout(args: &Args) -> Option { + args.timeout + .map(|secs| std::time::Duration::from_secs(secs as u64)) } /// Read per-stage optimization timings from `GET /collections/{c}/optimizations`. @@ -87,10 +107,11 @@ async fn fetch_optimization_stages( let (rest_uri, collection) = (rest_uri(args), args.collection_name.clone()); let api_key = std::env::var("QDRANT_API_KEY").ok(); + let timeout = optimization_http_timeout(args); // `ureq` is blocking; keep it off the async worker threads. let fetched = tokio::task::spawn_blocking(move || { - optimizations::fetch(&rest_uri, &collection, api_key.as_deref(), &baseline) + optimizations::fetch(&rest_uri, &collection, api_key.as_deref(), &baseline, timeout) }) .await; diff --git a/src/optimizations.rs b/src/optimizations.rs index 89bb14a..87f22e3 100644 --- a/src/optimizations.rs +++ b/src/optimizations.rs @@ -10,6 +10,7 @@ //! optimizations run. use std::collections::{BTreeMap, HashSet}; +use std::time::Duration; use anyhow::{Context, Result}; use serde::{Deserialize, Serialize}; @@ -18,6 +19,9 @@ use serde::{Deserialize, Serialize}; /// completed-optimization ring buffer. const DEFAULT_COMPLETED_LIMIT: usize = 16; +/// Bound the REST call when `--timeout` is unset; `ureq` applies none by default. +const DEFAULT_HTTP_TIMEOUT: Duration = Duration::from_secs(30); + // ------------------------- wire types (server shape) ---------------------- /// The REST envelope: `{result, status, time}`. @@ -243,8 +247,13 @@ impl Baseline { } /// Record which optimizations have already completed, to exclude them later. -pub fn baseline(rest_url: &str, collection: &str, api_key: Option<&str>) -> Result { - let response = get(rest_url, collection, api_key)?; +pub fn baseline( + rest_url: &str, + collection: &str, + api_key: Option<&str>, + timeout: Option, +) -> Result { + let response = get(rest_url, collection, api_key, timeout)?; Ok(Baseline( response .completed @@ -262,18 +271,28 @@ pub fn fetch( collection: &str, api_key: Option<&str>, baseline: &Baseline, + timeout: Option, ) -> Result { - let response = get(rest_url, collection, api_key)?; + let response = get(rest_url, collection, api_key, timeout)?; Ok(OptimizationsReport::from_response(&response, baseline)) } -fn get(rest_url: &str, collection: &str, api_key: Option<&str>) -> Result { +fn get( + rest_url: &str, + collection: &str, + api_key: Option<&str>, + timeout: Option, +) -> Result { let url = format!( "{}/collections/{collection}/optimizations?with=completed&completed_limit={DEFAULT_COMPLETED_LIMIT}", rest_url.trim_end_matches('/') ); - let agent = ureq::Agent::new_with_defaults(); + let agent = ureq::Agent::new_with_config( + ureq::config::Config::builder() + .timeout_global(Some(timeout.unwrap_or(DEFAULT_HTTP_TIMEOUT))) + .build(), + ); let mut request = agent.get(&url); if let Some(key) = api_key { request = request.header("api-key", key); From c5a9ee34d5de2eb4bd4b648fd10386ecb8e6eecd Mon Sep 17 00:00:00 2001 From: tellet-q Date: Wed, 15 Jul 2026 08:22:29 +0200 Subject: [PATCH 3/4] Fix formatter --- src/main.rs | 8 +++++++- 1 file changed, 7 insertions(+), 1 deletion(-) diff --git a/src/main.rs b/src/main.rs index 4e1851e..6503063 100644 --- a/src/main.rs +++ b/src/main.rs @@ -111,7 +111,13 @@ async fn fetch_optimization_stages( // `ureq` is blocking; keep it off the async worker threads. let fetched = tokio::task::spawn_blocking(move || { - optimizations::fetch(&rest_uri, &collection, api_key.as_deref(), &baseline, timeout) + optimizations::fetch( + &rest_uri, + &collection, + api_key.as_deref(), + &baseline, + timeout, + ) }) .await; From 77e19503cc6cd4c86891204932e5a605f8d819d6 Mon Sep 17 00:00:00 2001 From: tellet-q Date: Thu, 16 Jul 2026 09:18:21 +0200 Subject: [PATCH 4/4] Remove per-stage optimization scraping (unreliable ring-buffer read) The /collections/{c}/optimizations endpoint keeps only the last 16 completed optimizations, so a read-once-at-the-end sample silently undercounts on the large collections indexing benchmarks target. Total index time (results.index.wait_secs) is measured directly and stays; per-stage attribution belongs in the benchmark that needs it, polling the endpoint continuously. Co-Authored-By: Claude Opus 4.8 (1M context) --- DEVELOPMENT.md | 1 - README.md | 36 -- src/args/mod.rs | 4 - src/main.rs | 123 +---- src/optimizations.rs | 488 -------------------- src/results.rs | 23 +- tests/fixtures/optimizations_completed.json | 235 ---------- 7 files changed, 10 insertions(+), 900 deletions(-) delete mode 100644 src/optimizations.rs delete mode 100644 tests/fixtures/optimizations_completed.json diff --git a/DEVELOPMENT.md b/DEVELOPMENT.md index f488ad8..511f21c 100644 --- a/DEVELOPMENT.md +++ b/DEVELOPMENT.md @@ -13,7 +13,6 @@ src/ ├── upsert.rs # upsert request execution + timing ├── scroll.rs # scroll benchmarking processor ├── query.rs # search/scroll entry points -├── optimizations.rs # per-stage indexing timings (REST /collections/{c}/optimizations) ├── processor.rs # Processor trait + Timing measurement type ├── results.rs # unified `{config, results}` document written by --json ├── stats.rs # benchmark run loops (parallel/RPS), stats output, throttling diff --git a/README.md b/README.md index 9a5a958..c2dec65 100644 --- a/README.md +++ b/README.md @@ -152,42 +152,6 @@ and search configs. The CLI still controls *how* the benchmark runs (`-n`, `-p`, The flag-driven path (`bfb --scroll --keywords 100 …`) is still available when you prefer flat CLI flags over a YAML file. -### Per-stage indexing timings - -`bfb upload` and legacy runs already wait for a green index before exiting. That -wait now also reports the server's own breakdown of what the optimizer spent the -time on: - -```bash -bfb upload --file config.yaml -n 1M --json results.json -``` - -``` -Index ready in 20.255 seconds ---- Optimization stages (1 completed, 18.068 s total) --- -vector_index/main_graph: 16.663360 s total over 1 run(s), max 16.663360 s -copy_data: 0.336660 s total over 1 run(s), max 0.336660 s -payload_index/keyword:color: 0.133206 s total over 1 run(s), max 0.133206 s -``` - -Stages are keyed by their path in the optimizer's progress tree, so a name that -appears under two parents stays distinct: `payload_index/keyword:color` (building -the field index) is separate from `vector_index/additional_links/keyword:color` -(adding HNSW links for that field). Named vectors add a level of their own — -`vector_index/image/main_graph`. - -The server exposes a rolling window of the last 16 completed optimizations, not -the ones belonging to any particular run — so bfb snapshots which had already -finished *before* the phase and excludes them, rather than claiming earlier work -as its own. Only `done` optimizations are aggregated; cancelled or errored ones -stopped partway through their tree and are counted separately as `unfinished`. - -This data is only exposed over Qdrant's **REST** API, so bfb reads it on the REST -port, derived from `--uri` by mapping port 6334 → 6333. A failure here warns -rather than failing a run that already has its headline number; -`--skip-server-stats` turns it off. It lands under `results.index.optimizations` -in `--json`. - ### `schema` — print the upload-config file schema Print an annotated YAML reference enumerating every option accepted by an diff --git a/src/args/mod.rs b/src/args/mod.rs index 1fa58e4..eaac486 100644 --- a/src/args/mod.rs +++ b/src/args/mod.rs @@ -43,10 +43,6 @@ pub struct Args { #[clap(long, default_value = "http://localhost:6334", global = true)] pub uri: Vec, - /// Skip the REST-only extras: per-stage optimization timings after indexing - #[clap(long, default_value_t = false, global = true)] - pub skip_server_stats: bool, - /// Source of data to upload - fbin file. Random if not specified #[clap(long)] pub fbin: Option, diff --git a/src/main.rs b/src/main.rs index 6503063..7d7d8ec 100644 --- a/src/main.rs +++ b/src/main.rs @@ -15,7 +15,6 @@ mod config; mod dataset; mod fbin_reader; mod generators; -mod optimizations; mod processor; mod query; mod results; @@ -26,116 +25,12 @@ mod stats; mod upload; mod upsert; -/// Wait for the index, then read the server's per-stage optimization timings. -/// -/// `baseline` must have been captured *before* the work being timed started, so -/// optimizations left over from an earlier phase are not attributed to this one. -async fn wait_and_report_index( - args: &Args, - stopped: Arc, - baseline: optimizations::Baseline, -) -> Result { +/// Wait for the index and record how long it took. +async fn run_wait_index(args: &Args, stopped: Arc) -> Result { println!("Waiting for index to be ready..."); let wait_secs = collection::wait_index(args, stopped).await?; println!("Index ready in {wait_secs} seconds"); - - Ok(IndexPhase { - wait_secs, - optimizations: fetch_optimization_stages(args, baseline).await, - }) -} - -/// The REST base URL used for endpoints gRPC does not expose. Derived from -/// `--uri` by mapping Qdrant's gRPC port to its REST one. -fn rest_uri(args: &Args) -> String { - optimizations::rest_url_from_grpc(args.uri.first().expect("clap guarantees one --uri")) -} - -/// Snapshot the optimizations that have already completed, so the ones this -/// phase triggers can be told apart from them. -/// -/// Best-effort: on failure we warn and fall back to an empty baseline, which -/// over-counts (may attribute pre-existing optimizations to this phase) rather -/// than failing the run. -async fn optimization_baseline(args: &Args) -> optimizations::Baseline { - if args.skip_server_stats { - return optimizations::Baseline::none(); - } - let (rest_uri, collection) = (rest_uri(args), args.collection_name.clone()); - let api_key = std::env::var("QDRANT_API_KEY").ok(); - let timeout = optimization_http_timeout(args); - - let captured = tokio::task::spawn_blocking(move || { - optimizations::baseline(&rest_uri, &collection, api_key.as_deref(), timeout) - }) - .await; - - match captured { - Ok(Ok(baseline)) => baseline, - Ok(Err(err)) => { - eprintln!("Warning: could not capture optimization baseline: {err:#}"); - eprintln!(" (stage attribution may include pre-existing optimizations)"); - optimizations::Baseline::none() - } - Err(err) => { - eprintln!("Warning: optimization-baseline task failed: {err}"); - eprintln!(" (stage attribution may include pre-existing optimizations)"); - optimizations::Baseline::none() - } - } -} - -/// The bound applied to the best-effort optimization REST calls: the CLI -/// `--timeout` if set, otherwise `optimizations`' own fallback. -fn optimization_http_timeout(args: &Args) -> Option { - args.timeout - .map(|secs| std::time::Duration::from_secs(secs as u64)) -} - -/// Read per-stage optimization timings from `GET /collections/{c}/optimizations`. -/// -/// Best-effort: a benchmark that reached a green index has its headline number -/// already, so a REST failure (wrong port, no permission, older server) warns -/// rather than failing the run. -async fn fetch_optimization_stages( - args: &Args, - baseline: optimizations::Baseline, -) -> Option { - if args.skip_server_stats { - return None; - } - - let (rest_uri, collection) = (rest_uri(args), args.collection_name.clone()); - let api_key = std::env::var("QDRANT_API_KEY").ok(); - let timeout = optimization_http_timeout(args); - - // `ureq` is blocking; keep it off the async worker threads. - let fetched = tokio::task::spawn_blocking(move || { - optimizations::fetch( - &rest_uri, - &collection, - api_key.as_deref(), - &baseline, - timeout, - ) - }) - .await; - - match fetched { - Ok(Ok(report)) => { - report.print(); - Some(report) - } - Ok(Err(err)) => { - eprintln!("Warning: could not read optimization stages: {err:#}"); - eprintln!(" (pass --skip-server-stats to silence)"); - None - } - Err(err) => { - eprintln!("Warning: optimization-stage task failed: {err}"); - None - } - } + Ok(IndexPhase { wait_secs }) } /// `bfb scroll --file config.yaml`: YAML-config-driven scroll. @@ -207,17 +102,13 @@ async fn run_upload( collection::recreate_collection_from_config(&config, &args, stopped.clone()).await?; } - // Indexing can start during upload, so the baseline predates it. - let baseline = optimization_baseline(&args).await; - if !args.skip_upload && !args.skip_setup { results.results.upload = Some(upload::upload_with_config(&args, &config, stopped.clone()).await?); } if !args.skip_wait_index && !args.skip_setup { - results.results.index = - Some(wait_and_report_index(&args, stopped.clone(), baseline).await?); + results.results.index = Some(run_wait_index(&args, stopped.clone()).await?); } results.write_if_requested(&args) @@ -242,16 +133,12 @@ async fn run_benchmark(args: Args, stopped: Arc) -> Result<()> { collection::recreate_collection(&args, stopped.clone()).await?; } - // Indexing can start during upload, so the baseline predates it. - let baseline = optimization_baseline(&args).await; - if !args.skip_upload && !args.skip_setup { results.results.upload = Some(upload::upload_data(&args, stopped.clone()).await?); } if !args.skip_wait_index && !args.skip_setup { - results.results.index = - Some(wait_and_report_index(&args, stopped.clone(), baseline).await?); + results.results.index = Some(run_wait_index(&args, stopped.clone()).await?); } if args.search || args.search_quality { diff --git a/src/optimizations.rs b/src/optimizations.rs deleted file mode 100644 index 87f22e3..0000000 --- a/src/optimizations.rs +++ /dev/null @@ -1,488 +0,0 @@ -//! Per-stage indexing timings from `GET /collections/{c}/optimizations`. -//! -//! This is the one thing BFB cannot get over gRPC — the public API exposes it -//! only over REST (the gRPC `GetShardOptimizations` is an internal, node-to-node -//! RPC that ships this same JSON as opaque bytes). So we speak HTTP here, on the -//! REST port, reusing the `ureq` client already pulled in for dataset downloads. -//! -//! The server keeps the last 16 completed optimizations in a ring buffer, so -//! this must be read *after* indexing finishes but before too many further -//! optimizations run. - -use std::collections::{BTreeMap, HashSet}; -use std::time::Duration; - -use anyhow::{Context, Result}; -use serde::{Deserialize, Serialize}; - -/// Server default for `?completed_limit=`; also the depth of the server's -/// completed-optimization ring buffer. -const DEFAULT_COMPLETED_LIMIT: usize = 16; - -/// Bound the REST call when `--timeout` is unset; `ureq` applies none by default. -const DEFAULT_HTTP_TIMEOUT: Duration = Duration::from_secs(30); - -// ------------------------- wire types (server shape) ---------------------- - -/// The REST envelope: `{result, status, time}`. -#[derive(Debug, Deserialize)] -struct RestEnvelope { - result: T, -} - -#[derive(Debug, Default, Deserialize)] -pub struct OptimizationsResponse { - #[serde(default)] - pub running: Vec, - #[serde(default)] - pub completed: Option>, -} - -#[derive(Debug, Deserialize)] -pub struct Optimization { - /// Identifies this optimization run. Used to tell ours apart from those - /// that had already completed before the phase started. - pub uuid: String, - /// Name of the optimizer that ran (e.g. `indexing`, `merge`). - pub optimizer: String, - /// `"done"`, or a single-key object for `cancelled` / `error` (each carrying - /// a reason string), or `"optimizing"` while still running. - pub status: serde_json::Value, - pub progress: ProgressTree, -} - -impl Optimization { - /// Only `done` optimizations have trustworthy stage timings; a cancelled or - /// errored one stopped partway through its tree. - fn is_done(&self) -> bool { - self.status.as_str() == Some("done") - } -} - -/// A recursive stage tree. Each node carries its own wall-clock duration. -#[derive(Debug, Deserialize)] -pub struct ProgressTree { - pub name: String, - #[serde(default)] - pub duration_sec: Option, - #[serde(default)] - pub children: Vec, -} - -// ------------------------- reported types --------------------------------- - -/// Per-stage aggregate across every completed optimization. -#[derive(Debug, Clone, Default, Serialize, Deserialize, PartialEq)] -pub struct StageAggregate { - /// How many optimizations reported this stage. - pub count: usize, - pub total_secs: f64, - pub max_secs: f64, -} - -/// Per-stage breakdown of the optimizations the server ran, attached to the -/// index phase. -#[derive(Debug, Clone, Default, Serialize, Deserialize, PartialEq)] -pub struct OptimizationsReport { - /// Number of `done` optimizations aggregated below. - pub completed: usize, - /// Optimizations that had already completed before this phase started, and - /// so were excluded. See [`Baseline`]. - pub pre_existing_skipped: usize, - /// The server returned a full page of completed optimizations, so older - /// ones may have been evicted from its ring buffer before we read it — - /// meaning `stages` can undercount. Run fewer optimizations per phase. - pub possibly_truncated: bool, - /// Completed-but-not-`done` entries (cancelled or errored), whose partial - /// stage timings are excluded from `stages`. - pub unfinished: usize, - /// Number of optimizations still running when we read the endpoint. - pub running: usize, - /// Wall-clock seconds summed over every completed optimization's root node. - pub total_secs: f64, - /// Which optimizers ran, e.g. `indexing` → 5, `merge` → 1. - pub optimizers: BTreeMap, - /// Stage path → aggregate, e.g. `vector_index/main_graph`, `quantization`, - /// `payload_index/keyword:color`. - pub stages: BTreeMap, -} - -impl OptimizationsReport { - fn from_response(response: &OptimizationsResponse, baseline: &Baseline) -> Self { - let completed = response.completed.as_deref().unwrap_or_default(); - - let mut report = OptimizationsReport { - running: response.running.len(), - // The server caps the page, and its own buffer, at this many. - possibly_truncated: completed.len() >= DEFAULT_COMPLETED_LIMIT, - ..Default::default() - }; - - for optimization in completed { - // Optimizations that had already finished before the phase began - // belong to an earlier phase, not this one. - if baseline.contains(&optimization.uuid) { - report.pre_existing_skipped += 1; - continue; - } - if !optimization.is_done() { - report.unfinished += 1; - continue; - } - report.completed += 1; - *report - .optimizers - .entry(optimization.optimizer.clone()) - .or_default() += 1; - report.total_secs += optimization.progress.duration_sec.unwrap_or(0.0); - // The root node ("Segment Optimizing") is the total, already counted - // above; only its descendants are stages. - for child in &optimization.progress.children { - report.collect_stage(child, ""); - } - } - - report - } - - /// Fold `node` and its descendants into `stages`, keyed by path. - /// - /// Keys are paths rather than bare names because a name can repeat under - /// different parents with entirely different meanings: a payload field shows - /// up once under `payload_index` (building the field index) and again under - /// `additional_links` (adding HNSW links for it). Collapsing those into one - /// `keyword:color` bucket would sum two unrelated costs. - /// - /// Under `vector_index` the next level is the *vector name*, which is empty - /// for the unnamed default vector. Such levels are skipped rather than - /// recorded, so a default vector yields `vector_index/main_graph` (not - /// `vector_index//main_graph`) while a named one yields - /// `vector_index/image/main_graph`. - fn collect_stage(&mut self, node: &ProgressTree, parent_path: &str) { - let path = if node.name.is_empty() { - parent_path.to_string() - } else if parent_path.is_empty() { - node.name.clone() - } else { - format!("{parent_path}/{}", node.name) - }; - - if !node.name.is_empty() { - let secs = node.duration_sec.unwrap_or(0.0); - let stage = self.stages.entry(path.clone()).or_default(); - stage.count += 1; - stage.total_secs += secs; - stage.max_secs = stage.max_secs.max(secs); - } - - for child in &node.children { - self.collect_stage(child, &path); - } - } - - /// Print the stages slowest-first. - pub fn print(&self) { - if self.completed == 0 { - println!("No completed optimizations reported by the server"); - return; - } - println!( - "--- Optimization stages ({} completed, {:.3} s total) ---", - self.completed, self.total_secs - ); - if self.unfinished > 0 { - println!( - "({} cancelled/errored optimization(s) excluded)", - self.unfinished - ); - } - if self.possibly_truncated { - println!( - "(warning: the server's completed-optimization buffer was full; \ - older stages may be missing)" - ); - } - let mut stages: Vec<_> = self.stages.iter().collect(); - stages.sort_unstable_by(|(_, a), (_, b)| b.total_secs.total_cmp(&a.total_secs)); - for (name, stage) in stages { - println!( - "{name}: {:.6} s total over {} run(s), max {:.6} s", - stage.total_secs, stage.count, stage.max_secs - ); - } - } -} - -// ------------------------- fetching --------------------------------------- - -/// Derive the REST base URL from a gRPC `--uri`. -/// -/// Qdrant serves gRPC on 6334 and REST on 6333, so this is a one-port shift. -/// A `--uri` on any other port is returned unchanged. -pub fn rest_url_from_grpc(uri: &str) -> String { - let trimmed = uri.trim_end_matches('/'); - match trimmed.rsplit_once(":6334") { - Some((head, "")) => format!("{head}:6333"), - _ => trimmed.to_string(), - } -} - -/// The optimizations that had already completed when a phase began. -/// -/// The server reports a rolling window of completed optimizations, not the ones -/// belonging to any particular phase. Without a baseline, `bfb wait-index` would -/// happily attribute a preceding `bfb create-index`'s work to indexing. -#[derive(Debug, Default, Clone)] -pub struct Baseline(HashSet); - -impl Baseline { - fn contains(&self, uuid: &str) -> bool { - self.0.contains(uuid) - } - - /// An empty baseline: every completed optimization counts. - pub fn none() -> Self { - Baseline::default() - } -} - -/// Record which optimizations have already completed, to exclude them later. -pub fn baseline( - rest_url: &str, - collection: &str, - api_key: Option<&str>, - timeout: Option, -) -> Result { - let response = get(rest_url, collection, api_key, timeout)?; - Ok(Baseline( - response - .completed - .unwrap_or_default() - .into_iter() - .map(|optimization| optimization.uuid) - .collect(), - )) -} - -/// Read completed optimization stages for `collection`, excluding anything -/// already present in `baseline`. -pub fn fetch( - rest_url: &str, - collection: &str, - api_key: Option<&str>, - baseline: &Baseline, - timeout: Option, -) -> Result { - let response = get(rest_url, collection, api_key, timeout)?; - Ok(OptimizationsReport::from_response(&response, baseline)) -} - -fn get( - rest_url: &str, - collection: &str, - api_key: Option<&str>, - timeout: Option, -) -> Result { - let url = format!( - "{}/collections/{collection}/optimizations?with=completed&completed_limit={DEFAULT_COMPLETED_LIMIT}", - rest_url.trim_end_matches('/') - ); - - let agent = ureq::Agent::new_with_config( - ureq::config::Config::builder() - .timeout_global(Some(timeout.unwrap_or(DEFAULT_HTTP_TIMEOUT))) - .build(), - ); - let mut request = agent.get(&url); - if let Some(key) = api_key { - request = request.header("api-key", key); - } - - let body = request - .call() - .with_context(|| format!("failed to GET {url}"))? - .into_body() - .read_to_string() - .with_context(|| format!("failed to read response from {url}"))?; - - let envelope: RestEnvelope = serde_json::from_str(&body) - .with_context(|| format!("failed to parse optimizations response from {url}"))?; - Ok(envelope.result) -} - -#[cfg(test)] -mod tests { - use super::*; - - fn parse_fixture() -> OptimizationsResponse { - let text = include_str!("../tests/fixtures/optimizations_completed.json"); - let envelope: RestEnvelope = serde_json::from_str(text).unwrap(); - envelope.result - } - - #[test] - fn parses_real_server_response() { - let response = parse_fixture(); - assert_eq!(response.completed.as_ref().unwrap().len(), 2); - assert!(response.running.is_empty()); - assert_eq!( - response.completed.as_ref().unwrap()[0].optimizer, - "indexing" - ); - } - - #[test] - fn aggregates_the_stages_the_benchmarks_ask_for() { - let report = OptimizationsReport::from_response(&parse_fixture(), &Baseline::none()); - - assert_eq!(report.completed, 2); - assert_eq!(report.running, 0); - - // The stages the indexing benchmarks are written against. - for stage in ["vector_index/main_graph", "quantization", "payload_index"] { - assert!( - report.stages.contains_key(stage), - "missing stage {stage}; got {:?}", - report.stages.keys().collect::>() - ); - } - - // The fixture uses the unnamed default vector, so the vector-name level - // under `vector_index` is empty and must collapse out of the path. - let main_graph = &report.stages["vector_index/main_graph"]; - assert_eq!(main_graph.count, 2, "one per completed optimization"); - assert!(main_graph.total_secs > 0.0); - assert!(main_graph.max_secs <= main_graph.total_secs); - - // The unnamed wrapper node must not become a stage, nor leave an empty - // path segment behind. - assert!(!report.stages.contains_key("")); - assert!(report.stages.keys().all(|k| !k.contains("//"))); - - // The same field name appears under two different parents with two - // different meanings; they must not be summed into one bucket. - assert!(report.stages.contains_key("payload_index/keyword:a")); - assert!( - report - .stages - .contains_key("vector_index/additional_links/keyword:a") - ); - } - - #[test] - fn total_secs_sums_root_nodes_only() { - let report = OptimizationsReport::from_response(&parse_fixture(), &Baseline::none()); - // Root duration exceeds any single child stage, and is not the sum of - // all stages (which would double-count nested nodes). - assert!(report.total_secs > report.stages["vector_index/main_graph"].total_secs); - let stage_sum: f64 = report.stages.values().map(|s| s.total_secs).sum(); - assert!( - stage_sum > report.total_secs, - "nested stages overlap in time" - ); - } - - #[test] - fn records_which_optimizers_ran() { - let report = OptimizationsReport::from_response(&parse_fixture(), &Baseline::none()); - assert_eq!(report.optimizers.get("indexing"), Some(&2)); - assert_eq!(report.unfinished, 0); - } - - #[test] - fn cancelled_and_errored_optimizations_are_excluded() { - // A cancelled/errored entry stopped partway through its stage tree, so - // folding its durations in would understate the real stage cost. - let json = r#"{ - "running": [], - "completed": [ - {"uuid": "a", "optimizer": "indexing", "status": "done", - "progress": {"name": "Segment Optimizing", "duration_sec": 2.0, - "children": [{"name": "main_graph", "duration_sec": 1.5}]}}, - {"uuid": "b", "optimizer": "indexing", "status": {"cancelled": "shutdown"}, - "progress": {"name": "Segment Optimizing", "duration_sec": 0.3, - "children": [{"name": "main_graph", "duration_sec": 0.1}]}}, - {"uuid": "c", "optimizer": "merge", "status": {"error": "disk full"}, - "progress": {"name": "Segment Optimizing", "duration_sec": 0.2, "children": []}} - ] - }"#; - let response: OptimizationsResponse = serde_json::from_str(json).unwrap(); - let report = OptimizationsReport::from_response(&response, &Baseline::none()); - - assert_eq!(report.completed, 1); - assert_eq!(report.unfinished, 2); - assert_eq!(report.total_secs, 2.0, "only the `done` root counts"); - assert_eq!(report.stages["main_graph"].total_secs, 1.5); - assert_eq!(report.stages["main_graph"].count, 1); - assert_eq!(report.optimizers.get("merge"), None); - } - - #[test] - fn optimizations_completed_before_the_phase_are_excluded() { - // A `bfb create-index` run leaves completed optimizations behind; the - // `bfb wait-index` that follows must not claim them as indexing work. - let response = parse_fixture(); - let first = response.completed.as_ref().unwrap()[0].uuid.clone(); - let baseline = Baseline(HashSet::from([first])); - - let report = OptimizationsReport::from_response(&response, &baseline); - - assert_eq!(report.completed, 1, "only the new optimization counts"); - assert_eq!(report.pre_existing_skipped, 1); - assert_eq!(report.stages["vector_index/main_graph"].count, 1); - - // Without the baseline, both would be attributed to this phase. - let unfiltered = OptimizationsReport::from_response(&response, &Baseline::none()); - assert_eq!(unfiltered.completed, 2); - assert!(unfiltered.total_secs > report.total_secs); - } - - #[test] - fn a_full_page_of_completed_optimizations_flags_truncation() { - // The server keeps only the last 16; a full page means older stages may - // already be gone, so `stages` can undercount. - let one = r#"{"uuid": "%", "optimizer": "indexing", "status": "done", - "progress": {"name": "Segment Optimizing", "duration_sec": 1.0, "children": []}}"#; - let entries: Vec = (0..16).map(|i| one.replace('%', &i.to_string())).collect(); - let json = format!(r#"{{"running": [], "completed": [{}]}}"#, entries.join(",")); - - let response: OptimizationsResponse = serde_json::from_str(&json).unwrap(); - let report = OptimizationsReport::from_response(&response, &Baseline::none()); - - assert_eq!(report.completed, 16); - assert!(report.possibly_truncated); - } - - #[test] - fn a_partial_page_is_not_flagged_as_truncated() { - let report = OptimizationsReport::from_response(&parse_fixture(), &Baseline::none()); - assert!(!report.possibly_truncated); - } - - #[test] - fn empty_response_reports_nothing() { - let report = OptimizationsReport::from_response( - &OptimizationsResponse::default(), - &Baseline::none(), - ); - assert_eq!(report.completed, 0); - assert!(report.stages.is_empty()); - assert_eq!(report.total_secs, 0.0); - } - - #[test] - fn derives_the_rest_url_from_the_grpc_uri() { - assert_eq!( - rest_url_from_grpc("http://localhost:6334"), - "http://localhost:6334".replace("6334", "6333") - ); - assert_eq!( - rest_url_from_grpc("https://xyz.cloud.qdrant.io:6334/"), - "https://xyz.cloud.qdrant.io:6333" - ); - // A non-default port is left alone rather than silently rewritten. - assert_eq!(rest_url_from_grpc("http://host:7000"), "http://host:7000"); - // A port that merely *contains* 6334 must not match. - assert_eq!(rest_url_from_grpc("http://host:63340"), "http://host:63340"); - } -} diff --git a/src/results.rs b/src/results.rs index 1b05519..b64fff7 100644 --- a/src/results.rs +++ b/src/results.rs @@ -11,7 +11,6 @@ use anyhow::{Context, Result}; use serde::{Deserialize, Serialize}; use crate::args::Args; -use crate::optimizations::OptimizationsReport; use crate::processor::Timing; /// Top-level document: `{ config: {...}, results: {...} }`. @@ -88,13 +87,10 @@ pub struct UploadPhase { pub points_per_sec: f64, } -/// Time spent waiting for the collection to report a green index, plus the -/// server's own per-stage breakdown of the optimizations it ran. -#[derive(Debug, Clone, Default, Serialize, Deserialize, PartialEq)] +/// M2: time spent waiting for the collection to report a green index. +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)] pub struct IndexPhase { pub wait_secs: f64, - #[serde(skip_serializing_if = "Option::is_none")] - pub optimizations: Option, } /// A search or scroll phase: latency/throughput series plus their summaries. @@ -289,10 +285,7 @@ mod tests { #[test] fn skipped_phases_are_omitted_from_json() { let results = PhaseResults { - index: Some(IndexPhase { - wait_secs: 1.5, - ..Default::default() - }), + index: Some(IndexPhase { wait_secs: 1.5 }), ..Default::default() }; let json = serde_json::to_string(&results).unwrap(); @@ -338,10 +331,7 @@ mod tests { fn document_roundtrips() { let mut doc = doc(PhaseResults { upload: Some(UploadPhase::new(1.0, 100)), - index: Some(IndexPhase { - wait_secs: 0.5, - ..Default::default() - }), + index: Some(IndexPhase { wait_secs: 0.5 }), search: Some(phase(1.0)), scroll: None, }); @@ -397,10 +387,7 @@ mod tests { fn upload_only_run_emits_no_legacy_fields() { let mut doc = doc(PhaseResults { upload: Some(UploadPhase::new(1.0, 100)), - index: Some(IndexPhase { - wait_secs: 0.5, - ..Default::default() - }), + index: Some(IndexPhase { wait_secs: 0.5 }), ..Default::default() }); doc.populate_legacy_fields(); diff --git a/tests/fixtures/optimizations_completed.json b/tests/fixtures/optimizations_completed.json deleted file mode 100644 index 1b30a08..0000000 --- a/tests/fixtures/optimizations_completed.json +++ /dev/null @@ -1,235 +0,0 @@ -{ - "result": { - "summary": { - "queued_optimizations": 0, - "queued_segments": 0, - "queued_points": 0, - "idle_segments": 4 - }, - "running": [], - "completed": [ - { - "uuid": "21a7e712-dd1e-4069-9f06-b292e4997ef7", - "optimizer": "indexing", - "status": "done", - "segments": [ - { - "uuid": "1821de91-a124-4c57-bcd8-cdad6b535194", - "points_count": 15000 - }, - { - "uuid": "42488f00-c7f9-4e20-b59a-7273974397b8", - "points_count": 1012 - } - ], - "progress": { - "name": "Segment Optimizing", - "started_at": "2026-07-10T09:40:21.765178905Z", - "finished_at": "2026-07-10T09:40:23.145000753Z", - "duration_sec": 1.379821628, - "children": [ - { - "name": "copy_data", - "started_at": "2026-07-10T09:40:21.836839467Z", - "finished_at": "2026-07-10T09:40:21.895042118Z", - "duration_sec": 0.058201569 - }, - { - "name": "populate_vector_storages", - "started_at": "2026-07-10T09:40:21.895044172Z", - "finished_at": "2026-07-10T09:40:21.946601106Z", - "duration_sec": 0.051556734 - }, - { - "name": "wait_cpu_permit", - "started_at": "2026-07-10T09:40:21.946602378Z", - "finished_at": "2026-07-10T09:40:22.160921861Z", - "duration_sec": 0.214319082 - }, - { - "name": "quantization", - "started_at": "2026-07-10T09:40:22.192142510Z", - "finished_at": "2026-07-10T09:40:22.192145886Z", - "duration_sec": 2.795e-06 - }, - { - "name": "payload_index", - "started_at": "2026-07-10T09:40:22.231933459Z", - "finished_at": "2026-07-10T09:40:22.276830503Z", - "duration_sec": 0.044896443, - "children": [ - { - "name": "keyword:a", - "started_at": "2026-07-10T09:40:22.235260060Z", - "finished_at": "2026-07-10T09:40:22.276829041Z", - "duration_sec": 0.041560936 - } - ] - }, - { - "name": "vector_index", - "started_at": "2026-07-10T09:40:22.282012799Z", - "finished_at": "2026-07-10T09:40:23.121430494Z", - "duration_sec": 0.839417014, - "children": [ - { - "name": "", - "started_at": "2026-07-10T09:40:22.282019201Z", - "finished_at": "2026-07-10T09:40:23.121374811Z", - "duration_sec": 0.83935554, - "children": [ - { - "name": "migrate", - "started_at": "2026-07-10T09:40:22.284579186Z", - "finished_at": "2026-07-10T09:40:22.285133379Z", - "duration_sec": 0.000553562 - }, - { - "name": "main_graph", - "started_at": "2026-07-10T09:40:22.285133650Z", - "finished_at": "2026-07-10T09:40:23.082625933Z", - "duration_sec": 0.797491983, - "done": 15000, - "total": 15000 - }, - { - "name": "additional_links", - "started_at": "2026-07-10T09:40:23.082628588Z", - "finished_at": "2026-07-10T09:40:23.095550070Z", - "duration_sec": 0.012921402, - "children": [ - { - "name": "keyword:a", - "started_at": "2026-07-10T09:40:23.095520355Z", - "finished_at": "2026-07-10T09:40:23.095546554Z", - "duration_sec": 2.5608e-05, - "done": 0 - } - ] - } - ] - } - ] - }, - { - "name": "sparse_vector_index", - "started_at": "2026-07-10T09:40:23.121430985Z", - "finished_at": "2026-07-10T09:40:23.121431566Z", - "duration_sec": 5.01e-07 - } - ] - } - }, - { - "uuid": "93cc97c6-ace5-4d4d-ba32-c2144e2b5f7a", - "optimizer": "indexing", - "status": "done", - "segments": [ - { - "uuid": "c6d47929-9308-4a1a-b575-c7752c400630", - "points_count": 3500 - } - ], - "progress": { - "name": "Segment Optimizing", - "started_at": "2026-07-10T09:40:20.800774711Z", - "finished_at": "2026-07-10T09:40:22.123890285Z", - "duration_sec": 1.323115413, - "children": [ - { - "name": "copy_data", - "started_at": "2026-07-10T09:40:20.870915263Z", - "finished_at": "2026-07-10T09:40:20.888020114Z", - "duration_sec": 0.017103609 - }, - { - "name": "populate_vector_storages", - "started_at": "2026-07-10T09:40:20.888022348Z", - "finished_at": "2026-07-10T09:40:20.943550449Z", - "duration_sec": 0.055527811 - }, - { - "name": "wait_cpu_permit", - "started_at": "2026-07-10T09:40:20.943551912Z", - "finished_at": "2026-07-10T09:40:21.765102494Z", - "duration_sec": 0.821550302 - }, - { - "name": "quantization", - "started_at": "2026-07-10T09:40:21.793020004Z", - "finished_at": "2026-07-10T09:40:21.793023851Z", - "duration_sec": 2.895e-06 - }, - { - "name": "payload_index", - "started_at": "2026-07-10T09:40:21.815024641Z", - "finished_at": "2026-07-10T09:40:21.857925898Z", - "duration_sec": 0.042900546, - "children": [ - { - "name": "keyword:a", - "started_at": "2026-07-10T09:40:21.822396678Z", - "finished_at": "2026-07-10T09:40:21.857924255Z", - "duration_sec": 0.035526415 - } - ] - }, - { - "name": "vector_index", - "started_at": "2026-07-10T09:40:21.860372765Z", - "finished_at": "2026-07-10T09:40:22.099029295Z", - "duration_sec": 0.238655959, - "children": [ - { - "name": "", - "started_at": "2026-07-10T09:40:21.860377454Z", - "finished_at": "2026-07-10T09:40:22.098965377Z", - "duration_sec": 0.238587823, - "children": [ - { - "name": "migrate", - "started_at": "2026-07-10T09:40:21.861825747Z", - "finished_at": "2026-07-10T09:40:21.861825747Z", - "duration_sec": 0.0 - }, - { - "name": "main_graph", - "started_at": "2026-07-10T09:40:21.861826749Z", - "finished_at": "2026-07-10T09:40:22.080672230Z", - "duration_sec": 0.218845121, - "done": 5500, - "total": 5500 - }, - { - "name": "additional_links", - "started_at": "2026-07-10T09:40:22.080674735Z", - "finished_at": "2026-07-10T09:40:22.083122724Z", - "duration_sec": 0.002447788, - "children": [ - { - "name": "keyword:a", - "started_at": "2026-07-10T09:40:22.083107005Z", - "finished_at": "2026-07-10T09:40:22.083120840Z", - "duration_sec": 1.3505e-05, - "done": 0 - } - ] - } - ] - } - ] - }, - { - "name": "sparse_vector_index", - "started_at": "2026-07-10T09:40:22.099030157Z", - "finished_at": "2026-07-10T09:40:22.099030648Z", - "duration_sec": 3.81e-07 - } - ] - } - } - ] - }, - "status": "ok", - "time": 5.2477e-05 -} \ No newline at end of file