danny0405 commented on code in PR #20514: URL: https://github.com/apache/hudi/pull/20514#discussion_r4466575109
########## hudi-agent-gateway/tests/test_hudi_architect_cross_path.py: ########## @@ -0,0 +1,423 @@ +# 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. + +"""Cross-path invariants for the completed Hudi Architect Flink milestone.""" + +from __future__ import annotations + +import copy +import json +import subprocess +import sys +import tomllib +from pathlib import Path +from typing import Any + +import pytest + +GATEWAY_DIR = Path(__file__).resolve().parents[1] +REPO_ROOT = GATEWAY_DIR.parent +FIXTURE_DIR = GATEWAY_DIR / "tests" / "fixtures" / "hudi_architect" +PR6_PROFILES = FIXTURE_DIR / "flink_pr6" / "catalog_profiles.toml" +PR8_FIXTURE_DIR = FIXTURE_DIR / "flink_pr8" +SCENARIO_MATRIX = PR8_FIXTURE_DIR / "cross_path_scenarios.toml" +DESIGN_VALIDATOR = ( + REPO_ROOT / "hudi-ai-operator" / "skills" / "hudi-architect" + / "validate_flink_design.py" +) +EXISTING_TABLE_VALIDATOR = DESIGN_VALIDATOR.with_name( + "validate_flink_existing_table.py" +) +STATUS_PRIORITY = {"INCOMPLETE": 0, "REVIEW_REQUIRED": 1, "BLOCKED": 2} + + +def _read_json(path: Path) -> dict[str, Any]: + value = json.loads(path.read_text(encoding="utf-8")) + assert isinstance(value, dict) + return value + + +def _load_matrix() -> dict[str, Any]: + with SCENARIO_MATRIX.open("rb") as matrix_file: + return tomllib.load(matrix_file) + + +def _scenarios() -> list[dict[str, Any]]: + return _load_matrix()["scenario"] + + +def _scenario(case_id: str) -> dict[str, Any]: + return next(item for item in _scenarios() if item["case_id"] == case_id) + + +def _scenario_ids() -> list[str]: + return [scenario["case_id"] for scenario in _scenarios()] + + +def _write_input(tmp_path: Path, case_id: str, value: dict[str, Any]) -> Path: + input_path = tmp_path / f"{case_id.lower()}.json" + input_path.write_text(json.dumps(value), encoding="utf-8") + return input_path + + +def _catalog_profile(profile_id: str) -> dict[str, Any]: + with PR6_PROFILES.open("rb") as profile_file: + profiles = tomllib.load(profile_file)["profile"] + profile = next(item for item in profiles if item["id"] == profile_id) + design = _read_json(FIXTURE_DIR / profile["base_fixture"]) + 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 _direct_sql_hms(design: dict[str, Any]) -> dict[str, Any]: + result = copy.deepcopy(design) + result["safety"]["external_catalog"] = True + catalog: dict[str, Any] = { + "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", + }, + } + if result["write"]["table_type"] == "MERGE_ON_READ": + catalog["mor_sync_strategy"] = "ALL" + result["catalog"] = catalog + return result + + +def _multi_writer( + design: dict[str, Any], concurrency_mode: str +) -> dict[str, Any]: + result = copy.deepcopy(design) + result["safety"]["writer_model"] = "multi_writer" + result["writer_contract"] = { + "contract_schema": 1, + "primary_writer_id": "flink-orders-stream", + "additional_writers": [ + { + "writer_id": "spark-orders-backfill", + "role": "BACKFILL", + "relationship_to_primary": "INDEPENDENT_CONCURRENT", + "engine": "SPARK", + "engine_version": "3.5", + "hudi_version": "1.2.0", + "operation": "BULK_INSERT", + } + ], + "ownership": { + "schema_evolution_owner_id": "flink-orders-stream", + "table_configuration_owner_id": "flink-orders-stream", + }, + "observed_concurrency": { + "mode": concurrency_mode, + "lock_provider_class": ( + "org.apache.hudi.client.transaction.lock." + "FileSystemBasedLockProvider" + ), + "lock_configuration_consistency": "CONSISTENT", + }, + } + return result + + +def _scenario_input(case_id: str) -> tuple[str, dict[str, Any]]: + source = _scenario(case_id)["input"] + if source == "flink_pr2/stable_key.json": + return "design", _read_json(FIXTURE_DIR / "flink_pr2" / "stable_key.json") + if source == "catalog_profile:hudi_dfs_catalog_hive_metastore": + return "design", _catalog_profile("hudi_dfs_catalog_hive_metastore") + if source == "flink_pr4/mutable_mor.json+direct_sql_hive_metastore": + design = _read_json(FIXTURE_DIR / "flink_pr4" / "mutable_mor.json") + return "design", _direct_sql_hms(design) + if source == "flink_pr5/compatible_mutable_cow.json": + return "existing_table", _read_json( + FIXTURE_DIR / "flink_pr5" / "compatible_mutable_cow.json" + ) + if source == "flink_pr5/compatible_mutable_cow.json+direct_sql_hive_metastore": + envelope = _read_json( + FIXTURE_DIR / "flink_pr5" / "compatible_mutable_cow.json" + ) + envelope["desired_contract"] = _direct_sql_hms( + envelope["desired_contract"] + ) + return "existing_table", envelope + if source == "flink_pr3/mutable_cow.json+multi_writer_occ": + design = _read_json(FIXTURE_DIR / "flink_pr3" / "mutable_cow.json") + return "design", _multi_writer( + design, "OPTIMISTIC_CONCURRENCY_CONTROL" + ) + if source == "flink_pr4/mutable_mor.json+multi_writer_nbcc": + design = _read_json(FIXTURE_DIR / "flink_pr4" / "mutable_mor.json") + return "design", _multi_writer( + design, "NON_BLOCKING_CONCURRENCY_CONTROL" + ) + if source == ( + "flink_pr5/compatible_mutable_cow.json+" + "flink_pr8/existing_multi_writer_out_of_scope.json" + ): + envelope = _read_json( + FIXTURE_DIR / "flink_pr5" / "compatible_mutable_cow.json" + ) + envelope["evidence"].append( + _read_json(PR8_FIXTURE_DIR / "existing_multi_writer_out_of_scope.json") + ) + return "existing_table", envelope + if source == "flink_pr8/combined_safety_gates.json": + return "design", _read_json(PR8_FIXTURE_DIR / "combined_safety_gates.json") + if source == "flink_pr8/malformed_cross_path.json": + return "existing_table", _read_json( + PR8_FIXTURE_DIR / "malformed_cross_path.json" + ) + raise AssertionError(f"Unknown PR8 scenario input: {source}") + + +def _run_scenario( + tmp_path: Path, case_id: str +) -> tuple[subprocess.CompletedProcess[str], dict[str, Any], dict[str, Any]]: + validator_name, input_value = _scenario_input(case_id) + validator = ( + DESIGN_VALIDATOR + if validator_name == "design" + else EXISTING_TABLE_VALIDATOR + ) + result = subprocess.run( + [ + sys.executable, + str(validator), + "--input", + str(_write_input(tmp_path, case_id, input_value)), Review Comment: Non-blocking coverage suggestion: could we also exercise `--emit-sql` for the design scenarios? This helper always takes the JSON-output branch, so the matrix would still pass if the separate SQL-output branch in `validate_flink_design.main()` regressed and printed SQL for an ineligible composition. For X06/X07/X09, assert exit code 1, empty stdout, and the expected assessment on stderr; for X01/X02/X03, assert stdout matches `artifacts.combined_sql` (allowing the trailing newline). I checked the six scenarios manually and they currently behave correctly; this would preserve that boundary in CI. -- 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]
