MDEV-24978 : SIGABRT in __libc_message

Keyvalue can be longer than REC_VERSION_56_MAX_INDEX_COL_LEN
and this leads out-of-array reference. Use dynamic memory
allocation using actual max length of key value.

mysql-test/suite/galera/r/galera_fulltext.result

@@ -34,3 +34,31 @@ COUNT(f1) = 1000
 1
 DROP TABLE t1;
 DROP TABLE ten;
+connection node_1;
+SET @value=REPEAT (1,5001);
+CREATE TABLE t (a VARCHAR(5000),FULLTEXT (a)) engine=innodb;
+INSERT IGNORE INTO t VALUES(@value);
+Warnings:
+Warning	1265	Data truncated for column 'a' at row 1
+SELECT COUNT(*) FROM t;
+COUNT(*)
+1
+connection node_2;
+SELECT COUNT(*) FROM t;
+COUNT(*)
+1
+connection node_1;
+DROP TABLE t;
+CREATE TABLE t (a VARCHAR(5000)) engine=innodb;
+INSERT IGNORE INTO t VALUES(@value);
+Warnings:
+Warning	1265	Data truncated for column 'a' at row 1
+SELECT COUNT(*) FROM t;
+COUNT(*)
+1
+connection node_2;
+SELECT COUNT(*) FROM t;
+COUNT(*)
+1
+connection node_1;
+DROP TABLE t;

mysql-test/suite/galera/t/galera_fulltext.test

@@ -60,3 +60,26 @@ SELECT COUNT(f1) = 1000 FROM t1 WHERE MATCH(f1) AGAINST ('abcdefjhk');
 
 DROP TABLE t1;
 DROP TABLE ten;
+#
+# MDEV-24978 : SIGABRT in __libc_message
+#
+--connection node_1
+SET @value=REPEAT (1,5001);
+CREATE TABLE t (a VARCHAR(5000),FULLTEXT (a)) engine=innodb;
+INSERT IGNORE INTO t VALUES(@value);
+SELECT COUNT(*) FROM t;
+
+--connection node_2
+SELECT COUNT(*) FROM t;
+
+--connection node_1
+DROP TABLE t;
+CREATE TABLE t (a VARCHAR(5000)) engine=innodb;
+INSERT IGNORE INTO t VALUES(@value);
+SELECT COUNT(*) FROM t;
+
+--connection node_2
+SELECT COUNT(*) FROM t;
+
+--connection node_1
+DROP TABLE t;

storage/innobase/handler/ha_innodb.cc

@@ -6656,8 +6656,10 @@ wsrep_innobase_mysql_sort(
 	case MYSQL_TYPE_LONG_BLOB:
 	case MYSQL_TYPE_VARCHAR:
 	{
-		uchar tmp_str[REC_VERSION_56_MAX_INDEX_COL_LEN] = {'\0'};
-		uint tmp_length = REC_VERSION_56_MAX_INDEX_COL_LEN;
+		uchar *tmp_str;
+		uint tmp_length= ut_max(str_length,(uint)REC_VERSION_56_MAX_INDEX_COL_LEN);
+
+		tmp_str= static_cast<uchar *>(ut_zalloc_nokey(tmp_length));
 
 		/* Use the charset number to pick the right charset struct for
 		the comparison. Since the MySQL function get_charset may be
@@ -6680,7 +6682,6 @@ 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,
@@ -6704,6 +6705,7 @@ wsrep_innobase_mysql_sort(
 			ret_length = tmp_length;
 		}
 
+		ut_free(tmp_str);
 		break;
 	}
 	case MYSQL_TYPE_DECIMAL :
@@ -7055,7 +7057,7 @@ wsrep_store_key_val_for_row(
 	THD* 		thd,
 	TABLE*		table,
 	uint		keynr,	/*!< in: key number */
-	char*		buff,	/*!< in/out: buffer for the key value (in MySQL
+	uchar*		buff,	/*!< in/out: buffer for the key value (in MySQL
 				format) */
 	uint		buff_len,/*!< in: buffer length */
 	const uchar*	record,
@@ -7064,7 +7066,7 @@ wsrep_store_key_val_for_row(
 	KEY*		key_info	= table->key_info + keynr;
 	KEY_PART_INFO*	key_part	= key_info->key_part;
 	KEY_PART_INFO*	end		= key_part + key_info->user_defined_key_parts;
-	char*		buff_start	= buff;
+	uchar*		buff_start	= buff;
 	enum_field_types mysql_type;
 	Field*		field;
 	uint buff_space = buff_len;
@@ -7076,7 +7078,8 @@ wsrep_store_key_val_for_row(
 
 	for (; key_part != end; key_part++) {
 
-		uchar sorted[REC_VERSION_56_MAX_INDEX_COL_LEN] = {'\0'};
+		uchar *sorted= NULL;
+		uint max_len= 0;
 		ibool part_is_null = FALSE;
 
 		if (key_part->null_bit) {
@@ -7155,10 +7158,14 @@ wsrep_store_key_val_for_row(
 				true_len = key_len;
 			}
 
+			max_len= ut_max(true_len, (ulint)REC_VERSION_56_MAX_INDEX_COL_LEN);
+			sorted= static_cast<uchar *>(ut_zalloc_nokey(max_len));
 			memcpy(sorted, data, true_len);
 			true_len = wsrep_innobase_mysql_sort(
 				mysql_type, cs->number, sorted, true_len,
-				REC_VERSION_56_MAX_INDEX_COL_LEN);
+				max_len);
+			ut_ad(true_len <= max_len);
+
 			if (wsrep_protocol_version > 1) {
 				/* Note that we always reserve the maximum possible
 				length of the true VARCHAR in the key value, though
@@ -7243,11 +7250,13 @@ wsrep_store_key_val_for_row(
 				true_len = key_len;
 			}
 
+			max_len= ut_max(true_len, (ulint)REC_VERSION_56_MAX_INDEX_COL_LEN);
+			sorted= static_cast<uchar *>(ut_zalloc_nokey(max_len));
 			memcpy(sorted, blob_data, true_len);
 			true_len = wsrep_innobase_mysql_sort(
 				mysql_type, cs->number, sorted, true_len,
-				REC_VERSION_56_MAX_INDEX_COL_LEN);
-
+				max_len);
+			ut_ad(true_len <= max_len);
 
 			/* Note that we always reserve the maximum possible
 			length of the BLOB prefix in the key value. */
@@ -7323,10 +7332,13 @@ wsrep_store_key_val_for_row(
 								cs->mbmaxlen),
 							&error);
 				}
+				max_len= ut_max(true_len, (ulint)REC_VERSION_56_MAX_INDEX_COL_LEN);
+				sorted= static_cast<uchar *>(ut_zalloc_nokey(max_len));
 				memcpy(sorted, src_start, true_len);
 				true_len = wsrep_innobase_mysql_sort(
 					mysql_type, cs->number, sorted, true_len,
-					REC_VERSION_56_MAX_INDEX_COL_LEN);
+					max_len);
+				ut_ad(true_len <= max_len);
 
 				if (true_len > buff_space) {
 					fprintf (stderr,
@@ -7341,6 +7353,11 @@ wsrep_store_key_val_for_row(
 			buff       += true_len;
 			buff_space -= true_len;
 		}
+
+		if (sorted){
+			ut_free(sorted);
+			sorted= NULL;
+		}
 	}
 
 	ut_a(buff <= buff_start + buff_len);
@@ -10446,7 +10463,7 @@ wsrep_append_key(
 	trx_t 		*trx,
 	TABLE_SHARE 	*table_share,
 	TABLE 		*table,
-	const char*	key,
+	const uchar*	key,
 	uint16_t        key_len,
 	wsrep_key_type	key_type	/*!< in: access type of this key
 					(shared, exclusive, semi...) */
@@ -10550,8 +10567,8 @@ ha_innobase::wsrep_append_keys(
 
 	if (wsrep_protocol_version == 0) {
 		uint	len;
-		char 	keyval[WSREP_MAX_SUPPORTED_KEY_LENGTH+1] = {'\0'};
-		char 	*key 		= &keyval[0];
+		uchar 	keyval[WSREP_MAX_SUPPORTED_KEY_LENGTH+1] = {'\0'};
+		uchar 	*key 		= &keyval[0];
 		ibool    is_null;
 
 		len = wsrep_store_key_val_for_row(
@@ -10583,18 +10600,18 @@ ha_innobase::wsrep_append_keys(
 
 		for (i=0; i<table->s->keys; ++i) {
 			uint  len;
-			char  keyval0[WSREP_MAX_SUPPORTED_KEY_LENGTH+1] = {'\0'};
-			char  keyval1[WSREP_MAX_SUPPORTED_KEY_LENGTH+1] = {'\0'};
-			char* key0 		= &keyval0[1];
-			char* key1 		= &keyval1[1];
+			uchar  keyval0[WSREP_MAX_SUPPORTED_KEY_LENGTH+1] = {'\0'};
+			uchar  keyval1[WSREP_MAX_SUPPORTED_KEY_LENGTH+1] = {'\0'};
+			uchar* key0 		= &keyval0[1];
+			uchar* key1 		= &keyval1[1];
 			KEY*  key_info	= table->key_info + i;
 			ibool is_null;
 
 			dict_index_t* idx  = innobase_get_index(i);
 			dict_table_t* tab  = (idx) ? idx->table : NULL;
 
-			keyval0[0] = (char)i;
-			keyval1[0] = (char)i;
+			keyval0[0] = (uchar)i;
+			keyval1[0] = (uchar)i;
 
 			if (!tab) {
 				WSREP_WARN("MariaDB-InnoDB key mismatch %s %s",
@@ -10652,7 +10669,7 @@ ha_innobase::wsrep_append_keys(
 
 		wsrep_calc_row_hash(digest, record0, table, m_prebuilt, thd);
 		if ((rcode = wsrep_append_key(thd, trx, table_share, table,
-					      (const char*) digest, 16,
+					      (const uchar*) digest, 16,
 					      key_type))) {
 			DBUG_RETURN(rcode);
 		}
@@ -10662,7 +10679,7 @@ ha_innobase::wsrep_append_keys(
 				digest, record1, table, m_prebuilt, thd);
 			if ((rcode = wsrep_append_key(thd, trx, table_share,
 						      table,
-						      (const char*) digest,
+						      (const uchar*) digest,
 						      16, key_type))) {
 				DBUG_RETURN(rcode);
 			}