feat: group rejections for mandatory primary keys

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

@@ -28,6 +28,7 @@
     CopyEntity,
     DeferredFilter,
     EntityRemoval,
+    GroupIdentification,
     HeaderJoin,
     ImmediateFilter,
     InnerJoin,
@@ -344,6 +345,14 @@
         """
         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.
@@ -376,7 +385,7 @@
 
         """
 
     def identify_and_remove_orphans(
         self,
         working_directory: URI,
         entities: Entities,
@@ -403,7 +412,7 @@
             if isinstance(node, ChildHierarchyNode) and parent_entity_name is not None:
                 self.logger.info(f"Identifying orphans in {current_entity_name}")
 
                 join_expr = " AND ".join(
                     f"{parent_entity_name}.{k} = {current_entity_name}.{v}"
                     for k, v in node.join_fields.items()
                 )
@@ -436,36 +445,110 @@
                                     code=node.orphaned_records_error_code,
                                     message=node.orphaned_records_error_message,
                                     location=location,
-                                )
-                            )
+                                ),
+                            ),
                         )
                         for record in _orph_records:
-                            msg_writer.write_queue.put([
+                            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)
+
+        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 is not None:
+            del entities[ORPHANED_RECORD_ENTITY_NAME]
+
+        entities.update(entities)
+
+        return []
+
+    def identify_and_remove_missing_mandatory_groups(
+        self,
+        working_directory: URI,
+        entities: Entities,
+        entity_hierarchy: EntityHierarchy,
+        key_fields: Optional[dict[str, list[str]]] = None,
+    ) -> Messages:
+        """
+        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],
+        ):
+            """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:
+                    location = list(node.join_fields.values())[0]
+                    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=current_entity_name,
+                                    entity=parent_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,
+                                    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="Parent Missing",
+                                    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, orph_messages)
-
+                    process_node(child_node, current_entity_name)
 
         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:
-            del entities[ORPHANED_RECORD_ENTITY_NAME]
-
         entities.update(entities)
 
         return []

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.
 

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

@@ -34,6 +34,7 @@
     ColumnAddition,
     ColumnRemoval,
     ConfirmJoinHasMatch,
+    GroupIdentification,
     HeaderJoin,
     ImmediateFilter,
     InnerJoin,
@@ -384,6 +385,12 @@
         # 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.
 

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

@@ -33,13 +33,15 @@
     "CopyEntity",
     "DeferredFilter",
     "EntityRemoval",
+    "GroupIdentification",
     "HeaderJoin",
     "ImmediateFilter",
     "InnerJoin",
     "LeftJoin",
     "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."""
 

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

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
@@ -629,18 +629,23 @@
             else:
                 self._logger.info(f"Skipping {entity_name}. Marked original.")
                 filtered_entity = entity
-            projected = self._step_implementations.write_parquet(  # type: ignore
-                filtered_entity,
-                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
-                projected
-            )
+            # todo - Removing for now as double write causing spark write to crash - look into fix
+            # todo - main benefit of double write is that the execution plan to be truncated before
+            # todo - complex joins and checks performed in the orphan and group rejection.
+            # todo - ideally should only write twice if those steps are actually required.
+            # projected = self._step_implementations.write_parquet(  # type: ignore
+            #     filtered_entity,
+            #     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
+            #     projected
+            # )
+            entity_manager.entities[entity_name] = filtered_entity
 
         self.step_implementations.identify_and_remove_orphans(  # type: ignore
             working_directory,
@@ -649,6 +654,25 @@
             key_fields,
         )
 
+        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():
+            self._logger.info(f"Writing {entity_name} out to disk.")
+            self._step_implementations.write_parquet(  # type: ignore
+                entity,
+                fh.joinuri(
+                    self.processed_files_path,
+                    submission_info.submission_id,
+                    "business_rules",
+                    entity_name,
+                ),
+            )
+
         submission_status.number_of_records = self.get_entity_count(
             entity=entity_manager.entities[f"""Original{rules.global_variables.get(
                                               'entity',