Add fast execution helper support.

Added a new flag ``use_batch_mode`` to the psycopg2 dialect.  This flag
enables the use of psycopg2's ``psycopg2.extras.execute_batch``
extension when the :class:`.Engine` calls upon ``cursor.executemany()``.
This extension provides a critical performance increase by over an order of magnitude
when running INSERT statements in batch.  The flag is False by default
as it is considered to be experimental for now.

Change-Id: Ib88d28bc792958d47109f644ff1d08c897db4ff7
Fixes: #4109

1 files changed, 82 insertions, 1 deletions

diff --git a/test/dialect/postgresql/test_dialect.py b/test/dialect/postgresql/test_dialect.py
index e985718c7..29aa62e3f 100644
--- a/test/dialect/postgresql/test_dialect.py
+++ b/test/dialect/postgresql/test_dialect.py
@@ -9,7 +9,7 @@ import datetime
 from sqlalchemy import (
     Table, Column, select, MetaData, text, Integer, String, Sequence, Numeric,
     DateTime, BigInteger, func, extract, SmallInteger, TypeDecorator, literal,
-    cast)
+    cast, bindparam)
 from sqlalchemy import exc, schema
 from sqlalchemy.dialects.postgresql import base as postgresql
 import logging
@@ -84,6 +84,87 @@ class DialectTest(fixtures.TestBase):
         eq_(e.dialect.use_native_unicode, True)
 
 
+class BatchInsertsTest(fixtures.TablesTest):
+    __only_on__ = 'postgresql+psycopg2'
+    __backend__ = True
+
+    run_create_tables = "each"
+
+    @classmethod
+    def define_tables(cls, metadata):
+        Table(
+            'data', metadata,
+            Column('id', Integer, primary_key=True),
+            Column('x', String),
+            Column('y', String),
+            Column('z', Integer, server_default="5")
+        )
+
+    def setup(self):
+        super(BatchInsertsTest, self).setup()
+        self.engine = engines.testing_engine(options={"use_batch_mode": True})
+
+    def teardown(self):
+        self.engine.dispose()
+        super(BatchInsertsTest, self).teardown()
+
+    def test_insert(self):
+        with self.engine.connect() as conn:
+            conn.execute(
+                self.tables.data.insert(),
+                [
+                    {"x": "x1", "y": "y1"},
+                    {"x": "x2", "y": "y2"},
+                    {"x": "x3", "y": "y3"}
+                ]
+            )
+
+            eq_(
+                conn.execute(select([self.tables.data])).fetchall(),
+                [
+                    (1, "x1", "y1", 5),
+                    (2, "x2", "y2", 5),
+                    (3, "x3", "y3", 5)
+                ]
+            )
+
+    def test_not_sane_rowcount(self):
+        self.engine.connect().close()
+        assert not self.engine.dialect.supports_sane_multi_rowcount
+
+    def test_update(self):
+        with self.engine.connect() as conn:
+            conn.execute(
+                self.tables.data.insert(),
+                [
+                    {"x": "x1", "y": "y1"},
+                    {"x": "x2", "y": "y2"},
+                    {"x": "x3", "y": "y3"}
+                ]
+            )
+
+            conn.execute(
+                self.tables.data.update().
+                where(self.tables.data.c.x == bindparam('xval')).
+                values(y=bindparam('yval')),
+                [
+                    {"xval": "x1", "yval": "y5"},
+                    {"xval": "x3", "yval": "y6"}
+                ]
+            )
+            eq_(
+                conn.execute(
+                    select([self.tables.data]).
+                    order_by(self.tables.data.c.id)).
+                fetchall(),
+                [
+                    (1, "x1", "y5", 5),
+                    (2, "x2", "y2", 5),
+                    (3, "x3", "y6", 5)
+                ]
+            )
+
+
 class MiscBackendTest(
         fixtures.TestBase, AssertsExecutionResults, AssertsCompiledSQL):