A fix and a test case for

Bug#41756 "Strange error messages about locks from InnoDB".
      
In JT_EQ_REF (join_read_key()) access method, 
don't try to unlock rows in the handler, unless certain that 
a) they were locked
b) they are not used.

Unlocking of rows is done by the logic of the nested join loop,
and is unaware of the possible caching that the access method may
have. This could lead to double unlocking, when a row
was unlocked first after reading into the cache, and then 
when taken from cache, as well as to unlocking of rows which
were actually used (but taken from cache).
      
Delegate part of the unlocking logic to the access method,
and in JT_EQ_REF count how many times a record was actually 
used in the join. Unlock it only if it's usage count is 0.

Implemented review comments.

mysql-test/r/innodb_lock_wait_timeout_1.result:
  Update results (Bug41756).
mysql-test/t/innodb_lock_wait_timeout_1.test:
  Add a test case (Bug#41756).
sql/item_subselect.cc:
  Complete struct READ_RECORD initialization with a new
  member to unlock records.
sql/records.cc:
  Extend READ_RECORD API with a method to unlock read records.
sql/sql_select.cc:
  In JT_EQ_REF (join_read_key()) access method, 
  don't try to unlock rows in the handler, unless certain that 
  a) they were locked
  b) they are not used.
sql/sql_select.h:
  Add members to TABLE_REF to count TABLE_REF buffer usage count.
sql/structs.h:
  Update declarations.

5 files changed, 78 insertions, 6 deletions

diff --git a/sql/item_subselect.cc b/sql/item_subselect.cc
index 29db9eb0903..fa776ea3dca 100644
--- a/sql/item_subselect.cc
+++ b/sql/item_subselect.cc
@@ -1954,6 +1954,7 @@ int subselect_single_select_engine::exec()
               tab->read_record.record= tab->table->record[0];
               tab->read_record.thd= join->thd;
               tab->read_record.ref_length= tab->table->file->ref_length;
+              tab->read_record.unlock_row= rr_unlock_row;
               *(last_changed_tab++)= tab;
               break;
             }
diff --git a/sql/records.cc b/sql/records.cc
index 398dad42c4f..b6faf0227f9 100644
--- a/sql/records.cc
+++ b/sql/records.cc
@@ -62,6 +62,7 @@ void init_read_record_idx(READ_RECORD *info, THD *thd, TABLE *table,
   info->file=  table->file;
   info->record= table->record[0];
   info->print_error= print_error;
+  info->unlock_row= rr_unlock_row;
 
   table->status=0;			/* And it's always found */
   if (!table->file->inited)
@@ -187,6 +188,7 @@ void init_read_record(READ_RECORD *info,THD *thd, TABLE *table,
   }
   info->select=select;
   info->print_error=print_error;
+  info->unlock_row= rr_unlock_row;
   info->ignore_not_found_rows= 0;
   table->status=0;			/* And it's always found */
 
diff --git a/sql/sql_select.cc b/sql/sql_select.cc
index 854dd296039..cdaebef49f6 100644
--- a/sql/sql_select.cc
+++ b/sql/sql_select.cc
@@ -149,6 +149,7 @@ static int join_read_const_table(JOIN_TAB *tab, POSITION *pos);
 static int join_read_system(JOIN_TAB *tab);
 static int join_read_const(JOIN_TAB *tab);
 static int join_read_key(JOIN_TAB *tab);
+static void join_read_key_unlock_row(st_join_table *tab);
 static int join_read_always_key(JOIN_TAB *tab);
 static int join_read_last_key(JOIN_TAB *tab);
 static int join_no_more_records(READ_RECORD *info);
@@ -5628,7 +5629,9 @@ static bool create_ref_for_key(JOIN *join, JOIN_TAB *j, KEYUSE *org_keyuse,
   }
   j->ref.key_buff2=j->ref.key_buff+ALIGN_SIZE(length);
   j->ref.key_err=1;
+  j->ref.has_record= FALSE;
   j->ref.null_rejecting= 0;
+  j->ref.use_count= 0;
   keyuse=org_keyuse;
 
   store_key **ref_key= j->ref.key_copy;
@@ -6461,6 +6464,20 @@ make_join_select(JOIN *join,SQL_SELECT *select,COND *cond)
   DBUG_RETURN(0);
 }
 
+
+/**
+  The default implementation of unlock-row method of READ_RECORD,
+  used in all access methods.
+*/
+
+void rr_unlock_row(st_join_table *tab)
+{
+  READ_RECORD *info= &tab->read_record;
+  info->file->unlock_row();
+}
+
+
+
 static void
 make_join_readinfo(JOIN *join, ulonglong options)
 {
@@ -6476,6 +6493,7 @@ make_join_readinfo(JOIN *join, ulonglong options)
     TABLE *table=tab->table;
     tab->read_record.table= table;
     tab->read_record.file=table->file;
+    tab->read_record.unlock_row= rr_unlock_row;
     tab->next_select=sub_select;		/* normal select */
 
     /*
@@ -6521,6 +6539,7 @@ make_join_readinfo(JOIN *join, ulonglong options)
       delete tab->quick;
       tab->quick=0;
       tab->read_first_record= join_read_key;
+      tab->read_record.unlock_row= join_read_key_unlock_row;
       tab->read_record.read_record= join_no_more_records;
       if (table->covering_keys.is_set(tab->ref.key) &&
 	  !table->no_keyread)
@@ -11357,7 +11376,7 @@ evaluate_join_record(JOIN *join, JOIN_TAB *join_tab,
         return NESTED_LOOP_NO_MORE_ROWS;
     }
     else
-      join_tab->read_record.file->unlock_row();
+      join_tab->read_record.unlock_row(join_tab);
   }
   else
   {
@@ -11367,7 +11386,7 @@ evaluate_join_record(JOIN *join, JOIN_TAB *join_tab,
     */
     join->examined_rows++;
     join->thd->row_count++;
-    join_tab->read_record.file->unlock_row();
+    join_tab->read_record.unlock_row(join_tab);
   }
   return NESTED_LOOP_OK;
 }
@@ -11727,18 +11746,55 @@ join_read_key(JOIN_TAB *tab)
       table->status=STATUS_NOT_FOUND;
       return -1;
     }
+    /*
+      Moving away from the current record. Unlock the row
+      in the handler if it did not match the partial WHERE.
+    */
+    if (tab->ref.has_record && tab->ref.use_count == 0)
+    {
+      tab->read_record.file->unlock_row();
+      tab->ref.has_record= FALSE;
+    }
     error=table->file->index_read_map(table->record[0],
                                       tab->ref.key_buff,
                                       make_prev_keypart_map(tab->ref.key_parts),
                                       HA_READ_KEY_EXACT);
     if (error && error != HA_ERR_KEY_NOT_FOUND && error != HA_ERR_END_OF_FILE)
       return report_error(table, error);
+
+    if (! error)
+    {
+      tab->ref.has_record= TRUE;
+      tab->ref.use_count= 1;
+    }
+  }
+  else if (table->status == 0)
+  {
+    DBUG_ASSERT(tab->ref.has_record);
+    tab->ref.use_count++;
   }
   table->null_row=0;
   return table->status ? -1 : 0;
 }
 
 
+/**
+  Since join_read_key may buffer a record, do not unlock
+  it if it was not used in this invocation of join_read_key().
+  Only count locks, thus remembering if the record was left unused,
+  and unlock already when pruning the current value of
+  TABLE_REF buffer.
+  @sa join_read_key()
+*/
+
+static void
+join_read_key_unlock_row(st_join_table *tab)
+{
+  DBUG_ASSERT(tab->ref.use_count);
+  if (tab->ref.use_count)
+    tab->ref.use_count--;
+}
+
 /*
   ref access method implementation: "read_first" function
 
diff --git a/sql/sql_select.h b/sql/sql_select.h
index d2703004df2..c9cd3ecba42 100644
--- a/sql/sql_select.h
+++ b/sql/sql_select.h
@@ -58,6 +58,8 @@ class store_key;
 typedef struct st_table_ref
 {
   bool		key_err;
+  /** True if something was read into buffer in join_read_key.  */
+  bool          has_record;
   uint          key_parts;                ///< num of ...
   uint          key_length;               ///< length of key_buff
   int           key;                      ///< key no
@@ -85,6 +87,11 @@ typedef struct st_table_ref
   table_map	depend_map;		  ///< Table depends on these tables.
   /* null byte position in the key_buf. Used for REF_OR_NULL optimization */
   uchar          *null_ref_key;
+  /*
+    The number of times the record associated with this key was used
+    in the join.
+  */
+  ha_rows       use_count;
 } TABLE_REF;
 
 
diff --git a/sql/structs.h b/sql/structs.h
index a58c18f97c5..2546d241059 100644
--- a/sql/structs.h
+++ b/sql/structs.h
@@ -115,16 +115,22 @@ typedef struct st_reginfo {		/* Extra info about reg */
 } REGINFO;
 
 
-struct st_read_record;				/* For referense later */
 class SQL_SELECT;
 class THD;
 class handler;
+struct st_join_table;
+
+void rr_unlock_row(st_join_table *tab);
 
-typedef struct st_read_record {			/* Parameter to read_record */
+struct READ_RECORD {			/* Parameter to read_record */
+  typedef int (*Read_func)(READ_RECORD*);
+  typedef void (*Unlock_row_func)(st_join_table *);
   struct st_table *table;			/* Head-form */
   handler *file;
   struct st_table **forms;			/* head and ref forms */
-  int (*read_record)(struct st_read_record *);
+
+  Read_func read_record;
+  Unlock_row_func unlock_row;
   THD *thd;
   SQL_SELECT *select;
   uint cache_records;
@@ -136,7 +142,7 @@ typedef struct st_read_record {			/* Parameter to read_record */
   uchar	*cache,*cache_pos,*cache_end,*read_positions;
   IO_CACHE *io_cache;
   bool print_error, ignore_not_found_rows;
-} READ_RECORD;
+};
 
 
 /*