Merge of innobase changes to xtradb.

storage/xtradb/handler/ha_innodb.cc

@@ -5290,18 +5290,21 @@ innobase_mysql_cmp(
 }
 #ifdef WITH_WSREP
 extern "C" UNIV_INTERN
-void
+int
 wsrep_innobase_mysql_sort(
 /*===============*/
 					/* out: str contains sort string */
 	int		mysql_type,	/* in: MySQL type */
 	uint		charset_number,	/* in: number of the charset */
 	unsigned char*	str,		/* in: data field */
-	unsigned int	str_length)	/* in: data field length,
+	unsigned int	str_length,	/* in: data field length,
 					not UNIV_SQL_NULL */
+	unsigned int	buf_length)	/* in: total str buffer length */
+
 {
 	CHARSET_INFO*		charset;
 	enum_field_types	mysql_tp;
+	int ret_length =	str_length;
 
 	DBUG_ASSERT(str_length != UNIV_SQL_NULL);
 
@@ -5345,9 +5348,29 @@ wsrep_innobase_mysql_sort(
 		ut_a(str_length <= tmp_length);
 		memcpy(tmp_str, str, str_length);
 
-		tmp_length = charset->coll->strnxfrm(charset, str, str_length,
+		if (wsrep_protocol_version < 3) {
+			tmp_length = charset->coll->strnxfrm(
+				charset, str, str_length,
 				tmp_str, str_length);
 			DBUG_ASSERT(tmp_length <= str_length);
+		} else {
+			/* strnxfrm will expand the destination string,
+			   protocols < 3 truncated the sorted sring
+			   protocols > 3 gets full sorted sring
+			*/
+		  	/* 5.5 strnxfrm pads the tail with spaces and
+			   always returns the full destination buffer lenght
+			   we cannot know how many characters were converted
+			   using 2 * str length here as best guess
+			*/
+			uint dst_length = (str_length * 2 < tmp_length) ? 
+				(str_length * 2) : tmp_length; 
+			tmp_length = charset->coll->strnxfrm(
+				charset, str, dst_length,
+				tmp_str, str_length);
+			DBUG_ASSERT(tmp_length <= buf_length);
+			ret_length = tmp_length;
+		}
  
 		break;
 	}
@@ -5375,7 +5398,7 @@ wsrep_innobase_mysql_sort(
 		break;
 	}
 
-	return;
+	return ret_length;
 }
 #endif // WITH_WSREP
 /**************************************************************//**
@@ -5624,8 +5647,9 @@ wsrep_store_key_val_for_row(
 			}
 
 			memcpy(sorted, data, true_len);
-			wsrep_innobase_mysql_sort(
-			       mysql_type, cs->number, sorted, true_len);
+			true_len = wsrep_innobase_mysql_sort(
+				mysql_type, cs->number, sorted, true_len, 
+				REC_VERSION_56_MAX_INDEX_COL_LEN);
 
 			if (wsrep_protocol_version > 1) {
 				memcpy(buff, sorted, true_len);
@@ -5697,8 +5721,9 @@ wsrep_store_key_val_for_row(
 			}
 
 			memcpy(sorted, blob_data, true_len);
-			wsrep_innobase_mysql_sort(
-			       mysql_type, cs->number, sorted, true_len);
+			true_len = wsrep_innobase_mysql_sort(
+				mysql_type, cs->number, sorted, true_len,
+				REC_VERSION_56_MAX_INDEX_COL_LEN);
 
 			memcpy(buff, sorted, true_len);
 
@@ -5761,8 +5786,10 @@ wsrep_store_key_val_for_row(
 							&error);
 				}
 				memcpy(sorted, src_start, true_len);
-				wsrep_innobase_mysql_sort(
-					mysql_type, cs->number, sorted, true_len);
+				true_len = wsrep_innobase_mysql_sort(
+					mysql_type, cs->number, sorted, true_len,
+					REC_VERSION_56_MAX_INDEX_COL_LEN);
+
 				memcpy(buff, sorted, true_len);
 			} else {
 				memcpy(buff, src_start, true_len);
@@ -8312,7 +8339,7 @@ wsrep_append_foreign_key(
 			   wsrep_thd_query(thd) : "void");
 		return DB_ERROR;
 	}
-	byte  key[WSREP_MAX_SUPPORTED_KEY_LENGTH+1];
+	byte  key[WSREP_MAX_SUPPORTED_KEY_LENGTH+1] = {'\0'};
 	ulint len = WSREP_MAX_SUPPORTED_KEY_LENGTH;
 
 	dict_index_t *idx_target = (referenced) ? 
@@ -8482,11 +8509,11 @@ ha_innobase::wsrep_append_keys(
 		uint	len;
 		char 	keyval[WSREP_MAX_SUPPORTED_KEY_LENGTH+1] = {'\0'};
 		char 	*key 		= &keyval[0];
-		KEY	*key_info	= table->key_info;
 		ibool    is_null;
 
 		len = wsrep_store_key_val_for_row(
-			table, 0, key, key_info->key_length, record0, &is_null);
+			table, 0, key, WSREP_MAX_SUPPORTED_KEY_LENGTH, 
+			record0, &is_null);
 
 		if (!is_null) {
 			int rcode = wsrep_append_key(
@@ -8504,9 +8531,14 @@ ha_innobase::wsrep_append_keys(
 		uint i;
 		bool hasPK= false;
 
-		for (i=0; i<table->s->keys && !hasPK; ++i) {
+		for (i=0; i<table->s->keys; ++i) {
 			KEY*  key_info	= table->key_info + i;
-			if (key_info->flags & HA_NOSAME) hasPK = true;
+			if (key_info->flags & HA_NOSAME) {
+				hasPK = true;
+				if (i != table->s->primary_key) {
+					wsrep_thd_set_PA_safe(thd, FALSE);
+				}
+			}
 		}
 
 		for (i=0; i<table->s->keys; ++i) {
@@ -8529,13 +8561,15 @@ ha_innobase::wsrep_append_keys(
 					   table->s->table_name.str, 
 					   key_info->name);
 			}
+			/* !hasPK == table with no PK, must append all non-unique keys */
 			if (!hasPK || key_info->flags & HA_NOSAME ||
 			    ((tab &&
 			      dict_table_get_referenced_constraint(tab, idx)) ||
 			     (!tab && referenced_by_foreign_key()))) {
 
 				len = wsrep_store_key_val_for_row(
-					table, i, key0, key_info->key_length, 
+					table, i, key0, 
+					WSREP_MAX_SUPPORTED_KEY_LENGTH,
 					record0, &is_null);
 				if (!is_null) {
 					int rcode = wsrep_append_key(
@@ -8553,7 +8587,8 @@ ha_innobase::wsrep_append_keys(
 				}
 				if (record1) {
 					len = wsrep_store_key_val_for_row(
-						table, i, key1, key_info->key_length, 
+						table, i, key1, 
+						WSREP_MAX_SUPPORTED_KEY_LENGTH, 
 						record1, &is_null);
 					if (!is_null && memcmp(key0, key1, len)) {
 						int rcode = wsrep_append_key(
@@ -14175,6 +14210,7 @@ static int innobase_wsrep_set_checkpoint(handlerton* hton, const XID* xid)
                 trx_sysf_t* sys_header = trx_sysf_get(&mtr);
                 trx_sys_update_wsrep_checkpoint(xid, sys_header, &mtr);
                 mtr_commit(&mtr);
+                innobase_flush_logs(hton);
                 return 0;
         } else {
                 return 1;

storage/xtradb/include/ha_prototypes.h

@@ -306,12 +306,12 @@ UNIV_INTERN int wsrep_innobase_kill_one_trx(void *thd, trx_t *bf_trx, trx_t *vic
 my_bool wsrep_thd_is_BF(void *thd_ptr, my_bool sync);
 //int64_t wsrep_thd_trx_seqno(THD *thd);
 int wsrep_trx_order_before(void *thd1, void *thd2);
-void wsrep_innobase_mysql_sort(int mysql_type, uint charset_number,
-			       unsigned char* str, unsigned int str_length);
-//UNIV_INTERN
-int 
-wsrep_on(void *thd_ptr);
+int wsrep_innobase_mysql_sort(int mysql_type, uint charset_number,
+			      unsigned char* str, unsigned int str_length,
+			      unsigned int buf_length);
+int wsrep_on(void *thd_ptr);
 int wsrep_is_wsrep_xid(const void*);
+my_bool wsrep_thd_set_PA_safe(void *thd_ptr, my_bool safe);
 #endif /* WITH_WSREP */
 /**********************************************************************//**
 Get the current setting of the lower_case_table_names global parameter from

storage/xtradb/lock/lock0lock.c

@@ -1821,12 +1821,7 @@ lock_rec_create(
 	automatically of the gap type */
 
 	if (UNIV_UNLIKELY(heap_no == PAGE_HEAP_NO_SUPREMUM)) {
-#ifdef WITH_WSREP
-		ut_ad(!(type_mode & LOCK_REC_NOT_GAP) || 
-		      wsrep_thd_is_BF(trx->mysql_thd, FALSE));
-#else
 		ut_ad(!(type_mode & LOCK_REC_NOT_GAP));
-#endif /* WITH_WSREP */
 
 		type_mode = type_mode & ~(LOCK_GAP | LOCK_REC_NOT_GAP);
 	}
@@ -2116,12 +2111,7 @@ lock_rec_add_to_queue(
 	struct for a gap type lock */
 
 	if (UNIV_UNLIKELY(heap_no == PAGE_HEAP_NO_SUPREMUM)) {
-#ifdef WITH_WSREP
-		ut_ad(!(type_mode & LOCK_REC_NOT_GAP) || 
-		      wsrep_thd_is_BF(trx->mysql_thd, FALSE));
-#else
 		ut_ad(!(type_mode & LOCK_REC_NOT_GAP));
-#endif /* WITH_WSREP */
 
 		/* There should never be LOCK_REC_NOT_GAP on a supremum
 		record, but let us play safe */

storage/xtradb/rem/rem0rec.c

@@ -1961,10 +1961,10 @@ wsrep_rec_get_foreign_key(
 				key_len++;
 			}
 			memcpy(buf, data, len);
-			wsrep_innobase_mysql_sort(
+			*buf_len = wsrep_innobase_mysql_sort(
 				(int)(col_f->prtype & DATA_MYSQL_TYPE_MASK),
 				(uint)dtype_get_charset_coll(col_f->prtype),
-				buf, len);
+				buf, len, *buf_len);
 		} else { /* new protocol */
 			if (!(col_r->prtype & DATA_NOT_NULL)) {
 				*buf++ = 0;
@@ -1994,12 +1994,12 @@ wsrep_rec_get_foreign_key(
 			case DATA_MYSQL:
 				/* Copy the actual data */
 				ut_memcpy(buf, data, len);
-				wsrep_innobase_mysql_sort(
+				len = wsrep_innobase_mysql_sort(
 					(int)
 					(col_f->prtype & DATA_MYSQL_TYPE_MASK),
 					(uint)
 					dtype_get_charset_coll(col_f->prtype),
-					buf, len);
+					buf, len, *buf_len);
 				break;
 			case DATA_BLOB:
 			case DATA_BINARY:

storage/xtradb/row/row0ins.c

@@ -1772,13 +1772,6 @@ row_ins_scan_sec_index_for_duplicate(
 				lock_type, block, rec, index, offsets, thr);
 		} else {
 
-#ifdef WITH_WSREP
-			if (wsrep_thd_is_BF(thr_get_trx(thr)->mysql_thd, 0)) {
-				if (!(lock_type & LOCK_REC_NOT_GAP)) {
-					lock_type |= LOCK_REC_NOT_GAP;
-				}
-			}
-#endif /* WITH_WSREP */
 			err = row_ins_set_shared_rec_lock(
 				lock_type, block, rec, index, offsets, thr);
 		}