Add function current_schema and also tests for current_catalog and etc.

Author: kezhuw

src/sql/mod.rs

@@ -256,5 +256,19 @@ mod tests {
         assert_eq!(column_id.value(3), 2);
         assert_eq!(column_count.value(3), 3);
         assert_eq!(column_description.value(3), "NNNNNN");
+
+        let mut stream = executor
+            .execute_sql("select current_catalog, current_database() as database, current_schema(), inet_client_port()")
+            .await
+            .unwrap();
+        let record = stream.next().await.unwrap().unwrap();
+        let column_catalog = record.column(0).as_any().downcast_ref::<StringArray>().unwrap();
+        let column_database = record.column(1).as_any().downcast_ref::<StringArray>().unwrap();
+        let column_schema = record.column(2).as_any().downcast_ref::<StringArray>().unwrap();
+        let column_client_port = record.column(3).as_any().downcast_ref::<Int32Array>().unwrap();
+        assert_eq!(column_catalog.value(0), "test1");
+        assert_eq!(column_database.value(0), "test1");
+        assert_eq!(column_schema.value(0), "public");
+        assert_eq!(column_client_port.value(0), 0);
     }
 }

src/sql/postgresql/functions/mod.rs

@@ -28,6 +28,45 @@ use datafusion::logical_expr::{
 
 use crate::sql::context::SqlContext;
 
+#[derive(Debug)]
+pub struct CurrentSchema {
+    context: Arc<SqlContext>,
+    signature: Signature,
+    aliases: Vec<String>,
+}
+
+impl CurrentSchema {
+    pub fn new(context: Arc<SqlContext>) -> Self {
+        Self { context, signature: Signature::new(TypeSignature::Nullary, Volatility::Stable), aliases: vec![] }
+    }
+}
+
+impl ScalarUDFImpl for CurrentSchema {
+    fn as_any(&self) -> &(dyn Any + 'static) {
+        self
+    }
+
+    fn name(&self) -> &str {
+        "current_schema"
+    }
+
+    fn signature(&self) -> &Signature {
+        &self.signature
+    }
+
+    fn aliases(&self) -> &[String] {
+        &self.aliases
+    }
+
+    fn return_type(&self, _arg_types: &[DataType]) -> Result<DataType, DataFusionError> {
+        Ok(DataType::Utf8)
+    }
+
+    fn invoke_with_args(&self, _args: ScalarFunctionArgs) -> Result<ColumnarValue, DataFusionError> {
+        Ok(ColumnarValue::Scalar(ScalarValue::Utf8(self.context.current_schema().to_string().into())))
+    }
+}
+
 #[derive(Debug)]
 pub struct CurrentCatalog {
     context: Arc<SqlContext>,

src/sql/postgresql/mod.rs

@@ -74,6 +74,7 @@ use crate::tablet::TabletClient;
 pub fn create_all_scalar_functions(context: &Arc<SqlContext>) -> Vec<Arc<ScalarUDF>> {
     vec![
         Arc::new(ScalarUDF::new_from_impl(self::functions::CurrentCatalog::new(context.clone()))),
+        Arc::new(ScalarUDF::new_from_impl(self::functions::CurrentSchema::new(context.clone()))),
         Arc::new(ScalarUDF::new_from_impl(self::functions::CurrentUser::new(context.clone()))),
         Arc::new(ScalarUDF::new_from_impl(self::functions::InetClientPort::new(context.clone()))),
     ]