You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@hudi.apache.org by da...@apache.org on 2026/10/28 02:12:47 UTC
(hudi) branch master updated: feat(agent): add Flink catalog and metastore composition to Hudi Architect (#20419)
This is an automated email from the ASF dual-hosted git repository.
danny0405 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/hudi.git
The following commit(s) were added to refs/heads/master by this push:
new 82d724b93b55 feat(agent): add Flink catalog and metastore composition to Hudi Architect (#20419)
82d724b93b55 is described below
commit 82d724b93b552b822a968ba4cbc7fdf3360c4ccb
Author: fhan <aa...@gmail.com>
AuthorDate: Wed Oct 28 10:12:40 2026 +0800
feat(agent): add Flink catalog and metastore composition to Hudi Architect (#20419)
* feat(agent): add Flink catalog and metastore composition to Hudi Architect
* feat(agent): Harden Flink catalog composition validation and fixtures
---------
Co-authored-by: fhan <yf...@jd.com>
---
.../fixtures/hudi-architect-flink-1.20/pom.xml | 14 +
.../TestFlinkArchitectCatalogFixtures.java | 217 ++++++
.../hudi_architect/flink_pr6/catalog_profiles.toml | 52 ++
.../tests/test_hudi_architect_skill.py | 744 +++++++++++++++++-
hudi-ai-operator/skills/hudi-architect/README.md | 12 +-
hudi-ai-operator/skills/hudi-architect/RUNBOOK.md | 27 +-
hudi-ai-operator/skills/hudi-architect/SKILL.md | 30 +-
.../flink-1.20-hudi-1.2.0-capabilities.md | 44 +-
.../flink-1.20-hudi-1.2.0-capabilities.toml | 238 +++++-
.../references/flink-catalog-composition.md | 153 ++++
.../references/flink-config-templates.md | 34 +-
.../references/flink-decision-overrides.md | 48 +-
.../flink-existing-table-reconciliation.md | 18 +-
.../references/flink-question-flow.md | 39 +-
.../hudi-architect/references/flink-warnings.md | 99 ++-
.../hudi-architect/validate_flink_capabilities.py | 171 ++++-
.../skills/hudi-architect/validate_flink_design.py | 851 +++++++++++++++++++--
.../validate_flink_existing_table.py | 15 +-
18 files changed, 2697 insertions(+), 109 deletions(-)
diff --git a/hudi-agent-gateway/fixtures/hudi-architect-flink-1.20/pom.xml b/hudi-agent-gateway/fixtures/hudi-architect-flink-1.20/pom.xml
index 919f08edf04c..0396de718039 100644
--- a/hudi-agent-gateway/fixtures/hudi-architect-flink-1.20/pom.xml
+++ b/hudi-agent-gateway/fixtures/hudi-architect-flink-1.20/pom.xml
@@ -34,6 +34,7 @@
<hudi.version>1.2.0</hudi.version>
<flink.version>1.20.1</flink.version>
<hadoop.version>2.10.2</hadoop.version>
+ <hive.version>2.3.10</hive.version>
<junit.version>5.11.4</junit.version>
</properties>
@@ -66,6 +67,19 @@
<version>${hadoop.version}</version>
<scope>test</scope>
</dependency>
+ <dependency>
+ <groupId>org.apache.hadoop</groupId>
+ <artifactId>hadoop-mapreduce-client-core</artifactId>
+ <version>${hadoop.version}</version>
+ <scope>test</scope>
+ </dependency>
+ <!-- HoodieHiveCatalog exposes HiveConf in its released public constructor. -->
+ <dependency>
+ <groupId>org.apache.hive</groupId>
+ <artifactId>hive-exec</artifactId>
+ <version>${hive.version}</version>
+ <scope>test</scope>
+ </dependency>
<dependency>
<groupId>org.junit.jupiter</groupId>
<artifactId>junit-jupiter</artifactId>
diff --git a/hudi-agent-gateway/fixtures/hudi-architect-flink-1.20/src/test/java/org/apache/hudi/agent/architect/TestFlinkArchitectCatalogFixtures.java b/hudi-agent-gateway/fixtures/hudi-architect-flink-1.20/src/test/java/org/apache/hudi/agent/architect/TestFlinkArchitectCatalogFixtures.java
new file mode 100644
index 000000000000..77b5f7ab72f4
--- /dev/null
+++ b/hudi-agent-gateway/fixtures/hudi-architect-flink-1.20/src/test/java/org/apache/hudi/agent/architect/TestFlinkArchitectCatalogFixtures.java
@@ -0,0 +1,217 @@
+/*
+ * 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.
+ */
+
+package org.apache.hudi.agent.architect;
+
+import org.apache.hudi.configuration.FlinkOptions;
+import org.apache.hudi.exception.HoodieCatalogException;
+import org.apache.hudi.hive.HiveSyncConfigHolder;
+import org.apache.hudi.sink.utils.HiveSyncContext;
+import org.apache.hudi.sync.common.HoodieSyncConfig;
+import org.apache.hudi.table.catalog.CatalogOptions;
+import org.apache.hudi.table.catalog.HoodieCatalog;
+import org.apache.hudi.table.catalog.HoodieCatalogFactory;
+import org.apache.hudi.table.catalog.HoodieHiveCatalog;
+
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.configuration.ReadableConfig;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.table.api.DataTypes;
+import org.apache.flink.table.api.EnvironmentSettings;
+import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
+import org.apache.flink.table.catalog.Catalog;
+import org.apache.flink.table.catalog.ObjectPath;
+import org.apache.flink.table.factories.CatalogFactory;
+import org.apache.flink.types.Row;
+import org.apache.hadoop.hive.conf.HiveConf;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+import java.lang.reflect.Field;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Properties;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertInstanceOf;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/** Released-bundle evidence for the bounded PR6 catalog composition contract. */
+class TestFlinkArchitectCatalogFixtures {
+
+ @Test
+ void testPinnedCatalogFactorySelectsDfsAndHms() {
+ HoodieCatalogFactory factory = new HoodieCatalogFactory();
+
+ Catalog dfs = factory.createCatalog(context("lake", Map.of(
+ "mode", "dfs",
+ "catalog.path", "file:///tmp/lake")));
+ Catalog hms = factory.createCatalog(context("hms", Map.of(
+ "mode", "hms",
+ "hadoop.hive.metastore.uris", "thrift://hms.example:9083",
+ "table.external", "true")));
+
+ assertInstanceOf(HoodieCatalog.class, dfs);
+ assertInstanceOf(HoodieHiveCatalog.class, hms);
+ }
+
+ @Test
+ void testPinnedCatalogFactoryRejectsUnknownMode() {
+ HoodieCatalogFactory factory = new HoodieCatalogFactory();
+
+ HoodieCatalogException exception = assertThrows(
+ HoodieCatalogException.class,
+ () -> factory.createCatalog(context("unsupported", Map.of("mode", "unknown"))));
+
+ assertTrue(exception.getMessage().contains("supported modes: [hms, dfs]"));
+ }
+
+ @Test
+ void testPinnedDfsCatalogPathDerivation() {
+ Configuration options = new Configuration();
+ options.set(CatalogOptions.CATALOG_PATH, "file:///tmp/lake");
+ ExposedHoodieCatalog catalog = new ExposedHoodieCatalog("lake", options);
+
+ assertEquals(
+ "file:///tmp/lake/analytics/orders",
+ catalog.infer("file:///tmp/lake", new ObjectPath("analytics", "orders")));
+ }
+
+ @Test
+ void testPinnedHmsCatalogExternalFlag() throws Exception {
+ Configuration options = new Configuration();
+ options.set(CatalogOptions.TABLE_EXTERNAL, true);
+ HoodieHiveCatalog catalog = new HoodieHiveCatalog(
+ "hms", options, new HiveConf(), true);
+ Field external = HoodieHiveCatalog.class.getDeclaredField("external");
+ external.setAccessible(true);
+
+ assertTrue(external.getBoolean(catalog));
+ }
+
+ @Test
+ void testPinnedHmsCatalogRejectsEmbeddedMetastore() {
+ assertThrows(
+ IllegalArgumentException.class,
+ () -> new HoodieHiveCatalog("hms", new Configuration()));
+ }
+
+ @Test
+ void testPinnedDfsCatalogDdlAndInsertPlan(@TempDir Path tempDir) {
+ StreamExecutionEnvironment environment =
+ StreamExecutionEnvironment.getExecutionEnvironment();
+ StreamTableEnvironment tableEnv = StreamTableEnvironment.create(
+ environment, EnvironmentSettings.newInstance().inStreamingMode().build());
+ tableEnv.createTemporaryView(
+ "orders_source",
+ tableEnv.fromValues(
+ DataTypes.ROW(
+ DataTypes.FIELD("id", DataTypes.STRING().notNull()),
+ DataTypes.FIELD("payload", DataTypes.STRING())),
+ Row.of("order-1", "created")));
+ String catalogPath = tempDir.toUri().toString();
+ tableEnv.executeSql(
+ "CREATE CATALOG `lake` WITH ("
+ + "'type'='hudi','mode'='dfs','catalog.path'='"
+ + catalogPath
+ + "','default-database'='analytics')");
+ tableEnv.executeSql("USE CATALOG `lake`");
+ tableEnv.executeSql("CREATE DATABASE IF NOT EXISTS `analytics`");
+ tableEnv.executeSql("USE `analytics`");
+ tableEnv.executeSql(
+ "CREATE TABLE `orders` ("
+ + "`id` STRING NOT NULL, `payload` STRING, "
+ + "PRIMARY KEY (`id`) NOT ENFORCED) WITH ("
+ + "'connector'='hudi','table.type'='COPY_ON_WRITE',"
+ + "'write.operation'='insert','write.insert.cluster'='false')");
+
+ String plan = tableEnv.explainSql(
+ "INSERT INTO `orders` (`id`, `payload`) "
+ + "SELECT `id`, `payload` "
+ + "FROM `default_catalog`.`default_database`.`orders_source`");
+
+ assertTrue(plan.contains("lake.analytics.orders"), plan);
+ assertTrue(Files.exists(tempDir.resolve("analytics/orders/.hoodie")));
+ }
+
+ @Test
+ void testPinnedMorHiveSyncNamingDefaults() {
+ Configuration configuration = new Configuration();
+ configuration.set(FlinkOptions.PATH, "file:///tmp/orders");
+ configuration.set(FlinkOptions.TABLE_TYPE, "MERGE_ON_READ");
+ configuration.set(FlinkOptions.HIVE_SYNC_DB, "analytics");
+ configuration.set(FlinkOptions.HIVE_SYNC_TABLE, "orders");
+ Properties properties = HiveSyncContext.buildSyncConfig(configuration);
+
+ assertEquals(
+ "false",
+ properties.getProperty(HiveSyncConfigHolder.HIVE_IGNORE_EXCEPTIONS.key()));
+ assertEquals(
+ "false",
+ properties.getProperty(
+ HiveSyncConfigHolder.HIVE_SKIP_RO_SUFFIX_FOR_READ_OPTIMIZED_TABLE.key()));
+ assertEquals(
+ "true",
+ properties.getProperty(HiveSyncConfigHolder.HIVE_SUPPORT_TIMESTAMP_TYPE.key()));
+ assertEquals(
+ "ALL",
+ properties.getProperty(HiveSyncConfigHolder.HIVE_SYNC_TABLE_STRATEGY.key()));
+ assertEquals(
+ "true",
+ HoodieSyncConfig.META_SYNC_SNAPSHOT_WITH_TABLE_NAME.defaultValue());
+ }
+
+ private static CatalogFactory.Context context(String name, Map<String, String> values) {
+ Map<String, String> options = new HashMap<>(values);
+ Configuration configuration = Configuration.fromMap(options);
+ return new CatalogFactory.Context() {
+ @Override
+ public String getName() {
+ return name;
+ }
+
+ @Override
+ public Map<String, String> getOptions() {
+ return options;
+ }
+
+ @Override
+ public ReadableConfig getConfiguration() {
+ return configuration;
+ }
+
+ @Override
+ public ClassLoader getClassLoader() {
+ return TestFlinkArchitectCatalogFixtures.class.getClassLoader();
+ }
+ };
+ }
+
+ private static final class ExposedHoodieCatalog extends HoodieCatalog {
+ private ExposedHoodieCatalog(String name, Configuration options) {
+ super(name, options);
+ }
+
+ private String infer(String catalogPath, ObjectPath tablePath) {
+ return inferTablePath(catalogPath, tablePath);
+ }
+ }
+}
diff --git a/hudi-agent-gateway/tests/fixtures/hudi_architect/flink_pr6/catalog_profiles.toml b/hudi-agent-gateway/tests/fixtures/hudi_architect/flink_pr6/catalog_profiles.toml
new file mode 100644
index 000000000000..6fd7ba703b88
--- /dev/null
+++ b/hudi-agent-gateway/tests/fixtures/hudi_architect/flink_pr6/catalog_profiles.toml
@@ -0,0 +1,52 @@
+# 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.
+
+[[profile]]
+id = "direct_sql_none"
+base_fixture = "flink_pr2/stable_key.json"
+catalog = '''
+{"contract_schema":1,"resolution_mode":"direct_sql","external_discovery":"none","sync_owner":"none","data_lifecycle_owner":"application"}
+'''
+
+[[profile]]
+id = "direct_sql_hive_metastore"
+base_fixture = "flink_pr2/stable_key.json"
+catalog = '''
+{"contract_schema":1,"resolution_mode":"direct_sql","external_discovery":"hive_metastore","sync_owner":"flink_sink","data_lifecycle_owner":"application","database":"analytics","hive_configuration":{"source":"metastore_uris","location":"thrift://hms.example:9083"}}
+'''
+
+[[profile]]
+id = "hudi_dfs_catalog_none"
+base_fixture = "flink_pr3/mutable_cow.json"
+source_table = "default_catalog.default_database.orders_mutable_source"
+catalog = '''
+{"contract_schema":1,"resolution_mode":"hudi_dfs_catalog","external_discovery":"none","sync_owner":"none","data_lifecycle_owner":"catalog","catalog_name":"lake","database":"analytics","catalog_path":"file:///tmp/hudi-architect-catalog"}
+'''
+
+[[profile]]
+id = "hudi_dfs_catalog_hive_metastore"
+base_fixture = "flink_pr3/mutable_cow.json"
+source_table = "default_catalog.default_database.orders_mutable_source"
+catalog = '''
+{"contract_schema":1,"resolution_mode":"hudi_dfs_catalog","external_discovery":"hive_metastore","sync_owner":"flink_sink","data_lifecycle_owner":"catalog","catalog_name":"lake","database":"analytics","catalog_path":"file:///tmp/hudi-architect-catalog","hive_configuration":{"source":"metastore_uris","location":"thrift://hms.example:9083"}}
+'''
+
+[[profile]]
+id = "hudi_hms_catalog_hive_metastore"
+base_fixture = "flink_pr3/mutable_cow.json"
+source_table = "default_catalog.default_database.orders_mutable_source"
+catalog = '''
+{"contract_schema":1,"resolution_mode":"hudi_hms_catalog","external_discovery":"hive_metastore","sync_owner":"hudi_hms_catalog","data_lifecycle_owner":"application","catalog_name":"production_hudi","database":"analytics","hive_configuration":{"source":"hive_conf_dir","location":"/etc/hive/conf"},"table_external":true}
+'''
diff --git a/hudi-agent-gateway/tests/test_hudi_architect_skill.py b/hudi-agent-gateway/tests/test_hudi_architect_skill.py
index 4765386c3528..90dd4ff53b0e 100644
--- a/hudi-agent-gateway/tests/test_hudi_architect_skill.py
+++ b/hudi-agent-gateway/tests/test_hudi_architect_skill.py
@@ -21,6 +21,7 @@ from __future__ import annotations
import ast
import copy
import json
+import re
import subprocess
import sys
import tomllib
@@ -44,6 +45,7 @@ PR2_FIXTURE_DIR = GATEWAY_DIR / "tests" / "fixtures" / "hudi_architect" / "flink
PR3_FIXTURE_DIR = GATEWAY_DIR / "tests" / "fixtures" / "hudi_architect" / "flink_pr3"
PR4_FIXTURE_DIR = GATEWAY_DIR / "tests" / "fixtures" / "hudi_architect" / "flink_pr4"
PR5_FIXTURE_DIR = GATEWAY_DIR / "tests" / "fixtures" / "hudi_architect" / "flink_pr5"
+PR6_FIXTURE_DIR = GATEWAY_DIR / "tests" / "fixtures" / "hudi_architect" / "flink_pr6"
ASF_LICENSE_MARKER = "Licensed to the Apache Software Foundation (ASF) under one"
PR2_JAVA_FIXTURE = (
GATEWAY_DIR
@@ -59,6 +61,7 @@ PR2_JAVA_FIXTURE = (
/ "architect"
/ "TestFlinkArchitectSqlFixtures.java"
)
+PR6_JAVA_FIXTURE = PR2_JAVA_FIXTURE.with_name("TestFlinkArchitectCatalogFixtures.java")
FLINK_REFERENCES = (
"flink-1.20-hudi-1.2.0-capabilities.md",
@@ -68,6 +71,7 @@ FLINK_REFERENCES = (
"flink-warnings.md",
"flink-config-templates.md",
"flink-existing-table-reconciliation.md",
+ "flink-catalog-composition.md",
)
@@ -151,6 +155,77 @@ def _write_design(tmp_path: Path, design: dict) -> Path:
return input_path
+def _pr6_profile(profile_id: str) -> dict:
+ profiles = _load_toml(PR6_FIXTURE_DIR / "catalog_profiles.toml")["profile"]
+ profile = next(item for item in profiles if item["id"] == profile_id)
+ base_path = GATEWAY_DIR / "tests" / "fixtures" / "hudi_architect" / profile[
+ "base_fixture"
+ ]
+ design = json.loads(_read(base_path))
+ catalog = json.loads(profile["catalog"])
+ design["catalog"] = catalog
+ design["safety"]["external_catalog"] = profile_id != "direct_sql_none"
+ if "source_table" in profile:
+ design["source"]["table"] = profile["source_table"]
+ if catalog["resolution_mode"] == "hudi_dfs_catalog":
+ design["table"]["path"] = (
+ f"{catalog['catalog_path']}/{catalog['database']}/{design['table']['name']}"
+ )
+ return design
+
+
+def _pr6_mor_profile(
+ resolution_mode: str, query_modes: list[str], *, hive_conf_source: str
+) -> dict:
+ design = json.loads(_read(PR4_FIXTURE_DIR / "mutable_mor.json"))
+ design["safety"]["external_catalog"] = True
+ design["read"]["query_modes"] = query_modes
+ design["read"]["read_optimized_freshness"] = (
+ "after_compaction" if "read_optimized" in query_modes else "NOT_APPLICABLE"
+ )
+ strategy = (
+ "ALL"
+ if set(query_modes) == {"snapshot", "read_optimized"}
+ else "RT" if query_modes == ["snapshot"] else "RO"
+ )
+ if resolution_mode == "direct_sql":
+ design["catalog"] = {
+ "contract_schema": 1,
+ "resolution_mode": "direct_sql",
+ "external_discovery": "hive_metastore",
+ "sync_owner": "flink_sink",
+ "data_lifecycle_owner": "application",
+ "database": "analytics",
+ "hive_configuration": {
+ "source": "metastore_uris",
+ "location": "thrift://hms.example:9083",
+ },
+ "mor_sync_strategy": strategy,
+ }
+ else:
+ hive_configuration = (
+ {"source": "classpath"}
+ if hive_conf_source == "classpath"
+ else {"source": "hive_conf_dir", "location": "/etc/hive/conf"}
+ )
+ design["catalog"] = {
+ "contract_schema": 1,
+ "resolution_mode": "hudi_hms_catalog",
+ "external_discovery": "hive_metastore",
+ "sync_owner": "hudi_hms_catalog",
+ "data_lifecycle_owner": "application",
+ "catalog_name": "production_hudi",
+ "database": "analytics",
+ "hive_configuration": hive_configuration,
+ "table_external": True,
+ "mor_sync_strategy": strategy,
+ }
+ design["source"]["table"] = (
+ f"default_catalog.default_database.{design['source']['table']}"
+ )
+ return design
+
+
def _compatible_existing_envelope(design: dict) -> dict:
desired = copy.deepcopy(design)
desired["safety"]["table_lifecycle"] = "existing"
@@ -305,7 +380,7 @@ def test_flink_baseline_has_bounded_append_and_mutable_contracts() -> None:
"fixture_version": "1.20.1",
}
assert PINNED_HUDI_REVISION in capability
- assert manifest["schema_version"] == 5
+ assert manifest["schema_version"] == 6
assert manifest["fixture_artifacts"] == {
"hudi_bundle": "org.apache.hudi:hudi-flink1.20-bundle:1.2.0",
"flink_version": "1.20.1",
@@ -365,6 +440,17 @@ def test_flink_baseline_has_bounded_append_and_mutable_contracts() -> None:
assert reconciliation["finding_status"][
"FLINK_EXISTING_TIMESTAMP_PARTITION_LAYOUT_UNVERIFIED"
] == "REVIEW_REQUIRED"
+ catalog_contract = manifest["catalog_composition_contract"]
+ assert catalog_contract["contract_schema"] == 1
+ assert catalog_contract["supported_design_contract_schemas"] == [1, 2, 3]
+ assert catalog_contract["catalog_freshness"] == "post_commit_eventual"
+ assert [profile["id"] for profile in manifest["catalog_profiles"]] == [
+ "direct_sql_none",
+ "direct_sql_hive_metastore",
+ "hudi_dfs_catalog_none",
+ "hudi_dfs_catalog_hive_metastore",
+ "hudi_hms_catalog_hive_metastore",
+ ]
assert "first executable" in capability
assert "`CONFIG_VALIDATED`" in decisions
assert "Executable eligible: true" in output
@@ -385,10 +471,19 @@ def test_flink_capability_validation_emits_pinned_evidence() -> None:
assert result.returncode == 0, result.stderr
assert json.loads(result.stdout) == {
"baseline_id": "hudi-1.2.0-flink-1.20",
- "capability_manifest_schema": 5,
+ "capability_manifest_schema": 6,
+ "catalog_contract_schema": 1,
+ "catalog_profile": "direct_sql_none",
"design_contract_schema": 1,
"flink_fixture_version": "1.20.1",
"hudi_source_revision": PINNED_HUDI_REVISION,
+ "supported_catalog_profiles": [
+ "direct_sql_none",
+ "direct_sql_hive_metastore",
+ "hudi_dfs_catalog_none",
+ "hudi_dfs_catalog_hive_metastore",
+ "hudi_hms_catalog_hive_metastore",
+ ],
"supported_design_contract_schemas": [1, 2, 3],
}
@@ -480,7 +575,7 @@ def test_flink_capability_manifest_records_executable_acceptance_contracts() ->
for node in ast.walk(ast.parse(_read(Path(__file__))))
if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef))
}
- java_fixture = _read(PR2_JAVA_FIXTURE)
+ java_fixture = _read(PR2_JAVA_FIXTURE) + _read(PR6_JAVA_FIXTURE)
effective_settings = {
setting["applies_to"]: {
candidate["key"]: candidate["value"]
@@ -545,6 +640,18 @@ def test_flink_capability_manifest_records_executable_acceptance_contracts() ->
"test_pr2_rejects_binary_routing_fields",
"testPinnedBinaryValuesUseObjectIdentityForRouting",
),
+ "FLINK_CATALOG_LIFECYCLE_UNRESOLVED": (
+ "test_pr6_hms_catalog_requires_external_table",
+ "testPinnedHmsCatalogExternalFlag",
+ ),
+ "FLINK_CATALOG_MODE_UNSUPPORTED": (
+ "test_pr6_rejects_unsupported_catalog_compositions",
+ "testPinnedCatalogFactoryRejectsUnknownMode",
+ ),
+ "FLINK_CATALOG_PATH_CONFLICT": (
+ "test_pr6_rejects_dfs_catalog_path_mismatch",
+ "testPinnedDfsCatalogPathDerivation",
+ ),
"FLINK_RECORD_KEY_FIELD_MISSING": (
"test_pr2_rejects_record_key_missing_from_append_schema",
"testPinnedFactorySkipsMissingRecordKeyCheckInAppendMode",
@@ -605,6 +712,10 @@ def test_flink_capability_manifest_records_executable_acceptance_contracts() ->
"test_pr4_rejects_strict_read_optimized_freshness",
"testPinnedMorQuerySemantics",
),
+ "FLINK_MOR_CATALOG_NAMING_UNVERIFIED": (
+ "test_pr6_rejects_mor_catalog_naming_drift",
+ "testPinnedMorHiveSyncNamingDefaults",
+ ),
"FLINK_PR4_WRITE_PATH_UNSUPPORTED": (
"test_pr4_valid_mutable_mor_design_renders_exact_executable_sql",
"testMutableMorSinkAndCompactionPlan",
@@ -720,10 +831,19 @@ def test_pr2_valid_designs_render_exact_executable_sql(fixture_name: str) -> Non
).rstrip("\n")
assert assessment["validation_evidence"] == {
"baseline_id": "hudi-1.2.0-flink-1.20",
- "capability_manifest_schema": 5,
+ "capability_manifest_schema": 6,
+ "catalog_contract_schema": 1,
+ "catalog_profile": "direct_sql_none",
"design_contract_schema": 1,
"flink_fixture_version": "1.20.1",
"hudi_source_revision": PINNED_HUDI_REVISION,
+ "supported_catalog_profiles": [
+ "direct_sql_none",
+ "direct_sql_hive_metastore",
+ "hudi_dfs_catalog_none",
+ "hudi_dfs_catalog_hive_metastore",
+ "hudi_hms_catalog_hive_metastore",
+ ],
"supported_design_contract_schemas": [1, 2, 3],
}
@@ -741,7 +861,7 @@ def test_pr3_valid_mutable_design_renders_exact_executable_sql() -> None:
PR3_FIXTURE_DIR / "mutable_cow.sql"
).rstrip("\n")
assert assessment["validation_evidence"]["design_contract_schema"] == 2
- assert assessment["validation_evidence"]["capability_manifest_schema"] == 5
+ assert assessment["validation_evidence"]["capability_manifest_schema"] == 6
def test_pr4_valid_mutable_mor_design_renders_exact_executable_sql() -> None:
@@ -758,14 +878,624 @@ def test_pr4_valid_mutable_mor_design_renders_exact_executable_sql() -> None:
).rstrip("\n")
assert assessment["validation_evidence"] == {
"baseline_id": "hudi-1.2.0-flink-1.20",
- "capability_manifest_schema": 5,
+ "capability_manifest_schema": 6,
+ "catalog_contract_schema": 1,
+ "catalog_profile": "direct_sql_none",
"design_contract_schema": 3,
"flink_fixture_version": "1.20.1",
"hudi_source_revision": PINNED_HUDI_REVISION,
+ "supported_catalog_profiles": [
+ "direct_sql_none",
+ "direct_sql_hive_metastore",
+ "hudi_dfs_catalog_none",
+ "hudi_dfs_catalog_hive_metastore",
+ "hudi_hms_catalog_hive_metastore",
+ ],
"supported_design_contract_schemas": [1, 2, 3],
}
+@pytest.mark.parametrize(
+ "profile_id",
+ [
+ "direct_sql_none",
+ "direct_sql_hive_metastore",
+ "hudi_dfs_catalog_none",
+ "hudi_dfs_catalog_hive_metastore",
+ "hudi_hms_catalog_hive_metastore",
+ ],
+)
+def test_pr6_supported_catalog_profiles_render_deterministically(
+ tmp_path: Path, profile_id: str
+) -> None:
+ design = _pr6_profile(profile_id)
+ result = _run_flink_design_validator(_write_design(tmp_path, design))
+
+ assert result.returncode == 0, result.stderr
+ assessment = json.loads(result.stdout)
+ assert assessment["status"] == "CONFIG_VALIDATED"
+ assert assessment["catalog_profile"] == profile_id
+ assert assessment["catalog_contract_schema"] == 1
+ assert assessment["validation_evidence"]["catalog_profile"] == profile_id
+ expected_checks = {
+ "direct_sql_none": [],
+ "direct_sql_hive_metastore": [
+ "verify_hive_configuration_available",
+ "verify_hms_connectivity",
+ "verify_table_registration_after_commit",
+ "verify_partition_registration_after_commit",
+ "verify_external_engine_visibility",
+ ],
+ "hudi_dfs_catalog_none": [
+ "verify_catalog_dependencies",
+ "verify_catalog_path_exists",
+ "verify_catalog_path_permissions",
+ "verify_catalog_database_registration",
+ ],
+ "hudi_dfs_catalog_hive_metastore": [
+ "verify_catalog_dependencies",
+ "verify_catalog_path_exists",
+ "verify_catalog_path_permissions",
+ "verify_hive_configuration_available",
+ "verify_hms_connectivity",
+ "verify_catalog_database_registration",
+ "verify_table_registration_after_commit",
+ "verify_partition_registration_after_commit",
+ "verify_external_engine_visibility",
+ ],
+ "hudi_hms_catalog_hive_metastore": [
+ "verify_catalog_dependencies",
+ "verify_hive_configuration_available",
+ "verify_hms_connectivity",
+ "verify_catalog_database_registration",
+ "verify_table_registration_after_commit",
+ "verify_partition_registration_after_commit",
+ "verify_external_engine_visibility",
+ ],
+ }
+ assert assessment["catalog_deployment_checks"] == expected_checks[profile_id]
+ sql = assessment["artifacts"]["combined_sql"]
+ table_ddl = assessment["artifacts"]["table_ddl"]
+ assert sql.index("SET 'execution.runtime-mode'") < sql.index("CREATE TABLE")
+ if profile_id.startswith("hudi_dfs_catalog"):
+ assert "CREATE CATALOG `lake`" in sql
+ assert "'mode' = 'dfs'" in sql
+ assert "'path' =" not in table_ddl
+ elif profile_id == "hudi_hms_catalog_hive_metastore":
+ assert "CREATE CATALOG `production_hudi`" in sql
+ assert "'mode' = 'hms'" in sql
+ assert "'table.external' = 'true'" in sql
+ assert design["table"]["path"] in table_ddl
+ assert "'hive_sync." not in table_ddl
+ else:
+ assert assessment["catalog_setup_sql"] == ""
+ if profile_id == "direct_sql_none":
+ assert "'hive_sync.enabled'" not in table_ddl
+ assert sql == _read_sql_golden(PR2_FIXTURE_DIR / "stable_key.sql").rstrip(
+ "\n"
+ )
+ else:
+ assert "'hive_sync.enabled' = 'true'" in table_ddl
+ if assessment["catalog_setup_sql"]:
+ ordered_markers = [
+ "SET 'execution.runtime-mode'",
+ "CREATE CATALOG",
+ "USE CATALOG",
+ "CREATE DATABASE",
+ "USE `analytics`",
+ "CREATE TABLE",
+ "INSERT INTO",
+ ]
+ positions = [sql.index(marker) for marker in ordered_markers]
+ assert positions == sorted(positions)
+ assert (
+ "FROM `default_catalog`.`default_database`.`orders_mutable_source`;"
+ in assessment["artifacts"]["insert_sql"]
+ )
+ if design["catalog"]["external_discovery"] == "hive_metastore":
+ assert assessment["effective_catalog_configuration"][
+ "catalog_freshness"
+ ] == "post_commit_eventual"
+ effective_sync = assessment["effective_catalog_configuration"]["hive_sync"]
+ assert effective_sync["enabled"] is True
+ assert effective_sync["ignore_exceptions"] is False
+ assert effective_sync["skip_ro_suffix"] is False
+ assert effective_sync["snapshot_with_table_name"] is True
+ assert effective_sync["support_timestamp"] is True
+ assert "verify_hms_connectivity" in assessment["catalog_deployment_checks"]
+ if assessment["effective_catalog_configuration"]["sync_owner"] == "flink_sink":
+ rendered_sync_keys = re.findall(r"'hive_sync\.[^']+'", table_ddl)
+ assert rendered_sync_keys == [
+ "'hive_sync.enabled'",
+ "'hive_sync.mode'",
+ "'hive_sync.metastore.uris'",
+ "'hive_sync.db'",
+ "'hive_sync.table'",
+ ]
+
+
+@pytest.mark.parametrize(
+ "profile_id",
+ [
+ "hudi_dfs_catalog_none",
+ "hudi_dfs_catalog_hive_metastore",
+ "hudi_hms_catalog_hive_metastore",
+ ],
+)
+def test_pr6_catalog_managed_profiles_require_fully_qualified_source(
+ tmp_path: Path, profile_id: str
+) -> None:
+ design = _pr6_profile(profile_id)
+ design["source"]["table"] = "orders_mutable_source"
+
+ result = _run_flink_design_validator(_write_design(tmp_path, design))
+ assessment = json.loads(result.stdout)
+
+ assert result.returncode == 1
+ assert assessment["status"] == "INCOMPLETE"
+ assert "FLINK_SOURCE_CONTRACT_REQUIRED" in assessment["finding_codes"]
+ assert "artifacts" not in assessment
+
+
+@pytest.mark.parametrize(
+ "profile_id",
+ [
+ "hudi_dfs_catalog_none",
+ "hudi_dfs_catalog_hive_metastore",
+ "hudi_hms_catalog_hive_metastore",
+ ],
+)
+def test_pr6_catalog_managed_profiles_reject_same_source_and_target(
+ tmp_path: Path, profile_id: str
+) -> None:
+ design = _pr6_profile(profile_id)
+ catalog = design["catalog"]
+ design["source"]["table"] = (
+ f"{catalog['catalog_name']}.{catalog['database']}.{design['table']['name']}"
+ )
+
+ result = _run_flink_design_validator(_write_design(tmp_path, design))
+ assessment = json.loads(result.stdout)
+
+ assert result.returncode == 1
+ assert assessment["status"] == "BLOCKED"
+ assert "FLINK_CATALOG_COMPOSITION_CONFLICT" in assessment["finding_codes"]
+ assert "artifacts" not in assessment
+
+
+@pytest.mark.parametrize(
+ ("partition_fields", "expected_status"),
+ [
+ (["partition_date", "region"], "CONFIG_VALIDATED"),
+ (["region", "partition_date"], "BLOCKED"),
+ ],
+)
+def test_pr6_hms_catalog_validates_physical_partition_order(
+ tmp_path: Path, partition_fields: list[str], expected_status: str
+) -> None:
+ design = _pr6_profile("hudi_hms_catalog_hive_metastore")
+ region = {"name": "region", "type": "STRING", "nullable": False}
+ design["table"]["columns"].append(region)
+ design["source"]["columns"].append(region)
+ design["table"]["partition_fields"] = partition_fields
+
+ result = _run_flink_design_validator(_write_design(tmp_path, design))
+ assessment = json.loads(result.stdout)
+
+ assert assessment["status"] == expected_status
+ if expected_status == "CONFIG_VALIDATED":
+ assert result.returncode == 0
+ assert "FLINK_CATALOG_COMPOSITION_CONFLICT" not in assessment["finding_codes"]
+ assert (
+ "PARTITIONED BY (`partition_date`, `region`)"
+ in assessment["artifacts"]["table_ddl"]
+ )
+ else:
+ assert result.returncode == 1
+ assert "FLINK_CATALOG_COMPOSITION_CONFLICT" in assessment["finding_codes"]
+ assert "artifacts" not in assessment
+
+
+@pytest.mark.parametrize(
+ ("query_modes", "strategy", "registered_names"),
+ [
+ (["snapshot"], "RT", ["orders_mutable_mor_hudi"]),
+ (["read_optimized"], "RO", ["orders_mutable_mor_hudi"]),
+ ],
+)
+def test_pr6_direct_mor_hive_sync_derives_strategy_and_names(
+ tmp_path: Path,
+ query_modes: list[str],
+ strategy: str,
+ registered_names: list[str],
+) -> None:
+ design = _pr6_mor_profile(
+ "direct_sql", query_modes, hive_conf_source="metastore_uris"
+ )
+ assessment = json.loads(
+ _run_flink_design_validator(_write_design(tmp_path, design)).stdout
+ )
+
+ assert assessment["status"] == "CONFIG_VALIDATED"
+ assert f"'hive_sync.table.strategy' = '{strategy}'" in assessment["artifacts"][
+ "table_ddl"
+ ]
+ assert assessment["effective_catalog_configuration"]["hive_sync"][
+ "registered_table_names"
+ ] == registered_names
+
+
+def test_pr6_hms_mor_all_uses_fixed_registration_names(tmp_path: Path) -> None:
+ design = _pr6_mor_profile(
+ "hudi_hms_catalog",
+ ["snapshot", "read_optimized"],
+ hive_conf_source="classpath",
+ )
+ result = _run_flink_design_validator(_write_design(tmp_path, design))
+
+ assert result.returncode == 0, result.stderr
+ assessment = json.loads(result.stdout)
+ assert assessment["catalog_profile"] == "hudi_hms_catalog_hive_metastore"
+ assert "'hive_sync.table.strategy' = 'ALL'" in assessment["artifacts"][
+ "table_ddl"
+ ]
+ assert "'hive_sync.enabled'" not in assessment["artifacts"]["table_ddl"]
+ assert assessment["effective_catalog_configuration"]["hive_sync"][
+ "registered_table_names"
+ ] == [
+ "orders_mutable_mor_hudi_ro",
+ "orders_mutable_mor_hudi_rt",
+ "orders_mutable_mor_hudi",
+ ]
+
+
+def test_pr6_rejects_dfs_catalog_path_mismatch(tmp_path: Path) -> None:
+ design = _pr6_profile("hudi_dfs_catalog_none")
+ design["table"]["path"] = "file:///tmp/a-different-table"
+
+ assessment = json.loads(
+ _run_flink_design_validator(_write_design(tmp_path, design)).stdout
+ )
+
+ assert assessment["status"] == "BLOCKED"
+ assert "FLINK_CATALOG_PATH_CONFLICT" in assessment["finding_codes"]
+ assert "artifacts" not in assessment
+
+
+@pytest.mark.parametrize(
+ ("catalog_patch", "finding_code"),
+ [
+ (
+ {
+ "resolution_mode": "hudi_hms_catalog",
+ "external_discovery": "none",
+ },
+ "FLINK_CATALOG_COMPOSITION_CONFLICT",
+ ),
+ ({"resolution_mode": "iceberg_catalog"}, "FLINK_CATALOG_MODE_UNSUPPORTED"),
+ (
+ {"external_discovery": "aws_glue"},
+ "FLINK_GLUE_CATALOG_UNVERIFIED",
+ ),
+ ],
+)
+def test_pr6_rejects_unsupported_catalog_compositions(
+ tmp_path: Path, catalog_patch: dict, finding_code: str
+) -> None:
+ design = _pr6_profile("hudi_hms_catalog_hive_metastore")
+ design["catalog"].update(catalog_patch)
+
+ assessment = json.loads(
+ _run_flink_design_validator(_write_design(tmp_path, design)).stdout
+ )
+
+ assert finding_code in assessment["finding_codes"]
+ assert assessment["executable_eligible"] is False
+
+
+def test_pr6_hms_catalog_requires_external_table(tmp_path: Path) -> None:
+ design = _pr6_profile("hudi_hms_catalog_hive_metastore")
+ design["catalog"]["table_external"] = False
+
+ assessment = json.loads(
+ _run_flink_design_validator(_write_design(tmp_path, design)).stdout
+ )
+
+ assert assessment["status"] == "BLOCKED"
+ assert "FLINK_CATALOG_COMPOSITION_CONFLICT" in assessment["finding_codes"]
+
+
+def test_pr6_hms_catalog_requires_external_table_decision(tmp_path: Path) -> None:
+ design = _pr6_profile("hudi_hms_catalog_hive_metastore")
+ design["catalog"].pop("table_external")
+
+ assessment = json.loads(
+ _run_flink_design_validator(_write_design(tmp_path, design)).stdout
+ )
+
+ assert assessment["status"] == "REVIEW_REQUIRED"
+ assert assessment["finding_codes"] == ["FLINK_CATALOG_LIFECYCLE_UNRESOLVED"]
+
+
+def test_pr6_hms_catalog_rejects_malformed_external_table_decision(
+ tmp_path: Path,
+) -> None:
+ design = _pr6_profile("hudi_hms_catalog_hive_metastore")
+ design["catalog"]["table_external"] = []
+
+ assessment = json.loads(
+ _run_flink_design_validator(_write_design(tmp_path, design)).stdout
+ )
+
+ assert assessment["status"] == "INCOMPLETE"
+ assert assessment["finding_codes"] == ["FLINK_CATALOG_CONTRACT_INVALID"]
+
+
+def test_pr6_requires_explicit_dfs_lifecycle_acceptance(tmp_path: Path) -> None:
+ design = _pr6_profile("hudi_dfs_catalog_none")
+ design["catalog"].pop("data_lifecycle_owner")
+
+ assessment = json.loads(
+ _run_flink_design_validator(_write_design(tmp_path, design)).stdout
+ )
+
+ assert assessment["status"] == "REVIEW_REQUIRED"
+ assert "FLINK_CATALOG_CONTRACT_INVALID" in assessment["finding_codes"]
+ assert "FLINK_CATALOG_LIFECYCLE_UNRESOLVED" in assessment["finding_codes"]
+ assert "artifacts" not in assessment
+
+
+def test_pr6_blocks_declined_dfs_lifecycle_ownership(tmp_path: Path) -> None:
+ design = _pr6_profile("hudi_dfs_catalog_none")
+ design["catalog"]["data_lifecycle_owner"] = "application"
+
+ assessment = json.loads(
+ _run_flink_design_validator(_write_design(tmp_path, design)).stdout
+ )
+
+ assert assessment["status"] == "BLOCKED"
+ assert assessment["finding_codes"] == ["FLINK_CATALOG_COMPOSITION_CONFLICT"]
+ assert "artifacts" not in assessment
+
+
+def test_pr6_rejects_mor_catalog_naming_drift(tmp_path: Path) -> None:
+ design = _pr6_mor_profile(
+ "direct_sql", ["snapshot", "read_optimized"], hive_conf_source="metastore_uris"
+ )
+ design["catalog"]["mor_sync_strategy"] = "RO"
+
+ assessment = json.loads(
+ _run_flink_design_validator(_write_design(tmp_path, design)).stdout
+ )
+
+ assert assessment["status"] == "REVIEW_REQUIRED"
+ assert "FLINK_MOR_CATALOG_NAMING_UNVERIFIED" in assessment["finding_codes"]
+
+
+def test_pr6_requires_catalog_contract_for_legacy_true_gate(tmp_path: Path) -> None:
+ design = json.loads(_read(PR2_FIXTURE_DIR / "stable_key.json"))
+ design["safety"]["external_catalog"] = True
+
+ assessment = json.loads(
+ _run_flink_design_validator(_write_design(tmp_path, design)).stdout
+ )
+
+ assert assessment["status"] == "INCOMPLETE"
+ assert assessment["finding_codes"] == ["FLINK_CATALOG_CONTRACT_REQUIRED"]
+
+
+def test_pr6_blocks_catalog_gate_and_owner_conflicts(tmp_path: Path) -> None:
+ design = _pr6_profile("direct_sql_hive_metastore")
+ design["safety"]["external_catalog"] = False
+ design["catalog"]["sync_owner"] = "hudi_hms_catalog"
+
+ assessment = json.loads(
+ _run_flink_design_validator(_write_design(tmp_path, design)).stdout
+ )
+
+ assert assessment["status"] == "BLOCKED"
+ assert assessment["finding_codes"] == ["FLINK_CATALOG_COMPOSITION_CONFLICT"]
+
+
+@pytest.mark.parametrize(
+ "hive_configuration",
+ [
+ {
+ "source": "metastore_uris",
+ "location": "thrift://user:secret@hms.example:9083",
+ },
+ {
+ "source": "metastore_uris",
+ "location": "thrift://hms.example:9083?token=secret",
+ },
+ {
+ "source": "metastore_uris",
+ "location": "thrift://hms.example:9083/password=secret",
+ },
+ {
+ "source": "metastore_uris",
+ "location": "thrift://hms.example:9083;password=secret",
+ },
+ {
+ "source": "metastore_uris",
+ "location": "thrift://hms.example:9083/path",
+ },
+ {
+ "source": "metastore_uris",
+ "location": "thrift://hms.example:badport",
+ },
+ {
+ "source": "metastore_uris",
+ "location": "thrift://hms.example",
+ },
+ {
+ "source": "metastore_uris",
+ "location": "thrift://hms.example:70000",
+ },
+ {"source": "metastore_uris", "location": "http://hms.example:9083"},
+ ],
+)
+def test_pr6_rejects_credentialed_or_non_thrift_metastore_uris(
+ tmp_path: Path, hive_configuration: dict
+) -> None:
+ design = _pr6_profile("direct_sql_hive_metastore")
+ design["catalog"]["hive_configuration"] = hive_configuration
+
+ assessment = json.loads(
+ _run_flink_design_validator(_write_design(tmp_path, design)).stdout
+ )
+
+ assert assessment["executable_eligible"] is False
+ assert "FLINK_CATALOG_CONTRACT_INVALID" in assessment["finding_codes"]
+ assert hive_configuration["location"] not in json.dumps(assessment)
+ if "secret" in str(hive_configuration):
+ assert "FLINK_CATALOG_CREDENTIAL_REJECTED" in assessment["finding_codes"]
+
+
+def test_pr6_accepts_multiple_well_formed_metastore_uris(tmp_path: Path) -> None:
+ design = _pr6_profile("direct_sql_hive_metastore")
+ design["catalog"]["hive_configuration"]["location"] = (
+ "thrift://hms-a.example:9083,thrift://[::1]:9084"
+ )
+
+ result = _run_flink_design_validator(_write_design(tmp_path, design))
+ assessment = json.loads(result.stdout)
+
+ assert result.returncode == 0, result.stderr
+ assert assessment["status"] == "CONFIG_VALIDATED"
+ assert (
+ "'hive_sync.metastore.uris' = "
+ "'thrift://hms-a.example:9083,thrift://[::1]:9084'"
+ in assessment["artifacts"]["table_ddl"]
+ )
+
+
+@pytest.mark.parametrize("catalog", [None, [], "hms"])
+def test_pr6_malformed_catalog_contract_fails_closed(
+ tmp_path: Path, catalog: object
+) -> None:
+ design = json.loads(_read(PR2_FIXTURE_DIR / "stable_key.json"))
+ design["safety"]["external_catalog"] = True
+ design["catalog"] = catalog
+
+ result = _run_flink_design_validator(_write_design(tmp_path, design))
+ assessment = json.loads(result.stdout)
+
+ assert result.returncode == 1
+ assert assessment["status"] == "INCOMPLETE"
+ assert "FLINK_CATALOG_CONTRACT_INVALID" in assessment["finding_codes"]
+ assert "FLINK_CATALOG_MODE_UNSUPPORTED" not in assessment["finding_codes"]
+ assert "artifacts" not in assessment
+
+
+@pytest.mark.parametrize(
+ ("field", "value"),
+ [
+ ("resolution_mode", []),
+ ("resolution_mode", {}),
+ ("external_discovery", []),
+ ("sync_owner", {}),
+ ("data_lifecycle_owner", []),
+ ],
+)
+def test_pr6_malformed_catalog_fields_fail_closed(
+ tmp_path: Path, field: str, value: object
+) -> None:
+ design = _pr6_profile("hudi_hms_catalog_hive_metastore")
+ design["catalog"][field] = value
+
+ result = _run_flink_design_validator(_write_design(tmp_path, design))
+ assessment = json.loads(result.stdout)
+
+ assert result.returncode == 1
+ assert "Traceback" not in result.stderr
+ assert "FLINK_CATALOG_CONTRACT_INVALID" in assessment["finding_codes"]
+ assert assessment["executable_eligible"] is False
+ assert "artifacts" not in assessment
+
+
+def test_pr6_existing_table_malformed_catalog_field_fails_closed(
+ tmp_path: Path,
+) -> None:
+ design = _pr6_profile("hudi_hms_catalog_hive_metastore")
+ design["catalog"]["resolution_mode"] = {}
+ envelope = _compatible_existing_envelope(design)
+ input_path = tmp_path / "existing-malformed-catalog.json"
+ input_path.write_text(json.dumps(envelope), encoding="utf-8")
+
+ result = _run_flink_existing_table_validator(input_path)
+ assessment = json.loads(result.stdout)
+
+ assert result.returncode == 1
+ assert "Traceback" not in result.stderr
+ assert "FLINK_CATALOG_CONTRACT_INVALID" in assessment["finding_codes"]
+ assert assessment["executable_eligible"] is False
+ assert "artifacts" not in assessment
+
+
+@pytest.mark.parametrize(
+ ("field", "value"),
+ [
+ ("resolution_mode", "iceberg_catalog"),
+ ("external_discovery", "custom_metastore"),
+ ("sync_owner", "another_job"),
+ ("data_lifecycle_owner", "unknown"),
+ ],
+)
+def test_pr6_unknown_catalog_modes_and_owners_are_blocked(
+ tmp_path: Path, field: str, value: str
+) -> None:
+ design = _pr6_profile("direct_sql_hive_metastore")
+ design["catalog"][field] = value
+
+ assessment = json.loads(
+ _run_flink_design_validator(_write_design(tmp_path, design)).stdout
+ )
+
+ assert assessment["status"] == "BLOCKED"
+ assert "FLINK_CATALOG_MODE_UNSUPPORTED" in assessment["finding_codes"]
+ assert "artifacts" not in assessment
+
+
+def test_pr6_malformed_mor_query_modes_fail_closed(tmp_path: Path) -> None:
+ design = _pr6_mor_profile(
+ "direct_sql", ["snapshot"], hive_conf_source="metastore_uris"
+ )
+ design["read"]["query_modes"] = [{}]
+
+ result = _run_flink_design_validator(_write_design(tmp_path, design))
+ assessment = json.loads(result.stdout)
+
+ assert result.returncode == 1
+ assert assessment["status"] == "REVIEW_REQUIRED"
+ assert "FLINK_MOR_CATALOG_NAMING_UNVERIFIED" in assessment["finding_codes"]
+ assert "artifacts" not in assessment
+
+
+def test_pr6_existing_catalog_mutation_remains_unverified(tmp_path: Path) -> None:
+ desired = _pr6_profile("hudi_hms_catalog_hive_metastore")
+ envelope = _compatible_existing_envelope(desired)
+ input_path = tmp_path / "existing.json"
+ input_path.write_text(json.dumps(envelope), encoding="utf-8")
+
+ result = _run_flink_existing_table_validator(input_path)
+ assessment = json.loads(result.stdout)
+
+ assert result.returncode == 1
+ assert assessment["status"] == "REVIEW_REQUIRED"
+ assert assessment["reconciliation_result"] == "UNVERIFIED"
+ assert assessment["change_disposition"] == "UNVERIFIED"
+ assert "FLINK_EXISTING_CATALOG_MUTATION_UNVERIFIED" in assessment["finding_codes"]
+ assert assessment["executable_eligible"] is False
+ assert assessment["catalog_contract_schema"] == 1
+ assert assessment["catalog_profile"] == "hudi_hms_catalog_hive_metastore"
+ assert assessment["catalog_setup_sql"] == ""
+ assert assessment["effective_catalog_configuration"]["catalog_freshness"] == (
+ "post_commit_eventual"
+ )
+ assert "verify_hms_connectivity" in assessment["catalog_deployment_checks"]
+
+
@pytest.mark.parametrize(
("field", "value"),
[
@@ -2400,7 +3130,7 @@ def test_safety_gate_order_is_stable() -> None:
"## F9 — Streaming checkpoint contract",
"## F10 — Mutable table type, query freshness, and MOR ownership",
"## F11 — Existing-table evidence and writer transition",
- "## PR2/PR3/PR4/PR5 validation and completion",
+ "## PR2/PR3/PR4/PR5/PR6 validation and completion",
]
offsets = [flow.index(heading) for heading in headings]
assert offsets == sorted(offsets)
diff --git a/hudi-ai-operator/skills/hudi-architect/README.md b/hudi-ai-operator/skills/hudi-architect/README.md
index edd5b6801a1c..fbd437b75272 100644
--- a/hudi-ai-operator/skills/hudi-architect/README.md
+++ b/hudi-ai-operator/skills/hudi-architect/README.md
@@ -41,7 +41,10 @@ Target Hudi version: **1.2.0**.
partitions, and the fixed global FLINK_STATE/bootstrap/no-TTL contract. MOR additionally fixes
same-writer asynchronous compaction, records snapshot/read-optimized freshness, and disables
clustering. It also provides a read-only, non-executable exact-match reconciliation report for
- an existing table backed by structured evidence. Other Flink designs remain fail-closed.
+ an existing table backed by structured evidence. For new tables, those same design schemas can
+ compose direct Hive sync, Hudi DFS Catalog, or external Hudi HMS Catalog through five pinned
+ profiles. Glue, managed HMS, existing-table catalog mutation, and live connectivity claims
+ remain fail-closed.
## How to invoke
@@ -88,6 +91,8 @@ Even without invoking as a Skill, the files in `references/` are readable design
- `flink-config-templates.md` — non-executable and bounded executable Flink output envelopes.
- `flink-existing-table-reconciliation.md` — evidence authority, comparison, and writer-transition
rules for the assessment-only existing-table path.
+- `flink-catalog-composition.md` — the five PR6 profiles, lifecycle/path rules, Hive sync ownership,
+ MOR naming, freshness boundary, and deployment checks.
The Skill itself is defined in `SKILL.md`.
@@ -129,6 +134,8 @@ The mutable schema-2 fixture is
`hudi-agent-gateway/tests/fixtures/hudi_architect/flink_pr3/mutable_cow.json`.
The mutable MOR schema-3 fixture is
`hudi-agent-gateway/tests/fixtures/hudi_architect/flink_pr4/mutable_mor.json`.
+PR6 profile inputs are recorded in
+`hudi-agent-gateway/tests/fixtures/hudi_architect/flink_pr6/catalog_profiles.toml`.
Validate a structured existing-table reconciliation contract without rendering SQL:
@@ -158,7 +165,8 @@ This is **Milestone 1 of a longer arc** — meant to be shareable and playable,
## What's out of scope in Milestone 1
- Flink append-only MOR, standalone compaction, detailed compaction tuning, clustering,
- existing-table SQL generation or takeover, external catalogs, multiple writers, alternative indexes, custom
+ existing-table SQL generation or catalog takeover, AWS Glue, managed/embedded HMS, live catalog
+ probing, multiple writers, alternative indexes, custom
mergers, retract normalization, partition movement, key-only deletes, full CDC history,
DataStream API, `HoodieFlinkStreamer`, source connector generation, and submit/deploy commands.
Those paths do not reuse the bounded SQL output as if it were generally applicable.
diff --git a/hudi-ai-operator/skills/hudi-architect/RUNBOOK.md b/hudi-ai-operator/skills/hudi-architect/RUNBOOK.md
index 9579026b6b9a..9647fc762956 100644
--- a/hudi-ai-operator/skills/hudi-architect/RUNBOOK.md
+++ b/hudi-ai-operator/skills/hudi-architect/RUNBOOK.md
@@ -26,9 +26,10 @@ A practical guide to running the Hudi Architect agent: what it does, how to set
bundle, and a runnable submit command — all pinned to **Hudi 1.2.0**. The Flink route is pinned to
**Hudi 1.2.0 / Flink 1.20** and remains fail-closed by default. It can emit validated Flink SQL for
bounded append-only COW, mutable COW, and mutable MOR new-table, single-writer,
-no-external-catalog paths. For an existing table, it can instead produce a read-only exact-match
-reconciliation report from provenance-bearing table, schema, writer, and timeline evidence; that
-report never contains SQL or mutation commands.
+streaming paths. New-table designs can optionally compose direct Hive sync, Hudi DFS Catalog, or
+external Hudi HMS Catalog through five pinned profiles. For an existing table, it can instead
+produce a read-only exact-match reconciliation report from provenance-bearing table, schema,
+writer, and timeline evidence; that report never contains SQL or mutation commands.
---
@@ -148,8 +149,10 @@ source changelog, fixed write mode, and a checkpoint interval of at least 1000 m
validation. Mutable input additionally requires a stable non-null key, event-time ordering,
normalized UPSERT changelog, full-row deletes when present, immutable partitions, and the fixed global
FLINK_STATE/bootstrap/no-TTL contract. Mutable MOR additionally requires same-writer asynchronous
-compaction, explicit snapshot/read-optimized freshness expectations, and disabled clustering. Its artifacts are runtime `SET`
-statements, Hudi `CREATE TABLE`, and an explicit-column `INSERT INTO`; deployment remains manual.
+compaction, explicit snapshot/read-optimized freshness expectations, and disabled clustering. Its
+artifacts are runtime `SET` statements, optional catalog setup SQL, Hudi `CREATE TABLE`, and an
+explicit-column `INSERT INTO`. The assessment retains effective catalog configuration and
+deployment-check codes separately; generated SQL is never executed and deployment remains manual.
An existing-table session emits comparisons, evidence provenance, findings, and deployment rechecks
with executable eligibility `false`, including when the static result is `COMPATIBLE`.
@@ -160,14 +163,16 @@ Then: land a first commit in a staging path, run your real read patterns against
Be aware of what the agent will *decline* to decide — it defers honestly rather than guessing, but plan for these yourself:
- **Flink executable coverage** — bounded append-only COW, mutable COW, and mutable MOR
- new-table, single-writer, no-external-catalog streaming SQL sinks are implemented. Append-only
- MOR, standalone compaction, detailed compaction tuning, clustering, existing-table SQL or takeover,
- catalogs, multiple writers, alternative indexes, custom mergers, retract normalization,
- partition movement, key-only deletes, full CDC history, DataStream API, `HoodieFlinkStreamer`,
- source connector generation, and submit/deploy commands remain out of scope.
+ new-table, single-writer streaming SQL sinks are implemented, with five direct SQL / DFS Catalog
+ / HMS Catalog compositions. Append-only MOR, standalone compaction, detailed compaction tuning,
+ clustering, existing-table SQL or takeover, existing-table catalog mutation, AWS Glue, managed
+ or embedded HMS, live catalog probing, multiple writers, alternative indexes, custom mergers,
+ retract normalization, partition movement, key-only deletes, full CDC history, DataStream API,
+ `HoodieFlinkStreamer`, source connector generation, and submit/deploy commands remain out of
+ scope.
- **Multi-writer contention tuning** — on the Spark route, the agent asks whether anything else writes the table (every tier but the lightest), derives the concurrency mode, picks a lock provider, and emits a complete, runnable block. What it does *not* do is tune for observed contention: lock retry/timeout values, conflict-retry counts, and early conflict detection stay at their defaults, because the right values depend on measured behavior rather than design-time facts. That is Operati [...]
-- **Catalog / metastore sync beyond the common paths** — on the Spark route, the agent asks which engines query the table and derives sync config for Hive Metastore, AWS Glue, BigQuery, and DataHub, including the partition-extractor and MOR `_ro`/`_rt` consequences. The current Flink path detects an external-catalog requirement but emits no catalog configuration. Not covered: Polaris beyond pointing at the Spark catalog config, Snowflake/Redshift-specific setup, and per-catalog auth (Ker [...]
+- **Catalog / metastore sync beyond the common paths** — on the Spark route, the agent asks which engines query the table and derives sync config for Hive Metastore, AWS Glue, BigQuery, and DataHub, including the partition-extractor and MOR `_ro`/`_rt` consequences. The Flink route emits only the five pinned direct SQL / DFS Catalog / external HMS Catalog profiles and treats catalog visibility as post-commit eventual. Not covered: Glue execution, live HMS checks, existing-table registrat [...]
- **Non-Kafka source configs** — Kafka sources get full source-class + schema-provider derivation; for DFS/JDBC/Pulsar/Kinesis sources you'll fill in the `--source-class` and source properties from Hudi docs.
- **Cross-format interoperability (Apache XTable)** — explained and linked, not configured. XTable translates Hudi metadata so the same files can be read as Iceberg or Delta, but it is a separate incubating project: its config keys live in its codebase rather than Hudi's, so this skill cannot machine-verify them the way it verifies every `hoodie.*` key it emits. The agent names the one durable coupling — **XTable's FAQ lists MOR as unsupported, so cross-format interop argues for CoW**, w [...]
- Also out of scope: benchmarking, record-level TTL, z-order/layout-optimization guidance, multi-table transactions, CONSISTENT_HASHING sizing, and versions other than 1.2.0.
diff --git a/hudi-ai-operator/skills/hudi-architect/SKILL.md b/hudi-ai-operator/skills/hudi-architect/SKILL.md
index b878daaeae9b..8d369d389a6d 100644
--- a/hudi-ai-operator/skills/hudi-architect/SKILL.md
+++ b/hudi-ai-operator/skills/hudi-architect/SKILL.md
@@ -37,7 +37,7 @@ Ask **workload questions, not Hudi questions.** Every decision falls into one of
**Never ask a question whose answer wouldn't change the recommendation.** If the workload is COW-only, don't ask about compaction. If the workload is unpartitioned, don't ask about cross-partition updates.
-**Gate optional sections with one plain question before opening them.** Where a whole group of questions exists to configure something the user may not want at all — catalog sync is the clearest case — ask that first, in outcome terms, and let a "no" close the section. Deriving the answer from three questions the user didn't need to answer is the same waste as one unnecessary question, multiplied. Two things make the gate honest: phrase it as an outcome rather than a Hudi feature ("does [...]
+**Gate optional sections with one plain question before opening them.** Where a whole group of questions exists to configure something the user may not want at all — catalog sync is the clearest case — ask that first, in outcome terms, and let a "no" close the section. Deriving the answer from three questions the user didn't need to answer is the same waste as one unnecessary question, multiplied. Two things make the gate honest: phrase it as an outcome rather than a Hudi feature ("does [...]
**Every question must be in user language, not Hudi language.** Users don't know what an RLI is; they know what a customer_id lookup is. Users don't know what a delta commit is; they know how often data lands.
@@ -67,6 +67,8 @@ Within the Flink route, load the remaining references only when needed:
- Load `references/flink-config-templates.md` only when producing the final safety assessment.
- Load `references/flink-existing-table-reconciliation.md` only after F1 confirms an existing
table; use it to collect provenance-bearing evidence and produce the assessment-only PR5 report.
+- Load `references/flink-catalog-composition.md` only when the user needs named catalog resolution
+ or external discovery; use its PR6 contract without changing design schema 1, 2, or 3.
The verified Flink baseline is Apache Hudi 1.2.0 at source revision
`f05c83f2b97732de7a558ff9b26959e1139c05f5` with Apache Flink 1.20; build and test fixtures use
@@ -77,18 +79,22 @@ that manifest with options discovered from the enclosing checkout. If the user e
another Hudi or Flink version, do not silently reuse the baseline: add a `REVIEW_REQUIRED` finding
unless a separate verified capability reference exists.
-The Flink implementation includes the PR1 safety foundation, three bounded executable paths, and
-one existing-table assessment path. PR2 append-only COW, PR3 mutable COW, and PR4 mutable MOR all
-require a new table, confirmed single writer, no external catalog, streaming checkpointing, and
-`CONFIG_VALIDATED` from `validate_flink_design.py`. Both mutable paths additionally require stable non-null identity, one
+The Flink implementation includes the PR1 safety foundation, three bounded executable paths, one
+existing-table assessment path, and PR6 catalog/metastore composition. PR2 append-only COW, PR3
+mutable COW, and PR4 mutable MOR all require a new table, confirmed single writer, streaming
+checkpointing, and `CONFIG_VALIDATED` from `validate_flink_design.py`. They retain direct SQL by
+default; PR6 may compose one of five pinned direct/DFS/HMS profiles without changing the table
+design schema. Both mutable paths additionally require stable non-null identity, one
event-time ordering field, normalized UPSERT changelog, full-row deletes when present, immutable
partitions, and fixed global FLINK_STATE/bootstrap/no-TTL settings. The MOR path further requires
snapshot/read-optimized expectations, compaction owned by the same Flink writer, fixed asynchronous
five-delta-commit compaction, and clustering disabled. PR5 can reconcile an explicit schema 1, 2,
or 3 desired contract with structured evidence from an existing table, but it never renders SQL or
-authorizes mutation. Append-only MOR, standalone table services, detailed compaction tuning,
-separate-writer or cross-engine takeover, catalog, multi-writer, alternative indexes, custom
-mergers, retract normalization, and full CDC history still fail closed.
+authorizes mutation. PR6 supports new-table direct Hive sync, Hudi DFS Catalog, and external Hudi
+HMS Catalog composition; AWS Glue, managed HMS, existing-table catalog mutation, and live service
+claims fail closed. Append-only MOR, standalone table services, detailed compaction tuning,
+separate-writer or cross-engine takeover, multi-writer, alternative indexes, custom mergers,
+retract normalization, and full CDC history still fail closed.
## Flow structure
@@ -98,7 +104,7 @@ The conversation has three parts:
2. **Rounds 1-3** — workload questions, gated conditionally by tier.
3. **Output** — engine-specific output: the established Spark artifacts, a non-executable Flink
assessment, a PR5 existing-table reconciliation report, or bounded validator-rendered
- PR2/PR3/PR4 SQL artifacts.
+ PR2/PR3/PR4 SQL artifacts, optionally composed with validated PR6 catalog setup SQL.
Load `references/question-flow.md` for the full round-by-round question list with conditional gating.
@@ -214,6 +220,10 @@ every stable identifier. If a finding remains, emit only the non-executable enve
safety gate passes, create the explicit JSON design contract described by `flink-question-flow.md`
and run `validate_flink_design.py`. Only its successful, canonical artifacts may be emitted as
runtime SQL, Hudi DDL, and the sink-side `INSERT INTO`; never modify them after validation.
+When a catalog block is present, retain the validator's `catalog_setup_sql`, effective catalog
+configuration, and deployment checks as separate machine fields; `--emit-sql` is only their
+canonical ordered composition. Catalog deployment checks never imply that a live service was
+probed, and the generated SQL is never executed by this skill.
For an existing table, create the same explicit desired contract, collect and confirm evidence per
`flink-existing-table-reconciliation.md`, and run `validate_flink_existing_table.py` instead. Emit
its report and deployment checks exactly as assessment data; never reinterpret `COMPATIBLE` as
@@ -272,6 +282,8 @@ fact `INCOMPLETE`; do not ask the user to reveal the secret.
- Emit executable Flink output while any safety or design-contract finding remains.
- Render SQL or mutation commands from an existing-table reconciliation result.
- Hand-write, repair, or extend SQL after `validate_flink_design.py` renders it.
+- Claim catalog/commit atomicity, execute generated catalog SQL, probe live HMS/Glue, or emit
+ existing-table catalog mutations.
- Treat a successful `HoodieTableFactory` construction as proof that the source changelog is
append-only.
diff --git a/hudi-ai-operator/skills/hudi-architect/references/flink-1.20-hudi-1.2.0-capabilities.md b/hudi-ai-operator/skills/hudi-architect/references/flink-1.20-hudi-1.2.0-capabilities.md
index 14d3fc5a6a16..26ac22defe8a 100644
--- a/hudi-ai-operator/skills/hudi-architect/references/flink-1.20-hudi-1.2.0-capabilities.md
+++ b/hudi-ai-operator/skills/hudi-architect/references/flink-1.20-hudi-1.2.0-capabilities.md
@@ -33,7 +33,8 @@ newer release, or an older supported connector as equivalent without its own ver
`flink-1.20-hudi-1.2.0-capabilities.toml` is the machine-readable validation input. It records the
full Hudi release commit, pinned Maven fixture artifacts, source paths and hashes, the option
-allowlist, the bounded design and reconciliation contracts, and acceptance evidence.
+allowlist, the bounded design, reconciliation, and catalog-composition contracts, and acceptance
+evidence.
Normal validation must read that checked-in manifest. It must not discover accepted Flink options
by scanning the enclosing checkout: the checkout may be `master` or another release and can contain
@@ -72,7 +73,13 @@ the current checkout contains it.
| Confirmed single-writer detection | Supported |
| Multi-writer or unknown writer model | `REVIEW_REQUIRED`; deferred to PR7 |
| External-catalog requirement detection | Supported |
-| Catalog or metastore composition | `REVIEW_REQUIRED`; deferred to PR6 |
+| Five direct SQL / DFS Catalog / HMS Catalog profiles | Supported for new tables through PR6 |
+| Direct SQL or DFS Catalog with sink-owned HMS sync | Supported |
+| HMS Catalog | Supported only for external, application-owned tables with HMS discovery |
+| DFS Catalog lifecycle | Catalog-owned; explicit acceptance required |
+| Hive catalog freshness | Post-commit eventual; no atomicity or SLA claim |
+| AWS Glue, embedded HMS, or managed HMS | `REVIEW_REQUIRED`; no executable output |
+| Existing-table catalog registration or mutation | `REVIEW_REQUIRED`; assessment-only |
| Physical-schema availability detection | Supported |
| Bounded scalar physical-schema validation | Supported |
| Binary payload columns | Supported |
@@ -112,6 +119,37 @@ the current checkout contains it.
| Flink 1.20.1 planner fixture | Supported with a local declared source contract |
| Spark and HoodieStreamer behavior | Must remain unchanged |
+## Implemented PR6 catalog composition contract
+
+PR6 adds catalog contract schema 1 alongside design schemas 1, 2, and 3. It supports exactly five
+profiles: direct SQL without discovery, direct SQL plus Hive Metastore discovery, Hudi DFS Catalog
+without discovery, Hudi DFS Catalog plus Hive Metastore discovery, and Hudi HMS Catalog plus Hive
+Metastore discovery. The legacy no-block, `external_catalog=false` path remains direct SQL and
+retains its existing SQL byte-for-byte.
+
+DFS Catalog requires `catalog.path`, derives `${catalog.path}/${database}/${table.name}`, and owns
+the data lifecycle; its generated table DDL omits `path`. HMS Catalog requires non-embedded Hive
+configuration, renders the desired table path, and is limited to `table.external=true` so the
+application owns data. Direct and DFS Hive discovery uses sink-owned `hive_sync.*` options; HMS
+Catalog supplies those values from catalog context and the validator checks their effective
+configuration without duplicating them.
+
+Sink-owned metastore URIs are limited to credential-free `thrift://host:port` endpoints with no
+path, query, or fragment. Catalog-managed streaming sources use a three-part identifier and cannot
+resolve to the same complete object as the new sink target. Missing lifecycle choices remain
+unresolved; explicit requests for DFS application ownership or managed HMS tables are conflicts.
+
+MOR registration derives `RT`, `RO`, or `ALL` from the declared query modes. With the pinned
+defaults (`skip_ro_suffix=false` and snapshot-with-base-name true), ALL registers `base_ro`,
+`base_rt`, and the base snapshot name. Catalog freshness is always recorded as
+`post_commit_eventual`: Hudi invokes Hive sync asynchronously after a successful checkpoint
+commit, so static validation does not claim atomic external visibility.
+
+The released-bundle Java fixture selects DFS and HMS catalog implementations, rejects unknown and
+embedded modes, checks DFS path derivation and external HMS posture, exercises DFS Catalog DDL and
+an insert plan, and verifies Hive sync/naming defaults. It does not start HMS or Glue. Live
+connectivity and visibility are deployment checks rather than static status findings.
+
## Implemented PR5 existing-table reconciliation contract
PR5 reuses the explicit schema-1, schema-2, or schema-3 desired design and changes only the table
@@ -227,7 +265,7 @@ the Python validator tests preserve the 999/1000 ms Architect boundary.
## Implemented PR3 mutable COW contract
The schema-2 path is deliberately narrower than the full Hudi upsert surface. It accepts only a
-new table, one writer, no external catalog, streaming checkpoints, stable non-null identity, one
+new table, one writer, streaming checkpoints, stable non-null identity, one
event-time ordering field, normalized UPSERT input, immutable partitions, and full-row deletes when
present. All three resolved replay answers are eligible; only an unresolved replay posture blocks
validation. The exact ordering-field value `no_precombine` is rejected because Hudi 1.2.0 resolves
diff --git a/hudi-ai-operator/skills/hudi-architect/references/flink-1.20-hudi-1.2.0-capabilities.toml b/hudi-ai-operator/skills/hudi-architect/references/flink-1.20-hudi-1.2.0-capabilities.toml
index 0c673b461478..2bbfa5913a98 100644
--- a/hudi-ai-operator/skills/hudi-architect/references/flink-1.20-hudi-1.2.0-capabilities.toml
+++ b/hudi-ai-operator/skills/hudi-architect/references/flink-1.20-hudi-1.2.0-capabilities.toml
@@ -13,9 +13,9 @@
# See the License for the specific language governing permissions and
# limitations under the License.
-schema_version = 5
+schema_version = 6
baseline_id = "hudi-1.2.0-flink-1.20"
-validation_surface = "flink-sql-sink-and-existing-table-reconciliation"
+validation_surface = "flink-sql-sink-existing-table-reconciliation-and-catalog-composition"
[hudi]
version = "1.2.0"
@@ -90,6 +90,83 @@ query_modes = ["snapshot", "read_optimized"]
read_optimized_freshness = "after_compaction"
clustering_enabled = false
+[catalog_composition_contract]
+contract_schema = 1
+supported_design_contract_schemas = [1, 2, 3]
+new_table_executable = true
+existing_table_assessment_only = true
+catalog_freshness = "post_commit_eventual"
+resolution_modes = ["direct_sql", "hudi_dfs_catalog", "hudi_hms_catalog"]
+external_discovery_modes = ["none", "hive_metastore", "aws_glue"]
+sync_owners = ["none", "flink_sink", "hudi_hms_catalog"]
+data_lifecycle_owners = ["application", "catalog"]
+deployment_checks = [
+ "verify_catalog_dependencies",
+ "verify_catalog_path_exists",
+ "verify_catalog_path_permissions",
+ "verify_hive_configuration_available",
+ "verify_hms_connectivity",
+ "verify_catalog_database_registration",
+ "verify_table_registration_after_commit",
+ "verify_partition_registration_after_commit",
+ "verify_mor_catalog_table_names",
+ "verify_external_engine_visibility",
+]
+
+[catalog_composition_contract.finding_status]
+FLINK_CATALOG_COMPOSITION_CONFLICT = "BLOCKED"
+FLINK_CATALOG_CONTRACT_INVALID = "INCOMPLETE"
+FLINK_CATALOG_CONTRACT_REQUIRED = "INCOMPLETE"
+FLINK_CATALOG_CREDENTIAL_REJECTED = "BLOCKED"
+FLINK_CATALOG_LIFECYCLE_UNRESOLVED = "REVIEW_REQUIRED"
+FLINK_CATALOG_MODE_UNSUPPORTED = "BLOCKED"
+FLINK_CATALOG_PATH_CONFLICT = "BLOCKED"
+FLINK_EXISTING_CATALOG_MUTATION_UNVERIFIED = "REVIEW_REQUIRED"
+FLINK_GLUE_CATALOG_UNVERIFIED = "REVIEW_REQUIRED"
+FLINK_MOR_CATALOG_NAMING_UNVERIFIED = "REVIEW_REQUIRED"
+
+[[catalog_profiles]]
+id = "direct_sql_none"
+resolution_mode = "direct_sql"
+external_discovery = "none"
+sync_owner = "none"
+data_lifecycle_owner = "application"
+
+[[catalog_profiles]]
+id = "direct_sql_hive_metastore"
+resolution_mode = "direct_sql"
+external_discovery = "hive_metastore"
+sync_owner = "flink_sink"
+data_lifecycle_owner = "application"
+
+[[catalog_profiles]]
+id = "hudi_dfs_catalog_none"
+resolution_mode = "hudi_dfs_catalog"
+external_discovery = "none"
+sync_owner = "none"
+data_lifecycle_owner = "catalog"
+
+[[catalog_profiles]]
+id = "hudi_dfs_catalog_hive_metastore"
+resolution_mode = "hudi_dfs_catalog"
+external_discovery = "hive_metastore"
+sync_owner = "flink_sink"
+data_lifecycle_owner = "catalog"
+
+[[catalog_profiles]]
+id = "hudi_hms_catalog_hive_metastore"
+resolution_mode = "hudi_hms_catalog"
+external_discovery = "hive_metastore"
+sync_owner = "hudi_hms_catalog"
+data_lifecycle_owner = "application"
+
+[catalog_effective_defaults]
+hive_sync_ignore_exceptions = false
+hive_sync_skip_ro_suffix = false
+hive_sync_support_timestamp = true
+meta_sync_snapshot_with_table_name = true
+hive_sync_table_strategy = "ALL"
+
[existing_table_reconciliation_contract]
reconciliation_schema = 1
supported_design_contract_schemas = [1, 2, 3]
@@ -305,6 +382,54 @@ path = "hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/table/cat
sha256 = "a2b3f87ad44e1475a6f4e06dfd7246436f035af256f28a4ebe2c622672a24aef"
purpose = "Flink catalog option keys"
+[[source_anchors]]
+id = "catalog_factory"
+path = "hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/table/catalog/HoodieCatalogFactory.java"
+sha256 = "a9af3a20c744cb2d4d3a4236ac739db25e45325e2d57b25f0ea975a7db6a3196"
+purpose = "Hudi DFS and HMS catalog mode selection"
+
+[[source_anchors]]
+id = "catalog_util"
+path = "hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/table/catalog/HoodieCatalogUtil.java"
+sha256 = "f52c4d415ba8eb29d217b5ecc62795be137255208558b2d5b12cbaed107d3cf3"
+purpose = "Hive configuration directory, classpath, and embedded-metastore behavior"
+
+[[source_anchors]]
+id = "dfs_catalog"
+path = "hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/table/catalog/HoodieCatalog.java"
+sha256 = "76e9bb3cfc9a1c30630104f9496ee88265b4c6632fcae1f3dec8ff34cda106f4"
+purpose = "DFS catalog path derivation and catalog-owned data lifecycle"
+
+[[source_anchors]]
+id = "hms_catalog"
+path = "hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/table/catalog/HoodieHiveCatalog.java"
+sha256 = "4dedbcded76ee472c721c483ba6a0f69e14aa56510453c0f36ad20808592492c"
+purpose = "HMS external-table lifecycle and effective Hive sync option supplementation"
+
+[[source_anchors]]
+id = "hive_sync_context"
+path = "hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/utils/HiveSyncContext.java"
+sha256 = "4d2eeba0a8403c88707329c0eb26ddaa26dc652c786bdfb22a827fa0255bf929"
+purpose = "Flink-to-Hive sync configuration mapping"
+
+[[source_anchors]]
+id = "stream_write_coordinator"
+path = "hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/StreamWriteOperatorCoordinator.java"
+sha256 = "f6918b1610485e92b3b3d136b0063433515a14a34be79f459501586daa25e69d"
+purpose = "Post-checkpoint-commit asynchronous Hive synchronization"
+
+[[source_anchors]]
+id = "hive_sync_tool"
+path = "hudi-sync/hudi-hive-sync/src/main/java/org/apache/hudi/hive/HiveSyncTool.java"
+sha256 = "a2574227af363fab34e304cc3f65607088fb54b40b93b1091fffa7e1a4841651"
+purpose = "COW and MOR RO, RT, and ALL table registration naming"
+
+[[source_anchors]]
+id = "sync_common_config"
+path = "hudi-sync/hudi-sync-common/src/main/java/org/apache/hudi/sync/common/HoodieSyncConfig.java"
+sha256 = "eb8862397e38332dbd3a61909449df8c94572b7053411f0ba8f058ffc73cda66"
+purpose = "Snapshot-with-base-table-name effective default"
+
[[source_anchors]]
id = "key_generator_options"
path = "hudi-common/src/main/java/org/apache/hudi/keygen/constant/KeyGeneratorOptions.java"
@@ -377,6 +502,37 @@ path = "hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/util/Stre
sha256 = "c771e137ebcd45c94a941857a246dc741e1dde2c0e0158900fd48fe8884f83e2"
purpose = "Write-side record merge mode inference and table-property persistence"
+# Catalog option types are tracked separately from table connector options.
+[[catalog_option_types]]
+key = "catalog.path"
+value_type = "string"
+source_anchors = ["catalog_options", "dfs_catalog"]
+source_symbol = "CATALOG_PATH"
+
+[[catalog_option_types]]
+key = "default-database"
+value_type = "string"
+source_anchors = ["catalog_options"]
+source_symbol = "DEFAULT_DATABASE"
+
+[[catalog_option_types]]
+key = "hive.conf.dir"
+value_type = "string"
+source_anchors = ["catalog_options", "catalog_util"]
+source_symbol = "HIVE_CONF_DIR"
+
+[[catalog_option_types]]
+key = "mode"
+value_type = "string"
+source_anchors = ["catalog_options", "catalog_factory"]
+source_symbol = "MODE"
+
+[[catalog_option_types]]
+key = "table.external"
+value_type = "boolean"
+source_anchors = ["catalog_options", "hms_catalog"]
+source_symbol = "TABLE_EXTERNAL"
+
# This is an allowlist for the bounded executable sink paths, not a claim that it
# enumerates every option available in Hudi 1.2.0.
[[verified_options]]
@@ -427,6 +583,60 @@ value_type = "string"
source_anchors = ["flink_options"]
source_symbol = "COMPACTION_TRIGGER_STRATEGY"
+[[verified_options]]
+key = "hive_sync.db"
+value_type = "string"
+source_anchors = ["flink_options", "hive_sync_context", "hms_catalog"]
+source_symbol = "HIVE_SYNC_DB"
+
+[[verified_options]]
+key = "hive_sync.enabled"
+value_type = "boolean"
+source_anchors = ["flink_options", "stream_write_coordinator", "hms_catalog"]
+source_symbol = "HIVE_SYNC_ENABLED"
+
+[[verified_options]]
+key = "hive_sync.ignore_exceptions"
+value_type = "boolean"
+source_anchors = ["flink_options", "hive_sync_context"]
+source_symbol = "HIVE_SYNC_IGNORE_EXCEPTIONS"
+
+[[verified_options]]
+key = "hive_sync.metastore.uris"
+value_type = "string"
+source_anchors = ["flink_options", "hive_sync_context", "hms_catalog"]
+source_symbol = "HIVE_SYNC_METASTORE_URIS"
+
+[[verified_options]]
+key = "hive_sync.mode"
+value_type = "string"
+source_anchors = ["flink_options", "hive_sync_context", "hms_catalog"]
+source_symbol = "HIVE_SYNC_MODE"
+
+[[verified_options]]
+key = "hive_sync.skip_ro_suffix"
+value_type = "boolean"
+source_anchors = ["flink_options", "hive_sync_context", "hive_sync_tool"]
+source_symbol = "HIVE_SYNC_SKIP_RO_SUFFIX"
+
+[[verified_options]]
+key = "hive_sync.support_timestamp"
+value_type = "boolean"
+source_anchors = ["flink_options", "hive_sync_context", "hms_catalog"]
+source_symbol = "HIVE_SYNC_SUPPORT_TIMESTAMP"
+
+[[verified_options]]
+key = "hive_sync.table"
+value_type = "string"
+source_anchors = ["flink_options", "hive_sync_context", "hms_catalog"]
+source_symbol = "HIVE_SYNC_TABLE"
+
+[[verified_options]]
+key = "hive_sync.table.strategy"
+value_type = "string"
+source_anchors = ["flink_options", "hive_sync_context", "hive_sync_tool"]
+source_symbol = "HIVE_SYNC_TABLE_STRATEGY"
+
[[verified_options]]
key = "hoodie.datasource.write.partitionpath.field"
value_type = "string"
@@ -736,3 +946,27 @@ id = "FLINK_EXISTING_TABLE_PROPERTY_CONFLICT"
requirement = "Compare existing-table identity and durable properties using pinned table-version and partition-field semantics."
python_test = "test_pr5_property_and_schema_conflicts_are_blocked_and_preserved"
java_test = "testPinnedExistingTableTypeOverridesConflictingOption"
+
+[[implemented_acceptance_checks]]
+id = "FLINK_CATALOG_MODE_UNSUPPORTED"
+requirement = "Reject catalog modes and compositions outside the five pinned PR6 profiles."
+python_test = "test_pr6_rejects_unsupported_catalog_compositions"
+java_test = "testPinnedCatalogFactoryRejectsUnknownMode"
+
+[[implemented_acceptance_checks]]
+id = "FLINK_CATALOG_PATH_CONFLICT"
+requirement = "Require the DFS catalog-derived table path to exactly match the desired table path."
+python_test = "test_pr6_rejects_dfs_catalog_path_mismatch"
+java_test = "testPinnedDfsCatalogPathDerivation"
+
+[[implemented_acceptance_checks]]
+id = "FLINK_CATALOG_LIFECYCLE_UNRESOLVED"
+requirement = "Limit the first HMS Catalog path to explicit external application-owned tables."
+python_test = "test_pr6_hms_catalog_requires_external_table"
+java_test = "testPinnedHmsCatalogExternalFlag"
+
+[[implemented_acceptance_checks]]
+id = "FLINK_MOR_CATALOG_NAMING_UNVERIFIED"
+requirement = "Pin MOR Hive registrations to RO, RT, or ALL names derived from query modes."
+python_test = "test_pr6_rejects_mor_catalog_naming_drift"
+java_test = "testPinnedMorHiveSyncNamingDefaults"
diff --git a/hudi-ai-operator/skills/hudi-architect/references/flink-catalog-composition.md b/hudi-ai-operator/skills/hudi-architect/references/flink-catalog-composition.md
new file mode 100644
index 000000000000..ca0f4f8538b0
--- /dev/null
+++ b/hudi-ai-operator/skills/hudi-architect/references/flink-catalog-composition.md
@@ -0,0 +1,153 @@
+<!--
+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.
+-->
+
+# Flink catalog and metastore composition
+
+Use this reference only after the user asks for named catalog resolution or for an external engine
+to discover the table. Catalog composition is orthogonal to design contract schemas 1, 2, and 3;
+it does not create a schema 4 or relax any PR2–PR5 safety invariant.
+
+The baseline is Hudi 1.2.0 at `f05c83f2b97732de7a558ff9b26959e1139c05f5`, Flink 1.20,
+and the Flink 1.20.1 fixture. Validate the checked-in manifest before using this reference.
+
+## Supported profiles
+
+| Profile | Table resolution | External discovery | Sync owner | Data lifecycle owner |
+|---|---|---|---|---|
+| `direct_sql_none` | Direct Flink SQL | None | None | Application |
+| `direct_sql_hive_metastore` | Direct Flink SQL | Hive Metastore | Flink sink | Application |
+| `hudi_dfs_catalog_none` | Hudi DFS Catalog | None | None | Catalog |
+| `hudi_dfs_catalog_hive_metastore` | Hudi DFS Catalog | Hive Metastore | Flink sink | Catalog |
+| `hudi_hms_catalog_hive_metastore` | Hudi HMS Catalog | Hive Metastore | Hudi HMS Catalog | Application |
+
+The legacy contract remains `direct_sql_none` when `safety.external_catalog=false` and no
+`catalog` block is supplied. If that gate is true, require an explicit block. A gate/profile or
+owner mismatch is a conflict, not a request for the validator to choose an architecture.
+
+AWS Glue, HMS without external discovery, managed HMS tables, unknown modes, multi-writer
+operation, existing-table catalog mutation, and unconfirmed DFS lifecycle are not executable.
+
+## Catalog contract
+
+Add an optional top-level block to an existing schema 1, 2, or 3 design:
+
+```json
+{
+ "catalog": {
+ "contract_schema": 1,
+ "resolution_mode": "hudi_hms_catalog",
+ "external_discovery": "hive_metastore",
+ "sync_owner": "hudi_hms_catalog",
+ "data_lifecycle_owner": "application",
+ "catalog_name": "production_hudi",
+ "database": "analytics",
+ "hive_configuration": {
+ "source": "hive_conf_dir",
+ "location": "/etc/hive/conf"
+ },
+ "table_external": true,
+ "mor_sync_strategy": "ALL"
+ }
+}
+```
+
+- `catalog_name` is required for DFS and HMS Catalog and forbidden for direct SQL.
+- `database` is required for catalog resolution or Hive discovery.
+- `catalog_path` is required only for DFS Catalog.
+- `hive_configuration` is required for Hive discovery and forbidden otherwise.
+- Direct/DFS sink-owned sync accepts credential-free `thrift://` `metastore_uris`. Each URI must
+ contain only a host and explicit valid port; user information, paths, query strings, fragments,
+ missing or invalid ports, and credential assignments are rejected before SQL is emitted.
+- HMS Catalog accepts an absolute `hive_conf_dir`, or `classpath` with no location. Embedded HMS
+ is not supported.
+- `table_external=true` is mandatory for HMS Catalog. The application owns its data lifecycle.
+- `mor_sync_strategy` applies only to schema 3 plus Hive discovery and is derived from query modes.
+- DFS and HMS Catalog profiles switch the active catalog and database before the INSERT. Their
+ `source.table` must therefore be a three-part `catalog.database.table` identifier so source
+ resolution does not depend on the caller's original session context. For the bounded streaming
+ new-table path, that identifier must not equal the composed target
+ `catalog_name.database.table.name`; same-table read-transform-write needs a separately bounded
+ input snapshot and is outside PR6.
+
+Do not duplicate the table name, table path, table type, schema, or partition fields in this block.
+They are derived from the existing desired contract.
+
+## Path and lifecycle rules
+
+Direct SQL keeps the explicit desired table path. DFS Catalog derives
+`${catalog.path}/${database}/${table.name}`; the desired path must match exactly, and the table DDL
+must not contain a misleading `path` option. The caller must explicitly accept catalog-owned data:
+Hudi 1.2.0 DFS Catalog recursively deletes data during a table drop. Architect never emits DROP.
+An omitted lifecycle decision remains unresolved, while an explicit application-owned declaration
+conflicts with the DFS Catalog behavior and is blocked.
+
+HMS Catalog must render the desired path in table DDL and must not fall back to a warehouse path.
+Only `table.external=true` is supported so catalog removal does not imply catalog-owned data.
+An omitted `table_external` decision remains unresolved; an explicitly false value requests the
+unsupported managed-table lifecycle and is blocked.
+For multiple partition fields, their order in `table.partition_fields` must match their relative
+order in `table.columns`; Hudi 1.2.0 HMS Catalog rejects a different `PARTITIONED BY` order.
+
+## Hive sync and MOR naming
+
+For direct SQL or DFS Catalog with Hive discovery, the Flink sink owns these table options:
+
+- `hive_sync.enabled=true`
+- `hive_sync.mode=hms`
+- `hive_sync.metastore.uris`
+- `hive_sync.db`
+- `hive_sync.table`
+
+HMS Catalog supplies those values from catalog context; do not duplicate them in table DDL.
+Validate the effective Hudi 1.2.0 defaults in both cases: ignore exceptions false, timestamp support
+true, RO suffix skipping false, and snapshot-with-base-name true. Partition fields come from the
+existing table contract.
+
+For MOR plus Hive discovery, derive the fixed table strategy and registration names:
+
+| Query requirement | Strategy | Registered names |
+|---|---|---|
+| Snapshot only | `RT` | base name |
+| Read optimized only | `RO` | base name |
+| Both | `ALL` | `base_ro`, `base_rt`, and base name |
+
+Custom suffix behavior, custom names, or a different snapshot-with-base-name default requires
+review. Never infer those changes from external-engine preferences.
+
+## Freshness and deployment boundary
+
+Hive sync runs asynchronously after a successful checkpoint commit. Record catalog freshness as
+`post_commit_eventual`; do not claim atomicity with the Hudi commit or an external-visibility SLA.
+Keep commit freshness, Hudi query freshness, and catalog freshness separate.
+
+`catalog_deployment_checks` are operator checks, not static validation failures. Depending on the
+profile they cover bundle dependencies, DFS path existence and permissions, Hive configuration,
+HMS connectivity, database/table/partition registration, MOR names, and external-engine
+visibility. The validator generates SQL but never executes it or probes a live service.
+
+## Existing tables and secrets
+
+An existing-table desired contract may contain a catalog block, but PR6 does not register, alter,
+rename, delete, or take over an existing catalog entry. Any non-legacy catalog composition is
+`FLINK_EXISTING_CATALOG_MUTATION_UNVERIFIED`, remains non-executable, and has an `UNVERIFIED`
+reconciliation and change disposition.
+
+Never request or render usernames, passwords, tokens, IAM material, keytabs, private keys, or
+credential-bearing URIs. Reference credential-free deployment configuration only; do not read or
+echo `hive-site.xml` contents.
diff --git a/hudi-ai-operator/skills/hudi-architect/references/flink-config-templates.md b/hudi-ai-operator/skills/hudi-architect/references/flink-config-templates.md
index 2398c8d6f52d..4b36cec91bd0 100644
--- a/hudi-ai-operator/skills/hudi-architect/references/flink-config-templates.md
+++ b/hudi-ai-operator/skills/hudi-architect/references/flink-config-templates.md
@@ -30,7 +30,9 @@ Verified baseline: Apache Hudi 1.2.0 / Apache Flink 1.20
Baseline ID: hudi-1.2.0-flink-1.20
Hudi source revision: f05c83f2b97732de7a558ff9b26959e1139c05f5
Fixture version: Apache Flink 1.20.1
-Capability manifest schema: 5
+Capability manifest schema: 6
+Catalog contract schema: 1
+Catalog profile: <one pinned profile or unresolved>
Design contract schema: <1 append-only COW | 2 mutable COW | 3 mutable MOR>
Confirmed facts:
@@ -63,7 +65,11 @@ Use only the machine result from `validate_flink_existing_table.py`:
Flink design status: <CONFIG_VALIDATED | INCOMPLETE | BLOCKED | REVIEW_REQUIRED>
Reconciliation result: <COMPATIBLE | CONFLICT | UNVERIFIED>
Verified baseline: Apache Hudi 1.2.0 / Apache Flink 1.20
-Capability manifest schema: 5
+Capability manifest schema: 6
+Catalog contract schema: 1
+Catalog profile: <direct_sql_none or requested non-legacy profile>
+Catalog setup SQL: <always empty for an existing table>
+Effective catalog configuration: <validator-produced mapping>
Reconciliation schema: 1
Desired design contract schema: <1 | 2 | 3>
Writer transition: <same_job_restore | cold_start_new_job | flink_writer_takeover |
@@ -84,6 +90,8 @@ Deployment checks:
- Confirm the single-writer inventory immediately before deployment.
- Recheck pending instants immediately before deployment.
- For a mutable cold start, verify index bootstrap before relying on the first checkpoint.
+- Retain the validator-produced catalog deployment checks for any non-legacy profile; these are
+ observations to perform, not permission to mutate an existing registration.
Executable output:
- Withheld because PR5 is assessment-only, including for COMPATIBLE results.
@@ -104,11 +112,14 @@ Verified baseline: Apache Hudi 1.2.0 / Apache Flink 1.20
Baseline ID: hudi-1.2.0-flink-1.20
Hudi source revision: f05c83f2b97732de7a558ff9b26959e1139c05f5
Fixture version: Apache Flink 1.20.1
-Capability manifest schema: 5
+Capability manifest schema: 6
+Catalog contract schema: 1
+Catalog profile: <direct_sql_none | direct_sql_hive_metastore | hudi_dfs_catalog_none |
+hudi_dfs_catalog_hive_metastore | hudi_hms_catalog_hive_metastore>
Design contract schema: <1 append-only COW | 2 mutable COW | 3 mutable MOR>
Confirmed facts:
-- New table; confirmed single writer; no current external-catalog requirement.
+- New table; confirmed single writer; explicit catalog/discovery outcome.
- Workload: <append-only or mutable, with confirmed replay behavior>.
- Identity: <stable key fields | explicitly accepted append-only auto-generated keys>.
- Mutable only: one event-time ordering field; immutable partition values; full-row delete posture.
@@ -137,13 +148,18 @@ Runtime contract:
Validation summary:
- <baseline evidence>
- <design-contract validation evidence>
+- <effective catalog configuration and post-commit-eventual freshness when applicable>
- <advisories, including stable-key or auto-key durability>
+Catalog deployment checks:
+- <validator-produced check codes; these do not claim live connectivity or change static status>
+
Executable eligible: true
```
-Then emit, in order, the validator-produced `runtime_sql`, `table_ddl`, and `insert_sql`. Do not
-retype or modify those artifacts after validation.
+Then emit the validator-produced `combined_sql`. Its canonical order is `runtime_sql`, optional
+`catalog_setup_sql`, `table_ddl`, and `insert_sql`; the JSON form retains all boundaries. Do not
+retype, modify, or execute those artifacts after validation.
## Canonical SQL contract
@@ -231,6 +247,12 @@ state at a visible completed instant; read-optimized readers expose compacted ba
current only after compaction. Five delta commits is a trigger, not a time SLA. Reader-side
`hoodie.datasource.query.type` is intentionally absent from sink DDL.
+For catalog profiles, apply the additional canonical rules in
+`flink-catalog-composition.md`. DFS Catalog setup precedes table DDL and the DDL omits `path`;
+external HMS Catalog includes the desired table path and does not duplicate its auto-supplied Hive
+sync options. Direct/DFS sink-owned discovery renders the five bounded `hive_sync.*` settings.
+MOR plus Hive discovery also renders the query-derived `hive_sync.table.strategy`.
+
## Deployment checks
Always retain these outside the static success claim:
diff --git a/hudi-ai-operator/skills/hudi-architect/references/flink-decision-overrides.md b/hudi-ai-operator/skills/hudi-architect/references/flink-decision-overrides.md
index 6a863960e441..e17bcacfc70f 100644
--- a/hudi-ai-operator/skills/hudi-architect/references/flink-decision-overrides.md
+++ b/hudi-ai-operator/skills/hudi-architect/references/flink-decision-overrides.md
@@ -232,7 +232,8 @@ to the executable contract just like `F10_SAFE_APPEND`.
## PR3 mutable COW executable invariants
-- Version-2 contracts cover only new-table, single-writer, no-external-catalog streaming COW.
+- Version-2 contracts cover only new-table, single-writer streaming COW. Catalog composition is
+ orthogonal and, when requested, must pass one of the five PR6 profiles.
- Identity is a stable, non-null primary key shared by source and target. Auto-key is blocked.
- The source declares normalized UPSERT changelog (`INSERT`, `UPDATE_AFTER`, `DELETE`) without
`UPDATE_BEFORE`. Deletes carry the full projected row, including key, ordering, and partition.
@@ -264,8 +265,9 @@ to the executable contract just like `F10_SAFE_APPEND`.
- Snapshot readers see base plus log state at a visible completed instant. Read-optimized readers
see compacted base-file state and advance only after compaction. Reader `query.type` is not a
sink option and is not rendered into the DDL.
-- Clustering is disabled. Detailed compaction tuning, append-only MOR, existing tables, catalogs,
- multi-writer operation, and live service validation remain outside this path.
+- Clustering is disabled. Detailed compaction tuning, append-only MOR, existing tables,
+ non-PR6 catalog profiles, multi-writer operation, and live service validation remain outside
+ this path.
## PR5 existing-table reconciliation invariants
@@ -291,6 +293,46 @@ to the executable contract just like `F10_SAFE_APPEND`.
must recheck writer and timeline state, and no SQL, mutation, upgrade, or migration command is
emitted.
+## PR6 catalog/metastore composition invariants
+
+- Catalog composition is orthogonal to design schemas 1, 2, and 3. It never selects table type,
+ identity, ordering, compaction, or a writer topology.
+- No catalog block plus `safety.external_catalog=false` remains the byte-stable
+ `direct_sql_none` path. A true gate without the block is
+ `FLINK_CATALOG_CONTRACT_REQUIRED`; a gate/profile mismatch is
+ `FLINK_CATALOG_COMPOSITION_CONFLICT`.
+- Only five profiles are executable: direct SQL without discovery, direct SQL plus HMS sync, DFS
+ Catalog without discovery, DFS Catalog plus HMS sync, and external HMS Catalog plus HMS
+ discovery. Unknown or incompatible modes are `FLINK_CATALOG_MODE_UNSUPPORTED` or
+ `FLINK_CATALOG_COMPOSITION_CONFLICT`.
+- DFS Catalog requires exact `${catalog.path}/${database}/${table.name}` path agreement. A
+ mismatch is `FLINK_CATALOG_PATH_CONFLICT`. Catalog-owned lifecycle must be explicit; otherwise
+ use `FLINK_CATALOG_LIFECYCLE_UNRESOLVED`. An explicit application-owned declaration is a blocked
+ `FLINK_CATALOG_COMPOSITION_CONFLICT`, not an unresolved decision.
+- HMS Catalog is executable only with `table.external=true`, application-owned data, a
+ non-embedded Hive configuration, and an explicit desired table path. A missing external-table
+ decision is `FLINK_CATALOG_LIFECYCLE_UNRESOLVED`, while explicit `table_external=false` is a
+ blocked `FLINK_CATALOG_COMPOSITION_CONFLICT`. Multiple HMS partition fields must keep the same
+ relative order in the physical schema and partition declaration; a mismatch is also
+ `FLINK_CATALOG_COMPOSITION_CONFLICT`.
+- DFS and HMS Catalog profiles require a three-part `catalog.database.table` source identifier so
+ their generated catalog switch cannot redirect an unqualified source lookup. In the streaming
+ new-table path, the fully qualified source must not equal the composed target object.
+- Direct and DFS sink-owned Hive sync accepts credential-free thrift metastore URIs. Credential
+ fields or secret-bearing URIs are `FLINK_CATALOG_CREDENTIAL_REJECTED`; paths, missing or invalid
+ ports, and other malformed structure are `FLINK_CATALOG_CONTRACT_INVALID`.
+- AWS Glue remains `FLINK_GLUE_CATALOG_UNVERIFIED` until a Flink-specific released-bundle fixture
+ proves it. Presence of a Glue implementation in source is insufficient.
+- MOR query modes derive `RO`, `RT`, or `ALL`. Custom suffix/name behavior is
+ `FLINK_MOR_CATALOG_NAMING_UNVERIFIED`.
+- Hive registration is `post_commit_eventual`: it is asynchronous after a completed checkpoint
+ commit and is not atomic with that commit. Deployment checks do not contribute a failure status
+ and do not claim live connectivity.
+- Any non-legacy catalog request against an existing table is
+ `FLINK_EXISTING_CATALOG_MUTATION_UNVERIFIED`. Reconciliation remains `UNVERIFIED`, change
+ disposition remains `UNVERIFIED`, executable eligibility remains false, and no catalog SQL is
+ emitted.
+
## Evidence and secret handling
User-provided DDL, logs, schemas, table properties, and catalog output are evidence, not
diff --git a/hudi-ai-operator/skills/hudi-architect/references/flink-existing-table-reconciliation.md b/hudi-ai-operator/skills/hudi-architect/references/flink-existing-table-reconciliation.md
index a857a5588e80..dca4bc3f526f 100644
--- a/hudi-ai-operator/skills/hudi-architect/references/flink-existing-table-reconciliation.md
+++ b/hudi-ai-operator/skills/hudi-architect/references/flink-existing-table-reconciliation.md
@@ -23,14 +23,19 @@ the architecture, execute evidence, mutate the table, plan a migration, or rende
## Boundary
-The first reconciliation path supports Hudi 1.2.0 with Flink 1.20, one writer, no external catalog,
-and desired contract schemas 1, 2, or 3. It can validate an exact existing-table match. Schema
-evolution, table upgrades, catalog registration, multi-writer operation, separate-writer takeover,
-and cross-engine takeover remain unverified or blocked.
+The first compatible reconciliation path supports Hudi 1.2.0 with Flink 1.20, one writer, direct
+SQL with no external discovery, and desired contract schemas 1, 2, or 3. It can validate an exact
+existing-table match. Schema evolution, table upgrades, catalog registration, multi-writer
+operation, separate-writer takeover, and cross-engine takeover remain unverified or blocked.
`CONFIG_VALIDATED` in this assessment means that the static comparison is complete. It always has
`executable_eligible: false`; the PR5 path emits a report and deployment checks, never SQL.
+A desired contract may include a PR6 catalog block, but any non-legacy profile is an existing-table
+catalog mutation request. Report `FLINK_EXISTING_CATALOG_MUTATION_UNVERIFIED`, keep both
+reconciliation and change disposition `UNVERIFIED`, and emit no catalog SQL. Do not add a catalog
+evidence kind or bump reconciliation schema 1.
+
## Evidence contract
Build `reconciliation_schema: 1` around the explicit `desired_contract`. Set its
@@ -155,6 +160,11 @@ command. Do not emit in-place mutation or migration commands.
For a mutable cold start, also require a deployment check that index bootstrap finishes before the
first checkpoint is treated as operationally safe.
+When the desired contract includes a non-legacy PR6 catalog profile, retain its catalog contract
+schema, profile, effective configuration, and deployment checks in the machine report. Keep
+`catalog_setup_sql` empty: those fields describe the requested state and remaining observations;
+they do not authorize registration, alteration, takeover, or any other catalog mutation.
+
## Running the validator
Run:
diff --git a/hudi-ai-operator/skills/hudi-architect/references/flink-question-flow.md b/hudi-ai-operator/skills/hudi-architect/references/flink-question-flow.md
index 4eaa46ffd255..c923c7b6c78d 100644
--- a/hudi-ai-operator/skills/hudi-architect/references/flink-question-flow.md
+++ b/hudi-ai-operator/skills/hudi-architect/references/flink-question-flow.md
@@ -22,6 +22,8 @@ HoodieStreamer requests.
The PR1 gates remain the entry to this flow. PR2, PR3, and PR4 add bounded new-table executable
paths. PR5 adds a read-only existing-table reconciliation path after the same workload decisions.
+PR6 composes five pinned direct SQL, DFS Catalog, and HMS Catalog profiles with those same design
+schemas; it does not add a design schema or choose a table architecture.
Maintain a findings list throughout this flow. Adding a finding does not end the interview while
another independent gate can still be evaluated. Ask dependent follow-ups only when their
@@ -100,13 +102,18 @@ Ask in outcome language:
> - **Not now, maybe later**
> - **Not sure**
-- No → record no current external-catalog requirement and continue.
-- Yes → `REVIEW_REQUIRED` with `FLINK_EXTERNAL_CATALOG_REVIEW`.
-- Not now, maybe later → continue, record a catalog revisit condition, and generate no catalog
- configuration.
+- No → default to direct SQL with no external discovery, do not ask catalog implementation
+ questions, and continue.
+- Yes → ask whether the desired outcome is only external name-based discovery, shared Hudi catalog
+ resolution, or both. Then collect only the fields needed by a supported PR6 profile: catalog
+ name/path and lifecycle acceptance for DFS, credential-free Hive configuration source and
+ database for HMS, and MOR query modes when applicable.
+- Not now, maybe later → continue with direct SQL and no catalog block; record that later mutation
+ against an existing table is not covered by PR6 and requires review.
- Not sure → `REVIEW_REQUIRED` with `FLINK_CATALOG_REQUIREMENT_UNRESOLVED`.
-Flink catalog and metastore composition arrives in PR6. Do not reuse the Spark sync templates.
+Phrase the follow-up in outcomes, not implementation jargon. Do not ask for credentials. Validate
+the resulting block with `flink-catalog-composition.md`; do not reuse Spark sync templates.
## F4 — Physical schema availability
@@ -221,6 +228,9 @@ Enter this stage only when F0-F6 have no status-contributing finding. Collect, w
- Partition fields, or an explicit unpartitioned decision.
- Stable record-key fields when F5 found a stable business key.
- The existing source-table identifier and its expected physical columns.
+- For DFS or HMS Catalog resolution, a three-part `catalog.database.table` source identifier; do
+ not preserve an unqualified source across the generated catalog switch, and confirm it is not
+ the same complete object as the new sink target.
- For mutable input: source primary-key fields, exactly one ordering field, whether deletes occur,
whether deletes contain the complete projected row, and whether partition values can change.
@@ -233,6 +243,10 @@ Avro-backed Hudi schema constraint. `TIME`, `TIMESTAMP`, and `TIMESTAMP_LTZ` pre
between 0 and 6 for the pinned connector. Reject violations instead of renaming a field or
reducing its precision.
+For HMS Catalog, multiple partition fields must appear in `table.partition_fields` in the same
+relative order as they appear in `table.columns`. Reject a mismatch instead of silently reordering
+the declared physical schema.
+
`BYTES`, `BINARY`, and `VARBINARY` remain valid payload types, but must not be used as stable
record-key or partition fields. The pinned Hudi 1.2.0 routing path converts their Java arrays to
object-identity strings, so equal byte sequences do not have deterministic routing values. Add
@@ -388,7 +402,7 @@ the report and deployment checks with executable eligibility `false`, and never
known table-property or schema conflict is `CONFLICT`; incomplete, contradictory, evolutionary,
or unsupported evidence is `UNVERIFIED`.
-## PR2/PR3/PR4/PR5 validation and completion
+## PR2/PR3/PR4/PR5/PR6 validation and completion
After all applicable questions, build a version-1 JSON contract for append-only COW, a version-2
contract for mutable COW, or a version-3 contract for mutable MOR. All contain confirmed safety
@@ -399,6 +413,13 @@ read-optimized freshness, compaction ownership and fixed schedule/execution sett
clustering. Do not insert placeholders. Record an explicitly empty
pass-through connector-option map; no renderer accepts an ignored option.
+When F3 selects catalog resolution or external discovery, add catalog contract schema 1 using only
+the confirmed PR6 fields. Do not change the design schema. For no catalog outcome, omit the block
+and retain `external_catalog=false`; the validator normalizes the legacy direct profile. For an
+existing table, a non-legacy block is accepted only to report
+`FLINK_EXISTING_CATALOG_MUTATION_UNVERIFIED`; do not collect a new evidence kind or infer that
+registration already exists.
+
For a new table, run:
```bash
@@ -407,7 +428,8 @@ python3 validate_flink_design.py --input <redacted-design-contract.json>
The validator is allowed to reject or serialize explicit decisions; it must not select
architecture decisions or fill missing values. Preserve every returned finding. Only a successful
-result may provide the canonical runtime SQL, Hudi DDL, and `INSERT INTO` and end with:
+result may provide canonical runtime SQL, optional catalog setup SQL, Hudi DDL, and `INSERT INTO`
+and end with:
```text
Status: CONFIG_VALIDATED
@@ -415,4 +437,5 @@ Executable eligible: true
```
Any validation failure withholds all SQL and reports executable eligibility as `false`. Consult
-`flink-decision-overrides.md` for precedence and `flink-config-templates.md` for both envelopes.
+`flink-decision-overrides.md` for precedence, `flink-catalog-composition.md` for PR6 profiles, and
+`flink-config-templates.md` for both envelopes.
diff --git a/hudi-ai-operator/skills/hudi-architect/references/flink-warnings.md b/hudi-ai-operator/skills/hudi-architect/references/flink-warnings.md
index 7f1a153dd9a6..210408fe2448 100644
--- a/hudi-ai-operator/skills/hudi-architect/references/flink-warnings.md
+++ b/hudi-ai-operator/skills/hudi-architect/references/flink-warnings.md
@@ -214,9 +214,11 @@ be treated as safe.
**Trigger:** An external engine or BI tool must discover the table through a catalog or metastore.
**Message:** Flink table resolution and external metastore synchronization are separate
-responsibilities that must be composed together. That composition is deferred to PR6.
+responsibilities. The outcome gate must be resolved into one explicit PR6 profile before SQL is
+eligible.
-**Action:** `REVIEW_REQUIRED`; generate no catalog or sync configuration.
+**Action:** `REVIEW_REQUIRED` at the outcome gate; collect the PR6 contract rather than guessing a
+catalog or sync owner.
## FLINK_CATALOG_REQUIREMENT_UNRESOLVED
@@ -418,7 +420,8 @@ rename it.
## FLINK_SOURCE_CONTRACT_REQUIRED
-**Trigger:** The source table, expected physical schema, or changelog semantics are unavailable.
+**Trigger:** The source table, expected physical schema, or changelog semantics are unavailable,
+or a catalog-managed profile supplies a source identifier that is not fully qualified.
**Message:** A sink-side example must identify the existing source table and its expected contract.
@@ -683,3 +686,93 @@ configuration or secret references; never paste live credentials into the design
**Action:** Continue with redacted evidence. If redaction makes a load-bearing fact unreadable, mark
that fact `INCOMPLETE` without asking for the secret value.
+
+## FLINK_CATALOG_CONTRACT_REQUIRED
+
+**Trigger:** `safety.external_catalog=true` but no versioned catalog block is supplied.
+
+**Message:** The boolean gate does not identify table resolution, discovery, sync ownership, or
+data lifecycle.
+
+**Action:** `INCOMPLETE`; collect the PR6 catalog contract and emit no SQL.
+
+## FLINK_CATALOG_CONTRACT_INVALID
+
+**Trigger:** The catalog block is malformed, contains inapplicable fields, or lacks a required
+identifier or Hive configuration source.
+
+**Message:** Static validation cannot reproduce the catalog design.
+
+**Action:** `INCOMPLETE`; correct the contract without selecting architecture on the user's behalf.
+
+## FLINK_CATALOG_MODE_UNSUPPORTED
+
+**Trigger:** Resolution, discovery, or sync ownership uses a mode outside the pinned PR6 surface.
+
+**Message:** Hudi 1.2.0 source contains more integration code than this Flink fixture proves.
+
+**Action:** `BLOCKED`; use one of the five supported profiles or add pinned evidence later.
+
+## FLINK_CATALOG_PATH_CONFLICT
+
+**Trigger:** A DFS Catalog desired path differs from
+`${catalog.path}/${database}/${table.name}`.
+
+**Message:** Hudi DFS Catalog ignores an explicit table path and derives its own location.
+
+**Action:** `BLOCKED`; reconcile the desired path before rendering catalog SQL.
+
+## FLINK_CATALOG_LIFECYCLE_UNRESOLVED
+
+**Trigger:** DFS catalog-owned deletion semantics or the HMS external-table decision is missing.
+
+**Message:** Catalog operations can otherwise imply a different data lifecycle than the design.
+
+**Action:** `REVIEW_REQUIRED`; collect the missing lifecycle decision and do not emit executable
+catalog SQL. An explicit conflicting decision uses `FLINK_CATALOG_COMPOSITION_CONFLICT` instead.
+
+## FLINK_CATALOG_COMPOSITION_CONFLICT
+
+**Trigger:** The legacy boolean gate, resolution/discovery pair, sync owner, lifecycle owner, or
+HMS physical-schema partition order contradicts another explicit field; an HMS managed table is
+explicitly requested; or the streaming new-table source resolves to the composed sink object.
+
+**Message:** The validator does not arbitrate conflicting architecture decisions.
+
+**Action:** `BLOCKED`; make the contract internally consistent.
+
+## FLINK_GLUE_CATALOG_UNVERIFIED
+
+**Trigger:** AWS Glue is selected for external discovery.
+
+**Message:** The pinned source contains Glue support, but PR6 has no Flink-specific released-bundle
+fixture or live deployment evidence for an executable path.
+
+**Action:** `REVIEW_REQUIRED`; emit no Glue configuration or SQL.
+
+## FLINK_MOR_CATALOG_NAMING_UNVERIFIED
+
+**Trigger:** MOR sync strategy disagrees with query modes, or custom suffix/snapshot naming is
+requested.
+
+**Message:** PR6 fixes RO, RT, and ALL registrations with RO suffix skipping disabled and snapshot
+with base name enabled.
+
+**Action:** `REVIEW_REQUIRED`; retain the requested naming as a later capability.
+
+## FLINK_CATALOG_CREDENTIAL_REJECTED
+
+**Trigger:** Catalog configuration contains a username, password, token, IAM/keytab material, or a
+credential-bearing metastore URI.
+
+**Message:** Architect contracts may reference credential-free deployment configuration only.
+
+**Action:** `BLOCKED`; remove the secret and use deployment-managed authentication.
+
+## FLINK_EXISTING_CATALOG_MUTATION_UNVERIFIED
+
+**Trigger:** An existing-table desired contract requests any non-legacy catalog composition.
+
+**Message:** PR6 does not register, alter, rename, delete, or take over existing catalog entries.
+
+**Action:** `REVIEW_REQUIRED`; return an `UNVERIFIED`, non-executable reconciliation with no SQL.
diff --git a/hudi-ai-operator/skills/hudi-architect/validate_flink_capabilities.py b/hudi-ai-operator/skills/hudi-architect/validate_flink_capabilities.py
index 49360efb13de..cc28beee9221 100644
--- a/hudi-ai-operator/skills/hudi-architect/validate_flink_capabilities.py
+++ b/hudi-ai-operator/skills/hudi-architect/validate_flink_capabilities.py
@@ -40,7 +40,7 @@ DEFAULT_MANIFEST = (
SKILL_DIR / "references" / "flink-1.20-hudi-1.2.0-capabilities.toml"
)
-EXPECTED_SCHEMA_VERSION = 5
+EXPECTED_SCHEMA_VERSION = 6
EXPECTED_BASELINE_ID = "hudi-1.2.0-flink-1.20"
EXPECTED_HUDI_VERSION = "1.2.0"
EXPECTED_HUDI_SOURCE_REVISION = "f05c83f2b97732de7a558ff9b26959e1139c05f5"
@@ -222,6 +222,117 @@ EXPECTED_RUNTIME_CONSTRAINTS = {
"checkpoint_interval_min_ms": 1000,
"flink_checkpoint_interval_min_ms": 10,
}
+EXPECTED_CATALOG_COMPOSITION_CONTRACT = {
+ "contract_schema": 1,
+ "supported_design_contract_schemas": [1, 2, 3],
+ "new_table_executable": True,
+ "existing_table_assessment_only": True,
+ "catalog_freshness": "post_commit_eventual",
+ "resolution_modes": ["direct_sql", "hudi_dfs_catalog", "hudi_hms_catalog"],
+ "external_discovery_modes": ["none", "hive_metastore", "aws_glue"],
+ "sync_owners": ["none", "flink_sink", "hudi_hms_catalog"],
+ "data_lifecycle_owners": ["application", "catalog"],
+ "deployment_checks": [
+ "verify_catalog_dependencies",
+ "verify_catalog_path_exists",
+ "verify_catalog_path_permissions",
+ "verify_hive_configuration_available",
+ "verify_hms_connectivity",
+ "verify_catalog_database_registration",
+ "verify_table_registration_after_commit",
+ "verify_partition_registration_after_commit",
+ "verify_mor_catalog_table_names",
+ "verify_external_engine_visibility",
+ ],
+ "finding_status": {
+ "FLINK_CATALOG_COMPOSITION_CONFLICT": "BLOCKED",
+ "FLINK_CATALOG_CONTRACT_INVALID": "INCOMPLETE",
+ "FLINK_CATALOG_CONTRACT_REQUIRED": "INCOMPLETE",
+ "FLINK_CATALOG_CREDENTIAL_REJECTED": "BLOCKED",
+ "FLINK_CATALOG_LIFECYCLE_UNRESOLVED": "REVIEW_REQUIRED",
+ "FLINK_CATALOG_MODE_UNSUPPORTED": "BLOCKED",
+ "FLINK_CATALOG_PATH_CONFLICT": "BLOCKED",
+ "FLINK_EXISTING_CATALOG_MUTATION_UNVERIFIED": "REVIEW_REQUIRED",
+ "FLINK_GLUE_CATALOG_UNVERIFIED": "REVIEW_REQUIRED",
+ "FLINK_MOR_CATALOG_NAMING_UNVERIFIED": "REVIEW_REQUIRED",
+ },
+}
+EXPECTED_CATALOG_PROFILES = [
+ {
+ "id": "direct_sql_none",
+ "resolution_mode": "direct_sql",
+ "external_discovery": "none",
+ "sync_owner": "none",
+ "data_lifecycle_owner": "application",
+ },
+ {
+ "id": "direct_sql_hive_metastore",
+ "resolution_mode": "direct_sql",
+ "external_discovery": "hive_metastore",
+ "sync_owner": "flink_sink",
+ "data_lifecycle_owner": "application",
+ },
+ {
+ "id": "hudi_dfs_catalog_none",
+ "resolution_mode": "hudi_dfs_catalog",
+ "external_discovery": "none",
+ "sync_owner": "none",
+ "data_lifecycle_owner": "catalog",
+ },
+ {
+ "id": "hudi_dfs_catalog_hive_metastore",
+ "resolution_mode": "hudi_dfs_catalog",
+ "external_discovery": "hive_metastore",
+ "sync_owner": "flink_sink",
+ "data_lifecycle_owner": "catalog",
+ },
+ {
+ "id": "hudi_hms_catalog_hive_metastore",
+ "resolution_mode": "hudi_hms_catalog",
+ "external_discovery": "hive_metastore",
+ "sync_owner": "hudi_hms_catalog",
+ "data_lifecycle_owner": "application",
+ },
+]
+EXPECTED_CATALOG_EFFECTIVE_DEFAULTS = {
+ "hive_sync_ignore_exceptions": False,
+ "hive_sync_skip_ro_suffix": False,
+ "hive_sync_support_timestamp": True,
+ "meta_sync_snapshot_with_table_name": True,
+ "hive_sync_table_strategy": "ALL",
+}
+EXPECTED_CATALOG_OPTION_TYPES = [
+ {
+ "key": "catalog.path",
+ "value_type": "string",
+ "source_anchors": ["catalog_options", "dfs_catalog"],
+ "source_symbol": "CATALOG_PATH",
+ },
+ {
+ "key": "default-database",
+ "value_type": "string",
+ "source_anchors": ["catalog_options"],
+ "source_symbol": "DEFAULT_DATABASE",
+ },
+ {
+ "key": "hive.conf.dir",
+ "value_type": "string",
+ "source_anchors": ["catalog_options", "catalog_util"],
+ "source_symbol": "HIVE_CONF_DIR",
+ },
+ {
+ "key": "mode",
+ "value_type": "string",
+ "source_anchors": ["catalog_options", "catalog_factory"],
+ "source_symbol": "MODE",
+ },
+ {
+ "key": "table.external",
+ "value_type": "boolean",
+ "source_anchors": ["catalog_options", "hms_catalog"],
+ "source_symbol": "TABLE_EXTERNAL",
+ },
+]
EXPECTED_EXISTING_TABLE_RECONCILIATION_CONTRACT = {
"reconciliation_schema": 1,
"supported_design_contract_schemas": [1, 2, 3],
@@ -343,6 +454,18 @@ EXPECTED_ACCEPTANCE_EVIDENCE = {
"test_pr2_rejects_binary_routing_fields",
"testPinnedBinaryValuesUseObjectIdentityForRouting",
),
+ "FLINK_CATALOG_MODE_UNSUPPORTED": (
+ "test_pr6_rejects_unsupported_catalog_compositions",
+ "testPinnedCatalogFactoryRejectsUnknownMode",
+ ),
+ "FLINK_CATALOG_PATH_CONFLICT": (
+ "test_pr6_rejects_dfs_catalog_path_mismatch",
+ "testPinnedDfsCatalogPathDerivation",
+ ),
+ "FLINK_CATALOG_LIFECYCLE_UNRESOLVED": (
+ "test_pr6_hms_catalog_requires_external_table",
+ "testPinnedHmsCatalogExternalFlag",
+ ),
"FLINK_RECORD_KEY_FIELD_MISSING": (
"test_pr2_rejects_record_key_missing_from_append_schema",
"testPinnedFactorySkipsMissingRecordKeyCheckInAppendMode",
@@ -403,6 +526,10 @@ EXPECTED_ACCEPTANCE_EVIDENCE = {
"test_pr4_rejects_strict_read_optimized_freshness",
"testPinnedMorQuerySemantics",
),
+ "FLINK_MOR_CATALOG_NAMING_UNVERIFIED": (
+ "test_pr6_rejects_mor_catalog_naming_drift",
+ "testPinnedMorHiveSyncNamingDefaults",
+ ),
"FLINK_PR4_WRITE_PATH_UNSUPPORTED": (
"test_pr4_valid_mutable_mor_design_renders_exact_executable_sql",
"testMutableMorSinkAndCompactionPlan",
@@ -427,7 +554,9 @@ def validate_manifest(manifest: dict[str, Any]) -> list[str]:
expected_scalars = {
"schema_version": EXPECTED_SCHEMA_VERSION,
"baseline_id": EXPECTED_BASELINE_ID,
- "validation_surface": "flink-sql-sink-and-existing-table-reconciliation",
+ "validation_surface": (
+ "flink-sql-sink-existing-table-reconciliation-and-catalog-composition"
+ ),
}
for key, expected in expected_scalars.items():
if manifest.get(key) != expected:
@@ -635,6 +764,33 @@ def validate_manifest(manifest: dict[str, Any]) -> list[str]:
"existing_table_reconciliation_contract does not match the bounded PR5 path"
)
+ catalog_contract = manifest.get("catalog_composition_contract")
+ if catalog_contract != EXPECTED_CATALOG_COMPOSITION_CONTRACT:
+ errors.append(
+ "catalog_composition_contract does not match the bounded PR6 path"
+ )
+
+ catalog_profiles = manifest.get("catalog_profiles")
+ if catalog_profiles != EXPECTED_CATALOG_PROFILES:
+ errors.append("catalog_profiles must contain exactly the five PR6 profiles")
+
+ catalog_defaults = manifest.get("catalog_effective_defaults")
+ if catalog_defaults != EXPECTED_CATALOG_EFFECTIVE_DEFAULTS:
+ errors.append(
+ "catalog_effective_defaults do not match the pinned Hudi 1.2.0 behavior"
+ )
+
+ catalog_options = manifest.get("catalog_option_types")
+ if catalog_options != EXPECTED_CATALOG_OPTION_TYPES:
+ errors.append("catalog_option_types do not match the pinned catalog surface")
+ else:
+ for index, option in enumerate(catalog_options):
+ unknown = sorted(set(option["source_anchors"]) - anchor_ids)
+ if unknown:
+ errors.append(
+ f"catalog_option_types[{index}] references unknown source anchors: {unknown}"
+ )
+
physical_types = manifest.get("physical_types")
if not isinstance(physical_types, dict):
errors.append("physical_types must be a table")
@@ -709,13 +865,19 @@ def supported_option_keys(manifest: dict[str, Any]) -> set[str]:
def validation_evidence(
- manifest: dict[str, Any], contract_schema: int | None = None
+ manifest: dict[str, Any],
+ contract_schema: int | None = None,
+ catalog_profile: str = "direct_sql_none",
) -> dict[str, Any]:
"""Build the stable baseline evidence included in every Flink assessment."""
return {
"baseline_id": manifest["baseline_id"],
"capability_manifest_schema": manifest["schema_version"],
+ "catalog_contract_schema": manifest["catalog_composition_contract"][
+ "contract_schema"
+ ],
+ "catalog_profile": catalog_profile,
"design_contract_schema": (
manifest["executable_contract"]["contract_schema"]
if contract_schema is None
@@ -728,6 +890,9 @@ def validation_evidence(
],
"flink_fixture_version": manifest["flink"]["fixture_version"],
"hudi_source_revision": manifest["hudi"]["source_revision"],
+ "supported_catalog_profiles": [
+ profile["id"] for profile in manifest["catalog_profiles"]
+ ],
}
diff --git a/hudi-ai-operator/skills/hudi-architect/validate_flink_design.py b/hudi-ai-operator/skills/hudi-architect/validate_flink_design.py
index 9e4ab18abafa..14310e851594 100644
--- a/hudi-ai-operator/skills/hudi-architect/validate_flink_design.py
+++ b/hudi-ai-operator/skills/hudi-architect/validate_flink_design.py
@@ -31,7 +31,7 @@ import re
import sys
from pathlib import Path
from typing import Any, TypeGuard
-from urllib.parse import SplitResult, parse_qsl, urlsplit
+from urllib.parse import SplitResult, parse_qsl, unquote, urlsplit
from validate_flink_capabilities import (
DEFAULT_MANIFEST,
@@ -44,6 +44,10 @@ from validate_flink_capabilities import (
APPEND_CONTRACT_SCHEMA_VERSION = 1
MUTABLE_CONTRACT_SCHEMA_VERSION = 2
MUTABLE_MOR_CONTRACT_SCHEMA_VERSION = 3
+CATALOG_CONTRACT_SCHEMA_VERSION = 1
+CATALOG_MANAGED_RESOLUTION_MODES = frozenset(
+ {"hudi_dfs_catalog", "hudi_hms_catalog"}
+)
SUPPORTED_CONTRACT_SCHEMAS = {
APPEND_CONTRACT_SCHEMA_VERSION,
MUTABLE_CONTRACT_SCHEMA_VERSION,
@@ -62,6 +66,14 @@ SENSITIVE_QUERY_KEY = re.compile(
r"(?:access[_-]?key|api[_-]?key|credential|password|secret|signature|token)",
re.IGNORECASE,
)
+SENSITIVE_CATALOG_KEY = re.compile(
+ r"(?:access[_-]?key|api[_-]?key|credential|iam|keytab|password|secret|token|username)",
+ re.IGNORECASE,
+)
+SENSITIVE_URI_ASSIGNMENT = re.compile(
+ r"(?:^|[/;,&])(?:access[_-]?key|api[_-]?key|credential|iam|keytab|password|secret|sig|signature|token|username)\s*=",
+ re.IGNORECASE,
+)
PLACEHOLDER = re.compile(r"(?:<[^>]+>|\$\{[^}]+\}|\b(?:TBD|TODO)\b)", re.IGNORECASE)
PARAMETERIZED_TYPE = re.compile(
r"^(?P<name>DECIMAL|CHAR|VARCHAR|BINARY|VARBINARY|TIME|TIMESTAMP|TIMESTAMP_LTZ)"
@@ -75,6 +87,13 @@ FINDING_STATUS = {
"FLINK_BASELINE_EVIDENCE_INVALID": "BLOCKED",
"FLINK_BINARY_ROUTING_FIELD_UNSUPPORTED": "BLOCKED",
"FLINK_CATALOG_REQUIREMENT_UNRESOLVED": "REVIEW_REQUIRED",
+ "FLINK_CATALOG_COMPOSITION_CONFLICT": "BLOCKED",
+ "FLINK_CATALOG_CONTRACT_INVALID": "INCOMPLETE",
+ "FLINK_CATALOG_CONTRACT_REQUIRED": "INCOMPLETE",
+ "FLINK_CATALOG_CREDENTIAL_REJECTED": "BLOCKED",
+ "FLINK_CATALOG_LIFECYCLE_UNRESOLVED": "REVIEW_REQUIRED",
+ "FLINK_CATALOG_MODE_UNSUPPORTED": "BLOCKED",
+ "FLINK_CATALOG_PATH_CONFLICT": "BLOCKED",
"FLINK_CHECKPOINTING_REQUIRED": "BLOCKED",
"FLINK_CHECKPOINT_INTERVAL_REQUIRED": "INCOMPLETE",
"FLINK_CHECKPOINT_INTERVAL_UNSUPPORTED": "BLOCKED",
@@ -84,6 +103,7 @@ FINDING_STATUS = {
"FLINK_EXISTING_EVIDENCE_INVALID": "INCOMPLETE",
"FLINK_EXISTING_EVIDENCE_REQUIRED": "INCOMPLETE",
"FLINK_EXISTING_ENGINE_TAKEOVER_UNVERIFIED": "REVIEW_REQUIRED",
+ "FLINK_EXISTING_CATALOG_MUTATION_UNVERIFIED": "REVIEW_REQUIRED",
"FLINK_EXISTING_SCHEMA_CONFLICT": "BLOCKED",
"FLINK_EXISTING_SCHEMA_EVOLUTION_UNVERIFIED": "REVIEW_REQUIRED",
"FLINK_EXISTING_TABLE_PROPERTY_CONFLICT": "BLOCKED",
@@ -94,6 +114,7 @@ FINDING_STATUS = {
"FLINK_EXISTING_WRITER_TRANSITION_UNRESOLVED": "REVIEW_REQUIRED",
"FLINK_EXISTING_WRITER_CONFIG_CONFLICT": "REVIEW_REQUIRED",
"FLINK_EXTERNAL_CATALOG_REVIEW": "REVIEW_REQUIRED",
+ "FLINK_GLUE_CATALOG_UNVERIFIED": "REVIEW_REQUIRED",
"FLINK_HUDI_METADATA_FIELD_CONFLICT": "BLOCKED",
"FLINK_LOAD_BEARING_VALUE_REQUIRED": "INCOMPLETE",
"FLINK_MULTI_WRITER_REVIEW": "REVIEW_REQUIRED",
@@ -114,6 +135,7 @@ FINDING_STATUS = {
"FLINK_MOR_COMPACTION_TUNING_UNSUPPORTED": "BLOCKED",
"FLINK_MOR_QUERY_REQUIREMENT_UNRESOLVED": "INCOMPLETE",
"FLINK_MOR_READ_OPTIMIZED_FRESHNESS_UNSUPPORTED": "BLOCKED",
+ "FLINK_MOR_CATALOG_NAMING_UNVERIFIED": "REVIEW_REQUIRED",
"FLINK_OPTION_NOT_VERIFIED": "REVIEW_REQUIRED",
"FLINK_PARTITION_FIELD_MISSING": "BLOCKED",
"FLINK_PHYSICAL_SCHEMA_REQUIRED": "INCOMPLETE",
@@ -189,6 +211,17 @@ def _check_keys(
)
+def _check_catalog_keys(
+ value: dict[str, Any], allowed: set[str], field: str, assessment: Assessment
+) -> None:
+ unknown = sorted(set(value) - allowed)
+ if unknown:
+ assessment.add(
+ "FLINK_CATALOG_CONTRACT_INVALID",
+ f"{field} contains unsupported fields: {unknown}",
+ )
+
+
def _list(value: Any, field: str, assessment: Assessment) -> list[Any]:
if isinstance(value, list):
return value
@@ -382,11 +415,638 @@ def _sql_string(value: str) -> str:
return "'" + value.replace("'", "''") + "'"
+def _has_sensitive_catalog_key(value: Any) -> bool:
+ if isinstance(value, dict):
+ return any(
+ SENSITIVE_CATALOG_KEY.search(str(key)) is not None
+ or _has_sensitive_catalog_key(item)
+ for key, item in value.items()
+ )
+ if isinstance(value, list):
+ return any(_has_sensitive_catalog_key(item) for item in value)
+ return False
+
+
+def _metastore_uris_valid(value: Any) -> bool:
+ if not _check_concrete_value(value):
+ return False
+ for raw_uri in value.split(","):
+ uri = raw_uri.strip()
+ try:
+ parsed = urlsplit(uri)
+ port = parsed.port
+ except ValueError:
+ return False
+ if (
+ parsed.scheme.lower() != "thrift"
+ or not parsed.hostname
+ or port is None
+ or not 1 <= port <= 65535
+ or parsed.username is not None
+ or parsed.password is not None
+ or parsed.path
+ or parsed.query
+ or parsed.fragment
+ ):
+ return False
+ return True
+
+
+def _metastore_uris_have_credentials(value: Any) -> bool:
+ if not isinstance(value, str):
+ return False
+ for raw_uri in value.split(","):
+ if SENSITIVE_URI_ASSIGNMENT.search(unquote(raw_uri)) is not None:
+ return True
+ try:
+ parsed = urlsplit(raw_uri.strip())
+ except ValueError:
+ continue
+ if parsed.username is not None or parsed.password is not None:
+ return True
+ parameters = [
+ *parse_qsl(parsed.query, keep_blank_values=True),
+ *parse_qsl(parsed.fragment, keep_blank_values=True),
+ ]
+ if any(_is_sensitive_query_key(key) for key, _ in parameters):
+ return True
+ return False
+
+
+def _catalog_profile_id(resolution_mode: Any, external_discovery: Any) -> str | None:
+ if not isinstance(resolution_mode, str) or not isinstance(external_discovery, str):
+ return None
+ profiles = {
+ ("direct_sql", "none"): "direct_sql_none",
+ ("direct_sql", "hive_metastore"): "direct_sql_hive_metastore",
+ ("hudi_dfs_catalog", "none"): "hudi_dfs_catalog_none",
+ ("hudi_dfs_catalog", "hive_metastore"):
+ "hudi_dfs_catalog_hive_metastore",
+ ("hudi_hms_catalog", "hive_metastore"):
+ "hudi_hms_catalog_hive_metastore",
+ }
+ return profiles.get((resolution_mode, external_discovery))
+
+
+def _mor_sync_strategy(contract: dict[str, Any]) -> str | None:
+ raw_read = contract.get("read")
+ read = raw_read if isinstance(raw_read, dict) else {}
+ raw_modes = read.get("query_modes")
+ modes = (
+ set(raw_modes)
+ if isinstance(raw_modes, list)
+ and all(isinstance(mode, str) for mode in raw_modes)
+ else set()
+ )
+ if modes == {"snapshot"}:
+ return "RT"
+ if modes == {"read_optimized"}:
+ return "RO"
+ if modes == {"snapshot", "read_optimized"}:
+ return "ALL"
+ return None
+
+
+def _registered_table_names(
+ table_name: Any, table_type: Any, strategy: str | None
+) -> list[str]:
+ if not _non_empty_string(table_name):
+ return []
+ base_name = table_name.split(".")[-1]
+ if table_type != "MERGE_ON_READ":
+ return [base_name]
+ if strategy == "RO":
+ return [base_name]
+ if strategy == "RT":
+ return [base_name]
+ if strategy == "ALL":
+ return [f"{base_name}_ro", f"{base_name}_rt", base_name]
+ return []
+
+
+def _assess_catalog(
+ contract: dict[str, Any],
+ manifest: dict[str, Any],
+ assessment: Assessment,
+ *,
+ external_catalog: Any,
+ expected_table_lifecycle: str,
+) -> dict[str, Any]:
+ """Validate the orthogonal PR6 catalog contract without selecting it."""
+
+ explicit = "catalog" in contract
+ raw_catalog = contract.get("catalog")
+ if not explicit:
+ if external_catalog is True:
+ assessment.add(
+ "FLINK_CATALOG_CONTRACT_REQUIRED",
+ "safety.external_catalog=true requires an explicit catalog contract",
+ )
+ legacy_profile = external_catalog is False
+ resolution = "direct_sql" if legacy_profile else "unresolved"
+ discovery = "none" if legacy_profile else "unresolved"
+ normalized_sync_owner = "none" if legacy_profile else "unresolved"
+ normalized_lifecycle_owner = "application" if legacy_profile else "unresolved"
+ return {
+ "contract_schema": CATALOG_CONTRACT_SCHEMA_VERSION,
+ "profile": "direct_sql_none" if legacy_profile else "unresolved",
+ "resolution_mode": resolution,
+ "external_discovery": discovery,
+ "sync_owner": normalized_sync_owner,
+ "data_lifecycle_owner": normalized_lifecycle_owner,
+ "catalog_setup_sql": "",
+ "effective_configuration": {
+ "catalog_freshness": "not_applicable",
+ "data_lifecycle_owner": normalized_lifecycle_owner,
+ "external_discovery": discovery,
+ "resolution_mode": resolution,
+ "sync_owner": normalized_sync_owner,
+ },
+ "deployment_checks": [],
+ }
+
+ if isinstance(raw_catalog, dict):
+ catalog = raw_catalog
+ else:
+ assessment.add(
+ "FLINK_CATALOG_CONTRACT_INVALID", "catalog must be an object"
+ )
+ catalog = {}
+ _check_catalog_keys(
+ catalog,
+ {
+ "catalog_name",
+ "catalog_path",
+ "contract_schema",
+ "data_lifecycle_owner",
+ "database",
+ "external_discovery",
+ "hive_configuration",
+ "mor_sync_strategy",
+ "resolution_mode",
+ "sync_owner",
+ "table_external",
+ },
+ "catalog",
+ assessment,
+ )
+ if _has_sensitive_catalog_key(catalog):
+ assessment.add(
+ "FLINK_CATALOG_CREDENTIAL_REJECTED",
+ "Catalog contracts must reference credential-free deployment configuration",
+ )
+ if catalog.get("contract_schema") != CATALOG_CONTRACT_SCHEMA_VERSION:
+ assessment.add(
+ "FLINK_CATALOG_CONTRACT_INVALID",
+ f"catalog.contract_schema must be {CATALOG_CONTRACT_SCHEMA_VERSION}",
+ )
+
+ resolution_mode = catalog.get("resolution_mode")
+ external_discovery = catalog.get("external_discovery")
+ sync_owner = catalog.get("sync_owner")
+ lifecycle_owner = catalog.get("data_lifecycle_owner")
+ resolution_modes = {
+ "direct_sql",
+ "hudi_dfs_catalog",
+ "hudi_hms_catalog",
+ }
+ discovery_modes = {"none", "hive_metastore", "aws_glue"}
+ sync_owners = {"none", "flink_sink", "hudi_hms_catalog"}
+ lifecycle_owners = {"application", "catalog"}
+ resolution_valid = isinstance(resolution_mode, str) and resolution_mode in resolution_modes
+ discovery_valid = (
+ isinstance(external_discovery, str)
+ and external_discovery in discovery_modes
+ )
+ sync_owner_valid = isinstance(sync_owner, str) and sync_owner in sync_owners
+ lifecycle_owner_valid = (
+ isinstance(lifecycle_owner, str) and lifecycle_owner in lifecycle_owners
+ )
+ catalog_managed_resolution = (
+ resolution_valid and resolution_mode in CATALOG_MANAGED_RESOLUTION_MODES
+ )
+ if not _non_empty_string(resolution_mode):
+ assessment.add(
+ "FLINK_CATALOG_CONTRACT_INVALID",
+ "catalog.resolution_mode must be explicit",
+ )
+ elif not resolution_valid:
+ assessment.add(
+ "FLINK_CATALOG_MODE_UNSUPPORTED",
+ "catalog.resolution_mode is outside the Hudi 1.2.0 PR6 profiles",
+ )
+ if not _non_empty_string(external_discovery):
+ assessment.add(
+ "FLINK_CATALOG_CONTRACT_INVALID",
+ "catalog.external_discovery must be explicit",
+ )
+ elif not discovery_valid:
+ assessment.add(
+ "FLINK_CATALOG_MODE_UNSUPPORTED",
+ "catalog.external_discovery is unsupported",
+ )
+ if not _non_empty_string(sync_owner):
+ assessment.add(
+ "FLINK_CATALOG_CONTRACT_INVALID",
+ "catalog.sync_owner must be explicit",
+ )
+ elif not sync_owner_valid:
+ assessment.add(
+ "FLINK_CATALOG_MODE_UNSUPPORTED", "catalog.sync_owner is unsupported"
+ )
+ if not _non_empty_string(lifecycle_owner):
+ assessment.add(
+ "FLINK_CATALOG_CONTRACT_INVALID",
+ "catalog.data_lifecycle_owner must be explicit",
+ )
+ elif not lifecycle_owner_valid:
+ assessment.add(
+ "FLINK_CATALOG_MODE_UNSUPPORTED",
+ "catalog.data_lifecycle_owner is unsupported",
+ )
+ if external_discovery == "aws_glue":
+ assessment.add(
+ "FLINK_GLUE_CATALOG_UNVERIFIED",
+ "AWS Glue is present in Hudi 1.2.0 but is not executable without a "
+ "Flink-specific fixture",
+ )
+
+ profile = _catalog_profile_id(resolution_mode, external_discovery)
+ if profile is None:
+ if external_discovery == "aws_glue":
+ profile = "aws_glue_unverified"
+ elif resolution_valid and discovery_valid:
+ assessment.add(
+ "FLINK_CATALOG_COMPOSITION_CONFLICT",
+ "The requested table-resolution and external-discovery composition is unsupported",
+ )
+ profile = "unsupported"
+ else:
+ profile = "unresolved"
+
+ expected_external_gate = (
+ resolution_mode != "direct_sql" or external_discovery != "none"
+ )
+ if (
+ resolution_valid
+ and discovery_valid
+ and isinstance(external_catalog, bool)
+ and external_catalog != expected_external_gate
+ ):
+ assessment.add(
+ "FLINK_CATALOG_COMPOSITION_CONFLICT",
+ "safety.external_catalog conflicts with the explicit catalog composition",
+ )
+
+ expected_owners = {
+ "direct_sql_none": ("none", "application"),
+ "direct_sql_hive_metastore": ("flink_sink", "application"),
+ "hudi_dfs_catalog_none": ("none", "catalog"),
+ "hudi_dfs_catalog_hive_metastore": ("flink_sink", "catalog"),
+ "hudi_hms_catalog_hive_metastore": ("hudi_hms_catalog", "application"),
+ }
+ if profile in expected_owners:
+ expected_sync_owner, expected_lifecycle_owner = expected_owners[profile]
+ if sync_owner_valid and sync_owner != expected_sync_owner:
+ assessment.add(
+ "FLINK_CATALOG_COMPOSITION_CONFLICT",
+ f"{profile} requires sync_owner={expected_sync_owner}",
+ )
+ if lifecycle_owner is None or (
+ isinstance(lifecycle_owner, str) and not lifecycle_owner.strip()
+ ):
+ assessment.add(
+ "FLINK_CATALOG_LIFECYCLE_UNRESOLVED",
+ f"{profile} requires an explicit data lifecycle owner",
+ )
+ elif lifecycle_owner_valid and lifecycle_owner != expected_lifecycle_owner:
+ assessment.add(
+ "FLINK_CATALOG_COMPOSITION_CONFLICT",
+ f"{profile} requires data_lifecycle_owner={expected_lifecycle_owner}",
+ )
+
+ catalog_name = catalog.get("catalog_name")
+ database = catalog.get("database")
+ catalog_path = catalog.get("catalog_path")
+ if resolution_mode == "direct_sql":
+ if "catalog_name" in catalog or "catalog_path" in catalog or "table_external" in catalog:
+ assessment.add(
+ "FLINK_CATALOG_CONTRACT_INVALID",
+ "Direct SQL forbids catalog_name, catalog_path, and table_external",
+ )
+ else:
+ if not _check_concrete_value(catalog_name):
+ assessment.add(
+ "FLINK_CATALOG_CONTRACT_INVALID",
+ "Catalog resolution requires a concrete catalog_name",
+ )
+ elif re.fullmatch(r"[A-Za-z_][A-Za-z0-9_]*", catalog_name) is None:
+ assessment.add(
+ "FLINK_CATALOG_CONTRACT_INVALID",
+ "catalog.catalog_name must be a simple Flink SQL identifier",
+ )
+ if resolution_mode != "direct_sql" or external_discovery == "hive_metastore":
+ if not _check_concrete_value(database):
+ assessment.add(
+ "FLINK_CATALOG_CONTRACT_INVALID",
+ "Catalog or Hive metastore composition requires a concrete database",
+ )
+ elif re.fullmatch(r"[A-Za-z_][A-Za-z0-9_]*", database) is None:
+ assessment.add(
+ "FLINK_CATALOG_CONTRACT_INVALID",
+ "catalog.database must be a simple Flink SQL identifier",
+ )
+ elif "database" in catalog:
+ assessment.add(
+ "FLINK_CATALOG_CONTRACT_INVALID",
+ "catalog.database is not applicable to direct SQL without external discovery",
+ )
+
+ raw_table = contract.get("table")
+ table: dict[str, Any] = raw_table if isinstance(raw_table, dict) else {}
+ table_name = table.get("name")
+ table_path = table.get("path")
+ if catalog_managed_resolution and (
+ not _non_empty_string(table_name) or "." in table_name
+ ):
+ assessment.add(
+ "FLINK_CATALOG_CONTRACT_INVALID",
+ "Catalog-managed table.name must be an unqualified identifier; catalog and "
+ "database are separate fields",
+ )
+ if resolution_mode == "hudi_dfs_catalog":
+ if not _check_concrete_value(catalog_path):
+ assessment.add(
+ "FLINK_CATALOG_CONTRACT_INVALID",
+ "DFS catalog resolution requires a concrete catalog_path",
+ )
+ elif path_error := _path_validation_error(catalog_path):
+ assessment.add("FLINK_CATALOG_CREDENTIAL_REJECTED", path_error)
+ if all(
+ _check_concrete_value(value)
+ for value in (catalog_path, database, table_name)
+ ):
+ expected_path = f"{catalog_path}/{database}/{table_name}"
+ if table_path != expected_path:
+ assessment.add(
+ "FLINK_CATALOG_PATH_CONFLICT",
+ "The desired table path must exactly equal "
+ "${catalog.path}/${database}/${table.name}",
+ )
+ if "table_external" in catalog:
+ assessment.add(
+ "FLINK_CATALOG_CONTRACT_INVALID",
+ "catalog.table_external applies only to hudi_hms_catalog",
+ )
+ elif "catalog_path" in catalog:
+ assessment.add(
+ "FLINK_CATALOG_CONTRACT_INVALID",
+ "catalog.catalog_path applies only to hudi_dfs_catalog",
+ )
+
+ if resolution_mode == "hudi_hms_catalog":
+ table_external = catalog.get("table_external")
+ if "table_external" not in catalog:
+ assessment.add(
+ "FLINK_CATALOG_LIFECYCLE_UNRESOLVED",
+ "The first HMS Catalog path requires table.external=true and "
+ "application-owned data",
+ )
+ elif not isinstance(table_external, bool):
+ assessment.add(
+ "FLINK_CATALOG_CONTRACT_INVALID",
+ "catalog.table_external must be explicitly true or false",
+ )
+ elif table_external is False:
+ assessment.add(
+ "FLINK_CATALOG_COMPOSITION_CONFLICT",
+ "The bounded HMS Catalog path does not support managed tables; "
+ "table.external must be true",
+ )
+ elif "table_external" in catalog:
+ assessment.add(
+ "FLINK_CATALOG_CONTRACT_INVALID",
+ "catalog.table_external applies only to hudi_hms_catalog",
+ )
+
+ raw_hive = catalog.get("hive_configuration")
+ hive_configuration: dict[str, Any] = {}
+ if external_discovery == "hive_metastore":
+ if isinstance(raw_hive, dict):
+ hive_configuration = raw_hive
+ else:
+ assessment.add(
+ "FLINK_CATALOG_CONTRACT_INVALID",
+ "catalog.hive_configuration must be an object",
+ )
+ _check_catalog_keys(
+ hive_configuration,
+ {"location", "source"},
+ "catalog.hive_configuration",
+ assessment,
+ )
+ source = hive_configuration.get("source")
+ location = hive_configuration.get("location")
+ if resolution_mode == "hudi_hms_catalog":
+ if source == "hive_conf_dir":
+ if not _check_concrete_value(location) or not Path(location).is_absolute():
+ assessment.add(
+ "FLINK_CATALOG_CONTRACT_INVALID",
+ "hive_conf_dir requires an absolute credential-free location",
+ )
+ elif _path_validation_error(location) is not None:
+ assessment.add(
+ "FLINK_CATALOG_CREDENTIAL_REJECTED",
+ "hive_conf_dir must not contain credentials or secret parameters",
+ )
+ elif source == "classpath":
+ if "location" in hive_configuration:
+ assessment.add(
+ "FLINK_CATALOG_CONTRACT_INVALID",
+ "Classpath Hive configuration must not include a location",
+ )
+ else:
+ assessment.add(
+ "FLINK_CATALOG_CONTRACT_INVALID",
+ "HMS Catalog supports hive_conf_dir or classpath Hive configuration",
+ )
+ else:
+ if _metastore_uris_have_credentials(location):
+ assessment.add(
+ "FLINK_CATALOG_CREDENTIAL_REJECTED",
+ "Hive metastore URIs must not contain credentials or secret parameters",
+ )
+ if source != "metastore_uris" or not _metastore_uris_valid(location):
+ assessment.add(
+ "FLINK_CATALOG_CONTRACT_INVALID",
+ "Direct and DFS sink-owned Hive sync requires credential-free thrift "
+ "metastore_uris",
+ )
+ elif "hive_configuration" in catalog:
+ assessment.add(
+ "FLINK_CATALOG_CONTRACT_INVALID",
+ "catalog.hive_configuration is forbidden when external_discovery=none",
+ )
+
+ mutable_mor = contract.get("contract_schema") == MUTABLE_MOR_CONTRACT_SCHEMA_VERSION
+ strategy = catalog.get("mor_sync_strategy")
+ expected_strategy = _mor_sync_strategy(contract) if mutable_mor else None
+ if mutable_mor and external_discovery == "hive_metastore":
+ if strategy != expected_strategy:
+ assessment.add(
+ "FLINK_MOR_CATALOG_NAMING_UNVERIFIED",
+ "MOR sync strategy must be derived from the declared query modes",
+ )
+ elif "mor_sync_strategy" in catalog:
+ assessment.add(
+ "FLINK_CATALOG_CONTRACT_INVALID",
+ "catalog.mor_sync_strategy applies only to mutable MOR with Hive discovery",
+ )
+
+ raw_write = contract.get("write")
+ write = raw_write if isinstance(raw_write, dict) else {}
+ raw_options = write.get("connector_options")
+ connector_options = raw_options if isinstance(raw_options, dict) else {}
+ if _has_sensitive_catalog_key(connector_options):
+ assessment.add(
+ "FLINK_CATALOG_CREDENTIAL_REJECTED",
+ "Connector options must not contain catalog credentials",
+ )
+ if (
+ connector_options.get("hive_sync.skip_ro_suffix") is True
+ or connector_options.get("hive_sync.skip_ro_suffix") in ("true", "TRUE")
+ or connector_options.get("hoodie.meta.sync.sync_snapshot_with_table_name")
+ is False
+ or connector_options.get("hoodie.meta.sync.sync_snapshot_with_table_name")
+ in ("false", "FALSE")
+ or (
+ "hive_sync.table.strategy" in connector_options
+ and connector_options.get("hive_sync.table.strategy") != expected_strategy
+ )
+ ):
+ assessment.add(
+ "FLINK_MOR_CATALOG_NAMING_UNVERIFIED",
+ "Custom MOR catalog naming is outside the fixed PR6 naming contract",
+ )
+
+ if expected_table_lifecycle == "existing" and profile != "direct_sql_none":
+ assessment.add(
+ "FLINK_EXISTING_CATALOG_MUTATION_UNVERIFIED",
+ "Existing-table registration, alteration, and catalog takeover remain assessment-only",
+ )
+
+ catalog_sql = ""
+ if catalog_managed_resolution:
+ options: list[tuple[str, Any]] = [("type", "hudi")]
+ if resolution_mode == "hudi_dfs_catalog":
+ options.extend([("mode", "dfs"), ("catalog.path", catalog_path)])
+ else:
+ options.extend([("mode", "hms"), ("table.external", "true")])
+ if hive_configuration.get("source") == "hive_conf_dir":
+ options.append(("hive.conf.dir", hive_configuration.get("location")))
+ options.append(("default-database", database))
+ if all(_check_concrete_value(item) for item in (catalog_name, database)) and all(
+ _check_concrete_value(value) for _, value in options
+ ):
+ option_sql = ",\n".join(
+ f" {_sql_string(key)} = {_sql_string(str(value))}"
+ for key, value in options
+ )
+ catalog_sql = (
+ f"CREATE CATALOG {_quote_identifier(str(catalog_name))} WITH "
+ f"(\n{option_sql}\n);\n"
+ f"USE CATALOG {_quote_identifier(str(catalog_name))};\n"
+ f"CREATE DATABASE IF NOT EXISTS {_quote_identifier(str(database))};\n"
+ f"USE {_quote_identifier(str(database))};"
+ )
+
+ table_type = write.get("table_type")
+ registered_names = _registered_table_names(table_name, table_type, strategy)
+ effective_configuration: dict[str, Any] = {
+ "catalog_freshness": (
+ "post_commit_eventual"
+ if external_discovery == "hive_metastore"
+ else "not_applicable"
+ ),
+ "data_lifecycle_owner": lifecycle_owner,
+ "external_discovery": external_discovery,
+ "resolution_mode": resolution_mode,
+ "sync_owner": sync_owner,
+ }
+ if external_discovery == "hive_metastore":
+ effective_configuration["hive_sync"] = {
+ "database": database,
+ "enabled": True,
+ "ignore_exceptions": False,
+ "metastore_uris": (
+ hive_configuration.get("location")
+ if hive_configuration.get("source") == "metastore_uris"
+ and _metastore_uris_valid(hive_configuration.get("location"))
+ else (
+ "unavailable"
+ if hive_configuration.get("source") == "metastore_uris"
+ else "from_hive_configuration"
+ )
+ ),
+ "mode": "hms",
+ "partition_fields": table.get("partition_fields", []),
+ "registered_table_names": registered_names,
+ "skip_ro_suffix": False,
+ "snapshot_with_table_name": True,
+ "support_timestamp": True,
+ "table": table_name.split(".")[-1]
+ if _non_empty_string(table_name)
+ else table_name,
+ "table_strategy": strategy if mutable_mor else "NOT_APPLICABLE",
+ }
+
+ required_checks: set[str] = set()
+ if catalog_managed_resolution:
+ required_checks.update(
+ {"verify_catalog_dependencies", "verify_catalog_database_registration"}
+ )
+ if resolution_mode == "hudi_dfs_catalog":
+ required_checks.update(
+ {"verify_catalog_path_exists", "verify_catalog_path_permissions"}
+ )
+ if external_discovery == "hive_metastore":
+ required_checks.update(
+ {
+ "verify_hive_configuration_available",
+ "verify_hms_connectivity",
+ "verify_table_registration_after_commit",
+ }
+ )
+ if table.get("partition_fields"):
+ required_checks.add("verify_partition_registration_after_commit")
+ if mutable_mor:
+ required_checks.add("verify_mor_catalog_table_names")
+ required_checks.add("verify_external_engine_visibility")
+ checks = [
+ check
+ for check in manifest["catalog_composition_contract"]["deployment_checks"]
+ if check in required_checks
+ ]
+
+ return {
+ "contract_schema": CATALOG_CONTRACT_SCHEMA_VERSION,
+ "profile": profile,
+ "resolution_mode": resolution_mode,
+ "external_discovery": external_discovery,
+ "sync_owner": sync_owner,
+ "data_lifecycle_owner": lifecycle_owner,
+ "catalog_setup_sql": catalog_sql,
+ "effective_configuration": effective_configuration,
+ "deployment_checks": checks,
+ }
+
+
def _render_sql(
contract: dict[str, Any],
table_columns: list[dict[str, Any]],
partition_fields: list[str],
record_key_fields: list[str],
+ catalog_state: dict[str, Any],
) -> dict[str, str]:
table = contract["table"]
source = contract["source"]
@@ -397,6 +1057,17 @@ def _render_sql(
MUTABLE_MOR_CONTRACT_SCHEMA_VERSION,
}
mutable_mor = contract["contract_schema"] == MUTABLE_MOR_CONTRACT_SCHEMA_VERSION
+ resolution_mode = catalog_state["resolution_mode"]
+ external_discovery = catalog_state["external_discovery"]
+ catalog_managed_resolution = (
+ isinstance(resolution_mode, str)
+ and resolution_mode in CATALOG_MANAGED_RESOLUTION_MODES
+ )
+ target_identifier = (
+ _quote_identifier(table["name"])
+ if catalog_managed_resolution
+ else _quote_qualified_identifier(table["name"])
+ )
definitions = [
f" {_quote_identifier(column['name'])} {column['type']}"
@@ -407,56 +1078,74 @@ def _render_sql(
key_sql = ", ".join(_quote_identifier(field) for field in record_key_fields)
definitions.append(f" PRIMARY KEY ({key_sql}) NOT ENFORCED")
- table_ddl = (
- f"CREATE TABLE {_quote_qualified_identifier(table['name'])} (\n"
- + ",\n".join(definitions)
- + "\n)"
- )
+ table_ddl = f"CREATE TABLE {target_identifier} (\n" + ",\n".join(definitions) + "\n)"
if partition_fields:
partition_sql = ", ".join(_quote_identifier(field) for field in partition_fields)
table_ddl += f"\nPARTITIONED BY ({partition_sql})"
+ table_options: list[tuple[str, str]] = [("connector", "hudi")]
+ if resolution_mode != "hudi_dfs_catalog":
+ table_options.append(("path", table["path"]))
if mutable:
ordering_field = contract["ordering"]["fields"][0]
table_type = "MERGE_ON_READ" if mutable_mor else "COPY_ON_WRITE"
- table_ddl += (
- "\nWITH (\n"
- " 'connector' = 'hudi',\n"
- f" 'path' = {_sql_string(table['path'])},\n"
- f" 'table.type' = '{table_type}',\n"
- " 'write.operation' = 'upsert',\n"
- f" 'ordering.fields' = {_sql_string(ordering_field)},\n"
- " 'hoodie.write.record.merge.mode' = 'EVENT_TIME_ORDERING',\n"
- " 'index.type' = 'FLINK_STATE',\n"
- " 'index.global.enabled' = 'true',\n"
- " 'index.state.ttl' = '0',\n"
- " 'index.bootstrap.enabled' = 'true',\n"
- " 'changelog.enabled' = 'false'"
+ table_options.extend(
+ [
+ ("table.type", table_type),
+ ("write.operation", "upsert"),
+ ("ordering.fields", ordering_field),
+ ("hoodie.write.record.merge.mode", "EVENT_TIME_ORDERING"),
+ ("index.type", "FLINK_STATE"),
+ ("index.global.enabled", "true"),
+ ("index.state.ttl", "0"),
+ ("index.bootstrap.enabled", "true"),
+ ("changelog.enabled", "false"),
+ ]
)
if mutable_mor:
- table_ddl += (
- ",\n 'compaction.schedule.enabled' = 'true',\n"
- " 'compaction.async.enabled' = 'true',\n"
- " 'compaction.operation.execute.async.enabled' = 'true',\n"
- " 'compaction.trigger.strategy' = 'num_commits',\n"
- " 'compaction.delta_commits' = '5',\n"
- " 'clustering.schedule.enabled' = 'false',\n"
- " 'clustering.async.enabled' = 'false'"
- )
- table_ddl += "\n);"
+ table_options.extend(
+ [
+ ("compaction.schedule.enabled", "true"),
+ ("compaction.async.enabled", "true"),
+ ("compaction.operation.execute.async.enabled", "true"),
+ ("compaction.trigger.strategy", "num_commits"),
+ ("compaction.delta_commits", "5"),
+ ("clustering.schedule.enabled", "false"),
+ ("clustering.async.enabled", "false"),
+ ]
+ )
else:
- table_ddl += (
- "\nWITH (\n"
- " 'connector' = 'hudi',\n"
- f" 'path' = {_sql_string(table['path'])},\n"
- " 'table.type' = 'COPY_ON_WRITE',\n"
- " 'write.operation' = 'insert',\n"
- " 'write.insert.cluster' = 'false'\n"
- ");"
+ table_options.extend(
+ [
+ ("table.type", "COPY_ON_WRITE"),
+ ("write.operation", "insert"),
+ ("write.insert.cluster", "false"),
+ ]
)
+ catalog = contract.get("catalog", {})
+ if external_discovery == "hive_metastore":
+ if catalog_state["sync_owner"] == "flink_sink":
+ hive_configuration = catalog["hive_configuration"]
+ table_options.extend(
+ [
+ ("hive_sync.enabled", "true"),
+ ("hive_sync.mode", "hms"),
+ ("hive_sync.metastore.uris", hive_configuration["location"]),
+ ("hive_sync.db", catalog["database"]),
+ ("hive_sync.table", table["name"].split(".")[-1]),
+ ]
+ )
+ if mutable_mor:
+ table_options.append(("hive_sync.table.strategy", catalog["mor_sync_strategy"]))
+
+ option_sql = ",\n".join(
+ f" {_sql_string(key)} = {_sql_string(value)}" for key, value in table_options
+ )
+ table_ddl += f"\nWITH (\n{option_sql}\n);"
+
field_sql = ",\n ".join(_quote_identifier(column["name"]) for column in table_columns)
insert_sql = (
- f"INSERT INTO {_quote_qualified_identifier(table['name'])} (\n {field_sql}\n)\n"
+ f"INSERT INTO {target_identifier} (\n {field_sql}\n)\n"
f"SELECT\n {field_sql}\n"
f"FROM {_quote_qualified_identifier(source['table'])};"
)
@@ -465,11 +1154,17 @@ def _render_sql(
"SET 'execution.checkpointing.interval' = "
f"'{runtime['checkpoint_interval_ms']} ms';"
)
+ catalog_setup_sql = catalog_state["catalog_setup_sql"]
+ statements = [runtime_sql]
+ if catalog_setup_sql:
+ statements.append(catalog_setup_sql)
+ statements.extend([table_ddl, insert_sql])
return {
"runtime_sql": runtime_sql,
+ "catalog_setup_sql": catalog_setup_sql,
"table_ddl": table_ddl,
"insert_sql": insert_sql,
- "combined_sql": f"{runtime_sql}\n\n{table_ddl}\n\n{insert_sql}",
+ "combined_sql": "\n\n".join(statements),
}
@@ -512,6 +1207,7 @@ def assess_design(
contract,
{
"baseline_id",
+ "catalog",
"contract_schema",
"flink_version",
"hudi_version",
@@ -615,10 +1311,15 @@ def assess_design(
elif writer_model != "single_writer":
assessment.add("FLINK_WRITER_MODEL_UNRESOLVED", "Writer topology is unresolved")
external_catalog = safety.get("external_catalog")
- if external_catalog is True:
- assessment.add("FLINK_EXTERNAL_CATALOG_REVIEW", "Catalog composition is deferred")
- elif external_catalog is not False:
+ if type(external_catalog) is not bool:
assessment.add("FLINK_CATALOG_REQUIREMENT_UNRESOLVED", "Catalog requirement is unresolved")
+ catalog_state = _assess_catalog(
+ contract,
+ manifest,
+ assessment,
+ external_catalog=external_catalog,
+ expected_table_lifecycle=expected_table_lifecycle,
+ )
mutability = safety.get("mutability")
if mutable_contract:
if mutability != "mutable":
@@ -693,6 +1394,23 @@ def assess_design(
f"Partition field {field!r} cannot use binary type "
f"{table_by_name[field]['type']!r} until deterministic encoding is verified",
)
+ if (
+ catalog_state["resolution_mode"] == "hudi_hms_catalog"
+ and partition_fields
+ and len(partition_fields) == len(set(partition_fields))
+ and all(field in table_by_name for field in partition_fields)
+ ):
+ schema_partition_order = [
+ column["name"]
+ for column in table_columns
+ if column["name"] in partition_fields
+ ]
+ if schema_partition_order != partition_fields:
+ assessment.add(
+ "FLINK_CATALOG_COMPOSITION_CONFLICT",
+ "HMS Catalog requires table.partition_fields to follow their order in "
+ "table.columns",
+ )
identity = _dict(contract.get("identity"), "identity", assessment)
_check_keys(
@@ -861,10 +1579,36 @@ def assess_design(
source_keys.update({"delete_payload", "emits_deletes", "primary_key_fields"})
_check_keys(source, source_keys, "source", assessment)
source_name = source.get("table")
- if not _check_concrete_value(source_name) or any(
- not part for part in str(source_name).split(".")
- ):
+ source_parts = source_name.split(".") if _check_concrete_value(source_name) else []
+ source_catalog_managed = (
+ isinstance(catalog_state["resolution_mode"], str)
+ and catalog_state["resolution_mode"] in CATALOG_MANAGED_RESOLUTION_MODES
+ )
+ if not source_parts or any(not part for part in source_parts):
assessment.add("FLINK_SOURCE_CONTRACT_REQUIRED", "A concrete source table is required")
+ elif source_catalog_managed and len(source_parts) != 3:
+ assessment.add(
+ "FLINK_SOURCE_CONTRACT_REQUIRED",
+ "Catalog-managed profiles require source.table to be a fully qualified "
+ "catalog.database.table identifier",
+ )
+ elif expected_table_lifecycle == "new" and source_catalog_managed:
+ raw_catalog = contract.get("catalog")
+ catalog = raw_catalog if isinstance(raw_catalog, dict) else {}
+ target_parts = [
+ catalog.get("catalog_name"),
+ catalog.get("database"),
+ table_name,
+ ]
+ if (
+ all(_check_concrete_value(part) for part in target_parts)
+ and source_parts == target_parts
+ ):
+ assessment.add(
+ "FLINK_CATALOG_COMPOSITION_CONFLICT",
+ "The streaming new-table path cannot read from and write to the same "
+ "catalog.database.table object",
+ )
source_columns = _columns(source.get("columns"), "source.columns", manifest, assessment)
if not source_columns:
assessment.add(
@@ -1304,14 +2048,27 @@ def assess_design(
result: dict[str, Any] = {
"status": status,
"executable_eligible": executable,
+ "catalog_contract_schema": catalog_state["contract_schema"],
+ "catalog_profile": catalog_state["profile"],
+ "catalog_setup_sql": catalog_state["catalog_setup_sql"] if executable else "",
+ "effective_catalog_configuration": catalog_state[
+ "effective_configuration"
+ ],
+ "catalog_deployment_checks": catalog_state["deployment_checks"],
"finding_codes": [finding["code"] for finding in assessment.findings],
"findings": assessment.findings,
"advisory_codes": assessment.advisories,
- "validation_evidence": validation_evidence(manifest, contract_schema),
+ "validation_evidence": validation_evidence(
+ manifest, contract_schema, catalog_state["profile"]
+ ),
}
if executable:
result["artifacts"] = _render_sql(
- contract, table_columns, partition_fields, record_key_fields
+ contract,
+ table_columns,
+ partition_fields,
+ record_key_fields,
+ catalog_state,
)
return result
diff --git a/hudi-ai-operator/skills/hudi-architect/validate_flink_existing_table.py b/hudi-ai-operator/skills/hudi-architect/validate_flink_existing_table.py
index f8a2069f96ae..5be76ab0eb67 100644
--- a/hudi-ai-operator/skills/hudi-architect/validate_flink_existing_table.py
+++ b/hudi-ai-operator/skills/hudi-architect/validate_flink_existing_table.py
@@ -1081,7 +1081,11 @@ def assess_reconciliation(
deployment_checks.append(
{"check": "verify_index_bootstrap_before_first_checkpoint", "required": True}
)
- baseline_evidence = validation_evidence(manifest, desired.get("contract_schema"))
+ baseline_evidence = validation_evidence(
+ manifest,
+ desired.get("contract_schema"),
+ desired_result.get("catalog_profile", "direct_sql_none"),
+ )
baseline_evidence["reconciliation_schema"] = RECONCILIATION_SCHEMA_VERSION
return {
"status": status,
@@ -1089,6 +1093,15 @@ def assess_reconciliation(
"reconciliation_complete": complete,
"writer_transition": transition,
"executable_eligible": False,
+ "catalog_contract_schema": desired_result.get("catalog_contract_schema"),
+ "catalog_profile": desired_result.get("catalog_profile", "direct_sql_none"),
+ "catalog_setup_sql": "",
+ "effective_catalog_configuration": desired_result.get(
+ "effective_catalog_configuration", {}
+ ),
+ "catalog_deployment_checks": desired_result.get(
+ "catalog_deployment_checks", []
+ ),
"finding_codes": finding_codes,
"findings": assessment.findings,
"advisory_codes": assessment.advisories,