fhan688 commented on code in PR #20198: URL: https://github.com/apache/hudi/pull/20198#discussion_r4227646613
########## 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: Thank you for catching this. The validator now blocks `BYTES`, `BINARY`, and `VARBINARY` only when used as record-key or partition fields, while keeping them valid for payload columns. I also added Python coverage for all three types and pinned Hudi 1.2.0 evidence reproducing the object-identity routing behavior. -- 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]
