Preserve document ordering for nested aggregations

src/aggregation/bucket/histogram/histogram.rs

@@ -1097,7 +1097,7 @@ mod tests {
     }
 
     #[test]
-    fn histogram_counts_repeated_documents_in_separate_calls() -> crate::Result<()> {
+    fn histogram_counts_documents_in_separate_calls() -> crate::Result<()> {
         use crate::aggregation::agg_data::{
             build_aggregations_data_from_req, build_segment_agg_collectors, AggregationsSegmentCtx,
         };
@@ -1110,6 +1110,7 @@ mod tests {
         let index = Index::create_in_ram(schema.build());
         let mut writer = index.writer_with_num_threads(1, 20_000_000)?;
         writer.add_document(doc!(value => 12u64, value => 15u64))?;
+        writer.add_document(doc!(value => 12u64, value => 15u64))?;
         writer.commit()?;
         let reader = index.reader()?;
         let searcher = reader.searcher();
@@ -1123,7 +1124,7 @@ mod tests {
         let mut collector = build_segment_agg_collectors(&mut ctx, agg_tree)?;
         collector.prepare_max_bucket(0, &ctx)?;
         collector.collect(0, &[0], &mut ctx)?;
-        collector.collect(0, &[0], &mut ctx)?;
+        collector.collect(0, &[1], &mut ctx)?;
         collector.flush(&mut ctx)?;
         let mut result = IntermediateAggregationResults::default();
         collector.add_intermediate_aggregation_result(&ctx, &mut result, 0)?;

src/aggregation/bucket/multi_terms/mod.rs

@@ -216,8 +216,8 @@ fn block_missing_value(missing: Option<&MultiTermsMissingAccessor>) -> Option<u6
         .map(|missing| missing.missing_value)
 }
 
-/// Fetches one field into the shared block accessor and forwards safe missing handling. Ordering
-/// is requested only by multi-terms, which needs values from different fields aligned by document.
+/// Fetches one field into the shared block accessor with missing values in document order,
+/// keeping values from different fields aligned by document.
 /// Returns whether decoding produced exactly one aligned value per document.
 #[inline]
 fn fetch_field_block(

src/aggregation/bucket/range.rs

@@ -703,7 +703,7 @@ mod tests {
     }
 
     #[test]
-    fn range_counts_repeated_documents_in_separate_calls() -> crate::Result<()> {
+    fn range_counts_documents_in_separate_calls() -> crate::Result<()> {
         use crate::aggregation::agg_data::{
             build_aggregations_data_from_req, build_segment_agg_collectors, AggregationsSegmentCtx,
         };
@@ -716,6 +716,7 @@ mod tests {
         let index = Index::create_in_ram(schema.build());
         let mut writer = index.writer_with_num_threads(1, 20_000_000)?;
         writer.add_document(doc!(value => 12u64, value => 15u64))?;
+        writer.add_document(doc!(value => 12u64, value => 15u64))?;
         writer.commit()?;
         let reader = index.reader()?;
         let searcher = reader.searcher();
@@ -729,7 +730,7 @@ mod tests {
         let mut collector = build_segment_agg_collectors(&mut ctx, agg_tree)?;
         collector.prepare_max_bucket(0, &ctx)?;
         collector.collect(0, &[0], &mut ctx)?;
-        collector.collect(0, &[0], &mut ctx)?;
+        collector.collect(0, &[1], &mut ctx)?;
         collector.flush(&mut ctx)?;
         let mut result = IntermediateAggregationResults::default();
         collector.add_intermediate_aggregation_result(&ctx, &mut result, 0)?;

src/aggregation/bucket/term_agg/mod.rs

@@ -1066,7 +1066,7 @@ impl<TermMap: TermAggregationMap, B: SubAggBuffer> SegmentAggregationCollector
                 docs,
                 &mut *req_data.accessor,
                 req_data.missing_value_for_accessor,
-                false,
+                self.sub_agg.is_some(),
             );
 
         if let Some(sub_agg) = &mut self.sub_agg {
@@ -3187,6 +3187,47 @@ mod tests {
         Ok(())
     }
 
+    #[test]
+    fn terms_missing_with_child_missing_sum() -> crate::Result<()> {
+        for group_values in [&[5u64][..], &[5u64, 6][..]] {
+            let mut schema_builder = Schema::builder();
+            let group = schema_builder.add_u64_field("group", FAST);
+            let metric = schema_builder.add_u64_field("metric", FAST);
+            let index = Index::create_in_ram(schema_builder.build());
+            let mut index_writer = index.writer_with_num_threads(1, 20_000_000)?;
+            // Both documents belong to group 5. Appending missing documents would produce
+            // [1, 0], which breaks the child's missing lookup.
+            index_writer.add_document(doc!(metric => 10u64))?;
+            let mut document = doc!();
+            for &value in group_values {
+                document.add_u64(group, value);
+            }
+            index_writer.add_document(document)?;
+            index_writer.commit()?;
+
+            let request: Aggregations = serde_json::from_value(json!({
+                "groups": {
+                    "terms": { "field": "group", "missing": 5 },
+                    "aggs": {
+                        "total": { "sum": { "field": "metric", "missing": 100 } }
+                    }
+                }
+            }))?;
+            let result = exec_request_with_query(request, &index, None)?;
+            let buckets = result["groups"]["buckets"].as_array().unwrap();
+            assert_eq!(buckets.len(), group_values.len());
+            assert_eq!(buckets[0]["key"], 5.0);
+            assert_eq!(buckets[0]["doc_count"], 2);
+            assert_eq!(buckets[0]["total"]["value"], 110.0);
+            if group_values.len() > 1 {
+                assert_eq!(buckets[1]["key"], 6.0);
+                assert_eq!(buckets[1]["doc_count"], 1);
+                assert_eq!(buckets[1]["total"]["value"], 100.0);
+            }
+        }
+        Ok(())
+    }
+
     #[test]
     fn terms_aggregation_missing_multi_value() -> crate::Result<()> {
         let mut schema_builder = Schema::builder();

src/aggregation/buffered_sub_aggs.rs

@@ -22,6 +22,13 @@ use crate::DocId;
 /// TODO: consider using a more advanced data structure for high cardinality
 /// aggregations.
 /// What this datastructure does in general is to group docs by bucket id.
+///
+/// Submissions can accumulate across parent collection calls. Both backends preserve
+/// submission order for each bucket without sorting or deduplicating, so callers must
+/// supply strictly increasing document IDs per bucket, as required by
+/// [`SegmentAggregationCollector`]. Otherwise a flush can pass unordered or duplicate
+/// document IDs to a child collector even when each parent call contained sorted,
+/// duplicate-free input.
 #[derive(Debug)]
 pub(crate) struct BufferedSubAggs<B: SubAggBuffer> {
     buffer: B,

src/aggregation/segment_agg_result.rs

@@ -23,6 +23,19 @@ impl BucketIdProvider {
 }
 
 /// A SegmentAggregationCollector is used to collect aggregation results.
+///
+/// # Document ordering
+///
+/// For each bucket, submitted document IDs must be strictly increasing across all
+/// calls to [`Self::collect`] and [`Self::collect_multiple`]. A document may be
+/// submitted to different buckets, but only once to any given bucket. Submissions
+/// to different buckets do not need to be ordered relative to each other.
+///
+/// Bucket collectors may buffer submissions across calls and combine them into
+/// one child collector call without sorting or deduplicating. Downstream value
+/// loading requires ascending, duplicate-free document IDs for contiguous-range
+/// decoding and ordered missing-value handling. Ordering only within each call
+/// is therefore insufficient.
 pub trait SegmentAggregationCollector: Debug {
     fn add_intermediate_aggregation_result(
         &mut self,
@@ -33,8 +46,9 @@ pub trait SegmentAggregationCollector: Debug {
 
     /// Collect docs for one bucket.
     ///
-    /// `docs` must be sorted ascending without duplicates. There is no ordering or
-    /// uniqueness guarantee between separate calls.
+    /// `docs` must be strictly increasing. Its first document ID, if any, must be
+    /// greater than every document ID previously submitted to `parent_bucket_id`
+    /// through either collection method. See the trait's document ordering contract.
     ///
     /// The caller must call `prepare_max_bucket` for at least `parent_bucket_id`
     /// before collecting.
@@ -49,13 +63,11 @@ pub trait SegmentAggregationCollector: Debug {
     /// Minimizes dynamic dispatch overhead when collecting many buckets.
     ///
     /// `bucket_ids` and `docs` must have equal lengths; each aligned pair is one
-    /// submission. Within each consecutive run of the same bucket ID, doc IDs
-    /// must be sorted ascending. No `(bucket_id, doc_id)` pair may occur more than
-    /// once in the call, including across separate runs. A doc ID may appear for
-    /// different buckets.
-    ///
-    /// No ordering is required between runs. There is no ordering or uniqueness
-    /// guarantee between separate calls to either collection method.
+    /// submission. For each bucket ID, document IDs must be strictly increasing
+    /// in submission order, including across nonconsecutive runs and previous
+    /// calls to either collection method. A document may appear for different
+    /// buckets; no ordering is required between different buckets. See the trait's
+    /// document ordering contract.
     ///
     /// The caller must call `prepare_max_bucket` for at least the largest submitted
     /// bucket ID before collecting.

src/aggregation/value_source/block_accessor.rs

@@ -148,12 +148,25 @@ impl ColumnBlockAccessor {
             return;
         }
 
-        for &doc in &self.missing_docids_cache {
-            let pos = self.docid_cache.partition_point(|&hit| hit < doc);
-            // TODO insert by back to avoid shifting the same elements
-            // self.missing_docids_cache.len() times
-            self.docid_cache.insert(pos, doc);
-            self.val_cache.insert(pos, missing);
+        // Merge backwards so each present value moves at most once, without overwriting
+        // unread values. Both document runs are already sorted.
+        let mut remaining_values = self.docid_cache.len();
+        let mut remaining_missing = self.missing_docids_cache.len();
+        let len = remaining_values + remaining_missing;
+        self.docid_cache.resize(len, 0);
+        self.val_cache.resize(len, missing);
+        while remaining_missing > 0 {
+            let doc = self.missing_docids_cache[remaining_missing - 1];
+            let target = remaining_values + remaining_missing - 1;
+            if remaining_values > 0 && self.docid_cache[remaining_values - 1] > doc {
+                remaining_values -= 1;
+                self.docid_cache[target] = self.docid_cache[remaining_values];
+                self.val_cache[target] = self.val_cache[remaining_values];
+            } else {
+                remaining_missing -= 1;
+                self.docid_cache[target] = doc;
+                self.val_cache[target] = missing;
+            }
         }
     }
 
@@ -592,6 +605,44 @@ mod tests {
         );
     }
 
+    #[test]
+    fn test_multivalued_missing_merge() {
+        let docs = [0, 2, 5, 9];
+        let mut accessor = ColumnBlockAccessor::default();
+        // Cover missing documents before, between, and after present documents, including
+        // entirely missing/present batches. Preserve repeated values and their order.
+        for present_docs in 0..1 << docs.len() {
+            let mut entries = Vec::new();
+            let mut expected = Vec::new();
+            for (idx, &doc) in docs.iter().enumerate() {
+                if present_docs & (1 << idx) == 0 {
+                    expected.push((doc, 99));
+                } else {
+                    let values = [(doc, 20), (doc, 10), (doc, 20)];
+                    entries.extend(values);
+                    expected.extend(values);
+                }
+            }
+            let mut source = TestValueSource {
+                cardinality: Cardinality::Multivalued,
+                entries,
+            };
+            accessor.fetch_block_with_missing_ordered(&docs, &mut source, Some(99), true);
+            assert_eq!(
+                accessor.iter_docid_vals(&docs).collect::<Vec<_>>(),
+                expected
+            );
+        }
+
+        let mut source = TestValueSource {
+            cardinality: Cardinality::Multivalued,
+            entries: Vec::new(),
+        };
+        accessor.fetch_block_with_missing_ordered(&[], &mut source, Some(99), true);
+        assert!(accessor.values().is_empty());
+        assert!(accessor.docids().is_empty());
+    }
+
     #[test]
     fn test_sort_values_per_doc_for_dedup() {
         let mut accessor = ColumnBlockAccessor::default();