feat: support prepare statement

datafusion/core/src/datasource/listing/helpers.rs

@@ -124,7 +124,8 @@ impl ExpressionVisitor for ApplicabilityVisitor<'_> {
             | Expr::Sort { .. }
             | Expr::WindowFunction { .. }
             | Expr::Wildcard
-            | Expr::QualifiedWildcard { .. } => {
+            | Expr::QualifiedWildcard { .. }
+            | Expr::Placeholder(_) => {
                 *self.is_applicable = false;
                 Recursion::Stop(self)
             }

datafusion/core/src/physical_plan/planner.rs

@@ -344,6 +344,9 @@ fn create_physical_name(e: &Expr, is_first_expr: bool) -> Result<String> {
         Expr::QualifiedWildcard { .. } => Err(DataFusionError::Internal(
             "Create physical name does not support qualified wildcard".to_string(),
         )),
+        Expr::Placeholder(_) => Err(DataFusionError::Internal(
+            "Create physical name does not support placeholder".to_string(),
+        )),
     }
 }
 
@@ -1031,6 +1034,14 @@ impl DefaultPhysicalPlanner {
                         "Unsupported logical plan: CreateExternalTable".to_string(),
                     ))
                 }
+                LogicalPlan::Prepare(_) => {
+                    // There is no default plan for "PREPARE" -- it must be
+                    // handled at a higher level (so that the appropriate
+                    // statement can be prepared)
+                    Err(DataFusionError::Internal(
+                        "Unsupported logical plan: Prepare".to_string(),
+                    ))
+                }
                 LogicalPlan::CreateCatalogSchema(_) => {
                     // There is no default plan for "CREATE SCHEMA".
                     // It must be handled at a higher level (so

datafusion/expr/src/expr.rs

@@ -89,6 +89,8 @@ pub enum Expr {
     Alias(Box<Expr>, String),
     /// A named reference to a qualified filed in a schema.
     Column(Column),
+    /// A place holder for parameters in a prepared statement.
+    Placeholder(String),
     /// A named reference to a variable in a registry.
     ScalarVariable(DataType, Vec<String>),
     /// A constant value.
@@ -528,6 +530,7 @@ impl Expr {
             Expr::Literal(..) => "Literal",
             Expr::Negative(..) => "Negative",
             Expr::Not(..) => "Not",
+            Expr::Placeholder(..) => "Placeholder",
             Expr::QualifiedWildcard { .. } => "QualifiedWildcard",
             Expr::ScalarFunction { .. } => "ScalarFunction",
             Expr::ScalarSubquery { .. } => "ScalarSubquery",
@@ -984,6 +987,7 @@ impl fmt::Debug for Expr {
                     )
                 }
             },
+            Expr::Placeholder(param) => write!(f, "{}", param),
         }
     }
 }
@@ -1269,6 +1273,7 @@ fn create_name(e: &Expr) -> Result<String> {
         Expr::QualifiedWildcard { .. } => Err(DataFusionError::Internal(
             "Create name does not support qualified wildcard".to_string(),
         )),
+        Expr::Placeholder(param) => Ok(format!("{}", param)),
     }
 }
 

datafusion/expr/src/expr_rewriter.rs

@@ -291,6 +291,7 @@ impl ExprRewritable for Expr {
                     key,
                 ))
             }
+            Expr::Placeholder(param) => Expr::Placeholder(param),
         };
 
         // now rewrite this expression itself

datafusion/expr/src/expr_schema.rs

@@ -127,6 +127,7 @@ impl ExprSchemable for Expr {
             Expr::Like { .. } | Expr::ILike { .. } | Expr::SimilarTo { .. } => {
                 Ok(DataType::Boolean)
             }
+            Expr::Placeholder(_) => Ok(DataType::Boolean),
             Expr::Wildcard => Err(DataFusionError::Internal(
                 "Wildcard expressions are not valid in a logical query plan".to_owned(),
             )),
@@ -198,7 +199,8 @@ impl ExprSchemable for Expr {
             | Expr::IsNotTrue(_)
             | Expr::IsNotFalse(_)
             | Expr::IsNotUnknown(_)
-            | Expr::Exists { .. } => Ok(false),
+            | Expr::Exists { .. }
+            | Expr::Placeholder(_) => Ok(false), // todo: Placeholder should return false?
             Expr::InSubquery { expr, .. } => expr.nullable(input_schema),
             Expr::ScalarSubquery(subquery) => {
                 Ok(subquery.subquery.schema().field(0).is_nullable())

datafusion/expr/src/expr_visitor.rs

@@ -133,7 +133,8 @@ impl ExprVisitable for Expr {
             | Expr::Exists { .. }
             | Expr::ScalarSubquery(_)
             | Expr::Wildcard
-            | Expr::QualifiedWildcard { .. } => Ok(visitor),
+            | Expr::QualifiedWildcard { .. }
+            | Expr::Placeholder(_) => Ok(visitor),
             Expr::BinaryExpr(BinaryExpr { left, right, .. }) => {
                 let visitor = left.accept(visitor)?;
                 right.accept(visitor)

datafusion/expr/src/logical_plan/builder.rs

@@ -45,6 +45,8 @@ use std::any::Any;
 use std::convert::TryFrom;
 use std::{collections::HashMap, sync::Arc};
 
+use super::Prepare;
+
 /// Default table name for unnamed table
 pub const UNNAMED_TABLE: &str = "?table?";
 
@@ -119,6 +121,7 @@ impl LogicalPlanBuilder {
     /// The column names are not specified by the SQL standard and different database systems do it differently,
     /// so it's usually better to override the default names with a table alias list.
     pub fn values(mut values: Vec<Vec<Expr>>) -> Result<Self> {
+        // todo: hanlde for Placeholder expr
         if values.is_empty() {
             return Err(DataFusionError::Plan("Values list cannot be empty".into()));
         }
@@ -292,6 +295,14 @@ impl LogicalPlanBuilder {
         )?)))
     }
 
+    pub fn prepare(&self, name: String, data_types: Vec<DataType>) -> Result<Self> {
+        Ok(Self::from(LogicalPlan::Prepare(Prepare {
+            name,
+            data_types,
+            input: Arc::new(self.plan.clone()),
+        })))
+    }
+
     /// Limit the number of rows returned
     ///
     /// `skip` - Number of rows to skip before fetch any row.

datafusion/expr/src/logical_plan/mod.rs

@@ -25,7 +25,7 @@ pub use plan::{
     Aggregate, Analyze, CreateCatalog, CreateCatalogSchema, CreateExternalTable,
     CreateMemoryTable, CreateView, CrossJoin, Distinct, DropTable, DropView,
     EmptyRelation, Explain, Extension, Filter, Join, JoinConstraint, JoinType, Limit,
-    LogicalPlan, Partitioning, PlanType, PlanVisitor, Projection, Repartition,
+    LogicalPlan, Partitioning, PlanType, PlanVisitor, Prepare, Projection, Repartition,
     SetVariable, Sort, StringifiedPlan, Subquery, SubqueryAlias, TableScan,
     ToStringifiedPlan, Union, Values, Window,
 };

datafusion/expr/src/logical_plan/plan.rs

@@ -108,6 +108,8 @@ pub enum LogicalPlan {
     Distinct(Distinct),
     /// Set a Variable
     SetVariable(SetVariable),
+    /// Prepare a statement
+    Prepare(Prepare),
 }
 
 impl LogicalPlan {
@@ -134,6 +136,7 @@ impl LogicalPlan {
             LogicalPlan::CreateExternalTable(CreateExternalTable { schema, .. }) => {
                 schema
             }
+            LogicalPlan::Prepare(Prepare { input, .. }) => input.schema(),
             LogicalPlan::Explain(explain) => &explain.schema,
             LogicalPlan::Analyze(analyze) => &analyze.schema,
             LogicalPlan::Extension(extension) => extension.node.schema(),
@@ -201,8 +204,9 @@ impl LogicalPlan {
             | LogicalPlan::Sort(Sort { input, .. })
             | LogicalPlan::CreateMemoryTable(CreateMemoryTable { input, .. })
             | LogicalPlan::CreateView(CreateView { input, .. })
-            | LogicalPlan::Filter(Filter { input, .. }) => input.all_schemas(),
-            LogicalPlan::Distinct(Distinct { input, .. }) => input.all_schemas(),
+            | LogicalPlan::Filter(Filter { input, .. })
+            | LogicalPlan::Distinct(Distinct { input, .. })
+            | LogicalPlan::Prepare(Prepare { input, .. }) => input.all_schemas(),
             LogicalPlan::DropTable(_)
             | LogicalPlan::DropView(_)
             | LogicalPlan::SetVariable(_) => vec![],
@@ -271,7 +275,8 @@ impl LogicalPlan {
             | LogicalPlan::Analyze(_)
             | LogicalPlan::Explain(_)
             | LogicalPlan::Union(_)
-            | LogicalPlan::Distinct(_) => {
+            | LogicalPlan::Distinct(_)
+            | LogicalPlan::Prepare(_) => {
                 vec![]
             }
         }
@@ -300,7 +305,8 @@ impl LogicalPlan {
             LogicalPlan::Explain(explain) => vec![&explain.plan],
             LogicalPlan::Analyze(analyze) => vec![&analyze.input],
             LogicalPlan::CreateMemoryTable(CreateMemoryTable { input, .. })
-            | LogicalPlan::CreateView(CreateView { input, .. }) => {
+            | LogicalPlan::CreateView(CreateView { input, .. })
+            | LogicalPlan::Prepare(Prepare { input, .. }) => {
                 vec![input]
             }
             // plans without inputs
@@ -448,9 +454,8 @@ impl LogicalPlan {
                 input.accept(visitor)?
             }
             LogicalPlan::CreateMemoryTable(CreateMemoryTable { input, .. })
-            | LogicalPlan::CreateView(CreateView { input, .. }) => {
-                input.accept(visitor)?
-            }
+            | LogicalPlan::CreateView(CreateView { input, .. })
+            | LogicalPlan::Prepare(Prepare { input, .. }) => input.accept(visitor)?,
             LogicalPlan::Extension(extension) => {
                 for input in extension.node.inputs() {
                     if !input.accept(visitor)? {
@@ -961,6 +966,11 @@ impl LogicalPlan {
                     LogicalPlan::Analyze { .. } => write!(f, "Analyze"),
                     LogicalPlan::Union(_) => write!(f, "Union"),
                     LogicalPlan::Extension(e) => e.node.fmt_for_explain(f),
+                    LogicalPlan::Prepare(Prepare {
+                        name, data_types, ..
+                    }) => {
+                        write!(f, "Prepare: {:?} {:?} ", name, data_types)
+                    }
                 }
             }
         }
@@ -1358,6 +1368,17 @@ pub struct CreateExternalTable {
     pub options: HashMap<String, String>,
 }
 
+/// Prepare a statement
+#[derive(Clone)]
+pub struct Prepare {
+    /// The name of the statement
+    pub name: String,
+    /// Data types of the parameters
+    pub data_types: Vec<DataType>,
+    /// The logical plan of the statements
+    pub input: Arc<LogicalPlan>,
+}
+
 /// Produces a relation with string representations of
 /// various parts of the plan
 #[derive(Clone)]

datafusion/expr/src/utils.rs

@@ -22,8 +22,8 @@ use crate::expr_visitor::{ExprVisitable, ExpressionVisitor, Recursion};
 use crate::logical_plan::builder::build_join_schema;
 use crate::logical_plan::{
     Aggregate, Analyze, CreateMemoryTable, CreateView, Distinct, Extension, Filter, Join,
-    Limit, Partitioning, Projection, Repartition, Sort, Subquery, SubqueryAlias, Union,
-    Values, Window,
+    Limit, Partitioning, Prepare, Projection, Repartition, Sort, Subquery, SubqueryAlias,
+    Union, Values, Window,
 };
 use crate::{Cast, Expr, ExprSchemable, LogicalPlan, LogicalPlanBuilder};
 use arrow::datatypes::{DataType, TimeUnit};
@@ -126,7 +126,8 @@ impl ExpressionVisitor for ColumnNameVisitor<'_> {
             | Expr::ScalarSubquery(_)
             | Expr::Wildcard
             | Expr::QualifiedWildcard { .. }
-            | Expr::GetIndexedField { .. } => {}
+            | Expr::GetIndexedField { .. }
+            | Expr::Placeholder(_) => {}
         }
         Ok(Recursion::Continue(self))
     }
@@ -575,6 +576,13 @@ pub fn from_plan(
             );
             Ok(plan.clone())
         }
+        LogicalPlan::Prepare(Prepare {
+            name, data_types, ..
+        }) => Ok(LogicalPlan::Prepare(Prepare {
+            name: name.clone(),
+            data_types: data_types.clone(),
+            input: Arc::new(inputs[0].clone()),
+        })),
         LogicalPlan::EmptyRelation(_)
         | LogicalPlan::TableScan { .. }
         | LogicalPlan::CreateExternalTable(_)

datafusion/optimizer/src/common_subexpr_eliminate.rs

@@ -240,7 +240,8 @@ impl OptimizerRule for CommonSubexprEliminate {
             | LogicalPlan::DropView(_)
             | LogicalPlan::SetVariable(_)
             | LogicalPlan::Distinct(_)
-            | LogicalPlan::Extension(_) => {
+            | LogicalPlan::Extension(_)
+            | LogicalPlan::Prepare(_) => {
                 // apply the optimization to all inputs of the plan
                 utils::optimize_children(self, plan, optimizer_config)
             }

datafusion/optimizer/src/projection_push_down.rs

@@ -392,7 +392,8 @@ fn optimize_plan(
         | LogicalPlan::SetVariable(_)
         | LogicalPlan::CrossJoin(_)
         | LogicalPlan::Distinct(_)
-        | LogicalPlan::Extension { .. } => {
+        | LogicalPlan::Extension { .. }
+        | LogicalPlan::Prepare(_) => {
             let expr = plan.expressions();
             // collect all required columns by this plan
             exprlist_to_columns(&expr, &mut new_required_columns)?;

datafusion/optimizer/src/simplify_expressions/expr_simplifier.rs

@@ -253,7 +253,8 @@ impl<'a> ConstEvaluator<'a> {
             | Expr::Sort { .. }
             | Expr::GroupingSet(_)
             | Expr::Wildcard
-            | Expr::QualifiedWildcard { .. } => false,
+            | Expr::QualifiedWildcard { .. }
+            | Expr::Placeholder(_) => false,
             Expr::ScalarFunction { fun, .. } => Self::volatility_ok(fun.volatility()),
             Expr::ScalarUDF { fun, .. } => Self::volatility_ok(fun.signature.volatility),
             Expr::Literal(_)

datafusion/proto/proto/datafusion.proto

@@ -71,6 +71,7 @@ message LogicalPlanNode {
     DistinctNode distinct = 23;
     ViewTableScanNode view_scan = 24;
     CustomTableScanNode custom_scan = 25;
+    PrepareNode prepare = 26;
   }
 }
 
@@ -180,6 +181,12 @@ message CreateExternalTableNode {
   map<string, string> options = 11;
 }
 
+message PrepareNode {
+  string name = 1;
+  repeated ArrowType data_types = 2;
+  LogicalPlanNode input = 3;
+ }
+
 message CreateCatalogSchemaNode {
   string schema_name = 1;
   bool if_not_exists = 2;
@@ -343,9 +350,15 @@ message LogicalExprNode {
     ILikeNode ilike = 32;
     SimilarToNode similar_to = 33;
 
+    PlaceholderNode placeholder = 34;
+
   }
 }
 
+message PlaceholderNode {
+  string param = 1;
+}
+
 message LogicalExprList {
   repeated LogicalExprNode expr = 1;
 }

datafusion/proto/src/from_proto.rs

@@ -19,7 +19,7 @@ use crate::protobuf::plan_type::PlanTypeEnum::{
     FinalLogicalPlan, FinalPhysicalPlan, InitialLogicalPlan, InitialPhysicalPlan,
     OptimizedLogicalPlan, OptimizedPhysicalPlan,
 };
-use crate::protobuf::{self};
+use crate::protobuf::{self, PlaceholderNode};
 use crate::protobuf::{
     CubeNode, GroupingSetNode, OptimizedLogicalPlanType, OptimizedPhysicalPlanType,
     RollupNode,
@@ -1184,6 +1184,9 @@ pub fn parse_expr(
                     .collect::<Result<Vec<_>, Error>>()?,
             )))
         }
+        ExprType::Placeholder(PlaceholderNode { param }) => {
+            Ok(Expr::Placeholder(param.clone()))
+        }
     }
 }