25 files changed, 624 insertions, 42 deletions

diff --git a/mysql-test/r/lock_multi.result b/mysql-test/r/lock_multi.result
index cd05fc1473f..a0e3d9fad06 100644
--- a/mysql-test/r/lock_multi.result
+++ b/mysql-test/r/lock_multi.result
@@ -96,6 +96,40 @@ alter table t1 auto_increment=0;
 alter table t1 auto_increment=0;
 unlock tables;
 drop table t1;
+CREATE TABLE t1 (
+a int(11) unsigned default NULL,
+b varchar(255) default NULL,
+UNIQUE KEY a (a),
+KEY b (b)
+);
+INSERT INTO t1 VALUES (1, 1), (2, 2), (3, 3);
+CREATE TABLE t2 SELECT * FROM t1;
+CREATE TABLE t3 SELECT * FROM t1;
+# test altering of columns that multiupdate doesn't use
+# normal mode
+# PS mode
+# test altering of columns that multiupdate uses
+# normal mode
+# PS mode
+DROP TABLE t1, t2, t3;
+CREATE TABLE t1( a INT, b INT );
+INSERT INTO t1 VALUES (1, 1), (2, 2), (3, 3), (4, 4);
+# 1. test regular tables
+# 1.1. test altering of columns that multiupdate doesn't use
+# 1.1.1. normal mode
+# 1.1.2. PS mode
+# 1.2. test altering of columns that multiupdate uses
+# 1.2.1. normal mode
+# 1.2.2. PS mode
+ALTER TABLE t1 ADD COLUMN a INT;
+# 2. test UNIONs
+# 2.1. test altering of columns that multiupdate doesn't use
+# 2.1.1. normal mode
+# 2.1.2. PS mode
+# 2.2. test altering of columns that multiupdate uses
+# 2.2.1. normal mode
+# 2.2.2. PS mode
+DROP TABLE t1;
 End of 5.0 tests
 create table t1 (i int);
 lock table t1 read;
diff --git a/mysql-test/suite/rpl/r/rpl_row_create_table.result b/mysql-test/suite/rpl/r/rpl_row_create_table.result
index c4cf8353bca..ad659c37b7f 100644
--- a/mysql-test/suite/rpl/r/rpl_row_create_table.result
+++ b/mysql-test/suite/rpl/r/rpl_row_create_table.result
@@ -430,4 +430,25 @@ a
 1
 2
 DROP TABLE t1;
+stop slave;
+drop table if exists t1,t2,t3,t4,t5,t6,t7,t8,t9;
+reset master;
+reset slave;
+drop table if exists t1,t2,t3,t4,t5,t6,t7,t8,t9;
+start slave;
+CREATE DATABASE mysqltest1;
+CREATE TABLE mysqltest1.without_select (f1 BIGINT);
+CREATE TABLE mysqltest1.with_select AS SELECT 1 AS f1;
+show binlog events from <binlog_start>;
+Log_name	Pos	Event_type	Server_id	End_log_pos	Info
+master-bin.000001	#	Query	#	#	CREATE DATABASE mysqltest1
+master-bin.000001	#	Query	#	#	use `test`; CREATE TABLE mysqltest1.without_select (f1 BIGINT)
+master-bin.000001	#	Query	#	#	use `test`; BEGIN
+master-bin.000001	#	Query	#	#	use `test`; CREATE TABLE `mysqltest1`.`with_select` (
+  `f1` int(1) NOT NULL DEFAULT '0'
+)
+master-bin.000001	#	Table_map	#	#	table_id: # (mysqltest1.with_select)
+master-bin.000001	#	Write_rows	#	#	table_id: # flags: STMT_END_F
+master-bin.000001	#	Query	#	#	use `test`; COMMIT
+DROP DATABASE mysqltest1;
 end of the tests
diff --git a/mysql-test/suite/rpl/t/rpl_row_create_table.test b/mysql-test/suite/rpl/t/rpl_row_create_table.test
index e5cdfa4341a..3fb5aa8e1f2 100644
--- a/mysql-test/suite/rpl/t/rpl_row_create_table.test
+++ b/mysql-test/suite/rpl/t/rpl_row_create_table.test
@@ -259,5 +259,22 @@ connection master;
 DROP TABLE t1;
 sync_slave_with_master;
 
+#
+# BUG#34707: Row based replication: slave creates table within wrong database
+#
+
+source include/master-slave-reset.inc;
+
+connection master;
+CREATE DATABASE mysqltest1;
+
+CREATE TABLE mysqltest1.without_select (f1 BIGINT);
+CREATE TABLE mysqltest1.with_select AS SELECT 1 AS f1;
+source include/show_binlog_events.inc;
+sync_slave_with_master;
+
+connection master;
+DROP DATABASE mysqltest1;
+sync_slave_with_master;
 
 --echo end of the tests
diff --git a/mysql-test/t/disabled.def b/mysql-test/t/disabled.def
index ada7cb3aa97..689f2fe8064 100644
--- a/mysql-test/t/disabled.def
+++ b/mysql-test/t/disabled.def
@@ -14,4 +14,4 @@ csv_alter_table      : Bug#33696 2008-01-21 pcrews no .result file - bug allows
 thread_cache_size_func : Bug#36733 main.thread_cache_size_func fails randomly
 log_tables.test                       : Bug #37798: main.log_tables fails randomly on powermacg5 and windows
 slow_query_log_func.test              : Bug #37962: *_func tests containing sleeps/race conditions
-
+events-bugs.test                      : Bug #39848, Bug #39863, Bug #39569, Bug #37774
diff --git a/mysql-test/t/lock_multi.test b/mysql-test/t/lock_multi.test
index 54c2e4e76ee..08747b0f92e 100644
--- a/mysql-test/t/lock_multi.test
+++ b/mysql-test/t/lock_multi.test
@@ -291,7 +291,328 @@ connection reader;
 reap;
 connection locker;
 drop table t1;
+
+#
+# Bug #38691: segfault/abort in ``UPDATE ...JOIN'' while
+#             ``FLUSH TABLES WITH READ LOCK''
+#
+
+--connection default
+CREATE TABLE t1 (
+  a int(11) unsigned default NULL,
+  b varchar(255) default NULL,
+  UNIQUE KEY a (a),
+  KEY b (b)
+);
+
+INSERT INTO t1 VALUES (1, 1), (2, 2), (3, 3);
+CREATE TABLE t2 SELECT * FROM t1;
+CREATE TABLE t3 SELECT * FROM t1;
+
+--echo # test altering of columns that multiupdate doesn't use
+
+--echo # normal mode
+
+--disable_query_log
+let $i = 100;
+while ($i) {
+--dec $i
+
+--connection writer
+  send UPDATE t2 INNER JOIN (t1 JOIN t3 USING(a)) USING(a)
+         SET a = NULL WHERE t1.b <> t2.b;
+
+--connection locker
+  ALTER TABLE t2 ADD COLUMN (c INT);
+  ALTER TABLE t2 DROP COLUMN c;
+
+--connection writer
+--reap
+}
+
+--echo # PS mode
+
+--connection writer
+PREPARE stmt FROM 'UPDATE t2 INNER JOIN (t1 JOIN t3 USING(a)) USING(a)
+                     SET a = NULL WHERE t1.b <> t2.b';
+
+let $i = 100;
+while ($i) {
+--dec $i
+
+--connection writer
+--send EXECUTE stmt
+
+--connection locker
+  ALTER TABLE t2 ADD COLUMN (c INT);
+  ALTER TABLE t2 DROP COLUMN c;
+
+--connection writer
+--reap
+}
+--enable_query_log
+
+
+--echo # test altering of columns that multiupdate uses
+
+--echo # normal mode
+
+--connection default
+
+--disable_query_log
+let $i = 100;
+while ($i) {
+  dec $i;
+
+--connection locker
+--error 0,1060
+  ALTER TABLE t2 ADD COLUMN a int(11) unsigned default NULL;
+  UPDATE t2 SET a=b;
+
+--connection writer
+--send UPDATE t2 INNER JOIN (t1 JOIN t3 USING(a)) USING(a) SET a = NULL WHERE t1.b <> t2.b
+
+--connection locker
+--error 0,1091
+  ALTER TABLE t2 DROP COLUMN a;
+
+--connection writer
+--error 0,1054
+--reap
+}
+--enable_query_log
+
+--echo # PS mode
+
+--disable_query_log
+let $i = 100;
+while ($i) {
+  dec $i;
+
+--connection locker
+--error 0,1060
+  ALTER TABLE t2 ADD COLUMN a int(11) unsigned default NULL;
+  UPDATE t2 SET a=b;
+
+--connection writer
+  PREPARE stmt FROM 'UPDATE t2 INNER JOIN (t1 JOIN t3 USING(a)) USING(a) SET a = NULL WHERE t1.b <> t2.b';
+--send EXECUTE stmt
+
+--connection locker
+--error 0,1091
+  ALTER TABLE t2 DROP COLUMN a;
+
+--connection writer
+--error 0,1054
+--reap
+
+}
+--enable_query_log
+--connection default
+DROP TABLE t1, t2, t3;
+
 #
+# Bug#38499: flush tables and multitable table update with derived table cause 
+#            crash
+#
+
+CREATE TABLE t1( a INT, b INT );
+INSERT INTO t1 VALUES (1, 1), (2, 2), (3, 3), (4, 4);
+
+--echo # 1. test regular tables
+--echo # 1.1. test altering of columns that multiupdate doesn't use
+--echo # 1.1.1. normal mode
+
+--disable_query_log
+let $i = 100;
+while ($i) {
+--dec $i
+
+--connection writer
+  send UPDATE t1, (SELECT 1 FROM t1 t1i) d SET a = 0 WHERE 1=0;
+
+--connection locker
+  ALTER TABLE t1 ADD COLUMN (c INT);
+  ALTER TABLE t1 DROP COLUMN c;
+
+--connection writer
+--reap
+}
+
+--echo # 1.1.2. PS mode
+
+--connection writer
+PREPARE stmt FROM 'UPDATE t1, (SELECT 1 FROM t1 t1i) d SET a = 0 WHERE 1=0';
+
+let $i = 100;
+while ($i) {
+--dec $i
+
+--connection writer
+--send EXECUTE stmt
+
+--connection locker
+  ALTER TABLE t1 ADD COLUMN (c INT);
+  ALTER TABLE t1 DROP COLUMN c;
+
+--connection writer
+--reap
+}
+--enable_query_log
+
+--echo # 1.2. test altering of columns that multiupdate uses
+--echo # 1.2.1. normal mode
+
+--connection default
+
+--disable_query_log
+let $i = 100;
+while ($i) {
+  dec $i;
+
+--connection locker
+--error 0,1060
+  ALTER TABLE t1 ADD COLUMN a int(11) unsigned default NULL;
+  UPDATE t1 SET a=b;
+
+--connection writer
+--send UPDATE t1, (SELECT 1 FROM t1 t1i) d SET a = 0 WHERE 1=0;
+
+--connection locker
+--error 0,1091
+  ALTER TABLE t1 DROP COLUMN a;
+
+--connection writer
+--error 0,1054 # unknown column error
+--reap
+}
+--enable_query_log
+
+--echo # 1.2.2. PS mode
+
+--disable_query_log
+let $i = 100;
+while ($i) {
+  dec $i;
+
+--connection locker
+--error 0,1060
+  ALTER TABLE t1 ADD COLUMN a INT;
+  UPDATE t1 SET a=b;
+
+--connection writer
+  PREPARE stmt FROM 'UPDATE t1, (SELECT 1 FROM t1 t1i) d SET a = 0 WHERE 1=0';
+--send EXECUTE stmt
+
+--connection locker
+--error 0,1091
+  ALTER TABLE t1 DROP COLUMN a;
+
+--connection writer
+--error 0,1054 # Unknown column 'a' in 'field list'
+--reap
+}
+--enable_query_log
+--connection default
+ALTER TABLE t1 ADD COLUMN a INT;
+
+--echo # 2. test UNIONs
+--echo # 2.1. test altering of columns that multiupdate doesn't use
+--echo # 2.1.1. normal mode
+
+--disable_query_log
+let $i = 100;
+while ($i) {
+--dec $i
+
+--connection writer
+  send UPDATE t1, ((SELECT 1 FROM t1 t1i) UNION (SELECT 2 FROM t1 t1ii)) e SET a = 0 WHERE 1=0;
+
+--connection locker
+  ALTER TABLE t1 ADD COLUMN (c INT);
+  ALTER TABLE t1 DROP COLUMN c;
+
+--connection writer
+--reap
+}
+
+--echo # 2.1.2. PS mode
+
+--connection writer
+PREPARE stmt FROM 'UPDATE t1, ((SELECT 1 FROM t1 t1i) UNION (SELECT 2 FROM t1 t1ii)) e SET a = 0 WHERE 1=0';
+
+let $i = 100;
+while ($i) {
+--dec $i
+
+--connection writer
+--send EXECUTE stmt
+
+--connection locker
+  ALTER TABLE t1 ADD COLUMN (c INT);
+  ALTER TABLE t1 DROP COLUMN c;
+
+--connection writer
+--reap
+}
+--enable_query_log
+
+--echo # 2.2. test altering of columns that multiupdate uses
+--echo # 2.2.1. normal mode
+
+--connection default
+
+--disable_query_log
+let $i = 100;
+while ($i) {
+  dec $i;
+
+--connection locker
+--error 0,1060
+  ALTER TABLE t1 ADD COLUMN a int(11) unsigned default NULL;
+  UPDATE t1 SET a=b;
+
+--connection writer
+--send UPDATE t1, ((SELECT 1 FROM t1 t1i) UNION (SELECT 2 FROM t1 t1ii)) e SET a = 0 WHERE 1=0;
+
+--connection locker
+--error 0,1091
+  ALTER TABLE t1 DROP COLUMN a;
+
+--connection writer
+--error 0,1054 # Unknown column 'a' in 'field list'
+--reap
+}
+--enable_query_log
+
+--echo # 2.2.2. PS mode
+
+--disable_query_log
+let $i = 100;
+while ($i) {
+  dec $i;
+
+--connection locker
+--error 0,1060
+  ALTER TABLE t1 ADD COLUMN a INT;
+  UPDATE t1 SET a=b;
+
+--connection writer
+  PREPARE stmt FROM 'UPDATE t1, ((SELECT 1 FROM t1 t1i) UNION (SELECT 2 FROM t1 t1ii)) e SET a = 0 WHERE 1=0';
+--send EXECUTE stmt
+
+--connection locker
+--error 0,1091
+  ALTER TABLE t1 DROP COLUMN a;
+
+--connection writer
+--error 0,1054 # Unknown column 'a' in 'field list'
+--reap
+}
+--enable_query_log
+--connection default
+DROP TABLE t1;
+
 --echo End of 5.0 tests
 
 
diff --git a/sql/handler.cc b/sql/handler.cc
index 981984c812d..a988c34b7ca 100644
--- a/sql/handler.cc
+++ b/sql/handler.cc
@@ -2752,8 +2752,53 @@ bool handler::get_error_message(int error, String* buf)
 }
 
 
+/**
+  Check for incompatible collation changes.
+   
+  @retval
+    HA_ADMIN_NEEDS_UPGRADE   Table may have data requiring upgrade.
+  @retval
+    0                        No upgrade required.
+*/
+
+int handler::check_collation_compatibility()
+{
+  ulong mysql_version= table->s->mysql_version;
+
+  if (mysql_version < 50048)
+  {
+    KEY *key= table->key_info;
+    KEY *key_end= key + table->s->keys;
+    for (; key < key_end; key++)
+    {
+      KEY_PART_INFO *key_part= key->key_part;
+      KEY_PART_INFO *key_part_end= key_part + key->key_parts;
+      for (; key_part < key_part_end; key_part++)
+      {
+        if (!key_part->fieldnr)
+          continue;
+        Field *field= table->field[key_part->fieldnr - 1];
+        uint cs_number= field->charset()->number;
+        if (mysql_version < 50048 &&
+            (cs_number == 11 || /* ascii_general_ci - bug #29499, bug #27562 */
+             cs_number == 41 || /* latin7_general_ci - bug #29461 */
+             cs_number == 42 || /* latin7_general_cs - bug #29461 */
+             cs_number == 20 || /* latin7_estonian_cs - bug #29461 */
+             cs_number == 21 || /* latin2_hungarian_ci - bug #29461 */
+             cs_number == 22 || /* koi8u_general_ci - bug #29461 */
+             cs_number == 23 || /* cp1251_ukrainian_ci - bug #29461 */
+             cs_number == 26))  /* cp1250_general_ci - bug #29461 */
+          return HA_ADMIN_NEEDS_UPGRADE;
+      }  
+    }  
+  }  
+  return 0;
+}
+
+
 int handler::ha_check_for_upgrade(HA_CHECK_OPT *check_opt)
 {
+  int error;
   KEY *keyinfo, *keyend;
   KEY_PART_INFO *keypart, *keypartend;
 
@@ -2782,6 +2827,10 @@ int handler::ha_check_for_upgrade(HA_CHECK_OPT *check_opt)
   }
   if (table->s->frm_version != FRM_VER_TRUE_VARCHAR)
     return HA_ADMIN_NEEDS_ALTER;
+
+  if ((error= check_collation_compatibility()))
+    return error;
+    
   return check_for_upgrade(check_opt);
 }
 
@@ -4375,6 +4424,8 @@ static int write_locked_table_maps(THD *thd)
                        (long) thd, (long) thd->lock,
                        (long) thd->locked_tables, (long) thd->extra_lock));
 
+  DBUG_PRINT("debug", ("get_binlog_table_maps(): %d", thd->get_binlog_table_maps()));
+
   if (thd->get_binlog_table_maps() == 0)
   {
     MYSQL_LOCK *locks[3];
diff --git a/sql/handler.h b/sql/handler.h
index b943e188962..b7d4d689d40 100644
--- a/sql/handler.h
+++ b/sql/handler.h
@@ -1211,6 +1211,7 @@ public:
   int ha_delete_row(const uchar * buf);
   void ha_release_auto_increment();
 
+  int check_collation_compatibility();
   int ha_check_for_upgrade(HA_CHECK_OPT *check_opt);
   /** to be actually called to get 'check()' functionality*/
   int ha_check(THD *thd, HA_CHECK_OPT *check_opt);
@@ -1732,6 +1733,12 @@ public:
     but we don't have a primary key
   */
   virtual void use_hidden_primary_key();
+  virtual uint alter_table_flags(uint flags)
+  {
+    if (ht->alter_table_flags)
+      return ht->alter_table_flags(flags);
+    return 0;
+  } 
 
 protected:
   /* Service methods for use by storage engines. */
diff --git a/sql/item.cc b/sql/item.cc
index 8a2871e723c..31f5bc06a9f 100644
--- a/sql/item.cc
+++ b/sql/item.cc
@@ -1804,14 +1804,16 @@ Item_field::Item_field(THD *thd, Name_resolution_context *context_arg,
     We need to copy db_name, table_name and field_name because they must
     be allocated in the statement memory, not in table memory (the table
     structure can go away and pop up again between subsequent executions
-    of a prepared statement).
+    of a prepared statement or after the close_tables_for_reopen() call
+    in mysql_multi_update_prepare()).
   */
-  if (thd->stmt_arena->is_stmt_prepare_or_first_sp_execute())
   {
     if (db_name)
       orig_db_name= thd->strdup(db_name);
-    orig_table_name= thd->strdup(table_name);
-    orig_field_name= thd->strdup(field_name);
+    if (table_name)
+      orig_table_name= thd->strdup(table_name);
+    if (field_name)
+      orig_field_name= thd->strdup(field_name);
     /*
       We don't restore 'name' in cleanup because it's not changed
       during execution. Still we need it to point to persistent
diff --git a/sql/log.cc b/sql/log.cc
index b5539525ea6..8ab2e892b27 100644
--- a/sql/log.cc
+++ b/sql/log.cc
@@ -1377,6 +1377,8 @@ binlog_end_trans(THD *thd, binlog_trx_data *trx_data,
                       FLAGSTR(thd->options, OPTION_NOT_AUTOCOMMIT),
                       FLAGSTR(thd->options, OPTION_BEGIN)));
 
+  thd->binlog_flush_pending_rows_event(TRUE);
+
   /*
     NULL denotes ROLLBACK with nothing to replicate: i.e., rollback of
     only transactional tables.  If the transaction contain changes to
@@ -1395,8 +1397,6 @@ binlog_end_trans(THD *thd, binlog_trx_data *trx_data,
       were, we would have to ensure that we're not ending a statement
       inside a stored function.
      */
-    thd->binlog_flush_pending_rows_event(TRUE);
-
     error= mysql_bin_log.write(thd, &trx_data->trans_log, end_ev);
     trx_data->reset();
 
diff --git a/sql/mysqld.cc b/sql/mysqld.cc
index 76deffec789..abb778394e6 100644
--- a/sql/mysqld.cc
+++ b/sql/mysqld.cc
@@ -7675,8 +7675,6 @@ mysqld_get_one_option(int optid,
     break;
   case 'l':
     WARN_DEPRECATED(NULL, "7.0", "--log", "'--general_log'/'--general_log_file'");
-    /* FALL-THROUGH */
-  case OPT_GENERAL_LOG_FILE:
     opt_log=1;
     break;
   case 'h':
@@ -7847,8 +7845,6 @@ mysqld_get_one_option(int optid,
 #endif /* HAVE_REPLICATION */
   case (int) OPT_SLOW_QUERY_LOG:
     WARN_DEPRECATED(NULL, "7.0", "--log_slow_queries", "'--slow_query_log'/'--slow_query_log_file'");
-    /* FALL-THROUGH */
-  case (int) OPT_SLOW_QUERY_LOG_FILE:
     opt_slow_log= 1;
     break;
 #ifdef WITH_CSV_STORAGE_ENGINE
diff --git a/sql/sp_head.cc b/sql/sp_head.cc
index a0eda42f0c5..534cd0a7ca1 100644
--- a/sql/sp_head.cc
+++ b/sql/sp_head.cc
@@ -1949,8 +1949,8 @@ sp_head::execute_procedure(THD *thd, List<Item> *args)
       thd->rollback_item_tree_changes();
     }
 
-    DBUG_PRINT("info",(" %.*s: eval args done",
-                       (int) m_name.length, m_name.str));
+    DBUG_PRINT("info",(" %.*s: eval args done", (int) m_name.length, 
+                       m_name.str));
   }
   if (!(m_flags & LOG_SLOW_STATEMENTS) && thd->enable_slow_log)
   {
diff --git a/sql/sql_base.cc b/sql/sql_base.cc
index c97a4b0fbda..658acadf226 100644
--- a/sql/sql_base.cc
+++ b/sql/sql_base.cc
@@ -5759,8 +5759,21 @@ find_field_in_natural_join(THD *thd, TABLE_LIST *table_ref, const char *name,
   {
     /* This is a base table. */
     DBUG_ASSERT(nj_col->view_field == NULL);
-    DBUG_ASSERT(nj_col->table_ref->table == nj_col->table_field->table);
-    found_field= nj_col->table_field;
+    /*
+      This fix_fields is not necessary (initially this item is fixed by
+      the Item_field constructor; after reopen_tables the Item_func_eq
+      calls fix_fields on that item), it's just a check during table
+      reopening for columns that was dropped by the concurrent connection.
+    */
+    if (!nj_col->table_field->fixed &&
+        nj_col->table_field->fix_fields(thd, (Item **)&nj_col->table_field))
+    {
+      DBUG_PRINT("info", ("column '%s' was dropped by the concurrent connection",
+                          nj_col->table_field->name));
+      DBUG_RETURN(NULL);
+    }
+    DBUG_ASSERT(nj_col->table_ref->table == nj_col->table_field->field->table);
+    found_field= nj_col->table_field->field;
     update_field_dependencies(thd, found_field, nj_col->table_ref->table);
   }
 
@@ -6685,7 +6698,7 @@ mark_common_columns(THD *thd, TABLE_LIST *table_ref_1, TABLE_LIST *table_ref_2,
     const char *field_name_1;
     /* true if field_name_1 is a member of using_fields */
     bool is_using_column_1;
-    if (!(nj_col_1= it_1.get_or_create_column_ref(leaf_1)))
+    if (!(nj_col_1= it_1.get_or_create_column_ref(thd, leaf_1)))
       goto err;
     field_name_1= nj_col_1->name();
     is_using_column_1= using_fields && 
@@ -6706,7 +6719,7 @@ mark_common_columns(THD *thd, TABLE_LIST *table_ref_1, TABLE_LIST *table_ref_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(leaf_2)))
+      if (!(cur_nj_col_2= it_2.get_or_create_column_ref(thd, leaf_2)))
         goto err;
       cur_field_name_2= cur_nj_col_2->name();
       DBUG_PRINT ("info", ("cur_field_name_2=%s.%s", 
@@ -7196,15 +7209,24 @@ static bool setup_natural_join_row_types(THD *thd,
   TABLE_LIST *left_neighbor;
   /* Table reference to the right of the current. */
   TABLE_LIST *right_neighbor= NULL;
+  bool save_first_natural_join_processing=
+    context->select_lex->first_natural_join_processing;
+
+  context->select_lex->first_natural_join_processing= FALSE;
 
   /* Note that tables in the list are in reversed order */
   for (left_neighbor= table_ref_it++; left_neighbor ; )
   {
     table_ref= left_neighbor;
     left_neighbor= table_ref_it++;
-    /* For stored procedures do not redo work if already done. */
-    if (context->select_lex->first_execution)
+    /* 
+      Do not redo work if already done:
+      1) for stored procedures,
+      2) for multitable update after lock failure and table reopening.
+    */
+    if (save_first_natural_join_processing)
     {
+      context->select_lex->first_natural_join_processing= FALSE;
       if (store_top_level_join_columns(thd, table_ref,
                                        left_neighbor, right_neighbor))
         return TRUE;
diff --git a/sql/sql_class.cc b/sql/sql_class.cc
index c4a4312ad85..abbb2de0a82 100644
--- a/sql/sql_class.cc
+++ b/sql/sql_class.cc
@@ -3558,6 +3558,29 @@ int THD::binlog_flush_pending_rows_event(bool stmt_end)
 }
 
 
+#if !defined(DBUG_OFF) && !defined(_lint)
+static const char *
+show_query_type(THD::enum_binlog_query_type qtype)
+{
+  switch (qtype) {
+  case THD::ROW_QUERY_TYPE:
+    return "ROW";
+  case THD::STMT_QUERY_TYPE:
+    return "STMT";
+  case THD::MYSQL_QUERY_TYPE:
+    return "MYSQL";
+  case THD::QUERY_TYPE_COUNT:
+  default:
+    DBUG_ASSERT(0 <= qtype && qtype < THD::QUERY_TYPE_COUNT);
+  }
+
+  static char buf[64];
+  sprintf(buf, "UNKNOWN#%d", qtype);
+  return buf;
+}
+#endif
+
+
 /*
   Member function that will log query, either row-based or
   statement-based depending on the value of the 'current_stmt_binlog_row_based'
@@ -3586,7 +3609,8 @@ int THD::binlog_query(THD::enum_binlog_query_type qtype, char const *query_arg,
                       THD::killed_state killed_status_arg)
 {
   DBUG_ENTER("THD::binlog_query");
-  DBUG_PRINT("enter", ("qtype: %d  query: '%s'", qtype, query_arg));
+  DBUG_PRINT("enter", ("qtype: %s  query: '%s'",
+                       show_query_type(qtype), query_arg));
   DBUG_ASSERT(query_arg && mysql_bin_log.is_open());
 
   /*
@@ -3625,6 +3649,9 @@ int THD::binlog_query(THD::enum_binlog_query_type qtype, char const *query_arg,
 
   switch (qtype) {
   case THD::ROW_QUERY_TYPE:
+    DBUG_PRINT("debug",
+               ("current_stmt_binlog_row_based: %d",
+                current_stmt_binlog_row_based));
     if (current_stmt_binlog_row_based)
       DBUG_RETURN(0);
     /* Otherwise, we fall through */
diff --git a/sql/sql_class.h b/sql/sql_class.h
index 82c162f923f..7696882bbcc 100644
--- a/sql/sql_class.h
+++ b/sql/sql_class.h
@@ -996,6 +996,22 @@ enum enum_thread_type
   SYSTEM_THREAD_EVENT_WORKER= 32
 };
 
+inline char const *
+show_system_thread(enum_thread_type thread)
+{
+#define RETURN_NAME_AS_STRING(NAME) case (NAME): return #NAME
+  switch (thread) {
+    RETURN_NAME_AS_STRING(NON_SYSTEM_THREAD);
+    RETURN_NAME_AS_STRING(SYSTEM_THREAD_DELAYED_INSERT);
+    RETURN_NAME_AS_STRING(SYSTEM_THREAD_SLAVE_IO);
+    RETURN_NAME_AS_STRING(SYSTEM_THREAD_SLAVE_SQL);
+    RETURN_NAME_AS_STRING(SYSTEM_THREAD_NDBCLUSTER_BINLOG);
+    RETURN_NAME_AS_STRING(SYSTEM_THREAD_EVENT_SCHEDULER);
+    RETURN_NAME_AS_STRING(SYSTEM_THREAD_EVENT_WORKER);
+  }
+#undef RETURN_NAME_AS_STRING
+  return "UNKNOWN"; /* keep gcc happy */
+}
 
 /**
   This class represents the interface for internal error handlers.
@@ -2092,6 +2108,10 @@ public:
 
       Don't reset binlog format for NDB binlog injector thread.
     */
+    DBUG_PRINT("debug",
+               ("temporary_tables: %p, in_sub_stmt: %d, system_thread: %s",
+                temporary_tables, in_sub_stmt,
+                show_system_thread(system_thread)));
     if ((temporary_tables == NULL) && (in_sub_stmt == 0) &&
         (system_thread != SYSTEM_THREAD_NDBCLUSTER_BINLOG))
     {
diff --git a/sql/sql_insert.cc b/sql/sql_insert.cc
index 268e040288e..8762d3dc8fa 100644
--- a/sql/sql_insert.cc
+++ b/sql/sql_insert.cc
@@ -3635,7 +3635,8 @@ select_create::binlog_show_create_table(TABLE **tables, uint count)
   tmp_table_list.table = *tables;
   query.length(0);      // Have to zero it since constructor doesn't
 
-  result= store_create_info(thd, &tmp_table_list, &query, create_info);
+  result= store_create_info(thd, &tmp_table_list, &query, create_info,
+                            /* show_database */ TRUE);
   DBUG_ASSERT(result == 0); /* store_create_info() always return 0 */
 
   if (mysql_bin_log.is_open())
diff --git a/sql/sql_lex.cc b/sql/sql_lex.cc
index 18c61910e89..4ac73baa992 100644
--- a/sql/sql_lex.cc
+++ b/sql/sql_lex.cc
@@ -1553,6 +1553,7 @@ void st_select_lex::init_query()
   subquery_in_having= explicit_limit= 0;
   is_item_list_lookup= 0;
   first_execution= 1;
+  first_natural_join_processing= 1;
   first_cond_optimization= 1;
   parsing_place= NO_MATTER;
   exclude_from_table_unique_test= no_wrap_view_item= FALSE;
diff --git a/sql/sql_lex.h b/sql/sql_lex.h
index 7674b5c7ac2..53ae984e795 100644
--- a/sql/sql_lex.h
+++ b/sql/sql_lex.h
@@ -667,6 +667,7 @@ public:
     case of an error during prepare the PS is not created.
   */
   bool first_execution;
+  bool first_natural_join_processing;
   bool first_cond_optimization;
   /* do not wrap view fields with Item_ref */
   bool no_wrap_view_item;
diff --git a/sql/sql_parse.cc b/sql/sql_parse.cc
index f293a6dc709..87088c6a8af 100644
--- a/sql/sql_parse.cc
+++ b/sql/sql_parse.cc
@@ -5490,6 +5490,10 @@ void mysql_reset_thd_for_next_command(THD *thd)
   */
   thd->reset_current_stmt_binlog_row_based();
 
+  DBUG_PRINT("debug",
+             ("current_stmt_binlog_row_based: %d",
+              thd->current_stmt_binlog_row_based));
+
   DBUG_VOID_RETURN;
 }
 
diff --git a/sql/sql_show.cc b/sql/sql_show.cc
index d732c3ab110..ca7e222e18d 100644
--- a/sql/sql_show.cc
+++ b/sql/sql_show.cc
@@ -619,7 +619,8 @@ mysqld_show_create(THD *thd, TABLE_LIST *table_list)
 
   if ((table_list->view ?
        view_store_create_info(thd, table_list, &buffer) :
-       store_create_info(thd, table_list, &buffer, NULL)))
+       store_create_info(thd, table_list, &buffer, NULL,
+                         FALSE /* show_database */)))
     DBUG_RETURN(TRUE);
 
   List<Item> field_list;
@@ -810,7 +811,8 @@ mysqld_dump_create_info(THD *thd, TABLE_LIST *table_list, int fd)
   DBUG_PRINT("enter",("table: %s",table_list->table->s->table_name.str));
 
   protocol->prepare_for_resend();
-  if (store_create_info(thd, table_list, packet, NULL))
+  if (store_create_info(thd, table_list, packet, NULL,
+                        FALSE /* show_database */))
     DBUG_RETURN(-1);
 
   if (fd < 0)
@@ -1062,7 +1064,7 @@ static bool get_field_default_value(THD *thd, TABLE *table,
  */
 
 int store_create_info(THD *thd, TABLE_LIST *table_list, String *packet,
-                      HA_CREATE_INFO *create_info_arg)
+                      HA_CREATE_INFO *create_info_arg, bool show_database)
 {
   List<Item> field_list;
   char tmp[MAX_FIELD_WIDTH], *for_str, buff[128], def_value_buf[MAX_FIELD_WIDTH];
@@ -1110,6 +1112,25 @@ int store_create_info(THD *thd, TABLE_LIST *table_list, String *packet,
       alias= share->table_name.str;
     }
   }
+
+  /*
+    Print the database before the table name if told to do that. The
+    database name is only printed in the event that it is different
+    from the current database.  The main reason for doing this is to
+    avoid having to update gazillions of tests and result files, but
+    it also saves a few bytes of the binary log.
+   */
+  if (show_database)
+  {
+    const LEX_STRING *const db=
+      table_list->schema_table ? &INFORMATION_SCHEMA_NAME : &table->s->db;
+    if (strcmp(db->str, thd->db) != 0)
+    {
+      append_identifier(thd, packet, db->str, db->length);
+      packet->append(STRING_WITH_LEN("."));
+    }
+  }
+
   append_identifier(thd, packet, alias, strlen(alias));
   packet->append(STRING_WITH_LEN(" (\n"));
   /*
diff --git a/sql/sql_show.h b/sql/sql_show.h
index d63217584b2..3baaef00a7d 100644
--- a/sql/sql_show.h
+++ b/sql/sql_show.h
@@ -33,7 +33,7 @@ find_files_result find_files(THD *thd, List<LEX_STRING> *files, const char *db,
                              const char *path, const char *wild, bool dir);
 
 int store_create_info(THD *thd, TABLE_LIST *table_list, String *packet,
-                      HA_CREATE_INFO  *create_info_arg);
+                      HA_CREATE_INFO  *create_info_arg, bool show_database);
 int view_store_create_info(THD *thd, TABLE_LIST *table, String *buff);
 
 int copy_event_to_schema_table(THD *thd, TABLE *sch_table, TABLE *event_table);
diff --git a/sql/sql_table.cc b/sql/sql_table.cc
index 4fdd88bbec3..ff9a968f7f2 100644
--- a/sql/sql_table.cc
+++ b/sql/sql_table.cc
@@ -5012,8 +5012,9 @@ bool mysql_create_like_table(THD* thd, TABLE_LIST* table, TABLE_LIST* src_table,
         }
         VOID(pthread_mutex_unlock(&LOCK_open));
 
-        IF_DBUG(int result=) store_create_info(thd, table, &query,
-                                               create_info);
+        IF_DBUG(int result=)
+          store_create_info(thd, table, &query,
+                            create_info, FALSE /* show_database */);
 
         DBUG_ASSERT(result == 0); // store_create_info() always return 0
         write_bin_log(thd, TRUE, query.ptr(), query.length());
diff --git a/sql/sql_union.cc b/sql/sql_union.cc
index 5f7e3c5b620..2f3282f58f6 100644
--- a/sql/sql_union.cc
+++ b/sql/sql_union.cc
@@ -399,7 +399,6 @@ bool st_select_lex_unit::prepare(THD *thd_arg, select_result *sel_result,
     }
     else
     {
-      DBUG_ASSERT(!thd->stmt_arena->is_conventional());
       /*
         We're in execution of a prepared statement or stored procedure:
         reset field items to point at fields from the created temporary table.
diff --git a/sql/sql_update.cc b/sql/sql_update.cc
index e2ed3371711..197e5290aba 100644
--- a/sql/sql_update.cc
+++ b/sql/sql_update.cc
@@ -1076,11 +1076,14 @@ reopen_tables:
   }
 
   /* now lock and fill tables */
-  if (lock_tables(thd, table_list, table_count, &need_reopen))
+  if (!thd->stmt_arena->is_stmt_prepare() &&
+      lock_tables(thd, table_list, table_count, &need_reopen))
   {
     if (!need_reopen)
       DBUG_RETURN(TRUE);
 
+    DBUG_PRINT("info", ("lock_tables failed, reopening"));
+
     /*
       We have to reopen tables since some of them were altered or dropped
       during lock_tables() or something was done with their triggers.
@@ -1096,6 +1099,34 @@ reopen_tables:
     for (TABLE_LIST *tbl= table_list; tbl; tbl= tbl->next_global)
       tbl->cleanup_items();
 
+    /*
+      To not to hog memory (as a result of the 
+      unit->reinit_exec_mechanism() call below):
+    */
+    lex->unit.cleanup();
+
+    for (SELECT_LEX *sl= lex->all_selects_list;
+        sl;
+        sl= sl->next_select_in_list())
+    {
+      SELECT_LEX_UNIT *unit= sl->master_unit();
+      unit->reinit_exec_mechanism(); // reset unit->prepared flags
+      /*
+        Reset 'clean' flag back to force normal execution of
+        unit->cleanup() in Prepared_statement::cleanup_stmt()
+        (call to lex->unit.cleanup() above sets this flag to TRUE).
+      */
+      unit->unclean();
+    }
+
+    /*
+      Also we need to cleanup Natural_join_column::table_field items.
+      To not to traverse a join tree we will cleanup whole
+      thd->free_list (in PS execution mode that list may not contain
+      items from 'fields' list, so the cleanup above is necessary to.
+    */
+    cleanup_items(thd->free_list);
+
     close_tables_for_reopen(thd, &table_list);
     goto reopen_tables;
   }
diff --git a/sql/table.cc b/sql/table.cc
index 40264a7cbb3..89714e4e47e 100644
--- a/sql/table.cc
+++ b/sql/table.cc
@@ -3390,7 +3390,7 @@ TABLE_LIST *TABLE_LIST::find_underlying_table(TABLE *table_to_find)
 }
 
 /*
-  cleunup items belonged to view fields translation table
+  cleanup items belonged to view fields translation table
 
   SYNOPSIS
     TABLE_LIST::cleanup_items()
@@ -3836,10 +3836,10 @@ Natural_join_column::Natural_join_column(Field_translator *field_param,
 }
 
 
-Natural_join_column::Natural_join_column(Field *field_param,
+Natural_join_column::Natural_join_column(Item_field *field_param,
                                          TABLE_LIST *tab)
 {
-  DBUG_ASSERT(tab->table == field_param->table);
+  DBUG_ASSERT(tab->table == field_param->field->table);
   table_field= field_param;
   view_field= NULL;
   table_ref= tab;
@@ -3867,7 +3867,7 @@ Item *Natural_join_column::create_item(THD *thd)
     return create_view_field(thd, table_ref, &view_field->item,
                              view_field->name);
   }
-  return new Item_field(thd, &thd->lex->current_select->context, table_field);
+  return table_field;
 }
 
 
@@ -3878,7 +3878,7 @@ Field *Natural_join_column::field()
     DBUG_ASSERT(table_field == NULL);
     return NULL;
   }
-  return table_field;
+  return table_field->field;
 }
 
 
@@ -4010,7 +4010,7 @@ void Field_iterator_natural_join::next()
   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);
+              cur_column_ref->table_field->field->table);
 }
 
 
@@ -4174,7 +4174,7 @@ GRANT_INFO *Field_iterator_table_ref::grant()
 */
 
 Natural_join_column *
-Field_iterator_table_ref::get_or_create_column_ref(TABLE_LIST *parent_table_ref)
+Field_iterator_table_ref::get_or_create_column_ref(THD *thd, TABLE_LIST *parent_table_ref)
 {
   Natural_join_column *nj_col;
   bool is_created= TRUE;
@@ -4187,7 +4187,11 @@ Field_iterator_table_ref::get_or_create_column_ref(TABLE_LIST *parent_table_ref)
   {
     /* The field belongs to a stored table. */
     Field *tmp_field= table_field_it.field();
-    nj_col= new Natural_join_column(tmp_field, table_ref);
+    Item_field *tmp_item=
+      new Item_field(thd, &thd->lex->current_select->context, tmp_field);
+    if (!tmp_item)
+      return NULL;
+    nj_col= new Natural_join_column(tmp_item, table_ref);
     field_count= table_ref->table->s->fields;
   }
   else if (field_it == &view_field_it)
@@ -4211,7 +4215,7 @@ Field_iterator_table_ref::get_or_create_column_ref(TABLE_LIST *parent_table_ref)
     DBUG_ASSERT(nj_col);
   }
   DBUG_ASSERT(!nj_col->table_field ||
-              nj_col->table_ref->table == nj_col->table_field->table);
+              nj_col->table_ref->table == nj_col->table_field->field->table);
 
   /*
     If the natural join column was just created add it to the list of
@@ -4276,7 +4280,7 @@ Field_iterator_table_ref::get_natural_column_ref()
   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));
+               nj_col->table_ref->table == nj_col->table_field->field->table));
   return nj_col;
 }
 
diff --git a/sql/table.h b/sql/table.h
index fc172dd2f19..ccd6b60664e 100644
--- a/sql/table.h
+++ b/sql/table.h
@@ -18,6 +18,7 @@
 
 class Item;				/* Needed by ORDER */
 class Item_subselect;
+class Item_field;
 class GRANT_TABLE;
 class st_select_lex_unit;
 class st_select_lex;
@@ -1016,7 +1017,7 @@ class Natural_join_column: public Sql_alloc
 {
 public:
   Field_translator *view_field;  /* Column reference of merge view. */
-  Field            *table_field; /* Column reference of table or temp view. */
+  Item_field       *table_field; /* Column reference of table or temp view. */
   TABLE_LIST *table_ref; /* Original base table/view reference. */
   /*
     True if a common join column of two NATURAL/USING join operands. Notice
@@ -1028,7 +1029,7 @@ public:
   bool is_common;
 public:
   Natural_join_column(Field_translator *field_param, TABLE_LIST *tab);
-  Natural_join_column(Field *field_param, TABLE_LIST *tab);
+  Natural_join_column(Item_field *field_param, TABLE_LIST *tab);
   const char *name();
   Item *create_item(THD *thd);
   Field *field();
@@ -1607,7 +1608,7 @@ 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(TABLE_LIST *parent_table_ref);
+  Natural_join_column *get_or_create_column_ref(THD *thd, TABLE_LIST *parent_table_ref);
   Natural_join_column *get_natural_column_ref();
 };