Issue facebook#57: Release row locks on statement errors

Summary:
Do exactly that: remember what locks we've held at the start of the
statement, and release all other locks if the statement is rolled back.

Differential Revision: https://reviews.facebook.net/D38313

storage/rocksdb/ha_rocksdb.cc

@@ -990,6 +990,9 @@ class Rdb_transaction
   /* Row locks taken by this transaction */
   Dynamic_array<Row_lock*> trx_locks;
 
+  /* Number of elements in trx_locks array when the current statement started */
+  int trx_locks_at_stmt_start;
+
   int timeout_sec; /* Cached value of @@rocksdb_lock_wait_timeout */
   int max_row_locks;
 
@@ -1101,13 +1104,33 @@ class Rdb_transaction
     /* Currently, we don't do anything here */
   }
 
+  /*
+    Start a statement inside a multi-statement transaction.
+
+    For hooking to start of statement that is its own transaction, see
+    ha_rocksdb::external_lock().
+  */
+  void start_stmt()
+  {
+    changes.start_stmt();
+    trx_locks_at_stmt_start= trx_locks.elements();
+  }
+
   /*
     This must be called when last statement is rolled back, but the transaction
     continues
   */
-  void rollback_stmt() { changes.rollback_stmt(); }
+  void rollback_stmt()
+  {
+    changes.rollback_stmt();
 
-  void start_stmt() { changes.start_stmt(); }
+    /* Release the locks taken by the last statement */
+    int size= trx_locks.elements();
+    DBUG_ASSERT(trx_locks_at_stmt_start <= size);
+    for (int i= trx_locks_at_stmt_start; i < size; i++)
+      row_locks.release_lock(pins, trx_locks.at(i));
+    trx_locks.elements(trx_locks_at_stmt_start);
+  }
   void rollback()
   {
     if (snapshot)
@@ -1122,7 +1145,7 @@ class Rdb_transaction
   }
 
   Rdb_transaction() :
-    snapshot(NULL), n_writes(0), pins(NULL)
+    snapshot(NULL), n_writes(0), pins(NULL), trx_locks_at_stmt_start(0)
   {
     changes.init();
   }
@@ -4562,6 +4585,8 @@ int ha_rocksdb::external_lock(THD *thd, int lock_type)
       So, we put this code here.
     */
     Rdb_transaction* trx= get_or_create_trx(thd);
+    trx->trx_locks_at_stmt_start= trx->trx_locks.elements();
+
     trx->set_params(THDVAR(table->in_use, lock_wait_timeout),
                     THDVAR(table->in_use, max_row_locks));
 
@@ -4588,6 +4613,16 @@ int ha_rocksdb::external_lock(THD *thd, int lock_type)
 }
 
 
+/*
+  @note
+  A quote from ha_innobase::start_stmt():
+  <quote>
+  MySQL calls this function at the start of each SQL statement inside LOCK
+  TABLES. Inside LOCK TABLES the ::external_lock method does not work to
+  mark SQL statement borders.
+  </quote>
+*/
+
 int ha_rocksdb::start_stmt(THD *thd, thr_lock_type lock_type)
 {
   if (thd_test_options(thd, OPTION_NOT_AUTOCOMMIT | OPTION_BEGIN))