"` for this entity.'
)
return self
diff --git a/src/dve/pipeline/pipeline.py b/src/dve/pipeline/pipeline.py
index 1737ebf8..abff55e3 100644
--- a/src/dve/pipeline/pipeline.py
+++ b/src/dve/pipeline/pipeline.py
@@ -759,6 +759,8 @@ def apply_business_rules( # pylint: disable=R0914,R0915
final_projection
)
+ self.step_implementations.clear_entity_cache() # type: ignore
+
fh.remove_prefix(
fh.joinuri(
self.processed_files_path, submission_info.submission_id, "temp_business_rules"
@@ -775,7 +777,7 @@ def apply_business_rules( # pylint: disable=R0914,R0915
self.processed_files_path,
submission_info.submission_id,
"data_contract",
- rules.global_variables.get('entity', submission_info.dataset_id)
+ rules.global_variables.get("entity", submission_info.dataset_id),
)
)
)
diff --git a/src/dve/reporting/__init__.py b/src/dve/reporting/__init__.py
index ab78e113..476e0d0d 100644
--- a/src/dve/reporting/__init__.py
+++ b/src/dve/reporting/__init__.py
@@ -1,2 +1,3 @@
"""Error reports module."""
+
# pylint: disable=R0801
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 ecef834b..aae56967 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,9 @@
"""Test DuckDB backend steps."""
# pylint: disable=redefined-outer-name,unused-import,line-too-long
+import datetime
+from io import StringIO
+import json
import tempfile
from pathlib import Path
from typing import Iterator, List, Optional, Set, Tuple, Type
@@ -884,3 +887,112 @@ def test_read_and_write_nested_parquet(nested_typecast_parquet):
"datetimefield": "TIMESTAMP",
"subfield": "STRUCT(id BIGINT, substrfield VARCHAR, subarrayfield DATE[])[]",
}
+
+def test_cache_management():
+ conn = DUCKDB_STEP_BACKEND.connection
+ with tempfile.NamedTemporaryFile(mode="w") as tf1, tempfile.NamedTemporaryFile(mode="w") as tf2:
+ td1 = [
+ {"greeting": "hi", "num_one": 2, "num_two": 4, "test_date": datetime.date(2020,5,1), "active": True},
+ {"greeting": "bonjour", "num_one": 3, "num_two": 9, "test_date": datetime.date(2025,7,4), "active": False},
+ ]
+ tf1.write(json.dumps(td1, default=str))
+
+ tf1.seek(0)
+
+ td1_schema = {"greeting": "STRING", "num_one": "BIGINT", "num_two" :"BIGINT", "test_date": "DATE", "active": "BOOLEAN"}
+
+ td2 = [
+ {"farewell": "aurevoir", "lots_of_nums": [1,2,3], "nested_field": {"nested_str": "test1", "nested_timestamp": datetime.datetime(2024,3,5,1,2,3)}},
+ {"farewell": "bye", "lots_of_nums": [4,5,6,7,8], "nested_field": {"nested_str": "test2", "nested_timestamp": datetime.datetime(2022,8,12,10,12,14)}},
+ ]
+
+ tf2.write(json.dumps(td2, default=str))
+
+ tf2.seek(0)
+
+ td2_schema = {"farewell": "STRING", "lots_of_nums": "BIGINT[]", "nested_field": "STRUCT(nested_str STRING, nested_timestamp TIMESTAMP)"}
+
+ em = EntityManager({})
+ em.entities["test_one"] = conn.read_json(tf1.name, columns=td1_schema)
+ em.entities["test_two"] = conn.read_json(tf2.name, columns=td2_schema)
+
+ DUCKDB_STEP_BACKEND.cache_entity("test_one", em.entities)
+
+ cached_tables = [rw["table_name"] for rw in conn.sql("SELECT table_name from duckdb_tables() WHERE database_name = 'temp'").pl().to_dicts()]
+
+ test_one_temp_name = list(filter(lambda x: x.startswith("test_one_"), cached_tables))[0]
+
+ assert "test_one" in DUCKDB_STEP_BACKEND.entity_cache_tracker
+ assert DUCKDB_STEP_BACKEND.entity_cache_tracker["test_one"] == test_one_temp_name
+ assert sorted(em.entities["test_one"].pl().to_dicts(), key=lambda x: x.get("test_date")) == td1
+ assert sorted(conn.table(test_one_temp_name).pl().to_dicts(), key=lambda x: x.get("test_date")) == td1
+ assert "test_two" not in DUCKDB_STEP_BACKEND.entity_cache_tracker
+
+ DUCKDB_STEP_BACKEND.cache_entity("test_two", em.entities)
+ DUCKDB_STEP_BACKEND._remove_cached_artifact("test_one")
+
+ cached_tables = [rw["table_name"] for rw in conn.sql("SELECT table_name from duckdb_tables() WHERE database_name = 'temp'").pl().to_dicts()]
+ test_two_temp_name = list(filter(lambda x: x.startswith("test_two_"), cached_tables))[0]
+
+ assert "test_two" in DUCKDB_STEP_BACKEND.entity_cache_tracker
+ assert DUCKDB_STEP_BACKEND.entity_cache_tracker["test_two"] in cached_tables
+ assert sorted(em.entities["test_two"].pl().to_dicts(), key=lambda x: x.get("farewell")) == td2
+ assert sorted(conn.table(test_two_temp_name).pl().to_dicts(), key=lambda x: x.get("farewell")) == td2
+ assert "test_one" not in DUCKDB_STEP_BACKEND.entity_cache_tracker
+
+ DUCKDB_STEP_BACKEND.clear_entity_cache()
+
+ cached_tables = [rw["table_name"] for rw in conn.sql("SELECT table_name from duckdb_tables() WHERE database_name = 'temp'").pl().to_dicts()]
+
+ assert not DUCKDB_STEP_BACKEND.entity_cache_tracker
+ assert not any(tbl.startswith("test_one_") for tbl in cached_tables)
+ assert not any(tbl.startswith("test_two_") for tbl in cached_tables)
+
+def test_cache_management_with_update():
+ conn = DUCKDB_STEP_BACKEND.connection
+ with tempfile.NamedTemporaryFile(mode="w") as tf, tempfile.NamedTemporaryFile(mode="w") as ef:
+ td = [
+ {"idx": 1, "lots_of_nums": [1,2,3], "nested_field": {"nested_str": "test1", "nested_timestamp": datetime.datetime(2024,3,5,1,2,3)}},
+ {"idx": 2, "lots_of_nums": [4,5,6,7,8], "nested_field": {"nested_str": "test2", "nested_timestamp": datetime.datetime(2022,8,12,10,12,14)}},
+ ]
+
+ tf.write(json.dumps(td, default=str))
+ tf.seek(0)
+ td_schema = {"idx": "BIGINT",
+ "lots_of_nums": "BIGINT[]",
+ "nested_field": "STRUCT(nested_str STRING, nested_timestamp TIMESTAMP)"}
+
+
+ em = EntityManager({})
+ em.entities["test_df"] = conn.read_json(tf.name, columns=td_schema)
+
+ DUCKDB_STEP_BACKEND.cache_entity("test_df", em.entities)
+
+ cached_tables = [rw["table_name"] for rw in conn.sql("SELECT table_name from duckdb_tables() WHERE database_name = 'temp'").pl().to_dicts()]
+
+ first_temp_name = list(filter(lambda x: x.startswith("test_df_"), cached_tables))[0]
+
+ assert "test_df" in DUCKDB_STEP_BACKEND.entity_cache_tracker
+ assert DUCKDB_STEP_BACKEND.entity_cache_tracker["test_df"] == first_temp_name
+ assert sorted(em.entities["test_df"].pl().to_dicts(), key=lambda x: x.get("idx")) == td
+
+ extra_data = [{"idx": 3, "lots_of_nums": [9], "nested_field": {"nested_str": "test3", "nested_timestamp": datetime.datetime(2024,1,9,3,2,1)}}]
+ ef.write(json.dumps(extra_data, default=str))
+ ef.seek(0)
+ em.entities["test_df"] = em.entities["test_df"].union(conn.read_json(ef.name, columns=td_schema))
+ DUCKDB_STEP_BACKEND.cache_entity("test_df", em.entities)
+
+ cached_tables = [rw["table_name"] for rw in conn.sql("SELECT table_name from duckdb_tables() WHERE database_name = 'temp'").pl().to_dicts()]
+ tables_of_interest = list(filter(lambda x: x.startswith("test_df_"), cached_tables))
+ assert len(tables_of_interest) == 1
+ assert first_temp_name not in tables_of_interest
+ second_temp_name = tables_of_interest[0]
+ assert sorted(em.entities["test_df"].pl().to_dicts(), key=lambda x: x.get("idx")) == td + extra_data
+ assert sorted(conn.table(second_temp_name).pl().to_dicts(), key=lambda x: x.get("idx")) == td + extra_data
+
+ DUCKDB_STEP_BACKEND.clear_entity_cache()
+
+ cached_tables = [rw["table_name"] for rw in conn.sql("SELECT table_name from duckdb_tables() WHERE database_name = 'temp'").pl().to_dicts()]
+
+ assert not DUCKDB_STEP_BACKEND.entity_cache_tracker
+ assert not any(tbl.startswith("test_df_") for tbl in cached_tables)
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 6e38c7a9..2a79d81b 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
@@ -1,6 +1,7 @@
"""Test Spark backend steps."""
# pylint: disable=redefined-outer-name,unused-import,line-too-long
+import datetime
from pathlib import Path
from typing import List, Optional, Set, Tuple, Type
@@ -9,6 +10,7 @@
from pyspark.sql.functions import col, lit
from pyspark.sql.types import (
ArrayType,
+ BooleanType,
DateType,
LongType,
Row,
@@ -852,3 +854,107 @@ def test_read_and_write_nested_parquet(nested_typecast_parquet):
),
]
)
+
+def test_cache_management():
+ spark = SPARK_STEP_BACKEND._spark_session
+ td1 = [
+ {"greeting": "hi", "num_one": 2, "num_two": 4, "test_date": datetime.date(2020,5,1), "active": True},
+ {"greeting": "bonjour", "num_one": 3, "num_two": 9, "test_date": datetime.date(2025,7,4), "active": False},
+ ]
+ td1_schema = StructType([
+ StructField("greeting", StringType()),
+ StructField("num_one", LongType()),
+ StructField("num_two", LongType()),
+ StructField("test_date", DateType()),
+ StructField("active", BooleanType())])
+ td2 = [
+ {"farewell": "aurevoir", "lots_of_nums": [1,2,3], "nested_field": {"nested_str": "test1", "nested_timestamp": datetime.datetime(2024,3,5,1,2,3)}},
+ {"farewell": "bye", "lots_of_nums": [4,5,6,7,8], "nested_field": {"nested_str": "test2", "nested_timestamp": datetime.datetime(2022,8,12,10,12,14)}},
+ ]
+ td2_schema = StructType(
+ [
+ StructField("farewell", StringType()),
+ StructField("lots_of_nums", ArrayType(LongType())),
+ StructField("nested_field", StructType([StructField("nested_str", StringType()), StructField("nested_timestamp", TimestampType())]))
+ ]
+ )
+ em = EntityManager({})
+ em.entities["test_one"] = spark.createDataFrame(td1, schema=td1_schema)
+ em.entities["test_two"] = spark.createDataFrame(td2, schema=td2_schema)
+
+ SPARK_STEP_BACKEND.cache_entity("test_one", em.entities)
+
+ cached_tables = (rw.tableName for rw in spark.sql("SHOW TABLES").collect())
+
+ test_one_temp_name = list(filter(lambda x: x.startswith("test_one_"), cached_tables))[0]
+
+ assert "test_one" in SPARK_STEP_BACKEND.entity_cache_tracker
+ assert SPARK_STEP_BACKEND.entity_cache_tracker["test_one"] == test_one_temp_name
+ assert sorted([rw.asDict(True) for rw in em.entities["test_one"].collect()], key=lambda x: x.get("test_date")) == td1
+ assert sorted([rw.asDict(True) for rw in spark.table(test_one_temp_name).collect()], key=lambda x: x.get("test_date")) == td1
+ assert "test_two" not in SPARK_STEP_BACKEND.entity_cache_tracker
+
+ SPARK_STEP_BACKEND.cache_entity("test_two", em.entities)
+ SPARK_STEP_BACKEND._remove_cached_artifact("test_one")
+
+ cached_tables = list(rw.tableName for rw in spark.sql("SHOW TABLES").collect())
+ test_two_temp_name = list(filter(lambda x: x.startswith("test_two_"), cached_tables))[0]
+
+ assert "test_two" in SPARK_STEP_BACKEND.entity_cache_tracker
+ assert SPARK_STEP_BACKEND.entity_cache_tracker["test_two"] in cached_tables
+ assert sorted([rw.asDict(True) for rw in em.entities["test_two"].collect()], key=lambda x: x.get("farewell")) == td2
+ assert sorted([rw.asDict(True) for rw in spark.table(test_two_temp_name).collect()], key=lambda x: x.get("farewell")) == td2
+ assert "test_one" not in SPARK_STEP_BACKEND.entity_cache_tracker
+
+ SPARK_STEP_BACKEND.clear_entity_cache()
+
+ cached_tables = list(rw.tableName for rw in spark.sql("SHOW TABLES").collect())
+
+ assert not SPARK_STEP_BACKEND.entity_cache_tracker
+ assert not any(tbl.startswith("test_one_") for tbl in cached_tables)
+ assert not any(tbl.startswith("test_two_") for tbl in cached_tables)
+
+def test_cache_management_with_update():
+ spark = SPARK_STEP_BACKEND._spark_session
+ td = [
+ {"idx": 1, "lots_of_nums": [1,2,3], "nested_field": {"nested_str": "test1", "nested_timestamp": datetime.datetime(2024,3,5,1,2,3)}},
+ {"idx": 2, "lots_of_nums": [4,5,6,7,8], "nested_field": {"nested_str": "test2", "nested_timestamp": datetime.datetime(2022,8,12,10,12,14)}},
+ ]
+ td_schema = StructType(
+ [
+ StructField("idx", LongType()),
+ StructField("lots_of_nums", ArrayType(LongType())),
+ StructField("nested_field", StructType([StructField("nested_str", StringType()), StructField("nested_timestamp", TimestampType())]))
+ ]
+ )
+ em = EntityManager({})
+ em.entities["test_df"] = spark.createDataFrame(td, schema=td_schema)
+
+ SPARK_STEP_BACKEND.cache_entity("test_df", em.entities)
+
+ cached_tables = (rw.tableName for rw in spark.sql("SHOW TABLES").collect())
+
+ first_temp_name = list(filter(lambda x: x.startswith("test_df_"), cached_tables))[0]
+
+ assert "test_df" in SPARK_STEP_BACKEND.entity_cache_tracker
+ assert SPARK_STEP_BACKEND.entity_cache_tracker["test_df"] == first_temp_name
+ assert sorted([rw.asDict(True) for rw in em.entities["test_df"].collect()], key=lambda x: x.get("idx")) == td
+
+ extra_data = [{"idx": 3, "lots_of_nums": [9], "nested_field": {"nested_str": "test3", "nested_timestamp": datetime.datetime(2024,1,9,3,2,1)}}]
+ em.entities["test_df"] = em.entities["test_df"].union(spark.createDataFrame(extra_data, schema=td_schema))
+ SPARK_STEP_BACKEND.cache_entity("test_df", em.entities)
+
+ cached_tables = (rw.tableName for rw in spark.sql("SHOW TABLES").collect())
+ tables_of_interest = list(filter(lambda x: x.startswith("test_df_"), cached_tables))
+ assert len(tables_of_interest) == 1
+ assert first_temp_name not in tables_of_interest
+ second_temp_name = tables_of_interest[0]
+ assert sorted([rw.asDict(True) for rw in em.entities["test_df"].collect()], key=lambda x: x.get("idx")) == td + extra_data
+ assert sorted([rw.asDict(True) for rw in spark.table(second_temp_name).collect()], key=lambda x: x.get("idx")) == td + extra_data
+
+ SPARK_STEP_BACKEND.clear_entity_cache()
+
+ cached_tables = list(rw.tableName for rw in spark.sql("SHOW TABLES").collect())
+
+ assert not SPARK_STEP_BACKEND.entity_cache_tracker
+ assert not any(tbl.startswith("test_df_") for tbl in cached_tables)
From df8bbe4e2f7b26b007b36a3fe144232aeee14a00 Mon Sep 17 00:00:00 2001
From: George Robertson <50412379+georgeRobertson@users.noreply.github.com>
Date: Thu, 8 Oct 2026 11:35:26 +0100
Subject: [PATCH 27/37] feat: add orphan and group rejections to spark backend
(#174)
---
src/dve/core_engine/backends/base/reader.py | 2 +-
src/dve/core_engine/backends/base/rules.py | 19 +-
src/dve/core_engine/backends/exceptions.py | 5 +-
.../implementations/duckdb/readers/csv.py | 9 +-
.../backends/implementations/duckdb/rules.py | 15 --
.../implementations/spark/readers/xml.py | 3 +-
.../backends/implementations/spark/rules.py | 107 ++++++++-
.../core_engine/backends/readers/utilities.py | 2 +-
src/dve/pipeline/pipeline.py | 3 +-
tests/features/flights.feature | 78 +++++++
.../flights/flights_spark.dischema.json | 212 ++++++++++++++++++
11 files changed, 422 insertions(+), 33 deletions(-)
create mode 100644 tests/testdata/flights/flights_spark.dischema.json
diff --git a/src/dve/core_engine/backends/base/reader.py b/src/dve/core_engine/backends/base/reader.py
index 93b97071..eeeca7f4 100644
--- a/src/dve/core_engine/backends/base/reader.py
+++ b/src/dve/core_engine/backends/base/reader.py
@@ -11,7 +11,7 @@
from dve.core_engine.backends.exceptions import (
CriticalMessageBearingError,
MessageBearingError,
- ReaderLacksEntityTypeSupport
+ ReaderLacksEntityTypeSupport,
)
from dve.core_engine.backends.types import EntityName, EntityType
from dve.core_engine.configuration.v1 import (
diff --git a/src/dve/core_engine/backends/base/rules.py b/src/dve/core_engine/backends/base/rules.py
index 2471426c..6200e013 100644
--- a/src/dve/core_engine/backends/base/rules.py
+++ b/src/dve/core_engine/backends/base/rules.py
@@ -427,7 +427,13 @@ def process_node(node: HierarchyNode) -> bool:
self.cache_entity(node.entity_name, entities)
- return len(_messages) > 0
+ _orph_count = len(_messages)
+
+ self.logger.info(
+ f"Found {_orph_count} orphan records between {node.entity_name} and {node.parent_entity}" # pylint: disable=C0301
+ )
+
+ return _orph_count > 0
entity_issues_found: dict[EntityName, bool] = {}
@@ -454,7 +460,7 @@ def process_node(node: HierarchyNode) -> bool:
return False
self.logger.info(
- f"Identifying that mandatory entity `{node.parent_entity}` has at least 1 valid child record" # pylint: disable=C0301
+ f"Identifying that mandatory entity `{node.parent_entity}` has at least 1 valid child record in {node.entity_name}" # pylint: disable=C0301
)
join_expr = " AND ".join(
@@ -493,7 +499,14 @@ def process_node(node: HierarchyNode) -> bool:
]
msg_writer.write_queue.put(_messages)
self.cache_entity(node.parent_entity, entities)
- return len(_messages) > 0
+
+ _no_valid_child_records: int = len(_messages)
+
+ self.logger.info(
+ f"Found {_no_valid_child_records} records with no valid children in {node.parent_entity}." # pylint: disable=C0301
+ )
+
+ return _no_valid_child_records > 0
entity_issues_found: dict[EntityName, bool] = {}
diff --git a/src/dve/core_engine/backends/exceptions.py b/src/dve/core_engine/backends/exceptions.py
index f8d93079..99808c8a 100644
--- a/src/dve/core_engine/backends/exceptions.py
+++ b/src/dve/core_engine/backends/exceptions.py
@@ -32,6 +32,7 @@ def __init__(self, *args: object, messages: Messages) -> None:
self.messages = messages
"""The messages to be returned as part of the error."""
+
class CriticalMessageBearingError(BackendError):
"""
A backend error that comes with a pre-created message.
@@ -44,6 +45,7 @@ def __init__(self, *args: object, message: FeedbackMessage) -> None:
self.message = message
"""The message to be returned as part of the error."""
+
class UnableToParseCSVError(CriticalMessageBearingError):
"""An error raised when unable to parse a CSV file"""
@@ -60,7 +62,8 @@ def __init__(
failure_type="submission",
is_informational=False,
error_type="csv read",
- error_message=error_message or "Unable to parse the CSV file. Please check the structure of your CSV.", # pylint: disable=C0301
+ error_message=error_message
+ or "Unable to parse the CSV file. Please check the structure of your CSV.", # pylint: disable=C0301
error_code=error_code or "MalformedCSV",
)
)
diff --git a/src/dve/core_engine/backends/implementations/duckdb/readers/csv.py b/src/dve/core_engine/backends/implementations/duckdb/readers/csv.py
index dedd66bd..488f8c85 100644
--- a/src/dve/core_engine/backends/implementations/duckdb/readers/csv.py
+++ b/src/dve/core_engine/backends/implementations/duckdb/readers/csv.py
@@ -123,7 +123,8 @@ def read_to_relation( # pylint: disable=unused-argument
raise UnableToParseCSVError(
entity_name="csv_structure",
error_code=self.ft_error_code,
- error_message=self.ft_error_message or "Unable to parse CSV file. Structure is likely malformed.", # pylint: disable=C0301
+ error_message=self.ft_error_message
+ or "Unable to parse CSV file. Structure is likely malformed.", # pylint: disable=C0301
) from exc
if self.null_empty_strings:
@@ -183,7 +184,8 @@ def read_to_relation( # pylint: disable=unused-argument
raise UnableToParseCSVError(
entity_name="csv_structure",
error_code=self.ft_error_code,
- error_message=self.ft_error_message or "Unable to parse CSV file. Structure is likely malformed.", # pylint: disable=C0301
+ error_message=self.ft_error_message
+ or "Unable to parse CSV file. Structure is likely malformed.", # pylint: disable=C0301
) from exc
if self.null_empty_strings:
@@ -200,7 +202,8 @@ def read_to_relation( # pylint: disable=unused-argument
raise UnableToParseCSVError(
entity_name="csv_structure",
error_code=self.ft_error_code,
- error_message=self.ft_error_message or "Found zero records after loading CSV. File is likely malformed.", # pylint: disable=C0301
+ error_message=self.ft_error_message
+ or "Found zero records after loading CSV. File is likely malformed.", # pylint: disable=C0301
)
return entity
diff --git a/src/dve/core_engine/backends/implementations/duckdb/rules.py b/src/dve/core_engine/backends/implementations/duckdb/rules.py
index 24910d33..766142aa 100644
--- a/src/dve/core_engine/backends/implementations/duckdb/rules.py
+++ b/src/dve/core_engine/backends/implementations/duckdb/rules.py
@@ -421,12 +421,6 @@ def identify_orphans(
.set_alias("orphan")
)
- if relation_is_empty(orphaned_rel):
- self.logger.info(
- f"Found 0 orphan records between {config.entity_name} and {config.target_name}"
- )
- return []
-
message_rel = (
entities[config.entity_name]
.set_alias(config.entity_name)
@@ -474,15 +468,6 @@ def check_mandatory_group(
missing_children_rel = joined_rel.filter("fk IS NULL")
filtered_rel = joined_rel.filter("fk IS NOT NULL").select(StarExpression(exclude=["fk"]))
- _no_valid_child_records: tuple[int] = missing_children_rel.count("*").fetchone() # type: ignore # pylint: disable=C0301
- if _no_valid_child_records:
- _no_valid_children = _no_valid_child_records[0]
- else:
- _no_valid_children = 0
- self.logger.info(
- f"Found {_no_valid_children} records with no valid children in {config.entity_name}."
- ) # pylint: disable=C0301
-
entities[config.entity_name] = filtered_rel
return duckdb_rel_to_dictionaries(missing_children_rel)
diff --git a/src/dve/core_engine/backends/implementations/spark/readers/xml.py b/src/dve/core_engine/backends/implementations/spark/readers/xml.py
index 275182e8..769c480e 100644
--- a/src/dve/core_engine/backends/implementations/spark/readers/xml.py
+++ b/src/dve/core_engine/backends/implementations/spark/readers/xml.py
@@ -44,6 +44,7 @@ def read_to_dataframe(
resource: URI,
entity_name: EntityName,
schema: type[BaseModel],
+ all_model_fields: Optional[set[str]] = None,
) -> DataFrame:
"""Stream an XML file into a Spark data frame"""
if not self.spark:
@@ -51,7 +52,7 @@ def read_to_dataframe(
spark_schema = get_type_from_annotation(schema)
return self.add_record_index(
self.spark.createDataFrame( # type: ignore
- list(self.read_to_py_iterator(resource, entity_name, schema)),
+ list(self.read_to_py_iterator(resource, entity_name, schema, all_model_fields)),
schema=spark_schema,
)
)
diff --git a/src/dve/core_engine/backends/implementations/spark/rules.py b/src/dve/core_engine/backends/implementations/spark/rules.py
index 5f218ec6..8cdc8a1b 100644
--- a/src/dve/core_engine/backends/implementations/spark/rules.py
+++ b/src/dve/core_engine/backends/implementations/spark/rules.py
@@ -1,7 +1,7 @@
"""Step implementations in Spark."""
# pylint: disable=R0801
-from collections.abc import Callable, Iterator
+from collections.abc import Callable, Iterable, Iterator
from typing import Optional
from uuid import uuid4
@@ -13,6 +13,7 @@
from dve.core_engine.backends.exceptions import ConstraintError
from dve.core_engine.backends.implementations.spark.spark_helpers import (
create_udf,
+ df_is_empty,
get_all_registered_udfs,
object_to_spark_literal,
spark_filter_contract_errors,
@@ -48,6 +49,7 @@
SemiJoin,
TableUnion,
)
+from dve.core_engine.constants import 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
@@ -341,16 +343,107 @@ def union(self, entities: SparkEntities, *, config: TableUnion) -> Messages:
return []
def identify_orphans(
- self, entities: SparkEntities, *, config: OrphanIdentification
- ) -> tuple[Messages, int]:
- # TODO - adjust this to new setup of identify and remove orphans
- raise NotImplementedError
+ self,
+ entities: SparkEntities,
+ *,
+ config: OrphanIdentification,
+ ) -> Iterable:
+ """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.
+
+ If there is already an 'IsOrphaned' column in the entity, this will be set to the
+ logical OR of its current value and the value it would have been set to otherwise.
+
+ """
+ source_df: DataFrame = entities[config.entity_name]
+ source_df = source_df.alias(config.entity_name)
+
+ if df_is_empty(source_df):
+ self.logger.info(f"{config.entity_name} is empty. Skipping orphan check.")
+ return
+
+ target_df: DataFrame = entities[config.target_name]
+ match_name = f"matched_{uuid4().hex}"
+ target_df = target_df.select("*", sf.lit(1).alias(match_name)).alias(config.target_name)
+
+ orphaned_df: DataFrame = (
+ source_df.join(target_df, on=sf.expr(config.join_condition), how="left")
+ .groupBy(f"{config.entity_name}.{config.id}")
+ .agg(
+ sf.first(f"{config.entity_name}.{RECORD_INDEX_COLUMN_NAME}").alias(
+ RECORD_INDEX_COLUMN_NAME
+ ), # pylint: disable=C0301
+ (sf.coalesce(sf.count(match_name), sf.lit(0)) == sf.lit(0)).alias("IsOrphaned"),
+ )
+ .filter(sf.col("IsOrphaned"))
+ .select(RECORD_INDEX_COLUMN_NAME)
+ .alias("orphan")
+ )
+
+ message_df = (
+ entities[config.entity_name]
+ .alias(config.entity_name)
+ .join(
+ orphaned_df,
+ sf.expr(
+ f"{config.entity_name}.{RECORD_INDEX_COLUMN_NAME} = orphan.{RECORD_INDEX_COLUMN_NAME}", # pylint: disable=C0301
+ ),
+ "semi",
+ )
+ )
+ filtered_rel = (
+ entities[config.entity_name]
+ .alias(config.entity_name)
+ .join(
+ orphaned_df,
+ sf.expr(
+ f"{config.entity_name}.{RECORD_INDEX_COLUMN_NAME} = orphan.{RECORD_INDEX_COLUMN_NAME}", # pylint: disable=C0301
+ ),
+ "anti",
+ )
+ )
+
+ entities[config.entity_name] = filtered_rel
+
+ for r in message_df.toLocalIterator():
+ yield r.asDict()
def check_mandatory_group(
self, entities: SparkEntities, *, config: GroupIdentification
) -> Iterator:
- # TODO - implement for spark
- raise NotImplementedError
+ """
+ Check that a mandatory key in an entity has at least one valid entry in the all the
+ child entities.
+ """
+ source_df: DataFrame = entities[config.entity_name]
+ source_df = source_df.alias(config.entity_name)
+ target_df: DataFrame = entities[config.target_name]
+ target_df = target_df.alias(config.target_name)
+
+ source_columns = [f"{config.entity_name}.{c.strip()}" for c in source_df.columns]
+ _pk, fk = config.join_condition.split("=")
+
+ joined_df = source_df.join(target_df, sf.expr(config.join_condition), "left").select(
+ *source_columns,
+ sf.col(fk.strip()).alias("fk"),
+ )
+
+ missing_children_df = joined_df.filter("fk IS NULL")
+ filtered_df = joined_df.filter("fk IS NOT NULL").select("*").drop(sf.col("fk"))
+
+ if not df_is_empty(missing_children_df):
+ _no_valid_children = missing_children_df.count()
+ else:
+ _no_valid_children = 0
+ self.logger.info(
+ f"Found {_no_valid_children} records with no valid children in {config.entity_name}."
+ )
+
+ entities[config.entity_name] = filtered_df
+
+ for r in missing_children_df.toLocalIterator():
+ yield r.asDict()
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/readers/utilities.py b/src/dve/core_engine/backends/readers/utilities.py
index 99ca6abc..86347dce 100644
--- a/src/dve/core_engine/backends/readers/utilities.py
+++ b/src/dve/core_engine/backends/readers/utilities.py
@@ -66,7 +66,7 @@ def raise_message_bearing_error_on_header_differences(
reporting_field="csv_header",
error_code=field_check_error_code,
error_message=field_check_error_message,
- )
+ ),
)
diff --git a/src/dve/pipeline/pipeline.py b/src/dve/pipeline/pipeline.py
index abff55e3..2cf00553 100644
--- a/src/dve/pipeline/pipeline.py
+++ b/src/dve/pipeline/pipeline.py
@@ -764,7 +764,8 @@ def apply_business_rules( # pylint: disable=R0914,R0915
fh.remove_prefix(
fh.joinuri(
self.processed_files_path, submission_info.submission_id, "temp_business_rules"
- )
+ ),
+ recursive=True,
)
self.check_mandatory_entities_have_records(
diff --git a/tests/features/flights.feature b/tests/features/flights.feature
index bcfad686..3bf2ee06 100644
--- a/tests/features/flights.feature
+++ b/tests/features/flights.feature
@@ -286,3 +286,81 @@ Feature: Pipeline tests using the flights dataset
And 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: A flights submission with a rejection on a node with two mandatory nodes (spark)
+ Given I submit the flights file multi_node_file_rejection.xml for processing
+ And A spark pipeline is configured with schema file 'flights_spark.dischema.json'
+ And I add initial audit entries for the submission
+ Then the latest audit record for the submission is marked with processing status file_transformation
+ When I run the file transformation phase
+ Then the country entity is stored as a parquet after the file_transformation phase
+ And the airport entity is stored as a parquet after the file_transformation phase
+ And the flights entity is stored as a parquet after the file_transformation phase
+ And the passengers entity is stored as a parquet after the file_transformation phase
+ And the latest audit record for the submission is marked with processing status data_contract
+ When I run the data contract phase
+ Then there are no file rejections from the data_contract phase
+ And there are no record rejections from the data_contract phase
+ When I run the business rules phase
+ Then there are errors with the following details and associated error_count from the business_rules phase
+ | ErrorType | Status | ErrorCode | error_count |
+ | record | error | StaffIDMissing | 7 |
+ | record | error | AirportHasNoStaff | 1 |
+ | record | error | FlightHasNoAirport | 1 |
+ | record | error | PassengerHasNoFlight | 1 |
+ And the final entities have the following row counts
+ | entity_name | row_count |
+ | country | 1 |
+ | airport | 1 |
+ | staff | 1 |
+ | flights | 1 |
+ | passengers | 1 |
+ When I run the error report phase
+ Then An error report is produced
+ And The statistics entry for the submission shows the following information
+ | parameter | value |
+ | record_count | 1 |
+ | number_submission_rejections | 0 |
+ | number_record_rejections | 10 |
+ | number_warnings | 0 |
+
+ Scenario: A flights submission with many types of rejections in a single submission (spark)
+ Given I submit the flights file flights_full_regression.xml for processing
+ And A spark pipeline is configured with schema file 'flights_spark.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 errors with the following details and associated error_count from the data_contract phase
+ | FailureType | ErrorCode | error_count |
+ | record | AirportIdIsMissing | 1 |
+ When I run the business rules phase
+ Then there are errors with the following details and associated error_count from the business_rules phase
+ | ErrorType | Status | ErrorCode | error_count |
+ | record | error | InvalidFlightDestination | 1 |
+ | record | error | PassengerNameMissing | 1 |
+ | record | error | StaffIDMissing | 4 |
+ | record | error | PassengerHasNoFlight | 3 |
+ | record | error | StaffHasNoAirport | 1 |
+ | record | error | FlightHasNoAirport | 2 |
+ | record | error | AirportHasNoStaff | 1 |
+ And the final entities have the following row counts
+ | entity_name | row_count |
+ | country | 1 |
+ | airport | 3 |
+ | staff | 3 |
+ | flights | 3 |
+ | passengers | 2 |
+ When I run the error report phase
+ Then An error report is produced
+ And The statistics entry for the submission shows the following information
+ | parameter | value |
+ | record_count | 1 |
+ | number_submission_rejections | 0 |
+ | number_record_rejections | 14 |
+ | number_warnings | 0 |
diff --git a/tests/testdata/flights/flights_spark.dischema.json b/tests/testdata/flights/flights_spark.dischema.json
new file mode 100644
index 00000000..b318f823
--- /dev/null
+++ b/tests/testdata/flights/flights_spark.dischema.json
@@ -0,0 +1,212 @@
+{
+ "contract": {
+ "schemas": {
+ "passengers": {
+ "fields": {
+ "flight_id": "int",
+ "passenger_id": "int",
+ "passenger_name": "str"
+ }
+ }
+ },
+ "error_details": "flights_data_contract_error_details.json",
+ "datasets": {
+ "country": {
+ "fields": {
+ "country_id": "int",
+ "country_name": "str"
+ },
+ "reader_config": {
+ ".xml": {
+ "reader": "SparkXMLStreamReader",
+ "kwargs": {
+ "record_tag": "country",
+ "root_tag": "country"
+ }
+ }
+ },
+ "key_field": "country_id",
+ "mandatory_fields": [
+ "country_id",
+ "country_name"
+ ]
+ },
+ "airport": {
+ "fields": {
+ "country_id": "int",
+ "airport_id": "int",
+ "airport_name": "str",
+ "postcode": "str"
+ },
+ "reader_config": {
+ ".xml": {
+ "reader": "SparkXMLStreamReader",
+ "kwargs": {
+ "record_tag": "airport",
+ "root_tag": "country"
+ }
+ }
+ },
+ "key_field": "airport_id",
+ "mandatory_fields": [
+ "airport_id"
+ ]
+ },
+ "staff": {
+ "fields": {
+ "airport_id": "int",
+ "staff_id": "int",
+ "staff_name": "str",
+ "role": "str"
+ },
+ "reader_config": {
+ ".xml": {
+ "reader": "SparkXMLStreamReader",
+ "kwargs": {
+ "record_tag": "staff_member",
+ "root_tag": "country"
+ }
+ }
+ },
+ "key_field": "staff_id"
+ },
+ "flights": {
+ "fields": {
+ "airport_id": "int",
+ "flight_id": "int",
+ "destination": "str"
+ },
+ "reader_config": {
+ ".xml": {
+ "reader": "SparkXMLStreamReader",
+ "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": "SparkXMLStreamReader",
+ "kwargs": {
+ "record_tag": "passenger",
+ "root_tag": "country"
+ }
+ }
+ },
+ "key_field": "passenger_id"
+ }
+ }
+ },
+ "transformations": {
+ "parameters": {
+ "entity": "country"
+ },
+ "filters": [
+ {
+ "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": "FlightIDMissing"
+ },
+ {
+ "entity": "flights",
+ "name": "invalid_destination",
+ "expression": "lower(destination) IN ('paris', 'madrid', 'new york', 'amsterdam', 'rome', 'dubai', 'dublin', 'lisbon', 'toronto')",
+ "failure_type": "record",
+ "failure_message": "Record Rejected - {{ destination }} is not a valid destination",
+ "reporting_field": "flight_id",
+ "reporting_entity": "flights",
+ "category": "Bad value",
+ "error_code": "InvalidFlightDestination"
+ },
+ {
+ "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": "PassengerNameMissing"
+ },
+ {
+ "entity": "staff",
+ "name": "staff_id_is_null",
+ "expression": "staff_id IS NOT NULL",
+ "failure_type": "record",
+ "failure_message": "Record Rejected - staff_id is missing",
+ "reporting_field": "passenger_name",
+ "reporting_entity": "passengers",
+ "category": "Blank",
+ "error_code": "StaffIDMissing"
+ }
+ ]
+ },
+ "entity_relationships": {
+ "country": {
+ "is_root_entity": true,
+ "mandatory": true,
+ "empty_entity_error_code": "NoValidCountries",
+ "empty_entity_error_message": "File Rejected - There are no valid country records"
+ },
+ "airport": {
+ "parent_entity": "country",
+ "join_fields": {
+ "country_id": "country_id"
+ },
+ "mandatory": true,
+ "missing_parent_id_error_code": "AirportHasNoCountry",
+ "missing_parent_id_error_message": "Record rejected - No valid country id found for airport",
+ "no_valid_records_error_code": "CountryHasNoAirport",
+ "no_valid_records_error_message": "Group rejected - Unable to find any valid airports",
+ "empty_entity_error_code": "NoValidAirports",
+ "empty_entity_error_message": "File Rejected - There are no valid airport records"
+ },
+ "staff": {
+ "parent_entity": "airport",
+ "join_fields": {
+ "airport_id": "airport_id"
+ },
+ "mandatory": true,
+ "missing_parent_id_error_code": "StaffHasNoAirport",
+ "missing_parent_id_error_message": "Record rejected - No valid airport id found for staff. Airport ID = {{ airport_id }}, Staff ID = {{ staff_id }}",
+ "no_valid_records_error_code": "AirportHasNoStaff",
+ "no_valid_records_error_message": "Group rejected - Airport has no valid staff. Airport ID = {{ airport_id }}",
+ "empty_entity_error_code": "NoValidStaff",
+ "empty_entity_error_message": "File Rejected - There are no valid staff records"
+ },
+ "flights": {
+ "parent_entity": "airport",
+ "join_fields": {
+ "airport_id": "airport_id"
+ },
+ "mandatory": false,
+ "missing_parent_id_error_code": "FlightHasNoAirport",
+ "missing_parent_id_error_message": "Record Rejected - No valid airport found for flight"
+ },
+ "passengers": {
+ "parent_entity": "flights",
+ "join_fields": {
+ "flight_id": "flight_id"
+ },
+ "mandatory": false,
+ "missing_parent_id_error_code": "PassengerHasNoFlight",
+ "missing_parent_id_error_message": "Record rejected - No valid flight found for passenger"
+ }
+ }
+}
\ No newline at end of file
From a342af77cdb6cc2d02b843d84c4fde2814311831 Mon Sep 17 00:00:00 2001
From: George Robertson <50412379+georgeRobertson@users.noreply.github.com>
Date: Thu, 8 Oct 2026 14:00:26 +0100
Subject: [PATCH 28/37] fix: group rejection not filtering out invalid records
correctly (#175)
---
src/dve/core_engine/backends/base/rules.py | 4 +++-
1 file changed, 3 insertions(+), 1 deletion(-)
diff --git a/src/dve/core_engine/backends/base/rules.py b/src/dve/core_engine/backends/base/rules.py
index 6200e013..9f19c024 100644
--- a/src/dve/core_engine/backends/base/rules.py
+++ b/src/dve/core_engine/backends/base/rules.py
@@ -513,7 +513,9 @@ def process_node(node: HierarchyNode) -> bool:
for tree in entity_hierarchy.entity_trees.values():
for node in tree.iterate_lowest_descendent_up():
if node.parent_entity and node.mandatory:
- entity_issues_found[node.parent_entity] = process_node(node)
+ result = process_node(node)
+ if result:
+ entity_issues_found[node.parent_entity] = result
return [], entity_issues_found
From ddfdf64ac473a8202e06fbf18970124589177168 Mon Sep 17 00:00:00 2001
From: George Robertson <50412379+georgeRobertson@users.noreply.github.com>
Date: Fri, 9 Oct 2026 10:43:02 +0100
Subject: [PATCH 29/37] build: upgrade urllib3 from 2.7 to 2.8 (#178)
---
poetry.lock | 8 ++++----
pyproject.toml | 2 +-
2 files changed, 5 insertions(+), 5 deletions(-)
diff --git a/poetry.lock b/poetry.lock
index 18f789ec..78d904f7 100644
--- a/poetry.lock
+++ b/poetry.lock
@@ -3347,14 +3347,14 @@ files = [
[[package]]
name = "urllib3"
-version = "2.7.0"
+version = "2.8.0"
description = "HTTP library with thread-safe connection pooling, file post, and more."
optional = false
python-versions = ">=3.10"
groups = ["main", "dev", "test"]
files = [
- {file = "urllib3-2.7.0-py3-none-any.whl", hash = "sha256:9fb4c81ebbb1ce9531cce37674bbc6f1360472bc18ca9a553ede278ef7276897"},
- {file = "urllib3-2.7.0.tar.gz", hash = "sha256:231e0ec3b63ceb14667c67be60f2f2c40a518cb38b03af60abc813da26505f4c"},
+ {file = "urllib3-2.8.0-py3-none-any.whl", hash = "sha256:0cf3cae568d36aa9576b28dfb35f11328f1cb974ca7647d9475ebb86c75ac6e3"},
+ {file = "urllib3-2.8.0.tar.gz", hash = "sha256:63bf2ead4c879426ebf22ef2a781eeb4aa3b4ae798a0435506f8687fd5bb9b63"},
]
[package.extras]
@@ -3638,4 +3638,4 @@ type = ["pytest-mypy (>=1.0.1) ; platform_python_implementation != \"PyPy\""]
[metadata]
lock-version = "2.1"
python-versions = ">=3.10,<3.13"
-content-hash = "6d996888c9d149407fb885b454745bf763f9cba56404a29148ffe0b9fd57b0de"
+content-hash = "2345c7671a4598e4421d8e71efdf2ad53343494a72bbec7703e26b461e73d794"
diff --git a/pyproject.toml b/pyproject.toml
index 5cac0282..49b4f30a 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -47,7 +47,7 @@ pyarrow = "23.0.1"
pydantic = "2.13.4"
pyspark = ">=3.5.0,<=3.5.5"
typing_extensions = "4.15.0"
-urllib3 = "2.7.0" # dependency of boto3 & botocore
+urllib3 = "2.8.0" # dependency of boto3 & botocore
[tool.poetry.group.dev]
optional = true
From a32a01b339b4d29a9eea5b3ba75300716827aaee Mon Sep 17 00:00:00 2001
From: George Robertson <50412379+georgeRobertson@users.noreply.github.com>
Date: Fri, 9 Oct 2026 11:20:21 +0100
Subject: [PATCH 30/37] fix: refactor the join_header method in duckdb to deal
with out of bound date values (#177)
---
.../backends/implementations/duckdb/rules.py | 22 ++++++++-----
.../test_duckdb/test_rules.py | 32 +++++++++++++++++++
2 files changed, 46 insertions(+), 8 deletions(-)
diff --git a/src/dve/core_engine/backends/implementations/duckdb/rules.py b/src/dve/core_engine/backends/implementations/duckdb/rules.py
index 766142aa..2dd6a048 100644
--- a/src/dve/core_engine/backends/implementations/duckdb/rules.py
+++ b/src/dve/core_engine/backends/implementations/duckdb/rules.py
@@ -10,6 +10,7 @@
ConstantExpression,
DuckDBPyConnection,
DuckDBPyRelation,
+ FunctionExpression,
StarExpression,
)
from duckdb.typing import DuckDBPyType
@@ -22,7 +23,6 @@
)
from dve.core_engine.backends.exceptions import ConstraintError
from dve.core_engine.backends.implementations.duckdb.duckdb_helpers import (
- DDBStruct,
ddb_filter_contract_errors,
duckdb_get_entity_count,
duckdb_read_parquet,
@@ -361,8 +361,7 @@ def join_header(self, entities: DuckDBEntities, *, config: HeaderJoin) -> Messag
target_rel: DuckDBPyRelation = entities[config.target_name]
target_rel = target_rel.set_alias(config.target_name)
- target_rows = target_rel.pl().to_struct("header").to_list()
- n_target_rows = len(target_rows)
+ n_target_rows = target_rel.shape[0]
if n_target_rows != 1:
raise ConstraintError(
f"Unable to join header {config.target_name!r} to {config.entity_name!r} "
@@ -373,11 +372,18 @@ def join_header(self, entities: DuckDBEntities, *, config: HeaderJoin) -> Messag
),
)
- target_schema = DDBStruct(dict(zip(target_rel.columns, target_rel.dtypes)))() # type: ignore # pylint:disable=C0301
-
- joined_rel = source_rel.select(
- StarExpression(exclude=[]),
- ConstantExpression(target_rows[0]).cast(target_schema).alias(config.header_column_name),
+ joined_rel = (
+ source_rel
+ .cross(target_rel)
+ .select(
+ StarExpression(
+ exclude=[f"{config.target_name}.{c}" for c in target_rel.columns]
+ ),
+ FunctionExpression(
+ "struct_pack",
+ *[ColumnExpression(f"{config.target_name}.{c}") for c in target_rel.columns],
+ ).alias(config.header_column_name)
+ )
)
entities[config.new_entity_name or config.entity_name] = joined_rel
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 aae56967..ea1629ca 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
@@ -70,6 +70,13 @@ def value_literal_1_header(duckdb_connection) -> Iterator[DuckDBPyRelation]:
yield duckdb_connection.sql("SELECT 1 AS Value")
+@pytest.fixture(scope="function")
+def value_mixed_header(duckdb_connection) -> Iterator[DuckDBPyRelation]:
+ yield duckdb_connection.sql(
+ "SELECT make_date(0000,1,1) as test_date, 'test' as test_str, 123 as test_int, make_date(10000,1,1) as test_date2"
+ )
+
+
def test_column_addition(planets_rel: DuckDBPyRelation):
"""Test that columns can be added to entities."""
entities = EntityManager({"planets": planets_rel})
@@ -566,6 +573,31 @@ def test_header_join_planets(
assert planet_row == actual_row, "More than header changed as result of header join"
+def test_header_join_with_out_of_range_date(
+ planets_rel: DuckDBPyRelation,
+ value_mixed_header: DuckDBPyRelation
+):
+ """Test that a header with an invalid date does not error"""
+ header_join = HeaderJoin(
+ entity_name="planets",
+ target_name="header",
+ new_entity_name="planets_header",
+ header_column_name="_Header"
+ )
+ entities = EntityManager({
+ "planets": planets_rel,
+ "header": value_mixed_header,
+ })
+ DUCKDB_STEP_BACKEND.evaluate(entities, config=header_join)
+
+ assert entities.entities["planets_header"].select("_header").distinct().shape[0] == 1
+ assert entities.entities["planets_header"].shape[0] == 9
+ assert (
+ entities.entities["planets_header"].select("typeof(_header)").fetchone()[0]
+ == 'STRUCT(test_date DATE, test_str VARCHAR, test_int INTEGER, test_date2 DATE)'
+ )
+
+
def test_header_multi_rows_raises(
planets_rel: DuckDBPyRelation, value_literal_1_header: DuckDBPyRelation
):
From 06eaee659b06df993e2bc3b4eccc83f0a18fc33a Mon Sep 17 00:00:00 2001
From: George Robertson <50412379+georgeRobertson@users.noreply.github.com>
Date: Fri, 9 Oct 2026 11:33:35 +0100
Subject: [PATCH 31/37] build: bump virtualenv version to 21.7.13 to resolve
vulns (#179)
---
poetry.lock | 23 +++++++++--------------
pyproject.toml | 3 ++-
2 files changed, 11 insertions(+), 15 deletions(-)
diff --git a/poetry.lock b/poetry.lock
index 78d904f7..e647ec18 100644
--- a/poetry.lock
+++ b/poetry.lock
@@ -2838,23 +2838,18 @@ six = ">=1.5"
[[package]]
name = "python-discovery"
-version = "1.2.0"
+version = "1.6.0"
description = "Python interpreter discovery"
optional = false
python-versions = ">=3.8"
groups = ["dev"]
files = [
- {file = "python_discovery-1.2.0-py3-none-any.whl", hash = "sha256:1e108f1bbe2ed0ef089823d28805d5ad32be8e734b86a5f212bf89b71c266e4a"},
- {file = "python_discovery-1.2.0.tar.gz", hash = "sha256:7d33e350704818b09e3da2bd419d37e21e7c30db6e0977bb438916e06b41b5b1"},
+ {file = "python_discovery-1.6.0-py3-none-any.whl", hash = "sha256:d4e244cf17b8b29819ed78003d55fbacf86eda23425b075454fff9271b79377a"},
+ {file = "python_discovery-1.6.0.tar.gz", hash = "sha256:6393b4eae1be8b2182670635e7baff89ac21cb9f8e86fd1ff40c7b1144febb4c"},
]
[package.dependencies]
filelock = ">=3.15.4"
-platformdirs = ">=4.3.6,<5"
-
-[package.extras]
-docs = ["furo (>=2025.12.19)", "sphinx (>=9.1)", "sphinx-autodoc-typehints (>=3.6.3)", "sphinxcontrib-mermaid (>=2)"]
-testing = ["covdefaults (>=2.3)", "coverage (>=7.5.4)", "pytest (>=8.3.5)", "pytest-mock (>=3.14)", "setuptools (>=75.1)"]
[[package]]
name = "pytokens"
@@ -3365,21 +3360,21 @@ zstd = ["backports-zstd (>=1.0.0) ; python_version < \"3.14\""]
[[package]]
name = "virtualenv"
-version = "21.2.1"
+version = "21.7.13"
description = "Virtual Python Environment builder"
optional = false
-python-versions = ">=3.8"
+python-versions = ">=3.9"
groups = ["dev"]
files = [
- {file = "virtualenv-21.2.1-py3-none-any.whl", hash = "sha256:bd16b49c53562b28cf1a3ad2f36edb805ad71301dee70ddc449e5c88a9f919a2"},
- {file = "virtualenv-21.2.1.tar.gz", hash = "sha256:b66ffe81301766c0d5e2208fc3576652c59d44e7b731fc5f5ed701c9b537fa78"},
+ {file = "virtualenv-21.7.13-py3-none-any.whl", hash = "sha256:1bea5af7463f59c4719db48fe739579a2a4f569c96f26c086edda85c96da9f59"},
+ {file = "virtualenv-21.7.13.tar.gz", hash = "sha256:0355558b6f33619aab31347e43643b0ebc97f61ea3acf617b2b69e1f8a843d11"},
]
[package.dependencies]
distlib = ">=0.3.7,<1"
filelock = {version = ">=3.24.2,<4", markers = "python_version >= \"3.10\""}
platformdirs = ">=3.9.1,<5"
-python-discovery = ">=1"
+python-discovery = ">=1.6"
typing-extensions = {version = ">=4.13.2", markers = "python_version < \"3.11\""}
[[package]]
@@ -3638,4 +3633,4 @@ type = ["pytest-mypy (>=1.0.1) ; platform_python_implementation != \"PyPy\""]
[metadata]
lock-version = "2.1"
python-versions = ">=3.10,<3.13"
-content-hash = "2345c7671a4598e4421d8e71efdf2ad53343494a72bbec7703e26b461e73d794"
+content-hash = "2090fe698ede57b0910f35a1701b8fe66bbcd69ffa72412597afcab95f4876b8"
diff --git a/pyproject.toml b/pyproject.toml
index 49b4f30a..6970281f 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -60,8 +60,9 @@ include-groups = [
commitizen = "4.9.1"
pre-commit = "4.3.0"
charset-normalizer = "3.4.6"
-python-discovery = "1.2.0"
+python-discovery = "1.6.0"
idna = "3.15" # dependency of requests
+virtualenv = "21.7.13"
[tool.poetry.group.test]
optional = true
From 5bc3eda87d3f7e6db7e02f6814d5e3c2eae5a323 Mon Sep 17 00:00:00 2001
From: georgeRobertson <50412379+georgeRobertson@users.noreply.github.com>
Date: Mon, 12 Oct 2026 11:37:37 +0100
Subject: [PATCH 32/37] docs: add udfs to docs
refactor the advanced docs section
---
.../package_documentation/auditing.md | 1 +
.../business_rules/index.md | 31 +++++++++++++++
.../{ => business_rules}/operations.md | 0
.../{ => business_rules}/refdata_loaders.md | 0
.../reference_data_types.md} | 0
.../business_rules/udfs.md | 10 +++++
.../{ => data_contract}/domain_types.md | 0
.../data_contract/index.md | 20 ++++++++++
.../{ => data_contract}/readers.md | 0
.../package_documentation/entity_hierarchy.md | 2 +-
.../feedback/feedback_messages.md | 9 +++++
.../package_documentation/feedback/index.md | 13 +++++++
.../feedback_messages.md | 12 ------
.../package_documentation/index.md | 38 +++++++++++++++++--
.../package_documentation/other/index.md | 13 +++++++
.../{ => other}/models.md | 4 +-
docs/user_guidance/business_rules.md | 24 +++++++++++-
docs/user_guidance/data_contract.md | 2 +-
docs/user_guidance/feedback_messages.md | 2 +-
docs/user_guidance/file_transformation.md | 2 +-
docs/user_guidance/implementations/duckdb.md | 6 +--
docs/user_guidance/implementations/spark.md | 2 +-
zensical.toml | 31 ++++++++-------
23 files changed, 182 insertions(+), 40 deletions(-)
create mode 100644 docs/advanced_guidance/package_documentation/business_rules/index.md
rename docs/advanced_guidance/package_documentation/{ => business_rules}/operations.md (100%)
rename docs/advanced_guidance/package_documentation/{ => business_rules}/refdata_loaders.md (100%)
rename docs/advanced_guidance/package_documentation/{refence_data_types.md => business_rules/reference_data_types.md} (100%)
create mode 100644 docs/advanced_guidance/package_documentation/business_rules/udfs.md
rename docs/advanced_guidance/package_documentation/{ => data_contract}/domain_types.md (100%)
create mode 100644 docs/advanced_guidance/package_documentation/data_contract/index.md
rename docs/advanced_guidance/package_documentation/{ => data_contract}/readers.md (100%)
create mode 100644 docs/advanced_guidance/package_documentation/feedback/feedback_messages.md
create mode 100644 docs/advanced_guidance/package_documentation/feedback/index.md
delete mode 100644 docs/advanced_guidance/package_documentation/feedback_messages.md
create mode 100644 docs/advanced_guidance/package_documentation/other/index.md
rename docs/advanced_guidance/package_documentation/{ => other}/models.md (51%)
diff --git a/docs/advanced_guidance/package_documentation/auditing.md b/docs/advanced_guidance/package_documentation/auditing.md
index a043e9ae..78230195 100644
--- a/docs/advanced_guidance/package_documentation/auditing.md
+++ b/docs/advanced_guidance/package_documentation/auditing.md
@@ -11,3 +11,4 @@
- get_all_business_rule_submissions
- get_all_error_report_submissions
- get_current_processing_info
+ show_root_toc_entry: false
diff --git a/docs/advanced_guidance/package_documentation/business_rules/index.md b/docs/advanced_guidance/package_documentation/business_rules/index.md
new file mode 100644
index 00000000..e64ed66e
--- /dev/null
+++ b/docs/advanced_guidance/package_documentation/business_rules/index.md
@@ -0,0 +1,31 @@
+---
+title: Business Rules Documentation
+---
+
+
+
+- :material-language-python:{ .lg .middle } __Operations__
+
+ ---
+
+ [:octicons-arrow-right-24: Read about the `Operation` objects here](operations.md)
+
+- :material-language-python:{ .lg .middle } __Reference Data Loaders__
+
+ ---
+
+ [:octicons-arrow-right-24: Read about the `Business Rules` objects here](refdata_loaders.md)
+
+- :material-language-python:{ .lg .middle } __Reference Data Types__
+
+ ---
+
+ [:octicons-arrow-right-24: Read about the `Data Contract` objects here](reference_data_types.md)
+
+- :material-language-python:{ .lg .middle } __User Defined Functions (UDFs)__
+
+ ---
+
+ [:octicons-arrow-right-24: Read about the `Submission Feedback` objects here](udfs.md)
+
+
\ No newline at end of file
diff --git a/docs/advanced_guidance/package_documentation/operations.md b/docs/advanced_guidance/package_documentation/business_rules/operations.md
similarity index 100%
rename from docs/advanced_guidance/package_documentation/operations.md
rename to docs/advanced_guidance/package_documentation/business_rules/operations.md
diff --git a/docs/advanced_guidance/package_documentation/refdata_loaders.md b/docs/advanced_guidance/package_documentation/business_rules/refdata_loaders.md
similarity index 100%
rename from docs/advanced_guidance/package_documentation/refdata_loaders.md
rename to docs/advanced_guidance/package_documentation/business_rules/refdata_loaders.md
diff --git a/docs/advanced_guidance/package_documentation/refence_data_types.md b/docs/advanced_guidance/package_documentation/business_rules/reference_data_types.md
similarity index 100%
rename from docs/advanced_guidance/package_documentation/refence_data_types.md
rename to docs/advanced_guidance/package_documentation/business_rules/reference_data_types.md
diff --git a/docs/advanced_guidance/package_documentation/business_rules/udfs.md b/docs/advanced_guidance/package_documentation/business_rules/udfs.md
new file mode 100644
index 00000000..6a994a27
--- /dev/null
+++ b/docs/advanced_guidance/package_documentation/business_rules/udfs.md
@@ -0,0 +1,10 @@
+
+!!! warning
+
+ Consider limiting how many UDFs you utilise in your rules as they can cause significant performance degredation when used on large datasets.
+
+::: dve.core_engine.functions.implementations
+ handler: python
+ options:
+ show_root_heading: true
+ heading_level: 1
diff --git a/docs/advanced_guidance/package_documentation/domain_types.md b/docs/advanced_guidance/package_documentation/data_contract/domain_types.md
similarity index 100%
rename from docs/advanced_guidance/package_documentation/domain_types.md
rename to docs/advanced_guidance/package_documentation/data_contract/domain_types.md
diff --git a/docs/advanced_guidance/package_documentation/data_contract/index.md b/docs/advanced_guidance/package_documentation/data_contract/index.md
new file mode 100644
index 00000000..df4bf23b
--- /dev/null
+++ b/docs/advanced_guidance/package_documentation/data_contract/index.md
@@ -0,0 +1,20 @@
+---
+title: Data Contract Documentation
+---
+
+
+
+- :material-language-python:{ .lg .middle } __Readers__
+
+ ---
+
+ [:octicons-arrow-right-24: Read about the `Operation` objects here](readers.md)
+
+- :material-language-python:{ .lg .middle } __Domain Types__
+
+ ---
+
+ [:octicons-arrow-right-24: Read about the `Business Rules` objects here](domain_types.md)
+
+
+
\ No newline at end of file
diff --git a/docs/advanced_guidance/package_documentation/readers.md b/docs/advanced_guidance/package_documentation/data_contract/readers.md
similarity index 100%
rename from docs/advanced_guidance/package_documentation/readers.md
rename to docs/advanced_guidance/package_documentation/data_contract/readers.md
diff --git a/docs/advanced_guidance/package_documentation/entity_hierarchy.md b/docs/advanced_guidance/package_documentation/entity_hierarchy.md
index 4080dcbe..b4f3ba48 100644
--- a/docs/advanced_guidance/package_documentation/entity_hierarchy.md
+++ b/docs/advanced_guidance/package_documentation/entity_hierarchy.md
@@ -2,4 +2,4 @@
handler: python
options:
show_root_heading: true
- heading_level: 2
+ heading_level: 1
diff --git a/docs/advanced_guidance/package_documentation/feedback/feedback_messages.md b/docs/advanced_guidance/package_documentation/feedback/feedback_messages.md
new file mode 100644
index 00000000..cfc279a6
--- /dev/null
+++ b/docs/advanced_guidance/package_documentation/feedback/feedback_messages.md
@@ -0,0 +1,9 @@
+
+::: dve.core_engine.message
+ handler: python
+ options:
+ heading_level: 2
+ members:
+ - FeedbackMessage
+ - UserMessage
+ show_root_toc_entry: false
diff --git a/docs/advanced_guidance/package_documentation/feedback/index.md b/docs/advanced_guidance/package_documentation/feedback/index.md
new file mode 100644
index 00000000..e0d810a2
--- /dev/null
+++ b/docs/advanced_guidance/package_documentation/feedback/index.md
@@ -0,0 +1,13 @@
+---
+title: User Feedback Documentation
+---
+
+
+
+- :material-language-python:{ .lg .middle } __Feedback Messages__
+
+ ---
+
+ [:octicons-arrow-right-24: Read about the `Operation` objects here](feedback_messages.md)
+
+
\ No newline at end of file
diff --git a/docs/advanced_guidance/package_documentation/feedback_messages.md b/docs/advanced_guidance/package_documentation/feedback_messages.md
deleted file mode 100644
index ede7aad8..00000000
--- a/docs/advanced_guidance/package_documentation/feedback_messages.md
+++ /dev/null
@@ -1,12 +0,0 @@
-
-::: dve.core_engine.message.FeedbackMessage
- handler: python
- options:
- show_root_heading: true
- heading_level: 2
-
-::: dve.core_engine.message.UserMessage
- handler: python
- options:
- show_root_heading: true
- heading_level: 2
diff --git a/docs/advanced_guidance/package_documentation/index.md b/docs/advanced_guidance/package_documentation/index.md
index c2e76c50..3c02ad0e 100644
--- a/docs/advanced_guidance/package_documentation/index.md
+++ b/docs/advanced_guidance/package_documentation/index.md
@@ -4,16 +4,46 @@ title: Package Documentation
-- :material-language-python:{ .lg .middle } __Pipeline__
+- :material-database-plus:{ .lg .middle } __Auditing__
---
- [:octicons-arrow-right-24: Read about the `Pipeline` objects here](pipeline.md)
+ [:octicons-arrow-right-24: Read about the `Auditing` objects here](auditing.md)
+
+- :material-clipboard-check:{ .lg .middle } __Business Rules__
+
+ ---
+
+ [:octicons-arrow-right-24: Read about the `Business Rules` objects here](./business_rules/index.md)
+
+- :material-file-document-check:{ .lg .middle } __Data Contract__
+
+ ---
+
+ [:octicons-arrow-right-24: Read about the `Data Contract` objects here](./data_contract/index.md)
+
+- :material-relation-one-to-many:{ .lg .middle } __Entity Relationships__
+
+ ---
+
+ [:octicons-arrow-right-24: Read about the `Entity Reliationships` objects here](entity_hierarchy.md)
+
+- :material-message-alert:{ .lg .middle } __Submission Feedback__
+
+ ---
+
+ [:octicons-arrow-right-24: Read about the `Submission Feedback` objects here](./feedback/index.md)
-- :material-set-all:{ .lg .middle } __Reference Data Loader__
+- :material-package-variant-plus:{ .lg .middle } __Additional Objects__
---
- [:octicons-arrow-right-24: Read about the `Reference Data Loader` objects here](refdata_loaders.md)
+ [:octicons-arrow-right-24: Read about the additional objects here](./other/index.md)
+
+- :material-pipe:{ .lg .middle } __Pipeline__
+
+ ---
+
+ [:octicons-arrow-right-24: Read about the `Pipeline` objects here](pipeline.md)
\ No newline at end of file
diff --git a/docs/advanced_guidance/package_documentation/other/index.md b/docs/advanced_guidance/package_documentation/other/index.md
new file mode 100644
index 00000000..bfb3a4e2
--- /dev/null
+++ b/docs/advanced_guidance/package_documentation/other/index.md
@@ -0,0 +1,13 @@
+---
+title: Additional Objects Documentation
+---
+
+
+
+- :material-code-braces-box:{ .lg .middle } __Models__
+
+ ---
+
+ [:octicons-arrow-right-24: Read about the Model objects here](./models.md)
+
+
diff --git a/docs/advanced_guidance/package_documentation/models.md b/docs/advanced_guidance/package_documentation/other/models.md
similarity index 51%
rename from docs/advanced_guidance/package_documentation/models.md
rename to docs/advanced_guidance/package_documentation/other/models.md
index 42e94202..0fc02050 100644
--- a/docs/advanced_guidance/package_documentation/models.md
+++ b/docs/advanced_guidance/package_documentation/other/models.md
@@ -2,5 +2,7 @@
::: dve.core_engine.models
handler: python
options:
- show_root_heading: true
heading_level: 2
+ members:
+ - SubmissionInfo
+ show_root_toc_entry: false
diff --git a/docs/user_guidance/business_rules.md b/docs/user_guidance/business_rules.md
index 0f4093f5..9ea21a12 100644
--- a/docs/user_guidance/business_rules.md
+++ b/docs/user_guidance/business_rules.md
@@ -209,7 +209,27 @@ The difference between modifiying the existing entity and adding a new one is si
### Operations
-For a full list of operations that you can perform during the pre-steps see [Advanced User Guidance: Operations](../advanced_guidance/package_documentation/operations.md).
+For a full list of operations that you can perform during the pre-steps see [Advanced User Guidance: Operations](../advanced_guidance/package_documentation/business_rules/operations.md).
+
+### User Defined Functions (UDFs)
+
+!!! warning
+
+ Consider limiting how many UDFs you utilise in your rules as they can cause significant performance degredation when used on large datasets.
+
+
+The DVE currently has a number of UDFs available to be called during a rule check. For instance, if you wanted to use the [NHS Number](https://www.datadictionary.nhs.uk/attributes/nhs_number.html#ariaid-title2) mod 11 check then you can simply call the udf during the expression definition - such as:
+
+```json title="example.dischema.json"
+ {
+ "entity": "example",
+ "name": "check NHS number passes mod 11 check",
+ "expression": "nhsno_mod11_check()"
+ ...
+ }
+```
+
+A full list of UDF functions available can be found [here](../advanced_guidance/package_documentation/business_rules/udfs.md).
## Post Rule
@@ -358,7 +378,7 @@ If your Business Rules are reliant on reference data, then you can add the `"ref
- Refdata entities are also immutable. So, if you need to modify them in any way, you will always need to create a new entity from it
-For latest supported reference data types, see [Advanced User Guidance: Reference Data Types](../advanced_guidance/package_documentation/refence_data_types.md).
+For latest supported reference data types, see [Advanced User Guidance: Reference Data Types](../advanced_guidance/package_documentation/business_rules/reference_data_types.md).
## Complex Rules
diff --git a/docs/user_guidance/data_contract.md b/docs/user_guidance/data_contract.md
index 13e72efa..937fed10 100644
--- a/docs/user_guidance/data_contract.md
+++ b/docs/user_guidance/data_contract.md
@@ -278,7 +278,7 @@ As you can see, we can set the "type" for several fields to `MyConstrainedString
#### Domain Types
-Domain types are custom Pydantic model types available with the DVE. Current Domain types available are `Postcode`, `NHSNumber`, `FormattedDatetime` etc. You can find the full list of Domain Types [here](../advanced_guidance/package_documentation/domain_types.md).
+Domain types are custom Pydantic model types available with the DVE. Current Domain types available are `Postcode`, `NHSNumber`, `FormattedDatetime` etc. You can find the full list of Domain Types [here](../advanced_guidance/package_documentation/data_contract/domain_types.md).
### Complex Types
diff --git a/docs/user_guidance/feedback_messages.md b/docs/user_guidance/feedback_messages.md
index ee812886..01128d27 100644
--- a/docs/user_guidance/feedback_messages.md
+++ b/docs/user_guidance/feedback_messages.md
@@ -28,4 +28,4 @@ If the DVE is crashing and the error message is either unreadable or is crashing
## Feedback Message Models
-Please refer to [Advanced User Guidance: Feedback Messages](../advanced_guidance/package_documentation/feedback_messages.md).
+Please refer to [Advanced User Guidance: Feedback Messages](../advanced_guidance/package_documentation/feedback/feedback_messages.md).
diff --git a/docs/user_guidance/file_transformation.md b/docs/user_guidance/file_transformation.md
index 41dc7de4..411b1e0f 100644
--- a/docs/user_guidance/file_transformation.md
+++ b/docs/user_guidance/file_transformation.md
@@ -155,7 +155,7 @@ The secondary use of the File Transformation stage is the ability to normalise y
```
!!! abstract ""
- You can read more about the readers and kwargs [here](../advanced_guidance/package_documentation/readers.md).
+ You can read more about the readers and kwargs [here](../advanced_guidance/package_documentation/data_contract/readers.md).
## Supported Formats
diff --git a/docs/user_guidance/implementations/duckdb.md b/docs/user_guidance/implementations/duckdb.md
index ac75c587..48448e35 100644
--- a/docs/user_guidance/implementations/duckdb.md
+++ b/docs/user_guidance/implementations/duckdb.md
@@ -36,7 +36,7 @@ Now you have the DuckDB connection object setup, you are ready to setup the requ
## Generating SubmissionInfo objects
-Before we utilise the DVE, we need to generate an iterable object containing `SubmissionInfo` objects. These objects effectively contain the necessery metadata for the DVE to work with a given submission. Here is an example function used to generate [SubmissionInfo](../../advanced_guidance/package_documentation/models.md#dve.core_engine.models.SubmissionInfo) objects from a given path:
+Before we utilise the DVE, we need to generate an iterable object containing `SubmissionInfo` objects. These objects effectively contain the necessery metadata for the DVE to work with a given submission. Here is an example function used to generate [SubmissionInfo](../../advanced_guidance/package_documentation/other/models.md#dve.core_engine.models.SubmissionInfo) objects from a given path:
```py
import glob
from datetime import date, datetime
@@ -112,7 +112,7 @@ DuckDBRefDataLoader.dataset_config_uri = Path("path", "to", "my", "rules").as_po
The connection passed into the `DuckDBRefDataLoader` object will then be able to use various DuckDB readers to load data from an existing table on the connection OR loading data from reference data persisted in either `parquet` or `pyarrow` format.
-If you want to learn more about the reference data loaders, you can view the advanced user guidance [here](../../advanced_guidance/package_documentation/refdata_loaders.md).
+If you want to learn more about the reference data loaders, you can view the advanced user guidance [here](../../advanced_guidance/package_documentation/business_rules/refdata_loaders.md).
Now we can move onto setting up the DuckDB DVE Pipeline object.
@@ -171,4 +171,4 @@ error_reports = dve_pipeline.cluster_pipeline_run()
For further details on the objects referenced above, you can use the following links to read more about the objects:
- [Pipeline Docs](../../advanced_guidance/package_documentation/pipeline.md)
-- [Reference Data Docs](../../advanced_guidance/package_documentation/refdata_loaders.md)
+- [Reference Data Docs](../../advanced_guidance/package_documentation/business_rules/refdata_loaders.md)
diff --git a/docs/user_guidance/implementations/spark.md b/docs/user_guidance/implementations/spark.md
index 6bbdb300..866460e0 100644
--- a/docs/user_guidance/implementations/spark.md
+++ b/docs/user_guidance/implementations/spark.md
@@ -187,4 +187,4 @@ error_reports = dve_pipeline.cluster_pipeline_run()
For further details on the objects referenced above, you can use the following links to read more about the objects:
- [Pipeline Docs](../../advanced_guidance/package_documentation/pipeline.md)
-- [Reference Data Docs](../../advanced_guidance/package_documentation/refdata_loaders.md)
+- [Reference Data Docs](../../advanced_guidance/package_documentation/business_rules/refdata_loaders.md)
diff --git a/zensical.toml b/zensical.toml
index 0141b6df..fc47343e 100644
--- a/zensical.toml
+++ b/zensical.toml
@@ -44,28 +44,33 @@ nav = [
"advanced_guidance/index.md",
{"DVE Package Documentation" = [
"advanced_guidance/package_documentation/index.md",
- {"Pipeline" = "advanced_guidance/package_documentation/pipeline.md"},
{"Auditing" = "advanced_guidance/package_documentation/auditing.md"},
- {"Data Contract" = [
- {"Readers" = "advanced_guidance/package_documentation/readers.md"},
- {"Domain Types" = "advanced_guidance/package_documentation/domain_types.md"},
- ]},
{"Business Rules" = [
+ "advanced_guidance/package_documentation/business_rules/index.md",
{"Rules" = [
- {"Operations" = "advanced_guidance/package_documentation/operations.md"},
+ {"Operations" = "advanced_guidance/package_documentation/business_rules/operations.md"},
]},
{"Refdata" = [
- {"Refdata Types" = "advanced_guidance/package_documentation/refence_data_types.md"},
- {"Refdata Loaders" = "advanced_guidance/package_documentation/refdata_loaders.md"},
+ {"Refdata Types" = "advanced_guidance/package_documentation/business_rules/reference_data_types.md"},
+ {"Refdata Loaders" = "advanced_guidance/package_documentation/business_rules/refdata_loaders.md"},
]},
- {"Entity Relationships" = [
- {"Entity Hierarchy" = "advanced_guidance/package_documentation/entity_hierarchy.md"}
- ]}
+ {"User Defined Functions (UDFs)" = "advanced_guidance/package_documentation/business_rules/udfs.md"},
]},
+ {"Data Contract" = [
+ "advanced_guidance/package_documentation/data_contract/index.md",
+ {"Readers" = "advanced_guidance/package_documentation/data_contract/readers.md"},
+ {"Domain Types" = "advanced_guidance/package_documentation/data_contract/domain_types.md"},
+ ]},
+ {"Entity Relationships" = "advanced_guidance/package_documentation/entity_hierarchy.md"},
{"Feedback" = [
- {"Feedback Messages" = "advanced_guidance/package_documentation/feedback_messages.md"},
+ "advanced_guidance/package_documentation/feedback/index.md",
+ {"Feedback Messages" = "advanced_guidance/package_documentation/feedback/feedback_messages.md"},
+ ]},
+ {"Other" = [
+ "advanced_guidance/package_documentation/other/index.md",
+ {"Core Models" = "advanced_guidance/package_documentation/other/models.md"},
]},
- {"Models" = "advanced_guidance/package_documentation/models.md"},
+ {"Pipeline" = "advanced_guidance/package_documentation/pipeline.md"},
]},
{"DVE Developer Guidance" = [
{"Implementing a new backend" = "advanced_guidance/new_backend.md"},
From 72d15bed72af818ce5ba2d28c377d9fe910bb60a Mon Sep 17 00:00:00 2001
From: stevenhsd <56357022+stevenhsd@users.noreply.github.com>
Date: Mon, 12 Oct 2026 12:01:35 +0100
Subject: [PATCH 33/37] * fix: ensure that null records are not included in
record counts when determining if entity empty (#182)
* test: additional tests with record index for entity empty check functions
---
.../implementations/duckdb/duckdb_helpers.py | 12 +++++---
.../backends/implementations/duckdb/rules.py | 18 ++++-------
.../implementations/spark/spark_helpers.py | 14 +++++++--
.../test_duckdb/test_duckdb_helpers.py | 13 ++++++++
.../test_spark/test_spark_helpers.py | 30 +++++++++++++++++++
5 files changed, 69 insertions(+), 18 deletions(-)
diff --git a/src/dve/core_engine/backends/implementations/duckdb/duckdb_helpers.py b/src/dve/core_engine/backends/implementations/duckdb/duckdb_helpers.py
index 55db0945..c25387fb 100644
--- a/src/dve/core_engine/backends/implementations/duckdb/duckdb_helpers.py
+++ b/src/dve/core_engine/backends/implementations/duckdb/duckdb_helpers.py
@@ -110,9 +110,13 @@ def get_all_existing_ddb_tables(connection: DuckDBPyConnection) -> tuple[str]:
def relation_is_empty(relation: DuckDBPyRelation) -> bool:
"""Check if a duckdb relation is empty"""
- if relation.limit(1).shape[0] > 0:
- return False
- return True
+ if relation.limit(1).shape[0] == 0:
+ return True
+
+ cl_all_null_check_exprs = [
+ f'bool_and("{cl}" is NULL)' for cl in relation.columns if cl != RECORD_INDEX_COLUMN_NAME
+ ]
+ return relation.aggregate(" AND ".join(cl_all_null_check_exprs)).fetchone()[0] # type: ignore
def get_duckdb_type_from_annotation(type_annotation: Any) -> DuckDBPyType:
@@ -329,7 +333,7 @@ def duckdb_get_entity_count(cls):
def _duckdb_check_entity_empty(self, entity: DuckDBPyRelation) -> bool: # pylint: disable=W0613
- return entity.shape[0] == 0
+ return relation_is_empty(entity)
def duckdb_check_entity_empty(cls):
diff --git a/src/dve/core_engine/backends/implementations/duckdb/rules.py b/src/dve/core_engine/backends/implementations/duckdb/rules.py
index 2dd6a048..187a880d 100644
--- a/src/dve/core_engine/backends/implementations/duckdb/rules.py
+++ b/src/dve/core_engine/backends/implementations/duckdb/rules.py
@@ -372,18 +372,12 @@ def join_header(self, entities: DuckDBEntities, *, config: HeaderJoin) -> Messag
),
)
- joined_rel = (
- source_rel
- .cross(target_rel)
- .select(
- StarExpression(
- exclude=[f"{config.target_name}.{c}" for c in target_rel.columns]
- ),
- FunctionExpression(
- "struct_pack",
- *[ColumnExpression(f"{config.target_name}.{c}") for c in target_rel.columns],
- ).alias(config.header_column_name)
- )
+ joined_rel = source_rel.cross(target_rel).select(
+ StarExpression(exclude=[f"{config.target_name}.{c}" for c in target_rel.columns]),
+ FunctionExpression(
+ "struct_pack",
+ *[ColumnExpression(f"{config.target_name}.{c}") for c in target_rel.columns],
+ ).alias(config.header_column_name),
)
entities[config.new_entity_name or config.entity_name] = joined_rel
diff --git a/src/dve/core_engine/backends/implementations/spark/spark_helpers.py b/src/dve/core_engine/backends/implementations/spark/spark_helpers.py
index 1257306e..f3ce03f4 100644
--- a/src/dve/core_engine/backends/implementations/spark/spark_helpers.py
+++ b/src/dve/core_engine/backends/implementations/spark/spark_helpers.py
@@ -359,7 +359,17 @@ def object_to_spark_literal(obj: SparkLiteralType) -> Column:
def df_is_empty(dataframe: DataFrame) -> bool:
"""Return whether a DataFrame is empty."""
- return not dataframe.select(lit(1)).head(1)
+
+ if not dataframe.select(lit(1)).head(1):
+ return True
+
+ return (
+ not dataframe.dropna(
+ how="all", subset=[cl for cl in dataframe.columns if cl != RECORD_INDEX_COLUMN_NAME]
+ )
+ .select(lit(1))
+ .head(1)
+ )
def _spark_read_parquet(self, path: URI, **kwargs) -> DataFrame:
@@ -457,7 +467,7 @@ def spark_get_entity_count(cls):
def _spark_check_entity_empty(self, entity: DataFrame) -> bool: # pylint: disable=W0613
- return entity.count() == 0
+ return df_is_empty(entity)
def spark_check_entity_empty(cls):
diff --git a/tests/test_core_engine/test_backends/test_implementations/test_duckdb/test_duckdb_helpers.py b/tests/test_core_engine/test_backends/test_implementations/test_duckdb/test_duckdb_helpers.py
index 0e81cf46..74b7b3a8 100644
--- a/tests/test_core_engine/test_backends/test_implementations/test_duckdb/test_duckdb_helpers.py
+++ b/tests/test_core_engine/test_backends/test_implementations/test_duckdb/test_duckdb_helpers.py
@@ -25,6 +25,7 @@
relation_is_empty
)
from dve.core_engine.backends.utilities import datetime_format_to_regex
+from dve.core_engine.constants import RECORD_INDEX_COLUMN_NAME
from dve.metadata_parser.domain_types import formatteddatetime
@pytest.fixture
@@ -266,6 +267,18 @@ def test_relation_is_empty(temp_ddb_conn: DuckDBPyConnection):
_, con = temp_ddb_conn
rel = con.sql("SELECT 'abc' AS test").filter("test IS NULL")
assert relation_is_empty(rel)
+
+ rel2 = con.sql("SELECT NULL as fld1, NULL as fld2 UNION ALL SELECT NULL as fld1, NULL as fld2")
+ assert relation_is_empty(rel2)
+
+ rel3 = con.sql("SELECT 1 as fld1, NULL as fld2 UNION ALL SELECT NULL as fld1, NULL as fld2")
+ assert not relation_is_empty(rel3)
+
+ rel4 = con.sql(f"SELECT 1 as {RECORD_INDEX_COLUMN_NAME}, NULL as fld1, NULL as fld2 UNION ALL SELECT 2, NULL, NULL")
+ assert relation_is_empty(rel4)
+
+ rel5 = con.sql(f"SELECT 1 as {RECORD_INDEX_COLUMN_NAME}, NULL as fld1, NULL as fld2 UNION ALL SELECT 2, NULL, 'hello'")
+ assert not relation_is_empty(rel5)
@pytest.mark.parametrize("date_format, expected_regex", [("%Y-%m-%d", r"^[0-9]{4}\-[0-9]{2}\-[0-9]{2}$"),
("%Y-%m-%dT%H:%M:%S", r"^[0-9]{4}\-[0-9]{2}\-[0-9]{2}T[0-9]{2}:[0-9]{2}:[0-9]{2}$"),
diff --git a/tests/test_core_engine/test_backends/test_implementations/test_spark/test_spark_helpers.py b/tests/test_core_engine/test_backends/test_implementations/test_spark/test_spark_helpers.py
index 41c9e93e..34ecb6ca 100644
--- a/tests/test_core_engine/test_backends/test_implementations/test_spark/test_spark_helpers.py
+++ b/tests/test_core_engine/test_backends/test_implementations/test_spark/test_spark_helpers.py
@@ -30,7 +30,9 @@
get_type_from_annotation,
object_to_spark_literal,
python_to_java_datetime_format,
+ df_is_empty
)
+from dve.core_engine.constants import RECORD_INDEX_COLUMN_NAME
from dve.metadata_parser.domain_types import conformatteddate
from .....fixtures import spark # pylint: disable=unused-import
@@ -354,3 +356,31 @@ def test_spark_filter_contract_errors(spark: SparkSession, example_data_contract
("%Y-%m-%dT%H:%M:%S", "yyyy-MM-dd'T'HH:mm:ss")])
def test_python_to_java_datetime_format(python_dt, expected_java_dt):
assert python_to_java_datetime_format(python_dt) == expected_java_dt
+
+
+def test_df_is_empty(spark):
+ df = spark.createDataFrame([], schema=StructType([StructField("col1", StringType()), StructField("col2", LongType())]))
+ assert df_is_empty(df)
+ df2 = spark.createDataFrame([(None, None), (None, None), (None, None)],
+ schema=StructType([StructField("col1", StringType()), StructField("col2", LongType())]))
+ assert df_is_empty(df2)
+ df3 = spark.createDataFrame([(None, None), (None, None), (None, 3)],
+ schema=StructType([StructField("col1", StringType()), StructField("col2", LongType())]))
+ assert not df_is_empty(df3)
+
+ _schema_w_ri = schema=StructType([
+ StructField(RECORD_INDEX_COLUMN_NAME, LongType()),
+ StructField("col1", StringType()),
+ StructField("col2", LongType())
+ ])
+
+ df4 = spark.createDataFrame([(1, None, None), (2, None, None), (3, None, None)],
+ schema=_schema_w_ri)
+ assert df_is_empty(df4)
+
+ df5 = df4.union(spark.createDataFrame([(4, None, 8)], schema=_schema_w_ri))
+ assert not df_is_empty(df5)
+
+
+
+
From 562ca2bcbbebc0c42112ccc6272ad80d7c6fa855 Mon Sep 17 00:00:00 2001
From: georgeRobertson <50412379+georgeRobertson@users.noreply.github.com>
Date: Mon, 12 Oct 2026 12:19:55 +0100
Subject: [PATCH 34/37] docs: fix incorrect text in some of the indexes
---
.../package_documentation/business_rules/index.md | 2 +-
.../package_documentation/data_contract/index.md | 4 ++--
2 files changed, 3 insertions(+), 3 deletions(-)
diff --git a/docs/advanced_guidance/package_documentation/business_rules/index.md b/docs/advanced_guidance/package_documentation/business_rules/index.md
index e64ed66e..0b1f3aea 100644
--- a/docs/advanced_guidance/package_documentation/business_rules/index.md
+++ b/docs/advanced_guidance/package_documentation/business_rules/index.md
@@ -26,6 +26,6 @@ title: Business Rules Documentation
---
- [:octicons-arrow-right-24: Read about the `Submission Feedback` objects here](udfs.md)
+ [:octicons-arrow-right-24: Read about the `UDFs` objects here](udfs.md)
\ No newline at end of file
diff --git a/docs/advanced_guidance/package_documentation/data_contract/index.md b/docs/advanced_guidance/package_documentation/data_contract/index.md
index df4bf23b..a801650d 100644
--- a/docs/advanced_guidance/package_documentation/data_contract/index.md
+++ b/docs/advanced_guidance/package_documentation/data_contract/index.md
@@ -8,13 +8,13 @@ title: Data Contract Documentation
---
- [:octicons-arrow-right-24: Read about the `Operation` objects here](readers.md)
+ [:octicons-arrow-right-24: Read about the `Readers` objects here](readers.md)
- :material-language-python:{ .lg .middle } __Domain Types__
---
- [:octicons-arrow-right-24: Read about the `Business Rules` objects here](domain_types.md)
+ [:octicons-arrow-right-24: Read about the `Domain Types` objects here](domain_types.md)
\ No newline at end of file
From 3c71d18d2bc4a46cbc74ec499705a5a508902945 Mon Sep 17 00:00:00 2001
From: georgeRobertson <50412379+georgeRobertson@users.noreply.github.com>
Date: Mon, 12 Oct 2026 12:30:53 +0100
Subject: [PATCH 35/37] refactor: removed ods udfs as they were functionally
useless
---
.../core_engine/functions/implementations.py | 110 ------------------
1 file changed, 110 deletions(-)
diff --git a/src/dve/core_engine/functions/implementations.py b/src/dve/core_engine/functions/implementations.py
index bae6d98b..9d5928d2 100644
--- a/src/dve/core_engine/functions/implementations.py
+++ b/src/dve/core_engine/functions/implementations.py
@@ -42,116 +42,6 @@ def date_in_current_financial_year(test_date: datetime.date) -> bool:
return test_date >= current_fiscal_year_start
-def is_valid_ods_code(check_ods_code: str) -> bool:
- """A test ODS code lookup, contains some actual valid codes"""
- if (not check_ods_code) or (not isinstance(check_ods_code, str)):
- return False
- valid_ods_codes = [
- "EE142976",
- "EE144430",
- "EE143473",
- "EE148112",
- "EE142863",
- "EE147862",
- "EE142472",
- "EE141208",
- "EE143149",
- "EE140862",
- "EE140319",
- "EE144899",
- "EE144475",
- "EE141850",
- "EE147934",
- "EE141068",
- "EE143825",
- "EE147805",
- "EE143489",
- "EE146813",
- "EE145703",
- "EE148295",
- "EE140156",
- "EE145502",
- "EE148396",
- "EE144126",
- "EE145590",
- "EE141566",
- "EE142081",
- "EE143640",
- "EE144911",
- "EE145935",
- "EE145279",
- "EE143156",
- "EE146556",
- "EE140781",
- "EE144734",
- "EE144841",
- "EE140419",
- "EE140040",
- "EE147342",
- "EE143330",
- "EE140926",
- "EE140926",
- "EE146438",
- "EE142137",
- "EE143856",
- "EE141067",
- "EE148534",
- "EE141310",
- "EE146899",
- "EE146996",
- "EE147487",
- "EE148447",
- "EE144311",
- "EE142147",
- "EE147605",
- "EE142117",
- "EE144087",
- "EE147326",
- "EE147614",
- "EE143703",
- "EE146135",
- "EE140782",
- "EE143603",
- "EE143554",
- "EE146659",
- "EE140321",
- "EE141185",
- "EE147648",
- "EE144527",
- "EE142680",
- "EE141620",
- "EE145274",
- "EE146251",
- "EE148209",
- "EE142574",
- "EE148162",
- "EE143118",
- "EE142977",
- "EE147798",
- "EE147902",
- "EE145780",
- "EE146992",
- "EE142916",
- "EE144777",
- "EE146935",
- "EE145586",
- "EE144570",
- "EE147122",
- "EE140874",
- "EE141338",
- "EE143244",
- ]
- return check_ods_code in valid_ods_codes
-
-
-def is_valid_national_org(check_org_code: str) -> bool:
- """Simple test org code lookup"""
- if (not check_org_code) or (not isinstance(check_org_code, str)):
- return False
- valid_org_codes = ["ORG01", "ORG02"]
- return check_org_code in valid_org_codes
-
-
def check_correct_numeric_signage(
val: Union[float, int, Decimal], expected_sign: str = "+/-"
) -> Optional[bool]:
From 54f704bfe6c3ceb8b6cd1474613cd1db7f007e09 Mon Sep 17 00:00:00 2001
From: stevenhsd <56357022+stevenhsd@users.noreply.github.com>
Date: Mon, 12 Oct 2026 22:00:13 +0100
Subject: [PATCH 36/37] fix: adjust regex check for timezone offsets, allowing
: to be included (#181)
---
.../implementations/spark/spark_helpers.py | 2 +-
src/dve/core_engine/backends/utilities.py | 3 +-
.../test_duckdb/test_duckdb_helpers.py | 29 +++++++++++++++----
.../test_spark/test_spark_helpers.py | 18 +++++++++++-
4 files changed, 43 insertions(+), 9 deletions(-)
diff --git a/src/dve/core_engine/backends/implementations/spark/spark_helpers.py b/src/dve/core_engine/backends/implementations/spark/spark_helpers.py
index f3ce03f4..5041095b 100644
--- a/src/dve/core_engine/backends/implementations/spark/spark_helpers.py
+++ b/src/dve/core_engine/backends/implementations/spark/spark_helpers.py
@@ -112,7 +112,7 @@ def __post_init__(self):
"%H": "HH",
"%M": "mm",
"%S": "ss",
- "%z": "XX",
+ "%z": "XXX",
"%Z": "z",
}
"""A mapping of python to java datetime component formats. Not exhaustive but will hopefully support
diff --git a/src/dve/core_engine/backends/utilities.py b/src/dve/core_engine/backends/utilities.py
index af2f7573..7cb625d4 100644
--- a/src/dve/core_engine/backends/utilities.py
+++ b/src/dve/core_engine/backends/utilities.py
@@ -40,7 +40,7 @@
"H": r"[0-9]{2}",
"M": r"[0-9]{2}",
"S": r"[0-9]{2}",
- "z": r"(\+|\-)?[0-9]+(\.[0-9]*)?",
+ "z": r"[+-][0-9]+(:[0-9]{2})?(\.[0-9]*)?",
"Z": r"[A-Z]{0,3}",
}
@@ -54,7 +54,6 @@ def datetime_format_to_regex(format_str: str) -> str:
Helper function to convert python datetime formats to regexp string checks for casting
purposes.
"""
-
tokens = list(format_str.replace("%", ""))
for idx, tkn in enumerate(tokens):
diff --git a/tests/test_core_engine/test_backends/test_implementations/test_duckdb/test_duckdb_helpers.py b/tests/test_core_engine/test_backends/test_implementations/test_duckdb/test_duckdb_helpers.py
index 74b7b3a8..58dd423a 100644
--- a/tests/test_core_engine/test_backends/test_implementations/test_duckdb/test_duckdb_helpers.py
+++ b/tests/test_core_engine/test_backends/test_implementations/test_duckdb/test_duckdb_helpers.py
@@ -26,7 +26,7 @@
)
from dve.core_engine.backends.utilities import datetime_format_to_regex
from dve.core_engine.constants import RECORD_INDEX_COLUMN_NAME
-from dve.metadata_parser.domain_types import formatteddatetime
+from dve.metadata_parser.domain_types import conformatteddate, formatteddatetime
@pytest.fixture
def casting_test_table(temp_ddb_conn):
@@ -225,8 +225,8 @@ def test_duckdb_rel_to_dictionaries(temp_ddb_conn: DuckDBPyConnection,
("date_test", datetime.date,"TRY_CAST(CASE WHEN REGEXP_FULL_MATCH(TRIM(\"date_test\"), '^[0-9]{4}\-[0-9]{2}\-[0-9]{2}$') THEN TRY_STRPTIME(TRIM(\"date_test\"), '%Y-%m-%d') ELSE NULL END as DATE)"),
("timestamp_test", datetime.datetime, "TRY_CAST(CASE WHEN REGEXP_FULL_MATCH(TRIM(\"timestamp_test\"), '^[0-9]{4}\-[0-9]{2}\-[0-9]{2}T[0-9]{2}:[0-9]{2}:[0-9]{2}$') THEN TRY_STRPTIME(TRIM(\"timestamp_test\"), '%Y-%m-%dT%H:%M:%S') ELSE NULL END as TIMESTAMP)"),
("list_int_field", list[int], "TRY_CAST(LIST_TRANSFORM(\"list_int_field\", x -> TRIM(\"x\")) as BIGINT[])"),
- ("basic_model", BasicModel, "TRY_CAST(STRUCT_PACK(\"str_field\":= TRIM(\"basic_model\".str_field),\"date_field\":= TRY_CAST(CASE WHEN REGEXP_FULL_MATCH(TRIM(\"basic_model\".date_field), '^[0-9]{4}\-[0-9]{2}\-[0-9]{2}$') THEN TRY_STRPTIME(TRIM(\"basic_model\".date_field), '%Y-%m-%d') ELSE NULL END as DATE),\"timestamp_field\":= TRY_CAST(CASE WHEN REGEXP_FULL_MATCH(TRIM(\"basic_model\".timestamp_field), '^[0-9]{2}/[0-9]{2}/[0-9]{4} [0-9]{2}:[0-9]{2}:[0-9]{2}(\+|\-)?[0-9]+(\.[0-9]*)?$') THEN TRY_STRPTIME(TRIM(\"basic_model\".timestamp_field), '%d/%m/%Y %H:%M:%S%z') ELSE NULL END as TIMESTAMP)) as STRUCT(str_field VARCHAR, date_field DATE, timestamp_field TIMESTAMP))"),
- ("another_model", AnotherModel, "TRY_CAST(STRUCT_PACK(\"unique_id\":= TRIM(\"another_model\".unique_id),\"basic_models\":= LIST_TRANSFORM(\"another_model\".basic_models, x -> STRUCT_PACK(\"str_field\":= TRIM(\"x\".str_field),\"date_field\":= TRY_CAST(CASE WHEN REGEXP_FULL_MATCH(TRIM(\"x\".date_field), '^[0-9]{4}\-[0-9]{2}\-[0-9]{2}$') THEN TRY_STRPTIME(TRIM(\"x\".date_field), '%Y-%m-%d') ELSE NULL END as DATE),\"timestamp_field\":= TRY_CAST(CASE WHEN REGEXP_FULL_MATCH(TRIM(\"x\".timestamp_field), '^[0-9]{2}/[0-9]{2}/[0-9]{4} [0-9]{2}:[0-9]{2}:[0-9]{2}(\+|\-)?[0-9]+(\.[0-9]*)?$') THEN TRY_STRPTIME(TRIM(\"x\".timestamp_field), '%d/%m/%Y %H:%M:%S%z') ELSE NULL END as TIMESTAMP)))) as STRUCT(unique_id BIGINT, basic_models STRUCT(str_field VARCHAR, date_field DATE, timestamp_field TIMESTAMP)[]))")])
+ ("basic_model", BasicModel, "TRY_CAST(STRUCT_PACK(\"str_field\":= TRIM(\"basic_model\".str_field),\"date_field\":= TRY_CAST(CASE WHEN REGEXP_FULL_MATCH(TRIM(\"basic_model\".date_field), '^[0-9]{4}\-[0-9]{2}\-[0-9]{2}$') THEN TRY_STRPTIME(TRIM(\"basic_model\".date_field), '%Y-%m-%d') ELSE NULL END as DATE),\"timestamp_field\":= TRY_CAST(CASE WHEN REGEXP_FULL_MATCH(TRIM(\"basic_model\".timestamp_field), '^[0-9]{2}/[0-9]{2}/[0-9]{4} [0-9]{2}:[0-9]{2}:[0-9]{2}[+-][0-9]+(:[0-9]{2})?(\.[0-9]*)?$') THEN TRY_STRPTIME(TRIM(\"basic_model\".timestamp_field), '%d/%m/%Y %H:%M:%S%z') ELSE NULL END as TIMESTAMP)) as STRUCT(str_field VARCHAR, date_field DATE, timestamp_field TIMESTAMP))"),
+ ("another_model", AnotherModel, "TRY_CAST(STRUCT_PACK(\"unique_id\":= TRIM(\"another_model\".unique_id),\"basic_models\":= LIST_TRANSFORM(\"another_model\".basic_models, x -> STRUCT_PACK(\"str_field\":= TRIM(\"x\".str_field),\"date_field\":= TRY_CAST(CASE WHEN REGEXP_FULL_MATCH(TRIM(\"x\".date_field), '^[0-9]{4}\-[0-9]{2}\-[0-9]{2}$') THEN TRY_STRPTIME(TRIM(\"x\".date_field), '%Y-%m-%d') ELSE NULL END as DATE),\"timestamp_field\":= TRY_CAST(CASE WHEN REGEXP_FULL_MATCH(TRIM(\"x\".timestamp_field), '^[0-9]{2}/[0-9]{2}/[0-9]{4} [0-9]{2}:[0-9]{2}:[0-9]{2}[+-][0-9]+(:[0-9]{2})?(\.[0-9]*)?$') THEN TRY_STRPTIME(TRIM(\"x\".timestamp_field), '%d/%m/%Y %H:%M:%S%z') ELSE NULL END as TIMESTAMP)))) as STRUCT(unique_id BIGINT, basic_models STRUCT(str_field VARCHAR, date_field DATE, timestamp_field TIMESTAMP)[]))")])
def test_get_duckdb_cast_statement_from_annotation(field_name, field_type, cast_statement):
assert get_duckdb_cast_statement_from_annotation(field_name, field_type) == cast_statement
@@ -283,6 +283,25 @@ def test_relation_is_empty(temp_ddb_conn: DuckDBPyConnection):
@pytest.mark.parametrize("date_format, expected_regex", [("%Y-%m-%d", r"^[0-9]{4}\-[0-9]{2}\-[0-9]{2}$"),
("%Y-%m-%dT%H:%M:%S", r"^[0-9]{4}\-[0-9]{2}\-[0-9]{2}T[0-9]{2}:[0-9]{2}:[0-9]{2}$"),
("%m/%d/%Y", r"^[0-9]{2}/[0-9]{2}/[0-9]{4}$"),
- ("%Y/%m/%d %H:%M:%S%z", r"^[0-9]{4}/[0-9]{2}/[0-9]{2} [0-9]{2}:[0-9]{2}:[0-9]{2}(\+|\-)?[0-9]+(\.[0-9]*)?$")])
+ ("%Y/%m/%d %H:%M:%S%z", r"^[0-9]{4}/[0-9]{2}/[0-9]{2} [0-9]{2}:[0-9]{2}:[0-9]{2}[+-][0-9]+(:[0-9]{2})?(\.[0-9]*)?$")])
+
def test_datetime_format_to_regex(date_format, expected_regex):
- assert datetime_format_to_regex(date_format) == expected_regex
\ No newline at end of file
+ assert datetime_format_to_regex(date_format) == expected_regex
+
+
+@pytest.mark.parametrize("datetime_str, datetime_format, expected_value", [("2020-06-11", "%Y-%m-%d", datetime.date(2020,6,11)),
+ ("15/11/2018", "%d/%m/%Y", datetime.date(2018,11,15)),
+ ("27/14/2027", "%d/%m/%Y", None),
+ ("2014-05-27", "%d/%m/%y", None),
+ ("2020-03-15T13:24:54+0000", "%Y-%m-%dT%H:%M:%S%z", datetime.datetime(2020, 3, 15, 13, 24, 54)),
+ ("2020-03-15T13:24:54+00:00", "%Y-%m-%dT%H:%M:%S%z", datetime.datetime(2020, 3, 15, 13, 24, 54))])
+def test_datetime_casting(temp_ddb_conn: DuckDBPyConnection, datetime_str, datetime_format, expected_value):
+ _, con = temp_ddb_conn
+ if isinstance(expected_value, datetime.datetime):
+ type_ = formatteddatetime(date_format=datetime_format)
+ else:
+ type_ = conformatteddate(date_format=datetime_format)
+ cast_statement = get_duckdb_cast_statement_from_annotation("test", type_)
+ rel = con.sql(f"SELECT '{datetime_str}' as test")
+ check = rel.project(cast_statement).fetchone()[0]
+ assert check == expected_value
\ No newline at end of file
diff --git a/tests/test_core_engine/test_backends/test_implementations/test_spark/test_spark_helpers.py b/tests/test_core_engine/test_backends/test_implementations/test_spark/test_spark_helpers.py
index 34ecb6ca..9f7c9282 100644
--- a/tests/test_core_engine/test_backends/test_implementations/test_spark/test_spark_helpers.py
+++ b/tests/test_core_engine/test_backends/test_implementations/test_spark/test_spark_helpers.py
@@ -33,7 +33,7 @@
df_is_empty
)
from dve.core_engine.constants import RECORD_INDEX_COLUMN_NAME
-from dve.metadata_parser.domain_types import conformatteddate
+from dve.metadata_parser.domain_types import conformatteddate, formatteddatetime
from .....fixtures import spark # pylint: disable=unused-import
@@ -380,6 +380,22 @@ def test_df_is_empty(spark):
df5 = df4.union(spark.createDataFrame([(4, None, 8)], schema=_schema_w_ri))
assert not df_is_empty(df5)
+
+@pytest.mark.parametrize("datetime_str, datetime_format, expected_value", [("2020-06-11", "%Y-%m-%d", dt.date(2020,6,11)),
+ ("15/11/2018", "%d/%m/%Y", dt.date(2018,11,15)),
+ ("27/14/2027", "%d/%m/%Y", None),
+ ("2014-05-27", "%d/%m/%y", None),
+ ("2020-03-15T13:24:54+00:00", "%Y-%m-%dT%H:%M:%S%z", dt.datetime(2020, 3, 15, 13, 24, 54))])
+def test_datetime_casting(spark, datetime_str, datetime_format, expected_value):
+
+ if isinstance(expected_value, dt.datetime):
+ type_ = formatteddatetime(date_format=datetime_format)
+ else:
+ type_ = conformatteddate(date_format=datetime_format)
+ cast_statement = get_spark_cast_statement_from_annotation("test", type_)
+ df = spark.createDataFrame([(datetime_str,)], schema=StructType([StructField("test", StringType())]))
+ check = df.select(cast_statement.alias("test")).first().test
+ assert check == expected_value
From 665e1fad7ee23a31623bf0b20c001f843f3c715d Mon Sep 17 00:00:00 2001
From: stevenhsd <56357022+stevenhsd@users.noreply.github.com>
Date: Tue, 13 Oct 2026 15:30:17 +0100
Subject: [PATCH 37/37] refactor: rename of error code and messages related to
empty entities post business rules (#187)
---
docs/user_guidance/entity_relationships.md | 4 +--
.../core_engine/configuration/v1/__init__.py | 4 +--
.../core_engine/configuration/v1/hierarchy.py | 8 ++---
src/dve/pipeline/pipeline.py | 4 +--
tests/test_core_engine/test_hierarchy.py | 32 +++++++++----------
tests/testdata/flights/flights.dischema.json | 12 +++----
.../flights/flights_spark.dischema.json | 8 ++---
7 files changed, 36 insertions(+), 36 deletions(-)
diff --git a/docs/user_guidance/entity_relationships.md b/docs/user_guidance/entity_relationships.md
index bf0f6c19..68fe76d7 100644
--- a/docs/user_guidance/entity_relationships.md
+++ b/docs/user_guidance/entity_relationships.md
@@ -48,8 +48,8 @@ There is also the functionality to customise errors related to either missing pa
- `missing_parent_id_error_message`: the error message to display if a record is rejected as it has no valid parent record
- `no_valid_records_error_code`: the error code to display if parent records are removed due to no valid children in a mandatory field
- `no_valid_records_error_message`: the error message to display if parent records are removed due to no valid children in a mandatory field
-- `empty_entity_error_code`: the error code to display if a __mandatory__ entity has any records post filtering
-- `empty_entity_error_message`: the error message to display if a __mandatory__ entity has any records post filtering
+- `empty_entity_post_br_error_code`: the error code to display if a __mandatory__ entity has any records post filtering
+- `empty_entity_post_br_error_message`: the error message to display if a __mandatory__ entity has any records post filtering
!!! note
Specifying root entities is __optional__. Root entities will be inferred based on their absence.
diff --git a/src/dve/core_engine/configuration/v1/__init__.py b/src/dve/core_engine/configuration/v1/__init__.py
index 634cc112..dbccdab6 100644
--- a/src/dve/core_engine/configuration/v1/__init__.py
+++ b/src/dve/core_engine/configuration/v1/__init__.py
@@ -116,9 +116,9 @@ class _LinkageConfig(BaseModel):
"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
- empty_entity_error_code: ErrorCode = "EmptyEntity"
+ empty_entity_post_br_error_code: ErrorCode = "EmptyEntity"
"""The error code to emit if a mandatory entity has no valid remaining records"""
- empty_entity_error_message: ErrorMessage = "no valid records remaining"
+ empty_entity_post_br_error_message: ErrorMessage = "no valid records remaining"
"""The error message to emit if a mandatory entity has no valid remaining records"""
@model_validator(mode="after")
diff --git a/src/dve/core_engine/configuration/v1/hierarchy.py b/src/dve/core_engine/configuration/v1/hierarchy.py
index 35997eb1..57b64506 100644
--- a/src/dve/core_engine/configuration/v1/hierarchy.py
+++ b/src/dve/core_engine/configuration/v1/hierarchy.py
@@ -26,8 +26,8 @@ class HierarchyNode(BaseModel):
missing_parent_id_error_message: Optional[ErrorMessage] = (
"Records removed due to no valid parent record"
)
- empty_entity_error_code: ErrorCode = "EmptyEntity"
- empty_entity_error_message: ErrorMessage = "no valid records remaining"
+ empty_entity_post_br_error_code: ErrorCode = "EmptyEntity"
+ empty_entity_post_br_error_message: ErrorMessage = "no valid records remaining"
@model_validator(mode="after")
def validate_empty_error_details(self):
@@ -36,8 +36,8 @@ def validate_empty_error_details(self):
non mandatory nodes/entities
"""
if not self.mandatory:
- self.empty_entity_error_code = None
- self.empty_entity_error_message = None
+ self.empty_entity_post_br_error_code = None
+ self.empty_entity_post_br_error_message = None
return self
def get_descendents(self) -> list["HierarchyNode"]:
diff --git a/src/dve/pipeline/pipeline.py b/src/dve/pipeline/pipeline.py
index 2cf00553..1d29b1b2 100644
--- a/src/dve/pipeline/pipeline.py
+++ b/src/dve/pipeline/pipeline.py
@@ -581,10 +581,10 @@ def check_mandatory_entities_have_records(
entity=entity_name,
record=None,
error_location=entity_name,
- error_message=node.empty_entity_error_message,
+ error_message=node.empty_entity_post_br_error_message,
failure_type="submission",
error_type="submission",
- error_code=node.empty_entity_error_code,
+ error_code=node.empty_entity_post_br_error_code,
category="Empty entity",
)
)
diff --git a/tests/test_core_engine/test_hierarchy.py b/tests/test_core_engine/test_hierarchy.py
index bbce74f0..43a0a8f7 100644
--- a/tests/test_core_engine/test_hierarchy.py
+++ b/tests/test_core_engine/test_hierarchy.py
@@ -269,8 +269,8 @@ def test_linkage_config_load():
"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,
- "empty_entity_error_code": null,
- "empty_entity_error_message": null,
+ "empty_entity_post_br_error_code": null,
+ "empty_entity_post_br_error_message": null,
"children": {
"ds_003": {
"parent_entity": "ds_001",
@@ -282,8 +282,8 @@ def test_linkage_config_load():
"no_valid_records_error_message": "parent record removed as no valid child records",
"missing_parent_id_error_code": "DS003NoParent",
"missing_parent_id_error_message": "record removed as no parent",
- "empty_entity_error_code": null,
- "empty_entity_error_message": null,
+ "empty_entity_post_br_error_code": null,
+ "empty_entity_post_br_error_message": null,
"children": {}
},
"ds_101": {
@@ -296,8 +296,8 @@ def test_linkage_config_load():
"no_valid_records_error_message": "{{ ds_001_id }} removed as no valid ds_101 records",
"missing_parent_id_error_code": "DS101NoParent",
"missing_parent_id_error_message": "record removed as no parent",
- "empty_entity_error_code": null,
- "empty_entity_error_message": null,
+ "empty_entity_post_br_error_code": null,
+ "empty_entity_post_br_error_message": null,
"children": {
"ds_201": {
"parent_entity": "ds_101",
@@ -309,8 +309,8 @@ def test_linkage_config_load():
"no_valid_records_error_message": "parent record removed as no valid child records",
"missing_parent_id_error_code": "DS201NoParent",
"missing_parent_id_error_message": "record removed as no parent",
- "empty_entity_error_code": null,
- "empty_entity_error_message": null,
+ "empty_entity_post_br_error_code": null,
+ "empty_entity_post_br_error_message": null,
"children": {
"ds_202": {
"parent_entity": "ds_201",
@@ -322,8 +322,8 @@ def test_linkage_config_load():
"no_valid_records_error_message": "parent record removed as no valid child records",
"missing_parent_id_error_code": "MissingParentRecord",
"missing_parent_id_error_message": "Records removed due to no valid parent record",
- "empty_entity_error_code": "EmptyEntity",
- "empty_entity_error_message": "no valid records remaining",
+ "empty_entity_post_br_error_code": "EmptyEntity",
+ "empty_entity_post_br_error_message": "no valid records remaining",
"children": {}
}
}
@@ -351,8 +351,8 @@ def test_linkage_config_load():
"no_valid_records_error_message": "{{ ds_001_id }} removed as no valid ds_101 records",
"missing_parent_id_error_code": "DS101NoParent",
"missing_parent_id_error_message": "record removed as no parent",
- "empty_entity_error_code": null,
- "empty_entity_error_message": null,
+ "empty_entity_post_br_error_code": null,
+ "empty_entity_post_br_error_message": null,
"children": {
"ds_201": {
"parent_entity": "ds_101",
@@ -364,8 +364,8 @@ def test_linkage_config_load():
"no_valid_records_error_message": "parent record removed as no valid child records",
"missing_parent_id_error_code": "DS201NoParent",
"missing_parent_id_error_message": "record removed as no parent",
- "empty_entity_error_code": null,
- "empty_entity_error_message": null,
+ "empty_entity_post_br_error_code": null,
+ "empty_entity_post_br_error_message": null,
"children": {
"ds_202": {
"parent_entity": "ds_201",
@@ -377,8 +377,8 @@ def test_linkage_config_load():
"no_valid_records_error_message": "parent record removed as no valid child records",
"missing_parent_id_error_code": "MissingParentRecord",
"missing_parent_id_error_message": "Records removed due to no valid parent record",
- "empty_entity_error_code": "EmptyEntity",
- "empty_entity_error_message": "no valid records remaining",
+ "empty_entity_post_br_error_code": "EmptyEntity",
+ "empty_entity_post_br_error_message": "no valid records remaining",
"children": {}
}
}
diff --git a/tests/testdata/flights/flights.dischema.json b/tests/testdata/flights/flights.dischema.json
index a0342cc0..a4ef67e1 100644
--- a/tests/testdata/flights/flights.dischema.json
+++ b/tests/testdata/flights/flights.dischema.json
@@ -161,8 +161,8 @@
"country": {
"is_root_entity": true,
"mandatory": true,
- "empty_entity_error_code": "NoValidCountries",
- "empty_entity_error_message": "File Rejected - There are no valid country records"
+ "empty_entity_post_br_error_code": "NoValidCountries",
+ "empty_entity_post_br_error_message": "File Rejected - There are no valid country records"
},
"airport": {
"parent_entity": "country",
@@ -174,8 +174,8 @@
"missing_parent_id_error_message": "Record rejected - No valid country id found for airport",
"no_valid_records_error_code": "CountryHasNoAirport",
"no_valid_records_error_message": "Group rejected - Unable to find any valid airports",
- "empty_entity_error_code": "NoValidAirports",
- "empty_entity_error_message": "File Rejected - There are no valid airport records"
+ "empty_entity_post_br_error_code": "NoValidAirports",
+ "empty_entity_post_br_error_message": "File Rejected - There are no valid airport records"
},
"staff": {
"parent_entity": "airport",
@@ -187,8 +187,8 @@
"missing_parent_id_error_message": "Record rejected - No valid airport id found for staff. Airport ID = {{ airport_id }}, Staff ID = {{ staff_id }}",
"no_valid_records_error_code": "AirportHasNoStaff",
"no_valid_records_error_message": "Group rejected - Airport has no valid staff. Airport ID = {{ airport_id }}",
- "empty_entity_error_code": "NoValidStaff",
- "empty_entity_error_message": "File Rejected - There are no valid staff records"
+ "empty_entity_post_br_error_code": "NoValidStaff",
+ "empty_entity_post_br_error_message": "File Rejected - There are no valid staff records"
},
"flights": {
"parent_entity": "airport",
diff --git a/tests/testdata/flights/flights_spark.dischema.json b/tests/testdata/flights/flights_spark.dischema.json
index b318f823..d94e07d3 100644
--- a/tests/testdata/flights/flights_spark.dischema.json
+++ b/tests/testdata/flights/flights_spark.dischema.json
@@ -174,8 +174,8 @@
"missing_parent_id_error_message": "Record rejected - No valid country id found for airport",
"no_valid_records_error_code": "CountryHasNoAirport",
"no_valid_records_error_message": "Group rejected - Unable to find any valid airports",
- "empty_entity_error_code": "NoValidAirports",
- "empty_entity_error_message": "File Rejected - There are no valid airport records"
+ "empty_entity_post_br_error_code": "NoValidAirports",
+ "empty_entity_post_br_error_message": "File Rejected - There are no valid airport records"
},
"staff": {
"parent_entity": "airport",
@@ -187,8 +187,8 @@
"missing_parent_id_error_message": "Record rejected - No valid airport id found for staff. Airport ID = {{ airport_id }}, Staff ID = {{ staff_id }}",
"no_valid_records_error_code": "AirportHasNoStaff",
"no_valid_records_error_message": "Group rejected - Airport has no valid staff. Airport ID = {{ airport_id }}",
- "empty_entity_error_code": "NoValidStaff",
- "empty_entity_error_message": "File Rejected - There are no valid staff records"
+ "empty_entity_post_br_error_code": "NoValidStaff",
+ "empty_entity_post_br_error_message": "File Rejected - There are no valid staff records"
},
"flights": {
"parent_entity": "airport",