mirror of
https://github.com/quickwit-oss/tantivy.git
synced 2026-08-18 03:58:21 +00:00
fix cache flush in aggregations
fixes #2992 ``` full terms_7 Memory: 37.2 KB Avg: 2.3958ms (+0.31%) Median: 2.3896ms (+0.18%) [2.3573ms .. 2.5176ms] terms_all_unique Memory: 10.8 MB Avg: 5.5144ms (-1.07%) Median: 5.4625ms (-1.98%) [5.3364ms .. 5.9712ms] terms_all_unique_order_by_key Memory: 10.8 MB Avg: 5.2614ms (-0.85%) Median: 5.2177ms (-1.21%) [5.0823ms .. 5.6316ms] terms_150_000 Memory: 2.7 MB Avg: 5.5335ms (-1.07%) Median: 5.5152ms (-1.06%) [5.4151ms .. 5.9654ms] terms_many_top_1000 Memory: 5.2 MB Avg: 8.3579ms (-1.53%) Median: 8.3604ms (-0.95%) [8.2184ms .. 8.5421ms] terms_many_order_by_term Memory: 2.7 MB Avg: 4.6713ms (-0.07%) Median: 4.6569ms (-0.15%) [4.5994ms .. 4.9115ms] terms_all_unique_with_avg_sub_agg Memory: 54.0 MB Avg: 17.4981ms (-2.43%) Median: 17.6075ms (-1.75%) [15.8166ms .. 18.9250ms] terms_status_with_avg_sub_agg Memory: 90.3 KB Avg: 5.6365ms (+7.77%) Median: 5.6255ms (+7.97%) [5.5489ms .. 5.8254ms] terms_status_with_terms_zipf_1000_sub_agg Memory: 318.5 KB (+56.52%) Avg: 4.4504ms (+11.55%) Median: 4.4436ms (+11.59%) [4.3858ms .. 4.5692ms] terms_zipf_1000_with_terms_status_sub_agg Memory: 684.9 KB Avg: 11.8606ms (+0.19%) Median: 11.8360ms (-0.02%) [11.7478ms .. 12.0609ms] terms_status_with_histogram Memory: 139.5 KB Avg: 2.4524ms (-1.09%) Median: 2.4521ms (-0.23%) [2.4179ms .. 2.5049ms] terms_status_with_date_histogram Memory: 136.7 KB Avg: 2.3407ms (-1.28%) Median: 2.3359ms (-1.06%) [2.3001ms .. 2.4310ms] terms_status_with_date_histogram_hard_bounds Memory: 136.1 KB Avg: 2.5113ms (-2.04%) Median: 2.5073ms (-0.97%) [2.4455ms .. 2.7280ms] terms_status_with_date_histogram_and_sibling_terms Memory: 137.3 KB Avg: 3.8695ms (-0.48%) Median: 3.8653ms (+0.11%) [3.8093ms .. 4.0528ms] terms_zipf_1000 Memory: 69.8 KB Avg: 2.2022ms (-1.95%) Median: 2.2026ms (-1.12%) [2.1705ms .. 2.2859ms] terms_zipf_1000_with_histogram Memory: 1.2 MB Avg: 20.4087ms (-0.02%) Median: 20.3665ms (+0.11%) [20.1912ms .. 20.7933ms] terms_zipf_1000_with_avg_sub_agg Memory: 472.0 KB Avg: 8.7387ms (-3.48%) Median: 8.7043ms (-3.44%) [8.6466ms .. 9.1396ms] terms_zipf_90 Memory: 55.3 KB Avg: 1.3784ms (-2.04%) Median: 1.3787ms (-1.61%) [1.3484ms .. 1.4611ms] terms_zipf_90_with_sum_sub_agg Memory: 367.6 KB Avg: 4.8520ms (+8.43%) Median: 4.8326ms (+8.94%) [4.8058ms .. 5.1278ms] terms_many_json_mixed_type_with_avg_sub_agg Memory: 17.8 MB Avg: 25.0853ms (-7.70%) Median: 25.0591ms (-7.12%) [24.8103ms .. 25.4936ms] terms_status_with_cardinality_agg Memory: 91.8 KB Avg: 3.3667ms (+1.47%) Median: 3.3690ms (+1.66%) [3.3311ms .. 3.4070ms] terms_100_buckets_with_cardinality_agg Memory: 9.9 MB Avg: 48.4768ms (-3.07%) Median: 48.3745ms (-3.38%) [48.1425ms .. 49.5503ms] ```
This commit is contained in:
Pascal Seitz
PSeitz
@@ -73,6 +73,8 @@ fn bench_agg(mut group: InputGroup<Index>) {
|
||||
register!(group, terms_zipf_1000);
|
||||
register!(group, terms_zipf_1000_with_histogram);
|
||||
register!(group, terms_zipf_1000_with_avg_sub_agg);
|
||||
register!(group, terms_zipf_90);
|
||||
register!(group, terms_zipf_90_with_sum_sub_agg);
|
||||
|
||||
register!(group, terms_many_json_mixed_type_with_avg_sub_agg);
|
||||
|
||||
@@ -495,6 +497,27 @@ fn terms_zipf_1000(index: &Index) {
|
||||
execute_agg(index, agg_req);
|
||||
}
|
||||
|
||||
fn terms_zipf_90(index: &Index) {
|
||||
let agg_req = json!({
|
||||
"my_texts": { "terms": { "field": "text_90_terms_zipf", "size": 100 } },
|
||||
});
|
||||
execute_agg(index, agg_req);
|
||||
}
|
||||
|
||||
// 90-term (low-cardinality Vec path) terms agg with a metric sub-agg. The skewed distribution keeps
|
||||
// a dominant bucket crossing the sub-agg flush threshold, exercising the buffer flush path.
|
||||
fn terms_zipf_90_with_sum_sub_agg(index: &Index) {
|
||||
let agg_req = json!({
|
||||
"my_texts": {
|
||||
"terms": { "field": "text_90_terms_zipf", "size": 100 },
|
||||
"aggs": {
|
||||
"sum_score": { "sum": { "field": "score" } }
|
||||
}
|
||||
},
|
||||
});
|
||||
execute_agg(index, agg_req);
|
||||
}
|
||||
|
||||
fn terms_many_json_mixed_type_with_avg_sub_agg(index: &Index) {
|
||||
let agg_req = json!({
|
||||
"my_texts": {
|
||||
@@ -762,6 +785,8 @@ fn get_test_index_bench(cardinality: Cardinality) -> tantivy::Result<Index> {
|
||||
let text_field_few_terms = schema_builder.add_text_field("text_few_terms", STRING | FAST);
|
||||
let text_field_few_terms_status =
|
||||
schema_builder.add_text_field("text_few_terms_status", STRING | FAST);
|
||||
let text_field_90_terms_zipf =
|
||||
schema_builder.add_text_field("text_90_terms_zipf", STRING | FAST);
|
||||
let text_field_1000_terms_zipf =
|
||||
schema_builder.add_text_field("text_1000_terms_zipf", STRING | FAST);
|
||||
let score_fieldtype = tantivy::schema::NumericOptions::default().set_fast();
|
||||
@@ -800,6 +825,12 @@ fn get_test_index_bench(cardinality: Cardinality) -> tantivy::Result<Index> {
|
||||
let terms_1000: Vec<String> = (1..=1000).map(|i| format!("term_{i}")).collect();
|
||||
let zipf_1000 = rand_distr::Zipf::new(1000.0, 1.1f64).unwrap();
|
||||
|
||||
// 90 terms (< MAX_NUM_TERMS_FOR_VEC), skewed via Zipf so a dominant bucket keeps crossing the
|
||||
// sub-agg flush threshold while minority buckets stay small. Exercises the low-cardinality
|
||||
// sub-agg buffer flush path (see issue #2992).
|
||||
let terms_90: Vec<String> = (1..=90).map(|i| format!("term_{i}")).collect();
|
||||
let zipf_90 = rand_distr::Zipf::new(90.0, 1.1f64).unwrap();
|
||||
|
||||
{
|
||||
let mut rng = StdRng::from_seed([1u8; 32]);
|
||||
let mut index_writer = index.writer_with_num_threads(1, 200_000_000)?;
|
||||
@@ -815,6 +846,8 @@ fn get_test_index_bench(cardinality: Cardinality) -> tantivy::Result<Index> {
|
||||
let idx_b = zipf_1000.sample(&mut rng) as usize - 1;
|
||||
let term_1000_a = &terms_1000[idx_a];
|
||||
let term_1000_b = &terms_1000[idx_b];
|
||||
let term_90_a = &terms_90[zipf_90.sample(&mut rng) as usize - 1];
|
||||
let term_90_b = &terms_90[zipf_90.sample(&mut rng) as usize - 1];
|
||||
index_writer.add_document(doc!(
|
||||
json_field => json!({"mixed_type": 10.0}),
|
||||
json_field => json!({"mixed_type": 10.0}),
|
||||
@@ -830,6 +863,8 @@ fn get_test_index_bench(cardinality: Cardinality) -> tantivy::Result<Index> {
|
||||
text_field_few_terms => "cool",
|
||||
text_field_few_terms_status => log_level_sample_a,
|
||||
text_field_few_terms_status => log_level_sample_b,
|
||||
text_field_90_terms_zipf => term_90_a.as_str(),
|
||||
text_field_90_terms_zipf => term_90_b.as_str(),
|
||||
text_field_1000_terms_zipf => term_1000_a.as_str(),
|
||||
text_field_1000_terms_zipf => term_1000_b.as_str(),
|
||||
score_field => 1u64,
|
||||
@@ -866,6 +901,7 @@ fn get_test_index_bench(cardinality: Cardinality) -> tantivy::Result<Index> {
|
||||
text_field_many_terms => many_terms_data.choose(&mut rng).unwrap().to_string(),
|
||||
text_field_few_terms => few_terms_data.choose(&mut rng).unwrap().to_string(),
|
||||
text_field_few_terms_status => status_field_data[log_level_distribution.sample(&mut rng)].0,
|
||||
text_field_90_terms_zipf => terms_90[zipf_90.sample(&mut rng) as usize - 1].as_str(),
|
||||
text_field_1000_terms_zipf => terms_1000[zipf_1000.sample(&mut rng) as usize - 1].as_str(),
|
||||
score_field => val as u64,
|
||||
score_field_f64 => lg_norm.sample(&mut rng),
|
||||
|
||||
@@ -664,6 +664,82 @@ fn test_aggregation_flushing_variants() {
|
||||
test_aggregation_flushing(true, true).unwrap();
|
||||
}
|
||||
|
||||
// Regression test for https://github.com/quickwit-oss/tantivy/issues/2992
|
||||
//
|
||||
// A skewed terms bucket over <100 terms uses the low-cardinality (Vec) sub-agg buffer. A dominant
|
||||
// term keeps crossing the periodic flush threshold (every 2048 docs), which used to drop the
|
||||
// cached doc ids of the minority buckets before collecting them — corrupting their metric
|
||||
// sub-aggregations while their doc counts stayed exact.
|
||||
#[test]
|
||||
fn test_terms_sub_agg_flushing_skewed_buckets() -> crate::Result<()> {
|
||||
use std::collections::HashMap;
|
||||
|
||||
// 89 minority terms + 1 dominant term = 90 distinct terms, staying below
|
||||
// MAX_NUM_TERMS_FOR_VEC (100) so the low-cardinality Vec sub-agg buffer is used.
|
||||
const NUM_MINORITY_TERMS: usize = 89;
|
||||
|
||||
let mut values: Vec<(f64, String)> = Vec::new();
|
||||
let mut minority_idx = 0usize;
|
||||
// Enough docs to cross the 2048 flush threshold multiple times. The minority docs are
|
||||
// sprinkled among the dominant ones so they land in different flush windows.
|
||||
for i in 0..5000u64 {
|
||||
if i % 25 == 0 {
|
||||
let term = format!("minority_{:02}", minority_idx % NUM_MINORITY_TERMS);
|
||||
minority_idx += 1;
|
||||
values.push(((i % 13 + 1) as f64, term));
|
||||
} else {
|
||||
values.push((7.0, "dominant".to_string()));
|
||||
}
|
||||
}
|
||||
|
||||
let mut truth: HashMap<String, (u64, f64)> = HashMap::new();
|
||||
for (score, term) in &values {
|
||||
let entry = truth.entry(term.clone()).or_insert((0, 0.0));
|
||||
entry.0 += 1;
|
||||
entry.1 += *score;
|
||||
}
|
||||
// Sanity check on the shape of the generated data.
|
||||
assert_eq!(truth.len(), NUM_MINORITY_TERMS + 1);
|
||||
|
||||
let index = get_test_index_from_values_and_terms(false, &[values])?;
|
||||
let reader = index.reader()?;
|
||||
|
||||
let agg_req: Aggregations = serde_json::from_value(json!({
|
||||
"my_terms": {
|
||||
"terms": { "field": "string_id", "size": 100 },
|
||||
"aggs": {
|
||||
"sum_score": { "sum": { "field": "score" } }
|
||||
}
|
||||
}
|
||||
}))
|
||||
.unwrap();
|
||||
|
||||
let collector = get_collector(agg_req);
|
||||
let searcher = reader.searcher();
|
||||
let agg_res = searcher.search(&AllQuery, &collector)?;
|
||||
let res: Value = serde_json::from_str(&serde_json::to_string(&agg_res)?)?;
|
||||
|
||||
let buckets = res["my_terms"]["buckets"].as_array().unwrap();
|
||||
// size 100 >= 90 distinct terms, so every bucket is returned.
|
||||
assert_eq!(buckets.len(), truth.len());
|
||||
for bucket in buckets {
|
||||
let key = bucket["key"].as_str().unwrap();
|
||||
let (true_count, true_sum) = truth[key];
|
||||
assert_eq!(
|
||||
bucket["doc_count"].as_u64().unwrap(),
|
||||
true_count,
|
||||
"doc_count mismatch for {key}"
|
||||
);
|
||||
assert_eq!(
|
||||
bucket["sum_score"]["value"].as_f64().unwrap(),
|
||||
true_sum,
|
||||
"sum sub-agg mismatch for {key}"
|
||||
);
|
||||
}
|
||||
|
||||
Ok(())
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_aggregation_level1_simple() -> crate::Result<()> {
|
||||
let index = get_test_index_2_segments(true)?;
|
||||
|
||||
@@ -2,7 +2,6 @@ use std::fmt::Debug;
|
||||
|
||||
use super::segment_agg_result::SegmentAggregationCollector;
|
||||
use crate::aggregation::agg_data::AggregationsSegmentCtx;
|
||||
use crate::aggregation::bucket::MAX_NUM_TERMS_FOR_VEC;
|
||||
use crate::aggregation::BucketId;
|
||||
use crate::DocId;
|
||||
|
||||
@@ -45,7 +44,6 @@ pub trait SubAggBuffer: Debug {
|
||||
&mut self,
|
||||
sub_agg: &mut Box<dyn SegmentAggregationCollector>,
|
||||
agg_data: &mut AggregationsSegmentCtx,
|
||||
force: bool,
|
||||
) -> crate::Result<()>;
|
||||
}
|
||||
|
||||
@@ -76,7 +74,7 @@ impl<Backend: SubAggBuffer + Debug> BufferedSubAggs<Backend> {
|
||||
) -> crate::Result<()> {
|
||||
if self.num_docs >= FLUSH_THRESHOLD {
|
||||
self.buffer
|
||||
.flush_local(&mut self.sub_agg_collector, agg_data, false)?;
|
||||
.flush_local(&mut self.sub_agg_collector, agg_data)?;
|
||||
self.num_docs = 0;
|
||||
}
|
||||
Ok(())
|
||||
@@ -86,7 +84,7 @@ impl<Backend: SubAggBuffer + Debug> BufferedSubAggs<Backend> {
|
||||
pub fn flush(&mut self, agg_data: &mut AggregationsSegmentCtx) -> crate::Result<()> {
|
||||
if self.num_docs != 0 {
|
||||
self.buffer
|
||||
.flush_local(&mut self.sub_agg_collector, agg_data, true)?;
|
||||
.flush_local(&mut self.sub_agg_collector, agg_data)?;
|
||||
self.num_docs = 0;
|
||||
}
|
||||
self.sub_agg_collector.flush(agg_data)?;
|
||||
@@ -150,7 +148,6 @@ impl SubAggBuffer for HighCardSubAggBuffer {
|
||||
&mut self,
|
||||
sub_agg: &mut Box<dyn SegmentAggregationCollector>,
|
||||
agg_data: &mut AggregationsSegmentCtx,
|
||||
_force: bool,
|
||||
) -> crate::Result<()> {
|
||||
let mut max_bucket = 0u32;
|
||||
for partition in self.partitions.iter() {
|
||||
@@ -210,34 +207,11 @@ impl SubAggBuffer for LowCardSubAggBuffer {
|
||||
&mut self,
|
||||
sub_agg: &mut Box<dyn SegmentAggregationCollector>,
|
||||
agg_data: &mut AggregationsSegmentCtx,
|
||||
force: bool,
|
||||
) -> crate::Result<()> {
|
||||
// Pre-aggregated: call collect per bucket.
|
||||
let max_bucket = (self.per_bucket_docs.len() as BucketId).saturating_sub(1);
|
||||
sub_agg.prepare_max_bucket(max_bucket, agg_data)?;
|
||||
// The threshold above which we flush buckets individually.
|
||||
// Note: We need to make sure that we don't lock ourselves into a situation where we hit
|
||||
// the FLUSH_THRESHOLD, but never flush any buckets. (except the final flush)
|
||||
let mut bucket_treshold = FLUSH_THRESHOLD / (self.per_bucket_docs.len().max(1) * 2);
|
||||
const _: () = {
|
||||
// MAX_NUM_TERMS_FOR_VEC threshold is used for term aggregations
|
||||
// Note: There may be other flexible values, for other aggregations, but we can use the
|
||||
// const value here as a upper bound. (better than nothing)
|
||||
let bucket_treshold_limit = FLUSH_THRESHOLD / (MAX_NUM_TERMS_FOR_VEC as usize * 2);
|
||||
assert!(
|
||||
bucket_treshold_limit > 0,
|
||||
"Bucket threshold must be greater than 0"
|
||||
);
|
||||
};
|
||||
if force {
|
||||
bucket_treshold = 0;
|
||||
}
|
||||
for (bucket_id, docs) in self
|
||||
.per_bucket_docs
|
||||
.iter()
|
||||
.enumerate()
|
||||
.filter(|(_, docs)| docs.len() > bucket_treshold)
|
||||
{
|
||||
for (bucket_id, docs) in self.per_bucket_docs.iter().enumerate() {
|
||||
sub_agg.collect(bucket_id as BucketId, docs, agg_data)?;
|
||||
}
|
||||
|
||||
|
||||
Reference in New Issue
Block a user