Feat: support Trino Delta Lake connector (#2330)

* Add trino delta lake connector support

* Clarify timestamp conversion explanation

* PR feedback

Author: treysp

Makefile

@@ -179,4 +179,4 @@ spark-pyspark-test:
 	pytest -n auto -m "spark_pyspark"
 
 trino-test:
-	pytest -n auto -m "trino or trino_iceberg"
+	pytest -n auto -m "trino or trino_iceberg or trino_delta"

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 |

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

sqlmesh/core/engine_adapter/mixins.py

@@ -224,8 +224,8 @@ def _build_view_properties_exp(
     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)
 

sqlmesh/core/engine_adapter/trino.py

@@ -24,12 +24,13 @@
     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 @@ def _df_to_source_queries(
                 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

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

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
       "

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=[