Default psycopg2 executemany mode to "values_only"

The psycopg2 dialect now defaults to using the very performant
``execute_values()`` psycopg2 extension for compiled INSERT statements,
and also impements RETURNING support when this extension is used.  This
allows INSERT statements that even include an autoincremented SERIAL
or IDENTITY value to run very fast while still being able to return the
newly generated primary key values.   The ORM will then integrate this
new feature in a separate change.

Implements RETURNING for insert with executemany

Adds support to return_defaults() mode and inserted_primary_key
to support mutiple INSERTed rows, via return_defauls_rows
and inserted_primary_key_rows accessors.

within default execution context, new cached compiler
getters are used to fetch primary keys from rows

inserted_primary_key now returns a plain tuple. this
is not yet a row-like object however this can be
added.

Adds distinct "values_only" and "batch" modes, as
"values" has a lot of benefits but "batch" breaks
cursor.rowcount

psycopg2 minimum version 2.7 so we can remove the
large number of checks for very old versions of
psycopg2

simplify tests to no longer distinguish between
native and non-native json

Fixes: #5401
Change-Id: Ic08fd3423d4c5d16ca50994460c0c234868bd61c

8 files changed, 146 insertions, 38 deletions

diff --git a/test/sql/test_defaults.py b/test/sql/test_defaults.py
index fa6c4d9a1..676c46db6 100644
--- a/test/sql/test_defaults.py
+++ b/test/sql/test_defaults.py
@@ -977,11 +977,11 @@ class PKDefaultTest(fixtures.TablesTest):
         with engine.begin() as conn:
             conn.execute(t2.insert(), nextid=1)
             r = conn.execute(t1.insert(), data="hi")
-            eq_([1], r.inserted_primary_key)
+            eq_((1,), r.inserted_primary_key)
 
             conn.execute(t2.insert(), nextid=2)
             r = conn.execute(t1.insert(), data="there")
-            eq_([2], r.inserted_primary_key)
+            eq_((2,), r.inserted_primary_key)
 
             r = conn.execute(date_table.insert())
             assert isinstance(r.inserted_primary_key[0], datetime.datetime)
@@ -1273,10 +1273,10 @@ class SpecialTypePKTest(fixtures.TestBase):
                 not testing.db.dialect.implicit_returning
                 or not implicit_returning
             ):
-                eq_(r.inserted_primary_key, [None])
+                eq_(r.inserted_primary_key, (None,))
             else:
                 eq_(
-                    r.inserted_primary_key, [expected_result],
+                    r.inserted_primary_key, (expected_result,),
                 )
 
             eq_(
@@ -1350,7 +1350,7 @@ class ServerDefaultsOnPKTest(fixtures.TestBase):
         )
         metadata.create_all(connection)
         r = connection.execute(t.insert(), dict(data="data"))
-        eq_(r.inserted_primary_key, [None])
+        eq_(r.inserted_primary_key, (None,))
         eq_(list(connection.execute(t.select())), [("key_one", "data")])
 
     @testing.requires.returning
@@ -1370,7 +1370,7 @@ class ServerDefaultsOnPKTest(fixtures.TestBase):
         )
         metadata.create_all(connection)
         r = connection.execute(t.insert(), dict(data="data"))
-        eq_(r.inserted_primary_key, ["key_one"])
+        eq_(r.inserted_primary_key, ("key_one",))
         eq_(list(connection.execute(t.select())), [("key_one", "data")])
 
     @testing.provide_metadata
@@ -1386,7 +1386,7 @@ class ServerDefaultsOnPKTest(fixtures.TestBase):
         assert t._autoincrement_column is None
         metadata.create_all(connection)
         r = connection.execute(t.insert(), dict(data="data"))
-        eq_(r.inserted_primary_key, [None])
+        eq_(r.inserted_primary_key, (None,))
         if testing.against("sqlite"):
             eq_(list(connection.execute(t.select())), [(1, "data")])
         else:
@@ -1425,7 +1425,7 @@ class ServerDefaultsOnPKTest(fixtures.TestBase):
         t2 = Table("x", m2, autoload_with=connection, implicit_returning=False)
 
         r = connection.execute(t2.insert(), dict(data="data"))
-        eq_(r.inserted_primary_key, [None])
+        eq_(r.inserted_primary_key, (None,))
         if testing.against("sqlite"):
             eq_(list(connection.execute(t2.select())), [(1, "data")])
         else:
@@ -1444,7 +1444,7 @@ class ServerDefaultsOnPKTest(fixtures.TestBase):
 
         metadata.create_all(connection)
         r = connection.execute(t.insert(), dict(data="data"))
-        eq_(r.inserted_primary_key, [5])
+        eq_(r.inserted_primary_key, (5,))
         eq_(list(connection.execute(t.select())), [(5, "data")])
 
 
diff --git a/test/sql/test_deprecations.py b/test/sql/test_deprecations.py
index 4f018fcc9..635f009f6 100644
--- a/test/sql/test_deprecations.py
+++ b/test/sql/test_deprecations.py
@@ -1511,7 +1511,6 @@ class DMLTest(fixtures.TestBase, AssertsCompiledSQL):
             "INSERT INTO test (col1, col2) VALUES (foo(:foo_1), "
             "(SELECT coalesce(max(foo.id)) AS coalesce_1 FROM "
             "foo))",
-            inline_flag=True,
         )
 
     def test_insert_inline_kw_default(self):
@@ -1529,9 +1528,7 @@ class DMLTest(fixtures.TestBase, AssertsCompiledSQL):
             stmt = table.insert(values={}, inline=True)
 
         self.assert_compile(
-            stmt,
-            "INSERT INTO sometable (foo) VALUES (foobar())",
-            inline_flag=True,
+            stmt, "INSERT INTO sometable (foo) VALUES (foobar())",
         )
 
         with testing.expect_deprecated_20(
@@ -1540,10 +1537,7 @@ class DMLTest(fixtures.TestBase, AssertsCompiledSQL):
             stmt = table.insert(inline=True)
 
         self.assert_compile(
-            stmt,
-            "INSERT INTO sometable (foo) VALUES (foobar())",
-            params={},
-            inline_flag=True,
+            stmt, "INSERT INTO sometable (foo) VALUES (foobar())", params={},
         )
 
     def test_update_inline_kw_defaults(self):
@@ -1572,7 +1566,6 @@ class DMLTest(fixtures.TestBase, AssertsCompiledSQL):
             "UPDATE test SET col1=foo(:foo_1), col2=(SELECT "
             "coalesce(max(foo.id)) AS coalesce_1 FROM foo), "
             "col3=:col3",
-            inline_flag=True,
         )
 
     def test_update_dialect_kwargs(self):
diff --git a/test/sql/test_functions.py b/test/sql/test_functions.py
index 4b3555391..b37cb15b7 100644
--- a/test/sql/test_functions.py
+++ b/test/sql/test_functions.py
@@ -776,6 +776,8 @@ class CompileTest(fixtures.TestBase, AssertsCompiledSQL):
         )
 
     def test_incorrect_none_type(self):
+        from sqlalchemy.sql.expression import FunctionElement
+
         class MissingType(FunctionElement):
             name = "mt"
             type = None
@@ -784,7 +786,7 @@ class CompileTest(fixtures.TestBase, AssertsCompiledSQL):
             TypeError,
             "Object None associated with '.type' attribute is "
             "not a TypeEngine class or object",
-            MissingType().compile,
+            lambda: column("x", MissingType()) == 5,
         )
 
     def test_as_comparison(self):
diff --git a/test/sql/test_insert.py b/test/sql/test_insert.py
index 8a067b65a..c7749e74c 100644
--- a/test/sql/test_insert.py
+++ b/test/sql/test_insert.py
@@ -140,7 +140,6 @@ class InsertTest(_InsertTestBase, fixtures.TablesTest, AssertsCompiledSQL):
             "INSERT INTO test (col1, col2) VALUES (foo(:foo_1), "
             "(SELECT coalesce(max(foo.id)) AS coalesce_1 FROM "
             "foo))",
-            inline_flag=False,
         )
 
         self.assert_compile(
@@ -148,7 +147,6 @@ class InsertTest(_InsertTestBase, fixtures.TablesTest, AssertsCompiledSQL):
             "INSERT INTO test (col1, col2) VALUES (foo(:foo_1), "
             "(SELECT coalesce(max(foo.id)) AS coalesce_1 FROM "
             "foo))",
-            inline_flag=True,
         )
 
     def test_generic_insert_bind_params_all_columns(self):
@@ -302,27 +300,23 @@ class InsertTest(_InsertTestBase, fixtures.TablesTest, AssertsCompiledSQL):
         self.assert_compile(
             table.insert().values(),
             "INSERT INTO sometable (foo) VALUES (foobar())",
-            inline_flag=False,
         )
 
         self.assert_compile(
             table.insert(),
             "INSERT INTO sometable (foo) VALUES (foobar())",
             params={},
-            inline_flag=False,
         )
 
         self.assert_compile(
             table.insert().values().inline(),
             "INSERT INTO sometable (foo) VALUES (foobar())",
-            inline_flag=True,
         )
 
         self.assert_compile(
             table.insert().inline(),
             "INSERT INTO sometable (foo) VALUES (foobar())",
             params={},
-            inline_flag=True,
         )
 
     def test_insert_returning_not_in_default(self):
diff --git a/test/sql/test_insert_exec.py b/test/sql/test_insert_exec.py
index a081766bc..becca12ff 100644
--- a/test/sql/test_insert_exec.py
+++ b/test/sql/test_insert_exec.py
@@ -260,7 +260,7 @@ class InsertExecTest(fixtures.TablesTest):
         )
         t.create(eng)
         r = eng.execute(t.insert().values(y=5))
-        eq_(r.inserted_primary_key, [0])
+        eq_(r.inserted_primary_key, (0,))
 
     @testing.fails_on(
         "sqlite", "sqlite autoincrement doesn't work with composite pks"
@@ -299,7 +299,7 @@ class InsertExecTest(fixtures.TablesTest):
         eq_(id_, 12)
 
         r = t6.insert().values(manual_id=id_).execute()
-        eq_(r.inserted_primary_key, [12, 1])
+        eq_(r.inserted_primary_key, (12, 1))
 
     def test_implicit_id_insert_select_columns(self):
         users = self.tables.users
@@ -409,7 +409,7 @@ class TableInsertTest(fixtures.TablesTest):
         self._test(
             t.insert().values(id=1, data="data", x=5),
             (1, "data", 5),
-            inserted_primary_key=[1],
+            inserted_primary_key=(1,),
         )
 
     def test_uppercase_inline(self):
@@ -417,7 +417,7 @@ class TableInsertTest(fixtures.TablesTest):
         self._test(
             t.insert().inline().values(id=1, data="data", x=5),
             (1, "data", 5),
-            inserted_primary_key=[1],
+            inserted_primary_key=(1,),
         )
 
     @testing.crashes(
@@ -429,7 +429,7 @@ class TableInsertTest(fixtures.TablesTest):
         self._test(
             t.insert().inline().values(data="data", x=5),
             (1, "data", 5),
-            inserted_primary_key=[None],
+            inserted_primary_key=(None,),
         )
 
     def test_uppercase_implicit(self):
@@ -437,7 +437,7 @@ class TableInsertTest(fixtures.TablesTest):
         self._test(
             t.insert().values(data="data", x=5),
             (testing.db.dialect.default_sequence_base, "data", 5),
-            inserted_primary_key=[testing.db.dialect.default_sequence_base],
+            inserted_primary_key=(testing.db.dialect.default_sequence_base,),
         )
 
     def test_uppercase_direct_params(self):
@@ -445,7 +445,7 @@ class TableInsertTest(fixtures.TablesTest):
         self._test(
             t.insert().values(id=1, data="data", x=5),
             (1, "data", 5),
-            inserted_primary_key=[1],
+            inserted_primary_key=(1,),
         )
 
     @testing.requires.returning
@@ -462,7 +462,7 @@ class TableInsertTest(fixtures.TablesTest):
         self._test(
             t.insert().values(id=1, data="data", x=5),
             (1, "data", 5),
-            inserted_primary_key=[],
+            inserted_primary_key=(),
         )
 
     @testing.requires.returning
@@ -481,7 +481,7 @@ class TableInsertTest(fixtures.TablesTest):
         self._test(
             t.insert().values(data="data", x=5),
             (testing.db.dialect.default_sequence_base, "data", 5),
-            inserted_primary_key=[],
+            inserted_primary_key=(),
         )
 
     @testing.requires.emulated_lastrowid_even_with_sequences
@@ -505,5 +505,5 @@ class TableInsertTest(fixtures.TablesTest):
         self._test(
             t.insert().inline().values(data="data", x=5),
             (testing.db.dialect.default_sequence_base, "data", 5),
-            inserted_primary_key=[],
+            inserted_primary_key=(),
         )
diff --git a/test/sql/test_returning.py b/test/sql/test_returning.py
index 90c21ed45..7d60dd475 100644
--- a/test/sql/test_returning.py
+++ b/test/sql/test_returning.py
@@ -441,6 +441,7 @@ class ReturnDefaultsTest(fixtures.TablesTest):
             dict(result.returned_defaults._mapping),
             {"id": 1, "data": None, "insdef": 0},
         )
+        eq_(result.inserted_primary_key, (1,))
 
     def test_update_non_default_plus_default(self, connection):
         t1 = self.tables.t1
@@ -464,6 +465,7 @@ class ReturnDefaultsTest(fixtures.TablesTest):
             dict(result.returned_defaults._mapping),
             {"id": 1, "data": None, "insdef": 0},
         )
+        eq_(result.inserted_primary_key, (1,))
 
     def test_update_all(self, connection):
         t1 = self.tables.t1
@@ -473,6 +475,125 @@ class ReturnDefaultsTest(fixtures.TablesTest):
         )
         eq_(dict(result.returned_defaults._mapping), {"upddef": 1})
 
+    @testing.requires.insert_executemany_returning
+    def test_insert_executemany_no_defaults_passed(self, connection):
+        t1 = self.tables.t1
+        result = connection.execute(
+            t1.insert().return_defaults(),
+            [
+                {"data": "d1"},
+                {"data": "d2"},
+                {"data": "d3"},
+                {"data": "d4"},
+                {"data": "d5"},
+                {"data": "d6"},
+            ],
+        )
+
+        eq_(
+            [row._mapping for row in result.returned_defaults_rows],
+            [
+                {"id": 1, "insdef": 0, "upddef": None},
+                {"id": 2, "insdef": 0, "upddef": None},
+                {"id": 3, "insdef": 0, "upddef": None},
+                {"id": 4, "insdef": 0, "upddef": None},
+                {"id": 5, "insdef": 0, "upddef": None},
+                {"id": 6, "insdef": 0, "upddef": None},
+            ],
+        )
+
+        eq_(
+            result.inserted_primary_key_rows,
+            [(1,), (2,), (3,), (4,), (5,), (6,)],
+        )
+        assert_raises_message(
+            sa_exc.InvalidRequestError,
+            "This statement was an executemany call; "
+            "if return defaults is supported",
+            lambda: result.returned_defaults,
+        )
+        assert_raises_message(
+            sa_exc.InvalidRequestError,
+            "This statement was an executemany call; "
+            "if primary key returning is supported",
+            lambda: result.inserted_primary_key,
+        )
+
+    @testing.requires.insert_executemany_returning
+    def test_insert_executemany_insdefault_passed(self, connection):
+        t1 = self.tables.t1
+        result = connection.execute(
+            t1.insert().return_defaults(),
+            [
+                {"data": "d1", "insdef": 11},
+                {"data": "d2", "insdef": 12},
+                {"data": "d3", "insdef": 13},
+                {"data": "d4", "insdef": 14},
+                {"data": "d5", "insdef": 15},
+                {"data": "d6", "insdef": 16},
+            ],
+        )
+
+        eq_(
+            [row._mapping for row in result.returned_defaults_rows],
+            [
+                {"id": 1, "upddef": None},
+                {"id": 2, "upddef": None},
+                {"id": 3, "upddef": None},
+                {"id": 4, "upddef": None},
+                {"id": 5, "upddef": None},
+                {"id": 6, "upddef": None},
+            ],
+        )
+
+        eq_(
+            result.inserted_primary_key_rows,
+            [(1,), (2,), (3,), (4,), (5,), (6,)],
+        )
+        assert_raises_message(
+            sa_exc.InvalidRequestError,
+            "This statement was an executemany call; "
+            "if return defaults is supported",
+            lambda: result.returned_defaults,
+        )
+        assert_raises_message(
+            sa_exc.InvalidRequestError,
+            "This statement was an executemany call; "
+            "if primary key returning is supported",
+            lambda: result.inserted_primary_key,
+        )
+
+    @testing.requires.insert_executemany_returning
+    def test_insert_executemany_only_pk_passed(self, connection):
+        t1 = self.tables.t1
+        result = connection.execute(
+            t1.insert().return_defaults(),
+            [
+                {"id": 10, "data": "d1"},
+                {"id": 11, "data": "d2"},
+                {"id": 12, "data": "d3"},
+                {"id": 13, "data": "d4"},
+                {"id": 14, "data": "d5"},
+                {"id": 15, "data": "d6"},
+            ],
+        )
+
+        eq_(
+            [row._mapping for row in result.returned_defaults_rows],
+            [
+                {"insdef": 0, "upddef": None},
+                {"insdef": 0, "upddef": None},
+                {"insdef": 0, "upddef": None},
+                {"insdef": 0, "upddef": None},
+                {"insdef": 0, "upddef": None},
+                {"insdef": 0, "upddef": None},
+            ],
+        )
+        eq_(
+            result.inserted_primary_key_rows,
+            [(10,), (11,), (12,), (13,), (14,), (15,)],
+        )
+
 
 class ImplicitReturningFlag(fixtures.TestBase):
     __backend__ = True
diff --git a/test/sql/test_sequences.py b/test/sql/test_sequences.py
index 126f852ba..8d894f9f3 100644
--- a/test/sql/test_sequences.py
+++ b/test/sql/test_sequences.py
@@ -401,7 +401,7 @@ class SequenceTest(fixtures.TestBase, testing.AssertsCompiledSQL):
         self.metadata.create_all(connection)
 
         result = connection.execute(t.insert())
-        eq_(result.inserted_primary_key, [1])
+        eq_(result.inserted_primary_key, (1,))
 
 
 class FutureSequenceTest(fixtures.FutureEngineMixin, SequenceTest):
diff --git a/test/sql/test_update.py b/test/sql/test_update.py
index 664862dcb..18e9da654 100644
--- a/test/sql/test_update.py
+++ b/test/sql/test_update.py
@@ -401,7 +401,6 @@ class UpdateTest(_UpdateFromTestBase, fixtures.TablesTest, AssertsCompiledSQL):
             "UPDATE test SET col1=foo(:foo_1), col2=(SELECT "
             "coalesce(max(foo.id)) AS coalesce_1 FROM foo), "
             "col3=:col3",
-            inline_flag=False,
         )
 
         self.assert_compile(
@@ -409,7 +408,6 @@ class UpdateTest(_UpdateFromTestBase, fixtures.TablesTest, AssertsCompiledSQL):
             "UPDATE test SET col1=foo(:foo_1), col2=(SELECT "
             "coalesce(max(foo.id)) AS coalesce_1 FROM foo), "
             "col3=:col3",
-            inline_flag=True,
         )
 
     def test_update_1(self):