danny0405 commented on code in PR #20198: URL: https://github.com/apache/hudi/pull/20198#discussion_r4213765712
########## hudi-agent-gateway/skills/hudi-architect/validate_flink_design.py: ########## @@ -0,0 +1,761 @@ +#!/usr/bin/env python3 +# +# 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. + +"""Validate and render the bounded Hudi Architect Flink PR2 design contract. + +The caller supplies every architecture decision. This program does not select +table type, identity, partitioning, replay behavior, or checkpoint cadence. It +only rejects incomplete or unsupported combinations and serializes a validated +contract into deterministic Flink SQL. +""" + +from __future__ import annotations + +import argparse +import json +import re +import sys +from pathlib import Path +from typing import Any, TypeGuard +from urllib.parse import parse_qsl, urlsplit + +from validate_flink_capabilities import ( + DEFAULT_MANIFEST, + load_manifest, + supported_option_keys, + validate_manifest, + validation_evidence, +) + +CONTRACT_SCHEMA_VERSION = 1 +MAX_TYPE_LENGTH = 2_147_483_647 +STATUS_PRIORITY = {"INCOMPLETE": 0, "REVIEW_REQUIRED": 1, "BLOCKED": 2} +SENSITIVE_QUERY_KEY = re.compile( + r"(?:access[_-]?key|api[_-]?key|credential|password|secret|signature|token)", + re.IGNORECASE, +) +PLACEHOLDER = re.compile(r"(?:<[^>]+>|\$\{[^}]+\}|\b(?:TBD|TODO)\b)", re.IGNORECASE) +PARAMETERIZED_TYPE = re.compile( + r"^(?P<name>DECIMAL|CHAR|VARCHAR|BINARY|VARBINARY|TIME|TIMESTAMP|TIMESTAMP_LTZ)" + r"\((?P<args>\d+(?:\s*,\s*\d+)?)\)$" +) + +FINDING_STATUS = { + "FLINK_APPEND_MODE_CLUSTERING_ENABLED": "BLOCKED", + "FLINK_AUTO_KEY_ACCEPTANCE_REQUIRED": "INCOMPLETE", + "FLINK_AUTO_KEY_DECLINED": "BLOCKED", + "FLINK_BASELINE_EVIDENCE_INVALID": "BLOCKED", + "FLINK_CATALOG_REQUIREMENT_UNRESOLVED": "REVIEW_REQUIRED", + "FLINK_CHECKPOINTING_REQUIRED": "BLOCKED", + "FLINK_CHECKPOINT_INTERVAL_REQUIRED": "INCOMPLETE", + "FLINK_DESIGN_CONTRACT_INVALID": "INCOMPLETE", + "FLINK_EXISTING_TABLE_DEFERRED": "BLOCKED", + "FLINK_EXTERNAL_CATALOG_REVIEW": "REVIEW_REQUIRED", + "FLINK_LOAD_BEARING_VALUE_REQUIRED": "INCOMPLETE", + "FLINK_MULTI_WRITER_REVIEW": "REVIEW_REQUIRED", + "FLINK_MUTABILITY_REQUIRED": "INCOMPLETE", + "FLINK_MUTABLE_COW_DEFERRED": "BLOCKED", + "FLINK_OPTION_NOT_VERIFIED": "REVIEW_REQUIRED", + "FLINK_PARTITION_FIELD_MISSING": "BLOCKED", + "FLINK_PHYSICAL_SCHEMA_REQUIRED": "INCOMPLETE", + "FLINK_PRIMARY_KEY_RECORD_KEY_CONFLICT": "BLOCKED", + "FLINK_RECORD_KEY_FIELD_MISSING": "BLOCKED", + "FLINK_RECORD_KEY_NULLABLE": "BLOCKED", + "FLINK_REPLAY_BEHAVIOR_UNRESOLVED": "REVIEW_REQUIRED", + "FLINK_REPLAY_IDEMPOTENCE_DEFERRED": "BLOCKED", + "FLINK_SOURCE_CHANGELOG_NOT_APPEND_ONLY": "BLOCKED", + "FLINK_SOURCE_CONTRACT_REQUIRED": "INCOMPLETE", + "FLINK_SOURCE_SCHEMA_MISMATCH": "BLOCKED", + "FLINK_TABLE_LIFECYCLE_REQUIRED": "INCOMPLETE", + "FLINK_VERSION_REQUIRED": "INCOMPLETE", + "FLINK_VERSION_UNVERIFIED": "REVIEW_REQUIRED", + "FLINK_WRITER_MODEL_UNRESOLVED": "REVIEW_REQUIRED", + "FLINK_PR2_WRITE_PATH_UNSUPPORTED": "BLOCKED", + "FLINK_SCHEMA_TYPE_UNVERIFIED": "REVIEW_REQUIRED", +} + + +class Assessment: + """Accumulate stable findings without discarding independent reasons.""" + + def __init__(self) -> None: + self.findings: list[dict[str, str]] = [] + self.advisories: list[str] = [] + + def add(self, code: str, message: str) -> None: + if code not in FINDING_STATUS: + raise ValueError(f"Unregistered finding code: {code}") + existing = next( + (finding for finding in self.findings if finding["code"] == code), None + ) + if existing is None: + self.findings.append( + {"code": code, "status": FINDING_STATUS[code], "message": message} + ) + elif message not in existing["message"]: + existing["message"] += f"; {message}" + + def advise(self, code: str) -> None: + if code not in self.advisories: + self.advisories.append(code) + + def final_status(self) -> str: + if not self.findings: + return "CONFIG_VALIDATED" + return max( + (finding["status"] for finding in self.findings), + key=STATUS_PRIORITY.__getitem__, + ) + + +def _dict(value: Any, field: str, assessment: Assessment) -> dict[str, Any]: + if isinstance(value, dict): + return value + assessment.add("FLINK_DESIGN_CONTRACT_INVALID", f"{field} must be an object") + return {} + + +def _check_keys( + value: dict[str, Any], allowed: set[str], field: str, assessment: Assessment +) -> None: + unknown = sorted(set(value) - allowed) + if unknown: + assessment.add( + "FLINK_DESIGN_CONTRACT_INVALID", + f"{field} contains unsupported fields: {unknown}", + ) + + +def _list(value: Any, field: str, assessment: Assessment) -> list[Any]: + if isinstance(value, list): + return value + assessment.add("FLINK_DESIGN_CONTRACT_INVALID", f"{field} must be an array") + return [] + + +def _non_empty_string(value: Any) -> TypeGuard[str]: + return isinstance(value, str) and bool(value.strip()) + + +def _normalize_type( + value: Any, manifest: dict[str, Any], assessment: Assessment +) -> str | None: + if not _non_empty_string(value): + assessment.add("FLINK_PHYSICAL_SCHEMA_REQUIRED", "Every field needs a Flink SQL type") + return None + normalized = " ".join(value.upper().split()) + if normalized == "INTEGER": + normalized = "INT" + if normalized in manifest["physical_types"]["simple"]: + return normalized + + matched = PARAMETERIZED_TYPE.fullmatch(normalized) + if matched is None or matched.group("name") not in manifest["physical_types"][ + "parameterized" + ]: + assessment.add( + "FLINK_SCHEMA_TYPE_UNVERIFIED", + f"Flink SQL type {value!r} is outside the bounded PR2 scalar surface", + ) + return None + + type_name = matched.group("name") + arguments = [int(part.strip()) for part in matched.group("args").split(",")] + valid = False + if type_name == "DECIMAL": + valid = ( + len(arguments) == 2 + and 1 <= arguments[0] <= 38 + and 0 <= arguments[1] <= arguments[0] + ) + elif type_name in {"CHAR", "VARCHAR", "BINARY", "VARBINARY"}: + valid = len(arguments) == 1 and 1 <= arguments[0] <= MAX_TYPE_LENGTH + elif type_name in {"TIME", "TIMESTAMP", "TIMESTAMP_LTZ"}: + valid = len(arguments) == 1 and 0 <= arguments[0] <= 9 + if not valid: + assessment.add( + "FLINK_SCHEMA_TYPE_UNVERIFIED", + f"Flink SQL type {value!r} has unsupported parameters", + ) + return None + return f"{type_name}({','.join(str(argument) for argument in arguments)})" + + +def _columns( + raw_columns: Any, + field: str, + manifest: dict[str, Any], + assessment: Assessment, +) -> list[dict[str, Any]]: + columns: list[dict[str, Any]] = [] + names: set[str] = set() + for index, raw_column in enumerate(_list(raw_columns, field, assessment)): + column = _dict(raw_column, f"{field}[{index}]", assessment) + _check_keys(column, {"name", "nullable", "type"}, f"{field}[{index}]", assessment) + name = column.get("name") + nullable = column.get("nullable") + if not _non_empty_string(name): + assessment.add("FLINK_PHYSICAL_SCHEMA_REQUIRED", f"{field}[{index}] needs a name") + continue Review Comment: [P2] Validate physical field names against the inferred Avro schema A nonempty name is not sufficient for this sink: renaming `payload` to `user-id` in both schemas of `auto_key.json` returns `CONFIG_VALIDATED`, and the renderer correctly backtick-quotes it, but planning the generated insert against the pinned artifacts fails with `SchemaParseException: Illegal character in: user-id`. The Hudi factory infers an Avro schema from the physical columns, whose field-name restrictions still apply to quoted SQL identifiers. Please reject unsupported target field names before emitting executable SQL and cover a quoted name such as this in the planner fixtures. ########## hudi-agent-gateway/skills/hudi-architect/validate_flink_design.py: ########## @@ -0,0 +1,761 @@ +#!/usr/bin/env python3 +# +# 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. + +"""Validate and render the bounded Hudi Architect Flink PR2 design contract. + +The caller supplies every architecture decision. This program does not select +table type, identity, partitioning, replay behavior, or checkpoint cadence. It +only rejects incomplete or unsupported combinations and serializes a validated +contract into deterministic Flink SQL. +""" + +from __future__ import annotations + +import argparse +import json +import re +import sys +from pathlib import Path +from typing import Any, TypeGuard +from urllib.parse import parse_qsl, urlsplit + +from validate_flink_capabilities import ( + DEFAULT_MANIFEST, + load_manifest, + supported_option_keys, + validate_manifest, + validation_evidence, +) + +CONTRACT_SCHEMA_VERSION = 1 +MAX_TYPE_LENGTH = 2_147_483_647 +STATUS_PRIORITY = {"INCOMPLETE": 0, "REVIEW_REQUIRED": 1, "BLOCKED": 2} +SENSITIVE_QUERY_KEY = re.compile( + r"(?:access[_-]?key|api[_-]?key|credential|password|secret|signature|token)", + re.IGNORECASE, +) +PLACEHOLDER = re.compile(r"(?:<[^>]+>|\$\{[^}]+\}|\b(?:TBD|TODO)\b)", re.IGNORECASE) +PARAMETERIZED_TYPE = re.compile( + r"^(?P<name>DECIMAL|CHAR|VARCHAR|BINARY|VARBINARY|TIME|TIMESTAMP|TIMESTAMP_LTZ)" + r"\((?P<args>\d+(?:\s*,\s*\d+)?)\)$" +) + +FINDING_STATUS = { + "FLINK_APPEND_MODE_CLUSTERING_ENABLED": "BLOCKED", + "FLINK_AUTO_KEY_ACCEPTANCE_REQUIRED": "INCOMPLETE", + "FLINK_AUTO_KEY_DECLINED": "BLOCKED", + "FLINK_BASELINE_EVIDENCE_INVALID": "BLOCKED", + "FLINK_CATALOG_REQUIREMENT_UNRESOLVED": "REVIEW_REQUIRED", + "FLINK_CHECKPOINTING_REQUIRED": "BLOCKED", + "FLINK_CHECKPOINT_INTERVAL_REQUIRED": "INCOMPLETE", + "FLINK_DESIGN_CONTRACT_INVALID": "INCOMPLETE", + "FLINK_EXISTING_TABLE_DEFERRED": "BLOCKED", + "FLINK_EXTERNAL_CATALOG_REVIEW": "REVIEW_REQUIRED", + "FLINK_LOAD_BEARING_VALUE_REQUIRED": "INCOMPLETE", + "FLINK_MULTI_WRITER_REVIEW": "REVIEW_REQUIRED", + "FLINK_MUTABILITY_REQUIRED": "INCOMPLETE", + "FLINK_MUTABLE_COW_DEFERRED": "BLOCKED", + "FLINK_OPTION_NOT_VERIFIED": "REVIEW_REQUIRED", + "FLINK_PARTITION_FIELD_MISSING": "BLOCKED", + "FLINK_PHYSICAL_SCHEMA_REQUIRED": "INCOMPLETE", + "FLINK_PRIMARY_KEY_RECORD_KEY_CONFLICT": "BLOCKED", + "FLINK_RECORD_KEY_FIELD_MISSING": "BLOCKED", + "FLINK_RECORD_KEY_NULLABLE": "BLOCKED", + "FLINK_REPLAY_BEHAVIOR_UNRESOLVED": "REVIEW_REQUIRED", + "FLINK_REPLAY_IDEMPOTENCE_DEFERRED": "BLOCKED", + "FLINK_SOURCE_CHANGELOG_NOT_APPEND_ONLY": "BLOCKED", + "FLINK_SOURCE_CONTRACT_REQUIRED": "INCOMPLETE", + "FLINK_SOURCE_SCHEMA_MISMATCH": "BLOCKED", + "FLINK_TABLE_LIFECYCLE_REQUIRED": "INCOMPLETE", + "FLINK_VERSION_REQUIRED": "INCOMPLETE", + "FLINK_VERSION_UNVERIFIED": "REVIEW_REQUIRED", + "FLINK_WRITER_MODEL_UNRESOLVED": "REVIEW_REQUIRED", + "FLINK_PR2_WRITE_PATH_UNSUPPORTED": "BLOCKED", + "FLINK_SCHEMA_TYPE_UNVERIFIED": "REVIEW_REQUIRED", +} + + +class Assessment: + """Accumulate stable findings without discarding independent reasons.""" + + def __init__(self) -> None: + self.findings: list[dict[str, str]] = [] + self.advisories: list[str] = [] + + def add(self, code: str, message: str) -> None: + if code not in FINDING_STATUS: + raise ValueError(f"Unregistered finding code: {code}") + existing = next( + (finding for finding in self.findings if finding["code"] == code), None + ) + if existing is None: + self.findings.append( + {"code": code, "status": FINDING_STATUS[code], "message": message} + ) + elif message not in existing["message"]: + existing["message"] += f"; {message}" + + def advise(self, code: str) -> None: + if code not in self.advisories: + self.advisories.append(code) + + def final_status(self) -> str: + if not self.findings: + return "CONFIG_VALIDATED" + return max( + (finding["status"] for finding in self.findings), + key=STATUS_PRIORITY.__getitem__, + ) + + +def _dict(value: Any, field: str, assessment: Assessment) -> dict[str, Any]: + if isinstance(value, dict): + return value + assessment.add("FLINK_DESIGN_CONTRACT_INVALID", f"{field} must be an object") + return {} + + +def _check_keys( + value: dict[str, Any], allowed: set[str], field: str, assessment: Assessment +) -> None: + unknown = sorted(set(value) - allowed) + if unknown: + assessment.add( + "FLINK_DESIGN_CONTRACT_INVALID", + f"{field} contains unsupported fields: {unknown}", + ) + + +def _list(value: Any, field: str, assessment: Assessment) -> list[Any]: + if isinstance(value, list): + return value + assessment.add("FLINK_DESIGN_CONTRACT_INVALID", f"{field} must be an array") + return [] + + +def _non_empty_string(value: Any) -> TypeGuard[str]: + return isinstance(value, str) and bool(value.strip()) + + +def _normalize_type( + value: Any, manifest: dict[str, Any], assessment: Assessment +) -> str | None: + if not _non_empty_string(value): + assessment.add("FLINK_PHYSICAL_SCHEMA_REQUIRED", "Every field needs a Flink SQL type") + return None + normalized = " ".join(value.upper().split()) + if normalized == "INTEGER": + normalized = "INT" + if normalized in manifest["physical_types"]["simple"]: + return normalized + + matched = PARAMETERIZED_TYPE.fullmatch(normalized) + if matched is None or matched.group("name") not in manifest["physical_types"][ + "parameterized" + ]: + assessment.add( + "FLINK_SCHEMA_TYPE_UNVERIFIED", + f"Flink SQL type {value!r} is outside the bounded PR2 scalar surface", + ) + return None + + type_name = matched.group("name") + arguments = [int(part.strip()) for part in matched.group("args").split(",")] + valid = False + if type_name == "DECIMAL": + valid = ( + len(arguments) == 2 + and 1 <= arguments[0] <= 38 + and 0 <= arguments[1] <= arguments[0] + ) + elif type_name in {"CHAR", "VARCHAR", "BINARY", "VARBINARY"}: + valid = len(arguments) == 1 and 1 <= arguments[0] <= MAX_TYPE_LENGTH + elif type_name in {"TIME", "TIMESTAMP", "TIMESTAMP_LTZ"}: + valid = len(arguments) == 1 and 0 <= arguments[0] <= 9 Review Comment: [P2] Restrict timestamp precision to the pinned connector limit Changing `payload` to `TIMESTAMP(9)` in both the source and target schemas of `auto_key.json` still returns `CONFIG_VALIDATED` and executable SQL. However, planning that SQL with the pinned Hudi 1.2.0 / Flink 1.20.1 artifacts fails in `HoodieSchemaConverter` with "it only supports precisions <= 6." `TIMESTAMP_LTZ(9)` fails the same way. Thus ordinary nanosecond schemas pass the executable gate even though the sink cannot be planned. Please validate the connector-supported precision limits, rather than only the Flink SQL range, and add boundary cases to the Python and pinned Java tests. -- 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]
