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 1bf775e Support REST catalog views and SQL functions (#494)
1bf775e is described below
commit 1bf775ead20d6864ede7555624933234cc6bcfcb
Author: Jingsong Lee <[email protected]>
AuthorDate: Fri Jul 10 21:57:15 2026 +0800
Support REST catalog views and SQL functions (#494)
---
bindings/python/src/context.rs | 8 +-
crates/integrations/datafusion/README.md | 26 +
crates/integrations/datafusion/src/catalog.rs | 422 +++++++++-
crates/integrations/datafusion/src/lib.rs | 1 +
crates/integrations/datafusion/src/sql_context.rs | 853 ++++++++++++++++++++-
crates/integrations/datafusion/src/sql_function.rs | 459 +++++++++++
.../integrations/datafusion/tests/read_tables.rs | 8 +-
.../datafusion/tests/sql_context_tests.rs | 8 +-
crates/paimon/src/api/api_response.rs | 118 ++-
crates/paimon/src/api/mod.rs | 5 +-
crates/paimon/src/api/resource_paths.rs | 78 ++
crates/paimon/src/api/rest_api.rs | 132 +++-
crates/paimon/src/catalog/function.rs | 163 ++++
crates/paimon/src/catalog/mod.rs | 36 +
crates/paimon/src/catalog/rest/rest_catalog.rs | 78 ++
crates/paimon/src/catalog/view.rs | 106 +++
crates/paimon/src/error.rs | 4 +
crates/paimon/tests/mock_server.rs | 200 ++++-
crates/paimon/tests/rest_api_test.rs | 126 ++-
crates/paimon/tests/rest_catalog_test.rs | 184 ++++-
crates/paimon/tests/rest_object_models_test.rs | 154 ++++
docs/src/sql.md | 54 ++
22 files changed, 3182 insertions(+), 41 deletions(-)
diff --git a/bindings/python/src/context.rs b/bindings/python/src/context.rs
index 8cddde0..d92d8ee 100644
--- a/bindings/python/src/context.rs
+++ b/bindings/python/src/context.rs
@@ -75,7 +75,13 @@ impl PaimonCatalog {
#[new]
fn new(catalog_options: HashMap<String, String>) -> PyResult<Self> {
let catalog = build_paimon_catalog(catalog_options)?;
- let provider =
Arc::new(PaimonCatalogProvider::new(Arc::clone(&catalog)));
+ let provider = Arc::new(PaimonCatalogProvider::new(
+ None,
+ Arc::clone(&catalog),
+ Default::default(),
+ Default::default(),
+ None,
+ ));
Ok(Self { catalog, provider })
}
diff --git a/crates/integrations/datafusion/README.md
b/crates/integrations/datafusion/README.md
index 92f8a29..cc7a3a1 100644
--- a/crates/integrations/datafusion/README.md
+++ b/crates/integrations/datafusion/README.md
@@ -24,4 +24,30 @@
This crate contains the integration of [Apache
DataFusion](https://datafusion.apache.org/) and [Apache
Paimon](https://paimon.apache.org/).
+## REST Catalog views and SQL functions
+
+`SQLContext` can read persistent views and SQL scalar functions that already
exist in a Paimon
+REST Catalog. No view or function DDL is added.
+
+- A persistent view is resolved lazily like a table. The `datafusion` dialect
is preferred and the
+ default view query is used when that dialect is absent. Unqualified
relations inside the view
+ resolve in the view's owning catalog and database.
+- A SQL function can be called as `function(args...)` in the current
catalog/database or as
+ `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.
+- Only reads and execution are supported. Lambda/file functions, named
arguments, multiple return
+ values, non-deterministic functions, and calls made directly through a raw
DataFusion
+ `SessionContext` are not supported.
+
+Use `SQLContext::sql` for function expansion:
+
+```rust,ignore
+let mut ctx = paimon_datafusion::SQLContext::new();
+ctx.register_catalog("paimon", rest_catalog).await?;
+
+let view = ctx.sql("SELECT * FROM analytics_view").await?;
+let function = ctx.sql("SELECT normalize_score(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/catalog.rs
b/crates/integrations/datafusion/src/catalog.rs
index 7e32bae..5bcdd10 100644
--- a/crates/integrations/datafusion/src/catalog.rs
+++ b/crates/integrations/datafusion/src/catalog.rs
@@ -17,17 +17,23 @@
//! Paimon catalog integration for DataFusion.
-use std::collections::HashMap;
+use std::collections::{HashMap, HashSet, VecDeque};
use std::fmt::Debug;
use std::sync::Arc;
use std::sync::RwLock;
use async_trait::async_trait;
use datafusion::catalog::{CatalogProvider, MemorySchemaProvider,
SchemaProvider};
-use datafusion::common::plan_datafusion_err;
+use datafusion::common::{plan_datafusion_err, Column};
use datafusion::datasource::TableProvider;
use datafusion::error::Result as DFResult;
-use paimon::catalog::{Catalog, Identifier};
+use datafusion::execution::SessionState;
+use datafusion::logical_expr::{expr_fn::cast, Expr, LogicalPlan,
LogicalPlanBuilder};
+use datafusion::sql::planner::IdentNormalizer;
+use datafusion::sql::sqlparser::ast::{Ident, ObjectName, Query, Statement,
Visit, Visitor};
+use datafusion::sql::sqlparser::dialect::GenericDialect;
+use datafusion::sql::sqlparser::parser::Parser;
+use paimon::catalog::{Catalog, Identifier, View};
use crate::error::to_datafusion_error;
use crate::runtime::{await_with_runtime, block_on_with_runtime};
@@ -35,12 +41,15 @@ use crate::system_tables;
use crate::table::PaimonTableProvider;
use crate::{BlobReaderRegistry, DynamicOptions};
+pub(crate) type SessionStateProvider = Arc<dyn Fn() -> Option<SessionState> +
Send + Sync>;
+
/// Provides an interface to manage and access multiple schemas (databases)
/// within a Paimon [`Catalog`].
///
/// This provider uses lazy loading - databases and tables are fetched
/// on-demand from the catalog, ensuring data is always fresh.
pub struct PaimonCatalogProvider {
+ catalog_name: Option<String>,
/// Reference to the Paimon catalog.
catalog: Arc<dyn Catalog>,
/// Session-scoped dynamic options shared with the SQL context.
@@ -54,6 +63,7 @@ pub struct PaimonCatalogProvider {
/// becoming invisible or stale, which is recoverable by re-registering it.
temp_tables: Arc<RwLock<HashMap<String, Arc<MemorySchemaProvider>>>>,
blob_reader_registry: BlobReaderRegistry,
+ session_state: Option<SessionStateProvider>,
}
impl Debug for PaimonCatalogProvider {
@@ -64,29 +74,20 @@ impl Debug for PaimonCatalogProvider {
impl PaimonCatalogProvider {
/// Creates a new [`PaimonCatalogProvider`].
- ///
- /// For standalone use without `SET`/`RESET` support.
- /// When used via [`SQLContext`], the handler creates the provider
- /// internally with shared dynamic options.
- pub fn new(catalog: Arc<dyn Catalog>) -> Self {
- PaimonCatalogProvider {
- catalog,
- dynamic_options: Default::default(),
- temp_tables: Arc::new(RwLock::new(HashMap::new())),
- blob_reader_registry: BlobReaderRegistry::default(),
- }
- }
-
- pub(crate) fn with_dynamic_options(
+ pub fn new(
+ catalog_name: Option<String>,
catalog: Arc<dyn Catalog>,
dynamic_options: DynamicOptions,
blob_reader_registry: BlobReaderRegistry,
+ session_state: Option<SessionStateProvider>,
) -> Self {
PaimonCatalogProvider {
+ catalog_name,
catalog,
dynamic_options,
temp_tables: Arc::new(RwLock::new(HashMap::new())),
blob_reader_registry,
+ session_state,
}
}
}
@@ -109,6 +110,8 @@ impl CatalogProvider for PaimonCatalogProvider {
let catalog = Arc::clone(&self.catalog);
let dynamic_options = Arc::clone(&self.dynamic_options);
let blob_reader_registry = self.blob_reader_registry.clone();
+ let catalog_name = self.catalog_name.clone();
+ let session_state = self.session_state.clone();
let name = name.to_string();
let temp_provider = {
@@ -120,20 +123,24 @@ impl CatalogProvider for PaimonCatalogProvider {
async move {
match catalog.get_database(&name).await {
Ok(_) => Some(Arc::new(PaimonSchemaProvider::new(
+ catalog_name,
Arc::clone(&catalog),
name,
dynamic_options,
temp_provider,
blob_reader_registry,
+ session_state,
)) as Arc<dyn SchemaProvider>),
Err(paimon::Error::DatabaseNotExist { .. }) => {
if temp_provider.is_some() {
Some(Arc::new(PaimonSchemaProvider::new(
+ catalog_name,
Arc::clone(&catalog),
name,
dynamic_options,
temp_provider,
blob_reader_registry,
+ session_state,
)) as Arc<dyn SchemaProvider>)
} else {
None
@@ -157,6 +164,8 @@ impl CatalogProvider for PaimonCatalogProvider {
let catalog = Arc::clone(&self.catalog);
let dynamic_options = Arc::clone(&self.dynamic_options);
let blob_reader_registry = self.blob_reader_registry.clone();
+ let catalog_name = self.catalog_name.clone();
+ let session_state = self.session_state.clone();
let name = name.to_string();
block_on_with_runtime(
async move {
@@ -165,11 +174,13 @@ impl CatalogProvider for PaimonCatalogProvider {
.await
.map_err(to_datafusion_error)?;
Ok(Some(Arc::new(PaimonSchemaProvider::new(
+ catalog_name,
Arc::clone(&catalog),
name,
dynamic_options,
None,
blob_reader_registry,
+ session_state,
)) as Arc<dyn SchemaProvider>))
},
"paimon catalog access thread panicked",
@@ -184,6 +195,8 @@ impl CatalogProvider for PaimonCatalogProvider {
let catalog = Arc::clone(&self.catalog);
let dynamic_options = Arc::clone(&self.dynamic_options);
let blob_reader_registry = self.blob_reader_registry.clone();
+ let catalog_name = self.catalog_name.clone();
+ let session_state = self.session_state.clone();
let name = name.to_string();
block_on_with_runtime(
async move {
@@ -192,11 +205,13 @@ impl CatalogProvider for PaimonCatalogProvider {
.await
.map_err(to_datafusion_error)?;
Ok(Some(Arc::new(PaimonSchemaProvider::new(
+ catalog_name,
Arc::clone(&catalog),
name,
dynamic_options,
None,
blob_reader_registry,
+ session_state,
)) as Arc<dyn SchemaProvider>))
},
"paimon catalog access thread panicked",
@@ -287,6 +302,7 @@ impl PaimonCatalogProvider {
///
/// Tables are loaded lazily when accessed via the `table()` method.
pub struct PaimonSchemaProvider {
+ catalog_name: Option<String>,
/// Reference to the Paimon catalog.
catalog: Arc<dyn Catalog>,
/// Database name this schema represents.
@@ -296,6 +312,7 @@ pub struct PaimonSchemaProvider {
/// Optional temporary in-memory provider for temp tables and views.
temp_provider: Option<Arc<MemorySchemaProvider>>,
blob_reader_registry: BlobReaderRegistry,
+ session_state: Option<SessionStateProvider>,
}
impl Debug for PaimonSchemaProvider {
@@ -308,20 +325,24 @@ impl Debug for PaimonSchemaProvider {
}
impl PaimonSchemaProvider {
- /// Creates a new [`PaimonSchemaProvider`] with shared dynamic options.
+ /// Creates a new [`PaimonSchemaProvider`].
pub fn new(
+ catalog_name: Option<String>,
catalog: Arc<dyn Catalog>,
database: String,
dynamic_options: DynamicOptions,
temp_provider: Option<Arc<MemorySchemaProvider>>,
blob_reader_registry: BlobReaderRegistry,
+ session_state: Option<SessionStateProvider>,
) -> Self {
PaimonSchemaProvider {
+ catalog_name,
catalog,
database,
dynamic_options,
temp_provider,
blob_reader_registry,
+ session_state,
}
}
}
@@ -335,13 +356,21 @@ impl SchemaProvider for PaimonSchemaProvider {
{
let db = database.clone();
async move {
- match catalog.list_tables(&db).await {
+ let mut names = match catalog.list_tables(&db).await {
Ok(names) => names,
Err(e) => {
log::error!("failed to list tables in '{}': {e}",
db);
vec![]
}
+ };
+ match catalog.list_views(&db).await {
+ Ok(views) => names.extend(views),
+ Err(paimon::Error::Unsupported { .. }) => {}
+ Err(error) => {
+ log::error!("failed to list views in '{}':
{error}", db);
+ }
}
+ names
}
},
"paimon catalog access thread panicked",
@@ -378,6 +407,8 @@ impl SchemaProvider for PaimonSchemaProvider {
let catalog = Arc::clone(&self.catalog);
let dynamic_options = Arc::clone(&self.dynamic_options);
let blob_reader_registry = self.blob_reader_registry.clone();
+ let catalog_name = self.catalog_name.clone();
+ let session_state = self.session_state.clone();
let identifier = Identifier::new(self.database.clone(),
object.table().to_string());
let branch = object.branch().map(str::to_string);
await_with_runtime(async move {
@@ -412,7 +443,51 @@ impl SchemaProvider for PaimonSchemaProvider {
};
Ok(Some(Arc::new(provider) as Arc<dyn TableProvider>))
}
- Err(paimon::Error::TableNotExist { .. }) => Ok(None),
+ Err(paimon::Error::TableNotExist { .. }) => {
+ if branch.is_some() {
+ return Ok(None);
+ }
+ let view = match catalog.get_view(&identifier).await {
+ Ok(view) => view,
+ Err(paimon::Error::ViewNotExist { .. })
+ | Err(paimon::Error::Unsupported { .. }) => return
Ok(None),
+ Err(error) => return Err(to_datafusion_error(error)),
+ };
+ let catalog_name = catalog_name.ok_or_else(|| {
+ plan_datafusion_err!(
+ "REST catalog view '{}' requires a session-aware
catalog provider",
+ identifier.full_name()
+ )
+ })?;
+ validate_view_dependencies(&catalog, &catalog_name,
&view).await?;
+ let mut state = session_state
+ .and_then(|provider| provider())
+ .ok_or_else(|| {
+ plan_datafusion_err!(
+ "DataFusion session is unavailable while
planning REST catalog view '{}'",
+ identifier.full_name()
+ )
+ })?;
+ state.config_mut().options_mut().catalog.default_catalog =
+ catalog_name.clone();
+ state.config_mut().options_mut().catalog.default_schema =
+ identifier.database().to_string();
+ let catalogs =
+ HashMap::from([(catalog_name.clone(),
Arc::clone(&catalog))]);
+ let query = crate::sql_function::expand_sql(
+ view.query_for("datafusion"),
+ &catalogs,
+ &catalog_name,
+ identifier.database(),
+ )
+ .await?;
+ let plan = state.create_logical_plan(&query).await?;
+ let plan = enforce_view_schema(plan, &view)?;
+ Ok(Some(Arc::new(datafusion::datasource::ViewTable::new(
+ plan,
+ Some(query),
+ )) as Arc<dyn TableProvider>))
+ }
Err(e) => Err(to_datafusion_error(e)),
}
})
@@ -458,7 +533,20 @@ impl SchemaProvider for PaimonSchemaProvider {
true
}
}
- Err(paimon::Error::TableNotExist { .. }) => false,
+ Err(paimon::Error::TableNotExist { .. }) => {
+ if branch.is_some() {
+ return false;
+ }
+ match catalog.get_view(&identifier).await {
+ Ok(_) => true,
+ Err(paimon::Error::ViewNotExist { .. })
+ | Err(paimon::Error::Unsupported { .. }) => false,
+ Err(error) => {
+ log::error!("failed to check view '{}':
{error}", identifier);
+ false
+ }
+ }
+ }
Err(e) => {
log::error!("failed to check table '{}': {e}",
identifier);
false
@@ -501,3 +589,295 @@ impl SchemaProvider for PaimonSchemaProvider {
)
}
}
+
+fn enforce_view_schema(plan: LogicalPlan, view: &View) ->
DFResult<LogicalPlan> {
+ let declared_fields = view.schema().fields();
+ let actual_fields = plan.schema().fields();
+ if actual_fields.len() != declared_fields.len() {
+ return Err(plan_datafusion_err!(
+ "REST catalog view '{}' declares {} fields but its query returns
{}",
+ view.full_name(),
+ declared_fields.len(),
+ actual_fields.len()
+ ));
+ }
+
+ let expressions = declared_fields
+ .iter()
+ .enumerate()
+ .map(|(index, declared)| {
+ let (qualifier, actual) = plan.schema().qualified_field(index);
+ let column = match qualifier {
+ Some(qualifier) => Column::new(Some(qualifier.clone()),
actual.name()),
+ None => Column::new_unqualified(actual.name()),
+ };
+ let target_type =
paimon::arrow::paimon_type_to_arrow(declared.data_type())
+ .map_err(to_datafusion_error)?;
+ Ok(cast(Expr::Column(column), target_type).alias(declared.name()))
+ })
+ .collect::<DFResult<Vec<_>>>()?;
+
+ LogicalPlanBuilder::from(plan).project(expressions)?.build()
+}
+
+const MAX_VIEW_DEPENDENCIES: usize = 64;
+
+async fn validate_view_dependencies(
+ catalog: &Arc<dyn Catalog>,
+ catalog_name: &str,
+ root: &View,
+) -> DFResult<()> {
+ let mut queue = VecDeque::from([root.clone()]);
+ let mut loaded = HashSet::from([root.identifier().clone()]);
+ let mut dependencies = HashMap::<Identifier, Vec<Identifier>>::new();
+
+ while let Some(view) = queue.pop_front() {
+ let candidates = view_relation_identifiers(&view, catalog_name)?;
+ let mut view_dependencies = Vec::new();
+ for identifier in candidates {
+ match catalog.get_table(&identifier).await {
+ Ok(_) => continue,
+ Err(paimon::Error::TableNotExist { .. })
+ | Err(paimon::Error::Unsupported { .. }) => {}
+ Err(error) => return Err(to_datafusion_error(error)),
+ }
+
+ let dependency = match catalog.get_view(&identifier).await {
+ Ok(view) => view,
+ Err(paimon::Error::ViewNotExist { .. })
+ | Err(paimon::Error::Unsupported { .. }) => continue,
+ Err(error) => return Err(to_datafusion_error(error)),
+ };
+ view_dependencies.push(identifier.clone());
+ if loaded.insert(identifier) {
+ if loaded.len() > MAX_VIEW_DEPENDENCIES {
+ return Err(plan_datafusion_err!(
+ "REST catalog view '{}' exceeds the dependency limit
of {}",
+ root.full_name(),
+ MAX_VIEW_DEPENDENCIES
+ ));
+ }
+ queue.push_back(dependency);
+ }
+ }
+ dependencies.insert(view.identifier().clone(), view_dependencies);
+
+ if let Some(cycle) = find_view_dependency_cycle(&dependencies) {
+ let path = cycle
+ .iter()
+ .map(Identifier::full_name)
+ .collect::<Vec<_>>()
+ .join(" -> ");
+ return Err(plan_datafusion_err!(
+ "recursive REST catalog view dependency detected: {path}"
+ ));
+ }
+ }
+ Ok(())
+}
+
+fn view_relation_identifiers(view: &View, catalog_name: &str) ->
DFResult<Vec<Identifier>> {
+ let statements =
+ Parser::parse_sql(&GenericDialect {},
view.query_for("datafusion")).map_err(|error| {
+ plan_datafusion_err!(
+ "Invalid SQL for REST catalog view '{}': {error}",
+ view.full_name()
+ )
+ })?;
+ if statements.len() != 1 {
+ return Err(plan_datafusion_err!(
+ "REST catalog view '{}' must contain exactly one SQL statement",
+ view.full_name()
+ ));
+ }
+ if !matches!(statements.first(), Some(Statement::Query(_))) {
+ return Err(plan_datafusion_err!(
+ "REST catalog view '{}' must contain a single read-only query",
+ view.full_name()
+ ));
+ }
+
+ let mut visitor = ViewRelationVisitor::new(catalog_name,
view.identifier().database());
+ let _: std::ops::ControlFlow<()> = statements.visit(&mut visitor);
+ Ok(visitor.identifiers)
+}
+
+type SqlIdentifierKey = String;
+
+struct QueryCteScope {
+ visible: HashSet<SqlIdentifierKey>,
+ cte_query_visibility: HashMap<usize, HashSet<SqlIdentifierKey>>,
+}
+
+struct ViewRelationVisitor<'a> {
+ catalog_name: &'a str,
+ current_database: &'a str,
+ scopes: Vec<QueryCteScope>,
+ identifiers: Vec<Identifier>,
+}
+
+impl<'a> ViewRelationVisitor<'a> {
+ fn new(catalog_name: &'a str, current_database: &'a str) -> Self {
+ Self {
+ catalog_name,
+ current_database,
+ scopes: Vec::new(),
+ identifiers: Vec::new(),
+ }
+ }
+}
+
+impl Visitor for ViewRelationVisitor<'_> {
+ type Break = ();
+
+ fn pre_visit_query(&mut self, query: &Query) ->
std::ops::ControlFlow<Self::Break> {
+ let query_address = query as *const Query as usize;
+ let inherited = self
+ .scopes
+ .last()
+ .map(|scope| {
+ scope
+ .cte_query_visibility
+ .get(&query_address)
+ .unwrap_or(&scope.visible)
+ .clone()
+ })
+ .unwrap_or_default();
+ let mut visible = inherited.clone();
+ let mut cte_query_visibility = HashMap::new();
+
+ if let Some(with) = &query.with {
+ let local_ctes = with
+ .cte_tables
+ .iter()
+ .map(|cte| sql_identifier_key(&cte.alias.name))
+ .collect::<Vec<_>>();
+ if with.recursive {
+ visible.extend(local_ctes);
+ for cte in &with.cte_tables {
+ cte_query_visibility
+ .insert(cte.query.as_ref() as *const Query as usize,
visible.clone());
+ }
+ } else {
+ for (cte, alias) in with.cte_tables.iter().zip(local_ctes) {
+ cte_query_visibility
+ .insert(cte.query.as_ref() as *const Query as usize,
visible.clone());
+ visible.insert(alias);
+ }
+ }
+ }
+
+ self.scopes.push(QueryCteScope {
+ visible,
+ cte_query_visibility,
+ });
+ std::ops::ControlFlow::Continue(())
+ }
+
+ fn post_visit_query(&mut self, _query: &Query) ->
std::ops::ControlFlow<Self::Break> {
+ self.scopes.pop();
+ std::ops::ControlFlow::Continue(())
+ }
+
+ fn pre_visit_relation(&mut self, relation: &ObjectName) ->
std::ops::ControlFlow<Self::Break> {
+ let is_cte = match relation.0.as_slice() {
+ [part] => part.as_ident().is_some_and(|identifier| {
+ self.scopes
+ .last()
+ .is_some_and(|scope|
scope.visible.contains(&sql_identifier_key(identifier)))
+ }),
+ _ => false,
+ };
+ if !is_cte {
+ if let Some(identifier) =
+ relation_identifier(relation, self.catalog_name,
self.current_database)
+ {
+ self.identifiers.push(identifier);
+ }
+ }
+ std::ops::ControlFlow::Continue(())
+ }
+}
+
+fn sql_identifier_key(identifier: &Ident) -> SqlIdentifierKey {
+ IdentNormalizer::default().normalize(identifier.clone())
+}
+
+fn relation_identifier(
+ relation: &ObjectName,
+ catalog_name: &str,
+ current_database: &str,
+) -> Option<Identifier> {
+ let parts = relation
+ .0
+ .iter()
+ .map(|part| part.as_ident().map(sql_identifier_key))
+ .collect::<Option<Vec<_>>>()?;
+ match parts.as_slice() {
+ [object] => Some(Identifier::new(current_database, object.as_str())),
+ [database, object] => Some(Identifier::new(database.as_str(),
object.as_str())),
+ [catalog, database, object] if catalog == catalog_name => {
+ Some(Identifier::new(database.as_str(), object.as_str()))
+ }
+ _ => None,
+ }
+}
+
+fn find_view_dependency_cycle(
+ dependencies: &HashMap<Identifier, Vec<Identifier>>,
+) -> Option<Vec<Identifier>> {
+ fn visit(
+ identifier: &Identifier,
+ dependencies: &HashMap<Identifier, Vec<Identifier>>,
+ finished: &mut HashSet<Identifier>,
+ path: &mut Vec<Identifier>,
+ ) -> Option<Vec<Identifier>> {
+ if let Some(start) = path.iter().position(|entry| entry == identifier)
{
+ let mut cycle = path[start..].to_vec();
+ cycle.push(identifier.clone());
+ return Some(cycle);
+ }
+ if finished.contains(identifier) {
+ return None;
+ }
+
+ path.push(identifier.clone());
+ if let Some(next_identifiers) = dependencies.get(identifier) {
+ for next in next_identifiers {
+ if let Some(cycle) = visit(next, dependencies, finished, path)
{
+ return Some(cycle);
+ }
+ }
+ }
+ path.pop();
+ finished.insert(identifier.clone());
+ None
+ }
+
+ let mut finished = HashSet::new();
+ for identifier in dependencies.keys() {
+ if let Some(cycle) = visit(identifier, dependencies, &mut finished,
&mut Vec::new()) {
+ return Some(cycle);
+ }
+ }
+ None
+}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+
+ #[test]
+ fn relation_identifiers_follow_datafusion_normalization() {
+ let relation = ObjectName(vec![
+
datafusion::sql::sqlparser::ast::ObjectNamePart::Identifier(Ident::new("PAIMON")),
+
datafusion::sql::sqlparser::ast::ObjectNamePart::Identifier(Ident::new("DEFAULT")),
+
datafusion::sql::sqlparser::ast::ObjectNamePart::Identifier(Ident::new("ANSWER_VIEW")),
+ ]);
+
+ assert_eq!(
+ relation_identifier(&relation, "paimon", "unused"),
+ Some(Identifier::new("default", "answer_view"))
+ );
+ }
+}
diff --git a/crates/integrations/datafusion/src/lib.rs
b/crates/integrations/datafusion/src/lib.rs
index 9c1d914..a60d52b 100644
--- a/crates/integrations/datafusion/src/lib.rs
+++ b/crates/integrations/datafusion/src/lib.rs
@@ -52,6 +52,7 @@ mod procedures;
mod relation_planner;
pub mod runtime;
mod sql_context;
+mod sql_function;
mod system_tables;
mod table;
mod table_function_args;
diff --git a/crates/integrations/datafusion/src/sql_context.rs
b/crates/integrations/datafusion/src/sql_context.rs
index 1513ea9..d136492 100644
--- a/crates/integrations/datafusion/src/sql_context.rs
+++ b/crates/integrations/datafusion/src/sql_context.rs
@@ -175,12 +175,17 @@ impl SQLContext {
Err(e) => return Err(DataFusionError::External(Box::new(e))),
}
}
+ let weak_state = self.ctx.state_weak_ref();
+ let session_state: crate::catalog::SessionStateProvider =
+ Arc::new(move || weak_state.upgrade().map(|state|
state.read().clone()));
self.ctx.register_catalog(
&catalog_name,
-
Arc::new(crate::catalog::PaimonCatalogProvider::with_dynamic_options(
+ Arc::new(crate::catalog::PaimonCatalogProvider::new(
+ Some(catalog_name.clone()),
catalog.clone(),
self.dynamic_options.clone(),
self.blob_reader_registry.clone(),
+ Some(session_state),
)),
);
register_table_functions(&self.ctx, &catalog,
default_db.unwrap_or("default"));
@@ -475,6 +480,24 @@ impl SQLContext {
)
.await
}
+ Statement::Query(_) | Statement::Explain { .. } => {
+ let current_catalog = self.current_catalog_name();
+ let current_database = self
+ .ctx
+ .state()
+ .config_options()
+ .catalog
+ .default_schema
+ .clone();
+ let expanded = crate::sql_function::expand_statement(
+ statements[0].clone(),
+ &self.catalogs,
+ ¤t_catalog,
+ ¤t_database,
+ )
+ .await?;
+ self.ctx.sql(&expanded.to_string()).await
+ }
_ => self.ctx.sql(sql).await,
}
}
@@ -602,8 +625,23 @@ impl SQLContext {
);
}
- // Execute the rewritten SQL; tracker auto-deregisters on drop
- self.ctx.sql(&rewritten_sql).await
+ // Execute the rewritten SQL; tracker auto-deregisters on drop.
+ let current_catalog = self.current_catalog_name();
+ let current_database = self
+ .ctx
+ .state()
+ .config_options()
+ .catalog
+ .default_schema
+ .clone();
+ let expanded = crate::sql_function::expand_sql(
+ &rewritten_sql,
+ &self.catalogs,
+ ¤t_catalog,
+ ¤t_database,
+ )
+ .await?;
+ self.ctx.sql(&expanded).await
}
/// Parse a timestamp string to milliseconds since epoch (using local
timezone).
@@ -2653,7 +2691,9 @@ mod tests {
use async_trait::async_trait;
use paimon::catalog::Database;
- use paimon::spec::{DataType as PaimonDataType, IntType, Schema as
PaimonSchema};
+ use paimon::spec::{
+ DataField as PaimonDataField, DataType as PaimonDataType, IntType,
Schema as PaimonSchema,
+ };
use paimon::table::Table;
// ==================== Mock Catalog ====================
@@ -2681,6 +2721,8 @@ mod tests {
struct MockCatalog {
calls: Mutex<Vec<CatalogCall>>,
existing_table: Mutex<Option<Table>>,
+ functions: Mutex<HashMap<Identifier, paimon::catalog::Function>>,
+ views: Mutex<HashMap<Identifier, paimon::catalog::View>>,
}
impl MockCatalog {
@@ -2688,12 +2730,28 @@ mod tests {
Self {
calls: Mutex::new(Vec::new()),
existing_table: Mutex::new(None),
+ functions: Mutex::new(HashMap::new()),
+ views: Mutex::new(HashMap::new()),
}
}
fn take_calls(&self) -> Vec<CatalogCall> {
std::mem::take(&mut *self.calls.lock().unwrap())
}
+
+ fn add_function(&self, function: paimon::catalog::Function) {
+ self.functions
+ .lock()
+ .unwrap()
+ .insert(function.identifier().clone(), function);
+ }
+
+ fn add_view(&self, view: paimon::catalog::View) {
+ self.views
+ .lock()
+ .unwrap()
+ .insert(view.identifier().clone(), view);
+ }
}
#[async_trait]
@@ -2710,9 +2768,7 @@ mod tests {
Ok(())
}
async fn get_database(&self, _name: &str) -> paimon::Result<Database> {
- Err(paimon::Error::DatabaseNotExist {
- database: _name.to_string(),
- })
+ Ok(Database::new(_name.to_string(), HashMap::new(), None))
}
async fn drop_database(
&self,
@@ -2779,6 +2835,53 @@ mod tests {
});
Ok(())
}
+
+ async fn list_functions(&self, database_name: &str) ->
paimon::Result<Vec<String>> {
+ Ok(self
+ .functions
+ .lock()
+ .unwrap()
+ .keys()
+ .filter(|identifier| identifier.database() == database_name)
+ .map(|identifier| identifier.object().to_string())
+ .collect())
+ }
+
+ async fn get_function(
+ &self,
+ identifier: &Identifier,
+ ) -> paimon::Result<paimon::catalog::Function> {
+ self.functions
+ .lock()
+ .unwrap()
+ .get(identifier)
+ .cloned()
+ .ok_or_else(|| paimon::Error::FunctionNotExist {
+ full_name: identifier.full_name(),
+ })
+ }
+
+ async fn list_views(&self, database_name: &str) ->
paimon::Result<Vec<String>> {
+ Ok(self
+ .views
+ .lock()
+ .unwrap()
+ .keys()
+ .filter(|identifier| identifier.database() == database_name)
+ .map(|identifier| identifier.object().to_string())
+ .collect())
+ }
+
+ async fn get_view(&self, identifier: &Identifier) ->
paimon::Result<paimon::catalog::View> {
+ self.views
+ .lock()
+ .unwrap()
+ .get(identifier)
+ .cloned()
+ .ok_or_else(|| paimon::Error::ViewNotExist {
+ full_name: identifier.full_name(),
+ })
+ }
}
async fn make_sql_context(catalog: Arc<MockCatalog>) -> SQLContext {
@@ -2787,6 +2890,742 @@ mod tests {
ctx
}
+ fn add_unary_sql_function(
+ catalog: &MockCatalog,
+ name: &str,
+ definition: &str,
+ deterministic: bool,
+ ) {
+ add_unary_sql_function_in_database(catalog, "default", name,
definition, deterministic);
+ }
+
+ fn add_unary_sql_function_in_database(
+ catalog: &MockCatalog,
+ database: &str,
+ name: &str,
+ definition: &str,
+ deterministic: bool,
+ ) {
+ let input_params: Vec<PaimonDataField> =
serde_json::from_value(serde_json::json!([
+ {"id": 0, "name": "x", "type": "BIGINT"}
+ ]))
+ .unwrap();
+ let return_params: Vec<PaimonDataField> =
serde_json::from_value(serde_json::json!([
+ {"id": 0, "name": "result", "type": "BIGINT"}
+ ]))
+ .unwrap();
+ catalog.add_function(paimon::catalog::Function::new(
+ Identifier::new(database, name),
+ Some(input_params),
+ Some(return_params),
+ deterministic,
+ HashMap::from([(
+ "datafusion".to_string(),
+ paimon::catalog::FunctionDefinition::Sql {
+ definition: definition.to_string(),
+ },
+ )]),
+ None,
+ HashMap::new(),
+ ));
+ }
+
+ fn add_plus_one_function(catalog: &MockCatalog) {
+ add_unary_sql_function(catalog, "plus_one", "x + 1", true);
+ }
+
+ fn add_constant_view(catalog: &MockCatalog) {
+ let schema = serde_json::from_value(serde_json::json!({
+ "fields": [
+ {"id": 0, "name": "answer", "type": "BIGINT"}
+ ],
+ "query": "SELECT CAST(0 AS BIGINT) AS answer",
+ "dialects": {
+ "datafusion": "SELECT CAST(42 AS INT) AS source_answer"
+ },
+ "comment": null,
+ "options": {}
+ }))
+ .unwrap();
+ catalog.add_view(paimon::catalog::View::new(
+ Identifier::new("default", "answer_view"),
+ schema,
+ ));
+ }
+
+ fn add_bigint_view(catalog: &MockCatalog, database: &str, name: &str,
query: &str) {
+ let schema = serde_json::from_value(serde_json::json!({
+ "fields": [
+ {"id": 0, "name": "answer", "type": "BIGINT"}
+ ],
+ "query": query,
+ "dialects": {},
+ "comment": null,
+ "options": {}
+ }))
+ .unwrap();
+ catalog.add_view(paimon::catalog::View::new(
+ Identifier::new(database, name),
+ schema,
+ ));
+ }
+
+ #[tokio::test]
+ async fn rest_catalog_view_is_planned_lazily() {
+ let catalog = Arc::new(MockCatalog::new());
+ add_constant_view(&catalog);
+ let ctx = make_sql_context(catalog).await;
+
+ let batches = ctx
+ .sql("SELECT * FROM answer_view")
+ .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 rest_catalog_view_can_call_bare_sql_function() {
+ let catalog = Arc::new(MockCatalog::new());
+ add_plus_one_function(&catalog);
+ let schema = serde_json::from_value(serde_json::json!({
+ "fields": [
+ {"id": 0, "name": "answer", "type": "BIGINT"}
+ ],
+ "query": "SELECT plus_one(41) AS answer",
+ "dialects": {},
+ "comment": null,
+ "options": {}
+ }))
+ .unwrap();
+ catalog.add_view(paimon::catalog::View::new(
+ Identifier::new("default", "function_view"),
+ schema,
+ ));
+ let ctx = make_sql_context(catalog).await;
+
+ let batches = ctx
+ .sql("SELECT * FROM function_view")
+ .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 rest_catalog_view_is_discoverable_from_schema_provider() {
+ let catalog = Arc::new(MockCatalog::new());
+ add_constant_view(&catalog);
+ let ctx = make_sql_context(catalog).await;
+
+ let schema = ctx
+ .ctx
+ .catalog("paimon")
+ .unwrap()
+ .schema("default")
+ .unwrap();
+
+ assert!(schema.table_names().contains(&"answer_view".to_string()));
+ assert!(schema.table_exist("answer_view"));
+ }
+
+ #[tokio::test]
+ async fn nested_rest_catalog_view_binds_bare_names_to_owning_database() {
+ let catalog = Arc::new(MockCatalog::new());
+ add_bigint_view(
+ &catalog,
+ "default",
+ "base_view",
+ "SELECT CAST(42 AS BIGINT) AS answer",
+ );
+ add_bigint_view(
+ &catalog,
+ "other",
+ "base_view",
+ "SELECT CAST(7 AS BIGINT) AS answer",
+ );
+ add_bigint_view(&catalog, "default", "outer_view", "SELECT * FROM
base_view");
+ let ctx = make_sql_context(catalog).await;
+ ctx.set_current_database("other").await.unwrap();
+
+ let batches = ctx
+ .sql("SELECT * FROM paimon.default.outer_view")
+ .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 recursive_rest_catalog_views_are_rejected() {
+ let catalog = Arc::new(MockCatalog::new());
+ add_bigint_view(
+ &catalog,
+ "default",
+ "first_view",
+ "SELECT * FROM second_view",
+ );
+ add_bigint_view(
+ &catalog,
+ "default",
+ "second_view",
+ "SELECT * FROM first_view",
+ );
+ let ctx = make_sql_context(catalog).await;
+
+ let error = tokio::time::timeout(
+ std::time::Duration::from_secs(2),
+ ctx.sql("SELECT * FROM first_view"),
+ )
+ .await
+ .expect("recursive view planning should terminate")
+ .unwrap_err()
+ .to_string();
+
+ assert!(
+ error.contains("recursive REST catalog view"),
+ "unexpected error: {error}"
+ );
+ }
+
+ #[tokio::test]
+ async fn rest_catalog_view_allows_cte_with_same_name() {
+ let catalog = Arc::new(MockCatalog::new());
+ add_bigint_view(
+ &catalog,
+ "default",
+ "cte_view",
+ "WITH wrapper AS (\
+ WITH cte_view AS (SELECT CAST(42 AS BIGINT) AS answer) \
+ SELECT * FROM cte_view\
+ ) SELECT * FROM wrapper",
+ );
+ let ctx = make_sql_context(catalog).await;
+
+ let batches = ctx
+ .sql("SELECT * FROM cte_view")
+ .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 rest_catalog_view_normalizes_cte_identifiers() {
+ let catalog = Arc::new(MockCatalog::new());
+ add_bigint_view(
+ &catalog,
+ "default",
+ "cte_view",
+ "WITH cte_view AS (SELECT CAST(42 AS BIGINT) AS answer) \
+ SELECT * FROM \"cte_view\"",
+ );
+ let ctx = make_sql_context(catalog).await;
+
+ let batches = ctx
+ .sql("SELECT * FROM cte_view")
+ .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 rest_catalog_view_rejects_non_query_sql() {
+ let catalog = Arc::new(MockCatalog::new());
+ add_bigint_view(
+ &catalog,
+ "default",
+ "unsafe_view",
+ "DELETE FROM missing_table",
+ );
+ let ctx = make_sql_context(catalog).await;
+
+ let error = ctx
+ .sql("SELECT * FROM unsafe_view")
+ .await
+ .unwrap_err()
+ .to_string();
+
+ assert!(
+ error.contains("read-only query"),
+ "unexpected error: {error}"
+ );
+ }
+
+ #[tokio::test]
+ async fn bare_rest_sql_function_is_expanded() {
+ let catalog = Arc::new(MockCatalog::new());
+ add_plus_one_function(&catalog);
+ let ctx = make_sql_context(catalog).await;
+
+ 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 rest_sql_function_normalizes_call_identifiers() {
+ let catalog = Arc::new(MockCatalog::new());
+ add_plus_one_function(&catalog);
+ let ctx = make_sql_context(catalog).await;
+
+ let batches = ctx
+ .sql("SELECT PAIMON.DEFAULT.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 rest_sql_function_accepts_outer_column_argument() {
+ let catalog = Arc::new(MockCatalog::new());
+ add_plus_one_function(&catalog);
+ let ctx = make_sql_context(catalog).await;
+
+ let batches = ctx
+ .sql("SELECT plus_one(x) AS answer FROM (VALUES (41)) AS t(x)")
+ .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 fully_qualified_rest_sql_function_is_expanded() {
+ let catalog = Arc::new(MockCatalog::new());
+ add_plus_one_function(&catalog);
+ let ctx = make_sql_context(catalog).await;
+
+ let batches = ctx
+ .sql("SELECT paimon.default.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 rest_sql_function_result_is_cast_to_declared_type() {
+ let catalog = Arc::new(MockCatalog::new());
+ add_unary_sql_function(&catalog, "narrow_body", "CAST(x AS INT)",
true);
+ let ctx = make_sql_context(catalog).await;
+
+ let batches = ctx
+ .sql("SELECT narrow_body(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), 41);
+ }
+
+ #[tokio::test]
+ async fn rest_sql_function_normalizes_definition_parameters() {
+ let catalog = Arc::new(MockCatalog::new());
+ add_unary_sql_function(&catalog, "uppercase_parameter", "X + 1", true);
+ let ctx = make_sql_context(catalog).await;
+
+ let batches = ctx
+ .sql("SELECT uppercase_parameter(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 rest_sql_function_preserves_quoted_metadata_parameter() {
+ let catalog = Arc::new(MockCatalog::new());
+ let input_params: Vec<PaimonDataField> =
serde_json::from_value(serde_json::json!([
+ {"id": 0, "name": "X", "type": "BIGINT"}
+ ]))
+ .unwrap();
+ let return_params: Vec<PaimonDataField> =
serde_json::from_value(serde_json::json!([
+ {"id": 0, "name": "result", "type": "BIGINT"}
+ ]))
+ .unwrap();
+ catalog.add_function(paimon::catalog::Function::new(
+ Identifier::new("default", "quoted_parameter"),
+ Some(input_params),
+ Some(return_params),
+ true,
+ HashMap::from([(
+ "datafusion".to_string(),
+ paimon::catalog::FunctionDefinition::Sql {
+ definition: "\"X\" + 1".to_string(),
+ },
+ )]),
+ None,
+ HashMap::new(),
+ ));
+ let ctx = make_sql_context(catalog).await;
+
+ let batches = ctx
+ .sql("SELECT quoted_parameter(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 rest_sql_function_is_expanded_in_explain() {
+ let catalog = Arc::new(MockCatalog::new());
+ add_plus_one_function(&catalog);
+ let ctx = make_sql_context(catalog).await;
+
+ ctx.sql("EXPLAIN SELECT plus_one(1)")
+ .await
+ .unwrap()
+ .collect()
+ .await
+ .unwrap();
+ }
+
+ #[tokio::test]
+ async fn rest_sql_function_is_expanded_in_time_travel_query() {
+ let temp_dir = tempfile::tempdir().unwrap();
+ let mut options = paimon::Options::new();
+ options.set(
+ paimon::CatalogOptions::WAREHOUSE,
+ temp_dir.path().to_string_lossy(),
+ );
+ let storage_catalog =
Arc::new(paimon::FileSystemCatalog::new(options).unwrap());
+ let mut setup = SQLContext::new();
+ setup
+ .register_catalog("paimon", storage_catalog.clone())
+ .await
+ .unwrap();
+ setup
+ .sql("CREATE TABLE paimon.default.time_travel_source (id INT)")
+ .await
+ .unwrap();
+ setup
+ .sql("INSERT INTO paimon.default.time_travel_source VALUES (41)")
+ .await
+ .unwrap()
+ .collect()
+ .await
+ .unwrap();
+
+ let catalog = Arc::new(MockCatalog::new());
+ *catalog.existing_table.lock().unwrap() = Some(
+ storage_catalog
+ .get_table(&Identifier::new("default", "time_travel_source"))
+ .await
+ .unwrap(),
+ );
+ add_plus_one_function(&catalog);
+ let ctx = make_sql_context(catalog).await;
+
+ let batches = ctx
+ .sql(
+ "SELECT plus_one(id) AS answer \
+ FROM time_travel_source VERSION AS OF 1",
+ )
+ .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 rest_sql_function_rejects_undeclared_identifiers() {
+ let catalog = Arc::new(MockCatalog::new());
+ add_unary_sql_function(&catalog, "captures_column", "x + y", true);
+ let ctx = make_sql_context(catalog).await;
+
+ let error = ctx
+ .sql("SELECT captures_column(41)")
+ .await
+ .unwrap_err()
+ .to_string();
+
+ assert!(
+ error.contains("undeclared identifier 'y'"),
+ "unexpected error: {error}"
+ );
+ }
+
+ #[tokio::test]
+ async fn non_deterministic_rest_sql_function_is_rejected() {
+ let catalog = Arc::new(MockCatalog::new());
+ add_unary_sql_function(&catalog, "unsafe_plus_one", "x + 1", false);
+ let ctx = make_sql_context(catalog).await;
+
+ let error = ctx
+ .sql("SELECT unsafe_plus_one(41)")
+ .await
+ .unwrap_err()
+ .to_string();
+
+ assert!(
+ error.contains("non-deterministic"),
+ "unexpected error: {error}"
+ );
+ }
+
+ #[tokio::test]
+ async fn rest_sql_function_without_single_return_is_rejected() {
+ let catalog = Arc::new(MockCatalog::new());
+ let input_params: Vec<PaimonDataField> =
serde_json::from_value(serde_json::json!([
+ {"id": 0, "name": "x", "type": "BIGINT"}
+ ]))
+ .unwrap();
+ catalog.add_function(paimon::catalog::Function::new(
+ Identifier::new("default", "missing_return"),
+ Some(input_params),
+ None,
+ true,
+ HashMap::from([(
+ "datafusion".to_string(),
+ paimon::catalog::FunctionDefinition::Sql {
+ definition: "x + 1".to_string(),
+ },
+ )]),
+ None,
+ HashMap::new(),
+ ));
+ let ctx = make_sql_context(catalog).await;
+
+ let error = ctx
+ .sql("SELECT missing_return(41)")
+ .await
+ .unwrap_err()
+ .to_string();
+
+ assert!(
+ error.contains("exactly one return parameter"),
+ "unexpected error: {error}"
+ );
+ }
+
+ #[tokio::test]
+ async fn nested_rest_sql_functions_are_expanded() {
+ let catalog = Arc::new(MockCatalog::new());
+ add_plus_one_function(&catalog);
+ add_unary_sql_function(&catalog, "plus_two", "plus_one(x) + 1", true);
+ let ctx = make_sql_context(catalog).await;
+
+ let batches = ctx
+ .sql("SELECT plus_two(40) 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 nested_rest_sql_function_binds_to_owning_database() {
+ let catalog = Arc::new(MockCatalog::new());
+ add_unary_sql_function(&catalog, "plus_one", "x + 100", true);
+ add_unary_sql_function_in_database(&catalog, "other", "plus_one", "x +
1", true);
+ add_unary_sql_function_in_database(&catalog, "other", "plus_two",
"plus_one(x) + 1", true);
+ let ctx = make_sql_context(catalog).await;
+
+ let batches = ctx
+ .sql("SELECT paimon.other.plus_two(40) 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 nested_rest_sql_function_normalizes_owning_database_reference() {
+ let catalog = Arc::new(MockCatalog::new());
+ add_unary_sql_function(&catalog, "plus_one", "x + 100", true);
+ add_unary_sql_function_in_database(&catalog, "other", "plus_one", "x +
1", true);
+ add_unary_sql_function_in_database(&catalog, "other", "plus_two",
"PLUS_ONE(x) + 1", true);
+ let ctx = make_sql_context(catalog).await;
+
+ let batches = ctx
+ .sql("SELECT paimon.other.plus_two(40) 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 recursive_rest_sql_functions_are_rejected() {
+ let catalog = Arc::new(MockCatalog::new());
+ add_unary_sql_function(&catalog, "first", "second(x)", true);
+ add_unary_sql_function(&catalog, "second", "first(x)", true);
+ let ctx = make_sql_context(catalog).await;
+
+ let error = ctx.sql("SELECT first(1)").await.unwrap_err().to_string();
+
+ assert!(error.contains("recursive"), "unexpected error: {error}");
+ }
+
+ #[tokio::test]
+ async fn branching_rest_sql_function_expansion_is_bounded() {
+ let catalog = Arc::new(MockCatalog::new());
+ for index in 0..3 {
+ let next = index + 1;
+ add_unary_sql_function(
+ &catalog,
+ &format!("f{index}"),
+ &format!("f{next}(x) + f{next}(x)"),
+ true,
+ );
+ }
+ add_unary_sql_function(&catalog, "f3", "x", true);
+ let ctx = make_sql_context(catalog).await;
+ let statement = Parser::parse_sql(&GenericDialect {}, "SELECT f0(1)")
+ .unwrap()
+ .remove(0);
+
+ let error = crate::sql_function::expand_statement_with_budget(
+ statement,
+ &ctx.catalogs,
+ &ctx.current_catalog_name(),
+ "default",
+ 4,
+ )
+ .await
+ .unwrap_err()
+ .to_string();
+
+ assert!(
+ error.contains("expansion budget"),
+ "unexpected error: {error}"
+ );
+ }
+
// ==================== register_catalog_with_default_db tests
====================
/// Counts get/create_database calls so tests can assert whether the
default-db
diff --git a/crates/integrations/datafusion/src/sql_function.rs
b/crates/integrations/datafusion/src/sql_function.rs
new file mode 100644
index 0000000..88d52be
--- /dev/null
+++ b/crates/integrations/datafusion/src/sql_function.rs
@@ -0,0 +1,459 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you 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 std::collections::{BTreeMap, HashMap, HashSet, VecDeque};
+use std::ops::ControlFlow;
+use std::sync::Arc;
+
+use datafusion::error::{DataFusionError, Result as DFResult};
+use datafusion::sql::planner::IdentNormalizer;
+use datafusion::sql::sqlparser::ast::{
+ visit_expressions, visit_expressions_mut, Expr as SqlExpr, FunctionArg,
FunctionArgExpr,
+ FunctionArguments, Ident, ObjectName, Statement,
+};
+use datafusion::sql::sqlparser::dialect::GenericDialect;
+use datafusion::sql::sqlparser::parser::Parser;
+use paimon::catalog::{Catalog, Function, Identifier};
+
+const MAX_EXPANSION_DEPTH: usize = 32;
+const MAX_FUNCTION_REFERENCES: usize = 1024;
+const MAX_EXPANDED_CALLS: usize = 1024;
+type FunctionReference = (String, Identifier);
+
+pub(crate) async fn expand_sql(
+ sql: &str,
+ catalogs: &HashMap<String, Arc<dyn Catalog>>,
+ current_catalog: &str,
+ current_database: &str,
+) -> DFResult<String> {
+ let mut statements = Parser::parse_sql(&GenericDialect {}, sql)
+ .map_err(|error| DataFusionError::Plan(format!("Invalid REST SQL:
{error}")))?;
+ if statements.len() != 1 {
+ return Err(DataFusionError::Plan(format!(
+ "REST SQL must contain exactly one statement, found {}",
+ statements.len()
+ )));
+ }
+ let statement = expand_statement(
+ statements.remove(0),
+ catalogs,
+ current_catalog,
+ current_database,
+ )
+ .await?;
+ Ok(statement.to_string())
+}
+
+pub(crate) async fn expand_statement(
+ statement: Statement,
+ catalogs: &HashMap<String, Arc<dyn Catalog>>,
+ current_catalog: &str,
+ current_database: &str,
+) -> DFResult<Statement> {
+ expand_statement_with_budget(
+ statement,
+ catalogs,
+ current_catalog,
+ current_database,
+ MAX_EXPANDED_CALLS,
+ )
+ .await
+}
+
+pub(crate) async fn expand_statement_with_budget(
+ mut statement: Statement,
+ catalogs: &HashMap<String, Arc<dyn Catalog>>,
+ current_catalog: &str,
+ current_database: &str,
+ max_expanded_calls: usize,
+) -> DFResult<Statement> {
+ let mut functions = HashMap::new();
+ let mut dependencies = HashMap::new();
+ let mut total_expanded_calls = 0;
+ for _ in 0..MAX_EXPANSION_DEPTH {
+ let mut references = BTreeMap::new();
+ let _: ControlFlow<()> = visit_expressions(&statement, |expr| {
+ if let SqlExpr::Function(function) = expr {
+ if let Some(reference) =
+ function_reference(function, current_catalog,
current_database)
+ {
+ references.insert(reference, ());
+ }
+ }
+ ControlFlow::Continue(())
+ });
+
+ let mut pending = references.into_keys().collect::<VecDeque<_>>();
+ while let Some(reference) = pending.pop_front() {
+ if functions.contains_key(&reference) {
+ continue;
+ }
+ if functions.len() >= MAX_FUNCTION_REFERENCES {
+ return Err(DataFusionError::Plan(format!(
+ "REST SQL function expansion exceeded the reference limit
of {MAX_FUNCTION_REFERENCES}"
+ )));
+ }
+ let Some(catalog) = catalogs.get(&reference.0) else {
+ 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)),
+ };
+ if let Some(function) = &function {
+ let nested = function_dependencies(function, &reference.0,
reference.1.database());
+ pending.extend(nested.iter().cloned());
+ dependencies.insert(reference.clone(), nested);
+ }
+ functions.insert(reference, function);
+ }
+
+ if let Some(cycle) = find_dependency_cycle(&dependencies) {
+ let path = cycle
+ .iter()
+ .map(|(catalog, identifier)| format!("{catalog}.{}",
identifier.full_name()))
+ .collect::<Vec<_>>()
+ .join(" -> ");
+ return Err(DataFusionError::Plan(format!(
+ "recursive REST SQL function dependency detected: {path}"
+ )));
+ }
+
+ let mut expanded_count = 0;
+ let flow = visit_expressions_mut(&mut statement, |expr| {
+ let SqlExpr::Function(call) = expr else {
+ return ControlFlow::Continue(());
+ };
+ let Some(reference) = function_reference(call, current_catalog,
current_database)
+ else {
+ return ControlFlow::Continue(());
+ };
+ let Some(Some(function)) = functions.get(&reference) else {
+ return ControlFlow::Continue(());
+ };
+ if total_expanded_calls >= max_expanded_calls {
+ return ControlFlow::Break(DataFusionError::Plan(format!(
+ "REST SQL function expansion budget of
{max_expanded_calls} calls exceeded"
+ )));
+ }
+ total_expanded_calls += 1;
+
+ match expand_call(function, call, &reference.0, &functions) {
+ Ok(expanded) => {
+ *expr = expanded;
+ expanded_count += 1;
+ ControlFlow::Continue(())
+ }
+ Err(error) => ControlFlow::Break(error),
+ }
+ });
+ if let ControlFlow::Break(error) = flow {
+ return Err(error);
+ }
+ if expanded_count == 0 {
+ return Ok(statement);
+ }
+ }
+
+ Err(DataFusionError::Plan(format!(
+ "REST SQL function expansion exceeded maximum depth of
{MAX_EXPANSION_DEPTH}"
+ )))
+}
+
+fn function_dependencies(
+ function: &Function,
+ current_catalog: &str,
+ current_database: &str,
+) -> Vec<FunctionReference> {
+ let Some(definition) = function
+ .definition("datafusion")
+ .and_then(paimon::catalog::FunctionDefinition::sql)
+ else {
+ return Vec::new();
+ };
+ let Ok(mut parser) = Parser::new(&GenericDialect
{}).try_with_sql(definition) else {
+ return Vec::new();
+ };
+ let Ok(body) = parser.parse_expr() else {
+ return Vec::new();
+ };
+ let mut dependencies = Vec::new();
+ let _: ControlFlow<()> = visit_expressions(&body, |expr| {
+ if let SqlExpr::Function(call) = expr {
+ if let Some(reference) = function_reference(call, current_catalog,
current_database) {
+ dependencies.push(reference);
+ }
+ }
+ ControlFlow::Continue(())
+ });
+ dependencies
+}
+
+fn find_dependency_cycle(
+ dependencies: &HashMap<FunctionReference, Vec<FunctionReference>>,
+) -> Option<Vec<FunctionReference>> {
+ fn visit(
+ reference: &FunctionReference,
+ dependencies: &HashMap<FunctionReference, Vec<FunctionReference>>,
+ finished: &mut HashSet<FunctionReference>,
+ path: &mut Vec<FunctionReference>,
+ ) -> Option<Vec<FunctionReference>> {
+ if let Some(start) = path.iter().position(|entry| entry == reference) {
+ let mut cycle = path[start..].to_vec();
+ cycle.push(reference.clone());
+ return Some(cycle);
+ }
+ if finished.contains(reference) {
+ return None;
+ }
+
+ path.push(reference.clone());
+ if let Some(next_references) = dependencies.get(reference) {
+ for next in next_references {
+ if let Some(cycle) = visit(next, dependencies, finished, path)
{
+ return Some(cycle);
+ }
+ }
+ }
+ path.pop();
+ finished.insert(reference.clone());
+ None
+ }
+
+ let mut finished = HashSet::new();
+ for reference in dependencies.keys() {
+ if let Some(cycle) = visit(reference, dependencies, &mut finished,
&mut Vec::new()) {
+ return Some(cycle);
+ }
+ }
+ None
+}
+
+fn function_reference(
+ function: &datafusion::sql::sqlparser::ast::Function,
+ current_catalog: &str,
+ current_database: &str,
+) -> Option<(String, Identifier)> {
+ let identifiers = function
+ .name
+ .0
+ .iter()
+ .map(|part| part.as_ident().map(normalize_identifier))
+ .collect::<Option<Vec<_>>>()?;
+ match identifiers.as_slice() {
+ [function] => Some((
+ current_catalog.to_string(),
+ Identifier::new(current_database, function),
+ )),
+ [catalog, database, function] => {
+ Some((catalog.clone(), Identifier::new(database, function)))
+ }
+ _ => None,
+ }
+}
+
+fn normalize_identifier(identifier: &Ident) -> String {
+ IdentNormalizer::default().normalize(identifier.clone())
+}
+
+fn expand_call(
+ function: &Function,
+ call: &datafusion::sql::sqlparser::ast::Function,
+ owner_catalog: &str,
+ functions: &HashMap<FunctionReference, Option<Function>>,
+) -> DFResult<SqlExpr> {
+ if !function.is_deterministic() {
+ return Err(DataFusionError::Plan(format!(
+ "REST SQL function '{}' is non-deterministic and is not supported",
+ function.full_name()
+ )));
+ }
+ if function
+ .return_params()
+ .map(<[paimon::spec::DataField]>::len)
+ != Some(1)
+ {
+ return Err(DataFusionError::Plan(format!(
+ "REST SQL function '{}' must declare exactly one return parameter",
+ function.full_name()
+ )));
+ }
+ let input_params = function.input_params().ok_or_else(|| {
+ DataFusionError::Plan(format!(
+ "REST SQL function '{}' has no input parameters",
+ function.full_name()
+ ))
+ })?;
+ let definition = function
+ .definition("datafusion")
+ .and_then(paimon::catalog::FunctionDefinition::sql)
+ .ok_or_else(|| {
+ DataFusionError::Plan(format!(
+ "REST SQL function '{}' has no datafusion SQL definition",
+ function.full_name()
+ ))
+ })?;
+
+ let FunctionArguments::List(args) = &call.args else {
+ return Err(DataFusionError::Plan(format!(
+ "REST SQL function '{}' requires positional arguments",
+ function.full_name()
+ )));
+ };
+ let values = args
+ .args
+ .iter()
+ .map(|arg| match arg {
+ FunctionArg::Unnamed(FunctionArgExpr::Expr(expr)) =>
Ok(expr.clone()),
+ _ => Err(DataFusionError::Plan(format!(
+ "REST SQL function '{}' requires positional expression
arguments",
+ function.full_name()
+ ))),
+ })
+ .collect::<DFResult<Vec<_>>>()?;
+ if values.len() != input_params.len() {
+ return Err(DataFusionError::Plan(format!(
+ "REST SQL function '{}' expects {} arguments but received {}",
+ function.full_name(),
+ input_params.len(),
+ values.len()
+ )));
+ }
+
+ let mut body = Parser::new(&GenericDialect {})
+ .try_with_sql(definition)
+ .map_err(|error| {
+ DataFusionError::Plan(format!("Invalid SQL function definition:
{error}"))
+ })?
+ .parse_expr()
+ .map_err(|error| {
+ DataFusionError::Plan(format!("Invalid SQL function definition:
{error}"))
+ })?;
+ let replacements: HashMap<String, SqlExpr> = input_params
+ .iter()
+ .zip(values)
+ .map(|(field, value)| (field.name().to_string(), value))
+ .collect();
+ let validation = visit_expressions(&body, |expr| match expr {
+ SqlExpr::Identifier(identifier)
+ if !replacements.contains_key(&normalize_identifier(identifier)) =>
+ {
+ ControlFlow::Break(identifier.value.clone())
+ }
+ SqlExpr::CompoundIdentifier(identifiers) => ControlFlow::Break(
+ identifiers
+ .iter()
+ .map(|identifier| identifier.value.as_str())
+ .collect::<Vec<_>>()
+ .join("."),
+ ),
+ _ => ControlFlow::Continue(()),
+ });
+ if let ControlFlow::Break(identifier) = validation {
+ return Err(DataFusionError::Plan(format!(
+ "REST SQL function '{}' references undeclared identifier
'{identifier}'",
+ function.full_name()
+ )));
+ }
+ let _: ControlFlow<()> = visit_expressions_mut(&mut body, |expr| {
+ let SqlExpr::Function(call) = expr else {
+ return ControlFlow::Continue(());
+ };
+ let Some(function_name) = bare_function_name(call) else {
+ return ControlFlow::Continue(());
+ };
+ let function_name = normalize_identifier(&function_name);
+ let reference = (
+ owner_catalog.to_string(),
+ Identifier::new(function.identifier().database(), &function_name),
+ );
+ if matches!(functions.get(&reference), Some(Some(_))) {
+ call.name = ObjectName::from(vec![
+ Ident::with_quote('"', owner_catalog),
+ Ident::with_quote('"', function.identifier().database()),
+ Ident::with_quote('"', function_name),
+ ]);
+ }
+ ControlFlow::Continue(())
+ });
+ let _: ControlFlow<()> = visit_expressions_mut(&mut body, |expr| {
+ if let SqlExpr::Identifier(identifier) = expr {
+ if let Some(replacement) =
replacements.get(&normalize_identifier(identifier)) {
+ *expr = replacement.clone();
+ }
+ }
+ ControlFlow::Continue(())
+ });
+
+ let return_type = function.return_params().expect("validated
above")[0].data_type();
+ let serialized_type = serde_json::to_value(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| {
+ DataFusionError::Plan(format!(
+ "Invalid return type for REST SQL function '{}': {error}",
+ function.full_name()
+ ))
+ })?
+ .parse_expr()
+ .map_err(|error| {
+ DataFusionError::Plan(format!(
+ "Invalid return type for REST SQL function '{}': {error}",
+ function.full_name()
+ ))
+ })
+}
+
+fn bare_function_name(function: &datafusion::sql::sqlparser::ast::Function) ->
Option<Ident> {
+ match function.name.0.as_slice() {
+ [part] => part.as_ident().cloned(),
+ _ => None,
+ }
+}
+
+#[cfg(test)]
+mod tests {
+ use std::collections::HashMap;
+
+ use super::expand_sql;
+
+ #[tokio::test]
+ async fn leaves_bare_functions_for_datafusion_without_a_paimon_catalog() {
+ let sql = "SELECT vector_from_json('[1, 2.5]')";
+
+ let expanded = expand_sql(sql, &HashMap::new(), "datafusion", "public")
+ .await
+ .unwrap();
+
+ assert_eq!(expanded, sql);
+ }
+}
diff --git a/crates/integrations/datafusion/tests/read_tables.rs
b/crates/integrations/datafusion/tests/read_tables.rs
index 55db55b..c0a5e6d 100644
--- a/crates/integrations/datafusion/tests/read_tables.rs
+++ b/crates/integrations/datafusion/tests/read_tables.rs
@@ -745,7 +745,13 @@ async fn test_query_via_catalog_provider() {
#[tokio::test]
async fn test_missing_database_returns_no_schema() {
let catalog = create_catalog();
- let provider = PaimonCatalogProvider::new(Arc::new(catalog));
+ let provider = PaimonCatalogProvider::new(
+ None,
+ Arc::new(catalog),
+ Default::default(),
+ Default::default(),
+ None,
+ );
assert!(
provider.schema("definitely_missing_database").is_none(),
diff --git a/crates/integrations/datafusion/tests/sql_context_tests.rs
b/crates/integrations/datafusion/tests/sql_context_tests.rs
index d8934ee..f04f184 100644
--- a/crates/integrations/datafusion/tests/sql_context_tests.rs
+++ b/crates/integrations/datafusion/tests/sql_context_tests.rs
@@ -622,7 +622,13 @@ async fn test_drop_schema() {
#[tokio::test]
async fn test_schema_names_via_catalog_provider() {
let (_tmp, catalog) = create_test_env();
- let provider = PaimonCatalogProvider::new(catalog.clone());
+ let provider = PaimonCatalogProvider::new(
+ None,
+ catalog.clone(),
+ Default::default(),
+ Default::default(),
+ None,
+ );
catalog
.create_database("db_a", false, Default::default())
diff --git a/crates/paimon/src/api/api_response.rs
b/crates/paimon/src/api/api_response.rs
index 092008a..9586c7f 100644
--- a/crates/paimon/src/api/api_response.rs
+++ b/crates/paimon/src/api/api_response.rs
@@ -22,7 +22,8 @@
use serde::{Deserialize, Serialize};
use std::collections::HashMap;
-use crate::spec::Schema;
+use crate::catalog::{Function, FunctionDefinition, ViewSchema};
+use crate::spec::{DataField, Schema};
/// Error response from REST API calls.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
@@ -130,6 +131,81 @@ pub struct GetTableResponse {
pub schema: Option<Schema>,
}
+/// Response for getting a persistent view.
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(rename_all = "camelCase")]
+pub struct GetViewResponse {
+ /// Audit information.
+ #[serde(flatten)]
+ pub audit: AuditRESTResponse,
+ /// The unique identifier of the view.
+ pub id: Option<String>,
+ /// The name of the view.
+ pub name: Option<String>,
+ /// Stored view schema and SQL representations.
+ pub schema: ViewSchema,
+}
+
+/// Response for getting a persistent function.
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(rename_all = "camelCase")]
+pub struct GetFunctionResponse {
+ /// Audit information.
+ #[serde(flatten)]
+ pub audit: AuditRESTResponse,
+ /// The unique identifier of the function.
+ pub uuid: Option<String>,
+ /// The name of the function.
+ pub name: Option<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.
+ #[serde(default)]
+ pub options: HashMap<String, String>,
+}
+
+impl GetFunctionResponse {
+ /// Create a response from a catalog function.
+ pub fn from_function(function: &Function, audit: AuditRESTResponse) ->
Self {
+ Self {
+ audit,
+ uuid: None,
+ name: Some(function.name().to_string()),
+ input_params: function.input_params().map(<[DataField]>::to_vec),
+ return_params: function.return_params().map(<[DataField]>::to_vec),
+ deterministic: function.is_deterministic(),
+ definitions: function.definitions().clone(),
+ comment: function.comment().map(ToString::to_string),
+ options: function.options().clone(),
+ }
+ }
+}
+
+impl GetViewResponse {
+ /// Create a new get-view response.
+ pub fn new(
+ id: Option<String>,
+ name: Option<String>,
+ schema: ViewSchema,
+ audit: AuditRESTResponse,
+ ) -> Self {
+ Self {
+ audit,
+ id,
+ name,
+ schema,
+ }
+ }
+}
+
impl GetTableResponse {
/// Create a new GetTableResponse.
#[allow(clippy::too_many_arguments)]
@@ -248,6 +324,46 @@ pub struct ListTablesResponse {
pub next_page_token: Option<String>,
}
+/// Response for listing persistent views.
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(rename_all = "camelCase")]
+pub struct ListViewsResponse {
+ /// View names.
+ pub views: Option<Vec<String>>,
+ /// Token for the next page.
+ pub next_page_token: Option<String>,
+}
+
+/// Response for listing persistent functions.
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(rename_all = "camelCase")]
+pub struct ListFunctionsResponse {
+ /// Function names.
+ pub functions: Option<Vec<String>>,
+ /// Token for the next page.
+ pub next_page_token: Option<String>,
+}
+
+impl ListFunctionsResponse {
+ /// Create a list-functions response.
+ pub fn new(functions: Vec<String>, next_page_token: Option<String>) ->
Self {
+ Self {
+ functions: Some(functions),
+ next_page_token,
+ }
+ }
+}
+
+impl ListViewsResponse {
+ /// Create a list-views response.
+ pub fn new(views: Vec<String>, next_page_token: Option<String>) -> Self {
+ Self {
+ views: Some(views),
+ next_page_token,
+ }
+ }
+}
+
impl ListTablesResponse {
/// Create a new ListTablesResponse.
pub fn new(tables: Option<Vec<String>>, next_page_token: Option<String>)
-> Self {
diff --git a/crates/paimon/src/api/mod.rs b/crates/paimon/src/api/mod.rs
index 7672b9a..a81067b 100644
--- a/crates/paimon/src/api/mod.rs
+++ b/crates/paimon/src/api/mod.rs
@@ -37,8 +37,9 @@ pub use api_request::{
// Re-export response types
pub use api_response::{
- AuditRESTResponse, ConfigResponse, ErrorResponse, GetDatabaseResponse,
GetTableResponse,
- GetTableTokenResponse, ListDatabasesResponse, ListPartitionsResponse,
ListTablesResponse,
+ AuditRESTResponse, ConfigResponse, ErrorResponse, GetDatabaseResponse,
GetFunctionResponse,
+ GetTableResponse, GetTableTokenResponse, GetViewResponse,
ListDatabasesResponse,
+ ListFunctionsResponse, ListPartitionsResponse, ListTablesResponse,
ListViewsResponse,
PagedList,
};
diff --git a/crates/paimon/src/api/resource_paths.rs
b/crates/paimon/src/api/resource_paths.rs
index 45505da..4983ae7 100644
--- a/crates/paimon/src/api/resource_paths.rs
+++ b/crates/paimon/src/api/resource_paths.rs
@@ -33,6 +33,8 @@ impl ResourcePaths {
const TABLES: &'static str = "tables";
const TABLE_DETAILS: &'static str = "table-details";
const PARTITIONS: &'static str = "partitions";
+ const VIEWS: &'static str = "views";
+ const FUNCTIONS: &'static str = "functions";
/// Create a new ResourcePaths with the given prefix.
pub fn new(prefix: &str) -> Self {
@@ -105,6 +107,46 @@ impl ResourcePaths {
)
}
+ /// Get the views endpoint path for a database.
+ pub fn views(&self, database_name: &str) -> String {
+ format!(
+ "{}/{}/{}/{}",
+ self.base_path,
+ Self::DATABASES,
+ RESTUtil::encode_string(database_name),
+ Self::VIEWS
+ )
+ }
+
+ /// Get a specific view endpoint path.
+ pub fn view(&self, database_name: &str, view_name: &str) -> String {
+ format!(
+ "{}/{}",
+ self.views(database_name),
+ RESTUtil::encode_string(view_name)
+ )
+ }
+
+ /// Get the functions endpoint path for a database.
+ pub fn functions(&self, database_name: &str) -> String {
+ format!(
+ "{}/{}/{}/{}",
+ self.base_path,
+ Self::DATABASES,
+ RESTUtil::encode_string(database_name),
+ Self::FUNCTIONS
+ )
+ }
+
+ /// Get a specific function endpoint path.
+ pub fn function(&self, database_name: &str, function_name: &str) -> String
{
+ format!(
+ "{}/{}",
+ self.functions(database_name),
+ RESTUtil::encode_string(function_name)
+ )
+ }
+
/// Get the table details endpoint path.
pub fn table_details(&self, database_name: &str) -> String {
format!(
@@ -204,4 +246,40 @@ mod tests {
fn test_config_path() {
assert_eq!(ResourcePaths::config(), "/v1/config");
}
+
+ #[test]
+ fn test_views_path_encodes_database() {
+ let paths = ResourcePaths::new("catalog");
+ assert_eq!(
+ paths.views("analytics db"),
+ "/v1/catalog/databases/analytics+db/views"
+ );
+ }
+
+ #[test]
+ fn test_view_path_encodes_object_name() {
+ let paths = ResourcePaths::new("catalog");
+ assert_eq!(
+ paths.view("analytics", "active ids"),
+ "/v1/catalog/databases/analytics/views/active+ids"
+ );
+ }
+
+ #[test]
+ fn test_functions_path_encodes_database() {
+ let paths = ResourcePaths::new("catalog");
+ assert_eq!(
+ paths.functions("analytics db"),
+ "/v1/catalog/databases/analytics+db/functions"
+ );
+ }
+
+ #[test]
+ fn test_function_path_encodes_object_name() {
+ let paths = ResourcePaths::new("catalog");
+ assert_eq!(
+ paths.function("analytics", "rectangle area"),
+ "/v1/catalog/databases/analytics/functions/rectangle+area"
+ );
+ }
}
diff --git a/crates/paimon/src/api/rest_api.rs
b/crates/paimon/src/api/rest_api.rs
index 8721f48..421a23d 100644
--- a/crates/paimon/src/api/rest_api.rs
+++ b/crates/paimon/src/api/rest_api.rs
@@ -33,8 +33,9 @@ use super::api_request::{
RenameTableRequest,
};
use super::api_response::{
- ConfigResponse, GetDatabaseResponse, GetTableResponse,
ListDatabasesResponse,
- ListPartitionsResponse, ListTablesResponse, PagedList,
+ ConfigResponse, GetDatabaseResponse, GetFunctionResponse,
GetTableResponse, GetViewResponse,
+ ListDatabasesResponse, ListFunctionsResponse, ListPartitionsResponse,
ListTablesResponse,
+ ListViewsResponse, PagedList,
};
use super::auth::{AuthProviderFactory, RESTAuthFunction};
use super::resource_paths::ResourcePaths;
@@ -88,6 +89,8 @@ impl RESTApi {
pub const PAGE_TOKEN: &'static str = "pageToken";
pub const DATABASE_NAME_PATTERN: &'static str = "databaseNamePattern";
pub const TABLE_NAME_PATTERN: &'static str = "tableNamePattern";
+ pub const VIEW_NAME_PATTERN: &'static str = "viewNamePattern";
+ pub const FUNCTION_NAME_PATTERN: &'static str = "functionNamePattern";
pub const TABLE_TYPE: &'static str = "tableType";
/// Create a new RESTApi from options.
@@ -392,6 +395,131 @@ impl RESTApi {
Ok(())
}
+ // ==================== View Operations ====================
+
+ /// List persistent views in a database.
+ pub async fn list_views(&self, database: &str) -> Result<Vec<String>> {
+ validate_non_empty(database, "database name")?;
+ let mut results = Vec::new();
+ let mut page_token = None;
+ loop {
+ let page = self
+ .list_views_paged(database, None, page_token.as_deref(), None)
+ .await?;
+ let is_empty = page.elements.is_empty();
+ results.extend(page.elements);
+ page_token = page.next_page_token;
+ if page_token.is_none() || is_empty {
+ break;
+ }
+ }
+ Ok(results)
+ }
+
+ /// List persistent views in a database with pagination.
+ pub async fn list_views_paged(
+ &self,
+ database: &str,
+ max_results: Option<u32>,
+ page_token: Option<&str>,
+ view_name_pattern: Option<&str>,
+ ) -> Result<PagedList<String>> {
+ validate_non_empty(database, "database name")?;
+ let path = self.resource_paths.views(database);
+ let mut params = Vec::new();
+ if let Some(max_results) = max_results {
+ params.push((Self::MAX_RESULTS, max_results.to_string()));
+ }
+ if let Some(page_token) = page_token {
+ params.push((Self::PAGE_TOKEN, page_token.to_string()));
+ }
+ if let Some(view_name_pattern) = view_name_pattern {
+ params.push((Self::VIEW_NAME_PATTERN,
view_name_pattern.to_string()));
+ }
+ let response: ListViewsResponse = if params.is_empty() {
+ self.client.get(&path, None::<&[(&str, &str)]>).await?
+ } else {
+ self.client.get(&path, Some(¶ms)).await?
+ };
+ Ok(PagedList::new(
+ response.views.unwrap_or_default(),
+ response.next_page_token,
+ ))
+ }
+
+ /// Get persistent view information.
+ pub async fn get_view(&self, identifier: &Identifier) ->
Result<GetViewResponse> {
+ let database = identifier.database();
+ let view = identifier.object();
+ validate_non_empty_multi(&[(database, "database name"), (view, "view
name")])?;
+ let path = self.resource_paths.view(database, view);
+ self.client.get(&path, None::<&[(&str, &str)]>).await
+ }
+
+ // ==================== Function Operations ====================
+
+ /// List persistent functions in a database.
+ pub async fn list_functions(&self, database: &str) -> Result<Vec<String>> {
+ validate_non_empty(database, "database name")?;
+ let mut results = Vec::new();
+ let mut page_token = None;
+ loop {
+ let page = self
+ .list_functions_paged(database, None, page_token.as_deref(),
None)
+ .await?;
+ let is_empty = page.elements.is_empty();
+ results.extend(page.elements);
+ page_token = page.next_page_token;
+ if page_token.is_none() || is_empty {
+ break;
+ }
+ }
+ Ok(results)
+ }
+
+ /// List persistent functions in a database with pagination.
+ pub async fn list_functions_paged(
+ &self,
+ database: &str,
+ max_results: Option<u32>,
+ page_token: Option<&str>,
+ function_name_pattern: Option<&str>,
+ ) -> Result<PagedList<String>> {
+ validate_non_empty(database, "database name")?;
+ let path = self.resource_paths.functions(database);
+ let mut params = Vec::new();
+ if let Some(max_results) = max_results {
+ params.push((Self::MAX_RESULTS, max_results.to_string()));
+ }
+ if let Some(page_token) = page_token {
+ params.push((Self::PAGE_TOKEN, page_token.to_string()));
+ }
+ if let Some(function_name_pattern) = function_name_pattern {
+ params.push((
+ Self::FUNCTION_NAME_PATTERN,
+ function_name_pattern.to_string(),
+ ));
+ }
+ let response: ListFunctionsResponse = if params.is_empty() {
+ self.client.get(&path, None::<&[(&str, &str)]>).await?
+ } else {
+ self.client.get(&path, Some(¶ms)).await?
+ };
+ Ok(PagedList::new(
+ response.functions.unwrap_or_default(),
+ response.next_page_token,
+ ))
+ }
+
+ /// Get persistent function information.
+ pub async fn get_function(&self, identifier: &Identifier) ->
Result<GetFunctionResponse> {
+ let database = identifier.database();
+ let function = identifier.object();
+ validate_non_empty_multi(&[(database, "database name"), (function,
"function name")])?;
+ let path = self.resource_paths.function(database, function);
+ self.client.get(&path, None::<&[(&str, &str)]>).await
+ }
+
// ==================== Partition Operations ====================
/// List all partitions of a table, paging internally.
diff --git a/crates/paimon/src/catalog/function.rs
b/crates/paimon/src/catalog/function.rs
new file mode 100644
index 0000000..37003de
--- /dev/null
+++ b/crates/paimon/src/catalog/function.rs
@@ -0,0 +1,163 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you 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 std::collections::HashMap;
+
+use serde::{Deserialize, Serialize};
+
+use crate::spec::DataField;
+
+use super::Identifier;
+
+/// A persistent catalog function.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct Function {
+ identifier: Identifier,
+ input_params: Option<Vec<DataField>>,
+ return_params: Option<Vec<DataField>>,
+ deterministic: bool,
+ definitions: HashMap<String, FunctionDefinition>,
+ comment: Option<String>,
+ options: HashMap<String, String>,
+}
+
+impl Function {
+ /// Create a catalog function from REST metadata.
+ #[allow(clippy::too_many_arguments)]
+ pub fn new(
+ identifier: Identifier,
+ input_params: Option<Vec<DataField>>,
+ return_params: Option<Vec<DataField>>,
+ deterministic: bool,
+ definitions: HashMap<String, FunctionDefinition>,
+ comment: Option<String>,
+ options: HashMap<String, String>,
+ ) -> Self {
+ Self {
+ identifier,
+ input_params,
+ return_params,
+ deterministic,
+ definitions,
+ comment,
+ options,
+ }
+ }
+
+ /// Function identifier.
+ pub fn identifier(&self) -> &Identifier {
+ &self.identifier
+ }
+
+ /// Unqualified function name.
+ pub fn name(&self) -> &str {
+ self.identifier.object()
+ }
+
+ /// Database-qualified function name.
+ pub fn full_name(&self) -> String {
+ self.identifier.full_name()
+ }
+
+ /// Declared input parameters, when present.
+ pub fn input_params(&self) -> Option<&[DataField]> {
+ self.input_params.as_deref()
+ }
+
+ /// Declared return parameters, when present.
+ pub fn return_params(&self) -> Option<&[DataField]> {
+ self.return_params.as_deref()
+ }
+
+ /// Whether the function is deterministic.
+ pub fn is_deterministic(&self) -> bool {
+ self.deterministic
+ }
+
+ /// Definition for an execution engine.
+ pub fn definition(&self, dialect: &str) -> Option<&FunctionDefinition> {
+ self.definitions.get(dialect)
+ }
+
+ /// All engine definitions.
+ pub fn definitions(&self) -> &HashMap<String, FunctionDefinition> {
+ &self.definitions
+ }
+
+ /// Optional function comment.
+ pub fn comment(&self) -> Option<&str> {
+ self.comment.as_deref()
+ }
+
+ /// Function options.
+ pub fn options(&self) -> &HashMap<String, String> {
+ &self.options
+ }
+}
+
+/// Engine-specific implementation of a catalog function.
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(tag = "type", rename_all = "lowercase")]
+pub enum FunctionDefinition {
+ /// A function loaded from external resources.
+ File {
+ #[serde(rename = "fileResources")]
+ file_resources: Vec<FunctionFileResource>,
+ language: String,
+ #[serde(rename = "className")]
+ class_name: String,
+ #[serde(rename = "functionName")]
+ function_name: String,
+ },
+ /// A scalar SQL expression.
+ Sql { definition: String },
+ /// A language-specific lambda expression.
+ Lambda {
+ definition: String,
+ language: String,
+ },
+}
+
+/// External resource referenced by a file function definition.
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(rename_all = "camelCase")]
+pub struct FunctionFileResource {
+ resource_type: String,
+ uri: String,
+}
+
+impl FunctionFileResource {
+ /// Resource type such as `jar`.
+ pub fn resource_type(&self) -> &str {
+ &self.resource_type
+ }
+
+ /// Resource URI.
+ pub fn uri(&self) -> &str {
+ &self.uri
+ }
+}
+
+impl FunctionDefinition {
+ /// Return the SQL expression when this is an SQL definition.
+ pub fn sql(&self) -> Option<&str> {
+ match self {
+ Self::Sql { definition } => Some(definition),
+ _ => None,
+ }
+ }
+}
diff --git a/crates/paimon/src/catalog/mod.rs b/crates/paimon/src/catalog/mod.rs
index 6d5db12..fa653a1 100644
--- a/crates/paimon/src/catalog/mod.rs
+++ b/crates/paimon/src/catalog/mod.rs
@@ -23,8 +23,10 @@
mod database;
mod factory;
mod filesystem;
+mod function;
mod partition_listing;
mod rest;
+mod view;
use std::collections::HashMap;
use std::fmt;
@@ -33,9 +35,11 @@ use crate::{Error, Result};
pub use database::*;
pub use factory::*;
pub use filesystem::*;
+pub use function::*;
pub use partition_listing::list_partitions_from_file_system;
pub use rest::*;
use serde::{Deserialize, Serialize};
+pub use view::*;
/// Splitter for system table names (e.g. `table$snapshots`).
pub const SYSTEM_TABLE_SPLITTER: &str = "$";
@@ -371,6 +375,38 @@ pub trait Catalog: Send + Sync {
ignore_if_not_exists: bool,
) -> Result<()>;
+ // ======================= view methods ===============================
+
+ /// List persistent view names in a database.
+ async fn list_views(&self, _database_name: &str) -> Result<Vec<String>> {
+ Err(Error::Unsupported {
+ message: "Catalog does not support views".to_string(),
+ })
+ }
+
+ /// Get a persistent view by identifier.
+ async fn get_view(&self, _identifier: &Identifier) -> Result<View> {
+ Err(Error::Unsupported {
+ message: "Catalog does not support views".to_string(),
+ })
+ }
+
+ // ======================= function methods ===============================
+
+ /// List persistent function names in a database.
+ async fn list_functions(&self, _database_name: &str) ->
Result<Vec<String>> {
+ Err(Error::Unsupported {
+ message: "Catalog does not support functions".to_string(),
+ })
+ }
+
+ /// Get a persistent function by identifier.
+ async fn get_function(&self, _identifier: &Identifier) -> Result<Function>
{
+ Err(Error::Unsupported {
+ message: "Catalog does not support functions".to_string(),
+ })
+ }
+
/// List partitions for a table.
///
/// Default impl scans the table's manifest entries via
diff --git a/crates/paimon/src/catalog/rest/rest_catalog.rs
b/crates/paimon/src/catalog/rest/rest_catalog.rs
index 4d87008..6901312 100644
--- a/crates/paimon/src/catalog/rest/rest_catalog.rs
+++ b/crates/paimon/src/catalog/rest/rest_catalog.rs
@@ -278,6 +278,49 @@ impl Catalog for RESTCatalog {
})
}
+ async fn list_views(&self, database_name: &str) -> Result<Vec<String>> {
+ self.api
+ .list_views(database_name)
+ .await
+ .map_err(|e| map_unsupported_endpoint(e, "view"))
+ }
+
+ async fn get_view(&self, identifier: &Identifier) ->
Result<crate::catalog::View> {
+ let response = self
+ .api
+ .get_view(identifier)
+ .await
+ .map_err(|e| map_rest_error_for_view(e, identifier))?;
+ Ok(crate::catalog::View::new(
+ identifier.clone(),
+ response.schema,
+ ))
+ }
+
+ async fn list_functions(&self, database_name: &str) -> Result<Vec<String>>
{
+ self.api
+ .list_functions(database_name)
+ .await
+ .map_err(|e| map_unsupported_endpoint(e, "function"))
+ }
+
+ async fn get_function(&self, identifier: &Identifier) ->
Result<crate::catalog::Function> {
+ let response = self
+ .api
+ .get_function(identifier)
+ .await
+ .map_err(|e| map_rest_error_for_function(e, identifier))?;
+ Ok(crate::catalog::Function::new(
+ identifier.clone(),
+ response.input_params,
+ response.return_params,
+ response.deterministic,
+ response.definitions,
+ response.comment,
+ response.options,
+ ))
+ }
+
async fn list_partitions(&self, identifier: &Identifier) ->
Result<Vec<Partition>> {
match self.api.list_partitions(identifier).await {
Ok(parts) => Ok(parts),
@@ -360,6 +403,41 @@ fn map_rest_error_for_table(err: Error, identifier:
&Identifier) -> Error {
}
}
+/// Map a REST API error to a catalog-level view error.
+fn map_rest_error_for_view(err: Error, identifier: &Identifier) -> Error {
+ match err {
+ Error::RestApi {
+ source: RestError::NoSuchResource { .. },
+ } => Error::ViewNotExist {
+ full_name: identifier.full_name(),
+ },
+ other => map_unsupported_endpoint(other, "view"),
+ }
+}
+
+/// Map a REST API error to a catalog-level function error.
+fn map_rest_error_for_function(err: Error, identifier: &Identifier) -> Error {
+ match err {
+ Error::RestApi {
+ source: RestError::NoSuchResource { .. },
+ } => Error::FunctionNotExist {
+ 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 {
+ source: RestError::NotImplemented { message },
+ } => Error::Unsupported {
+ message: format!("REST catalog {object_type} endpoint is not
supported: {message}"),
+ },
+ other => other,
+ }
+}
+
/// Execute a fallible operation and ignore a specific error variant.
///
/// If the operation succeeds, returns `Ok(())`.
diff --git a/crates/paimon/src/catalog/view.rs
b/crates/paimon/src/catalog/view.rs
new file mode 100644
index 0000000..d99f819
--- /dev/null
+++ b/crates/paimon/src/catalog/view.rs
@@ -0,0 +1,106 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you 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 std::collections::HashMap;
+
+use serde::{Deserialize, Serialize};
+
+use crate::spec::DataField;
+
+use super::Identifier;
+
+/// A persistent catalog view.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct View {
+ identifier: Identifier,
+ schema: ViewSchema,
+}
+
+impl View {
+ /// Create a catalog view from its identifier and stored schema.
+ pub fn new(identifier: Identifier, schema: ViewSchema) -> Self {
+ Self { identifier, schema }
+ }
+
+ /// View identifier.
+ pub fn identifier(&self) -> &Identifier {
+ &self.identifier
+ }
+
+ /// Unqualified view name.
+ pub fn name(&self) -> &str {
+ self.identifier.object()
+ }
+
+ /// Database-qualified view name.
+ pub fn full_name(&self) -> String {
+ self.identifier.full_name()
+ }
+
+ /// Stored schema and SQL representations.
+ pub fn schema(&self) -> &ViewSchema {
+ &self.schema
+ }
+
+ /// SQL representation for a dialect, falling back to the default query.
+ pub fn query_for(&self, dialect: &str) -> &str {
+ self.schema.query_for(dialect)
+ }
+}
+
+/// Schema and SQL representations stored for a catalog view.
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(rename_all = "camelCase")]
+pub struct ViewSchema {
+ fields: Vec<DataField>,
+ query: String,
+ #[serde(default)]
+ dialects: HashMap<String, String>,
+ comment: Option<String>,
+ #[serde(default)]
+ options: HashMap<String, String>,
+}
+
+impl ViewSchema {
+ /// Declared output fields of the view.
+ pub fn fields(&self) -> &[DataField] {
+ &self.fields
+ }
+
+ /// Default SQL representation of the view.
+ pub fn query(&self) -> &str {
+ &self.query
+ }
+
+ /// SQL representation for a dialect, falling back to the default query.
+ pub fn query_for(&self, dialect: &str) -> &str {
+ self.dialects
+ .get(dialect)
+ .map(String::as_str)
+ .unwrap_or(&self.query)
+ }
+
+ /// Optional view comment.
+ pub fn comment(&self) -> Option<&str> {
+ self.comment.as_deref()
+ }
+
+ /// View options.
+ pub fn options(&self) -> &HashMap<String, String> {
+ &self.options
+ }
+}
diff --git a/crates/paimon/src/error.rs b/crates/paimon/src/error.rs
index 56a4d3c..ddac874 100644
--- a/crates/paimon/src/error.rs
+++ b/crates/paimon/src/error.rs
@@ -102,6 +102,10 @@ pub enum Error {
TableAlreadyExist { full_name: String },
#[snafu(display("Table {} does not exist.", full_name))]
TableNotExist { full_name: String },
+ #[snafu(display("View {} does not exist.", full_name))]
+ ViewNotExist { full_name: String },
+ #[snafu(display("Function {} does not exist.", full_name))]
+ FunctionNotExist { 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 6a9f38d..99e11e6 100644
--- a/crates/paimon/tests/mock_server.rs
+++ b/crates/paimon/tests/mock_server.rs
@@ -35,14 +35,20 @@ use tokio::task::JoinHandle;
use paimon::api::{
AlterDatabaseRequest, AlterTableRequest, AuditRESTResponse,
ConfigResponse, ErrorResponse,
- GetDatabaseResponse, GetTableResponse, ListDatabasesResponse,
ListTablesResponse,
+ GetDatabaseResponse, GetFunctionResponse, GetTableResponse,
GetViewResponse,
+ ListDatabasesResponse, ListFunctionsResponse, ListTablesResponse,
ListViewsResponse,
RenameTableRequest, ResourcePaths,
};
+use paimon::catalog::Function;
#[derive(Clone, Debug, Default)]
struct MockState {
databases: HashMap<String, GetDatabaseResponse>,
tables: HashMap<String, GetTableResponse>,
+ views: HashMap<String, GetViewResponse>,
+ functions: HashMap<String, GetFunctionResponse>,
+ view_function_endpoints_unsupported: bool,
+ list_page_size: Option<usize>,
no_permission_databases: HashSet<String>,
no_permission_tables: HashSet<String>,
/// ECS metadata role name (for token loader testing)
@@ -51,6 +57,24 @@ struct MockState {
ecs_token: Option<serde_json::Value>,
}
+fn paginate_names(
+ names: Vec<String>,
+ params: &HashMap<String, String>,
+ page_size: Option<usize>,
+) -> (Vec<String>, Option<String>) {
+ let Some(page_size) = page_size else {
+ return (names, None);
+ };
+ let offset = params
+ .get("pageToken")
+ .and_then(|token| token.parse::<usize>().ok())
+ .unwrap_or(0)
+ .min(names.len());
+ let end = (offset + page_size).min(names.len());
+ let next_page_token = (end < names.len()).then(|| end.to_string());
+ (names[offset..end].to_vec(), next_page_token)
+}
+
#[derive(Clone)]
pub struct RESTServer {
warehouse: String,
@@ -315,6 +339,126 @@ impl RESTServer {
(StatusCode::OK, Json(response)).into_response()
}
+ /// Handle GET /databases/:db/views/:view - get a persistent view.
+ pub async fn get_view(
+ Path((db, view)): Path<(String, String)>,
+ Extension(state): Extension<Arc<RESTServer>>,
+ ) -> impl IntoResponse {
+ let s = state.inner.lock().unwrap();
+ if s.view_function_endpoints_unsupported {
+ let err = ErrorResponse::new(
+ Some("view".to_string()),
+ Some(view),
+ Some("Not Implemented".to_string()),
+ Some(501),
+ );
+ return (StatusCode::NOT_IMPLEMENTED, Json(err)).into_response();
+ }
+ let key = format!("{db}.{view}");
+ if let Some(response) = s.views.get(&key) {
+ (StatusCode::OK, Json(response.clone())).into_response()
+ } else {
+ let err = ErrorResponse::new(
+ Some("view".to_string()),
+ Some(view),
+ Some("Not Found".to_string()),
+ Some(404),
+ );
+ (StatusCode::NOT_FOUND, Json(err)).into_response()
+ }
+ }
+
+ /// Handle GET /databases/:db/views - list persistent views.
+ pub async fn list_views(
+ Path(db): Path<String>,
+ Query(params): Query<HashMap<String, String>>,
+ Extension(state): Extension<Arc<RESTServer>>,
+ ) -> impl IntoResponse {
+ let s = state.inner.lock().unwrap();
+ if s.view_function_endpoints_unsupported {
+ let err = ErrorResponse::new(
+ Some("view".to_string()),
+ None,
+ Some("Not Implemented".to_string()),
+ Some(501),
+ );
+ return (StatusCode::NOT_IMPLEMENTED, Json(err)).into_response();
+ }
+ let prefix = format!("{db}.");
+ let mut views: Vec<String> = s
+ .views
+ .keys()
+ .filter_map(|key|
key.strip_prefix(&prefix).map(ToString::to_string))
+ .collect();
+ views.sort();
+ let (views, next_page_token) = paginate_names(views, ¶ms,
s.list_page_size);
+ (
+ StatusCode::OK,
+ Json(ListViewsResponse::new(views, next_page_token)),
+ )
+ .into_response()
+ }
+
+ /// Handle GET /databases/:db/functions/:function - get a persistent
function.
+ pub async fn get_function(
+ Path((db, function)): Path<(String, String)>,
+ Extension(state): Extension<Arc<RESTServer>>,
+ ) -> impl IntoResponse {
+ let s = state.inner.lock().unwrap();
+ if s.view_function_endpoints_unsupported {
+ let err = ErrorResponse::new(
+ Some("function".to_string()),
+ Some(function),
+ Some("Not Implemented".to_string()),
+ Some(501),
+ );
+ return (StatusCode::NOT_IMPLEMENTED, Json(err)).into_response();
+ }
+ let key = format!("{db}.{function}");
+ if let Some(response) = s.functions.get(&key) {
+ (StatusCode::OK, Json(response.clone())).into_response()
+ } else {
+ let err = ErrorResponse::new(
+ Some("function".to_string()),
+ Some(function),
+ Some("Not Found".to_string()),
+ Some(404),
+ );
+ (StatusCode::NOT_FOUND, Json(err)).into_response()
+ }
+ }
+
+ /// Handle GET /databases/:db/functions - list persistent functions.
+ pub async fn list_functions(
+ Path(db): Path<String>,
+ Query(params): Query<HashMap<String, String>>,
+ Extension(state): Extension<Arc<RESTServer>>,
+ ) -> impl IntoResponse {
+ let s = state.inner.lock().unwrap();
+ if s.view_function_endpoints_unsupported {
+ let err = ErrorResponse::new(
+ Some("function".to_string()),
+ None,
+ Some("Not Implemented".to_string()),
+ Some(501),
+ );
+ return (StatusCode::NOT_IMPLEMENTED, Json(err)).into_response();
+ }
+ let prefix = format!("{db}.");
+ let mut functions: Vec<String> = s
+ .functions
+ .keys()
+ .filter_map(|key|
key.strip_prefix(&prefix).map(ToString::to_string))
+ .collect();
+ functions.sort();
+ let (functions, next_page_token) = paginate_names(functions, ¶ms,
s.list_page_size);
+ (
+ StatusCode::OK,
+ Json(ListFunctionsResponse::new(functions, next_page_token)),
+ )
+ .into_response()
+ }
+
/// Handle POST /databases/:db/tables - create a new table.
pub async fn create_table(
Path(db): Path<String>,
@@ -602,6 +746,44 @@ impl RESTServer {
});
}
+ /// Add a persistent view to the server state.
+ pub fn add_view(&self, database: &str, view: &str, schema:
paimon::catalog::ViewSchema) {
+ let mut s = self.inner.lock().unwrap();
+ let key = format!("{database}.{view}");
+ s.views.insert(
+ key,
+ GetViewResponse::new(
+ Some(view.to_string()),
+ Some(view.to_string()),
+ schema,
+ AuditRESTResponse::new(None, None, None, None, None),
+ ),
+ );
+ }
+
+ /// Add a persistent function to the server state.
+ pub fn add_function(&self, function: Function) {
+ let key = function.full_name();
+ let response = GetFunctionResponse::from_function(
+ &function,
+ AuditRESTResponse::new(None, None, None, None, None),
+ );
+ self.inner.lock().unwrap().functions.insert(key, response);
+ }
+
+ /// Force list-view and list-function handlers to paginate at this size.
+ pub fn set_list_page_size(&self, page_size: usize) {
+ self.inner.lock().unwrap().list_page_size = Some(page_size.max(1));
+ }
+
+ /// Make persistent view and function endpoints return HTTP 501.
+ pub fn set_view_function_endpoints_unsupported(&self) {
+ self.inner
+ .lock()
+ .unwrap()
+ .view_function_endpoints_unsupported = true;
+ }
+
/// Add a table with schema and path to the server state.
///
/// This is needed for `RESTCatalog::get_table` which requires
@@ -761,6 +943,22 @@ pub async fn start_mock_server(
.post(RESTServer::alter_table)
.delete(RESTServer::drop_table),
)
+ .route(
+ &format!("{prefix}/databases/:db/views"),
+ get(RESTServer::list_views),
+ )
+ .route(
+ &format!("{prefix}/databases/:db/views/:view"),
+ get(RESTServer::get_view),
+ )
+ .route(
+ &format!("{prefix}/databases/:db/functions"),
+ get(RESTServer::list_functions),
+ )
+ .route(
+ &format!("{prefix}/databases/:db/functions/:function"),
+ get(RESTServer::get_function),
+ )
.route(
&format!("{prefix}/tables/rename"),
post(RESTServer::rename_table),
diff --git a/crates/paimon/tests/rest_api_test.rs
b/crates/paimon/tests/rest_api_test.rs
index 74784b3..07593cd 100644
--- a/crates/paimon/tests/rest_api_test.rs
+++ b/crates/paimon/tests/rest_api_test.rs
@@ -25,8 +25,9 @@ use std::collections::HashMap;
use paimon::api::auth::{DLFECSTokenLoader, DLFToken, DLFTokenLoader};
use paimon::api::rest_api::RESTApi;
use paimon::api::ConfigResponse;
-use paimon::catalog::Identifier;
+use paimon::catalog::{Function, FunctionDefinition, Identifier, ViewSchema};
use paimon::common::Options;
+use paimon::spec::DataField;
use serde_json::json;
mod mock_server;
@@ -82,6 +83,129 @@ async fn test_list_databases() {
assert!(dbs.contains(&"prod_db".to_string()));
}
+#[tokio::test]
+async fn test_get_view() {
+ let ctx = setup_test_server(vec!["default"]).await;
+ let schema: ViewSchema = serde_json::from_value(json!({
+ "fields": [{"id": 0, "name": "id", "type": "INT"}],
+ "query": "SELECT id FROM source",
+ "dialects": {"datafusion": "SELECT id FROM source WHERE id > 0"},
+ "comment": null,
+ "options": {}
+ }))
+ .unwrap();
+ ctx.server.add_view("default", "active_ids", schema);
+
+ let response = ctx
+ .api
+ .get_view(&Identifier::new("default", "active_ids"))
+ .await
+ .unwrap();
+
+ assert_eq!(response.name.as_deref(), Some("active_ids"));
+ assert_eq!(response.id.as_deref(), Some("active_ids"));
+ assert_eq!(
+ response.schema.query_for("datafusion"),
+ "SELECT id FROM source WHERE id > 0"
+ );
+}
+
+#[tokio::test]
+async fn test_list_views() {
+ let ctx = setup_test_server(vec!["default"]).await;
+ ctx.server.set_list_page_size(1);
+ let schema: ViewSchema = serde_json::from_value(json!({
+ "fields": [{"id": 0, "name": "id", "type": "INT"}],
+ "query": "SELECT 1 AS id",
+ "dialects": {},
+ "comment": null,
+ "options": {}
+ }))
+ .unwrap();
+ ctx.server.add_view("default", "v2", schema.clone());
+ ctx.server.add_view("default", "v1", schema);
+
+ assert_eq!(
+ ctx.api.list_views("default").await.unwrap(),
+ vec!["v1", "v2"]
+ );
+}
+
+#[tokio::test]
+async fn test_get_function() {
+ let ctx = setup_test_server(vec!["default"]).await;
+ let input_params: Vec<DataField> = serde_json::from_value(json!([
+ {"id": 0, "name": "length", "type": "DOUBLE"},
+ {"id": 1, "name": "width", "type": "DOUBLE"}
+ ]))
+ .unwrap();
+ let return_params: Vec<DataField> = serde_json::from_value(json!([
+ {"id": 0, "name": "area", "type": "DOUBLE"}
+ ]))
+ .unwrap();
+ ctx.server.add_function(Function::new(
+ Identifier::new("default", "area"),
+ Some(input_params),
+ Some(return_params),
+ true,
+ HashMap::from([(
+ "datafusion".to_string(),
+ FunctionDefinition::Sql {
+ definition: "length * width".to_string(),
+ },
+ )]),
+ None,
+ HashMap::new(),
+ ));
+
+ let response = ctx
+ .api
+ .get_function(&Identifier::new("default", "area"))
+ .await
+ .unwrap();
+
+ assert_eq!(response.name.as_deref(), Some("area"));
+ assert_eq!(
+ response
+ .definitions
+ .get("datafusion")
+ .and_then(FunctionDefinition::sql),
+ Some("length * width")
+ );
+}
+
+#[tokio::test]
+async fn test_list_functions() {
+ let ctx = setup_test_server(vec!["default"]).await;
+ ctx.server.set_list_page_size(1);
+ for name in ["zeta", "alpha"] {
+ ctx.server.add_function(Function::new(
+ Identifier::new("default", name),
+ Some(Vec::new()),
+ Some(
+ serde_json::from_value(json!([
+ {"id": 0, "name": "result", "type": "INT"}
+ ]))
+ .unwrap(),
+ ),
+ true,
+ HashMap::from([(
+ "datafusion".to_string(),
+ FunctionDefinition::Sql {
+ definition: "1".to_string(),
+ },
+ )]),
+ None,
+ HashMap::new(),
+ ));
+ }
+
+ assert_eq!(
+ ctx.api.list_functions("default").await.unwrap(),
+ vec!["alpha", "zeta"]
+ );
+}
+
#[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 b8801b8..778337c 100644
--- a/crates/paimon/tests/rest_catalog_test.rs
+++ b/crates/paimon/tests/rest_catalog_test.rs
@@ -27,11 +27,11 @@ use arrow_array::{Array, BinaryArray, Int32Array,
Int64Array, RecordBatch, Strin
use arrow_schema::{DataType as ArrowDataType, Field as ArrowField, Schema as
ArrowSchema};
use futures::TryStreamExt;
use paimon::api::ConfigResponse;
-use paimon::catalog::{Catalog, Identifier, RESTCatalog};
+use paimon::catalog::{Catalog, Function, FunctionDefinition, Identifier,
RESTCatalog, ViewSchema};
use paimon::common::Options;
use paimon::spec::{
- BigIntType, BlobType, BlobViewStruct, DataType, Datum, IntType,
PredicateBuilder, Schema,
- SchemaChange, VarCharType,
+ BigIntType, BlobType, BlobViewStruct, DataField, DataType, Datum, IntType,
PredicateBuilder,
+ Schema, SchemaChange, VarCharType,
};
use paimon::{CatalogOptions, FileSystemCatalog, Table};
@@ -998,3 +998,181 @@ async fn test_catalog_multiple_databases_with_tables() {
assert_eq!(tables_db2.len(), 1);
assert!(tables_db2.contains(&"table1_db2".to_string()));
}
+
+#[tokio::test]
+async fn test_catalog_get_view() {
+ let ctx = setup_catalog(vec!["default"]).await;
+ let schema: ViewSchema = serde_json::from_value(serde_json::json!({
+ "fields": [{"id": 0, "name": "id", "type": "INT"}],
+ "query": "SELECT id FROM source",
+ "dialects": {"datafusion": "SELECT id FROM source WHERE id > 0"},
+ "comment": null,
+ "options": {}
+ }))
+ .unwrap();
+ ctx.server.add_view("default", "active_ids", schema);
+
+ let view = ctx
+ .catalog
+ .get_view(&Identifier::new("default", "active_ids"))
+ .await
+ .unwrap();
+
+ assert_eq!(view.full_name(), "default.active_ids");
+ assert_eq!(
+ view.query_for("datafusion"),
+ "SELECT id FROM source WHERE id > 0"
+ );
+}
+
+#[tokio::test]
+async fn test_catalog_list_views() {
+ let ctx = setup_catalog(vec!["default"]).await;
+ let schema: ViewSchema = serde_json::from_value(serde_json::json!({
+ "fields": [{"id": 0, "name": "id", "type": "INT"}],
+ "query": "SELECT 1 AS id",
+ "dialects": {},
+ "comment": null,
+ "options": {}
+ }))
+ .unwrap();
+ ctx.server.add_view("default", "v2", schema.clone());
+ ctx.server.add_view("default", "v1", schema);
+
+ assert_eq!(
+ ctx.catalog.list_views("default").await.unwrap(),
+ vec!["v1", "v2"]
+ );
+}
+
+#[tokio::test]
+async fn test_catalog_get_function() {
+ let ctx = setup_catalog(vec!["default"]).await;
+ let input_params: Vec<DataField> =
serde_json::from_value(serde_json::json!([
+ {"id": 0, "name": "x", "type": "INT"}
+ ]))
+ .unwrap();
+ let return_params: Vec<DataField> =
serde_json::from_value(serde_json::json!([
+ {"id": 0, "name": "result", "type": "INT"}
+ ]))
+ .unwrap();
+ ctx.server.add_function(Function::new(
+ Identifier::new("default", "plus_one"),
+ Some(input_params),
+ Some(return_params),
+ true,
+ HashMap::from([(
+ "datafusion".to_string(),
+ FunctionDefinition::Sql {
+ definition: "x + 1".to_string(),
+ },
+ )]),
+ None,
+ HashMap::new(),
+ ));
+
+ let function = ctx
+ .catalog
+ .get_function(&Identifier::new("default", "plus_one"))
+ .await
+ .unwrap();
+
+ assert_eq!(function.full_name(), "default.plus_one");
+ assert_eq!(
+ function
+ .definition("datafusion")
+ .and_then(FunctionDefinition::sql),
+ Some("x + 1")
+ );
+}
+
+#[tokio::test]
+async fn test_catalog_list_functions() {
+ let ctx = setup_catalog(vec!["default"]).await;
+ let return_params: Vec<DataField> =
serde_json::from_value(serde_json::json!([
+ {"id": 0, "name": "result", "type": "INT"}
+ ]))
+ .unwrap();
+ for name in ["zeta", "alpha"] {
+ ctx.server.add_function(Function::new(
+ Identifier::new("default", name),
+ Some(Vec::new()),
+ Some(return_params.clone()),
+ true,
+ HashMap::from([(
+ "datafusion".to_string(),
+ FunctionDefinition::Sql {
+ definition: "1".to_string(),
+ },
+ )]),
+ None,
+ HashMap::new(),
+ ));
+ }
+
+ assert_eq!(
+ ctx.catalog.list_functions("default").await.unwrap(),
+ vec!["alpha", "zeta"]
+ );
+}
+
+#[tokio::test]
+async fn test_catalog_get_missing_view_maps_error() {
+ let ctx = setup_catalog(vec!["default"]).await;
+
+ let error = ctx
+ .catalog
+ .get_view(&Identifier::new("default", "missing"))
+ .await
+ .unwrap_err();
+
+ assert!(matches!(
+ error,
+ paimon::Error::ViewNotExist { full_name } if full_name ==
"default.missing"
+ ));
+}
+
+#[tokio::test]
+async fn test_catalog_get_missing_function_maps_error() {
+ let ctx = setup_catalog(vec!["default"]).await;
+
+ let error = ctx
+ .catalog
+ .get_function(&Identifier::new("default", "missing"))
+ .await
+ .unwrap_err();
+
+ assert!(matches!(
+ error,
+ paimon::Error::FunctionNotExist { full_name } if full_name ==
"default.missing"
+ ));
+}
+
+#[tokio::test]
+async fn test_catalog_maps_unsupported_view_and_function_endpoints() {
+ let ctx = setup_catalog(vec!["default"]).await;
+ ctx.server.set_view_function_endpoints_unsupported();
+
+ assert!(matches!(
+ ctx.catalog.list_views("default").await.unwrap_err(),
+ paimon::Error::Unsupported { .. }
+ ));
+ assert!(matches!(
+ ctx.catalog
+ .get_view(&Identifier::new("default", "view"))
+ .await
+ .unwrap_err(),
+ paimon::Error::Unsupported { .. }
+ ));
+ assert!(matches!(
+ ctx.catalog.list_functions("default").await.unwrap_err(),
+ paimon::Error::Unsupported { .. }
+ ));
+ assert!(matches!(
+ ctx.catalog
+ .get_function(&Identifier::new("default", "function"))
+ .await
+ .unwrap_err(),
+ paimon::Error::Unsupported { .. }
+ ));
+}
diff --git a/crates/paimon/tests/rest_object_models_test.rs
b/crates/paimon/tests/rest_object_models_test.rs
new file mode 100644
index 0000000..3c16bbd
--- /dev/null
+++ b/crates/paimon/tests/rest_object_models_test.rs
@@ -0,0 +1,154 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you 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 std::collections::HashMap;
+
+use paimon::catalog::{Function, FunctionDefinition, Identifier, View,
ViewSchema};
+use paimon::spec::DataField;
+use serde_json::json;
+
+#[test]
+fn deserialize_view_schema_contract() {
+ let schema: ViewSchema = serde_json::from_value(json!({
+ "fields": [{"id": 0, "name": "id", "type": "INT"}],
+ "query": "SELECT id FROM source",
+ "dialects": {"datafusion": "SELECT id FROM source"},
+ "comment": "active ids",
+ "options": {"owner": "analytics"}
+ }))
+ .unwrap();
+
+ assert_eq!(schema.fields()[0].name(), "id");
+ assert_eq!(schema.query_for("datafusion"), "SELECT id FROM source");
+ assert_eq!(schema.comment(), Some("active ids"));
+ assert_eq!(
+ schema.options().get("owner").map(String::as_str),
+ Some("analytics")
+ );
+}
+
+#[test]
+fn view_binds_identifier_to_schema() {
+ let schema: ViewSchema = serde_json::from_value(json!({
+ "fields": [{"id": 0, "name": "id", "type": "INT"}],
+ "query": "SELECT id FROM source",
+ "dialects": {"datafusion": "SELECT id FROM source WHERE id > 0"},
+ "comment": null,
+ "options": {}
+ }))
+ .unwrap();
+ let view = View::new(Identifier::new("analytics", "active_ids"), schema);
+
+ assert_eq!(view.name(), "active_ids");
+ assert_eq!(view.full_name(), "analytics.active_ids");
+ assert_eq!(
+ view.query_for("datafusion"),
+ "SELECT id FROM source WHERE id > 0"
+ );
+}
+
+#[test]
+fn deserialize_sql_function_definition_contract() {
+ let definition: FunctionDefinition = serde_json::from_value(json!({
+ "type": "sql",
+ "definition": "length * width"
+ }))
+ .unwrap();
+
+ assert_eq!(definition.sql(), Some("length * width"));
+}
+
+#[test]
+fn deserialize_lambda_function_definition_contract() {
+ let definition: FunctionDefinition = serde_json::from_value(json!({
+ "type": "lambda",
+ "definition": "(x) -> x + 1",
+ "language": "java"
+ }))
+ .unwrap();
+
+ assert!(matches!(
+ definition,
+ FunctionDefinition::Lambda { definition, language }
+ if definition == "(x) -> x + 1" && language == "java"
+ ));
+}
+
+#[test]
+fn deserialize_file_function_definition_contract() {
+ let definition: FunctionDefinition = serde_json::from_value(json!({
+ "type": "file",
+ "fileResources": [{"resourceType": "jar", "uri":
"file:///functions.jar"}],
+ "language": "java",
+ "className": "com.example.Area",
+ "functionName": "eval"
+ }))
+ .unwrap();
+
+ assert!(matches!(
+ definition,
+ FunctionDefinition::File {
+ file_resources,
+ language,
+ class_name,
+ function_name,
+ } if file_resources.len() == 1
+ && file_resources[0].resource_type() == "jar"
+ && file_resources[0].uri() == "file:///functions.jar"
+ && language == "java"
+ && class_name == "com.example.Area"
+ && function_name == "eval"
+ ));
+}
+
+#[test]
+fn function_binds_identifier_signature_and_definition() {
+ let input_params: Vec<DataField> = serde_json::from_value(json!([
+ {"id": 0, "name": "length", "type": "DOUBLE"},
+ {"id": 1, "name": "width", "type": "DOUBLE"}
+ ]))
+ .unwrap();
+ let return_params: Vec<DataField> = serde_json::from_value(json!([
+ {"id": 0, "name": "area", "type": "DOUBLE"}
+ ]))
+ .unwrap();
+ let definitions = HashMap::from([(
+ "datafusion".to_string(),
+ FunctionDefinition::Sql {
+ definition: "length * width".to_string(),
+ },
+ )]);
+ let function = Function::new(
+ Identifier::new("analytics", "area"),
+ Some(input_params),
+ Some(return_params),
+ true,
+ definitions,
+ Some("rectangle area".to_string()),
+ HashMap::new(),
+ );
+
+ assert_eq!(function.full_name(), "analytics.area");
+ assert_eq!(function.input_params().unwrap()[0].name(), "length");
+ assert_eq!(function.return_params().unwrap()[0].name(), "area");
+ assert_eq!(
+ function
+ .definition("datafusion")
+ .and_then(FunctionDefinition::sql),
+ Some("length * width")
+ );
+}
diff --git a/docs/src/sql.md b/docs/src/sql.md
index 6651344..2d58ba6 100644
--- a/docs/src/sql.md
+++ b/docs/src/sql.md
@@ -76,6 +76,60 @@ catalogs; registering a catalog also registers the built-in
scalar function
feature is enabled) against it. It also manages session-scoped dynamic options
internally for `SET`/`RESET` support.
+### Existing 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 that already exist in the
+catalog. This integration is read-only: it does not add `CREATE`, `ALTER`, or
+`DROP` support for persistent views or functions.
+
+Persistent views resolve through the normal DataFusion catalog path, so they
+can be queried wherever a table can be used:
+
+```sql
+SELECT * FROM analytics_view;
+SELECT * FROM paimon.reporting.daily_orders;
+```
+
+For a view, the `datafusion` entry in `schema.dialects` is preferred. If that
+entry is absent, DataFusion uses the view's default `schema.query`. Unqualified
+relations inside the stored query resolve against the view's owning catalog and
+database, not the caller's current database. The REST-declared output fields
+are authoritative: query results are matched by position, renamed to the
+declared field names, and cast to the declared types. Recursive view
+dependencies are rejected during planning.
+
+REST SQL scalar functions support bare names in the current catalog/database
+and fully qualified three-part names:
+
+```sql
+SELECT normalize_score(score) FROM scores;
+SELECT paimon.reporting.normalize_score(score) FROM scores;
+```
+
+A function is executable only when all of the following are true:
+
+- `definitions.datafusion` exists and has `type: "sql"`;
+- input parameters are declared and the call supplies the exact number of
+ positional expression arguments;
+- exactly one return parameter is declared;
+- the function is deterministic;
+- the SQL definition is a scalar expression that references inputs by their
+ declared parameter names.
+
+Nested REST SQL functions are supported. Bare function names inside a stored
+definition resolve in that function's owning catalog/database. Recursive
+function dependencies, missing DataFusion SQL definitions, undeclared
+identifiers, named arguments, and incompatible return types fail during
+planning. If no REST function exists for a bare name, normal DataFusion
+built-in or registered-function resolution continues.
+
+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.
+
## Data Types
The following SQL data types are supported in CREATE TABLE and mapped to their
corresponding Paimon types:
