From 21ae09c0b44075aff7f61eb38893e6ff9d520bc5 Mon Sep 17 00:00:00 2001 From: Paul Masurel Date: Tue, 1 Sep 2026 15:10:19 +0200 Subject: [PATCH] Refactoring to introduce calculated fields. (#3066) * Make aggregation column block accessor private Move ColumnBlockAccessor out of tantivy-columnar and into the aggregation implementation. Keep its behavior and tests intact while removing it from the public columnar API. * Refactor aggregation value block access * Following comment Removing Fused from the element that are private to the module. Re-added optimization in stats for single doc requests to avoid perf regression Renamed Fused -> Flattened --------- Co-authored-by: Paul Masurel --- benches/agg_bench.rs | 8 +- columnar/src/lib.rs | 2 - src/aggregation/agg_data.rs | 6 +- .../src => src/aggregation}/block_accessor.rs | 327 +++++++++++++----- src/aggregation/bucket/histogram/histogram.rs | 17 +- src/aggregation/bucket/multi_terms/mod.rs | 27 +- src/aggregation/bucket/range.rs | 5 +- ...stogram.rs => flattened_term_histogram.rs} | 157 +++++---- src/aggregation/bucket/term_agg/mod.rs | 16 +- src/aggregation/metric/stats.rs | 7 +- src/aggregation/mod.rs | 3 + src/store/reader.rs | 4 +- 12 files changed, 360 insertions(+), 219 deletions(-) rename {columnar/src => src/aggregation}/block_accessor.rs (57%) rename src/aggregation/bucket/term_agg/{term_histogram.rs => flattened_term_histogram.rs} (88%) diff --git a/benches/agg_bench.rs b/benches/agg_bench.rs index b9f55b564..c02a4bc32 100644 --- a/benches/agg_bench.rs +++ b/benches/agg_bench.rs @@ -500,7 +500,7 @@ fn terms_status_with_date_histogram() -> AggregationRequest { }) } -/// Same fused terms × date_histogram, but with `hard_bounds`. The timestamps span 0..120h; the +/// Same flattened terms × date_histogram, but with `hard_bounds`. The timestamps span 0..120h; the /// bounds drop only the first and last hour (ms: 1h=3_600_000, 119h=428_400_000), so almost every /// doc is in-bounds. This exercises the collector's hard-bounds path: `bounds.contains` runs per /// doc (the `all_docs_in_bounds` short-circuit is off) and the rare out-of-bounds doc takes the @@ -562,9 +562,9 @@ fn terms_status_with_date_histogram_hard_bounds() -> AggregationRequest { }) } -/// Same fused terms × date_histogram, but with a sibling terms aggregation next to it. The fused -/// fast path should still trigger for `my_texts` (sibling aggregations are independent top-level -/// aggregations, so they don't change its eligibility). +/// Same flattened terms × date_histogram, but with a sibling terms aggregation next to it. The +/// flattened fast path should still trigger for `my_texts` (sibling aggregations are independent +/// top-level aggregations, so they don't change its eligibility). fn terms_status_with_date_histogram_and_sibling_terms() -> AggregationRequest { json!({ "my_texts": { diff --git a/columnar/src/lib.rs b/columnar/src/lib.rs index 2db89ef28..1da8d9604 100644 --- a/columnar/src/lib.rs +++ b/columnar/src/lib.rs @@ -24,7 +24,6 @@ extern crate more_asserts; use std::fmt::Display; use std::io; -mod block_accessor; mod column; pub mod column_index; pub mod column_values; @@ -35,7 +34,6 @@ mod iterable; pub(crate) mod utils; mod value; -pub use block_accessor::ColumnBlockAccessor; pub use column::{BytesColumn, Column, StrColumn}; pub use column_index::ColumnIndex; pub use column_values::{ diff --git a/src/aggregation/agg_data.rs b/src/aggregation/agg_data.rs index 0027c17d4..475cef2ea 100644 --- a/src/aggregation/agg_data.rs +++ b/src/aggregation/agg_data.rs @@ -1,6 +1,6 @@ use std::sync::Arc; -use columnar::{Column, ColumnBlockAccessor, ColumnType, StrColumn}; +use columnar::{Column, ColumnType, StrColumn}; use common::BitSet; use rustc_hash::FxHashSet; use serde::Serialize; @@ -29,7 +29,7 @@ use crate::aggregation::metric::{ use crate::aggregation::segment_agg_result::{ GenericSegmentAggregationResultsCollector, SegmentAggregationCollector, }; -use crate::aggregation::{f64_to_fastfield_u64, AggContextParams, Key}; +use crate::aggregation::{f64_to_fastfield_u64, AggContextParams, ColumnBlockAccessor, Key}; use crate::{SegmentOrdinal, SegmentReader}; #[derive(Default)] @@ -39,7 +39,7 @@ pub struct AggregationsSegmentCtx { /// Request data for each aggregation type. pub per_request: PerRequestAggSegCtx, pub context: AggContextParams, - pub column_block_accessor: ColumnBlockAccessor, + pub(crate) column_block_accessor: ColumnBlockAccessor, } impl AggregationsSegmentCtx { diff --git a/columnar/src/block_accessor.rs b/src/aggregation/block_accessor.rs similarity index 57% rename from columnar/src/block_accessor.rs rename to src/aggregation/block_accessor.rs index ec6abe7e2..2eb592f1b 100644 --- a/columnar/src/block_accessor.rs +++ b/src/aggregation/block_accessor.rs @@ -1,86 +1,117 @@ use std::cmp::Ordering; -use crate::{Column, DocId, RowId}; +use columnar::{Cardinality, Column, RowId}; -#[derive(Debug, Default, Clone)] -pub struct ColumnBlockAccessor { - val_cache: Vec, - docid_cache: Vec, - missing_docids_cache: Vec, - row_id_cache: Vec, +use crate::DocId; + +/// A source of values for a block of documents. +/// +/// Implementations replace the contents of `values` and, for non-full sources, `docids`. The +/// returned cardinality describes how the two buffers are aligned. Full sources must return one +/// value per input document in the same order. Optional and multivalued sources must populate +/// `docids` with one document id per value. +pub(crate) trait BlockValueSource { + fn load_block( + &self, + docs: &[DocId], + values: &mut Vec, + docids: &mut Vec, + row_ids: &mut Vec, + ) -> Cardinality; } -impl - ColumnBlockAccessor -{ +/// Buffers the values associated with a block of documents loaded from a [`BlockValueSource`]. +/// +/// Regardless of their original types, values are loaded in their `u64` representation using the +/// associated monotonic mapping. +#[derive(Debug, Default, Clone)] +pub(crate) struct ColumnBlockAccessor { + /// Values loaded for the latest document block, in monotonic `u64` representation. + val_cache: Vec, + /// Document ID corresponding to each value in `val_cache` for a non-full source. + /// + /// A document can occur more than once for a multivalued source. For a full source this buffer + /// is ignored because `val_cache` is aligned directly with the requested document block. + docid_cache: Vec, + /// Scratch buffer used to identify documents for which a missing value must be inserted. + missing_docids_cache: Vec, + /// Scratch buffer available to sources for translating document IDs into value row IDs. + row_id_cache: Vec, + /// Cardinality reported by the source that loaded the latest block. + /// For the moment this is reporting the cardinality of the full column, not + /// something specific to the block. + cardinality: Cardinality, +} + +impl BlockValueSource for Column { #[inline] - pub fn fetch_block<'a>(&'a mut self, docs: &'a [u32], accessor: &Column) { - self.fetch_block_with_is_full(docs, accessor, accessor.index.get_cardinality().is_full()); + fn load_block( + &self, + docs: &[DocId], + values: &mut Vec, + docids: &mut Vec, + row_ids: &mut Vec, + ) -> Cardinality { + let cardinality = self.index.get_cardinality(); + if cardinality.is_full() { + load_full_column_values(docs, self, values); + } else { + docids.clear(); + row_ids.clear(); + self.row_ids_for_docs(docs, docids, row_ids); + values.resize(row_ids.len(), 0u64); + self.values.get_vals(row_ids, values); + } + cardinality + } +} + +impl ColumnBlockAccessor { + #[inline] + pub(crate) fn fetch_block(&mut self, docs: &[DocId], source: &impl BlockValueSource) { + self.cardinality = source.load_block( + docs, + &mut self.val_cache, + &mut self.docid_cache, + &mut self.row_id_cache, + ); } - /// Like [`Self::fetch_block`] but takes the column's fullness instead of querying - /// `accessor.index.get_cardinality()` each call — for callers that know it up front (e.g. - /// checked once at construction). `is_full` must equal - /// `accessor.index.get_cardinality().is_full()`. + /// Fetches a physical column known to be full without querying its cardinality. + /// + /// This direct-column-only entry point is reserved for specialized collectors whose + /// construction already proved the column is full. #[inline] - pub fn fetch_block_with_is_full<'a>( - &'a mut self, - docs: &'a [u32], - accessor: &Column, - is_full: bool, - ) { - if is_full { - // Skip the resize when already the right length (common case: fixed-size blocks). - if self.val_cache.len() != docs.len() { - self.val_cache.resize(docs.len(), T::default()); - } - // When the docs form a contiguous ascending run we can fetch the values - // as a single range. This lets codecs (e.g. bitpacked) bulk-decode the - // slice instead of gathering value-by-value, and avoids per-value dynamic - // dispatch. `docs` is always sorted ascending and free of duplicates here, - // so comparing the endpoints is enough to detect contiguity. - if is_contiguous(docs) { - accessor - .values - .get_range(docs[0] as u64, &mut self.val_cache); - } else { - accessor.values.get_vals(docs, &mut self.val_cache); - } - } else { - self.docid_cache.clear(); - self.row_id_cache.clear(); - accessor.row_ids_for_docs(docs, &mut self.docid_cache, &mut self.row_id_cache); - self.val_cache.resize(self.row_id_cache.len(), T::default()); - accessor - .values - .get_vals(&self.row_id_cache, &mut self.val_cache); - } + pub(crate) fn fetch_full_column_block(&mut self, docs: &[DocId], accessor: &Column) { + debug_assert!(accessor.index.get_cardinality().is_full()); + load_full_column_values(docs, accessor, &mut self.val_cache); + self.cardinality = Cardinality::Full; } /// Fetches a block and appends `missing_opt` for documents without a value. #[inline] - pub fn fetch_block_with_missing( + pub(crate) fn fetch_block_with_missing( &mut self, - docs: &[u32], - accessor: &Column, - missing_opt: Option, + docs: &[DocId], + source: &impl BlockValueSource, + missing_opt: Option, ) { - self.fetch_block_with_missing_ordered(docs, accessor, missing_opt, false); + self.fetch_block_with_missing_ordered(docs, source, missing_opt, false) } /// Fetches a block and adds `missing_opt` for documents without a value. When `ordered` is /// true, the missing entries are inserted in document order instead of appended as a second /// run. #[inline] - pub fn fetch_block_with_missing_ordered( + pub(crate) fn fetch_block_with_missing_ordered( &mut self, - docs: &[u32], - accessor: &Column, - missing_opt: Option, + docs: &[DocId], + source: &impl BlockValueSource, + missing_opt: Option, ordered: bool, ) { - self.fetch_block(docs, accessor); - let cardinality = accessor.index.get_cardinality(); + self.fetch_block(docs, source); + let cardinality = self.cardinality; // no missing values if cardinality.is_full() { return; @@ -142,17 +173,15 @@ impl /// This is necessary for correct document counting in aggregations, /// where multi-valued fields can produce duplicate entries that inflate counts. #[inline] - pub fn fetch_block_with_missing_unique_per_doc( + pub(crate) fn fetch_block_with_missing_unique_per_doc( &mut self, - docs: &[u32], - accessor: &Column, - missing: Option, + docs: &[DocId], + source: &impl BlockValueSource, + missing: Option, ordered: bool, - ) where - T: Ord, - { - self.fetch_block_with_missing_ordered(docs, accessor, missing, ordered); - if accessor.index.get_cardinality().is_multivalue() { + ) { + self.fetch_block_with_missing_ordered(docs, source, missing, ordered); + if self.cardinality.is_multivalue() { self.dedup_docid_val_pairs(); } } @@ -165,8 +194,7 @@ impl /// if it has more than 2 elements, then deduplicate adjacent pairs. /// /// Skips entirely if no doc_id appears more than once in the block. - fn dedup_docid_val_pairs(&mut self) - where T: Ord { + fn dedup_docid_val_pairs(&mut self) { if self.docid_cache.len() <= 1 { return; } @@ -211,18 +239,30 @@ impl /// Returns the values fetched by the last `fetch_block*` call. #[inline] - pub fn values(&self) -> &[T] { + pub(crate) fn values(&self) -> &[u64] { &self.val_cache } /// Returns the document IDs corresponding to [`Self::values`] for a non-full column. #[inline] - pub fn docids(&self) -> &[DocId] { + pub(crate) fn docids(&self) -> &[DocId] { &self.docid_cache } + /// Returns whether the last fetched block contains exactly one aligned value per input doc. #[inline] - pub fn iter_vals(&self) -> impl ExactSizeIterator + '_ { + pub(crate) fn has_one_value_per_doc(&self, docs: &[DocId]) -> bool { + self.val_cache.len() == docs.len() + && (self.cardinality.is_full() || self.docid_cache == docs) + } + + #[inline] + pub(crate) fn is_multivalued(&self) -> bool { + self.cardinality.is_multivalue() + } + + #[inline] + pub(crate) fn iter_vals(&self) -> impl ExactSizeIterator + '_ { self.val_cache.iter().cloned() } @@ -231,14 +271,13 @@ impl /// The passed in `docs` slice needs to be the same slice that was passed to `fetch_block` or /// `fetch_block_with_missing`. /// - /// The docs is used if the column is full (each docs has exactly one value), otherwise the - /// internal docid vec is used for the iterator, which e.g. may contain duplicate docs. - pub fn iter_docid_vals<'a>( + /// The docs are used if the source is full (each doc has exactly one value); otherwise the + /// internal docid vec is used and may contain duplicate docs. + pub(crate) fn iter_docid_vals<'a>( &'a self, - docs: &'a [u32], - accessor: &Column, - ) -> impl Iterator + 'a + use<'a, T> { - if accessor.index.get_cardinality().is_full() { + docs: &'a [DocId], + ) -> impl Iterator + 'a { + if self.cardinality.is_full() { docs.iter().cloned().zip(self.val_cache.iter().cloned()) } else { self.docid_cache @@ -249,6 +288,21 @@ impl } } +#[inline] +fn load_full_column_values(docs: &[DocId], accessor: &Column, values: &mut Vec) { + // Skip the resize when already the right length (common case: fixed-size blocks). + if values.len() != docs.len() { + values.resize(docs.len(), 0u64); + } + // When the docs form a contiguous ascending run we can fetch the values as a single range. + // This lets codecs (e.g. bitpacked) bulk-decode the slice instead of gathering value-by-value. + if is_contiguous(docs) { + accessor.values.get_range(docs[0] as u64, values); + } else { + accessor.values.get_vals(docs, values); + } +} + /// Returns true if `docs` is a contiguous ascending run `[d, d + 1, ..., d + n - 1]`. /// /// Assumes `docs` is sorted ascending and free of duplicates (the invariant for the @@ -304,6 +358,33 @@ fn find_missing_docs(docs: &[u32], hits: &[u32], output: &mut Vec) { mod tests { use super::*; + struct TestValueSource { + cardinality: Cardinality, + entries: Vec<(DocId, u64)>, + } + + impl BlockValueSource for TestValueSource { + fn load_block( + &self, + docs: &[DocId], + values: &mut Vec, + docids: &mut Vec, + row_ids: &mut Vec, + ) -> Cardinality { + values.clear(); + docids.clear(); + row_ids.clear(); + if self.cardinality.is_full() { + assert_eq!(self.entries.len(), docs.len()); + values.extend(self.entries.iter().map(|(_, value)| *value)); + } else { + docids.extend(self.entries.iter().map(|(doc, _)| *doc)); + values.extend(self.entries.iter().map(|(_, value)| *value)); + } + self.cardinality + } + } + #[test] fn test_find_missing_docs() { let docs: Vec = vec![1, 2, 3, 4, 5, 6, 7, 8, 9, 10]; @@ -337,11 +418,65 @@ mod tests { assert_eq!(missing_docs, vec![1, 2, 3, 4, 5]); } + #[test] + fn test_source_neutral_full_block_alignment() { + let docs = [2, 4, 8]; + let source = TestValueSource { + cardinality: Cardinality::Full, + entries: vec![(2, 20), (4, 40), (8, 80)], + }; + let mut accessor = ColumnBlockAccessor::default(); + + accessor.fetch_block(&docs, &source); + + assert!(accessor.has_one_value_per_doc(&docs)); + assert_eq!( + accessor.iter_docid_vals(&docs).collect::>(), + vec![(2, 20), (4, 40), (8, 80)] + ); + } + + #[test] + fn test_source_neutral_optional_block_with_missing() { + let docs = [0, 1, 2, 4]; + let source = TestValueSource { + cardinality: Cardinality::Optional, + entries: vec![(1, 10), (4, 40)], + }; + let mut accessor = ColumnBlockAccessor::default(); + + accessor.fetch_block_with_missing_ordered(&docs, &source, Some(99), true); + + assert!(accessor.has_one_value_per_doc(&docs)); + assert_eq!( + accessor.iter_docid_vals(&docs).collect::>(), + vec![(0, 99), (1, 10), (2, 99), (4, 40)] + ); + } + + #[test] + fn test_source_neutral_multivalue_block_deduplication() { + let docs = [0, 1]; + let source = TestValueSource { + cardinality: Cardinality::Multivalued, + entries: vec![(0, 3), (0, 1), (0, 3), (1, 5), (1, 5)], + }; + let mut accessor = ColumnBlockAccessor::default(); + + accessor.fetch_block_with_missing_unique_per_doc(&docs, &source, None, false); + + assert!(!accessor.has_one_value_per_doc(&docs)); + assert_eq!( + accessor.iter_docid_vals(&docs).collect::>(), + vec![(0, 1), (0, 3), (1, 5)] + ); + } + #[test] fn test_fetch_block_with_missing_ordered() { - use crate::column_index::{ColumnIndex, OptionalIndex}; - use crate::column_values::{ - ALL_U64_CODEC_TYPES, serialize_and_load_u64_based_column_values, + use columnar::column_index::{ColumnIndex, OptionalIndex}; + use columnar::column_values::{ + serialize_and_load_u64_based_column_values, ALL_U64_CODEC_TYPES, }; let vals = [10u64, 40, 70]; @@ -352,7 +487,7 @@ mod tests { values, }; let docs = [0, 1, 2, 4, 7, 8]; - let mut accessor = ColumnBlockAccessor::::default(); + let mut accessor = ColumnBlockAccessor::default(); accessor.fetch_block_with_missing_ordered(&docs, &column, Some(99), true); @@ -361,14 +496,14 @@ mod tests { vec![99, 10, 99, 40, 70, 99] ); assert_eq!( - accessor.iter_docid_vals(&docs, &column).collect::>(), + accessor.iter_docid_vals(&docs).collect::>(), vec![(0, 99), (1, 10), (2, 99), (4, 40), (7, 70), (8, 99)] ); } #[test] fn test_dedup_docid_val_pairs_consecutive() { - let mut accessor = ColumnBlockAccessor::::default(); + let mut accessor = ColumnBlockAccessor::default(); accessor.docid_cache = vec![0, 0, 2, 3]; accessor.val_cache = vec![10, 10, 10, 10]; accessor.dedup_docid_val_pairs(); @@ -379,7 +514,7 @@ mod tests { #[test] fn test_dedup_docid_val_pairs_non_consecutive() { // (0,1), (0,2), (0,1) — duplicate value not adjacent - let mut accessor = ColumnBlockAccessor::::default(); + let mut accessor = ColumnBlockAccessor::default(); accessor.docid_cache = vec![0, 0, 0]; accessor.val_cache = vec![1, 2, 1]; accessor.dedup_docid_val_pairs(); @@ -390,7 +525,7 @@ mod tests { #[test] fn test_dedup_docid_val_pairs_multi_doc() { // doc 0: values [3, 1, 3], doc 1: values [5, 5] - let mut accessor = ColumnBlockAccessor::::default(); + let mut accessor = ColumnBlockAccessor::default(); accessor.docid_cache = vec![0, 0, 0, 1, 1]; accessor.val_cache = vec![3, 1, 3, 5, 5]; accessor.dedup_docid_val_pairs(); @@ -400,7 +535,7 @@ mod tests { #[test] fn test_dedup_docid_val_pairs_no_duplicates() { - let mut accessor = ColumnBlockAccessor::::default(); + let mut accessor = ColumnBlockAccessor::default(); accessor.docid_cache = vec![0, 0, 1]; accessor.val_cache = vec![1, 2, 3]; accessor.dedup_docid_val_pairs(); @@ -410,7 +545,7 @@ mod tests { #[test] fn test_dedup_docid_val_pairs_single_element() { - let mut accessor = ColumnBlockAccessor::::default(); + let mut accessor = ColumnBlockAccessor::default(); accessor.docid_cache = vec![0]; accessor.val_cache = vec![1]; accessor.dedup_docid_val_pairs(); @@ -430,9 +565,9 @@ mod tests { #[test] fn test_fetch_block_contiguous_and_gather_match() { - use crate::column_index::ColumnIndex; - use crate::column_values::{ - ALL_U64_CODEC_TYPES, serialize_and_load_u64_based_column_values, + use columnar::column_index::ColumnIndex; + use columnar::column_values::{ + serialize_and_load_u64_based_column_values, ALL_U64_CODEC_TYPES, }; let vals: Vec = (0..200u64).map(|i| i * 7 + 3).collect(); @@ -443,14 +578,14 @@ mod tests { values, }; - let check = |accessor: &mut ColumnBlockAccessor, docs: &[u32]| { + let check = |accessor: &mut ColumnBlockAccessor, docs: &[u32]| { accessor.fetch_block(docs, &column); - let got: Vec<(u32, u64)> = accessor.iter_docid_vals(docs, &column).collect(); + let got: Vec<(u32, u64)> = accessor.iter_docid_vals(docs).collect(); let expected: Vec<(u32, u64)> = docs.iter().map(|&d| (d, vals[d as usize])).collect(); assert_eq!(got, expected); }; - let mut accessor = ColumnBlockAccessor::::default(); + let mut accessor = ColumnBlockAccessor::default(); // Contiguous block -> get_range fast path. check(&mut accessor, &(10..74).collect::>()); // Non-contiguous block -> get_vals gather path. diff --git a/src/aggregation/bucket/histogram/histogram.rs b/src/aggregation/bucket/histogram/histogram.rs index 0942a08f7..e97a9ca5c 100644 --- a/src/aggregation/bucket/histogram/histogram.rs +++ b/src/aggregation/bucket/histogram/histogram.rs @@ -494,10 +494,7 @@ impl SegmentAggregationCollector for SegmentHistogramCollector< .fetch_block(docs, &req.accessor); // special path for nested buckets if let Some(sub_agg) = &mut self.sub_agg { - for (doc, val) in agg_data - .column_block_accessor - .iter_docid_vals(docs, &req.accessor) - { + for (doc, val) in agg_data.column_block_accessor.iter_docid_vals(docs) { let val = f64_from_fastfield_u64(val, req.field_type); if bounds.contains(val) { let bucket = store.get_or_create( @@ -633,9 +630,9 @@ impl SegmentHistogramCollector { impl SegmentHistogramCollector<()> { /// Builds a histogram collector whose parent `t` is a dense histogram filled from /// `counts[t * num_time_buckets .. (t + 1) * num_time_buckets]` (row-major), consolidating each - /// cell's count lanes. Used by the fused terms×histogram collector to turn its flat 2D counters - /// into the regular intermediate result, so cross-segment merging is shared with the general - /// path. + /// cell's count lanes. Used by the flattened terms×histogram collector to turn its flat 2D + /// counters into the regular intermediate result, so cross-segment merging is shared with the + /// general path. pub(crate) fn from_dense_rows( req_data: HistogramAggReqData, base_pos: i64, @@ -688,7 +685,7 @@ fn normalize_histogram_req(req_data: &mut HistogramAggReqData) -> crate::Result< req_data.offset = req_data.req.offset.unwrap_or(0.0); // Drop `hard_bounds` that can't exclude any value (the column's range already sits inside // them): the per-doc `bounds.contains` check is then a no-op, so collapsing to the unbounded - // sentinel lets the histogram hot loop skip it and the fused term×histogram path derive + // sentinel lets the histogram hot loop skip it and the flattened term×histogram path derive // per-term counts from the grid. Only this collect-time filter is touched — empty-bucket // emission reads `req.hard_bounds` directly (see `get_req_min_max`), and `hard_bounds` only // ever clips that range, so a wider-than-data bound leaves the result unchanged. @@ -707,7 +704,7 @@ fn normalize_histogram_req(req_data: &mut HistogramAggReqData) -> crate::Result< /// Clones and normalizes (resolving interval/offset/bounds) the histogram request at `node`, and /// returns it together with its dense bucket range — or `None` if the column has no usable range. -/// Used by the fused terms×histogram collector, which then owns the normalized request. +/// Used by the flattened terms×histogram collector, which then owns the normalized request. pub(crate) fn prepare_histogram_dense_range( agg_data: &AggregationsSegmentCtx, node: &AggRefNode, @@ -1440,7 +1437,7 @@ mod tests { /// `hard_bounds` wider than the data (here with mid-interval edges, to cover the "bound cuts a /// bucket" case) can't exclude any value, so the result must be identical to the same request /// without bounds. Guards the normalization that collapses such bounds to the unbounded - /// sentinel so the hot loop / fused path can skip the per-doc bounds check. + /// sentinel so the hot loop / flattened path can skip the per-doc bounds check. fn histogram_non_binding_hard_bounds_test_with_opt(merge_segments: bool) -> crate::Result<()> { let values = vec![10.0, 12.0, 14.0, 16.0, 10.0, 13.0, 10.0, 12.0]; let index = get_test_index_from_values(merge_segments, &values)?; diff --git a/src/aggregation/bucket/multi_terms/mod.rs b/src/aggregation/bucket/multi_terms/mod.rs index 60fd3b5f5..3c40b1d70 100644 --- a/src/aggregation/bucket/multi_terms/mod.rs +++ b/src/aggregation/bucket/multi_terms/mod.rs @@ -4,8 +4,8 @@ use std::sync::Arc; use columnar::column_values::CompactSpaceU64Accessor; use columnar::{ - Column, ColumnBlockAccessor, ColumnType, Dictionary, MonotonicallyMappableToU128, - MonotonicallyMappableToU64, NumericalValue, StrColumn, + Column, ColumnType, Dictionary, MonotonicallyMappableToU128, MonotonicallyMappableToU64, + NumericalValue, StrColumn, }; use rustc_hash::FxHashMap; use serde::{Deserialize, Serialize}; @@ -30,7 +30,7 @@ use crate::aggregation::intermediate_agg_result::{ IntermediateKey, IntermediateTermBucketEntry, PruneMode, }; use crate::aggregation::segment_agg_result::{BucketIdProvider, SegmentAggregationCollector}; -use crate::aggregation::{f64_to_fastfield_u64, format_date, BucketId, Key}; +use crate::aggregation::{f64_to_fastfield_u64, format_date, BucketId, ColumnBlockAccessor, Key}; use crate::{DocId, TantivyError}; /// Multi-terms aggregation: one bucket per unique combination of values across N term fields. @@ -225,7 +225,7 @@ fn fetch_field_block( docs: &[crate::DocId], field: &MultiTermsFieldAccessor, missing: Option<&MultiTermsMissingAccessor>, - block_accessor: &mut ColumnBlockAccessor, + block_accessor: &mut ColumnBlockAccessor, ) -> bool { let missing_value = block_missing_value(missing); block_accessor.fetch_block_with_missing_unique_per_doc( @@ -234,10 +234,7 @@ fn fetch_field_block( missing_value, true, ); - if block_accessor.values().len() != docs.len() { - return false; - } - !field.column.get_cardinality().is_multivalue() || block_accessor.docids() == docs + block_accessor.has_one_value_per_doc(docs) } /// Packing operations used by the unified collector. @@ -461,9 +458,8 @@ fn missing_value_for_doc( fn expand_partial_combinations_for_field( packing: &Packing, field_idx: usize, - field: &MultiTermsFieldAccessor, missing: Option<&MultiTermsMissingAccessor>, - block_accessor: &ColumnBlockAccessor, + block_accessor: &ColumnBlockAccessor, keys_buf: &mut Vec, alive_docs: &mut Vec, doc_ids_per_partial_combination: &mut Vec, @@ -483,9 +479,7 @@ fn expand_partial_combinations_for_field( ); { - let mut field_values = block_accessor - .iter_docid_vals(alive_docs, &field.column) - .peekable(); + let mut field_values = block_accessor.iter_docid_vals(alive_docs).peekable(); let mut doc_values = SmallVec::<[u64; 2]>::new(); let mut combination_start = 0; @@ -575,7 +569,7 @@ where fn build_keys_from_full_fields( &mut self, docs: &[DocId], - block_accessor: &mut ColumnBlockAccessor, + block_accessor: &mut ColumnBlockAccessor, ) { for (field_idx, field) in self.req_data.fields.iter().enumerate() { fetch_field_block( @@ -595,7 +589,7 @@ where fn build_keys_from_non_full_fields( &mut self, docs: &[DocId], - block_accessor: &mut ColumnBlockAccessor, + block_accessor: &mut ColumnBlockAccessor, ) { self.alive_docs.clear(); self.alive_docs.extend_from_slice(docs); @@ -632,7 +626,7 @@ where // Until expansion, sparse single-value fields can filter keys in place. if self.doc_ids_per_partial_combination.is_empty() - && !field.column.get_cardinality().is_multivalue() + && !block_accessor.is_multivalued() && missing.is_none() { let mut source_idx = 0usize; @@ -660,7 +654,6 @@ where expand_partial_combinations_for_field( &self.packing, field_idx, - field, missing, block_accessor, &mut self.keys_buf, diff --git a/src/aggregation/bucket/range.rs b/src/aggregation/bucket/range.rs index abfdec330..a66ea8225 100644 --- a/src/aggregation/bucket/range.rs +++ b/src/aggregation/bucket/range.rs @@ -285,10 +285,7 @@ impl SegmentAggregationCollector for SegmentRangeCollector { let buckets = &mut self.parent_buckets[parent_bucket_id as usize]; - for (doc, val) in agg_data - .column_block_accessor - .iter_docid_vals(docs, &self.req_data.accessor) - { + for (doc, val) in agg_data.column_block_accessor.iter_docid_vals(docs) { let bucket_pos = get_bucket_pos(val, buckets); let bucket = &mut buckets[bucket_pos]; bucket.bucket.doc_count += 1; diff --git a/src/aggregation/bucket/term_agg/term_histogram.rs b/src/aggregation/bucket/term_agg/flattened_term_histogram.rs similarity index 88% rename from src/aggregation/bucket/term_agg/term_histogram.rs rename to src/aggregation/bucket/term_agg/flattened_term_histogram.rs index 05578f780..675171448 100644 --- a/src/aggregation/bucket/term_agg/term_histogram.rs +++ b/src/aggregation/bucket/term_agg/flattened_term_histogram.rs @@ -1,12 +1,12 @@ -//! Fused collector for the very common shape `terms` (low cardinality) × a single +//! Flattened collector for the very common shape `terms` (low cardinality) × a single //! `histogram`/`date_histogram` sub-aggregation with nothing nested below it. //! -//! See [`SegmentTermHistogramCollector`] for the approach and [`maybe_build_collector`] for the -//! conditions under which it is used. +//! See [`FlattenedTermHistogramCollector`] for the approach and +//! [`maybe_build_flattened_collector`] for the conditions under which it is used. use std::fmt::Debug; -use columnar::{Column, ColumnBlockAccessor, ColumnType}; +use columnar::{Column, ColumnType}; use super::{ Bucket, SegmentTermCollector, TermsAggReqData, VecTermBuckets, MAX_NUM_BUCKETS_FOR_COUNT_LANES, @@ -22,14 +22,14 @@ use crate::aggregation::intermediate_agg_result::{ IntermediateAggregationResult, IntermediateAggregationResults, }; use crate::aggregation::segment_agg_result::{BucketIdProvider, SegmentAggregationCollector}; -use crate::aggregation::{f64_from_fastfield_u64, BucketId}; +use crate::aggregation::{f64_from_fastfield_u64, BucketId, ColumnBlockAccessor}; -/// Maximum number of physical counters in the fused flat grid. Above this the grid would be too +/// Maximum number of physical counters in the flattened flat grid. Above this the grid would be too /// large/cache-unfriendly, so we fall back to the general buffered path. Count lanes are included /// in this limit: `num_terms × num_time_buckets × LANES` may not exceed it. /// /// Since we are only at the top-level, this won't be multiplied by any parent buckets. -const MAX_FUSED_GRID_COUNTERS: usize = 16_384; +const MAX_FLATTENED_GRID_COUNTERS: usize = 16_384; /// Scalar storage for grids whose term cardinality is too high for count lanes to pay off. const SINGLE_COUNT_LANE: usize = 1; @@ -41,7 +41,7 @@ const NUM_LARGE_LINEAR_BUCKETS: usize = 8; trait BucketResolver: Debug + 'static { /// Fetches the histogram values needed for this block. Resolvers that do not inspect the - /// histogram column (notably [`SingleBucketResolver`]) leave this as a no-op. + /// histogram column (notably [`FlattenedSingleBucketResolver`]) leave this as a no-op. fn prepare_block(&mut self, docs: &[crate::DocId]); /// Number of logical histogram buckets produced by this resolver. @@ -81,11 +81,21 @@ fn increment_grid_count( /// Resolver for a histogram whose entire value range maps to one bucket. It deliberately owns no /// block accessor: collecting this shape does not read or decode the histogram column at all. -#[derive(Debug, Default)] +#[derive(Debug)] struct SingleBucketResolver { next_count_lane: usize, } +impl SingleBucketResolver { + fn new(hist_req_data: &HistogramAggReqData) -> Self { + assert!( + hist_req_data.accessor.get_cardinality().is_full(), + "SingleBucketResolver requires a full histogram column" + ); + Self { next_count_lane: 0 } + } +} + impl BucketResolver for SingleBucketResolver { #[inline] fn prepare_block(&mut self, _docs: &[crate::DocId]) {} @@ -121,10 +131,9 @@ impl BucketResolver for SingleBucketResolver { /// calculation for histograms that do not use a specialized resolver. #[derive(Debug)] struct ComputedBucketResolver { - hist_block: ColumnBlockAccessor, + hist_block: ColumnBlockAccessor, next_count_lane: usize, accessor: Column, - is_full: bool, field_type: ColumnType, interval: f64, offset: f64, @@ -135,11 +144,14 @@ struct ComputedBucketResolver { impl ComputedBucketResolver { fn new(hist_req_data: &HistogramAggReqData, base_pos: i64, num_buckets: usize) -> Self { + assert!( + hist_req_data.accessor.get_cardinality().is_full(), + "ComputedBucketResolver requires a full histogram column" + ); Self { hist_block: ColumnBlockAccessor::default(), next_count_lane: 0, accessor: hist_req_data.accessor.clone(), - is_full: hist_req_data.accessor.get_cardinality().is_full(), field_type: hist_req_data.field_type, interval: hist_req_data.req.interval, offset: hist_req_data.offset, @@ -154,7 +166,7 @@ impl BucketResolver for ComputedBucketResolver { #[inline] fn prepare_block(&mut self, docs: &[crate::DocId]) { self.hist_block - .fetch_block_with_is_full(docs, &self.accessor, self.is_full); + .fetch_full_column_block(docs, &self.accessor); } #[inline] @@ -213,7 +225,7 @@ impl BucketResolver for ComputedBucketResolver { /// `u64` space, then scanned linearly. `NUM_BUCKETS` is fixed so the optimizer can unroll the scan. #[derive(Debug)] struct LinearBucketResolver { - hist_block: ColumnBlockAccessor, + hist_block: ColumnBlockAccessor, next_count_lane: usize, accessor: Column, boundaries: [u64; NUM_BUCKETS], @@ -227,6 +239,10 @@ impl LinearBucketResolver { num_time_buckets: usize, ) -> Option { assert!(num_time_buckets > 1 && num_time_buckets <= NUM_BUCKETS); + assert!( + hist_req_data.accessor.get_cardinality().is_full(), + "LinearBucketResolver requires a full histogram column" + ); let max_encoded_value = hist_req_data.accessor.max_value(); // Padding must compare false for every column value. There is no such `u64` sentinel when // the column contains `u64::MAX`, so that edge case uses the computed resolver instead. @@ -266,7 +282,7 @@ impl BucketResolver for LinearBucketResolver { +struct FlattenedTermHistogramCollector { /// Per-term count of docs *outside* `hard_bounds` (still in `doc_count`, but in no bucket). /// Per-term total = this + the term's `counts` row-sum; left empty when there are no hard /// bounds (every doc is in-bounds, so there's no remainder to track). @@ -363,17 +379,14 @@ struct SegmentTermHistogramCollector { hist_req_data: HistogramAggReqData, /// Private term block accessor. The bucket resolver owns a histogram block accessor when it /// needs one; the single-bucket resolver deliberately does not. - term_block: ColumnBlockAccessor, + term_block: ColumnBlockAccessor, bucket_resolver: R, /// No hard bounds, so every doc is in-bounds. all_docs_in_bounds: bool, - /// The term column is full (a fused-path precondition); cached so `collect` skips the - /// per-block cardinality lookup in `fetch_block`. - term_is_full: bool, } impl SegmentAggregationCollector - for SegmentTermHistogramCollector + for FlattenedTermHistogramCollector { fn add_intermediate_aggregation_result( &mut self, @@ -383,7 +396,7 @@ impl SegmentAggregationCollector ) -> crate::Result<()> { debug_assert_eq!( parent_bucket_id, 0, - "fused term-histogram collector is top-level only" + "flattened term-histogram collector is top-level only" ); // Expand the flat grid back into the regular structures and reuse the shared builders, so // ordering/cut-off/dict handling and cross-segment merging match the general path exactly. @@ -436,16 +449,13 @@ impl SegmentAggregationCollector ) -> crate::Result<()> { debug_assert_eq!( parent_bucket_id, 0, - "fused term-histogram collector is top-level only" + "flattened term-histogram collector is top-level only" ); // The term column is always needed. The resolver fetches the histogram column only when // bucket selection depends on its values; `SingleBucketResolver` makes this a no-op. - self.term_block.fetch_block_with_is_full( - docs, - &self.terms_req_data.accessor, - self.term_is_full, - ); + self.term_block + .fetch_full_column_block(docs, &self.terms_req_data.accessor); self.bucket_resolver.prepare_block(docs); // Keep separate bounded and unbounded entry points so the common path has no bounds branch, @@ -489,13 +499,13 @@ impl SegmentAggregationCollector } } -/// Builds the fused terms×histogram collector for a single top-level parent, when the shape is +/// Builds the flattened terms×histogram collector for a single top-level parent, when the shape is /// eligible. Returns `Ok(None)` to fall back to the general buffered terms path. /// /// Eligibility: top-level, low-cardinality terms over a full column with no missing/include-exclude /// handling; a single `histogram`/`date_histogram` leaf (no nesting below it) over a full column; -/// and a physical counter grid no larger than [`MAX_FUSED_GRID_COUNTERS`]. -pub(super) fn maybe_build_collector( +/// and a physical counter grid no larger than [`MAX_FLATTENED_GRID_COUNTERS`]. +pub(super) fn maybe_build_flattened_collector( agg_data: &mut AggregationsSegmentCtx, node: &AggRefNode, terms_req_data: &TermsAggReqData, @@ -507,7 +517,7 @@ pub(super) fn maybe_build_collector( // no-op (`fetch_block_with_missing` early-returns on full columns), so we needn't check for it. // // We don't cap the term cardinality here: the flat grid is bounded by the total physical - // counter count (`num_terms * num_time_buckets * LANES <= MAX_FUSED_GRID_COUNTERS`) checked + // counter count (`num_terms * num_time_buckets * LANES <= MAX_FLATTENED_GRID_COUNTERS`) checked // below, which subsumes it. // // We only allow this at the top-level, since we don't know how many buckets are created. We @@ -536,10 +546,10 @@ pub(super) fn maybe_build_collector( return Ok(None); } - // Clone + normalize the histogram request and get its dense bucket range; only take the fused - // path when the physical counter grid is small enough. Very small logical grids use multiple - // counters per cell; larger grids retain scalar cells to avoid paying for lanes when writes are - // already spread across many locations. + // Clone + normalize the histogram request and get its dense bucket range; only take the + // flattened path when the physical counter grid is small enough. Very small logical grids use + // multiple counters per cell; larger grids retain scalar cells to avoid paying for lanes when + // writes are already spread across many locations. let Some((hist_req_data, range)) = prepare_histogram_dense_range(agg_data, &node.children[0])? else { return Ok(None); @@ -552,12 +562,12 @@ pub(super) fn maybe_build_collector( } else { SINGLE_COUNT_LANE }; - if num_grid_cells.saturating_mul(num_count_lanes) > MAX_FUSED_GRID_COUNTERS { + if num_grid_cells.saturating_mul(num_count_lanes) > MAX_FLATTENED_GRID_COUNTERS { return Ok(None); } let collector = if use_count_lanes { - build_collector::( + build_flattened_collector::( agg_data, terms_req_data, hist_req_data, @@ -566,7 +576,7 @@ pub(super) fn maybe_build_collector( range.base_pos, )? } else { - build_collector::( + build_flattened_collector::( agg_data, terms_req_data, hist_req_data, @@ -578,7 +588,7 @@ pub(super) fn maybe_build_collector( Ok(Some(collector)) } -fn build_collector( +fn build_flattened_collector( agg_data: &mut AggregationsSegmentCtx, terms_req_data: &TermsAggReqData, hist_req_data: HistogramAggReqData, @@ -586,18 +596,19 @@ fn build_collector( num_time_buckets: usize, base_pos: i64, ) -> crate::Result> { - const { assert!(LANES > 0, "a fused grid needs at least one count lane") }; + const { assert!(LANES > 0, "a flattened grid needs at least one count lane") }; let all_docs_in_bounds = hist_req_data.bounds.min == f64::MIN && hist_req_data.bounds.max == f64::MAX; if all_docs_in_bounds && num_time_buckets == 1 { - return build_collector_with_resolver::( + let resolver = SingleBucketResolver::new(&hist_req_data); + return build_flattened_collector_with_resolver::( agg_data, terms_req_data, hist_req_data, num_terms, base_pos, - SingleBucketResolver::default(), + resolver, ); } if all_docs_in_bounds && num_time_buckets <= NUM_SMALL_LINEAR_BUCKETS { @@ -606,7 +617,7 @@ fn build_collector( base_pos, num_time_buckets, ) { - return build_collector_with_resolver::<_, LANES>( + return build_flattened_collector_with_resolver::<_, LANES>( agg_data, terms_req_data, hist_req_data, @@ -621,7 +632,7 @@ fn build_collector( base_pos, num_time_buckets, ) { - return build_collector_with_resolver::<_, LANES>( + return build_flattened_collector_with_resolver::<_, LANES>( agg_data, terms_req_data, hist_req_data, @@ -633,7 +644,7 @@ fn build_collector( } let resolver = ComputedBucketResolver::new(&hist_req_data, base_pos, num_time_buckets); - build_collector_with_resolver::<_, LANES>( + build_flattened_collector_with_resolver::<_, LANES>( agg_data, terms_req_data, hist_req_data, @@ -643,7 +654,7 @@ fn build_collector( ) } -fn build_collector_with_resolver( +fn build_flattened_collector_with_resolver( agg_data: &mut AggregationsSegmentCtx, terms_req_data: &TermsAggReqData, hist_req_data: HistogramAggReqData, @@ -670,7 +681,7 @@ fn build_collector_with_resolver( .limits .add_memory_consumed(memory_consumption as u64)?; - Ok(Box::new(SegmentTermHistogramCollector:: { + Ok(Box::new(FlattenedTermHistogramCollector:: { term_counts, counts, base_pos, @@ -679,7 +690,6 @@ fn build_collector_with_resolver( term_block: ColumnBlockAccessor::default(), bucket_resolver, all_docs_in_bounds, - term_is_full: terms_req_data.accessor.get_cardinality().is_full(), })) } @@ -692,17 +702,17 @@ mod tests { }; use crate::aggregation::AggregationLimitsGuard; - /// Hand-computed correctness check for the fused terms×histogram fast path - /// ([`super::SegmentTermHistogramCollector`]): low-cardinality terms × a histogram leaf over + /// Hand-computed correctness check for the flattened terms×histogram fast path + /// ([`super::FlattenedTermHistogramCollector`]): low-cardinality terms × a histogram leaf over /// full columns, exercised single- and multi-segment. #[test] - fn fused_term_histogram_test() -> crate::Result<()> { - fused_term_histogram_with_opt(false)?; - fused_term_histogram_with_opt(true)?; + fn flattened_term_histogram_test() -> crate::Result<()> { + flattened_term_histogram_with_opt(false)?; + flattened_term_histogram_with_opt(true)?; Ok(()) } - fn fused_term_histogram_with_opt(merge_segments: bool) -> crate::Result<()> { + fn flattened_term_histogram_with_opt(merge_segments: bool) -> crate::Result<()> { // 300 docs: term = {a, b, c} by i % 3, histogram value = i % 20 (interval 1 => buckets // 0..19). gcd(3, 20) = 1, so every (term, bucket) pair occurs exactly 300 / 60 = 5 times. let docs: Vec<(f64, String)> = (0..300u64) @@ -713,7 +723,8 @@ mod tests { ) }) .collect(); - // Two segments, to also exercise cross-segment merging of the fused per-term histograms. + // Two segments, to also exercise cross-segment merging of the flattened per-term + // histograms. let segments = vec![docs[..150].to_vec(), docs[150..].to_vec()]; let index = get_test_index_from_values_and_terms(merge_segments, &segments)?; @@ -747,7 +758,7 @@ mod tests { /// A histogram whose values all map to one bucket uses the resolver that counts terms without /// reading the histogram column. #[test] - fn fused_term_histogram_single_bucket_resolver() -> crate::Result<()> { + fn flattened_term_histogram_single_bucket_resolver() -> crate::Result<()> { // The ten distinct values all map to bucket 0 at interval 10. With three terms coprime to // the value count, each term occurs 30 times across the 90 documents. let docs: Vec<(f64, String)> = (0..90usize) @@ -781,7 +792,7 @@ mod tests { /// Four and eight histogram buckets take their corresponding fixed-size linear resolvers. /// Negative values also exercise monotonic `f64` fast-field boundaries on both sides of zero. #[test] - fn fused_term_histogram_linear_bucket_resolver() -> crate::Result<()> { + fn flattened_term_histogram_linear_bucket_resolver() -> crate::Result<()> { for num_buckets in [4usize, 8] { // Three terms are coprime with both bucket counts, so every pair occurs 10 times. let docs: Vec<(f64, String)> = (0..3 * num_buckets * 10) @@ -823,11 +834,11 @@ mod tests { Ok(()) } - /// A `missing` config on a *full* term column still takes the fused path (the string sentinel - /// is just `col_max + 1`, so the column stays low-cardinality). Since no doc is missing, the - /// real term buckets must be exactly as without `missing`. + /// A `missing` config on a *full* term column still takes the flattened path (the string + /// sentinel is just `col_max + 1`, so the column stays low-cardinality). Since no doc is + /// missing, the real term buckets must be exactly as without `missing`. #[test] - fn fused_term_histogram_with_missing_on_full_column() -> crate::Result<()> { + fn flattened_term_histogram_with_missing_on_full_column() -> crate::Result<()> { let docs: Vec<(f64, String)> = (0..300u64) .map(|i| { ( @@ -867,7 +878,7 @@ mod tests { /// Term cardinality is not what gates fusing: the flat grid is bounded by the total cell count /// (`num_terms * num_time_buckets`), not the term count, so many terms still fuse. #[test] - fn fused_term_histogram_many_terms() -> crate::Result<()> { + fn flattened_term_histogram_many_terms() -> crate::Result<()> { let num_terms = 150usize; let docs_per_term = 2usize; // All docs share histogram value 0 (a single bucket), so the grid is 150 x 1 = 150 cells. @@ -909,7 +920,7 @@ mod tests { /// is the case where the per-doc `term_counts` increment cannot be replaced by the grid /// row-sum. #[test] - fn fused_term_histogram_with_hard_bounds() -> crate::Result<()> { + fn flattened_term_histogram_with_hard_bounds() -> crate::Result<()> { // 300 docs: term = {a, b, c} by i % 3, value = i % 20. Per term: 100 docs, each value in // 0..=19 occurring 5 times. let docs: Vec<(f64, String)> = (0..300u64) @@ -965,7 +976,7 @@ mod tests { /// histogram row-sum. This is the case that previously fell back to the per-doc counter only /// because `bounds != [MIN, MAX]`. #[test] - fn fused_term_histogram_with_non_binding_hard_bounds() -> crate::Result<()> { + fn flattened_term_histogram_with_non_binding_hard_bounds() -> crate::Result<()> { // 300 docs: term = {a, b, c} by i % 3, value = i % 20. Data values span [0, 19]. let docs: Vec<(f64, String)> = (0..300u64) .map(|i| { @@ -1011,12 +1022,12 @@ mod tests { Ok(()) } - /// Regression: with hard bounds the fused path allocates `term_counts` (one `u32`/term) on top - /// of the grid, and that allocation must be charged to the memory limit. With many terms and a - /// single time bucket the two are equal in size, so a limit admitting the grid alone but not - /// grid + `term_counts` must fail. + /// Regression: with hard bounds the flattened path allocates `term_counts` (one `u32`/term) on + /// top of the grid, and that allocation must be charged to the memory limit. With many terms + /// and a single time bucket the two are equal in size, so a limit admitting the grid alone but + /// not grid + `term_counts` must fail. #[test] - fn fused_term_histogram_hard_bounds_charges_term_counts() -> crate::Result<()> { + fn flattened_term_histogram_hard_bounds_charges_term_counts() -> crate::Result<()> { // 16k distinct terms, one doc each; values alternate in/out of the single-bucket bounds // [5, 5] so the bounds bind and `term_counts` is allocated. num_terms=16000, // num_time_buckets=1 => `counts` and `term_counts` are ~64 KB each. diff --git a/src/aggregation/bucket/term_agg/mod.rs b/src/aggregation/bucket/term_agg/mod.rs index 510c35a84..e9be561ab 100644 --- a/src/aggregation/bucket/term_agg/mod.rs +++ b/src/aggregation/bucket/term_agg/mod.rs @@ -30,7 +30,7 @@ use crate::aggregation::{format_date, BucketId, Key}; use crate::error::DataCorruption; use crate::TantivyError; -mod term_histogram; +mod flattened_term_histogram; /// Contains all information required by the SegmentTermCollector to perform the /// terms aggregation on a segment. @@ -427,9 +427,10 @@ pub(crate) fn build_segment_term_collector( let max_column_val: u64 = col_max_value.max(terms_req_data.missing_value_for_accessor.unwrap_or(0u64)); - // Fused fast path: low-cardinality terms × a single `histogram`/`date_histogram` leaf over full - // columns with a small enough bucket grid. Anything else falls through to the general path. - if let Some(collector) = term_histogram::maybe_build_collector( + // Flattened fast path: low-cardinality terms × a single `histogram`/`date_histogram` leaf over + // full columns with a small enough bucket grid. Anything else falls through to the general + // path. + if let Some(collector) = flattened_term_histogram::maybe_build_flattened_collector( req_data, node, &terms_req_data, @@ -1070,9 +1071,7 @@ impl SegmentAggregationCollector if let Some(sub_agg) = &mut self.sub_agg { let term_buckets = &mut self.parent_buckets[parent_bucket_id as usize]; - let it = agg_data - .column_block_accessor - .iter_docid_vals(docs, &req_data.accessor); + let it = agg_data.column_block_accessor.iter_docid_vals(docs); if let Some(allowed_bs) = req_data.allowed_term_ids.as_ref() { let it = it.filter(move |&(_doc, term_id)| allowed_bs.contains(term_id as u32)); Self::collect_terms_with_docs( @@ -1896,7 +1895,8 @@ mod tests { res["my_texts"]["buckets"][1]["key"], serde_json::Value::Null ); - assert_eq!(res["my_texts"]["sum_other_doc_count"], 0); // TODO sum_other_doc_count with min_doc_count + assert_eq!(res["my_texts"]["sum_other_doc_count"], 0); // TODO sum_other_doc_count with + // min_doc_count Ok(()) } diff --git a/src/aggregation/metric/stats.rs b/src/aggregation/metric/stats.rs index eb15427a1..06a39f6d1 100644 --- a/src/aggregation/metric/stats.rs +++ b/src/aggregation/metric/stats.rs @@ -266,7 +266,7 @@ impl SegmentAggregationCollector return Err(TantivyError::InvalidArgument(format!( "Unsupported stats type for stats aggregation: {:?}", self.collecting_for - ))) + ))); } }; @@ -285,6 +285,11 @@ impl SegmentAggregationCollector docs: &[crate::DocId], agg_data: &mut AggregationsSegmentCtx, ) -> crate::Result<()> { + // Fast path: the caller hands us a single doc, which is the dominant case when a metric + // sub-agg sits under a high-cardinality bucket agg. Streaming straight from the column + // skips the block accessor's buffers entirely. + // Only valid without a missing value: `values_for_doc` yields nothing for a doc without a + // value, so the substitute would be silently dropped. // TODO: remove once we fetch all values for all bucket ids in one go if docs.len() == 1 && self.missing_u64.is_none() { collect_stats::( diff --git a/src/aggregation/mod.rs b/src/aggregation/mod.rs index a36bbc2c2..4c1820189 100644 --- a/src/aggregation/mod.rs +++ b/src/aggregation/mod.rs @@ -132,6 +132,7 @@ mod agg_data; mod agg_limits; pub mod agg_req; pub mod agg_result; +mod block_accessor; pub mod bucket; pub(crate) mod buffered_sub_aggs; mod collector; @@ -143,6 +144,8 @@ pub mod metric; mod segment_agg_result; use std::fmt::Display; +pub(crate) use block_accessor::ColumnBlockAccessor; + #[cfg(test)] mod agg_tests; diff --git a/src/store/reader.rs b/src/store/reader.rs index a4105abec..82c1fc303 100644 --- a/src/store/reader.rs +++ b/src/store/reader.rs @@ -304,7 +304,9 @@ impl StoreReader { let mut curr_checkpoint = checkpoint_block_iter.next(); let mut curr_block = curr_checkpoint .as_ref() - .map(|checkpoint| self.read_block(checkpoint).map_err(|e| e.kind())); // map error in order to enable cloning + .map(|checkpoint| self.read_block(checkpoint).map_err(|e| e.kind())); // map error in + // order to enable + // cloning let mut doc_pos = 0; (0..last_doc_id) .filter_map(move |doc_id| {