feat: add facility to add additional checks at end of file transforma…

docs/advanced_guidance/json_schemas/contract/components/base_entity.schema.json

@@ -26,8 +26,15 @@
             "items": {
                 "type": "string"
             }
+        },
+        "reader_additional_checks": {
+            "description": "A mapping of additional checks to perform on entities after initial read",
+            "type": "object",
+            "additionalProperties": {
+                "$ref": "reader_additional_checks.json"
         }
-    },
+    }
+},
 
     "required": ["fields"]
 }

docs/advanced_guidance/json_schemas/contract/components/reader_additional_checks.schema.json

@@ -0,0 +1,22 @@
+{
+    "$schema": "https://json-schema.org/draft/2020-12/schema",
+    "$id": "data-ingest:contract/components/reader_additional_checks.schema.json",
+    "title": "reader_additional_checks",
+    "description": "Additional checks to perform on initially read entities",
+    "type": "object",
+    "properties": {
+        "error_code": {
+            "description": "The code to be used for the additional check specified",
+            "type": "string"
+        },
+        "error_message": {
+            "description": "The message to be displayed for the additional check specified.",
+            "type": "string"
+        }
+    },
+    "required": [
+        "error_code",
+        "error_message"
+    ],
+    "additionalProperties": false
+}

src/dve/core_engine/backends/base/reader.py

@@ -10,6 +10,10 @@
 
 from dve.core_engine.backends.exceptions import MessageBearingError, ReaderLacksEntityTypeSupport
 from dve.core_engine.backends.types import EntityName, EntityType
+from dve.core_engine.configuration.v1 import (
+    AllowedAdditionalReaderChecks,
+    _ReaderAdditionalChecksConfig,
+)
 from dve.core_engine.message import FeedbackMessage
 from dve.core_engine.type_hints import URI, ArbitraryFunction, WrapDecorator
 from dve.parser.file_handling.service import open_stream
@@ -109,29 +113,58 @@ def read_to_entity_type(
         entity_name: EntityName,
         schema: type[BaseModel],
         all_model_fields: Optional[set[str]] = None,
-    ) -> EntityType:
+        additional_checks: Optional[
+            dict[AllowedAdditionalReaderChecks, _ReaderAdditionalChecksConfig]
+        ] = None,
+    ):
         """Read to the specified entity type, if supported.
 
         NOTE: Simple types should either be returned as strings (if present) or
         `None`. Format validation, casting, and parsing should be done in the
         data contract.
 
         """
-        if entity_name == Iterator[dict[str, Any]]:
-            return self.read_to_py_iterator(
+        additional_checks = additional_checks or {}
+
+        self.raise_if_not_sensible_file(resource, entity_name)
+
+        if entity_type == Iterator[dict[str, Any]]:
+            entity = self.read_to_py_iterator(
                 resource, entity_name, schema, all_model_fields  # type: ignore
             )
 
-        self.raise_if_not_sensible_file(resource, entity_name)
+        else:
 
-        try:
-            reader_func = self.__read_methods__[entity_type]
-        except KeyError as err:
-            raise ReaderLacksEntityTypeSupport(entity_type=entity_type) from err
+            try:
+                reader_func = self.__read_methods__[entity_type]
+            except KeyError as err:
+                raise ReaderLacksEntityTypeSupport(entity_type=entity_type) from err
 
-        return reader_func(
-            self, resource, entity_name, schema, all_model_fields=all_model_fields  # type: ignore
-        )
+            entity = reader_func(
+                self,
+                resource,
+                entity_name,
+                schema,
+                all_model_fields=all_model_fields,  # type: ignore
+            )
+
+        if config := additional_checks.get("check_empty"):
+            if self.check_entity_empty(entity):
+                raise MessageBearingError(
+                    f"The mandatory entity {entity_name} is empty",
+                    messages=[
+                        FeedbackMessage(
+                            entity=entity_name,
+                            record=None,
+                            failure_type="submission",
+                            error_location=entity_name,
+                            error_code=config.error_code,
+                            error_message=config.error_message,
+                        )
+                    ],
+                )
+
+        return entity
 
     def add_record_index(self, entity: EntityType, **kwargs) -> EntityType:
         """Add a record index to the entity"""
@@ -141,6 +174,10 @@ def drop_record_index(self, entity: EntityType, **kwargs) -> EntityType:
         """Drop a record index to the entity"""
         raise NotImplementedError(f"drop_record_index not implemented in {self.__class__}")
 
+    def check_entity_empty(self, entity: EntityType) -> bool:
+        """Determine if the entity supplied is empty"""
+        raise NotImplementedError(f"check_entity_empty not implemented in {self.__class__}")
+
     def write_parquet(
         self,
         entity: EntityType,

src/dve/core_engine/backends/base/rules.py

@@ -448,8 +448,7 @@ def process_node(node: HierarchyNode):
                                 record=record,  # type: ignore
                                 error_location=location,
                                 error_message=template_object(
-                                    node.missing_parent_id_error_message,
-                                    record
+                                    node.missing_parent_id_error_message, record
                                 ),
                                 failure_type="record",
                                 error_type="record",
@@ -522,10 +521,7 @@ def process_node(node: HierarchyNode) -> bool:
                         entity=node.parent_entity,
                         record=record,  # type: ignore
                         error_location=location,
-                        error_message=template_object(
-                            node.no_valid_records_error_message,
-                            record
-                        ),
+                        error_message=template_object(node.no_valid_records_error_message, record),
                         failure_type="record",
                         error_type="record",
                         error_code=node.no_valid_records_error_code,

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

@@ -322,6 +322,16 @@ def duckdb_get_entity_count(cls):
     return cls
 
 
+def _duckdb_check_entity_empty(self, entity: DuckDBPyRelation) -> bool:  # pylint: disable=W0613
+    return entity.shape[0] == 0
+
+
+def duckdb_check_entity_empty(cls):
+    """Class decorator to check whether a supplied entity is empty"""
+    cls.check_entity_empty = _duckdb_check_entity_empty
+    return cls
+
+
 def get_all_registered_udfs(connection: DuckDBPyConnection) -> set[str]:
     """Function to supply the names of a registered functions stored in the supplied
     duckdb connection. Creates the temp table used to store registered functions (if not exists).

src/dve/core_engine/backends/implementations/duckdb/readers/csv.py

@@ -22,6 +22,7 @@
     UnableToParseCSVError,
 )
 from dve.core_engine.backends.implementations.duckdb.duckdb_helpers import (
+    duckdb_check_entity_empty,
     duckdb_record_index,
     duckdb_write_parquet,
     get_duckdb_type_from_annotation,
@@ -35,6 +36,7 @@
 from dve.parser.file_handling import get_content_length
 
 
+@duckdb_check_entity_empty
 @duckdb_record_index
 @duckdb_write_parquet
 class DuckDBCSVReader(CSVFileReader):

src/dve/core_engine/backends/implementations/duckdb/readers/json.py

@@ -10,6 +10,7 @@
 
 from dve.core_engine.backends.base.reader import BaseFileReader, read_function
 from dve.core_engine.backends.implementations.duckdb.duckdb_helpers import (
+    duckdb_check_entity_empty,
     duckdb_record_index,
     duckdb_write_parquet,
     get_duckdb_type_from_annotation,
@@ -18,6 +19,7 @@
 from dve.core_engine.type_hints import URI, EntityName
 
 
+@duckdb_check_entity_empty
 @duckdb_record_index
 @duckdb_write_parquet
 class DuckDBJSONReader(BaseFileReader):

src/dve/core_engine/backends/implementations/duckdb/readers/xml.py

@@ -10,7 +10,10 @@
 
 from dve.core_engine.backends.base.reader import read_function
 from dve.core_engine.backends.exceptions import MessageBearingError
-from dve.core_engine.backends.implementations.duckdb.duckdb_helpers import duckdb_write_parquet
+from dve.core_engine.backends.implementations.duckdb.duckdb_helpers import (
+    duckdb_check_entity_empty,
+    duckdb_write_parquet,
+)
 from dve.core_engine.backends.readers.xml import XMLStreamReader
 from dve.core_engine.backends.utilities import (
     get_polars_type_from_annotation,
@@ -20,6 +23,7 @@
 from dve.core_engine.type_hints import URI
 
 
+@duckdb_check_entity_empty
 @polars_record_index
 @duckdb_write_parquet
 class DuckDBXMLStreamReader(XMLStreamReader):

src/dve/core_engine/backends/implementations/spark/readers/csv.py

@@ -12,6 +12,7 @@
 from dve.core_engine.backends.exceptions import EmptyFileError
 from dve.core_engine.backends.implementations.spark.spark_helpers import (
     get_type_from_annotation,
+    spark_check_entity_empty,
     spark_record_index,
     spark_write_parquet,
 )
@@ -20,6 +21,7 @@
 from dve.parser.file_handling import get_content_length
 
 
+@spark_check_entity_empty
 @spark_record_index
 @spark_write_parquet
 class SparkCSVReader(CSVFileReader):

src/dve/core_engine/backends/implementations/spark/readers/json.py

@@ -11,13 +11,15 @@
 from dve.core_engine.backends.exceptions import EmptyFileError
 from dve.core_engine.backends.implementations.spark.spark_helpers import (
     get_type_from_annotation,
+    spark_check_entity_empty,
     spark_record_index,
     spark_write_parquet,
 )
 from dve.core_engine.type_hints import URI, EntityName
 from dve.parser.file_handling import get_content_length
 
 
+@spark_check_entity_empty
 @spark_record_index
 @spark_write_parquet
 class SparkJSONReader(BaseFileReader):

src/dve/core_engine/backends/implementations/spark/readers/xml.py

@@ -17,6 +17,7 @@
 from dve.core_engine.backends.implementations.spark.spark_helpers import (
     df_is_empty,
     get_type_from_annotation,
+    spark_check_entity_empty,
     spark_record_index,
     spark_write_parquet,
 )
@@ -29,6 +30,7 @@
 """The mode to use when parsing XML files with Spark."""
 
 
+@spark_check_entity_empty
 @spark_record_index
 @spark_write_parquet
 class SparkXMLStreamReader(XMLStreamReader):
@@ -55,6 +57,7 @@ def read_to_dataframe(
         )
 
 
+@spark_check_entity_empty
 @spark_record_index
 @spark_write_parquet
 class SparkXMLReader(BasicXMLFileReader):  # pylint: disable=too-many-instance-attributes

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

@@ -456,6 +456,16 @@ def spark_get_entity_count(cls):
     return cls
 
 
+def _spark_check_entity_empty(self, entity: DataFrame) -> bool:  # pylint: disable=W0613
+    return entity.count() == 0
+
+
+def spark_check_entity_empty(cls):
+    """Class decorator to check whether a supplied entity is empty"""
+    cls.check_entity_empty = _spark_check_entity_empty
+    return cls
+
+
 def get_all_registered_udfs(spark: SparkSession) -> set[str]:
     """Function to supply the names of a registered functions stored in the supplied
     spark session.

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

@@ -57,6 +57,8 @@
 """The operation """
 RuleType = type[AbstractStep]
 """The metadata step type implemented by the rule."""
+AllowedAdditionalReaderChecks = Literal["check_empty"]
+"""Additional checks to be performed in the file_transformation stage"""
 
 
 class _BaseTypeDefintion(BaseModel):
@@ -146,6 +148,11 @@ class _SchemaConfig(BaseModel):
     """A list of the field names within the schema which _must_ be provided."""
 
 
+class _ReaderAdditionalChecksConfig(BaseModel):
+    error_code: str
+    error_message: str
+
+
 class _ReaderConfig(BaseModel):  # type: ignore
     """Reader configuration options for a model."""
 
@@ -166,6 +173,10 @@ class _ModelConfig(_SchemaConfig):
     """A single key field to be used by the model."""
     reader_config: dict[Extension, _ReaderConfig]
     """Reader configuration options for the model."""
+    reader_additional_checks: dict[AllowedAdditionalReaderChecks, _ReaderAdditionalChecksConfig] = (
+        Field(default_factory=dict)
+    )
+    """Additional checks to be performed after the entity is read"""
     aliases: dict[FieldName, FieldName] = Field(default_factory=dict)
     """An alias field name mapping."""
 

src/dve/pipeline/pipeline.py

@@ -238,11 +238,14 @@ def write_file_to_parquet(
                             model_name,
                             stringify_model(model),  # type: ignore
                             get_all_model_fields(models.values()),  # type: ignore
+                            dataset[model_name].reader_additional_checks,
                         ),
                         f"{out}{model_name}",
                     )
             except MessageBearingError as exc:
-                self._logger.error(f"Unable to process {model_name}", exc_info=exc)
+                self._logger.error(
+                    f"While processing {model_name}, an issue was encountered", exc_info=exc
+                )
                 errors.extend(exc.messages)
 
         return list(dict.fromkeys(errors))  # remove any duplicate errors

tests/features/flights.feature

@@ -263,3 +263,17 @@ Feature: Pipeline tests using the flights dataset
             | number_submission_rejections | 0     |
             | number_record_rejections     | 14    |
             | number_warnings              | 0     |
+
+    Scenario: A flights submission where mandatory entity has no records submitted
+        Given I submit the flights file only_country_id.xml for processing
+        And A duckdb pipeline is configured with schema file 'flights_add_reader_checks.dischema.json'
+        And I add initial audit entries for the submission
+        Then the latest audit record for the submission is marked with processing status file_transformation
+        When I run the file transformation phase
+        Then the country entity is stored as a parquet after the file_transformation phase
+        And there are errors with the following details and associated error_count from the file_transformation phase
+            | FailureType | ErrorCode    | error_count |
+            | submission  | AIRPORTEMPTY | 1           |
+        And the latest audit record for the submission is marked with processing status error_report
+        When I run the error report phase
+        Then An error report is produced

tests/features/steps/steps_pipeline.py

@@ -33,8 +33,8 @@
 
 from utilities import (
     load_errors_from_service,
-    get_test_file_path,
     SERVICE_TO_STORAGE_PATH_MAPPING,
+    get_test_file_path,
     get_all_errors_df,
 )
 

tests/features/steps/utilities.py

@@ -49,7 +49,7 @@ def load_errors_from_service(processing_folder: Path, service: str) -> pl.DataFr
     err_location = Path(
         processing_folder,
         "errors",
-        f"{SERVICE_TO_STORAGE_PATH_MAPPING.get(service, service)}_errors.jsonl",
+        f"{service}_errors.jsonl",
     )
     msgs = []
     try: