fhan688 commented on code in PR #20419:
URL: https://github.com/apache/hudi/pull/20419#discussion_r4387187159
##########
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",
+ )
Review Comment:
Thanks for pointing this out. I added validation to require HMS partition
fields to follow their physical schema order. An inconsistent order is now
blocked with `FLINK_CATALOG_COMPOSITION_CONFLICT`, with regression coverage
added.
--
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]
