diff --git a/docs/docs/concepts/rest/management-api.md b/docs/docs/concepts/rest/management-api.md index 751eb6f3afa6..e90e9a71d412 100644 --- a/docs/docs/concepts/rest/management-api.md +++ b/docs/docs/concepts/rest/management-api.md @@ -231,3 +231,329 @@ schema-incompatible predicate or transform must also fail closed rather than omi This experimental contract deliberately does not define governed tags, catalog/database policy inheritance, or tag-driven matching. Those features need explicit match conditions and conflict rules before being added. + +## Spark SQL procedures + +The following examples assume a Spark catalog named `paimon`. Replace it with the catalog name in +`spark.sql.catalog.`. + +### Grant permissions + +`grant_permission` returns one row with `result = true` when the server accepts the assignment. + +Grant permission to create databases in the catalog: + +```sql +CALL paimon.sys.grant_permission( + resource_type => 'CATALOG', + access => 'CREATEDATABASE', + principal => 'role:catalog_user' +); +``` + +Grant read access to every applicable object currently or subsequently created in the catalog. This +does not grant catalog-level operations such as `CREATEDATABASE`: + +```sql +CALL paimon.sys.grant_permission( + resource_type => 'CATALOG_ALL', + access => 'SELECT', + principal => 'role:catalog_reader' +); +``` + +Grant permission to create views in a database with an optional expiration time: + +```sql +CALL paimon.sys.grant_permission( + resource_type => 'DATABASE', + database => 'sales', + access => 'CREATEVIEW', + principal => 'role:data_engineer', + expire_time => '2027-01-01T00:00:00Z' +); +``` + +Grant write access to every applicable table currently or subsequently created in one database. +`DATABASE_ALL` requires `database` but does not accept a table, function, or view locator: + +```sql +CALL paimon.sys.grant_permission( + resource_type => 'DATABASE_ALL', + database => 'sales', + access => 'UPDATE', + principal => 'role:sales_writer' +); +``` + +Grant table, function, and view access with the matching locator: + +```sql +CALL paimon.sys.grant_permission( + resource_type => 'TABLE', + database => 'sales', + table => 'orders', + access => 'SELECT', + principal => 'user:alice' +); + +CALL paimon.sys.grant_permission( + resource_type => 'FUNCTION', + database => 'sales', + function => 'calculate_tax', + access => 'SELECT', + principal => 'role:analyst' +); + +CALL paimon.sys.grant_permission( + resource_type => 'VIEW', + database => 'sales', + view => 'daily_orders', + access => 'SELECT', + principal => 'service:reporting_job' +); +``` + +Grant access to selected columns. This requires table query authorization; named arguments are +recommended because the two column range modes are mutually exclusive: + +```sql +ALTER TABLE paimon.sales.orders +SET TBLPROPERTIES ('query-auth.enabled' = 'true'); + +CALL paimon.sys.grant_permission( + resource_type => 'COLUMN', + database => 'sales', + table => 'orders', + access => 'SELECT', + principal => 'role:analyst', + column_names => array('order_id', 'region') +); +``` + +Use `excluded_column_names` for a denylist. Repeating the grant replaces the preceding allowlist in +one operation: + +```sql +CALL paimon.sys.grant_permission( + resource_type => 'COLUMN', + database => 'sales', + table => 'orders', + access => 'SELECT', + principal => 'role:analyst', + excluded_column_names => array('email', 'phone_number') +); +``` + +### List permissions + +`list_permissions` always addresses one exact resource or explicit descendant scope. Omit optional +filters to list every direct assignment on it; effective assignments inherited from a scope are not +synthesized: + +```sql +CALL paimon.sys.list_permissions( + resource_type => 'TABLE', + database => 'sales', + table => 'orders' +); +``` + +Filter by principal or access: + +```sql +CALL paimon.sys.list_permissions( + resource_type => 'TABLE', + database => 'sales', + table => 'orders', + principal => 'role:analyst', + access => 'SELECT' +); +``` + +List the column range attached to a principal. The result exposes `column_names` and +`excluded_column_names` as `ARRAY` columns, with exactly one populated for a `COLUMN` +assignment: + +```sql +CALL paimon.sys.list_permissions( + resource_type => 'COLUMN', + database => 'sales', + table => 'orders', + principal => 'role:analyst', + access => 'SELECT' +); +``` + +The `next_page_token` output is opaque; pass it back unchanged with the same filters: + +```sql +CALL paimon.sys.list_permissions( + resource_type => 'TABLE', + database => 'sales', + table => 'orders', + max_results => 50, + page_token => 'opaque-token-from-previous-row' +); +``` + +### Revoke permissions + +Supply the same three identity fields used by the grant. `expire_time` is not part of identity. + +```sql +CALL paimon.sys.revoke_permission( + resource_type => 'TABLE', + database => 'sales', + table => 'orders', + access => 'SELECT', + principal => 'role:sales_reader' +); +``` + +Repeating the same call succeeds even when the assignment is already absent. + +Column revocation uses the containing table identity and removes the complete range: + +```sql +CALL paimon.sys.revoke_permission( + resource_type => 'COLUMN', + database => 'sales', + table => 'orders', + access => 'SELECT', + principal => 'role:analyst' +); +``` + +### Create row-filter policies + +Before attaching any policy, enable table query authorization: + +```sql +ALTER TABLE paimon.sales.orders +SET TBLPROPERTIES ('query-auth.enabled' = 'true'); +``` + +`create_policy` accepts the canonical `principal` and a serialized Paimon `Predicate`. The JSON +below is the same representation accepted in one `AuthTableQueryResponse.filter` entry. Named +arguments are recommended because row-filter and column-mask definitions use different fields: + +```sql +CALL paimon.sys.create_policy( + database => 'sales', + table => 'orders', + policy_type => 'ROW_FILTER', + principal => 'group:analysts', + predicate_json => '{"kind":"LEAF","transform":{"name":"FIELD_REF","fieldRef":{"index":1,"name":"region","type":"STRING"}},"function":"EQUAL","literals":["APAC"]}' +); +``` + +The call fails if that principal already has a row filter on the table. Drop the existing policy +before creating a different definition for the same identity. Create another policy for a second +principal with a separate call. + +### Create column-masking policies + +For column masking, `on_column` identifies the protected column and `transform_json` is the same +serialized Paimon `Transform` representation used as an +`AuthTableQueryResponse.columnMasking` value. This example replaces every visible phone number +with a fixed string: + +```sql +CALL paimon.sys.create_policy( + database => 'sales', + table => 'customers', + policy_type => 'COLUMN_MASKING', + principal => 'role:support', + on_column => 'phone_number', + transform_json => '{"name":"CONCAT","inputs":["****"]}' +); +``` + +A transform may reference table fields by name. The server remaps their indices to the current +schema, rejects missing fields, and verifies that the result type matches `on_column`: + +```sql +CALL paimon.sys.create_policy( + database => 'sales', + table => 'customers', + policy_type => 'COLUMN_MASKING', + principal => 'group:support', + on_column => 'email', + transform_json => '{"name":"CONCAT","inputs":[{"index":1,"name":"region","type":"STRING"},"-masked"]}' +); +``` + +`predicate_json` is required only for `ROW_FILTER`. `on_column` and `transform_json` are required +only for `COLUMN_MASKING`. JSON containing a single quote must escape it as `''` inside the SQL +string literal. + +### List policies + +List every policy directly attached to one table: + +```sql +CALL paimon.sys.list_policies( + database => 'sales', + table => 'orders' +); +``` + +Filter by policy type or principal. A `column` filter is valid only with +`policy_type => 'COLUMN_MASKING'`: + +```sql +CALL paimon.sys.list_policies( + database => 'sales', + table => 'orders', + policy_type => 'ROW_FILTER', + principal => 'group:analysts' +); +``` + +The output columns are `database`, `table`, `policy_type`, `principal`, `predicate_json`, +`on_column`, `transform_json`, and `next_page_token`. A row filter has only `predicate_json`; a +column mask has only `on_column` and `transform_json`. Pass an opaque continuation token back +unchanged with the same filters: + +```sql +CALL paimon.sys.list_policies( + database => 'sales', + table => 'orders', + max_results => 50, + page_token => 'opaque-token-from-previous-row' +); +``` + +Management listing follows the existing Paimon pagination contract: an empty page terminates +pagination and therefore has no continuation token. Each Spark procedure returns exactly the page +selected by `page_token`; pass a non-null `next_page_token` back unchanged to retrieve the next page. + +### Drop policies + +Drop an existing policy: + +```sql +CALL paimon.sys.drop_policy( + database => 'sales', + table => 'orders', + policy_type => 'ROW_FILTER', + principal => 'group:analysts' +); +``` + +By default an absent policy is an error. Set `if_exists => true` for an idempotent operation: + +```sql +CALL paimon.sys.drop_policy( + database => 'sales', + table => 'orders', + policy_type => 'ROW_FILTER', + principal => 'group:analysts', + if_exists => true +); +``` + +Creating, dropping, or inspecting permissions and policies requires the server to +authorize the caller for `GRANT` on the relevant resource. Authentication, principal +membership, policy persistence, schema validation, and audit logging remain REST server concerns. diff --git a/paimon-spark/paimon-spark-common/src/main/antlr4/org.apache.spark.sql.catalyst.parser.extensions/PaimonSqlExtensions.g4 b/paimon-spark/paimon-spark-common/src/main/antlr4/org.apache.spark.sql.catalyst.parser.extensions/PaimonSqlExtensions.g4 index 8c2e45b34ec2..bbd2728d9d91 100644 --- a/paimon-spark/paimon-spark-common/src/main/antlr4/org.apache.spark.sql.catalyst.parser.extensions/PaimonSqlExtensions.g4 +++ b/paimon-spark/paimon-spark-common/src/main/antlr4/org.apache.spark.sql.catalyst.parser.extensions/PaimonSqlExtensions.g4 @@ -167,6 +167,7 @@ overwriteClause expression : constant | stringMap + | stringArray ; constant @@ -180,6 +181,10 @@ stringMap : MAP '(' constant (',' constant)* ')' ; +stringArray + : ARRAY '(' (constant (',' constant)*)? ')' + ; + booleanValue : TRUE | FALSE ; @@ -214,7 +219,7 @@ nonReserved : ALTER | AS | CALL | CREATE | DAYS | DELETE | EXISTS | HOURS | IF | LIKE | NOT | OF | OR | TABLE | REPLACE | RETAIN | VERSION | TAG | TRUE | FALSE - | MAP + | ARRAY | MAP | COPY | INTO | FROM | FILE_FORMAT | PATTERN | FORCE | ON_ERROR | ABORT_STATEMENT | CONTINUE | SKIP_FILE | OVERWRITE | CSV | JSON @@ -249,6 +254,7 @@ TRUE: 'TRUE'; FALSE: 'FALSE'; MAP: 'MAP'; +ARRAY: 'ARRAY'; COPY: 'COPY'; INTO: 'INTO'; diff --git a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/SparkProcedures.java b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/SparkProcedures.java index 60b5747e3db7..866a54e802a9 100644 --- a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/SparkProcedures.java +++ b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/SparkProcedures.java @@ -29,16 +29,21 @@ import org.apache.paimon.spark.procedure.CreateBranchProcedure; import org.apache.paimon.spark.procedure.CreateFunctionProcedure; import org.apache.paimon.spark.procedure.CreateGlobalIndexProcedure; +import org.apache.paimon.spark.procedure.CreatePolicyProcedure; import org.apache.paimon.spark.procedure.CreateTagFromTimestampProcedure; import org.apache.paimon.spark.procedure.CreateTagProcedure; import org.apache.paimon.spark.procedure.DeleteBranchProcedure; import org.apache.paimon.spark.procedure.DeleteTagProcedure; import org.apache.paimon.spark.procedure.DropFunctionProcedure; import org.apache.paimon.spark.procedure.DropGlobalIndexProcedure; +import org.apache.paimon.spark.procedure.DropPolicyProcedure; import org.apache.paimon.spark.procedure.ExpirePartitionsProcedure; import org.apache.paimon.spark.procedure.ExpireSnapshotsProcedure; import org.apache.paimon.spark.procedure.ExpireTagsProcedure; import org.apache.paimon.spark.procedure.FastForwardProcedure; +import org.apache.paimon.spark.procedure.GrantPermissionProcedure; +import org.apache.paimon.spark.procedure.ListPermissionsProcedure; +import org.apache.paimon.spark.procedure.ListPoliciesProcedure; import org.apache.paimon.spark.procedure.MarkPartitionDoneProcedure; import org.apache.paimon.spark.procedure.MaterializeDeletionVectorsProcedure; import org.apache.paimon.spark.procedure.MergeBranchProcedure; @@ -57,6 +62,7 @@ import org.apache.paimon.spark.procedure.ReplaceTagProcedure; import org.apache.paimon.spark.procedure.RescaleProcedure; import org.apache.paimon.spark.procedure.ResetConsumerProcedure; +import org.apache.paimon.spark.procedure.RevokePermissionProcedure; import org.apache.paimon.spark.procedure.RewriteFileIndexProcedure; import org.apache.paimon.spark.procedure.RollbackProcedure; import org.apache.paimon.spark.procedure.RollbackToTimestampProcedure; @@ -134,6 +140,12 @@ private static Map> initProcedureBuilders() { procedureBuilders.put("rewrite_file_index", RewriteFileIndexProcedure::builder); procedureBuilders.put("copy", CopyFilesProcedure::builder); procedureBuilders.put("reassign_row_id", ReassignRowIdProcedure::builder); + procedureBuilders.put("grant_permission", GrantPermissionProcedure::builder); + procedureBuilders.put("revoke_permission", RevokePermissionProcedure::builder); + procedureBuilders.put("list_permissions", ListPermissionsProcedure::builder); + procedureBuilders.put("create_policy", CreatePolicyProcedure::builder); + procedureBuilders.put("drop_policy", DropPolicyProcedure::builder); + procedureBuilders.put("list_policies", ListPoliciesProcedure::builder); return procedureBuilders.build(); } } diff --git a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/BasePermissionProcedure.java b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/BasePermissionProcedure.java new file mode 100644 index 000000000000..8a4ba0b069f7 --- /dev/null +++ b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/BasePermissionProcedure.java @@ -0,0 +1,130 @@ +/* + * 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. + */ + +package org.apache.paimon.spark.procedure; + +import org.apache.paimon.catalog.Catalog; +import org.apache.paimon.catalog.DelegateCatalog; +import org.apache.paimon.management.PermissionAssignment; +import org.apache.paimon.management.PermissionColumns; +import org.apache.paimon.management.PermissionManagement; +import org.apache.paimon.management.PermissionResource; +import org.apache.paimon.management.PolicyManagement; +import org.apache.paimon.management.ResourceType; +import org.apache.paimon.rest.RESTCatalog; +import org.apache.paimon.spark.catalog.WithPaimonCatalog; + +import org.apache.spark.sql.connector.catalog.TableCatalog; + +import javax.annotation.Nullable; + +import java.util.Arrays; +import java.util.Locale; + +import static org.apache.paimon.utils.Preconditions.checkArgument; + +/** Shared REST catalog lookup and argument validation for management procedures. */ +abstract class BasePermissionProcedure extends BaseProcedure { + + protected BasePermissionProcedure(TableCatalog tableCatalog) { + super(tableCatalog); + } + + protected PermissionManagement permissionManagement() { + return restCatalog().permissionManagement(); + } + + protected PolicyManagement policyManagement() { + return restCatalog().policyManagement(); + } + + private RESTCatalog restCatalog() { + checkArgument( + tableCatalog() instanceof WithPaimonCatalog, + "Catalog '%s' is not a Paimon catalog.", + tableCatalog().name()); + Catalog root = + DelegateCatalog.rootCatalog(((WithPaimonCatalog) tableCatalog()).paimonCatalog()); + checkArgument( + root instanceof RESTCatalog, + "Catalog '%s' does not support permission or policy management.", + tableCatalog().name()); + return (RESTCatalog) root; + } + + protected static PermissionAssignment assignment( + ResourceType resourceType, + String access, + String principal, + @Nullable String database, + @Nullable String table, + @Nullable String function, + @Nullable String view, + @Nullable PermissionColumns columns, + @Nullable String expireTime) { + return new PermissionAssignment( + resource(resourceType, database, table, function, view), + access, + principal, + columns, + emptyToNull(expireTime)); + } + + protected static PermissionResource resource( + ResourceType resourceType, + @Nullable String database, + @Nullable String table, + @Nullable String function, + @Nullable String view) { + return new PermissionResource( + resourceType, + emptyToNull(database), + emptyToNull(table), + emptyToNull(function), + emptyToNull(view)); + } + + protected static > E enumValue( + String value, Class enumClass, String argument) { + checkArgument(!isBlank(value), "%s cannot be empty.", argument); + try { + return Enum.valueOf(enumClass, value.toUpperCase(Locale.ROOT)); + } catch (IllegalArgumentException e) { + throw new IllegalArgumentException( + String.format( + "Invalid %s '%s'. Expected one of %s.", + argument, value, Arrays.toString(enumClass.getEnumConstants())), + e); + } + } + + @Nullable + protected static > E optionalEnum( + @Nullable String value, Class enumClass, String argument) { + return isBlank(value) ? null : enumValue(value, enumClass, argument); + } + + @Nullable + protected static String emptyToNull(@Nullable String value) { + return isBlank(value) ? null : value; + } + + protected static boolean isBlank(@Nullable String value) { + return value == null || value.trim().isEmpty(); + } +} diff --git a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/BasePolicyProcedure.java b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/BasePolicyProcedure.java new file mode 100644 index 000000000000..e96d31df7749 --- /dev/null +++ b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/BasePolicyProcedure.java @@ -0,0 +1,62 @@ +/* + * 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. + */ + +package org.apache.paimon.spark.procedure; + +import org.apache.paimon.management.ColumnMask; +import org.apache.paimon.management.DataPolicy; +import org.apache.paimon.management.PermissionResource; +import org.apache.paimon.management.PolicyType; +import org.apache.paimon.management.ResourceType; +import org.apache.paimon.management.RowFilter; + +import org.apache.spark.sql.connector.catalog.TableCatalog; + +import javax.annotation.Nullable; + +import static org.apache.paimon.utils.Preconditions.checkArgument; + +/** Shared typed argument conversion for table policy procedures. */ +abstract class BasePolicyProcedure extends BasePermissionProcedure { + + protected BasePolicyProcedure(TableCatalog tableCatalog) { + super(tableCatalog); + } + + protected static DataPolicy policy( + String database, + String table, + PolicyType policyType, + String principal, + @Nullable String predicate, + @Nullable String onColumn, + @Nullable String transform) { + PermissionResource resource = tableResource(database, table); + if (policyType == PolicyType.ROW_FILTER) { + checkArgument(isBlank(onColumn), "ROW_FILTER policy cannot specify on_column."); + checkArgument(isBlank(transform), "ROW_FILTER policy cannot specify transform."); + return DataPolicy.rowFilter(resource, new RowFilter(predicate), principal); + } + checkArgument(isBlank(predicate), "COLUMN_MASKING policy cannot specify predicate."); + return DataPolicy.columnMask(resource, new ColumnMask(onColumn, transform), principal); + } + + protected static PermissionResource tableResource(String database, String table) { + return resource(ResourceType.TABLE, database, table, null, null); + } +} diff --git a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/CreatePolicyProcedure.java b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/CreatePolicyProcedure.java new file mode 100644 index 000000000000..52b56a55faf9 --- /dev/null +++ b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/CreatePolicyProcedure.java @@ -0,0 +1,100 @@ +/* + * 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. + */ + +package org.apache.paimon.spark.procedure; + +import org.apache.paimon.management.DataPolicy; +import org.apache.paimon.management.PolicyManagement.PolicyAlreadyExistException; +import org.apache.paimon.management.PolicyType; + +import org.apache.spark.sql.catalyst.InternalRow; +import org.apache.spark.sql.connector.catalog.TableCatalog; +import org.apache.spark.sql.types.DataTypes; +import org.apache.spark.sql.types.Metadata; +import org.apache.spark.sql.types.StructField; +import org.apache.spark.sql.types.StructType; + +import static org.apache.spark.sql.types.DataTypes.StringType; + +/** Creates a table row-filter or column-mask policy. */ +public class CreatePolicyProcedure extends BasePolicyProcedure { + + private static final ProcedureParameter[] PARAMETERS = + new ProcedureParameter[] { + ProcedureParameter.required("database", StringType), + ProcedureParameter.required("table", StringType), + ProcedureParameter.required("policy_type", StringType), + ProcedureParameter.required("principal", StringType), + ProcedureParameter.optional("predicate_json", StringType), + ProcedureParameter.optional("on_column", StringType), + ProcedureParameter.optional("transform_json", StringType) + }; + + private static final StructType OUTPUT_TYPE = + new StructType( + new StructField[] { + new StructField("result", DataTypes.BooleanType, false, Metadata.empty()) + }); + + private CreatePolicyProcedure(TableCatalog tableCatalog) { + super(tableCatalog); + } + + @Override + public ProcedureParameter[] parameters() { + return PARAMETERS; + } + + @Override + public StructType outputType() { + return OUTPUT_TYPE; + } + + @Override + public InternalRow[] call(InternalRow args) { + DataPolicy policy = + policy( + args.getString(0), + args.getString(1), + enumValue(args.getString(2), PolicyType.class, PARAMETERS[2].name()), + args.getString(3), + args.isNullAt(4) ? null : args.getString(4), + args.isNullAt(5) ? null : args.getString(5), + args.isNullAt(6) ? null : args.getString(6)); + try { + policyManagement().createPolicy(policy); + } catch (PolicyAlreadyExistException e) { + throw new RuntimeException(e); + } + return new InternalRow[] {newInternalRow(true)}; + } + + public static ProcedureBuilder builder() { + return new Builder() { + @Override + protected CreatePolicyProcedure doBuild() { + return new CreatePolicyProcedure(tableCatalog()); + } + }; + } + + @Override + public String description() { + return "CreatePolicyProcedure"; + } +} diff --git a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/DropPolicyProcedure.java b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/DropPolicyProcedure.java new file mode 100644 index 000000000000..71780d1a84f9 --- /dev/null +++ b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/DropPolicyProcedure.java @@ -0,0 +1,94 @@ +/* + * 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. + */ + +package org.apache.paimon.spark.procedure; + +import org.apache.paimon.management.PermissionResource; +import org.apache.paimon.management.PolicyType; + +import org.apache.spark.sql.catalyst.InternalRow; +import org.apache.spark.sql.connector.catalog.TableCatalog; +import org.apache.spark.sql.types.DataTypes; +import org.apache.spark.sql.types.Metadata; +import org.apache.spark.sql.types.StructField; +import org.apache.spark.sql.types.StructType; + +import static org.apache.spark.sql.types.DataTypes.BooleanType; +import static org.apache.spark.sql.types.DataTypes.StringType; + +/** Drops one principal's row-filter or column-masking policy. */ +public class DropPolicyProcedure extends BasePolicyProcedure { + + private static final ProcedureParameter[] PARAMETERS = + new ProcedureParameter[] { + ProcedureParameter.required("database", StringType), + ProcedureParameter.required("table", StringType), + ProcedureParameter.required("policy_type", StringType), + ProcedureParameter.required("principal", StringType), + ProcedureParameter.optional("column", StringType), + ProcedureParameter.optional("if_exists", BooleanType) + }; + + private static final StructType OUTPUT_TYPE = + new StructType( + new StructField[] { + new StructField("result", DataTypes.BooleanType, false, Metadata.empty()) + }); + + private DropPolicyProcedure(TableCatalog tableCatalog) { + super(tableCatalog); + } + + @Override + public ProcedureParameter[] parameters() { + return PARAMETERS; + } + + @Override + public StructType outputType() { + return OUTPUT_TYPE; + } + + @Override + public InternalRow[] call(InternalRow args) { + PermissionResource resource = tableResource(args.getString(0), args.getString(1)); + PolicyType type = enumValue(args.getString(2), PolicyType.class, PARAMETERS[2].name()); + policyManagement() + .dropPolicy( + resource, + type, + args.getString(3), + args.isNullAt(4) ? null : args.getString(4), + !args.isNullAt(5) && args.getBoolean(5)); + return new InternalRow[] {newInternalRow(true)}; + } + + public static ProcedureBuilder builder() { + return new Builder() { + @Override + protected DropPolicyProcedure doBuild() { + return new DropPolicyProcedure(tableCatalog()); + } + }; + } + + @Override + public String description() { + return "DropPolicyProcedure"; + } +} diff --git a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/GrantPermissionProcedure.java b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/GrantPermissionProcedure.java new file mode 100644 index 000000000000..14cbef3b7c1e --- /dev/null +++ b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/GrantPermissionProcedure.java @@ -0,0 +1,139 @@ +/* + * 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. + */ + +package org.apache.paimon.spark.procedure; + +import org.apache.paimon.management.PermissionAssignment; +import org.apache.paimon.management.PermissionColumns; +import org.apache.paimon.management.ResourceType; + +import org.apache.spark.sql.catalyst.InternalRow; +import org.apache.spark.sql.catalyst.util.ArrayData; +import org.apache.spark.sql.connector.catalog.TableCatalog; +import org.apache.spark.sql.types.Metadata; +import org.apache.spark.sql.types.StructField; +import org.apache.spark.sql.types.StructType; +import org.apache.spark.unsafe.types.UTF8String; + +import javax.annotation.Nullable; + +import java.util.ArrayList; +import java.util.List; + +import static org.apache.spark.sql.types.DataTypes.StringType; +import static org.apache.spark.sql.types.DataTypes.createArrayType; + +/** Grants a permission through a REST catalog. */ +public class GrantPermissionProcedure extends BasePermissionProcedure { + + private static final ProcedureParameter[] PARAMETERS = + new ProcedureParameter[] { + ProcedureParameter.required("resource_type", StringType), + ProcedureParameter.required("access", StringType), + ProcedureParameter.required("principal", StringType), + ProcedureParameter.optional("database", StringType), + ProcedureParameter.optional("table", StringType), + ProcedureParameter.optional("function", StringType), + ProcedureParameter.optional("view", StringType), + ProcedureParameter.optional("expire_time", StringType), + ProcedureParameter.optional("column_names", createArrayType(StringType)), + ProcedureParameter.optional("excluded_column_names", createArrayType(StringType)) + }; + + private static final StructType OUTPUT_TYPE = + new StructType( + new StructField[] { + new StructField( + "result", + org.apache.spark.sql.types.DataTypes.BooleanType, + false, + Metadata.empty()) + }); + + private GrantPermissionProcedure(TableCatalog tableCatalog) { + super(tableCatalog); + } + + @Override + public ProcedureParameter[] parameters() { + return PARAMETERS; + } + + @Override + public StructType outputType() { + return OUTPUT_TYPE; + } + + @Override + public InternalRow[] call(InternalRow args) { + ResourceType resourceType = + enumValue(args.getString(0), ResourceType.class, PARAMETERS[0].name()); + PermissionAssignment assignment = + assignment( + resourceType, + args.getString(1), + args.getString(2), + args.isNullAt(3) ? null : args.getString(3), + args.isNullAt(4) ? null : args.getString(4), + args.isNullAt(5) ? null : args.getString(5), + args.isNullAt(6) ? null : args.getString(6), + columns(args, 8, 9), + args.isNullAt(7) ? null : args.getString(7)); + + permissionManagement().grantPermission(assignment); + return new InternalRow[] {newInternalRow(true)}; + } + + @Nullable + private static PermissionColumns columns( + InternalRow args, int columnNamesPos, int excludedColumnNamesPos) { + List columnNames = stringArray(args, columnNamesPos); + List excludedColumnNames = stringArray(args, excludedColumnNamesPos); + return columnNames == null && excludedColumnNames == null + ? null + : new PermissionColumns(columnNames, excludedColumnNames); + } + + @Nullable + private static List stringArray(InternalRow args, int position) { + if (args.isNullAt(position)) { + return null; + } + ArrayData array = args.getArray(position); + List values = new ArrayList<>(array.numElements()); + for (int i = 0; i < array.numElements(); i++) { + UTF8String value = array.isNullAt(i) ? null : array.getUTF8String(i); + values.add(value == null ? null : value.toString()); + } + return values; + } + + public static ProcedureBuilder builder() { + return new Builder() { + @Override + protected GrantPermissionProcedure doBuild() { + return new GrantPermissionProcedure(tableCatalog()); + } + }; + } + + @Override + public String description() { + return "GrantPermissionProcedure"; + } +} diff --git a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/ListPermissionsProcedure.java b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/ListPermissionsProcedure.java new file mode 100644 index 000000000000..70d256e5c99c --- /dev/null +++ b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/ListPermissionsProcedure.java @@ -0,0 +1,160 @@ +/* + * 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. + */ + +package org.apache.paimon.spark.procedure; + +import org.apache.paimon.PagedList; +import org.apache.paimon.management.ListPermissionsRequest; +import org.apache.paimon.management.PermissionAssignment; +import org.apache.paimon.management.PermissionColumns; +import org.apache.paimon.management.ResourceType; + +import org.apache.spark.sql.catalyst.InternalRow; +import org.apache.spark.sql.catalyst.util.GenericArrayData; +import org.apache.spark.sql.connector.catalog.TableCatalog; +import org.apache.spark.sql.types.Metadata; +import org.apache.spark.sql.types.StructField; +import org.apache.spark.sql.types.StructType; +import org.apache.spark.unsafe.types.UTF8String; + +import java.util.List; + +import static org.apache.spark.sql.types.DataTypes.IntegerType; +import static org.apache.spark.sql.types.DataTypes.StringType; +import static org.apache.spark.sql.types.DataTypes.createArrayType; + +/** Lists direct permissions on an exact target. */ +public class ListPermissionsProcedure extends BasePermissionProcedure { + + private static final ProcedureParameter[] PARAMETERS = + new ProcedureParameter[] { + ProcedureParameter.required("resource_type", StringType), + ProcedureParameter.optional("database", StringType), + ProcedureParameter.optional("table", StringType), + ProcedureParameter.optional("function", StringType), + ProcedureParameter.optional("view", StringType), + ProcedureParameter.optional("principal", StringType), + ProcedureParameter.optional("access", StringType), + ProcedureParameter.optional("max_results", IntegerType), + ProcedureParameter.optional("page_token", StringType) + }; + + private static final StructType OUTPUT_TYPE = + new StructType( + new StructField[] { + field("resource_type", StringType, false), + field("database", StringType, true), + field("table", StringType, true), + field("function", StringType, true), + field("view", StringType, true), + field("access", StringType, false), + field("principal", StringType, false), + field("column_names", createArrayType(StringType), true), + field("excluded_column_names", createArrayType(StringType), true), + field("expire_time", StringType, true), + field("next_page_token", StringType, true) + }); + + private ListPermissionsProcedure(TableCatalog tableCatalog) { + super(tableCatalog); + } + + @Override + public ProcedureParameter[] parameters() { + return PARAMETERS; + } + + @Override + public StructType outputType() { + return OUTPUT_TYPE; + } + + @Override + public InternalRow[] call(InternalRow args) { + ResourceType resourceType = + enumValue(args.getString(0), ResourceType.class, PARAMETERS[0].name()); + Integer maxResults = args.isNullAt(7) ? null : args.getInt(7); + ListPermissionsRequest request = + new ListPermissionsRequest( + resourceType, + args.isNullAt(1) ? null : emptyToNull(args.getString(1)), + args.isNullAt(2) ? null : emptyToNull(args.getString(2)), + args.isNullAt(3) ? null : emptyToNull(args.getString(3)), + args.isNullAt(4) ? null : emptyToNull(args.getString(4)), + args.isNullAt(5) ? null : emptyToNull(args.getString(5)), + args.isNullAt(6) ? null : emptyToNull(args.getString(6)), + args.isNullAt(8) ? null : emptyToNull(args.getString(8)), + maxResults); + PagedList page = permissionManagement().listPermissions(request); + List assignments = page.getElements(); + if (assignments == null || assignments.isEmpty()) { + return new InternalRow[0]; + } + + InternalRow[] rows = new InternalRow[assignments.size()]; + for (int i = 0; i < assignments.size(); i++) { + PermissionAssignment assignment = assignments.get(i); + PermissionColumns columns = assignment.getColumns(); + rows[i] = + newInternalRow( + string(assignment.getResource().getType().name()), + string(assignment.getResource().getDatabase()), + string(assignment.getResource().getTable()), + string(assignment.getResource().getFunction()), + string(assignment.getResource().getView()), + string(assignment.getAccess()), + string(assignment.getPrincipal()), + stringArray(columns == null ? null : columns.getColumnNames()), + stringArray(columns == null ? null : columns.getExcludedColumnNames()), + string(assignment.getExpireTime()), + string(page.getNextPageToken())); + } + return rows; + } + + private static StructField field( + String name, org.apache.spark.sql.types.DataType type, boolean nullable) { + return new StructField(name, type, nullable, Metadata.empty()); + } + + private static UTF8String string(String value) { + return value == null ? null : UTF8String.fromString(value); + } + + private static GenericArrayData stringArray(List values) { + if (values == null) { + return null; + } + return new GenericArrayData( + values.stream().map(ListPermissionsProcedure::string).toArray()); + } + + public static ProcedureBuilder builder() { + return new Builder() { + @Override + protected ListPermissionsProcedure doBuild() { + return new ListPermissionsProcedure(tableCatalog()); + } + }; + } + + @Override + public String description() { + return "ListPermissionsProcedure"; + } +} diff --git a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/ListPoliciesProcedure.java b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/ListPoliciesProcedure.java new file mode 100644 index 000000000000..fa4d8a244fff --- /dev/null +++ b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/ListPoliciesProcedure.java @@ -0,0 +1,140 @@ +/* + * 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. + */ + +package org.apache.paimon.spark.procedure; + +import org.apache.paimon.PagedList; +import org.apache.paimon.management.ColumnMask; +import org.apache.paimon.management.DataPolicy; +import org.apache.paimon.management.ListPoliciesRequest; +import org.apache.paimon.management.PolicyType; +import org.apache.paimon.management.RowFilter; + +import org.apache.spark.sql.catalyst.InternalRow; +import org.apache.spark.sql.connector.catalog.TableCatalog; +import org.apache.spark.sql.types.Metadata; +import org.apache.spark.sql.types.StructField; +import org.apache.spark.sql.types.StructType; +import org.apache.spark.unsafe.types.UTF8String; + +import java.util.List; + +import static org.apache.spark.sql.types.DataTypes.IntegerType; +import static org.apache.spark.sql.types.DataTypes.StringType; + +/** Lists policies attached to an exact table. */ +public class ListPoliciesProcedure extends BasePolicyProcedure { + + private static final ProcedureParameter[] PARAMETERS = + new ProcedureParameter[] { + ProcedureParameter.required("database", StringType), + ProcedureParameter.required("table", StringType), + ProcedureParameter.optional("policy_type", StringType), + ProcedureParameter.optional("principal", StringType), + ProcedureParameter.optional("column", StringType), + ProcedureParameter.optional("max_results", IntegerType), + ProcedureParameter.optional("page_token", StringType) + }; + + private static final StructType OUTPUT_TYPE = + new StructType( + new StructField[] { + field("database", StringType, false), + field("table", StringType, false), + field("policy_type", StringType, false), + field("principal", StringType, false), + field("predicate_json", StringType, true), + field("on_column", StringType, true), + field("transform_json", StringType, true), + field("next_page_token", StringType, true) + }); + + private ListPoliciesProcedure(TableCatalog tableCatalog) { + super(tableCatalog); + } + + @Override + public ProcedureParameter[] parameters() { + return PARAMETERS; + } + + @Override + public StructType outputType() { + return OUTPUT_TYPE; + } + + @Override + public InternalRow[] call(InternalRow args) { + ListPoliciesRequest request = + new ListPoliciesRequest( + tableResource(args.getString(0), args.getString(1)), + optionalEnum( + args.isNullAt(2) ? null : args.getString(2), + PolicyType.class, + PARAMETERS[2].name()), + args.isNullAt(3) ? null : args.getString(3), + args.isNullAt(4) ? null : args.getString(4), + args.isNullAt(6) ? null : args.getString(6), + args.isNullAt(5) ? null : args.getInt(5)); + PagedList page = policyManagement().listPolicies(request); + List policies = page.getElements(); + if (policies == null || policies.isEmpty()) { + return new InternalRow[0]; + } + InternalRow[] rows = new InternalRow[policies.size()]; + for (int i = 0; i < policies.size(); i++) { + DataPolicy policy = policies.get(i); + RowFilter rowFilter = policy.getRowFilter(); + ColumnMask columnMask = policy.getColumnMask(); + rows[i] = + newInternalRow( + string(policy.getResource().getDatabase()), + string(policy.getResource().getTable()), + string(policy.type().name()), + string(policy.getPrincipal()), + string(rowFilter == null ? null : rowFilter.getPredicate()), + string(columnMask == null ? null : columnMask.getOnColumn()), + string(columnMask == null ? null : columnMask.getTransform()), + string(page.getNextPageToken())); + } + return rows; + } + + private static StructField field( + String name, org.apache.spark.sql.types.DataType type, boolean nullable) { + return new StructField(name, type, nullable, Metadata.empty()); + } + + private static UTF8String string(String value) { + return value == null ? null : UTF8String.fromString(value); + } + + public static ProcedureBuilder builder() { + return new Builder() { + @Override + protected ListPoliciesProcedure doBuild() { + return new ListPoliciesProcedure(tableCatalog()); + } + }; + } + + @Override + public String description() { + return "ListPoliciesProcedure"; + } +} diff --git a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/RevokePermissionProcedure.java b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/RevokePermissionProcedure.java new file mode 100644 index 000000000000..516c8d1ae126 --- /dev/null +++ b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/RevokePermissionProcedure.java @@ -0,0 +1,96 @@ +/* + * 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. + */ + +package org.apache.paimon.spark.procedure; + +import org.apache.paimon.management.PermissionResource; +import org.apache.paimon.management.ResourceType; + +import org.apache.spark.sql.catalyst.InternalRow; +import org.apache.spark.sql.connector.catalog.TableCatalog; +import org.apache.spark.sql.types.DataTypes; +import org.apache.spark.sql.types.Metadata; +import org.apache.spark.sql.types.StructField; +import org.apache.spark.sql.types.StructType; + +import static org.apache.spark.sql.types.DataTypes.StringType; + +/** Revokes a permission by its resource identity. */ +public class RevokePermissionProcedure extends BasePermissionProcedure { + + private static final ProcedureParameter[] PARAMETERS = + new ProcedureParameter[] { + ProcedureParameter.required("resource_type", StringType), + ProcedureParameter.required("access", StringType), + ProcedureParameter.required("principal", StringType), + ProcedureParameter.optional("database", StringType), + ProcedureParameter.optional("table", StringType), + ProcedureParameter.optional("function", StringType), + ProcedureParameter.optional("view", StringType) + }; + + private static final StructType OUTPUT_TYPE = + new StructType( + new StructField[] { + new StructField("result", DataTypes.BooleanType, false, Metadata.empty()) + }); + + private RevokePermissionProcedure(TableCatalog tableCatalog) { + super(tableCatalog); + } + + @Override + public ProcedureParameter[] parameters() { + return PARAMETERS; + } + + @Override + public StructType outputType() { + return OUTPUT_TYPE; + } + + @Override + public InternalRow[] call(InternalRow args) { + ResourceType resourceType = + enumValue(args.getString(0), ResourceType.class, PARAMETERS[0].name()); + PermissionResource resource = + resource( + resourceType, + args.isNullAt(3) ? null : args.getString(3), + args.isNullAt(4) ? null : args.getString(4), + args.isNullAt(5) ? null : args.getString(5), + args.isNullAt(6) ? null : args.getString(6)); + + permissionManagement().revokePermission(resource, args.getString(1), args.getString(2)); + return new InternalRow[] {newInternalRow(true)}; + } + + public static ProcedureBuilder builder() { + return new Builder() { + @Override + protected RevokePermissionProcedure doBuild() { + return new RevokePermissionProcedure(tableCatalog()); + } + }; + } + + @Override + public String description() { + return "RevokePermissionProcedure"; + } +} diff --git a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/spark/sql/catalyst/parser/extensions/AbstractPaimonSparkSqlExtensionsParser.scala b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/spark/sql/catalyst/parser/extensions/AbstractPaimonSparkSqlExtensionsParser.scala index 7770143c8012..6cede203d4b0 100644 --- a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/spark/sql/catalyst/parser/extensions/AbstractPaimonSparkSqlExtensionsParser.scala +++ b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/spark/sql/catalyst/parser/extensions/AbstractPaimonSparkSqlExtensionsParser.scala @@ -71,6 +71,7 @@ abstract class AbstractPaimonSparkSqlExtensionsParser(val delegate: ParserInterf PaimonSqlExtensionsParser.TAG, PaimonSqlExtensionsParser.TRUE, PaimonSqlExtensionsParser.FALSE, + PaimonSqlExtensionsParser.ARRAY, PaimonSqlExtensionsParser.MAP, PaimonSqlExtensionsParser.COPY, PaimonSqlExtensionsParser.INTO, diff --git a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/PaimonSparkTestWithRestCatalogBase.scala b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/PaimonSparkTestWithRestCatalogBase.scala index b8b8d53e21b0..12b5c8febecc 100644 --- a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/PaimonSparkTestWithRestCatalogBase.scala +++ b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/PaimonSparkTestWithRestCatalogBase.scala @@ -30,7 +30,7 @@ import java.util.UUID class PaimonSparkTestWithRestCatalogBase extends PaimonSparkTestBase { - private var restCatalogServer: RESTCatalogServer = _ + protected var restCatalogServer: RESTCatalogServer = _ private var serverUrl: String = _ protected var warehouse: String = _ private val initToken = "init_token" @@ -43,13 +43,25 @@ class PaimonSparkTestWithRestCatalogBase extends PaimonSparkTestBase { "paimon", CatalogOptions.WAREHOUSE.key, warehouse), - ImmutableMap.of()) + ImmutableMap.of() + ) val authProvider = new BearTokenAuthProvider(initToken) restCatalogServer = new RESTCatalogServer(tempDBDir.getCanonicalPath, authProvider, config, warehouse) restCatalogServer.start() serverUrl = restCatalogServer.getUrl super.beforeAll() + Seq("analyst", "first", "second", "reader", "function_reader").foreach( + restCatalogServer.registerManagementPrincipal) + restCatalogServer.registerManagementPrincipal("analysts") + restCatalogServer.registerManagementPrincipal("admin") + spark.sql("CREATE DATABASE IF NOT EXISTS paimon.sales") + spark.sql("""CREATE TABLE IF NOT EXISTS paimon.sales.orders ( + | id INT, + | region STRING, + | email STRING) + |TBLPROPERTIES ('query-auth.enabled' = 'true') + |""".stripMargin) } override protected def afterAll(): Unit = { diff --git a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/procedure/PermissionProcedureTest.scala b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/procedure/PermissionProcedureTest.scala new file mode 100644 index 000000000000..3e34fe56f24b --- /dev/null +++ b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/procedure/PermissionProcedureTest.scala @@ -0,0 +1,787 @@ +/* + * 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. + */ + +package org.apache.paimon.spark.procedure + +import org.apache.paimon.catalog.Identifier +import org.apache.paimon.data.BinaryString +import org.apache.paimon.management.{PermissionResource, ResourceType} +import org.apache.paimon.predicate.{ConcatTransform, Equal, FieldRef, FieldTransform, LeafPredicate} +import org.apache.paimon.spark.{PaimonSparkTestBase, PaimonSparkTestWithRestCatalogBase} +import org.apache.paimon.types.DataTypes +import org.apache.paimon.utils.JsonSerdeUtil + +import org.apache.spark.sql.Row +import org.assertj.core.api.Assertions.assertThat + +import java.util.{Arrays, Collections} + +/** End-to-end tests for permission and policy management procedures. */ +class PermissionProcedureTest extends PaimonSparkTestWithRestCatalogBase { + + test("grant, list and idempotently revoke a permission") { + checkAnswer( + spark.sql("""CALL sys.grant_permission( + | resource_type => 'table', + | access => 'select', + | principal => 'analyst', + | database => 'sales', + | table => 'orders', + | expire_time => '2027-01-01T00:00:00Z') + |""".stripMargin), + Row(true) + ) + + checkAnswer( + spark.sql("""CALL sys.grant_permission( + | resource_type => 'TABLE', + | access => 'SELECT', + | principal => 'analyst', + | database => 'sales', + | table => 'orders', + | expire_time => '2028-01-01T00:00:00Z') + |""".stripMargin), + Row(true) + ) + + val listed = spark.sql("""CALL sys.list_permissions( + | resource_type => 'TABLE', + | database => 'sales', + | table => 'orders', + | principal => 'analyst') + |""".stripMargin) + assertThat(listed.columns).containsExactly( + "resource_type", + "database", + "table", + "function", + "view", + "access", + "principal", + "column_names", + "excluded_column_names", + "expire_time", + "next_page_token") + val assignments = listed.collect() + assertThat(assignments).hasSize(1) + val assignment = assignments.head + assertThat(assignment.getString(0)).isEqualTo("TABLE") + assertThat(assignment.getString(1)).isEqualTo("sales") + assertThat(assignment.getString(2)).isEqualTo("orders") + assertThat(assignment.getString(5)).isEqualTo("SELECT") + assertThat(assignment.getString(6)).isEqualTo("analyst") + assertThat(assignment.isNullAt(7)).isTrue + assertThat(assignment.isNullAt(8)).isTrue + assertThat(assignment.getString(9)).isEqualTo("2028-01-01T00:00:00Z") + assertThat(assignment.isNullAt(10)).isTrue + + val revoke = """CALL sys.revoke_permission( + | resource_type => 'TABLE', + | access => 'SELECT', + | principal => 'analyst', + | database => 'sales', + | table => 'orders') + |""".stripMargin + checkAnswer(spark.sql(revoke), Row(true)) + checkAnswer(spark.sql(revoke), Row(true)) + checkAnswer( + spark.sql("""CALL sys.list_permissions( + | resource_type => 'TABLE', + | database => 'sales', + | table => 'orders', + | principal => 'analyst') + |""".stripMargin), + Nil + ) + } + + test("list permissions supports opaque pagination tokens") { + grantCatalogPermission("first") + grantCatalogPermission("second") + + val first = spark + .sql("CALL sys.list_permissions(resource_type => 'CATALOG', max_results => 1)") + .head() + assertThat(first.getString(6)).isEqualTo("first") + val pageToken = first.getString(10) + assertThat(pageToken).isNotEmpty.isNotEqualTo("1") + + spark.sql( + "CALL sys.revoke_permission(resource_type => 'CATALOG', " + + "access => 'CREATEDATABASE', principal => 'first')") + + val second = spark + .sql( + "CALL sys.list_permissions(resource_type => 'CATALOG', max_results => 1, " + + s"page_token => ${sqlLiteral(pageToken)})" + ) + .head() + assertThat(second.getString(6)).isEqualTo("second") + assertThat(second.isNullAt(10)).isTrue + } + + test("grant and list explicit descendant scopes") { + checkAnswer( + spark.sql("""CALL sys.grant_permission( + | resource_type => 'CATALOG_ALL', + | access => 'SELECT', + | principal => 'analyst') + |""".stripMargin), + Row(true) + ) + val catalogAll = spark + .sql( + "CALL sys.list_permissions(resource_type => 'CATALOG_ALL', principal => 'analyst')" + ) + .head() + assertThat(catalogAll.getString(0)).isEqualTo("CATALOG_ALL") + assertThat(catalogAll.isNullAt(1)).isTrue + assertThat(catalogAll.getString(5)).isEqualTo("SELECT") + + checkAnswer( + spark.sql("""CALL sys.grant_permission( + | resource_type => 'DATABASE_ALL', + | database => 'sales', + | access => 'UPDATE', + | principal => 'analyst') + |""".stripMargin), + Row(true) + ) + val databaseAll = spark + .sql("""CALL sys.list_permissions( + | resource_type => 'DATABASE_ALL', + | database => 'sales', + | principal => 'analyst') + |""".stripMargin) + .head() + assertThat(databaseAll.getString(0)).isEqualTo("DATABASE_ALL") + assertThat(databaseAll.getString(1)).isEqualTo("sales") + assertThat(databaseAll.isNullAt(2)).isTrue + assertThat(databaseAll.getString(5)).isEqualTo("UPDATE") + } + + test("grant, replace, list and enforce column permissions") { + restCatalogServer.setQueryPrincipals(Collections.singleton("analyst")) + + checkAnswer( + spark.sql("""CALL sys.grant_permission( + | resource_type => 'COLUMN', + | access => 'SELECT', + | principal => 'analyst', + | database => 'sales', + | table => 'orders', + | column_names => array('id', 'region')) + |""".stripMargin), + Row(true) + ) + + val included = spark.sql("""CALL sys.list_permissions( + | resource_type => 'COLUMN', + | database => 'sales', + | table => 'orders', + | principal => 'analyst') + |""".stripMargin) + assertThat(included.columns).containsExactly( + "resource_type", + "database", + "table", + "function", + "view", + "access", + "principal", + "column_names", + "excluded_column_names", + "expire_time", + "next_page_token") + assertThat(included.head().getSeq[String](7)).isEqualTo(Seq("id", "region")) + assertThat(included.head().isNullAt(8)).isTrue + + checkAnswer(spark.sql("SELECT id, region FROM paimon.sales.orders"), Nil) + val deniedEmail = intercept[Exception] { + spark.sql("SELECT email FROM paimon.sales.orders").collect() + } + assertThat(deniedEmail.getMessage).contains("permission") + + checkAnswer( + spark.sql("""CALL sys.grant_permission( + | resource_type => 'COLUMN', + | access => 'SELECT', + | principal => 'analyst', + | database => 'sales', + | table => 'orders', + | excluded_column_names => array('region')) + |""".stripMargin), + Row(true) + ) + checkAnswer(spark.sql("SELECT id, email FROM paimon.sales.orders"), Nil) + val deniedRegion = intercept[Exception] { + spark.sql("SELECT region FROM paimon.sales.orders").collect() + } + assertThat(deniedRegion.getMessage).contains("permission") + + restCatalogServer.registerManagementPrincipal("limited") + restCatalogServer.setQueryPrincipals(new java.util.HashSet(Arrays.asList("analyst", "limited"))) + checkAnswer( + spark.sql("""CALL sys.grant_permission( + | resource_type => 'COLUMN', access => 'SELECT', + | principal => 'limited', database => 'sales', table => 'orders', + | column_names => array('id', 'region')) + |""".stripMargin), + Row(true) + ) + checkAnswer(spark.sql("SELECT id FROM paimon.sales.orders"), Nil) + val deniedByIntersection = intercept[Exception] { + spark.sql("SELECT email FROM paimon.sales.orders").collect() + } + assertThat(deniedByIntersection.getMessage).contains("permission") + + checkAnswer( + spark.sql("""CALL sys.revoke_permission( + | resource_type => 'COLUMN', access => 'SELECT', + | principal => 'analyst', database => 'sales', table => 'orders') + |""".stripMargin), + Row(true) + ) + } + + test("column permission validates query authorization and referenced columns") { + spark.sql("CREATE TABLE paimon.sales.disabled_columns (id INT)") + val disabled = intercept[Exception] { + spark + .sql("""CALL sys.grant_permission( + | resource_type => 'COLUMN', access => 'SELECT', + | principal => 'analyst', database => 'sales', table => 'disabled_columns', + | column_names => array('id')) + |""".stripMargin) + .collect() + } + assertThat(disabled.getMessage).contains("query-auth.enabled=true") + + val missing = intercept[Exception] { + spark + .sql("""CALL sys.grant_permission( + | resource_type => 'COLUMN', access => 'SELECT', + | principal => 'analyst', database => 'sales', table => 'orders', + | excluded_column_names => array('missing')) + |""".stripMargin) + .collect() + } + assertThat(missing.getMessage).contains("Permission column does not exist") + } + + test("column permissions follow table and schema lifecycle") { + restCatalogServer.setQueryPrincipals(Collections.singleton("analyst")) + spark.sql("""CREATE TABLE paimon.sales.column_lifecycle ( + | id INT, + | secret STRING) + |TBLPROPERTIES ('query-auth.enabled' = 'true') + |""".stripMargin) + spark.sql("INSERT INTO paimon.sales.column_lifecycle VALUES (1, 's1')") + checkAnswer( + spark.sql("""CALL sys.grant_permission( + | resource_type => 'COLUMN', access => 'SELECT', principal => 'analyst', + | database => 'sales', table => 'column_lifecycle', + | excluded_column_names => array('secret')) + |""".stripMargin), + Row(true) + ) + + spark.sql( + "ALTER TABLE paimon.sales.column_lifecycle RENAME TO paimon.sales.renamed_column_lifecycle") + var assignment = spark + .sql( + "CALL sys.list_permissions(resource_type => 'COLUMN', database => 'sales', " + + "table => 'renamed_column_lifecycle', principal => 'analyst')") + .head() + assertThat(assignment.getString(2)).isEqualTo("renamed_column_lifecycle") + assertThat(assignment.getSeq[String](8)).isEqualTo(Seq("secret")) + + val disableAuth = intercept[Exception] { + spark + .sql("ALTER TABLE paimon.sales.renamed_column_lifecycle " + + "SET TBLPROPERTIES ('query-auth.enabled' = 'false')") + .collect() + } + assertThat(disableAuth.getMessage).contains("Cannot disable query-auth.enabled") + + spark.sql( + "ALTER TABLE paimon.sales.renamed_column_lifecycle " + + "RENAME COLUMN secret TO private_secret") + assignment = spark + .sql( + "CALL sys.list_permissions(resource_type => 'COLUMN', database => 'sales', " + + "table => 'renamed_column_lifecycle', principal => 'analyst')") + .head() + assertThat(assignment.getSeq[String](8)).isEqualTo(Seq("private_secret")) + val denied = intercept[Exception] { + spark.sql("SELECT private_secret FROM paimon.sales.renamed_column_lifecycle").collect() + } + assertThat(denied.getMessage).contains("permission") + + spark.sql("""CALL sys.grant_permission( + | resource_type => 'COLUMN', access => 'SELECT', principal => 'analyst', + | database => 'sales', table => 'renamed_column_lifecycle', + | column_names => array('private_secret')) + |""".stripMargin) + val dropOnlyAllowedColumn = intercept[Exception] { + spark + .sql("ALTER TABLE paimon.sales.renamed_column_lifecycle DROP COLUMN private_secret") + .collect() + } + assertThat(dropOnlyAllowedColumn.getMessage).contains("Cannot drop every allowed column") + + spark.sql("""CALL sys.grant_permission( + | resource_type => 'COLUMN', access => 'SELECT', principal => 'analyst', + | database => 'sales', table => 'renamed_column_lifecycle', + | excluded_column_names => array('private_secret')) + |""".stripMargin) + spark.sql("ALTER TABLE paimon.sales.renamed_column_lifecycle DROP COLUMN private_secret") + checkAnswer( + spark.sql( + "CALL sys.list_permissions(resource_type => 'COLUMN', database => 'sales', " + + "table => 'renamed_column_lifecycle', principal => 'analyst')"), + Nil + ) + + grantTablePermission("renamed_column_lifecycle", "analyst") + spark.sql("DROP TABLE paimon.sales.renamed_column_lifecycle") + spark.sql("CREATE TABLE paimon.sales.renamed_column_lifecycle (id INT)") + checkAnswer( + spark.sql( + "CALL sys.list_permissions(resource_type => 'TABLE', database => 'sales', " + + "table => 'renamed_column_lifecycle', principal => 'analyst')"), + Nil + ) + } + + test("create, reject duplicates, apply and idempotently drop table data policies") { + restCatalogServer.setQueryPrincipals(Collections.singleton("analyst")) + spark.sql( + "INSERT OVERWRITE paimon.sales.orders VALUES " + + "(1, 'APAC', 'apac@example.com'), (2, 'EMEA', 'emea@example.com')") + + val apacFilter = stringEqualsPredicate(1, "region", "APAC") + val emeaFilter = stringEqualsPredicate(1, "region", "EMEA") + val emailMask = concatFieldTransform(1, "region", "-masked") + + checkAnswer( + spark.sql(s"""CALL sys.create_policy( + | database => 'sales', + | table => 'orders', + | policy_type => 'ROW_FILTER', + | principal => 'analyst', + | predicate_json => ${sqlLiteral(apacFilter)}) + |""".stripMargin), + Row(true) + ) + + val listed = spark.sql("""CALL sys.list_policies( + | database => 'sales', + | table => 'orders', + | policy_type => 'ROW_FILTER', + | principal => 'analyst') + |""".stripMargin) + assertThat(listed.columns).containsExactly( + "database", + "table", + "policy_type", + "principal", + "predicate_json", + "on_column", + "transform_json", + "next_page_token") + val direct = listed.head() + assertThat(direct.getString(2)).isEqualTo("ROW_FILTER") + assertThat(direct.getString(3)).isEqualTo("analyst") + assertThat(direct.getString(4)).contains("\"name\":\"region\"") + assertThat(direct.isNullAt(5)).isTrue + assertThat(direct.isNullAt(6)).isTrue + checkAnswer( + spark.sql("SELECT id, region, email FROM paimon.sales.orders ORDER BY id"), + Row(1, "APAC", "apac@example.com") + ) + + val duplicate = intercept[Exception] { + spark + .sql(s"""CALL sys.create_policy( + | database => 'sales', + | table => 'orders', + | policy_type => 'ROW_FILTER', + | principal => 'analyst', + | predicate_json => ${sqlLiteral(emeaFilter)}) + |""".stripMargin) + .collect() + } + assertThat(duplicate.getMessage).contains("already exists") + val unchanged = spark + .sql("""CALL sys.list_policies( + | database => 'sales', + | table => 'orders', + | policy_type => 'ROW_FILTER', + | principal => 'analyst') + |""".stripMargin) + .head() + assertThat(unchanged.getString(4)).contains("APAC").doesNotContain("EMEA") + + checkAnswer( + spark.sql(s"""CALL sys.create_policy( + | database => 'sales', + | table => 'orders', + | policy_type => 'COLUMN_MASKING', + | principal => 'analyst', + | on_column => 'email', + | transform_json => ${sqlLiteral(emailMask)}) + |""".stripMargin), + Row(true) + ) + + val mask = spark + .sql("""CALL sys.list_policies( + | database => 'sales', + | table => 'orders', + | policy_type => 'COLUMN_MASKING') + |""".stripMargin) + .head() + assertThat(mask.getString(2)).isEqualTo("COLUMN_MASKING") + assertThat(mask.getString(3)).isEqualTo("analyst") + assertThat(mask.isNullAt(4)).isTrue + assertThat(mask.getString(5)).isEqualTo("email") + assertThat(mask.getString(6)).contains("-masked") + + checkAnswer( + spark.sql("SELECT id, region, email FROM paimon.sales.orders ORDER BY id"), + Row(1, "APAC", "APAC-masked") + ) + checkAnswer( + spark.sql("SELECT email FROM paimon.sales.orders"), + Row("APAC-masked") + ) + checkAnswer( + spark.sql("""CALL sys.drop_policy( + | database => 'sales', + | table => 'orders', + | policy_type => 'ROW_FILTER', + | principal => 'analyst', + | if_exists => true) + |""".stripMargin), + Row(true) + ) + checkAnswer( + spark.sql("""CALL sys.drop_policy( + | database => 'sales', + | table => 'orders', + | policy_type => 'ROW_FILTER', + | principal => 'analyst', + | if_exists => true) + |""".stripMargin), + Row(true) + ) + checkAnswer( + spark.sql("""CALL sys.list_policies( + | database => 'sales', + | table => 'orders', + | policy_type => 'ROW_FILTER', + | principal => 'analyst') + |""".stripMargin), + Nil + ) + spark.sql( + "CALL sys.drop_policy(database => 'sales', table => 'orders', " + + "policy_type => 'COLUMN_MASKING', " + + "principal => 'analyst', column => 'email', if_exists => true)") + } + + test("policy creation validates query authorization, JSON and columns") { + val validFilter = intEqualsPredicate(0, "id", 1) + spark.sql("CREATE TABLE paimon.sales.disabled_orders (id INT)") + + val disabled = intercept[Exception] { + spark + .sql(s"""CALL sys.create_policy( + | database => 'sales', table => 'disabled_orders', + | policy_type => 'ROW_FILTER', predicate_json => ${sqlLiteral(validFilter)}, + | principal => 'analyst') + |""".stripMargin) + .collect() + } + assertThat(disabled.getMessage).contains("query-auth.enabled=true") + + val malformed = intercept[Exception] { + spark + .sql("""CALL sys.create_policy( + | database => 'sales', table => 'orders', + | policy_type => 'ROW_FILTER', predicate_json => '{bad', + | principal => 'analyst') + |""".stripMargin) + .collect() + } + assertThat(malformed.getMessage).contains("Unexpected character") + + val unknownColumnFilter = stringEqualsPredicate(0, "unknown", "APAC") + val missingColumn = intercept[Exception] { + spark + .sql(s"""CALL sys.create_policy( + | database => 'sales', table => 'orders', + | policy_type => 'ROW_FILTER', predicate_json => ${sqlLiteral(unknownColumnFilter)}, + | principal => 'analyst') + |""".stripMargin) + .collect() + } + assertThat(missingColumn.getMessage).contains("column unknown") + + val missingPrincipal = intercept[Exception] { + spark + .sql(s"""CALL sys.create_policy( + | database => 'sales', table => 'orders', + | policy_type => 'ROW_FILTER', predicate_json => ${sqlLiteral(validFilter)}, + | principal => 'missing') + |""".stripMargin) + .collect() + } + assertThat(missingPrincipal.getMessage).contains("principal does not exist") + + val jsonNull = intercept[Exception] { + spark + .sql("""CALL sys.create_policy( + | database => 'sales', table => 'orders', + | policy_type => 'ROW_FILTER', predicate_json => 'null', + | principal => 'analyst') + |""".stripMargin) + .collect() + } + assertThat(jsonNull.getMessage).contains("JSON null") + + val mixedDefinition = intercept[Exception] { + spark + .sql(s"""CALL sys.create_policy( + | database => 'sales', table => 'orders', + | policy_type => 'ROW_FILTER', predicate_json => ${sqlLiteral(validFilter)}, + | transform_json => ${sqlLiteral(constantStringTransform("****"))}, + | principal => 'analyst') + |""".stripMargin) + .collect() + } + assertThat(mixedDefinition.getMessage).contains("cannot specify transform") + } + + test("list policies supports opaque pagination tokens") { + val filter = intEqualsPredicate(0, "id", 1) + Seq("first", "second").foreach { + principal => + checkAnswer( + spark.sql(s"""CALL sys.create_policy( + | database => 'sales', table => 'orders', + | policy_type => 'ROW_FILTER', principal => '$principal', + | predicate_json => ${sqlLiteral(filter)}) + |""".stripMargin), + Row(true) + ) + } + try { + val first = spark + .sql("CALL sys.list_policies(database => 'sales', table => 'orders', max_results => 1)") + .head() + assertThat(first.getString(3)).isEqualTo("first") + val pageToken = first.getString(7) + assertThat(pageToken).isNotEmpty.isNotEqualTo("1") + + spark.sql( + "CALL sys.drop_policy(database => 'sales', table => 'orders', " + + "policy_type => 'ROW_FILTER', principal => 'first', if_exists => true)") + + val second = spark + .sql( + "CALL sys.list_policies(database => 'sales', table => 'orders', " + + s"max_results => 1, page_token => ${sqlLiteral(pageToken)})") + .head() + assertThat(second.getString(3)).isEqualTo("second") + assertThat(second.isNullAt(7)).isTrue + } finally { + Seq("first", "second").foreach { + principal => + spark.sql( + "CALL sys.drop_policy(database => 'sales', table => 'orders', " + + s"policy_type => 'ROW_FILTER', principal => '$principal', if_exists => true)") + } + } + } + + test("table lifecycle preserves policy enforcement and rejects unsafe schema changes") { + restCatalogServer.setQueryPrincipals(Collections.singleton("analyst")) + spark.sql("""CREATE TABLE paimon.sales.lifecycle_orders ( + | id INT, + | region STRING) + |TBLPROPERTIES ('query-auth.enabled' = 'true') + |""".stripMargin) + spark.sql("INSERT INTO paimon.sales.lifecycle_orders VALUES (1, 'APAC'), (2, 'EMEA')") + val lifecycleFilter = stringEqualsPredicate(1, "region", "APAC") + checkAnswer( + spark.sql(s"""CALL sys.create_policy( + | database => 'sales', table => 'lifecycle_orders', + | policy_type => 'ROW_FILTER', predicate_json => ${sqlLiteral(lifecycleFilter)}, + | principal => 'analyst') + |""".stripMargin), + Row(true) + ) + + spark.sql("ALTER TABLE paimon.sales.lifecycle_orders RENAME TO paimon.sales.renamed_orders") + assertThat( + spark + .sql("CALL sys.list_policies(database => 'sales', table => 'renamed_orders', " + + "policy_type => 'ROW_FILTER', principal => 'analyst')") + .head() + .getString(1)).isEqualTo("renamed_orders") + assertThat( + paimonCatalog + .authTableQuery(Identifier.create("sales", "renamed_orders"), null) + .extractPredicate()).isNotNull + + val disableAuth = intercept[Exception] { + spark + .sql("ALTER TABLE paimon.sales.renamed_orders " + + "SET TBLPROPERTIES ('query-auth.enabled' = 'false')") + .collect() + } + assertThat(disableAuth.getMessage).contains("Cannot disable query-auth.enabled") + + val renameColumn = intercept[Exception] { + spark + .sql("ALTER TABLE paimon.sales.renamed_orders RENAME COLUMN region TO area") + .collect() + } + assertThat(renameColumn.getMessage).contains("column region") + + spark.sql("DROP TABLE paimon.sales.renamed_orders") + spark.sql("""CREATE TABLE paimon.sales.renamed_orders ( + | id INT, + | region STRING) + |TBLPROPERTIES ('query-auth.enabled' = 'true') + |""".stripMargin) + spark.sql("INSERT INTO paimon.sales.renamed_orders VALUES (1, 'APAC'), (2, 'EMEA')") + checkAnswer( + spark.sql("SELECT id, region FROM paimon.sales.renamed_orders ORDER BY id"), + Seq(Row(1, "APAC"), Row(2, "EMEA")) + ) + checkAnswer( + spark.sql( + "CALL sys.list_policies(database => 'sales', table => 'renamed_orders', " + + "policy_type => 'ROW_FILTER', principal => 'analyst')"), + Nil + ) + assertThat( + paimonCatalog + .authTableQuery(Identifier.create("sales", "renamed_orders"), null) + .extractPredicate()).isNull + } + + test("management endpoints enforce target authorization") { + val resource = + new PermissionResource(ResourceType.TABLE, "sales", "orders", null, null) + restCatalogServer.denyManagementPermission(resource) + try { + val permissionError = intercept[Exception] { + spark + .sql("""CALL sys.grant_permission( + | resource_type => 'TABLE', access => 'SELECT', + | principal => 'analyst', + | database => 'sales', table => 'orders') + |""".stripMargin) + .collect() + } + assertThat(permissionError.getMessage).contains("cannot manage permissions") + + val policyError = intercept[Exception] { + spark + .sql("CALL sys.list_policies(database => 'sales', table => 'orders')") + .collect() + } + assertThat(policyError.getMessage).contains("cannot manage permissions") + } finally { + restCatalogServer.allowManagementPermission(resource) + } + } + + private def grantCatalogPermission(principal: String): Unit = { + checkAnswer( + spark.sql(s"""CALL sys.grant_permission( + | resource_type => 'CATALOG', + | access => 'CREATEDATABASE', + | principal => '$principal') + |""".stripMargin), + Row(true) + ) + } + + private def grantTablePermission(table: String, principal: String): Unit = { + checkAnswer( + spark.sql(s"""CALL sys.grant_permission( + | resource_type => 'TABLE', access => 'SELECT', principal => '$principal', + | database => 'sales', table => '$table') + |""".stripMargin), + Row(true) + ) + } + + private def stringEqualsPredicate(index: Int, column: String, constant: String): String = { + JsonSerdeUtil.toFlatJson( + LeafPredicate.of( + new FieldTransform(new FieldRef(index, column, DataTypes.STRING())), + Equal.INSTANCE, + Collections.singletonList(BinaryString.fromString(constant)))) + } + + private def intEqualsPredicate(index: Int, column: String, constant: Int): String = { + JsonSerdeUtil.toFlatJson( + LeafPredicate.of( + new FieldTransform(new FieldRef(index, column, DataTypes.INT())), + Equal.INSTANCE, + Collections.singletonList(Integer.valueOf(constant)))) + } + + private def constantStringTransform(constant: String): String = { + JsonSerdeUtil.toFlatJson( + new ConcatTransform(Collections.singletonList(BinaryString.fromString(constant)))) + } + + private def concatFieldTransform(index: Int, column: String, suffix: String): String = { + JsonSerdeUtil.toFlatJson( + new ConcatTransform(Arrays + .asList(new FieldRef(index, column, DataTypes.STRING()), BinaryString.fromString(suffix)))) + } + + private def sqlLiteral(value: String): String = { + "'" + value.replace("'", "''") + "'" + } +} + +/** Management procedures must fail clearly for catalogs without the REST capability. */ +class PermissionProcedureUnsupportedCatalogTest extends PaimonSparkTestBase { + + test("filesystem catalog does not expose permission management") { + val error = intercept[IllegalArgumentException] { + spark + .sql("""CALL sys.grant_permission( + | resource_type => 'CATALOG', + | access => 'CREATEDATABASE', + | principal => 'admin') + |""".stripMargin) + .collect() + } + assertThat(error.getMessage).contains("does not support permission or policy management") + } +} diff --git a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/CatalogQualifiedCreateTableLikeTest.scala b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/CatalogQualifiedCreateTableLikeTest.scala index c4eb2cd6443a..350462fc1ca0 100644 --- a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/CatalogQualifiedCreateTableLikeTest.scala +++ b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/CatalogQualifiedCreateTableLikeTest.scala @@ -139,6 +139,11 @@ class CatalogQualifiedCreateTableLikeTest extends PaimonSparkTestBase { Assertions.assertEquals("skip_file", skipFileCommand.targetIdent.name()) Assertions.assertEquals(Seq("test"), skipFileCommand.targetIdent.namespace().toSeq) + val arrayCommand = + parseCreateTableLikeCommand("CREATE TABLE paimon.test.array LIKE paimon.test.source_tbl") + Assertions.assertEquals("array", arrayCommand.targetIdent.name()) + Assertions.assertEquals(Seq("test"), arrayCommand.targetIdent.namespace().toSeq) + val nestedIdentifierCommand = parseCreateTableLikeCommand( "CREATE TABLE paimon.test.extra.target_tbl LIKE paimon.test.extra.source_tbl")