fhan688 commented on code in PR #20514: URL: https://github.com/apache/hudi/pull/20514#discussion_r4474877421
########## 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: Thanks for the suggestion. I added `--emit-sql` coverage for all six design scenarios, verifying the emitted SQL for eligible paths and enforcing empty stdout with the expected assessment on stderr for ineligible paths. The focused suite now passes with 33 tests. -- 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]
