From 1eea189ba0825ec8a1f899021f35f36ede131312 Mon Sep 17 00:00:00 2001 From: Dan Wolfson Date: Sat, 7 Nov 2026 23:01:53 +0000 Subject: [PATCH] fix(dr-egeria): cache Egeria 404-002 ambiguous lookups; skip display-name pre-check for action commands Step 4a's display-name pre-check only recognised "Multiple elements found" as ambiguous. Egeria's OMAG-GENERIC-HANDLERS-404-002 ("Multiple ... entities where found") was never cached, so every Create Subscription Type repeated a ~35 s lookup (~30 min per Data Hub document). - resolve_element_guid: treat 404-002 / "entities where found" as ambiguous and cache it once per batch (silently, preserving the previous non-fatal behaviour). - Step 4a is skipped when supports_target_element_lookup() is False (action commands never upsert by display name). Signed-off-by: Dan Wolfson Co-Authored-By: Claude Sonnet 5.5 Signed-off-by: Dan Wolfson --- md_processing/v2/processors.py | 31 ++++++- .../test_dre_precheck_ambiguous.py | 85 +++++++++++++++++++ 2 files changed, 115 insertions(+), 1 deletion(-) create mode 100644 tests/micro-tests/test_dre_precheck_ambiguous.py diff --git a/md_processing/v2/processors.py b/md_processing/v2/processors.py index 239cff00..786123a3 100644 --- a/md_processing/v2/processors.py +++ b/md_processing/v2/processors.py @@ -27,6 +27,16 @@ # up, ambiguous" (_AMBIGUOUS) are never confused with each other. _NOT_CACHED = object() _AMBIGUOUS = object() +# Egeria's own OMAG-GENERIC-HANDLERS-404-002 "Multiple entities where +# found" answer: ambiguous, cached once per batch like _AMBIGUOUS, but replayed +# silently (no error on the command) -- it has always been non-fatal, and a +# display name shared by many elements is legitimate for most Creates. +_AMBIGUOUS_SILENT = object() + + +def _is_egeria_multiple_entities_error(exc: BaseException) -> bool: + text = str(exc) + return "404-002" in text or "entities where found" in text class AsyncBaseCommandProcessor(ABC): @@ -699,8 +709,13 @@ async def _execute(self) -> Dict[str, Any]: # so there's nothing for this ambiguity check to catch). Skipping it # avoids a redundant, expensive ambiguous-Display-Name lookup on # every Create that already fully disambiguates its target. + # Action-style commands (supports_target_element_lookup() False, e.g. + # Create Subscription Type) never upsert by Display Name, so the + # check is pointless for them -- and costs ~1s per candidate when the + # name is shared. if ( - not self.as_is_element + self.supports_target_element_lookup() + and not self.as_is_element and not self._user_supplied_qualified_name and self.command.verb in ["Create", "Define", "Register", "Add", "Upsert"] ): @@ -1525,6 +1540,9 @@ def _cache_if_not_found(key: tuple, value: Any) -> None: logger.debug(f"resolve_element_guid: cache hit (ambiguous, Pass 1) for {pass1_key!r}") _report_ambiguous("Pass 1, cached") return None + elif cached1 is _AMBIGUOUS_SILENT: + logger.debug(f"resolve_element_guid: cache hit (ambiguous 404-002, Pass 1) for {pass1_key!r}") + return None elif cached1 is not _NOT_CACHED: logger.debug(f"resolve_element_guid: cache hit for {pass1_key!r} -> {cached1!r}") res = cached1 @@ -1538,6 +1556,10 @@ def _cache_if_not_found(key: tuple, value: Any) -> None: query_cache[pass1_key] = _AMBIGUOUS _report_ambiguous("Pass 1") return None + if _is_egeria_multiple_entities_error(e): + query_cache[pass1_key] = _AMBIGUOUS_SILENT + logger.debug(f"resolve_element_guid: '{name_or_guid}' is ambiguous (Pass 1): {e}") + return None if tech_type and self._is_unsupported_type_lookup_error(e): error_id = self._extract_egeria_error_id(e) or "unknown-error-id" if tech_type not in unsupported_type_warnings: @@ -1564,6 +1586,9 @@ def _cache_if_not_found(key: tuple, value: Any) -> None: logger.debug(f"resolve_element_guid: cache hit (ambiguous, Pass 2) for {pass2_key!r}") _report_ambiguous("Pass 2, cached") return None + elif cached2 is _AMBIGUOUS_SILENT: + logger.debug(f"resolve_element_guid: cache hit (ambiguous 404-002, Pass 2) for {pass2_key!r}") + return None elif cached2 is not _NOT_CACHED: logger.debug(f"resolve_element_guid: cache hit for {pass2_key!r} -> {cached2!r}") res = cached2 @@ -1576,6 +1601,10 @@ def _cache_if_not_found(key: tuple, value: Any) -> None: query_cache[pass2_key] = _AMBIGUOUS _report_ambiguous("Pass 2") return None + if _is_egeria_multiple_entities_error(e): + query_cache[pass2_key] = _AMBIGUOUS_SILENT + logger.debug(f"resolve_element_guid: '{name_or_guid}' is ambiguous (Pass 2): {e}") + return None logger.debug(f"SDK strict lookup (Pass 2) failed for '{name_or_guid}': {e}") if self.context.get("directive") == "validate": print_basic_exception(e) diff --git a/tests/micro-tests/test_dre_precheck_ambiguous.py b/tests/micro-tests/test_dre_precheck_ambiguous.py new file mode 100644 index 00000000..b70fa6a4 --- /dev/null +++ b/tests/micro-tests/test_dre_precheck_ambiguous.py @@ -0,0 +1,85 @@ +""" +Display-name pre-check (processors.py step 4a) performance fixes: + +1. Egeria's OMAG-GENERIC-HANDLERS-404-002 "Multiple ... entities where found" + answer is classified as ambiguous and cached once per batch (previously only + the SDK's "Multiple elements found" wording was, so every command repeated + the ~1s/candidate lookup). +2. Action-style commands (supports_target_element_lookup() False, e.g. Create + Subscription Type) skip the display-name pre-check entirely. +""" +from typing import Any, Dict, List + +import pytest + +from pyegeria import NO_ELEMENTS_FOUND +from pyegeria.core._exceptions import PyegeriaException + +from md_processing.v2.extraction import DrECommand +from md_processing.v2.processors import AsyncBaseCommandProcessor +from md_processing.v2.dispatcher import V2Dispatcher +from md_processing.md_processing_utils.md_processing_constants import load_commands + +_SHARED = "Weekly refresh subscription" +_MSG = ("OMAG-GENERIC-HANDLERS-404-002 Multiple OpenMetadataRoot entities where found " + "for name Weekly refresh subscription") + + +class _Client: + def __init__(self): + self.calls: List[str] = [] + + async def __async_get_guid__(self, qualified_name=None, display_name=None, property_name="qualifiedName", + tech_type=None, **kwargs): + name = display_name or qualified_name + if name == _SHARED: + self.calls.append(name) + raise PyegeriaException(_MSG) + return NO_ELEMENTS_FOUND + + +class _Create(AsyncBaseCommandProcessor): + async def apply_changes(self) -> str: + qn = self.parsed_output["qualified_name"] + self.parsed_output["guid"] = f"guid::{qn}" + return f"created {qn}" + + +class _ActionCreate(_Create): + def supports_target_element_lookup(self) -> bool: + return False + + async def fetch_as_is(self): + return None + + +def _cmds(n): + return [DrECommand(verb="Create", object_type="Project", + attributes={"Display Name": _SHARED}, + raw_block=f"## Create Project\n### Display Name\n{_SHARED}\n### Description\nd{i}\n") + for i in range(n)] + + +async def _run(proc_cls): + load_commands() + client = _Client() + d = V2Dispatcher(client) + d.register("Create Project", proc_cls) + results = await d.dispatch_batch(_cmds(4), {"directive": "process"}) + return client, results + + +@pytest.mark.asyncio +async def test_404_002_ambiguous_is_cached_once_per_batch(): + client, results = await _run(_Create) + assert all(r["status"] == "success" for r in results) + # Step 4a probes the shared name; typed pass + untyped pass only for the + # first command (two distinct queries), cached silently for the rest. + assert len(client.calls) == 1 + + +@pytest.mark.asyncio +async def test_action_commands_skip_display_name_precheck(): + client, results = await _run(_ActionCreate) + assert all(r["status"] == "success" for r in results) + assert client.calls == []