From 36446d7be43bc3b0ef3d64151458d021a85520d1 Mon Sep 17 00:00:00 2001
From: georgeRobertson <50412379+georgeRobertson@users.noreply.github.com>
Date: Fri, 11 Sep 2026 20:28:32 +0100
Subject: [PATCH 1/7] feat: group rejections for mandatory primary keys
---
src/dve/core_engine/backends/base/rules.py | 115 +++++++++++++++---
.../backends/implementations/duckdb/rules.py | 40 ++++++
.../backends/implementations/spark/rules.py | 7 ++
.../core_engine/backends/metadata/rules.py | 10 ++
src/dve/core_engine/type_hints.py | 4 +-
src/dve/pipeline/pipeline.py | 50 ++++++--
tests/features/flights.feature | 56 +++++++++
tests/features/steps/steps_pipeline.py | 11 +-
tests/testdata/flights/flights.dischema.json | 23 +++-
.../flights/invalid_flight_destination.xml | 49 ++++++++
tests/testdata/flights/only_country_id.xml | 5 +
11 files changed, 330 insertions(+), 40 deletions(-)
create mode 100644 tests/testdata/flights/invalid_flight_destination.xml
create mode 100644 tests/testdata/flights/only_country_id.xml
diff --git a/src/dve/core_engine/backends/base/rules.py b/src/dve/core_engine/backends/base/rules.py
index 1340e32..a6413c4 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.
@@ -436,36 +445,110 @@ 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([
+ 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 []
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/pipeline/pipeline.py b/src/dve/pipeline/pipeline.py
index eaf7661..3f214fe 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
@@ -629,18 +629,23 @@ def apply_business_rules( # pylint: disable=R0914
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 @@ def apply_business_rules( # pylint: disable=R0914
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',
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
From a1b49bc4930094cafc8305dc05301c5028d05b50 Mon Sep 17 00:00:00 2001
From: georgeRobertson <50412379+georgeRobertson@users.noreply.github.com>
Date: Mon, 14 Sep 2026 10:57:35 +0100
Subject: [PATCH 2/7] refactor: add initial write back into pipeline
---
src/dve/pipeline/pipeline.py | 34 ++++++++++++++++------------------
1 file changed, 16 insertions(+), 18 deletions(-)
diff --git a/src/dve/pipeline/pipeline.py b/src/dve/pipeline/pipeline.py
index 3f214fe..799b2e9 100644
--- a/src/dve/pipeline/pipeline.py
+++ b/src/dve/pipeline/pipeline.py
@@ -629,23 +629,18 @@ def apply_business_rules( # pylint: disable=R0914
else:
self._logger.info(f"Skipping {entity_name}. Marked original.")
filtered_entity = entity
- # 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
+ projected = self._step_implementations.write_parquet( # type: ignore
+ filtered_entity,
+ fh.joinuri(
+ self.processed_files_path,
+ submission_info.submission_id,
+ "temp_business_rules",
+ entity_name,
+ ),
+ )
+ entity_manager.entities[entity_name] = self.step_implementations.read_parquet( # type: ignore
+ projected
+ )
self.step_implementations.identify_and_remove_orphans( # type: ignore
working_directory,
@@ -663,7 +658,7 @@ def apply_business_rules( # pylint: disable=R0914
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
+ final_projection = self._step_implementations.write_parquet( # type: ignore
entity,
fh.joinuri(
self.processed_files_path,
@@ -672,6 +667,9 @@ def apply_business_rules( # pylint: disable=R0914
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(
From f2a91f36b4d053ae95a41721e7407d1e7950ae74 Mon Sep 17 00:00:00 2001
From: georgeRobertson <50412379+georgeRobertson@users.noreply.github.com>
Date: Mon, 14 Sep 2026 17:10:32 +0100
Subject: [PATCH 3/7] refactor: add condition to final write to be more
performant
---
src/dve/core_engine/backends/base/rules.py | 19 ++++++----
src/dve/parser/file_handling/service.py | 7 ++--
src/dve/pipeline/pipeline.py | 42 +++++++++++++++-------
src/dve/pipeline/utils.py | 2 +-
4 files changed, 49 insertions(+), 21 deletions(-)
diff --git a/src/dve/core_engine/backends/base/rules.py b/src/dve/core_engine/backends/base/rules.py
index a6413c4..6247037 100644
--- a/src/dve/core_engine/backends/base/rules.py
+++ b/src/dve/core_engine/backends/base/rules.py
@@ -391,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.
@@ -469,16 +469,19 @@ def process_node(
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 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,
@@ -486,7 +489,7 @@ def identify_and_remove_missing_mandatory_groups(
entities: Entities,
entity_hierarchy: EntityHierarchy,
key_fields: Optional[dict[str, list[str]]] = None,
- ) -> Messages:
+ ) -> tuple[Messages, bool]:
"""
Identify that an entity with a mandatory key has at least one valid child record.
"""
@@ -494,6 +497,7 @@ def identify_and_remove_missing_mandatory_groups(
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
@@ -514,6 +518,7 @@ def process_node(
key_fields=key_fields,
logger=self.logger,
) as msg_writer:
+ processed = True
location = list(node.join_fields.values())[0]
missing_children_records = self.check_mandatory_group(
entities=entities,
@@ -544,14 +549,16 @@ def process_node(
if node.children:
for child_node in node.children:
- process_node(child_node, current_entity_name)
+ 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)
+ process_node(root_node, parent_entity_name=None, processed=processed)
entities.update(entities)
- return []
+ return [], processed
# pylint: disable=R0912,R0914
def apply_sync_filters(
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 799b2e9..326ca7a 100644
--- a/src/dve/pipeline/pipeline.py
+++ b/src/dve/pipeline/pipeline.py
@@ -642,14 +642,14 @@ 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,
)
- self.step_implementations.identify_and_remove_missing_mandatory_groups( # type: ignore
+ _, orph_or_group = self.step_implementations.identify_and_remove_missing_mandatory_groups( # type: ignore
working_directory,
entity_manager.entities,
entity_hierarchy,
@@ -657,16 +657,34 @@ def apply_business_rules( # pylint: disable=R0914
)
for entity_name, entity in entity_manager.entities.items():
- 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,
- ),
- )
+ 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
)
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]
From 4a0288d169a4b6742d53dac3a58e74bcb3725947 Mon Sep 17 00:00:00 2001
From: georgeRobertson <50412379+georgeRobertson@users.noreply.github.com>
Date: Mon, 14 Sep 2026 17:29:38 +0100
Subject: [PATCH 4/7] style: sonar feedback
M
---
src/dve/core_engine/backends/base/rules.py | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
diff --git a/src/dve/core_engine/backends/base/rules.py b/src/dve/core_engine/backends/base/rules.py
index 6247037..741d566 100644
--- a/src/dve/core_engine/backends/base/rules.py
+++ b/src/dve/core_engine/backends/base/rules.py
@@ -519,7 +519,7 @@ def process_node(
logger=self.logger,
) as msg_writer:
processed = True
- location = list(node.join_fields.values())[0]
+ location = next(iter(node.join_fields.values()))
missing_children_records = self.check_mandatory_group(
entities=entities,
config=GroupIdentification(
From 3114b4fac5a20f3b51d69d41c7947e93f7e0f77f Mon Sep 17 00:00:00 2001
From: georgeRobertson <50412379+georgeRobertson@users.noreply.github.com>
Date: Tue, 15 Sep 2026 13:25:38 +0100
Subject: [PATCH 5/7] docs: update jsonschema for entity relationships to
include group rej code+message
---
.../json_schemas/entity_relationships.schema.json | 6 ++++++
1 file changed, 6 insertions(+)
diff --git a/docs/advanced_guidance/json_schemas/entity_relationships.schema.json b/docs/advanced_guidance/json_schemas/entity_relationships.schema.json
index 9d4e363..ff35309 100644
--- a/docs/advanced_guidance/json_schemas/entity_relationships.schema.json
+++ b/docs/advanced_guidance/json_schemas/entity_relationships.schema.json
@@ -25,6 +25,12 @@
},
"missing_parent_id_error_message": {
"type": "string"
+ },
+ "no_valid_records_error_code": {
+ "type": "string"
+ },
+ "no_valid_records_error_message": {
+ "type": "string"
}
},
"required": [
From 9863565757b205987deda11b435ce58560f7e6e1 Mon Sep 17 00:00:00 2001
From: georgeRobertson <50412379+georgeRobertson@users.noreply.github.com>
Date: Tue, 15 Sep 2026 13:38:19 +0100
Subject: [PATCH 6/7] fix: remove childHierarchy object that's no longer valid
---
src/dve/core_engine/backends/base/rules.py | 4 ++--
1 file changed, 2 insertions(+), 2 deletions(-)
diff --git a/src/dve/core_engine/backends/base/rules.py b/src/dve/core_engine/backends/base/rules.py
index 362293c..b220631 100644
--- a/src/dve/core_engine/backends/base/rules.py
+++ b/src/dve/core_engine/backends/base/rules.py
@@ -494,14 +494,14 @@ def identify_and_remove_missing_mandatory_groups(
"""
def process_node(
- node: HierarchyNode | ChildHierarchyNode,
+ node: HierarchyNode,
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:
+ if 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
From 953f1a3dcaf1264c664fa5e6caed82de91499fbe Mon Sep 17 00:00:00 2001
From: georgeRobertson <50412379+georgeRobertson@users.noreply.github.com>
Date: Tue, 15 Sep 2026 14:03:40 +0100
Subject: [PATCH 7/7] fix: change processed in orphan and group rejections to
work as intended
---
src/dve/core_engine/backends/base/rules.py | 25 +++++++++++-----------
1 file changed, 13 insertions(+), 12 deletions(-)
diff --git a/src/dve/core_engine/backends/base/rules.py b/src/dve/core_engine/backends/base/rules.py
index b220631..26de282 100644
--- a/src/dve/core_engine/backends/base/rules.py
+++ b/src/dve/core_engine/backends/base/rules.py
@@ -397,14 +397,11 @@ def identify_and_remove_orphans(
def process_node(
node: HierarchyNode,
parent_entity_name: Optional[EntityName],
- orph_messages: Messages | None = None,
- ):
+ processed: bool = False,
+ ) -> bool:
"""Recursive helper to process a node and its children."""
current_entity_name = node.entity_name
- if orph_messages is None:
- orph_messages = []
-
if parent_entity_name is not None:
self.logger.info(f"Identifying orphans in {current_entity_name}")
@@ -427,6 +424,7 @@ def process_node(
self.logger.info(
f"Removing records with missing parent from {current_entity_name}"
)
+ processed = True
location = list(node.join_fields.values())[0]
with BackgroundMessageWriter(
working_directory=working_directory,
@@ -466,16 +464,17 @@ def process_node(
if node.children:
for child_node in node.children:
- process_node(child_node, current_entity_name, orph_messages)
+ processed = process_node(child_node, current_entity_name, processed)
+
+ return processed
processed = False
for root_node in entity_hierarchy.entity_trees.values():
- process_node(root_node, parent_entity_name=None)
+ processed = process_node(root_node, parent_entity_name=None, processed=processed)
_orph_rel = entities.get(ORPHANED_RECORD_ENTITY_NAME)
if _orph_rel is not None:
- processed = True
del entities[ORPHANED_RECORD_ENTITY_NAME]
entities.update(entities)
@@ -496,8 +495,8 @@ def identify_and_remove_missing_mandatory_groups(
def process_node(
node: HierarchyNode,
parent_entity_name: Optional[EntityName],
- processed: Optional[bool],
- ):
+ processed: bool = False,
+ ) -> bool:
"""Recursive helper to process a node and its children."""
current_entity_name = node.entity_name
@@ -548,12 +547,14 @@ def process_node(
if node.children:
for child_node in node.children:
- process_node(child_node, current_entity_name, processed)
+ processed = process_node(child_node, current_entity_name, processed)
+
+ return processed
processed = False
for root_node in entity_hierarchy.entity_trees.values():
- process_node(root_node, parent_entity_name=None, processed=processed)
+ processed = process_node(root_node, parent_entity_name=None, processed=processed)
entities.update(entities)