Repository: TobikoData/sqlmesh
Pull request #2308 (2024-03-20 00:53:51)

Fix: Create a temporary view to source the model schema instead of parsing the Redshift plan
This simplifies the implementation and eliminates the need to parser the planner's output to determine column types.


Patch:
diff --git a/Makefile b/Makefile
index 069c2d84..bc921bcc 100644
--- a/Makefile
+++ b/Makefile
@@ -179,4 +179,4 @@ spark-pyspark-test:
 	pytest -n auto -m "spark_pyspark"
 
 trino-test:
-	pytest -n auto -m "trino or trino_iceberg"
\ No newline at end of file
+	pytest -n auto -m "trino or trino_iceberg or trino_delta"
\ No newline at end of file
diff --git a/docs/integrations/engines/trino.md b/docs/integrations/engines/trino.md
index 61ec1ae5..db1c16ae 100644
--- a/docs/integrations/engines/trino.md
+++ b/docs/integrations/engines/trino.md
@@ -17,7 +17,7 @@ pip install "trino[external-authentication-token-cache]"
 
 ### Trino Connector Support
 
-The trino engine adapter has been tested against the [Hive Connector](https://trino.io/docs/current/connector/hive.html) and the [Iceberg Connector](https://trino.io/docs/current/connector/iceberg.html).
+The trino engine adapter has been tested against the [Hive Connector](https://trino.io/docs/current/connector/hive.html), [Iceberg Connector](https://trino.io/docs/current/connector/iceberg.html), and [Delta Lake Connector](https://trino.io/docs/current/connector/delta-lake.html).
 
 Please let us know on [Slack](https://tobikodata.com/slack) if you are wanting to use another connector or have tried another connector.
 
@@ -49,6 +49,17 @@ iceberg.catalog.type=hive_metastore
 
 The `jdbc`, `rest` and `nessie` catalogs do not support views and are thus incompatible with SQLMesh.
 
+#### Delta Lake Connector Configuration
+
+The Trino adapter Delta Lake connector has only been tested with the Hive metastore catalog type.
+
+The [properties file](https://trino.io/docs/current/connector/delta-lake.html#general-configuration) must include the Hive metastore URI and catalog name in addition to any other [general properties](https://trino.io/docs/current/object-storage/metastores.html#general-metastore-properties).
+
+``` properties linenums="1"
+hive.metastore.uri=thrift://example.net:9083
+delta.hive-catalog-name=datalake_delta # example catalog name, can be any valid string
+```
+
 ### Connection options
 
 | Option               | Description                                                                                                                                                               |  Type  | Required |
diff --git a/sqlmesh/core/engine_adapter/base_postgres.py b/sqlmesh/core/engine_adapter/base_postgres.py
index 63ef895a..049f84ee 100644
--- a/sqlmesh/core/engine_adapter/base_postgres.py
+++ b/sqlmesh/core/engine_adapter/base_postgres.py
@@ -35,10 +35,11 @@ class BasePostgresEngineAdapter(EngineAdapter):
         sql = (
             exp.select("column_name", "data_type")
             .from_(self.COLUMNS_TABLE)
-            .where(
-                f"table_name = '{table.alias_or_name}' AND table_schema = '{table.args['db'].name}'"
-            )
+            .where(f"table_name = '{table.alias_or_name}'")
         )
+        if table.args.get("db"):
+            sql = sql.where(f"table_schema = '{table.args['db'].name}'")
+
         self.execute(sql)
         resp = self.cursor.fetchall()
         if not resp:
diff --git a/sqlmesh/core/engine_adapter/mixins.py b/sqlmesh/core/engine_adapter/mixins.py
index c425e1c1..2d8ecd77 100644
--- a/sqlmesh/core/engine_adapter/mixins.py
+++ b/sqlmesh/core/engine_adapter/mixins.py
@@ -224,8 +224,8 @@ class HiveMetastoreTablePropertiesMixin(EngineAdapter):
     def _truncate_comment(
         self, comment: str, length: t.Optional[int], escape_backslash: bool = False
     ) -> str:
-        # iceberg does not have a comment length limit
-        if self.current_catalog_type == "iceberg":
+        # iceberg and delta do not have a comment length limit
+        if self.current_catalog_type in ("iceberg", "delta"):
             return comment
         return super()._truncate_comment(comment, length, escape_backslash)
 
diff --git a/sqlmesh/core/engine_adapter/redshift.py b/sqlmesh/core/engine_adapter/redshift.py
index 9ddaf8a6..639d08d9 100644
--- a/sqlmesh/core/engine_adapter/redshift.py
+++ b/sqlmesh/core/engine_adapter/redshift.py
@@ -4,7 +4,6 @@ import typing as t
 
 import pandas as pd
 from sqlglot import exp
-from sqlglot.errors import ErrorLevel
 
 from sqlmesh.core.dialect import to_schema
 from sqlmesh.core.engine_adapter.base_postgres import BasePostgresEngineAdapter
@@ -20,7 +19,7 @@ from sqlmesh.core.engine_adapter.shared import (
     SourceQuery,
     set_catalog,
 )
-from sqlmesh.utils.errors import SQLMeshError
+from sqlmesh.utils import random_id
 
 if t.TYPE_CHECKING:
     from sqlmesh.core._typing import SchemaName, TableName
@@ -121,68 +120,36 @@ class RedshiftEngineAdapter(
             and statement.expression.args.get("limit") is not None
             and statement.expression.args["limit"].expression.this == "0"
         ):
+            assert not isinstance(table_name_or_schema, exp.Schema)
             # redshift has a bug where CTAS statements have non determistic types. if a limit
             # is applied to a ctas statement, VARCHAR types default to 1 in some instances.
             # this checks the explain plain from redshift and tries to detect when these optimizer
             # bugs occur and force a cast
-            explain_statement = statement.copy()
-            for select in explain_statement.find_all(exp.Select):
-                if select.args.get("from"):
-                    select.set("limit", None)
-                    select.set("where", None)
+            select_statement = statement.expression.copy()
+            for select_or_union in select_statement.find_all(exp.Select, exp.Union):
+                select_or_union.set("limit", None)
+                select_or_union.set("where", None)
+
+            temp_view_name = exp.table_(f"#sqlmesh__{random_id()}")
+            self.create_view(
+                temp_view_name, select_statement, replace=False, no_schema_binding=False
+            )
+            columns_to_types_from_view = self.columns(temp_view_name)
 
-            explain_statement_sql = explain_statement.sql(
-                dialect=self.dialect, identify=True, unsupported_level=ErrorLevel.IGNORE, copy=False
+            schema = self._build_schema_exp(
+                exp.to_table(table_name_or_schema),
+                columns_to_types_from_view,
             )
-            plan = parse_plan(
-                "\n".join(r[0] for r in self.fetchall(f"EXPLAIN VERBOSE {explain_statement_sql}"))
+            statement = super()._build_create_table_exp(
+                schema,
+                None,
+                exists=exists,
+                replace=replace,
+                columns_to_types=columns_to_types_from_view,
+                table_description=table_description,
+                **kwargs,
             )
 
-            if plan:
-                select = exp.Select().from_(statement.expression.subquery("_subquery"))
-                statement.expression.replace(select)
-
-                for target in plan["targetlist"]:  # type: ignore
-                    if target["name"] == "TARGETENTRY":
-                        resdom = target["resdom"]
-                        resname = resdom["resname"]
-                        if resname == "<>":
-                            # A synthetic column added by Redshift to compute a window function.
-                            continue
-                        if resname == "? column ?":
-                            table_name_str = (
-                                table_name_or_schema
-                                if isinstance(table_name_or_schema, str)
-                                else table_name_or_schema.sql(dialect=self.dialect)
-                            )
-                            raise SQLMeshError(f"Missing column name for table '{table_name_str}'")
-                        # https://github.com/postgres/postgres/blob/master/src/include/catalog/pg_type.dat
-                        restype = resdom["restype"]
-                        data_type: t.Optional[str] = None
-                        if restype == "1043":
-                            size = (
-                                int(resdom["restypmod"]) - 4
-                                if resdom["restypmod"] != "- 1"
-                                else "MAX"
-                            )
-                            # Cast NULL instead of the original projection to trick the planner into assigning a
-                            # correct type to the column.
-                            data_type = f"VARCHAR({size})"
-                        else:
-                            data_type = REDSHIFT_PLAN_TYPE_MAPPINGS.get(restype)
-
-                        if data_type:
-                            select.select(
-                                exp.cast(
-                                    exp.null(),
-                                    data_type,
-                                    dialect=self.dialect,
-                                ).as_(resname),
-                                copy=False,
-                            )
-                        else:
-                            select.select(resname, copy=False)
-
         return statement
 
     def create_view(
@@ -209,7 +176,7 @@ class RedshiftEngineAdapter(
             materialized,
             table_description=table_description,
             column_descriptions=column_descriptions,
-            no_schema_binding=True,
+            no_schema_binding=create_kwargs.pop("no_schema_binding", True),
             **create_kwargs,
         )
 
@@ -309,90 +276,3 @@ class RedshiftEngineAdapter(
             )
             for row in df.itertuples()
         ]
-
-
-def parse_plan(plan: str) -> t.Optional[t.Dict]:
-    """Parse the output of a redshift explain verbose query plan into a Python dict."""
-    from sqlglot import Tokenizer, TokenType
-    from sqlglot.tokens import Token
-
-    tokens = Tokenizer().tokenize(plan)
-    i = 0
-    terminal_tokens = {TokenType.L_PAREN, TokenType.R_PAREN, TokenType.R_BRACE, TokenType.COLON}
-
-    def curr() -> t.Optional[TokenType]:
-        return tokens[i].token_type if i < len(tokens) else None
-
-    def advance() -> Token:
-        nonlocal i
-        i += 1
-        return tokens[i - 1]
-
-    def match(token_type: TokenType, raise_unmatched: bool = False) -> t.Optional[Token]:
-        if curr() == token_type:
-            return advance()
-        if raise_unmatched:
-            raise Exception(f"Expected {token_type}")
-        return None
-
-    def parse_value() -> t.Any:
-        if match(TokenType.L_PAREN):
-            values = []
-            while not match(TokenType.R_PAREN):
-                values.append(parse_value())
-            return values
-
-        nested = parse_nested()
-
-        if nested:
-            return nested
-
-        value = []
-
-        while not curr() in terminal_tokens:
-            value.append(advance().text)
-
-        return " ".join(value)
-
-    def parse_nested() -> t.Optional[t.Dict]:
-        if not match(TokenType.L_BRACE):
-            return None
-        query_plan = {}
-        query_plan["name"] = advance().text
-
-        while match(TokenType.COLON):
-            key = advance().text
-
-            while match(TokenType.DOT):
-                key += f".{advance().text}"
-
-            query_plan[key] = parse_value()
-
-        match(TokenType.R_BRACE, True)
-        return query_plan
-
-    while curr():
-        nested = parse_nested()
-
-        if nested and nested.get("name") in ("RESULT", "SEQSCAN"):
-            return nested
-        advance()
-    return None
-
-
-# https://github.com/postgres/postgres/blob/master/src/include/catalog/pg_type.dat
-REDSHIFT_PLAN_TYPE_MAPPINGS = {
-    "16": "BOOL",
-    "18": "CHAR",
-    "21": "SMALLINT",
-    "23": "INT",
-    "20": "BIGINT",
-    "1700": "NUMERIC",
-    "700": "FLOAT",
-    "701": "DOUBLE",
-    "1114": "TIMESTAMP",
-    "1184": "TIMESTAMPTZ",
-    "1083": "TIME",
-    "1266": "TIMETZ",
-    "1082": "DATE",
-}
diff --git a/sqlmesh/core/engine_adapter/trino.py b/sqlmesh/core/engine_adapter/trino.py
index 62808c1b..8834464e 100644
--- a/sqlmesh/core/engine_adapter/trino.py
+++ b/sqlmesh/core/engine_adapter/trino.py
@@ -24,12 +24,13 @@ from sqlmesh.core.engine_adapter.shared import (
     SourceQuery,
     set_catalog,
 )
+from sqlmesh.utils.date import TimeLike
 
 if t.TYPE_CHECKING:
     from trino.dbapi import Connection as TrinoConnection
 
     from sqlmesh.core._typing import SchemaName, TableName
-    from sqlmesh.core.engine_adapter._typing import DF
+    from sqlmesh.core.engine_adapter._typing import DF, QueryOrDF
 
 
 @set_catalog()
@@ -183,3 +184,80 @@ class TrinoEngineAdapter(
                 df[column] = pd.to_datetime(df[column]).map(lambda x: x.isoformat(" "))  # type: ignore
 
         return super()._df_to_source_queries(df, columns_to_types, batch_size, target_table)
+
+    def _build_schema_exp(
+        self,
+        table: exp.Table,
+        columns_to_types: t.Dict[str, exp.DataType],
+        column_descriptions: t.Optional[t.Dict[str, str]] = None,
+        expressions: t.Optional[t.List[exp.PrimaryKey]] = None,
+        is_view: bool = False,
+    ) -> exp.Schema:
+        if self.current_catalog_type == "delta_lake":
+            columns_to_types = self._to_delta_ts(columns_to_types)
+
+        return super()._build_schema_exp(
+            table, columns_to_types, column_descriptions, expressions, is_view
+        )
+
+    def _scd_type_2(
+        self,
+        target_table: TableName,
+        source_table: QueryOrDF,
+        unique_key: t.Sequence[exp.Expression],
+        valid_from_name: str,
+        valid_to_name: str,
+        execution_time: TimeLike,
+        invalidate_hard_deletes: bool = True,
+        updated_at_name: t.Optional[str] = None,
+        check_columns: t.Optional[t.Union[exp.Star, t.Sequence[exp.Column]]] = None,
+        updated_at_as_valid_from: bool = False,
+        execution_time_as_valid_from: bool = False,
+        columns_to_types: t.Optional[t.Dict[str, exp.DataType]] = None,
+        table_description: t.Optional[str] = None,
+        column_descriptions: t.Optional[t.Dict[str, str]] = None,
+    ) -> None:
+        if columns_to_types and self.current_catalog_type == "delta_lake":
+            columns_to_types = self._to_delta_ts(columns_to_types)
+
+        return super()._scd_type_2(
+            target_table,
+            source_table,
+            unique_key,
+            valid_from_name,
+            valid_to_name,
+            execution_time,
+            invalidate_hard_deletes,
+            updated_at_name,
+            check_columns,
+            updated_at_as_valid_from,
+            execution_time_as_valid_from,
+            columns_to_types,
+            table_description,
+            column_descriptions,
+        )
+
+    # delta_lake only supports two timestamp data types. This method converts other
+    # timestamp types to those two for use in DDL statements. Trino/delta automatically
+    # converts the data values to the correct type on write, so we only need to handle
+    # the column types in DDL.
+    # - `timestamp(6)` for non-timezone-aware
+    # - `timestamp(3) with time zone` for timezone-aware
+    # https://trino.io/docs/current/connector/delta-lake.html#delta-lake-to-trino-type-mapping
+    def _to_delta_ts(
+        self, columns_to_types: t.Dict[str, exp.DataType]
+    ) -> t.Dict[str, exp.DataType]:
+        ts6 = exp.DataType.build("timestamp(6)")
+        ts3_tz = exp.DataType.build("timestamp(3) with time zone")
+
+        delta_columns_to_types = {
+            k: ts6 if v.is_type(exp.DataType.Type.TIMESTAMP) else v
+            for k, v in columns_to_types.items()
+        }
+
+        delta_columns_to_types = {
+            k: ts3_tz if v.is_type(exp.DataType.Type.TIMESTAMPTZ) else v
+            for k, v in delta_columns_to_types.items()
+        }
+
+        return delta_columns_to_types
diff --git a/sqlmesh/core/snapshot/evaluator.py b/sqlmesh/core/snapshot/evaluator.py
index dc185a3b..98e47840 100644
--- a/sqlmesh/core/snapshot/evaluator.py
+++ b/sqlmesh/core/snapshot/evaluator.py
@@ -1094,10 +1094,12 @@ class MaterializableStrategy(PromotableStrategy):
                 column_descriptions=model.column_descriptions if is_table_deployable else None,
             )
 
-            # only sql models have queries that can be tested
-            # additionally, we always create temp tables and sometimes
-            # we additionally created prod tables, so we only need to test one.
-            if model.is_sql and not is_table_deployable:
+            # Only sql models have queries that can be tested.
+            # Additionally, we always create temp tables and sometimes we additionally created prod tables,
+            # we need to make sure that we only dry run once.
+            # We also need to make sure that we don't dry run on Redshift because its planner / optimizer sometimes
+            # breaks on our CTAS queries due to us relying on the WHERE FALSE LIMIT 0 combo.
+            if model.is_sql and not is_table_deployable and self.adapter.dialect != "redshift":
                 logger.info("Dry running model '%s'", model.name)
                 self.adapter.fetchall(ctas_query)
         else:


Test patch:
diff --git a/pytest.ini b/pytest.ini
index 38a6c5da..915c8004 100644
--- a/pytest.ini
+++ b/pytest.ini
@@ -31,6 +31,7 @@ markers =
     spark: test for Spark
     trino: test for Trino (Hive connector)
     trino_iceberg: test for Trino (Iceberg connector)
+    trino_delta: test for Trino (Delta connector)
 addopts = -n 0 --dist=loadgroup
 
 # Set this to True to enable logging during tests
diff --git a/tests/core/engine_adapter/config.yaml b/tests/core/engine_adapter/config.yaml
index d90ac8bd..fd29007e 100644
--- a/tests/core/engine_adapter/config.yaml
+++ b/tests/core/engine_adapter/config.yaml
@@ -30,6 +30,18 @@ gateways:
       register_comments: true
     state_connection:
       type: duckdb
+  inttest_trino_delta:
+    connection:
+      type: trino
+      host: localhost
+      port: 8080
+      user: admin
+      catalog: datalake_delta
+      http_scheme: http
+      retries: 20
+      register_comments: true
+    state_connection:
+      type: duckdb
   inttest_spark:
     connection:
       type: spark
diff --git a/tests/core/engine_adapter/docker-compose.yaml b/tests/core/engine_adapter/docker-compose.yaml
index 1888fc9e..117a7b1e 100644
--- a/tests/core/engine_adapter/docker-compose.yaml
+++ b/tests/core/engine_adapter/docker-compose.yaml
@@ -60,6 +60,17 @@ services:
     volumes:
       - ./trino/initdb.sql:/docker-entrypoint-initdb.d/initdb.sql
 
+  # A second metastore DB is needed because testing all of hive/iceberg/delta
+  # creates too many connections for a single postgres DB.
+  trino_iceberg_delta_metastore_db:
+    image: postgres
+    hostname: trino_iceberg_delta_metastore_db
+    environment:
+      POSTGRES_USER: hive
+      POSTGRES_PASSWORD: hive
+    volumes:
+      - ./trino/initdb.sql:/docker-entrypoint-initdb.d/initdb.sql
+
   trino-datalake-hive-metastore:
     hostname: trino-datalake-hive-metastore
     image: 'starburstdata/hive:3.1.2-e.15'
@@ -91,13 +102,26 @@ services:
     image: 'starburstdata/hive:3.1.2-e.15'
     environment:
       HIVE_METASTORE_DRIVER: org.postgresql.Driver
-      HIVE_METASTORE_JDBC_URL: jdbc:postgresql://trino_metastore_db:5432/datalake_iceberg_metastore
+      HIVE_METASTORE_JDBC_URL: jdbc:postgresql://trino_iceberg_delta_metastore_db:5432/datalake_iceberg_metastore
       HIVE_METASTORE_USER: hive
       HIVE_METASTORE_PASSWORD: hive
       HIVE_METASTORE_WAREHOUSE_DIR: s3://trino/datalake_iceberg
       <<: *hive_metastore_environments
     depends_on:
-      - trino_metastore_db
+      - trino_iceberg_delta_metastore_db
+
+  trino-datalake-delta-hive-metastore:
+    hostname: trino-datalake-delta-hive-metastore
+    image: 'starburstdata/hive:3.1.2-e.15'
+    environment:
+      HIVE_METASTORE_DRIVER: org.postgresql.Driver
+      HIVE_METASTORE_JDBC_URL: jdbc:postgresql://trino_iceberg_delta_metastore_db:5432/datalake_delta_metastore
+      HIVE_METASTORE_USER: hive
+      HIVE_METASTORE_PASSWORD: hive
+      HIVE_METASTORE_WAREHOUSE_DIR: s3://trino/datalake_delta
+      <<: *hive_metastore_environments
+    depends_on:
+      - trino_iceberg_delta_metastore_db
 
   # Spark Stack
   spark:
@@ -154,8 +178,10 @@ services:
       /usr/bin/mc config --quiet host add myminio http://minio:9000 minio minio123;
       /usr/bin/mc mb --quiet myminio/trino/datalake;
       /usr/bin/mc mb --quiet myminio/trino/datalake_iceberg;
+      /usr/bin/mc mb --quiet myminio/trino/datalake_delta;
       /usr/bin/mc mb --quiet myminio/trino/testing;
       /usr/bin/mc mb --quiet myminio/trino/testing_iceberg;
+      /usr/bin/mc mb --quiet myminio/trino/testing_delta;
       /usr/bin/mc mb --quiet myminio/spark/datalake;
       /usr/bin/mc mb --quiet myminio/spark/testing
       "
diff --git a/tests/core/engine_adapter/test_integration.py b/tests/core/engine_adapter/test_integration.py
index 9c79ee4b..3b97f936 100644
--- a/tests/core/engine_adapter/test_integration.py
+++ b/tests/core/engine_adapter/test_integration.py
@@ -449,6 +449,15 @@ def config() -> Config:
                 pytest.mark.xdist_group("engine_integration_trino_iceberg"),
             ],
         ),
+        pytest.param(
+            "trino_delta",
+            marks=[
+                pytest.mark.docker,
+                pytest.mark.engine,
+                pytest.mark.trino_delta,
+                pytest.mark.xdist_group("engine_integration_trino_delta"),
+            ],
+        ),
         pytest.param(
             "spark",
             marks=[
diff --git a/tests/core/engine_adapter/test_redshift.py b/tests/core/engine_adapter/test_redshift.py
index 91044531..df13be88 100644
--- a/tests/core/engine_adapter/test_redshift.py
+++ b/tests/core/engine_adapter/test_redshift.py
@@ -8,7 +8,6 @@ from sqlglot import expressions as exp
 from sqlglot import parse_one
 
 from sqlmesh.core.engine_adapter import RedshiftEngineAdapter
-from sqlmesh.core.engine_adapter.redshift import parse_plan
 from tests.core.engine_adapter import to_sql_calls
 
 pytestmark = [pytest.mark.engine, pytest.mark.redshift]
@@ -33,82 +32,42 @@ def test_columns(adapter: t.Callable):
 def test_create_table_from_query_exists_no_if_not_exists(
     adapter: t.Callable, mocker: MockerFixture
 ):
+    mocker.patch(
+        "sqlmesh.core.engine_adapter.redshift.random_id",
+        return_value="test_random_id",
+    )
+
     mocker.patch(
         "sqlmesh.core.engine_adapter.redshift.RedshiftEngineAdapter.table_exists",
         return_value=True,
     )
 
-    mocker.patch(
-        "sqlmesh.core.engine_adapter.redshift.parse_plan",
+    columns_mock = mocker.patch(
+        "sqlmesh.core.engine_adapter.redshift.RedshiftEngineAdapter.columns",
         return_value={
-            "targetlist": [
-                {
-                    "name": "TARGETENTRY",
-                    "resdom": {
-                        "resname": "a",
-                        "restype": "1043",
-                        "restypmod": "- 1",
-                    },
-                },
-                {
-                    "name": "TARGETENTRY",
-                    "resdom": {
-                        "resname": "b",
-                        "restype": "1043",
-                        "restypmod": "64",
-                    },
-                },
-                {
-                    "name": "TARGETENTRY",
-                    "resdom": {
-                        "resname": "c",
-                        "restype": "1043",
-                        "restypmod": "- 1",
-                    },
-                },
-                {
-                    "name": "TARGETENTRY",
-                    "resdom": {
-                        "resname": "d",
-                        "restype": "1043",
-                        "restypmod": "- 1",
-                    },
-                },
-                {
-                    "name": "TARGETENTRY",
-                    "resdom": {
-                        "resname": "e",
-                        "restype": "1114",
-                        "restypmod": "- 1",
-                    },
-                },
-                {
-                    "name": "TARGETENTRY",
-                    "resdom": {
-                        "resname": "f",
-                        "restype": "0000",  # Unknown type
-                        "restypmod": "- 1",
-                    },
-                },
-            ]
+            "a": exp.DataType.build("VARCHAR(MAX)", dialect="redshift"),
+            "b": exp.DataType.build("VARCHAR(60)", dialect="redshift"),
+            "c": exp.DataType.build("VARCHAR(MAX)", dialect="redshift"),
+            "d": exp.DataType.build("VARCHAR(MAX)", dialect="redshift"),
+            "e": exp.DataType.build("TIMESTAMP", dialect="redshift"),
         },
     )
 
     adapter.ctas(
-        table_name="test_table",
+        table_name="test_schema.test_table",
         query_or_df=parse_one(
-            "SELECT a, b, x + 1 AS c, d AS d, e, f FROM (SELECT * FROM table WHERE FALSE LIMIT 0) WHERE d > 0 AND FALSE LIMIT 0"
+            "SELECT a, b, x + 1 AS c, d AS d, e FROM (SELECT * FROM table WHERE FALSE LIMIT 0) WHERE d > 0 AND FALSE LIMIT 0"
         ),
         exists=False,
     )
 
     assert to_sql_calls(adapter) == [
-        'EXPLAIN VERBOSE CREATE TABLE "test_table" AS SELECT "a", "b", "x" + 1 AS "c", "d" AS "d", "e", "f" FROM (SELECT * FROM "table")',
-        'CREATE TABLE "test_table" AS SELECT CAST(NULL AS VARCHAR(MAX)) AS "a", CAST(NULL AS VARCHAR(60)) AS "b", CAST(NULL '
-        'AS VARCHAR(MAX)) AS "c", CAST(NULL AS VARCHAR(MAX)) AS "d", CAST(NULL AS TIMESTAMP) AS "e", "f" FROM (SELECT "a", "b", "x" + 1 '
-        'AS "c", "d" AS "d", "e", "f" FROM (SELECT * FROM "table" WHERE FALSE LIMIT 0) WHERE "d" > 0 AND FALSE LIMIT 0) AS "_subquery"',
+        'CREATE VIEW "#sqlmesh__test_random_id" AS SELECT "a", "b", "x" + 1 AS "c", "d" AS "d", "e" FROM (SELECT * FROM "table")',
+        'CREATE TABLE "test_schema"."test_table" ("a" VARCHAR(MAX), "b" VARCHAR(60), "c" VARCHAR(MAX), "d" VARCHAR(MAX), "e" TIMESTAMP)',
     ]
 
+    columns_mock.assert_called_once_with(exp.table_("#sqlmesh__test_random_id"))
+
 
 def test_create_table_from_query_exists_and_if_not_exists(
     adapter: t.Callable, mocker: MockerFixture
@@ -293,675 +252,3 @@ def test_create_view(adapter: t.Callable):
         'DROP VIEW IF EXISTS "test_view" CASCADE',
         'CREATE VIEW "test_view" ("a", "b") AS SELECT "cola" FROM "table" WITH NO SCHEMA BINDING',
     ]
-
-
-def test_parse_plan():
-    plan = parse_plan(
-        """
-{ RESULT
-  :startup_cost 0.00
-  :total_cost 0.07
-  :plan_rows 2
-  :node_id 1
-  :parent_id 0
-  :plan_width 32
-  :best_pathkeys <>
-  :dist_info.dist_strategy DS_DIST_NONE
-  :dist_info.dist_keys <>
-  :dist_info.seqscan_is_distributed false
-  :dist_info.topology 1966505242
-  :phys_properties <>
-  :reqd_phys_properties <>
-  :targetlist (
-    { TARGETENTRY
-    :resdom
-      { RESDOM
-      :resno 1
-      :restype 25
-      :restypmod 36
-      :resname cola
-      :ressortgroupref 0
-      :resorigtbl 0
-      :resorigcol 0
-      :resjunk false
-      }
-    :expr
-      { FUNCEXPR
-      :funcid 2311
-      :funcresulttype 25
-      :funcretset false
-      :funcformat 0
-      :args (
-        { RELABELTYPE
-        :arg
-          { VAR
-          :varno 65001
-          :varattno 1
-          :vartype 1043
-          :vartypmod -1
-          :varlevelsup 0
-          :varnoold 2
-          :varoattno 1
-          }
-        :resulttype 25
-        :resulttypmod -1
-        :relabelformat 0
-        }
-      )
-      }
-    }
-  )
-  :non_null_target (b)
-  :qual <>
-  :lefttree
-    { SUBQUERYSCAN
-    :startup_cost 0.00
-    :total_cost 0.06
-    :plan_rows 2
-    :node_id 2
-    :parent_id 1
-    :plan_width 32
-    :best_pathkeys <>
-    :dist_info.dist_strategy DS_DIST_ERR
-    :dist_info.dist_keys <>
-    :dist_info.seqscan_is_distributed false
-    :dist_info.topology 1966505242
-    :phys_properties <>
-    :reqd_phys_properties <>
-    :targetlist (
-      { TARGETENTRY
-      :resdom
-        { RESDOM
-        :resno 1
-        :restype 1043
-        :restypmod -1
-        :resname <>
-        :ressortgroupref 0
-        :resorigtbl 0
-        :resorigcol 0
-        :resjunk false
-        }
-      :expr
-        { VAR
-        :varno 2
-        :varattno 1
-        :vartype 1043
-        :vartypmod -1
-        :varlevelsup 0
-        :varnoold 2
-        :varoattno 1
-        }
-      }
-    )
-    :non_null_target (b)
-    :qual <>
-    :lefttree <>
-    :righttree <>
-    :initPlan <>
-    :extParam (b)
-    :allParam (b)
-    :nParamExec 0
-    :scanrelid 2
-    :tuples 0
-    :subplan
-      { APPEND
-      :startup_cost 0.00
-      :total_cost 0.04
-      :plan_rows 2
-      :node_id 3
-      :parent_id 2
-      :plan_width 0
-      :best_pathkeys <>
-      :dist_info.dist_strategy DS_DIST_ERR
-      :dist_info.dist_keys <>
-      :dist_info.seqscan_is_distributed false
-      :dist_info.topology 1966505242
-      :phys_properties <>
-      :reqd_phys_properties <>
-      :targetlist (
-        { TARGETENTRY
-        :resdom
-          { RESDOM
-          :resno 1
-          :restype 1043
-          :restypmod 5
-          :resname cola
-          :ressortgroupref 0
-          :resorigtbl 0
-          :resorigcol 0
-          :resjunk false
-          }
-        :expr
-          { VAR
-          :varno 0
-          :varattno 1
-          :vartype 1043
-          :vartypmod 5
-          :varlevelsup 0
-          :varnoold 0
-          :varoattno 1
-          }
-        }
-      )
-      :non_null_target (b)
-      :qual <>
-      :lefttree <>
-      :righttree <>
-      :initPlan <>
-      :extParam (b)
-      :allParam (b)
-      :nParamExec 0
-      :appendplans (
-        { NETWORK
-        :startup_cost 0.00
-        :total_cost 0.02
-        :plan_rows 1
-        :node_id 4
-        :parent_id 3
-        :plan_width 0
-        :best_pathkeys <>
-        :dist_info.dist_strategy DS_DIST_ERR
-        :dist_info.dist_keys <>
-        :dist_info.seqscan_is_distributed false
-        :dist_info.topology 1966505242
-        :phys_properties <>
-        :reqd_phys_properties <>
-        :targetlist (
-          { TARGETENTRY
-          :resdom
-            { RESDOM
-            :resno 1
-            :restype 1043
-            :restypmod 5
-            :resname cola
-            :ressortgroupref 0
-            :resorigtbl 0
-            :resorigcol 0
-            :resjunk false
-            }
-          :expr
-            { CONST
-            :consttype 1043
-            :constlen -1
-            :constbyval false
-            :constisnull false
-            :constvalue 5 [ 5 0 0 0 49 ]
-            }
-          }
-        )
-        :non_null_target (b)
-        :qual <>
-        :lefttree
-          { SUBQUERYSCAN
-          :startup_cost 0.00
-          :total_cost 0.02
-          :plan_rows 1
-          :node_id 5
-          :parent_id 4
-          :plan_width 0
-          :best_pathkeys <>
-          :dist_info.dist_strategy DS_DIST_ERR
-          :dist_info.dist_keys <>
-          :dist_info.seqscan_is_distributed false
-          :dist_info.topology 1966505242
-          :phys_properties <>
-          :reqd_phys_properties <>
-          :targetlist (
-            { TARGETENTRY
-            :resdom
-              { RESDOM
-              :resno 1
-              :restype 1043
-              :restypmod 5
-              :resname cola
-              :ressortgroupref 0
-              :resorigtbl 0
-              :resorigcol 0
-              :resjunk false
-              }
-            :expr
-              { FUNCEXPR
-              :funcid 669
-              :funcresulttype 1043
-              :funcretset false
-              :funcformat 2
-              :args (
-                { CONST
-                :consttype 1043
-                :constlen -1
-                :constbyval false
-                :constisnull false
-                :constvalue 5 [ 5 0 0 0 49 ]
-                }
-                { CONST
-                :consttype 23
-                :constlen 4
-                :constbyval true
-                :constisnull false
-                :constvalue 4 [ 5 0 0 0 0 0 0 0 ]
-                }
-                { CONST
-                :consttype 16
-                :constlen 1
-                :constbyval true
-                :constisnull false
-                :constvalue 1 [ 0 0 0 0 0 0 0 0 ]
-                }
-              )
-              }
-            }
-          )
-          :non_null_target (b)
-          :qual <>
-          :lefttree <>
-          :righttree <>
-          :initPlan <>
-          :extParam (b)
-          :allParam (b)
-          :nParamExec 0
-          :scanrelid 1
-          :tuples 0
-          :subplan
-            { RESULT
-            :startup_cost 0.00
-            :total_cost 0.01
-            :plan_rows 1
-            :node_id 6
-            :parent_id 5
-            :plan_width 0
-            :best_pathkeys <>
-            :dist_info.dist_strategy DS_DIST_NONE
-            :dist_info.dist_keys <>
-            :dist_info.seqscan_is_distributed false
-            :dist_info.topology 1966505242
-            :phys_properties <>
-            :reqd_phys_properties <>
-            :targetlist (
-              { TARGETENTRY
-              :resdom
-                { RESDOM
-                :resno 1
-                :restype 1043
-                :restypmod 204
-                :resname cola
-                :ressortgroupref 0
-                :resorigtbl 0
-                :resorigcol 0
-                :resjunk false
-                }
-              :expr
-                { CONST
-                :consttype 1043
-                :constlen -1
-                :constbyval false
-                :constisnull false
-                :constvalue 5 [ 5 0 0 0 49 ]
-                }
-              }
-            )
-            :non_null_target (b)
-            :qual <>
-            :lefttree <>
-            :righttree <>
-            :initPlan <>
-            :extParam (b)
-            :allParam (b)
-            :nParamExec 0
-            :resconstantqual <>
-            :scanrelid 0
-            :values_lists <>
-            }
-          }
-        :righttree <>
-        :initPlan <>
-        :extParam (b)
-        :allParam (b)
-        :nParamExec 0
-        :dataMovement Distribute
-        }
-        { NETWORK
-        :startup_cost 0.00
-        :total_cost 0.02
-        :plan_rows 1
-        :node_id 7
-        :parent_id 3
-        :plan_width 0
-        :best_pathkeys <>
-        :dist_info.dist_strategy DS_DIST_ERR
-        :dist_info.dist_keys <>
-        :dist_info.seqscan_is_distributed false
-        :dist_info.topology 1966505242
-        :phys_properties <>
-        :reqd_phys_properties <>
-        :targetlist (
-          { TARGETENTRY
-          :resdom
-            { RESDOM
-            :resno 1
-            :restype 1043
-            :restypmod 5
-            :resname cola
-            :ressortgroupref 0
-            :resorigtbl 0
-            :resorigcol 0
-            :resjunk false
-            }
-          :expr
-            { CONST
-            :consttype 1043
-            :constlen -1
-            :constbyval false
-            :constisnull false
-            :constvalue 5 [ 5 0 0 0 51 ]
-            }
-          }
-        )
-        :non_null_target (b)
-        :qual <>
-        :lefttree
-          { SUBQUERYSCAN
-          :startup_cost 0.00
-          :total_cost 0.02
-          :plan_rows 1
-          :node_id 8
-          :parent_id 7
-          :plan_width 0
-          :best_pathkeys <>
-          :dist_info.dist_strategy DS_DIST_ERR
-          :dist_info.dist_keys <>
-          :dist_info.seqscan_is_distributed false
-          :dist_info.topology 1966505242
-          :phys_properties <>
-          :reqd_phys_properties <>
-          :targetlist (
-            { TARGETENTRY
-            :resdom
-              { RESDOM
-              :resno 1
-              :restype 1043
-              :restypmod 5
-              :resname cola
-              :ressortgroupref 0
-              :resorigtbl 0
-              :resorigcol 0
-              :resjunk false
-              }
-            :expr
-              { FUNCEXPR
-              :funcid 669
-              :funcresulttype 1043
-              :funcretset false
-              :funcformat 2
-              :args (
-                { CONST
-                :consttype 1043
-                :constlen -1
-                :constbyval false
-                :constisnull false
-                :constvalue 5 [ 5 0 0 0 51 ]
-                }
-                { CONST
-                :consttype 23
-                :constlen 4
-                :constbyval true
-                :constisnull false
-                :constvalue 4 [ 5 0 0 0 0 0 0 0 ]
-                }
-                { CONST
-                :consttype 16
-                :constlen 1
-                :constbyval true
-                :constisnull false
-                :constvalue 1 [ 0 0 0 0 0 0 0 0 ]
-                }
-              )
-              }
-            }
-          )
-          :non_null_target (b)
-          :qual <>
-          :lefttree <>
-          :righttree <>
-          :initPlan <>
-          :extParam (b)
-          :allParam (b)
-          :nParamExec 0
-          :scanrelid 2
-          :tuples 0
-          :subplan
-            { RESULT
-            :startup_cost 0.00
-            :total_cost 0.01
-            :plan_rows 1
-            :node_id 9
-            :parent_id 8
-            :plan_width 0
-            :best_pathkeys <>
-            :dist_info.dist_strategy DS_DIST_NONE
-            :dist_info.dist_keys <>
-            :dist_info.seqscan_is_distributed false
-            :dist_info.topology 1966505242
-            :phys_properties <>
-            :reqd_phys_properties <>
-            :targetlist (
-              { TARGETENTRY
-              :resdom
-                { RESDOM
-                :resno 1
-                :restype 1043
-                :restypmod 204
-                :resname varchar
-                :ressortgroupref 0
-                :resorigtbl 0
-                :resorigcol 0
-                :resjunk false
-                }
-              :expr
-                { CONST
-                :consttype 1043
-                :constlen -1
-                :constbyval false
-                :constisnull false
-                :constvalue 5 [ 5 0 0 0 51 ]
-                }
-              }
-            )
-            :non_null_target (b)
-            :qual <>
-            :lefttree <>
-            :righttree <>
-            :initPlan <>
-            :extParam (b)
-            :allParam (b)
-            :nParamExec 0
-            :resconstantqual <>
-            :scanrelid 0
-            :values_lists <>
-            }
-          }
-        :righttree <>
-        :initPlan <>
-        :extParam (b)
-        :allParam (b)
-        :nParamExec 0
-        :dataMovement Distribute
-        }
-      )
-      :isTarget false
-      :streamable false
-      }
-    }
-  :righttree <>
-  :initPlan <>
-  :extParam (b)
-  :allParam (b)
-  :nParamExec 0
-  :resconstantqual (
-    { CONST
-    :consttype 16
-    :constlen 1
-    :constbyval true
-    :constisnull false
-    :constvalue 1 [ 0 0 0 0 0 0 0 0 ]
-    }
-    { CONST
-    :consttype 16
-    :constlen 1
-    :constbyval true
-    :constisnull false
-    :constvalue 1 [ 0 0 0 0 0 0 0 0 ]
-    }
-  )
-  :scanrelid 0
-  :values_lists <>
-  }
-    """
-    )
-    assert plan["targetlist"] == [
-        {
-            "expr": {
-                "args": [
-                    {
-                        "arg": {
-                            "name": "VAR",
-                            "varattno": "1",
-                            "varlevelsup": "0",
-                            "varno": "65001",
-                            "varnoold": "2",
-                            "varoattno": "1",
-                            "vartype": "1043",
-                            "vartypmod": "- 1",
-                        },
-                        "name": "RELABELTYPE",
-                        "relabelformat": "0",
-                        "resulttype": "25",
-                        "resulttypmod": "- 1",
-                    }
-                ],
-                "funcformat": "0",
-                "funcid": "2311",
-                "funcresulttype": "25",
-                "funcretset": "false",
-                "name": "FUNCEXPR",
-            },
-            "name": "TARGETENTRY",
-            "resdom": {
-                "name": "RESDOM",
-                "resjunk": "false",
-                "resname": "cola",
-                "resno": "1",
-                "resorigcol": "0",
-                "resorigtbl": "0",
-                "ressortgroupref": "0",
-                "restype": "25",
-                "restypmod": "36",
-            },
-        },
-    ]
-
-
-def test_parse_plan_limit():
-    plan = parse_plan(
-        """
-{ LIMIT
-  :startup_cost 0.00
-  :total_cost 0.00
-  :plan_rows 1
-  :node_id 1
-  :parent_id 0
-  :plan_width 0
-  :best_pathkeys <>
-  :phys_properties <>
-  :reqd_phys_properties <>
-  :targetlist (
-    { TARGETENTRY
-    :resdom
-      { RESDOM
-      :resno 1
-      :restype 20
-      :restypmod -1
-      :resname ?column?
-      :ressortgroupref 0
-      :resorigtbl 0
-      :resorigcol 0
-      :resjunk false
-      }
-    :expr
-      { CONST
-      :consttype 20
-      :constlen 8
-      :constbyval true
-      :constisnull true
-      :constvalue <>
-      }
-    }
-  )
-  }
-
-XN Limit  (cost=0.00..0.00 rows=1 width=0)
-  ->  XN Result  (cost=0.00..0.01 rows=1 width=0)
-        One-Time Filter: false
-
-  { RESULT
-  :targetlist (
-    { TARGETENTRY
-    :resdom
-      { RESDOM
-      :resno 1
-      :restype 1043
-      :restypmod 36
-      :resname surrogate_key
-      :ressortgroupref 0
-      :resorigtbl 0
-      :resorigcol 0
-      :resjunk false
-      }
-    :expr
-      { VAR
-      :varno 65001
-      :varattno 1
-      :vartype 1043
-      :vartypmod 36
-      :varlevelsup 0
-      :varnoold 1
-      :varoattno 1
-      }
-    }
-  )
-  :scanrelid 0
-  :values_lists <>
-  }
-        """
-    )
-
-    assert plan == {
-        "name": "RESULT",
-        "scanrelid": "0",
-        "targetlist": [
-            {
-                "expr": {
-                    "name": "VAR",
-                    "varattno": "1",
-                    "varlevelsup": "0",
-                    "varno": "65001",
-                    "varnoold": "1",
-                    "varoattno": "1",
-                    "vartype": "1043",
-                    "vartypmod": "36",
-                },
-                "name": "TARGETENTRY",
-                "resdom": {
-                    "name": "RESDOM",
-                    "resjunk": "false",
-                    "resname": "surrogate_key",
-                    "resno": "1",
-                    "resorigcol": "0",
-                    "resorigtbl": "0",
-                    "ressortgroupref": "0",
-                    "restype": "1043",
-                    "restypmod": "36",
-                },
-            }
-        ],
-        "values_lists": "<>",
-    }
diff --git a/tests/core/engine_adapter/test_trino.py b/tests/core/engine_adapter/test_trino.py
index 925c581f..b76fa22c 100644
--- a/tests/core/engine_adapter/test_trino.py
+++ b/tests/core/engine_adapter/test_trino.py
@@ -21,6 +21,8 @@ def trino_mocked_engine_adapter(
     def mock_catalog_type(catalog_name):
         if "iceberg" in catalog_name:
             return "iceberg"
+        if "delta" in catalog_name:
+            return "delta"
         return "hive"
 
     mocker.patch(
@@ -40,7 +42,12 @@ def test_set_current_catalog(trino_mocked_engine_adapter: TrinoEngineAdapter):
     ]
 
 
-def test_get_catalog_type(trino_mocked_engine_adapter: TrinoEngineAdapter, mocker: MockerFixture):
+@pytest.mark.trino_iceberg
+@pytest.mark.trino_delta
+@pytest.mark.parametrize("storage_type", ["iceberg", "delta"])
+def test_get_catalog_type(
+    trino_mocked_engine_adapter: TrinoEngineAdapter, mocker: MockerFixture, storage_type: str
+):
     adapter = trino_mocked_engine_adapter
     mocker.patch(
         "sqlmesh.core.engine_adapter.trino.TrinoEngineAdapter.get_current_catalog",
@@ -51,12 +58,13 @@ def test_get_catalog_type(trino_mocked_engine_adapter: TrinoEngineAdapter, mocke
     assert adapter.get_catalog_type("foo") == TrinoEngineAdapter.DEFAULT_CATALOG_TYPE
     assert adapter.get_catalog_type("datalake_hive") == "hive"
     assert adapter.get_catalog_type("datalake_iceberg") == "iceberg"
+    assert adapter.get_catalog_type("datalake_delta") == "delta"
 
     mocker.patch(
         "sqlmesh.core.engine_adapter.trino.TrinoEngineAdapter.get_current_catalog",
-        return_value="system_iceberg",
+        return_value=f"system_{storage_type}",
     )
-    assert adapter.current_catalog_type == "iceberg"
+    assert adapter.current_catalog_type == storage_type
 
 
 def test_get_catalog_type_cached(
@@ -89,16 +97,18 @@ def test_get_catalog_type_cached(
     assert fetchone_mock.call_count == 2
 
 
-def test_partitioned_by_hive(
-    trino_mocked_engine_adapter: TrinoEngineAdapter, mocker: MockerFixture
+@pytest.mark.trino_delta
+@pytest.mark.parametrize("storage_type", ["hive", "delta"])
+def test_partitioned_by_hive_delta(
+    trino_mocked_engine_adapter: TrinoEngineAdapter, mocker: MockerFixture, storage_type: str
 ):
     adapter = trino_mocked_engine_adapter
 
     mocker.patch(
         "sqlmesh.core.engine_adapter.trino.TrinoEngineAdapter.get_current_catalog",
-        return_value="datalake_hive",
+        return_value=f"datalake_{storage_type}",
     )
-    assert adapter.get_catalog_type("datalake_hive") == "hive"
+    assert adapter.get_catalog_type(f"datalake_{storage_type}") == storage_type
 
     columns_to_types = {
         "cola": exp.DataType.build("INT"),
@@ -115,6 +125,7 @@ def test_partitioned_by_hive(
     ]
 
 
+@pytest.mark.trino_iceberg
 def test_partitioned_by_iceberg(
     trino_mocked_engine_adapter: TrinoEngineAdapter, mocker: MockerFixture
 ):
@@ -141,6 +152,7 @@ def test_partitioned_by_iceberg(
     ]
 
 
+@pytest.mark.trino_iceberg
 def test_partitioned_by_iceberg_transforms(
     trino_mocked_engine_adapter: TrinoEngineAdapter, mocker: MockerFixture
 ):
@@ -184,6 +196,7 @@ def test_partitioned_by_iceberg_transforms(
     ]
 
 
+@pytest.mark.trino_iceberg
 def test_partitioned_by_with_multiple_catalogs_same_server(
     trino_mocked_engine_adapter: TrinoEngineAdapter, mocker: MockerFixture
 ):
@@ -194,6 +207,11 @@ def test_partitioned_by_with_multiple_catalogs_same_server(
         "colb": exp.DataType.build("TEXT"),
     }
 
+    mocker.patch(
+        "sqlmesh.core.engine_adapter.trino.TrinoEngineAdapter.get_current_catalog",
+        return_value="system",
+    )
+
     adapter.create_table(
         "datalake.test_schema.test_table", columns_to_types, partitioned_by=[exp.to_column("colb")]
     )
@@ -204,6 +222,11 @@ def test_partitioned_by_with_multiple_catalogs_same_server(
         partitioned_by=[exp.to_column("colb")],
     )
 
+    mocker.patch(
+        "sqlmesh.core.engine_adapter.trino.TrinoEngineAdapter.get_current_catalog",
+        return_value="iceberg",
+    )
+
     adapter.create_table(
         "datalake_iceberg.test_schema.test_table",
         columns_to_types,
@@ -287,17 +310,22 @@ def test_comments_hive(mocker: MockerFixture, make_mocked_engine_adapter: t.Call
     ]
 
 
-def test_comments_iceberg(mocker: MockerFixture, make_mocked_engine_adapter: t.Callable):
+@pytest.mark.trino_iceberg
+@pytest.mark.trino_delta
+@pytest.mark.parametrize("storage_type", ["iceberg", "delta"])
+def test_comments_iceberg_delta(
+    mocker: MockerFixture, make_mocked_engine_adapter: t.Callable, storage_type: str
+):
     adapter = make_mocked_engine_adapter(TrinoEngineAdapter)
 
     current_catalog_mock = mocker.patch(
         "sqlmesh.core.engine_adapter.trino.TrinoEngineAdapter.get_current_catalog"
     )
-    current_catalog_mock.return_value = "iceberg"
+    current_catalog_mock.return_value = storage_type
     catalog_type_mock = mocker.patch(
         "sqlmesh.core.engine_adapter.trino.TrinoEngineAdapter.get_catalog_type"
     )
-    catalog_type_mock.return_value = "iceberg"
+    catalog_type_mock.return_value = storage_type
 
     allowed_table_comment_length = TrinoEngineAdapter.MAX_TABLE_COMMENT_LENGTH
     truncated_table_comment = "a" * allowed_table_comment_length
@@ -348,3 +376,27 @@ def test_comments_iceberg(mocker: MockerFixture, make_mocked_engine_adapter: t.C
         f"""COMMENT ON TABLE "test_table" IS '{long_table_comment}'""",
         f"""COMMENT ON COLUMN "test_table"."a" IS '{long_column_comment}'""",
     ]
+
+
+@pytest.mark.trino_delta
+def test_delta_timestamps(make_mocked_engine_adapter: t.Callable):
+    adapter = make_mocked_engine_adapter(TrinoEngineAdapter)
+
+    ts6 = exp.DataType.build("timestamp(6)")
+    ts3_tz = exp.DataType.build("timestamp(3) with time zone")
+
+    columns_to_types = {
+        "ts": exp.DataType.build("TIMESTAMP"),
+        "ts_1": exp.DataType.build("TIMESTAMP(1)"),
+        "ts_tz": exp.DataType.build("TIMESTAMP WITH TIME ZONE"),
+        "ts_tz_1": exp.DataType.build("TIMESTAMP(1) WITH TIME ZONE"),
+    }
+
+    delta_columns_to_types = adapter._to_delta_ts(columns_to_types)
+
+    assert delta_columns_to_types == {
+        "ts": ts6,
+        "ts_1": ts6,
+        "ts_tz": ts3_tz,
+        "ts_tz_1": ts3_tz,
+    }
diff --git a/tests/core/engine_adapter/trino/catalog/datalake_delta.properties b/tests/core/engine_adapter/trino/catalog/datalake_delta.properties
new file mode 100644
index 00000000..a2ee5c99
--- /dev/null
+++ b/tests/core/engine_adapter/trino/catalog/datalake_delta.properties
@@ -0,0 +1,11 @@
+connector.name=delta_lake
+hive.metastore.uri=thrift://trino-datalake-delta-hive-metastore:9083
+hive.metastore-cache-ttl=0s
+hive.metastore-refresh-interval=5s
+hive.metastore-timeout=10s
+hive.s3.endpoint=http://minio:9000
+hive.s3.aws-access-key=minio
+hive.s3.aws-secret-key=minio123
+hive.s3.path-style-access=true
+delta.enable-non-concurrent-writes=true
+delta.hive-catalog-name=datalake_delta
\ No newline at end of file
diff --git a/tests/core/engine_adapter/trino/initdb.sql b/tests/core/engine_adapter/trino/initdb.sql
index 023f3602..ae71ef66 100644
--- a/tests/core/engine_adapter/trino/initdb.sql
+++ b/tests/core/engine_adapter/trino/initdb.sql
@@ -1,4 +1,6 @@
 create database datalake_metastore;
 create database datalake_iceberg_metastore;
+create database datalake_delta_metastore;
 create database testing_metastore;
-create database testing_iceberg_metastore;
\ No newline at end of file
+create database testing_iceberg_metastore;
+create database testing_delta_metastore;
\ No newline at end of file