Move unbound placeholder error to physical-expr

Previously, these errors would occurr during optimization. Now that we
allow unbound placeholders through the optimizer they fail when creating
the physical plan instead.

Author: davisp

datafusion/core/tests/dataframe/mod.rs

@@ -1985,6 +1985,7 @@ async fn test_array_agg() -> Result<()> {
 async fn test_dataframe_placeholder_missing_param_values() -> Result<()> {
     let ctx = SessionContext::new();
 
+    // Creating LogicalPlans with placeholders should work.
     let df = ctx
         .read_empty()
         .unwrap()
@@ -2006,17 +2007,16 @@ async fn test_dataframe_placeholder_missing_param_values() -> Result<()> {
         "\n\nexpected:\n\n{expected:#?}\nactual:\n\n{actual:#?}\n\n"
     );
 
-    // The placeholder is not replaced with a value,
-    // so the filter data type is not know, i.e. a = $0.
-    // Therefore, the optimization fails.
-    let optimized_plan = ctx.state().optimize(logical_plan);
-    assert!(optimized_plan.is_err());
-    assert!(optimized_plan
-        .unwrap_err()
-        .to_string()
-        .contains("Placeholder type for '$0' could not be resolved. Make sure that the placeholder is bound to a concrete type, e.g. by providing parameter values."));
-
-    // Prodiving a parameter value should resolve the error
+    // Executing LogicalPlans with placeholders that don't have bound values
+    // should fail.
+    let results = df.collect().await;
+    let err_mesg = results.unwrap_err().strip_backtrace();
+    assert_eq!(
+        err_mesg,
+        "Execution error: Placeholder '$0' was not provided a value for execution."
+    );
+
+    // Providing a parameter value should resolve the error
     let df = ctx
         .read_empty()
         .unwrap()
@@ -2040,12 +2040,14 @@ async fn test_dataframe_placeholder_missing_param_values() -> Result<()> {
         "\n\nexpected:\n\n{expected:#?}\nactual:\n\n{actual:#?}\n\n"
     );
 
-    let optimized_plan = ctx.state().optimize(logical_plan);
-    assert!(optimized_plan.is_ok());
+    // N.B., the test is basically `SELECT 1 as a WHERE a = 3;` which returns no results.
+    #[rustfmt::skip]
+    let expected = [
+        "++",
+        "++"
+    ];
 
-    let actual = optimized_plan.unwrap().display_indent_schema().to_string();
-    let expected = "EmptyRelation [a:Int32]";
-    assert_eq!(expected, actual);
+    assert_batches_eq!(expected, &df.collect().await.unwrap());
 
     Ok(())
 }
@@ -2054,27 +2056,33 @@ async fn test_dataframe_placeholder_missing_param_values() -> Result<()> {
 async fn test_dataframe_placeholder_column_parameter() -> Result<()> {
     let ctx = SessionContext::new();
 
+    // Creating LogicalPlans with placeholders should work
     let df = ctx.read_empty().unwrap().select_exprs(&["$1"]).unwrap();
-
     let logical_plan = df.logical_plan();
     let formatted = logical_plan.display_indent_schema().to_string();
     let actual: Vec<&str> = formatted.trim().lines().collect();
-    let expected = vec!["Projection: $1 [$1:Null;N]", "  EmptyRelation []"];
+
+    #[rustfmt::skip]
+    let expected = vec![
+        "Projection: $1 [$1:Null;N]",
+        "  EmptyRelation []"
+    ];
+
     assert_eq!(
         expected, actual,
         "\n\nexpected:\n\n{expected:#?}\nactual:\n\n{actual:#?}\n\n"
     );
 
-    // The placeholder is not replaced with a value,
-    // so the filter data type is not known, i.e. a = $0.
-    // Therefore, the optimization fails.
-    let optimized_plan = ctx.state().optimize(logical_plan);
-    assert!(optimized_plan
-        .unwrap_err()
-        .to_string()
-        .contains("Placeholder type for '$1' could not be resolved. Make sure that the placeholder is bound to a concrete type, e.g. by providing parameter values."));
+    // Executing LogicalPlans with placeholders that don't have bound values
+    // should fail.
+    let results = df.collect().await;
+    let err_mesg = results.unwrap_err().strip_backtrace();
+    assert_eq!(
+        err_mesg,
+        "Execution error: Placeholder '$1' was not provided a value for execution."
+    );
 
-    // Prodiving a parameter value should resolve the error
+    // Providing a parameter value should resolve the error
     let df = ctx
         .read_empty()
         .unwrap()
@@ -2095,16 +2103,16 @@ async fn test_dataframe_placeholder_column_parameter() -> Result<()> {
         "\n\nexpected:\n\n{expected:#?}\nactual:\n\n{actual:#?}\n\n"
     );
 
-    let optimized_plan = ctx.state().optimize(logical_plan);
-    assert!(optimized_plan.is_ok());
-
-    let formatted = optimized_plan.unwrap().display_indent_schema().to_string();
-    let actual: Vec<&str> = formatted.trim().lines().collect();
-    let expected = vec![
-        "Projection: Int32(3) AS $1 [$1:Int32]",
-        "  EmptyRelation []",
+    #[rustfmt::skip]
+    let expected = [
+        "+----+",
+        "| $1 |",
+        "+----+",
+        "| 3  |",
+        "+----+"
     ];
-    assert_eq!(expected, actual);
+
+    assert_batches_eq!(expected, &df.collect().await.unwrap());
 
     Ok(())
 }
@@ -2113,6 +2121,7 @@ async fn test_dataframe_placeholder_column_parameter() -> Result<()> {
 async fn test_dataframe_placeholder_like_expression() -> Result<()> {
     let ctx = SessionContext::new();
 
+    // Creating LogicalPlans with placeholders should work
     let df = ctx
         .read_empty()
         .unwrap()
@@ -2134,16 +2143,16 @@ async fn test_dataframe_placeholder_like_expression() -> Result<()> {
         "\n\nexpected:\n\n{expected:#?}\nactual:\n\n{actual:#?}\n\n"
     );
 
-    // The placeholder is not replaced with a value,
-    // so the filter data type is not known, i.e. a = $0.
-    // Therefore, the optimization fails.
-    let optimized_plan = ctx.state().optimize(logical_plan);
-    assert!(optimized_plan
-        .unwrap_err()
-        .to_string()
-        .contains("Placeholder type for '$1' could not be resolved. Make sure that the placeholder is bound to a concrete type, e.g. by providing parameter values."));
+    // Executing LogicalPlans with placeholders that don't have bound values
+    // should fail.
+    let results = df.collect().await;
+    let err_mesg = results.unwrap_err().strip_backtrace();
+    assert_eq!(
+        err_mesg,
+        "Execution error: Placeholder '$1' was not provided a value for execution."
+    );
 
-    // Prodiving a parameter value should resolve the error
+    // Providing a parameter value should resolve the error
     let df = ctx
         .read_empty()
         .unwrap()
@@ -2167,16 +2176,16 @@ async fn test_dataframe_placeholder_like_expression() -> Result<()> {
         "\n\nexpected:\n\n{expected:#?}\nactual:\n\n{actual:#?}\n\n"
     );
 
-    let optimized_plan = ctx.state().optimize(logical_plan);
-    assert!(optimized_plan.is_ok());
-
-    let formatted = optimized_plan.unwrap().display_indent_schema().to_string();
-    let actual: Vec<&str> = formatted.trim().lines().collect();
-    let expected = vec![
-        "Projection: Utf8(\"foo\") AS a [a:Utf8]",
-        "  EmptyRelation []",
+    #[rustfmt::skip]
+    let expected = [
+        "+-----+",
+        "| a   |",
+        "+-----+",
+        "| foo |",
+        "+-----+"
     ];
-    assert_eq!(expected, actual);
+
+    assert_batches_eq!(expected, &df.collect().await.unwrap());
 
     Ok(())
 }

datafusion/core/tests/sql/select.rs

@@ -229,6 +229,98 @@ async fn test_parameter_invalid_types() -> Result<()> {
     Ok(())
 }
 
+#[tokio::test]
+async fn test_positional_parameter_not_bound() -> Result<()> {
+    let ctx = SessionContext::new();
+    let signed_ints: Int32Array = vec![-1, 0, 1].into();
+    let unsigned_ints: UInt64Array = vec![1, 2, 3].into();
+    let batch = RecordBatch::try_from_iter(vec![
+        ("signed", Arc::new(signed_ints) as ArrayRef),
+        ("unsigned", Arc::new(unsigned_ints) as ArrayRef),
+    ])?;
+    ctx.register_batch("test", batch)?;
+
+    let query = "SELECT signed, unsigned FROM test \
+            WHERE $1 >= signed AND signed <= $2 \
+            AND unsigned <= $3 AND unsigned = $4";
+
+    let results = ctx.sql(query).await?.collect().await;
+
+    assert_eq!(
+        results.unwrap_err().strip_backtrace(),
+        "Execution error: Placeholder '$1' was not provided a value for execution."
+    );
+
+    let results = ctx
+        .sql(query)
+        .await?
+        .with_param_values(vec![
+            ScalarValue::from(4_i32),
+            ScalarValue::from(-1_i64),
+            ScalarValue::from(2_i32),
+            ScalarValue::from("1"),
+        ])?
+        .collect()
+        .await?;
+
+    let expected = [
+        "+--------+----------+",
+        "| signed | unsigned |",
+        "+--------+----------+",
+        "| -1     | 1        |",
+        "+--------+----------+",
+    ];
+    assert_batches_sorted_eq!(expected, &results);
+
+    Ok(())
+}
+
+#[tokio::test]
+async fn test_named_parameter_not_bound() -> Result<()> {
+    let ctx = SessionContext::new();
+    let signed_ints: Int32Array = vec![-1, 0, 1].into();
+    let unsigned_ints: UInt64Array = vec![1, 2, 3].into();
+    let batch = RecordBatch::try_from_iter(vec![
+        ("signed", Arc::new(signed_ints) as ArrayRef),
+        ("unsigned", Arc::new(unsigned_ints) as ArrayRef),
+    ])?;
+    ctx.register_batch("test", batch)?;
+
+    let query = "SELECT signed, unsigned FROM test \
+            WHERE $foo >= signed AND signed <= $bar \
+            AND unsigned <= $baz AND unsigned = $str";
+
+    let results = ctx.sql(query).await?.collect().await;
+
+    assert_eq!(
+        results.unwrap_err().strip_backtrace(),
+        "Execution error: Placeholder '$foo' was not provided a value for execution."
+    );
+
+    let results = ctx
+        .sql(query)
+        .await?
+        .with_param_values(vec![
+            ("foo", ScalarValue::from(4_i32)),
+            ("bar", ScalarValue::from(-1_i64)),
+            ("baz", ScalarValue::from(2_i32)),
+            ("str", ScalarValue::from("1")),
+        ])?
+        .collect()
+        .await?;
+
+    let expected = [
+        "+--------+----------+",
+        "| signed | unsigned |",
+        "+--------+----------+",
+        "| -1     | 1        |",
+        "+--------+----------+",
+    ];
+    assert_batches_sorted_eq!(expected, &results);
+
+    Ok(())
+}
+
 #[tokio::test]
 async fn test_version_function() {
     let expected_version = format!(

datafusion/expr/src/logical_plan/plan.rs

@@ -1501,26 +1501,6 @@ impl LogicalPlan {
         .map(|_| param_types)
     }
 
-    /// Walk the logical plan, error out if any `Placeholder` tokens with DataType::Null
-    pub fn validate_parameter_types(&self) -> Result<(), DataFusionError> {
-        self.apply_with_subqueries(|plan| {
-            plan.apply_expressions(|expr| {
-                expr.apply(|expr| {
-                    if let Expr::Placeholder(Placeholder { id, data_type }) = expr {
-                        if data_type.is_none() {
-                            plan_err!(
-                                "Placeholder type for '{id}' could not be resolved. Make sure that the \
-                                placeholder is bound to a concrete type, e.g. by providing \
-                                parameter values.")?;
-                        }
-                    }
-                    Ok(TreeNodeRecursion::Continue)
-                })
-            })
-        })
-        .map(|_| ())
-    }
-
     // ------------
     // Various implementations for printing out LogicalPlans
     // ------------

datafusion/optimizer/src/optimizer.rs

@@ -357,7 +357,6 @@ impl Optimizer {
     {
         let start_time = Instant::now();
         let options = config.options();
-        plan.validate_parameter_types()?;
         let mut new_plan = plan;
 
         let mut previous_plans = HashSet::with_capacity(16);

datafusion/physical-expr/src/planner.rs

@@ -28,7 +28,7 @@ use datafusion_common::{
     exec_err, not_impl_err, plan_err, DFSchema, Result, ScalarValue, ToDFSchema,
 };
 use datafusion_expr::execution_props::ExecutionProps;
-use datafusion_expr::expr::{Alias, Cast, InList, ScalarFunction};
+use datafusion_expr::expr::{Alias, Cast, InList, Placeholder, ScalarFunction};
 use datafusion_expr::var_provider::is_system_variables;
 use datafusion_expr::var_provider::VarType;
 use datafusion_expr::{
@@ -361,6 +361,9 @@ pub fn create_physical_expr(
                 expressions::in_list(value_expr, list_exprs, negated, input_schema)
             }
         },
+        Expr::Placeholder(Placeholder { id, .. }) => {
+            exec_err!("Placeholder '{id}' was not provided a value for execution.")
+        }
         other => {
             not_impl_err!("Physical plan does not support logical expression {other:?}")
         }

datafusion/sqllogictest/test_files/prepare.slt

@@ -88,6 +88,16 @@ EXECUTE my_plan('j%');
 statement ok
 DEALLOCATE my_plan
 
+# Check for missing parameters
+statement ok
+PREPARE my_plan AS SELECT * FROM person WHERE id < $1;
+
+statement error No value found for placeholder with id $1
+EXECUTE my_plan
+
+statement ok
+DEALLOCATE my_plan
+
 statement ok
 PREPARE my_plan(STRING, STRING) AS SELECT * FROM (VALUES(1, $1), (2, $2)) AS t (num, letter);