branches/zip: Implement a limit for the size of undo log records.

innodb-index.test: Add a test with a large number of externally stored
columns.  Check that there may not be prefix indexes on too many columns.

dict_index_too_big_for_undo(): New function: Check if the undo log may
overflow.

dict_index_add_to_cache(): Return DB_SUCCESS or DB_TOO_BIG_RECORD.
Postpone the creation and linking of some data structures, so that
when dict_index_too_big_for_undo() holds, it will be easier to clean up.
Check the return status in all callers.

dict/dict0boot.c

@@ -215,6 +215,7 @@ dict_boot(void)
 	dict_hdr_t*	dict_hdr;
 	mem_heap_t*	heap;
 	mtr_t		mtr;
+	ulint		error;
 
 	mtr_start(&mtr);
 
@@ -272,9 +273,11 @@ dict_boot(void)
 
 	index->id = DICT_TABLES_ID;
 
-	dict_index_add_to_cache(table, index,
-				mtr_read_ulint(dict_hdr + DICT_HDR_TABLES,
+	error = dict_index_add_to_cache(table, index,
+					mtr_read_ulint(dict_hdr
+						       + DICT_HDR_TABLES,
 						       MLOG_4BYTES, &mtr));
+	ut_a(error == DB_SUCCESS);
 
 	/*-------------------------*/
 	index = dict_mem_index_create("SYS_TABLES", "ID_IND",
@@ -282,9 +285,11 @@ dict_boot(void)
 	dict_mem_index_add_field(index, "ID", 0);
 
 	index->id = DICT_TABLE_IDS_ID;
-	dict_index_add_to_cache(table, index,
-				mtr_read_ulint(dict_hdr + DICT_HDR_TABLE_IDS,
+	error = dict_index_add_to_cache(table, index,
+					mtr_read_ulint(dict_hdr
+						       + DICT_HDR_TABLE_IDS,
 						       MLOG_4BYTES, &mtr));
+	ut_a(error == DB_SUCCESS);
 
 	/*-------------------------*/
 	table = dict_mem_table_create("SYS_COLUMNS", DICT_HDR_SPACE, 7, 0);
@@ -311,9 +316,11 @@ dict_boot(void)
 	dict_mem_index_add_field(index, "POS", 0);
 
 	index->id = DICT_COLUMNS_ID;
-	dict_index_add_to_cache(table, index,
-				mtr_read_ulint(dict_hdr + DICT_HDR_COLUMNS,
+	error = dict_index_add_to_cache(table, index,
+					mtr_read_ulint(dict_hdr
+						       + DICT_HDR_COLUMNS,
 						       MLOG_4BYTES, &mtr));
+	ut_a(error == DB_SUCCESS);
 
 	/*-------------------------*/
 	table = dict_mem_table_create("SYS_INDEXES", DICT_HDR_SPACE, 7, 0);
@@ -350,9 +357,11 @@ dict_boot(void)
 	dict_mem_index_add_field(index, "ID", 0);
 
 	index->id = DICT_INDEXES_ID;
-	dict_index_add_to_cache(table, index,
-				mtr_read_ulint(dict_hdr + DICT_HDR_INDEXES,
+	error = dict_index_add_to_cache(table, index,
+					mtr_read_ulint(dict_hdr
+						       + DICT_HDR_INDEXES,
 						       MLOG_4BYTES, &mtr));
+	ut_a(error == DB_SUCCESS);
 
 	/*-------------------------*/
 	table = dict_mem_table_create("SYS_FIELDS", DICT_HDR_SPACE, 3, 0);
@@ -374,9 +383,11 @@ dict_boot(void)
 	dict_mem_index_add_field(index, "POS", 0);
 
 	index->id = DICT_FIELDS_ID;
-	dict_index_add_to_cache(table, index,
-				mtr_read_ulint(dict_hdr + DICT_HDR_FIELDS,
+	error = dict_index_add_to_cache(table, index,
+					mtr_read_ulint(dict_hdr
+						       + DICT_HDR_FIELDS,
 						       MLOG_4BYTES, &mtr));
+	ut_a(error == DB_SUCCESS);
 
 	mtr_commit(&mtr);
 	/*-------------------------*/

dict/dict0crea.c

@@ -1088,12 +1088,16 @@ dict_create_index_step(
 
 		dulint	index_id = node->index->id;
 
-		dict_index_add_to_cache(node->table, node->index, FIL_NULL);
+		err = dict_index_add_to_cache(node->table, node->index,
+					      FIL_NULL);
 
 		node->index = dict_index_get_if_in_cache_low(index_id);
-		ut_a(node->index);
+		ut_a(!node->index == (err != DB_SUCCESS));
 
-		err = DB_SUCCESS;
+		if (err != DB_SUCCESS) {
+
+			goto function_exit;
+		}
 
 		node->state = INDEX_CREATE_INDEX_TREE;
 	}

dict/dict0dict.c

@@ -1223,12 +1223,115 @@ dict_col_name_is_reserved(
 	return(FALSE);
 }
 
+/********************************************************************
+If an undo log record for this table might not fit on a single page,
+return TRUE. */
+static
+ibool
+dict_index_too_big_for_undo(
+/*========================*/
+						/* out: TRUE if the undo log
+						record could become too big */
+	const dict_table_t*	table,		/* in: table */
+	const dict_index_t*	new_index)	/* in: index */
+{
+	/* Make sure that all column prefixes will fit in the undo log record
+	in trx_undo_page_report_modify() right after trx_undo_page_init(). */
+
+	ulint			i;
+	const dict_index_t*	clust_index
+		= dict_table_get_first_index(table);
+	ulint			undo_page_len
+		= TRX_UNDO_PAGE_HDR - TRX_UNDO_PAGE_HDR_SIZE
+		+ 2 /* next record pointer */
+		+ 1 /* type_cmpl */
+		+ 11 /* trx->undo_no */ - 11 /* table->id */
+		+ 1 /* rec_get_info_bits() */
+		+ 11 /* DB_TRX_ID */
+		+ 11 /* DB_ROLL_PTR */
+		+ 10 + FIL_PAGE_DATA_END /* trx_undo_left() */
+		+ 2/* pointer to previous undo log record */;
+
+	if (UNIV_UNLIKELY(!clust_index)) {
+		ut_a(dict_index_is_clust(new_index));
+		clust_index = new_index;
+	}
+
+	/* Add the size of the ordering columns in the
+	clustered index. */
+	for (i = 0; i < clust_index->n_uniq; i++) {
+		const dict_col_t*	col
+			= dict_index_get_nth_col(clust_index, i);
+
+		/* Use the maximum output size of
+		mach_write_compressed(), although the encoded
+		length should always fit in 2 bytes. */
+		undo_page_len += 5 + dict_col_get_max_size(col);
+	}
+
+	/* Add the old values of the columns to be updated.
+	First, the amount and the numbers of the columns.
+	These are written by mach_write_compressed() whose
+	maximum output length is 5 bytes.  However, given that
+	the quantities are below REC_MAX_N_FIELDS (10 bits),
+	the maximum length is 2 bytes per item. */
+	undo_page_len += 2 * (dict_table_get_n_cols(table) + 1);
+
+	for (i = 0; i < clust_index->n_def; i++) {
+		const dict_col_t*	col
+			= dict_index_get_nth_col(clust_index, i);
+		ulint			max_size
+			= dict_col_get_max_size(col);
+		ulint			fixed_size
+			= dict_col_get_fixed_size(col);
+
+		if (fixed_size) {
+			/* Fixed-size columns are stored locally. */
+			max_size = fixed_size;
+		} else if (max_size <= BTR_EXTERN_FIELD_REF_SIZE * 2) {
+			/* Short columns are stored locally. */
+		} else if (!col->ord_part) {
+			/* See if col->ord_part would be set
+			because of new_index. */
+			ulint	j;
+
+			for (j = 0; j < new_index->n_uniq; j++) {
+				if (dict_index_get_nth_col(
+					    new_index, j) == col) {
+
+					goto is_ord_part;
+				}
+			}
+
+			/* This is not an ordering column in any index.
+			Thus, it can be stored completely externally. */
+			max_size = BTR_EXTERN_FIELD_REF_SIZE;
+		} else {
+is_ord_part:
+			/* This is an ordering column in some index.
+			A long enough prefix must be written to the
+			undo log.  See trx_undo_page_fetch_ext(). */
+
+			if (max_size > REC_MAX_INDEX_COL_LEN) {
+				max_size = REC_MAX_INDEX_COL_LEN;
+			}
+
+			max_size += BTR_EXTERN_FIELD_REF_SIZE;
+		}
+
+		undo_page_len += 5 + max_size;
+	}
+
+	return(undo_page_len >= UNIV_PAGE_SIZE);
+}
+
 /**************************************************************************
 Adds an index to the dictionary cache. */
 
-void
+ulint
 dict_index_add_to_cache(
 /*====================*/
+				/* out: DB_SUCCESS or DB_TOO_BIG_RECORD */
 	dict_table_t*	table,	/* in: table on which the index is */
 	dict_index_t*	index,	/* in, own: index; NOTE! The index memory
 				object is freed in this function! */
@@ -1258,32 +1361,66 @@ dict_index_add_to_cache(
 		new_index = dict_index_build_internal_non_clust(table, index);
 	}
 
-	new_index->search_info = btr_search_info_create(new_index->heap);
-
 	/* Set the n_fields value in new_index to the actual defined
 	number of fields in the cache internal representation */
 
 	new_index->n_fields = new_index->n_def;
 
-	/* Add the new index as the last index for the table */
-
-	UT_LIST_ADD_LAST(indexes, table->indexes, new_index);
-	new_index->table = table;
-	new_index->table_name = table->name;
-
-	/* Increment the ord_part counts in columns which are ordering */
-
 	if (UNIV_UNLIKELY(index->type & DICT_UNIVERSAL)) {
 		n_ord = new_index->n_fields;
 	} else {
-		n_ord = dict_index_get_n_unique(new_index);
+		n_ord = new_index->n_uniq;
 	}
 
+	for (i = 0; i < n_ord; i++) {
+		const dict_field_t*	field
+			= dict_index_get_nth_field(new_index, i);
+		const dict_col_t*	col
+			= dict_field_get_col(field);
+
+		/* In dtuple_convert_big_rec(), variable-length columns
+		that are longer than BTR_EXTERN_FIELD_REF_SIZE * 2
+		may be chosen for external storage.  If the column appears
+		in an ordering column of an index, a longer prefix of
+		REC_MAX_INDEX_COL_LEN will be copied to the undo log
+		by trx_undo_page_report_modify() and
+		trx_undo_page_fetch_ext().  It suffices to check the
+		capacity of the undo log whenever new_index includes
+		a column prefix on a column that may be stored externally. */
+
+		if (field->prefix_len /* prefix index */
+		    && !col->ord_part /* not yet ordering column */
+		    && !dict_col_get_fixed_size(col) /* variable-length */
+		    && dict_col_get_max_size(col)
+		    > BTR_EXTERN_FIELD_REF_SIZE * 2 /* long enough */) {
+
+			if (dict_index_too_big_for_undo(table, new_index)) {
+				/* An undo log record might not fit in
+				a single page.  Refuse to create this index. */
+				dict_mem_index_free(new_index);
+				dict_mem_index_free(index);
+				return(DB_TOO_BIG_RECORD);
+			}
+
+			break;
+		}
+	}
+
+	/* Flag the ordering columns */
+
 	for (i = 0; i < n_ord; i++) {
 
 		dict_index_get_nth_field(new_index, i)->col->ord_part = 1;
 	}
 
+	/* Add the new index as the last index for the table */
+
+	UT_LIST_ADD_LAST(indexes, table->indexes, new_index);
+	new_index->table = table;
+	new_index->table_name = table->name;
+
+	new_index->search_info = btr_search_info_create(new_index->heap);
+
 	new_index->stat_index_size = 1;
 	new_index->stat_n_leaf_pages = 1;
 
@@ -1308,6 +1445,8 @@ dict_index_add_to_cache(
 	dict_sys->size += mem_heap_get_size(new_index->heap);
 
 	dict_mem_index_free(index);
+
+	return(DB_SUCCESS);
 }
 
 /**************************************************************************

dict/dict0load.c

@@ -728,7 +728,16 @@ dict_load_indexes(
 			index->id = id;
 
 			dict_load_fields(index, heap);
-			dict_index_add_to_cache(table, index, page_no);
+			error = dict_index_add_to_cache(table, index, page_no);
+			/* The data dictionary tables should never contain
+			invalid index definitions.  If we ignored this error
+			and simply did not load this index definition, the
+			.frm file would disagree with the index definitions
+			inside InnoDB. */
+			if (UNIV_UNLIKELY(error != DB_SUCCESS)) {
+
+				goto func_exit;
+			}
 		}
 
 next_rec:

ibuf/ibuf0ibuf.c

@@ -469,6 +469,7 @@ ibuf_data_init_for_space(
 	dict_table_t*	table;
 	dict_index_t*	index;
 	ulint		n_used;
+	ulint		error;
 
 	ut_a(space == 0);
 
@@ -547,7 +548,9 @@ ibuf_data_init_for_space(
 
 	index->id = ut_dulint_add(DICT_IBUF_ID_MIN, space);
 
-	dict_index_add_to_cache(table, index, FSP_IBUF_TREE_ROOT_PAGE_NO);
+	error = dict_index_add_to_cache(table, index,
+					FSP_IBUF_TREE_ROOT_PAGE_NO);
+	ut_a(error == DB_SUCCESS);
 
 	data->index = dict_table_get_first_index(table);
 

include/dict0dict.h

@@ -661,9 +661,10 @@ dict_index_find_on_id_low(
 /**************************************************************************
 Adds an index to the dictionary cache. */
 
-void
+ulint
 dict_index_add_to_cache(
 /*====================*/
+				/* out: DB_SUCCESS or error code */
 	dict_table_t*	table,	/* in: table on which the index is */
 	dict_index_t*	index,	/* in, own: index; NOTE! The index memory
 				object is freed in this function! */

mysql-test/innodb-index.result

@@ -885,3 +885,78 @@ a
 66
 commit;
 drop table t1;
+create table t1(a blob,b blob,c blob,d blob,e blob,f blob,g blob,h blob,
+i blob,j blob,k blob,l blob,m blob,n blob,o blob,p blob,
+q blob,r blob,s blob,t blob,u blob)
+engine=innodb;
+create index t1a on t1 (a(1));
+create index t1b on t1 (b(1));
+create index t1c on t1 (c(1));
+create index t1d on t1 (d(1));
+create index t1e on t1 (e(1));
+create index t1f on t1 (f(1));
+create index t1g on t1 (g(1));
+create index t1h on t1 (h(1));
+create index t1i on t1 (i(1));
+create index t1j on t1 (j(1));
+create index t1k on t1 (k(1));
+create index t1l on t1 (l(1));
+create index t1m on t1 (m(1));
+create index t1n on t1 (n(1));
+create index t1o on t1 (o(1));
+create index t1p on t1 (p(1));
+create index t1q on t1 (q(1));
+create index t1r on t1 (r(1));
+create index t1s on t1 (s(1));
+create index t1t on t1 (t(1));
+create index t1u on t1 (u(1));
+ERROR HY000: Too big row
+create index t1ut on t1 (u(1), t(1));
+ERROR HY000: Too big row
+create index t1st on t1 (s(1), t(1));
+show create table t1;
+Table	Create Table
+t1	CREATE TABLE `t1` (
+  `a` blob,
+  `b` blob,
+  `c` blob,
+  `d` blob,
+  `e` blob,
+  `f` blob,
+  `g` blob,
+  `h` blob,
+  `i` blob,
+  `j` blob,
+  `k` blob,
+  `l` blob,
+  `m` blob,
+  `n` blob,
+  `o` blob,
+  `p` blob,
+  `q` blob,
+  `r` blob,
+  `s` blob,
+  `t` blob,
+  `u` blob,
+  KEY `t1a` (`a`(1)),
+  KEY `t1b` (`b`(1)),
+  KEY `t1c` (`c`(1)),
+  KEY `t1d` (`d`(1)),
+  KEY `t1e` (`e`(1)),
+  KEY `t1f` (`f`(1)),
+  KEY `t1g` (`g`(1)),
+  KEY `t1h` (`h`(1)),
+  KEY `t1i` (`i`(1)),
+  KEY `t1j` (`j`(1)),
+  KEY `t1k` (`k`(1)),
+  KEY `t1l` (`l`(1)),
+  KEY `t1m` (`m`(1)),
+  KEY `t1n` (`n`(1)),
+  KEY `t1o` (`o`(1)),
+  KEY `t1p` (`p`(1)),
+  KEY `t1q` (`q`(1)),
+  KEY `t1r` (`r`(1)),
+  KEY `t1s` (`s`(1)),
+  KEY `t1t` (`t`(1)),
+  KEY `t1st` (`s`(1),`t`(1))
+) ENGINE=InnoDB DEFAULT CHARSET=latin1

mysql-test/innodb-index.test

@@ -314,3 +314,38 @@ disconnect a;
 disconnect b;
 
 drop table t1;
+
+# Test creating a table that could lead to undo log overflow.
+# In the undo log, we write a 768-byte prefix (REC_MAX_INDEX_COL_LEN)
+# of each externally stored column that appears as a column prefix in an index.
+# For this test case, it would suffice to write 1 byte, though.
+create table t1(a blob,b blob,c blob,d blob,e blob,f blob,g blob,h blob,
+		i blob,j blob,k blob,l blob,m blob,n blob,o blob,p blob,
+		q blob,r blob,s blob,t blob,u blob)
+	engine=innodb;
+create index t1a on t1 (a(1));
+create index t1b on t1 (b(1));
+create index t1c on t1 (c(1));
+create index t1d on t1 (d(1));
+create index t1e on t1 (e(1));
+create index t1f on t1 (f(1));
+create index t1g on t1 (g(1));
+create index t1h on t1 (h(1));
+create index t1i on t1 (i(1));
+create index t1j on t1 (j(1));
+create index t1k on t1 (k(1));
+create index t1l on t1 (l(1));
+create index t1m on t1 (m(1));
+create index t1n on t1 (n(1));
+create index t1o on t1 (o(1));
+create index t1p on t1 (p(1));
+create index t1q on t1 (q(1));
+create index t1r on t1 (r(1));
+create index t1s on t1 (s(1));
+create index t1t on t1 (t(1));
+--error 139
+create index t1u on t1 (u(1));
+--error 139
+create index t1ut on t1 (u(1), t(1));
+create index t1st on t1 (s(1), t(1));
+show create table t1;