[flink] Support grant, revoke and list_permissions procedures

[Stephen0421]

Purpose

Add the Flink procedures sys.grant_permission, sys.revoke_permission, and sys.list_permissions, matching the existing Spark procedures.

They call RESTCatalog.permissionManagement() and do not change the REST API or the Catalog interface. A filesystem catalog fails with Catalog does not support permission management.

These procedures require Flink 1.19 or later. Column ranges are passed as comma-separated names, for example column_names => 'order_id,region'. list_permissions still returns column_names and excluded_column_names as ARRAY<STRING>. A column name that itself contains a comma cannot be represented.

CALL sys.grant_permission(
  resource_type => 'TABLE',
  `database` => 'sales',
  `table` => 'orders',
  access => 'SELECT',
  principal => 'user:alice'
);

Test

• PermissionProcedureITCase: grant, replace, list, and idempotent revoke; pagination; CATALOG_ALL and DATABASE_ALL; column allowlist and denylist; positional 8-argument and 10-argument calls
• Invalid resource_type, empty principal, both column lists, and column names on a non-COLUMN resource are rejected
• PermissionProcedureUnsupportedCatalogITCase: a filesystem catalog is rejected

[JingsongLi]

[P2] Preserve column identity before granting permissions

BasePermissionProcedure.parseColumnNames trims every token before sending it to the REST catalog. Column names are exact identifiers in PermissionColumns and the REST permission contract, so this can successfully grant a different column range than the caller requested.

I reproduced this through actual Flink 1.20 SQL and the REST catalog test server:

CREATE TABLE mydb.space_columns (`secret` STRING, ` secret ` STRING)
WITH ('query-auth.enabled' = 'true');

CALL sys.grant_permission(
  resource_type => 'COLUMN', access => 'SELECT', principal => 'analyst',
  `database` => 'mydb', `table` => 'space_columns',
  excluded_column_names => ' secret '
);

DESCRIBE preserves the second column as secret, but sys.list_permissions returns excluded_column_names = ['secret']. The operation reports success while excluding the other column. Under the existing exact-name denylist contract, the requested spaced column remains allowed. The allowlist path has the corresponding wrong-target problem.

Please use an input representation that preserves exact names, or reject inputs that cannot be represented without changing their identity before making the grant. Documenting only the comma limitation does not cover this silent rewrite.

Validation: the seven existing permission procedure integration cases pass with standard JDK 8 Maven checks. An additional real SQL identity assertion fails (['secret'] versus [' secret ']). Independent review confirmed that the REST DTO and server preserve exact column names. No production permissions were changed.

[Stephen0421]

[P2] Preserve column identity before granting permissions

BasePermissionProcedure.parseColumnNames trims every token before sending it to the REST catalog. Column names are exact identifiers in PermissionColumns and the REST permission contract, so this can successfully grant a different column range than the caller requested.

I reproduced this through actual Flink 1.20 SQL and the REST catalog test server:

CREATE TABLE mydb.space_columns (`secret` STRING, ` secret ` STRING)
WITH ('query-auth.enabled' = 'true');

CALL sys.grant_permission(
  resource_type => 'COLUMN', access => 'SELECT', principal => 'analyst',
  `database` => 'mydb', `table` => 'space_columns',
  excluded_column_names => ' secret '
);

DESCRIBE preserves the second column as secret, but sys.list_permissions returns excluded_column_names = ['secret']. The operation reports success while excluding the other column. Under the existing exact-name denylist contract, the requested spaced column remains allowed. The allowlist path has the corresponding wrong-target problem.

Please use an input representation that preserves exact names, or reject inputs that cannot be represented without changing their identity before making the grant. Documenting only the comma limitation does not cover this silent rewrite.

Validation: the seven existing permission procedure integration cases pass with standard JDK 8 Maven checks. An additional real SQL identity assertion fails (['secret'] versus [' secret ']). Independent review confirmed that the REST DTO and server preserve exact column names. No production permissions were changed.

Column names are now a single JSON array of exact strings, for example column_names => '["order_id", "region"]'. Each element is passed through unchanged, including leading or trailing spaces and embedded commas. excluded_column_names => '[" secret "]' therefore excludes secret rather than secret.

Flink procedure arguments have to be literals, so these parameters stay STRING instead of ARRAY. An ARRAY argument fails while parsing the call. The value must be one complete JSON array: parsing uses FAIL_ON_TRAILING_TOKENS, so ["a"],["b"] is rejected before the grant and does not drop the second array.

PermissionProcedureITCase covers the spaced-name round trip, and rejects both the old bare string ' secret ' and a trailing second array.