Extend Vec term storage to 20k with eager/lazy bucket ids

Raise MAX_NUM_TERMS_FOR_VEC to 20_000 so the dense Vec term storage
(direct-indexed, no hashing/paging) is used for low/moderate cardinality
both with and without sub-aggregations, not just the <100 low-card case.
Split the old single threshold into MAX_NUM_TERMS_FOR_LOWCARD_SUBAGG (100,
still gates the Vec + LowCard-buffer pairing) and MAX_NUM_TERMS_FOR_VEC.

VecTermBuckets is now generic over a compile-time `const LAZY: bool`:
- eager (default): assigns all ids up front in `new`, branchless term_entry
- lazy: assigns on first occurrence, keeping the sub-agg bucket range down
  to the terms that actually occur (no blowup for sparse ordinal spaces)

A new BucketIdSlot::ASSIGNS_ID const plus the LAZY const fully gate the
first-seen branch out at monomorphization, so the only branchy path is the
deliberately-lazy <BucketId, true> instantiation. The with-sub-agg Vec path
picks eager below MAX_NUM_TERMS_FOR_EAGER_BUCKET_IDS (4096) and lazy above,
avoiding the ~6-11% hot-loop regression eager fixes while keeping large
sparse cases compact.

Biggest impact on terms_zipf_1000_only with -40%
```
full
terms_7                                               Memory: 46.5 KB              Avg: 2.3566ms (+4.29%)      Median: 2.3566ms (+4.29%)      [2.3566ms .. 2.3566ms]
terms_all_unique                                      Memory: 11.5 MB              Avg: 4.8532ms (-5.74%)      Median: 4.8532ms (-5.74%)      [4.8532ms .. 4.8532ms]
terms_all_unique_order_by_key                         Memory: 11.5 MB              Avg: 4.8448ms (-0.53%)      Median: 4.8448ms (-0.53%)      [4.8448ms .. 4.8448ms]
terms_150_000                                         Memory: 2.7 MB               Avg: 5.5295ms (+0.64%)      Median: 5.5295ms (+0.64%)      [5.5295ms .. 5.5295ms]
terms_many_top_1000                                   Memory: 5.3 MB               Avg: 8.4738ms (+0.58%)      Median: 8.4738ms (+0.58%)      [8.4738ms .. 8.4738ms]
terms_many_order_by_term                              Memory: 2.7 MB (-0.01%)      Avg: 4.5959ms (-0.15%)      Median: 4.5959ms (-0.15%)      [4.5959ms .. 4.5959ms]
terms_many_with_top_hits                              Memory: 48.9 MB (-0.00%)     Avg: 85.8206ms (-5.46%)     Median: 85.8206ms (-5.46%)     [85.8206ms .. 85.8206ms]
terms_all_unique_with_avg_sub_agg                     Memory: 54.7 MB (-0.00%)     Avg: 17.8747ms (+8.16%)     Median: 17.8747ms (+8.16%)     [17.8747ms .. 17.8747ms]
terms_many_with_avg_sub_agg                           Memory: 13.5 MB (-0.00%)     Avg: 14.7797ms (+2.54%)     Median: 14.7797ms (+2.54%)     [14.7797ms .. 14.7797ms]
terms_status_with_avg_sub_agg                         Memory: 92.1 KB (-0.35%)     Avg: 5.1467ms (+0.72%)      Median: 5.1467ms (+0.72%)      [5.1467ms .. 5.1467ms]
terms_status_with_terms_zipf_1000_sub_agg             Memory: 213.2 KB             Avg: 3.9576ms (-2.73%)      Median: 3.9576ms (-2.73%)      [3.9576ms .. 3.9576ms]
terms_zipf_1000_with_terms_status_sub_agg             Memory: 726.5 KB (+0.28%)    Avg: 12.1432ms (+3.30%)     Median: 12.1432ms (+3.30%)     [12.1432ms .. 12.1432ms]
terms_status_with_histogram                           Memory: 139.0 KB             Avg: 2.4717ms (+2.45%)      Median: 2.4717ms (+2.45%)      [2.4717ms .. 2.4717ms]
terms_status_with_date_histogram                      Memory: 145.9 KB             Avg: 2.3644ms (-0.44%)      Median: 2.3644ms (-0.44%)      [2.3644ms .. 2.3644ms]
terms_status_with_date_histogram_hard_bounds          Memory: 145.4 KB             Avg: 2.5705ms (+4.70%)      Median: 2.5705ms (+4.70%)      [2.5705ms .. 2.5705ms]
terms_status_with_date_histogram_and_sibling_terms    Memory: 143.3 KB             Avg: 3.8946ms (+1.61%)      Median: 3.8946ms (+1.61%)      [3.8946ms .. 3.8946ms]
terms_zipf_1000_only                                  Memory: 75.8 KB (-0.02%)     Avg: 1.2684ms (-41.50%)     Median: 1.2684ms (-41.50%)     [1.2684ms .. 1.2684ms]
terms_zipf_1000_with_histogram                        Memory: 1.2 MB (+0.17%)      Avg: 20.9323ms (+4.08%)     Median: 20.9323ms (+4.08%)     [20.9323ms .. 20.9323ms]
terms_zipf_1000_with_avg_sub_agg                      Memory: 486.7 KB (+2.95%)    Avg: 8.5297ms (-1.66%)      Median: 8.5297ms (-1.66%)      [8.5297ms .. 8.5297ms]
terms_many_json_mixed_type_with_avg_sub_agg           Memory: 17.9 MB              Avg: 27.0210ms (+1.81%)     Median: 27.0210ms (+1.81%)     [27.0210ms .. 27.0210ms]
terms_status_with_cardinality_agg                     Memory: 93.9 KB              Avg: 3.2022ms (-0.64%)      Median: 3.2022ms (-0.64%)      [3.2022ms .. 3.2022ms]
terms_100_buckets_with_cardinality_agg                Memory: 9.8 MB (+0.31%)      Avg: 49.3188ms (-1.70%)     Median: 49.3188ms (-1.70%)     [49.3188ms .. 49.3188ms]
terms_many_with_single_term_order_by_card             Memory: 48.9 MB              Avg: 77.2127ms (-5.02%)     Median: 77.2127ms (-5.02%)     [77.2127ms .. 77.2127ms]
terms_many_with_single_term_2_order_by_card           Memory: 40.6 MB (-0.02%)     Avg: 45.8004ms (-17.57%)    Median: 45.8004ms (-17.57%)    [45.8004ms .. 45.8004ms]
```

Author: PSeitz

src/aggregation/bucket/histogram/histogram.rs

@@ -256,6 +256,10 @@ impl HistogramBounds {
 /// the zero-sized `()` when it does not. Without sub aggregations the id is never read, so storing
 /// `()` drops the id bytes per bucket and turns id assignment into a no-op.
 pub trait BucketIdSlot: Copy + Default + std::fmt::Debug + PartialEq + 'static {
+    /// Whether this slot carries a real id, i.e. `assign` produces a meaningful value. `false` for
+    /// the `()` slot, letting callers gate id-assignment code out at compile time (e.g. skip the
+    /// lazy-assign branch in a hot loop entirely rather than relying on the optimizer).
+    const ASSIGNS_ID: bool;
     /// Assigns the next id from the provider, called once when a bucket is first filled.
     fn assign(provider: &mut BucketIdProvider) -> Self;
     /// Resolves to the `BucketId` for sub-aggregation bookkeeping.
@@ -265,6 +269,7 @@ pub trait BucketIdSlot: Copy + Default + std::fmt::Debug + PartialEq + 'static {
     fn to_bucket_id(self) -> BucketId;
 }
 impl BucketIdSlot for BucketId {
+    const ASSIGNS_ID: bool = true;
     #[inline(always)]
     fn assign(provider: &mut BucketIdProvider) -> Self {
         provider.next_bucket_id()
@@ -275,6 +280,7 @@ impl BucketIdSlot for BucketId {
     }
 }
 impl BucketIdSlot for () {
+    const ASSIGNS_ID: bool = false;
     #[inline(always)]
     fn assign(_provider: &mut BucketIdProvider) -> Self {}
     #[inline(always)]

src/aggregation/bucket/term_agg/mod.rs

@@ -335,15 +335,47 @@ impl TermsAggregationInternal {
     }
 }
 
-/// The treshold for maximum number of terms to use a Vec-backed bucket storage.
+/// Threshold for using dense `Vec` storage ([`VecTermBuckets`]) for term buckets. Below this many
+/// term ordinals the `Vec` (direct-indexed, no hashing/paging) beats the paged/hashed storages.
+/// It preallocates `num_terms` slots, so the ceiling also bounds its memory
+/// (`num_terms * size_of::<Bucket>`, ≤ ~160KB here).
+///
 /// TODO: Benchmark to validate the threshold
-pub const MAX_NUM_TERMS_FOR_VEC: u64 = 100;
+pub const MAX_NUM_TERMS_FOR_VEC: u64 = 20_000;
+
+/// Threshold below which a terms agg *with sub-aggregations* pairs the `Vec` storage with the
+/// [`LowCardSubAggBuffer`] (which buffers docs in a per-bucket `Vec`). That layout only pays off
+/// for a handful of buckets, so it is far lower than [`MAX_NUM_TERMS_FOR_VEC`]; between the two
+/// thresholds the `Vec` storage pairs with the partitioned `HighCard` buffer instead.
+pub const MAX_NUM_TERMS_FOR_LOWCARD_SUBAGG: u64 = 100;
+
+/// Above this term count, generate sub-aggregation bucket IDs on first use to avoid IDs for unseen
+/// terms.
+///
+/// TODO: Benchmark this threshold.
+const LAZY_BUCKET_ID_GENERATION_THRESHOLD: u64 = 4_096;
 
 /// Average docs-per-bucket below which term counts cluster too tightly (mostly 1s and 2s) for
 /// `select_nth_unstable` to beat `sort_unstable`'s adaptive paths, so we fall back to a full sort.
 /// This is very low on purpose, and meant to catch unique or mostly unique terms.
 const DOCS_PER_BUCKET_QUICKSELECT_THRESHOLD: u64 = 2;
 
+/// Charge a `Vec`-backed term storage's eager, up-front allocation against the memory limit.
+///
+/// The paged/hashed storages grow lazily during `collect` and are charged there via deltas, but
+/// [`VecTermBuckets`] preallocates all `num_terms` slots at construction, so its memory would
+/// otherwise escape the limit entirely (relevant now that the `Vec` threshold reaches
+/// [`MAX_NUM_TERMS_FOR_VEC`]).
+fn add_memory_consumption<M: TermAggregationMap>(
+    term_buckets: &M,
+    req_data: &mut AggregationsSegmentCtx,
+) -> crate::Result<()> {
+    req_data
+        .context
+        .limits
+        .add_memory_consumed(term_buckets.get_memory_consumption() as u64)
+}
+
 /// Build a concrete `SegmentTermCollector` with either a Vec- or HashMap-backed
 /// bucket storage, depending on the column type and aggregation level.
 pub(crate) fn build_segment_term_collector(
@@ -413,39 +445,37 @@ pub(crate) fn build_segment_term_collector(
     // `num_terms`. `saturating_add` guards the HashMap fallback, where `max_column_val` is a raw
     // numeric column value that can reach `u64::MAX`; term ordinals never come close.
     let num_terms = max_column_val.saturating_add(1);
+
+    // Very low cardinality *with* sub aggregations: dense `Vec` storage paired with the `LowCard`
+    // buffer, which groups docs in a per-bucket `Vec`. That layout only fits a handful of buckets
+    // (docs arrive in doc order, so the partitioned `HighCard` buffer would merely merge
+    // consecutive same-bucket docs), so it is gated far below the general `Vec` threshold.
+    if is_top_level && has_sub_aggregations && max_column_val < MAX_NUM_TERMS_FOR_LOWCARD_SUBAGG {
+        let term_buckets = VecTermBuckets::<BucketId>::new(num_terms, &mut bucket_id_provider);
+        add_memory_consumption(&term_buckets, req_data)?;
+        let collector: SegmentTermCollector<_, LowCardSubAggBuffer> = SegmentTermCollector {
+            parent_buckets: vec![term_buckets],
+            sub_agg: sub_agg_collector.map(LowCardBufferedSubAggs::new),
+            bucket_id_provider,
+            max_term_id: max_column_val,
+            terms_req_data,
+        };
+        return Ok(Box::new(collector));
+    }
+
+    // Everything below uses the partitioned `HighCard` sub-agg buffer, so it is built once here.
+    let sub_agg = sub_agg_collector.map(BufferedSubAggs::new);
     if is_top_level && max_column_val < MAX_NUM_TERMS_FOR_VEC {
-        // Low cardinality: dense `Vec` storage. With sub aggregations it pairs with the `LowCard`
-        // buffer, which groups docs in a per-bucket `Vec` — a better fit for the few buckets here
-        // than the partitioned `HighCard` buffer the branches below use (docs arrive in doc order,
-        // so `HighCard` would only merge consecutive same-bucket docs).
+        // Low/moderate cardinality: dense `Vec` storage, direct-indexed by term ordinal — no
+        // hashing or page indirection. Used both with and without sub aggregations.
         if has_sub_aggregations {
-            let term_buckets = VecTermBuckets::<BucketId>::new(num_terms, &mut bucket_id_provider);
-            let collector: SegmentTermCollector<_, LowCardSubAggBuffer> = SegmentTermCollector {
-                parent_buckets: vec![term_buckets],
-                sub_agg: sub_agg_collector.map(LowCardBufferedSubAggs::new),
-                bucket_id_provider,
-                max_term_id: max_column_val,
-                terms_req_data,
-            };
-            Ok(Box::new(collector))
-        } else {
-            let term_buckets = VecTermBuckets::<()>::new(num_terms, &mut bucket_id_provider);
-            Ok(boxed_high_card_collector(
-                term_buckets,
-                None,
-                bucket_id_provider,
-                max_column_val,
-                terms_req_data,
-            ))
-        }
-    } else {
-        // Higher cardinality: every remaining storage uses the partitioned `HighCard` sub-agg
-        // buffer, so it is built once here.
-        let sub_agg = sub_agg_collector.map(BufferedSubAggs::new);
-        if max_column_val < 8_000_000 && is_top_level {
-            if has_sub_aggregations {
+            // Eager id assignment (branchless `term_entry`) while the bucket count is small;
+            // switch to lazy above the threshold to keep the sub-agg bucket range compact for
+            // large/sparse ordinal spaces. See `LAZY_BUCKET_ID_GENERATION_THRESHOLD`.
+            if num_terms > LAZY_BUCKET_ID_GENERATION_THRESHOLD {
                 let term_buckets =
-                    PagedTermMap::<BucketId>::new(num_terms, &mut bucket_id_provider);
+                    VecTermBuckets::<BucketId, true>::new(num_terms, &mut bucket_id_provider);
+                add_memory_consumption(&term_buckets, req_data)?;
                 Ok(boxed_high_card_collector(
                     term_buckets,
                     sub_agg,
@@ -454,7 +484,9 @@ pub(crate) fn build_segment_term_collector(
                     terms_req_data,
                 ))
             } else {
-                let term_buckets = PagedTermMap::<()>::new(num_terms, &mut bucket_id_provider);
+                let term_buckets =
+                    VecTermBuckets::<BucketId, false>::new(num_terms, &mut bucket_id_provider);
+                add_memory_consumption(&term_buckets, req_data)?;
                 Ok(boxed_high_card_collector(
                     term_buckets,
                     sub_agg,
@@ -463,8 +495,22 @@ pub(crate) fn build_segment_term_collector(
                     terms_req_data,
                 ))
             }
-        } else if has_sub_aggregations {
-            let term_buckets = HashMapTermBuckets::<BucketId>::default();
+        } else {
+            // No sub aggregations: the `()` slot carries no id, so eager vs lazy is moot and
+            // `term_entry` is branchless either way.
+            let term_buckets = VecTermBuckets::<()>::new(num_terms, &mut bucket_id_provider);
+            add_memory_consumption(&term_buckets, req_data)?;
+            Ok(boxed_high_card_collector(
+                term_buckets,
+                sub_agg,
+                bucket_id_provider,
+                max_column_val,
+                terms_req_data,
+            ))
+        }
+    } else if is_top_level && max_column_val < 8_000_000 {
+        if has_sub_aggregations {
+            let term_buckets = PagedTermMap::<BucketId>::new(num_terms, &mut bucket_id_provider);
             Ok(boxed_high_card_collector(
                 term_buckets,
                 sub_agg,
@@ -473,7 +519,7 @@ pub(crate) fn build_segment_term_collector(
                 terms_req_data,
             ))
         } else {
-            let term_buckets = HashMapTermBuckets::<()>::default();
+            let term_buckets = PagedTermMap::<()>::new(num_terms, &mut bucket_id_provider);
             Ok(boxed_high_card_collector(
                 term_buckets,
                 sub_agg,
@@ -482,6 +528,24 @@ pub(crate) fn build_segment_term_collector(
                 terms_req_data,
             ))
         }
+    } else if has_sub_aggregations {
+        let term_buckets = HashMapTermBuckets::<BucketId>::default();
+        Ok(boxed_high_card_collector(
+            term_buckets,
+            sub_agg,
+            bucket_id_provider,
+            max_column_val,
+            terms_req_data,
+        ))
+    } else {
+        let term_buckets = HashMapTermBuckets::<()>::default();
+        Ok(boxed_high_card_collector(
+            term_buckets,
+            sub_agg,
+            bucket_id_provider,
+            max_column_val,
+            terms_req_data,
+        ))
     }
 }
 
@@ -731,13 +795,17 @@ impl<B: BucketIdSlot> TermAggregationMap for HashMapTermBuckets<B> {
     }
 }
 
-/// An optimized term map implementation for a compact set of term ordinals.
+/// A term map backed by a `Vec`, indexed directly by term ordinal.
+///
+/// `LAZY_BUCKET_ID_GENERATION` defers sub-aggregation bucket ID generation until first use.
 #[derive(Clone, Debug)]
-struct VecTermBuckets<B> {
+struct VecTermBuckets<B, const LAZY_BUCKET_ID_GENERATION: bool = false> {
     buckets: Vec<Bucket<B>>,
 }
 
-impl<B: BucketIdSlot> TermAggregationMap for VecTermBuckets<B> {
+impl<B: BucketIdSlot, const LAZY_BUCKET_ID_GENERATION: bool> TermAggregationMap
+    for VecTermBuckets<B, LAZY_BUCKET_ID_GENERATION>
+{
     type Slot = B;
 
     const SORTED_BY_ORD: bool = true;
@@ -753,7 +821,7 @@ impl<B: BucketIdSlot> TermAggregationMap for VecTermBuckets<B> {
 
     /// Add an occurrence of the given term id.
     #[inline(always)]
-    fn term_entry(&mut self, term_id: u64, _bucket_id_provider: &mut BucketIdProvider) -> B {
+    fn term_entry(&mut self, term_id: u64, bucket_id_provider: &mut BucketIdProvider) -> B {
         let term_id_usize = term_id as usize;
         debug_assert!(
             term_id_usize < self.buckets.len(),
@@ -762,6 +830,13 @@ impl<B: BucketIdSlot> TermAggregationMap for VecTermBuckets<B> {
             self.buckets.len()
         );
         let bucket = unsafe { self.buckets.get_unchecked_mut(term_id_usize) };
+        // Lazy mode: assign the id the first time a term is seen (`count == 0`). Both
+        // `LAZY_BUCKET_ID_GENERATION` and `B::ASSIGNS_ID` are consts, so in eager mode (or for the
+        // `()` slot) the whole branch is gated out at monomorphization, leaving just the `count`
+        // bump.
+        if LAZY_BUCKET_ID_GENERATION && B::ASSIGNS_ID && bucket.count == 0 {
+            bucket.bucket_id = B::assign(bucket_id_provider);
+        }
         bucket.count += 1;
         bucket.bucket_id
     }
@@ -776,13 +851,17 @@ impl<B: BucketIdSlot> TermAggregationMap for VecTermBuckets<B> {
     }
 
     fn new(num_terms: u64, bucket_id_provider: &mut BucketIdProvider) -> Self {
-        VecTermBuckets {
-            // Assigns an id per term up front from the shared provider; for the `()` slot this is a
-            // no-op and leaves the provider untouched.
-            buckets: std::iter::repeat_with(|| Bucket::new(bucket_id_provider))
-                .take(num_terms as usize)
-                .collect(),
-        }
+        let num_terms = num_terms as usize;
+        let buckets = if LAZY_BUCKET_ID_GENERATION {
+            // Empty buckets; ids are assigned lazily in `term_entry`, so the provider is untouched.
+            vec![Bucket::default(); num_terms]
+        } else {
+            // Eager: assign an id to every ordinal up front (a no-op for the `()` slot).
+            std::iter::repeat_with(|| Bucket::new(bucket_id_provider))
+                .take(num_terms)
+                .collect()
+        };
+        VecTermBuckets { buckets }
     }
 }
 
@@ -2566,6 +2645,69 @@ mod tests {
         Ok(())
     }
 
+    // Eager bucket ids: below LAZY_BUCKET_ID_GENERATION_THRESHOLD but above
+    // MAX_NUM_TERMS_FOR_LOWCARD_SUBAGG, so the `Vec` + `HighCard` buffer path assigns ids eagerly.
+    #[test]
+    fn terms_aggregation_sub_agg_vec_highcard_eager_ids() -> crate::Result<()> {
+        terms_agg_sub_agg_bucket_id_linkage(500)
+    }
+
+    // Lazy bucket ids: above LAZY_BUCKET_ID_GENERATION_THRESHOLD (4096), still below
+    // MAX_NUM_TERMS_FOR_VEC, so the `Vec` + `HighCard` path assigns ids lazily on first occurrence.
+    #[test]
+    fn terms_aggregation_sub_agg_vec_highcard_lazy_ids() -> crate::Result<()> {
+        terms_agg_sub_agg_bucket_id_linkage(5000)
+    }
+
+    /// Builds `num_terms` distinct terms (each with a known score and a varying doc count),
+    /// inserted in reverse term order so any lazily-assigned bucket ids (first-seen order) are
+    /// fully de-correlated from the term ordinals (sorted order). Verifies per-term doc counts
+    /// and metric sub-agg values, guarding the bucket-id -> sub-agg linkage on the `Vec` +
+    /// `HighCard` path.
+    fn terms_agg_sub_agg_bucket_id_linkage(num_terms: u64) -> crate::Result<()> {
+        let mut docs: Vec<(f64, String)> = Vec::new();
+        for k in (0..num_terms).rev() {
+            let count = (k % 3) + 1;
+            for _ in 0..count {
+                docs.push((k as f64, format!("term_{k:06}")));
+            }
+        }
+        let index = get_test_index_from_values_and_terms(false, &[docs])?;
+
+        let agg_req: Aggregations = serde_json::from_value(json!({
+            "my_texts": {
+                "terms": {
+                    "field": "string_id",
+                    "size": num_terms,
+                    "order": { "_key": "asc" }
+                },
+                "aggs": {
+                    "sum_score": { "sum": { "field": "score" } },
+                    "avg_score": { "avg": { "field": "score" } }
+                }
+            }
+        }))
+        .unwrap();
+
+        let res = exec_request(agg_req, &index)?;
+        let buckets = &res["my_texts"]["buckets"];
+        for k in 0..num_terms {
+            let count = (k % 3) + 1;
+            let bucket = &buckets[k as usize];
+            assert_eq!(bucket["key"], format!("term_{k:06}"), "key at {k}");
+            assert_eq!(bucket["doc_count"], count, "doc_count at {k}");
+            assert_eq!(
+                bucket["sum_score"]["value"],
+                (k * count) as f64,
+                "sum at {k}"
+            );
+            assert_eq!(bucket["avg_score"]["value"], k as f64, "avg at {k}");
+        }
+        assert_eq!(res["my_texts"]["sum_other_doc_count"], 0);
+
+        Ok(())
+    }
+
     #[test]
     fn terms_aggregation_different_tokenizer_on_ff_test() -> crate::Result<()> {
         let terms = vec!["Hello Hello", "Hallo Hallo", "Hallo Hallo"];

src/aggregation/bucket/term_agg/term_histogram.rs

@@ -389,9 +389,8 @@ mod tests {
         Ok(())
     }
 
-    /// Term cardinality above the general path's `MAX_NUM_TERMS_FOR_VEC` (100) still fuses: the
-    /// flat grid is bounded by the total cell count (`num_terms * num_time_buckets`), not the
-    /// term count.
+    /// Term cardinality is not what gates fusing: the flat grid is bounded by the total cell count
+    /// (`num_terms * num_time_buckets`), not the term count, so many terms still fuse.
     #[test]
     fn fused_term_histogram_many_terms() -> crate::Result<()> {
         let num_terms = 150usize;