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

2 files changed, 66 insertions, 22 deletions

diff --git a/examples/performance/__init__.py b/examples/performance/__init__.py
index bdf8e841d..31b1a152a 100644
--- a/examples/performance/__init__.py
+++ b/examples/performance/__init__.py
@@ -268,9 +268,9 @@ class Profiler:
 
     def run(self):
         if self.test:
-            tests = [fn for fn in self.tests if fn.__name__ == self.test]
+            tests = [fn for fn in self.tests if fn.__name__ in self.test]
             if not tests:
-                raise ValueError("No such test: %s" % self.test)
+                raise ValueError("No such test(s): %s" % self.test)
         else:
             tests = self.tests
 
@@ -333,7 +333,9 @@ class Profiler:
                 except ImportError:
                     pass
 
-        parser.add_argument("--test", type=str, help="run specific test name")
+        parser.add_argument(
+            "--test", nargs="+", type=str, help="run specific test(s)"
+        )
 
         parser.add_argument(
             "--dburl",
diff --git a/examples/performance/bulk_inserts.py b/examples/performance/bulk_inserts.py
index 49469581d..3046481d1 100644
--- a/examples/performance/bulk_inserts.py
+++ b/examples/performance/bulk_inserts.py
@@ -1,25 +1,29 @@
-"""This series of tests illustrates different ways to INSERT a large number
-of rows in bulk.
-
+from __future__ import annotations
 
-"""
 from sqlalchemy import bindparam
 from sqlalchemy import Column
 from sqlalchemy import create_engine
+from sqlalchemy import Identity
+from sqlalchemy import insert
 from sqlalchemy import Integer
+from sqlalchemy import select
 from sqlalchemy import String
-from sqlalchemy.ext.declarative import declarative_base
+from sqlalchemy.orm import declarative_base
 from sqlalchemy.orm import Session
 from . import Profiler
 
+"""This series of tests illustrates different ways to INSERT a large number
+of rows in bulk.
+
+
+"""
 
 Base = declarative_base()
-engine = None
 
 
 class Customer(Base):
     __tablename__ = "customer"
-    id = Column(Integer, primary_key=True)
+    id = Column(Integer, Identity(), primary_key=True)
     name = Column(String(255))
     description = Column(String(255))
 
@@ -37,7 +41,8 @@ def setup_database(dburl, echo, num):
 
 @Profiler.profile
 def test_flush_no_pk(n):
-    """Individual INSERT statements via the ORM, calling upon last row id"""
+    """INSERT statements via the ORM (batched with RETURNING if available),
+    fetching generated row id"""
     session = Session(bind=engine)
     for chunk in range(0, n, 1000):
         session.add_all(
@@ -55,7 +60,8 @@ def test_flush_no_pk(n):
 
 @Profiler.profile
 def test_bulk_save_return_pks(n):
-    """Individual INSERT statements in "bulk", but calling upon last row id"""
+    """INSERT statements in "bulk" (batched with RETURNING if available),
+    fetching generated row id"""
     session = Session(bind=engine)
     session.bulk_save_objects(
         [
@@ -106,12 +112,12 @@ def test_bulk_save(n):
 
 
 @Profiler.profile
-def test_bulk_insert_mappings(n):
-    """Batched INSERT statements via the ORM "bulk", using dictionaries."""
+def test_orm_insert(n):
+    """A single Core INSERT run through the Session"""
     session = Session(bind=engine)
-    session.bulk_insert_mappings(
-        Customer,
-        [
+    session.execute(
+        insert(Customer),
+        params=[
             dict(
                 name="customer name %d" % i,
                 description="customer description %d" % i,
@@ -123,11 +129,30 @@ def test_bulk_insert_mappings(n):
 
 
 @Profiler.profile
-def test_core_insert(n):
-    """A single Core INSERT construct inserting mappings in bulk."""
-    conn = engine.connect()
-    conn.execute(
-        Customer.__table__.insert(),
+def test_orm_insert_w_fetch(n):
+    """A single Core INSERT w executemany run through the Session, fetching
+    back new Customer objects into a list"""
+    session = Session(bind=engine)
+    result = session.execute(
+        select(Customer).from_statement(insert(Customer).returning(Customer)),
+        params=[
+            dict(
+                name="customer name %d" % i,
+                description="customer description %d" % i,
+            )
+            for i in range(n)
+        ],
+    )
+    customers = result.scalars().all()  # noqa: F841
+    session.commit()
+
+
+@Profiler.profile
+def test_bulk_insert_mappings(n):
+    """Batched INSERT statements via the ORM "bulk", using dictionaries."""
+    session = Session(bind=engine)
+    session.bulk_insert_mappings(
+        Customer,
         [
             dict(
                 name="customer name %d" % i,
@@ -136,6 +161,23 @@ def test_core_insert(n):
             for i in range(n)
         ],
     )
+    session.commit()
+
+
+@Profiler.profile
+def test_core_insert(n):
+    """A single Core INSERT construct inserting mappings in bulk."""
+    with engine.begin() as conn:
+        conn.execute(
+            Customer.__table__.insert(),
+            [
+                dict(
+                    name="customer name %d" % i,
+                    description="customer description %d" % i,
+                )
+                for i in range(n)
+            ],
+        )
 
 
 @Profiler.profile