~drizzle-trunk/drizzle/development

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

Loads to the memory cache database object definitions

from dictionary tables

(c) 1996 Innobase Oy

Created 4/24/1996 Heikki Tuuri

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

#include "dict0load.h"

#ifdef UNIV_NONINL

#include "dict0load.ic"

#endif

#include "btr0pcur.h"

#include "btr0btr.h"

#include "page0page.h"

#include "mach0data.h"

#include "dict0dict.h"

#include "dict0boot.h"

#include "rem0cmp.h"

#include "srv0start.h"

#include "srv0srv.h"

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

Returns TRUE if index's i'th column's name is 'name' .*/

static

ibool

name_of_col_is(

/*===========*/

				/* out: */

	dict_table_t*	table,	/* in: table */

	dict_index_t*	index,	/* in: index */

	ulint		i,	/* in:  */

	const char*	name)	/* in: name to compare to */

{

	ulint	tmp = dict_col_get_no(dict_field_get_col(

					      dict_index_get_nth_field(

						      index, i)));

	return(strcmp(name, dict_table_get_col_name(table, tmp)) == 0);

}

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

Finds the first table name in the given database. */

char*

dict_get_first_table_name_in_db(

/*============================*/

				/* out, own: table name, NULL if

				does not exist; the caller must

				free the memory in the string! */

	const char*	name)	/* in: database name which ends in '/' */

{

	dict_table_t*	sys_tables;

	btr_pcur_t	pcur;

	dict_index_t*	sys_index;

	dtuple_t*	tuple;

	mem_heap_t*	heap;

	dfield_t*	dfield;

	rec_t*		rec;

	byte*		field;

	ulint		len;

	mtr_t		mtr;

	ut_ad(mutex_own(&(dict_sys->mutex)));

	heap = mem_heap_create(1000);

	mtr_start(&mtr);

	sys_tables = dict_table_get_low("SYS_TABLES");

	sys_index = UT_LIST_GET_FIRST(sys_tables->indexes);

	ut_a(!dict_table_is_comp(sys_tables));

	tuple = dtuple_create(heap, 1);

	dfield = dtuple_get_nth_field(tuple, 0);

	dfield_set_data(dfield, name, ut_strlen(name));

	dict_index_copy_types(tuple, sys_index, 1);

	btr_pcur_open_on_user_rec(sys_index, tuple, PAGE_CUR_GE,

				  BTR_SEARCH_LEAF, &pcur, &mtr);

loop:

	rec = btr_pcur_get_rec(&pcur);

	if (!btr_pcur_is_on_user_rec(&pcur, &mtr)) {

		/* Not found */

		btr_pcur_close(&pcur);

		mtr_commit(&mtr);

		mem_heap_free(heap);

		return(NULL);

	}

	field = rec_get_nth_field_old(rec, 0, &len);

	if (len < strlen(name)

	    || ut_memcmp(name, field, strlen(name)) != 0) {

		/* Not found */

		btr_pcur_close(&pcur);

		mtr_commit(&mtr);

		mem_heap_free(heap);

		return(NULL);

	}

	if (!rec_get_deleted_flag(rec, 0)) {

		/* We found one */

		char*	table_name = mem_strdupl((char*) field, len);

		btr_pcur_close(&pcur);

		mtr_commit(&mtr);

		mem_heap_free(heap);

		return(table_name);

	}

	btr_pcur_move_to_next_user_rec(&pcur, &mtr);

	goto loop;

}

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

Prints to the standard output information on all tables found in the data

dictionary system table. */

void

dict_print(void)

/*============*/

{

	dict_table_t*	sys_tables;

	dict_index_t*	sys_index;

	dict_table_t*	table;

	btr_pcur_t	pcur;

	rec_t*		rec;

	byte*		field;

	ulint		len;

	mtr_t		mtr;

	/* Enlarge the fatal semaphore wait timeout during the InnoDB table

	monitor printout */

	mutex_enter(&kernel_mutex);

	srv_fatal_semaphore_wait_threshold += 7200; /* 2 hours */

	mutex_exit(&kernel_mutex);

	mutex_enter(&(dict_sys->mutex));

	mtr_start(&mtr);

	sys_tables = dict_table_get_low("SYS_TABLES");

	sys_index = UT_LIST_GET_FIRST(sys_tables->indexes);

	btr_pcur_open_at_index_side(TRUE, sys_index, BTR_SEARCH_LEAF, &pcur,

				    TRUE, &mtr);

loop:

	btr_pcur_move_to_next_user_rec(&pcur, &mtr);

	rec = btr_pcur_get_rec(&pcur);

	if (!btr_pcur_is_on_user_rec(&pcur, &mtr)) {

		/* end of index */

		btr_pcur_close(&pcur);

		mtr_commit(&mtr);

		mutex_exit(&(dict_sys->mutex));

		/* Restore the fatal semaphore wait timeout */

		mutex_enter(&kernel_mutex);

		srv_fatal_semaphore_wait_threshold -= 7200; /* 2 hours */

		mutex_exit(&kernel_mutex);

		return;

	}

	field = rec_get_nth_field_old(rec, 0, &len);

	if (!rec_get_deleted_flag(rec, 0)) {

		/* We found one */

		char*	table_name = mem_strdupl((char*) field, len);

		btr_pcur_store_position(&pcur, &mtr);

		mtr_commit(&mtr);

		table = dict_table_get_low(table_name);

		mem_free(table_name);

		if (table == NULL) {

			fputs("InnoDB: Failed to load table ", stderr);

			ut_print_namel(stderr, NULL, TRUE, (char*) field, len);

			putc('\n', stderr);

		} else {

			/* The table definition was corrupt if there

			is no index */

			if (dict_table_get_first_index(table)) {

				dict_update_statistics_low(table, TRUE);

			}

			dict_table_print_low(table);

		}

		mtr_start(&mtr);

		btr_pcur_restore_position(BTR_SEARCH_LEAF, &pcur, &mtr);

	}

	goto loop;

}

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

In a crash recovery we already have all the tablespace objects created.

This function compares the space id information in the InnoDB data dictionary

to what we already read with fil_load_single_table_tablespaces().

In a normal startup, we create the tablespace objects for every table in

InnoDB's data dictionary, if the corresponding .ibd file exists.

We also scan the biggest space id, and store it to fil_system. */

void

dict_check_tablespaces_and_store_max_id(

/*====================================*/

	ibool	in_crash_recovery)	/* in: are we doing a crash recovery */

{

	dict_table_t*	sys_tables;

	dict_index_t*	sys_index;

	btr_pcur_t	pcur;

	rec_t*		rec;

	byte*		field;

	ulint		len;

	ulint		space_id;

	ulint		max_space_id	= 0;

	mtr_t		mtr;

	mutex_enter(&(dict_sys->mutex));

	mtr_start(&mtr);

	sys_tables = dict_table_get_low("SYS_TABLES");

	sys_index = UT_LIST_GET_FIRST(sys_tables->indexes);

	ut_a(!dict_table_is_comp(sys_tables));

	btr_pcur_open_at_index_side(TRUE, sys_index, BTR_SEARCH_LEAF, &pcur,

				    TRUE, &mtr);

loop:

	btr_pcur_move_to_next_user_rec(&pcur, &mtr);

	rec = btr_pcur_get_rec(&pcur);

	if (!btr_pcur_is_on_user_rec(&pcur, &mtr)) {

		/* end of index */

		btr_pcur_close(&pcur);

		mtr_commit(&mtr);

		/* We must make the tablespace cache aware of the biggest

		known space id */

		/* printf("Biggest space id in data dictionary %lu\n",

		max_space_id); */

		fil_set_max_space_id_if_bigger(max_space_id);

		mutex_exit(&(dict_sys->mutex));

		return;

	}

	field = rec_get_nth_field_old(rec, 0, &len);

	if (!rec_get_deleted_flag(rec, 0)) {

		/* We found one */

		char*	name = mem_strdupl((char*) field, len);

		field = rec_get_nth_field_old(rec, 9, &len);

		ut_a(len == 4);

		space_id = mach_read_from_4(field);

		btr_pcur_store_position(&pcur, &mtr);

		mtr_commit(&mtr);

		if (space_id != 0 && in_crash_recovery) {

			/* Check that the tablespace (the .ibd file) really

			exists; print a warning to the .err log if not */

			fil_space_for_table_exists_in_mem(space_id, name,

							  FALSE, TRUE, TRUE);

		}

		if (space_id != 0 && !in_crash_recovery) {

			/* It is a normal database startup: create the space

			object and check that the .ibd file exists. */

			fil_open_single_table_tablespace(FALSE, space_id,

							 name);

		}

		mem_free(name);

		if (space_id > max_space_id) {

			max_space_id = space_id;

		}

		mtr_start(&mtr);

		btr_pcur_restore_position(BTR_SEARCH_LEAF, &pcur, &mtr);

	}

	goto loop;

}

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

Loads definitions for table columns. */

static

void

dict_load_columns(

/*==============*/

	dict_table_t*	table,	/* in: table */

	mem_heap_t*	heap)	/* in: memory heap for temporary storage */

{

	dict_table_t*	sys_columns;

	dict_index_t*	sys_index;

	btr_pcur_t	pcur;

	dtuple_t*	tuple;

	dfield_t*	dfield;

	rec_t*		rec;

	byte*		field;

	ulint		len;

	byte*		buf;

	char*		name;

	ulint		mtype;

	ulint		prtype;

	ulint		col_len;

	ulint		i;

	mtr_t		mtr;

	ut_ad(mutex_own(&(dict_sys->mutex)));

	mtr_start(&mtr);

	sys_columns = dict_table_get_low("SYS_COLUMNS");

	sys_index = UT_LIST_GET_FIRST(sys_columns->indexes);

	ut_a(!dict_table_is_comp(sys_columns));

	tuple = dtuple_create(heap, 1);

	dfield = dtuple_get_nth_field(tuple, 0);

	buf = mem_heap_alloc(heap, 8);

	mach_write_to_8(buf, table->id);

	dfield_set_data(dfield, buf, 8);

	dict_index_copy_types(tuple, sys_index, 1);

	btr_pcur_open_on_user_rec(sys_index, tuple, PAGE_CUR_GE,

				  BTR_SEARCH_LEAF, &pcur, &mtr);

	for (i = 0; i + DATA_N_SYS_COLS < (ulint) table->n_cols; i++) {

		rec = btr_pcur_get_rec(&pcur);

		ut_a(btr_pcur_is_on_user_rec(&pcur, &mtr));

		ut_a(!rec_get_deleted_flag(rec, 0));

		field = rec_get_nth_field_old(rec, 0, &len);

		ut_ad(len == 8);

		ut_a(ut_dulint_cmp(table->id, mach_read_from_8(field)) == 0);

		field = rec_get_nth_field_old(rec, 1, &len);

		ut_ad(len == 4);

		ut_a(i == mach_read_from_4(field));

		ut_a(name_of_col_is(sys_columns, sys_index, 4, "NAME"));

		field = rec_get_nth_field_old(rec, 4, &len);

		name = mem_heap_strdupl(heap, (char*) field, len);

		field = rec_get_nth_field_old(rec, 5, &len);

		mtype = mach_read_from_4(field);

		field = rec_get_nth_field_old(rec, 6, &len);

		prtype = mach_read_from_4(field);

		if (dtype_get_charset_coll(prtype) == 0

		    && dtype_is_string_type(mtype)) {

			/* The table was created with < 4.1.2. */

			if (dtype_is_binary_string_type(mtype, prtype)) {

				/* Use the binary collation for

				string columns of binary type. */

				prtype = dtype_form_prtype(

					prtype,

					DATA_DRIZZLE_BINARY_CHARSET_COLL);

			} else {

				/* Use the default charset for

				other than binary columns. */

				prtype = dtype_form_prtype(

					prtype,

					data_mysql_default_charset_coll);

			}

		}

		field = rec_get_nth_field_old(rec, 7, &len);

		col_len = mach_read_from_4(field);

		ut_a(name_of_col_is(sys_columns, sys_index, 8, "PREC"));

		dict_mem_table_add_col(table, heap, name,

				       mtype, prtype, col_len);

		btr_pcur_move_to_next_user_rec(&pcur, &mtr);

	}

	btr_pcur_close(&pcur);

	mtr_commit(&mtr);

}

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

Report that an index field or index for a table has been delete marked. */

static

void

dict_load_report_deleted_index(

/*===========================*/

	const char*	name,	/* in: table name */

	ulint		field)	/* in: index field, or ULINT_UNDEFINED */

{

	fprintf(stderr, "InnoDB: Error: data dictionary entry"

		" for table %s is corrupt!\n", name);

	if (field != ULINT_UNDEFINED) {

		fprintf(stderr,

			"InnoDB: Index field %lu is delete marked.\n", field);

	} else {

		fputs("InnoDB: An index is delete marked.\n", stderr);

	}

}

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

Loads definitions for index fields. */

static

void

dict_load_fields(

/*=============*/

	dict_table_t*	table,	/* in: table */

	dict_index_t*	index,	/* in: index whose fields to load */

	mem_heap_t*	heap)	/* in: memory heap for temporary storage */

{

	dict_table_t*	sys_fields;

	dict_index_t*	sys_index;

	btr_pcur_t	pcur;

	dtuple_t*	tuple;

	dfield_t*	dfield;

	ulint		pos_and_prefix_len;

	ulint		prefix_len;

	rec_t*		rec;

	byte*		field;

	ulint		len;

	byte*		buf;

	ulint		i;

	mtr_t		mtr;

	ut_ad(mutex_own(&(dict_sys->mutex)));

	mtr_start(&mtr);

	sys_fields = dict_table_get_low("SYS_FIELDS");

	sys_index = UT_LIST_GET_FIRST(sys_fields->indexes);

	ut_a(!dict_table_is_comp(sys_fields));

	tuple = dtuple_create(heap, 1);

	dfield = dtuple_get_nth_field(tuple, 0);

	buf = mem_heap_alloc(heap, 8);

	mach_write_to_8(buf, index->id);

	dfield_set_data(dfield, buf, 8);

	dict_index_copy_types(tuple, sys_index, 1);

	btr_pcur_open_on_user_rec(sys_index, tuple, PAGE_CUR_GE,

				  BTR_SEARCH_LEAF, &pcur, &mtr);

	for (i = 0; i < index->n_fields; i++) {

		rec = btr_pcur_get_rec(&pcur);

		ut_a(btr_pcur_is_on_user_rec(&pcur, &mtr));

		if (rec_get_deleted_flag(rec, 0)) {

			dict_load_report_deleted_index(table->name, i);

		}

		field = rec_get_nth_field_old(rec, 0, &len);

		ut_ad(len == 8);

		ut_a(ut_memcmp(buf, field, len) == 0);

		field = rec_get_nth_field_old(rec, 1, &len);

		ut_a(len == 4);

		/* The next field stores the field position in the index

		and a possible column prefix length if the index field

		does not contain the whole column. The storage format is

		like this: if there is at least one prefix field in the index,

		then the HIGH 2 bytes contain the field number (== i) and the

		low 2 bytes the prefix length for the field. Otherwise the

		field number (== i) is contained in the 2 LOW bytes. */

		pos_and_prefix_len = mach_read_from_4(field);

		ut_a((pos_and_prefix_len & 0xFFFFUL) == i

		     || (pos_and_prefix_len & 0xFFFF0000UL) == (i << 16));

		if ((i == 0 && pos_and_prefix_len > 0)

		    || (pos_and_prefix_len & 0xFFFF0000UL) > 0) {

			prefix_len = pos_and_prefix_len & 0xFFFFUL;

		} else {

			prefix_len = 0;

		}

		ut_a(name_of_col_is(sys_fields, sys_index, 4, "COL_NAME"));

		field = rec_get_nth_field_old(rec, 4, &len);

		dict_mem_index_add_field(index,

					 mem_heap_strdupl(heap,

							  (char*) field, len),

					 prefix_len);

		btr_pcur_move_to_next_user_rec(&pcur, &mtr);

	}

	btr_pcur_close(&pcur);

	mtr_commit(&mtr);

}

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

Loads definitions for table indexes. Adds them to the data dictionary

cache. */

static

ulint

dict_load_indexes(

/*==============*/

				/* out: DB_SUCCESS if ok, DB_CORRUPTION

				if corruption of dictionary table or

				DB_UNSUPPORTED if table has unknown index

				type */

	dict_table_t*	table,	/* in: table */

	mem_heap_t*	heap)	/* in: memory heap for temporary storage */

{

	dict_table_t*	sys_indexes;

	dict_index_t*	sys_index;

	dict_index_t*	index;

	btr_pcur_t	pcur;

	dtuple_t*	tuple;

	dfield_t*	dfield;

	rec_t*		rec;

	byte*		field;

	ulint		len;

	ulint		name_len;

	char*		name_buf;

	ulint		type;

	ulint		space;

	ulint		page_no;

	ulint		n_fields;

	byte*		buf;

	ibool		is_sys_table;

	dulint		id;

	mtr_t		mtr;

	ulint		error = DB_SUCCESS;

	ut_ad(mutex_own(&(dict_sys->mutex)));

	if ((ut_dulint_get_high(table->id) == 0)

	    && (ut_dulint_get_low(table->id) < DICT_HDR_FIRST_ID)) {

		is_sys_table = TRUE;

	} else {

		is_sys_table = FALSE;

	}

	mtr_start(&mtr);

	sys_indexes = dict_table_get_low("SYS_INDEXES");

	sys_index = UT_LIST_GET_FIRST(sys_indexes->indexes);

	ut_a(!dict_table_is_comp(sys_indexes));

	tuple = dtuple_create(heap, 1);

	dfield = dtuple_get_nth_field(tuple, 0);

	buf = mem_heap_alloc(heap, 8);

	mach_write_to_8(buf, table->id);

	dfield_set_data(dfield, buf, 8);

	dict_index_copy_types(tuple, sys_index, 1);

	btr_pcur_open_on_user_rec(sys_index, tuple, PAGE_CUR_GE,

				  BTR_SEARCH_LEAF, &pcur, &mtr);

	for (;;) {

		if (!btr_pcur_is_on_user_rec(&pcur, &mtr)) {

			break;

		}

		rec = btr_pcur_get_rec(&pcur);

		field = rec_get_nth_field_old(rec, 0, &len);

		ut_ad(len == 8);

		if (ut_memcmp(buf, field, len) != 0) {

			break;

		}

		if (rec_get_deleted_flag(rec, 0)) {

			dict_load_report_deleted_index(table->name,

						       ULINT_UNDEFINED);

			error = DB_CORRUPTION;

			goto func_exit;

		}

		field = rec_get_nth_field_old(rec, 1, &len);

		ut_ad(len == 8);

		id = mach_read_from_8(field);

		ut_a(name_of_col_is(sys_indexes, sys_index, 4, "NAME"));

		field = rec_get_nth_field_old(rec, 4, &name_len);

		name_buf = mem_heap_strdupl(heap, (char*) field, name_len);

		field = rec_get_nth_field_old(rec, 5, &len);

		n_fields = mach_read_from_4(field);

		field = rec_get_nth_field_old(rec, 6, &len);

		type = mach_read_from_4(field);

		field = rec_get_nth_field_old(rec, 7, &len);

		space = mach_read_from_4(field);

		ut_a(name_of_col_is(sys_indexes, sys_index, 8, "PAGE_NO"));

		field = rec_get_nth_field_old(rec, 8, &len);

		page_no = mach_read_from_4(field);

		/* We check for unsupported types first, so that the

		subsequent checks are relevant for the supported types. */

		if (type & ~(DICT_CLUSTERED | DICT_UNIQUE)) {

			fprintf(stderr,

				"InnoDB: Error: unknown type %lu"

				" of index %s of table %s\n",

				(ulong) type, name_buf, table->name);

			error = DB_UNSUPPORTED;

			goto func_exit;

		} else if (page_no == FIL_NULL) {

			fprintf(stderr,

				"InnoDB: Error: trying to load index %s"

				" for table %s\n"

				"InnoDB: but the index tree has been freed!\n",

				name_buf, table->name);

			error = DB_CORRUPTION;

			goto func_exit;

		} else if ((type & DICT_CLUSTERED) == 0

			    && NULL == dict_table_get_first_index(table)) {

			fprintf(stderr,

				"InnoDB: Error: trying to load index %s"

				" for table %s\n"

				"InnoDB: but the first index"

				" is not clustered!\n",

				name_buf, table->name);

			error = DB_CORRUPTION;

			goto func_exit;

		} else if (is_sys_table

			   && ((type & DICT_CLUSTERED)

			       || ((table == dict_sys->sys_tables)

				   && (name_len == (sizeof "ID_IND") - 1)

				   && (0 == ut_memcmp(name_buf,

						      "ID_IND", name_len))))) {

			/* The index was created in memory already at booting

			of the database server */

		} else {

			index = dict_mem_index_create(table->name, name_buf,

						      space, type, n_fields);

			index->id = id;

			dict_load_fields(table, index, heap);

			dict_index_add_to_cache(table, index, page_no);

		}

		btr_pcur_move_to_next_user_rec(&pcur, &mtr);

	}

func_exit:

	btr_pcur_close(&pcur);

	mtr_commit(&mtr);

	return(error);

}

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

Loads a table definition and also all its index definitions, and also

the cluster definition if the table is a member in a cluster. Also loads

all foreign key constraints where the foreign key is in the table or where

a foreign key references columns in this table. Adds all these to the data

dictionary cache. */

dict_table_t*

dict_load_table(

/*============*/

				/* out: table, NULL if does not exist;

				if the table is stored in an .ibd file,

				but the file does not exist,

				then we set the ibd_file_missing flag TRUE

				in the table object we return */

	const char*	name)	/* in: table name in the

				databasename/tablename format */

{

	ibool		ibd_file_missing	= FALSE;

	dict_table_t*	table;

	dict_table_t*	sys_tables;

	btr_pcur_t	pcur;

	dict_index_t*	sys_index;

	dtuple_t*	tuple;

	mem_heap_t*	heap;

	dfield_t*	dfield;

	rec_t*		rec;

	byte*		field;

	ulint		len;

	ulint		space;

	ulint		n_cols;

	ulint		flags;

	ulint		err;

	mtr_t		mtr;

	ut_ad(mutex_own(&(dict_sys->mutex)));

	heap = mem_heap_create(32000);

	mtr_start(&mtr);

	sys_tables = dict_table_get_low("SYS_TABLES");

	sys_index = UT_LIST_GET_FIRST(sys_tables->indexes);

	ut_a(!dict_table_is_comp(sys_tables));

	tuple = dtuple_create(heap, 1);

	dfield = dtuple_get_nth_field(tuple, 0);

	dfield_set_data(dfield, name, ut_strlen(name));

	dict_index_copy_types(tuple, sys_index, 1);

	btr_pcur_open_on_user_rec(sys_index, tuple, PAGE_CUR_GE,

				  BTR_SEARCH_LEAF, &pcur, &mtr);

	rec = btr_pcur_get_rec(&pcur);

	if (!btr_pcur_is_on_user_rec(&pcur, &mtr)

	    || rec_get_deleted_flag(rec, 0)) {

		/* Not found */

err_exit:

		btr_pcur_close(&pcur);

		mtr_commit(&mtr);

		mem_heap_free(heap);

		return(NULL);

	}

	field = rec_get_nth_field_old(rec, 0, &len);

	/* Check if the table name in record is the searched one */

	if (len != ut_strlen(name) || ut_memcmp(name, field, len) != 0) {

		goto err_exit;

	}

	ut_a(name_of_col_is(sys_tables, sys_index, 9, "SPACE"));

	field = rec_get_nth_field_old(rec, 9, &len);

	space = mach_read_from_4(field);

	/* Check if the tablespace exists and has the right name */

	if (space != 0) {

		if (fil_space_for_table_exists_in_mem(space, name, FALSE,

						      FALSE, FALSE)) {

			/* Ok; (if we did a crash recovery then the tablespace

			can already be in the memory cache) */

		} else {

			/* In >= 4.1.9, InnoDB scans the data dictionary also

			at a normal mysqld startup. It is an error if the

			space object does not exist in memory. */

			ut_print_timestamp(stderr);

			fprintf(stderr,

				"  InnoDB: error: space object of table %s,\n"

				"InnoDB: space id %lu did not exist in memory."

				" Retrying an open.\n",

				name, (ulong)space);

			/* Try to open the tablespace */

			if (!fil_open_single_table_tablespace(TRUE,

							      space, name)) {

				/* We failed to find a sensible tablespace

				file */

				ibd_file_missing = TRUE;

			}

		}

	}

	ut_a(name_of_col_is(sys_tables, sys_index, 4, "N_COLS"));

	field = rec_get_nth_field_old(rec, 4, &len);

	n_cols = mach_read_from_4(field);