Generalize RETURNING and suppor for MariaDB / SQLite

As almost every dialect supports RETURNING now, RETURNING
is also made more of a default assumption.

* the default compiler generates a RETURNING clause now
  when specified; CompileError is no longer raised.
* The dialect-level implicit_returning parameter now has
  no effect.   It's not fully clear if there are real world
  cases relying on the dialect-level parameter, so we will see
  once 2.0 is released.   ORM-level RETURNING can be disabled
  at the table level, and perhaps "implicit returning" should
  become an ORM-level option at some point as that's where
  it applies.
* Altered ORM update() / delete() to respect table-level
  implicit returning for fetch.
* Since MariaDB doesnt support UPDATE returning, "full_returning"
  is now split into insert_returning, update_returning, delete_returning
* Crazy new thing.  Dialects that have *both* cursor.lastrowid
  *and* returning.   so now we can pick between them for SQLite
  and mariadb.  so, we are trying to keep it on .lastrowid for
  simple inserts with an autoincrement column, this helps with
  some edge case test scenarios and i bet .lastrowid is faster
  anyway.  any return_defaults() / multiparams etc then we
  use returning
* SQLite decided they dont want to return rows that match in
  ON CONFLICT.  this is flat out wrong, but for now we need to
  work with it.

Fixes: #6195
Fixes: #7011
Closes: #7047
Pull-request: https://github.com/sqlalchemy/sqlalchemy/pull/7047
Pull-request-sha: d25d5ea3abe094f282c53c7dd87f5f53a9e85248

Co-authored-by: Mike Bayer <mike_mp@zzzcomputing.com>
Change-Id: I9908ce0ff7bdc50bd5b27722081767c31c19a950

23 files changed, 379 insertions, 154 deletions

diff --git a/lib/sqlalchemy/dialects/mssql/base.py b/lib/sqlalchemy/dialects/mssql/base.py
index 4295e0ed0..12f495d6e 100644
--- a/lib/sqlalchemy/dialects/mssql/base.py
+++ b/lib/sqlalchemy/dialects/mssql/base.py
@@ -2807,8 +2807,9 @@ class MSDialect(default.DefaultDialect):
     max_identifier_length = 128
     schema_name = "dbo"
 
-    implicit_returning = True
-    full_returning = True
+    insert_returning = True
+    update_returning = True
+    delete_returning = True
 
     colspecs = {
         sqltypes.DateTime: _MSDateTime,
diff --git a/lib/sqlalchemy/dialects/mssql/pyodbc.py b/lib/sqlalchemy/dialects/mssql/pyodbc.py
index 28cca56f7..6d64fdc3e 100644
--- a/lib/sqlalchemy/dialects/mssql/pyodbc.py
+++ b/lib/sqlalchemy/dialects/mssql/pyodbc.py
@@ -522,6 +522,8 @@ class MSDialect_pyodbc(PyODBCConnector, MSDialect):
     # mssql still has problems with this on Linux
     supports_sane_rowcount_returning = False
 
+    favor_returning_over_lastrowid = True
+
     execution_ctx_cls = MSExecutionContext_pyodbc
 
     colspecs = util.update_copy(
diff --git a/lib/sqlalchemy/dialects/mysql/base.py b/lib/sqlalchemy/dialects/mysql/base.py
index b585ea992..68653d976 100644
--- a/lib/sqlalchemy/dialects/mysql/base.py
+++ b/lib/sqlalchemy/dialects/mysql/base.py
@@ -488,6 +488,37 @@ available.
 
         :class:`_mysql.match`
 
+INSERT/DELETE...RETURNING
+-------------------------
+
+The MariaDB dialect supports 10.5+'s ``INSERT..RETURNING`` and
+``DELETE..RETURNING`` (10.0+) syntaxes.   ``INSERT..RETURNING`` may be used
+automatically in some cases in order to fetch newly generated identifiers in
+place of the traditional approach of using ``cursor.lastrowid``, however
+``cursor.lastrowid`` is currently still preferred for simple single-statement
+cases for its better performance.
+
+To specify an explicit ``RETURNING`` clause, use the
+:meth:`._UpdateBase.returning` method on a per-statement basis::
+
+    # INSERT..RETURNING
+    result = connection.execute(
+        table.insert().
+        values(name='foo').
+        returning(table.c.col1, table.c.col2)
+    )
+    print(result.all())
+
+    # DELETE..RETURNING
+    result = connection.execute(
+        table.delete().
+        where(table.c.name=='foo').
+        returning(table.c.col1, table.c.col2)
+    )
+    print(result.all())
+
+.. versionadded:: 2.0  Added support for MariaDB RETURNING
+
 .. _mysql_insert_on_duplicate_key_update:
 
 INSERT...ON DUPLICATE KEY UPDATE (Upsert)
@@ -2500,7 +2531,9 @@ class MySQLDialect(default.DefaultDialect):
 
         server_version_info = tuple(version)
 
-        self._set_mariadb(server_version_info and is_mariadb, val)
+        self._set_mariadb(
+            server_version_info and is_mariadb, server_version_info
+        )
 
         if not is_mariadb:
             self._mariadb_normalized_version_info = server_version_info
@@ -2522,7 +2555,7 @@ class MySQLDialect(default.DefaultDialect):
         if not is_mariadb and self.is_mariadb:
             raise exc.InvalidRequestError(
                 "MySQL version %s is not a MariaDB variant."
-                % (server_version_info,)
+                % (".".join(map(str, server_version_info)),)
             )
         if is_mariadb:
             self.preparer = MariaDBIdentifierPreparer
@@ -2717,6 +2750,14 @@ class MySQLDialect(default.DefaultDialect):
             not self.is_mariadb and self.server_version_info >= (8,)
         )
 
+        self.delete_returning = (
+            self.is_mariadb and self.server_version_info >= (10, 0, 5)
+        )
+
+        self.insert_returning = (
+            self.is_mariadb and self.server_version_info >= (10, 5)
+        )
+
         self._warn_for_known_db_issues()
 
     def _warn_for_known_db_issues(self):
diff --git a/lib/sqlalchemy/dialects/oracle/base.py b/lib/sqlalchemy/dialects/oracle/base.py
index 37b81e1dd..faac0deb7 100644
--- a/lib/sqlalchemy/dialects/oracle/base.py
+++ b/lib/sqlalchemy/dialects/oracle/base.py
@@ -293,40 +293,16 @@ added in a future release.
 RETURNING Support
 -----------------
 
-The Oracle database supports a limited form of RETURNING, in order to retrieve
-result sets of matched rows from INSERT, UPDATE and DELETE statements.
-Oracle's RETURNING..INTO syntax only supports one row being returned, as it
-relies upon OUT parameters in order to function.  In addition, supported
-DBAPIs have further limitations (see :ref:`cx_oracle_returning`).
+The Oracle database supports RETURNING fully for INSERT, UPDATE and DELETE
+statements that are invoked with a single collection of bound parameters
+(that is, a ``cursor.execute()`` style statement; SQLAlchemy does not generally
+support RETURNING with :term:`executemany` statements).  Multiple rows may be
+returned as well.
 
-SQLAlchemy's "implicit returning" feature, which employs RETURNING within an
-INSERT and sometimes an UPDATE statement in order to fetch newly generated
-primary key values and other SQL defaults and expressions, is normally enabled
-on the Oracle backend.  By default, "implicit returning" typically only
-fetches the value of a single ``nextval(some_seq)`` expression embedded into
-an INSERT in order to increment a sequence within an INSERT statement and get
-the value back at the same time. To disable this feature across the board,
-specify ``implicit_returning=False`` to :func:`_sa.create_engine`::
+.. versionchanged:: 2.0  the Oracle backend has full support for RETURNING
+   on parity with other backends.
 
-    engine = create_engine("oracle+cx_oracle://scott:tiger@dsn",
-                           implicit_returning=False)
 
-Implicit returning can also be disabled on a table-by-table basis as a table
-option::
-
-    # Core Table
-    my_table = Table("my_table", metadata, ..., implicit_returning=False)
-
-
-    # declarative
-    class MyClass(Base):
-        __tablename__ = 'my_table'
-        __table_args__ = {"implicit_returning": False}
-
-.. seealso::
-
-    :ref:`cx_oracle_returning` - additional cx_oracle-specific restrictions on
-    implicit returning.
 
 ON UPDATE CASCADE
 -----------------
@@ -1572,8 +1548,9 @@ class OracleDialect(default.DefaultDialect):
     supports_alter = True
     max_identifier_length = 128
 
-    implicit_returning = True
-    full_returning = True
+    insert_returning = True
+    update_returning = True
+    delete_returning = True
 
     div_is_floordiv = False
 
diff --git a/lib/sqlalchemy/dialects/postgresql/base.py b/lib/sqlalchemy/dialects/postgresql/base.py
index 146e59c4d..83e46151f 100644
--- a/lib/sqlalchemy/dialects/postgresql/base.py
+++ b/lib/sqlalchemy/dialects/postgresql/base.py
@@ -44,8 +44,6 @@ subsequent insert. Note that when an
 apply; no RETURNING clause is emitted nor is the sequence pre-executed in this
 case.
 
-To force the usage of RETURNING by default off, specify the flag
-``implicit_returning=False`` to :func:`_sa.create_engine`.
 
 PostgreSQL 10 and above IDENTITY columns
 ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
@@ -2351,16 +2349,6 @@ class PGCompiler(compiler.SQLCompiler):
 
         return tmp
 
-    def returning_clause(
-        self, stmt, returning_cols, *, populate_result_map, **kw
-    ):
-        columns = [
-            self._label_returning_column(stmt, c, populate_result_map)
-            for c in expression._select_iterables(returning_cols)
-        ]
-
-        return "RETURNING " + ", ".join(columns)
-
     def visit_substring_func(self, func, **kw):
         s = self.process(func.clauses.clauses[0], **kw)
         start = self.process(func.clauses.clauses[1], **kw)
@@ -3207,8 +3195,9 @@ class PGDialect(default.DefaultDialect):
     execution_ctx_cls = PGExecutionContext
     inspector = PGInspector
 
-    implicit_returning = True
-    full_returning = True
+    update_returning = True
+    delete_returning = True
+    insert_returning = True
 
     connection_characteristics = (
         default.DefaultDialect.connection_characteristics
@@ -3274,7 +3263,9 @@ class PGDialect(default.DefaultDialect):
         super(PGDialect, self).initialize(connection)
 
         if self.server_version_info <= (8, 2):
-            self.full_returning = self.implicit_returning = False
+            self.delete_returning = (
+                self.update_returning
+            ) = self.insert_returning = False
 
         self.supports_native_enum = self.server_version_info >= (8, 3)
         if not self.supports_native_enum:
diff --git a/lib/sqlalchemy/dialects/postgresql/psycopg.py b/lib/sqlalchemy/dialects/postgresql/psycopg.py
index 7ec26cb4e..90bae61e1 100644
--- a/lib/sqlalchemy/dialects/postgresql/psycopg.py
+++ b/lib/sqlalchemy/dialects/postgresql/psycopg.py
@@ -256,7 +256,7 @@ class PGDialect_psycopg(_PGDialect_common_psycopg):
 
         # PGDialect.initialize() checks server version for <= 8.2 and sets
         # this flag to False if so
-        if not self.full_returning:
+        if not self.insert_returning:
             self.insert_executemany_returning = False
 
         # HSTORE can't be registered until we have a connection so that
diff --git a/lib/sqlalchemy/dialects/postgresql/psycopg2.py b/lib/sqlalchemy/dialects/postgresql/psycopg2.py
index f5d84a5a3..3f4ee2a20 100644
--- a/lib/sqlalchemy/dialects/postgresql/psycopg2.py
+++ b/lib/sqlalchemy/dialects/postgresql/psycopg2.py
@@ -613,7 +613,7 @@ class PGDialect_psycopg2(_PGDialect_common_psycopg):
 
         # PGDialect.initialize() checks server version for <= 8.2 and sets
         # this flag to False if so
-        if not self.full_returning:
+        if not self.insert_returning:
             self.insert_executemany_returning = False
             self.executemany_mode = EXECUTEMANY_PLAIN
 
diff --git a/lib/sqlalchemy/dialects/sqlite/base.py b/lib/sqlalchemy/dialects/sqlite/base.py
index 2ce298436..fdcd1340b 100644
--- a/lib/sqlalchemy/dialects/sqlite/base.py
+++ b/lib/sqlalchemy/dialects/sqlite/base.py
@@ -221,6 +221,46 @@ by *not even emitting BEGIN* until the first write operation.
 
     :ref:`dbapi_autocommit`
 
+INSERT/UPDATE/DELETE...RETURNING
+---------------------------------
+
+The SQLite dialect supports SQLite 3.35's  ``INSERT|UPDATE|DELETE..RETURNING``
+syntax.   ``INSERT..RETURNING`` may be used
+automatically in some cases in order to fetch newly generated identifiers in
+place of the traditional approach of using ``cursor.lastrowid``, however
+``cursor.lastrowid`` is currently still preferred for simple single-statement
+cases for its better performance.
+
+To specify an explicit ``RETURNING`` clause, use the
+:meth:`._UpdateBase.returning` method on a per-statement basis::
+
+    # INSERT..RETURNING
+    result = connection.execute(
+        table.insert().
+        values(name='foo').
+        returning(table.c.col1, table.c.col2)
+    )
+    print(result.all())
+
+    # UPDATE..RETURNING
+    result = connection.execute(
+        table.update().
+        where(table.c.name=='foo').
+        values(name='bar').
+        returning(table.c.col1, table.c.col2)
+    )
+    print(result.all())
+
+    # DELETE..RETURNING
+    result = connection.execute(
+        table.delete().
+        where(table.c.name=='foo').
+        returning(table.c.col1, table.c.col2)
+    )
+    print(result.all())
+
+.. versionadded:: 2.0  Added support for SQLite RETURNING
+
 SAVEPOINT Support
 ----------------------------
 
@@ -1280,6 +1320,19 @@ class SQLiteCompiler(compiler.SQLCompiler):
                 "%s is not a valid extract argument." % extract.field
             ) from err
 
+    def returning_clause(
+        self,
+        stmt,
+        returning_cols,
+        *,
+        populate_result_map,
+        **kw,
+    ):
+        kw["include_table"] = False
+        return super().returning_clause(
+            stmt, returning_cols, populate_result_map=populate_result_map, **kw
+        )
+
     def limit_clause(self, select, **kw):
         text = ""
         if select._limit_clause is not None:
@@ -1372,6 +1425,11 @@ class SQLiteCompiler(compiler.SQLCompiler):
 
         return target_text
 
+    def visit_insert(self, insert_stmt, **kw):
+        if insert_stmt._post_values_clause is not None:
+            kw["disable_implicit_returning"] = True
+        return super().visit_insert(insert_stmt, **kw)
+
     def visit_on_conflict_do_nothing(self, on_conflict, **kw):
 
         target_text = self._on_conflict_target(on_conflict, **kw)
@@ -1831,6 +1889,9 @@ class SQLiteDialect(default.DefaultDialect):
     supports_default_values = True
     supports_default_metavalue = False
 
+    # https://github.com/python/cpython/issues/93421
+    supports_sane_rowcount_returning = False
+
     supports_empty_insert = False
     supports_cast = True
     supports_multivalues_insert = True
@@ -1944,6 +2005,11 @@ class SQLiteDialect(default.DefaultDialect):
                 14,
             )
 
+            if self.dbapi.sqlite_version_info >= (3, 35):
+                self.update_returning = (
+                    self.delete_returning
+                ) = self.insert_returning = True
+
     _isolation_lookup = util.immutabledict(
         {"READ UNCOMMITTED": 1, "SERIALIZABLE": 0}
     )
diff --git a/lib/sqlalchemy/engine/create.py b/lib/sqlalchemy/engine/create.py
index 68a6b81e2..36119ab24 100644
--- a/lib/sqlalchemy/engine/create.py
+++ b/lib/sqlalchemy/engine/create.py
@@ -57,7 +57,7 @@ def create_engine(
     execution_options: _ExecuteOptions = ...,
     future: Literal[True],
     hide_parameters: bool = ...,
-    implicit_returning: bool = ...,
+    implicit_returning: Literal[True] = ...,
     isolation_level: _IsolationLevel = ...,
     json_deserializer: Callable[..., Any] = ...,
     json_serializer: Callable[..., Any] = ...,
@@ -266,18 +266,12 @@ def create_engine(url: Union[str, "_url.URL"], **kwargs: Any) -> Engine:
             :ref:`dbengine_logging` - further detail on how to configure
             logging.
 
-    :param implicit_returning=True:  Legacy flag that when set to ``False``
-        will disable the use of ``RETURNING`` on supporting backends where it
-        would normally be used to fetch newly generated primary key values for
-        single-row INSERT statements that do not otherwise specify a RETURNING
-        clause.  This behavior applies primarily to the PostgreSQL, Oracle,
-        SQL Server backends.
-
-        .. warning:: this flag originally allowed the "implicit returning"
-           feature to be *enabled* back when it was very new and there was not
-           well-established database support.  In modern SQLAlchemy, this flag
-           should **always be set to True**.  Some SQLAlchemy features will
-           fail to function properly if this flag is set to ``False``.
+    :param implicit_returning=True:  Legacy parameter that may only be set
+        to True. In SQLAlchemy 2.0, this parameter does nothing. In order to
+        disable "implicit returning" for statements invoked by the ORM,
+        configure this on a per-table basis using the
+        :paramref:`.Table.implicit_returning` parameter.
+
 
     :param isolation_level: optional string name of an isolation level
         which will be set on all new connections unconditionally.
diff --git a/lib/sqlalchemy/engine/cursor.py b/lib/sqlalchemy/engine/cursor.py
index ec1e1abe1..7947456af 100644
--- a/lib/sqlalchemy/engine/cursor.py
+++ b/lib/sqlalchemy/engine/cursor.py
@@ -1817,7 +1817,7 @@ class CursorResult(Result[_T]):
 
     def merge(self, *others: Result[Any]) -> MergedResult[Any]:
         merged_result = super().merge(*others)
-        setup_rowcounts = not self._metadata.returns_rows
+        setup_rowcounts = self.context._has_rowcount
         if setup_rowcounts:
             merged_result.rowcount = sum(
                 cast("CursorResult[Any]", result).rowcount
diff --git a/lib/sqlalchemy/engine/default.py b/lib/sqlalchemy/engine/default.py
index bcbe83f3f..6b76601ff 100644
--- a/lib/sqlalchemy/engine/default.py
+++ b/lib/sqlalchemy/engine/default.py
@@ -57,6 +57,7 @@ from ..sql.compiler import DDLCompiler
 from ..sql.compiler import SQLCompiler
 from ..sql.elements import quoted_name
 from ..sql.schema import default_is_scalar
+from ..util.typing import Literal
 
 if typing.TYPE_CHECKING:
     from types import ModuleType
@@ -135,9 +136,11 @@ class DefaultDialect(Dialect):
     preexecute_autoincrement_sequences = False
     supports_identity_columns = False
     postfetch_lastrowid = True
+    favor_returning_over_lastrowid = False
     insert_null_pk_still_autoincrements = False
-    implicit_returning = False
-    full_returning = False
+    update_returning = False
+    delete_returning = False
+    insert_returning = False
     insert_executemany_returning = False
 
     cte_follows_insert = False
@@ -258,7 +261,7 @@ class DefaultDialect(Dialect):
         paramstyle: Optional[_ParamStyle] = None,
         isolation_level: Optional[_IsolationLevel] = None,
         dbapi: Optional[ModuleType] = None,
-        implicit_returning: Optional[bool] = None,
+        implicit_returning: Literal[True] = True,
         supports_native_boolean: Optional[bool] = None,
         max_identifier_length: Optional[int] = None,
         label_length: Optional[int] = None,
@@ -296,8 +299,6 @@ class DefaultDialect(Dialect):
             self.paramstyle = self.dbapi.paramstyle
         else:
             self.paramstyle = self.default_paramstyle
-        if implicit_returning is not None:
-            self.implicit_returning = implicit_returning
         self.positional = self.paramstyle in ("qmark", "format", "numeric")
         self.identifier_preparer = self.preparer(self)
         self._on_connect_isolation_level = isolation_level
@@ -324,6 +325,18 @@ class DefaultDialect(Dialect):
         self.label_length = label_length
         self.compiler_linting = compiler_linting
 
+    @util.deprecated_property(
+        "2.0",
+        "full_returning is deprecated, please use insert_returning, "
+        "update_returning, delete_returning",
+    )
+    def full_returning(self):
+        return (
+            self.insert_returning
+            and self.update_returning
+            and self.delete_returning
+        )
+
     @util.memoized_property
     def loaded_dbapi(self) -> ModuleType:
         if self.dbapi is None:
@@ -771,7 +784,6 @@ class StrCompileDialect(DefaultDialect):
     supports_sequences = True
     sequences_optional = True
     preexecute_autoincrement_sequences = False
-    implicit_returning = False
 
     supports_native_boolean = True
 
@@ -806,6 +818,8 @@ class DefaultExecutionContext(ExecutionContext):
 
     _soft_closed = False
 
+    _has_rowcount = False
+
     # a hook for SQLite's translation of
     # result column names
     # NOTE: pyhive is using this hook, can't remove it :(
@@ -1450,6 +1464,7 @@ class DefaultExecutionContext(ExecutionContext):
             # is testing this, and psycopg will no longer return
             # rowcount after cursor is closed.
             result.rowcount
+            self._has_rowcount = True
 
             row = result.fetchone()
             if row is not None:
@@ -1465,7 +1480,12 @@ class DefaultExecutionContext(ExecutionContext):
             # no results, get rowcount
             # (which requires open cursor on some drivers)
             result.rowcount
+            self._has_rowcount = True
             result._soft_close()
+        elif self.isupdate or self.isdelete:
+            result.rowcount
+            self._has_rowcount = True
+
         return result
 
     @util.memoized_property
@@ -1479,7 +1499,6 @@ class DefaultExecutionContext(ExecutionContext):
         getter = cast(
             SQLCompiler, self.compiled
         )._inserted_primary_key_from_lastrowid_getter
-
         lastrowid = self.get_lastrowid()
         return [getter(lastrowid, self.compiled_parameters[0])]
 
diff --git a/lib/sqlalchemy/engine/interfaces.py b/lib/sqlalchemy/engine/interfaces.py
index 4020af354..cd6efb904 100644
--- a/lib/sqlalchemy/engine/interfaces.py
+++ b/lib/sqlalchemy/engine/interfaces.py
@@ -737,14 +737,32 @@ class Dialect(EventTarget):
       PostgreSQL.
     """
 
-    implicit_returning: bool
-    """For dialects that support RETURNING, indicate RETURNING may be used
-    to fetch newly generated primary key values and other defaults from
-    an INSERT statement automatically.
+    insert_returning: bool
+    """if the dialect supports RETURNING with INSERT
 
-    .. seealso::
+    .. versionadded:: 2.0
+
+    """
+
+    update_returning: bool
+    """if the dialect supports RETURNING with UPDATE
+
+    .. versionadded:: 2.0
+
+    """
+
+    delete_returning: bool
+    """if the dialect supports RETURNING with DELETE
+
+    .. versionadded:: 2.0
+
+    """
+
+    favor_returning_over_lastrowid: bool
+    """for backends that support both a lastrowid and a RETURNING insert
+    strategy, favor RETURNING for simple single-int pk inserts.
 
-        :paramref:`_schema.Table.implicit_returning`
+    cursor.lastrowid tends to be more performant on most backends.
 
     """
 
diff --git a/lib/sqlalchemy/ext/horizontal_shard.py b/lib/sqlalchemy/ext/horizontal_shard.py
index 5588fd587..7afe2343d 100644
--- a/lib/sqlalchemy/ext/horizontal_shard.py
+++ b/lib/sqlalchemy/ext/horizontal_shard.py
@@ -253,5 +253,4 @@ def execute_and_instances(orm_context):
         for shard_id in session.execute_chooser(orm_context):
             result_ = iter_for_shard(shard_id, load_options, update_options)
             partial.append(result_)
-
         return partial[0].merge(*partial[1:])
diff --git a/lib/sqlalchemy/orm/persistence.py b/lib/sqlalchemy/orm/persistence.py
index 56e7cca1a..0c035e7cf 100644
--- a/lib/sqlalchemy/orm/persistence.py
+++ b/lib/sqlalchemy/orm/persistence.py
@@ -39,6 +39,7 @@ from .. import exc as sa_exc
 from .. import future
 from .. import sql
 from .. import util
+from ..engine import Dialect
 from ..engine import result as _result
 from ..sql import coercions
 from ..sql import expression
@@ -57,6 +58,7 @@ from ..sql.selectable import LABEL_STYLE_TABLENAME_PLUS_COL
 
 if TYPE_CHECKING:
     from .mapper import Mapper
+    from .session import ORMExecuteState
     from .session import SessionTransaction
     from .state import InstanceState
 
@@ -1103,7 +1105,8 @@ def _emit_insert_statements(
             or (
                 has_all_defaults
                 or not base_mapper.eager_defaults
-                or not connection.dialect.implicit_returning
+                or not base_mapper.local_table.implicit_returning
+                or not connection.dialect.insert_returning
             )
             and has_all_pks
             and not hasvalue
@@ -1118,7 +1121,6 @@ def _emit_insert_statements(
             c = connection.execute(
                 statement, multiparams, execution_options=execution_options
             )
-
             if bookkeeping:
                 for (
                     (
@@ -1803,6 +1805,10 @@ class BulkUDCompileState(CompileState):
         _refresh_identity_token = None
 
     @classmethod
+    def can_use_returning(cls, dialect: Dialect, mapper: Mapper[Any]) -> bool:
+        raise NotImplementedError()
+
+    @classmethod
     def orm_pre_session_exec(
         cls,
         session,
@@ -2093,9 +2099,10 @@ class BulkUDCompileState(CompileState):
         )
         select_stmt._where_criteria = statement._where_criteria
 
-        def skip_for_full_returning(orm_context):
+        def skip_for_returning(orm_context: ORMExecuteState) -> Any:
             bind = orm_context.session.get_bind(**orm_context.bind_arguments)
-            if bind.dialect.full_returning:
+
+            if cls.can_use_returning(bind.dialect, mapper):
                 return _result.null_result()
             else:
                 return None
@@ -2105,7 +2112,7 @@ class BulkUDCompileState(CompileState):
             params,
             execution_options=execution_options,
             bind_arguments=bind_arguments,
-            _add_event=skip_for_full_returning,
+            _add_event=skip_for_returning,
         )
         matched_rows = result.fetchall()
 
@@ -2283,10 +2290,9 @@ class BulkORMUpdate(ORMDMLState, UpdateDMLState, BulkUDCompileState):
         # if we are against a lambda statement we might not be the
         # topmost object that received per-execute annotations
 
-        if (
-            compiler._annotations.get("synchronize_session", None) == "fetch"
-            and compiler.dialect.full_returning
-        ):
+        if compiler._annotations.get(
+            "synchronize_session", None
+        ) == "fetch" and self.can_use_returning(compiler.dialect, mapper):
             if new_stmt._returning:
                 raise sa_exc.InvalidRequestError(
                     "Can't use synchronize_session='fetch' "
@@ -2299,6 +2305,12 @@ class BulkORMUpdate(ORMDMLState, UpdateDMLState, BulkUDCompileState):
         return self
 
     @classmethod
+    def can_use_returning(cls, dialect: Dialect, mapper: Mapper[Any]) -> bool:
+        return (
+            dialect.update_returning and mapper.local_table.implicit_returning
+        )
+
+    @classmethod
     def _get_crud_kv_pairs(cls, statement, kv_iterator):
         plugin_subject = statement._propagate_attrs["plugin_subject"]
 
@@ -2478,12 +2490,9 @@ class BulkORMDelete(ORMDMLState, DeleteDMLState, BulkUDCompileState):
         if new_crit:
             statement = statement.where(*new_crit)
 
-        if (
-            mapper
-            and compiler._annotations.get("synchronize_session", None)
-            == "fetch"
-            and compiler.dialect.full_returning
-        ):
+        if compiler._annotations.get(
+            "synchronize_session", None
+        ) == "fetch" and self.can_use_returning(compiler.dialect, mapper):
             statement = statement.returning(*mapper.primary_key)
 
         DeleteDMLState.__init__(self, statement, compiler, **kw)
@@ -2491,6 +2500,12 @@ class BulkORMDelete(ORMDMLState, DeleteDMLState, BulkUDCompileState):
         return self
 
     @classmethod
+    def can_use_returning(cls, dialect: Dialect, mapper: Mapper[Any]) -> bool:
+        return (
+            dialect.delete_returning and mapper.local_table.implicit_returning
+        )
+
+    @classmethod
     def _do_post_synchronize_evaluate(cls, session, result, update_options):
 
         session._remove_newly_deleted(
diff --git a/lib/sqlalchemy/sql/compiler.py b/lib/sqlalchemy/sql/compiler.py
index 3685751b0..78c6af38b 100644
--- a/lib/sqlalchemy/sql/compiler.py
+++ b/lib/sqlalchemy/sql/compiler.py
@@ -3482,7 +3482,7 @@ class SQLCompiler(Compiled):
         )
 
     def _label_returning_column(
-        self, stmt, column, populate_result_map, column_clause_args=None
+        self, stmt, column, populate_result_map, column_clause_args=None, **kw
     ):
         """Render a column with necessary labels inside of a RETURNING clause.
 
@@ -3499,6 +3499,7 @@ class SQLCompiler(Compiled):
             populate_result_map,
             False,
             {} if column_clause_args is None else column_clause_args,
+            **kw,
         )
 
     def _label_select_column(
@@ -3514,6 +3515,7 @@ class SQLCompiler(Compiled):
         within_columns_clause=True,
         column_is_repeated=False,
         need_column_expressions=False,
+        include_table=True,
     ):
         """produce labeled columns present in a select()."""
         impl = column.type.dialect_impl(self.dialect)
@@ -3661,6 +3663,7 @@ class SQLCompiler(Compiled):
         column_clause_args.update(
             within_columns_clause=within_columns_clause,
             add_to_result_map=add_to_result_map,
+            include_table=include_table,
         )
         return result_expr._compiler_dispatch(self, **column_clause_args)
 
@@ -4218,10 +4221,12 @@ class SQLCompiler(Compiled):
         populate_result_map: bool,
         **kw: Any,
     ) -> str:
-        raise exc.CompileError(
-            "RETURNING is not supported by this "
-            "dialect's statement compiler."
-        )
+        columns = [
+            self._label_returning_column(stmt, c, populate_result_map, **kw)
+            for c in base._select_iterables(returning_cols)
+        ]
+
+        return "RETURNING " + ", ".join(columns)
 
     def limit_clause(self, select, **kw):
         text = ""
diff --git a/lib/sqlalchemy/sql/crud.py b/lib/sqlalchemy/sql/crud.py
index 913e4d433..81151a26b 100644
--- a/lib/sqlalchemy/sql/crud.py
+++ b/lib/sqlalchemy/sql/crud.py
@@ -568,6 +568,7 @@ def _scan_cols(
                 _col_bind_name,
                 implicit_returning,
                 implicit_return_defaults,
+                postfetch_lastrowid,
                 values,
                 autoincrement_col,
                 insert_null_pk_still_autoincrements,
@@ -649,6 +650,7 @@ def _append_param_parameter(
     _col_bind_name,
     implicit_returning,
     implicit_return_defaults,
+    postfetch_lastrowid,
     values,
     autoincrement_col,
     insert_null_pk_still_autoincrements,
@@ -668,11 +670,12 @@ def _append_param_parameter(
             and c is autoincrement_col
         ):
             # support use case for #7998, fetch autoincrement cols
-            # even if value was given
-            if implicit_returning:
-                compiler.implicit_returning.append(c)
-            elif compiler.dialect.postfetch_lastrowid:
+            # even if value was given.
+
+            if postfetch_lastrowid:
                 compiler.postfetch_lastrowid = True
+            elif implicit_returning:
+                compiler.implicit_returning.append(c)
 
         value = _create_bind_param(
             compiler,
@@ -1281,7 +1284,12 @@ def _get_stmt_parameter_tuples_params(
 
 
 def _get_returning_modifiers(compiler, stmt, compile_state, toplevel):
+    """determines RETURNING strategy, if any, for the statement.
+
+    This is where it's determined what we need to fetch from the
+    INSERT or UPDATE statement after it's invoked.
 
+    """
     need_pks = (
         toplevel
         and compile_state.isinsert
@@ -1296,19 +1304,58 @@ def _get_returning_modifiers(compiler, stmt, compile_state, toplevel):
         and not stmt._returning
         and not compile_state._has_multi_parameters
     )
+
+    # check if we have access to simple cursor.lastrowid.  we can use that
+    # after the INSERT if that's all we need.
+    postfetch_lastrowid = (
+        need_pks
+        and compiler.dialect.postfetch_lastrowid
+        and stmt.table._autoincrement_column is not None
+    )
+
+    # see if we want to add RETURNING to an INSERT in order to get
+    # primary key columns back.  This would be instead of postfetch_lastrowid
+    # if that's set.
     implicit_returning = (
+        # statement itself can veto it
         need_pks
-        and compiler.dialect.implicit_returning
-        and stmt.table.implicit_returning
+        # the dialect can veto it if it just doesnt support RETURNING
+        # with INSERT
+        and compiler.dialect.insert_returning
+        # user-defined implicit_returning on Table can veto it
+        and compile_state._primary_table.implicit_returning
+        # the compile_state can veto it (SQlite uses this to disable
+        # RETURNING for an ON CONFLICT insert, as SQLite does not return
+        # for rows that were updated, which is wrong)
+        and compile_state._supports_implicit_returning
+        and (
+            # since we support MariaDB and SQLite which also support lastrowid,
+            # decide if we should use lastrowid or RETURNING.  for insert
+            # that didnt call return_defaults() and has just one set of
+            # parameters, we can use lastrowid.   this is more "traditional"
+            # and a lot of weird use cases are supported by it.
+            # SQLite lastrowid times 3x faster than returning,
+            # Mariadb lastrowid 2x faster than returning
+            (
+                not postfetch_lastrowid
+                or compiler.dialect.favor_returning_over_lastrowid
+            )
+            or compile_state._has_multi_parameters
+            or stmt._return_defaults
+        )
     )
 
+    if implicit_returning:
+        postfetch_lastrowid = False
+
     if compile_state.isinsert:
         implicit_return_defaults = implicit_returning and stmt._return_defaults
     elif compile_state.isupdate:
         implicit_return_defaults = (
-            compiler.dialect.implicit_returning
-            and stmt.table.implicit_returning
-            and stmt._return_defaults
+            stmt._return_defaults
+            and compile_state._primary_table.implicit_returning
+            and compile_state._supports_implicit_returning
+            and compiler.dialect.update_returning
         )
     else:
         # this line is unused, currently we are always
@@ -1321,7 +1368,6 @@ def _get_returning_modifiers(compiler, stmt, compile_state, toplevel):
         else:
             implicit_return_defaults = set(stmt._return_defaults_columns)
 
-    postfetch_lastrowid = need_pks and compiler.dialect.postfetch_lastrowid
     return (
         need_pks,
         implicit_returning,
diff --git a/lib/sqlalchemy/sql/dml.py b/lib/sqlalchemy/sql/dml.py
index e63a34454..28ea512a7 100644
--- a/lib/sqlalchemy/sql/dml.py
+++ b/lib/sqlalchemy/sql/dml.py
@@ -119,6 +119,8 @@ class DMLState(CompileState):
     _ordered_values: Optional[List[Tuple[_DMLColumnElement, Any]]] = None
     _parameter_ordering: Optional[List[_DMLColumnElement]] = None
     _has_multi_parameters = False
+    _primary_table: FromClause
+    _supports_implicit_returning = True
 
     isupdate = False
     isdelete = False
@@ -182,11 +184,14 @@ class DMLState(CompileState):
             for k, v in kv_iterator
         ]
 
-    def _make_extra_froms(self, statement: DMLWhereBase) -> List[FromClause]:
+    def _make_extra_froms(
+        self, statement: DMLWhereBase
+    ) -> Tuple[FromClause, List[FromClause]]:
         froms: List[FromClause] = []
 
         all_tables = list(sql_util.tables_from_leftmost(statement.table))
-        seen = {all_tables[0]}
+        primary_table = all_tables[0]
+        seen = {primary_table}
 
         for crit in statement._where_criteria:
             for item in _from_objects(crit):
@@ -195,7 +200,7 @@ class DMLState(CompileState):
                 seen.update(item._cloned_set)
 
         froms.extend(all_tables[1:])
-        return froms
+        return primary_table, froms
 
     def _process_multi_values(self, statement: ValuesBase) -> None:
         if not statement._supports_multi_parameters:
@@ -286,8 +291,18 @@ class InsertDMLState(DMLState):
 
     include_table_with_column_exprs = False
 
-    def __init__(self, statement: Insert, compiler: SQLCompiler, **kw: Any):
+    def __init__(
+        self,
+        statement: Insert,
+        compiler: SQLCompiler,
+        disable_implicit_returning: bool = False,
+        **kw: Any,
+    ):
         self.statement = statement
+        self._primary_table = statement.table
+
+        if disable_implicit_returning:
+            self._supports_implicit_returning = False
 
         self.isinsert = True
         if statement._select_names:
@@ -306,6 +321,7 @@ class UpdateDMLState(DMLState):
 
     def __init__(self, statement: Update, compiler: SQLCompiler, **kw: Any):
         self.statement = statement
+
         self.isupdate = True
         if statement._ordered_values is not None:
             self._process_ordered_values(statement)
@@ -313,7 +329,9 @@ class UpdateDMLState(DMLState):
             self._process_values(statement)
         elif statement._multi_values:
             self._process_multi_values(statement)
-        self._extra_froms = ef = self._make_extra_froms(statement)
+        t, ef = self._make_extra_froms(statement)
+        self._primary_table = t
+        self._extra_froms = ef
 
         self.is_multitable = mt = ef
 
@@ -330,7 +348,9 @@ class DeleteDMLState(DMLState):
         self.statement = statement
 
         self.isdelete = True
-        self._extra_froms = self._make_extra_froms(statement)
+        t, ef = self._make_extra_froms(statement)
+        self._primary_table = t
+        self._extra_froms = ef
 
 
 SelfUpdateBase = typing.TypeVar("SelfUpdateBase", bound="UpdateBase")
diff --git a/lib/sqlalchemy/sql/schema.py b/lib/sqlalchemy/sql/schema.py
index 598bacc59..447e102ed 100644
--- a/lib/sqlalchemy/sql/schema.py
+++ b/lib/sqlalchemy/sql/schema.py
@@ -639,10 +639,13 @@ class Table(
 
 
         :param implicit_returning: True by default - indicates that
-            RETURNING can be used by default to fetch newly inserted primary key
-            values, for backends which support this.  Note that
-            :func:`_sa.create_engine` also provides an ``implicit_returning``
-            flag.
+            RETURNING can be used, typically by the ORM, in order to fetch
+            server-generated values such as primary key values and
+            server side defaults, on those backends which support RETURNING.
+
+            In modern SQLAlchemy there is generally no reason to alter this
+            setting, except in the case of some backends such as SQL Server
+            when INSERT triggers are used for that table.
 
         :param include_columns: A list of strings indicating a subset of
             columns to be loaded via the ``autoload`` operation; table columns who
diff --git a/lib/sqlalchemy/sql/selectable.py b/lib/sqlalchemy/sql/selectable.py
index 53dcf51c7..eebefb877 100644
--- a/lib/sqlalchemy/sql/selectable.py
+++ b/lib/sqlalchemy/sql/selectable.py
@@ -1635,6 +1635,10 @@ class AliasedReturnsRows(NoInit, NamedFromClause):
 
         return name
 
+    @util.ro_non_memoized_property
+    def implicit_returning(self):
+        return self.element.implicit_returning  # type: ignore
+
     @property
     def original(self):
         """Legacy for dialects that are referring to Alias.original."""
diff --git a/lib/sqlalchemy/testing/assertsql.py b/lib/sqlalchemy/testing/assertsql.py
index a6e3c8764..4416fe630 100644
--- a/lib/sqlalchemy/testing/assertsql.py
+++ b/lib/sqlalchemy/testing/assertsql.py
@@ -67,10 +67,13 @@ class CursorSQL(SQLMatchRule):
 
 
 class CompiledSQL(SQLMatchRule):
-    def __init__(self, statement, params=None, dialect="default"):
+    def __init__(
+        self, statement, params=None, dialect="default", enable_returning=False
+    ):
         self.statement = statement
         self.params = params
         self.dialect = dialect
+        self.enable_returning = enable_returning
 
     def _compare_sql(self, execute_observed, received_statement):
         stmt = re.sub(r"[\n\t]", "", self.statement)
@@ -82,14 +85,14 @@ class CompiledSQL(SQLMatchRule):
             # this is currently what tests are expecting
             # dialect.supports_default_values = True
             dialect.supports_default_metavalue = True
+
+            if self.enable_returning:
+                dialect.insert_returning = (
+                    dialect.update_returning
+                ) = dialect.delete_returning = True
             return dialect
         else:
-            # ugh
-            if self.dialect == "postgresql":
-                params = {"implicit_returning": True}
-            else:
-                params = {}
-            return url.URL.create(self.dialect).get_dialect()(**params)
+            return url.URL.create(self.dialect).get_dialect()()
 
     def _received_statement(self, execute_observed):
         """reconstruct the statement and params in terms
@@ -221,12 +224,15 @@ class CompiledSQL(SQLMatchRule):
 
 
 class RegexSQL(CompiledSQL):
-    def __init__(self, regex, params=None, dialect="default"):
+    def __init__(
+        self, regex, params=None, dialect="default", enable_returning=False
+    ):
         SQLMatchRule.__init__(self)
         self.regex = re.compile(regex)
         self.orig_regex = regex
         self.params = params
         self.dialect = dialect
+        self.enable_returning = enable_returning
 
     def _failure_message(self, execute_observed, expected_params):
         return (
diff --git a/lib/sqlalchemy/testing/fixtures.py b/lib/sqlalchemy/testing/fixtures.py
index ae7a42488..d0e7d8f3c 100644
--- a/lib/sqlalchemy/testing/fixtures.py
+++ b/lib/sqlalchemy/testing/fixtures.py
@@ -90,6 +90,20 @@ class TestBase:
         conn.close()
 
     @config.fixture()
+    def close_result_when_finished(self):
+        to_close = []
+
+        def go(result):
+            to_close.append(result)
+
+        yield go
+        for r in to_close:
+            try:
+                r.close()
+            except:
+                pass
+
+    @config.fixture()
     def registry(self, metadata):
         reg = registry(
             metadata=metadata,
diff --git a/lib/sqlalchemy/testing/requirements.py b/lib/sqlalchemy/testing/requirements.py
index 4fff6546e..4f9c73cf6 100644
--- a/lib/sqlalchemy/testing/requirements.py
+++ b/lib/sqlalchemy/testing/requirements.py
@@ -365,15 +365,30 @@ class SuiteRequirements(Requirements):
         return exclusions.open()
 
     @property
-    def full_returning(self):
-        """target platform supports RETURNING completely, including
-        multiple rows returned.
+    def delete_returning(self):
+        """target platform supports DELETE ... RETURNING."""
 
-        """
+        return exclusions.only_if(
+            lambda config: config.db.dialect.delete_returning,
+            "%(database)s %(does_support)s 'DELETE ... RETURNING'",
+        )
+
+    @property
+    def insert_returning(self):
+        """target platform supports INSERT ... RETURNING."""
+
+        return exclusions.only_if(
+            lambda config: config.db.dialect.insert_returning,
+            "%(database)s %(does_support)s 'INSERT ... RETURNING'",
+        )
+
+    @property
+    def update_returning(self):
+        """target platform supports UPDATE ... RETURNING."""
 
         return exclusions.only_if(
-            lambda config: config.db.dialect.full_returning,
-            "%(database)s %(does_support)s 'RETURNING of multiple rows'",
+            lambda config: config.db.dialect.update_returning,
+            "%(database)s %(does_support)s 'UPDATE ... RETURNING'",
         )
 
     @property
@@ -391,21 +406,6 @@ class SuiteRequirements(Requirements):
         )
 
     @property
-    def returning(self):
-        """target platform supports RETURNING for at least one row.
-
-        .. seealso::
-
-            :attr:`.Requirements.full_returning`
-
-        """
-
-        return exclusions.only_if(
-            lambda config: config.db.dialect.implicit_returning,
-            "%(database)s %(does_support)s 'RETURNING of a single row'",
-        )
-
-    @property
     def tuple_in(self):
         """Target platform supports the syntax
         "(x, y) IN ((x1, y1), (x2, y2), ...)"
diff --git a/lib/sqlalchemy/testing/suite/test_insert.py b/lib/sqlalchemy/testing/suite/test_insert.py
index f0e4bfcc6..2307d3b3f 100644
--- a/lib/sqlalchemy/testing/suite/test_insert.py
+++ b/lib/sqlalchemy/testing/suite/test_insert.py
@@ -125,10 +125,14 @@ class InsertBehaviorTest(fixtures.TablesTest):
         # case, the row had to have been consumed at least.
         assert not r.returns_rows or r.fetchone() is None
 
-    @requirements.returning
+    @requirements.insert_returning
     def test_autoclose_on_insert_implicit_returning(self, connection):
         r = connection.execute(
-            self.tables.autoinc_pk.insert(), dict(data="some data")
+            # return_defaults() ensures RETURNING will be used,
+            # new in 2.0 as sqlite/mariadb offer both RETURNING and
+            # cursor.lastrowid
+            self.tables.autoinc_pk.insert().return_defaults(),
+            dict(data="some data"),
         )
         assert r._soft_closed
         assert not r.closed
@@ -295,7 +299,7 @@ class InsertBehaviorTest(fixtures.TablesTest):
 
 class ReturningTest(fixtures.TablesTest):
     run_create_tables = "each"
-    __requires__ = "returning", "autoincrement_insert"
+    __requires__ = "insert_returning", "autoincrement_insert"
     __backend__ = True
 
     def _assert_round_trip(self, table, conn):