Manual merge from mysql-trunk-bugfixing.

Conflicts:
  - BUILD/SETUP.sh
  - mysql-test/mysql-test-run.pl
  - mysql-test/r/partition_error.result
  - mysql-test/t/disabled.def
  - mysql-test/t/partition_error.test
  - sql/share/errmsg-utf8.txt

71 files changed, 1160 insertions, 950 deletions

diff --git a/sql/debug_sync.cc b/sql/debug_sync.cc
index d16fa4b2468..dde6267331f 100644
--- a/sql/debug_sync.cc
+++ b/sql/debug_sync.cc
@@ -387,6 +387,13 @@ static st_debug_sync_globals debug_sync_global; /* All globals in one object */
 */
 extern "C" void (*debug_sync_C_callback_ptr)(const char *, size_t);
 
+/**
+  Callbacks from C files.
+*/
+C_MODE_START
+static void debug_sync_C_callback(const char *, size_t);
+static int debug_sync_qsort_cmp(const void *, const void *);
+C_MODE_END
 
 /**
   Callback for debug sync, to be used by C files. See thr_lock.c for example.
@@ -422,8 +429,8 @@ extern "C" void (*debug_sync_C_callback_ptr)(const char *, size_t);
 static void debug_sync_C_callback(const char *sync_point_name,
                                   size_t name_len)
 {
-  if (unlikely(opt_debug_sync_timeout))                            
-    debug_sync(current_thd, sync_point_name, name_len);   
+  if (unlikely(opt_debug_sync_timeout))
+    debug_sync(current_thd, sync_point_name, name_len);
 }
 
 #ifdef HAVE_PSI_INTERFACE
diff --git a/sql/derror.cc b/sql/derror.cc
index 04a82860d45..7f1435e89c1 100644
--- a/sql/derror.cc
+++ b/sql/derror.cc
@@ -32,11 +32,12 @@
 static void init_myfunc_errs(void);
 
 
-const char **get_server_errmsgs()
+C_MODE_START
+static const char **get_server_errmsgs()
 {
   return CURRENT_THD_ERRMSGS;
 }
-
+C_MODE_END
 
 /**
   Read messages from errorfile.
diff --git a/sql/field.cc b/sql/field.cc
index d3de8bf6c6d..bfaaf10b141 100644
--- a/sql/field.cc
+++ b/sql/field.cc
@@ -1009,17 +1009,20 @@ test_if_important_data(CHARSET_INFO *cs, const char *str, const char *strend)
    Used below. In an anonymous namespace to not clash with definitions
    in other files.
  */
-namespace {
-  int compare(unsigned int a, unsigned int b)
-  {
-    if (a < b)
-      return -1;
-    if (b < a)
-      return 1;
-    return 0;
-}
+
+CPP_UNNAMED_NS_START
+
+int compare(unsigned int a, unsigned int b)
+{
+  if (a < b)
+    return -1;
+  if (b < a)
+    return 1;
+  return 0;
 }
 
+CPP_UNNAMED_NS_END
+
 /**
   Detect Item_result by given field type of UNION merge result.
 
@@ -9112,7 +9115,7 @@ void Create_field::create_length_to_internal_length(void)
 void Create_field::init_for_tmp_table(enum_field_types sql_type_arg,
                                       uint32 length_arg, uint32 decimals_arg,
                                       bool maybe_null, bool is_unsigned,
-                                      uint pack_length)
+                                      uint pack_length_arg)
 {
   DBUG_ENTER("Create_field::init_for_tmp_table");
 
@@ -9125,7 +9128,7 @@ void Create_field::init_for_tmp_table(enum_field_types sql_type_arg,
   geom_type= Field::GEOM_GEOMETRY;
 
   DBUG_PRINT("enter", ("sql_type: %d, length: %u, pack_length: %u",
-                       sql_type_arg, length_arg, pack_length));
+                       sql_type_arg, length_arg, pack_length_arg));
 
   /*
     These pack flags are crafted to get it correctly through the
@@ -9189,8 +9192,8 @@ void Create_field::init_for_tmp_table(enum_field_types sql_type_arg,
   case MYSQL_TYPE_GEOMETRY:
     // If you are going to use the above types, you have to pass a
     // pack_length as parameter. Assert that is really done.
-    DBUG_ASSERT(pack_length != ~0U);
-    pack_flag|= pack_length_to_packflag(pack_length);
+    DBUG_ASSERT(pack_length_arg != ~0U);
+    pack_flag|= pack_length_to_packflag(pack_length_arg);
     break;
   default:
     /* Nothing */
@@ -9944,6 +9947,39 @@ Create_field::Create_field(Field *old_field,Field *orig_field)
 
 
 /**
+  maximum possible character length for blob.
+  
+  This method is used in Item_field::set_field to calculate
+  max_length for Item.
+  
+  For example:
+    CREATE TABLE t2 SELECT CONCAT(tinyblob_utf8_column) FROM t1;
+  must create a "VARCHAR(255) CHARACTER SET utf8" column.
+  
+  @return
+    length
+*/
+
+uint32 Field_blob::char_length()
+{
+  switch (packlength)
+  {
+  case 1:
+    return 255;
+  case 2:
+    return 65535;
+  case 3:
+    return 16777215;
+  case 4:
+    return (uint32) 4294967295U;
+  default:
+    DBUG_ASSERT(0); // we should never go here
+    return 0;
+  }
+}
+
+
+/**
   maximum possible display length for blob.
 
   @return
diff --git a/sql/field.h b/sql/field.h
index 66b13d02b89..46d8a2aa6d9 100644
--- a/sql/field.h
+++ b/sql/field.h
@@ -499,7 +499,7 @@ public:
   longlong convert_decimal2longlong(const my_decimal *val, bool unsigned_flag,
                                     int *err);
   /* The max. number of characters */
-  inline uint32 char_length() const
+  virtual uint32 char_length()
   {
     return field_length / charset()->mbmaxlen;
   }
@@ -1813,6 +1813,7 @@ public:
   bool has_charset(void) const
   { return charset() == &my_charset_bin ? FALSE : TRUE; }
   uint32 max_display_length();
+  uint32 char_length();
   uint is_equal(Create_field *new_field);
   inline bool in_read_set() { return bitmap_is_set(table->read_set, field_index); }
   inline bool in_write_set() { return bitmap_is_set(table->write_set, field_index); }
diff --git a/sql/ha_ndbcluster.cc b/sql/ha_ndbcluster.cc
index 38324f3cf19..9f003174d2e 100644
--- a/sql/ha_ndbcluster.cc
+++ b/sql/ha_ndbcluster.cc
@@ -4249,7 +4249,7 @@ int ha_ndbcluster::info(uint flag)
 }
 
 
-void ha_ndbcluster::get_dynamic_partition_info(PARTITION_INFO *stat_info,
+void ha_ndbcluster::get_dynamic_partition_info(PARTITION_STATS *stat_info,
                                                uint part_id)
 {
   /* 
@@ -4257,7 +4257,7 @@ void ha_ndbcluster::get_dynamic_partition_info(PARTITION_INFO *stat_info,
      implement ndb function which retrives the statistics
      about ndb partitions.
   */
-  bzero((char*) stat_info, sizeof(PARTITION_INFO));
+  bzero((char*) stat_info, sizeof(PARTITION_STATS));
   return;
 }
 
@@ -4628,7 +4628,7 @@ int ha_ndbcluster::start_statement(THD *thd,
   trans_register_ha(thd, FALSE, ndbcluster_hton);
   if (!thd_ndb->trans)
   {
-    if (thd->in_multi_stmt_transaction())
+    if (thd->in_multi_stmt_transaction_mode())
       trans_register_ha(thd, TRUE, ndbcluster_hton);
     DBUG_PRINT("trans",("Starting transaction"));      
     thd_ndb->trans= ndb->startTransaction();
@@ -4698,7 +4698,7 @@ int ha_ndbcluster::init_handler_for_statement(THD *thd, Thd_ndb *thd_ndb)
   }
 #endif
 
-  if (thd->in_multi_stmt_transaction())
+  if (thd->in_multi_stmt_transaction_mode())
   {
     const void *key= m_table;
     HASH_SEARCH_STATE state;
@@ -4782,7 +4782,7 @@ int ha_ndbcluster::external_lock(THD *thd, int lock_type)
     if (opt_ndb_cache_check_time && m_rows_changed)
     {
       DBUG_PRINT("info", ("Rows has changed and util thread is running"));
-      if (thd->in_multi_stmt_transaction())
+      if (thd->in_multi_stmt_transaction_mode())
       {
         DBUG_PRINT("info", ("Add share to list of tables to be invalidated"));
         /* NOTE push_back allocates memory using transactions mem_root! */
@@ -4801,7 +4801,7 @@ int ha_ndbcluster::external_lock(THD *thd, int lock_type)
       DBUG_PRINT("trans", ("Last external_lock"));
       PRINT_OPTION_FLAGS(thd);
 
-      if (!thd->in_multi_stmt_transaction())
+      if (!thd->in_multi_stmt_transaction_mode())
       {
         if (thd_ndb->trans)
         {
@@ -4911,7 +4911,7 @@ static int ndbcluster_commit(handlerton *hton, THD *thd, bool all)
   PRINT_OPTION_FLAGS(thd);
   DBUG_PRINT("enter", ("Commit %s", (all ? "all" : "stmt")));
   thd_ndb->start_stmt_count= 0;
-  if (trans == NULL || (!all && thd->in_multi_stmt_transaction()))
+  if (trans == NULL || (!all && thd->in_multi_stmt_transaction_mode()))
   {
     /*
       An odditity in the handler interface is that commit on handlerton
@@ -4981,7 +4981,7 @@ static int ndbcluster_rollback(handlerton *hton, THD *thd, bool all)
   DBUG_ASSERT(ndb);
   thd_ndb->start_stmt_count= 0;
   if (trans == NULL || (!all &&
-      thd->in_multi_stmt_transaction()))
+      thd->in_multi_stmt_transaction_mode()))
   {
     /* Ignore end-of-statement until real rollback or commit is called */
     DBUG_PRINT("info", ("Rollback before start or end-of-statement only"));
@@ -8271,7 +8271,7 @@ ndbcluster_cache_retrieval_allowed(THD *thd,
   DBUG_ENTER("ndbcluster_cache_retrieval_allowed");
   DBUG_PRINT("enter", ("dbname: %s, tabname: %s", dbname, tabname));
 
-  if (thd->in_multi_stmt_transaction())
+  if (thd->in_multi_stmt_transaction_mode())
   {
     DBUG_PRINT("exit", ("No, don't use cache in transaction"));
     DBUG_RETURN(FALSE);
@@ -8339,7 +8339,7 @@ ha_ndbcluster::register_query_cache_table(THD *thd,
   DBUG_ENTER("ha_ndbcluster::register_query_cache_table");
   DBUG_PRINT("enter",("dbname: %s, tabname: %s", m_dbname, m_tabname));
 
-  if (thd->in_multi_stmt_transaction())
+  if (thd->in_multi_stmt_transaction_mode())
   {
     DBUG_PRINT("exit", ("Can't register table during transaction"));
     DBUG_RETURN(FALSE);
diff --git a/sql/ha_ndbcluster.h b/sql/ha_ndbcluster.h
index 17ba7451538..de1e36b61d2 100644
--- a/sql/ha_ndbcluster.h
+++ b/sql/ha_ndbcluster.h
@@ -275,7 +275,7 @@ class ha_ndbcluster: public handler
   ha_rows estimate_rows_upper_bound()
     { return HA_POS_ERROR; }
   int info(uint);
-  void get_dynamic_partition_info(PARTITION_INFO *stat_info, uint part_id);
+  void get_dynamic_partition_info(PARTITION_STATS *stat_info, uint part_id);
   int extra(enum ha_extra_function operation);
   int extra_opt(enum ha_extra_function operation, ulong cache_size);
   int reset();
diff --git a/sql/ha_partition.cc b/sql/ha_partition.cc
index ac05a699c7d..a6f106be3be 100644
--- a/sql/ha_partition.cc
+++ b/sql/ha_partition.cc
@@ -59,6 +59,7 @@
 #include "sql_table.h"                        // tablename_to_filename
 #include "key.h"
 #include "sql_plugin.h"
+#include "table.h"                           /* HA_DATA_PARTITION */
 
 #include "debug_sync.h"
 
@@ -2510,12 +2511,11 @@ err1:
   A destructor for partition-specific TABLE_SHARE data.
 */
 
-void ha_data_partition_destroy(void *ha_data)
+void ha_data_partition_destroy(HA_DATA_PARTITION* ha_part_data)
 {
-  if (ha_data)
+  if (ha_part_data)
   {
-    HA_DATA_PARTITION *ha_data_partition= (HA_DATA_PARTITION*) ha_data;
-    pthread_mutex_destroy(&ha_data_partition->mutex);
+    mysql_mutex_destroy(&ha_part_data->LOCK_auto_inc);
   }
 }
 
@@ -2655,28 +2655,30 @@ int ha_partition::open(const char *name, int mode, uint test_if_locked)
     goto err_handler;
 
   /*
-    Use table_share->ha_data to share auto_increment_value among all handlers
-    for the same table.
+    Use table_share->ha_part_data to share auto_increment_value among
+    all handlers for the same table.
   */
   if (is_not_tmp_table)
     mysql_mutex_lock(&table_share->LOCK_ha_data);
-  if (!table_share->ha_data)
+  if (!table_share->ha_part_data)
   {
-    HA_DATA_PARTITION *ha_data;
     /* currently only needed for auto_increment */
-    table_share->ha_data= ha_data= (HA_DATA_PARTITION*)
+    table_share->ha_part_data= (HA_DATA_PARTITION*)
                                    alloc_root(&table_share->mem_root,
                                               sizeof(HA_DATA_PARTITION));
-    if (!ha_data)
+    if (!table_share->ha_part_data)
     {
       if (is_not_tmp_table)
         mysql_mutex_unlock(&table_share->LOCK_ha_data);
       goto err_handler;
     }
-    DBUG_PRINT("info", ("table_share->ha_data 0x%p", ha_data));
-    bzero(ha_data, sizeof(HA_DATA_PARTITION));
-    table_share->ha_data_destroy= ha_data_partition_destroy;
-    pthread_mutex_init(&ha_data->mutex, MY_MUTEX_INIT_FAST);
+    DBUG_PRINT("info", ("table_share->ha_part_data 0x%p",
+                        table_share->ha_part_data));
+    bzero(table_share->ha_part_data, sizeof(HA_DATA_PARTITION));
+    table_share->ha_part_data_destroy= ha_data_partition_destroy;
+    mysql_mutex_init(key_PARTITION_LOCK_auto_inc,
+                     &table_share->ha_part_data->LOCK_auto_inc,
+                     MY_MUTEX_INIT_FAST);
   }
   if (is_not_tmp_table)
     mysql_mutex_unlock(&table_share->LOCK_ha_data);
@@ -3101,7 +3103,6 @@ int ha_partition::write_row(uchar * buf)
   longlong func_value;
   bool have_auto_increment= table->next_number_field && buf == table->record[0];
   my_bitmap_map *old_map;
-  HA_DATA_PARTITION *ha_data= (HA_DATA_PARTITION*) table_share->ha_data;
   THD *thd= ha_thd();
   timestamp_auto_set_type orig_timestamp_type= table->timestamp_field_type;
 #ifdef NOT_NEEDED
@@ -3121,8 +3122,8 @@ int ha_partition::write_row(uchar * buf)
   */
   if (have_auto_increment)
   {
-    if (!ha_data->auto_inc_initialized &&
-        !table->s->next_number_keypart)
+    if (!table_share->ha_part_data->auto_inc_initialized &&
+        !table_share->next_number_keypart)
     {
       /*
         If auto_increment in table_share is not initialized, start by
@@ -3262,7 +3263,7 @@ int ha_partition::update_row(const uchar *old_data, uchar *new_data)
 exit:
   /*
     if updating an auto_increment column, update
-    table_share->ha_data->next_auto_inc_val if needed.
+    table_share->ha_part_data->next_auto_inc_val if needed.
     (not to be used if auto_increment on secondary field in a multi-column
     index)
     mysql_update does not set table->next_number_field, so we use
@@ -3271,8 +3272,7 @@ exit:
   if (table->found_next_number_field && new_data == table->record[0] &&
       !table->s->next_number_keypart)
   {
-    HA_DATA_PARTITION *ha_data= (HA_DATA_PARTITION*) table_share->ha_data;
-    if (!ha_data->auto_inc_initialized)
+    if (!table_share->ha_part_data->auto_inc_initialized)
       info(HA_STATUS_AUTO);
     set_auto_increment_if_higher(table->found_next_number_field);
   }
@@ -3364,11 +3364,10 @@ int ha_partition::delete_all_rows()
   if (thd->lex->sql_command == SQLCOM_TRUNCATE)
   {
     Alter_info *alter_info= &thd->lex->alter_info;
-    HA_DATA_PARTITION *ha_data= (HA_DATA_PARTITION*) table_share->ha_data;
     /* TRUNCATE also means resetting auto_increment */
     lock_auto_increment();
-    ha_data->next_auto_inc_val= 0;
-    ha_data->auto_inc_initialized= FALSE;
+    table_share->ha_part_data->next_auto_inc_val= 0;
+    table_share->ha_part_data->auto_inc_initialized= FALSE;
     unlock_auto_increment();
     if (alter_info->flags & ALTER_ADMIN_PARTITION)
     {
@@ -4195,10 +4194,9 @@ int ha_partition::index_read_map(uchar *buf, const uchar *key,
 int ha_partition::common_index_read(uchar *buf, bool have_start_key)
 {
   int error;
-  uint key_len;
+  uint UNINIT_VAR(key_len); /* used if have_start_key==TRUE */
   bool reverse_order= FALSE;
   DBUG_ENTER("ha_partition::common_index_read");
-  LINT_INIT(key_len); /* used if have_start_key==TRUE */
 
   DBUG_PRINT("info", ("m_ordered %u m_ordered_scan_ong %u have_start_key %u",
                       m_ordered, m_ordered_scan_ongoing, have_start_key));
@@ -5084,22 +5082,22 @@ int ha_partition::info(uint flag)
   if (flag & HA_STATUS_AUTO)
   {
     bool auto_inc_is_first_in_idx= (table_share->next_number_keypart == 0);
-    HA_DATA_PARTITION *ha_data= (HA_DATA_PARTITION*) table_share->ha_data;
     DBUG_PRINT("info", ("HA_STATUS_AUTO"));
     if (!table->found_next_number_field)
       stats.auto_increment_value= 0;
-    else if (ha_data->auto_inc_initialized)
+    else if (table_share->ha_part_data->auto_inc_initialized)
     {
       lock_auto_increment();
-      stats.auto_increment_value= ha_data->next_auto_inc_val;
+      stats.auto_increment_value= table_share->ha_part_data->next_auto_inc_val;
       unlock_auto_increment();
     }
     else
     {
       lock_auto_increment();
       /* to avoid two concurrent initializations, check again when locked */
-      if (ha_data->auto_inc_initialized)
-        stats.auto_increment_value= ha_data->next_auto_inc_val;
+      if (table_share->ha_part_data->auto_inc_initialized)
+        stats.auto_increment_value=
+                                 table_share->ha_part_data->next_auto_inc_val;
       else
       {
         handler *file, **file_array;
@@ -5119,10 +5117,11 @@ int ha_partition::info(uint flag)
         stats.auto_increment_value= auto_increment_value;
         if (auto_inc_is_first_in_idx)
         {
-          set_if_bigger(ha_data->next_auto_inc_val, auto_increment_value);
-          ha_data->auto_inc_initialized= TRUE;
+          set_if_bigger(table_share->ha_part_data->next_auto_inc_val,
+                        auto_increment_value);
+          table_share->ha_part_data->auto_inc_initialized= TRUE;
           DBUG_PRINT("info", ("initializing next_auto_inc_val to %lu",
-                              (ulong) ha_data->next_auto_inc_val));
+                       (ulong) table_share->ha_part_data->next_auto_inc_val));
         }
       }
       unlock_auto_increment();
@@ -5297,7 +5296,7 @@ int ha_partition::info(uint flag)
 }
 
 
-void ha_partition::get_dynamic_partition_info(PARTITION_INFO *stat_info,
+void ha_partition::get_dynamic_partition_info(PARTITION_STATS *stat_info,
                                               uint part_id)
 {
   handler *file= m_file[part_id];
@@ -6512,11 +6511,10 @@ int ha_partition::reset_auto_increment(ulonglong value)
 {
   handler **file= m_file;
   int res;
-  HA_DATA_PARTITION *ha_data= (HA_DATA_PARTITION*) table_share->ha_data;
   DBUG_ENTER("ha_partition::reset_auto_increment");
   lock_auto_increment();
-  ha_data->auto_inc_initialized= FALSE;
-  ha_data->next_auto_inc_val= 0;
+  table_share->ha_part_data->auto_inc_initialized= FALSE;
+  table_share->ha_part_data->next_auto_inc_val= 0;
   do
   {
     if ((res= (*file)->ha_reset_auto_increment(value)) != 0)
@@ -6530,7 +6528,7 @@ int ha_partition::reset_auto_increment(ulonglong value)
 /**
   This method is called by update_auto_increment which in turn is called
   by the individual handlers as part of write_row. We use the
-  table_share->ha_data->next_auto_inc_val, or search all
+  table_share->ha_part_data->next_auto_inc_val, or search all
   partitions for the highest auto_increment_value if not initialized or
   if auto_increment field is a secondary part of a key, we must search
   every partition when holding a mutex to be sure of correctness.
@@ -6583,13 +6581,12 @@ void ha_partition::get_auto_increment(ulonglong offset, ulonglong increment,
   else
   {
     THD *thd= ha_thd();
-    HA_DATA_PARTITION *ha_data= (HA_DATA_PARTITION*) table_share->ha_data;
     /*
       This is initialized in the beginning of the first write_row call.
     */
-    DBUG_ASSERT(ha_data->auto_inc_initialized);
+    DBUG_ASSERT(table_share->ha_part_data->auto_inc_initialized);
     /*
-      Get a lock for handling the auto_increment in table_share->ha_data
+      Get a lock for handling the auto_increment in table_share->ha_part_data
       for avoiding two concurrent statements getting the same number.
     */ 
 
@@ -6616,8 +6613,9 @@ void ha_partition::get_auto_increment(ulonglong offset, ulonglong increment,
     }
 
     /* this gets corrected (for offset/increment) in update_auto_increment */
-    *first_value= ha_data->next_auto_inc_val;
-    ha_data->next_auto_inc_val+= nb_desired_values * increment;
+    *first_value= table_share->ha_part_data->next_auto_inc_val;
+    table_share->ha_part_data->next_auto_inc_val+=
+                                              nb_desired_values * increment;
 
     unlock_auto_increment();
     DBUG_PRINT("info", ("*first_value: %lu", (ulong) *first_value));
@@ -6637,10 +6635,9 @@ void ha_partition::release_auto_increment()
   }
   else if (next_insert_id)
   {
-    HA_DATA_PARTITION *ha_data= (HA_DATA_PARTITION*) table_share->ha_data;
     ulonglong next_auto_inc_val;
     lock_auto_increment();
-    next_auto_inc_val= ha_data->next_auto_inc_val;
+    next_auto_inc_val= table_share->ha_part_data->next_auto_inc_val;
     /*
       If the current auto_increment values is lower than the reserved
       value, and the reserved value was reserved by this thread,
@@ -6655,10 +6652,10 @@ void ha_partition::release_auto_increment()
         with SET INSERT_ID, i.e. forced/non generated values.
       */
       if (thd->auto_inc_intervals_forced.maximum() < next_insert_id)
-        ha_data->next_auto_inc_val= next_insert_id;
+        table_share->ha_part_data->next_auto_inc_val= next_insert_id;
     }
-    DBUG_PRINT("info", ("ha_data->next_auto_inc_val: %lu",
-                        (ulong) ha_data->next_auto_inc_val));
+    DBUG_PRINT("info", ("table_share->ha_part_data->next_auto_inc_val: %lu",
+                        (ulong) table_share->ha_part_data->next_auto_inc_val));
 
     /* Unlock the multi row statement lock taken in get_auto_increment */
     if (auto_increment_safe_stmt_log_lock)
@@ -6758,127 +6755,6 @@ int ha_partition::indexes_are_disabled(void)
 }
 
 
-/****************************************************************************
-                MODULE Partition Share
-****************************************************************************/
-/*
-  Service routines for ... methods.
--------------------------------------------------------------------------
-  Variables for partition share methods. A hash used to track open tables.
-  A mutex for the hash table and an init variable to check if hash table
-  is initialized.
-  There is also a constant ending of the partition handler file name.
-*/
-
-#ifdef NOT_USED
-static HASH partition_open_tables;
-static mysql_mutex_t partition_mutex;
-static int partition_init= 0;
-
-
-/*
-  Function we use in the creation of our hash to get key.
-*/
-
-static uchar *partition_get_key(PARTITION_SHARE *share, size_t *length,
-			       my_bool not_used __attribute__ ((unused)))
-{
-  *length= share->table_name_length;
-  return (uchar *) share->table_name;
-}
-
-/*
-  Example of simple lock controls. The "share" it creates is structure we
-  will pass to each partition handler. Do you have to have one of these?
-  Well, you have pieces that are used for locking, and they are needed to
-  function.
-*/
-
-static PARTITION_SHARE *get_share(const char *table_name, TABLE *table)
-{
-  PARTITION_SHARE *share;
-  uint length;
-  char *tmp_name;
-
-  /*
-    So why does this exist? There is no way currently to init a storage
-    engine.
-    Innodb and BDB both have modifications to the server to allow them to
-    do this. Since you will not want to do this, this is probably the next
-    best method.
-  */
-  if (!partition_init)
-  {
-    /* Hijack a mutex for init'ing the storage engine */
-    mysql_mutex_lock(&LOCK_mysql_create_db);
-    if (!partition_init)
-    {
-      partition_init++;
-      mysql_mutex_init(INSTRUMENT_ME, &partition_mutex, MY_MUTEX_INIT_FAST);
-      (void) hash_init(&partition_open_tables, system_charset_info, 32, 0, 0,
-		       (hash_get_key) partition_get_key, 0, 0);
-    }
-    mysql_mutex_unlock(&LOCK_mysql_create_db);
-  }
-  mysql_mutex_lock(&partition_mutex);
-  length= (uint) strlen(table_name);
-
-  if (!(share= (PARTITION_SHARE *) hash_search(&partition_open_tables,
-					       (uchar *) table_name, length)))
-  {
-    if (!(share= (PARTITION_SHARE *)
-	  my_multi_malloc(MYF(MY_WME | MY_ZEROFILL),
-			  &share, (uint) sizeof(*share),
-			  &tmp_name, (uint) length + 1, NullS)))
-    {
-      mysql_mutex_unlock(&partition_mutex);
-      return NULL;
-    }
-
-    share->use_count= 0;
-    share->table_name_length= length;
-    share->table_name= tmp_name;
-    strmov(share->table_name, table_name);
-    if (my_hash_insert(&partition_open_tables, (uchar *) share))
-      goto error;
-    thr_lock_init(&share->lock);
-    mysql_mutex_init(INSTRUMENT_ME, &share->mutex, MY_MUTEX_INIT_FAST);
-  }
-  share->use_count++;
-  mysql_mutex_unlock(&partition_mutex);
-
-  return share;
-
-error:
-  mysql_mutex_unlock(&partition_mutex);
-  my_free((uchar*) share, MYF(0));
-
-  return NULL;
-}
-
-
-/*
-  Free lock controls. We call this whenever we close a table. If the table
-  had the last reference to the share then we free memory associated with
-  it.
-*/
-
-static int free_share(PARTITION_SHARE *share)
-{
-  mysql_mutex_lock(&partition_mutex);
-  if (!--share->use_count)
-  {
-    hash_delete(&partition_open_tables, (uchar *) share);
-    thr_lock_delete(&share->lock);
-    mysql_mutex_destroy(&share->mutex);
-    my_free((uchar*) share, MYF(0));
-  }
-  mysql_mutex_unlock(&partition_mutex);
-
-  return 0;
-}
-#endif /* NOT_USED */
-
 struct st_mysql_storage_engine partition_storage_engine=
 { MYSQL_HANDLERTON_INTERFACE_VERSION };
 
diff --git a/sql/ha_partition.h b/sql/ha_partition.h
index 9f499e7b4a9..eec3cb71537 100644
--- a/sql/ha_partition.h
+++ b/sql/ha_partition.h
@@ -29,31 +29,6 @@ enum partition_keywords
   PKW_COLUMNS
 };
 
-/*
-  PARTITION_SHARE is a structure that will be shared amoung all open handlers
-  The partition implements the minimum of what you will probably need.
-*/
-
-#ifdef NOT_USED
-typedef struct st_partition_share
-{
-  char *table_name;
-  uint table_name_length, use_count;
-  mysql_mutex_t mutex;
-  THR_LOCK lock;
-} PARTITION_SHARE;
-#endif
-
-/**
-  Partition specific ha_data struct.
-  @todo: move all partition specific data from TABLE_SHARE here.
-*/
-typedef struct st_ha_data_partition
-{
-  ulonglong next_auto_inc_val;                 /**< first non reserved value */
-  bool auto_inc_initialized;
-  pthread_mutex_t mutex;
-} HA_DATA_PARTITION;
 
 #define PARTITION_BYTES_IN_POS 2
 #define PARTITION_ENABLED_TABLE_FLAGS (HA_FILE_BASED | HA_REC_NOT_IN_SEQ)
@@ -522,7 +497,7 @@ public:
     -------------------------------------------------------------------------
   */
   virtual int info(uint);
-  void get_dynamic_partition_info(PARTITION_INFO *stat_info,
+  void get_dynamic_partition_info(PARTITION_STATS *stat_info,
                                   uint part_id);
   virtual int extra(enum ha_extra_function operation);
   virtual int extra_opt(enum ha_extra_function operation, ulong cachesize);
@@ -944,16 +919,16 @@ private:
     /* lock already taken */
     if (auto_increment_safe_stmt_log_lock)
       return;
-    DBUG_ASSERT(table_share->ha_data && !auto_increment_lock);
+    DBUG_ASSERT(table_share->ha_part_data && !auto_increment_lock);
     if(table_share->tmp_table == NO_TMP_TABLE)
     {
       auto_increment_lock= TRUE;
-      mysql_mutex_lock(&table_share->LOCK_ha_data);
+      mysql_mutex_lock(&table_share->ha_part_data->LOCK_auto_inc);
     }
   }
   virtual void unlock_auto_increment()
   {
-    DBUG_ASSERT(table_share->ha_data);
+    DBUG_ASSERT(table_share->ha_part_data);
     /*
       If auto_increment_safe_stmt_log_lock is true, we have to keep the lock.
       It will be set to false and thus unlocked at the end of the statement by
@@ -961,20 +936,19 @@ private:
     */
     if(auto_increment_lock && !auto_increment_safe_stmt_log_lock)
     {
-      mysql_mutex_unlock(&table_share->LOCK_ha_data);
+      mysql_mutex_unlock(&table_share->ha_part_data->LOCK_auto_inc);
       auto_increment_lock= FALSE;
     }
   }
   virtual void set_auto_increment_if_higher(Field *field)
   {
-    HA_DATA_PARTITION *ha_data= (HA_DATA_PARTITION*) table_share->ha_data;
     ulonglong nr= (((Field_num*) field)->unsigned_flag ||
                    field->val_int() > 0) ? field->val_int() : 0;
     lock_auto_increment();
-    DBUG_ASSERT(ha_data->auto_inc_initialized == TRUE);
+    DBUG_ASSERT(table_share->ha_part_data->auto_inc_initialized == TRUE);
     /* must check when the mutex is taken */
-    if (nr >= ha_data->next_auto_inc_val)
-      ha_data->next_auto_inc_val= nr + 1;
+    if (nr >= table_share->ha_part_data->next_auto_inc_val)
+      table_share->ha_part_data->next_auto_inc_val= nr + 1;
     unlock_auto_increment();
   }
 
diff --git a/sql/handler.cc b/sql/handler.cc
index d641133c57a..11f684a8010 100644
--- a/sql/handler.cc
+++ b/sql/handler.cc
@@ -299,13 +299,14 @@ handler *get_ha_partition(partition_info *part_info)
 #endif
 
 
-const char **handler_errmsgs;
+static const char **handler_errmsgs;
 
-
-const char **get_handler_errmsgs()
+C_MODE_START
+static const char **get_handler_errmsgs()
 {
   return handler_errmsgs;
 }
+C_MODE_END
 
 
 /**
@@ -1249,7 +1250,14 @@ end:
 /**
   @note
   This function does not care about global read lock. A caller should.
+
+  @param[in]  all  Is set in case of explicit commit
+                   (COMMIT statement), or implicit commit
+                   issued by DDL. Is not set when called
+                   at the end of statement, even if
+                   autocommit=1.
 */
+
 int ha_commit_one_phase(THD *thd, bool all)
 {
   int error=0;
@@ -1257,9 +1265,15 @@ int ha_commit_one_phase(THD *thd, bool all)
   /*
     "real" is a nick name for a transaction for which a commit will
     make persistent changes. E.g. a 'stmt' transaction inside a 'all'
-    transation is not 'real': even though it's possible to commit it,
+    transaction is not 'real': even though it's possible to commit it,
     the changes are not durable as they might be rolled back if the
     enclosing 'all' transaction is rolled back.
+    We establish the value of 'is_real_trans' by checking
+    if it's an explicit COMMIT/BEGIN statement, or implicit
+    commit issued by DDL (all == TRUE), or if we're running
+    in autocommit mode (it's only in the autocommit mode
+    ha_commit_one_phase() can be called with an empty
+    transaction.all.ha_list, see why in trans_register_ha()).
   */
   bool is_real_trans=all || thd->transaction.all.ha_list == 0;
   Ha_trx_info *ha_info= trans->ha_list, *ha_info_next;
@@ -1307,9 +1321,15 @@ int ha_rollback_trans(THD *thd, bool all)
   /*
     "real" is a nick name for a transaction for which a commit will
     make persistent changes. E.g. a 'stmt' transaction inside a 'all'
-    transation is not 'real': even though it's possible to commit it,
+    transaction is not 'real': even though it's possible to commit it,
     the changes are not durable as they might be rolled back if the
     enclosing 'all' transaction is rolled back.
+    We establish the value of 'is_real_trans' by checking
+    if it's an explicit COMMIT or BEGIN statement, or implicit
+    commit issued by DDL (in these cases all == TRUE),
+    or if we're running in autocommit mode (it's only in the autocommit mode
+    ha_commit_one_phase() is called with an empty
+    transaction.all.ha_list, see why in trans_register_ha()).
   */
   bool is_real_trans=all || thd->transaction.all.ha_list == 0;
   DBUG_ENTER("ha_rollback_trans");
@@ -1362,7 +1382,7 @@ int ha_rollback_trans(THD *thd, bool all)
     if (all)
       thd->variables.tx_isolation=thd->session_tx_isolation;
   }
-  /* Always cleanup. Even if there nht==0. There may be savepoints. */
+  /* Always cleanup. Even if nht==0. There may be savepoints. */
   if (is_real_trans)
     thd->transaction.cleanup();
   if (all)
@@ -3533,7 +3553,7 @@ int handler::index_next_same(uchar *buf, const uchar *key, uint keylen)
 }
 
 
-void handler::get_dynamic_partition_info(PARTITION_INFO *stat_info,
+void handler::get_dynamic_partition_info(PARTITION_STATS *stat_info,
                                          uint part_id)
 {
   info(HA_STATUS_CONST | HA_STATUS_TIME | HA_STATUS_VARIABLE |
diff --git a/sql/handler.h b/sql/handler.h
index 4464f4f3920..ad26534d91d 100644
--- a/sql/handler.h
+++ b/sql/handler.h
@@ -961,7 +961,7 @@ typedef struct {
   ulong check_time;
   ulong update_time;
   ulonglong check_sum;
-} PARTITION_INFO;
+} PARTITION_STATS;
 
 #define UNDEF_NODEGROUP 65535
 class Item;
@@ -1560,7 +1560,7 @@ public:
     { return (ha_rows) 10; }
   virtual void position(const uchar *record)=0;
   virtual int info(uint)=0; // see my_base.h for full description
-  virtual void get_dynamic_partition_info(PARTITION_INFO *stat_info,
+  virtual void get_dynamic_partition_info(PARTITION_STATS *stat_info,
                                           uint part_id);
   virtual int extra(enum ha_extra_function operation)
   { return 0; }
diff --git a/sql/init.cc b/sql/init.cc
index c72787300b7..e43b12787ab 100644
--- a/sql/init.cc
+++ b/sql/init.cc
@@ -33,7 +33,7 @@ void unireg_init(ulong options)
 {
   DBUG_ENTER("unireg_init");
 
-  MYSYS_PROGRAM_DONT_USE_CURSES();
+  error_handler_hook = my_message_stderr;
   abort_loop=0;
 
   my_disable_async_io=1;		/* aioread is only in shared library */
diff --git a/sql/item.cc b/sql/item.cc
index 5905c3ee090..c59a17a0ea3 100644
--- a/sql/item.cc
+++ b/sql/item.cc
@@ -1,4 +1,4 @@
-/* Copyright 2000-2008 MySQL AB, 2008 Sun Microsystems, Inc.
+/* Copyright (c) 2000, 2010 Oracle and/or its affiliates. All rights reserved.
 
    This program is free software; you can redistribute it and/or modify
    it under the terms of the GNU General Public License as published by
@@ -3434,9 +3434,9 @@ Item_param::set_param_type_and_swap_value(Item_param *src)
 bool
 Item_param::set_value(THD *thd, sp_rcontext *ctx, Item **it)
 {
-  Item *value= *it;
+  Item *arg= *it;
 
-  if (value->is_null())
+  if (arg->is_null())
   {
     set_null();
     return FALSE;
@@ -3444,12 +3444,12 @@ Item_param::set_value(THD *thd, sp_rcontext *ctx, Item **it)
 
   null_value= FALSE;
 
-  switch (value->result_type()) {
+  switch (arg->result_type()) {
   case STRING_RESULT:
   {
     char str_buffer[STRING_BUFFER_USUAL_SIZE];
     String sv_buffer(str_buffer, sizeof(str_buffer), &my_charset_bin);
-    String *sv= value->val_str(&sv_buffer);
+    String *sv= arg->val_str(&sv_buffer);
 
     if (!sv)
       return TRUE;
@@ -3466,19 +3466,19 @@ Item_param::set_value(THD *thd, sp_rcontext *ctx, Item **it)
   }
 
   case REAL_RESULT:
-    set_double(value->val_real());
+    set_double(arg->val_real());
       param_type= MYSQL_TYPE_DOUBLE;
     break;
 
   case INT_RESULT:
-    set_int(value->val_int(), value->max_length);
+    set_int(arg->val_int(), arg->max_length);
     param_type= MYSQL_TYPE_LONG;
     break;
 
   case DECIMAL_RESULT:
   {
     my_decimal dv_buf;
-    my_decimal *dv= value->val_decimal(&dv_buf);
+    my_decimal *dv= arg->val_decimal(&dv_buf);
 
     if (!dv)
       return TRUE;
@@ -3498,8 +3498,8 @@ Item_param::set_value(THD *thd, sp_rcontext *ctx, Item **it)
     return FALSE;
   }
 
-  item_result_type= value->result_type();
-  item_type= value->type();
+  item_result_type= arg->result_type();
+  item_type= arg->type();
   return FALSE;
 }
 
@@ -7431,7 +7431,7 @@ void Item_cache_int::store(Item *item, longlong val_arg)
 String *Item_cache_int::val_str(String *str)
 {
   DBUG_ASSERT(fixed == 1);
-  if (!value_cached && !cache_value())
+  if (!has_value())
     return NULL;
   str->set(value, default_charset());
   return str;
@@ -7441,7 +7441,7 @@ String *Item_cache_int::val_str(String *str)
 my_decimal *Item_cache_int::val_decimal(my_decimal *decimal_val)
 {
   DBUG_ASSERT(fixed == 1);
-  if (!value_cached && !cache_value())
+  if (!has_value())
     return NULL;
   int2my_decimal(E_DEC_FATAL_ERROR, value, unsigned_flag, decimal_val);
   return decimal_val;
@@ -7450,7 +7450,7 @@ my_decimal *Item_cache_int::val_decimal(my_decimal *decimal_val)
 double Item_cache_int::val_real()
 {
   DBUG_ASSERT(fixed == 1);
-  if (!value_cached && !cache_value())
+  if (!has_value())
     return 0.0;
   return (double) value;
 }
@@ -7458,7 +7458,7 @@ double Item_cache_int::val_real()
 longlong Item_cache_int::val_int()
 {
   DBUG_ASSERT(fixed == 1);
-  if (!value_cached && !cache_value())
+  if (!has_value())
     return 0;
   return value;
 }
@@ -7514,7 +7514,7 @@ String *Item_cache_datetime::val_str(String *str)
 my_decimal *Item_cache_datetime::val_decimal(my_decimal *decimal_val)
 {
   DBUG_ASSERT(fixed == 1);
-  if (!value_cached && !cache_value_int())
+  if (!has_value())
     return NULL;
   int2my_decimal(E_DEC_FATAL_ERROR, int_value, unsigned_flag, decimal_val);
   return decimal_val;
@@ -7550,7 +7550,7 @@ bool Item_cache_real::cache_value()
 double Item_cache_real::val_real()
 {
   DBUG_ASSERT(fixed == 1);
-  if (!value_cached && !cache_value())
+  if (!has_value())
     return 0.0;
   return value;
 }
@@ -7558,7 +7558,7 @@ double Item_cache_real::val_real()
 longlong Item_cache_real::val_int()
 {
   DBUG_ASSERT(fixed == 1);
-  if (!value_cached && !cache_value())
+  if (!has_value())
     return 0;
   return (longlong) rint(value);
 }
@@ -7567,7 +7567,7 @@ longlong Item_cache_real::val_int()
 String* Item_cache_real::val_str(String *str)
 {
   DBUG_ASSERT(fixed == 1);
-  if (!value_cached && !cache_value())
+  if (!has_value())
     return NULL;
   str->set_real(value, decimals, default_charset());
   return str;
@@ -7577,7 +7577,7 @@ String* Item_cache_real::val_str(String *str)
 my_decimal *Item_cache_real::val_decimal(my_decimal *decimal_val)
 {
   DBUG_ASSERT(fixed == 1);
-  if (!value_cached && !cache_value())
+  if (!has_value())
     return NULL;
   double2my_decimal(E_DEC_FATAL_ERROR, value, decimal_val);
   return decimal_val;
@@ -7599,7 +7599,7 @@ double Item_cache_decimal::val_real()
 {
   DBUG_ASSERT(fixed);
   double res;
-  if (!value_cached && !cache_value())
+  if (!has_value())
     return 0.0;
   my_decimal2double(E_DEC_FATAL_ERROR, &decimal_value, &res);
   return res;
@@ -7609,7 +7609,7 @@ longlong Item_cache_decimal::val_int()
 {
   DBUG_ASSERT(fixed);
   longlong res;
-  if (!value_cached && !cache_value())
+  if (!has_value())
     return 0;
   my_decimal2int(E_DEC_FATAL_ERROR, &decimal_value, unsigned_flag, &res);
   return res;
@@ -7618,7 +7618,7 @@ longlong Item_cache_decimal::val_int()
 String* Item_cache_decimal::val_str(String *str)
 {
   DBUG_ASSERT(fixed);
-  if (!value_cached && !cache_value())
+  if (!has_value())
     return NULL;
   my_decimal_round(E_DEC_FATAL_ERROR, &decimal_value, decimals, FALSE,
                    &decimal_value);
@@ -7629,7 +7629,7 @@ String* Item_cache_decimal::val_str(String *str)
 my_decimal *Item_cache_decimal::val_decimal(my_decimal *val)
 {
   DBUG_ASSERT(fixed);
-  if (!value_cached && !cache_value())
+  if (!has_value())
     return NULL;
   return &decimal_value;
 }
@@ -7665,7 +7665,7 @@ double Item_cache_str::val_real()
   DBUG_ASSERT(fixed == 1);
   int err_not_used;
   char *end_not_used;
-  if (!value_cached && !cache_value())
+  if (!has_value())
     return 0.0;
   if (value)
     return my_strntod(value->charset(), (char*) value->ptr(),
@@ -7678,7 +7678,7 @@ longlong Item_cache_str::val_int()
 {
   DBUG_ASSERT(fixed == 1);
   int err;
-  if (!value_cached && !cache_value())
+  if (!has_value())
     return 0;
   if (value)
     return my_strntoll(value->charset(), value->ptr(),
@@ -7691,7 +7691,7 @@ longlong Item_cache_str::val_int()
 String* Item_cache_str::val_str(String *str)
 {
   DBUG_ASSERT(fixed == 1);
-  if (!value_cached && !cache_value())
+  if (!has_value())
     return 0;
   return value;
 }
@@ -7700,7 +7700,7 @@ String* Item_cache_str::val_str(String *str)
 my_decimal *Item_cache_str::val_decimal(my_decimal *decimal_val)
 {
   DBUG_ASSERT(fixed == 1);
-  if (!value_cached && !cache_value())
+  if (!has_value())
     return NULL;
   if (value)
     string2my_decimal(E_DEC_FATAL_ERROR, value, decimal_val);
@@ -7712,7 +7712,7 @@ my_decimal *Item_cache_str::val_decimal(my_decimal *decimal_val)
 
 int Item_cache_str::save_in_field(Field *field, bool no_conversions)
 {
-  if (!value_cached && !cache_value())
+  if (!has_value())
     return 0;
   int res= Item_cache::save_in_field(field, no_conversions);
   return (is_varbinary && field->type() == MYSQL_TYPE_STRING &&
diff --git a/sql/item.h b/sql/item.h
index 5f4f96f97d3..e18fa43037a 100644
--- a/sql/item.h
+++ b/sql/item.h
@@ -34,6 +34,15 @@ void item_init(void);			/* Init item functions */
 class Item_field;
 class user_var_entry;
 
+
+static inline uint32
+char_to_byte_length_safe(uint32 char_length_arg, uint32 mbmaxlen_arg)
+{
+   ulonglong tmp= ((ulonglong) char_length_arg) * mbmaxlen_arg;
+   return (tmp > UINT_MAX32) ? (uint32) UINT_MAX32 : (uint32) tmp;
+}
+
+
 /*
    "Declared Type Collation"
    A combination of collation and its derivation.
@@ -1171,16 +1180,23 @@ public:
   { return max_length / collation.collation->mbmaxlen; }
   void fix_length_and_charset(uint32 max_char_length_arg, CHARSET_INFO *cs)
   {
-    max_length= max_char_length_arg * cs->mbmaxlen;
+    max_length= char_to_byte_length_safe(max_char_length_arg, cs->mbmaxlen);
     collation.collation= cs;
   }
   void fix_char_length(uint32 max_char_length_arg)
-  { max_length= max_char_length_arg * collation.collation->mbmaxlen; }
+  {
+    max_length= char_to_byte_length_safe(max_char_length_arg,
+                                         collation.collation->mbmaxlen);
+  }
   void fix_length_and_charset_datetime(uint32 max_char_length_arg)
   {
     collation.set(&my_charset_numeric, DERIVATION_NUMERIC, MY_REPERTOIRE_ASCII);
     fix_char_length(max_char_length_arg);
   }
+  /*
+    Return TRUE if the item points to a column of an outer-joined table.
+  */
+  virtual bool is_outer_field() const { DBUG_ASSERT(fixed); return FALSE; }
 };
 
 
@@ -1694,6 +1710,11 @@ public:
   int fix_outer_field(THD *thd, Field **field, Item **reference);
   virtual Item *update_value_transformer(uchar *select_arg);
   virtual void print(String *str, enum_query_type query_type);
+  bool is_outer_field() const
+  {
+    DBUG_ASSERT(fixed);
+    return field->table->pos_in_table_list->outer_join;
+  }
   Field::geometry_type get_geometry_type() const
   {
     DBUG_ASSERT(field_type() == MYSQL_TYPE_GEOMETRY);
@@ -2506,7 +2527,13 @@ public:
     DBUG_ASSERT(fixed);
     return (*ref)->get_time(ltime);
   }
-  bool basic_const_item() { return (*ref)->basic_const_item(); }
+  virtual bool basic_const_item() const { return (*ref)->basic_const_item(); }
+  bool is_outer_field() const
+  {
+    DBUG_ASSERT(fixed);
+    DBUG_ASSERT(ref);
+    return (*ref)->is_outer_field();
+  }
 
 };
 
@@ -3191,6 +3218,15 @@ public:
   {
     return this == item;
   }
+  /**
+     Check if saved item has a non-NULL value.
+     Will cache value of saved item if not already done. 
+     @return TRUE if cached value is non-NULL.
+   */
+  bool has_value()
+  {
+    return (value_cached || cache_value()) && !null_value;
+  }
   virtual void store(Item *item);
   virtual bool cache_value()= 0;
   bool basic_const_item() const
@@ -3358,6 +3394,7 @@ public:
     cmp_context= STRING_RESULT;
   }
 
+  virtual void store(Item *item) { Item_cache::store(item); }
   void store(Item *item, longlong val_arg);
   double val_real();
   longlong val_int();
diff --git a/sql/item_cmpfunc.cc b/sql/item_cmpfunc.cc
index 19e8385539f..3c871bc0663 100644
--- a/sql/item_cmpfunc.cc
+++ b/sql/item_cmpfunc.cc
@@ -5466,7 +5466,21 @@ void Item_equal::update_const()
   Item *item;
   while ((item= it++))
   {
-    if (item->const_item())
+    if (item->const_item() &&
+        /*
+          Don't propagate constant status of outer-joined column.
+          Such a constant status here is a result of:
+            a) empty outer-joined table: in this case such a column has a
+               value of NULL; but at the same time other arguments of
+               Item_equal don't have to be NULLs and the value of the whole
+               multiple equivalence expression doesn't have to be NULL or FALSE
+               because of the outer join nature;
+          or
+            b) outer-joined table contains only 1 row: the result of
+               this column is equal to a row field value *or* NULL.
+          Both values are inacceptable as Item_equal constants.
+        */
+        !item->is_outer_field())
     {
       it.remove();
       add(item);
@@ -5505,7 +5519,8 @@ void Item_equal::update_used_tables()
   {
     item->update_used_tables();
     used_tables_cache|= item->used_tables();
-    const_item_cache&= item->const_item();
+    /* see commentary at Item_equal::update_const() */
+    const_item_cache&= item->const_item() && !item->is_outer_field();
   }
 }
 
diff --git a/sql/item_sum.cc b/sql/item_sum.cc
index 917acb0e908..15927c4b11e 100644
--- a/sql/item_sum.cc
+++ b/sql/item_sum.cc
@@ -2827,6 +2827,7 @@ String *Item_sum_udf_str::val_str(String *str)
   @retval  1 : key1 > key2 
 */
 
+extern "C"
 int group_concat_key_cmp_with_distinct(void* arg, const void* key1, 
                                        const void* key2)
 {
@@ -2861,6 +2862,7 @@ int group_concat_key_cmp_with_distinct(void* arg, const void* key1,
   function of sort for syntax: GROUP_CONCAT(expr,... ORDER BY col,... )
 */
 
+extern "C"
 int group_concat_key_cmp_with_order(void* arg, const void* key1, 
                                     const void* key2)
 {
@@ -2905,13 +2907,16 @@ int group_concat_key_cmp_with_order(void* arg, const void* key1,
   Append data from current leaf to item->result.
 */
 
-int dump_leaf_key(uchar* key, element_count count __attribute__((unused)),
-                  Item_func_group_concat *item)
+extern "C"
+int dump_leaf_key(void* key_arg, element_count count __attribute__((unused)),
+                  void* item_arg)
 {
+  Item_func_group_concat *item= (Item_func_group_concat *) item_arg;
   TABLE *table= item->table;
   String tmp((char *)table->record[1], table->s->reclength,
              default_charset_info);
   String tmp2;
+  uchar *key= (uchar *) key_arg;
   String *result= &item->result;
   Item **arg= item->args, **arg_end= item->args + item->arg_count_field;
   uint old_length= result->length();
@@ -3385,8 +3390,7 @@ String* Item_func_group_concat::val_str(String* str)
     return 0;
   if (no_appended && tree)
     /* Tree is used for sorting as in ORDER BY */
-    tree_walk(tree, (tree_walk_action)&dump_leaf_key, (void*)this,
-              left_root_right);
+    tree_walk(tree, &dump_leaf_key, this, left_root_right);
   return &result;
 }
 
diff --git a/sql/item_sum.h b/sql/item_sum.h
index 5f2c4f166e6..99fcb14d160 100644
--- a/sql/item_sum.h
+++ b/sql/item_sum.h
@@ -303,6 +303,8 @@ class st_select_lex;
 
 class Item_sum :public Item_result_field
 {
+  friend class Aggregator_distinct;
+
 protected:
   /**
     Aggregator class instance. Not set initially. Allocated only after
@@ -1317,6 +1319,16 @@ public:
 
 #endif /* HAVE_DLOPEN */
 
+C_MODE_START
+int group_concat_key_cmp_with_distinct(void* arg, const void* key1,
+                                       const void* key2);
+int group_concat_key_cmp_with_order(void* arg, const void* key1,
+                                    const void* key2);
+int dump_leaf_key(void* key_arg,
+                  element_count count __attribute__((unused)),
+                  void* item_arg);
+C_MODE_END
+
 class Item_func_group_concat : public Item_sum
 {
   TMP_TABLE_PARAM *tmp_table_param;
@@ -1356,9 +1368,9 @@ class Item_func_group_concat : public Item_sum
                                                 const void* key2);
   friend int group_concat_key_cmp_with_order(void* arg, const void* key1,
 					     const void* key2);
-  friend int dump_leaf_key(uchar* key,
+  friend int dump_leaf_key(void* key_arg,
                            element_count count __attribute__((unused)),
-			   Item_func_group_concat *group_concat_item);
+			   void* item_arg);
 
 public:
   Item_func_group_concat(Name_resolution_context *context_arg,
diff --git a/sql/key.cc b/sql/key.cc
index d593850ca10..582334620ad 100644
--- a/sql/key.cc
+++ b/sql/key.cc
@@ -354,6 +354,16 @@ void key_unpack(String *to,TABLE *table,uint idx)
     {
       CHARSET_INFO *cs= field->charset();
       field->val_str(&tmp);
+      /*
+        For BINARY(N) strip trailing zeroes to make
+        the error message nice-looking
+      */
+      if (field->binary() &&  field->type() == MYSQL_TYPE_STRING && tmp.length())
+      {
+        const char *tmp_end= tmp.ptr() + tmp.length();
+        while (tmp_end > tmp.ptr() && !*--tmp_end);
+        tmp.length(tmp_end - tmp.ptr() + 1);
+      }
       if (cs->mbmaxlen > 1 &&
           table->field[key_part->fieldnr - 1]->field_length !=
           key_part->length)
diff --git a/sql/lock.cc b/sql/lock.cc
index 758ea6cf914..3f13f15454a 100644
--- a/sql/lock.cc
+++ b/sql/lock.cc
@@ -533,20 +533,6 @@ void mysql_lock_remove(THD *thd, MYSQL_LOCK *locked,TABLE *table)
   }
 }
 
-/* Downgrade all locks on a table to new WRITE level from WRITE_ONLY */
-
-void mysql_lock_downgrade_write(THD *thd, TABLE *table,
-                                thr_lock_type new_lock_type)
-{
-  MYSQL_LOCK *locked;
-  if ((locked = get_lock_data(thd, &table, 1, GET_LOCK_UNLOCK)))
-  {
-    for (uint i=0; i < locked->lock_count; i++)
-      thr_downgrade_write_lock(locked->locks[i], new_lock_type);
-    my_free((uchar*) locked,MYF(0));
-  }
-}
-
 
 /** Abort all other threads waiting to get lock in table. */
 
diff --git a/sql/lock.h b/sql/lock.h
index 19b23f1f42b..f7c19913675 100644
--- a/sql/lock.h
+++ b/sql/lock.h
@@ -53,52 +53,22 @@ typedef struct st_mysql_lock MYSQL_LOCK;
                             MYSQL_OPEN_HAS_MDL_LOCK)
 
 
-#include "thr_lock.h"                           /* thr_lock_type */
-
-struct TABLE_LIST;
-class THD;
-struct TABLE;
-typedef struct st_mysql_lock MYSQL_LOCK;
-
 MYSQL_LOCK *mysql_lock_tables(THD *thd, TABLE **table, uint count, uint flags);
 void mysql_unlock_tables(THD *thd, MYSQL_LOCK *sql_lock);
 void mysql_unlock_read_tables(THD *thd, MYSQL_LOCK *sql_lock);
 void mysql_unlock_some_tables(THD *thd, TABLE **table,uint count);
 void mysql_lock_remove(THD *thd, MYSQL_LOCK *locked,TABLE *table);
 void mysql_lock_abort(THD *thd, TABLE *table, bool upgrade_lock);
-void mysql_lock_downgrade_write(THD *thd, TABLE *table,
-                                thr_lock_type new_lock_type);
 bool mysql_lock_abort_for_thread(THD *thd, TABLE *table);
 MYSQL_LOCK *mysql_lock_merge(MYSQL_LOCK *a,MYSQL_LOCK *b);
 TABLE_LIST *mysql_lock_have_duplicate(THD *thd, TABLE_LIST *needle,
                                       TABLE_LIST *haystack);
-bool lock_global_read_lock(THD *thd);
-void unlock_global_read_lock(THD *thd);
-bool wait_if_global_read_lock(THD *thd, bool abort_on_refresh,
-                              bool is_not_commit);
-void start_waiting_global_read_lock(THD *thd);
-bool make_global_read_lock_block_commit(THD *thd);
-bool set_protect_against_global_read_lock(void);
-void unset_protect_against_global_read_lock(void);
-/* Lock based on stored routine name */
-bool lock_routine_name(THD *thd, bool is_function, const char *db,
-                       const char *name);
 void broadcast_refresh(void);
-
 /* Lock based on name */
-int lock_and_wait_for_table_name(THD *thd, TABLE_LIST *table_list);
-int lock_table_name(THD *thd, TABLE_LIST *table_list, bool check_in_use);
-void unlock_table_name(THD *thd, TABLE_LIST *table_list);
-bool wait_for_locked_table_names(THD *thd, TABLE_LIST *table_list);
 bool lock_table_names(THD *thd, TABLE_LIST *table_list);
 void unlock_table_names(THD *thd);
-bool lock_table_names_exclusively(THD *thd, TABLE_LIST *table_list);
-bool is_table_name_exclusively_locked_by_this_thread(THD *thd, 
-                                                     TABLE_LIST *table_list);
-bool is_table_name_exclusively_locked_by_this_thread(THD *thd, uchar *key,
-                                                     int key_length);
-void broadcast_refresh(void);
-
-
+/* Lock based on stored routine name */
+bool lock_routine_name(THD *thd, bool is_function, const char *db,
+                       const char *name);
 
 #endif /* LOCK_INCLUDED */
diff --git a/sql/log.cc b/sql/log.cc
index a80e054178f..fd17e04b212 100644
--- a/sql/log.cc
+++ b/sql/log.cc
@@ -1669,7 +1669,7 @@ static int binlog_commit(handlerton *hton, THD *thd, bool all)
   DBUG_PRINT("debug",
              ("all: %d, in_transaction: %s, all.modified_non_trans_table: %s, stmt.modified_non_trans_table: %s",
               all,
-              YESNO(thd->in_multi_stmt_transaction()),
+              YESNO(thd->in_multi_stmt_transaction_mode()),
               YESNO(thd->transaction.all.modified_non_trans_table),
               YESNO(thd->transaction.stmt.modified_non_trans_table)));
 
@@ -4254,7 +4254,7 @@ bool use_trans_cache(const THD* thd, bool is_transactional)
 */
 bool ending_trans(THD* thd, const bool all)
 {
-  return (all || (!all && !thd->in_multi_stmt_transaction()));
+  return (all || (!all && !thd->in_multi_stmt_transaction_mode()));
 }
 
 /**
@@ -4357,7 +4357,7 @@ THD::binlog_start_trans_and_stmt()
       cache_mngr->trx_cache.get_prev_position() == MY_OFF_T_UNDEF)
   {
     this->binlog_set_stmt_begin();
-    if (in_multi_stmt_transaction())
+    if (in_multi_stmt_transaction_mode())
       trans_register_ha(this, TRUE, binlog_hton);
     trans_register_ha(this, FALSE, binlog_hton);
     /*
diff --git a/sql/log_event.cc b/sql/log_event.cc
index 5b2b86a9e42..cefc49f8c8a 100644
--- a/sql/log_event.cc
+++ b/sql/log_event.cc
@@ -2465,95 +2465,62 @@ Query_log_event::Query_log_event(THD* thd_arg, const char* query_arg,
   else
     time_zone_len= 0;
 
+  LEX *lex= thd->lex;
   /*
-    In what follows, we decide whether to write to the binary log or to use a
-    cache.
+    TRUE defines that either a trx-cache or stmt-cache must be used
+    and wrapped by a BEGIN...COMMIT. Otherwise, the statement will
+    be written directly to the binary log without being wrapped by
+    a BEGIN...COMMIT.
+
+    Note that a cache will not be used if the parameter direct is
+    TRUE.
   */
-  LEX *lex= thd->lex;
-  bool implicit_commit= FALSE;
-  bool force_trans= FALSE;
+  bool use_cache= FALSE;
+  /*
+    TRUE defines that the trx-cache must be used and by consequence
+    the use_cache is TRUE.
+
+    Note that a cache will not be used if the parameter direct is
+    TRUE.
+  */
+  bool trx_cache= FALSE;
   cache_type= Log_event::EVENT_INVALID_CACHE;
+
   switch (lex->sql_command)
   {
-    case SQLCOM_ALTER_DB:
-    case SQLCOM_CREATE_FUNCTION:
-    case SQLCOM_DROP_FUNCTION:
-    case SQLCOM_DROP_PROCEDURE:
-    case SQLCOM_INSTALL_PLUGIN:
-    case SQLCOM_UNINSTALL_PLUGIN:
-    case SQLCOM_ALTER_TABLESPACE:
-      implicit_commit= TRUE;
-      break;
     case SQLCOM_DROP_TABLE:
-      force_trans= lex->drop_temporary && thd->in_multi_stmt_transaction();
-      implicit_commit= !force_trans;
-      break;
-    case SQLCOM_ALTER_TABLE:
+      use_cache= trx_cache= (lex->drop_temporary &&
+                            thd->in_multi_stmt_transaction_mode());
+    break;
+
     case SQLCOM_CREATE_TABLE:
-      force_trans= (lex->create_info.options & HA_LEX_CREATE_TMP_TABLE) &&
-                    thd->in_multi_stmt_transaction();
-      implicit_commit= !force_trans &&
-                       !(lex->select_lex.item_list.elements &&
-                         thd->is_current_stmt_binlog_format_row());
+      use_cache= trx_cache=
+                 ((lex->create_info.options & HA_LEX_CREATE_TMP_TABLE) &&
+                   thd->in_multi_stmt_transaction_mode()) ||
+                 (lex->select_lex.item_list.elements &&
+                  thd->is_current_stmt_binlog_format_row());
       break;
     case SQLCOM_SET_OPTION:
-      implicit_commit= (lex->autocommit ? TRUE : FALSE);
-      break;
-    /*
-      Replace what follows after CF_AUTO_COMMIT_TRANS is backported by:
-
-      default:
-        implicit_commit= ((sql_command_flags[lex->sql_command] &
-                      CF_AUTO_COMMIT_TRANS));
+      use_cache= trx_cache= (lex->autocommit ? FALSE : TRUE);
       break;
-    */
-    case SQLCOM_CREATE_INDEX:
-    case SQLCOM_TRUNCATE:
-    case SQLCOM_CREATE_DB:
-    case SQLCOM_DROP_DB:
-    case SQLCOM_ALTER_DB_UPGRADE:
-    case SQLCOM_RENAME_TABLE:
-    case SQLCOM_DROP_INDEX:
-    case SQLCOM_CREATE_VIEW:
-    case SQLCOM_DROP_VIEW:
-    case SQLCOM_CREATE_TRIGGER:
-    case SQLCOM_DROP_TRIGGER:
-    case SQLCOM_CREATE_EVENT:
-    case SQLCOM_ALTER_EVENT:
-    case SQLCOM_DROP_EVENT:
-    case SQLCOM_REPAIR:
-    case SQLCOM_OPTIMIZE:
-    case SQLCOM_ANALYZE:
-    case SQLCOM_CREATE_USER:
-    case SQLCOM_DROP_USER:
-    case SQLCOM_RENAME_USER:
-    case SQLCOM_REVOKE_ALL:
-    case SQLCOM_REVOKE:
-    case SQLCOM_GRANT:
-    case SQLCOM_CREATE_PROCEDURE:
-    case SQLCOM_CREATE_SPFUNCTION:
-    case SQLCOM_ALTER_PROCEDURE:
-    case SQLCOM_ALTER_FUNCTION:
-    case SQLCOM_ASSIGN_TO_KEYCACHE:
-    case SQLCOM_PRELOAD_KEYS:
-    case SQLCOM_FLUSH:
-    case SQLCOM_RESET:
-    case SQLCOM_CHECK:
-      implicit_commit= TRUE;
+    case SQLCOM_RELEASE_SAVEPOINT:
+    case SQLCOM_ROLLBACK_TO_SAVEPOINT:
+    case SQLCOM_SAVEPOINT:
+      use_cache= trx_cache= TRUE;
       break;
     default:
-      implicit_commit= FALSE;
+      use_cache= sqlcom_can_generate_row_events(thd);
       break;
   }
 
-  if (implicit_commit || direct)
+  if (!use_cache || direct)
   {
     cache_type= Log_event::EVENT_NO_CACHE;
   }
   else
   {
-    cache_type= ((using_trans || stmt_has_updated_trans_table(thd) ||
-                 force_trans || thd->thread_temporary_used)
+    cache_type= ((using_trans || stmt_has_updated_trans_table(thd)
+                  || trx_cache || thd->thread_temporary_used)
                  ? Log_event::EVENT_TRANSACTIONAL_CACHE :
                  Log_event::EVENT_STMT_CACHE);
   }
@@ -3351,11 +3318,12 @@ compare_errors:
     */
     actual_error= thd->is_error() ? thd->stmt_da->sql_errno() : 0;
     DBUG_PRINT("info",("expected_error: %d  sql_errno: %d",
- 		       expected_error, actual_error));
+                       expected_error, actual_error));
+
     if ((expected_error && expected_error != actual_error &&
          !concurrency_error_code(expected_error)) &&
- 	!ignored_error_code(actual_error) &&
- 	!ignored_error_code(expected_error))
+        !ignored_error_code(actual_error) &&
+        !ignored_error_code(expected_error))
     {
       rli->report(ERROR_LEVEL, 0,
                       "\
@@ -3373,9 +3341,9 @@ Default database: '%s'. Query: '%s'",
       If we get the same error code as expected and it is not a concurrency
       issue, or should be ignored.
     */
-    else if ((expected_error == actual_error && 
+    else if ((expected_error == actual_error &&
               !concurrency_error_code(expected_error)) ||
- 	     ignored_error_code(actual_error))
+             ignored_error_code(actual_error))
     {
       DBUG_PRINT("info",("error ignored"));
       clear_all_errors(thd, const_cast<Relay_log_info*>(rli));
@@ -3394,7 +3362,7 @@ Default database: '%s'. Query: '%s'",
       If we expected a non-zero error code and get nothing and, it is a concurrency
       issue or should be ignored.
     */
-    else if (expected_error && !actual_error && 
+    else if (expected_error && !actual_error &&
              (concurrency_error_code(expected_error) ||
               ignored_error_code(expected_error)))
       trans_rollback_stmt(thd);
@@ -3435,12 +3403,13 @@ Default database: '%s'. Query: '%s'",
     */
   } /* End of if (db_ok(... */
 
-  {/**
-      The following failure injecion works in cooperation with tests 
+  {
+    /**
+      The following failure injecion works in cooperation with tests
       setting @@global.debug= 'd,stop_slave_middle_group'.
-      The sql thread receives the killed status and will proceed 
+      The sql thread receives the killed status and will proceed
       to shutdown trying to finish incomplete events group.
-   */
+    */
     DBUG_EXECUTE_IF("stop_slave_middle_group",
                     if (strcmp("COMMIT", query) != 0 &&
                         strcmp("BEGIN", query) != 0)
@@ -3455,7 +3424,7 @@ end:
     Probably we have set thd->query, thd->db, thd->catalog to point to places
     in the data_buf of this event. Now the event is going to be deleted
     probably, so data_buf will be freed, so the thd->... listed above will be
-    pointers to freed memory. 
+    pointers to freed memory.
     So we must set them to 0, so that those bad pointers values are not later
     used. Note that "cleanup" queries like automatic DROP TEMPORARY TABLE
     don't suffer from these assignments to 0 as DROP TEMPORARY
@@ -3465,7 +3434,7 @@ end:
   thd->set_db(NULL, 0);                 /* will free the current database */
   thd->set_query(NULL, 0);
   DBUG_PRINT("info", ("end: query= 0"));
-  close_thread_tables(thd);      
+  close_thread_tables(thd);
   /*
     As a disk space optimization, future masters will not log an event for
     LAST_INSERT_ID() if that function returned 0 (and thus they will be able
@@ -3767,7 +3736,7 @@ Format_description_log_event(uint8 binlog_ver, const char* server_ver)
     */
     if (post_header_len)
     {
-#ifndef DBUG_OFF      
+#ifndef DBUG_OFF
       // Allows us to sanity-check that all events initialized their
       // events (see the end of this 'if' block).
       memset(post_header_len, 255, number_of_event_types*sizeof(uint8));
@@ -4223,7 +4192,7 @@ void Load_log_event::print_query(bool need_db, const char *cs, char *buf,
 
   pos= strmov(pos, "LOAD DATA ");
 
-  if (thd->lex->lock_option == TL_WRITE_CONCURRENT_INSERT)
+  if (is_concurrent)
     pos= strmov(pos, "CONCURRENT ");
 
   if (fn_start)
@@ -4365,6 +4334,7 @@ bool Load_log_event::write_data_body(IO_CACHE* file)
 Load_log_event::Load_log_event(THD *thd_arg, sql_exchange *ex,
 			       const char *db_arg, const char *table_name_arg,
 			       List<Item> &fields_arg,
+                               bool is_concurrent_arg,
 			       enum enum_duplicates handle_dup,
 			       bool ignore, bool using_trans)
   :Log_event(thd_arg,
@@ -4375,7 +4345,8 @@ Load_log_event::Load_log_event(THD *thd_arg, sql_exchange *ex,
    num_fields(0),fields(0),
    field_lens(0),field_block_len(0),
    table_name(table_name_arg ? table_name_arg : ""),
-   db(db_arg), fname(ex->file_name), local_fname(FALSE)
+   db(db_arg), fname(ex->file_name), local_fname(FALSE),
+   is_concurrent(is_concurrent_arg)
 {
   time_t end_time;
   time(&end_time);
@@ -4456,7 +4427,13 @@ Load_log_event::Load_log_event(const char *buf, uint event_len,
                                const Format_description_log_event *description_event)
   :Log_event(buf, description_event), num_fields(0), fields(0),
    field_lens(0),field_block_len(0),
-   table_name(0), db(0), fname(0), local_fname(FALSE)
+   table_name(0), db(0), fname(0), local_fname(FALSE),
+   /*
+     Load_log_event which comes from the binary log does not contain
+     information about the type of insert which was used on the master.
+     Assume that it was an ordinary, non-concurrent LOAD DATA.
+    */
+   is_concurrent(FALSE)
 {
   DBUG_ENTER("Load_log_event");
   /*
@@ -4613,9 +4590,9 @@ void Load_log_event::print(FILE* file_arg, PRINT_EVENT_INFO* print_event_info,
     for (i = 0; i < num_fields; i++)
     {
       if (i)
-	my_b_printf(&cache, ",");
+        my_b_printf(&cache, ",");
       my_b_printf(&cache, "%s", field);
-	  
+
       field += field_lens[i]  + 1;
     }
     my_b_printf(&cache, ")");
@@ -4785,9 +4762,7 @@ int Load_log_event::do_apply_event(NET* net, Relay_log_info const *rli,
       thd->set_query(load_data_query, (uint) (end - load_data_query));
 
       if (sql_ex.opt_flags & REPLACE_FLAG)
-      {
-	handle_dup= DUP_REPLACE;
-      }
+        handle_dup= DUP_REPLACE;
       else if (sql_ex.opt_flags & IGNORE_FLAG)
       {
         ignore= 1;
@@ -4796,14 +4771,14 @@ int Load_log_event::do_apply_event(NET* net, Relay_log_info const *rli,
       else
       {
         /*
-	  When replication is running fine, if it was DUP_ERROR on the
+          When replication is running fine, if it was DUP_ERROR on the
           master then we could choose IGNORE here, because if DUP_ERROR
           suceeded on master, and data is identical on the master and slave,
           then there should be no uniqueness errors on slave, so IGNORE is
           the same as DUP_ERROR. But in the unlikely case of uniqueness errors
           (because the data on the master and slave happen to be different
-	  (user error or bug), we want LOAD DATA to print an error message on
-	  the slave to discover the problem.
+          (user error or bug), we want LOAD DATA to print an error message on
+          the slave to discover the problem.
 
           If reading from net (a 3.23 master), mysql_load() will change this
           to IGNORE.
@@ -4835,7 +4810,7 @@ int Load_log_event::do_apply_event(NET* net, Relay_log_info const *rli,
 
       ex.opt_enclosed = (sql_ex.opt_flags & OPT_ENCLOSED_FLAG);
       if (sql_ex.empty_flags & FIELD_TERM_EMPTY)
-	ex.field_term->length(0);
+        ex.field_term->length(0);
 
       ex.skip_lines = skip_lines;
       List<Item> field_list;
@@ -4844,12 +4819,10 @@ int Load_log_event::do_apply_event(NET* net, Relay_log_info const *rli,
       thd->variables.pseudo_thread_id= thread_id;
       if (net)
       {
-	// mysql_load will use thd->net to read the file
-	thd->net.vio = net->vio;
-	/*
-	  Make sure the client does not get confused about the packet sequence
-	*/
-	thd->net.pkt_nr = net->pkt_nr;
+        // mysql_load will use thd->net to read the file
+        thd->net.vio = net->vio;
+        // Make sure the client does not get confused about the packet sequence
+        thd->net.pkt_nr = net->pkt_nr;
       }
       /*
         It is safe to use tmp_list twice because we are not going to
@@ -4861,7 +4834,7 @@ int Load_log_event::do_apply_event(NET* net, Relay_log_info const *rli,
         thd->is_slave_error= 1;
       if (thd->cuted_fields)
       {
-	/* log_pos is the position of the LOAD event in the master log */
+        /* log_pos is the position of the LOAD event in the master log */
         sql_print_warning("Slave: load data infile on table '%s' at "
                           "log position %s in log '%s' produced %ld "
                           "warning(s). Default database: '%s'",
@@ -5609,10 +5582,10 @@ User_var_log_event(const char* buf,
   {
     type= (Item_result) buf[UV_VAL_IS_NULL];
     charset_number= uint4korr(buf + UV_VAL_IS_NULL + UV_VAL_TYPE_SIZE);
-    val_len= uint4korr(buf + UV_VAL_IS_NULL + UV_VAL_TYPE_SIZE + 
-		       UV_CHARSET_NUMBER_SIZE);
+    val_len= uint4korr(buf + UV_VAL_IS_NULL + UV_VAL_TYPE_SIZE +
+                       UV_CHARSET_NUMBER_SIZE);
     val= (char *) (buf + UV_VAL_IS_NULL + UV_VAL_TYPE_SIZE +
-		   UV_CHARSET_NUMBER_SIZE + UV_VAL_LEN_SIZE);
+                   UV_CHARSET_NUMBER_SIZE + UV_VAL_LEN_SIZE);
 
     /**
       We need to check if this is from an old server
@@ -5696,9 +5669,9 @@ bool User_var_log_event::write(IO_CACHE* file)
 
   return (write_header(file, event_length) ||
           my_b_safe_write(file, (uchar*) buf, sizeof(buf))   ||
-	  my_b_safe_write(file, (uchar*) name, name_len)     ||
-	  my_b_safe_write(file, (uchar*) buf1, buf1_length) ||
-	  my_b_safe_write(file, pos, val_len) ||
+          my_b_safe_write(file, (uchar*) name, name_len)     ||
+          my_b_safe_write(file, (uchar*) buf1, buf1_length) ||
+          my_b_safe_write(file, pos, val_len) ||
           my_b_safe_write(file, &flags, unsigned_len));
 }
 #endif
@@ -5972,7 +5945,7 @@ Slave_log_event::Slave_log_event(THD* thd_arg,
   master_log_len = strlen(rli->group_master_log_name);
   // on OOM, just do not initialize the structure and print the error
   if ((mem_pool = (char*)my_malloc(get_data_size() + 1,
-				   MYF(MY_WME))))
+                                   MYF(MY_WME))))
   {
     master_host = mem_pool + SL_MASTER_HOST_OFFSET ;
     memcpy(master_host, mi->host, master_host_len + 1);
@@ -5981,7 +5954,7 @@ Slave_log_event::Slave_log_event(THD* thd_arg,
     master_port = mi->port;
     master_pos = rli->group_master_log_pos;
     DBUG_PRINT("info", ("master_log: %s  pos: %lu", master_log,
-			(ulong) master_pos));
+                        (ulong) master_pos));
   }
   else
     sql_print_error("Out of memory while recording slave event");
@@ -6146,11 +6119,14 @@ int Stop_log_event::do_update_pos(Relay_log_info *rli)
 Create_file_log_event::
 Create_file_log_event(THD* thd_arg, sql_exchange* ex,
 		      const char* db_arg, const char* table_name_arg,
-		      List<Item>& fields_arg, enum enum_duplicates handle_dup,
+                      List<Item>& fields_arg,
+                      bool is_concurrent_arg,
+                      enum enum_duplicates handle_dup,
                       bool ignore,
 		      uchar* block_arg, uint block_len_arg, bool using_trans)
-  :Load_log_event(thd_arg,ex,db_arg,table_name_arg,fields_arg,handle_dup, ignore,
-		  using_trans),
+  :Load_log_event(thd_arg, ex, db_arg, table_name_arg, fields_arg,
+                  is_concurrent_arg,
+                  handle_dup, ignore, using_trans),
    fake_base(0), block(block_arg), event_buf(0), block_len(block_len_arg),
    file_id(thd_arg->file_id = mysql_bin_log.next_file_id())
 {
diff --git a/sql/log_event.h b/sql/log_event.h
index 1b5c8968210..bd95c74b6c5 100644
--- a/sql/log_event.h
+++ b/sql/log_event.h
@@ -2091,6 +2091,17 @@ public:
   uint32 skip_lines;
   sql_ex_info sql_ex;
   bool local_fname;
+  /**
+    Indicates that this event corresponds to LOAD DATA CONCURRENT,
+
+    @note Since Load_log_event event coming from the binary log
+          lacks information whether LOAD DATA on master was concurrent
+          or not, this flag is only set to TRUE for an auxiliary
+          Load_log_event object which is used in mysql_load() to
+          re-construct LOAD DATA statement from function parameters,
+          for logging.
+  */
+  bool is_concurrent;
 
   /* fname doesn't point to memory inside Log_event::temp_buf  */
   void set_fname_outside_temp_buf(const char *afname, uint alen)
@@ -2111,7 +2122,9 @@ public:
 
   Load_log_event(THD* thd, sql_exchange* ex, const char* db_arg,
 		 const char* table_name_arg,
-		 List<Item>& fields_arg, enum enum_duplicates handle_dup, bool ignore,
+		 List<Item>& fields_arg,
+                 bool is_concurrent_arg,
+                 enum enum_duplicates handle_dup, bool ignore,
 		 bool using_trans);
   void set_fields(const char* db, List<Item> &fields_arg,
                   Name_resolution_context *context);
@@ -2730,6 +2743,7 @@ public:
   Create_file_log_event(THD* thd, sql_exchange* ex, const char* db_arg,
 			const char* table_name_arg,
 			List<Item>& fields_arg,
+                        bool is_concurrent_arg,
 			enum enum_duplicates handle_dup, bool ignore,
 			uchar* block_arg, uint block_len_arg,
 			bool using_trans);
diff --git a/sql/mdl.h b/sql/mdl.h
index 2fb21a5aa18..89a679be264 100644
--- a/sql/mdl.h
+++ b/sql/mdl.h
@@ -718,7 +718,7 @@ void mdl_destroy();
 extern bool mysql_notify_thread_having_shared_lock(THD *thd, THD *in_use,
                                                    bool needs_thr_lock_abort);
 extern void mysql_ha_flush(THD *thd);
-extern "C" const char *set_thd_proc_info(THD *thd, const char *info,
+extern "C" const char *set_thd_proc_info(void *thd_arg, const char *info,
                                          const char *calling_function,
                                          const char *calling_file,
                                          const unsigned int calling_line);
diff --git a/sql/mysqld.cc b/sql/mysqld.cc
index 20a168dc5ae..2770f056938 100644
--- a/sql/mysqld.cc
+++ b/sql/mysqld.cc
@@ -857,12 +857,15 @@ void Buffered_logs::print()
 /** Logs reported before a logger is available. */
 static Buffered_logs buffered_logs;
 
+#ifndef EMBEDDED_LIBRARY
 /**
   Error reporter that buffer log messages.
   @param level          log message level
   @param format         log message format string
 */
-void buffered_option_error_reporter(enum loglevel level, const char *format, ...)
+C_MODE_START
+static void buffered_option_error_reporter(enum loglevel level,
+                                           const char *format, ...)
 {
   va_list args;
   char buffer[1024];
@@ -872,6 +875,8 @@ void buffered_option_error_reporter(enum loglevel level, const char *format, ...
   va_end(args);
   buffered_logs.buffer(level, buffer);
 }
+C_MODE_END
+#endif /* !EMBEDDED_LIBRARY */
 #endif /* WITH_PERFSCHEMA_STORAGE_ENGINE */
 
 static my_socket unix_sock,ip_sock;
@@ -973,7 +978,6 @@ uint connection_count= 0;
 
 pthread_handler_t signal_hand(void *arg);
 static int mysql_init_variables(void);
-extern "C" void option_error_reporter(enum loglevel level, const char *format, ...);
 static int get_options(int *argc_ptr, char ***argv_ptr);
 static bool add_terminator(DYNAMIC_ARRAY *options);
 extern "C" my_bool mysqld_get_one_option(int, const struct my_option *, char *);
@@ -3608,7 +3612,6 @@ static int init_common_variables()
   if (item_create_init())
     return 1;
   item_init();
-  mysys_uses_curses=0;
 #ifdef USE_REGEX
   my_regex_init(&my_charset_latin1);
 #endif
@@ -4028,9 +4031,8 @@ static int init_server_components()
     }
   }
 
-  proc_info_hook= (const char *(*)(void *, const char *, const char *,
-                                   const char *, const unsigned int))
-                  set_thd_proc_info;
+  proc_info_hook= set_thd_proc_info;
+
 #ifdef WITH_PERFSCHEMA_STORAGE_ENGINE
   /*
     Parsing the performance schema command line option may have reported
@@ -5351,11 +5353,11 @@ inline void kill_broken_server()
 
 void handle_connections_sockets()
 {
-  my_socket sock,new_sock;
+  my_socket UNINIT_VAR(sock), UNINIT_VAR(new_sock);
   uint error_count=0;
   THD *thd;
   struct sockaddr_storage cAddr;
-  int ip_flags=0,socket_flags=0,flags,retval;
+  int ip_flags=0,socket_flags=0,flags=0,retval;
   st_vio *vio_tmp;
 #ifdef HAVE_POLL
   int socket_count= 0;
@@ -5367,8 +5369,6 @@ void handle_connections_sockets()
 
   DBUG_ENTER("handle_connections_sockets");
 
-  LINT_INIT(new_sock);
-
 #ifndef HAVE_POLL
   FD_ZERO(&clientFDs);
 #endif
@@ -7430,10 +7430,7 @@ mysqld_get_one_option(int optid,
 
 /** Handle arguments for multiple key caches. */
 
-extern "C" int mysql_getopt_value(uchar **value,
-                                  const char *keyname, uint key_length,
-                                  const struct my_option *option,
-                                  int *error);
+C_MODE_START
 
 static uchar* *
 mysql_getopt_value(const char *keyname, uint key_length,
@@ -7469,7 +7466,7 @@ mysql_getopt_value(const char *keyname, uint key_length,
   return option->value;
 }
 
-void option_error_reporter(enum loglevel level, const char *format, ...)
+static void option_error_reporter(enum loglevel level, const char *format, ...)
 {
   va_list args;
   va_start(args, format);
@@ -7483,6 +7480,7 @@ void option_error_reporter(enum loglevel level, const char *format, ...)
   va_end(args);
 }
 
+C_MODE_END
 
 /**
   Get server options from the command line,
@@ -7983,8 +7981,9 @@ PSI_mutex_key key_BINLOG_LOCK_index, key_BINLOG_LOCK_prep_xids,
   key_master_info_data_lock, key_master_info_run_lock,
   key_mutex_slave_reporting_capability_err_lock, key_relay_log_info_data_lock,
   key_relay_log_info_log_space_lock, key_relay_log_info_run_lock,
-  key_structure_guard_mutex, key_TABLE_SHARE_LOCK_ha_data, key_LOCK_error_messages,
-  key_LOG_INFO_lock, key_LOCK_thread_count;
+  key_structure_guard_mutex, key_TABLE_SHARE_LOCK_ha_data,
+  key_LOCK_error_messages, key_LOG_INFO_lock, key_LOCK_thread_count,
+  key_PARTITION_LOCK_auto_inc;
 
 static PSI_mutex_info all_server_mutexes[]=
 {
@@ -8038,7 +8037,8 @@ static PSI_mutex_info all_server_mutexes[]=
   { &key_TABLE_SHARE_LOCK_ha_data, "TABLE_SHARE::LOCK_ha_data", 0},
   { &key_LOCK_error_messages, "LOCK_error_messages", PSI_FLAG_GLOBAL},
   { &key_LOG_INFO_lock, "LOG_INFO::lock", 0},
-  { &key_LOCK_thread_count, "LOCK_thread_count", PSI_FLAG_GLOBAL}
+  { &key_LOCK_thread_count, "LOCK_thread_count", PSI_FLAG_GLOBAL},
+  { &key_PARTITION_LOCK_auto_inc, "HA_DATA_PARTITION::LOCK_auto_inc", 0}
 };
 
 PSI_rwlock_key key_rwlock_LOCK_grant, key_rwlock_LOCK_logger,
diff --git a/sql/mysqld.h b/sql/mysqld.h
index 9a6e43e4321..e14cd15ceb8 100644
--- a/sql/mysqld.h
+++ b/sql/mysqld.h
@@ -242,8 +242,8 @@ extern PSI_mutex_key key_BINLOG_LOCK_index, key_BINLOG_LOCK_prep_xids,
   key_master_info_data_lock, key_master_info_run_lock,
   key_mutex_slave_reporting_capability_err_lock, key_relay_log_info_data_lock,
   key_relay_log_info_log_space_lock, key_relay_log_info_run_lock,
-  key_structure_guard_mutex, key_TABLE_SHARE_LOCK_ha_data, key_LOCK_error_messages,
-  key_LOCK_thread_count;
+  key_structure_guard_mutex, key_TABLE_SHARE_LOCK_ha_data,
+  key_LOCK_error_messages, key_LOCK_thread_count, key_PARTITION_LOCK_auto_inc;
 
 extern PSI_rwlock_key key_rwlock_LOCK_grant, key_rwlock_LOCK_logger,
   key_rwlock_LOCK_sys_init_connect, key_rwlock_LOCK_sys_init_slave,
diff --git a/sql/opt_range.cc b/sql/opt_range.cc
index 5e985625c78..9363b637862 100644
--- a/sql/opt_range.cc
+++ b/sql/opt_range.cc
@@ -1539,6 +1539,29 @@ QUICK_ROR_UNION_SELECT::QUICK_ROR_UNION_SELECT(THD *thd_param,
 
 
 /*
+  Comparison function to be used QUICK_ROR_UNION_SELECT::queue priority
+  queue.
+
+  SYNPOSIS
+    QUICK_ROR_UNION_SELECT_queue_cmp()
+      arg   Pointer to QUICK_ROR_UNION_SELECT
+      val1  First merged select
+      val2  Second merged select
+*/
+
+C_MODE_START
+
+static int QUICK_ROR_UNION_SELECT_queue_cmp(void *arg, uchar *val1, uchar *val2)
+{
+  QUICK_ROR_UNION_SELECT *self= (QUICK_ROR_UNION_SELECT*)arg;
+  return self->head->file->cmp_ref(((QUICK_SELECT_I*)val1)->last_rowid,
+                                   ((QUICK_SELECT_I*)val2)->last_rowid);
+}
+
+C_MODE_END
+
+
+/*
   Do post-constructor initialization.
   SYNOPSIS
     QUICK_ROR_UNION_SELECT::init()
@@ -1552,7 +1575,7 @@ int QUICK_ROR_UNION_SELECT::init()
 {
   DBUG_ENTER("QUICK_ROR_UNION_SELECT::init");
   if (init_queue(&queue, quick_selects.elements, 0,
-                 FALSE , QUICK_ROR_UNION_SELECT::queue_cmp,
+                 FALSE , QUICK_ROR_UNION_SELECT_queue_cmp,
                  (void*) this))
   {
     bzero(&queue, sizeof(QUEUE));
@@ -1567,25 +1590,6 @@ int QUICK_ROR_UNION_SELECT::init()
 
 
 /*
-  Comparison function to be used QUICK_ROR_UNION_SELECT::queue priority
-  queue.
-
-  SYNPOSIS
-    QUICK_ROR_UNION_SELECT::queue_cmp()
-      arg   Pointer to QUICK_ROR_UNION_SELECT
-      val1  First merged select
-      val2  Second merged select
-*/
-
-int QUICK_ROR_UNION_SELECT::queue_cmp(void *arg, uchar *val1, uchar *val2)
-{
-  QUICK_ROR_UNION_SELECT *self= (QUICK_ROR_UNION_SELECT*)arg;
-  return self->head->file->cmp_ref(((QUICK_SELECT_I*)val1)->last_rowid,
-                                   ((QUICK_SELECT_I*)val2)->last_rowid);
-}
-
-
-/*
   Initialize quick select for row retrieval.
   SYNOPSIS
     reset()
diff --git a/sql/opt_range.h b/sql/opt_range.h
index 85d59671b42..72f2eb4b51d 100644
--- a/sql/opt_range.h
+++ b/sql/opt_range.h
@@ -657,7 +657,6 @@ public:
   bool have_prev_rowid; /* true if prev_rowid has valid data */
   uint rowid_length;    /* table rowid length */
 private:
-  static int queue_cmp(void *arg, uchar *val1, uchar *val2);
   bool scans_inited; 
 };
 
diff --git a/sql/partition_info.cc b/sql/partition_info.cc
index b6189ed3fc4..5b0b681c3a6 100644
--- a/sql/partition_info.cc
+++ b/sql/partition_info.cc
@@ -1,4 +1,4 @@
-/* Copyright (C) 2006-2008 MySQL AB, Sun Microsystems Inc. 2008-2009
+/* Copyright (c) 2006, 2010 Oracle and/or its affiliates. All rights reserved.
 
    This program is free software; you can redistribute it and/or modify
    it under the terms of the GNU General Public License as published by
@@ -603,12 +603,12 @@ bool partition_info::check_engine_mix(handlerton *engine_type,
 {
   handlerton *old_engine_type= engine_type;
   bool first= TRUE;
-  uint num_parts= partitions.elements;
+  uint n_parts= partitions.elements;
   DBUG_ENTER("partition_info::check_engine_mix");
   DBUG_PRINT("info", ("in: engine_type = %s, table_engine_set = %u",
                        ha_resolve_storage_engine_name(engine_type),
                        table_engine_set));
-  if (num_parts)
+  if (n_parts)
   {
     List_iterator<partition_element> part_it(partitions);
     uint i= 0;
@@ -621,7 +621,7 @@ bool partition_info::check_engine_mix(handlerton *engine_type,
       if (is_sub_partitioned() &&
           part_elem->subpartitions.elements)
       {
-        uint num_subparts= part_elem->subpartitions.elements;
+        uint n_subparts= part_elem->subpartitions.elements;
         uint j= 0;
         List_iterator<partition_element> sub_it(part_elem->subpartitions);
         do
@@ -633,7 +633,7 @@ bool partition_info::check_engine_mix(handlerton *engine_type,
           if (check_engine_condition(sub_elem, table_engine_set,
                                      &engine_type, &first))
             goto error;
-        } while (++j < num_subparts);
+        } while (++j < n_subparts);
         /* ensure that the partition also has correct engine */
         if (check_engine_condition(part_elem, table_engine_set,
                                    &engine_type, &first))
@@ -642,7 +642,7 @@ bool partition_info::check_engine_mix(handlerton *engine_type,
       else if (check_engine_condition(part_elem, table_engine_set,
                                       &engine_type, &first))
         goto error;
-    } while (++i < num_parts);
+    } while (++i < n_parts);
   }
   DBUG_PRINT("info", ("engine_type = %s",
                        ha_resolve_storage_engine_name(engine_type)));
@@ -701,12 +701,11 @@ bool partition_info::check_range_constants(THD *thd)
   if (column_list)
   {
     part_column_list_val *loc_range_col_array;
-    part_column_list_val *current_largest_col_val;
+    part_column_list_val *UNINIT_VAR(current_largest_col_val);
     uint num_column_values= part_field_list.elements;
     uint size_entries= sizeof(part_column_list_val) * num_column_values;
     range_col_array= (part_column_list_val*)sql_calloc(num_parts *
                                                        size_entries);
-    LINT_INIT(current_largest_col_val);
     if (unlikely(range_col_array == NULL))
     {
       mem_alloc_error(num_parts * size_entries);
@@ -739,12 +738,10 @@ bool partition_info::check_range_constants(THD *thd)
   }
   else
   {
-    longlong current_largest;
+    longlong UNINIT_VAR(current_largest);
     longlong part_range_value;
     bool signed_flag= !part_expr->unsigned_flag;
 
-    LINT_INIT(current_largest);
-
     part_result_type= INT_RESULT;
     range_int_array= (longlong*)sql_alloc(num_parts * sizeof(longlong));
     if (unlikely(range_int_array == NULL))
@@ -805,7 +802,8 @@ range_not_increasing_error:
     -1                 a < b
 */
 
-int partition_info::list_part_cmp(const void* a, const void* b)
+extern "C"
+int partition_info_list_part_cmp(const void* a, const void* b)
 {
   longlong a1= ((LIST_PART_ENTRY*)a)->list_value;
   longlong b1= ((LIST_PART_ENTRY*)b)->list_value;
@@ -817,7 +815,14 @@ int partition_info::list_part_cmp(const void* a, const void* b)
     return 0;
 }
 
- /*
+
+int partition_info::list_part_cmp(const void* a, const void* b)
+{
+  return partition_info_list_part_cmp(a, b);
+}
+
+
+/*
   Compare two lists of column values in RANGE/LIST partitioning
   SYNOPSIS
     compare_column_values()
@@ -829,8 +834,9 @@ int partition_info::list_part_cmp(const void* a, const void* b)
     +1                       First argument is larger
 */
 
-int partition_info::compare_column_values(const void *first_arg,
-                                          const void *second_arg)
+extern "C"
+int partition_info_compare_column_values(const void *first_arg,
+                                         const void *second_arg)
 {
   const part_column_list_val *first= (part_column_list_val*)first_arg;
   const part_column_list_val *second= (part_column_list_val*)second_arg;
@@ -866,6 +872,14 @@ int partition_info::compare_column_values(const void *first_arg,
   return 0;
 }
 
+
+int partition_info::compare_column_values(const void *first_arg,
+                                          const void *second_arg)
+{
+  return partition_info_compare_column_values(first_arg, second_arg);
+}
+
+
 /*
   This routine allocates an array for all list constants to achieve a fast
   check what partition a certain value belongs to. At the same time it does
@@ -894,10 +908,11 @@ bool partition_info::check_list_constants(THD *thd)
   part_elem_value *list_value;
   bool result= TRUE;
   longlong type_add, calc_value;
-  void *curr_value, *prev_value;
+  void *curr_value;
+  void *UNINIT_VAR(prev_value);
   partition_element* part_def;
   bool found_null= FALSE;
-  int (*compare_func)(const void *, const void*);
+  qsort_cmp compare_func;
   void *ptr;
   List_iterator<partition_element> list_func_it(partitions);
   DBUG_ENTER("partition_info::check_list_constants");
@@ -954,7 +969,7 @@ bool partition_info::check_list_constants(THD *thd)
     part_column_list_val *loc_list_col_array;
     loc_list_col_array= (part_column_list_val*)ptr;
     list_col_array= (part_column_list_val*)ptr;
-    compare_func= compare_column_values;
+    compare_func= partition_info_compare_column_values;
     i= 0;
     do
     {
@@ -974,7 +989,7 @@ bool partition_info::check_list_constants(THD *thd)
   }
   else
   {
-    compare_func= list_part_cmp;
+    compare_func= partition_info_list_part_cmp;
     list_array= (LIST_PART_ENTRY*)ptr;
     i= 0;
     /*
@@ -1009,7 +1024,6 @@ bool partition_info::check_list_constants(THD *thd)
              compare_func);
 
     i= 0;
-    LINT_INIT(prev_value);
     do
     {
       DBUG_ASSERT(i < num_list_values);
@@ -1032,6 +1046,30 @@ end:
   DBUG_RETURN(result);
 }
 
+/**
+  Check if we allow DATA/INDEX DIRECTORY, if not warn and set them to NULL.
+
+  @param thd  THD also containing sql_mode (looks from MODE_NO_DIR_IN_CREATE).
+  @param part_elem partition_element to check.
+*/
+static void warn_if_dir_in_part_elem(THD *thd, partition_element *part_elem)
+{
+#ifdef HAVE_READLINK
+  if (!my_use_symdir || (thd->variables.sql_mode & MODE_NO_DIR_IN_CREATE))
+#endif
+  {
+    if (part_elem->data_file_name)
+      push_warning_printf(thd, MYSQL_ERROR::WARN_LEVEL_WARN,
+                          WARN_OPTION_IGNORED, ER(WARN_OPTION_IGNORED),
+                          "DATA DIRECTORY");
+    if (part_elem->index_file_name)
+      push_warning_printf(thd, MYSQL_ERROR::WARN_LEVEL_WARN,
+                          WARN_OPTION_IGNORED, ER(WARN_OPTION_IGNORED),
+                          "INDEX DIRECTORY");
+    part_elem->data_file_name= part_elem->index_file_name= NULL;
+  }
+}
+
 
 /*
   This code is used early in the CREATE TABLE and ALTER TABLE process.
@@ -1169,20 +1207,7 @@ bool partition_info::check_partition_info(THD *thd, handlerton **eng_type,
     do
     {
       partition_element *part_elem= part_it++;
-#ifdef HAVE_READLINK
-      if (!my_use_symdir || (thd->variables.sql_mode & MODE_NO_DIR_IN_CREATE))
-#endif
-      {
-        if (part_elem->data_file_name)
-          push_warning_printf(thd, MYSQL_ERROR::WARN_LEVEL_WARN,
-                              WARN_OPTION_IGNORED, ER(WARN_OPTION_IGNORED),
-                              "DATA DIRECTORY");
-        if (part_elem->index_file_name)
-          push_warning_printf(thd, MYSQL_ERROR::WARN_LEVEL_WARN,
-                              WARN_OPTION_IGNORED, ER(WARN_OPTION_IGNORED),
-                              "INDEX DIRECTORY");
-        part_elem->data_file_name= part_elem->index_file_name= NULL;
-      }
+      warn_if_dir_in_part_elem(thd, part_elem);
       if (!is_sub_partitioned())
       {
         if (part_elem->engine_type == NULL)
@@ -1208,6 +1233,7 @@ bool partition_info::check_partition_info(THD *thd, handlerton **eng_type,
         do
         {
           sub_elem= sub_it++;
+          warn_if_dir_in_part_elem(thd, sub_elem);
           if (check_table_name(sub_elem->partition_name,
                                strlen(sub_elem->partition_name), FALSE))
           {
@@ -1306,15 +1332,15 @@ end:
   RETURN VALUES
 */
 
-void partition_info::print_no_partition_found(TABLE *table)
+void partition_info::print_no_partition_found(TABLE *table_arg)
 {
   char buf[100];
   char *buf_ptr= (char*)&buf;
   TABLE_LIST table_list;
 
   bzero(&table_list, sizeof(table_list));
-  table_list.db= table->s->db.str;
-  table_list.table_name= table->s->table_name.str;
+  table_list.db= table_arg->s->db.str;
+  table_list.table_name= table_arg->s->table_name.str;
 
   if (check_single_table_access(current_thd,
                                 SELECT_ACL, &table_list, TRUE))
@@ -1328,13 +1354,13 @@ void partition_info::print_no_partition_found(TABLE *table)
       buf_ptr= (char*)"from column_list";
     else
     {
-      my_bitmap_map *old_map= dbug_tmp_use_all_columns(table, table->read_set);
+      my_bitmap_map *old_map= dbug_tmp_use_all_columns(table_arg, table_arg->read_set);
       if (part_expr->null_value)
         buf_ptr= (char*)"NULL";
       else
         longlong2str(err_value, buf,
                      part_expr->unsigned_flag ? 10 : -10);
-      dbug_tmp_restore_column_map(table->read_set, old_map);
+      dbug_tmp_restore_column_map(table_arg->read_set, old_map);
     }
     my_error(ER_NO_PARTITION_FOR_GIVEN_VALUE, MYF(0), buf_ptr);
   }
@@ -1994,7 +2020,7 @@ bool partition_info::fix_column_value_functions(THD *thd,
                                                 part_elem_value *val,
                                                 uint part_id)
 {
-  uint num_columns= part_field_list.elements;
+  uint n_columns= part_field_list.elements;
   bool result= FALSE;
   uint i;
   part_column_list_val *col_val= val->col_val_array;
@@ -2004,7 +2030,7 @@ bool partition_info::fix_column_value_functions(THD *thd,
   {
     DBUG_RETURN(FALSE);
   }
-  for (i= 0; i < num_columns; col_val++, i++)
+  for (i= 0; i < n_columns; col_val++, i++)
   {
     Item *column_item= col_val->item_expression;
     Field *field= part_field_array[i];
diff --git a/sql/partition_info.h b/sql/partition_info.h
index 479714a3928..b196d0b59a2 100644
--- a/sql/partition_info.h
+++ b/sql/partition_info.h
@@ -151,8 +151,6 @@ public:
   char *part_func_string;
   char *subpart_func_string;
 
-  const char *part_state;
-
   partition_element *curr_part_elem;
   partition_element *current_partition;
   part_elem_value *curr_list_val;
@@ -173,7 +171,6 @@ public:
   partition_type subpart_type;
 
   uint part_info_len;
-  uint part_state_len;
   uint part_func_len;
   uint subpart_func_len;
 
@@ -226,13 +223,12 @@ public:
     list_array(NULL), err_value(0),
     part_info_string(NULL),
     part_func_string(NULL), subpart_func_string(NULL),
-    part_state(NULL),
     curr_part_elem(NULL), current_partition(NULL),
     curr_list_object(0), num_columns(0),
     default_engine_type(NULL),
     part_result_type(INT_RESULT),
     part_type(NOT_A_PARTITION), subpart_type(NOT_A_PARTITION),
-    part_info_len(0), part_state_len(0),
+    part_info_len(0),
     part_func_len(0), subpart_func_len(0),
     num_parts(0), num_subparts(0),
     count_curr_subparts(0), part_error_code(0),
diff --git a/sql/protocol.cc b/sql/protocol.cc
index eeb248012ab..ac78ac88ec6 100644
--- a/sql/protocol.cc
+++ b/sql/protocol.cc
@@ -747,8 +747,7 @@ bool Protocol::send_result_set_metadata(List<Item> *list, uint flags)
       else
       {
         /* With conversion */
-        ulonglong max_length;
-        uint32 field_length;
+        uint32 field_length, max_length;
         int2store(pos, thd_charset->number);
         /*
           For TEXT/BLOB columns, field_length describes the maximum data
@@ -771,9 +770,8 @@ bool Protocol::send_result_set_metadata(List<Item> *list, uint flags)
                      field.type <= MYSQL_TYPE_BLOB) ?
                      field.length / item->collation.collation->mbminlen :
                      field.length / item->collation.collation->mbmaxlen;
-        max_length*= thd_charset->mbmaxlen;
-        field_length= (max_length > UINT_MAX32) ? 
-          UINT_MAX32 : (uint32) max_length;
+        field_length= char_to_byte_length_safe(max_length,
+                                               thd_charset->mbmaxlen);
         int4store(pos + 2, field_length);
       }
       pos[6]= field.type;
diff --git a/sql/rpl_handler.cc b/sql/rpl_handler.cc
index 9cb5391075d..be0a61bcae2 100644
--- a/sql/rpl_handler.cc
+++ b/sql/rpl_handler.cc
@@ -89,21 +89,24 @@ int get_user_var_str(const char *name, char *value,
 
 int delegates_init()
 {
-  static unsigned long trans_mem[sizeof(Trans_delegate) / sizeof(unsigned long) + 1];
-  static unsigned long storage_mem[sizeof(Binlog_storage_delegate) / sizeof(unsigned long) + 1];
+  static Aligned_char_array<sizeof(Trans_delegate)>           trans_mem;
+  static Aligned_char_array<sizeof(Binlog_storage_delegate)>  storage_mem;
 #ifdef HAVE_REPLICATION
-  static unsigned long transmit_mem[sizeof(Binlog_transmit_delegate) / sizeof(unsigned long) + 1];
-  static unsigned long relay_io_mem[sizeof(Binlog_relay_IO_delegate)/ sizeof(unsigned long) + 1];
+  static Aligned_char_array<sizeof(Binlog_transmit_delegate)> transmit_mem;
+  static Aligned_char_array<sizeof(Binlog_relay_IO_delegate)> relay_io_mem;
 #endif
-  
-  if (!(transaction_delegate= new (trans_mem) Trans_delegate)
+
+  if (!(transaction_delegate= new (trans_mem.arr()) Trans_delegate)
       || (!transaction_delegate->is_inited())
-      || !(binlog_storage_delegate= new (storage_mem) Binlog_storage_delegate)
+      || !(binlog_storage_delegate=
+           new (storage_mem.arr()) Binlog_storage_delegate)
       || (!binlog_storage_delegate->is_inited())
 #ifdef HAVE_REPLICATION
-      || !(binlog_transmit_delegate= new (transmit_mem) Binlog_transmit_delegate)
+      || !(binlog_transmit_delegate=
+           new (transmit_mem.arr()) Binlog_transmit_delegate)
       || (!binlog_transmit_delegate->is_inited())
-      || !(binlog_relay_io_delegate= new (relay_io_mem) Binlog_relay_IO_delegate)
+      || !(binlog_relay_io_delegate=
+           new (relay_io_mem.arr()) Binlog_relay_IO_delegate)
       || (!binlog_relay_io_delegate->is_inited())
 #endif /* HAVE_REPLICATION */
       )
diff --git a/sql/rpl_mi.cc b/sql/rpl_mi.cc
index 10bf752e140..9b0450b3f02 100644
--- a/sql/rpl_mi.cc
+++ b/sql/rpl_mi.cc
@@ -131,8 +131,10 @@ enum {
   LINE_FOR_MASTER_SSL_VERIFY_SERVER_CERT= 15,
   /* 6.0 added value of master_heartbeat_period */
   LINE_FOR_MASTER_HEARTBEAT_PERIOD= 16,
+  /* MySQL Cluster 6.3 added master_bind */
+  LINE_FOR_MASTER_BIND = 17,
   /* 6.0 added value of master_ignore_server_id */
-  LINE_FOR_REPLICATE_IGNORE_SERVER_IDS= 17,
+  LINE_FOR_REPLICATE_IGNORE_SERVER_IDS= 18,
   /* Number of lines currently used when saving master info file */
   LINES_IN_MASTER_INFO= LINE_FOR_REPLICATE_IGNORE_SERVER_IDS
 };
@@ -240,6 +242,7 @@ file '%s')", fname);
     int ssl= 0, ssl_verify_server_cert= 0;
     float master_heartbeat_period= 0.0;
     char *first_non_digit;
+    char dummy_buf[HOSTNAME_LENGTH+1];
 
     /*
        Starting from 4.1.x master.info has new format. Now its
@@ -329,6 +332,13 @@ file '%s')", fname);
           init_floatvar_from_file(&master_heartbeat_period, &mi->file, 0.0))
         goto errwithmsg;
       /*
+	Starting from MySQL Cluster 6.3 master_bind might be in the file
+	(this is just a reservation to avoid future upgrade problems) 
+       */
+      if (lines >= LINE_FOR_MASTER_BIND &&
+	  init_strvar_from_file(dummy_buf, sizeof(dummy_buf), &mi->file, ""))
+	  goto errwithmsg;
+      /*
         Starting from 6.0 list of server_id of ignorable servers might be
         in the file
       */
@@ -480,14 +490,14 @@ int flush_master_info(Master_info* mi,
   my_sprintf(heartbeat_buf, (heartbeat_buf, "%.3f", mi->heartbeat_period));
   my_b_seek(file, 0L);
   my_b_printf(file,
-              "%u\n%s\n%s\n%s\n%s\n%s\n%d\n%d\n%d\n%s\n%s\n%s\n%s\n%s\n%d\n%s\n%s\n",
+              "%u\n%s\n%s\n%s\n%s\n%s\n%d\n%d\n%d\n%s\n%s\n%s\n%s\n%s\n%d\n%s\n%s\n%s\n",
               LINES_IN_MASTER_INFO,
               mi->master_log_name, llstr(mi->master_log_pos, lbuf),
               mi->host, mi->user,
               mi->password, mi->port, mi->connect_retry,
               (int)(mi->ssl), mi->ssl_ca, mi->ssl_capath, mi->ssl_cert,
               mi->ssl_cipher, mi->ssl_key, mi->ssl_verify_server_cert,
-              heartbeat_buf, ignore_server_ids_buf);
+              heartbeat_buf, "", ignore_server_ids_buf);
   my_free(ignore_server_ids_buf, MYF(0));
   err= flush_io_cache(file);
   if (sync_masterinfo_period && !err && 
diff --git a/sql/rpl_utility.h b/sql/rpl_utility.h
index cf28d2c8e29..25f2a60bece 100644
--- a/sql/rpl_utility.h
+++ b/sql/rpl_utility.h
@@ -233,7 +233,7 @@ struct RPL_TABLE_LIST
 
 
 /* Anonymous namespace for template functions/classes */
-namespace {
+CPP_UNNAMED_NS_START
 
   /*
     Smart pointer that will automatically call my_afree (a macro) when
@@ -260,7 +260,7 @@ namespace {
     Obj* get() { return m_ptr; }
   };
 
-}
+CPP_UNNAMED_NS_END
 #endif
 
 // NB. number of printed bit values is limited to sizeof(buf) - 1
diff --git a/sql/share/errmsg-utf8.txt b/sql/share/errmsg-utf8.txt
index e5ab2332e20..9e7fdbfeae5 100644
--- a/sql/share/errmsg-utf8.txt
+++ b/sql/share/errmsg-utf8.txt
@@ -6335,7 +6335,13 @@ ER_BINLOG_UNSAFE_MULTIPLE_ENGINES_AND_SELF_LOGGING_ENGINE
   eng "Mixing self-logging and non-self-logging engines in a statement is unsafe."
 
 ER_BINLOG_UNSAFE_MIXED_STATEMENT
-  eng "Statements that read from both transactional (or a temporary table of any engine type) and non-transactional tables and write to any of them are unsafe."
+  eng "Statement accesses nontransactional table as well as transactional or temporary table, and writes to any of them."
+
+ER_INSIDE_TRANSACTION_PREVENTS_SWITCH_SQL_LOG_BIN
+  eng "Cannot modify @@session.sql_log_bin inside a transaction"
+
+ER_STORED_FUNCTION_PREVENTS_SWITCH_SQL_LOG_BIN
+  eng "Cannot change the sql_log_bin inside a stored function or trigger"
 
 ER_FAILED_READ_FROM_PAR_FILE
   eng "Failed to read from the .par file"
diff --git a/sql/sp_head.cc b/sql/sp_head.cc
index e833a540032..9395146f18f 100644
--- a/sql/sp_head.cc
+++ b/sql/sp_head.cc
@@ -125,10 +125,10 @@ sp_get_item_value(THD *thd, Item *item, String *str)
   case STRING_RESULT:
     {
       String *result= item->val_str(str);
-      
+
       if (!result)
         return NULL;
-      
+
       {
         char buf_holder[STRING_BUFFER_USUAL_SIZE];
         String buf(buf_holder, sizeof(buf_holder), result->charset());
@@ -366,7 +366,7 @@ sp_eval_expr(THD *thd, Field *result_field, Item **expr_item_ptr)
 
     Save original values and restore them after save.
   */
-  
+
   thd->count_cuted_fields= CHECK_FIELD_ERROR_FOR_NULL;
   thd->abort_on_warning=
     thd->variables.sql_mode &
@@ -465,7 +465,7 @@ check_routine_name(LEX_STRING *ident)
 {
   if (!ident || !ident->str || !ident->str[0] ||
       ident->str[ident->length-1] == ' ')
-  { 
+  {
     my_error(ER_SP_WRONG_NAME, MYF(0), ident->str);
     return TRUE;
   }
@@ -502,7 +502,7 @@ sp_head::operator new(size_t size) throw()
   DBUG_RETURN(sp);
 }
 
-void 
+void
 sp_head::operator delete(void *ptr, size_t size) throw()
 {
   DBUG_ENTER("sp_head::operator delete");
@@ -718,7 +718,7 @@ create_typelib(MEM_ROOT *mem_root, Create_field *field_def, List<String> *src)
       String *tmp= it++;
 
       if (String::needs_conversion(tmp->length(), tmp->charset(),
-      				   cs, &dummy))
+                                   cs, &dummy))
       {
         uint cnv_errs;
         conv.copy(tmp->ptr(), tmp->length(), tmp->charset(), cs, &cnv_errs);
@@ -815,7 +815,7 @@ sp_head::create_result_field(uint field_max_length, const char *field_name,
 
   if (field)
     field->init(table);
-  
+
   DBUG_RETURN(field);
 }
 
@@ -844,7 +844,7 @@ int cmp_splocal_locations(Item_splocal * const *a, Item_splocal * const *b)
   Statements that have is_update_query(stmt) == FALSE (e.g. SELECTs) are not
   written into binary log. Instead we catch function calls the statement
   makes and write it into binary log separately (see #3).
-  
+
   2. PROCEDURE calls
 
   CALL statements are not written into binary log. Instead
@@ -857,8 +857,8 @@ int cmp_splocal_locations(Item_splocal * const *a, Item_splocal * const *b)
     This substitution is done in subst_spvars().
 
   3. FUNCTION calls
-  
-  In sp_head::execute_function(), we check 
+
+  In sp_head::execute_function(), we check
    * If this function invocation is done from a statement that is written
      into the binary log.
    * If there were any attempts to write events to the binary log during
@@ -866,28 +866,28 @@ int cmp_splocal_locations(Item_splocal * const *a, Item_splocal * const *b)
 
    If the answers are No and Yes, we write the function call into the binary
    log as "SELECT spfunc(<param1value>, <param2value>, ...)"
-  
-  
+
+
   4. Miscellaneous issues.
-  
-  4.1 User variables. 
+
+  4.1 User variables.
 
   When we call mysql_bin_log.write() for an SP statement, thd->user_var_events
-  must hold set<{var_name, value}> pairs for all user variables used during 
+  must hold set<{var_name, value}> pairs for all user variables used during
   the statement execution.
   This set is produced by tracking user variable reads during statement
-  execution. 
+  execution.
 
   For SPs, this has the following implications:
-  1) thd->user_var_events may contain events from several SP statements and 
-     needs to be valid after exection of these statements was finished. In 
+  1) thd->user_var_events may contain events from several SP statements and
+     needs to be valid after exection of these statements was finished. In
      order to achieve that, we
      * Allocate user_var_events array elements on appropriate mem_root (grep
        for user_var_events_alloc).
      * Use is_query_in_union() to determine if user_var_event is created.
-     
+
   2) We need to empty thd->user_var_events after we have wrote a function
-     call. This is currently done by making 
+     call. This is currently done by making
      reset_dynamic(&thd->user_var_events);
      calls in several different places. (TODO cosider moving this into
      mysql_bin_log.write() function)
@@ -906,7 +906,7 @@ 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.
 
-  The binlog-suitable string is produced by replacing references to SP local 
+  The binlog-suitable string is produced by replacing references to SP local
   variables with NAME_CONST('sp_var_name', value) calls.
 
   @param thd        Current thread.
@@ -943,11 +943,11 @@ subst_spvars(THD *thd, sp_instr *instr, LEX_STRING *query_str)
   }
   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);
 
-  /* 
+  /*
     Construct a statement string where SP local var refs are replaced
     with "NAME_CONST(name, value)"
   */
@@ -955,7 +955,7 @@ subst_spvars(THD *thd, sp_instr *instr, LEX_STRING *query_str)
   cur= query_str->str;
   prev_pos= res= 0;
   thd->query_name_consts= 0;
-  
+
   for (Item_splocal **splocal= sp_vars_uses.front(); 
        splocal < sp_vars_uses.back(); splocal++)
   {
@@ -965,13 +965,13 @@ subst_spvars(THD *thd, sp_instr *instr, LEX_STRING *query_str)
     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;
-    
+
     res|= (*splocal)->fix_fields(thd, (Item **) splocal);
-    if (res) 
+    if (res)
       break;
 
     if ((*splocal)->limit_clause_param)
@@ -998,7 +998,7 @@ subst_spvars(THD *thd, sp_instr *instr, LEX_STRING *query_str)
     res|= qbuf.append(')');
     if (res)
       break;
-      
+
     thd->query_name_consts++;
   }
   res|= qbuf.append(cur + prev_pos, query_str->length - prev_pos);
@@ -1024,16 +1024,14 @@ subst_spvars(THD *thd, sp_instr *instr, LEX_STRING *query_str)
 }
 
 
-/*
+/**
   Return appropriate error about recursion limit reaching
 
-  SYNOPSIS
-    sp_head::recursion_level_error()
-    thd		Thread handle
+  @param thd  Thread handle
 
-  NOTE
-    For functions and triggers we return error about prohibited recursion.
-    For stored procedures we return about reaching recursion limit.
+  @remark For functions and triggers we return error about
+          prohibited recursion. For stored procedures we
+          return about reaching recursion limit.
 */
 
 void sp_head::recursion_level_error(THD *thd)
@@ -1053,7 +1051,7 @@ void sp_head::recursion_level_error(THD *thd)
   Execute the routine. The main instruction jump loop is there.
   Assume the parameters already set.
   @todo
-    - Will write this SP statement into binlog separately 
+    - Will write this SP statement into binlog separately
     (TODO: consider changing the condition to "not inside event union")
 
   @retval
@@ -1217,10 +1215,10 @@ sp_head::execute(THD *thd)
   do
   {
     sp_instr *i;
-    uint hip;			// Handler ip
+    uint hip;
 
 #if defined(ENABLED_PROFILING)
-    /* 
+    /*
      Treat each "instr" of a routine as discrete unit that could be profiled.
      Profiling only records information for segments of code that set the
      source of the query, and almost all kinds of instructions in s-p do not.
@@ -1229,7 +1227,8 @@ sp_head::execute(THD *thd)
     thd->profiling.start_new_query("continuing inside routine");
 #endif
 
-    i = get_instr(ip);	// Returns NULL when we're done.
+    /* get_instr returns NULL when we're done. */
+    i = get_instr(ip);
     if (i == NULL)
     {
 #if defined(ENABLED_PROFILING)
@@ -1240,10 +1239,13 @@ sp_head::execute(THD *thd)
 
     DBUG_PRINT("execute", ("Instruction %u", ip));
 
-    /* Don't change NOW() in FUNCTION or TRIGGER */
+    /*
+      Make current_time() et al work. But don't change NOW() in FUNCTION
+      or TRIGGER.
+    */
     if (!thd->in_sub_stmt)
-      thd->set_time();		// Make current_time() et al work
-    
+      thd->set_time();
+
     /*
       We have to set thd->stmt_arena before executing the instruction
       to store in the instruction free_list all new items, created
@@ -1251,10 +1253,10 @@ sp_head::execute(THD *thd)
       items made during other permanent subquery transformations).
     */
     thd->stmt_arena= i;
-    
-    /* 
-      Will write this SP statement into binlog separately 
-      (TODO: consider changing the condition to "not inside event union")
+
+    /*
+      Will write this SP statement into binlog separately.
+      TODO: consider changing the condition to "not inside event union".
     */
     if (thd->locked_tables_mode <= LTM_LOCK_TABLES)
       thd->user_var_events_alloc= thd->mem_root;
@@ -1263,8 +1265,8 @@ sp_head::execute(THD *thd)
 
     if (i->free_list)
       cleanup_items(i->free_list);
-    
-    /* 
+
+    /*
       If we've set thd->user_var_events_alloc to mem_root of this SP
       statement, clean all the events allocated in it.
     */
@@ -1276,7 +1278,7 @@ sp_head::execute(THD *thd)
 
     /* we should cleanup free_list and memroot, used by instruction */
     thd->cleanup_after_query();
-    free_root(&execute_mem_root, MYF(0));    
+    free_root(&execute_mem_root, MYF(0));
 
     /*
       Check if an exception has occurred and a handler has been found
@@ -1291,7 +1293,7 @@ sp_head::execute(THD *thd)
 
       switch (ctx->found_handler(& hip, & handler_index)) {
       case SP_HANDLER_NONE:
-	break;
+        break;
       case SP_HANDLER_CONTINUE:
         thd->restore_active_arena(&execute_arena, &backup_arena);
         thd->set_n_backup_active_arena(&execute_arena, &backup_arena);
@@ -1300,15 +1302,15 @@ sp_head::execute(THD *thd)
       default:
         if (ctx->end_partial_result_set)
           thd->protocol->end_partial_result_set(thd);
-	ip= hip;
-	err_status= FALSE;
-	ctx->clear_handler();
-	ctx->enter_handler(hip, handler_index);
+        ip= hip;
+        err_status= FALSE;
+        ctx->clear_handler();
+        ctx->enter_handler(hip, handler_index);
         thd->clear_error();
         thd->is_fatal_error= 0;
-	thd->killed= THD::NOT_KILLED;
+        thd->killed= THD::NOT_KILLED;
         thd->mysys_var->abort= 0;
-	continue;
+        continue;
       }
 
       ctx->end_partial_result_set= FALSE;
@@ -1351,7 +1353,7 @@ sp_head::execute(THD *thd)
 
  done:
   DBUG_PRINT("info", ("err_status: %d  killed: %d  is_slave_error: %d  report_error: %d",
-		      err_status, thd->killed, thd->is_slave_error,
+                      err_status, thd->killed, thd->is_slave_error,
                       thd->is_error()));
 
   if (thd->killed)
@@ -1839,10 +1841,10 @@ err_with_cleanup:
 
 
 /**
-  Execute a procedure. 
+  Execute a procedure.
 
   The function does the following steps:
-   - Set all parameters 
+   - Set all parameters
    - changes security context for SUID routines
    - call sp_head::execute
    - copy back values of INOUT and OUT parameters
@@ -1880,14 +1882,14 @@ sp_head::execute_procedure(THD *thd, List<Item> *args)
 
   save_spcont= octx= thd->spcont;
   if (! octx)
-  {				// Create a temporary old context
-    if (!(octx= new sp_rcontext(m_pcont, NULL, octx)) ||
-        octx->init(thd))
+  {
+    /* Create a temporary old context. */
+    if (!(octx= new sp_rcontext(m_pcont, NULL, octx)) || octx->init(thd))
     {
       delete octx; /* Delete octx if it was init() that failed. */
       DBUG_RETURN(TRUE);
     }
-    
+
 #ifndef DBUG_OFF
     octx->sp= 0;
 #endif
@@ -2119,6 +2121,9 @@ sp_head::reset_lex(THD *thd)
   sublex->interval_list.empty();
   sublex->type= 0;
 
+  /* Reset part of parser state which needs this. */
+  thd->m_parser_state->m_yacc.reset_before_substatement();
+
   DBUG_RETURN(FALSE);
 }
 
@@ -2144,7 +2149,7 @@ sp_head::restore_lex(THD *thd)
 
   oldlex= (LEX *)m_lex.pop();
   if (! oldlex)
-    DBUG_RETURN(FALSE);			// Nothing to restore
+    DBUG_RETURN(FALSE); // Nothing to restore
 
   oldlex->trg_table_fields.push_back(&sublex->trg_table_fields);
 
@@ -2297,7 +2302,7 @@ sp_head::do_cont_backpatch()
 
 void
 sp_head::set_info(longlong created, longlong modified,
-		  st_sp_chistics *chistics, ulong sql_mode)
+                  st_sp_chistics *chistics, ulong sql_mode)
 {
   m_created= created;
   m_modified= modified;
@@ -2307,8 +2312,8 @@ sp_head::set_info(longlong created, longlong modified,
     m_chistics->comment.str= 0;
   else
     m_chistics->comment.str= strmake_root(mem_root,
-					  m_chistics->comment.str,
-					  m_chistics->comment.length);
+                                          m_chistics->comment.str,
+                                          m_chistics->comment.length);
   m_sql_mode= sql_mode;
 }
 
@@ -2349,7 +2354,7 @@ sp_head::reset_thd_mem_root(THD *thd)
   DBUG_PRINT("info", ("mem_root 0x%lx moved to thd mem root 0x%lx",
                       (ulong) &mem_root, (ulong) &thd->mem_root));
   free_list= thd->free_list; // Keep the old list
-  thd->free_list= NULL;	// Start a new one
+  thd->free_list= NULL; // Start a new one
   m_thd= thd;
   DBUG_VOID_RETURN;
 }
@@ -2358,13 +2363,13 @@ void
 sp_head::restore_thd_mem_root(THD *thd)
 {
   DBUG_ENTER("sp_head::restore_thd_mem_root");
-  Item *flist= free_list;	// The old list
+  Item *flist= free_list;       // The old list
   set_query_arena(thd);         // Get new free_list and mem_root
   state= INITIALIZED_FOR_SP;
 
   DBUG_PRINT("info", ("mem_root 0x%lx returned from thd mem root 0x%lx",
                       (ulong) &mem_root, (ulong) &thd->mem_root));
-  thd->free_list= flist;	// Restore the old one
+  thd->free_list= flist;        // Restore the old one
   thd->mem_root= m_thd_root;
   m_thd= NULL;
   DBUG_VOID_RETURN;
@@ -2374,10 +2379,10 @@ sp_head::restore_thd_mem_root(THD *thd)
 /**
   Check if a user has access right to a routine.
 
-  @param thd		    Thread handler
-  @param sp		    SP
-  @param full_access       Set to 1 if the user has SELECT right to the
-                           'mysql.proc' able or is the owner of the routine
+  @param thd          Thread handler
+  @param sp           SP
+  @param full_access  Set to 1 if the user has SELECT right to the
+                      'mysql.proc' able or is the owner of the routine
   @retval
     false ok
   @retval
@@ -2506,8 +2511,6 @@ sp_head::show_create_routine(THD *thd, int type)
 }
 
 
-
-
 /**
   Add instruction to SP.
 
@@ -2567,11 +2570,11 @@ void sp_head::optimize()
       if (src != dst)
       {
         /* Move the instruction and update prev. jumps */
-	sp_instr *ibp;
-	List_iterator_fast<sp_instr> li(bp);
+        sp_instr *ibp;
+        List_iterator_fast<sp_instr> li(bp);
 
-	set_dynamic(&m_instr, (uchar*)&i, dst);
-	while ((ibp= li++))
+        set_dynamic(&m_instr, (uchar*)&i, dst);
+        while ((ibp= li++))
         {
           sp_instr_opt_meta *im= static_cast<sp_instr_opt_meta *>(ibp);
           im->set_destination(src, dst);
@@ -2666,7 +2669,7 @@ sp_head::show_routine_code(THD *thd)
 
   for (ip= 0; (i = get_instr(ip)) ; ip++)
   {
-    /* 
+    /*
       Consistency check. If these are different something went wrong
       during optimization.
     */
@@ -2729,7 +2732,7 @@ 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.
@@ -2891,9 +2894,8 @@ sp_instr_stmt::execute(THD *thd, uint *nextp)
     if (unlikely((thd->variables.option_bits & OPTION_LOG_OFF)==0))
       general_log_write(thd, COM_QUERY, thd->query(), thd->query_length());
 
-    if (query_cache_send_result_to_client(thd,
-          thd->query(), 
-          thd->query_length()) <= 0)
+    if (query_cache_send_result_to_client(thd, thd->query(),
+                                          thd->query_length()) <= 0)
     {
       res= m_lex_keeper.reset_lex_and_exec_core(thd, nextp, FALSE, this);
 
@@ -2996,7 +2998,7 @@ sp_instr_set::exec_core(THD *thd, uint *nextp)
       /* If this also failed, let's abort. */
 
       sp_rcontext *spcont= thd->spcont;
-    
+
       thd->spcont= NULL;           /* Avoid handlers */
       my_error(ER_OUT_OF_RESOURCES, MYF(0));
       spcont->clear_handler();
@@ -3100,7 +3102,7 @@ uint
 sp_instr_jump::opt_mark(sp_head *sp, List<sp_instr> *leads)
 {
   m_dest= opt_shortcut_jump(sp, this);
-  if (m_dest != m_ip+1)		/* Jumping to following instruction? */
+  if (m_dest != m_ip+1)   /* Jumping to following instruction? */
     marked= 1;
   m_optdest= sp->get_instr(m_dest);
   return m_dest;
@@ -3130,9 +3132,9 @@ void
 sp_instr_jump::opt_move(uint dst, List<sp_instr> *bp)
 {
   if (m_dest > m_ip)
-    bp->push_back(this);	// Forward
+    bp->push_back(this);      // Forward
   else if (m_optdest)
-    m_dest= m_optdest->m_ip;	// Backward
+    m_dest= m_optdest->m_ip;  // Backward
   m_ip= dst;
 }
 
@@ -3405,7 +3407,7 @@ uint
 sp_instr_hreturn::opt_mark(sp_head *sp, List<sp_instr> *leads)
 {
   marked= 1;
-  
+
   if (m_dest)
   {
     /*
@@ -3413,7 +3415,7 @@ sp_instr_hreturn::opt_mark(sp_head *sp, List<sp_instr> *leads)
      */
     return m_dest;
   }
-  
+
   /*
     This is a CONTINUE handler; next instruction step will come from
     the handler stack and not from opt_mark.
@@ -3730,14 +3732,14 @@ sp_instr_set_case_expr::exec_core(THD *thd, uint *nextp)
     */
 
     Item *null_item= new Item_null();
-    
+
     if (!null_item ||
         thd->spcont->set_case_expr(thd, m_case_expr_id, &null_item))
     {
       /* If this also failed, we have to abort. */
 
       sp_rcontext *spcont= thd->spcont;
-    
+
       thd->spcont= NULL;           /* Avoid handlers */
       my_error(ER_OUT_OF_RESOURCES, MYF(0));
       spcont->clear_handler();
@@ -3903,13 +3905,13 @@ sp_head::merge_table_list(THD *thd, TABLE_LIST *table, LEX *lex_for_tmp_check)
       }
       else
       {
-	if (!(tab= (SP_TABLE *)thd->calloc(sizeof(SP_TABLE))))
-	  return FALSE;
-	if (lex_for_tmp_check->sql_command == SQLCOM_CREATE_TABLE &&
-	    lex_for_tmp_check->query_tables == table &&
-	    lex_for_tmp_check->create_info.options & HA_LEX_CREATE_TMP_TABLE)
+        if (!(tab= (SP_TABLE *)thd->calloc(sizeof(SP_TABLE))))
+          return FALSE;
+        if (lex_for_tmp_check->sql_command == SQLCOM_CREATE_TABLE &&
+            lex_for_tmp_check->query_tables == table &&
+            lex_for_tmp_check->create_info.options & HA_LEX_CREATE_TMP_TABLE)
         {
-	  tab->temp= TRUE;
+          tab->temp= TRUE;
           tab->qname.length= tlen - alen - 1;
         }
         else
@@ -3922,7 +3924,7 @@ 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;
-	if (my_hash_insert(&m_sptabs, (uchar *)tab))
+        if (my_hash_insert(&m_sptabs, (uchar *)tab))
           return FALSE;
       }
     }
@@ -4029,8 +4031,8 @@ sp_head::add_used_tables_to_table_list(THD *thd,
 
 TABLE_LIST *
 sp_add_to_query_tables(THD *thd, LEX *lex,
-		       const char *db, const char *name,
-		       thr_lock_type locktype)
+                       const char *db, const char *name,
+                       thr_lock_type locktype)
 {
   TABLE_LIST *table;
 
diff --git a/sql/spatial.cc b/sql/spatial.cc
index 34b1f3d1f0c..fcf06119db9 100644
--- a/sql/spatial.cc
+++ b/sql/spatial.cc
@@ -128,6 +128,16 @@ Geometry::Class_info *Geometry::find_class(const char *name, uint32 len)
 }
 
 
+Geometry *Geometry::create_by_typeid(Geometry_buffer *buffer, int type_id)
+{
+  Class_info *ci;
+  if (!(ci= find_class((int) type_id)))
+    return NULL;
+  (*ci->m_create_func)(buffer->buf.arr());
+  return my_reinterpret_cast(Geometry *)(buffer->buf.arr());
+}
+
+
 Geometry *Geometry::construct(Geometry_buffer *buffer,
                               const char *data, uint32 data_len)
 {
diff --git a/sql/spatial.h b/sql/spatial.h
index a4bce47d0e5..aabbb7a1b97 100644
--- a/sql/spatial.h
+++ b/sql/spatial.h
@@ -267,14 +267,7 @@ public:
   virtual int geometry_n(uint32 num, String *result) const { return -1; }
 
 public:
-  static Geometry *create_by_typeid(Geometry_buffer *buffer, int type_id)
-  {
-    Class_info *ci;
-    if (!(ci= find_class((int) type_id)))
-      return NULL;
-    (*ci->m_create_func)((void *)buffer);
-    return my_reinterpret_cast(Geometry *)(buffer);
-  }
+  static Geometry *create_by_typeid(Geometry_buffer *buffer, int type_id);
 
   static Geometry *construct(Geometry_buffer *buffer,
                              const char *data, uint32 data_len);
@@ -532,10 +525,9 @@ public:
   const Class_info *get_class_info() const;
 };
 
-const int geometry_buffer_size= sizeof(Gis_point);
 struct Geometry_buffer
 {
-  void *arr[(geometry_buffer_size - 1)/sizeof(void *) + 1];
+  Aligned_char_array<sizeof(Gis_point)> buf;
 };
 
 #endif /*HAVE_SPATAIAL*/
diff --git a/sql/sql_acl.cc b/sql/sql_acl.cc
index 5a7fb7f154a..ec25e4cb68b 100644
--- a/sql/sql_acl.cc
+++ b/sql/sql_acl.cc
@@ -3159,6 +3159,12 @@ int mysql_table_grant(THD *thd, TABLE_LIST *table_list,
   */
   Query_tables_list backup;
   thd->lex->reset_n_backup_query_tables_list(&backup);
+  /*
+    Restore Query_tables_list::sql_command value, which was reset
+    above, as the code writing query to the binary log assumes that
+    this value corresponds to the statement being executed.
+  */
+  thd->lex->sql_command= backup.sql_command;
   if (open_and_lock_tables(thd, tables, FALSE, MYSQL_LOCK_IGNORE_TIMEOUT))
   {						// Should never happen
     close_thread_tables(thd);			/* purecov: deadcode */
diff --git a/sql/sql_base.cc b/sql/sql_base.cc
index fd0dfc5189a..325f054db02 100644
--- a/sql/sql_base.cc
+++ b/sql/sql_base.cc
@@ -1564,7 +1564,7 @@ void close_thread_tables(THD *thd)
     - If in autocommit mode, or outside a transactional context,
     automatically release metadata locks of the current statement.
   */
-  if (! thd->in_multi_stmt_transaction() &&
+  if (! thd->in_multi_stmt_transaction_mode() &&
       ! (thd->state_flags & Open_tables_state::BACKUPS_AVAIL))
   {
     thd->mdl_context.release_transactional_locks();
@@ -3789,7 +3789,7 @@ end_with_lock_open:
 Open_table_context::Open_table_context(THD *thd, ulong timeout)
   :m_action(OT_NO_ACTION),
    m_start_of_statement_svp(thd->mdl_context.mdl_savepoint()),
-   m_has_locks((thd->in_multi_stmt_transaction() &&
+   m_has_locks((thd->in_multi_stmt_transaction_mode() &&
                 thd->mdl_context.has_locks()) ||
                 thd->mdl_context.trans_sentinel()),
    m_global_mdl_request(NULL),
@@ -3969,7 +3969,8 @@ recover_from_failed_open(THD *thd, MDL_request *mdl_request,
   Return a appropriate read lock type given a table object.
 
   @param thd Thread context
-  @param table TABLE object for table to be locked
+  @param prelocking_ctx Prelocking context.
+  @param table_list     Table list element for table to be locked.
 
   @remark Due to a statement-based replication limitation, statements such as
           INSERT INTO .. SELECT FROM .. and CREATE TABLE .. SELECT FROM need
@@ -3978,20 +3979,44 @@ recover_from_failed_open(THD *thd, MDL_request *mdl_request,
           source table. If such a statement gets applied on the slave before
           the INSERT .. SELECT statement finishes, data on the master could
           differ from data on the slave and end-up with a discrepancy between
-          the binary log and table state. Furthermore, this does not apply to
-          I_S and log tables as it's always unsafe to replicate such tables
-          under statement-based replication as the table on the slave might
-          contain other data (ie: general_log is enabled on the slave). The
-          statement will be marked as unsafe for SBR in decide_logging_format().
+          the binary log and table state.
+          This also applies to SELECT/SET/DO statements which use stored
+          functions. Calls to such functions are going to be logged as a
+          whole and thus should be serialized against concurrent changes
+          to tables used by those functions. This can be avoided if functions
+          only read data but doing so requires more complex analysis than it
+          is done now.
+          Furthermore, this does not apply to I_S and log tables as it's
+          always unsafe to replicate such tables under statement-based
+          replication as the table on the slave might contain other data
+          (ie: general_log is enabled on the slave). The statement will
+          be marked as unsafe for SBR in decide_logging_format().
+  @remark Note that even in prelocked mode it is important to correctly
+          determine lock type value. In this mode lock type is passed to
+          handler::start_stmt() method and can be used by storage engine,
+          for example, to determine what kind of row locks it should acquire
+          when reading data from the table.
 */
 
-thr_lock_type read_lock_type_for_table(THD *thd, TABLE *table)
+thr_lock_type read_lock_type_for_table(THD *thd,
+                                       Query_tables_list *prelocking_ctx,
+                                       TABLE_LIST *table_list)
 {
-  bool log_on= mysql_bin_log.is_open() && (thd->variables.option_bits & OPTION_BIN_LOG);
+  /*
+    In cases when this function is called for a sub-statement executed in
+    prelocked mode we can't rely on OPTION_BIN_LOG flag in THD::options
+    bitmap to determine that binary logging is turned on as this bit can
+    be cleared before executing sub-statement. So instead we have to look
+    at THD::variables::sql_log_bin member.
+  */
+  bool log_on= mysql_bin_log.is_open() && thd->variables.sql_log_bin;
   ulong binlog_format= thd->variables.binlog_format;
   if ((log_on == FALSE) || (binlog_format == BINLOG_FORMAT_ROW) ||
-      (table->s->table_category == TABLE_CATEGORY_LOG) ||
-      (table->s->table_category == TABLE_CATEGORY_PERFORMANCE))
+      (table_list->table->s->table_category == TABLE_CATEGORY_LOG) ||
+      (table_list->table->s->table_category == TABLE_CATEGORY_PERFORMANCE) ||
+      !(is_update_query(prelocking_ctx->sql_command) ||
+        table_list->prelocking_placeholder ||
+        (thd->locked_tables_mode > LTM_LOCK_TABLES)))
     return TL_READ;
   else
     return TL_READ_NO_INSERT;
@@ -4342,7 +4367,7 @@ open_and_process_table(THD *thd, LEX *lex, TABLE_LIST *tables,
       tables->table->reginfo.lock_type= thd->update_lock_default;
     else if (tables->lock_type == TL_READ_DEFAULT)
       tables->table->reginfo.lock_type=
-        read_lock_type_for_table(thd, tables->table);
+        read_lock_type_for_table(thd, lex, tables);
     else
       tables->table->reginfo.lock_type= tables->lock_type;
   }
@@ -4995,35 +5020,49 @@ handle_view(THD *thd, Query_tables_list *prelocking_ctx,
 }
 
 
-/*
+/**
   Check that lock is ok for tables; Call start stmt if ok
 
-  SYNOPSIS
-    check_lock_and_start_stmt()
-    thd			Thread handle
-    table_list		Table to check
-    lock_type		Lock used for table
+  @param thd             Thread handle.
+  @param prelocking_ctx  Prelocking context.
+  @param table_list      Table list element for table to be checked.
 
-  RETURN VALUES
-  0	ok
-  1	error
+  @retval FALSE - Ok.
+  @retval TRUE  - Error.
 */
 
-static bool check_lock_and_start_stmt(THD *thd, TABLE *table,
-				      thr_lock_type lock_type)
+static bool check_lock_and_start_stmt(THD *thd,
+                                      Query_tables_list *prelocking_ctx,
+                                      TABLE_LIST *table_list)
 {
   int error;
+  thr_lock_type lock_type;
   DBUG_ENTER("check_lock_and_start_stmt");
 
+  /*
+    TL_WRITE_DEFAULT and TL_READ_DEFAULT are supposed to be parser only
+    types of locks so they should be converted to appropriate other types
+    to be passed to storage engine. The exact lock type passed to the
+    engine is important as, for example, InnoDB uses it to determine
+    what kind of row locks should be acquired when executing statement
+    in prelocked mode or under LOCK TABLES with @@innodb_table_locks = 0.
+  */
+  if (table_list->lock_type == TL_WRITE_DEFAULT)
+    lock_type= thd->update_lock_default;
+  else if (table_list->lock_type == TL_READ_DEFAULT)
+    lock_type= read_lock_type_for_table(thd, prelocking_ctx, table_list);
+  else
+    lock_type= table_list->lock_type;
+
   if ((int) lock_type >= (int) TL_WRITE_ALLOW_READ &&
-      (int) table->reginfo.lock_type < (int) TL_WRITE_ALLOW_READ)
+      (int) table_list->table->reginfo.lock_type < (int) TL_WRITE_ALLOW_READ)
   {
-    my_error(ER_TABLE_NOT_LOCKED_FOR_WRITE, MYF(0),table->alias);
+    my_error(ER_TABLE_NOT_LOCKED_FOR_WRITE, MYF(0), table_list->alias);
     DBUG_RETURN(1);
   }
-  if ((error=table->file->start_stmt(thd, lock_type)))
+  if ((error= table_list->table->file->start_stmt(thd, lock_type)))
   {
-    table->file->print_error(error,MYF(0));
+    table_list->table->file->print_error(error, MYF(0));
     DBUG_RETURN(1);
   }
   DBUG_RETURN(0);
@@ -5168,7 +5207,7 @@ TABLE *open_ltable(THD *thd, TABLE_LIST *table_list, thr_lock_type lock_type,
     table->grant= table_list->grant;
     if (thd->locked_tables_mode)
     {
-      if (check_lock_and_start_stmt(thd, table, lock_type))
+      if (check_lock_and_start_stmt(thd, thd->lex, table_list))
 	table= 0;
     }
     else
@@ -5396,7 +5435,7 @@ bool lock_tables(THD *thd, TABLE_LIST *tables, uint count,
         if (!table->placeholder())
         {
           table->table->query_id= thd->query_id;
-          if (check_lock_and_start_stmt(thd, table->table, table->lock_type))
+          if (check_lock_and_start_stmt(thd, thd->lex, table))
           {
             mysql_unlock_tables(thd, thd->lock);
             thd->lock= 0;
@@ -5450,7 +5489,7 @@ bool lock_tables(THD *thd, TABLE_LIST *tables, uint count,
         }
       }
 
-      if (check_lock_and_start_stmt(thd, table->table, table->lock_type))
+      if (check_lock_and_start_stmt(thd, thd->lex, table))
       {
 	DBUG_RETURN(TRUE);
       }
diff --git a/sql/sql_base.h b/sql/sql_base.h
index 4d6be93b532..0fe70e4bc9d 100644
--- a/sql/sql_base.h
+++ b/sql/sql_base.h
@@ -56,9 +56,6 @@ enum enum_resolution_type {
   RESOLVED_AGAINST_ALIAS
 };
 
-enum enum_open_table_action {OT_NO_ACTION= 0, OT_BACK_OFF_AND_RETRY,
-                             OT_DISCOVER, OT_REPAIR};
-
 enum find_item_error_report_type {REPORT_ALL_ERRORS, REPORT_EXCEPT_NOT_FOUND,
 				  IGNORE_ERRORS, REPORT_EXCEPT_NON_UNIQUE,
                                   IGNORE_EXCEPT_NON_UNIQUE};
@@ -123,7 +120,9 @@ TABLE *open_temporary_table(THD *thd, const char *path, const char *db,
 TABLE *find_locked_table(TABLE *list, const char *db, const char *table_name);
 TABLE *find_write_locked_table(TABLE *list, const char *db,
                                const char *table_name);
-thr_lock_type read_lock_type_for_table(THD *thd, TABLE *table);
+thr_lock_type read_lock_type_for_table(THD *thd,
+                                       Query_tables_list *prelocking_ctx,
+                                       TABLE_LIST *table_list);
 
 my_bool mysql_rm_tmp_tables(void);
 bool rm_temporary_table(handlerton *base, char *path);
diff --git a/sql/sql_cache.cc b/sql/sql_cache.cc
index 1e4161dfa1c..92d54c8e71b 100644
--- a/sql/sql_cache.cc
+++ b/sql/sql_cache.cc
@@ -1177,7 +1177,7 @@ void Query_cache::store_query(THD *thd, TABLE_LIST *tables_used)
     DBUG_ASSERT(flags.protocol_type != (unsigned int) Protocol::PROTOCOL_LOCAL);
     flags.more_results_exists= test(thd->server_status &
                                     SERVER_MORE_RESULTS_EXISTS);
-    flags.in_trans= test(thd->server_status & SERVER_STATUS_IN_TRANS);
+    flags.in_trans= thd->in_active_multi_stmt_transaction();
     flags.autocommit= test(thd->server_status & SERVER_STATUS_AUTOCOMMIT);
     flags.pkt_nr= net->pkt_nr;
     flags.character_set_client_num=
@@ -1470,7 +1470,7 @@ Query_cache::send_result_to_client(THD *thd, char *sql, uint query_length)
   flags.protocol_type= (unsigned int) thd->protocol->type();
   flags.more_results_exists= test(thd->server_status &
                                   SERVER_MORE_RESULTS_EXISTS);
-  flags.in_trans= test(thd->server_status & SERVER_STATUS_IN_TRANS);
+  flags.in_trans= thd->in_active_multi_stmt_transaction();
   flags.autocommit= test(thd->server_status & SERVER_STATUS_AUTOCOMMIT);
   flags.pkt_nr= thd->net.pkt_nr;
   flags.character_set_client_num= thd->variables.character_set_client->number;
@@ -1541,7 +1541,7 @@ def_week_frmt: %lu, in_trans: %d, autocommit: %d",
   }
   DBUG_PRINT("qcache", ("Query have result 0x%lx", (ulong) query));
 
-  if (thd->in_multi_stmt_transaction() &&
+  if (thd->in_multi_stmt_transaction_mode() &&
       (query->tables_type() & HA_CACHE_TBL_TRANSACT))
   {
     DBUG_PRINT("qcache",
@@ -1698,7 +1698,7 @@ void Query_cache::invalidate(THD *thd, TABLE_LIST *tables_used,
   if (is_disabled())
     DBUG_VOID_RETURN;
 
-  using_transactions= using_transactions && thd->in_multi_stmt_transaction();
+  using_transactions= using_transactions && thd->in_multi_stmt_transaction_mode();
   for (; tables_used; tables_used= tables_used->next_local)
   {
     DBUG_ASSERT(!using_transactions || tables_used->table!=0);
@@ -1782,7 +1782,7 @@ void Query_cache::invalidate(THD *thd, TABLE *table,
   if (is_disabled())
     DBUG_VOID_RETURN;
 
-  using_transactions= using_transactions && thd->in_multi_stmt_transaction();
+  using_transactions= using_transactions && thd->in_multi_stmt_transaction_mode();
   if (using_transactions && 
       (table->file->table_cache_type() == HA_CACHE_TBL_TRANSACT))
     thd->add_changed_table(table);
@@ -1800,7 +1800,7 @@ void Query_cache::invalidate(THD *thd, const char *key, uint32  key_length,
   if (is_disabled())
    DBUG_VOID_RETURN;
 
-  using_transactions= using_transactions && thd->in_multi_stmt_transaction();
+  using_transactions= using_transactions && thd->in_multi_stmt_transaction_mode();
   if (using_transactions) // used for innodb => has_transactions() is TRUE
     thd->add_changed_table(key, key_length);
   else
@@ -3572,7 +3572,7 @@ Query_cache::is_cacheable(THD *thd, size_t query_len, const char *query,
                                                 tables_type)))
       DBUG_RETURN(0);
 
-    if (thd->in_multi_stmt_transaction() &&
+    if (thd->in_multi_stmt_transaction_mode() &&
 	((*tables_type)&HA_CACHE_TBL_TRANSACT))
     {
       DBUG_PRINT("qcache", ("not in autocommin mode"));
diff --git a/sql/sql_class.cc b/sql/sql_class.cc
index 049975bccd9..ac092756a74 100644
--- a/sql/sql_class.cc
+++ b/sql/sql_class.cc
@@ -261,11 +261,13 @@ int thd_tablespace_op(const THD *thd)
 
 
 extern "C"
-const char *set_thd_proc_info(THD *thd, const char *info,
+const char *set_thd_proc_info(void *thd_arg, const char *info,
                               const char *calling_function,
                               const char *calling_file,
                               const unsigned int calling_line)
 {
+  THD *thd= (THD *) thd_arg;
+
   if (!thd)
     thd= current_thd;
 
@@ -491,7 +493,6 @@ THD::THD()
    rli_fake(0),
    lock_id(&main_lock_id),
    user_time(0), in_sub_stmt(0),
-   sql_log_bin_toplevel(false),
    binlog_unsafe_warning_flags(0), binlog_table_maps(0),
    table_map_for_update(0),
    arg_of_last_insert_id_function(FALSE),
@@ -960,7 +961,11 @@ void THD::init(void)
   update_charset();
   reset_current_stmt_binlog_format_row();
   bzero((char *) &status_var, sizeof(status_var));
-  sql_log_bin_toplevel= variables.option_bits & OPTION_BIN_LOG;
+
+  if (variables.sql_log_bin)
+    variables.option_bits|= OPTION_BIN_LOG;
+  else
+    variables.option_bits&= ~OPTION_BIN_LOG;
 
 #if defined(ENABLED_DEBUG_SYNC)
   /* Initialize the Debug Sync Facility. See debug_sync.cc. */
@@ -1474,7 +1479,7 @@ void THD::add_changed_table(TABLE *table)
 {
   DBUG_ENTER("THD::add_changed_table(table)");
 
-  DBUG_ASSERT(in_multi_stmt_transaction() && table->file->has_transactions());
+  DBUG_ASSERT(in_multi_stmt_transaction_mode() && table->file->has_transactions());
   add_changed_table(table->s->table_cache_key.str,
                     (long) table->s->table_cache_key.length);
   DBUG_VOID_RETURN;
@@ -3189,6 +3194,11 @@ extern "C" bool thd_binlog_filter_ok(const MYSQL_THD thd)
 {
   return binlog_filter->db_ok(thd->db);
 }
+
+extern "C" bool thd_sqlcom_can_generate_row_events(const MYSQL_THD thd)
+{
+  return sqlcom_can_generate_row_events(thd);
+}
 #endif // INNODB_COMPATIBILITY_HOOKS */
 
 /****************************************************************************
@@ -3914,7 +3924,8 @@ int THD::decide_logging_format(TABLE_LIST *tables)
         */
         my_error((error= ER_BINLOG_ROW_INJECTION_AND_STMT_ENGINE), MYF(0));
       }
-      else if (variables.binlog_format == BINLOG_FORMAT_ROW)
+      else if (variables.binlog_format == BINLOG_FORMAT_ROW &&
+               sqlcom_can_generate_row_events(this))
       {
         /*
           2. Error: Cannot modify table that uses a storage engine
@@ -3952,7 +3963,8 @@ int THD::decide_logging_format(TABLE_LIST *tables)
           */
           my_error((error= ER_BINLOG_ROW_INJECTION_AND_STMT_MODE), MYF(0));
         }
-        else if ((flags_write_all_set & HA_BINLOG_STMT_CAPABLE) == 0)
+        else if ((flags_write_all_set & HA_BINLOG_STMT_CAPABLE) == 0 &&
+                 sqlcom_can_generate_row_events(this))
         {
           /*
             5. Error: Cannot modify table that uses a storage engine
@@ -4197,7 +4209,9 @@ field_type_name(enum_field_types type)
 #endif
 
 
-namespace {
+/* Declare in unnamed namespace. */
+CPP_UNNAMED_NS_START
+
   /**
      Class to handle temporary allocation of memory for row data.
 
@@ -4316,8 +4330,8 @@ namespace {
     uchar *m_memory;
     uchar *m_ptr[2];
   };
-}
 
+CPP_UNNAMED_NS_END
 
 int THD::binlog_write_row(TABLE* table, bool is_trans, 
                           MY_BITMAP const* cols, size_t colcnt, 
@@ -4608,8 +4622,13 @@ int THD::binlog_query(THD::enum_binlog_query_type qtype, char const *query_arg,
     because the warnings should be printed only if the statement is
     actually logged. When executing decide_logging_format(), we cannot
     know for sure if the statement will be logged.
+
+    Besides, we should not try to print these warnings if it is not
+    possible to write statements to the binary log as it happens when
+    the execution is inside a function, or generaly speaking, when
+    the variables.option_bits & OPTION_BIN_LOG is false.
   */
-  if (sql_log_bin_toplevel)
+  if (variables.option_bits & OPTION_BIN_LOG)
     issue_unsafe_warnings();
 
   switch (qtype) {
diff --git a/sql/sql_class.h b/sql/sql_class.h
index 93433ff85c8..25b136bc4ca 100644
--- a/sql/sql_class.h
+++ b/sql/sql_class.h
@@ -428,6 +428,7 @@ typedef struct system_variables
 
   uint binlog_format; ///< binlog format for this thd (see enum_binlog_format)
   my_bool binlog_direct_non_trans_update;
+  my_bool sql_log_bin;
   uint completion_type;
   uint query_cache_type;
   uint tx_isolation;
@@ -1672,8 +1673,6 @@ public:
 
   /* <> 0 if we are inside of trigger or stored function. */
   uint in_sub_stmt;
-  /* TRUE when the current top has SQL_LOG_BIN ON */
-  bool sql_log_bin_toplevel;
 
   /* container for handler's private per-connection data */
   Ha_data ha_data[MAX_HA];
@@ -2119,8 +2118,6 @@ public:
   char	     scramble[SCRAMBLE_LENGTH+1];
 
   bool       slave_thread, one_shot_set;
-  bool	     locked, some_tables_deleted;
-  bool       last_cuted_field;
   bool	     no_errors, password;
   /**
     Set to TRUE if execution of the current compound statement
@@ -2369,10 +2366,6 @@ public:
   {
     return limit_found_rows;
   }
-  inline bool active_transaction()
-  {
-    return server_status & SERVER_STATUS_IN_TRANS;
-  }
   /**
     Returns TRUE if session is in a multi-statement transaction mode.
 
@@ -2383,11 +2376,60 @@ public:
     OPTION_BEGIN: Regardless of the autocommit status, a multi-statement
     transaction can be explicitly started with the statements "START
     TRANSACTION", "BEGIN [WORK]", "[COMMIT | ROLLBACK] AND CHAIN", etc.
+
+    Note: this doesn't tell you whether a transaction is active.
+    A session can be in multi-statement transaction mode, and yet
+    have no active transaction, e.g., in case of:
+    set @@autocommit=0;
+    set @a= 3;                                     <-- these statements don't
+    set transaction isolation level serializable;  <-- start an active
+    flush tables;                                  <-- transaction
+
+    I.e. for the above scenario this function returns TRUE, even
+    though no active transaction has begun.
+    @sa in_active_multi_stmt_transaction()
   */
-  inline bool in_multi_stmt_transaction()
+  inline bool in_multi_stmt_transaction_mode()
   {
     return variables.option_bits & (OPTION_NOT_AUTOCOMMIT | OPTION_BEGIN);
   }
+  /**
+    TRUE if the session is in a multi-statement transaction mode
+    (@sa in_multi_stmt_transaction_mode()) *and* there is an
+    active transaction, i.e. there is an explicit start of a
+    transaction with BEGIN statement, or implicit with a
+    statement that uses a transactional engine.
+
+    For example, these scenarios don't start an active transaction
+    (even though the server is in multi-statement transaction mode):
+
+    set @@autocommit=0;
+    select * from nontrans_table;
+    set @var=TRUE;
+    flush tables;
+
+    Note, that even for a statement that starts a multi-statement
+    transaction (i.e. select * from trans_table), this
+    flag won't be set until we open the statement's tables
+    and the engines register themselves for the transaction
+    (see trans_register_ha()),
+    hence this method is reliable to use only after
+    open_tables() has completed.
+
+    Why do we need a flag?
+    ----------------------
+    We need to maintain a (at first glance redundant)
+    session flag, rather than looking at thd->transaction.all.ha_list
+    because of explicit start of a transaction with BEGIN. 
+
+    I.e. in case of
+    BEGIN;
+    select * from nontrans_t1; <-- in_active_multi_stmt_transaction() is true
+  */
+  inline bool in_active_multi_stmt_transaction()
+  {
+    return server_status & SERVER_STATUS_IN_TRANS;
+  }
   inline bool fill_derived_tables()
   {
     return !stmt_arena->is_stmt_prepare() && !lex->only_view_structure();
@@ -3582,6 +3624,12 @@ public:
  */
 #define CF_PROTECT_AGAINST_GRL  (1U << 10)
 
+/**
+  Identifies statements that may generate row events
+  and that may end up in the binary log.
+*/
+#define CF_CAN_GENERATE_ROW_EVENTS (1U << 11)
+
 /* Bits in server_command_flags */
 
 /**
@@ -3642,7 +3690,7 @@ inline bool add_group_to_list(THD *thd, Item *item, bool asc)
   three calling-info parameters.
 */
 extern "C"
-const char *set_thd_proc_info(THD *thd, const char *info,
+const char *set_thd_proc_info(void *thd_arg, const char *info,
                               const char *calling_func,
                               const char *calling_file,
                               const unsigned int calling_line);
diff --git a/sql/sql_const.h b/sql/sql_const.h
index 72f34ed6be8..dca66628ddb 100644
--- a/sql/sql_const.h
+++ b/sql/sql_const.h
@@ -136,7 +136,6 @@
 #ifndef MYSQLD_NET_RETRY_COUNT
 #define MYSQLD_NET_RETRY_COUNT  10	///< Abort read after this many int.
 #endif
-#define TEMP_POOL_SIZE          128
 
 #define QUERY_ALLOC_BLOCK_SIZE		8192
 #define QUERY_ALLOC_PREALLOC_SIZE   	8192
@@ -146,11 +145,8 @@
 #define ACL_ALLOC_BLOCK_SIZE		1024
 #define UDF_ALLOC_BLOCK_SIZE		1024
 #define TABLE_ALLOC_BLOCK_SIZE		1024
-#define BDB_LOG_ALLOC_BLOCK_SIZE	1024
 #define WARN_ALLOC_BLOCK_SIZE		2048
 #define WARN_ALLOC_PREALLOC_SIZE	1024
-#define PROFILE_ALLOC_BLOCK_SIZE  2048
-#define PROFILE_ALLOC_PREALLOC_SIZE 1024
 
 /*
   The following parameters is to decide when to use an extra cache to
@@ -194,8 +190,6 @@
 */
 #define MATCHING_ROWS_IN_OTHER_TABLE 10
 
-#define RAID_BLOCK_SIZE 1024
-
 #define MY_CHARSET_BIN_MB_MAXLEN 1
 
 /** Don't pack string keys shorter than this (if PACK_KEYS=1 isn't used). */
diff --git a/sql/sql_delete.cc b/sql/sql_delete.cc
index 25e470f56ea..2e86315d072 100644
--- a/sql/sql_delete.cc
+++ b/sql/sql_delete.cc
@@ -25,8 +25,7 @@
 #include "sql_cache.h"                          // query_cache_*
 #include "sql_base.h"                           // open_temprary_table
 #include "sql_table.h"                         // build_table_filename
-#include "lock.h"                              // lock_and_wait_for_table_name,
-                                               // unlock_table_name
+#include "lock.h"                              // unlock_table_name
 #include "sql_view.h"             // check_key_in_view, mysql_frm_type
 #include "sql_parse.h"            // mysql_init_select
 #include "sql_acl.h"              // *_ACL
diff --git a/sql/sql_lex.cc b/sql/sql_lex.cc
index b8971811f15..88cba22d115 100644
--- a/sql/sql_lex.cc
+++ b/sql/sql_lex.cc
@@ -377,7 +377,6 @@ void lex_start(THD *thd)
   lex->subqueries= FALSE;
   lex->view_prepare_mode= FALSE;
   lex->derived_tables= 0;
-  lex->lock_option= TL_READ;
   lex->safe_to_cache_query= 1;
   lex->leaf_tables_insert= 0;
   lex->parsing_options.reset();
@@ -390,7 +389,6 @@ void lex_start(THD *thd)
   lex->select_lex.ftfunc_list= &lex->select_lex.ftfunc_list_alloc;
   lex->select_lex.group_list.empty();
   lex->select_lex.order_list.empty();
-  lex->sql_command= SQLCOM_END;
   lex->duplicates= DUP_ERROR;
   lex->ignore= 0;
   lex->spname= NULL;
@@ -1735,7 +1733,6 @@ void st_select_lex::init_query()
   exclude_from_table_unique_test= no_wrap_view_item= FALSE;
   nest_level= 0;
   link_next= 0;
-  lock_option= TL_READ_DEFAULT;
 }
 
 void st_select_lex::init_select()
@@ -2247,6 +2244,7 @@ void LEX::cleanup_lex_after_parse_error(THD *thd)
 
 void Query_tables_list::reset_query_tables_list(bool init)
 {
+  sql_command= SQLCOM_END;
   if (!init && query_tables)
   {
     TABLE_LIST *table= query_tables;
@@ -2309,8 +2307,7 @@ void Query_tables_list::destroy_query_tables_list()
 */
 
 LEX::LEX()
-  :result(0),
-   sql_command(SQLCOM_END), option_type(OPT_DEFAULT), is_lex_started(0)
+  :result(0), option_type(OPT_DEFAULT), is_lex_started(0)
 {
 
   my_init_dynamic_array2(&plugins, sizeof(plugin_ref),
diff --git a/sql/sql_lex.h b/sql/sql_lex.h
index e9b5ca00953..40bd3875793 100644
--- a/sql/sql_lex.h
+++ b/sql/sql_lex.h
@@ -739,14 +739,6 @@ public:
 
   List<udf_func>     udf_list;                  /* udf function calls stack */
 
-  /**
-    Per sub-query locking strategy.
-    Note: This variable might interfer with the corresponding statement-level
-    variable Lex::lock_option because on how different parser rules depend
-    on eachother.
-  */
-  thr_lock_type lock_option;
-
   /* 
     This is a copy of the original JOIN USING list that comes from
     the parser. The parser :
@@ -1005,8 +997,11 @@ extern const LEX_STRING empty_lex_str;
 
 
 /*
-  Class representing list of all tables used by statement.
-  It also contains information about stored functions used by statement
+  Class representing list of all tables used by statement and other
+  information which is necessary for opening and locking its tables,
+  like SQL command for this statement.
+
+  Also contains information about stored functions used by statement
   since during its execution we may have to add all tables used by its
   stored functions/triggers to this list in order to pre-open and lock
   them.
@@ -1018,6 +1013,13 @@ extern const LEX_STRING empty_lex_str;
 class Query_tables_list
 {
 public:
+  /**
+    SQL command for this statement. Part of this class since the
+    process of opening and locking tables for the statement needs
+    this information to determine correct type of lock for some of
+    the tables.
+  */
+  enum_sql_command sql_command;
   /* Global list of all tables used by this statement */
   TABLE_LIST *query_tables;
   /* Pointer to next_global member of last element in the previous list. */
@@ -1920,7 +1922,6 @@ struct LEX: public Query_tables_list
     the variable can contain 0 or 1 for each nest level.
   */
   nesting_map allow_sum_func;
-  enum_sql_command sql_command;
 
   Sql_statement *m_stmt;
 
@@ -1932,7 +1933,6 @@ struct LEX: public Query_tables_list
   */
   bool expr_allows_subselect;
 
-  thr_lock_type lock_option;
   enum SSL_type ssl_type;			/* defined in violite.h */
   enum enum_duplicates duplicates;
   enum enum_tx_isolation tx_isolation;
@@ -2249,11 +2249,21 @@ public:
     yacc_yyss= NULL;
     yacc_yyvs= NULL;
     m_set_signal_info.clear();
+    m_lock_type= TL_READ_DEFAULT;
   }
 
   ~Yacc_state();
 
   /**
+    Reset part of the state which needs resetting before parsing
+    substatement.
+  */
+  void reset_before_substatement()
+  {
+    m_lock_type= TL_READ_DEFAULT;
+  }
+
+  /**
     Bison internal state stack, yyss, when dynamically allocated using
     my_yyoverflow().
   */
@@ -2271,6 +2281,25 @@ public:
   */
   Set_signal_information m_set_signal_info;
 
+  /**
+    Type of lock to be used for tables being added to the statement's
+    table list in table_factor, table_alias_ref, single_multi and
+    table_wild_one rules.
+    Statements which use these rules but require lock type different
+    from one specified by this member have to override it by using
+    st_select_lex::set_lock_for_tables() method.
+
+    The default value of this member is TL_READ_DEFAULT. The only two
+    cases in which we change it are:
+    - When parsing SELECT HIGH_PRIORITY.
+    - Rule for DELETE. In which we use this member to pass information
+      about type of lock from delete to single_multi part of rule.
+
+    We should try to avoid introducing new use cases as we would like
+    to get rid of this member eventually.
+  */
+  thr_lock_type m_lock_type;
+
   /*
     TODO: move more attributes from the LEX structure here.
   */
diff --git a/sql/sql_load.cc b/sql/sql_load.cc
index 6e8913b171d..2c42f29ae71 100644
--- a/sql/sql_load.cc
+++ b/sql/sql_load.cc
@@ -1,4 +1,4 @@
-/* Copyright (C) 2000-2006 MySQL AB, 2008-2009 Sun Microsystems, Inc
+/* Copyright (c) 2000, 2010 Oracle and/or its affiliates. All rights reserved.
 
    This program is free software; you can redistribute it and/or modify
    it under the terms of the GNU General Public License as published by
@@ -132,6 +132,7 @@ static int read_xml_field(THD *thd, COPY_INFO &info, TABLE_LIST *table_list,
 static bool write_execute_load_query_log_event(THD *thd, sql_exchange* ex,
                                                const char* db_arg, /* table's database */
                                                const char* table_name_arg,
+                                               bool is_concurrent,
                                                enum enum_duplicates duplicates,
                                                bool ignore,
                                                bool transactional_table,
@@ -184,6 +185,7 @@ int mysql_load(THD *thd,sql_exchange *ex,TABLE_LIST *table_list,
   char *tdb= thd->db ? thd->db : db;		// Result is never null
   ulong skip_lines= ex->skip_lines;
   bool transactional_table;
+  bool is_concurrent;
   THD::killed_state killed_status= THD::NOT_KILLED;
   DBUG_ENTER("mysql_load");
 
@@ -245,6 +247,7 @@ int mysql_load(THD *thd,sql_exchange *ex,TABLE_LIST *table_list,
 
   table= table_list->table;
   transactional_table= table->file->has_transactions();
+  is_concurrent= (table_list->lock_type == TL_WRITE_CONCURRENT_INSERT);
 
   if (!fields_vars.elements)
   {
@@ -557,6 +560,7 @@ int mysql_load(THD *thd,sql_exchange *ex,TABLE_LIST *table_list,
             (void) write_execute_load_query_log_event(thd, ex,
                                                       table_list->db, 
                                                       table_list->table_name,
+                                                      is_concurrent,
                                                       handle_duplicates, ignore,
                                                       transactional_table,
                                                       errcode);
@@ -604,6 +608,7 @@ int mysql_load(THD *thd,sql_exchange *ex,TABLE_LIST *table_list,
         int errcode= query_error_code(thd, killed_status == THD::NOT_KILLED);
         error= write_execute_load_query_log_event(thd, ex,
                                                   table_list->db, table_list->table_name,
+                                                  is_concurrent,
                                                   handle_duplicates, ignore,
                                                   transactional_table,
                                                   errcode);
@@ -632,6 +637,7 @@ err:
 static bool write_execute_load_query_log_event(THD *thd, sql_exchange* ex,
                                                const char* db_arg,  /* table's database */
                                                const char* table_name_arg,
+                                               bool is_concurrent,
                                                enum enum_duplicates duplicates,
                                                bool ignore,
                                                bool transactional_table,
@@ -667,8 +673,8 @@ static bool write_execute_load_query_log_event(THD *thd, sql_exchange* ex,
     tbl= string_buf.c_ptr_safe();
   }
 
-  Load_log_event       lle(thd, ex, tdb, tbl, fv, duplicates,
-                           ignore, transactional_table);
+  Load_log_event       lle(thd, ex, tdb, tbl, fv, is_concurrent,
+                           duplicates, ignore, transactional_table);
 
   /*
     force in a LOCAL if there was one in the original.
@@ -1702,7 +1708,7 @@ bool READ_INFO::find_start_of_fields()
 /*
   Clear taglist from tags with a specified level
 */
-int READ_INFO::clear_level(int level)
+int READ_INFO::clear_level(int level_arg)
 {
   DBUG_ENTER("READ_INFO::read_xml clear_level");
   List_iterator<XML_TAG> xmlit(taglist);
@@ -1711,7 +1717,7 @@ int READ_INFO::clear_level(int level)
   
   while ((tag= xmlit++))
   {
-     if(tag->level >= level)
+     if(tag->level >= level_arg)
      {
        xmlit.remove();
        delete tag;
diff --git a/sql/sql_parse.cc b/sql/sql_parse.cc
index 90302c0f7ac..80a41fae9b9 100644
--- a/sql/sql_parse.cc
+++ b/sql/sql_parse.cc
@@ -247,8 +247,19 @@ void init_update_queries(void)
   /* Initialize the sql command flags array. */
   memset(sql_command_flags, 0, sizeof(sql_command_flags));
 
+  /*
+    In general, DDL statements do not generate row events and do not go
+    through a cache before being written to the binary log. However, the
+    CREATE TABLE...SELECT is an exception because it may generate row
+    events. For that reason,  the SQLCOM_CREATE_TABLE  which represents
+    a CREATE TABLE, including the CREATE TABLE...SELECT, has the
+    CF_CAN_GENERATE_ROW_EVENTS flag. The distinction between a regular
+    CREATE TABLE and the CREATE TABLE...SELECT is made in other parts of
+    the code, in particular in the Query_log_event's constructor.
+  */
   sql_command_flags[SQLCOM_CREATE_TABLE]=   CF_CHANGES_DATA | CF_REEXECUTION_FRAGILE |
-                                            CF_AUTO_COMMIT_TRANS | CF_PROTECT_AGAINST_GRL;
+                                            CF_AUTO_COMMIT_TRANS | CF_PROTECT_AGAINST_GRL |
+                                            CF_CAN_GENERATE_ROW_EVENTS;
   sql_command_flags[SQLCOM_CREATE_INDEX]=   CF_CHANGES_DATA | CF_AUTO_COMMIT_TRANS;
   sql_command_flags[SQLCOM_ALTER_TABLE]=    CF_CHANGES_DATA | CF_WRITE_LOGS_COMMAND |
                                             CF_AUTO_COMMIT_TRANS | CF_PROTECT_AGAINST_GRL;
@@ -256,7 +267,8 @@ void init_update_queries(void)
                                             CF_AUTO_COMMIT_TRANS;
   sql_command_flags[SQLCOM_DROP_TABLE]=     CF_CHANGES_DATA | CF_AUTO_COMMIT_TRANS;
   sql_command_flags[SQLCOM_LOAD]=           CF_CHANGES_DATA | CF_REEXECUTION_FRAGILE |
-                                            CF_PROTECT_AGAINST_GRL;
+                                            CF_PROTECT_AGAINST_GRL |
+                                            CF_CAN_GENERATE_ROW_EVENTS;
   sql_command_flags[SQLCOM_CREATE_DB]=      CF_CHANGES_DATA | CF_AUTO_COMMIT_TRANS;
   sql_command_flags[SQLCOM_DROP_DB]=        CF_CHANGES_DATA | CF_AUTO_COMMIT_TRANS;
   sql_command_flags[SQLCOM_ALTER_DB_UPGRADE]= CF_AUTO_COMMIT_TRANS;
@@ -275,22 +287,32 @@ void init_update_queries(void)
   sql_command_flags[SQLCOM_DROP_TRIGGER]=   CF_AUTO_COMMIT_TRANS;
 
   sql_command_flags[SQLCOM_UPDATE]=	    CF_CHANGES_DATA | CF_REEXECUTION_FRAGILE |
-                                            CF_PROTECT_AGAINST_GRL;
+                                            CF_PROTECT_AGAINST_GRL |
+                                            CF_CAN_GENERATE_ROW_EVENTS;
   sql_command_flags[SQLCOM_UPDATE_MULTI]=   CF_CHANGES_DATA | CF_REEXECUTION_FRAGILE |
-                                            CF_PROTECT_AGAINST_GRL;
+                                            CF_PROTECT_AGAINST_GRL |
+                                            CF_CAN_GENERATE_ROW_EVENTS;
   sql_command_flags[SQLCOM_INSERT]=	    CF_CHANGES_DATA | CF_REEXECUTION_FRAGILE |
-                                            CF_PROTECT_AGAINST_GRL;
+                                            CF_PROTECT_AGAINST_GRL |
+                                            CF_CAN_GENERATE_ROW_EVENTS;
   sql_command_flags[SQLCOM_INSERT_SELECT]=  CF_CHANGES_DATA | CF_REEXECUTION_FRAGILE |
-                                            CF_PROTECT_AGAINST_GRL;
+                                            CF_PROTECT_AGAINST_GRL |
+                                            CF_CAN_GENERATE_ROW_EVENTS;
   sql_command_flags[SQLCOM_DELETE]=         CF_CHANGES_DATA | CF_REEXECUTION_FRAGILE |
-                                            CF_PROTECT_AGAINST_GRL;
+                                            CF_PROTECT_AGAINST_GRL |
+                                            CF_CAN_GENERATE_ROW_EVENTS;
   sql_command_flags[SQLCOM_DELETE_MULTI]=   CF_CHANGES_DATA | CF_REEXECUTION_FRAGILE |
-                                            CF_PROTECT_AGAINST_GRL;
-  sql_command_flags[SQLCOM_REPLACE]=        CF_CHANGES_DATA | CF_REEXECUTION_FRAGILE;
-  sql_command_flags[SQLCOM_REPLACE_SELECT]= CF_CHANGES_DATA | CF_REEXECUTION_FRAGILE;
-  sql_command_flags[SQLCOM_SELECT]=         CF_REEXECUTION_FRAGILE;
+                                            CF_PROTECT_AGAINST_GRL |
+                                            CF_CAN_GENERATE_ROW_EVENTS;
+  sql_command_flags[SQLCOM_REPLACE]=        CF_CHANGES_DATA | CF_REEXECUTION_FRAGILE |
+                                            CF_CAN_GENERATE_ROW_EVENTS;
+  sql_command_flags[SQLCOM_REPLACE_SELECT]= CF_CHANGES_DATA | CF_REEXECUTION_FRAGILE |
+                                            CF_CAN_GENERATE_ROW_EVENTS;
+  sql_command_flags[SQLCOM_SELECT]=         CF_REEXECUTION_FRAGILE |
+                                            CF_CAN_GENERATE_ROW_EVENTS;
   sql_command_flags[SQLCOM_SET_OPTION]=     CF_REEXECUTION_FRAGILE | CF_AUTO_COMMIT_TRANS;
-  sql_command_flags[SQLCOM_DO]=             CF_REEXECUTION_FRAGILE;
+  sql_command_flags[SQLCOM_DO]=             CF_REEXECUTION_FRAGILE |
+                                            CF_CAN_GENERATE_ROW_EVENTS;
 
   sql_command_flags[SQLCOM_SHOW_STATUS_PROC]= CF_STATUS_COMMAND | CF_REEXECUTION_FRAGILE;
   sql_command_flags[SQLCOM_SHOW_STATUS]=      CF_STATUS_COMMAND | CF_REEXECUTION_FRAGILE;
@@ -365,7 +387,9 @@ void init_update_queries(void)
     last called (or executed) statement is preserved.
     See mysql_execute_command() for how CF_ROW_COUNT is used.
   */
-  sql_command_flags[SQLCOM_CALL]=      CF_REEXECUTION_FRAGILE;
+  sql_command_flags[SQLCOM_CALL]=      CF_REEXECUTION_FRAGILE |
+                                       CF_CAN_GENERATE_ROW_EVENTS;
+  sql_command_flags[SQLCOM_EXECUTE]=   CF_CAN_GENERATE_ROW_EVENTS;
 
   /*
     The following admin table operations are allowed
@@ -390,7 +414,12 @@ void init_update_queries(void)
   sql_command_flags[SQLCOM_CHECK]=              CF_AUTO_COMMIT_TRANS;
 }
 
-
+bool sqlcom_can_generate_row_events(const THD *thd)
+{
+  return (sql_command_flags[thd->lex->sql_command] &
+          CF_CAN_GENERATE_ROW_EVENTS);
+}
+ 
 bool is_update_query(enum enum_sql_command command)
 {
   DBUG_ASSERT(command >= 0 && command <= SQLCOM_END);
@@ -2682,6 +2711,10 @@ case SQLCOM_PREPARE:
         */
         lex->unlink_first_table(&link_to_local);
 
+        /* So that CREATE TEMPORARY TABLE gets to binlog at commit/rollback */
+        if (create_info.options & HA_LEX_CREATE_TMP_TABLE)
+          thd->variables.option_bits|= OPTION_KEEP_LOG;
+
         /*
           select_create is currently not re-execution friendly and
           needs to be created for every execution of a PS/SP.
@@ -2792,7 +2825,7 @@ end_with_restore_list:
     client thread has locked tables
   */
   if (thd->locked_tables_mode ||
-      thd->active_transaction() || thd->global_read_lock.is_acquired())
+      thd->in_active_multi_stmt_transaction() || thd->global_read_lock.is_acquired())
   {
     my_message(ER_LOCK_OR_ACTIVE_TRANSACTION,
                ER(ER_LOCK_OR_ACTIVE_TRANSACTION), MYF(0));
@@ -3300,7 +3333,7 @@ end_with_restore_list:
       Don't allow this within a transaction because we want to use
       re-generate table
     */
-    if (thd->active_transaction())
+    if (thd->in_active_multi_stmt_transaction())
     {
       my_message(ER_LOCK_OR_ACTIVE_TRANSACTION,
                  ER(ER_LOCK_OR_ACTIVE_TRANSACTION), MYF(0));
@@ -4730,6 +4763,9 @@ finish:
     thd->global_read_lock.start_waiting_global_read_lock(thd);
   }
 
+  DBUG_ASSERT(!thd->in_active_multi_stmt_transaction() ||
+               thd->in_multi_stmt_transaction_mode());
+
   if (stmt_causes_implicit_commit(thd, CF_IMPLICIT_COMMIT_END))
   {
     /* If commit fails, we should be able to reset the OK status. */
@@ -5543,7 +5579,7 @@ void THD::reset_for_next_command()
     OPTION_STATUS_NO_TRANS_UPDATE | OPTION_KEEP_LOG to not get warnings
     in ha_rollback_trans() about some tables couldn't be rolled back.
   */
-  if (!thd->in_multi_stmt_transaction())
+  if (!thd->in_multi_stmt_transaction_mode())
   {
     thd->variables.option_bits&= ~OPTION_KEEP_LOG;
     thd->transaction.all.modified_non_trans_table= FALSE;
@@ -5729,7 +5765,6 @@ void mysql_init_multi_delete(LEX *lex)
   lex->select_lex.select_limit= 0;
   lex->unit.select_limit_cnt= HA_POS_ERROR;
   lex->select_lex.table_list.save_and_clear(&lex->auxiliary_table_list);
-  lex->lock_option= TL_READ_DEFAULT;
   lex->query_tables= 0;
   lex->query_tables_last= &lex->query_tables;
 }
diff --git a/sql/sql_parse.h b/sql/sql_parse.h
index e1543a09549..6d968033ccd 100644
--- a/sql/sql_parse.h
+++ b/sql/sql_parse.h
@@ -79,6 +79,7 @@ bool check_host_name(LEX_STRING *str);
 bool check_identifier_name(LEX_STRING *str, uint max_char_length,
                            uint err_code, const char *param_for_err_msg);
 bool mysql_test_parse_for_slave(THD *thd,char *inBuf,uint length);
+bool sqlcom_can_generate_row_events(const THD *thd);
 bool is_update_query(enum enum_sql_command command);
 bool is_log_table_write_query(enum enum_sql_command command);
 bool alloc_query(THD *thd, const char *packet, uint packet_length);
diff --git a/sql/sql_partition.cc b/sql/sql_partition.cc
index a3007d1875c..fa9c698622b 100644
--- a/sql/sql_partition.cc
+++ b/sql/sql_partition.cc
@@ -3096,7 +3096,7 @@ int get_partition_id_list_col(partition_info *part_info,
     }
     else
     {
-      *part_id= (uint32)list_col_array[list_index].partition_id;
+      *part_id= (uint32)list_col_array[list_index*num_columns].partition_id;
       DBUG_RETURN(0);
     }
   }
@@ -4186,7 +4186,6 @@ void get_partition_set(const TABLE *table, uchar *buf, const uint index,
 
 bool mysql_unpack_partition(THD *thd,
                             const char *part_buf, uint part_info_len,
-                            const char *part_state, uint part_state_len,
                             TABLE* table, bool is_create_table_ind,
                             handlerton *default_db_type,
                             bool *work_part_info_used)
@@ -4222,8 +4221,6 @@ bool mysql_unpack_partition(THD *thd,
     goto end;
   }
   part_info= lex.part_info;
-  part_info->part_state= part_state;
-  part_info->part_state_len= part_state_len;
   DBUG_PRINT("info", ("Parse: %s", part_buf));
   if (parse_sql(thd, & parser_state, NULL) ||
       part_info->fix_parser_data(thd))
@@ -4587,7 +4584,7 @@ uint prep_alter_part_table(THD *thd, TABLE *table, Alter_info *alter_info,
     partition_info *tab_part_info= table->part_info;
     partition_info *alt_part_info= thd->work_part_info;
     uint flags= 0;
-    bool is_last_partition_reorged;
+    bool is_last_partition_reorged= FALSE;
     part_elem_value *tab_max_elem_val= NULL;
     part_elem_value *alt_max_elem_val= NULL;
     longlong tab_max_range= 0, alt_max_range= 0;
@@ -7647,7 +7644,7 @@ uint32 get_next_partition_id_range(PARTITION_ITERATOR* part_iter)
   DESCRIPTION
     This implementation of PARTITION_ITERATOR::get_next() is special for 
     LIST partitioning: it enumerates partition ids in
-    part_info->list_array[i] (list_col_array[i] for COLUMNS LIST
+    part_info->list_array[i] (list_col_array[i*cols] for COLUMNS LIST
     partitioning) where i runs over [min_idx, max_idx] interval.
     The function conforms to partition_iter_func type.
 
@@ -7673,9 +7670,12 @@ uint32 get_next_partition_id_list(PARTITION_ITERATOR *part_iter)
   {
     partition_info *part_info= part_iter->part_info;
     uint32 num_part= part_iter->part_nums.cur++;
-    return part_info->column_list ?
-           part_info->list_col_array[num_part].partition_id :
-           part_info->list_array[num_part].partition_id;
+    if (part_info->column_list)
+    {
+      uint num_columns= part_info->part_field_list.elements;
+      return part_info->list_col_array[num_part*num_columns].partition_id;
+    }
+    return part_info->list_array[num_part].partition_id;
   }
 }
 
diff --git a/sql/sql_partition.h b/sql/sql_partition.h
index 8542074e718..2e0ea740555 100644
--- a/sql/sql_partition.h
+++ b/sql/sql_partition.h
@@ -64,20 +64,6 @@ typedef struct st_lock_param_type
   partition_info *part_info;
 } ALTER_PARTITION_PARAM_TYPE;
 
-
-/*typedef struct {
-  ulonglong data_file_length;
-  ulonglong max_data_file_length;
-  ulonglong index_file_length;
-  ulonglong delete_length;
-  ha_rows records;
-  ulong mean_rec_length;
-  time_t create_time;
-  time_t check_time;
-  time_t update_time;
-  ulonglong check_sum;
-} PARTITION_INFO;
-*/
 typedef struct {
   longlong list_value;
   uint32 partition_id;
@@ -126,7 +112,6 @@ void get_full_part_id_from_key(const TABLE *table, uchar *buf,
                                part_id_range *part_spec);
 bool mysql_unpack_partition(THD *thd, const char *part_buf,
                             uint part_info_len,
-                            const char *part_state, uint part_state_len,
                             TABLE *table, bool is_create_table_ind,
                             handlerton *default_db_type,
                             bool *work_part_info_used);
diff --git a/sql/sql_prepare.cc b/sql/sql_prepare.cc
index 739b6576a99..c2d3c595d95 100644
--- a/sql/sql_prepare.cc
+++ b/sql/sql_prepare.cc
@@ -1708,7 +1708,7 @@ static bool mysql_test_create_table(Prepared_statement *stmt)
 
     res= select_like_stmt_test(stmt, 0, 0);
 
-    lex->link_first_table_back(create_table, &link_to_local);
+    lex->link_first_table_back(create_table, link_to_local);
   }
   else
   {
@@ -3246,7 +3246,7 @@ bool Prepared_statement::prepare(const char *packet, uint packet_len)
     locks have already been released and our savepoint points
     to ticket which has been released as well.
   */
-  if (thd->in_multi_stmt_transaction())
+  if (thd->in_multi_stmt_transaction_mode())
     thd->mdl_context.rollback_to_savepoint(mdl_savepoint);
   thd->restore_backup_statement(this, &stmt_backup);
   thd->stmt_arena= old_stmt_arena;
diff --git a/sql/sql_priv.h b/sql/sql_priv.h
index eeefd3cac04..8601b10b9bf 100644
--- a/sql/sql_priv.h
+++ b/sql/sql_priv.h
@@ -129,6 +129,12 @@ extern char err_shared_dir[];
 */
 #define TMP_TABLE_FORCE_MYISAM          (1ULL << 32)
 #define OPTION_PROFILING                (1ULL << 33)
+/**
+  Indicates that this is a HIGH_PRIORITY SELECT.
+  Currently used only for printing of such selects.
+  Type of locks to be acquired is specified directly.
+*/
+#define SELECT_HIGH_PRIORITY            (1ULL << 34)     // SELECT, user
 
 
 /* The rest of the file is included in the server only */
@@ -181,10 +187,6 @@ extern char err_shared_dir[];
 
 #define BINLOG_DUMP_NON_BLOCK   1
 
-/* sql_show.cc:show_log_files() */
-#define SHOW_LOG_STATUS_FREE "FREE"
-#define SHOW_LOG_STATUS_INUSE "IN USE"
-
 /*
   Some defines for exit codes for ::is_equal class functions.
 */
diff --git a/sql/sql_rename.cc b/sql/sql_rename.cc
index d387010141c..ea95b59b0c2 100644
--- a/sql/sql_rename.cc
+++ b/sql/sql_rename.cc
@@ -54,7 +54,7 @@ bool mysql_rename_tables(THD *thd, TABLE_LIST *table_list, bool silent)
     if the user is trying to to do this in a transcation context
   */
 
-  if (thd->locked_tables_mode || thd->active_transaction())
+  if (thd->locked_tables_mode || thd->in_active_multi_stmt_transaction())
   {
     my_message(ER_LOCK_OR_ACTIVE_TRANSACTION,
                ER(ER_LOCK_OR_ACTIVE_TRANSACTION), MYF(0));
diff --git a/sql/sql_select.cc b/sql/sql_select.cc
index 3f04adf066c..45e30bb5be3 100644
--- a/sql/sql_select.cc
+++ b/sql/sql_select.cc
@@ -44,7 +44,7 @@
 #include "sql_partition.h"       // make_used_partitions_str
 #include "sql_acl.h"             // *_ACL
 #include "sql_test.h"            // print_where, print_keyuse_array,
-                                 // print_sjm, print_plan
+                                 // print_sjm, print_plan, TEST_join
 #include "records.h"             // init_read_record, end_read_record
 #include "filesort.h"            // filesort_free_buffers
 #include "sql_union.h"           // mysql_union
@@ -90,8 +90,10 @@ static bool best_extension_by_limited_search(JOIN *join,
                                              double read_time, uint depth,
                                              uint prune_level);
 static uint determine_search_depth(JOIN* join);
+C_MODE_START
 static int join_tab_cmp(const void* ptr1, const void* ptr2);
 static int join_tab_cmp_straight(const void* ptr1, const void* ptr2);
+C_MODE_END
 /*
   TODO: 'find_best' is here only temporarily until 'greedy_search' is
   tested and approved.
@@ -17252,8 +17254,7 @@ void st_select_lex::print(THD *thd, String *str, enum_query_type query_type)
   /* First add options */
   if (options & SELECT_STRAIGHT_JOIN)
     str->append(STRING_WITH_LEN("straight_join "));
-  if ((thd->lex->lock_option == TL_READ_HIGH_PRIORITY) &&
-      (this == &thd->lex->select_lex))
+  if (options & SELECT_HIGH_PRIORITY)
     str->append(STRING_WITH_LEN("high_priority "));
   if (options & SELECT_DISTINCT)
     str->append(STRING_WITH_LEN("distinct "));
diff --git a/sql/sql_select.h b/sql/sql_select.h
index ccf88c2cc5c..2c44dba74c3 100644
--- a/sql/sql_select.h
+++ b/sql/sql_select.h
@@ -597,7 +597,6 @@ typedef struct st_select_check {
 } SELECT_CHECK;
 
 extern const char *join_type_str[];
-void TEST_join(JOIN *join);
 
 /* Extern functions in sql_select.cc */
 bool store_val_in_field(Field *field, Item *val, enum_check_fields check_flag);
diff --git a/sql/sql_show.cc b/sql/sql_show.cc
index 8daf7a73a80..16a17744279 100644
--- a/sql/sql_show.cc
+++ b/sql/sql_show.cc
@@ -18,6 +18,7 @@
 
 #include "my_global.h"                          /* NO_EMBEDDED_ACCESS_CHECKS */
 #include "sql_priv.h"
+#include "debug_sync.h"
 #include "unireg.h"
 #include "sql_acl.h"                        // fill_schema_*_privileges
 #include "sql_select.h"                         // For select_describe
@@ -1944,10 +1945,13 @@ int fill_schema_processlist(THD* thd, TABLE_LIST* tables, COND* cond)
 
 static DYNAMIC_ARRAY all_status_vars;
 static bool status_vars_inited= 0;
+
+C_MODE_START
 static int show_var_cmp(const void *var1, const void *var2)
 {
   return strcmp(((SHOW_VAR*)var1)->name, ((SHOW_VAR*)var2)->name);
 }
+C_MODE_END
 
 /*
   deletes all the SHOW_UNDEF elements from the array and calls
@@ -3296,12 +3300,17 @@ static int fill_schema_table_from_frm(THD *thd, TABLE_LIST *tables,
     goto end_share;
   }
 
+  if (!open_table_from_share(thd, share, table_name->str, 0,
+                             (EXTRA_RECORD | OPEN_FRM_FILE_ONLY),
+                             thd->open_options, &tbl, FALSE))
   {
     tbl.s= share;
     table_list.table= &tbl;
     table_list.view= (LEX*) share->is_view;
     res= schema_table->process_table(thd, &table_list, table,
                                      res, db_name, table_name);
+    free_root(&tbl.mem_root, MYF(0));
+    my_free((char*) tbl.alias, MYF(MY_ALLOW_ZERO_PTR));
   }
 
 end_share:
@@ -3345,7 +3354,6 @@ int get_all_tables(THD *thd, TABLE_LIST *tables, COND *cond)
   LEX *lex= thd->lex;
   TABLE *table= tables->table;
   SELECT_LEX *old_all_select_lex= lex->all_selects_list;
-  enum_sql_command save_sql_command= lex->sql_command;
   SELECT_LEX *lsel= tables->schema_select_lex;
   ST_SCHEMA_TABLE *schema_table= tables->schema_table;
   SELECT_LEX sel;
@@ -3381,6 +3389,12 @@ int get_all_tables(THD *thd, TABLE_LIST *tables, COND *cond)
 
   lex->view_prepare_mode= TRUE;
   lex->reset_n_backup_query_tables_list(&query_tables_list_backup);
+  /*
+    Restore Query_tables_list::sql_command value, which was reset
+    above, as ST_SCHEMA_TABLE::process_table() functions often rely
+    that this value reflects which SHOW statement is executed.
+  */
+  lex->sql_command= query_tables_list_backup.sql_command;
 
   /*
     We should not introduce deadlocks even if we already have some
@@ -3544,7 +3558,7 @@ int get_all_tables(THD *thd, TABLE_LIST *tables, COND *cond)
                    (MYSQL_OPEN_IGNORE_FLUSH |
                     MYSQL_OPEN_FORCE_SHARED_HIGH_PRIO_MDL |
                     (can_deadlock ? MYSQL_OPEN_FAIL_ON_MDL_CONFLICT : 0)));
-            lex->sql_command= save_sql_command;
+            lex->sql_command= query_tables_list_backup.sql_command;
             /*
               XXX:  show_table_list has a flag i_is_requested,
               and when it's set, open_normal_and_derived_tables()
@@ -3603,7 +3617,6 @@ err:
   lex->derived_tables= derived_tables;
   lex->all_selects_list= old_all_select_lex;
   lex->view_prepare_mode= save_view_prepare_mode;
-  lex->sql_command= save_sql_command;
   DBUG_RETURN(error);
 }
 
@@ -3751,7 +3764,7 @@ static int get_schema_tables_record(THD *thd, TABLE_LIST *tables,
     }
 #ifdef WITH_PARTITION_STORAGE_ENGINE
     if (share->db_type() == partition_hton &&
-        share->partition_info_len)
+        share->partition_info_str_len)
     {
       tmp_db_type= share->default_part_db_type;
       is_partitioned= TRUE;
@@ -4025,7 +4038,6 @@ static int get_schema_column_record(THD *thd, TABLE_LIST *tables,
   const char *wild= lex->wild ? lex->wild->ptr() : NullS;
   CHARSET_INFO *cs= system_charset_info;
   TABLE *show_table;
-  TABLE_SHARE *show_table_share;
   Field **ptr, *field, *timestamp_field;
   int count;
   DBUG_ENTER("get_schema_column_record");
@@ -4048,37 +4060,11 @@ static int get_schema_column_record(THD *thd, TABLE_LIST *tables,
   }
 
   show_table= tables->table;
-  show_table_share= show_table->s;
   count= 0;
-
-  if (tables->view || tables->schema_table)
-  {
-    ptr= show_table->field;
-    timestamp_field= show_table->timestamp_field;
-    show_table->use_all_columns();               // Required for default
-  }
-  else
-  {
-    ptr= show_table_share->field;
-    timestamp_field= show_table_share->timestamp_field;
-    /*
-      read_set may be inited in case of
-      temporary table
-    */
-    if (!show_table->read_set)
-    {
-      /* to satisfy 'field->val_str' ASSERTs */
-      uchar *bitmaps;
-      uint bitmap_size= show_table_share->column_bitmap_size;
-      if (!(bitmaps= (uchar*) alloc_root(thd->mem_root, bitmap_size)))
-        DBUG_RETURN(0);
-      bitmap_init(&show_table->def_read_set,
-                  (my_bitmap_map*) bitmaps, show_table_share->fields, FALSE);
-      bitmap_set_all(&show_table->def_read_set);
-      show_table->read_set= &show_table->def_read_set;
-    }
-    bitmap_set_all(show_table->read_set);
-  }
+  ptr= show_table->field;
+  timestamp_field= show_table->timestamp_field;
+  show_table->use_all_columns();               // Required for default
+  restore_record(show_table, s->default_values);
 
   for (; (field= *ptr) ; ptr++)
   {
@@ -4087,9 +4073,7 @@ static int get_schema_column_record(THD *thd, TABLE_LIST *tables,
     String type(tmp,sizeof(tmp), system_charset_info);
     char *end;
 
-    /* to satisfy 'field->val_str' ASSERTs */
-    field->table= show_table;
-    show_table->in_use= thd;
+    DEBUG_SYNC(thd, "get_schema_column");
 
     if (wild && wild[0] &&
         wild_case_compare(system_charset_info, field->field_name,wild))
@@ -5240,7 +5224,8 @@ static int get_schema_key_column_usage_record(THD *thd,
 
 
 #ifdef WITH_PARTITION_STORAGE_ENGINE
-static void collect_partition_expr(List<char> &field_list, String *str)
+static void collect_partition_expr(THD *thd, List<char> &field_list,
+                                   String *str)
 {
   List_iterator<char> part_it(field_list);
   ulong no_fields= field_list.elements;
@@ -5248,7 +5233,7 @@ static void collect_partition_expr(List<char> &field_list, String *str)
   str->length(0);
   while ((field_str= part_it++))
   {
-    str->append(field_str);
+    append_identifier(thd, str, field_str, strlen(field_str));
     if (--no_fields != 0)
       str->append(",");
   }
@@ -5315,7 +5300,7 @@ static void store_schema_partitions_record(THD *thd, TABLE *schema_table,
 {
   TABLE* table= schema_table;
   CHARSET_INFO *cs= system_charset_info;
-  PARTITION_INFO stat_info;
+  PARTITION_STATS stat_info;
   MYSQL_TIME time;
   file->get_dynamic_partition_info(&stat_info, part_id);
   table->field[0]->store(STRING_WITH_LEN("def"), cs);
@@ -5515,7 +5500,7 @@ static int get_schema_partitions_record(THD *thd, TABLE_LIST *tables,
     }
     else if (part_info->list_of_part_fields)
     {
-      collect_partition_expr(part_info->part_field_list, &tmp_str);
+      collect_partition_expr(thd, part_info->part_field_list, &tmp_str);
       table->field[9]->store(tmp_str.ptr(), tmp_str.length(), cs);
     }
     table->field[9]->set_notnull();
@@ -5544,7 +5529,7 @@ static int get_schema_partitions_record(THD *thd, TABLE_LIST *tables,
       }
       else if (part_info->list_of_subpart_fields)
       {
-        collect_partition_expr(part_info->subpart_field_list, &tmp_str);
+        collect_partition_expr(thd, part_info->subpart_field_list, &tmp_str);
         table->field[10]->store(tmp_str.ptr(), tmp_str.length(), cs);
       }
       table->field[10]->set_notnull();
diff --git a/sql/sql_string.cc b/sql/sql_string.cc
index 9fbc06b7529..762eebba031 100644
--- a/sql/sql_string.cc
+++ b/sql/sql_string.cc
@@ -24,13 +24,6 @@
 #include <m_string.h>
 #include <m_ctype.h>
 #include <mysql_com.h>
-/*
-  The following extern declarations are ok as these are interface functions
-  required by the string function
-*/
-
-extern uchar* sql_alloc(unsigned size);
-extern void sql_element_free(void *ptr);
 
 #include "sql_string.h"
 
diff --git a/sql/sql_table.cc b/sql/sql_table.cc
index 9177573971e..5052e29ac30 100644
--- a/sql/sql_table.cc
+++ b/sql/sql_table.cc
@@ -1782,11 +1782,12 @@ bool mysql_write_frm(ALTER_PARTITION_PARAM_TYPE *lpt, uint flags)
           error= 1;
           goto err;
         }
-        share->partition_info= tmp_part_syntax_str;
+        share->partition_info_str= tmp_part_syntax_str;
       }
       else
-        memcpy((char*) share->partition_info, part_syntax_buf, syntax_len + 1);
-      share->partition_info_len= part_info->part_info_len= syntax_len;
+        memcpy((char*) share->partition_info_str, part_syntax_buf,
+               syntax_len + 1);
+      share->partition_info_str_len= part_info->part_info_len= syntax_len;
       part_info->part_info_string= part_syntax_buf;
     }
 #endif
@@ -1929,8 +1930,8 @@ int mysql_rm_table_part2(THD *thd, TABLE_LIST *tables, bool if_exists,
 			 bool dont_log_query)
 {
   TABLE_LIST *table;
-  char path[FN_REFLEN + 1], *alias;
-  uint path_length;
+  char path[FN_REFLEN + 1], *alias= NULL;
+  uint path_length= 0;
   String wrong_tables;
   int error= 0;
   int non_temp_tables_count= 0;
@@ -1939,9 +1940,6 @@ int mysql_rm_table_part2(THD *thd, TABLE_LIST *tables, bool if_exists,
   String built_tmp_query;
   DBUG_ENTER("mysql_rm_table_part2");
 
-  LINT_INIT(alias);
-  LINT_INIT(path_length);
-
   if (thd->is_current_stmt_binlog_format_row() && !dont_log_query)
   {
     built_query.set_charset(system_charset_info);
@@ -4839,6 +4837,7 @@ static bool mysql_admin_table(THD* thd, TABLE_LIST* tables,
       /* purecov: begin inspected */
       char buff[FN_REFLEN + MYSQL_ERRMSG_SIZE];
       size_t length;
+      enum_sql_command save_sql_command= lex->sql_command;
       DBUG_PRINT("admin", ("sending error message"));
       protocol->prepare_for_resend();
       protocol->store(table_name, system_charset_info);
@@ -4852,6 +4851,11 @@ static bool mysql_admin_table(THD* thd, TABLE_LIST* tables,
       close_thread_tables(thd);
       thd->mdl_context.release_transactional_locks();
       lex->reset_query_tables_list(FALSE);
+      /*
+        Restore Query_tables_list::sql_command value to make statement
+        safe for re-execution.
+      */
+      lex->sql_command= save_sql_command;
       table->table=0;				// For query cache
       if (protocol->write())
 	goto err;
@@ -5049,7 +5053,7 @@ send_result_message:
         /* Clear the ticket released in close_thread_tables(). */
         table->mdl_request.ticket= NULL;
         DEBUG_SYNC(thd, "ha_admin_open_ltable");
-        if (table->table= open_ltable(thd, table, lock_type, 0))
+        if ((table->table= open_ltable(thd, table, lock_type, 0)))
         {
           result_code= table->table->file->ha_analyze(thd, check_opt);
           if (result_code == HA_ADMIN_ALREADY_DONE)
@@ -6584,7 +6588,7 @@ bool mysql_alter_table(THD *thd,char *new_db, char *new_name,
       if the user is trying to to do this in a transcation context
     */
 
-    if (thd->locked_tables_mode || thd->active_transaction())
+    if (thd->locked_tables_mode || thd->in_active_multi_stmt_transaction())
     {
       my_message(ER_LOCK_OR_ACTIVE_TRANSACTION,
                  ER(ER_LOCK_OR_ACTIVE_TRANSACTION), MYF(0));
diff --git a/sql/sql_test.cc b/sql/sql_test.cc
index d34aee854d0..43d203e6498 100644
--- a/sql/sql_test.cc
+++ b/sql/sql_test.cc
@@ -76,7 +76,7 @@ print_where(COND *cond,const char *info, enum_query_type query_type)
 	/* This is for debugging purposes */
 
 
-void print_cached_tables(void)
+static void print_cached_tables(void)
 {
   uint idx,count,unused;
   TABLE_SHARE *share;
@@ -341,6 +341,11 @@ print_plan(JOIN* join, uint idx, double record_count, double read_time,
 
 #endif
 
+C_MODE_START
+static int dl_compare(const void *p1, const void *p2);
+static int print_key_cache_status(const char *name, KEY_CACHE *key_cache);
+C_MODE_END
+
 typedef struct st_debug_lock
 {
   ulong thread_id;
@@ -350,8 +355,13 @@ typedef struct st_debug_lock
   enum thr_lock_type type;
 } TABLE_LOCK_INFO;
 
-static int dl_compare(TABLE_LOCK_INFO *a,TABLE_LOCK_INFO *b)
+static int dl_compare(const void *p1, const void *p2)
 {
+  TABLE_LOCK_INFO *a, *b;
+
+  a= (TABLE_LOCK_INFO *) p1;
+  b= (TABLE_LOCK_INFO *) p2;
+
   if (a->thread_id > b->thread_id)
     return 1;
   if (a->thread_id < b->thread_id)
@@ -401,9 +411,10 @@ static void push_locks_into_array(DYNAMIC_ARRAY *ar, THR_LOCK_DATA *data,
   function so that we can easily add this if we ever need this.
 */
 
-static void display_table_locks(void) 
+static void display_table_locks(void)
 {
   LIST *list;
+  void *saved_base;
   DYNAMIC_ARRAY saved_table_locks;
 
   (void) my_init_dynamic_array(&saved_table_locks,sizeof(TABLE_LOCK_INFO), table_cache_count + 20,50);
@@ -424,13 +435,17 @@ static void display_table_locks(void)
     mysql_mutex_unlock(&lock->mutex);
   }
   mysql_mutex_unlock(&THR_LOCK_lock);
-  if (!saved_table_locks.elements) goto end;
-  
-  qsort((uchar*) dynamic_element(&saved_table_locks,0,TABLE_LOCK_INFO *),saved_table_locks.elements,sizeof(TABLE_LOCK_INFO),(qsort_cmp) dl_compare);
+
+  if (!saved_table_locks.elements)
+    goto end;
+
+  saved_base= dynamic_element(&saved_table_locks, 0, TABLE_LOCK_INFO *);
+  my_qsort(saved_base, saved_table_locks.elements, sizeof(TABLE_LOCK_INFO),
+           dl_compare);
   freeze_size(&saved_table_locks);
 
   puts("\nThread database.table_name          Locked/Waiting        Lock_type\n");
-  
+
   unsigned int i;
   for (i=0 ; i < saved_table_locks.elements ; i++)
   {
diff --git a/sql/sql_test.h b/sql/sql_test.h
index 539e89ec949..d7fcc126cb2 100644
--- a/sql/sql_test.h
+++ b/sql/sql_test.h
@@ -26,8 +26,8 @@ typedef struct st_sort_field SORT_FIELD;
 
 #ifndef DBUG_OFF
 void print_where(COND *cond,const char *info, enum_query_type query_type);
-void print_cached_tables(void);
 void TEST_filesort(SORT_FIELD *sortorder,uint s_length);
+void TEST_join(JOIN *join);
 void print_plan(JOIN* join,uint idx, double record_count, double read_time,
                 double current_read_time, const char *info);
 void dump_TABLE_LIST_graph(SELECT_LEX *select_lex, TABLE_LIST* tl);
diff --git a/sql/sql_trigger.cc b/sql/sql_trigger.cc
index 701a2ec93c2..9ce62d9f2a4 100644
--- a/sql/sql_trigger.cc
+++ b/sql/sql_trigger.cc
@@ -411,6 +411,13 @@ bool mysql_create_or_drop_trigger(THD *thd, TABLE_LIST *tables, bool create)
       destructive changes necessary to open the trigger's table.
     */
     thd->lex->reset_n_backup_query_tables_list(&backup);
+    /*
+      Restore Query_tables_list::sql_command, which was
+      reset above, as the code that writes the query to the
+      binary log assumes that this value corresponds to the
+      statement that is being executed.
+    */
+    thd->lex->sql_command= backup.sql_command;
 
     if (add_table_for_trigger(thd, thd->lex->spname, if_exists, & tables))
       goto end;
diff --git a/sql/sql_update.cc b/sql/sql_update.cc
index fe6e0994959..e7cc133e01b 100644
--- a/sql/sql_update.cc
+++ b/sql/sql_update.cc
@@ -1056,7 +1056,7 @@ int mysql_multi_update_prepare(THD *thd)
         be write-locked (for example, trigger to be invoked might try
         to update this table).
       */
-      tl->lock_type= read_lock_type_for_table(thd, table);
+      tl->lock_type= read_lock_type_for_table(thd, lex, tl);
       tl->updating= 0;
       /* Update TABLE::lock_type accordingly. */
       if (!tl->placeholder() && !using_lock_tables)
diff --git a/sql/sql_yacc.yy b/sql/sql_yacc.yy
index f9478b8c5f5..9f20a4ccf71 100644
--- a/sql/sql_yacc.yy
+++ b/sql/sql_yacc.yy
@@ -29,6 +29,7 @@
 #define YYLEX_PARAM yythd
 #define YYTHD ((THD *)yythd)
 #define YYLIP (& YYTHD->m_parser_state->m_lip)
+#define YYPS (& YYTHD->m_parser_state->m_yacc)
 
 #define MYSQL_YACC
 #define YYINITDEPTH 100
@@ -4937,7 +4938,6 @@ create_select:
           SELECT_SYM
           {
             LEX *lex=Lex;
-            lex->lock_option= TL_READ_DEFAULT;
             if (lex->sql_command == SQLCOM_INSERT)
               lex->sql_command= SQLCOM_INSERT_SELECT;
             else if (lex->sql_command == SQLCOM_REPLACE)
@@ -7302,7 +7302,6 @@ select_lock_type:
           {
             LEX *lex=Lex;
             lex->current_select->set_lock_for_tables(TL_WRITE);
-            lex->current_select->lock_option= TL_WRITE;
             lex->safe_to_cache_query=0;
             lex->protect_against_global_read_lock= TRUE;
           }
@@ -7311,7 +7310,6 @@ select_lock_type:
             LEX *lex=Lex;
             lex->current_select->
               set_lock_for_tables(TL_READ_WITH_SHARED_LOCKS);
-            lex->current_select->lock_option= TL_READ_WITH_SHARED_LOCKS;
             lex->safe_to_cache_query=0;
           }
         ;
@@ -9221,7 +9219,7 @@ table_factor:
           {
             if (!($$= Select->add_table_to_list(YYTHD, $2, $3,
                                                 Select->get_table_join_options(),
-                                                Lex->lock_option,
+                                                YYPS->m_lock_type,
                                                 Select->pop_index_hints())))
               MYSQL_YYABORT;
             Select->add_joined_table($$);
@@ -10278,7 +10276,7 @@ table_alias_ref:
           {
             if (!Select->add_table_to_list(YYTHD, $1, NULL,
                                            TL_OPTION_UPDATING | TL_OPTION_ALIAS,
-                                           Lex->lock_option ))
+                                           YYPS->m_lock_type))
               MYSQL_YYABORT;
           }
         ;
@@ -10303,8 +10301,6 @@ insert:
             lex->sql_command= SQLCOM_INSERT;
             lex->duplicates= DUP_ERROR; 
             mysql_init_select(lex);
-            /* for subselects */
-            lex->lock_option= TL_READ_DEFAULT;
           }
           insert_lock_option
           opt_ignore insert2
@@ -10495,7 +10491,6 @@ update:
             LEX *lex= Lex;
             mysql_init_select(lex);
             lex->sql_command= SQLCOM_UPDATE;
-            lex->lock_option= TL_UNLOCK; /* Will be set later */
             lex->duplicates= DUP_ERROR; 
           }
           opt_low_priority opt_ignore join_table_list
@@ -10562,7 +10557,7 @@ delete:
             LEX *lex= Lex;
             lex->sql_command= SQLCOM_DELETE;
             mysql_init_select(lex);
-            lex->lock_option= TL_WRITE_DEFAULT;
+            YYPS->m_lock_type= TL_WRITE_DEFAULT;
             lex->ignore= 0;
             lex->select_lex.init_order();
           }
@@ -10573,20 +10568,27 @@ single_multi:
           FROM table_ident
           {
             if (!Select->add_table_to_list(YYTHD, $2, NULL, TL_OPTION_UPDATING,
-                                           Lex->lock_option))
+                                           YYPS->m_lock_type))
               MYSQL_YYABORT;
+            YYPS->m_lock_type= TL_READ_DEFAULT;
           }
           where_clause opt_order_clause
           delete_limit_clause {}
         | table_wild_list
-          { mysql_init_multi_delete(Lex); }
+          {
+            mysql_init_multi_delete(Lex);
+            YYPS->m_lock_type= TL_READ_DEFAULT;
+          }
           FROM join_table_list where_clause
           {
             if (multi_delete_set_locks_and_link_aux_tables(Lex))
               MYSQL_YYABORT;
           }
         | FROM table_alias_ref_list
-          { mysql_init_multi_delete(Lex); }
+          {
+            mysql_init_multi_delete(Lex);
+            YYPS->m_lock_type= TL_READ_DEFAULT;
+          }
           USING join_table_list where_clause
           {
             if (multi_delete_set_locks_and_link_aux_tables(Lex))
@@ -10609,7 +10611,7 @@ table_wild_one:
                                            ti,
                                            NULL,
                                            TL_OPTION_UPDATING | TL_OPTION_ALIAS,
-                                           Lex->lock_option))
+                                           YYPS->m_lock_type))
               MYSQL_YYABORT;
           }
         | ident '.' ident opt_wild
@@ -10621,7 +10623,7 @@ table_wild_one:
                                            ti,
                                            NULL,
                                            TL_OPTION_UPDATING | TL_OPTION_ALIAS,
-                                           Lex->lock_option))
+                                           YYPS->m_lock_type))
               MYSQL_YYABORT;
           }
         ;
@@ -10638,7 +10640,7 @@ opt_delete_options:
 
 opt_delete_option:
           QUICK        { Select->options|= OPTION_QUICK; }
-        | LOW_PRIORITY { Lex->lock_option= TL_WRITE_LOW_PRIORITY; }
+        | LOW_PRIORITY { YYPS->m_lock_type= TL_WRITE_LOW_PRIORITY; }
         | IGNORE_SYM   { Lex->ignore= 1; }
         ;
 
@@ -10724,7 +10726,6 @@ show:
           {
             LEX *lex=Lex;
             lex->wild=0;
-            lex->lock_option= TL_READ;
             mysql_init_select(lex);
             lex->current_select->parsing_place= SELECT_LIST;
             bzero((char*) &lex->create_info,sizeof(lex->create_info));
@@ -11077,7 +11078,6 @@ describe:
           describe_command table_ident
           {
             LEX *lex= Lex;
-            lex->lock_option= TL_READ;
             mysql_init_select(lex);
             lex->current_select->parsing_place= SELECT_LIST;
             lex->sql_command= SQLCOM_SHOW_FIELDS;
@@ -11291,7 +11291,6 @@ load:
           {
             LEX *lex=Lex;
             lex->sql_command= SQLCOM_LOAD;
-            lex->lock_option= $4;
             lex->local_file=  $5;
             lex->duplicates= DUP_ERROR;
             lex->ignore= 0;
@@ -11302,7 +11301,7 @@ load:
           {
             LEX *lex=Lex;
             if (!Select->add_table_to_list(YYTHD, $12, NULL, TL_OPTION_UPDATING,
-                                           lex->lock_option))
+                                           $4))
               MYSQL_YYABORT;
             lex->field_list.empty();
             lex->update_list.empty();
@@ -13734,17 +13733,6 @@ subselect_start:
 subselect_end:
           {
             LEX *lex=Lex;
-            /*
-              Set the required lock level for the tables associated with the
-              current sub-select. This will overwrite previous lock options set
-              using st_select_lex::add_table_to_list in any of the following
-              rules: single_multi, table_wild_one, load_data, table_alias_ref,
-              table_factor.
-              The default lock level is TL_READ_DEFAULT but it can be modified
-              with query options specific for a certain (sub-)SELECT.
-            */
-            lex->current_select->
-              set_lock_for_tables(lex->current_select->lock_option);
 
             lex->pop_context();
             SELECT_LEX *child= lex->current_select;
@@ -13776,8 +13764,8 @@ query_expression_option:
           {
             if (check_simple_select())
               MYSQL_YYABORT;
-            Lex->lock_option= TL_READ_HIGH_PRIORITY;
-            Lex->current_select->lock_option= TL_READ_HIGH_PRIORITY;
+            YYPS->m_lock_type= TL_READ_HIGH_PRIORITY;
+            Select->options|= SELECT_HIGH_PRIORITY;
           }
         | DISTINCT         { Select->options|= SELECT_DISTINCT; }
         | SQL_SMALL_RESULT { Select->options|= SELECT_SMALL_RESULT; }
diff --git a/sql/sys_vars.cc b/sql/sys_vars.cc
index 5f75dd4adf5..34a9401a41e 100644
--- a/sql/sys_vars.cc
+++ b/sql/sys_vars.cc
@@ -295,7 +295,7 @@ static bool binlog_format_check(sys_var *self, THD *thd, set_var *var)
   /*
     Make the session variable 'binlog_format' read-only inside a transaction.
   */
-  if (thd->active_transaction())
+  if (thd->in_active_multi_stmt_transaction())
   {
     my_error(ER_INSIDE_TRANSACTION_PREVENTS_SWITCH_BINLOG_FORMAT, MYF(0));
     return true;
@@ -348,7 +348,7 @@ static bool binlog_direct_check(sys_var *self, THD *thd, set_var *var)
      Makes the session variable 'binlog_direct_non_transactional_updates'
      read-only inside a transaction.
    */
-   if (thd->active_transaction())
+   if (thd->in_active_multi_stmt_transaction())
    {
      my_error(ER_INSIDE_TRANSACTION_PREVENTS_SWITCH_BINLOG_DIRECT, MYF(0));
      return true;
@@ -1428,7 +1428,7 @@ static my_bool read_only;
 static bool check_read_only(sys_var *self, THD *thd, set_var *var)
 {
   /* Prevent self dead-lock */
-  if (thd->locked_tables_mode || thd->active_transaction())
+  if (thd->locked_tables_mode || thd->in_active_multi_stmt_transaction())
   {
     my_error(ER_LOCK_OR_ACTIVE_TRANSACTION, MYF(0));
     return true;
@@ -2013,15 +2013,20 @@ static Sys_var_ulong Sys_thread_pool_size(
        VALID_RANGE(1, 16384), DEFAULT(20), BLOCK_SIZE(0));
 #endif
 
-//  Can't change the 'next' tx_isolation if we are already in a transaction
+/**
+  Can't change the 'next' tx_isolation if we are already in a
+  transaction.
+*/
+
 static bool check_tx_isolation(sys_var *self, THD *thd, set_var *var)
 {
-  if (var->type == OPT_DEFAULT && (thd->server_status & SERVER_STATUS_IN_TRANS))
+  if (var->type == OPT_DEFAULT && thd->in_active_multi_stmt_transaction())
   {
+    DBUG_ASSERT(thd->in_multi_stmt_transaction_mode());
     my_error(ER_CANT_CHANGE_TX_ISOLATION, MYF(0));
-    return true;
+    return TRUE;
   }
-  return false;
+  return FALSE;
 }
 
 /*
@@ -2034,6 +2039,7 @@ static bool fix_tx_isolation(sys_var *self, THD *thd, enum_var_type type)
     thd->session_tx_isolation= (enum_tx_isolation)thd->variables.tx_isolation;
   return false;
 }
+
 // NO_CMD_LINE - different name of the option
 static Sys_var_enum Sys_tx_isolation(
        "tx_isolation", "Default transaction isolation level",
@@ -2237,17 +2243,69 @@ static Sys_var_bit Sys_log_off(
        SESSION_VAR(option_bits), NO_CMD_LINE, OPTION_LOG_OFF,
        DEFAULT(FALSE), NO_MUTEX_GUARD, NOT_IN_BINLOG, ON_CHECK(check_has_super));
 
-static bool fix_sql_log_bin(sys_var *self, THD *thd, enum_var_type type)
+/**
+  This function sets the session variable thd->variables.sql_log_bin 
+  to reflect changes to @@session.sql_log_bin.
+
+  @param[IN] self   A pointer to the sys_var, i.e. Sys_log_binlog.
+  @param[IN] type   The type either session or global.
+
+  @return @c FALSE.
+*/
+static bool fix_sql_log_bin_after_update(sys_var *self, THD *thd,
+                                         enum_var_type type)
 {
-  if (type != OPT_GLOBAL && !thd->in_sub_stmt)
-    thd->sql_log_bin_toplevel= thd->variables.option_bits & OPTION_BIN_LOG;
-  return false;
+  if (type == OPT_SESSION)
+  {
+    if (thd->variables.sql_log_bin)
+      thd->variables.option_bits |= OPTION_BIN_LOG;
+    else
+      thd->variables.option_bits &= ~OPTION_BIN_LOG;
+  }
+  return FALSE;
 }
-static Sys_var_bit Sys_log_binlog(
+
+/**
+  This function checks if the sql_log_bin can be changed,
+  what is possible if:
+    - the user is a super user;
+    - the set is not called from within a function/trigger;
+    - there is no on-going transaction.
+
+  @param[IN] self   A pointer to the sys_var, i.e. Sys_log_binlog.
+  @param[IN] var    A pointer to the set_var created by the parser.
+
+  @return @c FALSE if the change is allowed, otherwise @c TRUE.
+*/
+static bool check_sql_log_bin(sys_var *self, THD *thd, set_var *var)
+{
+  if (check_has_super(self, thd, var))
+    return TRUE;
+
+  if (var->type == OPT_GLOBAL)
+    return FALSE;
+
+  /* If in a stored function/trigger, it's too late to change sql_log_bin. */
+  if (thd->in_sub_stmt)
+  {
+    my_error(ER_STORED_FUNCTION_PREVENTS_SWITCH_SQL_LOG_BIN, MYF(0));
+    return TRUE;
+  }
+  /* Make the session variable 'sql_log_bin' read-only inside a transaction. */
+  if (thd->in_active_multi_stmt_transaction())
+  {
+    my_error(ER_INSIDE_TRANSACTION_PREVENTS_SWITCH_SQL_LOG_BIN, MYF(0));
+    return TRUE;
+  }
+
+  return FALSE;
+}
+
+static Sys_var_mybool Sys_log_binlog(
        "sql_log_bin", "sql_log_bin",
-       SESSION_VAR(option_bits), NO_CMD_LINE, OPTION_BIN_LOG,
-       DEFAULT(TRUE), NO_MUTEX_GUARD, NOT_IN_BINLOG, ON_CHECK(check_has_super),
-       ON_UPDATE(fix_sql_log_bin));
+       SESSION_VAR(sql_log_bin), NO_CMD_LINE,
+       DEFAULT(TRUE), NO_MUTEX_GUARD, NOT_IN_BINLOG, ON_CHECK(check_sql_log_bin),
+       ON_UPDATE(fix_sql_log_bin_after_update));
 
 static Sys_var_bit Sys_sql_warnings(
        "sql_warnings", "sql_warnings",
@@ -2746,8 +2804,8 @@ static Sys_var_mybool Sys_log_slow(
 static bool fix_log_state(sys_var *self, THD *thd, enum_var_type type)
 {
   bool res;
-  my_bool *newvalptr, newval, oldval;
-  uint log_type;
+  my_bool *UNINIT_VAR(newvalptr), newval, UNINIT_VAR(oldval);
+  uint UNINIT_VAR(log_type);
 
   if (self == &Sys_general_log || self == &Sys_log)
   {
diff --git a/sql/table.cc b/sql/table.cc
index 12ac8b11f97..6c0a4a4c674 100644
--- a/sql/table.cc
+++ b/sql/table.cc
@@ -976,28 +976,28 @@ static int open_binary_frm(THD *thd, TABLE_SHARE *share, uchar *head,
     }
     if (next_chunk + 5 < buff_end)
     {
-      uint32 partition_info_len = uint4korr(next_chunk);
+      uint32 partition_info_str_len = uint4korr(next_chunk);
 #ifdef WITH_PARTITION_STORAGE_ENGINE
       if ((share->partition_info_buffer_size=
-             share->partition_info_len= partition_info_len))
+             share->partition_info_str_len= partition_info_str_len))
       {
-        if (!(share->partition_info= (char*)
+        if (!(share->partition_info_str= (char*)
               memdup_root(&share->mem_root, next_chunk + 4,
-                          partition_info_len + 1)))
+                          partition_info_str_len + 1)))
         {
           my_free(buff, MYF(0));
           goto err;
         }
       }
 #else
-      if (partition_info_len)
+      if (partition_info_str_len)
       {
         DBUG_PRINT("info", ("WITH_PARTITION_STORAGE_ENGINE is not defined"));
         my_free(buff, MYF(0));
         goto err;
       }
 #endif
-      next_chunk+= 5 + partition_info_len;
+      next_chunk+= 5 + partition_info_str_len;
     }
 #if MYSQL_VERSION_ID < 50200
     if (share->mysql_version >= 50106 && share->mysql_version <= 50109)
@@ -1660,6 +1660,10 @@ static int open_binary_frm(THD *thd, TABLE_SHARE *share, uchar *head,
   my_hash_free(&share->name_hash);
   if (share->ha_data_destroy)
     share->ha_data_destroy(share->ha_data);
+#ifdef WITH_PARTITION_STORAGE_ENGINE
+  if (share->ha_part_data_destroy)
+    share->ha_part_data_destroy(share->ha_part_data);
+#endif
 
   open_table_error(share, error, share->open_errno, errarg);
   DBUG_RETURN(error);
@@ -1851,7 +1855,7 @@ int open_table_from_share(THD *thd, TABLE_SHARE *share, const char *alias,
   }
 
 #ifdef WITH_PARTITION_STORAGE_ENGINE
-  if (share->partition_info_len && outparam->file)
+  if (share->partition_info_str_len && outparam->file)
   {
   /*
     In this execution we must avoid calling thd->change_item_tree since
@@ -1872,10 +1876,8 @@ int open_table_from_share(THD *thd, TABLE_SHARE *share, const char *alias,
     bool tmp;
     bool work_part_info_used;
 
-    tmp= mysql_unpack_partition(thd, share->partition_info,
-                                share->partition_info_len,
-                                share->part_state,
-                                share->part_state_len,
+    tmp= mysql_unpack_partition(thd, share->partition_info_str,
+                                share->partition_info_str_len,
                                 outparam, is_create_table,
                                 share->default_part_db_type,
                                 &work_part_info_used);
diff --git a/sql/table.h b/sql/table.h
index 0725d62b286..34f71d6bba4 100644
--- a/sql/table.h
+++ b/sql/table.h
@@ -488,6 +488,19 @@ typedef struct st_table_field_def
 } TABLE_FIELD_DEF;
 
 
+#ifdef WITH_PARTITION_STORAGE_ENGINE
+/**
+  Partition specific ha_data struct.
+*/
+typedef struct st_ha_data_partition
+{
+  bool auto_inc_initialized;
+  mysql_mutex_t LOCK_auto_inc;                 /**< protecting auto_inc val */
+  ulonglong next_auto_inc_val;                 /**< first non reserved value */
+} HA_DATA_PARTITION;
+#endif
+
+
 class Table_check_intact
 {
 protected:
@@ -627,13 +640,11 @@ struct TABLE_SHARE
   int cached_row_logging_check;
 
 #ifdef WITH_PARTITION_STORAGE_ENGINE
-  /** @todo: Move into *ha_data for partitioning */
+  /* filled in when reading from frm */
   bool auto_partitioned;
-  const char *partition_info;
-  uint  partition_info_len;
+  const char *partition_info_str;
+  uint  partition_info_str_len;
   uint  partition_info_buffer_size;
-  const char *part_state;
-  uint part_state_len;
   handlerton *default_part_db_type;
 #endif
 
@@ -653,6 +664,14 @@ struct TABLE_SHARE
   void *ha_data;
   void (*ha_data_destroy)(void *); /* An optional destructor for ha_data */
 
+#ifdef WITH_PARTITION_STORAGE_ENGINE
+  /** place to store partition specific data, LOCK_ha_data hold while init. */
+  HA_DATA_PARTITION *ha_part_data;
+  /* Destructor for ha_part_data */
+  void (*ha_part_data_destroy)(HA_DATA_PARTITION *);
+#endif
+
+
   /** Instrumentation for this table share. */
   PSI_table_share *m_psi;
 
diff --git a/sql/thr_malloc.cc b/sql/thr_malloc.cc
index 638f3bbb9f1..7696f28081d 100644
--- a/sql/thr_malloc.cc
+++ b/sql/thr_malloc.cc
@@ -109,10 +109,6 @@ void* sql_memdup(const void *ptr, size_t len)
   return pos;
 }
 
-void sql_element_free(void *ptr __attribute__((unused)))
-{} /* purecov: deadcode */
-
-
 
 char *sql_strmake_with_convert(const char *str, size_t arg_length,
 			       CHARSET_INFO *from_cs,
diff --git a/sql/thr_malloc.h b/sql/thr_malloc.h
index a655884b8b4..6b372a285a2 100644
--- a/sql/thr_malloc.h
+++ b/sql/thr_malloc.h
@@ -27,7 +27,6 @@ void *sql_calloc(size_t);
 char *sql_strdup(const char *str);
 char *sql_strmake(const char *str, size_t len);
 void *sql_memdup(const void * ptr, size_t size);
-void sql_element_free(void *ptr);
 char *sql_strmake_with_convert(const char *str, size_t arg_length,
 			       CHARSET_INFO *from_cs,
 			       size_t max_res_length,
diff --git a/sql/transaction.cc b/sql/transaction.cc
index ff4eabc2b0f..5047de1ccdc 100644
--- a/sql/transaction.cc
+++ b/sql/transaction.cc
@@ -169,7 +169,7 @@ bool trans_commit_implicit(THD *thd)
   if (trans_check(thd))
     DBUG_RETURN(TRUE);
 
-  if (thd->in_multi_stmt_transaction() ||
+  if (thd->in_multi_stmt_transaction_mode() ||
       (thd->variables.option_bits & OPTION_TABLE_LOCK))
   {
     /* Safety if one did "drop table" on locked tables */
@@ -305,7 +305,7 @@ bool trans_savepoint(THD *thd, LEX_STRING name)
   SAVEPOINT **sv, *newsv;
   DBUG_ENTER("trans_savepoint");
 
-  if (!(thd->in_multi_stmt_transaction() || thd->in_sub_stmt) ||
+  if (!(thd->in_multi_stmt_transaction_mode() || thd->in_sub_stmt) ||
       !opt_using_transactions)
     DBUG_RETURN(FALSE);
 
@@ -467,7 +467,7 @@ bool trans_xa_start(THD *thd)
     my_error(ER_XAER_INVAL, MYF(0));
   else if (xa_state != XA_NOTR)
     my_error(ER_XAER_RMFAIL, MYF(0), xa_state_names[xa_state]);
-  else if (thd->locked_tables_mode || thd->active_transaction())
+  else if (thd->locked_tables_mode || thd->in_active_multi_stmt_transaction())
     my_error(ER_XAER_OUTSIDE, MYF(0));
   else if (xid_cache_search(thd->lex->xid))
     my_error(ER_XAER_DUPID, MYF(0));