Add support for "real" sequences in mssql

Added support for "CREATE SEQUENCE" and full :class:`.Sequence` support for
Microsoft SQL Server.  This removes the deprecated feature of using
:class:`.Sequence` objects to manipulate IDENTITY characteristics which
should now be performed using ``mssql_identity_start`` and
``mssql_identity_increment`` as documented at :ref:`mssql_identity`. The
change includes a new parameter :paramref:`.Sequence.data_type` to
accommodate SQL Server's choice of datatype, which for that backend
includes INTEGER and BIGINT.   The default starting value for SQL Server's
version of :class:`.Sequence` has been set at 1; this default is now
emitted within the CREATE SEQUENCE DDL for all backends.

Fixes: #4235
Fixes: #4633
Change-Id: I6aa55c441e8146c2f002e2e201a7f645e667b916

4 files changed, 141 insertions, 110 deletions

diff --git a/test/dialect/mssql/test_compiler.py b/test/dialect/mssql/test_compiler.py
index b7a06c8e3..99308132b 100644
--- a/test/dialect/mssql/test_compiler.py
+++ b/test/dialect/mssql/test_compiler.py
@@ -13,7 +13,6 @@ from sqlalchemy import MetaData
 from sqlalchemy import PrimaryKeyConstraint
 from sqlalchemy import schema
 from sqlalchemy import select
-from sqlalchemy import Sequence
 from sqlalchemy import sql
 from sqlalchemy import String
 from sqlalchemy import Table
@@ -1157,52 +1156,6 @@ class CompileTest(fixtures.TestBase, AssertsCompiledSQL):
             "PRIMARY KEY (id))",
         )
 
-    def test_sequence_start_0(self):
-        metadata = MetaData()
-        tbl = Table(
-            "test",
-            metadata,
-            Column("id", Integer, Sequence("", 0), primary_key=True),
-        )
-        with testing.expect_deprecated(
-            "Use of Sequence with SQL Server in order to affect "
-        ):
-            self.assert_compile(
-                schema.CreateTable(tbl),
-                "CREATE TABLE test (id INTEGER NOT NULL IDENTITY(0,1), "
-                "PRIMARY KEY (id))",
-            )
-
-    def test_sequence_non_primary_key(self):
-        metadata = MetaData()
-        tbl = Table(
-            "test",
-            metadata,
-            Column("id", Integer, Sequence("", start=5), primary_key=False),
-        )
-        with testing.expect_deprecated(
-            "Use of Sequence with SQL Server in order to affect "
-        ):
-            self.assert_compile(
-                schema.CreateTable(tbl),
-                "CREATE TABLE test (id INTEGER NOT NULL IDENTITY(5,1))",
-            )
-
-    def test_sequence_ignore_nullability(self):
-        metadata = MetaData()
-        tbl = Table(
-            "test",
-            metadata,
-            Column("id", Integer, Sequence("", start=5), nullable=True),
-        )
-        with testing.expect_deprecated(
-            "Use of Sequence with SQL Server in order to affect "
-        ):
-            self.assert_compile(
-                schema.CreateTable(tbl),
-                "CREATE TABLE test (id INTEGER NOT NULL IDENTITY(5,1))",
-            )
-
     def test_table_pkc_clustering(self):
         metadata = MetaData()
         tbl = Table(
diff --git a/test/dialect/mssql/test_query.py b/test/dialect/mssql/test_query.py
index 2ad6c161d..77ecef3a3 100644
--- a/test/dialect/mssql/test_query.py
+++ b/test/dialect/mssql/test_query.py
@@ -245,7 +245,7 @@ class QueryUnicodeTest(fixtures.TestBase):
     @testing.requires.mssql_freetds
     @testing.requires.python2
     @testing.provide_metadata
-    def test_convert_unicode(self):
+    def test_convert_unicode(self, connection):
         meta = self.metadata
         t1 = Table(
             "unitest_table",
@@ -253,26 +253,27 @@ class QueryUnicodeTest(fixtures.TestBase):
             Column("id", Integer, primary_key=True),
             Column("descr", mssql.MSText()),
         )
-        meta.create_all()
-        with testing.db.connect() as con:
-            con.execute(
-                ue(
-                    "insert into unitest_table values ('abc \xc3\xa9 def')"
-                ).encode("UTF-8")
+        meta.create_all(connection)
+        connection.execute(
+            ue("insert into unitest_table values ('abc \xc3\xa9 def')").encode(
+                "UTF-8"
             )
-            r = con.execute(t1.select()).first()
-            assert isinstance(r[1], util.text_type), (
-                "%s is %s instead of unicode, working on %s"
-                % (r[1], type(r[1]), meta.bind)
-            )
-            eq_(r[1], util.ue("abc \xc3\xa9 def"))
+        )
+        r = connection.execute(t1.select()).first()
+        assert isinstance(r[1], util.text_type), (
+            "%s is %s instead of unicode, working on %s"
+            % (r[1], type(r[1]), meta.bind)
+        )
+        eq_(r[1], util.ue("abc \xc3\xa9 def"))
 
 
 class QueryTest(testing.AssertsExecutionResults, fixtures.TestBase):
     __only_on__ = "mssql"
     __backend__ = True
 
-    def test_fetchid_trigger(self):
+    @testing.provide_metadata
+    def test_fetchid_trigger(self, connection):
+        # TODO: investigate test hang on mssql when connection fixture is used
         """
         Verify identity return value on inserting to a trigger table.
 
@@ -299,12 +300,12 @@ class QueryTest(testing.AssertsExecutionResults, fixtures.TestBase):
         with the init parameter 'implicit_returning = False'.
         """
 
-        # todo: this same test needs to be tried in a multithreaded context
+        # TODO: this same test needs to be tried in a multithreaded context
         #      with multiple threads inserting to the same table.
-        # todo: check whether this error also occurs with clients other
+        # TODO: check whether this error also occurs with clients other
         #      than the SQL Server Native Client. Maybe an assert_raises
         #      test should be written.
-        meta = MetaData(testing.db)
+        meta = self.metadata
         t1 = Table(
             "t1",
             meta,
@@ -323,24 +324,31 @@ class QueryTest(testing.AssertsExecutionResults, fixtures.TestBase):
             Column("id", Integer, mssql_identity_start=200, primary_key=True),
             Column("descr", String(200)),
         )
-        meta.create_all()
-        con = testing.db.connect()
-        con.exec_driver_sql(
-            """create trigger paj on t1 for insert as
-            insert into t2 (descr) select descr from inserted"""
+
+        event.listen(
+            meta,
+            "after_create",
+            DDL(
+                "create trigger paj on t1 for insert as "
+                "insert into t2 (descr) select descr from inserted"
+            ),
         )
 
-        try:
-            tr = con.begin()
-            r = con.execute(t2.insert(), descr="hello")
-            self.assert_(r.inserted_primary_key == [200])
-            r = con.execute(t1.insert(), descr="hello")
-            self.assert_(r.inserted_primary_key == [100])
+        # this DROP is not actually needed since SQL Server transactional
+        # DDL is reverting it with the connection fixture.  however,
+        # since we can use "if exists" it's safe to have this here in
+        # case things change.
+        event.listen(
+            meta, "before_drop", DDL("""drop trigger if exists paj""")
+        )
 
-        finally:
-            tr.commit()
-            con.exec_driver_sql("""drop trigger paj""")
-            meta.drop_all()
+        # seems to work with all linux drivers + backend.  not sure
+        # if windows drivers / servers have different behavior here.
+        meta.create_all(connection)
+        r = connection.execute(t2.insert(), descr="hello")
+        self.assert_(r.inserted_primary_key == [200])
+        r = connection.execute(t1.insert(), descr="hello")
+        self.assert_(r.inserted_primary_key == [100])
 
     @testing.provide_metadata
     def _test_disable_scope_identity(self):
@@ -430,8 +438,8 @@ class QueryTest(testing.AssertsExecutionResults, fixtures.TestBase):
             options=dict(legacy_schema_aliasing=False)
         )
         meta.bind = eng
-        con = eng.connect()
-        con.exec_driver_sql("create schema paj")
+        conn = eng.connect()
+        conn.exec_driver_sql("create schema paj")
 
         @event.listens_for(meta, "after_drop")
         def cleanup(target, connection, **kw):
@@ -440,17 +448,17 @@ class QueryTest(testing.AssertsExecutionResults, fixtures.TestBase):
         tbl = Table(
             "test", meta, Column("id", Integer, primary_key=True), schema="paj"
         )
-        tbl.create()
-        tbl.insert().execute({"id": 1})
-        eq_(tbl.select().scalar(), 1)
+        tbl.create(conn)
+        conn.execute(tbl.insert(), {"id": 1})
+        eq_(conn.scalar(tbl.select()), 1)
 
     @testing.provide_metadata
     def test_insertid_schema_legacy(self):
         meta = self.metadata
         eng = engines.testing_engine(options=dict(legacy_schema_aliasing=True))
         meta.bind = eng
-        con = eng.connect()
-        con.exec_driver_sql("create schema paj")
+        conn = eng.connect()
+        conn.exec_driver_sql("create schema paj")
 
         @event.listens_for(meta, "after_drop")
         def cleanup(target, connection, **kw):
@@ -464,7 +472,7 @@ class QueryTest(testing.AssertsExecutionResults, fixtures.TestBase):
         eq_(tbl.select().scalar(), 1)
 
     @testing.provide_metadata
-    def test_returning_no_autoinc(self):
+    def test_returning_no_autoinc(self, connection):
         meta = self.metadata
         table = Table(
             "t1",
@@ -472,12 +480,11 @@ class QueryTest(testing.AssertsExecutionResults, fixtures.TestBase):
             Column("id", Integer, primary_key=True),
             Column("data", String(50)),
         )
-        table.create()
-        result = (
+        table.create(connection)
+        result = connection.execute(
             table.insert()
             .values(id=1, data=func.lower("SomeString"))
             .returning(table.c.id, table.c.data)
-            .execute()
         )
         eq_(result.fetchall(), [(1, "somestring")])
 
@@ -488,8 +495,8 @@ class QueryTest(testing.AssertsExecutionResults, fixtures.TestBase):
             options=dict(legacy_schema_aliasing=False)
         )
         meta.bind = eng
-        con = eng.connect()
-        con.exec_driver_sql("create schema paj")
+        conn = eng.connect()
+        conn.exec_driver_sql("create schema paj")
 
         @event.listens_for(meta, "after_drop")
         def cleanup(target, connection, **kw):
@@ -498,19 +505,19 @@ class QueryTest(testing.AssertsExecutionResults, fixtures.TestBase):
         tbl = Table(
             "test", meta, Column("id", Integer, primary_key=True), schema="paj"
         )
-        tbl.create()
-        tbl.insert().execute({"id": 1})
-        eq_(tbl.select().scalar(), 1)
-        tbl.delete(tbl.c.id == 1).execute()
-        eq_(tbl.select().scalar(), None)
+        tbl.create(conn)
+        conn.execute(tbl.insert(), {"id": 1})
+        eq_(conn.scalar(tbl.select()), 1)
+        conn.execute(tbl.delete(tbl.c.id == 1))
+        eq_(conn.scalar(tbl.select()), None)
 
     @testing.provide_metadata
     def test_delete_schema_legacy(self):
         meta = self.metadata
         eng = engines.testing_engine(options=dict(legacy_schema_aliasing=True))
         meta.bind = eng
-        con = eng.connect()
-        con.exec_driver_sql("create schema paj")
+        conn = eng.connect()
+        conn.exec_driver_sql("create schema paj")
 
         @event.listens_for(meta, "after_drop")
         def cleanup(target, connection, **kw):
@@ -519,20 +526,20 @@ class QueryTest(testing.AssertsExecutionResults, fixtures.TestBase):
         tbl = Table(
             "test", meta, Column("id", Integer, primary_key=True), schema="paj"
         )
-        tbl.create()
-        tbl.insert().execute({"id": 1})
-        eq_(tbl.select().scalar(), 1)
-        tbl.delete(tbl.c.id == 1).execute()
-        eq_(tbl.select().scalar(), None)
+        tbl.create(conn)
+        conn.execute(tbl.insert(), {"id": 1})
+        eq_(conn.scalar(tbl.select()), 1)
+        conn.execute(tbl.delete(tbl.c.id == 1))
+        eq_(conn.scalar(tbl.select()), None)
 
     @testing.provide_metadata
-    def test_insertid_reserved(self):
+    def test_insertid_reserved(self, connection):
         meta = self.metadata
         table = Table("select", meta, Column("col", Integer, primary_key=True))
-        table.create()
+        table.create(connection)
 
-        table.insert().execute(col=7)
-        eq_(table.select().scalar(), 7)
+        connection.execute(table.insert(), {"col": 7})
+        eq_(connection.scalar(table.select()), 7)
 
 
 class Foo(object):
diff --git a/test/dialect/mssql/test_sequence.py b/test/dialect/mssql/test_sequence.py
new file mode 100644
index 000000000..61f315fa2
--- /dev/null
+++ b/test/dialect/mssql/test_sequence.py
@@ -0,0 +1,70 @@
+from decimal import Decimal
+
+from sqlalchemy import BIGINT
+from sqlalchemy import Column
+from sqlalchemy import DECIMAL
+from sqlalchemy import Integer
+from sqlalchemy import Sequence
+from sqlalchemy import String
+from sqlalchemy import Table
+from sqlalchemy.future import select
+from sqlalchemy.testing import eq_
+from sqlalchemy.testing import fixtures
+
+
+class SequenceTest(fixtures.TablesTest):
+    __only_on__ = "mssql"
+    __backend__ = True
+
+    @classmethod
+    def define_tables(cls, metadata):
+        Table(
+            "int_seq_t",
+            metadata,
+            Column("id", Integer, default=Sequence("int_seq")),
+            Column("txt", String(50)),
+        )
+
+        Table(
+            "bigint_seq_t",
+            metadata,
+            Column(
+                "id",
+                BIGINT,
+                default=Sequence(
+                    "bigint_seq", data_type=BIGINT, start=3000000000
+                ),
+            ),
+            Column("txt", String(50)),
+        )
+
+        Table(
+            "decimal_seq_t",
+            metadata,
+            Column(
+                "id",
+                DECIMAL(10, 0),
+                default=Sequence(
+                    "decimal_seq", data_type=DECIMAL(10, 0), start=3000000000,
+                ),
+            ),
+            Column("txt", String(50)),
+        )
+
+    def test_int_seq(self, connection):
+        t = self.tables.int_seq_t
+        connection.execute(t.insert({"txt": "int_seq test"}))
+        result = connection.scalar(select(t.c.id))
+        eq_(result, 1)
+
+    def test_bigint_seq(self, connection):
+        t = self.tables.bigint_seq_t
+        connection.execute(t.insert({"txt": "bigint_seq test"}))
+        result = connection.scalar(select(t.c.id))
+        eq_(result, 3000000000)
+
+    def test_decimal_seq(self, connection):
+        t = self.tables.decimal_seq_t
+        connection.execute(t.insert({"txt": "decimal_seq test"}))
+        result = connection.scalar(select(t.c.id))
+        eq_(result, Decimal("3000000000"))
diff --git a/test/dialect/mssql/test_types.py b/test/dialect/mssql/test_types.py
index f0ddc4564..37b83e911 100644
--- a/test/dialect/mssql/test_types.py
+++ b/test/dialect/mssql/test_types.py
@@ -561,9 +561,10 @@ class TypeRoundTripTest(
             metadata,
             Column(
                 "id",
-                Integer,
+                testing.db.dialect.sequence_default_column_type,
                 Sequence("numeric_id_seq", optional=True),
                 primary_key=True,
+                autoincrement=False,
             ),
             Column(
                 "numericcol", Numeric(precision=38, scale=20, asdecimal=True)
@@ -651,7 +652,7 @@ class TypeRoundTripTest(
             metadata,
             Column(
                 "id",
-                Integer,
+                testing.db.dialect.sequence_default_column_type,
                 Sequence("numeric_id_seq", optional=True),
                 primary_key=True,
             ),