diff --git a/benches/bool_queries_with_range.rs b/benches/bool_queries_with_range.rs index 32f8eef680..3b7c3f916d 100644 --- a/benches/bool_queries_with_range.rs +++ b/benches/bool_queries_with_range.rs @@ -1,11 +1,11 @@ -use binggan::{black_box, BenchGroup, BenchRunner}; +use binggan::{black_box, BenchRunner}; use rand::prelude::*; use rand::rngs::StdRng; use rand::SeedableRng; use tantivy::collector::{Collector, Count, TopDocs}; use tantivy::query::{Query, QueryParser}; -use tantivy::schema::{Schema, FAST, INDEXED, TEXT}; -use tantivy::{doc, Index, Order, ReloadPolicy, Searcher}; +use tantivy::schema::{Field, Schema, SchemaBuilder, FAST, INDEXED, TEXT}; +use tantivy::{Index, Order, ReloadPolicy, Searcher, TantivyDocument}; #[derive(Clone)] struct BenchIndex { @@ -15,16 +15,303 @@ struct BenchIndex { query_parser: QueryParser, } -fn build_shared_indices(num_docs: usize, p_title_a: f32, distribution: &str) -> BenchIndex { - // Unified schema +const NUMERIC_NUM_VALUES: u64 = 1_000; + +/// Value distribution of a numeric range field. Both distributions are low cardinality (few +/// distinct values, so many docs share each) and start at 0, which lets a target selectivity be +/// turned into concrete `[0, high]` bounds. +#[derive(Clone, Copy)] +enum Distribution { + /// Uniform random in `0..1_000` (~5k docs per value at 5M docs). + Rand, + /// A coarse `doc_id / 10_000` trend with deterministic high/low noise. The noise keeps the + /// timestamp-query shape while ensuring the fast field selects the bitpacked codec. + Asc, +} + +impl Distribution { + /// Distinct values the field takes; the denominator turning a `[0, high]` range into a + /// selectivity. + fn num_values(self) -> u64 { + match self { + Distribution::Rand => NUMERIC_NUM_VALUES, + Distribution::Asc => NUMERIC_NUM_VALUES, + } + } + + /// Inclusive `[low, high]` bounds for a range matching ~`target` of documents. At least one + /// value is always kept, so a target below `1 / num_values` floors there rather than emptying. + fn bounds_for(self, target: f64) -> (u64, u64) { + let count = (target * self.num_values() as f64).round().max(1.0) as u64; + (0, count - 1) + } +} + +const ASC_BLOCK_LEN: usize = 10_000; +const ASC_NOISE_OFFSET: u64 = NUMERIC_NUM_VALUES / 2; + +/// The numeric range fields and the distribution each draws from. Every field has an indexed-only +/// and an indexed+fast variant so range queries can be compared with and without a fast field. +const NUMERIC_FIELDS: [(&str, Distribution); 4] = [ + ("num_rand", Distribution::Rand), + ("num_asc", Distribution::Asc), + ("num_rand_fast", Distribution::Rand), + ("num_asc_fast", Distribution::Asc), +]; + +// The clustered layout ([`TermLayout::Clustered`]) decides each doc by thresholding a smooth, +// slowly-drifting latent field: "a" where the field is high (see [`TitleField::next_token`]). The +// threshold is set from `p(a)` so the marginal is preserved regardless of how bursty the drift is; +// these knobs shape only the *arrangement* of the "a" docs. Values were picked (and checked by +// simulation) to keep the realized marginal within a few % of `p(a)` while giving moderate, +// realistic clustering. +// +/// Correlation length of the coarsest drift octave, as a fraction of the corpus. Finer octaves +/// refine at 1/4 this length each, so bursts span a range of sizes rather than one fixed scale. +/// Small enough that hundreds of bursts fit in the corpus, which is what keeps the marginal stable. +const CLUSTERED_DRIFT_SCALE: f64 = 0.001; +/// Octaves summed to build the drift. More = more scales of structure; 3 is plenty of texture. +const CLUSTERED_OCTAVES: usize = 3; +/// Fraction of the latent field's (unit) variance carried by the smooth drift; the rest is per-doc +/// noise. Higher = burstier (neighbouring docs agree more); lower = closer to uniform. ~0.3 gives +/// moderate over-dispersion (a few×–20× a uniform sprinkle) — clustered, but not pathologically so. +const CLUSTERED_SIGNAL_FRAC: f64 = 0.3; + +/// How the "a"/"b" tokens of a title field are laid out across documents. Both layouts hit the same +/// marginal `p(a)` — they differ only in where the "a" docs land, which is what an intersection +/// actually sees. +#[derive(Clone, Copy, PartialEq)] +enum TermLayout { + /// Each doc is independently "a" with probability `p_a`: the "a" docs are scattered uniformly. + Uniform, + /// The "a" docs arrive in bursts (see [`TitleField::next_token`]). Real attributes rarely + /// sprinkle evenly — a log level spikes, a topic is ingested together — so their matching + /// docs cluster. Bursts are statistically denser regions, not solid runs, and a background + /// floor keeps some "a" docs scattered in between. + Clustered, +} + +impl TermLayout { + /// Short label for group names, distinguishing the two layouts at the same `p_a`/range. + fn tag(self) -> &'static str { + match self { + TermLayout::Uniform => "uniform", + TermLayout::Clustered => "clustered", + } + } +} + +/// Derives a schema-safe title field name from a term probability and layout, e.g. `(0.001, +/// Uniform)` -> `title_0_1pct` and `(0.001, Clustered)` -> `title_clustered_0_1pct`. A title field +/// is populated with token "a" for a `p_a` fraction of documents, so `:a` matches ~`p_a` of +/// them. +fn title_field_name(p_a: f64, layout: TermLayout) -> String { + let pct = format_pct(p_a).replace('.', "_").replace('%', "pct"); + match layout { + TermLayout::Uniform => format!("title_{}", pct), + TermLayout::Clustered => format!("title_clustered_{}", pct), + } +} + +/// One AR(1) octave of the clustered-intensity drift: a unit-variance value that each step relaxes +/// toward 0 by `rho` and is nudged by fresh noise. High `rho` = slow drift (long, coarse bursts); +/// low `rho` = fast wiggle (short detail). Summing octaves with different `rho` yields bursts at a +/// range of sizes. +struct Ar1Octave { + rho: f64, + state: f64, +} + +/// A title field plus the generator for its "a"/"b" token stream. Produced tokens hit the marginal +/// `p(a)` under either [`TermLayout`]; see [`Self::next_token`]. +struct TitleField { + field: Field, + layout: TermLayout, + /// Target marginal probability of "a". + p_a: f64, + // --- Clustered layout only (unused for Uniform) --- + /// Latent-field value a doc must exceed to be "a". Set to `Φ⁻¹(1 - p_a)` for the unit-variance + /// field below, so `P(exceed) = p_a` and the marginal is `p_a` however bursty the drift is. + threshold: f64, + /// Amplitude per octave, so the summed drift carries `CLUSTERED_SIGNAL_FRAC` of the unit + /// variance. + octave_weight: f64, + /// s.d. of the per-doc noise carrying the remaining `1 - CLUSTERED_SIGNAL_FRAC` of the + /// variance. + noise_sd: f64, + octaves: Vec, +} + +impl TitleField { + fn new(p_a: f64, layout: TermLayout, field: Field, num_docs: usize) -> Self { + // Octave j has correlation length CLUSTERED_DRIFT_SCALE * num_docs / 4^j; rho = e^(-1/len) + // is the per-doc retention producing that length. Splitting the signal variance + // equally across octaves (weight = sqrt(signal/octaves)) plus noise variance (1 - + // signal) makes the latent field unit-variance, so the threshold is a plain + // standard-normal quantile. + let octaves = (0..CLUSTERED_OCTAVES) + .map(|j| { + let len = (CLUSTERED_DRIFT_SCALE * num_docs as f64 / 4f64.powi(j as i32)).max(1.0); + Ar1Octave { + rho: (-1.0 / len).exp(), + state: 0.0, + } + }) + .collect(); + TitleField { + field, + layout, + p_a, + threshold: inverse_normal_cdf(1.0 - p_a), + octave_weight: (CLUSTERED_SIGNAL_FRAC / CLUSTERED_OCTAVES as f64).sqrt(), + noise_sd: (1.0 - CLUSTERED_SIGNAL_FRAC).sqrt(), + octaves, + } + } + + /// Draws this document's token. Uniform: independently "a" with probability `p_a`. Clustered: + /// advance the drift, add per-doc noise to get a unit-variance latent value, and emit "a" when + /// it clears `threshold`. Because the drift moves slowly, neighbouring docs land on the + /// same side of the threshold together — that is the clustering; the noise softens the + /// boundary and scatters a background of "a" docs through the cold stretches (and "b" docs + /// through the hot ones). + fn next_token(&mut self, rng: &mut StdRng) -> &'static str { + match self.layout { + TermLayout::Uniform => { + if rng.random_bool(self.p_a) { + "a" + } else { + "b" + } + } + TermLayout::Clustered => { + let mut latent = self.noise_sd * standard_normal(rng); + for octave in &mut self.octaves { + let noise = (1.0 - octave.rho * octave.rho).sqrt() * standard_normal(rng); + octave.state = octave.rho * octave.state + noise; + latent += self.octave_weight * octave.state; + } + if latent > self.threshold { + "a" + } else { + "b" + } + } + } + } +} + +/// A standard-normal sample via Box–Muller, so the clustered drift needs no extra dependency. +fn standard_normal(rng: &mut StdRng) -> f64 { + let u1 = 1.0 - rng.random::(); // shift to (0,1] so ln is finite + let u2 = rng.random::(); + (-2.0 * u1.ln()).sqrt() * (2.0 * std::f64::consts::PI * u2).cos() +} + +/// Inverse standard-normal CDF (the probit / quantile function) via Acklam's rational +/// approximation, accurate to ~1e-9 over `p ∈ (0, 1)` — enough to place the clustered-layout +/// threshold. Only called once per field, so speed is irrelevant. +fn inverse_normal_cdf(p: f64) -> f64 { + // Coefficients for the central and tail regions of the approximation. + const A: [f64; 6] = [ + -3.969683028665376e+01, + 2.209460984245205e+02, + -2.759285104469687e+02, + 1.383577518672690e+02, + -3.066479806614716e+01, + 2.506628277459239e+00, + ]; + const B: [f64; 5] = [ + -5.447609879822406e+01, + 1.615858368580409e+02, + -1.556989798598866e+02, + 6.680131188771972e+01, + -1.328068155288572e+01, + ]; + const C: [f64; 6] = [ + -7.784894002430293e-03, + -3.223964580411365e-01, + -2.400758277161838e+00, + -2.549732539343734e+00, + 4.374664141464968e+00, + 2.938163982698783e+00, + ]; + const D: [f64; 4] = [ + 7.784695709041462e-03, + 3.224671290700398e-01, + 2.445134137142996e+00, + 3.754408661907416e+00, + ]; + const P_LOW: f64 = 0.02425; + if p < P_LOW { + let q = (-2.0 * p.ln()).sqrt(); + (((((C[0] * q + C[1]) * q + C[2]) * q + C[3]) * q + C[4]) * q + C[5]) + / ((((D[0] * q + D[1]) * q + D[2]) * q + D[3]) * q + 1.0) + } else if p <= 1.0 - P_LOW { + let q = p - 0.5; + let r = q * q; + (((((A[0] * r + A[1]) * r + A[2]) * r + A[3]) * r + A[4]) * r + A[5]) * q + / (((((B[0] * r + B[1]) * r + B[2]) * r + B[3]) * r + B[4]) * r + 1.0) + } else { + let q = (-2.0 * (1.0 - p).ln()).sqrt(); + -(((((C[0] * q + C[1]) * q + C[2]) * q + C[3]) * q + C[4]) * q + C[5]) + / ((((D[0] * q + D[1]) * q + D[2]) * q + D[3]) * q + 1.0) + } +} + +/// The numeric field handles, in an indexed and an indexed+fast variant of both distributions. +struct NumericFields { + rand: Field, + asc: Field, + rand_fast: Field, + asc_fast: Field, +} + +impl NumericFields { + fn add(builder: &mut SchemaBuilder) -> Self { + NumericFields { + rand: builder.add_u64_field("num_rand", INDEXED), + asc: builder.add_u64_field("num_asc", INDEXED), + rand_fast: builder.add_u64_field("num_rand_fast", INDEXED | FAST), + asc_fast: builder.add_u64_field("num_asc_fast", INDEXED | FAST), + } + } + + fn add_to(&self, doc: &mut TantivyDocument, rng: &mut StdRng, doc_id: usize) { + // The fast and non-fast variants must hold identical data to be comparable. + let rand = rng.random_range(0u64..NUMERIC_NUM_VALUES); + // A perfectly ascending field selects the blockwise-linear codec. Add deterministic + // high/low noise across the value domain so bitpacking wins, without consuming RNG state + // or losing the coarse ascending trend. + let noise = ((doc_id as u64).wrapping_mul(0x9E37_79B9_7F4A_7C15) >> 63) * ASC_NOISE_OFFSET; + let asc = (doc_id / ASC_BLOCK_LEN) as u64 + noise; + doc.add_u64(self.rand, rand); + doc.add_u64(self.asc, asc); + doc.add_u64(self.rand_fast, rand); + doc.add_u64(self.asc_fast, asc); + } +} + +/// Builds one shared index holding a title field for each requested `(p(a), layout)` plus the +/// numeric range fields. Building once lets all scenarios query the same corpus instead of each +/// rebuilding a near-identical index. +fn build_shared_index(num_docs: usize, term_specs: &[(f64, TermLayout)]) -> BenchIndex { let mut schema_builder = Schema::builder(); - let f_title = schema_builder.add_text_field("title", TEXT); - let f_num_rand = schema_builder.add_u64_field("num_rand", INDEXED); - let f_num_asc = schema_builder.add_u64_field("num_asc", INDEXED); - let f_num_rand_fast = schema_builder.add_u64_field("num_rand_fast", INDEXED | FAST); - let f_num_asc_fast = schema_builder.add_u64_field("num_asc_fast", INDEXED | FAST); - let schema = schema_builder.build(); - let index = Index::create_in_ram(schema.clone()); + // One title field per distinct (probability, layout); querying `:a` matches ~p of + // documents. + let mut title_fields: Vec = Vec::new(); + for &(p_a, layout) in term_specs { + if title_fields + .iter() + .any(|tf| tf.p_a == p_a && tf.layout == layout) + { + continue; // same (probability, layout) -> same field, build it once + } + let field = schema_builder.add_text_field(&title_field_name(p_a, layout), TEXT); + title_fields.push(TitleField::new(p_a, layout, field, num_docs)); + } + let numeric = NumericFields::add(&mut schema_builder); + let index = Index::create_in_ram(schema_builder.build()); // Populate index with stable RNG for reproducibility. let mut rng = StdRng::from_seed([7u8; 32]); @@ -32,58 +319,14 @@ fn build_shared_indices(num_docs: usize, p_title_a: f32, distribution: &str) -> { let mut writer = index.writer_with_num_threads(1, 4_000_000_000).unwrap(); - match distribution { - "dense" => { - for doc_id in 0..num_docs { - // Always add title to avoid empty documents - let title_token = if rng.random_bool(p_title_a as f64) { - "a" - } else { - "b" - }; - - let num_rand = rng.random_range(0u64..1000u64); - - let num_asc = (doc_id / 10000) as u64; - - writer - .add_document(doc!( - f_title=>title_token, - f_num_rand=>num_rand, - f_num_asc=>num_asc, - f_num_rand_fast=>num_rand, - f_num_asc_fast=>num_asc, - )) - .unwrap(); - } - } - "sparse" => { - for doc_id in 0..num_docs { - // Always add title to avoid empty documents - let title_token = if rng.random_bool(p_title_a as f64) { - "a" - } else { - "b" - }; - - let num_rand = rng.random_range(0u64..10000000u64); - - let num_asc = doc_id as u64; - - writer - .add_document(doc!( - f_title=>title_token, - f_num_rand=>num_rand, - f_num_asc=>num_asc, - f_num_rand_fast=>num_rand, - f_num_asc_fast=>num_asc, - )) - .unwrap(); - } - } - _ => { - panic!("Unsupported distribution type"); + for doc_id in 0..num_docs { + let mut doc = TantivyDocument::default(); + for title_field in &mut title_fields { + let token = title_field.next_token(&mut rng); + doc.add_text(title_field.field, token); } + numeric.add_to(&mut doc, &mut rng, doc_id); + writer.add_document(doc).unwrap(); } writer.commit().unwrap(); } @@ -96,154 +339,205 @@ fn build_shared_indices(num_docs: usize, p_title_a: f32, distribution: &str) -> .unwrap(); let searcher = reader.searcher(); - // Build query parser for title field - let qp_title = QueryParser::for_index(&index, vec![f_title]); + // Queries always qualify their field, so the default fields are irrelevant. + let query_parser = + QueryParser::for_index(&index, title_fields.iter().map(|tf| tf.field).collect()); BenchIndex { index, searcher, - query_parser: qp_title, + query_parser, + } +} + +/// Formats a fraction as a percentage, with enough precision to stay legible for both coarse +/// (`50%`) and highly selective (`0.0001%`) ranges. +fn format_pct(frac: f64) -> String { + let pct = frac * 100.0; + if pct >= 1.0 { + format!("{:.0}%", pct) + } else if pct >= 0.1 { + format!("{:.1}%", pct) + } else { + format!("{:.4}%", pct) } } fn main() { - // Prepare corpora with varying scenarios - let scenarios = vec![ - ( - "dense and 0.1% a".to_string(), - 5_000_000, - 0.001, - "dense", - 0, - 9, - ), - ("dense and 1% a".to_string(), 5_000_000, 0.01, "dense", 0, 9), - ("dense and 10% a".to_string(), 5_000_000, 0.1, "dense", 0, 9), - ( - "dense and 50% a".to_string(), - 5_000_000, - 0.5, - "dense", - 0, - 500, - ), - ( - "sparse and 50% a".to_string(), - 5_000_000, - 0.99, - "sparse", - 0, - 9, - ), + // Two independent knobs, both surfaced in the group name: `p(a)` (term selectivity, drives the + // title field) and a target range selectivity (drives the numeric range bounds, computed per + // field to hit it). Kept as an explicit paired list rather than a full matrix so the combos + // stay curated, including the 18% term / 0.2% range pair that reproduces the regression. + let scenarios = [ + (0.0001, 0.8), + (0.001, 0.8), + (0.01, 0.8), + (0.1, 0.8), + (0.5, 0.8), + (0.7, 0.8), + (0.0001, 0.1), + (0.001, 0.1), + (0.01, 0.1), + (0.1, 0.1), + (0.5, 0.1), + (0.7, 0.1), + // Keeps the seek heuristic near its point-lookup/scan boundary. + (0.0001, 0.0001), + (0.001, 0.0001), + (0.01, 0.0001), + (0.1, 0.0001), + (0.18, 0.0001), + (0.5, 0.0001), + (0.7, 0.0001), ]; - let mut runner = BenchRunner::new(); - for (scenario_id, num_docs, p_title_a, num_rand_distribution, range_low, range_high) in - scenarios - { - // Build index for this scenario - let bench_index = build_shared_indices(num_docs, p_title_a, num_rand_distribution); - - // Create benchmark group - let mut group = runner.new_group(); + // Each scenario is run under both term layouts: `uniform` ("a" scattered evenly) and + // `clustered` ("a" arriving in organic bursts, same marginal `p(a)`). Real corpora look + // like the latter, and where the matching docs land is exactly what an intersection's + // block-skipping sees. + let layouts = [TermLayout::Uniform, TermLayout::Clustered]; - // Now set the name (this moves scenario_id) - group.set_name(scenario_id); - - // Define all four field types - let field_names = ["num_rand", "num_asc", "num_rand_fast", "num_asc_fast"]; - - let term = "a"; + // Build a single shared corpus with a title field for each (probability, layout). + let mut term_specs: Vec<(f64, TermLayout)> = Vec::new(); + for &layout in &layouts { + for &(p_a, _) in &scenarios { + if !term_specs.iter().any(|&(p, l)| p == p_a && l == layout) { + term_specs.push((p_a, layout)); + } + } + } + let bench_index = build_shared_index(5_000_000, &term_specs); - // Generate all combinations of term and field names - let mut queries = Vec::new(); - for &field_name in &field_names { - let query_str = format!( - "{} AND {}:[{} TO {}]", - term, field_name, range_low, range_high + // Precompute each group's name and queries once; every collector runner reuses them. Layouts + // are interleaved per scenario so a scenario's `uniform` and `clustered` groups sit next to + // each other in the output for easy comparison. + let mut prepared: Vec<(String, Vec)> = Vec::new(); + for &(p_a, range_sel) in &scenarios { + for &layout in &layouts { + let group_name = format!( + "{} a {}, {} range", + format_pct(p_a), + layout.tag(), + format_pct(range_sel) ); - queries.push((query_str, field_name.to_string())); + prepared.push(( + group_name, + build_term_and_range_queries(p_a, range_sel, layout), + )); } + } - let query_str = format!( - "{}:[{} TO {}] AND {}:[{} TO {}]", - "num_rand_fast", range_low, range_high, "num_asc_fast", range_low, range_high - ); - queries.push((query_str, "num_asc_fast".to_string())); - - // Run all benchmark tasks for each query and its corresponding field name - for (query_str, field_name) in queries { - run_benchmark_tasks(&mut group, &bench_index, &query_str, &field_name); + // range∩range queries use no term, so they belong in their own groups keyed by range + // selectivity only (one per distinct value) rather than under a misleading `p(a)` group. + let mut range_sels: Vec = Vec::new(); + for &(_, range_sel) in &scenarios { + if !range_sels.contains(&range_sel) { + range_sels.push(range_sel); } - - group.run(); } -} + for range_sel in range_sels { + let group_name = format!("{} range, range∩range", format_pct(range_sel)); + prepared.push((group_name, vec![build_range_intersection_query(range_sel)])); + } -/// Run all benchmark tasks for a given query string and field name -fn run_benchmark_tasks( - bench_group: &mut BenchGroup, - bench_index: &BenchIndex, - query_str: &str, - field_name: &str, -) { - // Test count - add_bench_task(bench_group, bench_index, query_str, Count, "count"); - - // Test all results - add_bench_task( - bench_group, - bench_index, - query_str, - (Count, TopDocs::with_limit(1000).order_by_score()), - "all_results", + // A separate runner per collector type: the collector heads the output section (via the runner + // name) instead of being repeated in every task name. + run_collector( + BenchRunner::with_name("count"), + &bench_index, + &prepared, + false, + |_| Count, + ); + run_collector( + BenchRunner::with_name("cnt+top_score"), + &bench_index, + &prepared, + false, + |_| (Count, TopDocs::with_limit(100).order_by_score()), + ); + run_collector( + BenchRunner::with_name("top100_asc"), + &bench_index, + &prepared, + true, + |field| TopDocs::with_limit(100).order_by_fast_field::(field.to_string(), Order::Asc), + ); + run_collector( + BenchRunner::with_name("top100_desc"), + &bench_index, + &prepared, + true, + |field| TopDocs::with_limit(100).order_by_fast_field::(field.to_string(), Order::Desc), ); +} - // Test top 100 by the field (if it's a FAST field) - if field_name.ends_with("_fast") { - // Ascending order - { - let collector_name = format!("top100_by_{}_asc", field_name); - let field_name_owned = field_name.to_string(); - add_bench_task( - bench_group, - bench_index, - query_str, - TopDocs::with_limit(100).order_by_fast_field::(field_name_owned, Order::Asc), - &collector_name, - ); - } +/// `(query_str, label, field_name)`: `query_str` is parsed and executed, `label` names the task, +/// `field_name` is the field used to pick fast-field collectors. +type Query3 = (String, String, String); - // Descending order - { - let collector_name = format!("top100_by_{}_desc", field_name); - let field_name_owned = field_name.to_string(); - add_bench_task( - bench_group, - bench_index, - query_str, - TopDocs::with_limit(100).order_by_fast_field::(field_name_owned, Order::Desc), - &collector_name, - ); - } - } +/// `title:a AND field:[range]` for each numeric field, at the scenario's range selectivity, using +/// the title field of the given layout. The range selectivity and layout are stated in the group +/// name, so labels only carry the field. +fn build_term_and_range_queries(p_a: f64, range_sel: f64, layout: TermLayout) -> Vec { + let title_field = title_field_name(p_a, layout); + NUMERIC_FIELDS + .iter() + .map(|&(field_name, dist)| { + let (low, high) = dist.bounds_for(range_sel); + let query_str = format!("{}:a AND {}:[{} TO {}]", title_field, field_name, low, high); + let label = format!("a_AND_{}", field_name); + (query_str, label, field_name.to_string()) + }) + .collect() +} + +/// Intersects the two fast range fields (`num_rand_fast AND num_asc_fast`) at `range_sel`. No term +/// is involved, hence its own group. +fn build_range_intersection_query(range_sel: f64) -> Query3 { + let (rand_low, rand_high) = Distribution::Rand.bounds_for(range_sel); + let (asc_low, asc_high) = Distribution::Asc.bounds_for(range_sel); + let query_str = format!( + "num_rand_fast:[{} TO {}] AND num_asc_fast:[{} TO {}]", + rand_low, rand_high, asc_low, asc_high + ); + ( + query_str, + "num_rand_fast_AND_num_asc_fast".to_string(), + "num_asc_fast".to_string(), + ) } -fn add_bench_task( - bench_group: &mut BenchGroup, +/// Runs one collector over every group in its own named runner. `make_collector` builds the +/// collector for a query's field; `only_fast` skips non-fast fields (fast-field ordering needs a +/// fast field). +fn run_collector( + mut runner: BenchRunner, bench_index: &BenchIndex, - query_str: &str, - collector: C, - collector_name: &str, -) { - let task_name = format!("{}_{}", query_str.replace(" ", "_"), collector_name); - let query = bench_index.query_parser.parse_query(query_str).unwrap(); - let search_task = SearchTask { - searcher: bench_index.searcher.clone(), - collector, - query, - }; - bench_group.register(task_name, move |_| black_box(search_task.run())); + prepared: &[(String, Vec)], + only_fast: bool, + make_collector: F, +) where + C: Collector + 'static, + F: Fn(&str) -> C, +{ + for (group_name, queries) in prepared { + let mut group = runner.new_group(); + group.set_name(group_name); + for (query_str, label, field_name) in queries { + if only_fast && !field_name.ends_with("_fast") { + continue; + } + let query = bench_index.query_parser.parse_query(query_str).unwrap(); + let search_task = SearchTask { + searcher: bench_index.searcher.clone(), + collector: make_collector(field_name), + query, + }; + group.register(label.clone(), move |_| black_box(search_task.run())); + } + group.run(); + } } struct SearchTask { diff --git a/columnar/src/column/mod.rs b/columnar/src/column/mod.rs index f6a50b45f2..96b9c7ddb4 100644 --- a/columnar/src/column/mod.rs +++ b/columnar/src/column/mod.rs @@ -76,6 +76,14 @@ impl Column { } } + /// Returns the total number of values stored in the column. + /// + /// Unlike [`Self::num_docs`], this counts every value in multivalued columns and does not count + /// documents without a value. + pub fn num_values(&self) -> RowId { + self.values.num_vals() + } + pub fn min_value(&self) -> T { self.values.min_value() } diff --git a/columnar/src/tests.rs b/columnar/src/tests.rs index 891fdfe716..187b2b69e3 100644 --- a/columnar/src/tests.rs +++ b/columnar/src/tests.rs @@ -89,6 +89,7 @@ fn test_dataframe_writer_u64_multivalued() { crate::Cardinality::Multivalued ); assert_eq!(divisor_col.num_docs(), 7); + assert_eq!(divisor_col.num_values(), 6); } #[test] diff --git a/src/query/range_query/fast_field_range_doc_set.rs b/src/query/range_query/fast_field_range_doc_set.rs index 28e4db2148..264f807f51 100644 --- a/src/query/range_query/fast_field_range_doc_set.rs +++ b/src/query/range_query/fast_field_range_doc_set.rs @@ -1,8 +1,28 @@ use core::fmt::Debug; +use std::net::Ipv6Addr; use std::ops::RangeInclusive; use columnar::Column; +/// Maps supported column values into an ordered integer space for range-width estimation. +pub(crate) trait RangeDocSetValue: + Send + Sync + PartialOrd + Copy + Debug + 'static +{ + fn to_u128(self) -> u128; +} + +impl RangeDocSetValue for u64 { + fn to_u128(self) -> u128 { + self as u128 + } +} + +impl RangeDocSetValue for Ipv6Addr { + fn to_u128(self) -> u128 { + u128::from_be_bytes(self.octets()) + } +} + use crate::docset::SeekDangerResult; use crate::{DocId, DocSet, TERMINATED}; @@ -35,7 +55,7 @@ impl VecCursor { fn last_doc(&self) -> Option { self.docs.last().cloned() } - fn is_empty(&self) -> bool { + fn is_consumed(&self) -> bool { self.current().is_none() } } @@ -55,13 +75,26 @@ pub(crate) struct RangeDocSet { /// should load small chunks. When the seeks are small, we can employ the same strategy as on /// a full scan. fetch_horizon: u32, + /// Upper bound for `fetch_horizon`. This can be changed while the docset is running. + /// Must remain greater than zero. + max_fetch_horizon: u32, /// Current batch of loaded docs. loaded_docs: VecCursor, last_seek_pos_opt: Option, + /// Rolling confidence that `seek_danger` targets are clustered: each small hop builds it up to + /// a cap, each large hop erodes it. Once it clears `MIN_RUN_TO_SCAN` the seeks are dense + /// enough that one forward-scanned block serves many targets more cheaply than a point + /// lookup each. Unlike a cumulative distance sum it has no periodic reset, so a sustained + /// dense run keeps scanning while a few isolated jumps only nick it (see + /// [`DocSet::seek_danger`]). + seek_cluster_run: u32, } const DEFAULT_FETCH_HORIZON: u32 = 128; -impl RangeDocSet { +const DEFAULT_MAX_FETCH_HORIZON: u32 = 100_000; +const RANGE_DOCSET_COST_PER_HIT: f64 = 0.5; + +impl RangeDocSet { pub(crate) fn new(value_range: RangeInclusive, column: Column) -> Self { if *value_range.start() > column.max_value() || *value_range.end() < column.min_value() { return Self { @@ -70,7 +103,9 @@ impl RangeDocSet { loaded_docs: VecCursor::new(), next_fetch_start: TERMINATED, fetch_horizon: DEFAULT_FETCH_HORIZON, + max_fetch_horizon: DEFAULT_MAX_FETCH_HORIZON, last_seek_pos_opt: None, + seek_cluster_run: 0, }; } @@ -80,15 +115,41 @@ impl RangeDocSet { loaded_docs: VecCursor::new(), next_fetch_start: 0, fetch_horizon: DEFAULT_FETCH_HORIZON, + max_fetch_horizon: DEFAULT_MAX_FETCH_HORIZON, last_seek_pos_opt: None, + seek_cluster_run: 0, }; range_docset.reset_fetch_range(); range_docset.fetch_block(); range_docset } + /// Estimates matching values by assuming values are uniformly distributed between the + /// column's minimum and maximum. + fn estimated_num_hits(&self) -> f64 { + if self.column.num_values() == 0 { + return 0.0; + } + + let column_min = self.column.min_value().to_u128(); + let column_max = self.column.max_value().to_u128(); + let range_start = (*self.value_range.start()).to_u128().max(column_min); + let range_end = (*self.value_range.end()).to_u128().min(column_max); + if range_start > range_end { + return 0.0; + } + + let column_width = (column_max - column_min) as f64 + 1.0; + let range_width = (range_end - range_start) as f64 + 1.0; + self.column.num_values() as f64 * range_width / column_width + } + + fn set_fetch_horizon(&mut self, fetch_horizon: u32) { + self.fetch_horizon = fetch_horizon.min(self.max_fetch_horizon); + } + fn reset_fetch_range(&mut self) { - self.fetch_horizon = DEFAULT_FETCH_HORIZON; + self.set_fetch_horizon(DEFAULT_FETCH_HORIZON); } /// Returns true if more data could be fetched @@ -96,14 +157,13 @@ impl RangeDocSet { if self.next_fetch_start >= self.column.num_docs() { return; } - const MAX_HORIZON: u32 = 100_000; - while self.loaded_docs.is_empty() { - let finished_to_end = self.fetch_horizon(self.fetch_horizon); + while self.loaded_docs.is_consumed() { + let finished_to_end = self.do_fetch_horizon(); if finished_to_end { break; } // Fetch more data, increase horizon. Horizon only gets reset when doing a seek. - self.fetch_horizon = (self.fetch_horizon * 2).min(MAX_HORIZON); + self.set_fetch_horizon(self.fetch_horizon.saturating_mul(2)); } } @@ -117,11 +177,12 @@ impl RangeDocSet { } /// Fetches a block for docid range [next_fetch_start .. next_fetch_start + HORIZON] - fn fetch_horizon(&mut self, horizon: u32) -> bool { + fn do_fetch_horizon(&mut self) -> bool { + let horizon = self.fetch_horizon; let mut finished_to_end = false; let num_docs = self.column.num_docs(); - let mut fetch_end = self.next_fetch_start + horizon; + let mut fetch_end = self.next_fetch_start.saturating_add(horizon); if fetch_end >= num_docs { fetch_end = num_docs; finished_to_end = true; @@ -129,6 +190,10 @@ impl RangeDocSet { let last_doc = self.loaded_docs.last_doc(); let doc_buffer: &mut Vec = self.loaded_docs.get_cleared_data(); + + // TODO: for very sparse columns (e.g. 0.1%), we could load the values in the column and + // translate them back to docids, instead of starting at the docids. That way we + // should be able to extend fetch_end for cheap. self.column.get_docids_for_value_range( self.value_range.clone(), self.next_fetch_start..fetch_end, @@ -143,9 +208,46 @@ impl RangeDocSet { finished_to_end } + + /// Specialized `fetch_block` for seek_danger. Unlike the regular fetch_block, it does not + /// double the horizon until it finds a hit. + fn fetch_block_seek_danger(&mut self, target: DocId) -> SeekDangerResult { + self.next_fetch_start = self.next_fetch_start.max(target); + + while self.loaded_docs.current().is_some_and(|doc| doc < target) { + self.loaded_docs.next(); + } + + // We want to limit the number of docs we fetch, so we don't scan the whole column if the + // target is far away. + // There are exceptions to this. + // 1. If we intersect term "a" that has a lot of docs with a range query that has few hits, + // we want to scan more, so we don't get called repeatedly with small incremental + // targets. + // 2. If the range column is mostly empty, checking is relatively cheap, so we can scan + // more. + if self.loaded_docs.is_consumed() { + self.do_fetch_horizon(); + } + + match self.loaded_docs.current() { + Some(doc) if doc == target => SeekDangerResult::Found, + Some(doc) => SeekDangerResult::SeekLowerBound(doc), + None if self.next_fetch_start >= self.column.num_docs() => { + SeekDangerResult::SeekLowerBound(TERMINATED) + } + None => { + // Since the caller will use this a new lowerbound this messes up our + // last_seek_pos_opt, distance computation so we artificially set it + // to the returned lower bound, which is the next_fetch_start. + self.last_seek_pos_opt = Some(self.next_fetch_start); + SeekDangerResult::SeekLowerBound(self.next_fetch_start) + } + } + } } -impl DocSet for RangeDocSet { +impl DocSet for RangeDocSet { #[inline] fn advance(&mut self) -> DocId { if let Some(docid) = self.loaded_docs.next() { @@ -169,13 +271,12 @@ impl DocSet for RangeDocSe /// of `DocSet` should support it. /// /// Calling `seek(TERMINATED)` is also legal and is the normal way to consume a `DocSet`. + #[inline(never)] fn seek(&mut self, target: DocId) -> DocId { if self.is_last_seek_distance_large(target) { self.reset_fetch_range(); } - if target > self.next_fetch_start { - self.next_fetch_start = target; - } + self.next_fetch_start = self.next_fetch_start.max(target); let mut doc = self.doc(); debug_assert!(doc <= target); while doc < target { @@ -194,10 +295,71 @@ impl DocSet for RangeDocSe return SeekDangerResult::SeekLowerBound(TERMINATED); } - if self.is_last_seek_distance_large(target) { + // A scan miss can return an actual matching doc as its lower bound. The intersection then + // calls us again with that doc. It is already loaded, and this self-generated hop says + // nothing about the density of the driving docset. + if self.loaded_docs.current() == Some(target) { + self.last_seek_pos_opt = Some(target); + return SeekDangerResult::Found; + } + + let distance_to_last_seek = self + .last_seek_pos_opt + .map(|last_seek_pos| target.saturating_sub(last_seek_pos)) + .unwrap_or(u32::MAX); + // The point lookup is more expensive than scanning forward, so once the seeks look + // clustered we switch to scanning forward instead of doing a point lookup per target. + // + // We can't look into the future to see if the next seek is also small, but if we change + // the API in tantivy to operate on blocks of DocIds or have access to the callers docset, + // we can do a better job here. + // + // A fetch_block over 128 documents costs about as much as ~24 point lookups, putting the + // break-even spacing near five documents. + // + // These values below are empirically evaluated on the bool_queries_with_range benchmark and + // a real dataset. + const SCORE_DELTA_BY_HOP: [i8; 7] = [1, 1, 1, 0, -2, -4, -6]; + const MIN_RUN_TO_SCAN: u32 = 16; + // Cap the so sustained large hops switch back to point lookups. + const MAX_RUN: u32 = 32 * MIN_RUN_TO_SCAN; + let hop_bucket = distance_to_last_seek + .saturating_sub(1) + .min(SCORE_DELTA_BY_HOP.len() as u32 - 1) as usize; + let score_delta = SCORE_DELTA_BY_HOP[hop_bucket] as i32; + self.seek_cluster_run = self + .seek_cluster_run + .saturating_add_signed(score_delta) + .min(MAX_RUN); + if self.seek_cluster_run == MIN_RUN_TO_SCAN { + // Start a newly detected dense run with a bounded speculative scan. Productive blocks + // grow the horizon in `fetch_block_seek_danger`. self.reset_fetch_range(); } self.last_seek_pos_opt = Some(target); + if self.seek_cluster_run >= MIN_RUN_TO_SCAN { + return self.fetch_block_seek_danger(target); + } + + // If the target is already in the loaded docs, we can just return Found without doing a + // point lookup. This also leaves the cursor positioned on `target`, so the docset stays in + // a valid state and a following `advance()` resumes the scan right after it. + if self + .loaded_docs + .last_doc() + .map(|doc| doc >= target) + .unwrap_or(false) + { + // Iterate through the loaded docs to find the target or the next doc after it. + while let Some(doc) = self.loaded_docs.current() { + if doc == target { + return SeekDangerResult::Found; + } else if doc > target { + return SeekDangerResult::SeekLowerBound(doc); + } + self.loaded_docs.next(); + } + } let is_match = self .column @@ -224,18 +386,13 @@ impl DocSet for RangeDocSe /// Returns a best-effort hint of the /// cost to drive the docset. fn cost(&self) -> u64 { - // Advancing the docset is pretty expensive since it scans the whole column, there is no - // index currently (will change with an kd-tree) - // Since we use SIMD to scan the fast field range query we lower the cost a little bit, - // assuming that we hit 10% of the docs like in size_hint. - // - // If we would return a cost higher than num_docs, we would never choose ff range query as - // the driver in a DocSet, when intersecting a term query with a fast field. But - // it's the faster choice when the term query has a lot of docids and the range - // query has not. + // We have two cost: + // * a fixed cost of scanning the fast field, which is 10% of the column size + // * another fixed cost of start scanning the column, which is 100 + // * a cost for each hit, we assume a uniform distribution of the values in the column // - // Ideally this would take the fast field codec into account - (self.column.num_docs() as f64 * 0.8) as u64 + 100 + (self.column.num_values() as u64 / 10) + + (self.estimated_num_hits() * RANGE_DOCSET_COST_PER_HIT) as u64 } } @@ -289,6 +446,17 @@ mod tests { RangeDocSet::new(value_range, build_u64_column(num_docs, values_for_doc)) } + #[test] + fn cost_estimates_hits_from_range_and_column_bounds() { + let column = build_u64_column(1_000, |i| vec![(i % 100) as u64]); + + assert_eq!(RangeDocSet::new(20..=39, column.clone()).cost(), 100); + assert_eq!(RangeDocSet::new(50..=50, column.clone()).cost(), 5); + assert_eq!(RangeDocSet::new(0..=99, column.clone()).cost(), 500); + assert_eq!(RangeDocSet::new(90..=200, column.clone()).cost(), 50); + assert_eq!(RangeDocSet::new(200..=300, column).cost(), 0); + } + #[test] fn seek_danger_found_leaves_valid_state() { // Even docs match the range, odd docs do not. @@ -346,6 +514,113 @@ mod tests { assert_eq!(docset.advance(), TERMINATED); } + #[test] + fn seek_danger_keeps_point_lookups_for_selective_targets() { + // Only every tenth document matches. Widely spaced candidates should stay on the point + // lookup path: its miss lower bound is target + 1 rather than the next actual match. + let mut docset = range_docset(1..=1, 5_000, |i| vec![(i % 10 == 0) as u64]); + + for target in [1_001, 2_001, 3_001, 4_001] { + assert_eq!( + docset.seek_danger(target), + SeekDangerResult::SeekLowerBound(target + 1) + ); + } + assert_eq!(docset.seek_cluster_run, 0); + } + + #[test] + fn seek_danger_scan_miss_does_not_expand_past_one_horizon() { + // The range only matches an early prefix. A locally dense burst much later can trigger scan + // mode, but that speculative scan must not expand through the entire empty suffix. + let mut docset = range_docset(1..=1, 5_000, |i| vec![(i < 100) as u64]); + for target in 500..516 { + assert_eq!( + docset.seek_danger(target), + SeekDangerResult::SeekLowerBound(target + 1) + ); + } + + // This call reaches MIN_RUN_TO_SCAN. Only [516, 644) is inspected; 644 is a valid lower + // bound because that entire horizon was proven empty. + assert_eq!( + docset.seek_danger(516), + SeekDangerResult::SeekLowerBound(644) + ); + assert_eq!(docset.next_fetch_start, 644); + assert_eq!(docset.fetch_horizon, 128); + + // The empty first horizon rejects the scan-mode transition and returns to point lookups. + assert_eq!( + docset.seek_danger(644), + SeekDangerResult::SeekLowerBound(645) + ); + } + + #[test] + fn seek_danger_scan_returns_a_real_match_after_exhausted_block() { + let mut docset = range_docset(1..=1, 1_000, |i| vec![(i % 10 == 0) as u64]); + while docset.loaded_docs.current().is_some() { + docset.loaded_docs.next(); + } + assert_eq!(docset.next_fetch_start, 128); + + // Enter scan mode in the known-empty tail of the exhausted block. The bounded seek must + // continue with the next block and return a real match with a valid cursor. + docset.last_seek_pos_opt = Some(124); + docset.seek_cluster_run = 15; + assert_eq!( + docset.seek_danger(125), + SeekDangerResult::SeekLowerBound(130) + ); + assert_eq!(docset.doc(), 130); + assert_eq!(docset.seek_danger(130), SeekDangerResult::Found); + } + + #[test] + fn seek_danger_dense_scan_keeps_growing_fetch_horizon() { + // Every tenth document matches, while the supplied candidates are dense. After enough + // adjacent candidates, seek_danger should switch from target + 1 point-lookup lower bounds + // to a scan returning the next real match. + let mut docset = range_docset(1..=1, 5_000, |i| vec![(i % 10 == 0) as u64]); + for target in 1_000..1_016 { + let _ = docset.seek_danger(target); + } + let first_scanned_match = match docset.seek_danger(1_016) { + SeekDangerResult::SeekLowerBound(doc) => doc, + SeekDangerResult::Found => panic!("1016 does not match"), + }; + assert_eq!(first_scanned_match, 1_020); + + // The lower bound is generated by our own scan. Confirming it must neither look it up nor + // penalize the density score for the ten-document hop. + let score_before_confirmation = docset.seek_cluster_run; + assert_eq!( + docset.seek_danger(first_scanned_match), + SeekDangerResult::Found + ); + assert_eq!(docset.seek_cluster_run, score_before_confirmation); + + // Consume enough scan results to cross a block boundary. The horizon should grow from 128 + // to 256 and then 512; resetting it on every seek_danger call would leave it at 256. + let mut current = first_scanned_match; + while current <= 1_150 { + let mut candidate = current + 1; + loop { + match docset.seek_danger(candidate) { + SeekDangerResult::Found => { + current = candidate; + break; + } + SeekDangerResult::SeekLowerBound(lower_bound) => { + candidate = lower_bound; + } + } + } + } + assert!(docset.fetch_horizon >= 512); + } + #[test] fn seek_danger_matches_seek() { // Cross-check seek_danger against the true next match for every target, on a column with a