danny0405 commented on code in PR #20198: URL: https://github.com/apache/hudi/pull/20198#discussion_r4226334114
########## hudi-agent-gateway/skills/hudi-architect/validate_flink_design.py: ########## @@ -0,0 +1,810 @@ +#!/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_CHECKPOINT_INTERVAL_UNSUPPORTED": "BLOCKED", + "FLINK_DESIGN_CONTRACT_INVALID": "INCOMPLETE", + "FLINK_EXISTING_TABLE_DEFERRED": "BLOCKED", + "FLINK_EXTERNAL_CATALOG_REVIEW": "REVIEW_REQUIRED", + "FLINK_HUDI_METADATA_FIELD_CONFLICT": "BLOCKED", + "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_SCHEMA_FIELD_NAME_UNSUPPORTED": "BLOCKED", + "FLINK_SOURCE_CHANGELOG_NOT_APPEND_ONLY": "BLOCKED", + "FLINK_SOURCE_CONTRACT_REQUIRED": "INCOMPLETE", + "FLINK_SOURCE_SCHEMA_MISMATCH": "BLOCKED", + "FLINK_TABLE_LIFECYCLE_REQUIRED": "INCOMPLETE", + "FLINK_TEMPORAL_PRECISION_UNSUPPORTED": "BLOCKED", + "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 manifest["physical_schema_constraints"]["temporal_types"]: + constraints = manifest["physical_schema_constraints"] + if len(arguments) == 1 and not ( + constraints["temporal_precision_min"] + <= arguments[0] + <= constraints["temporal_precision_max"] + ): + assessment.add( + "FLINK_TEMPORAL_PRECISION_UNSUPPORTED", + f"Hudi 1.2.0 supports {type_name} precision only between " + f"{constraints['temporal_precision_min']} and " + f"{constraints['temporal_precision_max']}", + ) + return None + valid = len(arguments) == 1 + 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, + *, + reject_hudi_metadata_fields: bool = False, +) -> 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 + if re.fullmatch( + manifest["physical_schema_constraints"]["field_name_pattern"], name + ) is None: + assessment.add( + "FLINK_SCHEMA_FIELD_NAME_UNSUPPORTED", + f"{field}[{index}].name {name!r} cannot be represented by the " + "Avro-backed Hudi sink schema", + ) + continue + reserved_names = manifest["physical_schema_constraints"][ + "reserved_target_field_names" + ] + if reject_hudi_metadata_fields and name.lower() in reserved_names: + assessment.add( + "FLINK_HUDI_METADATA_FIELD_CONFLICT", + f"{field}[{index}].name {name!r} is reserved for Hudi metadata", + ) + if name in names: + assessment.add( + "FLINK_DESIGN_CONTRACT_INVALID", f"Duplicate field {name!r} in {field}" + ) + continue + names.add(name) + if not isinstance(nullable, bool): + assessment.add( + "FLINK_PHYSICAL_SCHEMA_REQUIRED", + f"{field}[{index}].nullable must be explicitly true or false", + ) + continue + normalized_type = _normalize_type(column.get("type"), manifest, assessment) + if normalized_type is None: + continue + columns.append({"name": name, "type": normalized_type, "nullable": nullable}) + if not columns: + assessment.add("FLINK_PHYSICAL_SCHEMA_REQUIRED", f"{field} must contain physical fields") + return columns + + +def _string_list(value: Any, field: str, assessment: Assessment) -> list[str]: + values = _list(value, field, assessment) + if not all(_non_empty_string(item) for item in values): + assessment.add("FLINK_DESIGN_CONTRACT_INVALID", f"{field} must contain field names") + return [] + if len(values) != len(set(values)): + assessment.add("FLINK_DESIGN_CONTRACT_INVALID", f"{field} must not contain duplicates") + return values + + +def _check_concrete_value(value: Any) -> TypeGuard[str]: + return _non_empty_string(value) and PLACEHOLDER.search(value) is None + + +def _path_validation_error(path: str) -> str | None: + if any(ord(character) < 32 for character in path): + return "The target path must not contain control characters" + try: + parsed = urlsplit(path) + except ValueError: + return "The target path must be a valid URI or filesystem path" + if parsed.username is not None or parsed.password is not None: Review Comment: [P2] Distinguish Azure filesystem authorities from URI credentials An otherwise valid design with `table.path = "abfss://[email protected]/orders"` returns `INCOMPLETE` because `urlsplit()` exposes `container` as `username`. In the standard ABFS/ABFSS URI format this component names the filesystem, not a credential, so this condition blocks ordinary Azure targets. Please make the authority check scheme-aware while retaining rejection of actual credentials. Azure documents the format here: https://learn.microsoft.com/en-us/previous-versions/azure/storage/blobs/data-lake-storage-introduction-abfs-uri ########## hudi-agent-gateway/skills/hudi-architect/validate_flink_design.py: ########## @@ -0,0 +1,810 @@ +#!/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_CHECKPOINT_INTERVAL_UNSUPPORTED": "BLOCKED", + "FLINK_DESIGN_CONTRACT_INVALID": "INCOMPLETE", + "FLINK_EXISTING_TABLE_DEFERRED": "BLOCKED", + "FLINK_EXTERNAL_CATALOG_REVIEW": "REVIEW_REQUIRED", + "FLINK_HUDI_METADATA_FIELD_CONFLICT": "BLOCKED", + "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_SCHEMA_FIELD_NAME_UNSUPPORTED": "BLOCKED", + "FLINK_SOURCE_CHANGELOG_NOT_APPEND_ONLY": "BLOCKED", + "FLINK_SOURCE_CONTRACT_REQUIRED": "INCOMPLETE", + "FLINK_SOURCE_SCHEMA_MISMATCH": "BLOCKED", + "FLINK_TABLE_LIFECYCLE_REQUIRED": "INCOMPLETE", + "FLINK_TEMPORAL_PRECISION_UNSUPPORTED": "BLOCKED", + "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 manifest["physical_schema_constraints"]["temporal_types"]: + constraints = manifest["physical_schema_constraints"] + if len(arguments) == 1 and not ( + constraints["temporal_precision_min"] + <= arguments[0] + <= constraints["temporal_precision_max"] + ): + assessment.add( + "FLINK_TEMPORAL_PRECISION_UNSUPPORTED", + f"Hudi 1.2.0 supports {type_name} precision only between " + f"{constraints['temporal_precision_min']} and " + f"{constraints['temporal_precision_max']}", + ) + return None + valid = len(arguments) == 1 + 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, + *, + reject_hudi_metadata_fields: bool = False, +) -> 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 + if re.fullmatch( + manifest["physical_schema_constraints"]["field_name_pattern"], name + ) is None: + assessment.add( + "FLINK_SCHEMA_FIELD_NAME_UNSUPPORTED", + f"{field}[{index}].name {name!r} cannot be represented by the " + "Avro-backed Hudi sink schema", + ) + continue + reserved_names = manifest["physical_schema_constraints"][ + "reserved_target_field_names" + ] + if reject_hudi_metadata_fields and name.lower() in reserved_names: + assessment.add( + "FLINK_HUDI_METADATA_FIELD_CONFLICT", + f"{field}[{index}].name {name!r} is reserved for Hudi metadata", + ) + if name in names: + assessment.add( + "FLINK_DESIGN_CONTRACT_INVALID", f"Duplicate field {name!r} in {field}" + ) + continue + names.add(name) + if not isinstance(nullable, bool): + assessment.add( + "FLINK_PHYSICAL_SCHEMA_REQUIRED", + f"{field}[{index}].nullable must be explicitly true or false", + ) + continue + normalized_type = _normalize_type(column.get("type"), manifest, assessment) + if normalized_type is None: + continue + columns.append({"name": name, "type": normalized_type, "nullable": nullable}) + if not columns: + assessment.add("FLINK_PHYSICAL_SCHEMA_REQUIRED", f"{field} must contain physical fields") + return columns + + +def _string_list(value: Any, field: str, assessment: Assessment) -> list[str]: + values = _list(value, field, assessment) + if not all(_non_empty_string(item) for item in values): + assessment.add("FLINK_DESIGN_CONTRACT_INVALID", f"{field} must contain field names") + return [] + if len(values) != len(set(values)): + assessment.add("FLINK_DESIGN_CONTRACT_INVALID", f"{field} must not contain duplicates") + return values + + +def _check_concrete_value(value: Any) -> TypeGuard[str]: + return _non_empty_string(value) and PLACEHOLDER.search(value) is None + + +def _path_validation_error(path: str) -> str | None: + if any(ord(character) < 32 for character in path): + return "The target path must not contain control characters" + try: + parsed = urlsplit(path) + except ValueError: + return "The target path must be a valid URI or filesystem path" + if parsed.username is not None or parsed.password is not None: + return "The target path must not contain credentials or secret query parameters" + parameters = [*parse_qsl(parsed.query), *parse_qsl(parsed.fragment)] + if any(SENSITIVE_QUERY_KEY.search(key) for key, _ in parameters): + return "The target path must not contain credentials or secret query parameters" + return None + + +def _quote_identifier(identifier: str) -> str: + return "`" + identifier.replace("`", "``") + "`" + + +def _quote_qualified_identifier(identifier: str) -> str: + return ".".join(_quote_identifier(part) for part in identifier.split(".")) + + +def _sql_string(value: str) -> str: + return "'" + value.replace("'", "''") + "'" + + +def _render_sql( + contract: dict[str, Any], + table_columns: list[dict[str, Any]], + partition_fields: list[str], + record_key_fields: list[str], +) -> dict[str, str]: + table = contract["table"] + source = contract["source"] + runtime = contract["runtime"] + identity = contract["identity"] + + definitions = [ + f" {_quote_identifier(column['name'])} {column['type']}" + + ("" if column["nullable"] else " NOT NULL") + for column in table_columns + ] + if identity["mode"] == "stable_key": + key_sql = ", ".join(_quote_identifier(field) for field in record_key_fields) + definitions.append(f" PRIMARY KEY ({key_sql}) NOT ENFORCED") + + table_ddl = ( + f"CREATE TABLE {_quote_qualified_identifier(table['name'])} (\n" + + ",\n".join(definitions) + + "\n)" + ) + if partition_fields: + partition_sql = ", ".join(_quote_identifier(field) for field in partition_fields) + table_ddl += f"\nPARTITIONED BY ({partition_sql})" + table_ddl += ( + "\nWITH (\n" + " 'connector' = 'hudi',\n" + f" 'path' = {_sql_string(table['path'])},\n" + " 'table.type' = 'COPY_ON_WRITE',\n" + " 'write.operation' = 'insert',\n" + " 'write.insert.cluster' = 'false'\n" + ");" + ) + + field_sql = ",\n ".join(_quote_identifier(column["name"]) for column in table_columns) + insert_sql = ( + f"INSERT INTO {_quote_qualified_identifier(table['name'])} (\n {field_sql}\n)\n" + f"SELECT\n {field_sql}\n" + f"FROM {_quote_qualified_identifier(source['table'])};" + ) + runtime_sql = ( + "SET 'execution.runtime-mode' = 'streaming';\n" + "SET 'execution.checkpointing.interval' = " + f"'{runtime['checkpoint_interval_ms']} ms';" + ) + return { + "runtime_sql": runtime_sql, + "table_ddl": table_ddl, + "insert_sql": insert_sql, + "combined_sql": f"{runtime_sql}\n\n{table_ddl}\n\n{insert_sql}", + } + + +def assess_design( + contract: dict[str, Any], manifest: dict[str, Any] +) -> dict[str, Any]: + """Validate one explicit PR2 contract and return deterministic evidence.""" + + assessment = Assessment() + manifest_errors = validate_manifest(manifest) + if manifest_errors: + assessment.add( + "FLINK_BASELINE_EVIDENCE_INVALID", "; ".join(manifest_errors) + ) + return { + "status": assessment.final_status(), + "executable_eligible": False, + "finding_codes": [finding["code"] for finding in assessment.findings], + "findings": assessment.findings, + "advisory_codes": assessment.advisories, + "validation_evidence": { + "design_contract_schema": CONTRACT_SCHEMA_VERSION, + }, + } + + _check_keys( + contract, + { + "baseline_id", + "contract_schema", + "flink_version", + "hudi_version", + "identity", + "runtime", + "safety", + "source", + "table", + "write", + }, + "contract", + assessment, + ) + if contract.get("contract_schema") != CONTRACT_SCHEMA_VERSION: + assessment.add( + "FLINK_DESIGN_CONTRACT_INVALID", + f"contract_schema must be {CONTRACT_SCHEMA_VERSION}", + ) + if contract.get("baseline_id") != manifest.get("baseline_id"): + assessment.add("FLINK_VERSION_UNVERIFIED", "The capability baseline does not match") + if not _non_empty_string(contract.get("hudi_version")) or not _non_empty_string( + contract.get("flink_version") + ): + assessment.add("FLINK_VERSION_REQUIRED", "Concrete Hudi and Flink versions are required") + else: + if contract["hudi_version"] != manifest["hudi"]["version"]: + assessment.add("FLINK_VERSION_UNVERIFIED", "The Hudi version is outside the baseline") + if not re.fullmatch(r"1\.20(?:\.\d+|\.x)?", contract["flink_version"]): + assessment.add("FLINK_VERSION_UNVERIFIED", "The Flink version is outside the baseline") + + safety = _dict(contract.get("safety"), "safety", assessment) + _check_keys( + safety, + { + "external_catalog", + "mutability", + "replay_behavior", + "table_lifecycle", + "writer_model", + }, + "safety", + assessment, + ) + lifecycle = safety.get("table_lifecycle") + if lifecycle is None: + assessment.add("FLINK_TABLE_LIFECYCLE_REQUIRED", "Table lifecycle is required") + elif lifecycle == "existing": + assessment.add("FLINK_EXISTING_TABLE_DEFERRED", "PR2 supports only a new table") + elif lifecycle != "new": + assessment.add("FLINK_TABLE_LIFECYCLE_REQUIRED", "Table lifecycle is unresolved") + writer_model = safety.get("writer_model") + if writer_model is None: + assessment.add("FLINK_WRITER_MODEL_UNRESOLVED", "Writer topology is required") + elif writer_model == "multi_writer": + assessment.add("FLINK_MULTI_WRITER_REVIEW", "PR2 supports only a confirmed single writer") + elif writer_model != "single_writer": + assessment.add("FLINK_WRITER_MODEL_UNRESOLVED", "Writer topology is unresolved") + external_catalog = safety.get("external_catalog") + if external_catalog is True: + assessment.add("FLINK_EXTERNAL_CATALOG_REVIEW", "Catalog composition is deferred") + elif external_catalog is not False: + assessment.add("FLINK_CATALOG_REQUIREMENT_UNRESOLVED", "Catalog requirement is unresolved") + mutability = safety.get("mutability") + if mutability == "mutable": + assessment.add("FLINK_MUTABLE_COW_DEFERRED", "Mutable COW is deferred") + elif mutability != "append_only": + assessment.add("FLINK_MUTABILITY_REQUIRED", "Append-only input must be confirmed") + replay_behavior = safety.get("replay_behavior") + if replay_behavior == "must_collapse": + assessment.add("FLINK_REPLAY_IDEMPOTENCE_DEFERRED", "Replay deduplication needs upsert") + elif replay_behavior not in {"cannot_occur", "duplicates_acceptable"}: + assessment.add("FLINK_REPLAY_BEHAVIOR_UNRESOLVED", "Replay behavior is unresolved") + + table = _dict(contract.get("table"), "table", assessment) + _check_keys(table, {"columns", "name", "partition_fields", "path"}, "table", assessment) + table_name = table.get("name") + table_path = table.get("path") + if not _check_concrete_value(table_name) or any( + not part for part in str(table_name).split(".") + ): + assessment.add("FLINK_LOAD_BEARING_VALUE_REQUIRED", "A concrete target table is required") + if not _check_concrete_value(table_path): + assessment.add("FLINK_LOAD_BEARING_VALUE_REQUIRED", "A concrete target path is required") + elif path_error := _path_validation_error(table_path): + assessment.add( + "FLINK_LOAD_BEARING_VALUE_REQUIRED", + path_error, + ) + table_columns = _columns( + table.get("columns"), + "table.columns", + manifest, + assessment, + reject_hudi_metadata_fields=True, + ) + table_by_name = {column["name"]: column for column in table_columns} + partition_fields = _string_list( + table.get("partition_fields"), "table.partition_fields", assessment + ) + for field in partition_fields: + if field not in table_by_name: + assessment.add( + "FLINK_PARTITION_FIELD_MISSING", + f"Partition field {field!r} is absent from the physical schema", + ) + + identity = _dict(contract.get("identity"), "identity", assessment) + _check_keys( + identity, + { + "auto_key_accepted", + "mode", + "record_key_fields", + "record_key_option_fields", + }, + "identity", + assessment, + ) + identity_mode = identity.get("mode") + record_key_fields = _string_list( + identity.get("record_key_fields"), "identity.record_key_fields", assessment + ) + option_fields = _string_list( + identity.get("record_key_option_fields"), + "identity.record_key_option_fields", + assessment, + ) + raw_write = contract.get("write") + raw_connector_options = ( + raw_write.get("connector_options") if isinstance(raw_write, dict) else None + ) + if isinstance(raw_connector_options, dict): + raw_record_key = raw_connector_options.get( + "hoodie.datasource.write.recordkey.field" + ) + if raw_record_key is not None: + if not _non_empty_string(raw_record_key): + assessment.add( + "FLINK_DESIGN_CONTRACT_INVALID", + "The record-key connector option must contain field names", + ) + else: + raw_option_fields = [field.strip() for field in raw_record_key.split(",")] + if not all(raw_option_fields) or len(raw_option_fields) != len( + set(raw_option_fields) + ): + assessment.add( + "FLINK_DESIGN_CONTRACT_INVALID", + "The record-key connector option must contain unique field names", + ) + else: + if option_fields and option_fields != raw_option_fields: + assessment.add( + "FLINK_DESIGN_CONTRACT_INVALID", + "Normalized and raw record-key option fields do not match", + ) + option_fields = list(dict.fromkeys([*option_fields, *raw_option_fields])) + for field in option_fields: + if field not in table_by_name: + assessment.add( + "FLINK_RECORD_KEY_FIELD_MISSING", + f"Record-key option field {field!r} is absent from the physical schema", + ) + if identity_mode == "stable_key": + assessment.advise("FLINK_STABLE_KEY_NOT_IDEMPOTENT") + if "auto_key_accepted" in identity: + assessment.add( + "FLINK_DESIGN_CONTRACT_INVALID", + "identity.auto_key_accepted applies only to auto_key mode", + ) + if not record_key_fields: + assessment.add("FLINK_PHYSICAL_SCHEMA_REQUIRED", "Stable-key fields are required") + for field in record_key_fields: + if field not in table_by_name: + assessment.add( + "FLINK_RECORD_KEY_FIELD_MISSING", + f"Record-key field {field!r} is absent from the physical schema", + ) + elif table_by_name[field]["nullable"]: Review Comment: [P2] Reject binary columns in record-key and partition positions Changing the stable-key fixture’s `id` type to `BYTES`, `BINARY(16)`, or `VARBINARY(16)` in both schemas still returns `CONFIG_VALIDATED`; the same is true for the partition field. In the pinned Hudi 1.2.0 append path, `BulkInsertWriterHelper` uses `RowDataKeyGen`, which converts these byte arrays to Java object identity strings. Two additional tests against the released bundle reproduced equal binary values producing different record keys and different partition directories (`[B@...`). This breaks the accepted stable-key contract and can scatter equal partition values across directories. Please reject binary types in these positions until a deterministic encoding is supported and verified; they can remain valid payload columns. ########## hudi-agent-gateway/skills/hudi-architect/validate_flink_design.py: ########## @@ -0,0 +1,810 @@ +#!/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_CHECKPOINT_INTERVAL_UNSUPPORTED": "BLOCKED", + "FLINK_DESIGN_CONTRACT_INVALID": "INCOMPLETE", + "FLINK_EXISTING_TABLE_DEFERRED": "BLOCKED", + "FLINK_EXTERNAL_CATALOG_REVIEW": "REVIEW_REQUIRED", + "FLINK_HUDI_METADATA_FIELD_CONFLICT": "BLOCKED", + "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_SCHEMA_FIELD_NAME_UNSUPPORTED": "BLOCKED", + "FLINK_SOURCE_CHANGELOG_NOT_APPEND_ONLY": "BLOCKED", + "FLINK_SOURCE_CONTRACT_REQUIRED": "INCOMPLETE", + "FLINK_SOURCE_SCHEMA_MISMATCH": "BLOCKED", + "FLINK_TABLE_LIFECYCLE_REQUIRED": "INCOMPLETE", + "FLINK_TEMPORAL_PRECISION_UNSUPPORTED": "BLOCKED", + "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 manifest["physical_schema_constraints"]["temporal_types"]: + constraints = manifest["physical_schema_constraints"] + if len(arguments) == 1 and not ( + constraints["temporal_precision_min"] + <= arguments[0] + <= constraints["temporal_precision_max"] + ): + assessment.add( + "FLINK_TEMPORAL_PRECISION_UNSUPPORTED", + f"Hudi 1.2.0 supports {type_name} precision only between " + f"{constraints['temporal_precision_min']} and " + f"{constraints['temporal_precision_max']}", + ) + return None + valid = len(arguments) == 1 + 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, + *, + reject_hudi_metadata_fields: bool = False, +) -> 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 + if re.fullmatch( + manifest["physical_schema_constraints"]["field_name_pattern"], name + ) is None: + assessment.add( + "FLINK_SCHEMA_FIELD_NAME_UNSUPPORTED", + f"{field}[{index}].name {name!r} cannot be represented by the " + "Avro-backed Hudi sink schema", + ) + continue + reserved_names = manifest["physical_schema_constraints"][ + "reserved_target_field_names" + ] + if reject_hudi_metadata_fields and name.lower() in reserved_names: + assessment.add( + "FLINK_HUDI_METADATA_FIELD_CONFLICT", + f"{field}[{index}].name {name!r} is reserved for Hudi metadata", + ) + if name in names: + assessment.add( + "FLINK_DESIGN_CONTRACT_INVALID", f"Duplicate field {name!r} in {field}" + ) + continue + names.add(name) + if not isinstance(nullable, bool): + assessment.add( + "FLINK_PHYSICAL_SCHEMA_REQUIRED", + f"{field}[{index}].nullable must be explicitly true or false", + ) + continue + normalized_type = _normalize_type(column.get("type"), manifest, assessment) + if normalized_type is None: + continue + columns.append({"name": name, "type": normalized_type, "nullable": nullable}) + if not columns: + assessment.add("FLINK_PHYSICAL_SCHEMA_REQUIRED", f"{field} must contain physical fields") + return columns + + +def _string_list(value: Any, field: str, assessment: Assessment) -> list[str]: + values = _list(value, field, assessment) + if not all(_non_empty_string(item) for item in values): + assessment.add("FLINK_DESIGN_CONTRACT_INVALID", f"{field} must contain field names") + return [] + if len(values) != len(set(values)): + assessment.add("FLINK_DESIGN_CONTRACT_INVALID", f"{field} must not contain duplicates") + return values + + +def _check_concrete_value(value: Any) -> TypeGuard[str]: + return _non_empty_string(value) and PLACEHOLDER.search(value) is None + + +def _path_validation_error(path: str) -> str | None: + if any(ord(character) < 32 for character in path): + return "The target path must not contain control characters" + try: + parsed = urlsplit(path) + except ValueError: + return "The target path must be a valid URI or filesystem path" + if parsed.username is not None or parsed.password is not None: + return "The target path must not contain credentials or secret query parameters" + parameters = [*parse_qsl(parsed.query), *parse_qsl(parsed.fragment)] + if any(SENSITIVE_QUERY_KEY.search(key) for key, _ in parameters): Review Comment: [P2] Reject SAS signatures before emitting target paths The sensitive-query regex recognizes `signature` but not Azure SAS’s `sig` parameter. Replacing the fixture path with `https://account.blob.core.windows.net/container/orders?sv=2025-01-05&sp=rw&sig=EXAMPLE_SIGNATURE` still returns `CONFIG_VALIDATED` and copies the signature verbatim into the generated SQL. A real SAS URL would therefore expose its authorization token in artifacts advertised as credential-free. Please recognize SAS signatures (or reject unsupported query-bearing target paths) and add a regression covering `sig=`. The authorization parameter is documented here: https://learn.microsoft.com/en-us/rest/api/storageservices/create-service-sas ########## hudi-agent-gateway/skills/hudi-architect/validate_flink_design.py: ########## @@ -0,0 +1,810 @@ +#!/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_CHECKPOINT_INTERVAL_UNSUPPORTED": "BLOCKED", + "FLINK_DESIGN_CONTRACT_INVALID": "INCOMPLETE", + "FLINK_EXISTING_TABLE_DEFERRED": "BLOCKED", + "FLINK_EXTERNAL_CATALOG_REVIEW": "REVIEW_REQUIRED", + "FLINK_HUDI_METADATA_FIELD_CONFLICT": "BLOCKED", + "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_SCHEMA_FIELD_NAME_UNSUPPORTED": "BLOCKED", + "FLINK_SOURCE_CHANGELOG_NOT_APPEND_ONLY": "BLOCKED", + "FLINK_SOURCE_CONTRACT_REQUIRED": "INCOMPLETE", + "FLINK_SOURCE_SCHEMA_MISMATCH": "BLOCKED", + "FLINK_TABLE_LIFECYCLE_REQUIRED": "INCOMPLETE", + "FLINK_TEMPORAL_PRECISION_UNSUPPORTED": "BLOCKED", + "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 manifest["physical_schema_constraints"]["temporal_types"]: + constraints = manifest["physical_schema_constraints"] + if len(arguments) == 1 and not ( + constraints["temporal_precision_min"] + <= arguments[0] + <= constraints["temporal_precision_max"] + ): + assessment.add( + "FLINK_TEMPORAL_PRECISION_UNSUPPORTED", + f"Hudi 1.2.0 supports {type_name} precision only between " + f"{constraints['temporal_precision_min']} and " + f"{constraints['temporal_precision_max']}", + ) + return None + valid = len(arguments) == 1 + 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, + *, + reject_hudi_metadata_fields: bool = False, +) -> 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 + if re.fullmatch( + manifest["physical_schema_constraints"]["field_name_pattern"], name + ) is None: + assessment.add( + "FLINK_SCHEMA_FIELD_NAME_UNSUPPORTED", + f"{field}[{index}].name {name!r} cannot be represented by the " + "Avro-backed Hudi sink schema", + ) + continue + reserved_names = manifest["physical_schema_constraints"][ + "reserved_target_field_names" + ] + if reject_hudi_metadata_fields and name.lower() in reserved_names: + assessment.add( + "FLINK_HUDI_METADATA_FIELD_CONFLICT", + f"{field}[{index}].name {name!r} is reserved for Hudi metadata", + ) + if name in names: + assessment.add( + "FLINK_DESIGN_CONTRACT_INVALID", f"Duplicate field {name!r} in {field}" + ) + continue + names.add(name) + if not isinstance(nullable, bool): + assessment.add( + "FLINK_PHYSICAL_SCHEMA_REQUIRED", + f"{field}[{index}].nullable must be explicitly true or false", + ) + continue + normalized_type = _normalize_type(column.get("type"), manifest, assessment) + if normalized_type is None: + continue + columns.append({"name": name, "type": normalized_type, "nullable": nullable}) + if not columns: + assessment.add("FLINK_PHYSICAL_SCHEMA_REQUIRED", f"{field} must contain physical fields") + return columns + + +def _string_list(value: Any, field: str, assessment: Assessment) -> list[str]: + values = _list(value, field, assessment) + if not all(_non_empty_string(item) for item in values): + assessment.add("FLINK_DESIGN_CONTRACT_INVALID", f"{field} must contain field names") + return [] + if len(values) != len(set(values)): + assessment.add("FLINK_DESIGN_CONTRACT_INVALID", f"{field} must not contain duplicates") + return values + + +def _check_concrete_value(value: Any) -> TypeGuard[str]: + return _non_empty_string(value) and PLACEHOLDER.search(value) is None + + +def _path_validation_error(path: str) -> str | None: + if any(ord(character) < 32 for character in path): + return "The target path must not contain control characters" + try: + parsed = urlsplit(path) + except ValueError: + return "The target path must be a valid URI or filesystem path" + if parsed.username is not None or parsed.password is not None: + return "The target path must not contain credentials or secret query parameters" + parameters = [*parse_qsl(parsed.query), *parse_qsl(parsed.fragment)] + if any(SENSITIVE_QUERY_KEY.search(key) for key, _ in parameters): + return "The target path must not contain credentials or secret query parameters" + return None + + +def _quote_identifier(identifier: str) -> str: + return "`" + identifier.replace("`", "``") + "`" + + +def _quote_qualified_identifier(identifier: str) -> str: + return ".".join(_quote_identifier(part) for part in identifier.split(".")) + + +def _sql_string(value: str) -> str: + return "'" + value.replace("'", "''") + "'" + + +def _render_sql( + contract: dict[str, Any], + table_columns: list[dict[str, Any]], + partition_fields: list[str], + record_key_fields: list[str], +) -> dict[str, str]: + table = contract["table"] + source = contract["source"] + runtime = contract["runtime"] + identity = contract["identity"] + + definitions = [ + f" {_quote_identifier(column['name'])} {column['type']}" + + ("" if column["nullable"] else " NOT NULL") + for column in table_columns + ] + if identity["mode"] == "stable_key": + key_sql = ", ".join(_quote_identifier(field) for field in record_key_fields) + definitions.append(f" PRIMARY KEY ({key_sql}) NOT ENFORCED") + + table_ddl = ( + f"CREATE TABLE {_quote_qualified_identifier(table['name'])} (\n" + + ",\n".join(definitions) + + "\n)" + ) + if partition_fields: + partition_sql = ", ".join(_quote_identifier(field) for field in partition_fields) + table_ddl += f"\nPARTITIONED BY ({partition_sql})" + table_ddl += ( + "\nWITH (\n" + " 'connector' = 'hudi',\n" + f" 'path' = {_sql_string(table['path'])},\n" + " 'table.type' = 'COPY_ON_WRITE',\n" + " 'write.operation' = 'insert',\n" + " 'write.insert.cluster' = 'false'\n" + ");" + ) + + field_sql = ",\n ".join(_quote_identifier(column["name"]) for column in table_columns) + insert_sql = ( + f"INSERT INTO {_quote_qualified_identifier(table['name'])} (\n {field_sql}\n)\n" + f"SELECT\n {field_sql}\n" + f"FROM {_quote_qualified_identifier(source['table'])};" + ) + runtime_sql = ( + "SET 'execution.runtime-mode' = 'streaming';\n" + "SET 'execution.checkpointing.interval' = " + f"'{runtime['checkpoint_interval_ms']} ms';" + ) + return { + "runtime_sql": runtime_sql, + "table_ddl": table_ddl, + "insert_sql": insert_sql, + "combined_sql": f"{runtime_sql}\n\n{table_ddl}\n\n{insert_sql}", + } + + +def assess_design( + contract: dict[str, Any], manifest: dict[str, Any] +) -> dict[str, Any]: + """Validate one explicit PR2 contract and return deterministic evidence.""" + + assessment = Assessment() + manifest_errors = validate_manifest(manifest) + if manifest_errors: + assessment.add( + "FLINK_BASELINE_EVIDENCE_INVALID", "; ".join(manifest_errors) + ) + return { + "status": assessment.final_status(), + "executable_eligible": False, + "finding_codes": [finding["code"] for finding in assessment.findings], + "findings": assessment.findings, + "advisory_codes": assessment.advisories, + "validation_evidence": { + "design_contract_schema": CONTRACT_SCHEMA_VERSION, + }, + } + + _check_keys( + contract, + { + "baseline_id", + "contract_schema", + "flink_version", + "hudi_version", + "identity", + "runtime", + "safety", + "source", + "table", + "write", + }, + "contract", + assessment, + ) + if contract.get("contract_schema") != CONTRACT_SCHEMA_VERSION: + assessment.add( + "FLINK_DESIGN_CONTRACT_INVALID", + f"contract_schema must be {CONTRACT_SCHEMA_VERSION}", + ) + if contract.get("baseline_id") != manifest.get("baseline_id"): + assessment.add("FLINK_VERSION_UNVERIFIED", "The capability baseline does not match") + if not _non_empty_string(contract.get("hudi_version")) or not _non_empty_string( + contract.get("flink_version") + ): + assessment.add("FLINK_VERSION_REQUIRED", "Concrete Hudi and Flink versions are required") + else: + if contract["hudi_version"] != manifest["hudi"]["version"]: + assessment.add("FLINK_VERSION_UNVERIFIED", "The Hudi version is outside the baseline") + if not re.fullmatch(r"1\.20(?:\.\d+|\.x)?", contract["flink_version"]): + assessment.add("FLINK_VERSION_UNVERIFIED", "The Flink version is outside the baseline") + + safety = _dict(contract.get("safety"), "safety", assessment) + _check_keys( + safety, + { + "external_catalog", + "mutability", + "replay_behavior", + "table_lifecycle", + "writer_model", + }, + "safety", + assessment, + ) + lifecycle = safety.get("table_lifecycle") + if lifecycle is None: + assessment.add("FLINK_TABLE_LIFECYCLE_REQUIRED", "Table lifecycle is required") + elif lifecycle == "existing": + assessment.add("FLINK_EXISTING_TABLE_DEFERRED", "PR2 supports only a new table") + elif lifecycle != "new": + assessment.add("FLINK_TABLE_LIFECYCLE_REQUIRED", "Table lifecycle is unresolved") + writer_model = safety.get("writer_model") + if writer_model is None: + assessment.add("FLINK_WRITER_MODEL_UNRESOLVED", "Writer topology is required") + elif writer_model == "multi_writer": + assessment.add("FLINK_MULTI_WRITER_REVIEW", "PR2 supports only a confirmed single writer") + elif writer_model != "single_writer": + assessment.add("FLINK_WRITER_MODEL_UNRESOLVED", "Writer topology is unresolved") + external_catalog = safety.get("external_catalog") + if external_catalog is True: + assessment.add("FLINK_EXTERNAL_CATALOG_REVIEW", "Catalog composition is deferred") + elif external_catalog is not False: + assessment.add("FLINK_CATALOG_REQUIREMENT_UNRESOLVED", "Catalog requirement is unresolved") + mutability = safety.get("mutability") + if mutability == "mutable": + assessment.add("FLINK_MUTABLE_COW_DEFERRED", "Mutable COW is deferred") + elif mutability != "append_only": + assessment.add("FLINK_MUTABILITY_REQUIRED", "Append-only input must be confirmed") + replay_behavior = safety.get("replay_behavior") + if replay_behavior == "must_collapse": + assessment.add("FLINK_REPLAY_IDEMPOTENCE_DEFERRED", "Replay deduplication needs upsert") + elif replay_behavior not in {"cannot_occur", "duplicates_acceptable"}: Review Comment: [P2] Validate replay_behavior’s type before set membership Setting `safety.replay_behavior` to `[]` or `{}` in an otherwise valid fixture raises an uncaught `TypeError` here because these JSON values are unhashable. The CLI then emits a traceback instead of the structured assessment, losing any independently accumulated findings and final status. Please check that this value is a string before testing membership and report malformed values through the existing assessment mechanism; cover both array and object inputs. -- 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]
