Bug #16004999 ASSERT STATE == TRX_STATE_NOT_STARTED, UNLOCK_ROW()

Problem:

During the index intersect access method, the SQL layer will access one row,
that satisfies a set of conditions, using an index i1.  And then it will try to
access the same row, with other set of conditions using the next index i2.  If
the fetch from i2 fails (we are talking about an error situation here and not
simply an unmatched row situation), then it will unlock the row accessed via
i1.  This will work in all situations except deadlock error.

When a deadlock happens, InnoDB will rollback the transaction.  InnoDB intimates
the SQL layer about this through the THD::transaction_rollback_request member.
But this is not currently used by the SQL layer.

Solution:

When an error happens, the SQL layer must check the 
THD::transaction_rollback_request member, before calling handler::unlock_row().
We have also added a debug assert in ha_innobase::unlock_row() checking that
it must be called only when the transaction is in active state.

rb#1773 approved by Marko and Sunny.

sql/opt_range.cc

@@ -8469,9 +8469,13 @@ int QUICK_ROR_INTERSECT_SELECT::get_next()
 
       do
       {
+        DBUG_EXECUTE_IF("innodb_quick_report_deadlock",
+                        DBUG_SET("+d,innodb_report_deadlock"););
         if ((error= quick->get_next()))
         {
-          quick_with_last_rowid->file->unlock_row();
+          /* On certain errors like deadlock, trx might be rolled back.*/
+          if (!current_thd->transaction_rollback_request)
+            quick_with_last_rowid->file->unlock_row();
           DBUG_RETURN(error);
         }
         quick->file->position(quick->record);
@@ -8494,7 +8498,9 @@ int QUICK_ROR_INTERSECT_SELECT::get_next()
             quick->file->unlock_row(); /* row not in range; unlock */
             if ((error= quick->get_next()))
             {
-              quick_with_last_rowid->file->unlock_row();
+              /* On certain errors like deadlock, trx might be rolled back.*/
+              if (!current_thd->transaction_rollback_request)
+                quick_with_last_rowid->file->unlock_row();
               DBUG_RETURN(error);
             }
           }

storage/innobase/handler/ha_innodb.cc

@@ -5596,6 +5596,8 @@ ha_innobase::unlock_row(void)
 {
 	DBUG_ENTER("ha_innobase::unlock_row");
 
+	ut_ad(prebuilt->trx->conc_state == TRX_ACTIVE);
+
 	/* Consistent read does not take any locks, thus there is
 	nothing to unlock. */
 

storage/innobase/lock/lock0lock.c

@@ -1997,6 +1997,8 @@ lock_rec_lock_fast(
 	      || mode - (LOCK_MODE_MASK & mode) == 0
 	      || mode - (LOCK_MODE_MASK & mode) == LOCK_REC_NOT_GAP);
 
+	DBUG_EXECUTE_IF("innodb_report_deadlock", return(LOCK_REC_FAIL););
+
 	lock = lock_rec_get_first_on_page(block);
 
 	trx = thr_get_trx(thr);
@@ -2074,6 +2076,8 @@ lock_rec_lock_slow(
 
 	trx = thr_get_trx(thr);
 
+	DBUG_EXECUTE_IF("innodb_report_deadlock", return(DB_DEADLOCK););
+
 	lock = lock_rec_has_expl(mode, block, heap_no, trx);
 	if (lock) {
 		if (lock->type_mode & LOCK_CONV_BY_OTHER) {
@@ -4124,6 +4128,7 @@ lock_rec_unlock(
 
 	ut_ad(trx && rec);
 	ut_ad(block->frame == page_align(rec));
+	ut_ad(trx->conc_state == TRX_ACTIVE);
 
 	heap_no = page_rec_get_heap_no(rec);