implement batched INSERT..VALUES () () for executemany

the feature is enabled for all built in backends
when RETURNING is used,
except for Oracle that doesn't need it,  and on
psycopg2 and mssql+pyodbc it is used for all INSERT statements,
not just those that use RETURNING.

third party dialects would need to opt in to the new feature
by setting use_insertmanyvalues to True.

Also adds dialect-level guards against using returning
with executemany where we dont have an implementation to
suit it.   execute single w/ returning still defers to the
server without us checking.

Fixes: #6047
Fixes: #7907
Change-Id: I3936d3c00003f02e322f2e43fb949d0e6e568304

4 files changed, 566 insertions, 57 deletions

diff --git a/lib/sqlalchemy/engine/base.py b/lib/sqlalchemy/engine/base.py
index 2b9cf602a..1b07acab5 100644
--- a/lib/sqlalchemy/engine/base.py
+++ b/lib/sqlalchemy/engine/base.py
@@ -12,6 +12,7 @@ import typing
 from typing import Any
 from typing import Callable
 from typing import cast
+from typing import Iterable
 from typing import Iterator
 from typing import List
 from typing import Mapping
@@ -29,6 +30,7 @@ from .interfaces import BindTyping
 from .interfaces import ConnectionEventsTarget
 from .interfaces import DBAPICursor
 from .interfaces import ExceptionContext
+from .interfaces import ExecuteStyle
 from .interfaces import ExecutionContext
 from .util import _distill_params_20
 from .util import _distill_raw_params
@@ -438,6 +440,20 @@ class Connection(ConnectionEventsTarget, inspection.Inspectable["Inspector"]):
             :ref:`orm_queryguide_yield_per` - in the :ref:`queryguide_toplevel`
             describing the ORM version of ``yield_per``
 
+        :param insertmanyvalues_page_size: number of rows to format into an
+            INSERT statement when the statement uses "insertmanyvalues" mode,
+            which is a paged form of bulk insert that is used for many backends
+            when using :term:`executemany` execution typically in conjunction
+            with RETURNING. Defaults to 1000. May also be modified on a
+            per-engine basis using the
+            :paramref:`_sa.create_engine.insertmanyvalues_page_size` parameter.
+
+            .. versionadded:: 2.0
+
+            .. seealso::
+
+                :ref:`engine_insertmanyvalues`
+
         :param schema_translate_map: Available on: :class:`_engine.Connection`,
           :class:`_engine.Engine`, :class:`_sql.Executable`.
 
@@ -1795,8 +1811,39 @@ class Connection(ConnectionEventsTarget, inspection.Inspectable["Inspector"]):
 
         context.pre_exec()
 
+        if context.execute_style is ExecuteStyle.INSERTMANYVALUES:
+            return self._exec_insertmany_context(
+                dialect,
+                context,
+            )
+        else:
+            return self._exec_single_context(
+                dialect, context, statement, parameters
+            )
+
+    def _exec_single_context(
+        self,
+        dialect: Dialect,
+        context: ExecutionContext,
+        statement: Union[str, Compiled],
+        parameters: Optional[_AnyMultiExecuteParams],
+    ) -> CursorResult[Any]:
+        """continue the _execute_context() method for a single DBAPI
+        cursor.execute() or cursor.executemany() call.
+
+        """
         if dialect.bind_typing is BindTyping.SETINPUTSIZES:
-            context._set_input_sizes()
+            generic_setinputsizes = context._prepare_set_input_sizes()
+
+            if generic_setinputsizes:
+                try:
+                    dialect.do_set_input_sizes(
+                        context.cursor, generic_setinputsizes, context
+                    )
+                except BaseException as e:
+                    self._handle_dbapi_exception(
+                        e, str(statement), parameters, None, context
+                    )
 
         cursor, str_statement, parameters = (
             context.cursor,
@@ -1840,13 +1887,13 @@ class Connection(ConnectionEventsTarget, inspection.Inspectable["Inspector"]):
                 )
             else:
                 self._log_info(
-                    "[%s] [SQL parameters hidden due to hide_parameters=True]"
-                    % (stats,)
+                    "[%s] [SQL parameters hidden due to hide_parameters=True]",
+                    stats,
                 )
 
         evt_handled: bool = False
         try:
-            if context.executemany:
+            if context.execute_style is ExecuteStyle.EXECUTEMANY:
                 effective_parameters = cast(
                     "_CoreMultiExecuteParams", effective_parameters
                 )
@@ -1862,7 +1909,10 @@ class Connection(ConnectionEventsTarget, inspection.Inspectable["Inspector"]):
                             break
                 if not evt_handled:
                     self.dialect.do_executemany(
-                        cursor, str_statement, effective_parameters, context
+                        cursor,
+                        str_statement,
+                        effective_parameters,
+                        context,
                     )
             elif not effective_parameters and context.no_parameters:
                 if self.dialect._has_events:
@@ -1914,6 +1964,151 @@ class Connection(ConnectionEventsTarget, inspection.Inspectable["Inspector"]):
 
         return result
 
+    def _exec_insertmany_context(
+        self,
+        dialect: Dialect,
+        context: ExecutionContext,
+    ) -> CursorResult[Any]:
+        """continue the _execute_context() method for an "insertmanyvalues"
+        operation, which will invoke DBAPI
+        cursor.execute() one or more times with individual log and
+        event hook calls.
+
+        """
+
+        if dialect.bind_typing is BindTyping.SETINPUTSIZES:
+            generic_setinputsizes = context._prepare_set_input_sizes()
+        else:
+            generic_setinputsizes = None
+
+        cursor, str_statement, parameters = (
+            context.cursor,
+            context.statement,
+            context.parameters,
+        )
+
+        effective_parameters = parameters
+
+        engine_events = self._has_events or self.engine._has_events
+        if self.dialect._has_events:
+            do_execute_dispatch: Iterable[
+                Any
+            ] = self.dialect.dispatch.do_execute
+        else:
+            do_execute_dispatch = ()
+
+        if self._echo:
+            stats = context._get_cache_stats() + " (insertmanyvalues)"
+        for (
+            sub_stmt,
+            sub_params,
+            setinputsizes,
+            batchnum,
+            totalbatches,
+        ) in dialect._deliver_insertmanyvalues_batches(
+            cursor,
+            str_statement,
+            effective_parameters,
+            generic_setinputsizes,
+            context,
+        ):
+
+            if setinputsizes:
+                try:
+                    dialect.do_set_input_sizes(
+                        context.cursor, setinputsizes, context
+                    )
+                except BaseException as e:
+                    self._handle_dbapi_exception(
+                        e,
+                        sql_util._long_statement(sub_stmt),
+                        sub_params,
+                        None,
+                        context,
+                    )
+
+            if engine_events:
+                for fn in self.dispatch.before_cursor_execute:
+                    sub_stmt, sub_params = fn(
+                        self,
+                        cursor,
+                        sub_stmt,
+                        sub_params,
+                        context,
+                        True,
+                    )
+
+            if self._echo:
+
+                self._log_info(sql_util._long_statement(sub_stmt))
+
+                if batchnum > 1:
+                    stats = (
+                        f"insertmanyvalues batch {batchnum} "
+                        f"of {totalbatches}"
+                    )
+
+                if not self.engine.hide_parameters:
+                    self._log_info(
+                        "[%s] %r",
+                        stats,
+                        sql_util._repr_params(
+                            sub_params,
+                            batches=10,
+                            ismulti=False,
+                        ),
+                    )
+                else:
+                    self._log_info(
+                        "[%s] [SQL parameters hidden due to "
+                        "hide_parameters=True]",
+                        stats,
+                    )
+
+            try:
+                for fn in do_execute_dispatch:
+                    if fn(
+                        cursor,
+                        sub_stmt,
+                        sub_params,
+                        context,
+                    ):
+                        break
+                else:
+                    dialect.do_execute(cursor, sub_stmt, sub_params, context)
+
+            except BaseException as e:
+                self._handle_dbapi_exception(
+                    e,
+                    sql_util._long_statement(sub_stmt),
+                    sub_params,
+                    cursor,
+                    context,
+                    is_sub_exec=True,
+                )
+
+            if engine_events:
+                self.dispatch.after_cursor_execute(
+                    self,
+                    cursor,
+                    str_statement,
+                    effective_parameters,
+                    context,
+                    context.executemany,
+                )
+
+        try:
+            context.post_exec()
+
+            result = context._setup_result_proxy()
+
+        except BaseException as e:
+            self._handle_dbapi_exception(
+                e, str_statement, effective_parameters, cursor, context
+            )
+
+        return result
+
     def _cursor_execute(
         self,
         cursor: DBAPICursor,
@@ -1983,6 +2178,7 @@ class Connection(ConnectionEventsTarget, inspection.Inspectable["Inspector"]):
         parameters: Optional[_AnyExecuteParams],
         cursor: Optional[DBAPICursor],
         context: Optional[ExecutionContext],
+        is_sub_exec: bool = False,
     ) -> NoReturn:
         exc_info = sys.exc_info()
 
@@ -2001,6 +2197,11 @@ class Connection(ConnectionEventsTarget, inspection.Inspectable["Inspector"]):
 
         invalidate_pool_on_disconnect = not is_exit_exception
 
+        ismulti: bool = (
+            not is_sub_exec and context.executemany
+            if context is not None
+            else False
+        )
         if self._reentrant_error:
             raise exc.DBAPIError.instance(
                 statement,
@@ -2009,7 +2210,7 @@ class Connection(ConnectionEventsTarget, inspection.Inspectable["Inspector"]):
                 self.dialect.loaded_dbapi.Error,
                 hide_parameters=self.engine.hide_parameters,
                 dialect=self.dialect,
-                ismulti=context.executemany if context is not None else None,
+                ismulti=ismulti,
             ).with_traceback(exc_info[2]) from e
         self._reentrant_error = True
         try:
@@ -2030,9 +2231,7 @@ class Connection(ConnectionEventsTarget, inspection.Inspectable["Inspector"]):
                     hide_parameters=self.engine.hide_parameters,
                     connection_invalidated=self._is_disconnect,
                     dialect=self.dialect,
-                    ismulti=context.executemany
-                    if context is not None
-                    else None,
+                    ismulti=ismulti,
                 )
             else:
                 sqlalchemy_exception = None
diff --git a/lib/sqlalchemy/engine/create.py b/lib/sqlalchemy/engine/create.py
index 36119ab24..a9b388d71 100644
--- a/lib/sqlalchemy/engine/create.py
+++ b/lib/sqlalchemy/engine/create.py
@@ -58,6 +58,7 @@ def create_engine(
     future: Literal[True],
     hide_parameters: bool = ...,
     implicit_returning: Literal[True] = ...,
+    insertmanyvalues_page_size: int = ...,
     isolation_level: _IsolationLevel = ...,
     json_deserializer: Callable[..., Any] = ...,
     json_serializer: Callable[..., Any] = ...,
@@ -79,6 +80,7 @@ def create_engine(
     pool_use_lifo: bool = ...,
     plugins: List[str] = ...,
     query_cache_size: int = ...,
+    use_insertmanyvalues: bool = ...,
     **kwargs: Any,
 ) -> Engine:
     ...
@@ -273,6 +275,23 @@ def create_engine(url: Union[str, "_url.URL"], **kwargs: Any) -> Engine:
         :paramref:`.Table.implicit_returning` parameter.
 
 
+    :param insertmanyvalues_page_size: number of rows to format into an
+     INSERT statement when the statement uses "insertmanyvalues" mode, which is
+     a paged form of bulk insert that is used for many backends when using
+     :term:`executemany` execution typically in conjunction with RETURNING.
+     Defaults to 1000, but may also be subject to dialect-specific limiting
+     factors which may override this value on a per-statement basis.
+
+     .. versionadded:: 2.0
+
+     .. seealso::
+
+        :ref:`engine_insertmanyvalues`
+
+        :ref:`engine_insertmanyvalues_page_size`
+
+        :paramref:`_engine.Connection.execution_options.insertmanyvalues_page_size`
+
     :param isolation_level: optional string name of an isolation level
         which will be set on all new connections unconditionally.
         Isolation levels are typically some subset of the string names
@@ -508,6 +527,15 @@ def create_engine(url: Union[str, "_url.URL"], **kwargs: Any) -> Engine:
 
      .. versionadded:: 1.4
 
+    :param use_insertmanyvalues: True by default, use the "insertmanyvalues"
+     execution style for INSERT..RETURNING statements by default.
+
+     .. versionadded:: 2.0
+
+     .. seealso::
+
+        :ref:`engine_insertmanyvalues`
+
     """  # noqa
 
     if "strategy" in kwargs:
diff --git a/lib/sqlalchemy/engine/default.py b/lib/sqlalchemy/engine/default.py
index 3a53f8157..11ab713d0 100644
--- a/lib/sqlalchemy/engine/default.py
+++ b/lib/sqlalchemy/engine/default.py
@@ -35,6 +35,7 @@ from typing import Set
 from typing import Tuple
 from typing import Type
 from typing import TYPE_CHECKING
+from typing import Union
 import weakref
 
 from . import characteristics
@@ -44,6 +45,7 @@ from .base import Connection
 from .interfaces import CacheStats
 from .interfaces import DBAPICursor
 from .interfaces import Dialect
+from .interfaces import ExecuteStyle
 from .interfaces import ExecutionContext
 from .reflection import ObjectKind
 from .reflection import ObjectScope
@@ -52,13 +54,16 @@ from .. import exc
 from .. import pool
 from .. import util
 from ..sql import compiler
+from ..sql import dml
 from ..sql import expression
 from ..sql import type_api
 from ..sql._typing import is_tuple_type
+from ..sql.base import _NoArg
 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 Final
 from ..util.typing import Literal
 
 if typing.TYPE_CHECKING:
@@ -146,7 +151,6 @@ class DefaultDialect(Dialect):
     update_returning_multifrom = False
     delete_returning_multifrom = False
     insert_returning = False
-    insert_executemany_returning = False
 
     cte_follows_insert = False
 
@@ -208,6 +212,10 @@ class DefaultDialect(Dialect):
     supports_default_metavalue = False
     """dialect supports INSERT... VALUES (DEFAULT) syntax"""
 
+    default_metavalue_token = "DEFAULT"
+    """for INSERT... VALUES (DEFAULT) syntax, the token to put in the
+    parenthesis."""
+
     # not sure if this is a real thing but the compiler will deliver it
     # if this is the only flag enabled.
     supports_empty_insert = True
@@ -215,6 +223,13 @@ class DefaultDialect(Dialect):
 
     supports_multivalues_insert = False
 
+    use_insertmanyvalues: bool = False
+
+    use_insertmanyvalues_wo_returning: bool = False
+
+    insertmanyvalues_page_size: int = 1000
+    insertmanyvalues_max_parameters = 32700
+
     supports_is_distinct_from = True
 
     supports_server_side_cursors = False
@@ -272,6 +287,8 @@ class DefaultDialect(Dialect):
         supports_native_boolean: Optional[bool] = None,
         max_identifier_length: Optional[int] = None,
         label_length: Optional[int] = None,
+        insertmanyvalues_page_size: Union[_NoArg, int] = _NoArg.NO_ARG,
+        use_insertmanyvalues: Optional[bool] = None,
         # util.deprecated_params decorator cannot render the
         # Linting.NO_LINTING constant
         compiler_linting: Linting = int(compiler.NO_LINTING),  # type: ignore
@@ -332,6 +349,12 @@ class DefaultDialect(Dialect):
         self.label_length = label_length
         self.compiler_linting = compiler_linting
 
+        if use_insertmanyvalues is not None:
+            self.use_insertmanyvalues = use_insertmanyvalues
+
+        if insertmanyvalues_page_size is not _NoArg.NO_ARG:
+            self.insertmanyvalues_page_size = insertmanyvalues_page_size
+
     @util.deprecated_property(
         "2.0",
         "full_returning is deprecated, please use insert_returning, "
@@ -344,6 +367,17 @@ class DefaultDialect(Dialect):
             and self.delete_returning
         )
 
+    @property
+    def insert_executemany_returning(self):
+        return (
+            self.insert_returning
+            and self.supports_multivalues_insert
+            and self.use_insertmanyvalues
+        )
+
+    update_executemany_returning = False
+    delete_executemany_returning = False
+
     @util.memoized_property
     def loaded_dbapi(self) -> ModuleType:
         if self.dbapi is None:
@@ -682,6 +716,27 @@ class DefaultDialect(Dialect):
     def do_release_savepoint(self, connection, name):
         connection.execute(expression.ReleaseSavepointClause(name))
 
+    def _deliver_insertmanyvalues_batches(
+        self, cursor, statement, parameters, generic_setinputsizes, context
+    ):
+        context = cast(DefaultExecutionContext, context)
+        compiled = cast(SQLCompiler, context.compiled)
+
+        is_returning: Final[bool] = bool(compiled.effective_returning)
+        batch_size = context.execution_options.get(
+            "insertmanyvalues_page_size", self.insertmanyvalues_page_size
+        )
+
+        if is_returning:
+            context._insertmanyvalues_rows = result = []
+
+        for batch_rec in compiled._deliver_insertmanyvalues_batches(
+            statement, parameters, generic_setinputsizes, batch_size
+        ):
+            yield batch_rec
+            if is_returning:
+                result.extend(cursor.fetchall())
+
     def do_executemany(self, cursor, statement, parameters, context=None):
         cursor.executemany(statement, parameters)
 
@@ -936,7 +991,8 @@ class DefaultExecutionContext(ExecutionContext):
     is_text = False
     isddl = False
 
-    executemany = False
+    execute_style: ExecuteStyle = ExecuteStyle.EXECUTE
+
     compiled: Optional[Compiled] = None
     result_column_struct: Optional[
         Tuple[List[ResultColumnsEntry], bool, bool, bool, bool]
@@ -982,6 +1038,8 @@ class DefaultExecutionContext(ExecutionContext):
 
     _empty_dict_params = cast("Mapping[str, Any]", util.EMPTY_DICT)
 
+    _insertmanyvalues_rows: Optional[List[Tuple[Any, ...]]] = None
+
     @classmethod
     def _init_ddl(
         cls,
@@ -1061,23 +1119,55 @@ class DefaultExecutionContext(ExecutionContext):
             compiled._loose_column_name_matching,
         )
 
-        self.isinsert = compiled.isinsert
-        self.isupdate = compiled.isupdate
-        self.isdelete = compiled.isdelete
+        self.isinsert = ii = compiled.isinsert
+        self.isupdate = iu = compiled.isupdate
+        self.isdelete = id_ = compiled.isdelete
         self.is_text = compiled.isplaintext
 
-        if self.isinsert or self.isupdate or self.isdelete:
+        if ii or iu or id_:
             if TYPE_CHECKING:
                 assert isinstance(compiled.statement, UpdateBase)
             self.is_crud = True
-            self._is_explicit_returning = bool(compiled.statement._returning)
-            self._is_implicit_returning = is_implicit_returning = bool(
+            self._is_explicit_returning = ier = bool(
+                compiled.statement._returning
+            )
+            self._is_implicit_returning = iir = is_implicit_returning = bool(
                 compiled.implicit_returning
             )
             assert not (
                 is_implicit_returning and compiled.statement._returning
             )
 
+            if (ier or iir) and compiled.for_executemany:
+                if ii and not self.dialect.insert_executemany_returning:
+                    raise exc.InvalidRequestError(
+                        f"Dialect {self.dialect.dialect_description} with "
+                        f"current server capabilities does not support "
+                        "INSERT..RETURNING when executemany is used"
+                    )
+                elif (
+                    ii
+                    and self.dialect.use_insertmanyvalues
+                    and not compiled._insertmanyvalues
+                ):
+                    raise exc.InvalidRequestError(
+                        'Statement does not have "insertmanyvalues" '
+                        "enabled, can't use INSERT..RETURNING with "
+                        "executemany in this case."
+                    )
+                elif iu and not self.dialect.update_executemany_returning:
+                    raise exc.InvalidRequestError(
+                        f"Dialect {self.dialect.dialect_description} with "
+                        f"current server capabilities does not support "
+                        "UPDATE..RETURNING when executemany is used"
+                    )
+                elif id_ and not self.dialect.delete_executemany_returning:
+                    raise exc.InvalidRequestError(
+                        f"Dialect {self.dialect.dialect_description} with "
+                        f"current server capabilities does not support "
+                        "DELETE..RETURNING when executemany is used"
+                    )
+
         if not parameters:
             self.compiled_parameters = [
                 compiled.construct_params(
@@ -1096,7 +1186,11 @@ class DefaultExecutionContext(ExecutionContext):
                 for grp, m in enumerate(parameters)
             ]
 
-            self.executemany = len(parameters) > 1
+            if len(parameters) > 1:
+                if self.isinsert and compiled._insertmanyvalues:
+                    self.execute_style = ExecuteStyle.INSERTMANYVALUES
+                else:
+                    self.execute_style = ExecuteStyle.EXECUTEMANY
 
         self.unicode_statement = compiled.string
 
@@ -1238,7 +1332,8 @@ class DefaultExecutionContext(ExecutionContext):
                 dialect.execute_sequence_format(p) for p in parameters
             ]
 
-        self.executemany = len(parameters) > 1
+        if len(parameters) > 1:
+            self.execute_style = ExecuteStyle.EXECUTEMANY
 
         self.statement = self.unicode_statement = statement
 
@@ -1293,6 +1388,13 @@ class DefaultExecutionContext(ExecutionContext):
         else:
             return "unknown"
 
+    @property
+    def executemany(self):
+        return self.execute_style in (
+            ExecuteStyle.EXECUTEMANY,
+            ExecuteStyle.INSERTMANYVALUES,
+        )
+
     @util.memoized_property
     def identifier_preparer(self):
         if self.compiled:
@@ -1555,7 +1657,23 @@ class DefaultExecutionContext(ExecutionContext):
     def _setup_dml_or_text_result(self):
         compiled = cast(SQLCompiler, self.compiled)
 
+        strategy = self.cursor_fetch_strategy
+
         if self.isinsert:
+            if (
+                self.execute_style is ExecuteStyle.INSERTMANYVALUES
+                and compiled.effective_returning
+            ):
+                strategy = _cursor.FullyBufferedCursorFetchStrategy(
+                    self.cursor,
+                    initial_buffer=self._insertmanyvalues_rows,
+                    # maintain alt cursor description if set by the
+                    # dialect, e.g. mssql preserves it
+                    alternate_description=(
+                        strategy.alternate_cursor_description
+                    ),
+                )
+
             if compiled.postfetch_lastrowid:
                 self.inserted_primary_key_rows = (
                     self._setup_ins_pk_from_lastrowid()
@@ -1564,7 +1682,6 @@ class DefaultExecutionContext(ExecutionContext):
             # the default inserted_primary_key_rows accessor will
             # return an "empty" primary key collection when accessed.
 
-        strategy = self.cursor_fetch_strategy
         if self._is_server_side and strategy is _cursor._DEFAULT_FETCH:
             strategy = _cursor.BufferedRowCursorFetchStrategy(
                 self.cursor, self.execution_options
@@ -1675,8 +1792,11 @@ class DefaultExecutionContext(ExecutionContext):
             cast(SQLCompiler, self.compiled).postfetch
         )
 
-    def _set_input_sizes(self):
-        """Given a cursor and ClauseParameters, call the appropriate
+    def _prepare_set_input_sizes(
+        self,
+    ) -> Optional[List[Tuple[str, Any, TypeEngine[Any]]]]:
+        """Given a cursor and ClauseParameters, prepare arguments
+        in order to call the appropriate
         style of ``setinputsizes()`` on the cursor, using DB-API types
         from the bind parameter's ``TypeEngine`` objects.
 
@@ -1691,14 +1811,14 @@ class DefaultExecutionContext(ExecutionContext):
 
         """
         if self.isddl or self.is_text:
-            return
+            return None
 
         compiled = cast(SQLCompiler, self.compiled)
 
         inputsizes = compiled._get_set_input_sizes_lookup()
 
         if inputsizes is None:
-            return
+            return None
 
         dialect = self.dialect
 
@@ -1775,12 +1895,8 @@ class DefaultExecutionContext(ExecutionContext):
                 generic_inputsizes.append(
                     (escaped_name, dbtype, bindparam.type)
                 )
-        try:
-            dialect.do_set_input_sizes(self.cursor, generic_inputsizes, self)
-        except BaseException as e:
-            self.root_connection._handle_dbapi_exception(
-                e, None, None, None, self
-            )
+
+        return generic_inputsizes
 
     def _exec_default(self, column, default, type_):
         if default.is_sequence:
@@ -1906,7 +2022,7 @@ class DefaultExecutionContext(ExecutionContext):
         assert compile_state is not None
         if (
             isolate_multiinsert_groups
-            and self.isinsert
+            and dml.isinsert(compile_state)
             and compile_state._has_multi_parameters
         ):
             if column._is_multiparam_column:
diff --git a/lib/sqlalchemy/engine/interfaces.py b/lib/sqlalchemy/engine/interfaces.py
index 01b266d68..fb59acbd0 100644
--- a/lib/sqlalchemy/engine/interfaces.py
+++ b/lib/sqlalchemy/engine/interfaces.py
@@ -18,6 +18,7 @@ from typing import ClassVar
 from typing import Collection
 from typing import Dict
 from typing import Iterable
+from typing import Iterator
 from typing import List
 from typing import Mapping
 from typing import MutableMapping
@@ -62,6 +63,7 @@ if TYPE_CHECKING:
     from ..sql.elements import ClauseElement
     from ..sql.schema import Column
     from ..sql.schema import DefaultGenerator
+    from ..sql.schema import SchemaItem
     from ..sql.schema import Sequence as Sequence_SchemaItem
     from ..sql.sqltypes import Integer
     from ..sql.type_api import _TypeMemoDict
@@ -80,6 +82,28 @@ class CacheStats(Enum):
     NO_DIALECT_SUPPORT = 4
 
 
+class ExecuteStyle(Enum):
+    """indicates the :term:`DBAPI` cursor method that will be used to invoke
+    a statement."""
+
+    EXECUTE = 0
+    """indicates cursor.execute() will be used"""
+
+    EXECUTEMANY = 1
+    """indicates cursor.executemany() will be used."""
+
+    INSERTMANYVALUES = 2
+    """indicates cursor.execute() will be used with an INSERT where the
+    VALUES expression will be expanded to accommodate for multiple
+    parameter sets
+
+    .. seealso::
+
+        :ref:`engine_insertmanyvalues`
+
+    """
+
+
 class DBAPIConnection(Protocol):
     """protocol representing a :pep:`249` database connection.
 
@@ -235,6 +259,8 @@ _ImmutableExecuteOptions = immutabledict[str, Any]
 
 _ParamStyle = Literal["qmark", "numeric", "named", "format", "pyformat"]
 
+_GenericSetInputSizesType = List[Tuple[str, Any, "TypeEngine[Any]"]]
+
 _IsolationLevel = Literal[
     "SERIALIZABLE",
     "REPEATABLE READ",
@@ -608,6 +634,8 @@ class Dialect(EventTarget):
     driver: str
     """identifying name for the dialect's DBAPI"""
 
+    dialect_description: str
+
     dbapi: Optional[ModuleType]
     """A reference to the DBAPI module object itself.
 
@@ -748,23 +776,125 @@ class Dialect(EventTarget):
       executemany.
     """
 
+    supports_empty_insert: bool
+    """dialect supports INSERT () VALUES (), i.e. a plain INSERT with no
+    columns in it.
+
+    This is not usually supported; an "empty" insert is typically
+    suited using either "INSERT..DEFAULT VALUES" or
+    "INSERT ... (col) VALUES (DEFAULT)".
+
+    """
+
     supports_default_values: bool
     """dialect supports INSERT... DEFAULT VALUES syntax"""
 
     supports_default_metavalue: bool
-    """dialect supports INSERT... VALUES (DEFAULT) syntax"""
+    """dialect supports INSERT...(col) VALUES (DEFAULT) syntax.
 
-    supports_empty_insert: bool
-    """dialect supports INSERT () VALUES ()"""
+    Most databases support this in some way, e.g. SQLite supports it using
+    ``VALUES (NULL)``.    MS SQL Server supports the syntax also however
+    is the only included dialect where we have this disabled, as
+    MSSQL does not support the field for the IDENTITY column, which is
+    usually where we like to make use of the feature.
+
+    """
+
+    default_metavalue_token: str = "DEFAULT"
+    """for INSERT... VALUES (DEFAULT) syntax, the token to put in the
+    parenthesis.
+
+    E.g. for SQLite this is the keyword "NULL".
+
+    """
 
     supports_multivalues_insert: bool
     """Target database supports INSERT...VALUES with multiple value
-    sets"""
+    sets, i.e. INSERT INTO table (cols) VALUES (...), (...), (...), ...
+
+    """
+
+    insert_executemany_returning: bool
+    """dialect / driver / database supports some means of providing
+    INSERT...RETURNING support when dialect.do_executemany() is used.
+
+    """
+
+    update_executemany_returning: bool
+    """dialect supports UPDATE..RETURNING with executemany."""
+
+    delete_executemany_returning: bool
+    """dialect supports DELETE..RETURNING with executemany."""
+
+    use_insertmanyvalues: bool
+    """if True, indicates "insertmanyvalues" functionality should be used
+    to allow for ``insert_executemany_returning`` behavior, if possible.
+
+    In practice, setting this to True means:
+
+    if ``supports_multivalues_insert``, ``insert_returning`` and
+    ``use_insertmanyvalues`` are all True, the SQL compiler will produce
+    an INSERT that will be interpreted by the :class:`.DefaultDialect`
+    as an :attr:`.ExecuteStyle.INSERTMANYVALUES` execution that allows
+    for INSERT of many rows with RETURNING by rewriting a single-row
+    INSERT statement to have multiple VALUES clauses, also executing
+    the statement multiple times for a series of batches when large numbers
+    of rows are given.
+
+    The parameter is False for the default dialect, and is set to
+    True for SQLAlchemy internal dialects SQLite, MySQL/MariaDB, PostgreSQL,
+    SQL Server.   It remains at False for Oracle, which provides native
+    "executemany with RETURNING" support and also does not support
+    ``supports_multivalues_insert``.    For MySQL/MariaDB, those MySQL
+    dialects that don't support RETURNING will not report
+    ``insert_executemany_returning`` as True.
+
+    .. versionadded:: 2.0
+
+    .. seealso::
+
+        :ref:`engine_insertmanyvalues`
+
+    """
+
+    use_insertmanyvalues_wo_returning: bool
+    """if True, and use_insertmanyvalues is also True, INSERT statements
+    that don't include RETURNING will also use "insertmanyvalues".
+
+    .. versionadded:: 2.0
+
+    """
+
+    insertmanyvalues_page_size: int
+    """Number of rows to render into an individual INSERT..VALUES() statement
+    for :attr:`.ExecuteStyle.INSERTMANYVALUES` executions.
+
+    The default dialect defaults this to 1000.
+
+    .. versionadded:: 2.0
+
+    .. seealso::
+
+        :paramref:`_engine.Connection.execution_options.insertmanyvalues_page_size` -
+        execution option available on :class:`_engine.Connection`, statements
+
+    """  # noqa: E501
+
+    insertmanyvalues_max_parameters: int
+    """Alternate to insertmanyvalues_page_size, will additionally limit
+    page size based on number of parameters total in the statement.
+
+
+    """
 
     preexecute_autoincrement_sequences: bool
     """True if 'implicit' primary key functions must be executed separately
-      in order to get their value.   This is currently oriented towards
-      PostgreSQL.
+      in order to get their value, if RETURNING is not used.
+
+      This is currently oriented towards PostgreSQL when the
+      ``implicit_returning=False`` parameter is used on a :class:`.Table`
+      object.
+
     """
 
     insert_returning: bool
@@ -810,6 +940,13 @@ class Dialect(EventTarget):
 
     """
 
+    supports_identity_columns: bool
+    """target database supports IDENTITY"""
+
+    cte_follows_insert: bool
+    """target database, when given a CTE with an INSERT statement, needs
+    the CTE to be below the INSERT"""
+
     colspecs: MutableMapping[Type["TypeEngine[Any]"], Type["TypeEngine[Any]"]]
     """A dictionary of TypeEngine classes from sqlalchemy.types mapped
       to subclasses that are specific to the dialect class.  This
@@ -860,7 +997,7 @@ class Dialect(EventTarget):
     """
 
     construct_arguments: Optional[
-        List[Tuple[Type[ClauseElement], Mapping[str, Any]]]
+        List[Tuple[Type[Union[SchemaItem, ClauseElement]], Mapping[str, Any]]]
     ] = None
     """Optional set of argument specifiers for various SQLAlchemy
     constructs, typically schema items.
@@ -1007,19 +1144,6 @@ class Dialect(EventTarget):
 
     _bind_typing_render_casts: bool
 
-    supports_identity_columns: bool
-    """target database supports IDENTITY"""
-
-    cte_follows_insert: bool
-    """target database, when given a CTE with an INSERT statement, needs
-    the CTE to be below the INSERT"""
-
-    insert_executemany_returning: bool
-    """dialect / driver / database supports some means of providing RETURNING
-    support when dialect.do_executemany() is used.
-
-    """
-
     _type_memos: MutableMapping[TypeEngine[Any], "_TypeMemoDict"]
 
     def _builtin_onconnect(self) -> Optional[_ListenerFnType]:
@@ -1826,8 +1950,8 @@ class Dialect(EventTarget):
     def do_set_input_sizes(
         self,
         cursor: DBAPICursor,
-        list_of_tuples: List[Tuple[str, Any, TypeEngine[Any]]],
-        context: "ExecutionContext",
+        list_of_tuples: _GenericSetInputSizesType,
+        context: ExecutionContext,
     ) -> Any:
         """invoke the cursor.setinputsizes() method with appropriate arguments
 
@@ -1961,12 +2085,35 @@ class Dialect(EventTarget):
 
         raise NotImplementedError()
 
+    def _deliver_insertmanyvalues_batches(
+        self,
+        cursor: DBAPICursor,
+        statement: str,
+        parameters: _DBAPIMultiExecuteParams,
+        generic_setinputsizes: Optional[_GenericSetInputSizesType],
+        context: ExecutionContext,
+    ) -> Iterator[
+        Tuple[
+            str,
+            _DBAPISingleExecuteParams,
+            _GenericSetInputSizesType,
+            int,
+            int,
+        ]
+    ]:
+        """convert executemany parameters for an INSERT into an iterator
+        of statement/single execute values, used by the insertmanyvalues
+        feature.
+
+        """
+        raise NotImplementedError()
+
     def do_executemany(
         self,
         cursor: DBAPICursor,
         statement: str,
         parameters: _DBAPIMultiExecuteParams,
-        context: Optional["ExecutionContext"] = None,
+        context: Optional[ExecutionContext] = None,
     ) -> None:
         """Provide an implementation of ``cursor.executemany(statement,
         parameters)``."""
@@ -2743,7 +2890,9 @@ class ExecutionContext:
 
     These are always stored as a list of parameter entries.  A single-element
     list corresponds to a ``cursor.execute()`` call and a multiple-element
-    list corresponds to ``cursor.executemany()``.
+    list corresponds to ``cursor.executemany()``, except in the case
+    of :attr:`.ExecuteStyle.INSERTMANYVALUES` which will use
+    ``cursor.execute()`` one or more times.
 
     """
 
@@ -2756,8 +2905,23 @@ class ExecutionContext:
     isupdate: bool
     """True if the statement is an UPDATE."""
 
+    execute_style: ExecuteStyle
+    """the style of DBAPI cursor method that will be used to execute
+    a statement.
+
+    .. versionadded:: 2.0
+
+    """
+
     executemany: bool
-    """True if the parameters have determined this to be an executemany"""
+    """True if the context has a list of more than one parameter set.
+
+    Historically this attribute links to whether ``cursor.execute()`` or
+    ``cursor.executemany()`` will be used.  It also can now mean that
+    "insertmanyvalues" may be used which indicates one or more
+    ``cursor.execute()`` calls.
+
+    """
 
     prefetch_cols: util.generic_fn_descriptor[Optional[Sequence[Column[Any]]]]
     """a list of Column objects for which a client-side default
@@ -2824,7 +2988,9 @@ class ExecutionContext:
     ) -> Any:
         raise NotImplementedError()
 
-    def _set_input_sizes(self) -> None:
+    def _prepare_set_input_sizes(
+        self,
+    ) -> Optional[List[Tuple[str, Any, TypeEngine[Any]]]]:
         raise NotImplementedError()
 
     def _get_cache_stats(self) -> str: