Fixed a bug in UPDATE statement with no index column in where condition

locks all rows (BUG #3300). When using innobase_locks_unsafe_for_binlog
option InnoDB does not take locks for those rows which do not
belong to the result set or werent changed by the query. This fix removes
unnecessary locks also from SELECT and DELETE queries.

Author: jan@hundin.mysql.fi

innobase/include/row0mysql.h

@@ -233,6 +233,17 @@ row_update_for_mysql(
 					the MySQL format */
 	row_prebuilt_t*	prebuilt);	/* in: prebuilt struct in MySQL
 					handle */
+
+/*************************************************************************
+Does an unlock of a row for MySQL. */
+
+int
+row_unlock_for_mysql(
+/*=================*/
+					/* out: error code or DB_SUCCESS */
+	row_prebuilt_t*	prebuilt);	/* in: prebuilt struct in MySQL
+					handle */
+
 /*************************************************************************
 Creates an query graph node of 'update' type to be used in the MySQL
 interface. */

innobase/include/trx0trx.h

@@ -423,6 +423,8 @@ struct trx_struct{
 	lock_t*		auto_inc_lock;	/* possible auto-inc lock reserved by
 					the transaction; note that it is also
 					in the lock list trx_locks */
+	ibool		trx_create_lock;/* this is TRUE if we have created a
+					new lock for a record accessed */
 	ulint		n_lock_table_exp;/* number of explicit table locks
 					(LOCK TABLES) reserved by the
 					transaction, stored in trx_locks */

innobase/lock/lock0lock.c

@@ -1617,6 +1617,9 @@ lock_rec_create(
 
 	HASH_INSERT(lock_t, hash, lock_sys->rec_hash,
 					lock_rec_fold(space, page_no), lock); 
+	/* Note that we have create a new lock */
+	trx->trx_create_lock = TRUE;
+
 	if (type_mode & LOCK_WAIT) {
 
 		lock_set_lock_and_trx_wait(lock, trx);
@@ -1791,6 +1794,15 @@ lock_rec_add_to_queue(
 
 	if (similar_lock && !somebody_waits && !(type_mode & LOCK_WAIT)) {
 
+		/* If the nth bit of a record lock is already set then we
+		do not set a new lock bit, otherwice we set */
+
+		if (lock_rec_get_nth_bit(similar_lock, heap_no)) {
+			trx->trx_create_lock = FALSE;
+		} else {
+			trx->trx_create_lock = TRUE;
+		}
+
 		lock_rec_set_nth_bit(similar_lock, heap_no);
 
 		return(similar_lock);
@@ -1822,6 +1834,7 @@ lock_rec_lock_fast(
 {
 	lock_t*	lock;
 	ulint	heap_no;
+	trx_t*  trx;
 
 #ifdef UNIV_SYNC_DEBUG
 	ut_ad(mutex_own(&kernel_mutex));
@@ -1840,9 +1853,12 @@ lock_rec_lock_fast(
 
 	lock = lock_rec_get_first_on_page(rec);
 
+	trx = thr_get_trx(thr);
+	trx->trx_create_lock = FALSE;
+
 	if (lock == NULL) {
 		if (!impl) {
-			lock_rec_create(mode, rec, index, thr_get_trx(thr));
+			lock_rec_create(mode, rec, index, trx);
 		}
 
 		return(TRUE);
@@ -1853,13 +1869,23 @@ lock_rec_lock_fast(
 		return(FALSE);
 	}
 
-	if (lock->trx != thr_get_trx(thr)
+	if (lock->trx != trx
 				|| lock->type_mode != (mode | LOCK_REC)
 				|| lock_rec_get_n_bits(lock) <= heap_no) {
 	    	return(FALSE);
 	}
 
 	if (!impl) {
+
+		/* If the nth bit of a record lock is already set then we
+		do not set a new lock bit, otherwice we set */
+
+		if (lock_rec_get_nth_bit(lock, heap_no)) {
+			trx->trx_create_lock = FALSE;
+		} else {
+			trx->trx_create_lock = TRUE;
+		}
+
 		lock_rec_set_nth_bit(lock, heap_no);
 	}
 

innobase/row/row0mysql.c

@@ -1186,6 +1186,57 @@ row_update_for_mysql(
 	return((int) err);
 }
 
+/*************************************************************************
+Does an unlock of a row for MySQL. */
+
+int
+row_unlock_for_mysql(
+/*=================*/
+					/* out: error code or DB_SUCCESS */
+	row_prebuilt_t*	prebuilt)	/* in: prebuilt struct in MySQL
+					handle */
+{
+	rec_t*		rec;
+	btr_pcur_t*	cur		= prebuilt->pcur;
+	trx_t*		trx		= prebuilt->trx;
+	mtr_t           mtr;
+	
+	ut_ad(prebuilt && trx);
+	ut_ad(trx->mysql_thread_id == os_thread_get_curr_id());
+		
+	trx->op_info = "unlock_row";
+	
+	if (srv_locks_unsafe_for_binlog) {
+		if (trx->trx_create_lock == TRUE) {
+
+			mtr_start(&mtr);
+			
+			/* Restore a cursor position and find a record */
+			btr_pcur_restore_position(BTR_SEARCH_LEAF, cur, &mtr);
+			rec = btr_pcur_get_rec(cur);
+
+			if (rec) {
+
+				lock_rec_reset_and_release_wait(rec);
+			} else {
+				fputs("InnoDB: Error: "
+				      "Record for the lock not found\n",
+				      stderr);
+				mem_analyze_corruption((byte*) trx);
+				ut_error;
+			}
+
+			trx->trx_create_lock = FALSE;
+			mtr_commit(&mtr);
+		}
+		
+	}
+			
+	trx->op_info = "";
+	
+	return(DB_SUCCESS);
+}
+
 /**************************************************************************
 Does a cascaded delete or set null in a foreign key operation. */
 

sql/ha_innodb.cc

@@ -2690,6 +2690,32 @@ ha_innobase::delete_row(
 	DBUG_RETURN(error);
 }
 
+/**************************************************************************
+Deletes a lock set to a row */
+
+void
+ha_innobase::unlock_row(void)
+/*=========================*/
+{
+	row_prebuilt_t*	prebuilt = (row_prebuilt_t*) innobase_prebuilt;
+
+	DBUG_ENTER("ha_innobase::unlock_row");
+
+	ut_ad(prebuilt->trx ==
+		(trx_t*) current_thd->transaction.all.innobase_tid);
+
+	if (last_query_id != user_thd->query_id) {
+		ut_print_timestamp(stderr);
+		fprintf(stderr,
+"  InnoDB: Error: last_query_id is %lu != user_thd_query_id is %lu\n",
+			(ulong)last_query_id, (ulong)user_thd->query_id);
+		mem_analyze_corruption((byte *) prebuilt->trx);
+		ut_error;
+	}
+
+	row_unlock_for_mysql(prebuilt);
+}
+
 /**********************************************************************
 Initializes a handle to use an index. */
 

sql/ha_innodb.h

@@ -120,6 +120,7 @@ class ha_innobase: public handler
   	int write_row(byte * buf);
   	int update_row(const byte * old_data, byte * new_data);
   	int delete_row(const byte * buf);
+	void unlock_row();
 
   	int index_init(uint index);
   	int index_end();