Merge bk-internal.mysql.com:/home/bk/mysql-5.1

into  bodhi.(none):/opt/local/work/mysql-5.1-27430


Makefile.am:
  Auto merged
include/my_global.h:
  Auto merged
mysql-test/include/mix1.inc:
  Auto merged
sql/item.cc:
  Auto merged
sql/my_decimal.h:
  Auto merged
sql/mysqld.cc:
  Auto merged
sql/set_var.cc:
  Auto merged
sql/set_var.h:
  Auto merged
sql/sp.cc:
  Auto merged
sql/sp_head.cc:
  Auto merged
sql/sql_base.cc:
  Auto merged
sql/sql_class.cc:
  Auto merged
sql/sql_class.h:
  Auto merged
sql/sql_parse.cc:
  Auto merged
sql/sql_table.cc:
  Auto merged
sql/sql_update.cc:
  Auto merged
sql/share/errmsg.txt:
  Auto merged
sql/sql_yacc.yy:
  Auto merged
libmysqld/CMakeLists.txt:
  Manual merge.
libmysqld/lib_sql.cc:
  Manual merge.
mysql-test/t/disabled.def:
  Manual merge.

23 files changed, 1115 insertions, 494 deletions

diff --git a/sql/item.cc b/sql/item.cc
index 76f3a818826..96408a70bdd 100644
--- a/sql/item.cc
+++ b/sql/item.cc
@@ -3161,6 +3161,49 @@ void Item_param::print(String *str, enum_query_type query_type)
 }
 
 
+/**
+  Preserve the original parameter types and values
+  when re-preparing a prepared statement.
+
+  @details Copy parameter type information and conversion
+  function pointers from a parameter of the old statement
+  to the corresponding parameter of the new one.
+
+  Move parameter values from the old parameters to the new
+  one. We simply "exchange" the values, which allows
+  to save on allocation and character set conversion in
+  case a parameter is a string or a blob/clob.
+
+  The old parameter gets the value of this one, which
+  ensures that all memory of this parameter is freed
+  correctly.
+
+  @param[in]  src   parameter item of the original
+                    prepared statement
+*/
+
+void
+Item_param::set_param_type_and_swap_value(Item_param *src)
+{
+  unsigned_flag= src->unsigned_flag;
+  param_type= src->param_type;
+  set_param_func= src->set_param_func;
+  item_type= src->item_type;
+  item_result_type= src->item_result_type;
+
+  collation.set(src->collation);
+  maybe_null= src->maybe_null;
+  null_value= src->null_value;
+  max_length= src->max_length;
+  decimals= src->decimals;
+  state= src->state;
+  value= src->value;
+
+  decimal_value.swap(src->decimal_value);
+  str_value.swap(src->str_value);
+  str_value_ptr.swap(src->str_value_ptr);
+}
+
 /****************************************************************************
   Item_copy_string
 ****************************************************************************/
diff --git a/sql/item.h b/sql/item.h
index cd8bb4faccb..465d6f4d54c 100644
--- a/sql/item.h
+++ b/sql/item.h
@@ -1684,6 +1684,7 @@ public:
   bool eq(const Item *item, bool binary_cmp) const;
   /** Item is a argument to a limit clause. */
   bool limit_clause_param;
+  void set_param_type_and_swap_value(Item_param *from);
 };
 
 
diff --git a/sql/my_decimal.h b/sql/my_decimal.h
index e73e0d4080d..0e79f70ab4e 100644
--- a/sql/my_decimal.h
+++ b/sql/my_decimal.h
@@ -114,6 +114,14 @@ public:
   bool sign() const { return decimal_t::sign; }
   void sign(bool s) { decimal_t::sign= s; }
   uint precision() const { return intg + frac; }
+
+  /** Swap two my_decimal values */
+  void swap(my_decimal &rhs)
+  {
+    swap_variables(my_decimal, *this, rhs);
+    /* Swap the buffer pointers back */
+    swap_variables(decimal_digit_t *, buf, rhs.buf);
+  }
 };
 
 
diff --git a/sql/mysql_priv.h b/sql/mysql_priv.h
index 69f214d95f3..0060e7873e1 100644
--- a/sql/mysql_priv.h
+++ b/sql/mysql_priv.h
@@ -259,6 +259,21 @@ protected:
 #define USER_VARS_HASH_SIZE     16
 #define TABLE_OPEN_CACHE_MIN    64
 #define TABLE_OPEN_CACHE_DEFAULT 64
+#define TABLE_DEF_CACHE_DEFAULT 256
+/**
+  We must have room for at least 256 table definitions in the table
+  cache, since otherwise there is no chance prepared
+  statements that use these many tables can work.
+  Prepared statements use table definition cache ids (table_map_id)
+  as table version identifiers. If the table definition
+  cache size is less than the number of tables used in a statement,
+  the contents of the table definition cache is guaranteed to rotate
+  between a prepare and execute. This leads to stable validation
+  errors. In future we shall use more stable version identifiers,
+  for now the only solution is to ensure that the table definition
+  cache can contain at least all tables of a given statement.
+*/
+#define TABLE_DEF_CACHE_MIN     256
 
 /* 
  Value of 9236 discovered through binary search 2006-09-26 on Ubuntu Dapper
@@ -670,6 +685,31 @@ const char *set_thd_proc_info(THD *thd, const char *info,
                               const char *calling_file, 
                               const unsigned int calling_line);
 
+/**
+  Enumerate possible types of a table from re-execution
+  standpoint.
+  TABLE_LIST class has a member of this type.
+  At prepared statement prepare, this member is assigned a value
+  as of the current state of the database. Before (re-)execution
+  of a prepared statement, we check that the value recorded at
+  prepare matches the type of the object we obtained from the
+  table definition cache.
+
+  @sa check_and_update_table_version()
+  @sa Execute_observer
+  @sa Prepared_statement::reprepare()
+*/
+
+enum enum_table_ref_type
+{
+  /** Initial value set by the parser */
+  TABLE_REF_NULL= 0,
+  TABLE_REF_VIEW,
+  TABLE_REF_BASE_TABLE,
+  TABLE_REF_I_S_TABLE,
+  TABLE_REF_TMP_TABLE
+};
+
 /*
   External variables
 */
diff --git a/sql/mysqld.cc b/sql/mysqld.cc
index 0c1ad3ffc69..e34356215e2 100644
--- a/sql/mysqld.cc
+++ b/sql/mysqld.cc
@@ -1957,9 +1957,11 @@ static BOOL WINAPI console_event_handler( DWORD type )
        between main thread doing initialization and CTRL-C thread doing
        cleanup, which can result into crash.
      */
+#ifndef EMBEDDED_LIBRARY
      if(hEventShutdown)
        kill_mysql();
      else
+#endif
        sql_print_warning("CTRL-C ignored during startup");
      DBUG_RETURN(TRUE);
   }
@@ -2843,6 +2845,7 @@ int my_message_sql(uint error, const char *str, myf MyFlags)
       by the stored procedures code.
     */
     if (thd->spcont &&
+        ! (MyFlags & ME_NO_SP_HANDLER) &&
         thd->spcont->handle_error(error, MYSQL_ERROR::WARN_LEVEL_ERROR, thd))
     {
       /*
@@ -2852,7 +2855,8 @@ int my_message_sql(uint error, const char *str, myf MyFlags)
       DBUG_RETURN(0);
     }
 
-    if (!thd->no_warnings_for_error)
+    if (!thd->no_warnings_for_error &&
+        !(MyFlags & ME_NO_WARNING_FOR_ERROR))
     {
       /*
         Suppress infinite recursion if there a memory allocation error
@@ -3092,6 +3096,7 @@ SHOW_VAR com_status_vars[]= {
   {"stmt_execute",         (char*) offsetof(STATUS_VAR, com_stmt_execute), SHOW_LONG_STATUS},
   {"stmt_fetch",           (char*) offsetof(STATUS_VAR, com_stmt_fetch), SHOW_LONG_STATUS},
   {"stmt_prepare",         (char*) offsetof(STATUS_VAR, com_stmt_prepare), SHOW_LONG_STATUS},
+  {"stmt_reprepare",       (char*) offsetof(STATUS_VAR, com_stmt_reprepare), SHOW_LONG_STATUS},
   {"stmt_reset",           (char*) offsetof(STATUS_VAR, com_stmt_reset), SHOW_LONG_STATUS},
   {"stmt_send_long_data",  (char*) offsetof(STATUS_VAR, com_stmt_send_long_data), SHOW_LONG_STATUS},
   {"truncate",             (char*) offsetof(STATUS_VAR, com_stat[(uint) SQLCOM_TRUNCATE]), SHOW_LONG_STATUS},
@@ -3180,7 +3185,7 @@ static int init_common_variables(const char *conf_file_name, int argc,
     We have few debug-only commands in com_status_vars, only visible in debug
     builds. for simplicity we enable the assert only in debug builds
 
-    There are 7 Com_ variables which don't have corresponding SQLCOM_ values:
+    There are 8 Com_ variables which don't have corresponding SQLCOM_ values:
     (TODO strictly speaking they shouldn't be here, should not have Com_ prefix
     that is. Perhaps Stmt_ ? Comstmt_ ? Prepstmt_ ?)
 
@@ -3189,6 +3194,7 @@ static int init_common_variables(const char *conf_file_name, int argc,
       Com_stmt_execute         => com_stmt_execute
       Com_stmt_fetch           => com_stmt_fetch
       Com_stmt_prepare         => com_stmt_prepare
+      Com_stmt_reprepare       => com_stmt_reprepare
       Com_stmt_reset           => com_stmt_reset
       Com_stmt_send_long_data  => com_stmt_send_long_data
 
@@ -3197,7 +3203,7 @@ static int init_common_variables(const char *conf_file_name, int argc,
     of SQLCOM_ constants.
   */
   compile_time_assert(sizeof(com_status_vars)/sizeof(com_status_vars[0]) - 1 ==
-                     SQLCOM_END + 7);
+                     SQLCOM_END + 8);
 #endif
 
   load_defaults(conf_file_name, groups, &argc, &argv);
@@ -6766,7 +6772,8 @@ The minimum value for this variable is 4096.",
   {"table_definition_cache", OPT_TABLE_DEF_CACHE,
    "The number of cached table definitions.",
    (uchar**) &table_def_size, (uchar**) &table_def_size,
-   0, GET_ULONG, REQUIRED_ARG, 128, 1, 512*1024L, 0, 1, 0},
+   0, GET_ULONG, REQUIRED_ARG, TABLE_DEF_CACHE_DEFAULT, TABLE_DEF_CACHE_MIN,
+   512*1024L, 0, 1, 0},
   {"table_open_cache", OPT_TABLE_OPEN_CACHE,
    "The number of cached open tables.",
    (uchar**) &table_cache_size, (uchar**) &table_cache_size, 0, GET_ULONG,
diff --git a/sql/opt_range.cc b/sql/opt_range.cc
index e5709f418f7..b428909d9b7 100644
--- a/sql/opt_range.cc
+++ b/sql/opt_range.cc
@@ -2314,9 +2314,6 @@ int SQL_SELECT::test_quick_select(THD *thd, key_map keys_to_use,
           table deletes.
         */
         if ((thd->lex->sql_command != SQLCOM_DELETE))
-#ifdef NOT_USED
-          if ((thd->lex->sql_command != SQLCOM_UPDATE))
-#endif
         {
           /*
             Get best non-covering ROR-intersection plan and prepare data for
diff --git a/sql/set_var.cc b/sql/set_var.cc
index fd81dc1a867..7d9d88f0281 100644
--- a/sql/set_var.cc
+++ b/sql/set_var.cc
@@ -1667,6 +1667,14 @@ bool sys_var::check_set(THD *thd, set_var *var, TYPELIB *enum_names)
       strmov(buff, "NULL");
       goto err;
     }
+
+    if (!m_allow_empty_value &&
+        res->length() == 0)
+    {
+      buff[0]= 0;
+      goto err;
+    }
+
     var->save_result.ulong_value= ((ulong)
 				   find_set(enum_names, res->c_ptr(),
 					    res->length(),
@@ -1682,10 +1690,19 @@ bool sys_var::check_set(THD *thd, set_var *var, TYPELIB *enum_names)
   else
   {
     ulonglong tmp= var->value->val_int();
-   /*
-     For when the enum is made to contain 64 elements, as 1ULL<<64 is
-     undefined, we guard with a "count<64" test.
-   */
+
+    if (!m_allow_empty_value &&
+        tmp == 0)
+    {
+      buff[0]= '0';
+      buff[1]= 0;
+      goto err;
+    }
+
+    /*
+      For when the enum is made to contain 64 elements, as 1ULL<<64 is
+      undefined, we guard with a "count<64" test.
+    */
     if (unlikely((tmp >= ((ULL(1)) << enum_names->count)) &&
                  (enum_names->count < 64)))
     {
@@ -2385,32 +2402,51 @@ static int  sys_check_log_path(THD *thd,  set_var *var)
   MY_STAT f_stat;
   String str(buff, sizeof(buff), system_charset_info), *res;
   const char *log_file_str;
-      
+  size_t path_length;
+
   if (!(res= var->value->val_str(&str)))
     goto err;
 
   log_file_str= res->c_ptr();
   bzero(&f_stat, sizeof(MY_STAT));
 
-  (void) unpack_filename(path, log_file_str);
-  if (my_stat(path, &f_stat, MYF(0)))
+  path_length= unpack_filename(path, log_file_str);
+
+  if (!path_length)
   {
-    /* Check if argument is a file and we have 'write' permission */
-    if (!MY_S_ISREG(f_stat.st_mode) ||
-        !(f_stat.st_mode & MY_S_IWRITE))
-      goto err;
+    /* File name is empty. */
+
+    goto err;
   }
-  else
+
+  if (my_stat(path, &f_stat, MYF(0)))
   {
-    size_t path_length;
     /*
-      Check if directory exists and 
-      we have permission to create file & write to file
+      A file system object exists. Check if argument is a file and we have
+      'write' permission.
     */
-    (void) dirname_part(path, log_file_str, &path_length);
-    if (my_access(path, (F_OK|W_OK)))
+
+    if (!MY_S_ISREG(f_stat.st_mode) ||
+        !(f_stat.st_mode & MY_S_IWRITE))
       goto err;
+
+    return 0;
   }
+
+  /* Get dirname of the file path. */
+  (void) dirname_part(path, log_file_str, &path_length);
+
+  /* Dirname is empty if file path is relative. */
+  if (!path_length)
+    return 0;
+
+  /*
+    Check if directory exists and we have permission to create file and
+    write to file.
+  */
+  if (my_access(path, (F_OK|W_OK)))
+    goto err;
+
   return 0;
 
 err:
diff --git a/sql/set_var.h b/sql/set_var.h
index 290d1bf0eb1..603c3114d0e 100644
--- a/sql/set_var.h
+++ b/sql/set_var.h
@@ -74,7 +74,8 @@ public:
   sys_var(const char *name_arg, sys_after_update_func func= NULL,
           Binlog_status_enum binlog_status_arg= NOT_IN_BINLOG)
     :name(name_arg), after_update(func), no_support_one_shot(1),
-    binlog_status(binlog_status_arg)
+    binlog_status(binlog_status_arg),
+    m_allow_empty_value(TRUE)
   {}
   virtual ~sys_var() {}
   void chain_sys_var(sys_var_chain *chain_arg)
@@ -109,8 +110,16 @@ public:
   virtual bool is_readonly() const { return 0; }
   virtual sys_var_pluginvar *cast_pluginvar() { return 0; }
 
+protected:
+  void set_allow_empty_value(bool allow_empty_value)
+  {
+    m_allow_empty_value= allow_empty_value;
+  }
+
 private:
   const Binlog_status_enum binlog_status;
+
+  bool m_allow_empty_value;
 };
 
 
@@ -878,8 +887,11 @@ public:
   sys_var_log_output(sys_var_chain *chain, const char *name_arg, ulong *value_arg,
                      TYPELIB *typelib, sys_after_update_func func)
     :sys_var(name_arg,func), value(value_arg), enum_names(typelib)
-  { chain_sys_var(chain); }
-  bool check(THD *thd, set_var *var)
+  {
+    chain_sys_var(chain);
+    set_allow_empty_value(FALSE);
+  }
+  virtual bool check(THD *thd, set_var *var)
   {
     return check_set(thd, var, enum_names);
   }
diff --git a/sql/share/errmsg.txt b/sql/share/errmsg.txt
index 5b9d0ce8617..3fd35fd0c76 100644
--- a/sql/share/errmsg.txt
+++ b/sql/share/errmsg.txt
@@ -6121,8 +6121,15 @@ ER_NO_FORMAT_DESCRIPTION_EVENT_BEFORE_BINLOG_STATEMENT
   eng "The BINLOG statement of type `%s` was not preceded by a format description BINLOG statement."
 ER_SLAVE_CORRUPT_EVENT
   eng "Corrupted replication event was detected"
+
 ER_LOAD_DATA_INVALID_COLUMN
   eng "Invalid column reference (%-.64s) in LOAD DATA"
 
 ER_LOG_PURGE_NO_FILE  
 	eng "Being purged log %s was not found"
+
+ER_NEED_REPREPARE
+  eng "Prepared statement needs to be re-prepared"
+
+ER_PS_REBIND
+  eng "Prepared statement result set has changed, a rebind needed"
diff --git a/sql/sp.cc b/sql/sp.cc
index 8637174efde..a1e5e1291ae 100644
--- a/sql/sp.cc
+++ b/sql/sp.cc
@@ -1076,210 +1076,6 @@ sp_update_routine(THD *thd, int type, sp_name *name, st_sp_chistics *chistics)
 }
 
 
-struct st_used_field
-{
-  const char *field_name;
-  uint field_length;
-  enum enum_field_types field_type;
-  Field *field;
-};
-
-static struct st_used_field init_fields[]=
-{
-  { "Db",                     NAME_CHAR_LEN, MYSQL_TYPE_STRING,    0},
-  { "Name",                   NAME_CHAR_LEN, MYSQL_TYPE_STRING,    0},
-  { "Type",                               9, MYSQL_TYPE_STRING,    0},
-  { "Definer",          USER_HOST_BUFF_SIZE, MYSQL_TYPE_STRING,    0},
-  { "Modified",                           0, MYSQL_TYPE_TIMESTAMP, 0},
-  { "Created",                            0, MYSQL_TYPE_TIMESTAMP, 0},
-  { "Security_type",                      1, MYSQL_TYPE_STRING,    0},
-  { "Comment",                NAME_CHAR_LEN, MYSQL_TYPE_STRING,    0},
-  { "character_set_client", MY_CS_NAME_SIZE, MYSQL_TYPE_STRING,    0},
-  { "collation_connection", MY_CS_NAME_SIZE, MYSQL_TYPE_STRING,    0},
-  { "Database Collation",   MY_CS_NAME_SIZE, MYSQL_TYPE_STRING,    0},
-  { 0,                                    0, MYSQL_TYPE_STRING,    0}
-};
-
-
-static int
-print_field_values(THD *thd, TABLE *table,
-		   struct st_used_field *used_fields,
-		   int type, const char *wild)
-{
-  Protocol *protocol= thd->protocol;
-
-  if (table->field[MYSQL_PROC_MYSQL_TYPE]->val_int() == type)
-  {
-    String db_string;
-    String name_string;
-    struct st_used_field *used_field= used_fields;
-
-    if (get_field(thd->mem_root, used_field->field, &db_string))
-      db_string.set_ascii("", 0);
-    used_field+= 1;
-    get_field(thd->mem_root, used_field->field, &name_string);
-
-    if (!wild || !wild[0] || !wild_compare(name_string.ptr(), wild, 0))
-    {
-      protocol->prepare_for_resend();
-      protocol->store(&db_string);
-      protocol->store(&name_string);
-      for (used_field++;
-	   used_field->field_name;
-	   used_field++)
-      {
-	switch (used_field->field_type) {
-	case MYSQL_TYPE_TIMESTAMP:
-	  {
-	    MYSQL_TIME tmp_time;
-
-	    bzero((char *)&tmp_time, sizeof(tmp_time));
-	    ((Field_timestamp *) used_field->field)->get_time(&tmp_time);
-	    protocol->store(&tmp_time);
-	  }
-	  break;
-	default:
-	  {
-	    String tmp_string;
-
-	    get_field(thd->mem_root, used_field->field, &tmp_string);
-	    protocol->store(&tmp_string);
-	  }
-	  break;
-	}
-      }
-      if (protocol->write())
-	return SP_INTERNAL_ERROR;
-    }
-  }
-
-  return SP_OK;
-}
-
-
-/**
-  Implement SHOW STATUS statement for stored routines.
-
-  @param thd          Thread context.
-  @param type         Stored routine type
-                      (TYPE_ENUM_PROCEDURE or TYPE_ENUM_FUNCTION)
-  @param name_pattern Stored routine name pattern.
-
-  @return Error code. SP_OK is returned on success. Other SP_ constants are
-  used to indicate about errors.
-*/
-
-int
-sp_show_status_routine(THD *thd, int type, const char *name_pattern)
-{
-  TABLE *table;
-  TABLE_LIST tables;
-  int res;
-  DBUG_ENTER("sp_show_status_routine");
-
-  DBUG_ASSERT(type == TYPE_ENUM_PROCEDURE ||
-              type == TYPE_ENUM_FUNCTION);
-
-  memset(&tables, 0, sizeof(tables));
-  tables.db= (char*)"mysql";
-  tables.table_name= tables.alias= (char*)"proc";
-
-  if (! (table= open_ltable(thd, &tables, TL_READ, 0)))
-  {
-    res= SP_OPEN_TABLE_FAILED;
-    goto done;
-  }
-  else
-  {
-    Item *item;
-    List<Item> field_list;
-    struct st_used_field *used_field;
-    TABLE_LIST *leaves= 0;
-    st_used_field used_fields[array_elements(init_fields)];
-
-    table->use_all_columns();
-    memcpy((char*) used_fields, (char*) init_fields, sizeof(used_fields));
-    /* Init header */
-    for (used_field= &used_fields[0];
-	 used_field->field_name;
-	 used_field++)
-    {
-      switch (used_field->field_type) {
-      case MYSQL_TYPE_TIMESTAMP:
-	item= new Item_return_date_time(used_field->field_name,
-                                        MYSQL_TYPE_DATETIME);
-	field_list.push_back(item);
-	break;
-      default:
-        item= new Item_empty_string(used_field->field_name,
-                                    used_field->field_length);
-	field_list.push_back(item);
-	break;
-      }
-    }
-    /* Print header */
-    if (thd->protocol->send_fields(&field_list, Protocol::SEND_NUM_ROWS |
-                                                Protocol::SEND_EOF))
-    {
-      res= SP_INTERNAL_ERROR;
-      goto err_case;
-    }
-
-    /*
-      Init fields
-
-      tables is not VIEW for sure => we can pass 0 as condition
-    */
-    thd->lex->select_lex.context.resolve_in_table_list_only(&tables);
-    setup_tables(thd, &thd->lex->select_lex.context,
-                 &thd->lex->select_lex.top_join_list,
-                 &tables, &leaves, FALSE);
-    for (used_field= &used_fields[0];
-	 used_field->field_name;
-	 used_field++)
-    {
-      Item_field *field= new Item_field(&thd->lex->select_lex.context,
-                                        "mysql", "proc",
-					used_field->field_name);
-      if (!field ||
-          !(used_field->field= find_field_in_tables(thd, field, &tables, NULL,
-						    0, REPORT_ALL_ERRORS, 1,
-                                                    TRUE)))
-      {
-	res= SP_INTERNAL_ERROR;
-	goto err_case1;
-      }
-    }
-
-    table->file->ha_index_init(0, 1);
-    if ((res= table->file->index_first(table->record[0])))
-    {
-      res= (res == HA_ERR_END_OF_FILE) ? 0 : SP_INTERNAL_ERROR;
-      goto err_case1;
-    }
-
-    do
-    {
-      res= print_field_values(thd, table, used_fields, type, name_pattern);
-
-      if (res)
-	goto err_case1;
-    }
-    while (!table->file->index_next(table->record[0]));
-
-    res= SP_OK;
-  }
-
-err_case1:
-  my_eof(thd);
-err_case:
-  table->file->ha_index_end();
-  close_thread_tables(thd);
-done:
-  DBUG_RETURN(res);
-}
-
-
 /**
   Drop all routines in database 'db'
 
diff --git a/sql/sp.h b/sql/sp.h
index 31173e1f90c..75088ea0b83 100644
--- a/sql/sp.h
+++ b/sql/sp.h
@@ -49,9 +49,6 @@ bool
 sp_show_create_routine(THD *thd, int type, sp_name *name);
 
 int
-sp_show_status_routine(THD *thd, int type, const char *wild);
-
-int
 sp_create_routine(THD *thd, int type, sp_head *sp);
 
 int
diff --git a/sql/sp_head.cc b/sql/sp_head.cc
index e342a3e9451..c450c9dd25d 100644
--- a/sql/sp_head.cc
+++ b/sql/sp_head.cc
@@ -1072,6 +1072,7 @@ sp_head::execute(THD *thd)
   LEX *old_lex;
   Item_change_list old_change_list;
   String old_packet;
+  Reprepare_observer *save_reprepare_observer= thd->m_reprepare_observer;
 
   Object_creation_ctx *saved_creation_ctx;
 
@@ -1139,6 +1140,25 @@ sp_head::execute(THD *thd)
   thd->variables.sql_mode= m_sql_mode;
   save_abort_on_warning= thd->abort_on_warning;
   thd->abort_on_warning= 0;
+  /**
+    When inside a substatement (a stored function or trigger
+    statement), clear the metadata observer in THD, if any.
+    Remember the value of the observer here, to be able
+    to restore it when leaving the substatement.
+
+    We reset the observer to suppress errors when a substatement
+    uses temporary tables. If a temporary table does not exist
+    at start of the main statement, it's not prelocked
+    and thus is not validated with other prelocked tables.
+
+    Later on, when the temporary table is opened, metadata
+    versions mismatch, expectedly.
+
+    The proper solution for the problem is to re-validate tables
+    of substatements (Bug#12257, Bug#27011, Bug#32868, Bug#33000),
+    but it's not implemented yet.
+  */
+  thd->m_reprepare_observer= 0;
 
   /*
     It is also more efficient to save/restore current thd->lex once when
@@ -1301,6 +1321,7 @@ sp_head::execute(THD *thd)
   thd->derived_tables= old_derived_tables;
   thd->variables.sql_mode= save_sql_mode;
   thd->abort_on_warning= save_abort_on_warning;
+  thd->m_reprepare_observer= save_reprepare_observer;
 
   thd->stmt_arena= old_arena;
   state= EXECUTED;
diff --git a/sql/sql_acl.cc b/sql/sql_acl.cc
index 92eb68dc2fa..0b33abcfba2 100644
--- a/sql/sql_acl.cc
+++ b/sql/sql_acl.cc
@@ -695,6 +695,8 @@ my_bool acl_reload(THD *thd)
   tables[0].next_local= tables[0].next_global= tables+1;
   tables[1].next_local= tables[1].next_global= tables+2;
   tables[0].lock_type=tables[1].lock_type=tables[2].lock_type=TL_READ;
+  tables[0].skip_temporary= tables[1].skip_temporary=
+    tables[2].skip_temporary= TRUE;
 
   if (simple_open_n_lock_tables(thd, tables))
   {
@@ -3537,7 +3539,7 @@ static my_bool grant_load_procs_priv(TABLE *p_table)
   bool check_no_resolve= specialflag & SPECIAL_NO_RESOLVE;
   MEM_ROOT **save_mem_root_ptr= my_pthread_getspecific_ptr(MEM_ROOT**,
                                                            THR_MALLOC);
-  DBUG_ENTER("grant_load");
+  DBUG_ENTER("grant_load_procs_priv");
   (void) hash_init(&proc_priv_hash,system_charset_info,
                    0,0,0, (hash_get_key) get_grant_table,
                    0,0);
@@ -3721,6 +3723,7 @@ static my_bool grant_reload_procs_priv(THD *thd)
   table.alias= table.table_name= (char*) "procs_priv";
   table.db= (char *) "mysql";
   table.lock_type= TL_READ;
+  table.skip_temporary= 1;
 
   if (simple_open_n_lock_tables(thd, &table))
   {
@@ -3786,7 +3789,7 @@ my_bool grant_reload(THD *thd)
   tables[0].db= tables[1].db= (char *) "mysql";
   tables[0].next_local= tables[0].next_global= tables+1;
   tables[0].lock_type= tables[1].lock_type= TL_READ;
-
+  tables[0].skip_temporary= tables[1].skip_temporary= TRUE;
   /*
     To avoid deadlocks we should obtain table locks before
     obtaining LOCK_grant rwlock.
diff --git a/sql/sql_base.cc b/sql/sql_base.cc
index b680bbc1308..d25057789ac 100644
--- a/sql/sql_base.cc
+++ b/sql/sql_base.cc
@@ -345,26 +345,9 @@ TABLE_SHARE *get_table_share(THD *thd, TABLE_LIST *table_list, char *key,
 
   if (!(share= alloc_table_share(table_list, key, key_length)))
   {
-#ifdef WAITING_FOR_TABLE_DEF_CACHE_STAGE_3
-    pthread_mutex_unlock(&LOCK_open);
-#endif
     DBUG_RETURN(0);
   }
 
-#ifdef WAITING_FOR_TABLE_DEF_CACHE_STAGE_3
-  // We need a write lock to be able to add a new entry
-  pthread_mutex_unlock(&LOCK_open);
-  pthread_mutex_lock(&LOCK_open);
-  /* Check that another thread didn't insert the same table in between */
-  if ((old_share= hash_search(&table_def_cache, (uchar*) key, key_length)))
-  {
-    (void) pthread_mutex_lock(&share->mutex);
-    free_table_share(share);
-    share= old_share;
-    goto found;
-  }
-#endif
-
   /*
     Lock mutex to be able to read table definition from file without
     conflicts
@@ -388,29 +371,11 @@ TABLE_SHARE *get_table_share(THD *thd, TABLE_LIST *table_list, char *key,
 
   if (my_hash_insert(&table_def_cache, (uchar*) share))
   {
-#ifdef WAITING_FOR_TABLE_DEF_CACHE_STAGE_3
-    pthread_mutex_unlock(&LOCK_open);    
-    (void) pthread_mutex_unlock(&share->mutex);
-#endif
     free_table_share(share);
     DBUG_RETURN(0);				// return error
   }
-#ifdef WAITING_FOR_TABLE_DEF_CACHE_STAGE_3
-  pthread_mutex_unlock(&LOCK_open);
-#endif
   if (open_table_def(thd, share, db_flags))
   {
-#ifdef WAITING_FOR_TABLE_DEF_CACHE_STAGE_3
-    /*
-      No such table or wrong table definition file
-      Lock first the table cache and then the mutex.
-      This will ensure that no other thread is using the share
-      structure.
-    */
-    (void) pthread_mutex_unlock(&share->mutex);
-    (void) pthread_mutex_lock(&LOCK_open);
-    (void) pthread_mutex_lock(&share->mutex);
-#endif
     *error= share->error;
     (void) hash_delete(&table_def_cache, (uchar*) share);
     DBUG_RETURN(0);
@@ -429,9 +394,6 @@ found:
 
   /* We must do a lock to ensure that the structure is initialized */
   (void) pthread_mutex_lock(&share->mutex);
-#ifdef WAITING_FOR_TABLE_DEF_CACHE_STAGE_3
-  pthread_mutex_unlock(&LOCK_open);
-#endif
   if (share->error)
   {
     /* Table definition contained an error */
@@ -618,52 +580,6 @@ void release_table_share(TABLE_SHARE *share, enum release_type type)
   }
   pthread_mutex_unlock(&share->mutex);
   DBUG_VOID_RETURN;
-
-
-#ifdef WAITING_FOR_TABLE_DEF_CACHE_STAGE_3
-  if (to_be_deleted)
-  {
-    /*
-      We must try again with new locks as we must get LOCK_open
-      before share->mutex
-    */
-    pthread_mutex_unlock(&share->mutex);
-    pthread_mutex_lock(&LOCK_open);
-    pthread_mutex_lock(&share->mutex);
-    if (!share->ref_count)
-    {						// No one is using this now
-      TABLE_SHARE *name_lock;
-      if (share->replace_with_name_lock && (name_lock=get_name_lock(share)))
-      {
-	/*
-	  This code is execured when someone does FLUSH TABLES while on has
-	  locked tables.
-	 */
-	(void) hash_search(&def_cache,(uchar*) key,key_length);
-	hash_replace(&def_cache, def_cache.current_record,(uchar*) name_lock);
-      }
-      else
-      {
-	/* Remove table definition */
-	hash_delete(&def_cache,(uchar*) share);
-      }
-      pthread_mutex_unlock(&LOCK_open);
-      free_table_share(share);
-    }
-    else
-    {
-      pthread_mutex_unlock(&LOCK_open);
-      if (type == RELEASE_WAIT_FOR_DROP)
-	wait_for_table(share, "Waiting for close");
-      else
-	pthread_mutex_unlock(&share->mutex);
-    }
-  }
-  else if (type == RELEASE_WAIT_FOR_DROP)
-    wait_for_table(share, "Waiting for close");
-  else
-    pthread_mutex_unlock(&share->mutex);
-#endif
 }
 
 
@@ -3801,6 +3717,73 @@ void assign_new_table_id(TABLE_SHARE *share)
   DBUG_VOID_RETURN;
 }
 
+
+/**
+  Compare metadata versions of an element obtained from the table
+  definition cache and its corresponding node in the parse tree.
+
+  @details If the new and the old values mismatch, invoke
+  Metadata_version_observer.
+  At prepared statement prepare, all TABLE_LIST version values are
+  NULL and we always have a mismatch. But there is no observer set
+  in THD, and therefore no error is reported. Instead, we update
+  the value in the parse tree, effectively recording the original
+  version.
+  At prepared statement execute, an observer may be installed.  If
+  there is a version mismatch, we push an error and return TRUE.
+
+  For conventional execution (no prepared statements), the
+  observer is never installed.
+
+  @sa Execute_observer
+  @sa check_prepared_statement() to see cases when an observer is installed
+  @sa TABLE_LIST::is_table_ref_id_equal()
+  @sa TABLE_SHARE::get_table_ref_id()
+
+  @param[in]      thd         used to report errors
+  @param[in,out]  tables      TABLE_LIST instance created by the parser
+                              Metadata version information in this object
+                              is updated upon success.
+  @param[in]      table_share an element from the table definition cache
+
+  @retval  TRUE  an error, which has been reported
+  @retval  FALSE success, version in TABLE_LIST has been updated
+*/
+
+bool
+check_and_update_table_version(THD *thd,
+                               TABLE_LIST *tables, TABLE_SHARE *table_share)
+{
+  if (! tables->is_table_ref_id_equal(table_share))
+  {
+    if (thd->m_reprepare_observer &&
+        thd->m_reprepare_observer->report_error(thd))
+    {
+      /*
+        Version of the table share is different from the
+        previous execution of the prepared statement, and it is
+        unacceptable for this SQLCOM. Error has been reported.
+      */
+      DBUG_ASSERT(thd->is_error());
+      return TRUE;
+    }
+    /* Always maintain the latest version and type */
+    tables->set_table_ref_id(table_share);
+  }
+
+#ifndef DBUG_OFF
+  /* Spuriously reprepare each statement. */
+  if (_db_strict_keyword_("reprepare_each_statement") &&
+      thd->m_reprepare_observer && thd->stmt_arena->is_reprepared == FALSE)
+  {
+    thd->m_reprepare_observer->report_error(thd);
+    return TRUE;
+  }
+#endif
+
+  return FALSE;
+}
+
 /*
   Load a table definition from file and open unireg table
 
@@ -3846,6 +3829,12 @@ retry:
 
   if (share->is_view)
   {
+    /*
+      This table is a view. Validate its metadata version: in particular,
+      that it was a view when the statement was prepared.
+    */
+    if (check_and_update_table_version(thd, table_list, share))
+      goto err;
     if (table_list->i_s_requested_object &  OPEN_TABLE_ONLY)
       goto err;
 
@@ -3863,6 +3852,26 @@ retry:
     release_table_share(share, RELEASE_NORMAL);
     DBUG_RETURN((flags & OPEN_VIEW_NO_PARSE)? -1 : 0);
   }
+  else if (table_list->view)
+  {
+    /*
+      We're trying to open a table for what was a view.
+      This can only happen during (re-)execution.
+      At prepared statement prepare the view has been opened and
+      merged into the statement parse tree. After that, someone
+      performed a DDL and replaced the view with a base table.
+      Don't try to open the table inside a prepared statement,
+      invalidate it instead.
+
+      Note, the assert below is known to fail inside stored
+      procedures (Bug#27011).
+    */
+    DBUG_ASSERT(thd->m_reprepare_observer);
+    check_and_update_table_version(thd, table_list, share);
+    /* Always an error. */
+    DBUG_ASSERT(thd->is_error());
+    goto err;
+  }
 
   if (table_list->i_s_requested_object &  OPEN_VIEW_ONLY)
     goto err;
@@ -4464,8 +4473,18 @@ int open_tables(THD *thd, TABLE_LIST **start, uint *counter, uint flags)
     */
     if (tables->schema_table)
     {
-      if (!mysql_schema_table(thd, thd->lex, tables))
+      /*
+        If this information_schema table is merged into a mergeable
+        view, ignore it for now -- it will be filled when its respective
+        TABLE_LIST is processed. This code works only during re-execution.
+      */
+      if (tables->view)
+        goto process_view_routines;
+      if (!mysql_schema_table(thd, thd->lex, tables) &&
+          !check_and_update_table_version(thd, tables, tables->table->s))
+      {
         continue;
+      }
       DBUG_RETURN(-1);
     }
     (*counter)++;
@@ -4613,6 +4632,13 @@ int open_tables(THD *thd, TABLE_LIST **start, uint *counter, uint flags)
     }
     tables->table->grant= tables->grant;
 
+    /* Check and update metadata version of a base table. */
+    if (check_and_update_table_version(thd, tables, tables->table->s))
+    {
+      result= -1;
+      goto err;
+    }
+
     /* Attach MERGE children if not locked already. */
     DBUG_PRINT("tcache", ("is parent: %d  is child: %d",
                           test(tables->table->child_l),
@@ -4671,7 +4697,11 @@ process_view_routines:
       error happens on a MERGE child, clear the parents TABLE reference.
     */
     if (tables->parent_l)
+    {
+      if (tables->parent_l->next_global == tables->parent_l->table->child_l)
+        tables->parent_l->next_global= *tables->parent_l->table->child_last_l;
       tables->parent_l->table= NULL;
+    }
     tables->table= NULL;
   }
   DBUG_PRINT("tcache", ("returning: %d", result));
diff --git a/sql/sql_class.cc b/sql/sql_class.cc
index 5ee4f3dea02..afa9777e00a 100644
--- a/sql/sql_class.cc
+++ b/sql/sql_class.cc
@@ -198,6 +198,19 @@ bool foreign_key_prefix(Key *a, Key *b)
 ** Thread specific functions
 ****************************************************************************/
 
+/** Push an error to the error stack and return TRUE for now. */
+
+bool
+Reprepare_observer::report_error(THD *thd)
+{
+  my_error(ER_NEED_REPREPARE, MYF(ME_NO_WARNING_FOR_ERROR|ME_NO_SP_HANDLER));
+
+  m_invalidated= TRUE;
+
+  return TRUE;
+}
+
+
 Open_tables_state::Open_tables_state(ulong version_arg)
   :version(version_arg), state_flags(0U)
 {
@@ -2770,7 +2783,8 @@ void THD::restore_backup_open_tables_state(Open_tables_state *backup)
   DBUG_ASSERT(open_tables == 0 && temporary_tables == 0 &&
               handler_tables == 0 && derived_tables == 0 &&
               lock == 0 && locked_tables == 0 &&
-              prelocked_mode == NON_PRELOCKED);
+              prelocked_mode == NON_PRELOCKED &&
+              m_reprepare_observer == NULL);
   set_open_tables_state(backup);
   DBUG_VOID_RETURN;
 }
diff --git a/sql/sql_class.h b/sql/sql_class.h
index 446b7f163d9..a86b4d36d00 100644
--- a/sql/sql_class.h
+++ b/sql/sql_class.h
@@ -23,6 +23,51 @@
 #include "log.h"
 #include "rpl_tblmap.h"
 
+/**
+  An interface that is used to take an action when
+  the locking module notices that a table version has changed
+  since the last execution. "Table" here may refer to any kind of
+  table -- a base table, a temporary table, a view or an
+  information schema table.
+
+  When we open and lock tables for execution of a prepared
+  statement, we must verify that they did not change
+  since statement prepare. If some table did change, the statement
+  parse tree *may* be no longer valid, e.g. in case it contains
+  optimizations that depend on table metadata.
+
+  This class provides an interface (a method) that is
+  invoked when such a situation takes place.
+  The implementation of the method simply reports an error, but
+  the exact details depend on the nature of the SQL statement.
+
+  At most 1 instance of this class is active at a time, in which
+  case THD::m_reprepare_observer is not NULL.
+
+  @sa check_and_update_table_version() for details of the
+  version tracking algorithm 
+
+  @sa Open_tables_state::m_reprepare_observer for the life cycle
+  of metadata observers.
+*/
+
+class Reprepare_observer
+{
+public:
+  /**
+    Check if a change of metadata is OK. In future
+    the signature of this method may be extended to accept the old
+    and the new versions, but since currently the check is very
+    simple, we only need the THD to report an error.
+  */
+  bool report_error(THD *thd);
+  bool is_invalidated() const { return m_invalidated; }
+  void reset_reprepare_observer() { m_invalidated= FALSE; }
+private:
+  bool m_invalidated;
+};
+
+
 class Relay_log_info;
 
 class Query_log_event;
@@ -406,6 +451,7 @@ typedef struct system_status_var
   ulong filesort_scan_count;
   /* Prepared statements and binary protocol */
   ulong com_stmt_prepare;
+  ulong com_stmt_reprepare;
   ulong com_stmt_execute;
   ulong com_stmt_send_long_data;
   ulong com_stmt_fetch;
@@ -436,7 +482,7 @@ void free_tmp_table(THD *thd, TABLE *entry);
 
 /* The following macro is to make init of Query_arena simpler */
 #ifndef DBUG_OFF
-#define INIT_ARENA_DBUG_INFO is_backup_arena= 0
+#define INIT_ARENA_DBUG_INFO is_backup_arena= 0; is_reprepared= FALSE;
 #else
 #define INIT_ARENA_DBUG_INFO
 #endif
@@ -452,6 +498,7 @@ public:
   MEM_ROOT *mem_root;                   // Pointer to current memroot
 #ifndef DBUG_OFF
   bool is_backup_arena; /* True if this arena is used for backup. */
+  bool is_reprepared;
 #endif
   /*
     The states relfects three diffrent life cycles for three
@@ -789,6 +836,20 @@ class Open_tables_state
 {
 public:
   /**
+    As part of class THD, this member is set during execution
+    of a prepared statement. When it is set, it is used
+    by the locking subsystem to report a change in table metadata.
+
+    When Open_tables_state part of THD is reset to open
+    a system or INFORMATION_SCHEMA table, the member is cleared
+    to avoid spurious ER_NEED_REPREPARE errors -- system and
+    INFORMATION_SCHEMA tables are not subject to metadata version
+    tracking.
+    @sa check_and_update_table_version()
+  */
+  Reprepare_observer *m_reprepare_observer;
+
+  /**
     List of regular tables in use by this thread. Contains temporary and
     base tables that were opened with @see open_tables().
   */
@@ -891,6 +952,7 @@ public:
     extra_lock= lock= locked_tables= 0;
     prelocked_mode= NON_PRELOCKED;
     state_flags= 0U;
+    m_reprepare_observer= NULL;
   }
 };
 
@@ -2812,6 +2874,20 @@ public:
 #define CF_STATUS_COMMAND	4
 #define CF_SHOW_TABLE_COMMAND	8
 #define CF_WRITE_LOGS_COMMAND  16
+/**
+  Must be set for SQL statements that may contain
+  Item expressions and/or use joins and tables.
+  Indicates that the parse tree of such statement may
+  contain rule-based optimizations that depend on metadata
+  (i.e. number of columns in a table), and consequently
+  that the statement must be re-prepared whenever
+  referenced metadata changes. Must not be set for
+  statements that themselves change metadata, e.g. RENAME,
+  ALTER and other DDL, since otherwise will trigger constant
+  reprepare. Consequently, complex item expressions and
+  joins are currently prohibited in these statements.
+*/
+#define CF_REEXECUTION_FRAGILE 32
 
 /* Functions in sql_class.cc */
 
diff --git a/sql/sql_cursor.cc b/sql/sql_cursor.cc
index 5c4e93d4c74..d33680e1e0b 100644
--- a/sql/sql_cursor.cc
+++ b/sql/sql_cursor.cc
@@ -111,7 +111,8 @@ class Select_materialize: public select_union
   select_result *result; /**< the result object of the caller (PS or SP) */
 public:
   Materialized_cursor *materialized_cursor;
-  Select_materialize(select_result *result_arg) :result(result_arg) {}
+  Select_materialize(select_result *result_arg)
+    :result(result_arg), materialized_cursor(0) {}
   virtual bool send_fields(List<Item> &list, uint flags);
 };
 
diff --git a/sql/sql_parse.cc b/sql/sql_parse.cc
index ddcec92a0d3..ee251e6511b 100644
--- a/sql/sql_parse.cc
+++ b/sql/sql_parse.cc
@@ -200,45 +200,56 @@ void init_update_queries(void)
 {
   bzero((uchar*) &sql_command_flags, sizeof(sql_command_flags));
 
-  sql_command_flags[SQLCOM_CREATE_TABLE]=   CF_CHANGES_DATA;
+  sql_command_flags[SQLCOM_CREATE_TABLE]=   CF_CHANGES_DATA | CF_REEXECUTION_FRAGILE;
   sql_command_flags[SQLCOM_CREATE_INDEX]=   CF_CHANGES_DATA;
   sql_command_flags[SQLCOM_ALTER_TABLE]=    CF_CHANGES_DATA | CF_WRITE_LOGS_COMMAND;
   sql_command_flags[SQLCOM_TRUNCATE]=       CF_CHANGES_DATA | CF_WRITE_LOGS_COMMAND;
   sql_command_flags[SQLCOM_DROP_TABLE]=     CF_CHANGES_DATA;
-  sql_command_flags[SQLCOM_LOAD]=           CF_CHANGES_DATA;
+  sql_command_flags[SQLCOM_LOAD]=           CF_CHANGES_DATA | CF_REEXECUTION_FRAGILE;
   sql_command_flags[SQLCOM_CREATE_DB]=      CF_CHANGES_DATA;
   sql_command_flags[SQLCOM_DROP_DB]=        CF_CHANGES_DATA;
   sql_command_flags[SQLCOM_RENAME_TABLE]=   CF_CHANGES_DATA;
   sql_command_flags[SQLCOM_BACKUP_TABLE]=   CF_CHANGES_DATA;
   sql_command_flags[SQLCOM_RESTORE_TABLE]=  CF_CHANGES_DATA;
   sql_command_flags[SQLCOM_DROP_INDEX]=     CF_CHANGES_DATA;
-  sql_command_flags[SQLCOM_CREATE_VIEW]=    CF_CHANGES_DATA;
+  sql_command_flags[SQLCOM_CREATE_VIEW]=    CF_CHANGES_DATA | CF_REEXECUTION_FRAGILE;
   sql_command_flags[SQLCOM_DROP_VIEW]=      CF_CHANGES_DATA;
   sql_command_flags[SQLCOM_CREATE_EVENT]=   CF_CHANGES_DATA;
   sql_command_flags[SQLCOM_ALTER_EVENT]=    CF_CHANGES_DATA;
   sql_command_flags[SQLCOM_DROP_EVENT]=     CF_CHANGES_DATA;
 
-  sql_command_flags[SQLCOM_UPDATE]=	    CF_CHANGES_DATA | CF_HAS_ROW_COUNT;
-  sql_command_flags[SQLCOM_UPDATE_MULTI]=   CF_CHANGES_DATA | CF_HAS_ROW_COUNT;
-  sql_command_flags[SQLCOM_INSERT]=	    CF_CHANGES_DATA | CF_HAS_ROW_COUNT;
-  sql_command_flags[SQLCOM_INSERT_SELECT]=  CF_CHANGES_DATA | CF_HAS_ROW_COUNT;
-  sql_command_flags[SQLCOM_DELETE]=         CF_CHANGES_DATA | CF_HAS_ROW_COUNT;
-  sql_command_flags[SQLCOM_DELETE_MULTI]=   CF_CHANGES_DATA | CF_HAS_ROW_COUNT;
-  sql_command_flags[SQLCOM_REPLACE]=        CF_CHANGES_DATA | CF_HAS_ROW_COUNT;
-  sql_command_flags[SQLCOM_REPLACE_SELECT]= CF_CHANGES_DATA | CF_HAS_ROW_COUNT;
-
-  sql_command_flags[SQLCOM_SHOW_STATUS_PROC]= CF_STATUS_COMMAND;
-  sql_command_flags[SQLCOM_SHOW_STATUS]=      CF_STATUS_COMMAND;
-  sql_command_flags[SQLCOM_SHOW_DATABASES]=   CF_STATUS_COMMAND;
-  sql_command_flags[SQLCOM_SHOW_TRIGGERS]=    CF_STATUS_COMMAND;
-  sql_command_flags[SQLCOM_SHOW_EVENTS]=      CF_STATUS_COMMAND;
-  sql_command_flags[SQLCOM_SHOW_OPEN_TABLES]= CF_STATUS_COMMAND;
+  sql_command_flags[SQLCOM_UPDATE]=	    CF_CHANGES_DATA | CF_HAS_ROW_COUNT |
+                                            CF_REEXECUTION_FRAGILE;
+  sql_command_flags[SQLCOM_UPDATE_MULTI]=   CF_CHANGES_DATA | CF_HAS_ROW_COUNT |
+                                            CF_REEXECUTION_FRAGILE;
+  sql_command_flags[SQLCOM_INSERT]=	    CF_CHANGES_DATA | CF_HAS_ROW_COUNT |
+                                            CF_REEXECUTION_FRAGILE;
+  sql_command_flags[SQLCOM_INSERT_SELECT]=  CF_CHANGES_DATA | CF_HAS_ROW_COUNT |
+                                            CF_REEXECUTION_FRAGILE;
+  sql_command_flags[SQLCOM_DELETE]=         CF_CHANGES_DATA | CF_HAS_ROW_COUNT |
+                                            CF_REEXECUTION_FRAGILE;
+  sql_command_flags[SQLCOM_DELETE_MULTI]=   CF_CHANGES_DATA | CF_HAS_ROW_COUNT |
+                                            CF_REEXECUTION_FRAGILE;
+  sql_command_flags[SQLCOM_REPLACE]=        CF_CHANGES_DATA | CF_HAS_ROW_COUNT |
+                                            CF_REEXECUTION_FRAGILE;
+  sql_command_flags[SQLCOM_REPLACE_SELECT]= CF_CHANGES_DATA | CF_HAS_ROW_COUNT |
+                                            CF_REEXECUTION_FRAGILE;
+  sql_command_flags[SQLCOM_SELECT]=         CF_REEXECUTION_FRAGILE;
+  sql_command_flags[SQLCOM_SET_OPTION]=     CF_REEXECUTION_FRAGILE;
+  sql_command_flags[SQLCOM_DO]=             CF_REEXECUTION_FRAGILE;
+
+  sql_command_flags[SQLCOM_SHOW_STATUS_PROC]= CF_STATUS_COMMAND | CF_REEXECUTION_FRAGILE;
+  sql_command_flags[SQLCOM_SHOW_STATUS]=      CF_STATUS_COMMAND | CF_REEXECUTION_FRAGILE;
+  sql_command_flags[SQLCOM_SHOW_DATABASES]=   CF_STATUS_COMMAND | CF_REEXECUTION_FRAGILE;
+  sql_command_flags[SQLCOM_SHOW_TRIGGERS]=    CF_STATUS_COMMAND | CF_REEXECUTION_FRAGILE;
+  sql_command_flags[SQLCOM_SHOW_EVENTS]=      CF_STATUS_COMMAND | CF_REEXECUTION_FRAGILE;
+  sql_command_flags[SQLCOM_SHOW_OPEN_TABLES]= CF_STATUS_COMMAND | CF_REEXECUTION_FRAGILE;
   sql_command_flags[SQLCOM_SHOW_PLUGINS]=     CF_STATUS_COMMAND;
-  sql_command_flags[SQLCOM_SHOW_FIELDS]=      CF_STATUS_COMMAND;
-  sql_command_flags[SQLCOM_SHOW_KEYS]=        CF_STATUS_COMMAND;
-  sql_command_flags[SQLCOM_SHOW_VARIABLES]=   CF_STATUS_COMMAND;
-  sql_command_flags[SQLCOM_SHOW_CHARSETS]=    CF_STATUS_COMMAND;
-  sql_command_flags[SQLCOM_SHOW_COLLATIONS]=  CF_STATUS_COMMAND;
+  sql_command_flags[SQLCOM_SHOW_FIELDS]=      CF_STATUS_COMMAND | CF_REEXECUTION_FRAGILE;
+  sql_command_flags[SQLCOM_SHOW_KEYS]=        CF_STATUS_COMMAND | CF_REEXECUTION_FRAGILE;
+  sql_command_flags[SQLCOM_SHOW_VARIABLES]=   CF_STATUS_COMMAND | CF_REEXECUTION_FRAGILE;
+  sql_command_flags[SQLCOM_SHOW_CHARSETS]=    CF_STATUS_COMMAND | CF_REEXECUTION_FRAGILE;
+  sql_command_flags[SQLCOM_SHOW_COLLATIONS]=  CF_STATUS_COMMAND | CF_REEXECUTION_FRAGILE;
   sql_command_flags[SQLCOM_SHOW_NEW_MASTER]= CF_STATUS_COMMAND;
   sql_command_flags[SQLCOM_SHOW_BINLOGS]= CF_STATUS_COMMAND;
   sql_command_flags[SQLCOM_SHOW_SLAVE_HOSTS]= CF_STATUS_COMMAND;
@@ -262,7 +273,7 @@ void init_update_queries(void)
   sql_command_flags[SQLCOM_SHOW_CREATE_PROC]=  CF_STATUS_COMMAND;
   sql_command_flags[SQLCOM_SHOW_CREATE_FUNC]=  CF_STATUS_COMMAND;
   sql_command_flags[SQLCOM_SHOW_CREATE_TRIGGER]=  CF_STATUS_COMMAND;
-  sql_command_flags[SQLCOM_SHOW_STATUS_FUNC]=  CF_STATUS_COMMAND;
+  sql_command_flags[SQLCOM_SHOW_STATUS_FUNC]=  CF_STATUS_COMMAND | CF_REEXECUTION_FRAGILE;
   sql_command_flags[SQLCOM_SHOW_PROC_CODE]=  CF_STATUS_COMMAND;
   sql_command_flags[SQLCOM_SHOW_FUNC_CODE]=  CF_STATUS_COMMAND;
   sql_command_flags[SQLCOM_SHOW_CREATE_EVENT]=  CF_STATUS_COMMAND;
@@ -270,9 +281,11 @@ void init_update_queries(void)
   sql_command_flags[SQLCOM_SHOW_PROFILE]= CF_STATUS_COMMAND;
 
    sql_command_flags[SQLCOM_SHOW_TABLES]=       (CF_STATUS_COMMAND |
-                                               CF_SHOW_TABLE_COMMAND);
+                                                 CF_SHOW_TABLE_COMMAND |
+                                                 CF_REEXECUTION_FRAGILE);
   sql_command_flags[SQLCOM_SHOW_TABLE_STATUS]= (CF_STATUS_COMMAND |
-                                                CF_SHOW_TABLE_COMMAND);
+                                                CF_SHOW_TABLE_COMMAND |
+                                                CF_REEXECUTION_FRAGILE);
 
   /*
     The following is used to preserver CF_ROW_COUNT during the
@@ -280,7 +293,7 @@ void init_update_queries(void)
     last called (or executed) statement is preserved.
     See mysql_execute_command() for how CF_ROW_COUNT is used.
   */
-  sql_command_flags[SQLCOM_CALL]= 		CF_HAS_ROW_COUNT;
+  sql_command_flags[SQLCOM_CALL]= 		CF_HAS_ROW_COUNT | CF_REEXECUTION_FRAGILE;
   sql_command_flags[SQLCOM_EXECUTE]= 		CF_HAS_ROW_COUNT;
 
   /*
@@ -4315,20 +4328,6 @@ create_sp_error:
       }
       break;
     }
-#ifdef NOT_USED
-  case SQLCOM_SHOW_STATUS_PROC:
-    {
-      res= sp_show_status_routine(thd, TYPE_ENUM_PROCEDURE,
-                                  (lex->wild ? lex->wild->ptr() : NullS));
-      break;
-    }
-  case SQLCOM_SHOW_STATUS_FUNC:
-    {
-      res= sp_show_status_routine(thd, TYPE_ENUM_FUNCTION,
-                                  (lex->wild ? lex->wild->ptr() : NullS));
-      break;
-    }
-#endif
 #ifndef DBUG_OFF
   case SQLCOM_SHOW_PROC_CODE:
   case SQLCOM_SHOW_FUNC_CODE:
diff --git a/sql/sql_prepare.cc b/sql/sql_prepare.cc
index 091cc18114b..3036b36f8fa 100644
--- a/sql/sql_prepare.cc
+++ b/sql/sql_prepare.cc
@@ -155,21 +155,28 @@ public:
   virtual void cleanup_stmt();
   bool set_name(LEX_STRING *name);
   inline void close_cursor() { delete cursor; cursor= 0; }
-
+  inline bool is_in_use() { return flags & (uint) IS_IN_USE; }
+  inline bool is_protocol_text() const { return protocol == &thd->protocol_text; }
   bool prepare(const char *packet, uint packet_length);
-  bool execute(String *expanded_query, bool open_cursor);
+  bool execute_loop(String *expanded_query,
+                    bool open_cursor,
+                    uchar *packet_arg, uchar *packet_end_arg);
   /* Destroy this statement */
-  bool deallocate();
+  void deallocate();
 private:
   /**
-    Store the parsed tree of a prepared statement here.
-  */
-  LEX main_lex;
-  /**
     The memory root to allocate parsed tree elements (instances of Item,
     SELECT_LEX and other classes).
   */
   MEM_ROOT main_mem_root;
+private:
+  bool set_db(const char *db, uint db_length);
+  bool set_parameters(String *expanded_query,
+                      uchar *packet, uchar *packet_end);
+  bool execute(String *expanded_query, bool open_cursor);
+  bool reprepare();
+  bool validate_metadata(Prepared_statement  *copy);
+  void swap_prepared_statement(Prepared_statement *copy);
 };
 
 
@@ -198,7 +205,7 @@ inline bool is_param_null(const uchar *pos, ulong param_no)
 */
 
 static Prepared_statement *
-find_prepared_statement(THD *thd, ulong id, const char *where)
+find_prepared_statement(THD *thd, ulong id)
 {
   /*
     To strictly separate namespaces of SQL prepared statements and C API
@@ -208,12 +215,8 @@ find_prepared_statement(THD *thd, ulong id, const char *where)
   Statement *stmt= thd->stmt_map.find(id);
 
   if (stmt == 0 || stmt->type() != Query_arena::PREPARED_STATEMENT)
-  {
-    char llbuf[22];
-    my_error(ER_UNKNOWN_STMT_HANDLER, MYF(0), sizeof(llbuf), llstr(id, llbuf),
-             where);
-    return 0;
-  }
+    return NULL;
+
   return (Prepared_statement *) stmt;
 }
 
@@ -945,6 +948,55 @@ static bool emb_insert_params_with_log(Prepared_statement *stmt,
 
 #endif /*!EMBEDDED_LIBRARY*/
 
+/**
+  Setup data conversion routines using an array of parameter
+  markers from the original prepared statement.
+  Swap the parameter data of the original prepared
+  statement to the new one.
+
+  Used only when we re-prepare a prepared statement.
+  There are two reasons for this function to exist:
+
+  1) In the binary client/server protocol, parameter metadata
+  is sent only at first execute. Consequently, if we need to
+  reprepare a prepared statement at a subsequent execution,
+  we may not have metadata information in the packet.
+  In that case we use the parameter array of the original
+  prepared statement to setup parameter types of the new
+  prepared statement.
+
+  2) In the binary client/server protocol, we may supply
+  long data in pieces. When the last piece is supplied,
+  we assemble the pieces and convert them from client
+  character set to the connection character set. After
+  that the parameter value is only available inside
+  the parameter, the original pieces are lost, and thus
+  we can only assign the corresponding parameter of the
+  reprepared statement from the original value.
+
+  @param[out]  param_array_dst  parameter markers of the new statement
+  @param[in]   param_array_src  parameter markers of the original
+                                statement
+  @param[in]   param_count      total number of parameters. Is the
+                                same in src and dst arrays, since
+                                the statement query is the same
+
+  @return this function never fails
+*/
+
+static void
+swap_parameter_array(Item_param **param_array_dst,
+                     Item_param **param_array_src,
+                     uint param_count)
+{
+  Item_param **dst= param_array_dst;
+  Item_param **src= param_array_src;
+  Item_param **end= param_array_dst + param_count;
+
+  for (; dst < end; ++src, ++dst)
+    (*dst)->set_param_type_and_swap_value(*src);
+}
+
 
 /**
   Assign prepared statement parameters from user variables.
@@ -1268,7 +1320,7 @@ error:
 */
 
 static int mysql_test_select(Prepared_statement *stmt,
-                             TABLE_LIST *tables, bool text_protocol)
+                             TABLE_LIST *tables)
 {
   THD *thd= stmt->thd;
   LEX *lex= stmt->lex;
@@ -1304,7 +1356,7 @@ static int mysql_test_select(Prepared_statement *stmt,
   */
   if (unit->prepare(thd, 0, 0))
     goto error;
-  if (!lex->describe && !text_protocol)
+  if (!lex->describe && !stmt->is_protocol_text())
   {
     /* Make copy of item list, as change_columns may change it */
     List<Item> fields(lex->select_lex.item_list);
@@ -1396,6 +1448,43 @@ error:
 
 
 /**
+  Validate and prepare for execution CALL statement expressions.
+
+  @param stmt               prepared statement
+  @param tables             list of tables used in this query
+  @param value_list         list of expressions
+
+  @retval FALSE             success
+  @retval TRUE              error, error message is set in THD
+*/
+
+static bool mysql_test_call_fields(Prepared_statement *stmt,
+                                   TABLE_LIST *tables,
+                                   List<Item> *value_list)
+{
+  DBUG_ENTER("mysql_test_call_fields");
+
+  List_iterator<Item> it(*value_list);
+  THD *thd= stmt->thd;
+  Item *item;
+
+  if (tables && check_table_access(thd, SELECT_ACL, tables, UINT_MAX, FALSE) ||
+      open_normal_and_derived_tables(thd, tables, 0))
+    goto err;
+
+  while ((item= it++))
+  {
+    if (!item->fixed && item->fix_fields(thd, it.ref()) ||
+        item->check_cols(1))
+      goto err;
+  }
+  DBUG_RETURN(FALSE);
+err:
+  DBUG_RETURN(TRUE);
+}
+
+
+/**
   Check internal SELECT of the prepared command.
 
   @param stmt                      prepared statement
@@ -1516,6 +1605,17 @@ static bool mysql_test_create_table(Prepared_statement *stmt)
 
     res= select_like_stmt_test(stmt, 0, 0);
   }
+  else if (lex->create_info.options & HA_LEX_CREATE_TABLE_LIKE)
+  {
+    /*
+      Check that the source table exist, and also record
+      its metadata version. Even though not strictly necessary,
+      we validate metadata of all CREATE TABLE statements,
+      which keeps metadata validation code simple.
+    */
+    if (open_normal_and_derived_tables(stmt->thd, lex->query_tables, 0))
+      DBUG_RETURN(TRUE);
+  }
 
   /* put tables back for PS rexecuting */
   lex->link_first_table_back(create_table, link_to_local);
@@ -1712,8 +1812,7 @@ static bool mysql_test_insert_select(Prepared_statement *stmt,
     TRUE              error, error message is set in THD (but not sent)
 */
 
-static bool check_prepared_statement(Prepared_statement *stmt,
-                                     bool text_protocol)
+static bool check_prepared_statement(Prepared_statement *stmt)
 {
   THD *thd= stmt->thd;
   LEX *lex= stmt->lex;
@@ -1754,9 +1853,23 @@ static bool check_prepared_statement(Prepared_statement *stmt,
   case SQLCOM_DELETE:
     res= mysql_test_delete(stmt, tables);
     break;
-
+  /* The following allow WHERE clause, so they must be tested like SELECT */
+  case SQLCOM_SHOW_DATABASES:
+  case SQLCOM_SHOW_TABLES:
+  case SQLCOM_SHOW_TRIGGERS:
+  case SQLCOM_SHOW_EVENTS:
+  case SQLCOM_SHOW_OPEN_TABLES:
+  case SQLCOM_SHOW_FIELDS:
+  case SQLCOM_SHOW_KEYS:
+  case SQLCOM_SHOW_COLLATIONS:
+  case SQLCOM_SHOW_CHARSETS:
+  case SQLCOM_SHOW_VARIABLES:
+  case SQLCOM_SHOW_STATUS:
+  case SQLCOM_SHOW_TABLE_STATUS:
+  case SQLCOM_SHOW_STATUS_PROC:
+  case SQLCOM_SHOW_STATUS_FUNC:
   case SQLCOM_SELECT:
-    res= mysql_test_select(stmt, tables, text_protocol);
+    res= mysql_test_select(stmt, tables);
     if (res == 2)
     {
       /* Statement and field info has already been sent */
@@ -1779,6 +1892,9 @@ static bool check_prepared_statement(Prepared_statement *stmt,
     res= mysql_test_do_fields(stmt, tables, lex->insert_list);
     break;
 
+  case SQLCOM_CALL:
+    res= mysql_test_call_fields(stmt, tables, &lex->value_list);
+    break;
   case SQLCOM_SET_OPTION:
     res= mysql_test_set_fields(stmt, tables, &lex->var_list);
     break;
@@ -1828,7 +1944,6 @@ static bool check_prepared_statement(Prepared_statement *stmt,
   case SQLCOM_DROP_INDEX:
   case SQLCOM_ROLLBACK:
   case SQLCOM_TRUNCATE:
-  case SQLCOM_CALL:
   case SQLCOM_DROP_VIEW:
   case SQLCOM_REPAIR:
   case SQLCOM_ANALYZE:
@@ -1871,8 +1986,8 @@ static bool check_prepared_statement(Prepared_statement *stmt,
     break;
   }
   if (res == 0)
-    DBUG_RETURN(text_protocol? FALSE : (send_prep_stmt(stmt, 0) ||
-                                        thd->protocol->flush()));
+    DBUG_RETURN(stmt->is_protocol_text() ?
+                FALSE : (send_prep_stmt(stmt, 0) || thd->protocol->flush()));
 error:
   DBUG_RETURN(TRUE);
 }
@@ -2121,8 +2236,13 @@ void mysql_sql_stmt_prepare(THD *thd)
       If there is a statement with the same name, remove it. It is ok to
       remove old and fail to insert a new one at the same time.
     */
-    if (stmt->deallocate())
+    if (stmt->is_in_use())
+    {
+      my_error(ER_PS_NO_RECURSION, MYF(0));
       DBUG_VOID_RETURN;
+    }
+
+    stmt->deallocate();
   }
 
   if (! (query= get_dynamic_sql_string(lex, &query_len)) ||
@@ -2308,11 +2428,9 @@ void mysql_stmt_execute(THD *thd, char *packet_arg, uint packet_length)
   ulong flags= (ulong) packet[4];
   /* Query text for binary, general or slow log, if any of them is open */
   String expanded_query;
-#ifndef EMBEDDED_LIBRARY
   uchar *packet_end= packet + packet_length;
-#endif
   Prepared_statement *stmt;
-  bool error;
+  bool open_cursor;
   DBUG_ENTER("mysql_stmt_execute");
 
   packet+= 9;                               /* stmt_id + 5 bytes of flags */
@@ -2320,8 +2438,13 @@ void mysql_stmt_execute(THD *thd, char *packet_arg, uint packet_length)
   /* First of all clear possible warnings from the previous command */
   mysql_reset_thd_for_next_command(thd);
 
-  if (!(stmt= find_prepared_statement(thd, stmt_id, "mysql_stmt_execute")))
+  if (!(stmt= find_prepared_statement(thd, stmt_id)))
+  {
+    char llbuf[22];
+    my_error(ER_UNKNOWN_STMT_HANDLER, MYF(0), sizeof(llbuf),
+             llstr(stmt_id, llbuf), "mysql_stmt_execute");
     DBUG_VOID_RETURN;
+  }
 
 #if defined(ENABLED_PROFILING) && defined(COMMUNITY_SERVER)
   thd->profiling.set_query_source(stmt->query, stmt->query_length);
@@ -2332,44 +2455,12 @@ void mysql_stmt_execute(THD *thd, char *packet_arg, uint packet_length)
   sp_cache_flush_obsolete(&thd->sp_proc_cache);
   sp_cache_flush_obsolete(&thd->sp_func_cache);
 
-#ifndef EMBEDDED_LIBRARY
-  if (stmt->param_count)
-  {
-    uchar *null_array= packet;
-    if (setup_conversion_functions(stmt, &packet, packet_end) ||
-        stmt->set_params(stmt, null_array, packet, packet_end,
-                         &expanded_query))
-      goto set_params_data_err;
-  }
-#else
-  /*
-    In embedded library we re-install conversion routines each time
-    we set params, and also we don't need to parse packet.
-    So we do it in one function.
-  */
-  if (stmt->param_count && stmt->set_params_data(stmt, &expanded_query))
-    goto set_params_data_err;
-#endif
-  if (!(specialflag & SPECIAL_NO_PRIOR))
-    my_pthread_setprio(pthread_self(),QUERY_PRIOR);
+  open_cursor= test(flags & (ulong) CURSOR_TYPE_READ_ONLY);
 
-  /*
-    If the free_list is not empty, we'll wrongly free some externally
-    allocated items when cleaning up after validation of the prepared
-    statement.
-  */
-  DBUG_ASSERT(thd->free_list == NULL);
+  stmt->execute_loop(&expanded_query, open_cursor, packet, packet_end);
 
-  error= stmt->execute(&expanded_query,
-                       test(flags & (ulong) CURSOR_TYPE_READ_ONLY));
-  if (!(specialflag & SPECIAL_NO_PRIOR))
-    my_pthread_setprio(pthread_self(), WAIT_PRIOR);
   DBUG_VOID_RETURN;
 
-set_params_data_err:
-  my_error(ER_WRONG_ARGUMENTS, MYF(0), "mysql_stmt_execute");
-  reset_stmt_params(stmt);
-  DBUG_VOID_RETURN;
 }
 
 
@@ -2415,24 +2506,8 @@ void mysql_sql_stmt_execute(THD *thd)
 
   DBUG_PRINT("info",("stmt: 0x%lx", (long) stmt));
 
-  /*
-    If the free_list is not empty, we'll wrongly free some externally
-    allocated items when cleaning up after validation of the prepared
-    statement.
-  */
-  DBUG_ASSERT(thd->free_list == NULL);
-
-  if (stmt->set_params_from_vars(stmt, lex->prepared_stmt_params,
-                                 &expanded_query))
-    goto set_params_data_err;
-
-  (void) stmt->execute(&expanded_query, FALSE);
-
-  DBUG_VOID_RETURN;
+  (void) stmt->execute_loop(&expanded_query, FALSE, NULL, NULL);
 
-set_params_data_err:
-  my_error(ER_WRONG_ARGUMENTS, MYF(0), "EXECUTE");
-  reset_stmt_params(stmt);
   DBUG_VOID_RETURN;
 }
 
@@ -2458,8 +2533,13 @@ void mysql_stmt_fetch(THD *thd, char *packet, uint packet_length)
   /* First of all clear possible warnings from the previous command */
   mysql_reset_thd_for_next_command(thd);
   status_var_increment(thd->status_var.com_stmt_fetch);
-  if (!(stmt= find_prepared_statement(thd, stmt_id, "mysql_stmt_fetch")))
+  if (!(stmt= find_prepared_statement(thd, stmt_id)))
+  {
+    char llbuf[22];
+    my_error(ER_UNKNOWN_STMT_HANDLER, MYF(0), sizeof(llbuf),
+             llstr(stmt_id, llbuf), "mysql_stmt_fetch");
     DBUG_VOID_RETURN;
+  }
 
   cursor= stmt->cursor;
   if (!cursor)
@@ -2520,8 +2600,13 @@ void mysql_stmt_reset(THD *thd, char *packet)
   mysql_reset_thd_for_next_command(thd);
 
   status_var_increment(thd->status_var.com_stmt_reset);
-  if (!(stmt= find_prepared_statement(thd, stmt_id, "mysql_stmt_reset")))
+  if (!(stmt= find_prepared_statement(thd, stmt_id)))
+  {
+    char llbuf[22];
+    my_error(ER_UNKNOWN_STMT_HANDLER, MYF(0), sizeof(llbuf),
+             llstr(stmt_id, llbuf), "mysql_stmt_reset");
     DBUG_VOID_RETURN;
+  }
 
   stmt->close_cursor();
 
@@ -2557,15 +2642,15 @@ void mysql_stmt_close(THD *thd, char *packet)
 
   thd->main_da.disable_status();
 
-  if (!(stmt= find_prepared_statement(thd, stmt_id, "mysql_stmt_close")))
+  if (!(stmt= find_prepared_statement(thd, stmt_id)))
     DBUG_VOID_RETURN;
 
   /*
     The only way currently a statement can be deallocated when it's
     in use is from within Dynamic SQL.
   */
-  DBUG_ASSERT(! (stmt->flags & (uint) Prepared_statement::IS_IN_USE));
-  (void) stmt->deallocate();
+  DBUG_ASSERT(! stmt->is_in_use());
+  stmt->deallocate();
   general_log_print(thd, thd->command, NullS);
 
   DBUG_VOID_RETURN;
@@ -2592,14 +2677,15 @@ void mysql_sql_stmt_close(THD *thd)
                       name->str));
 
   if (! (stmt= (Prepared_statement*) thd->stmt_map.find_by_name(name)))
-  {
     my_error(ER_UNKNOWN_STMT_HANDLER, MYF(0),
              name->length, name->str, "DEALLOCATE PREPARE");
-    return;
-  }
-
-  if (stmt->deallocate() == 0)
+  else if (stmt->is_in_use())
+    my_error(ER_PS_NO_RECURSION, MYF(0));
+  else
+  {
+    stmt->deallocate();
     my_ok(thd);
+  }
 }
 
 /**
@@ -2633,17 +2719,13 @@ void mysql_stmt_get_longdata(THD *thd, char *packet, ulong packet_length)
 #ifndef EMBEDDED_LIBRARY
   /* Minimal size of long data packet is 6 bytes */
   if (packet_length < MYSQL_LONG_DATA_HEADER)
-  {
-    my_error(ER_WRONG_ARGUMENTS, MYF(0), "mysql_stmt_send_long_data");
     DBUG_VOID_RETURN;
-  }
 #endif
 
   stmt_id= uint4korr(packet);
   packet+= 4;
 
-  if (!(stmt=find_prepared_statement(thd, stmt_id,
-                                     "mysql_stmt_send_long_data")))
+  if (!(stmt=find_prepared_statement(thd, stmt_id)))
     DBUG_VOID_RETURN;
 
   param_number= uint2korr(packet);
@@ -2729,7 +2811,7 @@ Select_fetch_protocol_binary::send_data(List<Item> &fields)
 ****************************************************************************/
 
 Prepared_statement::Prepared_statement(THD *thd_arg, Protocol *protocol_arg)
-  :Statement(&main_lex, &main_mem_root,
+  :Statement(NULL, &main_mem_root,
              INITIALIZED, ++thd_arg->statement_id_counter),
   thd(thd_arg),
   result(thd_arg),
@@ -2800,7 +2882,11 @@ Prepared_statement::~Prepared_statement()
     like Item_param, don't free everything until free_items()
   */
   free_items();
-  delete lex->result;
+  if (lex)
+  {
+    delete lex->result;
+    delete (st_lex_local *) lex;
+  }
   free_root(&main_mem_root, MYF(0));
   DBUG_VOID_RETURN;
 }
@@ -2836,6 +2922,34 @@ bool Prepared_statement::set_name(LEX_STRING *name_arg)
   return name.str == 0;
 }
 
+
+/**
+  Remember the current database.
+
+  We must reset/restore the current database during execution of
+  a prepared statement since it affects execution environment:
+  privileges, @@character_set_database, and other.
+
+  @return Returns an error if out of memory.
+*/
+
+bool
+Prepared_statement::set_db(const char *db_arg, uint db_length_arg)
+{
+  /* Remember the current database. */
+  if (db_arg && db_length_arg)
+  {
+    db= this->strmake(db_arg, db_length_arg);
+    db_length= db_length_arg;
+  }
+  else
+  {
+    db= NULL;
+    db_length= 0;
+  }
+  return db_arg != NULL && db == NULL;
+}
+
 /**************************************************************************
   Common parts of mysql_[sql]_stmt_prepare, mysql_[sql]_stmt_execute.
   Essentially, these functions do all the magic of preparing/executing
@@ -2877,6 +2991,12 @@ bool Prepared_statement::prepare(const char *packet, uint packet_len)
   */
   status_var_increment(thd->status_var.com_stmt_prepare);
 
+  if (! (lex= new (mem_root) st_lex_local))
+    DBUG_RETURN(TRUE);
+
+  if (set_db(thd->db, thd->db_length))
+    DBUG_RETURN(TRUE);
+
   /*
     alloc_query() uses thd->memroot && thd->query, so we should call
     both of backup_statement() and backup_query_arena() here.
@@ -2904,19 +3024,6 @@ bool Prepared_statement::prepare(const char *packet, uint packet_len)
 
   lex->set_trg_event_type_for_tables();
 
-  /* Remember the current database. */
-
-  if (thd->db && thd->db_length)
-  {
-    db= this->strmake(thd->db, thd->db_length);
-    db_length= thd->db_length;
-  }
-  else
-  {
-    db= NULL;
-    db_length= 0;
-  }
-
   /*
     While doing context analysis of the query (in check_prepared_statement)
     we allocate a lot of additional memory: for open tables, JOINs, derived
@@ -2940,7 +3047,7 @@ bool Prepared_statement::prepare(const char *packet, uint packet_len)
   */
 
   if (error == 0)
-    error= check_prepared_statement(this, name.str != 0);
+    error= check_prepared_statement(this);
 
   /*
     Currently CREATE PROCEDURE/TRIGGER/EVENT are prohibited in prepared
@@ -2987,6 +3094,306 @@ bool Prepared_statement::prepare(const char *packet, uint packet_len)
   DBUG_RETURN(error);
 }
 
+
+/**
+  Assign parameter values either from variables, in case of SQL PS
+  or from the execute packet.
+
+  @param expanded_query  a container with the original SQL statement.
+                         '?' placeholders will be replaced with
+                         their values in case of success.
+                         The result is used for logging and replication
+  @param packet          pointer to execute packet.
+                         NULL in case of SQL PS
+  @param packet_end      end of the packet. NULL in case of SQL PS
+
+  @todo Use a paremeter source class family instead of 'if's, and
+  support stored procedure variables.
+
+  @retval TRUE an error occurred when assigning a parameter (likely
+          a conversion error or out of memory, or malformed packet)
+  @retval FALSE success
+*/
+
+bool
+Prepared_statement::set_parameters(String *expanded_query,
+                                   uchar *packet, uchar *packet_end)
+{
+  bool is_sql_ps= packet == NULL;
+  bool res= FALSE;
+
+  if (is_sql_ps)
+  {
+    /* SQL prepared statement */
+    res= set_params_from_vars(this, thd->lex->prepared_stmt_params,
+                              expanded_query);
+  }
+  else if (param_count)
+  {
+#ifndef EMBEDDED_LIBRARY
+    uchar *null_array= packet;
+    res= (setup_conversion_functions(this, &packet, packet_end) ||
+          set_params(this, null_array, packet, packet_end, expanded_query));
+#else
+    /*
+      In embedded library we re-install conversion routines each time
+      we set parameters, and also we don't need to parse packet.
+      So we do it in one function.
+    */
+    res= set_params_data(this, expanded_query);
+#endif
+  }
+  if (res)
+  {
+    my_error(ER_WRONG_ARGUMENTS, MYF(0),
+             is_sql_ps ? "EXECUTE" : "mysql_stmt_execute");
+    reset_stmt_params(this);
+  }
+  return res;
+}
+
+
+/**
+  Execute a prepared statement. Re-prepare it a limited number
+  of times if necessary.
+
+  Try to execute a prepared statement. If there is a metadata
+  validation error, prepare a new copy of the prepared statement,
+  swap the old and the new statements, and try again.
+  If there is a validation error again, repeat the above, but
+  perform no more than MAX_REPREPARE_ATTEMPTS.
+
+  @note We have to try several times in a loop since we
+  release metadata locks on tables after prepared statement
+  prepare. Therefore, a DDL statement may sneak in between prepare
+  and execute of a new statement. If this happens repeatedly
+  more than MAX_REPREPARE_ATTEMPTS times, we give up.
+
+  In future we need to be able to keep the metadata locks between
+  prepare and execute, but right now open_and_lock_tables(), as
+  well as close_thread_tables() are buried deep inside
+  execution code (mysql_execute_command()).
+
+  @return TRUE if an error, FALSE if success
+  @retval  TRUE    either MAX_REPREPARE_ATTEMPTS has been reached,
+                   or some general error
+  @retval  FALSE   successfully executed the statement, perhaps
+                   after having reprepared it a few times.
+*/
+
+bool
+Prepared_statement::execute_loop(String *expanded_query,
+                                 bool open_cursor,
+                                 uchar *packet,
+                                 uchar *packet_end)
+{
+  const int MAX_REPREPARE_ATTEMPTS= 3;
+  Reprepare_observer reprepare_observer;
+  bool error;
+  int reprepare_attempt= 0;
+
+  if (set_parameters(expanded_query, packet, packet_end))
+    return TRUE;
+
+reexecute:
+  reprepare_observer.reset_reprepare_observer();
+
+  /*
+    If the free_list is not empty, we'll wrongly free some externally
+    allocated items when cleaning up after validation of the prepared
+    statement.
+  */
+  DBUG_ASSERT(thd->free_list == NULL);
+
+  /*
+    Install the metadata observer. If some metadata version is
+    different from prepare time and an observer is installed,
+    the observer method will be invoked to push an error into
+    the error stack.
+  */
+  if (sql_command_flags[lex->sql_command] &
+      CF_REEXECUTION_FRAGILE)
+  {
+    DBUG_ASSERT(thd->m_reprepare_observer == NULL);
+    thd->m_reprepare_observer = &reprepare_observer;
+  }
+
+  if (!(specialflag & SPECIAL_NO_PRIOR))
+    my_pthread_setprio(pthread_self(),QUERY_PRIOR);
+
+  error= execute(expanded_query, open_cursor) || thd->is_error();
+
+  if (!(specialflag & SPECIAL_NO_PRIOR))
+    my_pthread_setprio(pthread_self(), WAIT_PRIOR);
+
+  thd->m_reprepare_observer= NULL;
+
+  if (error && !thd->is_fatal_error && !thd->killed &&
+      reprepare_observer.is_invalidated() &&
+      reprepare_attempt++ < MAX_REPREPARE_ATTEMPTS)
+  {
+    DBUG_ASSERT(thd->main_da.sql_errno() == ER_NEED_REPREPARE);
+    thd->clear_error();
+
+    error= reprepare();
+
+    if (! error)                                /* Success */
+      goto reexecute;
+  }
+  reset_stmt_params(this);
+
+  return error;
+}
+
+
+/**
+  Reprepare this prepared statement.
+
+  Currently this is implemented by creating a new prepared
+  statement, preparing it with the original query and then
+  swapping the new statement and the original one.
+
+  @retval  TRUE   an error occurred. Possible errors include
+                  incompatibility of new and old result set
+                  metadata
+  @retval  FALSE  success, the statement has been reprepared
+*/
+
+bool
+Prepared_statement::reprepare()
+{
+  char saved_cur_db_name_buf[NAME_LEN+1];
+  LEX_STRING saved_cur_db_name=
+    { saved_cur_db_name_buf, sizeof(saved_cur_db_name_buf) };
+  LEX_STRING stmt_db_name= { db, db_length };
+  bool cur_db_changed;
+  bool error;
+
+  Prepared_statement copy(thd, &thd->protocol_text);
+
+  status_var_increment(thd->status_var.com_stmt_reprepare);
+
+  if (mysql_opt_change_db(thd, &stmt_db_name, &saved_cur_db_name, TRUE,
+                          &cur_db_changed))
+    return TRUE;
+
+  error= (name.str && copy.set_name(&name) ||
+          copy.prepare(query, query_length) ||
+          validate_metadata(&copy));
+
+  if (cur_db_changed)
+    mysql_change_db(thd, &saved_cur_db_name, TRUE);
+
+  if (! error)
+  {
+    swap_prepared_statement(&copy);
+    swap_parameter_array(param_array, copy.param_array, param_count);
+#ifndef DBUG_OFF
+    is_reprepared= TRUE;
+#endif
+    /*
+      Clear possible warnings during reprepare, it has to be completely
+      transparent to the user. We use mysql_reset_errors() since
+      there were no separate query id issued for re-prepare.
+      Sic: we can't simply silence warnings during reprepare, because if
+      it's failed, we need to return all the warnings to the user.
+    */
+    mysql_reset_errors(thd, TRUE);
+  }
+  return error;
+}
+
+
+/**
+  Validate statement result set metadata (if the statement returns
+  a result set).
+
+  Currently we only check that the number of columns of the result
+  set did not change.
+  This is a helper method used during re-prepare.
+
+  @param[in]  copy  the re-prepared prepared statement to verify
+                    the metadata of
+
+  @retval TRUE  error, ER_PS_NEED_REBIND is reported
+  @retval FALSE statement return no or compatible metadata
+*/
+
+
+bool Prepared_statement::validate_metadata(Prepared_statement *copy)
+{
+  /**
+    If this is an SQL prepared statement or EXPLAIN,
+    return FALSE -- the metadata of the original SELECT,
+    if any, has not been sent to the client.
+  */
+  if (is_protocol_text() || lex->describe)
+    return FALSE;
+
+  if (lex->select_lex.item_list.elements !=
+      copy->lex->select_lex.item_list.elements)
+  {
+    /** Column counts mismatch. */
+    my_error(ER_PS_REBIND, MYF(0));
+    return TRUE;
+  }
+
+  return FALSE;
+}
+
+
+/**
+  Replace the original prepared statement with a prepared copy.
+
+  This is a private helper that is used as part of statement
+  reprepare
+
+  @return This function does not return any errors.
+*/
+
+void
+Prepared_statement::swap_prepared_statement(Prepared_statement *copy)
+{
+  Statement tmp_stmt;
+
+  /* Swap memory roots. */
+  swap_variables(MEM_ROOT, main_mem_root, copy->main_mem_root);
+
+  /* Swap the arenas */
+  tmp_stmt.set_query_arena(this);
+  set_query_arena(copy);
+  copy->set_query_arena(&tmp_stmt);
+
+  /* Swap the statement parent classes */
+  tmp_stmt.set_statement(this);
+  set_statement(copy);
+  copy->set_statement(&tmp_stmt);
+
+  /* Swap ids back, we need the original id */
+  swap_variables(ulong, id, copy->id);
+  /* Swap mem_roots back, they must continue pointing at the main_mem_roots */
+  swap_variables(MEM_ROOT *, mem_root, copy->mem_root);
+  /*
+    Swap the old and the new parameters array. The old array
+    is allocated in the old arena.
+  */
+  swap_variables(Item_param **, param_array, copy->param_array);
+  /* Swap flags: this is perhaps unnecessary */
+  swap_variables(uint, flags, copy->flags);
+  /* Swap names, the old name is allocated in the wrong memory root */
+  swap_variables(LEX_STRING, name, copy->name);
+  /* Ditto */
+  swap_variables(char *, db, copy->db);
+
+  DBUG_ASSERT(db_length == copy->db_length);
+  DBUG_ASSERT(param_count == copy->param_count);
+  DBUG_ASSERT(thd == copy->thd);
+  last_error[0]= '\0';
+  last_errno= 0;
+  /* Do not swap protocols, the copy always has protocol_text */
+}
+
+
 /**
   Execute a prepared statement.
 
@@ -3149,10 +3556,7 @@ bool Prepared_statement::execute(String *expanded_query, bool open_cursor)
   DBUG_ASSERT(! (error && cursor));
 
   if (! cursor)
-  {
     cleanup_stmt();
-    reset_stmt_params(this);
-  }
 
   thd->set_statement(&stmt_backup);
   thd->stmt_arena= old_stmt_arena;
@@ -3186,16 +3590,10 @@ error:
 
 /** Common part of DEALLOCATE PREPARE and mysql_stmt_close. */
 
-bool Prepared_statement::deallocate()
+void Prepared_statement::deallocate()
 {
   /* We account deallocate in the same manner as mysql_stmt_close */
   status_var_increment(thd->status_var.com_stmt_close);
-  if (flags & (uint) IS_IN_USE)
-  {
-    my_error(ER_PS_NO_RECURSION, MYF(0));
-    return TRUE;
-  }
   /* Statement map calls delete stmt on erase */
   thd->stmt_map.erase(this);
-  return FALSE;
 }
diff --git a/sql/sql_table.cc b/sql/sql_table.cc
index 3ef510b8d89..7c847e0817c 100644
--- a/sql/sql_table.cc
+++ b/sql/sql_table.cc
@@ -4783,9 +4783,6 @@ bool mysql_create_like_table(THD* thd, TABLE_LIST* table, TABLE_LIST* src_table,
   DBUG_ENTER("mysql_create_like_table");
 
 
-  /* CREATE TABLE ... LIKE is not allowed for views. */
-  src_table->required_type= FRMTYPE_TABLE;
-
   /*
     By opening source table we guarantee that it exists and no concurrent
     DDL operation will mess with it. Later we also take an exclusive
diff --git a/sql/sql_update.cc b/sql/sql_update.cc
index 9d6a2b1b8c8..bfb71db22da 100644
--- a/sql/sql_update.cc
+++ b/sql/sql_update.cc
@@ -853,8 +853,9 @@ bool mysql_prepare_update(THD *thd, TABLE_LIST *table_list,
 			 Item **conds, uint order_num, ORDER *order)
 {
   Item *fake_conds= 0;
+#ifndef NO_EMBEDDED_ACCESS_CHECKS
   TABLE *table= table_list->table;
-  TABLE_LIST tables;
+#endif
   List<Item> all_fields;
   SELECT_LEX *select_lex= &thd->lex->select_lex;
   DBUG_ENTER("mysql_prepare_update");
@@ -878,9 +879,6 @@ bool mysql_prepare_update(THD *thd, TABLE_LIST *table_list,
   table_list->register_want_access(SELECT_ACL);
 #endif
 
-  bzero((char*) &tables,sizeof(tables));	// For ORDER BY
-  tables.table= table;
-  tables.alias= table_list->alias;
   thd->lex->allow_sum_func= 0;
 
   if (setup_tables_and_check_access(thd, &select_lex->context, 
diff --git a/sql/sql_yacc.yy b/sql/sql_yacc.yy
index 25549936f5a..54399c3791c 100644
--- a/sql/sql_yacc.yy
+++ b/sql/sql_yacc.yy
@@ -3582,20 +3582,30 @@ create2:
         | LIKE table_ident
           {
             THD *thd= YYTHD;
+            TABLE_LIST *src_table;
             LEX *lex= thd->lex;
 
             lex->create_info.options|= HA_LEX_CREATE_TABLE_LIKE;
-            if (!lex->select_lex.add_table_to_list(thd, $2, NULL, 0, TL_READ))
+            src_table= lex->select_lex.add_table_to_list(thd, $2, NULL, 0,
+                                                         TL_READ);
+            if (! src_table)
               MYSQL_YYABORT;
+            /* CREATE TABLE ... LIKE is not allowed for views. */
+            src_table->required_type= FRMTYPE_TABLE;
           }
         | '(' LIKE table_ident ')'
           {
             THD *thd= YYTHD;
+            TABLE_LIST *src_table;
             LEX *lex= thd->lex;
 
             lex->create_info.options|= HA_LEX_CREATE_TABLE_LIKE;
-            if (!lex->select_lex.add_table_to_list(thd, $3, NULL, 0, TL_READ))
+            src_table= lex->select_lex.add_table_to_list(thd, $3, NULL, 0,
+                                                         TL_READ);
+            if (! src_table)
               MYSQL_YYABORT;
+            /* CREATE TABLE ... LIKE is not allowed for views. */
+            src_table->required_type= FRMTYPE_TABLE;
           }
         ;
 
diff --git a/sql/table.h b/sql/table.h
index 0e0a35b022d..75ddaf69c10 100644
--- a/sql/table.h
+++ b/sql/table.h
@@ -438,6 +438,105 @@ typedef struct st_table_share
     return table_map_id;
   }
 
+  /**
+    Convert unrelated members of TABLE_SHARE to one enum
+    representing its type.
+
+    @todo perhaps we need to have a member instead of a function.
+  */
+  enum enum_table_ref_type get_table_ref_type() const
+  {
+    if (is_view)
+      return TABLE_REF_VIEW;
+    switch (tmp_table) {
+    case NO_TMP_TABLE:
+      return TABLE_REF_BASE_TABLE;
+    case SYSTEM_TMP_TABLE:
+      return TABLE_REF_I_S_TABLE;
+    default:
+      return TABLE_REF_TMP_TABLE;
+    }
+  }
+  /**
+    Return a table metadata version.
+     * for base tables, we return table_map_id.
+       It is assigned from a global counter incremented for each
+       new table loaded into the table definition cache (TDC).
+     * for temporary tables it's table_map_id again. But for
+       temporary tables table_map_id is assigned from
+       thd->query_id. The latter is assigned from a thread local
+       counter incremented for every new SQL statement. Since
+       temporary tables are thread-local, each temporary table
+       gets a unique id.
+     * for everything else (views, information schema tables),
+       the version id is zero.
+
+   This choice of version id is a large compromise
+   to have a working prepared statement validation in 5.1. In
+   future version ids will be persistent, as described in WL#4180.
+
+   Let's try to explain why and how this limited solution allows
+   to validate prepared statements.
+
+   Firstly, sets (in mathematical sense) of version numbers
+   never intersect for different table types. Therefore,
+   version id of a temporary table is never compared with
+   a version id of a view, and vice versa.
+
+   Secondly, for base tables, we know that each DDL flushes the
+   respective share from the TDC. This ensures that whenever
+   a table is altered or dropped and recreated, it gets a new
+   version id.
+   Unfortunately, since elements of the TDC are also flushed on
+   LRU basis, this choice of version ids leads to false positives.
+   E.g. when the TDC size is too small, we may have a SELECT
+   * FROM INFORMATION_SCHEMA.TABLES flush all its elements, which
+   in turn will lead to a validation error and a subsequent
+   reprepare of all prepared statements.  This is
+   considered acceptable, since as long as prepared statements are
+   automatically reprepared, spurious invalidation is only
+   a performance hit. Besides, no better simple solution exists.
+
+   For temporary tables, using thd->query_id ensures that if
+   a temporary table was altered or recreated, a new version id is
+   assigned. This suits validation needs very well and will perhaps
+   never change.
+
+   Metadata of information schema tables never changes.
+   Thus we can safely assume 0 for a good enough version id.
+
+   Views are a special and tricky case. A view is always inlined
+   into the parse tree of a prepared statement at prepare.
+   Thus, when we execute a prepared statement, the parse tree
+   will not get modified even if the view is replaced with another
+   view.  Therefore, we can safely choose 0 for version id of
+   views and effectively never invalidate a prepared statement
+   when a view definition is altered. Note, that this leads to
+   wrong binary log in statement-based replication, since we log
+   prepared statement execution in form Query_log_events
+   containing conventional statements. But since there is no
+   metadata locking for views, the very same problem exists for
+   conventional statements alone, as reported in Bug#25144. The only
+   difference between prepared and conventional execution is,
+   effectively, that for prepared statements the race condition
+   window is much wider.
+   In 6.0 we plan to support view metadata locking (WL#3726) and
+   extend table definition cache to cache views (WL#4298).
+   When this is done, views will be handled in the same fashion
+   as the base tables.
+
+   Finally, by taking into account table type, we always
+   track that a change has taken place when a view is replaced
+   with a base table, a base table is replaced with a temporary
+   table and so on.
+
+   @sa TABLE_LIST::is_table_ref_id_equal()
+  */
+  ulong get_table_ref_version() const
+  {
+    return (tmp_table == SYSTEM_TMP_TABLE || is_view) ? 0 : table_map_id;
+  }
+
 } TABLE_SHARE;
 
 
@@ -1233,6 +1332,33 @@ struct TABLE_LIST
     child_def_version= ~0UL;
   }
 
+  /**
+    Compare the version of metadata from the previous execution
+    (if any) with values obtained from the current table
+    definition cache element.
+
+    @sa check_and_update_table_version()
+  */
+  inline
+  bool is_table_ref_id_equal(TABLE_SHARE *s) const
+  {
+    return (m_table_ref_type == s->get_table_ref_type() &&
+            m_table_ref_version == s->get_table_ref_version());
+  }
+
+  /**
+    Record the value of metadata version of the corresponding
+    table definition cache element in this parse tree node.
+
+    @sa check_and_update_table_version()
+  */
+  inline
+  void set_table_ref_id(TABLE_SHARE *s)
+  {
+    m_table_ref_type= s->get_table_ref_type();
+    m_table_ref_version= s->get_table_ref_version();
+  }
+
 private:
   bool prep_check_option(THD *thd, uint8 check_opt_type);
   bool prep_where(THD *thd, Item **conds, bool no_where_clause);
@@ -1243,6 +1369,10 @@ private:
 
   /* Remembered MERGE child def version.  See top comment in ha_myisammrg.cc */
   ulong         child_def_version;
+  /** See comments for set_metadata_id() */
+  enum enum_table_ref_type m_table_ref_type;
+  /** See comments for set_metadata_id() */
+  ulong m_table_ref_version;
 };
 
 class Item;