plusplusjiajia commented on code in PR #10503:
URL: https://github.com/apache/paimon/pull/10503#discussion_r4266649220
##########
paimon-python/pypaimon/tests/rest/rest_server.py:
##########
@@ -587,6 +716,189 @@ def _handle_table_resource(self, method: str, path_parts:
List[str],
return self._mock_response(ErrorResponse(None, None, "Not Found",
404), 404)
return self._mock_response(ErrorResponse(None, None, "Not Found",
404), 404)
+ def set_query_principals(self, *principals: str) -> None:
+ self.query_principals = set(principals)
+
+ def _permissions_handle(self, method: str, resource_path: str, data: str,
+ parameters: Dict[str, str]) -> Tuple[str, int]:
+ if method == "GET" and resource_path ==
self.resource_paths.permissions():
+ target = PermissionResource(
+ parameters.get("resourceType"), parameters.get("database"),
+ parameters.get("table"), parameters.get("function"),
parameters.get("view"))
+ missing = self._missing_permission_resource(target)
+ if missing is not None:
+ return missing
+ matched = [assignment for assignment in
self.permission_store.values()
+ if assignment.get_resource() == target
+ and _matches(parameters, "principal",
assignment.get_principal())
+ and _matches(parameters, "access",
assignment.get_access())]
+ return self._mock_response(
+ ListPermissionsResponse(sorted(matched, key=_assignment_key)),
200)
+ if method == "POST" and resource_path ==
self.resource_paths.grant_permission():
+ assignment = JSON.from_json(data,
GrantPermissionRequest).assignment()
+ missing =
self._missing_permission_resource(assignment.get_resource())
+ if missing is not None:
+ return missing
+ invalid = self._invalid_column_grant(assignment)
+ if invalid is not None:
+ return invalid
+ self.permission_store[_assignment_key(assignment)] = assignment
+ return self._mock_response("", 200)
+ if method == "POST" and resource_path ==
self.resource_paths.revoke_permission():
+ request = JSON.from_json(data, RevokePermissionRequest)
+ missing = self._missing_permission_resource(request.get_resource())
+ if missing is not None:
+ return missing
+ self.permission_store.pop(_permission_key(
+ request.get_resource(), request.get_access(),
request.get_principal()), None)
+ return self._mock_response("", 200)
+ return self._mock_response(ErrorResponse(None, None, "Not Found",
404), 404)
+
+ def _missing_permission_resource(
+ self, resource: PermissionResource) -> Optional[Tuple[str, int]]:
+ resource_type = resource.get_type()
+ database = resource.get_database()
+ if resource_type in (ResourceType.CATALOG, ResourceType.CATALOG_ALL):
+ return None
+ exists = database in self.database_store
+ if resource_type in (ResourceType.TABLE, ResourceType.COLUMN):
+ exists = exists and Identifier.create(
+ database, resource.get_table()).get_full_name() in
self.table_metadata_store
+ elif resource_type == ResourceType.FUNCTION:
+ exists = exists and "{}.{}".format(
+ database, resource.get_function()) in self.function_store
+ elif resource_type == ResourceType.VIEW:
+ exists = False
+ if exists:
+ return None
+ return self._mock_response(ErrorResponse(
+ resource_type.name, _resource_name(resource),
+ "Permission resource does not exist.", 404), 404)
+
+ def _invalid_column_grant(self, assignment: PermissionAssignment) ->
Optional[Tuple[str, int]]:
+ resource = assignment.get_resource()
+ if resource.get_type() != ResourceType.COLUMN:
+ return None
+ schema = self.table_metadata_store[
+ Identifier.create(resource.get_database(),
resource.get_table()).get_full_name()].schema
+ if not _query_auth_enabled(schema):
+ return self._mock_response(ErrorResponse(
+ ErrorResponse.RESOURCE_TYPE_TABLE, _resource_name(resource),
+ "Column permissions require query-auth.enabled=true.", 409),
409)
+ names = {field.name for field in schema.fields}
+ for column in _granted_columns(assignment):
+ if column not in names:
+ return self._mock_response(ErrorResponse(
+ ErrorResponse.RESOURCE_TYPE_COLUMN, column,
+ "Permission column does not exist.", 404), 404)
+ return None
+
+ def _column_grants(self, identifier: Identifier) ->
List[PermissionAssignment]:
+ return [assignment for assignment in self.permission_store.values()
+ if assignment.get_resource().get_type() == ResourceType.COLUMN
+ and _on_table(assignment.get_resource(), identifier)]
+
+ def _policies_handle(self, method: str, data: str, identifier: Identifier,
+ parameters: Dict[str, str], drop: bool) -> Tuple[str,
int]:
+ resource = PermissionResource(
+ ResourceType.TABLE, identifier.get_database_name(),
identifier.get_table_name())
+ metadata = self.table_metadata_store.get(identifier.get_full_name())
+ if metadata is None:
+ return self._missing_permission_resource(resource)
+ policies = self._table_policies(identifier, metadata)
+ if method == "GET" and not drop:
+ matched = [DataPolicy(resource, policy.get_row_filter(),
policy.get_column_mask(),
+ policy.get_principal())
+ for policy in policies.values() if
_matches_policy(policy, parameters)]
+ return self._mock_response(ListPoliciesResponse(sorted(matched,
key=_policy_key)), 200)
+ if method == "POST" and not drop:
+ policy = JSON.from_json(data, PolicyRequest).policy(resource)
+ if not _query_auth_enabled(metadata.schema):
+ return self._mock_response(ErrorResponse(
+ ErrorResponse.RESOURCE_TYPE_TABLE,
identifier.get_full_name(),
+ "Data policies require the target table option
query-auth.enabled=true.",
+ 409), 409)
+ identity = _policy_identity(policy)
+ missing = _policy_columns(policy) - {field.name for field in
metadata.schema.fields}
Review Comment:
Thanks for the careful check! To keep this PR focused, it now covers only
permissions and the COLUMN grant check of AuthTableQuery, and the data policy
endpoints move to a follow-up PR. There a column mask must return its column's
type, your YEAR-on-`email` case is a rejection test, and the SQL tests use a
STRING mask on `email` and `NULL` on a BOOLEAN column.
##########
paimon-python/pypaimon/tests/rest/rest_server.py:
##########
@@ -197,6 +213,190 @@ def _dict_to_schema_change(change_dict: dict) ->
SchemaChange:
raise ValueError(f"Unknown schema change action: {action}")
+def _cursor_key(*parts: Optional[str]) -> str:
+ return "".join("-1:" if part is None else "{}:{}".format(len(part), part)
for part in parts)
+
+
+def _permission_key(resource: PermissionResource, access: str, principal: str)
-> str:
+ return _cursor_key(resource.get_type().name, resource.get_database(),
resource.get_table(),
+ resource.get_function(), resource.get_view(),
+ PermissionAccess.canonicalize_for(resource, access),
principal)
+
+
+def _assignment_key(assignment: PermissionAssignment) -> str:
+ return _permission_key(
+ assignment.get_resource(), assignment.get_access(),
assignment.get_principal())
+
+
+def _policy_identity(policy: DataPolicy) -> Tuple[str, str, Optional[str]]:
+ mask = policy.get_column_mask()
+ return policy.type().name, policy.get_principal(), None if mask is None
else mask.get_on_column()
+
+
+def _policy_key(policy: DataPolicy) -> str:
+ return _cursor_key(*_policy_identity(policy))
+
+
+def _policy_name(policy_type: str, principal: str, column: Optional[str]) ->
str:
+ return ":".join(part for part in (policy_type, principal, column) if part
is not None)
+
+
+def _resource_name(resource: PermissionResource) -> str:
+ name = resource.get_table() or resource.get_function() or
resource.get_view()
+ return ".".join(part for part in (resource.get_database(), name) if part)
or "catalog"
+
+
+def _transform_columns(transform: Dict[str, Any]) -> set:
+ refs = list(transform.get("inputs") or []) + [transform.get("fieldRef")]
+ return {ref["name"] for ref in refs if isinstance(ref, dict)}
+
+
+def _leaf_transforms(predicate: Dict[str, Any]) -> List[Dict[str, Any]]:
+ if predicate["kind"] == "COMPOUND":
+ children = predicate.get("children")
+ if not predicate.get("function") or not children or None in children:
+ raise ValueError("Compound predicate needs a function and at least
one predicate.")
+ return [transform for child in children for transform in
_leaf_transforms(child)]
+ return [predicate["transform"]]
+
+
+def _policy_columns(policy: DataPolicy) -> set:
+ mask = policy.get_column_mask()
+ if mask is None:
+ predicate = json.loads(policy.get_row_filter().get_predicate())
+ return set().union(*map(_transform_columns,
_leaf_transforms(predicate)))
+ return {mask.get_on_column()} |
_transform_columns(json.loads(mask.get_transform()))
+
+
+_DATE_PARTS = ("YEAR", "QUARTER", "MONTH", "WEEK", "YEAR_OF_WEEK", "DAY",
"DAY_OF_WEEK", "ISO_DAY_OF_WEEK",
+ "WEEKDAY", "DAY_OF_YEAR", "HOUR", "MINUTE", "SECOND")
+_TRANSFORM_INPUTS = {
+ "CONCAT": ("s", 0, None, "STRING"), "CONCAT_WS": ("s", 2, None, "STRING"),
+ "UPPER": ("s", 1, 1, "STRING"), "LOWER": ("s", 1, 1, "STRING"), "TRIM":
("s", 1, 2, "STRING"),
+ "TRANSLATE": ("s", 3, 3, "STRING"), "SUBSTRING": ("a", 2, 3, "STRING"),
+ "OVERLAY": ("ssii", 3, 4, "STRING"), "PAD": ("sis", 3, 3, "STRING"),
+ "LENGTH": ("s", 1, 1, "INT"), "BIT_LENGTH": ("s", 1, 1, "INT"),
+ "DATE_ADD": ("di", 2, 2, "DATE"), "DATE_DIFF": ("dd", 2, 2, "INT"),
"DATE_TRUNC": ("ds", 2, 2, "DATE"),
+}
+_TRANSFORM_FLAGS = {"TRIM": ("trimFlag", ("LEADING", "TRAILING", "BOTH")),
"PAD": ("direction", ("LEFT", "RIGHT"))}
+_INPUT_ROOTS = {"s": casts.CHARACTER_STRING, "i": casts.INTEGER_NUMERIC, "d":
{casts.DATE}}
+_TYPE_DEFAULTS = {casts.CHAR: (1,), casts.VARCHAR: (1,), casts.BINARY: (1,),
casts.VARBINARY: (1,),
+ casts.DECIMAL: (10, 0), casts.TIME: (0,), casts.TIMESTAMP:
(6,), casts.TIMESTAMP_LTZ: (6,)}
+
+
+def _logical_type(data_type: Optional[DataType]) -> Any:
+ if not isinstance(data_type, AtomicType):
+ return data_type
+ spelling = data_type.type.upper()
+ root = casts.TIMESTAMP_LTZ if "LOCAL TIME ZONE" in spelling else
casts._root(data_type)
+ found = (2147483647,) if spelling in ("STRING", "BYTES") else
tuple(map(int, re.findall(r"\d+", spelling)))
+ return root, found + _TYPE_DEFAULTS.get(root, ())[len(found):],
data_type.nullable
+
+
+def _input_fits(kind: str, value: Any, columns: Dict[str, DataType]) -> bool:
+ if value is None or kind == "a":
+ return True
+ if isinstance(value, dict):
+ return casts._root(columns[value["name"]]) in _INPUT_ROOTS[kind]
+ if kind == "s":
+ return isinstance(value, str)
+ return isinstance(value, int) and not isinstance(value, bool)
+
+
+def _transform_type(transform: Dict[str, Any], columns: Dict[str, DataType])
-> Optional[DataType]:
+ name = transform["name"]
+ if "fieldRef" in transform:
+ source = columns[transform["fieldRef"]["name"]]
+ if name == "CAST":
+ target = DataTypeParser.parse_data_type(transform["type"])
+ return target if casts.supports_cast(source, target) else None
+ if name in _DATE_PARTS:
+ return AtomicType("INT") if casts._root(source) in (casts.DATE,
casts.TIMESTAMP) else None
+ for part in transform.get("path") or []:
+ source = next((field.type for field in getattr(source, "fields",
()) if field.name == part), None)
+ nested = name == "NESTED_FIELD_REF" and transform.get("path")
+ return source if name == "FIELD_REF" or nested else None
+ if name == "NULL":
+ return AtomicType("BOOLEAN")
+ if name in _TRANSFORM_FLAGS:
+ flag, values = _TRANSFORM_FLAGS[name]
+ if transform.get(flag) not in values:
+ return None
+ kinds, low, high, output = _TRANSFORM_INPUTS.get(name, ("", 1, 0, None))
+ inputs = transform.get("inputs") or []
+ if not low <= len(inputs) <= (len(inputs) if high is None else high):
+ return None
+ fits = all(_input_fits(kinds[min(position, len(kinds) - 1)], value,
columns)
+ for position, value in enumerate(inputs))
+ return AtomicType(output) if fits else None
+
+
+def _policy_type_error(policy: DataPolicy, schema: TableSchema) ->
Optional[str]:
+ columns = {field.name: field.type for field in schema.fields}
+ mask = policy.get_column_mask()
+ if mask is None:
+ predicate = json.loads(policy.get_row_filter().get_predicate())
+ invalid = [transform["name"] for transform in
_leaf_transforms(predicate)
+ if _transform_type(transform, columns) is None]
+ return "Row filter transform {} does not fit the table
schema.".format(invalid[0]) if invalid else None
Review Comment:
Thanks! This moves to the same follow-up PR, where predicate functions are
checked against Java's leaf and compound functions and an unknown function is a
rejection test.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
