#2953 merge update_multiple to main refs[t:2953]

git-svn-id: file:///svn/toku/tokudb@25124 c7de825b-a66e-492c-adef-691d508d4ae1

buildheader/db.h_4_1

@@ -256,7 +256,7 @@ struct __toku_db_env {
   int (*put_multiple)                         (DB_ENV *env, DB *src_db, DB_TXN *txn,
                                              const DBT *key, const DBT *val,
                                              uint32_t num_dbs, DB **db_array, DBT *keys, DBT *vals, uint32_t *flags_array,
-                                             void *extra) /* Insert into multiple dbs */;
+                                             void *extra) /* insert into multiple DBs */;
   int (*set_generate_row_callback_for_put)    (DB_ENV *env, 
                                              int (*generate_row_for_put)(DB *dest_db, DB *src_db,
                                                                          DBT *dest_key, DBT *dest_val,
@@ -265,17 +265,23 @@ struct __toku_db_env {
   int (*del_multiple)                         (DB_ENV *env, DB *src_db, DB_TXN *txn,
                                              const DBT *key, const DBT *val,
                                              uint32_t num_dbs, DB **db_array, DBT *keys, uint32_t *flags_array,
-                                             void *extra) /* Insert into multiple dbs */;
+                                             void *extra) /* delete from multiple DBs */;
   int (*set_generate_row_callback_for_del)    (DB_ENV *env, 
                                              int (*generate_row_for_del)(DB *dest_db, DB *src_db,
                                                                          DBT *dest_key,
                                                                          const DBT *src_key, const DBT *src_val,
                                                                          void *extra));
+  int (*update_multiple)                      (DB_ENV *env, DB *src_db, DB_TXN *txn,
+                                             const DBT *old_src_key, const DBT *old_src_data,
+                                             const DBT *new_src_key, const DBT *new_src_data,
+                                             uint32_t num_dbs, DB **db_array,
+                                             uint32_t num_dbts, DBT *keys, DBT *vals,
+                                             void *extra) /* update multiple DBs */;
   int (*get_redzone)                          (DB_ENV *env, int *redzone) /* get the redzone limit */;
   int (*set_redzone)                          (DB_ENV *env, int redzone) /* set the redzone limit in percent of total space */;
   int (*set_lk_max_memory)                    (DB_ENV *env, uint64_t max);
   int (*get_lk_max_memory)                    (DB_ENV *env, uint64_t *max);
-  void* __toku_dummy0[16];
+  void* __toku_dummy0[15];
   char __toku_dummy1[64];
   void *api1_internal; /* 32-bit offset=212 size=4, 64=bit offset=360 size=8 */
   void* __toku_dummy2[7];

buildheader/db.h_4_3

@@ -258,7 +258,7 @@ struct __toku_db_env {
   int (*put_multiple)                         (DB_ENV *env, DB *src_db, DB_TXN *txn,
                                              const DBT *key, const DBT *val,
                                              uint32_t num_dbs, DB **db_array, DBT *keys, DBT *vals, uint32_t *flags_array,
-                                             void *extra) /* Insert into multiple dbs */;
+                                             void *extra) /* insert into multiple DBs */;
   int (*set_generate_row_callback_for_put)    (DB_ENV *env, 
                                              int (*generate_row_for_put)(DB *dest_db, DB *src_db,
                                                                          DBT *dest_key, DBT *dest_val,
@@ -267,17 +267,23 @@ struct __toku_db_env {
   int (*del_multiple)                         (DB_ENV *env, DB *src_db, DB_TXN *txn,
                                              const DBT *key, const DBT *val,
                                              uint32_t num_dbs, DB **db_array, DBT *keys, uint32_t *flags_array,
-                                             void *extra) /* Insert into multiple dbs */;
+                                             void *extra) /* delete from multiple DBs */;
   int (*set_generate_row_callback_for_del)    (DB_ENV *env, 
                                              int (*generate_row_for_del)(DB *dest_db, DB *src_db,
                                                                          DBT *dest_key,
                                                                          const DBT *src_key, const DBT *src_val,
                                                                          void *extra));
+  int (*update_multiple)                      (DB_ENV *env, DB *src_db, DB_TXN *txn,
+                                             const DBT *old_src_key, const DBT *old_src_data,
+                                             const DBT *new_src_key, const DBT *new_src_data,
+                                             uint32_t num_dbs, DB **db_array,
+                                             uint32_t num_dbts, DBT *keys, DBT *vals,
+                                             void *extra) /* update multiple DBs */;
   int (*get_redzone)                          (DB_ENV *env, int *redzone) /* get the redzone limit */;
   int (*set_redzone)                          (DB_ENV *env, int redzone) /* set the redzone limit in percent of total space */;
   int (*set_lk_max_memory)                    (DB_ENV *env, uint64_t max);
   int (*get_lk_max_memory)                    (DB_ENV *env, uint64_t *max);
-  void* __toku_dummy0[16];
+  void* __toku_dummy0[15];
   char __toku_dummy1[96];
   void *api1_internal; /* 32-bit offset=244 size=4, 64=bit offset=392 size=8 */
   void* __toku_dummy2[7];

buildheader/db.h_4_4

@@ -258,7 +258,7 @@ struct __toku_db_env {
   int (*put_multiple)                         (DB_ENV *env, DB *src_db, DB_TXN *txn,
                                              const DBT *key, const DBT *val,
                                              uint32_t num_dbs, DB **db_array, DBT *keys, DBT *vals, uint32_t *flags_array,
-                                             void *extra) /* Insert into multiple dbs */;
+                                             void *extra) /* insert into multiple DBs */;
   int (*set_generate_row_callback_for_put)    (DB_ENV *env, 
                                              int (*generate_row_for_put)(DB *dest_db, DB *src_db,
                                                                          DBT *dest_key, DBT *dest_val,
@@ -267,17 +267,23 @@ struct __toku_db_env {
   int (*del_multiple)                         (DB_ENV *env, DB *src_db, DB_TXN *txn,
                                              const DBT *key, const DBT *val,
                                              uint32_t num_dbs, DB **db_array, DBT *keys, uint32_t *flags_array,
-                                             void *extra) /* Insert into multiple dbs */;
+                                             void *extra) /* delete from multiple DBs */;
   int (*set_generate_row_callback_for_del)    (DB_ENV *env, 
                                              int (*generate_row_for_del)(DB *dest_db, DB *src_db,
                                                                          DBT *dest_key,
                                                                          const DBT *src_key, const DBT *src_val,
                                                                          void *extra));
+  int (*update_multiple)                      (DB_ENV *env, DB *src_db, DB_TXN *txn,
+                                             const DBT *old_src_key, const DBT *old_src_data,
+                                             const DBT *new_src_key, const DBT *new_src_data,
+                                             uint32_t num_dbs, DB **db_array,
+                                             uint32_t num_dbts, DBT *keys, DBT *vals,
+                                             void *extra) /* update multiple DBs */;
   int (*get_redzone)                          (DB_ENV *env, int *redzone) /* get the redzone limit */;
   int (*set_redzone)                          (DB_ENV *env, int redzone) /* set the redzone limit in percent of total space */;
   int (*set_lk_max_memory)                    (DB_ENV *env, uint64_t max);
   int (*get_lk_max_memory)                    (DB_ENV *env, uint64_t *max);
-  void* __toku_dummy0[31];
+  void* __toku_dummy0[30];
   char __toku_dummy1[128];
   void *api1_internal; /* 32-bit offset=336 size=4, 64=bit offset=544 size=8 */
   void* __toku_dummy2[7];

buildheader/db.h_4_5

@@ -257,7 +257,7 @@ struct __toku_db_env {
   int (*put_multiple)                         (DB_ENV *env, DB *src_db, DB_TXN *txn,
                                              const DBT *key, const DBT *val,
                                              uint32_t num_dbs, DB **db_array, DBT *keys, DBT *vals, uint32_t *flags_array,
-                                             void *extra) /* Insert into multiple dbs */;
+                                             void *extra) /* insert into multiple DBs */;
   void *app_private; /* 32-bit offset=52 size=4, 64=bit offset=104 size=8 */
   int (*set_generate_row_callback_for_put)    (DB_ENV *env, 
                                              int (*generate_row_for_put)(DB *dest_db, DB *src_db,
@@ -267,17 +267,23 @@ struct __toku_db_env {
   int (*del_multiple)                         (DB_ENV *env, DB *src_db, DB_TXN *txn,
                                              const DBT *key, const DBT *val,
                                              uint32_t num_dbs, DB **db_array, DBT *keys, uint32_t *flags_array,
-                                             void *extra) /* Insert into multiple dbs */;
+                                             void *extra) /* delete from multiple DBs */;
   int (*set_generate_row_callback_for_del)    (DB_ENV *env, 
                                              int (*generate_row_for_del)(DB *dest_db, DB *src_db,
                                                                          DBT *dest_key,
                                                                          const DBT *src_key, const DBT *src_val,
                                                                          void *extra));
+  int (*update_multiple)                      (DB_ENV *env, DB *src_db, DB_TXN *txn,
+                                             const DBT *old_src_key, const DBT *old_src_data,
+                                             const DBT *new_src_key, const DBT *new_src_data,
+                                             uint32_t num_dbs, DB **db_array,
+                                             uint32_t num_dbts, DBT *keys, DBT *vals,
+                                             void *extra) /* update multiple DBs */;
   int (*get_redzone)                          (DB_ENV *env, int *redzone) /* get the redzone limit */;
   int (*set_redzone)                          (DB_ENV *env, int redzone) /* set the redzone limit in percent of total space */;
   int (*set_lk_max_memory)                    (DB_ENV *env, uint64_t max);
   int (*get_lk_max_memory)                    (DB_ENV *env, uint64_t *max);
-  void* __toku_dummy0[31];
+  void* __toku_dummy0[30];
   char __toku_dummy1[128];
   void *api1_internal; /* 32-bit offset=336 size=4, 64=bit offset=544 size=8 */
   void* __toku_dummy2[8];

buildheader/db.h_4_6

@@ -258,7 +258,7 @@ struct __toku_db_env {
   int (*put_multiple)                         (DB_ENV *env, DB *src_db, DB_TXN *txn,
                                              const DBT *key, const DBT *val,
                                              uint32_t num_dbs, DB **db_array, DBT *keys, DBT *vals, uint32_t *flags_array,
-                                             void *extra) /* Insert into multiple dbs */;
+                                             void *extra) /* insert into multiple DBs */;
   void *app_private; /* 32-bit offset=52 size=4, 64=bit offset=104 size=8 */
   int (*set_generate_row_callback_for_put)    (DB_ENV *env, 
                                              int (*generate_row_for_put)(DB *dest_db, DB *src_db,
@@ -268,17 +268,23 @@ struct __toku_db_env {
   int (*del_multiple)                         (DB_ENV *env, DB *src_db, DB_TXN *txn,
                                              const DBT *key, const DBT *val,
                                              uint32_t num_dbs, DB **db_array, DBT *keys, uint32_t *flags_array,
-                                             void *extra) /* Insert into multiple dbs */;
+                                             void *extra) /* delete from multiple DBs */;
   int (*set_generate_row_callback_for_del)    (DB_ENV *env, 
                                              int (*generate_row_for_del)(DB *dest_db, DB *src_db,
                                                                          DBT *dest_key,
                                                                          const DBT *src_key, const DBT *src_val,
                                                                          void *extra));
+  int (*update_multiple)                      (DB_ENV *env, DB *src_db, DB_TXN *txn,
+                                             const DBT *old_src_key, const DBT *old_src_data,
+                                             const DBT *new_src_key, const DBT *new_src_data,
+                                             uint32_t num_dbs, DB **db_array,
+                                             uint32_t num_dbts, DBT *keys, DBT *vals,
+                                             void *extra) /* update multiple DBs */;
   int (*get_redzone)                          (DB_ENV *env, int *redzone) /* get the redzone limit */;
   int (*set_redzone)                          (DB_ENV *env, int redzone) /* set the redzone limit in percent of total space */;
   int (*set_lk_max_memory)                    (DB_ENV *env, uint64_t max);
   int (*get_lk_max_memory)                    (DB_ENV *env, uint64_t *max);
-  void* __toku_dummy0[32];
+  void* __toku_dummy0[31];
   char __toku_dummy1[144];
   void *api1_internal; /* 32-bit offset=356 size=4, 64=bit offset=568 size=8 */
   void* __toku_dummy2[8];

buildheader/make_db_h.c

@@ -556,7 +556,7 @@ int main (int argc __attribute__((__unused__)), char *const argv[] __attribute__
                              "int (*put_multiple)                         (DB_ENV *env, DB *src_db, DB_TXN *txn,\n"
                              "                                             const DBT *key, const DBT *val,\n"
                              "                                             uint32_t num_dbs, DB **db_array, DBT *keys, DBT *vals, uint32_t *flags_array,\n"
-                             "                                             void *extra) /* Insert into multiple dbs */",
+                             "                                             void *extra) /* insert into multiple DBs */",
                              "int (*set_generate_row_callback_for_put)    (DB_ENV *env, \n"
                              "                                             int (*generate_row_for_put)(DB *dest_db, DB *src_db,\n"
                              "                                                                         DBT *dest_key, DBT *dest_val,\n"
@@ -565,12 +565,18 @@ int main (int argc __attribute__((__unused__)), char *const argv[] __attribute__
                              "int (*del_multiple)                         (DB_ENV *env, DB *src_db, DB_TXN *txn,\n"
                              "                                             const DBT *key, const DBT *val,\n"
                              "                                             uint32_t num_dbs, DB **db_array, DBT *keys, uint32_t *flags_array,\n"
-                             "                                             void *extra) /* Insert into multiple dbs */",
+                             "                                             void *extra) /* delete from multiple DBs */",
                              "int (*set_generate_row_callback_for_del)    (DB_ENV *env, \n"
                              "                                             int (*generate_row_for_del)(DB *dest_db, DB *src_db,\n"
                              "                                                                         DBT *dest_key,\n"
                              "                                                                         const DBT *src_key, const DBT *src_val,\n"
                              "                                                                         void *extra))",
+                             "int (*update_multiple)                      (DB_ENV *env, DB *src_db, DB_TXN *txn,\n"
+                             "                                             const DBT *old_src_key, const DBT *old_src_data,\n"
+                             "                                             const DBT *new_src_key, const DBT *new_src_data,\n"
+                             "                                             uint32_t num_dbs, DB **db_array,\n"
+                             "                                             uint32_t num_dbts, DBT *keys, DBT *vals,\n"
+                             "                                             void *extra) /* update multiple DBs */",
                              "int (*get_redzone)                          (DB_ENV *env, int *redzone) /* get the redzone limit */",
                              "int (*set_redzone)                          (DB_ENV *env, int redzone) /* set the redzone limit in percent of total space */",
                              "int (*set_lk_max_memory)                    (DB_ENV *env, uint64_t max)",

buildheader/tdb.h

@@ -258,7 +258,7 @@ struct __toku_db_env {
   int (*put_multiple)                         (DB_ENV *env, DB *src_db, DB_TXN *txn,
                                              const DBT *key, const DBT *val,
                                              uint32_t num_dbs, DB **db_array, DBT *keys, DBT *vals, uint32_t *flags_array,
-                                             void *extra) /* Insert into multiple dbs */;
+                                             void *extra) /* insert into multiple DBs */;
   void *app_private;
   int (*set_generate_row_callback_for_put)    (DB_ENV *env, 
                                              int (*generate_row_for_put)(DB *dest_db, DB *src_db,
@@ -268,12 +268,18 @@ struct __toku_db_env {
   int (*del_multiple)                         (DB_ENV *env, DB *src_db, DB_TXN *txn,
                                              const DBT *key, const DBT *val,
                                              uint32_t num_dbs, DB **db_array, DBT *keys, uint32_t *flags_array,
-                                             void *extra) /* Insert into multiple dbs */;
+                                             void *extra) /* delete from multiple DBs */;
   int (*set_generate_row_callback_for_del)    (DB_ENV *env, 
                                              int (*generate_row_for_del)(DB *dest_db, DB *src_db,
                                                                          DBT *dest_key,
                                                                          const DBT *src_key, const DBT *src_val,
                                                                          void *extra));
+  int (*update_multiple)                      (DB_ENV *env, DB *src_db, DB_TXN *txn,
+                                             const DBT *old_src_key, const DBT *old_src_data,
+                                             const DBT *new_src_key, const DBT *new_src_data,
+                                             uint32_t num_dbs, DB **db_array,
+                                             uint32_t num_dbts, DBT *keys, DBT *vals,
+                                             void *extra) /* update multiple DBs */;
   int (*get_redzone)                          (DB_ENV *env, int *redzone) /* get the redzone limit */;
   int (*set_redzone)                          (DB_ENV *env, int redzone) /* set the redzone limit in percent of total space */;
   int (*set_lk_max_memory)                    (DB_ENV *env, uint64_t max);

include/db.h

@@ -258,7 +258,7 @@ struct __toku_db_env {
   int (*put_multiple)                         (DB_ENV *env, DB *src_db, DB_TXN *txn,
                                              const DBT *key, const DBT *val,
                                              uint32_t num_dbs, DB **db_array, DBT *keys, DBT *vals, uint32_t *flags_array,
-                                             void *extra) /* Insert into multiple dbs */;
+                                             void *extra) /* insert into multiple DBs */;
   void *app_private;
   int (*set_generate_row_callback_for_put)    (DB_ENV *env, 
                                              int (*generate_row_for_put)(DB *dest_db, DB *src_db,
@@ -268,12 +268,18 @@ struct __toku_db_env {
   int (*del_multiple)                         (DB_ENV *env, DB *src_db, DB_TXN *txn,
                                              const DBT *key, const DBT *val,
                                              uint32_t num_dbs, DB **db_array, DBT *keys, uint32_t *flags_array,
-                                             void *extra) /* Insert into multiple dbs */;
+                                             void *extra) /* delete from multiple DBs */;
   int (*set_generate_row_callback_for_del)    (DB_ENV *env, 
                                              int (*generate_row_for_del)(DB *dest_db, DB *src_db,
                                                                          DBT *dest_key,
                                                                          const DBT *src_key, const DBT *src_val,
                                                                          void *extra));
+  int (*update_multiple)                      (DB_ENV *env, DB *src_db, DB_TXN *txn,
+                                             const DBT *old_src_key, const DBT *old_src_data,
+                                             const DBT *new_src_key, const DBT *new_src_data,
+                                             uint32_t num_dbs, DB **db_array,
+                                             uint32_t num_dbts, DBT *keys, DBT *vals,
+                                             void *extra) /* update multiple DBs */;
   int (*get_redzone)                          (DB_ENV *env, int *redzone) /* get the redzone limit */;
   int (*set_redzone)                          (DB_ENV *env, int redzone) /* set the redzone limit in percent of total space */;
   int (*set_lk_max_memory)                    (DB_ENV *env, uint64_t max);

newbrt/brt.c

@@ -2638,6 +2638,22 @@ toku_brt_load(BRT brt, TOKUTXN txn, char const * new_iname, int do_fsync, LSN *l
     return r;
 }
 
+int 
+toku_brt_log_put (TOKUTXN txn, BRT brt, const DBT *key, const DBT *val) {
+    int r = 0;
+    TOKULOGGER logger = toku_txn_logger(txn);
+    if (logger && brt->h->txnid_that_suppressed_recovery_logs == TXNID_NONE) {
+        BYTESTRING keybs = {.len=key->size, .data=key->data};
+        BYTESTRING valbs = {.len=val->size, .data=val->data};
+        TXNID xid = toku_txn_get_txnid(txn);
+        // if (type == BRT_INSERT)
+            r = toku_log_enq_insert(logger, (LSN*)0, 0, toku_cachefile_filenum(brt->cf), xid, keybs, valbs);
+        // else
+            // r = toku_log_enq_insert_no_overwrite(logger, (LSN*)0, 0, toku_cachefile_filenum(brt->cf), xid, keybs, valbs);
+    }
+    return r;
+}
+
 int
 toku_brt_log_put_multiple (TOKUTXN txn, BRT src_brt, BRT *brts, int num_brts, const DBT *key, const DBT *val) {
     int r = 0;
@@ -2666,7 +2682,8 @@ toku_brt_log_put_multiple (TOKUTXN txn, BRT src_brt, BRT *brts, int num_brts, co
     return r;
 }
 
-int toku_brt_maybe_insert (BRT brt, DBT *key, DBT *val, TOKUTXN txn, BOOL oplsn_valid, LSN oplsn, int do_logging, enum brt_msg_type type) {
+int 
+toku_brt_maybe_insert (BRT brt, DBT *key, DBT *val, TOKUTXN txn, BOOL oplsn_valid, LSN oplsn, int do_logging, enum brt_msg_type type) {
     lazy_assert(type==BRT_INSERT || type==BRT_INSERT_NO_OVERWRITE);
     int r = 0;
     XIDS message_xids = xids_get_root_xids(); //By default use committed messages
@@ -2714,6 +2731,18 @@ int toku_brt_delete(BRT brt, DBT *key, TOKUTXN txn) {
     return toku_brt_maybe_delete(brt, key, txn, FALSE, ZERO_LSN, TRUE);
 }
 
+int
+toku_brt_log_del(TOKUTXN txn, BRT brt, const DBT *key) {
+    int r = 0;
+    TOKULOGGER logger = toku_txn_logger(txn);
+    if (logger && brt->h->txnid_that_suppressed_recovery_logs == TXNID_NONE) {
+        BYTESTRING keybs = {.len=key->size, .data=key->data};
+        TXNID xid = toku_txn_get_txnid(txn);
+        r = toku_log_enq_delete_any(logger, (LSN*)0, 0, toku_cachefile_filenum(brt->cf), xid, keybs);
+    }
+    return r;
+}
+
 int
 toku_brt_log_del_multiple (TOKUTXN txn, BRT src_brt, BRT *brts, int num_brts, const DBT *key, const DBT *val) {
     int r = 0;
@@ -2742,7 +2771,8 @@ toku_brt_log_del_multiple (TOKUTXN txn, BRT src_brt, BRT *brts, int num_brts, co
     return r;
 }
 
-int toku_brt_maybe_delete(BRT brt, DBT *key, TOKUTXN txn, BOOL oplsn_valid, LSN oplsn, int do_logging) {
+int 
+toku_brt_maybe_delete(BRT brt, DBT *key, TOKUTXN txn, BOOL oplsn_valid, LSN oplsn, int do_logging) {
     int r;
     XIDS message_xids = xids_get_root_xids(); //By default use committed messages
     TXNID xid = toku_txn_get_txnid(txn);

newbrt/brt.h

@@ -74,7 +74,9 @@ int toku_brt_maybe_insert (BRT brt, DBT *k, DBT *v, TOKUTXN txn, BOOL oplsn_vali
 int toku_brt_load_recovery(TOKUTXN txn, char const * old_iname, char const * new_iname, int do_fsync, int do_log, LSN *load_lsn)  __attribute__ ((warn_unused_result));
 int toku_brt_load(BRT brt, TOKUTXN txn, char const * new_iname, int do_fsync, LSN *get_lsn)  __attribute__ ((warn_unused_result));
 int toku_brt_log_put_multiple (TOKUTXN txn, BRT src_brt, BRT *brts, int num_brts, const DBT *key, const DBT *val)  __attribute__ ((warn_unused_result));
+int toku_brt_log_put (TOKUTXN txn, BRT brt, const DBT *key, const DBT *val)  __attribute__ ((warn_unused_result));
 int toku_brt_log_del_multiple (TOKUTXN txn, BRT src_brt, BRT *brts, int num_brts, const DBT *key, const DBT *val) __attribute__ ((warn_unused_result));
+int toku_brt_log_del (TOKUTXN txn, BRT brt, const DBT *key) __attribute__ ((warn_unused_result));
 
 // Effect: Delete a key from a brt
 // Returns 0 if successful

newbrt/tests/keyrange-unflat.c

@@ -75,10 +75,14 @@ test_main (int argc , const char *argv[]) {
     int ncpus = 0;
     for (int i = 1; i < argc; i++) {
         const char *arg = argv[i];
-        if (strcmp(arg, "-v") == 0) {
+        if (strcmp(arg, "-v") == 0 || strcmp(arg, "--verbose") == 0) {
             verbose++;
             continue;
         }
+        if (strcmp(arg, "-q") == 0 || strcmp(arg, "--quiet") == 0) {
+            verbose = 0;
+            continue;
+        }
         if (strcmp(arg, "--ncpus") == 0 && i+1 < argc) {
             ncpus = atoi(argv[++i]);
             continue;

src/export.map

@@ -71,6 +71,7 @@
 
    toku_do_assert;
    toku_do_assert_fail;
+   toku_do_assert_zero_fail;
    toku_set_assert_on_write_enospc;
 
    toku_test_db_redirect_dictionary;

src/tests/Makefile

@@ -88,6 +88,8 @@ BDB_DONTRUN_TESTS = \
         checkpoint_stress \
         checkpoint_truncate_1 \
 	cursor-isolation \
+	del-simple \
+	del-multiple \
         diskfull \
         env-put-multiple \
         env_startup \
@@ -117,14 +119,19 @@ BDB_DONTRUN_TESTS = \
         powerfail \
 	preload-3.1-db \
         progress \
+	put-multiple \
         recover-2483 \
         recover-compare-db \
         recover-compare-db-descriptor \
+	recover-del-multiple \
+	recover-del-multiple-abort \
         recover-delboth-after-checkpoint \
         recover-delboth-checkpoint \
         recover-fclose-in-checkpoint \
         recover-loader-test \
         recover-lsn-filter-multiple \
+	recover-put-multiple \
+	recover-put-multiple-abort \
         recover-put-multiple-fdelete-all \
         recover-put-multiple-fdelete-some \
         recover-split-checkpoint \
@@ -133,6 +140,8 @@ BDB_DONTRUN_TESTS = \
 	recover-test-logsuppress-put \
 	recover-upgrade-db-descriptor-multihandle \
 	recover-upgrade-db-descriptor \
+	recover-update-multiple \
+	recover-update-multiple-abort \
         recovery_fileops_stress \
         recovery_fileops_unit \
         recovery_stress \
@@ -167,6 +176,9 @@ BDB_DONTRUN_TESTS = \
         test_txn_nested4 \
         test_txn_nested5 \
         transactional_fileops \
+	update-multiple-nochange \
+	update-multiple-key0 \
+	update-multiple-data-diagonal \
 	upgrade-test-1 \
 	upgrade-test-2 \
 	upgrade-test-3 \

src/tests/del-multiple.c

+#include "test.h"
+
+// verify that del_multiple deletes the correct key from N dictionaries
+// verify that del_multiple locks the correct key for N dictionaries
+
+static int
+get_key(int i, int dbnum) {
+    return htonl(i + dbnum);
+}
+
+static void
+get_data(int *v, int i, int ndbs) {
+    for (int dbnum = 0; dbnum < ndbs; dbnum++) {
+        v[dbnum] = get_key(i, dbnum);
+    }
+}
+
+static int
+del_callback(DB *dest_db, DB *src_db, DBT *dest_key, const DBT *src_key, const DBT *src_data, void *extra) {
+    dest_db = dest_db; src_db = src_db; dest_key = dest_key; src_key = src_key; src_data = src_data;
+    assert(src_db == NULL);
+    assert(extra == NULL);
+
+    unsigned int dbnum;
+    assert(dest_db->descriptor->dbt.size == sizeof dbnum);
+    memcpy(&dbnum, dest_db->descriptor->dbt.data, sizeof dbnum);
+    assert(dbnum < src_data->size / sizeof (int));
+
+    int *pri_data = (int *) src_data->data;
+
+    assert(dest_key->flags == 0);
+    dest_key->size = sizeof (int);
+    dest_key->data = &pri_data[dbnum];
+    
+    return 0;
+}
+
+static void
+verify_locked(DB_ENV *env, DB *db, int k) {
+    int r;
+    DB_TXN *txn = NULL;
+    r = env->txn_begin(env, NULL, &txn, 0); assert_zero(r);
+    DBT key; dbt_init(&key, &k, sizeof k);
+    r = db->del(db, txn, &key, DB_DELETE_ANY); assert(r == DB_LOCK_NOTGRANTED);
+    r = txn->abort(txn); assert_zero(r);
+}
+
+static void
+verify_empty(DB_ENV *env, DB *db) {
+    int r;
+    DB_TXN *txn = NULL;
+    r = env->txn_begin(env, NULL, &txn, 0); assert_zero(r);
+
+    DBC *cursor = NULL;
+    r = db->cursor(db, txn, &cursor, 0); assert_zero(r);
+    int i;
+    for (i = 0; ; i++) {
+        DBT key; memset(&key, 0, sizeof key);
+        DBT val; memset(&val, 0, sizeof val);
+        r = cursor->c_get(cursor, &key, &val, DB_NEXT);
+        if (r != 0)
+            break;
+    }
+    assert_zero(i);
+    r = cursor->c_close(cursor); assert_zero(r);
+    r = txn->commit(txn, 0); assert_zero(r);
+}
+
+static void
+verify_del_multiple(DB_ENV *env, DB *db[], int ndbs, int nrows) {
+    int r;
+    DB_TXN *deltxn = NULL;
+    r = env->txn_begin(env, NULL, &deltxn, 0); assert_zero(r);
+    for (int i = 0; i < nrows; i++) {
+        int k = get_key(i, 0);
+        DBT pri_key; dbt_init(&pri_key, &k, sizeof k);
+        int v[ndbs]; get_data(v, i, ndbs);
+        DBT pri_data; dbt_init(&pri_data, &v[0], sizeof v);
+        DBT keys[ndbs]; memset(keys, 0, sizeof keys);
+        uint32_t flags[ndbs]; memset(flags, 0, sizeof flags);
+        r = env->del_multiple(env, NULL, deltxn, &pri_key, &pri_data, ndbs, db, keys, flags, NULL); assert_zero(r);
+        for (int dbnum = 0; dbnum < ndbs; dbnum++) 
+            verify_locked(env, db[dbnum], get_key(i, dbnum));
+    }
+    r = deltxn->commit(deltxn, 0); assert_zero(r);
+    for (int dbnum = 0; dbnum < ndbs; dbnum++) 
+        verify_empty(env, db[dbnum]);
+}
+
+static void
+populate_primary(DB_ENV *env, DB *db, int ndbs, int nrows) {
+    int r;
+    DB_TXN *txn = NULL;
+    r = env->txn_begin(env, NULL, &txn, 0); assert_zero(r);
+
+    // populate
+    for (int i = 0; i < nrows; i++) {
+        int k = get_key(i, 0);
+        int v[ndbs]; get_data(v, i, ndbs);
+        DBT key; dbt_init(&key, &k, sizeof k);
+        DBT val; dbt_init(&val, &v[0], sizeof v);
+        r = db->put(db, txn, &key, &val, DB_YESOVERWRITE); assert_zero(r);
+    }
+
+    r = txn->commit(txn, 0); assert_zero(r);
+}
+
+static void
+populate_secondary(DB_ENV *env, DB *db, int dbnum, int nrows) {
+    int r;
+    DB_TXN *txn = NULL;
+    r = env->txn_begin(env, NULL, &txn, 0); assert_zero(r);
+
+    // populate
+    for (int i = 0; i < nrows; i++) {
+        int k = get_key(i, dbnum);
+        DBT key; dbt_init(&key, &k, sizeof k);
+        DBT val; dbt_init(&val, NULL, 0);
+        r = db->put(db, txn, &key, &val, DB_YESOVERWRITE); assert_zero(r);
+    }
+
+    r = txn->commit(txn, 0); assert_zero(r);
+}
+
+static void
+run_test(int ndbs, int nrows) {
+    int r;
+    DB_ENV *env = NULL;
+    r = db_env_create(&env, 0); assert_zero(r);
+
+    r = env->set_generate_row_callback_for_del(env, del_callback); assert_zero(r);
+
+    r = env->open(env, ENVDIR, DB_INIT_MPOOL|DB_CREATE|DB_THREAD |DB_INIT_LOCK|DB_INIT_LOG|DB_INIT_TXN|DB_PRIVATE, S_IRWXU+S_IRWXG+S_IRWXO); assert_zero(r);
+
+    DB *db[ndbs];
+    for (int dbnum = 0; dbnum < ndbs; dbnum++) {
+        r = db_create(&db[dbnum], env, 0); assert_zero(r);
+
+        DBT dbt_dbnum; dbt_init(&dbt_dbnum, &dbnum, sizeof dbnum);
+        r = db[dbnum]->set_descriptor(db[dbnum], 1, &dbt_dbnum); assert_zero(r);
+
+        char dbname[32]; sprintf(dbname, "%d.tdb", dbnum);
+        r = db[dbnum]->open(db[dbnum], NULL, dbname, NULL, DB_BTREE, DB_AUTO_COMMIT+DB_CREATE, S_IRWXU+S_IRWXG+S_IRWXO); assert_zero(r);
+    }
+
+    for (int dbnum = 0; dbnum < ndbs; dbnum++) {
+        if (dbnum == 0)
+            populate_primary(env, db[dbnum], ndbs, nrows);
+        else
+            populate_secondary(env, db[dbnum], dbnum, nrows);
+    }
+
+    verify_del_multiple(env, db, ndbs, nrows);
+
+    for (int dbnum = 0; dbnum < ndbs; dbnum++) 
+        r = db[dbnum]->close(db[dbnum], 0); assert_zero(r);
+
+    r = env->close(env, 0); assert_zero(r);
+}
+
+int
+test_main(int argc, char * const argv[]) {
+    int r;
+    int ndbs = 2;
+    int nrows = 2;
+
+    // parse_args(argc, argv);
+    for (int i = 1; i < argc; i++) {
+        char * const arg = argv[i];
+        if (strcmp(arg, "-v") == 0) {
+            verbose++;
+            continue;
+        }
+        if (strcmp(arg, "-q") == 0) {
+            verbose = 0;
+            continue;
+        }
+        if (strcmp(arg, "--ndbs") == 0 && i+1 < argc) {
+            ndbs = atoi(argv[++i]);
+            continue;
+        }
+        if (strcmp(arg, "--nrows") == 0 && i+1 < argc) {
+            nrows = atoi(argv[++i]);
+            continue;
+        }
+    }
+
+    r = system("rm -rf " ENVDIR); assert_zero(r);
+    r = toku_os_mkdir(ENVDIR, S_IRWXU+S_IRWXG+S_IRWXO); assert_zero(r);
+
+    run_test(ndbs, nrows);
+
+    return 0;
+}
+

src/tests/del-simple.c

+#include "test.h"
+
+// verify that deletes on single dictionaries delete the correct key and create the correct lock
+
+static void
+verify_locked(DB_ENV *env, DB *db, int k) {
+    int r;
+    DB_TXN *txn = NULL;
+    r = env->txn_begin(env, NULL, &txn, 0); assert_zero(r);
+    DBT key; dbt_init(&key, &k, sizeof k);
+    r = db->del(db, txn, &key, DB_DELETE_ANY); assert(r == DB_LOCK_NOTGRANTED);
+    r = txn->abort(txn); assert_zero(r);
+}
+
+static void
+verify_empty(DB_ENV *env, DB *db) {
+    int r;
+    DB_TXN *txn = NULL;
+    r = env->txn_begin(env, NULL, &txn, 0); assert_zero(r);
+
+    DBC *cursor = NULL;
+    r = db->cursor(db, txn, &cursor, 0); assert_zero(r);
+    int i;
+    for (i = 0; ; i++) {
+        DBT key; memset(&key, 0, sizeof key);
+        DBT val; memset(&val, 0, sizeof val);
+        r = cursor->c_get(cursor, &key, &val, DB_NEXT);
+        if (r != 0)
+            break;
+    }
+    assert_zero(i);
+    r = cursor->c_close(cursor); assert_zero(r);
+    r = txn->commit(txn, 0); assert_zero(r);
+}
+
+static void
+verify_del(DB_ENV *env, DB *db, int nrows) {
+    int r;
+    DB_TXN *deltxn = NULL;
+    r = env->txn_begin(env, NULL, &deltxn, 0); assert_zero(r);
+    for (int i = 0; i < nrows; i++) {
+        int k = htonl(i);
+        DBT key; dbt_init(&key, &k, sizeof k);
+        r = db->del(db, deltxn, &key, DB_DELETE_ANY); assert_zero(r);
+        verify_locked(env, db, k);
+    }
+    r = deltxn->commit(deltxn, 0); assert_zero(r);
+    verify_empty(env, db);
+}
+
+static void
+test_del(int nrows) {
+    int r;
+    DB_ENV *env = NULL;
+    r = db_env_create(&env, 0); assert_zero(r);
+
+    r = env->open(env, ENVDIR, DB_INIT_MPOOL|DB_CREATE|DB_THREAD |DB_INIT_LOCK|DB_INIT_LOG|DB_INIT_TXN|DB_PRIVATE, S_IRWXU+S_IRWXG+S_IRWXO); assert_zero(r);
+
+    DB *db = NULL;
+    r = db_create(&db, env, 0); assert_zero(r);
+
+    r = db->open(db, NULL, "test.db", NULL, DB_BTREE, DB_AUTO_COMMIT+DB_CREATE, S_IRWXU+S_IRWXG+S_IRWXO); assert_zero(r);
+
+    DB_TXN *txn = NULL;
+    r = env->txn_begin(env, NULL, &txn, 0); assert_zero(r);
+
+    // populate
+    for (int i = 0; i < nrows; i++) {
+        int k = htonl(i);
+        int v = i;
+        DBT key; dbt_init(&key, &k, sizeof k);
+        DBT val; dbt_init(&val, &v, sizeof v);
+        r = db->put(db, txn, &key, &val, DB_YESOVERWRITE); assert_zero(r);
+    }
+
+    r = txn->commit(txn, 0); assert_zero(r);
+
+    verify_del(env, db, nrows);
+
+    r = db->close(db, 0); assert_zero(r);
+
+    r = env->close(env, 0); assert_zero(r);
+}
+
+int
+test_main(int argc, char * const argv[]) {
+    int r;
+    int nrows = 2;
+
+    // parse_args(argc, argv);
+    for (int i = 1; i < argc; i++) {
+        char * const arg = argv[i];
+        if (strcmp(arg, "-v") == 0) {
+            verbose++;
+            continue;
+        }
+        if (strcmp(arg, "-q") == 0) {
+            verbose = 0;
+            continue;
+        }
+        if (strcmp(arg, "--nrows") == 0 && i+1 < argc) {
+            nrows = atoi(argv[++i]);
+            continue;
+        }
+    }
+
+    r = system("rm -rf " ENVDIR); assert_zero(r);
+    r = toku_os_mkdir(ENVDIR, S_IRWXU+S_IRWXG+S_IRWXO); assert_zero(r);
+
+    test_del(nrows);
+
+    return 0;
+}
+

src/tests/put-multiple.c

+#include "test.h"
+
+// verify that put_multiple inserts the correct rows into N dictionaries
+// verify that pu_multiple locks the correct keys for N dictionaries
+
+static int
+get_key(int i, int dbnum) {
+    return htonl(i + dbnum);
+}
+
+static void
+get_data(int *v, int i, int ndbs) {
+    for (int dbnum = 0; dbnum < ndbs; dbnum++) {
+        v[dbnum] = get_key(i, dbnum);
+    }
+}
+
+static int
+put_callback(DB *dest_db, DB *src_db, DBT *dest_key, DBT *dest_data, const DBT *src_key, const DBT *src_data, void *extra) {
+    dest_db = dest_db; src_db = src_db; dest_key = dest_key; dest_data = dest_data; src_key = src_key; src_data = src_data;
+    assert(src_db == NULL);
+    assert(extra == NULL);
+
+    unsigned int dbnum;
+    assert(dest_db->descriptor->dbt.size == sizeof dbnum);
+    memcpy(&dbnum, dest_db->descriptor->dbt.data, sizeof dbnum);
+    assert(dbnum < src_data->size / sizeof (int));
+
+    int *pri_data = (int *) src_data->data;
+
+    assert(dest_key->flags == 0);
+    dest_key->size = sizeof (int);
+    dest_key->data = &pri_data[dbnum];
+
+    if (dbnum == 0) {
+        dest_data->size = src_data->size;
+        dest_data->data = src_data->data;
+    } else {
+        dest_data->size = 0;
+    }
+    
+    return 0;
+}
+
+static void
+verify_locked(DB_ENV *env, DB *db, int k) {
+    int r;
+    DB_TXN *txn = NULL;
+    r = env->txn_begin(env, NULL, &txn, 0); assert_zero(r);
+    DBT key; dbt_init(&key, &k, sizeof k);
+    r = db->del(db, txn, &key, DB_DELETE_ANY); assert(r == DB_LOCK_NOTGRANTED);
+    r = txn->abort(txn); assert_zero(r);
+}
+
+static void
+verify_seq(DB_ENV *env, DB *db, int dbnum, int ndbs, int nrows) {
+    int r;
+    DB_TXN *txn = NULL;
+    r = env->txn_begin(env, NULL, &txn, 0); assert_zero(r);
+
+    DBC *cursor = NULL;
+    r = db->cursor(db, txn, &cursor, 0); assert_zero(r);
+    int i;
+    for (i = 0; ; i++) {
+        DBT key; memset(&key, 0, sizeof key);
+        DBT val; memset(&val, 0, sizeof val);
+        r = cursor->c_get(cursor, &key, &val, DB_NEXT);
+        if (r != 0)
+            break;
+        int k;
+        assert(key.size == sizeof k);
+        memcpy(&k, key.data, key.size);
+        assert(k == get_key(i, dbnum));
+
+        if (dbnum == 0) {
+            assert(val.size == ndbs * sizeof (int));
+            int v[ndbs]; get_data(v, i, ndbs);
+            assert(memcmp(val.data, v, val.size) == 0);
+        } else
+            assert(val.size == 0);
+    }
+    assert(i == nrows);
+    r = cursor->c_close(cursor); assert_zero(r);
+    r = txn->commit(txn, 0); assert_zero(r);
+}
+
+static void
+verify(DB_ENV *env, DB *db[], int ndbs, int nrows) {
+    for (int dbnum = 0; dbnum < ndbs; dbnum++) 
+        verify_seq(env, db[dbnum], dbnum, ndbs, nrows);
+}
+
+static void
+populate(DB_ENV *env, DB *db[], int ndbs, int nrows) {
+    int r;
+    DB_TXN *txn = NULL;
+    r = env->txn_begin(env, NULL, &txn, 0); assert_zero(r);
+
+    // populate
+    for (int i = 0; i < nrows; i++) {
+        int k = get_key(i, 0);
+        int v[ndbs]; get_data(v, i, ndbs);
+        DBT pri_key; dbt_init(&pri_key, &k, sizeof k);
+        DBT pri_val; dbt_init(&pri_val, &v[0], sizeof v);
+        DBT keys[ndbs]; memset(keys, 0, sizeof keys);
+        DBT vals[ndbs]; memset(vals, 0, sizeof vals);
+        uint32_t flags[ndbs]; memset(flags, 0, sizeof flags);
+        r = env->put_multiple(env, NULL, txn, &pri_key, &pri_val, ndbs, db, keys, vals, flags, NULL); 
+        assert_zero(r);
+        for (int dbnum = 0; dbnum < ndbs; dbnum++)
+            verify_locked(env, db[dbnum], get_key(i, dbnum));
+    }
+
+    r = txn->commit(txn, 0); assert_zero(r);
+}
+
+static void
+run_test(int ndbs, int nrows) {
+    int r;
+    DB_ENV *env = NULL;
+    r = db_env_create(&env, 0); assert_zero(r);
+
+    r = env->set_generate_row_callback_for_put(env, put_callback); assert_zero(r);
+
+    r = env->open(env, ENVDIR, DB_INIT_MPOOL|DB_CREATE|DB_THREAD |DB_INIT_LOCK|DB_INIT_LOG|DB_INIT_TXN|DB_PRIVATE, S_IRWXU+S_IRWXG+S_IRWXO); assert_zero(r);
+
+    DB *db[ndbs];
+    for (int dbnum = 0; dbnum < ndbs; dbnum++) {
+        r = db_create(&db[dbnum], env, 0); assert_zero(r);
+
+        DBT dbt_dbnum; dbt_init(&dbt_dbnum, &dbnum, sizeof dbnum);
+        r = db[dbnum]->set_descriptor(db[dbnum], 1, &dbt_dbnum); 
+        assert_zero(r);
+
+        char dbname[32]; sprintf(dbname, "%d.tdb", dbnum);
+        r = db[dbnum]->open(db[dbnum], NULL, dbname, NULL, DB_BTREE, DB_AUTO_COMMIT+DB_CREATE, S_IRWXU+S_IRWXG+S_IRWXO); 
+        assert_zero(r);
+    }
+
+    populate(env, db, ndbs, nrows);
+
+    verify(env, db, ndbs, nrows);
+
+    for (int dbnum = 0; dbnum < ndbs; dbnum++) 
+        r = db[dbnum]->close(db[dbnum], 0); assert_zero(r);
+
+    r = env->close(env, 0); assert_zero(r);
+}
+
+int
+test_main(int argc, char * const argv[]) {
+    int r;
+    int ndbs = 2;
+    int nrows = 2;
+
+    // parse_args(argc, argv);
+    for (int i = 1; i < argc; i++) {
+        char * const arg = argv[i];
+        if (strcmp(arg, "-v") == 0) {
+            verbose++;
+            continue;
+        }
+        if (strcmp(arg, "-q") == 0) {
+            verbose = 0;
+            continue;
+        }
+        if (strcmp(arg, "--ndbs") == 0 && i+1 < argc) {
+            ndbs = atoi(argv[++i]);
+            continue;
+        }
+        if (strcmp(arg, "--nrows") == 0 && i+1 < argc) {
+            nrows = atoi(argv[++i]);
+            continue;
+        }
+    }
+
+    r = system("rm -rf " ENVDIR); assert_zero(r);
+    r = toku_os_mkdir(ENVDIR, S_IRWXU+S_IRWXG+S_IRWXO); assert_zero(r);
+
+    run_test(ndbs, nrows);
+
+    return 0;
+}
+

src/tests/recover-del-multiple-abort.c

+#include "test.h"
+
+// verify recovery of a delete multiple log entry
+
+static const int envflags = DB_INIT_MPOOL|DB_CREATE|DB_THREAD |DB_INIT_LOCK|DB_INIT_LOG|DB_INIT_TXN|DB_PRIVATE;
+
+static int
+get_key(int i, int dbnum) {
+    return htonl(i + dbnum);
+}
+
+static void
+get_data(int *v, int i, int ndbs) {
+    for (int dbnum = 0; dbnum < ndbs; dbnum++) {
+        v[dbnum] = get_key(i, dbnum);
+    }
+}
+
+static int
+put_callback(DB *dest_db, DB *src_db, DBT *dest_key, DBT *dest_data, const DBT *src_key, const DBT *src_data, void *extra) {
+    dest_db = dest_db; src_db = src_db; dest_key = dest_key; dest_data = dest_data; src_key = src_key; src_data = src_data;
+    assert(src_db == NULL);
+    assert(extra == NULL);
+
+    unsigned int dbnum;
+    assert(dest_db->descriptor->dbt.size == sizeof dbnum);
+    memcpy(&dbnum, dest_db->descriptor->dbt.data, sizeof dbnum);
+    assert(dbnum < src_data->size / sizeof (int));
+
+    int *pri_data = (int *) src_data->data;
+
+    switch (dest_key->flags) {
+    case 0:
+        dest_key->size = sizeof (int);
+        dest_key->data = &pri_data[dbnum];
+        break;
+    case DB_DBT_REALLOC:
+        dest_key->size = sizeof (int);
+        dest_key->data = toku_realloc(dest_key->data, dest_key->size);
+        memcpy(dest_key->data, &pri_data[dbnum], dest_key->size);
+        break;
+    default:
+        assert(0);
+    }
+
+    if (dest_data) {
+        switch (dest_data->flags) {
+        case 0:
+            if (dbnum == 0) {
+                dest_data->size = src_data->size;
+                dest_data->data = src_data->data;
+            } else {
+                dest_data->size = 0;
+            }
+            break;
+        case DB_DBT_REALLOC:
+            assert(0);
+        default:
+            assert(0);
+        }
+    }
+    
+    return 0;
+}
+
+static int
+del_callback(DB *dest_db, DB *src_db, DBT *dest_key, const DBT *src_key, const DBT *src_data, void *extra) {
+    return put_callback(dest_db, src_db, dest_key, NULL, src_key, src_data, extra);
+}
+
+static void
+run_test(int ndbs, int nrows) {
+    int r;
+    r = system("rm -rf " ENVDIR); assert_zero(r);
+    r = toku_os_mkdir(ENVDIR, S_IRWXU+S_IRWXG+S_IRWXO); assert_zero(r);
+
+    DB_ENV *env;
+    r = db_env_create(&env, 0);                                                         assert_zero(r);
+    r = env->set_generate_row_callback_for_put(env, put_callback);                      assert_zero(r);
+    r = env->set_generate_row_callback_for_del(env, del_callback);                      assert_zero(r);
+    r = env->open(env, ENVDIR, envflags, S_IRWXU+S_IRWXG+S_IRWXO);                      assert_zero(r);
+
+    DB *db[ndbs];
+    for (int dbnum = 0; dbnum < ndbs; dbnum++) {
+        r = db_create(&db[dbnum], env, 0);                                                        
+        assert_zero(r);
+        DBT dbt_dbnum; dbt_init(&dbt_dbnum, &dbnum, sizeof dbnum);
+        r = db[dbnum]->set_descriptor(db[dbnum], 1, &dbt_dbnum); 
+        assert_zero(r);
+        char dbname[32]; sprintf(dbname, "%d.tdb", dbnum);
+        r = db[dbnum]->open(db[dbnum], NULL, dbname, NULL, DB_BTREE, DB_AUTO_COMMIT|DB_CREATE, 0666);    
+        assert_zero(r);
+    }
+
+    DB_TXN *txn;
+    r = env->txn_begin(env, NULL, &txn, 0);                                             assert_zero(r);
+
+    for (int i = 0; i < nrows; i++) {
+        int k = get_key(i, 0);
+        int v[ndbs]; get_data(v, i, ndbs);
+        DBT pri_key; dbt_init(&pri_key, &k, sizeof k);
+        DBT pri_val; dbt_init(&pri_val, &v[0], sizeof v);
+        DBT keys[ndbs]; memset(keys, 0, sizeof keys);
+        DBT vals[ndbs]; memset(vals, 0, sizeof vals);
+        uint32_t flags[ndbs]; memset(flags, 0, sizeof flags);
+        r = env->put_multiple(env, NULL, txn, &pri_key, &pri_val, ndbs, db, keys, vals, flags, NULL); 
+        assert_zero(r);
+    }
+
+    r = txn->commit(txn, 0); assert_zero(r);
+
+    r = env->txn_checkpoint(env, 0, 0, 0);                                              assert_zero(r);
+
+    r = env->txn_begin(env, NULL, &txn, 0);                                             assert_zero(r);
+    
+    for (int i = 0; i < nrows; i++) {
+        int k = get_key(i, 0);
+        DBT pri_key; dbt_init(&pri_key, &k, sizeof k);
+        int v[ndbs]; get_data(v, i, ndbs);
+        DBT pri_data; dbt_init(&pri_data, &v[0], sizeof v);
+        DBT keys[ndbs]; memset(keys, 0, sizeof keys);
+        uint32_t flags[ndbs]; memset(flags, 0, sizeof flags);
+        r = env->del_multiple(env, NULL, txn, &pri_key, &pri_data, ndbs, db, keys, flags, NULL);
+        assert_zero(r);
+    }
+
+    toku_hard_crash_on_purpose();
+}
+
+static void
+verify_seq(DB_ENV *env, DB *db, int dbnum, int ndbs, int nrows) {
+    int r;
+    DB_TXN *txn = NULL;
+    r = env->txn_begin(env, NULL, &txn, 0); assert_zero(r);
+
+    DBC *cursor = NULL;
+    r = db->cursor(db, txn, &cursor, 0); assert_zero(r);
+    int i;
+    for (i = 0; ; i++) {
+        DBT key; memset(&key, 0, sizeof key);
+        DBT val; memset(&val, 0, sizeof val);
+        r = cursor->c_get(cursor, &key, &val, DB_NEXT);
+        if (r != 0)
+            break;
+        int k;
+        assert(key.size == sizeof k);
+        memcpy(&k, key.data, key.size);
+        assert(k == get_key(i, dbnum));
+
+        if (dbnum == 0) {
+            assert(val.size == ndbs * sizeof (int));
+            int v[ndbs]; get_data(v, i, ndbs);
+            assert(memcmp(val.data, v, val.size) == 0);
+        } else
+            assert(val.size == 0);
+    }
+    assert(i == nrows);
+    r = cursor->c_close(cursor); assert_zero(r);
+    r = txn->commit(txn, 0); assert_zero(r);
+}
+
+static void
+verify_all(DB_ENV *env, int ndbs, int nrows) {
+    int r;
+    for (int dbnum = 0; dbnum < ndbs; dbnum++) {
+        DB *db = NULL;
+        r = db_create(&db, env, 0);                                                        
+        assert_zero(r);
+        char dbname[32]; sprintf(dbname, "%d.tdb", dbnum);
+        r = db->open(db, NULL, dbname, NULL, DB_BTREE, DB_AUTO_COMMIT|DB_CREATE, 0666);    
+        assert_zero(r);
+        verify_seq(env, db, dbnum, ndbs, nrows);
+        r = db->close(db, 0); 
+        assert_zero(r);
+    }
+}
+
+static void
+run_recover(int ndbs, int nrows) {
+    int r;
+
+    DB_ENV *env;
+    r = db_env_create(&env, 0);                                                             assert_zero(r);
+    r = env->set_generate_row_callback_for_put(env, put_callback);                          assert_zero(r);
+    r = env->set_generate_row_callback_for_del(env, del_callback);                          assert_zero(r);
+    r = env->open(env, ENVDIR, envflags|DB_RECOVER, S_IRWXU+S_IRWXG+S_IRWXO);               assert_zero(r);
+    verify_all(env, ndbs, nrows);
+    r = env->close(env, 0);                                                                 assert_zero(r);
+}
+
+static int
+usage(void) {
+    return 1;
+}
+
+int
+test_main (int argc, char * const argv[]) {
+    BOOL do_test = FALSE;
+    BOOL do_recover = FALSE;
+    int ndbs = 2;
+    int nrows = 1;
+
+    for (int i = 1; i < argc; i++) {
+        char * const arg = argv[i];
+        if (strcmp(arg, "-v") == 0) {
+            verbose++;
+            continue;
+        }
+        if (strcmp(arg, "-q") == 0) {
+            verbose--;
+            if (verbose < 0)
+                verbose = 0;
+            continue;
+        }
+        if (strcmp(arg, "--test") == 0) {
+            do_test = TRUE;
+            continue;
+        }
+        if (strcmp(arg, "--recover") == 0) {
+            do_recover = TRUE;
+            continue;
+        }
+        if (strcmp(arg, "--ndbs") == 0 && i+1 < argc) {
+            ndbs = atoi(argv[++i]);
+            continue;
+        }
+        if (strcmp(arg, "--nrows") == 0 && i+1 < argc) {
+            nrows = atoi(argv[++i]);
+            continue;
+        }
+        if (strcmp(arg, "--help") == 0) {
+            return usage();
+        }
+    }
+    
+    if (do_test)
+        run_test(ndbs, nrows);
+    if (do_recover)
+        run_recover(ndbs, nrows);
+
+    return 0;
+}

src/tests/recover-del-multiple.c

+#include "test.h"
+
+// verify recovery of a delete multiple log entry
+
+static const int envflags = DB_INIT_MPOOL|DB_CREATE|DB_THREAD |DB_INIT_LOCK|DB_INIT_LOG|DB_INIT_TXN|DB_PRIVATE;
+
+static int
+get_key(int i, int dbnum) {
+    return htonl(i + dbnum);
+}
+
+static void
+get_data(int *v, int i, int ndbs) {
+    for (int dbnum = 0; dbnum < ndbs; dbnum++) {
+        v[dbnum] = get_key(i, dbnum);
+    }
+}
+
+static int
+put_callback(DB *dest_db, DB *src_db, DBT *dest_key, DBT *dest_data, const DBT *src_key, const DBT *src_data, void *extra) {
+    dest_db = dest_db; src_db = src_db; dest_key = dest_key; dest_data = dest_data; src_key = src_key; src_data = src_data;
+    assert(src_db == NULL);
+    assert(extra == NULL);
+
+    unsigned int dbnum;
+    assert(dest_db->descriptor->dbt.size == sizeof dbnum);
+    memcpy(&dbnum, dest_db->descriptor->dbt.data, sizeof dbnum);
+    assert(dbnum < src_data->size / sizeof (int));
+
+    int *pri_data = (int *) src_data->data;
+
+    switch (dest_key->flags) {
+    case 0:
+        dest_key->size = sizeof (int);
+        dest_key->data = &pri_data[dbnum];
+        break;
+    case DB_DBT_REALLOC:
+        dest_key->size = sizeof (int);
+        dest_key->data = toku_realloc(dest_key->data, dest_key->size);
+        memcpy(dest_key->data, &pri_data[dbnum], dest_key->size);
+        break;
+    default:
+        assert(0);
+    }
+
+    if (dest_data) {
+        switch (dest_data->flags) {
+        case 0:
+            if (dbnum == 0) {
+                dest_data->size = src_data->size;
+                dest_data->data = src_data->data;
+            } else {
+                dest_data->size = 0;
+            }
+            break;
+        case DB_DBT_REALLOC:
+            assert(0);
+        default:
+            assert(0);
+        }
+    }
+    
+    return 0;
+}
+
+static int
+del_callback(DB *dest_db, DB *src_db, DBT *dest_key, const DBT *src_key, const DBT *src_data, void *extra) {
+    return put_callback(dest_db, src_db, dest_key, NULL, src_key, src_data, extra);
+}
+
+static void
+run_test(int ndbs, int nrows) {
+    int r;
+    r = system("rm -rf " ENVDIR); assert_zero(r);
+    r = toku_os_mkdir(ENVDIR, S_IRWXU+S_IRWXG+S_IRWXO); assert_zero(r);
+
+    DB_ENV *env;
+    r = db_env_create(&env, 0);                                                         assert_zero(r);
+    r = env->set_generate_row_callback_for_put(env, put_callback);                      assert_zero(r);
+    r = env->set_generate_row_callback_for_del(env, del_callback);                      assert_zero(r);
+    r = env->open(env, ENVDIR, envflags, S_IRWXU+S_IRWXG+S_IRWXO);                      assert_zero(r);
+
+    DB *db[ndbs];
+    for (int dbnum = 0; dbnum < ndbs; dbnum++) {
+        r = db_create(&db[dbnum], env, 0);                                                        
+        assert_zero(r);
+        DBT dbt_dbnum; dbt_init(&dbt_dbnum, &dbnum, sizeof dbnum);
+        r = db[dbnum]->set_descriptor(db[dbnum], 1, &dbt_dbnum); 
+        assert_zero(r);
+        char dbname[32]; sprintf(dbname, "%d.tdb", dbnum);
+        r = db[dbnum]->open(db[dbnum], NULL, dbname, NULL, DB_BTREE, DB_AUTO_COMMIT|DB_CREATE, 0666);    
+        assert_zero(r);
+    }
+
+    DB_TXN *txn;
+    r = env->txn_begin(env, NULL, &txn, 0);                                             assert_zero(r);
+
+    for (int i = 0; i < nrows; i++) {
+        int k = get_key(i, 0);
+        int v[ndbs]; get_data(v, i, ndbs);
+        DBT pri_key; dbt_init(&pri_key, &k, sizeof k);
+        DBT pri_val; dbt_init(&pri_val, &v[0], sizeof v);
+        DBT keys[ndbs]; memset(keys, 0, sizeof keys);
+        DBT vals[ndbs]; memset(vals, 0, sizeof vals);
+        uint32_t flags[ndbs]; memset(flags, 0, sizeof flags);
+        r = env->put_multiple(env, NULL, txn, &pri_key, &pri_val, ndbs, db, keys, vals, flags, NULL); 
+        assert_zero(r);
+    }
+
+    r = txn->commit(txn, 0); assert_zero(r);
+
+    r = env->txn_checkpoint(env, 0, 0, 0);                                              assert_zero(r);
+
+    r = env->txn_begin(env, NULL, &txn, 0);                                             assert_zero(r);
+    
+    for (int i = 0; i < nrows; i++) {
+        int k = get_key(i, 0);
+        DBT pri_key; dbt_init(&pri_key, &k, sizeof k);
+        int v[ndbs]; get_data(v, i, ndbs);
+        DBT pri_data; dbt_init(&pri_data, &v[0], sizeof v);
+        DBT keys[ndbs]; memset(keys, 0, sizeof keys);
+        uint32_t flags[ndbs]; memset(flags, 0, sizeof flags);
+        r = env->del_multiple(env, NULL, txn, &pri_key, &pri_data, ndbs, db, keys, flags, NULL);
+        assert_zero(r);
+    }
+
+    r = txn->commit(txn, 0);                                                        assert_zero(r);
+
+    toku_hard_crash_on_purpose();
+}
+
+
+static void
+verify_empty(DB_ENV *env, DB *db) {
+    int r;
+    DB_TXN *txn = NULL;
+    r = env->txn_begin(env, NULL, &txn, 0); assert_zero(r);
+
+    DBC *cursor = NULL;
+    r = db->cursor(db, txn, &cursor, 0); assert_zero(r);
+
+    DBT key; memset(&key, 0, sizeof key);
+    DBT val; memset(&val, 0, sizeof val);
+    r = cursor->c_get(cursor, &key, &val, DB_NEXT);
+    assert(r == DB_NOTFOUND);
+
+    r = cursor->c_close(cursor); assert_zero(r);
+    r = txn->commit(txn, 0); assert_zero(r);
+}
+
+static void
+verify_all(DB_ENV *env, int ndbs) {
+    int r;
+    for (int dbnum = 0; dbnum < ndbs; dbnum++) {
+        DB *db = NULL;
+        r = db_create(&db, env, 0);                                                        
+        assert_zero(r);
+        char dbname[32]; sprintf(dbname, "%d.tdb", dbnum);
+        r = db->open(db, NULL, dbname, NULL, DB_BTREE, DB_AUTO_COMMIT|DB_CREATE, 0666);    
+        assert_zero(r);
+        verify_empty(env, db);
+        r = db->close(db, 0); 
+        assert_zero(r);
+    }
+}
+
+static void
+run_recover(int ndbs, int UU(nrows)) {
+    int r;
+
+    DB_ENV *env;
+    r = db_env_create(&env, 0);                                                             assert_zero(r);
+    r = env->set_generate_row_callback_for_put(env, put_callback);                          assert_zero(r);
+    r = env->set_generate_row_callback_for_del(env, del_callback);                          assert_zero(r);
+    r = env->open(env, ENVDIR, envflags|DB_RECOVER, S_IRWXU+S_IRWXG+S_IRWXO);               assert_zero(r);
+    verify_all(env, ndbs);
+    r = env->close(env, 0);                                                                 assert_zero(r);
+}
+
+static int
+usage(void) {
+    return 1;
+}
+
+int
+test_main (int argc, char * const argv[]) {
+    BOOL do_test = FALSE;
+    BOOL do_recover = FALSE;
+    int ndbs = 2;
+    int nrows = 1;
+
+    for (int i = 1; i < argc; i++) {
+        char * const arg = argv[i];
+        if (strcmp(arg, "-v") == 0) {
+            verbose++;
+            continue;
+        }
+        if (strcmp(arg, "-q") == 0) {
+            verbose--;
+            if (verbose < 0)
+                verbose = 0;
+            continue;
+        }
+        if (strcmp(arg, "--test") == 0) {
+            do_test = TRUE;
+            continue;
+        }
+        if (strcmp(arg, "--recover") == 0) {
+            do_recover = TRUE;
+            continue;
+        }
+        if (strcmp(arg, "--ndbs") == 0 && i+1 < argc) {
+            ndbs = atoi(argv[++i]);
+            continue;
+        }
+        if (strcmp(arg, "--nrows") == 0 && i+1 < argc) {
+            nrows = atoi(argv[++i]);
+            continue;
+        }
+        if (strcmp(arg, "--help") == 0) {
+            return usage();
+        }
+    }
+    
+    if (do_test)
+        run_test(ndbs, nrows);
+    if (do_recover)
+        run_recover(ndbs, nrows);
+
+    return 0;
+}

src/tests/recover-put-multiple-abort.c

+#include "test.h"
+
+// verify recovery of a put multiple log entry
+
+static const int envflags = DB_INIT_MPOOL|DB_CREATE|DB_THREAD |DB_INIT_LOCK|DB_INIT_LOG|DB_INIT_TXN|DB_PRIVATE;
+
+static int
+get_key(int i, int dbnum) {
+    return htonl(i + dbnum);
+}
+
+static void
+get_data(int *v, int i, int ndbs) {
+    for (int dbnum = 0; dbnum < ndbs; dbnum++) {
+        v[dbnum] = get_key(i, dbnum);
+    }
+}
+
+static int
+put_callback(DB *dest_db, DB *src_db, DBT *dest_key, DBT *dest_data, const DBT *src_key, const DBT *src_data, void *extra) {
+    dest_db = dest_db; src_db = src_db; dest_key = dest_key; dest_data = dest_data; src_key = src_key; src_data = src_data;
+    assert(src_db == NULL);
+    assert(extra == NULL);
+
+    unsigned int dbnum;
+    assert(dest_db->descriptor->dbt.size == sizeof dbnum);
+    memcpy(&dbnum, dest_db->descriptor->dbt.data, sizeof dbnum);
+    assert(dbnum < src_data->size / sizeof (int));
+
+    int *pri_data = (int *) src_data->data;
+
+    switch (dest_key->flags) {
+    case 0:
+        dest_key->size = sizeof (int);
+        dest_key->data = &pri_data[dbnum];
+        break;
+    case DB_DBT_REALLOC:
+        dest_key->size = sizeof (int);
+        dest_key->data = toku_realloc(dest_key->data, dest_key->size);
+        memcpy(dest_key->data, &pri_data[dbnum], dest_key->size);
+        break;
+    default:
+        assert(0);
+    }
+
+    if (dest_data) {
+        switch (dest_data->flags) {
+        case 0:
+            if (dbnum == 0) {
+                dest_data->size = src_data->size;
+                dest_data->data = src_data->data;
+            } else {
+                dest_data->size = 0;
+            }
+            break;
+        case DB_DBT_REALLOC:
+            assert(0);
+        default:
+            assert(0);
+        }
+    }
+    
+    return 0;
+}
+
+static int
+del_callback(DB *dest_db, DB *src_db, DBT *dest_key, const DBT *src_key, const DBT *src_data, void *extra) {
+    return put_callback(dest_db, src_db, dest_key, NULL, src_key, src_data, extra);
+}
+
+static void
+run_test(int ndbs, int nrows) {
+    int r;
+    r = system("rm -rf " ENVDIR); assert_zero(r);
+    r = toku_os_mkdir(ENVDIR, S_IRWXU+S_IRWXG+S_IRWXO); assert_zero(r);
+
+    DB_ENV *env;
+    r = db_env_create(&env, 0);                                                         assert_zero(r);
+    r = env->set_generate_row_callback_for_put(env, put_callback);                      assert_zero(r);
+    r = env->set_generate_row_callback_for_del(env, del_callback);                      assert_zero(r);
+    r = env->open(env, ENVDIR, envflags, S_IRWXU+S_IRWXG+S_IRWXO);                      assert_zero(r);
+
+    DB *db[ndbs];
+    for (int dbnum = 0; dbnum < ndbs; dbnum++) {
+        r = db_create(&db[dbnum], env, 0);                                                        
+        assert_zero(r);
+        DBT dbt_dbnum; dbt_init(&dbt_dbnum, &dbnum, sizeof dbnum);
+        r = db[dbnum]->set_descriptor(db[dbnum], 1, &dbt_dbnum); 
+        assert_zero(r);
+        char dbname[32]; sprintf(dbname, "%d.tdb", dbnum);
+        r = db[dbnum]->open(db[dbnum], NULL, dbname, NULL, DB_BTREE, DB_AUTO_COMMIT|DB_CREATE, 0666);    
+        assert_zero(r);
+    }
+
+    r = env->txn_checkpoint(env, 0, 0, 0);                                              assert_zero(r);
+
+    DB_TXN *txn;
+    r = env->txn_begin(env, NULL, &txn, 0);                                             assert_zero(r);
+
+    for (int i = 0; i < nrows; i++) {
+        int k = get_key(i, 0);
+        int v[ndbs]; get_data(v, i, ndbs);
+        DBT pri_key; dbt_init(&pri_key, &k, sizeof k);
+        DBT pri_val; dbt_init(&pri_val, &v[0], sizeof v);
+        DBT keys[ndbs]; memset(keys, 0, sizeof keys);
+        DBT vals[ndbs]; memset(vals, 0, sizeof vals);
+        uint32_t flags[ndbs]; memset(flags, 0, sizeof flags);
+        r = env->put_multiple(env, NULL, txn, &pri_key, &pri_val, ndbs, db, keys, vals, flags, NULL); 
+        assert_zero(r);
+    }
+
+    toku_hard_crash_on_purpose();
+}
+
+
+static void
+verify_empty(DB_ENV *env, DB *db) {
+    int r;
+    DB_TXN *txn = NULL;
+    r = env->txn_begin(env, NULL, &txn, 0); assert_zero(r);
+
+    DBC *cursor = NULL;
+    r = db->cursor(db, txn, &cursor, 0); assert_zero(r);
+
+    DBT key; memset(&key, 0, sizeof key);
+    DBT val; memset(&val, 0, sizeof val);
+    r = cursor->c_get(cursor, &key, &val, DB_NEXT);
+    assert(r == DB_NOTFOUND);
+
+    r = cursor->c_close(cursor); assert_zero(r);
+    r = txn->commit(txn, 0); assert_zero(r);
+}
+
+static void
+verify_all(DB_ENV *env, int ndbs) {
+    int r;
+    for (int dbnum = 0; dbnum < ndbs; dbnum++) {
+        DB *db = NULL;
+        r = db_create(&db, env, 0);                                                        
+        assert_zero(r);
+        char dbname[32]; sprintf(dbname, "%d.tdb", dbnum);
+        r = db->open(db, NULL, dbname, NULL, DB_BTREE, DB_AUTO_COMMIT|DB_CREATE, 0666);    
+        assert_zero(r);
+        verify_empty(env, db);
+        r = db->close(db, 0); 
+        assert_zero(r);
+    }
+}
+
+static void
+run_recover(int ndbs, int UU(nrows)) {
+    int r;
+
+    DB_ENV *env;
+    r = db_env_create(&env, 0);                                                             assert_zero(r);
+    r = env->set_generate_row_callback_for_put(env, put_callback);                          assert_zero(r);
+    r = env->set_generate_row_callback_for_del(env, del_callback);                          assert_zero(r);
+    r = env->open(env, ENVDIR, envflags|DB_RECOVER, S_IRWXU+S_IRWXG+S_IRWXO);               assert_zero(r);
+    verify_all(env, ndbs);
+    r = env->close(env, 0);                                                                 assert_zero(r);
+}
+
+static int
+usage(void) {
+    return 1;
+}
+
+int
+test_main (int argc, char * const argv[]) {
+    BOOL do_test = FALSE;
+    BOOL do_recover = FALSE;
+    int ndbs = 2;
+    int nrows = 1;
+
+    for (int i = 1; i < argc; i++) {
+        char * const arg = argv[i];
+        if (strcmp(arg, "-v") == 0) {
+            verbose++;
+            continue;
+        }
+        if (strcmp(arg, "-q") == 0) {
+            verbose--;
+            if (verbose < 0)
+                verbose = 0;
+            continue;
+        }
+        if (strcmp(arg, "--test") == 0) {
+            do_test = TRUE;
+            continue;
+        }
+        if (strcmp(arg, "--recover") == 0) {
+            do_recover = TRUE;
+            continue;
+        }
+        if (strcmp(arg, "--ndbs") == 0 && i+1 < argc) {
+            ndbs = atoi(argv[++i]);
+            continue;
+        }
+        if (strcmp(arg, "--nrows") == 0 && i+1 < argc) {
+            nrows = atoi(argv[++i]);
+            continue;
+        }
+        if (strcmp(arg, "--help") == 0) {
+            return usage();
+        }
+    }
+    
+    if (do_test)
+        run_test(ndbs, nrows);
+    if (do_recover)
+        run_recover(ndbs, nrows);
+
+    return 0;
+}

src/tests/recover-put-multiple.c

+#include "test.h"
+
+// verify recovery of a put multiple log entry
+
+static const int envflags = DB_INIT_MPOOL|DB_CREATE|DB_THREAD |DB_INIT_LOCK|DB_INIT_LOG|DB_INIT_TXN|DB_PRIVATE;
+
+static int
+get_key(int i, int dbnum) {
+    return htonl(i + dbnum);
+}
+
+static void
+get_data(int *v, int i, int ndbs) {
+    for (int dbnum = 0; dbnum < ndbs; dbnum++) {
+        v[dbnum] = get_key(i, dbnum);
+    }
+}
+
+static int
+put_callback(DB *dest_db, DB *src_db, DBT *dest_key, DBT *dest_data, const DBT *src_key, const DBT *src_data, void *extra) {
+    dest_db = dest_db; src_db = src_db; dest_key = dest_key; dest_data = dest_data; src_key = src_key; src_data = src_data;
+    assert(src_db == NULL);
+    assert(extra == NULL);
+
+    unsigned int dbnum;
+    assert(dest_db->descriptor->dbt.size == sizeof dbnum);
+    memcpy(&dbnum, dest_db->descriptor->dbt.data, sizeof dbnum);
+    assert(dbnum < src_data->size / sizeof (int));
+
+    int *pri_data = (int *) src_data->data;
+
+    switch (dest_key->flags) {
+    case 0:
+        dest_key->size = sizeof (int);
+        dest_key->data = &pri_data[dbnum];
+        break;
+    case DB_DBT_REALLOC:
+        dest_key->size = sizeof (int);
+        dest_key->data = toku_realloc(dest_key->data, dest_key->size);
+        memcpy(dest_key->data, &pri_data[dbnum], dest_key->size);
+        break;
+    default:
+        assert(0);
+    }
+
+    if (dest_data) {
+        switch (dest_data->flags) {
+        case 0:
+            if (dbnum == 0) {
+                dest_data->size = src_data->size;
+                dest_data->data = src_data->data;
+            } else
+                dest_data->size = 0;
+            break;
+        case DB_DBT_REALLOC:
+            if (dbnum == 0) {
+                dest_data->size = src_data->size;
+                dest_data->data = toku_realloc(dest_data->data, dest_data->size);
+                memcpy(dest_data->data, src_data->data, dest_data->size);
+            } else
+                dest_data->size = 0;
+            break;
+        default:
+            assert(0);
+        }
+    }
+    
+    return 0;
+}
+
+static int
+del_callback(DB *dest_db, DB *src_db, DBT *dest_key, const DBT *src_key, const DBT *src_data, void *extra) {
+    return put_callback(dest_db, src_db, dest_key, NULL, src_key, src_data, extra);
+}
+
+static void
+run_test(int ndbs, int nrows) {
+    int r;
+    r = system("rm -rf " ENVDIR); assert_zero(r);
+    r = toku_os_mkdir(ENVDIR, S_IRWXU+S_IRWXG+S_IRWXO); assert_zero(r);
+
+    DB_ENV *env;
+    r = db_env_create(&env, 0);                                                         assert_zero(r);
+    r = env->set_generate_row_callback_for_put(env, put_callback);                      assert_zero(r);
+    r = env->set_generate_row_callback_for_del(env, del_callback);                      assert_zero(r);
+    r = env->open(env, ENVDIR, envflags, S_IRWXU+S_IRWXG+S_IRWXO);                      assert_zero(r);
+
+    DB *db[ndbs];
+    for (int dbnum = 0; dbnum < ndbs; dbnum++) {
+        r = db_create(&db[dbnum], env, 0);                                                        
+        assert_zero(r);
+        DBT dbt_dbnum; dbt_init(&dbt_dbnum, &dbnum, sizeof dbnum);
+        r = db[dbnum]->set_descriptor(db[dbnum], 1, &dbt_dbnum); 
+        assert_zero(r);
+        char dbname[32]; sprintf(dbname, "%d.tdb", dbnum);
+        r = db[dbnum]->open(db[dbnum], NULL, dbname, NULL, DB_BTREE, DB_AUTO_COMMIT|DB_CREATE, 0666);    
+        assert_zero(r);
+    }
+
+    r = env->txn_checkpoint(env, 0, 0, 0);                                              assert_zero(r);
+
+    DB_TXN *txn;
+    r = env->txn_begin(env, NULL, &txn, 0);                                             assert_zero(r);
+
+    for (int i = 0; i < nrows; i++) {
+        int k = get_key(i, 0);
+        int v[ndbs]; get_data(v, i, ndbs);
+        DBT pri_key; dbt_init(&pri_key, &k, sizeof k);
+        DBT pri_val; dbt_init(&pri_val, &v[0], sizeof v);
+        DBT keys[ndbs]; memset(keys, 0, sizeof keys);
+        DBT vals[ndbs]; memset(vals, 0, sizeof vals);
+        uint32_t flags[ndbs]; memset(flags, 0, sizeof flags);
+        r = env->put_multiple(env, NULL, txn, &pri_key, &pri_val, ndbs, db, keys, vals, flags, NULL); 
+        assert_zero(r);
+    }
+
+    r = txn->commit(txn, 0);                                                        assert_zero(r);
+
+    toku_hard_crash_on_purpose();
+}
+
+static void
+verify_seq(DB_ENV *env, DB *db, int dbnum, int ndbs, int nrows) {
+    int r;
+    DB_TXN *txn = NULL;
+    r = env->txn_begin(env, NULL, &txn, 0); assert_zero(r);
+
+    DBC *cursor = NULL;
+    r = db->cursor(db, txn, &cursor, 0); assert_zero(r);
+    int i;
+    for (i = 0; ; i++) {
+        DBT key; memset(&key, 0, sizeof key);
+        DBT val; memset(&val, 0, sizeof val);
+        r = cursor->c_get(cursor, &key, &val, DB_NEXT);
+        if (r != 0)
+            break;
+        int k;
+        assert(key.size == sizeof k);
+        memcpy(&k, key.data, key.size);
+        assert(k == get_key(i, dbnum));
+
+        if (dbnum == 0) {
+            assert(val.size == ndbs * sizeof (int));
+            int v[ndbs]; get_data(v, i, ndbs);
+            assert(memcmp(val.data, v, val.size) == 0);
+        } else
+            assert(val.size == 0);
+    }
+    assert(i == nrows);
+    r = cursor->c_close(cursor); assert_zero(r);
+    r = txn->commit(txn, 0); assert_zero(r);
+}
+
+static void
+verify_all(DB_ENV *env, int ndbs, int nrows) {
+    int r;
+    for (int dbnum = 0; dbnum < ndbs; dbnum++) {
+        DB *db = NULL;
+        r = db_create(&db, env, 0);                                                        
+        assert_zero(r);
+        char dbname[32]; sprintf(dbname, "%d.tdb", dbnum);
+        r = db->open(db, NULL, dbname, NULL, DB_BTREE, DB_AUTO_COMMIT|DB_CREATE, 0666);    
+        assert_zero(r);
+        verify_seq(env, db, dbnum, ndbs, nrows);
+        r = db->close(db, 0); 
+        assert_zero(r);
+    }
+}
+
+static void
+run_recover(int ndbs, int nrows) {
+    int r;
+
+    DB_ENV *env;
+    r = db_env_create(&env, 0);                                                             assert_zero(r);
+    r = env->set_generate_row_callback_for_put(env, put_callback);                          assert_zero(r);
+    r = env->set_generate_row_callback_for_del(env, del_callback);                          assert_zero(r);
+    r = env->open(env, ENVDIR, envflags|DB_RECOVER, S_IRWXU+S_IRWXG+S_IRWXO);               assert_zero(r);
+    verify_all(env, ndbs, nrows);
+    r = env->close(env, 0);                                                                 assert_zero(r);
+}
+
+static int
+usage(void) {
+    return 1;
+}
+
+int
+test_main (int argc, char * const argv[]) {
+    BOOL do_test = FALSE;
+    BOOL do_recover = FALSE;
+    int ndbs = 2;
+    int nrows = 1;
+
+    for (int i = 1; i < argc; i++) {
+        char * const arg = argv[i];
+        if (strcmp(arg, "-v") == 0) {
+            verbose++;
+            continue;
+        }
+        if (strcmp(arg, "-q") == 0) {
+            verbose--;
+            if (verbose < 0)
+                verbose = 0;
+            continue;
+        }
+        if (strcmp(arg, "--test") == 0) {
+            do_test = TRUE;
+            continue;
+        }
+        if (strcmp(arg, "--recover") == 0) {
+            do_recover = TRUE;
+            continue;
+        }
+        if (strcmp(arg, "--ndbs") == 0 && i+1 < argc) {
+            ndbs = atoi(argv[++i]);
+            continue;
+        }
+        if (strcmp(arg, "--nrows") == 0 && i+1 < argc) {
+            nrows = atoi(argv[++i]);
+            continue;
+        }
+        if (strcmp(arg, "--help") == 0) {
+            return usage();
+        }
+    }
+    
+    if (do_test)
+        run_test(ndbs, nrows);
+    if (do_recover)
+        run_recover(ndbs, nrows);
+
+    return 0;
+}

src/tests/recover-update-multiple-abort.c

+#include "test.h"
+
+// verify recovery of some update multiple operations
+
+static const int envflags = DB_INIT_MPOOL|DB_CREATE|DB_THREAD |DB_INIT_LOCK|DB_INIT_LOG|DB_INIT_TXN|DB_PRIVATE;
+
+static int
+get_key(int i, int dbnum) {
+    return htonl(2*(i + dbnum));
+}
+
+static int
+get_new_key(int i, int dbnum) {
+    return htonl(2*(i + dbnum) + 1);
+}
+
+static void
+get_data(int *v, int i, int ndbs) {
+    for (int dbnum = 0; dbnum < ndbs; dbnum++) {
+        v[dbnum] = get_key(i, dbnum);
+    }
+}
+
+static void
+get_new_data(int *v, int i, int ndbs) {
+    for (int dbnum = 0; dbnum < ndbs; dbnum++) {
+        if ((i % ndbs) == dbnum)
+            v[dbnum] = get_new_key(i, dbnum);
+        else
+            v[dbnum] = get_key(i, dbnum);
+    }
+}
+
+static int
+put_callback(DB *dest_db, DB *src_db, DBT *dest_key, DBT *dest_data, const DBT *src_key, const DBT *src_data, void *extra) {
+    dest_db = dest_db; src_db = src_db; dest_key = dest_key; dest_data = dest_data; src_key = src_key; src_data = src_data;
+    assert(src_db == NULL);
+    assert(extra == NULL);
+
+    unsigned int dbnum;
+    assert(dest_db->descriptor->dbt.size == sizeof dbnum);
+    memcpy(&dbnum, dest_db->descriptor->dbt.data, sizeof dbnum);
+    assert(dbnum < src_data->size / sizeof (int));
+
+    int *pri_key = (int *) src_key->data;
+    int *pri_data = (int *) src_data->data;
+
+    switch (dest_key->flags) {
+    case 0:
+        dest_key->size = sizeof (int);
+        dest_key->data = dbnum == 0 ? &pri_key[dbnum] : &pri_data[dbnum];
+        break;
+    case DB_DBT_REALLOC:
+        dest_key->size = sizeof (int);
+        dest_key->data = toku_realloc(dest_key->data, dest_key->size);
+        memcpy(dest_key->data, dbnum == 0 ? &pri_key[dbnum] : &pri_data[dbnum], dest_key->size);
+        break;
+    default:
+        assert(0);
+    }
+
+    if (dest_data) {
+        switch (dest_data->flags) {
+        case 0:
+            if (dbnum == 0) {
+                dest_data->size = src_data->size;
+                dest_data->data = src_data->data;
+            } else
+                dest_data->size = 0;
+            break;
+        case DB_DBT_REALLOC:
+            if (dbnum == 0) {
+                dest_data->size = src_data->size;
+                dest_data->data = toku_realloc(dest_data->data, dest_data->size);
+                memcpy(dest_data->data, src_data->data, dest_data->size);
+            } else
+                dest_data->size = 0;
+            break;
+        default:
+            assert(0);
+        }
+    }
+    
+    return 0;
+}
+
+static int
+del_callback(DB *dest_db, DB *src_db, DBT *dest_key, const DBT *src_key, const DBT *src_data, void *extra) {
+    return put_callback(dest_db, src_db, dest_key, NULL, src_key, src_data, extra);
+}
+
+static void
+update_diagonal(DB_ENV *env, DB_TXN *txn, DB *db[], int ndbs, int nrows) {
+    assert(ndbs > 0);
+    int r;
+    for (int i = 0; i < nrows; i++) {
+
+        // update the data i % ndbs col from x to x+1
+
+        int k = get_key(i, 0);
+        DBT old_key; dbt_init(&old_key, &k, sizeof k);
+        DBT new_key = old_key;
+
+        int v[ndbs]; get_data(v, i, ndbs);
+        DBT old_data; dbt_init(&old_data, &v[0], sizeof v);
+        
+        int newv[ndbs]; get_new_data(newv, i, ndbs);
+        DBT new_data; dbt_init(&new_data, &newv[0], sizeof newv);
+  
+        int ndbts = 2 * ndbs;
+        DBT keys[ndbts]; memset(keys, 0, sizeof keys);
+        DBT vals[ndbts]; memset(vals, 0, sizeof vals);
+
+        r = env->update_multiple(env, NULL, txn, &old_key, &old_data, &new_key, &new_data, ndbs, db, ndbts, keys, vals, NULL);
+        assert_zero(r);
+    }
+}
+
+static void
+populate_primary(DB_ENV *env, DB *db, int ndbs, int nrows) {
+    int r;
+    DB_TXN *txn = NULL;
+    r = env->txn_begin(env, NULL, &txn, 0); assert_zero(r);
+
+    // populate
+    for (int i = 0; i < nrows; i++) {
+        int k = get_key(i, 0);
+        int v[ndbs]; get_data(v, i, ndbs);
+        DBT key; dbt_init(&key, &k, sizeof k);
+        DBT val; dbt_init(&val, &v[0], sizeof v);
+        r = db->put(db, txn, &key, &val, DB_YESOVERWRITE); assert_zero(r);
+    }
+
+    r = txn->commit(txn, 0); assert_zero(r);
+}
+
+static void
+populate_secondary(DB_ENV *env, DB *db, int dbnum, int nrows) {
+    int r;
+    DB_TXN *txn = NULL;
+    r = env->txn_begin(env, NULL, &txn, 0); assert_zero(r);
+
+    // populate
+    for (int i = 0; i < nrows; i++) {
+        int k = get_key(i, dbnum);
+        DBT key; dbt_init(&key, &k, sizeof k);
+        DBT val; dbt_init(&val, NULL, 0);
+        r = db->put(db, txn, &key, &val, DB_YESOVERWRITE); assert_zero(r);
+    }
+
+    r = txn->commit(txn, 0); assert_zero(r);
+}
+
+
+static void
+run_test(int ndbs, int nrows) {
+    int r;
+    r = system("rm -rf " ENVDIR); assert_zero(r);
+    r = toku_os_mkdir(ENVDIR, S_IRWXU+S_IRWXG+S_IRWXO); assert_zero(r);
+
+    DB_ENV *env;
+    r = db_env_create(&env, 0);                                                         assert_zero(r);
+    r = env->set_generate_row_callback_for_put(env, put_callback);                      assert_zero(r);
+    r = env->set_generate_row_callback_for_del(env, del_callback);                      assert_zero(r);
+    r = env->open(env, ENVDIR, envflags, S_IRWXU+S_IRWXG+S_IRWXO);                      assert_zero(r);
+
+    DB *db[ndbs];
+    for (int dbnum = 0; dbnum < ndbs; dbnum++) {
+        r = db_create(&db[dbnum], env, 0);                                                        
+        assert_zero(r);
+        DBT dbt_dbnum; dbt_init(&dbt_dbnum, &dbnum, sizeof dbnum);
+        r = db[dbnum]->set_descriptor(db[dbnum], 1, &dbt_dbnum); 
+        assert_zero(r);
+        char dbname[32]; sprintf(dbname, "%d.tdb", dbnum);
+        r = db[dbnum]->open(db[dbnum], NULL, dbname, NULL, DB_BTREE, DB_AUTO_COMMIT|DB_CREATE, 0666);    
+        assert_zero(r);
+    }
+
+    r = env->txn_checkpoint(env, 0, 0, 0);                                              assert_zero(r);
+
+    for (int dbnum = 0; dbnum < ndbs; dbnum++) {
+        if (dbnum == 0)
+            populate_primary(env, db[dbnum], ndbs, nrows);
+        else
+            populate_secondary(env, db[dbnum], dbnum, nrows);
+    }
+
+    r = env->txn_checkpoint(env, 0, 0, 0);                                              assert_zero(r);
+
+    // update multiple key0
+    DB_TXN *txn = NULL;
+    r = env->txn_begin(env, NULL, &txn, 0); assert_zero(r);
+
+    update_diagonal(env, txn, db, ndbs, nrows);
+
+    toku_hard_crash_on_purpose();
+}
+
+static void
+verify_seq(DB_ENV *env, DB *db, int dbnum, int ndbs, int nrows) {
+    int r;
+    DB_TXN *txn = NULL;
+    r = env->txn_begin(env, NULL, &txn, 0); assert_zero(r);
+
+    DBC *cursor = NULL;
+    r = db->cursor(db, txn, &cursor, 0); assert_zero(r);
+    int i;
+    for (i = 0; ; i++) {
+        DBT key; memset(&key, 0, sizeof key);
+        DBT val; memset(&val, 0, sizeof val);
+        r = cursor->c_get(cursor, &key, &val, DB_NEXT);
+        if (r != 0)
+            break;
+        int k;
+        int expectk = get_key(i, dbnum);
+        assert(key.size == sizeof k);
+        memcpy(&k, key.data, key.size);
+        assert(k == expectk);
+
+        if (dbnum == 0) {
+            assert(val.size == ndbs * sizeof (int));
+            int v[ndbs]; get_data(v, i, ndbs);
+            assert(memcmp(val.data, v, val.size) == 0);
+        } else
+            assert(val.size == 0);
+    }
+    assert(i == nrows);
+    r = cursor->c_close(cursor); assert_zero(r);
+    r = txn->commit(txn, 0); assert_zero(r);
+}
+
+static void
+verify_all(DB_ENV *env, int ndbs, int nrows) {
+    int r;
+    for (int dbnum = 0; dbnum < ndbs; dbnum++) {
+        DB *db = NULL;
+        r = db_create(&db, env, 0);                                                        
+        assert_zero(r);
+        char dbname[32]; sprintf(dbname, "%d.tdb", dbnum);
+        r = db->open(db, NULL, dbname, NULL, DB_BTREE, DB_AUTO_COMMIT|DB_CREATE, 0666);    
+        assert_zero(r);
+        verify_seq(env, db, dbnum, ndbs, nrows);
+        r = db->close(db, 0); 
+        assert_zero(r);
+    }
+}
+
+static void
+run_recover(int ndbs, int nrows) {
+    int r;
+
+    DB_ENV *env;
+    r = db_env_create(&env, 0);                                                             assert_zero(r);
+    r = env->set_generate_row_callback_for_put(env, put_callback);                          assert_zero(r);
+    r = env->set_generate_row_callback_for_del(env, del_callback);                          assert_zero(r);
+    r = env->open(env, ENVDIR, envflags|DB_RECOVER, S_IRWXU+S_IRWXG+S_IRWXO);               assert_zero(r);
+    verify_all(env, ndbs, nrows);
+    r = env->close(env, 0);                                                                 assert_zero(r);
+}
+
+static int
+usage(void) {
+    return 1;
+}
+
+int
+test_main (int argc, char * const argv[]) {
+    BOOL do_test = FALSE;
+    BOOL do_recover = FALSE;
+    int ndbs = 2;
+    int nrows = 2;
+
+    for (int i = 1; i < argc; i++) {
+        char * const arg = argv[i];
+        if (strcmp(arg, "-v") == 0) {
+            verbose++;
+            continue;
+        }
+        if (strcmp(arg, "-q") == 0) {
+            verbose--;
+            if (verbose < 0)
+                verbose = 0;
+            continue;
+        }
+        if (strcmp(arg, "--test") == 0) {
+            do_test = TRUE;
+            continue;
+        }
+        if (strcmp(arg, "--recover") == 0) {
+            do_recover = TRUE;
+            continue;
+        }
+        if (strcmp(arg, "--ndbs") == 0 && i+1 < argc) {
+            ndbs = atoi(argv[++i]);
+            continue;
+        }
+        if (strcmp(arg, "--nrows") == 0 && i+1 < argc) {
+            nrows = atoi(argv[++i]);
+            continue;
+        }
+        if (strcmp(arg, "--help") == 0) {
+            return usage();
+        }
+    }
+    
+    if (do_test)
+        run_test(ndbs, nrows);
+    if (do_recover)
+        run_recover(ndbs, nrows);
+
+    return 0;
+}

src/tests/update-multiple-data-diagonal.c

+#include "test.h"
+
+// verify that update_multiple where we change the data in row[i] col[j] from x to x+1
+
+static int
+get_key(int i, int dbnum) {
+    return htonl(2*(i + dbnum));
+}
+
+static int
+get_new_key(int i, int dbnum) {
+    return htonl(2*(i + dbnum) + 1);
+}
+
+static void
+get_data(int *v, int i, int ndbs) {
+    for (int dbnum = 0; dbnum < ndbs; dbnum++) {
+        v[dbnum] = get_key(i, dbnum);
+    }
+}
+
+static void
+get_new_data(int *v, int i, int ndbs) {
+    for (int dbnum = 0; dbnum < ndbs; dbnum++) {
+        if ((i % ndbs) == dbnum)
+            v[dbnum] = get_new_key(i, dbnum);
+        else
+            v[dbnum] = get_key(i, dbnum);
+    }
+}
+
+static int
+put_callback(DB *dest_db, DB *src_db, DBT *dest_key, DBT *dest_data, const DBT *src_key, const DBT *src_data, void *extra) {
+    dest_db = dest_db; src_db = src_db; dest_key = dest_key; dest_data = dest_data; src_key = src_key; src_data = src_data;
+    assert(src_db == NULL);
+    assert(extra == NULL);
+
+    unsigned int dbnum;
+    assert(dest_db->descriptor->dbt.size == sizeof dbnum);
+    memcpy(&dbnum, dest_db->descriptor->dbt.data, sizeof dbnum);
+    assert(dbnum < src_data->size / sizeof (int));
+
+    int *pri_key = (int *) src_key->data;
+    int *pri_data = (int *) src_data->data;
+
+    switch (dest_key->flags) {
+    case 0:
+        dest_key->size = sizeof (int);
+        dest_key->data = dbnum == 0 ? &pri_key[dbnum] : &pri_data[dbnum];
+        break;
+    case DB_DBT_REALLOC:
+        dest_key->size = sizeof (int);
+        dest_key->data = toku_realloc(dest_key->data, dest_key->size);
+        memcpy(dest_key->data, dbnum == 0 ? &pri_key[dbnum] : &pri_data[dbnum], dest_key->size);
+        break;
+    default:
+        assert(0);
+    }
+
+    if (dest_data) {
+        switch (dest_data->flags) {
+        case 0:
+            if (dbnum == 0) {
+                dest_data->size = src_data->size;
+                dest_data->data = src_data->data;
+            } else
+                dest_data->size = 0;
+            break;
+        case DB_DBT_REALLOC:
+            if (dbnum == 0) {
+                dest_data->size = src_data->size;
+                dest_data->data = toku_realloc(dest_data->data, dest_data->size);
+                memcpy(dest_data->data, src_data->data, dest_data->size);
+            } else
+                dest_data->size = 0;
+            break;
+        default:
+            assert(0);
+        }
+    }
+    
+    return 0;
+}
+
+static int
+del_callback(DB *dest_db, DB *src_db, DBT *dest_key, const DBT *src_key, const DBT *src_data, void *extra) {
+    return put_callback(dest_db, src_db, dest_key, NULL, src_key, src_data, extra);
+}
+
+#if 0
+static void
+verify_locked(DB_ENV *env, DB *db, int k) {
+    int r;
+    DB_TXN *txn = NULL;
+    r = env->txn_begin(env, NULL, &txn, 0); assert_zero(r);
+    DBT key; dbt_init(&key, &k, sizeof k);
+    r = db->del(db, txn, &key, DB_DELETE_ANY); assert(r == DB_LOCK_NOTGRANTED);
+    r = txn->abort(txn); assert_zero(r);
+}
+
+static void
+verify_empty(DB_ENV *env, DB *db) {
+    int r;
+    DB_TXN *txn = NULL;
+    r = env->txn_begin(env, NULL, &txn, 0); assert_zero(r);
+
+    DBC *cursor = NULL;
+    r = db->cursor(db, txn, &cursor, 0); assert_zero(r);
+    int i;
+    for (i = 0; ; i++) {
+        DBT key; memset(&key, 0, sizeof key);
+        DBT val; memset(&val, 0, sizeof val);
+        r = cursor->c_get(cursor, &key, &val, DB_NEXT);
+        if (r != 0)
+            break;
+    }
+    assert_zero(i);
+    r = cursor->c_close(cursor); assert_zero(r);
+    r = txn->commit(txn, 0); assert_zero(r);
+}
+#endif
+
+static void
+verify_seq(DB_ENV *env, DB *db, int dbnum, int ndbs, int nrows) {
+    int r;
+    DB_TXN *txn = NULL;
+    r = env->txn_begin(env, NULL, &txn, 0); assert_zero(r);
+
+    DBC *cursor = NULL;
+    r = db->cursor(db, txn, &cursor, 0); assert_zero(r);
+    int i;
+    for (i = 0; ; i++) {
+        DBT key; memset(&key, 0, sizeof key);
+        DBT val; memset(&val, 0, sizeof val);
+        r = cursor->c_get(cursor, &key, &val, DB_NEXT);
+        if (r != 0)
+            break;
+        int k;
+        int expectk;
+        if (dbnum == 0 || (i % ndbs) != dbnum)
+            expectk = get_key(i, dbnum);
+        else
+            expectk = get_new_key(i, dbnum);
+     
+        assert(key.size == sizeof k);
+        memcpy(&k, key.data, key.size);
+        assert(k == expectk);
+
+        if (dbnum == 0) {
+            assert(val.size == ndbs * sizeof (int));
+            int v[ndbs]; get_new_data(v, i, ndbs);
+            assert(memcmp(val.data, v, val.size) == 0);
+        } else
+            assert(val.size == 0);
+    }
+    assert(i == nrows); // if (i != nrows) printf("%s:%d %d %d\n", __FUNCTION__, __LINE__, i, nrows); // assert(i == nrows);
+    r = cursor->c_close(cursor); assert_zero(r);
+    r = txn->commit(txn, 0); assert_zero(r);
+}
+
+static void
+update_diagonal(DB_ENV *env, DB *db[], int ndbs, int nrows) {
+    assert(ndbs > 0);
+    int r;
+    DB_TXN *txn = NULL;
+    r = env->txn_begin(env, NULL, &txn, 0); assert_zero(r);
+    for (int i = 0; i < nrows; i++) {
+
+        // update the data i % ndbs col from x to x+1
+
+        int k = get_key(i, 0);
+        DBT old_key; dbt_init(&old_key, &k, sizeof k);
+        DBT new_key = old_key;
+
+        int v[ndbs]; get_data(v, i, ndbs);
+        DBT old_data; dbt_init(&old_data, &v[0], sizeof v);
+        
+        int newv[ndbs]; get_new_data(newv, i, ndbs);
+        DBT new_data; dbt_init(&new_data, &newv[0], sizeof newv);
+  
+        int ndbts = 2 * ndbs;
+        DBT keys[ndbts]; memset(keys, 0, sizeof keys);
+        DBT vals[ndbts]; memset(vals, 0, sizeof vals);
+
+        r = env->update_multiple(env, NULL, txn, &old_key, &old_data, &new_key, &new_data, ndbs, db, ndbts, keys, vals, NULL);
+        assert_zero(r);
+    }
+    r = txn->commit(txn, 0); assert_zero(r);
+}
+
+static void
+populate_primary(DB_ENV *env, DB *db, int ndbs, int nrows) {
+    int r;
+    DB_TXN *txn = NULL;
+    r = env->txn_begin(env, NULL, &txn, 0); assert_zero(r);
+
+    // populate
+    for (int i = 0; i < nrows; i++) {
+        int k = get_key(i, 0);
+        int v[ndbs]; get_data(v, i, ndbs);
+        DBT key; dbt_init(&key, &k, sizeof k);
+        DBT val; dbt_init(&val, &v[0], sizeof v);
+        r = db->put(db, txn, &key, &val, DB_YESOVERWRITE); assert_zero(r);
+    }
+
+    r = txn->commit(txn, 0); assert_zero(r);
+}
+
+static void
+populate_secondary(DB_ENV *env, DB *db, int dbnum, int nrows) {
+    int r;
+    DB_TXN *txn = NULL;
+    r = env->txn_begin(env, NULL, &txn, 0); assert_zero(r);
+
+    // populate
+    for (int i = 0; i < nrows; i++) {
+        int k = get_key(i, dbnum);
+        DBT key; dbt_init(&key, &k, sizeof k);
+        DBT val; dbt_init(&val, NULL, 0);
+        r = db->put(db, txn, &key, &val, DB_YESOVERWRITE); assert_zero(r);
+    }
+
+    r = txn->commit(txn, 0); assert_zero(r);
+}
+
+static void
+run_test(int ndbs, int nrows) {
+    int r;
+    DB_ENV *env = NULL;
+    r = db_env_create(&env, 0); assert_zero(r);
+
+    r = env->set_generate_row_callback_for_put(env, put_callback); assert_zero(r);
+    r = env->set_generate_row_callback_for_del(env, del_callback); assert_zero(r);
+
+    r = env->open(env, ENVDIR, DB_INIT_MPOOL|DB_CREATE|DB_THREAD |DB_INIT_LOCK|DB_INIT_LOG|DB_INIT_TXN|DB_PRIVATE, S_IRWXU+S_IRWXG+S_IRWXO); assert_zero(r);
+
+    DB *db[ndbs];
+    for (int dbnum = 0; dbnum < ndbs; dbnum++) {
+        r = db_create(&db[dbnum], env, 0); assert_zero(r);
+
+        DBT dbt_dbnum; dbt_init(&dbt_dbnum, &dbnum, sizeof dbnum);
+        r = db[dbnum]->set_descriptor(db[dbnum], 1, &dbt_dbnum); assert_zero(r);
+
+        char dbname[32]; sprintf(dbname, "%d.tdb", dbnum);
+        r = db[dbnum]->open(db[dbnum], NULL, dbname, NULL, DB_BTREE, DB_AUTO_COMMIT+DB_CREATE, S_IRWXU+S_IRWXG+S_IRWXO); assert_zero(r);
+    }
+
+    for (int dbnum = 0; dbnum < ndbs; dbnum++) {
+        if (dbnum == 0)
+            populate_primary(env, db[dbnum], ndbs, nrows);
+        else
+            populate_secondary(env, db[dbnum], dbnum, nrows);
+    }
+
+    update_diagonal(env, db, ndbs, nrows);
+    for (int dbnum = 0; dbnum < ndbs; dbnum++) 
+        verify_seq(env, db[dbnum], dbnum, ndbs, nrows);
+    for (int dbnum = 0; dbnum < ndbs; dbnum++) 
+        r = db[dbnum]->close(db[dbnum], 0); assert_zero(r);
+
+    r = env->close(env, 0); assert_zero(r);
+}
+
+int
+test_main(int argc, char * const argv[]) {
+    int r;
+    int ndbs = 2;
+    int nrows = 2;
+
+    // parse_args(argc, argv);
+    for (int i = 1; i < argc; i++) {
+        char * const arg = argv[i];
+        if (strcmp(arg, "-v") == 0) {
+            verbose++;
+            continue;
+        }
+        if (strcmp(arg, "-q") == 0) {
+            verbose = 0;
+            continue;
+        }
+        if (strcmp(arg, "--ndbs") == 0 && i+1 < argc) {
+            ndbs = atoi(argv[++i]);
+            continue;
+        }
+        if (strcmp(arg, "--nrows") == 0 && i+1 < argc) {
+            nrows = atoi(argv[++i]);
+            continue;
+        }
+    }
+
+    r = system("rm -rf " ENVDIR); assert_zero(r);
+    r = toku_os_mkdir(ENVDIR, S_IRWXU+S_IRWXG+S_IRWXO); assert_zero(r);
+
+    run_test(ndbs, nrows);
+
+    return 0;
+}
+

src/tests/update-multiple-key0.c

+#include "test.h"
+
+// verify that update_multiple where we only change key0
+
+static int
+get_key(int i, int dbnum) {
+    return htonl(i + dbnum);
+}
+
+static void
+get_data(int *v, int i, int ndbs) {
+    for (int dbnum = 0; dbnum < ndbs; dbnum++) {
+        v[dbnum] = get_key(i, dbnum);
+    }
+}
+
+static int
+put_callback(DB *dest_db, DB *src_db, DBT *dest_key, DBT *dest_data, const DBT *src_key, const DBT *src_data, void *extra) {
+    dest_db = dest_db; src_db = src_db; dest_key = dest_key; dest_data = dest_data; src_key = src_key; src_data = src_data;
+    assert(src_db == NULL);
+    assert(extra == NULL);
+
+    unsigned int dbnum;
+    assert(dest_db->descriptor->dbt.size == sizeof dbnum);
+    memcpy(&dbnum, dest_db->descriptor->dbt.data, sizeof dbnum);
+    assert(dbnum < src_data->size / sizeof (int));
+
+    int *pri_key = (int *) src_key->data;
+    int *pri_data = (int *) src_data->data;
+
+    switch (dest_key->flags) {
+    case 0:
+        dest_key->size = sizeof (int);
+        dest_key->data = dbnum == 0 ? &pri_key[dbnum] : &pri_data[dbnum];
+        break;
+    case DB_DBT_REALLOC:
+        dest_key->size = sizeof (int);
+        dest_key->data = toku_realloc(dest_key->data, dest_key->size);
+        memcpy(dest_key->data, dbnum == 0 ? &pri_key[dbnum] : &pri_data[dbnum], dest_key->size);
+        break;
+    default:
+        assert(0);
+    }
+
+    if (dest_data) {
+        switch (dest_data->flags) {
+        case 0:
+            if (dbnum == 0) {
+                dest_data->size = src_data->size;
+                dest_data->data = src_data->data;
+            } else
+                dest_data->size = 0;
+            break;
+        case DB_DBT_REALLOC:
+            if (dbnum == 0) {
+                dest_data->size = src_data->size;
+                dest_data->data = toku_realloc(dest_data->data, dest_data->size);
+                memcpy(dest_data->data, src_data->data, dest_data->size);
+            } else
+                dest_data->size = 0;
+            break;
+        default:
+            assert(0);
+        }
+    }
+    
+    return 0;
+}
+
+static int
+del_callback(DB *dest_db, DB *src_db, DBT *dest_key, const DBT *src_key, const DBT *src_data, void *extra) {
+    return put_callback(dest_db, src_db, dest_key, NULL, src_key, src_data, extra);
+}
+
+static void
+verify_locked(DB_ENV *env, DB *db, int k) {
+    int r;
+    DB_TXN *txn = NULL;
+    r = env->txn_begin(env, NULL, &txn, 0); assert_zero(r);
+    DBT key; dbt_init(&key, &k, sizeof k);
+    r = db->del(db, txn, &key, DB_DELETE_ANY); assert(r == DB_LOCK_NOTGRANTED);
+    r = txn->abort(txn); assert_zero(r);
+}
+
+#if 0
+static void
+verify_empty(DB_ENV *env, DB *db) {
+    int r;
+    DB_TXN *txn = NULL;
+    r = env->txn_begin(env, NULL, &txn, 0); assert_zero(r);
+
+    DBC *cursor = NULL;
+    r = db->cursor(db, txn, &cursor, 0); assert_zero(r);
+    int i;
+    for (i = 0; ; i++) {
+        DBT key; memset(&key, 0, sizeof key);
+        DBT val; memset(&val, 0, sizeof val);
+        r = cursor->c_get(cursor, &key, &val, DB_NEXT);
+        if (r != 0)
+            break;
+    }
+    assert_zero(i);
+    r = cursor->c_close(cursor); assert_zero(r);
+    r = txn->commit(txn, 0); assert_zero(r);
+}
+#endif
+
+static void
+verify_seq(DB_ENV *env, DB *db, int dbnum, int ndbs, int nrows) {
+    int r;
+    DB_TXN *txn = NULL;
+    r = env->txn_begin(env, NULL, &txn, 0); assert_zero(r);
+
+    DBC *cursor = NULL;
+    r = db->cursor(db, txn, &cursor, 0); assert_zero(r);
+    int i;
+    for (i = 0; ; i++) {
+        DBT key; memset(&key, 0, sizeof key);
+        DBT val; memset(&val, 0, sizeof val);
+        r = cursor->c_get(cursor, &key, &val, DB_NEXT);
+        if (r != 0)
+            break;
+        int k;
+        int expectk = dbnum == 0 ? get_key(i + nrows, dbnum) : get_key(i, dbnum);
+        assert(key.size == sizeof k);
+        memcpy(&k, key.data, key.size);
+        assert(k == expectk);
+
+        if (dbnum == 0) {
+            assert(val.size == ndbs * sizeof (int));
+            int v[ndbs]; get_data(v, i, ndbs);
+            assert(memcmp(val.data, v, val.size) == 0);
+        } else
+            assert(val.size == 0);
+    }
+    assert(i == nrows);
+    r = cursor->c_close(cursor); assert_zero(r);
+    r = txn->commit(txn, 0); assert_zero(r);
+}
+
+static void
+update_key0(DB_ENV *env, DB *db[], int ndbs, int nrows) {
+    assert(ndbs > 0);
+    int r;
+    DB_TXN *txn = NULL;
+    r = env->txn_begin(env, NULL, &txn, 0); assert_zero(r);
+    for (int i = 0; i < nrows; i++) {
+
+        // update where new key0 = old key0 + nrows
+
+        int k = get_key(i, 0);
+        DBT old_key; dbt_init(&old_key, &k, sizeof k);
+        int newk = get_key(i + nrows, 0);
+        DBT new_key; dbt_init(&new_key, &newk, sizeof newk);
+
+        int v[ndbs]; get_data(v, i, ndbs);
+        DBT old_data; dbt_init(&old_data, &v[0], sizeof v);
+        DBT new_data = old_data;
+      
+        int ndbts = 2 * ndbs;
+        DBT keys[ndbts]; memset(keys, 0, sizeof keys);
+        DBT vals[ndbts]; memset(vals, 0, sizeof vals);
+
+        r = env->update_multiple(env, NULL, txn, &old_key, &old_data, &new_key, &new_data, ndbs, db, ndbts, keys, vals, NULL);
+        assert_zero(r);
+
+        verify_locked(env, db[0], k);
+        verify_locked(env, db[0], newk);
+    }
+    r = txn->commit(txn, 0); assert_zero(r);
+}
+
+static void
+populate_primary(DB_ENV *env, DB *db, int ndbs, int nrows) {
+    int r;
+    DB_TXN *txn = NULL;
+    r = env->txn_begin(env, NULL, &txn, 0); assert_zero(r);
+
+    // populate
+    for (int i = 0; i < nrows; i++) {
+        int k = get_key(i, 0);
+        int v[ndbs]; get_data(v, i, ndbs);
+        DBT key; dbt_init(&key, &k, sizeof k);
+        DBT val; dbt_init(&val, &v[0], sizeof v);
+        r = db->put(db, txn, &key, &val, DB_YESOVERWRITE); assert_zero(r);
+    }
+
+    r = txn->commit(txn, 0); assert_zero(r);
+}
+
+static void
+populate_secondary(DB_ENV *env, DB *db, int dbnum, int nrows) {
+    int r;
+    DB_TXN *txn = NULL;
+    r = env->txn_begin(env, NULL, &txn, 0); assert_zero(r);
+
+    // populate
+    for (int i = 0; i < nrows; i++) {
+        int k = get_key(i, dbnum);
+        DBT key; dbt_init(&key, &k, sizeof k);
+        DBT val; dbt_init(&val, NULL, 0);
+        r = db->put(db, txn, &key, &val, DB_YESOVERWRITE); assert_zero(r);
+    }
+
+    r = txn->commit(txn, 0); assert_zero(r);
+}
+
+static void
+run_test(int ndbs, int nrows) {
+    int r;
+    DB_ENV *env = NULL;
+    r = db_env_create(&env, 0); assert_zero(r);
+
+    r = env->set_generate_row_callback_for_put(env, put_callback); assert_zero(r);
+    r = env->set_generate_row_callback_for_del(env, del_callback); assert_zero(r);
+
+    r = env->open(env, ENVDIR, DB_INIT_MPOOL|DB_CREATE|DB_THREAD |DB_INIT_LOCK|DB_INIT_LOG|DB_INIT_TXN|DB_PRIVATE, S_IRWXU+S_IRWXG+S_IRWXO); assert_zero(r);
+
+    DB *db[ndbs];
+    for (int dbnum = 0; dbnum < ndbs; dbnum++) {
+        r = db_create(&db[dbnum], env, 0); assert_zero(r);
+
+        DBT dbt_dbnum; dbt_init(&dbt_dbnum, &dbnum, sizeof dbnum);
+        r = db[dbnum]->set_descriptor(db[dbnum], 1, &dbt_dbnum); assert_zero(r);
+
+        char dbname[32]; sprintf(dbname, "%d.tdb", dbnum);
+        r = db[dbnum]->open(db[dbnum], NULL, dbname, NULL, DB_BTREE, DB_AUTO_COMMIT+DB_CREATE, S_IRWXU+S_IRWXG+S_IRWXO); assert_zero(r);
+    }
+
+    for (int dbnum = 0; dbnum < ndbs; dbnum++) {
+        if (dbnum == 0)
+            populate_primary(env, db[dbnum], ndbs, nrows);
+        else
+            populate_secondary(env, db[dbnum], dbnum, nrows);
+    }
+
+    update_key0(env, db, ndbs, nrows);
+    for (int dbnum = 0; dbnum < ndbs; dbnum++) 
+        verify_seq(env, db[dbnum], dbnum, ndbs, nrows);
+
+    for (int dbnum = 0; dbnum < ndbs; dbnum++) 
+        r = db[dbnum]->close(db[dbnum], 0); assert_zero(r);
+
+    r = env->close(env, 0); assert_zero(r);
+}
+
+int
+test_main(int argc, char * const argv[]) {
+    int r;
+    int ndbs = 2;
+    int nrows = 2;
+
+    // parse_args(argc, argv);
+    for (int i = 1; i < argc; i++) {
+        char * const arg = argv[i];
+        if (strcmp(arg, "-v") == 0) {
+            verbose++;
+            continue;
+        }
+        if (strcmp(arg, "-q") == 0) {
+            verbose = 0;
+            continue;
+        }
+        if (strcmp(arg, "--ndbs") == 0 && i+1 < argc) {
+            ndbs = atoi(argv[++i]);
+            continue;
+        }
+        if (strcmp(arg, "--nrows") == 0 && i+1 < argc) {
+            nrows = atoi(argv[++i]);
+            continue;
+        }
+    }
+
+    r = system("rm -rf " ENVDIR); assert_zero(r);
+    r = toku_os_mkdir(ENVDIR, S_IRWXU+S_IRWXG+S_IRWXO); assert_zero(r);
+
+    run_test(ndbs, nrows);
+
+    return 0;
+}
+

src/tests/update-multiple-nochange.c

+#include "test.h"
+
+// verify that update_multiple where new row = old row
+
+static int
+get_key(int i, int dbnum) {
+    return htonl(i + dbnum);
+}
+
+static void
+get_data(int *v, int i, int ndbs) {
+    for (int dbnum = 0; dbnum < ndbs; dbnum++) {
+        v[dbnum] = get_key(i, dbnum);
+    }
+}
+
+static int
+put_callback(DB *dest_db, DB *src_db, DBT *dest_key, DBT *dest_data, const DBT *src_key, const DBT *src_data, void *extra) {
+    dest_db = dest_db; src_db = src_db; dest_key = dest_key; dest_data = dest_data; src_key = src_key; src_data = src_data;
+    assert(src_db == NULL);
+    assert(extra == NULL);
+
+    unsigned int dbnum;
+    assert(dest_db->descriptor->dbt.size == sizeof dbnum);
+    memcpy(&dbnum, dest_db->descriptor->dbt.data, sizeof dbnum);
+    assert(dbnum < src_data->size / sizeof (int));
+
+    int *pri_data = (int *) src_data->data;
+
+    switch (dest_key->flags) {
+    case 0:
+        dest_key->size = sizeof (int);
+        dest_key->data = &pri_data[dbnum];
+        break;
+    case DB_DBT_REALLOC:
+        dest_key->size = sizeof (int);
+        dest_key->data = toku_realloc(dest_key->data, dest_key->size);
+        memcpy(dest_key->data, &pri_data[dbnum], dest_key->size);
+        break;
+    default:
+        assert(0);
+    }
+
+    if (dest_data) {
+        switch (dest_data->flags) {
+        case 0:
+            if (dbnum == 0) {
+                dest_data->size = src_data->size;
+                dest_data->data = src_data->data;
+            } else
+                dest_data->size = 0;
+            break;
+        case DB_DBT_REALLOC:
+            if (dbnum == 0) {
+                dest_data->size = src_data->size;
+                dest_data->data = toku_realloc(dest_data->data, dest_data->size);
+                memcpy(dest_data->data, src_data->data, dest_data->size);
+            } else
+                dest_data->size = 0;
+            break;
+        default:
+            assert(0);
+        }
+    }
+    
+    return 0;
+}
+
+static int
+del_callback(DB *dest_db, DB *src_db, DBT *dest_key, const DBT *src_key, const DBT *src_data, void *extra) {
+    return put_callback(dest_db, src_db, dest_key, NULL, src_key, src_data, extra);
+}
+
+#if 0
+static void
+verify_locked(DB_ENV *env, DB *db, int k) {
+    int r;
+    DB_TXN *txn = NULL;
+    r = env->txn_begin(env, NULL, &txn, 0); assert_zero(r);
+    DBT key; dbt_init(&key, &k, sizeof k);
+    r = db->del(db, txn, &key, DB_DELETE_ANY); assert(r == DB_LOCK_NOTGRANTED);
+    r = txn->abort(txn); assert_zero(r);
+}
+
+static void
+verify_empty(DB_ENV *env, DB *db) {
+    int r;
+    DB_TXN *txn = NULL;
+    r = env->txn_begin(env, NULL, &txn, 0); assert_zero(r);
+
+    DBC *cursor = NULL;
+    r = db->cursor(db, txn, &cursor, 0); assert_zero(r);
+    int i;
+    for (i = 0; ; i++) {
+        DBT key; memset(&key, 0, sizeof key);
+        DBT val; memset(&val, 0, sizeof val);
+        r = cursor->c_get(cursor, &key, &val, DB_NEXT);
+        if (r != 0)
+            break;
+    }
+    assert_zero(i);
+    r = cursor->c_close(cursor); assert_zero(r);
+    r = txn->commit(txn, 0); assert_zero(r);
+}
+#endif
+
+static void
+verify_seq(DB_ENV *env, DB *db, int dbnum, int ndbs, int nrows) {
+    int r;
+    DB_TXN *txn = NULL;
+    r = env->txn_begin(env, NULL, &txn, 0); assert_zero(r);
+
+    DBC *cursor = NULL;
+    r = db->cursor(db, txn, &cursor, 0); assert_zero(r);
+    int i;
+    for (i = 0; ; i++) {
+        DBT key; memset(&key, 0, sizeof key);
+        DBT val; memset(&val, 0, sizeof val);
+        r = cursor->c_get(cursor, &key, &val, DB_NEXT);
+        if (r != 0)
+            break;
+        int k;
+        assert(key.size == sizeof k);
+        memcpy(&k, key.data, key.size);
+        assert(k == get_key(i, dbnum));
+
+        if (dbnum == 0) {
+            assert(val.size == ndbs * sizeof (int));
+            int v[ndbs]; get_data(v, i, ndbs);
+            assert(memcmp(val.data, v, val.size) == 0);
+        } else
+            assert(val.size == 0);
+    }
+    assert(i == nrows);
+    r = cursor->c_close(cursor); assert_zero(r);
+    r = txn->commit(txn, 0); assert_zero(r);
+}
+
+static void
+verify(DB_ENV *env, DB *db[], int ndbs, int nrows) {
+    int r;
+    DB_TXN *txn = NULL;
+    r = env->txn_begin(env, NULL, &txn, 0); assert_zero(r);
+    for (int i = 0; i < nrows; i++) {
+
+        // update where new row = old row
+
+        int k = get_key(i, 0);
+        DBT old_key; dbt_init(&old_key, &k, sizeof k);
+        DBT new_key = old_key;
+        int v[ndbs]; get_data(v, i, ndbs);
+        DBT old_data; dbt_init(&old_data, &v[0], sizeof v);
+        DBT new_data = old_data;
+        
+        int ndbts = 2 * ndbs;
+        DBT keys[ndbts]; memset(keys, 0, sizeof keys);
+        DBT vals[ndbts]; memset(vals, 0, sizeof vals);
+
+        r = env->update_multiple(env, NULL, txn, &old_key, &old_data, &new_key, &new_data, ndbs, db, ndbts, keys, vals, NULL);
+        assert_zero(r);
+    }
+    r = txn->commit(txn, 0); assert_zero(r);
+    for (int dbnum = 0; dbnum < ndbs; dbnum++) 
+        verify_seq(env, db[dbnum], dbnum, ndbs, nrows);
+}
+
+static void
+populate_primary(DB_ENV *env, DB *db, int ndbs, int nrows) {
+    int r;
+    DB_TXN *txn = NULL;
+    r = env->txn_begin(env, NULL, &txn, 0); assert_zero(r);
+
+    // populate
+    for (int i = 0; i < nrows; i++) {
+        int k = get_key(i, 0);
+        int v[ndbs]; get_data(v, i, ndbs);
+        DBT key; dbt_init(&key, &k, sizeof k);
+        DBT val; dbt_init(&val, &v[0], sizeof v);
+        r = db->put(db, txn, &key, &val, DB_YESOVERWRITE); assert_zero(r);
+    }
+
+    r = txn->commit(txn, 0); assert_zero(r);
+}
+
+static void
+populate_secondary(DB_ENV *env, DB *db, int dbnum, int nrows) {
+    int r;
+    DB_TXN *txn = NULL;
+    r = env->txn_begin(env, NULL, &txn, 0); assert_zero(r);
+
+    // populate
+    for (int i = 0; i < nrows; i++) {
+        int k = get_key(i, dbnum);
+        DBT key; dbt_init(&key, &k, sizeof k);
+        DBT val; dbt_init(&val, NULL, 0);
+        r = db->put(db, txn, &key, &val, DB_YESOVERWRITE); assert_zero(r);
+    }
+
+    r = txn->commit(txn, 0); assert_zero(r);
+}
+
+static void
+run_test(int ndbs, int nrows) {
+    int r;
+    DB_ENV *env = NULL;
+    r = db_env_create(&env, 0); assert_zero(r);
+
+    r = env->set_generate_row_callback_for_put(env, put_callback); assert_zero(r);
+    r = env->set_generate_row_callback_for_del(env, del_callback); assert_zero(r);
+
+    r = env->open(env, ENVDIR, DB_INIT_MPOOL|DB_CREATE|DB_THREAD |DB_INIT_LOCK|DB_INIT_LOG|DB_INIT_TXN|DB_PRIVATE, S_IRWXU+S_IRWXG+S_IRWXO); assert_zero(r);
+
+    DB *db[ndbs];
+    for (int dbnum = 0; dbnum < ndbs; dbnum++) {
+        r = db_create(&db[dbnum], env, 0); assert_zero(r);
+
+        DBT dbt_dbnum; dbt_init(&dbt_dbnum, &dbnum, sizeof dbnum);
+        r = db[dbnum]->set_descriptor(db[dbnum], 1, &dbt_dbnum); assert_zero(r);
+
+        char dbname[32]; sprintf(dbname, "%d.tdb", dbnum);
+        r = db[dbnum]->open(db[dbnum], NULL, dbname, NULL, DB_BTREE, DB_AUTO_COMMIT+DB_CREATE, S_IRWXU+S_IRWXG+S_IRWXO); assert_zero(r);
+    }
+
+    for (int dbnum = 0; dbnum < ndbs; dbnum++) {
+        if (dbnum == 0)
+            populate_primary(env, db[dbnum], ndbs, nrows);
+        else
+            populate_secondary(env, db[dbnum], dbnum, nrows);
+    }
+
+    verify(env, db, ndbs, nrows);
+
+    for (int dbnum = 0; dbnum < ndbs; dbnum++) 
+        r = db[dbnum]->close(db[dbnum], 0); assert_zero(r);
+
+    r = env->close(env, 0); assert_zero(r);
+}
+
+int
+test_main(int argc, char * const argv[]) {
+    int r;
+    int ndbs = 2;
+    int nrows = 2;
+
+    // parse_args(argc, argv);
+    for (int i = 1; i < argc; i++) {
+        char * const arg = argv[i];
+        if (strcmp(arg, "-v") == 0) {
+            verbose++;
+            continue;
+        }
+        if (strcmp(arg, "-q") == 0) {
+            verbose = 0;
+            continue;
+        }
+        if (strcmp(arg, "--ndbs") == 0 && i+1 < argc) {
+            ndbs = atoi(argv[++i]);
+            continue;
+        }
+        if (strcmp(arg, "--nrows") == 0 && i+1 < argc) {
+            nrows = atoi(argv[++i]);
+            continue;
+        }
+    }
+
+    r = system("rm -rf " ENVDIR); assert_zero(r);
+    r = toku_os_mkdir(ENVDIR, S_IRWXU+S_IRWXG+S_IRWXO); assert_zero(r);
+
+    run_test(ndbs, nrows);
+
+    return 0;
+}
+