fix: refactor the join_header method in duckdb to deal with out of bound date values

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

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