From 61693134bece4df76a17fa5d26f262572dcdfd0f Mon Sep 17 00:00:00 2001 From: Pascal Seitz Date: Thu, 9 Jul 2026 17:50:43 +0200 Subject: [PATCH] fix cache flush in aggregations fixes #2992 ``` full terms_7 Memory: 37.2 KB Avg: 2.3958ms (+0.31%) Median: 2.3896ms (+0.18%) [2.3573ms .. 2.5176ms] terms_all_unique Memory: 10.8 MB Avg: 5.5144ms (-1.07%) Median: 5.4625ms (-1.98%) [5.3364ms .. 5.9712ms] terms_all_unique_order_by_key Memory: 10.8 MB Avg: 5.2614ms (-0.85%) Median: 5.2177ms (-1.21%) [5.0823ms .. 5.6316ms] terms_150_000 Memory: 2.7 MB Avg: 5.5335ms (-1.07%) Median: 5.5152ms (-1.06%) [5.4151ms .. 5.9654ms] terms_many_top_1000 Memory: 5.2 MB Avg: 8.3579ms (-1.53%) Median: 8.3604ms (-0.95%) [8.2184ms .. 8.5421ms] terms_many_order_by_term Memory: 2.7 MB Avg: 4.6713ms (-0.07%) Median: 4.6569ms (-0.15%) [4.5994ms .. 4.9115ms] terms_all_unique_with_avg_sub_agg Memory: 54.0 MB Avg: 17.4981ms (-2.43%) Median: 17.6075ms (-1.75%) [15.8166ms .. 18.9250ms] terms_status_with_avg_sub_agg Memory: 90.3 KB Avg: 5.6365ms (+7.77%) Median: 5.6255ms (+7.97%) [5.5489ms .. 5.8254ms] terms_status_with_terms_zipf_1000_sub_agg Memory: 318.5 KB (+56.52%) Avg: 4.4504ms (+11.55%) Median: 4.4436ms (+11.59%) [4.3858ms .. 4.5692ms] terms_zipf_1000_with_terms_status_sub_agg Memory: 684.9 KB Avg: 11.8606ms (+0.19%) Median: 11.8360ms (-0.02%) [11.7478ms .. 12.0609ms] terms_status_with_histogram Memory: 139.5 KB Avg: 2.4524ms (-1.09%) Median: 2.4521ms (-0.23%) [2.4179ms .. 2.5049ms] terms_status_with_date_histogram Memory: 136.7 KB Avg: 2.3407ms (-1.28%) Median: 2.3359ms (-1.06%) [2.3001ms .. 2.4310ms] terms_status_with_date_histogram_hard_bounds Memory: 136.1 KB Avg: 2.5113ms (-2.04%) Median: 2.5073ms (-0.97%) [2.4455ms .. 2.7280ms] terms_status_with_date_histogram_and_sibling_terms Memory: 137.3 KB Avg: 3.8695ms (-0.48%) Median: 3.8653ms (+0.11%) [3.8093ms .. 4.0528ms] terms_zipf_1000 Memory: 69.8 KB Avg: 2.2022ms (-1.95%) Median: 2.2026ms (-1.12%) [2.1705ms .. 2.2859ms] terms_zipf_1000_with_histogram Memory: 1.2 MB Avg: 20.4087ms (-0.02%) Median: 20.3665ms (+0.11%) [20.1912ms .. 20.7933ms] terms_zipf_1000_with_avg_sub_agg Memory: 472.0 KB Avg: 8.7387ms (-3.48%) Median: 8.7043ms (-3.44%) [8.6466ms .. 9.1396ms] terms_zipf_90 Memory: 55.3 KB Avg: 1.3784ms (-2.04%) Median: 1.3787ms (-1.61%) [1.3484ms .. 1.4611ms] terms_zipf_90_with_sum_sub_agg Memory: 367.6 KB Avg: 4.8520ms (+8.43%) Median: 4.8326ms (+8.94%) [4.8058ms .. 5.1278ms] terms_many_json_mixed_type_with_avg_sub_agg Memory: 17.8 MB Avg: 25.0853ms (-7.70%) Median: 25.0591ms (-7.12%) [24.8103ms .. 25.4936ms] terms_status_with_cardinality_agg Memory: 91.8 KB Avg: 3.3667ms (+1.47%) Median: 3.3690ms (+1.66%) [3.3311ms .. 3.4070ms] terms_100_buckets_with_cardinality_agg Memory: 9.9 MB Avg: 48.4768ms (-3.07%) Median: 48.3745ms (-3.38%) [48.1425ms .. 49.5503ms] ``` --- benches/agg_bench.rs | 36 +++++++++++++ src/aggregation/agg_tests.rs | 76 ++++++++++++++++++++++++++++ src/aggregation/buffered_sub_aggs.rs | 32 ++---------- 3 files changed, 115 insertions(+), 29 deletions(-) diff --git a/benches/agg_bench.rs b/benches/agg_bench.rs index d89c9b5bc..e5bb87170 100644 --- a/benches/agg_bench.rs +++ b/benches/agg_bench.rs @@ -73,6 +73,8 @@ fn bench_agg(mut group: InputGroup) { register!(group, terms_zipf_1000); register!(group, terms_zipf_1000_with_histogram); register!(group, terms_zipf_1000_with_avg_sub_agg); + register!(group, terms_zipf_90); + register!(group, terms_zipf_90_with_sum_sub_agg); register!(group, terms_many_json_mixed_type_with_avg_sub_agg); @@ -495,6 +497,27 @@ fn terms_zipf_1000(index: &Index) { execute_agg(index, agg_req); } +fn terms_zipf_90(index: &Index) { + let agg_req = json!({ + "my_texts": { "terms": { "field": "text_90_terms_zipf", "size": 100 } }, + }); + execute_agg(index, agg_req); +} + +// 90-term (low-cardinality Vec path) terms agg with a metric sub-agg. The skewed distribution keeps +// a dominant bucket crossing the sub-agg flush threshold, exercising the buffer flush path. +fn terms_zipf_90_with_sum_sub_agg(index: &Index) { + let agg_req = json!({ + "my_texts": { + "terms": { "field": "text_90_terms_zipf", "size": 100 }, + "aggs": { + "sum_score": { "sum": { "field": "score" } } + } + }, + }); + execute_agg(index, agg_req); +} + fn terms_many_json_mixed_type_with_avg_sub_agg(index: &Index) { let agg_req = json!({ "my_texts": { @@ -762,6 +785,8 @@ fn get_test_index_bench(cardinality: Cardinality) -> tantivy::Result { let text_field_few_terms = schema_builder.add_text_field("text_few_terms", STRING | FAST); let text_field_few_terms_status = schema_builder.add_text_field("text_few_terms_status", STRING | FAST); + let text_field_90_terms_zipf = + schema_builder.add_text_field("text_90_terms_zipf", STRING | FAST); let text_field_1000_terms_zipf = schema_builder.add_text_field("text_1000_terms_zipf", STRING | FAST); let score_fieldtype = tantivy::schema::NumericOptions::default().set_fast(); @@ -800,6 +825,12 @@ fn get_test_index_bench(cardinality: Cardinality) -> tantivy::Result { let terms_1000: Vec = (1..=1000).map(|i| format!("term_{i}")).collect(); let zipf_1000 = rand_distr::Zipf::new(1000.0, 1.1f64).unwrap(); + // 90 terms (< MAX_NUM_TERMS_FOR_VEC), skewed via Zipf so a dominant bucket keeps crossing the + // sub-agg flush threshold while minority buckets stay small. Exercises the low-cardinality + // sub-agg buffer flush path (see issue #2992). + let terms_90: Vec = (1..=90).map(|i| format!("term_{i}")).collect(); + let zipf_90 = rand_distr::Zipf::new(90.0, 1.1f64).unwrap(); + { let mut rng = StdRng::from_seed([1u8; 32]); let mut index_writer = index.writer_with_num_threads(1, 200_000_000)?; @@ -815,6 +846,8 @@ fn get_test_index_bench(cardinality: Cardinality) -> tantivy::Result { let idx_b = zipf_1000.sample(&mut rng) as usize - 1; let term_1000_a = &terms_1000[idx_a]; let term_1000_b = &terms_1000[idx_b]; + let term_90_a = &terms_90[zipf_90.sample(&mut rng) as usize - 1]; + let term_90_b = &terms_90[zipf_90.sample(&mut rng) as usize - 1]; index_writer.add_document(doc!( json_field => json!({"mixed_type": 10.0}), json_field => json!({"mixed_type": 10.0}), @@ -830,6 +863,8 @@ fn get_test_index_bench(cardinality: Cardinality) -> tantivy::Result { text_field_few_terms => "cool", text_field_few_terms_status => log_level_sample_a, text_field_few_terms_status => log_level_sample_b, + text_field_90_terms_zipf => term_90_a.as_str(), + text_field_90_terms_zipf => term_90_b.as_str(), text_field_1000_terms_zipf => term_1000_a.as_str(), text_field_1000_terms_zipf => term_1000_b.as_str(), score_field => 1u64, @@ -866,6 +901,7 @@ fn get_test_index_bench(cardinality: Cardinality) -> tantivy::Result { text_field_many_terms => many_terms_data.choose(&mut rng).unwrap().to_string(), text_field_few_terms => few_terms_data.choose(&mut rng).unwrap().to_string(), text_field_few_terms_status => status_field_data[log_level_distribution.sample(&mut rng)].0, + text_field_90_terms_zipf => terms_90[zipf_90.sample(&mut rng) as usize - 1].as_str(), text_field_1000_terms_zipf => terms_1000[zipf_1000.sample(&mut rng) as usize - 1].as_str(), score_field => val as u64, score_field_f64 => lg_norm.sample(&mut rng), diff --git a/src/aggregation/agg_tests.rs b/src/aggregation/agg_tests.rs index 09d9752e9..49b14898e 100644 --- a/src/aggregation/agg_tests.rs +++ b/src/aggregation/agg_tests.rs @@ -664,6 +664,82 @@ fn test_aggregation_flushing_variants() { test_aggregation_flushing(true, true).unwrap(); } +// Regression test for https://github.com/quickwit-oss/tantivy/issues/2992 +// +// A skewed terms bucket over <100 terms uses the low-cardinality (Vec) sub-agg buffer. A dominant +// term keeps crossing the periodic flush threshold (every 2048 docs), which used to drop the +// cached doc ids of the minority buckets before collecting them — corrupting their metric +// sub-aggregations while their doc counts stayed exact. +#[test] +fn test_terms_sub_agg_flushing_skewed_buckets() -> crate::Result<()> { + use std::collections::HashMap; + + // 89 minority terms + 1 dominant term = 90 distinct terms, staying below + // MAX_NUM_TERMS_FOR_VEC (100) so the low-cardinality Vec sub-agg buffer is used. + const NUM_MINORITY_TERMS: usize = 89; + + let mut values: Vec<(f64, String)> = Vec::new(); + let mut minority_idx = 0usize; + // Enough docs to cross the 2048 flush threshold multiple times. The minority docs are + // sprinkled among the dominant ones so they land in different flush windows. + for i in 0..5000u64 { + if i % 25 == 0 { + let term = format!("minority_{:02}", minority_idx % NUM_MINORITY_TERMS); + minority_idx += 1; + values.push(((i % 13 + 1) as f64, term)); + } else { + values.push((7.0, "dominant".to_string())); + } + } + + let mut truth: HashMap = HashMap::new(); + for (score, term) in &values { + let entry = truth.entry(term.clone()).or_insert((0, 0.0)); + entry.0 += 1; + entry.1 += *score; + } + // Sanity check on the shape of the generated data. + assert_eq!(truth.len(), NUM_MINORITY_TERMS + 1); + + let index = get_test_index_from_values_and_terms(false, &[values])?; + let reader = index.reader()?; + + let agg_req: Aggregations = serde_json::from_value(json!({ + "my_terms": { + "terms": { "field": "string_id", "size": 100 }, + "aggs": { + "sum_score": { "sum": { "field": "score" } } + } + } + })) + .unwrap(); + + let collector = get_collector(agg_req); + let searcher = reader.searcher(); + let agg_res = searcher.search(&AllQuery, &collector)?; + let res: Value = serde_json::from_str(&serde_json::to_string(&agg_res)?)?; + + let buckets = res["my_terms"]["buckets"].as_array().unwrap(); + // size 100 >= 90 distinct terms, so every bucket is returned. + assert_eq!(buckets.len(), truth.len()); + for bucket in buckets { + let key = bucket["key"].as_str().unwrap(); + let (true_count, true_sum) = truth[key]; + assert_eq!( + bucket["doc_count"].as_u64().unwrap(), + true_count, + "doc_count mismatch for {key}" + ); + assert_eq!( + bucket["sum_score"]["value"].as_f64().unwrap(), + true_sum, + "sum sub-agg mismatch for {key}" + ); + } + + Ok(()) +} + #[test] fn test_aggregation_level1_simple() -> crate::Result<()> { let index = get_test_index_2_segments(true)?; diff --git a/src/aggregation/buffered_sub_aggs.rs b/src/aggregation/buffered_sub_aggs.rs index 0e8c76706..c3555c832 100644 --- a/src/aggregation/buffered_sub_aggs.rs +++ b/src/aggregation/buffered_sub_aggs.rs @@ -2,7 +2,6 @@ use std::fmt::Debug; use super::segment_agg_result::SegmentAggregationCollector; use crate::aggregation::agg_data::AggregationsSegmentCtx; -use crate::aggregation::bucket::MAX_NUM_TERMS_FOR_VEC; use crate::aggregation::BucketId; use crate::DocId; @@ -45,7 +44,6 @@ pub trait SubAggBuffer: Debug { &mut self, sub_agg: &mut Box, agg_data: &mut AggregationsSegmentCtx, - force: bool, ) -> crate::Result<()>; } @@ -76,7 +74,7 @@ impl BufferedSubAggs { ) -> crate::Result<()> { if self.num_docs >= FLUSH_THRESHOLD { self.buffer - .flush_local(&mut self.sub_agg_collector, agg_data, false)?; + .flush_local(&mut self.sub_agg_collector, agg_data)?; self.num_docs = 0; } Ok(()) @@ -86,7 +84,7 @@ impl BufferedSubAggs { pub fn flush(&mut self, agg_data: &mut AggregationsSegmentCtx) -> crate::Result<()> { if self.num_docs != 0 { self.buffer - .flush_local(&mut self.sub_agg_collector, agg_data, true)?; + .flush_local(&mut self.sub_agg_collector, agg_data)?; self.num_docs = 0; } self.sub_agg_collector.flush(agg_data)?; @@ -150,7 +148,6 @@ impl SubAggBuffer for HighCardSubAggBuffer { &mut self, sub_agg: &mut Box, agg_data: &mut AggregationsSegmentCtx, - _force: bool, ) -> crate::Result<()> { let mut max_bucket = 0u32; for partition in self.partitions.iter() { @@ -210,34 +207,11 @@ impl SubAggBuffer for LowCardSubAggBuffer { &mut self, sub_agg: &mut Box, agg_data: &mut AggregationsSegmentCtx, - force: bool, ) -> crate::Result<()> { // Pre-aggregated: call collect per bucket. let max_bucket = (self.per_bucket_docs.len() as BucketId).saturating_sub(1); sub_agg.prepare_max_bucket(max_bucket, agg_data)?; - // The threshold above which we flush buckets individually. - // Note: We need to make sure that we don't lock ourselves into a situation where we hit - // the FLUSH_THRESHOLD, but never flush any buckets. (except the final flush) - let mut bucket_treshold = FLUSH_THRESHOLD / (self.per_bucket_docs.len().max(1) * 2); - const _: () = { - // MAX_NUM_TERMS_FOR_VEC threshold is used for term aggregations - // Note: There may be other flexible values, for other aggregations, but we can use the - // const value here as a upper bound. (better than nothing) - let bucket_treshold_limit = FLUSH_THRESHOLD / (MAX_NUM_TERMS_FOR_VEC as usize * 2); - assert!( - bucket_treshold_limit > 0, - "Bucket threshold must be greater than 0" - ); - }; - if force { - bucket_treshold = 0; - } - for (bucket_id, docs) in self - .per_bucket_docs - .iter() - .enumerate() - .filter(|(_, docs)| docs.len() > bucket_treshold) - { + for (bucket_id, docs) in self.per_bucket_docs.iter().enumerate() { sub_agg.collect(bucket_id as BucketId, docs, agg_data)?; }