Add support for identity columns

Added the :class:`_schema.Identity` construct that can be used to
configure identity columns rendered with GENERATED { ALWAYS |
BY DEFAULT } AS IDENTITY. Currently the supported backends are
PostgreSQL >= 10, Oracle >= 12 and MSSQL (with different syntax
and a subset of functionalities).

Fixes: #5362
Fixes: #5324
Fixes: #5360

Change-Id: Iecea6f3ceb36821e8b96f0b61049b580507a1875

5 files changed, 298 insertions, 93 deletions

diff --git a/test/dialect/mssql/test_compiler.py b/test/dialect/mssql/test_compiler.py
index 83a610888..67120e8fe 100644
--- a/test/dialect/mssql/test_compiler.py
+++ b/test/dialect/mssql/test_compiler.py
@@ -2,8 +2,10 @@
 from sqlalchemy import Column
 from sqlalchemy import Computed
 from sqlalchemy import delete
+from sqlalchemy import exc
 from sqlalchemy import extract
 from sqlalchemy import func
+from sqlalchemy import Identity
 from sqlalchemy import Index
 from sqlalchemy import insert
 from sqlalchemy import Integer
@@ -27,6 +29,7 @@ from sqlalchemy.dialects.mssql.base import try_cast
 from sqlalchemy.sql import column
 from sqlalchemy.sql import quoted_name
 from sqlalchemy.sql import table
+from sqlalchemy.testing import assert_raises_message
 from sqlalchemy.testing import AssertsCompiledSQL
 from sqlalchemy.testing import eq_
 from sqlalchemy.testing import fixtures
@@ -1116,96 +1119,6 @@ class CompileTest(fixtures.TestBase, AssertsCompiledSQL):
             checkparams={"x_1": 5, "param_1": 0},
         )
 
-    def test_primary_key_no_identity(self):
-        metadata = MetaData()
-        tbl = Table(
-            "test",
-            metadata,
-            Column("id", Integer, autoincrement=False, primary_key=True),
-        )
-        self.assert_compile(
-            schema.CreateTable(tbl),
-            "CREATE TABLE test (id INTEGER NOT NULL, " "PRIMARY KEY (id))",
-        )
-
-    def test_primary_key_defaults_to_identity(self):
-        metadata = MetaData()
-        tbl = Table("test", metadata, Column("id", Integer, primary_key=True))
-        self.assert_compile(
-            schema.CreateTable(tbl),
-            "CREATE TABLE test (id INTEGER NOT NULL IDENTITY(1,1), "
-            "PRIMARY KEY (id))",
-        )
-
-    def test_identity_no_primary_key(self):
-        metadata = MetaData()
-        tbl = Table(
-            "test", metadata, Column("id", Integer, autoincrement=True)
-        )
-        self.assert_compile(
-            schema.CreateTable(tbl),
-            "CREATE TABLE test (id INTEGER NOT NULL IDENTITY(1,1)" ")",
-        )
-
-    def test_identity_separate_from_primary_key(self):
-        metadata = MetaData()
-        tbl = Table(
-            "test",
-            metadata,
-            Column("id", Integer, autoincrement=False, primary_key=True),
-            Column("x", Integer, autoincrement=True),
-        )
-        self.assert_compile(
-            schema.CreateTable(tbl),
-            "CREATE TABLE test (id INTEGER NOT NULL, "
-            "x INTEGER NOT NULL IDENTITY(1,1), "
-            "PRIMARY KEY (id))",
-        )
-
-    def test_identity_illegal_two_autoincrements(self):
-        metadata = MetaData()
-        tbl = Table(
-            "test",
-            metadata,
-            Column("id", Integer, autoincrement=True),
-            Column("id2", Integer, autoincrement=True),
-        )
-        # this will be rejected by the database, just asserting this is what
-        # the two autoincrements will do right now
-        self.assert_compile(
-            schema.CreateTable(tbl),
-            "CREATE TABLE test (id INTEGER NOT NULL IDENTITY(1,1), "
-            "id2 INTEGER NOT NULL IDENTITY(1,1))",
-        )
-
-    def test_identity_start_0(self):
-        metadata = MetaData()
-        tbl = Table(
-            "test",
-            metadata,
-            Column("id", Integer, mssql_identity_start=0, primary_key=True),
-        )
-        self.assert_compile(
-            schema.CreateTable(tbl),
-            "CREATE TABLE test (id INTEGER NOT NULL IDENTITY(0,1), "
-            "PRIMARY KEY (id))",
-        )
-
-    def test_identity_increment_5(self):
-        metadata = MetaData()
-        tbl = Table(
-            "test",
-            metadata,
-            Column(
-                "id", Integer, mssql_identity_increment=5, primary_key=True
-            ),
-        )
-        self.assert_compile(
-            schema.CreateTable(tbl),
-            "CREATE TABLE test (id INTEGER NOT NULL IDENTITY(1,5), "
-            "PRIMARY KEY (id))",
-        )
-
     def test_table_pkc_clustering(self):
         metadata = MetaData()
         tbl = Table(
@@ -1388,6 +1301,240 @@ class CompileTest(fixtures.TestBase, AssertsCompiledSQL):
         )
 
 
+class CompileIdentityTest(fixtures.TestBase, AssertsCompiledSQL):
+    __dialect__ = mssql.dialect()
+
+    def assert_compile_with_warning(self, *args, **kwargs):
+        with testing.expect_deprecated(
+            "The dialect options 'mssql_identity_start' and "
+            "'mssql_identity_increment' are deprecated. "
+            "Use the 'Identity' object instead."
+        ):
+            return self.assert_compile(*args, **kwargs)
+
+    def test_primary_key_no_identity(self):
+        metadata = MetaData()
+        tbl = Table(
+            "test",
+            metadata,
+            Column("id", Integer, autoincrement=False, primary_key=True),
+        )
+        self.assert_compile(
+            schema.CreateTable(tbl),
+            "CREATE TABLE test (id INTEGER NOT NULL, PRIMARY KEY (id))",
+        )
+
+    def test_primary_key_defaults_to_identity(self):
+        metadata = MetaData()
+        tbl = Table("test", metadata, Column("id", Integer, primary_key=True))
+        self.assert_compile(
+            schema.CreateTable(tbl),
+            "CREATE TABLE test (id INTEGER NOT NULL IDENTITY, "
+            "PRIMARY KEY (id))",
+        )
+
+    def test_primary_key_with_identity_object(self):
+        metadata = MetaData()
+        tbl = Table(
+            "test",
+            metadata,
+            Column(
+                "id",
+                Integer,
+                Identity(start=3, increment=42),
+                primary_key=True,
+            ),
+        )
+        self.assert_compile(
+            schema.CreateTable(tbl),
+            "CREATE TABLE test (id INTEGER NOT NULL IDENTITY(3,42), "
+            "PRIMARY KEY (id))",
+        )
+
+    def test_identity_no_primary_key(self):
+        metadata = MetaData()
+        tbl = Table(
+            "test", metadata, Column("id", Integer, autoincrement=True)
+        )
+        self.assert_compile(
+            schema.CreateTable(tbl),
+            "CREATE TABLE test (id INTEGER NOT NULL IDENTITY)",
+        )
+
+    def test_identity_object_no_primary_key(self):
+        metadata = MetaData()
+        tbl = Table(
+            "test", metadata, Column("id", Integer, Identity(increment=42)),
+        )
+        self.assert_compile(
+            schema.CreateTable(tbl),
+            "CREATE TABLE test (id INTEGER NOT NULL IDENTITY(1,42))",
+        )
+
+    def test_identity_object_1_1(self):
+        metadata = MetaData()
+        tbl = Table(
+            "test",
+            metadata,
+            Column("id", Integer, Identity(start=1, increment=1)),
+        )
+        self.assert_compile(
+            schema.CreateTable(tbl),
+            "CREATE TABLE test (id INTEGER NOT NULL IDENTITY(1,1))",
+        )
+
+    def test_identity_object_no_primary_key_non_nullable(self):
+        metadata = MetaData()
+        tbl = Table(
+            "test",
+            metadata,
+            Column("id", Integer, Identity(start=3), nullable=False,),
+        )
+        self.assert_compile(
+            schema.CreateTable(tbl),
+            "CREATE TABLE test (id INTEGER NOT NULL IDENTITY(3,1)" ")",
+        )
+
+    def test_identity_separate_from_primary_key(self):
+        metadata = MetaData()
+        tbl = Table(
+            "test",
+            metadata,
+            Column("id", Integer, autoincrement=False, primary_key=True),
+            Column("x", Integer, autoincrement=True),
+        )
+        self.assert_compile(
+            schema.CreateTable(tbl),
+            "CREATE TABLE test (id INTEGER NOT NULL, "
+            "x INTEGER NOT NULL IDENTITY, "
+            "PRIMARY KEY (id))",
+        )
+
+    def test_identity_object_separate_from_primary_key(self):
+        metadata = MetaData()
+        tbl = Table(
+            "test",
+            metadata,
+            Column("id", Integer, autoincrement=False, primary_key=True),
+            Column("x", Integer, Identity(start=3, increment=42),),
+        )
+        self.assert_compile(
+            schema.CreateTable(tbl),
+            "CREATE TABLE test (id INTEGER NOT NULL, "
+            "x INTEGER NOT NULL IDENTITY(3,42), "
+            "PRIMARY KEY (id))",
+        )
+
+    def test_identity_illegal_two_autoincrements(self):
+        metadata = MetaData()
+        tbl = Table(
+            "test",
+            metadata,
+            Column("id", Integer, autoincrement=True),
+            Column("id2", Integer, autoincrement=True),
+        )
+        # this will be rejected by the database, just asserting this is what
+        # the two autoincrements will do right now
+        self.assert_compile(
+            schema.CreateTable(tbl),
+            "CREATE TABLE test (id INTEGER NOT NULL IDENTITY, "
+            "id2 INTEGER NOT NULL IDENTITY)",
+        )
+
+    def test_identity_object_illegal_two_autoincrements(self):
+        metadata = MetaData()
+        tbl = Table(
+            "test",
+            metadata,
+            Column(
+                "id",
+                Integer,
+                Identity(start=3, increment=42),
+                autoincrement=True,
+            ),
+            Column("id2", Integer, Identity(start=7, increment=2),),
+        )
+        # this will be rejected by the database, just asserting this is what
+        # the two autoincrements will do right now
+        self.assert_compile(
+            schema.CreateTable(tbl),
+            "CREATE TABLE test (id INTEGER NOT NULL IDENTITY(3,42), "
+            "id2 INTEGER NOT NULL IDENTITY(7,2))",
+        )
+
+    def test_identity_start_0(self):
+        metadata = MetaData()
+        tbl = Table(
+            "test",
+            metadata,
+            Column("id", Integer, mssql_identity_start=0, primary_key=True),
+        )
+        self.assert_compile_with_warning(
+            schema.CreateTable(tbl),
+            "CREATE TABLE test (id INTEGER NOT NULL IDENTITY(0,1), "
+            "PRIMARY KEY (id))",
+        )
+
+    def test_identity_increment_5(self):
+        metadata = MetaData()
+        tbl = Table(
+            "test",
+            metadata,
+            Column(
+                "id", Integer, mssql_identity_increment=5, primary_key=True
+            ),
+        )
+        self.assert_compile_with_warning(
+            schema.CreateTable(tbl),
+            "CREATE TABLE test (id INTEGER NOT NULL IDENTITY(1,5), "
+            "PRIMARY KEY (id))",
+        )
+
+    @testing.combinations(
+        schema.CreateTable(
+            Table(
+                "test",
+                MetaData(),
+                Column(
+                    "id",
+                    Integer,
+                    Identity(start=2, increment=2),
+                    mssql_identity_start=0,
+                ),
+            )
+        ),
+        schema.CreateTable(
+            Table(
+                "test1",
+                MetaData(),
+                Column(
+                    "id2",
+                    Integer,
+                    Identity(start=3, increment=3),
+                    mssql_identity_increment=5,
+                ),
+            )
+        ),
+    )
+    def test_identity_options_ignored_with_identity_object(self, create_table):
+        assert_raises_message(
+            exc.CompileError,
+            "Cannot specify options 'mssql_identity_start' and/or "
+            "'mssql_identity_increment' while also using the "
+            "'Identity' construct.",
+            create_table.compile,
+            dialect=self.__dialect__,
+        )
+
+    def test_identity_object_no_options(self):
+        metadata = MetaData()
+        tbl = Table("test", metadata, Column("id", Integer, Identity()),)
+        self.assert_compile(
+            schema.CreateTable(tbl),
+            "CREATE TABLE test (id INTEGER NOT NULL IDENTITY)",
+        )
+
+
 class SchemaTest(fixtures.TestBase):
     def setup(self):
         t = Table(
diff --git a/test/dialect/mssql/test_query.py b/test/dialect/mssql/test_query.py
index 1e73b8b49..d8f2a4a0e 100644
--- a/test/dialect/mssql/test_query.py
+++ b/test/dialect/mssql/test_query.py
@@ -6,6 +6,7 @@ from sqlalchemy import desc
 from sqlalchemy import event
 from sqlalchemy import ForeignKey
 from sqlalchemy import func
+from sqlalchemy import Identity
 from sqlalchemy import Integer
 from sqlalchemy import literal
 from sqlalchemy import MetaData
@@ -309,7 +310,7 @@ class QueryTest(testing.AssertsExecutionResults, fixtures.TestBase):
         t1 = Table(
             "t1",
             meta,
-            Column("id", Integer, mssql_identity_start=100, primary_key=True),
+            Column("id", Integer, Identity(start=100), primary_key=True),
             Column("descr", String(200)),
             # the following flag will prevent the
             # MSSQLCompiler.returning_clause from getting called,
@@ -321,7 +322,7 @@ class QueryTest(testing.AssertsExecutionResults, fixtures.TestBase):
         t2 = Table(
             "t2",
             meta,
-            Column("id", Integer, mssql_identity_start=200, primary_key=True),
+            Column("id", Integer, Identity(start=200), primary_key=True),
             Column("descr", String(200)),
         )
 
diff --git a/test/dialect/mssql/test_reflection.py b/test/dialect/mssql/test_reflection.py
index 176d3d2ec..6e4038eb4 100644
--- a/test/dialect/mssql/test_reflection.py
+++ b/test/dialect/mssql/test_reflection.py
@@ -133,7 +133,10 @@ class ReflectionTest(fixtures.TestBase, ComparesTables, AssertsCompiledSQL):
                 primary_key=True,
             ),
         )
-        table.create()
+        with testing.expect_deprecated(
+            "The dialect options 'mssql_identity_start' and"
+        ):
+            table.create()
 
         meta2 = MetaData(testing.db)
         table2 = Table("identity_test", meta2, autoload=True)
diff --git a/test/dialect/oracle/test_compiler.py b/test/dialect/oracle/test_compiler.py
index 21a3b04ba..97a204630 100644
--- a/test/dialect/oracle/test_compiler.py
+++ b/test/dialect/oracle/test_compiler.py
@@ -6,6 +6,7 @@ from sqlalchemy import exc
 from sqlalchemy import except_
 from sqlalchemy import ForeignKey
 from sqlalchemy import func
+from sqlalchemy import Identity
 from sqlalchemy import Index
 from sqlalchemy import Integer
 from sqlalchemy import literal
@@ -1249,6 +1250,44 @@ class CompileTest(fixtures.TestBase, AssertsCompiledSQL):
             dialect=oracle.dialect(),
         )
 
+    def test_column_identity(self):
+        # all other tests are in test_identity_column.py
+        m = MetaData()
+        t = Table(
+            "t",
+            m,
+            Column("y", Integer, Identity(always=True, start=4, increment=7)),
+        )
+        self.assert_compile(
+            schema.CreateTable(t),
+            "CREATE TABLE t (y INTEGER GENERATED ALWAYS AS IDENTITY "
+            "(INCREMENT BY 7 START WITH 4))",
+        )
+
+    def test_column_identity_no_generated(self):
+        m = MetaData()
+        t = Table("t", m, Column("y", Integer, Identity(always=None)))
+        self.assert_compile(
+            schema.CreateTable(t),
+            "CREATE TABLE t (y INTEGER GENERATED  AS IDENTITY)",
+        )
+
+    @testing.combinations(
+        (True, True, "ALWAYS ON NULL"),  # this would error when executed
+        (False, None, "BY DEFAULT"),
+        (False, False, "BY DEFAULT"),
+        (False, True, "BY DEFAULT ON NULL"),
+    )
+    def test_column_identity_on_null(self, always, on_null, text):
+        m = MetaData()
+        t = Table(
+            "t", m, Column("y", Integer, Identity(always, on_null=on_null))
+        )
+        self.assert_compile(
+            schema.CreateTable(t),
+            "CREATE TABLE t (y INTEGER GENERATED %s AS IDENTITY)" % text,
+        )
+
 
 class SequenceTest(fixtures.TestBase, AssertsCompiledSQL):
     def test_basic(self):
diff --git a/test/dialect/postgresql/test_compiler.py b/test/dialect/postgresql/test_compiler.py
index ce285007f..aa355549b 100644
--- a/test/dialect/postgresql/test_compiler.py
+++ b/test/dialect/postgresql/test_compiler.py
@@ -8,6 +8,7 @@ from sqlalchemy import delete
 from sqlalchemy import Enum
 from sqlalchemy import exc
 from sqlalchemy import func
+from sqlalchemy import Identity
 from sqlalchemy import Index
 from sqlalchemy import Integer
 from sqlalchemy import MetaData
@@ -1689,6 +1690,20 @@ class CompileTest(fixtures.TestBase, AssertsCompiledSQL):
             dialect=postgresql.dialect(),
         )
 
+    def test_column_identity(self):
+        # all other tests are in test_identity_column.py
+        m = MetaData()
+        t = Table(
+            "t",
+            m,
+            Column("y", Integer, Identity(always=True, start=4, increment=7)),
+        )
+        self.assert_compile(
+            schema.CreateTable(t),
+            "CREATE TABLE t (y INTEGER GENERATED ALWAYS AS IDENTITY "
+            "(INCREMENT BY 7 START WITH 4))",
+        )
+
 
 class InsertOnConflictTest(fixtures.TestBase, AssertsCompiledSQL):
     __dialect__ = postgresql.dialect()