Manual merge from the mysql-5.1-bugteam.

30 files changed, 666 insertions, 188 deletions

diff --git a/sql/event_db_repository.cc b/sql/event_db_repository.cc
index 8faab5023da..9f3863eb2b0 100644
--- a/sql/event_db_repository.cc
+++ b/sql/event_db_repository.cc
@@ -26,7 +26,7 @@
 */
 
 static
-const TABLE_FIELD_W_TYPE event_table_fields[ET_FIELD_COUNT] =
+const TABLE_FIELD_TYPE event_table_fields[ET_FIELD_COUNT] =
 {
   {
     { C_STRING_WITH_LEN("db") },
@@ -151,6 +151,24 @@ const TABLE_FIELD_W_TYPE event_table_fields[ET_FIELD_COUNT] =
   }
 };
 
+static const TABLE_FIELD_DEF
+  event_table_def= {ET_FIELD_COUNT, event_table_fields};
+
+class Event_db_intact : public Table_check_intact
+{
+protected:
+  void report_error(uint, const char *fmt, ...)
+  {
+    va_list args;
+    va_start(args, fmt);
+    error_log_print(ERROR_LEVEL, fmt, args);
+    va_end(args);
+  }
+};
+
+/** In case of an error, a message is printed to the error log. */
+static Event_db_intact table_intact;
+
 
 /**
   Puts some data common to CREATE and ALTER EVENT into a row.
@@ -1117,10 +1135,8 @@ Event_db_repository::check_system_tables(THD *thd)
   }
   else
   {
-    if (table_check_intact(tables.table, MYSQL_DB_FIELD_COUNT,
-                           mysql_db_table_fields))
+    if (table_intact.check(tables.table, &mysql_db_table_def))
       ret= 1;
-    /* in case of an error, the message is printed inside table_check_intact */
 
     close_thread_tables(thd);
   }
@@ -1154,9 +1170,8 @@ Event_db_repository::check_system_tables(THD *thd)
   }
   else
   {
-    if (table_check_intact(tables.table, ET_FIELD_COUNT, event_table_fields))
+    if (table_intact.check(tables.table, &event_table_def))
       ret= 1;
-    /* in case of an error, the message is printed inside table_check_intact */
     close_thread_tables(thd);
   }
 
diff --git a/sql/item_create.cc b/sql/item_create.cc
index 7991d9adf82..53aa8081da1 100644
--- a/sql/item_create.cc
+++ b/sql/item_create.cc
@@ -3524,6 +3524,7 @@ Create_func_get_lock Create_func_get_lock::s_singleton;
 Item*
 Create_func_get_lock::create(THD *thd, Item *arg1, Item *arg2)
 {
+  thd->lex->set_stmt_unsafe();
   thd->lex->uncacheable(UNCACHEABLE_SIDEEFFECT);
   return new (thd->mem_root) Item_func_get_lock(arg1, arg2);
 }
@@ -3635,6 +3636,7 @@ Create_func_is_free_lock Create_func_is_free_lock::s_singleton;
 Item*
 Create_func_is_free_lock::create(THD *thd, Item *arg1)
 {
+  thd->lex->set_stmt_unsafe();
   thd->lex->uncacheable(UNCACHEABLE_SIDEEFFECT);
   return new (thd->mem_root) Item_func_is_free_lock(arg1);
 }
@@ -3645,6 +3647,7 @@ Create_func_is_used_lock Create_func_is_used_lock::s_singleton;
 Item*
 Create_func_is_used_lock::create(THD *thd, Item *arg1)
 {
+  thd->lex->set_stmt_unsafe();
   thd->lex->uncacheable(UNCACHEABLE_SIDEEFFECT);
   return new (thd->mem_root) Item_func_is_used_lock(arg1);
 }
@@ -3961,6 +3964,8 @@ Create_func_master_pos_wait::create_native(THD *thd, LEX_STRING name,
   Item *func= NULL;
   int arg_count= 0;
 
+  thd->lex->set_stmt_unsafe();
+
   if (item_list != NULL)
     arg_count= item_list->elements;
 
@@ -4203,6 +4208,7 @@ Create_func_release_lock Create_func_release_lock::s_singleton;
 Item*
 Create_func_release_lock::create(THD *thd, Item *arg1)
 {
+  thd->lex->set_stmt_unsafe();
   thd->lex->uncacheable(UNCACHEABLE_SIDEEFFECT);
   return new (thd->mem_root) Item_func_release_lock(arg1);
 }
@@ -4325,6 +4331,7 @@ Create_func_sleep Create_func_sleep::s_singleton;
 Item*
 Create_func_sleep::create(THD *thd, Item *arg1)
 {
+  thd->lex->set_stmt_unsafe();
   thd->lex->uncacheable(UNCACHEABLE_SIDEEFFECT);
   return new (thd->mem_root) Item_func_sleep(arg1);
 }
@@ -4591,6 +4598,7 @@ Create_func_version Create_func_version::s_singleton;
 Item*
 Create_func_version::create(THD *thd)
 {
+  thd->lex->set_stmt_unsafe();
   return new (thd->mem_root) Item_static_string_func("version()",
                                                      server_version,
                                                      (uint) strlen(server_version),
diff --git a/sql/item_strfunc.cc b/sql/item_strfunc.cc
index 39c4b8e7033..183a628f8e4 100644
--- a/sql/item_strfunc.cc
+++ b/sql/item_strfunc.cc
@@ -1827,8 +1827,9 @@ String *Item_func_database::val_str(String *str)
 
 
 /**
-  @todo
-  make USER() replicate properly (currently it is replicated to "")
+  @note USER() is replicated correctly if binlog_format=ROW or (as of
+  BUG#28086) binlog_format=MIXED, but is incorrectly replicated to ''
+  if binlog_format=STATEMENT.
 */
 bool Item_func_user::init(const char *user, const char *host)
 {
diff --git a/sql/item_timefunc.cc b/sql/item_timefunc.cc
index b5037c08b3c..b293145cc27 100644
--- a/sql/item_timefunc.cc
+++ b/sql/item_timefunc.cc
@@ -386,7 +386,7 @@ static bool extract_date_time(DATE_TIME_FORMAT *format,
 	if (tmp - val > 6)
 	  tmp= (char*) val + 6;
 	l_time->second_part= (int) my_strtoll10(val, &tmp, &error);
-	frac_part= 6 - (uint) (tmp - val);
+	frac_part= 6 - (int) (tmp - val);
 	if (frac_part > 0)
 	  l_time->second_part*= (ulong) log_10_int[frac_part];
 	val= tmp;
@@ -876,9 +876,9 @@ static bool get_interval_info(const char *str,uint length,CHARSET_INFO *cs,
       value= value*LL(10) + (longlong) (*str - '0');
     if (transform_msec && i == count - 1) // microseconds always last
     {
-      long msec_length= 6 - (uint) (str - start);
+      int msec_length= 6 - (int)(str - start);
       if (msec_length > 0)
-	value*= (long) log_10_int[msec_length];
+        value*= (long)log_10_int[msec_length];
     }
     values[i]= value;
     while (str != end && !my_isdigit(cs,*str))
diff --git a/sql/log.cc b/sql/log.cc
index 057f5e8cd7d..b4c9e5eb4cc 100644
--- a/sql/log.cc
+++ b/sql/log.cc
@@ -5650,9 +5650,8 @@ int TC_LOG_BINLOG::recover(IO_CACHE *log, Format_description_log_event *fdle)
       Xid_log_event *xev=(Xid_log_event *)ev;
       uchar *x= (uchar *) memdup_root(&mem_root, (uchar*) &xev->xid,
                                       sizeof(xev->xid));
-      if (! x)
+      if (!x || my_hash_insert(&xids, x))
         goto err2;
-      my_hash_insert(&xids, x);
     }
     delete ev;
   }
diff --git a/sql/log_event.cc b/sql/log_event.cc
index 2cb253c9c56..5e49f7b3312 100644
--- a/sql/log_event.cc
+++ b/sql/log_event.cc
@@ -8450,13 +8450,17 @@ Rows_log_event::write_row(const Relay_log_info *const rli,
   auto_afree_ptr<char> key(NULL);
 
   /* fill table->record[0] with default values */
-
+  bool abort_on_warnings= (rli->sql_thd->variables.sql_mode &
+                           (MODE_STRICT_TRANS_TABLES | MODE_STRICT_ALL_TABLES));
   if ((error= prepare_record(table, m_width,
-                             TRUE /* check if columns have def. values */)))
+                             table->file->ht->db_type != DB_TYPE_NDBCLUSTER,
+                             abort_on_warnings, m_curr_row == m_rows_buf)))
     DBUG_RETURN(error);
   
   /* unpack row into table->record[0] */
-  error= unpack_current_row(rli); // TODO: how to handle errors?
+  if ((error= unpack_current_row(rli, abort_on_warnings)))
+    DBUG_RETURN(error);
+
   if (m_curr_row == m_rows_buf)
   {
     /* this is the first row to be inserted, we estimate the rows with
@@ -9253,8 +9257,12 @@ Update_rows_log_event::do_exec_row(const Relay_log_info *const rli)
 
   store_record(m_table,record[1]);
 
+  bool abort_on_warnings= (rli->sql_thd->variables.sql_mode &
+                           (MODE_STRICT_TRANS_TABLES | MODE_STRICT_ALL_TABLES));
   m_curr_row= m_curr_row_end;
-  error= unpack_current_row(rli); // this also updates m_curr_row_end
+  /* this also updates m_curr_row_end */
+  if ((error= unpack_current_row(rli, abort_on_warnings)))
+    return error;
 
   /*
     Now we have the right row to update.  The old row (the one we're
diff --git a/sql/log_event.h b/sql/log_event.h
index 31d4a7480c2..0b4c63a73af 100644
--- a/sql/log_event.h
+++ b/sql/log_event.h
@@ -3541,12 +3541,16 @@ protected:
   int write_row(const Relay_log_info *const, const bool);
 
   // Unpack the current row into m_table->record[0]
-  int unpack_current_row(const Relay_log_info *const rli)
+  int unpack_current_row(const Relay_log_info *const rli,
+                         const bool abort_on_warning= TRUE)
   { 
     DBUG_ASSERT(m_table);
+
+    bool first_row= (m_curr_row == m_rows_buf);
     ASSERT_OR_RETURN_ERROR(m_curr_row < m_rows_end, HA_ERR_CORRUPT_EVENT);
     int const result= ::unpack_row(rli, m_table, m_width, m_curr_row, &m_cols,
-                                   &m_curr_row_end, &m_master_reclength);
+                                   &m_curr_row_end, &m_master_reclength,
+                                   abort_on_warning, first_row);
     if (m_curr_row_end > m_rows_end)
       my_error(ER_SLAVE_CORRUPT_EVENT, MYF(0));
     ASSERT_OR_RETURN_ERROR(m_curr_row_end <= m_rows_end, HA_ERR_CORRUPT_EVENT);
diff --git a/sql/mysqld.cc b/sql/mysqld.cc
index a3fffdfebbe..1926d447ef7 100644
--- a/sql/mysqld.cc
+++ b/sql/mysqld.cc
@@ -5302,12 +5302,16 @@ pthread_handler_t handle_connections_sockets(void *arg __attribute__((unused)))
 pthread_handler_t handle_connections_namedpipes(void *arg)
 {
   HANDLE hConnectedPipe;
-  OVERLAPPED connectOverlapped = {0};
+  OVERLAPPED connectOverlapped= {0};
   THD *thd;
   my_thread_init();
   DBUG_ENTER("handle_connections_namedpipes");
-  connectOverlapped.hEvent = CreateEvent(NULL, TRUE, FALSE, NULL);
-
+  connectOverlapped.hEvent= CreateEvent(NULL, TRUE, FALSE, NULL);
+  if (!connectOverlapped.hEvent)
+  {
+    sql_print_error("Can't create event, last error=%u", GetLastError());
+    unireg_abort(1);
+  }
   DBUG_PRINT("general",("Waiting for named pipe connections."));
   while (!abort_loop)
   {
@@ -5330,7 +5334,8 @@ pthread_handler_t handle_connections_namedpipes(void *arg)
     {
       CloseHandle(hPipe);
       if ((hPipe= CreateNamedPipe(pipe_name,
-                                  PIPE_ACCESS_DUPLEX|FILE_FLAG_OVERLAPPED,
+                                  PIPE_ACCESS_DUPLEX |
+                                  FILE_FLAG_OVERLAPPED,
                                   PIPE_TYPE_BYTE |
                                   PIPE_READMODE_BYTE |
                                   PIPE_WAIT,
@@ -5350,7 +5355,8 @@ pthread_handler_t handle_connections_namedpipes(void *arg)
     hConnectedPipe = hPipe;
     /* create new pipe for new connection */
     if ((hPipe = CreateNamedPipe(pipe_name,
-				 PIPE_ACCESS_DUPLEX|FILE_FLAG_OVERLAPPED,
+                 PIPE_ACCESS_DUPLEX |
+                 FILE_FLAG_OVERLAPPED,
 				 PIPE_TYPE_BYTE |
 				 PIPE_READMODE_BYTE |
 				 PIPE_WAIT,
diff --git a/sql/opt_range.cc b/sql/opt_range.cc
index 5f9bae22c70..9d8d3bd43d9 100644
--- a/sql/opt_range.cc
+++ b/sql/opt_range.cc
@@ -6671,6 +6671,7 @@ key_or(RANGE_OPT_PARAM *param, SEL_ARG *key1,SEL_ARG *key2)
     else if ((cmp=tmp->cmp_max_to_min(key2)) < 0)
     {						// Found tmp.max < key2.min
       SEL_ARG *next=tmp->next;
+      /* key1 on the left of key2 non-overlapping */
       if (cmp == -2 && eq_tree(tmp->next_key_part,key2->next_key_part))
       {
 	// Join near ranges like tmp.max < 0 and key2.min >= 0
@@ -6699,6 +6700,7 @@ key_or(RANGE_OPT_PARAM *param, SEL_ARG *key1,SEL_ARG *key2)
       int tmp_cmp;
       if ((tmp_cmp=tmp->cmp_min_to_max(key2)) > 0) // if tmp.min > key2.max
       {
+        /* tmp is on the right of key2 non-overlapping */
 	if (tmp_cmp == 2 && eq_tree(tmp->next_key_part,key2->next_key_part))
 	{					// ranges are connected
 	  tmp->copy_min_to_min(key2);
@@ -6733,25 +6735,52 @@ key_or(RANGE_OPT_PARAM *param, SEL_ARG *key1,SEL_ARG *key2)
       }
     }
 
-    // tmp.max >= key2.min && tmp.min <= key.max  (overlapping ranges)
+    /* 
+      tmp.min >= key2.min && tmp.min <= key.max  (overlapping ranges)
+      key2.min <= tmp.min <= key2.max 
+    */  
     if (eq_tree(tmp->next_key_part,key2->next_key_part))
     {
       if (tmp->is_same(key2))
       {
+        /* 
+          Found exact match of key2 inside key1. 
+          Use the relevant range in key1.
+        */
 	tmp->merge_flags(key2);			// Copy maybe flags
 	key2->increment_use_count(-1);		// Free not used tree
       }
       else
       {
 	SEL_ARG *last=tmp;
+        SEL_ARG *first=tmp;
+        /* 
+          Find the last range in tmp that overlaps key2 and has the same 
+          condition on the rest of the keyparts.
+        */
 	while (last->next && last->next->cmp_min_to_max(key2) <= 0 &&
 	       eq_tree(last->next->next_key_part,key2->next_key_part))
 	{
+          /*
+            We've found the last overlapping key1 range in last.
+            This means that the ranges between (and including) the 
+            first overlapping range (tmp) and the last overlapping range
+            (last) are fully nested into the current range of key2 
+            and can safely be discarded. We just need the minimum endpoint
+            of the first overlapping range (tmp) so we can compare it with
+            the minimum endpoint of the enclosing key2 range.
+          */
 	  SEL_ARG *save=last;
 	  last=last->next;
 	  key1=key1->tree_delete(save);
 	}
-        last->copy_min(tmp);
+        /*
+          The tmp range (the first overlapping range) could have been discarded
+          by the previous loop. We should re-direct tmp to the new united range 
+          that's taking its place.
+        */
+        tmp= last;
+        last->copy_min(first);
         bool full_range= last->copy_min(key2);
         if (!full_range)
         {
@@ -9822,7 +9851,11 @@ check_group_min_max_predicates(COND *cond, Item_field *min_max_arg_item,
     }
     else if (cur_arg->const_item())
     {
-      DBUG_RETURN(TRUE);
+      /*
+        For predicates of the form "const OP expr" we also have to check 'expr'
+        to make a decision.
+      */
+      continue;
     }
     else
       DBUG_RETURN(FALSE);
diff --git a/sql/repl_failsafe.cc b/sql/repl_failsafe.cc
index 312499bb4ee..c6a05e93bf4 100644
--- a/sql/repl_failsafe.cc
+++ b/sql/repl_failsafe.cc
@@ -559,7 +559,12 @@ HOSTS";
 	goto err;
       }
       si->server_id = log_server_id;
-      my_hash_insert(&slave_list, (uchar*)si);
+      if (my_hash_insert(&slave_list, (uchar*)si))
+      {
+        error= "the slave is out of memory";
+        pthread_mutex_unlock(&LOCK_slave_list);
+        goto err;
+      }
     }
     strmake(si->host, row[1], sizeof(si->host)-1);
     si->port = atoi(row[port_ind]);
diff --git a/sql/rpl_record.cc b/sql/rpl_record.cc
index 14a80cbb4b6..8e80620dd2c 100644
--- a/sql/rpl_record.cc
+++ b/sql/rpl_record.cc
@@ -180,7 +180,8 @@ int
 unpack_row(Relay_log_info const *rli,
            TABLE *table, uint const colcnt,
            uchar const *const row_data, MY_BITMAP const *cols,
-           uchar const **const row_end, ulong *const master_reclength)
+           uchar const **const row_end, ulong *const master_reclength,
+           const bool abort_on_warning, const bool first_row)
 {
   DBUG_ENTER("unpack_row");
   DBUG_ASSERT(row_data);
@@ -224,8 +225,35 @@ unpack_row(Relay_log_info const *rli,
       /* Field...::unpack() cannot return 0 */
       DBUG_ASSERT(pack_ptr != NULL);
 
-      if ((null_bits & null_mask) && f->maybe_null())
-        f->set_null();
+      if (null_bits & null_mask)
+      {
+        if (f->maybe_null())
+        {
+          DBUG_PRINT("debug", ("Was NULL; null mask: 0x%x; null bits: 0x%x",
+                               null_mask, null_bits));
+          f->set_null();
+        }
+        else
+        {
+          MYSQL_ERROR::enum_warning_level error_type=
+            MYSQL_ERROR::WARN_LEVEL_NOTE;
+          if (abort_on_warning && (table->file->has_transactions() ||
+                                   first_row))
+          {
+            error = HA_ERR_ROWS_EVENT_APPLY;
+            error_type= MYSQL_ERROR::WARN_LEVEL_ERROR;
+          }
+          else
+          {
+            f->set_default();
+            error_type= MYSQL_ERROR::WARN_LEVEL_WARN;
+          }
+          push_warning_printf(current_thd, error_type,
+                              ER_BAD_NULL_ERROR,
+                              ER(ER_BAD_NULL_ERROR),
+                              f->field_name);
+        }
+      }
       else
       {
         f->set_notnull();
@@ -305,13 +333,17 @@ unpack_row(Relay_log_info const *rli,
   @param table  Table whose record[0] buffer is prepared. 
   @param skip   Number of columns for which default/nullable check 
                 should be skipped.
-  @param check  Indicates if errors should be raised when checking 
-                default/nullable field properties.
+  @param check  Specifies if lack of default error needs checking.
+  @param abort_on_warning
+                Controls how to react on lack of a field's default.
+                The parameter mimics the master side one for
+                @c check_that_all_fields_are_given_values.
                 
   @returns 0 on success or a handler level error code
  */ 
 int prepare_record(TABLE *const table, 
-                   const uint skip, const bool check)
+                   const uint skip, const bool check,
+                   const bool abort_on_warning, const bool first_row)
 {
   DBUG_ENTER("prepare_record");
 
@@ -326,17 +358,37 @@ int prepare_record(TABLE *const table,
   if (skip >= table->s->fields || !check)
     DBUG_RETURN(0);
 
-  /* Checking if exists default/nullable fields in the default values. */
-
-  for (Field **field_ptr= table->field+skip ; *field_ptr ; ++field_ptr)
+  /*
+    For fields the extra fields on the slave, we check if they have a default.
+    The check follows the same rules as the INSERT query without specifying an
+    explicit value for a field not having the explicit default 
+    (@c check_that_all_fields_are_given_values()).
+  */
+  for (Field **field_ptr= table->field+skip; *field_ptr; ++field_ptr)
   {
     uint32 const mask= NOT_NULL_FLAG | NO_DEFAULT_VALUE_FLAG;
     Field *const f= *field_ptr;
-
-    if (((f->flags & mask) == mask))
+    if ((f->flags &  NO_DEFAULT_VALUE_FLAG) &&
+        (f->real_type() != MYSQL_TYPE_ENUM))
     {
-      my_error(ER_NO_DEFAULT_FOR_FIELD, MYF(0), f->field_name);
-      error = HA_ERR_ROWS_EVENT_APPLY;
+
+      MYSQL_ERROR::enum_warning_level error_type=
+        MYSQL_ERROR::WARN_LEVEL_NOTE;
+      if (abort_on_warning && (table->file->has_transactions() ||
+                               first_row))
+      {
+        error= HA_ERR_ROWS_EVENT_APPLY;
+        error_type= MYSQL_ERROR::WARN_LEVEL_ERROR;
+      }
+      else
+      {
+        f->set_default();
+        error_type= MYSQL_ERROR::WARN_LEVEL_WARN;
+      }
+      push_warning_printf(current_thd, error_type,
+                          ER_NO_DEFAULT_FOR_FIELD,
+                          ER(ER_NO_DEFAULT_FOR_FIELD),
+                          f->field_name);
     }
   }
 
diff --git a/sql/rpl_record.h b/sql/rpl_record.h
index f9e64f0ab1d..6e8838f82b3 100644
--- a/sql/rpl_record.h
+++ b/sql/rpl_record.h
@@ -27,10 +27,13 @@ size_t pack_row(TABLE* table, MY_BITMAP const* cols,
 int unpack_row(Relay_log_info const *rli,
                TABLE *table, uint const colcnt,
                uchar const *const row_data, MY_BITMAP const *cols,
-               uchar const **const row_end, ulong *const master_reclength);
+               uchar const **const row_end, ulong *const master_reclength,
+               const bool abort_on_warning= TRUE, const bool first_row= TRUE);
 
 // Fill table's record[0] with default values.
-int prepare_record(TABLE *const, const uint =0, const bool =FALSE);
+int prepare_record(TABLE *const table, const uint skip, const bool check,
+                   const bool abort_on_warning= TRUE,
+                   const bool first_row= TRUE);
 #endif
 
 #endif
diff --git a/sql/rpl_tblmap.cc b/sql/rpl_tblmap.cc
index a004c354263..6ef9a8623fe 100644
--- a/sql/rpl_tblmap.cc
+++ b/sql/rpl_tblmap.cc
@@ -119,7 +119,13 @@ int table_mapping::set_table(ulong table_id, TABLE* table)
   }
   e->table_id= table_id;
   e->table= table;
-  my_hash_insert(&m_table_ids,(uchar *)e);
+  if (my_hash_insert(&m_table_ids,(uchar *)e))
+  {
+    /* we add this entry to the chain of free (free for use) entries */
+    e->next= m_free;
+    m_free= e;
+    DBUG_RETURN(ERR_MEMORY_ALLOCATION);
+  }
 
   DBUG_PRINT("info", ("tid %lu -> table 0x%lx (%s)", 
 		      table_id, (long) e->table,
diff --git a/sql/sp.cc b/sql/sp.cc
index fd420732628..d3c5dfb96d0 100644
--- a/sql/sp.cc
+++ b/sql/sp.cc
@@ -70,6 +70,122 @@ enum
   MYSQL_PROC_FIELD_COUNT
 };
 
+static const
+TABLE_FIELD_TYPE proc_table_fields[MYSQL_PROC_FIELD_COUNT] =
+{
+  {
+    { C_STRING_WITH_LEN("db") },
+    { C_STRING_WITH_LEN("char(64)") },
+    { C_STRING_WITH_LEN("utf8") }
+  },
+  {
+    { C_STRING_WITH_LEN("name") },
+    { C_STRING_WITH_LEN("char(64)") },
+    { C_STRING_WITH_LEN("utf8") }
+  },
+  {
+    { C_STRING_WITH_LEN("type") },
+    { C_STRING_WITH_LEN("enum('FUNCTION','PROCEDURE')") },
+    { NULL, 0 }
+  },
+  {
+    { C_STRING_WITH_LEN("specific_name") },
+    { C_STRING_WITH_LEN("char(64)") },
+    { C_STRING_WITH_LEN("utf8") }
+  },
+  {
+    { C_STRING_WITH_LEN("language") },
+    { C_STRING_WITH_LEN("enum('SQL')") },
+    { NULL, 0 }
+  },
+  {
+    { C_STRING_WITH_LEN("sql_data_access") },
+    { C_STRING_WITH_LEN("enum('CONTAINS_SQL','NO_SQL','READS_SQL_DATA','MODIFIES_SQL_DATA')") },
+    { NULL, 0 }
+  },
+  {
+    { C_STRING_WITH_LEN("is_deterministic") },
+    { C_STRING_WITH_LEN("enum('YES','NO')") },
+    { NULL, 0 }
+  },
+  {
+    { C_STRING_WITH_LEN("security_type") },
+    { C_STRING_WITH_LEN("enum('INVOKER','DEFINER')") },
+    { NULL, 0 }
+  },
+  {
+    { C_STRING_WITH_LEN("param_list") },
+    { C_STRING_WITH_LEN("blob") },
+    { NULL, 0 }
+  },
+
+  {
+    { C_STRING_WITH_LEN("returns") },
+    { C_STRING_WITH_LEN("longblob") },
+    { NULL, 0 }
+  },
+  {
+    { C_STRING_WITH_LEN("body") },
+    { C_STRING_WITH_LEN("longblob") },
+    { NULL, 0 }
+  },
+  {
+    { C_STRING_WITH_LEN("definer") },
+    { C_STRING_WITH_LEN("char(77)") },
+    { C_STRING_WITH_LEN("utf8") }
+  },
+  {
+    { C_STRING_WITH_LEN("created") },
+    { C_STRING_WITH_LEN("timestamp") },
+    { NULL, 0 }
+  },
+  {
+    { C_STRING_WITH_LEN("modified") },
+    { C_STRING_WITH_LEN("timestamp") },
+    { NULL, 0 }
+  },
+  {
+    { C_STRING_WITH_LEN("sql_mode") },
+    { C_STRING_WITH_LEN("set('REAL_AS_FLOAT','PIPES_AS_CONCAT','ANSI_QUOTES',"
+    "'IGNORE_SPACE','NOT_USED','ONLY_FULL_GROUP_BY','NO_UNSIGNED_SUBTRACTION',"
+    "'NO_DIR_IN_CREATE','POSTGRESQL','ORACLE','MSSQL','DB2','MAXDB',"
+    "'NO_KEY_OPTIONS','NO_TABLE_OPTIONS','NO_FIELD_OPTIONS','MYSQL323','MYSQL40',"
+    "'ANSI','NO_AUTO_VALUE_ON_ZERO','NO_BACKSLASH_ESCAPES','STRICT_TRANS_TABLES',"
+    "'STRICT_ALL_TABLES','NO_ZERO_IN_DATE','NO_ZERO_DATE','INVALID_DATES',"
+    "'ERROR_FOR_DIVISION_BY_ZERO','TRADITIONAL','NO_AUTO_CREATE_USER',"
+    "'HIGH_NOT_PRECEDENCE','NO_ENGINE_SUBSTITUTION','PAD_CHAR_TO_FULL_LENGTH')") },
+    { NULL, 0 }
+  },
+  {
+    { C_STRING_WITH_LEN("comment") },
+    { C_STRING_WITH_LEN("char(64)") },
+    { C_STRING_WITH_LEN("utf8") }
+  },
+  {
+    { C_STRING_WITH_LEN("character_set_client") },
+    { C_STRING_WITH_LEN("char(32)") },
+    { C_STRING_WITH_LEN("utf8") }
+  },
+  {
+    { C_STRING_WITH_LEN("collation_connection") },
+    { C_STRING_WITH_LEN("char(32)") },
+    { C_STRING_WITH_LEN("utf8") }
+  },
+  {
+    { C_STRING_WITH_LEN("db_collation") },
+    { C_STRING_WITH_LEN("char(32)") },
+    { C_STRING_WITH_LEN("utf8") }
+  },
+  {
+    { C_STRING_WITH_LEN("body_utf8") },
+    { C_STRING_WITH_LEN("longblob") },
+    { NULL, 0 }
+  }
+};
+
+static const TABLE_FIELD_DEF
+  proc_table_def= {MYSQL_PROC_FIELD_COUNT, proc_table_fields};
+
 /*************************************************************************/
 
 /**
@@ -247,6 +363,50 @@ Stored_routine_creation_ctx::load_from_db(THD *thd,
 
 /*************************************************************************/
 
+class Proc_table_intact : public Table_check_intact
+{
+private:
+  bool m_print_once;
+
+public:
+  Proc_table_intact() : m_print_once(TRUE) {}
+
+protected:
+  void report_error(uint code, const char *fmt, ...);
+};
+
+
+/**
+  Report failure to validate the mysql.proc table definition.
+  Print a message to the error log only once.
+*/
+
+void Proc_table_intact::report_error(uint code, const char *fmt, ...)
+{
+  va_list args;
+  char buf[512];
+
+  va_start(args, fmt);
+  my_vsnprintf(buf, sizeof(buf), fmt, args);
+  va_end(args);
+
+  if (code)
+    my_message(code, buf, MYF(0));
+  else
+    my_error(ER_CANNOT_LOAD_FROM_TABLE, MYF(0), "proc");
+
+  if (m_print_once)
+  {
+    m_print_once= FALSE;
+    sql_print_error("%s", buf);
+  }
+};
+
+
+/** Single instance used to control printing to the error log. */
+static Proc_table_intact proc_table_intact;
+
+
 /**
   Open the mysql.proc table for read.
 
@@ -266,15 +426,17 @@ TABLE *open_proc_table_for_read(THD *thd, Open_tables_state *backup)
   DBUG_ENTER("open_proc_table_for_read");
 
   TABLE_LIST table;
-  bzero((char*) &table, sizeof(table));
-  table.db= (char*) "mysql";
-  table.table_name= table.alias= (char*)"proc";
-  table.lock_type= TL_READ;
+  table.init_one_table("mysql", "proc", TL_READ);
 
-  if (!open_system_tables_for_read(thd, &table, backup))
+  if (open_system_tables_for_read(thd, &table, backup))
+    DBUG_RETURN(NULL);
+
+  if (!proc_table_intact.check(table.table, &proc_table_def))
     DBUG_RETURN(table.table);
-  else
-    DBUG_RETURN(0);
+
+  close_system_tables(thd, backup);
+
+  DBUG_RETURN(NULL);
 }
 
 
@@ -296,13 +458,19 @@ static TABLE *open_proc_table_for_update(THD *thd)
 {
   DBUG_ENTER("open_proc_table_for_update");
 
-  TABLE_LIST table;
-  bzero((char*) &table, sizeof(table));
-  table.db= (char*) "mysql";
-  table.table_name= table.alias= (char*)"proc";
-  table.lock_type= TL_WRITE;
+  TABLE *table;
+  TABLE_LIST table_list;
+  table_list.init_one_table("mysql", "proc", TL_WRITE);
+
+  if (!(table= open_system_table_for_update(thd, &table_list)))
+    DBUG_RETURN(NULL);
 
-  DBUG_RETURN(open_system_table_for_update(thd, &table));
+  if (!proc_table_intact.check(table, &proc_table_def))
+    DBUG_RETURN(table);
+
+  close_thread_tables(thd);
+
+  DBUG_RETURN(NULL);
 }
 
 
@@ -1506,7 +1674,8 @@ static bool add_used_routine(LEX *lex, Query_arena *arena,
     rn->key.length= key->length;
     rn->key.str= (char *)rn + sizeof(Sroutine_hash_entry);
     memcpy(rn->key.str, key->str, key->length + 1);
-    my_hash_insert(&lex->sroutines, (uchar *)rn);
+    if (my_hash_insert(&lex->sroutines, (uchar *)rn))
+      return FALSE;
     lex->sroutines_list.link_in_list((uchar *)rn, (uchar **)&rn->next);
     rn->belong_to_view= belong_to_view;
     return TRUE;
@@ -1584,16 +1753,24 @@ void sp_remove_not_own_routines(LEX *lex)
     dependant on time of life of elements from source hash. It also
     won't touch lists linking elements in source and destination
     hashes.
+
+  @returns
+    @return TRUE Failure
+    @return FALSE Success
 */
 
-void sp_update_sp_used_routines(HASH *dst, HASH *src)
+bool sp_update_sp_used_routines(HASH *dst, HASH *src)
 {
   for (uint i=0 ; i < src->records ; i++)
   {
     Sroutine_hash_entry *rt= (Sroutine_hash_entry *)hash_element(src, i);
     if (!hash_search(dst, (uchar *)rt->key.str, rt->key.length))
-      my_hash_insert(dst, (uchar *)rt);
+    {
+      if (my_hash_insert(dst, (uchar *)rt))
+        return TRUE;
+    }
   }
+  return FALSE;
 }
 
 
diff --git a/sql/sp.h b/sql/sp.h
index 5a190c5480e..876287d9704 100644
--- a/sql/sp.h
+++ b/sql/sp.h
@@ -69,7 +69,7 @@ void sp_get_prelocking_info(THD *thd, bool *need_prelocking,
 void sp_add_used_routine(LEX *lex, Query_arena *arena,
                          sp_name *rt, char rt_type);
 void sp_remove_not_own_routines(LEX *lex);
-void sp_update_sp_used_routines(HASH *dst, HASH *src);
+bool sp_update_sp_used_routines(HASH *dst, HASH *src);
 int sp_cache_routines_and_add_tables(THD *thd, LEX *lex,
                                      bool first_no_prelock);
 int sp_cache_routines_and_add_tables_for_view(THD *thd, LEX *lex,
diff --git a/sql/sp_cache.cc b/sql/sp_cache.cc
index 64898915b7e..b8209a373a2 100644
--- a/sql/sp_cache.cc
+++ b/sql/sp_cache.cc
@@ -36,10 +36,16 @@ public:
   sp_cache();
   ~sp_cache();
 
-  inline void insert(sp_head *sp)
+  /**
+   Inserts a sp_head object into a hash table.
+
+   @returns Success status
+     @return TRUE Failure
+     @return FALSE Success
+  */
+  inline bool insert(sp_head *sp)
   {
-    /* TODO: why don't we check return value? */
-    my_hash_insert(&m_hashtable, (const uchar *)sp);
+    return my_hash_insert(&m_hashtable, (const uchar *)sp);
   }
 
   inline sp_head *lookup(char *name, uint namelen)
diff --git a/sql/sp_head.cc b/sql/sp_head.cc
index 940af540fd7..78539034677 100644
--- a/sql/sp_head.cc
+++ b/sql/sp_head.cc
@@ -2091,8 +2091,18 @@ sp_head::reset_lex(THD *thd)
   DBUG_RETURN(FALSE);
 }
 
-/// Restore lex during parsing, after we have parsed a sub statement.
-void
+
+/**
+  Restore lex during parsing, after we have parsed a sub statement.
+
+  @param thd Thread handle
+
+  @return
+    @retval TRUE failure
+    @retval FALSE success
+*/
+
+bool
 sp_head::restore_lex(THD *thd)
 {
   DBUG_ENTER("sp_head::restore_lex");
@@ -2103,7 +2113,7 @@ sp_head::restore_lex(THD *thd)
 
   oldlex= (LEX *)m_lex.pop();
   if (! oldlex)
-    return;			// Nothing to restore
+    DBUG_RETURN(FALSE);			// Nothing to restore
 
   oldlex->trg_table_fields.push_back(&sublex->trg_table_fields);
 
@@ -2119,7 +2129,8 @@ sp_head::restore_lex(THD *thd)
     Add routines which are used by statement to respective set for
     this routine.
   */
-  sp_update_sp_used_routines(&m_sroutines, &sublex->sroutines);
+  if (sp_update_sp_used_routines(&m_sroutines, &sublex->sroutines))
+    DBUG_RETURN(TRUE);
   /*
     Merge tables used by this statement (but not by its functions or
     procedures) to multiset of tables used by this routine.
@@ -2131,7 +2142,7 @@ sp_head::restore_lex(THD *thd)
     delete sublex;
   }
   thd->lex= oldlex;
-  DBUG_VOID_RETURN;
+  DBUG_RETURN(FALSE);
 }
 
 /**
@@ -3876,7 +3887,8 @@ sp_head::merge_table_list(THD *thd, TABLE_LIST *table, LEX *lex_for_tmp_check)
         tab->lock_type= table->lock_type;
         tab->lock_count= tab->query_lock_count= 1;
         tab->trg_event_map= table->trg_event_map;
-	my_hash_insert(&m_sptabs, (uchar *)tab);
+	if (my_hash_insert(&m_sptabs, (uchar *)tab))
+          return FALSE;
       }
     }
   return TRUE;
diff --git a/sql/sp_head.h b/sql/sp_head.h
index dd11f8693ac..00c96d44f70 100644
--- a/sql/sp_head.h
+++ b/sql/sp_head.h
@@ -340,7 +340,7 @@ public:
 
     @todo Conflicting comment in sp_head.cc
   */
-  void
+  bool
   restore_lex(THD *thd);
 
   /// Put the instruction on the backpatch list, associated with the label.
diff --git a/sql/sql_acl.cc b/sql/sql_acl.cc
index 0592bb3be1d..eb91f66d114 100644
--- a/sql/sql_acl.cc
+++ b/sql/sql_acl.cc
@@ -31,9 +31,8 @@
 #include "sp_head.h"
 #include "sp.h"
 
-time_t mysql_db_table_last_check= 0L;
-
-TABLE_FIELD_W_TYPE mysql_db_table_fields[MYSQL_DB_FIELD_COUNT] = {
+static const
+TABLE_FIELD_TYPE mysql_db_table_fields[MYSQL_DB_FIELD_COUNT] = {
   {
     { C_STRING_WITH_LEN("Host") },            
     { C_STRING_WITH_LEN("char(60)") },
@@ -146,6 +145,8 @@ TABLE_FIELD_W_TYPE mysql_db_table_fields[MYSQL_DB_FIELD_COUNT] = {
   }
 };
 
+const TABLE_FIELD_DEF
+  mysql_db_table_def= {MYSQL_DB_FIELD_COUNT, mysql_db_table_fields};
 
 #ifndef NO_EMBEDDED_ACCESS_CHECKS
 
@@ -2404,7 +2405,12 @@ GRANT_TABLE::GRANT_TABLE(TABLE *form, TABLE *col_privs)
         privs = cols = 0;			/* purecov: deadcode */
         return;				/* purecov: deadcode */
       }
-      my_hash_insert(&hash_columns, (uchar *) mem_check);
+      if (my_hash_insert(&hash_columns, (uchar *) mem_check))
+      {
+        /* Invalidate this entry */
+        privs= cols= 0;
+        return;
+      }
     } while (!col_privs->file->index_next(col_privs->record[0]) &&
              !key_cmp_if_same(col_privs,key,0,key_prefix_len));
     col_privs->file->ha_index_end();
@@ -2609,7 +2615,11 @@ static int replace_column_table(GRANT_TABLE *g_t,
 	goto end;				/* purecov: inspected */
       }
       grant_column= new GRANT_COLUMN(column->column,privileges);
-      my_hash_insert(&g_t->hash_columns,(uchar*) grant_column);
+      if (my_hash_insert(&g_t->hash_columns,(uchar*) grant_column))
+      {
+        result= -1;
+        goto end;
+      }
     }
   }
 
@@ -3134,12 +3144,12 @@ int mysql_table_grant(THD *thd, TABLE_LIST *table_list,
 				     Str->user.str, table_name,
 				     rights,
 				     column_priv);
-      if (!grant_table)				// end of memory
+      if (!grant_table ||
+        my_hash_insert(&column_priv_hash,(uchar*) grant_table))
       {
 	result= TRUE;				/* purecov: deadcode */
 	continue;				/* purecov: deadcode */
       }
-      my_hash_insert(&column_priv_hash,(uchar*) grant_table);
     }
 
     /* If revoke_grant, calculate the new column privilege for tables_priv */
@@ -3343,12 +3353,13 @@ bool mysql_routine_grant(THD *thd, TABLE_LIST *table_list, bool is_proc,
       grant_name= new GRANT_NAME(Str->host.str, db_name,
 				 Str->user.str, table_name,
 				 rights, TRUE);
-      if (!grant_name)
+      if (!grant_name ||
+        my_hash_insert(is_proc ?
+                       &proc_priv_hash : &func_priv_hash,(uchar*) grant_name))
       {
         result= TRUE;
 	continue;
       }
-      my_hash_insert(is_proc ? &proc_priv_hash : &func_priv_hash,(uchar*) grant_name);
     }
 
     if (replace_routine_table(thd, grant_name, tables[1].table, *Str,
diff --git a/sql/sql_acl.h b/sql/sql_acl.h
index a8090fba2e7..4c835e2718c 100644
--- a/sql/sql_acl.h
+++ b/sql/sql_acl.h
@@ -159,8 +159,7 @@ enum mysql_db_table_field
   MYSQL_DB_FIELD_COUNT
 };
 
-extern TABLE_FIELD_W_TYPE mysql_db_table_fields[];
-extern time_t mysql_db_table_last_check;
+extern const TABLE_FIELD_DEF mysql_db_table_def;
 
 /* Classes */
 
diff --git a/sql/sql_base.cc b/sql/sql_base.cc
index be31ffe04dd..cf8a0b32764 100644
--- a/sql/sql_base.cc
+++ b/sql/sql_base.cc
@@ -2935,7 +2935,12 @@ TABLE *open_table(THD *thd, TABLE_LIST *table_list, MEM_ROOT *mem_root,
     DBUG_PRINT("info", ("inserting table '%s'.'%s' 0x%lx into the cache",
                         table->s->db.str, table->s->table_name.str,
                         (long) table));
-    VOID(my_hash_insert(&open_cache,(uchar*) table));
+    if (my_hash_insert(&open_cache,(uchar*) table))
+    {
+      my_free(table, MYF(0));
+      VOID(pthread_mutex_unlock(&LOCK_open));
+      DBUG_RETURN(NULL);
+    }
   }
 
   check_unused();				// Debugging call
diff --git a/sql/sql_cache.cc b/sql/sql_cache.cc
index 7190c63d057..8114e0221d6 100644
--- a/sql/sql_cache.cc
+++ b/sql/sql_cache.cc
@@ -422,12 +422,16 @@ TYPELIB query_cache_type_typelib=
   effect by another thread. This enables a quick path in execution to skip waits
   when the outcome is known.
 
+  @param use_timeout TRUE if the lock can abort because of a timeout.
+
+  @note use_timeout is optional and default value is FALSE.
+
   @return
    @retval FALSE An exclusive lock was taken
    @retval TRUE The locking attempt failed
 */
 
-bool Query_cache::try_lock(void)
+bool Query_cache::try_lock(bool use_timeout)
 {
   bool interrupt= FALSE;
   DBUG_ENTER("Query_cache::try_lock");
@@ -457,7 +461,26 @@ bool Query_cache::try_lock(void)
     else
     {
       DBUG_ASSERT(m_cache_lock_status == Query_cache::LOCKED);
-      pthread_cond_wait(&COND_cache_status_changed, &structure_guard_mutex);
+      /*
+        To prevent send_result_to_client() and query_cache_insert() from
+        blocking execution for too long a timeout is put on the lock.
+      */
+      if (use_timeout)
+      {
+        struct timespec waittime;
+        set_timespec_nsec(waittime,(ulong)(50000000L));  /* Wait for 50 msec */
+        int res= pthread_cond_timedwait(&COND_cache_status_changed,
+                                        &structure_guard_mutex,&waittime);
+        if (res == ETIMEDOUT)
+        {
+          interrupt= TRUE;
+          break;
+        }
+      }
+      else
+      {
+        pthread_cond_wait(&COND_cache_status_changed, &structure_guard_mutex);
+      }
     }
   }
   pthread_mutex_unlock(&structure_guard_mutex);
@@ -1191,8 +1214,14 @@ def_week_frmt: %lu, in_trans: %d, autocommit: %d",
       A table- or a full flush operation can potentially take a long time to
       finish. We choose not to wait for them and skip caching statements
       instead.
+
+      In case the wait time can't be determined there is an upper limit which
+      causes try_lock() to abort with a time out.
+
+      The 'TRUE' parameter indicate that the lock is allowed to timeout
+
     */
-    if (try_lock())
+    if (try_lock(TRUE))
       DBUG_VOID_RETURN;
     if (query_cache_size == 0)
     {
@@ -1390,8 +1419,10 @@ Query_cache::send_result_to_client(THD *thd, char *sql, uint query_length)
     Try to obtain an exclusive lock on the query cache. If the cache is
     disabled or if a full cache flush is in progress, the attempt to
     get the lock is aborted.
+
+    The 'TRUE' parameter indicate that the lock is allowed to timeout
   */
-  if (try_lock())
+  if (try_lock(TRUE))
     goto err;
 
   if (query_cache_size == 0)
diff --git a/sql/sql_cache.h b/sql/sql_cache.h
index 777ddd39280..44fc3123b98 100644
--- a/sql/sql_cache.h
+++ b/sql/sql_cache.h
@@ -485,7 +485,7 @@ protected:
 			const char *name);
   my_bool in_blocks(Query_cache_block * point);
 
-  bool try_lock(void);
+  bool try_lock(bool use_timeout= FALSE);
   void lock(void);
   void lock_and_suspend(void);
   void unlock(void);
diff --git a/sql/sql_delete.cc b/sql/sql_delete.cc
index b80138af7f2..6b9a83e695b 100644
--- a/sql/sql_delete.cc
+++ b/sql/sql_delete.cc
@@ -426,7 +426,8 @@ cleanup:
   }
   DBUG_ASSERT(transactional_table || !deleted || thd->transaction.stmt.modified_non_trans_table);
   free_underlaid_joins(thd, select_lex);
-  if (error < 0 || (thd->lex->ignore && !thd->is_fatal_error))
+  if (error < 0 || 
+      (thd->lex->ignore && !thd->is_error() && !thd->is_fatal_error))
   {
     /*
       If a TRUNCATE TABLE was issued, the number of rows should be reported as
@@ -1089,6 +1090,7 @@ bool mysql_truncate(THD *thd, TABLE_LIST *table_list, bool dont_send_ok)
   TABLE *table;
   bool error;
   uint path_length;
+  bool is_temporary_table= false;
   DBUG_ENTER("mysql_truncate");
 
   bzero((char*) &create_info,sizeof(create_info));
@@ -1099,6 +1101,7 @@ bool mysql_truncate(THD *thd, TABLE_LIST *table_list, bool dont_send_ok)
   /* If it is a temporary table, close and regenerate it */
   if (!dont_send_ok && (table= find_temporary_table(thd, table_list)))
   {
+    is_temporary_table= true;
     handlerton *table_type= table->s->db_type();
     TABLE_SHARE *share= table->s;
     if (!ha_check_storage_engine_flag(table_type, HTON_CAN_RECREATE))
@@ -1162,11 +1165,9 @@ end:
   {
     if (!error)
     {
-      /*
-        TRUNCATE must always be statement-based binlogged (not row-based) so
-        we don't test current_stmt_binlog_row_based.
-      */
-      write_bin_log(thd, TRUE, thd->query(), thd->query_length());
+      /* In RBR, the statement is not binlogged if the table is temporary. */
+      if (!is_temporary_table || !thd->current_stmt_binlog_row_based)
+        write_bin_log(thd, TRUE, thd->query(), thd->query_length());
       my_ok(thd);		// This should return record count
     }
     VOID(pthread_mutex_lock(&LOCK_open));
diff --git a/sql/sql_insert.cc b/sql/sql_insert.cc
index a4785418d4c..6281dd8168a 100644
--- a/sql/sql_insert.cc
+++ b/sql/sql_insert.cc
@@ -500,6 +500,22 @@ bool open_and_lock_for_insert_delayed(THD *thd, TABLE_LIST *table_list)
   DBUG_ENTER("open_and_lock_for_insert_delayed");
 
 #ifndef EMBEDDED_LIBRARY
+  if (thd->locked_tables && thd->global_read_lock)
+  {
+    /*
+      If this connection has the global read lock, the handler thread
+      will not be able to lock the table. It will wait for the global
+      read lock to go away, but this will never happen since the
+      connection thread will be stuck waiting for the handler thread
+      to open and lock the table.
+      If we are not in locked tables mode, INSERT will seek protection
+      against the global read lock (and fail), thus we will only get
+      to this point in locked tables mode.
+    */
+    my_error(ER_CANT_UPDATE_WITH_READLOCK, MYF(0));
+    DBUG_RETURN(TRUE);
+  }
+
   if (delayed_get_table(thd, table_list))
     DBUG_RETURN(TRUE);
 
diff --git a/sql/sql_partition.cc b/sql/sql_partition.cc
index 04d22174c8e..0decd3e1a91 100644
--- a/sql/sql_partition.cc
+++ b/sql/sql_partition.cc
@@ -196,26 +196,27 @@ bool partition_default_handling(TABLE *table, partition_info *part_info,
 {
   DBUG_ENTER("partition_default_handling");
 
-  if (part_info->use_default_no_partitions)
+  if (!is_create_table_ind)
   {
-    if (!is_create_table_ind &&
-        table->file->get_no_parts(normalized_path, &part_info->no_parts))
+    if (part_info->use_default_no_partitions)
     {
-      DBUG_RETURN(TRUE);
+      if (table->file->get_no_parts(normalized_path, &part_info->no_parts))
+      {
+        DBUG_RETURN(TRUE);
+      }
     }
-  }
-  else if (part_info->is_sub_partitioned() &&
-           part_info->use_default_no_subpartitions)
-  {
-    uint no_parts;
-    if (!is_create_table_ind &&
-        (table->file->get_no_parts(normalized_path, &no_parts)))
+    else if (part_info->is_sub_partitioned() &&
+             part_info->use_default_no_subpartitions)
     {
-      DBUG_RETURN(TRUE);
+      uint no_parts;
+      if (table->file->get_no_parts(normalized_path, &no_parts))
+      {
+        DBUG_RETURN(TRUE);
+      }
+      DBUG_ASSERT(part_info->no_parts > 0);
+      DBUG_ASSERT((no_parts % part_info->no_parts) == 0);
+      part_info->no_subparts= no_parts / part_info->no_parts;
     }
-    DBUG_ASSERT(part_info->no_parts > 0);
-    part_info->no_subparts= no_parts / part_info->no_parts;
-    DBUG_ASSERT((no_parts % part_info->no_parts) == 0);
   }
   part_info->set_up_defaults_for_partitioning(table->file,
                                               (ulonglong)0, (uint)0);
@@ -905,6 +906,8 @@ bool fix_fields_part_func(THD *thd, Item* func_expr, TABLE *table,
   char* db_name;
   char db_name_string[FN_REFLEN];
   bool save_use_only_table_context;
+  uint8 saved_full_group_by_flag;
+  nesting_map saved_allow_sum_func;
   DBUG_ENTER("fix_fields_part_func");
 
   if (part_info->fixed)
@@ -974,9 +977,19 @@ bool fix_fields_part_func(THD *thd, Item* func_expr, TABLE *table,
   save_use_only_table_context= thd->lex->use_only_table_context;
   thd->lex->use_only_table_context= TRUE;
   thd->lex->current_select->cur_pos_in_select_list= UNDEF_POS;
+  saved_full_group_by_flag= thd->lex->current_select->full_group_by_flag;
+  saved_allow_sum_func= thd->lex->allow_sum_func;
+  thd->lex->allow_sum_func= 0;
   
   error= func_expr->fix_fields(thd, (Item**)&func_expr);
 
+  /*
+    Restore full_group_by_flag and allow_sum_func,
+    fix_fields should not affect mysql_select later, see Bug#46923.
+  */
+  thd->lex->current_select->full_group_by_flag= saved_full_group_by_flag;
+  thd->lex->allow_sum_func= saved_allow_sum_func;
+
   thd->lex->use_only_table_context= save_use_only_table_context;
 
   context->table_list= save_table_list;
@@ -1679,7 +1692,7 @@ bool fix_partition_func(THD *thd, TABLE *table,
   if (((part_info->part_type != HASH_PARTITION ||
       part_info->list_of_part_fields == FALSE) &&
       check_part_func_fields(part_info->part_field_array, TRUE)) ||
-      (part_info->list_of_part_fields == FALSE &&
+      (part_info->list_of_subpart_fields == FALSE &&
        part_info->is_sub_partitioned() &&
        check_part_func_fields(part_info->subpart_field_array, TRUE)))
   {
diff --git a/sql/sql_select.cc b/sql/sql_select.cc
index c7d51a96d45..2272b75ee29 100644
--- a/sql/sql_select.cc
+++ b/sql/sql_select.cc
@@ -1304,7 +1304,7 @@ JOIN::optimize()
 	     ((select_distinct || !simple_order || !simple_group) ||
 	      (group_list && order) ||
 	      test(select_options & OPTION_BUFFER_RESULT))) ||
-             rollup.state != ROLLUP::STATE_NONE && select_distinct);
+             (rollup.state != ROLLUP::STATE_NONE && select_distinct));
 
   // No cache for MATCH
   make_join_readinfo(this,
@@ -2134,17 +2134,13 @@ JOIN::exec()
 	    DBUG_VOID_RETURN;
 	if (!curr_table->select->cond)
 	  curr_table->select->cond= sort_table_cond;
-	else					// This should never happen
+	else
 	{
 	  if (!(curr_table->select->cond=
 		new Item_cond_and(curr_table->select->cond,
 				  sort_table_cond)))
 	    DBUG_VOID_RETURN;
-	  /*
-	    Item_cond_and do not need fix_fields for execution, its parameters
-	    are fixed or do not need fix_fields, too
-	  */
-	  curr_table->select->cond->quick_fix_field();
+	  curr_table->select->cond->fix_fields(thd, 0);
 	}
 	curr_table->select_cond= curr_table->select->cond;
 	curr_table->select_cond->top_level_item();
@@ -14014,7 +14010,10 @@ static int remove_dup_with_hash_index(THD *thd, TABLE *table,
 	goto err;
     }
     else
-      (void) my_hash_insert(&hash, org_key_pos);
+    {
+      if (my_hash_insert(&hash, org_key_pos))
+        goto err;
+    }
     key_pos+=extra_length;
   }
   my_free((char*) key_buffer,MYF(0));
diff --git a/sql/sql_yacc.yy b/sql/sql_yacc.yy
index 86bef83cf1a..90bd7395732 100644
--- a/sql/sql_yacc.yy
+++ b/sql/sql_yacc.yy
@@ -2330,8 +2330,8 @@ sp_decl:
             }
 
             pctx->declare_var_boundary(0);
-            lex->sphead->restore_lex(YYTHD);
-
+            if (lex->sphead->restore_lex(YYTHD))
+              MYSQL_YYABORT;
             $$.vars= $2;
             $$.conds= $$.hndlrs= $$.curs= 0;
           }
@@ -2441,7 +2441,8 @@ sp_cursor_stmt:
             }
             lex->sp_lex_in_use= TRUE;
             $$= lex;
-            lex->sphead->restore_lex(YYTHD);
+            if (lex->sphead->restore_lex(YYTHD))
+              MYSQL_YYABORT;
           }
         ;
 
@@ -2660,7 +2661,8 @@ sp_proc_stmt_statement:
                     sp->add_instr(i))
                 MYSQL_YYABORT;
             }
-            sp->restore_lex(thd);
+            if (sp->restore_lex(thd))
+              MYSQL_YYABORT;
           }
         ;
 
@@ -2688,7 +2690,8 @@ sp_proc_stmt_return:
                 MYSQL_YYABORT;
               sp->m_flags|= sp_head::HAS_RETURN;
             }
-            sp->restore_lex(YYTHD);
+            if (sp->restore_lex(YYTHD))
+              MYSQL_YYABORT;
           }
         ;
 
@@ -2928,7 +2931,8 @@ sp_if:
                 sp->add_cont_backpatch(i) ||
                 sp->add_instr(i))
               MYSQL_YYABORT;
-            sp->restore_lex(YYTHD);
+            if (sp->restore_lex(YYTHD))
+              MYSQL_YYABORT;
           }
           sp_proc_stmts1
           {
@@ -2974,7 +2978,9 @@ simple_case_stmt:
             if (case_stmt_action_expr(lex, $3))
               MYSQL_YYABORT;
 
-            lex->sphead->restore_lex(YYTHD); /* For expr $3 */
+            /* For expr $3 */
+            if (lex->sphead->restore_lex(YYTHD))
+              MYSQL_YYABORT;
           }
           simple_when_clause_list
           else_clause_opt
@@ -3024,7 +3030,9 @@ simple_when_clause:
             LEX *lex= Lex;
             if (case_stmt_action_when(lex, $3, true))
               MYSQL_YYABORT;
-            lex->sphead->restore_lex(YYTHD); /* For expr $3 */
+            /* For expr $3 */
+            if (lex->sphead->restore_lex(YYTHD))
+              MYSQL_YYABORT;
           }
           THEN_SYM
           sp_proc_stmts1
@@ -3045,7 +3053,9 @@ searched_when_clause:
             LEX *lex= Lex;
             if (case_stmt_action_when(lex, $3, false))
               MYSQL_YYABORT;
-            lex->sphead->restore_lex(YYTHD); /* For expr $3 */
+            /* For expr $3 */
+            if (lex->sphead->restore_lex(YYTHD))
+              MYSQL_YYABORT;
           }
           THEN_SYM
           sp_proc_stmts1
@@ -3222,7 +3232,8 @@ sp_unlabeled_control:
                 sp->new_cont_backpatch(i) ||
                 sp->add_instr(i))
               MYSQL_YYABORT;
-            sp->restore_lex(YYTHD);
+            if (sp->restore_lex(YYTHD))
+              MYSQL_YYABORT;
           }
           sp_proc_stmts1 END WHILE_SYM
           {
@@ -3248,7 +3259,8 @@ sp_unlabeled_control:
             if (i == NULL ||
                 lex->sphead->add_instr(i))
               MYSQL_YYABORT;
-            lex->sphead->restore_lex(YYTHD);
+            if (lex->sphead->restore_lex(YYTHD))
+              MYSQL_YYABORT;
             /* We can shortcut the cont_backpatch here */
             i->m_cont_dest= ip+1;
           }
@@ -7523,6 +7535,14 @@ function_call_nonkeyword:
           }
         | SYSDATE optional_braces
           {
+            /*
+              Unlike other time-related functions, SYSDATE() is
+              replication-unsafe because it is not affected by the
+              TIMESTAMP variable.  It is unsafe even if
+              sysdate_is_now=1, because the slave may have
+              sysdate_is_now=0.
+            */
+            Lex->set_stmt_unsafe();
             if (global_system_variables.sysdate_is_now == 0)
               $$= new (YYTHD->mem_root) Item_func_sysdate_local();
             else
@@ -11769,7 +11789,8 @@ option_type_value:
                 if (sp->add_instr(i))
                   MYSQL_YYABORT;
               }
-              lex->sphead->restore_lex(thd);
+              if (lex->sphead->restore_lex(thd))
+                MYSQL_YYABORT;
             }
           }
         ;
diff --git a/sql/table.cc b/sql/table.cc
index 45099c1a4fb..0ca81294458 100644
--- a/sql/table.cc
+++ b/sql/table.cc
@@ -1316,8 +1316,16 @@ static int open_binary_frm(THD *thd, TABLE_SHARE *share, uchar *head,
       share->timestamp_field_offset= i;
 
     if (use_hash)
-      (void) my_hash_insert(&share->name_hash,
-                            (uchar*) field_ptr); // never fail
+      if (my_hash_insert(&share->name_hash, (uchar*) field_ptr) )
+      {
+        /*
+          Set return code 8 here to indicate that an error has
+          occurred but that the error message already has been
+          sent (OOM).
+        */
+        error= 8; 
+        goto err;
+      }
   }
   *field_ptr=0;					// End marker
 
@@ -2804,34 +2812,38 @@ bool check_column_name(const char *name)
                   and such errors never reach the user.
 */
 
-my_bool
-table_check_intact(TABLE *table, const uint table_f_count,
-                   const TABLE_FIELD_W_TYPE *table_def)
+bool
+Table_check_intact::check(TABLE *table, const TABLE_FIELD_DEF *table_def)
 {
   uint i;
   my_bool error= FALSE;
-  my_bool fields_diff_count;
+  const TABLE_FIELD_TYPE *field_def= table_def->field;
   DBUG_ENTER("table_check_intact");
   DBUG_PRINT("info",("table: %s  expected_count: %d",
-                     table->alias, table_f_count));
+                     table->alias, table_def->count));
+
+  /* Whether the table definition has already been validated. */
+  if (table->s->table_field_def_cache == table_def)
+    DBUG_RETURN(FALSE);
 
-  fields_diff_count= (table->s->fields != table_f_count);
-  if (fields_diff_count)
+  if (table->s->fields != table_def->count)
   {
     DBUG_PRINT("info", ("Column count has changed, checking the definition"));
 
     /* previous MySQL version */
     if (MYSQL_VERSION_ID > table->s->mysql_version)
     {
-      sql_print_error(ER(ER_COL_COUNT_DOESNT_MATCH_PLEASE_UPDATE),
-                      table->alias, table_f_count, table->s->fields,
-                      table->s->mysql_version, MYSQL_VERSION_ID);
+      report_error(ER_COL_COUNT_DOESNT_MATCH_PLEASE_UPDATE,
+                   ER(ER_COL_COUNT_DOESNT_MATCH_PLEASE_UPDATE),
+                   table->alias, table_def->count, table->s->fields,
+                   table->s->mysql_version, MYSQL_VERSION_ID);
       DBUG_RETURN(TRUE);
     }
     else if (MYSQL_VERSION_ID == table->s->mysql_version)
     {
-      sql_print_error(ER(ER_COL_COUNT_DOESNT_MATCH_CORRUPTED), table->alias,
-                      table_f_count, table->s->fields);
+      report_error(ER_COL_COUNT_DOESNT_MATCH_CORRUPTED,
+                   ER(ER_COL_COUNT_DOESNT_MATCH_CORRUPTED), table->alias,
+                   table_def->count, table->s->fields);
       DBUG_RETURN(TRUE);
     }
     /*
@@ -2843,7 +2855,7 @@ table_check_intact(TABLE *table, const uint table_f_count,
     */
   }
   char buffer[STRING_BUFFER_USUAL_SIZE];
-  for (i=0 ; i < table_f_count; i++, table_def++)
+  for (i=0 ; i < table_def->count; i++, field_def++)
   {
     String sql_type(buffer, sizeof(buffer), system_charset_info);
     sql_type.length(0);
@@ -2851,18 +2863,18 @@ table_check_intact(TABLE *table, const uint table_f_count,
     {
       Field *field= table->field[i];
 
-      if (strncmp(field->field_name, table_def->name.str,
-                  table_def->name.length))
+      if (strncmp(field->field_name, field_def->name.str,
+                  field_def->name.length))
       {
         /*
           Name changes are not fatal, we use ordinal numbers to access columns.
           Still this can be a sign of a tampered table, output an error
           to the error log.
         */
-        sql_print_error("Incorrect definition of table %s.%s: "
-                        "expected column '%s' at position %d, found '%s'.",
-                        table->s->db.str, table->alias, table_def->name.str, i,
-                        field->field_name);
+        report_error(0, "Incorrect definition of table %s.%s: "
+                     "expected column '%s' at position %d, found '%s'.",
+                     table->s->db.str, table->alias, field_def->name.str, i,
+                     field->field_name);
       }
       field->sql_type(sql_type);
       /*
@@ -2882,47 +2894,51 @@ table_check_intact(TABLE *table, const uint table_f_count,
         the new table definition is backward compatible with the
         original one.
        */
-      if (strncmp(sql_type.c_ptr_safe(), table_def->type.str,
-                  table_def->type.length - 1))
+      if (strncmp(sql_type.c_ptr_safe(), field_def->type.str,
+                  field_def->type.length - 1))
       {
-        sql_print_error("Incorrect definition of table %s.%s: "
-                        "expected column '%s' at position %d to have type "
-                        "%s, found type %s.", table->s->db.str, table->alias,
-                        table_def->name.str, i, table_def->type.str,
-                        sql_type.c_ptr_safe());
+        report_error(0, "Incorrect definition of table %s.%s: "
+                     "expected column '%s' at position %d to have type "
+                     "%s, found type %s.", table->s->db.str, table->alias,
+                     field_def->name.str, i, field_def->type.str,
+                     sql_type.c_ptr_safe());
         error= TRUE;
       }
-      else if (table_def->cset.str && !field->has_charset())
+      else if (field_def->cset.str && !field->has_charset())
       {
-        sql_print_error("Incorrect definition of table %s.%s: "
-                        "expected the type of column '%s' at position %d "
-                        "to have character set '%s' but the type has no "
-                        "character set.", table->s->db.str, table->alias,
-                        table_def->name.str, i, table_def->cset.str);
+        report_error(0, "Incorrect definition of table %s.%s: "
+                     "expected the type of column '%s' at position %d "
+                     "to have character set '%s' but the type has no "
+                     "character set.", table->s->db.str, table->alias,
+                     field_def->name.str, i, field_def->cset.str);
         error= TRUE;
       }
-      else if (table_def->cset.str &&
-               strcmp(field->charset()->csname, table_def->cset.str))
+      else if (field_def->cset.str &&
+               strcmp(field->charset()->csname, field_def->cset.str))
       {
-        sql_print_error("Incorrect definition of table %s.%s: "
-                        "expected the type of column '%s' at position %d "
-                        "to have character set '%s' but found "
-                        "character set '%s'.", table->s->db.str, table->alias,
-                        table_def->name.str, i, table_def->cset.str,
-                        field->charset()->csname);
+        report_error(0, "Incorrect definition of table %s.%s: "
+                     "expected the type of column '%s' at position %d "
+                     "to have character set '%s' but found "
+                     "character set '%s'.", table->s->db.str, table->alias,
+                     field_def->name.str, i, field_def->cset.str,
+                     field->charset()->csname);
         error= TRUE;
       }
     }
     else
     {
-      sql_print_error("Incorrect definition of table %s.%s: "
-                      "expected column '%s' at position %d to have type %s "
-                      " but the column is not found.",
-                      table->s->db.str, table->alias,
-                      table_def->name.str, i, table_def->type.str);
+      report_error(0, "Incorrect definition of table %s.%s: "
+                   "expected column '%s' at position %d to have type %s "
+                   " but the column is not found.",
+                   table->s->db.str, table->alias,
+                   field_def->name.str, i, field_def->type.str);
       error= TRUE;
     }
   }
+
+  if (! error)
+    table->s->table_field_def_cache= table_def;
+
   DBUG_RETURN(error);
 }
 
diff --git a/sql/table.h b/sql/table.h
index e4a382c799f..eae261cc97d 100644
--- a/sql/table.h
+++ b/sql/table.h
@@ -285,6 +285,36 @@ typedef enum enum_table_category TABLE_CATEGORY;
 TABLE_CATEGORY get_table_category(const LEX_STRING *db,
                                   const LEX_STRING *name);
 
+
+typedef struct st_table_field_type
+{
+  LEX_STRING name;
+  LEX_STRING type;
+  LEX_STRING cset;
+} TABLE_FIELD_TYPE;
+
+
+typedef struct st_table_field_def
+{
+  uint count;
+  const TABLE_FIELD_TYPE *field;
+} TABLE_FIELD_DEF;
+
+
+class Table_check_intact
+{
+protected:
+  virtual void report_error(uint code, const char *fmt, ...)= 0;
+
+public:
+  Table_check_intact() {}
+  virtual ~Table_check_intact() {}
+
+  /** Checks whether a table is intact. */
+  bool check(TABLE *table, const TABLE_FIELD_DEF *table_def);
+};
+
+
 /*
   This structure is shared between different table objects. There is one
   instance of table share per one table in the database.
@@ -421,6 +451,18 @@ typedef struct st_table_share
   handlerton *default_part_db_type;
 #endif
 
+  /**
+    Cache the checked structure of this table.
+
+    The pointer data is used to describe the structure that
+    a instance of the table must have. Each element of the
+    array specifies a field that must exist on the table.
+
+    The pointer is cached in order to perform the check only
+    once -- when the table is loaded from the disk.
+  */
+  const TABLE_FIELD_DEF *table_field_def_cache;
+
   /** place to store storage engine specific data */
   void *ha_data;
 
@@ -1662,17 +1704,6 @@ typedef struct st_open_table_list{
   uint32 in_use,locked;
 } OPEN_TABLE_LIST;
 
-typedef struct st_table_field_w_type
-{
-  LEX_STRING name;
-  LEX_STRING type;
-  LEX_STRING cset;
-} TABLE_FIELD_W_TYPE;
-
-
-my_bool
-table_check_intact(TABLE *table, const uint table_f_count,
-                   const TABLE_FIELD_W_TYPE *table_def);
 
 static inline my_bitmap_map *tmp_use_all_columns(TABLE *table,
                                                  MY_BITMAP *bitmap)