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

25 files changed, 456 insertions, 224 deletions

diff --git a/doc/build/changelog/migration_14.rst b/doc/build/changelog/migration_14.rst
index 9c7d5fa55..6702f86f0 100644
--- a/doc/build/changelog/migration_14.rst
+++ b/doc/build/changelog/migration_14.rst
@@ -1599,4 +1599,31 @@ was not created.
 
     :ref:`defaults_sequences`
 
-:ticket:`4976`
\ No newline at end of file
+:ticket:`4976`
+
+
+.. _change_4235:
+
+Added Sequence support distinct from IDENTITY to SQL Server
+-----------------------------------------------------------
+
+The :class:`.Sequence` construct is now fully functional with Microsoft
+SQL Server.  When applied to a :class:`.Column`, the DDL for the table will
+no longer include IDENTITY keywords and instead will rely upon "CREATE SEQUENCE"
+to ensure a sequence is present which will then be used for INSERT statements
+on the table.
+
+The :class:`.Sequence` prior to version 1.3 was used to control parameters for
+the IDENTITY column in SQL Server; this usage emitted deprecation warnings
+throughout 1.3 and is now removed in 1.4.  For control of paramters for an
+IDENTITY column, the ``mssql_identity_start`` and ``mssql_identity_increment``
+parameters should be used; see the MSSQL dialect documentation linked below.
+
+
+.. seealso::
+
+    :ref:`mssql_identity`
+
+:ticket:`4235`
+
+:ticket:`4633`
diff --git a/doc/build/changelog/unreleased_14/4235.rst b/doc/build/changelog/unreleased_14/4235.rst
new file mode 100644
index 000000000..acd1ff1ce
--- /dev/null
+++ b/doc/build/changelog/unreleased_14/4235.rst
@@ -0,0 +1,18 @@
+.. change::
+    :tags: sql, mssql
+    :tickets: 4235, 4633
+
+    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, BIGINT, and DECIMAL(n, 0).   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.
+
+    .. seealso::
+
+        :ref:`change_4235`
\ No newline at end of file
diff --git a/lib/sqlalchemy/dialects/mssql/base.py b/lib/sqlalchemy/dialects/mssql/base.py
index 5e0704597..9a38a7881 100644
--- a/lib/sqlalchemy/dialects/mssql/base.py
+++ b/lib/sqlalchemy/dialects/mssql/base.py
@@ -22,7 +22,8 @@ behavior for an integer primary key column, described at
 the first
 integer primary key column in a :class:`_schema.Table`
 will be considered to be the
-identity column and will generate DDL as such::
+identity column - unless it is associated with a :class:`.Sequence` - and will
+generate DDL as such::
 
     from sqlalchemy import Table, MetaData, Column, Integer
 
@@ -76,6 +77,10 @@ is set to ``False`` on any integer primary key column::
    the ``mssql_identity_start`` and ``mssql_identity_increment`` parameters
    documented at :ref:`mssql_identity`.
 
+.. versionchanged::  1.4   Removed the ability to use a :class:`.Sequence`
+   object to modify IDENTITY characteristics. :class:`.Sequence` objects
+   now only manipulate true T-SQL SEQUENCE types.
+
 .. note::
 
     There can only be one IDENTITY column on the table.  When using
@@ -216,6 +221,17 @@ how SQLAlchemy handles this:
 This
 is an auxiliary use case suitable for testing and bulk insert scenarios.
 
+SEQUENCE support
+----------------
+
+The :class:`.Sequence` object now creates "real" sequences, i.e.,
+``CREATE SEQUENCE``. To provide compatibility with other dialects,
+:class:`.Sequence` defaults to a data type of Integer and a start value of 1,
+even though the T-SQL defaults are BIGINT and -9223372036854775808,
+respectively.
+
+.. versionadded:: 1.4.0
+
 MAX on VARCHAR / NVARCHAR
 -------------------------
 
@@ -701,6 +717,7 @@ import re
 from . import information_schema as ischema
 from ... import exc
 from ... import schema as sa_schema
+from ... import Sequence
 from ... import sql
 from ... import types as sqltypes
 from ... import util
@@ -713,6 +730,7 @@ from ...sql import expression
 from ...sql import func
 from ...sql import quoted_name
 from ...sql import util as sql_util
+from ...sql.type_api import to_instance
 from ...types import BIGINT
 from ...types import BINARY
 from ...types import CHAR
@@ -1465,18 +1483,20 @@ class MSExecutionContext(default.DefaultExecutionContext):
 
         if self.isinsert:
             tbl = self.compiled.statement.table
-            seq_column = tbl._autoincrement_column
-            insert_has_sequence = seq_column is not None
+            id_column = tbl._autoincrement_column
+            insert_has_identity = (id_column is not None) and (
+                not isinstance(id_column.default, Sequence)
+            )
 
-            if insert_has_sequence:
+            if insert_has_identity:
                 compile_state = self.compiled.compile_state
                 self._enable_identity_insert = (
-                    seq_column.key in self.compiled_parameters[0]
+                    id_column.key in self.compiled_parameters[0]
                 ) or (
                     compile_state._dict_parameters
                     and (
-                        seq_column.key in compile_state._dict_parameters
-                        or seq_column in compile_state._dict_parameters
+                        id_column.key in compile_state._dict_parameters
+                        or id_column in compile_state._dict_parameters
                     )
                 )
 
@@ -1485,7 +1505,7 @@ class MSExecutionContext(default.DefaultExecutionContext):
 
             self._select_lastrowid = (
                 not self.compiled.inline
-                and insert_has_sequence
+                and insert_has_identity
                 and not self.compiled.returning
                 and not self._enable_identity_insert
                 and not self.executemany
@@ -1570,6 +1590,23 @@ class MSExecutionContext(default.DefaultExecutionContext):
             except Exception:
                 pass
 
+    def get_result_cursor_strategy(self, result):
+        if self._result_strategy:
+            return self._result_strategy
+        else:
+            return super(MSExecutionContext, self).get_result_cursor_strategy(
+                result
+            )
+
+    def fire_sequence(self, seq, type_):
+        return self._execute_scalar(
+            (
+                "SELECT NEXT VALUE FOR %s"
+                % self.dialect.identifier_preparer.format_sequence(seq)
+            ),
+            type_,
+        )
+
 
 class MSSQLCompiler(compiler.SQLCompiler):
     returning_precedes_values = True
@@ -1972,6 +2009,9 @@ class MSSQLCompiler(compiler.SQLCompiler):
             self.process(binary.right),
         )
 
+    def visit_sequence(self, seq, **kw):
+        return "NEXT VALUE FOR %s" % self.preparer.format_sequence(seq)
+
 
 class MSSQLStrictCompiler(MSSQLCompiler):
 
@@ -2050,41 +2090,16 @@ class MSDDLCompiler(compiler.DDLCompiler):
                 "in order to generate DDL"
             )
 
-        # install an IDENTITY Sequence if we either a sequence or an implicit
-        # IDENTITY column
-        if isinstance(column.default, sa_schema.Sequence):
-
-            if (
-                column.default.start is not None
-                or column.default.increment is not None
-                or column is not column.table._autoincrement_column
-            ):
-                util.warn_deprecated(
-                    "Use of Sequence with SQL Server in order to affect the "
-                    "parameters of the IDENTITY value is deprecated, as "
-                    "Sequence "
-                    "will correspond to an actual SQL Server "
-                    "CREATE SEQUENCE in "
-                    "a future release.  Please use the mssql_identity_start "
-                    "and mssql_identity_increment parameters.",
-                    version="1.3",
-                )
-            if column.default.start == 0:
-                start = 0
-            else:
-                start = column.default.start or 1
-
-            colspec += " IDENTITY(%s,%s)" % (
-                start,
-                column.default.increment or 1,
-            )
-        elif (
+        if (
             column is column.table._autoincrement_column
             or column.autoincrement is True
         ):
-            start = column.dialect_options["mssql"]["identity_start"]
-            increment = column.dialect_options["mssql"]["identity_increment"]
-            colspec += " IDENTITY(%s,%s)" % (start, increment)
+            if not isinstance(column.default, Sequence):
+                start = column.dialect_options["mssql"]["identity_start"]
+                increment = column.dialect_options["mssql"][
+                    "identity_increment"
+                ]
+                colspec += " IDENTITY(%s,%s)" % (start, increment)
         else:
             default = self.get_column_default_string(column)
             if default is not None:
@@ -2203,6 +2218,18 @@ class MSDDLCompiler(compiler.DDLCompiler):
             text += " PERSISTED"
         return text
 
+    def visit_create_sequence(self, create, **kw):
+
+        if create.element.data_type is not None:
+            data_type = create.element.data_type
+        else:
+            data_type = to_instance(self.dialect.sequence_default_column_type)
+
+        prefix = " AS %s" % self.type_compiler.process(data_type)
+        return super(MSDDLCompiler, self).visit_create_sequence(
+            create, prefix=prefix, **kw
+        )
+
 
 class MSIdentifierPreparer(compiler.IdentifierPreparer):
     reserved_words = RESERVED_WORDS
@@ -2363,6 +2390,12 @@ class MSDialect(default.DefaultDialect):
 
     ischema_names = ischema_names
 
+    supports_sequences = True
+    # T-SQL's actual default is BIGINT
+    sequence_default_column_type = INTEGER
+    # T-SQL's actual default is -9223372036854775808
+    default_sequence_base = 1
+
     supports_native_boolean = False
     non_native_boolean_check_constraint = False
     supports_unicode_binds = True
@@ -2565,6 +2598,21 @@ class MSDialect(default.DefaultDialect):
 
         return c.first() is not None
 
+    @_db_plus_owner
+    def has_sequence(self, connection, sequencename, dbname, owner, schema):
+        sequences = ischema.sequences
+
+        s = sql.select([sequences.c.sequence_name]).where(
+            sequences.c.sequence_name == sequencename
+        )
+
+        if owner:
+            s = s.where(sequences.c.sequence_schema == owner)
+
+        c = connection.execute(s)
+
+        return c.first() is not None
+
     @reflection.cache
     def get_schema_names(self, connection, **kw):
         s = sql.select(
diff --git a/lib/sqlalchemy/dialects/mssql/information_schema.py b/lib/sqlalchemy/dialects/mssql/information_schema.py
index e9ab6f4f3..6cdde8386 100644
--- a/lib/sqlalchemy/dialects/mssql/information_schema.py
+++ b/lib/sqlalchemy/dialects/mssql/information_schema.py
@@ -173,3 +173,12 @@ computed_columns = Table(
     Column("definition", CoerceUnicode),
     schema="sys",
 )
+
+sequences = Table(
+    "SEQUENCES",
+    ischema,
+    Column("SEQUENCE_CATALOG", CoerceUnicode, key="sequence_catalog"),
+    Column("SEQUENCE_SCHEMA", CoerceUnicode, key="sequence_schema"),
+    Column("SEQUENCE_NAME", CoerceUnicode, key="sequence_name"),
+    schema="INFORMATION_SCHEMA",
+)
diff --git a/lib/sqlalchemy/dialects/postgresql/base.py b/lib/sqlalchemy/dialects/postgresql/base.py
index a85a36bb7..441e77a37 100644
--- a/lib/sqlalchemy/dialects/postgresql/base.py
+++ b/lib/sqlalchemy/dialects/postgresql/base.py
@@ -2446,6 +2446,7 @@ class PGDialect(default.DefaultDialect):
 
     supports_sequences = True
     sequences_optional = True
+    sequence_default_column_type = INTEGER
     preexecute_autoincrement_sequences = True
     postfetch_lastrowid = False
 
diff --git a/lib/sqlalchemy/engine/default.py b/lib/sqlalchemy/engine/default.py
index d0f5cfe96..c682a8ee1 100644
--- a/lib/sqlalchemy/engine/default.py
+++ b/lib/sqlalchemy/engine/default.py
@@ -23,6 +23,7 @@ from . import cursor as _cursor
 from . import interfaces
 from .. import event
 from .. import exc
+from .. import Integer
 from .. import pool
 from .. import processors
 from .. import types as sqltypes
@@ -58,6 +59,8 @@ class DefaultDialect(interfaces.Dialect):
     # not cx_oracle.
     execute_sequence_format = tuple
 
+    sequence_default_column_type = Integer
+
     supports_views = True
     supports_sequences = False
     sequences_optional = False
diff --git a/lib/sqlalchemy/sql/compiler.py b/lib/sqlalchemy/sql/compiler.py
index fc66ca517..4bd19e04b 100644
--- a/lib/sqlalchemy/sql/compiler.py
+++ b/lib/sqlalchemy/sql/compiler.py
@@ -3743,14 +3743,17 @@ class DDLCompiler(Compiled):
             drop.element, use_table=True
         )
 
-    def visit_create_sequence(self, create, **kw):
+    def visit_create_sequence(self, create, prefix=None, **kw):
         text = "CREATE SEQUENCE %s" % self.preparer.format_sequence(
             create.element
         )
+        if prefix:
+            text += prefix
         if create.element.increment is not None:
             text += " INCREMENT BY %d" % create.element.increment
-        if create.element.start is not None:
-            text += " START WITH %d" % create.element.start
+        if create.element.start is None:
+            create.element.start = self.dialect.default_sequence_base
+        text += " START WITH %d" % create.element.start
         if create.element.minvalue is not None:
             text += " MINVALUE %d" % create.element.minvalue
         if create.element.maxvalue is not None:
diff --git a/lib/sqlalchemy/sql/schema.py b/lib/sqlalchemy/sql/schema.py
index 263f579de..ee411174c 100644
--- a/lib/sqlalchemy/sql/schema.py
+++ b/lib/sqlalchemy/sql/schema.py
@@ -49,6 +49,7 @@ from .elements import ColumnElement
 from .elements import quoted_name
 from .elements import TextClause
 from .selectable import TableClause
+from .type_api import to_instance
 from .visitors import InternalTraversal
 from .. import event
 from .. import exc
@@ -2312,6 +2313,7 @@ class IdentityOptions(object):
         cycle=None,
         cache=None,
         order=None,
+        data_type=None,
     ):
         """Construct a :class:`.IdentityOptions` object.
 
@@ -2330,7 +2332,8 @@ class IdentityOptions(object):
          sequence which are calculated in advance.
         :param order: optional boolean value; if true, renders the
          ORDER keyword.
-         name.
+        :param data_type: The type to be returned by the sequence.
+
         """
         self.start = start
         self.increment = increment
@@ -2341,6 +2344,10 @@ class IdentityOptions(object):
         self.cycle = cycle
         self.cache = cache
         self.order = order
+        if data_type is not None:
+            self.data_type = to_instance(data_type)
+        else:
+            self.data_type = None
 
 
 class Sequence(IdentityOptions, roles.StatementRole, DefaultGenerator):
@@ -2393,6 +2400,7 @@ class Sequence(IdentityOptions, roles.StatementRole, DefaultGenerator):
         schema=None,
         cache=None,
         order=None,
+        data_type=None,
         optional=False,
         quote=None,
         metadata=None,
@@ -2402,6 +2410,7 @@ class Sequence(IdentityOptions, roles.StatementRole, DefaultGenerator):
         """Construct a :class:`.Sequence` object.
 
         :param name: the name of the sequence.
+
         :param start: the starting index of the sequence.  This value is
          used when the CREATE SEQUENCE command is emitted to the database
          as the value of the "START WITH" clause.   If ``None``, the
@@ -2478,6 +2487,12 @@ class Sequence(IdentityOptions, roles.StatementRole, DefaultGenerator):
 
          .. versionadded:: 1.1.12
 
+        :param data_type: The type to be returned by the sequence, for
+         dialects that allow us to choose between INTEGER, BIGINT, etc.
+         (e.g., mssql).
+
+         .. versionadded:: 1.4.0
+
         :param optional: boolean value, when ``True``, indicates that this
          :class:`.Sequence` object only needs to be explicitly generated
          on backends that don't provide another way to generate primary
@@ -2542,6 +2557,7 @@ class Sequence(IdentityOptions, roles.StatementRole, DefaultGenerator):
             cycle=cycle,
             cache=cache,
             order=order,
+            data_type=data_type,
         )
         self.name = quoted_name(name, quote)
         self.optional = optional
diff --git a/lib/sqlalchemy/testing/plugin/plugin_base.py b/lib/sqlalchemy/testing/plugin/plugin_base.py
index bf168efce..b31a4ff3e 100644
--- a/lib/sqlalchemy/testing/plugin/plugin_base.py
+++ b/lib/sqlalchemy/testing/plugin/plugin_base.py
@@ -477,6 +477,8 @@ def _prep_testing_database(options, file_config):
                         )
                     )
 
+            # TODO: need to do a get_sequences and drop them also after tables
+
 
 @post
 def _reverse_topological(options, file_config):
diff --git a/lib/sqlalchemy/testing/requirements.py b/lib/sqlalchemy/testing/requirements.py
index 31011a970..2d51e7c9b 100644
--- a/lib/sqlalchemy/testing/requirements.py
+++ b/lib/sqlalchemy/testing/requirements.py
@@ -19,6 +19,7 @@ import platform
 import sys
 
 from . import exclusions
+from . import fails_on_everything_except
 from .. import util
 
 
@@ -1189,3 +1190,13 @@ class SuiteRequirements(Requirements):
             lambda config: not config.db.dialect.supports_is_distinct_from,
             "driver doesn't support an IS DISTINCT FROM construct",
         )
+
+    @property
+    def emulated_lastrowid_even_with_sequences(self):
+        """"target dialect retrieves cursor.lastrowid or an equivalent
+        after an insert() construct executes, even if the table has a
+        Sequence on it..
+        """
+        return fails_on_everything_except(
+            "mysql", "sqlite+pysqlite", "sqlite+pysqlcipher", "sybase",
+        )
diff --git a/lib/sqlalchemy/testing/suite/test_insert.py b/lib/sqlalchemy/testing/suite/test_insert.py
index 92e38ab20..65741941f 100644
--- a/lib/sqlalchemy/testing/suite/test_insert.py
+++ b/lib/sqlalchemy/testing/suite/test_insert.py
@@ -41,7 +41,9 @@ class LastrowidTest(fixtures.TablesTest):
 
     def _assert_round_trip(self, table, conn):
         row = conn.execute(table.select()).first()
-        eq_(row, (conn.engine.dialect.default_sequence_base, "some data"))
+        eq_(
+            row, (conn.dialect.default_sequence_base, "some data",),
+        )
 
     def test_autoincrement_on_insert(self, connection):
 
@@ -282,7 +284,9 @@ class ReturningTest(fixtures.TablesTest):
 
     def _assert_round_trip(self, table, conn):
         row = conn.execute(table.select()).first()
-        eq_(row, (conn.engine.dialect.default_sequence_base, "some data"))
+        eq_(
+            row, (conn.dialect.default_sequence_base, "some data",),
+        )
 
     @classmethod
     def define_tables(cls, metadata):
diff --git a/lib/sqlalchemy/testing/suite/test_sequence.py b/lib/sqlalchemy/testing/suite/test_sequence.py
index db5582c21..dda447c0d 100644
--- a/lib/sqlalchemy/testing/suite/test_sequence.py
+++ b/lib/sqlalchemy/testing/suite/test_sequence.py
@@ -23,7 +23,7 @@ class SequenceTest(fixtures.TablesTest):
         Table(
             "seq_pk",
             metadata,
-            Column("id", Integer, Sequence("tab_id_seq"), primary_key=True),
+            Column("id", Integer, Sequence("tab_id_seq"), primary_key=True,),
             Column("data", String(50)),
         )
 
@@ -33,7 +33,7 @@ class SequenceTest(fixtures.TablesTest):
             Column(
                 "id",
                 Integer,
-                Sequence("tab_id_seq", optional=True),
+                Sequence("tab_id_seq", data_type=Integer, optional=True),
                 primary_key=True,
             ),
             Column("data", String(50)),
@@ -45,11 +45,11 @@ class SequenceTest(fixtures.TablesTest):
 
     def test_insert_lastrowid(self, connection):
         r = connection.execute(self.tables.seq_pk.insert(), data="some data")
-        eq_(r.inserted_primary_key, [1])
+        eq_(r.inserted_primary_key, [testing.db.dialect.default_sequence_base])
 
     def test_nextval_direct(self, connection):
         r = connection.execute(self.tables.seq_pk.c.id.default)
-        eq_(r, 1)
+        eq_(r, testing.db.dialect.default_sequence_base)
 
     @requirements.sequences_optional
     def test_optional_seq(self, connection):
@@ -60,7 +60,7 @@ class SequenceTest(fixtures.TablesTest):
 
     def _assert_round_trip(self, table, conn):
         row = conn.execute(table.select()).first()
-        eq_(row, (1, "some data"))
+        eq_(row, (testing.db.dialect.default_sequence_base, "some data"))
 
 
 class SequenceCompilerTest(testing.AssertsCompiledSQL, fixtures.TestBase):
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,
             ),
diff --git a/test/engine/test_execute.py b/test/engine/test_execute.py
index cc4bb1417..41f94aa98 100644
--- a/test/engine/test_execute.py
+++ b/test/engine/test_execute.py
@@ -1001,7 +1001,7 @@ class MockStrategyTest(fixtures.TestBase):
             "testtable",
             metadata,
             Column(
-                "pk", Integer, Sequence("testtable_pk_seq"), primary_key=True
+                "pk", Integer, Sequence("testtable_pk_seq"), primary_key=True,
             ),
         )
 
@@ -1862,7 +1862,12 @@ class EngineEventsTest(fixtures.TestBase):
         t = Table(
             "t",
             self.metadata,
-            Column("x", Integer, Sequence("t_id_seq"), primary_key=True),
+            Column(
+                "x",
+                testing.db.dialect.sequence_default_column_type,
+                Sequence("t_id_seq"),
+                primary_key=True,
+            ),
             implicit_returning=False,
         )
         self.metadata.create_all(engine)
diff --git a/test/orm/test_session.py b/test/orm/test_session.py
index fa4ffb0e6..635f5ee1a 100644
--- a/test/orm/test_session.py
+++ b/test/orm/test_session.py
@@ -51,7 +51,7 @@ class ExecutionTest(_fixtures.FixtureTest):
         seq.create(testing.db)
         try:
             sess = create_session(bind=testing.db)
-            eq_(sess.execute(seq), 1)
+            eq_(sess.execute(seq), testing.db.dialect.default_sequence_base)
         finally:
             seq.drop(testing.db)
 
diff --git a/test/requirements.py b/test/requirements.py
index 969b4ea83..ec597109b 100644
--- a/test/requirements.py
+++ b/test/requirements.py
@@ -1630,3 +1630,18 @@ class DefaultRequirements(SuiteRequirements):
     @property
     def supports_for_update_of(self):
         return only_if(lambda config: config.db.dialect.supports_for_update_of)
+
+    @property
+    def sequences_in_other_clauses(self):
+        """sequences allowed in WHERE, GROUP BY, HAVING, etc."""
+        return skip_if(["mssql", "oracle"])
+
+    @property
+    def supports_lastrowid_for_expressions(self):
+        """sequences allowed in WHERE, GROUP BY, HAVING, etc."""
+        return skip_if("mssql")
+
+    @property
+    def supports_sequence_for_autoincrement_column(self):
+        """for mssql, autoincrement means IDENTITY, not sequence"""
+        return skip_if("mssql")
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,),
                 ],
             )