Port r103 from braches/5.0 to trunk.

Fixed a bug on unlock_row. In a unlock_row we may unlock
only the latest lock granted to this transaction to the row.

include/lock0lock.h

@@ -64,14 +64,6 @@ lock_clust_rec_some_has_impl(
 	dict_index_t*	index,	/* in: clustered index */
 	const ulint*	offsets);/* in: rec_get_offsets(rec, index) */
 /*****************************************************************
-Resets the lock bits for a single record. Releases transactions
-waiting for lock requests here. */
-
-void
-lock_rec_reset_and_release_wait(
-/*============================*/
-	rec_t*	rec);	/* in: record whose locks bits should be reset */
-/*****************************************************************
 Makes a record to inherit the locks of another record as gap type
 locks, but does not reset the lock bits of the other record. Also
 waiting lock requests on rec are inherited as GRANTED gap locks. */
@@ -427,6 +419,18 @@ lock_is_on_table(
 /*=============*/
 				/* out: TRUE if there are lock(s) */
 	dict_table_t*	table);	/* in: database table in dictionary cache */
+/*****************************************************************
+Removes a granted record lock of a transaction from the queue and grants
+locks to other transactions waiting in the queue if they now are entitled
+to a lock. */
+
+void
+lock_rec_unlock(
+/*============*/
+	trx_t*	trx,  		/* in: transaction that has set a record
+				lock */
+	rec_t*	rec,		/* in: record */
+	ulint	lock_mode);	/* in: LOCK_S or LOCK_X */
 /*************************************************************************
 Releases a table lock.
 Releases possible other transactions waiting for this lock. */

lock/lock0lock.c

@@ -2386,7 +2386,7 @@ lock_rec_free_all_from_discard_page(
 /*****************************************************************
 Resets the lock bits for a single record. Releases transactions waiting for
 lock requests here. */
-
+static
 void
 lock_rec_reset_and_release_wait(
 /*============================*/
@@ -3748,6 +3748,75 @@ lock_table_dequeue(
 
 /*=========================== LOCK RELEASE ==============================*/
 
+/*****************************************************************
+Removes a granted record lock of a transaction from the queue and grants
+locks to other transactions waiting in the queue if they now are entitled
+to a lock. */
+
+void
+lock_rec_unlock(
+/*============*/
+	trx_t*	trx,  		/* in: transaction that has set a record
+				lock */
+	rec_t*	rec,		/* in: record */
+	ulint	lock_mode)	/* in: LOCK_S or LOCK_X */
+{
+	lock_t* lock;
+	lock_t* release_lock;
+	ulint heap_no;
+
+	ut_ad(trx && rec);
+
+	mutex_enter(&kernel_mutex);
+
+	heap_no = rec_get_heap_no(rec, page_rec_is_comp(rec));
+
+	lock = lock_rec_get_first(rec);
+
+	/* Find the last lock with the same lock_mode and transaction
+	from the record. */
+
+	while (lock != NULL) {
+		if (lock->trx == trx && lock_get_mode(lock) == lock_mode) {
+			release_lock = lock;
+			ut_a(!lock_get_wait(lock));
+		}
+
+		lock = lock_rec_get_next(rec, lock);
+	}
+
+	/* If a record lock is found, release the record lock */
+
+	if(UNIV_LIKELY(release_lock != NULL)) {
+		lock_rec_reset_nth_bit(release_lock, heap_no);
+	} else {
+		mutex_exit(&kernel_mutex);
+		ut_print_timestamp(stderr);
+		fprintf(stderr,
+"  InnoDB: Error: unlock row could not find a %lu mode lock on the record\n",
+			(ulong)lock_mode);
+
+		return;
+	}
+
+	/* Check if we can now grant waiting lock requests */
+
+	lock = lock_rec_get_first(rec);
+
+	while (lock != NULL) {
+		if (lock_get_wait(lock)
+			&& !lock_rec_has_to_wait_in_queue(lock)) {
+
+			/* Grant the lock */
+			lock_grant(lock);
+		}
+
+		lock = lock_rec_get_next(rec, lock);
+	}
+
+	mutex_exit(&kernel_mutex);
+} 
+
 /*************************************************************************
 Releases a table lock.
 Releases possible other transactions waiting for this lock. */

mysql-test/innodb_unsafe_binlog.result

@@ -14,3 +14,22 @@ select ml.* from t1 as ml left join t2 as mm on (mm.id=ml.id)
 where mm.id is null lock in share mode;
 id	f_id	f
 drop table t1,t2;
+create table t1(a int not null, b int, primary key(a)) engine=innodb;
+insert into t1 values(1,1),(2,2),(3,1),(4,2),(5,1),(6,2);
+commit;
+set autocommit = 0;
+select * from t1 lock in share mode;
+a	b
+1	1
+2	2
+3	1
+4	2
+5	1
+6	2
+update t1 set b = 5 where b = 1;
+set autocommit = 0;
+select * from t1 where a = 2 and b = 2 for update;
+ERROR HY000: Lock wait timeout exceeded; try restarting transaction
+commit;
+commit;
+drop table t1;

mysql-test/innodb_unsafe_binlog.test

 -- source include/have_innodb.inc
 #
-# Note that these test work only on 5.0 because these test need
-# innodb_locks_unsafe_for_binlog option implemented.
+# Note that these tests uses a innodb_locks_unsafe_for_binlog option.
 #
 #
 # Test cases for a bug #15650
@@ -24,3 +23,33 @@ WHERE mm.id IS NULL;
 select ml.* from t1 as ml left join t2 as mm on (mm.id=ml.id)
 where mm.id is null lock in share mode;
 drop table t1,t2;
+
+#
+# Test case for unlock row bug where unlock releases all locks granted for
+# a row. Only the latest lock should be released.
+#
+
+connect (a,localhost,root,,);
+connect (b,localhost,root,,);
+connection a;
+create table t1(a int not null, b int, primary key(a)) engine=innodb;
+insert into t1 values(1,1),(2,2),(3,1),(4,2),(5,1),(6,2);
+commit;
+set autocommit = 0; 
+select * from t1 lock in share mode;
+update t1 set b = 5 where b = 1;
+connection b;
+set autocommit = 0;
+#
+# S-lock to records (2,2),(4,2), and (6,2) should not be released in a update
+#
+--error 1205
+select * from t1 where a = 2 and b = 2 for update;
+connection a;
+commit;
+connection b;
+commit;
+drop table t1;
+disconnect a;
+disconnect b;
+

row/row0mysql.c

@@ -1486,11 +1486,7 @@ row_unlock_for_mysql(
 
 		rec = btr_pcur_get_rec(pcur);
 
-		mutex_enter(&kernel_mutex);
-
-		lock_rec_reset_and_release_wait(rec);
-
-		mutex_exit(&kernel_mutex);
+		lock_rec_unlock(trx, rec, prebuilt->select_lock_type);
 
 		mtr_commit(&mtr);
 
@@ -1520,11 +1516,7 @@ row_unlock_for_mysql(
 
 		rec = btr_pcur_get_rec(clust_pcur);
 
-		mutex_enter(&kernel_mutex);
-
-		lock_rec_reset_and_release_wait(rec);
-
-		mutex_exit(&kernel_mutex);
+		lock_rec_unlock(trx, rec, prebuilt->select_lock_type);
 
 		mtr_commit(&mtr);
 	}