Propertly ignore ``Identity`` in MySQL and MariaDb.

Ensure that the MySQL and MariaDB dialect ignore the
:class:`_sql.Identity` construct while rendering the
``AUTO_INCREMENT`` keyword in a create table.

The Oracle and PostgreSQL compiler was updated to not render
:class:`_sql.Identity` if the database version does not support it
(Oracle < 12 and PostgreSQL < 10). Previously it was rendered regardless
of the database version.

Fixes: #6338
Change-Id: I2ca0902fdd7b4be4fc1a563cf5585504cbea9360

7 files changed, 61 insertions, 7 deletions

diff --git a/lib/sqlalchemy/dialects/mysql/base.py b/lib/sqlalchemy/dialects/mysql/base.py
index d4c70a78e..88c03a011 100644
--- a/lib/sqlalchemy/dialects/mysql/base.py
+++ b/lib/sqlalchemy/dialects/mysql/base.py
@@ -1933,7 +1933,10 @@ class MySQLDDLCompiler(compiler.DDLCompiler):
         if (
             column.table is not None
             and column is column.table._autoincrement_column
-            and column.server_default is None
+            and (
+                column.server_default is None
+                or isinstance(column.server_default, sa_schema.Identity)
+            )
             and not (
                 self.dialect.supports_sequences
                 and isinstance(column.default, sa_schema.Sequence)
diff --git a/lib/sqlalchemy/dialects/oracle/base.py b/lib/sqlalchemy/dialects/oracle/base.py
index 9571eb46f..57f55c182 100644
--- a/lib/sqlalchemy/dialects/oracle/base.py
+++ b/lib/sqlalchemy/dialects/oracle/base.py
@@ -1457,6 +1457,7 @@ class OracleDialect(default.DefaultDialect):
     supports_default_values = False
     supports_default_metavalue = True
     supports_empty_insert = False
+    supports_identity_columns = True
 
     statement_compiler = OracleCompiler
     ddl_compiler = OracleDDLCompiler
@@ -1513,6 +1514,8 @@ class OracleDialect(default.DefaultDialect):
             self.colspecs.pop(sqltypes.Interval)
             self.use_ansi = False
 
+        self.supports_identity_columns = self.server_version_info >= (12,)
+
     def _get_effective_compat_server_version_info(self, connection):
         # dialect does not need compat levels below 12.2, so don't query
         # in those cases
diff --git a/lib/sqlalchemy/dialects/postgresql/base.py b/lib/sqlalchemy/dialects/postgresql/base.py
index 47a933479..26b025b1a 100644
--- a/lib/sqlalchemy/dialects/postgresql/base.py
+++ b/lib/sqlalchemy/dialects/postgresql/base.py
@@ -2464,6 +2464,11 @@ class PGDDLCompiler(compiler.DDLCompiler):
         if isinstance(impl_type, sqltypes.TypeDecorator):
             impl_type = impl_type.impl
 
+        has_identity = (
+            column.identity is not None
+            and self.dialect.supports_identity_columns
+        )
+
         if (
             column.primary_key
             and column is column.table._autoincrement_column
@@ -2471,7 +2476,7 @@ class PGDDLCompiler(compiler.DDLCompiler):
                 self.dialect.supports_smallserial
                 or not isinstance(impl_type, sqltypes.SmallInteger)
             )
-            and column.identity is None
+            and not has_identity
             and (
                 column.default is None
                 or (
@@ -2498,12 +2503,12 @@ class PGDDLCompiler(compiler.DDLCompiler):
 
         if column.computed is not None:
             colspec += " " + self.process(column.computed)
-        if column.identity is not None:
+        if has_identity:
             colspec += " " + self.process(column.identity)
 
-        if not column.nullable and not column.identity:
+        if not column.nullable and not has_identity:
             colspec += " NOT NULL"
-        elif column.nullable and column.identity:
+        elif column.nullable and has_identity:
             colspec += " NULL"
         return colspec
 
@@ -3086,6 +3091,8 @@ class PGDialect(default.DefaultDialect):
 
     supports_empty_insert = False
     supports_multivalues_insert = True
+    supports_identity_columns = True
+
     default_paramstyle = "pyformat"
     ischema_names = ischema_names
     colspecs = colspecs
@@ -3193,6 +3200,7 @@ class PGDialect(default.DefaultDialect):
             9,
             2,
         )
+        self.supports_identity_columns = self.server_version_info >= (10,)
 
     def on_connect(self):
         if self.isolation_level is not None:
diff --git a/lib/sqlalchemy/engine/default.py b/lib/sqlalchemy/engine/default.py
index a917228ad..aa7e4e5e9 100644
--- a/lib/sqlalchemy/engine/default.py
+++ b/lib/sqlalchemy/engine/default.py
@@ -74,6 +74,7 @@ class DefaultDialect(interfaces.Dialect):
     supports_sequences = False
     sequences_optional = False
     preexecute_autoincrement_sequences = False
+    supports_identity_columns = False
     postfetch_lastrowid = True
     implicit_returning = False
     full_returning = False
@@ -808,6 +809,8 @@ class StrCompileDialect(DefaultDialect):
 
     supports_statement_cache = True
 
+    supports_identity_columns = True
+
     supports_sequences = True
     sequences_optional = True
     preexecute_autoincrement_sequences = False
diff --git a/lib/sqlalchemy/sql/compiler.py b/lib/sqlalchemy/sql/compiler.py
index 6168248ff..bd93f5199 100644
--- a/lib/sqlalchemy/sql/compiler.py
+++ b/lib/sqlalchemy/sql/compiler.py
@@ -4257,10 +4257,15 @@ class DDLCompiler(Compiled):
         if column.computed is not None:
             colspec += " " + self.process(column.computed)
 
-        if column.identity is not None:
+        if (
+            column.identity is not None
+            and self.dialect.supports_identity_columns
+        ):
             colspec += " " + self.process(column.identity)
 
-        if not column.nullable and not column.identity:
+        if not column.nullable and (
+            not column.identity or not self.dialect.supports_identity_columns
+        ):
             colspec += " NOT NULL"
         return colspec
 
diff --git a/lib/sqlalchemy/testing/requirements.py b/lib/sqlalchemy/testing/requirements.py
index 8a70cc692..673fa15cd 100644
--- a/lib/sqlalchemy/testing/requirements.py
+++ b/lib/sqlalchemy/testing/requirements.py
@@ -1417,3 +1417,10 @@ class SuiteRequirements(Requirements):
         or ties. basically this is "not mssql"
         """
         return exclusions.closed()
+
+    @property
+    def autoincrement_without_sequence(self):
+        """If autoincrement=True on a column does not require an explicit
+        sequence. This should be false only for oracle.
+        """
+        return exclusions.open()
diff --git a/lib/sqlalchemy/testing/suite/test_select.py b/lib/sqlalchemy/testing/suite/test_select.py
index 7b35dc3fa..8f3412929 100644
--- a/lib/sqlalchemy/testing/suite/test_select.py
+++ b/lib/sqlalchemy/testing/suite/test_select.py
@@ -1442,6 +1442,31 @@ class IdentityColumnTest(fixtures.TablesTest):
         assert_raises((DatabaseError, ProgrammingError), fn)
 
 
+class IdentityAutoincrementTest(fixtures.TablesTest):
+    __backend__ = True
+    __requires__ = ("autoincrement_without_sequence",)
+
+    @classmethod
+    def define_tables(cls, metadata):
+        Table(
+            "tbl",
+            metadata,
+            Column(
+                "id",
+                Integer,
+                Identity(),
+                primary_key=True,
+                autoincrement=True,
+            ),
+            Column("desc", String(100)),
+        )
+
+    def test_autoincrement_with_identity(self, connection):
+        res = connection.execute(self.tables.tbl.insert(), {"desc": "row"})
+        res = connection.execute(self.tables.tbl.select()).first()
+        eq_(res, (1, "row"))
+
+
 class ExistsTest(fixtures.TablesTest):
     __backend__ = True