JingsongLi commented on code in PR #10503:
URL: https://github.com/apache/paimon/pull/10503#discussion_r4261552271
##########
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:
[P2] Reject unknown row-filter predicate functions
A row filter with a valid FIELD_REF but an unknown predicate function is
accepted here because only its transforms are validated. I reproduced this
through the public REST management client on this head: creating a LEAF with
function="DOES_NOT_EXIST", literals=[1], and FIELD_REF(id BIGINT) returns
success, stores the policy, and returns it from query auth. Java
RESTCatalogServerUtils.parseRowFilter deserializes Predicate; the same JDK 8
probe fails with "Could not resolve leaf function", which validatePolicy maps
to HTTP 400. A typo in a policy can therefore pass these new integration
endpoints even though the Java server rejects it. Validate the leaf/compound
predicate functions before storing the policy and add an HTTP rejection
regression.
--
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]
