Tentative implementation of

WL#4165 Prepared statements: validation 
WL#4166 Prepared statements: automatic re-prepare
Fixes
Bug#27430 Crash in subquery code when in PS and table DDL changed after PREPARE
Bug#27690 Re-execution of prepared statement after table was replaced with a view crashes
Bug#27420 A combination of PS and view operations cause error + assertion on shutdown

The basic idea of the patch is to keep track of table metadata between
prepared statement prepare and execute. If some table used in the statement
has changed, the prepared statement is re-prepared before execution.

See WL#4165 and WL#4166 contents and comments in the code for details
of the implementation.


include/my_global.h:
  Remove 'register' keyword to avoid warnings when swapping large structures
  that don't fit into a register. Any modern compiler is capable of placing
  a variable in a register when that would benefit performance.
mysql-test/r/ps_1general.result:
  Update test results: since now we re-prepare automatically,
  more correct results are produced in prepare-ddl-execute scenario.
mysql-test/r/query_cache_merge.result:
  Ensure that the table definition cache is large enough for
  the test to pass in --ps-protocol
mysql-test/r/trigger.result:
  Update test results to reflect automatic statement reprepare.
mysql-test/t/disabled.def:
  Enable ps_ddl.test, which now passes.
mysql-test/t/ps_1general.test:
  Since now we re-execute prepared statements after DDL successfully,
  change the test to produce repeatable results. Remove expectancy of
  an error in one place where now we automatically reprepare the prepared
  statement.
mysql-test/t/query_cache_merge.test:
  Ensure the table definition cache is large enough for the test to pass
  in --ps-protocol
mysql-test/t/trigger.test:
  Sinc
sql/item.cc:
  Implement Item_param "copy" functionality, used at re-prepare of
  a prepared statement.
  We copy the type of the original parameter, and move the assigned value,
  if any. Sic, the value is "moved", since it can be quite big --
  e.g. in case we deal with a LONG DATA parameter.
  It's essential to move the value from the old parameter since
  at the time of re-prepare the client packet with the necessary information
  may be not available.
sql/item.h:
  Declare a new method used for reprepare.
sql/my_decimal.h:
  Implement "swap()" functionality of class my_decimal to be
  able to easily swap two decimal values.
sql/mysql_priv.h:
  Declare enum_metadata_type.
sql/mysqld.cc:
  Implement a status variable for the number of reprepared statements.
sql/sql_base.cc:
  Implement metadata version validation.
sql/share/errmsg.txt:
  Add two new error messages: ER_NEED_REPREPARE and ER_PS_REBIND.
  The first error (theoretically) never reaches the user.
  It is issued by the metadata validation framework when a metadata version
  has changed between prepare and execute. Later on it's intercepted
  and the statement is automatically re-prepared. Only if the error
  has occurred repeatedly MAX_REPREPARE_ATTEMTS (3) times do we
  return it to the user.
  
  The second error is issued when after re-prepare we discover
  that the metadata we sent over to the client using the binary
  protocol differs drammatically from the new result set metadata 
  that the reprepared statement produces (e.g. number of result
  set columns is different).
sql/sql_class.cc:
  Implement metadata version validation framework.
sql/sql_class.h:
  Declarations for metadata version validation framework.
sql/sql_parse.cc:
  Mark commands for which we must invalidate and reprepare a prepared
  statement when metadata has changed.
sql/sql_prepare.cc:
  Implement WL#4165 and WL#4166 (limited support of metadata validation
  and re-prepare).
sql/table.h:
  Implement metadata validation.
tests/mysql_client_test.c:
  Add a test case for WL#4166

12 files changed, 882 insertions, 98 deletions

diff --git a/sql/item.cc b/sql/item.cc
index 883c293f645..04c75107d14 100644
--- a/sql/item.cc
+++ b/sql/item.cc
@@ -3161,6 +3161,45 @@ void Item_param::print(String *str, enum_query_type query_type)
 }
 
 
+/**
+  Preserve the original parameter types and values
+  when re-preparing a prepared statement.
+
+  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.
+
+  @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.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 1885036f42b..b096abaf8ae 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 26f253a6f8e..491a4c8ca1b 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_metadata_version()
+  @sa Execute_observer
+  @sa Prepared_statement::reprepare()
+*/
+
+enum enum_metadata_type
+{
+  /** Initial value set by the parser */
+  METADATA_NULL= 0,
+  METADATA_VIEW,
+  METADATA_BASE_TABLE,
+  METADATA_I_S_TABLE,
+  METADATA_TMP_TABLE
+};
+
 /*
   External variables
 */
diff --git a/sql/mysqld.cc b/sql/mysqld.cc
index 36a5b1a10d9..d93100a068f 100644
--- a/sql/mysqld.cc
+++ b/sql/mysqld.cc
@@ -3089,6 +3089,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},
@@ -3177,7 +3178,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_ ?)
 
@@ -3186,6 +3187,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
 
@@ -3194,7 +3196,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);
@@ -6757,7 +6759,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/share/errmsg.txt b/sql/share/errmsg.txt
index 7990baa6bb7..684d1ce9421 100644
--- a/sql/share/errmsg.txt
+++ b/sql/share/errmsg.txt
@@ -6120,3 +6120,9 @@ 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_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/sql_base.cc b/sql/sql_base.cc
index f0faf034027..62a723953ed 100644
--- a/sql/sql_base.cc
+++ b/sql/sql_base.cc
@@ -3717,6 +3717,72 @@ 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.
+
+  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_metadata_version_equal()
+  @sa TABLE_SHARE::get_metadata_version()
+
+  @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_metadata_version_equal(table_share))
+  {
+    if (thd->m_metadata_observer &&
+        thd->m_metadata_observer->check_metadata_change(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.
+      */
+      return TRUE;
+    }
+    /* Always maintain the latest version */
+    tables->set_metadata_version(table_share);
+  }
+#if 0
+#ifndef DBUG_OFF
+  /* Spuriously reprepare each statement. */
+  if (thd->m_metadata_observer && thd->stmt_arena->is_reprepared == FALSE)
+  {
+    thd->m_metadata_observer->check_metadata_change(thd);
+    return TRUE;
+  }
+#endif
+#endif
+  return FALSE;
+}
+
 /*
   Load a table definition from file and open unireg table
 
@@ -3762,6 +3828,8 @@ retry:
 
   if (share->is_view)
   {
+    if (check_and_update_table_version(thd, table_list, share))
+      goto err;
     if (table_list->i_s_requested_object &  OPEN_TABLE_ONLY)
       goto err;
 
@@ -3779,6 +3847,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_metadata_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;
@@ -4375,8 +4463,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)++;
@@ -4524,6 +4622,12 @@ int open_tables(THD *thd, TABLE_LIST **start, uint *counter, uint flags)
     }
     tables->table->grant= tables->grant;
 
+    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),
@@ -4582,7 +4686,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 7c57e3fc40e..6a4ef0781a2 100644
--- a/sql/sql_class.cc
+++ b/sql/sql_class.cc
@@ -359,6 +359,10 @@ char *thd_security_context(THD *thd, char *buffer, unsigned int length,
   return thd->strmake(str.ptr(), str.length());
 }
 
+Metadata_version_observer::~Metadata_version_observer()
+{
+}
+
 /**
   Clear this diagnostics area. 
 
@@ -2767,7 +2771,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_metadata_observer == NULL);
   set_open_tables_state(backup);
   DBUG_VOID_RETURN;
 }
@@ -2871,6 +2876,7 @@ void THD::reset_sub_statement_state(Sub_statement_state *backup,
     first_successful_insert_id_in_prev_stmt;
   backup->first_successful_insert_id_in_cur_stmt= 
     first_successful_insert_id_in_cur_stmt;
+  backup->m_metadata_observer= m_metadata_observer;
 
   if ((!lex->requires_prelocking() || is_update_query(lex->sql_command)) &&
       !current_stmt_binlog_row_based)
@@ -2890,6 +2896,7 @@ void THD::reset_sub_statement_state(Sub_statement_state *backup,
   cuted_fields= 0;
   transaction.savepoints= 0;
   first_successful_insert_id_in_cur_stmt= 0;
+  m_metadata_observer= 0;
 }
 
 
@@ -2938,6 +2945,7 @@ void THD::restore_sub_statement_state(Sub_statement_state *backup)
   */
   examined_row_count+= backup->examined_row_count;
   cuted_fields+=       backup->cuted_fields;
+  m_metadata_observer= backup->m_metadata_observer;
 }
 
 
diff --git a/sql/sql_class.h b/sql/sql_class.h
index 2bfcc43a454..837b74b3b60 100644
--- a/sql/sql_class.h
+++ b/sql/sql_class.h
@@ -23,6 +23,53 @@
 #include "log.h"
 #include "rpl_tblmap.h"
 
+/**
+  An abstract interface that can be 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 abstract interface (a method) that is
+  invoked when such a situation takes place.
+  The implementation of the interface in most cases 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_metadata_observer is not NULL.
+
+  @sa check_and_update_metadata_version() for details of the
+  version tracking algorithm 
+
+  @sa Execute_observer for details of how we detect that
+  a metadata change is fatal and a re-prepare is necessary
+
+  @sa Open_tables_state::m_metadata_observer for the life cycle
+  of metadata observers.
+*/
+
+class Metadata_version_observer
+{
+protected:
+  virtual ~Metadata_version_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.
+  */
+  virtual bool check_metadata_change(THD *thd)= 0;
+};
+
+
 class Relay_log_info;
 
 class Query_log_event;
@@ -406,6 +453,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 +484,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 +500,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 +838,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_metadata_version()
+  */
+  Metadata_version_observer *m_metadata_observer;
+
+  /**
     List of regular tables in use by this thread. Contains temporary and
     base tables that were opened with @see open_tables().
   */
@@ -891,6 +954,7 @@ public:
     extra_lock= lock= locked_tables= 0;
     prelocked_mode= NON_PRELOCKED;
     state_flags= 0U;
+    m_metadata_observer= NULL;
   }
 };
 
@@ -919,6 +983,25 @@ public:
   bool enable_slow_log;
   bool last_insert_id_used;
   SAVEPOINT *savepoints;
+  /**
+    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.
+  */
+  Metadata_version_observer *m_metadata_observer;
 };
 
 
@@ -2777,6 +2860,7 @@ public:
 #define CF_STATUS_COMMAND	4
 #define CF_SHOW_TABLE_COMMAND	8
 #define CF_WRITE_LOGS_COMMAND  16
+#define CF_REEXECUTION_FRAGILE 32
 
 /* Functions in sql_class.cc */
 
diff --git a/sql/sql_parse.cc b/sql/sql_parse.cc
index c28fea57c64..faf48b38fd0 100644
--- a/sql/sql_parse.cc
+++ b/sql/sql_parse.cc
@@ -218,14 +218,23 @@ void init_update_queries(void)
   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_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_SHOW_STATUS_PROC]= CF_STATUS_COMMAND;
   sql_command_flags[SQLCOM_SHOW_STATUS]=      CF_STATUS_COMMAND;
diff --git a/sql/sql_prepare.cc b/sql/sql_prepare.cc
index c221e29e1d0..7550226adff 100644
--- a/sql/sql_prepare.cc
+++ b/sql/sql_prepare.cc
@@ -116,6 +116,39 @@ public:
 #endif
 };
 
+/**
+  If a metadata changed, report a respective error to trigger
+  re-prepare of a prepared statement.
+*/
+
+class Execute_observer: public Metadata_version_observer
+{
+public:
+  virtual bool check_metadata_change(THD *thd);
+  /** Set to TRUE if metadata of some used table has changed since prepare */
+  bool m_invalidated;
+};
+
+/**
+  Push an error to the error stack and return TRUE for now.
+  In future we must take special care of statements like CREATE
+  TABLE ... SELECT.  Should we re-prepare such statements every
+  time?
+*/
+
+bool
+Execute_observer::check_metadata_change(THD *thd)
+{
+  bool save_thd_no_warnings_for_error= thd->no_warnings_for_error;
+  DBUG_ENTER("Execute_observer::notify_about_metadata_change");
+
+  thd->no_warnings_for_error= TRUE;
+  my_error(ER_NEED_REPREPARE, MYF(0));
+  thd->no_warnings_for_error= save_thd_no_warnings_for_error;
+  m_invalidated= TRUE;
+  DBUG_RETURN(TRUE);
+}
+
 /****************************************************************************/
 
 /**
@@ -156,20 +189,27 @@ public:
   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 */
   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);
 };
 
 
@@ -941,6 +981,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.
+  Move 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.
@@ -1264,7 +1353,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;
@@ -1300,7 +1389,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);
@@ -1708,8 +1797,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;
@@ -1752,7 +1840,7 @@ static bool check_prepared_statement(Prepared_statement *stmt,
     break;
 
   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 */
@@ -1867,8 +1955,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);
 }
@@ -2309,11 +2397,10 @@ 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 */
@@ -2338,44 +2425,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;
 }
 
 
@@ -2421,24 +2476,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;
 }
 
@@ -2742,7 +2781,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(0, &main_mem_root,
              INITIALIZED, ++thd_arg->statement_id_counter),
   thd(thd_arg),
   result(thd_arg),
@@ -2813,7 +2852,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;
 }
@@ -2849,6 +2892,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
@@ -2890,6 +2961,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.
@@ -2917,19 +2994,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
@@ -2953,7 +3017,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
@@ -3000,6 +3064,293 @@ bool Prepared_statement::prepare(const char *packet, uint packet_len)
   DBUG_RETURN(error);
 }
 
+
+bool
+Prepared_statement::set_parameters(String *expanded_query,
+                                   uchar *packet, uchar *packet_end)
+{
+  bool is_sql_ps= packet == NULL;
+
+  if (is_sql_ps)
+  {
+    /* SQL prepared statement */
+    if (set_params_from_vars(this, thd->lex->prepared_stmt_params,
+                                   expanded_query))
+      goto set_params_data_err;
+  }
+  else if (param_count)
+  {
+#ifndef EMBEDDED_LIBRARY
+    uchar *null_array= packet;
+    if (setup_conversion_functions(this, &packet, packet_end) ||
+        set_params(this, 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 (set_params_data(this, expanded_query))
+      goto set_params_data_err;
+#endif
+  }
+  return FALSE;
+set_params_data_err:
+  my_error(ER_WRONG_ARGUMENTS, MYF(0),
+           is_sql_ps ? "EXECUTE" : "mysql_stmt_execute");
+  reset_stmt_params(this);
+  return TRUE;
+}
+
+
+/**
+  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;
+  Execute_observer execute_observer;
+  bool error;
+  int reprepare_attempt= 0;
+
+  if (set_parameters(expanded_query, packet, packet_end))
+    return TRUE;
+
+reexecute:
+  execute_observer.m_invalidated= FALSE;
+
+  /*
+    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)
+  {
+    thd->m_metadata_observer= &execute_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_metadata_observer= 0;
+
+  if (error && !thd->is_fatal_error && !thd->killed &&
+      execute_observer.m_invalidated &&
+      reprepare_attempt++ < MAX_REPREPARE_ATTEMPTS)
+  {
+    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 };
+  Prepared_statement *copy;
+  bool cur_db_changed;
+  bool error= TRUE;
+
+  status_var_increment(thd->status_var.com_stmt_reprepare);
+
+  copy= new Prepared_statement(thd, &thd->protocol_text);
+
+  if (! copy)
+    return TRUE;
+
+  if (mysql_opt_change_db(thd, &stmt_db_name, &saved_cur_db_name, TRUE,
+                          &cur_db_changed))
+    goto end;
+
+  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);
+    is_reprepared= TRUE;
+    /*
+      Clear possible warnigns during re-prepare, 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.
+    */
+    mysql_reset_errors(thd, TRUE);
+  }
+end:
+  delete copy;
+  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)
+{
+  List_iterator_fast<Item> it_org(lex->select_lex.item_list);
+  List_iterator_fast<Item> it_new(copy->lex->select_lex.item_list);
+  Item *item_org;
+  Item *item_new;
+
+  /**
+    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.
 
@@ -3162,10 +3513,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;
diff --git a/sql/table.h b/sql/table.h
index d448485a117..2b3c9b8d5b1 100644
--- a/sql/table.h
+++ b/sql/table.h
@@ -437,6 +437,105 @@ typedef struct st_table_share
     return table_map_id;
   }
 
+  /**
+    Convert unrelated members of TABLE_SHARE to one enum
+    representing its metadata type.
+
+    @todo perhaps we need to have a member instead of a function.
+  */
+  enum enum_metadata_type get_metadata_type() const
+  {
+    if (is_view)
+      return METADATA_VIEW;
+    switch (tmp_table) {
+    case NO_TMP_TABLE:
+      return METADATA_BASE_TABLE;
+    case SYSTEM_TMP_TABLE:
+      return METADATA_I_S_TABLE;
+    default:
+      return METADATA_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, spaces (in mathematical sense) of version numbers
+   never intersect for different metadata types. Therefore,
+   version id of a temporary table is never compared with
+   a version id of a view or a temporary table, 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 metadata 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_metadata_version_equal()
+  */
+  ulong get_metadata_version() const
+  {
+    return tmp_table == SYSTEM_TMP_TABLE || is_view ? 0 : table_map_id;
+  }
+
 } TABLE_SHARE;
 
 
@@ -1232,6 +1331,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_metadata_version()
+  */
+  inline
+  bool is_metadata_version_equal(TABLE_SHARE *s) const
+  {
+    return (m_metadata_type == s->get_metadata_type() &&
+            m_metadata_version == s->get_metadata_version());
+  }
+
+  /**
+    Record the value of metadata version of the corresponding
+    table definition cache element in this parse tree node.
+
+    @sa check_and_update_metadata_version()
+  */
+  inline
+  void set_metadata_version(TABLE_SHARE *s)
+  {
+    m_metadata_type= s->get_metadata_type();
+    m_metadata_version= s->get_metadata_version();
+  }
+
 private:
   bool prep_check_option(THD *thd, uint8 check_opt_type);
   bool prep_where(THD *thd, Item **conds, bool no_where_clause);
@@ -1242,6 +1368,10 @@ private:
 
   /* Remembered MERGE child def version.  See top comment in ha_myisammrg.cc */
   ulong         child_def_version;
+  /** See comments for set_metadata_version() */
+  enum enum_metadata_type m_metadata_type;
+  /** See comments for set_metadata_version() */
+  ulong m_metadata_version;
 };
 
 class Item;