mirror of
https://github.com/GreptimeTeam/greptimedb.git
synced 2026-10-13 07:22:32 +00:00
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)
This commit is contained in:
discord9
@@ -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<CacheManager> {
|
||||
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<CacheManager>,
|
||||
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,
|
||||
|
||||
@@ -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<BTreeMap<ColumnId, Vec<InListPredicate>>> {
|
||||
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<Item = i64>) -> 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_");
|
||||
|
||||
Reference in New Issue
Block a user