Implement bulk database operation support for Oracle and SQL Server

18 files changed, 1223 insertions, 379 deletions

diff --git a/odb/mssql/binding.hxx b/odb/mssql/binding.hxx
index 76aa2b2..1ff6f50 100644
--- a/odb/mssql/binding.hxx
+++ b/odb/mssql/binding.hxx
@@ -24,10 +24,23 @@ namespace odb
       typedef mssql::bind bind_type;
       typedef mssql::change_callback change_callback_type;
 
-      binding (): bind (0), count (0), version (0), change_callback (0) {}
+      binding ()
+        : bind (0), count (0), version (0),
+          batch (0), skip (0), status (0),
+          change_callback (0) {}
 
       binding (bind_type* b, std::size_t n)
-        : bind (b), count (n), version (0), change_callback (0)
+        : bind (b), count (n), version (0),
+          batch (1), skip (0), status (0),
+          change_callback (0)
+      {
+      }
+
+      binding (bind_type* b, std::size_t n,
+               std::size_t bt, std::size_t s, SQLUSMALLINT* st)
+        : bind (b), count (n), version (0),
+          batch (bt), skip (s), status (st),
+          change_callback (0)
       {
       }
 
@@ -35,6 +48,10 @@ namespace odb
       std::size_t count;
       std::size_t version;
 
+      std::size_t batch;
+      std::size_t skip;
+      SQLUSMALLINT* status; // Batch status array.
+
       change_callback_type* change_callback;
 
     private:
diff --git a/odb/mssql/container-statements.hxx b/odb/mssql/container-statements.hxx
index 5a7e52c..adc09ac 100644
--- a/odb/mssql/container-statements.hxx
+++ b/odb/mssql/container-statements.hxx
@@ -124,6 +124,7 @@ namespace odb
               insert_image_binding_,
               false,
               false,
+              0,
               false));
 
         return *insert_;
@@ -287,7 +288,7 @@ namespace odb
               update_text_,
               this->versioned_, // Process if versioned.
               update_image_binding_,
-              false,
+              0,
               false));
 
         return *update_;
diff --git a/odb/mssql/database.hxx b/odb/mssql/database.hxx
index 4ac85d2..fc13974 100644
--- a/odb/mssql/database.hxx
+++ b/odb/mssql/database.hxx
@@ -153,6 +153,10 @@ namespace odb
 
       template <typename T>
       typename object_traits<T>::id_type
+      persist (const T& object);
+
+      template <typename T>
+      typename object_traits<T>::id_type
       persist (T* obj_ptr);
 
       template <typename T, template <typename> class P>
@@ -175,6 +179,13 @@ namespace odb
       typename object_traits<T>::id_type
       persist (const typename object_traits<T>::pointer_type& obj_ptr);
 
+      // Bulk persist. Can be a range of references or pointers (including
+      // smart pointers) to objects.
+      //
+      template <typename I>
+      void
+      persist (I begin, I end, bool continue_failed = true);
+
       // Load an object. Throw object_not_persistent if not found.
       //
       template <typename T>
@@ -261,6 +272,13 @@ namespace odb
       void
       update (const typename object_traits<T>::pointer_type& obj_ptr);
 
+      // Bulk update. Can be a range of references or pointers (including
+      // smart pointers) to objects.
+      //
+      template <typename I>
+      void
+      update (I begin, I end, bool continue_failed = true);
+
       // Update a section of an object. Throws the section_not_loaded
       // exception if the section is not loaded. Note also that this
       // function does not clear the changed flag if it is set.
@@ -304,6 +322,19 @@ namespace odb
       void
       erase (const typename object_traits<T>::pointer_type& obj_ptr);
 
+      // Bulk erase.
+      //
+      template <typename T, typename I>
+      void
+      erase (I id_begin, I id_end, bool continue_failed = true);
+
+      // Can be a range of references or pointers (including smart pointers)
+      // to objects.
+      //
+      template <typename I>
+      void
+      erase (I obj_begin, I obj_end, bool continue_failed = true);
+
       // Erase multiple objects matching a query predicate.
       //
       template <typename T>
diff --git a/odb/mssql/database.ixx b/odb/mssql/database.ixx
index 93919f6..fc43934 100644
--- a/odb/mssql/database.ixx
+++ b/odb/mssql/database.ixx
@@ -27,6 +27,13 @@ namespace odb
 
     template <typename T>
     inline typename object_traits<T>::id_type database::
+    persist (const T& obj)
+    {
+      return persist_<const T, id_mssql> (obj);
+    }
+
+    template <typename T>
+    inline typename object_traits<T>::id_type database::
     persist (T* p)
     {
       typedef typename object_traits<T>::pointer_type object_pointer;
@@ -93,6 +100,13 @@ namespace odb
       return persist_<T, id_mssql> (pobj);
     }
 
+    template <typename I>
+    inline void database::
+    persist (I b, I e, bool cont)
+    {
+      persist_<I, id_mssql> (b, e, cont);
+    }
+
     template <typename T>
     inline typename object_traits<T>::pointer_type database::
     load (const typename object_traits<T>::id_type& id)
@@ -254,6 +268,13 @@ namespace odb
       update_<T, id_mssql> (pobj);
     }
 
+    template <typename I>
+    inline void database::
+    update (I b, I e, bool cont)
+    {
+      update_<I, id_mssql> (b, e, cont);
+    }
+
     template <typename T>
     inline void database::
     update (const T& obj, const section& s)
@@ -343,6 +364,20 @@ namespace odb
       erase_<T, id_mssql> (pobj);
     }
 
+    template <typename T, typename I>
+    inline void database::
+    erase (I idb, I ide, bool cont)
+    {
+      erase_id_<I, T, id_mssql> (idb, ide, cont);
+    }
+
+    template <typename I>
+    inline void database::
+    erase (I ob, I oe, bool cont)
+    {
+      erase_object_<I, id_mssql> (ob, oe, cont);
+    }
+
     template <typename T>
     inline unsigned long long database::
     erase_query ()
diff --git a/odb/mssql/error.cxx b/odb/mssql/error.cxx
index f6be6ea..0c41d32 100644
--- a/odb/mssql/error.cxx
+++ b/odb/mssql/error.cxx
@@ -20,7 +20,9 @@ namespace odb
                      SQLHANDLE h,
                      SQLSMALLINT htype,
                      connection* conn,
-                     bool end_tran)
+                     bool end_tran,
+                     size_t pos,
+                     multiple_exceptions* mex)
     {
       // First see if we have one of the errors indicated via the
       // return error code.
@@ -166,14 +168,41 @@ namespace odb
 
       for (SQLSMALLINT i (1);; ++i)
       {
-         r = SQLGetDiagRecA (htype,
-                             h,
-                             i,
-                             (SQLCHAR*) sqlstate,
-                             &native_code,
-                             (SQLCHAR*) msg,
-                             sizeof (msg),
-                             &msg_size);
+        // If this is for a batch, filter out based on row association.
+        // Here we only ignore records that have the associated row
+        // number and this number doesn't match ours. In particular,
+        // this means that all the un-associated records which will be
+        // duplicated for all the failed rows, which seems like the
+        // correct thing to do.
+        //
+        if (mex != 0)
+        {
+          SQLLEN n;
+          r = SQLGetDiagField (htype,
+                               h,
+                               i,
+                               SQL_DIAG_ROW_NUMBER,
+                               &n,
+                               0,
+                               0);
+
+          if (r == SQL_NO_DATA)
+              break;
+          else if (SQL_SUCCEEDED (r) &&
+                   n != SQL_NO_ROW_NUMBER &&
+                   n != SQL_ROW_NUMBER_UNKNOWN &&
+                   n != static_cast<SQLLEN> (pos + 1)) // 1-based
+            continue;
+        }
+
+        r = SQLGetDiagRecA (htype,
+                            h,
+                            i,
+                            (SQLCHAR*) sqlstate,
+                            &native_code,
+                            (SQLCHAR*) msg,
+                            sizeof (msg),
+                            &msg_size);
 
         if (r == SQL_NO_DATA)
           break;
@@ -203,27 +232,36 @@ namespace odb
       if (e.size () == 0)
         e.append (0, "?????", "no diagnostic record (using wrong handle?)");
 
-      throw e;
+      if (mex == 0)
+        throw e;
+      else
+        // It could be that some of these errors are fatal. I guess we
+        // will just have to learn from experience which ones are. The
+        // client code can always treat specific error codes as fatal.
+        //
+        mex->insert (pos, e);
     }
 
     void
     translate_error (SQLRETURN r, connection& c, bool end_tran)
     {
-      translate_error (r, c.handle (), SQL_HANDLE_DBC, &c, end_tran);
+      translate_error (r, c.handle (), SQL_HANDLE_DBC, &c, end_tran, 0, 0);
     }
 
     void
     translate_error (SQLRETURN r,
                      connection& c,
-                     const auto_handle<SQL_HANDLE_STMT>& h)
+                     const auto_handle<SQL_HANDLE_STMT>& h,
+                     size_t pos,
+                     multiple_exceptions* mex)
     {
-      translate_error (r, h, SQL_HANDLE_STMT, &c, false);
+      translate_error (r, h, SQL_HANDLE_STMT, &c, false, pos, mex);
     }
 
     void
     translate_error (SQLRETURN r, SQLHANDLE h, SQLSMALLINT htype)
     {
-      translate_error (r, h, htype, 0, false);
+      translate_error (r, h, htype, 0, false, 0, 0);
     }
   }
 }
diff --git a/odb/mssql/error.hxx b/odb/mssql/error.hxx
index 69ff6d4..59ba3bf 100644
--- a/odb/mssql/error.hxx
+++ b/odb/mssql/error.hxx
@@ -6,10 +6,11 @@
 #define ODB_MSSQL_ERROR_HXX
 
 #include <odb/pre.hxx>
+#include <cstddef> // std::size_t
 
 #include <odb/mssql/mssql-fwd.hxx>
 #include <odb/mssql/version.hxx>
-#include <odb/mssql/forward.hxx>     // connection
+#include <odb/mssql/forward.hxx>     // connection, multiple_exceptions
 #include <odb/mssql/auto-handle.hxx>
 
 #include <odb/mssql/details/export.hxx>
@@ -18,13 +19,17 @@ namespace odb
 {
   namespace mssql
   {
+    // Translate ODBC error given a handle and throw (or return, in case
+    // multiple_exceptions is not NULL) an appropriate exception.
+    //
     LIBODB_MSSQL_EXPORT void
     translate_error (SQLRETURN, connection&, bool end_tran = false);
 
     LIBODB_MSSQL_EXPORT void
     translate_error (SQLRETURN,
                      connection&,
-                     const auto_handle<SQL_HANDLE_STMT>&);
+                     const auto_handle<SQL_HANDLE_STMT>&,
+                     std::size_t pos = 0, multiple_exceptions* = 0);
 
     LIBODB_MSSQL_EXPORT void
     translate_error (SQLRETURN, SQLHANDLE, SQLSMALLINT htype);
diff --git a/odb/mssql/exceptions.cxx b/odb/mssql/exceptions.cxx
index 33aaba4..44cdf2e 100644
--- a/odb/mssql/exceptions.cxx
+++ b/odb/mssql/exceptions.cxx
@@ -57,6 +57,12 @@ namespace odb
       return what_.c_str ();
     }
 
+    database_exception* database_exception::
+    clone () const
+    {
+      return new database_exception (*this);
+    }
+
     //
     // cli_exception
     //
@@ -78,6 +84,12 @@ namespace odb
       return what_.c_str ();
     }
 
+    cli_exception* cli_exception::
+    clone () const
+    {
+      return new cli_exception (*this);
+    }
+
     //
     // long_data_reload
     //
@@ -88,5 +100,11 @@ namespace odb
       return "attempt to re-load object or view with long data "
         "from query result";
     }
+
+    long_data_reload* long_data_reload::
+    clone () const
+    {
+      return new long_data_reload (*this);
+    }
   }
 }
diff --git a/odb/mssql/exceptions.hxx b/odb/mssql/exceptions.hxx
index c503070..7d44a7b 100644
--- a/odb/mssql/exceptions.hxx
+++ b/odb/mssql/exceptions.hxx
@@ -79,6 +79,9 @@ namespace odb
       virtual const char*
       what () const throw ();
 
+      virtual database_exception*
+      clone () const;
+
     public:
       ~database_exception () throw ();
 
@@ -105,6 +108,9 @@ namespace odb
       virtual const char*
       what () const throw ();
 
+      virtual cli_exception*
+      clone () const;
+
     private:
       std::string what_;
     };
@@ -113,6 +119,9 @@ namespace odb
     {
       virtual const char*
       what () const throw ();
+
+      virtual long_data_reload*
+      clone () const;
     };
 
     namespace core
diff --git a/odb/mssql/no-id-object-statements.hxx b/odb/mssql/no-id-object-statements.hxx
index 981bc9f..4ead693 100644
--- a/odb/mssql/no-id-object-statements.hxx
+++ b/odb/mssql/no-id-object-statements.hxx
@@ -49,9 +49,9 @@ namespace odb
       // Object image.
       //
       image_type&
-      image ()
+      image (std::size_t i = 0)
       {
-        return image_;
+        return image_[i];
       }
 
       // Insert binding.
@@ -89,7 +89,8 @@ namespace odb
               object_traits::versioned, // Process if versioned.
               insert_image_binding_,
               false,
-              object_traits::rowversion,
+              false,
+              0,
               false));
 
         return *persist_;
@@ -110,7 +111,8 @@ namespace odb
       no_id_object_statements& operator= (const no_id_object_statements&);
 
     private:
-      image_type image_;
+      image_type image_[object_traits::batch];
+      SQLUSMALLINT status_[object_traits::batch];
 
       // Select binding.
       //
diff --git a/odb/mssql/no-id-object-statements.txx b/odb/mssql/no-id-object-statements.txx
index 0d92686..65dc8e9 100644
--- a/odb/mssql/no-id-object-statements.txx
+++ b/odb/mssql/no-id-object-statements.txx
@@ -19,13 +19,17 @@ namespace odb
     no_id_object_statements (connection_type& conn)
         : statements_base (conn),
           select_image_binding_ (select_image_bind_, select_column_count),
-          insert_image_binding_ (insert_image_bind_, insert_column_count)
+          insert_image_binding_ (insert_image_bind_,
+                                 insert_column_count,
+                                 object_traits::batch,
+                                 sizeof (image_type),
+                                 status_)
     {
-      image_.version = 0;
+      image_[0].version = 0;
       select_image_version_ = 0;
       insert_image_version_ = 0;
 
-      select_image_binding_.change_callback = image_.change_callback ();
+      select_image_binding_.change_callback = image_[0].change_callback ();
 
       std::memset (insert_image_bind_, 0, sizeof (insert_image_bind_));
       std::memset (select_image_bind_, 0, sizeof (select_image_bind_));
diff --git a/odb/mssql/polymorphic-object-statements.hxx b/odb/mssql/polymorphic-object-statements.hxx
index 381c607..49b5438 100644
--- a/odb/mssql/polymorphic-object-statements.hxx
+++ b/odb/mssql/polymorphic-object-statements.hxx
@@ -310,7 +310,7 @@ namespace odb
               insert_image_binding_,
               false,
               false,
-              false));
+              0));
 
         return *persist_;
       }
@@ -345,7 +345,7 @@ namespace odb
               object_traits::update_statement,
               object_traits::versioned, // Process if versioned.
               update_image_binding_,
-              false,
+              0,
               false));
 
         return *update_;
@@ -373,6 +373,7 @@ namespace odb
         return extra_statement_cache_.get (
           conn_,
           image_,
+          id_image (),
           id_image_binding (),
           &id_image_binding ()); // Note, not id+version.
       }
@@ -411,8 +412,9 @@ namespace odb
       root_statements_type& root_statements_;
       base_statements_type& base_statements_;
 
-      extra_statement_cache_ptr<extra_statement_cache_type, image_type>
-      extra_statement_cache_;
+      extra_statement_cache_ptr<extra_statement_cache_type,
+                                image_type,
+                                id_image_type> extra_statement_cache_;
 
       image_type image_;
 
diff --git a/odb/mssql/section-statements.hxx b/odb/mssql/section-statements.hxx
index 43919d5..8bb1650 100644
--- a/odb/mssql/section-statements.hxx
+++ b/odb/mssql/section-statements.hxx
@@ -36,6 +36,7 @@ namespace odb
       typedef ST traits;
 
       typedef typename traits::image_type image_type;
+      typedef typename traits::id_image_type id_image_type;
 
       typedef mssql::select_statement select_statement_type;
       typedef mssql::update_statement update_statement_type;
@@ -43,7 +44,7 @@ namespace odb
       typedef mssql::connection connection_type;
 
       section_statements (connection_type&,
-                          image_type&,
+                          image_type&, id_image_type&,
                           binding& id, binding& idv);
 
       connection_type&
@@ -61,6 +62,9 @@ namespace odb
       image_type&
       image () {return image_;}
 
+      id_image_type&
+      id_image () {return id_image_;}
+
       const binding&
       id_binding () {return id_binding_;}
 
@@ -128,7 +132,7 @@ namespace odb
               traits::update_statement,
               traits::versioned, // Process if versioned.
               update_image_binding_,
-              traits::rowversion,
+              (traits::rowversion ? &idv_binding_ : 0),
               false));
 
         return *update_;
@@ -155,6 +159,7 @@ namespace odb
       // These come from object_statements.
       //
       image_type& image_;
+      id_image_type& id_image_;
       binding& id_binding_;
       binding& idv_binding_;
 
diff --git a/odb/mssql/section-statements.txx b/odb/mssql/section-statements.txx
index 358c216..319b0b4 100644
--- a/odb/mssql/section-statements.txx
+++ b/odb/mssql/section-statements.txx
@@ -11,11 +11,12 @@ namespace odb
     template <typename T, typename ST>
     section_statements<T, ST>::
     section_statements (connection_type& conn,
-                        image_type& im,
+                        image_type& im, id_image_type& idim,
                         binding& id, binding& idv)
         : conn_ (conn),
           svm_ (0),
           image_ (im),
+          id_image_ (idim),
           id_binding_ (id),
           idv_binding_ (idv),
           select_image_binding_ (select_image_bind_,
diff --git a/odb/mssql/simple-object-statements.hxx b/odb/mssql/simple-object-statements.hxx
index ab43439..77b1c34 100644
--- a/odb/mssql/simple-object-statements.hxx
+++ b/odb/mssql/simple-object-statements.hxx
@@ -39,49 +39,56 @@ namespace odb
     // deleter function which will be initialized during allocation
     // (at that point we know that the cache class is defined).
     //
-    template <typename T, typename I>
+    template <typename T, typename I, typename ID>
     struct extra_statement_cache_ptr
     {
       typedef I image_type;
+      typedef ID id_image_type;
       typedef mssql::connection connection_type;
 
       extra_statement_cache_ptr (): p_ (0) {}
       ~extra_statement_cache_ptr ()
       {
         if (p_ != 0)
-          (this->*deleter_) (0, 0, 0, 0);
+          (this->*deleter_) (0, 0, 0, 0, 0);
       }
 
       T&
-      get (connection_type& c, image_type& im, binding& id, binding* idv)
+      get (connection_type& c,
+           image_type& im, id_image_type& idim,
+           binding& id, binding* idv)
       {
         if (p_ == 0)
-          allocate (&c, &im, &id, (idv != 0 ? idv : &id));
+          allocate (&c, &im, &idim, &id, (idv != 0 ? idv : &id));
 
         return *p_;
       }
 
     private:
       void
-      allocate (connection_type*, image_type*, binding*, binding*);
+      allocate (connection_type*,
+                image_type*, id_image_type*,
+                binding*, binding*);
 
     private:
       T* p_;
       void (extra_statement_cache_ptr::*deleter_) (
-        connection_type*, image_type*, binding*, binding*);
+        connection_type*, image_type*, id_image_type*, binding*, binding*);
     };
 
-    template <typename T, typename I>
-    void extra_statement_cache_ptr<T, I>::
-    allocate (connection_type* c, image_type* im, binding* id, binding* idv)
+    template <typename T, typename I, typename ID>
+    void extra_statement_cache_ptr<T, I, ID>::
+    allocate (connection_type* c,
+              image_type* im, id_image_type* idim,
+              binding* id, binding* idv)
     {
       // To reduce object code size, this function acts as both allocator
       // and deleter.
       //
       if (p_ == 0)
       {
-        p_ = new T (*c, *im, *id, *idv);
-        deleter_ = &extra_statement_cache_ptr<T, I>::allocate;
+        p_ = new T (*c, *im, *idim, *id, *idv);
+        deleter_ = &extra_statement_cache_ptr<T, I, ID>::allocate;
       }
       else
         delete p_;
@@ -155,7 +162,7 @@ namespace odb
       typedef T object_type;
       typedef object_traits_impl<object_type, id_mssql> object_traits;
 
-      optimistic_data (bind*);
+      optimistic_data (bind*, std::size_t skip, SQLUSMALLINT* status);
 
       binding*
       id_image_binding () {return &id_image_binding_;}
@@ -170,7 +177,7 @@ namespace odb
     template <typename T>
     struct optimistic_data<T, false>
     {
-      optimistic_data (bind*) {}
+      optimistic_data (bind*, std::size_t, SQLUSMALLINT*) {}
 
       binding*
       id_image_binding () {return 0;}
@@ -276,9 +283,9 @@ namespace odb
       // Object image.
       //
       image_type&
-      image ()
+      image (std::size_t i = 0)
       {
-        return image_;
+        return images_[i].obj;
       }
 
       // Insert binding.
@@ -323,7 +330,7 @@ namespace odb
       // Object id image and binding.
       //
       id_image_type&
-      id_image () {return id_image_;}
+      id_image (std::size_t i = 0) {return images_[i].id;}
 
       std::size_t
       id_image_version () const {return id_image_version_;}
@@ -339,7 +346,7 @@ namespace odb
       // at the same time.
       //
       binding&
-      optimistic_id_image_binding () {return od_.id_image_binding_;}
+      optimistic_id_image_binding () {return *od_.id_image_binding ();}
 
       // Statements.
       //
@@ -355,6 +362,9 @@ namespace odb
               insert_image_binding_,
               object_traits::auto_id,
               object_traits::rowversion,
+              (object_traits::rowversion
+               ? &optimistic_id_image_binding ()
+               : (object_traits::auto_id ? &id_image_binding () : 0)),
               false));
 
         return *persist_;
@@ -385,9 +395,10 @@ namespace odb
             new (details::shared) update_statement_type (
               conn_,
               object_traits::update_statement,
-              object_traits::versioned, // Process if versioned.
+              true,                            // Unique (0 or 1).
+              object_traits::versioned,        // Process if versioned.
               update_image_binding_,
-              object_traits::rowversion,
+              object_traits::rowversion ? &optimistic_id_image_binding () : 0,
               false));
 
         return *update_;
@@ -401,6 +412,7 @@ namespace odb
             new (details::shared) delete_statement_type (
               conn_,
               object_traits::erase_statement,
+              true, // Unique (0 or 1 affected rows).
               id_image_binding_,
               false));
 
@@ -429,7 +441,9 @@ namespace odb
       extra_statement_cache ()
       {
         return extra_statement_cache_.get (
-          conn_, image_, id_image_binding_, od_.id_image_binding ());
+          conn_,
+          images_[0].obj, images_[0].id,
+          id_image_binding_, od_.id_image_binding ());
       }
 
     public:
@@ -476,10 +490,22 @@ namespace odb
       template <typename T1>
       friend class polymorphic_derived_object_statements;
 
-      extra_statement_cache_ptr<extra_statement_cache_type, image_type>
-      extra_statement_cache_;
+      extra_statement_cache_ptr<extra_statement_cache_type,
+                                image_type,
+                                id_image_type> extra_statement_cache_;
+
+      // The UPDATE statement uses both the object and id image. Keep
+      // them next to each other so that the same skip distance can
+      // be used in batch binding.
+      //
+      struct images
+      {
+        image_type obj;
+        id_image_type id;
+      };
 
-      image_type image_;
+      images images_[object_traits::batch];
+      SQLUSMALLINT status_[object_traits::batch];
 
       // Select binding.
       //
@@ -508,10 +534,9 @@ namespace odb
       bind update_image_bind_[update_column_count + id_column_count +
                               managed_optimistic_column_count];
 
-      // Id image binding (only used as a parameter). Uses the suffix in
-      // the update bind.
+      // Id image binding (only used as a parameter or in OUTPUT for
+      // auto id and version). Uses the suffix in the update bind.
       //
-      id_image_type id_image_;
       std::size_t id_image_version_;
       binding id_image_binding_;
 
diff --git a/odb/mssql/simple-object-statements.txx b/odb/mssql/simple-object-statements.txx
index 84ed5db..1cdffb2 100644
--- a/odb/mssql/simple-object-statements.txx
+++ b/odb/mssql/simple-object-statements.txx
@@ -20,11 +20,14 @@ namespace odb
 
     template <typename T>
     optimistic_data<T, true>::
-    optimistic_data (bind* b)
+    optimistic_data (bind* b, std::size_t skip, SQLUSMALLINT* status)
         : id_image_binding_ (
             b,
             object_traits::id_column_count +
-            object_traits::managed_optimistic_column_count)
+            object_traits::managed_optimistic_column_count,
+            object_traits::batch,
+            skip,
+            status)
     {
     }
 
@@ -43,24 +46,37 @@ namespace odb
     object_statements (connection_type& conn)
         : object_statements_base (conn),
           select_image_binding_ (select_image_bind_, select_column_count),
-          insert_image_binding_ (insert_image_bind_, insert_column_count),
+          insert_image_binding_ (insert_image_bind_,
+                                 insert_column_count,
+                                 object_traits::batch,
+                                 sizeof (images),
+                                 status_),
           update_image_binding_ (update_image_bind_,
                                  update_column_count + id_column_count +
-                                 managed_optimistic_column_count),
+                                 managed_optimistic_column_count,
+                                 object_traits::batch,
+                                 sizeof (images),
+                                 status_),
           id_image_binding_ (update_image_bind_ + update_column_count,
-                             id_column_count),
-          od_ (update_image_bind_ + update_column_count)
+                             id_column_count,
+                             object_traits::batch,
+                             sizeof (images),
+                             status_),
+          od_ (update_image_bind_ + update_column_count,
+               sizeof (images),
+               status_)
     {
-      image_.version = 0;
+      images_[0].obj.version = 0; // @@ TODO [0]
+      images_[0].id.version = 0;  // @@ TODO
+
       select_image_version_ = 0;
       insert_image_version_ = 0;
       update_image_version_ = 0;
       update_id_image_version_ = 0;
-
-      id_image_.version = 0;
       id_image_version_ = 0;
 
-      select_image_binding_.change_callback = image_.change_callback ();
+      select_image_binding_.change_callback =
+        images_[0].obj.change_callback ();
 
       std::memset (insert_image_bind_, 0, sizeof (insert_image_bind_));
       std::memset (update_image_bind_, 0, sizeof (update_image_bind_));
diff --git a/odb/mssql/statement.cxx b/odb/mssql/statement.cxx
index e87374b..1d4b204 100644
--- a/odb/mssql/statement.cxx
+++ b/odb/mssql/statement.cxx
@@ -2,7 +2,7 @@
 // copyright : Copyright (c) 2005-2013 Code Synthesis Tools CC
 // license   : ODB NCUEL; see accompanying LICENSE file
 
-#include <cstring> // std::strlen, std::strstr
+#include <cstring> // std::strlen, std::strstr, std::memset, std::memcpy
 #include <cassert>
 
 #include <odb/tracer.hxx>
@@ -27,7 +27,7 @@ namespace odb
       SQL_BIT,                // bind::bit
       SQL_TINYINT,            // bind::tinyint
       SQL_SMALLINT,           // bind::smallint
-      SQL_INTEGER,            // bind::integer
+      SQL_INTEGER,            // bind::int_
       SQL_BIGINT,             // bind::bigint
 
       SQL_DECIMAL,            // bind::decimal
@@ -62,7 +62,7 @@ namespace odb
       SQL_C_BIT,            // bind::bit
       SQL_C_UTINYINT,       // bind::tinyint
       SQL_C_SSHORT,         // bind::smallint
-      SQL_C_SLONG,          // bind::integer
+      SQL_C_SLONG,          // bind::int_
       SQL_C_SBIGINT,        // bind::bigint
 
       SQL_C_NUMERIC,        // bind::decimal
@@ -90,6 +90,41 @@ namespace odb
       SQL_C_BINARY          // bind::rowversion
     };
 
+    // Mapping of bind::buffer_type to fixed buffer capacity values.
+    //
+    static const SQLLEN capacity_lookup [bind::last] =
+    {
+      1,                       // bind::bit
+      1,                       // bind::tinyint
+      2,                       // bind::smallint
+      4,                       // bind::int_
+      8,                       // bind::bigint
+
+      sizeof (decimal),        // bind::decimal
+      4,                       // bind::smallmoney
+      8,                       // bind::money
+
+      4,                       // bind::float4
+      8,                       // bind::float8
+
+      0,                       // bind::string
+      0,                       // bind::long_string
+
+      0,                       // bind::nstring
+      0,                       // bind::long_nstring
+
+      0,                       // bind::binary
+      0,                       // bind::long_binary
+
+      sizeof (date),           // bind::date
+      sizeof (time),           // bind::time
+      sizeof (datetime),       // bind::datetime
+      sizeof (datetimeoffset), // bind::datetimeoffset
+
+      16,                      // bind::uniqueidentifier
+      8                        // bind::rowversion
+    };
+
     //
     // statement
     //
@@ -289,7 +324,7 @@ namespace odb
         case bind::long_string:
         case bind::long_binary:
           {
-            buf = (SQLPOINTER) b;
+            buf = (SQLPOINTER) b->buffer;
             col_size = b->capacity != 0
               ? (SQLULEN) b->capacity
               : SQL_SS_LENGTH_UNLIMITED;
@@ -297,7 +332,7 @@ namespace odb
           }
         case bind::long_nstring:
           {
-            buf = (SQLPOINTER) b;
+            buf = (SQLPOINTER) b->buffer;
             col_size = b->capacity != 0
               ? (SQLULEN) b->capacity / 2 // In characters, not bytes.
               : SQL_SS_LENGTH_UNLIMITED;
@@ -429,56 +464,10 @@ namespace odb
         if (b->buffer == 0) // Skip NULL entries.
           continue;
 
-        SQLLEN cap (0);
+        SQLLEN cap (capacity_lookup[b->type]);
 
         switch (b->type)
         {
-        case bind::bit:
-        case bind::tinyint:
-          {
-            cap = 1;
-            break;
-          }
-        case bind::smallint:
-          {
-            cap = 2;
-            break;
-          }
-        case bind::int_:
-          {
-            cap = 4;
-            break;
-          }
-        case bind::bigint:
-          {
-            cap = 8;
-            break;
-          }
-        case bind::decimal:
-          {
-            cap = (SQLLEN) sizeof (decimal);
-            break;
-          }
-        case bind::smallmoney:
-          {
-            cap = 4;
-            break;
-          }
-        case bind::money:
-          {
-            cap = 8;
-            break;
-          }
-        case bind::float4:
-          {
-            cap = 4;
-            break;
-          }
-        case bind::float8:
-          {
-            cap = 8;
-            break;
-          }
         case bind::string:
         case bind::nstring:
         case bind::binary:
@@ -495,41 +484,13 @@ namespace odb
             long_count++;
             continue;
           }
-        case bind::date:
-          {
-            cap = (SQLLEN) sizeof (date);
-            break;
-          }
-        case bind::time:
-          {
-            cap = (SQLLEN) sizeof (time);
-            break;
-          }
-        case bind::datetime:
-          {
-            cap = (SQLLEN) sizeof (datetime);
-            break;
-          }
-        case bind::datetimeoffset:
-          {
-            cap = (SQLLEN) sizeof (datetimeoffset);
-            break;
-          }
-        case bind::uniqueidentifier:
-          {
-            cap = 16;
-            break;
-          }
-        case bind::rowversion:
-          {
-            cap = 8;
-            break;
-          }
         case bind::last:
           {
             assert (false);
             break;
           }
+        default:
+          break;
         }
 
         r = SQLBindCol (stmt_,
@@ -566,10 +527,13 @@ namespace odb
         if (tmp_buf.capacity () == 0)
           tmp_buf.capacity (4096);
 
-        bind* b;
+        long_callback* pcb;
         for (;;)
         {
-          r = SQLParamData (stmt_, (SQLPOINTER*) &b);
+          // ODBC seems to already offset the returned pointer for us
+          // in case of a batch.
+          //
+          r = SQLParamData (stmt_, (SQLPOINTER*) &pcb);
 
           // If we get anything other than SQL_NEED_DATA, then this is
           // the return code of SQLExecute().
@@ -577,12 +541,11 @@ namespace odb
           if (r != SQL_NEED_DATA)
             break;
 
-          long_callback& cb (*static_cast<long_callback*> (b->buffer));
-
           // Store the pointer to the long_callback struct in buf on the
           // first call to the callback. This allows the callback to
           // redirect further calls to some other callback.
           //
+          long_callback cb (*pcb);
           const void* buf (&cb);
 
           size_t position (0);
@@ -670,7 +633,7 @@ namespace odb
           cbp = nb + (p - ob);
         }
 
-        long_callback& cb (*static_cast<long_callback*> (cbp));
+        long_callback cb (*static_cast<long_callback*> (cbp));
 
         // First determine if the value is NULL as well as try to
         // get the total data size.
@@ -755,6 +718,175 @@ namespace odb
     }
 
     //
+    // bulk_statement
+    //
+    bulk_statement::
+    ~bulk_statement () {}
+
+    void bulk_statement::
+    init (size_t skip)
+    {
+      // Setup row-wise batch operation. We set the actual number of
+      // parameter sets in the batch in execute().
+      //
+      SQLRETURN r;
+
+      r = SQLSetStmtAttr (stmt_,
+                          SQL_ATTR_PARAM_BIND_TYPE,
+                          (SQLPOINTER) skip,
+                          0);
+
+      if (!SQL_SUCCEEDED (r))
+        translate_error (r, conn_, stmt_);
+
+      r = SQLSetStmtAttr (stmt_,
+                          SQL_ATTR_PARAMS_PROCESSED_PTR,
+                          (SQLPOINTER) &processed_,
+                          0);
+
+      if (!SQL_SUCCEEDED (r))
+        translate_error (r, conn_, stmt_);
+
+      r = SQLSetStmtAttr (stmt_,
+                          SQL_ATTR_PARAM_STATUS_PTR,
+                          (SQLPOINTER) status_,
+                          0);
+
+      if (!SQL_SUCCEEDED (r))
+        translate_error (r, conn_, stmt_);
+    }
+
+    SQLRETURN bulk_statement::
+    execute (size_t n, multiple_exceptions* mex)
+    {
+      mex_ = mex;
+
+      if (status_ != 0)
+      {
+        SQLRETURN r (SQLSetStmtAttr (stmt_,
+                                     SQL_ATTR_PARAMSET_SIZE,
+                                     (SQLPOINTER) n,
+                                     0));
+
+        if (!SQL_SUCCEEDED (r))
+          translate_error (r, conn_, stmt_);
+
+        // Some SQL* functions would only update the status in case of
+        // an error.
+        //
+        memset (status_, 0, sizeof (status_[0]) * n);
+      }
+
+      processed_ = 0;
+      SQLRETURN r (statement::execute ());
+      bool ok (SQL_SUCCEEDED (r) || r == SQL_NO_DATA);
+
+      // If we have a batch of 1 parameter set, SQL Server ODBC driver
+      // returns the error via SQLExecute() rather than via the status
+      // array even if we set all the attributes necessary for row-wise
+      // binding. So what we are going to do here is convert this case
+      // to the batch way of reporting errors (not that we also check
+      // processed_ so that we only do this is the parameter set was
+      // actually attempted).
+      //
+      if (!ok && status_ != 0 && n == 1 && processed_ == 1)
+      {
+        status_[0] = SQL_PARAM_ERROR;
+        r = SQL_SUCCESS;
+        ok = true;
+      }
+
+      // If the statement failed as a whole, assume no parameter sets
+      // were attempted in case of a batch. Otherwise, the documentation
+      // says that the native client driver keeps processing remaining
+      // sets even in case of an error.
+      //
+      i_ = 0;
+      n_ = (ok ? n : (status_ == 0 ? 1 : 0));
+
+      if (mex_ != 0)
+      {
+        mex_->current (i_);
+        mex_->attempted (processed_);
+      }
+
+      if (!ok)
+      {
+        if (mex_ != 0)
+          mex_->fatal (true); // An incomplete batch is always fatal.
+
+        return r;
+      }
+
+      return r;
+    }
+
+    size_t bulk_statement::
+    extract_errors ()
+    {
+      size_t e (0);
+
+      for (size_t i (0); i != n_; ++i)
+      {
+        if (status_[i] != SQL_PARAM_SUCCESS &&
+            status_[i] != SQL_PARAM_SUCCESS_WITH_INFO)
+        {
+          translate_error (SQL_ERROR, conn_, stmt_, i, mex_);
+          e++;
+        }
+      }
+
+      return e;
+    }
+
+    unsigned long long bulk_statement::
+    affected (SQLRETURN r, size_t errors, bool unique)
+    {
+      unsigned long long rows (0);
+
+      // SQL_NO_DATA indicates that the statement hasn't affected any rows.
+      //
+      if (r != SQL_NO_DATA)
+      {
+        SQLLEN n;
+        r = SQLRowCount (stmt_, &n);
+
+        if (!SQL_SUCCEEDED (r))
+          translate_error (r, conn_, stmt_);
+
+        // If all the parameter sets failed, then the returned count is -1,
+        // which means "not available" according to the documentation.
+        //
+        rows = (n != -1 ? static_cast<unsigned long long> (n) : 0);
+      }
+
+      if (n_ > 1) // Batch.
+      {
+        if (rows != 0) // Some rows did get affected.
+        {
+          // Subtract the parameter sets that failed since they haven't
+          // affected any rows.
+          //
+          size_t p (n_ - errors);
+
+          if (p > 1) // True batch.
+          {
+            if (unique) // Each can affect 0 or 1 row.
+            {
+              rows = (p == static_cast<size_t> (rows)
+                      ? 1
+                      : result_unknown);
+            }
+            else
+              rows = result_unknown;
+          }
+        }
+      }
+
+      return rows;
+    }
+
+    //
     // select_statement
     //
     select_statement::
@@ -949,16 +1081,19 @@ namespace odb
                       bool process,
                       binding& param,
                       bool returning_id,
-                      bool returning_version)
-        : statement (conn,
-                     text, statement_insert,
-                     (process ? &param : 0), false),
+                      bool returning_version,
+                      binding* returning)
+        : bulk_statement (conn,
+                          text, statement_insert,
+                          (process ? &param : 0), false,
+                          param.batch, param.skip, param.status),
           returning_id_ (returning_id),
-          returning_version_ (returning_version)
+          returning_version_ (returning_version),
+          ret_ (returning)
     {
       bind_param (param.bind, param.count);
 
-      if (returning_id_ || returning_version_)
+      if (ret_ != 0)
         init_result ();
     }
 
@@ -969,20 +1104,31 @@ namespace odb
                       binding& param,
                       bool returning_id,
                       bool returning_version,
+                      binding* returning,
                       bool copy_text)
-        : statement (conn,
-                     text, statement_insert,
-                     (process ? &param : 0), false,
-                     copy_text),
+        : bulk_statement (conn,
+                          text, statement_insert,
+                          (process ? &param : 0), false,
+                          param.batch, param.skip, param.status,
+                          copy_text),
           returning_id_ (returning_id),
-          returning_version_ (returning_version)
+          returning_version_ (returning_version),
+          ret_ (returning)
     {
       bind_param (param.bind, param.count);
 
-      if (returning_id_ || returning_version_)
+      if (ret_ != 0)
         init_result ();
     }
 
+    template <typename T>
+    inline T*
+    offset (T* base, size_t count, size_t size)
+    {
+      return reinterpret_cast<T*> (
+        reinterpret_cast<char*> (base) + count * size);
+    }
+
     void insert_statement::
     init_result ()
     {
@@ -993,20 +1139,39 @@ namespace odb
       // for one of the inserted columns is supplied at execution
       // (long data).
       //
-      batch_ = strstr (text_, "OUTPUT INSERTED.") == 0 &&
-        strstr (text_, "output inserted.") == 0;
-
+      text_batch_ = (strstr (text_, "OUTPUT INSERTED.") == 0 &&
+                     strstr (text_, "output inserted.") == 0);
+
+      // It might seem logical to set up the array of results if this is a
+      // batch (i.e., the SQL_ATTR_ROW_BIND_TYPE, SQL_ATTR_ROW_ARRAY_SIZE).
+      // This won't work because what we are getting is multiple result
+      // sets (each containing a single row) and not multiple rows. As a
+      // result, the SQL Server ODBC driver will always store the data in
+      // the first element of our array. A bit counter-intuitive.
+      //
+      // At the same time it would be conceptually cleaner to have the
+      // returned data extracted into the batch array instead of always
+      // the first element. This is also how other database runtimes (e.g.,
+      // Oracle) behave. So what we are going to do here is emulate this
+      // by making the ODBC driver store the data into the last element
+      // of the batch array and then copying it into the right place
+      // after processing each result set (see fetch() below).
+      //
+      SQLRETURN r;
       SQLUSMALLINT col (1);
 
+      size_t last (ret_->batch - 1);
+
       if (returning_id_)
       {
-        SQLRETURN r (
-          SQLBindCol (stmt_,
-                      col++,
-                      SQL_C_SBIGINT,
-                      (SQLPOINTER) &id_,
-                      sizeof (id_),
-                      &id_size_ind_));
+        bind& b (ret_->bind[0]); // Auto id is the first element.
+
+        r = SQLBindCol (stmt_,
+                        col++,
+                        c_type_lookup[b.type],
+                        (SQLPOINTER) offset (b.buffer, last, ret_->skip),
+                        capacity_lookup[b.type],
+                        offset (b.size_ind, last, ret_->skip));
 
         if (!SQL_SUCCEEDED (r))
           translate_error (r, conn_, stmt_);
@@ -1014,84 +1179,173 @@ namespace odb
 
       if (returning_version_)
       {
-        SQLRETURN r (
-          SQLBindCol (stmt_,
-                      col++,
-                      SQL_C_BINARY,
-                      (SQLPOINTER) &version_,
-                      sizeof (version_),
-                      &version_size_ind_));
+        bind& b (ret_->bind[ret_->count - 1]); // Version is the last element.
+
+        r = SQLBindCol (stmt_,
+                        col++,
+                        c_type_lookup[b.type],
+                        (SQLPOINTER) offset (b.buffer, last, ret_->skip),
+                        capacity_lookup[b.type],
+                        offset (b.size_ind, last, ret_->skip));
 
         if (!SQL_SUCCEEDED (r))
           translate_error (r, conn_, stmt_);
       }
     }
 
-    bool insert_statement::
-    execute ()
+    size_t insert_statement::
+    execute (size_t n, multiple_exceptions* mex)
     {
-      SQLRETURN r (statement::execute ());
+      // The batch INSERT works in two different ways, depending on
+      // whether we have the OUTPUT clause. If there is no OUTPUT, then
+      // all the parameter sets are processed inside the SQLExecute()
+      // call. If, however, there is OUTPUT, then the sets are
+      // processed one at a time as we consume the results with
+      // the SQLMoreResults() call below. Thus we in effect have
+      // two counts: the "processed so far" as set by the API
+      // (SQL_ATTR_PARAMS_PROCESSED_PTR) and the "to be processed"
+      // (value in n_). Note that in the OUTPUT case if there is an
+      // error, the processed count seems to jump by 2 for some reason.
+      //
+      // The OUTPUT case can be handled in two different ways: we can
+      // "execute" (with SQLMoreResults()) each set as the user moves
+      // from one result to the next (result() call). The advantage of
+      // this approach is that the returned data ends up in the right
+      // place automatically. The potential drawback is that the total
+      // affected row count will only be available at the end. As a
+      // result, this approach probably won't work if we need to handle,
+      // say, UPDATE with OUTPUT (SQLRowCount() does not return an
+      // intermediate total, at least not for INSERT).
+      //
+      // The alternative implementation would call SQLMoreResults()
+      // inside execute() until all the parameter sets are executed.
+      // In this case we will have to copy the extracted data into
+      // the right place in the bindings (or update the binding before
+      // each call to SQLMoreResults()). It is also not clear whether
+      // the diagnostic records for the failed sets would accumulate.
+      // If not, those will have to be stashed into mex on each
+      // iteration.
+      //
+      SQLRETURN r (bulk_statement::execute (n, mex));
 
+      // Statement failed as a whole, assume no parameter sets were
+      // attempted in case of a batch.
+      //
       if (!SQL_SUCCEEDED (r))
       {
-        // Translate the integrity contraint violation (SQLSTATE 23000)
-        // to the flase return value. This code is similar to that found
-        // in translate_error().
-        //
-        char sqlstate[SQL_SQLSTATE_SIZE + 1];
-        SQLINTEGER native_code;
-        SQLSMALLINT msg_size;
+        fetch (r);
+        return n_;
+      }
 
-        bool cv (false);
+      if (status_ == 0) // Non-batch case.
+        fetch (SQL_SUCCESS);
+      else
+        fetch (status_[i_] == SQL_PARAM_SUCCESS ||
+               status_[i_] == SQL_PARAM_SUCCESS_WITH_INFO
+               ? SQL_SUCCESS : SQL_ERROR);
+
+      return n_;
+    }
 
-        for (SQLSMALLINT i (1);; ++i)
+    void insert_statement::
+    fetch (SQLRETURN r)
+    {
+      result_ = true;
+
+      if (!SQL_SUCCEEDED (r))
+      {
+        // An auto-assigned object id should never cause a duplicate primary
+        // key.
+        //
+        if (!returning_id_)
         {
-          SQLRETURN r (SQLGetDiagRecA (SQL_HANDLE_STMT,
-                                       stmt_,
-                                       i,
-                                       (SQLCHAR*) sqlstate,
-                                       &native_code,
-                                       0,
-                                       0,
-                                       &msg_size));
+          // Translate the integrity contraint violation (SQLSTATE 23000)
+          // to the false result value. This code is similar to that found
+          // in translate_error().
+          //
+          char sqlstate[SQL_SQLSTATE_SIZE + 1];
+          SQLINTEGER native_code;
+          SQLSMALLINT msg_size;
 
-          if (r == SQL_NO_DATA)
-            break;
-          else if (SQL_SUCCEEDED (r))
+          bool cv (false);
+
+          for (SQLSMALLINT i (1);; ++i)
           {
-            string s (sqlstate);
+            SQLRETURN r;
 
-            if (s == "23000")      // Integrity contraint violation.
-              cv = true;
-            else if (s != "01000") // General warning.
+            // Filter based on row association.
+            //
+            if (mex_ != 0)
+            {
+              SQLLEN n;
+              r = SQLGetDiagField (SQL_HANDLE_STMT,
+                                   stmt_,
+                                   i,
+                                   SQL_DIAG_ROW_NUMBER,
+                                   &n,
+                                   0,
+                                   0);
+
+              if (r == SQL_NO_DATA)
+                break;
+              else if (!SQL_SUCCEEDED (r))
+                continue;
+
+              if (n == SQL_NO_ROW_NUMBER ||
+                  n == SQL_ROW_NUMBER_UNKNOWN ||
+                  n != static_cast<SQLLEN> (i_ + 1)) // 1-based
+                continue;
+            }
+
+            r= SQLGetDiagRecA (SQL_HANDLE_STMT,
+                               stmt_,
+                               i,
+                               (SQLCHAR*) sqlstate,
+                               &native_code,
+                               0,
+                               0,
+                               &msg_size);
+
+            if (r == SQL_NO_DATA)
+              break;
+            else if (SQL_SUCCEEDED (r))
+            {
+              string s (sqlstate);
+
+              if (s == "23000")      // Integrity contraint violation.
+                cv = true;
+              else if (s != "01000") // General warning.
+              {
+                // Some other code.
+                //
+                cv = false;
+                break;
+              }
+            }
+            else // SQLGetDiagRec() failure.
             {
-              // Some other code.
-              //
               cv = false;
               break;
             }
           }
-          else
-          {
-            // SQLGetDiagRec() failure.
-            //
-            cv = false;
-            break;
-          }
+
+          if (cv)
+            result_ = false;
         }
 
-        if (cv)
-          return false;
-        else
-          translate_error (r, conn_, stmt_);
+        if (result_)
+        {
+          translate_error (r, conn_, stmt_, i_, mex_); // Can return.
+          result_ = false; // Prevent id/version extraction below or.
+        }
       }
 
       // Fetch the row containing the id/version if this statement is
       // returning.
       //
-      if (returning_id_ || returning_version_)
+      if (result_ && ret_ != 0)
       {
-        if (batch_)
+        if (text_batch_)
         {
           r = SQLMoreResults (stmt_);
 
@@ -1111,21 +1365,99 @@ namespace odb
         if (r != SQL_NO_DATA && !SQL_SUCCEEDED (r))
           translate_error (r, conn_, stmt_);
 
-        {
-          SQLRETURN r (SQLCloseCursor (stmt_)); // Don't overwrite r.
-
-          if (!SQL_SUCCEEDED (r))
-            translate_error (r, conn_, stmt_);
-        }
-
         if (r == SQL_NO_DATA)
           throw database_exception (
             0,
             "?????",
             "result set expected from a statement with the OUTPUT clause");
+
+        // See init_result() for details on what's going here.
+        //
+        size_t last (ret_->batch - 1);
+        if (i_ != last)
+        {
+          if (returning_id_)
+          {
+            bind& b (ret_->bind[0]); // Auto id is the first element.
+
+            memcpy (offset (b.buffer, i_, ret_->skip),
+                    offset (b.buffer, last, ret_->skip),
+                    capacity_lookup[b.type]);
+
+            memcpy (offset (b.size_ind, i_, ret_->skip),
+                    offset (b.size_ind, last, ret_->skip),
+                    sizeof (*b.size_ind));
+          }
+
+          if (returning_version_)
+          {
+            bind& b (ret_->bind[ret_->count - 1]); // Version is the last.
+
+            memcpy (offset (b.buffer, i_, ret_->skip),
+                    offset (b.buffer, last, ret_->skip),
+                    capacity_lookup[b.type]);
+
+            memcpy (offset (b.size_ind, i_, ret_->skip),
+                    offset (b.size_ind, last, ret_->skip),
+                    sizeof (*b.size_ind));
+          }
+        }
       }
+    }
 
-      return true;
+    bool insert_statement::
+    result (size_t i)
+    {
+      assert ((i_ == i || i_ + 1 == i) && i < n_);
+
+      SQLRETURN r;
+
+      // Get to the next result set if necessary.
+      //
+      if (i != i_)
+      {
+        mex_->current (++i_); // mex cannot be NULL since this is a batch.
+
+        // Only in case of the OUTPUT clause do we have multiple result sets.
+        //
+        if (ret_ != 0)
+        {
+          r = SQLMoreResults (stmt_);
+
+          // The actually processed count could have changed (see execute()).
+          //
+          mex_->attempted (processed_);
+
+          if (r == SQL_NO_DATA)
+          {
+            throw database_exception (
+              0,
+              "?????",
+              "multiple result sets expected from an array of parameters");
+          }
+        }
+
+        fetch (status_[i_] == SQL_PARAM_SUCCESS ||
+               status_[i_] == SQL_PARAM_SUCCESS_WITH_INFO
+               ? SQL_SUCCESS : SQL_ERROR);
+      }
+
+      // Close the cursor if we are done.
+      //
+      if (ret_ != 0 && i_ + 1 == n_)
+      {
+        // Use SQLFreeStmt(SQL_CLOSE) instead of SQLCloseCursor() to avoid
+        // an error if a cursor is not open. This seem to happen if the
+        // statement failure was translated to a parameter set failure in
+        // bulk_statement for batches of one.
+        //
+        r = SQLFreeStmt (stmt_, SQL_CLOSE);
+
+        if (!SQL_SUCCEEDED (r))
+          translate_error (r, conn_, stmt_);
+      }
+
+      return result_;
     }
 
     //
@@ -1142,19 +1474,33 @@ namespace odb
                       const string& text,
                       bool process,
                       binding& param,
-                      bool returning_version)
-        : statement (conn,
-                     text, statement_update,
-                     (process ? &param : 0), false),
-          returning_version_ (returning_version)
+                      binding* returning)
+        : bulk_statement (conn,
+                          text, statement_update,
+                          (process ? &param : 0), false,
+                          param.batch, param.skip, param.status),
+          unique_ (false),
+          returning_ (returning != 0)
     {
-      if (!empty ())
-      {
-        bind_param (param.bind, param.count);
+      assert (param.batch == 1); // Specify unique_hint explicitly.
+      init (param, returning);
+    }
 
-        if (returning_version_)
-          init_result ();
-      }
+    update_statement::
+    update_statement (connection_type& conn,
+                      const string& text,
+                      bool unique,
+                      bool process,
+                      binding& param,
+                      binding* returning)
+        : bulk_statement (conn,
+                          text, statement_update,
+                          (process ? &param : 0), false,
+                          param.batch, param.skip, param.status),
+          unique_ (unique),
+          returning_ (returning != 0)
+    {
+      init (param, returning);
     }
 
     update_statement::
@@ -1162,85 +1508,132 @@ namespace odb
                       const char* text,
                       bool process,
                       binding& param,
-                      bool returning_version,
+                      binding* returning,
                       bool copy_text)
-        : statement (conn,
-                     text, statement_update,
-                     (process ? &param : 0), false,
-                     copy_text),
-          returning_version_ (returning_version)
+        : bulk_statement (conn,
+                          text, statement_update,
+                          (process ? &param : 0), false,
+                          param.batch, param.skip, param.status,
+                          copy_text),
+          unique_ (false),
+          returning_ (returning != 0)
     {
-      if (!empty ())
-      {
-        bind_param (param.bind, param.count);
-
-        if (returning_version_)
-          init_result ();
-      }
+      assert (param.batch == 1); // Specify unique_hint explicitly.
+      init (param, returning);
     }
 
-    void update_statement::
-    init_result ()
+    update_statement::
+    update_statement (connection_type& conn,
+                      const char* text,
+                      bool unique,
+                      bool process,
+                      binding& param,
+                      binding* returning,
+                      bool copy_text)
+        : bulk_statement (conn,
+                          text, statement_update,
+                          (process ? &param : 0), false,
+                          param.batch, param.skip, param.status,
+                          copy_text),
+          unique_ (unique),
+          returning_ (returning != 0)
     {
-      SQLRETURN r (
-        SQLBindCol (stmt_,
-                    1,
-                    SQL_C_BINARY,
-                    (SQLPOINTER) &version_,
-                    sizeof (version_),
-                    &version_size_ind_));
-
-      if (!SQL_SUCCEEDED (r))
-        translate_error (r, conn_, stmt_);
+      init (param, returning);
     }
 
-    unsigned long long update_statement::
-    execute ()
+    void update_statement::
+    init (binding& param, binding* ret)
     {
-      SQLRETURN r (statement::execute ());
+      if (!empty ())
+      {
+        bind_param (param.bind, param.count);
 
-      // SQL_NO_DATA indicates that the statement hasn't affected any rows.
-      //
-      if (r == SQL_NO_DATA)
-        return 0;
+        if (ret != 0)
+        {
+          bind& b (ret->bind[ret->count - 1]); // Version is the last element.
 
-      if (!SQL_SUCCEEDED (r))
-        translate_error (r, conn_, stmt_);
+          SQLRETURN r (SQLBindCol (stmt_,
+                                   1,
+                                   c_type_lookup[b.type],
+                                   (SQLPOINTER) b.buffer,
+                                   capacity_lookup[b.type],
+                                   b.size_ind));
+
+          if (!SQL_SUCCEEDED (r))
+            translate_error (r, conn_, stmt_);
+        }
+      }
+    }
 
-      // Get the number of affected rows.
+    size_t update_statement::
+    execute (size_t n, multiple_exceptions* mex)
+    {
+      // In batch UPDATE without the OUTPUT clause (which is the
+      // only kind we currently support) all the parameter sets
+      // are processed inside SQLExecute() and the total count of
+      // affected rows is available after it returns.
       //
-      SQLLEN rows;
-      r = SQLRowCount (stmt_, &rows);
+      assert (!returning_ || status_ == 0);
 
-      if (!SQL_SUCCEEDED (r))
-        translate_error (r, conn_, stmt_);
+      SQLRETURN r (bulk_statement::execute (n, mex));
 
-      // Fetch the row containing the version if this statement is
-      // returning. We still need to close the cursor even if we
-      // haven't updated any rows.
+      // Statement failed as a whole, assume no parameter sets were
+      // attempted in case of a batch.
       //
-      if (returning_version_)
+      if (!(SQL_SUCCEEDED (r) || r == SQL_NO_DATA))
       {
-        r = SQLFetch (stmt_);
-
-        if (r != SQL_NO_DATA && !SQL_SUCCEEDED (r))
-          translate_error (r, conn_, stmt_);
+        translate_error (r, conn_, stmt_, 0, mex_);
+        return n_;
+      }
 
+      if (status_ == 0) // Non-batch case.
+      {
+        // Fetch the row containing the data if this statement is
+        // returning. We still need to close the cursor even if we
+        // haven't updated any rows.
+        //
+        if (returning_)
         {
-          SQLRETURN r (SQLCloseCursor (stmt_)); // Don't overwrite r.
+          r = SQLFetch (stmt_);
 
-          if (!SQL_SUCCEEDED (r))
+          if (r != SQL_NO_DATA && !SQL_SUCCEEDED (r))
             translate_error (r, conn_, stmt_);
+
+          // We have to get the result after fetching the OUTPUT data
+          // but before closing the cursor.
+          //
+          result_ = affected (SQL_SUCCESS, 0, unique_);
+
+          {
+            SQLRETURN r (SQLCloseCursor (stmt_)); // Don't overwrite r.
+
+            if (!SQL_SUCCEEDED (r))
+              translate_error (r, conn_, stmt_);
+          }
+
+          if (result_ != 0 && r == SQL_NO_DATA)
+            throw database_exception (
+              0,
+              "?????",
+              "result set expected from a statement with the OUTPUT clause");
         }
+        else
+          result_ = affected (r, 0, unique_);
+      }
+      else
+      {
+        // Extract error information for failed parameter sets. If we do
+        // this after calling SQLRowCount(), all the diagnostics records
+        // that we need will be gone.
+        //
+        size_t errors (extract_errors ());
 
-        if (rows != 0 && r == SQL_NO_DATA)
-          throw database_exception (
-            0,
-            "?????",
-            "result set expected from a statement with the OUTPUT clause");
+        // Figure out the affected row count.
+        //
+        result_ = affected (r, errors, unique_);
       }
 
-      return static_cast<unsigned long long> (rows);
+      return n_;
     }
 
     //
@@ -1256,9 +1649,26 @@ namespace odb
     delete_statement (connection_type& conn,
                       const string& text,
                       binding& param)
-        : statement (conn,
-                     text, statement_delete,
-                     0, false)
+        : bulk_statement (conn,
+                          text, statement_delete,
+                          0, false,
+                          param.batch, param.skip, param.status),
+          unique_ (false)
+    {
+      assert (param.batch == 1); // Specify unique_hint explicitly.
+      bind_param (param.bind, param.count);
+    }
+
+    delete_statement::
+    delete_statement (connection_type& conn,
+                      const string& text,
+                      bool unique,
+                      binding& param)
+        : bulk_statement (conn,
+                          text, statement_delete,
+                          0, false,
+                          param.batch, param.skip, param.status),
+          unique_ (unique)
     {
       bind_param (param.bind, param.count);
     }
@@ -1268,36 +1678,64 @@ namespace odb
                       const char* text,
                       binding& param,
                       bool copy_text)
-        : statement (conn,
-                     text, statement_delete,
-                     0, false,
-                     copy_text)
+        : bulk_statement (conn,
+                          text, statement_delete,
+                          0, false,
+                          param.batch, param.skip, param.status,
+                          copy_text),
+          unique_ (false)
     {
+      assert (param.batch == 1); // Specify unique_hint explicitly.
       bind_param (param.bind, param.count);
     }
 
-    unsigned long long delete_statement::
-    execute ()
+    delete_statement::
+    delete_statement (connection_type& conn,
+                      const char* text,
+                      bool unique,
+                      binding& param,
+                      bool copy_text)
+        : bulk_statement (conn,
+                          text, statement_delete,
+                          0, false,
+                          param.batch, param.skip, param.status,
+                          copy_text),
+          unique_ (unique)
     {
-      SQLRETURN r (statement::execute ());
+      bind_param (param.bind, param.count);
+    }
 
-      // SQL_NO_DATA indicates that the statement hasn't affected any rows.
+    size_t delete_statement::
+    execute (size_t n, multiple_exceptions* mex)
+    {
+      // In batch DELETE without the OUTPUT clause (which is the
+      // only kind we currently support) all the parameter sets
+      // are processed inside SQLExecute() and the total count of
+      // affected rows is available after it returns.
       //
-      if (r == SQL_NO_DATA)
-        return 0;
 
-      if (!SQL_SUCCEEDED (r))
-        translate_error (r, conn_, stmt_);
+      SQLRETURN r (bulk_statement::execute (n, mex));
 
-      // Get the number of affected rows.
+      // Statement failed as a whole, assume no parameter sets were
+      // attempted in case of a batch.
       //
-      SQLLEN rows;
-      r = SQLRowCount (stmt_, &rows);
+      if (!(SQL_SUCCEEDED (r) || r == SQL_NO_DATA))
+      {
+        translate_error (r, conn_, stmt_, 0, mex_);
+        return n_;
+      }
 
-      if (!SQL_SUCCEEDED (r))
-        translate_error (r, conn_, stmt_);
+      // Extract error information for failed parameter sets. If we do
+      // this after calling SQLRowCount(), all the diagnostics records
+      // that we need will be gone.
+      //
+      size_t errors (status_ != 0 ? extract_errors () : 0);
+
+      // Figure out the affected row count.
+      //
+      result_ = affected (r, errors, unique_);
 
-      return static_cast<unsigned long long> (rows);
+      return n_;
     }
   }
 }
diff --git a/odb/mssql/statement.hxx b/odb/mssql/statement.hxx
index 36c87f1..4ed279d 100644
--- a/odb/mssql/statement.hxx
+++ b/odb/mssql/statement.hxx
@@ -11,6 +11,7 @@
 #include <cstddef>  // std::size_t
 
 #include <odb/statement.hxx>
+#include <odb/exceptions.hxx>
 
 #include <odb/mssql/version.hxx>
 #include <odb/mssql/forward.hxx>
@@ -126,6 +127,62 @@ namespace odb
       auto_handle<SQL_HANDLE_STMT> stmt_;
     };
 
+    class LIBODB_MSSQL_EXPORT bulk_statement: public statement
+    {
+    public:
+      virtual
+      ~bulk_statement () = 0;
+
+    protected:
+      bulk_statement (connection_type&,
+                      const std::string& text,
+                      statement_kind,
+                      const binding* process,
+                      bool optimize,
+                      std::size_t batch,
+                      std::size_t skip,
+                      SQLUSMALLINT* status);
+
+      bulk_statement (connection_type&,
+                      const char* text,
+                      statement_kind,
+                      const binding* process,
+                      bool optimize,
+                      std::size_t batch,
+                      std::size_t skip,
+                      SQLUSMALLINT* status,
+                      bool copy_text);
+
+      // Call SQLExecute() and set up the batch tracking variables (see
+      // below). Note that this function does not treat SQL_NO_DATA as
+      // an error since for DELETE and UPDATE statements this is a
+      // shortcut notation for zero rows affected.
+      //
+      SQLRETURN
+      execute (std::size_t n, multiple_exceptions*);
+
+      // Return the number of failed parameter sets.
+      //
+      std::size_t
+      extract_errors ();
+
+      static const unsigned long long result_unknown = ~0ULL;
+
+      unsigned long long
+      affected (SQLRETURN, std::size_t errors, bool unique);
+
+    private:
+      void
+      init (std::size_t skip);
+
+    protected:
+      SQLULEN processed_;    // Number of parameter sets processed so far.
+      SQLUSMALLINT* status_; // Parameter sets status array.
+      std::size_t n_;        // Actual batch size.
+      std::size_t i_;        // Position in result.
+      multiple_exceptions* mex_;
+    };
+
     class LIBODB_MSSQL_EXPORT select_statement: public statement
     {
     public:
@@ -237,7 +294,7 @@ namespace odb
       select_statement* s_;
     };
 
-    class LIBODB_MSSQL_EXPORT insert_statement: public statement
+    class LIBODB_MSSQL_EXPORT insert_statement: public bulk_statement
     {
     public:
       virtual
@@ -248,7 +305,8 @@ namespace odb
                         bool process_text,
                         binding& param,
                         bool returning_id,
-                        bool returning_version);
+                        bool returning_version,
+                        binding* returning);
 
       insert_statement (connection_type& conn,
                         const char* text,
@@ -256,23 +314,30 @@ namespace odb
                         binding& param,
                         bool returning_id,
                         bool returning_version,
+                        binding* returning,
                         bool copy_text = true);
 
-      // Return true if successful and false if the row is a duplicate.
+      // Return the number of parameter sets (out of n) that were attempted.
+      //
+      std::size_t
+      execute (std::size_t n, multiple_exceptions& mex)
+      {
+        return execute (n, &mex);
+      }
+
+      // Return true if successful and false if this row is a duplicate.
       // All other errors are reported by throwing exceptions.
       //
       bool
-      execute ();
+      result (std::size_t i);
 
-      unsigned long long
-      id ()
+      bool
+      execute ()
       {
-        return id_;
+        execute (1, 0);
+        return result (0);
       }
 
-      unsigned long long
-      version ();
-
     private:
       insert_statement (const insert_statement&);
       insert_statement& operator= (const insert_statement&);
@@ -281,42 +346,97 @@ namespace odb
       void
       init_result ();
 
+      std::size_t
+      execute (std::size_t, multiple_exceptions*);
+
+      void
+      fetch (SQLRETURN);
+
     private:
       bool returning_id_;
       bool returning_version_;
-      bool batch_;
+      binding* ret_;
+      bool text_batch_;
 
-      unsigned long long id_;
-      SQLLEN id_size_ind_;
-
-      unsigned char version_[8];
-      SQLLEN version_size_ind_;
+      bool result_;
     };
 
-    class LIBODB_MSSQL_EXPORT update_statement: public statement
+    class LIBODB_MSSQL_EXPORT update_statement: public bulk_statement
     {
     public:
       virtual
       ~update_statement ();
 
+      // SQL Server native client ODBC driver does not expose individual
+      // affected row counts for batch operations, even though it says it
+      // does (SQLGetInfo(SQL_PARAM_ARRAY_ROW_COUNTS) returns SQL_PARC_BATCH).
+      // Instead, it adds them all up and returns a single count. This is
+      // bad news for us.
+      //
+      // In case of updating by primary key (the affected row count is
+      // either 1 or 0), we can recognize the presumably successful case
+      // where the total affected row count is equal to the batch size
+      // (we can also recognize the "all unsuccessful" case where the
+      // total affected row count is 0). The unique_hint argument in the
+      // constructors below indicates whether this is a "0 or 1" UPDATE
+      // statement.
+      //
+      // In all other situations (provided this is a batch), the result()
+      // function below returns the special result_unknown value.
+      //
       update_statement (connection_type& conn,
                         const std::string& text,
                         bool process,
                         binding& param,
-                        bool returning_version);
+                        binding* returning);
+
+      update_statement (connection_type& conn,
+                        const std::string& text,
+                        bool unique_hint,
+                        bool process,
+                        binding& param,
+                        binding* returning);
 
       update_statement (connection_type& conn,
                         const char* text,
                         bool process,
                         binding& param,
-                        bool returning_version,
+                        binding* returning,
                         bool copy_text = true);
 
+      update_statement (connection_type& conn,
+                        const char* text,
+                        bool unique_hint,
+                        bool process,
+                        binding& param,
+                        binding* returning,
+                        bool copy_text = true);
+
+      // Return the number of parameter sets (out of n) that were attempted.
+      //
+      std::size_t
+      execute (std::size_t n, multiple_exceptions& mex)
+      {
+        return execute (n, &mex);
+      }
+
+      // Return the number of rows affected (deleted) by the parameter
+      // set. If this is a batch (n > 1 in execute() call above) and it
+      // is impossible to determine the affected row count for each
+      // parameter set, then this function returns result_unknown. All
+      // other errors are reported by throwing exceptions.
+      //
+      using bulk_statement::result_unknown;
+
       unsigned long long
-      execute ();
+      result (std::size_t i);
 
       unsigned long long
-      version ();
+      execute ()
+      {
+        execute (1, 0);
+        return result (0);
+      }
 
     private:
       update_statement (const update_statement&);
@@ -324,23 +444,48 @@ namespace odb
 
     private:
       void
-      init_result ();
+      init (binding& param, binding* ret);
+
+      std::size_t
+      execute (std::size_t, multiple_exceptions*);
 
     private:
-      bool returning_version_;
+      bool unique_;
+      bool returning_;
 
-      unsigned char version_[8];
-      SQLLEN version_size_ind_;
+      unsigned long long result_;
     };
 
-    class LIBODB_MSSQL_EXPORT delete_statement: public statement
+    class LIBODB_MSSQL_EXPORT delete_statement: public bulk_statement
     {
     public:
       virtual
       ~delete_statement ();
 
+      // SQL Server native client ODBC driver does not expose individual
+      // affected row counts for batch operations, even though it says it
+      // does (SQLGetInfo(SQL_PARAM_ARRAY_ROW_COUNTS) returns SQL_PARC_BATCH).
+      // Instead, it adds them all up and returns a single count. This is
+      // bad news for us.
+      //
+      // In case of deleting by primary key (the affected row count is
+      // either 1 or 0), we can recognize the presumably successful case
+      // where the total affected row count is equal to the batch size
+      // (we can also recognize the "all unsuccessful" case where the
+      // total affected row count is 0). The unique_hint argument in the
+      // constructors below indicates whether this is a "0 or 1" DELETE
+      // statement.
+      //
+      // In all other situations (provided this is a batch), the result()
+      // function below returns the special result_unknown value.
+      //
+      delete_statement (connection_type& conn,
+                        const std::string& text,
+                        binding& param);
+
       delete_statement (connection_type& conn,
                         const std::string& text,
+                        bool unique_hint,
                         binding& param);
 
       delete_statement (connection_type& conn,
@@ -348,12 +493,49 @@ namespace odb
                         binding& param,
                         bool copy_text = true);
 
+      delete_statement (connection_type& conn,
+                        const char* text,
+                        bool unique_hint,
+                        binding& param,
+                        bool copy_text = true);
+
+      // Return the number of parameter sets (out of n) that were attempted.
+      //
+      std::size_t
+      execute (std::size_t n, multiple_exceptions& mex)
+      {
+        return execute (n, &mex);
+      }
+
+      // Return the number of rows affected (deleted) by the parameter
+      // set. If this is a batch (n > 1 in execute() call above) and it
+      // is impossible to determine the affected row count for each
+      // parameter set, then this function returns result_unknown. All
+      // other errors are reported by throwing exceptions.
+      //
+      using bulk_statement::result_unknown;
+
       unsigned long long
-      execute ();
+      result (std::size_t i);
+
+      unsigned long long
+      execute ()
+      {
+        execute (1, 0);
+        return result (0);
+      }
 
     private:
       delete_statement (const delete_statement&);
       delete_statement& operator= (const delete_statement&);
+
+    private:
+      std::size_t
+      execute (std::size_t, multiple_exceptions*);
+
+    private:
+      bool unique_;
+      unsigned long long result_;
     };
   }
 }
diff --git a/odb/mssql/statement.ixx b/odb/mssql/statement.ixx
index 89be46f..56e0ca4 100644
--- a/odb/mssql/statement.ixx
+++ b/odb/mssql/statement.ixx
@@ -6,44 +6,59 @@ namespace odb
 {
   namespace mssql
   {
-    inline unsigned long long insert_statement::
-    version ()
+    inline bulk_statement::
+    bulk_statement (connection_type& c,
+                    const std::string& text,
+                    statement_kind k,
+                    const binding* process,
+                    bool optimize,
+                    std::size_t batch,
+                    std::size_t skip,
+                    SQLUSMALLINT* status)
+        : statement (c, text, k, process, optimize),
+          status_ (batch == 1 ? 0 : status)
     {
-      unsigned long long r;
-
-      // The value is in the big-endian format.
-      //
-      unsigned char* p (reinterpret_cast<unsigned char*> (&r));
-      p[0] = version_[7];
-      p[1] = version_[6];
-      p[2] = version_[5];
-      p[3] = version_[4];
-      p[4] = version_[3];
-      p[5] = version_[2];
-      p[6] = version_[1];
-      p[7] = version_[0];
+      if (status_ != 0 && !empty ())
+        init (skip);
+    }
 
-      return r;
+    inline bulk_statement::
+    bulk_statement (connection_type& c,
+                    const char* text,
+                    statement_kind k,
+                    const binding* process,
+                    bool optimize,
+                    std::size_t batch,
+                    std::size_t skip,
+                    SQLUSMALLINT* status,
+                    bool copy_text)
+        : statement (c, text, k, process, optimize, copy_text),
+          status_ (batch == 1 ? 0 : status)
+    {
+      if (status_ != 0 && !empty ())
+        init (skip);
     }
 
+    // update_statement
+    //
     inline unsigned long long update_statement::
-    version ()
+    result (std::size_t i)
     {
-      unsigned long long r;
+      if (i != i_)
+        mex_->current (++i_); // mex cannot be NULL since this is a batch.
 
-      // The value is in the big-endian format.
-      //
-      unsigned char* p (reinterpret_cast<unsigned char*> (&r));
-      p[0] = version_[7];
-      p[1] = version_[6];
-      p[2] = version_[5];
-      p[3] = version_[4];
-      p[4] = version_[3];
-      p[5] = version_[2];
-      p[6] = version_[1];
-      p[7] = version_[0];
+      return result_;
+    }
+
+    // delete_statement
+    //
+    inline unsigned long long delete_statement::
+    result (std::size_t i)
+    {
+      if (i != i_)
+        mex_->current (++i_); // mex cannot be NULL since this is a batch.
 
-      return r;
+      return result_;
     }
   }
 }