diff --git a/src/query/src/dist_plan/analyzer/test.rs b/src/query/src/dist_plan/analyzer/test.rs index a689d94c282..207de7d43fd 100644 --- a/src/query/src/dist_plan/analyzer/test.rs +++ b/src/query/src/dist_plan/analyzer/test.rs @@ -530,6 +530,72 @@ fn dictionary_literals_in_values_keep_their_schema() { ); } +#[test] +fn like_filters_are_pushed_down() { + let test_table = TestTable::table_with_name(0, "t".to_string()); + let table_provider = Arc::new(DfTableProviderAdapter::new(test_table)); + let table_source = Arc::new(DefaultTableSource::new(table_provider.clone() as _)); + let ctx = SessionContext::new(); + ctx.register_table(TableReference::bare("t"), table_provider as _) + .unwrap(); + + let like = |negated, pattern: &str, escape_char, case_insensitive| { + Expr::Like(datafusion_expr::expr::Like::new( + negated, + Box::new(col("t.pk1")), + Box::new(lit(pattern.to_string())), + escape_char, + case_insensitive, + )) + }; + // Substrait has no negated LIKE, so `NOT LIKE` comes back as `NOT (.. LIKE ..)`. + for (predicate, decoded) in [ + ( + like(false, "% timeout %", None, false), + like(false, "% timeout %", None, false), + ), + ( + like(true, "% timeout %", None, false), + Expr::Not(Box::new(like(false, "% timeout %", None, false))), + ), + ( + like(false, "% timeout %", None, true), + like(false, "% timeout %", None, true), + ), + ( + like(false, "100!% done", Some('!'), false), + like(false, "100!% done", Some('!'), false), + ), + ] { + let plan = LogicalPlanBuilder::scan_with_filters("t", table_source.clone(), None, vec![]) + .unwrap() + .filter(predicate.clone()) + .unwrap() + .build() + .unwrap(); + + let result = DistPlannerAnalyzer {} + .analyze(plan, &ConfigOptions::default()) + .unwrap(); + let remote_input = find_merge_scan(&result).unwrap().input(); + assert!( + has_filter_above_table_scan(remote_input, "t", &predicate), + "{result}" + ); + + let bytes = DFLogicalSubstraitConvertor + .encode(remote_input, DefaultSerializer) + .unwrap(); + let decoded_plan = + futures::executor::block_on(DFLogicalSubstraitConvertor.decode(bytes, ctx.state())) + .unwrap(); + assert!( + has_filter_above_table_scan(&decoded_plan, "t", &decoded), + "{decoded_plan}" + ); + } +} + fn try_encode_decode_substrait(plan: &LogicalPlan, state: SessionState) { let sub_plan_bytes = substrait::DFLogicalSubstraitConvertor .encode(plan, crate::query_engine::DefaultSerializer) diff --git a/src/query/src/dist_plan/commutativity.rs b/src/query/src/dist_plan/commutativity.rs index d02b746a8bf..b69589a7cfe 100644 --- a/src/query/src/dist_plan/commutativity.rs +++ b/src/query/src/dist_plan/commutativity.rs @@ -113,7 +113,6 @@ impl Categorizer { } Commutativity::Commutative } - // TODO(ruihang): Change this to Commutative once Like is supported in substrait LogicalPlan::Filter(filter) => Self::check_expr(&filter.predicate), LogicalPlan::Window(_) => Commutativity::Unimplemented, LogicalPlan::Aggregate(aggr) => { @@ -276,6 +275,7 @@ impl Categorizer { | Expr::Between(_) | Expr::Exists(_) | Expr::InList(_) + | Expr::Like(_) | Expr::Case(_) => Commutativity::Commutative, // Annotation collection must not affect distribution. Keep scalar UDFs pushdownable; // TODO: Preserve datanode PromQL annotations through plan decoding, return them in @@ -283,8 +283,7 @@ impl Categorizer { Expr::ScalarFunction(_) => Commutativity::Commutative, Expr::AggregateFunction(_udaf) => Commutativity::Commutative, - Expr::Like(_) - | Expr::SimilarTo(_) + Expr::SimilarTo(_) | Expr::IsUnknown(_) | Expr::IsNotUnknown(_) | Expr::WindowFunction(_) diff --git a/tests/cases/standalone/common/function/fulltext_index_pruning.result b/tests/cases/standalone/common/function/fulltext_index_pruning.result index 10e4b56e2b6..b357d561984 100644 --- a/tests/cases/standalone/common/function/fulltext_index_pruning.result +++ b/tests/cases/standalone/common/function/fulltext_index_pruning.result @@ -219,6 +219,27 @@ SELECT ts, msg FROM ft_en_bloom WHERE msg LIKE '%user-123%' ORDER BY ts; | 1970-01-01T00:00:00.007 | 用户user-123登录 | +-------------------------+------------------+ +-- The `LIKE` filter runs on the datanode, after the index prunes the row group. +-- SQLNESS REPLACE (-+) - +-- SQLNESS REPLACE (\s\s+) _ +-- SQLNESS REPLACE (peers.*) REDACTED +-- SQLNESS REPLACE SeqScan:.*("filters":\s\[[^\]]*\]).*("rg_fulltext_filtered":\d+,\s"rows_fulltext_filtered":\d+).* SeqScan:_${1}_${2} +-- SQLNESS REPLACE (metrics=.*) REDACTED +EXPLAIN ANALYZE VERBOSE SELECT ts, msg FROM ft_en_bloom WHERE msg LIKE 'nomatch %'; + ++-+-+-+ +| stage | node | plan_| ++-+-+-+ +| 0_| 0_|_CooperativeExec REDACTED +|_|_|_MergeScanExec: REDACTED +|_|_|_| +| 1_| 0_|_FilterExec: msg@1 LIKE nomatch % REDACTED +|_|_|_CooperativeExec REDACTED +|_|_|_SeqScan:_"filters": ["msg LIKE Utf8(\"nomatch %\")"]_"rg_fulltext_filtered":1, "rows_fulltext_filtered":8 +|_|_|_| +|_|_| Total rows: 0_| ++-+-+-+ + SELECT ts, matches_term(lower(msg), 'quota') AS "matches_term(lower(msg), 'quota')", matches_term(lower(msg), 'vol') AS "matches_term(lower(msg), 'vol')", lower(msg) LIKE '%disk quota exceeded%' AS "lower(msg) LIKE '%disk quota exceeded%'", lower(msg) LIKE 'disk quota%' AS "lower(msg) LIKE 'disk quota%'" FROM ft_en_bloom ORDER BY ts; +-------------------------+-----------------------------------+---------------------------------+-----------------------------------------+-------------------------------+ diff --git a/tests/cases/standalone/common/function/fulltext_index_pruning.sql b/tests/cases/standalone/common/function/fulltext_index_pruning.sql index f4e711b94e0..b063c92d518 100644 --- a/tests/cases/standalone/common/function/fulltext_index_pruning.sql +++ b/tests/cases/standalone/common/function/fulltext_index_pruning.sql @@ -64,6 +64,14 @@ SELECT ts, msg FROM ft_en_bloom WHERE msg LIKE '连接timeout.%' ORDER BY ts; SELECT ts, msg FROM ft_en_bloom WHERE msg LIKE '%user-123%' ORDER BY ts; +-- The `LIKE` filter runs on the datanode, after the index prunes the row group. +-- SQLNESS REPLACE (-+) - +-- SQLNESS REPLACE (\s\s+) _ +-- SQLNESS REPLACE (peers.*) REDACTED +-- SQLNESS REPLACE SeqScan:.*("filters":\s\[[^\]]*\]).*("rg_fulltext_filtered":\d+,\s"rows_fulltext_filtered":\d+).* SeqScan:_${1}_${2} +-- SQLNESS REPLACE (metrics=.*) REDACTED +EXPLAIN ANALYZE VERBOSE SELECT ts, msg FROM ft_en_bloom WHERE msg LIKE 'nomatch %'; + SELECT ts, matches_term(lower(msg), 'quota') AS "matches_term(lower(msg), 'quota')", matches_term(lower(msg), 'vol') AS "matches_term(lower(msg), 'vol')", lower(msg) LIKE '%disk quota exceeded%' AS "lower(msg) LIKE '%disk quota exceeded%'", lower(msg) LIKE 'disk quota%' AS "lower(msg) LIKE 'disk quota%'" FROM ft_en_bloom ORDER BY ts; SELECT ts, msg FROM ft_en_bloom WHERE matches_term(lower(msg), 'quota') ORDER BY ts;