1 files changed, 160 insertions, 65 deletions

diff --git a/innobase/row/row0ins.c b/innobase/row/row0ins.c
index 60dbf059673..5ca1ee51cbd 100644
--- a/innobase/row/row0ins.c
+++ b/innobase/row/row0ins.c
@@ -50,6 +50,20 @@ innobase_invalidate_query_cache(
 	ulint	full_name_len);	/* in: full name length where also the null
 				chars count */
 
+/**********************************************************************
+This function returns true if 
+
+1) SQL-query in the current thread
+is either REPLACE or LOAD DATA INFILE REPLACE. 
+
+2) SQL-query in the current thread
+is INSERT ON DUPLICATE KEY UPDATE.
+
+NOTE that /mysql/innobase/row/row0ins.c must contain the 
+prototype for this function ! */
+
+ibool
+innobase_query_is_update(void);
 
 /*************************************************************************
 Creates an insert node struct. */
@@ -252,7 +266,7 @@ row_ins_sec_index_entry_by_modify(
 	heap = mem_heap_create(1024);
 	
 	update = row_upd_build_sec_rec_difference_binary(cursor->index,
-							entry, rec, heap); 
+				entry, rec, thr_get_trx(thr), heap);
 	if (mode == BTR_MODIFY_LEAF) {
 		/* Try an optimistic updating of the record, keeping changes
 		within the page */
@@ -316,7 +330,7 @@ row_ins_clust_index_entry_by_modify(
 	roll_ptr */
 	
 	update = row_upd_build_difference_binary(cursor->index, entry, ext_vec,
-						n_ext_vec, rec, heap); 
+			n_ext_vec, rec, thr_get_trx(thr), heap);
 	if (mode == BTR_MODIFY_LEAF) {
 		/* Try optimistic updating of the record, keeping changes
 		within the page */
@@ -543,7 +557,7 @@ static
 void
 row_ins_foreign_report_err(
 /*=======================*/
-	char*		errstr,		/* in: error string from the viewpoint
+	const char*	errstr,		/* in: error string from the viewpoint
 					of the parent table */
 	que_thr_t*	thr,		/* in: query thread whose run_node
 					is an update node */
@@ -554,29 +568,30 @@ row_ins_foreign_report_err(
 					table */
 {
 	FILE*	ef	= dict_foreign_err_file;
+	trx_t*	trx	= thr_get_trx(thr);
 
 	mutex_enter(&dict_foreign_err_mutex);
 	rewind(ef);
 	ut_print_timestamp(ef);
 	fputs(" Transaction:\n", ef);
-	trx_print(ef, thr_get_trx(thr));
+	trx_print(ef, trx);
 
 	fputs("Foreign key constraint fails for table ", ef);
-	ut_print_name(ef, foreign->foreign_table_name);
+	ut_print_name(ef, trx, foreign->foreign_table_name);
 	fputs(":\n", ef);
-	dict_print_info_on_foreign_key_in_create_format(ef, foreign);
+	dict_print_info_on_foreign_key_in_create_format(ef, trx, foreign);
 	putc('\n', ef);
 	fputs(errstr, ef);
 	fputs(" in parent table, in index ", ef);
-	ut_print_name(ef, foreign->referenced_index->name);
+	ut_print_name(ef, trx, foreign->referenced_index->name);
 	if (entry) {
 		fputs(" tuple:\n", ef);
 		dtuple_print(ef, entry);
 	}
 	fputs("\nBut in child table ", ef);
-	ut_print_name(ef, foreign->foreign_table_name);
+	ut_print_name(ef, trx, foreign->foreign_table_name);
 	fputs(", in index ", ef);
-	ut_print_name(ef, foreign->foreign_index->name);
+	ut_print_name(ef, trx, foreign->foreign_index->name);
 	if (rec) {
 		fputs(", there is a record:\n", ef);
 		rec_print(ef, rec);
@@ -612,19 +627,19 @@ row_ins_foreign_report_add_err(
 	fputs(" Transaction:\n", ef);
 	trx_print(ef, trx);
 	fputs("Foreign key constraint fails for table ", ef);
-	ut_print_name(ef, foreign->foreign_table_name);
+	ut_print_name(ef, trx, foreign->foreign_table_name);
 	fputs(":\n", ef);
-	dict_print_info_on_foreign_key_in_create_format(ef, foreign);
+	dict_print_info_on_foreign_key_in_create_format(ef, trx, foreign);
 	fputs("\nTrying to add in child table, in index ", ef);
-	ut_print_name(ef, foreign->foreign_index->name);
+	ut_print_name(ef, trx, foreign->foreign_index->name);
 	if (entry) {
 		fputs(" tuple:\n", ef);
 		dtuple_print(ef, entry);
 	}
 	fputs("\nBut in parent table ", ef);
-	ut_print_name(ef, foreign->referenced_table_name);
+	ut_print_name(ef, trx, foreign->referenced_table_name);
 	fputs(", in index ", ef);
-	ut_print_name(ef, foreign->referenced_index->name);
+	ut_print_name(ef, trx, foreign->referenced_index->name);
 	fputs(",\nthe closest match we can find is record:\n", ef);
 	if (rec && page_rec_is_supremum(rec)) {
 		/* If the cursor ended on a supremum record, it is better
@@ -704,9 +719,13 @@ row_ins_foreign_check_on_constraint(
 	ulint		n_to_update;
 	ulint		err;
 	ulint		i;
+	trx_t*		trx;
+
 	
 	ut_a(thr && foreign && pcur && mtr);
 
+	trx = thr_get_trx(thr);
+
 	/* Since we are going to delete or update a row, we have to invalidate
 	the MySQL query cache for table */
 
@@ -718,7 +737,7 @@ row_ins_foreign_check_on_constraint(
 			(DICT_FOREIGN_ON_DELETE_CASCADE
 			 | DICT_FOREIGN_ON_DELETE_SET_NULL))) {
 
-		row_ins_foreign_report_err((char*)"Trying to delete",
+		row_ins_foreign_report_err("Trying to delete",
 					thr, foreign,
 					btr_pcur_get_rec(pcur), entry);
 
@@ -731,7 +750,7 @@ row_ins_foreign_check_on_constraint(
 
 		/* This is an UPDATE */
 			 
-		row_ins_foreign_report_err((char*)"Trying to update",
+		row_ins_foreign_report_err("Trying to update",
 					thr, foreign,
 					btr_pcur_get_rec(pcur), entry);
 
@@ -792,7 +811,7 @@ row_ins_foreign_check_on_constraint(
 	        err = DB_ROW_IS_REFERENCED;
 
 		row_ins_foreign_report_err(
-(char*)"Trying an update, possibly causing a cyclic cascaded update\n"
+"Trying an update, possibly causing a cyclic cascaded update\n"
 "in the child table,", thr, foreign, btr_pcur_get_rec(pcur), entry);
 
 		goto nonstandard_exit_func;
@@ -845,7 +864,7 @@ row_ins_foreign_check_on_constraint(
 			fputs(
 			"InnoDB: error in cascade of a foreign key op\n"
 			"InnoDB: ", stderr);
-			dict_index_name_print(stderr, index);
+			dict_index_name_print(stderr, trx, index);
 
 			fputs("\n"
 				"InnoDB: record ", stderr);
@@ -927,7 +946,7 @@ row_ins_foreign_check_on_constraint(
 		        err = DB_ROW_IS_REFERENCED;
 
 			row_ins_foreign_report_err(
-(char*)"Trying a cascaded update where the updated value in the child\n"
+"Trying a cascaded update where the updated value in the child\n"
 "table would not fit in the length of the column, or the value would\n"
 "be NULL and the column is declared as not NULL in the child table,",
 			thr, foreign, btr_pcur_get_rec(pcur), entry);
@@ -1037,6 +1056,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
@@ -1064,7 +1110,6 @@ row_ins_check_foreign_constraint(
 	dict_table_t*	check_table;
 	dict_index_t*	check_index;
 	ulint		n_fields_cmp;
-	ibool		unique_search;
 	rec_t*		rec;
 	btr_pcur_t	pcur;
 	ibool		moved;
@@ -1072,6 +1117,7 @@ row_ins_check_foreign_constraint(
 	ulint		err;
 	ulint		i;
 	mtr_t		mtr;
+	trx_t*		trx = thr_get_trx(thr);
 
 run_again:
 #ifdef UNIV_SYNC_DEBUG
@@ -1080,7 +1126,7 @@ run_again:
 
 	err = DB_SUCCESS;
 
-	if (thr_get_trx(thr)->check_foreigns == FALSE) {
+	if (trx->check_foreigns == FALSE) {
 		/* The user has suppressed foreign key checks currently for
 		this session */
 
@@ -1131,26 +1177,26 @@ run_again:
 		check_index = foreign->foreign_index;
 	}
 
-	if (check_table == NULL) {
+	if (check_table == NULL || check_table->ibd_file_missing) {
 		if (check_ref) {
 			FILE*	ef = dict_foreign_err_file;
 			mutex_enter(&dict_foreign_err_mutex);
 			rewind(ef);
 			ut_print_timestamp(ef);
 			fputs(" Transaction:\n", ef);
-			trx_print(ef, thr_get_trx(thr));
+			trx_print(ef, trx);
 			fputs("Foreign key constraint fails for table ", ef);
-			ut_print_name(ef, foreign->foreign_table_name);
+			ut_print_name(ef, trx, foreign->foreign_table_name);
 			fputs(":\n", ef);
 			dict_print_info_on_foreign_key_in_create_format(ef,
-				foreign);
+					trx, foreign);
 			fputs("\nTrying to add to index ", ef);
-			ut_print_name(ef, foreign->foreign_index->name);
+			ut_print_name(ef, trx, foreign->foreign_index->name);
 			fputs(" tuple:\n", ef);
 			dtuple_print(ef, entry);
 			fputs("\nBut the parent table ", ef);
-			ut_print_name(ef, foreign->referenced_table_name);
-			fputs(" does not currently exist!\n", ef);
+			ut_print_name(ef, trx, foreign->referenced_table_name);
+		fputs("\nor its .ibd file does not currently exist!\n", ef);
 			mutex_exit(&dict_foreign_err_mutex);
 
 			return(DB_NO_REFERENCED_ROW);
@@ -1181,14 +1227,6 @@ run_again:
 
 	dtuple_set_n_fields_cmp(entry, foreign->n_fields);
 
-	if (dict_index_get_n_unique(check_index) <= foreign->n_fields) {
-		/* We can just set a LOCK_REC_NOT_GAP type lock */
-	
-		unique_search = TRUE;
-	} else {
-		unique_search = FALSE;
-	}
-
 	btr_pcur_open(check_index, entry, PAGE_CUR_GE,
 					BTR_SEARCH_LEAF, &pcur, &mtr);
 
@@ -1226,17 +1264,13 @@ run_again:
 					break;
 				}
 			} else {
-				/* Found a matching record */
+				/* Found a matching record. Lock only
+				a record because we can allow inserts
+				into gaps */
 				
-				if (unique_search) {
-					err = row_ins_set_shared_rec_lock(
-							LOCK_REC_NOT_GAP,
-							rec, check_index, thr);
-				} else {
-					err = row_ins_set_shared_rec_lock(
-							LOCK_ORDINARY,
-							rec, check_index, thr);
-				}
+				err = row_ins_set_shared_rec_lock(
+						LOCK_REC_NOT_GAP,
+						rec, check_index, thr);
 				
 				if (err != DB_SUCCESS) {
 
@@ -1262,7 +1296,7 @@ run_again:
 					}
 				} else {
 					row_ins_foreign_report_err(
-					(char*)"Trying to delete or update",
+						"Trying to delete or update",
 						thr, foreign, rec, entry);
 
 					err = DB_ROW_IS_REFERENCED;
@@ -1282,7 +1316,7 @@ run_again:
 			if (check_ref) {			
 				err = DB_NO_REFERENCED_ROW;
 				row_ins_foreign_report_add_err(
-					thr_get_trx(thr), foreign, rec, entry);
+					trx, foreign, rec, entry);
 			} else {
 				err = DB_SUCCESS;
 			}
@@ -1298,7 +1332,7 @@ next_rec:
 			if (check_ref) {			
 				rec = btr_pcur_get_rec(&pcur);
 				row_ins_foreign_report_add_err(
-					thr_get_trx(thr), foreign, rec, entry);
+					trx, foreign, rec, entry);
 				err = DB_NO_REFERENCED_ROW;
 			} else {
 				err = DB_SUCCESS;
@@ -1317,18 +1351,18 @@ next_rec:
 
 do_possible_lock_wait:
 	if (err == DB_LOCK_WAIT) {
-		thr_get_trx(thr)->error_state = err;
+		trx->error_state = err;
 
 		que_thr_stop_for_mysql(thr);
 
 		srv_suspend_mysql_thread(thr);
 	
-		if (thr_get_trx(thr)->error_state == DB_SUCCESS) {
+		if (trx->error_state == DB_SUCCESS) {
 
 		        goto run_again;
 		}
 
-		err = thr_get_trx(thr)->error_state;
+		err = trx->error_state;
 	}
 
 	return(err);
@@ -1492,7 +1526,8 @@ row_ins_scan_sec_index_for_duplicate(
 	ulint		err		= DB_SUCCESS;
 	ibool		moved;
 	mtr_t		mtr;
-
+	trx_t*		trx;
+	
 	n_unique = dict_index_get_n_unique(index);
 
 	/* If the secondary index is unique, but one of the fields in the
@@ -1529,8 +1564,23 @@ 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);
+
+		if (innobase_query_is_update()) {
+
+			/* If the SQL-query will update or replace
+			duplicate key we will take X-lock for 
+			duplicates ( REPLACE, LOAD DATAFILE REPLACE, 
+			INSERT ON DUPLICATE KEY UPDATE). */
+			
+			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) {
 
@@ -1629,9 +1679,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);
+
+			if (innobase_query_is_update()) {
+
+				/* If the SQL-query will update or replace
+				duplicate key we will take X-lock for 
+				duplicates ( REPLACE, LOAD DATAFILE REPLACE, 
+				INSERT ON DUPLICATE KEY UPDATE). */
+				
+				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);
@@ -1652,8 +1717,24 @@ 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);
+
+			if (innobase_query_is_update()) {
+
+				/* If the SQL-query will update or replace
+				duplicate key we will take X-lock for 
+				duplicates ( REPLACE, LOAD DATAFILE REPLACE, 
+				INSERT ON DUPLICATE KEY UPDATE). */
+
+				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);
@@ -1747,6 +1828,7 @@ row_ins_index_entry_low(
 	ulint		modify = 0; /* remove warning */
 	rec_t*		insert_rec;
 	rec_t*		rec;
+	rec_t*		first_rec;
 	ulint		err;
 	ulint		n_unique;
 	big_rec_t*	big_rec			= NULL;
@@ -1779,6 +1861,14 @@ row_ins_index_entry_low(
 		goto function_exit;
 	}	
 					
+	first_rec = page_rec_get_next(page_get_infimum_rec(
+			buf_frame_align(btr_cur_get_rec(&cursor))));
+
+	if (!page_rec_is_supremum(first_rec)) {
+		ut_a((rec_get_n_fields(first_rec))
+					== dtuple_get_n_fields(entry));
+	}
+
 	n_unique = dict_index_get_n_unique(index);
 
 	if (index->type & DICT_UNIQUE && (cursor.up_match >= n_unique
@@ -1945,6 +2035,7 @@ row_ins_index_entry_set_vals(
 	dfield_t*	row_field;
 	ulint		n_fields;
 	ulint		i;
+	dtype_t*        cur_type;
 
 	ut_ad(entry && row);
 
@@ -1958,10 +2049,14 @@ row_ins_index_entry_set_vals(
 
 		/* Check column prefix indexes */
 		if (ind_field->prefix_len > 0
-		    && dfield_get_len(row_field) != UNIV_SQL_NULL
-		    && dfield_get_len(row_field) > ind_field->prefix_len) {
-		    
-		        field->len = ind_field->prefix_len;
+		    && dfield_get_len(row_field) != UNIV_SQL_NULL) {
+
+			cur_type = dict_col_get_type(
+				dict_field_get_col(ind_field));
+
+			field->len = dtype_get_at_most_n_mbchars(cur_type,
+				  ind_field->prefix_len,
+				  dfield_get_len(row_field), row_field->data);
 		} else {
 		        field->len = row_field->len;
 		}
@@ -2246,4 +2341,4 @@ error_handling:
 	}
 
 	return(thr);
-} 
+}