Add type tests

Author: danthelion

.github/workflows/ci.yml

@@ -34,7 +34,7 @@ jobs:
           python-version: ${{ matrix.python-version }}
 
       - name: Build the stack
-        run: docker-compose up -d mysql postgres presto
+        run: docker-compose up -d mysql postgres presto trino
 
       - name: Install Poetry
         run: pip install poetry
@@ -46,4 +46,5 @@ jobs:
         env:
             DATADIFF_SNOWFLAKE_URI: '${{ secrets.DATADIFF_SNOWFLAKE_URI }}'
             DATADIFF_PRESTO_URI: '${{ secrets.DATADIFF_PRESTO_URI }}'
+            DATADIFF_TRINO_URI: '${{ secrets.DATADIFF_TRINO_URI }}'
         run: poetry run unittest-parallel -j 16

README.md

@@ -169,6 +169,8 @@ While you may install them manually, we offer an easy way to install them along
 
 - `pip install 'data-diff[oracle]'`
 
+- `pip install 'data-diff[trino]'`
+
 - For BigQuery, see: https://pypi.org/project/google-cloud-bigquery/
 
 

data_diff/databases/trino.py

@@ -54,13 +54,15 @@ def _query(self, sql_code: str) -> list:
         c.execute(sql_code)
         if sql_code.lower().startswith("select"):
             return c.fetchall()
+        if re.match(r"(insert|create|truncate|drop)", sql_code, re.IGNORECASE):
+            return c.fetchone()
 
     def close(self):
         self._conn.close()
 
     def normalize_timestamp(self, value: str, coltype: TemporalType) -> str:
         if coltype.rounds:
-            s = f"date_format(cast({coltype.precision} as timestamp(6)), '%Y-%m-%d %H:%i:%S.%f')"
+            s = f"date_format(cast({value} as timestamp({coltype.precision})), '%Y-%m-%d %H:%i:%S.%f')"
         else:
             s = f"date_format(cast({value} as timestamp(6)), '%Y-%m-%d %H:%i:%S.%f')"
 

dev/trino-conf/etc/catalog/jms.properties

@@ -0,0 +1 @@
+connector.name=jmx

docker-compose.yml

@@ -70,7 +70,7 @@ services:
         ports:
             - '8080:8080'
         volumes:
-            - ./dev/trino-conf/etc:/etc/trino
+            - ./dev/trino-conf/etc:/etc/trino:ro
         networks:
             - local
 

tests/test_database_types.py

@@ -375,7 +375,27 @@ def __iter__(self):
         "uuid": [
             "STRING",
         ]
-    }
+    },
+    db.Trino: {
+        "int": [
+            "int",
+            "bigint",
+        ],
+        "datetime": [
+            "timestamp",
+            "timestamp with time zone",
+        ],
+        "float": [
+            "real",
+            "double",
+            "decimal(10,2)",
+            "decimal(30,6)",
+        ],
+        "uuid": [
+            "varchar",
+            "char(100)",
+        ],
+    },
 }
 
 
@@ -460,7 +480,7 @@ def _insert_to_table(conn, table, values, type):
 
         if isinstance(sample, (float, Decimal, int)):
             value = str(sample)
-        elif isinstance(sample, datetime) and isinstance(conn, (db.Presto, db.Oracle)):
+        elif isinstance(sample, datetime) and isinstance(conn, (db.Presto, db.Oracle, db.Trino)):
             value = f"timestamp '{sample}'"
         elif isinstance(sample, datetime) and isinstance(conn, db.BigQuery) and type == 'datetime':
             value = f"cast(timestamp '{sample}' as datetime)"
@@ -492,7 +512,7 @@ def _insert_to_table(conn, table, values, type):
 def _create_indexes(conn, table):
     # It is unfortunate that Presto doesn't support creating indexes...
     # Technically we could create it in the backing Postgres behind the scenes.
-    if isinstance(conn, (db.Snowflake, db.Redshift, db.Presto, db.BigQuery, db.Databricks)):
+    if isinstance(conn, (db.Snowflake, db.Redshift, db.Presto, db.BigQuery, db.Databricks, db.Trino)):
         return
 
     try:
@@ -583,7 +603,7 @@ def test_types(self, source_db, target_db, source_type, target_type, type_catego
         insertion_source_duration = time.time() - start
 
         values_in_source = PaginatedTable(src_table, src_conn)
-        if source_db is db.Presto:
+        if source_db is db.Presto or source_db is db.Trino:
             if source_type.startswith("decimal"):
                 values_in_source = ((a, Decimal(b)) for a, b in values_in_source)
             elif source_type.startswith("timestamp"):