feat: placeholders in execution plans

datafusion/physical-expr/src/expressions/mod.rs

@@ -32,6 +32,7 @@ mod literal;
 mod negative;
 mod no_op;
 mod not;
+mod placeholder;
 mod try_cast;
 mod unknown_column;
 
@@ -54,5 +55,6 @@ pub use literal::{Literal, lit};
 pub use negative::{NegativeExpr, negative};
 pub use no_op::NoOp;
 pub use not::{NotExpr, not};
+pub use placeholder::{PlaceholderExpr, has_placeholders, placeholder};
 pub use try_cast::{TryCastExpr, try_cast};
 pub use unknown_column::UnKnownColumn;

datafusion/physical-expr/src/expressions/placeholder.rs

@@ -0,0 +1,126 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you 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.
+
+//! Placeholder expression.
+
+use std::{
+    any::Any,
+    fmt::{self, Formatter},
+    sync::Arc,
+};
+
+use arrow::{
+    array::RecordBatch,
+    datatypes::{DataType, Field, FieldRef, Schema},
+};
+use datafusion_common::{
+    DataFusionError, Result, exec_datafusion_err, tree_node::TreeNode,
+};
+use datafusion_expr::ColumnarValue;
+use datafusion_physical_expr_common::physical_expr::PhysicalExpr;
+use std::hash::Hash;
+
+/// Physical expression representing a placeholder parameter (e.g., $1, $2, or named parameters) in
+/// the physical plan.
+///
+/// This expression serves as a placeholder that will be resolved to a literal value during
+/// execution. It should not be evaluated directly.
+#[derive(Debug, PartialEq, Eq, Hash)]
+pub struct PlaceholderExpr {
+    /// Placeholder id, e.g. $1 or $a.
+    pub id: String,
+    /// Derived from expression where placeholder is met.
+    pub field: Option<FieldRef>,
+}
+
+impl PlaceholderExpr {
+    /// Create a new placeholder expression.
+    pub fn new(id: String, data_type: DataType) -> Self {
+        let field = Arc::new(Field::new("", data_type, true));
+        Self::new_with_field(id, field)
+    }
+
+    /// Create a new placeholders expression from a field.
+    pub fn new_with_field(id: String, field: FieldRef) -> Self {
+        Self {
+            id,
+            field: Some(field),
+        }
+    }
+
+    /// Create a new placeholder expression without a specified data type.
+    pub fn new_without_data_type(id: String) -> Self {
+        Self { id, field: None }
+    }
+
+    fn execution_error(&self) -> DataFusionError {
+        exec_datafusion_err!(
+            "Placeholder '{}' was not provided a value for execution.",
+            self.id
+        )
+    }
+}
+
+/// Create a placeholder expression.
+pub fn placeholder<I: Into<String>>(id: I, data_type: DataType) -> Arc<dyn PhysicalExpr> {
+    Arc::new(PlaceholderExpr::new(id.into(), data_type))
+}
+
+/// Returns `true` if expression has placeholders.
+pub fn has_placeholders(expr: &Arc<dyn PhysicalExpr>) -> bool {
+    expr.exists(|e| Ok(e.as_any().is::<PlaceholderExpr>()))
+        .expect("do not return errors")
+}
+
+impl fmt::Display for PlaceholderExpr {
+    fn fmt(&self, f: &mut Formatter) -> fmt::Result {
+        write!(f, "{}", self.id)
+    }
+}
+
+impl PhysicalExpr for PlaceholderExpr {
+    fn as_any(&self) -> &dyn Any {
+        self
+    }
+
+    fn return_field(&self, _input_schema: &Schema) -> Result<FieldRef> {
+        self.field
+            .as_ref()
+            .map(Arc::clone)
+            .ok_or_else(|| self.execution_error())
+    }
+
+    fn evaluate(&self, _batch: &RecordBatch) -> Result<ColumnarValue> {
+        Err(self.execution_error())
+    }
+
+    fn children(&self) -> Vec<&Arc<dyn PhysicalExpr>> {
+        vec![]
+    }
+
+    fn with_new_children(
+        self: Arc<Self>,
+        children: Vec<Arc<dyn PhysicalExpr>>,
+    ) -> Result<Arc<dyn PhysicalExpr>> {
+        assert!(children.is_empty());
+        Ok(self)
+    }
+
+    fn fmt_sql(&self, f: &mut Formatter<'_>) -> fmt::Result {
+        fmt::Display::fmt(self, f)
+    }
+}

datafusion/physical-expr/src/planner.rs

@@ -20,10 +20,10 @@ use std::sync::Arc;
 use crate::ScalarFunctionExpr;
 use crate::{
     PhysicalExpr,
-    expressions::{self, Column, Literal, binary, like, similar_to},
+    expressions::{self, Column, Literal, PlaceholderExpr, binary, like, similar_to},
 };
 
-use arrow::datatypes::Schema;
+use arrow::datatypes::{DataType, Schema};
 use datafusion_common::config::ConfigOptions;
 use datafusion_common::metadata::FieldMetadata;
 use datafusion_common::{
@@ -288,16 +288,28 @@ pub fn create_physical_expr(
                 };
             Ok(expressions::case(expr, when_then_expr, else_expr)?)
         }
-        Expr::Cast(Cast { expr, data_type }) => expressions::cast(
-            create_physical_expr(expr, input_dfschema, execution_props)?,
-            input_schema,
-            data_type.clone(),
-        ),
-        Expr::TryCast(TryCast { expr, data_type }) => expressions::try_cast(
-            create_physical_expr(expr, input_dfschema, execution_props)?,
-            input_schema,
-            data_type.clone(),
-        ),
+        Expr::Cast(Cast { expr, data_type }) => {
+            let mut expr = create_physical_expr(expr, input_dfschema, execution_props)?;
+            if let Some(placeholder) = expr.as_any().downcast_ref::<PlaceholderExpr>()
+                && placeholder.field.is_none()
+            {
+                expr = expressions::placeholder(&placeholder.id, data_type.clone());
+            }
+
+            expressions::cast(expr, input_schema, data_type.clone())
+        }
+        Expr::TryCast(TryCast { expr, data_type }) => {
+            let mut expr = create_physical_expr(expr, input_dfschema, execution_props)?;
+            if let Some(placeholder) = expr.as_any().downcast_ref::<PlaceholderExpr>()
+                && placeholder.field.is_none()
+            {
+                // To maintain try_cast behavior, we initially resolve the placeholder with the
+                // Utf8 data type.
+                expr = expressions::placeholder(&placeholder.id, DataType::Utf8);
+            }
+
+            expressions::try_cast(expr, input_schema, data_type.clone())
+        }
         Expr::Not(expr) => {
             expressions::not(create_physical_expr(expr, input_dfschema, execution_props)?)
         }
@@ -381,9 +393,13 @@ 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.")
-        }
+        Expr::Placeholder(Placeholder { id, field }) => match field {
+            Some(field) => Ok(Arc::new(PlaceholderExpr::new_with_field(
+                id.clone(),
+                Arc::clone(field),
+            ))),
+            None => Ok(Arc::new(PlaceholderExpr::new_without_data_type(id.clone()))),
+        },
         other => {
             not_impl_err!("Physical plan does not support logical expression {other:?}")
         }

datafusion/physical-expr/src/simplifier/const_evaluator.rs

@@ -27,7 +27,7 @@ use datafusion_common::{Result, ScalarValue};
 use datafusion_expr_common::columnar_value::ColumnarValue;
 
 use crate::PhysicalExpr;
-use crate::expressions::{Column, Literal};
+use crate::expressions::{Column, Literal, PlaceholderExpr};
 
 /// Simplify expressions that consist only of literals by evaluating them.
 ///
@@ -81,7 +81,10 @@ fn can_evaluate_as_constant(expr: &Arc<dyn PhysicalExpr>) -> bool {
     let mut can_evaluate = true;
 
     expr.apply(|e| {
-        if e.as_any().is::<Column>() || e.is_volatile_node() {
+        if e.as_any().is::<Column>()
+            || e.is_volatile_node()
+            || e.as_any().is::<PlaceholderExpr>()
+        {
             can_evaluate = false;
             Ok(TreeNodeRecursion::Stop)
         } else {

datafusion/proto/proto/datafusion.proto

@@ -896,6 +896,8 @@ message PhysicalExprNode {
     UnknownColumn unknown_column = 20;
 
     PhysicalHashExprNode hash_expr = 21;
+
+    PhysicalPlaceholderNode placeholder_expr = 22;
   }
 }
 
@@ -1023,6 +1025,11 @@ message PhysicalHashExprNode {
   string description = 6;
 }
 
+message PhysicalPlaceholderNode {
+  string id = 1;
+  optional datafusion_common.Field field = 2;
+}
+
 message FilterExecNode {
   PhysicalPlanNode input = 1;
   PhysicalExprNode expr = 2;