fix(query): push LIKE filters down into MergeScan remote plans (#9648)

Signed-off-by: Dennis Zhuang <killme2008@gmail.com>

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)

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(_)

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;
 
 +-------------------------+-----------------------------------+---------------------------------+-----------------------------------------+-------------------------------+

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;