perf(mito2): skip remaining column indexes after empty row groups

Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com>

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(), &region_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(),
+            &region_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(&region_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,
+            &region_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(&region_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() {

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 {

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(