mirror of
https://github.com/quickwit-oss/tantivy.git
synced 2026-10-06 03:42:36 +00:00
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 <paul.masurel@datadoghq.com>
This commit is contained in:
co-authored by
Paul Masurel
parent
35a6187ed1
commit
21ae09c0b4
@@ -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": {
|
||||
|
||||
@@ -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::{
|
||||
|
||||
@@ -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<u64>,
|
||||
pub(crate) column_block_accessor: ColumnBlockAccessor,
|
||||
}
|
||||
|
||||
impl AggregationsSegmentCtx {
|
||||
|
||||
@@ -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<T> {
|
||||
val_cache: Vec<T>,
|
||||
docid_cache: Vec<DocId>,
|
||||
missing_docids_cache: Vec<DocId>,
|
||||
row_id_cache: Vec<RowId>,
|
||||
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<u64>,
|
||||
docids: &mut Vec<DocId>,
|
||||
row_ids: &mut Vec<RowId>,
|
||||
) -> Cardinality;
|
||||
}
|
||||
|
||||
impl<T: PartialOrd + Copy + std::fmt::Debug + Send + Sync + 'static + Default>
|
||||
ColumnBlockAccessor<T>
|
||||
{
|
||||
/// 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<u64>,
|
||||
/// 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<DocId>,
|
||||
/// Scratch buffer used to identify documents for which a missing value must be inserted.
|
||||
missing_docids_cache: Vec<DocId>,
|
||||
/// Scratch buffer available to sources for translating document IDs into value row IDs.
|
||||
row_id_cache: Vec<RowId>,
|
||||
/// 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<u64> {
|
||||
#[inline]
|
||||
pub fn fetch_block<'a>(&'a mut self, docs: &'a [u32], accessor: &Column<T>) {
|
||||
self.fetch_block_with_is_full(docs, accessor, accessor.index.get_cardinality().is_full());
|
||||
fn load_block(
|
||||
&self,
|
||||
docs: &[DocId],
|
||||
values: &mut Vec<u64>,
|
||||
docids: &mut Vec<DocId>,
|
||||
row_ids: &mut Vec<RowId>,
|
||||
) -> 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<T>,
|
||||
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<u64>) {
|
||||
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<T>,
|
||||
missing_opt: Option<T>,
|
||||
docs: &[DocId],
|
||||
source: &impl BlockValueSource,
|
||||
missing_opt: Option<u64>,
|
||||
) {
|
||||
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<T>,
|
||||
missing_opt: Option<T>,
|
||||
docs: &[DocId],
|
||||
source: &impl BlockValueSource,
|
||||
missing_opt: Option<u64>,
|
||||
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<T: PartialOrd + Copy + std::fmt::Debug + Send + Sync + 'static + Default>
|
||||
/// 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<T>,
|
||||
missing: Option<T>,
|
||||
docs: &[DocId],
|
||||
source: &impl BlockValueSource,
|
||||
missing: Option<u64>,
|
||||
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<T: PartialOrd + Copy + std::fmt::Debug + Send + Sync + 'static + Default>
|
||||
/// 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<T: PartialOrd + Copy + std::fmt::Debug + Send + Sync + 'static + Default>
|
||||
|
||||
/// 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<Item = T> + '_ {
|
||||
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<Item = u64> + '_ {
|
||||
self.val_cache.iter().cloned()
|
||||
}
|
||||
|
||||
@@ -231,14 +271,13 @@ impl<T: PartialOrd + Copy + std::fmt::Debug + Send + Sync + 'static + Default>
|
||||
/// 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<T>,
|
||||
) -> impl Iterator<Item = (DocId, T)> + 'a + use<'a, T> {
|
||||
if accessor.index.get_cardinality().is_full() {
|
||||
docs: &'a [DocId],
|
||||
) -> impl Iterator<Item = (DocId, u64)> + 'a {
|
||||
if self.cardinality.is_full() {
|
||||
docs.iter().cloned().zip(self.val_cache.iter().cloned())
|
||||
} else {
|
||||
self.docid_cache
|
||||
@@ -249,6 +288,21 @@ impl<T: PartialOrd + Copy + std::fmt::Debug + Send + Sync + 'static + Default>
|
||||
}
|
||||
}
|
||||
|
||||
#[inline]
|
||||
fn load_full_column_values(docs: &[DocId], accessor: &Column<u64>, values: &mut Vec<u64>) {
|
||||
// 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<u32>) {
|
||||
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<u64>,
|
||||
docids: &mut Vec<DocId>,
|
||||
row_ids: &mut Vec<RowId>,
|
||||
) -> 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<u32> = 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<_>>(),
|
||||
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<_>>(),
|
||||
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<_>>(),
|
||||
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::<u64>::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::<Vec<_>>(),
|
||||
accessor.iter_docid_vals(&docs).collect::<Vec<_>>(),
|
||||
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::<u64>::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::<u64>::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::<u64>::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::<u64>::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::<u64>::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<u64> = (0..200u64).map(|i| i * 7 + 3).collect();
|
||||
@@ -443,14 +578,14 @@ mod tests {
|
||||
values,
|
||||
};
|
||||
|
||||
let check = |accessor: &mut ColumnBlockAccessor<u64>, 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::<u64>::default();
|
||||
let mut accessor = ColumnBlockAccessor::default();
|
||||
// Contiguous block -> get_range fast path.
|
||||
check(&mut accessor, &(10..74).collect::<Vec<u32>>());
|
||||
// Non-contiguous block -> get_vals gather path.
|
||||
@@ -494,10 +494,7 @@ impl<B: BucketIdSlot> 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<B: BucketIdSlot> SegmentHistogramCollector<B> {
|
||||
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<const LANES: usize>(
|
||||
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)?;
|
||||
|
||||
@@ -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<u64>,
|
||||
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: MultiTermsPacking>(
|
||||
packing: &Packing,
|
||||
field_idx: usize,
|
||||
field: &MultiTermsFieldAccessor,
|
||||
missing: Option<&MultiTermsMissingAccessor>,
|
||||
block_accessor: &ColumnBlockAccessor<u64>,
|
||||
block_accessor: &ColumnBlockAccessor,
|
||||
keys_buf: &mut Vec<Packing::PackingType>,
|
||||
alive_docs: &mut Vec<DocId>,
|
||||
doc_ids_per_partial_combination: &mut Vec<DocId>,
|
||||
@@ -483,9 +479,7 @@ fn expand_partial_combinations_for_field<Packing: MultiTermsPacking>(
|
||||
);
|
||||
|
||||
{
|
||||
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<u64>,
|
||||
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<u64>,
|
||||
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,
|
||||
|
||||
@@ -285,10 +285,7 @@ impl<B: SubAggBuffer> SegmentAggregationCollector for SegmentRangeCollector<B> {
|
||||
|
||||
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;
|
||||
|
||||
+84
-73
@@ -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<const LANES: usize>(
|
||||
|
||||
/// 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<u64>,
|
||||
hist_block: ColumnBlockAccessor,
|
||||
next_count_lane: usize,
|
||||
accessor: Column<u64>,
|
||||
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<const NUM_BUCKETS: usize> {
|
||||
hist_block: ColumnBlockAccessor<u64>,
|
||||
hist_block: ColumnBlockAccessor,
|
||||
next_count_lane: usize,
|
||||
accessor: Column<u64>,
|
||||
boundaries: [u64; NUM_BUCKETS],
|
||||
@@ -227,6 +239,10 @@ impl<const NUM_BUCKETS: usize> LinearBucketResolver<NUM_BUCKETS> {
|
||||
num_time_buckets: usize,
|
||||
) -> Option<Self> {
|
||||
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<const NUM_BUCKETS: usize> BucketResolver for LinearBucketResolver<NUM_BUCKE
|
||||
#[inline]
|
||||
fn prepare_block(&mut self, docs: &[crate::DocId]) {
|
||||
self.hist_block
|
||||
.fetch_block_with_is_full(docs, &self.accessor, true);
|
||||
.fetch_full_column_block(docs, &self.accessor);
|
||||
}
|
||||
|
||||
#[inline]
|
||||
@@ -328,9 +344,9 @@ fn first_encoded_value_for_bucket(
|
||||
encoded_lower_bound
|
||||
}
|
||||
|
||||
/// Fused collector for `terms` (low cardinality) × a single `histogram`/`date_histogram` leaf with
|
||||
/// nothing nested below it, when the resulting counter grid is small (see
|
||||
/// [`MAX_FUSED_GRID_COUNTERS`]).
|
||||
/// Flattened collector for `terms` (low cardinality) × a single `histogram`/`date_histogram` leaf
|
||||
/// with nothing nested below it, when the resulting counter grid is small (see
|
||||
/// [`MAX_FLATTENED_GRID_COUNTERS`]).
|
||||
///
|
||||
/// It keeps a flat, fully dense 2D counter grid
|
||||
/// (`counts[term * num_time_buckets + bucket][lane]`) and a per-term total. Cycling writes through
|
||||
@@ -344,7 +360,7 @@ fn first_encoded_value_for_bucket(
|
||||
/// handed to the shared intermediate-result builders, so cross-segment merging is identical to the
|
||||
/// general path.
|
||||
#[derive(Debug)]
|
||||
struct SegmentTermHistogramCollector<R: BucketResolver, const LANES: usize> {
|
||||
struct FlattenedTermHistogramCollector<R: BucketResolver, const LANES: usize> {
|
||||
/// 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<R: BucketResolver, const LANES: usize> {
|
||||
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<u64>,
|
||||
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<R: BucketResolver, const LANES: usize> SegmentAggregationCollector
|
||||
for SegmentTermHistogramCollector<R, LANES>
|
||||
for FlattenedTermHistogramCollector<R, LANES>
|
||||
{
|
||||
fn add_intermediate_aggregation_result(
|
||||
&mut self,
|
||||
@@ -383,7 +396,7 @@ impl<R: BucketResolver, const LANES: usize> 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<R: BucketResolver, const LANES: usize> 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<R: BucketResolver, const LANES: usize> 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::<NUM_LOW_CARD_COUNT_LANES>(
|
||||
build_flattened_collector::<NUM_LOW_CARD_COUNT_LANES>(
|
||||
agg_data,
|
||||
terms_req_data,
|
||||
hist_req_data,
|
||||
@@ -566,7 +576,7 @@ pub(super) fn maybe_build_collector(
|
||||
range.base_pos,
|
||||
)?
|
||||
} else {
|
||||
build_collector::<SINGLE_COUNT_LANE>(
|
||||
build_flattened_collector::<SINGLE_COUNT_LANE>(
|
||||
agg_data,
|
||||
terms_req_data,
|
||||
hist_req_data,
|
||||
@@ -578,7 +588,7 @@ pub(super) fn maybe_build_collector(
|
||||
Ok(Some(collector))
|
||||
}
|
||||
|
||||
fn build_collector<const LANES: usize>(
|
||||
fn build_flattened_collector<const LANES: usize>(
|
||||
agg_data: &mut AggregationsSegmentCtx,
|
||||
terms_req_data: &TermsAggReqData,
|
||||
hist_req_data: HistogramAggReqData,
|
||||
@@ -586,18 +596,19 @@ fn build_collector<const LANES: usize>(
|
||||
num_time_buckets: usize,
|
||||
base_pos: i64,
|
||||
) -> crate::Result<Box<dyn SegmentAggregationCollector>> {
|
||||
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::<SingleBucketResolver, LANES>(
|
||||
let resolver = SingleBucketResolver::new(&hist_req_data);
|
||||
return build_flattened_collector_with_resolver::<SingleBucketResolver, LANES>(
|
||||
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<const LANES: usize>(
|
||||
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<const LANES: usize>(
|
||||
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<const LANES: usize>(
|
||||
}
|
||||
|
||||
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<const LANES: usize>(
|
||||
)
|
||||
}
|
||||
|
||||
fn build_collector_with_resolver<R: BucketResolver, const LANES: usize>(
|
||||
fn build_flattened_collector_with_resolver<R: BucketResolver, const LANES: usize>(
|
||||
agg_data: &mut AggregationsSegmentCtx,
|
||||
terms_req_data: &TermsAggReqData,
|
||||
hist_req_data: HistogramAggReqData,
|
||||
@@ -670,7 +681,7 @@ fn build_collector_with_resolver<R: BucketResolver, const LANES: usize>(
|
||||
.limits
|
||||
.add_memory_consumed(memory_consumption as u64)?;
|
||||
|
||||
Ok(Box::new(SegmentTermHistogramCollector::<R, LANES> {
|
||||
Ok(Box::new(FlattenedTermHistogramCollector::<R, LANES> {
|
||||
term_counts,
|
||||
counts,
|
||||
base_pos,
|
||||
@@ -679,7 +690,6 @@ fn build_collector_with_resolver<R: BucketResolver, const LANES: usize>(
|
||||
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.
|
||||
@@ -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<TermMap: TermAggregationMap, B: SubAggBuffer> 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(())
|
||||
}
|
||||
|
||||
|
||||
@@ -266,7 +266,7 @@ impl<const COLUMN_TYPE_ID: u8> SegmentAggregationCollector
|
||||
return Err(TantivyError::InvalidArgument(format!(
|
||||
"Unsupported stats type for stats aggregation: {:?}",
|
||||
self.collecting_for
|
||||
)))
|
||||
)));
|
||||
}
|
||||
};
|
||||
|
||||
@@ -285,6 +285,11 @@ impl<const COLUMN_TYPE_ID: u8> 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::<COLUMN_TYPE_ID>(
|
||||
|
||||
@@ -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;
|
||||
|
||||
|
||||
+3
-1
@@ -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| {
|
||||
|
||||
Reference in New Issue
Block a user