Take X-lock for duplicate keys in REPLACE command.

innobase/lock/lock0lock.c

@@ -755,9 +755,13 @@ lock_rec_has_to_wait(
 	ulint	type_mode,/* in: precise mode of the new lock to set:
 			LOCK_S or LOCK_X, possibly ORed to
 			LOCK_GAP or LOCK_REC_NOT_GAP, LOCK_INSERT_INTENTION */
-	lock_t*	lock2)	/* in: another record lock; NOTE that it is assumed
+	lock_t*	lock2,	/* in: another record lock; NOTE that it is assumed
 			that this has a lock bit set on the same record as
 			in the new lock we are setting */
+        ibool lock_is_on_supremum)  /* in: TRUE if we are setting the lock
+                                       on the 'supremum' record of an index page: we know
+                                       then that the lock request is really for a 'gap' type
+                                       lock */
 {
 	ut_ad(trx && lock2);
 	ut_ad(lock_get_type(lock2) == LOCK_REC);
@@ -769,6 +773,15 @@ lock_rec_has_to_wait(
 		/* We have somewhat complex rules when gap type record locks
 		cause waits */
 
+	        if( lock_is_on_supremum && !(type_mode & LOCK_INSERT_INTENTION))
+                {
+		        /* Lock on the supremum record is really a 'gap' type lock
+                           and gap type lock do not need to wait if it
+                           is not LOCK_INSERT_INTENSION type lock */
+
+                       return(FALSE);
+                }
+
 		if ((type_mode & LOCK_REC_NOT_GAP)
 						&& lock_rec_get_gap(lock2)) {
 			/* Lock on just the record does not need to wait for
@@ -828,9 +841,10 @@ lock_has_to_wait(
 				     		lock_get_mode(lock2))) {
 		if (lock_get_type(lock1) == LOCK_REC) {
 			ut_ad(lock_get_type(lock2) == LOCK_REC);
-				
+
+
 			return(lock_rec_has_to_wait(lock1->trx,
-						lock1->type_mode, lock2));
+			       lock1->type_mode, lock2,(ibool)lock_rec_get_nth_bit(lock1,1)));
 		}
 
 		return(TRUE);
@@ -1419,7 +1433,7 @@ lock_rec_other_has_conflicting(
 	lock = lock_rec_get_first(rec);
 
 	while (lock) {
-		if (lock_rec_has_to_wait(trx, mode, lock)) {
+		if (lock_rec_has_to_wait(trx, mode, lock, (ibool)page_rec_is_supremum(rec))) {
 
 			return(lock);
 		}

innobase/row/row0ins.c

@@ -1022,6 +1022,33 @@ row_ins_set_shared_rec_lock(
 
 	return(err);
 }
+
+/*************************************************************************
+Sets a exclusive lock on a record. Used in locking possible duplicate key
+records */
+static
+ulint
+row_ins_set_exclusive_rec_lock(
+/*========================*/
+				/* out: DB_SUCCESS or error code */
+	ulint		type, 	/* in: LOCK_ORDINARY, LOCK_GAP, or
+				LOCK_REC_NOT_GAP type lock */
+	rec_t*		rec,	/* in: record */
+	dict_index_t*	index,	/* in: index */
+	que_thr_t*	thr)	/* in: query thread */	
+{
+	ulint	err;
+
+	if (index->type & DICT_CLUSTERED) {
+		err = lock_clust_rec_read_check_and_lock(0, rec, index, LOCK_X,
+								type, thr);
+	} else {
+		err = lock_sec_rec_read_check_and_lock(0, rec, index, LOCK_X,
+								type, thr);
+	}
+
+	return(err);
+}
 	
 /*******************************************************************
 Checks if foreign key constraint fails for an index entry. Sets shared locks
@@ -1451,6 +1478,8 @@ row_ins_scan_sec_index_for_duplicate(
 	ulint		err		= DB_SUCCESS;
 	ibool		moved;
 	mtr_t		mtr;
+        trx_t           *trx;
+        ibool           success;
 
 	n_unique = dict_index_get_n_unique(index);
 
@@ -1488,8 +1517,29 @@ row_ins_scan_sec_index_for_duplicate(
 				
 		/* Try to place a lock on the index record */
 
-		err = row_ins_set_shared_rec_lock(LOCK_ORDINARY, rec, index,
-									thr);
+                /* The manual defines the REPLACE semantics that it 
+                   is either an INSERT or DELETE(s) for duplicate key
+                   + INSERT. Therefore, we should take X-lock for
+                   duplicates.
+		*/
+
+		trx = thr_get_trx(thr);*/* Get Transaction */
+
+                /* Is the first word in MySQL query REPLACE ? */
+
+		ut_ad(trx)
+		dict_accept(*trx->mysql_query_str, "REPLACE", &success);
+
+                if (success) {
+
+		    err = row_ins_set_exclusive_rec_lock(
+			LOCK_ORDINARY,rec,index,thr);
+		} else {
+
+		      err = row_ins_set_shared_rec_lock(
+                         LOCK_ORDINARY, rec, index,thr);
+		}
+
 
 		if (err != DB_SUCCESS) {
 
@@ -1556,6 +1606,7 @@ row_ins_duplicate_error_in_clust(
 	page_t*	page;
 	ulint	n_unique;
 	trx_t*	trx	= thr_get_trx(thr);
+        ibool   success;
 
 	UT_NOT_USED(mtr);
 	
@@ -1588,9 +1639,28 @@ row_ins_duplicate_error_in_clust(
 			is needed in logical logging of MySQL to make
 			sure that in roll-forward we get the same duplicate
 			errors as in original execution */
-		
-			err = row_ins_set_shared_rec_lock(LOCK_REC_NOT_GAP,
-						rec, cursor->index, thr);
+
+                        /* The manual defines the REPLACE semantics that it 
+                           is either an INSERT or DELETE(s) for duplicate key
+                           + INSERT. Therefore, we should take X-lock for
+                           duplicates.
+		        */
+
+                        
+                        /* Is the first word in MySQL query REPLACE ? */
+
+		         dict_accept(*trx->mysql_query_str, "REPLACE", &success);
+
+                        if (success) {
+
+		            err = row_ins_set_exclusive_rec_lock(
+			       LOCK_REC_NOT_GAP,rec,cursor->index,thr);
+		        } else {
+		  
+			    err = row_ins_set_shared_rec_lock(
+                               LOCK_REC_NOT_GAP,rec, cursor->index, thr);
+			} 
+
 			if (err != DB_SUCCESS) {
 					
 				return(err);
@@ -1611,8 +1681,28 @@ row_ins_duplicate_error_in_clust(
 
 		if (rec != page_get_supremum_rec(page)) {
 
-			err = row_ins_set_shared_rec_lock(LOCK_REC_NOT_GAP,
-						rec, cursor->index, thr);
+
+                        /* The manual defines the REPLACE semantics that it 
+                           is either an INSERT or DELETE(s) for duplicate key
+                           + INSERT. Therefore, we should take X-lock for
+                           duplicates.
+		        */
+
+                        
+                        /* Is the first word in MySQL query REPLACE ? */
+
+                        dict_accept(*trx->mysql_query_str, "REPLACE", &success);
+
+                        if (success) {
+
+		            err = row_ins_set_exclusive_rec_lock(
+			       LOCK_REC_NOT_GAP,rec,cursor->index,thr);
+		        } else {
+
+			    err = row_ins_set_shared_rec_lock(
+                               LOCK_REC_NOT_GAP,rec, cursor->index, thr);
+                        }
+
 			if (err != DB_SUCCESS) {
 					
 				return(err);