You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@hudi.apache.org by "fhan688 (via GitHub)" <gi...@apache.org> on 2026/10/19 07:50:52 UTC
[PR] feat(agent): add existing-table reconciliation to Hudi Architect [hudi]
fhan688 opened a new pull request, #20391:
URL: https://github.com/apache/hudi/pull/20391
### Describe the issue this Pull Request addresses
Related to #19779.
The previous Flink SQL work established bounded design contracts for append-only COW, mutable COW, and mutable MOR sinks. Applying these contracts to an existing Hudi table additionally requires checking persisted table metadata, physical schema, and writer-transition evidence.
This PR adds an assessment-only reconciliation path. It distinguishes verified matches, known conflicts, and unresolved evidence without treating an existing-table assessment as permission to generate or execute SQL.
### Summary and Changelog
- Add a deterministic existing-table reconciliation validator for the pinned Hudi 1.2.0 / Flink 1.20 baseline, reusing desired contract schemas 1, 2, and 3.
- Introduce reconciliation schema 1 with structured evidence, unique evidence IDs, observation timestamps, source-artifact SHA-256 values, and explicit table-path scope.
- Compare the desired contract against authoritative Hudi table configuration and physical schema. Flink DDL cannot override these sources, and conflicting evidence remains visible.
- Keep persisted table properties separate from writer configuration and time-sensitive operational observations.
- Require recoverable-state evidence and matching applicable writer settings for same-job restores. Require the previous writer to be stopped for cold starts and retain an index-bootstrap deployment check for mutable workloads.
- Preserve independent findings and select one deterministic final status. Distinguish schema conflicts from unverified schema evolution and unknown key-generator semantics.
- Report `COMPATIBLE`, `CONFLICT`, or `UNVERIFIED`, together with a bounded change disposition. Classify an assessment as `DEPLOYMENT_CONFIGURATION_CHANGE` only when writer-configuration differences are the only remaining findings.
- Keep existing-table assessments non-executable, including successful matches. Do not generate SQL, table mutations, or migration commands.
- Extend the capability manifest to schema 5 and update the skill instructions, reference documentation, regression tests, and Gateway CI checks.
Validation performed locally:
- Python 3.11: 371 tests passed and 4 live-stack tests skipped.
- Python 3.12: 371 tests passed and 4 live-stack tests skipped.
- Pinned Java fixtures: 23 tests passed, including existing-table type precedence and record-key/ordering-field enrichment.
- Ruff, mypy, capability source verification, configuration-key validation, CLI smoke checks, Helm lint/template checks, and skill validation passed.
- Independent Agent forward evaluation covered incomplete restore evidence, concurrent-writer findings combined with missing evidence, and an executable-SQL request for a compatible existing table.
### Impact
This adds a user-facing assessment capability to Hudi Architect. It does not change Hudi runtime behavior, storage formats, connector configuration defaults, or public runtime APIs.
Existing-table reconciliation consumes normalized, user-confirmed evidence; it does not inspect a live deployment or independently authenticate the supplied artifacts.
`CONFIG_VALIDATED` means the bounded static assessment is complete. It does not establish production readiness, and `executable_eligible` remains false.
Separate-writer takeover, cross-engine takeover, schema evolution, table upgrades, external catalogs, multi-writer operation, and unresolved pending instants are not automatically approved. Writer inventory and timeline state must be rechecked before deployment.
The existing new-table SQL generation paths remain covered by regression tests.
### Risk Level
medium
Incorrect reconciliation could produce misleading compatibility guidance. The implementation mitigates this through explicit evidence requirements, source precedence, conservative handling of unsupported cases, preservation of independent findings, and an assessment-only output boundary.
Verification includes positive and negative reconciliation tests, pinned Hudi/Flink Java behavior tests, full Gateway regression tests on Python 3.11 and 3.12, and independent Agent forward evaluation.
Live-stack integration tests were not run. The synthetic fixtures and static assessments do not prove production recovery or deployment safety.
### Documentation Update
Updated the Hudi Architect skill instructions, README, runbook, capability references, question flow, decision guidance, warnings, and output templates.
Added a dedicated existing-table reconciliation reference documenting evidence requirements, writer transitions, comparison semantics, deployment checks, and unsupported cases.
No Hudi runtime configuration options or defaults are changed. This PR does not include a Hudi website documentation update.
### Contributor's checklist
- [ ] Read through [contributor's guide](https://hudi.apache.org/contribute/how-to-contribute)
- [x] Enough context is provided in the sections above
- [x] Adequate tests were added if applicable
--
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(agent): add existing-table reconciliation to Hudi Architect [hudi]
Posted by "codecov-commenter (via GitHub)" <gi...@apache.org>.
codecov-commenter commented on PR #20391:
URL: https://github.com/apache/hudi/pull/20391#issuecomment-6264338610
## [Codecov](https://app.codecov.io/gh/apache/hudi/pull/20391?dropdown=coverage&src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) Report
:white_check_mark: All modified and coverable lines are covered by tests.
:white_check_mark: Project coverage is 80.75%. Comparing base ([`e95ecf5`](https://app.codecov.io/gh/apache/hudi/commit/e95ecf581c5457cbca9708fad265df2a2db7925e?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)) to head ([`a19576e`](https://app.codecov.io/gh/apache/hudi/commit/a19576e0119fd911b98e4cc473f73bc69503962f?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)).
:warning: Report is 2 commits behind head on master.
<details><summary>Additional details and impacted files</summary>
```diff
@@ Coverage Diff @@
## master #20391 +/- ##
=========================================
Coverage 80.75% 80.75%
- Complexity 35315 35316 +1
=========================================
Files 2562 2562
Lines 144143 144143
Branches 17560 17560
=========================================
+ Hits 116407 116409 +2
+ Misses 19760 19759 -1
+ Partials 7976 7975 -1
```
| [Components](https://app.codecov.io/gh/apache/hudi/pull/20391/components?src=pr&el=components&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Coverage Δ | |
|---|---|---|
| [hudi-common](https://app.codecov.io/gh/apache/hudi/pull/20391/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `84.18% <ø> (ø)` | |
| [hudi-client](https://app.codecov.io/gh/apache/hudi/pull/20391/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `83.99% <ø> (+<0.01%)` | :arrow_up: |
| [hudi-flink](https://app.codecov.io/gh/apache/hudi/pull/20391/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `86.07% <ø> (ø)` | |
| [hudi-spark-datasource](https://app.codecov.io/gh/apache/hudi/pull/20391/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `74.19% <ø> (ø)` | |
| [hudi-utilities](https://app.codecov.io/gh/apache/hudi/pull/20391/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `78.35% <ø> (ø)` | |
| [hudi-cli](https://app.codecov.io/gh/apache/hudi/pull/20391/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `70.80% <ø> (ø)` | |
| [hudi-hadoop](https://app.codecov.io/gh/apache/hudi/pull/20391/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `71.17% <ø> (ø)` | |
| [hudi-sync](https://app.codecov.io/gh/apache/hudi/pull/20391/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `76.78% <ø> (ø)` | |
| [hudi-io](https://app.codecov.io/gh/apache/hudi/pull/20391/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `81.57% <ø> (ø)` | |
| [hudi-timeline-service](https://app.codecov.io/gh/apache/hudi/pull/20391/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `83.41% <ø> (ø)` | |
| [hudi-cloud](https://app.codecov.io/gh/apache/hudi/pull/20391/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `81.01% <ø> (ø)` | |
| [hudi-kafka-connect](https://app.codecov.io/gh/apache/hudi/pull/20391/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `57.90% <ø> (ø)` | |
| [Flag](https://app.codecov.io/gh/apache/hudi/pull/20391/flags?src=pr&el=flags&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Coverage Δ | |
|---|---|---|
| [hadoop-mr-java-client](https://app.codecov.io/gh/apache/hudi/pull/20391/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `43.94% <ø> (+<0.01%)` | :arrow_up: |
Flags with carried forward coverage won't be shown. [Click here](https://docs.codecov.io/docs/carryforward-flags?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#carryforward-flags-in-the-pull-request-comment) to find out more.
[see 1 file with indirect coverage changes](https://app.codecov.io/gh/apache/hudi/pull/20391/indirect-changes?src=pr&el=tree-more&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)
</details>
<details><summary> :rocket: New features to boost your workflow: </summary>
- :snowflake: [Test Analytics](https://docs.codecov.com/docs/test-analytics): Detect flaky tests, report on failures, and find test suite problems.
- :package: [JS Bundle Analysis](https://docs.codecov.com/docs/javascript-bundle-analysis): Save yourself from yourself by tracking and limiting bundle sizes in JS merges.
</details>
--
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(agent): add existing-table reconciliation to Hudi Architect [hudi]
Posted by "danny0405 (via GitHub)" <gi...@apache.org>.
danny0405 commented on code in PR #20391:
URL: https://github.com/apache/hudi/pull/20391#discussion_r4332527259
##########
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:
[P2] Exclude evidence for another table from authoritative comparisons
The observations are populated before the scope check, and remain usable after that check fails. Reproduction: in `compatible_mutable_cow.json`, change only the schema evidence's `table_path` to `file:///tmp/unrelated-table` and its `id` type to `BIGINT`. The report becomes `BLOCKED` / `CONFLICT` / `MIGRATION_REQUIRED` because `_compare_schema` treats that unrelated table's schema as authoritative for the requested table. There is actually no schema evidence establishing a conflict at the requested path. Please retain the scope finding/provenance but exclude out-of-scope facts from target-table comparisons, so this case remains unverified/incomplete rather than claiming a migration is required.
##########
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)
+ if scoped_paths:
+ first_path = scoped_paths[0][0]
+ source_ids = tuple(item[1] for item in scoped_paths)
+ if any(item[0] != first_path for item in scoped_paths[1:]):
+ assessment.add(
+ "FLINK_EXISTING_EVIDENCE_CONFLICT",
+ "Evidence items do not identify one common table path",
+ )
+ else:
+ observations.setdefault("evidence_scope", {})["table_path"] = Observation(
+ first_path, source_ids
+ )
+ return observations, summaries
+
+
+def _add_comparison(
+ comparisons: list[dict[str, Any]],
+ field: str,
+ desired: Any,
+ observed: Observation | None,
+ *,
+ mismatch_code: str,
+ assessment: Assessment,
+ mismatch_result: str = "CONFLICT",
+) -> None:
+ if observed is None:
+ comparisons.append(
+ {
+ "desired": desired,
+ "field": field,
+ "observed": None,
+ "result": "UNVERIFIED",
+ "source_ids": [],
+ }
+ )
+ assessment.add(
+ "FLINK_EXISTING_EVIDENCE_REQUIRED", f"Evidence is required for {field}"
+ )
+ return
+ result = "MATCH" if desired == observed.value else mismatch_result
+ comparisons.append(
+ {
+ "desired": desired,
+ "field": field,
+ "observed": observed.value,
+ "result": result,
+ "source_ids": list(observed.source_ids),
+ }
+ )
+ if result != "MATCH":
+ assessment.add(mismatch_code, f"Existing-table value differs at {field}")
+
+
+def _expected_key_generator(desired: dict[str, Any]) -> set[str]:
+ identity = desired.get("identity")
+ table = desired.get("table")
+ identity = identity if isinstance(identity, dict) else {}
+ table = table if isinstance(table, dict) else {}
+ partition_fields = table.get("partition_fields")
+ partition_fields = partition_fields if isinstance(partition_fields, list) else []
+ record_keys = identity.get("record_key_fields", [])
+ record_keys = record_keys if isinstance(record_keys, list) else []
+ if not partition_fields:
+ return {"NON_PARTITION", "NON_PARTITION_AVRO"}
+ if len(record_keys) <= 1 and len(partition_fields) <= 1:
+ if _partition_field_type(table) in {"DATE", "TIMESTAMP"}:
+ return {"TIMESTAMP", "TIMESTAMP_AVRO"}
+ return {"SIMPLE", "SIMPLE_AVRO"}
+ return {"COMPLEX", "COMPLEX_AVRO"}
+
+
+def _partition_field_type(table: dict[str, Any]) -> str | None:
+ partitions = table.get("partition_fields")
+ columns = table.get("columns")
+ if not isinstance(partitions, list) or len(partitions) != 1 or not isinstance(columns, list):
+ return None
+ for column in columns:
+ if not isinstance(column, dict) or column.get("name") != partitions[0]:
+ continue
+ field_type = _normalize_type(column.get("type")) or ""
+ if field_type == "DATE":
+ return "DATE"
+ if field_type.startswith("TIMESTAMP"):
+ return "TIMESTAMP"
+ return None
+
+
+def _expected_timestamp_settings(table: dict[str, Any]) -> dict[str, str]:
+ partitions = table.get("partition_fields")
+ columns = table.get("columns")
+ if not isinstance(partitions, list) or len(partitions) != 1 or not isinstance(columns, list):
+ return {}
+ for column in columns:
+ if not isinstance(column, dict) or column.get("name") != partitions[0]:
+ continue
+ field_type = _normalize_type(column.get("type")) or ""
+ if field_type == "DATE":
+ return {
+ "timestamp_type": "SCALAR",
+ "input_time_unit": "DAYS",
+ "output_date_format": "yyyyMMdd",
+ "output_timezone": "UTC",
+ }
+ if field_type in {"TIMESTAMP(0)", "TIMESTAMP(3)"}:
+ return {
+ "timestamp_type": (
+ "UNIX_TIMESTAMP" if field_type.endswith("(0)") else "EPOCHMILLISECONDS"
+ ),
+ "output_date_format": "yyyyMMddHH",
+ "output_timezone": "UTC",
+ }
+ return {}
+
+
+def _compare_schema(
+ desired_columns: Any,
+ observed: Observation | None,
+ assessment: Assessment,
+ comparisons: list[dict[str, Any]],
+) -> None:
+ if observed is None:
+ _add_comparison(
+ comparisons,
+ "schema.columns",
+ desired_columns,
+ None,
+ mismatch_code="FLINK_EXISTING_SCHEMA_CONFLICT",
+ assessment=assessment,
+ )
+ return
+ desired = [
+ {
+ "name": column.get("name"),
+ "type": _normalize_type(column.get("type")),
+ "nullable": column.get("nullable"),
+ }
+ for column in (desired_columns if isinstance(desired_columns, list) else [])
+ if isinstance(column, dict) and _non_empty_string(column.get("name"))
+ ]
+ actual = observed.value
+ result = "MATCH"
+ mismatch_code: str | None = None
+ if desired != actual:
+ desired_by_name = {
+ column["name"]: column for column in desired if isinstance(column["name"], str)
+ }
+ actual_by_name = {column["name"]: column for column in actual}
+ missing_required = sorted(
+ name
+ for name in set(desired_by_name) - set(actual_by_name)
+ if desired_by_name[name]["nullable"] is not True
+ )
+ missing_nullable = sorted(
+ name
+ for name in set(desired_by_name) - set(actual_by_name)
+ if desired_by_name[name]["nullable"] is True
+ )
+ incompatible_types = sorted(
+ name
+ for name in set(desired_by_name) & set(actual_by_name)
+ if desired_by_name[name]["type"] != actual_by_name[name]["type"]
+ )
+ nullable_conflicts = sorted(
+ name
+ for name in set(desired_by_name) & set(actual_by_name)
+ if desired_by_name[name]["nullable"] is True
+ and actual_by_name[name]["nullable"] is False
+ )
+ nullable_evolution = sorted(
+ name
+ for name in set(desired_by_name) & set(actual_by_name)
+ if desired_by_name[name]["nullable"] is False
+ and actual_by_name[name]["nullable"] is True
+ )
+ extra = sorted(set(actual_by_name) - set(desired_by_name))
+ if missing_required or incompatible_types or nullable_conflicts:
+ result = "CONFLICT"
+ mismatch_code = "FLINK_EXISTING_SCHEMA_CONFLICT"
+ if missing_nullable or nullable_evolution or extra:
+ assessment.add(
+ "FLINK_EXISTING_SCHEMA_EVOLUTION_UNVERIFIED",
+ "Schema additions, nullability changes or extra physical columns require review",
+ )
+ if result != "CONFLICT":
+ result = "UNVERIFIED"
+ mismatch_code = "FLINK_EXISTING_SCHEMA_EVOLUTION_UNVERIFIED"
+ comparisons.append(
+ {
+ "desired": desired,
+ "field": "schema.columns",
+ "observed": actual,
+ "result": result,
+ "source_ids": list(observed.source_ids),
+ }
+ )
+ if mismatch_code:
+ assessment.add(mismatch_code, "Existing physical schema differs from the desired schema")
+
+
+def _compare_ddl_with_authority(
+ observations: dict[str, dict[str, Observation]], assessment: Assessment
+) -> None:
+ ddl = observations.get("flink_table_ddl", {})
+ table = {
+ **observations.get("hoodie_properties", {}),
+ **observations.get("hudi_table_identity", {}),
+ }
+ schema = observations.get("hudi_table_schema", {})
+ for name in sorted(set(ddl) & set(table)):
+ if ddl[name].value != table[name].value:
+ assessment.add(
+ "FLINK_EXISTING_EVIDENCE_CONFLICT",
+ f"Flink DDL disagrees with authoritative table evidence for {name}",
+ )
+ if "columns" in ddl and "columns" in schema and ddl["columns"].value != schema["columns"].value:
+ assessment.add(
+ "FLINK_EXISTING_EVIDENCE_CONFLICT",
+ "Flink DDL disagrees with the authoritative Hudi table schema",
+ )
+
+
+def _expected_writer_state(desired: dict[str, Any]) -> dict[str, Any]:
+ """Return the bounded writer state that must survive a stateful restore."""
+
+ raw_write = desired.get("write")
+ write = raw_write if isinstance(raw_write, dict) else {}
+ expected: dict[str, Any] = {
+ "design_contract_schema": desired.get("contract_schema"),
+ "operation": str(write.get("operation", "")).upper(),
+ }
+ if desired.get("contract_schema") in (2, 3):
+ raw_index = write.get("index")
+ index = raw_index if isinstance(raw_index, dict) else {}
+ expected.update(
+ {
+ "index_bootstrap_enabled": index.get("bootstrap_enabled"),
+ "index_global": index.get("global"),
+ "index_state_ttl_days": index.get("state_ttl_days"),
+ "index_type": str(index.get("type", "")).upper(),
+ }
+ )
+ if desired.get("contract_schema") == 3:
+ raw_services = desired.get("table_services")
+ services = raw_services if isinstance(raw_services, dict) else {}
+ raw_compaction = services.get("compaction")
+ compaction = raw_compaction if isinstance(raw_compaction, dict) else {}
+ expected["compaction_owner"] = str(compaction.get("owner", "")).upper()
+ return expected
+
+
+def assess_reconciliation(
+ envelope: dict[str, Any], manifest: dict[str, Any]
+) -> dict[str, Any]:
+ """Return one deterministic, non-executable existing-table assessment."""
+
+ assessment = Assessment()
+ manifest_errors = validate_manifest(manifest)
+ if manifest_errors:
+ assessment.add("FLINK_BASELINE_EVIDENCE_INVALID", "; ".join(manifest_errors))
+ return {
+ "status": assessment.final_status(),
+ "reconciliation_result": "UNVERIFIED",
+ "reconciliation_complete": False,
+ "executable_eligible": False,
+ "finding_codes": [finding["code"] for finding in assessment.findings],
+ "findings": assessment.findings,
+ "comparisons": [],
+ "evidence_summary": [],
+ }
+
+ _check_keys(
+ envelope,
+ {
+ "baseline_id",
+ "desired_contract",
+ "evidence",
+ "reconciliation_schema",
+ "writer_transition",
+ },
+ "reconciliation",
+ assessment,
+ )
+ schema_version = envelope.get("reconciliation_schema")
+ if type(schema_version) is not int or schema_version != RECONCILIATION_SCHEMA_VERSION:
+ assessment.add(
+ "FLINK_EXISTING_EVIDENCE_INVALID", "reconciliation_schema must be 1"
+ )
+ if envelope.get("baseline_id") != manifest.get("baseline_id"):
+ assessment.add(
+ "FLINK_VERSION_UNVERIFIED", "The reconciliation baseline does not match"
+ )
+
+ desired = envelope.get("desired_contract")
+ if not isinstance(desired, dict):
+ assessment.add(
+ "FLINK_EXISTING_EVIDENCE_INVALID", "desired_contract must be an object"
+ )
+ desired = {}
+ desired_result = assess_design(
+ desired,
+ manifest,
+ expected_table_lifecycle="existing",
+ render_sql=False,
+ )
+ for finding in desired_result["findings"]:
+ assessment.add(finding["code"], finding["message"])
+ for advisory in desired_result["advisory_codes"]:
+ assessment.advise(advisory)
+
+ observations, evidence_summary = _parse_evidence(
+ envelope.get("evidence"), assessment
+ )
+ for kind in sorted(REQUIRED_EVIDENCE_KINDS - set(observations)):
+ assessment.add(
+ "FLINK_EXISTING_EVIDENCE_REQUIRED", f"{kind} evidence is required"
+ )
+ table_observations = observations.get("hoodie_properties", {})
+ for fact in sorted(REQUIRED_TABLE_FACTS - set(table_observations)):
+ assessment.add(
+ "FLINK_EXISTING_EVIDENCE_REQUIRED",
+ f"hoodie_properties.{fact} is required",
+ )
+ _compare_ddl_with_authority(observations, assessment)
+
+ comparisons: list[dict[str, Any]] = []
+ desired_table = desired.get("table", {}) if isinstance(desired.get("table"), dict) else {}
+ desired_identity = (
+ desired.get("identity", {}) if isinstance(desired.get("identity"), dict) else {}
+ )
+ desired_ordering = (
+ desired.get("ordering", {}) if isinstance(desired.get("ordering"), dict) else {}
+ )
+ desired_write = desired.get("write", {}) if isinstance(desired.get("write"), dict) else {}
+ property_expectations = {
+ "base_file_format": manifest["existing_table_reconciliation_contract"][
+ "supported_base_file_formats"
+ ][0],
+ "ordering_fields": desired_ordering.get("fields", []),
+ "partition_fields": desired_table.get("partition_fields", []),
+ "record_key_fields": desired_identity.get("record_key_fields", []),
+ "table_name": desired_table.get("name"),
+ "table_type": desired_write.get("table_type"),
+ "initial_table_version": manifest["existing_table_reconciliation_contract"][
+ "supported_initial_table_versions"
+ ][0],
+ "table_version": manifest["existing_table_reconciliation_contract"][
+ "supported_table_versions"
+ ][0],
+ "timeline_layout_version": manifest["existing_table_reconciliation_contract"][
+ "supported_timeline_layout_versions"
+ ][0],
+ }
+ if desired.get("contract_schema") in (2, 3):
+ property_expectations["record_merge_mode"] = desired_write.get("record_merge_mode")
+ if desired_table.get("partition_fields"):
+ property_expectations.update(
+ {"hive_style_partitioning": False, "url_encode_partitioning": False}
+ )
+ _add_comparison(
+ comparisons,
+ "table_identity.table_path",
+ desired_table.get("path"),
+ observations.get("hudi_table_identity", {}).get("table_path"),
+ mismatch_code="FLINK_EXISTING_TABLE_PROPERTY_CONFLICT",
+ assessment=assessment,
+ )
+ _add_comparison(
+ comparisons,
+ "evidence_scope.table_path",
+ desired_table.get("path"),
+ observations.get("evidence_scope", {}).get("table_path"),
+ mismatch_code="FLINK_EXISTING_EVIDENCE_CONFLICT",
+ mismatch_result="UNVERIFIED",
+ assessment=assessment,
+ )
+ for name, expected in property_expectations.items():
+ mismatch_code = (
+ "FLINK_EXISTING_TABLE_VERSION_UNVERIFIED"
+ if name in {"initial_table_version", "table_version", "timeline_layout_version"}
+ else "FLINK_EXISTING_TABLE_PROPERTY_CONFLICT"
+ )
+ mismatch_result = (
+ "UNVERIFIED"
+ if name in {"initial_table_version", "table_version", "timeline_layout_version"}
+ else "CONFLICT"
+ )
+ _add_comparison(
+ comparisons,
+ f"table_config.{name}",
+ expected,
+ table_observations.get(name),
+ mismatch_code=mismatch_code,
+ mismatch_result=mismatch_result,
+ assessment=assessment,
+ )
+
+ expected_key_generators = _expected_key_generator(desired)
+ if "TIMESTAMP_AVRO" in expected_key_generators:
+ timestamp_expected = _expected_timestamp_settings(desired_table)
+ timestamp_observed = observations.get("flink_writer_config", {}).get(
+ "timestamp_key_generator"
+ )
+ if not timestamp_expected:
+ comparisons.append(
+ {
+ "desired": None,
+ "field": "writer_config.timestamp_key_generator",
+ "observed": timestamp_observed.value if timestamp_observed else None,
+ "result": "UNVERIFIED",
+ "source_ids": (
+ list(timestamp_observed.source_ids) if timestamp_observed else []
+ ),
+ }
+ )
+ assessment.add(
+ "FLINK_EXISTING_TABLE_PROPERTY_UNVERIFIED",
+ "Datetime partition precision is outside the pinned writer-setting mappings",
+ )
+ elif timestamp_observed is not None and not set(timestamp_expected) <= set(
+ timestamp_observed.value
+ ):
+ assessment.add(
+ "FLINK_EXISTING_EVIDENCE_REQUIRED",
+ "Timestamp key-generator evidence is missing required settings",
+ )
+ timestamp_observed = None
+ if timestamp_expected:
+ _add_comparison(
+ comparisons,
+ "writer_config.timestamp_key_generator",
+ timestamp_expected,
+ timestamp_observed,
+ mismatch_code="FLINK_EXISTING_WRITER_CONFIG_CONFLICT",
Review Comment:
[P2] Keep partition-layout differences out of deployment-only changes
Changing only the fixture's observed `timestamp_key_generator.output_date_format` from `yyyyMMdd` to `yyyy-MM-dd` produces `DEPLOYMENT_CONFIGURATION_CHANGE`. These are the prior writer's effective settings, so they describe how existing partition paths were written: the same date maps to `20261019` versus `2026-10-19`. Changing the next writer's configuration does not rewrite those existing paths, and can make writes for the same logical partition target a different directory. Could timestamp partition-format/timezone differences have a separate finding that retains layout/migration review (`UNVERIFIED` unless migration requirements are established), rather than sharing the finding that is reduced to deployment-only at the end? Please cover this case without asserting `DEPLOYMENT_CONFIGURATION_CHANGE`.
--
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(agent): add existing-table reconciliation to Hudi Architect [hudi]
Posted by "danny0405 (via GitHub)" <gi...@apache.org>.
danny0405 merged PR #20391:
URL: https://github.com/apache/hudi/pull/20391
--
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(agent): add existing-table reconciliation to Hudi Architect [hudi]
Posted by "fhan688 (via GitHub)" <gi...@apache.org>.
fhan688 commented on code in PR #20391:
URL: https://github.com/apache/hudi/pull/20391#discussion_r4334134600
##########
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)
+ if scoped_paths:
+ first_path = scoped_paths[0][0]
+ source_ids = tuple(item[1] for item in scoped_paths)
+ if any(item[0] != first_path for item in scoped_paths[1:]):
+ assessment.add(
+ "FLINK_EXISTING_EVIDENCE_CONFLICT",
+ "Evidence items do not identify one common table path",
+ )
+ else:
+ observations.setdefault("evidence_scope", {})["table_path"] = Observation(
+ first_path, source_ids
+ )
+ return observations, summaries
+
+
+def _add_comparison(
+ comparisons: list[dict[str, Any]],
+ field: str,
+ desired: Any,
+ observed: Observation | None,
+ *,
+ mismatch_code: str,
+ assessment: Assessment,
+ mismatch_result: str = "CONFLICT",
+) -> None:
+ if observed is None:
+ comparisons.append(
+ {
+ "desired": desired,
+ "field": field,
+ "observed": None,
+ "result": "UNVERIFIED",
+ "source_ids": [],
+ }
+ )
+ assessment.add(
+ "FLINK_EXISTING_EVIDENCE_REQUIRED", f"Evidence is required for {field}"
+ )
+ return
+ result = "MATCH" if desired == observed.value else mismatch_result
+ comparisons.append(
+ {
+ "desired": desired,
+ "field": field,
+ "observed": observed.value,
+ "result": result,
+ "source_ids": list(observed.source_ids),
+ }
+ )
+ if result != "MATCH":
+ assessment.add(mismatch_code, f"Existing-table value differs at {field}")
+
+
+def _expected_key_generator(desired: dict[str, Any]) -> set[str]:
+ identity = desired.get("identity")
+ table = desired.get("table")
+ identity = identity if isinstance(identity, dict) else {}
+ table = table if isinstance(table, dict) else {}
+ partition_fields = table.get("partition_fields")
+ partition_fields = partition_fields if isinstance(partition_fields, list) else []
+ record_keys = identity.get("record_key_fields", [])
+ record_keys = record_keys if isinstance(record_keys, list) else []
+ if not partition_fields:
+ return {"NON_PARTITION", "NON_PARTITION_AVRO"}
+ if len(record_keys) <= 1 and len(partition_fields) <= 1:
+ if _partition_field_type(table) in {"DATE", "TIMESTAMP"}:
+ return {"TIMESTAMP", "TIMESTAMP_AVRO"}
+ return {"SIMPLE", "SIMPLE_AVRO"}
+ return {"COMPLEX", "COMPLEX_AVRO"}
+
+
+def _partition_field_type(table: dict[str, Any]) -> str | None:
+ partitions = table.get("partition_fields")
+ columns = table.get("columns")
+ if not isinstance(partitions, list) or len(partitions) != 1 or not isinstance(columns, list):
+ return None
+ for column in columns:
+ if not isinstance(column, dict) or column.get("name") != partitions[0]:
+ continue
+ field_type = _normalize_type(column.get("type")) or ""
+ if field_type == "DATE":
+ return "DATE"
+ if field_type.startswith("TIMESTAMP"):
+ return "TIMESTAMP"
+ return None
+
+
+def _expected_timestamp_settings(table: dict[str, Any]) -> dict[str, str]:
+ partitions = table.get("partition_fields")
+ columns = table.get("columns")
+ if not isinstance(partitions, list) or len(partitions) != 1 or not isinstance(columns, list):
+ return {}
+ for column in columns:
+ if not isinstance(column, dict) or column.get("name") != partitions[0]:
+ continue
+ field_type = _normalize_type(column.get("type")) or ""
+ if field_type == "DATE":
+ return {
+ "timestamp_type": "SCALAR",
+ "input_time_unit": "DAYS",
+ "output_date_format": "yyyyMMdd",
+ "output_timezone": "UTC",
+ }
+ if field_type in {"TIMESTAMP(0)", "TIMESTAMP(3)"}:
+ return {
+ "timestamp_type": (
+ "UNIX_TIMESTAMP" if field_type.endswith("(0)") else "EPOCHMILLISECONDS"
+ ),
+ "output_date_format": "yyyyMMddHH",
+ "output_timezone": "UTC",
+ }
+ return {}
+
+
+def _compare_schema(
+ desired_columns: Any,
+ observed: Observation | None,
+ assessment: Assessment,
+ comparisons: list[dict[str, Any]],
+) -> None:
+ if observed is None:
+ _add_comparison(
+ comparisons,
+ "schema.columns",
+ desired_columns,
+ None,
+ mismatch_code="FLINK_EXISTING_SCHEMA_CONFLICT",
+ assessment=assessment,
+ )
+ return
+ desired = [
+ {
+ "name": column.get("name"),
+ "type": _normalize_type(column.get("type")),
+ "nullable": column.get("nullable"),
+ }
+ for column in (desired_columns if isinstance(desired_columns, list) else [])
+ if isinstance(column, dict) and _non_empty_string(column.get("name"))
+ ]
+ actual = observed.value
+ result = "MATCH"
+ mismatch_code: str | None = None
+ if desired != actual:
+ desired_by_name = {
+ column["name"]: column for column in desired if isinstance(column["name"], str)
+ }
+ actual_by_name = {column["name"]: column for column in actual}
+ missing_required = sorted(
+ name
+ for name in set(desired_by_name) - set(actual_by_name)
+ if desired_by_name[name]["nullable"] is not True
+ )
+ missing_nullable = sorted(
+ name
+ for name in set(desired_by_name) - set(actual_by_name)
+ if desired_by_name[name]["nullable"] is True
+ )
+ incompatible_types = sorted(
+ name
+ for name in set(desired_by_name) & set(actual_by_name)
+ if desired_by_name[name]["type"] != actual_by_name[name]["type"]
+ )
+ nullable_conflicts = sorted(
+ name
+ for name in set(desired_by_name) & set(actual_by_name)
+ if desired_by_name[name]["nullable"] is True
+ and actual_by_name[name]["nullable"] is False
+ )
+ nullable_evolution = sorted(
+ name
+ for name in set(desired_by_name) & set(actual_by_name)
+ if desired_by_name[name]["nullable"] is False
+ and actual_by_name[name]["nullable"] is True
+ )
+ extra = sorted(set(actual_by_name) - set(desired_by_name))
+ if missing_required or incompatible_types or nullable_conflicts:
+ result = "CONFLICT"
+ mismatch_code = "FLINK_EXISTING_SCHEMA_CONFLICT"
+ if missing_nullable or nullable_evolution or extra:
+ assessment.add(
+ "FLINK_EXISTING_SCHEMA_EVOLUTION_UNVERIFIED",
+ "Schema additions, nullability changes or extra physical columns require review",
+ )
+ if result != "CONFLICT":
+ result = "UNVERIFIED"
+ mismatch_code = "FLINK_EXISTING_SCHEMA_EVOLUTION_UNVERIFIED"
+ comparisons.append(
+ {
+ "desired": desired,
+ "field": "schema.columns",
+ "observed": actual,
+ "result": result,
+ "source_ids": list(observed.source_ids),
+ }
+ )
+ if mismatch_code:
+ assessment.add(mismatch_code, "Existing physical schema differs from the desired schema")
+
+
+def _compare_ddl_with_authority(
+ observations: dict[str, dict[str, Observation]], assessment: Assessment
+) -> None:
+ ddl = observations.get("flink_table_ddl", {})
+ table = {
+ **observations.get("hoodie_properties", {}),
+ **observations.get("hudi_table_identity", {}),
+ }
+ schema = observations.get("hudi_table_schema", {})
+ for name in sorted(set(ddl) & set(table)):
+ if ddl[name].value != table[name].value:
+ assessment.add(
+ "FLINK_EXISTING_EVIDENCE_CONFLICT",
+ f"Flink DDL disagrees with authoritative table evidence for {name}",
+ )
+ if "columns" in ddl and "columns" in schema and ddl["columns"].value != schema["columns"].value:
+ assessment.add(
+ "FLINK_EXISTING_EVIDENCE_CONFLICT",
+ "Flink DDL disagrees with the authoritative Hudi table schema",
+ )
+
+
+def _expected_writer_state(desired: dict[str, Any]) -> dict[str, Any]:
+ """Return the bounded writer state that must survive a stateful restore."""
+
+ raw_write = desired.get("write")
+ write = raw_write if isinstance(raw_write, dict) else {}
+ expected: dict[str, Any] = {
+ "design_contract_schema": desired.get("contract_schema"),
+ "operation": str(write.get("operation", "")).upper(),
+ }
+ if desired.get("contract_schema") in (2, 3):
+ raw_index = write.get("index")
+ index = raw_index if isinstance(raw_index, dict) else {}
+ expected.update(
+ {
+ "index_bootstrap_enabled": index.get("bootstrap_enabled"),
+ "index_global": index.get("global"),
+ "index_state_ttl_days": index.get("state_ttl_days"),
+ "index_type": str(index.get("type", "")).upper(),
+ }
+ )
+ if desired.get("contract_schema") == 3:
+ raw_services = desired.get("table_services")
+ services = raw_services if isinstance(raw_services, dict) else {}
+ raw_compaction = services.get("compaction")
+ compaction = raw_compaction if isinstance(raw_compaction, dict) else {}
+ expected["compaction_owner"] = str(compaction.get("owner", "")).upper()
+ return expected
+
+
+def assess_reconciliation(
+ envelope: dict[str, Any], manifest: dict[str, Any]
+) -> dict[str, Any]:
+ """Return one deterministic, non-executable existing-table assessment."""
+
+ assessment = Assessment()
+ manifest_errors = validate_manifest(manifest)
+ if manifest_errors:
+ assessment.add("FLINK_BASELINE_EVIDENCE_INVALID", "; ".join(manifest_errors))
+ return {
+ "status": assessment.final_status(),
+ "reconciliation_result": "UNVERIFIED",
+ "reconciliation_complete": False,
+ "executable_eligible": False,
+ "finding_codes": [finding["code"] for finding in assessment.findings],
+ "findings": assessment.findings,
+ "comparisons": [],
+ "evidence_summary": [],
+ }
+
+ _check_keys(
+ envelope,
+ {
+ "baseline_id",
+ "desired_contract",
+ "evidence",
+ "reconciliation_schema",
+ "writer_transition",
+ },
+ "reconciliation",
+ assessment,
+ )
+ schema_version = envelope.get("reconciliation_schema")
+ if type(schema_version) is not int or schema_version != RECONCILIATION_SCHEMA_VERSION:
+ assessment.add(
+ "FLINK_EXISTING_EVIDENCE_INVALID", "reconciliation_schema must be 1"
+ )
+ if envelope.get("baseline_id") != manifest.get("baseline_id"):
+ assessment.add(
+ "FLINK_VERSION_UNVERIFIED", "The reconciliation baseline does not match"
+ )
+
+ desired = envelope.get("desired_contract")
+ if not isinstance(desired, dict):
+ assessment.add(
+ "FLINK_EXISTING_EVIDENCE_INVALID", "desired_contract must be an object"
+ )
+ desired = {}
+ desired_result = assess_design(
+ desired,
+ manifest,
+ expected_table_lifecycle="existing",
+ render_sql=False,
+ )
+ for finding in desired_result["findings"]:
+ assessment.add(finding["code"], finding["message"])
+ for advisory in desired_result["advisory_codes"]:
+ assessment.advise(advisory)
+
+ observations, evidence_summary = _parse_evidence(
+ envelope.get("evidence"), assessment
+ )
+ for kind in sorted(REQUIRED_EVIDENCE_KINDS - set(observations)):
+ assessment.add(
+ "FLINK_EXISTING_EVIDENCE_REQUIRED", f"{kind} evidence is required"
+ )
+ table_observations = observations.get("hoodie_properties", {})
+ for fact in sorted(REQUIRED_TABLE_FACTS - set(table_observations)):
+ assessment.add(
+ "FLINK_EXISTING_EVIDENCE_REQUIRED",
+ f"hoodie_properties.{fact} is required",
+ )
+ _compare_ddl_with_authority(observations, assessment)
+
+ comparisons: list[dict[str, Any]] = []
+ desired_table = desired.get("table", {}) if isinstance(desired.get("table"), dict) else {}
+ desired_identity = (
+ desired.get("identity", {}) if isinstance(desired.get("identity"), dict) else {}
+ )
+ desired_ordering = (
+ desired.get("ordering", {}) if isinstance(desired.get("ordering"), dict) else {}
+ )
+ desired_write = desired.get("write", {}) if isinstance(desired.get("write"), dict) else {}
+ property_expectations = {
+ "base_file_format": manifest["existing_table_reconciliation_contract"][
+ "supported_base_file_formats"
+ ][0],
+ "ordering_fields": desired_ordering.get("fields", []),
+ "partition_fields": desired_table.get("partition_fields", []),
+ "record_key_fields": desired_identity.get("record_key_fields", []),
+ "table_name": desired_table.get("name"),
+ "table_type": desired_write.get("table_type"),
+ "initial_table_version": manifest["existing_table_reconciliation_contract"][
+ "supported_initial_table_versions"
+ ][0],
+ "table_version": manifest["existing_table_reconciliation_contract"][
+ "supported_table_versions"
+ ][0],
+ "timeline_layout_version": manifest["existing_table_reconciliation_contract"][
+ "supported_timeline_layout_versions"
+ ][0],
+ }
+ if desired.get("contract_schema") in (2, 3):
+ property_expectations["record_merge_mode"] = desired_write.get("record_merge_mode")
+ if desired_table.get("partition_fields"):
+ property_expectations.update(
+ {"hive_style_partitioning": False, "url_encode_partitioning": False}
+ )
+ _add_comparison(
+ comparisons,
+ "table_identity.table_path",
+ desired_table.get("path"),
+ observations.get("hudi_table_identity", {}).get("table_path"),
+ mismatch_code="FLINK_EXISTING_TABLE_PROPERTY_CONFLICT",
+ assessment=assessment,
+ )
+ _add_comparison(
+ comparisons,
+ "evidence_scope.table_path",
+ desired_table.get("path"),
+ observations.get("evidence_scope", {}).get("table_path"),
+ mismatch_code="FLINK_EXISTING_EVIDENCE_CONFLICT",
+ mismatch_result="UNVERIFIED",
+ assessment=assessment,
+ )
+ for name, expected in property_expectations.items():
+ mismatch_code = (
+ "FLINK_EXISTING_TABLE_VERSION_UNVERIFIED"
+ if name in {"initial_table_version", "table_version", "timeline_layout_version"}
+ else "FLINK_EXISTING_TABLE_PROPERTY_CONFLICT"
+ )
+ mismatch_result = (
+ "UNVERIFIED"
+ if name in {"initial_table_version", "table_version", "timeline_layout_version"}
+ else "CONFLICT"
+ )
+ _add_comparison(
+ comparisons,
+ f"table_config.{name}",
+ expected,
+ table_observations.get(name),
+ mismatch_code=mismatch_code,
+ mismatch_result=mismatch_result,
+ assessment=assessment,
+ )
+
+ expected_key_generators = _expected_key_generator(desired)
+ if "TIMESTAMP_AVRO" in expected_key_generators:
+ timestamp_expected = _expected_timestamp_settings(desired_table)
+ timestamp_observed = observations.get("flink_writer_config", {}).get(
+ "timestamp_key_generator"
+ )
+ if not timestamp_expected:
+ comparisons.append(
+ {
+ "desired": None,
+ "field": "writer_config.timestamp_key_generator",
+ "observed": timestamp_observed.value if timestamp_observed else None,
+ "result": "UNVERIFIED",
+ "source_ids": (
+ list(timestamp_observed.source_ids) if timestamp_observed else []
+ ),
+ }
+ )
+ assessment.add(
+ "FLINK_EXISTING_TABLE_PROPERTY_UNVERIFIED",
+ "Datetime partition precision is outside the pinned writer-setting mappings",
+ )
+ elif timestamp_observed is not None and not set(timestamp_expected) <= set(
+ timestamp_observed.value
+ ):
+ assessment.add(
+ "FLINK_EXISTING_EVIDENCE_REQUIRED",
+ "Timestamp key-generator evidence is missing required settings",
+ )
+ timestamp_observed = None
+ if timestamp_expected:
+ _add_comparison(
+ comparisons,
+ "writer_config.timestamp_key_generator",
+ timestamp_expected,
+ timestamp_observed,
+ mismatch_code="FLINK_EXISTING_WRITER_CONFIG_CONFLICT",
Review Comment:
Thanks for catching this. I separated timestamp partition-layout differences from writer-configuration changes. Differences in timestamp type, input unit, output format, or timezone now remain `UNVERIFIED` and no longer produce `DEPLOYMENT_CONFIGURATION_CHANGE`. I also added regression coverage for these cases.
--
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(agent): add existing-table reconciliation to Hudi Architect [hudi]
Posted by "fhan688 (via GitHub)" <gi...@apache.org>.
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org