fhan688 commented on code in PR #20391: URL: https://github.com/apache/hudi/pull/20391#discussion_r4334140907
########## hudi-agent-gateway/skills/hudi-architect/validate_flink_existing_table.py: ########## @@ -0,0 +1,1119 @@ +#!/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. + +"""Reconcile a bounded Flink design with structured existing-table evidence. + +This is deliberately an assessment-only path. It never reads a live table, +executes supplied evidence, mutates table state, or renders SQL. The caller +must supply an explicit PR2, PR3, or PR4 desired contract plus normalized, +provenance-bearing observations. +""" + +from __future__ import annotations + +import argparse +import json +import re +import sys +from dataclasses import dataclass +from datetime import datetime +from pathlib import Path +from typing import Any, TypeGuard + +from validate_flink_capabilities import ( + DEFAULT_MANIFEST, + load_manifest, + validate_manifest, + validation_evidence, +) +from validate_flink_design import Assessment, _path_validation_error, assess_design + +RECONCILIATION_SCHEMA_VERSION = 1 +SHA256_PATTERN = re.compile(r"[0-9a-f]{64}") +RFC3339_PATTERN = re.compile( + r"\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}(?:\.\d+)?(?:Z|[+-]\d{2}:\d{2})" +) +FIELD_NAME_PATTERN = re.compile(r"[A-Za-z_][A-Za-z0-9_]*") +ALLOWED_TRANSITIONS = { + "same_job_restore", + "cold_start_new_job", + "flink_writer_takeover", + "cross_engine_takeover", + "unknown", +} +ALLOWED_EVIDENCE_FACTS = { + "hoodie_properties": { + "base_file_format", + "hive_style_partitioning", + "initial_table_version", + "key_generator_type", + "ordering_fields", + "partition_fields", + "record_key_fields", + "record_merge_mode", + "table_name", + "table_type", + "table_version", + "timeline_layout_version", + "url_encode_partitioning", + }, + "hudi_table_identity": {"table_path"}, + "hudi_table_schema": {"columns"}, + "flink_table_ddl": { + "columns", + "ordering_fields", + "partition_fields", + "record_key_fields", + "table_name", + "table_path", + "table_type", + }, + "flink_writer_config": { + "checkpoint_recovery_available", + "compaction_owner", + "design_contract_schema", + "index_bootstrap_enabled", + "index_global", + "index_state_ttl_days", + "index_type", + "operation", + "timestamp_key_generator", + }, + "timeline_summary": {"pending_instants"}, + "writer_inventory": { + "concurrent_writers", + "previous_writer_engine", + "previous_writer_stopped", + }, +} +REQUIRED_EVIDENCE_KINDS = { + "hoodie_properties", + "hudi_table_identity", + "hudi_table_schema", + "timeline_summary", + "writer_inventory", +} +REQUIRED_TABLE_FACTS = { + "base_file_format", + "key_generator_type", + "initial_table_version", + "ordering_fields", + "partition_fields", + "record_key_fields", + "table_name", + "table_type", + "table_version", + "timeline_layout_version", +} +LIST_FACTS = { + "ordering_fields", + "partition_fields", + "pending_instants", + "record_key_fields", +} +UPPERCASE_FACTS = { + "base_file_format", + "key_generator_type", + "compaction_owner", + "index_type", + "operation", + "previous_writer_engine", + "record_merge_mode", + "table_type", +} +KNOWN_KEY_GENERATORS = { + "AUTO_RECORD", + "AUTO_RECORD_AVRO", + "COMPLEX", + "COMPLEX_AVRO", + "GLOBAL_DELETE", + "GLOBAL_DELETE_AVRO", + "NON_PARTITION", + "NON_PARTITION_AVRO", + "SIMPLE", + "SIMPLE_AVRO", + "TIMESTAMP", + "TIMESTAMP_AVRO", +} +SENSITIVE_KEY = re.compile( + r"(?:access[_-]?key|api[_-]?key|credential|password|secret|signature|token)", + re.IGNORECASE, +) + + +@dataclass(frozen=True) +class Observation: + """One normalized value and the evidence items that establish it.""" + + value: Any + source_ids: tuple[str, ...] + + +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_EXISTING_EVIDENCE_INVALID", + f"{field} contains unsupported fields: {unknown}", + ) + + +def _non_empty_string(value: Any) -> TypeGuard[str]: + return isinstance(value, str) and bool(value.strip()) + + +def _string_list(value: Any, field: str, assessment: Assessment) -> list[str] | None: + if not isinstance(value, list) or not all(_non_empty_string(item) for item in value): + assessment.add( + "FLINK_EXISTING_EVIDENCE_INVALID", + f"{field} must be an array of non-empty strings", + ) + return None + if len(value) != len(set(value)): + assessment.add( + "FLINK_EXISTING_EVIDENCE_INVALID", + f"{field} must not contain duplicates", + ) + return None + return value + + +def _valid_observed_at(value: Any) -> bool: + if not isinstance(value, str) or RFC3339_PATTERN.fullmatch(value) is None: + return False + try: + parsed = datetime.fromisoformat(value.replace("Z", "+00:00")) + except ValueError: + return False + return parsed.tzinfo is not None + + +def _normalize_type(value: Any) -> str | None: + if not _non_empty_string(value): + return None + normalized = " ".join(value.upper().split()) + normalized = re.sub(r"\s*,\s*", ",", normalized) + normalized = re.sub(r"\s*\(\s*", "(", normalized) + normalized = re.sub(r"\s*\)", ")", normalized) + return "INT" if normalized == "INTEGER" else normalized + + +def _normalize_columns( + value: Any, field: str, assessment: Assessment +) -> list[dict[str, Any]] | None: + if not isinstance(value, list) or not value: + assessment.add( + "FLINK_EXISTING_EVIDENCE_INVALID", + f"{field} must be a non-empty array", + ) + return None + columns: list[dict[str, Any]] = [] + names: set[str] = set() + for index, raw_column in enumerate(value): + item_field = f"{field}[{index}]" + if not isinstance(raw_column, dict): + assessment.add( + "FLINK_EXISTING_EVIDENCE_INVALID", f"{item_field} must be an object" + ) + continue + _check_keys(raw_column, {"name", "nullable", "type"}, item_field, assessment) + name = raw_column.get("name") + normalized_type = _normalize_type(raw_column.get("type")) + nullable = raw_column.get("nullable") + if ( + not _non_empty_string(name) + or FIELD_NAME_PATTERN.fullmatch(name) is None + or normalized_type is None + or not isinstance(nullable, bool) + ): + assessment.add( + "FLINK_EXISTING_EVIDENCE_INVALID", + f"{item_field} needs a valid name, type, and explicit nullable flag", + ) + continue + if name in names: + assessment.add( + "FLINK_EXISTING_EVIDENCE_INVALID", + f"{field} contains duplicate field {name!r}", + ) + continue + names.add(name) + columns.append({"name": name, "type": normalized_type, "nullable": nullable}) + return columns if len(columns) == len(value) else None + + +def _normalize_fact( + kind: str, name: str, value: Any, field: str, assessment: Assessment +) -> Any | None: + if name in LIST_FACTS: + values = _string_list(value, field, assessment) + if values is None: + return None + if name == "partition_fields": + # HoodieTableConfig stores custom partition-key types after a colon. + return [item.split(":", 1)[0] for item in values] + return values + if name == "columns": + return _normalize_columns(value, field, assessment) + if name == "timestamp_key_generator": + allowed = {"timestamp_type", "input_time_unit", "output_date_format", "output_timezone"} + if ( + not isinstance(value, dict) + or not set(value) <= allowed + or not all(_non_empty_string(item) for item in value.values()) + ): + assessment.add( + "FLINK_EXISTING_EVIDENCE_INVALID", + f"{field} must contain only normalized timestamp key-generator settings", + ) + return None + return value + if name in { + "design_contract_schema", + "index_state_ttl_days", + "initial_table_version", + "table_version", + "timeline_layout_version", + }: + if not isinstance(value, int) or isinstance(value, bool) or value < 0: + assessment.add( + "FLINK_EXISTING_EVIDENCE_INVALID", + f"{field} must be a non-negative integer", + ) + return None + return value + if name in { + "checkpoint_recovery_available", + "concurrent_writers", + "previous_writer_stopped", + "hive_style_partitioning", + "index_bootstrap_enabled", + "index_global", + "url_encode_partitioning", + }: + if not isinstance(value, bool): + assessment.add( + "FLINK_EXISTING_EVIDENCE_INVALID", f"{field} must be a boolean" + ) + return None + return value + if not _non_empty_string(value): + assessment.add( + "FLINK_EXISTING_EVIDENCE_INVALID", f"{field} must be a non-empty string" + ) + return None + normalized = value.strip() + if name == "table_path" and _path_validation_error(normalized) is not None: + assessment.add( + "FLINK_EXISTING_EVIDENCE_INVALID", + f"{field} must be a credential-free path", + ) + return None + return normalized.upper() if name in UPPERCASE_FACTS else normalized + + +def _parse_evidence( + raw_evidence: Any, assessment: Assessment +) -> tuple[dict[str, dict[str, Observation]], list[dict[str, Any]]]: + observations: dict[str, dict[str, Observation]] = {} + summaries: list[dict[str, Any]] = [] + if not isinstance(raw_evidence, list): + assessment.add("FLINK_EXISTING_EVIDENCE_INVALID", "evidence must be an array") + return observations, summaries + + evidence_ids: set[str] = set() + collected: dict[tuple[str, str], list[tuple[Any, str]]] = {} + scoped_paths: list[tuple[str, str]] = [] + for index, raw_item in enumerate(raw_evidence): + field = f"evidence[{index}]" + if not isinstance(raw_item, dict): + assessment.add("FLINK_EXISTING_EVIDENCE_INVALID", f"{field} must be an object") + continue + _check_keys( + raw_item, + {"evidence_id", "facts", "kind", "observed_at", "sha256", "table_path"}, + field, + assessment, + ) + evidence_id = raw_item.get("evidence_id") + kind = raw_item.get("kind") + observed_at = raw_item.get("observed_at") + digest = raw_item.get("sha256") + table_path = raw_item.get("table_path") + facts = raw_item.get("facts") + valid_header = True + if not _non_empty_string(evidence_id) or evidence_id in evidence_ids: + assessment.add( + "FLINK_EXISTING_EVIDENCE_INVALID", + f"{field}.evidence_id must be a unique non-empty string", + ) + valid_header = False + else: + evidence_ids.add(evidence_id) + supported_kind = isinstance(kind, str) and kind in ALLOWED_EVIDENCE_FACTS + if not supported_kind: + assessment.add( + "FLINK_EXISTING_EVIDENCE_INVALID", + f"{field}.kind is not supported", + ) + valid_header = False + if not _valid_observed_at(observed_at): + assessment.add( + "FLINK_EXISTING_EVIDENCE_INVALID", + f"{field}.observed_at must be RFC 3339 with a timezone", + ) + valid_header = False + if not isinstance(digest, str) or SHA256_PATTERN.fullmatch(digest) is None: + assessment.add( + "FLINK_EXISTING_EVIDENCE_INVALID", + f"{field}.sha256 must be a lowercase SHA-256", + ) + valid_header = False + if not _non_empty_string(table_path) or _path_validation_error(table_path) is not None: + assessment.add( + "FLINK_EXISTING_EVIDENCE_INVALID", + f"{field}.table_path must be a credential-free path", + ) + valid_header = False + if not isinstance(facts, dict): + assessment.add( + "FLINK_EXISTING_EVIDENCE_INVALID", f"{field}.facts must be an object" + ) + continue + if any(SENSITIVE_KEY.search(str(key)) for key in facts): + assessment.add( + "FLINK_EXISTING_EVIDENCE_INVALID", + f"{field}.facts contains a credential-shaped field", + ) + continue + if isinstance(kind, str) and supported_kind: + _check_keys(facts, ALLOWED_EVIDENCE_FACTS[kind], f"{field}.facts", assessment) + if not valid_header or not isinstance(kind, str) or not isinstance(evidence_id, str): + continue + assert isinstance(table_path, str) + scoped_paths.append((table_path, evidence_id)) + normalized_facts: list[str] = [] + for name in sorted(facts): + if name not in ALLOWED_EVIDENCE_FACTS[kind]: + continue + normalized = _normalize_fact( + kind, name, facts[name], f"{field}.facts.{name}", assessment + ) + if normalized is None: + continue + collected.setdefault((kind, name), []).append((normalized, evidence_id)) + normalized_facts.append(name) + summaries.append( + { + "evidence_id": evidence_id, + "fact_names": normalized_facts, + "kind": kind, + "observed_at": observed_at, + "sha256": digest, + "table_path": table_path, + } + ) + + for (kind, name), candidates in sorted(collected.items()): + first_value = candidates[0][0] + source_ids = tuple(candidate[1] for candidate in candidates) + if any(candidate[0] != first_value for candidate in candidates[1:]): + assessment.add( + "FLINK_EXISTING_EVIDENCE_CONFLICT", + f"Evidence sources {', '.join(source_ids)} disagree about {kind}.{name}", + ) + continue + observations.setdefault(kind, {})[name] = Observation(first_value, source_ids) Review Comment: Thanks for pointing this out. Evidence is now scoped against the desired table path before observations are aggregated. Out-of-scope evidence retains its provenance and scope finding but is excluded from target-table comparisons, so the reproduced case remains unverified instead of incorrectly requiring migration. I added the corresponding regression test. -- 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]
