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]
```

Author: PSeitz

benches/agg_bench.rs

@@ -73,6 +73,8 @@ fn bench_agg(mut group: InputGroup<Index>) {
     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<Index> {
     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<Index> {
     let terms_1000: Vec<String> = (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<String> = (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<Index> {
             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<Index> {
                 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<Index> {
                 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),

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<String, (u64, f64)> = 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)?;

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<dyn SegmentAggregationCollector>,
         agg_data: &mut AggregationsSegmentCtx,
-        force: bool,
     ) -> crate::Result<()>;
 }
 
@@ -76,7 +74,7 @@ impl<Backend: SubAggBuffer + Debug> BufferedSubAggs<Backend> {
     ) -> 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<Backend: SubAggBuffer + Debug> BufferedSubAggs<Backend> {
     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<dyn SegmentAggregationCollector>,
         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<dyn SegmentAggregationCollector>,
         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)?;
         }