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 5de0ba34 feat(datafusion): add REST management procedures (#837)
5de0ba34 is described below
commit 5de0ba34df50f4806430e4a19cd27dfa9ee569a7
Author: Jiajia Li <[email protected]>
AuthorDate: Tue Sep 22 14:16:51 2026 +0800
feat(datafusion): add REST management procedures (#837)
---
.github/workflows/ci.yml | 2 +
crates/integrations/datafusion/src/procedures.rs | 539 +++++++++++++-
.../datafusion/tests/rest_management_procedures.rs | 817 +++++++++++++++++++++
crates/paimon/src/catalog/filesystem.rs | 6 +
crates/paimon/src/catalog/mod.rs | 7 +
crates/paimon/src/catalog/rest/rest_catalog.rs | 3 +
docs/src/sql.md | 117 +++
7 files changed, 1488 insertions(+), 3 deletions(-)
diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml
index 5f2336a3..1b84a707 100644
--- a/.github/workflows/ci.yml
+++ b/.github/workflows/ci.yml
@@ -267,6 +267,8 @@ jobs:
--features fulltext,vortex
-- --ignored --exact
+ # The test filter selects tests, not targets, so without `--test` cargo
links all 26
+ # integration binaries with `vortex` enabled just to run one ignored
test in one of them.
- name: DataFusion Lumina Build Query E2E Test
if: matrix.suite == 'lumina'
run: >
diff --git a/crates/integrations/datafusion/src/procedures.rs
b/crates/integrations/datafusion/src/procedures.rs
index 04941451..3ea1f6c6 100644
--- a/crates/integrations/datafusion/src/procedures.rs
+++ b/crates/integrations/datafusion/src/procedures.rs
@@ -29,13 +29,27 @@
//! - `CALL sys.drop_global_index(table => '...', index_column => '...',
index_type => 'btree')` (also 'bitmap', 'multivalue', 'fm', 'lumina', or a
vindex type such as 'ivf-pq')
//! - `CALL sys.create_lumina_index(table => '...', index_column => '...')`
//!
+//! REST management procedures (REST catalogs only, mirroring Java's
+//! `RESTCatalog.permissionManagement()` / `policyManagement()`):
+//! - `CALL sys.grant_permission(resource_type => '...', access => '...',
principal => '...'[, database, table, function, view, expire_time,
column_names, excluded_column_names])`
+//! - `CALL sys.revoke_permission(resource_type => '...', access => '...',
principal => '...'[, database, table, function, view])`
+//! - `CALL sys.list_permissions(resource_type => '...'[, database, table,
function, view, principal, access, max_results, page_token])`
+//! - `CALL sys.create_policy(database => '...', table => '...', policy_type
=> '...', principal => '...'[, predicate_json, on_column, transform_json])`
+//! - `CALL sys.drop_policy(database => '...', table => '...', policy_type =>
'...', principal => '...'[, column, if_exists])`
+//! - `CALL sys.list_policies(database => '...', table => '...'[, policy_type,
principal, column, max_results, page_token])`
+//!
//! The `index_type` argument of the three global index procedures is
//! case-insensitive and surrounding whitespace is ignored.
+//!
+//! `column_names` and `excluded_column_names` are comma-separated lists, not
SQL arrays:
+//! `column_names => 'id, region'`. Java's Spark procedures take
`ARRAY<STRING>` there, but a
+//! DataFusion CALL argument is always a scalar, so this crate uses the same
comma convention as
+//! `delete_tag`'s `tag` argument.
use std::collections::HashMap;
use std::sync::Arc;
-use datafusion::arrow::array::StringArray;
+use datafusion::arrow::array::{ArrayRef, StringArray};
use datafusion::arrow::datatypes::{DataType as ArrowDataType, Field, Schema};
use datafusion::arrow::record_batch::RecordBatch;
use datafusion::error::{DataFusionError, Result as DFResult};
@@ -44,7 +58,11 @@ use datafusion::sql::sqlparser::ast::{
Expr as SqlExpr, Function, FunctionArg, FunctionArgExpr,
FunctionArgOperator,
FunctionArguments, ObjectName, Value as SqlValue,
};
-use paimon::catalog::{Catalog, Identifier};
+use paimon::api::{
+ ColumnMask, DataPolicy, ListPermissionsRequest, ListPoliciesRequest,
PermissionAssignment,
+ PermissionColumns, PermissionResource, PolicyType, ResourceType, RowFilter,
+};
+use paimon::catalog::{Catalog, Identifier, RESTCatalog};
use paimon::lumina::LUMINA_IDENTIFIER;
use paimon::spec::Snapshot;
use paimon::table::{
@@ -141,6 +159,71 @@ async fn earlier_or_equal_from_all(
}
}
+/// The parameter names Java declares for each management procedure, in Java's
order.
+fn management_parameters(proc_name: &str) -> Option<&'static [&'static str]> {
+ Some(match proc_name {
+ "grant_permission" => &[
+ "resource_type",
+ "access",
+ "principal",
+ "database",
+ "table",
+ "function",
+ "view",
+ "expire_time",
+ "column_names",
+ "excluded_column_names",
+ ],
+ "revoke_permission" => &[
+ "resource_type",
+ "access",
+ "principal",
+ "database",
+ "table",
+ "function",
+ "view",
+ ],
+ "list_permissions" => &[
+ "resource_type",
+ "database",
+ "table",
+ "function",
+ "view",
+ "principal",
+ "access",
+ "max_results",
+ "page_token",
+ ],
+ "create_policy" => &[
+ "database",
+ "table",
+ "policy_type",
+ "principal",
+ "predicate_json",
+ "on_column",
+ "transform_json",
+ ],
+ "drop_policy" => &[
+ "database",
+ "table",
+ "policy_type",
+ "principal",
+ "column",
+ "if_exists",
+ ],
+ "list_policies" => &[
+ "database",
+ "table",
+ "policy_type",
+ "principal",
+ "column",
+ "max_results",
+ "page_token",
+ ],
+ _ => return None,
+ })
+}
+
pub async fn execute_call(
ctx: &SessionContext,
catalogs: &HashMap<String, Arc<dyn Catalog>>,
@@ -154,6 +237,17 @@ pub async fn execute_call(
.ok_or_else(|| DataFusionError::Plan(format!("Unknown catalog
'{catalog_name}'")))?;
let args = extract_named_args(&func.args)?;
+ // Java rejects an argument name no parameter declares, so a typo cannot
be dropped in
+ // silence. `expiretime => ...` on a grant would otherwise send a
permanent one.
+ if let Some(declared) = management_parameters(&proc_name) {
+ if let Some(unknown) = args.keys().find(|key|
!declared.contains(&key.as_str())) {
+ return Err(DataFusionError::Plan(format!(
+ "Argument {unknown} is unknown. Expected one of [{}].",
+ declared.join(", ")
+ )));
+ }
+ }
+
match proc_name.as_str() {
"create_tag" => proc_create_tag(ctx, catalog, catalog_name,
&args).await,
"delete_tag" => proc_delete_tag(ctx, catalog, catalog_name,
&args).await,
@@ -167,6 +261,12 @@ pub async fn execute_call(
"create_global_index" => proc_create_global_index(ctx, catalog,
catalog_name, &args).await,
"drop_global_index" => proc_drop_global_index(ctx, catalog,
catalog_name, &args).await,
"create_lumina_index" => proc_create_lumina_index(ctx, catalog,
catalog_name, &args).await,
+ "grant_permission" => proc_grant_permission(ctx, catalog,
catalog_name, &args).await,
+ "revoke_permission" => proc_revoke_permission(ctx, catalog,
catalog_name, &args).await,
+ "list_permissions" => proc_list_permissions(ctx, catalog,
catalog_name, &args).await,
+ "create_policy" => proc_create_policy(ctx, catalog, catalog_name,
&args).await,
+ "drop_policy" => proc_drop_policy(ctx, catalog, catalog_name,
&args).await,
+ "list_policies" => proc_list_policies(ctx, catalog, catalog_name,
&args).await,
_ => Err(DataFusionError::Plan(format!(
"Unknown procedure: {proc_name}"
))),
@@ -210,7 +310,14 @@ fn extract_named_args(args: &FunctionArguments) ->
DFResult<HashMap<String, Stri
operator: FunctionArgOperator::RightArrow,
} => {
let value = expr_to_string(expr)?;
- map.insert(name.value.to_lowercase(), value);
+ let name = name.value.to_lowercase();
+ // Java `PaimonProcedureResolver.buildNameToArgumentMap`
rejects a repeat
+ // rather than letting the last one win.
+ if map.insert(name.clone(), value).is_some() {
+ return Err(DataFusionError::Plan(format!(
+ "Procedure argument {name} is duplicated."
+ )));
+ }
}
_ => return Err(DataFusionError::Plan(
"CALL procedures require named arguments with '=>' syntax,
e.g. table => 'db.t'"
@@ -656,6 +763,432 @@ fn parse_key_value_options(options: &str) ->
DFResult<HashMap<String, String>> {
Ok(parsed)
}
+// ==================== REST management procedures (Java #9410)
====================
+//
+// Parameter names follow Java's; divergences are called out where they occur.
+
+/// The REST catalog behind `catalog`, which is where permission and policy
management lives.
+/// Java does the same check with `DelegateCatalog.rootCatalog(...) instanceof
RESTCatalog`.
+fn rest_catalog<'a>(
+ catalog: &'a Arc<dyn Catalog>,
+ catalog_name: &str,
+) -> DFResult<&'a RESTCatalog> {
+ catalog
+ .as_any()
+ .and_then(|any| any.downcast_ref::<RESTCatalog>())
+ .ok_or_else(|| {
+ DataFusionError::Plan(format!(
+ "Catalog '{catalog_name}' does not support permission or
policy management."
+ ))
+ })
+}
+
+/// Java's blankness, which is `String.trim()`: only `<= U+0020` counts.
`str::trim` would also
+/// strip the rest of Unicode whitespace and disagree on a non-breaking space.
(`listagg.rs` has
+/// a third variant using `Character.isWhitespace`; none of the three are
interchangeable.)
+fn is_blank(value: &str) -> bool {
+ value.chars().all(|ch| ch <= ' ')
+}
+
+/// Java `emptyToNull`: a blank value is absent.
+fn opt_arg<'a>(args: &'a HashMap<String, String>, name: &str) -> Option<&'a
str> {
+ args.get(name)
+ .map(String::as_str)
+ .filter(|value| !is_blank(value))
+}
+
+/// Java `BasePermissionProcedure.enumValue`.
+/// The values Java prints in `Invalid <arg> '<value>'. Expected one of
<values>.`
+trait ProcedureEnum: std::str::FromStr<Err = paimon::Error> + Sized {
+ fn allowed() -> Vec<&'static str>;
+}
+
+impl ProcedureEnum for ResourceType {
+ fn allowed() -> Vec<&'static str> {
+ ResourceType::VALUES
+ .iter()
+ .map(ResourceType::as_str)
+ .collect()
+ }
+}
+
+impl ProcedureEnum for PolicyType {
+ fn allowed() -> Vec<&'static str> {
+ vec![
+ PolicyType::RowFilter.as_str(),
+ PolicyType::ColumnMasking.as_str(),
+ ]
+ }
+}
+
+fn enum_arg<T>(args: &HashMap<String, String>, name: &str) -> DFResult<T>
+where
+ T: ProcedureEnum,
+{
+ let value = require_arg(args, name)?;
+ // Java neither trims nor accepts blank, so ' TABLE ' must stay an error
here too.
+ if is_blank(value) {
+ return Err(DataFusionError::Plan(format!("{name} cannot be empty.")));
+ }
+ value.parse().map_err(|_| {
+ DataFusionError::Plan(format!(
+ "Invalid {name} '{value}'. Expected one of [{}].",
+ T::allowed().join(", ")
+ ))
+ })
+}
+
+/// Same as [`enum_arg`], but the argument may be absent or blank.
+fn opt_enum_arg<T>(args: &HashMap<String, String>, name: &str) ->
DFResult<Option<T>>
+where
+ T: ProcedureEnum,
+{
+ match opt_arg(args, name) {
+ None => Ok(None),
+ Some(_) => enum_arg(args, name).map(Some),
+ }
+}
+
+/// Comma-separated, like `delete_tag`'s `tag`. Java declares `ARRAY<STRING>`,
but a
+/// DataFusion CALL argument is always a scalar.
+fn comma_list(args: &HashMap<String, String>, name: &str) ->
Option<Vec<String>> {
+ args.get(name).map(|raw| {
+ raw.split(',')
+ // Java's trim, as everywhere else here: a non-breaking space is
part of the name.
+ .map(|value| value.trim_matches(|ch| ch <= ' '))
+ .filter(|value| !value.is_empty())
+ .map(str::to_string)
+ .collect()
+ })
+}
+
+fn bool_arg(args: &HashMap<String, String>, name: &str) -> DFResult<bool> {
+ // Java's `ProcedureParameter.optional(..., BooleanType)` leaves a missing
value as false.
+ match args.get(name) {
+ None => Ok(false),
+ Some(value) => match value.trim().to_ascii_lowercase().as_str() {
+ "true" => Ok(true),
+ "false" => Ok(false),
+ _ => Err(DataFusionError::Plan(format!(
+ "Invalid {name} '{value}'. Expected 'true' or 'false'"
+ ))),
+ },
+ }
+}
+
+fn max_results_arg(args: &HashMap<String, String>) -> DFResult<Option<u32>> {
+ args.get("max_results")
+ .map(|value| {
+ value
+ .trim()
+ .parse()
+ .map_err(|_| DataFusionError::Plan(format!("Invalid
max_results: '{value}'")))
+ })
+ .transpose()
+}
+
+/// `resource_type` plus whichever locators it needs (Java
`BasePermissionProcedure.resource`).
+fn permission_resource(args: &HashMap<String, String>) ->
DFResult<PermissionResource> {
+ PermissionResource::new(
+ enum_arg::<ResourceType>(args, "resource_type")?,
+ opt_arg(args, "database"),
+ opt_arg(args, "table"),
+ opt_arg(args, "function"),
+ opt_arg(args, "view"),
+ )
+ .map_err(to_datafusion_error)
+}
+
+/// The `TABLE` resource a policy hangs off (Java
`BasePolicyProcedure.tableResource`).
+fn policy_table_resource(args: &HashMap<String, String>) ->
DFResult<PermissionResource> {
+ PermissionResource::new(
+ ResourceType::Table,
+ Some(require_arg(args, "database")?),
+ Some(require_arg(args, "table")?),
+ None,
+ None,
+ )
+ .map_err(to_datafusion_error)
+}
+
+/// Java hands both lists to `PermissionColumns` and lets it reject having
both; each Rust
+/// constructor takes one list, so that case is rejected here instead.
+fn permission_columns(args: &HashMap<String, String>) ->
DFResult<Option<PermissionColumns>> {
+ match (
+ comma_list(args, "column_names"),
+ comma_list(args, "excluded_column_names"),
+ ) {
+ (None, None) => Ok(None),
+ (Some(names), None) => PermissionColumns::names(names)
+ .map(Some)
+ .map_err(to_datafusion_error),
+ (None, Some(excluded)) => PermissionColumns::excluded(excluded)
+ .map(Some)
+ .map_err(to_datafusion_error),
+ (Some(_), Some(_)) => Err(DataFusionError::Plan(
+ "columns must contain exactly one of column_names or
excluded_column_names."
+ .to_string(),
+ )),
+ }
+}
+
+fn utf8_result(
+ ctx: &SessionContext,
+ fields: &[(&str, bool)],
+ rows: Vec<Vec<Option<String>>>,
+) -> DFResult<DataFrame> {
+ debug_assert!(
+ rows.iter().all(|row| row.len() == fields.len()),
+ "every row must have one cell per declared column"
+ );
+ let schema = Arc::new(Schema::new(
+ fields
+ .iter()
+ .map(|(name, nullable)| Field::new(*name, ArrowDataType::Utf8,
*nullable))
+ .collect::<Vec<_>>(),
+ ));
+ let columns = (0..fields.len())
+ .map(|column| {
+ Arc::new(
+ rows.iter()
+ .map(|row| row[column].clone())
+ .collect::<StringArray>(),
+ ) as ArrayRef
+ })
+ .collect::<Vec<_>>();
+ ctx.read_batch(RecordBatch::try_new(schema, columns)?)
+}
+
+fn joined(values: Option<&[String]>) -> Option<String> {
+ values.map(|values| values.join(","))
+}
+
+async fn proc_grant_permission(
+ ctx: &SessionContext,
+ catalog: &Arc<dyn Catalog>,
+ catalog_name: &str,
+ args: &HashMap<String, String>,
+) -> DFResult<DataFrame> {
+ let rest = rest_catalog(catalog, catalog_name)?;
+ let assignment = PermissionAssignment::new(
+ permission_resource(args)?,
+ require_arg(args, "access")?,
+ require_arg(args, "principal")?,
+ permission_columns(args)?,
+ opt_arg(args, "expire_time"),
+ )
+ .map_err(to_datafusion_error)?;
+ rest.grant_permission(&assignment)
+ .await
+ .map_err(to_datafusion_error)?;
+ // Java returns boolean `true`; every write procedure here answers
`ok_result` instead.
+ ok_result(ctx)
+}
+
+async fn proc_revoke_permission(
+ ctx: &SessionContext,
+ catalog: &Arc<dyn Catalog>,
+ catalog_name: &str,
+ args: &HashMap<String, String>,
+) -> DFResult<DataFrame> {
+ let rest = rest_catalog(catalog, catalog_name)?;
+ rest.revoke_permission(
+ &permission_resource(args)?,
+ require_arg(args, "access")?,
+ require_arg(args, "principal")?,
+ )
+ .await
+ .map_err(to_datafusion_error)?;
+ ok_result(ctx)
+}
+
+async fn proc_list_permissions(
+ ctx: &SessionContext,
+ catalog: &Arc<dyn Catalog>,
+ catalog_name: &str,
+ args: &HashMap<String, String>,
+) -> DFResult<DataFrame> {
+ let rest = rest_catalog(catalog, catalog_name)?;
+ let request = ListPermissionsRequest {
+ resource: permission_resource(args)?,
+ principal: opt_arg(args, "principal").map(str::to_string),
+ access: opt_arg(args, "access").map(str::to_string),
+ max_results: max_results_arg(args)?,
+ page_token: opt_arg(args, "page_token").map(str::to_string),
+ };
+ let page = rest
+ .list_permissions_paged(&request)
+ .await
+ .map_err(to_datafusion_error)?;
+
+ // Java parity, quirk included: `next_page_token` repeats on every row, so
an empty page
+ // returns zero rows and loses it. See `ListPermissionsProcedure.call`.
+ let rows = page
+ .elements
+ .iter()
+ .map(|assignment| {
+ let resource = assignment.resource();
+ let columns = assignment.columns();
+ vec![
+ Some(resource.resource_type().to_string()),
+ resource.database_name().map(str::to_string),
+ resource.table_name().map(str::to_string),
+ resource.function_name().map(str::to_string),
+ resource.view_name().map(str::to_string),
+ Some(assignment.access().to_string()),
+ Some(assignment.principal().to_string()),
+ // Comma-joined, matching how `column_names` is passed in.
+ joined(columns.and_then(PermissionColumns::column_names)),
+
joined(columns.and_then(PermissionColumns::excluded_column_names)),
+ assignment.expire_time().map(str::to_string),
+ page.next_page_token.clone(),
+ ]
+ })
+ .collect();
+
+ utf8_result(
+ ctx,
+ &[
+ ("resource_type", false),
+ ("database", true),
+ ("table", true),
+ ("function", true),
+ ("view", true),
+ ("access", false),
+ ("principal", false),
+ ("column_names", true),
+ ("excluded_column_names", true),
+ ("expire_time", true),
+ ("next_page_token", true),
+ ],
+ rows,
+ )
+}
+
+async fn proc_create_policy(
+ ctx: &SessionContext,
+ catalog: &Arc<dyn Catalog>,
+ catalog_name: &str,
+ args: &HashMap<String, String>,
+) -> DFResult<DataFrame> {
+ let rest = rest_catalog(catalog, catalog_name)?;
+ let resource = policy_table_resource(args)?;
+ let principal = require_arg(args, "principal")?;
+ let predicate = opt_arg(args, "predicate_json");
+ let on_column = opt_arg(args, "on_column");
+ let transform = opt_arg(args, "transform_json");
+
+ // Java `BasePolicyProcedure.policy`: each policy type rejects the other's
fields.
+ let policy = match enum_arg::<PolicyType>(args, "policy_type")? {
+ PolicyType::RowFilter => {
+ for (value, name) in [(on_column, "on_column"), (transform,
"transform_json")] {
+ if value.is_some() {
+ return Err(DataFusionError::Plan(format!(
+ "ROW_FILTER policy cannot specify {name}."
+ )));
+ }
+ }
+ let row_filter =
+
RowFilter::new(predicate.unwrap_or_default()).map_err(to_datafusion_error)?;
+ DataPolicy::new_row_filter(resource, row_filter, principal)
+ }
+ PolicyType::ColumnMasking => {
+ if predicate.is_some() {
+ return Err(DataFusionError::Plan(
+ "COLUMN_MASKING policy cannot specify
predicate_json.".to_string(),
+ ));
+ }
+ let column_mask =
+ ColumnMask::new(on_column.unwrap_or_default(),
transform.unwrap_or_default())
+ .map_err(to_datafusion_error)?;
+ DataPolicy::new_column_mask(resource, column_mask, principal)
+ }
+ }
+ .map_err(to_datafusion_error)?;
+
+ rest.create_policy(&policy)
+ .await
+ .map_err(to_datafusion_error)?;
+ ok_result(ctx)
+}
+
+async fn proc_drop_policy(
+ ctx: &SessionContext,
+ catalog: &Arc<dyn Catalog>,
+ catalog_name: &str,
+ args: &HashMap<String, String>,
+) -> DFResult<DataFrame> {
+ let rest = rest_catalog(catalog, catalog_name)?;
+ rest.drop_policy(
+ &policy_table_resource(args)?,
+ enum_arg::<PolicyType>(args, "policy_type")?,
+ require_arg(args, "principal")?,
+ opt_arg(args, "column"),
+ bool_arg(args, "if_exists")?,
+ )
+ .await
+ .map_err(to_datafusion_error)?;
+ ok_result(ctx)
+}
+
+async fn proc_list_policies(
+ ctx: &SessionContext,
+ catalog: &Arc<dyn Catalog>,
+ catalog_name: &str,
+ args: &HashMap<String, String>,
+) -> DFResult<DataFrame> {
+ let rest = rest_catalog(catalog, catalog_name)?;
+ let request = ListPoliciesRequest {
+ resource: policy_table_resource(args)?,
+ policy_type: opt_enum_arg(args, "policy_type")?,
+ principal: opt_arg(args, "principal").map(str::to_string),
+ column: opt_arg(args, "column").map(str::to_string),
+ max_results: max_results_arg(args)?,
+ page_token: opt_arg(args, "page_token").map(str::to_string),
+ };
+ let page = rest
+ .list_policies_paged(&request)
+ .await
+ .map_err(to_datafusion_error)?;
+
+ // Same Java pagination quirk as `list_permissions`.
+ let rows = page
+ .elements
+ .iter()
+ .map(|policy| {
+ let resource = policy.resource();
+ let column_mask = policy.column_mask();
+ vec![
+ resource.database_name().map(str::to_string),
+ resource.table_name().map(str::to_string),
+ Some(policy.policy_type().to_string()),
+ Some(policy.principal().to_string()),
+ policy
+ .row_filter()
+ .map(|filter| filter.predicate().to_string()),
+ column_mask.map(|mask| mask.on_column().to_string()),
+ column_mask.map(|mask| mask.transform().to_string()),
+ page.next_page_token.clone(),
+ ]
+ })
+ .collect();
+
+ utf8_result(
+ ctx,
+ &[
+ ("database", false),
+ ("table", false),
+ ("policy_type", false),
+ ("principal", false),
+ ("predicate_json", true),
+ ("on_column", true),
+ ("transform_json", true),
+ ("next_page_token", true),
+ ],
+ rows,
+ )
+}
+
fn ok_result(ctx: &SessionContext) -> DFResult<DataFrame> {
let schema = Arc::new(Schema::new(vec![Field::new(
"result",
diff --git a/crates/integrations/datafusion/tests/rest_management_procedures.rs
b/crates/integrations/datafusion/tests/rest_management_procedures.rs
new file mode 100644
index 00000000..a26a405b
--- /dev/null
+++ b/crates/integrations/datafusion/tests/rest_management_procedures.rs
@@ -0,0 +1,817 @@
+// 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.
+
+mod common;
+
+#[path = "../../../paimon/tests/mock_server.rs"]
+mod mock_server;
+
+use std::collections::HashMap;
+use std::sync::Arc;
+
+use arrow_array::{Array, RecordBatch};
+use paimon::api::ConfigResponse;
+use paimon::catalog::RESTCatalog;
+use paimon::{CatalogOptions, Options};
+use paimon_datafusion::SQLContext;
+
+use mock_server::{start_mock_server, RESTServer};
+
+const DATABASE: &str = "sales";
+const TABLE: &str = "orders";
+const WAREHOUSE: &str = "test_warehouse";
+
+async fn setup() -> (tempfile::TempDir, RESTServer, SQLContext) {
+ let temp_dir = tempfile::tempdir().unwrap();
+ let server = start_mock_server(
+ WAREHOUSE.to_string(),
+ temp_dir.path().to_string_lossy().into_owned(),
+ ConfigResponse::new(HashMap::from([(
+ CatalogOptions::PREFIX.to_string(),
+ "mock-test".to_string(),
+ )])),
+ vec![DATABASE.to_string()],
+ )
+ .await;
+ server.add_table(DATABASE, TABLE);
+
+ let mut options = Options::new();
+ options.set(CatalogOptions::URI, server.url().unwrap());
+ options.set(CatalogOptions::WAREHOUSE, WAREHOUSE);
+ options.set(CatalogOptions::TOKEN_PROVIDER, "bear");
+ options.set(CatalogOptions::TOKEN, "test-token");
+ let catalog = Arc::new(RESTCatalog::new(options, true).await.unwrap());
+ let mut context = SQLContext::new();
+ context.register_catalog("paimon", catalog).await.unwrap();
+ (temp_dir, server, context)
+}
+
+async fn call(context: &SQLContext, sql: &str) -> Vec<RecordBatch> {
+ context
+ .sql(sql)
+ .await
+ .unwrap_or_else(|e| panic!("planning '{sql}' failed: {e}"))
+ .collect()
+ .await
+ .unwrap_or_else(|e| panic!("'{sql}' failed: {e}"))
+}
+
+async fn columns_of(context: &SQLContext, sql: &str) -> Vec<String> {
+ context
+ .sql(sql)
+ .await
+ .unwrap()
+ .schema()
+ .fields()
+ .iter()
+ .map(|field| field.name().clone())
+ .collect()
+}
+
+fn rows(batches: &[RecordBatch]) -> Vec<Vec<Option<String>>> {
+ batches
+ .iter()
+ .flat_map(|batch| {
+ (0..batch.num_rows()).map(move |row| {
+ (0..batch.num_columns())
+ .map(|column| {
+ let array = batch.column(column);
+ array
+ .is_valid(row)
+ .then(|| common::string_value(array.as_ref(),
row).to_string())
+ })
+ .collect()
+ })
+ })
+ .collect()
+}
+
+fn cell(rows: &[Vec<Option<String>>], row: usize, column: usize) ->
Option<&str> {
+ rows[row][column].as_deref()
+}
+
+async fn assert_ok_row(context: &SQLContext, sql: &str) {
+ let batches = call(context, sql).await;
+ assert_eq!(rows(&batches), vec![vec![Some("OK".to_string())]], "{sql}");
+}
+
+async fn grant_table_select(context: &SQLContext, principal: &str) {
+ assert_ok_row(
+ context,
+ &format!(
+ "CALL sys.grant_permission(resource_type => 'TABLE', access =>
'SELECT', \
+ principal => '{principal}', database => '{DATABASE}', table =>
'{TABLE}')"
+ ),
+ )
+ .await;
+}
+
+async fn create_row_filter(context: &SQLContext, principal: &str, predicate:
&str) {
+ assert_ok_row(
+ context,
+ &format!(
+ "CALL sys.create_policy(database => '{DATABASE}', table =>
'{TABLE}', \
+ policy_type => 'ROW_FILTER', principal => '{principal}', \
+ predicate_json => '{predicate}')"
+ ),
+ )
+ .await;
+}
+
+#[tokio::test]
+async fn test_grant_permission_sends_assignment() {
+ let (_tmp, server, context) = setup().await;
+
+ assert_ok_row(
+ &context,
+ &format!(
+ "CALL sys.grant_permission(resource_type => 'TABLE', access =>
'select', \
+ principal => 'role:analyst', database => '{DATABASE}', table =>
'{TABLE}', \
+ expire_time => '2028-01-01T00:00:00Z')"
+ ),
+ )
+ .await;
+
+ let bodies = server.grant_permission_bodies();
+ assert_eq!(bodies.len(), 1);
+ assert_eq!(
+ bodies[0],
+ serde_json::json!({
+ "resource": {"type": "TABLE", "database": "sales", "table":
"orders"},
+ "access": "SELECT",
+ "principal": "role:analyst",
+ "expireTime": "2028-01-01T00:00:00Z",
+ })
+ );
+ assert_eq!(server.permissions().len(), 1);
+}
+
+#[tokio::test]
+async fn test_revoke_permission_sends_identity() {
+ let (_tmp, server, context) = setup().await;
+ grant_table_select(&context, "role:analyst").await;
+
+ assert_ok_row(
+ &context,
+ &format!(
+ "CALL sys.revoke_permission(resource_type => 'TABLE', access =>
'SELECT', \
+ principal => 'role:analyst', database => '{DATABASE}', table =>
'{TABLE}')"
+ ),
+ )
+ .await;
+
+ let bodies = server.revoke_permission_bodies();
+ assert_eq!(bodies.len(), 1);
+ assert_eq!(
+ bodies[0],
+ serde_json::json!({
+ "resource": {"type": "TABLE", "database": "sales", "table":
"orders"},
+ "access": "SELECT",
+ "principal": "role:analyst",
+ })
+ );
+ assert!(server.permissions().is_empty());
+}
+
+#[tokio::test]
+async fn test_list_permissions_columns_and_rows() {
+ let (_tmp, server, context) = setup().await;
+ assert_ok_row(
+ &context,
+ &format!(
+ "CALL sys.grant_permission(resource_type => 'TABLE', access =>
'SELECT', \
+ principal => 'role:analyst', database => '{DATABASE}', table =>
'{TABLE}', \
+ expire_time => '2028-01-01T00:00:00Z')"
+ ),
+ )
+ .await;
+
+ let sql = format!(
+ "CALL sys.list_permissions(resource_type => 'TABLE', database =>
'{DATABASE}', \
+ table => '{TABLE}', principal => 'role:analyst', access => 'SELECT')"
+ );
+ assert_eq!(
+ columns_of(&context, &sql).await,
+ vec![
+ "resource_type",
+ "database",
+ "table",
+ "function",
+ "view",
+ "access",
+ "principal",
+ "column_names",
+ "excluded_column_names",
+ "expire_time",
+ "next_page_token",
+ ]
+ );
+
+ let listed = rows(&call(&context, &sql).await);
+ assert_eq!(
+ listed,
+ vec![vec![
+ Some("TABLE".to_string()),
+ Some("sales".to_string()),
+ Some("orders".to_string()),
+ None,
+ None,
+ Some("SELECT".to_string()),
+ Some("role:analyst".to_string()),
+ None,
+ None,
+ Some("2028-01-01T00:00:00Z".to_string()),
+ None,
+ ]]
+ );
+
+ let query = server.list_permissions_queries().pop().unwrap();
+ assert_eq!(query.get("resourceType").unwrap(), "TABLE");
+ assert_eq!(query.get("database").unwrap(), "sales");
+ assert_eq!(query.get("table").unwrap(), "orders");
+ assert_eq!(query.get("principal").unwrap(), "role:analyst");
+ assert_eq!(query.get("access").unwrap(), "SELECT");
+}
+
+#[tokio::test]
+async fn test_create_policy_row_filter_and_column_mask() {
+ let (_tmp, server, context) = setup().await;
+
+ create_row_filter(
+ &context,
+ "role:analyst",
+ r#"{"kind":"LEAF","name":"region"}"#,
+ )
+ .await;
+ assert_ok_row(
+ &context,
+ &format!(
+ "CALL sys.create_policy(database => '{DATABASE}', table =>
'{TABLE}', \
+ policy_type => 'COLUMN_MASKING', principal => 'role:support', \
+ on_column => 'email', transform_json =>
'{{\"name\":\"CONCAT\"}}')"
+ ),
+ )
+ .await;
+
+ let bodies = server.create_policy_bodies();
+ assert_eq!(bodies.len(), 2);
+ assert_eq!(
+ bodies[0],
+ serde_json::json!({
+ "rowFilter": {"predicate": r#"{"kind":"LEAF","name":"region"}"#},
+ "principal": "role:analyst",
+ })
+ );
+ assert_eq!(
+ bodies[1],
+ serde_json::json!({
+ "columnMask": {"onColumn": "email", "transform":
r#"{"name":"CONCAT"}"#},
+ "principal": "role:support",
+ })
+ );
+ assert_eq!(server.table_policies(DATABASE, TABLE).len(), 2);
+}
+
+#[tokio::test]
+async fn test_drop_policy_sends_identity() {
+ let (_tmp, server, context) = setup().await;
+ create_row_filter(&context, "role:analyst", r#"{"kind":"LEAF"}"#).await;
+
+ assert_ok_row(
+ &context,
+ &format!(
+ "CALL sys.drop_policy(database => '{DATABASE}', table =>
'{TABLE}', \
+ policy_type => 'ROW_FILTER', principal => 'role:analyst')"
+ ),
+ )
+ .await;
+
+ assert_eq!(
+ server.drop_policy_bodies(),
+ vec![serde_json::json!({"type": "ROW_FILTER", "principal":
"role:analyst"})]
+ );
+ assert!(server.table_policies(DATABASE, TABLE).is_empty());
+}
+
+#[tokio::test]
+async fn test_list_policies_columns_and_rows() {
+ let (_tmp, server, context) = setup().await;
+ create_row_filter(&context, "role:analyst", r#"{"kind":"LEAF"}"#).await;
+ assert_ok_row(
+ &context,
+ &format!(
+ "CALL sys.create_policy(database => '{DATABASE}', table =>
'{TABLE}', \
+ policy_type => 'COLUMN_MASKING', principal => 'role:support', \
+ on_column => 'email', transform_json =>
'{{\"name\":\"CONCAT\"}}')"
+ ),
+ )
+ .await;
+
+ let sql = format!("CALL sys.list_policies(database => '{DATABASE}', table
=> '{TABLE}')");
+ assert_eq!(
+ columns_of(&context, &sql).await,
+ vec![
+ "database",
+ "table",
+ "policy_type",
+ "principal",
+ "predicate_json",
+ "on_column",
+ "transform_json",
+ "next_page_token",
+ ]
+ );
+
+ let listed = rows(&call(&context, &sql).await);
+ assert_eq!(listed.len(), 2);
+ assert_eq!(
+ listed[0],
+ vec![
+ Some("sales".to_string()),
+ Some("orders".to_string()),
+ Some("ROW_FILTER".to_string()),
+ Some("role:analyst".to_string()),
+ Some(r#"{"kind":"LEAF"}"#.to_string()),
+ None,
+ None,
+ None,
+ ]
+ );
+ assert_eq!(
+ listed[1],
+ vec![
+ Some("sales".to_string()),
+ Some("orders".to_string()),
+ Some("COLUMN_MASKING".to_string()),
+ Some("role:support".to_string()),
+ None,
+ Some("email".to_string()),
+ Some(r#"{"name":"CONCAT"}"#.to_string()),
+ None,
+ ]
+ );
+
+ assert_eq!(
+ server.list_policies_queries().pop().unwrap(),
+ HashMap::new()
+ );
+}
+
+#[tokio::test]
+async fn test_list_policies_filters() {
+ let (_tmp, server, context) = setup().await;
+ create_row_filter(&context, "role:analyst", r#"{"kind":"LEAF"}"#).await;
+ assert_ok_row(
+ &context,
+ &format!(
+ "CALL sys.create_policy(database => '{DATABASE}', table =>
'{TABLE}', \
+ policy_type => 'COLUMN_MASKING', principal => 'role:support', \
+ on_column => 'email', transform_json =>
'{{\"name\":\"CONCAT\"}}')"
+ ),
+ )
+ .await;
+
+ let listed = rows(
+ &call(
+ &context,
+ &format!(
+ "CALL sys.list_policies(database => '{DATABASE}', table =>
'{TABLE}', \
+ policy_type => 'COLUMN_MASKING', principal => 'role:support',
column => 'email')"
+ ),
+ )
+ .await,
+ );
+ assert_eq!(listed.len(), 1);
+ assert_eq!(cell(&listed, 0, 3), Some("role:support"));
+
+ let query = server.list_policies_queries().pop().unwrap();
+ assert_eq!(query.get("type").unwrap(), "COLUMN_MASKING");
+ assert_eq!(query.get("principal").unwrap(), "role:support");
+ assert_eq!(query.get("column").unwrap(), "email");
+}
+
+#[tokio::test]
+async fn test_list_permissions_pagination_repeats_token_on_every_row() {
+ let (_tmp, server, context) = setup().await;
+ for principal in ["role:a", "role:b", "role:c"] {
+ grant_table_select(&context, principal).await;
+ }
+
+ let first = rows(
+ &call(
+ &context,
+ &format!(
+ "CALL sys.list_permissions(resource_type => 'TABLE', database
=> '{DATABASE}', \
+ table => '{TABLE}', max_results => 2)"
+ ),
+ )
+ .await,
+ );
+ assert_eq!(first.len(), 2);
+ let token = cell(&first, 0, 10).expect("next_page_token").to_string();
+ assert_eq!(cell(&first, 1, 10), Some(token.as_str()));
+ assert_eq!(cell(&first, 0, 6), Some("role:a"));
+ assert_eq!(cell(&first, 1, 6), Some("role:b"));
+
+ let second = rows(
+ &call(
+ &context,
+ &format!(
+ "CALL sys.list_permissions(resource_type => 'TABLE', database
=> '{DATABASE}', \
+ table => '{TABLE}', max_results => 2, page_token => '{token}')"
+ ),
+ )
+ .await,
+ );
+ assert_eq!(second.len(), 1);
+ assert_eq!(cell(&second, 0, 6), Some("role:c"));
+ assert_eq!(cell(&second, 0, 10), None);
+
+ let query = server.list_permissions_queries().pop().unwrap();
+ assert_eq!(query.get("maxResults").unwrap(), "2");
+ assert_eq!(query.get("pageToken").unwrap(), &token);
+
+ let past_end = call(
+ &context,
+ &format!(
+ "CALL sys.list_permissions(resource_type => 'TABLE', database =>
'{DATABASE}', \
+ table => '{TABLE}', max_results => 2, page_token => '3')"
+ ),
+ )
+ .await;
+ assert_eq!(rows(&past_end), Vec::<Vec<Option<String>>>::new());
+}
+
+#[tokio::test]
+async fn test_list_policies_pagination_repeats_token_on_every_row() {
+ let (_tmp, _server, context) = setup().await;
+ for principal in ["role:a", "role:b", "role:c"] {
+ create_row_filter(&context, principal, r#"{"kind":"LEAF"}"#).await;
+ }
+
+ let first = rows(
+ &call(
+ &context,
+ &format!(
+ "CALL sys.list_policies(database => '{DATABASE}', table =>
'{TABLE}', max_results => 2)"
+ ),
+ )
+ .await,
+ );
+ assert_eq!(first.len(), 2);
+ let token = cell(&first, 0, 7).expect("next_page_token").to_string();
+ assert_eq!(cell(&first, 1, 7), Some(token.as_str()));
+
+ let second = rows(
+ &call(
+ &context,
+ &format!(
+ "CALL sys.list_policies(database => '{DATABASE}', table =>
'{TABLE}', \
+ max_results => 2, page_token => '{token}')"
+ ),
+ )
+ .await,
+ );
+ assert_eq!(second.len(), 1);
+ assert_eq!(cell(&second, 0, 3), Some("role:c"));
+ assert_eq!(cell(&second, 0, 7), None);
+}
+
+#[tokio::test]
+async fn test_column_lists_are_comma_separated_and_java_trimmed() {
+ let (_tmp, server, context) = setup().await;
+
+ let empty = call(
+ &context,
+ &format!("CALL sys.list_policies(database => '{DATABASE}', table =>
'{TABLE}')"),
+ )
+ .await;
+ assert_eq!(rows(&empty), Vec::<Vec<Option<String>>>::new());
+ let list = format!(
+ "CALL sys.list_permissions(resource_type => 'COLUMN', database =>
'{DATABASE}', \
+ table => '{TABLE}')"
+ );
+
+ assert_ok_row(
+ &context,
+ &format!(
+ "CALL sys.grant_permission(resource_type => 'COLUMN', access =>
'SELECT', \
+ principal => 'role:analyst', database => '{DATABASE}', table =>
'{TABLE}', \
+ column_names => ' id , region ,\u{a0}')"
+ ),
+ )
+ .await;
+ assert_eq!(
+ server.grant_permission_bodies()[0]["columns"],
+ serde_json::json!({"columnNames": ["id", "region", "\u{a0}"]})
+ );
+ let listed = rows(&call(&context, &list).await);
+ assert_eq!(cell(&listed, 0, 7), Some("id,region,\u{a0}"));
+ assert_eq!(cell(&listed, 0, 8), None);
+
+ assert_ok_row(
+ &context,
+ &format!(
+ "CALL sys.grant_permission(resource_type => 'COLUMN', access =>
'SELECT', \
+ principal => 'role:analyst', database => '{DATABASE}', table =>
'{TABLE}', \
+ excluded_column_names => 'email,\tphone ')"
+ ),
+ )
+ .await;
+ assert_eq!(
+ server.grant_permission_bodies()[1]["columns"],
+ serde_json::json!({"excludedColumnNames": ["email", "phone"]})
+ );
+ let listed = rows(&call(&context, &list).await);
+ assert_eq!(listed.len(), 1);
+ assert_eq!(cell(&listed, 0, 7), None);
+ assert_eq!(cell(&listed, 0, 8), Some("email,phone"));
+}
+
+#[tokio::test]
+async fn test_both_column_lists_rejected() {
+ let (_tmp, _server, context) = setup().await;
+ common::assert_sql_error(
+ &context,
+ &format!(
+ "CALL sys.grant_permission(resource_type => 'COLUMN', access =>
'SELECT', \
+ principal => 'role:analyst', database => '{DATABASE}', table =>
'{TABLE}', \
+ column_names => 'id', excluded_column_names => 'email')"
+ ),
+ "exactly one of column_names or excluded_column_names",
+ )
+ .await;
+}
+
+#[tokio::test]
+async fn test_drop_policy_if_exists_swallows_missing_policy() {
+ let (_tmp, _server, context) = setup().await;
+
+ assert_ok_row(
+ &context,
+ &format!(
+ "CALL sys.drop_policy(database => '{DATABASE}', table =>
'{TABLE}', \
+ policy_type => 'ROW_FILTER', principal => 'role:ghost', if_exists
=> true)"
+ ),
+ )
+ .await;
+
+ common::assert_sql_error(
+ &context,
+ &format!(
+ "CALL sys.drop_policy(database => '{DATABASE}', table =>
'{TABLE}', \
+ policy_type => 'ROW_FILTER', principal => 'role:ghost', if_exists
=> false)"
+ ),
+ "Policy does not exist",
+ )
+ .await;
+
+ common::assert_sql_error(
+ &context,
+ &format!(
+ "CALL sys.drop_policy(database => '{DATABASE}', table =>
'{TABLE}', \
+ policy_type => 'ROW_FILTER', principal => 'role:ghost')"
+ ),
+ "Policy does not exist",
+ )
+ .await;
+
+ common::assert_sql_error(
+ &context,
+ &format!(
+ "CALL sys.drop_policy(database => '{DATABASE}', table =>
'{TABLE}', \
+ policy_type => 'ROW_FILTER', principal => 'role:ghost', if_exists
=> 'yes')"
+ ),
+ "Invalid if_exists 'yes'",
+ )
+ .await;
+}
+
+#[tokio::test]
+async fn test_policy_type_rejects_the_other_type_fields() {
+ let (_tmp, _server, context) = setup().await;
+
+ common::assert_sql_error(
+ &context,
+ &format!(
+ "CALL sys.create_policy(database => '{DATABASE}', table =>
'{TABLE}', \
+ policy_type => 'ROW_FILTER', principal => 'p', predicate_json =>
'{{}}', \
+ on_column => 'email')"
+ ),
+ "ROW_FILTER policy cannot specify on_column.",
+ )
+ .await;
+
+ common::assert_sql_error(
+ &context,
+ &format!(
+ "CALL sys.create_policy(database => '{DATABASE}', table =>
'{TABLE}', \
+ policy_type => 'COLUMN_MASKING', principal => 'p', predicate_json
=> '{{}}', \
+ on_column => 'email', transform_json => '{{}}')"
+ ),
+ "COLUMN_MASKING policy cannot specify predicate_json.",
+ )
+ .await;
+}
+
+#[tokio::test]
+async fn test_non_rest_catalog_is_rejected_by_every_procedure() {
+ let (_tmp, catalog) = common::create_test_env();
+ let context = common::create_sql_context(catalog).await;
+
+ for sql in [
+ "CALL sys.grant_permission(resource_type => 'CATALOG', access =>
'CREATEDATABASE', principal => 'admin')",
+ "CALL sys.revoke_permission(resource_type => 'CATALOG', access =>
'CREATEDATABASE', principal => 'admin')",
+ "CALL sys.list_permissions(resource_type => 'CATALOG')",
+ "CALL sys.create_policy(database => 'sales', table => 'orders',
policy_type => 'ROW_FILTER', principal => 'p', predicate_json => '{}')",
+ "CALL sys.drop_policy(database => 'sales', table => 'orders',
policy_type => 'ROW_FILTER', principal => 'p')",
+ "CALL sys.list_policies(database => 'sales', table => 'orders')",
+ ] {
+ common::assert_sql_error(
+ &context,
+ sql,
+ "does not support permission or policy management",
+ )
+ .await;
+ }
+}
+
+#[tokio::test]
+async fn test_argument_validation_follows_java() {
+ let (_tmp, server, context) = setup().await;
+
+ let error = context
+ .sql(&format!(
+ "CALL sys.grant_permission(resource_type => 'TABLE', database =>
'{DATABASE}', \
+ table => '{TABLE}', access => 'SELECT', principal => 'analyst', \
+ expire_time => '\u{a0}')"
+ ))
+ .await
+ .expect_err("a non-blank expire_time that is not a timestamp must be
rejected")
+ .to_string();
+ assert!(error.contains("ISO-8601"), "{error}");
+
+ context
+ .sql(&format!(
+ "CALL sys.grant_permission(resource_type => 'TABLE', database =>
'{DATABASE}', \
+ table => '{TABLE}', access => 'SELECT', principal => 'analyst', \
+ expire_time => '\u{0}')"
+ ))
+ .await
+ .unwrap();
+ let body = server.grant_permission_bodies().pop().unwrap();
+ assert!(
+ body.get("expireTime").is_none(),
+ "a NUL-only expire_time is blank to Java and must not be sent: {body}"
+ );
+
+ let error = context
+ .sql("CALL sys.list_permissions(resource_type => ' TABLE ')")
+ .await
+ .unwrap_err()
+ .to_string();
+ assert!(error.contains("Invalid resource_type"), "{error}");
+ let error = context
+ .sql("CALL sys.list_permissions(resource_type => ' ')")
+ .await
+ .unwrap_err()
+ .to_string();
+ assert!(error.contains("resource_type cannot be empty"), "{error}");
+
+ for (sql, expected) in [
+ (
+ "CALL sys.grant_permission(resource_type => 'TABEL', access =>
'SELECT', \
+ principal => 'role:analyst')"
+ .to_string(),
+ "Invalid resource_type 'TABEL'. Expected one of [CATALOG,
CATALOG_ALL, \
+ DATABASE, DATABASE_ALL, TABLE, COLUMN, VIEW, FUNCTION].",
+ ),
+ (
+ format!(
+ "CALL sys.list_policies(database => '{DATABASE}', table =>
'{TABLE}', \
+ policy_type => 'ROW_FILTERS')"
+ ),
+ "Invalid policy_type 'ROW_FILTERS'. Expected one of [ROW_FILTER,
COLUMN_MASKING].",
+ ),
+ (
+ "CALL sys.list_permissions(database => 'sales')".to_string(),
+ "Missing required argument: 'resource_type'",
+ ),
+ ] {
+ common::assert_sql_error(&context, &sql, expected).await;
+ }
+}
+
+#[tokio::test]
+async fn test_unknown_and_duplicate_arguments_are_rejected() {
+ let (_tmp, server, context) = setup().await;
+
+ let grant = |extra: &str| {
+ format!(
+ "CALL sys.grant_permission(resource_type => 'TABLE', database =>
'{DATABASE}', \
+ table => '{TABLE}', access => 'SELECT', principal =>
'role:analyst', {extra})"
+ )
+ };
+
+ common::assert_sql_error(
+ &context,
+ &grant("expiretime => '2030-01-01T00:00:00Z'"),
+ "Argument expiretime is unknown.",
+ )
+ .await;
+
+ common::assert_sql_error(
+ &context,
+ &grant("principal => 'role:other'"),
+ "Procedure argument principal is duplicated.",
+ )
+ .await;
+
+ common::assert_sql_error(
+ &context,
+ &format!(
+ "CALL sys.list_policies(database => '{DATABASE}', table =>
'{TABLE}', \
+ page_tokn => '1')"
+ ),
+ "Argument page_tokn is unknown.",
+ )
+ .await;
+
+ assert!(server.grant_permission_bodies().is_empty());
+ assert!(server.list_policies_queries().is_empty());
+}
+
+#[tokio::test]
+async fn test_view_and_function_resources_round_trip() {
+ let (_tmp, server, context) = setup().await;
+
+ for (resource_type, locator, name) in [("VIEW", "view", "v1"),
("FUNCTION", "function", "f1")] {
+ assert_ok_row(
+ &context,
+ &format!(
+ "CALL sys.grant_permission(resource_type => '{resource_type}',
\
+ database => '{DATABASE}', {locator} => '{name}', access =>
'SELECT', \
+ principal => 'role:analyst')"
+ ),
+ )
+ .await;
+
+ let listed = rows(
+ &call(
+ &context,
+ &format!(
+ "CALL sys.list_permissions(resource_type =>
'{resource_type}', \
+ database => '{DATABASE}', {locator} => '{name}')"
+ ),
+ )
+ .await,
+ );
+ assert_eq!(cell(&listed, 0, 0), Some(resource_type));
+ assert_eq!(cell(&listed, 0, 1), Some(DATABASE));
+ assert_eq!(cell(&listed, 0, 2), None);
+ // Column 3 is `function` and column 4 is `view`, in Java's order.
+ let (function, view) = (cell(&listed, 0, 3), cell(&listed, 0, 4));
+ if resource_type == "VIEW" {
+ assert_eq!((function, view), (None, Some(name)));
+ } else {
+ assert_eq!((function, view), (Some(name), None));
+ }
+ }
+ assert_eq!(server.grant_permission_bodies().len(), 2);
+}
+
+#[tokio::test]
+async fn test_all_three_procedure_name_forms_resolve() {
+ let (_tmp, server, context) = setup().await;
+
+ for name in [
+ "grant_permission",
+ "sys.grant_permission",
+ "paimon.sys.grant_permission",
+ ] {
+ assert_ok_row(
+ &context,
+ &format!(
+ "CALL {name}(resource_type => 'TABLE', access => 'SELECT', \
+ principal => 'role:analyst', database => '{DATABASE}', table
=> '{TABLE}')"
+ ),
+ )
+ .await;
+ }
+ assert_eq!(server.grant_permission_bodies().len(), 3);
+ assert_eq!(server.permissions().len(), 1);
+}
diff --git a/crates/paimon/src/catalog/filesystem.rs
b/crates/paimon/src/catalog/filesystem.rs
index 743f0028..d660697f 100644
--- a/crates/paimon/src/catalog/filesystem.rs
+++ b/crates/paimon/src/catalog/filesystem.rs
@@ -288,6 +288,12 @@ impl FileSystemCatalog {
#[async_trait]
impl Catalog for FileSystemCatalog {
+ /// Opts in so that a failed downcast means "wrong catalog", not "never
opted in".
+ /// Do not delete: it is what makes the REST-only capability checks test
what they claim.
+ fn as_any(&self) -> Option<&dyn std::any::Any> {
+ Some(self)
+ }
+
async fn list_databases(&self) -> Result<Vec<String>> {
let dirs = self.list_directories(&self.warehouse).await?;
Ok(dirs
diff --git a/crates/paimon/src/catalog/mod.rs b/crates/paimon/src/catalog/mod.rs
index 0c66f8be..b610680c 100644
--- a/crates/paimon/src/catalog/mod.rs
+++ b/crates/paimon/src/catalog/mod.rs
@@ -347,6 +347,13 @@ impl LoadedTable {
/// Corresponds to
[org.apache.paimon.catalog.Catalog](https://github.com/apache/paimon/blob/release-1.3/paimon-core/src/main/java/org/apache/paimon/catalog/Catalog.java).
#[async_trait]
pub trait Catalog: Send + Sync {
+ /// Hook for catalog-specific APIs that are not on this trait, as Java
reaches `RESTCatalog`
+ /// through `DelegateCatalog.rootCatalog(...) instanceof RESTCatalog`.
Defaults to `None`;
+ /// override as `Some(self)` in a catalog that has such APIs.
+ fn as_any(&self) -> Option<&dyn std::any::Any> {
+ None
+ }
+
// ======================= database methods ===============================
/// List names of all databases in this catalog.
diff --git a/crates/paimon/src/catalog/rest/rest_catalog.rs
b/crates/paimon/src/catalog/rest/rest_catalog.rs
index 724dbcdb..a7e410cb 100644
--- a/crates/paimon/src/catalog/rest/rest_catalog.rs
+++ b/crates/paimon/src/catalog/rest/rest_catalog.rs
@@ -198,6 +198,9 @@ impl RESTCatalog {
#[async_trait]
impl Catalog for RESTCatalog {
+ fn as_any(&self) -> Option<&dyn std::any::Any> {
+ Some(self)
+ }
// ======================= database methods ===============================
async fn list_databases(&self) -> Result<Vec<String>> {
diff --git a/docs/src/sql.md b/docs/src/sql.md
index 7848f91c..6cad8cfe 100644
--- a/docs/src/sql.md
+++ b/docs/src/sql.md
@@ -1356,6 +1356,123 @@ CALL sys.create_lumina_index(
);
```
+### grant_permission
+
+Grant an access to a principal on a REST catalog. Only the REST catalog
supports
+permission and policy management:
+
+```sql
+CALL sys.grant_permission(
+ resource_type => 'TABLE',
+ database => 'sales',
+ table => 'orders',
+ access => 'SELECT',
+ principal => 'user:alice'
+);
+```
+
+`resource_type` is one of `CATALOG`, `CATALOG_ALL`, `DATABASE`, `DATABASE_ALL`,
+`TABLE`, `COLUMN`, `VIEW` or `FUNCTION`, and decides which locators are
required:
+`DATABASE` and `DATABASE_ALL` need `database`, while `TABLE`, `COLUMN`, `VIEW`
+and `FUNCTION` need `database` plus the matching `table`, `view` or `function`.
+`expire_time` is an ISO-8601 UTC instant ending in `Z`.
+
+A `COLUMN` grant narrows an access to part of a table and takes exactly one of
+`column_names` or `excluded_column_names`. **Both are comma-separated strings
+here**, where Java's Spark procedure declares `ARRAY<STRING>`, because a `CALL`
+argument in this engine is always a scalar:
+
+```sql
+CALL sys.grant_permission(
+ resource_type => 'COLUMN',
+ database => 'sales',
+ table => 'orders',
+ access => 'SELECT',
+ principal => 'role:analyst',
+ column_names => 'id, region'
+);
+```
+
+### revoke_permission
+
+Remove an access from a principal. Revoking one that was never granted
succeeds:
+
+```sql
+CALL sys.revoke_permission(
+ resource_type => 'TABLE',
+ database => 'sales',
+ table => 'orders',
+ access => 'SELECT',
+ principal => 'user:alice'
+);
+```
+
+### list_permissions
+
+List the assignments on an exact resource, optionally filtered by `principal`
+and `access`, and paged with `max_results` and `page_token`. Only grants made
on
+that exact resource are returned, so a `DATABASE_ALL` or `COLUMN` grant that
also
+covers the table does not appear:
+
+```sql
+CALL sys.list_permissions(
+ resource_type => 'TABLE',
+ database => 'sales',
+ table => 'orders'
+);
+```
+
+Every row carries `next_page_token`, so an empty page returns no rows and no
+token. The `column_names` and `excluded_column_names` columns are comma-joined.
+
+### create_policy
+
+Attach a row filter or a column mask to a table for one principal. A policy
+carries exactly one of the two, so `ROW_FILTER` takes `predicate_json` and
+rejects `on_column` and `transform_json`, while `COLUMN_MASKING` takes
+`on_column` and `transform_json` and rejects `predicate_json`:
+
+```sql
+CALL sys.create_policy(
+ database => 'sales',
+ table => 'orders',
+ policy_type => 'ROW_FILTER',
+ principal => 'role:analyst',
+ predicate_json =>
'{"kind":"LEAF","transform":{"name":"FIELD_REF","fieldRef":{"index":1,"name":"region","type":"STRING"}},"function":"EQUAL","literals":["APAC"]}'
+);
+```
+
+`predicate_json` and `transform_json` are serialized Paimon `Predicate` and
+`Transform` values, not SQL. They are the same JSON the catalog returns for a
+table's query authorization.
+
+### drop_policy
+
+Remove a policy by its identity, which is the table, the principal, the type,
+and for a column mask the column. `if_exists => 'true'` swallows a missing
+policy, but not a missing table:
+
+```sql
+CALL sys.drop_policy(
+ database => 'sales',
+ table => 'orders',
+ policy_type => 'COLUMN_MASKING',
+ principal => 'role:analyst',
+ column => 'email',
+ if_exists => 'true'
+);
+```
+
+### list_policies
+
+List the policies on a table, optionally filtered by `policy_type` and
+`principal`, and paged like `list_permissions`. A `column` filter is only
+meaningful for a column mask, so it requires `policy_type => 'COLUMN_MASKING'`:
+
+```sql
+CALL sys.list_policies(database => 'sales', table => 'orders');
+```
+
## Queries
### Basic Queries
