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

6 files changed, 98 insertions, 59 deletions

diff --git a/test/sql/test_compiler.py b/test/sql/test_compiler.py
index 20f31ba1e..df52a62c0 100644
--- a/test/sql/test_compiler.py
+++ b/test/sql/test_compiler.py
@@ -4180,19 +4180,19 @@ class DDLTest(fixtures.TestBase, AssertsCompiledSQL):
 
         self.assert_compile(
             schema.CreateSequence(s1),
-            "CREATE SEQUENCE [SCHEMA__none].s1",
+            "CREATE SEQUENCE [SCHEMA__none].s1 START WITH 1",
             schema_translate_map=schema_translate_map,
         )
 
         self.assert_compile(
             schema.CreateSequence(s2),
-            "CREATE SEQUENCE [SCHEMA_foo].s2",
+            "CREATE SEQUENCE [SCHEMA_foo].s2 START WITH 1",
             schema_translate_map=schema_translate_map,
         )
 
         self.assert_compile(
             schema.CreateSequence(s3),
-            "CREATE SEQUENCE [SCHEMA_bar].s3",
+            "CREATE SEQUENCE [SCHEMA_bar].s3 START WITH 1",
             schema_translate_map=schema_translate_map,
         )
 
diff --git a/test/sql/test_defaults.py b/test/sql/test_defaults.py
index 7352810ae..fa6c4d9a1 100644
--- a/test/sql/test_defaults.py
+++ b/test/sql/test_defaults.py
@@ -998,7 +998,7 @@ class PKIncrementTest(fixtures.TablesTest):
             metadata,
             Column(
                 "id",
-                Integer,
+                testing.db.dialect.sequence_default_column_type,
                 Sequence("ai_id_seq", optional=True),
                 primary_key=True,
             ),
@@ -1036,11 +1036,24 @@ class PKIncrementTest(fixtures.TablesTest):
         self.assert_(last not in ids)
         ids.add(last)
 
-        eq_(ids, set([1, 2, 3, 4]))
+        eq_(
+            ids,
+            set(
+                range(
+                    testing.db.dialect.default_sequence_base,
+                    testing.db.dialect.default_sequence_base + 4,
+                )
+            ),
+        )
 
         eq_(
             list(bind.execute(aitable.select().order_by(aitable.c.id))),
-            [(1, 1, None), (2, None, "row 2"), (3, 3, "row 3"), (4, 4, None)],
+            [
+                (testing.db.dialect.default_sequence_base, 1, None),
+                (testing.db.dialect.default_sequence_base + 1, None, "row 2"),
+                (testing.db.dialect.default_sequence_base + 2, 3, "row 3"),
+                (testing.db.dialect.default_sequence_base + 3, 4, None),
+            ],
         )
 
     def test_autoincrement_autocommit(self):
@@ -1164,6 +1177,7 @@ class AutoIncrementTest(fixtures.TestBase):
         )
         return dataset_no_autoinc
 
+    @testing.skip_if(testing.requires.sequences)
     def test_col_w_optional_sequence_non_autoinc_no_firing(
         self, dataset_no_autoinc, connection
     ):
@@ -1215,7 +1229,7 @@ class SpecialTypePKTest(fixtures.TestBase):
     @classmethod
     def setup_class(cls):
         class MyInteger(TypeDecorator):
-            impl = Integer
+            impl = testing.db.dialect.sequence_default_column_type
 
             def process_bind_param(self, value, dialect):
                 if value is None:
@@ -1248,6 +1262,12 @@ class SpecialTypePKTest(fixtures.TestBase):
             t.create(conn)
             r = conn.execute(t.insert().values(data=5))
 
+            expected_result = "INT_" + str(
+                testing.db.dialect.default_sequence_base
+                if (arg and isinstance(arg[0], Sequence))
+                else 1
+            )
+
             # we don't pre-fetch 'server_default'.
             if "server_default" in kw and (
                 not testing.db.dialect.implicit_returning
@@ -1255,9 +1275,13 @@ class SpecialTypePKTest(fixtures.TestBase):
             ):
                 eq_(r.inserted_primary_key, [None])
             else:
-                eq_(r.inserted_primary_key, ["INT_1"])
+                eq_(
+                    r.inserted_primary_key, [expected_result],
+                )
 
-            eq_(conn.execute(t.select()).first(), ("INT_1", 5))
+            eq_(
+                conn.execute(t.select()).first(), (expected_result, 5),
+            )
 
     def test_plain(self):
         # among other things, tests that autoincrement
diff --git a/test/sql/test_functions.py b/test/sql/test_functions.py
index 317c4677a..4b3555391 100644
--- a/test/sql/test_functions.py
+++ b/test/sql/test_functions.py
@@ -991,7 +991,7 @@ class ExecuteTest(fixtures.TestBase):
             meta,
             Column(
                 "id",
-                Integer,
+                testing.db.dialect.sequence_default_column_type,
                 Sequence("t1idseq", optional=True),
                 primary_key=True,
             ),
@@ -1002,7 +1002,7 @@ class ExecuteTest(fixtures.TestBase):
             meta,
             Column(
                 "id",
-                Integer,
+                testing.db.dialect.sequence_default_column_type,
                 Sequence("t2idseq", optional=True),
                 primary_key=True,
             ),
diff --git a/test/sql/test_insert_exec.py b/test/sql/test_insert_exec.py
index 9f9525f3c..a081766bc 100644
--- a/test/sql/test_insert_exec.py
+++ b/test/sql/test_insert_exec.py
@@ -350,7 +350,12 @@ class TableInsertTest(fixtures.TablesTest):
         Table(
             "foo",
             metadata,
-            Column("id", Integer, Sequence("t_id_seq"), primary_key=True),
+            Column(
+                "id",
+                testing.db.dialect.sequence_default_column_type,
+                Sequence("t_id_seq"),
+                primary_key=True,
+            ),
             Column("data", String(50)),
             Column("x", Integer),
         )
@@ -396,7 +401,7 @@ class TableInsertTest(fixtures.TablesTest):
             t.insert().values(
                 id=func.next_value(Sequence("t_id_seq")), data="data", x=5
             ),
-            (1, "data", 5),
+            (testing.db.dialect.default_sequence_base, "data", 5),
         )
 
     def test_uppercase(self):
@@ -431,8 +436,8 @@ class TableInsertTest(fixtures.TablesTest):
         t = self.tables.foo
         self._test(
             t.insert().values(data="data", x=5),
-            (1, "data", 5),
-            inserted_primary_key=[1],
+            (testing.db.dialect.default_sequence_base, "data", 5),
+            inserted_primary_key=[testing.db.dialect.default_sequence_base],
         )
 
     def test_uppercase_direct_params(self):
@@ -452,9 +457,6 @@ class TableInsertTest(fixtures.TablesTest):
             returning=(1, 5),
         )
 
-    @testing.fails_on(
-        "mssql", "lowercase table doesn't support identity insert disable"
-    )
     def test_direct_params(self):
         t = self._fixture()
         self._test(
@@ -463,27 +465,26 @@ class TableInsertTest(fixtures.TablesTest):
             inserted_primary_key=[],
         )
 
-    @testing.fails_on(
-        "mssql", "lowercase table doesn't support identity insert disable"
-    )
     @testing.requires.returning
     def test_direct_params_returning(self):
         t = self._fixture()
         self._test(
             t.insert().values(id=1, data="data", x=5).returning(t.c.id, t.c.x),
-            (1, "data", 5),
-            returning=(1, 5),
+            (testing.db.dialect.default_sequence_base, "data", 5),
+            returning=(testing.db.dialect.default_sequence_base, 5),
         )
 
+    @testing.requires.emulated_lastrowid_even_with_sequences
     @testing.requires.emulated_lastrowid
     def test_implicit_pk(self):
         t = self._fixture()
         self._test(
             t.insert().values(data="data", x=5),
-            (1, "data", 5),
+            (testing.db.dialect.default_sequence_base, "data", 5),
             inserted_primary_key=[],
         )
 
+    @testing.requires.emulated_lastrowid_even_with_sequences
     @testing.requires.emulated_lastrowid
     def test_implicit_pk_multi_rows(self):
         t = self._fixture()
@@ -497,11 +498,12 @@ class TableInsertTest(fixtures.TablesTest):
             [(1, "d1", 5), (2, "d2", 6), (3, "d3", 7)],
         )
 
+    @testing.requires.emulated_lastrowid_even_with_sequences
     @testing.requires.emulated_lastrowid
     def test_implicit_pk_inline(self):
         t = self._fixture()
         self._test(
             t.insert().inline().values(data="data", x=5),
-            (1, "data", 5),
+            (testing.db.dialect.default_sequence_base, "data", 5),
             inserted_primary_key=[],
         )
diff --git a/test/sql/test_returning.py b/test/sql/test_returning.py
index 5f655db6d..f856c15a4 100644
--- a/test/sql/test_returning.py
+++ b/test/sql/test_returning.py
@@ -234,7 +234,7 @@ class SequenceReturningTest(fixtures.TestBase):
         table = Table(
             "tables",
             meta,
-            Column("id", Integer, seq, primary_key=True),
+            Column("id", Integer, seq, primary_key=True,),
             Column("data", String(50)),
         )
         with testing.db.connect() as conn:
@@ -248,8 +248,11 @@ class SequenceReturningTest(fixtures.TestBase):
         r = connection.execute(
             table.insert().values(data="hi").returning(table.c.id)
         )
-        assert r.first() == (1,)
-        assert connection.execute(seq) == 2
+        eq_(r.first(), tuple([testing.db.dialect.default_sequence_base]))
+        eq_(
+            connection.execute(seq),
+            testing.db.dialect.default_sequence_base + 1,
+        )
 
 
 class KeyReturningTest(fixtures.TestBase, AssertsExecutionResults):
diff --git a/test/sql/test_sequences.py b/test/sql/test_sequences.py
index 1d78c0904..126f852ba 100644
--- a/test/sql/test_sequences.py
+++ b/test/sql/test_sequences.py
@@ -26,7 +26,8 @@ class SequenceDDLTest(fixtures.TestBase, testing.AssertsCompiledSQL):
 
     def test_create_drop_ddl(self):
         self.assert_compile(
-            CreateSequence(Sequence("foo_seq")), "CREATE SEQUENCE foo_seq"
+            CreateSequence(Sequence("foo_seq")),
+            "CREATE SEQUENCE foo_seq START WITH 1",
         )
 
         self.assert_compile(
@@ -36,7 +37,7 @@ class SequenceDDLTest(fixtures.TestBase, testing.AssertsCompiledSQL):
 
         self.assert_compile(
             CreateSequence(Sequence("foo_seq", increment=2)),
-            "CREATE SEQUENCE foo_seq INCREMENT BY 2",
+            "CREATE SEQUENCE foo_seq INCREMENT BY 2 START WITH 1",
         )
 
         self.assert_compile(
@@ -81,12 +82,12 @@ class SequenceDDLTest(fixtures.TestBase, testing.AssertsCompiledSQL):
 
         self.assert_compile(
             CreateSequence(Sequence("foo_seq", cache=1000, order=True)),
-            "CREATE SEQUENCE foo_seq CACHE 1000 ORDER",
+            "CREATE SEQUENCE foo_seq START WITH 1 CACHE 1000 ORDER",
         )
 
         self.assert_compile(
             CreateSequence(Sequence("foo_seq", order=True)),
-            "CREATE SEQUENCE foo_seq ORDER",
+            "CREATE SEQUENCE foo_seq START WITH 1 ORDER",
         )
 
         self.assert_compile(
@@ -111,7 +112,7 @@ class LegacySequenceExecTest(fixtures.TestBase):
         """asserts return of next_value is an int"""
 
         assert isinstance(ret, util.int_types)
-        assert ret > 0
+        assert ret >= testing.db.dialect.default_sequence_base
 
     def test_implicit_connectionless(self):
         s = Sequence("my_sequence", metadata=MetaData(testing.db))
@@ -169,7 +170,7 @@ class SequenceExecTest(fixtures.TestBase):
         """asserts return of next_value is an int"""
 
         assert isinstance(ret, util.int_types)
-        assert ret > 0
+        assert ret >= testing.db.dialect.default_sequence_base
 
     def test_execute(self, connection):
         s = Sequence("my_sequence")
@@ -202,7 +203,7 @@ class SequenceExecTest(fixtures.TestBase):
         s = Sequence("my_sequence")
         self._assert_seq_result(connection.scalar(select([s.next_value()])))
 
-    @testing.fails_on("oracle", "ORA-02287: sequence number not allowed here")
+    @testing.requires.sequences_in_other_clauses
     @testing.provide_metadata
     def test_func_embedded_whereclause(self, connection):
         """test can use next_value() in whereclause"""
@@ -224,43 +225,44 @@ class SequenceExecTest(fixtures.TestBase):
         """test can use next_value() in values() of _ValuesBase"""
 
         metadata = self.metadata
-        t1 = Table("t", metadata, Column("x", Integer))
+        t1 = Table("t", metadata, Column("x", Integer),)
         t1.create(testing.db)
         s = Sequence("my_sequence")
         connection.execute(t1.insert().values(x=s.next_value()))
         self._assert_seq_result(connection.scalar(t1.select()))
 
-    @testing.requires.supports_lastrowid
+    @testing.requires.no_lastrowid_support
     @testing.provide_metadata
-    def test_inserted_pk_no_returning_w_lastrowid(self):
-        """test inserted_primary_key contains the pk when
-        pk_col=next_value(), lastrowid is supported."""
+    def test_inserted_pk_no_returning_no_lastrowid(self):
+        """test inserted_primary_key contains [None] when
+        pk_col=next_value(), implicit returning is not used."""
 
         metadata = self.metadata
         t1 = Table("t", metadata, Column("x", Integer, primary_key=True))
         t1.create(testing.db)
+
         e = engines.testing_engine(options={"implicit_returning": False})
         s = Sequence("my_sequence")
-
         with e.connect() as conn:
             r = conn.execute(t1.insert().values(x=s.next_value()))
-            self._assert_seq_result(r.inserted_primary_key[0])
+            eq_(r.inserted_primary_key, [None])
 
-    @testing.requires.no_lastrowid_support
+    @testing.requires.supports_lastrowid
+    @testing.requires.supports_lastrowid_for_expressions
     @testing.provide_metadata
-    def test_inserted_pk_no_returning_no_lastrowid(self):
-        """test inserted_primary_key contains [None] when
-        pk_col=next_value(), implicit returning is not used."""
+    def test_inserted_pk_no_returning_w_lastrowid(self):
+        """test inserted_primary_key contains the pk when
+        pk_col=next_value(), lastrowid is supported."""
 
         metadata = self.metadata
-        t1 = Table("t", metadata, Column("x", Integer, primary_key=True))
+        t1 = Table("t", metadata, Column("x", Integer, primary_key=True,),)
         t1.create(testing.db)
-
         e = engines.testing_engine(options={"implicit_returning": False})
         s = Sequence("my_sequence")
+
         with e.connect() as conn:
             r = conn.execute(t1.insert().values(x=s.next_value()))
-            eq_(r.inserted_primary_key, [None])
+            self._assert_seq_result(r.inserted_primary_key[0])
 
     @testing.requires.returning
     @testing.provide_metadata
@@ -270,7 +272,7 @@ class SequenceExecTest(fixtures.TestBase):
 
         metadata = self.metadata
         s = Sequence("my_sequence")
-        t1 = Table("t", metadata, Column("x", Integer, primary_key=True))
+        t1 = Table("t", metadata, Column("x", Integer, primary_key=True,),)
         t1.create(testing.db)
 
         e = engines.testing_engine(options={"implicit_returning": True})
@@ -298,7 +300,7 @@ class SequenceTest(fixtures.TestBase, testing.AssertsCompiledSQL):
         try:
             with testing.db.connect() as conn:
                 values = [conn.execute(seq) for i in range(3)]
-                start = seq.start or 1
+                start = seq.start or testing.db.dialect.default_sequence_base
                 inc = seq.increment or 1
                 eq_(values, list(range(start, start + inc * 3, inc)))
 
@@ -380,6 +382,7 @@ class SequenceTest(fixtures.TestBase, testing.AssertsCompiledSQL):
         assert not self._has_sequence(connection, "s2")
 
     @testing.requires.returning
+    @testing.requires.supports_sequence_for_autoincrement_column
     @testing.provide_metadata
     def test_freestanding_sequence_via_autoinc(self, connection):
         t = Table(
@@ -416,7 +419,11 @@ class TableBoundSequenceTest(fixtures.TablesTest):
             "cartitems",
             metadata,
             Column(
-                "cart_id", Integer, Sequence("cart_id_seq"), primary_key=True
+                "cart_id",
+                Integer,
+                Sequence("cart_id_seq"),
+                primary_key=True,
+                autoincrement=False,
             ),
             Column("description", String(40)),
             Column("createdate", sa.DateTime()),
@@ -426,7 +433,7 @@ class TableBoundSequenceTest(fixtures.TablesTest):
         Table(
             "Manager",
             metadata,
-            Column("obj_id", Integer, Sequence("obj_id_seq")),
+            Column("obj_id", Integer, Sequence("obj_id_seq"),),
             Column("name", String(128)),
             Column(
                 "id",
@@ -443,7 +450,8 @@ class TableBoundSequenceTest(fixtures.TablesTest):
         connection.execute(cartitems.insert(), dict(description="there"))
         r = connection.execute(cartitems.insert(), dict(description="lala"))
 
-        eq_(r.inserted_primary_key[0], 3)
+        expected = 2 + testing.db.dialect.default_sequence_base
+        eq_(r.inserted_primary_key[0], expected)
 
         eq_(
             connection.scalar(
@@ -451,7 +459,7 @@ class TableBoundSequenceTest(fixtures.TablesTest):
                     cartitems.c.description == "lala"
                 ),
             ),
-            3,
+            expected,
         )
 
     def test_seq_nonpk(self):
@@ -471,15 +479,17 @@ class TableBoundSequenceTest(fixtures.TablesTest):
             conn.execute(
                 sometable.insert(), [{"name": "name3"}, {"name": "name4"}]
             )
+
+            dsb = testing.db.dialect.default_sequence_base
             eq_(
                 list(
                     conn.execute(sometable.select().order_by(sometable.c.id))
                 ),
                 [
-                    (1, "somename", 1),
-                    (2, "someother", 2),
-                    (3, "name3", 3),
-                    (4, "name4", 4),
+                    (dsb, "somename", dsb,),
+                    (dsb + 1, "someother", dsb + 1,),
+                    (dsb + 2, "name3", dsb + 2,),
+                    (dsb + 3, "name4", dsb + 3,),
                 ],
             )