From efe75d408b77312facd46adf411497bdfb375690 Mon Sep 17 00:00:00 2001 From: discord9 Date: Thu, 6 Aug 2026 16:45:19 +0800 Subject: [PATCH] fix(mito2): fail open when Bloom IN predicate has non-literal or unencodable members (#8709) * fix(mito2): fail open when Bloom IN predicate has non-literal or unencodable members collect_in_list previously filtered out non-literal and encoding-failed IN members and could build a partial hard-pruning predicate, so scanning with the Bloom filter enabled could prune rows that actually match the query (false-negative results). Now any non-literal member or any encoding failure disables Bloom pruning for the whole IN expression (fail open), while independent AND subpredicates such as `col = 42` are still extracted as before. Encoding errors are logged and ignored, so queries never fail. Adds builder unit tests covering pure-literal, pure-nonliteral, mixed literal+null, all-null, mixed-nonliteral-with-AND, and encoding-failure cases, plus an engine-level witness test asserting Bloom-on and Bloom-off scans return identical rows for a mixed IN filter. Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com> * test(mito2): rename bloom filter tests to semantic names without bug-id prefixes Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com> --------- Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com> (cherry picked from commit 70ddf1685eb63d6faca2d0009fe157369c950f73) (cherry picked from commit 5cbab008ffcdd83c0c5603b81448e1a9b2790254) --- src/mito2/src/engine/index_build_test.rs | 215 +++++++++++++++++- .../sst/index/bloom_filter/applier/builder.rs | 147 ++++++++++-- 2 files changed, 339 insertions(+), 23 deletions(-) diff --git a/src/mito2/src/engine/index_build_test.rs b/src/mito2/src/engine/index_build_test.rs index 90815711649..ce950f8a246 100644 --- a/src/mito2/src/engine/index_build_test.rs +++ b/src/mito2/src/engine/index_build_test.rs @@ -24,6 +24,8 @@ use api::v1::{Rows, SemanticType}; use async_trait::async_trait; use common_base::readable_size::ReadableSize; use common_recordbatch::RecordBatches; +use datafusion_expr::expr::InList; +use datafusion_expr::{Expr, col, lit}; use datatypes::arrow::array::AsArray; use datatypes::arrow::datatypes::TimestampMillisecondType; use datatypes::prelude::ConcreteDataType; @@ -38,14 +40,16 @@ use store_api::storage::{RegionId, ScanRequest}; use tokio::sync::{Notify, Semaphore}; use crate::cache::file_cache::{FileType, IndexKey}; +use crate::cache::{CacheManager, CacheStrategy}; use crate::config::{IndexBuildMode, MitoConfig, Mode}; use crate::engine::MitoEngine; use crate::engine::compaction_test::put_and_flush; use crate::engine::listener::{EventListener, GateIndexBuildListener, IndexBuildListener}; use crate::manifest::action::RegionEdit; -use crate::read::scan_region::Scanner; +use crate::read::scan_region::{ScanRegion, Scanner}; use crate::sst::file::{FileMeta, RegionFileId, RegionIndexId}; use crate::sst::location; +use crate::test_util::batch_util::sort_batches_and_print; use crate::test_util::{ CreateRequestBuilder, TestEnv, build_rows, flush_region, put_rows, reopen_region, rows_schema, }; @@ -98,6 +102,215 @@ fn assert_listener_counts( assert_eq!(listener.finish_count(), expected_success_count); } +fn bloom_filter_test_cache() -> Arc { + const CACHE_SIZE: u64 = 64 * 1024; + + Arc::new( + CacheManager::builder() + .index_metadata_size(CACHE_SIZE) + .index_content_size(CACHE_SIZE) + .index_content_page_size(CACHE_SIZE) + .puffin_metadata_size(CACHE_SIZE) + .build(), + ) +} + +async fn scan_with_bloom_filter( + engine: &MitoEngine, + region_id: RegionId, + request: ScanRequest, + cache_manager: Arc, + ignore_bloom_filter: bool, +) -> RecordBatches { + let region = engine.get_region(region_id).unwrap(); + let scanner = ScanRegion::new( + region.version(), + region.access_layer.clone(), + request, + CacheStrategy::EnableAll(cache_manager), + ) + .with_ignore_bloom_filter(ignore_bloom_filter) + .scanner() + .await + .unwrap(); + + RecordBatches::try_collect(scanner.scan().await.unwrap()) + .await + .unwrap() +} + +#[tokio::test] +async fn bloom_mixed_in_index_on_off_equivalent() { + let mut env = TestEnv::with_prefix("bloom_mixed_in_index_on_off_equivalent_").await; + let listener = Arc::new(IndexBuildListener::default()); + let engine = env + .create_engine_with( + async_build_mode_config(true), + None, + Some(listener.clone()), + None, + ) + .await; + let region_id = RegionId::new(1, 1); + + env.get_schema_metadata_manager() + .register_region_table_info( + region_id.table_id(), + "test_table", + "test_catalog", + "test_schema", + None, + env.get_kv_backend(), + ) + .await; + + let mut request = CreateRequestBuilder::new().build(); + let bloom_options = + SkippingIndexOptions::new_unchecked(1, 0.0001, SkippingIndexType::BloomFilter); + request.column_metadatas[0] + .column_schema + .set_skipping_options(&bloom_options) + .unwrap(); + let column_schemas = rows_schema(&request); + engine + .handle_request(region_id, RegionRequest::Create(request)) + .await + .unwrap(); + + put_rows( + &engine, + region_id, + Rows { + schema: column_schemas.clone(), + rows: build_rows(0, 2), + }, + ) + .await; + flush_region(&engine, region_id, None).await; + listener.wait_finish(1).await; + + let scanner = engine + .scanner(region_id, ScanRequest::default()) + .await + .unwrap(); + assert_eq!(scanner.num_files(), 1); + assert_eq!(num_of_index_files(&engine, &scanner, region_id).await, 1); + assert!( + engine + .all_index_metas() + .await + .iter() + .any(|meta| meta.index_type == "bloom_filter") + ); + + put_rows( + &engine, + region_id, + Rows { + schema: column_schemas, + rows: build_rows(2, 4), + }, + ) + .await; + let scanner = engine + .scanner(region_id, ScanRequest::default()) + .await + .unwrap(); + assert_eq!(scanner.num_files(), 1); + assert_eq!(scanner.num_memtables(), 1); + + let mixed_in = Expr::InList(InList { + expr: Box::new(col("tag_0")), + list: vec![ + lit("definitely_absent_0"), + lit("definitely_absent_1"), + lit("definitely_absent_2"), + lit("definitely_absent_3"), + col("tag_0"), + ], + negated: false, + }); + let filtered_request = ScanRequest { + filters: vec![mixed_in], + ..Default::default() + }; + let cache_manager = bloom_filter_test_cache(); + + let indexed = scan_with_bloom_filter( + &engine, + region_id, + filtered_request.clone(), + cache_manager.clone(), + false, + ) + .await; + let unindexed = scan_with_bloom_filter( + &engine, + region_id, + filtered_request.clone(), + cache_manager, + true, + ) + .await; + let memtable_only = RecordBatches::try_collect( + engine + .scan_to_stream( + region_id, + ScanRequest { + skip_sst_files: true, + ..filtered_request.clone() + }, + ) + .await + .unwrap(), + ) + .await + .unwrap(); + let full_scan = RecordBatches::try_collect( + engine + .scan_to_stream(region_id, ScanRequest::default()) + .await + .unwrap(), + ) + .await + .unwrap(); + + let indexed_rows = sort_batches_and_print(&indexed, &["ts"]); + let unindexed_rows = sort_batches_and_print(&unindexed, &["ts"]); + let memtable_rows = sort_batches_and_print(&memtable_only, &["ts"]); + let full_rows = sort_batches_and_print(&full_scan, &["ts"]); + assert_eq!( + memtable_rows, + "\ ++-------+---------+---------------------+ +| tag_0 | field_0 | ts | ++-------+---------+---------------------+ +| 2 | 2.0 | 1970-01-01T00:00:02 | +| 3 | 3.0 | 1970-01-01T00:00:03 | ++-------+---------+---------------------+" + ); + assert_eq!( + full_rows, + "\ ++-------+---------+---------------------+ +| tag_0 | field_0 | ts | ++-------+---------+---------------------+ +| 0 | 0.0 | 1970-01-01T00:00:00 | +| 1 | 1.0 | 1970-01-01T00:00:01 | +| 2 | 2.0 | 1970-01-01T00:00:02 | +| 3 | 3.0 | 1970-01-01T00:00:03 | ++-------+---------+---------------------+" + ); + assert_eq!( + unindexed_rows, full_rows, + "disabling the Bloom filter must retain all matching SST rows" + ); + assert_eq!( + indexed_rows, unindexed_rows, + "Bloom filtering must not prune rows matched by a nonliteral IN member" + ); +} + #[derive(Clone, Copy, Debug, PartialEq, Eq)] enum IndexPublicationPhase { BeforeManifestCommit, diff --git a/src/mito2/src/sst/index/bloom_filter/applier/builder.rs b/src/mito2/src/sst/index/bloom_filter/applier/builder.rs index beb26d1bf7c..e5968b5abb2 100644 --- a/src/mito2/src/sst/index/bloom_filter/applier/builder.rs +++ b/src/mito2/src/sst/index/bloom_filter/applier/builder.rs @@ -204,21 +204,13 @@ impl<'a> BloomFilterIndexApplierBuilder<'a> { return Ok(()); }; - // Convert all non-null literals to predicates - let predicates = in_list - .list - .iter() - .filter_map(Self::nonnull_lit) - .map(|lit| encode_lit(lit, data_type.clone())); - - // Collect successful conversions let mut valid_predicates = BTreeSet::new(); - for predicate in predicates { - match predicate { - Ok(p) => { - valid_predicates.insert(p); - } - Err(e) => warn!(e; "Failed to convert value in InList"), + for expr in &in_list.list { + let Expr::Literal(lit, _) = expr else { + return Ok(()); + }; + if !lit.is_null() { + valid_predicates.insert(encode_lit(lit, data_type.clone())?); } } @@ -311,14 +303,6 @@ impl<'a> BloomFilterIndexApplierBuilder<'a> { Ok(false) } - /// Helper function to get non-null literal value - fn nonnull_lit(expr: &Expr) -> Option<&ScalarValue> { - match expr { - Expr::Literal(lit, _) if !lit.is_null() => Some(lit), - _ => None, - } - } - /// Helper function to get the column and literal value from an equality expr (column = lit) fn eq_expr_col_lit<'b>( left: &'b Expr, @@ -435,6 +419,125 @@ mod tests { i.lit() } + fn build_bloom_predicates(exprs: &[Expr]) -> Option>> { + let (_d, factory) = PuffinManagerFactory::new_for_test_block("bloom_builder_"); + let metadata = test_region_metadata(); + BloomFilterIndexApplierBuilder::new( + "test".to_string(), + PathType::Bare, + test_object_store(), + &metadata, + factory, + ) + .build(exprs) + .unwrap() + .map(|applier| (*applier.default_predicates).clone()) + } + + fn int64_inlist_predicate(values: impl IntoIterator) -> InListPredicate { + InListPredicate { + list: values + .into_iter() + .map(|value| { + encode_lit( + &ScalarValue::Int64(Some(value)), + ConcreteDataType::int64_datatype(), + ) + .unwrap() + }) + .collect(), + } + } + + #[test] + fn bloom_pure_literal_in_extracts_exact_predicate() { + let expr = Expr::InList(InList { + expr: Box::new(column("column2")), + list: vec![int64_lit(1), int64_lit(2), int64_lit(3)], + negated: false, + }); + + assert_eq!( + build_bloom_predicates(&[expr]), + Some(BTreeMap::from([( + 2, + vec![int64_inlist_predicate([1, 2, 3])] + )])) + ); + } + + #[test] + fn bloom_pure_nonliteral_in_does_not_extract_predicate() { + let expr = Expr::InList(InList { + expr: Box::new(column("column1")), + list: vec![column("column1")], + negated: false, + }); + + assert_eq!(build_bloom_predicates(&[expr]), None); + } + + #[test] + fn bloom_mixed_literal_null_in_extracts_exact_predicate() { + let expr = Expr::InList(InList { + expr: Box::new(column("column2")), + list: vec![ + int64_lit(1), + Expr::Literal(ScalarValue::Int64(None), None), + int64_lit(3), + ], + negated: false, + }); + + assert_eq!( + build_bloom_predicates(&[expr]), + Some(BTreeMap::from([(2, vec![int64_inlist_predicate([1, 3])])])) + ); + } + + #[test] + fn bloom_all_literal_null_in_does_not_extract_predicate() { + let expr = Expr::InList(InList { + expr: Box::new(column("column2")), + list: vec![ + Expr::Literal(ScalarValue::Int64(None), None), + Expr::Literal(ScalarValue::Int64(None), None), + ], + negated: false, + }); + + assert_eq!(build_bloom_predicates(&[expr]), None); + } + + #[test] + fn bloom_mixed_nonliteral_in_keeps_only_independent_predicate() { + let expr = Expr::BinaryExpr(BinaryExpr { + left: Box::new(Expr::InList(InList { + expr: Box::new(column("column1")), + list: vec!["definitely_absent".lit(), column("column1")], + negated: false, + })), + op: Operator::And, + right: Box::new(column("column2").eq(int64_lit(42))), + }); + + assert_eq!( + build_bloom_predicates(&[expr]), + Some(BTreeMap::from([(2, vec![int64_inlist_predicate([42])])])) + ); + } + + #[test] + fn bloom_encoding_failure_in_does_not_extract_predicate() { + let expr = Expr::InList(InList { + expr: Box::new(column("column2")), + list: vec![int64_lit(1), "not_an_int64".lit()], + negated: false, + }); + + assert_eq!(build_bloom_predicates(&[expr]), None); + } + #[test] fn test_build_with_in_list() { let (_d, factory) = PuffinManagerFactory::new_for_test_block("test_build_with_in_list_");