From 1383e7b692e23dfd8a331301d2ac0d884671ef64 Mon Sep 17 00:00:00 2001 From: Cursor Agent Date: Wed, 2 Sep 2026 11:14:48 +0000 Subject: [PATCH 1/4] Add serverless multi-collection benchmark mode Introduce `bfb serverless {upload,clear,query}` for Qdrant Serverless: lazily create per-tenant collections, route traffic with uniform/zipf distributions, and drive the new QdrantServerless rust-client API. --- Cargo.lock | 5 +- DEVELOPMENT.md | 1 + README.md | 45 +++++++ examples/serverless-upload.yaml | 26 ++++ src/args/mod.rs | 8 +- src/main.rs | 4 + src/serverless/args.rs | 87 ++++++++++++ src/serverless/clear.rs | 40 ++++++ src/serverless/client.rs | 80 +++++++++++ src/serverless/collections.rs | 139 +++++++++++++++++++ src/serverless/convert.rs | 109 +++++++++++++++ src/serverless/distribution.rs | 95 +++++++++++++ src/serverless/mod.rs | 40 ++++++ src/serverless/query.rs | 231 ++++++++++++++++++++++++++++++++ src/serverless/upload.rs | 191 ++++++++++++++++++++++++++ 15 files changed, 1097 insertions(+), 4 deletions(-) create mode 100644 examples/serverless-upload.yaml create mode 100644 src/serverless/args.rs create mode 100644 src/serverless/clear.rs create mode 100644 src/serverless/client.rs create mode 100644 src/serverless/collections.rs create mode 100644 src/serverless/convert.rs create mode 100644 src/serverless/distribution.rs create mode 100644 src/serverless/mod.rs create mode 100644 src/serverless/query.rs create mode 100644 src/serverless/upload.rs diff --git a/Cargo.lock b/Cargo.lock index ff39960..36cf61b 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1472,11 +1472,10 @@ dependencies = [ [[package]] name = "qdrant-client" version = "1.16.1-dev" -source = "git+https://github.com/qdrant/rust-client?branch=dev#738227709dab2a6473f3d60cc845bf2313f9dfea" +source = "git+https://github.com/qdrant/rust-client?branch=dev#b6bacf7cdc50dc0a64ed6ff7e6d159bc186e903a" dependencies = [ "anyhow", "derive_builder", - "futures", "parking_lot", "prost", "prost-types", @@ -1881,7 +1880,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "32497e9a4c7b38532efcdebeef879707aa9f794296a4f0244f6f69e9bc8574bd" dependencies = [ "fastrand", - "getrandom 0.4.2", + "getrandom 0.3.4", "once_cell", "rustix", "windows-sys 0.61.2", diff --git a/DEVELOPMENT.md b/DEVELOPMENT.md index 99e3f57..3edfe2e 100644 --- a/DEVELOPMENT.md +++ b/DEVELOPMENT.md @@ -9,6 +9,7 @@ src/ ├── collection/ # collection (re)creation: from CLI flags / from YAML config ├── generators/ # data generation: points (legacy & config), queries, random primitives ├── search/ # search benchmarking processors (flag-driven & config-driven) +├── serverless/ # multi-collection mode for Qdrant Serverless (upload/clear/query) ├── upload.rs # upload pipeline (parallelism, batching, progress) ├── upsert.rs # upsert request execution + timing ├── scroll.rs # scroll benchmarking processor diff --git a/README.md b/README.md index d8ff8bb..ddecf7d 100644 --- a/README.md +++ b/README.md @@ -366,6 +366,51 @@ 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. +### `serverless` — multi-collection benchmarks against Qdrant Serverless + +Serverless uses a **collection per tenant**, so the interesting workload is +traffic spread across many collections rather than one shared collection with +tenant payload. `bfb serverless` talks to the space through the +[`QdrantServerless`](https://github.com/qdrant/rust-client) client (API key via +`QDRANT_API_KEY`, URI defaults to port 443). + +```bash +# Upload 10M points across 100 collections (created lazily on first upsert) +bfb serverless upload \ + --uri https://serverless.example.cloud.qdrant.io \ + --collection-prefix benchmark- \ + --collections-count 100 \ + --distribution uniform \ + --total-points 10M \ + --config-file examples/serverless-upload.yaml \ + -b 64 -p 8 + +# Query existing collections; vector shape is inferred from collection config +bfb serverless query \ + --uri https://serverless.example.cloud.qdrant.io \ + --collection-prefix benchmark- \ + --distribution zipf \ + -n 10k -p 8 + +# Tear down everything with that prefix +bfb serverless clear \ + --uri https://serverless.example.cloud.qdrant.io \ + --collection-prefix benchmark- +``` + +| Flag | Meaning | +|------|---------| +| `--collection-prefix` | Shared name prefix (`benchmark-` → `benchmark-0` … `benchmark-99`) | +| `--collections-count` | How many collection slots to spread upload across | +| `--distribution` | `uniform` or `zipf` — how points/queries are routed across slots | +| `--total-points` | Upload only: total points across all collections (falls back to `-n`) | +| `--config-file` / `--file` | Upload: YAML collection shape (same schema as `bfb upload --file`) | + +Collections are **not** created up front. On first upsert to a slot, bfb creates +it from the upload YAML (only the tenant-facing subset: dense/sparse vectors and +payload indexes — HNSW/quantization/on-disk knobs are ignored). A registry +prints how many collections preexisting / created / queryable after the run. + ### `schema` — print the upload-config file schema Print an annotated YAML reference enumerating every option accepted by an diff --git a/examples/serverless-upload.yaml b/examples/serverless-upload.yaml new file mode 100644 index 0000000..cff7e26 --- /dev/null +++ b/examples/serverless-upload.yaml @@ -0,0 +1,26 @@ +# Minimal upload shape for `bfb serverless upload`. +# +# bfb serverless upload \ +# --uri https://serverless.example.cloud.qdrant.io \ +# --collection-prefix benchmark- \ +# --collections-count 100 \ +# --distribution uniform \ +# --total-points 1M \ +# --config-file examples/serverless-upload.yaml +# +# Only the tenant-facing subset is sent to serverless (dense/sparse vectors + +# payload indexes). HNSW / quantization / on-disk knobs below are ignored by +# the converter — keep them if you also reuse this file with `bfb upload`. + +collection: + vectors: + - size: 128 + distance: cosine + source: random + + fields: + - name: color + type: keyword + source: + type: random + cardinality: 100 diff --git a/src/args/mod.rs b/src/args/mod.rs index 9e5a19f..88d221d 100644 --- a/src/args/mod.rs +++ b/src/args/mod.rs @@ -550,6 +550,12 @@ pub enum Command { #[clap(value_enum)] shell: clap_complete::Shell, }, + + /// Benchmark Qdrant Serverless (multi-collection / per-tenant mode). + /// + /// Spreads upload and query traffic across a range of collections instead + /// of a single shared collection. See `bfb serverless --help`. + Serverless(crate::serverless::ServerlessArgs), } /// `--file` or `--example` (exactly one required). Flattened into upload / @@ -603,7 +609,7 @@ impl Args { } } -fn parse_number(n: &str) -> Result { +pub(crate) fn parse_number(n: &str) -> Result { parse_number_impl(n) .and_then(|v| v.try_into().ok()) .ok_or_else(|| format!("Invalid number: {n}")) diff --git a/src/main.rs b/src/main.rs index 1da7930..de2f03f 100644 --- a/src/main.rs +++ b/src/main.rs @@ -23,6 +23,7 @@ mod save_jsonl; mod scroll; mod search; mod self_update; +mod serverless; mod stats; mod upload; mod upsert; @@ -129,6 +130,9 @@ async fn run_benchmark(args: Args, stopped: Arc) -> Result<()> { Some(Command::Search(search_args)) => return run_search(args, search_args, stopped).await, Some(Command::Upload(upload_args)) => return run_upload(args, upload_args, stopped).await, Some(Command::Scroll(scroll_args)) => return run_scroll(args, scroll_args, stopped).await, + Some(Command::Serverless(serverless_args)) => { + return serverless::run(args, serverless_args, stopped).await; + } // `Schema` / `SelfUpdate` / `Completions` are handled before the runtime // starts; `None` falls through. Some( diff --git a/src/serverless/args.rs b/src/serverless/args.rs new file mode 100644 index 0000000..de041ab --- /dev/null +++ b/src/serverless/args.rs @@ -0,0 +1,87 @@ +//! CLI for `bfb serverless {upload,clear,query}`. + +use clap::{Args as ClapArgs, Subcommand, ValueEnum}; + +use super::distribution::Distribution; + +/// `bfb serverless` — multi-collection benchmarks against Qdrant Serverless. +#[derive(ClapArgs, Debug, Clone)] +pub struct ServerlessArgs { + #[command(subcommand)] + pub command: ServerlessCommand, +} + +#[derive(Subcommand, Debug, Clone)] +pub enum ServerlessCommand { + /// Upload points across a range of collections (created lazily on first use). + Upload(ServerlessUploadArgs), + + /// Delete every collection whose name starts with `--collection-prefix`. + Clear(ServerlessClearArgs), + + /// Run queries routed across existing collections matching the prefix. + Query(ServerlessQueryArgs), +} + +#[derive(ClapArgs, Debug, Clone)] +pub struct ServerlessUploadArgs { + /// Prefix shared by every collection name (`benchmark-` → `benchmark-0` …). + #[clap(long)] + pub collection_prefix: String, + + /// How many collections to spread points across. + #[clap(long, value_parser = crate::args::parse_number)] + pub collections_count: usize, + + /// How points are allocated across collections. + #[clap(long, value_enum, default_value_t = DistributionArg::Uniform)] + pub distribution: DistributionArg, + + /// Total number of points to upload across all collections. + /// Falls back to the global `-n` / `--num-vectors` when omitted. + #[clap(long, value_parser = crate::args::parse_number)] + pub total_points: Option, + + /// Path to a YAML upload-shape config (`bfb upload --file` schema). + /// Alias of `--file` for the Notion CLI wording. + #[clap(long = "config-file", visible_alias = "file", value_name = "PATH")] + pub config_file: String, +} + +#[derive(ClapArgs, Debug, Clone)] +pub struct ServerlessClearArgs { + /// Delete every collection whose name starts with this prefix. + #[clap(long)] + pub collection_prefix: String, +} + +#[derive(ClapArgs, Debug, Clone)] +pub struct ServerlessQueryArgs { + /// Query every existing collection whose name starts with this prefix. + #[clap(long)] + pub collection_prefix: String, + + /// How queries are routed across matching collections. + #[clap(long, value_enum, default_value_t = DistributionArg::Uniform)] + pub distribution: DistributionArg, + + /// Optional YAML search-shape config. When omitted, vector shape is read + /// from an existing collection's serverless config. + #[clap(long = "config-file", visible_alias = "file", value_name = "PATH")] + pub config_file: Option, +} + +#[derive(Debug, Clone, Copy, ValueEnum, PartialEq, Eq)] +pub enum DistributionArg { + Uniform, + Zipf, +} + +impl From for Distribution { + fn from(value: DistributionArg) -> Self { + match value { + DistributionArg::Uniform => Distribution::Uniform, + DistributionArg::Zipf => Distribution::Zipf, + } + } +} diff --git a/src/serverless/clear.rs b/src/serverless/clear.rs new file mode 100644 index 0000000..38e17de --- /dev/null +++ b/src/serverless/clear.rs @@ -0,0 +1,40 @@ +//! `bfb serverless clear` — delete collections matching a prefix. + +use anyhow::Result; + +use super::args::ServerlessClearArgs; +use super::client; +use super::collections::list_matching; +use crate::args::Args; + +pub async fn run(args: &Args, clear: ServerlessClearArgs) -> Result<()> { + let client = client::random_client(args)?; + let names = list_matching(&client, &clear.collection_prefix).await?; + + if names.is_empty() { + println!( + "No collections matched prefix {:?}", + clear.collection_prefix + ); + return Ok(()); + } + + println!( + "Deleting {} collection(s) with prefix {:?}", + names.len(), + clear.collection_prefix + ); + + let mut deleted = 0usize; + for name in &names { + let ok = client.delete_collection(name).await?; + if ok { + deleted += 1; + println!(" deleted {name}"); + } else { + println!(" skipped {name} (already gone)"); + } + } + println!("Deleted {deleted}/{} collections", names.len()); + Ok(()) +} diff --git a/src/serverless/client.rs b/src/serverless/client.rs new file mode 100644 index 0000000..146d3bb --- /dev/null +++ b/src/serverless/client.rs @@ -0,0 +1,80 @@ +//! Build [`QdrantServerless`] clients from shared BFB [`Args`]. + +use std::time::Duration; + +use anyhow::Result; +use qdrant_client::serverless::QdrantServerless; +use rand::RngExt; +use rand::prelude::SliceRandom; +use tracing::warn; + +use crate::args::Args; + +fn choose_owned(mut items: Vec) -> T { + let mut rng = rand::rng(); + let id = rng.random_range(0..items.len()); + items.swap_remove(id) +} + +/// One serverless client per (`uri` × `connections`) pair, matching regular BFB. +pub fn create_clients(args: &Args) -> Result> { + let api_key = std::env::var("QDRANT_API_KEY").ok(); + let mut clients = Vec::new(); + + for _ in 0..args.connections { + for uri in &args.uri { + let mut builder = QdrantServerless::from_url(uri); + if let Some(timeout) = args.timeout { + let channel_timeout = Duration::from_secs(timeout as u64 + 5); + builder = builder + .timeout(channel_timeout) + .connect_timeout(channel_timeout); + } + if let Some(api_key) = &api_key { + builder = builder.api_key(api_key.as_str()); + } + clients.push(builder.build()?); + } + } + Ok(clients) +} + +pub fn random_client(args: &Args) -> Result { + Ok(choose_owned(create_clients(args)?)) +} + +/// Try the request on every client in random order, retrying `args.retries` +/// times with `args.retry_interval` between rounds. +pub async fn retry_with_clients< + 'a, + R, + T: std::future::Future>, +>( + clients: &'a [QdrantServerless], + args: &Args, + mut call: impl FnMut(&'a QdrantServerless) -> T, +) -> anyhow::Result { + let mut rng = rand::rng(); + let mut permutation = (0..clients.len()).collect::>(); + let mut previous_err: Option = None; + + for attempt in 0..=args.retries { + permutation.shuffle(&mut rng); + for client_id in &permutation { + let client = &clients[*client_id]; + match call(client).await { + Ok(v) => return Ok(v), + Err(err) => previous_err = Some(err.into()), + } + } + + if attempt < args.retries { + if let Some(err) = &previous_err { + warn!("Request failed at attempt {}: {err}", attempt + 1); + } + tokio::time::sleep(Duration::from_secs_f32(args.retry_interval.max(0.0))).await; + } + } + + Err(previous_err.unwrap_or_else(|| anyhow::anyhow!("No clients"))) +} diff --git a/src/serverless/collections.rs b/src/serverless/collections.rs new file mode 100644 index 0000000..ff4d38d --- /dev/null +++ b/src/serverless/collections.rs @@ -0,0 +1,139 @@ +//! Track which serverless collections existed before a run, which were created +//! during upload, and which can be queried afterwards. + +use std::collections::HashSet; +use std::sync::Mutex; + +use anyhow::{Context, Result}; +use qdrant_client::serverless::{CollectionConfig, QdrantServerless}; +use tokio::sync::Mutex as AsyncMutex; + +/// Name of the `i`-th collection under `prefix` (`benchmark-` + `0` → `benchmark-0`). +pub fn collection_name(prefix: &str, index: usize) -> String { + format!("{prefix}{index}") +} + +/// All collections currently in the space whose name starts with `prefix`. +pub async fn list_matching(client: &QdrantServerless, prefix: &str) -> Result> { + let mut names: Vec = client + .list_collections() + .await + .context("list_collections")? + .into_iter() + .map(|c| c.collection_name) + .filter(|n| n.starts_with(prefix)) + .collect(); + names.sort(); + Ok(names) +} + +/// Registry for one serverless upload/query experiment. +pub struct CollectionRegistry { + prefix: String, + /// Present in the space before the experiment started. + preexisting: HashSet, + /// Created by this process during upload. + created: Mutex>, + /// Have received at least one successful upsert (or existed with points). + queryable: Mutex>, + /// Config used when lazily creating a missing collection. + create_config: CollectionConfig, + /// Serializes create RPCs so parallel workers do not race on the same name. + create_lock: AsyncMutex<()>, +} + +impl CollectionRegistry { + pub async fn bootstrap( + client: &QdrantServerless, + prefix: &str, + collections_count: usize, + create_config: CollectionConfig, + ) -> Result { + let preexisting: HashSet = + list_matching(client, prefix).await?.into_iter().collect(); + + println!( + "Serverless registry: prefix={prefix:?} slots={collections_count} preexisting={}", + preexisting.len() + ); + for name in preexisting.iter().take(5) { + println!(" preexisting: {name}"); + } + if preexisting.len() > 5 { + println!(" … {} more", preexisting.len() - 5); + } + + Ok(Self { + prefix: prefix.to_string(), + preexisting, + created: Mutex::new(HashSet::new()), + queryable: Mutex::new(HashSet::new()), + create_config, + create_lock: AsyncMutex::new(()), + }) + } + + pub fn name(&self, index: usize) -> String { + collection_name(&self.prefix, index) + } + + /// Ensure collection `index` exists, creating it lazily on first use. + /// Returns the collection name. + pub async fn ensure(&self, client: &QdrantServerless, index: usize) -> Result { + let name = self.name(index); + + // Fast path: already known to exist (preexisting or created this run). + { + let created = self.created.lock().unwrap(); + if self.preexisting.contains(&name) || created.contains(&name) { + return Ok(name); + } + } + + // Slow path: serialize creates so parallel upserts of the same slot + // do not all race `create_collection`. + let _guard = self.create_lock.lock().await; + + // Re-check under the lock. + { + let created = self.created.lock().unwrap(); + if self.preexisting.contains(&name) || created.contains(&name) { + return Ok(name); + } + } + + let info = client + .get_collection(&name) + .await + .with_context(|| format!("get_collection {name}"))?; + + if info.exists { + // Appeared between bootstrap and now (another process / prior run). + // Do not claim we created it. + return Ok(name); + } + + client + .create_collection(&name, self.create_config.clone()) + .await + .with_context(|| format!("create_collection {name}"))?; + self.created.lock().unwrap().insert(name.clone()); + println!("Created collection {name}"); + Ok(name) + } + + pub fn mark_queryable(&self, name: &str) { + self.queryable.lock().unwrap().insert(name.to_string()); + } + + pub fn summary(&self) { + let created = self.created.lock().unwrap(); + let queryable = self.queryable.lock().unwrap(); + println!( + "Serverless registry summary: preexisting={} created={} queryable={}", + self.preexisting.len(), + created.len(), + queryable.len() + ); + } +} diff --git a/src/serverless/convert.rs b/src/serverless/convert.rs new file mode 100644 index 0000000..87ffb2c --- /dev/null +++ b/src/serverless/convert.rs @@ -0,0 +1,109 @@ +//! Convert a BFB [`UploadConfig`] into a serverless [`CollectionConfig`]. +//! +//! Serverless only accepts the tenant-facing shape (dense/sparse vectors + +//! payload indexes). Storage knobs from the upload YAML (HNSW, quantization, +//! on-disk placement, …) are ignored — the serverless manager decides those. + +use anyhow::{Result, bail}; +use qdrant_client::serverless::{ + BoolIndex, CollectionConfig, DenseVectorConfig, Distance, FloatIndex, GeoIndex, IntegerIndex, + KeywordIndex, PayloadIndex, SparseVectorConfig, TextIndex, Tokenizer, UuidIndex, +}; + +use crate::config::payload::PayloadType; +use crate::config::vector::{DistanceKind, ModifierKind}; +use crate::config::{TokenizerKind, UploadConfig}; + +/// Map an upload-shape YAML into the serverless create-collection config. +pub fn to_serverless_config(upload: &UploadConfig) -> Result { + let c = &upload.collection; + if c.vectors.is_empty() && c.sparse_vectors.is_empty() { + bail!("config must define at least one dense or sparse vector"); + } + + let mut config = CollectionConfig::new(); + + for v in &c.vectors { + let dense = DenseVectorConfig::new(v.size, map_distance(v.distance)) + .multivector(v.multivector.is_some()); + match &v.name { + Some(name) => config = config.named_dense_vector(name.clone(), dense), + None => config = config.dense_vector(dense), + } + } + + for s in &c.sparse_vectors { + let sparse = SparseVectorConfig::new().use_idf(s.modifier == ModifierKind::Idf); + config = config.named_sparse_vector(s.name.clone(), sparse); + } + + for field in &c.fields { + if !field.index { + continue; + } + let index: PayloadIndex = match field.kind { + PayloadType::Keyword => KeywordIndex.into(), + PayloadType::Integer => IntegerIndex::new() + .lookup(true) + .range(field.range_index) + .into(), + PayloadType::Float => FloatIndex.into(), + PayloadType::Bool => BoolIndex.into(), + PayloadType::Uuid => UuidIndex.into(), + PayloadType::Geo => GeoIndex.into(), + PayloadType::Text => { + let mut text = TextIndex::new(); + if let Some(tok) = field.tokenizer { + text = text.tokenizer(map_tokenizer(tok)); + } + text.into() + } + PayloadType::Datetime => qdrant_client::serverless::DatetimeIndex.into(), + }; + config = config.payload_index(field.name.clone(), index); + } + + Ok(config) +} + +fn map_distance(d: DistanceKind) -> Distance { + match d { + DistanceKind::Cosine => Distance::Cosine, + DistanceKind::Dot => Distance::Dot, + DistanceKind::Euclid => Distance::Euclid, + DistanceKind::Manhattan => Distance::Manhattan, + } +} + +fn map_tokenizer(t: TokenizerKind) -> Tokenizer { + match t { + TokenizerKind::Word => Tokenizer::Word, + TokenizerKind::Whitespace => Tokenizer::Whitespace, + TokenizerKind::Prefix => Tokenizer::Prefix, + TokenizerKind::Multilingual => Tokenizer::Multilingual, + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn converts_basic_upload_config() { + let yaml = r#" +collection: + vectors: + - size: 128 + distance: cosine + fields: + - name: color + type: keyword +"#; + let upload = crate::config::parse(yaml, "test").unwrap(); + let cfg = to_serverless_config(&upload).unwrap(); + assert_eq!(cfg.dense_vectors.len(), 1); + assert!(cfg.dense_vectors.contains_key("")); + assert_eq!(cfg.dense_vectors[""].size, 128); + assert!(cfg.payload_indexes.contains_key("color")); + } +} diff --git a/src/serverless/distribution.rs b/src/serverless/distribution.rs new file mode 100644 index 0000000..31a9330 --- /dev/null +++ b/src/serverless/distribution.rs @@ -0,0 +1,95 @@ +//! Uniform / Zipf sampling over a fixed set of collection slots. + +use rand::Rng; +use rand::RngExt; +use rand::prelude::Distribution as RandDistribution; +use rand_distr::Zipf; + +/// How work (points or queries) is spread across collections. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum Distribution { + Uniform, + /// Zipf with exponent 1.03 over ranks `1..=n` (same default as payload text). + Zipf, +} + +/// Pick a collection index in `0..n` according to [`Distribution`]. +#[derive(Debug, Clone)] +pub struct CollectionPicker { + n: usize, + zipf: Option>, +} + +impl CollectionPicker { + pub fn new(n: usize, distribution: Distribution) -> anyhow::Result { + anyhow::ensure!(n > 0, "collections-count must be > 0"); + let zipf = match distribution { + Distribution::Uniform => None, + Distribution::Zipf => Some( + Zipf::new(n as f64, 1.03) + .map_err(|e| anyhow::anyhow!("failed to build Zipf(n={n}): {e}"))?, + ), + }; + Ok(Self { n, zipf }) + } + + /// Sample a collection index in `0..n`. + pub fn pick(&self, rng: &mut impl Rng) -> usize { + match &self.zipf { + None => rng.random_range(0..self.n), + // Zipf samples in `1..=n`; convert to 0-based. + Some(zipf) => ((zipf.sample(rng) as usize).saturating_sub(1)).min(self.n - 1), + } + } + + /// Allocate `total` points across `n` collections according to the + /// distribution. Returns a vec of length `n` summing to `total`. + pub fn allocate(&self, total: usize, rng: &mut impl Rng) -> Vec { + if self.n == 1 { + return vec![total]; + } + match self.zipf { + None => { + // Even split; remainder goes to the first slots. + let base = total / self.n; + let rem = total % self.n; + (0..self.n).map(|i| base + usize::from(i < rem)).collect() + } + Some(_) => { + // Monte-Carlo allocation: sample `total` ranks. + let mut counts = vec![0usize; self.n]; + for _ in 0..total { + counts[self.pick(rng)] += 1; + } + counts + } + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + use rand::SeedableRng; + use rand::rngs::StdRng; + + #[test] + fn uniform_allocate_sums() { + let picker = CollectionPicker::new(10, Distribution::Uniform).unwrap(); + let mut rng = StdRng::seed_from_u64(1); + let counts = picker.allocate(1000, &mut rng); + assert_eq!(counts.len(), 10); + assert_eq!(counts.iter().sum::(), 1000); + assert!(counts.iter().all(|&c| c == 100)); + } + + #[test] + fn zipf_allocate_sums_and_skews() { + let picker = CollectionPicker::new(10, Distribution::Zipf).unwrap(); + let mut rng = StdRng::seed_from_u64(42); + let counts = picker.allocate(10_000, &mut rng); + assert_eq!(counts.iter().sum::(), 10_000); + // Rank 0 should get strictly more than the last rank on average. + assert!(counts[0] > counts[9]); + } +} diff --git a/src/serverless/mod.rs b/src/serverless/mod.rs new file mode 100644 index 0000000..0b757b5 --- /dev/null +++ b/src/serverless/mod.rs @@ -0,0 +1,40 @@ +//! Serverless benchmarking mode. +//! +//! Unlike the regular single-collection workflow, serverless mode spreads work +//! across a *range* of collections (one per tenant). Collections are created +//! lazily on first upsert; the registry tracks which existed before the run, +//! which were created during upload, and which are queryable afterwards. +//! +//! ```text +//! bfb serverless upload --collection-prefix benchmark- --collections-count 100 \ +//! --distribution uniform --total-points 10M --config-file config.yaml +//! bfb serverless clear --collection-prefix benchmark- +//! bfb serverless query --collection-prefix benchmark- --distribution zipf -n 10k +//! ``` + +mod args; +mod clear; +mod client; +mod collections; +mod convert; +mod distribution; +mod query; +mod upload; + +pub use args::{ServerlessArgs, ServerlessCommand}; + +use std::sync::Arc; +use std::sync::atomic::AtomicBool; + +use anyhow::Result; + +use crate::args::Args; + +/// Dispatch a `bfb serverless …` subcommand. +pub async fn run(args: Args, serverless: ServerlessArgs, stopped: Arc) -> Result<()> { + match serverless.command { + ServerlessCommand::Upload(upload_args) => upload::run(&args, upload_args, stopped).await, + ServerlessCommand::Clear(clear_args) => clear::run(&args, clear_args).await, + ServerlessCommand::Query(query_args) => query::run(&args, query_args, stopped).await, + } +} diff --git a/src/serverless/query.rs b/src/serverless/query.rs new file mode 100644 index 0000000..dbc5e16 --- /dev/null +++ b/src/serverless/query.rs @@ -0,0 +1,231 @@ +//! `bfb serverless query` — route searches across existing collections. + +use std::sync::Arc; +use std::sync::Mutex; +use std::sync::atomic::{AtomicBool, Ordering}; +use std::time::Instant; + +use anyhow::{Context, Result, bail}; +use futures::stream::StreamExt; +use indicatif::{MultiProgress, ProgressBar, ProgressDrawTarget, ProgressStyle}; +use qdrant_client::qdrant::{QueryPointsBuilder, VectorInput}; +use rand::Rng; +use rand::RngExt; + +use super::args::ServerlessQueryArgs; +use super::client::{self, retry_with_clients}; +use super::collections::list_matching; +use super::distribution::CollectionPicker; +use crate::args::Args; +use crate::generators::random::random_dense_vector; +use crate::processor::Timing; +use crate::stats::{print_stats, throttler}; +use qdrant_client::serverless::CollectionConfig; + +/// Shape used when no search YAML is provided: taken from a live collection. +struct InferredShape { + /// `(vector_name, size)` — empty name is the default unnamed vector. + dense: Vec<(String, u64)>, +} + +impl InferredShape { + fn from_config(config: &CollectionConfig) -> Result { + let mut dense: Vec<(String, u64)> = config + .dense_vectors + .iter() + .map(|(name, cfg)| (name.clone(), cfg.size)) + .collect(); + dense.sort_by(|a, b| a.0.cmp(&b.0)); + if dense.is_empty() { + bail!("collection has no dense vectors to query"); + } + Ok(Self { dense }) + } + + fn random_query(&self, rng: &mut impl Rng) -> (Option, Vec) { + let idx = rng.random_range(0..self.dense.len()); + let (name, size) = &self.dense[idx]; + let vector = random_dense_vector(rng, *size as usize, false); + let using = if name.is_empty() { + None + } else { + Some(name.clone()) + }; + (using, vector) + } +} + +pub async fn run(args: &Args, query: ServerlessQueryArgs, stopped: Arc) -> Result<()> { + let clients = client::create_clients(args)?; + let names = list_matching(&clients[0], &query.collection_prefix).await?; + if names.is_empty() { + bail!( + "no collections matched prefix {:?} — run `bfb serverless upload` first", + query.collection_prefix + ); + } + + println!( + "Querying {} collection(s) with prefix {:?} ({:?})", + names.len(), + query.collection_prefix, + query.distribution + ); + + // Guess vector shape from an existing collection's config when no search + // YAML is given (Notion: "guess shape of the vectors from collection config"). + if query.config_file.is_some() { + // Search-shape YAML for serverless is not wired yet; infer instead. + eprintln!( + "note: --config-file on `serverless query` is accepted but ignored for now; \ + vector shape is inferred from collection {:?}", + names[0] + ); + } + let shape = infer_shape(&clients[0], &names[0]).await?; + + let picker = CollectionPicker::new(names.len(), query.distribution.into())?; + let num_queries = args.num_vectors.unwrap_or(10_000); + + let logger = env_logger::Builder::from_default_env().build(); + let multiprogress = MultiProgress::new(); + indicatif_log_bridge::LogWrapper::new(multiprogress.clone(), logger) + .try_init() + .ok(); + + let bar = multiprogress.add(ProgressBar::new(num_queries as u64)); + bar.set_style( + ProgressStyle::default_bar() + .template("{msg} [{elapsed_precise}] {wide_bar} [{per_sec:>3}] {pos}/{len} (eta:{eta})") + .expect("progress style"), + ); + bar.set_draw_target(ProgressDrawTarget::stdout_with_hz(2)); + let bar = Arc::new(bar); + + let server_timings = Arc::new(Mutex::new(Vec::::new())); + let full_timings = Arc::new(Mutex::new(Vec::::new())); + let rps_timings = Arc::new(Mutex::new(Vec::::new())); + let start = Instant::now(); + + let parallel = if args.rps.is_some() { + 1 + } else { + args.parallel.max(1) + }; + let throttler = throttler(args.rps.map(|r| r as f32).or(args.throttle)); + let stopped_flag = stopped.clone(); + let clients = Arc::new(clients); + let names = Arc::new(names); + let shape = Arc::new(shape); + let args_owned = args.clone(); + + futures::stream::iter(0..num_queries) + .take_while(|_| { + let s = stopped_flag.clone(); + async move { !s.load(Ordering::Relaxed) } + }) + .zip(throttler) + .for_each_concurrent(parallel, |(req_id, _)| { + let clients = clients.clone(); + let names = names.clone(); + let shape = shape.clone(); + let bar = bar.clone(); + let server_timings = server_timings.clone(); + let full_timings = full_timings.clone(); + let rps_timings = rps_timings.clone(); + let args = args_owned.clone(); + let stopped = stopped_flag.clone(); + let picker = picker.clone(); + + async move { + if stopped.load(Ordering::Relaxed) { + return; + } + + let mut rng = rand::rng(); + let coll_idx = picker.pick(&mut rng); + let collection = names[coll_idx].clone(); + let (using, vector) = shape.random_query(&mut rng); + + let mut builder = QueryPointsBuilder::new(collection.clone()) + .query(VectorInput::new_dense(vector)) + .limit(args.search_limit as u64) + .with_payload(args.search_with_payload) + .with_vectors(args.search_with_vectors); + if let Some(name) = using { + builder = builder.using(name); + } + if let Some(timeout) = args.timeout { + builder = builder.timeout(timeout as u64); + } + let request = builder.build(); + + let req_start = Instant::now(); + let res = retry_with_clients(&clients, &args, |c| c.query(request.clone())).await; + let full = req_start.elapsed().as_secs_f32(); + + match res { + Ok(resp) => { + let delay = start.elapsed().as_millis() as u32; + server_timings.lock().unwrap().push(Timing { + delay_millis: delay, + value: resp.time as f32, + }); + full_timings.lock().unwrap().push(Timing { + delay_millis: delay, + value: full, + }); + // Instantaneous RPS estimate from inter-arrival of completions. + let elapsed = start.elapsed().as_secs_f32().max(1e-6); + rps_timings.lock().unwrap().push(Timing { + delay_millis: delay, + value: (req_id as f32 + 1.0) / elapsed, + }); + if resp.time > args.timing_threshold { + bar.println(format!("Slow query on {collection}: {:?}", resp.time)); + } + bar.inc(1); + } + Err(e) => { + bar.println(format!("query failed on {collection}: {e:?}")); + if !args.ignore_errors { + stopped.store(true, Ordering::Relaxed); + } + } + } + } + }) + .await; + + bar.finish_and_clear(); + let elapsed = start.elapsed().as_secs_f64(); + println!( + "Serverless query finished in {elapsed:.2}s ({:.0} qps wall)", + num_queries as f64 / elapsed.max(1e-9) + ); + + let mut server = server_timings.lock().unwrap().clone(); + let mut full = full_timings.lock().unwrap().clone(); + let mut rps = rps_timings.lock().unwrap().clone(); + print_stats(args, &mut server, "server time", true); + print_stats(args, &mut full, "full time", true); + print_stats(args, &mut rps, "rps", false); + Ok(()) +} + +async fn infer_shape( + client: &qdrant_client::serverless::QdrantServerless, + name: &str, +) -> Result { + let info = client + .get_collection(name) + .await + .with_context(|| format!("get_collection {name}"))?; + if !info.exists { + bail!("collection {name} no longer exists"); + } + let config = info.config.ok_or_else(|| { + anyhow::anyhow!("collection {name} has no config; re-upload or pass a search config") + })?; + InferredShape::from_config(&config) +} diff --git a/src/serverless/upload.rs b/src/serverless/upload.rs new file mode 100644 index 0000000..fe63fb2 --- /dev/null +++ b/src/serverless/upload.rs @@ -0,0 +1,191 @@ +//! `bfb serverless upload` — spread points across lazily-created collections. + +use std::cmp::min; +use std::sync::Arc; +use std::sync::atomic::{AtomicBool, Ordering}; +use std::time::Instant; + +use anyhow::{Context, Result}; +use futures::stream::StreamExt; +use indicatif::{MultiProgress, ProgressBar, ProgressDrawTarget, ProgressStyle}; +use qdrant_client::qdrant::UpsertPointsBuilder; +use tokio::time::sleep; + +use super::args::ServerlessUploadArgs; +use super::client::{self, retry_with_clients}; +use super::collections::CollectionRegistry; +use super::convert::to_serverless_config; +use super::distribution::CollectionPicker; +use crate::args::Args; +use crate::config; +use crate::generators::{ConfigGenerator, PointGenerator}; +use crate::stats::throttler; + +pub async fn run( + args: &Args, + upload: ServerlessUploadArgs, + stopped: Arc, +) -> Result<()> { + anyhow::ensure!( + upload.collections_count > 0, + "--collections-count must be > 0" + ); + + let yaml = std::fs::read_to_string(&upload.config_file) + .with_context(|| format!("read config {}", upload.config_file))?; + let upload_config = config::parse(&yaml, &upload.config_file)?; + let serverless_config = to_serverless_config(&upload_config)?; + + let total_points = upload.total_points.or(args.num_vectors).unwrap_or(100_000); + + let clients = client::create_clients(args)?; + let registry = Arc::new( + CollectionRegistry::bootstrap( + &clients[0], + &upload.collection_prefix, + upload.collections_count, + serverless_config, + ) + .await?, + ); + + let picker = CollectionPicker::new(upload.collections_count, upload.distribution.into())?; + let mut rng = rand::rng(); + let per_collection = picker.allocate(total_points, &mut rng); + + println!( + "Uploading {total_points} points across {} collections ({:?})", + upload.collections_count, upload.distribution + ); + for (i, &n) in per_collection.iter().enumerate().take(5) { + println!(" {} → {n} points", registry.name(i)); + } + if per_collection.len() > 5 { + println!(" …"); + } + + let generator: Arc = Arc::new(ConfigGenerator::new(&upload_config)?); + + let logger = env_logger::Builder::from_default_env().build(); + let multiprogress = MultiProgress::new(); + indicatif_log_bridge::LogWrapper::new(multiprogress.clone(), logger) + .try_init() + .ok(); + + let bar = multiprogress.add(ProgressBar::new(total_points as u64)); + bar.set_style( + ProgressStyle::default_bar() + .template("{msg} [{elapsed_precise}] {wide_bar} [{per_sec:>3}] {pos}/{len} (eta:{eta})") + .expect("progress style"), + ); + bar.set_draw_target(ProgressDrawTarget::stdout_with_hz(2)); + let bar = Arc::new(bar); + + // Flatten (collection_idx, local_point_id) into a work list of batches. + // Each batch stays within one collection so upserts stay simple. + let mut batches: Vec<(usize, u64, usize)> = Vec::new(); // (coll_idx, start_id, count) + for (coll_idx, &count) in per_collection.iter().enumerate() { + if count == 0 { + continue; + } + let mut remaining = count; + let mut start = 0u64; + while remaining > 0 { + let n = min(args.batch_size, remaining); + batches.push((coll_idx, start, n)); + start += n as u64; + remaining -= n; + } + } + + let started = Instant::now(); + let parallel = if args.rps.is_some() { + // RPS mode: one in-flight stream controlled by the throttler below. + 1 + } else { + args.parallel.max(1) + }; + + let throttler = throttler(args.rps.map(|r| r as f32).or(args.throttle)); + let stopped_flag = stopped.clone(); + let clients = Arc::new(clients); + let args = args.clone(); + + futures::stream::iter(batches.into_iter().enumerate()) + .take_while(|_| { + let s = stopped_flag.clone(); + async move { !s.load(Ordering::Relaxed) } + }) + .zip(throttler) + .for_each_concurrent(parallel, |((batch_no, (coll_idx, start_id, count)), _)| { + let clients = clients.clone(); + let registry = registry.clone(); + let generator = generator.clone(); + let bar = bar.clone(); + let args = args.clone(); + let stopped = stopped_flag.clone(); + + async move { + if stopped.load(Ordering::Relaxed) { + return; + } + + let client = &clients[batch_no % clients.len()]; + let name = match registry.ensure(client, coll_idx).await { + Ok(n) => n, + Err(e) => { + bar.println(format!("ensure collection failed: {e:?}")); + if !args.ignore_errors { + stopped.store(true, Ordering::Relaxed); + } + return; + } + }; + + let mut points = Vec::with_capacity(count); + for i in 0..count { + points.push(generator.make_point(start_id + i as u64)); + } + + let mut request = + UpsertPointsBuilder::new(name.clone(), points).wait(args.wait_on_upsert); + if let Some(timeout) = args.timeout { + request = request.timeout(timeout as u64); + } + let request = request.build(); + + let res = + retry_with_clients(&clients, &args, |c| c.upsert_points(request.clone())).await; + + match res { + Ok(resp) => { + if resp.time > args.timing_threshold { + bar.println(format!("Slow upsert on {name}: {:?}", resp.time)); + } + registry.mark_queryable(&name); + bar.inc(count as u64); + } + Err(e) => { + bar.println(format!("upsert failed on {name}: {e:?}")); + if !args.ignore_errors { + stopped.store(true, Ordering::Relaxed); + } + } + } + + if let Some(delay_millis) = args.delay { + sleep(std::time::Duration::from_millis(delay_millis as u64)).await; + } + } + }) + .await; + + bar.finish_and_clear(); + let elapsed = started.elapsed().as_secs_f64(); + println!( + "Serverless upload finished in {elapsed:.2}s ({:.0} points/s)", + total_points as f64 / elapsed.max(1e-9) + ); + registry.summary(); + Ok(()) +} From 48ae548ebca68edcba0603be2ab8f9d0088d52de Mon Sep 17 00:00:00 2001 From: Cursor Agent Date: Wed, 2 Sep 2026 11:19:57 +0000 Subject: [PATCH 2/4] Register serverless-upload in built-in examples catalog The examples/ directory must match EXAMPLES; CI failed because the new YAML was present on disk but not listed. --- src/config/examples.rs | 5 +++++ 1 file changed, 5 insertions(+) diff --git a/src/config/examples.rs b/src/config/examples.rs index 2585045..458952b 100644 --- a/src/config/examples.rs +++ b/src/config/examples.rs @@ -90,6 +90,11 @@ pub static EXAMPLES: &[Example] = &[ Upload, "Full LAION-400M corpus (~410 parts, streamed with cache: evict)" ), + example!( + "serverless-upload", + Upload, + "Minimal upload shape for `bfb serverless upload` (dense + keyword)" + ), example!( "search-config", Search, From 2d32a4aa416f98543f057f7773ce57f455d0b00d Mon Sep 17 00:00:00 2001 From: generall Date: Wed, 2 Sep 2026 15:47:24 +0200 Subject: [PATCH 3/4] Fix serverless mode review findings - --rps paced requests one at a time; upload now uses an interval pacer with unbounded concurrency and query runs through the shared stats::process loop - upload/query failures set the stop flag but exited 0; the first error is now returned - upload throughput reported the requested count instead of acknowledged points; RPS/QPS series now come from the progress bar like regular search - accept --file / --example (ConfigArgs) so the registered serverless-upload example is reachable; wire an optional search YAML into `serverless query` via ConfigSearchGenerator (dense, sparse, filters, IDF corpus, dataset query sets) - derive a query template per dense and sparse vector when no YAML is given, skip empty collections, default to the shared 100k request count - record upsert latency, honour --jsonl-updates, and populate BenchmarkResults so --json / --jsonl-searches / --jsonl-rps work - per-slot create locks, verify preexisting collections against the YAML's vector shapes, lay point ids out contiguously across collections from --offset and cap totals by the dataset size - make client::retry_with_clients generic and drop the serverless copy; build serverless clients from the shared get_config - remove the unused queryable set and Notion references; document that the precision tier is left at the service default - keep only the qdrant-client bump in Cargo.lock Co-Authored-By: Claude Fable 5.1 Claude-Session: https://claude.ai/code/session_01Qdso7ZDKX1kUnn7VbPpVAE --- Cargo.lock | 2 +- README.md | 39 ++- examples/serverless-upload.yaml | 6 +- src/client.rs | 14 +- src/serverless/args.rs | 37 ++- src/serverless/clear.rs | 10 +- src/serverless/client.rs | 83 +----- src/serverless/collections.rs | 189 ++++++++---- src/serverless/convert.rs | 5 +- src/serverless/mod.rs | 6 +- src/serverless/query.rs | 490 ++++++++++++++++++++------------ src/serverless/upload.rs | 351 +++++++++++++++-------- 12 files changed, 780 insertions(+), 452 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 36cf61b..fd6dc49 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1880,7 +1880,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "32497e9a4c7b38532efcdebeef879707aa9f794296a4f0244f6f69e9bc8574bd" dependencies = [ "fastrand", - "getrandom 0.3.4", + "getrandom 0.4.2", "once_cell", "rustix", "windows-sys 0.61.2", diff --git a/README.md b/README.md index ddecf7d..a68f165 100644 --- a/README.md +++ b/README.md @@ -371,8 +371,10 @@ you prefer flat CLI flags over a YAML file. Serverless uses a **collection per tenant**, so the interesting workload is traffic spread across many collections rather than one shared collection with tenant payload. `bfb serverless` talks to the space through the -[`QdrantServerless`](https://github.com/qdrant/rust-client) client (API key via -`QDRANT_API_KEY`, URI defaults to port 443). +[`QdrantServerless`](https://github.com/qdrant/rust-client) client. It takes +the same connection flags as the rest of bfb (`--uri`, `--connections`, +`--timeout`, API key via `QDRANT_API_KEY`); a `--uri` without an explicit port +defaults to 443. ```bash # Upload 10M points across 100 collections (created lazily on first upsert) @@ -382,15 +384,21 @@ bfb serverless upload \ --collections-count 100 \ --distribution uniform \ --total-points 10M \ - --config-file examples/serverless-upload.yaml \ + --example serverless-upload \ -b 64 -p 8 -# Query existing collections; vector shape is inferred from collection config +# Query existing collections; query shape is derived from a collection's config bfb serverless query \ --uri https://serverless.example.cloud.qdrant.io \ --collection-prefix benchmark- \ --distribution zipf \ - -n 10k -p 8 + -n 10k --rps 200 --json results.json + +# Same, but with a search YAML (filters, sparse queries, dataset query sets …) +bfb serverless query \ + --uri https://serverless.example.cloud.qdrant.io \ + --collection-prefix benchmark- \ + --file search.yaml -n 10k -p 8 # Tear down everything with that prefix bfb serverless clear \ @@ -401,15 +409,28 @@ bfb serverless clear \ | Flag | Meaning | |------|---------| | `--collection-prefix` | Shared name prefix (`benchmark-` → `benchmark-0` … `benchmark-99`) | -| `--collections-count` | How many collection slots to spread upload across | +| `--collections-count` | Upload only: how many collection slots to spread points across | | `--distribution` | `uniform` or `zipf` — how points/queries are routed across slots | | `--total-points` | Upload only: total points across all collections (falls back to `-n`) | -| `--config-file` / `--file` | Upload: YAML collection shape (same schema as `bfb upload --file`) | +| `--file` / `--example` | Upload: collection shape (`bfb upload` schema, required). Query: request shape (`bfb search` schema, optional) | + +The run itself is controlled by the usual global flags: `-p` / `--rps` / +`--throttle`, `-b`, `--search-batch-size`, `--search-limit`, `--retries`, +`--ignore-errors`, `--jsonl-updates` / `--jsonl-searches` / `--jsonl-rps`, and +`--json` for the unified results document. Collections are **not** created up front. On first upsert to a slot, bfb creates it from the upload YAML (only the tenant-facing subset: dense/sparse vectors and -payload indexes — HNSW/quantization/on-disk knobs are ignored). A registry -prints how many collections preexisting / created / queryable after the run. +payload indexes — HNSW/quantization/on-disk knobs are ignored and the precision +tier is left at the service default). A slot that already exists is reused +after checking its vectors match the YAML. Point ids are contiguous across +collections (starting at `--offset`), so with a dataset source every +collection receives a different slice of the data. A registry prints how many +collections were preexisting / created / used after the run. + +`query` only targets collections that hold points. Without a search YAML it +issues random dense and sparse queries matching the vectors it finds in the +first collection's config. ### `schema` — print the upload-config file schema diff --git a/examples/serverless-upload.yaml b/examples/serverless-upload.yaml index cff7e26..9c7a588 100644 --- a/examples/serverless-upload.yaml +++ b/examples/serverless-upload.yaml @@ -6,11 +6,11 @@ # --collections-count 100 \ # --distribution uniform \ # --total-points 1M \ -# --config-file examples/serverless-upload.yaml +# --example serverless-upload # # Only the tenant-facing subset is sent to serverless (dense/sparse vectors + -# payload indexes). HNSW / quantization / on-disk knobs below are ignored by -# the converter — keep them if you also reuse this file with `bfb upload`. +# payload indexes). HNSW / quantization / on-disk knobs are ignored by the +# converter — keep them if you also reuse a file with `bfb upload`. collection: vectors: diff --git a/src/client.rs b/src/client.rs index 495a532..b88028c 100644 --- a/src/client.rs +++ b/src/client.rs @@ -56,10 +56,18 @@ pub fn create_clients(args: &Args) -> Result> { /// Try the request on every client in random order, retrying `args.retries` /// times with `args.retry_interval` between rounds. -pub async fn retry_with_clients<'a, R, T: std::future::Future>>( - clients: &'a [Qdrant], +/// +/// Generic over the client type so the regular [`Qdrant`] client and the +/// serverless client share one retry policy. +pub async fn retry_with_clients< + 'a, + C, + R, + T: std::future::Future>, +>( + clients: &'a [C], args: &Args, - mut call: impl FnMut(&'a Qdrant) -> T, + mut call: impl FnMut(&'a C) -> T, ) -> anyhow::Result { let mut rng = rand::rng(); let mut permutation = (0..clients.len()).collect::>(); diff --git a/src/serverless/args.rs b/src/serverless/args.rs index de041ab..0e5eb44 100644 --- a/src/serverless/args.rs +++ b/src/serverless/args.rs @@ -3,6 +3,7 @@ use clap::{Args as ClapArgs, Subcommand, ValueEnum}; use super::distribution::Distribution; +use crate::args::ConfigArgs; /// `bfb serverless` — multi-collection benchmarks against Qdrant Serverless. #[derive(ClapArgs, Debug, Clone)] @@ -42,10 +43,10 @@ pub struct ServerlessUploadArgs { #[clap(long, value_parser = crate::args::parse_number)] pub total_points: Option, - /// Path to a YAML upload-shape config (`bfb upload --file` schema). - /// Alias of `--file` for the Notion CLI wording. - #[clap(long = "config-file", visible_alias = "file", value_name = "PATH")] - pub config_file: String, + /// Upload-shape YAML: `--file ` or `--example ` (same schema + /// as `bfb upload`). + #[clap(flatten)] + pub config: ConfigArgs, } #[derive(ClapArgs, Debug, Clone)] @@ -65,10 +66,30 @@ pub struct ServerlessQueryArgs { #[clap(long, value_enum, default_value_t = DistributionArg::Uniform)] pub distribution: DistributionArg, - /// Optional YAML search-shape config. When omitted, vector shape is read - /// from an existing collection's serverless config. - #[clap(long = "config-file", visible_alias = "file", value_name = "PATH")] - pub config_file: Option, + /// Optional search-shape YAML (same schema as `bfb search`). When omitted, + /// one random dense or sparse query template is derived from a matching + /// collection's config. + #[clap(flatten)] + pub config: OptionalConfigArgs, +} + +/// `--file` or `--example`, both optional (at most one). +#[derive(ClapArgs, Debug, Clone)] +#[group(multiple = false)] +pub struct OptionalConfigArgs { + /// Path to a YAML config file + #[clap(long)] + pub file: Option, + + /// Built-in example name (`bfb examples` lists them) + #[clap(long, value_name = "NAME")] + pub example: Option, +} + +impl OptionalConfigArgs { + pub fn is_some(&self) -> bool { + self.file.is_some() || self.example.is_some() + } } #[derive(Debug, Clone, Copy, ValueEnum, PartialEq, Eq)] diff --git a/src/serverless/clear.rs b/src/serverless/clear.rs index 38e17de..232127f 100644 --- a/src/serverless/clear.rs +++ b/src/serverless/clear.rs @@ -3,13 +3,17 @@ use anyhow::Result; use super::args::ServerlessClearArgs; -use super::client; +use super::client::single_client; use super::collections::list_matching; use crate::args::Args; pub async fn run(args: &Args, clear: ServerlessClearArgs) -> Result<()> { - let client = client::random_client(args)?; - let names = list_matching(&client, &clear.collection_prefix).await?; + let client = single_client(args)?; + let names: Vec = list_matching(&client, &clear.collection_prefix) + .await? + .into_iter() + .map(|c| c.collection_name) + .collect(); if names.is_empty() { println!( diff --git a/src/serverless/client.rs b/src/serverless/client.rs index 146d3bb..07cad72 100644 --- a/src/serverless/client.rs +++ b/src/serverless/client.rs @@ -1,80 +1,27 @@ //! Build [`QdrantServerless`] clients from shared BFB [`Args`]. - -use std::time::Duration; +//! +//! Reuses the regular client configuration (`--uri` × `--connections`, +//! `--timeout`, `QDRANT_API_KEY`) so both modes are configured the same way. use anyhow::Result; use qdrant_client::serverless::QdrantServerless; -use rand::RngExt; -use rand::prelude::SliceRandom; -use tracing::warn; use crate::args::Args; - -fn choose_owned(mut items: Vec) -> T { - let mut rng = rand::rng(); - let id = rng.random_range(0..items.len()); - items.swap_remove(id) -} +use crate::client::get_config; /// One serverless client per (`uri` × `connections`) pair, matching regular BFB. pub fn create_clients(args: &Args) -> Result> { - let api_key = std::env::var("QDRANT_API_KEY").ok(); - let mut clients = Vec::new(); - - for _ in 0..args.connections { - for uri in &args.uri { - let mut builder = QdrantServerless::from_url(uri); - if let Some(timeout) = args.timeout { - let channel_timeout = Duration::from_secs(timeout as u64 + 5); - builder = builder - .timeout(channel_timeout) - .connect_timeout(channel_timeout); - } - if let Some(api_key) = &api_key { - builder = builder.api_key(api_key.as_str()); - } - clients.push(builder.build()?); - } - } - Ok(clients) + get_config(args) + .into_iter() + .map(|config| Ok(QdrantServerless::new(config)?)) + .collect() } -pub fn random_client(args: &Args) -> Result { - Ok(choose_owned(create_clients(args)?)) -} - -/// Try the request on every client in random order, retrying `args.retries` -/// times with `args.retry_interval` between rounds. -pub async fn retry_with_clients< - 'a, - R, - T: std::future::Future>, ->( - clients: &'a [QdrantServerless], - args: &Args, - mut call: impl FnMut(&'a QdrantServerless) -> T, -) -> anyhow::Result { - let mut rng = rand::rng(); - let mut permutation = (0..clients.len()).collect::>(); - let mut previous_err: Option = None; - - for attempt in 0..=args.retries { - permutation.shuffle(&mut rng); - for client_id in &permutation { - let client = &clients[*client_id]; - match call(client).await { - Ok(v) => return Ok(v), - Err(err) => previous_err = Some(err.into()), - } - } - - if attempt < args.retries { - if let Some(err) = &previous_err { - warn!("Request failed at attempt {}: {err}", attempt + 1); - } - tokio::time::sleep(Duration::from_secs_f32(args.retry_interval.max(0.0))).await; - } - } - - Err(previous_err.unwrap_or_else(|| anyhow::anyhow!("No clients"))) +/// A single client for one-off administrative calls (`clear`, listing). +pub fn single_client(args: &Args) -> Result { + let config = get_config(args) + .into_iter() + .next() + .ok_or_else(|| anyhow::anyhow!("no --uri given"))?; + Ok(QdrantServerless::new(config)?) } diff --git a/src/serverless/collections.rs b/src/serverless/collections.rs index ff4d38d..a48a669 100644 --- a/src/serverless/collections.rs +++ b/src/serverless/collections.rs @@ -1,11 +1,11 @@ -//! Track which serverless collections existed before a run, which were created -//! during upload, and which can be queried afterwards. +//! Track which serverless collections existed before a run and which were +//! created during upload, creating missing ones lazily on first upsert. use std::collections::HashSet; use std::sync::Mutex; -use anyhow::{Context, Result}; -use qdrant_client::serverless::{CollectionConfig, QdrantServerless}; +use anyhow::{Context, Result, bail}; +use qdrant_client::serverless::{CollectionConfig, CollectionSummary, QdrantServerless}; use tokio::sync::Mutex as AsyncMutex; /// Name of the `i`-th collection under `prefix` (`benchmark-` + `0` → `benchmark-0`). @@ -13,33 +13,83 @@ pub fn collection_name(prefix: &str, index: usize) -> String { format!("{prefix}{index}") } -/// All collections currently in the space whose name starts with `prefix`. -pub async fn list_matching(client: &QdrantServerless, prefix: &str) -> Result> { - let mut names: Vec = client +/// All collections currently in the space whose name starts with `prefix`, +/// sorted by name. +pub async fn list_matching( + client: &QdrantServerless, + prefix: &str, +) -> Result> { + let mut summaries: Vec = client .list_collections() .await .context("list_collections")? .into_iter() - .map(|c| c.collection_name) - .filter(|n| n.starts_with(prefix)) + .filter(|c| c.collection_name.starts_with(prefix)) .collect(); - names.sort(); - Ok(names) + summaries.sort_by(|a, b| a.collection_name.cmp(&b.collection_name)); + Ok(summaries) } -/// Registry for one serverless upload/query experiment. +/// Check that every vector the upload config will send exists in `existing` +/// with the same shape. Extra vectors or indexes on the existing collection +/// are fine; a missing or mismatched one would make every upsert fail, so +/// report it up front instead. +pub fn check_vector_shape( + name: &str, + existing: &CollectionConfig, + wanted: &CollectionConfig, +) -> Result<()> { + for (vector, want) in &wanted.dense_vectors { + let label = if vector.is_empty() { + "default dense vector".to_string() + } else { + format!("dense vector {vector:?}") + }; + let Some(have) = existing.dense_vectors.get(vector) else { + bail!( + "collection {name} has no {label}; delete it (`bfb serverless clear`) or use a matching config" + ); + }; + if have.size != want.size || have.distance != want.distance { + bail!( + "collection {name}: {label} is {}d/{:?} but the config wants {}d/{:?}", + have.size, + have.distance, + want.size, + want.distance + ); + } + if have.multivector != want.multivector { + bail!( + "collection {name}: {label} multivector={} but the config wants {}", + have.multivector, + want.multivector + ); + } + } + for vector in wanted.sparse_vectors.keys() { + if !existing.sparse_vectors.contains_key(vector) { + bail!("collection {name} has no sparse vector {vector:?}"); + } + } + Ok(()) +} + +/// Registry for one serverless upload experiment. pub struct CollectionRegistry { prefix: String, /// Present in the space before the experiment started. preexisting: HashSet, + /// Known to exist and match the config: created by this process, or + /// preexisting and verified on first use. + ready: Mutex>, /// Created by this process during upload. created: Mutex>, - /// Have received at least one successful upsert (or existed with points). - queryable: Mutex>, /// Config used when lazily creating a missing collection. create_config: CollectionConfig, - /// Serializes create RPCs so parallel workers do not race on the same name. - create_lock: AsyncMutex<()>, + /// One lock per slot, so parallel workers on the same collection do not + /// race `create_collection` while other slots proceed unblocked. + slot_locks: Vec>, } impl CollectionRegistry { @@ -49,14 +99,19 @@ impl CollectionRegistry { collections_count: usize, create_config: CollectionConfig, ) -> Result { - let preexisting: HashSet = - list_matching(client, prefix).await?.into_iter().collect(); + let preexisting: HashSet = list_matching(client, prefix) + .await? + .into_iter() + .map(|c| c.collection_name) + .collect(); println!( "Serverless registry: prefix={prefix:?} slots={collections_count} preexisting={}", preexisting.len() ); - for name in preexisting.iter().take(5) { + let mut shown: Vec<&String> = preexisting.iter().collect(); + shown.sort(); + for name in shown.iter().take(5) { println!(" preexisting: {name}"); } if preexisting.len() > 5 { @@ -66,10 +121,12 @@ impl CollectionRegistry { Ok(Self { prefix: prefix.to_string(), preexisting, + ready: Mutex::new(HashSet::new()), created: Mutex::new(HashSet::new()), - queryable: Mutex::new(HashSet::new()), create_config, - create_lock: AsyncMutex::new(()), + slot_locks: (0..collections_count) + .map(|_| AsyncMutex::new(())) + .collect(), }) } @@ -77,29 +134,20 @@ impl CollectionRegistry { collection_name(&self.prefix, index) } - /// Ensure collection `index` exists, creating it lazily on first use. - /// Returns the collection name. + /// Ensure collection `index` exists with a compatible vector shape, + /// creating it lazily on first use. Returns the collection name. pub async fn ensure(&self, client: &QdrantServerless, index: usize) -> Result { let name = self.name(index); - // Fast path: already known to exist (preexisting or created this run). - { - let created = self.created.lock().unwrap(); - if self.preexisting.contains(&name) || created.contains(&name) { - return Ok(name); - } + if self.ready.lock().unwrap().contains(&name) { + return Ok(name); } - // Slow path: serialize creates so parallel upserts of the same slot - // do not all race `create_collection`. - let _guard = self.create_lock.lock().await; + let _guard = self.slot_locks[index].lock().await; - // Re-check under the lock. - { - let created = self.created.lock().unwrap(); - if self.preexisting.contains(&name) || created.contains(&name) { - return Ok(name); - } + // Re-check under the slot lock. + if self.ready.lock().unwrap().contains(&name) { + return Ok(name); } let info = client @@ -108,32 +156,65 @@ impl CollectionRegistry { .with_context(|| format!("get_collection {name}"))?; if info.exists { - // Appeared between bootstrap and now (another process / prior run). - // Do not claim we created it. - return Ok(name); + // Preexisting, or appeared between bootstrap and now (another + // process / prior run). Do not claim we created it, but make sure + // the upload config fits it before sending points. + if let Some(existing) = &info.config { + check_vector_shape(&name, existing, &self.create_config)?; + } + if !self.preexisting.contains(&name) { + println!("Collection {name} appeared during the run; reusing it"); + } + } else { + client + .create_collection(&name, self.create_config.clone()) + .await + .with_context(|| format!("create_collection {name}"))?; + self.created.lock().unwrap().insert(name.clone()); + println!("Created collection {name}"); } - client - .create_collection(&name, self.create_config.clone()) - .await - .with_context(|| format!("create_collection {name}"))?; - self.created.lock().unwrap().insert(name.clone()); - println!("Created collection {name}"); + self.ready.lock().unwrap().insert(name.clone()); Ok(name) } - pub fn mark_queryable(&self, name: &str) { - self.queryable.lock().unwrap().insert(name.to_string()); - } - pub fn summary(&self) { let created = self.created.lock().unwrap(); - let queryable = self.queryable.lock().unwrap(); + let ready = self.ready.lock().unwrap(); println!( - "Serverless registry summary: preexisting={} created={} queryable={}", + "Serverless registry summary: preexisting={} created={} used={}", self.preexisting.len(), created.len(), - queryable.len() + ready.len() ); } } + +#[cfg(test)] +mod tests { + use super::*; + use qdrant_client::serverless::{DenseVectorConfig, Distance, SparseVectorConfig}; + + fn dense(size: u64) -> CollectionConfig { + CollectionConfig::new().dense_vector(DenseVectorConfig::new(size, Distance::Cosine)) + } + + #[test] + fn matching_shape_passes_and_extra_vectors_are_fine() { + let existing = dense(128).named_sparse_vector("text", SparseVectorConfig::new()); + check_vector_shape("c", &existing, &dense(128)).unwrap(); + } + + #[test] + fn size_mismatch_is_reported() { + let err = check_vector_shape("c", &dense(128), &dense(256)).unwrap_err(); + assert!(err.to_string().contains("128d"), "{err}"); + } + + #[test] + fn missing_sparse_vector_is_reported() { + let wanted = dense(128).named_sparse_vector("text", SparseVectorConfig::new()); + let err = check_vector_shape("c", &dense(128), &wanted).unwrap_err(); + assert!(err.to_string().contains("sparse vector \"text\""), "{err}"); + } +} diff --git a/src/serverless/convert.rs b/src/serverless/convert.rs index 87ffb2c..5dc10eb 100644 --- a/src/serverless/convert.rs +++ b/src/serverless/convert.rs @@ -2,7 +2,10 @@ //! //! Serverless only accepts the tenant-facing shape (dense/sparse vectors + //! payload indexes). Storage knobs from the upload YAML (HNSW, quantization, -//! on-disk placement, …) are ignored — the serverless manager decides those. +//! on-disk placement, datatype, …) are ignored — the serverless manager +//! decides those. The serverless `precision_tier` is deliberately left unset +//! (service default): there is no faithful mapping from BFB's quantization / +//! datatype settings onto the tiers. use anyhow::{Result, bail}; use qdrant_client::serverless::{ diff --git a/src/serverless/mod.rs b/src/serverless/mod.rs index 0b757b5..a8411a0 100644 --- a/src/serverless/mod.rs +++ b/src/serverless/mod.rs @@ -2,12 +2,12 @@ //! //! Unlike the regular single-collection workflow, serverless mode spreads work //! across a *range* of collections (one per tenant). Collections are created -//! lazily on first upsert; the registry tracks which existed before the run, -//! which were created during upload, and which are queryable afterwards. +//! lazily on first upsert; the registry tracks which existed before the run +//! and which were created during upload. //! //! ```text //! bfb serverless upload --collection-prefix benchmark- --collections-count 100 \ -//! --distribution uniform --total-points 10M --config-file config.yaml +//! --distribution uniform --total-points 10M --example serverless-upload //! bfb serverless clear --collection-prefix benchmark- //! bfb serverless query --collection-prefix benchmark- --distribution zipf -n 10k //! ``` diff --git a/src/serverless/query.rs b/src/serverless/query.rs index dbc5e16..f90292e 100644 --- a/src/serverless/query.rs +++ b/src/serverless/query.rs @@ -1,4 +1,11 @@ //! `bfb serverless query` — route searches across existing collections. +//! +//! Requests are generated by the same [`ConfigSearchGenerator`] as +//! `bfb search`, either from a search YAML (`--file` / `--example`) or from a +//! template derived from a matching collection's vector config. Each request +//! picks a collection according to `--distribution` and sends +//! `--search-batch-size` queries to it; the run itself (parallelism, `--rps`, +//! stats, `--jsonl-*`, `--json`) is driven by the shared [`process`] loop. use std::sync::Arc; use std::sync::Mutex; @@ -6,65 +13,257 @@ use std::sync::atomic::{AtomicBool, Ordering}; use std::time::Instant; use anyhow::{Context, Result, bail}; -use futures::stream::StreamExt; -use indicatif::{MultiProgress, ProgressBar, ProgressDrawTarget, ProgressStyle}; -use qdrant_client::qdrant::{QueryPointsBuilder, VectorInput}; -use rand::Rng; -use rand::RngExt; +use indicatif::ProgressBar; +use qdrant_client::qdrant::{ + IdfParamsBuilder, Query, QueryBatchPointsBuilder, QueryPointsBuilder, SearchParamsBuilder, + VectorInput, +}; +use qdrant_client::serverless::{CollectionConfig, QdrantServerless}; use super::args::ServerlessQueryArgs; -use super::client::{self, retry_with_clients}; +use super::client::create_clients; use super::collections::list_matching; use super::distribution::CollectionPicker; use crate::args::Args; -use crate::generators::random::random_dense_vector; -use crate::processor::Timing; -use crate::stats::{print_stats, throttler}; -use qdrant_client::serverless::CollectionConfig; - -/// Shape used when no search YAML is provided: taken from a live collection. -struct InferredShape { - /// `(vector_name, size)` — empty name is the default unnamed vector. - dense: Vec<(String, u64)>, +use crate::client::retry_with_clients; +use crate::config::examples::{ExampleKind, resolve}; +use crate::config::search::{SearchCollectionConfig, SearchConfig, SearchRequestConfig}; +use crate::config::{DatatypeKind, SparseSource, VectorSource}; +use crate::generators::ConfigSearchGenerator; +use crate::generators::queries::GeneratedQuery; +use crate::processor::{Processor, Timing}; +use crate::results::BenchmarkResults; +use crate::stats::process; + +#[derive(Default)] +struct QueryStats { + server_timings: Vec, + full_timings: Vec, + qps: Vec, + rps: Vec, } -impl InferredShape { - fn from_config(config: &CollectionConfig) -> Result { - let mut dense: Vec<(String, u64)> = config - .dense_vectors - .iter() - .map(|(name, cfg)| (name.clone(), cfg.size)) - .collect(); - dense.sort_by(|a, b| a.0.cmp(&b.0)); - if dense.is_empty() { - bail!("collection has no dense vectors to query"); - } - Ok(Self { dense }) - } +struct ServerlessQueryProcessor { + args: Args, + stopped: Arc, + clients: Vec, + names: Vec, + picker: CollectionPicker, + generator: ConfigSearchGenerator, + start_timestamp_millis: f64, + start_time: Instant, + stats: Mutex, +} - fn random_query(&self, rng: &mut impl Rng) -> (Option, Vec) { - let idx = rng.random_range(0..self.dense.len()); - let (name, size) = &self.dense[idx]; - let vector = random_dense_vector(rng, *size as usize, false); - let using = if name.is_empty() { - None +impl ServerlessQueryProcessor { + fn build_query( + &self, + generated: GeneratedQuery, + collection: &str, + ) -> Result { + let (vector, using) = if let Some((values, indices, name)) = generated.sparse { + (VectorInput::new_sparse(indices.data, values), Some(name)) + } else if let Some((values, name)) = generated.dense { + (VectorInput::new_dense(values), name) } else { - Some(name.clone()) + bail!("search config request must produce a dense or sparse vector"); }; - (using, vector) + + let mut params = SearchParamsBuilder::default().exact(self.args.search_exact); + if let Some(hnsw_ef) = self.args.search_hnsw_ef { + params = params.hnsw_ef(hnsw_ef as u64); + } + if let Some(corpus) = generated.idf_corpus { + params = params.idf(IdfParamsBuilder::default().corpus(corpus)); + } + + let mut builder = QueryPointsBuilder::new(collection) + .query(Query::new_nearest(vector)) + .params(params) + .limit(self.args.search_limit as u64) + .with_payload(self.args.search_with_payload) + .with_vectors(self.args.search_with_vectors); + if let Some(name) = using.filter(|n| !n.is_empty()) { + builder = builder.using(name); + } + if let Some(filter) = generated.filter { + builder = builder.filter(filter); + } + Ok(builder) + } + + async fn search(&self, req_id: usize, progress_bar: &ProgressBar) -> Result<()> { + if self.stopped.load(Ordering::Relaxed) { + return Ok(()); + } + + let mut rng = rand::rng(); + let collection = &self.names[self.picker.pick(&mut rng)]; + let template_idx = self.generator.random_template_idx(&mut rng); + + let queries = (0..self.args.search_batch_size) + .map(|_| { + let generated = self + .generator + .make_query_for(template_idx, req_id, &mut rng); + self.build_query(generated, collection).map(|b| b.build()) + }) + .collect::>>()?; + + let mut request = QueryBatchPointsBuilder::new(collection, queries); + if let Some(timeout) = self.args.timeout { + request = request.timeout(timeout as u64); + } + let request = request.build(); + + let start = Instant::now(); + let res = retry_with_clients(&self.clients, &self.args, |c| { + c.query_batch(request.clone()) + }) + .await + .with_context(|| format!("query on {collection}"))?; + let elapsed = start.elapsed().as_secs_f32(); + let delay_millis = self.start_time.elapsed().as_millis() as u32; + + if res.time > self.args.timing_threshold { + progress_bar.println(format!("Slow query on {collection}: {:?}", res.time)); + } + + let mut stats = self.stats.lock().unwrap(); + stats.server_timings.push(Timing { + delay_millis, + value: res.time as f32, + }); + stats.full_timings.push(Timing { + delay_millis, + value: elapsed, + }); + stats.qps.push(Timing { + delay_millis, + value: progress_bar.per_sec() as f32, + }); + stats.rps.push(Timing { + delay_millis, + value: (progress_bar.per_sec() / self.args.search_batch_size as f64) as f32, + }); + Ok(()) + } +} + +impl Processor for ServerlessQueryProcessor { + async fn make_request( + &self, + req_id: usize, + _args: &Args, + progress_bar: &ProgressBar, + ) -> Result<()> { + self.search(req_id, progress_bar).await + } + + fn start_timestamp_millis(&self) -> f64 { + self.start_timestamp_millis + } + + fn server_timings(&self) -> Vec { + self.stats.lock().unwrap().server_timings.clone() + } + + fn qps(&self) -> Vec { + self.stats.lock().unwrap().qps.clone() + } + + fn rps(&self) -> Vec { + self.stats.lock().unwrap().rps.clone() + } + + fn full_timings(&self) -> Vec { + self.stats.lock().unwrap().full_timings.clone() + } + + fn get_batch_size(&self) -> usize { + self.args.search_batch_size + } +} + +/// Derive a search config from a live collection: one random-query template +/// per dense and sparse vector, no filters. +fn infer_search_config(name: &str, config: &CollectionConfig) -> Result { + let mut requests = Vec::new(); + + let mut dense: Vec<_> = config.dense_vectors.iter().collect(); + dense.sort_by(|a, b| a.0.cmp(b.0)); + for (vector, cfg) in dense { + requests.push(SearchRequestConfig::Dense { + using: Some(vector.clone()).filter(|n| !n.is_empty()), + size: cfg.size, + datatype: DatatypeKind::default(), + source: VectorSource::Random, + filters: Vec::new(), + }); + } + + let mut sparse: Vec<_> = config.sparse_vectors.keys().collect(); + sparse.sort(); + for vector in sparse { + requests.push(SearchRequestConfig::Sparse { + using: vector.clone(), + source: SparseSource::default(), + filters: Vec::new(), + idf_corpus: Vec::new(), + }); + } + + if requests.is_empty() { + bail!("collection {name} has no vectors to query"); } + Ok(SearchConfig { + collection: SearchCollectionConfig { + name: name.to_string(), + }, + requests, + }) +} + +async fn fetch_config(client: &QdrantServerless, name: &str) -> Result { + let info = client + .get_collection(name) + .await + .with_context(|| format!("get_collection {name}"))?; + if !info.exists { + bail!("collection {name} no longer exists"); + } + info.config.ok_or_else(|| { + anyhow::anyhow!("collection {name} has no config; re-upload or pass a search config") + }) } pub async fn run(args: &Args, query: ServerlessQueryArgs, stopped: Arc) -> Result<()> { - let clients = client::create_clients(args)?; - let names = list_matching(&clients[0], &query.collection_prefix).await?; - if names.is_empty() { + let clients = create_clients(args)?; + let summaries = list_matching(&clients[0], &query.collection_prefix).await?; + if summaries.is_empty() { bail!( "no collections matched prefix {:?} — run `bfb serverless upload` first", query.collection_prefix ); } + let (names, empty): (Vec<_>, Vec<_>) = summaries + .into_iter() + .partition(|c| c.point_count != Some(0)); + let names: Vec = names.into_iter().map(|c| c.collection_name).collect(); + if !empty.is_empty() { + println!( + "Skipping {} empty collection(s) matching the prefix", + empty.len() + ); + } + if names.is_empty() { + bail!( + "every collection matching prefix {:?} is empty — run `bfb serverless upload` first", + query.collection_prefix + ); + } + println!( "Querying {} collection(s) with prefix {:?} ({:?})", names.len(), @@ -72,160 +271,81 @@ pub async fn run(args: &Args, query: ServerlessQueryArgs, stopped: Arc3}] {pos}/{len} (eta:{eta})") - .expect("progress style"), - ); - bar.set_draw_target(ProgressDrawTarget::stdout_with_hz(2)); - let bar = Arc::new(bar); + results.results.search = Some(process(&args, stopped, processor).await?); + results.write_if_requested(&args) +} - let server_timings = Arc::new(Mutex::new(Vec::::new())); - let full_timings = Arc::new(Mutex::new(Vec::::new())); - let rps_timings = Arc::new(Mutex::new(Vec::::new())); - let start = Instant::now(); +#[cfg(test)] +mod tests { + use super::*; + use qdrant_client::serverless::{DenseVectorConfig, Distance, SparseVectorConfig}; - let parallel = if args.rps.is_some() { - 1 - } else { - args.parallel.max(1) - }; - let throttler = throttler(args.rps.map(|r| r as f32).or(args.throttle)); - let stopped_flag = stopped.clone(); - let clients = Arc::new(clients); - let names = Arc::new(names); - let shape = Arc::new(shape); - let args_owned = args.clone(); - - futures::stream::iter(0..num_queries) - .take_while(|_| { - let s = stopped_flag.clone(); - async move { !s.load(Ordering::Relaxed) } - }) - .zip(throttler) - .for_each_concurrent(parallel, |(req_id, _)| { - let clients = clients.clone(); - let names = names.clone(); - let shape = shape.clone(); - let bar = bar.clone(); - let server_timings = server_timings.clone(); - let full_timings = full_timings.clone(); - let rps_timings = rps_timings.clone(); - let args = args_owned.clone(); - let stopped = stopped_flag.clone(); - let picker = picker.clone(); - - async move { - if stopped.load(Ordering::Relaxed) { - return; - } - - let mut rng = rand::rng(); - let coll_idx = picker.pick(&mut rng); - let collection = names[coll_idx].clone(); - let (using, vector) = shape.random_query(&mut rng); - - let mut builder = QueryPointsBuilder::new(collection.clone()) - .query(VectorInput::new_dense(vector)) - .limit(args.search_limit as u64) - .with_payload(args.search_with_payload) - .with_vectors(args.search_with_vectors); - if let Some(name) = using { - builder = builder.using(name); - } - if let Some(timeout) = args.timeout { - builder = builder.timeout(timeout as u64); - } - let request = builder.build(); - - let req_start = Instant::now(); - let res = retry_with_clients(&clients, &args, |c| c.query(request.clone())).await; - let full = req_start.elapsed().as_secs_f32(); - - match res { - Ok(resp) => { - let delay = start.elapsed().as_millis() as u32; - server_timings.lock().unwrap().push(Timing { - delay_millis: delay, - value: resp.time as f32, - }); - full_timings.lock().unwrap().push(Timing { - delay_millis: delay, - value: full, - }); - // Instantaneous RPS estimate from inter-arrival of completions. - let elapsed = start.elapsed().as_secs_f32().max(1e-6); - rps_timings.lock().unwrap().push(Timing { - delay_millis: delay, - value: (req_id as f32 + 1.0) / elapsed, - }); - if resp.time > args.timing_threshold { - bar.println(format!("Slow query on {collection}: {:?}", resp.time)); - } - bar.inc(1); - } - Err(e) => { - bar.println(format!("query failed on {collection}: {e:?}")); - if !args.ignore_errors { - stopped.store(true, Ordering::Relaxed); - } - } - } + #[test] + fn infers_one_template_per_vector() { + let config = CollectionConfig::new() + .dense_vector(DenseVectorConfig::new(64, Distance::Dot)) + .named_dense_vector("image", DenseVectorConfig::new(512, Distance::Cosine)) + .named_sparse_vector("text", SparseVectorConfig::new()); + let search = infer_search_config("c", &config).unwrap(); + assert_eq!(search.collection.name, "c"); + assert_eq!(search.requests.len(), 3); + match &search.requests[0] { + SearchRequestConfig::Dense { using, size, .. } => { + assert_eq!(using.as_deref(), None); + assert_eq!(*size, 64); } - }) - .await; - - bar.finish_and_clear(); - let elapsed = start.elapsed().as_secs_f64(); - println!( - "Serverless query finished in {elapsed:.2}s ({:.0} qps wall)", - num_queries as f64 / elapsed.max(1e-9) - ); - - let mut server = server_timings.lock().unwrap().clone(); - let mut full = full_timings.lock().unwrap().clone(); - let mut rps = rps_timings.lock().unwrap().clone(); - print_stats(args, &mut server, "server time", true); - print_stats(args, &mut full, "full time", true); - print_stats(args, &mut rps, "rps", false); - Ok(()) -} + other => panic!("unexpected {other:?}"), + } + match &search.requests[1] { + SearchRequestConfig::Dense { using, size, .. } => { + assert_eq!(using.as_deref(), Some("image")); + assert_eq!(*size, 512); + } + other => panic!("unexpected {other:?}"), + } + assert!( + matches!(&search.requests[2], SearchRequestConfig::Sparse { using, .. } if using == "text") + ); + } -async fn infer_shape( - client: &qdrant_client::serverless::QdrantServerless, - name: &str, -) -> Result { - let info = client - .get_collection(name) - .await - .with_context(|| format!("get_collection {name}"))?; - if !info.exists { - bail!("collection {name} no longer exists"); + #[test] + fn vectorless_collection_is_rejected() { + assert!(infer_search_config("c", &CollectionConfig::new()).is_err()); } - let config = info.config.ok_or_else(|| { - anyhow::anyhow!("collection {name} has no config; re-upload or pass a search config") - })?; - InferredShape::from_config(&config) } diff --git a/src/serverless/upload.rs b/src/serverless/upload.rs index fe63fb2..909377a 100644 --- a/src/serverless/upload.rs +++ b/src/serverless/upload.rs @@ -1,25 +1,151 @@ //! `bfb serverless upload` — spread points across lazily-created collections. use std::cmp::min; -use std::sync::Arc; use std::sync::atomic::{AtomicBool, Ordering}; -use std::time::Instant; +use std::sync::{Arc, Mutex}; +use std::time::{Duration, Instant}; use anyhow::{Context, Result}; -use futures::stream::StreamExt; +use futures::stream::{Stream, StreamExt}; use indicatif::{MultiProgress, ProgressBar, ProgressDrawTarget, ProgressStyle}; use qdrant_client::qdrant::UpsertPointsBuilder; +use qdrant_client::serverless::QdrantServerless; use tokio::time::sleep; +use tokio_stream::wrappers::IntervalStream; use super::args::ServerlessUploadArgs; -use super::client::{self, retry_with_clients}; +use super::client::create_clients; use super::collections::CollectionRegistry; use super::convert::to_serverless_config; use super::distribution::CollectionPicker; use crate::args::Args; +use crate::client::retry_with_clients; use crate::config; +use crate::config::examples::{ExampleKind, resolve}; +use crate::dataset; use crate::generators::{ConfigGenerator, PointGenerator}; -use crate::stats::throttler; +use crate::processor::Timing; +use crate::results::{BenchmarkResults, UploadPhase}; +use crate::save_jsonl::save_timings_as_jsonl; +use crate::stats::{print_stats, throttler}; + +/// One upsert request: `count` points of collection slot `coll_idx`, with +/// point ids starting at `start_id` (unique across the whole upload). +#[derive(Debug, Clone, Copy)] +struct Batch { + coll_idx: usize, + start_id: u64, + count: usize, +} + +/// Split the per-collection allocation into upsert batches. Point ids are +/// laid out contiguously across collections starting at `offset`, so with a +/// dataset source every collection gets a different slice of the data. +fn plan_batches(per_collection: &[usize], batch_size: usize, offset: usize) -> Vec { + let mut batches = Vec::new(); + let mut next_id = offset as u64; + for (coll_idx, &count) in per_collection.iter().enumerate() { + let mut remaining = count; + while remaining > 0 { + let n = min(batch_size, remaining); + batches.push(Batch { + coll_idx, + start_id: next_id, + count: n, + }); + next_id += n as u64; + remaining -= n; + } + } + batches +} + +/// Request pacing: `--rps` fires at a fixed interval regardless of how many +/// requests are in flight (missed ticks are skipped, not burst); otherwise +/// the regular `--throttle` stream. +fn pacer(args: &Args) -> Box + Unpin> { + match args.rps.filter(|rps| *rps > 0.0 && rps.is_finite()) { + Some(rps) => { + let mut interval = tokio::time::interval(Duration::from_secs_f64(1.0 / rps)); + interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip); + Box::new(IntervalStream::new(interval).map(|_| ())) + } + None => throttler(args.throttle), + } +} + +/// Records the first failure of a run; later ones are only logged. +#[derive(Default)] +struct FirstError(Mutex>); + +impl FirstError { + fn record(&self, err: anyhow::Error) { + let mut slot = self.0.lock().unwrap(); + if slot.is_none() { + *slot = Some(err); + } + } + + fn into_result(self) -> Result<()> { + match self.0.into_inner().unwrap() { + Some(err) => Err(err), + None => Ok(()), + } + } +} + +struct Uploader { + args: Args, + clients: Vec, + registry: CollectionRegistry, + generator: Box, + bar: ProgressBar, + started: Instant, + /// Server-reported upsert latency per batch. + timings: Mutex>, + /// Points acknowledged by the server. + uploaded: Mutex, +} + +impl Uploader { + async fn upsert(&self, batch_no: usize, batch: Batch) -> Result<()> { + let client = &self.clients[batch_no % self.clients.len()]; + let name = self.registry.ensure(client, batch.coll_idx).await?; + + let points: Vec<_> = (0..batch.count as u64) + .map(|i| self.generator.make_point(batch.start_id + i)) + .collect(); + + let mut request = + UpsertPointsBuilder::new(name.clone(), points).wait(self.args.wait_on_upsert); + if let Some(timeout) = self.args.timeout { + request = request.timeout(timeout as u64); + } + let request = request.build(); + + let resp = retry_with_clients(&self.clients, &self.args, |c| { + c.upsert_points(request.clone()) + }) + .await + .with_context(|| format!("upsert into {name}"))?; + + self.timings.lock().unwrap().push(Timing { + delay_millis: self.started.elapsed().as_millis() as u32, + value: resp.time as f32, + }); + *self.uploaded.lock().unwrap() += batch.count; + if resp.time > self.args.timing_threshold { + self.bar + .println(format!("Slow upsert on {name}: {:?}", resp.time)); + } + self.bar.inc(batch.count as u64); + + if let Some(delay_millis) = self.args.delay { + sleep(Duration::from_millis(delay_millis as u64)).await; + } + Ok(()) + } +} pub async fn run( args: &Args, @@ -31,27 +157,38 @@ pub async fn run( "--collections-count must be > 0" ); - let yaml = std::fs::read_to_string(&upload.config_file) - .with_context(|| format!("read config {}", upload.config_file))?; - let upload_config = config::parse(&yaml, &upload.config_file)?; + let resolved = resolve( + upload.config.file.as_deref(), + upload.config.example.as_deref(), + ExampleKind::Upload, + )?; + let upload_config = config::parse(&resolved.yaml, &resolved.origin)?; let serverless_config = to_serverless_config(&upload_config)?; - let total_points = upload.total_points.or(args.num_vectors).unwrap_or(100_000); - - let clients = client::create_clients(args)?; - let registry = Arc::new( - CollectionRegistry::bootstrap( - &clients[0], - &upload.collection_prefix, - upload.collections_count, - serverless_config, - ) - .await?, - ); + // Dataset-backed configs cap the total at what the dataset holds. + let total_points = dataset::resolve_num_vectors( + upload.total_points.or(args.num_vectors), + args.offset, + &upload_config, + &dataset::default_datasets_dir(), + )?; + + let mut args = args.clone(); + args.num_vectors = Some(total_points); + args.collection_name = format!("{}*", upload.collection_prefix); + let mut results = BenchmarkResults::new(&args, Some(resolved.origin)); + + let clients = create_clients(&args)?; + let registry = CollectionRegistry::bootstrap( + &clients[0], + &upload.collection_prefix, + upload.collections_count, + serverless_config, + ) + .await?; let picker = CollectionPicker::new(upload.collections_count, upload.distribution.into())?; - let mut rng = rand::rng(); - let per_collection = picker.allocate(total_points, &mut rng); + let per_collection = picker.allocate(total_points, &mut rand::rng()); println!( "Uploading {total_points} points across {} collections ({:?})", @@ -64,7 +201,8 @@ pub async fn run( println!(" …"); } - let generator: Arc = Arc::new(ConfigGenerator::new(&upload_config)?); + let batches = plan_batches(&per_collection, args.batch_size, args.offset); + let generator: Box = Box::new(ConfigGenerator::new(&upload_config)?); let logger = env_logger::Builder::from_default_env().build(); let multiprogress = MultiProgress::new(); @@ -79,113 +217,98 @@ pub async fn run( .expect("progress style"), ); bar.set_draw_target(ProgressDrawTarget::stdout_with_hz(2)); - let bar = Arc::new(bar); - // Flatten (collection_idx, local_point_id) into a work list of batches. - // Each batch stays within one collection so upserts stay simple. - let mut batches: Vec<(usize, u64, usize)> = Vec::new(); // (coll_idx, start_id, count) - for (coll_idx, &count) in per_collection.iter().enumerate() { - if count == 0 { - continue; - } - let mut remaining = count; - let mut start = 0u64; - while remaining > 0 { - let n = min(args.batch_size, remaining); - batches.push((coll_idx, start, n)); - start += n as u64; - remaining -= n; - } - } - - let started = Instant::now(); - let parallel = if args.rps.is_some() { - // RPS mode: one in-flight stream controlled by the throttler below. - 1 + // `--rps` decides the send rate on its own, so concurrency is unbounded; + // otherwise `-p` bounds the number of in-flight batches. + let concurrency = if args.rps.is_some() { + None } else { - args.parallel.max(1) + Some(args.parallel.max(1)) }; + let pacer = pacer(&args); + let start_timestamp_millis = std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .unwrap() + .as_millis() as f64; - let throttler = throttler(args.rps.map(|r| r as f32).or(args.throttle)); - let stopped_flag = stopped.clone(); - let clients = Arc::new(clients); - let args = args.clone(); + let uploader = Uploader { + args: args.clone(), + clients, + registry, + generator, + bar: bar.clone(), + started: Instant::now(), + timings: Mutex::new(Vec::new()), + uploaded: Mutex::new(0), + }; + let first_error = FirstError::default(); futures::stream::iter(batches.into_iter().enumerate()) - .take_while(|_| { - let s = stopped_flag.clone(); - async move { !s.load(Ordering::Relaxed) } - }) - .zip(throttler) - .for_each_concurrent(parallel, |((batch_no, (coll_idx, start_id, count)), _)| { - let clients = clients.clone(); - let registry = registry.clone(); - let generator = generator.clone(); - let bar = bar.clone(); - let args = args.clone(); - let stopped = stopped_flag.clone(); - + .take_while(|_| futures::future::ready(!stopped.load(Ordering::Relaxed))) + .zip(pacer) + .for_each_concurrent(concurrency, |((batch_no, batch), _)| { + let uploader = &uploader; + let first_error = &first_error; + let stopped = &stopped; + let args = &args; async move { if stopped.load(Ordering::Relaxed) { return; } - - let client = &clients[batch_no % clients.len()]; - let name = match registry.ensure(client, coll_idx).await { - Ok(n) => n, - Err(e) => { - bar.println(format!("ensure collection failed: {e:?}")); - if !args.ignore_errors { - stopped.store(true, Ordering::Relaxed); - } - return; - } - }; - - let mut points = Vec::with_capacity(count); - for i in 0..count { - points.push(generator.make_point(start_id + i as u64)); - } - - let mut request = - UpsertPointsBuilder::new(name.clone(), points).wait(args.wait_on_upsert); - if let Some(timeout) = args.timeout { - request = request.timeout(timeout as u64); - } - let request = request.build(); - - let res = - retry_with_clients(&clients, &args, |c| c.upsert_points(request.clone())).await; - - match res { - Ok(resp) => { - if resp.time > args.timing_threshold { - bar.println(format!("Slow upsert on {name}: {:?}", resp.time)); - } - registry.mark_queryable(&name); - bar.inc(count as u64); - } - Err(e) => { - bar.println(format!("upsert failed on {name}: {e:?}")); - if !args.ignore_errors { - stopped.store(true, Ordering::Relaxed); - } + if let Err(err) = uploader.upsert(batch_no, batch).await { + uploader.bar.println(format!("Error: {err:?}")); + if !args.ignore_errors { + first_error.record(err); + stopped.store(true, Ordering::Relaxed); } } - - if let Some(delay_millis) = args.delay { - sleep(std::time::Duration::from_millis(delay_millis as u64)).await; - } } }) .await; - bar.finish_and_clear(); - let elapsed = started.elapsed().as_secs_f64(); + let duration_secs = uploader.started.elapsed().as_secs_f64(); + if stopped.load(Ordering::Relaxed) { + bar.abandon(); + } else { + bar.finish(); + } + + let uploaded = *uploader.uploaded.lock().unwrap(); + let phase = UploadPhase::new(duration_secs, uploaded); println!( - "Serverless upload finished in {elapsed:.2}s ({:.0} points/s)", - total_points as f64 / elapsed.max(1e-9) + "Uploaded {} points in {:.3} s ({:.0} points/s)", + phase.num_points, phase.duration_secs, phase.points_per_sec ); - registry.summary(); - Ok(()) + uploader.registry.summary(); + + let mut timings = uploader.timings.into_inner().unwrap(); + println!("--- Upsert timings ---"); + print_stats(&args, &mut timings, "upsert time", true); + if let Some(jsonl_path) = &args.jsonl_updates { + save_timings_as_jsonl( + jsonl_path, + args.absolute_time.unwrap_or(false), + &timings, + start_timestamp_millis, + "upsert_latency", + )?; + } + + first_error.into_result()?; + results.results.upload = Some(phase); + results.write_if_requested(&args) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn batches_cover_every_point_with_unique_ids() { + let batches = plan_batches(&[5, 0, 7], 3, 100); + let counts: Vec<_> = batches.iter().map(|b| (b.coll_idx, b.count)).collect(); + assert_eq!(counts, vec![(0, 3), (0, 2), (2, 3), (2, 3), (2, 1)]); + let ids: Vec<_> = batches.iter().map(|b| b.start_id).collect(); + assert_eq!(ids, vec![100, 103, 105, 108, 111]); + } } From 01615b69635f2efd1c734e5fbc5a5a6790870acc Mon Sep 17 00:00:00 2001 From: generall Date: Wed, 2 Sep 2026 15:51:09 +0200 Subject: [PATCH 4/4] Add `bfb serverless list` Lists collections (optionally filtered by --collection-prefix) with their point counts and a total, so a space can be inspected before querying or clearing it. Co-Authored-By: Claude Fable 5.1 Claude-Session: https://claude.ai/code/session_01Qdso7ZDKX1kUnn7VbPpVAE --- README.md | 7 +++++- src/serverless/args.rs | 11 ++++++++++ src/serverless/list.rs | 49 ++++++++++++++++++++++++++++++++++++++++++ src/serverless/mod.rs | 3 +++ 4 files changed, 69 insertions(+), 1 deletion(-) create mode 100644 src/serverless/list.rs diff --git a/README.md b/README.md index a68f165..8f3ab0a 100644 --- a/README.md +++ b/README.md @@ -400,6 +400,11 @@ bfb serverless query \ --collection-prefix benchmark- \ --file search.yaml -n 10k -p 8 +# Show what is there: one line per collection with its point count +bfb serverless list \ + --uri https://serverless.example.cloud.qdrant.io \ + --collection-prefix benchmark- + # Tear down everything with that prefix bfb serverless clear \ --uri https://serverless.example.cloud.qdrant.io \ @@ -408,7 +413,7 @@ bfb serverless clear \ | Flag | Meaning | |------|---------| -| `--collection-prefix` | Shared name prefix (`benchmark-` → `benchmark-0` … `benchmark-99`) | +| `--collection-prefix` | Shared name prefix (`benchmark-` → `benchmark-0` … `benchmark-99`). Optional for `list`, which then shows the whole space | | `--collections-count` | Upload only: how many collection slots to spread points across | | `--distribution` | `uniform` or `zipf` — how points/queries are routed across slots | | `--total-points` | Upload only: total points across all collections (falls back to `-n`) | diff --git a/src/serverless/args.rs b/src/serverless/args.rs index 0e5eb44..6bd4b66 100644 --- a/src/serverless/args.rs +++ b/src/serverless/args.rs @@ -20,6 +20,9 @@ pub enum ServerlessCommand { /// Delete every collection whose name starts with `--collection-prefix`. Clear(ServerlessClearArgs), + /// List collections and their point counts. + List(ServerlessListArgs), + /// Run queries routed across existing collections matching the prefix. Query(ServerlessQueryArgs), } @@ -56,6 +59,14 @@ pub struct ServerlessClearArgs { pub collection_prefix: String, } +#[derive(ClapArgs, Debug, Clone)] +pub struct ServerlessListArgs { + /// Only list collections whose name starts with this prefix. + /// Omit to list every collection in the space. + #[clap(long, default_value = "")] + pub collection_prefix: String, +} + #[derive(ClapArgs, Debug, Clone)] pub struct ServerlessQueryArgs { /// Query every existing collection whose name starts with this prefix. diff --git a/src/serverless/list.rs b/src/serverless/list.rs new file mode 100644 index 0000000..41c600e --- /dev/null +++ b/src/serverless/list.rs @@ -0,0 +1,49 @@ +//! `bfb serverless list` — collections matching a prefix and their point counts. + +use anyhow::Result; + +use super::args::ServerlessListArgs; +use super::client::single_client; +use super::collections::list_matching; +use crate::args::Args; + +pub async fn run(args: &Args, list: ServerlessListArgs) -> Result<()> { + let client = single_client(args)?; + let summaries = list_matching(&client, &list.collection_prefix).await?; + + if summaries.is_empty() { + if list.collection_prefix.is_empty() { + println!("No collections in the space"); + } else { + println!("No collections matched prefix {:?}", list.collection_prefix); + } + return Ok(()); + } + + let width = summaries + .iter() + .map(|c| c.collection_name.len()) + .max() + .unwrap_or(0); + let mut total: u64 = 0; + let mut unknown = 0usize; + for c in &summaries { + match c.point_count { + Some(n) => { + total += n; + println!("{: { + unknown += 1; + println!("{: 0 { + print!(" (+{unknown} with unknown count)"); + } + println!(); + Ok(()) +} diff --git a/src/serverless/mod.rs b/src/serverless/mod.rs index a8411a0..2258f72 100644 --- a/src/serverless/mod.rs +++ b/src/serverless/mod.rs @@ -8,6 +8,7 @@ //! ```text //! bfb serverless upload --collection-prefix benchmark- --collections-count 100 \ //! --distribution uniform --total-points 10M --example serverless-upload +//! bfb serverless list --collection-prefix benchmark- //! bfb serverless clear --collection-prefix benchmark- //! bfb serverless query --collection-prefix benchmark- --distribution zipf -n 10k //! ``` @@ -18,6 +19,7 @@ mod client; mod collections; mod convert; mod distribution; +mod list; mod query; mod upload; @@ -35,6 +37,7 @@ pub async fn run(args: Args, serverless: ServerlessArgs, stopped: Arc upload::run(&args, upload_args, stopped).await, ServerlessCommand::Clear(clear_args) => clear::run(&args, clear_args).await, + ServerlessCommand::List(list_args) => list::run(&args, list_args).await, ServerlessCommand::Query(query_args) => query::run(&args, query_args, stopped).await, } }