fix: adapt composite aggregation for Quickwit compatibility

Fixes on top of PR #2714 ("Add composite aggregation") to make it work
with Quickwit's current codebase and Postcard serialization:

- Rewrite SegmentCompositeCollector to match current
  SegmentAggregationCollector trait signatures (collect,
  add_intermediate_aggregation_result, prepare_max_bucket)
- Remove Clone derive from CompositeBucketCollector (incompatible
  with dyn SegmentAggregationCollector)
- Add custom serde for FxHashMap entries in
  IntermediateCompositeBucketResult (Postcard requires known sequence
  length)
- Rewrite AfterKey Serialize/Deserialize to output raw values instead
  of internal "type:value" format, matching Elasticsearch wire format
- Remove unused imports and tracing::warn calls

Made-with: Cursor

benches/agg_bench.rs

@@ -1,5 +1,6 @@
 use binggan::plugins::PeakMemAllocPlugin;
 use binggan::{black_box, InputGroup, PeakMemAlloc, INSTRUMENTED_SYSTEM};
+use common::DateTime;
 use rand::distr::weighted::WeightedIndex;
 use rand::rngs::StdRng;
 use rand::seq::IndexedRandom;
@@ -70,6 +71,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);
 
@@ -313,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();
@@ -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())?
@@ -593,6 +670,7 @@ fn get_test_index_bench(cardinality: Cardinality) -> tantivy::Result<Index> {
                 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 {
@@ -430,6 +448,26 @@ impl CompactSpaceDecompressor {
         Ok(decompressor)
     }
 
+    /// Finds the next compact space value for a given u128 value
+    pub fn u128_to_next_compact(&self, value: u128) -> CompactHit {
+        // Try to convert to compact space
+        match self.u128_to_compact(value) {
+            // Value is in compact space, return its compact representation
+            Ok(compact) => CompactHit::Exact(compact),
+            // Value is not in compact space
+            Err(pos) => {
+                if pos >= self.params.compact_space.ranges_mapping.len() {
+                    // Value is beyond all ranges, no next value exists
+                    CompactHit::AfterLast
+                } else {
+                    // Get the next range and return its start compact value
+                    let next_range = &self.params.compact_space.ranges_mapping[pos];
+                    CompactHit::Next(next_range.compact_start)
+                }
+            }
+        }
+    }
+
     /// Converting to compact space for the decompressor is more complex, since we may get values
     /// which are outside the compact space. e.g. if we map
     /// 1000 => 5
@@ -823,6 +861,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/accessor_helpers.rs

@@ -95,11 +95,21 @@ pub(crate) fn get_all_ff_reader_or_empty(
     allowed_column_types: Option<&[ColumnType]>,
     fallback_type: ColumnType,
 ) -> crate::Result<Vec<(columnar::Column<u64>, ColumnType)>> {
-    let ff_fields = reader.fast_fields();
-    let mut ff_field_with_type =
-        ff_fields.u64_lenient_for_type_all(allowed_column_types, field_name)?;
+    let mut ff_field_with_type = get_all_ff_readers(reader, field_name, allowed_column_types)?;
     if ff_field_with_type.is_empty() {
         ff_field_with_type.push((Column::build_empty_column(reader.num_docs()), fallback_type));
     }
     Ok(ff_field_with_type)
 }
+
+/// Get all fast field reader.
+pub(crate) fn get_all_ff_readers(
+    reader: &SegmentReader,
+    field_name: &str,
+    allowed_column_types: Option<&[ColumnType]>,
+) -> crate::Result<Vec<(columnar::Column<u64>, ColumnType)>> {
+    let ff_fields = reader.fast_fields();
+    let ff_field_with_type =
+        ff_fields.u64_lenient_for_type_all(allowed_column_types, field_name)?;
+    Ok(ff_field_with_type)
+}

src/aggregation/agg_data.rs

@@ -9,11 +9,12 @@ use crate::aggregation::accessor_helpers::{
     get_numeric_or_date_column_types,
 };
 use crate::aggregation::agg_req::{Aggregation, AggregationVariants, Aggregations};
+pub use crate::aggregation::bucket::{CompositeAggReqData, CompositeSourceAccessors};
 use crate::aggregation::bucket::{
-    build_segment_filter_collector, build_segment_range_collector, FilterAggReqData,
-    HistogramAggReqData, HistogramBounds, IncludeExcludeParam, MissingTermAggReqData,
-    RangeAggReqData, SegmentHistogramCollector, TermMissingAgg, TermsAggReqData, TermsAggregation,
-    TermsAggregationInternal,
+    build_segment_filter_collector, build_segment_range_collector, CompositeAggregation,
+    FilterAggReqData, HistogramAggReqData, HistogramBounds, IncludeExcludeParam,
+    MissingTermAggReqData, RangeAggReqData, SegmentCompositeCollector, SegmentHistogramCollector,
+    TermMissingAgg, TermsAggReqData, TermsAggregation, TermsAggregationInternal,
 };
 use crate::aggregation::metric::{
     build_segment_stats_collector, AverageAggregation, CardinalityAggReqData,
@@ -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 ----------
 
@@ -130,8 +143,14 @@ impl AggregationsSegmentCtx {
             .as_deref_mut()
             .expect("histogram_req_data slot is empty (taken)")
     }
+    #[inline]
+    pub(crate) fn get_composite_req_data_mut(&mut self, idx: usize) -> &mut CompositeAggReqData {
+        self.per_request.composite_req_data[idx]
+            .as_deref_mut()
+            .expect("composite_req_data slot is empty (taken)")
+    }
 
-    // ---------- take / put (terms, histogram, range) ----------
+    // ---------- take / put (terms, histogram, range, composite) ----------
 
     /// Move out the boxed Histogram request at `idx`, leaving `None`.
     #[inline]
@@ -181,6 +200,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
@@ -200,6 +238,8 @@ pub struct PerRequestAggSegCtx {
     pub range_req_data: Vec<Option<Box<RangeAggReqData>>>,
     /// FilterAggReqData contains the request data for a filter aggregation.
     pub filter_req_data: Vec<Option<Box<FilterAggReqData>>>,
+    /// CompositeAggReqData contains the request data for a composite aggregation.
+    pub composite_req_data: Vec<Option<Box<CompositeAggReqData>>>,
     /// Shared by avg, min, max, sum, stats, extended_stats, count
     pub stats_metric_req_data: Vec<MetricAggReqData>,
     /// CardinalityAggReqData contains the request data for a cardinality aggregation.
@@ -255,6 +295,11 @@ impl PerRequestAggSegCtx {
                 .iter()
                 .map(|t| t.get_memory_consumption())
                 .sum::<usize>()
+            + self
+                .composite_req_data
+                .iter()
+                .map(|t| t.as_ref().unwrap().get_memory_consumption())
+                .sum::<usize>()
             + self.agg_tree.len() * std::mem::size_of::<AggRefNode>()
     }
 
@@ -291,6 +336,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 +467,9 @@ 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(SegmentCompositeCollector::from_req_and_validate(
+            req, node,
+        )?)),
     }
 }
 
@@ -447,6 +500,7 @@ pub enum AggKind {
     DateHistogram,
     Range,
     Filter,
+    Composite,
 }
 
 impl AggKind {
@@ -462,6 +516,7 @@ impl AggKind {
             AggKind::DateHistogram => "DateHistogram",
             AggKind::Range => "Range",
             AggKind::Filter => "Filter",
+            AggKind::Composite => "Composite",
         }
     }
 }
@@ -740,6 +795,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,
+        )?]),
     }
 }
 
@@ -935,6 +998,35 @@ fn build_terms_or_cardinality_nodes(
     Ok(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,
+    })
+}
+
 /// Builds a single BitSet of allowed term ordinals for a string dictionary column according to
 /// include/exclude parameters.
 fn build_allowed_term_ids_for_str(

src/aggregation/agg_req.rs

@@ -40,6 +40,7 @@ use super::metric::{
     MaxAggregation, MinAggregation, PercentilesAggregationReq, StatsAggregation, SumAggregation,
     TopHitsAggregationReq,
 };
+use crate::aggregation::bucket::CompositeAggregation;
 
 /// The top-level aggregation request structure, which contains [`Aggregation`] and their user
 /// defined names. It is also used in buckets aggregations to define sub-aggregations.
@@ -134,6 +135,9 @@ pub enum AggregationVariants {
     /// Filter documents into a single bucket.
     #[serde(rename = "filter")]
     Filter(FilterAggregation),
+    /// Put data into multi level paginated buckets.
+    #[serde(rename = "composite")]
+    Composite(CompositeAggregation),
 
     // Metric aggregation types
     /// Computes the average of the extracted values.
@@ -180,6 +184,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_map| source_map.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 +223,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

@@ -13,6 +13,8 @@ use super::metric::{
     ExtendedStats, PercentilesMetricResult, SingleMetricResult, Stats, TopHitsMetricResult,
 };
 use super::{AggregationError, Key};
+use crate::aggregation::bucket::AfterKey;
+use crate::aggregation::intermediate_agg_result::CompositeIntermediateKey;
 use crate::TantivyError;
 
 #[derive(Clone, Default, Debug, PartialEq, Serialize, Deserialize)]
@@ -158,6 +160,16 @@ pub enum BucketResult {
     },
     /// This is the filter result - a single bucket with sub-aggregations
     Filter(FilterBucketResult),
+    /// This is the composite aggregation result
+    Composite {
+        /// The buckets
+        ///
+        /// See [`CompositeAggregation`](super::bucket::CompositeAggregation)
+        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 +191,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 +352,130 @@ pub struct FilterBucketResult {
     #[serde(flatten)]
     pub sub_aggregations: AggregationResults,
 }
+
+/// The JSON mappable key to identify a composite bucket.
+///
+/// This is similar to `Key`, but composite keys can also be boolean and null.
+///
+/// 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,
+            (
+                Self::Bool(_)
+                | Self::Str(_)
+                | Self::F64(_)
+                | Self::I64(_)
+                | Self::U64(_)
+                | Self::Null,
+                _,
+            ) => false,
+        }
+    }
+}
+impl From<CompositeIntermediateKey> for CompositeKey {
+    fn from(value: CompositeIntermediateKey) -> Self {
+        match value {
+            CompositeIntermediateKey::Str(s) => Self::Str(s),
+            CompositeIntermediateKey::IpAddr(s) => {
+                // Prefer to use the IPv4 representation if possible
+                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), // Convert ns to ms
+            CompositeIntermediateKey::Null => Self::Null,
+        }
+    }
+}
+
+/// This is the default entry for a bucket, which contains a composite key, count, and optionally
+/// sub-aggregations.
+///   ...
+///     "my_composite": {
+///       "buckets": [
+///         {
+///           "key": {
+///             "date": 1494201600000,
+///             "product": "rocky"
+///           },
+///           "doc_count": 5
+///         },
+///         {
+///           "key": {
+///             "date": 1494201600000,
+///             "product": "balboa"
+///           },
+///           "doc_count": 2
+///         },
+///         {
+///           "key": {
+///             "date": 1494201700000,
+///             "product": "john"
+///           },
+///           "doc_count": 3
+///         }
+///       ]
+///    }
+///    ...
+/// }
+/// ```
+#[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,515 @@
+use std::fmt::Debug;
+use std::net::Ipv6Addr;
+
+use columnar::column_values::{CompactHit, CompactSpaceU64Accessor};
+use columnar::{Column, ColumnType, MonotonicallyMappableToU64, StrColumn, TermOrdHit};
+
+use crate::aggregation::accessor_helpers::{get_all_ff_readers, 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 =
+                    get_all_ff_readers(reader, &source.field, Some(&allowed_column_types))?;
+
+                // Sort columns by their pagination order and determine which to skip
+                columns_and_types.sort_by_key(|(_, col_type)| 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)> = get_all_ff_readers(
+                    reader,
+                    &source.field,
+                    Some(get_numeric_or_date_column_types()),
+                )?;
+                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 =
+                    get_all_ff_readers(reader, &source.field, Some(&[ColumnType::DateTime]))?;
+                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<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 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<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,140 @@
+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 {}: {}",
+            timestamp_ns,
+            e.to_string()
+        ))
+    })
+}
+
+/// 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 {}: {}",
+            timestamp_ns,
+            e.to_string()
+        ))
+    })
+}
+
+/// 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,595 @@
+use std::fmt::Debug;
+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::intermediate_agg_result::{
+    CompositeIntermediateKey, IntermediateAggregationResult, IntermediateAggregationResults,
+    IntermediateBucketResult, IntermediateCompositeBucketEntry, IntermediateCompositeBucketResult,
+};
+use crate::aggregation::segment_agg_result::SegmentAggregationCollector;
+use crate::aggregation::BucketId;
+use crate::TantivyError;
+
+#[derive(Debug)]
+struct CompositeBucketCollector {
+    count: u32,
+}
+
+impl CompositeBucketCollector {
+    fn new() -> Self {
+        CompositeBucketCollector { count: 0 }
+    }
+    #[inline]
+    fn collect(&mut self) {
+        self.count += 1;
+    }
+}
+
+/// The value is represented as a tuple of:
+/// - the column index or missing value sentinel
+///   - if the value is present, store the accessor index + 1
+///   - if the value is missing, store 0 (for missing first) or u8::MAX (for missing last)
+/// - the fast field value u64 representation
+///   - 0 if the field is missing
+///   - regular u64 repr if the ordering is ascending
+///   - bitwise NOT of the u64 repr if the ordering is descending
+#[derive(Clone, Copy, Debug, PartialEq, Eq, PartialOrd, Ord, Default, Hash)]
+struct InternalValueRepr(u8, u64);
+
+impl InternalValueRepr {
+    #[inline]
+    fn new_term(raw: u64, accessor_idx: u8, order: Order) -> Self {
+        match order {
+            Order::Asc => InternalValueRepr(accessor_idx + 1, raw),
+            Order::Desc => InternalValueRepr(accessor_idx + 1, !raw),
+        }
+    }
+    /// For histogram, the source column does not matter
+    #[inline]
+    fn new_histogram(raw: u64, order: Order) -> Self {
+        match order {
+            Order::Asc => InternalValueRepr(1, raw),
+            Order::Desc => InternalValueRepr(1, !raw),
+        }
+    }
+    #[inline]
+    fn new_missing(order: Order, missing_order: MissingOrder) -> Self {
+        let column_idx = match (missing_order, order) {
+            (MissingOrder::First, _) => 0,
+            (MissingOrder::Last, _) => u8::MAX,
+            (MissingOrder::Default, Order::Asc) => 0,
+            (MissingOrder::Default, Order::Desc) => u8::MAX,
+        };
+        InternalValueRepr(column_idx, 0)
+    }
+    #[inline]
+    fn decode(self, order: Order) -> Option<(u8, u64)> {
+        if self.0 == u8::MAX || self.0 == 0 {
+            return None;
+        }
+        match order {
+            Order::Asc => Some((self.0 - 1, self.1)),
+            Order::Desc => Some((self.0 - 1, !self.1)),
+        }
+    }
+}
+
+/// 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 {
+    buckets: DynArrayHeapMap<InternalValueRepr, CompositeBucketCollector>,
+    accessor_idx: 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.into_intermediate_bucket_result(agg_data)?;
+        results.push(
+            name,
+            IntermediateAggregationResult::Bucket(IntermediateBucketResult::Composite { buckets }),
+        )?;
+
+        Ok(())
+    }
+
+    #[inline]
+    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 sub_level_values = SmallVec::new();
+            recursive_key_visitor(
+                *doc,
+                agg_data,
+                &composite_agg_data,
+                0,
+                &mut sub_level_values,
+                &mut self.buckets,
+                true,
+            )?;
+        }
+        agg_data.put_back_composite_req_data(self.accessor_idx, composite_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 prepare_max_bucket(
+        &mut self,
+        _max_bucket: BucketId,
+        _agg_data: &AggregationsSegmentCtx,
+    ) -> crate::Result<()> {
+        Ok(())
+    }
+
+    fn flush(&mut self, _agg_data: &mut AggregationsSegmentCtx) -> crate::Result<()> {
+        Ok(())
+    }
+}
+
+impl SegmentCompositeCollector {
+    fn get_memory_consumption(&self) -> u64 {
+        self.buckets.memory_consumption()
+    }
+
+    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)?;
+
+        if !node.children.is_empty() {
+            let _sub_aggregation = build_segment_agg_collectors(req_data, &node.children)?;
+        }
+
+        let composite_req_data = req_data.get_composite_req_data(node.idx_in_req_data);
+        Ok(SegmentCompositeCollector {
+            buckets: DynArrayHeapMap::try_new(composite_req_data.req.sources.len())?,
+            accessor_idx: node.idx_in_req_data,
+        })
+    }
+
+    #[inline]
+    pub(crate) fn into_intermediate_bucket_result(
+        &mut self,
+        agg_data: &AggregationsSegmentCtx,
+    ) -> crate::Result<IntermediateCompositeBucketResult> {
+        let mut dict: FxHashMap<Vec<CompositeIntermediateKey>, IntermediateCompositeBucketEntry> =
+            Default::default();
+        dict.reserve(self.buckets.size());
+        let composite_data = agg_data.get_composite_req_data(self.accessor_idx);
+        let buckets = std::mem::replace(
+            &mut self.buckets,
+            DynArrayHeapMap::try_new(composite_data.req.sources.len())
+                .expect("already validated source count"),
+        );
+        for (key_internal_repr, agg) in buckets.into_iter() {
+            let key = resolve_key(&key_internal_repr, composite_data)?;
+            let sub_aggregation_res = IntermediateAggregationResults::default();
+
+            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(),
+        ));
+    }
+    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.len() > MAX_DYN_ARRAY_SIZE {
+            return Err(TantivyError::InvalidArgument(format!(
+                "composite aggregation source supports maximum {MAX_DYN_ARRAY_SIZE} 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(
+    agg_data: &mut AggregationsSegmentCtx,
+    composite_agg_data: &CompositeAggReqData,
+    buckets: &mut DynArrayHeapMap<InternalValueRepr, CompositeBucketCollector>,
+    key: &[InternalValueRepr],
+) -> crate::Result<()> {
+    if (buckets.size() as u32) < composite_agg_data.req.size {
+        buckets
+            .get_or_insert_with(key, CompositeBucketCollector::new)
+            .collect();
+        return Ok(());
+    }
+
+    if let Some(entry) = buckets.get_mut(key) {
+        entry.collect();
+        return Ok(());
+    }
+
+    if let Some(highest_key) = buckets.peek_highest() {
+        if key < highest_key {
+            buckets.evict_highest();
+            buckets
+                .get_or_insert_with(key, CompositeBucketCollector::new)
+                .collect();
+        }
+    }
+
+    let _ = agg_data;
+    Ok(())
+}
+
+/// 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
+        .into_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)
+}
+
+/// Depth-first walk of the accessors to build the composite key combinations
+/// and update the buckets.
+fn recursive_key_visitor(
+    doc_id: crate::DocId,
+    agg_data: &mut AggregationsSegmentCtx,
+    composite_agg_data: &CompositeAggReqData,
+    source_idx_for_recursion: usize,
+    sub_level_values: &mut SmallVec<[InternalValueRepr; MAX_DYN_ARRAY_SIZE]>,
+    buckets: &mut DynArrayHeapMap<InternalValueRepr, CompositeBucketCollector>,
+    is_on_after_key: bool,
+) -> crate::Result<()> {
+    if source_idx_for_recursion == composite_agg_data.req.sources.len() {
+        if !is_on_after_key {
+            collect_bucket_with_limit(
+                agg_data,
+                composite_agg_data,
+                buckets,
+                sub_level_values,
+            )?;
+        }
+        return Ok(());
+    }
+
+    let current_level_accessors = &composite_agg_data.composite_accessors[source_idx_for_recursion];
+    let current_level_source = &composite_agg_data.req.sources[source_idx_for_recursion];
+    let mut missing = true;
+    for (accessor_idx, accessor) in current_level_accessors.accessors.iter().enumerate() {
+        let values = accessor.column.values_for_doc(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;
+                        }
+                    }
+                    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);
+                    recursive_key_visitor(
+                        doc_id,
+                        agg_data,
+                        composite_agg_data,
+                        source_idx_for_recursion + 1,
+                        sub_level_values,
+                        buckets,
+                        is_on_after_key && still_on_after_key,
+                    )?;
+                    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;
+                        }
+                    }
+                    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);
+                    recursive_key_visitor(
+                        doc_id,
+                        agg_data,
+                        composite_agg_data,
+                        source_idx_for_recursion + 1,
+                        sub_level_values,
+                        buckets,
+                        is_on_after_key && still_on_after_key,
+                    )?;
+                    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;
+                        }
+                    }
+                    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);
+                    recursive_key_visitor(
+                        doc_id,
+                        agg_data,
+                        composite_agg_data,
+                        source_idx_for_recursion + 1,
+                        sub_level_values,
+                        buckets,
+                        is_on_after_key && still_on_after_key,
+                    )?;
+                    sub_level_values.pop();
+                }
+            };
+        }
+    }
+    if missing && current_level_source.missing_bucket() {
+        if is_on_after_key && current_level_accessors.skip_missing {
+            return Ok(());
+        }
+        sub_level_values.push(InternalValueRepr::new_missing(
+            current_level_source.order(),
+            current_level_source.missing_order(),
+        ));
+        recursive_key_visitor(
+            doc_id,
+            agg_data,
+            composite_agg_data,
+            source_idx_for_recursion + 1,
+            sub_level_values,
+            buckets,
+            is_on_after_key && current_level_accessors.is_after_key_explicit_missing,
+        )?;
+        sub_level_values.pop();
+    }
+    Ok(())
+}

src/aggregation/bucket/composite/map.rs

@@ -0,0 +1,364 @@
+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)),
+        )
+    }
+
+    fn values_mut<'a>(&'a mut self) -> Box<dyn Iterator<Item = &'a mut V> + 'a> {
+        Box::new(self.buckets.values_mut())
+    }
+}
+
+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(),
+        }
+    }
+
+    /// Returns an iterator over mutable references to the values in the map.
+    pub(super) fn values_mut(&mut self) -> impl Iterator<Item = &mut V> {
+        match &mut self.0 {
+            DynArrayHeapMapInner::Dim1(map) => map.values_mut(),
+            DynArrayHeapMapInner::Dim2(map) => map.values_mut(),
+            DynArrayHeapMapInner::Dim3(map) => map.values_mut(),
+            DynArrayHeapMapInner::Dim4(map) => map.values_mut(),
+            DynArrayHeapMapInner::Dim5(map) => map.values_mut(),
+            DynArrayHeapMapInner::Dim6(map) => map.values_mut(),
+            DynArrayHeapMapInner::Dim7(map) => map.values_mut(),
+            DynArrayHeapMapInner::Dim8(map) => map.values_mut(),
+            DynArrayHeapMapInner::Dim9(map) => map.values_mut(),
+            DynArrayHeapMapInner::Dim10(map) => map.values_mut(),
+            DynArrayHeapMapInner::Dim11(map) => map.values_mut(),
+            DynArrayHeapMapInner::Dim12(map) => map.values_mut(),
+            DynArrayHeapMapInner::Dim13(map) => map.values_mut(),
+            DynArrayHeapMapInner::Dim14(map) => map.values_mut(),
+            DynArrayHeapMapInner::Dim15(map) => map.values_mut(),
+            DynArrayHeapMapInner::Dim16(map) => map.values_mut(),
+        }
+    }
+}
+
+#[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[..]));
+
+        // mutable iterator
+        {
+            let mut mut_iter = map.values_mut();
+            let v = mut_iter.next().unwrap();
+            assert_eq!(*v, "a");
+            *v = "c";
+            assert_eq!(mut_iter.next(), None);
+        }
+
+        // into_iter
+        let mut iter = map.into_iter();
+        let (k, v) = iter.next().unwrap();
+        assert_eq!(k.as_slice(), &key1);
+        assert_eq!(v, "c");
+        assert_eq!(iter.next(), None);
+    }
+}

src/aggregation/bucket/composite/numeric_types.rs

@@ -0,0 +1,460 @@
+/// This modules 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 modules 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) {
+            return ProjectedNumber::Next(i64::MIN);
+        } else if value >= (i64::MAX as f64) {
+            return 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

@@ -25,9 +25,12 @@ 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::{
+    composite_intermediate_key_ordering, CompositeAggregation, MissingOrder,
+    TermsAggregationInternal,
 };
-use crate::aggregation::bucket::TermsAggregationInternal;
 use crate::aggregation::metric::CardinalityCollector;
 use crate::TantivyError;
 
@@ -246,6 +249,11 @@ pub(crate) fn empty_from_req(req: &Aggregation) -> IntermediateAggregationResult
                 is_date_agg: true,
             })
         }
+        Composite(_) => {
+            IntermediateAggregationResult::Bucket(IntermediateBucketResult::Composite {
+                buckets: Default::default(),
+            })
+        }
         Average(_) => IntermediateAggregationResult::Metric(IntermediateMetricResult::Average(
             IntermediateAverage::default(),
         )),
@@ -473,6 +481,11 @@ pub enum IntermediateBucketResult {
         /// Sub-aggregation results
         sub_aggregations: IntermediateAggregationResults,
     },
+    /// Composite aggregation
+    Composite {
+        /// The composite buckets
+        buckets: IntermediateCompositeBucketResult,
+    },
 }
 
 impl IntermediateBucketResult {
@@ -568,6 +581,13 @@ impl IntermediateBucketResult {
                     sub_aggregations: final_sub_aggregations,
                 }))
             }
+            IntermediateBucketResult::Composite { buckets } => buckets.into_final_result(
+                req.agg
+                    .as_composite()
+                    .expect("unexpected aggregation, expected composite aggregation"),
+                req.sub_aggregation(),
+                limits,
+            ),
         }
     }
 
@@ -634,6 +654,16 @@ impl IntermediateBucketResult {
                 *doc_count_left += doc_count_right;
                 sub_aggs_left.merge_fruits(sub_aggs_right)?;
             }
+            (
+                IntermediateBucketResult::Composite {
+                    buckets: buckets_left,
+                },
+                IntermediateBucketResult::Composite {
+                    buckets: buckets_right,
+                },
+            ) => {
+                buckets_left.merge_fruits(buckets_right)?;
+            }
             (IntermediateBucketResult::Range(_), _) => {
                 panic!("try merge on different types")
             }
@@ -646,6 +676,9 @@ impl IntermediateBucketResult {
             (IntermediateBucketResult::Filter { .. }, _) => {
                 panic!("try merge on different types")
             }
+            (IntermediateBucketResult::Composite { .. }, _) => {
+                panic!("try merge on different types")
+            }
         }
         Ok(())
     }
@@ -889,6 +922,212 @@ pub struct IntermediateTermBucketEntry {
     pub sub_aggregation: IntermediateAggregationResults,
 }
 
+/// 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 {
+    #[serde(
+        serialize_with = "serialize_composite_entries",
+        deserialize_with = "deserialize_composite_entries"
+    )]
+    pub(crate) entries: FxHashMap<Vec<CompositeIntermediateKey>, IntermediateCompositeBucketEntry>,
+    pub(crate) target_size: u32,
+    pub(crate) orders: Vec<(Order, MissingOrder)>,
+}
+
+fn serialize_composite_entries<S>(
+    entries: &FxHashMap<Vec<CompositeIntermediateKey>, IntermediateCompositeBucketEntry>,
+    serializer: S,
+) -> Result<S::Ok, S::Error>
+where
+    S: serde::Serializer,
+{
+    use serde::ser::SerializeSeq;
+    let mut seq = serializer.serialize_seq(Some(entries.len()))?;
+    for (k, v) in entries {
+        seq.serialize_element(&(k, v))?;
+    }
+    seq.end()
+}
+
+fn deserialize_composite_entries<'de, D>(
+    deserializer: D,
+) -> Result<FxHashMap<Vec<CompositeIntermediateKey>, IntermediateCompositeBucketEntry>, D::Error>
+where
+    D: serde::Deserializer<'de>,
+{
+    let vec: Vec<(Vec<CompositeIntermediateKey>, IntermediateCompositeBucketEntry)> =
+        serde::Deserialize::deserialize(deserializer)?;
+    Ok(vec.into_iter().collect())
+}
+
+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 {
+            // 2x factor used to avoid trimming too often (expensive operation)
+            // an optimal threshold could probably be figured out
+            self.trim()?;
+        }
+        Ok(())
+    }
+
+    /// Trim the composite buckets to the target size, according to the ordering.
+    ///
+    /// Returns an error if the ordering comparison fails.
+    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, _)| {
+        // Only attempt sorting if we haven't encountered an error yet
+        if sort_error.is_some() {
+            return Ordering::Equal; // Return a default, we'll handle the error after sorting
+        }
+
+        for i in 0..orders.len() {
+            match composite_intermediate_key_ordering(
+                &left_key[i],
+                &right_key[i],
+                orders[i].0,
+                orders[i].1,
+            ) {
+                Ok(ordering) if ordering != Ordering::Equal => return ordering,
+                Ok(_) => continue, // Equal, try next key
+                Err(err) => {
+                    sort_error = Some(err);
+                    break;
+                }
+            }
+        }
+        Ordering::Equal
+    });
+
+    // If we encountered an error during sorting, return it now
+    if let Some(err) = sort_error {
+        return Err(err);
+    }
+
+    entries.truncate(target_size as usize);
+    Ok(entries)
+}
+
 impl MergeFruits for IntermediateTermBucketEntry {
     fn merge_fruits(&mut self, other: IntermediateTermBucketEntry) -> crate::Result<()> {
         self.doc_count += other.doc_count;

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;