~drizzle-trunk/drizzle/development

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

Copyright (C) 1996, 2010, Innobase Oy. All Rights Reserved.

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., 51 Franklin

St, Fifth Floor, Boston, MA 02110-1301 USA

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

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

@file row/row0ins.c

Insert into a table

Created 4/20/1996 Heikki Tuuri

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

#include "row0ins.h"

#ifdef UNIV_NONINL

#include "row0ins.ic"

#endif

#include "ha_prototypes.h"

#include "dict0dict.h"

#include "dict0boot.h"

#include "trx0undo.h"

#include "btr0btr.h"

#include "btr0cur.h"

#include "mach0data.h"

#include "que0que.h"

#include "row0upd.h"

#include "row0sel.h"

#include "row0row.h"

#include "rem0cmp.h"

#include "lock0lock.h"

#include "log0log.h"

#include "eval0eval.h"

#include "data0data.h"

#include "usr0sess.h"

#include "buf0lru.h"

#define	ROW_INS_PREV	1

#define	ROW_INS_NEXT	2

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

IMPORTANT NOTE: Any operation that generates redo MUST check that there

is enough space in the redo log before for that operation. This is

done by calling log_free_check(). The reason for checking the

availability of the redo log space before the start of the operation is

that we MUST not hold any synchonization objects when performing the

check.

If you make a change in this module make sure that no codepath is

introduced where a call to log_free_check() is bypassed. */

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

Creates an insert node struct.

@return	own: insert node struct */

UNIV_INTERN

ins_node_t*

ins_node_create(

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

	ulint		ins_type,	/*!< in: INS_VALUES, ... */

	dict_table_t*	table,		/*!< in: table where to insert */

	mem_heap_t*	heap)		/*!< in: mem heap where created */

{

	ins_node_t*	node;

	node = static_cast<ins_node_t *>(mem_heap_alloc(heap, sizeof(ins_node_t)));

	node->common.type = QUE_NODE_INSERT;

	node->ins_type = ins_type;

	node->state = INS_NODE_SET_IX_LOCK;

	node->table = table;

	node->index = NULL;

	node->entry = NULL;

	node->select = NULL;

	node->trx_id = 0;

	node->entry_sys_heap = mem_heap_create(128);

	node->magic_n = INS_NODE_MAGIC_N;

	return(node);

}

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

Creates an entry template for each index of a table. */

UNIV_INTERN

void

ins_node_create_entry_list(

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

	ins_node_t*	node)	/*!< in: row insert node */

{

	dict_index_t*	index;

	dtuple_t*	entry;

	ut_ad(node->entry_sys_heap);

	UT_LIST_INIT(node->entry_list);

	index = dict_table_get_first_index(node->table);

	while (index != NULL) {

		entry = row_build_index_entry(node->row, NULL, index,

					      node->entry_sys_heap);

		UT_LIST_ADD_LAST(tuple_list, node->entry_list, entry);

		index = dict_table_get_next_index(index);

	}

}

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

Adds system field buffers to a row. */

static

void

row_ins_alloc_sys_fields(

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

	ins_node_t*	node)	/*!< in: insert node */

{

	dtuple_t*		row;

	dict_table_t*		table;

	mem_heap_t*		heap;

	const dict_col_t*	col;

	dfield_t*		dfield;

	byte*			ptr;

	row = node->row;

	table = node->table;

	heap = node->entry_sys_heap;

	ut_ad(row && table && heap);

	ut_ad(dtuple_get_n_fields(row) == dict_table_get_n_cols(table));

	/* 1. Allocate buffer for row id */

	col = dict_table_get_sys_col(table, DATA_ROW_ID);

	dfield = dtuple_get_nth_field(row, dict_col_get_no(col));

	ptr = static_cast<byte *>(mem_heap_zalloc(heap, DATA_ROW_ID_LEN));

	dfield_set_data(dfield, ptr, DATA_ROW_ID_LEN);

	node->row_id_buf = ptr;

	/* 3. Allocate buffer for trx id */

	col = dict_table_get_sys_col(table, DATA_TRX_ID);

	dfield = dtuple_get_nth_field(row, dict_col_get_no(col));

	ptr = static_cast<byte *>(mem_heap_zalloc(heap, DATA_TRX_ID_LEN));

	dfield_set_data(dfield, ptr, DATA_TRX_ID_LEN);

	node->trx_id_buf = ptr;

	/* 4. Allocate buffer for roll ptr */

	col = dict_table_get_sys_col(table, DATA_ROLL_PTR);

	dfield = dtuple_get_nth_field(row, dict_col_get_no(col));

	ptr = static_cast<byte *>(mem_heap_zalloc(heap, DATA_ROLL_PTR_LEN));

	dfield_set_data(dfield, ptr, DATA_ROLL_PTR_LEN);

}

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

Sets a new row to insert for an INS_DIRECT node. This function is only used

if we have constructed the row separately, which is a rare case; this

function is quite slow. */

UNIV_INTERN

void

ins_node_set_new_row(

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

	ins_node_t*	node,	/*!< in: insert node */

	dtuple_t*	row)	/*!< in: new row (or first row) for the node */

{

	node->state = INS_NODE_SET_IX_LOCK;

	node->index = NULL;

	node->entry = NULL;

	node->row = row;

	mem_heap_empty(node->entry_sys_heap);

	/* Create templates for index entries */

	ins_node_create_entry_list(node);

	/* Allocate from entry_sys_heap buffers for sys fields */

	row_ins_alloc_sys_fields(node);

	/* As we allocated a new trx id buf, the trx id should be written

	there again: */

	node->trx_id = 0;

}

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

Does an insert operation by updating a delete-marked existing record

in the index. This situation can occur if the delete-marked record is

kept in the index for consistent reads.

@return	DB_SUCCESS or error code */

static

ulint

row_ins_sec_index_entry_by_modify(

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

	ulint		mode,	/*!< in: BTR_MODIFY_LEAF or BTR_MODIFY_TREE,

				depending on whether mtr holds just a leaf

				latch or also a tree latch */

	btr_cur_t*	cursor,	/*!< in: B-tree cursor */

	const dtuple_t*	entry,	/*!< in: index entry to insert */

	que_thr_t*	thr,	/*!< in: query thread */

	mtr_t*		mtr)	/*!< in: mtr; must be committed before

				latching any further pages */

{

	big_rec_t*	dummy_big_rec;

	mem_heap_t*	heap;

	upd_t*		update;

	rec_t*		rec;

	ulint		err;

	rec = btr_cur_get_rec(cursor);

	ut_ad(!dict_index_is_clust(cursor->index));

	ut_ad(rec_get_deleted_flag(rec,

				   dict_table_is_comp(cursor->index->table)));

	/* We know that in the alphabetical ordering, entry and rec are

	identified. But in their binary form there may be differences if

	there are char fields in them. Therefore we have to calculate the

	difference. */

	heap = mem_heap_create(1024);

	update = row_upd_build_sec_rec_difference_binary(

		cursor->index, entry, rec, thr_get_trx(thr), heap);

	if (mode == BTR_MODIFY_LEAF) {

		/* Try an optimistic updating of the record, keeping changes

		within the page */

		err = btr_cur_optimistic_update(BTR_KEEP_SYS_FLAG, cursor,

						update, 0, thr, mtr);

		switch (err) {

		case DB_OVERFLOW:

		case DB_UNDERFLOW:

		case DB_ZIP_OVERFLOW:

			err = DB_FAIL;

		}

	} else {

		ut_a(mode == BTR_MODIFY_TREE);

		if (buf_LRU_buf_pool_running_out()) {

			err = DB_LOCK_TABLE_FULL;

			goto func_exit;

		}

		err = btr_cur_pessimistic_update(BTR_KEEP_SYS_FLAG, cursor,

						 &heap, &dummy_big_rec, update,

						 0, thr, mtr);

		ut_ad(!dummy_big_rec);

	}

func_exit:

	mem_heap_free(heap);

	return(err);

}

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

Does an insert operation by delete unmarking and updating a delete marked

existing record in the index. This situation can occur if the delete marked

record is kept in the index for consistent reads.

@return	DB_SUCCESS, DB_FAIL, or error code */

static

ulint

row_ins_clust_index_entry_by_modify(

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

	ulint		mode,	/*!< in: BTR_MODIFY_LEAF or BTR_MODIFY_TREE,

				depending on whether mtr holds just a leaf

				latch or also a tree latch */

	btr_cur_t*	cursor,	/*!< in: B-tree cursor */

	mem_heap_t**	heap,	/*!< in/out: pointer to memory heap, or NULL */

	big_rec_t**	big_rec,/*!< out: possible big rec vector of fields

				which have to be stored externally by the

				caller */

	const dtuple_t*	entry,	/*!< in: index entry to insert */

	que_thr_t*	thr,	/*!< in: query thread */

	mtr_t*		mtr)	/*!< in: mtr; must be committed before

				latching any further pages */

{

	rec_t*		rec;

	upd_t*		update;

	ulint		err;

	ut_ad(dict_index_is_clust(cursor->index));

	*big_rec = NULL;

	rec = btr_cur_get_rec(cursor);

	ut_ad(rec_get_deleted_flag(rec,

				   dict_table_is_comp(cursor->index->table)));

	if (!*heap) {

		*heap = mem_heap_create(1024);

	}

	/* Build an update vector containing all the fields to be modified;

	NOTE that this vector may NOT contain system columns trx_id or

	roll_ptr */

	update = row_upd_build_difference_binary(cursor->index, entry, rec,

						 thr_get_trx(thr), *heap);

	if (mode == BTR_MODIFY_LEAF) {

		/* Try optimistic updating of the record, keeping changes

		within the page */

		err = btr_cur_optimistic_update(0, cursor, update, 0, thr,

						mtr);

		switch (err) {

		case DB_OVERFLOW:

		case DB_UNDERFLOW:

		case DB_ZIP_OVERFLOW:

			err = DB_FAIL;

		}

	} else {

		ut_a(mode == BTR_MODIFY_TREE);

		if (buf_LRU_buf_pool_running_out()) {

			return(DB_LOCK_TABLE_FULL);

		}

		err = btr_cur_pessimistic_update(0, cursor,

						 heap, big_rec, update,

						 0, thr, mtr);

	}

	return(err);

}

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

Returns TRUE if in a cascaded update/delete an ancestor node of node

updates (not DELETE, but UPDATE) table.

@return	TRUE if an ancestor updates table */

static

ibool

row_ins_cascade_ancestor_updates_table(

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

	que_node_t*	node,	/*!< in: node in a query graph */

	dict_table_t*	table)	/*!< in: table */

{

	que_node_t*	parent;

	upd_node_t*	upd_node;

	parent = que_node_get_parent(node);

	while (que_node_get_type(parent) == QUE_NODE_UPDATE) {

		upd_node = static_cast<upd_node_t *>(parent);

		if (upd_node->table == table && upd_node->is_delete == FALSE) {

			return(TRUE);

		}

		parent = que_node_get_parent(parent);

		ut_a(parent);

	}

	return(FALSE);

}

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

Returns the number of ancestor UPDATE or DELETE nodes of a

cascaded update/delete node.

@return	number of ancestors */

static

ulint

row_ins_cascade_n_ancestors(

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

	que_node_t*	node)	/*!< in: node in a query graph */

{

	que_node_t*	parent;

	ulint		n_ancestors = 0;

	parent = que_node_get_parent(node);

	while (que_node_get_type(parent) == QUE_NODE_UPDATE) {

		n_ancestors++;

		parent = que_node_get_parent(parent);

		ut_a(parent);

	}

	return(n_ancestors);

}

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

Calculates the update vector node->cascade->update for a child table in

a cascaded update.

@return number of fields in the calculated update vector; the value

can also be 0 if no foreign key fields changed; the returned value is

ULINT_UNDEFINED if the column type in the child table is too short to

fit the new value in the parent table: that means the update fails */

static

ulint

row_ins_cascade_calc_update_vec(

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

	upd_node_t*	node,		/*!< in: update node of the parent

					table */

	dict_foreign_t*	foreign,	/*!< in: foreign key constraint whose

					type is != 0 */

	mem_heap_t*	heap)		/*!< in: memory heap to use as

					temporary storage */

{

	upd_node_t*	cascade		= node->cascade_node;

	dict_table_t*	table		= foreign->foreign_table;

	dict_index_t*	index		= foreign->foreign_index;

	upd_t*		update;

	upd_field_t*	ufield;

	dict_table_t*	parent_table;

	dict_index_t*	parent_index;

	upd_t*		parent_update;

	upd_field_t*	parent_ufield;

	ulint		n_fields_updated;

	ulint		parent_field_no;

	ulint		i;

	ulint		j;

	ut_a(node);

	ut_a(foreign);

	ut_a(cascade);

	ut_a(table);

	ut_a(index);

	/* Calculate the appropriate update vector which will set the fields

	in the child index record to the same value (possibly padded with

	spaces if the column is a fixed length CHAR or FIXBINARY column) as

	the referenced index record will get in the update. */

	parent_table = node->table;

	ut_a(parent_table == foreign->referenced_table);

	parent_index = foreign->referenced_index;

	parent_update = node->update;

	update = cascade->update;

	update->info_bits = 0;

	update->n_fields = foreign->n_fields;

	n_fields_updated = 0;

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

		parent_field_no = dict_table_get_nth_col_pos(

			parent_table,

			dict_index_get_nth_col_no(parent_index, i));

		for (j = 0; j < parent_update->n_fields; j++) {

			parent_ufield = parent_update->fields + j;

			if (parent_ufield->field_no == parent_field_no) {

				ulint			min_size;

				const dict_col_t*	col;

				ulint			ufield_len;

				col = dict_index_get_nth_col(index, i);

				/* A field in the parent index record is

				updated. Let us make the update vector

				field for the child table. */

				ufield = update->fields + n_fields_updated;

				ufield->field_no

					= dict_table_get_nth_col_pos(

					table, dict_col_get_no(col));

				ufield->exp = NULL;

				ufield->new_val = parent_ufield->new_val;

				ufield_len = dfield_get_len(&ufield->new_val);

				/* Clear the "external storage" flag */

				dfield_set_len(&ufield->new_val, ufield_len);

				/* Do not allow a NOT NULL column to be

				updated as NULL */

				if (dfield_is_null(&ufield->new_val)

				    && (col->prtype & DATA_NOT_NULL)) {

					return(ULINT_UNDEFINED);

				}

				/* If the new value would not fit in the

				column, do not allow the update */

				if (!dfield_is_null(&ufield->new_val)

				    && dtype_get_at_most_n_mbchars(

					col->prtype, col->mbminmaxlen,

					col->len,

					ufield_len,

					static_cast<const char *>(dfield_get_data(&ufield->new_val)))

				    < ufield_len) {

					return(ULINT_UNDEFINED);

				}

				/* If the parent column type has a different

				length than the child column type, we may

				need to pad with spaces the new value of the

				child column */

				min_size = dict_col_get_min_size(col);

				/* Because UNIV_SQL_NULL (the marker

				of SQL NULL values) exceeds all possible

				values of min_size, the test below will

				not hold for SQL NULL columns. */

				if (min_size > ufield_len) {

					byte*	pad;

					ulint	pad_len;

					byte*	padded_data;

					ulint	mbminlen;

					padded_data = static_cast<unsigned char *>(mem_heap_alloc(

						heap, min_size));

					pad = padded_data + ufield_len;

					pad_len = min_size - ufield_len;

					memcpy(padded_data,

					       dfield_get_data(&ufield

							       ->new_val),

					       ufield_len);

					mbminlen = dict_col_get_mbminlen(col);

					ut_ad(!(ufield_len % mbminlen));

					ut_ad(!(min_size % mbminlen));

					if (mbminlen == 1

					    && dtype_get_charset_coll(

						    col->prtype)

					    == DATA_MYSQL_BINARY_CHARSET_COLL) {

						/* Do not pad BINARY columns */

						return(ULINT_UNDEFINED);

					}

					row_mysql_pad_col(mbminlen,

							  pad, pad_len);

					dfield_set_data(&ufield->new_val,

							padded_data, min_size);

				}

				n_fields_updated++;

			}

		}

	}

	update->n_fields = n_fields_updated;

	return(n_fields_updated);

}

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

Set detailed error message associated with foreign key errors for

the given transaction. */

static

void

row_ins_set_detailed(

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

	trx_t*		trx,		/*!< in: transaction */

	dict_foreign_t*	foreign)	/*!< in: foreign key constraint */

{

	mutex_enter(&srv_misc_tmpfile_mutex);

	rewind(srv_misc_tmpfile);

	if (os_file_set_eof(srv_misc_tmpfile)) {

		ut_print_name(srv_misc_tmpfile, trx, TRUE,

			      foreign->foreign_table_name);

		dict_print_info_on_foreign_key_in_create_format(

			srv_misc_tmpfile, trx, foreign, FALSE);

		trx_set_detailed_error_from_file(trx, srv_misc_tmpfile);

	} else {

		trx_set_detailed_error(trx, "temp file operation failed");

	}

	mutex_exit(&srv_misc_tmpfile_mutex);

}

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

Reports a foreign key error associated with an update or a delete of a

parent table index entry. */

static

void

row_ins_foreign_report_err(

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

	const char*	errstr,		/*!< in: error string from the viewpoint

					of the parent table */

	que_thr_t*	thr,		/*!< in: query thread whose run_node

					is an update node */

	dict_foreign_t*	foreign,	/*!< in: foreign key constraint */

	const rec_t*	rec,		/*!< in: a matching index record in the

					child table */

	const dtuple_t*	entry)		/*!< in: index entry in the parent

					table */

{

	FILE*	ef	= dict_foreign_err_file;

	trx_t*	trx	= thr_get_trx(thr);

	row_ins_set_detailed(trx, foreign);

	mutex_enter(&dict_foreign_err_mutex);

	rewind(ef);

	ut_print_timestamp(ef);

	fputs(" Transaction:\n", ef);

	trx_print(ef, trx, 600);

	fputs("Foreign key constraint fails for table ", ef);

	ut_print_name(ef, trx, TRUE, foreign->foreign_table_name);

	fputs(":\n", ef);

	dict_print_info_on_foreign_key_in_create_format(ef, trx, foreign,

							TRUE);

	putc('\n', ef);

	fputs(errstr, ef);

	fputs(" in parent table, in index ", ef);

	ut_print_name(ef, trx, FALSE, foreign->referenced_index->name);

	if (entry) {

		fputs(" tuple:\n", ef);

		dtuple_print(ef, entry);

	}

	fputs("\nBut in child table ", ef);

	ut_print_name(ef, trx, TRUE, foreign->foreign_table_name);

	fputs(", in index ", ef);

	ut_print_name(ef, trx, FALSE, foreign->foreign_index->name);

	if (rec) {

		fputs(", there is a record:\n", ef);

		rec_print(ef, rec, foreign->foreign_index);

	} else {

		fputs(", the record is not available\n", ef);

	}

	putc('\n', ef);

	mutex_exit(&dict_foreign_err_mutex);

}

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

Reports a foreign key error to dict_foreign_err_file when we are trying

to add an index entry to a child table. Note that the adding may be the result

of an update, too. */

static

void

row_ins_foreign_report_add_err(

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

	trx_t*		trx,		/*!< in: transaction */

	dict_foreign_t*	foreign,	/*!< in: foreign key constraint */

	const rec_t*	rec,		/*!< in: a record in the parent table:

					it does not match entry because we

					have an error! */

	const dtuple_t*	entry)		/*!< in: index entry to insert in the

					child table */

{

	FILE*	ef	= dict_foreign_err_file;

	row_ins_set_detailed(trx, foreign);

	mutex_enter(&dict_foreign_err_mutex);

	rewind(ef);

	ut_print_timestamp(ef);

	fputs(" Transaction:\n", ef);

	trx_print(ef, trx, 600);

	fputs("Foreign key constraint fails for table ", ef);

	ut_print_name(ef, trx, TRUE, foreign->foreign_table_name);

	fputs(":\n", ef);

	dict_print_info_on_foreign_key_in_create_format(ef, trx, foreign,

							TRUE);

	fputs("\nTrying to add in child table, in index ", ef);

	ut_print_name(ef, trx, FALSE, foreign->foreign_index->name);

	if (entry) {

		fputs(" tuple:\n", ef);

		/* TODO: DB_TRX_ID and DB_ROLL_PTR may be uninitialized.

		It would be better to only display the user columns. */

		dtuple_print(ef, entry);

	}

	fputs("\nBut in parent table ", ef);

	ut_print_name(ef, trx, TRUE, foreign->referenced_table_name);

	fputs(", in index ", ef);

	ut_print_name(ef, trx, FALSE, foreign->referenced_index->name);

	fputs(",\nthe closest match we can find is record:\n", ef);

	if (rec && page_rec_is_supremum(rec)) {

		/* If the cursor ended on a supremum record, it is better

		to report the previous record in the error message, so that

		the user gets a more descriptive error message. */

		rec = page_rec_get_prev_const(rec);

	}

	if (rec) {

		rec_print(ef, rec, foreign->referenced_index);

	}

	putc('\n', ef);

	mutex_exit(&dict_foreign_err_mutex);

}

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

Invalidate the query cache for the given table. */

static

void

row_ins_invalidate_query_cache(

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

	que_thr_t*	unused,		/*!< in: query thread whose run_node

					is an update node */

	const char*	name)		/*!< in: table name prefixed with

					database name and a '/' character */

{

	char*	buf;

	char*	ptr;

	ulint	len = strlen(name) + 1;

        (void)unused;

	buf = mem_strdupl(name, len);

	ptr = strchr(buf, '/');

	ut_a(ptr);

	*ptr = '\0';

	mem_free(buf);

}

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

Perform referential actions or checks when a parent row is deleted or updated

and the constraint had an ON DELETE or ON UPDATE condition which was not

RESTRICT.

@return	DB_SUCCESS, DB_LOCK_WAIT, or error code */

static

ulint

row_ins_foreign_check_on_constraint(

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

	que_thr_t*	thr,		/*!< in: query thread whose run_node

					is an update node */

	dict_foreign_t*	foreign,	/*!< in: foreign key constraint whose

					type is != 0 */

	btr_pcur_t*	pcur,		/*!< in: cursor placed on a matching

					index record in the child table */

	dtuple_t*	entry,		/*!< in: index entry in the parent

					table */

	mtr_t*		mtr)		/*!< in: mtr holding the latch of pcur

					page */

{

	upd_node_t*	node;

	upd_node_t*	cascade;

	dict_table_t*	table		= foreign->foreign_table;

	dict_index_t*	index;

	dict_index_t*	clust_index;

	dtuple_t*	ref;

	mem_heap_t*	upd_vec_heap	= NULL;

	const rec_t*	rec;

	const rec_t*	clust_rec;

	const buf_block_t* clust_block;

	upd_t*		update;

	ulint		n_to_update;

	ulint		err;

	ulint		i;

	trx_t*		trx;

	mem_heap_t*	tmp_heap	= NULL;

	ut_a(thr);

	ut_a(foreign);

	ut_a(pcur);

	ut_a(mtr);

	trx = thr_get_trx(thr);

	/* Since we are going to delete or update a row, we have to invalidate

	the MySQL query cache for table. A deadlock of threads is not possible

	here because the caller of this function does not hold any latches with

	the sync0sync.h rank above the kernel mutex. The query cache mutex has