test: add example failing test for recursive datatype DDLs

Author: NickCrews

duckdb_engine/tests/test_datatypes.py

@@ -1,4 +1,5 @@
 import decimal
+import enum
 import json
 import warnings
 from typing import Any, Dict, Type
@@ -9,6 +10,7 @@
 from snapshottest.module import SnapshotTest
 from sqlalchemy import (
     Column,
+    Enum,
     Integer,
     Interval,
     MetaData,
@@ -28,7 +30,7 @@
 from sqlalchemy.types import FLOAT, JSON
 
 from .._supports import duckdb_version, has_uhugeint_support
-from ..datatypes import Map, Struct, types
+from ..datatypes import Map, Struct, Union, types
 
 
 @mark.parametrize("coltype", types)
@@ -234,6 +236,47 @@ class Entry(base):
     assert result.outer == outer
 
 
+def test_double_nested_type_ddl(engine: Engine, session: Session) -> None:
+    """If we create a table with a nested type, then all the children types,
+    (such as ENUMS) need to have already been created with an eg CREATE TYPE ...
+    DDL statement."""
+    importorskip("duckdb", "0.5.0")  # nested types require at least duckdb 0.5.0
+    base = declarative_base()
+
+    class Severity(enum.Enum):
+        LOW = "L"
+        MEDIUM = "M"
+        HIGH = "H"
+
+    class Entry(base):
+        __tablename__ = "test_struct"
+
+        id = Column(Integer, primary_key=True, default=0)
+        struct = Column(Struct({"severity": Enum(Severity)}))
+        map = Column(Map(String, Enum(Severity)))
+        union = Column(Union({"age": Integer, "severity": Enum(Severity)}))
+
+    base.metadata.create_all(bind=engine)
+
+    struct = {"struct": {"severity": "L"}}
+    session.add(Entry(struct=struct))  # type: ignore[call-arg]
+    session.commit()
+    result = session.query(Entry).one()
+    assert result.struct == struct
+
+    map = {"one": "L", "two": "M"}
+    session.add(Entry(map=map))  # type: ignore[call-arg]
+    session.commit()
+    result = session.query(Entry).one()
+    assert result.map == map
+
+    union = {"age": 42}
+    session.add(Entry(union=union))  # type: ignore[call-arg]
+    session.commit()
+    result = session.query(Entry).one()
+    assert result.union == union
+
+
 def test_interval(engine: Engine, snapshot: SnapshotTest) -> None:
     test_table = Table("test_table", MetaData(), Column("duration", Interval))