This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/paimon-rust.git
The following commit(s) were added to refs/heads/main by this push:
new 009039f feat(datafusion): support REST Catalog CREATE FUNCTION (#499)
009039f is described below
commit 009039f11c3668cbd2994e18758d4344cc62e6e4
Author: Jingsong Lee <[email protected]>
AuthorDate: Sat Jul 11 16:59:45 2026 +0800
feat(datafusion): support REST Catalog CREATE FUNCTION (#499)
---
crates/integrations/datafusion/README.md | 23 +-
crates/integrations/datafusion/src/sql_context.rs | 972 ++++++++++++++++++++-
crates/integrations/datafusion/src/sql_function.rs | 75 +-
crates/integrations/datafusion/src/table/mod.rs | 2 +-
crates/paimon/src/api/api_request.rs | 39 +-
crates/paimon/src/api/mod.rs | 4 +-
crates/paimon/src/api/rest_api.rs | 19 +-
crates/paimon/src/catalog/mod.rs | 15 +
crates/paimon/src/catalog/rest/rest_catalog.rs | 32 +
crates/paimon/src/error.rs | 2 +
crates/paimon/tests/mock_server.rs | 68 +-
crates/paimon/tests/rest_api_test.rs | 35 +
crates/paimon/tests/rest_catalog_test.rs | 135 +++
crates/paimon/tests/rest_object_models_test.rs | 69 +-
docs/src/sql.md | 71 +-
15 files changed, 1512 insertions(+), 49 deletions(-)
diff --git a/crates/integrations/datafusion/README.md
b/crates/integrations/datafusion/README.md
index 0f6a664..c4fa4d2 100644
--- a/crates/integrations/datafusion/README.md
+++ b/crates/integrations/datafusion/README.md
@@ -26,11 +26,16 @@ This crate contains the integration of [Apache
DataFusion](https://datafusion.ap
## REST Catalog views and SQL functions
-`SQLContext` can read persistent views and SQL scalar functions in a Paimon
REST Catalog. It can
-also create persistent REST Catalog views:
+`SQLContext` can read, execute, and create persistent views and SQL scalar
functions in a Paimon
+REST Catalog:
```sql
CREATE VIEW [IF NOT EXISTS] view_name [(column_name, ...)] AS query;
+
+CREATE FUNCTION [IF NOT EXISTS] function_name([parameter_name data_type, ...])
+RETURNS data_type
+[LANGUAGE SQL]
+RETURN scalar_expression;
```
- A persistent view is resolved lazily like a table. The `datafusion` dialect
is preferred and the
@@ -44,10 +49,15 @@ CREATE VIEW [IF NOT EXISTS] view_name [(column_name, ...)]
AS query;
`catalog.database.function(args...)`. Its `definitions.datafusion` value
must be a scalar SQL
expression, it must be deterministic, and it must declare its input
parameters and exactly one
return parameter.
+- `CREATE FUNCTION` requires named parameters, one return type, and a scalar
`RETURN` expression.
+ `LANGUAGE SQL` is optional and SQL is the default, matching Databricks
syntax. Determinism is
+ inferred and validated from the planned expression before sending the REST
create request. Bare,
+ two-part, and three-part creation targets are supported; calls remain
limited to bare and
+ three-part names.
- `CREATE OR REPLACE VIEW`, materialized/secure views, comments/options,
persistent `ALTER VIEW` /
- `DROP VIEW`, and persistent function DDL are not supported. Lambda/file
functions, named
- arguments, multiple return values, non-deterministic functions, and calls
made directly through
- a raw DataFusion `SessionContext` are also not supported.
+ `DROP VIEW`, `CREATE OR REPLACE/ALTER/TEMPORARY FUNCTION`, and persistent
`ALTER FUNCTION` /
+ `DROP FUNCTION` are not supported. Lambda/file,
aggregate/table/multi-return, non-deterministic,
+ Stable/Volatile, and non-SQL functions are also not supported.
Use `SQLContext::sql` for function expansion:
@@ -56,8 +66,9 @@ let mut ctx = paimon_datafusion::SQLContext::new();
ctx.register_catalog("paimon", rest_catalog).await?;
ctx.sql("CREATE VIEW daily_scores AS SELECT normalize_score(score) AS score
FROM scores").await?;
+ctx.sql("CREATE FUNCTION plus_one(x BIGINT) RETURNS BIGINT RETURN x +
1").await?;
let view = ctx.sql("SELECT * FROM analytics_view").await?;
-let function = ctx.sql("SELECT normalize_score(score) FROM scores").await?;
+let function = ctx.sql("SELECT plus_one(score) FROM scores").await?;
```
See the [documentation](https://paimon.apache.org/docs/rust/datafusion/) for
getting started guide and more details.
diff --git a/crates/integrations/datafusion/src/sql_context.rs
b/crates/integrations/datafusion/src/sql_context.rs
index bc00552..ec72013 100644
--- a/crates/integrations/datafusion/src/sql_context.rs
+++ b/crates/integrations/datafusion/src/sql_context.rs
@@ -31,6 +31,7 @@
//! - `ALTER TABLE db.t RENAME TO new_name`
//! - `ALTER TABLE db.t DROP PARTITION (col = val, ...)`
//! - `CREATE VIEW [IF NOT EXISTS] view [(col, ...)] AS query`
+//! - `CREATE FUNCTION name(args) RETURNS type [LANGUAGE SQL] RETURN
expression`
//! - `TRUNCATE TABLE db.t`
//! - `TRUNCATE TABLE db.t PARTITION (col = val, ...)`
@@ -44,20 +45,25 @@ use datafusion::arrow::array::{
use datafusion::arrow::compute::cast;
use datafusion::arrow::datatypes::{DataType as ArrowDataType, Field, Schema};
use datafusion::arrow::record_batch::RecordBatch;
+use datafusion::common::tree_node::{TreeNode, TreeNodeRecursion};
use datafusion::common::TableReference;
use datafusion::datasource::{MemTable, TableProvider};
use datafusion::error::{DataFusionError, Result as DFResult};
use datafusion::execution::SessionStateBuilder;
+use datafusion::logical_expr::{Expr as LogicalExpr, LogicalPlan, Volatility};
use datafusion::prelude::{DataFrame, SessionContext};
use datafusion::sql::planner::IdentNormalizer;
use datafusion::sql::sqlparser::ast::{
- AlterTableOperation, BinaryLength, CharacterLength, ColumnDef,
ColumnOption, CreateTable,
- CreateTableOptions, CreateView, Delete, Expr as SqlExpr, FromTable,
Insert, Merge, ObjectName,
- ObjectType, RenameTableNameKind, Reset, ResetStatement, Set,
ShowCreateObject, SqlOption,
- Statement, TableFactor, TableObject, Truncate, Update, Value as SqlValue,
+ AlterTableOperation, BinaryLength, CharacterLength, ColumnDef,
ColumnOption, CreateFunction,
+ CreateFunctionBody, CreateTable, CreateTableOptions, CreateView, Delete,
Expr as SqlExpr,
+ FromTable, FunctionBehavior, FunctionReturnType, Insert, Merge,
ObjectName, ObjectType,
+ RenameTableNameKind, Reset, ResetStatement, Set, ShowCreateObject,
SqlOption, Statement,
+ TableFactor, TableObject, Truncate, Update, Value as SqlValue,
};
use datafusion::sql::sqlparser::dialect::GenericDialect;
+use datafusion::sql::sqlparser::keywords::Keyword;
use datafusion::sql::sqlparser::parser::Parser;
+use datafusion::sql::sqlparser::tokenizer::{Token, Tokenizer};
use futures::StreamExt;
use paimon::catalog::{parse_object_name, Catalog, Identifier};
use paimon::spec::{
@@ -356,8 +362,7 @@ impl SQLContext {
return self.handle_time_travel_query(&rewritten_sql).await;
}
- let statements = Parser::parse_sql(&GenericDialect {}, &rewritten_sql)
- .map_err(|e| DataFusionError::Plan(format!("SQL parse error:
{e}")))?;
+ let statements = parse_sql_statements(&rewritten_sql)?;
if statements.len() != 1 {
return Err(DataFusionError::Plan(
@@ -450,6 +455,13 @@ impl SQLContext {
}
}
}
+ Statement::CreateFunction(create_function) => {
+ if self.is_paimon_function_name(&create_function.name) {
+ self.handle_create_function(create_function).await
+ } else {
+ self.ctx.sql(sql).await
+ }
+ }
Statement::Drop {
object_type,
if_exists,
@@ -1430,6 +1442,157 @@ impl SQLContext {
}
}
+ async fn handle_create_function(
+ &self,
+ create_function: &CreateFunction,
+ ) -> DFResult<DataFrame> {
+ validate_persistent_create_function(create_function)?;
+ if create_function
+ .language
+ .as_ref()
+ .is_some_and(|language|
!language.value.eq_ignore_ascii_case("sql"))
+ {
+ return Err(DataFusionError::Plan(
+ "CREATE FUNCTION only supports LANGUAGE SQL".to_string(),
+ ));
+ }
+ if matches!(
+ create_function.behavior,
+ Some(FunctionBehavior::Stable | FunctionBehavior::Volatile)
+ ) {
+ return Err(DataFusionError::Plan(
+ "CREATE FUNCTION only supports deterministic SQL
functions".to_string(),
+ ));
+ }
+ let FunctionReturnType::DataType(return_type) = create_function
+ .return_type
+ .as_ref()
+ .ok_or_else(|| DataFusionError::Plan("CREATE FUNCTION requires
RETURNS".to_string()))?
+ else {
+ return Err(DataFusionError::Plan(
+ "CREATE FUNCTION SETOF return types are not
supported".to_string(),
+ ));
+ };
+ let CreateFunctionBody::Return(body) = create_function
+ .function_body
+ .as_ref()
+ .ok_or_else(|| DataFusionError::Plan("CREATE FUNCTION requires
RETURN".to_string()))?
+ else {
+ return Err(DataFusionError::Plan(
+ "CREATE FUNCTION only supports a RETURN
expression".to_string(),
+ ));
+ };
+
+ let mut parameter_names = HashSet::new();
+ let input_params = create_function
+ .args
+ .as_deref()
+ .unwrap_or_default()
+ .iter()
+ .enumerate()
+ .map(|(id, argument)| {
+ if argument.mode.is_some() || argument.default_expr.is_some() {
+ return Err(DataFusionError::Plan(
+ "CREATE FUNCTION argument modes and defaults are not
supported".to_string(),
+ ));
+ }
+ let name = argument
+ .name
+ .clone()
+ .map(normalize_create_function_argument_name)
+ .ok_or_else(|| {
+ DataFusionError::Plan(
+ "CREATE FUNCTION arguments must have
names".to_string(),
+ )
+ })?;
+ if !parameter_names.insert(name.clone()) {
+ return Err(DataFusionError::Plan(format!(
+ "duplicate function argument name '{name}'"
+ )));
+ }
+ Ok(PaimonDataField::new(
+ id as i32,
+ name,
+ sql_data_type_to_paimon_type(&argument.data_type, true)?,
+ ))
+ })
+ .collect::<DFResult<Vec<_>>>()?;
+ let return_params = vec![PaimonDataField::new(
+ 0,
+ "result".to_string(),
+ sql_data_type_to_paimon_type(return_type, true)?,
+ )];
+ let (catalog, catalog_name, identifier) =
+ self.resolve_catalog_and_function(&create_function.name)?;
+ let function = paimon::catalog::Function::new(
+ identifier,
+ Some(input_params),
+ Some(return_params),
+ true,
+ HashMap::from([(
+ "datafusion".to_string(),
+ paimon::catalog::FunctionDefinition::Sql {
+ definition: body.to_string(),
+ },
+ )]),
+ None,
+ HashMap::new(),
+ );
+ self.validate_create_function(&function, &catalog_name)
+ .await?;
+ catalog
+ .create_function(&function, create_function.if_not_exists)
+ .await
+ .map_err(to_datafusion_error)?;
+ ok_result(&self.ctx)
+ }
+
+ async fn validate_create_function(
+ &self,
+ function: &paimon::catalog::Function,
+ catalog_name: &str,
+ ) -> DFResult<()> {
+ let arguments = function
+ .input_params()
+ .unwrap_or_default()
+ .iter()
+ .map(|field| {
+ let sql_type =
+
crate::table::data_type_to_sql(field.data_type()).map_err(|error| {
+ DataFusionError::Plan(format!(
+ "Invalid CREATE FUNCTION argument type '{:?}':
{error}",
+ field.data_type()
+ ))
+ })?;
+ Ok(format!("CAST(NULL AS {sql_type})"))
+ })
+ .collect::<DFResult<Vec<_>>>()?
+ .join(", ");
+ let quote = |identifier: &str| format!("\"{}\"",
identifier.replace('"', "\"\""));
+ let validation_sql = format!(
+ "SELECT {}.{}.{}({arguments})",
+ quote(catalog_name),
+ quote(function.identifier().database()),
+ quote(function.name())
+ );
+ let expanded = crate::sql_function::expand_sql_with_candidate(
+ &validation_sql,
+ &self.catalogs,
+ catalog_name,
+ function.identifier().database(),
+ function,
+ )
+ .await?;
+ let mut state = self.ctx.state();
+ state.config_mut().options_mut().catalog.default_catalog =
catalog_name.to_string();
+ state.config_mut().options_mut().catalog.default_schema =
+ function.identifier().database().to_string();
+ let logical_plan = state.create_logical_plan(&expanded).await?;
+ validate_immutable_scalar_plan(&logical_plan)?;
+ state.create_physical_plan(&logical_plan).await?;
+ Ok(())
+ }
+
async fn handle_drop_partitions(
&self,
catalog: &Arc<dyn Catalog>,
@@ -1496,6 +1659,77 @@ impl SQLContext {
self.catalogs.contains_key(&catalog_name)
}
+ fn is_paimon_function_name(&self, name: &ObjectName) -> bool {
+ let Some(parts) = name
+ .0
+ .iter()
+ .map(|part| {
+ part.as_ident()
+ .map(|identifier|
IdentNormalizer::default().normalize(identifier.clone()))
+ })
+ .collect::<Option<Vec<_>>>()
+ else {
+ return false;
+ };
+ let catalog_name = match parts.as_slice() {
+ [catalog, _, _] => catalog.clone(),
+ [_] | [_, _] => self.current_catalog_name(),
+ _ => return false,
+ };
+ self.catalogs.contains_key(&catalog_name)
+ }
+
+ fn resolve_catalog_and_function(
+ &self,
+ name: &ObjectName,
+ ) -> DFResult<(Arc<dyn Catalog>, String, Identifier)> {
+ let parts = name
+ .0
+ .iter()
+ .map(|part| {
+ part.as_ident()
+ .cloned()
+ .map(|identifier|
IdentNormalizer::default().normalize(identifier))
+ .ok_or_else(|| {
+ DataFusionError::Plan(format!("Invalid function
reference: {name}"))
+ })
+ })
+ .collect::<DFResult<Vec<_>>>()?;
+ match parts.as_slice() {
+ [catalog_name, database, function] => {
+ let catalog = self.catalogs.get(catalog_name).ok_or_else(|| {
+ DataFusionError::Plan(format!("Unknown catalog
'{catalog_name}'"))
+ })?;
+ Ok((
+ Arc::clone(catalog),
+ catalog_name.clone(),
+ Identifier::new(database, function),
+ ))
+ }
+ [database, function] => Ok((
+ self.current_catalog()?,
+ self.current_catalog_name(),
+ Identifier::new(database, function),
+ )),
+ [function] => Ok((
+ self.current_catalog()?,
+ self.current_catalog_name(),
+ Identifier::new(
+ self.ctx
+ .state()
+ .config_options()
+ .catalog
+ .default_schema
+ .clone(),
+ function,
+ ),
+ )),
+ _ => Err(DataFusionError::Plan(format!(
+ "Invalid function reference: {name}"
+ ))),
+ }
+ }
+
/// Resolve an ObjectName like `catalog.db.table` or `db.table` to a
catalog and Identifier.
fn resolve_catalog_and_table(
&self,
@@ -1569,6 +1803,182 @@ impl SQLContext {
}
}
+fn validate_immutable_scalar_plan(plan: &LogicalPlan) -> DFResult<()> {
+ let mut violation = None;
+ plan.apply(|node| {
+ node.apply_expressions(|expression| {
+ expression.apply(|expression| {
+ match expression {
+ LogicalExpr::ScalarFunction(function)
+ if function.func.signature().volatility !=
Volatility::Immutable =>
+ {
+ violation = Some(format!(
+ "CREATE FUNCTION body uses non-immutable function
'{}'",
+ function.func.name()
+ ));
+ }
+ LogicalExpr::HigherOrderFunction(function)
+ if function.func.signature().volatility !=
Volatility::Immutable =>
+ {
+ violation = Some(format!(
+ "CREATE FUNCTION body uses non-immutable function
'{}'",
+ function.func.name()
+ ));
+ }
+ LogicalExpr::AggregateFunction(_) |
LogicalExpr::WindowFunction(_) => {
+ violation = Some(
+ "CREATE FUNCTION body must be a scalar expression;
aggregate and window functions are not supported"
+ .to_string(),
+ );
+ }
+ LogicalExpr::Exists(_)
+ | LogicalExpr::InSubquery(_)
+ | LogicalExpr::SetComparison(_)
+ | LogicalExpr::ScalarSubquery(_) => {
+ violation = Some(
+ "CREATE FUNCTION body must be a scalar expression;
subqueries are not supported"
+ .to_string(),
+ );
+ }
+ LogicalExpr::Unnest(_) => {
+ violation = Some(
+ "CREATE FUNCTION body must be a scalar expression;
UNNEST is not supported"
+ .to_string(),
+ );
+ }
+ _ => {}
+ }
+ Ok(TreeNodeRecursion::Continue)
+ })?;
+ Ok(TreeNodeRecursion::Continue)
+ })?;
+ Ok(TreeNodeRecursion::Continue)
+ })?;
+ if let Some(message) = violation {
+ return Err(DataFusionError::Plan(message));
+ }
+ Ok(())
+}
+
+fn normalize_create_function_argument_name(
+ identifier: datafusion::sql::sqlparser::ast::Ident,
+) -> String {
+ if identifier.quote_style.is_none() {
+ if let Some(value) = identifier
+ .value
+ .strip_prefix('"')
+ .and_then(|value| value.strip_suffix('"'))
+ {
+ return value.replace("\"\"", "\"");
+ }
+ }
+ IdentNormalizer::default().normalize(identifier)
+}
+
+fn validate_persistent_create_function(create_function: &CreateFunction) ->
DFResult<()> {
+ let unsupported = if create_function.or_alter {
+ Some("CREATE OR ALTER FUNCTION is not supported")
+ } else if create_function.or_replace {
+ Some("CREATE OR REPLACE FUNCTION is not supported")
+ } else if create_function.temporary {
+ Some("CREATE TEMPORARY FUNCTION is not supported")
+ } else if create_function.args.is_none() {
+ Some("CREATE FUNCTION requires a parenthesized argument list")
+ } else if create_function.called_on_null.is_some() {
+ Some("CREATE FUNCTION NULL INPUT clauses are not supported")
+ } else if create_function.parallel.is_some() {
+ Some("CREATE FUNCTION PARALLEL clauses are not supported")
+ } else if create_function.security.is_some() {
+ Some("CREATE FUNCTION SECURITY clauses are not supported")
+ } else if !create_function.set_params.is_empty() {
+ Some("CREATE FUNCTION SET clauses are not supported")
+ } else if create_function.using.is_some() {
+ Some("CREATE FUNCTION USING clauses are not supported")
+ } else if create_function.determinism_specifier.is_some() {
+ Some("CREATE FUNCTION determinism specifiers are not supported")
+ } else if create_function.options.is_some() {
+ Some("CREATE FUNCTION OPTIONS clauses are not supported")
+ } else if create_function.remote_connection.is_some() {
+ Some("CREATE FUNCTION REMOTE clauses are not supported")
+ } else {
+ None
+ };
+ if let Some(message) = unsupported {
+ return Err(DataFusionError::Plan(message.to_string()));
+ }
+ Ok(())
+}
+
+fn parse_sql_statements(sql: &str) -> DFResult<Vec<Statement>> {
+ let dialect = GenericDialect {};
+ let mut tokens = Tokenizer::new(&dialect, sql)
+ .tokenize_with_location()
+ .map_err(|error| DataFusionError::Plan(format!("SQL parse error:
{error}")))?;
+ let significant = tokens
+ .iter()
+ .enumerate()
+ .filter_map(|(index, token)| match &token.token {
+ Token::Whitespace(_) => None,
+ Token::Word(word) => Some((index, word.keyword)),
+ _ => Some((index, Keyword::NoKeyword)),
+ })
+ .take(5)
+ .collect::<Vec<_>>();
+ let create_function_if_not_exists = matches!(
+ significant.as_slice(),
+ [
+ (_, Keyword::CREATE),
+ (_, Keyword::FUNCTION),
+ (_, Keyword::IF),
+ (_, Keyword::NOT),
+ (_, Keyword::EXISTS)
+ ]
+ );
+ let create_function_or_alter = matches!(
+ significant.as_slice(),
+ [
+ (_, Keyword::CREATE),
+ (_, Keyword::OR),
+ (_, Keyword::ALTER),
+ (_, Keyword::FUNCTION),
+ ..
+ ]
+ );
+ if create_function_if_not_exists {
+ let removed = significant[2..=4]
+ .iter()
+ .map(|(index, _)| *index)
+ .collect::<HashSet<_>>();
+ tokens = tokens
+ .into_iter()
+ .enumerate()
+ .filter(|(index, _)| !removed.contains(index))
+ .map(|(_, token)| token)
+ .collect();
+ }
+ let mut statements = Parser::new(&dialect)
+ .with_tokens_with_locations(tokens)
+ .parse_statements()
+ .map_err(|error| DataFusionError::Plan(format!("SQL parse error:
{error}")))?;
+ if create_function_if_not_exists {
+ let Some(Statement::CreateFunction(create_function)) =
statements.first_mut() else {
+ return Err(DataFusionError::Plan(
+ "SQL parse error: invalid CREATE FUNCTION IF NOT EXISTS
statement".to_string(),
+ ));
+ };
+ create_function.if_not_exists = true;
+ }
+ if create_function_or_alter {
+ let Some(Statement::CreateFunction(create_function)) =
statements.first_mut() else {
+ return Err(DataFusionError::Plan(
+ "SQL parse error: invalid CREATE OR ALTER FUNCTION
statement".to_string(),
+ ));
+ };
+ create_function.or_alter = true;
+ }
+ Ok(statements)
+}
+
fn validate_persistent_create_view(create_view: &CreateView) -> DFResult<()> {
let unsupported = if create_view.or_alter {
Some("CREATE OR ALTER VIEW is not supported")
@@ -2970,6 +3380,24 @@ mod tests {
.collect())
}
+ async fn create_function(
+ &self,
+ function: &paimon::catalog::Function,
+ ignore_if_exists: bool,
+ ) -> paimon::Result<()> {
+ let mut functions = self.functions.lock().unwrap();
+ if functions.contains_key(function.identifier()) {
+ if ignore_if_exists {
+ return Ok(());
+ }
+ return Err(paimon::Error::FunctionAlreadyExist {
+ full_name: function.full_name(),
+ });
+ }
+ functions.insert(function.identifier().clone(), function.clone());
+ Ok(())
+ }
+
async fn get_function(
&self,
identifier: &Identifier,
@@ -3143,6 +3571,538 @@ mod tests {
assert_eq!(answers.value(0), 42);
}
+ #[tokio::test]
+ async fn persistent_rest_catalog_function_can_be_created_and_called() {
+ let catalog = Arc::new(MockCatalog::new());
+ let ctx = make_sql_context(Arc::clone(&catalog)).await;
+
+ ctx.sql(
+ "CREATE FUNCTION plus_one(x BIGINT) RETURNS BIGINT \
+ LANGUAGE SQL IMMUTABLE RETURN x + 1",
+ )
+ .await
+ .unwrap();
+
+ let stored = catalog
+ .get_function(&Identifier::new("default", "plus_one"))
+ .await
+ .unwrap();
+ assert_eq!(stored.input_params().unwrap()[0].id(), 0);
+ assert_eq!(stored.input_params().unwrap()[0].name(), "x");
+ assert!(stored.input_params().unwrap()[0].data_type().is_nullable());
+ assert_eq!(stored.return_params().unwrap()[0].id(), 0);
+ assert_eq!(stored.return_params().unwrap()[0].name(), "result");
+ assert!(stored.return_params().unwrap()[0].data_type().is_nullable());
+
+ let batches = ctx
+ .sql("SELECT plus_one(41) AS answer")
+ .await
+ .unwrap()
+ .collect()
+ .await
+ .unwrap();
+ let answers = batches[0]
+ .column_by_name("answer")
+ .unwrap()
+ .as_any()
+ .downcast_ref::<Int64Array>()
+ .unwrap();
+ assert_eq!(answers.value(0), 42);
+ }
+
+ #[tokio::test]
+ async fn
persistent_rest_catalog_function_uses_databricks_default_sql_syntax() {
+ let catalog = Arc::new(MockCatalog::new());
+ let ctx = make_sql_context(Arc::clone(&catalog)).await;
+
+ ctx.sql("CREATE FUNCTION plus_one(x BIGINT) RETURNS BIGINT RETURN x +
1")
+ .await
+ .unwrap();
+
+ let stored = catalog
+ .get_function(&Identifier::new("default", "plus_one"))
+ .await
+ .unwrap();
+ assert!(stored.is_deterministic());
+
+ let batches = ctx
+ .sql("SELECT plus_one(41) AS answer")
+ .await
+ .unwrap()
+ .collect()
+ .await
+ .unwrap();
+ let answers = batches[0]
+ .column_by_name("answer")
+ .unwrap()
+ .as_any()
+ .downcast_ref::<Int64Array>()
+ .unwrap();
+ assert_eq!(answers.value(0), 42);
+ }
+
+ #[tokio::test]
+ async fn persistent_rest_catalog_function_supports_array_argument() {
+ let catalog = Arc::new(MockCatalog::new());
+ let ctx = make_sql_context(Arc::clone(&catalog)).await;
+
+ ctx.sql(
+ "CREATE FUNCTION array_answer(x ARRAY<BIGINT>) \
+ RETURNS BIGINT RETURN 42",
+ )
+ .await
+ .unwrap();
+
+ assert!(catalog
+ .get_function(&Identifier::new("default", "array_answer"))
+ .await
+ .is_ok());
+ }
+
+ #[tokio::test]
+ async fn persistent_rest_catalog_function_supports_array_return_type() {
+ let catalog = Arc::new(MockCatalog::new());
+ let ctx = make_sql_context(catalog).await;
+
+ ctx.sql(
+ "CREATE FUNCTION singleton(x BIGINT) \
+ RETURNS ARRAY<BIGINT> RETURN make_array(x)",
+ )
+ .await
+ .unwrap();
+
+ let batches = ctx
+ .sql("SELECT singleton(42) AS answer")
+ .await
+ .unwrap()
+ .collect()
+ .await
+ .unwrap();
+ assert_eq!(batches[0].num_rows(), 1);
+ assert!(matches!(
+ batches[0]
+ .schema()
+ .field_with_name("answer")
+ .unwrap()
+ .data_type(),
+ ArrowDataType::List(_)
+ ));
+ }
+
+ #[tokio::test]
+ async fn persistent_rest_catalog_function_normalizes_unquoted_bare_name() {
+ let catalog = Arc::new(MockCatalog::new());
+ let ctx = make_sql_context(Arc::clone(&catalog)).await;
+
+ ctx.sql(
+ "CREATE FUNCTION PlusOne(X BIGINT) RETURNS BIGINT \
+ LANGUAGE SQL IMMUTABLE RETURN X + 1",
+ )
+ .await
+ .unwrap();
+
+ let stored = catalog
+ .get_function(&Identifier::new("default", "plusone"))
+ .await
+ .unwrap();
+ assert_eq!(stored.input_params().unwrap()[0].name(), "x");
+ let batches = ctx
+ .sql("SELECT plusone(41) AS answer")
+ .await
+ .unwrap()
+ .collect()
+ .await
+ .unwrap();
+ let answers = batches[0]
+ .column_by_name("answer")
+ .unwrap()
+ .as_any()
+ .downcast_ref::<Int64Array>()
+ .unwrap();
+ assert_eq!(answers.value(0), 42);
+ }
+
+ #[tokio::test]
+ async fn persistent_rest_catalog_function_preserves_quoted_names() {
+ let catalog = Arc::new(MockCatalog::new());
+ let ctx = make_sql_context(Arc::clone(&catalog)).await;
+
+ ctx.sql(
+ "CREATE FUNCTION \"PlusOne\"(\"Input\" BIGINT) RETURNS BIGINT \
+ LANGUAGE SQL IMMUTABLE RETURN \"Input\" + 1",
+ )
+ .await
+ .unwrap();
+
+ let stored = catalog
+ .get_function(&Identifier::new("default", "PlusOne"))
+ .await
+ .unwrap();
+ assert_eq!(stored.input_params().unwrap()[0].name(), "Input");
+ let batches = ctx
+ .sql("SELECT \"PlusOne\"(41) AS answer")
+ .await
+ .unwrap()
+ .collect()
+ .await
+ .unwrap();
+ let answers = batches[0]
+ .column_by_name("answer")
+ .unwrap()
+ .as_any()
+ .downcast_ref::<Int64Array>()
+ .unwrap();
+ assert_eq!(answers.value(0), 42);
+ }
+
+ #[tokio::test]
+ async fn
persistent_rest_catalog_function_if_not_exists_preserves_existing_function() {
+ let catalog = Arc::new(MockCatalog::new());
+ let ctx = make_sql_context(Arc::clone(&catalog)).await;
+ ctx.sql(
+ "CREATE FUNCTION answer() RETURNS BIGINT \
+ LANGUAGE SQL IMMUTABLE RETURN 1",
+ )
+ .await
+ .unwrap();
+
+ ctx.sql(
+ "CrEaTe /* keep comments */ FuNcTiOn IF /* gap */ NOT EXISTS
answer() \
+ RETURNS BIGINT LANGUAGE SQL IMMUTABLE RETURN 2",
+ )
+ .await
+ .unwrap();
+
+ let stored = catalog
+ .get_function(&Identifier::new("default", "answer"))
+ .await
+ .unwrap();
+ assert_eq!(
+ stored
+ .definition("datafusion")
+ .and_then(paimon::catalog::FunctionDefinition::sql),
+ Some("1")
+ );
+ }
+
+ #[tokio::test]
+ async fn
persistent_rest_catalog_function_if_not_exists_validates_proposed_body() {
+ let catalog = Arc::new(MockCatalog::new());
+ let ctx = make_sql_context(Arc::clone(&catalog)).await;
+ ctx.sql(
+ "CREATE FUNCTION answer() RETURNS BIGINT \
+ LANGUAGE SQL IMMUTABLE RETURN 1",
+ )
+ .await
+ .unwrap();
+
+ let error = ctx
+ .sql(
+ "CREATE FUNCTION IF NOT EXISTS answer() RETURNS BIGINT \
+ LANGUAGE SQL IMMUTABLE RETURN undeclared + 1",
+ )
+ .await
+ .unwrap_err();
+
+ assert!(error.to_string().contains("undeclared identifier"));
+ let stored = catalog
+ .get_function(&Identifier::new("default", "answer"))
+ .await
+ .unwrap();
+ assert_eq!(
+ stored
+ .definition("datafusion")
+ .and_then(paimon::catalog::FunctionDefinition::sql),
+ Some("1")
+ );
+ }
+
+ #[tokio::test]
+ async fn
persistent_rest_catalog_function_uses_owning_database_for_dependencies() {
+ let catalog = Arc::new(MockCatalog::new());
+ add_unary_sql_function_in_database(&catalog, "default", "plus_one", "x
+ 1", true);
+ add_unary_sql_function_in_database(&catalog, "other", "plus_one", "x +
100", true);
+ let ctx = make_sql_context(Arc::clone(&catalog)).await;
+ ctx.set_current_database("other").await.unwrap();
+
+ ctx.sql(
+ "CREATE FUNCTION paimon.default.wrapper(x BIGINT) RETURNS BIGINT \
+ LANGUAGE SQL IMMUTABLE RETURN plus_one(x)",
+ )
+ .await
+ .unwrap();
+
+ let batches = ctx
+ .sql("SELECT paimon.default.wrapper(41) AS answer")
+ .await
+ .unwrap()
+ .collect()
+ .await
+ .unwrap();
+ let answers = batches[0]
+ .column_by_name("answer")
+ .unwrap()
+ .as_any()
+ .downcast_ref::<Int64Array>()
+ .unwrap();
+ assert_eq!(answers.value(0), 42);
+ }
+
+ #[tokio::test]
+ async fn
persistent_rest_catalog_function_rejects_nondeterministic_dependency() {
+ let catalog = Arc::new(MockCatalog::new());
+ add_unary_sql_function(&catalog, "unstable", "x + 1", false);
+ let ctx = make_sql_context(Arc::clone(&catalog)).await;
+
+ let error = ctx
+ .sql(
+ "CREATE FUNCTION wrapper(x BIGINT) RETURNS BIGINT \
+ LANGUAGE SQL IMMUTABLE RETURN unstable(x)",
+ )
+ .await
+ .unwrap_err();
+
+ assert!(error.to_string().contains("non-deterministic"));
+ assert!(matches!(
+ catalog
+ .get_function(&Identifier::new("default", "wrapper"))
+ .await,
+ Err(paimon::Error::FunctionNotExist { .. })
+ ));
+ }
+
+ #[tokio::test]
+ async fn
persistent_rest_catalog_function_rejects_direct_recursion_before_create() {
+ let catalog = Arc::new(MockCatalog::new());
+ let ctx = make_sql_context(Arc::clone(&catalog)).await;
+
+ let error = ctx
+ .sql(
+ "CREATE FUNCTION abs(x BIGINT) RETURNS BIGINT \
+ LANGUAGE SQL IMMUTABLE RETURN abs(x)",
+ )
+ .await
+ .unwrap_err();
+
+ assert!(
+ error.to_string().contains("recursive REST SQL function"),
+ "unexpected error: {error}"
+ );
+ assert!(matches!(
+ catalog
+ .get_function(&Identifier::new("default", "abs"))
+ .await,
+ Err(paimon::Error::FunctionNotExist { .. })
+ ));
+ }
+
+ #[tokio::test]
+ async fn
persistent_rest_catalog_function_rejects_indirect_recursion_before_create() {
+ let catalog = Arc::new(MockCatalog::new());
+ add_unary_sql_function(&catalog, "existing", "candidate(x)", true);
+ let ctx = make_sql_context(Arc::clone(&catalog)).await;
+
+ let error = ctx
+ .sql(
+ "CREATE FUNCTION candidate(x BIGINT) RETURNS BIGINT \
+ LANGUAGE SQL IMMUTABLE RETURN existing(x)",
+ )
+ .await
+ .unwrap_err();
+
+ assert!(
+ error.to_string().contains("recursive REST SQL function"),
+ "unexpected error: {error}"
+ );
+ assert!(matches!(
+ catalog
+ .get_function(&Identifier::new("default", "candidate"))
+ .await,
+ Err(paimon::Error::FunctionNotExist { .. })
+ ));
+ }
+
+ #[tokio::test]
+ async fn
persistent_rest_catalog_function_rejects_volatile_datafusion_function() {
+ let catalog = Arc::new(MockCatalog::new());
+ let ctx = make_sql_context(Arc::clone(&catalog)).await;
+
+ let error = ctx
+ .sql(
+ "CREATE FUNCTION random_value() RETURNS DOUBLE \
+ LANGUAGE SQL IMMUTABLE RETURN random()",
+ )
+ .await
+ .unwrap_err();
+
+ assert!(
+ error
+ .to_string()
+ .contains("non-immutable function 'random'"),
+ "unexpected error: {error}"
+ );
+ assert!(matches!(
+ catalog
+ .get_function(&Identifier::new("default", "random_value"))
+ .await,
+ Err(paimon::Error::FunctionNotExist { .. })
+ ));
+ }
+
+ #[tokio::test]
+ async fn persistent_rest_catalog_function_rejects_unsupported_clauses() {
+ let cases = [
+ (
+ "CREATE OR REPLACE FUNCTION invalid() RETURNS BIGINT LANGUAGE
SQL IMMUTABLE RETURN 1",
+ "OR REPLACE",
+ ),
+ (
+ "CREATE OR ALTER FUNCTION invalid() RETURNS BIGINT LANGUAGE
SQL IMMUTABLE RETURN 1",
+ "OR ALTER",
+ ),
+ (
+ "CREATE TEMPORARY FUNCTION invalid() RETURNS BIGINT LANGUAGE
SQL IMMUTABLE RETURN 1",
+ "TEMPORARY",
+ ),
+ (
+ "CREATE FUNCTION invalid() RETURNS BIGINT LANGUAGE SQL
IMMUTABLE CALLED ON NULL INPUT RETURN 1",
+ "NULL INPUT",
+ ),
+ (
+ "CREATE FUNCTION invalid() RETURNS BIGINT LANGUAGE SQL
IMMUTABLE PARALLEL SAFE RETURN 1",
+ "PARALLEL",
+ ),
+ (
+ "CREATE FUNCTION invalid() RETURNS BIGINT LANGUAGE SQL
IMMUTABLE SECURITY INVOKER RETURN 1",
+ "SECURITY",
+ ),
+ (
+ "CREATE FUNCTION invalid() RETURNS BIGINT LANGUAGE SQL
IMMUTABLE SET search_path TO public RETURN 1",
+ "SET",
+ ),
+ ];
+
+ for (sql, clause) in cases {
+ let catalog = Arc::new(MockCatalog::new());
+ let ctx = make_sql_context(catalog).await;
+ let error = match ctx.sql(sql).await {
+ Ok(_) => panic!("expected error for {clause}: {sql}"),
+ Err(error) => error,
+ };
+ assert!(
+ error.to_string().contains(clause),
+ "expected error for {clause}, got: {error}"
+ );
+ }
+ }
+
+ #[tokio::test]
+ async fn persistent_rest_catalog_function_rejects_non_scalar_bodies() {
+ let cases = [
+ (
+ "CREATE FUNCTION invalid() RETURNS BIGINT LANGUAGE SQL
IMMUTABLE RETURN count(1)",
+ "aggregate and window",
+ ),
+ (
+ "CREATE FUNCTION invalid() RETURNS BIGINT LANGUAGE SQL
IMMUTABLE RETURN row_number() OVER ()",
+ "aggregate and window",
+ ),
+ (
+ "CREATE FUNCTION invalid() RETURNS BIGINT LANGUAGE SQL
IMMUTABLE RETURN (SELECT 1)",
+ "subqueries",
+ ),
+ ];
+
+ for (sql, expected) in cases {
+ let catalog = Arc::new(MockCatalog::new());
+ let ctx = make_sql_context(catalog).await;
+ let error = match ctx.sql(sql).await {
+ Ok(_) => panic!("expected error containing {expected}: {sql}"),
+ Err(error) => error,
+ };
+ assert!(
+ error.to_string().contains(expected),
+ "expected {expected}, got: {error}"
+ );
+ }
+ }
+
+ #[tokio::test]
+ async fn
persistent_rest_catalog_function_rejects_invalid_signature_and_body_forms() {
+ let cases = [
+ (
+ "CREATE FUNCTION invalid() RETURNS BIGINT LANGUAGE PYTHON
RETURN 1",
+ "LANGUAGE SQL",
+ ),
+ (
+ "CREATE FUNCTION invalid() RETURNS BIGINT LANGUAGE SQL STABLE
RETURN 1",
+ "deterministic SQL",
+ ),
+ (
+ "CREATE FUNCTION invalid() RETURNS BIGINT LANGUAGE SQL
IMMUTABLE AS '1'",
+ "RETURN expression",
+ ),
+ (
+ "CREATE FUNCTION invalid() RETURNS SETOF BIGINT LANGUAGE SQL
IMMUTABLE RETURN 1",
+ "SETOF",
+ ),
+ (
+ "CREATE FUNCTION invalid(BIGINT) RETURNS BIGINT LANGUAGE SQL
IMMUTABLE RETURN 1",
+ "must have names",
+ ),
+ (
+ "CREATE FUNCTION invalid(IN x BIGINT) RETURNS BIGINT LANGUAGE
SQL IMMUTABLE RETURN x",
+ "modes and defaults",
+ ),
+ (
+ "CREATE FUNCTION invalid(x BIGINT = 1) RETURNS BIGINT LANGUAGE
SQL IMMUTABLE RETURN x",
+ "modes and defaults",
+ ),
+ (
+ "CREATE FUNCTION invalid(X BIGINT, x BIGINT) RETURNS BIGINT
LANGUAGE SQL IMMUTABLE RETURN x",
+ "duplicate function argument",
+ ),
+ ];
+
+ for (sql, expected) in cases {
+ let catalog = Arc::new(MockCatalog::new());
+ let ctx = make_sql_context(catalog).await;
+ let error = match ctx.sql(sql).await {
+ Ok(_) => panic!("expected error containing {expected}: {sql}"),
+ Err(error) => error,
+ };
+ assert!(
+ error.to_string().contains(expected),
+ "expected {expected}, got: {error}"
+ );
+ }
+ }
+
+ #[tokio::test]
+ async fn
persistent_rest_catalog_function_rejects_incompatible_return_type() {
+ let catalog = Arc::new(MockCatalog::new());
+ let ctx = make_sql_context(Arc::clone(&catalog)).await;
+
+ let error = ctx
+ .sql(
+ "CREATE FUNCTION invalid() RETURNS BIGINT \
+ LANGUAGE SQL IMMUTABLE RETURN named_struct('value', 1)",
+ )
+ .await
+ .unwrap_err();
+
+ assert!(
+ error.to_string().to_ascii_lowercase().contains("cast"),
+ "unexpected error: {error}"
+ );
+ assert!(matches!(
+ catalog
+ .get_function(&Identifier::new("default", "invalid"))
+ .await,
+ Err(paimon::Error::FunctionNotExist { .. })
+ ));
+ }
+
#[tokio::test]
async fn persistent_rest_catalog_view_infers_type_and_nullability() {
let catalog = Arc::new(MockCatalog::new());
diff --git a/crates/integrations/datafusion/src/sql_function.rs
b/crates/integrations/datafusion/src/sql_function.rs
index 88d52be..4d6af48 100644
--- a/crates/integrations/datafusion/src/sql_function.rs
+++ b/crates/integrations/datafusion/src/sql_function.rs
@@ -39,6 +39,16 @@ pub(crate) async fn expand_sql(
catalogs: &HashMap<String, Arc<dyn Catalog>>,
current_catalog: &str,
current_database: &str,
+) -> DFResult<String> {
+ expand_sql_internal(sql, catalogs, current_catalog, current_database,
None).await
+}
+
+async fn expand_sql_internal(
+ sql: &str,
+ catalogs: &HashMap<String, Arc<dyn Catalog>>,
+ current_catalog: &str,
+ current_database: &str,
+ candidate: Option<&Function>,
) -> DFResult<String> {
let mut statements = Parser::parse_sql(&GenericDialect {}, sql)
.map_err(|error| DataFusionError::Plan(format!("Invalid REST SQL:
{error}")))?;
@@ -48,16 +58,35 @@ pub(crate) async fn expand_sql(
statements.len()
)));
}
- let statement = expand_statement(
+ let statement = expand_statement_internal(
statements.remove(0),
catalogs,
current_catalog,
current_database,
+ MAX_EXPANDED_CALLS,
+ candidate,
)
.await?;
Ok(statement.to_string())
}
+pub(crate) async fn expand_sql_with_candidate(
+ sql: &str,
+ catalogs: &HashMap<String, Arc<dyn Catalog>>,
+ current_catalog: &str,
+ current_database: &str,
+ candidate: &Function,
+) -> DFResult<String> {
+ expand_sql_internal(
+ sql,
+ catalogs,
+ current_catalog,
+ current_database,
+ Some(candidate),
+ )
+ .await
+}
+
pub(crate) async fn expand_statement(
statement: Statement,
catalogs: &HashMap<String, Arc<dyn Catalog>>,
@@ -75,12 +104,33 @@ pub(crate) async fn expand_statement(
}
pub(crate) async fn expand_statement_with_budget(
+ statement: Statement,
+ catalogs: &HashMap<String, Arc<dyn Catalog>>,
+ current_catalog: &str,
+ current_database: &str,
+ max_expanded_calls: usize,
+) -> DFResult<Statement> {
+ expand_statement_internal(
+ statement,
+ catalogs,
+ current_catalog,
+ current_database,
+ max_expanded_calls,
+ None,
+ )
+ .await
+}
+
+async fn expand_statement_internal(
mut statement: Statement,
catalogs: &HashMap<String, Arc<dyn Catalog>>,
current_catalog: &str,
current_database: &str,
max_expanded_calls: usize,
+ candidate: Option<&Function>,
) -> DFResult<Statement> {
+ let candidate_reference =
+ candidate.map(|function| (current_catalog.to_string(),
function.identifier().clone()));
let mut functions = HashMap::new();
let mut dependencies = HashMap::new();
let mut total_expanded_calls = 0;
@@ -111,11 +161,15 @@ pub(crate) async fn expand_statement_with_budget(
functions.insert(reference, None);
continue;
};
- let function = match catalog.get_function(&reference.1).await {
- Ok(function) => Some(function),
- Err(paimon::Error::FunctionNotExist { .. })
- | Err(paimon::Error::Unsupported { .. }) => None,
- Err(error) => return Err(crate::to_datafusion_error(error)),
+ let function = if candidate_reference.as_ref() == Some(&reference)
{
+ candidate.cloned()
+ } else {
+ match catalog.get_function(&reference.1).await {
+ Ok(function) => Some(function),
+ Err(paimon::Error::FunctionNotExist { .. })
+ | Err(paimon::Error::Unsupported { .. }) => None,
+ Err(error) => return
Err(crate::to_datafusion_error(error)),
+ }
};
if let Some(function) = &function {
let nested = function_dependencies(function, &reference.0,
reference.1.database());
@@ -403,19 +457,12 @@ fn expand_call(
});
let return_type = function.return_params().expect("validated
above")[0].data_type();
- let serialized_type = serde_json::to_value(return_type).map_err(|error| {
+ let sql_type = crate::table::data_type_to_sql(return_type).map_err(|error|
{
DataFusionError::Plan(format!(
"Invalid return type for REST SQL function '{}': {error}",
function.full_name()
))
})?;
- let sql_type = serialized_type.as_str().ok_or_else(|| {
- DataFusionError::Plan(format!(
- "REST SQL function '{}' has a return type that cannot be
represented in SQL",
- function.full_name()
- ))
- })?;
- let sql_type = sql_type.strip_suffix(" NOT NULL").unwrap_or(sql_type);
Parser::new(&GenericDialect {})
.try_with_sql(&format!("CAST(({body}) AS {sql_type})"))
.map_err(|error| {
diff --git a/crates/integrations/datafusion/src/table/mod.rs
b/crates/integrations/datafusion/src/table/mod.rs
index 9b698f4..9e333ac 100644
--- a/crates/integrations/datafusion/src/table/mod.rs
+++ b/crates/integrations/datafusion/src/table/mod.rs
@@ -209,7 +209,7 @@ fn quote_string_literal(text: &str) -> String {
/// at the top of a column definition, not nested inside `MAP`, `ARRAY`, or
/// `STRUCT` arguments. Callers that render a column should append `NOT NULL`
/// themselves when the field is non-nullable; recursive calls below must not.
-fn data_type_to_sql(data_type: &DataType) -> DFResult<String> {
+pub(crate) fn data_type_to_sql(data_type: &DataType) -> DFResult<String> {
match data_type {
DataType::Boolean(_) => Ok("BOOLEAN".to_string()),
DataType::TinyInt(_) => Ok("TINYINT".to_string()),
diff --git a/crates/paimon/src/api/api_request.rs
b/crates/paimon/src/api/api_request.rs
index fabcf4a..018673c 100644
--- a/crates/paimon/src/api/api_request.rs
+++ b/crates/paimon/src/api/api_request.rs
@@ -23,8 +23,8 @@ use serde::{Deserialize, Serialize};
use std::collections::HashMap;
use crate::{
- catalog::{Identifier, ViewSchema},
- spec::{Schema, SchemaChange},
+ catalog::{Function, FunctionDefinition, Identifier, ViewSchema},
+ spec::{DataField, Schema, SchemaChange},
};
/// Request to create a new database.
@@ -115,6 +115,41 @@ impl CreateViewRequest {
}
}
+/// Request to create a persistent function.
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(rename_all = "camelCase")]
+pub struct CreateFunctionRequest {
+ /// Unqualified function name.
+ pub name: String,
+ /// Declared input parameters.
+ pub input_params: Option<Vec<DataField>>,
+ /// Declared return parameters.
+ pub return_params: Option<Vec<DataField>>,
+ /// Whether the function is deterministic.
+ pub deterministic: bool,
+ /// Engine-specific function definitions.
+ pub definitions: HashMap<String, FunctionDefinition>,
+ /// Optional function comment.
+ pub comment: Option<String>,
+ /// Function options.
+ pub options: HashMap<String, String>,
+}
+
+impl CreateFunctionRequest {
+ /// Create a request from a catalog function.
+ pub fn from_function(function: &Function) -> Self {
+ Self {
+ name: function.name().to_string(),
+ input_params: function.input_params().map(<[_]>::to_vec),
+ return_params: function.return_params().map(<[_]>::to_vec),
+ deterministic: function.is_deterministic(),
+ definitions: function.definitions().clone(),
+ comment: function.comment().map(str::to_string),
+ options: function.options().clone(),
+ }
+ }
+}
+
/// Request to alter a table's schema.
///
/// Wire-compatible with Java Paimon's `AlterTableRequest` (`{"changes":
[...]}`).
diff --git a/crates/paimon/src/api/mod.rs b/crates/paimon/src/api/mod.rs
index 643b1ec..fb2da9f 100644
--- a/crates/paimon/src/api/mod.rs
+++ b/crates/paimon/src/api/mod.rs
@@ -31,8 +31,8 @@ mod api_response;
// Re-export request types
pub use api_request::{
- AlterDatabaseRequest, AlterTableRequest, CreateDatabaseRequest,
CreateTableRequest,
- CreateViewRequest, RenameTableRequest,
+ AlterDatabaseRequest, AlterTableRequest, CreateDatabaseRequest,
CreateFunctionRequest,
+ CreateTableRequest, CreateViewRequest, RenameTableRequest,
};
// Re-export response types
diff --git a/crates/paimon/src/api/rest_api.rs
b/crates/paimon/src/api/rest_api.rs
index 37f89fd..5d60b33 100644
--- a/crates/paimon/src/api/rest_api.rs
+++ b/crates/paimon/src/api/rest_api.rs
@@ -23,14 +23,14 @@
use std::collections::HashMap;
use crate::api::rest_client::HttpClient;
-use crate::catalog::{Identifier, ViewSchema};
+use crate::catalog::{Function, Identifier, ViewSchema};
use crate::common::{CatalogOptions, Options};
use crate::spec::{Partition, PartitionStatistics, Schema, SchemaChange,
Snapshot};
use crate::Result;
use super::api_request::{
- AlterDatabaseRequest, AlterTableRequest, CreateDatabaseRequest,
CreateTableRequest,
- CreateViewRequest, RenameTableRequest,
+ AlterDatabaseRequest, AlterTableRequest, CreateDatabaseRequest,
CreateFunctionRequest,
+ CreateTableRequest, CreateViewRequest, RenameTableRequest,
};
use super::api_response::{
ConfigResponse, GetDatabaseResponse, GetFunctionResponse,
GetTableResponse, GetViewResponse,
@@ -469,6 +469,19 @@ impl RESTApi {
// ==================== Function Operations ====================
+ /// Create a persistent function.
+ pub async fn create_function(&self, function: &Function) -> Result<()> {
+ let database = function.identifier().database();
+ validate_non_empty_multi(&[
+ (database, "database name"),
+ (function.name(), "function name"),
+ ])?;
+ let path = self.resource_paths.functions(database);
+ let request = CreateFunctionRequest::from_function(function);
+ let _resp: serde_json::Value = self.client.post(&path,
&request).await?;
+ Ok(())
+ }
+
/// List persistent functions in a database.
pub async fn list_functions(&self, database: &str) -> Result<Vec<String>> {
validate_non_empty(database, "database name")?;
diff --git a/crates/paimon/src/catalog/mod.rs b/crates/paimon/src/catalog/mod.rs
index 8d17300..516b830 100644
--- a/crates/paimon/src/catalog/mod.rs
+++ b/crates/paimon/src/catalog/mod.rs
@@ -405,6 +405,21 @@ pub trait Catalog: Send + Sync {
// ======================= function methods ===============================
+ /// Create a persistent function.
+ ///
+ /// * `ignore_if_exists` - if true, do nothing when the function already
exists;
+ /// if false, return [`crate::Error::FunctionAlreadyExist`].
+ ///
+ /// # Errors
+ /// * [`crate::Error::DatabaseNotExist`] - database in the function
identifier does not exist.
+ /// * [`crate::Error::FunctionAlreadyExist`] - function already exists when
+ /// `ignore_if_exists` is false.
+ async fn create_function(&self, _function: &Function, _ignore_if_exists:
bool) -> Result<()> {
+ Err(Error::Unsupported {
+ message: "Catalog does not support functions".to_string(),
+ })
+ }
+
/// List persistent function names in a database.
async fn list_functions(&self, _database_name: &str) ->
Result<Vec<String>> {
Err(Error::Unsupported {
diff --git a/crates/paimon/src/catalog/rest/rest_catalog.rs
b/crates/paimon/src/catalog/rest/rest_catalog.rs
index 58eb63b..9fbb388 100644
--- a/crates/paimon/src/catalog/rest/rest_catalog.rs
+++ b/crates/paimon/src/catalog/rest/rest_catalog.rs
@@ -320,6 +320,21 @@ impl Catalog for RESTCatalog {
.map_err(|e| map_unsupported_endpoint(e, "function"))
}
+ async fn create_function(
+ &self,
+ function: &crate::catalog::Function,
+ ignore_if_exists: bool,
+ ) -> Result<()> {
+ let result = self
+ .api
+ .create_function(function)
+ .await
+ .map_err(|error| map_rest_error_for_create_function(error,
function.identifier()));
+ ignore_error_if(result, |error| {
+ ignore_if_exists && matches!(error, Error::FunctionAlreadyExist {
.. })
+ })
+ }
+
async fn get_function(&self, identifier: &Identifier) ->
Result<crate::catalog::Function> {
let response = self
.api
@@ -460,6 +475,23 @@ fn map_rest_error_for_function(err: Error, identifier:
&Identifier) -> Error {
}
}
+/// Map a REST API error from creating a persistent function.
+fn map_rest_error_for_create_function(err: Error, identifier: &Identifier) ->
Error {
+ match err {
+ Error::RestApi {
+ source: RestError::NoSuchResource { .. },
+ } => Error::DatabaseNotExist {
+ database: identifier.database().to_string(),
+ },
+ Error::RestApi {
+ source: RestError::AlreadyExists { .. },
+ } => Error::FunctionAlreadyExist {
+ full_name: identifier.full_name(),
+ },
+ other => map_unsupported_endpoint(other, "function"),
+ }
+}
+
fn map_unsupported_endpoint(err: Error, object_type: &str) -> Error {
match err {
Error::RestApi {
diff --git a/crates/paimon/src/error.rs b/crates/paimon/src/error.rs
index 507ee3a..39969ee 100644
--- a/crates/paimon/src/error.rs
+++ b/crates/paimon/src/error.rs
@@ -108,6 +108,8 @@ pub enum Error {
ViewNotExist { full_name: String },
#[snafu(display("Function {} does not exist.", full_name))]
FunctionNotExist { full_name: String },
+ #[snafu(display("Function {} already exists.", full_name))]
+ FunctionAlreadyExist { full_name: String },
#[snafu(display("Column {} already exists in table {}.", column,
full_name))]
ColumnAlreadyExist { full_name: String, column: String },
#[snafu(display("Column {} does not exist in table {}.", column,
full_name))]
diff --git a/crates/paimon/tests/mock_server.rs
b/crates/paimon/tests/mock_server.rs
index 35a8a3f..21e9520 100644
--- a/crates/paimon/tests/mock_server.rs
+++ b/crates/paimon/tests/mock_server.rs
@@ -34,12 +34,13 @@ use std::sync::{Arc, Mutex};
use tokio::task::JoinHandle;
use paimon::api::{
- AlterDatabaseRequest, AlterTableRequest, AuditRESTResponse,
ConfigResponse, CreateViewRequest,
- ErrorResponse, GetDatabaseResponse, GetFunctionResponse, GetTableResponse,
GetViewResponse,
- ListDatabasesResponse, ListFunctionsResponse, ListTablesResponse,
ListViewsResponse,
- RenameTableRequest, ResourcePaths,
+ AlterDatabaseRequest, AlterTableRequest, AuditRESTResponse, ConfigResponse,
+ CreateFunctionRequest, CreateViewRequest, ErrorResponse,
GetDatabaseResponse,
+ GetFunctionResponse, GetTableResponse, GetViewResponse,
ListDatabasesResponse,
+ ListFunctionsResponse, ListTablesResponse, ListViewsResponse,
RenameTableRequest,
+ ResourcePaths,
};
-use paimon::catalog::Function;
+use paimon::catalog::{Function, Identifier};
#[derive(Clone, Debug, Default)]
struct MockState {
@@ -505,6 +506,61 @@ impl RESTServer {
.into_response()
}
+ /// Handle POST /databases/:db/functions - create a persistent function.
+ pub async fn create_function(
+ Path(db): Path<String>,
+ Extension(state): Extension<Arc<RESTServer>>,
+ Json(request): Json<CreateFunctionRequest>,
+ ) -> impl IntoResponse {
+ let mut s = state.inner.lock().unwrap();
+ let function_name = request.name.clone();
+ if s.view_function_endpoints_unsupported {
+ let err = ErrorResponse::new(
+ Some("function".to_string()),
+ Some(function_name),
+ Some("Not Implemented".to_string()),
+ Some(501),
+ );
+ return (StatusCode::NOT_IMPLEMENTED, Json(err)).into_response();
+ }
+ if !s.databases.contains_key(&db) {
+ let err = ErrorResponse::new(
+ Some("database".to_string()),
+ Some(db),
+ Some("Not Found".to_string()),
+ Some(404),
+ );
+ return (StatusCode::NOT_FOUND, Json(err)).into_response();
+ }
+ let key = format!("{db}.{function_name}");
+ if s.functions.contains_key(&key) {
+ let err = ErrorResponse::new(
+ Some("function".to_string()),
+ Some(function_name),
+ Some("Already Exists".to_string()),
+ Some(409),
+ );
+ return (StatusCode::CONFLICT, Json(err)).into_response();
+ }
+ let function = Function::new(
+ Identifier::new(&db, &request.name),
+ request.input_params,
+ request.return_params,
+ request.deterministic,
+ request.definitions,
+ request.comment,
+ request.options,
+ );
+ s.functions.insert(
+ key,
+ GetFunctionResponse::from_function(
+ &function,
+ AuditRESTResponse::new(None, None, None, None, None),
+ ),
+ );
+ (StatusCode::OK, Json(json!({"function":
function_name}))).into_response()
+ }
+
/// Handle POST /databases/:db/tables - create a new table.
pub async fn create_table(
Path(db): Path<String>,
@@ -999,7 +1055,7 @@ pub async fn start_mock_server(
)
.route(
&format!("{prefix}/databases/:db/functions"),
- get(RESTServer::list_functions),
+ get(RESTServer::list_functions).post(RESTServer::create_function),
)
.route(
&format!("{prefix}/databases/:db/functions/:function"),
diff --git a/crates/paimon/tests/rest_api_test.rs
b/crates/paimon/tests/rest_api_test.rs
index d90e633..e020b6c 100644
--- a/crates/paimon/tests/rest_api_test.rs
+++ b/crates/paimon/tests/rest_api_test.rs
@@ -230,6 +230,41 @@ async fn test_list_functions() {
);
}
+#[tokio::test]
+async fn test_create_function() {
+ let ctx = setup_test_server(vec!["default"]).await;
+ let function = Function::new(
+ Identifier::new("default", "plus_one"),
+ Some(
+ serde_json::from_value(json!([
+ {"id": 0, "name": "x", "type": "BIGINT"}
+ ]))
+ .unwrap(),
+ ),
+ Some(
+ serde_json::from_value(json!([
+ {"id": 0, "name": "result", "type": "BIGINT"}
+ ]))
+ .unwrap(),
+ ),
+ true,
+ HashMap::from([(
+ "datafusion".to_string(),
+ FunctionDefinition::Sql {
+ definition: "x + 1".to_string(),
+ },
+ )]),
+ None,
+ HashMap::new(),
+ );
+
+ ctx.api.create_function(&function).await.unwrap();
+
+ let response = ctx.api.get_function(function.identifier()).await.unwrap();
+ assert_eq!(response.name.as_deref(), Some("plus_one"));
+ assert_eq!(response.input_params.unwrap()[0].name(), "x");
+}
+
#[tokio::test]
async fn test_create_database() {
let ctx = setup_test_server(vec!["default"]).await;
diff --git a/crates/paimon/tests/rest_catalog_test.rs
b/crates/paimon/tests/rest_catalog_test.rs
index 722a7b2..849b699 100644
--- a/crates/paimon/tests/rest_catalog_test.rs
+++ b/crates/paimon/tests/rest_catalog_test.rs
@@ -1187,6 +1187,123 @@ async fn test_catalog_get_function() {
);
}
+#[tokio::test]
+async fn test_catalog_create_function() {
+ let ctx = setup_catalog(vec!["default"]).await;
+ let function = Function::new(
+ Identifier::new("default", "plus_one"),
+ Some(
+ serde_json::from_value(serde_json::json!([
+ {"id": 0, "name": "x", "type": "BIGINT"}
+ ]))
+ .unwrap(),
+ ),
+ Some(
+ serde_json::from_value(serde_json::json!([
+ {"id": 0, "name": "result", "type": "BIGINT"}
+ ]))
+ .unwrap(),
+ ),
+ true,
+ HashMap::from([(
+ "datafusion".to_string(),
+ FunctionDefinition::Sql {
+ definition: "x + 1".to_string(),
+ },
+ )]),
+ None,
+ HashMap::new(),
+ );
+
+ ctx.catalog.create_function(&function, false).await.unwrap();
+
+ let stored = ctx
+ .catalog
+ .get_function(function.identifier())
+ .await
+ .unwrap();
+ assert_eq!(stored, function);
+}
+
+#[tokio::test]
+async fn test_catalog_create_function_already_exists() {
+ let ctx = setup_catalog(vec!["default"]).await;
+ let function = Function::new(
+ Identifier::new("default", "answer"),
+ Some(Vec::new()),
+ Some(
+ serde_json::from_value(serde_json::json!([
+ {"id": 0, "name": "result", "type": "INT"}
+ ]))
+ .unwrap(),
+ ),
+ true,
+ HashMap::from([(
+ "datafusion".to_string(),
+ FunctionDefinition::Sql {
+ definition: "42".to_string(),
+ },
+ )]),
+ None,
+ HashMap::new(),
+ );
+ ctx.catalog.create_function(&function, false).await.unwrap();
+
+ let error = ctx
+ .catalog
+ .create_function(&function, false)
+ .await
+ .unwrap_err();
+
+ assert!(matches!(
+ error,
+ paimon::Error::FunctionAlreadyExist { full_name }
+ if full_name == "default.answer"
+ ));
+}
+
+#[tokio::test]
+async fn test_catalog_create_function_ignore_if_exists() {
+ let ctx = setup_catalog(vec!["default"]).await;
+ let function = Function::new(
+ Identifier::new("default", "answer"),
+ Some(Vec::new()),
+ Some(Vec::new()),
+ true,
+ HashMap::new(),
+ None,
+ HashMap::new(),
+ );
+ ctx.catalog.create_function(&function, false).await.unwrap();
+
+ ctx.catalog.create_function(&function, true).await.unwrap();
+}
+
+#[tokio::test]
+async fn test_catalog_create_function_missing_database() {
+ let ctx = setup_catalog(vec!["default"]).await;
+ let function = Function::new(
+ Identifier::new("missing_db", "answer"),
+ Some(Vec::new()),
+ Some(Vec::new()),
+ true,
+ HashMap::new(),
+ None,
+ HashMap::new(),
+ );
+
+ let error = ctx
+ .catalog
+ .create_function(&function, false)
+ .await
+ .unwrap_err();
+
+ assert!(matches!(
+ error,
+ paimon::Error::DatabaseNotExist { database } if database ==
"missing_db"
+ ));
+}
+
#[tokio::test]
async fn test_catalog_list_functions() {
let ctx = setup_catalog(vec!["default"]).await;
@@ -1286,6 +1403,24 @@ async fn
test_catalog_maps_unsupported_view_and_function_endpoints() {
ctx.catalog.list_functions("default").await.unwrap_err(),
paimon::Error::Unsupported { .. }
));
+ assert!(matches!(
+ ctx.catalog
+ .create_function(
+ &Function::new(
+ Identifier::new("default", "function"),
+ Some(Vec::new()),
+ Some(Vec::new()),
+ true,
+ HashMap::new(),
+ None,
+ HashMap::new(),
+ ),
+ false,
+ )
+ .await
+ .unwrap_err(),
+ paimon::Error::Unsupported { .. }
+ ));
assert!(matches!(
ctx.catalog
.get_function(&Identifier::new("default", "function"))
diff --git a/crates/paimon/tests/rest_object_models_test.rs
b/crates/paimon/tests/rest_object_models_test.rs
index 6e2806a..f4e3af1 100644
--- a/crates/paimon/tests/rest_object_models_test.rs
+++ b/crates/paimon/tests/rest_object_models_test.rs
@@ -17,7 +17,7 @@
use std::collections::HashMap;
-use paimon::api::CreateViewRequest;
+use paimon::api::{CreateFunctionRequest, CreateViewRequest};
use paimon::catalog::{Function, FunctionDefinition, Identifier, View,
ViewSchema};
use paimon::spec::DataField;
use serde_json::json;
@@ -209,3 +209,70 @@ fn function_binds_identifier_signature_and_definition() {
Some("length * width")
);
}
+
+#[test]
+fn create_function_request_serialization_contract() {
+ let input_params: Vec<DataField> = serde_json::from_value(json!([
+ {"id": 0, "name": "x", "type": "BIGINT"}
+ ]))
+ .unwrap();
+ let return_params: Vec<DataField> = serde_json::from_value(json!([
+ {"id": 0, "name": "result", "type": "BIGINT"}
+ ]))
+ .unwrap();
+ let function = Function::new(
+ Identifier::new("analytics", "plus_one"),
+ Some(input_params),
+ Some(return_params),
+ true,
+ HashMap::from([(
+ "datafusion".to_string(),
+ FunctionDefinition::Sql {
+ definition: "x + 1".to_string(),
+ },
+ )]),
+ None,
+ HashMap::new(),
+ );
+
+ assert_eq!(
+
serde_json::to_value(CreateFunctionRequest::from_function(&function)).unwrap(),
+ json!({
+ "name": "plus_one",
+ "inputParams": [{"id": 0, "name": "x", "type": "BIGINT"}],
+ "returnParams": [{"id": 0, "name": "result", "type": "BIGINT"}],
+ "deterministic": true,
+ "definitions": {
+ "datafusion": {"type": "sql", "definition": "x + 1"}
+ },
+ "comment": null,
+ "options": {}
+ })
+ );
+}
+
+#[test]
+fn create_function_request_serializes_zero_arguments_as_empty_array() {
+ let function = Function::new(
+ Identifier::new("analytics", "answer"),
+ Some(Vec::new()),
+ Some(
+ serde_json::from_value(json!([
+ {"id": 0, "name": "result", "type": "BIGINT"}
+ ]))
+ .unwrap(),
+ ),
+ true,
+ HashMap::from([(
+ "datafusion".to_string(),
+ FunctionDefinition::Sql {
+ definition: "42".to_string(),
+ },
+ )]),
+ None,
+ HashMap::new(),
+ );
+
+ let json =
serde_json::to_value(CreateFunctionRequest::from_function(&function)).unwrap();
+ assert_eq!(json["inputParams"], json!([]));
+}
diff --git a/docs/src/sql.md b/docs/src/sql.md
index 0a6073f..50623e4 100644
--- a/docs/src/sql.md
+++ b/docs/src/sql.md
@@ -40,9 +40,9 @@ Mosaic support is always available and currently read-only.
SQL queries can read
SQL support has two layers:
- DataFusion provides the parser, query planner, optimizer, execution engine,
expressions, scalar functions, aggregate functions, and window functions. SQL
statements that `SQLContext` does not intercept are delegated to DataFusion.
This includes the DataFusion SQL surface for `SELECT` queries, CTEs (including
recursive CTEs), subqueries, joins including `LATERAL` joins, SQL lambda
functions, grouping, `HAVING`, window clauses, `QUALIFY`, set operations,
`ORDER BY`, `LIMIT`/`OFFSET`, `EX [...]
-- Paimon-specific table management and row-level writes are implemented by
`SQLContext`. This includes Paimon `CREATE TABLE`, `ALTER TABLE`, `DROP TABLE`,
`CREATE TEMPORARY TABLE`, `CREATE TEMPORARY VIEW`, REST Catalog persistent
`CREATE VIEW`, `DROP TEMPORARY TABLE` / `VIEW`, `INSERT OVERWRITE ...
PARTITION`, `UPDATE`, `DELETE`, `MERGE INTO`, `TRUNCATE TABLE`, `ALTER TABLE
... DROP PARTITION`, `CALL sys.*`, Paimon time travel, and `SET` / `RESET
'paimon.*'`.
+- Paimon-specific table management and row-level writes are implemented by
`SQLContext`. This includes Paimon `CREATE TABLE`, `ALTER TABLE`, `DROP TABLE`,
`CREATE TEMPORARY TABLE`, `CREATE TEMPORARY VIEW`, REST Catalog persistent
`CREATE VIEW` and `CREATE FUNCTION`, `DROP TEMPORARY TABLE` / `VIEW`, `INSERT
OVERWRITE ... PARTITION`, `UPDATE`, `DELETE`, `MERGE INTO`, `TRUNCATE TABLE`,
`ALTER TABLE ... DROP PARTITION`, `CALL sys.*`, Paimon time travel, and `SET` /
`RESET 'paimon.*'`.
-Not every DataFusion DDL/DML statement maps to a Paimon table operation. For
Paimon catalogs, `CREATE EXTERNAL TABLE`, `LOCATION`, `CREATE MATERIALIZED
VIEW`, `CREATE FUNCTION`, and persistent `CREATE TABLE AS SELECT` are rejected
or not implemented. DataFusion `COPY` can export query results to files; it
does not create or commit Paimon table files.
+Not every DataFusion DDL/DML statement maps to a Paimon table operation. For
Paimon catalogs, `CREATE EXTERNAL TABLE`, `LOCATION`, `CREATE MATERIALIZED
VIEW`, and persistent `CREATE TABLE AS SELECT` are rejected or not implemented.
Persistent `CREATE FUNCTION` is supported only for the REST Catalog SQL scalar
form documented below. DataFusion `COPY` can export query results to files; it
does not create or commit Paimon table files.
For the exact delegated SQL grammar, see the [DataFusion SQL
Reference](https://datafusion.apache.org/user-guide/sql/index.html).
@@ -78,9 +78,10 @@ internally for `SET`/`RESET` support.
### REST Catalog Views and SQL Functions
-When the registered catalog is a Paimon REST Catalog, `SQLContext` can read and
-execute persistent views and SQL scalar functions in the catalog. It can also
-create persistent views with this syntax:
+When the registered catalog is a Paimon REST Catalog, `SQLContext` can read,
+execute, and create persistent views and SQL scalar functions.
+
+Create a persistent view with this syntax:
```sql
CREATE VIEW [IF NOT EXISTS] view_name [(column_name, ...)] AS query;
@@ -109,8 +110,7 @@ dialect definition.
Persistent `CREATE VIEW` is currently implemented by REST Catalog. Other
catalog implementations may return `Unsupported`. `CREATE OR REPLACE VIEW`,
materialized/secure views, view comments or options, vendor-specific modifiers,
-and persistent `ALTER VIEW` / `DROP VIEW` are not supported. Persistent
-`CREATE FUNCTION`, `ALTER FUNCTION`, and `DROP FUNCTION` are also not
supported.
+and persistent `ALTER VIEW` / `DROP VIEW` are not supported.
Persistent views resolve through the normal DataFusion catalog path, so they
can be queried wherever a table can be used:
@@ -136,6 +136,57 @@ SELECT normalize_score(score) FROM scores;
SELECT paimon.reporting.normalize_score(score) FROM scores;
```
+Create a persistent REST SQL scalar function with this syntax:
+
+```sql
+CREATE FUNCTION [IF NOT EXISTS] function_name(
+ [parameter_name data_type, ...]
+)
+RETURNS data_type
+[LANGUAGE SQL]
+RETURN scalar_expression;
+```
+
+For example:
+
+```sql
+CREATE FUNCTION paimon.reporting.add_tax(amount DECIMAL(12, 2))
+RETURNS DECIMAL(12, 2)
+RETURN amount * DECIMAL '1.10';
+
+SELECT add_tax(total) FROM orders;
+```
+
+Bare, two-part (`database.function`), and three-part
+(`catalog.database.function`) names are accepted as creation targets. Unquoted
+names are normalized and quoted names are preserved. Calls remain limited to
+bare or three-part names; two-part function calls are not supported.
+
+Parameters must be named and cannot have modes or defaults. Zero parameters
+are allowed. Inputs and the single return value are stored as nullable fields;
+parameter IDs start at zero and the return field has ID `0` and name `result`.
+The canonical, unexpanded `RETURN` expression is stored in
+`definitions.datafusion` with `type: "sql"`.
+
+`LANGUAGE SQL` is optional and SQL is the default, matching Databricks SQL
+function syntax. `IMMUTABLE` is not required; when omitted, determinism is
+inferred from the planned expression. An explicit `IMMUTABLE` clause remains
+accepted for compatibility.
+
+Before the REST create request is sent, `SQLContext` expands dependencies using
+the new function as a candidate, validates argument substitution and the
+declared return cast, and builds both logical and physical DataFusion plans in
+the function's owning catalog/database. This rejects undeclared identifiers,
+recursive dependencies (including indirect recursion), non-deterministic REST
+dependencies, subqueries/table access, aggregate or window functions,
+Stable/Volatile DataFusion functions, and incompatible return types. The
+function is stored as deterministic only after the planned expression passes
+these checks.
+
+`IF NOT EXISTS` still validates the proposed definition first. The REST server
+then handles the create atomically; only an already-existing function error is
+ignored.
+
A function is executable only when all of the following are true:
- `definitions.datafusion` exists and has `type: "sql"`;
@@ -157,7 +208,11 @@ Function expansion is implemented by `SQLContext::sql`.
Queries executed
directly through a raw DataFusion `SessionContext` do not expand REST SQL
functions. Two-part function names such as `database.function(...)`, lambda or
file definitions, aggregate/table/multi-return functions, and non-deterministic
-functions are not supported by this read-only integration.
+functions are not supported. `CREATE OR REPLACE/ALTER/TEMPORARY FUNCTION`,
+non-SQL bodies, `STABLE`/`VOLATILE`, null-input/parallel/security/SET clauses,
+options/remote functions, and persistent `ALTER FUNCTION` / `DROP FUNCTION`
+are also not supported. Catalog implementations other than REST Catalog may
+return `Unsupported` for persistent function creation.
## Data Types
