Deprecate create_engine.implicit_returning

The :paramref:`_sa.create_engine.implicit_returning` parameter is
deprecated on the :func:`_sa.create_engine` function only; the parameter
remains available on the :class:`_schema.Table` object. This parameter was
originally intended to enable the "implicit returning" feature of
SQLAlchemy when it was first developed and was not enabled by default.
Under modern use, there's no reason this parameter should be disabled, and
it has been observed to cause confusion as it degrades performance and
makes it more difficult for the ORM to retrieve recently inserted server
defaults. The parameter remains available on :class:`_schema.Table` to
specifically suit database-level edge cases which make RETURNING
infeasible, the sole example currently being SQL Server's limitation that
INSERT RETURNING may not be used on a table that has INSERT triggers on it.

Also removed from the Oracle dialect some logic that would upgrade
an Oracle 8/8i server version to use implicit returning if the
parameter were explictly passed; these versions of Oracle
still support RETURNING so the feature is now enabled for all
Oracle versions.

Fixes: #6962
Change-Id: Ib338e300cd7c8026c3083043f645084a8211aed8

18 files changed, 835 insertions, 736 deletions

diff --git a/doc/build/changelog/unreleased_20/6962.rst b/doc/build/changelog/unreleased_20/6962.rst
new file mode 100644
index 000000000..fcc3c1997
--- /dev/null
+++ b/doc/build/changelog/unreleased_20/6962.rst
@@ -0,0 +1,28 @@
+.. change::
+    :tags: deprecations, engine
+    :tickets: 6962
+
+    The :paramref:`_sa.create_engine.implicit_returning` parameter is
+    deprecated on the :func:`_sa.create_engine` function only; the parameter
+    remains available on the :class:`_schema.Table` object. This parameter was
+    originally intended to enable the "implicit returning" feature of
+    SQLAlchemy when it was first developed and was not enabled by default.
+    Under modern use, there's no reason this parameter should be disabled, and
+    it has been observed to cause confusion as it degrades performance and
+    makes it more difficult for the ORM to retrieve recently inserted server
+    defaults. The parameter remains available on :class:`_schema.Table` to
+    specifically suit database-level edge cases which make RETURNING
+    infeasible, the sole example currently being SQL Server's limitation that
+    INSERT RETURNING may not be used on a table that has INSERT triggers on it.
+
+
+.. change::
+    :tags: bug, oracle
+    :tickets: 6962
+
+    Related to the deprecation for
+    :paramref:`_sa.create_engine.implicit_returning`, the "implicit_returning"
+    feature is now enabled for the Oracle dialect in all cases; previously, the
+    feature would be turned off when an Oracle 8/8i version were detected,
+    however online documentation indicates both versions support the same
+    RETURNING syntax as modern versions.
diff --git a/lib/sqlalchemy/dialects/oracle/base.py b/lib/sqlalchemy/dialects/oracle/base.py
index 9e62b931d..5d9c7beec 100644
--- a/lib/sqlalchemy/dialects/oracle/base.py
+++ b/lib/sqlalchemy/dialects/oracle/base.py
@@ -1443,6 +1443,8 @@ class OracleDialect(default.DefaultDialect):
     supports_unicode_binds = False
     max_identifier_length = 128
 
+    implicit_returning = True
+
     supports_simple_order_by_label = False
     cte_follows_insert = True
 
@@ -1508,9 +1510,11 @@ class OracleDialect(default.DefaultDialect):
     def initialize(self, connection):
         super(OracleDialect, self).initialize(connection)
 
-        self.implicit_returning = self.__dict__.get(
-            "implicit_returning", self.server_version_info > (10,)
-        )
+        # Oracle 8i has RETURNING:
+        # https://docs.oracle.com/cd/A87860_01/doc/index.htm
+
+        # so does Oracle8:
+        # https://docs.oracle.com/cd/A64702_01/doc/index.htm
 
         if self._is_oracle_8:
             self.colspecs = self.colspecs.copy()
diff --git a/lib/sqlalchemy/engine/create.py b/lib/sqlalchemy/engine/create.py
index 6e5a07098..b10e72601 100644
--- a/lib/sqlalchemy/engine/create.py
+++ b/lib/sqlalchemy/engine/create.py
@@ -34,6 +34,11 @@ from ..sql import compiler
         'expressions, or an "empty set" SELECT, at statement execution'
         "time.",
     ),
+    implicit_returning=(
+        "2.0",
+        "The :paramref:`_sa.create_engine.implicit_returning` parameter "
+        "is deprecated and will be removed in a future release. ",
+    ),
 )
 def create_engine(url, **kwargs):
     """Create a new :class:`_engine.Engine` instance.
@@ -350,8 +355,6 @@ def create_engine(url, **kwargs):
 
             :paramref:`_engine.Connection.execution_options.logging_token`
 
-
-
     :param max_identifier_length: integer; override the max_identifier_length
         determined by the dialect.  if ``None`` or zero, has no effect.  This
         is the database's configured maximum number of characters that may be
@@ -411,13 +414,11 @@ def create_engine(url, **kwargs):
        "sqlalchemy.pool" logger. Defaults to a hexstring of the object's
        id.
 
-
        .. seealso::
 
             :ref:`dbengine_logging` - further detail on how to configure
             logging.
 
-
     :param pool_pre_ping: boolean, if True will enable the connection pool
         "pre-ping" feature that tests connections for liveness upon
         each checkout.
diff --git a/lib/sqlalchemy/engine/interfaces.py b/lib/sqlalchemy/engine/interfaces.py
index 38d2c7a57..48b7e35a0 100644
--- a/lib/sqlalchemy/engine/interfaces.py
+++ b/lib/sqlalchemy/engine/interfaces.py
@@ -94,14 +94,6 @@ class Dialect(object):
       in order to get their value.   This is currently oriented towards
       PostgreSQL.
 
-    ``implicit_returning``
-      use RETURNING or equivalent during INSERT execution in order to load
-      newly generated primary keys and other column defaults in one execution,
-      which are then available via inserted_primary_key.
-      If an insert statement has returning() specified explicitly,
-      the "implicit" functionality is not used and inserted_primary_key
-      will not be available.
-
     ``colspecs``
       A dictionary of TypeEngine classes from sqlalchemy.types mapped
       to subclasses that are specific to the dialect class.  This
diff --git a/lib/sqlalchemy/testing/assertsql.py b/lib/sqlalchemy/testing/assertsql.py
index ca0bc6726..9816c99b7 100644
--- a/lib/sqlalchemy/testing/assertsql.py
+++ b/lib/sqlalchemy/testing/assertsql.py
@@ -186,7 +186,9 @@ class CompiledSQL(SQLMatchRule):
             self.is_consumed = True
             self.errormessage = None
         else:
-            self.errormessage = self._failure_message(params) % {
+            self.errormessage = self._failure_message(
+                execute_observed, params
+            ) % {
                 "received_statement": _received_statement,
                 "received_parameters": _received_parameters,
             }
@@ -203,7 +205,7 @@ class CompiledSQL(SQLMatchRule):
         else:
             return None
 
-    def _failure_message(self, expected_params):
+    def _failure_message(self, execute_observed, expected_params):
         return (
             "Testing for compiled statement\n%r partial params %s, "
             "received\n%%(received_statement)r with params "
@@ -223,7 +225,7 @@ class RegexSQL(CompiledSQL):
         self.params = params
         self.dialect = dialect
 
-    def _failure_message(self, expected_params):
+    def _failure_message(self, execute_observed, expected_params):
         return (
             "Testing for compiled statement ~%r partial params %s, "
             "received %%(received_statement)r with params "
@@ -263,11 +265,8 @@ class DialectSQL(CompiledSQL):
 
         return received_stmt, execute_observed.context.compiled_parameters
 
-    def _compare_sql(self, execute_observed, received_statement):
+    def _dialect_adjusted_statement(self, paramstyle):
         stmt = re.sub(r"[\n\t]", "", self.statement)
-        # convert our comparison statement to have the
-        # paramstyle of the received
-        paramstyle = execute_observed.context.dialect.paramstyle
         if paramstyle == "pyformat":
             stmt = re.sub(r":([\w_]+)", r"%(\1)s", stmt)
         else:
@@ -280,9 +279,27 @@ class DialectSQL(CompiledSQL):
             elif paramstyle == "numeric":
                 repl = None
             stmt = re.sub(r":([\w_]+)", repl, stmt)
+        return stmt
 
+    def _compare_sql(self, execute_observed, received_statement):
+        paramstyle = execute_observed.context.dialect.paramstyle
+        stmt = self._dialect_adjusted_statement(paramstyle)
         return received_statement == stmt
 
+    def _failure_message(self, execute_observed, expected_params):
+        paramstyle = execute_observed.context.dialect.paramstyle
+        return (
+            "Testing for compiled statement\n%r partial params %s, "
+            "received\n%%(received_statement)r with params "
+            "%%(received_parameters)r"
+            % (
+                self._dialect_adjusted_statement(paramstyle).replace(
+                    "%", "%%"
+                ),
+                repr(expected_params).replace("%", "%%"),
+            )
+        )
+
 
 class CountStatements(AssertRule):
     def __init__(self, count):
diff --git a/lib/sqlalchemy/testing/suite/test_insert.py b/lib/sqlalchemy/testing/suite/test_insert.py
index 3c22f50b2..080bb486e 100644
--- a/lib/sqlalchemy/testing/suite/test_insert.py
+++ b/lib/sqlalchemy/testing/suite/test_insert.py
@@ -1,5 +1,3 @@
-from .. import config
-from .. import engines
 from .. import fixtures
 from ..assertions import eq_
 from ..config import requirements
@@ -19,8 +17,6 @@ class LastrowidTest(fixtures.TablesTest):
 
     __requires__ = "implements_get_lastrowid", "autoincrement_insert"
 
-    __engine_options__ = {"implicit_returning": False}
-
     @classmethod
     def define_tables(cls, metadata):
         Table(
@@ -30,6 +26,7 @@ class LastrowidTest(fixtures.TablesTest):
                 "id", Integer, primary_key=True, test_needs_autoincrement=True
             ),
             Column("data", String(50)),
+            implicit_returning=False,
         )
 
         Table(
@@ -37,6 +34,7 @@ class LastrowidTest(fixtures.TablesTest):
             metadata,
             Column("id", Integer, primary_key=True, autoincrement=False),
             Column("data", String(50)),
+            implicit_returning=False,
         )
 
     def _assert_round_trip(self, table, conn):
@@ -110,18 +108,10 @@ class InsertBehaviorTest(fixtures.TablesTest):
         )
 
     @requirements.autoincrement_insert
-    def test_autoclose_on_insert(self):
-        if requirements.returning.enabled:
-            engine = engines.testing_engine(
-                options={"implicit_returning": False}
-            )
-        else:
-            engine = config.db
-
-        with engine.begin() as conn:
-            r = conn.execute(
-                self.tables.autoinc_pk.insert(), dict(data="some data")
-            )
+    def test_autoclose_on_insert(self, connection):
+        r = connection.execute(
+            self.tables.autoinc_pk.insert(), dict(data="some data")
+        )
         assert r._soft_closed
         assert not r.closed
         assert r.is_insert
@@ -306,8 +296,6 @@ class ReturningTest(fixtures.TablesTest):
     __requires__ = "returning", "autoincrement_insert"
     __backend__ = True
 
-    __engine_options__ = {"implicit_returning": True}
-
     def _assert_round_trip(self, table, conn):
         row = conn.execute(table.select()).first()
         eq_(
diff --git a/test/dialect/mssql/test_types.py b/test/dialect/mssql/test_types.py
index a96b74c0a..57a1f1284 100644
--- a/test/dialect/mssql/test_types.py
+++ b/test/dialect/mssql/test_types.py
@@ -1046,43 +1046,49 @@ class TypeRoundTripTest(
                         col.type.length, binary_table.c[col.name].type.length
                     )
 
-    def test_autoincrement(self, metadata, connection):
+    @testing.combinations(True, False, argnames="implicit_returning")
+    def test_autoincrement(self, metadata, connection, implicit_returning):
         Table(
             "ai_1",
             metadata,
             Column("int_y", Integer, primary_key=True, autoincrement=True),
             Column("int_n", Integer, DefaultClause("0"), primary_key=True),
+            implicit_returning=implicit_returning,
         )
         Table(
             "ai_2",
             metadata,
             Column("int_y", Integer, primary_key=True, autoincrement=True),
             Column("int_n", Integer, DefaultClause("0"), primary_key=True),
+            implicit_returning=implicit_returning,
         )
         Table(
             "ai_3",
             metadata,
             Column("int_n", Integer, DefaultClause("0"), primary_key=True),
             Column("int_y", Integer, primary_key=True, autoincrement=True),
+            implicit_returning=implicit_returning,
         )
-
         Table(
             "ai_4",
             metadata,
             Column("int_n", Integer, DefaultClause("0"), primary_key=True),
             Column("int_n2", Integer, DefaultClause("0"), primary_key=True),
+            implicit_returning=implicit_returning,
         )
         Table(
             "ai_5",
             metadata,
             Column("int_y", Integer, primary_key=True, autoincrement=True),
             Column("int_n", Integer, DefaultClause("0"), primary_key=True),
+            implicit_returning=implicit_returning,
         )
         Table(
             "ai_6",
             metadata,
             Column("o1", String(1), DefaultClause("x"), primary_key=True),
             Column("int_y", Integer, primary_key=True, autoincrement=True),
+            implicit_returning=implicit_returning,
         )
         Table(
             "ai_7",
@@ -1090,12 +1096,14 @@ class TypeRoundTripTest(
             Column("o1", String(1), DefaultClause("x"), primary_key=True),
             Column("o2", String(1), DefaultClause("x"), primary_key=True),
             Column("int_y", Integer, autoincrement=True, primary_key=True),
+            implicit_returning=implicit_returning,
         )
         Table(
             "ai_8",
             metadata,
             Column("o1", String(1), DefaultClause("x"), primary_key=True),
             Column("o2", String(1), DefaultClause("x"), primary_key=True),
+            implicit_returning=implicit_returning,
         )
         metadata.create_all(connection)
 
@@ -1124,29 +1132,18 @@ class TypeRoundTripTest(
                     eq_(col.autoincrement, "auto")
                     is_not(tbl._autoincrement_column, col)
 
-            eng = [
-                engines.testing_engine(options={"implicit_returning": False}),
-                engines.testing_engine(options={"implicit_returning": True}),
-            ]
-
-            for counter, engine in enumerate(eng):
-                connection.execute(tbl.insert())
-                if "int_y" in tbl.c:
-                    eq_(
-                        connection.execute(select(tbl.c.int_y)).scalar(),
-                        counter + 1,
-                    )
-                    assert (
-                        list(connection.execute(tbl.select()).first()).count(
-                            counter + 1
-                        )
-                        == 1
-                    )
-                else:
-                    assert 1 not in list(
-                        connection.execute(tbl.select()).first()
-                    )
-                connection.execute(tbl.delete())
+            connection.execute(tbl.insert())
+            if "int_y" in tbl.c:
+                eq_(
+                    connection.execute(select(tbl.c.int_y)).scalar(),
+                    1,
+                )
+                assert (
+                    list(connection.execute(tbl.select()).first()).count(1)
+                    == 1
+                )
+            else:
+                assert 1 not in list(connection.execute(tbl.select()).first())
 
 
 class StringTest(fixtures.TestBase, AssertsCompiledSQL):
diff --git a/test/dialect/oracle/test_dialect.py b/test/dialect/oracle/test_dialect.py
index f287a9a0b..8b6f6206b 100644
--- a/test/dialect/oracle/test_dialect.py
+++ b/test/dialect/oracle/test_dialect.py
@@ -524,17 +524,16 @@ class CompatFlagsTest(fixtures.TestBase, AssertsCompiledSQL):
         assert not dialect._use_nchar_for_unicode
 
         dialect.initialize(Mock())
-        assert not dialect.implicit_returning
+
+        # oracle 8 / 8i support returning
+        assert dialect.implicit_returning
+
         assert not dialect._supports_char_length
         assert not dialect.use_ansi
         self.assert_compile(String(50), "VARCHAR2(50)", dialect=dialect)
         self.assert_compile(Unicode(50), "VARCHAR2(50)", dialect=dialect)
         self.assert_compile(UnicodeText(), "CLOB", dialect=dialect)
 
-        dialect = self._dialect((8, 2, 5), implicit_returning=True)
-        dialect.initialize(testing.db.connect())
-        assert dialect.implicit_returning
-
     def test_default_flags(self):
         """test with no initialization or server version info"""
 
diff --git a/test/dialect/postgresql/test_compiler.py b/test/dialect/postgresql/test_compiler.py
index 5ada6f592..46adc62e5 100644
--- a/test/dialect/postgresql/test_compiler.py
+++ b/test/dialect/postgresql/test_compiler.py
@@ -48,8 +48,6 @@ from sqlalchemy.sql import literal_column
 from sqlalchemy.sql import operators
 from sqlalchemy.sql import table
 from sqlalchemy.sql import util as sql_util
-from sqlalchemy.testing import engines
-from sqlalchemy.testing import eq_
 from sqlalchemy.testing import fixtures
 from sqlalchemy.testing.assertions import assert_raises
 from sqlalchemy.testing.assertions import assert_raises_message
@@ -61,7 +59,7 @@ from sqlalchemy.util import u
 
 
 class SequenceTest(fixtures.TestBase, AssertsCompiledSQL):
-    __prefer__ = "postgresql"
+    __dialect__ = "postgresql"
 
     def test_format(self):
         seq = Sequence("my_seq_no_schema")
@@ -81,27 +79,6 @@ class SequenceTest(fixtures.TestBase, AssertsCompiledSQL):
             == '"Some_Schema"."My_Seq"'
         )
 
-    @testing.only_on("postgresql", "foo")
-    @testing.provide_metadata
-    def test_reverse_eng_name(self):
-        metadata = self.metadata
-        engine = engines.testing_engine(options=dict(implicit_returning=False))
-        for tname, cname in [
-            ("tb1" * 30, "abc"),
-            ("tb2", "abc" * 30),
-            ("tb3" * 30, "abc" * 30),
-            ("tb4", "abc"),
-        ]:
-            t = Table(
-                tname[:57],
-                metadata,
-                Column(cname[:57], Integer, primary_key=True),
-            )
-            t.create(engine)
-            with engine.begin() as conn:
-                r = conn.execute(t.insert())
-                eq_(r.inserted_primary_key, (1,))
-
     @testing.combinations(
         (None, ""),
         (Integer, "AS INTEGER "),
diff --git a/test/dialect/postgresql/test_dialect.py b/test/dialect/postgresql/test_dialect.py
index e7c4ebb7c..88784d0a3 100644
--- a/test/dialect/postgresql/test_dialect.py
+++ b/test/dialect/postgresql/test_dialect.py
@@ -1198,6 +1198,26 @@ $$ LANGUAGE plpgsql;
         connection.execute(text("CREATE SEQUENCE fooseq"))
         t.create(connection, checkfirst=True)
 
+    @testing.combinations(True, False, argnames="implicit_returning")
+    def test_sequence_detection_tricky_names(
+        self, metadata, connection, implicit_returning
+    ):
+        for tname, cname in [
+            ("tb1" * 30, "abc"),
+            ("tb2", "abc" * 30),
+            ("tb3" * 30, "abc" * 30),
+            ("tb4", "abc"),
+        ]:
+            t = Table(
+                tname[:57],
+                metadata,
+                Column(cname[:57], Integer, primary_key=True),
+                implicit_returning=implicit_returning,
+            )
+            t.create(connection)
+            r = connection.execute(t.insert())
+            eq_(r.inserted_primary_key, (1,))
+
     @testing.provide_metadata
     def test_schema_roundtrips(self, connection):
         meta = self.metadata
diff --git a/test/dialect/postgresql/test_query.py b/test/dialect/postgresql/test_query.py
index a1e9c4657..04bce4e22 100644
--- a/test/dialect/postgresql/test_query.py
+++ b/test/dialect/postgresql/test_query.py
@@ -45,27 +45,25 @@ class InsertTest(fixtures.TestBase, AssertsExecutionResults):
     __only_on__ = "postgresql"
     __backend__ = True
 
-    def setup_test(self):
-        self.metadata = MetaData()
+    @testing.combinations((False,), (True,), argnames="implicit_returning")
+    def test_foreignkey_missing_insert(
+        self, metadata, connection, implicit_returning
+    ):
 
-    def teardown_test(self):
-        with testing.db.begin() as conn:
-            self.metadata.drop_all(conn)
-
-    @testing.combinations((False,), (True,))
-    def test_foreignkey_missing_insert(self, implicit_returning):
-        engine = engines.testing_engine(
-            options={"implicit_returning": implicit_returning}
+        Table(
+            "t1",
+            metadata,
+            Column("id", Integer, primary_key=True),
+            implicit_returning=implicit_returning,
         )
-
-        Table("t1", self.metadata, Column("id", Integer, primary_key=True))
         t2 = Table(
             "t2",
-            self.metadata,
+            metadata,
             Column("id", Integer, ForeignKey("t1.id"), primary_key=True),
+            implicit_returning=implicit_returning,
         )
 
-        self.metadata.create_all(engine)
+        metadata.create_all(connection)
 
         # want to ensure that "null value in column "id" violates not-
         # null constraint" is raised (IntegrityError on psycoopg2, but
@@ -75,54 +73,36 @@ class InsertTest(fixtures.TestBase, AssertsExecutionResults):
         # the case here due to the foreign key.
 
         with expect_warnings(".*has no Python-side or server-side default.*"):
-            with engine.begin() as conn:
-                assert_raises(
-                    (exc.IntegrityError, exc.ProgrammingError),
-                    conn.execute,
-                    t2.insert(),
-                )
-
-    def test_sequence_insert(self):
-        table = Table(
-            "testtable",
-            self.metadata,
-            Column("id", Integer, Sequence("my_seq"), primary_key=True),
-            Column("data", String(30)),
-        )
-        self.metadata.create_all(testing.db)
-        self._assert_data_with_sequence(table, "my_seq")
+            assert_raises(
+                (exc.IntegrityError, exc.ProgrammingError),
+                connection.execute,
+                t2.insert(),
+            )
 
-    @testing.requires.returning
-    def test_sequence_returning_insert(self):
+    @testing.combinations(True, False, argnames="implicit_returning")
+    def test_sequence_insert(self, metadata, connection, implicit_returning):
         table = Table(
             "testtable",
-            self.metadata,
+            metadata,
             Column("id", Integer, Sequence("my_seq"), primary_key=True),
             Column("data", String(30)),
+            implicit_returning=implicit_returning,
         )
-        self.metadata.create_all(testing.db)
-        self._assert_data_with_sequence_returning(table, "my_seq")
-
-    def test_opt_sequence_insert(self):
-        table = Table(
-            "testtable",
-            self.metadata,
-            Column(
-                "id",
-                Integer,
-                Sequence("my_seq", optional=True),
-                primary_key=True,
-            ),
-            Column("data", String(30)),
-        )
-        self.metadata.create_all(testing.db)
-        self._assert_data_autoincrement(table)
+        metadata.create_all(connection)
+        if implicit_returning:
+            self._assert_data_with_sequence_returning(
+                connection, table, "my_seq"
+            )
+        else:
+            self._assert_data_with_sequence(connection, table, "my_seq")
 
-    @testing.requires.returning
-    def test_opt_sequence_returning_insert(self):
+    @testing.combinations(True, False, argnames="implicit_returning")
+    def test_opt_sequence_insert(
+        self, metadata, connection, implicit_returning
+    ):
         table = Table(
             "testtable",
-            self.metadata,
+            metadata,
             Column(
                 "id",
                 Integer,
@@ -130,78 +110,85 @@ class InsertTest(fixtures.TestBase, AssertsExecutionResults):
                 primary_key=True,
             ),
             Column("data", String(30)),
+            implicit_returning=implicit_returning,
         )
-        self.metadata.create_all(testing.db)
-        self._assert_data_autoincrement_returning(table)
-
-    def test_autoincrement_insert(self):
-        table = Table(
-            "testtable",
-            self.metadata,
-            Column("id", Integer, primary_key=True),
-            Column("data", String(30)),
-        )
-        self.metadata.create_all(testing.db)
-        self._assert_data_autoincrement(table)
+        metadata.create_all(connection)
+        if implicit_returning:
+            self._assert_data_autoincrement_returning(connection, table)
+        else:
+            self._assert_data_autoincrement(connection, table)
 
-    @testing.requires.returning
-    def test_autoincrement_returning_insert(self):
+    @testing.combinations(True, False, argnames="implicit_returning")
+    def test_autoincrement_insert(
+        self, metadata, connection, implicit_returning
+    ):
         table = Table(
             "testtable",
-            self.metadata,
+            metadata,
             Column("id", Integer, primary_key=True),
             Column("data", String(30)),
+            implicit_returning=implicit_returning,
         )
-        self.metadata.create_all(testing.db)
-        self._assert_data_autoincrement_returning(table)
+        metadata.create_all(connection)
+        if implicit_returning:
+            self._assert_data_autoincrement_returning(connection, table)
+        else:
+            self._assert_data_autoincrement(connection, table)
 
-    def test_noautoincrement_insert(self):
+    @testing.combinations(True, False, argnames="implicit_returning")
+    def test_noautoincrement_insert(
+        self, metadata, connection, implicit_returning
+    ):
         table = Table(
             "testtable",
-            self.metadata,
+            metadata,
             Column("id", Integer, primary_key=True, autoincrement=False),
             Column("data", String(30)),
+            implicit_returning=implicit_returning,
         )
-        self.metadata.create_all(testing.db)
-        self._assert_data_noautoincrement(table)
-
-    def _assert_data_autoincrement(self, table):
-        engine = engines.testing_engine(options={"implicit_returning": False})
+        metadata.create_all(connection)
+        self._assert_data_noautoincrement(connection, table)
 
-        with self.sql_execution_asserter(engine) as asserter:
+    def _assert_data_autoincrement(self, connection, table):
+        """
+        invoked by:
+        * test_opt_sequence_insert
+        * test_autoincrement_insert
+        """
 
-            with engine.begin() as conn:
-                # execute with explicit id
+        with self.sql_execution_asserter(connection) as asserter:
+            conn = connection
+            # execute with explicit id
 
-                r = conn.execute(table.insert(), {"id": 30, "data": "d1"})
-                eq_(r.inserted_primary_key, (30,))
+            r = conn.execute(table.insert(), {"id": 30, "data": "d1"})
+            eq_(r.inserted_primary_key, (30,))
 
-                # execute with prefetch id
+            # execute with prefetch id
 
-                r = conn.execute(table.insert(), {"data": "d2"})
-                eq_(r.inserted_primary_key, (1,))
+            r = conn.execute(table.insert(), {"data": "d2"})
+            eq_(r.inserted_primary_key, (1,))
 
-                # executemany with explicit ids
+            # executemany with explicit ids
 
-                conn.execute(
-                    table.insert(),
-                    [
-                        {"id": 31, "data": "d3"},
-                        {"id": 32, "data": "d4"},
-                    ],
-                )
+            conn.execute(
+                table.insert(),
+                [
+                    {"id": 31, "data": "d3"},
+                    {"id": 32, "data": "d4"},
+                ],
+            )
 
-                # executemany, uses SERIAL
+            # executemany, uses SERIAL
 
-                conn.execute(table.insert(), [{"data": "d5"}, {"data": "d6"}])
+            conn.execute(table.insert(), [{"data": "d5"}, {"data": "d6"}])
 
-                # single execute, explicit id, inline
+            # single execute, explicit id, inline
 
-                conn.execute(table.insert().inline(), {"id": 33, "data": "d7"})
+            conn.execute(table.insert().inline(), {"id": 33, "data": "d7"})
 
-                # single execute, inline, uses SERIAL
+            # single execute, inline, uses SERIAL
 
-                conn.execute(table.insert().inline(), {"data": "d8"})
+            conn.execute(table.insert().inline(), {"data": "d8"})
 
         asserter.assert_(
             DialectSQL(
@@ -229,44 +216,44 @@ class InsertTest(fixtures.TestBase, AssertsExecutionResults):
             ),
         )
 
-        with engine.begin() as conn:
-            eq_(
-                conn.execute(table.select()).fetchall(),
-                [
-                    (30, "d1"),
-                    (1, "d2"),
-                    (31, "d3"),
-                    (32, "d4"),
-                    (2, "d5"),
-                    (3, "d6"),
-                    (33, "d7"),
-                    (4, "d8"),
-                ],
-            )
+        eq_(
+            conn.execute(table.select()).fetchall(),
+            [
+                (30, "d1"),
+                (1, "d2"),
+                (31, "d3"),
+                (32, "d4"),
+                (2, "d5"),
+                (3, "d6"),
+                (33, "d7"),
+                (4, "d8"),
+            ],
+        )
 
-            conn.execute(table.delete())
+        conn.execute(table.delete())
 
         # test the same series of events using a reflected version of
         # the table
 
         m2 = MetaData()
-        table = Table(table.name, m2, autoload_with=engine)
-
-        with self.sql_execution_asserter(engine) as asserter:
-            with engine.begin() as conn:
-                conn.execute(table.insert(), {"id": 30, "data": "d1"})
-                r = conn.execute(table.insert(), {"data": "d2"})
-                eq_(r.inserted_primary_key, (5,))
-                conn.execute(
-                    table.insert(),
-                    [
-                        {"id": 31, "data": "d3"},
-                        {"id": 32, "data": "d4"},
-                    ],
-                )
-                conn.execute(table.insert(), [{"data": "d5"}, {"data": "d6"}])
-                conn.execute(table.insert().inline(), {"id": 33, "data": "d7"})
-                conn.execute(table.insert().inline(), {"data": "d8"})
+        table = Table(
+            table.name, m2, autoload_with=connection, implicit_returning=False
+        )
+
+        with self.sql_execution_asserter(connection) as asserter:
+            conn.execute(table.insert(), {"id": 30, "data": "d1"})
+            r = conn.execute(table.insert(), {"data": "d2"})
+            eq_(r.inserted_primary_key, (5,))
+            conn.execute(
+                table.insert(),
+                [
+                    {"id": 31, "data": "d3"},
+                    {"id": 32, "data": "d4"},
+                ],
+            )
+            conn.execute(table.insert(), [{"data": "d5"}, {"data": "d6"}])
+            conn.execute(table.insert().inline(), {"id": 33, "data": "d7"})
+            conn.execute(table.insert().inline(), {"data": "d8"})
 
         asserter.assert_(
             DialectSQL(
@@ -293,59 +280,61 @@ class InsertTest(fixtures.TestBase, AssertsExecutionResults):
                 "INSERT INTO testtable (data) VALUES (:data)", [{"data": "d8"}]
             ),
         )
-        with engine.begin() as conn:
-            eq_(
-                conn.execute(table.select()).fetchall(),
-                [
-                    (30, "d1"),
-                    (5, "d2"),
-                    (31, "d3"),
-                    (32, "d4"),
-                    (6, "d5"),
-                    (7, "d6"),
-                    (33, "d7"),
-                    (8, "d8"),
-                ],
-            )
-            conn.execute(table.delete())
 
-    def _assert_data_autoincrement_returning(self, table):
-        engine = engines.testing_engine(options={"implicit_returning": True})
+        eq_(
+            conn.execute(table.select()).fetchall(),
+            [
+                (30, "d1"),
+                (5, "d2"),
+                (31, "d3"),
+                (32, "d4"),
+                (6, "d5"),
+                (7, "d6"),
+                (33, "d7"),
+                (8, "d8"),
+            ],
+        )
 
-        with self.sql_execution_asserter(engine) as asserter:
-            with engine.begin() as conn:
+    def _assert_data_autoincrement_returning(self, connection, table):
+        """
+        invoked by:
+        * test_opt_sequence_returning_insert
+        * test_autoincrement_returning_insert
+        """
+        with self.sql_execution_asserter(connection) as asserter:
+            conn = connection
 
-                # execute with explicit id
+            # execute with explicit id
 
-                r = conn.execute(table.insert(), {"id": 30, "data": "d1"})
-                eq_(r.inserted_primary_key, (30,))
+            r = conn.execute(table.insert(), {"id": 30, "data": "d1"})
+            eq_(r.inserted_primary_key, (30,))
 
-                # execute with prefetch id
+            # execute with prefetch id
 
-                r = conn.execute(table.insert(), {"data": "d2"})
-                eq_(r.inserted_primary_key, (1,))
+            r = conn.execute(table.insert(), {"data": "d2"})
+            eq_(r.inserted_primary_key, (1,))
 
-                # executemany with explicit ids
+            # executemany with explicit ids
 
-                conn.execute(
-                    table.insert(),
-                    [
-                        {"id": 31, "data": "d3"},
-                        {"id": 32, "data": "d4"},
-                    ],
-                )
+            conn.execute(
+                table.insert(),
+                [
+                    {"id": 31, "data": "d3"},
+                    {"id": 32, "data": "d4"},
+                ],
+            )
 
-                # executemany, uses SERIAL
+            # executemany, uses SERIAL
 
-                conn.execute(table.insert(), [{"data": "d5"}, {"data": "d6"}])
+            conn.execute(table.insert(), [{"data": "d5"}, {"data": "d6"}])
 
-                # single execute, explicit id, inline
+            # single execute, explicit id, inline
 
-                conn.execute(table.insert().inline(), {"id": 33, "data": "d7"})
+            conn.execute(table.insert().inline(), {"id": 33, "data": "d7"})
 
-                # single execute, inline, uses SERIAL
+            # single execute, inline, uses SERIAL
 
-                conn.execute(table.insert().inline(), {"data": "d8"})
+            conn.execute(table.insert().inline(), {"data": "d8"})
 
         asserter.assert_(
             DialectSQL(
@@ -374,43 +363,46 @@ class InsertTest(fixtures.TestBase, AssertsExecutionResults):
             ),
         )
 
-        with engine.begin() as conn:
-            eq_(
-                conn.execute(table.select()).fetchall(),
-                [
-                    (30, "d1"),
-                    (1, "d2"),
-                    (31, "d3"),
-                    (32, "d4"),
-                    (2, "d5"),
-                    (3, "d6"),
-                    (33, "d7"),
-                    (4, "d8"),
-                ],
-            )
-            conn.execute(table.delete())
+        eq_(
+            conn.execute(table.select()).fetchall(),
+            [
+                (30, "d1"),
+                (1, "d2"),
+                (31, "d3"),
+                (32, "d4"),
+                (2, "d5"),
+                (3, "d6"),
+                (33, "d7"),
+                (4, "d8"),
+            ],
+        )
+        conn.execute(table.delete())
 
         # test the same series of events using a reflected version of
         # the table
 
         m2 = MetaData()
-        table = Table(table.name, m2, autoload_with=engine)
-
-        with self.sql_execution_asserter(engine) as asserter:
-            with engine.begin() as conn:
-                conn.execute(table.insert(), {"id": 30, "data": "d1"})
-                r = conn.execute(table.insert(), {"data": "d2"})
-                eq_(r.inserted_primary_key, (5,))
-                conn.execute(
-                    table.insert(),
-                    [
-                        {"id": 31, "data": "d3"},
-                        {"id": 32, "data": "d4"},
-                    ],
-                )
-                conn.execute(table.insert(), [{"data": "d5"}, {"data": "d6"}])
-                conn.execute(table.insert().inline(), {"id": 33, "data": "d7"})
-                conn.execute(table.insert().inline(), {"data": "d8"})
+        table = Table(
+            table.name,
+            m2,
+            autoload_with=connection,
+            implicit_returning=True,
+        )
+
+        with self.sql_execution_asserter(connection) as asserter:
+            conn.execute(table.insert(), {"id": 30, "data": "d1"})
+            r = conn.execute(table.insert(), {"data": "d2"})
+            eq_(r.inserted_primary_key, (5,))
+            conn.execute(
+                table.insert(),
+                [
+                    {"id": 31, "data": "d3"},
+                    {"id": 32, "data": "d4"},
+                ],
+            )
+            conn.execute(table.insert(), [{"data": "d5"}, {"data": "d6"}])
+            conn.execute(table.insert().inline(), {"id": 33, "data": "d7"})
+            conn.execute(table.insert().inline(), {"data": "d8"})
 
         asserter.assert_(
             DialectSQL(
@@ -439,39 +431,40 @@ class InsertTest(fixtures.TestBase, AssertsExecutionResults):
             ),
         )
 
-        with engine.begin() as conn:
-            eq_(
-                conn.execute(table.select()).fetchall(),
+        eq_(
+            conn.execute(table.select()).fetchall(),
+            [
+                (30, "d1"),
+                (5, "d2"),
+                (31, "d3"),
+                (32, "d4"),
+                (6, "d5"),
+                (7, "d6"),
+                (33, "d7"),
+                (8, "d8"),
+            ],
+        )
+
+    def _assert_data_with_sequence(self, connection, table, seqname):
+        """
+        invoked by:
+        * test_sequence_insert
+        """
+
+        with self.sql_execution_asserter(connection) as asserter:
+            conn = connection
+            conn.execute(table.insert(), {"id": 30, "data": "d1"})
+            conn.execute(table.insert(), {"data": "d2"})
+            conn.execute(
+                table.insert(),
                 [
-                    (30, "d1"),
-                    (5, "d2"),
-                    (31, "d3"),
-                    (32, "d4"),
-                    (6, "d5"),
-                    (7, "d6"),
-                    (33, "d7"),
-                    (8, "d8"),
+                    {"id": 31, "data": "d3"},
+                    {"id": 32, "data": "d4"},
                 ],
             )
-            conn.execute(table.delete())
-
-    def _assert_data_with_sequence(self, table, seqname):
-        engine = engines.testing_engine(options={"implicit_returning": False})
-
-        with self.sql_execution_asserter(engine) as asserter:
-            with engine.begin() as conn:
-                conn.execute(table.insert(), {"id": 30, "data": "d1"})
-                conn.execute(table.insert(), {"data": "d2"})
-                conn.execute(
-                    table.insert(),
-                    [
-                        {"id": 31, "data": "d3"},
-                        {"id": 32, "data": "d4"},
-                    ],
-                )
-                conn.execute(table.insert(), [{"data": "d5"}, {"data": "d6"}])
-                conn.execute(table.insert().inline(), {"id": 33, "data": "d7"})
-                conn.execute(table.insert().inline(), {"data": "d8"})
+            conn.execute(table.insert(), [{"data": "d5"}, {"data": "d6"}])
+            conn.execute(table.insert().inline(), {"id": 33, "data": "d7"})
+            conn.execute(table.insert().inline(), {"data": "d8"})
 
         asserter.assert_(
             DialectSQL(
@@ -502,41 +495,40 @@ class InsertTest(fixtures.TestBase, AssertsExecutionResults):
                 [{"data": "d8"}],
             ),
         )
-        with engine.begin() as conn:
-            eq_(
-                conn.execute(table.select()).fetchall(),
+        eq_(
+            conn.execute(table.select()).fetchall(),
+            [
+                (30, "d1"),
+                (1, "d2"),
+                (31, "d3"),
+                (32, "d4"),
+                (2, "d5"),
+                (3, "d6"),
+                (33, "d7"),
+                (4, "d8"),
+            ],
+        )
+
+    def _assert_data_with_sequence_returning(self, connection, table, seqname):
+        """
+        invoked by:
+        * test_sequence_returning_insert
+        """
+
+        with self.sql_execution_asserter(connection) as asserter:
+            conn = connection
+            conn.execute(table.insert(), {"id": 30, "data": "d1"})
+            conn.execute(table.insert(), {"data": "d2"})
+            conn.execute(
+                table.insert(),
                 [
-                    (30, "d1"),
-                    (1, "d2"),
-                    (31, "d3"),
-                    (32, "d4"),
-                    (2, "d5"),
-                    (3, "d6"),
-                    (33, "d7"),
-                    (4, "d8"),
+                    {"id": 31, "data": "d3"},
+                    {"id": 32, "data": "d4"},
                 ],
             )
-
-        # cant test reflection here since the Sequence must be
-        # explicitly specified
-
-    def _assert_data_with_sequence_returning(self, table, seqname):
-        engine = engines.testing_engine(options={"implicit_returning": True})
-
-        with self.sql_execution_asserter(engine) as asserter:
-            with engine.begin() as conn:
-                conn.execute(table.insert(), {"id": 30, "data": "d1"})
-                conn.execute(table.insert(), {"data": "d2"})
-                conn.execute(
-                    table.insert(),
-                    [
-                        {"id": 31, "data": "d3"},
-                        {"id": 32, "data": "d4"},
-                    ],
-                )
-                conn.execute(table.insert(), [{"data": "d5"}, {"data": "d6"}])
-                conn.execute(table.insert().inline(), {"id": 33, "data": "d7"})
-                conn.execute(table.insert().inline(), {"data": "d8"})
+            conn.execute(table.insert(), [{"data": "d5"}, {"data": "d6"}])
+            conn.execute(table.insert().inline(), {"id": 33, "data": "d7"})
+            conn.execute(table.insert().inline(), {"data": "d8"})
 
         asserter.assert_(
             DialectSQL(
@@ -568,35 +560,34 @@ class InsertTest(fixtures.TestBase, AssertsExecutionResults):
             ),
         )
 
-        with engine.begin() as conn:
-            eq_(
-                conn.execute(table.select()).fetchall(),
-                [
-                    (30, "d1"),
-                    (1, "d2"),
-                    (31, "d3"),
-                    (32, "d4"),
-                    (2, "d5"),
-                    (3, "d6"),
-                    (33, "d7"),
-                    (4, "d8"),
-                ],
-            )
-
-            # cant test reflection here since the Sequence must be
-            # explicitly specified
+        eq_(
+            connection.execute(table.select()).fetchall(),
+            [
+                (30, "d1"),
+                (1, "d2"),
+                (31, "d3"),
+                (32, "d4"),
+                (2, "d5"),
+                (3, "d6"),
+                (33, "d7"),
+                (4, "d8"),
+            ],
+        )
 
-    def _assert_data_noautoincrement(self, table):
-        engine = engines.testing_engine(options={"implicit_returning": False})
+    def _assert_data_noautoincrement(self, connection, table):
+        """
+        invoked by:
+        * test_noautoincrement_insert
+        """
 
         # turning off the cache because we are checking for compile-time
         # warnings
-        engine = engine.execution_options(compiled_cache=None)
+        connection.execution_options(compiled_cache=None)
 
-        with engine.begin() as conn:
-            conn.execute(table.insert(), {"id": 30, "data": "d1"})
+        conn = connection
+        conn.execute(table.insert(), {"id": 30, "data": "d1"})
 
-        with engine.begin() as conn:
+        with conn.begin_nested() as nested:
             with expect_warnings(
                 ".*has no Python-side or server-side default.*"
             ):
@@ -606,8 +597,9 @@ class InsertTest(fixtures.TestBase, AssertsExecutionResults):
                     table.insert(),
                     {"data": "d2"},
                 )
+            nested.rollback()
 
-        with engine.begin() as conn:
+        with conn.begin_nested() as nested:
             with expect_warnings(
                 ".*has no Python-side or server-side default.*"
             ):
@@ -617,8 +609,9 @@ class InsertTest(fixtures.TestBase, AssertsExecutionResults):
                     table.insert(),
                     [{"data": "d2"}, {"data": "d3"}],
                 )
+            nested.rollback()
 
-        with engine.begin() as conn:
+        with conn.begin_nested() as nested:
             with expect_warnings(
                 ".*has no Python-side or server-side default.*"
             ):
@@ -628,8 +621,9 @@ class InsertTest(fixtures.TestBase, AssertsExecutionResults):
                     table.insert(),
                     {"data": "d2"},
                 )
+            nested.rollback()
 
-        with engine.begin() as conn:
+        with conn.begin_nested() as nested:
             with expect_warnings(
                 ".*has no Python-side or server-side default.*"
             ):
@@ -639,28 +633,29 @@ class InsertTest(fixtures.TestBase, AssertsExecutionResults):
                     table.insert(),
                     [{"data": "d2"}, {"data": "d3"}],
                 )
+            nested.rollback()
 
-        with engine.begin() as conn:
-            conn.execute(
-                table.insert(),
-                [{"id": 31, "data": "d2"}, {"id": 32, "data": "d3"}],
-            )
-            conn.execute(table.insert().inline(), {"id": 33, "data": "d4"})
-            eq_(
-                conn.execute(table.select()).fetchall(),
-                [(30, "d1"), (31, "d2"), (32, "d3"), (33, "d4")],
-            )
-            conn.execute(table.delete())
+        conn.execute(
+            table.insert(),
+            [{"id": 31, "data": "d2"}, {"id": 32, "data": "d3"}],
+        )
+        conn.execute(table.insert().inline(), {"id": 33, "data": "d4"})
+        eq_(
+            conn.execute(table.select()).fetchall(),
+            [(30, "d1"), (31, "d2"), (32, "d3"), (33, "d4")],
+        )
+        conn.execute(table.delete())
 
         # test the same series of events using a reflected version of
         # the table
 
         m2 = MetaData()
-        table = Table(table.name, m2, autoload_with=engine)
-        with engine.begin() as conn:
-            conn.execute(table.insert(), {"id": 30, "data": "d1"})
+        table = Table(table.name, m2, autoload_with=connection)
+        conn = connection
+
+        conn.execute(table.insert(), {"id": 30, "data": "d1"})
 
-        with engine.begin() as conn:
+        with conn.begin_nested() as nested:
             with expect_warnings(
                 ".*has no Python-side or server-side default.*"
             ):
@@ -670,8 +665,9 @@ class InsertTest(fixtures.TestBase, AssertsExecutionResults):
                     table.insert(),
                     {"data": "d2"},
                 )
+            nested.rollback()
 
-        with engine.begin() as conn:
+        with conn.begin_nested() as nested:
             with expect_warnings(
                 ".*has no Python-side or server-side default.*"
             ):
@@ -681,17 +677,17 @@ class InsertTest(fixtures.TestBase, AssertsExecutionResults):
                     table.insert(),
                     [{"data": "d2"}, {"data": "d3"}],
                 )
+            nested.rollback()
 
-        with engine.begin() as conn:
-            conn.execute(
-                table.insert(),
-                [{"id": 31, "data": "d2"}, {"id": 32, "data": "d3"}],
-            )
-            conn.execute(table.insert().inline(), {"id": 33, "data": "d4"})
-            eq_(
-                conn.execute(table.select()).fetchall(),
-                [(30, "d1"), (31, "d2"), (32, "d3"), (33, "d4")],
-            )
+        conn.execute(
+            table.insert(),
+            [{"id": 31, "data": "d2"}, {"id": 32, "data": "d3"}],
+        )
+        conn.execute(table.insert().inline(), {"id": 33, "data": "d4"})
+        eq_(
+            conn.execute(table.select()).fetchall(),
+            [(30, "d1"), (31, "d2"), (32, "d3"), (33, "d4")],
+        )
 
 
 class MatchTest(fixtures.TablesTest, AssertsCompiledSQL):
diff --git a/test/engine/test_deprecations.py b/test/engine/test_deprecations.py
index 6e7169d12..fd5148316 100644
--- a/test/engine/test_deprecations.py
+++ b/test/engine/test_deprecations.py
@@ -7,6 +7,7 @@ from sqlalchemy import engine
 from sqlalchemy import event
 from sqlalchemy import exc
 from sqlalchemy import ForeignKey
+from sqlalchemy import insert
 from sqlalchemy import inspect
 from sqlalchemy import Integer
 from sqlalchemy import MetaData
@@ -22,6 +23,7 @@ from sqlalchemy.engine.base import Engine
 from sqlalchemy.engine.mock import MockConnection
 from sqlalchemy.testing import assert_raises
 from sqlalchemy.testing import assert_raises_message
+from sqlalchemy.testing import assertions
 from sqlalchemy.testing import config
 from sqlalchemy.testing import engines
 from sqlalchemy.testing import eq_
@@ -32,6 +34,7 @@ from sqlalchemy.testing import is_instance_of
 from sqlalchemy.testing import is_true
 from sqlalchemy.testing import mock
 from sqlalchemy.testing.assertions import expect_deprecated
+from sqlalchemy.testing.assertions import expect_raises_message
 from sqlalchemy.testing.engines import testing_engine
 from sqlalchemy.testing.mock import Mock
 from sqlalchemy.testing.schema import Column
@@ -545,3 +548,69 @@ class EngineEventsTest(fixtures.TestBase):
             with e1.connect() as conn:
                 result = conn.execute(select(1))
                 result.close()
+
+
+ce_implicit_returning = (
+    r"The create_engine.implicit_returning parameter is deprecated "
+    r"and will be removed in a future release."
+)
+
+
+class ImplicitReturningFlagTest(fixtures.TestBase):
+    __backend__ = True
+
+    @testing.combinations(True, False, None, argnames="implicit_returning")
+    def test_implicit_returning_engine_parameter(self, implicit_returning):
+        if implicit_returning is None:
+            e = engines.testing_engine()
+        else:
+            with assertions.expect_deprecated(ce_implicit_returning):
+                e = engines.testing_engine(
+                    options={"implicit_returning": implicit_returning}
+                )
+
+        if implicit_returning is None:
+            eq_(
+                e.dialect.implicit_returning,
+                testing.db.dialect.implicit_returning,
+            )
+        else:
+            eq_(e.dialect.implicit_returning, implicit_returning)
+
+        t = Table(
+            "t",
+            MetaData(),
+            Column("id", Integer, primary_key=True),
+            Column("data", String(50)),
+        )
+
+        t2 = Table(
+            "t",
+            MetaData(),
+            Column("id", Integer, primary_key=True),
+            Column("data", String(50)),
+            implicit_returning=False,
+        )
+
+        with e.connect() as conn:
+            stmt = insert(t).values(data="data")
+
+            if implicit_returning:
+                if not testing.requires.returning.enabled:
+                    with expect_raises_message(
+                        exc.CompileError, "RETURNING is not supported"
+                    ):
+                        stmt.compile(conn)
+                else:
+                    eq_(stmt.compile(conn).returning, [t.c.id])
+            elif (
+                implicit_returning is None
+                and testing.db.dialect.implicit_returning
+            ):
+                eq_(stmt.compile(conn).returning, [t.c.id])
+            else:
+                eq_(stmt.compile(conn).returning, [])
+
+            # table setting it to False disables it
+            stmt2 = insert(t2).values(data="data")
+            eq_(stmt2.compile(conn).returning, [])
diff --git a/test/engine/test_execute.py b/test/engine/test_execute.py
index afe95ba82..5a5f0e8e6 100644
--- a/test/engine/test_execute.py
+++ b/test/engine/test_execute.py
@@ -2010,10 +2010,8 @@ class EngineEventsTest(fixtures.TestBase):
         # TODO: this test is kind of a mess
 
         for engine in [
-            engines.testing_engine(options=dict(implicit_returning=False)),
-            engines.testing_engine(
-                options=dict(implicit_returning=False)
-            ).connect(),
+            engines.testing_engine(),
+            engines.testing_engine().connect(),
         ]:
             event.listen(engine, "before_execute", execute)
             event.listen(engine, "before_cursor_execute", cursor_execute)
@@ -2028,6 +2026,7 @@ class EngineEventsTest(fixtures.TestBase):
                     default=func.lower("Foo"),
                     primary_key=True,
                 ),
+                implicit_returning=False,
             )
 
             if isinstance(engine, Connection):
diff --git a/test/requirements.py b/test/requirements.py
index 28a283b12..876507387 100644
--- a/test/requirements.py
+++ b/test/requirements.py
@@ -1643,6 +1643,12 @@ class DefaultRequirements(SuiteRequirements):
         return skip_if("mssql")
 
     @property
+    def supports_autoincrement_w_composite_pk(self):
+        """integer autoincrement works for tables with composite primary
+        keys"""
+        return fails_if("sqlite")
+
+    @property
     def identity_columns(self):
         return only_if(["postgresql >= 10", "oracle >= 12", "mssql"])
 
diff --git a/test/sql/test_defaults.py b/test/sql/test_defaults.py
index 31cc15155..aa3728a98 100644
--- a/test/sql/test_defaults.py
+++ b/test/sql/test_defaults.py
@@ -20,7 +20,6 @@ from sqlalchemy.sql import select
 from sqlalchemy.sql import text
 from sqlalchemy.testing import assert_raises_message
 from sqlalchemy.testing import AssertsCompiledSQL
-from sqlalchemy.testing import engines
 from sqlalchemy.testing import eq_
 from sqlalchemy.testing import expect_warnings
 from sqlalchemy.testing import fixtures
@@ -921,68 +920,70 @@ class CTEDefaultTest(fixtures.TablesTest):
         eq_(conn.execute(select(q.c.x, q.c.y)).first(), expected)
 
 
-class PKDefaultTest(fixtures.TablesTest):
+class PKDefaultTest(fixtures.TestBase):
     __requires__ = ("subqueries",)
     __backend__ = True
 
-    @classmethod
-    def define_tables(cls, metadata):
-        t2 = Table("t2", metadata, Column("nextid", Integer))
+    @testing.fixture
+    def table_fixture(self, metadata, connection):
+        def go(implicit_returning):
+            t2 = Table(
+                "t2",
+                metadata,
+                Column("nextid", Integer),
+                implicit_returning=implicit_returning,
+            )
 
-        Table(
-            "t1",
-            metadata,
-            Column(
-                "id",
-                Integer,
-                primary_key=True,
-                default=sa.select(func.max(t2.c.nextid)).scalar_subquery(),
-            ),
-            Column("data", String(30)),
-        )
+            t1 = Table(
+                "t1",
+                metadata,
+                Column(
+                    "id",
+                    Integer,
+                    primary_key=True,
+                    default=sa.select(func.max(t2.c.nextid)).scalar_subquery(),
+                ),
+                Column("data", String(30)),
+                implicit_returning=implicit_returning,
+            )
 
-        Table(
-            "date_table",
-            metadata,
-            Column(
-                "date_id",
-                DateTime(timezone=True),
-                default=text("current_timestamp"),
-                primary_key=True,
-            ),
-        )
+            date_table = Table(
+                "date_table",
+                metadata,
+                Column(
+                    "date_id",
+                    DateTime(timezone=True),
+                    default=text("current_timestamp"),
+                    primary_key=True,
+                ),
+                implicit_returning=implicit_returning,
+            )
 
-    @testing.requires.returning
-    def test_with_implicit_returning(self):
-        self._test(True)
+            metadata.create_all(connection)
+            return t1, t2, date_table
 
-    def test_regular(self):
-        self._test(False)
+        return go
 
-    def _test(self, returning):
-        t2, t1, date_table = (
-            self.tables.t2,
-            self.tables.t1,
-            self.tables.date_table,
-        )
+    @testing.combinations(
+        (True, testing.requires.returning),
+        (False,),
+        argnames="implicit_returning",
+    )
+    def test_pk_default(self, connection, table_fixture, implicit_returning):
+        t1, t2, date_table = table_fixture(implicit_returning)
 
-        if not returning and not testing.db.dialect.implicit_returning:
-            engine = testing.db
-        else:
-            engine = engines.testing_engine(
-                options={"implicit_returning": returning}
-            )
-        with engine.begin() as conn:
-            conn.execute(t2.insert(), dict(nextid=1))
-            r = conn.execute(t1.insert(), dict(data="hi"))
-            eq_((1,), r.inserted_primary_key)
+        conn = connection
+
+        conn.execute(t2.insert(), dict(nextid=1))
+        r = conn.execute(t1.insert(), dict(data="hi"))
+        eq_((1,), r.inserted_primary_key)
 
-            conn.execute(t2.insert(), dict(nextid=2))
-            r = conn.execute(t1.insert(), dict(data="there"))
-            eq_((2,), r.inserted_primary_key)
+        conn.execute(t2.insert(), dict(nextid=2))
+        r = conn.execute(t1.insert(), dict(data="there"))
+        eq_((2,), r.inserted_primary_key)
 
-            r = conn.execute(date_table.insert())
-            assert isinstance(r.inserted_primary_key[0], datetime.datetime)
+        r = conn.execute(date_table.insert())
+        assert isinstance(r.inserted_primary_key[0], datetime.datetime)
 
 
 class PKIncrementTest(fixtures.TablesTest):
diff --git a/test/sql/test_insert_exec.py b/test/sql/test_insert_exec.py
index e0e3b60a9..ef1a3be09 100644
--- a/test/sql/test_insert_exec.py
+++ b/test/sql/test_insert_exec.py
@@ -5,17 +5,16 @@ from sqlalchemy import func
 from sqlalchemy import INT
 from sqlalchemy import Integer
 from sqlalchemy import literal
-from sqlalchemy import MetaData
 from sqlalchemy import Sequence
 from sqlalchemy import sql
 from sqlalchemy import String
 from sqlalchemy import testing
 from sqlalchemy import VARCHAR
 from sqlalchemy.testing import assert_raises_message
-from sqlalchemy.testing import engines
 from sqlalchemy.testing import eq_
 from sqlalchemy.testing import fixtures
 from sqlalchemy.testing import is_
+from sqlalchemy.testing import mock
 from sqlalchemy.testing.schema import Column
 from sqlalchemy.testing.schema import Table
 
@@ -89,10 +88,10 @@ class InsertExecTest(fixtures.TablesTest):
             ],
         )
 
-    def _test_lastrow_accessor(self, table_, values, assertvalues):
+    def _test_lastrow_accessor(self, connection, table_, values, assertvalues):
         """Tests the inserted_primary_key and lastrow_has_id() functions."""
 
-        def insert_values(engine, table_, values):
+        def insert_values(table_, values):
             """
             Inserts a row into a table, returns the full list of values
             INSERTed including defaults that fired off on the DB side and
@@ -100,71 +99,62 @@ class InsertExecTest(fixtures.TablesTest):
             """
 
             # verify implicit_returning is working
-            if engine.dialect.implicit_returning:
+            if (
+                connection.dialect.implicit_returning
+                and table_.implicit_returning
+            ):
                 ins = table_.insert()
-                comp = ins.compile(engine, column_keys=list(values))
+                comp = ins.compile(connection, column_keys=list(values))
                 if not set(values).issuperset(
                     c.key for c in table_.primary_key
                 ):
                     is_(bool(comp.returning), True)
 
-            with engine.begin() as connection:
-                result = connection.execute(table_.insert(), values)
-                ret = values.copy()
-
-                ipk = result.inserted_primary_key
-                for col, id_ in zip(table_.primary_key, ipk):
-                    ret[col.key] = id_
-
-                if result.lastrow_has_defaults():
-                    criterion = and_(
-                        *[
-                            col == id_
-                            for col, id_ in zip(
-                                table_.primary_key, result.inserted_primary_key
-                            )
-                        ]
-                    )
-                    row = connection.execute(
-                        table_.select().where(criterion)
-                    ).first()
-                    for c in table_.c:
-                        ret[c.key] = row._mapping[c]
+            result = connection.execute(table_.insert(), values)
+            ret = values.copy()
+
+            ipk = result.inserted_primary_key
+            for col, id_ in zip(table_.primary_key, ipk):
+                ret[col.key] = id_
+
+            if result.lastrow_has_defaults():
+                criterion = and_(
+                    *[
+                        col == id_
+                        for col, id_ in zip(
+                            table_.primary_key, result.inserted_primary_key
+                        )
+                    ]
+                )
+                row = connection.execute(
+                    table_.select().where(criterion)
+                ).first()
+                for c in table_.c:
+                    ret[c.key] = row._mapping[c]
             return ret, ipk
 
-        if testing.against("postgresql", "oracle", "mssql"):
-            assert testing.db.dialect.implicit_returning
-
-        if testing.db.dialect.implicit_returning:
-            test_engines = [
-                engines.testing_engine(options={"implicit_returning": False}),
-                engines.testing_engine(options={"implicit_returning": True}),
-            ]
-        else:
-            test_engines = [testing.db]
-
-        for engine in test_engines:
-            try:
-                table_.create(bind=engine, checkfirst=True)
-                i, ipk = insert_values(engine, table_, values)
-                eq_(i, assertvalues)
+        table_.create(connection, checkfirst=True)
+        i, ipk = insert_values(table_, values)
+        eq_(i, assertvalues)
 
-                # named tuple tests
-                for col in table_.primary_key:
-                    eq_(getattr(ipk, col.key), assertvalues[col.key])
-                    eq_(ipk._mapping[col.key], assertvalues[col.key])
+        # named tuple tests
+        for col in table_.primary_key:
+            eq_(getattr(ipk, col.key), assertvalues[col.key])
+            eq_(ipk._mapping[col.key], assertvalues[col.key])
 
-                eq_(
-                    ipk._fields, tuple([col.key for col in table_.primary_key])
-                )
-
-            finally:
-                table_.drop(bind=engine)
+        eq_(ipk._fields, tuple([col.key for col in table_.primary_key]))
 
-    @testing.skip_if("sqlite")
-    def test_lastrow_accessor_one(self):
-        metadata = MetaData()
+    @testing.requires.supports_autoincrement_w_composite_pk
+    @testing.combinations(
+        (True, testing.requires.returning),
+        (False,),
+        argnames="implicit_returning",
+    )
+    def test_lastrow_accessor_one(
+        self, metadata, connection, implicit_returning
+    ):
         self._test_lastrow_accessor(
+            connection,
             Table(
                 "t1",
                 metadata,
@@ -175,15 +165,23 @@ class InsertExecTest(fixtures.TablesTest):
                     test_needs_autoincrement=True,
                 ),
                 Column("foo", String(30), primary_key=True),
+                implicit_returning=implicit_returning,
             ),
             {"foo": "hi"},
             {"id": 1, "foo": "hi"},
         )
 
-    @testing.skip_if("sqlite")
-    def test_lastrow_accessor_two(self):
-        metadata = MetaData()
+    @testing.requires.supports_autoincrement_w_composite_pk
+    @testing.combinations(
+        (True, testing.requires.returning),
+        (False,),
+        argnames="implicit_returning",
+    )
+    def test_lastrow_accessor_two(
+        self, metadata, connection, implicit_returning
+    ):
         self._test_lastrow_accessor(
+            connection,
             Table(
                 "t2",
                 metadata,
@@ -195,29 +193,45 @@ class InsertExecTest(fixtures.TablesTest):
                 ),
                 Column("foo", String(30), primary_key=True),
                 Column("bar", String(30), server_default="hi"),
+                implicit_returning=implicit_returning,
             ),
             {"foo": "hi"},
             {"id": 1, "foo": "hi", "bar": "hi"},
         )
 
-    def test_lastrow_accessor_three(self):
-        metadata = MetaData()
+    @testing.combinations(
+        (True, testing.requires.returning),
+        (False,),
+        argnames="implicit_returning",
+    )
+    def test_lastrow_accessor_three(
+        self, metadata, connection, implicit_returning
+    ):
         self._test_lastrow_accessor(
+            connection,
             Table(
                 "t3",
                 metadata,
                 Column("id", String(40), primary_key=True),
                 Column("foo", String(30), primary_key=True),
                 Column("bar", String(30)),
+                implicit_returning=implicit_returning,
             ),
             {"id": "hi", "foo": "thisisfoo", "bar": "thisisbar"},
             {"id": "hi", "foo": "thisisfoo", "bar": "thisisbar"},
         )
 
     @testing.requires.sequences
-    def test_lastrow_accessor_four(self):
-        metadata = MetaData()
+    @testing.combinations(
+        (True, testing.requires.returning),
+        (False,),
+        argnames="implicit_returning",
+    )
+    def test_lastrow_accessor_four(
+        self, metadata, connection, implicit_returning
+    ):
         self._test_lastrow_accessor(
+            connection,
             Table(
                 "t4",
                 metadata,
@@ -229,15 +243,23 @@ class InsertExecTest(fixtures.TablesTest):
                 ),
                 Column("foo", String(30), primary_key=True),
                 Column("bar", String(30), server_default="hi"),
+                implicit_returning=implicit_returning,
             ),
             {"foo": "hi", "id": 1},
             {"id": 1, "foo": "hi", "bar": "hi"},
         )
 
     @testing.requires.sequences
-    def test_lastrow_accessor_four_a(self):
-        metadata = MetaData()
+    @testing.combinations(
+        (True, testing.requires.returning),
+        (False,),
+        argnames="implicit_returning",
+    )
+    def test_lastrow_accessor_four_a(
+        self, metadata, connection, implicit_returning
+    ):
         self._test_lastrow_accessor(
+            connection,
             Table(
                 "t4",
                 metadata,
@@ -248,28 +270,44 @@ class InsertExecTest(fixtures.TablesTest):
                     primary_key=True,
                 ),
                 Column("foo", String(30)),
+                implicit_returning=implicit_returning,
             ),
             {"foo": "hi"},
             {"id": 1, "foo": "hi"},
         )
 
-    def test_lastrow_accessor_five(self):
-        metadata = MetaData()
+    @testing.combinations(
+        (True, testing.requires.returning),
+        (False,),
+        argnames="implicit_returning",
+    )
+    def test_lastrow_accessor_five(
+        self, metadata, connection, implicit_returning
+    ):
         self._test_lastrow_accessor(
+            connection,
             Table(
                 "t5",
                 metadata,
                 Column("id", String(10), primary_key=True),
                 Column("bar", String(30), server_default="hi"),
+                implicit_returning=implicit_returning,
             ),
             {"id": "id1"},
             {"id": "id1", "bar": "hi"},
         )
 
-    @testing.skip_if("sqlite")
-    def test_lastrow_accessor_six(self):
-        metadata = MetaData()
+    @testing.requires.supports_autoincrement_w_composite_pk
+    @testing.combinations(
+        (True, testing.requires.returning),
+        (False,),
+        argnames="implicit_returning",
+    )
+    def test_lastrow_accessor_six(
+        self, metadata, connection, implicit_returning
+    ):
         self._test_lastrow_accessor(
+            connection,
             Table(
                 "t6",
                 metadata,
@@ -280,6 +318,7 @@ class InsertExecTest(fixtures.TablesTest):
                     test_needs_autoincrement=True,
                 ),
                 Column("bar", Integer, primary_key=True),
+                implicit_returning=implicit_returning,
             ),
             {"bar": 0},
             {"id": 1, "bar": 0},
@@ -287,35 +326,28 @@ class InsertExecTest(fixtures.TablesTest):
 
     # TODO: why not in the sqlite suite?
     @testing.only_on("sqlite+pysqlite")
-    @testing.provide_metadata
-    def test_lastrowid_zero(self):
+    def test_lastrowid_zero(self, metadata, connection):
         from sqlalchemy.dialects import sqlite
 
-        eng = engines.testing_engine()
-
         class ExcCtx(sqlite.base.SQLiteExecutionContext):
             def get_lastrowid(self):
                 return 0
 
-        eng.dialect.execution_ctx_cls = ExcCtx
         t = Table(
             "t",
             self.metadata,
             Column("x", Integer, primary_key=True),
             Column("y", Integer),
         )
-        with eng.begin() as conn:
-            t.create(conn)
-            r = conn.execute(t.insert().values(y=5))
+        t.create(connection)
+        with mock.patch.object(
+            connection.dialect, "execution_ctx_cls", ExcCtx
+        ):
+            r = connection.execute(t.insert().values(y=5))
             eq_(r.inserted_primary_key, (0,))
 
-    @testing.fails_on(
-        "sqlite", "sqlite autoincrement doesn't work with composite pks"
-    )
-    @testing.provide_metadata
-    def test_misordered_lastrow(self, connection):
-        metadata = self.metadata
-
+    @testing.requires.supports_autoincrement_w_composite_pk
+    def test_misordered_lastrow(self, connection, metadata):
         related = Table(
             "related",
             metadata,
diff --git a/test/sql/test_returning.py b/test/sql/test_returning.py
index 4069416d4..138e7a4c6 100644
--- a/test/sql/test_returning.py
+++ b/test/sql/test_returning.py
@@ -16,7 +16,6 @@ from sqlalchemy import update
 from sqlalchemy.testing import assert_raises_message
 from sqlalchemy.testing import AssertsCompiledSQL
 from sqlalchemy.testing import AssertsExecutionResults
-from sqlalchemy.testing import engines
 from sqlalchemy.testing import eq_
 from sqlalchemy.testing import fixtures
 from sqlalchemy.testing.schema import Column
@@ -767,35 +766,3 @@ class ReturnDefaultsTest(fixtures.TablesTest):
             result.inserted_primary_key_rows,
             [(10,), (11,), (12,), (13,), (14,), (15,)],
         )
-
-
-class ImplicitReturningFlag(fixtures.TestBase):
-    __backend__ = True
-
-    def test_flag_turned_off(self):
-        e = engines.testing_engine(options={"implicit_returning": False})
-        assert e.dialect.implicit_returning is False
-        c = e.connect()
-        c.close()
-        assert e.dialect.implicit_returning is False
-
-    def test_flag_turned_on(self):
-        e = engines.testing_engine(options={"implicit_returning": True})
-        assert e.dialect.implicit_returning is True
-        c = e.connect()
-        c.close()
-        assert e.dialect.implicit_returning is True
-
-    def test_flag_turned_default(self):
-        supports = [False]
-
-        def go():
-            supports[0] = True
-
-        testing.requires.returning(go)()
-        e = engines.testing_engine()
-
-        # version detection on connect sets it
-        c = e.connect()
-        c.close()
-        assert e.dialect.implicit_returning is supports[0]
diff --git a/test/sql/test_sequences.py b/test/sql/test_sequences.py
index d6906f9e6..c5c76ad08 100644
--- a/test/sql/test_sequences.py
+++ b/test/sql/test_sequences.py
@@ -11,7 +11,6 @@ from sqlalchemy.schema import CreateSequence
 from sqlalchemy.schema import DropSequence
 from sqlalchemy.sql import select
 from sqlalchemy.testing import assert_raises_message
-from sqlalchemy.testing import engines
 from sqlalchemy.testing import eq_
 from sqlalchemy.testing import fixtures
 from sqlalchemy.testing import is_false
@@ -180,29 +179,30 @@ class SequenceExecTest(fixtures.TestBase):
         connection.execute(t1.insert().values(x=s.next_value()))
         self._assert_seq_result(connection.scalar(t1.select()))
 
-    @testing.provide_metadata
-    def test_inserted_pk_no_returning(self):
+    def test_inserted_pk_no_returning(self, metadata, connection):
         """test inserted_primary_key contains [None] when
         pk_col=next_value(), implicit returning is not used."""
 
         # I'm not really sure what this test wants to accomlish.
 
-        metadata = self.metadata
-        t1 = Table("t", metadata, Column("x", Integer, primary_key=True))
+        t1 = Table(
+            "t",
+            metadata,
+            Column("x", Integer, primary_key=True),
+            implicit_returning=False,
+        )
         s = Sequence("my_sequence_here", metadata=metadata)
 
-        e = engines.testing_engine(options={"implicit_returning": False})
-        with e.begin() as conn:
-
-            t1.create(conn)
-            s.create(conn)
+        conn = connection
+        t1.create(conn)
+        s.create(conn)
 
-            r = conn.execute(t1.insert().values(x=s.next_value()))
+        r = conn.execute(t1.insert().values(x=s.next_value()))
 
-            if testing.requires.emulated_lastrowid_even_with_sequences.enabled:
-                eq_(r.inserted_primary_key, (1,))
-            else:
-                eq_(r.inserted_primary_key, (None,))
+        if testing.requires.emulated_lastrowid_even_with_sequences.enabled:
+            eq_(r.inserted_primary_key, (1,))
+        else:
+            eq_(r.inserted_primary_key, (None,))
 
     @testing.combinations(
         ("implicit_returning",),
@@ -213,43 +213,40 @@ class SequenceExecTest(fixtures.TestBase):
         argnames="returning",
     )
     @testing.requires.multivalues_inserts
-    def test_seq_multivalues_inline(self, metadata, testing_engine, returning):
+    def test_seq_multivalues_inline(self, metadata, connection, returning):
+        _implicit_returning = "no_implicit_returning" not in returning
         t1 = Table(
             "t",
             metadata,
             Column("x", Integer, Sequence("my_seq"), primary_key=True),
             Column("data", String(50)),
+            implicit_returning=_implicit_returning,
         )
 
-        e = engines.testing_engine(
-            options={
-                "implicit_returning": "no_implicit_returning" not in returning
-            }
+        metadata.create_all(connection)
+        conn = connection
+
+        stmt = t1.insert().values(
+            [{"data": "d1"}, {"data": "d2"}, {"data": "d3"}]
         )
-        metadata.create_all(e)
-        with e.begin() as conn:
+        if returning == "explicit_returning":
+            stmt = stmt.returning(t1.c.x)
+        elif "return_defaults" in returning:
+            stmt = stmt.return_defaults()
 
-            stmt = t1.insert().values(
-                [{"data": "d1"}, {"data": "d2"}, {"data": "d3"}]
-            )
-            if returning == "explicit_returning":
-                stmt = stmt.returning(t1.c.x)
-            elif "return_defaults" in returning:
-                stmt = stmt.return_defaults()
-
-            r = conn.execute(stmt)
-            if returning == "explicit_returning":
-                eq_(r.all(), [(1,), (2,), (3,)])
-            elif "return_defaults" in returning:
-                eq_(r.returned_defaults_rows, None)
+        r = conn.execute(stmt)
+        if returning == "explicit_returning":
+            eq_(r.all(), [(1,), (2,), (3,)])
+        elif "return_defaults" in returning:
+            eq_(r.returned_defaults_rows, None)
 
-                # TODO: not sure what this is
-                eq_(r.inserted_primary_key_rows, [(None,)])
+            # TODO: not sure what this is
+            eq_(r.inserted_primary_key_rows, [(None,)])
 
-            eq_(
-                conn.execute(t1.select().order_by(t1.c.x)).all(),
-                [(1, "d1"), (2, "d2"), (3, "d3")],
-            )
+        eq_(
+            conn.execute(t1.select().order_by(t1.c.x)).all(),
+            [(1, "d1"), (2, "d2"), (3, "d3")],
+        )
 
     @testing.combinations(
         ("implicit_returning",),
@@ -272,54 +269,49 @@ class SequenceExecTest(fixtures.TestBase):
         argnames="returning",
     )
     def test_seq_multivalues_executemany(
-        self, metadata, testing_engine, returning
+        self, connection, metadata, returning
     ):
+        _implicit_returning = "no_implicit_returning" not in returning
         t1 = Table(
             "t",
             metadata,
             Column("x", Integer, Sequence("my_seq"), primary_key=True),
             Column("data", String(50)),
+            implicit_returning=_implicit_returning,
         )
 
-        e = engines.testing_engine(
-            options={
-                "implicit_returning": "no_implicit_returning" not in returning
-            }
-        )
-        metadata.create_all(e)
-        with e.begin() as conn:
+        metadata.create_all(connection)
+        conn = connection
 
-            stmt = t1.insert()
-            if returning == "explicit_returning":
-                stmt = stmt.returning(t1.c.x)
-            elif "return_defaults" in returning:
-                stmt = stmt.return_defaults()
+        stmt = t1.insert()
+        if returning == "explicit_returning":
+            stmt = stmt.returning(t1.c.x)
+        elif "return_defaults" in returning:
+            stmt = stmt.return_defaults()
 
-            r = conn.execute(
-                stmt, [{"data": "d1"}, {"data": "d2"}, {"data": "d3"}]
-            )
-            if returning == "explicit_returning":
-                eq_(r.all(), [(1,), (2,), (3,)])
-            elif "return_defaults" in returning:
-                if "no_implicit_returning" in returning:
-                    eq_(r.returned_defaults_rows, None)
-                    eq_(r.inserted_primary_key_rows, [(1,), (2,), (3,)])
-                else:
-                    eq_(r.returned_defaults_rows, [(1,), (2,), (3,)])
-                    eq_(r.inserted_primary_key_rows, [(1,), (2,), (3,)])
-
-            eq_(
-                conn.execute(t1.select().order_by(t1.c.x)).all(),
-                [(1, "d1"), (2, "d2"), (3, "d3")],
-            )
+        r = conn.execute(
+            stmt, [{"data": "d1"}, {"data": "d2"}, {"data": "d3"}]
+        )
+        if returning == "explicit_returning":
+            eq_(r.all(), [(1,), (2,), (3,)])
+        elif "return_defaults" in returning:
+            if "no_implicit_returning" in returning:
+                eq_(r.returned_defaults_rows, None)
+                eq_(r.inserted_primary_key_rows, [(1,), (2,), (3,)])
+            else:
+                eq_(r.returned_defaults_rows, [(1,), (2,), (3,)])
+                eq_(r.inserted_primary_key_rows, [(1,), (2,), (3,)])
+
+        eq_(
+            conn.execute(t1.select().order_by(t1.c.x)).all(),
+            [(1, "d1"), (2, "d2"), (3, "d3")],
+        )
 
     @testing.requires.returning
-    @testing.provide_metadata
-    def test_inserted_pk_implicit_returning(self):
+    def test_inserted_pk_implicit_returning(self, connection, metadata):
         """test inserted_primary_key contains the result when
         pk_col=next_value(), when implicit returning is used."""
 
-        metadata = self.metadata
         s = Sequence("my_sequence")
         t1 = Table(
             "t",
@@ -329,13 +321,12 @@ class SequenceExecTest(fixtures.TestBase):
                 Integer,
                 primary_key=True,
             ),
+            implicit_returning=True,
         )
-        t1.create(testing.db)
+        t1.create(connection)
 
-        e = engines.testing_engine(options={"implicit_returning": True})
-        with e.begin() as conn:
-            r = conn.execute(t1.insert().values(x=s.next_value()))
-            self._assert_seq_result(r.inserted_primary_key[0])
+        r = connection.execute(t1.insert().values(x=s.next_value()))
+        self._assert_seq_result(r.inserted_primary_key[0])
 
 
 class SequenceTest(fixtures.TestBase, testing.AssertsCompiledSQL):
@@ -501,42 +492,56 @@ class TableBoundSequenceTest(fixtures.TablesTest):
     __requires__ = ("sequences",)
     __backend__ = True
 
-    @classmethod
-    def define_tables(cls, metadata):
-        Table(
-            "cartitems",
-            metadata,
-            Column(
-                "cart_id",
-                Integer,
-                Sequence("cart_id_seq"),
-                primary_key=True,
-                autoincrement=False,
-            ),
-            Column("description", String(40)),
-            Column("createdate", sa.DateTime()),
-        )
+    @testing.fixture
+    def table_fixture(self, metadata, connection, implicit_returning):
+        def go(implicit_returning):
+            cartitems = Table(
+                "cartitems",
+                metadata,
+                Column(
+                    "cart_id",
+                    Integer,
+                    Sequence("cart_id_seq"),
+                    primary_key=True,
+                    autoincrement=False,
+                ),
+                Column("description", String(40)),
+                Column("createdate", sa.DateTime()),
+                implicit_returning=implicit_returning,
+            )
 
-        # a little bit of implicit case sensitive naming test going on here
-        Table(
-            "Manager",
-            metadata,
-            Column(
-                "obj_id",
-                Integer,
-                Sequence("obj_id_seq"),
-            ),
-            Column("name", String(128)),
-            Column(
-                "id",
-                Integer,
-                Sequence("Manager_id_seq", optional=True),
-                primary_key=True,
-            ),
-        )
+            # a little bit of implicit case sensitive naming test going on here
+            Manager = Table(
+                "Manager",
+                metadata,
+                Column(
+                    "obj_id",
+                    Integer,
+                    Sequence("obj_id_seq"),
+                ),
+                Column("name", String(128)),
+                Column(
+                    "id",
+                    Integer,
+                    Sequence("Manager_id_seq", optional=True),
+                    primary_key=True,
+                ),
+                implicit_returning=implicit_returning,
+            )
+            metadata.create_all(connection)
+            return Manager, cartitems
 
-    def test_insert_via_seq(self, connection):
-        cartitems = self.tables.cartitems
+        return go
+
+    @testing.combinations(
+        (True, testing.requires.returning),
+        (False,),
+        argnames="implicit_returning",
+    )
+    def test_insert_via_seq(
+        self, table_fixture, connection, implicit_returning
+    ):
+        Manager, cartitems = table_fixture(implicit_returning)
 
         connection.execute(cartitems.insert(), dict(description="hi"))
         connection.execute(cartitems.insert(), dict(description="there"))
@@ -554,52 +559,53 @@ class TableBoundSequenceTest(fixtures.TablesTest):
             expected,
         )
 
-    def test_seq_nonpk(self):
+    @testing.combinations(
+        (True, testing.requires.returning),
+        (False,),
+        argnames="implicit_returning",
+    )
+    def test_seq_nonpk(self, connection, table_fixture, implicit_returning):
         """test sequences fire off as defaults on non-pk columns"""
 
-        sometable = self.tables.Manager
-
-        engine = engines.testing_engine(options={"implicit_returning": False})
+        sometable, cartitems = table_fixture(implicit_returning)
 
-        with engine.begin() as conn:
-            result = conn.execute(sometable.insert(), dict(name="somename"))
+        conn = connection
+        result = conn.execute(sometable.insert(), dict(name="somename"))
 
-            eq_(result.postfetch_cols(), [sometable.c.obj_id])
+        eq_(result.postfetch_cols(), [sometable.c.obj_id])
 
-            result = conn.execute(sometable.insert(), dict(name="someother"))
+        result = conn.execute(sometable.insert(), dict(name="someother"))
 
-            conn.execute(
-                sometable.insert(), [{"name": "name3"}, {"name": "name4"}]
-            )
+        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))
+        dsb = testing.db.dialect.default_sequence_base
+        eq_(
+            list(conn.execute(sometable.select().order_by(sometable.c.id))),
+            [
+                (
+                    dsb,
+                    "somename",
+                    dsb,
                 ),
-                [
-                    (
-                        dsb,
-                        "somename",
-                        dsb,
-                    ),
-                    (
-                        dsb + 1,
-                        "someother",
-                        dsb + 1,
-                    ),
-                    (
-                        dsb + 2,
-                        "name3",
-                        dsb + 2,
-                    ),
-                    (
-                        dsb + 3,
-                        "name4",
-                        dsb + 3,
-                    ),
-                ],
-            )
+                (
+                    dsb + 1,
+                    "someother",
+                    dsb + 1,
+                ),
+                (
+                    dsb + 2,
+                    "name3",
+                    dsb + 2,
+                ),
+                (
+                    dsb + 3,
+                    "name4",
+                    dsb + 3,
+                ),
+            ],
+        )
 
 
 class SequenceAsServerDefaultTest(