feat(query): support inverted_filter to omit the filter executor (#12934)

* feat(query): support inverted_filter to omit the filter executor

* feat(query): support inverted_filter to omit the filter executor

* feat(query): support inverted_filter to omit the filter executor

* feat(query): support inverted_filter to omit the filter executor

* feat(query): support inverted_filter to omit the filter executor

* feat(query): address comments

Author: sundy-li

src/common/storage/src/metrics/common.rs

@@ -0,0 +1,30 @@
+// Copyright 2021 Datafuse Labs
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+use common_metrics::register_counter;
+use common_metrics::Counter;
+use lazy_static::lazy_static;
+
+lazy_static! {
+    static ref OMIT_FILTER_ROWGROUPS: Counter = register_counter("omit_filter_rowgroups");
+    static ref OMIT_FILTER_ROWS: Counter = register_counter("omit_filter_rows");
+}
+
+pub fn metrics_inc_omit_filter_rowgroups(c: u64) {
+    OMIT_FILTER_ROWGROUPS.inc_by(c);
+}
+
+pub fn metrics_inc_omit_filter_rows(c: u64) {
+    OMIT_FILTER_ROWS.inc_by(c);
+}

src/common/storage/src/metrics/mod.rs

@@ -12,6 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+pub mod common;
 pub mod copy;
 pub mod merge_into;
 mod storage_metrics;

src/query/catalog/src/plan/pushdown.rs

@@ -77,9 +77,9 @@ pub struct PushDownInfo {
     /// The difference with `projection` is the removal of the source columns
     /// which were only used to generate virtual columns.
     pub output_columns: Option<Projection>,
-    /// Optional filter expression plan
+    /// Optional filter and reverse filter expression plan
     /// Assumption: expression's data type must be `DataType::Boolean`.
-    pub filter: Option<RemoteExpr<String>>,
+    pub filters: Option<Filters>,
     pub is_deterministic: bool,
     /// Optional prewhere information
     /// used for prewhere optimization
@@ -96,6 +96,12 @@ pub struct PushDownInfo {
     pub agg_index: Option<AggIndexInfo>,
 }
 
+#[derive(serde::Serialize, serde::Deserialize, Clone, Debug, PartialEq, Eq)]
+pub struct Filters {
+    pub filter: RemoteExpr<String>,
+    pub inverted_filter: RemoteExpr<String>,
+}
+
 /// TopK is a wrapper for topk push down items.
 /// We only take the first column in order_by as the topk column.
 #[derive(Debug, Clone)]

src/query/catalog/src/table.rs

@@ -465,14 +465,6 @@ pub struct NavigationDescriptor {
     pub point: NavigationPoint,
 }
 
-#[derive(Debug, Clone)]
-pub struct DeletionFilters {
-    // the filter expression for the deletion
-    pub filter: RemoteExpr<String>,
-    // just "not(filter)"
-    pub inverted_filter: RemoteExpr<String>,
-}
-
 use std::collections::HashMap;
 
 #[derive(serde::Serialize, serde::Deserialize, Clone, Debug, Default)]

src/query/expression/src/expression.rs

@@ -124,14 +124,17 @@ pub enum Expr<Index: ColumnIndex = usize> {
 ///
 /// The remote node will recover the `Arc` pointer within `FunctionCall` by looking
 /// up the function registry with the `FunctionID`.
-#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+#[derive(Debug, Clone, Educe, Serialize, Deserialize)]
+#[educe(PartialEq, Eq, Hash)]
 pub enum RemoteExpr<Index: ColumnIndex = usize> {
     Constant {
+        #[educe(Hash(ignore), PartialEq(ignore), Eq(ignore))]
         span: Span,
         scalar: Scalar,
         data_type: DataType,
     },
     ColumnRef {
+        #[educe(Hash(ignore), PartialEq(ignore), Eq(ignore))]
         span: Span,
         id: Index,
         data_type: DataType,
@@ -140,19 +143,22 @@ pub enum RemoteExpr<Index: ColumnIndex = usize> {
         display_name: String,
     },
     Cast {
+        #[educe(Hash(ignore), PartialEq(ignore), Eq(ignore))]
         span: Span,
         is_try: bool,
         expr: Box<RemoteExpr<Index>>,
         dest_type: DataType,
     },
     FunctionCall {
+        #[educe(Hash(ignore), PartialEq(ignore), Eq(ignore))]
         span: Span,
         id: FunctionID,
         generics: Vec<DataType>,
         args: Vec<RemoteExpr<Index>>,
         return_type: DataType,
     },
     UDFServerCall {
+        #[educe(Hash(ignore), PartialEq(ignore), Eq(ignore))]
         span: Span,
         func_name: String,
         server_addr: String,

src/query/service/src/interpreters/common/mod.rs

@@ -25,5 +25,6 @@ pub use refresh_aggregating_index::hook_refresh_agg_index;
 pub use refresh_aggregating_index::RefreshAggIndexDesc;
 pub use table::check_referenced_computed_columns;
 pub use util::check_deduplicate_label;
+pub use util::create_push_down_filters;
 
 pub use self::metrics::*;

src/query/service/src/interpreters/common/util.rs

@@ -14,11 +14,17 @@
 
 use std::sync::Arc;
 
+use common_catalog::plan::Filters;
 use common_catalog::table_context::TableContext;
 use common_exception::Result;
+use common_expression::type_check::check_function;
+use common_functions::BUILTIN_FUNCTIONS;
 use common_meta_kvapi::kvapi::KVApi;
 use common_users::UserApiProvider;
 
+use crate::sql::executor::cast_expr_to_non_null_boolean;
+use crate::sql::ScalarExpr;
+
 /// Checks if a duplicate label exists in the meta store.
 ///
 /// # Arguments
@@ -41,3 +47,22 @@ pub async fn check_deduplicate_label(ctx: Arc<dyn TableContext>) -> Result<bool>
         }
     }
 }
+
+pub fn create_push_down_filters(scalar: &ScalarExpr) -> Result<Filters> {
+    let filter = cast_expr_to_non_null_boolean(
+        scalar
+            .as_expr()?
+            .project_column_ref(|col| col.column_name.clone()),
+    )?;
+
+    let remote_filter = filter.as_remote_expr();
+
+    // prepare the inverse filter expression
+    let remote_inverted_filter =
+        check_function(None, "not", &[], &[filter], &BUILTIN_FUNCTIONS)?.as_remote_expr();
+
+    Ok(Filters {
+        filter: remote_filter,
+        inverted_filter: remote_inverted_filter,
+    })
+}

src/query/service/src/interpreters/interpreter_delete.rs

@@ -18,7 +18,6 @@ use std::sync::Arc;
 
 use common_base::runtime::GlobalIORuntime;
 use common_catalog::plan::Partitions;
-use common_catalog::table::DeletionFilters;
 use common_exception::ErrorCode;
 use common_exception::Result;
 use common_expression::types::DataType;
@@ -30,7 +29,6 @@ use common_functions::BUILTIN_FUNCTIONS;
 use common_meta_app::schema::CatalogInfo;
 use common_meta_app::schema::TableInfo;
 use common_sql::binder::ColumnBindingBuilder;
-use common_sql::executor::cast_expr_to_non_null_boolean;
 use common_sql::executor::DeletePartial;
 use common_sql::executor::Exchange;
 use common_sql::executor::FragmentKind;
@@ -60,6 +58,7 @@ use log::debug;
 use storages_common_table_meta::meta::TableSnapshot;
 use table_lock::TableLockHandlerWrapper;
 
+use crate::interpreters::common::create_push_down_filters;
 use crate::interpreters::Interpreter;
 use crate::interpreters::SelectInterpreter;
 use crate::pipelines::executor::ExecutorSettings;
@@ -164,36 +163,15 @@ impl Interpreter for DeleteInterpreter {
 
         let (filters, col_indices) = if let Some(scalar) = selection {
             // prepare the filter expression
-            let filter = cast_expr_to_non_null_boolean(
-                scalar
-                    .as_expr()?
-                    .project_column_ref(|col| col.column_name.clone()),
-            )?
-            .as_remote_expr();
-
-            let expr = filter.as_expr(&BUILTIN_FUNCTIONS);
+            let filters = create_push_down_filters(&scalar)?;
+
+            let expr = filters.filter.as_expr(&BUILTIN_FUNCTIONS);
             if !expr.is_deterministic(&BUILTIN_FUNCTIONS) {
                 return Err(ErrorCode::Unimplemented(
                     "Delete must have deterministic predicate",
                 ));
             }
 
-            // prepare the inverse filter expression
-            let inverted_filter = {
-                let inverse = ScalarExpr::FunctionCall(common_sql::planner::plans::FunctionCall {
-                    span: None,
-                    func_name: "not".to_string(),
-                    params: vec![],
-                    arguments: vec![scalar.clone()],
-                });
-                cast_expr_to_non_null_boolean(
-                    inverse
-                        .as_expr()?
-                        .project_column_ref(|col| col.column_name.clone()),
-                )?
-                .as_remote_expr()
-            };
-
             let col_indices: Vec<usize> = if !self.plan.subquery_desc.is_empty() {
                 let mut col_indices = HashSet::new();
                 for subquery_desc in &self.plan.subquery_desc {
@@ -203,13 +181,7 @@ impl Interpreter for DeleteInterpreter {
             } else {
                 scalar.used_columns().into_iter().collect()
             };
-            (
-                Some(DeletionFilters {
-                    filter,
-                    inverted_filter,
-                }),
-                col_indices,
-            )
+            (Some(filters), col_indices)
         } else {
             (None, vec![])
         };

src/query/service/src/interpreters/interpreter_table_recluster.rs

@@ -16,9 +16,12 @@ use std::sync::Arc;
 use std::time::Duration;
 use std::time::SystemTime;
 
+use common_catalog::plan::Filters;
 use common_catalog::plan::PushDownInfo;
 use common_exception::ErrorCode;
 use common_exception::Result;
+use common_expression::type_check::check_function;
+use common_functions::BUILTIN_FUNCTIONS;
 use log::error;
 use log::info;
 use log::warn;
@@ -31,6 +34,7 @@ use crate::pipelines::Pipeline;
 use crate::pipelines::PipelineBuildResult;
 use crate::sessions::QueryContext;
 use crate::sessions::TableContext;
+use crate::sql::executor::cast_expr_to_non_null_boolean;
 use crate::sql::plans::ReclusterTablePlan;
 
 pub struct ReclusterTableInterpreter {
@@ -68,13 +72,23 @@ impl Interpreter for ReclusterTableInterpreter {
 
         // Build extras via push down scalar
         let extras = if let Some(scalar) = &plan.push_downs {
-            let filter = scalar
-                .as_expr()?
-                .project_column_ref(|col| col.column_name.clone())
-                .as_remote_expr();
+            // prepare the filter expression
+            let filter = cast_expr_to_non_null_boolean(
+                scalar
+                    .as_expr()?
+                    .project_column_ref(|col| col.column_name.clone()),
+            )?;
+            // prepare the inverse filter expression
+            let inverted_filter =
+                check_function(None, "not", &[], &[filter.clone()], &BUILTIN_FUNCTIONS)?;
+
+            let filters = Filters {
+                filter: filter.as_remote_expr(),
+                inverted_filter: inverted_filter.as_remote_expr(),
+            };
 
             Some(PushDownInfo {
-                filter: Some(filter),
+                filters: Some(filters),
                 ..PushDownInfo::default()
             })
         } else {

src/query/service/src/table_functions/numbers/numbers_table.rs

@@ -137,7 +137,7 @@ impl Table for NumbersTable {
         let mut limit = None;
 
         if let Some(extras) = &push_downs {
-            if extras.limit.is_some() && extras.filter.is_none() && extras.order_by.is_empty() {
+            if extras.limit.is_some() && extras.filters.is_none() && extras.order_by.is_empty() {
                 // It is allowed to have an error when we can't get sort columns from the expression. For
                 // example 'select number from numbers(10) order by number+4 limit 10', the column 'number+4'
                 // doesn't exist in the numbers table.