fhan688 commented on code in PR #20198: URL: https://github.com/apache/hudi/pull/20198#discussion_r4215673272
########## 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: Thanks for catching this. I’ve restricted `TIME`, `TIMESTAMP`, and `TIMESTAMP_LTZ` to precision 0–6 before SQL emission. I also added Python 6/7/9 boundary coverage and pinned Hudi 1.2.0/Flink 1.20.1 tests; the additional `TIME` case is covered directly against `HoodieSchemaConverter`. -- 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]
