Add Filter::try_new with validation

benchmarks/expected-plans/q6.txt

@@ -1,5 +1,5 @@
 Projection: SUM(lineitem.l_extendedprice * lineitem.l_discount) AS revenue
   Aggregate: groupBy=[[]], aggr=[[SUM(lineitem.l_extendedprice * lineitem.l_discount)]]
     Projection: CAST(lineitem.l_discount AS Decimal128(30, 15)) AS CAST(lineitem.l_discount AS Decimal128(30, 15))lineitem.l_discount, lineitem.l_quantity, lineitem.l_extendedprice, lineitem.l_discount, lineitem.l_shipdate
-      Filter: lineitem.l_shipdate >= Date32("8766") AND lineitem.l_shipdate < Date32("9131") AND CAST(lineitem.l_discount AS Decimal128(30, 15)) AS lineitem.l_discount >= Decimal128(Some(49999999999999),30,15) AND CAST(lineitem.l_discount AS Decimal128(30, 15)) AS lineitem.l_discount <= Decimal128(Some(69999999999999),30,15) AND lineitem.l_quantity < Decimal128(Some(2400),15,2)
+      Filter: lineitem.l_shipdate >= Date32("8766") AND lineitem.l_shipdate < Date32("9131") AND CAST(lineitem.l_discount AS Decimal128(30, 15)) >= Decimal128(Some(49999999999999),30,15) AND CAST(lineitem.l_discount AS Decimal128(30, 15)) <= Decimal128(Some(69999999999999),30,15) AND lineitem.l_quantity < Decimal128(Some(2400),15,2)
         TableScan: lineitem projection=[l_quantity, l_extendedprice, l_discount, l_shipdate]

datafusion/core/src/physical_plan/planner.rs

@@ -28,8 +28,8 @@ use crate::datasource::source_as_provider;
 use crate::execution::context::{ExecutionProps, SessionState};
 use crate::logical_expr::utils::generate_sort_key;
 use crate::logical_expr::{
-    Aggregate, Distinct, EmptyRelation, Filter, Join, Projection, Sort, SubqueryAlias,
-    TableScan, Window,
+    Aggregate, Distinct, EmptyRelation, Join, Projection, Sort, SubqueryAlias, TableScan,
+    Window,
 };
 use crate::logical_plan::{
     unalias, unnormalize_cols, CrossJoin, DFSchema, Expr, LogicalPlan,
@@ -756,15 +756,13 @@ impl DefaultPhysicalPlanner {
                         input_exec,
                     )?) )
                 }
-                LogicalPlan::Filter(Filter {
-                    input, predicate, ..
-                }) => {
-                    let physical_input = self.create_initial_plan(input, session_state).await?;
+                LogicalPlan::Filter(filter) => {
+                    let physical_input = self.create_initial_plan(filter.input(), session_state).await?;
                     let input_schema = physical_input.as_ref().schema();
-                    let input_dfschema = input.as_ref().schema();
+                    let input_dfschema = filter.input().schema();
 
                     let runtime_expr = self.create_physical_expr(
-                        predicate,
+                        filter.predicate(),
                         input_dfschema,
                         &input_schema,
                         session_state,
@@ -1696,16 +1694,18 @@ mod tests {
     use arrow::record_batch::RecordBatch;
     use datafusion_common::{DFField, DFSchema, DFSchemaRef};
     use datafusion_expr::expr::GroupingSet;
-    use datafusion_expr::sum;
-    use datafusion_expr::{col, lit};
+    use datafusion_expr::{col, lit, sum};
     use fmt::Debug;
     use std::collections::HashMap;
     use std::convert::TryFrom;
     use std::{any::Any, fmt};
 
     fn make_session_state() -> SessionState {
         let runtime = Arc::new(RuntimeEnv::default());
-        SessionState::with_config_rt(SessionConfig::new(), runtime)
+        let config = SessionConfig::new();
+        // TODO we should really test that no optimizer rules are failing here
+        // let config = config.set_bool(crate::config::OPT_OPTIMIZER_SKIP_FAILED_RULES, false);
+        SessionState::with_config_rt(config, runtime)
     }
 
     async fn plan(logical_plan: &LogicalPlan) -> Result<Arc<dyn ExecutionPlan>> {
@@ -1972,6 +1972,11 @@ mod tests {
         let expected = "expr: [(InListExpr { expr: Column { name: \"c1\", index: 0 }, list: [Literal { value: Utf8(\"a\") }, Literal { value: Utf8(\"1\") }], negated: false, set: None }";
         assert!(format!("{:?}", execution_plan).contains(expected));
 
+        Ok(())
+    }
+
+    #[tokio::test]
+    async fn in_list_types_struct_literal() -> Result<()> {
         // expression: "a in (struct::null, 'a')"
         let list = vec![struct_literal(), lit("a")];
 

datafusion/core/tests/sql/select.rs

@@ -1211,14 +1211,19 @@ async fn boolean_literal() -> Result<()> {
 
 #[tokio::test]
 async fn unprojected_filter() {
-    let ctx = SessionContext::new();
+    let config = SessionConfig::new();
+    let ctx = SessionContext::with_config(config);
     let df = ctx.read_table(table_with_sequence(1, 3).unwrap()).unwrap();
 
     let df = df
-        .select(vec![col("i") + col("i")])
-        .unwrap()
         .filter(col("i").gt(lit(2)))
+        .unwrap()
+        .select(vec![col("i") + col("i")])
         .unwrap();
+
+    let plan = df.to_logical_plan().unwrap();
+    println!("{}", plan.display_indent());
+
     let results = df.collect().await.unwrap();
 
     let expected = vec![

datafusion/expr/Cargo.toml

@@ -38,4 +38,5 @@ path = "src/lib.rs"
 ahash = { version = "0.8", default-features = false, features = ["runtime-rng"] }
 arrow = { version = "24.0.0", default-features = false }
 datafusion-common = { path = "../common", version = "13.0.0" }
+log = "^0.4"
 sqlparser = "0.25"

datafusion/expr/src/expr.rs

@@ -486,6 +486,14 @@ impl Expr {
         Expr::Alias(Box::new(self), name.to_owned())
     }
 
+    /// Remove an alias from an expression if one exists.
+    pub fn unalias(self) -> Expr {
+        match self {
+            Expr::Alias(expr, _) => expr.as_ref().clone(),
+            _ => self,
+        }
+    }
+
     /// Return `self IN <list>` if `negated` is false, otherwise
     /// return `self NOT IN <list>`.a
     pub fn in_list(self, list: Vec<Expr>, negated: bool) -> Expr {

datafusion/expr/src/logical_plan/builder.rs

@@ -289,10 +289,10 @@ impl LogicalPlanBuilder {
     /// Apply a filter
     pub fn filter(&self, expr: impl Into<Expr>) -> Result<Self> {
         let expr = normalize_col(expr.into(), &self.plan)?;
-        Ok(Self::from(LogicalPlan::Filter(Filter {
-            predicate: expr,
-            input: Arc::new(self.plan.clone()),
-        })))
+        Ok(Self::from(LogicalPlan::Filter(Filter::try_new(
+            expr,
+            Arc::new(self.plan.clone()),
+        )?)))
     }
 
     /// Limit the number of rows returned

datafusion/expr/src/logical_plan/plan.rs

@@ -15,17 +15,17 @@
 // specific language governing permissions and limitations
 // under the License.
 
+///! Logical plan types
 use crate::logical_plan::builder::validate_unique_names;
 use crate::logical_plan::display::{GraphvizVisitor, IndentVisitor};
 use crate::logical_plan::extension::UserDefinedLogicalNode;
 use crate::utils::{
     exprlist_to_fields, grouping_set_expr_count, grouping_set_to_exprlist,
 };
-use crate::{Expr, TableProviderFilterPushDown, TableSource};
+use crate::{Expr, ExprSchemable, TableProviderFilterPushDown, TableSource};
 use arrow::datatypes::{DataType, Field, Schema, SchemaRef};
 use datafusion_common::{plan_err, Column, DFSchema, DFSchemaRef, DataFusionError};
 use std::collections::HashSet;
-///! Logical plan types
 use std::fmt::{self, Debug, Display, Formatter};
 use std::hash::{Hash, Hasher};
 use std::sync::Arc;
@@ -1148,18 +1148,59 @@ pub struct SubqueryAlias {
 #[derive(Clone)]
 pub struct Filter {
     /// The predicate expression, which must have Boolean type.
-    pub predicate: Expr,
+    predicate: Expr,
     /// The incoming logical plan
-    pub input: Arc<LogicalPlan>,
+    input: Arc<LogicalPlan>,
 }
 
 impl Filter {
+    /// Create a new filter operator.
+    pub fn try_new(
+        predicate: Expr,
+        input: Arc<LogicalPlan>,
+    ) -> datafusion_common::Result<Self> {
+        // Filter predicates must return a boolean value so we try and validate that here.
+        // Note that it is not always possible to resolve the predicate expression during plan
+        // construction (such as with correlated subqueries) so we make a best effort here and
+        // ignore errors resolving the expression against the schema.
+        if let Ok(predicate_type) = predicate.get_type(input.schema()) {
+            if predicate_type != DataType::Boolean {
+                return Err(DataFusionError::Plan(format!(
+                    "Cannot create filter with non-boolean predicate '{}' returning {}",
+                    predicate, predicate_type
+                )));
+            }
+        }
+
+        // filter predicates should not be aliased
+        if let Expr::Alias(expr, alias) = predicate {
+            return Err(DataFusionError::Plan(format!(
+                "Attempted to create Filter predicate with \
+                expression `{}` aliased as '{}'. Filter predicates should not be \
+                aliased.",
+                expr, alias
+            )));
+        }
+
+        Ok(Self { predicate, input })
+    }
+
     pub fn try_from_plan(plan: &LogicalPlan) -> datafusion_common::Result<&Filter> {
         match plan {
             LogicalPlan::Filter(it) => Ok(it),
             _ => plan_err!("Could not coerce into Filter!"),
         }
     }
+
+    /// Access the filter predicate expression
+    pub fn predicate(&self) -> &Expr {
+        &self.predicate
+    }
+
+    /// Access the filter input plan
+    pub fn input(&self) -> &Arc<LogicalPlan> {
+        &self.input
+    }
 }
 
 /// Window its input based on a set of window spec and window function (e.g. SUM or RANK)

datafusion/expr/src/utils.rs

@@ -17,6 +17,7 @@
 
 //! Expression utilities
 
+use crate::expr_rewriter::{ExprRewritable, ExprRewriter, RewriteRecursion};
 use crate::expr_visitor::{ExprVisitable, ExpressionVisitor, Recursion};
 use crate::logical_plan::builder::build_join_schema;
 use crate::logical_plan::{
@@ -380,10 +381,51 @@ pub fn from_plan(
                 .map(|s| s.to_vec())
                 .collect::<Vec<_>>(),
         })),
-        LogicalPlan::Filter { .. } => Ok(LogicalPlan::Filter(Filter {
-            predicate: expr[0].clone(),
-            input: Arc::new(inputs[0].clone()),
-        })),
+        LogicalPlan::Filter { .. } => {
+            assert_eq!(1, expr.len());
+            let predicate = expr[0].clone();
+
+            // filter predicates should not contain aliased expressions so we remove any aliases
+            // before this logic was added we would have aliases within filters such as for
+            // benchmark q6:
+            //
+            // lineitem.l_shipdate >= Date32(\"8766\")
+            // AND lineitem.l_shipdate < Date32(\"9131\")
+            // AND CAST(lineitem.l_discount AS Decimal128(30, 15)) AS lineitem.l_discount >=
+            // Decimal128(Some(49999999999999),30,15)
+            // AND CAST(lineitem.l_discount AS Decimal128(30, 15)) AS lineitem.l_discount <=
+            // Decimal128(Some(69999999999999),30,15)
+            // AND lineitem.l_quantity < Decimal128(Some(2400),15,2)
+
+            struct RemoveAliases {}
+
+            impl ExprRewriter for RemoveAliases {
+                fn pre_visit(&mut self, expr: &Expr) -> Result<RewriteRecursion> {
+                    match expr {
+                        Expr::Exists { .. }
+                        | Expr::ScalarSubquery(_)
+                        | Expr::InSubquery { .. } => {
+                            // subqueries could contain aliases so we don't recurse into those
+                            Ok(RewriteRecursion::Stop)
+                        }
+                        Expr::Alias(_, _) => Ok(RewriteRecursion::Mutate),
+                        _ => Ok(RewriteRecursion::Continue),
+                    }
+                }
+
+                fn mutate(&mut self, expr: Expr) -> Result<Expr> {
+                    Ok(expr.unalias())
+                }
+            }
+
+            let mut remove_aliases = RemoveAliases {};
+            let predicate = predicate.rewrite(&mut remove_aliases)?;
+
+            Ok(LogicalPlan::Filter(Filter::try_new(
+                predicate,
+                Arc::new(inputs[0].clone()),
+            )?))
+        }
         LogicalPlan::Repartition(Repartition {
             partitioning_scheme,
             ..

datafusion/optimizer/src/common_subexpr_eliminate.rs

@@ -114,7 +114,9 @@ fn optimize(
                 alias.clone(),
             )?))
         }
-        LogicalPlan::Filter(Filter { predicate, input }) => {
+        LogicalPlan::Filter(filter) => {
+            let input = filter.input();
+            let predicate = filter.predicate();
             let input_schema = Arc::clone(input.schema());
             let all_schemas: Vec<DFSchemaRef> =
                 plan.all_schemas().into_iter().cloned().collect();
@@ -131,16 +133,16 @@ fn optimize(
             let (mut new_expr, new_input) = rewrite_expr(
                 &[&[predicate.clone()]],
                 &[&[id_array]],
-                input,
+                filter.input(),
                 &mut expr_set,
                 optimizer_config,
             )?;
 
             if let Some(predicate) = pop_expr(&mut new_expr)?.pop() {
-                Ok(LogicalPlan::Filter(Filter {
+                Ok(LogicalPlan::Filter(Filter::try_new(
                     predicate,
-                    input: Arc::new(new_input),
-                }))
+                    Arc::new(new_input),
+                )?))
             } else {
                 Err(DataFusionError::Internal(
                     "Failed to pop predicate expr".to_string(),

datafusion/optimizer/src/decorrelate_where_exists.rs

@@ -76,19 +76,19 @@ impl OptimizerRule for DecorrelateWhereExists {
         optimizer_config: &mut OptimizerConfig,
     ) -> datafusion_common::Result<LogicalPlan> {
         match plan {
-            LogicalPlan::Filter(Filter {
-                predicate,
-                input: filter_input,
-            }) => {
+            LogicalPlan::Filter(filter) => {
+                let predicate = filter.predicate();
+                let filter_input = filter.input();
+
                 // Apply optimizer rule to current input
                 let optimized_input = self.optimize(filter_input, optimizer_config)?;
 
                 let (subqueries, other_exprs) =
                     self.extract_subquery_exprs(predicate, optimizer_config)?;
-                let optimized_plan = LogicalPlan::Filter(Filter {
-                    predicate: predicate.clone(),
-                    input: Arc::new(optimized_input),
-                });
+                let optimized_plan = LogicalPlan::Filter(Filter::try_new(
+                    predicate.clone(),
+                    Arc::new(optimized_input),
+                )?);
                 if subqueries.is_empty() {
                     // regular filter, no subquery exists clause here
                     return Ok(optimized_plan);
@@ -153,20 +153,21 @@ fn optimize_exists(
 
     // split into filters
     let mut subqry_filter_exprs = vec![];
-    split_conjunction(&subqry_filter.predicate, &mut subqry_filter_exprs);
+    split_conjunction(subqry_filter.predicate(), &mut subqry_filter_exprs);
     verify_not_disjunction(&subqry_filter_exprs)?;
 
     // Grab column names to join on
     let (col_exprs, other_subqry_exprs) =
-        find_join_exprs(subqry_filter_exprs, subqry_filter.input.schema())?;
+        find_join_exprs(subqry_filter_exprs, subqry_filter.input().schema())?;
     let (outer_cols, subqry_cols, join_filters) =
-        exprs_to_join_cols(&col_exprs, subqry_filter.input.schema(), false)?;
+        exprs_to_join_cols(&col_exprs, subqry_filter.input().schema(), false)?;
     if subqry_cols.is_empty() || outer_cols.is_empty() {
         plan_err!("cannot optimize non-correlated subquery")?;
     }
 
     // build subquery side of join - the thing the subquery was querying
-    let mut subqry_plan = LogicalPlanBuilder::from((*subqry_filter.input).clone());
+    let mut subqry_plan =
+        LogicalPlanBuilder::from(subqry_filter.input().as_ref().clone());
     if let Some(expr) = combine_filters(&other_subqry_exprs) {
         subqry_plan = subqry_plan.filter(expr)? // if the subquery had additional expressions, restore them
     }