Composite agg merge (#2856)

Add composite aggregation

Co-authored-by: Remi Dettai <remi.dettai@sekoia.io>
Co-authored-by: Paul Masurel <paul.masurel@datadoghq.com>

Cargo.toml

@@ -11,7 +11,7 @@ repository = "https://github.com/quickwit-oss/tantivy"
 readme = "README.md"
 keywords = ["search", "information", "retrieval"]
 edition = "2021"
-rust-version = "1.85"
+rust-version = "1.86"
 exclude = ["benches/*.json", "benches/*.txt"]
 
 [dependencies]

benches/agg_bench.rs

@@ -10,7 +10,7 @@ use tantivy::aggregation::agg_req::Aggregations;
 use tantivy::aggregation::AggregationCollector;
 use tantivy::query::{AllQuery, TermQuery};
 use tantivy::schema::{IndexRecordOption, Schema, TextFieldIndexing, FAST, STRING};
-use tantivy::{doc, Index, Term};
+use tantivy::{doc, DateTime, Index, Term};
 
 #[global_allocator]
 pub static GLOBAL: &PeakMemAlloc<std::alloc::System> = &INSTRUMENTED_SYSTEM;
@@ -70,6 +70,12 @@ fn bench_agg(mut group: InputGroup<Index>) {
 
     register!(group, terms_many_json_mixed_type_with_avg_sub_agg);
 
+    register!(group, composite_term_many_page_1000);
+    register!(group, composite_term_many_page_1000_with_avg_sub_agg);
+    register!(group, composite_term_few);
+    register!(group, composite_histogram);
+    register!(group, composite_histogram_calendar);
+
     register!(group, cardinality_agg);
     register!(group, terms_status_with_cardinality_agg);
 
@@ -314,6 +320,75 @@ fn terms_many_json_mixed_type_with_avg_sub_agg(index: &Index) {
     execute_agg(index, agg_req);
 }
 
+fn composite_term_few(index: &Index) {
+    let agg_req = json!({
+        "my_ctf": {
+            "composite": {
+                "sources": [
+                    { "text_few_terms": { "terms": { "field": "text_few_terms" } } }
+                ],
+                "size": 1000
+            }
+        },
+    });
+    execute_agg(index, agg_req);
+}
+fn composite_term_many_page_1000(index: &Index) {
+    let agg_req = json!({
+        "my_ctmp1000": {
+            "composite": {
+                "sources": [
+                    { "text_many_terms": { "terms": { "field": "text_many_terms" } } }
+                ],
+                "size": 1000
+            }
+        },
+    });
+    execute_agg(index, agg_req);
+}
+fn composite_term_many_page_1000_with_avg_sub_agg(index: &Index) {
+    let agg_req = json!({
+        "my_ctmp1000wasa": {
+            "composite": {
+                "sources": [
+                    { "text_many_terms": { "terms": { "field": "text_many_terms" } } }
+                ],
+                "size": 1000,
+            },
+            "aggs": {
+                "average_f64": { "avg": { "field": "score_f64" } }
+            }
+        },
+    });
+    execute_agg(index, agg_req);
+}
+fn composite_histogram(index: &Index) {
+    let agg_req = json!({
+        "my_ch": {
+            "composite": {
+                "sources": [
+                    { "f64_histogram": { "histogram": { "field": "score_f64", "interval": 1 } } }
+                ],
+                "size": 1000
+            }
+        },
+    });
+    execute_agg(index, agg_req);
+}
+fn composite_histogram_calendar(index: &Index) {
+    let agg_req = json!({
+        "my_chc": {
+            "composite": {
+                "sources": [
+                    { "time_histogram": { "date_histogram": { "field": "timestamp", "calendar_interval": "month" } } }
+                ],
+                "size": 1000
+            }
+        },
+    });
+    execute_agg(index, agg_req);
+}
+
 fn execute_agg(index: &Index, agg_req: serde_json::Value) {
     let agg_req: Aggregations = serde_json::from_value(agg_req).unwrap();
     let collector = get_collector(agg_req);
@@ -496,6 +571,7 @@ fn get_test_index_bench(cardinality: Cardinality) -> tantivy::Result<Index> {
     let text_field_all_unique_terms =
         schema_builder.add_text_field("text_all_unique_terms", STRING | FAST);
     let text_field_many_terms = schema_builder.add_text_field("text_many_terms", STRING | FAST);
+    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_1000_terms_zipf =
@@ -504,6 +580,7 @@ fn get_test_index_bench(cardinality: Cardinality) -> tantivy::Result<Index> {
     let score_field = schema_builder.add_u64_field("score", score_fieldtype.clone());
     let score_field_f64 = schema_builder.add_f64_field("score_f64", score_fieldtype.clone());
     let score_field_i64 = schema_builder.add_i64_field("score_i64", score_fieldtype);
+    let date_field = schema_builder.add_date_field("timestamp", FAST);
     // use tmp dir
     let index = if reuse_index {
         Index::create_in_dir("agg_bench", schema_builder.build())?
@@ -523,6 +600,7 @@ fn get_test_index_bench(cardinality: Cardinality) -> tantivy::Result<Index> {
     let log_level_distribution =
         WeightedIndex::new(status_field_data.iter().map(|item| item.1)).unwrap();
 
+    let few_terms_data = ["INFO", "ERROR", "WARN", "DEBUG"];
     let lg_norm = rand_distr::LogNormal::new(2.996f64, 0.979f64).unwrap();
 
     let many_terms_data = (0..150_000)
@@ -558,6 +636,8 @@ fn get_test_index_bench(cardinality: Cardinality) -> tantivy::Result<Index> {
                 text_field_all_unique_terms => "coolo",
                 text_field_many_terms => "cool",
                 text_field_many_terms => "cool",
+                text_field_few_terms => "cool",
+                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_1000_terms_zipf => term_1000_a.as_str(),
@@ -588,11 +668,13 @@ fn get_test_index_bench(cardinality: Cardinality) -> tantivy::Result<Index> {
                 json_field => json,
                 text_field_all_unique_terms => format!("unique_term_{}", rng.random::<u64>()),
                 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_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),
                 score_field_i64 => val as i64,
+                date_field => DateTime::from_timestamp_millis((val * 1_000_000.) as i64),
             ))?;
             if cardinality == Cardinality::OptionalSparse {
                 for _ in 0..20 {

columnar/src/column_values/mod.rs

@@ -31,7 +31,7 @@ pub use u64_based::{
     serialize_and_load_u64_based_column_values, serialize_u64_based_column_values,
 };
 pub use u128_based::{
-    CompactSpaceU64Accessor, open_u128_as_compact_u64, open_u128_mapped,
+    CompactHit, CompactSpaceU64Accessor, open_u128_as_compact_u64, open_u128_mapped,
     serialize_column_values_u128,
 };
 pub use vec_column::VecColumn;

columnar/src/column_values/u128_based/compact_space/mod.rs

@@ -292,6 +292,19 @@ impl BinarySerializable for IPCodecParams {
     }
 }
 
+/// Represents the result of looking up a u128 value in the compact space.
+///
+/// If a value is outside the compact space, the next compact value is returned.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum CompactHit {
+    /// The value exists in the compact space
+    Exact(u32),
+    /// The value does not exist in the compact space, but the next higher value does
+    Next(u32),
+    /// The value is greater than the maximum compact value
+    AfterLast,
+}
+
 /// Exposes the compact space compressed values as u64.
 ///
 /// This allows faster access to the values, as u64 is faster to work with than u128.
@@ -309,6 +322,11 @@ impl CompactSpaceU64Accessor {
     pub fn compact_to_u128(&self, compact: u32) -> u128 {
         self.0.compact_to_u128(compact)
     }
+
+    /// Finds the next compact space value for a given u128 value.
+    pub fn u128_to_next_compact(&self, value: u128) -> CompactHit {
+        self.0.u128_to_next_compact(value)
+    }
 }
 
 impl ColumnValues<u64> for CompactSpaceU64Accessor {
@@ -441,6 +459,21 @@ impl CompactSpaceDecompressor {
         self.params.compact_space.u128_to_compact(value)
     }
 
+    /// Finds the next compact space value for a given u128 value.
+    pub fn u128_to_next_compact(&self, value: u128) -> CompactHit {
+        match self.u128_to_compact(value) {
+            Ok(compact) => CompactHit::Exact(compact),
+            Err(pos) => {
+                if pos >= self.params.compact_space.ranges_mapping.len() {
+                    CompactHit::AfterLast
+                } else {
+                    let next_range = &self.params.compact_space.ranges_mapping[pos];
+                    CompactHit::Next(next_range.compact_start)
+                }
+            }
+        }
+    }
+
     fn compact_to_u128(&self, compact: u32) -> u128 {
         self.params.compact_space.compact_to_u128(compact)
     }
@@ -823,6 +856,41 @@ mod tests {
         let _data = test_aux_vals(vals);
     }
 
+    #[test]
+    fn test_u128_to_next_compact() {
+        let vals = &[100u128, 200u128, 1_000_000_000u128, 1_000_000_100u128];
+        let mut data = test_aux_vals(vals);
+
+        let _header = U128Header::deserialize(&mut data);
+        let decomp = CompactSpaceDecompressor::open(data).unwrap();
+
+        // Test value that's already in a range
+        let compact_100 = decomp.u128_to_compact(100).unwrap();
+        assert_eq!(
+            decomp.u128_to_next_compact(100),
+            CompactHit::Exact(compact_100)
+        );
+
+        // Test value between two ranges
+        let compact_million = decomp.u128_to_compact(1_000_000_000).unwrap();
+        assert_eq!(
+            decomp.u128_to_next_compact(250),
+            CompactHit::Next(compact_million)
+        );
+
+        // Test value before the first range
+        assert_eq!(
+            decomp.u128_to_next_compact(50),
+            CompactHit::Next(compact_100)
+        );
+
+        // Test value after the last range
+        assert_eq!(
+            decomp.u128_to_next_compact(10_000_000_000),
+            CompactHit::AfterLast
+        );
+    }
+
     use proptest::prelude::*;
 
     fn num_strategy() -> impl Strategy<Value = u128> {

columnar/src/column_values/u128_based/mod.rs

@@ -7,7 +7,7 @@ mod compact_space;
 
 use common::{BinarySerializable, OwnedBytes, VInt};
 pub use compact_space::{
-    CompactSpaceCompressor, CompactSpaceDecompressor, CompactSpaceU64Accessor,
+    CompactHit, CompactSpaceCompressor, CompactSpaceDecompressor, CompactSpaceU64Accessor,
 };
 
 use crate::column_values::monotonic_map_column;

columnar/src/lib.rs

@@ -59,7 +59,7 @@ pub struct RowAddr {
     pub row_id: RowId,
 }
 
-pub use sstable::Dictionary;
+pub use sstable::{Dictionary, TermOrdHit};
 pub type Streamer<'a> = sstable::Streamer<'a, VoidSSTable>;
 
 pub use common::DateTime;

src/aggregation/agg_data.rs

@@ -10,9 +10,10 @@ use crate::aggregation::accessor_helpers::{
 };
 use crate::aggregation::agg_req::{Aggregation, AggregationVariants, Aggregations};
 use crate::aggregation::bucket::{
-    build_segment_filter_collector, build_segment_range_collector, FilterAggReqData,
-    HistogramAggReqData, HistogramBounds, IncludeExcludeParam, MissingTermAggReqData,
-    RangeAggReqData, SegmentHistogramCollector, TermMissingAgg, TermsAggReqData, TermsAggregation,
+    build_segment_filter_collector, build_segment_range_collector, CompositeAggReqData,
+    CompositeAggregation, CompositeSourceAccessors, FilterAggReqData, HistogramAggReqData,
+    HistogramBounds, IncludeExcludeParam, MissingTermAggReqData, RangeAggReqData,
+    SegmentHistogramCollector, TermMissingAgg, TermsAggReqData, TermsAggregation,
     TermsAggregationInternal,
 };
 use crate::aggregation::metric::{
@@ -73,6 +74,12 @@ impl AggregationsSegmentCtx {
         self.per_request.filter_req_data.push(Some(Box::new(data)));
         self.per_request.filter_req_data.len() - 1
     }
+    pub(crate) fn push_composite_req_data(&mut self, data: CompositeAggReqData) -> usize {
+        self.per_request
+            .composite_req_data
+            .push(Some(Box::new(data)));
+        self.per_request.composite_req_data.len() - 1
+    }
 
     #[inline]
     pub(crate) fn get_term_req_data(&self, idx: usize) -> &TermsAggReqData {
@@ -108,6 +115,12 @@ impl AggregationsSegmentCtx {
             .as_deref()
             .expect("range_req_data slot is empty (taken)")
     }
+    #[inline]
+    pub(crate) fn get_composite_req_data(&self, idx: usize) -> &CompositeAggReqData {
+        self.per_request.composite_req_data[idx]
+            .as_deref()
+            .expect("composite_req_data slot is empty (taken)")
+    }
 
     // ---------- mutable getters ----------
 
@@ -181,6 +194,25 @@ impl AggregationsSegmentCtx {
         debug_assert!(self.per_request.filter_req_data[idx].is_none());
         self.per_request.filter_req_data[idx] = Some(value);
     }
+
+    /// Move out the Composite request at `idx`.
+    #[inline]
+    pub(crate) fn take_composite_req_data(&mut self, idx: usize) -> Box<CompositeAggReqData> {
+        self.per_request.composite_req_data[idx]
+            .take()
+            .expect("composite_req_data slot is empty (taken)")
+    }
+
+    /// Put back a Composite request into an empty slot at `idx`.
+    #[inline]
+    pub(crate) fn put_back_composite_req_data(
+        &mut self,
+        idx: usize,
+        value: Box<CompositeAggReqData>,
+    ) {
+        debug_assert!(self.per_request.composite_req_data[idx].is_none());
+        self.per_request.composite_req_data[idx] = Some(value);
+    }
 }
 
 /// Each type of aggregation has its own request data struct. This struct holds
@@ -208,6 +240,8 @@ pub struct PerRequestAggSegCtx {
     pub top_hits_req_data: Vec<TopHitsAggReqData>,
     /// MissingTermAggReqData contains the request data for a missing term aggregation.
     pub missing_term_req_data: Vec<MissingTermAggReqData>,
+    /// CompositeAggReqData contains the request data for a composite aggregation.
+    pub composite_req_data: Vec<Option<Box<CompositeAggReqData>>>,
 
     /// Request tree used to build collectors.
     pub agg_tree: Vec<AggRefNode>,
@@ -255,6 +289,11 @@ impl PerRequestAggSegCtx {
                 .iter()
                 .map(|t| t.get_memory_consumption())
                 .sum::<usize>()
+            + self
+                .composite_req_data
+                .iter()
+                .map(|b| b.as_ref().map(|d| d.get_memory_consumption()).unwrap_or(0))
+                .sum::<usize>()
             + self.agg_tree.len() * std::mem::size_of::<AggRefNode>()
     }
 
@@ -291,6 +330,11 @@ impl PerRequestAggSegCtx {
                 .expect("filter_req_data slot is empty (taken)")
                 .name
                 .as_str(),
+            AggKind::Composite => self.composite_req_data[idx]
+                .as_deref()
+                .expect("composite_req_data slot is empty (taken)")
+                .name
+                .as_str(),
         }
     }
 
@@ -417,6 +461,11 @@ pub(crate) fn build_segment_agg_collector(
         )?)),
         AggKind::Range => Ok(build_segment_range_collector(req, node)?),
         AggKind::Filter => build_segment_filter_collector(req, node),
+        AggKind::Composite => Ok(Box::new(
+            crate::aggregation::bucket::SegmentCompositeCollector::from_req_and_validate(
+                req, node,
+            )?,
+        )),
     }
 }
 
@@ -447,6 +496,7 @@ pub enum AggKind {
     DateHistogram,
     Range,
     Filter,
+    Composite,
 }
 
 impl AggKind {
@@ -462,6 +512,7 @@ impl AggKind {
             AggKind::DateHistogram => "DateHistogram",
             AggKind::Range => "Range",
             AggKind::Filter => "Filter",
+            AggKind::Composite => "Composite",
         }
     }
 }
@@ -709,6 +760,14 @@ fn build_nodes(
                 children,
             }])
         }
+        AggregationVariants::Composite(composite_req) => Ok(vec![build_composite_node(
+            agg_name,
+            reader,
+            segment_ordinal,
+            data,
+            &req.sub_aggregation,
+            composite_req,
+        )?]),
         AggregationVariants::Filter(filter_req) => {
             // Build the query and evaluator upfront
             let schema = reader.schema();
@@ -743,6 +802,35 @@ fn build_nodes(
     }
 }
 
+fn build_composite_node(
+    agg_name: &str,
+    reader: &SegmentReader,
+    _segment_ordinal: SegmentOrdinal,
+    data: &mut AggregationsSegmentCtx,
+    sub_aggs: &Aggregations,
+    req: &CompositeAggregation,
+) -> crate::Result<AggRefNode> {
+    let mut composite_accessors = Vec::with_capacity(req.sources.len());
+    for source in &req.sources {
+        let source_after_key_opt = req.after.get(source.name()).map(|k| &k.0);
+        let source_accessor =
+            CompositeSourceAccessors::build_for_source(reader, source, source_after_key_opt)?;
+        composite_accessors.push(source_accessor);
+    }
+    let agg = CompositeAggReqData {
+        name: agg_name.to_string(),
+        req: req.clone(),
+        composite_accessors,
+    };
+    let idx = data.push_composite_req_data(agg);
+    let children = build_children(sub_aggs, reader, _segment_ordinal, data)?;
+    Ok(AggRefNode {
+        kind: AggKind::Composite,
+        idx_in_req_data: idx,
+        children,
+    })
+}
+
 fn build_children(
     aggs: &Aggregations,
     reader: &SegmentReader,

src/aggregation/agg_req.rs

@@ -32,8 +32,8 @@ use rustc_hash::FxHashMap;
 use serde::{Deserialize, Serialize};
 
 use super::bucket::{
-    DateHistogramAggregationReq, FilterAggregation, HistogramAggregation, RangeAggregation,
-    TermsAggregation,
+    CompositeAggregation, DateHistogramAggregationReq, FilterAggregation, HistogramAggregation,
+    RangeAggregation, TermsAggregation,
 };
 use super::metric::{
     AverageAggregation, CardinalityAggregationReq, CountAggregation, ExtendedStatsAggregation,
@@ -134,6 +134,9 @@ pub enum AggregationVariants {
     /// Filter documents into a single bucket.
     #[serde(rename = "filter")]
     Filter(FilterAggregation),
+    /// Multi-dimensional, paginable bucket aggregation.
+    #[serde(rename = "composite")]
+    Composite(CompositeAggregation),
 
     // Metric aggregation types
     /// Computes the average of the extracted values.
@@ -180,6 +183,11 @@ impl AggregationVariants {
             AggregationVariants::Histogram(histogram) => vec![histogram.field.as_str()],
             AggregationVariants::DateHistogram(histogram) => vec![histogram.field.as_str()],
             AggregationVariants::Filter(filter) => filter.get_fast_field_names(),
+            AggregationVariants::Composite(composite) => composite
+                .sources
+                .iter()
+                .map(|source| source.field())
+                .collect(),
             AggregationVariants::Average(avg) => vec![avg.field_name()],
             AggregationVariants::Count(count) => vec![count.field_name()],
             AggregationVariants::Max(max) => vec![max.field_name()],
@@ -214,6 +222,12 @@ impl AggregationVariants {
             _ => None,
         }
     }
+    pub(crate) fn as_composite(&self) -> Option<&CompositeAggregation> {
+        match &self {
+            AggregationVariants::Composite(composite) => Some(composite),
+            _ => None,
+        }
+    }
     pub(crate) fn as_percentile(&self) -> Option<&PercentilesAggregationReq> {
         match &self {
             AggregationVariants::Percentiles(percentile_req) => Some(percentile_req),

src/aggregation/agg_result.rs

@@ -9,10 +9,12 @@ use rustc_hash::FxHashMap;
 use serde::{Deserialize, Serialize};
 
 use super::bucket::GetDocCount;
+use super::intermediate_agg_result::CompositeIntermediateKey;
 use super::metric::{
     ExtendedStats, PercentilesMetricResult, SingleMetricResult, Stats, TopHitsMetricResult,
 };
 use super::{AggregationError, Key};
+use crate::aggregation::bucket::AfterKey;
 use crate::TantivyError;
 
 #[derive(Clone, Default, Debug, PartialEq, Serialize, Deserialize)]
@@ -158,6 +160,14 @@ pub enum BucketResult {
     },
     /// This is the filter result - a single bucket with sub-aggregations
     Filter(FilterBucketResult),
+    /// This is the composite result
+    Composite {
+        /// The buckets
+        buckets: Vec<CompositeBucketEntry>,
+        /// The key to start after when paginating
+        #[serde(skip_serializing_if = "FxHashMap::is_empty")]
+        after_key: FxHashMap<String, AfterKey>,
+    },
 }
 
 impl BucketResult {
@@ -179,6 +189,9 @@ impl BucketResult {
                 // Only count sub-aggregation buckets
                 filter_result.sub_aggregations.get_bucket_count()
             }
+            BucketResult::Composite { buckets, .. } => {
+                buckets.iter().map(|bucket| bucket.get_bucket_count()).sum()
+            }
         }
     }
 }
@@ -337,3 +350,87 @@ pub struct FilterBucketResult {
     #[serde(flatten)]
     pub sub_aggregations: AggregationResults,
 }
+
+/// Note the type information loss compared to `CompositeIntermediateKey`.
+/// Pagination is performed using `AfterKey`, which encodes type information.
+#[derive(Clone, Debug, Serialize, Deserialize)]
+#[serde(untagged)]
+pub enum CompositeKey {
+    /// Boolean key
+    Bool(bool),
+    /// String key
+    Str(String),
+    /// `i64` key
+    I64(i64),
+    /// `u64` key
+    U64(u64),
+    /// `f64` key
+    F64(f64),
+    /// Null key
+    Null,
+}
+impl Eq for CompositeKey {}
+impl std::hash::Hash for CompositeKey {
+    fn hash<H: std::hash::Hasher>(&self, state: &mut H) {
+        core::mem::discriminant(self).hash(state);
+        match self {
+            Self::Bool(val) => val.hash(state),
+            Self::Str(text) => text.hash(state),
+            Self::F64(val) => val.to_bits().hash(state),
+            Self::U64(val) => val.hash(state),
+            Self::I64(val) => val.hash(state),
+            Self::Null => {}
+        }
+    }
+}
+impl PartialEq for CompositeKey {
+    fn eq(&self, other: &Self) -> bool {
+        match (self, other) {
+            (Self::Bool(l), Self::Bool(r)) => l == r,
+            (Self::Str(l), Self::Str(r)) => l == r,
+            (Self::F64(l), Self::F64(r)) => l.to_bits() == r.to_bits(),
+            (Self::I64(l), Self::I64(r)) => l == r,
+            (Self::U64(l), Self::U64(r)) => l == r,
+            (Self::Null, Self::Null) => true,
+            _ => false,
+        }
+    }
+}
+impl From<CompositeIntermediateKey> for CompositeKey {
+    fn from(value: CompositeIntermediateKey) -> Self {
+        match value {
+            CompositeIntermediateKey::Str(s) => Self::Str(s),
+            CompositeIntermediateKey::IpAddr(s) => {
+                if let Some(ip) = s.to_ipv4_mapped() {
+                    Self::Str(ip.to_string())
+                } else {
+                    Self::Str(s.to_string())
+                }
+            }
+            CompositeIntermediateKey::F64(f) => Self::F64(f),
+            CompositeIntermediateKey::Bool(f) => Self::Bool(f),
+            CompositeIntermediateKey::U64(f) => Self::U64(f),
+            CompositeIntermediateKey::I64(f) => Self::I64(f),
+            CompositeIntermediateKey::DateTime(f) => Self::I64(f / 1_000_000), // ns to ms
+            CompositeIntermediateKey::Null => Self::Null,
+        }
+    }
+}
+
+/// Composite bucket entry with a multi-dimensional key.
+#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
+pub struct CompositeBucketEntry {
+    /// The identifier of the bucket.
+    pub key: FxHashMap<String, CompositeKey>,
+    /// Number of documents in the bucket.
+    pub doc_count: u64,
+    #[serde(flatten)]
+    /// Sub-aggregations in this bucket.
+    pub sub_aggregation: AggregationResults,
+}
+
+impl CompositeBucketEntry {
+    pub(crate) fn get_bucket_count(&self) -> u64 {
+        1 + self.sub_aggregation.get_bucket_count()
+    }
+}

src/aggregation/bucket/composite/accessors.rs

@@ -0,0 +1,518 @@
+use std::net::Ipv6Addr;
+
+use columnar::column_values::{CompactHit, CompactSpaceU64Accessor};
+use columnar::{Column, ColumnType, MonotonicallyMappableToU64, StrColumn, TermOrdHit};
+
+use crate::aggregation::accessor_helpers::get_numeric_or_date_column_types;
+use crate::aggregation::bucket::composite::numeric_types::num_proj;
+use crate::aggregation::bucket::composite::numeric_types::num_proj::ProjectedNumber;
+use crate::aggregation::bucket::composite::ToTypePaginationOrder;
+use crate::aggregation::bucket::{
+    parse_into_milliseconds, CalendarInterval, CompositeAggregation, CompositeAggregationSource,
+    MissingOrder, Order,
+};
+use crate::aggregation::intermediate_agg_result::CompositeIntermediateKey;
+use crate::{SegmentReader, TantivyError};
+
+/// Contains all information required by the SegmentCompositeCollector to perform the
+/// composite aggregation on a segment.
+pub struct CompositeAggReqData {
+    /// The name of the aggregation.
+    pub name: String,
+    /// The normalized term aggregation request.
+    pub req: CompositeAggregation,
+    /// Accessors for each source, each source can have multiple accessors (columns).
+    pub composite_accessors: Vec<CompositeSourceAccessors>,
+}
+
+impl CompositeAggReqData {
+    /// Estimate the memory consumption of this struct in bytes.
+    pub fn get_memory_consumption(&self) -> usize {
+        std::mem::size_of::<Self>()
+            + self.composite_accessors.len() * std::mem::size_of::<CompositeSourceAccessors>()
+    }
+}
+
+/// Accessors for a single column in a composite source.
+pub struct CompositeAccessor {
+    /// The fast field column
+    pub column: Column<u64>,
+    /// The column type
+    pub column_type: ColumnType,
+    /// Term dictionary if the column type is Str
+    ///
+    /// Only used by term sources
+    pub str_dict_column: Option<StrColumn>,
+    /// Parsed date interval for date histogram sources
+    pub date_histogram_interval: PrecomputedDateInterval,
+}
+
+/// Accessors to all the columns that belong to the field of a composite source.
+pub struct CompositeSourceAccessors {
+    /// The accessors for this source
+    pub accessors: Vec<CompositeAccessor>,
+    /// The key after which to start collecting results. Applies to the first
+    /// column of the source.
+    pub after_key: PrecomputedAfterKey,
+
+    /// The column index the after_key applies to. The after_key only applies to
+    /// one column. Columns before should be skipped. Columns after should be
+    /// kept without comparison to the after_key.
+    pub after_key_accessor_idx: usize,
+
+    /// Whether to skip missing values because of the after_key. Skipping only
+    /// applies if the value for previous columns were exactly equal to the
+    /// corresponding after keys (is_on_after_key).
+    pub skip_missing: bool,
+
+    /// The after key was set to null to indicate that the last collected key
+    /// was a missing value.
+    pub is_after_key_explicit_missing: bool,
+}
+
+impl CompositeSourceAccessors {
+    /// Creates a new set of accessors for the composite source.
+    ///
+    /// Precomputes some values to make collection faster.
+    pub fn build_for_source(
+        reader: &SegmentReader,
+        source: &CompositeAggregationSource,
+        // First option is None when no after key was set in the query, the
+        // second option is None when the after key was set but its value for
+        // this source was set to `null`
+        source_after_key_opt: Option<&CompositeIntermediateKey>,
+    ) -> crate::Result<Self> {
+        let is_after_key_explicit_missing = source_after_key_opt
+            .map(|after_key| matches!(after_key, CompositeIntermediateKey::Null))
+            .unwrap_or(false);
+        let mut skip_missing = false;
+        if let Some(CompositeIntermediateKey::Null) = source_after_key_opt {
+            if !source.missing_bucket() {
+                return Err(TantivyError::InvalidArgument(
+                    "the 'after' key for a source cannot be null when 'missing_bucket' is false"
+                        .to_string(),
+                ));
+            }
+        } else if source_after_key_opt.is_some() {
+            // if missing buckets come first and we have a non null after key, we skip missing
+            if MissingOrder::First == source.missing_order() {
+                skip_missing = true;
+            }
+            if MissingOrder::Default == source.missing_order() && Order::Asc == source.order() {
+                skip_missing = true;
+            }
+        };
+
+        match source {
+            CompositeAggregationSource::Terms(source) => {
+                let allowed_column_types = [
+                    ColumnType::I64,
+                    ColumnType::U64,
+                    ColumnType::F64,
+                    ColumnType::Str,
+                    ColumnType::DateTime,
+                    ColumnType::Bool,
+                    ColumnType::IpAddr,
+                    // ColumnType::Bytes Unsupported
+                ];
+                let mut columns_and_types = reader
+                    .fast_fields()
+                    .u64_lenient_for_type_all(Some(&allowed_column_types), &source.field)?;
+
+                // Sort columns by their pagination order and determine which to skip
+                columns_and_types.sort_by_key(|(_, col_type): &(Column, ColumnType)| {
+                    col_type.column_pagination_order()
+                });
+                if source.order == Order::Desc {
+                    columns_and_types.reverse();
+                }
+                let after_key_accessor_idx = find_first_column_to_collect(
+                    &columns_and_types,
+                    source_after_key_opt,
+                    source.missing_order,
+                    source.order,
+                )?;
+
+                let source_collectors: Vec<CompositeAccessor> = columns_and_types
+                    .into_iter()
+                    .map(|(column, column_type)| {
+                        Ok(CompositeAccessor {
+                            column,
+                            column_type,
+                            str_dict_column: reader.fast_fields().str(&source.field)?,
+                            date_histogram_interval: PrecomputedDateInterval::NotApplicable,
+                        })
+                    })
+                    .collect::<crate::Result<_>>()?;
+
+                let after_key = if let Some(first_col) =
+                    source_collectors.get(after_key_accessor_idx)
+                {
+                    match source_after_key_opt {
+                        Some(after_key) => PrecomputedAfterKey::precompute(
+                            first_col,
+                            after_key,
+                            &source.field,
+                            source.missing_order,
+                            source.order,
+                        )?,
+                        None => {
+                            precompute_missing_after_key(false, source.missing_order, source.order)
+                        }
+                    }
+                } else {
+                    // if no columns, we don't care about the after_key
+                    PrecomputedAfterKey::Next(0)
+                };
+
+                Ok(CompositeSourceAccessors {
+                    accessors: source_collectors,
+                    is_after_key_explicit_missing,
+                    skip_missing,
+                    after_key,
+                    after_key_accessor_idx,
+                })
+            }
+            CompositeAggregationSource::Histogram(source) => {
+                let column_and_types: Vec<(Column, ColumnType)> =
+                    reader.fast_fields().u64_lenient_for_type_all(
+                        Some(get_numeric_or_date_column_types()),
+                        &source.field,
+                    )?;
+                let source_collectors: Vec<CompositeAccessor> = column_and_types
+                    .into_iter()
+                    .map(|(column, column_type)| {
+                        Ok(CompositeAccessor {
+                            column,
+                            column_type,
+                            str_dict_column: None,
+                            date_histogram_interval: PrecomputedDateInterval::NotApplicable,
+                        })
+                    })
+                    .collect::<crate::Result<_>>()?;
+                let after_key = match source_after_key_opt {
+                    Some(CompositeIntermediateKey::F64(key)) => {
+                        let normalized_key = *key / source.interval;
+                        num_proj::f64_to_i64(normalized_key).into()
+                    }
+                    Some(CompositeIntermediateKey::Null) => {
+                        precompute_missing_after_key(true, source.missing_order, source.order)
+                    }
+                    None => precompute_missing_after_key(true, source.missing_order, source.order),
+                    _ => {
+                        return Err(crate::TantivyError::InvalidArgument(
+                            "After key type invalid for interval composite source".to_string(),
+                        ));
+                    }
+                };
+                Ok(CompositeSourceAccessors {
+                    accessors: source_collectors,
+                    is_after_key_explicit_missing,
+                    skip_missing,
+                    after_key,
+                    after_key_accessor_idx: 0,
+                })
+            }
+            CompositeAggregationSource::DateHistogram(source) => {
+                let column_and_types = reader
+                    .fast_fields()
+                    .u64_lenient_for_type_all(Some(&[ColumnType::DateTime]), &source.field)?;
+                let date_histogram_interval =
+                    PrecomputedDateInterval::from_date_histogram_source_intervals(
+                        &source.fixed_interval,
+                        source.calendar_interval,
+                    )?;
+                let source_collectors: Vec<CompositeAccessor> = column_and_types
+                    .into_iter()
+                    .map(|(column, column_type)| {
+                        Ok(CompositeAccessor {
+                            column,
+                            column_type,
+                            str_dict_column: None,
+                            date_histogram_interval,
+                        })
+                    })
+                    .collect::<crate::Result<_>>()?;
+                let after_key = match source_after_key_opt {
+                    Some(CompositeIntermediateKey::DateTime(key)) => {
+                        PrecomputedAfterKey::Exact(key.to_u64())
+                    }
+                    Some(CompositeIntermediateKey::Null) => {
+                        precompute_missing_after_key(true, source.missing_order, source.order)
+                    }
+                    None => precompute_missing_after_key(true, source.missing_order, source.order),
+                    _ => {
+                        return Err(crate::TantivyError::InvalidArgument(
+                            "After key type invalid for interval composite source".to_string(),
+                        ));
+                    }
+                };
+                Ok(CompositeSourceAccessors {
+                    accessors: source_collectors,
+                    is_after_key_explicit_missing,
+                    skip_missing,
+                    after_key,
+                    after_key_accessor_idx: 0,
+                })
+            }
+        }
+    }
+}
+
+/// Finds the index of the first column we should start collecting from to
+/// resume the pagination from the after_key.
+fn find_first_column_to_collect<T>(
+    sorted_columns: &[(T, ColumnType)],
+    after_key_opt: Option<&CompositeIntermediateKey>,
+    missing_order: MissingOrder,
+    order: Order,
+) -> crate::Result<usize> {
+    let after_key = match after_key_opt {
+        None => return Ok(0), // No pagination, start from beginning
+        Some(key) => key,
+    };
+    // Handle null after_key (we were on a missing value last time)
+    if matches!(after_key, CompositeIntermediateKey::Null) {
+        return match (missing_order, order) {
+            // Missing values come first, so all columns remain
+            (MissingOrder::First, _) | (MissingOrder::Default, Order::Asc) => Ok(0),
+            // Missing values come last, so all columns are done
+            (MissingOrder::Last, _) | (MissingOrder::Default, Order::Desc) => {
+                Ok(sorted_columns.len())
+            }
+        };
+    }
+    // Find the first column whose type order matches or follows the after_key's
+    // type in the pagination sequence
+    let after_key_column_order = after_key.column_pagination_order();
+    for (idx, (_, col_type)) in sorted_columns.iter().enumerate() {
+        let col_order = col_type.column_pagination_order();
+        let is_first_to_collect = match order {
+            Order::Asc => col_order >= after_key_column_order,
+            Order::Desc => col_order <= after_key_column_order,
+        };
+        if is_first_to_collect {
+            return Ok(idx);
+        }
+    }
+    // All columns are before the after_key, nothing left to collect
+    Ok(sorted_columns.len())
+}
+
+fn precompute_missing_after_key(
+    is_after_key_explicit_missing: bool,
+    missing_order: MissingOrder,
+    order: Order,
+) -> PrecomputedAfterKey {
+    let after_last = PrecomputedAfterKey::AfterLast;
+    let before_first = PrecomputedAfterKey::Next(0);
+    match (is_after_key_explicit_missing, missing_order, order) {
+        (true, MissingOrder::First, Order::Asc) => before_first,
+        (true, MissingOrder::First, Order::Desc) => after_last,
+        (true, MissingOrder::Last, Order::Asc) => after_last,
+        (true, MissingOrder::Last, Order::Desc) => before_first,
+        (true, MissingOrder::Default, Order::Asc) => before_first,
+        (true, MissingOrder::Default, Order::Desc) => after_last,
+        (false, _, Order::Asc) => before_first,
+        (false, _, Order::Desc) => after_last,
+    }
+}
+
+/// A parsed representation of the date interval for date histogram sources
+#[derive(Clone, Copy, Debug)]
+pub enum PrecomputedDateInterval {
+    /// This is not a date histogram source
+    NotApplicable,
+    /// Source was configured with a fixed interval
+    FixedNanoseconds(i64),
+    /// Source was configured with a calendar interval
+    Calendar(CalendarInterval),
+}
+
+impl PrecomputedDateInterval {
+    /// Validates the date histogram source interval fields and parses a date interval from them.
+    pub fn from_date_histogram_source_intervals(
+        fixed_interval: &Option<String>,
+        calendar_interval: Option<CalendarInterval>,
+    ) -> crate::Result<Self> {
+        match (fixed_interval, calendar_interval) {
+            (Some(_), Some(_)) | (None, None) => Err(TantivyError::InvalidArgument(
+                "date histogram source must one and only one of fixed_interval or \
+                 calendar_interval set"
+                    .to_string(),
+            )),
+            (Some(fixed_interval), None) => {
+                let fixed_interval_ms = parse_into_milliseconds(fixed_interval)?;
+                Ok(PrecomputedDateInterval::FixedNanoseconds(
+                    fixed_interval_ms * 1_000_000,
+                ))
+            }
+            (None, Some(calendar_interval)) => {
+                Ok(PrecomputedDateInterval::Calendar(calendar_interval))
+            }
+        }
+    }
+}
+
+/// The after key projected to the u64 column space
+///
+/// Some column types (term, IP) might not have an exact representation of the
+/// specified after key
+#[derive(Debug)]
+pub enum PrecomputedAfterKey {
+    /// The after key could be exactly represented in the column space.
+    Exact(u64),
+    /// The after key could not be exactly represented exactly represented, so
+    /// this is the next closest one.
+    Next(u64),
+    /// The after key could not be represented in the column space, it is
+    /// greater than all value
+    AfterLast,
+}
+
+impl From<CompactHit> for PrecomputedAfterKey {
+    fn from(hit: CompactHit) -> Self {
+        match hit {
+            CompactHit::Exact(ord) => PrecomputedAfterKey::Exact(ord as u64),
+            CompactHit::Next(ord) => PrecomputedAfterKey::Next(ord as u64),
+            CompactHit::AfterLast => PrecomputedAfterKey::AfterLast,
+        }
+    }
+}
+
+impl From<TermOrdHit> for PrecomputedAfterKey {
+    fn from(hit: TermOrdHit) -> Self {
+        match hit {
+            TermOrdHit::Exact(ord) => PrecomputedAfterKey::Exact(ord),
+            // TermOrdHit represents AfterLast as Next(u64::MAX), we keep it as is
+            TermOrdHit::Next(ord) => PrecomputedAfterKey::Next(ord),
+        }
+    }
+}
+
+impl<T: MonotonicallyMappableToU64> From<ProjectedNumber<T>> for PrecomputedAfterKey {
+    fn from(num: ProjectedNumber<T>) -> Self {
+        match num {
+            ProjectedNumber::Exact(number) => PrecomputedAfterKey::Exact(number.to_u64()),
+            ProjectedNumber::Next(number) => PrecomputedAfterKey::Next(number.to_u64()),
+            ProjectedNumber::AfterLast => PrecomputedAfterKey::AfterLast,
+        }
+    }
+}
+
+// /!\ These operators only makes sense if both values are in the same column space
+impl PrecomputedAfterKey {
+    pub fn equals(&self, column_value: u64) -> bool {
+        match self {
+            PrecomputedAfterKey::Exact(v) => *v == column_value,
+            PrecomputedAfterKey::Next(_) => false,
+            PrecomputedAfterKey::AfterLast => false,
+        }
+    }
+
+    pub fn gt(&self, column_value: u64) -> bool {
+        match self {
+            PrecomputedAfterKey::Exact(v) => *v > column_value,
+            PrecomputedAfterKey::Next(v) => *v > column_value,
+            PrecomputedAfterKey::AfterLast => true,
+        }
+    }
+
+    pub fn lt(&self, column_value: u64) -> bool {
+        match self {
+            PrecomputedAfterKey::Exact(v) => *v < column_value,
+            // a value equal to the next is greater than the after key
+            PrecomputedAfterKey::Next(v) => *v <= column_value,
+            PrecomputedAfterKey::AfterLast => false,
+        }
+    }
+
+    fn precompute_ip_addr(column: &Column<u64>, key: &Ipv6Addr) -> crate::Result<Self> {
+        let compact_space_accessor = column
+            .values
+            .clone()
+            .downcast_arc::<CompactSpaceU64Accessor>()
+            .map_err(|_| {
+                TantivyError::AggregationError(crate::aggregation::AggregationError::InternalError(
+                    "type mismatch: could not downcast to CompactSpaceU64Accessor".to_string(),
+                ))
+            })?;
+        let ip_u128 = key.to_bits();
+        let ip_next_compact = compact_space_accessor.u128_to_next_compact(ip_u128);
+        Ok(ip_next_compact.into())
+    }
+
+    fn precompute_term_ord(
+        str_dict_column: &Option<StrColumn>,
+        key: &str,
+        field: &str,
+    ) -> crate::Result<Self> {
+        let dict = str_dict_column
+            .as_ref()
+            .expect("dictionary missing for str accessor")
+            .dictionary();
+        let next_ord = dict.term_ord_or_next(key).map_err(|_| {
+            TantivyError::InvalidArgument(format!(
+                "failed to lookup after_key '{}' for field '{}'",
+                key, field
+            ))
+        })?;
+        Ok(next_ord.into())
+    }
+
+    /// Projects the after key into the column space of the given accessor.
+    ///
+    /// The computed after key will not take care of skipping entire columns
+    /// when the after key type is ordered after the accessor's type, that
+    /// should be performed earlier.
+    pub fn precompute(
+        composite_accessor: &CompositeAccessor,
+        source_after_key: &CompositeIntermediateKey,
+        field: &str,
+        missing_order: MissingOrder,
+        order: Order,
+    ) -> crate::Result<Self> {
+        use CompositeIntermediateKey as CIKey;
+        let precomputed_key = match (composite_accessor.column_type, source_after_key) {
+            (ColumnType::Bytes, _) => panic!("unsupported"),
+            // null after key
+            (_, CIKey::Null) => precompute_missing_after_key(false, missing_order, order),
+            // numerical
+            (ColumnType::I64, CIKey::I64(k)) => PrecomputedAfterKey::Exact(k.to_u64()),
+            (ColumnType::I64, CIKey::U64(k)) => num_proj::u64_to_i64(*k).into(),
+            (ColumnType::I64, CIKey::F64(k)) => num_proj::f64_to_i64(*k).into(),
+            (ColumnType::U64, CIKey::I64(k)) => num_proj::i64_to_u64(*k).into(),
+            (ColumnType::U64, CIKey::U64(k)) => PrecomputedAfterKey::Exact(*k),
+            (ColumnType::U64, CIKey::F64(k)) => num_proj::f64_to_u64(*k).into(),
+            (ColumnType::F64, CIKey::I64(k)) => num_proj::i64_to_f64(*k).into(),
+            (ColumnType::F64, CIKey::U64(k)) => num_proj::u64_to_f64(*k).into(),
+            (ColumnType::F64, CIKey::F64(k)) => PrecomputedAfterKey::Exact(k.to_u64()),
+            // boolean
+            (ColumnType::Bool, CIKey::Bool(key)) => PrecomputedAfterKey::Exact(key.to_u64()),
+            // string
+            (ColumnType::Str, CIKey::Str(key)) => PrecomputedAfterKey::precompute_term_ord(
+                &composite_accessor.str_dict_column,
+                key,
+                field,
+            )?,
+            // date time
+            (ColumnType::DateTime, CIKey::DateTime(key)) => {
+                PrecomputedAfterKey::Exact(key.to_u64())
+            }
+            // ip address
+            (ColumnType::IpAddr, CIKey::IpAddr(key)) => {
+                PrecomputedAfterKey::precompute_ip_addr(&composite_accessor.column, key)?
+            }
+            // assume the column's type is ordered after the after_key's type
+            _ => PrecomputedAfterKey::keep_all(order),
+        };
+        Ok(precomputed_key)
+    }
+
+    fn keep_all(order: Order) -> Self {
+        match order {
+            Order::Asc => PrecomputedAfterKey::Next(0),
+            Order::Desc => PrecomputedAfterKey::Next(u64::MAX),
+        }
+    }
+}

src/aggregation/bucket/composite/calendar_interval.rs

@@ -0,0 +1,138 @@
+use time::convert::{Day, Nanosecond};
+use time::{Time, UtcDateTime};
+
+const NS_IN_DAY: i64 = Nanosecond::per_t::<i128>(Day) as i64;
+
+/// Computes the timestamp in nanoseconds corresponding to the beginning of the
+/// year (January 1st at midnight UTC).
+pub(super) fn try_year_bucket(timestamp_ns: i64) -> crate::Result<i64> {
+    year_bucket_using_time_crate(timestamp_ns).map_err(|e| {
+        crate::TantivyError::InvalidArgument(format!(
+            "Failed to compute year bucket for timestamp {}: {e}",
+            timestamp_ns
+        ))
+    })
+}
+
+/// Computes the timestamp in nanoseconds corresponding to the beginning of the
+/// month (1st at midnight UTC).
+pub(super) fn try_month_bucket(timestamp_ns: i64) -> crate::Result<i64> {
+    month_bucket_using_time_crate(timestamp_ns).map_err(|e| {
+        crate::TantivyError::InvalidArgument(format!(
+            "Failed to compute month bucket for timestamp {}: {e}",
+            timestamp_ns
+        ))
+    })
+}
+
+/// Computes the timestamp in nanoseconds corresponding to the beginning of the
+/// week (Monday at midnight UTC).
+pub(super) fn week_bucket(timestamp_ns: i64) -> i64 {
+    // 1970-01-01 was a Thursday (weekday = 4)
+    let days_since_epoch = timestamp_ns.div_euclid(NS_IN_DAY);
+    // Find the weekday: 0=Monday, ..., 6=Sunday
+    let weekday = (days_since_epoch + 3).rem_euclid(7);
+    let monday_days_since_epoch = days_since_epoch - weekday;
+    monday_days_since_epoch * NS_IN_DAY
+}
+
+fn year_bucket_using_time_crate(timestamp_ns: i64) -> Result<i64, time::Error> {
+    let timestamp_ns = UtcDateTime::from_unix_timestamp_nanos(timestamp_ns as i128)?
+        .replace_ordinal(1)?
+        .replace_time(Time::MIDNIGHT)
+        .unix_timestamp_nanos();
+    Ok(timestamp_ns as i64)
+}
+
+fn month_bucket_using_time_crate(timestamp_ns: i64) -> Result<i64, time::Error> {
+    let timestamp_ns = UtcDateTime::from_unix_timestamp_nanos(timestamp_ns as i128)?
+        .replace_day(1)?
+        .replace_time(Time::MIDNIGHT)
+        .unix_timestamp_nanos();
+    Ok(timestamp_ns as i64)
+}
+
+#[cfg(test)]
+mod tests {
+    use std::i64;
+
+    use time::format_description::well_known::Iso8601;
+    use time::UtcDateTime;
+
+    use super::*;
+
+    fn ts_ns(iso: &str) -> i64 {
+        UtcDateTime::parse(iso, &Iso8601::DEFAULT)
+            .unwrap()
+            .unix_timestamp_nanos() as i64
+    }
+
+    #[test]
+    fn test_year_bucket() {
+        let ts = ts_ns("1970-01-01T00:00:00Z");
+        let res = try_year_bucket(ts).unwrap();
+        assert_eq!(res, ts_ns("1970-01-01T00:00:00Z"));
+
+        let ts = ts_ns("1970-06-01T10:00:01.010Z");
+        let res = try_year_bucket(ts).unwrap();
+        assert_eq!(res, ts_ns("1970-01-01T00:00:00Z"));
+
+        let ts = ts_ns("2008-12-31T23:59:59.999999999Z"); // leap year
+        let res = try_year_bucket(ts).unwrap();
+        assert_eq!(res, ts_ns("2008-01-01T00:00:00Z"));
+
+        let ts = ts_ns("2008-01-01T00:00:00Z"); // leap year
+        let res = try_year_bucket(ts).unwrap();
+        assert_eq!(res, ts_ns("2008-01-01T00:00:00Z"));
+
+        let ts = ts_ns("2010-12-31T23:59:59.999999999Z");
+        let res = try_year_bucket(ts).unwrap();
+        assert_eq!(res, ts_ns("2010-01-01T00:00:00Z"));
+
+        let ts = ts_ns("1972-06-01T00:10:00Z");
+        let res = try_year_bucket(ts).unwrap();
+        assert_eq!(res, ts_ns("1972-01-01T00:00:00Z"));
+    }
+
+    #[test]
+    fn test_month_bucket() {
+        let ts = ts_ns("1970-01-15T00:00:00Z");
+        let res = try_month_bucket(ts).unwrap();
+        assert_eq!(res, ts_ns("1970-01-01T00:00:00Z"));
+
+        let ts = ts_ns("1970-02-01T00:00:00Z");
+        let res = try_month_bucket(ts).unwrap();
+        assert_eq!(res, ts_ns("1970-02-01T00:00:00Z"));
+
+        let ts = ts_ns("2000-01-31T23:59:59.999999999Z");
+        let res = try_month_bucket(ts).unwrap();
+        assert_eq!(res, ts_ns("2000-01-01T00:00:00Z"));
+    }
+
+    #[test]
+    fn test_week_bucket() {
+        let ts = ts_ns("1970-01-05T00:00:00Z"); // Monday
+        let res = week_bucket(ts);
+        assert_eq!(res, ts_ns("1970-01-05T00:00:00Z"));
+
+        let ts = ts_ns("1970-01-05T23:59:59Z"); // Monday
+        let res = week_bucket(ts);
+        assert_eq!(res, ts_ns("1970-01-05T00:00:00Z"));
+
+        let ts = ts_ns("1970-01-07T01:13:00Z"); // Wednesday
+        let res = week_bucket(ts);
+        assert_eq!(res, ts_ns("1970-01-05T00:00:00Z"));
+
+        let ts = ts_ns("1970-01-11T23:59:59.999999999Z"); // Sunday
+        let res = week_bucket(ts);
+        assert_eq!(res, ts_ns("1970-01-05T00:00:00Z"));
+
+        let ts = ts_ns("2025-10-16T10:41:59.010Z"); // Thursday
+        let res = week_bucket(ts);
+        assert_eq!(res, ts_ns("2025-10-13T00:00:00Z"));
+
+        let ts = ts_ns("1970-01-01T00:00:00Z"); // Thursday
+        let res = week_bucket(ts);
+        assert_eq!(res, ts_ns("1969-12-29T00:00:00Z")); // Negative
+    }
+}

src/aggregation/bucket/composite/collector.rs

@@ -0,0 +1,652 @@
+use std::fmt::Debug;
+use std::mem;
+use std::net::Ipv6Addr;
+
+use columnar::column_values::CompactSpaceU64Accessor;
+use columnar::{
+    Column, ColumnType, Dictionary, MonotonicallyMappableToU128, MonotonicallyMappableToU64,
+    NumericalValue, StrColumn,
+};
+use rustc_hash::FxHashMap;
+use smallvec::SmallVec;
+
+use crate::aggregation::agg_data::{
+    build_segment_agg_collectors, AggRefNode, AggregationsSegmentCtx,
+};
+use crate::aggregation::bucket::composite::accessors::{
+    CompositeAccessor, CompositeAggReqData, PrecomputedDateInterval,
+};
+use crate::aggregation::bucket::composite::calendar_interval;
+use crate::aggregation::bucket::composite::map::{DynArrayHeapMap, MAX_DYN_ARRAY_SIZE};
+use crate::aggregation::bucket::{
+    CalendarInterval, CompositeAggregationSource, MissingOrder, Order,
+};
+use crate::aggregation::cached_sub_aggs::{CachedSubAggs, HighCardSubAggCache};
+use crate::aggregation::intermediate_agg_result::{
+    CompositeIntermediateKey, IntermediateAggregationResult, IntermediateAggregationResults,
+    IntermediateBucketResult, IntermediateCompositeBucketEntry, IntermediateCompositeBucketResult,
+};
+use crate::aggregation::segment_agg_result::{BucketIdProvider, SegmentAggregationCollector};
+use crate::aggregation::BucketId;
+use crate::TantivyError;
+
+#[derive(Clone, Debug)]
+struct CompositeBucketCollector {
+    count: u32,
+    bucket_id: BucketId,
+}
+
+/// Compact sortable representation of a single source value within a composite key.
+///
+/// The struct encodes both the column identity and the fast field value in a way
+/// that preserves the desired sort order via the derived `Ord` implementation
+/// (fields are compared top-to-bottom: `sort_key` first, then `encoded_value`).
+///
+/// ## `sort_key` encoding
+/// - `0` — missing value, sorted first
+/// - `1..=254` — present value; the original accessor index is `sort_key - 1`
+/// - `u8::MAX` (255) — missing value, sorted last
+///
+/// ## `encoded_value` encoding
+/// - `0` when the field is missing
+/// - The raw u64 fast-field representation when order is ascending
+/// - Bitwise NOT of the raw u64 when order is descending
+#[derive(Clone, Copy, Debug, PartialEq, Eq, PartialOrd, Ord, Default, Hash)]
+struct InternalValueRepr {
+    /// Column index biased by +1 (so 0 and u8::MAX are reserved for missing sentinels).
+    sort_key: u8,
+    /// Fast field value, possibly bit-flipped for descending order.
+    encoded_value: u64,
+}
+
+impl InternalValueRepr {
+    #[inline]
+    fn new_term(raw: u64, accessor_idx: u8, order: Order) -> Self {
+        let encoded_value = match order {
+            Order::Asc => raw,
+            Order::Desc => !raw,
+        };
+        InternalValueRepr {
+            sort_key: accessor_idx + 1,
+            encoded_value,
+        }
+    }
+
+    /// For histogram sources the column index is irrelevant (always 1).
+    #[inline]
+    fn new_histogram(raw: u64, order: Order) -> Self {
+        let encoded_value = match order {
+            Order::Asc => raw,
+            Order::Desc => !raw,
+        };
+        InternalValueRepr {
+            sort_key: 1,
+            encoded_value,
+        }
+    }
+
+    #[inline]
+    fn new_missing(order: Order, missing_order: MissingOrder) -> Self {
+        let sort_key = match (missing_order, order) {
+            (MissingOrder::First, _) | (MissingOrder::Default, Order::Asc) => 0,
+            (MissingOrder::Last, _) | (MissingOrder::Default, Order::Desc) => u8::MAX,
+        };
+        InternalValueRepr {
+            sort_key,
+            encoded_value: 0,
+        }
+    }
+
+    /// Decode back to `(accessor_idx, raw_value)`.
+    /// Returns `None` when the value represents a missing field.
+    #[inline]
+    fn decode(self, order: Order) -> Option<(u8, u64)> {
+        if self.sort_key == 0 || self.sort_key == u8::MAX {
+            return None;
+        }
+        let raw = match order {
+            Order::Asc => self.encoded_value,
+            Order::Desc => !self.encoded_value,
+        };
+        Some((self.sort_key - 1, raw))
+    }
+}
+
+/// The collector puts values from the fast field into the correct buckets and
+/// does a conversion to the correct datatype.
+#[derive(Debug)]
+pub struct SegmentCompositeCollector {
+    /// One DynArrayHeapMap per parent bucket.
+    parent_buckets: Vec<DynArrayHeapMap<InternalValueRepr, CompositeBucketCollector>>,
+    accessor_idx: usize,
+    sub_agg: Option<CachedSubAggs<HighCardSubAggCache>>,
+    bucket_id_provider: BucketIdProvider,
+    /// Number of sources, needed when creating new DynArrayHeapMaps.
+    num_sources: usize,
+}
+
+impl SegmentAggregationCollector for SegmentCompositeCollector {
+    fn add_intermediate_aggregation_result(
+        &mut self,
+        agg_data: &AggregationsSegmentCtx,
+        results: &mut IntermediateAggregationResults,
+        parent_bucket_id: BucketId,
+    ) -> crate::Result<()> {
+        let name = agg_data
+            .get_composite_req_data(self.accessor_idx)
+            .name
+            .clone();
+
+        let buckets = self.add_intermediate_bucket_result(agg_data, parent_bucket_id)?;
+        results.push(
+            name,
+            IntermediateAggregationResult::Bucket(IntermediateBucketResult::Composite { buckets }),
+        )?;
+
+        Ok(())
+    }
+
+    fn collect(
+        &mut self,
+        parent_bucket_id: BucketId,
+        docs: &[crate::DocId],
+        agg_data: &mut AggregationsSegmentCtx,
+    ) -> crate::Result<()> {
+        let mem_pre = self.get_memory_consumption();
+        let composite_agg_data = agg_data.take_composite_req_data(self.accessor_idx);
+
+        for doc in docs {
+            let mut visitor = CompositeKeyVisitor {
+                doc_id: *doc,
+                composite_agg_data: &composite_agg_data,
+                buckets: &mut self.parent_buckets[parent_bucket_id as usize],
+                sub_agg: &mut self.sub_agg,
+                bucket_id_provider: &mut self.bucket_id_provider,
+                sub_level_values: SmallVec::new(),
+            };
+            visitor.visit(0, true)?;
+        }
+        agg_data.put_back_composite_req_data(self.accessor_idx, composite_agg_data);
+
+        if let Some(sub_agg) = &mut self.sub_agg {
+            sub_agg.check_flush_local(agg_data)?;
+        }
+
+        let mem_delta = self.get_memory_consumption() - mem_pre;
+        if mem_delta > 0 {
+            agg_data.context.limits.add_memory_consumed(mem_delta)?;
+        }
+
+        Ok(())
+    }
+
+    fn flush(&mut self, agg_data: &mut AggregationsSegmentCtx) -> crate::Result<()> {
+        if let Some(sub_agg) = &mut self.sub_agg {
+            sub_agg.flush(agg_data)?;
+        }
+        Ok(())
+    }
+
+    fn prepare_max_bucket(
+        &mut self,
+        max_bucket: BucketId,
+        _agg_data: &AggregationsSegmentCtx,
+    ) -> crate::Result<()> {
+        let required_len = max_bucket as usize + 1;
+        while self.parent_buckets.len() < required_len {
+            let map = DynArrayHeapMap::try_new(self.num_sources)?;
+            self.parent_buckets.push(map);
+        }
+        Ok(())
+    }
+}
+
+impl SegmentCompositeCollector {
+    fn get_memory_consumption(&self) -> u64 {
+        self.parent_buckets
+            .iter()
+            .map(|m| m.memory_consumption())
+            .sum()
+    }
+
+    pub(crate) fn from_req_and_validate(
+        req_data: &mut AggregationsSegmentCtx,
+        node: &AggRefNode,
+    ) -> crate::Result<Self> {
+        validate_req(req_data, node.idx_in_req_data)?;
+
+        let has_sub_aggregations = !node.children.is_empty();
+        let sub_agg = if has_sub_aggregations {
+            let sub_agg_collector = build_segment_agg_collectors(req_data, &node.children)?;
+            Some(CachedSubAggs::new(sub_agg_collector))
+        } else {
+            None
+        };
+
+        let composite_req_data = req_data.get_composite_req_data(node.idx_in_req_data);
+        let num_sources = composite_req_data.req.sources.len();
+
+        Ok(SegmentCompositeCollector {
+            parent_buckets: vec![DynArrayHeapMap::try_new(num_sources)?],
+            accessor_idx: node.idx_in_req_data,
+            sub_agg,
+            bucket_id_provider: BucketIdProvider::default(),
+            num_sources,
+        })
+    }
+
+    #[inline]
+    fn add_intermediate_bucket_result(
+        &mut self,
+        agg_data: &AggregationsSegmentCtx,
+        parent_bucket_id: BucketId,
+    ) -> crate::Result<IntermediateCompositeBucketResult> {
+        let empty_map = DynArrayHeapMap::try_new(self.num_sources)?;
+        let heap_map = mem::replace(
+            &mut self.parent_buckets[parent_bucket_id as usize],
+            empty_map,
+        );
+
+        let mut dict: FxHashMap<Vec<CompositeIntermediateKey>, IntermediateCompositeBucketEntry> =
+            Default::default();
+        dict.reserve(heap_map.size());
+        let composite_data = agg_data.get_composite_req_data(self.accessor_idx);
+        for (key_internal_repr, agg) in heap_map.into_iter() {
+            let key = resolve_key(&key_internal_repr, composite_data)?;
+            let mut sub_aggregation_res = IntermediateAggregationResults::default();
+            if let Some(sub_agg) = &mut self.sub_agg {
+                sub_agg
+                    .get_sub_agg_collector()
+                    .add_intermediate_aggregation_result(
+                        agg_data,
+                        &mut sub_aggregation_res,
+                        agg.bucket_id,
+                    )?;
+            }
+
+            dict.insert(
+                key,
+                IntermediateCompositeBucketEntry {
+                    doc_count: agg.count,
+                    sub_aggregation: sub_aggregation_res,
+                },
+            );
+        }
+
+        Ok(IntermediateCompositeBucketResult {
+            entries: dict,
+            target_size: composite_data.req.size,
+            orders: composite_data
+                .req
+                .sources
+                .iter()
+                .map(|source| match source {
+                    CompositeAggregationSource::Terms(t) => (t.order, t.missing_order),
+                    CompositeAggregationSource::Histogram(h) => (h.order, h.missing_order),
+                    CompositeAggregationSource::DateHistogram(d) => (d.order, d.missing_order),
+                })
+                .collect(),
+        })
+    }
+}
+
+fn validate_req(req_data: &mut AggregationsSegmentCtx, accessor_idx: usize) -> crate::Result<()> {
+    let composite_data = req_data.get_composite_req_data(accessor_idx);
+    let req = &composite_data.req;
+    if req.sources.is_empty() {
+        return Err(TantivyError::InvalidArgument(
+            "composite aggregation must have at least one source".to_string(),
+        ));
+    }
+    if req.size == 0 {
+        return Err(TantivyError::InvalidArgument(
+            "composite aggregation 'size' must be > 0".to_string(),
+        ));
+    }
+
+    if composite_data.composite_accessors.len() > MAX_DYN_ARRAY_SIZE {
+        return Err(TantivyError::InvalidArgument(format!(
+            "composite aggregation source supports maximum {MAX_DYN_ARRAY_SIZE} sources",
+        )));
+    }
+
+    let column_types_for_sources = composite_data.composite_accessors.iter().map(|item| {
+        item.accessors
+            .iter()
+            .map(|a| a.column_type)
+            .collect::<Vec<_>>()
+    });
+
+    for column_types in column_types_for_sources {
+        if column_types.contains(&ColumnType::Bytes) {
+            return Err(TantivyError::InvalidArgument(
+                "composite aggregation does not support 'bytes' field type".to_string(),
+            ));
+        }
+    }
+    Ok(())
+}
+
+fn collect_bucket_with_limit(
+    doc_id: crate::DocId,
+    limit_num_buckets: usize,
+    buckets: &mut DynArrayHeapMap<InternalValueRepr, CompositeBucketCollector>,
+    key: &[InternalValueRepr],
+    sub_agg: &mut Option<CachedSubAggs<HighCardSubAggCache>>,
+    bucket_id_provider: &mut BucketIdProvider,
+) {
+    let mut record_in_bucket = |bucket: &mut CompositeBucketCollector| {
+        bucket.count += 1;
+        if let Some(sub_agg) = sub_agg {
+            sub_agg.push(bucket.bucket_id, doc_id);
+        }
+    };
+
+    // We still have room for buckets, just insert
+    if buckets.size() < limit_num_buckets {
+        let bucket = buckets.get_or_insert_with(key, || CompositeBucketCollector {
+            count: 0,
+            bucket_id: bucket_id_provider.next_bucket_id(),
+        });
+        record_in_bucket(bucket);
+        return;
+    }
+
+    // Map is full, but we can still update the bucket if it already exists
+    if let Some(bucket) = buckets.get_mut(key) {
+        record_in_bucket(bucket);
+        return;
+    }
+
+    // Check if the item qualifies to enter the top-k, and evict the highest if it does
+    if let Some(highest_key) = buckets.peek_highest() {
+        if key < highest_key {
+            buckets.evict_highest();
+            let bucket = buckets.get_or_insert_with(key, || CompositeBucketCollector {
+                count: 0,
+                bucket_id: bucket_id_provider.next_bucket_id(),
+            });
+            record_in_bucket(bucket);
+        }
+    }
+}
+
+/// Converts the composite key from its internal column space representation
+/// (segment specific) into its intermediate form.
+fn resolve_key(
+    internal_key: &[InternalValueRepr],
+    agg_data: &CompositeAggReqData,
+) -> crate::Result<Vec<CompositeIntermediateKey>> {
+    internal_key
+        .iter()
+        .enumerate()
+        .map(|(idx, val)| {
+            resolve_internal_value_repr(
+                *val,
+                &agg_data.req.sources[idx],
+                &agg_data.composite_accessors[idx].accessors,
+            )
+        })
+        .collect()
+}
+
+fn resolve_internal_value_repr(
+    internal_value_repr: InternalValueRepr,
+    source: &CompositeAggregationSource,
+    composite_accessors: &[CompositeAccessor],
+) -> crate::Result<CompositeIntermediateKey> {
+    let decoded_value_opt = match source {
+        CompositeAggregationSource::Terms(source) => internal_value_repr.decode(source.order),
+        CompositeAggregationSource::Histogram(source) => internal_value_repr.decode(source.order),
+        CompositeAggregationSource::DateHistogram(source) => {
+            internal_value_repr.decode(source.order)
+        }
+    };
+    let Some((decoded_accessor_idx, val)) = decoded_value_opt else {
+        return Ok(CompositeIntermediateKey::Null);
+    };
+    let key = match source {
+        CompositeAggregationSource::Terms(_) => {
+            let CompositeAccessor {
+                column_type,
+                str_dict_column,
+                column,
+                ..
+            } = &composite_accessors[decoded_accessor_idx as usize];
+            resolve_term(val, column_type, str_dict_column, column)?
+        }
+        CompositeAggregationSource::Histogram(source) => {
+            CompositeIntermediateKey::F64(i64::from_u64(val) as f64 * source.interval)
+        }
+        CompositeAggregationSource::DateHistogram(_) => {
+            CompositeIntermediateKey::DateTime(i64::from_u64(val))
+        }
+    };
+
+    Ok(key)
+}
+
+fn resolve_term(
+    val: u64,
+    column_type: &ColumnType,
+    str_dict_column: &Option<StrColumn>,
+    column: &Column,
+) -> crate::Result<CompositeIntermediateKey> {
+    let key = if *column_type == ColumnType::Str {
+        let fallback_dict = Dictionary::empty();
+        let term_dict = str_dict_column
+            .as_ref()
+            .map(|el| el.dictionary())
+            .unwrap_or_else(|| &fallback_dict);
+
+        let mut buffer = Vec::new();
+        term_dict.ord_to_term(val, &mut buffer)?;
+        CompositeIntermediateKey::Str(
+            String::from_utf8(buffer.to_vec()).expect("could not convert to String"),
+        )
+    } else if *column_type == ColumnType::DateTime {
+        let val = i64::from_u64(val);
+        CompositeIntermediateKey::DateTime(val)
+    } else if *column_type == ColumnType::Bool {
+        let val = bool::from_u64(val);
+        CompositeIntermediateKey::Bool(val)
+    } else if *column_type == ColumnType::IpAddr {
+        let compact_space_accessor = column
+            .values
+            .clone()
+            .downcast_arc::<CompactSpaceU64Accessor>()
+            .map_err(|_| {
+                TantivyError::AggregationError(crate::aggregation::AggregationError::InternalError(
+                    "Type mismatch: Could not downcast to CompactSpaceU64Accessor".to_string(),
+                ))
+            })?;
+        let val: u128 = compact_space_accessor.compact_to_u128(val as u32);
+        let val = Ipv6Addr::from_u128(val);
+        CompositeIntermediateKey::IpAddr(val)
+    } else if *column_type == ColumnType::U64 {
+        CompositeIntermediateKey::U64(val)
+    } else if *column_type == ColumnType::I64 {
+        CompositeIntermediateKey::I64(i64::from_u64(val))
+    } else {
+        let val = f64::from_u64(val);
+        let val: NumericalValue = val.into();
+
+        match val.normalize() {
+            NumericalValue::U64(val) => CompositeIntermediateKey::U64(val),
+            NumericalValue::I64(val) => CompositeIntermediateKey::I64(val),
+            NumericalValue::F64(val) => CompositeIntermediateKey::F64(val),
+        }
+    };
+    Ok(key)
+}
+
+/// Browse through the cardinal product obtained by the different values of the doc composite key
+/// sources.
+///
+/// For each of those tuple-key, that are after the limit key, we call collect_bucket_with_limit.
+struct CompositeKeyVisitor<'a> {
+    doc_id: crate::DocId,
+    composite_agg_data: &'a CompositeAggReqData,
+    buckets: &'a mut DynArrayHeapMap<InternalValueRepr, CompositeBucketCollector>,
+    sub_agg: &'a mut Option<CachedSubAggs<HighCardSubAggCache>>,
+    bucket_id_provider: &'a mut BucketIdProvider,
+    sub_level_values: SmallVec<[InternalValueRepr; MAX_DYN_ARRAY_SIZE]>,
+}
+
+impl CompositeKeyVisitor<'_> {
+    /// Depth-first walk of the accessors to build the composite key combinations
+    /// and update the buckets.
+    ///
+    /// `source_idx` is the current source index in the recursion.
+    /// `is_on_after_key` tracks whether we still need to consider the after_key
+    /// for pruning at this level and below.
+    fn visit(&mut self, source_idx: usize, is_on_after_key: bool) -> crate::Result<()> {
+        if source_idx == self.composite_agg_data.req.sources.len() {
+            if !is_on_after_key {
+                collect_bucket_with_limit(
+                    self.doc_id,
+                    self.composite_agg_data.req.size as usize,
+                    self.buckets,
+                    &self.sub_level_values,
+                    self.sub_agg,
+                    self.bucket_id_provider,
+                );
+            }
+            return Ok(());
+        }
+
+        let current_level_accessors = &self.composite_agg_data.composite_accessors[source_idx];
+        let current_level_source = &self.composite_agg_data.req.sources[source_idx];
+        let mut missing = true;
+        for (accessor_idx, accessor) in current_level_accessors.accessors.iter().enumerate() {
+            let values = accessor.column.values_for_doc(self.doc_id);
+            for value in values {
+                missing = false;
+                match current_level_source {
+                    CompositeAggregationSource::Terms(_) => {
+                        let preceeds_after_key_type =
+                            accessor_idx < current_level_accessors.after_key_accessor_idx;
+                        if is_on_after_key && preceeds_after_key_type {
+                            break;
+                        }
+                        let matches_after_key_type =
+                            accessor_idx == current_level_accessors.after_key_accessor_idx;
+
+                        if matches_after_key_type && is_on_after_key {
+                            let should_skip = match current_level_source.order() {
+                                Order::Asc => current_level_accessors.after_key.gt(value),
+                                Order::Desc => current_level_accessors.after_key.lt(value),
+                            };
+                            if should_skip {
+                                continue;
+                            }
+                        }
+                        self.sub_level_values.push(InternalValueRepr::new_term(
+                            value,
+                            accessor_idx as u8,
+                            current_level_source.order(),
+                        ));
+                        let still_on_after_key = matches_after_key_type
+                            && current_level_accessors.after_key.equals(value);
+                        self.visit(source_idx + 1, is_on_after_key && still_on_after_key)?;
+                        self.sub_level_values.pop();
+                    }
+                    CompositeAggregationSource::Histogram(source) => {
+                        let float_value = match accessor.column_type {
+                            ColumnType::U64 => value as f64,
+                            ColumnType::I64 => i64::from_u64(value) as f64,
+                            ColumnType::DateTime => i64::from_u64(value) as f64 / 1_000_000.,
+                            ColumnType::F64 => f64::from_u64(value),
+                            _ => {
+                                panic!(
+                                    "unexpected type {:?}. This should not happen",
+                                    accessor.column_type
+                                )
+                            }
+                        };
+                        let bucket_index = (float_value / source.interval).floor() as i64;
+                        let bucket_value = i64::to_u64(bucket_index);
+                        if is_on_after_key {
+                            let should_skip = match current_level_source.order() {
+                                Order::Asc => current_level_accessors.after_key.gt(bucket_value),
+                                Order::Desc => current_level_accessors.after_key.lt(bucket_value),
+                            };
+                            if should_skip {
+                                continue;
+                            }
+                        }
+                        self.sub_level_values.push(InternalValueRepr::new_histogram(
+                            bucket_value,
+                            current_level_source.order(),
+                        ));
+                        let still_on_after_key =
+                            current_level_accessors.after_key.equals(bucket_value);
+                        self.visit(source_idx + 1, is_on_after_key && still_on_after_key)?;
+                        self.sub_level_values.pop();
+                    }
+                    CompositeAggregationSource::DateHistogram(_) => {
+                        let value_ns = match accessor.column_type {
+                            ColumnType::DateTime => i64::from_u64(value),
+                            _ => {
+                                panic!(
+                                    "unexpected type {:?}. This should not happen",
+                                    accessor.column_type
+                                )
+                            }
+                        };
+                        let bucket_index = match accessor.date_histogram_interval {
+                            PrecomputedDateInterval::FixedNanoseconds(fixed_interval_ns) => {
+                                (value_ns / fixed_interval_ns) * fixed_interval_ns
+                            }
+                            PrecomputedDateInterval::Calendar(CalendarInterval::Year) => {
+                                calendar_interval::try_year_bucket(value_ns)?
+                            }
+                            PrecomputedDateInterval::Calendar(CalendarInterval::Month) => {
+                                calendar_interval::try_month_bucket(value_ns)?
+                            }
+                            PrecomputedDateInterval::Calendar(CalendarInterval::Week) => {
+                                calendar_interval::week_bucket(value_ns)
+                            }
+                            PrecomputedDateInterval::NotApplicable => {
+                                panic!("interval not precomputed for date histogram source")
+                            }
+                        };
+                        let bucket_value = i64::to_u64(bucket_index);
+                        if is_on_after_key {
+                            let should_skip = match current_level_source.order() {
+                                Order::Asc => current_level_accessors.after_key.gt(bucket_value),
+                                Order::Desc => current_level_accessors.after_key.lt(bucket_value),
+                            };
+                            if should_skip {
+                                continue;
+                            }
+                        }
+                        self.sub_level_values.push(InternalValueRepr::new_histogram(
+                            bucket_value,
+                            current_level_source.order(),
+                        ));
+                        let still_on_after_key =
+                            current_level_accessors.after_key.equals(bucket_value);
+                        self.visit(source_idx + 1, is_on_after_key && still_on_after_key)?;
+                        self.sub_level_values.pop();
+                    }
+                };
+            }
+        }
+        if missing && current_level_source.missing_bucket() {
+            if is_on_after_key && current_level_accessors.skip_missing {
+                return Ok(());
+            }
+            self.sub_level_values.push(InternalValueRepr::new_missing(
+                current_level_source.order(),
+                current_level_source.missing_order(),
+            ));
+            self.visit(
+                source_idx + 1,
+                is_on_after_key && current_level_accessors.is_after_key_explicit_missing,
+            )?;
+            self.sub_level_values.pop();
+        }
+        Ok(())
+    }
+}

src/aggregation/bucket/composite/map.rs

@@ -0,0 +1,329 @@
+use std::collections::BinaryHeap;
+use std::fmt::Debug;
+use std::hash::Hash;
+
+use rustc_hash::FxHashMap;
+use smallvec::SmallVec;
+
+use crate::TantivyError;
+
+/// Map backed by a hash map for fast access and a binary heap to track the
+/// highest key. The key is an array of fixed size S.
+#[derive(Clone, Debug)]
+struct ArrayHeapMap<K: Ord, V, const S: usize> {
+    pub(crate) buckets: FxHashMap<[K; S], V>,
+    pub(crate) heap: BinaryHeap<[K; S]>,
+}
+
+impl<K: Ord, V, const S: usize> Default for ArrayHeapMap<K, V, S> {
+    fn default() -> Self {
+        ArrayHeapMap {
+            buckets: FxHashMap::default(),
+            heap: BinaryHeap::default(),
+        }
+    }
+}
+
+impl<K: Eq + Hash + Clone + Ord, V, const S: usize> ArrayHeapMap<K, V, S> {
+    /// Panics if the length of `key` is not S.
+    fn get_or_insert_with<F: FnOnce() -> V>(&mut self, key: &[K], f: F) -> &mut V {
+        let key_array: &[K; S] = key.try_into().expect("Key length mismatch");
+        self.buckets.entry(key_array.clone()).or_insert_with(|| {
+            self.heap.push(key_array.clone());
+            f()
+        })
+    }
+
+    /// Panics if the length of `key` is not S.
+    fn get_mut(&mut self, key: &[K]) -> Option<&mut V> {
+        let key_array: &[K; S] = key.try_into().expect("Key length mismatch");
+        self.buckets.get_mut(key_array)
+    }
+
+    fn peek_highest(&self) -> Option<&[K]> {
+        self.heap.peek().map(|k_array| k_array.as_slice())
+    }
+
+    fn evict_highest(&mut self) {
+        if let Some(highest) = self.heap.pop() {
+            self.buckets.remove(&highest);
+        }
+    }
+
+    fn memory_consumption(&self) -> u64 {
+        let key_size = std::mem::size_of::<[K; S]>();
+        let map_size = (key_size + std::mem::size_of::<V>()) * self.buckets.capacity();
+        let heap_size = key_size * self.heap.capacity();
+        (map_size + heap_size) as u64
+    }
+}
+
+impl<K: Copy + Ord + Clone + 'static, V: 'static, const S: usize> ArrayHeapMap<K, V, S> {
+    fn into_iter(self) -> Box<dyn Iterator<Item = (SmallVec<[K; MAX_DYN_ARRAY_SIZE]>, V)>> {
+        Box::new(
+            self.buckets
+                .into_iter()
+                .map(|(k, v)| (SmallVec::from_slice(&k), v)),
+        )
+    }
+}
+
+pub(super) const MAX_DYN_ARRAY_SIZE: usize = 16;
+const MAX_DYN_ARRAY_SIZE_PLUS_ONE: usize = MAX_DYN_ARRAY_SIZE + 1;
+
+/// A map optimized for memory footprint, fast access and efficient eviction of
+/// the highest key.
+///
+/// Keys are inlined arrays of size 1 to [MAX_DYN_ARRAY_SIZE] but for a given
+/// instance the key size is fixed. This allows to avoid heap allocations for the
+/// keys.
+#[derive(Clone, Debug)]
+pub(super) struct DynArrayHeapMap<K: Ord, V>(DynArrayHeapMapInner<K, V>);
+
+/// Wrapper around ArrayHeapMap to dynamically dispatch on the array size.
+#[derive(Clone, Debug)]
+enum DynArrayHeapMapInner<K: Ord, V> {
+    Dim1(ArrayHeapMap<K, V, 1>),
+    Dim2(ArrayHeapMap<K, V, 2>),
+    Dim3(ArrayHeapMap<K, V, 3>),
+    Dim4(ArrayHeapMap<K, V, 4>),
+    Dim5(ArrayHeapMap<K, V, 5>),
+    Dim6(ArrayHeapMap<K, V, 6>),
+    Dim7(ArrayHeapMap<K, V, 7>),
+    Dim8(ArrayHeapMap<K, V, 8>),
+    Dim9(ArrayHeapMap<K, V, 9>),
+    Dim10(ArrayHeapMap<K, V, 10>),
+    Dim11(ArrayHeapMap<K, V, 11>),
+    Dim12(ArrayHeapMap<K, V, 12>),
+    Dim13(ArrayHeapMap<K, V, 13>),
+    Dim14(ArrayHeapMap<K, V, 14>),
+    Dim15(ArrayHeapMap<K, V, 15>),
+    Dim16(ArrayHeapMap<K, V, 16>),
+}
+
+impl<K: Ord, V> DynArrayHeapMap<K, V> {
+    /// Creates a new heap map with dynamic array keys of size `key_dimension`.
+    pub(super) fn try_new(key_dimension: usize) -> crate::Result<Self> {
+        let inner = match key_dimension {
+            0 => {
+                return Err(TantivyError::InvalidArgument(
+                    "DynArrayHeapMap dimension must be at least 1".to_string(),
+                ))
+            }
+            1 => DynArrayHeapMapInner::Dim1(ArrayHeapMap::default()),
+            2 => DynArrayHeapMapInner::Dim2(ArrayHeapMap::default()),
+            3 => DynArrayHeapMapInner::Dim3(ArrayHeapMap::default()),
+            4 => DynArrayHeapMapInner::Dim4(ArrayHeapMap::default()),
+            5 => DynArrayHeapMapInner::Dim5(ArrayHeapMap::default()),
+            6 => DynArrayHeapMapInner::Dim6(ArrayHeapMap::default()),
+            7 => DynArrayHeapMapInner::Dim7(ArrayHeapMap::default()),
+            8 => DynArrayHeapMapInner::Dim8(ArrayHeapMap::default()),
+            9 => DynArrayHeapMapInner::Dim9(ArrayHeapMap::default()),
+            10 => DynArrayHeapMapInner::Dim10(ArrayHeapMap::default()),
+            11 => DynArrayHeapMapInner::Dim11(ArrayHeapMap::default()),
+            12 => DynArrayHeapMapInner::Dim12(ArrayHeapMap::default()),
+            13 => DynArrayHeapMapInner::Dim13(ArrayHeapMap::default()),
+            14 => DynArrayHeapMapInner::Dim14(ArrayHeapMap::default()),
+            15 => DynArrayHeapMapInner::Dim15(ArrayHeapMap::default()),
+            16 => DynArrayHeapMapInner::Dim16(ArrayHeapMap::default()),
+            MAX_DYN_ARRAY_SIZE_PLUS_ONE.. => {
+                return Err(TantivyError::InvalidArgument(format!(
+                    "DynArrayHeapMap supports maximum {MAX_DYN_ARRAY_SIZE} dimensions, got \
+                     {key_dimension}",
+                )))
+            }
+        };
+        Ok(DynArrayHeapMap(inner))
+    }
+
+    /// Number of elements in the map. This is not the dimension of the keys.
+    pub(super) fn size(&self) -> usize {
+        match &self.0 {
+            DynArrayHeapMapInner::Dim1(map) => map.buckets.len(),
+            DynArrayHeapMapInner::Dim2(map) => map.buckets.len(),
+            DynArrayHeapMapInner::Dim3(map) => map.buckets.len(),
+            DynArrayHeapMapInner::Dim4(map) => map.buckets.len(),
+            DynArrayHeapMapInner::Dim5(map) => map.buckets.len(),
+            DynArrayHeapMapInner::Dim6(map) => map.buckets.len(),
+            DynArrayHeapMapInner::Dim7(map) => map.buckets.len(),
+            DynArrayHeapMapInner::Dim8(map) => map.buckets.len(),
+            DynArrayHeapMapInner::Dim9(map) => map.buckets.len(),
+            DynArrayHeapMapInner::Dim10(map) => map.buckets.len(),
+            DynArrayHeapMapInner::Dim11(map) => map.buckets.len(),
+            DynArrayHeapMapInner::Dim12(map) => map.buckets.len(),
+            DynArrayHeapMapInner::Dim13(map) => map.buckets.len(),
+            DynArrayHeapMapInner::Dim14(map) => map.buckets.len(),
+            DynArrayHeapMapInner::Dim15(map) => map.buckets.len(),
+            DynArrayHeapMapInner::Dim16(map) => map.buckets.len(),
+        }
+    }
+}
+
+impl<K: Ord + Hash + Clone, V> DynArrayHeapMap<K, V> {
+    /// Get a mutable reference to the value corresponding to `key` or inserts a new
+    /// value created by calling `f`.
+    ///
+    /// Panics if the length of `key` does not match the key dimension of the map.
+    pub(super) fn get_or_insert_with<F: FnOnce() -> V>(&mut self, key: &[K], f: F) -> &mut V {
+        match &mut self.0 {
+            DynArrayHeapMapInner::Dim1(map) => map.get_or_insert_with(key, f),
+            DynArrayHeapMapInner::Dim2(map) => map.get_or_insert_with(key, f),
+            DynArrayHeapMapInner::Dim3(map) => map.get_or_insert_with(key, f),
+            DynArrayHeapMapInner::Dim4(map) => map.get_or_insert_with(key, f),
+            DynArrayHeapMapInner::Dim5(map) => map.get_or_insert_with(key, f),
+            DynArrayHeapMapInner::Dim6(map) => map.get_or_insert_with(key, f),
+            DynArrayHeapMapInner::Dim7(map) => map.get_or_insert_with(key, f),
+            DynArrayHeapMapInner::Dim8(map) => map.get_or_insert_with(key, f),
+            DynArrayHeapMapInner::Dim9(map) => map.get_or_insert_with(key, f),
+            DynArrayHeapMapInner::Dim10(map) => map.get_or_insert_with(key, f),
+            DynArrayHeapMapInner::Dim11(map) => map.get_or_insert_with(key, f),
+            DynArrayHeapMapInner::Dim12(map) => map.get_or_insert_with(key, f),
+            DynArrayHeapMapInner::Dim13(map) => map.get_or_insert_with(key, f),
+            DynArrayHeapMapInner::Dim14(map) => map.get_or_insert_with(key, f),
+            DynArrayHeapMapInner::Dim15(map) => map.get_or_insert_with(key, f),
+            DynArrayHeapMapInner::Dim16(map) => map.get_or_insert_with(key, f),
+        }
+    }
+
+    /// Returns a mutable reference to the value corresponding to `key`.
+    ///
+    /// Panics if the length of `key` does not match the key dimension of the map.
+    pub fn get_mut(&mut self, key: &[K]) -> Option<&mut V> {
+        match &mut self.0 {
+            DynArrayHeapMapInner::Dim1(map) => map.get_mut(key),
+            DynArrayHeapMapInner::Dim2(map) => map.get_mut(key),
+            DynArrayHeapMapInner::Dim3(map) => map.get_mut(key),
+            DynArrayHeapMapInner::Dim4(map) => map.get_mut(key),
+            DynArrayHeapMapInner::Dim5(map) => map.get_mut(key),
+            DynArrayHeapMapInner::Dim6(map) => map.get_mut(key),
+            DynArrayHeapMapInner::Dim7(map) => map.get_mut(key),
+            DynArrayHeapMapInner::Dim8(map) => map.get_mut(key),
+            DynArrayHeapMapInner::Dim9(map) => map.get_mut(key),
+            DynArrayHeapMapInner::Dim10(map) => map.get_mut(key),
+            DynArrayHeapMapInner::Dim11(map) => map.get_mut(key),
+            DynArrayHeapMapInner::Dim12(map) => map.get_mut(key),
+            DynArrayHeapMapInner::Dim13(map) => map.get_mut(key),
+            DynArrayHeapMapInner::Dim14(map) => map.get_mut(key),
+            DynArrayHeapMapInner::Dim15(map) => map.get_mut(key),
+            DynArrayHeapMapInner::Dim16(map) => map.get_mut(key),
+        }
+    }
+
+    /// Returns a reference to the highest key in the map.
+    pub(super) fn peek_highest(&self) -> Option<&[K]> {
+        match &self.0 {
+            DynArrayHeapMapInner::Dim1(map) => map.peek_highest(),
+            DynArrayHeapMapInner::Dim2(map) => map.peek_highest(),
+            DynArrayHeapMapInner::Dim3(map) => map.peek_highest(),
+            DynArrayHeapMapInner::Dim4(map) => map.peek_highest(),
+            DynArrayHeapMapInner::Dim5(map) => map.peek_highest(),
+            DynArrayHeapMapInner::Dim6(map) => map.peek_highest(),
+            DynArrayHeapMapInner::Dim7(map) => map.peek_highest(),
+            DynArrayHeapMapInner::Dim8(map) => map.peek_highest(),
+            DynArrayHeapMapInner::Dim9(map) => map.peek_highest(),
+            DynArrayHeapMapInner::Dim10(map) => map.peek_highest(),
+            DynArrayHeapMapInner::Dim11(map) => map.peek_highest(),
+            DynArrayHeapMapInner::Dim12(map) => map.peek_highest(),
+            DynArrayHeapMapInner::Dim13(map) => map.peek_highest(),
+            DynArrayHeapMapInner::Dim14(map) => map.peek_highest(),
+            DynArrayHeapMapInner::Dim15(map) => map.peek_highest(),
+            DynArrayHeapMapInner::Dim16(map) => map.peek_highest(),
+        }
+    }
+
+    /// Removes the entry with the highest key from the map.
+    pub(super) fn evict_highest(&mut self) {
+        match &mut self.0 {
+            DynArrayHeapMapInner::Dim1(map) => map.evict_highest(),
+            DynArrayHeapMapInner::Dim2(map) => map.evict_highest(),
+            DynArrayHeapMapInner::Dim3(map) => map.evict_highest(),
+            DynArrayHeapMapInner::Dim4(map) => map.evict_highest(),
+            DynArrayHeapMapInner::Dim5(map) => map.evict_highest(),
+            DynArrayHeapMapInner::Dim6(map) => map.evict_highest(),
+            DynArrayHeapMapInner::Dim7(map) => map.evict_highest(),
+            DynArrayHeapMapInner::Dim8(map) => map.evict_highest(),
+            DynArrayHeapMapInner::Dim9(map) => map.evict_highest(),
+            DynArrayHeapMapInner::Dim10(map) => map.evict_highest(),
+            DynArrayHeapMapInner::Dim11(map) => map.evict_highest(),
+            DynArrayHeapMapInner::Dim12(map) => map.evict_highest(),
+            DynArrayHeapMapInner::Dim13(map) => map.evict_highest(),
+            DynArrayHeapMapInner::Dim14(map) => map.evict_highest(),
+            DynArrayHeapMapInner::Dim15(map) => map.evict_highest(),
+            DynArrayHeapMapInner::Dim16(map) => map.evict_highest(),
+        }
+    }
+
+    pub(crate) fn memory_consumption(&self) -> u64 {
+        match &self.0 {
+            DynArrayHeapMapInner::Dim1(map) => map.memory_consumption(),
+            DynArrayHeapMapInner::Dim2(map) => map.memory_consumption(),
+            DynArrayHeapMapInner::Dim3(map) => map.memory_consumption(),
+            DynArrayHeapMapInner::Dim4(map) => map.memory_consumption(),
+            DynArrayHeapMapInner::Dim5(map) => map.memory_consumption(),
+            DynArrayHeapMapInner::Dim6(map) => map.memory_consumption(),
+            DynArrayHeapMapInner::Dim7(map) => map.memory_consumption(),
+            DynArrayHeapMapInner::Dim8(map) => map.memory_consumption(),
+            DynArrayHeapMapInner::Dim9(map) => map.memory_consumption(),
+            DynArrayHeapMapInner::Dim10(map) => map.memory_consumption(),
+            DynArrayHeapMapInner::Dim11(map) => map.memory_consumption(),
+            DynArrayHeapMapInner::Dim12(map) => map.memory_consumption(),
+            DynArrayHeapMapInner::Dim13(map) => map.memory_consumption(),
+            DynArrayHeapMapInner::Dim14(map) => map.memory_consumption(),
+            DynArrayHeapMapInner::Dim15(map) => map.memory_consumption(),
+            DynArrayHeapMapInner::Dim16(map) => map.memory_consumption(),
+        }
+    }
+}
+
+impl<K: Ord + Clone + Copy + 'static, V: 'static> DynArrayHeapMap<K, V> {
+    /// Turns this map into an iterator over key-value pairs.
+    pub fn into_iter(self) -> impl Iterator<Item = (SmallVec<[K; MAX_DYN_ARRAY_SIZE]>, V)> {
+        match self.0 {
+            DynArrayHeapMapInner::Dim1(map) => map.into_iter(),
+            DynArrayHeapMapInner::Dim2(map) => map.into_iter(),
+            DynArrayHeapMapInner::Dim3(map) => map.into_iter(),
+            DynArrayHeapMapInner::Dim4(map) => map.into_iter(),
+            DynArrayHeapMapInner::Dim5(map) => map.into_iter(),
+            DynArrayHeapMapInner::Dim6(map) => map.into_iter(),
+            DynArrayHeapMapInner::Dim7(map) => map.into_iter(),
+            DynArrayHeapMapInner::Dim8(map) => map.into_iter(),
+            DynArrayHeapMapInner::Dim9(map) => map.into_iter(),
+            DynArrayHeapMapInner::Dim10(map) => map.into_iter(),
+            DynArrayHeapMapInner::Dim11(map) => map.into_iter(),
+            DynArrayHeapMapInner::Dim12(map) => map.into_iter(),
+            DynArrayHeapMapInner::Dim13(map) => map.into_iter(),
+            DynArrayHeapMapInner::Dim14(map) => map.into_iter(),
+            DynArrayHeapMapInner::Dim15(map) => map.into_iter(),
+            DynArrayHeapMapInner::Dim16(map) => map.into_iter(),
+        }
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn test_dyn_array_heap_map() {
+        let mut map = DynArrayHeapMap::<u32, &str>::try_new(2).unwrap();
+        // insert
+        let key1 = [1u32, 2u32];
+        let key2 = [2u32, 1u32];
+        map.get_or_insert_with(&key1, || "a");
+        map.get_or_insert_with(&key2, || "b");
+        assert_eq!(map.size(), 2);
+
+        // evict highest
+        assert_eq!(map.peek_highest(), Some(&key2[..]));
+        map.evict_highest();
+        assert_eq!(map.size(), 1);
+        assert_eq!(map.peek_highest(), Some(&key1[..]));
+
+        // into_iter
+        let mut iter = map.into_iter();
+        let (k, v) = iter.next().unwrap();
+        assert_eq!(k.as_slice(), &key1);
+        assert_eq!(v, "a");
+        assert_eq!(iter.next(), None);
+    }
+}

src/aggregation/bucket/composite/numeric_types.rs

@@ -0,0 +1,460 @@
+/// This module helps comparing numerical values of different types (i64, u64
+/// and f64).
+pub(super) mod num_cmp {
+    use std::cmp::Ordering;
+
+    use crate::TantivyError;
+
+    pub fn cmp_i64_f64(left_i: i64, right_f: f64) -> crate::Result<Ordering> {
+        if right_f.is_nan() {
+            return Err(TantivyError::InvalidArgument(
+                "NaN comparison is not supported".to_string(),
+            ));
+        }
+
+        // If right_f is < i64::MIN then left_i > right_f (i64::MIN=-2^63 can be
+        // exactly represented as f64)
+        if right_f < i64::MIN as f64 {
+            return Ok(Ordering::Greater);
+        }
+        // If right_f is >= i64::MAX then left_i < right_f (i64::MAX=2^63-1 cannot
+        // be exactly represented as f64)
+        if right_f >= i64::MAX as f64 {
+            return Ok(Ordering::Less);
+        }
+
+        // Now right_f is in (i64::MIN, i64::MAX), so `right_f as i64` is
+        // well-defined (truncation toward 0)
+        let right_as_i = right_f as i64;
+
+        let result = match left_i.cmp(&right_as_i) {
+            Ordering::Less => Ordering::Less,
+            Ordering::Greater => Ordering::Greater,
+            Ordering::Equal => {
+                // they have the same integer part, compare the fraction
+                let rem = right_f - (right_as_i as f64);
+                if rem == 0.0 {
+                    Ordering::Equal
+                } else if right_f > 0.0 {
+                    Ordering::Less
+                } else {
+                    Ordering::Greater
+                }
+            }
+        };
+        Ok(result)
+    }
+
+    pub fn cmp_u64_f64(left_u: u64, right_f: f64) -> crate::Result<Ordering> {
+        if right_f.is_nan() {
+            return Err(TantivyError::InvalidArgument(
+                "NaN comparison is not supported".to_string(),
+            ));
+        }
+
+        // Negative floats are always less than any u64 >= 0
+        if right_f < 0.0 {
+            return Ok(Ordering::Greater);
+        }
+
+        // If right_f is >= u64::MAX then left_u < right_f (u64::MAX=2^64-1 cannot be exactly)
+        let max_as_f = u64::MAX as f64;
+        if right_f > max_as_f {
+            return Ok(Ordering::Less);
+        }
+
+        // Now right_f is in (0, u64::MAX), so `right_f as u64` is well-defined
+        // (truncation toward 0)
+        let right_as_u = right_f as u64;
+
+        let result = match left_u.cmp(&right_as_u) {
+            Ordering::Less => Ordering::Less,
+            Ordering::Greater => Ordering::Greater,
+            Ordering::Equal => {
+                // they have the same integer part, compare the fraction
+                let rem = right_f - (right_as_u as f64);
+                if rem == 0.0 {
+                    Ordering::Equal
+                } else {
+                    Ordering::Less
+                }
+            }
+        };
+        Ok(result)
+    }
+
+    pub fn cmp_i64_u64(left_i: i64, right_u: u64) -> Ordering {
+        if left_i < 0 {
+            Ordering::Less
+        } else {
+            let left_as_u = left_i as u64;
+            left_as_u.cmp(&right_u)
+        }
+    }
+}
+
+/// This module helps projecting numerical values to other numerical types.
+/// When the target value space cannot exactly represent the source value, the
+/// next representable value is returned (or AfterLast if the source value is
+/// larger than the largest representable value).
+///
+/// All functions in this module assume that f64 values are not NaN.
+pub(super) mod num_proj {
+    #[derive(Debug, PartialEq)]
+    pub enum ProjectedNumber<T> {
+        Exact(T),
+        Next(T),
+        AfterLast,
+    }
+
+    pub fn i64_to_u64(value: i64) -> ProjectedNumber<u64> {
+        if value < 0 {
+            ProjectedNumber::Next(0)
+        } else {
+            ProjectedNumber::Exact(value as u64)
+        }
+    }
+
+    pub fn u64_to_i64(value: u64) -> ProjectedNumber<i64> {
+        if value > i64::MAX as u64 {
+            ProjectedNumber::AfterLast
+        } else {
+            ProjectedNumber::Exact(value as i64)
+        }
+    }
+
+    pub fn f64_to_u64(value: f64) -> ProjectedNumber<u64> {
+        if value < 0.0 {
+            ProjectedNumber::Next(0)
+        } else if value > u64::MAX as f64 {
+            ProjectedNumber::AfterLast
+        } else if value.fract() == 0.0 {
+            ProjectedNumber::Exact(value as u64)
+        } else {
+            // casting f64 to u64 truncates toward zero
+            ProjectedNumber::Next(value as u64 + 1)
+        }
+    }
+
+    pub fn f64_to_i64(value: f64) -> ProjectedNumber<i64> {
+        if value < (i64::MIN as f64) {
+            ProjectedNumber::Next(i64::MIN)
+        } else if value >= (i64::MAX as f64) {
+            ProjectedNumber::AfterLast
+        } else if value.fract() == 0.0 {
+            ProjectedNumber::Exact(value as i64)
+        } else if value > 0.0 {
+            // casting f64 to i64 truncates toward zero
+            ProjectedNumber::Next(value as i64 + 1)
+        } else {
+            ProjectedNumber::Next(value as i64)
+        }
+    }
+
+    pub fn i64_to_f64(value: i64) -> ProjectedNumber<f64> {
+        let value_f = value as f64;
+        let k_roundtrip = value_f as i64;
+        if k_roundtrip == value {
+            // between -2^53 and 2^53 all i64 are exactly represented as f64
+            ProjectedNumber::Exact(value_f)
+        } else {
+            // for very large/small i64 values, it is approximated to the closest f64
+            if k_roundtrip > value {
+                ProjectedNumber::Next(value_f)
+            } else {
+                ProjectedNumber::Next(value_f.next_up())
+            }
+        }
+    }
+
+    pub fn u64_to_f64(value: u64) -> ProjectedNumber<f64> {
+        let value_f = value as f64;
+        let k_roundtrip = value_f as u64;
+        if k_roundtrip == value {
+            // between 0 and 2^53 all u64 are exactly represented as f64
+            ProjectedNumber::Exact(value_f)
+        } else if k_roundtrip > value {
+            ProjectedNumber::Next(value_f)
+        } else {
+            ProjectedNumber::Next(value_f.next_up())
+        }
+    }
+}
+
+#[cfg(test)]
+mod num_cmp_tests {
+    use std::cmp::Ordering;
+
+    use super::num_cmp::*;
+
+    #[test]
+    fn test_cmp_u64_f64() {
+        // Basic comparisons
+        assert_eq!(cmp_u64_f64(5, 5.0).unwrap(), Ordering::Equal);
+        assert_eq!(cmp_u64_f64(5, 6.0).unwrap(), Ordering::Less);
+        assert_eq!(cmp_u64_f64(6, 5.0).unwrap(), Ordering::Greater);
+        assert_eq!(cmp_u64_f64(0, 0.0).unwrap(), Ordering::Equal);
+        assert_eq!(cmp_u64_f64(0, 0.1).unwrap(), Ordering::Less);
+
+        // Negative float values should always be less than any u64
+        assert_eq!(cmp_u64_f64(0, -0.1).unwrap(), Ordering::Greater);
+        assert_eq!(cmp_u64_f64(5, -5.0).unwrap(), Ordering::Greater);
+        assert_eq!(cmp_u64_f64(u64::MAX, -1e20).unwrap(), Ordering::Greater);
+
+        // Tests with extreme values
+        assert_eq!(cmp_u64_f64(u64::MAX, 1e20).unwrap(), Ordering::Less);
+
+        // Precision edge cases: large u64 that loses precision when converted to f64
+        // => 2^54, exactly represented as f64
+        let large_f64 = 18_014_398_509_481_984.0;
+        let large_u64 = 18_014_398_509_481_984;
+        // prove that large_u64 is exactly represented as f64
+        assert_eq!(large_u64 as f64, large_f64);
+        assert_eq!(cmp_u64_f64(large_u64, large_f64).unwrap(), Ordering::Equal);
+        // => (2^54 + 1) cannot be exactly represented in f64
+        let large_u64_plus_1 = 18_014_398_509_481_985;
+        // prove that it is represented as f64 by large_f64
+        assert_eq!(large_u64_plus_1 as f64, large_f64);
+        assert_eq!(
+            cmp_u64_f64(large_u64_plus_1, large_f64).unwrap(),
+            Ordering::Greater
+        );
+        // => (2^54 - 1) cannot be exactly represented in f64
+        let large_u64_minus_1 = 18_014_398_509_481_983;
+        // prove that it is also represented as f64 by large_f64
+        assert_eq!(large_u64_minus_1 as f64, large_f64);
+        assert_eq!(
+            cmp_u64_f64(large_u64_minus_1, large_f64).unwrap(),
+            Ordering::Less
+        );
+
+        // NaN comparison results in an error
+        assert!(cmp_u64_f64(0, f64::NAN).is_err());
+    }
+
+    #[test]
+    fn test_cmp_i64_f64() {
+        // Basic comparisons
+        assert_eq!(cmp_i64_f64(5, 5.0).unwrap(), Ordering::Equal);
+        assert_eq!(cmp_i64_f64(5, 6.0).unwrap(), Ordering::Less);
+        assert_eq!(cmp_i64_f64(6, 5.0).unwrap(), Ordering::Greater);
+        assert_eq!(cmp_i64_f64(-5, -5.0).unwrap(), Ordering::Equal);
+        assert_eq!(cmp_i64_f64(-5, -4.0).unwrap(), Ordering::Less);
+        assert_eq!(cmp_i64_f64(-4, -5.0).unwrap(), Ordering::Greater);
+        assert_eq!(cmp_i64_f64(-5, 5.0).unwrap(), Ordering::Less);
+        assert_eq!(cmp_i64_f64(5, -5.0).unwrap(), Ordering::Greater);
+        assert_eq!(cmp_i64_f64(0, -0.1).unwrap(), Ordering::Greater);
+        assert_eq!(cmp_i64_f64(0, 0.1).unwrap(), Ordering::Less);
+        assert_eq!(cmp_i64_f64(-1, -0.5).unwrap(), Ordering::Less);
+        assert_eq!(cmp_i64_f64(-1, 0.0).unwrap(), Ordering::Less);
+        assert_eq!(cmp_i64_f64(0, 0.0).unwrap(), Ordering::Equal);
+
+        // Tests with extreme values
+        assert_eq!(cmp_i64_f64(i64::MAX, 1e20).unwrap(), Ordering::Less);
+        assert_eq!(cmp_i64_f64(i64::MIN, -1e20).unwrap(), Ordering::Greater);
+
+        // Precision edge cases: large i64 that loses precision when converted to f64
+        // => 2^54, exactly represented as f64
+        let large_f64 = 18_014_398_509_481_984.0;
+        let large_i64 = 18_014_398_509_481_984;
+        // prove that large_i64 is exactly represented as f64
+        assert_eq!(large_i64 as f64, large_f64);
+        assert_eq!(cmp_i64_f64(large_i64, large_f64).unwrap(), Ordering::Equal);
+        // => (1_i64 << 54) + 1 cannot be exactly represented in f64
+        let large_i64_plus_1 = 18_014_398_509_481_985;
+        // prove that it is represented as f64 by large_f64
+        assert_eq!(large_i64_plus_1 as f64, large_f64);
+        assert_eq!(
+            cmp_i64_f64(large_i64_plus_1, large_f64).unwrap(),
+            Ordering::Greater
+        );
+        // => (1_i64 << 54) - 1 cannot be exactly represented in f64
+        let large_i64_minus_1 = 18_014_398_509_481_983;
+        // prove that it is also represented as f64 by large_f64
+        assert_eq!(large_i64_minus_1 as f64, large_f64);
+        assert_eq!(
+            cmp_i64_f64(large_i64_minus_1, large_f64).unwrap(),
+            Ordering::Less
+        );
+
+        // Same precision edge case but with negative values
+        // => -2^54, exactly represented as f64
+        let large_neg_f64 = -18_014_398_509_481_984.0;
+        let large_neg_i64 = -18_014_398_509_481_984;
+        // prove that large_neg_i64 is exactly represented as f64
+        assert_eq!(large_neg_i64 as f64, large_neg_f64);
+        assert_eq!(
+            cmp_i64_f64(large_neg_i64, large_neg_f64).unwrap(),
+            Ordering::Equal
+        );
+        // => (-2^54 + 1) cannot be exactly represented in f64
+        let large_neg_i64_plus_1 = -18_014_398_509_481_985;
+        // prove that it is represented as f64 by large_neg_f64
+        assert_eq!(large_neg_i64_plus_1 as f64, large_neg_f64);
+        assert_eq!(
+            cmp_i64_f64(large_neg_i64_plus_1, large_neg_f64).unwrap(),
+            Ordering::Less
+        );
+        // => (-2^54 - 1) cannot be exactly represented in f64
+        let large_neg_i64_minus_1 = -18_014_398_509_481_983;
+        // prove that it is also represented as f64 by large_neg_f64
+        assert_eq!(large_neg_i64_minus_1 as f64, large_neg_f64);
+        assert_eq!(
+            cmp_i64_f64(large_neg_i64_minus_1, large_neg_f64).unwrap(),
+            Ordering::Greater
+        );
+
+        // NaN comparison results in an error
+        assert!(cmp_i64_f64(0, f64::NAN).is_err());
+    }
+
+    #[test]
+    fn test_cmp_i64_u64() {
+        // Test with negative i64 values (should always be less than any u64)
+        assert_eq!(cmp_i64_u64(-1, 0), Ordering::Less);
+        assert_eq!(cmp_i64_u64(i64::MIN, 0), Ordering::Less);
+        assert_eq!(cmp_i64_u64(i64::MIN, u64::MAX), Ordering::Less);
+
+        // Test with positive i64 values
+        assert_eq!(cmp_i64_u64(0, 0), Ordering::Equal);
+        assert_eq!(cmp_i64_u64(1, 0), Ordering::Greater);
+        assert_eq!(cmp_i64_u64(1, 1), Ordering::Equal);
+        assert_eq!(cmp_i64_u64(0, 1), Ordering::Less);
+        assert_eq!(cmp_i64_u64(5, 10), Ordering::Less);
+        assert_eq!(cmp_i64_u64(10, 5), Ordering::Greater);
+
+        // Test with values near i64::MAX and u64 conversion
+        assert_eq!(cmp_i64_u64(i64::MAX, i64::MAX as u64), Ordering::Equal);
+        assert_eq!(cmp_i64_u64(i64::MAX, (i64::MAX as u64) + 1), Ordering::Less);
+        assert_eq!(cmp_i64_u64(i64::MAX, u64::MAX), Ordering::Less);
+    }
+}
+
+#[cfg(test)]
+mod num_proj_tests {
+    use super::num_proj::{self, ProjectedNumber};
+
+    #[test]
+    fn test_i64_to_u64() {
+        assert_eq!(num_proj::i64_to_u64(-1), ProjectedNumber::Next(0));
+        assert_eq!(num_proj::i64_to_u64(i64::MIN), ProjectedNumber::Next(0));
+        assert_eq!(num_proj::i64_to_u64(0), ProjectedNumber::Exact(0));
+        assert_eq!(num_proj::i64_to_u64(42), ProjectedNumber::Exact(42));
+        assert_eq!(
+            num_proj::i64_to_u64(i64::MAX),
+            ProjectedNumber::Exact(i64::MAX as u64)
+        );
+    }
+
+    #[test]
+    fn test_u64_to_i64() {
+        assert_eq!(num_proj::u64_to_i64(0), ProjectedNumber::Exact(0));
+        assert_eq!(num_proj::u64_to_i64(42), ProjectedNumber::Exact(42));
+        assert_eq!(
+            num_proj::u64_to_i64(i64::MAX as u64),
+            ProjectedNumber::Exact(i64::MAX)
+        );
+        assert_eq!(
+            num_proj::u64_to_i64((i64::MAX as u64) + 1),
+            ProjectedNumber::AfterLast
+        );
+        assert_eq!(num_proj::u64_to_i64(u64::MAX), ProjectedNumber::AfterLast);
+    }
+
+    #[test]
+    fn test_f64_to_u64() {
+        assert_eq!(num_proj::f64_to_u64(-1e25), ProjectedNumber::Next(0));
+        assert_eq!(num_proj::f64_to_u64(-0.1), ProjectedNumber::Next(0));
+        assert_eq!(num_proj::f64_to_u64(1e20), ProjectedNumber::AfterLast);
+        assert_eq!(
+            num_proj::f64_to_u64(f64::INFINITY),
+            ProjectedNumber::AfterLast
+        );
+        assert_eq!(num_proj::f64_to_u64(0.0), ProjectedNumber::Exact(0));
+        assert_eq!(num_proj::f64_to_u64(42.0), ProjectedNumber::Exact(42));
+        assert_eq!(num_proj::f64_to_u64(0.5), ProjectedNumber::Next(1));
+        assert_eq!(num_proj::f64_to_u64(42.1), ProjectedNumber::Next(43));
+    }
+
+    #[test]
+    fn test_f64_to_i64() {
+        assert_eq!(num_proj::f64_to_i64(-1e20), ProjectedNumber::Next(i64::MIN));
+        assert_eq!(
+            num_proj::f64_to_i64(f64::NEG_INFINITY),
+            ProjectedNumber::Next(i64::MIN)
+        );
+        assert_eq!(num_proj::f64_to_i64(1e20), ProjectedNumber::AfterLast);
+        assert_eq!(
+            num_proj::f64_to_i64(f64::INFINITY),
+            ProjectedNumber::AfterLast
+        );
+        assert_eq!(num_proj::f64_to_i64(0.0), ProjectedNumber::Exact(0));
+        assert_eq!(num_proj::f64_to_i64(42.0), ProjectedNumber::Exact(42));
+        assert_eq!(num_proj::f64_to_i64(-42.0), ProjectedNumber::Exact(-42));
+        assert_eq!(num_proj::f64_to_i64(0.5), ProjectedNumber::Next(1));
+        assert_eq!(num_proj::f64_to_i64(42.1), ProjectedNumber::Next(43));
+        assert_eq!(num_proj::f64_to_i64(-0.5), ProjectedNumber::Next(0));
+        assert_eq!(num_proj::f64_to_i64(-42.1), ProjectedNumber::Next(-42));
+    }
+
+    #[test]
+    fn test_i64_to_f64() {
+        assert_eq!(num_proj::i64_to_f64(0), ProjectedNumber::Exact(0.0));
+        assert_eq!(num_proj::i64_to_f64(42), ProjectedNumber::Exact(42.0));
+        assert_eq!(num_proj::i64_to_f64(-42), ProjectedNumber::Exact(-42.0));
+
+        let max_exact = 9_007_199_254_740_992; // 2^53
+        assert_eq!(
+            num_proj::i64_to_f64(max_exact),
+            ProjectedNumber::Exact(max_exact as f64)
+        );
+
+        // Test values that cannot be exactly represented as f64 (integers above 2^53)
+        let large_i64 = 9_007_199_254_740_993; // 2^53 + 1
+        let closest_f64 = 9_007_199_254_740_992.0;
+        assert_eq!(large_i64 as f64, closest_f64);
+        if let ProjectedNumber::Next(val) = num_proj::i64_to_f64(large_i64) {
+            // Verify that the returned float is different from the direct cast
+            assert!(val > closest_f64);
+            assert!(val - closest_f64 < 2. * f64::EPSILON * closest_f64);
+        } else {
+            panic!("Expected ProjectedNumber::Next for large_i64");
+        }
+
+        // Test with very large negative value
+        let large_neg_i64 = -9_007_199_254_740_993; // -(2^53 + 1)
+        let closest_neg_f64 = -9_007_199_254_740_992.0;
+        assert_eq!(large_neg_i64 as f64, closest_neg_f64);
+        if let ProjectedNumber::Next(val) = num_proj::i64_to_f64(large_neg_i64) {
+            // Verify that the returned float is the closest representable f64
+            assert_eq!(val, closest_neg_f64);
+        } else {
+            panic!("Expected ProjectedNumber::Next for large_neg_i64");
+        }
+    }
+
+    #[test]
+    fn test_u64_to_f64() {
+        assert_eq!(num_proj::u64_to_f64(0), ProjectedNumber::Exact(0.0));
+        assert_eq!(num_proj::u64_to_f64(42), ProjectedNumber::Exact(42.0));
+
+        // Test the largest u64 value that can be exactly represented as f64 (2^53)
+        let max_exact = 9_007_199_254_740_992; // 2^53
+        assert_eq!(
+            num_proj::u64_to_f64(max_exact),
+            ProjectedNumber::Exact(max_exact as f64)
+        );
+
+        // Test values that cannot be exactly represented as f64 (integers above 2^53)
+        let large_u64 = 9_007_199_254_740_993; // 2^53 + 1
+        let closest_f64 = 9_007_199_254_740_992.0;
+        assert_eq!(large_u64 as f64, closest_f64);
+        if let ProjectedNumber::Next(val) = num_proj::u64_to_f64(large_u64) {
+            // Verify that the returned float is different from the direct cast
+            assert!(val > closest_f64);
+            assert!(val - closest_f64 < 2. * f64::EPSILON * closest_f64);
+        } else {
+            panic!("Expected ProjectedNumber::Next for large_u64");
+        }
+    }
+}

src/aggregation/bucket/histogram/date_histogram.rs

@@ -207,7 +207,7 @@ fn parse_offset_into_milliseconds(input: &str) -> Result<i64, AggregationError>
     }
 }
 
-fn parse_into_milliseconds(input: &str) -> Result<i64, AggregationError> {
+pub(crate) fn parse_into_milliseconds(input: &str) -> Result<i64, AggregationError> {
     let split_boundary = input
         .as_bytes()
         .iter()

src/aggregation/bucket/mod.rs

@@ -22,6 +22,7 @@
 //! - [Range](RangeAggregation)
 //! - [Terms](TermsAggregation)
 
+mod composite;
 mod filter;
 mod histogram;
 mod range;
@@ -31,6 +32,7 @@ mod term_missing_agg;
 use std::collections::HashMap;
 use std::fmt;
 
+pub use composite::*;
 pub use filter::*;
 pub use histogram::*;
 pub use range::*;

src/aggregation/intermediate_agg_result.rs

@@ -15,8 +15,9 @@ use serde::{Deserialize, Serialize};
 use super::agg_req::{Aggregation, AggregationVariants, Aggregations};
 use super::agg_result::{AggregationResult, BucketResult, MetricResult, RangeBucketEntry};
 use super::bucket::{
-    cut_off_buckets, get_agg_name_and_property, intermediate_histogram_buckets_to_final_buckets,
-    GetDocCount, Order, OrderTarget, RangeAggregation, TermsAggregation,
+    composite_intermediate_key_ordering, cut_off_buckets, get_agg_name_and_property,
+    intermediate_histogram_buckets_to_final_buckets, CompositeAggregation, GetDocCount,
+    MissingOrder, Order, OrderTarget, RangeAggregation, TermsAggregation,
 };
 use super::metric::{
     IntermediateAverage, IntermediateCount, IntermediateExtendedStats, IntermediateMax,
@@ -25,7 +26,7 @@ use super::metric::{
 use super::segment_agg_result::AggregationLimitsGuard;
 use super::{format_date, AggregationError, Key, SerializedKey};
 use crate::aggregation::agg_result::{
-    AggregationResults, BucketEntries, BucketEntry, FilterBucketResult,
+    AggregationResults, BucketEntries, BucketEntry, CompositeBucketEntry, FilterBucketResult,
 };
 use crate::aggregation::bucket::TermsAggregationInternal;
 use crate::aggregation::metric::CardinalityCollector;
@@ -280,6 +281,11 @@ pub(crate) fn empty_from_req(req: &Aggregation) -> IntermediateAggregationResult
             doc_count: 0,
             sub_aggregations: IntermediateAggregationResults::default(),
         }),
+        Composite(_) => {
+            IntermediateAggregationResult::Bucket(IntermediateBucketResult::Composite {
+                buckets: IntermediateCompositeBucketResult::default(),
+            })
+        }
     }
 }
 
@@ -473,6 +479,11 @@ pub enum IntermediateBucketResult {
         /// Sub-aggregation results
         sub_aggregations: IntermediateAggregationResults,
     },
+    /// Composite aggregation
+    Composite {
+        /// The composite buckets
+        buckets: IntermediateCompositeBucketResult,
+    },
 }
 
 impl IntermediateBucketResult {
@@ -568,6 +579,13 @@ impl IntermediateBucketResult {
                     sub_aggregations: final_sub_aggregations,
                 }))
             }
+            IntermediateBucketResult::Composite { buckets } => {
+                let composite_req = req
+                    .agg
+                    .as_composite()
+                    .expect("unexpected aggregation, expected composite aggregation");
+                buckets.into_final_result(composite_req, req.sub_aggregation(), limits)
+            }
         }
     }
 
@@ -634,6 +652,16 @@ impl IntermediateBucketResult {
                 *doc_count_left += doc_count_right;
                 sub_aggs_left.merge_fruits(sub_aggs_right)?;
             }
+            (
+                IntermediateBucketResult::Composite {
+                    buckets: composite_left,
+                },
+                IntermediateBucketResult::Composite {
+                    buckets: composite_right,
+                },
+            ) => {
+                composite_left.merge_fruits(composite_right)?;
+            }
             (IntermediateBucketResult::Range(_), _) => {
                 panic!("try merge on different types")
             }
@@ -646,6 +674,9 @@ impl IntermediateBucketResult {
             (IntermediateBucketResult::Filter { .. }, _) => {
                 panic!("try merge on different types")
             }
+            (IntermediateBucketResult::Composite { .. }, _) => {
+                panic!("try merge on different types")
+            }
         }
         Ok(())
     }
@@ -914,6 +945,176 @@ impl MergeFruits for IntermediateHistogramBucketEntry {
     }
 }
 
+/// Entry for the composite bucket.
+pub type IntermediateCompositeBucketEntry = IntermediateTermBucketEntry;
+
+/// The fully typed key for composite aggregation
+#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
+pub enum CompositeIntermediateKey {
+    /// Bool key
+    Bool(bool),
+    /// String key
+    Str(String),
+    /// Float key
+    F64(f64),
+    /// Signed integer key
+    I64(i64),
+    /// Unsigned integer key
+    U64(u64),
+    /// DateTime key, nanoseconds since epoch
+    DateTime(i64),
+    /// IP Address key
+    IpAddr(Ipv6Addr),
+    /// Missing value key
+    Null,
+}
+
+impl Eq for CompositeIntermediateKey {}
+
+impl std::hash::Hash for CompositeIntermediateKey {
+    fn hash<H: std::hash::Hasher>(&self, state: &mut H) {
+        core::mem::discriminant(self).hash(state);
+        match self {
+            CompositeIntermediateKey::Bool(val) => val.hash(state),
+            CompositeIntermediateKey::Str(text) => text.hash(state),
+            CompositeIntermediateKey::F64(val) => val.to_bits().hash(state),
+            CompositeIntermediateKey::U64(val) => val.hash(state),
+            CompositeIntermediateKey::I64(val) => val.hash(state),
+            CompositeIntermediateKey::DateTime(val) => val.hash(state),
+            CompositeIntermediateKey::IpAddr(val) => val.hash(state),
+            CompositeIntermediateKey::Null => {}
+        }
+    }
+}
+
+/// Composite aggregation page.
+#[derive(Default, Clone, Debug, PartialEq, Serialize, Deserialize)]
+pub struct IntermediateCompositeBucketResult {
+    pub(crate) entries: FxHashMap<Vec<CompositeIntermediateKey>, IntermediateCompositeBucketEntry>,
+    pub(crate) target_size: u32,
+    pub(crate) orders: Vec<(Order, MissingOrder)>,
+}
+
+impl IntermediateCompositeBucketResult {
+    pub(crate) fn into_final_result(
+        self,
+        req: &CompositeAggregation,
+        sub_aggregation_req: &Aggregations,
+        limits: &mut AggregationLimitsGuard,
+    ) -> crate::Result<BucketResult> {
+        let trimmed_entry_vec =
+            trim_composite_buckets(self.entries, &self.orders, self.target_size)?;
+        let after_key = if trimmed_entry_vec.len() == req.size as usize {
+            trimmed_entry_vec
+                .last()
+                .map(|bucket| {
+                    let (intermediate_key, _entry) = bucket;
+                    intermediate_key
+                        .iter()
+                        .enumerate()
+                        .map(|(idx, intermediate_key)| {
+                            let source = &req.sources[idx];
+                            (source.name().to_string(), intermediate_key.clone().into())
+                        })
+                        .collect()
+                })
+                .unwrap()
+        } else {
+            FxHashMap::default()
+        };
+
+        let buckets = trimmed_entry_vec
+            .into_iter()
+            .map(|(intermediate_key, entry)| {
+                let key = intermediate_key
+                    .into_iter()
+                    .enumerate()
+                    .map(|(idx, intermediate_key)| {
+                        let source = &req.sources[idx];
+                        (source.name().to_string(), intermediate_key.into())
+                    })
+                    .collect();
+                Ok(CompositeBucketEntry {
+                    key,
+                    doc_count: entry.doc_count as u64,
+                    sub_aggregation: entry
+                        .sub_aggregation
+                        .into_final_result_internal(sub_aggregation_req, limits)?,
+                })
+            })
+            .collect::<crate::Result<Vec<_>>>()?;
+
+        Ok(BucketResult::Composite { after_key, buckets })
+    }
+
+    fn merge_fruits(&mut self, other: IntermediateCompositeBucketResult) -> crate::Result<()> {
+        merge_maps(&mut self.entries, other.entries)?;
+        if self.entries.len() as u32 > 2 * self.target_size {
+            self.trim()?;
+        }
+        Ok(())
+    }
+
+    /// Trim the composite buckets to the target size, according to the ordering.
+    pub(crate) fn trim(&mut self) -> crate::Result<()> {
+        if self.entries.len() as u32 <= self.target_size {
+            return Ok(());
+        }
+
+        let sorted_entries = trim_composite_buckets(
+            std::mem::take(&mut self.entries),
+            &self.orders,
+            self.target_size,
+        )?;
+
+        self.entries = sorted_entries.into_iter().collect();
+        Ok(())
+    }
+}
+
+fn trim_composite_buckets(
+    entries: FxHashMap<Vec<CompositeIntermediateKey>, IntermediateCompositeBucketEntry>,
+    orders: &[(Order, MissingOrder)],
+    target_size: u32,
+) -> crate::Result<
+    Vec<(
+        Vec<CompositeIntermediateKey>,
+        IntermediateCompositeBucketEntry,
+    )>,
+> {
+    let mut entries: Vec<_> = entries.into_iter().collect();
+    let mut sort_error: Option<TantivyError> = None;
+    entries.sort_by(|(left_key, _), (right_key, _)| {
+        if sort_error.is_some() {
+            return Ordering::Equal;
+        }
+
+        for idx in 0..orders.len() {
+            match composite_intermediate_key_ordering(
+                &left_key[idx],
+                &right_key[idx],
+                orders[idx].0,
+                orders[idx].1,
+            ) {
+                Ok(ordering) if ordering != Ordering::Equal => return ordering,
+                Ok(_) => continue,
+                Err(err) => {
+                    sort_error = Some(err);
+                    break;
+                }
+            }
+        }
+        Ordering::Equal
+    });
+
+    if let Some(err) = sort_error {
+        return Err(err);
+    }
+
+    entries.truncate(target_size as usize);
+    Ok(entries)
+}
+
 #[cfg(test)]
 mod tests {
     use std::collections::HashMap;

src/directory/composite_file.rs

@@ -167,6 +167,7 @@ impl CompositeFile {
             .map(|byte_range| self.data.slice(byte_range.clone()))
     }
 
+    /// Returns the space usage per field in this composite file.
     pub fn space_usage(&self, schema: &Schema) -> PerFieldSpaceUsage {
         let mut fields = Vec::new();
         for (&field_addr, byte_range) in &self.offsets_index {

src/indexer/segment_updater.rs

@@ -403,7 +403,8 @@ impl SegmentUpdater {
             // from the different drives.
             //
             // Segment 1 from disk 1, Segment 1 from disk 2, etc.
-            committed_segment_metas.sort_by_key(|segment_meta| std::cmp::Reverse(segment_meta.max_doc()));
+            committed_segment_metas
+                .sort_by_key(|segment_meta| std::cmp::Reverse(segment_meta.max_doc()));
             let index_meta = IndexMeta {
                 index_settings: index.settings().clone(),
                 segments: committed_segment_metas,
@@ -648,9 +649,6 @@ impl SegmentUpdater {
                                     merge_operation.segment_ids(),
                                     advance_deletes_err
                                 );
-                                assert!(!cfg!(test), "Merge failed.");
-
-                                // ... cancel merge
                                 // `merge_operations` are tracked. As it is dropped, the
                                 // the segment_ids will be available again for merge.
                                 return Err(advance_deletes_err);
@@ -705,12 +703,12 @@ mod tests {
     use crate::collector::TopDocs;
     use crate::directory::RamDirectory;
     use crate::fastfield::AliveBitSet;
+    use crate::index::{SegmentId, SegmentMetaInventory};
     use crate::indexer::merge_policy::tests::MergeWheneverPossible;
     use crate::indexer::merger::IndexMerger;
     use crate::indexer::segment_updater::merge_filtered_segments;
     use crate::query::QueryParser;
     use crate::schema::*;
-    use crate::index::{SegmentId, SegmentMetaInventory};
     use crate::{Directory, DocAddress, Index, Segment};
 
     #[test]
@@ -718,7 +716,7 @@ mod tests {
         // Regression test: -(max_doc as i32) overflows for max_doc >= 2^31.
         // Using std::cmp::Reverse avoids this.
         let inventory = SegmentMetaInventory::default();
-        let mut metas = vec![
+        let mut metas = [
             inventory.new_segment_meta(SegmentId::generate_random(), 100),
             inventory.new_segment_meta(SegmentId::generate_random(), (1u32 << 31) - 1),
             inventory.new_segment_meta(SegmentId::generate_random(), 50_000),

src/termdict/fst_termdict/term_info_store.rs

@@ -48,8 +48,7 @@ impl BinarySerializable for TermInfoBlockMeta {
 }
 
 impl FixedSize for TermInfoBlockMeta {
-    const SIZE_IN_BYTES: usize =
-        u64::SIZE_IN_BYTES + TermInfo::SIZE_IN_BYTES + 3 * u8::SIZE_IN_BYTES;
+    const SIZE_IN_BYTES: usize = u64::SIZE_IN_BYTES + TermInfo::SIZE_IN_BYTES + 3;
 }
 
 impl TermInfoBlockMeta {

sstable/src/lib.rs

@@ -51,7 +51,7 @@ mod sstable_index_v3;
 pub use sstable_index_v3::{BlockAddr, SSTableIndex, SSTableIndexBuilder, SSTableIndexV3};
 mod sstable_index_v2;
 pub(crate) mod vint;
-pub use dictionary::Dictionary;
+pub use dictionary::{Dictionary, TermOrdHit};
 pub use streamer::{Streamer, StreamerBuilder};
 
 mod block_reader;

sstable/src/sstable_index_v3.rs

@@ -553,7 +553,7 @@ impl FixedSize for BlockAddrBlockMetadata {
     const SIZE_IN_BYTES: usize = u64::SIZE_IN_BYTES
         + BlockStartAddr::SIZE_IN_BYTES
         + 2 * u32::SIZE_IN_BYTES
-        + 2 * u8::SIZE_IN_BYTES
+        + 2
         + u16::SIZE_IN_BYTES;
 }