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 4e67c89 feat(datafusion): support REST Catalog CREATE VIEW (#497)
4e67c89 is described below
commit 4e67c892dce06784430735e78e077c1e0e8c1e9c
Author: Jingsong Lee <[email protected]>
AuthorDate: Sat Jul 11 08:56:32 2026 +0800
feat(datafusion): support REST Catalog CREATE VIEW (#497)
---
crates/integrations/datafusion/README.md | 20 +-
crates/integrations/datafusion/src/sql_context.rs | 482 +++++++++++++++++++++-
crates/paimon/src/api/api_request.rs | 19 +-
crates/paimon/src/api/mod.rs | 2 +-
crates/paimon/src/api/rest_api.rs | 15 +-
crates/paimon/src/catalog/mod.rs | 12 +
crates/paimon/src/catalog/rest/rest_catalog.rs | 33 ++
crates/paimon/src/catalog/view.rs | 17 +
crates/paimon/src/error.rs | 2 +
crates/paimon/tests/mock_server.rs | 52 ++-
crates/paimon/tests/rest_api_test.rs | 24 ++
crates/paimon/tests/rest_catalog_test.rs | 118 ++++++
crates/paimon/tests/rest_object_models_test.rs | 57 +++
docs/src/sql.md | 41 +-
14 files changed, 864 insertions(+), 30 deletions(-)
diff --git a/crates/integrations/datafusion/README.md
b/crates/integrations/datafusion/README.md
index cc7a3a1..0f6a664 100644
--- a/crates/integrations/datafusion/README.md
+++ b/crates/integrations/datafusion/README.md
@@ -26,19 +26,28 @@ This crate contains the integration of [Apache
DataFusion](https://datafusion.ap
## REST Catalog views and SQL functions
-`SQLContext` can read persistent views and SQL scalar functions that already
exist in a Paimon
-REST Catalog. No view or function DDL is added.
+`SQLContext` can read persistent views and SQL scalar functions in a Paimon
REST Catalog. It can
+also create persistent REST Catalog views:
+
+```sql
+CREATE VIEW [IF NOT EXISTS] view_name [(column_name, ...)] AS query;
+```
- 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.
+- `CREATE VIEW` infers stored field types and nullability from the defining
query. An optional
+ column list overrides names only and must match the query output.
Unqualified relations and REST
+ SQL functions are planned in the new view's owning catalog and database. `IF
NOT EXISTS` is
+ handled atomically by the catalog.
- 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.
+- `CREATE OR REPLACE VIEW`, materialized/secure views, comments/options,
persistent `ALTER VIEW` /
+ `DROP VIEW`, and persistent function DDL are not supported. Lambda/file
functions, named
+ arguments, multiple return values, non-deterministic functions, and calls
made directly through
+ a raw DataFusion `SessionContext` are also not supported.
Use `SQLContext::sql` for function expansion:
@@ -46,6 +55,7 @@ Use `SQLContext::sql` for function expansion:
let mut ctx = paimon_datafusion::SQLContext::new();
ctx.register_catalog("paimon", rest_catalog).await?;
+ctx.sql("CREATE VIEW daily_scores AS SELECT normalize_score(score) AS score
FROM scores").await?;
let view = ctx.sql("SELECT * FROM analytics_view").await?;
let function = ctx.sql("SELECT normalize_score(score) FROM scores").await?;
```
diff --git a/crates/integrations/datafusion/src/sql_context.rs
b/crates/integrations/datafusion/src/sql_context.rs
index d136492..bc00552 100644
--- a/crates/integrations/datafusion/src/sql_context.rs
+++ b/crates/integrations/datafusion/src/sql_context.rs
@@ -30,10 +30,11 @@
//! - `ALTER TABLE db.t RENAME COLUMN old TO new`
//! - `ALTER TABLE db.t RENAME TO new_name`
//! - `ALTER TABLE db.t DROP PARTITION (col = val, ...)`
+//! - `CREATE VIEW [IF NOT EXISTS] view [(col, ...)] AS query`
//! - `TRUNCATE TABLE db.t`
//! - `TRUNCATE TABLE db.t PARTITION (col = val, ...)`
-use std::collections::HashMap;
+use std::collections::{HashMap, HashSet};
use std::sync::Arc;
use datafusion::arrow::array::{
@@ -48,6 +49,7 @@ use datafusion::datasource::{MemTable, TableProvider};
use datafusion::error::{DataFusionError, Result as DFResult};
use datafusion::execution::SessionStateBuilder;
use datafusion::prelude::{DataFrame, SessionContext};
+use datafusion::sql::planner::IdentNormalizer;
use datafusion::sql::sqlparser::ast::{
AlterTableOperation, BinaryLength, CharacterLength, ColumnDef,
ColumnOption, CreateTable,
CreateTableOptions, CreateView, Delete, Expr as SqlExpr, FromTable,
Insert, Merge, ObjectName,
@@ -1348,16 +1350,14 @@ impl SQLContext {
));
}
- let view_name = create_view.name.to_string();
- let table_ref: TableReference = view_name.as_str().into();
- let (catalog, database, name) =
self.resolve_temp_table_name(table_ref)?;
-
- // Use DataFusion's SQL planner to convert the sqlparser Query into a
LogicalPlan
let query_sql = create_view.query.to_string();
- let df = self.ctx.sql(&query_sql).await?;
- let logical_plan = df.logical_plan().clone();
if create_view.temporary {
+ let view_name = create_view.name.to_string();
+ let table_ref: TableReference = view_name.as_str().into();
+ let (catalog, database, name) =
self.resolve_temp_table_name(table_ref)?;
+ let df = self.ctx.sql(&query_sql).await?;
+ let logical_plan = df.logical_plan().clone();
if create_view.if_not_exists
&&
self.temp_table_exist(format!("{catalog}.{database}.{name}"))?
{
@@ -1368,10 +1368,65 @@ impl SQLContext {
self.register_temp_table(format!("{catalog}.{database}.{name}"),
Arc::new(view_table))?;
ok_result(&self.ctx)
} else {
- Err(DataFusionError::Plan(
- "CREATE VIEW (non-temporary) is not supported. Use CREATE
TEMPORARY VIEW instead."
- .to_string(),
- ))
+ validate_persistent_create_view(create_view)?;
+ let (catalog, catalog_name, identifier) =
+ self.resolve_catalog_and_table(&create_view.name)?;
+ let mut state = self.ctx.state();
+ state.config_mut().options_mut().catalog.default_catalog =
catalog_name.clone();
+ state.config_mut().options_mut().catalog.default_schema =
+ identifier.database().to_string();
+ let expanded_query = crate::sql_function::expand_sql(
+ &query_sql,
+ &self.catalogs,
+ &catalog_name,
+ identifier.database(),
+ )
+ .await?;
+ let logical_plan =
state.create_logical_plan(&expanded_query).await?;
+ let mut arrow_fields = logical_plan
+ .schema()
+ .as_arrow()
+ .fields()
+ .iter()
+ .map(|field| field.as_ref().clone())
+ .collect::<Vec<_>>();
+ if !create_view.columns.is_empty() && create_view.columns.len() !=
arrow_fields.len() {
+ return Err(DataFusionError::Plan(format!(
+ "view column list has {} columns but query produces {}
columns",
+ create_view.columns.len(),
+ arrow_fields.len()
+ )));
+ }
+ let column_names = create_view
+ .columns
+ .iter()
+ .map(|column|
IdentNormalizer::default().normalize(column.name.clone()))
+ .collect::<Vec<_>>();
+ let mut unique_names = HashSet::with_capacity(column_names.len());
+ for name in &column_names {
+ if !unique_names.insert(name.clone()) {
+ return Err(DataFusionError::Plan(format!(
+ "duplicate view column name '{name}'"
+ )));
+ }
+ }
+ for (field, name) in arrow_fields.iter_mut().zip(column_names) {
+ *field = field.clone().with_name(name);
+ }
+ let fields = paimon::arrow::arrow_fields_to_paimon(&arrow_fields)
+ .map_err(to_datafusion_error)?;
+ let schema = paimon::catalog::ViewSchema::new(
+ fields,
+ query_sql.clone(),
+ HashMap::from([("datafusion".to_string(), query_sql)]),
+ None,
+ HashMap::new(),
+ );
+ catalog
+ .create_view(&identifier, schema, create_view.if_not_exists)
+ .await
+ .map_err(to_datafusion_error)?;
+ ok_result(&self.ctx)
}
}
@@ -1514,6 +1569,74 @@ impl SQLContext {
}
}
+fn validate_persistent_create_view(create_view: &CreateView) -> DFResult<()> {
+ let unsupported = if create_view.or_alter {
+ Some("CREATE OR ALTER VIEW is not supported")
+ } else if create_view.or_replace {
+ Some("CREATE OR REPLACE VIEW is not supported")
+ } else if create_view.secure {
+ Some("CREATE SECURE VIEW is not supported")
+ } else if create_view.copy_grants {
+ Some("CREATE VIEW COPY GRANTS is not supported")
+ } else if create_view.name_before_not_exists {
+ Some("CREATE VIEW with the name before IF NOT EXISTS is not supported")
+ } else {
+ match &create_view.options {
+ CreateTableOptions::None => None,
+ CreateTableOptions::With(_) => Some("CREATE VIEW WITH options are
not supported"),
+ CreateTableOptions::Options(_) => Some("CREATE VIEW OPTIONS are
not supported"),
+ _ => Some("CREATE VIEW options are not supported"),
+ }
+ };
+ if let Some(message) = unsupported {
+ return Err(DataFusionError::Plan(message.to_string()));
+ }
+ if create_view.comment.is_some() {
+ return Err(DataFusionError::Plan(
+ "CREATE VIEW COMMENT is not supported".to_string(),
+ ));
+ }
+ if !create_view.cluster_by.is_empty() {
+ return Err(DataFusionError::Plan(
+ "CREATE VIEW CLUSTER BY is not supported".to_string(),
+ ));
+ }
+ if create_view.to.is_some() {
+ return Err(DataFusionError::Plan(
+ "CREATE VIEW TO is not supported".to_string(),
+ ));
+ }
+ if create_view.with_no_schema_binding {
+ return Err(DataFusionError::Plan(
+ "CREATE VIEW WITH NO SCHEMA BINDING is not supported".to_string(),
+ ));
+ }
+ if create_view.params.is_some() {
+ return Err(DataFusionError::Plan(
+ "CREATE VIEW view parameters are not supported".to_string(),
+ ));
+ }
+ if create_view
+ .columns
+ .iter()
+ .any(|column| column.data_type.is_some())
+ {
+ return Err(DataFusionError::Plan(
+ "CREATE VIEW column data types are not supported".to_string(),
+ ));
+ }
+ if create_view
+ .columns
+ .iter()
+ .any(|column| column.options.is_some())
+ {
+ return Err(DataFusionError::Plan(
+ "CREATE VIEW column options are not supported".to_string(),
+ ));
+ }
+ Ok(())
+}
+
/// Quick check whether the SQL looks like a CREATE TABLE statement.
/// Skips leading whitespace, `--` line comments, and `/* */` block comments.
fn looks_like_create_table(sql: &str) -> bool {
@@ -2882,6 +3005,28 @@ mod tests {
full_name: identifier.full_name(),
})
}
+
+ async fn create_view(
+ &self,
+ identifier: &Identifier,
+ schema: paimon::catalog::ViewSchema,
+ ignore_if_exists: bool,
+ ) -> paimon::Result<()> {
+ let mut views = self.views.lock().unwrap();
+ if views.contains_key(identifier) {
+ if ignore_if_exists {
+ return Ok(());
+ }
+ return Err(paimon::Error::ViewAlreadyExist {
+ full_name: identifier.full_name(),
+ });
+ }
+ views.insert(
+ identifier.clone(),
+ paimon::catalog::View::new(identifier.clone(), schema),
+ );
+ Ok(())
+ }
}
async fn make_sql_context(catalog: Arc<MockCatalog>) -> SQLContext {
@@ -2970,6 +3115,319 @@ mod tests {
));
}
+ #[tokio::test]
+ async fn persistent_rest_catalog_view_can_be_created_and_read() {
+ let catalog = Arc::new(MockCatalog::new());
+ let ctx = make_sql_context(catalog).await;
+
+ ctx.sql(
+ "CREATE VIEW answer_view AS \
+ SELECT CAST(42 AS BIGINT) AS answer",
+ )
+ .await
+ .unwrap();
+
+ 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 persistent_rest_catalog_view_infers_type_and_nullability() {
+ let catalog = Arc::new(MockCatalog::new());
+ let ctx = make_sql_context(Arc::clone(&catalog)).await;
+
+ ctx.sql(
+ "CREATE VIEW paimon.default.inferred_view AS \
+ SELECT CAST(1 AS BIGINT) AS required, CAST(NULL AS BIGINT) AS
optional",
+ )
+ .await
+ .unwrap();
+
+ let view = catalog
+ .get_view(&Identifier::new("default", "inferred_view"))
+ .await
+ .unwrap();
+ let fields = view.schema().fields();
+ assert!(matches!(fields[0].data_type(), PaimonDataType::BigInt(_)));
+ assert!(!fields[0].data_type().is_nullable());
+ assert!(matches!(fields[1].data_type(), PaimonDataType::BigInt(_)));
+ assert!(fields[1].data_type().is_nullable());
+ }
+
+ #[tokio::test]
+ async fn
persistent_rest_catalog_view_expands_function_in_owning_database() {
+ let catalog = Arc::new(MockCatalog::new());
+ add_unary_sql_function_in_database(&catalog, "default", "plus_one", "x
+ 1", true);
+ add_unary_sql_function_in_database(&catalog, "other", "plus_one", "x +
100", true);
+ let ctx = make_sql_context(Arc::clone(&catalog)).await;
+ ctx.set_current_database("other").await.unwrap();
+
+ ctx.sql(
+ "CREATE VIEW paimon.default.function_view AS \
+ SELECT plus_one(41) AS answer",
+ )
+ .await
+ .unwrap();
+
+ let view = catalog
+ .get_view(&Identifier::new("default", "function_view"))
+ .await
+ .unwrap();
+ assert!(view.query_for("datafusion").contains("plus_one(41)"));
+ let batches = ctx
+ .sql("SELECT * FROM paimon.default.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 persistent_rest_catalog_view_binds_to_owning_database() {
+ let catalog = Arc::new(MockCatalog::new());
+ add_bigint_view(
+ &catalog,
+ "default",
+ "base_view",
+ "SELECT CAST(42 AS BIGINT) AS answer",
+ );
+ let other_schema = serde_json::from_value(serde_json::json!({
+ "fields": [
+ {"id": 0, "name": "answer", "type": "BIGINT"},
+ {"id": 1, "name": "extra", "type": "BIGINT"}
+ ],
+ "query": "SELECT CAST(7 AS BIGINT) AS answer, CAST(8 AS BIGINT) AS
extra",
+ "dialects": {},
+ "comment": null,
+ "options": {}
+ }))
+ .unwrap();
+ catalog.add_view(paimon::catalog::View::new(
+ Identifier::new("other", "base_view"),
+ other_schema,
+ ));
+ let ctx = make_sql_context(Arc::clone(&catalog)).await;
+ ctx.set_current_database("other").await.unwrap();
+
+ ctx.sql("CREATE VIEW paimon.default.created_view AS SELECT * FROM
base_view")
+ .await
+ .unwrap();
+
+ let view = catalog
+ .get_view(&Identifier::new("default", "created_view"))
+ .await
+ .unwrap();
+ assert_eq!(view.schema().fields().len(), 1);
+ assert_eq!(view.schema().fields()[0].name(), "answer");
+ }
+
+ #[tokio::test]
+ async fn persistent_rest_catalog_view_columns_override_names() {
+ let catalog = Arc::new(MockCatalog::new());
+ let ctx = make_sql_context(Arc::clone(&catalog)).await;
+
+ ctx.sql(
+ "CREATE VIEW paimon.default.named_view (renamed) \
+ AS SELECT CAST(42 AS BIGINT) AS answer",
+ )
+ .await
+ .unwrap();
+
+ let view = catalog
+ .get_view(&Identifier::new("default", "named_view"))
+ .await
+ .unwrap();
+ assert_eq!(view.schema().fields()[0].name(), "renamed");
+ }
+
+ #[tokio::test]
+ async fn persistent_rest_catalog_view_rejects_column_count_mismatch() {
+ let catalog = Arc::new(MockCatalog::new());
+ let ctx = make_sql_context(catalog).await;
+
+ let error = ctx
+ .sql(
+ "CREATE VIEW paimon.default.invalid_view (first, second) \
+ AS SELECT CAST(42 AS BIGINT) AS answer",
+ )
+ .await
+ .unwrap_err();
+
+ assert!(error
+ .to_string()
+ .contains("view column list has 2 columns but query produces 1
columns"));
+ }
+
+ #[tokio::test]
+ async fn persistent_rest_catalog_view_rejects_duplicate_column_names() {
+ let catalog = Arc::new(MockCatalog::new());
+ let ctx = make_sql_context(catalog).await;
+
+ let error = ctx
+ .sql(
+ "CREATE VIEW paimon.default.invalid_view (duplicate,
duplicate) \
+ AS SELECT CAST(42 AS BIGINT), CAST(7 AS BIGINT)",
+ )
+ .await
+ .unwrap_err();
+
+ assert!(
+ error
+ .to_string()
+ .contains("duplicate view column name 'duplicate'"),
+ "unexpected error: {error}"
+ );
+ }
+
+ #[tokio::test]
+ async fn persistent_rest_catalog_view_rejects_duplicate_inferred_names() {
+ let catalog = Arc::new(MockCatalog::new());
+ let ctx = make_sql_context(catalog).await;
+
+ let error = ctx
+ .sql(
+ "CREATE VIEW paimon.default.invalid_view AS \
+ SELECT CAST(42 AS BIGINT) AS duplicate, \
+ CAST(7 AS BIGINT) AS duplicate",
+ )
+ .await
+ .unwrap_err();
+
+ assert!(
+ error
+ .to_string()
+ .contains("Projections require unique expression names"),
+ "unexpected error: {error}"
+ );
+ }
+
+ #[tokio::test]
+ async fn
persistent_rest_catalog_view_if_not_exists_preserves_existing_view() {
+ let catalog = Arc::new(MockCatalog::new());
+ let ctx = make_sql_context(Arc::clone(&catalog)).await;
+
+ ctx.sql(
+ "CREATE VIEW paimon.default.existing_view \
+ AS SELECT CAST(1 AS BIGINT) AS answer",
+ )
+ .await
+ .unwrap();
+ ctx.sql(
+ "CREATE VIEW IF NOT EXISTS paimon.default.existing_view \
+ AS SELECT CAST(2 AS BIGINT) AS answer",
+ )
+ .await
+ .unwrap();
+
+ let view = catalog
+ .get_view(&Identifier::new("default", "existing_view"))
+ .await
+ .unwrap();
+ assert!(view.query_for("datafusion").contains("CAST(1 AS BIGINT)"));
+ }
+
+ #[tokio::test]
+ async fn persistent_rest_catalog_view_rejects_or_replace() {
+ let catalog = Arc::new(MockCatalog::new());
+ let ctx = make_sql_context(catalog).await;
+
+ let error = ctx
+ .sql(
+ "CREATE OR REPLACE VIEW paimon.default.invalid_view \
+ AS SELECT CAST(1 AS BIGINT) AS answer",
+ )
+ .await
+ .unwrap_err();
+
+ assert!(error
+ .to_string()
+ .contains("CREATE OR REPLACE VIEW is not supported"));
+ }
+
+ #[tokio::test]
+ async fn persistent_rest_catalog_view_rejects_unsupported_clauses() {
+ let cases = [
+ (
+ "CREATE OR ALTER VIEW paimon.default.invalid_view AS SELECT 1",
+ "OR ALTER",
+ ),
+ (
+ "CREATE SECURE VIEW paimon.default.invalid_view AS SELECT 1",
+ "SECURE",
+ ),
+ (
+ "CREATE VIEW paimon.default.invalid_view COPY GRANTS AS SELECT
1",
+ "COPY GRANTS",
+ ),
+ (
+ "CREATE VIEW paimon.default.invalid_view IF NOT EXISTS AS
SELECT 1",
+ "name before IF NOT EXISTS",
+ ),
+ (
+ "CREATE VIEW paimon.default.invalid_view WITH ('key' =
'value') AS SELECT 1",
+ "WITH options",
+ ),
+ (
+ "CREATE VIEW paimon.default.invalid_view OPTIONS(key =
'value') AS SELECT 1",
+ "OPTIONS",
+ ),
+ (
+ "CREATE VIEW paimon.default.invalid_view COMMENT = 'comment'
AS SELECT 1",
+ "COMMENT",
+ ),
+ (
+ "CREATE VIEW paimon.default.invalid_view CLUSTER BY (answer)
AS SELECT 1 AS answer",
+ "CLUSTER BY",
+ ),
+ (
+ "CREATE VIEW paimon.default.invalid_view TO default.sink AS
SELECT 1",
+ "TO",
+ ),
+ (
+ "CREATE VIEW paimon.default.invalid_view AS SELECT 1 WITH NO
SCHEMA BINDING",
+ "WITH NO SCHEMA BINDING",
+ ),
+ (
+ "CREATE ALGORITHM = MERGE VIEW paimon.default.invalid_view AS
SELECT 1",
+ "view parameters",
+ ),
+ (
+ "CREATE VIEW paimon.default.invalid_view (answer COMMENT
'comment') AS SELECT 1",
+ "column options",
+ ),
+ ];
+
+ for (sql, clause) in cases {
+ let catalog = Arc::new(MockCatalog::new());
+ let ctx = make_sql_context(catalog).await;
+ let error = ctx.sql(sql).await.unwrap_err();
+ assert!(
+ error.to_string().contains(clause),
+ "expected error for {clause}, got: {error}"
+ );
+ }
+ }
+
#[tokio::test]
async fn rest_catalog_view_is_planned_lazily() {
let catalog = Arc::new(MockCatalog::new());
diff --git a/crates/paimon/src/api/api_request.rs
b/crates/paimon/src/api/api_request.rs
index 28e4bac..fabcf4a 100644
--- a/crates/paimon/src/api/api_request.rs
+++ b/crates/paimon/src/api/api_request.rs
@@ -23,7 +23,7 @@ use serde::{Deserialize, Serialize};
use std::collections::HashMap;
use crate::{
- catalog::Identifier,
+ catalog::{Identifier, ViewSchema},
spec::{Schema, SchemaChange},
};
@@ -98,6 +98,23 @@ impl CreateTableRequest {
}
}
+/// Request to create a persistent view.
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(rename_all = "camelCase")]
+pub struct CreateViewRequest {
+ /// The identifier for the view to create.
+ pub identifier: Identifier,
+ /// The schema and SQL definitions for the view.
+ pub schema: ViewSchema,
+}
+
+impl CreateViewRequest {
+ /// Create a new create-view request.
+ pub fn new(identifier: Identifier, schema: ViewSchema) -> Self {
+ Self { identifier, schema }
+ }
+}
+
/// Request to alter a table's schema.
///
/// Wire-compatible with Java Paimon's `AlterTableRequest` (`{"changes":
[...]}`).
diff --git a/crates/paimon/src/api/mod.rs b/crates/paimon/src/api/mod.rs
index a81067b..643b1ec 100644
--- a/crates/paimon/src/api/mod.rs
+++ b/crates/paimon/src/api/mod.rs
@@ -32,7 +32,7 @@ mod api_response;
// Re-export request types
pub use api_request::{
AlterDatabaseRequest, AlterTableRequest, CreateDatabaseRequest,
CreateTableRequest,
- RenameTableRequest,
+ CreateViewRequest, RenameTableRequest,
};
// Re-export response types
diff --git a/crates/paimon/src/api/rest_api.rs
b/crates/paimon/src/api/rest_api.rs
index 421a23d..37f89fd 100644
--- a/crates/paimon/src/api/rest_api.rs
+++ b/crates/paimon/src/api/rest_api.rs
@@ -23,14 +23,14 @@
use std::collections::HashMap;
use crate::api::rest_client::HttpClient;
-use crate::catalog::Identifier;
+use crate::catalog::{Identifier, ViewSchema};
use crate::common::{CatalogOptions, Options};
use crate::spec::{Partition, PartitionStatistics, Schema, SchemaChange,
Snapshot};
use crate::Result;
use super::api_request::{
AlterDatabaseRequest, AlterTableRequest, CreateDatabaseRequest,
CreateTableRequest,
- RenameTableRequest,
+ CreateViewRequest, RenameTableRequest,
};
use super::api_response::{
ConfigResponse, GetDatabaseResponse, GetFunctionResponse,
GetTableResponse, GetViewResponse,
@@ -397,6 +397,17 @@ impl RESTApi {
// ==================== View Operations ====================
+ /// Create a persistent view.
+ pub async fn create_view(&self, identifier: &Identifier, schema:
ViewSchema) -> Result<()> {
+ let database = identifier.database();
+ let view = identifier.object();
+ validate_non_empty_multi(&[(database, "database name"), (view, "view
name")])?;
+ let path = self.resource_paths.views(database);
+ let request = CreateViewRequest::new(identifier.clone(), schema);
+ let _resp: serde_json::Value = self.client.post(&path,
&request).await?;
+ Ok(())
+ }
+
/// List persistent views in a database.
pub async fn list_views(&self, database: &str) -> Result<Vec<String>> {
validate_non_empty(database, "database name")?;
diff --git a/crates/paimon/src/catalog/mod.rs b/crates/paimon/src/catalog/mod.rs
index fa653a1..8d17300 100644
--- a/crates/paimon/src/catalog/mod.rs
+++ b/crates/paimon/src/catalog/mod.rs
@@ -377,6 +377,18 @@ pub trait Catalog: Send + Sync {
// ======================= view methods ===============================
+ /// Create a persistent view.
+ async fn create_view(
+ &self,
+ _identifier: &Identifier,
+ _schema: ViewSchema,
+ _ignore_if_exists: bool,
+ ) -> Result<()> {
+ Err(Error::Unsupported {
+ message: "Catalog does not support views".to_string(),
+ })
+ }
+
/// List persistent view names in a database.
async fn list_views(&self, _database_name: &str) -> Result<Vec<String>> {
Err(Error::Unsupported {
diff --git a/crates/paimon/src/catalog/rest/rest_catalog.rs
b/crates/paimon/src/catalog/rest/rest_catalog.rs
index 6901312..58eb63b 100644
--- a/crates/paimon/src/catalog/rest/rest_catalog.rs
+++ b/crates/paimon/src/catalog/rest/rest_catalog.rs
@@ -278,6 +278,22 @@ impl Catalog for RESTCatalog {
})
}
+ async fn create_view(
+ &self,
+ identifier: &Identifier,
+ schema: crate::catalog::ViewSchema,
+ ignore_if_exists: bool,
+ ) -> Result<()> {
+ let result = self
+ .api
+ .create_view(identifier, schema)
+ .await
+ .map_err(|error| map_rest_error_for_create_view(error,
identifier));
+ ignore_error_if(result, |error| {
+ ignore_if_exists && matches!(error, Error::ViewAlreadyExist { .. })
+ })
+ }
+
async fn list_views(&self, database_name: &str) -> Result<Vec<String>> {
self.api
.list_views(database_name)
@@ -403,9 +419,26 @@ fn map_rest_error_for_table(err: Error, identifier:
&Identifier) -> Error {
}
}
+/// Map a REST API error from creating a persistent view.
+fn map_rest_error_for_create_view(err: Error, identifier: &Identifier) ->
Error {
+ match err {
+ Error::RestApi {
+ source: RestError::NoSuchResource { .. },
+ } => Error::DatabaseNotExist {
+ database: identifier.database().to_string(),
+ },
+ other => map_rest_error_for_view(other, identifier),
+ }
+}
+
/// 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::AlreadyExists { .. },
+ } => Error::ViewAlreadyExist {
+ full_name: identifier.full_name(),
+ },
Error::RestApi {
source: RestError::NoSuchResource { .. },
} => Error::ViewNotExist {
diff --git a/crates/paimon/src/catalog/view.rs
b/crates/paimon/src/catalog/view.rs
index d99f819..9921cf0 100644
--- a/crates/paimon/src/catalog/view.rs
+++ b/crates/paimon/src/catalog/view.rs
@@ -76,6 +76,23 @@ pub struct ViewSchema {
}
impl ViewSchema {
+ /// Create a persistent view schema.
+ pub fn new(
+ fields: Vec<DataField>,
+ query: String,
+ dialects: HashMap<String, String>,
+ comment: Option<String>,
+ options: HashMap<String, String>,
+ ) -> Self {
+ Self {
+ fields,
+ query,
+ dialects,
+ comment,
+ options,
+ }
+ }
+
/// Declared output fields of the view.
pub fn fields(&self) -> &[DataField] {
&self.fields
diff --git a/crates/paimon/src/error.rs b/crates/paimon/src/error.rs
index ddac874..507ee3a 100644
--- a/crates/paimon/src/error.rs
+++ b/crates/paimon/src/error.rs
@@ -102,6 +102,8 @@ pub enum Error {
TableAlreadyExist { full_name: String },
#[snafu(display("Table {} does not exist.", full_name))]
TableNotExist { full_name: String },
+ #[snafu(display("View {} already exists.", full_name))]
+ ViewAlreadyExist { full_name: String },
#[snafu(display("View {} does not exist.", full_name))]
ViewNotExist { full_name: String },
#[snafu(display("Function {} does not exist.", full_name))]
diff --git a/crates/paimon/tests/mock_server.rs
b/crates/paimon/tests/mock_server.rs
index 99e11e6..35a8a3f 100644
--- a/crates/paimon/tests/mock_server.rs
+++ b/crates/paimon/tests/mock_server.rs
@@ -34,8 +34,8 @@ use std::sync::{Arc, Mutex};
use tokio::task::JoinHandle;
use paimon::api::{
- AlterDatabaseRequest, AlterTableRequest, AuditRESTResponse,
ConfigResponse, ErrorResponse,
- GetDatabaseResponse, GetFunctionResponse, GetTableResponse,
GetViewResponse,
+ AlterDatabaseRequest, AlterTableRequest, AuditRESTResponse,
ConfigResponse, CreateViewRequest,
+ ErrorResponse, GetDatabaseResponse, GetFunctionResponse, GetTableResponse,
GetViewResponse,
ListDatabasesResponse, ListFunctionsResponse, ListTablesResponse,
ListViewsResponse,
RenameTableRequest, ResourcePaths,
};
@@ -399,6 +399,52 @@ impl RESTServer {
.into_response()
}
+ /// Handle POST /databases/:db/views - create a persistent view.
+ pub async fn create_view(
+ Path(db): Path<String>,
+ Extension(state): Extension<Arc<RESTServer>>,
+ Json(request): Json<CreateViewRequest>,
+ ) -> impl IntoResponse {
+ let mut s = state.inner.lock().unwrap();
+ let view = request.identifier.object().to_string();
+ 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();
+ }
+ if !s.databases.contains_key(&db) {
+ let err = ErrorResponse::new(
+ Some("database".to_string()),
+ Some(db.clone()),
+ Some("Not Found".to_string()),
+ Some(404),
+ );
+ return (StatusCode::NOT_FOUND, Json(err)).into_response();
+ }
+ let key = format!("{db}.{view}");
+ if s.views.contains_key(&key) {
+ let err = ErrorResponse::new(
+ Some("view".to_string()),
+ Some(view),
+ Some("Already Exists".to_string()),
+ Some(409),
+ );
+ return (StatusCode::CONFLICT, Json(err)).into_response();
+ }
+ let response = GetViewResponse::new(
+ Some(view.clone()),
+ Some(view),
+ request.schema,
+ AuditRESTResponse::new(None, None, None, None, None),
+ );
+ s.views.insert(key, response);
+ (StatusCode::OK, Json(serde_json::json!(""))).into_response()
+ }
+
/// Handle GET /databases/:db/functions/:function - get a persistent
function.
pub async fn get_function(
Path((db, function)): Path<(String, String)>,
@@ -945,7 +991,7 @@ pub async fn start_mock_server(
)
.route(
&format!("{prefix}/databases/:db/views"),
- get(RESTServer::list_views),
+ get(RESTServer::list_views).post(RESTServer::create_view),
)
.route(
&format!("{prefix}/databases/:db/views/:view"),
diff --git a/crates/paimon/tests/rest_api_test.rs
b/crates/paimon/tests/rest_api_test.rs
index 07593cd..d90e633 100644
--- a/crates/paimon/tests/rest_api_test.rs
+++ b/crates/paimon/tests/rest_api_test.rs
@@ -110,6 +110,30 @@ async fn test_get_view() {
);
}
+#[tokio::test]
+async fn test_create_view() {
+ let ctx = setup_test_server(vec!["default"]).await;
+ let schema = ViewSchema::new(
+ serde_json::from_value(json!([
+ {"id": 0, "name": "id", "type": "INT"}
+ ]))
+ .unwrap(),
+ "SELECT id FROM source".to_string(),
+ HashMap::from([(
+ "datafusion".to_string(),
+ "SELECT id FROM source".to_string(),
+ )]),
+ None,
+ HashMap::new(),
+ );
+ let identifier = Identifier::new("default", "active_ids");
+
+ ctx.api.create_view(&identifier, schema).await.unwrap();
+
+ let response = ctx.api.get_view(&identifier).await.unwrap();
+ assert_eq!(response.schema.query(), "SELECT id FROM source");
+}
+
#[tokio::test]
async fn test_list_views() {
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 778337c..722a7b2 100644
--- a/crates/paimon/tests/rest_catalog_test.rs
+++ b/crates/paimon/tests/rest_catalog_test.rs
@@ -1025,6 +1025,107 @@ async fn test_catalog_get_view() {
);
}
+#[tokio::test]
+async fn test_catalog_create_view() {
+ let ctx = setup_catalog(vec!["default"]).await;
+ let schema = ViewSchema::new(
+ serde_json::from_value(serde_json::json!([
+ {"id": 0, "name": "id", "type": "INT"}
+ ]))
+ .unwrap(),
+ "SELECT id FROM source".to_string(),
+ HashMap::from([(
+ "datafusion".to_string(),
+ "SELECT id FROM source".to_string(),
+ )]),
+ None,
+ HashMap::new(),
+ );
+ let identifier = Identifier::new("default", "active_ids");
+
+ ctx.catalog
+ .create_view(&identifier, schema, false)
+ .await
+ .unwrap();
+
+ let view = ctx.catalog.get_view(&identifier).await.unwrap();
+ assert_eq!(view.query_for("datafusion"), "SELECT id FROM source");
+}
+
+#[tokio::test]
+async fn test_catalog_create_view_missing_database() {
+ let ctx = setup_catalog(vec!["default"]).await;
+ let identifier = Identifier::new("missing_db", "active_ids");
+ let schema = ViewSchema::new(
+ Vec::new(),
+ "SELECT 1".to_string(),
+ HashMap::new(),
+ None,
+ HashMap::new(),
+ );
+
+ let error = ctx
+ .catalog
+ .create_view(&identifier, schema, false)
+ .await
+ .unwrap_err();
+
+ assert!(matches!(
+ error,
+ paimon::Error::DatabaseNotExist { database } if database ==
"missing_db"
+ ));
+}
+
+#[tokio::test]
+async fn test_catalog_create_view_already_exists() {
+ let ctx = setup_catalog(vec!["default"]).await;
+ let identifier = Identifier::new("default", "active_ids");
+ let schema = ViewSchema::new(
+ Vec::new(),
+ "SELECT 1".to_string(),
+ HashMap::new(),
+ None,
+ HashMap::new(),
+ );
+ ctx.catalog
+ .create_view(&identifier, schema.clone(), false)
+ .await
+ .unwrap();
+
+ let error = ctx
+ .catalog
+ .create_view(&identifier, schema, false)
+ .await
+ .unwrap_err();
+
+ assert!(matches!(
+ error,
+ paimon::Error::ViewAlreadyExist { full_name } if full_name ==
"default.active_ids"
+ ));
+}
+
+#[tokio::test]
+async fn test_catalog_create_view_ignore_if_exists() {
+ let ctx = setup_catalog(vec!["default"]).await;
+ let identifier = Identifier::new("default", "active_ids");
+ let schema = ViewSchema::new(
+ Vec::new(),
+ "SELECT 1".to_string(),
+ HashMap::new(),
+ None,
+ HashMap::new(),
+ );
+ ctx.catalog
+ .create_view(&identifier, schema.clone(), false)
+ .await
+ .unwrap();
+
+ ctx.catalog
+ .create_view(&identifier, schema, true)
+ .await
+ .unwrap();
+}
+
#[tokio::test]
async fn test_catalog_list_views() {
let ctx = setup_catalog(vec!["default"]).await;
@@ -1153,6 +1254,23 @@ 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
+ .create_view(
+ &Identifier::new("default", "view"),
+ ViewSchema::new(
+ Vec::new(),
+ "SELECT 1".to_string(),
+ HashMap::new(),
+ None,
+ HashMap::new(),
+ ),
+ false,
+ )
+ .await
+ .unwrap_err(),
+ paimon::Error::Unsupported { .. }
+ ));
assert!(matches!(
ctx.catalog.list_views("default").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
index 3c16bbd..6e2806a 100644
--- a/crates/paimon/tests/rest_object_models_test.rs
+++ b/crates/paimon/tests/rest_object_models_test.rs
@@ -17,10 +17,67 @@
use std::collections::HashMap;
+use paimon::api::CreateViewRequest;
use paimon::catalog::{Function, FunctionDefinition, Identifier, View,
ViewSchema};
use paimon::spec::DataField;
use serde_json::json;
+#[test]
+fn construct_view_schema_contract() {
+ let fields: Vec<DataField> = serde_json::from_value(json!([
+ {"id": 0, "name": "id", "type": "INT"}
+ ]))
+ .unwrap();
+ let schema = ViewSchema::new(
+ fields,
+ "SELECT id FROM source".to_string(),
+ HashMap::from([(
+ "datafusion".to_string(),
+ "SELECT id FROM source".to_string(),
+ )]),
+ None,
+ HashMap::new(),
+ );
+
+ assert_eq!(schema.fields()[0].name(), "id");
+ assert_eq!(schema.query_for("datafusion"), "SELECT id FROM source");
+}
+
+#[test]
+fn create_view_request_serialization_contract() {
+ let fields: Vec<DataField> = serde_json::from_value(json!([
+ {"id": 0, "name": "id", "type": "INT"}
+ ]))
+ .unwrap();
+ let request = CreateViewRequest::new(
+ Identifier::new("analytics", "active_ids"),
+ ViewSchema::new(
+ fields,
+ "SELECT id FROM source".to_string(),
+ HashMap::from([(
+ "datafusion".to_string(),
+ "SELECT id FROM source".to_string(),
+ )]),
+ None,
+ HashMap::new(),
+ ),
+ );
+
+ assert_eq!(
+ serde_json::to_value(request).unwrap(),
+ json!({
+ "identifier": {"database": "analytics", "object": "active_ids"},
+ "schema": {
+ "fields": [{"id": 0, "name": "id", "type": "INT"}],
+ "query": "SELECT id FROM source",
+ "dialects": {"datafusion": "SELECT id FROM source"},
+ "comment": null,
+ "options": {}
+ }
+ })
+ );
+}
+
#[test]
fn deserialize_view_schema_contract() {
let schema: ViewSchema = serde_json::from_value(json!({
diff --git a/docs/src/sql.md b/docs/src/sql.md
index 2d58ba6..0a6073f 100644
--- a/docs/src/sql.md
+++ b/docs/src/sql.md
@@ -40,9 +40,9 @@ Mosaic support is always available and currently read-only.
SQL queries can read
SQL support has two layers:
- DataFusion provides the parser, query planner, optimizer, execution engine,
expressions, scalar functions, aggregate functions, and window functions. SQL
statements that `SQLContext` does not intercept are delegated to DataFusion.
This includes the DataFusion SQL surface for `SELECT` queries, CTEs (including
recursive CTEs), subqueries, joins including `LATERAL` joins, SQL lambda
functions, grouping, `HAVING`, window clauses, `QUALIFY`, set operations,
`ORDER BY`, `LIMIT`/`OFFSET`, `EX [...]
-- Paimon-specific table management and row-level writes are implemented by
`SQLContext`. This includes Paimon `CREATE TABLE`, `ALTER TABLE`, `DROP TABLE`,
`CREATE TEMPORARY TABLE`, `CREATE TEMPORARY VIEW`, `DROP TEMPORARY TABLE` /
`VIEW`, `INSERT OVERWRITE ... PARTITION`, `UPDATE`, `DELETE`, `MERGE INTO`,
`TRUNCATE TABLE`, `ALTER TABLE ... DROP PARTITION`, `CALL sys.*`, Paimon time
travel, and `SET` / `RESET 'paimon.*'`.
+- Paimon-specific table management and row-level writes are implemented by
`SQLContext`. This includes Paimon `CREATE TABLE`, `ALTER TABLE`, `DROP TABLE`,
`CREATE TEMPORARY TABLE`, `CREATE TEMPORARY VIEW`, REST Catalog persistent
`CREATE VIEW`, `DROP TEMPORARY TABLE` / `VIEW`, `INSERT OVERWRITE ...
PARTITION`, `UPDATE`, `DELETE`, `MERGE INTO`, `TRUNCATE TABLE`, `ALTER TABLE
... DROP PARTITION`, `CALL sys.*`, Paimon time travel, and `SET` / `RESET
'paimon.*'`.
-Not every DataFusion DDL/DML statement maps to a Paimon table operation. For
Paimon catalogs, `CREATE EXTERNAL TABLE`, `LOCATION`, persistent `CREATE VIEW`,
`CREATE MATERIALIZED VIEW`, and persistent `CREATE TABLE AS SELECT` are
rejected or not implemented. DataFusion `COPY` can export query results to
files; it does not create or commit Paimon table files.
+Not every DataFusion DDL/DML statement maps to a Paimon table operation. For
Paimon catalogs, `CREATE EXTERNAL TABLE`, `LOCATION`, `CREATE MATERIALIZED
VIEW`, `CREATE FUNCTION`, and persistent `CREATE TABLE AS SELECT` are rejected
or not implemented. DataFusion `COPY` can export query results to files; it
does not create or commit Paimon table files.
For the exact delegated SQL grammar, see the [DataFusion SQL
Reference](https://datafusion.apache.org/user-guide/sql/index.html).
@@ -76,12 +76,41 @@ 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
+### 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.
+execute persistent views and SQL scalar functions in the catalog. It can also
+create persistent views with this syntax:
+
+```sql
+CREATE VIEW [IF NOT EXISTS] view_name [(column_name, ...)] AS query;
+```
+
+For example:
+
+```sql
+CREATE VIEW paimon.reporting.daily_orders (order_date, order_count) AS
+SELECT order_date, COUNT(*)
+FROM orders
+GROUP BY order_date;
+```
+
+The defining query is planned before the view is created. Its output types and
+nullability become the stored REST view schema, with field IDs assigned from
+zero. An optional column list changes only the output names and must contain
+exactly one unique name per query column. `IF NOT EXISTS` is passed to the
+catalog so the REST server handles concurrent creates atomically.
+
+Unqualified relations and REST SQL functions in the defining query resolve in
+the new view's owning catalog and database, not the session's current database.
+The canonical query is stored as both the default query and the `datafusion`
+dialect definition.
+
+Persistent `CREATE VIEW` is currently implemented by REST Catalog. Other
+catalog implementations may return `Unsupported`. `CREATE OR REPLACE VIEW`,
+materialized/secure views, view comments or options, vendor-specific modifiers,
+and persistent `ALTER VIEW` / `DROP VIEW` are not supported. Persistent
+`CREATE FUNCTION`, `ALTER FUNCTION`, and `DROP FUNCTION` are also not
supported.
Persistent views resolve through the normal DataFusion catalog path, so they
can be queried wherever a table can be used:
