fix(dr-egeria): cache Egeria 404-002 ambiguous lookups; skip display-name pre-check for action commands

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 <Type> 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)

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 == []