* add Parameter expr in binder

* infer parameter type in bind
* add interface to bind param for BoundStatement
* add related test

Author: ZENOTME

src/frontend/src/binder/bind_param.rs

@@ -0,0 +1,151 @@
+// Copyright 2023 RisingWave Labs
+//
+// Licensed 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.
+
+use bytes::Bytes;
+use pgwire::types::Format;
+use risingwave_common::error::{Result, RwError};
+use risingwave_common::types::ScalarImpl;
+
+use super::statement::RewriteExprsRecursive;
+use super::BoundStatement;
+use crate::expr::{Expr, ExprImpl, ExprRewriter, Literal};
+
+/// Rewrites parameter expressions to literals.
+pub(crate) struct ParamRewriter {
+    pub(crate) params: Vec<Bytes>,
+    pub(crate) param_formats: Vec<Format>,
+    pub(crate) error: Option<RwError>,
+}
+
+impl ExprRewriter for ParamRewriter {
+    fn rewrite_expr(&mut self, expr: ExprImpl) -> ExprImpl {
+        if self.error.is_some() {
+            return expr;
+        }
+        match expr {
+            ExprImpl::InputRef(inner) => self.rewrite_input_ref(*inner),
+            ExprImpl::Literal(inner) => self.rewrite_literal(*inner),
+            ExprImpl::FunctionCall(inner) => self.rewrite_function_call(*inner),
+            ExprImpl::AggCall(inner) => self.rewrite_agg_call(*inner),
+            ExprImpl::Subquery(inner) => self.rewrite_subquery(*inner),
+            ExprImpl::CorrelatedInputRef(inner) => self.rewrite_correlated_input_ref(*inner),
+            ExprImpl::TableFunction(inner) => self.rewrite_table_function(*inner),
+            ExprImpl::WindowFunction(inner) => self.rewrite_window_function(*inner),
+            ExprImpl::UserDefinedFunction(inner) => self.rewrite_user_defined_function(*inner),
+            ExprImpl::Parameter(inner) => self.rewrite_parameter(*inner),
+        }
+    }
+
+    fn rewrite_parameter(&mut self, parameter: crate::expr::Parameter) -> ExprImpl {
+        let data_type = parameter.return_type();
+
+        // original parameter.index is 1-based.
+        let parameter_index = (parameter.index - 1) as usize;
+
+        let format = self.param_formats[parameter_index];
+        let scalar = {
+            let res = match format {
+                Format::Text => {
+                    let value = self.params[parameter_index].clone();
+                    ScalarImpl::from_text(&value, &data_type)
+                }
+                Format::Binary => {
+                    let value = self.params[parameter_index].clone();
+                    ScalarImpl::from_binary(&value, &data_type)
+                }
+            };
+
+            match res {
+                Ok(datum) => datum,
+                Err(e) => {
+                    self.error = Some(e);
+                    return parameter.into();
+                }
+            }
+        };
+        Literal::new(Some(scalar), data_type).into()
+    }
+}
+
+impl BoundStatement {
+    pub fn bind_parameter(
+        mut self,
+        params: Vec<Bytes>,
+        param_formats: Vec<Format>,
+    ) -> Result<BoundStatement> {
+        let mut rewriter = ParamRewriter {
+            params,
+            param_formats,
+            error: None,
+        };
+
+        self.rewrite_exprs_recursive(&mut rewriter);
+
+        if let Some(err) = rewriter.error {
+            return Err(err);
+        }
+
+        Ok(self)
+    }
+}
+
+#[cfg(test)]
+mod test {
+    use bytes::Bytes;
+    use pgwire::types::Format;
+    use risingwave_sqlparser::test_utils::parse_sql_statements;
+
+    use crate::binder::test_utils::mock_binder;
+    use crate::binder::BoundStatement;
+
+    fn create_expect_bound(sql: &str) -> BoundStatement {
+        let mut binder = mock_binder();
+        let stmt = parse_sql_statements(sql).unwrap().remove(0);
+        binder.bind(stmt).unwrap()
+    }
+
+    fn create_actual_bound(
+        sql: &str,
+        params: Vec<Bytes>,
+        param_formats: Vec<Format>,
+    ) -> BoundStatement {
+        let mut binder = mock_binder();
+        let stmt = parse_sql_statements(sql).unwrap().remove(0);
+        let bound = binder.bind(stmt).unwrap();
+        bound.bind_parameter(params, param_formats).unwrap()
+    }
+
+    fn expect_actual_eq(expect: BoundStatement, actual: BoundStatement) {
+        println!("expect: {:?}", expect);
+        println!("actual: {:?}", actual);
+        // Use debug format to compare. May modify in future.
+        assert!(format!("{:?}", expect) == format!("{:?}", actual));
+    }
+
+    #[tokio::test]
+    async fn bind_basic_select() {
+        expect_actual_eq(
+            create_expect_bound("select 1::int4"),
+            create_actual_bound("select $1::int4", vec!["1".into()], vec![Format::Text]),
+        );
+    }
+
+    #[tokio::test]
+    async fn bind_basic_value() {
+        expect_actual_eq(
+            create_expect_bound("values(1::int4)"),
+            create_actual_bound("values($1::int4)", vec!["1".into()], vec![Format::Text]),
+        );
+    }
+}

src/frontend/src/binder/delete.rs

@@ -16,6 +16,7 @@ use risingwave_common::catalog::{Schema, TableVersionId};
 use risingwave_common::error::Result;
 use risingwave_sqlparser::ast::{Expr, ObjectName, SelectItem};
 
+use super::statement::RewriteExprsRecursive;
 use super::{Binder, BoundBaseTable};
 use crate::catalog::TableId;
 use crate::expr::ExprImpl;
@@ -48,6 +49,19 @@ pub struct BoundDelete {
     pub returning_schema: Option<Schema>,
 }
 
+impl RewriteExprsRecursive for BoundDelete {
+    fn rewrite_exprs_recursive(&mut self, rewriter: &mut impl crate::expr::ExprRewriter) {
+        self.selection =
+            std::mem::take(&mut self.selection).map(|expr| rewriter.rewrite_expr(expr));
+
+        let new_returning_list = std::mem::take(&mut self.returning_list)
+            .into_iter()
+            .map(|expr| rewriter.rewrite_expr(expr))
+            .collect::<Vec<_>>();
+        self.returning_list = new_returning_list;
+    }
+}
+
 impl Binder {
     pub(super) fn bind_delete(
         &mut self,

src/frontend/src/binder/expr/mod.rs

@@ -23,7 +23,7 @@ use risingwave_sqlparser::ast::{
 };
 
 use crate::binder::Binder;
-use crate::expr::{Expr as _, ExprImpl, ExprType, FunctionCall, SubqueryKind};
+use crate::expr::{Expr as _, ExprImpl, ExprType, FunctionCall, Parameter, SubqueryKind};
 
 mod binary_op;
 mod column;
@@ -123,6 +123,7 @@ impl Binder {
                 start,
                 count,
             } => self.bind_overlay(*expr, *new_substring, *start, count),
+            Expr::Parameter { index } => self.bind_parameter(index),
             _ => Err(ErrorCode::NotImplemented(
                 format!("unsupported expression {:?}", expr),
                 112.into(),
@@ -297,6 +298,10 @@ impl Binder {
         FunctionCall::new(ExprType::Overlay, args).map(|f| f.into())
     }
 
+    fn bind_parameter(&mut self, index: u64) -> Result<ExprImpl> {
+        Ok(Parameter::new(index, self.param_types.clone()).into())
+    }
+
     /// Bind `expr (not) between low and high`
     pub(super) fn bind_between(
         &mut self,

src/frontend/src/binder/insert.rs

@@ -21,6 +21,7 @@ use risingwave_common::types::DataType;
 use risingwave_common::util::iter_util::ZipEqFast;
 use risingwave_sqlparser::ast::{Ident, ObjectName, Query, SelectItem, SetExpr};
 
+use super::statement::RewriteExprsRecursive;
 use super::{BoundQuery, BoundSetExpr};
 use crate::binder::Binder;
 use crate::catalog::TableId;
@@ -66,6 +67,24 @@ pub struct BoundInsert {
     pub returning_schema: Option<Schema>,
 }
 
+impl RewriteExprsRecursive for BoundInsert {
+    fn rewrite_exprs_recursive(&mut self, rewriter: &mut impl crate::expr::ExprRewriter) {
+        self.source.rewrite_exprs_recursive(rewriter);
+
+        let new_cast_exprs = std::mem::take(&mut self.cast_exprs)
+            .into_iter()
+            .map(|expr| rewriter.rewrite_expr(expr))
+            .collect::<Vec<_>>();
+        self.cast_exprs = new_cast_exprs;
+
+        let new_returning_list = std::mem::take(&mut self.returning_list)
+            .into_iter()
+            .map(|expr| rewriter.rewrite_expr(expr))
+            .collect::<Vec<_>>();
+        self.returning_list = new_returning_list;
+    }
+}
+
 impl Binder {
     pub(super) fn bind_insert(
         &mut self,

src/frontend/src/binder/mod.rs

@@ -13,13 +13,15 @@
 // limitations under the License.
 
 use std::collections::HashMap;
-use std::sync::Arc;
+use std::sync::{Arc, RwLock};
 
 use risingwave_common::error::Result;
 use risingwave_common::session_config::SearchPath;
+use risingwave_common::types::DataType;
 use risingwave_sqlparser::ast::Statement;
 
 mod bind_context;
+mod bind_param;
 mod create;
 mod delete;
 mod expr;
@@ -90,6 +92,8 @@ pub struct Binder {
 
     /// `ShareId`s identifying shared views.
     shared_views: HashMap<ViewId, ShareId>,
+
+    param_types: Arc<RwLock<HashMap<u64, Option<DataType>>>>,
 }
 
 impl Binder {
@@ -114,6 +118,7 @@ impl Binder {
             search_path: session.config().get_search_path(),
             in_create_mv,
             shared_views: HashMap::new(),
+            param_types: Arc::new(RwLock::new(HashMap::new())),
         }
     }
 
@@ -130,6 +135,10 @@ impl Binder {
         self.bind_statement(stmt)
     }
 
+    pub fn export_param_types(&self) -> HashMap<u64, Option<DataType>> {
+        self.param_types.read().unwrap().clone()
+    }
+
     fn push_context(&mut self) {
         let new_context = std::mem::take(&mut self.context);
         self.context.cte_to_relation = new_context.cte_to_relation.clone();

src/frontend/src/binder/query.rs

@@ -20,8 +20,9 @@ use risingwave_common::error::{ErrorCode, Result};
 use risingwave_common::types::DataType;
 use risingwave_sqlparser::ast::{Cte, Expr, Fetch, OrderByExpr, Query, Value, With};
 
+use super::statement::RewriteExprsRecursive;
 use crate::binder::{Binder, BoundSetExpr};
-use crate::expr::{CorrelatedId, Depth, ExprImpl};
+use crate::expr::{CorrelatedId, Depth, ExprImpl, ExprRewriter};
 use crate::optimizer::property::{Direction, FieldOrder};
 
 /// A validated sql query, including order and union.
@@ -96,6 +97,18 @@ impl BoundQuery {
     }
 }
 
+impl RewriteExprsRecursive for BoundQuery {
+    fn rewrite_exprs_recursive(&mut self, rewriter: &mut impl ExprRewriter) {
+        let new_extra_order_exprs = std::mem::take(&mut self.extra_order_exprs)
+            .into_iter()
+            .map(|expr| rewriter.rewrite_expr(expr))
+            .collect::<Vec<_>>();
+        self.extra_order_exprs = new_extra_order_exprs;
+
+        self.body.rewrite_exprs_recursive(rewriter);
+    }
+}
+
 impl Binder {
     /// Bind a [`Query`].
     ///

src/frontend/src/binder/relation/join.rs

@@ -19,6 +19,7 @@ use risingwave_sqlparser::ast::{
 };
 
 use crate::binder::bind_context::BindContext;
+use crate::binder::statement::RewriteExprsRecursive;
 use crate::binder::{Binder, Relation, COLUMN_GROUP_PREFIX};
 use crate::expr::ExprImpl;
 
@@ -30,6 +31,14 @@ pub struct BoundJoin {
     pub cond: ExprImpl,
 }
 
+impl RewriteExprsRecursive for BoundJoin {
+    fn rewrite_exprs_recursive(&mut self, rewriter: &mut impl crate::expr::ExprRewriter) {
+        self.left.rewrite_exprs_recursive(rewriter);
+        self.right.rewrite_exprs_recursive(rewriter);
+        self.cond = rewriter.rewrite_expr(std::mem::take(&mut self.cond));
+    }
+}
+
 impl Binder {
     pub(crate) fn bind_vec_table_with_joins(
         &mut self,

src/frontend/src/binder/relation/mod.rs

@@ -27,6 +27,7 @@ use risingwave_sqlparser::ast::{
 
 use self::watermark::is_watermark_func;
 use super::bind_context::ColumnBinding;
+use super::statement::RewriteExprsRecursive;
 use crate::binder::{Binder, BoundSetExpr};
 use crate::catalog::system_catalog::pg_catalog::{
     PG_GET_KEYWORDS_FUNC_NAME, PG_KEYWORDS_TABLE_NAME,
@@ -64,6 +65,26 @@ pub enum Relation {
     Share(Box<BoundShare>),
 }
 
+impl RewriteExprsRecursive for Relation {
+    fn rewrite_exprs_recursive(&mut self, rewriter: &mut impl crate::expr::ExprRewriter) {
+        match self {
+            Relation::Subquery(inner) => inner.rewrite_exprs_recursive(rewriter),
+            Relation::Join(inner) => inner.rewrite_exprs_recursive(rewriter),
+            Relation::WindowTableFunction(inner) => inner.rewrite_exprs_recursive(rewriter),
+            Relation::Watermark(inner) => inner.rewrite_exprs_recursive(rewriter),
+            Relation::Share(inner) => inner.rewrite_exprs_recursive(rewriter),
+            Relation::TableFunction(inner) => {
+                let new_args = std::mem::take(&mut inner.args)
+                    .into_iter()
+                    .map(|expr| rewriter.rewrite_expr(expr))
+                    .collect();
+                inner.args = new_args;
+            }
+            _ => {}
+        }
+    }
+}
+
 impl Relation {
     pub fn contains_sys_table(&self) -> bool {
         match self {

src/frontend/src/binder/relation/share.rs

@@ -12,6 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+use crate::binder::statement::RewriteExprsRecursive;
 use crate::binder::{Relation, ShareId};
 
 /// Share a relation during binding and planning.
@@ -21,3 +22,9 @@ pub struct BoundShare {
     pub(crate) share_id: ShareId,
     pub(crate) input: Relation,
 }
+
+impl RewriteExprsRecursive for BoundShare {
+    fn rewrite_exprs_recursive(&mut self, rewriter: &mut impl crate::expr::ExprRewriter) {
+        self.input.rewrite_exprs_recursive(rewriter);
+    }
+}