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

1 files changed, 111 insertions, 13 deletions

diff --git a/lib/sqlalchemy/sql/util.py b/lib/sqlalchemy/sql/util.py
index 0400ab3fe..55c6a35f8 100644
--- a/lib/sqlalchemy/sql/util.py
+++ b/lib/sqlalchemy/sql/util.py
@@ -598,6 +598,21 @@ class _repr_row(_repr_base):
         )
 
 
+class _long_statement(str):
+    def __str__(self) -> str:
+        lself = len(self)
+        if lself > 500:
+            lleft = 250
+            lright = 100
+            trunc = lself - lleft - lright
+            return (
+                f"{self[0:lleft]} ... {trunc} "
+                f"characters truncated ... {self[-lright:]}"
+            )
+        else:
+            return str.__str__(self)
+
+
 class _repr_params(_repr_base):
     """Provide a string view of bound parameters.
 
@@ -606,12 +621,13 @@ class _repr_params(_repr_base):
 
     """
 
-    __slots__ = "params", "batches", "ismulti"
+    __slots__ = "params", "batches", "ismulti", "max_params"
 
     def __init__(
         self,
         params: Optional[_AnyExecuteParams],
         batches: int,
+        max_params: int = 100,
         max_chars: int = 300,
         ismulti: Optional[bool] = None,
     ):
@@ -619,6 +635,7 @@ class _repr_params(_repr_base):
         self.ismulti = ismulti
         self.batches = batches
         self.max_chars = max_chars
+        self.max_params = max_params
 
     def __repr__(self) -> str:
         if self.ismulti is None:
@@ -693,29 +710,110 @@ class _repr_params(_repr_base):
         else:
             return "(%s)" % elements
 
+    def _get_batches(self, params: Iterable[Any]) -> Any:
+
+        lparams = list(params)
+        lenparams = len(lparams)
+        if lenparams > self.max_params:
+            lleft = self.max_params // 2
+            return (
+                lparams[0:lleft],
+                lparams[-lleft:],
+                lenparams - self.max_params,
+            )
+        else:
+            return lparams, None, None
+
     def _repr_params(
         self,
         params: _AnySingleExecuteParams,
         typ: int,
     ) -> str:
-        trunc = self.trunc
         if typ is self._DICT:
-            return "{%s}" % (
+            return self._repr_param_dict(
+                cast("_CoreSingleExecuteParams", params)
+            )
+        elif typ is self._TUPLE:
+            return self._repr_param_tuple(cast("Sequence[Any]", params))
+        else:
+            return self._repr_param_list(params)
+
+    def _repr_param_dict(self, params: _CoreSingleExecuteParams) -> str:
+        trunc = self.trunc
+        (
+            items_first_batch,
+            items_second_batch,
+            trunclen,
+        ) = self._get_batches(params.items())
+
+        if items_second_batch:
+            text = "{%s" % (
                 ", ".join(
-                    "%r: %s" % (key, trunc(value))
-                    for key, value in cast(
-                        "_CoreSingleExecuteParams", params
-                    ).items()
+                    f"{key!r}: {trunc(value)}"
+                    for key, value in items_first_batch
                 )
             )
-        elif typ is self._TUPLE:
-            seq_params = cast("Sequence[Any]", params)
-            return "(%s%s)" % (
-                ", ".join(trunc(value) for value in seq_params),
-                "," if len(seq_params) == 1 else "",
+            text += f" ... {trunclen} parameters truncated ... "
+            text += "%s}" % (
+                ", ".join(
+                    f"{key!r}: {trunc(value)}"
+                    for key, value in items_second_batch
+                )
             )
         else:
-            return "[%s]" % (", ".join(trunc(value) for value in params))
+            text = "{%s}" % (
+                ", ".join(
+                    f"{key!r}: {trunc(value)}"
+                    for key, value in items_first_batch
+                )
+            )
+        return text
+
+    def _repr_param_tuple(self, params: "Sequence[Any]") -> str:
+        trunc = self.trunc
+
+        (
+            items_first_batch,
+            items_second_batch,
+            trunclen,
+        ) = self._get_batches(params)
+
+        if items_second_batch:
+            text = "(%s" % (
+                ", ".join(trunc(value) for value in items_first_batch)
+            )
+            text += f" ... {trunclen} parameters truncated ... "
+            text += "%s)" % (
+                ", ".join(trunc(value) for value in items_second_batch),
+            )
+        else:
+            text = "(%s%s)" % (
+                ", ".join(trunc(value) for value in items_first_batch),
+                "," if len(items_first_batch) == 1 else "",
+            )
+        return text
+
+    def _repr_param_list(self, params: _AnySingleExecuteParams) -> str:
+        trunc = self.trunc
+        (
+            items_first_batch,
+            items_second_batch,
+            trunclen,
+        ) = self._get_batches(params)
+
+        if items_second_batch:
+            text = "[%s" % (
+                ", ".join(trunc(value) for value in items_first_batch)
+            )
+            text += f" ... {trunclen} parameters truncated ... "
+            text += "%s]" % (
+                ", ".join(trunc(value) for value in items_second_batch)
+            )
+        else:
+            text = "[%s]" % (
+                ", ".join(trunc(value) for value in items_first_batch)
+            )
+        return text
 
 
 def adapt_criterion_to_null(crit: _CE, nulls: Collection[Any]) -> _CE: