Merge "Deprecate create_engine.implicit_returning" into main

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 229a54b95..f9f23c963 100644
--- a/lib/sqlalchemy/dialects/oracle/base.py
+++ b/lib/sqlalchemy/dialects/oracle/base.py
@@ -1440,6 +1440,8 @@ class OracleDialect(default.DefaultDialect):
     supports_alter = True
     max_identifier_length = 128
 
+    implicit_returning = True
+
     supports_simple_order_by_label = False
     cte_follows_insert = True
 
@@ -1505,9 +1507,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 e6da1d8e6..efcd2b530 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.
@@ -278,8 +283,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
@@ -339,13 +342,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 ccf771f81..52162a286 100644
--- a/test/dialect/oracle/test_dialect.py
+++ b/test/dialect/oracle/test_dialect.py
@@ -525,17 +525,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 c12f4a50a..47a4728b4 100644
--- a/test/dialect/postgresql/test_dialect.py
+++ b/test/dialect/postgresql/test_dialect.py
@@ -1192,6 +1192,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 956524abd..d0cbe1a77 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 750a2dd98..3ca763c32 100644
--- a/test/engine/test_execute.py
+++ b/test/engine/test_execute.py
@@ -2014,10 +2014,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)
@@ -2032,6 +2030,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 92e59d9a8..2011744fb 100644
--- a/test/sql/test_defaults.py
+++ b/test/sql/test_defaults.py
@@ -19,7 +19,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
@@ -918,68 +917,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(