diff --git a/src/aggregation/bucket/histogram/histogram.rs b/src/aggregation/bucket/histogram/histogram.rs index f0dd84a64..fb59cb398 100644 --- a/src/aggregation/bucket/histogram/histogram.rs +++ b/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)?; diff --git a/src/aggregation/bucket/multi_terms/mod.rs b/src/aggregation/bucket/multi_terms/mod.rs index 917e56af4..b6759d47f 100644 --- a/src/aggregation/bucket/multi_terms/mod.rs +++ b/src/aggregation/bucket/multi_terms/mod.rs @@ -216,8 +216,8 @@ fn block_missing_value(missing: Option<&MultiTermsMissingAccessor>) -> Option 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)?; diff --git a/src/aggregation/bucket/term_agg/mod.rs b/src/aggregation/bucket/term_agg/mod.rs index 5ebb75d35..71b3ca7cb 100644 --- a/src/aggregation/bucket/term_agg/mod.rs +++ b/src/aggregation/bucket/term_agg/mod.rs @@ -1066,7 +1066,7 @@ impl 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(); diff --git a/src/aggregation/buffered_sub_aggs.rs b/src/aggregation/buffered_sub_aggs.rs index c3555c832..9ab6a6d8c 100644 --- a/src/aggregation/buffered_sub_aggs.rs +++ b/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 { buffer: B, diff --git a/src/aggregation/segment_agg_result.rs b/src/aggregation/segment_agg_result.rs index 2db57f55c..291addbe7 100644 --- a/src/aggregation/segment_agg_result.rs +++ 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. diff --git a/src/aggregation/value_source/block_accessor.rs b/src/aggregation/value_source/block_accessor.rs index c9d57c95b..20423f55c 100644 --- a/src/aggregation/value_source/block_accessor.rs +++ b/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::>(), + 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();