Skip to content

Commit 874c030

Browse files
georgeRobertsongeorgeRobertson
committed
fix: allow group rejections to work correctly on multiple mandatory nodes
1 parent 434ca4a commit 874c030

11 files changed

Lines changed: 232 additions & 457 deletions

File tree

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

Lines changed: 47 additions & 52 deletions
Original file line numberDiff line numberDiff line change
@@ -493,67 +493,62 @@ def identify_and_remove_missing_mandatory_groups(
493493

494494
def process_node(
495495
node: HierarchyNode,
496-
parent_entity_name: Optional[EntityName],
497496
processed: bool = False,
498497
) -> bool:
499-
"""Recursive helper to process a node and its children."""
500-
current_entity_name = node.entity_name
501-
502-
if parent_entity_name is not None:
503-
self.logger.info(
504-
f"Identifying that {current_entity_name} has at least 1 valid child record"
505-
) # pylint: disable=C0301
506-
507-
join_expr = " AND ".join(
508-
f"{parent_entity_name}.{k} = {current_entity_name}.{v}"
509-
for k, v in node.join_fields.items()
498+
"""Identify at least one valid child for a mandatory entity at a given node."""
499+
if node.parent_entity is None or not node.mandatory:
500+
return processed
501+
502+
processed = True
503+
504+
self.logger.info(
505+
f"Identifying that mandatory entity `{node.parent_entity}` has at least 1 valid child record" # pylint: disable=C0301
506+
)
507+
508+
join_expr = " AND ".join(
509+
f"{node.parent_entity}.{k} = {node.entity_name}.{v}"
510+
for k, v in node.join_fields.items()
511+
)
512+
513+
with BackgroundMessageWriter(
514+
working_directory=working_directory,
515+
dve_stage=self.__stage_name__,
516+
key_fields=key_fields,
517+
logger=self.logger,
518+
) as msg_writer:
519+
location = next(iter(node.join_fields.values()))
520+
missing_children_records = self.check_mandatory_group(
521+
entities=entities,
522+
config=GroupIdentification(
523+
entity_name=node.parent_entity,
524+
target_name=node.entity_name,
525+
join_condition=join_expr,
526+
),
510527
)
511-
512-
with BackgroundMessageWriter(
513-
working_directory=working_directory,
514-
dve_stage=self.__stage_name__,
515-
key_fields=key_fields,
516-
logger=self.logger,
517-
) as msg_writer:
518-
processed = True
519-
location = next(iter(node.join_fields.values()))
520-
missing_children_records = self.check_mandatory_group(
521-
entities=entities,
522-
config=GroupIdentification(
523-
entity_name=parent_entity_name,
524-
target_name=node.entity_name,
525-
join_condition=join_expr,
526-
mandatory=node.mandatory, # type: ignore
527-
),
528+
for record in missing_children_records:
529+
msg_writer.write_queue.put(
530+
[
531+
FeedbackMessage(
532+
entity=node.parent_entity,
533+
record=record, # type: ignore
534+
error_location=location,
535+
error_message=node.no_valid_records_error_message,
536+
failure_type="record",
537+
error_type="record",
538+
error_code=node.no_valid_records_error_code,
539+
reporting_field=location,
540+
category="Children missing",
541+
)
542+
]
528543
)
529-
for record in missing_children_records:
530-
msg_writer.write_queue.put(
531-
[
532-
FeedbackMessage(
533-
entity=parent_entity_name,
534-
record=record, # type: ignore
535-
error_location=location,
536-
error_message=node.no_valid_records_error_message,
537-
failure_type="submission" if node.mandatory else "record",
538-
error_type="submission" if node.mandatory else "record",
539-
error_code=node.no_valid_records_error_code,
540-
reporting_field=location,
541-
category="Children missing",
542-
is_informational=not node.mandatory, # type: ignore
543-
)
544-
]
545-
)
546-
547-
if node.children:
548-
for child_node in node.children:
549-
processed = process_node(child_node, current_entity_name, processed)
550544

551545
return processed
552546

553547
processed = False
554548

555-
for root_node in entity_hierarchy.entity_trees.values():
556-
processed = process_node(root_node, parent_entity_name=None, processed=processed)
549+
for tree in entity_hierarchy.entity_trees.values():
550+
for node in tree.iterate_lowest_descendent_up():
551+
processed = process_node(node, processed)
557552

558553
entities.update(entities)
559554

‎src/dve/core_engine/backends/implementations/duckdb/rules.py‎

Lines changed: 1 addition & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -469,13 +469,10 @@ def check_mandatory_group(
469469
joined_rel = source_rel.join(target_rel, config.join_condition, "left").select(
470470
*source_columns,
471471
ColumnExpression(fk.strip()).alias("fk"),
472-
ConstantExpression(config.mandatory).alias("mandatory"),
473472
)
474473

475474
missing_children_rel = joined_rel.filter("fk IS NULL")
476-
filtered_rel = joined_rel.filter("fk IS NOT NULL and not mandatory").select(
477-
StarExpression(exclude=["fk", "mandatory"])
478-
)
475+
filtered_rel = joined_rel.filter("fk IS NOT NULL").select(StarExpression(exclude=["fk"]))
479476

480477
_no_valid_child_records: tuple[int] = missing_children_rel.count("*").fetchone() # type: ignore # pylint: disable=C0301
481478
if _no_valid_child_records:

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

Lines changed: 0 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -570,9 +570,6 @@ class OrphanRemoval(BaseStep):
570570
class GroupIdentification(AbstractConditionalJoin):
571571
"""Identify mandatory records which do not have any valid child records"""
572572

573-
mandatory: bool
574-
"""Whether the primary key is mandatory and whether the record should be stripped."""
575-
576573

577574
class Rule(BaseModel):
578575
"""A rule, made up of multiple steps."""

‎src/dve/pipeline/pipeline.py‎

Lines changed: 9 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -545,7 +545,7 @@ def data_contract_step(
545545

546546
return processed_files, failed_processing
547547

548-
def apply_business_rules( # pylint: disable=R0914
548+
def apply_business_rules( # pylint: disable=R0914,R0915
549549
self, submission_info: SubmissionInfo, submission_status: Optional[SubmissionStatus] = None
550550
) -> tuple[SubmissionInfo, SubmissionStatus]:
551551
"""Apply the business rules to a given submission, the submission may have failed at the
@@ -657,6 +657,14 @@ def apply_business_rules( # pylint: disable=R0914
657657
key_fields,
658658
)
659659

660+
# Perform a second time incase the mandatory groups result in new orphans
661+
_, orph_or_group = self.step_implementations.identify_and_remove_orphans( # type: ignore
662+
working_directory,
663+
entity_manager.entities,
664+
entity_hierarchy,
665+
key_fields,
666+
)
667+
660668
for entity_name, entity in entity_manager.entities.items():
661669
if orph_or_group:
662670
self._logger.info(f"Writing {entity_name} out to disk.")

‎tests/features/flights.feature‎

Lines changed: 66 additions & 62 deletions
Original file line numberDiff line numberDiff line change
@@ -22,12 +22,12 @@ Feature: Pipeline tests using the flights dataset
2222
And there are no record rejections from the business_rules phase
2323
When I run the error report phase
2424
Then An error report is produced
25-
# TODO - fix the stats calculations as they're currently incorrect for hiearchical datasets
26-
# And The statistics entry for the submission shows the following information
27-
# | parameter | value |
28-
# | record_count | 1 |
29-
# | number_file_rejections | 0 |
30-
# | number_record_rejections | 0 |
25+
And The statistics entry for the submission shows the following information
26+
| parameter | value |
27+
| record_count | 1 |
28+
| number_submission_rejections | 0 |
29+
| number_record_rejections | 0 |
30+
| number_warnings | 0 |
3131

3232
Scenario: A flights submission where the root record is rejected
3333
Given I submit the flights file missing_country_id.xml for processing
@@ -44,21 +44,24 @@ Feature: Pipeline tests using the flights dataset
4444
When I run the data contract phase
4545
Then there are no file rejections from the data_contract phase
4646
And there is 1 record rejection from the data_contract phase
47+
# And there are errors with the following details and associated error_count from the data_contract phase
48+
# | ErrorType | ErrorCode | error_count |
49+
# | record | CountryIdIsMissing | 1 |
4750
When I run the business rules phase
4851
Then there are errors with the following details and associated error_count from the business_rules phase
49-
| ErrorType | ErrorCode | error_count |
50-
| record | AG1 | 3 |
51-
| record | SG1 | 15 |
52-
| record | FG1 | 10 |
53-
| record | PG1 | 25 |
52+
| ErrorType | ErrorCode | error_count |
53+
| record | AirportHasNoCountry | 3 |
54+
| record | StaffHasNoAirport | 15 |
55+
| record | FlightHasNoAirport | 10 |
56+
| record | PassengerHasNoFlight | 25 |
5457
When I run the error report phase
5558
Then An error report is produced
56-
# TODO - fix the stats calculations as they're currently incorrect for hiearchical datasets
57-
# And The statistics entry for the submission shows the following information
58-
# | parameter | value |
59-
# | record_count | 1 |
60-
# | number_file_rejections | 0 |
61-
# | number_record_rejections | 1 |
59+
And The statistics entry for the submission shows the following information
60+
| parameter | value |
61+
| record_count | 1 |
62+
| number_submission_rejections | 0 |
63+
| number_record_rejections | 54 |
64+
| number_warnings | 0 |
6265

6366
Scenario: A flights submission where a child primary key is rejected
6467
Given I submit the flights file missing_flight_id.xml for processing
@@ -77,51 +80,47 @@ Feature: Pipeline tests using the flights dataset
7780
And there are no record rejections from the data_contract phase
7881
When I run the business rules phase
7982
Then there are errors with the following details and associated error_count from the business_rules phase
80-
| ErrorType | ErrorCode | error_count |
81-
| record | F1 | 1 |
82-
| record | PG1 | 3 |
83+
| ErrorType | ErrorCode | error_count |
84+
| record | FlightIDMissing | 1 |
85+
| record | PassengerHasNoFlight | 3 |
8386
When I run the error report phase
8487
Then An error report is produced
85-
#TODO - fix the stats calculations as they're currently incorrect for hiearchical datasets
86-
# And The statistics entry for the submission shows the following information
87-
# | parameter | value |
88-
# | record_count | 1 |
89-
# | number_file_rejections | 0 |
90-
# | number_record_rejections | 1 |
88+
And The statistics entry for the submission shows the following information
89+
| parameter | value |
90+
| record_count | 1 |
91+
| number_submission_rejections | 0 |
92+
| number_record_rejections | 4 |
93+
| number_warnings | 0 |
9194

92-
Scenario: A flights submission with a mixture of group and record rejections
93-
Given I submit the flights file mixture_of_group_rej_and_bi_rej.xml for processing
95+
Scenario: A flights submission with no valid airports record on submission
96+
Given I submit the flights file only_country_id.xml for processing
9497
And A duckdb pipeline is configured with schema file 'flights.dischema.json'
9598
And I add initial audit entries for the submission
9699
Then the latest audit record for the submission is marked with processing status file_transformation
97100
When I run the file transformation phase
98101
Then the country entity is stored as a parquet after the file_transformation phase
99102
And the airport entity is stored as a parquet after the file_transformation phase
100103
And the flights entity is stored as a parquet after the file_transformation phase
101-
And the staff entity is stored as a parquet after the file_transformation phase
102104
And the passengers entity is stored as a parquet after the file_transformation phase
103105
And the latest audit record for the submission is marked with processing status data_contract
104106
When I run the data contract phase
105107
Then there are no file rejections from the data_contract phase
106108
And there are no record rejections from the data_contract phase
107109
When I run the business rules phase
108110
Then there are errors with the following details and associated error_count from the business_rules phase
109-
| ErrorType | ErrorCode | error_count |
110-
| record | F1 | 1 |
111-
| record | PG1 | 3 |
112-
| record | P1 | 1 |
113-
| record | S1 | 7 |
111+
| ErrorType | ErrorCode | error_count |
112+
| record | CountryHasNoAirport | 1 |
114113
When I run the error report phase
115114
Then An error report is produced
116-
# TODO - fix the stats calculations as they're currently incorrect for hiearchical datasets
117-
# And The statistics entry for the submission shows the following information
118-
# | parameter | value |
119-
# | record_count | 1 |
120-
# | number_file_rejections | 0 |
121-
# | number_record_rejections | 1 |
115+
And The statistics entry for the submission shows the following information
116+
| parameter | value |
117+
| record_count | 1 |
118+
| number_submission_rejections | 0 |
119+
| number_record_rejections | 1 |
120+
| number_warnings | 0 |
122121

123-
Scenario: A flights submission with no valid airports record on submission
124-
Given I submit the flights file only_country_id.xml for processing
122+
Scenario: A flights submission with a rejection on a node with one mandatory node
123+
Given I submit the flights file singular_node_rejections.xml for processing
125124
And A duckdb pipeline is configured with schema file 'flights.dischema.json'
126125
And I add initial audit entries for the submission
127126
Then the latest audit record for the submission is marked with processing status file_transformation
@@ -136,19 +135,22 @@ Feature: Pipeline tests using the flights dataset
136135
And there are no record rejections from the data_contract phase
137136
When I run the business rules phase
138137
Then there are errors with the following details and associated error_count from the business_rules phase
139-
| ErrorType | ErrorCode | error_count |
140-
| submission | C2 | 1 |
138+
| ErrorType | Status | ErrorCode | error_count |
139+
| record | error | InvalidFlightDestination | 2 |
140+
| record | error | PassengerHasNoFlight | 4 |
141+
| record | error | AirportHasNoStaff | 1 |
142+
| record | error | CountryHasNoAirport | 1 |
141143
When I run the error report phase
142144
Then An error report is produced
143-
# TODO - fix the stats calculations as they're currently incorrect for hiearchical datasets
144-
# And The statistics entry for the submission shows the following information
145-
# | parameter | value |
146-
# | record_count | 1 |
147-
# | number_file_rejections | 0 |
148-
# | number_record_rejections | 1 |
145+
And The statistics entry for the submission shows the following information
146+
| parameter | value |
147+
| record_count | 1 |
148+
| number_submission_rejections | 0 |
149+
| number_record_rejections | 8 |
150+
| number_warnings | 0 |
149151

150-
Scenario: A flights submission with a mixture of group and orphan record rejections
151-
Given I submit the flights file invalid_flight_destination.xml for processing
152+
Scenario: A flights submission with a rejection on a node with two mandatory nodes
153+
Given I submit the flights file multi_node_file_rejection.xml for processing
152154
And A duckdb pipeline is configured with schema file 'flights.dischema.json'
153155
And I add initial audit entries for the submission
154156
Then the latest audit record for the submission is marked with processing status file_transformation
@@ -157,21 +159,23 @@ Feature: Pipeline tests using the flights dataset
157159
And the airport entity is stored as a parquet after the file_transformation phase
158160
And the flights entity is stored as a parquet after the file_transformation phase
159161
And the passengers entity is stored as a parquet after the file_transformation phase
162+
And the passengers entity is stored as a parquet after the file_transformation phase
160163
And the latest audit record for the submission is marked with processing status data_contract
161164
When I run the data contract phase
162165
Then there are no file rejections from the data_contract phase
163166
And there are no record rejections from the data_contract phase
164167
When I run the business rules phase
165168
Then there are errors with the following details and associated error_count from the business_rules phase
166-
| ErrorType | Status | ErrorCode | error_count |
167-
| record | error | F2 | 2 |
168-
| record | error | PG1 | 4 |
169-
# | record | informational | A1 | 1 |
169+
| ErrorType | Status | ErrorCode | error_count |
170+
| record | error | StaffIDMissing | 6 |
171+
| record | error | AirportHasNoStaff | 1 |
172+
| record | error | FlightHasNoAirport | 1 |
173+
| record | error | PassengerHasNoFlight | 1 |
170174
When I run the error report phase
171175
Then An error report is produced
172-
# TODO - fix the stats calculations as they're currently incorrect for hiearchical datasets
173-
# And The statistics entry for the submission shows the following information
174-
# | parameter | value |
175-
# | record_count | 1 |
176-
# | number_file_rejections | 0 |
177-
# | number_record_rejections | 1 |
176+
And The statistics entry for the submission shows the following information
177+
| parameter | value |
178+
| record_count | 1 |
179+
| number_submission_rejections | 0 |
180+
| number_record_rejections | 9 |
181+
| number_warnings | 0 |

0 commit comments

Comments
 (0)