fix(query): fix prewhere cannot push down for native storage

scripts/ci/ci-run-sqllogic-tests-native.sh

@@ -19,4 +19,4 @@ fi
 echo "Run suites using argument: $RUN_DIR"
 
 echo "Starting databend-sqllogic tests"
-target/${BUILD_PROFILE}/databend-sqllogictests --handlers ${TEST_HANDLERS} ${RUN_DIR} --skip_dir management,mode,explain,tpch --enable_sandbox --parallel 8
+target/${BUILD_PROFILE}/databend-sqllogictests --handlers ${TEST_HANDLERS} ${RUN_DIR} --skip_dir management,cluster,explain,tpch --enable_sandbox --parallel 8

scripts/ci/ci-run-sqllogic-tests.sh

@@ -19,4 +19,4 @@ fi
 echo "Run suites using argument: $RUN_DIR"
 
 echo "Starting databend-sqllogic tests"
-target/${BUILD_PROFILE}/databend-sqllogictests --handlers ${TEST_HANDLERS} ${RUN_DIR} --skip_dir management --enable_sandbox --parallel 8
+target/${BUILD_PROFILE}/databend-sqllogictests --handlers ${TEST_HANDLERS} ${RUN_DIR} --skip_dir management,explain_native --enable_sandbox --parallel 8

src/query/sql/src/planner/optimizer/rule/factory.rs

@@ -55,7 +55,9 @@ impl RuleFactory {
         match id {
             RuleID::EliminateEvalScalar => Ok(Box::new(RuleEliminateEvalScalar::new())),
             RuleID::PushDownFilterUnion => Ok(Box::new(RulePushDownFilterUnion::new())),
-            RuleID::PushDownFilterEvalScalar => Ok(Box::new(RulePushDownFilterEvalScalar::new())),
+            RuleID::PushDownFilterEvalScalar => {
+                Ok(Box::new(RulePushDownFilterEvalScalar::new(metadata)))
+            }
             RuleID::PushDownFilterJoin => Ok(Box::new(RulePushDownFilterJoin::new(metadata))),
             RuleID::PushDownFilterScan => Ok(Box::new(RulePushDownFilterScan::new(metadata))),
             RuleID::PushDownFilterSort => Ok(Box::new(RulePushDownFilterSort::new())),

src/query/sql/src/planner/optimizer/rule/rewrite/rule_push_down_prewhere.rs

@@ -12,7 +12,9 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+use common_exception::ErrorCode;
 use common_exception::Result;
+use common_expression::TableSchemaRef;
 
 use crate::optimizer::rule::Rule;
 use crate::optimizer::ColumnSet;
@@ -24,19 +26,19 @@ use crate::plans::Prewhere;
 use crate::plans::RelOp;
 use crate::plans::ScalarExpr;
 use crate::plans::Scan;
+use crate::IndexType;
 use crate::MetadataRef;
 
 pub struct RulePushDownPrewhere {
     id: RuleID,
-    metadata: MetadataRef,
     patterns: Vec<SExpr>,
+    metadata: MetadataRef,
 }
 
 impl RulePushDownPrewhere {
     pub fn new(metadata: MetadataRef) -> Self {
         Self {
             id: RuleID::PushDownPrewhere,
-            metadata,
             patterns: vec![SExpr::create_unary(
                 PatternPlan {
                     plan_type: RelOp::Filter,
@@ -49,56 +51,78 @@ impl RulePushDownPrewhere {
                     .into(),
                 ),
             )],
+            metadata,
         }
     }
 
     /// will throw error if the bound column ref is not in the table, such as subquery
-    fn collect_columns_impl(expr: &ScalarExpr, columns: &mut ColumnSet) -> Option<()> {
+    fn collect_columns_impl(
+        table_index: IndexType,
+        schema: &TableSchemaRef,
+        expr: &ScalarExpr,
+        columns: &mut ColumnSet,
+    ) -> Result<()> {
         match expr {
             ScalarExpr::BoundColumnRef(column) => {
-                column.column.table_name.as_ref()?;
-                columns.insert(column.column.index);
-                Some(())
+                if let Some(index) = &column.column.table_index {
+                    if table_index == *index
+                        && schema.index_of(column.column.column_name.as_str()).is_ok()
+                    {
+                        columns.insert(column.column.index);
+                        return Ok(());
+                    }
+                }
+                return Err(ErrorCode::Unimplemented("Column is not in the table"));
             }
             ScalarExpr::AndExpr(and) => {
-                Self::collect_columns_impl(and.left.as_ref(), columns)?;
-                Self::collect_columns_impl(and.right.as_ref(), columns)
+                Self::collect_columns_impl(table_index, schema, and.left.as_ref(), columns)?;
+                Self::collect_columns_impl(table_index, schema, and.right.as_ref(), columns)?;
             }
             ScalarExpr::OrExpr(or) => {
-                Self::collect_columns_impl(or.left.as_ref(), columns)?;
-                Self::collect_columns_impl(or.right.as_ref(), columns)
+                Self::collect_columns_impl(table_index, schema, or.left.as_ref(), columns)?;
+                Self::collect_columns_impl(table_index, schema, or.right.as_ref(), columns)?;
+            }
+            ScalarExpr::NotExpr(not) => {
+                Self::collect_columns_impl(table_index, schema, not.argument.as_ref(), columns)?
             }
-            ScalarExpr::NotExpr(not) => Self::collect_columns_impl(not.argument.as_ref(), columns),
             ScalarExpr::ComparisonExpr(cmp) => {
-                Self::collect_columns_impl(cmp.left.as_ref(), columns)?;
-                Self::collect_columns_impl(cmp.right.as_ref(), columns)
+                Self::collect_columns_impl(table_index, schema, cmp.left.as_ref(), columns)?;
+                Self::collect_columns_impl(table_index, schema, cmp.right.as_ref(), columns)?;
             }
             ScalarExpr::FunctionCall(func) => {
                 for arg in func.arguments.iter() {
-                    Self::collect_columns_impl(arg, columns)?;
+                    Self::collect_columns_impl(table_index, schema, arg, columns)?;
                 }
-                Some(())
             }
             ScalarExpr::CastExpr(cast) => {
-                Self::collect_columns_impl(cast.argument.as_ref(), columns)
+                Self::collect_columns_impl(table_index, schema, cast.argument.as_ref(), columns)?;
+            }
+            ScalarExpr::ConstantExpr(_) => {}
+            _ => {
+                // SubqueryExpr and AggregateFunction will not appear in Filter-LogicalGet
+                return Err(ErrorCode::Unimplemented(format!(
+                    "Prewhere don't support expr {:?}",
+                    expr
+                )));
             }
-            // 1. ConstantExpr is not collected.
-            // 2. SubqueryExpr and AggregateFunction will not appear in Filter-LogicalGet
-            _ => None,
         }
+        Ok(())
     }
 
     // analyze if the expression can be moved to prewhere
-    fn collect_columns(expr: &ScalarExpr) -> Option<ColumnSet> {
+    fn collect_columns(
+        table_index: IndexType,
+        schema: &TableSchemaRef,
+        expr: &ScalarExpr,
+    ) -> Option<ColumnSet> {
         let mut columns = ColumnSet::new();
         // columns in subqueries are not considered
-        Self::collect_columns_impl(expr, &mut columns)?;
+        Self::collect_columns_impl(table_index, schema, expr, &mut columns).ok()?;
 
         Some(columns)
     }
 
     pub fn prewhere_optimize(&self, s_expr: &SExpr) -> Result<SExpr> {
-        let filter: Filter = s_expr.plan().clone().try_into()?;
         let mut get: Scan = s_expr.child(0)?.plan().clone().try_into()?;
         let metadata = self.metadata.read().clone();
 
@@ -107,13 +131,14 @@ impl RulePushDownPrewhere {
             // cannot optimize
             return Ok(s_expr.clone());
         }
+        let filter: Filter = s_expr.plan().clone().try_into()?;
 
         let mut prewhere_columns = ColumnSet::new();
         let mut prewhere_pred = Vec::new();
 
         // filter.predicates are already split by AND
         for pred in filter.predicates.iter() {
-            match Self::collect_columns(pred) {
+            match Self::collect_columns(get.table_index, &table.schema(), pred) {
                 Some(columns) => {
                     prewhere_pred.push(pred.clone());
                     prewhere_columns.extend(&columns);

tests/sqllogictests/suites/base/06_show/06_0015_show_columns

@@ -43,7 +43,7 @@ EvalScalar
         ├── expressions: [columns.name (#0), columns.database (#1), columns.table (#2), columns.data_type (#4), columns.default_expression (#6), columns.is_nullable (#7), NULL, NULL, NULL, NULL, NULL]
         ├── estimated rows: 0.00
         └── Filter
-            ├── filters: [columns.table_schema (#1) = "showcolumn", columns.table_name (#2) = "t3"]
+            ├── filters: [columns.database (#1) = "showcolumn", columns.table (#2) = "t3"]
             ├── estimated rows: 0.00
             └── TableScan
                 ├── table: default.system.columns

tests/sqllogictests/suites/mode/standalone/explain/explain.test

@@ -976,6 +976,9 @@ Query (children 1)
                 ├── prune_page=>Literal Boolean(true)
                 └── refresh_meta_cache=>Literal Boolean(true)
 
+statement ok
+drop table if exists t4
+
 statement ok
 create table t4(a int, b string);
 
@@ -997,6 +1000,9 @@ EvalScalar
         ├── push downs: [filters: [and_filters(CAST(t4.a (#0) = 1 AS Boolean NULL), TRY_CAST(get(try_parse_json(t4.b (#1)), "bb") AS String NULL) = "xx")], limit: NONE]
         └── estimated rows: 0.00
 
+statement ok
+drop view if exists v4
+
 statement ok
 create view v4 as select a as a, try_cast(get(try_parse_json(b), 'bb') as varchar) as b from t4;
 
@@ -1031,7 +1037,7 @@ EvalScalar
     ├── expressions: [t4.a (#0), TRY_CAST(get(try_parse_json(t4.b (#1)), "bb") AS String NULL)]
     ├── estimated rows: 0.00
     └── Filter
-        ├── filters: [v4.a (#0) > 100]
+        ├── filters: [t4.a (#0) > 100]
         ├── estimated rows: 0.00
         └── TableScan
             ├── table: default.default.t4