diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 217fccafa..f652b8fe8 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -267,11 +267,13 @@ 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: > cargo test --locked -p paimon-datafusion - --features vortex + --features vortex --test read_tables vector_search_tests::test_lumina_build_then_vector_search_query -- --ignored --exact diff --git a/crates/integrations/datafusion/src/procedures.rs b/crates/integrations/datafusion/src/procedures.rs index 04941451d..3ea1f6c68 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` 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>, @@ -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 { 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> { 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, + catalog_name: &str, +) -> DFResult<&'a RESTCatalog> { + catalog + .as_any() + .and_then(|any| any.downcast_ref::()) + .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, 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 ''. Expected one of .` +trait ProcedureEnum: std::str::FromStr + 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(args: &HashMap, name: &str) -> DFResult +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(args: &HashMap, name: &str) -> DFResult> +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`, but a +/// DataFusion CALL argument is always a scalar. +fn comma_list(args: &HashMap, name: &str) -> Option> { + 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, name: &str) -> DFResult { + // 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) -> DFResult> { + 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) -> DFResult { + PermissionResource::new( + enum_arg::(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) -> DFResult { + 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) -> DFResult> { + 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>>, +) -> DFResult { + 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::>(), + )); + let columns = (0..fields.len()) + .map(|column| { + Arc::new( + rows.iter() + .map(|row| row[column].clone()) + .collect::(), + ) as ArrayRef + }) + .collect::>(); + ctx.read_batch(RecordBatch::try_new(schema, columns)?) +} + +fn joined(values: Option<&[String]>) -> Option { + values.map(|values| values.join(",")) +} + +async fn proc_grant_permission( + ctx: &SessionContext, + catalog: &Arc, + catalog_name: &str, + args: &HashMap, +) -> DFResult { + 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, + catalog_name: &str, + args: &HashMap, +) -> DFResult { + 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, + catalog_name: &str, + args: &HashMap, +) -> DFResult { + 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, + catalog_name: &str, + args: &HashMap, +) -> DFResult { + 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::(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, + catalog_name: &str, + args: &HashMap, +) -> DFResult { + let rest = rest_catalog(catalog, catalog_name)?; + rest.drop_policy( + &policy_table_resource(args)?, + enum_arg::(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, + catalog_name: &str, + args: &HashMap, +) -> DFResult { + 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 { 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 000000000..a26a405bd --- /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 { + 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 { + context + .sql(sql) + .await + .unwrap() + .schema() + .fields() + .iter() + .map(|field| field.name().clone()) + .collect() +} + +fn rows(batches: &[RecordBatch]) -> Vec>> { + 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>], 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::>>::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::>>::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 743f00286..d660697f7 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> { 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 0c66f8bef..b610680c3 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 724dbcdbb..a7e410cbb 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> { diff --git a/docs/src/sql.md b/docs/src/sql.md index 8dc81360f..b9e3d2f7e 100644 --- a/docs/src/sql.md +++ b/docs/src/sql.md @@ -1306,6 +1306,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`, 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