feat: support current_schema and session_user

e2e_test/batch/catalog/sysinfo.slt.part

@@ -0,0 +1,5 @@
+query T
+SELECT current_schema();
+----
+public
+

src/frontend/src/binder/expr/function.rs

@@ -16,6 +16,7 @@ use std::iter::once;
 use std::str::FromStr;
 
 use itertools::Itertools;
+use risingwave_common::catalog::DEFAULT_SCHEMA_NAME;
 use risingwave_common::error::{ErrorCode, Result};
 use risingwave_common::types::DataType;
 use risingwave_expr::expr::AggKind;
@@ -142,7 +143,7 @@ impl Binder {
             "octet_length" => ExprType::OctetLength,
             "bit_length" => ExprType::BitLength,
             "regexp_match" => ExprType::RegexpMatch,
-            // special
+            // System information operations.
             "pg_typeof" if inputs.len() == 1 => {
                 let input = &inputs[0];
                 let v = match input.is_unknown() {
@@ -154,6 +155,14 @@ impl Binder {
             "current_database" if inputs.is_empty() => {
                 return Ok(ExprImpl::literal_varchar(self.db_name.clone()));
             }
+            "current_schema" if inputs.is_empty() => {
+                return Ok(ExprImpl::literal_varchar(DEFAULT_SCHEMA_NAME.to_string()));
+            }
+            "session_user" if inputs.is_empty() => {
+                return Ok(ExprImpl::literal_varchar(
+                    self.auth_context.user_name.clone(),
+                ));
+            }
             // internal
             "rw_vnode" => ExprType::Vnode,
             _ => {

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

@@ -17,8 +17,8 @@ use risingwave_common::catalog::{ColumnDesc, ColumnId};
 use risingwave_common::error::{ErrorCode, Result};
 use risingwave_common::types::DataType;
 use risingwave_sqlparser::ast::{
-    BinaryOperator, DataType as AstDataType, DateTimeField, Expr, Query, StructField,
-    TrimWhereField, UnaryOperator,
+    BinaryOperator, DataType as AstDataType, DateTimeField, Expr, Function, ObjectName, Query,
+    StructField, TrimWhereField, UnaryOperator,
 };
 
 use crate::binder::Binder;
@@ -41,7 +41,20 @@ impl Binder {
             }
             Expr::Row(exprs) => self.bind_row(exprs),
             // input ref
-            Expr::Identifier(ident) => self.bind_column(&[ident]),
+            Expr::Identifier(ident) => {
+                if ["session_user", "current_schema"]
+                    .iter()
+                    .any(|e| ident.real_value().as_str() == *e)
+                {
+                    // Rewrite a system variable to a function call, e.g. `SELECT current_schema;`
+                    // will be rewritten to `SELECT current_schema();`.
+                    // NOTE: Here we don't 100% follow the behavior of Postgres, as it doesn't
+                    // allow `session_user()` while we do.
+                    self.bind_function(Function::no_arg(ObjectName(vec![ident])))
+                } else {
+                    self.bind_column(&[ident])
+                }
+            }
             Expr::CompoundIdentifier(idents) => self.bind_column(&idents),
             Expr::FieldIdentifier(field_expr, idents) => {
                 self.bind_single_field_column(*field_expr, &idents)

src/frontend/src/binder/expr/value.rs

@@ -250,8 +250,8 @@ mod tests {
     use crate::binder::test_utils::mock_binder;
     use crate::expr::{Expr, ExprImpl, ExprType, FunctionCall};
 
-    #[test]
-    fn test_bind_value() {
+    #[tokio::test]
+    async fn test_bind_value() {
         use std::str::FromStr;
 
         use super::*;
@@ -337,8 +337,8 @@ mod tests {
         assert_eq!(expr.return_type(), DataType::Int32);
     }
 
-    #[test]
-    fn test_bind_interval() {
+    #[tokio::test]
+    async fn test_bind_interval() {
         use super::*;
 
         let mut binder = mock_binder();

src/frontend/src/binder/mod.rs

@@ -13,6 +13,7 @@
 // limitations under the License.
 
 use std::collections::HashMap;
+use std::sync::Arc;
 
 use risingwave_common::error::Result;
 use risingwave_sqlparser::ast::{Statement, TableAlias};
@@ -47,13 +48,15 @@ pub use update::BoundUpdate;
 pub use values::BoundValues;
 
 use crate::catalog::catalog_service::CatalogReadGuard;
+use crate::session::{AuthContext, SessionImpl};
 
 /// `Binder` binds the identifiers in AST to columns in relations
 pub struct Binder {
     // TODO: maybe we can only lock the database, but not the whole catalog.
     catalog: CatalogReadGuard,
     db_name: String,
     context: BindContext,
+    auth_context: Arc<AuthContext>,
     /// A stack holding contexts of outer queries when binding a subquery.
     /// It also holds all of the lateral contexts for each respective
     /// subquery.
@@ -73,11 +76,12 @@ pub struct Binder {
 }
 
 impl Binder {
-    pub fn new(catalog: CatalogReadGuard, db_name: String) -> Binder {
+    pub fn new(session: &SessionImpl) -> Binder {
         Binder {
-            catalog,
-            db_name,
+            catalog: session.env().catalog_reader().read_guard(),
+            db_name: session.database().to_string(),
             context: BindContext::new(),
+            auth_context: session.auth_context(),
             upper_subquery_contexts: vec![],
             lateral_contexts: vec![],
             next_subquery_id: 0,
@@ -149,23 +153,12 @@ impl Binder {
 
 #[cfg(test)]
 pub mod test_utils {
-    use std::sync::Arc;
-
-    use parking_lot::RwLock;
-
     use super::Binder;
-    use crate::catalog::catalog_service::CatalogReader;
-    use crate::catalog::root_catalog::Catalog;
+    use crate::session::SessionImpl;
 
-    #[cfg(test)]
-    pub fn mock_binder_with_catalog(catalog: Catalog, db_name: String) -> Binder {
-        let catalog = Arc::new(RwLock::new(catalog));
-        let catalog_reader = CatalogReader::new(catalog);
-        Binder::new(catalog_reader.read_guard(), db_name)
-    }
     #[cfg(test)]
     pub fn mock_binder() -> Binder {
-        mock_binder_with_catalog(Catalog::default(), "".to_string())
+        Binder::new(&SessionImpl::mock())
     }
 }
 

src/frontend/src/binder/values.rs

@@ -92,8 +92,8 @@ mod tests {
     use crate::binder::test_utils::mock_binder;
     use crate::expr::Expr as _;
 
-    #[test]
-    fn test_bind_values() {
+    #[tokio::test]
+    async fn test_bind_values() {
         let mut binder = mock_binder();
 
         // Test i32 -> decimal.

src/frontend/src/handler/create_mv.rs

@@ -61,10 +61,7 @@ pub fn gen_create_mv_plan(
     };
 
     let bound = {
-        let mut binder = Binder::new(
-            session.env().catalog_reader().read_guard(),
-            session.database().to_string(),
-        );
+        let mut binder = Binder::new(session);
         binder.bind_query(*query)?
     };
 

src/frontend/src/handler/dml.rs

@@ -29,10 +29,7 @@ pub async fn handle_dml(context: OptimizerContext, stmt: Statement) -> Result<Pg
     let session = context.session_ctx.clone();
 
     let bound = {
-        let mut binder = Binder::new(
-            session.env().catalog_reader().read_guard(),
-            session.database().to_string(),
-        );
+        let mut binder = Binder::new(&session);
         binder.bind(stmt)?
     };
 

src/frontend/src/handler/explain.rs

@@ -85,10 +85,7 @@ pub(super) fn handle_explain(
 
         stmt => {
             let bound = {
-                let mut binder = Binder::new(
-                    session.env().catalog_reader().read_guard(),
-                    session.database().to_string(),
-                );
+                let mut binder = Binder::new(&session);
                 binder.bind(stmt)?
             };
 

src/frontend/src/handler/query.rs

@@ -38,10 +38,7 @@ pub async fn handle_query(
     let session = context.session_ctx.clone();
 
     let bound = {
-        let mut binder = Binder::new(
-            session.env().catalog_reader().read_guard(),
-            session.database().to_string(),
-        );
+        let mut binder = Binder::new(&session);
         binder.bind(stmt)?
     };
 

src/frontend/src/session.rs

@@ -628,10 +628,7 @@ fn infer(session: Arc<SessionImpl>, stmt: Statement, sql: &str) -> Result<Vec<Pg
     let session = context.session_ctx.clone();
 
     let bound = {
-        let mut binder = Binder::new(
-            session.env().catalog_reader().read_guard(),
-            session.database().to_string(),
-        );
+        let mut binder = Binder::new(&session);
         binder.bind(stmt)?
     };
 

src/frontend/src/test_utils.rs

@@ -117,10 +117,7 @@ impl LocalFrontend {
             let session = self.session_ref();
 
             let bound = {
-                let mut binder = Binder::new(
-                    session.env().catalog_reader().read_guard(),
-                    session.database().to_string(),
-                );
+                let mut binder = Binder::new(&session);
                 binder.bind(Statement::Query(query.clone()))?
             };
             Planner::new(OptimizerContext::new(session, Arc::from(raw_sql.as_str())).into())

src/frontend/test_runner/src/lib.rs

@@ -330,10 +330,7 @@ impl TestCase {
         let mut ret = TestCaseResult::default();
 
         let bound = {
-            let mut binder = Binder::new(
-                session.env().catalog_reader().read_guard(),
-                session.database().to_string(),
-            );
+            let mut binder = Binder::new(&session);
             match binder.bind(stmt.clone()) {
                 Ok(bound) => bound,
                 Err(err) => {

src/frontend/test_runner/tests/testdata/sysinfo_funcs.yaml

@@ -0,0 +1,16 @@
+# This file is automatically generated. See `src/frontend/test_runner/README.md` for more information.
+- sql: |
+    select current_schema();
+  batch_plan: |
+    BatchProject { exprs: ['public':Varchar] }
+      BatchValues { rows: [[]] }
+- sql: |
+    select current_schema;
+  batch_plan: |
+    BatchProject { exprs: ['public':Varchar] }
+      BatchValues { rows: [[]] }
+- sql: |
+    select session_user;
+  batch_plan: |
+    BatchProject { exprs: ['root':Varchar] }
+      BatchValues { rows: [[]] }

src/sqlparser/src/ast/mod.rs

@@ -1604,6 +1604,19 @@ pub struct Function {
     pub filter: Option<Box<Expr>>,
 }
 
+impl Function {
+    pub fn no_arg(name: ObjectName) -> Self {
+        Self {
+            name,
+            args: vec![],
+            over: None,
+            distinct: false,
+            order_by: vec![],
+            filter: None,
+        }
+    }
+}
+
 impl fmt::Display for Function {
     fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
         write!(