fhan688 commented on code in PR #20419:
URL: https://github.com/apache/hudi/pull/20419#discussion_r4387181543
##########
hudi-ai-operator/skills/hudi-architect/validate_flink_design.py:
##########
@@ -382,11 +408,620 @@ 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)
+ except ValueError:
+ return False
+ if (
+ parsed.scheme.lower() != "thrift"
+ or not parsed.hostname
+ or parsed.username is not None
+ or parsed.password is not None
+ 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(","):
+ 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
+ )
+ 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 not _non_empty_string(lifecycle_owner):
+ assessment.add(
+ "FLINK_CATALOG_LIFECYCLE_UNRESOLVED",
+ f"{profile} requires an explicit data lifecycle owner",
+ )
+ elif lifecycle_owner_valid and lifecycle_owner !=
expected_lifecycle_owner:
+ code = (
+ "FLINK_CATALOG_LIFECYCLE_UNRESOLVED"
+ if resolution_mode == "hudi_dfs_catalog"
+ else "FLINK_CATALOG_COMPOSITION_CONFLICT"
+ )
+ assessment.add(
+ code,
+ 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 resolution_mode in {"hudi_dfs_catalog", "hudi_hms_catalog"} 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":
+ if catalog.get("table_external") is not True:
+ assessment.add(
+ "FLINK_CATALOG_LIFECYCLE_UNRESOLVED",
+ "The first HMS Catalog path requires table.external=true and "
+ "application-owned data",
+ )
+ 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 resolution_mode in {"hudi_dfs_catalog", "hudi_hms_catalog"}:
+ 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))};"
Review Comment:
Thanks for the guidance. Catalog-managed profiles now require a fully
qualified `catalog.database.table` source, and the Flink fixture registers the
source before switching to the target catalog so that source resolution remains
correct.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
