Merge jlindstrom@build.mysql.com:/home/bk/mysql-4.1

into hundin.mysql.fi:/home/jan/mysql-4.1

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_GAP))
+			&& !(type_mode & LOCK_INSERT_INTENTION)) {
+			/* Gap type locks without LOCK_INSERT_INTENTION flag
+			do not need to wait for anything. This is because different 
+			users can have conflicting lock types on gaps. */
+						  
+			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,13 @@ lock_has_to_wait(
 				     		lock_get_mode(lock2))) {
 		if (lock_get_type(lock1) == LOCK_REC) {
 			ut_ad(lock_get_type(lock2) == LOCK_REC);
-				
+
+			/* If this lock request is for a supremum record
+			then the second bit on the lock bitmap is set */
+			
 			return(lock_rec_has_to_wait(lock1->trx,
-						lock1->type_mode, lock2));
+						lock1->type_mode, lock2,
+						lock_rec_get_nth_bit(lock1,1)));
 		}
 
 		return(TRUE);
@@ -1419,7 +1436,8 @@ 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,
+			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,24 @@ 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);
+		trx = thr_get_trx(thr);      
+		ut_ad(trx);
+		dict_accept(*trx->mysql_query_str, "REPLACE", &success);
+
+		if (success) {
+
+			/* 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 */
+			
+			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 +1601,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 +1634,24 @@ 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);
+
+			dict_accept(*trx->mysql_query_str, "REPLACE", &success);
+
+			if (success) {
+
+				/* 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 */
+				
+				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 +1672,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);
@@ -2214,4 +2295,4 @@ row_ins_step(
 	}
 
 	return(thr);
-} 
+}