Add "render_nulls" flag to bulk_insert as optional performance optimization

Currently, ``Session.bulk_insert_mappings`` omits NULL values which
causes it to break up batches of inserts based on which batches
contain NULL and which do not.

By adding this flag, the same columns are rendered in the INSERT
for all rows allowing them to be batched.  The downside is that
server-side defaults are omitted.

Doctext-author: Mike Bayer <mike_mp@zzzcomputing.com>
Change-Id: Iec5969304d4bdbf57290b200331bde02254aa3a5
Pull-request: https://github.com/zzzeek/sqlalchemy/pull/243

4 files changed, 73 insertions, 12 deletions

diff --git a/doc/build/changelog/changelog_11.rst b/doc/build/changelog/changelog_11.rst
index 789a241d0..297be5d2d 100644
--- a/doc/build/changelog/changelog_11.rst
+++ b/doc/build/changelog/changelog_11.rst
@@ -22,6 +22,15 @@
     :version: 1.1.0b1
 
     .. change::
+        :tags: feature, orm
+
+        Added new flag :paramref:`.Session.bulk_insert_mappings.render_nulls`
+        which allows an ORM bulk INSERT to occur with NULL values rendered;
+        this bypasses server side defaults, however allows all statements
+        to be formed with the same set of columns, allowing them to be
+        batched.  Pull request courtesy Tobias Sauerwein.
+
+    .. change::
         :tags: feature, mssql
 
         The ``mssql_clustered`` flag available on :class:`.UniqueConstraint`,
diff --git a/lib/sqlalchemy/orm/persistence.py b/lib/sqlalchemy/orm/persistence.py
index a5e0d9d95..0b029f466 100644
--- a/lib/sqlalchemy/orm/persistence.py
+++ b/lib/sqlalchemy/orm/persistence.py
@@ -25,7 +25,8 @@ from . import loading
 
 
 def _bulk_insert(
-        mapper, mappings, session_transaction, isstates, return_defaults):
+        mapper, mappings, session_transaction, isstates, return_defaults,
+        render_nulls):
     base_mapper = mapper.base_mapper
 
     cached_connections = _cached_connection_dict(base_mapper)
@@ -58,7 +59,8 @@ def _bulk_insert(
             has_all_defaults in _collect_insert_commands(table, (
                 (None, mapping, mapper, connection)
                 for mapping in mappings),
-                bulk=True, return_defaults=return_defaults
+                bulk=True, return_defaults=return_defaults,
+                render_nulls=render_nulls
             )
         )
         _emit_insert_statements(base_mapper, None,
@@ -365,7 +367,7 @@ def _organize_states_for_delete(base_mapper, states, uowtransaction):
 
 def _collect_insert_commands(
         table, states_to_insert,
-        bulk=False, return_defaults=False):
+        bulk=False, return_defaults=False, render_nulls=False):
     """Identify sets of values to use in INSERT statements for a
     list of states.
 
@@ -384,7 +386,7 @@ def _collect_insert_commands(
         for propkey in set(propkey_to_col).intersection(state_dict):
             value = state_dict[propkey]
             col = propkey_to_col[propkey]
-            if value is None and propkey not in eval_none:
+            if value is None and propkey not in eval_none and not render_nulls:
                 continue
             elif not bulk and isinstance(value, sql.ClauseElement):
                 value_params[col.key] = value
diff --git a/lib/sqlalchemy/orm/session.py b/lib/sqlalchemy/orm/session.py
index 1cf1bdb24..a7440bf40 100644
--- a/lib/sqlalchemy/orm/session.py
+++ b/lib/sqlalchemy/orm/session.py
@@ -2261,9 +2261,10 @@ class Session(_SessionClassMethods):
         ):
             self._bulk_save_mappings(
                 mapper, states, isupdate, True,
-                return_defaults, update_changed_only)
+                return_defaults, update_changed_only, False)
 
-    def bulk_insert_mappings(self, mapper, mappings, return_defaults=False):
+    def bulk_insert_mappings(
+            self, mapper, mappings, return_defaults=False, render_nulls=False):
         """Perform a bulk insert of the given list of mapping dictionaries.
 
         The bulk insert feature allows plain Python dictionaries to be used as
@@ -2316,6 +2317,29 @@ class Session(_SessionClassMethods):
          reason this flag should be set as the returned default information
          is not used.
 
+        :param render_nulls: When True, a value of ``None`` will result
+         in a NULL value being included in the INSERT statement, rather
+         than the column being omitted from the INSERT.   This allows all
+         the rows being INSERTed to have the identical set of columns which
+         allows the full set of rows to be batched to the DBAPI.  Normally,
+         each column-set that contains a different combination of NULL values
+         than the previous row must omit a different series of columns from
+         the rendered INSERT statement, which means it must be emitted as a
+         separate statement.   By passing this flag, the full set of rows
+         are guaranteed to be batchable into one batch; the cost however is
+         that server-side defaults which are invoked by an omitted column will
+         be skipped, so care must be taken to ensure that these are not
+         necessary.
+
+         .. warning::
+
+            When this flag is set, **server side default SQL values will
+            not be invoked** for those columns that are inserted as NULL;
+            the NULL value will be sent explicitly.   Care must be taken
+            to ensure that no server-side default functions need to be
+            invoked for the operation as a whole.
+
+         .. versionadded:: 1.1
 
         .. seealso::
 
@@ -2327,7 +2351,8 @@ class Session(_SessionClassMethods):
 
         """
         self._bulk_save_mappings(
-            mapper, mappings, False, False, return_defaults, False)
+            mapper, mappings, False, False,
+            return_defaults, False, render_nulls)
 
     def bulk_update_mappings(self, mapper, mappings):
         """Perform a bulk update of the given list of mapping dictionaries.
@@ -2376,11 +2401,12 @@ class Session(_SessionClassMethods):
             :meth:`.Session.bulk_save_objects`
 
         """
-        self._bulk_save_mappings(mapper, mappings, True, False, False, False)
+        self._bulk_save_mappings(
+            mapper, mappings, True, False, False, False, False)
 
     def _bulk_save_mappings(
             self, mapper, mappings, isupdate, isstates,
-            return_defaults, update_changed_only):
+            return_defaults, update_changed_only, render_nulls):
         mapper = _class_to_mapper(mapper)
         self._flushing = True
 
@@ -2393,7 +2419,8 @@ class Session(_SessionClassMethods):
                     isstates, update_changed_only)
             else:
                 persistence._bulk_insert(
-                    mapper, mappings, transaction, isstates, return_defaults)
+                    mapper, mappings, transaction,
+                    isstates, return_defaults, render_nulls)
             transaction.commit()
 
         except:
diff --git a/test/orm/test_bulk.py b/test/orm/test_bulk.py
index 878560cf6..0a51a5ad3 100644
--- a/test/orm/test_bulk.py
+++ b/test/orm/test_bulk.py
@@ -17,11 +17,12 @@ class BulkInsertUpdateTest(BulkTest, _fixtures.FixtureTest):
 
     @classmethod
     def setup_mappers(cls):
-        User, Address = cls.classes("User", "Address")
-        u, a = cls.tables("users", "addresses")
+        User, Address, Order = cls.classes("User", "Address", "Order")
+        u, a, o = cls.tables("users", "addresses", "orders")
 
         mapper(User, u)
         mapper(Address, a)
+        mapper(Order, o)
 
     def test_bulk_save_return_defaults(self):
         User, = self.classes("User",)
@@ -155,6 +156,28 @@ class BulkInsertUpdateTest(BulkTest, _fixtures.FixtureTest):
             )
         )
 
+    def test_bulk_insert_render_nulls(self):
+        Order, = self.classes("Order",)
+
+        s = Session()
+        with self.sql_execution_asserter() as asserter:
+            s.bulk_insert_mappings(
+                Order,
+                [{'id': 1, 'description': 'u1new'},
+                 {'id': 2, 'description': None},
+                 {'id': 3, 'description': 'u3new'}],
+                render_nulls=True
+            )
+
+        asserter.assert_(
+            CompiledSQL(
+                "INSERT INTO orders (id, description) VALUES (:id, :description)",
+                [{'id': 1, 'description': 'u1new'},
+                 {'id': 2, 'description': None},
+                 {'id': 3, 'description': 'u3new'}]
+            )
+        )
+
 
 class BulkUDPostfetchTest(BulkTest, fixtures.MappedTest):
     @classmethod