diff --git a/src/dve/core_engine/backends/base/rules.py b/src/dve/core_engine/backends/base/rules.py index 1340e32..741d566 100644 --- a/src/dve/core_engine/backends/base/rules.py +++ b/src/dve/core_engine/backends/base/rules.py @@ -28,6 +28,7 @@ CopyEntity, DeferredFilter, EntityRemoval, + GroupIdentification, HeaderJoin, ImmediateFilter, InnerJoin, @@ -344,6 +345,14 @@ def remove_orphans(self, entities: Entities, *, config: OrphanRemoval) -> Iterat """ raise NotImplementedError + @abstractmethod + def check_mandatory_group(self, entities: Entities, *, config: GroupIdentification) -> Iterator: + """ + Check that a mandatory key in an entity has at least one valid entry in the all the child + entities. + """ + raise NotImplementedError + @abstractmethod def union(self, entities: Entities, *, config: TableUnion) -> Messages: """Union two entities together, taking the columns from each by name. @@ -382,7 +391,7 @@ def identify_and_remove_orphans( entities: Entities, entity_hierarchy: EntityHierarchy, key_fields: Optional[dict[str, list[str]]] = None, - ) -> Messages: + ) -> tuple[Messages, bool]: """ Identifies and removes orphan records by traversing the EntityHierarchy object. An orphan is a child record whose parent FK does not exist in the parent entity. @@ -436,39 +445,120 @@ def process_node( code=node.orphaned_records_error_code, message=node.orphaned_records_error_message, location=location, - ) - ) + ), + ), ) for record in _orph_records: - msg_writer.write_queue.put([ - FeedbackMessage( - entity=current_entity_name, - record=record, # type: ignore - error_location=location, - error_message=node.orphaned_records_error_message, - failure_type="record", - error_type="record", - error_code=node.orphaned_records_error_code, - reporting_field=location, - category="Parent Missing", - ) - ]) + msg_writer.write_queue.put( + [ + FeedbackMessage( + entity=current_entity_name, + record=record, # type: ignore + error_location=location, + error_message=node.orphaned_records_error_message, + failure_type="record", + error_type="record", + error_code=node.orphaned_records_error_code, + reporting_field=location, + category="Parent Missing", + ) + ] + ) if node.children: for child_node in node.children: process_node(child_node, current_entity_name, orph_messages) + processed = False for root_node in entity_hierarchy.entity_trees.values(): process_node(root_node, parent_entity_name=None) _orph_rel = entities.get(ORPHANED_RECORD_ENTITY_NAME) - if _orph_rel: + if _orph_rel is not None: + processed = True del entities[ORPHANED_RECORD_ENTITY_NAME] entities.update(entities) - return [] + return [], processed + + def identify_and_remove_missing_mandatory_groups( + self, + working_directory: URI, + entities: Entities, + entity_hierarchy: EntityHierarchy, + key_fields: Optional[dict[str, list[str]]] = None, + ) -> tuple[Messages, bool]: + """ + Identify that an entity with a mandatory key has at least one valid child record. + """ + + def process_node( + node: HierarchyNode | ChildHierarchyNode, + parent_entity_name: Optional[EntityName], + processed: Optional[bool], + ): + """Recursive helper to process a node and its children.""" + current_entity_name = node.entity_name + + if isinstance(node, ChildHierarchyNode) and parent_entity_name is not None: + self.logger.info( + f"Identifying that {current_entity_name} has at least 1 valid child record" + ) # pylint: disable=C0301 + + join_expr = " AND ".join( + f"{parent_entity_name}.{k} = {current_entity_name}.{v}" + for k, v in node.join_fields.items() + ) + + with BackgroundMessageWriter( + working_directory=working_directory, + dve_stage=self.__stage_name__, + key_fields=key_fields, + logger=self.logger, + ) as msg_writer: + processed = True + location = next(iter(node.join_fields.values())) + missing_children_records = self.check_mandatory_group( + entities=entities, + config=GroupIdentification( + entity_name=parent_entity_name, + target_name=node.entity_name, + join_condition=join_expr, + mandatory=node.mandatory, # type: ignore + ), + ) + for record in missing_children_records: + msg_writer.write_queue.put( + [ + FeedbackMessage( + entity=parent_entity_name, + record=record, # type: ignore + error_location=location, + error_message=node.no_valid_records_error_message, + failure_type="submission" if node.mandatory else "record", + error_type="submission" if node.mandatory else "record", + error_code=node.no_valid_records_error_code, + reporting_field=location, + category="Children missing", + is_informational=not node.mandatory, # type: ignore + ) + ] + ) + + if node.children: + for child_node in node.children: + process_node(child_node, current_entity_name, processed) + + processed = False + + for root_node in entity_hierarchy.entity_trees.values(): + process_node(root_node, parent_entity_name=None, processed=processed) + + entities.update(entities) + + return [], processed # pylint: disable=R0912,R0914 def apply_sync_filters( diff --git a/src/dve/core_engine/backends/implementations/duckdb/rules.py b/src/dve/core_engine/backends/implementations/duckdb/rules.py index 4479846..0fb5dd7 100644 --- a/src/dve/core_engine/backends/implementations/duckdb/rules.py +++ b/src/dve/core_engine/backends/implementations/duckdb/rules.py @@ -43,6 +43,7 @@ Aggregation, AntiJoin, ConfirmJoinHasMatch, + GroupIdentification, HeaderJoin, ImmediateFilter, InnerJoin, @@ -450,6 +451,45 @@ def remove_orphans(self, entities: DuckDBEntities, *, config: OrphanRemoval) -> orphan_rel.filter(f"entity_name = '{config.entity_name}'") ) + def check_mandatory_group( + self, entities: DuckDBEntities, *, config: GroupIdentification + ) -> Iterator: + """ + Check that a mandatory key in an entity has at least one valid entry in the all the + child entities. + """ + source_rel: DuckDBPyRelation = entities[config.entity_name] + source_rel = source_rel.set_alias(config.entity_name) + target_rel: DuckDBPyRelation = entities[config.target_name] + target_rel = target_rel.set_alias(config.target_name) + + source_columns = [f"{config.entity_name}.{c.strip()}" for c in source_rel.columns] + _pk, fk = config.join_condition.split("=") + + joined_rel = source_rel.join(target_rel, config.join_condition, "left").select( + *source_columns, + ColumnExpression(fk.strip()).alias("fk"), + ConstantExpression(config.mandatory).alias("mandatory"), + ) + + missing_children_rel = joined_rel.filter("fk IS NULL") + filtered_rel = joined_rel.filter("fk IS NOT NULL and not mandatory").select( + StarExpression(exclude=["fk", "mandatory"]) + ) + + _no_valid_child_records: tuple[int] = missing_children_rel.count("*").fetchone() # type: ignore # pylint: disable=C0301 + if _no_valid_child_records: + _no_valid_children = _no_valid_child_records[0] + else: + _no_valid_children = 0 + self.logger.info( + f"Found {_no_valid_children} records with no valid children in {config.entity_name}." + ) # pylint: disable=C0301 + + entities[config.entity_name] = filtered_rel + + return duckdb_rel_to_dictionaries(missing_children_rel) + def union(self, entities: DuckDBEntities, *, config: TableUnion) -> Messages: """Union two entities together, taking the columns from each by name. diff --git a/src/dve/core_engine/backends/implementations/spark/rules.py b/src/dve/core_engine/backends/implementations/spark/rules.py index ff15b52..cec1156 100644 --- a/src/dve/core_engine/backends/implementations/spark/rules.py +++ b/src/dve/core_engine/backends/implementations/spark/rules.py @@ -34,6 +34,7 @@ ColumnAddition, ColumnRemoval, ConfirmJoinHasMatch, + GroupIdentification, HeaderJoin, ImmediateFilter, InnerJoin, @@ -384,6 +385,12 @@ def remove_orphans( # TODO - implement for spark raise NotImplementedError + def check_mandatory_group( + self, entities: SparkEntities, *, config: GroupIdentification + ) -> Iterator: + # TODO - implement for spark + raise NotImplementedError + def filter(self, entities: SparkEntities, *, config: ImmediateFilter) -> Messages: """Filter an entity immediately, and do not emit any messages. diff --git a/src/dve/core_engine/backends/metadata/rules.py b/src/dve/core_engine/backends/metadata/rules.py index 9b96a14..4c26213 100644 --- a/src/dve/core_engine/backends/metadata/rules.py +++ b/src/dve/core_engine/backends/metadata/rules.py @@ -33,6 +33,7 @@ "CopyEntity", "DeferredFilter", "EntityRemoval", + "GroupIdentification", "HeaderJoin", "ImmediateFilter", "InnerJoin", @@ -40,6 +41,7 @@ "OneToOneJoin", "OneToOneJoin", "OrphanIdentification", + "OrphanRemoval", "ParentMetadata", "RenameEntity", "Rule", @@ -553,6 +555,7 @@ class OrphanIdentification(AbstractConditionalJoin): """ + Step = Union[AbstractStep, Literal["sync"]] """A step within a rule. This is either a rule config or the literal string 'sync'.""" @@ -564,6 +567,13 @@ class OrphanRemoval(BaseStep): """The reporting information for the row removal.""" +class GroupIdentification(AbstractConditionalJoin): + """Identify mandatory records which do not have any valid child records""" + + mandatory: bool + """Whether the primary key is mandatory and whether the record should be stripped.""" + + class Rule(BaseModel): """A rule, made up of multiple steps.""" diff --git a/src/dve/core_engine/type_hints.py b/src/dve/core_engine/type_hints.py index e369ff4..48d9eeb 100644 --- a/src/dve/core_engine/type_hints.py +++ b/src/dve/core_engine/type_hints.py @@ -133,7 +133,9 @@ """A string indicating the field that the error pertains to.""" FieldValue = Optional[Any] """The value that caused the error.""" -ErrorCategory = Literal["Blank", "Wrong format", "Bad value", "Bad file", "Parent Missing"] +ErrorCategory = Literal[ + "Blank", "Wrong format", "Bad value", "Bad file", "Parent Missing", "Children missing" +] """A string indicating the category of the error.""" RecordIndex = Optional[int] """The record index that the error relates to (if applicable)""" diff --git a/src/dve/parser/file_handling/service.py b/src/dve/parser/file_handling/service.py index 9ee9d9f..fbdc8ab 100644 --- a/src/dve/parser/file_handling/service.py +++ b/src/dve/parser/file_handling/service.py @@ -273,9 +273,12 @@ def copy_resource(source_uri: URI, target_uri: URI, overwrite: bool = False) -> _transfer_resource(source_uri, target_uri, overwrite, "copy") -def move_resource(source_uri: URI, target_uri: URI, overwrite: bool = False) -> None: - """Move a resource from one location to another.""" +def move_resource(source_uri: URI, target_uri: URI, overwrite: bool = False) -> URI: + """ + Move a resource from one location to another. Returns the target_uri. + """ _transfer_resource(source_uri, target_uri, overwrite, "move") + return target_uri def create_directory(target_uri: URI): diff --git a/src/dve/pipeline/pipeline.py b/src/dve/pipeline/pipeline.py index eaf7661..326ca7a 100644 --- a/src/dve/pipeline/pipeline.py +++ b/src/dve/pipeline/pipeline.py @@ -1,4 +1,4 @@ -# pylint: disable=protected-access,too-many-instance-attributes,too-many-arguments,line-too-long +# pylint: disable=protected-access,too-many-instance-attributes,too-many-arguments,line-too-long,too-many-lines """Generic Pipeline object to define how DVE should be interacted with.""" import json @@ -634,7 +634,7 @@ def apply_business_rules( # pylint: disable=R0914 fh.joinuri( self.processed_files_path, submission_info.submission_id, - "business_rules", + "temp_business_rules", entity_name, ), ) @@ -642,13 +642,53 @@ def apply_business_rules( # pylint: disable=R0914 projected ) - self.step_implementations.identify_and_remove_orphans( # type: ignore + _, orph_or_group = self.step_implementations.identify_and_remove_orphans( # type: ignore working_directory, entity_manager.entities, entity_hierarchy, key_fields, ) + _, orph_or_group = self.step_implementations.identify_and_remove_missing_mandatory_groups( # type: ignore + working_directory, + entity_manager.entities, + entity_hierarchy, + key_fields, + ) + + for entity_name, entity in entity_manager.entities.items(): + if orph_or_group: + self._logger.info(f"Writing {entity_name} out to disk.") + final_projection = self._step_implementations.write_parquet( # type: ignore + entity, + fh.joinuri( + self.processed_files_path, + submission_info.submission_id, + "business_rules", + entity_name, + ), + ) + else: + self._logger.info(f"Moving {entity_name} from temp_business_rules to business_rules") + final_projection = fh.move_resource( + source_uri=fh.joinuri( + self.processed_files_path, + submission_info.submission_id, + "temp_business_rules", + entity_name + ), + target_uri=fh.joinuri( + self.processed_files_path, + submission_info.submission_id, + "business_rules", + entity_name + ) + ) + + entity_manager.entities[entity_name] = self.step_implementations.read_parquet( # type: ignore + final_projection + ) + submission_status.number_of_records = self.get_entity_count( entity=entity_manager.entities[f"""Original{rules.global_variables.get( 'entity', diff --git a/src/dve/pipeline/utils.py b/src/dve/pipeline/utils.py index e6122c2..babcd47 100644 --- a/src/dve/pipeline/utils.py +++ b/src/dve/pipeline/utils.py @@ -68,7 +68,7 @@ def unpersist_all_rdds(spark: SparkSession): rdd.unpersist() -def deadletter_file(source_uri: URI) -> None: +def deadletter_file(source_uri: URI) -> URI | None: """Move files that can't be processed to a deadletter location""" try: source_parent: URI = source_uri.rsplit("/", 1)[0] diff --git a/tests/features/flights.feature b/tests/features/flights.feature index ea710e9..f1954c6 100644 --- a/tests/features/flights.feature +++ b/tests/features/flights.feature @@ -114,3 +114,59 @@ Feature: Pipeline tests using the flights dataset # | record_count | 1 | # | number_file_rejections | 0 | # | number_record_rejections | 1 | + + Scenario: A flights submission with no valid airports record on submission + Given I submit the flights file only_country_id.xml for processing + And A duckdb pipeline is configured with schema file 'flights.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 the airport entity is stored as a parquet after the file_transformation phase + And the flights entity is stored as a parquet after the file_transformation phase + And the passengers entity is stored as a parquet after the file_transformation phase + And the latest audit record for the submission is marked with processing status data_contract + When I run the data contract phase + Then there are no file rejections from the data_contract phase + And there are no record rejections from the data_contract phase + When I run the business rules phase + Then there are errors with the following details and associated error_count from the business_rules phase + | ErrorType | ErrorCode | error_count | + | submission | C2 | 1 | + When I run the error report phase + Then An error report is produced + # TODO - fix the stats calculations as they're currently incorrect for hiearchical datasets + # And The statistics entry for the submission shows the following information + # | parameter | value | + # | record_count | 1 | + # | number_file_rejections | 0 | + # | number_record_rejections | 1 | + + Scenario: A flights submission with a mixture of group and orphan record rejections + Given I submit the flights file invalid_flight_destination.xml for processing + And A duckdb pipeline is configured with schema file 'flights.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 the airport entity is stored as a parquet after the file_transformation phase + And the flights entity is stored as a parquet after the file_transformation phase + And the passengers entity is stored as a parquet after the file_transformation phase + And the latest audit record for the submission is marked with processing status data_contract + When I run the data contract phase + Then there are no file rejections from the data_contract phase + And there are no record rejections from the data_contract phase + When I run the business rules phase + Then there are errors with the following details and associated error_count from the business_rules phase + | ErrorType | Status | ErrorCode | error_count | + | record | error | F2 | 2 | + | record | error | PG1 | 4 | + | record | informational | A1 | 1 | + When I run the error report phase + Then An error report is produced +# TODO - fix the stats calculations as they're currently incorrect for hiearchical datasets +# And The statistics entry for the submission shows the following information +# | parameter | value | +# | record_count | 1 | +# | number_file_rejections | 0 | +# | number_record_rejections | 1 | diff --git a/tests/features/steps/steps_pipeline.py b/tests/features/steps/steps_pipeline.py index c72c873..c71bd45 100644 --- a/tests/features/steps/steps_pipeline.py +++ b/tests/features/steps/steps_pipeline.py @@ -183,8 +183,10 @@ def check_error_record_details_from_service(context: Context, service:str): message_df = load_errors_from_service(processing_path, service) for err_details in error_details: filter_expr, error_count = err_details - assert message_df.filter(filter_expr).shape[0] == error_count - + assert message_df.filter(filter_expr).shape[0] == error_count, message_df.select( + *[pl.col(c) for c in table.headings if c not in ["error_count"]] + ) + @given("A {implementation} pipeline is configured") @given("A {implementation} pipeline is configured with schema file '{schema_file_name}'") @@ -317,8 +319,3 @@ def create_refdata_tables(context: Context, database: str): pipeline._connection.sql(f"ATTACH '{ref_db_file}' AS {database}") for tbl, source in refdata_tables.items(): pipeline._connection.read_parquet(source).to_table(f"{database}.{tbl}") - - - - - diff --git a/tests/testdata/flights/flights.dischema.json b/tests/testdata/flights/flights.dischema.json index faac68f..8ab9dd7 100644 --- a/tests/testdata/flights/flights.dischema.json +++ b/tests/testdata/flights/flights.dischema.json @@ -104,6 +104,17 @@ "category": "Blank", "error_code": "F1" }, + { + "entity": "flights", + "name": "flight_missing_id", + "expression": "lower(destination) IN ('paris', 'madrid')", + "failure_type": "record", + "failure_message": "Record Rejected - {{ destination }} is not a valid destination", + "reporting_field": "flight_id", + "reporting_entity": "flights", + "category": "Bad value", + "error_code": "F2" + }, { "entity": "passengers", "name": "passenger_name_is_null", @@ -125,7 +136,9 @@ }, "mandatory": true, "orphaned_records_error_code": "AG1", - "orphaned_records_error_message": "Group rejected - No valid country group found country" + "orphaned_records_error_message": "Group rejected - No valid country group found country", + "no_valid_records_error_code": "C2", + "no_valid_records_error_message": "File rejected - No valid child entries found for this mandatory key" }, "flights": { "parent_entity": "airport", @@ -134,7 +147,9 @@ }, "mandatory": false, "orphaned_records_error_code": "FG1", - "orphaned_records_error_message": "Group rejected - No valid airport group found for airport" + "orphaned_records_error_message": "Group rejected - No valid airport group found for airport", + "no_valid_records_error_code": "A1", + "no_valid_records_error_message": "Warning - No valid child entries found for this mandatory key" }, "passengers": { "parent_entity": "flights", @@ -143,7 +158,9 @@ }, "mandatory": false, "orphaned_records_error_code": "PG1", - "orphaned_records_error_message": "Group rejected - No valid flight group found for passenger" + "orphaned_records_error_message": "Group rejected - No valid flight group found for passenger", + "no_valid_records_error_code": "F3", + "no_valid_records_error_message": "Warning - No valid child entries found for this mandatory key" } } } \ No newline at end of file diff --git a/tests/testdata/flights/invalid_flight_destination.xml b/tests/testdata/flights/invalid_flight_destination.xml new file mode 100644 index 0000000..134ee89 --- /dev/null +++ b/tests/testdata/flights/invalid_flight_destination.xml @@ -0,0 +1,49 @@ + + + 1 + England + + + 1 + 1 + Heathrow + TW6 1EW + + + 1 + 1 + Mars + + + 1 + 1 + John + + + 1 + 2 + Jane + + + + + 1 + 2 + Venus + + + 2 + 3 + Homer + + + 2 + 4 + Marge + + + + + + + \ No newline at end of file diff --git a/tests/testdata/flights/only_country_id.xml b/tests/testdata/flights/only_country_id.xml new file mode 100644 index 0000000..b0ebd07 --- /dev/null +++ b/tests/testdata/flights/only_country_id.xml @@ -0,0 +1,5 @@ + + + 1 + England + \ No newline at end of file