Merge tsmith@bk-internal.mysql.com:/home/bk/mysql-5.1-opt

into  ramayana.hindu.god:/home/tsmith/m/bk/maint/51


mysql-test/r/show_check.result:
  Auto merged
mysql-test/t/disabled.def:
  Auto merged
mysql-test/t/show_check.test:
  Auto merged
sql/handler.cc:
  Auto merged
sql/handler.h:
  Auto merged
sql/log.cc:
  Auto merged
sql/set_var.cc:
  Auto merged
sql/sql_base.cc:
  Auto merged
sql/sql_class.cc:
  Auto merged
sql/sql_class.h:
  Auto merged
sql/sql_delete.cc:
  Auto merged
sql/sql_insert.cc:
  Auto merged
sql/sql_parse.cc:
  Auto merged
sql/sql_select.cc:
  Auto merged
sql/sql_show.cc:
  Auto merged
sql/sql_table.cc:
  Auto merged
sql/sql_yacc.yy:
  Auto merged
sql/table.cc:
  Auto merged

23 files changed, 328 insertions, 191 deletions

diff --git a/sql/ha_ndbcluster.cc b/sql/ha_ndbcluster.cc
index dd29eb17059..48712600e79 100644
--- a/sql/ha_ndbcluster.cc
+++ b/sql/ha_ndbcluster.cc
@@ -4366,7 +4366,7 @@ int ha_ndbcluster::external_lock(THD *thd, int lock_type)
     {
       m_transaction_on= FALSE;
       /* Would be simpler if has_transactions() didn't always say "yes" */
-      thd->no_trans_update.all= thd->no_trans_update.stmt= TRUE;
+      thd->transaction.all.modified_non_trans_table= thd->transaction.stmt.modified_non_trans_table= TRUE;
     }
     else if (!thd->transaction.on)
       m_transaction_on= FALSE;
diff --git a/sql/handler.cc b/sql/handler.cc
index f06ad282efa..54ae258d560 100644
--- a/sql/handler.cc
+++ b/sql/handler.cc
@@ -860,7 +860,7 @@ int ha_rollback_trans(THD *thd, bool all)
     the error log; but we don't want users to wonder why they have this
     message in the error log, so we don't send it.
   */
-  if (is_real_trans && thd->no_trans_update.all &&
+  if (is_real_trans && thd->transaction.all.modified_non_trans_table &&
       !thd->slave_thread && thd->killed != THD::KILL_CONNECTION)
     push_warning(thd, MYSQL_ERROR::WARN_LEVEL_WARN,
                  ER_WARNING_NOT_COMPLETE_ROLLBACK,
diff --git a/sql/handler.h b/sql/handler.h
index 7cf7bfe043d..ddea6c3a34c 100644
--- a/sql/handler.h
+++ b/sql/handler.h
@@ -728,6 +728,35 @@ typedef struct st_thd_trans
   bool        no_2pc;
   /* storage engines that registered themselves for this transaction */
   handlerton *ht[MAX_HA];
+  /* 
+    The purpose of this flag is to keep track of non-transactional
+    tables that were modified in scope of:
+    - transaction, when the variable is a member of
+    THD::transaction.all
+    - top-level statement or sub-statement, when the variable is a
+    member of THD::transaction.stmt
+    This member has the following life cycle:
+    * stmt.modified_non_trans_table is used to keep track of
+    modified non-transactional tables of top-level statements. At
+    the end of the previous statement and at the beginning of the session,
+    it is reset to FALSE.  If such functions
+    as mysql_insert, mysql_update, mysql_delete etc modify a
+    non-transactional table, they set this flag to TRUE.  At the
+    end of the statement, the value of stmt.modified_non_trans_table 
+    is merged with all.modified_non_trans_table and gets reset.
+    * all.modified_non_trans_table is reset at the end of transaction
+    
+    * Since we do not have a dedicated context for execution of a
+    sub-statement, to keep track of non-transactional changes in a
+    sub-statement, we re-use stmt.modified_non_trans_table. 
+    At entrance into a sub-statement, a copy of the value of
+    stmt.modified_non_trans_table (containing the changes of the
+    outer statement) is saved on stack. Then 
+    stmt.modified_non_trans_table is reset to FALSE and the
+    substatement is executed. Then the new value is merged with the
+    saved value.
+  */
+  bool modified_non_trans_table;
 } THD_TRANS;
 
 enum enum_tx_isolation { ISO_READ_UNCOMMITTED, ISO_READ_COMMITTED,
diff --git a/sql/item.cc b/sql/item.cc
index 711a21ecbec..593915ef172 100644
--- a/sql/item.cc
+++ b/sql/item.cc
@@ -334,6 +334,37 @@ int Item::save_date_in_field(Field *field)
 }
 
 
+/*
+  Store the string value in field directly
+
+  SYNOPSIS
+    Item::save_str_value_in_field()
+    field   a pointer to field where to store
+    result  the pointer to the string value to be stored
+
+  DESCRIPTION
+    The method is used by Item_*::save_in_field implementations
+    when we don't need to calculate the value to store
+    See Item_string::save_in_field() implementation for example
+
+  IMPLEMENTATION
+    Check if the Item is null and stores the NULL or the
+    result value in the field accordingly.
+
+  RETURN
+    Nonzero value if error
+*/
+
+int Item::save_str_value_in_field(Field *field, String *result)
+{
+  if (null_value)
+    return set_field_to_null(field);
+  field->set_notnull();
+  return field->store(result->ptr(), result->length(),
+		      collation.collation);
+}
+
+
 Item::Item():
   rsize(0), name(0), orig_name(0), name_length(0), fixed(0),
   is_autogenerated_name(TRUE),
@@ -1051,9 +1082,9 @@ bool Item_sp_variable::is_null()
 Item_splocal::Item_splocal(const LEX_STRING &sp_var_name,
                            uint sp_var_idx,
                            enum_field_types sp_var_type,
-                           uint pos_in_q)
+                           uint pos_in_q, uint len_in_q)
   :Item_sp_variable(sp_var_name.str, sp_var_name.length),
-   m_var_idx(sp_var_idx), pos_in_query(pos_in_q)
+   m_var_idx(sp_var_idx), pos_in_query(pos_in_q), len_in_query(len_in_q)
 {
   maybe_null= TRUE;
 
@@ -3046,16 +3077,6 @@ my_decimal *Item_copy_string::val_decimal(my_decimal *decimal_value)
 }
 
 
-
-int Item_copy_string::save_in_field(Field *field, bool no_conversions)
-{
-  if (null_value)
-    return set_field_to_null(field);
-  field->set_notnull();
-  return field->store(str_value.ptr(),str_value.length(),
-		      collation.collation);
-}
-
 /*
   Functions to convert item to field (for send_fields)
 */
@@ -4508,6 +4529,12 @@ int Item_null::save_safe_in_field(Field *field)
 }
 
 
+/*
+  This implementation can lose str_value content, so if the
+  Item uses str_value to store something, it should
+  reimplement it's ::save_in_field() as Item_string, for example, does
+*/
+
 int Item::save_in_field(Field *field, bool no_conversions)
 {
   int error;
@@ -4565,10 +4592,7 @@ int Item_string::save_in_field(Field *field, bool no_conversions)
 {
   String *result;
   result=val_str(&str_value);
-  if (null_value)
-    return set_field_to_null(field);
-  field->set_notnull();
-  return field->store(result->ptr(),result->length(),collation.collation);
+  return save_str_value_in_field(field, result);
 }
 
 
diff --git a/sql/item.h b/sql/item.h
index 432da6c3a1c..ab39505bf5f 100644
--- a/sql/item.h
+++ b/sql/item.h
@@ -653,6 +653,7 @@ public:
 
   int save_time_in_field(Field *field);
   int save_date_in_field(Field *field);
+  int save_str_value_in_field(Field *field, String *result);
 
   virtual Field *get_tmp_table_field() { return 0; }
   /* This is also used to create fields in CREATE ... SELECT: */
@@ -1053,9 +1054,18 @@ public:
     SP variable in query text.
   */
   uint pos_in_query;
+  /*
+    Byte length of SP variable name in the statement (see pos_in_query).
+    The value of this field may differ from the name_length value because
+    name_length contains byte length of UTF8-encoded item name, but
+    the query string (see sp_instr_stmt::m_query) is currently stored with
+    a charset from the SET NAMES statement.
+  */
+  uint len_in_query;
 
   Item_splocal(const LEX_STRING &sp_var_name, uint sp_var_idx,
-               enum_field_types sp_var_type, uint pos_in_q= 0);
+               enum_field_types sp_var_type,
+               uint pos_in_q= 0, uint len_in_q= 0);
 
   bool is_splocal() { return 1; } /* Needed for error checking */
 
@@ -2293,7 +2303,10 @@ public:
   my_decimal *val_decimal(my_decimal *);
   void make_field(Send_field *field) { item->make_field(field); }
   void copy();
-  int save_in_field(Field *field, bool no_conversions);
+  int save_in_field(Field *field, bool no_conversions)
+  {
+    return save_str_value_in_field(field, &str_value);
+  }
   table_map used_tables() const { return (table_map) 1L; }
   bool const_item() const { return 0; }
   bool is_null() { return null_value; }
diff --git a/sql/item_strfunc.h b/sql/item_strfunc.h
index 6d2d9c199c9..ea9517976a8 100644
--- a/sql/item_strfunc.h
+++ b/sql/item_strfunc.h
@@ -427,6 +427,10 @@ public:
   }
   const char *func_name() const { return "user"; }
   const char *fully_qualified_func_name() const { return "user()"; }
+  int save_in_field(Field *field, bool no_conversions)
+  {
+    return save_str_value_in_field(field, &str_value);
+  }
 };
 
 
diff --git a/sql/log.cc b/sql/log.cc
index dbb664c9d18..00ef726f037 100644
--- a/sql/log.cc
+++ b/sql/log.cc
@@ -1462,7 +1462,8 @@ static int binlog_rollback(handlerton *hton, THD *thd, bool all)
     table. Such cases should be rare (updating a
     non-transactional table inside a transaction...)
   */
-  if (unlikely(thd->no_trans_update.all || (thd->options & OPTION_KEEP_LOG)))
+  if (unlikely(thd->transaction.all.modified_non_trans_table || 
+               (thd->options & OPTION_KEEP_LOG)))
   {
     Query_log_event qev(thd, STRING_WITH_LEN("ROLLBACK"), TRUE, FALSE);
     qev.error_code= 0; // see comment in MYSQL_LOG::write(THD, IO_CACHE)
@@ -1516,7 +1517,8 @@ static int binlog_savepoint_rollback(handlerton *hton, THD *thd, void *sv)
     non-transactional table. Otherwise, truncate the binlog cache starting
     from the SAVEPOINT command.
   */
-  if (unlikely(thd->no_trans_update.all || (thd->options & OPTION_KEEP_LOG)))
+  if (unlikely(thd->transaction.all.modified_non_trans_table || 
+               (thd->options & OPTION_KEEP_LOG)))
   {
     int error=
       thd->binlog_query(THD::STMT_QUERY_TYPE,
@@ -3957,6 +3959,7 @@ bool MYSQL_BIN_LOG::write(THD *thd, IO_CACHE *cache, Log_event *commit_event)
   /* NULL would represent nothing to replicate after ROLLBACK */
   DBUG_ASSERT(commit_event != NULL);
 
+  DBUG_ASSERT(is_open());
   if (likely(is_open()))                       // Should always be true
   {
     /*
diff --git a/sql/set_var.cc b/sql/set_var.cc
index d21a4c18716..a18d45627dc 100644
--- a/sql/set_var.cc
+++ b/sql/set_var.cc
@@ -2558,14 +2558,14 @@ static bool set_option_autocommit(THD *thd, set_var *var)
     {
       /* We changed to auto_commit mode */
       thd->options&= ~(ulonglong) (OPTION_BEGIN | OPTION_KEEP_LOG);
-      thd->no_trans_update.all= FALSE;
+      thd->transaction.all.modified_non_trans_table= FALSE;
       thd->server_status|= SERVER_STATUS_AUTOCOMMIT;
       if (ha_commit(thd))
 	return 1;
     }
     else
     {
-      thd->no_trans_update.all= FALSE;
+      thd->transaction.all.modified_non_trans_table= FALSE;
       thd->server_status&= ~SERVER_STATUS_AUTOCOMMIT;
     }
   }
diff --git a/sql/sp_head.cc b/sql/sp_head.cc
index 9b67a89bed2..7b2073b8de3 100644
--- a/sql/sp_head.cc
+++ b/sql/sp_head.cc
@@ -349,13 +349,13 @@ sp_eval_expr(THD *thd, Field *result_field, Item **expr_item_ptr)
   
   enum_check_fields save_count_cuted_fields= thd->count_cuted_fields;
   bool save_abort_on_warning= thd->abort_on_warning;
-  bool save_no_trans_update_stmt= thd->no_trans_update.stmt;
+  bool save_stmt_modified_non_trans_table= thd->transaction.stmt.modified_non_trans_table;
 
   thd->count_cuted_fields= CHECK_FIELD_ERROR_FOR_NULL;
   thd->abort_on_warning=
     thd->variables.sql_mode &
     (MODE_STRICT_TRANS_TABLES | MODE_STRICT_ALL_TABLES);
-  thd->no_trans_update.stmt= FALSE;
+  thd->transaction.stmt.modified_non_trans_table= FALSE;
 
   /* Save the value in the field. Convert the value if needed. */
 
@@ -363,7 +363,7 @@ sp_eval_expr(THD *thd, Field *result_field, Item **expr_item_ptr)
 
   thd->count_cuted_fields= save_count_cuted_fields;
   thd->abort_on_warning= save_abort_on_warning;
-  thd->no_trans_update.stmt= save_no_trans_update_stmt;
+  thd->transaction.stmt.modified_non_trans_table= save_stmt_modified_non_trans_table;
 
   if (thd->net.report_error)
   {
@@ -858,7 +858,8 @@ int cmp_splocal_locations(Item_splocal * const *a, Item_splocal * const *b)
 
 
 /*
-  Replace thd->query{_length} with a string that one can write to the binlog.
+  Replace thd->query{_length} with a string that one can write to the binlog
+  or the query cache.
  
   SYNOPSIS
     subst_spvars()
@@ -870,7 +871,9 @@ int cmp_splocal_locations(Item_splocal * const *a, Item_splocal * const *b)
   DESCRIPTION
 
   The binlog-suitable string is produced by replacing references to SP local 
-  variables with NAME_CONST('sp_var_name', value) calls.
+  variables with NAME_CONST('sp_var_name', value) calls. To make this string
+  suitable for the query cache this function allocates some additional space
+  for the query cache flags.
  
   RETURN
     FALSE  on success
@@ -883,80 +886,89 @@ static bool
 subst_spvars(THD *thd, sp_instr *instr, LEX_STRING *query_str)
 {
   DBUG_ENTER("subst_spvars");
-  if (thd->prelocked_mode == NON_PRELOCKED && mysql_bin_log.is_open())
-  {
-    Dynamic_array<Item_splocal*> sp_vars_uses;
-    char *pbuf, *cur, buffer[512];
-    String qbuf(buffer, sizeof(buffer), &my_charset_bin);
-    int prev_pos, res;
 
-    /* Find all instances of Item_splocal used in this statement */
-    for (Item *item= instr->free_list; item; item= item->next)
-    {
-      if (item->is_splocal())
-      {
-        Item_splocal *item_spl= (Item_splocal*)item;
-        if (item_spl->pos_in_query)
-          sp_vars_uses.append(item_spl);
-      }
-    }
-    if (!sp_vars_uses.elements())
-      DBUG_RETURN(FALSE);
-      
-    /* Sort SP var refs by their occurences in the query */
-    sp_vars_uses.sort(cmp_splocal_locations);
+  Dynamic_array<Item_splocal*> sp_vars_uses;
+  char *pbuf, *cur, buffer[512];
+  String qbuf(buffer, sizeof(buffer), &my_charset_bin);
+  int prev_pos, res, buf_len;
 
-    /* 
-      Construct a statement string where SP local var refs are replaced
-      with "NAME_CONST(name, value)"
-    */
-    qbuf.length(0);
-    cur= query_str->str;
-    prev_pos= res= 0;
-    for (Item_splocal **splocal= sp_vars_uses.front(); 
-         splocal < sp_vars_uses.back(); splocal++)
+  /* Find all instances of Item_splocal used in this statement */
+  for (Item *item= instr->free_list; item; item= item->next)
+  {
+    if (item->is_splocal())
     {
-      Item *val;
-
-      char str_buffer[STRING_BUFFER_USUAL_SIZE];
-      String str_value_holder(str_buffer, sizeof(str_buffer),
-                              &my_charset_latin1);
-      String *str_value;
-      
-      /* append the text between sp ref occurences */
-      res|= qbuf.append(cur + prev_pos, (*splocal)->pos_in_query - prev_pos);
-      prev_pos= (*splocal)->pos_in_query + (*splocal)->m_name.length;
-      
-      /* append the spvar substitute */
-      res|= qbuf.append(STRING_WITH_LEN(" NAME_CONST('"));
-      res|= qbuf.append((*splocal)->m_name.str, (*splocal)->m_name.length);
-      res|= qbuf.append(STRING_WITH_LEN("',"));
-      res|= (*splocal)->fix_fields(thd, (Item **) splocal);
+      Item_splocal *item_spl= (Item_splocal*)item;
+      if (item_spl->pos_in_query)
+        sp_vars_uses.append(item_spl);
+    }
+  }
+  if (!sp_vars_uses.elements())
+    DBUG_RETURN(FALSE);
+    
+  /* Sort SP var refs by their occurences in the query */
+  sp_vars_uses.sort(cmp_splocal_locations);
 
-      if (res)
-        break;
+  /* 
+    Construct a statement string where SP local var refs are replaced
+    with "NAME_CONST(name, value)"
+  */
+  qbuf.length(0);
+  cur= query_str->str;
+  prev_pos= res= 0;
+  for (Item_splocal **splocal= sp_vars_uses.front(); 
+       splocal < sp_vars_uses.back(); splocal++)
+  {
+    Item *val;
+
+    char str_buffer[STRING_BUFFER_USUAL_SIZE];
+    String str_value_holder(str_buffer, sizeof(str_buffer),
+                            &my_charset_latin1);
+    String *str_value;
+    
+    /* append the text between sp ref occurences */
+    res|= qbuf.append(cur + prev_pos, (*splocal)->pos_in_query - prev_pos);
+    prev_pos= (*splocal)->pos_in_query + (*splocal)->len_in_query;
+    
+    /* append the spvar substitute */
+    res|= qbuf.append(STRING_WITH_LEN(" NAME_CONST('"));
+    res|= qbuf.append((*splocal)->m_name.str, (*splocal)->m_name.length);
+    res|= qbuf.append(STRING_WITH_LEN("',"));
+    res|= (*splocal)->fix_fields(thd, (Item **) splocal);
 
-      val= (*splocal)->this_item();
-      DBUG_PRINT("info", ("print 0x%lx", (long) val));
-      str_value= sp_get_item_value(thd, val, &str_value_holder);
-      if (str_value)
-        res|= qbuf.append(*str_value);
-      else
-        res|= qbuf.append(STRING_WITH_LEN("NULL"));
-      res|= qbuf.append(')');
-      if (res)
-        break;
-    }
-    res|= qbuf.append(cur + prev_pos, query_str->length - prev_pos);
     if (res)
-      DBUG_RETURN(TRUE);
+      break;
 
-    if (!(pbuf= thd->strmake(qbuf.ptr(), qbuf.length())))
-      DBUG_RETURN(TRUE);
+    val= (*splocal)->this_item();
+    DBUG_PRINT("info", ("print 0x%lx", (long) val));
+    str_value= sp_get_item_value(thd, val, &str_value_holder);
+    if (str_value)
+      res|= qbuf.append(*str_value);
+    else
+      res|= qbuf.append(STRING_WITH_LEN("NULL"));
+    res|= qbuf.append(')');
+    if (res)
+      break;
+  }
+  res|= qbuf.append(cur + prev_pos, query_str->length - prev_pos);
+  if (res)
+    DBUG_RETURN(TRUE);
 
-    thd->query= pbuf;
-    thd->query_length= qbuf.length();
+  /*
+    Allocate additional space at the end of the new query string for the
+    query_cache_send_result_to_client function.
+  */
+  buf_len= qbuf.length() + thd->db_length + 1 + QUERY_CACHE_FLAGS_SIZE + 1;
+  if ((pbuf= (char *) alloc_root(thd->mem_root, buf_len)))
+  {
+    memcpy(pbuf, qbuf.ptr(), qbuf.length());
+    pbuf[qbuf.length()]= 0;
   }
+  else
+    DBUG_RETURN(TRUE);
+
+  thd->query= pbuf;
+  thd->query_length= qbuf.length();
+
   DBUG_RETURN(FALSE);
 }
 
@@ -2535,6 +2547,13 @@ sp_lex_keeper::reset_lex_and_exec_core(THD *thd, uint *nextp,
   int res= 0;
   DBUG_ENTER("reset_lex_and_exec_core");
 
+  /* 
+    The flag is saved at the entry to the following substatement.
+    It's reset further in the common code part.
+    It's merged with the saved parent's value at the exit of this func.
+  */
+  bool parent_modified_non_trans_table= thd->transaction.stmt.modified_non_trans_table;
+  thd->transaction.stmt.modified_non_trans_table= FALSE;
   DBUG_ASSERT(!thd->derived_tables);
   DBUG_ASSERT(thd->change_list.is_empty());
   /*
@@ -2604,7 +2623,11 @@ sp_lex_keeper::reset_lex_and_exec_core(THD *thd, uint *nextp,
   /* Update the state of the active arena. */
   thd->stmt_arena->state= Query_arena::EXECUTED;
 
-
+  /*
+    Merge here with the saved parent's values
+    what is needed from the substatement gained
+  */
+  thd->transaction.stmt.modified_non_trans_table |= parent_modified_non_trans_table;
   /*
     Unlike for PS we should not call Item's destructors for newly created
     items after execution of each instruction in stored routine. This is
diff --git a/sql/sql_base.cc b/sql/sql_base.cc
index 4e62830a903..3e594d4da4b 100644
--- a/sql/sql_base.cc
+++ b/sql/sql_base.cc
@@ -4514,8 +4514,7 @@ find_field_in_view(THD *thd, TABLE_LIST *table_list,
               table_list->alias, name, item_name, (ulong) ref));
   Field_iterator_view field_it;
   field_it.set(table_list);
-  Query_arena *arena, backup;  
-  LINT_INIT(arena);
+  Query_arena *arena= 0, backup;  
   
   DBUG_ASSERT(table_list->schema_table_reformed ||
               (ref != 0 && table_list->view != 0));
@@ -4524,14 +4523,14 @@ find_field_in_view(THD *thd, TABLE_LIST *table_list,
     if (!my_strcasecmp(system_charset_info, field_it.name(), name))
     {
       // in PS use own arena or data will be freed after prepare
-      if (register_tree_change)
+      if (register_tree_change && thd->stmt_arena->is_stmt_prepare_or_first_sp_execute())
         arena= thd->activate_stmt_arena_if_needed(&backup);
       /*
         create_item() may, or may not create a new Item, depending on
         the column reference. See create_view_field() for details.
       */
       Item *item= field_it.create_item(thd);
-      if (register_tree_change && arena)
+      if (arena)
         thd->restore_active_arena(arena, &backup);
       
       if (!item)
diff --git a/sql/sql_class.cc b/sql/sql_class.cc
index a9a5e10a996..ba9d832dfa9 100644
--- a/sql/sql_class.cc
+++ b/sql/sql_class.cc
@@ -583,7 +583,7 @@ void THD::init(void)
   if (variables.sql_mode & MODE_NO_BACKSLASH_ESCAPES)
     server_status|= SERVER_STATUS_NO_BACKSLASH_ESCAPES;
   options= thd_startup_options;
-  no_trans_update.stmt= no_trans_update.all= FALSE;
+  transaction.all.modified_non_trans_table= transaction.stmt.modified_non_trans_table= FALSE;
   open_options=ha_open_options;
   update_lock_default= (variables.low_priority_updates ?
 			TL_WRITE_LOW_PRIORITY :
@@ -2574,7 +2574,7 @@ extern "C" int thd_slave_thread(const MYSQL_THD thd)
 
 extern "C" int thd_non_transactional_update(const MYSQL_THD thd)
 {
-  return(thd->no_trans_update.all);
+  return(thd->transaction.all.modified_non_trans_table);
 }
 
 extern "C" int thd_binlog_format(const MYSQL_THD thd)
diff --git a/sql/sql_class.h b/sql/sql_class.h
index 18ee426ae9c..00b68420f56 100644
--- a/sql/sql_class.h
+++ b/sql/sql_class.h
@@ -1423,10 +1423,6 @@ public:
   bool	     charset_is_system_charset, charset_is_collation_connection;
   bool       charset_is_character_set_filesystem;
   bool       enable_slow_log;   /* enable slow log for current statement */
-  struct {
-    bool all:1;
-    bool stmt:1;
-  } no_trans_update;
   bool	     abort_on_warning;
   bool 	     got_warning;       /* Set on call to push_warning() */
   bool	     no_warnings_for_error; /* no warnings on call to my_error() */
@@ -1713,7 +1709,7 @@ public:
   inline bool really_abort_on_warning()
   {
     return (abort_on_warning &&
-            (!no_trans_update.stmt ||
+            (!transaction.stmt.modified_non_trans_table ||
              (variables.sql_mode & MODE_STRICT_ALL_TABLES)));
   }
   void set_status_var_init();
diff --git a/sql/sql_delete.cc b/sql/sql_delete.cc
index 7d8fc119cf9..37f1325228d 100644
--- a/sql/sql_delete.cc
+++ b/sql/sql_delete.cc
@@ -341,6 +341,9 @@ cleanup:
   delete select;
   transactional_table= table->file->has_transactions();
 
+  if (!transactional_table && deleted > 0)
+    thd->transaction.stmt.modified_non_trans_table= TRUE;
+  
   /* See similar binlogging code in sql_update.cc, for comments */
   if ((error < 0) || (deleted && !transactional_table))
   {
@@ -364,9 +367,10 @@ cleanup:
 	error=1;
       }
     }
-    if (!transactional_table)
-      thd->no_trans_update.all= TRUE;
+    if (thd->transaction.stmt.modified_non_trans_table)
+      thd->transaction.all.modified_non_trans_table= TRUE;
   }
+  DBUG_ASSERT(transactional_table || !deleted || thd->transaction.stmt.modified_non_trans_table);
   free_underlaid_joins(thd, select_lex);
   if (transactional_table)
   {
@@ -677,20 +681,22 @@ bool multi_delete::send_data(List<Item> &values)
       if (table->triggers &&
           table->triggers->process_triggers(thd, TRG_EVENT_DELETE,
                                             TRG_ACTION_BEFORE, FALSE))
-	DBUG_RETURN(1);
+        DBUG_RETURN(1);
       table->status|= STATUS_DELETED;
       if (!(error=table->file->ha_delete_row(table->record[0])))
       {
-	deleted++;
+        deleted++;
+        if (!table->file->has_transactions())
+          thd->transaction.stmt.modified_non_trans_table= TRUE;
         if (table->triggers &&
             table->triggers->process_triggers(thd, TRG_EVENT_DELETE,
                                               TRG_ACTION_AFTER, FALSE))
-	  DBUG_RETURN(1);
+          DBUG_RETURN(1);
       }
       else
       {
-	table->file->print_error(error,MYF(0));
-	DBUG_RETURN(1);
+        table->file->print_error(error,MYF(0));
+        DBUG_RETURN(1);
       }
     }
     else
@@ -740,6 +746,7 @@ void multi_delete::send_error(uint errcode,const char *err)
     error= 1;
     send_eof();
   }
+  DBUG_ASSERT(!normal_tables || !deleted || thd->transaction.stmt.modified_non_trans_table);
   DBUG_VOID_RETURN;
 }
 
@@ -768,6 +775,7 @@ int multi_delete::do_deletes()
   for (; table_being_deleted;
        table_being_deleted= table_being_deleted->next_local, counter++)
   { 
+    ha_rows last_deleted= deleted;
     TABLE *table = table_being_deleted->table;
     if (tempfiles[counter]->get(table))
     {
@@ -814,6 +822,8 @@ int multi_delete::do_deletes()
         table->file->print_error(local_error,MYF(0));
       }
     }
+    if (last_deleted != deleted && !table->file->has_transactions())
+      thd->transaction.stmt.modified_non_trans_table= TRUE;
     end_read_record(&info);
     if (thd->killed && !local_error)
       local_error= 1;
@@ -852,7 +862,6 @@ bool multi_delete::send_eof()
   {
     query_cache_invalidate3(thd, delete_tables, 1);
   }
-
   if ((local_error == 0) || (deleted && normal_tables))
   {
     if (mysql_bin_log.is_open())
@@ -867,9 +876,11 @@ bool multi_delete::send_eof()
 	local_error=1;  // Log write failed: roll back the SQL statement
       }
     }
-    if (!transactional_tables)
-      thd->no_trans_update.all= TRUE;
+    if (thd->transaction.stmt.modified_non_trans_table)
+      thd->transaction.all.modified_non_trans_table= TRUE;
   }
+  DBUG_ASSERT(!normal_tables || !deleted || thd->transaction.stmt.modified_non_trans_table);
+
   /* Commit or rollback the current SQL statement */
   if (transactional_tables)
     if (ha_autocommit_or_rollback(thd,local_error > 0))
diff --git a/sql/sql_insert.cc b/sql/sql_insert.cc
index dc5db54e128..6c6a4721369 100644
--- a/sql/sql_insert.cc
+++ b/sql/sql_insert.cc
@@ -549,6 +549,7 @@ bool mysql_insert(THD *thd,TABLE_LIST *table_list,
   int error, res;
   bool transactional_table, joins_freed= FALSE;
   bool changed;
+  bool was_insert_delayed= (table_list->lock_type ==  TL_WRITE_DELAYED);
   uint value_count;
   ulong counter = 1;
   ulonglong id;
@@ -712,7 +713,6 @@ bool mysql_insert(THD *thd,TABLE_LIST *table_list,
   if (lock_type != TL_WRITE_DELAYED && !thd->prelocked_mode)
     table->file->ha_start_bulk_insert(values_list.elements);
 
-  thd->no_trans_update.stmt= FALSE;
   thd->abort_on_warning= (!ignore && (thd->variables.sql_mode &
                                        (MODE_STRICT_TRANS_TABLES |
                                         MODE_STRICT_ALL_TABLES)));
@@ -832,14 +832,16 @@ bool mysql_insert(THD *thd,TABLE_LIST *table_list,
     }
     transactional_table= table->file->has_transactions();
 
-    if ((changed= (info.copied || info.deleted || info.updated)))
+    if ((changed= (info.copied || info.deleted || info.updated)) ||
+        was_insert_delayed)
     {
       /*
         Invalidate the table in the query cache if something changed.
         For the transactional algorithm to work the invalidation must be
         before binlog writing and ha_autocommit_or_rollback
       */
-      query_cache_invalidate3(thd, table_list, 1);
+      if (changed)
+        query_cache_invalidate3(thd, table_list, 1);
       if (error <= 0 || !transactional_table)
       {
         if (mysql_bin_log.is_open())
@@ -880,10 +882,12 @@ bool mysql_insert(THD *thd,TABLE_LIST *table_list,
             error=1;
           }
         }
-        if (!transactional_table)
-          thd->no_trans_update.all= TRUE;
+        if (thd->transaction.stmt.modified_non_trans_table)
+          thd->transaction.all.modified_non_trans_table= TRUE;
       }
     }
+    DBUG_ASSERT(transactional_table || !changed || 
+                thd->transaction.stmt.modified_non_trans_table);
     if (transactional_table)
       error=ha_autocommit_or_rollback(thd,error);
     
@@ -1293,7 +1297,7 @@ static int last_uniq_key(TABLE *table,uint keynr)
     then both on update triggers will work instead. Similarly both on
     delete triggers will be invoked if we will delete conflicting records.
 
-    Sets thd->no_trans_update.stmt to TRUE if table which is updated didn't have
+    Sets thd->transaction.stmt.modified_non_trans_table to TRUE if table which is updated didn't have
     transactions.
 
   RETURN VALUE
@@ -1513,7 +1517,7 @@ int write_record(THD *thd, TABLE *table,COPY_INFO *info)
             goto err;
           info->deleted++;
           if (!table->file->has_transactions())
-            thd->no_trans_update.stmt= TRUE;
+            thd->transaction.stmt.modified_non_trans_table= TRUE;
           if (table->triggers &&
               table->triggers->process_triggers(thd, TRG_EVENT_DELETE,
                                                 TRG_ACTION_AFTER, TRUE))
@@ -1554,7 +1558,7 @@ ok_or_after_trg_err:
   if (key)
     my_safe_afree(key,table->s->max_unique_length,MAX_KEY_LENGTH);
   if (!table->file->has_transactions())
-    thd->no_trans_update.stmt= TRUE;
+    thd->transaction.stmt.modified_non_trans_table= TRUE;
   DBUG_RETURN(trg_error);
 
 err:
@@ -2919,7 +2923,6 @@ select_insert::prepare(List<Item> &values, SELECT_LEX_UNIT *u)
     table->file->extra(HA_EXTRA_WRITE_CAN_REPLACE);
   if (info.handle_duplicates == DUP_UPDATE)
     table->file->extra(HA_EXTRA_INSERT_WITH_UPDATE);
-  thd->no_trans_update.stmt= FALSE;
   thd->abort_on_warning= (!info.ignore &&
                           (thd->variables.sql_mode &
                            (MODE_STRICT_TRANS_TABLES |
@@ -3068,6 +3071,7 @@ bool select_insert::send_eof()
   int error;
   bool const trans_table= table->file->has_transactions();
   ulonglong id;
+  bool changed;
   DBUG_ENTER("select_insert::send_eof");
   DBUG_PRINT("enter", ("trans_table=%d, table_type='%s'",
                        trans_table, table->file->table_type()));
@@ -3076,24 +3080,18 @@ bool select_insert::send_eof()
   table->file->extra(HA_EXTRA_NO_IGNORE_DUP_KEY);
   table->file->extra(HA_EXTRA_WRITE_CANNOT_REPLACE);
 
-  if (info.copied || info.deleted || info.updated)
+  if (changed= (info.copied || info.deleted || info.updated))
   {
     /*
       We must invalidate the table in the query cache before binlog writing
       and ha_autocommit_or_rollback.
     */
     query_cache_invalidate3(thd, table, 1);
-    /*
-      Mark that we have done permanent changes if all of the below is true
-      - Table doesn't support transactions
-      - It's a normal (not temporary) table. (Changes to temporary tables
-        are not logged in RBR)
-      - We are using statement based replication
-    */
-    if (!trans_table &&
-        (!table->s->tmp_table || !thd->current_stmt_binlog_row_based))
-      thd->no_trans_update.all= TRUE;
-   }
+    if (thd->transaction.stmt.modified_non_trans_table)
+      thd->transaction.all.modified_non_trans_table= TRUE;
+  }
+  DBUG_ASSERT(trans_table || !changed || 
+              thd->transaction.stmt.modified_non_trans_table);
 
   /*
     Write to binlog before commiting transaction.  No statement will
@@ -3190,7 +3188,7 @@ void select_insert::abort() {
                             table->file->has_transactions(), FALSE);
         if (!thd->current_stmt_binlog_row_based && !table->s->tmp_table &&
             !can_rollback_data())
-          thd->no_trans_update.all= TRUE;
+          thd->transaction.all.modified_non_trans_table= TRUE;
         query_cache_invalidate3(thd, table, 1);
       }
     }
@@ -3514,7 +3512,6 @@ select_create::prepare(List<Item> &values, SELECT_LEX_UNIT *u)
     table->file->extra(HA_EXTRA_INSERT_WITH_UPDATE);
   if (!thd->prelocked_mode)
     table->file->ha_start_bulk_insert((ha_rows) 0);
-  thd->no_trans_update.stmt= FALSE;
   thd->abort_on_warning= (!info.ignore &&
                           (thd->variables.sql_mode &
                            (MODE_STRICT_TRANS_TABLES |
diff --git a/sql/sql_load.cc b/sql/sql_load.cc
index 3ffbdf83815..8bbe1e413b3 100644
--- a/sql/sql_load.cc
+++ b/sql/sql_load.cc
@@ -376,7 +376,6 @@ bool mysql_load(THD *thd,sql_exchange *ex,TABLE_LIST *table_list,
       table->file->ha_start_bulk_insert((ha_rows) 0);
     table->copy_blobs=1;
 
-    thd->no_trans_update.stmt= FALSE;
     thd->abort_on_warning= (!ignore &&
                             (thd->variables.sql_mode &
                              (MODE_STRICT_TRANS_TABLES |
@@ -410,7 +409,6 @@ bool mysql_load(THD *thd,sql_exchange *ex,TABLE_LIST *table_list,
     ha_autocommit_...
   */
   query_cache_invalidate3(thd, table_list, 0);
-
   if (error)
   {
     if (read_file_from_client)
@@ -469,8 +467,8 @@ bool mysql_load(THD *thd,sql_exchange *ex,TABLE_LIST *table_list,
   sprintf(name, ER(ER_LOAD_INFO), (ulong) info.records, (ulong) info.deleted,
 	  (ulong) (info.records - info.copied), (ulong) thd->cuted_fields);
 
-  if (!transactional_table)
-    thd->no_trans_update.all= TRUE;
+  if (thd->transaction.stmt.modified_non_trans_table)
+    thd->transaction.all.modified_non_trans_table= TRUE;
 #ifndef EMBEDDED_LIBRARY
   if (mysql_bin_log.is_open())
   {
@@ -506,6 +504,8 @@ bool mysql_load(THD *thd,sql_exchange *ex,TABLE_LIST *table_list,
   /* ok to client sent only after binlog write and engine commit */
   send_ok(thd, info.copied + info.deleted, 0L, name);
 err:
+  DBUG_ASSERT(transactional_table || !(info.copied || info.deleted) ||
+              thd->transaction.stmt.modified_non_trans_table);
   table->file->ha_release_auto_increment();
   if (thd->lock)
   {
@@ -552,7 +552,7 @@ read_fixed_length(THD *thd, COPY_INFO &info, TABLE_LIST *table_list,
   Item_field *sql_field;
   TABLE *table= table_list->table;
   ulonglong id;
-  bool no_trans_update_stmt, err;
+  bool err;
   DBUG_ENTER("read_fixed_length");
 
   id= 0;
@@ -580,7 +580,6 @@ read_fixed_length(THD *thd, COPY_INFO &info, TABLE_LIST *table_list,
 #ifdef HAVE_purify
     read_info.row_end[0]=0;
 #endif
-    no_trans_update_stmt= !table->file->has_transactions();
 
     restore_record(table, s->default_values);
     /*
@@ -650,7 +649,6 @@ read_fixed_length(THD *thd, COPY_INFO &info, TABLE_LIST *table_list,
     table->auto_increment_field_not_null= FALSE;
     if (err)
       DBUG_RETURN(1);
-    thd->no_trans_update.stmt= no_trans_update_stmt;
    
     /*
       We don't need to reset auto-increment field since we are restoring
@@ -685,12 +683,11 @@ read_sep_field(THD *thd, COPY_INFO &info, TABLE_LIST *table_list,
   TABLE *table= table_list->table;
   uint enclosed_length;
   ulonglong id;
-  bool no_trans_update_stmt, err;
+  bool err;
   DBUG_ENTER("read_sep_field");
 
   enclosed_length=enclosed.length();
   id= 0;
-  no_trans_update_stmt= !table->file->has_transactions();
 
   for (;;it.rewind())
   {
@@ -827,7 +824,6 @@ read_sep_field(THD *thd, COPY_INFO &info, TABLE_LIST *table_list,
       We don't need to reset auto-increment field since we are restoring
       its default value at the beginning of each loop iteration.
     */
-    thd->no_trans_update.stmt= no_trans_update_stmt;
     if (read_info.next_line())			// Skip to next line
       break;
     if (read_info.line_cuted)
diff --git a/sql/sql_parse.cc b/sql/sql_parse.cc
index 01a5467a0b9..737ec320a69 100644
--- a/sql/sql_parse.cc
+++ b/sql/sql_parse.cc
@@ -120,7 +120,7 @@ bool end_active_trans(THD *thd)
       error=1;
   }
   thd->options&= ~(OPTION_BEGIN | OPTION_KEEP_LOG);
-  thd->no_trans_update.all= FALSE;
+  thd->transaction.all.modified_non_trans_table= FALSE;
   DBUG_RETURN(error);
 }
 
@@ -223,7 +223,6 @@ void init_update_queries(void)
   sql_command_flags[SQLCOM_REPLACE_SELECT]= CF_CHANGES_DATA | CF_HAS_ROW_COUNT;
 
   sql_command_flags[SQLCOM_SHOW_STATUS_PROC]= CF_STATUS_COMMAND;
-  sql_command_flags[SQLCOM_SHOW_STATUS_FUNC]= CF_STATUS_COMMAND;
   sql_command_flags[SQLCOM_SHOW_STATUS]=      CF_STATUS_COMMAND;
   sql_command_flags[SQLCOM_SHOW_DATABASES]=   CF_STATUS_COMMAND;
   sql_command_flags[SQLCOM_SHOW_TRIGGERS]=    CF_STATUS_COMMAND;
@@ -235,10 +234,36 @@ void init_update_queries(void)
   sql_command_flags[SQLCOM_SHOW_VARIABLES]=   CF_STATUS_COMMAND;
   sql_command_flags[SQLCOM_SHOW_CHARSETS]=    CF_STATUS_COMMAND;
   sql_command_flags[SQLCOM_SHOW_COLLATIONS]=  CF_STATUS_COMMAND;
-  sql_command_flags[SQLCOM_SHOW_STATUS_PROC]= CF_STATUS_COMMAND;
-
-  sql_command_flags[SQLCOM_SHOW_TABLES]=       (CF_STATUS_COMMAND |
-                                                CF_SHOW_TABLE_COMMAND);
+  sql_command_flags[SQLCOM_SHOW_NEW_MASTER]= CF_STATUS_COMMAND;
+  sql_command_flags[SQLCOM_SHOW_BINLOGS]= CF_STATUS_COMMAND;
+  sql_command_flags[SQLCOM_SHOW_SLAVE_HOSTS]= CF_STATUS_COMMAND;
+  sql_command_flags[SQLCOM_SHOW_BINLOG_EVENTS]= CF_STATUS_COMMAND;
+  sql_command_flags[SQLCOM_SHOW_COLUMN_TYPES]= CF_STATUS_COMMAND;
+  sql_command_flags[SQLCOM_SHOW_STORAGE_ENGINES]= CF_STATUS_COMMAND;
+  sql_command_flags[SQLCOM_SHOW_AUTHORS]= CF_STATUS_COMMAND;
+  sql_command_flags[SQLCOM_SHOW_CONTRIBUTORS]= CF_STATUS_COMMAND;
+  sql_command_flags[SQLCOM_SHOW_PRIVILEGES]= CF_STATUS_COMMAND;
+  sql_command_flags[SQLCOM_SHOW_WARNS]= CF_STATUS_COMMAND;
+  sql_command_flags[SQLCOM_SHOW_ERRORS]= CF_STATUS_COMMAND;
+  sql_command_flags[SQLCOM_SHOW_ENGINE_STATUS]= CF_STATUS_COMMAND;
+  sql_command_flags[SQLCOM_SHOW_ENGINE_MUTEX]= CF_STATUS_COMMAND;
+  sql_command_flags[SQLCOM_SHOW_ENGINE_LOGS]= CF_STATUS_COMMAND;
+  sql_command_flags[SQLCOM_SHOW_PROCESSLIST]= CF_STATUS_COMMAND;
+  sql_command_flags[SQLCOM_SHOW_GRANTS]=  CF_STATUS_COMMAND;
+  sql_command_flags[SQLCOM_SHOW_CREATE_DB]=  CF_STATUS_COMMAND;
+  sql_command_flags[SQLCOM_SHOW_CREATE]=  CF_STATUS_COMMAND;
+  sql_command_flags[SQLCOM_SHOW_MASTER_STAT]=  CF_STATUS_COMMAND;
+  sql_command_flags[SQLCOM_SHOW_SLAVE_STAT]=  CF_STATUS_COMMAND;
+  sql_command_flags[SQLCOM_SHOW_CREATE_PROC]=  CF_STATUS_COMMAND;
+  sql_command_flags[SQLCOM_SHOW_CREATE_FUNC]=  CF_STATUS_COMMAND;
+  sql_command_flags[SQLCOM_SHOW_CREATE_TRIGGER]=  CF_STATUS_COMMAND;
+  sql_command_flags[SQLCOM_SHOW_STATUS_FUNC]=  CF_STATUS_COMMAND;
+  sql_command_flags[SQLCOM_SHOW_PROC_CODE]=  CF_STATUS_COMMAND;
+  sql_command_flags[SQLCOM_SHOW_FUNC_CODE]=  CF_STATUS_COMMAND;
+  sql_command_flags[SQLCOM_SHOW_CREATE_EVENT]=  CF_STATUS_COMMAND;
+
+   sql_command_flags[SQLCOM_SHOW_TABLES]=       (CF_STATUS_COMMAND |
+                                               CF_SHOW_TABLE_COMMAND);
   sql_command_flags[SQLCOM_SHOW_TABLE_STATUS]= (CF_STATUS_COMMAND |
                                                 CF_SHOW_TABLE_COMMAND);
 
@@ -572,7 +597,7 @@ int end_trans(THD *thd, enum enum_mysql_completiontype completion)
     thd->server_status&= ~SERVER_STATUS_IN_TRANS;
     res= ha_commit(thd);
     thd->options&= ~(ulong) (OPTION_BEGIN | OPTION_KEEP_LOG);
-    thd->no_trans_update.all= FALSE;
+    thd->transaction.all.modified_non_trans_table= FALSE;
     break;
   case COMMIT_RELEASE:
     do_release= 1; /* fall through */
@@ -590,7 +615,7 @@ int end_trans(THD *thd, enum enum_mysql_completiontype completion)
     if (ha_rollback(thd))
       res= -1;
     thd->options&= ~(ulong) (OPTION_BEGIN | OPTION_KEEP_LOG);
-    thd->no_trans_update.all= FALSE;
+    thd->transaction.all.modified_non_trans_table= FALSE;
     if (!res && (completion == ROLLBACK_AND_CHAIN))
       res= begin_trans(thd);
     break;
@@ -1342,7 +1367,8 @@ void log_slow_statement(THD *thd)
 	thd->variables.long_query_time ||
 	((thd->server_status &
 	  (SERVER_QUERY_NO_INDEX_USED | SERVER_QUERY_NO_GOOD_INDEX_USED)) &&
-	 opt_log_queries_not_using_indexes))
+         opt_log_queries_not_using_indexes &&
+          !(sql_command_flags[thd->lex->sql_command] & CF_STATUS_COMMAND)))
     {
       thd->status_var.long_query_count++;
       slow_log_print(thd, thd->query, thd->query_length, start_of_query);
@@ -1799,6 +1825,8 @@ mysql_execute_command(THD *thd)
 #endif
   status_var_increment(thd->status_var.com_stat[lex->sql_command]);
 
+  DBUG_ASSERT(thd->transaction.stmt.modified_non_trans_table == FALSE);
+  
   switch (lex->sql_command) {
   case SQLCOM_SHOW_EVENTS:
     if ((res= check_access(thd, EVENT_ACL, thd->lex->select_lex.db, 0, 0, 0,
@@ -3613,7 +3641,8 @@ end_with_restore_list:
         res= TRUE; // cannot happen
       else
       {
-        if (((thd->options & OPTION_KEEP_LOG) || thd->no_trans_update.all) &&
+        if (((thd->options & OPTION_KEEP_LOG) || 
+             thd->transaction.all.modified_non_trans_table) &&
             !thd->slave_thread)
           push_warning(thd, MYSQL_ERROR::WARN_LEVEL_WARN,
                        ER_WARNING_NOT_COMPLETE_ROLLBACK,
@@ -4193,8 +4222,8 @@ create_sp_error:
     thd->transaction.xid_state.xa_state=XA_ACTIVE;
     thd->transaction.xid_state.xid.set(thd->lex->xid);
     xid_cache_insert(&thd->transaction.xid_state);
+    thd->transaction.all.modified_non_trans_table= FALSE;
     thd->options= ((thd->options & ~(OPTION_KEEP_LOG)) | OPTION_BEGIN);
-    thd->no_trans_update.all= FALSE;
     thd->server_status|= SERVER_STATUS_IN_TRANS;
     send_ok(thd);
     break;
@@ -4288,7 +4317,7 @@ create_sp_error:
       break;
     }
     thd->options&= ~(OPTION_BEGIN | OPTION_KEEP_LOG);
-    thd->no_trans_update.all= FALSE;
+    thd->transaction.all.modified_non_trans_table= FALSE;
     thd->server_status&= ~SERVER_STATUS_IN_TRANS;
     xid_cache_delete(&thd->transaction.xid_state);
     thd->transaction.xid_state.xa_state=XA_NOTR;
@@ -4319,7 +4348,7 @@ create_sp_error:
     else
       send_ok(thd);
     thd->options&= ~(OPTION_BEGIN | OPTION_KEEP_LOG);
-    thd->no_trans_update.all= FALSE;
+    thd->transaction.all.modified_non_trans_table= FALSE;
     thd->server_status&= ~SERVER_STATUS_IN_TRANS;
     xid_cache_delete(&thd->transaction.xid_state);
     thd->transaction.xid_state.xa_state=XA_NOTR;
@@ -5170,7 +5199,7 @@ void mysql_reset_thd_for_next_command(THD *thd)
   if (!(thd->options & (OPTION_NOT_AUTOCOMMIT | OPTION_BEGIN)))
   {
     thd->options&= ~OPTION_KEEP_LOG;
-    thd->no_trans_update.all= FALSE;
+    thd->transaction.all.modified_non_trans_table= FALSE;
   }
   DBUG_ASSERT(thd->security_ctx== &thd->main_security_ctx);
   thd->thread_specific_used= FALSE;
diff --git a/sql/sql_select.cc b/sql/sql_select.cc
index b7757d3e5fc..f333e754ff7 100644
--- a/sql/sql_select.cc
+++ b/sql/sql_select.cc
@@ -1134,6 +1134,7 @@ JOIN::optimize()
     order=0;					// The output has only one row
     simple_order=1;
     select_distinct= 0;                       // No need in distinct for 1 row
+    group_optimized_away= 1;
   }
 
   calc_group_buffer(this, group_list);
@@ -11742,7 +11743,8 @@ end_send_group(JOIN *join, JOIN_TAB *join_tab __attribute__((unused)),
   if (!join->first_record || end_of_records ||
       (idx=test_if_group_changed(join->group_fields)) >= 0)
   {
-    if (join->first_record || (end_of_records && !join->group))
+    if (join->first_record || 
+        (end_of_records && !join->group && !join->group_optimized_away))
     {
       if (join->procedure)
 	join->procedure->end_group();
@@ -12313,7 +12315,6 @@ static int test_if_order_by_key(ORDER *order, TABLE *table, uint idx,
       */
       if (!on_primary_key &&
           (table->file->ha_table_flags() & HA_PRIMARY_KEY_IN_READ_INDEX) &&
-          ha_legacy_type(table->s->db_type()) == DB_TYPE_INNODB &&
           table->s->primary_key != MAX_KEY)
       {
         on_primary_key= TRUE;
diff --git a/sql/sql_select.h b/sql/sql_select.h
index 98f2a7829dd..64173a91162 100644
--- a/sql/sql_select.h
+++ b/sql/sql_select.h
@@ -321,6 +321,14 @@ public:
   ROLLUP rollup;				// Used with rollup
 
   bool select_distinct;				// Set if SELECT DISTINCT
+  /*
+    If we have the GROUP BY statement in the query,
+    but the group_list was emptied by optimizer, this
+    flag is TRUE.
+    It happens when fields in the GROUP BY are from
+    constant table
+  */
+  bool group_optimized_away;
 
   /*
     simple_xxxxx is set if ORDER/GROUP BY doesn't include any references
@@ -429,6 +437,7 @@ public:
     zero_result_cause= 0;
     optimized= 0;
     cond_equal= 0;
+    group_optimized_away= 0;
 
     all_fields= fields_arg;
     fields_list= fields_arg;
diff --git a/sql/sql_show.cc b/sql/sql_show.cc
index fdd75fbe844..a0c3355b49a 100644
--- a/sql/sql_show.cc
+++ b/sql/sql_show.cc
@@ -3667,7 +3667,7 @@ static int get_schema_views_record(THD *thd, TABLE_LIST *tables,
       Item *item;
       Item_field *field;
       /*
-        chech that at least one coulmn in view is updatable
+        check that at least one column in view is updatable
       */
       while ((item= it++))
       {
@@ -3678,6 +3678,8 @@ static int get_schema_views_record(THD *thd, TABLE_LIST *tables,
           break;
         }
       }
+      if (updatable_view && !tables->view->can_be_merged())
+        updatable_view= 0;
     }
     if (updatable_view)
       table->field[5]->store(STRING_WITH_LEN("YES"), cs);
diff --git a/sql/sql_table.cc b/sql/sql_table.cc
index a555bc0a58b..dd92c47d29b 100644
--- a/sql/sql_table.cc
+++ b/sql/sql_table.cc
@@ -6738,7 +6738,6 @@ copy_data_between_tables(TABLE *from,TABLE *to,
   alter_table_manage_keys(to, from->file->indexes_are_disabled(), keys_onoff);
 
   /* We can abort alter table for any table type */
-  thd->no_trans_update.stmt= FALSE;
   thd->abort_on_warning= !ignore && test(thd->variables.sql_mode &
                                          (MODE_STRICT_TRANS_TABLES |
                                           MODE_STRICT_ALL_TABLES));
diff --git a/sql/sql_update.cc b/sql/sql_update.cc
index 1016a23b5ae..cb3f2fece89 100644
--- a/sql/sql_update.cc
+++ b/sql/sql_update.cc
@@ -530,7 +530,6 @@ int mysql_update(THD *thd,
   thd->proc_info="Updating";
 
   transactional_table= table->file->has_transactions();
-  thd->no_trans_update.stmt= FALSE;
   thd->abort_on_warning= test(!ignore &&
                               (thd->variables.sql_mode &
                                (MODE_STRICT_TRANS_TABLES |
@@ -641,7 +640,6 @@ int mysql_update(THD *thd,
             updated++;
           else
             error= 0;
-          thd->no_trans_update.stmt= !transactional_table;
 
           if (table->triggers &&
               table->triggers->process_triggers(thd, TRG_EVENT_UPDATE,
@@ -717,6 +715,10 @@ int mysql_update(THD *thd,
   }
   dup_key_found= 0;
 
+  if (!transactional_table && updated > 0)
+    thd->transaction.stmt.modified_non_trans_table= TRUE;
+
+
   /*
     todo bug#27571: to avoid asynchronization of `error' and
     `error_code' of binlog event constructor
@@ -809,9 +811,10 @@ int mysql_update(THD *thd,
         error=1;				// Rollback update
       }
     }
-    if (!transactional_table)
-      thd->no_trans_update.all= TRUE;
+    if (thd->transaction.stmt.modified_non_trans_table)
+      thd->transaction.all.modified_non_trans_table= TRUE;
   }
+  DBUG_ASSERT(transactional_table || !updated || thd->transaction.stmt.modified_non_trans_table);
   free_underlaid_joins(thd, select_lex);
   if (transactional_table)
   {
@@ -1176,7 +1179,6 @@ bool mysql_multi_update(THD *thd,
 				 handle_duplicates, ignore)))
     DBUG_RETURN(TRUE);
 
-  thd->no_trans_update.stmt= FALSE;
   thd->abort_on_warning= test(thd->variables.sql_mode &
                               (MODE_STRICT_TRANS_TABLES |
                                MODE_STRICT_ALL_TABLES));
@@ -1549,9 +1551,8 @@ multi_update::~multi_update()
   if (copy_field)
     delete [] copy_field;
   thd->count_cuted_fields= CHECK_FIELD_IGNORE;		// Restore this setting
-  if (!trans_safe)  // todo: remove since redundant
-    thd->no_trans_update.all= TRUE;
-  DBUG_ASSERT(trans_safe || thd->no_trans_update.all);
+  DBUG_ASSERT(trans_safe || !updated || 
+              thd->transaction.all.modified_non_trans_table);
 }
 
 
@@ -1655,7 +1656,7 @@ bool multi_update::send_data(List<Item> &not_used_values)
           else
           {
             trans_safe= 0;
-            thd->no_trans_update.stmt= TRUE;
+            thd->transaction.stmt.modified_non_trans_table= TRUE;
           }
           if (table->triggers &&
               table->triggers->process_triggers(thd, TRG_EVENT_UPDATE,
@@ -1718,7 +1719,6 @@ void multi_update::send_error(uint errcode,const char *err)
 
   /* Something already updated so we have to invalidate cache */
   query_cache_invalidate3(thd, update_tables, 1);
-
   /*
     If all tables that has been updated are trans safe then just do rollback.
     If not attempt to do remaining updates.
@@ -1731,7 +1731,7 @@ void multi_update::send_error(uint errcode,const char *err)
   }
   else
   { 
-    DBUG_ASSERT(thd->no_trans_update.stmt);
+    DBUG_ASSERT(thd->transaction.stmt.modified_non_trans_table);
     if (do_update && table_count > 1)
     {
       /* Add warning here */
@@ -1742,7 +1742,7 @@ void multi_update::send_error(uint errcode,const char *err)
       VOID(do_updates(0));
     }
   }
-  if (thd->no_trans_update.stmt)
+  if (thd->transaction.stmt.modified_non_trans_table)
   {
     /*
       The query has to binlog because there's a modified non-transactional table
@@ -1755,9 +1755,9 @@ void multi_update::send_error(uint errcode,const char *err)
                         transactional_tables, FALSE);
     }
     if (!trans_safe)
-      thd->no_trans_update.all= TRUE;
+      thd->transaction.all.modified_non_trans_table= TRUE;
   }
-  DBUG_ASSERT(trans_safe || !updated || thd->no_trans_update.stmt);
+  DBUG_ASSERT(trans_safe || !updated || thd->transaction.stmt.modified_non_trans_table);
   
   if (transactional_tables)
   {
@@ -1900,7 +1900,7 @@ int multi_update::do_updates(bool from_send_error)
       else
       {
         trans_safe= 0;				// Can't do safe rollback
-        thd->no_trans_update.stmt= TRUE;
+        thd->transaction.stmt.modified_non_trans_table= TRUE;
       }
     }
     (void) table->file->ha_rnd_end();
@@ -1934,7 +1934,7 @@ err2:
     else
     {
       trans_safe= 0;
-      thd->no_trans_update.stmt= TRUE;
+      thd->transaction.stmt.modified_non_trans_table= TRUE;
     }
   }
   DBUG_RETURN(1);
@@ -1961,7 +1961,6 @@ bool multi_update::send_eof()
   {
     query_cache_invalidate3(thd, update_tables, 1);
   }
-
   /*
     Write the SQL statement to the binlog if we updated
     rows and we succeeded or if we updated some non
@@ -1971,8 +1970,9 @@ bool multi_update::send_eof()
     either from the query's list or via a stored routine: bug#13270,23333
   */
 
-  DBUG_ASSERT(trans_safe || !updated || thd->no_trans_update.stmt);
-  if (local_error == 0 || thd->no_trans_update.stmt)
+  DBUG_ASSERT(trans_safe || !updated || 
+              thd->transaction.stmt.modified_non_trans_table);
+  if (local_error == 0 || thd->transaction.stmt.modified_non_trans_table)
   {
     if (mysql_bin_log.is_open())
     {
@@ -1988,8 +1988,8 @@ bool multi_update::send_eof()
 	local_error= 1;				// Rollback update
       }
     }
-    if (!trans_safe)
-     thd->no_trans_update.all= TRUE;
+    if (thd->transaction.stmt.modified_non_trans_table)
+      thd->transaction.all.modified_non_trans_table= TRUE;
   }
 
   if (transactional_tables)
diff --git a/sql/sql_yacc.yy b/sql/sql_yacc.yy
index 81bb1c2dae9..591f2c46d4d 100644
--- a/sql/sql_yacc.yy
+++ b/sql/sql_yacc.yy
@@ -8729,6 +8729,8 @@ show_param:
 	    LEX *lex=Lex;
 	    lex->sql_command= SQLCOM_SHOW_STORAGE_ENGINES;
 	    WARN_DEPRECATED(yythd, "5.2", "SHOW TABLE TYPES", "'SHOW [STORAGE] ENGINES'");
+            if (prepare_schema_table(YYTHD, lex, 0, SCH_ENGINES))
+              MYSQL_YYABORT;
 	  }
 	| opt_storage ENGINES_SYM
 	  {
@@ -9503,7 +9505,8 @@ simple_ident:
             Item_splocal *splocal;
             splocal= new Item_splocal($1, spv->offset, spv->type,
                                       lip->get_tok_start_prev() -
-                                      lex->sphead->m_tmp_query);
+                                      lex->sphead->m_tmp_query,
+                                      lip->get_tok_end() - lip->get_tok_start_prev());
 #ifndef DBUG_OFF
             if (splocal)
               splocal->m_sp= lex->sphead;
diff --git a/sql/table.cc b/sql/table.cc
index f52c81d7889..5e036abe6b1 100644
--- a/sql/table.cc
+++ b/sql/table.cc
@@ -1391,8 +1391,7 @@ static int open_binary_frm(THD *thd, TABLE_SHARE *share, uchar *head,
           if (ha_option & HA_PRIMARY_KEY_IN_READ_INDEX)
           {
             field->part_of_key= share->keys_in_use;
-            if (ha_legacy_type(share->db_type()) == DB_TYPE_INNODB &&
-                field->part_of_sortkey.is_set(key))
+            if (field->part_of_sortkey.is_set(key))
               field->part_of_sortkey= share->keys_in_use;
           }
         }