Fix: Cast to TEXT in state sync instead of using custom text types (#1681)

Author: izeigerman

sqlmesh/core/state_sync/engine_adapter.py

@@ -57,7 +57,6 @@
 from sqlmesh.utils import major_minor, random_id
 from sqlmesh.utils.date import TimeLike, now_timestamp, time_like_to_str
 from sqlmesh.utils.errors import SQLMeshError
-from sqlmesh.utils.migration import blob_text_type, index_text_type
 
 logger = logging.getLogger(__name__)
 
@@ -104,20 +103,17 @@ def __init__(
         self.plan_dags_table = exp.table_("_plan_dags", db=self.schema)
         self.versions_table = exp.table_("_versions", db=self.schema)
 
-        index_type = index_text_type(self.engine_adapter.dialect)
-        blob_type = blob_text_type(self.engine_adapter.dialect)
-
         self._snapshot_columns_to_types = {
-            "name": exp.DataType.build(index_type),
-            "identifier": exp.DataType.build(index_type),
-            "version": exp.DataType.build(index_type),
-            "snapshot": exp.DataType.build(blob_type),
-            "kind_name": exp.DataType.build(index_type),
+            "name": exp.DataType.build("text"),
+            "identifier": exp.DataType.build("text"),
+            "version": exp.DataType.build("text"),
+            "snapshot": exp.DataType.build("text"),
+            "kind_name": exp.DataType.build("text"),
         }
 
         self._environment_columns_to_types = {
-            "name": exp.DataType.build(index_type),
-            "snapshots": exp.DataType.build(blob_type),
+            "name": exp.DataType.build("text"),
+            "snapshots": exp.DataType.build("text"),
             "start_at": exp.DataType.build("text"),
             "end_at": exp.DataType.build("text"),
             "plan_id": exp.DataType.build("text"),
@@ -129,17 +125,17 @@ def __init__(
         }
 
         self._seed_columns_to_types = {
-            "name": exp.DataType.build(index_type),
-            "identifier": exp.DataType.build(index_type),
-            "content": exp.DataType.build(blob_type),
+            "name": exp.DataType.build("text"),
+            "identifier": exp.DataType.build("text"),
+            "content": exp.DataType.build("text"),
         }
 
         self._interval_columns_to_types = {
-            "id": exp.DataType.build(index_type),
+            "id": exp.DataType.build("text"),
             "created_ts": exp.DataType.build("bigint"),
-            "name": exp.DataType.build(index_type),
-            "identifier": exp.DataType.build(index_type),
-            "version": exp.DataType.build(index_type),
+            "name": exp.DataType.build("text"),
+            "identifier": exp.DataType.build("text"),
+            "version": exp.DataType.build("text"),
             "start_ts": exp.DataType.build("bigint"),
             "end_ts": exp.DataType.build("bigint"),
             "is_dev": exp.DataType.build("boolean"),

sqlmesh/schedulers/airflow/plan.py

@@ -15,7 +15,6 @@
 from sqlmesh.schedulers.airflow import common
 from sqlmesh.utils.date import now, now_timestamp
 from sqlmesh.utils.errors import SQLMeshError
-from sqlmesh.utils.migration import blob_text_type, index_text_type
 
 
 class PlanDagState:
@@ -24,13 +23,10 @@ def __init__(self, engine_adapter: EngineAdapter, plan_dags_table: exp.Table):
 
         self._plan_dags_table = plan_dags_table
 
-        index_type = index_text_type(self.engine_adapter.dialect)
-        blob_type = blob_text_type(self.engine_adapter.dialect)
-
         self._plan_dag_columns_to_types = {
-            "request_id": exp.DataType.build(index_type),
-            "dag_id": exp.DataType.build(index_type),
-            "dag_spec": exp.DataType.build(blob_type),
+            "request_id": exp.DataType.build("text"),
+            "dag_id": exp.DataType.build("text"),
+            "dag_spec": exp.DataType.build("text"),
         }
 
     @classmethod