fix: enable passing of original_entity_override to fix instance where id cannot be sourced from current entity

Author: georgeRobertson

src/dve/core_engine/backends/implementations/duckdb/rules.py

@@ -518,7 +518,7 @@ def notify(self, entities: DuckDBEntities, *, config: Notification) -> Messages:
             messages.append(
                 FeedbackMessage(
                     entity=config.reporting.reporting_entity_override or config.entity_name,
-                    original_entity=config.entity_name,
+                    original_entity=config.reporting.original_entity_override or config.entity_name,
                     record=record,  # type: ignore
                     error_location=config.reporting.legacy_location,
                     error_message=template_object(config.reporting.message, record),  # type: ignore

src/dve/core_engine/backends/implementations/spark/rules.py

@@ -412,7 +412,7 @@ def notify(self, entities: SparkEntities, *, config: Notification) -> Messages:
                 # more complex extraction done in reporting module
                 FeedbackMessage(
                     entity=config.reporting.reporting_entity_override or config.entity_name,
-                    original_entity=config.entity_name,
+                    original_entity=config.reporting.original_entity_override or config.entity_name,
                     record=record.asDict(recursive=True),
                     error_location=config.reporting.legacy_location,
                     error_message=template_object(

src/dve/core_engine/backends/metadata/reporting.py

@@ -10,6 +10,7 @@
 
 from dve.core_engine.templating import template_object
 from dve.core_engine.type_hints import (
+    EntityName,
     ErrorCategory,
     ErrorCode,
     ErrorEmitValue,
@@ -102,6 +103,12 @@ class BaseReportingConfig(BaseModel):
     If this is not provided, the top level names from the `location` field
     will be used in the report.
 
+    """
+    original_entity_override: Optional[EntityName] = None
+    """
+    This should only be populated where you need to refer back to an originally defined entity that
+    contains the key field(s) of interest. This might be required in cases where you've created
+    lots of new entities during a complex rule.
     """
 
     def template(

src/dve/core_engine/configuration/v1/filters.py

@@ -24,6 +24,7 @@ class ConcreteFilterConfig(BaseModel):
     reporting_field: Optional[Union[str, list[str]]] = None
     reporting_field_name: Optional[str] = None
     category: ErrorCategory = "Bad value"
+    original_entity_override: Optional[str] = None
 
     def to_step(self) -> AbstractStep:
         """Create a deferred filter from the concrete filter config."""
@@ -37,6 +38,7 @@ def to_step(self) -> AbstractStep:
             legacy_is_informational=self.is_informational,
             legacy_location=self.error_location,
             legacy_reporting_field=self.reporting_field,
+            original_entity_override=self.original_entity_override,
         )
 
         return DeferredFilter(