From daa1938f1955960775edee088f7b8f0e6500e64f Mon Sep 17 00:00:00 2001
From: Tom Owers <owerstom@gmail.com>
Date: Wed, 30 Oct 2024 16:52:02 +0000
Subject: [PATCH] Fixes for DLT

---
 .../pipelines/rest_source/__init__.py         | 19 -------------------
 .../temporal/data_modeling/run_workflow.py    |  6 +++---
 .../warehouse/data_load/validate_schema.py    |  2 +-
 .../models/external_table_definitions.py      |  4 ++--
 4 files changed, 6 insertions(+), 25 deletions(-)

diff --git a/posthog/temporal/data_imports/pipelines/rest_source/__init__.py b/posthog/temporal/data_imports/pipelines/rest_source/__init__.py
index 5dceafd1d2aec..4fd019ce76753 100644
--- a/posthog/temporal/data_imports/pipelines/rest_source/__init__.py
+++ b/posthog/temporal/data_imports/pipelines/rest_source/__init__.py
@@ -372,22 +372,3 @@ def identity_func(x: Any) -> Any:
     if incremental_param.end:
         params[incremental_param.end] = transform(incremental_object.end_value)
     return params
-
-
-# XXX: This is a workaround pass test_dlt_init.py
-# since the source uses dlt.source as a function
-def _register_source(source_func: Callable[..., DltSource]) -> None:
-    import inspect
-    from dlt.common.configuration import get_fun_spec
-    from dlt.common.source import _SOURCES, SourceInfo
-
-    spec = get_fun_spec(source_func)
-    func_module = inspect.getmodule(source_func)
-    _SOURCES[source_func.__name__] = SourceInfo(
-        SPEC=spec,
-        f=source_func,
-        module=func_module,
-    )
-
-
-_register_source(rest_api_source)
diff --git a/posthog/temporal/data_modeling/run_workflow.py b/posthog/temporal/data_modeling/run_workflow.py
index 8629bb29cdf76..e5c0f43acad36 100644
--- a/posthog/temporal/data_modeling/run_workflow.py
+++ b/posthog/temporal/data_modeling/run_workflow.py
@@ -57,9 +57,9 @@
     "Int32": "bigint",
     "Int64": "bigint",
     "Tuple": "bigint",
-    "Array": "complex",
-    "Map": "complex",
-    "Tuple": "complex",
+    "Array": "json",
+    "Map": "json",
+    "Tuple": "json",
     "Bool": "bool",
     "Decimal": "decimal",
 }
diff --git a/posthog/warehouse/data_load/validate_schema.py b/posthog/warehouse/data_load/validate_schema.py
index c4c483a5f1fd9..17ed1c53190c6 100644
--- a/posthog/warehouse/data_load/validate_schema.py
+++ b/posthog/warehouse/data_load/validate_schema.py
@@ -50,7 +50,7 @@ def dlt_to_hogql_type(dlt_type: TDataType | None) -> str:
         hogql_type = IntegerDatabaseField
     elif dlt_type == "binary":
         raise Exception("DLT type 'binary' is not a supported column type")
-    elif dlt_type == "complex":
+    elif dlt_type == "json":
         hogql_type = StringJSONDatabaseField
     elif dlt_type == "decimal":
         hogql_type = IntegerDatabaseField
diff --git a/posthog/warehouse/models/external_table_definitions.py b/posthog/warehouse/models/external_table_definitions.py
index 9d76d4d993969..00704ec6c3994 100644
--- a/posthog/warehouse/models/external_table_definitions.py
+++ b/posthog/warehouse/models/external_table_definitions.py
@@ -707,8 +707,8 @@
     IntegerDatabaseField: "bigint",
     BooleanDatabaseField: "bool",
     DateTimeDatabaseField: "timestamp",
-    StringJSONDatabaseField: "complex",
-    StringArrayDatabaseField: "complex",
+    StringJSONDatabaseField: "json",
+    StringArrayDatabaseField: "json",
     FloatDatabaseField: "double",
     DateDatabaseField: "date",
 }