InnoDB: Make CREATE TABLE return error when the minimum row length

exceeds the maximum record size.  (Bug #5682)


innobase/data/data0type.c:
  Remove function dtype_str_needs_mysql_cmp().
  Document dtype_get_at_most_n_mbchars().
  dtype_get_at_most_n_mbchars(): Use mbminlen and mbmaxlen.
innobase/dict/dict0crea.c:
  dict_build_table_def_step(): Reject if minimum row size is too big.
innobase/include/data0type.h:
  Remove dtype_str_needs_mysql_cmp().
  Document dtype_get_at_most_n_mbchars().
  Add dtype_get_min_size().
innobase/include/data0type.ic:
  Add dtype_get_min_size().
innobase/include/row0mysql.h:
  row_mysql_store_col_in_innobase_format(): Add parameter comp,
  as we will only truncate UTF-8 CHAR(n) columns in row_format=compact.
innobase/include/row0mysql.ic:
  row_mysql_store_col_in_innobase_format(): Add parameter comp,
  as we will only truncate UTF-8 CHAR(n) columns in row_format=compact.
innobase/row/row0mysql.c:
  Pass parameter comp to row_mysql_store_col_in_innobase_format().
innobase/row/row0sel.c:
  Pass parameter comp to row_mysql_store_col_in_innobase_format().
  row_sel_field_store_in_mysql_format(): Undo the stripping of
  UTF-8 CHAR(n) columns by padding with spaces.

8 files changed, 175 insertions, 82 deletions

diff --git a/innobase/data/data0type.c b/innobase/data/data0type.c
index b8e4dd9c7f2..3fcd666b5a5 100644
--- a/innobase/data/data0type.c
+++ b/innobase/data/data0type.c
@@ -45,54 +45,33 @@ dtype_t		dtype_binary_val = {DATA_BINARY, 0, 0, 0, 0, 0};
 dtype_t* 	dtype_binary 	= &dtype_binary_val;
 
 /*************************************************************************
-Checks if a string type has to be compared by the MySQL comparison functions.
-InnoDB internally only handles binary byte string comparisons, as well as
-latin1_swedish_ci strings. For example, UTF-8 strings have to be compared
-by MySQL. */
-
-ibool
-dtype_str_needs_mysql_cmp(
-/*======================*/
-				/* out: TRUE if a string type that requires
-				comparison with MySQL functions */
-	dtype_t*	dtype)	/* in: type struct */
-{
-	if (dtype->mtype == DATA_MYSQL
-	    || dtype->mtype == DATA_VARMYSQL
-	    || (dtype->mtype == DATA_BLOB
-	        && 0 == (dtype->prtype & DATA_BINARY_TYPE)
-		&& dtype_get_charset_coll(dtype->prtype) !=
-				data_mysql_latin1_swedish_charset_coll)) {
-		return(TRUE);
-	}
-
-	return(FALSE);
-}
-
-/*************************************************************************
-For the documentation of this function, see innobase_get_at_most_n_mbchars()
-in ha_innodb.cc. */
+Determine how many bytes the first n characters of the given string occupy.
+If the string is shorter than n characters, returns the number of bytes
+the characters in the string occupy. */
 
 ulint
 dtype_get_at_most_n_mbchars(
 /*========================*/
-	dtype_t*	dtype,
-	ulint		prefix_len,
-	ulint		data_len,
-	const char*	str)
+					/* out: length of the prefix,
+					in bytes */
+	const dtype_t*	dtype,		/* in: data type */
+	ulint		prefix_len,	/* in: length of the requested
+					prefix, in characters, multiplied by
+					dtype_get_mbmaxlen(dtype) */
+	ulint		data_len,	/* in: length of str (in bytes) */
+	const char*	str)		/* in: the string whose prefix
+					length is being determined */
 {
 #ifndef UNIV_HOTBACKUP
 	ut_a(data_len != UNIV_SQL_NULL);
+	ut_a(!(prefix_len % dtype->mbmaxlen));
 
-	if (dtype_str_needs_mysql_cmp(dtype)) {
+	if (dtype->mbminlen != dtype->mbmaxlen) {
 		return(innobase_get_at_most_n_mbchars(
 				dtype_get_charset_coll(dtype->prtype),
 				prefix_len, data_len, str));
 	}
 
-	/* We assume here that the string types that InnoDB itself can compare
-	are single-byte charsets! */
-
 	if (prefix_len < data_len) {
 
 		return(prefix_len);
diff --git a/innobase/dict/dict0crea.c b/innobase/dict/dict0crea.c
index b6f79ad10b4..4926797721c 100644
--- a/innobase/dict/dict0crea.c
+++ b/innobase/dict/dict0crea.c
@@ -220,6 +220,8 @@ dict_build_table_def_step(
 	const char*	path_or_name;
 	ibool		is_path;
 	mtr_t		mtr;
+	ulint		i;
+	ulint		row_len;
 
 #ifdef UNIV_SYNC_DEBUG
 	ut_ad(mutex_own(&(dict_sys->mutex)));
@@ -231,6 +233,15 @@ dict_build_table_def_step(
 
 	thr_get_trx(thr)->table_id = table->id;
 
+	row_len = 0;
+	for (i = 0; i < table->n_def; i++) {
+		row_len += dtype_get_min_size(dict_col_get_type(
+						&table->cols[i]));
+	}
+	if (row_len > BTR_PAGE_MAX_REC_SIZE) {
+		return(DB_TOO_BIG_RECORD);
+	}
+
 	if (table->type == DICT_TABLE_CLUSTER_MEMBER) {
 
 		cluster_table = dict_table_get_low(table->cluster_name);
diff --git a/innobase/include/data0type.h b/innobase/include/data0type.h
index 4c327d0b7ed..174665ca1fa 100644
--- a/innobase/include/data0type.h
+++ b/innobase/include/data0type.h
@@ -145,28 +145,22 @@ store the charset-collation number; one byte is left unused, though */
 #define DATA_NEW_ORDER_NULL_TYPE_BUF_SIZE	6
 
 /*************************************************************************
-Checks if a string type has to be compared by the MySQL comparison functions.
-InnoDB internally only handles binary byte string comparisons, as well as
-latin1_swedish_ci strings. For example, UTF-8 strings have to be compared
-by MySQL. */
-
-ibool
-dtype_str_needs_mysql_cmp(
-/*======================*/
-				/* out: TRUE if a string type that requires
-				comparison with MySQL functions */
-	dtype_t*	dtype);	/* in: type struct */
-/*************************************************************************
-For the documentation of this function, see innobase_get_at_most_n_mbchars()
-in ha_innodb.cc. */
+Determine how many bytes the first n characters of the given string occupy.
+If the string is shorter than n characters, returns the number of bytes
+the characters in the string occupy. */
 
 ulint
 dtype_get_at_most_n_mbchars(
 /*========================*/
-	dtype_t*	dtype,
-	ulint		prefix_len,
-	ulint		data_len,
-	const char*	str);
+					/* out: length of the prefix,
+					in bytes */
+	const dtype_t*	dtype,		/* in: data type */
+	ulint		prefix_len,	/* in: length of the requested
+					prefix, in characters, multiplied by
+					dtype_get_mbmaxlen(dtype) */
+	ulint		data_len,	/* in: length of str (in bytes) */
+	const char*	str);		/* in: the string whose prefix
+					length is being determined */
 /*************************************************************************
 Checks if a data main type is a string type. Also a BLOB is considered a
 string type. */
@@ -306,6 +300,14 @@ dtype_get_fixed_size(
 				/* out: fixed size, or 0 */
 	dtype_t*	type);	/* in: type */
 /***************************************************************************
+Returns the minimum size of a data type. */
+UNIV_INLINE
+ulint
+dtype_get_min_size(
+/*===============*/
+				/* out: minimum size */
+	const dtype_t*	type);	/* in: type */
+/***************************************************************************
 Returns a stored SQL NULL size for a type. For fixed length types it is
 the fixed length of the type, otherwise 0. */
 UNIV_INLINE
diff --git a/innobase/include/data0type.ic b/innobase/include/data0type.ic
index 761f7b3da7f..47f35bdce67 100644
--- a/innobase/include/data0type.ic
+++ b/innobase/include/data0type.ic
@@ -314,6 +314,7 @@ dtype_new_read_for_order_and_null_size(
 	dtype_set_mblen(type);
 }
 
+#ifndef UNIV_HOTBACKUP
 /***************************************************************************
 Returns the size of a fixed size data type, 0 if not a fixed size type. */
 UNIV_INLINE
@@ -323,7 +324,6 @@ dtype_get_fixed_size(
 				/* out: fixed size, or 0 */
 	dtype_t*	type)	/* in: type */
 {
-#ifndef UNIV_HOTBACKUP
 	ulint	mtype;
 
 	mtype = dtype_get_mtype(type);
@@ -401,15 +401,66 @@ dtype_get_fixed_size(
 	}
 
 	return(0);
-#else /* UNIV_HOTBACKUP */
-	/* This function depends on MySQL code that is not included in
-	InnoDB Hot Backup builds.  Besides, this function should never
-	be called in InnoDB Hot Backup. */
-	ut_error;
-#endif /* UNIV_HOTBACKUP */
 }
 
 /***************************************************************************
+Returns the size of a fixed size data type, 0 if not a fixed size type. */
+UNIV_INLINE
+ulint
+dtype_get_min_size(
+/*===============*/
+				/* out: minimum size */
+	const dtype_t*	type)	/* in: type */
+{
+	switch (type->mtype) {
+	case DATA_SYS:
+#ifdef UNIV_DEBUG
+			switch (type->prtype & DATA_MYSQL_TYPE_MASK) {
+			default:
+				ut_ad(0);
+				return(0);
+			case DATA_ROW_ID:
+				ut_ad(type->len == DATA_ROW_ID_LEN);
+				break;
+			case DATA_TRX_ID:
+				ut_ad(type->len == DATA_TRX_ID_LEN);
+				break;
+			case DATA_ROLL_PTR:
+				ut_ad(type->len == DATA_ROLL_PTR_LEN);
+				break;
+			case DATA_MIX_ID:
+				ut_ad(type->len == DATA_MIX_ID_LEN);
+				break;
+			}
+#endif /* UNIV_DEBUG */
+	case DATA_CHAR:
+	case DATA_FIXBINARY:
+	case DATA_INT:
+	case DATA_FLOAT:
+	case DATA_DOUBLE:
+	case DATA_MYSQL:
+			if ((type->prtype & DATA_BINARY_TYPE)
+					|| type->mbminlen == type->mbmaxlen) {
+				return(type->len);
+			}
+			/* this is a variable-length character set */
+			ut_a(type->mbminlen > 0);
+			ut_a(type->mbmaxlen > type->mbminlen);
+			return(type->len * type->mbminlen / type->mbmaxlen);
+	case DATA_VARCHAR:
+	case DATA_BINARY:
+	case DATA_DECIMAL:
+	case DATA_VARMYSQL:
+	case DATA_BLOB:
+			return(0); 
+	default:	ut_error;
+	}
+
+	return(0);
+}
+#endif /* !UNIV_HOTBACKUP */
+
+/***************************************************************************
 Returns a stored SQL NULL size for a type. For fixed length types it is
 the fixed length of the type, otherwise 0. */
 UNIV_INLINE
diff --git a/innobase/include/row0mysql.h b/innobase/include/row0mysql.h
index 7ffa6ebf87d..48a9d9bc941 100644
--- a/innobase/include/row0mysql.h
+++ b/innobase/include/row0mysql.h
@@ -99,6 +99,7 @@ row_mysql_store_col_in_innobase_format(
 					as dfield is used! */
 	ulint		col_len,	/* in: MySQL column length */
 	ulint		type,		/* in: data type */
+	bool		comp,		/* in: TRUE=compact format */
 	ulint		is_unsigned);	/* in: != 0 if unsigned integer type */
 /********************************************************************
 Handles user errors and lock waits detected by the database engine. */
diff --git a/innobase/include/row0mysql.ic b/innobase/include/row0mysql.ic
index 1f5d0b0c990..910546e298c 100644
--- a/innobase/include/row0mysql.ic
+++ b/innobase/include/row0mysql.ic
@@ -67,6 +67,7 @@ row_mysql_store_col_in_innobase_format(
 					as dfield is used! */
 	ulint		col_len,	/* in: MySQL column length */
 	ulint		type,		/* in: data type */
+	bool		comp,		/* in: TRUE=compact format */
 	ulint		is_unsigned)	/* in: != 0 if unsigned integer type */
 {
 	byte*	ptr 	= mysql_data;
@@ -113,6 +114,37 @@ row_mysql_store_col_in_innobase_format(
 				col_len--;
 			}
 		}
+	} else if (comp && type == DATA_MYSQL
+			&& dtype_get_mbminlen(dfield_get_type(dfield)) == 1
+			&& dtype_get_mbmaxlen(dfield_get_type(dfield)) > 1) {
+		/* We assume that this CHAR field is encoded in a
+		variable-length character set where spaces have
+		1:1 correspondence to 0x20 bytes, such as UTF-8.
+
+		Consider a CHAR(n) field, a field of n characters.
+		It will contain between n*mbminlen and n*mbmaxlen bytes.
+		We will try to truncate it to n bytes by stripping
+		space padding.  If the field contains single-byte
+		characters only, it will be truncated to n characters.
+		Consider a CHAR(5) field containing the string ".a   "
+		where "." denotes a 3-byte character represented by
+		the bytes "$%&".  After our stripping, the string will
+		be stored as "$%&a " (5 bytes).  The string ".abc "
+		will be stored as "$%&abc" (6 bytes).
+
+		The space padding will be restored in row0sel.c, function
+		row_sel_field_store_in_mysql_format(). */
+
+		ulint		n_chars;
+		dtype_t*	dtype = dfield_get_type(dfield);
+
+		ut_a(!(dtype_get_len(dtype) % dtype_get_mbmaxlen(dtype)));
+		n_chars = dtype_get_len(dtype) / dtype_get_mbmaxlen(dtype);
+
+		/* Strip space padding. */
+		while (col_len > n_chars && ptr[col_len - 1] == 0x20) {
+			col_len--;
+		}
 	} else if (type == DATA_BLOB) {
 		ptr = row_mysql_read_blob_ref(&col_len, mysql_data, col_len);
 	}
diff --git a/innobase/row/row0mysql.c b/innobase/row/row0mysql.c
index 0ec261e798f..39c4b76f814 100644
--- a/innobase/row/row0mysql.c
+++ b/innobase/row/row0mysql.c
@@ -238,7 +238,8 @@ row_mysql_convert_row_to_innobase(
 						+ templ->mysql_col_offset,
 					mysql_rec + templ->mysql_col_offset,
 					templ->mysql_col_len,
-					templ->type, templ->is_unsigned);
+					templ->type, prebuilt->table->comp,
+					templ->is_unsigned);
 next_column:
 		;
 	} 
diff --git a/innobase/row/row0sel.c b/innobase/row/row0sel.c
index c0141f896ce..84a160cfc0d 100644
--- a/innobase/row/row0sel.c
+++ b/innobase/row/row0sel.c
@@ -2137,6 +2137,7 @@ row_sel_convert_mysql_key_to_innobase(
 		        row_mysql_store_col_in_innobase_format(
 					dfield, buf, key_ptr + data_offset,
 					data_len, type,
+					index->table->comp,
 					dfield_get_type(dfield)->prtype
 							& DATA_UNSIGNED);
 			buf += data_len;
@@ -2232,17 +2233,15 @@ row_sel_field_store_in_mysql_format(
 			are not in themselves stored here: the caller must
 			allocate and copy the BLOB into buffer before, and pass
 			the pointer to the BLOB in 'data' */
-	ulint	col_len,/* in: MySQL column length */
+	const mysql_row_templ_t* templ,	/* in: MySQL column template */
 	byte*	data,	/* in: data to store */
-	ulint	len,	/* in: length of the data */
-	ulint	type,	/* in: data type */
-	ulint	is_unsigned)/* in: != 0 if an unsigned integer type */
+	ulint	len)	/* in: length of the data */
 {
 	byte*	ptr;
 
 	ut_ad(len != UNIV_SQL_NULL);
 
-	if (type == DATA_INT) {
+	if (templ->type == DATA_INT) {
 		/* Convert integer data from Innobase to a little-endian
 		format, sign bit restored to normal */
 
@@ -2257,31 +2256,47 @@ row_sel_field_store_in_mysql_format(
 			data++;
 		}
 
-		if (!is_unsigned) {
+		if (!templ->is_unsigned) {
 			dest[len - 1] = (byte) (dest[len - 1] ^ 128);
 		}
 
-		ut_ad(col_len == len);
-	} else if (type == DATA_VARCHAR || type == DATA_VARMYSQL
-						|| type == DATA_BINARY) {
+		ut_ad(templ->mysql_col_len == len);
+	} else if (templ->type == DATA_VARCHAR || templ->type == DATA_VARMYSQL
+					|| templ->type == DATA_BINARY) {
 		/* Store the length of the data to the first two bytes of
 		dest; does not do anything yet because MySQL has
 		no real vars! */
 		
 		dest = row_mysql_store_var_len(dest, len);
 		ut_memcpy(dest, data, len);
-
-		/* ut_ad(col_len >= len + 2); No real var implemented in
-		MySQL yet! */
+#if 0
+		/* No real var implemented in MySQL yet! */
+		ut_ad(templ->mysql_col_len >= len + 2);
+#endif
 		
-	} else if (type == DATA_BLOB) {
+	} else if (templ->type == DATA_BLOB) {
 		/* Store a pointer to the BLOB buffer to dest: the BLOB was
 		already copied to the buffer in row_sel_store_mysql_rec */
 
-		row_mysql_store_blob_ref(dest, col_len, data, len);
+		row_mysql_store_blob_ref(dest, templ->mysql_col_len,
+							data, len);
 	} else {
-		ut_memcpy(dest, data, len);
-		ut_ad(col_len == len);
+		memcpy(dest, data, len);
+
+		ut_ad(templ->mysql_col_len >= len);
+		ut_ad(templ->mbmaxlen >= templ->mbminlen);
+
+		ut_ad(templ->mbmaxlen > templ->mbminlen
+			|| templ->mysql_col_len == len);
+		ut_ad(!templ->mbmaxlen
+			|| !(templ->mysql_col_len % templ->mbmaxlen));
+
+		if (templ->mbminlen != templ->mbmaxlen) {
+			/* Pad with spaces.  This undoes the stripping
+			done in row0mysql.ic, function
+			row_mysql_store_col_in_innobase_format(). */
+			memset(dest + len, 0x20, templ->mysql_col_len - len);
+		}
 	}
 }
 
@@ -2401,8 +2416,7 @@ row_sel_store_mysql_rec(
 		
 			row_sel_field_store_in_mysql_format(
 				mysql_rec + templ->mysql_col_offset,
-				templ->mysql_col_len, data, len,
-				templ->type, templ->is_unsigned);
+				templ, data, len);
 
 			if (templ->type == DATA_VARCHAR
 					|| templ->type == DATA_VARMYSQL
@@ -2487,7 +2501,7 @@ row_sel_store_mysql_rec(
 					len -= 2;
 				}
 			} else {
-				ut_ad(templ->mbminlen == 1);
+				ut_ad(!pad_char || templ->mbminlen == 1);
 				memset(mysql_rec + templ->mysql_col_offset,
 					pad_char, templ->mysql_col_len);
 			}
@@ -2855,9 +2869,11 @@ row_sel_push_cache_row_for_mysql(
 
 	ut_ad(prebuilt->fetch_cache_first == 0);
 
-	ut_a(row_sel_store_mysql_rec(
+	if (!row_sel_store_mysql_rec(
 			prebuilt->fetch_cache[prebuilt->n_fetch_cached],
-			prebuilt, rec, offsets));
+			prebuilt, rec, offsets)) {
+		ut_error;
+	}
 
 	prebuilt->n_fetch_cached++;
 }