From 7cddfea41bc01c3364de1a6e398ee24e3c4b7c77 Mon Sep 17 00:00:00 2001 From: tellet-q Date: Mon, 13 Jul 2026 14:42:17 +0200 Subject: [PATCH] feat: `bfb update-collection`; patch settings and measure the indexing it triggers Co-Authored-By: Claude Opus 4.8 (1M context) --- README.md | 36 ++++++++++++ examples/update-config.yaml | 28 +++++++++ src/args/mod.rs | 14 +++++ src/collection/mod.rs | 98 +++++++++++++++++++++++++++++++- src/config/mod.rs | 1 + src/config/update.rs | 109 ++++++++++++++++++++++++++++++++++++ src/main.rs | 30 ++++++++++ src/results.rs | 14 +++++ 8 files changed, 327 insertions(+), 3 deletions(-) create mode 100644 examples/update-config.yaml create mode 100644 src/config/update.rs diff --git a/README.md b/README.md index 428ed63..e0a3139 100644 --- a/README.md +++ b/README.md @@ -209,6 +209,42 @@ bfb drop-field-index --file config.yaml --field color bfb drop-field-index --file config.yaml # or every indexed field in the config ``` +### `update-collection` — patch settings, and measure what that triggers + +Only what the YAML declares is sent, so the rest of the collection config is +left alone — undeclared is not the same as `false` or `0`. Lowering +`indexing_threshold` starts indexing; changing `max_segment_size` starts a +merge. The patch request returns long before either finishes, so this waits for +the work it triggered and reports it, exactly as `bfb upload` does: + +```bash +bfb upload --file config.yaml -n 1M --skip-wait-index # indexing_threshold: 1000000000 +bfb update-collection --file examples/update-config.yaml --json index.json +``` + +```yaml +collection: + name: benchmark + optimizers: + indexing_threshold: 1 # start indexing now + # max_segment_size: 200000 # changing this triggers a merge + # hnsw: + # m: 32 + # payload_m: 16 +``` + +``` +Updated collection: optimizers.indexing_threshold=1 +Server reported changes: true +Index ready in 20.255 seconds +``` + +The patch itself lands under `results.update_collection` (including the server's +`applied` flag); the indexing it triggered lands under `results.index`, the same +place `bfb upload` puts it. `--skip-wait-index` fires the patch and returns +without waiting. A patch that declares no changes is an error rather than a +silent no-op. See [`examples/update-config.yaml`](examples/update-config.yaml). + ### `schema` — print the upload-config file schema Print an annotated YAML reference enumerating every option accepted by an diff --git a/examples/update-config.yaml b/examples/update-config.yaml new file mode 100644 index 0000000..fd9add2 --- /dev/null +++ b/examples/update-config.yaml @@ -0,0 +1,28 @@ +# Example bfb update-collection config. +# +# bfb update-collection --file examples/update-config.yaml +# +# Only what this file declares is sent; everything else on the collection is +# left alone. Undeclared is not the same as `false` or `0`. +# +# The canonical use is measuring indexing on its own: upload with indexing +# effectively disabled, then turn it on and time it. `update-collection` waits +# for the indexing its patch triggers and reports it under `index` in `--json`. +# +# bfb upload --file upload-config.yaml -n 1M --skip-wait-index +# bfb update-collection --file update-config.yaml --json index.json + +collection: + name: benchmark + + optimizers: + indexing_threshold: 1 # start indexing now + # max_segment_size: 200000 # changing this triggers a merge + # memmap_threshold: 100000 + # default_segment_number: 2 + + # hnsw: + # m: 32 + # payload_m: 16 + # ef_construct: 200 + # on_disk: true diff --git a/src/args/mod.rs b/src/args/mod.rs index 367220f..8f067d5 100644 --- a/src/args/mod.rs +++ b/src/args/mod.rs @@ -460,6 +460,13 @@ pub enum Command { /// the same data without re-uploading. DropFieldIndex(DropFieldIndexArgs), + /// Patch optimizer/HNSW settings on a live collection, and measure the + /// indexing it triggers. + /// + /// Only what the YAML declares is sent. Lower `indexing_threshold` to start + /// indexing on demand, or change `max_segment_size` to trigger a merge. + UpdateCollection(UpdateCollectionArgs), + /// Print the upload-config file schema: every option, its type, default, /// and allowed values, as an annotated YAML reference. Schema, @@ -498,6 +505,13 @@ pub struct SearchArgs { pub file: String, } +#[derive(clap::Args, Debug, Clone)] +pub struct UpdateCollectionArgs { + /// Path to the YAML update config file + #[clap(long)] + pub file: String, +} + #[derive(clap::Args, Debug, Clone)] pub struct ScrollArgs { /// Path to the YAML scroll config file diff --git a/src/collection/mod.rs b/src/collection/mod.rs index 3fc6c3e..09dfd56 100644 --- a/src/collection/mod.rs +++ b/src/collection/mod.rs @@ -1,5 +1,6 @@ //! Collection lifecycle: (re)creation — from CLI flags or a YAML config — -//! waiting for indexing, and building payload field indices. +//! waiting for indexing, building payload field indices, and patching settings +//! on a live collection. mod from_args; mod from_config; @@ -14,13 +15,15 @@ use std::time::{Duration, Instant}; use anyhow::{Result, bail}; use qdrant_client::qdrant::{ CollectionInfo, CollectionStatus, CreateFieldIndexCollection, - CreateFieldIndexCollectionBuilder, DeleteFieldIndexCollectionBuilder, + CreateFieldIndexCollectionBuilder, DeleteFieldIndexCollectionBuilder, HnswConfigDiffBuilder, + OptimizersConfigDiffBuilder, UpdateCollectionBuilder, }; use tokio::time::sleep; use crate::args::Args; use crate::client::random_client; -use crate::results::{CreateFieldIndexPhase, FieldIndexTiming}; +use crate::config::update::UpdateConfig; +use crate::results::{CreateFieldIndexPhase, FieldIndexTiming, UpdateCollectionPhase}; /// One field index to create: the built request plus the labels used to report it. pub struct FieldIndexSpec { @@ -155,3 +158,92 @@ fn report_index_progress(info: &CollectionInfo, elapsed: Duration) { elapsed.as_secs_f64() ); } + +/// Patch collection settings on a live collection. +/// +/// Only what the config declares is sent, so this can lower +/// `indexing_threshold` to start indexing on demand, or change +/// `max_segment_size` to trigger a merge, without disturbing anything else. +pub async fn update_collection( + args: &Args, + config: &UpdateConfig, +) -> Result { + let client = random_client(args)?; + let mut changed = Vec::new(); + + let mut optimizers = OptimizersConfigDiffBuilder::default(); + let mut optimizers_changed = false; + if let Some(patch) = &config.collection.optimizers { + if let Some(threshold) = patch.indexing_threshold { + optimizers = optimizers.indexing_threshold(threshold); + optimizers_changed = true; + changed.push(format!("optimizers.indexing_threshold={threshold}")); + } + if let Some(size) = patch.max_segment_size { + optimizers = optimizers.max_segment_size(size); + optimizers_changed = true; + changed.push(format!("optimizers.max_segment_size={size}")); + } + if let Some(threshold) = patch.memmap_threshold { + optimizers = optimizers.memmap_threshold(threshold); + optimizers_changed = true; + changed.push(format!("optimizers.memmap_threshold={threshold}")); + } + if let Some(segments) = patch.default_segment_number { + optimizers = optimizers.default_segment_number(segments); + optimizers_changed = true; + changed.push(format!("optimizers.default_segment_number={segments}")); + } + } + + let mut hnsw = HnswConfigDiffBuilder::default(); + let mut hnsw_changed = false; + if let Some(patch) = &config.collection.hnsw { + if let Some(m) = patch.m { + hnsw = hnsw.m(m); + hnsw_changed = true; + changed.push(format!("hnsw.m={m}")); + } + if let Some(ef) = patch.ef_construct { + hnsw = hnsw.ef_construct(ef); + hnsw_changed = true; + changed.push(format!("hnsw.ef_construct={ef}")); + } + if let Some(payload_m) = patch.payload_m { + hnsw = hnsw.payload_m(payload_m); + hnsw_changed = true; + changed.push(format!("hnsw.payload_m={payload_m}")); + } + if let Some(on_disk) = patch.on_disk { + hnsw = hnsw.on_disk(on_disk); + hnsw_changed = true; + changed.push(format!("hnsw.on_disk={on_disk}")); + } + } + + // `validate()` already rejected an empty patch, so `changed` is non-empty. + // Each section is only attached when it actually changed: an empty diff would + // be a no-op at best, and could still nudge the server to re-evaluate work + // the patch never asked for. + let mut builder = UpdateCollectionBuilder::new(args.collection_name.clone()); + if optimizers_changed { + builder = builder.optimizers_config(optimizers); + } + if hnsw_changed { + builder = builder.hnsw_config(hnsw); + } + + let started = Instant::now(); + let response = client.update_collection(builder).await?; + let duration_secs = started.elapsed().as_secs_f64(); + + println!("Updated collection: {}", changed.join(", ")); + println!("Server reported changes: {}", response.result); + + Ok(UpdateCollectionPhase { + duration_secs, + server_secs: response.time, + changed, + applied: response.result, + }) +} diff --git a/src/config/mod.rs b/src/config/mod.rs index 9009490..b309ca4 100644 --- a/src/config/mod.rs +++ b/src/config/mod.rs @@ -11,6 +11,7 @@ pub mod payload; pub mod schema; pub mod scroll; pub mod search; +pub mod update; pub mod vector; pub use collection::{CollectionConfig, IdType, QuantKind}; diff --git a/src/config/update.rs b/src/config/update.rs new file mode 100644 index 0000000..8d7459e --- /dev/null +++ b/src/config/update.rs @@ -0,0 +1,109 @@ +//! YAML patch configuration for `bfb update-collection --file config.yaml`. +//! +//! Every field is optional: only what the file declares is sent, so the rest of +//! the collection config is left alone. + +use anyhow::{Context, Result, bail}; +use serde::{Deserialize, Serialize}; + +use crate::config::default_collection_name; + +/// Top-level document: `{ collection: { name, hnsw, optimizers } }`. +#[derive(Debug, Clone, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct UpdateConfig { + pub collection: UpdateCollectionConfig, +} + +#[derive(Debug, Clone, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct UpdateCollectionConfig { + #[serde(default = "default_collection_name")] + pub name: String, + #[serde(default)] + pub hnsw: Option, + #[serde(default)] + pub optimizers: Option, +} + +/// HNSW settings to change. `on_disk` is an `Option` rather than a plain +/// `bool` so that "not declared" stays distinct from "set to false". +#[derive(Debug, Clone, Default, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct HnswPatch { + pub m: Option, + pub payload_m: Option, + pub ef_construct: Option, + pub on_disk: Option, +} + +#[derive(Debug, Clone, Default, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct OptimizersPatch { + pub default_segment_number: Option, + pub indexing_threshold: Option, + pub memmap_threshold: Option, + pub max_segment_size: Option, +} + +impl UpdateConfig { + pub fn validate(&self) -> Result<()> { + let hnsw = self.collection.hnsw.clone().unwrap_or_default(); + let optimizers = self.collection.optimizers.clone().unwrap_or_default(); + + let declares_nothing = hnsw.m.is_none() + && hnsw.payload_m.is_none() + && hnsw.ef_construct.is_none() + && hnsw.on_disk.is_none() + && optimizers.default_segment_number.is_none() + && optimizers.indexing_threshold.is_none() + && optimizers.memmap_threshold.is_none() + && optimizers.max_segment_size.is_none(); + + if declares_nothing { + bail!( + "update config declares no changes: set at least one field under \ + `hnsw` or `optimizers`" + ); + } + Ok(()) + } +} + +pub fn load(path: &str) -> Result { + let text = std::fs::read_to_string(path) + .with_context(|| format!("failed to read update config file {path}"))?; + let config: UpdateConfig = serde_yaml::from_str(&text) + .with_context(|| format!("failed to parse update config file {path}"))?; + config.validate()?; + Ok(config) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn parses_a_partial_patch() { + let config: UpdateConfig = serde_yaml::from_str( + "collection:\n name: x\n optimizers:\n indexing_threshold: 1\n", + ) + .unwrap(); + config.validate().unwrap(); + + assert_eq!(config.collection.name, "x"); + assert_eq!( + config.collection.optimizers.unwrap().indexing_threshold, + Some(1) + ); + // Undeclared sections stay absent rather than defaulting to something + // that would be sent to the server. + assert!(config.collection.hnsw.is_none()); + } + + #[test] + fn rejects_a_patch_that_changes_nothing() { + let config: UpdateConfig = serde_yaml::from_str("collection:\n name: x\n").unwrap(); + assert!(config.validate().is_err()); + } +} diff --git a/src/main.rs b/src/main.rs index 5aff127..e42c2e1 100644 --- a/src/main.rs +++ b/src/main.rs @@ -161,6 +161,33 @@ async fn run_create_field_index( results.write_if_requested(&args) } +/// `bfb update-collection --file config.yaml`: patch settings on a live +/// collection, then measure the indexing the patch triggers. +async fn run_update_collection( + args: Args, + update_args: args::UpdateCollectionArgs, + stopped: Arc, +) -> Result<()> { + let config = config::update::load(&update_args.file)?; + + let mut args = args; + args.collection_name = config.collection.name.clone(); + + // Phase-only: neither uploads nor queries, so no requested point count. + let mut results = BenchmarkResults::new(&args, Some(update_args.file.clone()), None); + results.results.update_collection = Some(collection::update_collection(&args, &config).await?); + + // Lowering `indexing_threshold` starts indexing; changing `max_segment_size` + // starts a merge. The patch request returns before either finishes, so the + // wait is what turns "applied: true" into a measurement. + if !args.skip_wait_index { + results.results.index = Some(run_wait_index(&args, stopped).await?); + results.results.memory = fetch_memory(&args).await; + } + + results.write_if_requested(&args) +} + /// `bfb drop-field-index`: remove field indices so they can be rebuilt and re-measured. async fn run_drop_field_index(args: Args, drop_args: args::DropFieldIndexArgs) -> Result<()> { let config = config::load(&drop_args.file)?; @@ -261,6 +288,9 @@ async fn run_benchmark(args: Args, stopped: Arc) -> Result<()> { Some(Command::DropFieldIndex(drop_args)) => { return run_drop_field_index(args, drop_args).await; } + Some(Command::UpdateCollection(update_args)) => { + return run_update_collection(args, update_args, stopped).await; + } // `Schema` is handled before the runtime starts; `None` falls through. Some(Command::Schema) | None => {} } diff --git a/src/results.rs b/src/results.rs index 1047f90..63bc88a 100644 --- a/src/results.rs +++ b/src/results.rs @@ -80,6 +80,8 @@ pub struct PhaseResults { #[serde(skip_serializing_if = "Option::is_none")] pub create_field_index: Option, #[serde(skip_serializing_if = "Option::is_none")] + pub update_collection: Option, + #[serde(skip_serializing_if = "Option::is_none")] pub search: Option, #[serde(skip_serializing_if = "Option::is_none")] pub scroll: Option, @@ -108,6 +110,18 @@ pub struct FieldIndexTiming { pub server_secs: f64, } +/// A mid-run `update_collection` patch. It reports the patch request itself; the +/// indexing it triggers is reported under `index`, as for any other trigger. +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)] +pub struct UpdateCollectionPhase { + pub duration_secs: f64, + pub server_secs: f64, + /// Human-readable list of the settings that were sent. + pub changed: Vec, + /// Whether the server reported that the operation made changes. + pub applied: bool, +} + /// M2: upload wall time and throughput. #[derive(Debug, Clone, Serialize, Deserialize, PartialEq)] pub struct UploadPhase {