Ignore primary key, foreign key and unique constraints for DDL generation

tests/unit/sqlalchemy/test_compiler.py

@@ -11,13 +11,15 @@
 # limitations under the License.
 import pytest
 from sqlalchemy import Column
+from sqlalchemy import ForeignKey
 from sqlalchemy import func
 from sqlalchemy import insert
 from sqlalchemy import Integer
 from sqlalchemy import MetaData
 from sqlalchemy import select
 from sqlalchemy import String
 from sqlalchemy import Table
+from sqlalchemy.exc import SAWarning
 from sqlalchemy.schema import CreateTable
 from sqlalchemy.sql import column
 from sqlalchemy.sql import table
@@ -40,6 +42,26 @@
     trino_catalog='other'
 )
 
+table_with_pk = Table(
+    'table_with_pk',
+    metadata,
+    Column('id', String, primary_key=True)
+)
+
+table_with_fk = Table(
+    'table_with_fk',
+    metadata,
+    Column('id', String, primary_key=True),
+    Column('fk', String, ForeignKey('table_with_pk.id'))
+)
+
+table_with_unique = Table(
+    'table_with_constraint',
+    metadata,
+    Column('id', String, primary_key=True),
+    Column('uniq', String, unique=True)
+)
+
 
 @pytest.fixture
 def dialect():
@@ -170,3 +192,24 @@ def test_try_cast(dialect):
     statement = select(try_cast(table_without_catalog.c.id, String))
     query = statement.compile(dialect=dialect)
     assert str(query) == 'SELECT try_cast("table".id as VARCHAR) AS id \nFROM "table"'
+
+
+def test_catalogs_create_table_with_pk(dialect):
+    with pytest.warns(SAWarning, match="Trino does not support PRIMARY KEY constraints. Constraint will be ignored."):
+        statement = CreateTable(table_with_pk)
+        query = statement.compile(dialect=dialect)
+        assert 'primary key' not in str(query).lower()
+
+
+def test_catalogs_create_table_with_fk(dialect):
+    with pytest.warns(SAWarning, match="Trino does not support FOREIGN KEY constraints. Constraint will be ignored."):
+        statement = CreateTable(table_with_fk)
+        query = statement.compile(dialect=dialect)
+        assert 'foreign key' not in str(query).lower()
+
+
+def test_catalogs_create_table_with_unique(dialect):
+    with pytest.warns(SAWarning, match="Trino does not support UNIQUE constraints. Constraint will be ignored."):
+        statement = CreateTable(table_with_unique)
+        query = statement.compile(dialect=dialect)
+        assert 'unique' not in str(query).lower()

trino/sqlalchemy/compiler.py

@@ -14,6 +14,7 @@
 from sqlalchemy.sql import sqltypes
 from sqlalchemy.sql.base import DialectKWArgs
 from sqlalchemy.sql.functions import GenericFunction
+from sqlalchemy.util import warn as SAWarn
 
 # https://trino.io/docs/current/language/reserved.html
 RESERVED_WORDS = {
@@ -181,7 +182,17 @@ def visit_try_cast(self, element, **kw):
 
 
 class TrinoDDLCompiler(compiler.DDLCompiler):
-    pass
+    def visit_foreign_key_constraint(self, constraint, **kw):
+        SAWarn("Trino does not support FOREIGN KEY constraints. Constraint will be ignored.")
+        return None
+
+    def visit_primary_key_constraint(self, constraint, **kw):
+        SAWarn("Trino does not support PRIMARY KEY constraints. Constraint will be ignored.")
+        return None
+
+    def visit_unique_constraint(self, constraint, **kw):
+        SAWarn("Trino does not support UNIQUE constraints. Constraint will be ignored.")
+        return None
 
 
 class TrinoTypeCompiler(compiler.GenericTypeCompiler):