add eager_defaults="auto" for inserts

Added a new default value for the :paramref:`.Mapper.eager_defaults`
parameter "auto", which will automatically fetch table default values
during a unit of work flush, if the dialect supports RETURNING for the
INSERT being run, as well as
:ref:`insertmanyvalues <engine_insertmanyvalues>` available. Eager fetches
for server-side UPDATE defaults, which are very uncommon, continue to only
take place if :paramref:`.Mapper.eager_defaults` is set to ``True``, as
there is no batch-RETURNING form for UPDATE statements.

Fixes: #8889
Change-Id: I84b91092a37c4cd216e060513acde3eb0298abe9

5 files changed, 110 insertions, 34 deletions

diff --git a/lib/sqlalchemy/dialects/mssql/base.py b/lib/sqlalchemy/dialects/mssql/base.py
index a0049c361..aa640727f 100644
--- a/lib/sqlalchemy/dialects/mssql/base.py
+++ b/lib/sqlalchemy/dialects/mssql/base.py
@@ -232,6 +232,7 @@ integer values in Python 3), use :class:`_types.TypeDecorator` as follows::
         )
         name = Column(String)
 
+.. _mssql_insert_behavior:
 
 INSERT behavior
 ^^^^^^^^^^^^^^^^
diff --git a/lib/sqlalchemy/orm/mapper.py b/lib/sqlalchemy/orm/mapper.py
index 7a7524621..90c463ce6 100644
--- a/lib/sqlalchemy/orm/mapper.py
+++ b/lib/sqlalchemy/orm/mapper.py
@@ -230,7 +230,7 @@ class Mapper(
         passive_updates: bool = True,
         passive_deletes: bool = False,
         confirm_deleted_rows: bool = True,
-        eager_defaults: bool = False,
+        eager_defaults: Literal[True, False, "auto"] = "auto",
         legacy_is_orphan: bool = False,
         _compiled_cache_size: int = 100,
     ):
@@ -336,14 +336,30 @@ class Mapper(
           value of server-generated default values after an INSERT or UPDATE,
           rather than leaving them as expired to be fetched on next access.
           This can be used for event schemes where the server-generated values
-          are needed immediately before the flush completes.   By default,
-          this scheme will emit an individual ``SELECT`` statement per row
-          inserted or updated, which note can add significant performance
-          overhead.  However, if the
-          target database supports :term:`RETURNING`, the default values will
-          be returned inline with the INSERT or UPDATE statement, which can
-          greatly enhance performance for an application that needs frequent
-          access to just-generated server defaults.
+          are needed immediately before the flush completes.
+
+          The fetch of values occurs either by using ``RETURNING`` inline
+          with the ``INSERT`` or ``UPDATE`` statement, or by adding an
+          additional ``SELECT`` statement subsequent to the ``INSERT`` or
+          ``UPDATE``, if the backend does not support ``RETURNING``.
+
+          The use of ``RETURNING`` is extremely performant in particular for
+          ``INSERT`` statements where SQLAlchemy can take advantage of
+          :ref:`insertmanyvalues <engine_insertmanyvalues>`, whereas the use of
+          an additional ``SELECT`` is relatively poor performing, adding
+          additional SQL round trips which would be unnecessary if these new
+          attributes are not to be accessed in any case.
+
+          For this reason, :paramref:`.Mapper.eager_defaults` defaults to the
+          string value ``"auto"``, which indicates that server defaults for
+          INSERT should be fetched using ``RETURNING`` if the backing database
+          supports it and if the dialect in use supports "insertmanyreturning"
+          for an INSERT statement. If the backing database does not support
+          ``RETURNING`` or "insertmanyreturning" is not available, server
+          defaults will not be fetched.
+
+          .. versionchanged:: 2.0.0b5 added the "auto" option for
+             :paramref:`.Mapper.eager_defaults`
 
           .. seealso::
 
@@ -352,6 +368,12 @@ class Mapper(
           .. versionchanged:: 0.9.0 The ``eager_defaults`` option can now
              make use of :term:`RETURNING` for backends which support it.
 
+          .. versionchanged:: 2.0.0  RETURNING now works with multiple rows
+             INSERTed at once using the
+             :ref:`insertmanyvalues <engine_insertmanyvalues>` feature, which
+             among other things allows the :paramref:`.Mapper.eager_defaults`
+             feature to be very performant on supporting backends.
+
         :param exclude_properties: A list or set of string column names to
           be excluded from mapping.
 
@@ -818,6 +840,18 @@ class Mapper(
             self._log("constructed")
             self._expire_memoizations()
 
+    def _prefer_eager_defaults(self, dialect, table):
+        if self.eager_defaults == "auto":
+            if not table.implicit_returning:
+                return False
+
+            return (
+                table in self._server_default_col_keys
+                and dialect.insert_executemany_returning
+            )
+        else:
+            return self.eager_defaults
+
     def _gen_cache_key(self, anon_map, bindparams):
         return (self,)
 
diff --git a/lib/sqlalchemy/orm/persistence.py b/lib/sqlalchemy/orm/persistence.py
index c236ad1cf..0eff4e1fa 100644
--- a/lib/sqlalchemy/orm/persistence.py
+++ b/lib/sqlalchemy/orm/persistence.py
@@ -381,7 +381,9 @@ def _collect_insert_commands(
             # compare to pk_keys_by_table
             has_all_pks = mapper._pk_keys_by_table[table].issubset(params)
 
-            if mapper.base_mapper.eager_defaults:
+            if mapper.base_mapper._prefer_eager_defaults(
+                connection.dialect, table
+            ):
                 has_all_defaults = mapper._server_default_col_keys[
                     table
                 ].issubset(params)
@@ -491,7 +493,7 @@ def _collect_update_commands(
                 ):
                     params[col.key] = value
 
-            if mapper.base_mapper.eager_defaults:
+            if mapper.base_mapper.eager_defaults is True:
                 has_all_defaults = (
                     mapper._server_onupdate_default_col_keys[table]
                 ).issubset(params)
@@ -787,7 +789,12 @@ def _emit_update_statements(
         if (
             bookkeeping
             and not has_all_defaults
-            and mapper.base_mapper.eager_defaults
+            and mapper.base_mapper.eager_defaults is True
+            # change as of #8889 - if RETURNING is not going to be used anyway,
+            # (applies to MySQL, MariaDB which lack UPDATE RETURNING) ensure
+            # we can do an executemany UPDATE which is more efficient
+            and table.implicit_returning
+            and connection.dialect.update_returning
         ):
             statement = statement.return_defaults(
                 *mapper._server_onupdate_default_cols[table]
@@ -808,7 +815,11 @@ def _emit_update_statements(
             assert_singlerow
             and connection.dialect.supports_sane_multi_rowcount
         )
-        allow_multirow = has_all_defaults and not needs_version_id
+
+        # change as of #8889 - if RETURNING is not going to be used anyway,
+        # (applies to MySQL, MariaDB which lack UPDATE RETURNING) ensure
+        # we can do an executemany UPDATE which is more efficient
+        allow_executemany = not return_defaults and not needs_version_id
 
         if hasvalue:
             for (
@@ -842,7 +853,7 @@ def _emit_update_statements(
                 rows += c.rowcount
                 check_rowcount = assert_singlerow
         else:
-            if not allow_multirow:
+            if not allow_executemany:
                 check_rowcount = assert_singlerow
                 for (
                     state,
@@ -991,7 +1002,9 @@ def _emit_insert_statements(
                 not bookkeeping
                 or (
                     has_all_defaults
-                    or not base_mapper.eager_defaults
+                    or not base_mapper._prefer_eager_defaults(
+                        connection.dialect, table
+                    )
                     or not table.implicit_returning
                     or not connection.dialect.insert_returning
                 )
@@ -1067,7 +1080,9 @@ def _emit_insert_statements(
             else:
                 do_executemany = False
 
-            if not has_all_defaults and base_mapper.eager_defaults:
+            if not has_all_defaults and base_mapper._prefer_eager_defaults(
+                connection.dialect, table
+            ):
                 statement = statement.return_defaults(
                     *mapper._server_default_cols[table]
                 )
@@ -1282,9 +1297,9 @@ def _emit_post_update_statements(
             assert_singlerow
             and connection.dialect.supports_sane_multi_rowcount
         )
-        allow_multirow = not needs_version_id or assert_multirow
+        allow_executemany = not needs_version_id or assert_multirow
 
-        if not allow_multirow:
+        if not allow_executemany:
             check_rowcount = assert_singlerow
             for state, state_dict, mapper_rec, connection, params in records:
 
@@ -1475,7 +1490,9 @@ def _finalize_insert_update_commands(base_mapper, uowtransaction, states):
         # it isn't expired.
         toload_now = []
 
-        if base_mapper.eager_defaults:
+        # this is specifically to emit a second SELECT for eager_defaults,
+        # so only if it's set to True, not "auto"
+        if base_mapper.eager_defaults is True:
             toload_now.extend(
                 state._unloaded_non_object.intersection(
                     mapper._server_default_plus_onupdate_propkeys
diff --git a/lib/sqlalchemy/testing/__init__.py b/lib/sqlalchemy/testing/__init__.py
index 993fc4954..6454750f5 100644
--- a/lib/sqlalchemy/testing/__init__.py
+++ b/lib/sqlalchemy/testing/__init__.py
@@ -56,6 +56,7 @@ from .config import requirements as requires
 from .config import skip_test
 from .config import Variation
 from .config import variation
+from .config import variation_fixture
 from .exclusions import _is_excluded
 from .exclusions import _server_version
 from .exclusions import against as _against
diff --git a/lib/sqlalchemy/testing/config.py b/lib/sqlalchemy/testing/config.py
index 6adcf5b64..b444eb39f 100644
--- a/lib/sqlalchemy/testing/config.py
+++ b/lib/sqlalchemy/testing/config.py
@@ -157,9 +157,33 @@ class Variation:
     def __str__(self):
         return f"{self._argname}={self._name!r}"
 
+    def __repr__(self):
+        return str(self)
+
     def fail(self) -> NoReturn:
         fail(f"Unknown {self}")
 
+    @classmethod
+    def idfn(cls, variation):
+        return variation.name
+
+    @classmethod
+    def generate_cases(cls, argname, cases):
+        case_names = [
+            argname if c is True else "not_" + argname if c is False else c
+            for c in cases
+        ]
+
+        typ = type(
+            argname,
+            (Variation,),
+            {
+                "__slots__": tuple(case_names),
+            },
+        )
+
+        return [typ(casename, argname, case_names) for casename in case_names]
+
 
 def variation(argname, cases):
     """a helper around testing.combinations that provides a single namespace
@@ -203,26 +227,17 @@ def variation(argname, cases):
         else (entry, None)
         for entry in cases
     ]
-    case_names = [
-        argname if c is True else "not_" + argname if c is False else c
-        for c, l in cases_plus_limitations
-    ]
 
-    typ = type(
-        argname,
-        (Variation,),
-        {
-            "__slots__": tuple(case_names),
-        },
+    variations = Variation.generate_cases(
+        argname, [c for c, l in cases_plus_limitations]
     )
-
     return combinations(
         *[
-            (casename, typ(casename, argname, case_names), limitation)
+            (variation._name, variation, limitation)
             if limitation is not None
-            else (casename, typ(casename, argname, case_names))
-            for casename, (case, limitation) in zip(
-                case_names, cases_plus_limitations
+            else (variation._name, variation)
+            for variation, (case, limitation) in zip(
+                variations, cases_plus_limitations
             )
         ],
         id_="ia",
@@ -230,6 +245,14 @@ def variation(argname, cases):
     )
 
 
+def variation_fixture(argname, cases, scope="function"):
+    return fixture(
+        params=Variation.generate_cases(argname, cases),
+        ids=Variation.idfn,
+        scope=scope,
+    )
+
+
 def fixture(*arg: Any, **kw: Any) -> Any:
     return _fixture_functions.fixture(*arg, **kw)