Merge istruewing@bk-internal.mysql.com:/home/bk/mysql-5.0

into  mysql.com:/home/mydev/mysql-5.0-bug10932


sql/mysql_priv.h:
  Auto merged

33 files changed, 702 insertions, 524 deletions

diff --git a/innobase/btr/btr0sea.c b/innobase/btr/btr0sea.c
index 7a4e92a672a..9b1e93fe700 100644
--- a/innobase/btr/btr0sea.c
+++ b/innobase/btr/btr0sea.c
@@ -904,6 +904,7 @@ btr_search_drop_page_hash_index(
 	ulint*		folds;
 	ulint		i;
 	mem_heap_t*	heap;
+	dict_index_t*	index;
 	ulint*		offsets;
 
 #ifdef UNIV_SYNC_DEBUG
@@ -932,11 +933,16 @@ btr_search_drop_page_hash_index(
 
 	n_fields = block->curr_n_fields;
 	n_bytes = block->curr_n_bytes;
+	index = block->index;
 
-	ut_a(n_fields + n_bytes > 0);
+	/* NOTE: The fields of block must not be accessed after
+	releasing btr_search_latch, as the index page might only
+	be s-latched! */
 
 	rw_lock_s_unlock(&btr_search_latch);
 	
+	ut_a(n_fields + n_bytes > 0);
+
 	n_recs = page_get_n_recs(page);
 
 	/* Calculate and cache fold values into an array for fast deletion
@@ -949,14 +955,6 @@ btr_search_drop_page_hash_index(
 	rec = page_get_infimum_rec(page);
 	rec = page_rec_get_next(rec);
 
-	if (!page_rec_is_supremum(rec)) {
-		ut_a(n_fields <= rec_get_n_fields(rec, block->index));
-
-		if (n_bytes > 0) {
-			ut_a(n_fields < rec_get_n_fields(rec, block->index));
-		}
-	}
-
 	tree_id = btr_page_get_index_id(page);
 	
 	prev_fold = 0;
@@ -964,18 +962,12 @@ btr_search_drop_page_hash_index(
 	heap = NULL;
 	offsets = NULL;
 
-	if (block->index == NULL) {
-
-		mem_analyze_corruption((byte*)block);
-
-		ut_a(block->index != NULL);
-	}
-
 	while (!page_rec_is_supremum(rec)) {
 		/* FIXME: in a mixed tree, not all records may have enough
 		ordering fields: */
-		offsets = rec_get_offsets(rec, block->index,
-				offsets, n_fields + (n_bytes > 0), &heap);
+		offsets = rec_get_offsets(rec, index, offsets,
+					n_fields + (n_bytes > 0), &heap);
+		ut_a(rec_offs_n_fields(offsets) == n_fields + (n_bytes > 0));
 		fold = rec_fold(rec, offsets, n_fields, n_bytes, tree_id);
 
 		if (fold == prev_fold && prev_fold != 0) {
diff --git a/innobase/dict/dict0dict.c b/innobase/dict/dict0dict.c
index fb95ffbd80c..8050eebddd8 100644
--- a/innobase/dict/dict0dict.c
+++ b/innobase/dict/dict0dict.c
@@ -2104,8 +2104,11 @@ dict_foreign_find_index(
 	dict_table_t*	table,	/* in: table */
 	const char**	columns,/* in: array of column names */
 	ulint		n_cols,	/* in: number of columns */
-	dict_index_t*	types_idx)/* in: NULL or an index to whose types the
-				column types must match */
+	dict_index_t*	types_idx, /* in: NULL or an index to whose types the
+				   column types must match */
+	ibool		check_charsets)	/* in: whether to check charsets.
+					only has an effect if types_idx !=
+					NULL. */
 {
 #ifndef UNIV_HOTBACKUP
 	dict_index_t*	index;
@@ -2135,7 +2138,8 @@ dict_foreign_find_index(
 
 				if (types_idx && !cmp_types_are_equal(
 				     dict_index_get_nth_type(index, i),
-				     dict_index_get_nth_type(types_idx, i))) {
+				     dict_index_get_nth_type(types_idx, i),
+				     check_charsets)) {
 
 				  	break;
 				}		
@@ -2212,7 +2216,8 @@ dict_foreign_add_to_cache(
 /*======================*/
 					/* out: DB_SUCCESS or error code */
 	dict_foreign_t*	foreign,	/* in, own: foreign key constraint */
-	ibool		check_types)	/* in: TRUE=check type compatibility */
+	ibool		check_charsets)	/* in: TRUE=check charset
+					compatibility */
 {
 	dict_table_t*	for_table;
 	dict_table_t*	ref_table;
@@ -2248,16 +2253,10 @@ dict_foreign_add_to_cache(
 	}
 
 	if (for_in_cache->referenced_table == NULL && ref_table) {
-		dict_index_t*	types_idx;
-		if (check_types) {
-			types_idx = for_in_cache->foreign_index;
-		} else {
-			types_idx = NULL;
-		}
 		index = dict_foreign_find_index(ref_table,
 			(const char**) for_in_cache->referenced_col_names,
 			for_in_cache->n_fields,
-			types_idx);
+			for_in_cache->foreign_index, check_charsets);
 
 		if (index == NULL) {
 			dict_foreign_error_report(ef, for_in_cache,
@@ -2281,16 +2280,10 @@ dict_foreign_add_to_cache(
 	}
 
 	if (for_in_cache->foreign_table == NULL && for_table) {
-		dict_index_t*	types_idx;
-		if (check_types) {
-			types_idx = for_in_cache->referenced_index;
-		} else {
-			types_idx = NULL;
-		}
 		index = dict_foreign_find_index(for_table,
 			(const char**) for_in_cache->foreign_col_names,
 			for_in_cache->n_fields,
-			types_idx);
+			for_in_cache->referenced_index, check_charsets);
 
 		if (index == NULL) {
 			dict_foreign_error_report(ef, for_in_cache,
@@ -3097,7 +3090,7 @@ col_loop1:
 	/* Try to find an index which contains the columns
 	as the first fields and in the right order */
 
-	index = dict_foreign_find_index(table, column_names, i, NULL);
+	index = dict_foreign_find_index(table, column_names, i, NULL, TRUE);
 
 	if (!index) {
 		mutex_enter(&dict_foreign_err_mutex);
@@ -3362,8 +3355,7 @@ try_find_index:
 
 	if (referenced_table) {
 		index = dict_foreign_find_index(referenced_table,
-						column_names, i,
-						foreign->foreign_index);
+			column_names, i, foreign->foreign_index, TRUE);
 		if (!index) {
 			dict_foreign_free(foreign);
 			mutex_enter(&dict_foreign_err_mutex);
diff --git a/innobase/dict/dict0load.c b/innobase/dict/dict0load.c
index 9bafcf33553..3281f9926f9 100644
--- a/innobase/dict/dict0load.c
+++ b/innobase/dict/dict0load.c
@@ -1091,7 +1091,7 @@ dict_load_foreign(
 				/* out: DB_SUCCESS or error code */
 	const char*	id,	/* in: foreign constraint id as a
 				null-terminated string */
-	ibool		check_types)/* in: TRUE=check type compatibility */
+	ibool		check_charsets)/* in: TRUE=check charset compatibility */
 {	
 	dict_foreign_t*	foreign;
 	dict_table_t*	sys_foreign;
@@ -1204,7 +1204,7 @@ dict_load_foreign(
 	a new foreign key constraint but loading one from the data
 	dictionary. */
 
-	return(dict_foreign_add_to_cache(foreign, check_types));
+	return(dict_foreign_add_to_cache(foreign, check_charsets));
 }
 
 /***************************************************************************
@@ -1219,7 +1219,8 @@ dict_load_foreigns(
 /*===============*/
 					/* out: DB_SUCCESS or error code */
 	const char*	table_name,	/* in: table name */
-	ibool		check_types)	/* in: TRUE=check type compatibility */
+	ibool		check_charsets)	/* in: TRUE=check charset
+					compatibility */
 {
 	btr_pcur_t	pcur;
 	mem_heap_t* 	heap;
@@ -1319,7 +1320,7 @@ loop:
 
 	/* Load the foreign constraint definition to the dictionary cache */
 	
-	err = dict_load_foreign(id, check_types);
+	err = dict_load_foreign(id, check_charsets);
 
 	if (err != DB_SUCCESS) {
 		btr_pcur_close(&pcur);
diff --git a/innobase/include/buf0buf.h b/innobase/include/buf0buf.h
index ae8d0411c12..24e7a71c02d 100644
--- a/innobase/include/buf0buf.h
+++ b/innobase/include/buf0buf.h
@@ -745,8 +745,6 @@ struct buf_block_struct{
 					buffer pool which are index pages,
 					but this flag is not set because
 					we do not keep track of all pages */
-	dict_index_t*	index;		/* index for which the adaptive
-					hash index has been created */
 	/* 2. Page flushing fields */
 
 	UT_LIST_NODE_T(buf_block_t) flush_list;
@@ -833,7 +831,7 @@ struct buf_block_struct{
 					records with the same prefix should be
 					indexed in the hash index */
 					
-	/* The following 4 fields are protected by btr_search_latch: */
+	/* The following 6 fields are protected by btr_search_latch: */
 
 	ibool		is_hashed;	/* TRUE if hash index has already been
 					built on this page; note that it does
@@ -850,6 +848,12 @@ struct buf_block_struct{
 	ulint		curr_side;	/* BTR_SEARCH_LEFT_SIDE or
 					BTR_SEARCH_RIGHT_SIDE in hash
 					indexing */
+	dict_index_t*	index;		/* Index for which the adaptive
+					hash index has been created.
+					This field may only be modified
+					while holding an s-latch or x-latch
+					on block->lock and an x-latch on
+					btr_search_latch. */
 	/* 6. Debug fields */
 #ifdef UNIV_SYNC_DEBUG
 	rw_lock_t	debug_latch;	/* in the debug version, each thread
diff --git a/innobase/include/dict0dict.h b/innobase/include/dict0dict.h
index 5215d51cabe..4396611e529 100644
--- a/innobase/include/dict0dict.h
+++ b/innobase/include/dict0dict.h
@@ -197,7 +197,8 @@ dict_foreign_add_to_cache(
 /*======================*/
 					/* out: DB_SUCCESS or error code */
 	dict_foreign_t*	foreign,	/* in, own: foreign key constraint */
-	ibool		check_types);	/* in: TRUE=check type compatibility */
+	ibool		check_charsets);/* in: TRUE=check charset
+					compatibility */
 /*************************************************************************
 Checks if a table is referenced by foreign keys. */
 
diff --git a/innobase/include/dict0load.h b/innobase/include/dict0load.h
index f13620bc6e8..741123614ab 100644
--- a/innobase/include/dict0load.h
+++ b/innobase/include/dict0load.h
@@ -82,7 +82,8 @@ dict_load_foreigns(
 /*===============*/
 					/* out: DB_SUCCESS or error code */
 	const char*	table_name,	/* in: table name */
-	ibool		check_types);	/* in: TRUE=check type compatibility */
+	ibool		check_charsets);/* in: TRUE=check charsets
+					compatibility */
 /************************************************************************
 Prints to the standard output information on all tables found in the data
 dictionary system table. */
diff --git a/innobase/include/rem0cmp.h b/innobase/include/rem0cmp.h
index 1b1ee26b809..f6762078cbc 100644
--- a/innobase/include/rem0cmp.h
+++ b/innobase/include/rem0cmp.h
@@ -24,7 +24,8 @@ cmp_types_are_equal(
 				/* out: TRUE if the types are considered
 				equal in comparisons */
 	dtype_t*	type1,	/* in: type 1 */
-	dtype_t*	type2);	/* in: type 2 */
+	dtype_t*	type2,	/* in: type 2 */
+	ibool		check_charsets); /* in: whether to check charsets */
 /*****************************************************************
 This function is used to compare two data fields for which we know the
 data type. */
diff --git a/innobase/rem/rem0cmp.c b/innobase/rem/rem0cmp.c
index 7c33476fb9e..6a463b7d4cf 100644
--- a/innobase/rem/rem0cmp.c
+++ b/innobase/rem/rem0cmp.c
@@ -99,7 +99,8 @@ cmp_types_are_equal(
 				/* out: TRUE if the types are considered
 				equal in comparisons */
 	dtype_t*	type1,	/* in: type 1 */
-	dtype_t*	type2)	/* in: type 2 */
+	dtype_t*	type2,	/* in: type 2 */
+	ibool		check_charsets) /* in: whether to check charsets */
 {
 	if (dtype_is_non_binary_string_type(type1->mtype, type1->prtype)
 	    && dtype_is_non_binary_string_type(type2->mtype, type2->prtype)) {
@@ -107,12 +108,12 @@ cmp_types_are_equal(
 		/* Both are non-binary string types: they can be compared if
 		and only if the charset-collation is the same */
 
-		if (dtype_get_charset_coll(type1->prtype)
-				== dtype_get_charset_coll(type2->prtype)) {
+		if (check_charsets) {
+			return(dtype_get_charset_coll(type1->prtype)
+			    == dtype_get_charset_coll(type2->prtype));
+		} else {
 			return(TRUE);
 		}
-
-		return(FALSE);
         }
 
 	if (dtype_is_binary_string_type(type1->mtype, type1->prtype)
diff --git a/innobase/row/row0mysql.c b/innobase/row/row0mysql.c
index 82f7daf2ed8..723e305b2ab 100644
--- a/innobase/row/row0mysql.c
+++ b/innobase/row/row0mysql.c
@@ -2132,7 +2132,7 @@ row_table_add_foreign_constraints(
 
 	if (err == DB_SUCCESS) {
 		/* Check that also referencing constraints are ok */
-		err = dict_load_foreigns(name, trx->check_foreigns);
+		err = dict_load_foreigns(name, TRUE);
 	}
 
 	if (err != DB_SUCCESS) {
@@ -3590,7 +3590,8 @@ row_rename_table_for_mysql(
 	mem_heap_t*	heap			= NULL;
 	const char**	constraints_to_drop	= NULL;
 	ulint		n_constraints_to_drop	= 0;
-        ibool           recovering_temp_table   = FALSE;
+	ibool           recovering_temp_table   = FALSE;
+	ibool		old_is_tmp, new_is_tmp;
 	ulint		len;
 	ulint		i;
         ibool		success;
@@ -3630,6 +3631,9 @@ row_rename_table_for_mysql(
 	trx->op_info = "renaming table";
 	trx_start_if_not_started(trx);
 
+	old_is_tmp = row_is_mysql_tmp_table_name(old_name);
+	new_is_tmp = row_is_mysql_tmp_table_name(new_name);
+	
 	if (row_mysql_is_recovered_tmp_table(new_name)) {
 
                 recovering_temp_table = TRUE;
@@ -3676,7 +3680,7 @@ row_rename_table_for_mysql(
 	len = (sizeof str1) + (sizeof str2) + (sizeof str3) + (sizeof str5) - 4
 		+ ut_strlenq(new_name, '\'') + ut_strlenq(old_name, '\'');
 
-	if (row_is_mysql_tmp_table_name(new_name)) {
+	if (new_is_tmp) {
 		db_name_len = dict_get_db_name_len(old_name) + 1;
 
 		/* MySQL is doing an ALTER TABLE command and it renames the
@@ -3829,7 +3833,7 @@ row_rename_table_for_mysql(
 		the table is stored in a single-table tablespace */
 
 		success = dict_table_rename_in_cache(table, new_name,
-				!row_is_mysql_tmp_table_name(new_name));
+				!new_is_tmp);
 		if (!success) {
 			trx->error_state = DB_SUCCESS;
 			trx_general_rollback_for_mysql(trx, FALSE, NULL);
@@ -3846,19 +3850,16 @@ row_rename_table_for_mysql(
 			goto funct_exit;
 		}
 
-		err = dict_load_foreigns(new_name, trx->check_foreigns);
-
-		if (row_is_mysql_tmp_table_name(old_name)) {
+		/* We only want to switch off some of the type checking in
+		an ALTER, not in a RENAME. */
+		
+		err = dict_load_foreigns(new_name,
+			old_is_tmp ? trx->check_foreigns : TRUE);
 
-			/* MySQL is doing an ALTER TABLE command and it
-			renames the created temporary table to the name
-			of the original table. In the ALTER TABLE we maybe
-			created some FOREIGN KEY constraints for the temporary
-			table. But we want to load also the foreign key
-			constraint definitions for the original table name. */
+		if (err != DB_SUCCESS) {
+			ut_print_timestamp(stderr);
 
-			if (err != DB_SUCCESS) {
-	    			ut_print_timestamp(stderr);
+			if (old_is_tmp) {
 				fputs("  InnoDB: Error: in ALTER TABLE ",
 					stderr);
 				ut_print_name(stderr, trx, new_name);
@@ -3866,36 +3867,23 @@ row_rename_table_for_mysql(
 	"InnoDB: has or is referenced in foreign key constraints\n"
 	"InnoDB: which are not compatible with the new table definition.\n",
 					stderr);
-
-				ut_a(dict_table_rename_in_cache(table,
-					old_name, FALSE));
-				trx->error_state = DB_SUCCESS;
-				trx_general_rollback_for_mysql(trx, FALSE,
-									NULL);
-				trx->error_state = DB_SUCCESS;
-			}
-		} else {
-			if (err != DB_SUCCESS) {
-
-	    			ut_print_timestamp(stderr);
-
+			} else {
 				fputs(
 				"  InnoDB: Error: in RENAME TABLE table ",
 					stderr);
 				ut_print_name(stderr, trx, new_name);
 				fputs("\n"
-     "InnoDB: is referenced in foreign key constraints\n"
-     "InnoDB: which are not compatible with the new table definition.\n",
+	"InnoDB: is referenced in foreign key constraints\n"
+	"InnoDB: which are not compatible with the new table definition.\n",
 					stderr);
-     
-				ut_a(dict_table_rename_in_cache(table,
-					old_name, FALSE));
-						
-				trx->error_state = DB_SUCCESS;
-				trx_general_rollback_for_mysql(trx, FALSE,
-									NULL);
-				trx->error_state = DB_SUCCESS;
 			}
+
+			ut_a(dict_table_rename_in_cache(table,
+					old_name, FALSE));
+			trx->error_state = DB_SUCCESS;
+			trx_general_rollback_for_mysql(trx, FALSE,
+				NULL);
+			trx->error_state = DB_SUCCESS;
 		}
 	}
 funct_exit:	
diff --git a/mysql-test/r/innodb.result b/mysql-test/r/innodb.result
index aabc83a71b8..0e0e6a20770 100644
--- a/mysql-test/r/innodb.result
+++ b/mysql-test/r/innodb.result
@@ -2437,7 +2437,9 @@ a	b
 20	NULL
 drop table t1;
 create table t1 (v varchar(65530), key(v));
-ERROR HY000: Can't create table './test/t1' (errno: 139)
+Warnings:
+Warning	1071	Specified key was too long; max key length is 767 bytes
+drop table t1;
 create table t1 (v varchar(65536));
 Warnings:
 Note	1246	Converting column 'v' from VARCHAR to TEXT
@@ -2577,22 +2579,49 @@ create table t8 (col1 blob, index(col1(767)))
 character set = latin1 engine = innodb;
 create table t9 (col1 varchar(512), col2 varchar(512), index(col1, col2))
 character set = latin1 engine = innodb;
+show create table t9;
+Table	Create Table
+t9	CREATE TABLE `t9` (
+  `col1` varchar(512) default NULL,
+  `col2` varchar(512) default NULL,
+  KEY `col1` (`col1`,`col2`)
+) ENGINE=InnoDB DEFAULT CHARSET=latin1
 drop table t1, t2, t3, t4, t5, t6, t7, t8, t9;
-create table t1 (col1 varchar(768), index (col1))
+create table t1 (col1 varchar(768), index(col1))
 character set = latin1 engine = innodb;
-ERROR HY000: Can't create table './test/t1.frm' (errno: 139)
-create table t2 (col1 varchar(768) primary key)
+Warnings:
+Warning	1071	Specified key was too long; max key length is 767 bytes
+create table t2 (col1 varbinary(768), index(col1))
 character set = latin1 engine = innodb;
-ERROR HY000: Can't create table './test/t2.frm' (errno: 139)
-create table t3 (col1 varbinary(768) primary key)
+Warnings:
+Warning	1071	Specified key was too long; max key length is 767 bytes
+create table t3 (col1 text, index(col1(768)))
 character set = latin1 engine = innodb;
-ERROR HY000: Can't create table './test/t3.frm' (errno: 139)
-create table t4 (col1 text, index(col1(768)))
+Warnings:
+Warning	1071	Specified key was too long; max key length is 767 bytes
+create table t4 (col1 blob, index(col1(768)))
 character set = latin1 engine = innodb;
-ERROR HY000: Can't create table './test/t4.frm' (errno: 139)
-create table t5 (col1 blob, index(col1(768)))
+Warnings:
+Warning	1071	Specified key was too long; max key length is 767 bytes
+show create table t1;
+Table	Create Table
+t1	CREATE TABLE `t1` (
+  `col1` varchar(768) default NULL,
+  KEY `col1` (`col1`(767))
+) ENGINE=InnoDB DEFAULT CHARSET=latin1
+drop table t1, t2, t3, t4;
+create table t1 (col1 varchar(768) primary key)
+character set = latin1 engine = innodb;
+ERROR 42000: Specified key was too long; max key length is 767 bytes
+create table t2 (col1 varbinary(768) primary key)
+character set = latin1 engine = innodb;
+ERROR 42000: Specified key was too long; max key length is 767 bytes
+create table t3 (col1 text, primary key(col1(768)))
+character set = latin1 engine = innodb;
+ERROR 42000: Specified key was too long; max key length is 767 bytes
+create table t4 (col1 blob, primary key(col1(768)))
 character set = latin1 engine = innodb;
-ERROR HY000: Can't create table './test/t5.frm' (errno: 139)
+ERROR 42000: Specified key was too long; max key length is 767 bytes
 CREATE TABLE t1
 (
 id INT PRIMARY KEY
@@ -2772,6 +2801,38 @@ insert into t2 values (4,_ucs2 0x05612020,_ucs2 0x05612020,'taken');
 drop table t1;
 drop table t2;
 commit;
+set foreign_key_checks=0;
+create table t2 (a int primary key, b int, foreign key (b) references t1(a)) engine = innodb;
+create table t1(a char(10) primary key, b varchar(20)) engine = innodb;
+ERROR HY000: Can't create table './test/t1.frm' (errno: 150)
+set foreign_key_checks=1;
+drop table t2;
+set foreign_key_checks=0;
+create table t1(a varchar(10) primary key) engine = innodb DEFAULT CHARSET=latin1;
+create table t2 (a varchar(10), foreign key (a) references t1(a)) engine = innodb DEFAULT CHARSET=utf8;
+ERROR HY000: Can't create table './test/t2.frm' (errno: 150)
+set foreign_key_checks=1;
+drop table t1;
+set foreign_key_checks=0;
+create table t2 (a varchar(10), foreign key (a) references t1(a)) engine = innodb;
+create table t1(a varchar(10) primary key) engine = innodb;
+alter table t1 modify column a int;
+Got one of the listed errors
+set foreign_key_checks=1;
+drop table t2,t1;
+set foreign_key_checks=0;
+create table t2 (a varchar(10), foreign key (a) references t1(a)) engine = innodb DEFAULT CHARSET=latin1;
+create table t1(a varchar(10) primary key) engine = innodb DEFAULT CHARSET=latin1;
+alter table t1 convert to character set utf8;
+set foreign_key_checks=1;
+drop table t2,t1;
+set foreign_key_checks=0;
+create table t2 (a varchar(10), foreign key (a) references t1(a)) engine = innodb DEFAULT CHARSET=latin1;
+create table t3(a varchar(10) primary key) engine = innodb DEFAULT CHARSET=utf8;
+rename table t3 to t1;
+ERROR HY000: Error on rename of './test/t3' to './test/t1' (errno: 150)
+set foreign_key_checks=1;
+drop table t2,t3;
 create table t1 (a varchar(255) character set utf8,
 b varchar(255) character set utf8,
 c varchar(255) character set utf8,
diff --git a/mysql-test/r/sp.result b/mysql-test/r/sp.result
index 6f0a8623a4c..c78ae13d8ee 100644
--- a/mysql-test/r/sp.result
+++ b/mysql-test/r/sp.result
@@ -4153,4 +4153,43 @@ A local variable in a nested compound statement takes precedence over table colu
 a - local variable in a nested compound statement
 A local variable in a nested compound statement takes precedence over table column in cursors
 a - local variable in a nested compound statement
+drop schema if exists mysqltest1|
+Warnings:
+Note	1008	Can't drop database 'mysqltest1'; database doesn't exist
+drop schema if exists mysqltest2|
+Warnings:
+Note	1008	Can't drop database 'mysqltest2'; database doesn't exist
+drop schema if exists mysqltest3|
+Warnings:
+Note	1008	Can't drop database 'mysqltest3'; database doesn't exist
+create schema mysqltest1|
+create schema mysqltest2|
+create schema mysqltest3|
+use mysqltest3|
+create procedure mysqltest1.p1 (out prequestid varchar(100))
+begin
+call mysqltest2.p2('call mysqltest3.p3(1, 2)');
+end|
+create procedure mysqltest2.p2(in psql text)
+begin
+declare lsql text;
+set @lsql= psql;
+prepare lstatement from @lsql;
+execute lstatement;
+deallocate prepare lstatement;
+end|
+create procedure mysqltest3.p3(in p1 int)
+begin
+select p1;
+end|
+call mysqltest1.p1(@rs)|
+ERROR 42000: Incorrect number of arguments for PROCEDURE mysqltest3.p3; expected 1, got 2
+call mysqltest1.p1(@rs)|
+ERROR 42000: Incorrect number of arguments for PROCEDURE mysqltest3.p3; expected 1, got 2
+call mysqltest1.p1(@rs)|
+ERROR 42000: Incorrect number of arguments for PROCEDURE mysqltest3.p3; expected 1, got 2
+drop schema if exists mysqltest1|
+drop schema if exists mysqltest2|
+drop schema if exists mysqltest3|
+use test|
 drop table t1,t2;
diff --git a/mysql-test/r/type_binary.result b/mysql-test/r/type_binary.result
index 1c60bf46608..597defb7a9b 100644
--- a/mysql-test/r/type_binary.result
+++ b/mysql-test/r/type_binary.result
@@ -114,3 +114,26 @@ drop table t1;
 select hex(cast(0x10 as binary(2)));
 hex(cast(0x10 as binary(2)))
 1000
+create table t1 (b binary(2), vb varbinary(2));
+insert into t1 values(0x4120, 0x4120);
+insert into t1 values(0x412020, 0x412020);
+Warnings:
+Warning	1265	Data truncated for column 'b' at row 1
+Warning	1265	Data truncated for column 'vb' at row 1
+drop table t1;
+create table t1 (c char(2), vc varchar(2));
+insert into t1 values(0x4120, 0x4120);
+insert into t1 values(0x412020, 0x412020);
+Warnings:
+Note	1265	Data truncated for column 'vc' at row 1
+drop table t1;
+set @old_sql_mode= @@sql_mode, sql_mode= 'traditional';
+create table t1 (b binary(2), vb varbinary(2));
+insert into t1 values(0x4120, 0x4120);
+insert into t1 values(0x412020, NULL);
+ERROR 22001: Data too long for column 'b' at row 1
+insert into t1 values(NULL, 0x412020);
+ERROR 22001: Data too long for column 'vb' at row 1
+drop table t1;
+set @@sql_mode= @old_sql_mode;
+End of 5.0 tests
diff --git a/mysql-test/t/innodb.test b/mysql-test/t/innodb.test
index a73ecf7c3eb..735deba2b05 100644
--- a/mysql-test/t/innodb.test
+++ b/mysql-test/t/innodb.test
@@ -1356,8 +1356,8 @@ source include/varchar.inc;
 # Clean up filename -- embedded server reports whole path without .frm,
 # regular server reports relative path with .frm (argh!)
 --replace_result \\ / $MYSQL_TEST_DIR . /var/master-data/ / t1.frm t1
---error 1005
 create table t1 (v varchar(65530), key(v));
+drop table t1;
 create table t1 (v varchar(65536));
 show create table t1;
 drop table t1;
@@ -1485,7 +1485,7 @@ CREATE TEMPORARY TABLE t2
 DROP TABLE t1;
 
 #
-# Test that index column max sizes are checked (bug #13315)
+# Test that index column max sizes are honored (bug #13315)
 #
 
 # prefix index
@@ -1512,22 +1512,36 @@ create table t8 (col1 blob, index(col1(767)))
 create table t9 (col1 varchar(512), col2 varchar(512), index(col1, col2))
  character set = latin1 engine = innodb;
 
+show create table t9;
+
 drop table t1, t2, t3, t4, t5, t6, t7, t8, t9;
 
---error 1005
-create table t1 (col1 varchar(768), index (col1))
+# these should have their index length trimmed
+create table t1 (col1 varchar(768), index(col1))
  character set = latin1 engine = innodb;
---error 1005
-create table t2 (col1 varchar(768) primary key)
+create table t2 (col1 varbinary(768), index(col1))
  character set = latin1 engine = innodb;
---error 1005
-create table t3 (col1 varbinary(768) primary key)
+create table t3 (col1 text, index(col1(768)))
  character set = latin1 engine = innodb;
---error 1005
-create table t4 (col1 text, index(col1(768)))
+create table t4 (col1 blob, index(col1(768)))
  character set = latin1 engine = innodb;
---error 1005
-create table t5 (col1 blob, index(col1(768)))
+
+show create table t1;
+
+drop table t1, t2, t3, t4;
+
+# these should be refused
+--error 1071
+create table t1 (col1 varchar(768) primary key)
+ character set = latin1 engine = innodb;
+--error 1071
+create table t2 (col1 varbinary(768) primary key)
+ character set = latin1 engine = innodb;
+--error 1071
+create table t3 (col1 text, primary key(col1(768)))
+ character set = latin1 engine = innodb;
+--error 1071
+create table t4 (col1 blob, primary key(col1(768)))
  character set = latin1 engine = innodb;
 
 #
@@ -1752,6 +1766,56 @@ drop table t1;
 drop table t2;
 commit;
 
+# tests for bugs #9802 and #13778
+
+# test that FKs between invalid types are not accepted
+
+set foreign_key_checks=0;
+create table t2 (a int primary key, b int, foreign key (b) references t1(a)) engine = innodb;
+-- error 1005
+create table t1(a char(10) primary key, b varchar(20)) engine = innodb;
+set foreign_key_checks=1;
+drop table t2;
+
+# test that FKs between different charsets are not accepted in CREATE even
+# when f_k_c is 0
+
+set foreign_key_checks=0;
+create table t1(a varchar(10) primary key) engine = innodb DEFAULT CHARSET=latin1;
+-- error 1005
+create table t2 (a varchar(10), foreign key (a) references t1(a)) engine = innodb DEFAULT CHARSET=utf8;
+set foreign_key_checks=1;
+drop table t1;
+
+# test that invalid datatype conversions with ALTER are not allowed
+
+set foreign_key_checks=0;
+create table t2 (a varchar(10), foreign key (a) references t1(a)) engine = innodb;
+create table t1(a varchar(10) primary key) engine = innodb;
+-- error 1025,1025
+alter table t1 modify column a int;
+set foreign_key_checks=1;
+drop table t2,t1;
+
+# test that charset conversions with ALTER are allowed when f_k_c is 0
+
+set foreign_key_checks=0;
+create table t2 (a varchar(10), foreign key (a) references t1(a)) engine = innodb DEFAULT CHARSET=latin1;
+create table t1(a varchar(10) primary key) engine = innodb DEFAULT CHARSET=latin1;
+alter table t1 convert to character set utf8;
+set foreign_key_checks=1;
+drop table t2,t1;
+
+# test that RENAME does not allow invalid charsets when f_k_c is 0
+
+set foreign_key_checks=0;
+create table t2 (a varchar(10), foreign key (a) references t1(a)) engine = innodb DEFAULT CHARSET=latin1;
+create table t3(a varchar(10) primary key) engine = innodb DEFAULT CHARSET=utf8;
+-- error 1025
+rename table t3 to t1;
+set foreign_key_checks=1;
+drop table t2,t3;
+
 #
 # Test that we can create a large (>1K) key
 #
diff --git a/mysql-test/t/sp.test b/mysql-test/t/sp.test
index 81f7609a468..b3760a1d7f5 100644
--- a/mysql-test/t/sp.test
+++ b/mysql-test/t/sp.test
@@ -4948,6 +4948,52 @@ end|
 call p1("a - stored procedure parameter")|
 
 #
+# A test case for Bug#15392 "Server crashes during prepared statement
+# execute": make sure that stored procedure check for error conditions
+# properly and do not continue execution if an error has been set. 
+#
+# It's necessary to use several DBs because in the original code
+# the successful return of mysql_change_db overrode the error from
+# execution.
+drop schema if exists mysqltest1|
+drop schema if exists mysqltest2|
+drop schema if exists mysqltest3|
+create schema mysqltest1|
+create schema mysqltest2|
+create schema mysqltest3|
+use mysqltest3|
+
+create procedure mysqltest1.p1 (out prequestid varchar(100))
+begin
+  call mysqltest2.p2('call mysqltest3.p3(1, 2)');
+end|
+
+create procedure mysqltest2.p2(in psql text)
+begin
+  declare lsql text;
+  set @lsql= psql;
+  prepare lstatement from @lsql;
+  execute lstatement;
+  deallocate prepare lstatement;
+end|
+
+create procedure mysqltest3.p3(in p1 int)
+begin
+  select p1;
+end|
+
+--error ER_SP_WRONG_NO_OF_ARGS
+call mysqltest1.p1(@rs)|
+--error ER_SP_WRONG_NO_OF_ARGS
+call mysqltest1.p1(@rs)|
+--error ER_SP_WRONG_NO_OF_ARGS
+call mysqltest1.p1(@rs)|
+drop schema if exists mysqltest1|
+drop schema if exists mysqltest2|
+drop schema if exists mysqltest3|
+use test|
+
+#
 # BUG#NNNN: New bug synopsis
 #
 #--disable_warnings
diff --git a/mysql-test/t/type_binary.test b/mysql-test/t/type_binary.test
index 451363c1ae4..1639aff4711 100644
--- a/mysql-test/t/type_binary.test
+++ b/mysql-test/t/type_binary.test
@@ -68,3 +68,27 @@ drop table t1;
 
 # check that cast appends trailing zeros
 select hex(cast(0x10 as binary(2)));
+
+#
+# Bug #14299: BINARY space truncation should cause warning or error
+# 
+create table t1 (b binary(2), vb varbinary(2));
+insert into t1 values(0x4120, 0x4120);
+insert into t1 values(0x412020, 0x412020);
+drop table t1;
+create table t1 (c char(2), vc varchar(2));
+insert into t1 values(0x4120, 0x4120);
+insert into t1 values(0x412020, 0x412020);
+drop table t1;
+
+set @old_sql_mode= @@sql_mode, sql_mode= 'traditional';
+create table t1 (b binary(2), vb varbinary(2));
+insert into t1 values(0x4120, 0x4120);
+--error ER_DATA_TOO_LONG
+insert into t1 values(0x412020, NULL);
+--error ER_DATA_TOO_LONG
+insert into t1 values(NULL, 0x412020);
+drop table t1;
+set @@sql_mode= @old_sql_mode;
+
+--echo End of 5.0 tests
diff --git a/sql/field.cc b/sql/field.cc
index 6de2a731030..3903d8323ad 100644
--- a/sql/field.cc
+++ b/sql/field.cc
@@ -5855,44 +5855,52 @@ int Field_string::store(const char *from,uint length,CHARSET_INFO *cs)
   char buff[STRING_BUFFER_USUAL_SIZE];
   String tmpstr(buff,sizeof(buff), &my_charset_bin);
   uint copy_length;
-  
+
   /* See the comment for Field_long::store(long long) */
   DBUG_ASSERT(table->in_use == current_thd);
-  
+
   /* Convert character set if necessary */
   if (String::needs_conversion(length, cs, field_charset, &not_used))
-  { 
+  {
     uint conv_errors;
     tmpstr.copy(from, length, cs, field_charset, &conv_errors);
     from= tmpstr.ptr();
-    length=  tmpstr.length();
+    length= tmpstr.length();
     if (conv_errors)
       error= 2;
   }
 
-  /* 
-    Make sure we don't break a multibyte sequence
-    as well as don't copy a malformed data.
-  */
+  /* Make sure we don't break a multibyte sequence or copy malformed data. */
   copy_length= field_charset->cset->well_formed_len(field_charset,
                                                     from,from+length,
                                                     field_length/
                                                     field_charset->mbmaxlen,
                                                     &well_formed_error);
   memcpy(ptr,from,copy_length);
-  if (copy_length < field_length)	// Append spaces if shorter
+
+  /* Append spaces if the string was shorter than the field. */
+  if (copy_length < field_length)
     field_charset->cset->fill(field_charset,ptr+copy_length,
-			      field_length-copy_length,
+                              field_length-copy_length,
                               field_charset->pad_char);
-  
+
+  /*
+    Check if we lost any important data (anything in a binary string,
+    or any non-space in others).
+  */
   if ((copy_length < length) && table->in_use->count_cuted_fields)
-  {					// Check if we loosed some info
-    const char *end=from+length;
-    from+= copy_length;
-    from+= field_charset->cset->scan(field_charset, from, end,
-				     MY_SEQ_SPACES);
-    if (from != end)
+  {
+    if (binary())
       error= 2;
+    else
+    {
+      const char *end=from+length;
+      from+= copy_length;
+      from+= field_charset->cset->scan(field_charset, from, end,
+                                       MY_SEQ_SPACES);
+      if (from != end)
+        error= 2;
+    }
   }
   if (error)
   {
@@ -6271,12 +6279,15 @@ int Field_varstring::store(const char *from,uint length,CHARSET_INFO *cs)
   if ((copy_length < length) && table->in_use->count_cuted_fields &&
       !error_code)
   {
-    const char *end= from + length;
-    from+= copy_length;
-    from+= field_charset->cset->scan(field_charset, from, end, MY_SEQ_SPACES);
-    /* If we lost only spaces then produce a NOTE, not a WARNING */
-    if (from == end)
-      level= MYSQL_ERROR::WARN_LEVEL_NOTE;
+    if (!binary())
+    {
+      const char *end= from + length;
+      from+= copy_length;
+      from+= field_charset->cset->scan(field_charset, from, end, MY_SEQ_SPACES);
+      /* If we lost only spaces then produce a NOTE, not a WARNING */
+      if (from == end)
+        level= MYSQL_ERROR::WARN_LEVEL_NOTE;
+    }
     error_code= WARN_DATA_TRUNCATED;
   }
   if (error_code)
diff --git a/sql/ha_innodb.cc b/sql/ha_innodb.cc
index 56e5fd8923f..6c8dc9f81f6 100644
--- a/sql/ha_innodb.cc
+++ b/sql/ha_innodb.cc
@@ -2519,6 +2519,12 @@ ha_innobase::open(
   	DBUG_RETURN(0);
 }
 
+uint
+ha_innobase::max_supported_key_part_length() const
+{
+	return(DICT_MAX_INDEX_COL_LEN - 1);
+}
+
 /**********************************************************************
 Closes a handle to an InnoDB table. */
 
@@ -4675,6 +4681,9 @@ create_index(
 				0, prefix_len);
 	}
 
+	/* Even though we've defined max_supported_key_part_length, we
+	still do our own checking using field_lengths to be absolutely
+	sure we don't create too long indexes. */
 	error = row_create_index_for_mysql(index, trx, field_lengths);
 
 	error = convert_error_code_to_mysql(error, NULL);
diff --git a/sql/ha_innodb.h b/sql/ha_innodb.h
index eb4e10e545f..58051624f89 100644
--- a/sql/ha_innodb.h
+++ b/sql/ha_innodb.h
@@ -110,7 +110,7 @@ class ha_innobase: public handler
 				but currently MySQL does not work with keys
 				whose size is > MAX_KEY_LENGTH */
   	uint max_supported_key_length() const { return 3500; }
-  	uint max_supported_key_part_length() const { return 3500; }
+  	uint max_supported_key_part_length() const;
 	const key_map *keys_to_use_for_scanning() { return &key_map_full; }
   	bool has_transactions()  { return 1;}
 
diff --git a/sql/item.cc b/sql/item.cc
index c45727d1e93..c15fd948b5a 100644
--- a/sql/item.cc
+++ b/sql/item.cc
@@ -5165,7 +5165,7 @@ void Item_trigger_field::setup_field(THD *thd, TABLE *table)
     set field_idx properly.
   */
   (void)find_field_in_table(thd, table, field_name, (uint) strlen(field_name),
-                            0, 0, &field_idx, 0);
+                            0, &field_idx);
   thd->set_query_id= save_set_query_id;
   triggers= table->triggers;
 }
diff --git a/sql/item.h b/sql/item.h
index c562e30d23d..ab9c93e207d 100644
--- a/sql/item.h
+++ b/sql/item.h
@@ -326,6 +326,48 @@ struct Name_resolution_context: Sql_alloc
 };
 
 
+/*
+  Store and restore the current state of a name resolution context.
+*/
+
+class Name_resolution_context_state
+{
+private:
+  TABLE_LIST *save_table_list;
+  TABLE_LIST *save_first_name_resolution_table;
+  TABLE_LIST *save_next_name_resolution_table;
+  bool        save_resolve_in_select_list;
+
+public:
+  TABLE_LIST *save_next_local;
+
+public:
+  /* Save the state of a name resolution context. */
+  void save_state(Name_resolution_context *context, TABLE_LIST *table_list)
+  {
+    save_table_list=                  context->table_list;
+    save_first_name_resolution_table= context->first_name_resolution_table;
+    save_next_name_resolution_table=  (context->first_name_resolution_table) ?
+                                      context->first_name_resolution_table->
+                                               next_name_resolution_table :
+                                      NULL;
+    save_resolve_in_select_list=      context->resolve_in_select_list;
+    save_next_local=                  table_list->next_local;
+  }
+
+  /* Restore a name resolution context from saved state. */
+  void restore_state(Name_resolution_context *context, TABLE_LIST *table_list)
+  {
+    table_list->next_local=                save_next_local;
+    context->table_list=                   save_table_list;
+    context->first_name_resolution_table=  save_first_name_resolution_table;
+    if (context->first_name_resolution_table)
+      context->first_name_resolution_table->
+               next_name_resolution_table= save_next_name_resolution_table;
+    context->resolve_in_select_list=       save_resolve_in_select_list;
+  }
+};
+
 /*************************************************************************/
 
 typedef bool (Item::*Item_processor)(byte *arg);
diff --git a/sql/mysql_priv.h b/sql/mysql_priv.h
index 9b7921b2179..9f391bf26c8 100644
--- a/sql/mysql_priv.h
+++ b/sql/mysql_priv.h
@@ -795,18 +795,15 @@ find_field_in_tables(THD *thd, Item_ident *item,
                      bool check_privileges, bool register_tree_change);
 Field *
 find_field_in_table_ref(THD *thd, TABLE_LIST *table_list,
-                        const char *name, const char *item_name,
-                        const char *table_name, const char *db_name,
-                        uint length, Item **ref,
-                        bool check_grants_table, bool check_grants_view,
-                        bool allow_rowid,
+                        const char *name, uint length,
+                        const char *item_name, const char *db_name,
+                        const char *table_name, Item **ref,
+                        bool check_privileges, bool allow_rowid,
                         uint *cached_field_index_ptr,
                         bool register_tree_change, TABLE_LIST **actual_table);
 Field *
-find_field_in_table(THD *thd, TABLE *table, const char *name,
-                    uint length, bool check_grants, bool allow_rowid,
-                    uint *cached_field_index_ptr,
-                    Security_context *sctx);
+find_field_in_table(THD *thd, TABLE *table, const char *name, uint length,
+                    bool allow_rowid, uint *cached_field_index_ptr);
 
 #ifdef HAVE_OPENSSL
 #include <openssl/des.h>
@@ -922,8 +919,9 @@ create_field * new_create_field(THD *thd, char *field_name, enum_field_types typ
 				uint uint_geom_type);
 void store_position_for_column(const char *name);
 bool add_to_list(THD *thd, SQL_LIST &list,Item *group,bool asc);
-Name_resolution_context *make_join_on_context(THD *thd, TABLE_LIST *left_op,
-                                              TABLE_LIST *right_op);
+bool push_new_name_resolution_context(THD *thd,
+                                      TABLE_LIST *left_op,
+                                      TABLE_LIST *right_op);
 void add_join_on(TABLE_LIST *b,Item *expr);
 void add_join_natural(TABLE_LIST *a,TABLE_LIST *b,List<String> *using_fields);
 bool add_proc_to_list(THD *thd, Item *item);
diff --git a/sql/sp_head.cc b/sql/sp_head.cc
index ff4f898924c..fcd220353fc 100644
--- a/sql/sp_head.cc
+++ b/sql/sp_head.cc
@@ -1137,10 +1137,12 @@ int sp_head::execute(THD *thd)
      original thd->db will then have been freed */
   if (dbchanged)
   {
-    /* No access check when changing back to where we came from.
-       (It would generate an error from mysql_change_db() when olddb=="") */
+    /*
+      No access check when changing back to where we came from.
+      (It would generate an error from mysql_change_db() when olddb=="")
+    */
     if (! thd->killed)
-      ret= mysql_change_db(thd, olddb, 1);
+      ret|= (int) mysql_change_db(thd, olddb, 1);
   }
   m_flags&= ~IS_INVOKED;
   DBUG_PRINT("info", ("first free for 0x%lx --: 0x%lx->0x%lx, level: %lu, flags %x",
@@ -1519,13 +1521,12 @@ int sp_head::execute_procedure(THD *thd, List<Item> *args)
 
 	    suv= new Item_func_set_user_var(guv->get_name(), item);
 	    /*
-	      we do not check suv->fixed, because it can't be fixed after
-	      creation
+              Item_func_set_user_var is not fixed after construction,
+              call fix_fields().
 	    */
-	    suv->fix_fields(thd, &item);
-	    suv->fix_length_and_dec();
-	    suv->check();
-	    suv->update();
+            if ((ret= test(!suv || suv->fix_fields(thd, &item) ||
+                           suv->check() || suv->update())))
+              break;
 	  }
 	}
       }
@@ -2097,7 +2098,7 @@ sp_lex_keeper::reset_lex_and_exec_core(THD *thd, uint *nextp,
 
     cleanup_items() is called in sp_head::execute()
   */
-  return res;
+  return res || thd->net.report_error;
 }
 
 
diff --git a/sql/sql_acl.cc b/sql/sql_acl.cc
index 46be74ae972..cd83efcac2c 100644
--- a/sql/sql_acl.cc
+++ b/sql/sql_acl.cc
@@ -2761,8 +2761,9 @@ bool mysql_table_grant(THD *thd, TABLE_LIST *table_list,
         uint unused_field_idx= NO_CACHED_FIELD_INDEX;
         TABLE_LIST *dummy;
         Field *f=find_field_in_table_ref(thd, table_list, column->column.ptr(),
+                                         column->column.length(),
                                          column->column.ptr(), NULL, NULL,
-                                         column->column.length(), 0, 1, 1, 0,
+                                         NULL, TRUE, FALSE,
                                          &unused_field_idx, FALSE, &dummy);
         if (f == (Field*)0)
         {
@@ -3616,11 +3617,28 @@ err:
 }
 
 
+/*
+  Check column rights in given security context
+
+  SYNOPSIS
+    check_grant_column()
+    thd                  thread handler
+    grant                grant information structure
+    db_name              db name
+    table_name           table  name
+    name                 column name
+    length               column name length
+    sctx                 security context
+
+  RETURN
+    FALSE OK
+    TRUE  access denied
+*/
+
 bool check_grant_column(THD *thd, GRANT_INFO *grant,
 			const char *db_name, const char *table_name,
-			const char *name, uint length, uint show_tables)
+			const char *name, uint length,  Security_context *sctx)
 {
-  Security_context *sctx= thd->security_ctx;
   GRANT_TABLE *grant_table;
   GRANT_COLUMN *grant_column;
   ulong want_access= grant->want_privilege & ~grant->privilege;
@@ -3651,28 +3669,74 @@ bool check_grant_column(THD *thd, GRANT_INFO *grant,
     rw_unlock(&LOCK_grant);
     DBUG_RETURN(0);
   }
-#ifdef NOT_USED
-  if (show_tables && (grant_column || grant->privilege & COL_ACLS))
-  {
-    rw_unlock(&LOCK_grant);			/* purecov: deadcode */
-    DBUG_RETURN(0);				/* purecov: deadcode */
-  }
-#endif
 
 err:
   rw_unlock(&LOCK_grant);
-  if (!show_tables)
+  char command[128];
+  get_privilege_desc(command, sizeof(command), want_access);
+  my_error(ER_COLUMNACCESS_DENIED_ERROR, MYF(0),
+           command,
+           sctx->priv_user,
+           sctx->host_or_ip,
+           name,
+           table_name);
+  DBUG_RETURN(1);
+}
+
+
+/*
+  Check the access right to a column depending on the type of table.
+
+  SYNOPSIS
+    check_column_grant_in_table_ref()
+    thd              thread handler
+    table_ref        table reference where to check the field
+    name             name of field to check
+    length           length of name
+
+  DESCRIPTION
+    Check the access rights to a column depending on the type of table
+    reference where the column is checked. The function provides a
+    generic interface to check column access rights that hides the
+    heterogeneity of the column representation - whether it is a view
+    or a stored table colum.
+
+  RETURN
+    FALSE OK
+    TRUE  access denied
+*/
+
+bool check_column_grant_in_table_ref(THD *thd, TABLE_LIST * table_ref,
+                                     const char *name, uint length)
+{
+  GRANT_INFO *grant;
+  const char *db_name;
+  const char *table_name;
+  Security_context *sctx= test(table_ref->security_ctx) ?
+                          table_ref->security_ctx : thd->security_ctx;
+
+  if (table_ref->view || table_ref->field_translation)
   {
-    char command[128];
-    get_privilege_desc(command, sizeof(command), want_access);
-    my_error(ER_COLUMNACCESS_DENIED_ERROR, MYF(0),
-             command,
-             sctx->priv_user,
-             sctx->host_or_ip,
-             name,
-             table_name);
+    /* View or derived information schema table. */
+    grant= &(table_ref->grant);
+    db_name= table_ref->view_db.str;
+    table_name= table_ref->view_name.str;
   }
-  DBUG_RETURN(1);
+  else
+  {
+    /* Normal or temporary table. */
+    TABLE *table= table_ref->table;
+    grant= &(table->grant);
+    db_name= table->s->db;
+    table_name= table->s->table_name;
+  }
+
+  if (grant->want_privilege)
+    return check_grant_column(thd, grant, db_name, table_name, name,
+                              length, sctx);
+  else
+    return FALSE;
+
 }
 
 
diff --git a/sql/sql_acl.h b/sql/sql_acl.h
index 0e50737f84c..c8fadb73b0c 100644
--- a/sql/sql_acl.h
+++ b/sql/sql_acl.h
@@ -204,7 +204,9 @@ bool check_grant(THD *thd, ulong want_access, TABLE_LIST *tables,
 		 uint show_command, uint number, bool dont_print_error);
 bool check_grant_column (THD *thd, GRANT_INFO *grant,
 			 const char *db_name, const char *table_name,
-			 const char *name, uint length, uint show_command=0);
+			 const char *name, uint length, Security_context *sctx);
+bool check_column_grant_in_table_ref(THD *thd, TABLE_LIST * table_ref,
+                                     const char *name, uint length);
 bool check_grant_all_columns(THD *thd, ulong want_access, GRANT_INFO *grant,
                              const char* db_name, const char *table_name,
                              Field_iterator *fields);
diff --git a/sql/sql_base.cc b/sql/sql_base.cc
index 35f4bd9a05f..0e1c1525c9e 100644
--- a/sql/sql_base.cc
+++ b/sql/sql_base.cc
@@ -2714,47 +2714,6 @@ static void update_field_dependencies(THD *thd, Field *field, TABLE *table)
 }
 
 
-#ifndef NO_EMBEDDED_ACCESS_CHECKS
-/*
-  Check column rights in given security context
-
-  SYNOPSIS
-    check_grant_column_in_sctx()
-    thd                  thread handler
-    grant                grant information structure
-    db                   db name
-    table                table  name
-    name                 column name
-    length               column name length
-    check_grants         need to check grants
-    sctx                 0 or security context
-
-  RETURN
-    FALSE OK
-    TRUE  access denied
-*/
-
-static bool check_grant_column_in_sctx(THD *thd, GRANT_INFO *grant,
-                                       const char *db, const char *table,
-                                       const char *name, uint length,
-                                       bool check_grants,
-                                       Security_context *sctx)
-{
-  if (!check_grants)
-    return FALSE;
-  Security_context *save_security_ctx= thd->security_ctx;
-  bool res;
-  if (sctx)
-  {
-    thd->security_ctx= sctx;
-  }
-  res= check_grant_column(thd, grant, db, table, name, length);
-  thd->security_ctx= save_security_ctx;
-  return res;
-}
-#endif
-
-
 /*
   Find a field by name in a view that uses merge algorithm.
 
@@ -2763,11 +2722,10 @@ static bool check_grant_column_in_sctx(THD *thd, GRANT_INFO *grant,
     thd				thread handler
     table_list			view to search for 'name'
     name			name of field
-    item_name                   name of item if it will be created (VIEW)
     length			length of name
+    item_name                   name of item if it will be created (VIEW)
     ref				expression substituted in VIEW should be passed
                                 using this reference (return view_ref_found)
-    check_grants		do check columns grants for view?
     register_tree_change        TRUE if ref is not stack variable and we
                                 need register changes in item tree
 
@@ -2779,8 +2737,8 @@ static bool check_grant_column_in_sctx(THD *thd, GRANT_INFO *grant,
 
 static Field *
 find_field_in_view(THD *thd, TABLE_LIST *table_list,
-                   const char *name, const char *item_name,
-                   uint length, Item **ref, bool check_grants,
+                   const char *name, uint length,
+                   const char *item_name, Item **ref,
                    bool register_tree_change)
 {
   DBUG_ENTER("find_field_in_view");
@@ -2797,24 +2755,13 @@ find_field_in_view(THD *thd, TABLE_LIST *table_list,
   {
     if (!my_strcasecmp(system_charset_info, field_it.name(), name))
     {
-      if (table_list->schema_table_reformed)
-        /*
-          Translation table items are always Item_fields and fixed already
-          ('mysql_schema_table' function). So we can return ->field. It is
-          used only for 'show & where' commands.
-        */
-        DBUG_RETURN(((Item_field*) (field_it.item()))->field);
-#ifndef NO_EMBEDDED_ACCESS_CHECKS
-      if (check_grant_column_in_sctx(thd, &table_list->grant,
-                                     table_list->view_db.str,
-                                     table_list->view_name.str, name, length,
-                                     check_grants,
-                                     table_list->security_ctx))
-        DBUG_RETURN(WRONG_GRANT);
-#endif
       // in PS use own arena or data will be freed after prepare
       if (register_tree_change)
         arena= thd->activate_stmt_arena_if_needed(&backup);
+      /*
+        create_item() may, or may not create a new Item, depending on
+        the column reference. See create_view_field() for details.
+      */
       Item *item= field_it.create_item(thd);
       if (register_tree_change && arena)
         thd->restore_active_arena(arena, &backup);
@@ -2856,7 +2803,6 @@ find_field_in_view(THD *thd, TABLE_LIST *table_list,
     length		 [in]  length of name
     ref                  [in/out] if 'name' is resolved to a view field, ref is
                                set to point to the found view field
-    check_grants	 [in]  do check columns grants?
     register_tree_change [in]  TRUE if ref is not stack variable and we
                                need register changes in item tree
     actual_table         [out] the original table reference where the field
@@ -2877,8 +2823,7 @@ find_field_in_view(THD *thd, TABLE_LIST *table_list,
 
 static Field *
 find_field_in_natural_join(THD *thd, TABLE_LIST *table_ref, const char *name,
-                           uint length, Item **ref, bool check_grants,
-                           bool register_tree_change,
+                           uint length, Item **ref, bool register_tree_change,
                            TABLE_LIST **actual_table)
 {
   List_iterator_fast<Natural_join_column>
@@ -2903,23 +2848,16 @@ find_field_in_natural_join(THD *thd, TABLE_LIST *table_ref, const char *name,
       break;
   }
 
-#ifndef NO_EMBEDDED_ACCESS_CHECKS
-  if (check_grants && nj_col->check_grants(thd, name, length))
-    DBUG_RETURN(WRONG_GRANT);
-#endif
-
   if (nj_col->view_field)
   {
     Item *item;
-    /*
-      The found field is a view field, we do as in find_field_in_view()
-      and return a pointer to pointer to the Item of that field.
-    */
     if (register_tree_change)
       arena= thd->activate_stmt_arena_if_needed(&backup);
-
+    /*
+      create_item() may, or may not create a new Item, depending on the
+      column reference. See create_view_field() for details.
+    */
     item= nj_col->create_item(thd);
-
     if (register_tree_change && arena)
       thd->restore_active_arena(arena, &backup);
 
@@ -2965,7 +2903,6 @@ find_field_in_natural_join(THD *thd, TABLE_LIST *table_ref, const char *name,
     table			table where to search for the field
     name			name of field
     length			length of name
-    check_grants		do check columns grants?
     allow_rowid			do allow finding of "_rowid" field?
     cached_field_index_ptr	cached position in field list (used to speedup
                                 lookup for fields in prepared tables)
@@ -2977,9 +2914,7 @@ find_field_in_natural_join(THD *thd, TABLE_LIST *table_ref, const char *name,
 
 Field *
 find_field_in_table(THD *thd, TABLE *table, const char *name, uint length,
-                    bool check_grants, bool allow_rowid,
-                    uint *cached_field_index_ptr,
-                    Security_context *sctx)
+                    bool allow_rowid, uint *cached_field_index_ptr)
 {
   Field **field_ptr, *field;
   uint cached_field_index= *cached_field_index_ptr;
@@ -3018,13 +2953,6 @@ find_field_in_table(THD *thd, TABLE *table, const char *name, uint length,
 
   update_field_dependencies(thd, field, table);
 
-#ifndef NO_EMBEDDED_ACCESS_CHECKS
-  if (check_grant_column_in_sctx(thd, &table->grant,
-                                 table->s->db, table->s->table_name,
-                                 name, length,
-                                 check_grants, sctx))
-    field= WRONG_GRANT;
-#endif
   DBUG_RETURN(field);
 }
 
@@ -3037,14 +2965,13 @@ find_field_in_table(THD *thd, TABLE *table, const char *name, uint length,
     thd			   [in]  thread handler
     table_list		   [in]  table reference to search
     name		   [in]  name of field
+    length		   [in]  field length of name
     item_name              [in]  name of item if it will be created (VIEW)
-    table_name             [in]  optional table name that qualifies the field
     db_name                [in]  optional database name that qualifies the
-    length		   [in]  field length of name
+    table_name             [in]  optional table name that qualifies the field
     ref		       [in/out] if 'name' is resolved to a view field, ref
                                  is set to point to the found view field
-    check_grants_table	   [in]  do check columns grants for table?
-    check_grants_view	   [in]  do check columns grants for view?
+    check_privileges       [in]  check privileges
     allow_rowid		   [in]  do allow finding of "_rowid" field?
     cached_field_index_ptr [in]  cached position in field list (used to
                                  speedup lookup for fields in prepared tables)
@@ -3074,11 +3001,11 @@ find_field_in_table(THD *thd, TABLE *table, const char *name, uint length,
 
 Field *
 find_field_in_table_ref(THD *thd, TABLE_LIST *table_list,
-                        const char *name, const char *item_name,
-                        const char *table_name, const char *db_name,
-                        uint length, Item **ref,
-                        bool check_grants_table, bool check_grants_view,
-                        bool allow_rowid, uint *cached_field_index_ptr,
+                        const char *name, uint length,
+                        const char *item_name, const char *db_name,
+                        const char *table_name, Item **ref,
+                        bool check_privileges, bool allow_rowid,
+                        uint *cached_field_index_ptr,
                         bool register_tree_change, TABLE_LIST **actual_table)
 {
   Field *fld;
@@ -3123,8 +3050,7 @@ find_field_in_table_ref(THD *thd, TABLE_LIST *table_list,
   if (table_list->field_translation)
   {
     /* 'table_list' is a view or an information schema table. */
-    if ((fld= find_field_in_view(thd, table_list, name, item_name, length,
-                                 ref, check_grants_view,
+    if ((fld= find_field_in_view(thd, table_list, name, length, item_name, ref,
                                  register_tree_change)))
       *actual_table= table_list;
   }
@@ -3133,20 +3059,9 @@ find_field_in_table_ref(THD *thd, TABLE_LIST *table_list,
     /* 'table_list' is a stored table. */
     DBUG_ASSERT(table_list->table);
     if ((fld= find_field_in_table(thd, table_list->table, name, length,
-                                  check_grants_table, allow_rowid,
-                                  cached_field_index_ptr,
-                                  table_list->security_ctx)))
+                                  allow_rowid,
+                                  cached_field_index_ptr)))
       *actual_table= table_list;
-#ifndef NO_EMBEDDED_ACCESS_CHECKS
-    /* check for views with temporary table algorithm */
-    if (check_grants_view && table_list->view &&
-        fld && fld != WRONG_GRANT &&
-        check_grant_column(thd, &table_list->grant,
-                           table_list->view_db.str,
-                           table_list->view_name.str,
-                           name, length))
-    fld= WRONG_GRANT;
-#endif
   }
   else
   {
@@ -3163,11 +3078,10 @@ find_field_in_table_ref(THD *thd, TABLE_LIST *table_list,
       TABLE_LIST *table;
       while ((table= it++))
       {
-        if ((fld= find_field_in_table_ref(thd, table, name, item_name,
-                                          table_name, db_name, length, ref,
-                                          check_grants_table,
-                                          check_grants_view,
-                                          allow_rowid, cached_field_index_ptr,
+        if ((fld= find_field_in_table_ref(thd, table, name, length, item_name,
+                                          db_name, table_name, ref,
+                                          check_privileges, allow_rowid,
+                                          cached_field_index_ptr,
                                           register_tree_change, actual_table)))
           DBUG_RETURN(fld);
       }
@@ -3180,11 +3094,16 @@ find_field_in_table_ref(THD *thd, TABLE_LIST *table_list,
       directly the top-most NATURAL/USING join.
     */
     fld= find_field_in_natural_join(thd, table_list, name, length, ref,
-                                    /* TIMOUR_TODO: check this with Sanja */
-                                    check_grants_table || check_grants_view,
                                     register_tree_change, actual_table);
   }
 
+#ifndef NO_EMBEDDED_ACCESS_CHECKS
+  /* Check if there are sufficient access rights to the found field. */
+  if (fld && check_privileges &&
+      check_column_grant_in_table_ref(thd, *actual_table, name, length))
+    fld= WRONG_GRANT;
+#endif
+
   DBUG_RETURN(fld);
 }
 
@@ -3266,21 +3185,11 @@ find_field_in_tables(THD *thd, Item_ident *item,
       */
     if (table_ref->table && !table_ref->view)
       found= find_field_in_table(thd, table_ref->table, name, length,
-                                 test(table_ref->table->
-                                      grant.want_privilege) &&
-                                 check_privileges,
-                                 1, &(item->cached_field_index),
-                                 table_ref->security_ctx);
+                                 TRUE, &(item->cached_field_index));
     else
-      found= find_field_in_table_ref(thd, table_ref, name, item->name,
-                                     NULL, NULL, length, ref,
-                                     (table_ref->table &&
-                                      test(table_ref->table->grant.
-                                           want_privilege) &&
-                                      check_privileges),
-                                     (test(table_ref->grant.want_privilege) &&
-                                      check_privileges),
-                                     1, &(item->cached_field_index),
+      found= find_field_in_table_ref(thd, table_ref, name, length, item->name,
+                                     NULL, NULL, ref, check_privileges,
+                                     TRUE, &(item->cached_field_index),
                                      register_tree_change,
                                      &actual_table);
     if (found)
@@ -3320,17 +3229,9 @@ find_field_in_tables(THD *thd, Item_ident *item,
   for (; cur_table != last_table ;
        cur_table= cur_table->next_name_resolution_table)
   {
-    Field *cur_field= find_field_in_table_ref(thd, cur_table, name, item->name,
-                                              table_name, db,
-                                              length, ref,
-                                              (cur_table->table &&
-                                               test(cur_table->table->grant.
-                                                    want_privilege) &&
-                                               check_privileges),
-                                              (test(cur_table->grant.
-                                                    want_privilege)
-                                               && check_privileges),
-                                              allow_rowid,
+    Field *cur_field= find_field_in_table_ref(thd, cur_table, name, length,
+                                              item->name, db, table_name, ref,
+                                              check_privileges, allow_rowid,
                                               &(item->cached_field_index),
                                               register_tree_change,
                                               &actual_table);
@@ -3738,7 +3639,7 @@ mark_common_columns(THD *thd, TABLE_LIST *table_ref_1, TABLE_LIST *table_ref_2,
   {
     bool is_created_1;
     bool found= FALSE;
-    if (!(nj_col_1= it_1.get_or_create_column_ref(thd, &is_created_1)))
+    if (!(nj_col_1= it_1.get_or_create_column_ref(&is_created_1)))
       goto err;
     field_name_1= nj_col_1->name();
 
@@ -3759,7 +3660,7 @@ mark_common_columns(THD *thd, TABLE_LIST *table_ref_1, TABLE_LIST *table_ref_2,
       bool is_created_2;
       Natural_join_column *cur_nj_col_2;
       const char *cur_field_name_2;
-      if (!(cur_nj_col_2= it_2.get_or_create_column_ref(thd, &is_created_2)))
+      if (!(cur_nj_col_2= it_2.get_or_create_column_ref(&is_created_2)))
         goto err;
       cur_field_name_2= cur_nj_col_2->name();
 
@@ -3951,13 +3852,7 @@ store_natural_using_join_columns(THD *thd, TABLE_LIST *natural_using_join,
   /* Append the columns of the first join operand. */
   for (it_1.set(table_ref_1); !it_1.end_of_fields(); it_1.next())
   {
-    if (!(nj_col_1= it_1.get_or_create_column_ref(thd, &is_created)))
-      goto err;
-    /*
-      The following assert checks that mark_common_columns() was run and
-      we created the list table_ref_1->join_columns.
-    */
-    DBUG_ASSERT(!is_created);
+    nj_col_1= it_1.get_natural_column_ref();
     if (nj_col_1->is_common)
     {
       natural_using_join->join_columns->push_back(nj_col_1);
@@ -4003,13 +3898,7 @@ store_natural_using_join_columns(THD *thd, TABLE_LIST *natural_using_join,
   /* Append the non-equi-join columns of the second join operand. */
   for (it_2.set(table_ref_2); !it_2.end_of_fields(); it_2.next())
   {
-    if (!(nj_col_2= it_2.get_or_create_column_ref(thd, &is_created)))
-      goto err;
-    /*
-      The following assert checks that mark_common_columns() was run and
-      we created the list table_ref_2->join_columns.
-    */
-    DBUG_ASSERT(!is_created);
+    nj_col_2= it_2.get_natural_column_ref();
     if (!nj_col_2->is_common)
       non_join_columns->push_back(nj_col_2);
     else
@@ -4743,8 +4632,7 @@ insert_fields(THD *thd, Name_resolution_context *context, const char *db_name,
             because it was already created and stored with the natural join.
           */
           Natural_join_column *nj_col;
-          if (!(nj_col= field_iterator.get_or_create_column_ref(thd,
-                                                                &is_created)))
+          if (!(nj_col= field_iterator.get_or_create_column_ref(&is_created)))
             DBUG_RETURN(TRUE);
           DBUG_ASSERT(nj_col->table_field && !is_created);
           field_table= nj_col->table_ref->table;
diff --git a/sql/sql_insert.cc b/sql/sql_insert.cc
index 5e9ca203632..8903f28be11 100644
--- a/sql/sql_insert.cc
+++ b/sql/sql_insert.cc
@@ -108,11 +108,7 @@ static int check_insert_fields(THD *thd, TABLE_LIST *table_list,
   {						// Part field list
     SELECT_LEX *select_lex= &thd->lex->select_lex;
     Name_resolution_context *context= &select_lex->context;
-    TABLE_LIST *save_next_local;
-    TABLE_LIST *save_table_list;
-    TABLE_LIST *save_first_name_resolution_table;
-    TABLE_LIST *save_next_name_resolution_table;
-    bool        save_resolve_in_select_list;
+    Name_resolution_context_state ctx_state;
     int res;
 
     if (fields.elements != values.elements)
@@ -125,14 +121,7 @@ static int check_insert_fields(THD *thd, TABLE_LIST *table_list,
     select_lex->no_wrap_view_item= TRUE;
 
     /* Save the state of the current name resolution context. */
-    save_table_list=                  context->table_list;
-    save_first_name_resolution_table= context->first_name_resolution_table;
-    save_next_name_resolution_table=  (context->first_name_resolution_table) ?
-                                      context->first_name_resolution_table->
-                                               next_name_resolution_table :
-                                      NULL;
-    save_resolve_in_select_list=      context->resolve_in_select_list;
-    save_next_local=                  table_list->next_local;
+    ctx_state.save_state(context, table_list);
 
     /*
       Perform name resolution only in the first table - 'table_list',
@@ -143,13 +132,7 @@ static int check_insert_fields(THD *thd, TABLE_LIST *table_list,
     res= setup_fields(thd, 0, fields, 1, 0, 0);
 
     /* Restore the current context. */
-    table_list->next_local=                save_next_local;
-    context->table_list=                   save_table_list;
-    context->first_name_resolution_table=  save_first_name_resolution_table;
-    if (context->first_name_resolution_table)
-      context->first_name_resolution_table->
-               next_name_resolution_table= save_next_name_resolution_table;
-    context->resolve_in_select_list=       save_resolve_in_select_list;
+    ctx_state.restore_state(context, table_list);
     thd->lex->select_lex.no_wrap_view_item= FALSE;
 
     if (res)
@@ -280,13 +263,10 @@ bool mysql_insert(THD *thd,TABLE_LIST *table_list,
   ulonglong id;
   COPY_INFO info;
   TABLE *table= 0;
-  TABLE_LIST *save_table_list;
-  TABLE_LIST *save_next_local;
-  TABLE_LIST *save_first_name_resolution_table;
-  TABLE_LIST *save_next_name_resolution_table;
   List_iterator_fast<List_item> its(values_list);
   List_item *values;
   Name_resolution_context *context;
+  Name_resolution_context_state ctx_state;
 #ifndef EMBEDDED_LIBRARY
   char *query= thd->query;
 #endif
@@ -367,13 +347,7 @@ bool mysql_insert(THD *thd,TABLE_LIST *table_list,
 
   context= &thd->lex->select_lex.context;
   /* Save the state of the current name resolution context. */
-  save_table_list=                  context->table_list;
-  save_first_name_resolution_table= context->first_name_resolution_table;
-  save_next_name_resolution_table=  (context->first_name_resolution_table) ?
-                                    context->first_name_resolution_table->
-                                             next_name_resolution_table :
-                                    NULL;
-  save_next_local=                  table_list->next_local;
+  ctx_state.save_state(context, table_list);
 
   /*
     Perform name resolution only in the first table - 'table_list',
@@ -397,16 +371,11 @@ bool mysql_insert(THD *thd,TABLE_LIST *table_list,
   its.rewind ();
  
   /* Restore the current context. */
-  table_list->next_local= save_next_local;
-  context->first_name_resolution_table= save_first_name_resolution_table;
-  if (context->first_name_resolution_table)
-    context->first_name_resolution_table->
-             next_name_resolution_table= save_next_name_resolution_table;
+  ctx_state.restore_state(context, table_list);
 
   /*
     Fill in the given fields and dump it to the table file
   */
-
   info.records= info.deleted= info.copied= info.updated= 0;
   info.ignore= ignore;
   info.handle_duplicates=duplic;
@@ -814,11 +783,7 @@ bool mysql_prepare_insert(THD *thd, TABLE_LIST *table_list,
 {
   SELECT_LEX *select_lex= &thd->lex->select_lex;
   Name_resolution_context *context= &select_lex->context;
-  TABLE_LIST *save_table_list;
-  TABLE_LIST *save_next_local;
-  TABLE_LIST *save_first_name_resolution_table;
-  TABLE_LIST *save_next_name_resolution_table;
-  bool        save_resolve_in_select_list;
+  Name_resolution_context_state ctx_state;
   bool insert_into_view= (table_list->view != 0);
   bool res= 0;
   DBUG_ENTER("mysql_prepare_insert");
@@ -858,15 +823,7 @@ bool mysql_prepare_insert(THD *thd, TABLE_LIST *table_list,
     DBUG_RETURN(TRUE);
 
   /* Save the state of the current name resolution context. */
-  save_table_list=                  context->table_list;
-  /* Here first_name_resolution_table points to the first select table. */
-  save_first_name_resolution_table= context->first_name_resolution_table;
-  save_next_name_resolution_table=  (context->first_name_resolution_table) ?
-                                    context->first_name_resolution_table->
-                                             next_name_resolution_table :
-                                    NULL;
-  save_resolve_in_select_list=      context->resolve_in_select_list;
-  save_next_local=                  table_list->next_local;
+  ctx_state.save_state(context, table_list);
 
   /*
     Perform name resolution only in the first table - 'table_list',
@@ -891,23 +848,17 @@ bool mysql_prepare_insert(THD *thd, TABLE_LIST *table_list,
     */       
     if (select_lex->group_list.elements == 0)
     {
-      context->table_list->next_local=       save_next_local;
+      context->table_list->next_local=       ctx_state.save_next_local;
       /* first_name_resolution_table was set by resolve_in_table_list_only() */
       context->first_name_resolution_table->
-        next_name_resolution_table=          save_next_local;
+        next_name_resolution_table=          ctx_state.save_next_local;
     }
     if (!res)
       res= setup_fields(thd, 0, update_values, 1, 0, 0);
   }
 
   /* Restore the current context. */
-  table_list->next_local= save_next_local;
-  context->table_list= save_table_list;
-  context->first_name_resolution_table= save_first_name_resolution_table;
-  if (context->first_name_resolution_table)
-    context->first_name_resolution_table->
-             next_name_resolution_table= save_next_name_resolution_table;
-  context->resolve_in_select_list= save_resolve_in_select_list;
+  ctx_state.restore_state(context, table_list);
 
   if (res)
     DBUG_RETURN(res);
@@ -2176,17 +2127,10 @@ select_insert::prepare(List<Item> &values, SELECT_LEX_UNIT *u)
   {
     /* Save the state of the current name resolution context. */
     Name_resolution_context *context= &lex->select_lex.context;
-    TABLE_LIST *save_table_list;
-    TABLE_LIST *save_next_local;
-    TABLE_LIST *save_first_name_resolution_table;
-    TABLE_LIST *save_next_name_resolution_table;
-    save_table_list=                  context->table_list;
-    save_first_name_resolution_table= context->first_name_resolution_table;
-    save_next_name_resolution_table=  (context->first_name_resolution_table) ?
-                                      context->first_name_resolution_table->
-                                               next_name_resolution_table :
-                                      NULL;
-    save_next_local= table_list->next_local;
+    Name_resolution_context_state ctx_state;
+
+    /* Save the state of the current name resolution context. */
+    ctx_state.save_state(context, table_list);
 
     /* Perform name resolution only in the first table - 'table_list'. */
     table_list->next_local= 0;
@@ -2202,20 +2146,15 @@ select_insert::prepare(List<Item> &values, SELECT_LEX_UNIT *u)
     */       
     if (lex->select_lex.group_list.elements == 0)
     {
-      context->table_list->next_local=       save_next_local;
+      context->table_list->next_local=       ctx_state.save_next_local;
       /* first_name_resolution_table was set by resolve_in_table_list_only() */
       context->first_name_resolution_table->
-        next_name_resolution_table=          save_next_local;
+        next_name_resolution_table=          ctx_state.save_next_local;
     }
     res= res || setup_fields(thd, 0, *info.update_values, 1, 0, 0);
 
     /* Restore the current context. */
-    table_list->next_local= save_next_local;
-    context->first_name_resolution_table= save_first_name_resolution_table;
-    if (context->first_name_resolution_table)
-      context->first_name_resolution_table->
-               next_name_resolution_table= save_next_name_resolution_table;
-
+    ctx_state.restore_state(context, table_list);
   }
 
   lex->current_select= lex_current_select_save;
diff --git a/sql/sql_lex.cc b/sql/sql_lex.cc
index ac2d7e82fcf..22ec8241367 100644
--- a/sql/sql_lex.cc
+++ b/sql/sql_lex.cc
@@ -1127,6 +1127,11 @@ void st_select_lex::init_query()
   /*
     Add the name resolution context of the current (sub)query to the
     stack of contexts for the whole query.
+    TODO:
+    push_context may return an error if there is no memory for a new
+    element in the stack, however this method has no return value,
+    thus push_context should be moved to a place where query
+    initialization is checked for failure.
   */
   parent_lex->push_context(&context);
   cond_count= with_wild= 0;
diff --git a/sql/sql_lex.h b/sql/sql_lex.h
index c3198beb4e9..f101b7c811b 100644
--- a/sql/sql_lex.h
+++ b/sql/sql_lex.h
@@ -1006,9 +1006,9 @@ typedef struct st_lex
   }
   void cleanup_after_one_table_open();
 
-  void push_context(Name_resolution_context *context)
+  bool push_context(Name_resolution_context *context)
   {
-    context_stack.push_front(context);
+    return context_stack.push_front(context);
   }
 
   void pop_context()
diff --git a/sql/sql_list.h b/sql/sql_list.h
index 285f1d6e501..b2bcc4ea401 100644
--- a/sql/sql_list.h
+++ b/sql/sql_list.h
@@ -266,10 +266,21 @@ protected:
     ls.elements= elm;
   }
 public:
-  base_list_iterator(base_list &list_par) 
-    :list(&list_par), el(&list_par.first), prev(0), current(0)
+  base_list_iterator() 
+    :list(0), el(0), prev(0), current(0)
   {}
 
+  base_list_iterator(base_list &list_par) 
+  { init(list_par); }
+
+  inline void init(base_list &list_par)
+  {
+    list= &list_par;
+    el= &list_par.first;
+    prev= 0;
+    current= 0;
+  }
+
   inline void *next(void)
   {
     prev=el;
@@ -364,6 +375,8 @@ template <class T> class List_iterator :public base_list_iterator
 {
 public:
   List_iterator(List<T> &a) : base_list_iterator(a) {}
+  List_iterator() : base_list_iterator() {}
+  inline void init(List<T> &a) { base_list_iterator::init(a); }
   inline T* operator++(int) { return (T*) base_list_iterator::next(); }
   inline T *replace(T *a)   { return (T*) base_list_iterator::replace(a); }
   inline T *replace(List<T> &a) { return (T*) base_list_iterator::replace(a); }
@@ -385,6 +398,8 @@ protected:
 
 public:
   inline List_iterator_fast(List<T> &a) : base_list_iterator(a) {}
+  inline List_iterator_fast() : base_list_iterator() {}
+  inline void init(List<T> &a) { base_list_iterator::init(a); }
   inline T* operator++(int) { return (T*) base_list_iterator::next_fast(); }
   inline void rewind(void)  { base_list_iterator::rewind(); }
   void sublist(List<T> &list_arg, uint el_arg)
diff --git a/sql/sql_parse.cc b/sql/sql_parse.cc
index 7430cbffeeb..fcc04ea8ced 100644
--- a/sql/sql_parse.cc
+++ b/sql/sql_parse.cc
@@ -6583,36 +6583,39 @@ void st_select_lex::set_lock_for_tables(thr_lock_type lock_type)
 
 
 /*
-  Create a new name resolution context for a JOIN ... ON clause.
+  Push a new name resolution context for a JOIN ... ON clause to the
+  context stack of a query block.
 
   SYNOPSIS
-    make_join_on_context()
+    push_new_name_resolution_context()
     thd       pointer to current thread
     left_op   left  operand of the JOIN
     right_op  rigth operand of the JOIN
 
   DESCRIPTION
     Create a new name resolution context for a JOIN ... ON clause,
-    and set the first and last leaves of the list of table references
-    to be used for name resolution.
+    set the first and last leaves of the list of table references
+    to be used for name resolution, and push the newly created
+    context to the stack of contexts of the query.
 
   RETURN
-    A new context if all is OK
-    NULL - if a memory allocation error occured
+    FALSE  if all is OK
+    TRUE   if a memory allocation error occured
 */
 
-Name_resolution_context *
-make_join_on_context(THD *thd, TABLE_LIST *left_op, TABLE_LIST *right_op)
+bool
+push_new_name_resolution_context(THD *thd,
+                                 TABLE_LIST *left_op, TABLE_LIST *right_op)
 {
   Name_resolution_context *on_context;
   if (!(on_context= new (thd->mem_root) Name_resolution_context))
-    return NULL;
+    return TRUE;
   on_context->init();
   on_context->first_name_resolution_table=
     left_op->first_leaf_for_name_resolution();
   on_context->last_name_resolution_table=
     right_op->last_leaf_for_name_resolution();
-  return on_context;
+  return thd->lex->push_context(on_context);
 }
 
 
diff --git a/sql/sql_yacc.yy b/sql/sql_yacc.yy
index fbcb3de7907..2c575a8173e 100644
--- a/sql/sql_yacc.yy
+++ b/sql/sql_yacc.yy
@@ -5185,10 +5185,8 @@ join_table:
           {
             YYERROR_UNLESS($1 && ($$=$3));
             /* Change the current name resolution context to a local context. */
-            Name_resolution_context *on_context;
-            if (!(on_context= make_join_on_context(YYTHD,$1,$3)))
+            if (push_new_name_resolution_context(YYTHD, $1, $3))
               YYABORT;
-            Lex->push_context(on_context);
           }
           expr
 	  {
@@ -5200,10 +5198,8 @@ join_table:
           {
             YYERROR_UNLESS($1 && ($$=$3));
             /* Change the current name resolution context to a local context. */
-            Name_resolution_context *on_context;
-            if (!(on_context= make_join_on_context(YYTHD,$1,$3)))
+            if (push_new_name_resolution_context(YYTHD, $1, $3))
               YYABORT;
-            Lex->push_context(on_context);
           }
           expr
           {
@@ -5230,10 +5226,8 @@ join_table:
           ON
           {
             /* Change the current name resolution context to a local context. */
-            Name_resolution_context *on_context;
-            if (!(on_context= make_join_on_context(YYTHD,$1,$5)))
+            if (push_new_name_resolution_context(YYTHD, $1, $5))
               YYABORT;
-            Lex->push_context(on_context);
           }
           expr
 	  {
@@ -5263,10 +5257,8 @@ join_table:
           ON
           {
             /* Change the current name resolution context to a local context. */
-            Name_resolution_context *on_context;
-            if (!(on_context= make_join_on_context(YYTHD,$1,$5)))
+            if (push_new_name_resolution_context(YYTHD, $1, $5))
               YYABORT;
-            Lex->push_context(on_context);
           }
           expr
           {
@@ -5327,10 +5319,9 @@ table_factor:
           ON
           {
             /* Change the current name resolution context to a local context. */
-            Name_resolution_context *on_context;
-            if (!(on_context= make_join_on_context(YYTHD,$3,$7)))
+            if (push_new_name_resolution_context(YYTHD, $3, $7))
               YYABORT;
-            Lex->push_context(on_context);
+
           }
           expr '}'
 	  {
diff --git a/sql/table.cc b/sql/table.cc
index 07ca236854c..268d7a0be49 100644
--- a/sql/table.cc
+++ b/sql/table.cc
@@ -2606,60 +2606,6 @@ GRANT_INFO *Natural_join_column::grant()
 }
 
 
-
-#ifndef NO_EMBEDDED_ACCESS_CHECKS
-
-/*
-  Check the access rights for the current join column.
-  columns.
-
-  SYNOPSIS
-    Natural_join_column::check_grants()
-
-  DESCRIPTION
-    Check the access rights to a column from a natural join in a generic
-    way that hides the heterogeneity of the column representation - whether
-    it is a view or a stored table colum.
-
-  RETURN
-    FALSE  The column can be accessed
-    TRUE   There are no access rights to all equivalent columns
-*/
-
-bool
-Natural_join_column::check_grants(THD *thd, const char *name, uint length)
-{
-  GRANT_INFO *grant;
-  const char *db_name;
-  const char *table_name;
-  Security_context *save_security_ctx= thd->security_ctx;
-  Security_context *new_sctx= table_ref->security_ctx;
-  bool res;
-
-  if (view_field)
-  {
-    DBUG_ASSERT(table_field == NULL);
-    grant= &(table_ref->grant);
-    db_name= table_ref->view_db.str;
-    table_name= table_ref->view_name.str;
-  }
-  else
-  {
-    DBUG_ASSERT(table_field && view_field == NULL);
-    grant= &(table_ref->table->grant);
-    db_name= table_ref->table->s->db;
-    table_name= table_ref->table->s->table_name;
-  }
-
-  if (new_sctx)
-    thd->security_ctx= new_sctx;
-  res= check_grant_column(thd, grant, db_name, table_name, name, length);
-  thd->security_ctx= save_security_ctx;
-  return res;
-}
-#endif
-
-
 void Field_iterator_view::set(TABLE_LIST *table)
 {
   DBUG_ASSERT(table->field_translation);
@@ -2702,8 +2648,9 @@ Item *create_view_field(THD *thd, TABLE_LIST *view, Item **field_ref,
   if (view->schema_table_reformed)
   {
     /*
-      In case of SHOW command (schema_table_reformed set) all items are
-      fixed
+      Translation table items are always Item_fields and already fixed
+      ('mysql_schema_table' function). So we can return directly the
+      field. This case happens only for 'show & where' commands.
     */
     DBUG_ASSERT(field && field->fixed);
     DBUG_RETURN(field);
@@ -2735,21 +2682,14 @@ Item *create_view_field(THD *thd, TABLE_LIST *view, Item **field_ref,
 void Field_iterator_natural_join::set(TABLE_LIST *table_ref)
 {
   DBUG_ASSERT(table_ref->join_columns);
-  delete column_ref_it;
-
-  /*
-    TODO: try not to allocate new iterator every time. If we have to,
-    then check for out of memory condition.
-  */
-  column_ref_it= new List_iterator_fast<Natural_join_column>
-                     (*(table_ref->join_columns));
-  cur_column_ref= (*column_ref_it)++;
+  column_ref_it.init(*(table_ref->join_columns));
+  cur_column_ref= column_ref_it++;
 }
 
 
 void Field_iterator_natural_join::next()
 {
-  cur_column_ref= (*column_ref_it)++;
+  cur_column_ref= column_ref_it++;
   DBUG_ASSERT(!cur_column_ref || ! cur_column_ref->table_field ||
               cur_column_ref->table_ref->table ==
               cur_column_ref->table_field->table);
@@ -2876,7 +2816,6 @@ GRANT_INFO *Field_iterator_table_ref::grant()
 
   SYNOPSIS
     Field_iterator_table_ref::get_or_create_column_ref()
-    thd         [in]  pointer to current thread
     is_created  [out] set to TRUE if the column was created,
                       FALSE if we return an already created colum
 
@@ -2889,7 +2828,7 @@ GRANT_INFO *Field_iterator_table_ref::grant()
 */
 
 Natural_join_column *
-Field_iterator_table_ref::get_or_create_column_ref(THD *thd, bool *is_created)
+Field_iterator_table_ref::get_or_create_column_ref(bool *is_created)
 {
   Natural_join_column *nj_col;
 
@@ -2923,6 +2862,41 @@ Field_iterator_table_ref::get_or_create_column_ref(THD *thd, bool *is_created)
 }
 
 
+/*
+  Return an existing reference to a column of a natural/using join.
+
+  SYNOPSIS
+    Field_iterator_table_ref::get_natural_column_ref()
+
+  DESCRIPTION
+    The method should be called in contexts where it is expected that
+    all natural join columns are already created, and that the column
+    being retrieved is a Natural_join_column.
+
+  RETURN
+    #     Pointer to a column of a natural join (or its operand)
+    NULL  No memory to allocate the column
+*/
+
+Natural_join_column *
+Field_iterator_table_ref::get_natural_column_ref()
+{
+  Natural_join_column *nj_col;
+
+  DBUG_ASSERT(field_it == &natural_join_it);
+  /*
+    The field belongs to a NATURAL join, therefore the column reference was
+    already created via one of the two constructor calls above. In this case
+    we just return the already created column reference.
+  */
+  nj_col= natural_join_it.column_ref();
+  DBUG_ASSERT(nj_col &&
+              (!nj_col->table_field ||
+               nj_col->table_ref->table == nj_col->table_field->table));
+  return nj_col;
+}
+
+
 /*****************************************************************************
 ** Instansiate templates
 *****************************************************************************/
diff --git a/sql/table.h b/sql/table.h
index d919b9a4203..ce0616a6833 100644
--- a/sql/table.h
+++ b/sql/table.h
@@ -407,9 +407,6 @@ public:
   const char *table_name();
   const char *db_name();
   GRANT_INFO *grant();
-#ifndef NO_EMBEDDED_ACCESS_CHECKS
-  bool check_grants(THD *thd, const char *name, uint length);
-#endif
 };
 
 
@@ -734,11 +731,11 @@ public:
 
 class Field_iterator_natural_join: public Field_iterator
 {
-  List_iterator_fast<Natural_join_column> *column_ref_it;
+  List_iterator_fast<Natural_join_column> column_ref_it;
   Natural_join_column *cur_column_ref;
 public:
-  Field_iterator_natural_join() :column_ref_it(NULL), cur_column_ref(NULL) {}
-  ~Field_iterator_natural_join() { delete column_ref_it; }
+  Field_iterator_natural_join() :cur_column_ref(NULL) {}
+  ~Field_iterator_natural_join() {}
   void set(TABLE_LIST *table);
   void next();
   bool end_of_fields() { return !cur_column_ref; }
@@ -785,7 +782,8 @@ public:
   GRANT_INFO *grant();
   Item *create_item(THD *thd) { return field_it->create_item(thd); }
   Field *field() { return field_it->field(); }
-  Natural_join_column *get_or_create_column_ref(THD *thd, bool *is_created);
+  Natural_join_column *get_or_create_column_ref(bool *is_created);
+  Natural_join_column *get_natural_column_ref();
 };