From de08575387ee18953380f5263b9bc1a427edb547 Mon Sep 17 00:00:00 2001 From: stevenhsd <56357022+stevenhsd@users.noreply.github.com> Date: Tue, 25 Aug 2026 16:40:11 +0100 Subject: [PATCH 1/7] feature: add entity hierarchy handling (#143) * feat: added proposals for new models and objects to store entity hierarchy information * docs: added json schema for entity relationships --- .../json_schemas/dataset.schema.json | 3 + .../entity_relationships.schema.json | 37 ++ .../implementations/spark/contract.py | 5 +- .../core_engine/configuration/v1/__init__.py | 36 +- .../core_engine/configuration/v1/hierarchy.py | 131 +++++++ tests/test_core_engine/test_hierarchy.py | 359 ++++++++++++++++++ 6 files changed, 567 insertions(+), 4 deletions(-) create mode 100644 docs/advanced_guidance/json_schemas/entity_relationships.schema.json create mode 100644 src/dve/core_engine/configuration/v1/hierarchy.py create mode 100644 tests/test_core_engine/test_hierarchy.py diff --git a/docs/advanced_guidance/json_schemas/dataset.schema.json b/docs/advanced_guidance/json_schemas/dataset.schema.json index 4e85011..af8b620 100644 --- a/docs/advanced_guidance/json_schemas/dataset.schema.json +++ b/docs/advanced_guidance/json_schemas/dataset.schema.json @@ -10,6 +10,9 @@ }, "transformations": { "$ref": "transformations/transformations.schema.json" + }, + "entity_relationships": { + "$ref": "entity_relationships.schema.json" } }, "required": [ diff --git a/docs/advanced_guidance/json_schemas/entity_relationships.schema.json b/docs/advanced_guidance/json_schemas/entity_relationships.schema.json new file mode 100644 index 0000000..c570c3c --- /dev/null +++ b/docs/advanced_guidance/json_schemas/entity_relationships.schema.json @@ -0,0 +1,37 @@ +{ + "$schema": "https://json-schema.org/draft/2020-12/schema", + "$id": "data-ingest:entity_relationships.schema.json", + "title": "entity_relationships", + "description": "Description of relationships to link normalised entities back to parent entities.", + "type": "object", + "patternProperties": { + "^[A-Za-z0-9_]+.$": { + "type": "object", + "properties": { + "parent_entity": { + "type": "string" + }, + "join_fields": { + "type": "object", + "additionalProperties": { + "type": "string" + } + }, + "mandatory": { + "type": "boolean" + }, + "orphaned_records_error_code": { + "type": "string" + }, + "orphaned_records_error_message": { + "type": "string" + } + }, + "required": [ + "parent_entity", + "join_fields" + ], + "additionalProperties": false + } + } +} \ No newline at end of file diff --git a/src/dve/core_engine/backends/implementations/spark/contract.py b/src/dve/core_engine/backends/implementations/spark/contract.py index d2fd9ae..432a731 100644 --- a/src/dve/core_engine/backends/implementations/spark/contract.py +++ b/src/dve/core_engine/backends/implementations/spark/contract.py @@ -156,8 +156,9 @@ def apply_data_contract( fld, fld_info.annotation ).alias(fld) if fld in record_df.columns - else lit(None).cast( - get_type_from_annotation(fld_info.annotation)).alias(fld) + else lit(None) + .cast(get_type_from_annotation(fld_info.annotation)) + .alias(fld) ) for fld, fld_info in entity_fields.items() ], diff --git a/src/dve/core_engine/configuration/v1/__init__.py b/src/dve/core_engine/configuration/v1/__init__.py index 959596f..10e245d 100644 --- a/src/dve/core_engine/configuration/v1/__init__.py +++ b/src/dve/core_engine/configuration/v1/__init__.py @@ -1,7 +1,7 @@ """The loader for the first JSON-based dataset configuration.""" import json -from typing import Any, Optional, Union +from typing import Any, Optional, Type, Union from pydantic import BaseModel, Field, PrivateAttr, validate_call from typing_extensions import Literal @@ -22,7 +22,14 @@ ) from dve.core_engine.configuration.v1.steps import StepConfigUnion from dve.core_engine.message import DataContractErrorDetail -from dve.core_engine.type_hints import EntityName, ErrorCategory, ErrorType, TemplateVariables +from dve.core_engine.type_hints import ( + EntityName, + ErrorCategory, + ErrorCode, + ErrorMessage, + ErrorType, + TemplateVariables, +) from dve.core_engine.validation import RowValidator from dve.parser.file_handling import joinuri, open_stream, resolve_location from dve.parser.type_hints import URI, Extension @@ -38,6 +45,8 @@ FieldName = str """The name of a field within a model/schema.""" +JoinFields = Optional[dict[str, str]] +"""The fields required ( parent > child ) to join a child entity back to the parent""" TypeOrDef = Union[ # pylint: disable=C0103 TypeName, "_CallableTypeDefinition", "_ModelTypeDefinition", "_TypeAliasDefinition" ] @@ -81,6 +90,27 @@ class _TypeAliasDefinition(_BaseTypeDefintion): """The name of the Python type.""" +class _LinkageConfig(BaseModel): + """Specify how to link entities back to parents if required""" + + parent_entity: EntityName + """The name of the parent entity""" + join_fields: JoinFields + """The fields that can be used to link back to the parent entity""" + mandatory: Optional[bool] = False + """If the entity is a child, is it a mandatory field of the parent""" + no_valid_records_error_code: Optional[ErrorCode] = "NoValidRecords" + """The error code to emit if the entity has no valid records and is mandatory in the parent entity""" # pylint: disable=C0301 + no_valid_records_error_message: Optional[ErrorMessage] = ( + "parent record removed as no valid child records" + ) + """The error message to emit if the entity has no valid records and is mandatory in the parent entity""" # pylint: disable=C0301 + orphaned_records_error_code: Optional[ErrorCode] = "OrphanedRecords" + """The error code to emit if the entity contains records that are orphaned by parent record rejections""" # pylint: disable=C0301 + orphaned_records_error_message: Optional[ErrorMessage] = "Orphaned records removed" + """The error code to emit if the entity contains records that are orphaned by parent record rejections""" # pylint: disable=C0301 + + class _SchemaConfig(BaseModel): """Configuration for a component schema within a dataset.""" @@ -177,6 +207,8 @@ class V1EngineConfig(BaseEngineConfig): default_factory=dict ) """Rule store rules from the loaded rule stores.""" + entity_relationships: dict[EntityName, _LinkageConfig] = Field(default_factory=dict) + """The parent-child relationships linking the defined entities""" @validate_call def _update_rule_store(self, rule_store: dict[RuleName, BusinessComponentSpecConfigUnion]): diff --git a/src/dve/core_engine/configuration/v1/hierarchy.py b/src/dve/core_engine/configuration/v1/hierarchy.py new file mode 100644 index 0000000..5964270 --- /dev/null +++ b/src/dve/core_engine/configuration/v1/hierarchy.py @@ -0,0 +1,131 @@ +"""Classes to help determine and store entity hierarchy information.""" + +import json +from typing import Any, Iterable, Optional, Union + +from pydantic import BaseModel, Field + +from dve.core_engine.configuration.v1 import V1EngineConfig, _LinkageConfig +from dve.core_engine.type_hints import EntityName, ErrorCode, ErrorMessage +from dve.metadata_parser.exc import EntityNotFoundError +from dve.parser.file_handling.service import open_stream +from dve.parser.type_hints import URI + + +class HierarchyNode(BaseModel): + """Stores entity hierarchy information""" + + entity_name: str + children: list["HierarchyNode"] = Field(default_factory=list) + + def get_descendents(self) -> list[str]: + """Recursively list all descendents of the node""" + descendents = [] + for node in self.children: + descendents.append(node.entity_name) + descendents.extend(node.get_descendents()) + return descendents + + def get_node(self, entity_name: str) -> Union["HierarchyNode", None]: + """Recursively search for node and return if found""" + node = None + if self.entity_name == entity_name: + return self + for child in self.children: + node = child.get_node(entity_name) + if node: + break + return node + + def add_child_node(self, parent_entity: str, child_info: "HierarchyNode") -> None: + """Add a child node if the parent exists in the hierarchy""" + try: + self.get_node(parent_entity).children.append(child_info) # type: ignore + except AttributeError as exc: + raise EntityNotFoundError( + f"Can't find parent node {parent_entity} in {self.entity_name}" + ) from exc + + def as_dict(self) -> dict[str, dict[str, Any]]: + """Get dictionary representation of entity hierarchy""" + child_dict = {} + for node in self.children: + child_dict.update(node.as_dict()) + + ret_dict = self.model_dump(exclude={"entity_name", "children"}) + ret_dict.update({"children": child_dict}) + + return {self.entity_name: ret_dict} + + +class ChildHierarchyNode(HierarchyNode): + """Stores child entity hierarchy information""" + + join_fields: dict[str, str] + mandatory: Optional[bool] = False + no_valid_records_error_code: Optional[ErrorCode] = "NoValidRecords" + no_valid_records_error_message: Optional[ErrorMessage] = ( + "parent record removed as no valid child records" + ) + orphaned_records_error_code: Optional[ErrorCode] = "OrphanedRecords" + orphaned_records_error_message: Optional[ErrorMessage] = "Orphaned records removed" + + +class EntityHierarchy: + """Determines and stores entity hierarchy information from config""" + + def __init__(self, entity_trees: dict[EntityName, HierarchyNode]): + self.entity_trees = entity_trees + + @staticmethod + def determine_trees( + all_datasets: Iterable[str], entity_relationships: dict[str, _LinkageConfig] + ) -> dict[EntityName, HierarchyNode]: + """Determine the entity hierarchy trees and store as HierarchyNodes""" + top_level_parents: dict[EntityName, HierarchyNode] = { + entity_name: HierarchyNode(entity_name=entity_name) + for entity_name in all_datasets + if entity_name not in entity_relationships + } + + for name, linkage_detail in entity_relationships.items(): + for main_entity, parent_node in top_level_parents.items(): + if ( + linkage_detail.parent_entity == main_entity + or linkage_detail.parent_entity in parent_node.get_descendents() + ): + parent_node.add_child_node( + linkage_detail.parent_entity, + ChildHierarchyNode( + entity_name=name, **linkage_detail.model_dump(exclude={"parent_entity"}) + ), + ) + break + else: + raise EntityNotFoundError( + f"Can't find parent entity {linkage_detail.parent_entity} defined to " + + f"establish hierarchy for {name} - please ensure it is defined above " + + "any child entities in the dischema." + ) + return top_level_parents + + @classmethod + def from_dischema(cls, dischema_uri: URI): + """Create entity hierarchy direct from dischema""" + with open_stream(dischema_uri) as dischema: + config_dict = json.load(dischema) + all_datasets = config_dict.get("contract", {}).get("datasets", {}).keys() + entity_relationships = { + k: _LinkageConfig(**v) for k, v in config_dict.get("entity_relationships", {}).items() + } + return cls(entity_trees=cls.determine_trees(all_datasets, entity_relationships)) + + @classmethod + def from_engine_config(cls, engine_config: V1EngineConfig): + """Create entity hierarchy direct from engine config""" + return cls( + entity_trees=cls.determine_trees( + all_datasets=engine_config.contract.datasets.keys(), + entity_relationships=engine_config.entity_relationships, + ) + ) diff --git a/tests/test_core_engine/test_hierarchy.py b/tests/test_core_engine/test_hierarchy.py new file mode 100644 index 0000000..1b04faf --- /dev/null +++ b/tests/test_core_engine/test_hierarchy.py @@ -0,0 +1,359 @@ +import json +import pytest +from tempfile import NamedTemporaryFile +from dve.core_engine.configuration.v1 import V1EngineConfig +from dve.core_engine.configuration.v1.hierarchy import EntityHierarchy + +CONFIG_WITHOUT_LINKAGE = """{ + "contract": { + "schemas": {}, + "datasets": { + "animals": { + "fields": { + "name": "str", + "height": "float", + "weight": "float", + "region": "str" + }, + "reader_config": { + ".xml": { + "reader": "DuckDBXMLStreamReader", + "kwargs": { + "record_tag": "animal", + "root_tag": "animals" + } + } + }, + "mandatory_fields": [ + "name" + ] + } + } + }, + "transformations": { + "filters": [ + { + "entity": "animals", + "name": "check_valid_region", + "expression": "lower(region) in ('africa', 'asia')", + "error_code": "ANE01", + "failure_message": "Record rejected - `{{ region }}` is not in a valid region." + }, + { + "entity": "animals", + "name": "check_for_pets", + "expression": "lower(name) != 'human'", + "error_code": "ANE02", + "failure_message": "Submission Rejected - 'Human' is not a valid animal.", + "failure_type": "submission" + }, + { + "entity": "animals", + "name": "check_valid_weight", + "expression": "weight > 0", + "error_code": "ANE03", + "failure_message": "Warning - `{{ weight }}` is below zero.", + "is_informational": true + } + ] + } +}""" + +CONFIG_WITH_LINKAGE = """{ + "contract": { + "schemas": {}, + "datasets": { + "ds_001": { + "fields": { + "ds_001_id": "str", + "patient_id": "str", + "address": "str", + "name": "str" + }, + "reader_config": { + ".xml": { + "reader": "DuckDBXMLStreamReader", + "kwargs": { + "record_tag": "001", + "root_tag": "header" + } + } + }, + "mandatory_fields": [ + "ds_001_id", + "patient_id", + "address", + "name" + ] + }, + "ds_002": { + "fields": { + "ds_002_id": "str", + "gp_name": "str", + "gp_address": "str" + }, + "reader_config": { + ".xml": { + "reader": "DuckDBXMLStreamReader", + "kwargs": { + "record_tag": "002", + "root_tag": "header" + } + } + }, + "mandatory_fields": [ + "ds_002_id", + "gp_name", + "gp_address" + ] + }, + "ds_003": { + "fields": { + "ds_003_id": "str", + "ds_001_id": "str", + "total_income": "int" + }, + "reader_config": { + ".xml": { + "reader": "DuckDBXMLStreamReader", + "kwargs": { + "record_tag": "003", + "root_tag": "header" + } + } + }, + "mandatory_fields": [ + "ds_003_id", + "ds_001_id" + ] + }, + "ds_101": { + "fields": { + "ds_001_id": "str", + "referral_id": "int", + "consultant_name": "str" + }, + "reader_config": { + ".xml": { + "reader": "DuckDBXMLStreamReader", + "kwargs": { + "record_tag": "101", + "root_tag": "header" + } + } + }, + "mandatory_fields": [ + "referral_id", + "ds_001_id" + ] + }, + "ds_201": { + "fields": { + "ds_201_id": "str", + "ds_101_id": "str", + "contact_date": "date" + }, + "reader_config": { + ".xml": { + "reader": "DuckDBXMLStreamReader", + "kwargs": { + "record_tag": "201", + "root_tag": "header" + } + } + }, + "mandatory_fields": [ + "ds_101_id", + "ds_201_id", + "contact_date" + ] + }, + "ds_202": { + "fields": { + "ds_202_id": "str", + "ds_201_id": "str", + "contact_name": "str" + }, + "reader_config": { + ".xml": { + "reader": "DuckDBXMLStreamReader", + "kwargs": { + "record_tag": "202", + "root_tag": "header" + } + } + }, + "mandatory_fields": [ + "ds_202_id", + "ds_201_id" + ] + } + } + }, + "transformations": { + "filters": [ + { + "entity": "001", + "name": "check_name", + "expression": "len(name) > 2", + "error_code": "CHECK1", + "failure_message": "Record rejected - `{{ name }}` is not valid." + } + ] + }, + "entity_relationships": { + "ds_003": { + "parent_entity": "ds_001", + "join_fields": {"ds_001_id": "ds_001_id"}, + "mandatory": false, + "orphaned_records_error_code": "DS003ORPHAN", + "orphaned_records_error_message": "record removed as orphaned" + }, + "ds_101": { + "parent_entity": "ds_001", + "join_fields": {"ds_001_id": "ds_001_id"}, + "mandatory_entity": true, + "no_valid_records_error_code": "DS101NOVALIDRECS", + "no_valid_records_error_message": "{{ ds_001_id }} removed as no valid ds_101 records", + "orphaned_records_error_code": "DS101ORPHAN", + "orphaned_records_error_message": "record removed as orphaned" + }, + "ds_201": { + "parent_entity": "ds_101", + "join_fields": {"referral_id": "ds_101_id"}, + "mandatory": false, + "orphaned_records_error_code": "DS201ORPHAN", + "orphaned_records_error_message": "record removed as orphaned" + }, + "ds_202": { + "parent_entity": "ds_201", + "join_fields": {"ds_201_id": "ds_201_id"}, + "mandatory": true + } + } +}""" + +def test_no_linkage_config_load(): + config = V1EngineConfig(location="", + **json.loads(CONFIG_WITHOUT_LINKAGE)) + assert len(config.contract.datasets) == 1 + hierarchy = EntityHierarchy.from_engine_config(config) + assert len(hierarchy.entity_trees) == 1 + assert not hierarchy.entity_trees.get("animals").children + + +def test_linkage_config_load(): + config = V1EngineConfig(location="", + **json.loads(CONFIG_WITH_LINKAGE)) + assert len(config.contract.datasets) == 6 + with NamedTemporaryFile("w") as tmp: + tmp.write(CONFIG_WITH_LINKAGE) + tmp.flush() + hierarchy = EntityHierarchy.from_dischema(tmp.name) + assert len(hierarchy.entity_trees) == 2 + assert not hierarchy.entity_trees.get("ds_002").children + assert len(hierarchy.entity_trees.get("ds_001").get_descendents()) == 4 + children_001 = sorted(hierarchy.entity_trees.get("ds_001").children, key=lambda x: x.entity_name) + dict_rep_001 = hierarchy.entity_trees.get("ds_001").as_dict() + assert len(children_001) == 2 + assert children_001[0].entity_name == "ds_003" + assert not children_001[0].children + assert children_001[1].entity_name == "ds_101" + assert dict_rep_001 == json.loads(""" + { + "ds_001": { + "children": { + "ds_003": { + "join_fields": { + "ds_001_id": "ds_001_id" + }, + "mandatory": false, + "no_valid_records_error_code": "NoValidRecords", + "no_valid_records_error_message": "parent record removed as no valid child records", + "orphaned_records_error_code": "DS003ORPHAN", + "orphaned_records_error_message": "record removed as orphaned", + "children": {} + }, + "ds_101": { + "join_fields": { + "ds_001_id": "ds_001_id" + }, + "mandatory": false, + "no_valid_records_error_code": "DS101NOVALIDRECS", + "no_valid_records_error_message": "{{ ds_001_id }} removed as no valid ds_101 records", + "orphaned_records_error_code": "DS101ORPHAN", + "orphaned_records_error_message": "record removed as orphaned", + "children": { + "ds_201": { + "join_fields": { + "referral_id": "ds_101_id" + }, + "mandatory": false, + "no_valid_records_error_code": "NoValidRecords", + "no_valid_records_error_message": "parent record removed as no valid child records", + "orphaned_records_error_code": "DS201ORPHAN", + "orphaned_records_error_message": "record removed as orphaned", + "children": { + "ds_202": { + "join_fields": { + "ds_201_id": "ds_201_id" + }, + "mandatory": true, + "no_valid_records_error_code": "NoValidRecords", + "no_valid_records_error_message": "parent record removed as no valid child records", + "orphaned_records_error_code": "OrphanedRecords", + "orphaned_records_error_message": "Orphaned records removed", + "children": {} + } + } + } + } + } + } + } + }""" + ) + + dict_rep_101 = dict_rep_001["ds_001"]["children"]["ds_101"] + children_101 = children_001[1].children + assert len(children_101) == 1 + assert children_101[0].entity_name == "ds_201" + assert children_101[0].children[0].entity_name == "ds_202" + assert not children_101[0].children[0].children + assert dict_rep_101 == json.loads(""" + { + "join_fields": { + "ds_001_id": "ds_001_id" + }, + "mandatory": false, + "no_valid_records_error_code": "DS101NOVALIDRECS", + "no_valid_records_error_message": "{{ ds_001_id }} removed as no valid ds_101 records", + "orphaned_records_error_code": "DS101ORPHAN", + "orphaned_records_error_message": "record removed as orphaned", + "children": { + "ds_201": { + "join_fields": { + "referral_id": "ds_101_id" + }, + "mandatory": false, + "no_valid_records_error_code": "NoValidRecords", + "no_valid_records_error_message": "parent record removed as no valid child records", + "orphaned_records_error_code": "DS201ORPHAN", + "orphaned_records_error_message": "record removed as orphaned", + "children": { + "ds_202": { + "join_fields": { + "ds_201_id": "ds_201_id" + }, + "mandatory": true, + "no_valid_records_error_code": "NoValidRecords", + "no_valid_records_error_message": "parent record removed as no valid child records", + "orphaned_records_error_code": "OrphanedRecords", + "orphaned_records_error_message": "Orphaned records removed", + "children": {} + } + } + } + } + }""") + \ No newline at end of file From 42fb3e49b806d85eaf77dac3683a5014ef51096f Mon Sep 17 00:00:00 2001 From: georgeRobertson <50412379+georgeRobertson@users.noreply.github.com> Date: Thu, 27 Aug 2026 11:41:17 +0100 Subject: [PATCH 2/7] build: upgrade duckdb to v1.4 --- README.md | 3 +- docs/user_guidance/install.md | 17 +-- poetry.lock | 102 ++++++++---------- pyproject.toml | 2 +- .../backends/implementations/duckdb/rules.py | 2 +- 5 files changed, 60 insertions(+), 66 deletions(-) diff --git a/README.md b/README.md index e94b29e..2e7690d 100644 --- a/README.md +++ b/README.md @@ -41,7 +41,8 @@ Below is a list of features that we would like to implement or have been request | Uplift to Python 3.11 | 0.2.0 | Yes | | Uplift Pyspark to 3.5 | 0.8.0 | Yes | | Allow DVE to run on Python 3.12+ | 0.8.0 | Yes | -| Upgrade to Pydantic 2.0 | 0.9.0 | Yes | +| Upgrade to Pydantic 2.0 | 0.9.0 | Yes | +| Upgrade DuckDB to v1.4 | 0.10.0 | Yes | | Uplift Pyspark to 4.0+ | TBA | No | | Polars upgrade to v1+ | TBA | No | | DuckDB upgrade to v1.5+ | TBA | No | diff --git a/docs/user_guidance/install.md b/docs/user_guidance/install.md index 85186cd..2c86b12 100644 --- a/docs/user_guidance/install.md +++ b/docs/user_guidance/install.md @@ -78,11 +78,12 @@ Once you have installed the DVE you are almost ready to use it. To be able to ru ## DVE Version Compatability Matrix -| DVE Version | Python Version | DuckDB Version | Spark Version | Pydantic Version | -| ------------ | -------------- | -------------- | --------------- | ---------------- | -| >=0.9.0 | >=3.10,<3.13 | 1.1.3 | >=3.5.0,<=3.5.5 | 2.13.4 | -| >=0.8.0 | >=3.10,<3.13 | 1.1.3 | 3.5.2 | 1.10.19 | -| >=0.7.2 | >=3.10,<3.12 | 1.1.* | 3.4.* | 1.10.16 | -| >=0.6 | >=3.10,<3.12 | 1.1.* | 3.4.* | 1.10.15 | -| >=0.2,<0.6 | >=3.10,<3.12 | 1.1.0 | 3.4.4 | 1.10.15 | -| 0.1 | >=3.7.2,<3.8 | 1.1.0 | 3.2.1 | 1.10.15 | +| DVE Version | Python Version | DuckDB Version | Spark Version | Pydantic Version | +| ------------ | -------------- | ---------------- | --------------- | ---------------- | +| >=0.10.0 | >=3.10,<1.13 | __>=1.4,<1.4.5__ | >=3.5.0,<=3.5.5 | 2.13.4 | +| >=0.9.0 | >=3.10,<3.13 | 1.1.3 | >=3.5.0,<=3.5.5 | __2.13.4__ | +| >=0.8.0 | >=3.10,<3.13 | __1.1.3__ | __3.5.2__ | 1.10.19 | +| >=0.7.2 | >=3.10,<3.12 | 1.1.* | 3.4.* | __1.10.16__ | +| >=0.6 | >=3.10,<3.12 | __1.1.*__ | __3.4.*__ | 1.10.15 | +| >=0.2,<0.6 | __>=3.10,<3.12__ | 1.1.0 | 3.4.4 | 1.10.15 | +| 0.1 | >=3.7.2,<3.8 | 1.1.0 | 3.2.1 | 1.10.15 | diff --git a/poetry.lock b/poetry.lock index 0fedfd4..95befcd 100644 --- a/poetry.lock +++ b/poetry.lock @@ -1144,66 +1144,58 @@ files = [ [[package]] name = "duckdb" -version = "1.1.3" +version = "1.4.4" description = "DuckDB in-process database" optional = false -python-versions = ">=3.7.0" +python-versions = ">=3.9.0" groups = ["main"] files = [ - {file = "duckdb-1.1.3-cp310-cp310-macosx_12_0_arm64.whl", hash = "sha256:1c0226dc43e2ee4cc3a5a4672fddb2d76fd2cf2694443f395c02dd1bea0b7fce"}, - {file = "duckdb-1.1.3-cp310-cp310-macosx_12_0_universal2.whl", hash = "sha256:7c71169fa804c0b65e49afe423ddc2dc83e198640e3b041028da8110f7cd16f7"}, - {file = "duckdb-1.1.3-cp310-cp310-macosx_12_0_x86_64.whl", hash = "sha256:872d38b65b66e3219d2400c732585c5b4d11b13d7a36cd97908d7981526e9898"}, - {file = "duckdb-1.1.3-cp310-cp310-manylinux_2_17_aarch64.manylinux2014_aarch64.whl", hash = "sha256:25fb02629418c0d4d94a2bc1776edaa33f6f6ccaa00bd84eb96ecb97ae4b50e9"}, - {file = "duckdb-1.1.3-cp310-cp310-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:9e3f5cd604e7c39527e6060f430769b72234345baaa0987f9500988b2814f5e4"}, - {file = "duckdb-1.1.3-cp310-cp310-manylinux_2_24_aarch64.manylinux_2_28_aarch64.whl", hash = "sha256:08935700e49c187fe0e9b2b86b5aad8a2ccd661069053e38bfaed3b9ff795efd"}, - {file = "duckdb-1.1.3-cp310-cp310-musllinux_1_2_x86_64.whl", hash = "sha256:f9b47036945e1db32d70e414a10b1593aec641bd4c5e2056873d971cc21e978b"}, - {file = "duckdb-1.1.3-cp310-cp310-win_amd64.whl", hash = "sha256:35c420f58abc79a68a286a20fd6265636175fadeca1ce964fc8ef159f3acc289"}, - {file = "duckdb-1.1.3-cp311-cp311-macosx_12_0_arm64.whl", hash = "sha256:4f0e2e5a6f5a53b79aee20856c027046fba1d73ada6178ed8467f53c3877d5e0"}, - {file = "duckdb-1.1.3-cp311-cp311-macosx_12_0_universal2.whl", hash = "sha256:911d58c22645bfca4a5a049ff53a0afd1537bc18fedb13bc440b2e5af3c46148"}, - {file = "duckdb-1.1.3-cp311-cp311-macosx_12_0_x86_64.whl", hash = "sha256:c443d3d502335e69fc1e35295fcfd1108f72cb984af54c536adfd7875e79cee5"}, - {file = "duckdb-1.1.3-cp311-cp311-manylinux_2_17_aarch64.manylinux2014_aarch64.whl", hash = "sha256:0a55169d2d2e2e88077d91d4875104b58de45eff6a17a59c7dc41562c73df4be"}, - {file = "duckdb-1.1.3-cp311-cp311-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:9d0767ada9f06faa5afcf63eb7ba1befaccfbcfdac5ff86f0168c673dd1f47aa"}, - {file = "duckdb-1.1.3-cp311-cp311-manylinux_2_24_aarch64.manylinux_2_28_aarch64.whl", hash = "sha256:51c6d79e05b4a0933672b1cacd6338f882158f45ef9903aef350c4427d9fc898"}, - {file = "duckdb-1.1.3-cp311-cp311-musllinux_1_2_x86_64.whl", hash = "sha256:183ac743f21c6a4d6adfd02b69013d5fd78e5e2cd2b4db023bc8a95457d4bc5d"}, - {file = "duckdb-1.1.3-cp311-cp311-win_amd64.whl", hash = "sha256:a30dd599b8090ea6eafdfb5a9f1b872d78bac318b6914ada2d35c7974d643640"}, - {file = "duckdb-1.1.3-cp312-cp312-macosx_12_0_arm64.whl", hash = "sha256:a433ae9e72c5f397c44abdaa3c781d94f94f4065bcbf99ecd39433058c64cb38"}, - {file = "duckdb-1.1.3-cp312-cp312-macosx_12_0_universal2.whl", hash = "sha256:d08308e0a46c748d9c30f1d67ee1143e9c5ea3fbcccc27a47e115b19e7e78aa9"}, - {file = "duckdb-1.1.3-cp312-cp312-macosx_12_0_x86_64.whl", hash = "sha256:5d57776539211e79b11e94f2f6d63de77885f23f14982e0fac066f2885fcf3ff"}, - {file = "duckdb-1.1.3-cp312-cp312-manylinux_2_17_aarch64.manylinux2014_aarch64.whl", hash = "sha256:e59087dbbb63705f2483544e01cccf07d5b35afa58be8931b224f3221361d537"}, - {file = "duckdb-1.1.3-cp312-cp312-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:4ebf5f60ddbd65c13e77cddb85fe4af671d31b851f125a4d002a313696af43f1"}, - {file = "duckdb-1.1.3-cp312-cp312-manylinux_2_24_aarch64.manylinux_2_28_aarch64.whl", hash = "sha256:e4ef7ba97a65bd39d66f2a7080e6fb60e7c3e41d4c1e19245f90f53b98e3ac32"}, - {file = "duckdb-1.1.3-cp312-cp312-musllinux_1_2_x86_64.whl", hash = "sha256:f58db1b65593ff796c8ea6e63e2e144c944dd3d51c8d8e40dffa7f41693d35d3"}, - {file = "duckdb-1.1.3-cp312-cp312-win_amd64.whl", hash = "sha256:e86006958e84c5c02f08f9b96f4bc26990514eab329b1b4f71049b3727ce5989"}, - {file = "duckdb-1.1.3-cp313-cp313-macosx_12_0_arm64.whl", hash = "sha256:0897f83c09356206ce462f62157ce064961a5348e31ccb2a557a7531d814e70e"}, - {file = "duckdb-1.1.3-cp313-cp313-macosx_12_0_universal2.whl", hash = "sha256:cddc6c1a3b91dcc5f32493231b3ba98f51e6d3a44fe02839556db2b928087378"}, - {file = "duckdb-1.1.3-cp313-cp313-macosx_12_0_x86_64.whl", hash = "sha256:1d9ab6143e73bcf17d62566e368c23f28aa544feddfd2d8eb50ef21034286f24"}, - {file = "duckdb-1.1.3-cp313-cp313-manylinux_2_17_aarch64.manylinux2014_aarch64.whl", hash = "sha256:2f073d15d11a328f2e6d5964a704517e818e930800b7f3fa83adea47f23720d3"}, - {file = "duckdb-1.1.3-cp313-cp313-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:d5724fd8a49e24d730be34846b814b98ba7c304ca904fbdc98b47fa95c0b0cee"}, - {file = "duckdb-1.1.3-cp313-cp313-manylinux_2_24_aarch64.manylinux_2_28_aarch64.whl", hash = "sha256:51e7dbd968b393343b226ab3f3a7b5a68dee6d3fe59be9d802383bf916775cb8"}, - {file = "duckdb-1.1.3-cp313-cp313-musllinux_1_2_x86_64.whl", hash = "sha256:00cca22df96aa3473fe4584f84888e2cf1c516e8c2dd837210daec44eadba586"}, - {file = "duckdb-1.1.3-cp313-cp313-win_amd64.whl", hash = "sha256:77f26884c7b807c7edd07f95cf0b00e6d47f0de4a534ac1706a58f8bc70d0d31"}, - {file = "duckdb-1.1.3-cp37-cp37m-manylinux_2_17_aarch64.manylinux2014_aarch64.whl", hash = "sha256:a4748635875fc3c19a7320a6ae7410f9295557450c0ebab6d6712de12640929a"}, - {file = "duckdb-1.1.3-cp37-cp37m-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:b74e121ab65dbec5290f33ca92301e3a4e81797966c8d9feef6efdf05fc6dafd"}, - {file = "duckdb-1.1.3-cp37-cp37m-manylinux_2_24_aarch64.manylinux_2_28_aarch64.whl", hash = "sha256:9c619e4849837c8c83666f2cd5c6c031300cd2601e9564b47aa5de458ff6e69d"}, - {file = "duckdb-1.1.3-cp37-cp37m-win_amd64.whl", hash = "sha256:0ba6baa0af33ded836b388b09433a69b8bec00263247f6bf0a05c65c897108d3"}, - {file = "duckdb-1.1.3-cp38-cp38-macosx_12_0_arm64.whl", hash = "sha256:ecb1dc9062c1cc4d2d88a5e5cd8cc72af7818ab5a3c0f796ef0ffd60cfd3efb4"}, - {file = "duckdb-1.1.3-cp38-cp38-macosx_12_0_universal2.whl", hash = "sha256:5ace6e4b1873afdd38bd6cc8fcf90310fb2d454f29c39a61d0c0cf1a24ad6c8d"}, - {file = "duckdb-1.1.3-cp38-cp38-macosx_12_0_x86_64.whl", hash = "sha256:a1fa0c502f257fa9caca60b8b1478ec0f3295f34bb2efdc10776fc731b8a6c5f"}, - {file = "duckdb-1.1.3-cp38-cp38-manylinux_2_17_aarch64.manylinux2014_aarch64.whl", hash = "sha256:6411e21a2128d478efbd023f2bdff12464d146f92bc3e9c49247240448ace5a6"}, - {file = "duckdb-1.1.3-cp38-cp38-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:c5336939d83837af52731e02b6a78a446794078590aa71fd400eb17f083dda3e"}, - {file = "duckdb-1.1.3-cp38-cp38-manylinux_2_24_aarch64.manylinux_2_28_aarch64.whl", hash = "sha256:f549af9f7416573ee48db1cf8c9d27aeed245cb015f4b4f975289418c6cf7320"}, - {file = "duckdb-1.1.3-cp38-cp38-win_amd64.whl", hash = "sha256:2141c6b28162199999075d6031b5d63efeb97c1e68fb3d797279d31c65676269"}, - {file = "duckdb-1.1.3-cp39-cp39-macosx_12_0_arm64.whl", hash = "sha256:09c68522c30fc38fc972b8a75e9201616b96ae6da3444585f14cf0d116008c95"}, - {file = "duckdb-1.1.3-cp39-cp39-macosx_12_0_universal2.whl", hash = "sha256:8ee97ec337794c162c0638dda3b4a30a483d0587deda22d45e1909036ff0b739"}, - {file = "duckdb-1.1.3-cp39-cp39-macosx_12_0_x86_64.whl", hash = "sha256:a1f83c7217c188b7ab42e6a0963f42070d9aed114f6200e3c923c8899c090f16"}, - {file = "duckdb-1.1.3-cp39-cp39-manylinux_2_17_aarch64.manylinux2014_aarch64.whl", hash = "sha256:1aa3abec8e8995a03ff1a904b0e66282d19919f562dd0a1de02f23169eeec461"}, - {file = "duckdb-1.1.3-cp39-cp39-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:80158f4c7c7ada46245837d5b6869a336bbaa28436fbb0537663fa324a2750cd"}, - {file = "duckdb-1.1.3-cp39-cp39-manylinux_2_24_aarch64.manylinux_2_28_aarch64.whl", hash = "sha256:647f17bd126170d96a38a9a6f25fca47ebb0261e5e44881e3782989033c94686"}, - {file = "duckdb-1.1.3-cp39-cp39-musllinux_1_2_x86_64.whl", hash = "sha256:252d9b17d354beb9057098d4e5d5698e091a4f4a0d38157daeea5fc0ec161670"}, - {file = "duckdb-1.1.3-cp39-cp39-win_amd64.whl", hash = "sha256:eeacb598120040e9591f5a4edecad7080853aa8ac27e62d280f151f8c862afa3"}, - {file = "duckdb-1.1.3.tar.gz", hash = "sha256:68c3a46ab08836fe041d15dcbf838f74a990d551db47cb24ab1c4576fc19351c"}, + {file = "duckdb-1.4.4-cp310-cp310-macosx_10_9_universal2.whl", hash = "sha256:e870a441cb1c41d556205deb665749f26347ed13b3a247b53714f5d589596977"}, + {file = "duckdb-1.4.4-cp310-cp310-macosx_10_9_x86_64.whl", hash = "sha256:49123b579e4a6323e65139210cd72dddc593a72d840211556b60f9703bda8526"}, + {file = "duckdb-1.4.4-cp310-cp310-macosx_11_0_arm64.whl", hash = "sha256:5e1933fac5293fea5926b0ee75a55b8cfe7f516d867310a5b251831ab61fe62b"}, + {file = "duckdb-1.4.4-cp310-cp310-manylinux_2_26_aarch64.manylinux_2_28_aarch64.whl", hash = "sha256:707530f6637e91dc4b8125260595299ec9dd157c09f5d16c4186c5988bfbd09a"}, + {file = "duckdb-1.4.4-cp310-cp310-manylinux_2_26_x86_64.manylinux_2_28_x86_64.whl", hash = "sha256:453b115f4777467f35103d8081770ac2f223fb5799178db5b06186e3ab51d1f2"}, + {file = "duckdb-1.4.4-cp310-cp310-win_amd64.whl", hash = "sha256:a3c8542db7ffb128aceb7f3b35502ebaddcd4f73f1227569306cc34bad06680c"}, + {file = "duckdb-1.4.4-cp311-cp311-macosx_10_9_universal2.whl", hash = "sha256:5ba684f498d4e924c7e8f30dd157da8da34c8479746c5011b6c0e037e9c60ad2"}, + {file = "duckdb-1.4.4-cp311-cp311-macosx_10_9_x86_64.whl", hash = "sha256:5536eb952a8aa6ae56469362e344d4e6403cc945a80bc8c5c2ebdd85d85eb64b"}, + {file = "duckdb-1.4.4-cp311-cp311-macosx_11_0_arm64.whl", hash = "sha256:47dd4162da6a2be59a0aef640eb08d6360df1cf83c317dcc127836daaf3b7f7c"}, + {file = "duckdb-1.4.4-cp311-cp311-manylinux_2_26_aarch64.manylinux_2_28_aarch64.whl", hash = "sha256:6cb357cfa3403910e79e2eb46c8e445bb1ee2fd62e9e9588c6b999df4256abc1"}, + {file = "duckdb-1.4.4-cp311-cp311-manylinux_2_26_x86_64.manylinux_2_28_x86_64.whl", hash = "sha256:4c25d5b0febda02b7944e94fdae95aecf952797afc8cb920f677b46a7c251955"}, + {file = "duckdb-1.4.4-cp311-cp311-win_amd64.whl", hash = "sha256:6703dd1bb650025b3771552333d305d62ddd7ff182de121483d4e042ea6e2e00"}, + {file = "duckdb-1.4.4-cp311-cp311-win_arm64.whl", hash = "sha256:bf138201f56e5d6fc276a25138341b3523e2f84733613fc43f02c54465619a95"}, + {file = "duckdb-1.4.4-cp312-cp312-macosx_10_13_universal2.whl", hash = "sha256:ddcfd9c6ff234da603a1edd5fd8ae6107f4d042f74951b65f91bc5e2643856b3"}, + {file = "duckdb-1.4.4-cp312-cp312-macosx_10_13_x86_64.whl", hash = "sha256:6792ca647216bd5c4ff16396e4591cfa9b4a72e5ad7cdd312cec6d67e8431a7c"}, + {file = "duckdb-1.4.4-cp312-cp312-macosx_11_0_arm64.whl", hash = "sha256:1f8d55843cc940e36261689054f7dfb6ce35b1f5b0953b0d355b6adb654b0d52"}, + {file = "duckdb-1.4.4-cp312-cp312-manylinux_2_26_aarch64.manylinux_2_28_aarch64.whl", hash = "sha256:c65d15c440c31e06baaebfd2c06d71ce877e132779d309f1edf0a85d23c07e92"}, + {file = "duckdb-1.4.4-cp312-cp312-manylinux_2_26_x86_64.manylinux_2_28_x86_64.whl", hash = "sha256:b297eff642503fd435a9de5a9cb7db4eccb6f61d61a55b30d2636023f149855f"}, + {file = "duckdb-1.4.4-cp312-cp312-win_amd64.whl", hash = "sha256:d525de5f282b03aa8be6db86b1abffdceae5f1055113a03d5b50cd2fb8cf2ef8"}, + {file = "duckdb-1.4.4-cp312-cp312-win_arm64.whl", hash = "sha256:50f2eb173c573811b44aba51176da7a4e5c487113982be6a6a1c37337ec5fa57"}, + {file = "duckdb-1.4.4-cp313-cp313-macosx_10_13_universal2.whl", hash = "sha256:337f8b24e89bc2e12dadcfe87b4eb1c00fd920f68ab07bc9b70960d6523b8bc3"}, + {file = "duckdb-1.4.4-cp313-cp313-macosx_10_13_x86_64.whl", hash = "sha256:0509b39ea7af8cff0198a99d206dca753c62844adab54e545984c2e2c1381616"}, + {file = "duckdb-1.4.4-cp313-cp313-macosx_11_0_arm64.whl", hash = "sha256:fb94de6d023de9d79b7edc1ae07ee1d0b4f5fa8a9dcec799650b5befdf7aafec"}, + {file = "duckdb-1.4.4-cp313-cp313-manylinux_2_26_aarch64.manylinux_2_28_aarch64.whl", hash = "sha256:0d636ceda422e7babd5e2f7275f6a0d1a3405e6a01873f00d38b72118d30c10b"}, + {file = "duckdb-1.4.4-cp313-cp313-manylinux_2_26_x86_64.manylinux_2_28_x86_64.whl", hash = "sha256:7df7351328ffb812a4a289732f500d621e7de9942a3a2c9b6d4afcf4c0e72526"}, + {file = "duckdb-1.4.4-cp313-cp313-win_amd64.whl", hash = "sha256:6fb1225a9ea5877421481d59a6c556a9532c32c16c7ae6ca8d127e2b878c9389"}, + {file = "duckdb-1.4.4-cp313-cp313-win_arm64.whl", hash = "sha256:f28a18cc790217e5b347bb91b2cab27aafc557c58d3d8382e04b4fe55d0c3f66"}, + {file = "duckdb-1.4.4-cp314-cp314-macosx_10_15_universal2.whl", hash = "sha256:25874f8b1355e96178079e37312c3ba6d61a2354f51319dae860cf21335c3a20"}, + {file = "duckdb-1.4.4-cp314-cp314-macosx_10_15_x86_64.whl", hash = "sha256:452c5b5d6c349dc5d1154eb2062ee547296fcbd0c20e9df1ed00b5e1809089da"}, + {file = "duckdb-1.4.4-cp314-cp314-macosx_11_0_arm64.whl", hash = "sha256:8e5c2d8a0452df55e092959c0bfc8ab8897ac3ea0f754cb3b0ab3e165cd79aff"}, + {file = "duckdb-1.4.4-cp314-cp314-manylinux_2_26_aarch64.manylinux_2_28_aarch64.whl", hash = "sha256:1af6e76fe8bd24875dc56dd8e38300d64dc708cd2e772f67b9fbc635cc3066a3"}, + {file = "duckdb-1.4.4-cp314-cp314-manylinux_2_26_x86_64.manylinux_2_28_x86_64.whl", hash = "sha256:d0440f59e0cd9936a9ebfcf7a13312eda480c79214ffed3878d75947fc3b7d6d"}, + {file = "duckdb-1.4.4-cp314-cp314-win_amd64.whl", hash = "sha256:59c8d76016dde854beab844935b1ec31de358d4053e792988108e995b18c08e7"}, + {file = "duckdb-1.4.4-cp314-cp314-win_arm64.whl", hash = "sha256:53cd6423136ab44383ec9955aefe7599b3fb3dd1fe006161e6396d8167e0e0d4"}, + {file = "duckdb-1.4.4-cp39-cp39-macosx_10_9_universal2.whl", hash = "sha256:8097201bc5fd0779d7fcc2f3f4736c349197235f4cb7171622936343a1aa8dbf"}, + {file = "duckdb-1.4.4-cp39-cp39-macosx_10_9_x86_64.whl", hash = "sha256:cd1be3d48577f5b40eb9706c6b2ae10edfe18e78eb28e31a3b922dcff1183597"}, + {file = "duckdb-1.4.4-cp39-cp39-macosx_11_0_arm64.whl", hash = "sha256:e041f2fbd6888da090eca96ac167a7eb62d02f778385dd9155ed859f1c6b6dc8"}, + {file = "duckdb-1.4.4-cp39-cp39-manylinux_2_26_aarch64.manylinux_2_28_aarch64.whl", hash = "sha256:7eec0bf271ac622e57b7f6554a27a6e7d1dd2f43d1871f7962c74bcbbede15ba"}, + {file = "duckdb-1.4.4-cp39-cp39-manylinux_2_26_x86_64.manylinux_2_28_x86_64.whl", hash = "sha256:5cdc4126ec925edf3112bc656ac9ed23745294b854935fa7a643a216e4455af6"}, + {file = "duckdb-1.4.4-cp39-cp39-win_amd64.whl", hash = "sha256:c9566a4ed834ec7999db5849f53da0a7ee83d86830c33f471bf0211a1148ca12"}, + {file = "duckdb-1.4.4.tar.gz", hash = "sha256:8bba52fd2acb67668a4615ee17ee51814124223de836d9e2fdcbc4c9021b3d3c"}, ] +[package.extras] +all = ["adbc-driver-manager", "fsspec", "ipython", "numpy", "pandas", "pyarrow"] + [[package]] name = "et-xmlfile" version = "2.0.0" @@ -3649,4 +3641,4 @@ type = ["pytest-mypy (>=1.0.1) ; platform_python_implementation != \"PyPy\""] [metadata] lock-version = "2.1" python-versions = ">=3.10,<3.13" -content-hash = "3c6b964ad86fe375ec1862480b207189085a8d1da8e5017a8545b78dc0ff469b" +content-hash = "a2df40ebbf2383a36c3031ca7d04330dd8e60130b291cc136d25c22647e13716" diff --git a/pyproject.toml b/pyproject.toml index 0ad62d3..40fc9a0 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -35,7 +35,7 @@ python = ">=3.10,<3.13" # breaking changes beyond 3.12 boto3 = ">=1.34.162,<1.36" # breaking change beyond 1.36 botocore = ">=1.34.162,<1.36" # breaking change beyond 1.36 delta-spark = ">=3.0.0,<=3.2.0" -duckdb = "1.1.3" # breaking changes beyond 1.1 +duckdb = ">=1.4,<1.4.5" Jinja2 = "3.1.6" lxml = "6.1.1" numpy = "1.26.4" diff --git a/src/dve/core_engine/backends/implementations/duckdb/rules.py b/src/dve/core_engine/backends/implementations/duckdb/rules.py index dc73dad..c4277d9 100644 --- a/src/dve/core_engine/backends/implementations/duckdb/rules.py +++ b/src/dve/core_engine/backends/implementations/duckdb/rules.py @@ -364,7 +364,7 @@ def join_header(self, entities: DuckDBEntities, *, config: HeaderJoin) -> Messag ), ) - target_schema = DDBStruct(dict(zip(target_rel.columns, target_rel.dtypes)))() + target_schema = DDBStruct(dict(zip(target_rel.columns, target_rel.dtypes)))() # type: ignore # pylint:disable=C0301 joined_rel = source_rel.select( StarExpression(exclude=[]), From a9c0872ac3bfd61e5a668b46f2a17019f652d1c2 Mon Sep 17 00:00:00 2001 From: georgeRobertson <50412379+georgeRobertson@users.noreply.github.com> Date: Thu, 27 Aug 2026 12:42:49 +0100 Subject: [PATCH 3/7] docs: add dev classifier --- pyproject.toml | 1 + 1 file changed, 1 insertion(+) diff --git a/pyproject.toml b/pyproject.toml index 40fc9a0..ddfd701 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -7,6 +7,7 @@ authors = [ ] readme = "README.md" classifiers = [ + "Development Status :: 4 - Beta", "Programming Language :: Python :: 3", "Programming Language :: Python :: 3.10", "Programming Language :: Python :: 3.11", From a3cafe737c360857c136afb219dff1146cfa59d6 Mon Sep 17 00:00:00 2001 From: George Robertson <50412379+georgeRobertson@users.noreply.github.com> Date: Fri, 11 Sep 2026 14:43:39 +0100 Subject: [PATCH 4/7] feat: add orphan record identification and removal (#150) --- src/dve/core_engine/backends/base/rules.py | 122 +++++++++++- .../backends/implementations/duckdb/rules.py | 80 +++++--- .../backends/implementations/spark/rules.py | 17 +- .../core_engine/backends/metadata/rules.py | 8 +- .../core_engine/configuration/v1/hierarchy.py | 10 +- src/dve/core_engine/constants.py | 3 + src/dve/core_engine/type_hints.py | 2 +- src/dve/pipeline/pipeline.py | 15 +- tests/features/flights.feature | 116 +++++++++++ .../test_duckdb/test_rules.py | 187 ++++++++++-------- .../test_spark/test_rules.py | 3 + tests/testdata/flights/flights.dischema.json | 149 ++++++++++++++ tests/testdata/flights/missing_country_id.xml | 48 +++++ tests/testdata/flights/missing_flight_id.xml | 48 +++++ .../mixture_of_group_rej_and_bi_rej.xml | 53 +++++ tests/testdata/flights/perfect_flights.xml | 49 +++++ 16 files changed, 785 insertions(+), 125 deletions(-) create mode 100644 tests/features/flights.feature create mode 100644 tests/testdata/flights/flights.dischema.json create mode 100644 tests/testdata/flights/missing_country_id.xml create mode 100644 tests/testdata/flights/missing_flight_id.xml create mode 100644 tests/testdata/flights/mixture_of_group_rej_and_bi_rej.xml create mode 100644 tests/testdata/flights/perfect_flights.xml diff --git a/src/dve/core_engine/backends/base/rules.py b/src/dve/core_engine/backends/base/rules.py index 9b6b4fe..1340e32 100644 --- a/src/dve/core_engine/backends/base/rules.py +++ b/src/dve/core_engine/backends/base/rules.py @@ -3,7 +3,7 @@ import logging from abc import ABC, abstractmethod from collections import defaultdict -from collections.abc import Iterable +from collections.abc import Iterable, Iterator from typing import Any, ClassVar, Generic, NoReturn, Optional, TypeVar from uuid import uuid4 @@ -17,6 +17,7 @@ ) from dve.core_engine.backends.base.core import get_entity_type from dve.core_engine.backends.exceptions import render_error +from dve.core_engine.backends.metadata.reporting import ReportingConfig from dve.core_engine.backends.metadata.rules import ( AbstractStep, Aggregation, @@ -34,6 +35,7 @@ Notification, OneToOneJoin, OrphanIdentification, + OrphanRemoval, ParentMetadata, RenameEntity, Rule, @@ -43,8 +45,15 @@ TableUnion, ) from dve.core_engine.backends.types import Entities, EntityType, StageSuccessful +from dve.core_engine.configuration.v1.hierarchy import ( + ChildHierarchyNode, + EntityHierarchy, + HierarchyNode, +) +from dve.core_engine.constants import ORPHANED_RECORD_ENTITY_NAME from dve.core_engine.exceptions import CriticalProcessingError from dve.core_engine.loggers import get_logger +from dve.core_engine.message import FeedbackMessage from dve.core_engine.type_hints import URI, DVEStageName, EntityName, Messages, TemplateVariables T_contra = TypeVar("T_contra", bound=AbstractStep, contravariant=True) @@ -307,7 +316,10 @@ def join_header(self, entities: Entities, *, config: HeaderJoin) -> Messages: """ raise NotImplementedError - def identify_orphans(self, entities: Entities, *, config: OrphanIdentification) -> Messages: + @abstractmethod + def identify_orphans( + self, entities: Entities, *, config: OrphanIdentification + ) -> tuple[Messages, int]: """Identify records in an entity which don't have at least one corresponding match in the target. A new boolean column will be added to `entity` ('IsOrphaned') indicating whether the condition matched. @@ -320,6 +332,18 @@ def identify_orphans(self, entities: Entities, *, config: OrphanIdentification) """ raise NotImplementedError + @abstractmethod + def remove_orphans(self, entities: Entities, *, config: OrphanRemoval) -> Iterator: + """ + Remove orphaned records from an entity based on the orphans found in + identify_orphans method. Returns a generator objects with the records removed + for generating feedback messages from. + + This may not be implemented by some backends. + + """ + raise NotImplementedError + @abstractmethod def union(self, entities: Entities, *, config: TableUnion) -> Messages: """Union two entities together, taking the columns from each by name. @@ -352,6 +376,100 @@ def notify(self, entities: Entities, *, config: Notification) -> Messages: """ + def identify_and_remove_orphans( + self, + working_directory: URI, + entities: Entities, + entity_hierarchy: EntityHierarchy, + key_fields: Optional[dict[str, list[str]]] = None, + ) -> Messages: + """ + 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. + Processes recursively: removes orphans at each level, then processes children. + """ + + def process_node( + node: HierarchyNode | ChildHierarchyNode, + parent_entity_name: Optional[EntityName], + orph_messages: Messages | None = None, + ): + """Recursive helper to process a node and its children.""" + current_entity_name = node.entity_name + + if orph_messages is None: + orph_messages = [] + + 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() + ) + + _, no_orphs = self.identify_orphans( + entities=entities, + config=OrphanIdentification( + id=list(node.join_fields.values())[0], + entity_name=current_entity_name, + target_name=parent_entity_name, + join_condition=join_expr, + ), + ) + + if no_orphs > 0: + self.logger.info(f"Removing orphan records from {current_entity_name}") + location = list(node.join_fields.values())[0] + with BackgroundMessageWriter( + working_directory=working_directory, + dve_stage=self.__stage_name__, + key_fields=key_fields, + logger=self.logger, + ) as msg_writer: + _orph_records = self.remove_orphans( + entities=entities, + config=OrphanRemoval( + entity_name=current_entity_name, + reporting=ReportingConfig( + emit="record_failure", + 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", + ) + ]) + + 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: + del entities[ORPHANED_RECORD_ENTITY_NAME] + + entities.update(entities) + + return [] + # pylint: disable=R0912,R0914 def apply_sync_filters( self, diff --git a/src/dve/core_engine/backends/implementations/duckdb/rules.py b/src/dve/core_engine/backends/implementations/duckdb/rules.py index c4277d9..4479846 100644 --- a/src/dve/core_engine/backends/implementations/duckdb/rules.py +++ b/src/dve/core_engine/backends/implementations/duckdb/rules.py @@ -1,6 +1,6 @@ """Business rule definitions for duckdb backend""" -from collections.abc import Callable +from collections.abc import Callable, Iterator from typing import get_type_hints from uuid import uuid4 @@ -50,9 +50,11 @@ Notification, OneToOneJoin, OrphanIdentification, + OrphanRemoval, SemiJoin, TableUnion, ) +from dve.core_engine.constants import ORPHANED_RECORD_ENTITY_NAME, RECORD_INDEX_COLUMN_NAME from dve.core_engine.functions import implementations as functions from dve.core_engine.message import FeedbackMessage from dve.core_engine.templating import template_object @@ -375,8 +377,11 @@ def join_header(self, entities: DuckDBEntities, *, config: HeaderJoin) -> Messag return [] def identify_orphans( - self, entities: DuckDBEntities, *, config: OrphanIdentification - ) -> Messages: + self, + entities: DuckDBEntities, + *, + config: OrphanIdentification, + ) -> tuple[Messages, int]: """Identify records in an entity which don't have at least one corresponding match in the target. A new boolean column will be added to `entity` ('IsOrphaned') indicating whether the condition matched. @@ -390,41 +395,60 @@ def identify_orphans( target_rel: DuckDBPyRelation = entities[config.target_name] target_rel = target_rel.set_alias(config.target_name) - key_name = f"key_{uuid4().hex}" - source_rel = source_rel.select(f"*, row_number() over () as {key_name}").set_alias( - config.entity_name - ) match_name = f"matched_{uuid4().hex}" target_rel = target_rel.select( StarExpression(exclude=[]), ConstantExpression(1).alias(match_name) ).set_alias(config.target_name) - joined_rel: DuckDBPyRelation = source_rel.join( - target_rel, condition=config.join_condition, how="left" - ).aggregate(f"{key_name}, coalesce(count({match_name})==0, TRUE) AS IsOrphaned") + pk, _fk = config.join_condition.split("=") - if "IsOrphaned" not in source_rel.columns: - result: DuckDBPyRelation = source_rel.join( - joined_rel, condition=key_name, how="left" - ).select(StarExpression(exclude=[key_name])) - else: - result = source_rel.set_alias("source").join( - joined_rel.set_alias("joined"), - condition=f"source.{key_name} = joined.{key_name}", - how="left", + orphaned_rel: DuckDBPyRelation = ( + source_rel.join(target_rel, condition=config.join_condition, how="left") + .aggregate( + f"{config.entity_name}.{RECORD_INDEX_COLUMN_NAME}, {config.entity_name}.{config.id}, coalesce(count({match_name}), 0)==0 AS IsOrphaned" # pylint: disable=C0301 ) + .filter("IsOrphaned") + .select( + RECORD_INDEX_COLUMN_NAME, + ConstantExpression(config.entity_name).alias("entity_name"), + ConstantExpression(pk.strip().rsplit(".")[1]).alias("pk"), + ColumnExpression(config.id).alias("pk_value"), # type: ignore + ) + .unique("*") + ) + _orph_records: tuple[int] = orphaned_rel.count(RECORD_INDEX_COLUMN_NAME).fetchone() # type: ignore # pylint: disable=C0301 + if _orph_records: + _no_orphans = _orph_records[0] + else: + _no_orphans = 0 + self.logger.info(f"Found {_no_orphans} orphaned records in {config.entity_name}.") - columns = {name: f"source.{name}" for name in source_rel.columns} - if "IsOrphaned" in source_rel.columns: - columns["IsOrphaned"] = ColumnExpression("source.IsOrphaned") | ColumnExpression("joined.IsOrphaned") # type: ignore # pylint: disable=line-too-long - columns.pop(key_name, None) - - result = result.select( - ",".join([f"{column} as {name}" for name, column in columns.items()]) + if entities.get(ORPHANED_RECORD_ENTITY_NAME) is not None: + entities[ORPHANED_RECORD_ENTITY_NAME] = entities[ORPHANED_RECORD_ENTITY_NAME].union( + orphaned_rel ) + else: + entities[ORPHANED_RECORD_ENTITY_NAME] = orphaned_rel + return [], _no_orphans + + def remove_orphans(self, entities: DuckDBEntities, *, config: OrphanRemoval) -> Iterator: + """Method to remove identified orphans in the orphan tracker entity.""" + orphan_rel = entities[ORPHANED_RECORD_ENTITY_NAME].set_alias("orphan") + filtered_rel = ( + entities[config.entity_name] + .set_alias(config.entity_name) + .join( + orphan_rel, + f"{config.entity_name}.{RECORD_INDEX_COLUMN_NAME} = orphan.{RECORD_INDEX_COLUMN_NAME}", # pylint: disable=C0301 + "anti", + ) + ) - entities[config.new_entity_name or config.entity_name] = result - return [] + entities[config.entity_name] = filtered_rel + + return duckdb_rel_to_dictionaries( + orphan_rel.filter(f"entity_name = '{config.entity_name}'") + ) 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 825ee15..ff15b52 100644 --- a/src/dve/core_engine/backends/implementations/spark/rules.py +++ b/src/dve/core_engine/backends/implementations/spark/rules.py @@ -1,6 +1,6 @@ """Step implementations in Spark.""" -from collections.abc import Callable +from collections.abc import Callable, Iterator from typing import Optional from uuid import uuid4 @@ -41,6 +41,7 @@ Notification, OneToOneJoin, OrphanIdentification, + OrphanRemoval, SelectColumns, SemiJoin, TableUnion, @@ -338,7 +339,8 @@ def union(self, entities: SparkEntities, *, config: TableUnion) -> Messages: def identify_orphans( self, entities: SparkEntities, *, config: OrphanIdentification - ) -> Messages: + ) -> tuple[Messages, int]: + # TODO - adjust this to new setup of identify and remove orphans source_df: DataFrame = entities[config.entity_name] source_df = source_df.alias(config.entity_name) target_df: DataFrame = entities[config.target_name] @@ -371,7 +373,16 @@ def identify_orphans( result = result.select(*[column.alias(name) for name, column in columns.items()]) entities[config.new_entity_name or config.entity_name] = result - return [] + return [], 0 + + def remove_orphans( + self, + entities: SparkEntities, + *, + config: OrphanRemoval, + ) -> 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 f3a6305..9b96a14 100644 --- a/src/dve/core_engine/backends/metadata/rules.py +++ b/src/dve/core_engine/backends/metadata/rules.py @@ -553,11 +553,17 @@ class OrphanIdentification(AbstractConditionalJoin): """ - Step = Union[AbstractStep, Literal["sync"]] """A step within a rule. This is either a rule config or the literal string 'sync'.""" +class OrphanRemoval(BaseStep): + """Remove an orphan record from the `entity`.""" + + reporting: ReportingConfig + """The reporting information for the row removal.""" + + class Rule(BaseModel): """A rule, made up of multiple steps.""" diff --git a/src/dve/core_engine/configuration/v1/hierarchy.py b/src/dve/core_engine/configuration/v1/hierarchy.py index 5964270..15dedcf 100644 --- a/src/dve/core_engine/configuration/v1/hierarchy.py +++ b/src/dve/core_engine/configuration/v1/hierarchy.py @@ -16,12 +16,12 @@ class HierarchyNode(BaseModel): """Stores entity hierarchy information""" entity_name: str - children: list["HierarchyNode"] = Field(default_factory=list) + children: Optional[list["ChildHierarchyNode"]] = Field(default_factory=list) def get_descendents(self) -> list[str]: """Recursively list all descendents of the node""" descendents = [] - for node in self.children: + for node in self.children: # type: ignore descendents.append(node.entity_name) descendents.extend(node.get_descendents()) return descendents @@ -31,7 +31,7 @@ def get_node(self, entity_name: str) -> Union["HierarchyNode", None]: node = None if self.entity_name == entity_name: return self - for child in self.children: + for child in self.children: # type: ignore node = child.get_node(entity_name) if node: break @@ -48,8 +48,8 @@ def add_child_node(self, parent_entity: str, child_info: "HierarchyNode") -> Non def as_dict(self) -> dict[str, dict[str, Any]]: """Get dictionary representation of entity hierarchy""" - child_dict = {} - for node in self.children: + child_dict: dict[str, dict[str, Any]] = {} + for node in self.children: # type: ignore child_dict.update(node.as_dict()) ret_dict = self.model_dump(exclude={"entity_name", "children"}) diff --git a/src/dve/core_engine/constants.py b/src/dve/core_engine/constants.py index a2a4a65..7afcae7 100644 --- a/src/dve/core_engine/constants.py +++ b/src/dve/core_engine/constants.py @@ -6,3 +6,6 @@ CONTRACT_ERROR_VALUE_FIELD_NAME: str = "__error_value" """The name of the field that can be used to extract the field value that caused a pydantic validation error""" + +ORPHANED_RECORD_ENTITY_NAME: str = "orphaned_records_tracker" +"""Name to keep track of identified orphaned records""" diff --git a/src/dve/core_engine/type_hints.py b/src/dve/core_engine/type_hints.py index 154ada6..e369ff4 100644 --- a/src/dve/core_engine/type_hints.py +++ b/src/dve/core_engine/type_hints.py @@ -133,7 +133,7 @@ """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"] +ErrorCategory = Literal["Blank", "Wrong format", "Bad value", "Bad file", "Parent 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 a9be3ff..eaf7661 100644 --- a/src/dve/pipeline/pipeline.py +++ b/src/dve/pipeline/pipeline.py @@ -34,6 +34,7 @@ from dve.core_engine.backends.readers.utilities import get_all_model_fields from dve.core_engine.backends.types import EntityType from dve.core_engine.backends.utilities import stringify_model +from dve.core_engine.configuration.v1.hierarchy import EntityHierarchy from dve.core_engine.exceptions import CriticalProcessingError from dve.core_engine.loggers import get_logger from dve.core_engine.message import FeedbackMessage @@ -596,8 +597,13 @@ def apply_business_rules( # pylint: disable=R0914 key_fields = {model: conf.reporting_fields for model, conf in model_config.items()} + entity_hierarchy = EntityHierarchy.from_engine_config(config) + _errors_uri, rules_success = self.step_implementations.apply_rules( # type: ignore - working_directory, entity_manager, rules, key_fields + working_directory, + entity_manager, + rules, + key_fields, ) rule_messages = load_feedback_messages( @@ -636,6 +642,13 @@ def apply_business_rules( # pylint: disable=R0914 projected ) + self.step_implementations.identify_and_remove_orphans( # type: ignore + working_directory, + entity_manager.entities, + entity_hierarchy, + key_fields, + ) + 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 new file mode 100644 index 0000000..ea710e9 --- /dev/null +++ b/tests/features/flights.feature @@ -0,0 +1,116 @@ +Feature: Pipeline tests using the flights dataset + Test hierarchical record rejection and ensuring that records are removed correctly including + any "orphan" records generated from record removal in parent entities. + + Scenario: A perfect flights file + Given I submit the flights file perfect_flights.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 no file rejections from the business_rules phase + And there are no record rejections from the business_rules phase + 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 | 0 | + + Scenario: A flights submission where the root record is rejected + Given I submit the flights file missing_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 | + | record | C1 | 1 | + | record | AG1 | 1 | + | record | FG1 | 2 | + | record | PG1 | 4 | + 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 where a child primary key is rejected + Given I submit the flights file missing_flight_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 | + | record | F1 | 1 | + | record | PG1 | 2 | + 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 record rejections + Given I submit the flights file mixture_of_group_rej_and_bi_rej.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 | + | record | F1 | 1 | + | record | PG1 | 2 | + | record | P1 | 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/test_core_engine/test_backends/test_implementations/test_duckdb/test_rules.py b/tests/test_core_engine/test_backends/test_implementations/test_duckdb/test_rules.py index 35007a9..e55228d 100644 --- a/tests/test_core_engine/test_backends/test_implementations/test_duckdb/test_rules.py +++ b/tests/test_core_engine/test_backends/test_implementations/test_duckdb/test_rules.py @@ -1,6 +1,7 @@ """Test DuckDB backend steps.""" # pylint: disable=redefined-outer-name,unused-import,line-too-long +import tempfile from pathlib import Path from typing import Iterator, List, Optional, Set, Tuple, Type @@ -39,6 +40,9 @@ SemiJoin, TableUnion, ) +from dve.core_engine.configuration.v1.hierarchy import ( + ChildHierarchyNode, EntityHierarchy, HierarchyNode +) from dve.core_engine.type_hints import MultipleExpressions from tests.test_core_engine.test_backends.fixtures import ( duckdb_connection, @@ -581,91 +585,106 @@ def test_header_multi_rows_raises( DUCKDB_STEP_BACKEND.join_header(entities, config=header_join) -def test_orphans_planets_satellites( - planets_rel: DuckDBPyRelation, largest_satellites_rel: DuckDBPyRelation -): - """Test a basic orphan idenfitication from satellites to planets.""" - # Each satellite _must_ have a planet. - join = OrphanIdentification( - entity_name="satellites", - target_name="planets", - join_condition="satellites.planet == planets.planet", - ) - entities = EntityManager( - { - "planets": planets_rel.filter(ColumnExpression("Planet") != ConstantExpression("Mars")), - "satellites": largest_satellites_rel, - } - ) - - DUCKDB_STEP_BACKEND.evaluate(entities, config=join) - actual_rel = ( - entities["satellites"] - .filter(ColumnExpression("IsOrphaned")) - .select(ColumnExpression("name")) - ) - actual_rows = sorted(actual_rel.df().to_dict(orient="records"), key=lambda row: row["name"]) - - expected_rel = largest_satellites_rel.filter( - ColumnExpression("Planet") == ConstantExpression("Mars") - ).select(ColumnExpression("name")) - expected_rows = sorted(expected_rel.df().to_dict(orient="records"), key=lambda row: row["name"]) - - assert actual_rows == expected_rows - - -def test_chained_orphans_planets_satellites( - planets_rel: DuckDBPyRelation, largest_satellites_rel: DuckDBPyRelation -): - """Test a basic chained orphan idenfitication from satellites to planets.""" - join = OrphanIdentification( - entity_name="satellites", - target_name="planets", - join_condition="satellites.planet == planets.planet", - ) - entities = EntityManager( - { - "planets": planets_rel.filter(ColumnExpression("planet") != ConstantExpression("Mars")), - "satellites": largest_satellites_rel, - } - ) - DUCKDB_STEP_BACKEND.evaluate(entities, config=join) - entities["planets"] = planets_rel.filter( - ColumnExpression("planet") != ConstantExpression("Earth") - ) - DUCKDB_STEP_BACKEND.evaluate(entities, config=join) - - actual_rel = ( - entities["satellites"] - .filter(ColumnExpression("IsOrphaned")) - .select(ColumnExpression("name")) - ) - actual_rows = sorted(actual_rel.df().to_dict(orient="records"), key=lambda row: row["name"]) - - expected_rel = largest_satellites_rel.filter( - ColumnExpression("Planet").isin(ConstantExpression("Mars"), ConstantExpression("Earth")) - ).select(ColumnExpression("name")) - expected_rows = sorted(expected_rel.df().to_dict(orient="records"), key=lambda row: row["name"]) - - assert actual_rows == expected_rows - - -def test_orphans_missing_entities_raises( - planets_rel: DuckDBPyRelation, satellites_rel: DuckDBPyRelation -): - """Test that trying to join orphans from missing entities raises correctly.""" - join = OrphanIdentification( - entity_name="planets", - target_name="satellites", - join_condition="planets.planet == satellites.planet", - ) +class TestOrphanRecords: + """ + Check that Orphan records identification and removal is working as expected. - entities = EntityManager({"planets": planets_rel}) - with pytest.raises(MissingEntity): - DUCKDB_STEP_BACKEND.identify_orphans(entities, config=join) - entities = EntityManager({"satellites": satellites_rel}) - with pytest.raises(MissingEntity): - DUCKDB_STEP_BACKEND.identify_orphans(entities, config=join) + Current scenarios are: + Flight ID 1 = Perfect Record - no orphans + Flight ID 2 = Record rejected at the flights entity, therefore, two expected orphans in passengers and food entities. + """ + mod_flights_df = pl.DataFrame([ + {'flight_id': 1, '__record_index__': 1}, + ]) + mod_passengers_df = pl.DataFrame([ + {'flight_id': 1, 'passenger_id': 1, '__record_index__': 1}, + {'flight_id': 1, 'passenger_id': 2, '__record_index__': 2}, + {'flight_id': 2, 'passenger_id': 3, '__record_index__': 3}, + ]) + mod_food_df = pl.DataFrame([ + {'passenger_id': 1, 'food_id': 1, '__record_index__': 1}, + {'passenger_id': 1, 'food_id': 2, '__record_index__': 2}, + {'passenger_id': 3, 'food_id': 3, '__record_index__': 3}, + ]) + + def test_identify_orphan_record_single_entity(self): + """Ensure that a single one-to-one check works to identify orphan records.""" + with duckdb.connect() as cnn: + cnn.register("mod_flights", self.mod_flights_df) + cnn.register("mod_passengers", self.mod_passengers_df) + + mod_entities = EntityManager( + entities={ + "flights": cnn.sql("SELECT * FROM mod_flights"), + "passengers": cnn.sql("SELECT * FROM mod_passengers"), + } + ) + + rules = DuckDBStepImplementations(connection=cnn) + _msgs = rules.identify_orphans( + mod_entities.entities, + config=OrphanIdentification( + id="flight_id", + entity_name="passengers", + target_name="flights", + join_condition="passengers.flight_id = flights.flight_id" + ) + ) + result = mod_entities["orphaned_records_tracker"] + assert result.count("*").fetchone()[0] == 1 # type: ignore + assert result.select("entity_name").unique("*").count("*").fetchone()[0] == 1 # type: ignore + + def test_identify_and_remove_orphans(self): + with duckdb.connect() as cnn: + cnn.register("mod_flights", self.mod_flights_df) + cnn.register("mod_passengers", self.mod_passengers_df) + cnn.register("mod_food", self.mod_food_df) + + mod_entities = EntityManager( + entities={ + "flights": cnn.sql("SELECT * FROM mod_flights"), + "passengers": cnn.sql("SELECT * FROM mod_passengers"), + "food": cnn.sql("SELECT * FROM mod_food"), + } + ) + + rules = DuckDBStepImplementations(connection=cnn) + hierarchy = EntityHierarchy({ + "flights": HierarchyNode( + entity_name="flights", + children=[ + ChildHierarchyNode( + entity_name="passengers", + children=[ + ChildHierarchyNode( + entity_name="food", + children=[], + join_fields={"passenger_id": "passenger_id"}, + mandatory=False + ) + ], + join_fields={"flight_id": "flight_id"}, + mandatory=True + ) + ] + ) + }) + + with tempfile.TemporaryDirectory() as wd: + rules.identify_and_remove_orphans( + wd, + mod_entities.entities, + hierarchy + ) + + flights_rel = mod_entities["flights"] + assert flights_rel.select("__record_index__").unique("*").count("*").fetchone()[0] == 1 # type: ignore + + passenger_rel = mod_entities["passengers"] + assert passenger_rel.select("__record_index__").unique("*").count("*").fetchone()[0] == 2 # type: ignore + + food_rel = mod_entities["food"] + assert food_rel.select("__record_index__").unique("*").count("*").fetchone()[0] == 2 # type: ignore def test_has_match_planets_satellites( diff --git a/tests/test_core_engine/test_backends/test_implementations/test_spark/test_rules.py b/tests/test_core_engine/test_backends/test_implementations/test_spark/test_rules.py index 673e611..654dc08 100644 --- a/tests/test_core_engine/test_backends/test_implementations/test_spark/test_rules.py +++ b/tests/test_core_engine/test_backends/test_implementations/test_spark/test_rules.py @@ -568,6 +568,7 @@ def test_header_multi_rows_raises(planets_df: DataFrame, value_literal_1_header: SPARK_STEP_BACKEND.join_header(entities, config=header_join) +@pytest.mark.skip(reason="Logic is no longer valid") def test_orphans_planets_satellites(planets_df: DataFrame, largest_satellites_df: DataFrame): """Test a basic orphan idenfitication from satellites to planets.""" # Each satellite _must_ have a planet. @@ -593,6 +594,7 @@ def test_orphans_planets_satellites(planets_df: DataFrame, largest_satellites_df assert actual_rows == expected_rows +@pytest.mark.skip(reason="Logic is no longer valid") def test_chained_orphans_planets_satellites( planets_df: DataFrame, largest_satellites_df: DataFrame ): @@ -623,6 +625,7 @@ def test_chained_orphans_planets_satellites( assert actual_rows == expected_rows +@pytest.mark.skip(reason="Logic is no longer valid") def test_orphans_missing_entities_raises(planets_df: DataFrame, satellites_df: DataFrame): """Test that trying to join orphans from missing entities raises correctly.""" join = OrphanIdentification( diff --git a/tests/testdata/flights/flights.dischema.json b/tests/testdata/flights/flights.dischema.json new file mode 100644 index 0000000..faac68f --- /dev/null +++ b/tests/testdata/flights/flights.dischema.json @@ -0,0 +1,149 @@ +{ + "contract": { + "schemas": { + "passengers": { + "fields": { + "flight_id": "int", + "passenger_id": "int", + "passenger_name": "str" + } + } + }, + "datasets": { + "country": { + "fields": { + "country_id": "int", + "country_name": "str" + }, + "reader_config": { + ".xml": { + "reader": "DuckDBXMLStreamReader", + "kwargs": { + "record_tag": "country", + "root_tag": "country" + } + } + }, + "key_field": "country_id" + }, + "airport": { + "fields": { + "country_id": "int", + "airport_id": "int", + "airport_name": "str", + "postcode": "str" + }, + "reader_config": { + ".xml": { + "reader": "DuckDBXMLStreamReader", + "kwargs": { + "record_tag": "airport", + "root_tag": "country" + } + } + }, + "key_field": "airport_id" + }, + "flights": { + "fields": { + "airport_id": "int", + "flight_id": "int", + "destination": "str" + }, + "reader_config": { + ".xml": { + "reader": "DuckDBXMLStreamReader", + "kwargs": { + "record_tag": "flight", + "root_tag": "country" + } + } + }, + "key_field": "flight_id" + }, + "passengers": { + "fields": { + "flight_id": "int", + "passenger_id": "int", + "passenger_name": "str" + }, + "reader_config": { + ".xml": { + "reader": "DuckDBXMLStreamReader", + "kwargs": { + "record_tag": "passenger", + "root_tag": "country" + } + } + }, + "key_field": "passenger_id" + } + } + }, + "transformations": { + "filters": [ + { + "entity": "country", + "name": "country_id_missing", + "expression": "country_id IS NOT NULL", + "failure_type": "record", + "failure_message": "Record Rejected - Country is missing an id", + "reporting_field": "country_id", + "reporting_entity": "country", + "category": "Blank", + "error_code": "C1" + }, + { + "entity": "flights", + "name": "flight_missing_id", + "expression": "flight_id IS NOT NULL", + "failure_type": "record", + "failure_message": "Record Rejected - Flight is missing an id", + "reporting_field": "flight_id", + "reporting_entity": "flights", + "category": "Blank", + "error_code": "F1" + }, + { + "entity": "passengers", + "name": "passenger_name_is_null", + "expression": "passenger_name IS NOT NULL", + "failure_type": "record", + "failure_message": "Record Rejected - Passenger Name is missing", + "reporting_field": "passenger_name", + "reporting_entity": "passengers", + "category": "Blank", + "error_code": "P1" + } + ] + }, + "entity_relationships": { + "airport": { + "parent_entity": "country", + "join_fields": { + "country_id": "country_id" + }, + "mandatory": true, + "orphaned_records_error_code": "AG1", + "orphaned_records_error_message": "Group rejected - No valid country group found country" + }, + "flights": { + "parent_entity": "airport", + "join_fields": { + "airport_id": "airport_id" + }, + "mandatory": false, + "orphaned_records_error_code": "FG1", + "orphaned_records_error_message": "Group rejected - No valid airport group found for airport" + }, + "passengers": { + "parent_entity": "flights", + "join_fields": { + "flight_id": "flight_id" + }, + "mandatory": false, + "orphaned_records_error_code": "PG1", + "orphaned_records_error_message": "Group rejected - No valid flight group found for passenger" + } + } +} \ No newline at end of file diff --git a/tests/testdata/flights/missing_country_id.xml b/tests/testdata/flights/missing_country_id.xml new file mode 100644 index 0000000..5c73640 --- /dev/null +++ b/tests/testdata/flights/missing_country_id.xml @@ -0,0 +1,48 @@ + + + England + + + 1 + 1 + Heathrow + TW6 1EW + + + 1 + 1 + Paris + + + 1 + 1 + John + + + 1 + 2 + Jane + + + + + 1 + 2 + Madrid + + + 2 + 3 + Homer + + + 2 + 4 + Marge + + + + + + + \ No newline at end of file diff --git a/tests/testdata/flights/missing_flight_id.xml b/tests/testdata/flights/missing_flight_id.xml new file mode 100644 index 0000000..f9969f1 --- /dev/null +++ b/tests/testdata/flights/missing_flight_id.xml @@ -0,0 +1,48 @@ + + + 1 + England + + + 1 + 1 + Heathrow + TW6 1EW + + + 1 + Paris + + + 1 + 1 + John + + + 1 + 2 + Jane + + + + + 1 + 2 + Madrid + + + 2 + 3 + Homer + + + 2 + 4 + Marge + + + + + + + \ No newline at end of file diff --git a/tests/testdata/flights/mixture_of_group_rej_and_bi_rej.xml b/tests/testdata/flights/mixture_of_group_rej_and_bi_rej.xml new file mode 100644 index 0000000..411b2da --- /dev/null +++ b/tests/testdata/flights/mixture_of_group_rej_and_bi_rej.xml @@ -0,0 +1,53 @@ + + + 1 + England + + + 1 + 1 + Heathrow + TW6 1EW + + + 1 + Paris + + + 1 + 1 + John + + + 1 + 2 + Jane + + + + + 1 + 2 + Madrid + + + 2 + 3 + Homer + + + 2 + 4 + Marge + + + 2 + 5 + + + + + + + + \ No newline at end of file diff --git a/tests/testdata/flights/perfect_flights.xml b/tests/testdata/flights/perfect_flights.xml new file mode 100644 index 0000000..a581a6f --- /dev/null +++ b/tests/testdata/flights/perfect_flights.xml @@ -0,0 +1,49 @@ + + + 1 + England + + + 1 + 1 + Heathrow + TW6 1EW + + + 1 + 1 + Paris + + + 1 + 1 + John + + + 1 + 2 + Jane + + + + + 1 + 2 + Madrid + + + 2 + 3 + Homer + + + 2 + 4 + Marge + + + + + + + \ No newline at end of file From de4e44d817672b20caaae0d87fa28fff19488532 Mon Sep 17 00:00:00 2001 From: stevenhsd <56357022+stevenhsd@users.noreply.github.com> Date: Tue, 15 Sep 2026 12:34:44 +0100 Subject: [PATCH 5/7] refactor: unify HierarchyNode and ChildHierarchyNode also tweaked background writer queue type and use in orphaned records implementation (#153) --- .../entity_relationships.schema.json | 4 +- src/dve/common/error_utils.py | 2 +- src/dve/core_engine/backends/base/rules.py | 36 ++++++------- .../core_engine/backends/metadata/rules.py | 1 + .../core_engine/configuration/v1/__init__.py | 38 ++++++++++--- .../core_engine/configuration/v1/hierarchy.py | 54 ++++++++++++------- src/dve/core_engine/constants.py | 4 +- .../test_duckdb/test_rules.py | 9 ++-- tests/test_core_engine/test_hierarchy.py | 46 +++++++++------- tests/testdata/flights/flights.dischema.json | 12 ++--- 10 files changed, 129 insertions(+), 77 deletions(-) diff --git a/docs/advanced_guidance/json_schemas/entity_relationships.schema.json b/docs/advanced_guidance/json_schemas/entity_relationships.schema.json index c570c3c..9d4e363 100644 --- a/docs/advanced_guidance/json_schemas/entity_relationships.schema.json +++ b/docs/advanced_guidance/json_schemas/entity_relationships.schema.json @@ -20,10 +20,10 @@ "mandatory": { "type": "boolean" }, - "orphaned_records_error_code": { + "missing_parent_id_error_code": { "type": "string" }, - "orphaned_records_error_message": { + "missing_parent_id_error_message": { "type": "string" } }, diff --git a/src/dve/common/error_utils.py b/src/dve/common/error_utils.py index 120c902..23ad714 100644 --- a/src/dve/common/error_utils.py +++ b/src/dve/common/error_utils.py @@ -5,7 +5,7 @@ import logging from collections.abc import Iterable from itertools import chain -from multiprocessing import Queue +from queue import Queue from threading import Thread from typing import Optional, Union diff --git a/src/dve/core_engine/backends/base/rules.py b/src/dve/core_engine/backends/base/rules.py index 1340e32..e24d165 100644 --- a/src/dve/core_engine/backends/base/rules.py +++ b/src/dve/core_engine/backends/base/rules.py @@ -45,11 +45,7 @@ TableUnion, ) from dve.core_engine.backends.types import Entities, EntityType, StageSuccessful -from dve.core_engine.configuration.v1.hierarchy import ( - ChildHierarchyNode, - EntityHierarchy, - HierarchyNode, -) +from dve.core_engine.configuration.v1.hierarchy import EntityHierarchy, HierarchyNode from dve.core_engine.constants import ORPHANED_RECORD_ENTITY_NAME from dve.core_engine.exceptions import CriticalProcessingError from dve.core_engine.loggers import get_logger @@ -390,7 +386,7 @@ def identify_and_remove_orphans( """ def process_node( - node: HierarchyNode | ChildHierarchyNode, + node: HierarchyNode, parent_entity_name: Optional[EntityName], orph_messages: Messages | None = None, ): @@ -400,7 +396,7 @@ def process_node( if orph_messages is None: orph_messages = [] - if isinstance(node, ChildHierarchyNode) and parent_entity_name is not None: + if parent_entity_name is not None: self.logger.info(f"Identifying orphans in {current_entity_name}") join_expr = " AND ".join( @@ -419,7 +415,9 @@ def process_node( ) if no_orphs > 0: - self.logger.info(f"Removing orphan records from {current_entity_name}") + self.logger.info( + f"Removing records with missing parent from {current_entity_name}" + ) location = list(node.join_fields.values())[0] with BackgroundMessageWriter( working_directory=working_directory, @@ -433,32 +431,34 @@ def process_node( entity_name=current_entity_name, reporting=ReportingConfig( emit="record_failure", - code=node.orphaned_records_error_code, - message=node.orphaned_records_error_message, + code=node.missing_parent_id_error_code, + message=node.missing_parent_id_error_message, location=location, - ) - ) + ), + ), ) - for record in _orph_records: - msg_writer.write_queue.put([ + # moved to batch the write - risky if large number of + msg_writer.write_queue.put( + [ FeedbackMessage( entity=current_entity_name, record=record, # type: ignore error_location=location, - error_message=node.orphaned_records_error_message, + error_message=node.missing_parent_id_error_message, failure_type="record", error_type="record", - error_code=node.orphaned_records_error_code, + error_code=node.missing_parent_id_error_code, reporting_field=location, category="Parent Missing", ) - ]) + for record in _orph_records + ] + ) 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) diff --git a/src/dve/core_engine/backends/metadata/rules.py b/src/dve/core_engine/backends/metadata/rules.py index 9b96a14..ae1c4e6 100644 --- a/src/dve/core_engine/backends/metadata/rules.py +++ b/src/dve/core_engine/backends/metadata/rules.py @@ -553,6 +553,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'.""" diff --git a/src/dve/core_engine/configuration/v1/__init__.py b/src/dve/core_engine/configuration/v1/__init__.py index 10e245d..b9b08b6 100644 --- a/src/dve/core_engine/configuration/v1/__init__.py +++ b/src/dve/core_engine/configuration/v1/__init__.py @@ -3,7 +3,8 @@ import json from typing import Any, Optional, Type, Union -from pydantic import BaseModel, Field, PrivateAttr, validate_call +from pydantic import BaseModel, Field, PrivateAttr, field_validator, model_validator, validate_call +from pydantic_core.core_schema import FieldValidationInfo from typing_extensions import Literal from dve.core_engine.backends.base.reference_data import ReferenceConfig, ReferenceConfigUnion @@ -93,11 +94,13 @@ class _TypeAliasDefinition(_BaseTypeDefintion): class _LinkageConfig(BaseModel): """Specify how to link entities back to parents if required""" - parent_entity: EntityName + parent_entity: Optional[EntityName] = None """The name of the parent entity""" - join_fields: JoinFields + join_fields: JoinFields = Field(default_factory=dict) """The fields that can be used to link back to the parent entity""" - mandatory: Optional[bool] = False + is_root_entity: bool = False + """Whether the entity is the highest level parent in a tree""" + mandatory: bool = False """If the entity is a child, is it a mandatory field of the parent""" no_valid_records_error_code: Optional[ErrorCode] = "NoValidRecords" """The error code to emit if the entity has no valid records and is mandatory in the parent entity""" # pylint: disable=C0301 @@ -105,11 +108,34 @@ class _LinkageConfig(BaseModel): "parent record removed as no valid child records" ) """The error message to emit if the entity has no valid records and is mandatory in the parent entity""" # pylint: disable=C0301 - orphaned_records_error_code: Optional[ErrorCode] = "OrphanedRecords" + missing_parent_id_error_code: Optional[ErrorCode] = "MissingParentRecord" """The error code to emit if the entity contains records that are orphaned by parent record rejections""" # pylint: disable=C0301 - orphaned_records_error_message: Optional[ErrorMessage] = "Orphaned records removed" + missing_parent_id_error_message: Optional[ErrorMessage] = ( + "Records removed due to no valid parent record" + ) """The error code to emit if the entity contains records that are orphaned by parent record rejections""" # pylint: disable=C0301 + @model_validator(mode="after") + def _check_root_no_parent_or_join_keys(self): + if self.is_root_entity and (self.parent_entity or self.join_fields): + raise ValueError( + "If entity is root, neither parent_entity nor join keys should be specified" + ) + return self + + @model_validator(mode="after") + def _check_root_mandatory(self): + if self.is_root_entity and not self.mandatory: + raise ValueError("If entity is root, it must be labelled mandatory") + return self + + @model_validator(mode="after") + def _check_parent_entity_with_join_keys(self): + if self.parent_entity or self.join_fields: + if not (self.parent_entity and self.join_fields): + raise ValueError("Both parent_entity and join_fields must be supplied if one is") + return self + class _SchemaConfig(BaseModel): """Configuration for a component schema within a dataset.""" diff --git a/src/dve/core_engine/configuration/v1/hierarchy.py b/src/dve/core_engine/configuration/v1/hierarchy.py index 15dedcf..55c6df8 100644 --- a/src/dve/core_engine/configuration/v1/hierarchy.py +++ b/src/dve/core_engine/configuration/v1/hierarchy.py @@ -16,7 +16,15 @@ class HierarchyNode(BaseModel): """Stores entity hierarchy information""" entity_name: str - children: Optional[list["ChildHierarchyNode"]] = Field(default_factory=list) + children: list["HierarchyNode"] = Field(default_factory=list) + mandatory: bool = False + join_fields: dict[str, str] = Field(default_factory=dict) + no_valid_records_error_code: ErrorCode = "NoValidRecords" + no_valid_records_error_message: ErrorMessage = "parent record removed as no valid child records" + missing_parent_id_error_code: Optional[ErrorCode] = "MissingParentRecord" + missing_parent_id_error_message: Optional[ErrorMessage] = ( + "Records removed due to no valid parent record" + ) def get_descendents(self) -> list[str]: """Recursively list all descendents of the node""" @@ -58,19 +66,6 @@ def as_dict(self) -> dict[str, dict[str, Any]]: return {self.entity_name: ret_dict} -class ChildHierarchyNode(HierarchyNode): - """Stores child entity hierarchy information""" - - join_fields: dict[str, str] - mandatory: Optional[bool] = False - no_valid_records_error_code: Optional[ErrorCode] = "NoValidRecords" - no_valid_records_error_message: Optional[ErrorMessage] = ( - "parent record removed as no valid child records" - ) - orphaned_records_error_code: Optional[ErrorCode] = "OrphanedRecords" - orphaned_records_error_message: Optional[ErrorMessage] = "Orphaned records removed" - - class EntityHierarchy: """Determines and stores entity hierarchy information from config""" @@ -82,12 +77,35 @@ def determine_trees( all_datasets: Iterable[str], entity_relationships: dict[str, _LinkageConfig] ) -> dict[EntityName, HierarchyNode]: """Determine the entity hierarchy trees and store as HierarchyNodes""" + root_entities: dict[str, _LinkageConfig] = dict( + filter(lambda x: x[1].is_root_entity, entity_relationships.items()) + ) top_level_parents: dict[EntityName, HierarchyNode] = { - entity_name: HierarchyNode(entity_name=entity_name) - for entity_name in all_datasets - if entity_name not in entity_relationships + entity_name: HierarchyNode( + entity_name=entity_name, + **config.model_dump( + exclude={ + "parent_entity", + "missing_parent_id_error_code", + "missing_parent_id_error_message", + } + ), + missing_parent_id_error_code=None, + missing_parent_id_error_message=None, + ) + for entity_name, config in root_entities.items() } + if default_roots := [ + entity_name for entity_name in all_datasets if entity_name not in entity_relationships + ]: + for entity_name in default_roots: + top_level_parents[entity_name] = HierarchyNode( + entity_name=entity_name, + missing_parent_id_error_code=None, + missing_parent_id_error_message=None, + ) + for name, linkage_detail in entity_relationships.items(): for main_entity, parent_node in top_level_parents.items(): if ( @@ -96,7 +114,7 @@ def determine_trees( ): parent_node.add_child_node( linkage_detail.parent_entity, - ChildHierarchyNode( + HierarchyNode( entity_name=name, **linkage_detail.model_dump(exclude={"parent_entity"}) ), ) diff --git a/src/dve/core_engine/constants.py b/src/dve/core_engine/constants.py index 7afcae7..3581088 100644 --- a/src/dve/core_engine/constants.py +++ b/src/dve/core_engine/constants.py @@ -7,5 +7,5 @@ """The name of the field that can be used to extract the field value that caused a pydantic validation error""" -ORPHANED_RECORD_ENTITY_NAME: str = "orphaned_records_tracker" -"""Name to keep track of identified orphaned records""" +ORPHANED_RECORD_ENTITY_NAME: str = "orphaned_record_tracker" +"""Name of entity to keep track of records where there is a missing parent record""" diff --git a/tests/test_core_engine/test_backends/test_implementations/test_duckdb/test_rules.py b/tests/test_core_engine/test_backends/test_implementations/test_duckdb/test_rules.py index e55228d..ddb9c81 100644 --- a/tests/test_core_engine/test_backends/test_implementations/test_duckdb/test_rules.py +++ b/tests/test_core_engine/test_backends/test_implementations/test_duckdb/test_rules.py @@ -40,8 +40,9 @@ SemiJoin, TableUnion, ) +from dve.core_engine.constants import ORPHANED_RECORD_ENTITY_NAME from dve.core_engine.configuration.v1.hierarchy import ( - ChildHierarchyNode, EntityHierarchy, HierarchyNode + EntityHierarchy, HierarchyNode ) from dve.core_engine.type_hints import MultipleExpressions from tests.test_core_engine.test_backends.fixtures import ( @@ -630,7 +631,7 @@ def test_identify_orphan_record_single_entity(self): join_condition="passengers.flight_id = flights.flight_id" ) ) - result = mod_entities["orphaned_records_tracker"] + result = mod_entities[ORPHANED_RECORD_ENTITY_NAME] assert result.count("*").fetchone()[0] == 1 # type: ignore assert result.select("entity_name").unique("*").count("*").fetchone()[0] == 1 # type: ignore @@ -653,10 +654,10 @@ def test_identify_and_remove_orphans(self): "flights": HierarchyNode( entity_name="flights", children=[ - ChildHierarchyNode( + HierarchyNode( entity_name="passengers", children=[ - ChildHierarchyNode( + HierarchyNode( entity_name="food", children=[], join_fields={"passenger_id": "passenger_id"}, diff --git a/tests/test_core_engine/test_hierarchy.py b/tests/test_core_engine/test_hierarchy.py index 1b04faf..0128a38 100644 --- a/tests/test_core_engine/test_hierarchy.py +++ b/tests/test_core_engine/test_hierarchy.py @@ -206,8 +206,8 @@ "parent_entity": "ds_001", "join_fields": {"ds_001_id": "ds_001_id"}, "mandatory": false, - "orphaned_records_error_code": "DS003ORPHAN", - "orphaned_records_error_message": "record removed as orphaned" + "missing_parent_id_error_code": "DS003NoParent", + "missing_parent_id_error_message": "record removed as no parent" }, "ds_101": { "parent_entity": "ds_001", @@ -215,15 +215,15 @@ "mandatory_entity": true, "no_valid_records_error_code": "DS101NOVALIDRECS", "no_valid_records_error_message": "{{ ds_001_id }} removed as no valid ds_101 records", - "orphaned_records_error_code": "DS101ORPHAN", - "orphaned_records_error_message": "record removed as orphaned" + "missing_parent_id_error_code": "DS101NoParent", + "missing_parent_id_error_message": "record removed as no parent" }, "ds_201": { "parent_entity": "ds_101", "join_fields": {"referral_id": "ds_101_id"}, "mandatory": false, - "orphaned_records_error_code": "DS201ORPHAN", - "orphaned_records_error_message": "record removed as orphaned" + "missing_parent_id_error_code": "DS201NoParent", + "missing_parent_id_error_message": "record removed as no parent" }, "ds_202": { "parent_entity": "ds_201", @@ -262,6 +262,12 @@ def test_linkage_config_load(): assert dict_rep_001 == json.loads(""" { "ds_001": { + "join_fields": {}, + "mandatory": false, + "no_valid_records_error_code": "NoValidRecords", + "no_valid_records_error_message": "parent record removed as no valid child records", + "missing_parent_id_error_code": null, + "missing_parent_id_error_message": null, "children": { "ds_003": { "join_fields": { @@ -270,8 +276,8 @@ def test_linkage_config_load(): "mandatory": false, "no_valid_records_error_code": "NoValidRecords", "no_valid_records_error_message": "parent record removed as no valid child records", - "orphaned_records_error_code": "DS003ORPHAN", - "orphaned_records_error_message": "record removed as orphaned", + "missing_parent_id_error_code": "DS003NoParent", + "missing_parent_id_error_message": "record removed as no parent", "children": {} }, "ds_101": { @@ -281,8 +287,8 @@ def test_linkage_config_load(): "mandatory": false, "no_valid_records_error_code": "DS101NOVALIDRECS", "no_valid_records_error_message": "{{ ds_001_id }} removed as no valid ds_101 records", - "orphaned_records_error_code": "DS101ORPHAN", - "orphaned_records_error_message": "record removed as orphaned", + "missing_parent_id_error_code": "DS101NoParent", + "missing_parent_id_error_message": "record removed as no parent", "children": { "ds_201": { "join_fields": { @@ -291,8 +297,8 @@ def test_linkage_config_load(): "mandatory": false, "no_valid_records_error_code": "NoValidRecords", "no_valid_records_error_message": "parent record removed as no valid child records", - "orphaned_records_error_code": "DS201ORPHAN", - "orphaned_records_error_message": "record removed as orphaned", + "missing_parent_id_error_code": "DS201NoParent", + "missing_parent_id_error_message": "record removed as no parent", "children": { "ds_202": { "join_fields": { @@ -301,8 +307,8 @@ def test_linkage_config_load(): "mandatory": true, "no_valid_records_error_code": "NoValidRecords", "no_valid_records_error_message": "parent record removed as no valid child records", - "orphaned_records_error_code": "OrphanedRecords", - "orphaned_records_error_message": "Orphaned records removed", + "missing_parent_id_error_code": "MissingParentRecord", + "missing_parent_id_error_message": "Records removed due to no valid parent record", "children": {} } } @@ -328,8 +334,8 @@ def test_linkage_config_load(): "mandatory": false, "no_valid_records_error_code": "DS101NOVALIDRECS", "no_valid_records_error_message": "{{ ds_001_id }} removed as no valid ds_101 records", - "orphaned_records_error_code": "DS101ORPHAN", - "orphaned_records_error_message": "record removed as orphaned", + "missing_parent_id_error_code": "DS101NoParent", + "missing_parent_id_error_message": "record removed as no parent", "children": { "ds_201": { "join_fields": { @@ -338,8 +344,8 @@ def test_linkage_config_load(): "mandatory": false, "no_valid_records_error_code": "NoValidRecords", "no_valid_records_error_message": "parent record removed as no valid child records", - "orphaned_records_error_code": "DS201ORPHAN", - "orphaned_records_error_message": "record removed as orphaned", + "missing_parent_id_error_code": "DS201NoParent", + "missing_parent_id_error_message": "record removed as no parent", "children": { "ds_202": { "join_fields": { @@ -348,8 +354,8 @@ def test_linkage_config_load(): "mandatory": true, "no_valid_records_error_code": "NoValidRecords", "no_valid_records_error_message": "parent record removed as no valid child records", - "orphaned_records_error_code": "OrphanedRecords", - "orphaned_records_error_message": "Orphaned records removed", + "missing_parent_id_error_code": "MissingParentRecord", + "missing_parent_id_error_message": "Records removed due to no valid parent record", "children": {} } } diff --git a/tests/testdata/flights/flights.dischema.json b/tests/testdata/flights/flights.dischema.json index faac68f..f4a7e0c 100644 --- a/tests/testdata/flights/flights.dischema.json +++ b/tests/testdata/flights/flights.dischema.json @@ -124,8 +124,8 @@ "country_id": "country_id" }, "mandatory": true, - "orphaned_records_error_code": "AG1", - "orphaned_records_error_message": "Group rejected - No valid country group found country" + "missing_parent_id_error_code": "AG1", + "missing_parent_id_error_message": "Group rejected - No valid country group found country" }, "flights": { "parent_entity": "airport", @@ -133,8 +133,8 @@ "airport_id": "airport_id" }, "mandatory": false, - "orphaned_records_error_code": "FG1", - "orphaned_records_error_message": "Group rejected - No valid airport group found for airport" + "missing_parent_id_error_code": "FG1", + "missing_parent_id_error_message": "Group rejected - No valid airport group found for airport" }, "passengers": { "parent_entity": "flights", @@ -142,8 +142,8 @@ "flight_id": "flight_id" }, "mandatory": false, - "orphaned_records_error_code": "PG1", - "orphaned_records_error_message": "Group rejected - No valid flight group found for passenger" + "missing_parent_id_error_code": "PG1", + "missing_parent_id_error_message": "Group rejected - No valid flight group found for passenger" } } } \ No newline at end of file From 516e5e47658991b4a9886aa94f487638a9793e54 Mon Sep 17 00:00:00 2001 From: George Robertson <50412379+georgeRobertson@users.noreply.github.com> Date: Tue, 15 Sep 2026 12:37:09 +0100 Subject: [PATCH 6/7] fix: add user feedback message for unsupported file type (#142) --- src/dve/core_engine/models.py | 4 ++- src/dve/pipeline/pipeline.py | 7 +++-- src/dve/pipeline/utils.py | 28 ++++++++++++++++- tests/features/planets.feature | 4 ++- tests/test_pipeline/test_pipeline_utils.py | 36 ++++++++++++++++++++++ 5 files changed, 73 insertions(+), 6 deletions(-) create mode 100644 tests/test_pipeline/test_pipeline_utils.py diff --git a/src/dve/core_engine/models.py b/src/dve/core_engine/models.py index bba2986..49dba23 100644 --- a/src/dve/core_engine/models.py +++ b/src/dve/core_engine/models.py @@ -82,7 +82,9 @@ def _ensure_just_file_stem( @property def file_name_with_ext(self): """Return file name with extension.""" - return f"{self.file_name}.{self.file_extension}" + if self.file_extension: + return f"{self.file_name}.{self.file_extension}" + return self.file_name @classmethod def from_metadata_file(cls, submission_id: str, metadata_uri: Location): diff --git a/src/dve/pipeline/pipeline.py b/src/dve/pipeline/pipeline.py index eaf7661..e062fb2 100644 --- a/src/dve/pipeline/pipeline.py +++ b/src/dve/pipeline/pipeline.py @@ -216,10 +216,10 @@ def write_file_to_parquet( for model_name, model in models.items(): self._logger.info(f"Transforming {model_name} to stringified parquet") - reader: BaseFileReader = load_reader( - dataset, model_name, ext, self.backend_reader_kwargs - ) try: + reader: BaseFileReader = load_reader( + dataset, model_name, ext, self.backend_reader_kwargs + ) if not entity_type: reader.write_parquet( reader.read_to_py_iterator( @@ -242,6 +242,7 @@ def write_file_to_parquet( f"{out}{model_name}", ) except MessageBearingError as exc: + self._logger.error(f"Unable to process {model_name}", exc_info=exc) errors.extend(exc.messages) return list(dict.fromkeys(errors)) # remove any duplicate errors diff --git a/src/dve/pipeline/utils.py b/src/dve/pipeline/utils.py index e6122c2..9163684 100644 --- a/src/dve/pipeline/utils.py +++ b/src/dve/pipeline/utils.py @@ -11,9 +11,11 @@ import dve.core_engine.backends.implementations.duckdb # pylint: disable=unused-import import dve.core_engine.backends.implementations.spark # pylint: disable=unused-import import dve.parser.file_handling as fh +from dve.core_engine.backends.exceptions import MessageBearingError from dve.core_engine.backends.readers import _READER_REGISTRY from dve.core_engine.configuration.v1 import SchemaName, V1EngineConfig, _ModelConfig from dve.core_engine.loggers import get_logger +from dve.core_engine.message import FeedbackMessage from dve.core_engine.type_hints import URI, SubmissionResult from dve.metadata_parser.model_generator import JSONtoPyd @@ -52,7 +54,31 @@ def load_reader( backend_reader_kwargs: Optional[dict[str, Any]] = None, ): """Loads the readers for the diven feed, model name and file extension""" - reader_config = dataset[model_name].reader_config[f".{file_extension.lower()}"] + try: + reader_config = dataset[model_name].reader_config[f".{file_extension.lower()}"] + except KeyError as exc: + if file_extension: + err_msg = ( + f"The supplied file extension `{file_extension}`" + +f" is not a supported file format for {model_name}." + ) + else: + err_msg = "No supplied file extension. Unable to parse file without a file extension." + + raise MessageBearingError( + f"The file extension provided ({file_extension}) is not supported for this collection.", + messages=[ + FeedbackMessage( + entity=model_name, + record=None, + failure_type="submission", + error_location="Whole File", + error_code="InvalidFileExtension", + error_message=err_msg, + ) + ], + ) from exc + reader = _READER_REGISTRY[reader_config.reader]( **reader_config.kwargs_, **backend_reader_kwargs if backend_reader_kwargs else {} ) diff --git a/tests/features/planets.feature b/tests/features/planets.feature index b37b60b..0c4b21d 100644 --- a/tests/features/planets.feature +++ b/tests/features/planets.feature @@ -43,7 +43,9 @@ Feature: Pipeline tests using the planets dataset 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 latest audit record for the submission is marked with processing status failed + Then 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 Scenario: Handle a file with duplicated extension provided (spark) Given I submit the planets file planets.csv.csv for processing diff --git a/tests/test_pipeline/test_pipeline_utils.py b/tests/test_pipeline/test_pipeline_utils.py new file mode 100644 index 0000000..fc28306 --- /dev/null +++ b/tests/test_pipeline/test_pipeline_utils.py @@ -0,0 +1,36 @@ +from dve.core_engine.backends.exceptions import MessageBearingError +from dve.core_engine.configuration.v1 import _ModelConfig, _ReaderConfig +from dve.pipeline.utils import load_reader + +import pytest + + +class TestLoadReader: + test_model_config = _ModelConfig( + fields={"test": "str"}, + reporting_fields=["test"], + key_field="test", + reader_config={ + ".csv": _ReaderConfig(reader="TestCsvReader"), + } + ) + + def test_invalid_load_reader_with_file_ext(self): + with pytest.raises(MessageBearingError) as exc_info: + load_reader( + {"test": self.test_model_config}, + "test_model", + "jpeg" + ) + + assert exc_info.value.messages[0].error_message == "The supplied file extension `jpeg` is not a supported file format for test_model." + + def test_invalid_load_reader_missing_file_ext(self): + with pytest.raises(MessageBearingError) as exc_info: + load_reader( + {"test": self.test_model_config}, + "test_model", + "" + ) + + assert exc_info.value.messages[0].error_message == "No supplied file extension. Unable to parse file without a file extension." From c2b8cf5811815eaf3b5b5ddd513135a9c4814365 Mon Sep 17 00:00:00 2001 From: George Robertson <50412379+georgeRobertson@users.noreply.github.com> Date: Wed, 16 Sep 2026 09:05:13 +0100 Subject: [PATCH 7/7] fix: change record rej ct to be eq to records processed when file rejected (#154) --- src/dve/pipeline/pipeline.py | 14 +++++++++----- src/dve/reporting/excel_report.py | 20 ++++++++++++-------- tests/features/animals.feature | 2 +- tests/features/movies.feature | 4 ++-- 4 files changed, 24 insertions(+), 16 deletions(-) diff --git a/src/dve/pipeline/pipeline.py b/src/dve/pipeline/pipeline.py index e062fb2..1739416 100644 --- a/src/dve/pipeline/pipeline.py +++ b/src/dve/pipeline/pipeline.py @@ -846,14 +846,18 @@ def error_report( .agg(pl.col("Count").sum()) # type: ignore .iter_rows(named=True) } + submission_rejections = err_types.get( + ErrorReportCategories.FILE_REJECTION.reporting_name, 0 + ) sub_stats = SubmissionStatisticsRecord( submission_id=submission_info.submission_id, record_count=submission_status.number_of_records, - number_submission_rejections=err_types.get( - ErrorReportCategories.FILE_REJECTION.reporting_name, 0 - ), - number_record_rejections=err_types.get( - ErrorReportCategories.RECORD_REJECTION.reporting_name, 0 + number_submission_rejections=submission_rejections, + number_record_rejections=( + submission_status.number_of_records + if submission_rejections > 0 else err_types.get( # type: ignore + ErrorReportCategories.RECORD_REJECTION.reporting_name, 0 + ) ), number_warnings=err_types.get(ErrorReportCategories.WARNING.reporting_name, 0), ) diff --git a/src/dve/reporting/excel_report.py b/src/dve/reporting/excel_report.py index 5876cdd..e0c4891 100644 --- a/src/dve/reporting/excel_report.py +++ b/src/dve/reporting/excel_report.py @@ -153,17 +153,21 @@ def _add_submission_info(self, status: str, summary: Worksheet): ), # pylint: disable=C0301 ] ) - if status not in ( + if status in ( ErrorReportStatus.PROCESSING_FAILED, ErrorReportStatus.FILE_REJECTION, ): - summary.append( - [ - "", - "Total Number of Records Rejected", - self.submission_status.number_of_records_rejected, - ] - ) + _records_rejected = self.submission_status.number_of_records + else: + _records_rejected = self.submission_status.number_of_records_rejected + summary.append( + [ + "", + "Total Number of Records Rejected", + _records_rejected, + ] + ) + summary.append(["", ""]) diff --git a/tests/features/animals.feature b/tests/features/animals.feature index d68ddbf..2b49184 100644 --- a/tests/features/animals.feature +++ b/tests/features/animals.feature @@ -55,5 +55,5 @@ Feature: Pipeline tests using the animal dataset | parameter | value | | record_count | 7 | | number_submission_rejections | 1 | - | number_record_rejections | 2 | + | number_record_rejections | 7 | | number_warnings | 1 | diff --git a/tests/features/movies.feature b/tests/features/movies.feature index 750975e..88c0218 100644 --- a/tests/features/movies.feature +++ b/tests/features/movies.feature @@ -42,7 +42,7 @@ Feature: Pipeline tests using the movies dataset | parameter | value | | record_count | 5 | | number_submission_rejections | 1 | - | number_record_rejections | 3 | + | number_record_rejections | 5 | | number_warnings | 2 | And the error aggregates are persisted @@ -81,7 +81,7 @@ Feature: Pipeline tests using the movies dataset | parameter | value | | record_count | 5 | | number_submission_rejections | 1 | - | number_record_rejections | 3 | + | number_record_rejections | 5 | | number_warnings | 2 | And the error aggregates are persisted