fix: ensure that original entity detials persisted in feedback messages for further processing (#168)

Author: stevenhsd

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

@@ -284,11 +284,11 @@ def _ddb_filter_contract_errors(
                 "RecordIndex": "INTEGER",
                 "FailureType": "STRING",
                 "Status": "STRING",
-                "Entity": "STRING",
+                "OriginalEntity": "STRING",
             },
         )
         .filter(
-            f"FailureType == 'record' AND Status != 'informational' AND Entity = '{entity_name}'"
+            f"FailureType == 'record' AND Status != 'informational' AND OriginalEntity = '{entity_name}'" # pylint: disable=C0301
         )  # pylint: disable=C0301
         .select("RecordIndex")
         .distinct()

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

@@ -415,14 +415,14 @@ def _spark_filter_contract_errors(
                     st.StructField("RecordIndex", st.IntegerType()),
                     st.StructField("FailureType", st.StringType()),
                     st.StructField("Status", st.StringType()),
-                    st.StructField("Entity", st.StringType()),
+                    st.StructField("OriginalEntity", st.StringType()),
                 ]
             ),
         )
         .filter(
             (sf.col("FailureType") == sf.lit("record"))
             & (sf.col("Status") != sf.lit("informational"))
-            & (sf.col("Entity") == sf.lit(entity_name))
+            & (sf.col("OriginalEntity") == sf.lit(entity_name))
         )
         .distinct()
         .orderBy(sf.asc(sf.col("RecordIndex")))

src/dve/core_engine/message.py

@@ -90,7 +90,7 @@ def extract_error_value(records, error_location):
 class UserMessage:
     """The structure of the message that is used to populate the error report."""
 
-    Entity: Optional[str]
+    ReportingEntity: Optional[str]
     """The entity that the message pertains to (if applicable)."""
     Key: Optional[str]
     "The key field(s) in string format to allow users to identify the record"
@@ -176,7 +176,8 @@ class FeedbackMessage:  # pylint: disable=too-many-instance-attributes
     """The category of the error."""
 
     HEADER: ClassVar[list[str]] = [
-        "Entity",
+        "ReportingEntity",
+        "OriginalEntity",
         "Key",
         "FailureType",
         "Status",
@@ -307,6 +308,7 @@ def to_row(
 
         return (
             self.entity,
+            self.original_entity,
             key,
             self.failure_type,
             "informational" if self.is_informational else "error",

src/dve/core_engine/type_hints.py

@@ -147,6 +147,7 @@
 """The record index that the error relates to (if applicable)"""
 
 MessageTuple = tuple[
+    Optional[EntityName],
     Optional[EntityName],
     Key,
     FailureType,

src/dve/pipeline/pipeline.py

@@ -908,7 +908,7 @@ def _get_error_dataframes(self, submission_id: str):
                     .alias("error_type")  # type: ignore
                 )
                 df = df.select(
-                    pl.col("Entity").alias("Table"),  # type: ignore
+                    pl.col("ReportingEntity").alias("Table"),  # type: ignore
                     pl.col("error_type").alias("Type"),  # type: ignore
                     pl.col("ErrorCode").alias("Error_Code"),  # type: ignore
                     pl.col("ReportingField").alias("Data_Item"),  # type: ignore

src/dve/reporting/error_report.py

@@ -91,7 +91,7 @@ def create_error_dataframe(errors: deque[FeedbackMessage], key_fields):
         .alias("error_type")
     )
     df = df.select(  # type: ignore
-        col("Entity").alias("Table"),  # type: ignore
+        col("ReportingEntity").alias("Table"),  # type: ignore
         col("error_type").alias("Type"),  # type: ignore
         col("ErrorCode").alias("Error_Code"),  # type: ignore
         col("ReportingField").alias("Data_Item"),  # type: ignore

tests/features/movies.feature

@@ -22,7 +22,7 @@ Feature: Pipeline tests using the movies dataset
         Then there is 1 submission rejection from the data_contract phase
         And there are 3 record rejections from the data_contract phase
         And there are errors with the following details and associated error_count from the data_contract phase
-            | Entity             | ErrorCode  | ErrorMessage                              | RecordIndex | error_count |
+            | ReportingEntity    | ErrorCode  | ErrorMessage                              | RecordIndex | error_count |
             | movies             | BLANKYEAR  | year not provided                         | 2           | 1           |
             | movies_rename_test | DODGYYEAR  | year value (NOT_A_NUMBER) is invalid      | 1           | 1           |
             | movies             | DODGYDATE  | date_joined value is not valid: daft_date | 1           | 1           |
@@ -61,10 +61,10 @@ Feature: Pipeline tests using the movies dataset
         Then there is 1 submission rejection from the data_contract phase
         And there are 3 record rejections from the data_contract phase
         And there are errors with the following details and associated error_count from the data_contract phase
-            | Entity             | ErrorCode | ErrorMessage                              | RecordIndex | error_count |
-            | movies             | BLANKYEAR | year not provided                         | 2           | 1           |
-            | movies_rename_test | DODGYYEAR | year value (NOT_A_NUMBER) is invalid      | 1           | 1           |
-            | movies             | DODGYDATE | date_joined value is not valid: daft_date | 1           | 1           |
+            | ReportingEntity    | ErrorCode  | ErrorMessage                              | RecordIndex | error_count |
+            | movies             | BLANKYEAR  | year not provided                         | 2           | 1           |
+            | movies_rename_test | DODGYYEAR  | year value (NOT_A_NUMBER) is invalid      | 1           | 1           |
+            | movies             | DODGYDATE  | date_joined value is not valid: daft_date | 1           | 1           |
             | movies             | BLANKTITLE | title should not be blank                 | 4           | 1           |
         And the movies entity is stored as a parquet after the data_contract phase
         And the latest audit record for the submission is marked with processing status business_rules

tests/features/steps/steps_pipeline.py

@@ -284,7 +284,7 @@ def check_rows_removed_with_error_code(context: Context, entity_name: str, error
     err_df = get_all_errors_df(context)
 
     recs_with_err_code = err_df.filter(
-        (pl.col("Entity").eq(entity_name)) & (pl.col("ErrorCode").eq(error_code))
+        (pl.col("ReportingEntity").eq(entity_name)) & (pl.col("ErrorCode").eq(error_code))
     ).shape[0]
     assert recs_with_err_code >= 1
 

tests/features/steps/utilities.py

@@ -15,7 +15,7 @@
 from dve.parser.type_hints import URI
 
 ERROR_DF_FIELDS: List[str] = [
-    "Entity",
+    "ReportingEntity",
     "Key",
     "ErrorCode",
     "FailureType",

tests/test_core_engine/test_backends/test_implementations/test_duckdb/test_data_contract.py

@@ -381,4 +381,4 @@ def test_duckdb_data_contract_custom_error_details(nested_all_string_parquet_w_e
     assert messages[0].ErrorMessage == "subfield id is invalid: subfield.id - WRONG"
     assert messages[1].ErrorCode == "TESTIDBAD"
     assert messages[1].ErrorMessage == "id is invalid: id - WRONG"
-    assert messages[1].Entity == "test_rename"
+    assert messages[1].ReportingEntity == "test_rename"