handler0alter.cc 31.6 KB
Newer Older
vasil's avatar
vasil committed
1 2
/*****************************************************************************

3
Copyright (c) 2005, 2010, Innobase Oy. All Rights Reserved.
vasil's avatar
vasil committed
4 5 6 7 8 9 10 11 12 13 14 15 16 17 18

This program is free software; you can redistribute it and/or modify it under
the terms of the GNU General Public License as published by the Free Software
Foundation; version 2 of the License.

This program is distributed in the hope that it will be useful, but WITHOUT
ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS
FOR A PARTICULAR PURPOSE. See the GNU General Public License for more details.

You should have received a copy of the GNU General Public License along with
this program; if not, write to the Free Software Foundation, Inc., 59 Temple
Place, Suite 330, Boston, MA 02111-1307 USA

*****************************************************************************/

19 20
/**************************************************//**
@file handler/handler0alter.cc
21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36
Smart ALTER TABLE
*******************************************************/

#include <mysql_priv.h>
#include <mysqld_error.h>

extern "C" {
#include "log0log.h"
#include "row0merge.h"
#include "srv0srv.h"
#include "trx0trx.h"
#include "trx0roll.h"
#include "ha_prototypes.h"
#include "handler0alter.h"
}

37 38
#include "ha_innodb.h"

39
/*************************************************************//**
40 41 42 43 44 45
Copies an InnoDB column to a MySQL field.  This function is
adapted from row_sel_field_store_in_mysql_format(). */
static
void
innobase_col_to_mysql(
/*==================*/
46 47 48 49
	const dict_col_t*	col,	/*!< in: InnoDB column */
	const uchar*		data,	/*!< in: InnoDB column data */
	ulint			len,	/*!< in: length of data, in bytes */
	Field*			field)	/*!< in/out: MySQL field */
50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99
{
	uchar*	ptr;
	uchar*	dest	= field->ptr;
	ulint	flen	= field->pack_length();

	switch (col->mtype) {
	case DATA_INT:
		ut_ad(len == flen);

		/* Convert integer data from Innobase to little-endian
		format, sign bit restored to normal */

		for (ptr = dest + len; ptr != dest; ) {
			*--ptr = *data++;
		}

		if (!(field->flags & UNSIGNED_FLAG)) {
			((byte*) dest)[len - 1] ^= 0x80;
		}

		break;

	case DATA_VARCHAR:
	case DATA_VARMYSQL:
	case DATA_BINARY:
		field->reset();

		if (field->type() == MYSQL_TYPE_VARCHAR) {
			/* This is a >= 5.0.3 type true VARCHAR. Store the
			length of the data to the first byte or the first
			two bytes of dest. */

			dest = row_mysql_store_true_var_len(
				dest, len, flen - field->key_length());
		}

		/* Copy the actual data */
		memcpy(dest, data, len);
		break;

	case 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, flen, data, len);
		break;

#ifdef UNIV_DEBUG
	case DATA_MYSQL:
		ut_ad(flen >= len);
100 101
		ut_ad(col->mbmaxlen >= col->mbminlen);
		ut_ad(col->mbmaxlen > col->mbminlen || flen == len);
102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124
		memcpy(dest, data, len);
		break;

	default:
	case DATA_SYS_CHILD:
	case DATA_SYS:
		/* These column types should never be shipped to MySQL. */
		ut_ad(0);

	case DATA_CHAR:
	case DATA_FIXBINARY:
	case DATA_FLOAT:
	case DATA_DOUBLE:
	case DATA_DECIMAL:
		/* Above are the valid column types for MySQL data. */
		ut_ad(flen == len);
#else /* UNIV_DEBUG */
	default:
#endif /* UNIV_DEBUG */
		memcpy(dest, data, len);
	}
}

125
/*************************************************************//**
126
Copies an InnoDB record to table->record[0]. */
127
extern "C" UNIV_INTERN
128 129 130
void
innobase_rec_to_mysql(
/*==================*/
131 132 133 134
	TABLE*			table,		/*!< in/out: MySQL table */
	const rec_t*		rec,		/*!< in: record */
	const dict_index_t*	index,		/*!< in: index */
	const ulint*		offsets)	/*!< in: rec_get_offsets(
135 136 137 138 139 140 141 142 143 144 145
						rec, index, ...) */
{
	uint	n_fields	= table->s->fields;
	uint	i;

	ut_ad(n_fields == dict_table_get_n_user_cols(index->table));

	for (i = 0; i < n_fields; i++) {
		Field*		field	= table->field[i];
		ulint		ipos;
		ulint		ilen;
146 147 148
		const uchar*	ifield;

		field->reset();
149 150 151 152

		ipos = dict_index_get_nth_col_pos(index, i);

		if (UNIV_UNLIKELY(ipos == ULINT_UNDEFINED)) {
153
null_field:
154 155 156 157 158 159 160 161 162
			field->set_null();
			continue;
		}

		ifield = rec_get_nth_field(rec, offsets, ipos, &ilen);

		/* Assign the NULL flag */
		if (ilen == UNIV_SQL_NULL) {
			ut_ad(field->real_maybe_null());
163
			goto null_field;
164
		}
165 166 167 168 169 170 171

		field->set_notnull();

		innobase_col_to_mysql(
			dict_field_get_col(
				dict_index_get_nth_field(index, ipos)),
			ifield, ilen, field);
172 173 174
	}
}

175
/*************************************************************//**
176
Resets table->record[0]. */
177
extern "C" UNIV_INTERN
178 179 180
void
innobase_rec_reset(
/*===============*/
181
	TABLE*			table)		/*!< in/out: MySQL table */
182 183 184 185 186 187 188 189 190
{
	uint	n_fields	= table->s->fields;
	uint	i;

	for (i = 0; i < n_fields; i++) {
		table->field[i]->set_default();
	}
}

191
/******************************************************************//**
192 193 194 195 196
Removes the filename encoding of a database and table name. */
static
void
innobase_convert_tablename(
/*=======================*/
197
	char*	s)	/*!< in: identifier; out: decoded identifier */
198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224
{
	uint	errors;

	char*	slash = strchr(s, '/');

	if (slash) {
		char*	t;
		/* Temporarily replace the '/' with NUL. */
		*slash = 0;
		/* Convert the database name. */
		strconvert(&my_charset_filename, s, system_charset_info,
			   s, slash - s + 1, &errors);

		t = s + strlen(s);
		ut_ad(slash >= t);
		/* Append a  '.' after the database name. */
		*t++ = '.';
		slash++;
		/* Convert the table name. */
		strconvert(&my_charset_filename, slash, system_charset_info,
			   t, slash - t + strlen(slash), &errors);
	} else {
		strconvert(&my_charset_filename, s,
			   system_charset_info, s, strlen(s), &errors);
	}
}

225
/*******************************************************************//**
226 227
This function checks that index keys are sensible.
@return	0 or error number */
228 229 230 231
static
int
innobase_check_index_keys(
/*======================*/
232 233 234 235 236
	const KEY*		key_info,	/*!< in: Indexes to be
						created */
	ulint			num_of_keys,	/*!< in: Number of
						indexes to be created */
	const dict_table_t*	table)		/*!< in: Existing indexes */
237 238 239
{
	ulint		key_num;

240 241
	ut_ad(key_info);
	ut_ad(num_of_keys);
242 243

	for (key_num = 0; key_num < num_of_keys; key_num++) {
244
		const KEY&	key = key_info[key_num];
245 246 247 248 249

		/* Check that the same index name does not appear
		twice in indexes to be created. */

		for (ulint i = 0; i < key_num; i++) {
250
			const KEY&	key2 = key_info[i];
251

252
			if (0 == strcmp(key.name, key2.name)) {
253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268
				my_error(ER_WRONG_NAME_FOR_INDEX, MYF(0),
					 key.name);

				return(ER_WRONG_NAME_FOR_INDEX);
			}
		}

		/* Check that the same index name does not already exist. */

		for (const dict_index_t* index
			     = dict_table_get_first_index(table);
		     index; index = dict_table_get_next_index(index)) {

			if (0 == strcmp(key.name, index->name)) {
				my_error(ER_WRONG_NAME_FOR_INDEX, MYF(0),
					 key.name);
269

270
				return(ER_WRONG_NAME_FOR_INDEX);
271 272 273 274 275
			}
		}

		/* Check that MySQL does not try to create a column
		prefix index field on an inappropriate data type and
276
		that the same column does not appear twice in the index. */
277

278 279 280 281 282 283
		for (ulint i = 0; i < key.key_parts; i++) {
			const KEY_PART_INFO&	key_part1
				= key.key_part[i];
			const Field*		field
				= key_part1.field;
			ibool			is_unsigned;
284

285 286 287 288 289 290 291 292 293 294 295 296 297 298 299 300 301 302 303 304
			switch (get_innobase_type_from_mysql_type(
					&is_unsigned, field)) {
			default:
				break;
			case DATA_INT:
			case DATA_FLOAT:
			case DATA_DOUBLE:
			case DATA_DECIMAL:
				if (field->type() == MYSQL_TYPE_VARCHAR) {
					if (key_part1.length
					    >= field->pack_length()
					    - ((Field_varstring*) field)
					    ->length_bytes) {
						break;
					}
				} else {
					if (key_part1.length
					    >= field->pack_length()) {
						break;
					}
305
				}
306

307 308
				my_error(ER_WRONG_KEY_COLUMN, MYF(0),
					 field->field_name);
309
				return(ER_WRONG_KEY_COLUMN);
310 311 312
			}

			for (ulint j = 0; j < i; j++) {
313 314
				const KEY_PART_INFO&	key_part2
					= key.key_part[j];
315

316 317 318
				if (strcmp(key_part1.field->field_name,
					   key_part2.field->field_name)) {
					continue;
319
				}
320

321 322
				my_error(ER_WRONG_KEY_COLUMN, MYF(0),
					 key_part1.field->field_name);
323
				return(ER_WRONG_KEY_COLUMN);
324 325 326 327
			}
		}
	}

328
	return(0);
329 330
}

331
/*******************************************************************//**
332 333 334 335 336
Create index field definition for key part */
static
void
innobase_create_index_field_def(
/*============================*/
337 338 339
	KEY_PART_INFO*		key_part,	/*!< in: MySQL key definition */
	mem_heap_t*		heap,		/*!< in: memory heap */
	merge_index_field_t*	index_field)	/*!< out: index field
340 341 342 343 344 345 346 347 348 349 350 351 352 353 354 355 356 357 358 359 360 361 362 363 364 365 366 367 368 369 370 371 372
						definition for key_part */
{
	Field*		field;
	ibool		is_unsigned;
	ulint		col_type;

	DBUG_ENTER("innobase_create_index_field_def");

	ut_ad(key_part);
	ut_ad(index_field);

	field = key_part->field;
	ut_a(field);

	col_type = get_innobase_type_from_mysql_type(&is_unsigned, field);

	if (DATA_BLOB == col_type
	    || (key_part->length < field->pack_length()
		&& field->type() != MYSQL_TYPE_VARCHAR)
	    || (field->type() == MYSQL_TYPE_VARCHAR
		&& key_part->length < field->pack_length()
			- ((Field_varstring*)field)->length_bytes)) {

		index_field->prefix_len = key_part->length;
	} else {
		index_field->prefix_len = 0;
	}

	index_field->field_name = mem_heap_strdup(heap, field->field_name);

	DBUG_VOID_RETURN;
}

373
/*******************************************************************//**
374 375 376 377 378
Create index definition for key */
static
void
innobase_create_index_def(
/*======================*/
379 380
	KEY*			key,		/*!< in: key definition */
	bool			new_primary,	/*!< in: TRUE=generating
381 382
						a new primary key
						on the table */
383
	bool			key_primary,	/*!< in: TRUE if this key
384
						is a primary key */
385 386
	merge_index_def_t*	index,		/*!< out: index definition */
	mem_heap_t*		heap)		/*!< in: heap where memory
387 388 389 390 391 392 393 394 395 396 397 398 399 400 401 402 403 404 405 406 407 408 409 410 411 412 413 414 415 416 417 418 419 420 421 422 423 424 425 426
						is allocated */
{
	ulint	i;
	ulint	len;
	ulint	n_fields = key->key_parts;
	char*	index_name;

	DBUG_ENTER("innobase_create_index_def");

	index->fields = (merge_index_field_t*) mem_heap_alloc(
		heap, n_fields * sizeof *index->fields);

	index->ind_type = 0;
	index->n_fields = n_fields;
	len = strlen(key->name) + 1;
	index->name = index_name = (char*) mem_heap_alloc(heap,
							  len + !new_primary);

	if (UNIV_LIKELY(!new_primary)) {
		*index_name++ = TEMP_INDEX_PREFIX;
	}

	memcpy(index_name, key->name, len);

	if (key->flags & HA_NOSAME) {
		index->ind_type |= DICT_UNIQUE;
	}

	if (key_primary) {
		index->ind_type |= DICT_CLUSTERED;
	}

	for (i = 0; i < n_fields; i++) {
		innobase_create_index_field_def(&key->key_part[i], heap,
						&index->fields[i]);
	}

	DBUG_VOID_RETURN;
}

427
/*******************************************************************//**
428 429 430 431 432
Copy index field definition */
static
void
innobase_copy_index_field_def(
/*==========================*/
433 434
	const dict_field_t*	field,		/*!< in: definition to copy */
	merge_index_field_t*	index_field)	/*!< out: copied definition */
435 436 437 438 439 440 441 442 443 444 445
{
	DBUG_ENTER("innobase_copy_index_field_def");
	DBUG_ASSERT(field != NULL);
	DBUG_ASSERT(index_field != NULL);

	index_field->field_name = field->name;
	index_field->prefix_len = field->prefix_len;

	DBUG_VOID_RETURN;
}

446
/*******************************************************************//**
447 448 449 450 451
Copy index definition for the index */
static
void
innobase_copy_index_def(
/*====================*/
452 453 454
	const dict_index_t*	index,	/*!< in: index definition to copy */
	merge_index_def_t*	new_index,/*!< out: Index definition */
	mem_heap_t*		heap)	/*!< in: heap where allocated */
455 456 457 458 459 460 461 462 463 464 465 466 467 468 469 470 471 472 473 474 475 476 477 478 479 480 481 482 483
{
	ulint	n_fields;
	ulint	i;

	DBUG_ENTER("innobase_copy_index_def");

	/* Note that we take only those fields that user defined to be
	in the index.  In the internal representation more colums were
	added and those colums are not copied .*/

	n_fields = index->n_user_defined_cols;

	new_index->fields = (merge_index_field_t*) mem_heap_alloc(
		heap, n_fields * sizeof *new_index->fields);

	/* When adding a PRIMARY KEY, we may convert a previous
	clustered index to a secondary index (UNIQUE NOT NULL). */
	new_index->ind_type = index->type & ~DICT_CLUSTERED;
	new_index->n_fields = n_fields;
	new_index->name = index->name;

	for (i = 0; i < n_fields; i++) {
		innobase_copy_index_field_def(&index->fields[i],
					      &new_index->fields[i]);
	}

	DBUG_VOID_RETURN;
}

484
/*******************************************************************//**
485 486 487 488 489 490 491 492 493 494 495 496 497 498
Create an index table where indexes are ordered as follows:

IF a new primary key is defined for the table THEN

	1) New primary key
	2) Original secondary indexes
	3) New secondary indexes

ELSE

	1) All new indexes in the order they arrive from MySQL

ENDIF

499 500

@return	key definitions or NULL */
501 502 503 504
static
merge_index_def_t*
innobase_create_key_def(
/*====================*/
505 506 507
	trx_t*		trx,		/*!< in: trx */
	const dict_table_t*table,		/*!< in: table definition */
	mem_heap_t*	heap,		/*!< in: heap where space for key
508
					definitions are allocated */
509 510
	KEY*		key_info,	/*!< in: Indexes to be created */
	ulint&		n_keys)		/*!< in/out: Number of indexes to
511 512 513 514 515 516 517 518 519 520 521 522 523 524 525 526 527 528 529 530
					be created */
{
	ulint			i = 0;
	merge_index_def_t*	indexdef;
	merge_index_def_t*	indexdefs;
	bool			new_primary;

	DBUG_ENTER("innobase_create_key_def");

	indexdef = indexdefs = (merge_index_def_t*)
		mem_heap_alloc(heap, sizeof *indexdef
			       * (n_keys + UT_LIST_GET_LEN(table->indexes)));

	/* If there is a primary key, it is always the first index
	defined for the table. */

	new_primary = !my_strcasecmp(system_charset_info,
				     key_info->name, "PRIMARY");

	/* If there is a UNIQUE INDEX consisting entirely of NOT NULL
531 532 533
	columns and if the index does not contain column prefix(es)
	(only prefix/part of the column is indexed), MySQL will treat the
	index as a PRIMARY KEY unless the table already has one. */
534 535

	if (!new_primary && (key_info->flags & HA_NOSAME)
536
	    && (!(key_info->flags & HA_KEY_HAS_PART_KEY_SEG))
537
	    && row_table_got_default_clust_index(table)) {
538
		uint    key_part = key_info->key_parts;
539 540 541 542 543 544 545 546 547 548 549 550 551 552 553 554 555 556 557 558 559 560 561 562 563 564 565 566 567 568 569 570 571 572 573 574 575 576 577 578 579 580 581 582 583 584 585 586 587 588 589 590 591 592

		new_primary = TRUE;

		while (key_part--) {
			if (key_info->key_part[key_part].key_type
			    & FIELDFLAG_MAYBE_NULL) {
				new_primary = FALSE;
				break;
			}
		}
	}

	if (new_primary) {
		const dict_index_t*	index;

		/* Create the PRIMARY key index definition */
		innobase_create_index_def(&key_info[i++], TRUE, TRUE,
					  indexdef++, heap);

		row_mysql_lock_data_dictionary(trx);

		index = dict_table_get_first_index(table);

		/* Copy the index definitions of the old table.  Skip
		the old clustered index if it is a generated clustered
		index or a PRIMARY KEY.  If the clustered index is a
		UNIQUE INDEX, it must be converted to a secondary index. */

		if (dict_index_get_nth_col(index, 0)->mtype == DATA_SYS
		    || !my_strcasecmp(system_charset_info,
				      index->name, "PRIMARY")) {
			index = dict_table_get_next_index(index);
		}

		while (index) {
			innobase_copy_index_def(index, indexdef++, heap);
			index = dict_table_get_next_index(index);
		}

		row_mysql_unlock_data_dictionary(trx);
	}

	/* Create definitions for added secondary indexes. */

	while (i < n_keys) {
		innobase_create_index_def(&key_info[i++], new_primary, FALSE,
					  indexdef++, heap);
	}

	n_keys = indexdef - indexdefs;

	DBUG_RETURN(indexdefs);
}

593
/*******************************************************************//**
594 595
Create a temporary tablename using query id, thread id, and id
@return	temporary tablename */
596 597 598 599
static
char*
innobase_create_temporary_tablename(
/*================================*/
600 601 602
	mem_heap_t*	heap,		/*!< in: memory heap */
	char		id,		/*!< in: identifier [0-9a-zA-Z] */
	const char*     table_name)	/*!< in: table name */
603 604 605 606 607 608 609 610 611 612 613 614 615 616 617
{
	char*			name;
	ulint			len;
	static const char	suffix[] = "@0023 "; /* "# " */

	len = strlen(table_name);

	name = (char*) mem_heap_alloc(heap, len + sizeof suffix);
	memcpy(name, table_name, len);
	memcpy(name + len, suffix, sizeof suffix);
	name[len + (sizeof suffix - 2)] = id;

	return(name);
}

618
/*******************************************************************//**
619 620
Create indexes.
@return	0 or error number */
621
UNIV_INTERN
622 623 624
int
ha_innobase::add_index(
/*===================*/
625 626 627
	TABLE*	table,		/*!< in: Table where indexes are created */
	KEY*	key_info,	/*!< in: Indexes to be created */
	uint	num_of_keys)	/*!< in: Number of indexes to be created */
628
{
629 630 631 632 633 634
	dict_index_t**	index;		/*!< Index to be created */
	dict_table_t*	innodb_table;	/*!< InnoDB table in dictionary */
	dict_table_t*	indexed_table;	/*!< Table where indexes are created */
	merge_index_def_t* index_defs;	/*!< Index definitions */
	mem_heap_t*     heap;		/*!< Heap for index definitions */
	trx_t*		trx;		/*!< Transaction */
635
	ulint		num_of_idx;
636 637
	ulint		num_created	= 0;
	ibool		dict_locked	= FALSE;
638
	ulint		new_primary;
639
	int		error;
640 641 642 643 644 645 646 647 648 649

	DBUG_ENTER("ha_innobase::add_index");
	ut_a(table);
	ut_a(key_info);
	ut_a(num_of_keys);

	if (srv_created_new_raw || srv_force_recovery) {
		DBUG_RETURN(HA_ERR_WRONG_COMMAND);
	}

marko's avatar
marko committed
650
	update_thd();
651 652 653 654 655

	heap = mem_heap_create(1024);

	/* In case MySQL calls this in the middle of a SELECT query, release
	possible adaptive hash latch to avoid deadlocks of threads. */
marko's avatar
marko committed
656
	trx_search_latch_release_if_reserved(prebuilt->trx);
657
	trx_start_if_not_started(prebuilt->trx);
658

659 660
	/* Create a background transaction for the operations on
	the data dictionary tables. */
661
	trx = innobase_trx_allocate(user_thd);
662 663 664 665 666
	trx_start_if_not_started(trx);

	innodb_table = indexed_table
		= dict_table_get(prebuilt->table->name, FALSE);

667 668 669 670 671
	if (UNIV_UNLIKELY(!innodb_table)) {
		error = HA_ERR_NO_SUCH_TABLE;
		goto err_exit;
	}

672 673 674 675 676
	/* Check if the index name is reserved. */
	if (innobase_index_name_is_reserved(trx, key_info, num_of_keys)) {
		error = -1;
	} else {
		/* Check that index keys are sensible */
677 678
		error = innobase_check_index_keys(key_info, num_of_keys,
						  innodb_table);
679
	}
680 681 682 683

	if (UNIV_UNLIKELY(error)) {
err_exit:
		mem_heap_free(heap);
684
		trx_general_rollback_for_mysql(trx, NULL);
685
		trx_free_for_mysql(trx);
686
		trx_commit_for_mysql(prebuilt->trx);
687 688 689 690 691 692 693 694 695 696 697 698
		DBUG_RETURN(error);
	}

	/* Create table containing all indexes to be built in this
	alter table add index so that they are in the correct order
	in the table. */

	num_of_idx = num_of_keys;

	index_defs = innobase_create_key_def(
		trx, innodb_table, heap, key_info, num_of_idx);

699 700
	new_primary = DICT_CLUSTERED & index_defs[0].ind_type;

701 702 703 704 705
	/* Allocate memory for dictionary index definitions */

	index = (dict_index_t**) mem_heap_alloc(
		heap, num_of_idx * sizeof *index);

706
	/* Flag this transaction as a dictionary operation, so that
707 708
	the data dictionary will be locked in crash recovery. */
	trx_set_dict_operation(trx, TRX_DICT_OP_INDEX);
709

710
	/* Acquire a lock on the table before creating any indexes. */
711
	error = row_merge_lock_table(prebuilt->trx, innodb_table,
712
				     new_primary ? LOCK_X : LOCK_S);
713 714 715 716 717 718

	if (UNIV_UNLIKELY(error != DB_SUCCESS)) {

		goto error_handling;
	}

719 720 721 722 723 724
	/* Latch the InnoDB data dictionary exclusively so that no deadlocks
	or lock waits can happen in it during an index create operation. */

	row_mysql_lock_data_dictionary(trx);
	dict_locked = TRUE;

725 726
	ut_d(dict_table_check_for_dup_indexes(innodb_table, FALSE));

727 728 729 730
	/* If a new primary key is defined for the table we need
	to drop the original table and rebuild all indexes. */

	if (UNIV_UNLIKELY(new_primary)) {
731 732 733 734
		/* This transaction should be the only one
		operating on the table. */
		ut_a(innodb_table->n_mysql_handles_opened == 1);

735 736 737 738
		char*	new_table_name = innobase_create_temporary_tablename(
			heap, '1', innodb_table->name);

		/* Clone the table. */
739
		trx_set_dict_operation(trx, TRX_DICT_OP_TABLE);
740 741 742 743 744 745 746 747 748 749 750 751 752 753 754
		indexed_table = row_merge_create_temporary_table(
			new_table_name, index_defs, innodb_table, trx);

		if (!indexed_table) {

			switch (trx->error_state) {
			case DB_TABLESPACE_ALREADY_EXISTS:
			case DB_DUPLICATE_KEY:
				innobase_convert_tablename(new_table_name);
				my_error(HA_ERR_TABLE_EXIST, MYF(0),
					 new_table_name);
				error = HA_ERR_TABLE_EXIST;
				break;
			default:
				error = convert_error_code_to_mysql(
755 756
					trx->error_state, innodb_table->flags,
					user_thd);
757 758
			}

759 760
			ut_d(dict_table_check_for_dup_indexes(innodb_table,
							      FALSE));
761 762 763 764 765 766 767 768 769 770 771 772 773 774 775 776 777 778 779 780 781 782
			row_mysql_unlock_data_dictionary(trx);
			goto err_exit;
		}

		trx->table_id = indexed_table->id;
	}

	/* Create the indexes in SYS_INDEXES and load into dictionary. */

	for (ulint i = 0; i < num_of_idx; i++) {

		index[i] = row_merge_create_index(trx, indexed_table,
						  &index_defs[i]);

		if (!index[i]) {
			error = trx->error_state;
			goto error_handling;
		}

		num_created++;
	}

783
	ut_ad(error == DB_SUCCESS);
784

785 786 787 788
	/* We will need to rebuild index translation table. Set
	valid index entry count in the translation table to zero */
	share->idx_trans_tbl.index_count = 0;

789
	/* Commit the data dictionary transaction in order to release
790 791 792 793 794
	the table locks on the system tables.  This means that if
	MySQL crashes while creating a new primary key inside
	row_merge_build_indexes(), indexed_table will not be dropped
	by trx_rollback_active().  It will have to be recovered or
	dropped by the database administrator. */
795 796
	trx_commit_for_mysql(trx);

797 798 799 800 801 802 803 804 805 806 807
	row_mysql_unlock_data_dictionary(trx);
	dict_locked = FALSE;

	ut_a(trx->n_active_thrs == 0);
	ut_a(UT_LIST_GET_LEN(trx->signals) == 0);

	if (UNIV_UNLIKELY(new_primary)) {
		/* A primary key is to be built.  Acquire an exclusive
		table lock also on the table that is being created. */
		ut_ad(indexed_table != innodb_table);

808 809
		error = row_merge_lock_table(prebuilt->trx, indexed_table,
					     LOCK_X);
810 811 812 813 814 815 816 817 818

		if (UNIV_UNLIKELY(error != DB_SUCCESS)) {

			goto error_handling;
		}
	}

	/* Read the clustered index of the table and build indexes
	based on this information using temporary files and merge sort. */
819 820
	error = row_merge_build_indexes(prebuilt->trx,
					innodb_table, indexed_table,
821
					index, num_of_idx, table);
822 823 824 825 826 827 828 829 830

error_handling:
	/* After an error, remove all those index definitions from the
	dictionary which were defined. */

	switch (error) {
		const char*	old_name;
		char*		tmp_name;
	case DB_SUCCESS:
831 832 833
		ut_a(!dict_locked);
		row_mysql_lock_data_dictionary(trx);
		dict_locked = TRUE;
834

835
		ut_d(dict_table_check_for_dup_indexes(prebuilt->table, TRUE));
836

837 838 839 840 841 842 843 844 845 846 847 848 849 850 851 852 853 854 855 856 857 858 859 860 861 862 863 864 865 866 867 868 869 870
		if (!new_primary) {
			error = row_merge_rename_indexes(trx, indexed_table);

			if (error != DB_SUCCESS) {
				row_merge_drop_indexes(trx, indexed_table,
						       index, num_created);
			}

			goto convert_error;
		}

		/* If a new primary key was defined for the table and
		there was no error at this point, we can now rename
		the old table as a temporary table, rename the new
		temporary table as the old table and drop the old table. */
		old_name = innodb_table->name;
		tmp_name = innobase_create_temporary_tablename(heap, '2',
							       old_name);

		error = row_merge_rename_tables(innodb_table, indexed_table,
						tmp_name, trx);

		if (error != DB_SUCCESS) {

			row_merge_drop_table(trx, indexed_table);

			switch (error) {
			case DB_TABLESPACE_ALREADY_EXISTS:
			case DB_DUPLICATE_KEY:
				innobase_convert_tablename(tmp_name);
				my_error(HA_ERR_TABLE_EXIST, MYF(0), tmp_name);
				error = HA_ERR_TABLE_EXIST;
				break;
			default:
871
				goto convert_error;
872 873 874 875
			}
			break;
		}

876
		trx_commit_for_mysql(prebuilt->trx);
877 878 879
		row_prebuilt_free(prebuilt, TRUE);
		prebuilt = row_create_prebuilt(indexed_table);

880
		indexed_table->n_mysql_handles_opened++;
881 882

		error = row_merge_drop_table(trx, innodb_table);
883
		innodb_table = indexed_table;
884 885
		goto convert_error;

886 887 888
	case DB_TOO_BIG_RECORD:
		my_error(HA_ERR_TO_BIG_ROW, MYF(0));
		goto error;
889 890 891 892
	case DB_PRIMARY_KEY_IS_NULL:
		my_error(ER_PRIMARY_CANT_HAVE_NULL, MYF(0));
		/* fall through */
	case DB_DUPLICATE_KEY:
893
error:
894 895 896 897
		prebuilt->trx->error_info = NULL;
		/* fall through */
	default:
		if (new_primary) {
898 899 900
			if (indexed_table != innodb_table) {
				row_merge_drop_table(trx, indexed_table);
			}
901
		} else {
902 903 904 905 906
			if (!dict_locked) {
				row_mysql_lock_data_dictionary(trx);
				dict_locked = TRUE;
			}

907 908 909 910 911
			row_merge_drop_indexes(trx, indexed_table,
					       index, num_created);
		}

convert_error:
912 913 914
		error = convert_error_code_to_mysql(error,
						    innodb_table->flags,
						    user_thd);
915 916 917 918
	}

	mem_heap_free(heap);
	trx_commit_for_mysql(trx);
919 920 921
	if (prebuilt->trx) {
		trx_commit_for_mysql(prebuilt->trx);
	}
922 923

	if (dict_locked) {
924
		ut_d(dict_table_check_for_dup_indexes(innodb_table, FALSE));
925 926 927 928 929 930 931 932 933 934 935
		row_mysql_unlock_data_dictionary(trx);
	}

	trx_free_for_mysql(trx);

	/* There might be work for utility threads.*/
	srv_active_wake_master_thread();

	DBUG_RETURN(error);
}

936
/*******************************************************************//**
937 938
Prepare to drop some indexes of a table.
@return	0 or error number */
939
UNIV_INTERN
940 941 942
int
ha_innobase::prepare_drop_index(
/*============================*/
943 944 945
	TABLE*	table,		/*!< in: Table where indexes are dropped */
	uint*	key_num,	/*!< in: Key nums to be dropped */
	uint	num_of_keys)	/*!< in: Number of keys to be dropped */
946 947 948 949 950 951 952 953 954 955 956 957 958
{
	trx_t*		trx;
	int		err = 0;
	uint 		n_key;

	DBUG_ENTER("ha_innobase::prepare_drop_index");
	ut_ad(table);
	ut_ad(key_num);
	ut_ad(num_of_keys);
	if (srv_created_new_raw || srv_force_recovery) {
		DBUG_RETURN(HA_ERR_WRONG_COMMAND);
	}

marko's avatar
marko committed
959
	update_thd();
960

marko's avatar
marko committed
961 962
	trx_search_latch_release_if_reserved(prebuilt->trx);
	trx = prebuilt->trx;
963 964 965 966

	/* Test and mark all the indexes to be dropped */

	row_mysql_lock_data_dictionary(trx);
967
	ut_d(dict_table_check_for_dup_indexes(prebuilt->table, FALSE));
968

969 970 971 972 973 974 975 976 977 978 979
	/* Check that none of the indexes have previously been flagged
	for deletion. */
	{
		const dict_index_t*	index
			= dict_table_get_first_index(prebuilt->table);
		do {
			ut_a(!index->to_be_dropped);
			index = dict_table_get_next_index(index);
		} while (index);
	}

980 981 982 983 984 985 986 987 988 989 990 991 992 993 994 995 996 997 998 999 1000 1001 1002 1003 1004 1005 1006 1007 1008 1009 1010 1011 1012 1013 1014 1015 1016 1017 1018 1019 1020 1021 1022 1023 1024 1025
	for (n_key = 0; n_key < num_of_keys; n_key++) {
		const KEY*	key;
		dict_index_t*	index;

		key = table->key_info + key_num[n_key];
		index = dict_table_get_index_on_name_and_min_id(
			prebuilt->table, key->name);

		if (!index) {
			sql_print_error("InnoDB could not find key n:o %u "
					"with name %s for table %s",
					key_num[n_key],
					key ? key->name : "NULL",
					prebuilt->table->name);

			err = HA_ERR_KEY_NOT_FOUND;
			goto func_exit;
		}

		/* Refuse to drop the clustered index.  It would be
		better to automatically generate a clustered index,
		but mysql_alter_table() will call this method only
		after ha_innobase::add_index(). */

		if (dict_index_is_clust(index)) {
			my_error(ER_REQUIRES_PRIMARY_KEY, MYF(0));
			err = -1;
			goto func_exit;
		}

		index->to_be_dropped = TRUE;
	}

	/* If FOREIGN_KEY_CHECK = 1 you may not drop an index defined
	for a foreign key constraint because InnoDB requires that both
	tables contain indexes for the constraint.  Note that CREATE
	INDEX id ON table does a CREATE INDEX and DROP INDEX, and we
	can ignore here foreign keys because a new index for the
	foreign key has already been created.

	We check for the foreign key constraints after marking the
	candidate indexes for deletion, because when we check for an
	equivalent foreign index we don't want to select an index that
	is later deleted. */

	if (trx->check_foreigns
marko's avatar
marko committed
1026
	    && thd_sql_command(user_thd) != SQLCOM_CREATE_INDEX) {
1027
		dict_index_t*	index;
1028

1029 1030 1031
		for (index = dict_table_get_first_index(prebuilt->table);
		     index;
		     index = dict_table_get_next_index(index)) {
1032
			dict_foreign_t*	foreign;
1033

1034 1035
			if (!index->to_be_dropped) {

1036
				continue;
1037
			}
1038 1039 1040 1041 1042 1043 1044 1045 1046 1047 1048 1049 1050 1051 1052 1053 1054 1055 1056 1057 1058 1059 1060 1061 1062 1063

			/* Check if the index is referenced. */
			foreign = dict_table_get_referenced_constraint(
				prebuilt->table, index);

			if (foreign) {
index_needed:
				trx_set_detailed_error(
					trx,
					"Index needed in foreign key "
					"constraint");

				trx->error_info = index;

				err = HA_ERR_DROP_INDEX_FK;
				break;
			} else {
				/* Check if this index references some
				other table */
				foreign = dict_table_get_foreign_constraint(
					prebuilt->table, index);

				if (foreign) {
					ut_a(foreign->foreign_index == index);

					/* Search for an equivalent index that
1064
					the foreign key constraint could use
1065
					if this index were to be deleted. */
1066 1067
					if (!dict_foreign_find_equiv_index(
						foreign)) {
1068 1069 1070 1071 1072

						goto index_needed;
					}
				}
			}
1073 1074 1075 1076 1077 1078
		}
	} else if (thd_sql_command(user_thd) == SQLCOM_CREATE_INDEX) {
		/* This is a drop of a foreign key constraint index that
		was created by MySQL when the constraint was added.  MySQL
		does this when the user creates an index explicitly which
		can be used in place of the automatically generated index. */
1079

1080 1081 1082 1083 1084 1085 1086 1087 1088 1089 1090 1091 1092 1093 1094 1095 1096 1097 1098 1099 1100 1101 1102 1103 1104 1105 1106 1107 1108 1109 1110 1111 1112 1113 1114 1115 1116 1117 1118
		dict_index_t*	index;

		for (index = dict_table_get_first_index(prebuilt->table);
		     index;
		     index = dict_table_get_next_index(index)) {
			dict_foreign_t*	foreign;

			if (!index->to_be_dropped) {

				continue;
			}

			/* Check if this index references some other table */
			foreign = dict_table_get_foreign_constraint(
				prebuilt->table, index);

			if (foreign == NULL) {

				continue;
			}

			ut_a(foreign->foreign_index == index);

			/* Search for an equivalent index that the
			foreign key constraint could use if this index
			were to be deleted. */

			if (!dict_foreign_find_equiv_index(foreign)) {
				trx_set_detailed_error(
					trx,
					"Index needed in foreign key "
					"constraint");

				trx->error_info = foreign->foreign_index;

				err = HA_ERR_DROP_INDEX_FK;
				break;
			}
		}
1119 1120 1121 1122
	}

func_exit:
	if (err) {
1123 1124 1125
		/* Undo our changes since there was some sort of error. */
		dict_index_t*	index
			= dict_table_get_first_index(prebuilt->table);
1126

1127 1128 1129 1130
		do {
			index->to_be_dropped = FALSE;
			index = dict_table_get_next_index(index);
		} while (index);
1131 1132
	}

1133
	ut_d(dict_table_check_for_dup_indexes(prebuilt->table, FALSE));
1134 1135 1136 1137 1138
	row_mysql_unlock_data_dictionary(trx);

	DBUG_RETURN(err);
}

1139
/*******************************************************************//**
1140 1141
Drop the indexes that were passed to a successful prepare_drop_index().
@return	0 or error number */
1142
UNIV_INTERN
1143 1144 1145
int
ha_innobase::final_drop_index(
/*==========================*/
1146
	TABLE*	table)		/*!< in: Table where indexes are dropped */
1147
{
1148 1149
	dict_index_t*	index;		/*!< Index to be dropped */
	trx_t*		trx;		/*!< Transaction */
1150
	int		err;
1151 1152 1153 1154 1155 1156 1157 1158

	DBUG_ENTER("ha_innobase::final_drop_index");
	ut_ad(table);

	if (srv_created_new_raw || srv_force_recovery) {
		DBUG_RETURN(HA_ERR_WRONG_COMMAND);
	}

marko's avatar
marko committed
1159
	update_thd();
1160

marko's avatar
marko committed
1161
	trx_search_latch_release_if_reserved(prebuilt->trx);
1162
	trx_start_if_not_started(prebuilt->trx);
1163 1164 1165

	/* Create a background transaction for the operations on
	the data dictionary tables. */
1166
	trx = innobase_trx_allocate(user_thd);
1167 1168 1169
	trx_start_if_not_started(trx);

	/* Flag this transaction as a dictionary operation, so that
1170 1171
	the data dictionary will be locked in crash recovery. */
	trx_set_dict_operation(trx, TRX_DICT_OP_INDEX);
1172

1173 1174 1175
	/* Lock the table exclusively, to ensure that no active
	transaction depends on an index that is being dropped. */
	err = convert_error_code_to_mysql(
1176
		row_merge_lock_table(prebuilt->trx, prebuilt->table, LOCK_X),
1177
		prebuilt->table->flags, user_thd);
1178

1179
	row_mysql_lock_data_dictionary(trx);
1180
	ut_d(dict_table_check_for_dup_indexes(prebuilt->table, FALSE));
1181

1182 1183
	if (UNIV_UNLIKELY(err)) {

1184 1185 1186 1187 1188 1189 1190
		/* Unmark the indexes to be dropped. */
		for (index = dict_table_get_first_index(prebuilt->table);
		     index; index = dict_table_get_next_index(index)) {

			index->to_be_dropped = FALSE;
		}

1191 1192 1193
		goto func_exit;
	}

1194 1195
	/* Drop indexes marked to be dropped */

1196 1197 1198 1199 1200 1201 1202 1203 1204 1205 1206 1207 1208 1209 1210
	index = dict_table_get_first_index(prebuilt->table);

	while (index) {
		dict_index_t*	next_index;

		next_index = dict_table_get_next_index(index);

		if (index->to_be_dropped) {

			row_merge_drop_index(index, prebuilt->table, trx);
		}

		index = next_index;
	}

1211 1212 1213 1214 1215 1216
	/* Check that all flagged indexes were dropped. */
	for (index = dict_table_get_first_index(prebuilt->table);
	     index; index = dict_table_get_next_index(index)) {
		ut_a(!index->to_be_dropped);
	}

1217 1218 1219 1220
	/* We will need to rebuild index translation table. Set
	valid index entry count in the translation table to zero */
	share->idx_trans_tbl.index_count = 0;

1221
func_exit:
1222
	ut_d(dict_table_check_for_dup_indexes(prebuilt->table, FALSE));
1223
	trx_commit_for_mysql(trx);
1224
	trx_commit_for_mysql(prebuilt->trx);
1225
	row_mysql_unlock_data_dictionary(trx);
1226 1227 1228 1229 1230 1231 1232

	/* Flush the log to reduce probability that the .frm files and
	the InnoDB data dictionary get out-of-sync if the user runs
	with innodb_flush_log_at_trx_commit = 0 */

	log_buffer_flush_to_disk();

1233 1234
	trx_free_for_mysql(trx);

1235 1236 1237 1238 1239
	/* Tell the InnoDB server that there might be work for
	utility threads: */

	srv_active_wake_master_thread();

1240
	DBUG_RETURN(err);
1241
}