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

1 files changed, 118 insertions, 86 deletions

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,