From da795705c03b062806bce65f34503ddb24ae03fb Mon Sep 17 00:00:00 2001 From: discord9 <55937128+discord9@users.noreply.github.com> Date: Mon, 28 Sep 2026 16:19:14 +0800 Subject: [PATCH] perf(mito2): skip remaining column indexes after empty row groups Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com> --- .../src/sst/index/bloom_filter/applier.rs | 99 ++++++++++++++ .../src/sst/index/fulltext_index/applier.rs | 6 +- .../src/sst/index/fulltext_index/creator.rs | 123 ++++++++++++++++++ 3 files changed, 227 insertions(+), 1 deletion(-) diff --git a/src/mito2/src/sst/index/bloom_filter/applier.rs b/src/mito2/src/sst/index/bloom_filter/applier.rs index 1faae466c89..1dd86ce961d 100644 --- a/src/mito2/src/sst/index/bloom_filter/applier.rs +++ b/src/mito2/src/sst/index/bloom_filter/applier.rs @@ -266,6 +266,9 @@ impl BloomFilterIndexApplier { .await .context(ApplyBloomFilterIndexSnafu)?; } + if output.iter().all(|(_, ranges)| ranges.is_empty()) { + break; + } } // adjust ranges to be based on row group @@ -651,6 +654,102 @@ mod tests { } } + #[tokio::test] + async fn test_bloom_filter_applier_short_circuits_after_empty_ranges() { + let region_metadata = mock_region_metadata(); + let (d, factory) = + PuffinManagerFactory::new_for_test_async("test_bloom_filter_applier_short_circuit_") + .await; + let object_store = mock_object_store(); + let intm_mgr = new_intm_mgr(d.path().to_string_lossy()).await; + let file_id = RegionIndexId::new( + RegionFileId::new(region_metadata.region_id, FileId::random()), + 0, + ); + let table_dir = "table_dir".to_string(); + let mut indexer = + BloomFilterIndexer::new(file_id.file_id(), ®ion_metadata, intm_mgr, Some(1024)) + .unwrap() + .unwrap(); + let mut batch = new_batch("tag1", 0..10); + indexer.update(&mut batch).await.unwrap(); + let mut batch = new_batch("tag2", 10..20); + indexer.update(&mut batch).await.unwrap(); + let manager = factory.build( + object_store.clone(), + RegionFilePathFactory::new(table_dir.clone(), PathType::Bare), + ); + let mut writer = manager.writer(&file_id).await.unwrap(); + indexer.finish(&mut writer).await.unwrap(); + writer.finish().await.unwrap(); + + let applier = BloomFilterIndexApplierBuilder::new( + table_dir, + PathType::Bare, + object_store.clone(), + ®ion_metadata, + factory.clone(), + ) + .build(&[ + col("tag_str").eq(lit("missing")), + col("field_u64").eq(lit(1u64)), + ]) + .unwrap() + .unwrap(); + let predicates = applier + .compatible_predicate_for_sst(®ion_metadata) + .unwrap(); + let mut metrics = BloomFilterIndexApplyMetrics::default(); + let result = applier + .apply( + file_id, + None, + &predicates, + [(5, true), (5, true), (5, true), (5, true)].into_iter(), + Some(&mut metrics), + ) + .await + .unwrap(); + + assert_eq!( + result, + vec![(0, vec![]), (1, vec![]), (2, vec![]), (3, vec![])] + ); + assert_eq!(metrics.blob_cache_miss, 1); + assert!(metrics.read_metrics.total_ranges > 0); + + let positive_applier = BloomFilterIndexApplierBuilder::new( + "table_dir".to_string(), + PathType::Bare, + object_store, + ®ion_metadata, + factory, + ) + .build(&[ + col("tag_str").eq(lit("tag1")), + col("field_u64").eq(lit(1u64)), + ]) + .unwrap() + .unwrap(); + let predicates = positive_applier + .compatible_predicate_for_sst(®ion_metadata) + .unwrap(); + let mut metrics = BloomFilterIndexApplyMetrics::default(); + let result = positive_applier + .apply( + file_id, + None, + &predicates, + [(5, true), (5, true), (5, true), (5, true)].into_iter(), + Some(&mut metrics), + ) + .await + .unwrap(); + assert_eq!(result[0].0, 0); + assert!(!result[0].1.is_empty()); + assert_eq!(metrics.blob_cache_miss, 2); + } + #[tokio::test] #[allow(clippy::single_range_in_vec_init)] async fn test_bloom_filter_applier() { diff --git a/src/mito2/src/sst/index/fulltext_index/applier.rs b/src/mito2/src/sst/index/fulltext_index/applier.rs index 4efb2a5ccb9..5fcacb9e801 100644 --- a/src/mito2/src/sst/index/fulltext_index/applier.rs +++ b/src/mito2/src/sst/index/fulltext_index/applier.rs @@ -381,7 +381,7 @@ impl FulltextIndexApplier { continue; } - applied |= self + let column_applied = self .apply_coarse_one_column( file_id, file_size_hint, @@ -391,6 +391,10 @@ impl FulltextIndexApplier { metrics.as_deref_mut(), ) .await?; + applied |= column_applied; + if column_applied && output.iter().all(|(_, ranges)| ranges.is_empty()) { + break; + } } if !applied { diff --git a/src/mito2/src/sst/index/fulltext_index/creator.rs b/src/mito2/src/sst/index/fulltext_index/creator.rs index 800e0b77a09..7c2ec1562da 100644 --- a/src/mito2/src/sst/index/fulltext_index/creator.rs +++ b/src/mito2/src/sst/index/fulltext_index/creator.rs @@ -803,6 +803,129 @@ mod tests { assert_eq!(row_ids, None); } + #[tokio::test] + async fn test_fulltext_coarse_short_circuits_empty_ranges() { + let prefix = "test_fulltext_coarse_short_circuit_"; + let (d, factory) = PuffinManagerFactory::new_for_test_async(prefix).await; + let table_dir = "table0".to_string(); + let object_store = mock_object_store(); + let metadata = mock_region_metadata(FulltextBackend::Bloom); + let intm_mgr = new_intm_mgr(d.path().to_string_lossy()).await; + let file_id = + RegionIndexId::new(RegionFileId::new(metadata.region_id, FileId::random()), 0); + let mut indexer = FulltextIndexer::new( + &metadata.region_id, + &file_id.file_id(), + &intm_mgr, + &metadata, + true, + 1024, + ) + .await + .unwrap() + .unwrap(); + let mut batch = new_batch(&[ + (Some("hello"), Some("hello"), None), + (Some("world"), Some("world"), None), + ]); + indexer.update(&mut batch).await.unwrap(); + + let manager = factory.build( + object_store.clone(), + RegionFilePathFactory::new(table_dir.clone(), PathType::Bare), + ); + let mut writer = manager.writer(&file_id).await.unwrap(); + indexer.finish(&mut writer).await.unwrap(); + writer.finish().await.unwrap(); + + let terms = |first: &str| { + BTreeMap::from([ + ( + 1, + FulltextRequest { + terms: vec![FulltextTerm { + col_lowered: false, + term: first.to_string(), + }], + ..Default::default() + }, + ), + ( + 2, + FulltextRequest { + terms: vec![FulltextTerm { + col_lowered: false, + term: "hello".to_string(), + }], + ..Default::default() + }, + ), + ]) + }; + let applier = FulltextIndexApplier::new( + table_dir.clone(), + PathType::Bare, + object_store.clone(), + terms("absent"), + factory.clone(), + ); + let mut metrics = + crate::sst::index::fulltext_index::applier::FulltextIndexApplyMetrics::default(); + let result = applier + .apply_coarse( + file_id, + None, + [(1, true), (1, true)].into_iter(), + Some(&mut metrics), + ) + .await + .unwrap(); + assert_eq!(result, Some(vec![(0, vec![]), (1, vec![])])); + assert_eq!(metrics.blob_cache_miss, 1); + + let not_applied = FulltextIndexApplier::new( + table_dir.clone(), + PathType::Bare, + object_store.clone(), + BTreeMap::from([( + 1, + FulltextRequest { + terms: vec![FulltextTerm { + col_lowered: true, + term: "hello".to_string(), + }], + ..Default::default() + }, + )]), + factory.clone(), + ) + .apply_coarse(file_id, None, [(1, true)].into_iter(), None) + .await + .unwrap(); + assert_eq!(not_applied, None); + + let positive_applier = FulltextIndexApplier::new( + table_dir, + PathType::Bare, + object_store, + terms("hello"), + factory, + ); + let mut positive_metrics = + crate::sst::index::fulltext_index::applier::FulltextIndexApplyMetrics::default(); + let result = positive_applier + .apply_coarse( + file_id, + None, + [(1, true), (1, true)].into_iter(), + Some(&mut positive_metrics), + ) + .await + .unwrap(); + assert!(result.unwrap().iter().any(|(_, ranges)| !ranges.is_empty())); + assert_eq!(positive_metrics.blob_cache_miss, 2); + } + #[tokio::test] async fn test_fulltext_index_basic_case_sensitive_bloom() { let applier_factory = build_fulltext_applier_factory(