Merge branch '10.2' into 10.3

61 files changed, 992 insertions, 629 deletions

diff --git a/sql/contributors.h b/sql/contributors.h
index a0d05af3fa6..69f8fa6bd4c 100644
--- a/sql/contributors.h
+++ b/sql/contributors.h
@@ -45,12 +45,14 @@ struct show_table_contributors_st show_table_contributors[]= {
   {"Visma", "https://visma.com", "Gold Sponsor of the MariaDB Foundation"},
   {"DBS", "https://dbs.com", "Gold Sponsor of the MariaDB Foundation"},
   {"IBM", "https://www.ibm.com", "Gold Sponsor of the MariaDB Foundation"},
+  {"Tencent Games", "http://game.qq.com/", "Gold Sponsor of the MariaDB Foundation"},
   {"Nexedi", "https://www.nexedi.com", "Silver Sponsor of the MariaDB Foundation"},
-  {"Acronis", "http://www.acronis.com", "Silver Sponsor of the MariaDB Foundation"},
+  {"Acronis", "https://www.acronis.com", "Silver Sponsor of the MariaDB Foundation"},
   {"Verkkokauppa.com", "https://www.verkkokauppa.com", "Bronze Sponsor of the MariaDB Foundation"},
   {"Virtuozzo", "https://virtuozzo.com", "Bronze Sponsor of the MariaDB Foundation"},
   {"Tencent Game DBA", "http://tencentdba.com/about", "Bronze Sponsor of the MariaDB Foundation"},
   {"Tencent TDSQL", "http://tdsql.org", "Bronze Sponsor of the MariaDB Foundation"},
+  {"Percona", "https://www.percona.com/", "Bronze Sponsor of the MariaDB Foundation"},
 
   /* Sponsors of important features */
   {"Google", "USA", "Sponsoring encryption, parallel replication and GTID"},
diff --git a/sql/events.cc b/sql/events.cc
index fcad115c199..5723cf2eaa4 100644
--- a/sql/events.cc
+++ b/sql/events.cc
@@ -463,6 +463,7 @@ Events::update_event(THD *thd, Event_parse_data *parse_data,
 
   if (check_access(thd, EVENT_ACL, parse_data->dbname.str, NULL, NULL, 0, 0))
     DBUG_RETURN(TRUE);
+
   WSREP_TO_ISOLATION_BEGIN(WSREP_MYSQL_DB, NULL, NULL);
 
   if (lock_object_name(thd, MDL_key::EVENT,
@@ -593,6 +594,7 @@ Events::drop_event(THD *thd, const LEX_CSTRING *dbname,
 
   if (check_access(thd, EVENT_ACL, dbname->str, NULL, NULL, 0, 0))
     DBUG_RETURN(TRUE);
+
   WSREP_TO_ISOLATION_BEGIN(WSREP_MYSQL_DB, NULL, NULL);
 
   /*
diff --git a/sql/field.cc b/sql/field.cc
index 7eb277e23d8..4ba31e17fa4 100644
--- a/sql/field.cc
+++ b/sql/field.cc
@@ -3075,6 +3075,12 @@ Field *Field_decimal::make_new_field(MEM_ROOT *root, TABLE *new_table,
 ** Field_new_decimal
 ****************************************************************************/
 
+static uint get_decimal_precision(uint len, uint8 dec, bool unsigned_val)
+{
+  uint precision= my_decimal_length_to_precision(len, dec, unsigned_val);
+  return MY_MIN(precision, DECIMAL_MAX_PRECISION);
+}
+
 Field_new_decimal::Field_new_decimal(uchar *ptr_arg,
                                      uint32 len_arg, uchar *null_ptr_arg,
                                      uchar null_bit_arg,
@@ -3085,8 +3091,7 @@ Field_new_decimal::Field_new_decimal(uchar *ptr_arg,
   :Field_num(ptr_arg, len_arg, null_ptr_arg, null_bit_arg,
              unireg_check_arg, field_name_arg, dec_arg, zero_arg, unsigned_arg)
 {
-  precision= my_decimal_length_to_precision(len_arg, dec_arg, unsigned_arg);
-  set_if_smaller(precision, DECIMAL_MAX_PRECISION);
+  precision= get_decimal_precision(len_arg, dec_arg, unsigned_arg);
   DBUG_ASSERT((precision <= DECIMAL_MAX_PRECISION) &&
               (dec <= DECIMAL_MAX_SCALE));
   bin_size= my_decimal_get_binary_size(precision, dec);
@@ -4528,7 +4533,7 @@ longlong Field_float::val_int(void)
 {
   float j;
   float4get(j,ptr);
-  return (longlong) rint(j);
+  return Converter_double_to_longlong(j, false).result();
 }
 
 
@@ -10289,12 +10294,9 @@ void Column_definition::create_length_to_internal_length_bit()
 
 void Column_definition::create_length_to_internal_length_newdecimal()
 {
-  key_length= pack_length=
-    my_decimal_get_binary_size(my_decimal_length_to_precision((uint) length,
-                                                              decimals,
-                                                              flags &
-                                                              UNSIGNED_FLAG),
-                               decimals);
+  DBUG_ASSERT(length < UINT_MAX32);
+  uint prec= get_decimal_precision((uint)length, decimals, flags & UNSIGNED_FLAG);
+  key_length= pack_length= my_decimal_get_binary_size(prec, decimals);
 }
 
 
diff --git a/sql/gcalc_slicescan.cc b/sql/gcalc_slicescan.cc
index ce1f4394ebd..9127bb95aeb 100644
--- a/sql/gcalc_slicescan.cc
+++ b/sql/gcalc_slicescan.cc
@@ -177,6 +177,17 @@ Gcalc_dyn_list::Gcalc_dyn_list(size_t blk_size, size_t sizeof_item):
 {}
 
 
+Gcalc_dyn_list::Gcalc_dyn_list(const Gcalc_dyn_list &dl)
+{
+  m_blk_size= dl.m_blk_size;
+  m_sizeof_item= dl.m_sizeof_item;
+  m_points_per_blk= dl.m_points_per_blk;
+  m_blk_hook= &m_first_blk;
+  m_free= NULL;
+  m_keep= NULL;
+}
+
+
 void Gcalc_dyn_list::format_blk(void* block)
 {
   Item *pi_end, *cur_pi, *first_pi;
diff --git a/sql/gcalc_slicescan.h b/sql/gcalc_slicescan.h
index b9516fc8d8c..ebf173c1a57 100644
--- a/sql/gcalc_slicescan.h
+++ b/sql/gcalc_slicescan.h
@@ -63,6 +63,7 @@ public:
   };
 
   Gcalc_dyn_list(size_t blk_size, size_t sizeof_item);
+  Gcalc_dyn_list(const Gcalc_dyn_list &dl);
   ~Gcalc_dyn_list();
   Item *new_item()
   {
@@ -229,6 +230,12 @@ public:
     Gcalc_dyn_list(blk_size, sizeof(Info)),
     m_hook(&m_first), m_n_points(0)
   {}
+
+  Gcalc_heap(const Gcalc_heap &gh) :
+    Gcalc_dyn_list(gh),
+    m_hook(&m_first), m_n_points(0)
+  {}
+
   void set_extent(double xmin, double xmax, double ymin, double ymax);
   Info *new_point_info(double x, double y, gcalc_shape_info shape);
   void free_point_info(Info *i, Gcalc_dyn_list::Item **i_hook);
diff --git a/sql/gcalc_tools.cc b/sql/gcalc_tools.cc
index 2aeaafc1a76..832a52db522 100644
--- a/sql/gcalc_tools.cc
+++ b/sql/gcalc_tools.cc
@@ -663,6 +663,17 @@ Gcalc_operation_reducer::Gcalc_operation_reducer(size_t blk_size) :
 {}
 
 
+Gcalc_operation_reducer::Gcalc_operation_reducer(
+                           const Gcalc_operation_reducer &gor) :
+  Gcalc_dyn_list(gor),
+#ifndef GCALC_DBUG_OFF
+  n_res_points(0),
+#endif /*GCALC_DBUG_OFF*/
+  m_res_hook((Gcalc_dyn_list::Item **)&m_result),
+  m_first_active_thread(NULL)
+{}
+
+
 void Gcalc_operation_reducer::init(Gcalc_function *fn, modes mode)
 {
   m_fn= fn;
diff --git a/sql/gcalc_tools.h b/sql/gcalc_tools.h
index 8bda3c144a6..4d5aec0d443 100644
--- a/sql/gcalc_tools.h
+++ b/sql/gcalc_tools.h
@@ -224,6 +224,7 @@ public:
   };
 
   Gcalc_operation_reducer(size_t blk_size=8192);
+  Gcalc_operation_reducer(const Gcalc_operation_reducer &gor);
   void init(Gcalc_function *fn, modes mode= default_mode);
   Gcalc_operation_reducer(Gcalc_function *fn, modes mode= default_mode,
 		       size_t blk_size=8192);
diff --git a/sql/ha_partition.cc b/sql/ha_partition.cc
index 6f647eb6cfd..40380ffeec5 100644
--- a/sql/ha_partition.cc
+++ b/sql/ha_partition.cc
@@ -10893,11 +10893,9 @@ int ha_partition::check_for_upgrade(HA_CHECK_OPT *check_opt)
           }
           m_part_info->key_algorithm= partition_info::KEY_ALGORITHM_51;
           if (skip_generation ||
-              !(part_buf= generate_partition_syntax(thd, m_part_info,
-                                                    &part_buf_len,
-                                                    true,
-                                                    NULL,
-                                                    NULL)) ||
+              !(part_buf= generate_partition_syntax_for_frm(thd, m_part_info,
+                                                            &part_buf_len,
+                                                            NULL, NULL)) ||
 	      print_admin_msg(thd, SQL_ADMIN_MSG_TEXT_SIZE + 1, "error",
 	                      table_share->db.str,
 	                      table->alias,
diff --git a/sql/handler.cc b/sql/handler.cc
index 306b0868d15..58253b90594 100644
--- a/sql/handler.cc
+++ b/sql/handler.cc
@@ -21,6 +21,7 @@
 */
 
 #include "mariadb.h"
+#include <inttypes.h>
 #include "sql_priv.h"
 #include "unireg.h"
 #include "rpl_rli.h"
@@ -4618,6 +4619,7 @@ handler::ha_create_partitioning_metadata(const char *name,
               (!old_name && strcmp(name, table_share->path.str)));
 
 
+  mark_trx_read_write();
   return create_partitioning_metadata(name, old_name, action_flag);
 }
 
@@ -6546,7 +6548,7 @@ void ha_fake_trx_id(THD *thd)
 
   if (thd->wsrep_ws_handle.trx_id != WSREP_UNDEFINED_TRX_ID)
   {
-    WSREP_DEBUG("fake trx id skipped: %lu", thd->wsrep_ws_handle.trx_id);
+    WSREP_DEBUG("fake trx id skipped: %" PRIu64, thd->wsrep_ws_handle.trx_id);
     DBUG_VOID_RETURN;
   }
 
diff --git a/sql/item.cc b/sql/item.cc
index 8fa3a0b741b..e72c2b80288 100644
--- a/sql/item.cc
+++ b/sql/item.cc
@@ -1647,7 +1647,14 @@ Query_fragment::Query_fragment(THD *thd, sp_head *sphead,
                                const char *start, const char *end)
 {
   DBUG_ASSERT(start <= end);
-  if (sphead)
+  if (thd->lex->clone_spec_offset)
+  {
+    Lex_input_stream *lip= (& thd->m_parser_state->m_lip);
+    DBUG_ASSERT(lip->get_buf() <= start);
+    DBUG_ASSERT(end <= lip->get_end_of_query());
+    set(start - lip->get_buf(), end - start);
+  }
+  else if (sphead)
   {
     if (sphead->m_tmp_query)
     {
@@ -4062,7 +4069,8 @@ Item_param::Item_param(THD *thd, const LEX_CSTRING *name_arg,
     For dynamic SQL, settability depends on the type of Item passed
     as an actual parameter. See Item_param::set_from_item().
   */
-  m_is_settable_routine_parameter(true)
+  m_is_settable_routine_parameter(true),
+  m_clones(thd->mem_root)
 {
   name= *name_arg;
   /*
@@ -4074,6 +4082,60 @@ Item_param::Item_param(THD *thd, const LEX_CSTRING *name_arg,
 }
 
 
+/* Add reference to Item_param used in a copy of CTE to its master as a clone */
+
+bool Item_param::add_as_clone(THD *thd)
+{
+  LEX *lex= thd->lex;
+  my_ptrdiff_t master_pos= pos_in_query + lex->clone_spec_offset;
+  List_iterator_fast<Item_param> it(lex->param_list);
+  Item_param *master_param;
+  while ((master_param = it++))
+  {
+    if (master_pos == master_param->pos_in_query)
+      return master_param->register_clone(this);
+  }
+  DBUG_ASSERT(false);
+  return false;
+}
+
+
+/* Update all clones of Item_param to sync their values with the item's value */
+
+void Item_param::sync_clones()
+{
+  Item_param **c_ptr= m_clones.begin();
+  Item_param **end= m_clones.end();
+  for ( ; c_ptr < end; c_ptr++)
+  {
+    Item_param *c= *c_ptr;
+    /* Scalar-type members: */
+    c->maybe_null= maybe_null;
+    c->null_value= null_value;
+    c->Type_std_attributes::operator=(*this);
+    c->Type_handler_hybrid_field_type::operator=(*this);
+    c->Type_geometry_attributes::operator=(*this);
+
+    c->state= state;
+    c->item_type= item_type;
+    c->m_empty_string_is_null= m_empty_string_is_null;
+
+    c->value.PValue_simple::operator=(value);
+    c->value.Type_handler_hybrid_field_type::operator=(value);
+    type_handler()->Item_param_setup_conversion(current_thd, c);
+
+    /* Class-type members: */
+    c->value.m_decimal= value.m_decimal;
+    /*
+      Note that String's assignment op properly sets m_is_alloced to 'false',
+      which is correct here: c->str_value doesn't own anything.
+    */
+    c->value.m_string= value.m_string;
+    c->value.m_string_ptr= value.m_string_ptr;
+  }
+}
+
+
 void Item_param::set_null()
 {
   DBUG_ENTER("Item_param::set_null");
@@ -4530,7 +4592,7 @@ longlong Item_param::PValue::val_int(const Type_std_attributes *attr) const
 {
   switch (type_handler()->cmp_type()) {
   case REAL_RESULT:
-    return (longlong) rint(real);
+    return Converter_double_to_longlong(real, attr->unsigned_flag).result();
   case INT_RESULT:
     return integer;
   case DECIMAL_RESULT:
@@ -5040,32 +5102,6 @@ bool Item_param::append_for_log(THD *thd, String *str)
   return str->append(*val);
 }
 
-/****************************************************************************
-  Item_copy
-****************************************************************************/
-
-Item_copy *Item_copy::create(THD *thd, Item *item)
-{
-  MEM_ROOT *mem_root= thd->mem_root;
-  switch (item->result_type())
-  {
-    case STRING_RESULT:
-      return new (mem_root) Item_copy_string(thd, item);
-    case REAL_RESULT:
-      return new (mem_root) Item_copy_float(thd, item);
-    case INT_RESULT:
-      return item->unsigned_flag ?
-        new (mem_root) Item_copy_uint(thd, item) :
-        new (mem_root) Item_copy_int(thd, item);
-    case DECIMAL_RESULT:
-      return new (mem_root) Item_copy_decimal(thd, item);
-    case TIME_RESULT:
-    case ROW_RESULT:
-      DBUG_ASSERT (0);
-  }
-  /* should not happen */
-  return NULL;
-}
 
 /****************************************************************************
   Item_copy_string
@@ -5123,156 +5159,6 @@ my_decimal *Item_copy_string::val_decimal(my_decimal *decimal_value)
 }
 
 
-/****************************************************************************
-  Item_copy_int
-****************************************************************************/
-
-void Item_copy_int::copy()
-{
-  cached_value= item->val_int();
-  null_value=item->null_value;
-}
-
-static int save_int_value_in_field (Field *, longlong, bool, bool);
-
-int Item_copy_int::save_in_field(Field *field, bool no_conversions)
-{
-  return save_int_value_in_field(field, cached_value, 
-                                 null_value, unsigned_flag);
-}
-
-
-String *Item_copy_int::val_str(String *str)
-{
-  if (null_value)
-    return (String *) 0;
-
-  str->set(cached_value, &my_charset_bin);
-  return str;
-}
-
-
-my_decimal *Item_copy_int::val_decimal(my_decimal *decimal_value)
-{
-  if (null_value)
-    return (my_decimal *) 0;
-
-  int2my_decimal(E_DEC_FATAL_ERROR, cached_value, unsigned_flag, decimal_value);
-  return decimal_value;
-}
-
-
-/****************************************************************************
-  Item_copy_uint
-****************************************************************************/
-
-String *Item_copy_uint::val_str(String *str)
-{
-  if (null_value)
-    return (String *) 0;
-
-  str->set((ulonglong) cached_value, &my_charset_bin);
-  return str;
-}
-
-
-/****************************************************************************
-  Item_copy_float
-****************************************************************************/
-
-String *Item_copy_float::val_str(String *str)
-{
-  if (null_value)
-    return (String *) 0;
-  else
-  {
-    double nr= val_real();
-    str->set_real(nr,decimals, &my_charset_bin);
-    return str;
-  }
-}
-
-
-my_decimal *Item_copy_float::val_decimal(my_decimal *decimal_value)
-{
-  if (null_value)
-    return (my_decimal *) 0;
-  else
-  {
-    double nr= val_real();
-    double2my_decimal(E_DEC_FATAL_ERROR, nr, decimal_value);
-    return decimal_value;
-  }
-}
-
-
-int Item_copy_float::save_in_field(Field *field, bool no_conversions)
-{
-  if (null_value)
-    return set_field_to_null(field);
-  field->set_notnull();
-  return field->store(cached_value);
-}
-
-
-/****************************************************************************
-  Item_copy_decimal
-****************************************************************************/
-
-int Item_copy_decimal::save_in_field(Field *field, bool no_conversions)
-{
-  if (null_value)
-    return set_field_to_null(field);
-  field->set_notnull();
-  return field->store_decimal(&cached_value);
-}
-
-
-String *Item_copy_decimal::val_str(String *result)
-{
-  if (null_value)
-    return (String *) 0;
-  result->set_charset(&my_charset_bin);
-  my_decimal2string(E_DEC_FATAL_ERROR, &cached_value, 0, 0, 0, result);
-  return result;
-}
-
-
-double Item_copy_decimal::val_real()
-{
-  if (null_value)
-    return 0.0;
-  else
-  {
-    double result;
-    my_decimal2double(E_DEC_FATAL_ERROR, &cached_value, &result);
-    return result;
-  }
-}
-
-
-longlong Item_copy_decimal::val_int()
-{
-  if (null_value)
-    return 0;
-  else
-  {
-    longlong result;
-    my_decimal2int(E_DEC_FATAL_ERROR, &cached_value, unsigned_flag, &result);
-    return result;
-  }
-}
-
-
-void Item_copy_decimal::copy()
-{
-  my_decimal *nr= item->val_decimal(&cached_value);
-  if (nr && nr != &cached_value)
-    my_decimal2decimal (nr, &cached_value);
-  null_value= item->null_value;
-}
-
-
 /*
   Functions to convert item to field (for send_result_set_metadata)
 */
@@ -9744,13 +9630,11 @@ void Item_trigger_field::cleanup()
 
 Item_result item_cmp_type(Item_result a,Item_result b)
 {
-  if (a == STRING_RESULT && b == STRING_RESULT)
-    return STRING_RESULT;
-  if (a == INT_RESULT && b == INT_RESULT)
-    return INT_RESULT;
-  else if (a == ROW_RESULT || b == ROW_RESULT)
+  if (a == b)
+    return a;
+  if (a == ROW_RESULT || b == ROW_RESULT)
     return ROW_RESULT;
-  else if (a == TIME_RESULT || b == TIME_RESULT)
+  if (a == TIME_RESULT || b == TIME_RESULT)
     return TIME_RESULT;
   if ((a == INT_RESULT || a == DECIMAL_RESULT) &&
       (b == INT_RESULT || b == DECIMAL_RESULT))
@@ -10196,7 +10080,7 @@ longlong Item_cache_real::val_int()
   DBUG_ASSERT(fixed == 1);
   if (!has_value())
     return 0;
-  return (longlong) rint(value);
+  return Converter_double_to_longlong(value, unsigned_flag).result();
 }
 
 
diff --git a/sql/item.h b/sql/item.h
index 48b060095af..6823d58a94c 100644
--- a/sql/item.h
+++ b/sql/item.h
@@ -28,6 +28,7 @@
 #include "field.h"                              /* Derivation */
 #include "sql_type.h"
 #include "sql_time.h"
+#include "mem_root_array.h"
 
 C_MODE_START
 #include <ma_dyncol.h>
@@ -3614,6 +3615,10 @@ public:
   bool check_vcol_func_processor(void *int_arg) {return FALSE;}
   Item *get_copy(THD *thd) { return 0; }
 
+  bool add_as_clone(THD *thd);
+  void sync_clones();
+  bool register_clone(Item_param *i) { return m_clones.push_back(i); }
+
 private:
   void invalid_default_param() const;
 
@@ -3631,6 +3636,12 @@ public:
 private:
   Send_field *m_out_param_info;
   bool m_is_settable_routine_parameter;
+  /*
+    Array of all references of this parameter marker used in a CTE to its clones
+    created for copies of this marker used the CTE's copies. It's used to
+    synchronize the actual value of the parameter with the values of the clones.
+  */
+  Mem_root_array<Item_param *, true> m_clones;
 };
 
 
@@ -5462,7 +5473,7 @@ public:
   Base class to implement typed value caching Item classes
 
   Item_copy_ classes are very similar to the corresponding Item_
-  classes (e.g. Item_copy_int is similar to Item_int) but they add
+  classes (e.g. Item_copy_string is similar to Item_string) but they add
   the following additional functionality to Item_ :
     1. Nullability
     2. Possibility to store the value not only on instantiation time,
@@ -5509,13 +5520,6 @@ protected:
   }
 
 public:
-  /** 
-    Factory method to create the appropriate subclass dependent on the type of 
-    the original item.
-
-    @param item      the original item.
-  */  
-  static Item_copy *create(THD *thd, Item *item);
 
   /** 
     Update the cache with the value of the original item
@@ -5583,107 +5587,6 @@ public:
 };
 
 
-class Item_copy_int : public Item_copy
-{
-protected:  
-  longlong cached_value; 
-public:
-  Item_copy_int(THD *thd, Item *i): Item_copy(thd, i) {}
-  int save_in_field(Field *field, bool no_conversions);
-
-  virtual String *val_str(String*);
-  virtual my_decimal *val_decimal(my_decimal *);
-  virtual double val_real()
-  {
-    return null_value ? 0.0 : (double) cached_value;
-  }
-  virtual longlong val_int()
-  {
-    return null_value ? 0 : cached_value;
-  }
-  bool get_date(MYSQL_TIME *ltime, ulonglong fuzzydate)
-  { return get_date_from_int(ltime, fuzzydate); }
-  virtual void copy();
-  Item *get_copy(THD *thd)
-  { return get_item_copy<Item_copy_int>(thd, this); }
-};
-
-
-class Item_copy_uint : public Item_copy_int
-{
-public:
-  Item_copy_uint(THD *thd, Item *item_arg): Item_copy_int(thd, item_arg)
-  {
-    unsigned_flag= 1;
-  }
-
-  String *val_str(String*);
-  double val_real()
-  {
-    return null_value ? 0.0 : (double) (ulonglong) cached_value;
-  }
-  Item *get_copy(THD *thd)
-  { return get_item_copy<Item_copy_uint>(thd, this); }
-};
-
-
-class Item_copy_float : public Item_copy
-{
-protected:  
-  double cached_value; 
-public:
-  Item_copy_float(THD *thd, Item *i): Item_copy(thd, i) {}
-  int save_in_field(Field *field, bool no_conversions);
-
-  String *val_str(String*);
-  my_decimal *val_decimal(my_decimal *);
-  double val_real()
-  {
-    return null_value ? 0.0 : cached_value;
-  }
-  longlong val_int()
-  {
-    return (longlong) rint(val_real());
-  }
-  bool get_date(MYSQL_TIME *ltime, ulonglong fuzzydate)
-  {
-    return get_date_from_real(ltime, fuzzydate);
-  }
-  void copy()
-  {
-    cached_value= item->val_real();
-    null_value= item->null_value;
-  }
-  Item *get_copy(THD *thd)
-  { return get_item_copy<Item_copy_float>(thd, this); }
-};
-
-
-class Item_copy_decimal : public Item_copy
-{
-protected:  
-  my_decimal cached_value;
-public:
-  Item_copy_decimal(THD *thd, Item *i): Item_copy(thd, i) {}
-  int save_in_field(Field *field, bool no_conversions);
-
-  String *val_str(String*);
-  my_decimal *val_decimal(my_decimal *) 
-  { 
-    return null_value ? NULL: &cached_value; 
-  }
-  double val_real();
-  longlong val_int();
-  bool get_date(MYSQL_TIME *ltime, ulonglong fuzzydate)
-  {
-    return get_date_from_decimal(ltime, fuzzydate);
-  }
-  void copy();
-  Item *get_copy(THD *thd)
-  { return get_item_copy<Item_copy_decimal>(thd, this); }
-};
-
-
 /*
   Cached_item_XXX objects are not exactly caches. They do the following:
 
diff --git a/sql/item_func.cc b/sql/item_func.cc
index 5693484ad3b..3dcfe4fbce8 100644
--- a/sql/item_func.cc
+++ b/sql/item_func.cc
@@ -902,7 +902,7 @@ String *Item_func_hybrid_field_type::val_str_from_real_op(String *str)
 
 longlong Item_func_hybrid_field_type::val_int_from_real_op()
 {
-  return (longlong) rint(real_op());
+  return Converter_double_to_longlong(real_op(), unsigned_flag).result();
 }
 
 my_decimal *
diff --git a/sql/item_func.h b/sql/item_func.h
index af29e14fe3a..1d19c7a43f1 100644
--- a/sql/item_func.h
+++ b/sql/item_func.h
@@ -393,7 +393,10 @@ public:
   String *val_str(String*str);
   my_decimal *val_decimal(my_decimal *decimal_value);
   longlong val_int()
-    { DBUG_ASSERT(fixed == 1); return (longlong) rint(val_real()); }
+  {
+    DBUG_ASSERT(fixed == 1);
+    return Converter_double_to_longlong(val_real(), unsigned_flag).result();
+  }
   bool get_date(MYSQL_TIME *ltime, ulonglong fuzzydate)
   { return get_date_from_real(ltime, fuzzydate); }
   const Type_handler *type_handler() const { return &type_handler_double; }
@@ -2096,7 +2099,8 @@ class Item_func_udf_float :public Item_udf_func
   longlong val_int()
   {
     DBUG_ASSERT(fixed == 1);
-    return (longlong) rint(Item_func_udf_float::val_real());
+    return Converter_double_to_longlong(Item_func_udf_float::val_real(),
+                                        unsigned_flag).result();
   }
   my_decimal *val_decimal(my_decimal *dec_buf)
   {
diff --git a/sql/item_jsonfunc.cc b/sql/item_jsonfunc.cc
index 2cda8b25a8a..230d954aa77 100644
--- a/sql/item_jsonfunc.cc
+++ b/sql/item_jsonfunc.cc
@@ -1622,13 +1622,15 @@ String *Item_func_json_array_append::val_str(String *str)
 
     if (je.value_type == JSON_VALUE_ARRAY)
     {
-      if (json_skip_level(&je))
+      int n_items;
+      if (json_skip_level_and_count(&je, &n_items))
         goto js_error;
 
       ar_end= je.s.c_str - je.sav_c_len;
       str_rest_len= js->length() - (ar_end - (const uchar *) js->ptr());
       str->q_append(js->ptr(), ar_end-(const uchar *) js->ptr());
-      str->append(", ", 2);
+      if (n_items)
+        str->append(", ", 2);
       if (append_json_value(str, args[n_arg+1], &tmp_val))
         goto return_null; /* Out of memory. */
 
@@ -2019,23 +2021,14 @@ continue_j2:
   else
   {
     const uchar *end1, *beg1, *end2, *beg2;
-    int empty_array= 0;
+    int n_items1=1, n_items2= 1;
 
     beg1= je1->value_begin;
 
     /* Merge as a single array. */
     if (je1->value_type == JSON_VALUE_ARRAY)
     {
-      int cur_level= je1->stack_p;
-      empty_array= 1;
-      while (json_scan_next(je1) == 0)
-      {
-        if (je1->stack_p < cur_level)
-          break;
-        empty_array= 0;
-      }
-
-      if (unlikely(je1->s.error))
+      if (json_skip_level_and_count(je1, &n_items1))
         return 1;
 
       end1= je1->s.c_str - je1->sav_c_len;
@@ -2054,8 +2047,7 @@ continue_j2:
         end1= je1->value_end;
     }
 
-    if (str->append((const char*) beg1, end1 - beg1) ||
-        (!empty_array && str->append(", ", 2)))
+    if (str->append((const char*) beg1, end1 - beg1))
       return 3;
 
     if (json_value_scalar(je2))
@@ -2066,15 +2058,22 @@ continue_j2:
     else
     {
       if (je2->value_type == JSON_VALUE_OBJECT)
+      {
         beg2= je2->value_begin;
+        if (json_skip_level(je2))
+          return 2;
+      }
       else
+      {
         beg2= je2->s.c_str;
-      if (json_skip_level(je2))
-        return 2;
+        if (json_skip_level_and_count(je2, &n_items2))
+          return 2;
+      }
       end2= je2->s.c_str;
     }
 
-    if (str->append((const char*) beg2, end2 - beg2))
+    if ((n_items1 && n_items2 && str->append(", ", 2)) ||
+        str->append((const char*) beg2, end2 - beg2))
       return 3;
 
     if (je2->value_type != JSON_VALUE_ARRAY &&
@@ -3057,7 +3056,7 @@ static int append_json_path(String *str, const json_path_t *p)
 String *Item_func_json_search::val_str(String *str)
 {
   String *js= args[0]->val_json(&tmp_js);
-  String *s_str= args[2]->val_str(&tmp_js);
+  String *s_str= args[2]->val_str(&tmp_path);
   json_engine_t je;
   json_path_t p, sav_path;
   uint n_arg;
@@ -3306,5 +3305,3 @@ int Arg_comparator::compare_e_json_str_basic(Item *j, Item *s)
 
   return MY_TEST(sortcmp(res1, res2, compare_collation()) == 0);
 }
-
-
diff --git a/sql/item_jsonfunc.h b/sql/item_jsonfunc.h
index af0995b9605..a4705f012f2 100644
--- a/sql/item_jsonfunc.h
+++ b/sql/item_jsonfunc.h
@@ -407,7 +407,7 @@ public:
 class Item_func_json_search: public Item_json_str_multipath
 {
 protected:
-  String tmp_js, esc_value;
+  String tmp_js, tmp_path, esc_value;
   bool mode_one;
   bool ooa_constant, ooa_parsed;
   int escape;
diff --git a/sql/item_timefunc.cc b/sql/item_timefunc.cc
index d29181a3446..4d17bc354d4 100644
--- a/sql/item_timefunc.cc
+++ b/sql/item_timefunc.cc
@@ -41,6 +41,7 @@
 #include "set_var.h"
 #include "sql_locale.h"          // MY_LOCALE my_locale_en_US
 #include "strfunc.h"             // check_word
+#include "sql_type_int.h"        // Longlong_hybrid
 #include "sql_time.h"            // make_truncated_value_warning,
                                  // get_date_from_daynr,
                                  // calc_weekday, calc_week,
@@ -2860,8 +2861,7 @@ bool Item_func_timediff::get_date(MYSQL_TIME *ltime, ulonglong fuzzy_date)
 bool Item_func_maketime::get_date(MYSQL_TIME *ltime, ulonglong fuzzy_date)
 {
   DBUG_ASSERT(fixed == 1);
-  bool overflow= 0;
-  longlong hour=   args[0]->val_int();
+  Longlong_hybrid hour(args[0]->val_int(), args[0]->unsigned_flag);
   longlong minute= args[1]->val_int();
   ulonglong second;
   ulong microsecond;
@@ -2873,32 +2873,23 @@ bool Item_func_maketime::get_date(MYSQL_TIME *ltime, ulonglong fuzzy_date)
 
   bzero(ltime, sizeof(*ltime));
   ltime->time_type= MYSQL_TIMESTAMP_TIME;
+  ltime->neg= hour.neg();
 
-  /* Check for integer overflows */
-  if (hour < 0)
+  if (hour.abs() <= TIME_MAX_HOUR)
   {
-    if (args[0]->unsigned_flag)
-      overflow= 1;
-    else
-      ltime->neg= 1;
-  }
-  if (-hour > TIME_MAX_HOUR || hour > TIME_MAX_HOUR)
-    overflow= 1;
-
-  if (!overflow)
-  {
-    ltime->hour=   (uint) ((hour < 0 ? -hour : hour));
+    ltime->hour=   (uint) hour.abs();
     ltime->minute= (uint) minute;
     ltime->second= (uint) second;
     ltime->second_part= microsecond;
   }
   else
   {
-    ltime->hour= TIME_MAX_HOUR;
-    ltime->minute= TIME_MAX_MINUTE;
-    ltime->second= TIME_MAX_SECOND;
+    // use check_time_range() to set ltime to the max value depending on dec
+    int unused;
+    ltime->hour= TIME_MAX_HOUR + 1;
+    check_time_range(ltime, decimals, &unused);
     char buf[28];
-    char *ptr= longlong10_to_str(hour, buf, args[0]->unsigned_flag ? 10 : -10);
+    char *ptr= longlong10_to_str(hour.value(), buf, hour.is_unsigned() ? 10 : -10);
     int len = (int)(ptr - buf) + sprintf(ptr, ":%02u:%02u", (uint)minute, (uint)second);
     make_truncated_value_warning(current_thd, Sql_condition::WARN_LEVEL_WARN,
                                  buf, len, MYSQL_TIMESTAMP_TIME,
diff --git a/sql/log.cc b/sql/log.cc
index 1ffa6f7588f..0fa65a05de3 100644
--- a/sql/log.cc
+++ b/sql/log.cc
@@ -1703,14 +1703,14 @@ static int binlog_close_connection(handlerton *hton, THD *thd)
     uchar *buf;
     size_t len=0;
     wsrep_write_cache_buf(cache, &buf, &len);
-    WSREP_WARN("binlog trx cache not empty (%lu bytes) @ connection close %lld",
-               (ulong) len, (longlong) thd->thread_id);
+    WSREP_WARN("binlog trx cache not empty (%zu bytes) @ connection close %lld",
+               len, (longlong) thd->thread_id);
     if (len > 0) wsrep_dump_rbr_buf(thd, buf, len);
 
     cache = cache_mngr->get_binlog_cache_log(false);
     wsrep_write_cache_buf(cache, &buf, &len);
-    WSREP_WARN("binlog stmt cache not empty (%lu bytes) @ connection close %lld",
-               (ulong) len, (longlong) thd->thread_id);
+    WSREP_WARN("binlog stmt cache not empty (%zu bytes) @ connection close %lld",
+               len, (longlong) thd->thread_id);
     if (len > 0) wsrep_dump_rbr_buf(thd, buf, len);
   }
 #endif /* WITH_WSREP */
diff --git a/sql/log_event.cc b/sql/log_event.cc
index c550adea26b..e10480fb015 100644
--- a/sql/log_event.cc
+++ b/sql/log_event.cc
@@ -9172,11 +9172,6 @@ User_var_log_event(const char* buf, uint event_len,
       we keep the flags set to UNDEF_F.
     */
     size_t bytes_read= (val + val_len) - buf_start;
-    if (bytes_read > size_t(event_len))
-    {
-      error= true;
-      goto err;
-    }
     if ((data_written - bytes_read) > 0)
     {
       flags= (uint) *(buf + UV_VAL_IS_NULL + UV_VAL_TYPE_SIZE +
diff --git a/sql/mysqld.cc b/sql/mysqld.cc
index f2b00831d6f..e4866f516ff 100644
--- a/sql/mysqld.cc
+++ b/sql/mysqld.cc
@@ -799,6 +799,7 @@ char *master_info_file;
 char *relay_log_info_file, *report_user, *report_password, *report_host;
 char *opt_relay_logname = 0, *opt_relaylog_index_name=0;
 char *opt_logname, *opt_slow_logname, *opt_bin_logname;
+char *opt_binlog_index_name=0;
 
 /* Static variables */
 
@@ -808,7 +809,6 @@ my_bool opt_expect_abort= 0, opt_bootstrap= 0;
 static my_bool opt_myisam_log;
 static int cleanup_done;
 static ulong opt_specialflag;
-static char *opt_binlog_index_name;
 char *mysql_home_ptr, *pidfile_name_ptr;
 /** Initial command line arguments (count), after load_defaults().*/
 static int defaults_argc;
diff --git a/sql/mysqld.h b/sql/mysqld.h
index 6c0c7308d91..d5cabd790b2 100644
--- a/sql/mysqld.h
+++ b/sql/mysqld.h
@@ -175,6 +175,7 @@ extern uint protocol_version, mysqld_port, dropping_tables;
 extern ulong delay_key_write_options;
 extern char *opt_logname, *opt_slow_logname, *opt_bin_logname, 
             *opt_relay_logname;
+extern char *opt_binlog_index_name;
 extern char *opt_backup_history_logname, *opt_backup_progress_logname,
             *opt_backup_settings_name;
 extern const char *log_output_str;
diff --git a/sql/opt_range.cc b/sql/opt_range.cc
index 0b29fcd2f11..636100f9c4e 100644
--- a/sql/opt_range.cc
+++ b/sql/opt_range.cc
@@ -2734,13 +2734,19 @@ bool create_key_parts_for_pseudo_indexes(RANGE_OPT_PARAM *param,
 
   for (field_ptr= table->field; *field_ptr; field_ptr++)
   {
-    if (bitmap_is_set(used_fields, (*field_ptr)->field_index))
+    Column_statistics* col_stats= (*field_ptr)->read_stats;
+    if (bitmap_is_set(used_fields, (*field_ptr)->field_index)
+       && col_stats && !col_stats->no_stat_values_provided()
+       && !((*field_ptr)->type() == MYSQL_TYPE_GEOMETRY))
       parts++;
   }
 
   KEY_PART *key_part;
   uint keys= 0;
 
+  if (!parts)
+    return TRUE;
+
   if (!(key_part= (KEY_PART *)  alloc_root(param->mem_root,
                                            sizeof(KEY_PART) * parts)))
     return TRUE;
@@ -2752,6 +2758,9 @@ bool create_key_parts_for_pseudo_indexes(RANGE_OPT_PARAM *param,
     if (bitmap_is_set(used_fields, (*field_ptr)->field_index))
     {
       Field *field= *field_ptr;
+      if (field->type() == MYSQL_TYPE_GEOMETRY)
+        continue;
+
       uint16 store_length;
       uint16 max_key_part_length= (uint16) table->file->max_key_part_length();
       key_part->key= keys;
@@ -2909,7 +2918,18 @@ bool calculate_cond_selectivity_for_table(THD *thd, TABLE *table, Item **cond)
 
   table->cond_selectivity= 1.0;
 
-  if (!*cond || table_records == 0)
+  if (table_records == 0)
+    DBUG_RETURN(FALSE);
+
+  QUICK_SELECT_I *quick;
+  if ((quick=table->reginfo.join_tab->quick) &&
+      quick->get_type() == QUICK_SELECT_I::QS_TYPE_GROUP_MIN_MAX)
+  {
+    table->cond_selectivity*= (quick->records/table_records);
+    DBUG_RETURN(FALSE);
+  }
+
+  if (!*cond)
     DBUG_RETURN(FALSE);
 
   if (table->pos_in_table_list->schema_table)
@@ -3026,7 +3046,8 @@ bool calculate_cond_selectivity_for_table(THD *thd, TABLE *table, Item **cond)
   */
 
   if (thd->variables.optimizer_use_condition_selectivity > 2 &&
-      !bitmap_is_clear_all(used_fields))
+      !bitmap_is_clear_all(used_fields) &&
+      thd->variables.use_stat_tables > 0)
   {
     PARAM param;
     MEM_ROOT alloc;
diff --git a/sql/sql_acl.cc b/sql/sql_acl.cc
index 88d1630d94b..5b38420223e 100644
--- a/sql/sql_acl.cc
+++ b/sql/sql_acl.cc
@@ -10989,9 +10989,8 @@ bool sp_grant_privileges(THD *thd, const char *sp_db, const char *sp_name,
   combo->user.str= (char *) sctx->priv_user;
 
   mysql_mutex_lock(&acl_cache->lock);
-
- if ((au= find_user_wild(combo->host.str= (char *) sctx->priv_host,
-                         combo->user.str)))
+  if ((au= find_user_exact(combo->host.str= (char *) sctx->priv_host,
+                           combo->user.str)))
     goto found_acl;
 
   mysql_mutex_unlock(&acl_cache->lock);
diff --git a/sql/sql_alter.cc b/sql/sql_alter.cc
index a96b0a5a32b..d2893b5fbff 100644
--- a/sql/sql_alter.cc
+++ b/sql/sql_alter.cc
@@ -476,15 +476,17 @@ bool Sql_cmd_alter_table::execute(THD *thd)
   thd->work_part_info= 0;
 #endif
 
-#ifdef WITH_WSREP
-  if ((!thd->is_current_stmt_binlog_format_row() ||
+  if (WSREP(thd) &&
+      (!thd->is_current_stmt_binlog_format_row() ||
        !thd->find_temporary_table(first_table)))
   {
-    WSREP_TO_ISOLATION_BEGIN(((lex->name.str) ? select_lex->db.str : NULL),
-                             ((lex->name.str) ? lex->name.str : NULL),
-                             first_table);
+    WSREP_TO_ISOLATION_BEGIN_ALTER((lex->name.str ? select_lex->db.str : NULL),
+                                   (lex->name.str ? lex->name.str : NULL),
+                                   first_table, &alter_info);
+
+    thd->variables.auto_increment_offset = 1;
+    thd->variables.auto_increment_increment = 1;
   }
-#endif /* WITH_WSREP */
 
   result= mysql_alter_table(thd, &select_lex->db, &lex->name,
                             &create_info,
@@ -497,10 +499,12 @@ bool Sql_cmd_alter_table::execute(THD *thd)
   DBUG_RETURN(result);
 
 #ifdef WITH_WSREP
-error:
-  WSREP_WARN("ALTER TABLE isolation failure");
-  DBUG_RETURN(TRUE);
-#endif /* WITH_WSREP */
+error: /* Used by WSREP_TO_ISOLATION_BEGIN_ALTER */
+#endif
+  {
+    WSREP_WARN("ALTER TABLE isolation failure");
+    DBUG_RETURN(TRUE);
+  }
 }
 
 bool Sql_cmd_discard_import_tablespace::execute(THD *thd)
diff --git a/sql/sql_base.cc b/sql/sql_base.cc
index 34aef26ebc2..d9e68d919ef 100644
--- a/sql/sql_base.cc
+++ b/sql/sql_base.cc
@@ -414,9 +414,10 @@ bool close_cached_tables(THD *thd, TABLE_LIST *tables,
     for (TABLE_LIST *table_list= tables_to_reopen; table_list;
          table_list= table_list->next_global)
     {
+      int err;
       /* A check that the table was locked for write is done by the caller. */
       TABLE *table= find_table_for_mdl_upgrade(thd, table_list->db.str,
-                                               table_list->table_name.str, TRUE);
+                                            table_list->table_name.str, &err);
 
       /* May return NULL if this table has already been closed via an alias. */
       if (! table)
@@ -648,6 +649,7 @@ close_all_tables_for_name(THD *thd, TABLE_SHARE *share,
   size_t key_length= share->table_cache_key.length;
   const char *db= key;
   const char *table_name= db + share->db.length + 1;
+  bool remove_from_locked_tables= extra != HA_EXTRA_NOT_USED;
 
   memcpy(key, share->table_cache_key.str, key_length);
 
@@ -661,7 +663,7 @@ close_all_tables_for_name(THD *thd, TABLE_SHARE *share,
     {
       thd->locked_tables_list.unlink_from_list(thd,
                                                table->pos_in_locked_tables,
-                                               extra != HA_EXTRA_NOT_USED);
+                                               remove_from_locked_tables);
       /* Inform handler that there is a drop table or a rename going on */
       if (extra != HA_EXTRA_NOT_USED && table->db_stat)
       {
@@ -1499,6 +1501,65 @@ static int set_partitions_as_used(TABLE_LIST *tl, TABLE *t)
 
 
 /**
+  Check if the given table is actually a VIEW that was LOCK-ed
+
+  @param thd            Thread context.
+  @param t              Table to check.
+
+  @retval TRUE  The 't'-table is a locked view
+                needed to remedy problem before retrying again.
+  @retval FALSE 't' was not locked, not a VIEW or an error happened.
+*/
+bool is_locked_view(THD *thd, TABLE_LIST *t)
+{
+  DBUG_ENTER("check_locked_view");
+  /*
+   Is this table a view and not a base table?
+   (it is work around to allow to open view with locked tables,
+   real fix will be made after definition cache will be made)
+
+   Since opening of view which was not explicitly locked by LOCK
+   TABLES breaks metadata locking protocol (potentially can lead
+   to deadlocks) it should be disallowed.
+  */
+  if (thd->mdl_context.is_lock_owner(MDL_key::TABLE, t->db.str,
+                                     t->table_name.str, MDL_SHARED))
+  {
+    char path[FN_REFLEN + 1];
+    build_table_filename(path, sizeof(path) - 1,
+                         t->db.str, t->table_name.str, reg_ext, 0);
+    /*
+      Note that we can't be 100% sure that it is a view since it's
+      possible that we either simply have not found unused TABLE
+      instance in THD::open_tables list or were unable to open table
+      during prelocking process (in this case in theory we still
+      should hold shared metadata lock on it).
+    */
+    if (dd_frm_is_view(thd, path))
+    {
+      /*
+        If parent_l of the table_list is non null then a merge table
+        has this view as child table, which is not supported.
+      */
+      if (t->parent_l)
+      {
+        my_error(ER_WRONG_MRG_TABLE, MYF(0));
+        DBUG_RETURN(FALSE);
+      }
+
+      if (!tdc_open_view(thd, t, CHECK_METADATA_VERSION))
+      {
+        DBUG_ASSERT(t->view != 0);
+        DBUG_RETURN(TRUE); // VIEW
+      }
+    }
+  }
+
+  DBUG_RETURN(FALSE);
+}
+
+
+/**
   Open a base table.
 
   @param thd            Thread context.
@@ -1652,49 +1713,10 @@ bool open_table(THD *thd, TABLE_LIST *table_list, Open_table_context *ot_ctx)
 #endif
       goto reset;
     }
-    /*
-      Is this table a view and not a base table?
-      (it is work around to allow to open view with locked tables,
-      real fix will be made after definition cache will be made)
 
-      Since opening of view which was not explicitly locked by LOCK
-      TABLES breaks metadata locking protocol (potentially can lead
-      to deadlocks) it should be disallowed.
-    */
-    if (thd->mdl_context.is_lock_owner(MDL_key::TABLE,
-                                       table_list->db.str,
-                                       table_list->table_name.str,
-                                       MDL_SHARED))
-    {
-      char path[FN_REFLEN + 1];
-      build_table_filename(path, sizeof(path) - 1,
-                           table_list->db.str, table_list->table_name.str, reg_ext, 0);
-      /*
-        Note that we can't be 100% sure that it is a view since it's
-        possible that we either simply have not found unused TABLE
-        instance in THD::open_tables list or were unable to open table
-        during prelocking process (in this case in theory we still
-        should hold shared metadata lock on it).
-      */
-      if (dd_frm_is_view(thd, path))
-      {
-        /*
-          If parent_l of the table_list is non null then a merge table
-          has this view as child table, which is not supported.
-        */
-        if (table_list->parent_l)
-        {
-          my_error(ER_WRONG_MRG_TABLE, MYF(0));
-          DBUG_RETURN(true);
-        }
+    if (is_locked_view(thd, table_list))
+      DBUG_RETURN(FALSE); // VIEW
 
-        if (!tdc_open_view(thd, table_list, CHECK_METADATA_VERSION))
-        {
-          DBUG_ASSERT(table_list->view != 0);
-          DBUG_RETURN(FALSE); // VIEW
-        }
-      }
-    }
     /*
       No table in the locked tables list. In case of explicit LOCK TABLES
       this can happen if a user did not include the table into the list.
@@ -2064,8 +2086,9 @@ TABLE *find_locked_table(TABLE *list, const char *db, const char *table_name)
    @param thd        Thread context
    @param db         Database name.
    @param table_name Name of table.
-   @param no_error   Don't emit error if no suitable TABLE
-                     instance were found.
+   @param p_error    In the case of an error (when the function returns NULL)
+                     the error number is stored there.
+                     If the p_error is NULL, function launches the error itself.
 
    @note This function checks if the connection holds a global IX
          metadata lock. If no such lock is found, it is not safe to
@@ -2078,15 +2101,15 @@ TABLE *find_locked_table(TABLE *list, const char *db, const char *table_name)
 */
 
 TABLE *find_table_for_mdl_upgrade(THD *thd, const char *db,
-                                  const char *table_name, bool no_error)
+                                  const char *table_name, int *p_error)
 {
   TABLE *tab= find_locked_table(thd->open_tables, db, table_name);
+  int error;
 
   if (unlikely(!tab))
   {
-    if (!no_error)
-      my_error(ER_TABLE_NOT_LOCKED, MYF(0), table_name);
-    return NULL;
+    error= ER_TABLE_NOT_LOCKED;
+    goto err_exit;
   }
 
   /*
@@ -2098,9 +2121,8 @@ TABLE *find_table_for_mdl_upgrade(THD *thd, const char *db,
   if (unlikely(!thd->mdl_context.is_lock_owner(MDL_key::GLOBAL, "", "",
                                                MDL_INTENTION_EXCLUSIVE)))
   {
-    if (!no_error)
-      my_error(ER_TABLE_NOT_LOCKED_FOR_WRITE, MYF(0), table_name);
-    return NULL;
+    error= ER_TABLE_NOT_LOCKED_FOR_WRITE;
+    goto err_exit;
   }
 
   while (tab->mdl_ticket != NULL &&
@@ -2108,10 +2130,21 @@ TABLE *find_table_for_mdl_upgrade(THD *thd, const char *db,
          (tab= find_locked_table(tab->next, db, table_name)))
     continue;
 
-  if (unlikely(!tab && !no_error))
-    my_error(ER_TABLE_NOT_LOCKED_FOR_WRITE, MYF(0), table_name);
+  if (unlikely(!tab))
+  {
+    error= ER_TABLE_NOT_LOCKED_FOR_WRITE;
+    goto err_exit;
+  }
 
   return tab;
+
+err_exit:
+  if (p_error)
+    *p_error= error;
+  else
+    my_error(error, MYF(0), table_name);
+
+  return NULL;
 }
 
 
@@ -3399,6 +3432,15 @@ open_and_process_table(THD *thd, LEX *lex, TABLE_LIST *tables,
     */
     if (tables->with)
     {
+      if (tables->is_recursive_with_table() &&
+          !tables->is_with_table_recursive_reference())
+      {
+        tables->with->rec_outer_references++;
+        With_element *with_elem= tables->with;
+        while ((with_elem= with_elem->get_next_mutually_recursive()) !=
+               tables->with)
+	  with_elem->rec_outer_references++;
+      }
       if (tables->set_as_with_table(thd, tables->with))
         DBUG_RETURN(1);
       else
@@ -3813,10 +3855,10 @@ lock_table_names(THD *thd, const DDL_options_st &options,
     mdl_requests.push_front(&global_request);
 
     if (create_table)
-    #ifdef WITH_WSREP
-      if (thd->lex->sql_command != SQLCOM_CREATE_TABLE && 
+#ifdef WITH_WSREP
+      if (thd->lex->sql_command != SQLCOM_CREATE_TABLE &&
           thd->wsrep_exec_mode != REPL_RECV)
-    #endif
+#endif
       lock_wait_timeout= 0;                     // Don't wait for timeout
   }
 
@@ -3921,7 +3963,8 @@ open_tables_check_upgradable_mdl(THD *thd, TABLE_LIST *tables_start,
       Note that find_table_for_mdl_upgrade() will report an error if
       no suitable ticket is found.
     */
-    if (!find_table_for_mdl_upgrade(thd, table->db.str, table->table_name.str, false))
+    if (!find_table_for_mdl_upgrade(thd, table->db.str, table->table_name.str,
+                                    NULL))
       return TRUE;
   }
 
@@ -4318,9 +4361,8 @@ handle_routine(THD *thd, Query_tables_list *prelocking_ctx,
   @note this can be changed to use a hash, instead of scanning the linked
   list, if the performance of this function will ever become an issue
 */
-static bool table_already_fk_prelocked(TABLE_LIST *tl, LEX_CSTRING *db,
-                                       LEX_CSTRING *table,
-                                       thr_lock_type lock_type)
+bool table_already_fk_prelocked(TABLE_LIST *tl, LEX_CSTRING *db,
+                                LEX_CSTRING *table, thr_lock_type lock_type)
 {
   for (; tl; tl= tl->next_global )
   {
@@ -6932,10 +6974,22 @@ store_natural_using_join_columns(THD *thd, TABLE_LIST *natural_using_join,
 
   result= FALSE;
 
-err:
   if (arena)
     thd->restore_active_arena(arena, &backup);
   DBUG_RETURN(result);
+
+err:
+  /*
+     Actually we failed to build join columns list, so we have to
+     clear it to avoid problems with half-build join on next run.
+     The list was created in mark_common_columns().
+   */
+  table_ref_1->remove_join_columns();
+  table_ref_2->remove_join_columns();
+
+  if (arena)
+    thd->restore_active_arena(arena, &backup);
+  DBUG_RETURN(TRUE);
 }
 
 
diff --git a/sql/sql_base.h b/sql/sql_base.h
index be93566be09..22247af07a8 100644
--- a/sql/sql_base.h
+++ b/sql/sql_base.h
@@ -126,6 +126,7 @@ TABLE *open_ltable(THD *thd, TABLE_LIST *table_list, thr_lock_type update,
                             MYSQL_OPEN_GET_NEW_TABLE |\
                             MYSQL_OPEN_HAS_MDL_LOCK)
 
+bool is_locked_view(THD *thd, TABLE_LIST *t);
 bool open_table(THD *thd, TABLE_LIST *table_list, Open_table_context *ot_ctx);
 
 bool get_key_map_from_key_list(key_map *map, TABLE *table,
@@ -141,6 +142,8 @@ thr_lock_type read_lock_type_for_table(THD *thd,
 my_bool mysql_rm_tmp_tables(void);
 void close_tables_for_reopen(THD *thd, TABLE_LIST **tables,
                              const MDL_savepoint &start_of_statement_svp);
+bool table_already_fk_prelocked(TABLE_LIST *tl, LEX_CSTRING *db,
+                                LEX_CSTRING *table, thr_lock_type lock_type);
 TABLE_LIST *find_table_in_list(TABLE_LIST *table,
                                TABLE_LIST *TABLE_LIST::*link,
                                const LEX_CSTRING *db_name,
@@ -300,7 +303,8 @@ bool tdc_open_view(THD *thd, TABLE_LIST *table_list, uint flags);
 
 TABLE *find_table_for_mdl_upgrade(THD *thd, const char *db,
                                   const char *table_name,
-                                  bool no_error);
+                                  int *p_error);
+void mark_tmp_table_for_reuse(TABLE *table);
 
 int dynamic_column_error_message(enum_dyncol_func_result rc);
 
diff --git a/sql/sql_class.h b/sql/sql_class.h
index 00a5c41f708..97c5bfa3c12 100644
--- a/sql/sql_class.h
+++ b/sql/sql_class.h
@@ -5580,10 +5580,16 @@ class select_union_recursive :public select_unit
   TABLE *first_rec_table_to_update;
   /* The temporary tables used for recursive table references */
   List<TABLE> rec_tables;
+  /*
+    The count of how many times cleanup() was called with cleaned==false
+    for the unit specifying the recursive CTE for which this object was created
+    or for the unit specifying a CTE that mutually recursive with this CTE.
+  */
+  uint cleanup_count;
 
   select_union_recursive(THD *thd_arg):
     select_unit(thd_arg),
-    incr_table(0), first_rec_table_to_update(0) {};
+    incr_table(0), first_rec_table_to_update(0), cleanup_count(0) {};
 
   int send_data(List<Item> &items);
   bool create_result_table(THD *thd, List<Item> *column_types,
diff --git a/sql/sql_cte.cc b/sql/sql_cte.cc
index 97bbf8f73bd..b05a688f40e 100644
--- a/sql/sql_cte.cc
+++ b/sql/sql_cte.cc
@@ -743,9 +743,10 @@ bool With_clause::prepare_unreferenced_elements(THD *thd)
   @brief
     Save the specification of the given with table as a string
 
-  @param thd        The context of the statement containing this with element
-  @param spec_start The beginning of the specification in the input string
-  @param spec_end   The end of the specification in the input string
+  @param thd         The context of the statement containing this with element
+  @param spec_start  The beginning of the specification in the input string
+  @param spec_end    The end of the specification in the input string
+  @param spec_offset The offset of the specification in the input string
 
   @details
     The method creates for a string copy of the specification used in this
@@ -757,10 +758,17 @@ bool With_clause::prepare_unreferenced_elements(THD *thd)
     true    on failure
 */
   
-bool With_element::set_unparsed_spec(THD *thd, char *spec_start, char *spec_end)
+bool With_element::set_unparsed_spec(THD *thd, char *spec_start, char *spec_end,
+                                     my_ptrdiff_t spec_offset)
 {
+  stmt_prepare_mode= thd->m_parser_state->m_lip.stmt_prepare_mode;
   unparsed_spec.length= spec_end - spec_start;
-  unparsed_spec.str= thd->strmake(spec_start, unparsed_spec.length);
+
+  if (stmt_prepare_mode || !thd->lex->sphead)
+    unparsed_spec.str= spec_start;
+  else
+    unparsed_spec.str= thd->strmake(spec_start, unparsed_spec.length);
+  unparsed_spec_offset= spec_offset;
 
   if (!unparsed_spec.str)
   {
@@ -830,13 +838,28 @@ st_select_lex_unit *With_element::clone_parsed_spec(THD *thd,
   TABLE_LIST *spec_tables_tail;
   st_select_lex *with_select;
 
+  char save_end= unparsed_spec.str[unparsed_spec.length];
+  ((char*) &unparsed_spec.str[unparsed_spec.length])[0]= '\0';
   if (parser_state.init(thd, (char*) unparsed_spec.str, (unsigned int)unparsed_spec.length))
     goto err;
+  parser_state.m_lip.stmt_prepare_mode= stmt_prepare_mode;
+  parser_state.m_lip.multi_statements= false;
+  parser_state.m_lip.m_digest= NULL;
+
   lex_start(thd);
+  lex->clone_spec_offset= unparsed_spec_offset;
+  lex->param_list= old_lex->param_list;
+  lex->sphead= old_lex->sphead;
+  lex->spname= old_lex->spname;
+  lex->spcont= old_lex->spcont;
+  lex->sp_chistics= old_lex->sp_chistics;
+
   lex->stmt_lex= old_lex;
   with_select= &lex->select_lex;
   with_select->select_number= ++thd->lex->stmt_lex->current_select_number;
   parse_status= parse_sql(thd, &parser_state, 0);
+  ((char*) &unparsed_spec.str[unparsed_spec.length])[0]= save_end;
+
   if (parse_status)
     goto err;
 
@@ -881,6 +904,7 @@ st_select_lex_unit *With_element::clone_parsed_spec(THD *thd,
                         with_select));
   if (check_dependencies_in_with_clauses(lex->with_clauses_list))
     res= NULL;
+  lex->sphead= NULL;    // in order not to delete lex->sphead
   lex_end(lex);
 err:
   if (arena)
diff --git a/sql/sql_cte.h b/sql/sql_cte.h
index 16b473f0665..bda62271649 100644
--- a/sql/sql_cte.h
+++ b/sql/sql_cte.h
@@ -90,6 +90,11 @@ private:
     It used to build clones of the specification if they are needed.
   */
   LEX_CSTRING unparsed_spec;
+  /* Offset of the specification in the input string */
+  my_ptrdiff_t unparsed_spec_offset;
+
+  /* True if the with element is used a prepared statement */
+  bool stmt_prepare_mode;
 
   /* Return the map where 1 is set only in the position for this element */
   table_map get_elem_map() { return (table_map) 1 << number; }
@@ -114,7 +119,14 @@ public:
     for the definition of this element
   */
   bool is_recursive;
-
+  /*
+    For a simple recursive CTE: the number of references to the CTE from
+    outside of the CTE specification.
+    For a CTE mutually recursive with other CTEs : the total number of
+    references to all these CTEs outside of their specification.
+    Each of these mutually recursive CTEs has the same value in this field.
+  */
+  uint rec_outer_references;
   /*
     Any non-recursive select in the specification of a recursive
     with element is a called anchor. In the case mutually recursive
@@ -158,7 +170,7 @@ public:
       top_level_dep_map(0), sq_rec_ref(NULL),
       next_mutually_recursive(NULL), references(0), 
       query_name(name), column_list(list), spec(unit),
-      is_recursive(false), with_anchor(false),
+      is_recursive(false), rec_outer_references(0), with_anchor(false),
       level(0), rec_result(NULL)
   { unit->with_element= this; }
 
@@ -185,7 +197,8 @@ public:
 
   TABLE_LIST *find_first_sq_rec_ref_in_select(st_select_lex *sel);
 
-  bool set_unparsed_spec(THD *thd, char *spec_start, char *spec_end);
+  bool set_unparsed_spec(THD *thd, char *spec_start, char *spec_end,
+                         my_ptrdiff_t spec_offset);
 
   st_select_lex_unit *clone_parsed_spec(THD *thd, TABLE_LIST *with_table);
 
diff --git a/sql/sql_delete.cc b/sql/sql_delete.cc
index a040ba9eb93..fc400252397 100644
--- a/sql/sql_delete.cc
+++ b/sql/sql_delete.cc
@@ -1212,6 +1212,7 @@ multi_delete::~multi_delete()
   {
     TABLE *table= table_being_deleted->table;
     table->no_keyread=0;
+    table->no_cache= 0;
   }
 
   for (uint counter= 0; counter < num_of_tables; counter++)
diff --git a/sql/sql_derived.cc b/sql/sql_derived.cc
index 6a2e956fbf3..589d0214292 100644
--- a/sql/sql_derived.cc
+++ b/sql/sql_derived.cc
@@ -1106,6 +1106,7 @@ bool mysql_derived_fill(THD *thd, LEX *lex, TABLE_LIST *derived)
   DBUG_ASSERT(derived->table && derived->table->is_created());
   select_unit *derived_result= derived->derived_result;
   SELECT_LEX *save_current_select= lex->current_select;
+  bool derived_recursive_is_filled= false;
 
   if (unit->executed && !derived_is_recursive &&
       (unit->uncacheable & UNCACHEABLE_DEPENDENT))
@@ -1134,6 +1135,7 @@ bool mysql_derived_fill(THD *thd, LEX *lex, TABLE_LIST *derived)
     {
       /* In this case all iteration are performed */
       res= derived->fill_recursive(thd);
+      derived_recursive_is_filled= true;
     }
   }
   else if (unit->is_unit_op())
@@ -1189,7 +1191,8 @@ bool mysql_derived_fill(THD *thd, LEX *lex, TABLE_LIST *derived)
     }
   }
 err:
-  if (res || (!lex->describe && !derived_is_recursive && !unit->uncacheable))
+  if (res || (!lex->describe && !unit->uncacheable &&
+              (!derived_is_recursive || derived_recursive_is_filled)))
     unit->cleanup();
   lex->current_select= save_current_select;
 
@@ -1433,4 +1436,3 @@ bool pushdown_cond_for_derived(THD *thd, Item *cond, TABLE_LIST *derived)
   thd->lex->current_select= save_curr_select;
   DBUG_RETURN(false);
 }
-
diff --git a/sql/sql_error.h b/sql/sql_error.h
index 88355e529b5..822503f89d3 100644
--- a/sql/sql_error.h
+++ b/sql/sql_error.h
@@ -18,11 +18,11 @@
 #define SQL_ERROR_H
 
 #include "sql_list.h" 	/* Sql_alloc, MEM_ROOT, list */
-#include "m_string.h"		/* LEX_STRING */
-#include "sql_string.h"       /* String */
-#include "sql_plist.h"        /* I_P_List */
-#include "mysql_com.h"        /* MYSQL_ERRMSG_SIZE */
-#include "my_time.h"          /* MYSQL_TIME */
+#include "sql_type_int.h" // Longlong_hybrid
+#include "sql_string.h"                        /* String */
+#include "sql_plist.h" /* I_P_List */
+#include "mysql_com.h" /* MYSQL_ERRMSG_SIZE */
+#include "my_time.h"   /* MYSQL_TIME */
 #include "decimal.h"
 
 class THD;
@@ -843,13 +843,11 @@ public:
   }
 };
 
-class ErrConvInteger : public ErrConv
+class ErrConvInteger : public ErrConv, public Longlong_hybrid
 {
-  longlong m_value;
-  bool m_unsigned;
 public:
   ErrConvInteger(longlong num_arg, bool unsigned_flag= false) :
-    ErrConv(), m_value(num_arg), m_unsigned(unsigned_flag) {}
+    ErrConv(), Longlong_hybrid(num_arg, unsigned_flag) {}
   const char *ptr() const
   {
     return m_unsigned ? ullstr(m_value, err_buffer) :
diff --git a/sql/sql_lex.cc b/sql/sql_lex.cc
index 1b1031d440b..6f60b479153 100644
--- a/sql/sql_lex.cc
+++ b/sql/sql_lex.cc
@@ -669,6 +669,7 @@ void LEX::start(THD *thd_arg)
   curr_with_clause= 0;
   with_clauses_list= 0;
   with_clauses_list_last_next= &with_clauses_list;
+  clone_spec_offset= 0;
   create_view= NULL;
   value_list.empty();
   update_list.empty();
@@ -6660,6 +6661,14 @@ Item *LEX::make_item_sysvar(THD *thd,
 }
 
 
+static bool param_push_or_clone(THD *thd, LEX *lex, Item_param *item)
+{
+  return !lex->clone_spec_offset ?
+         lex->param_list.push_back(item, thd->mem_root) :
+         item->add_as_clone(thd);
+}
+
+
 Item_param *LEX::add_placeholder(THD *thd, const LEX_CSTRING *name,
                                  const char *start, const char *end)
 {
@@ -6677,7 +6686,7 @@ Item_param *LEX::add_placeholder(THD *thd, const LEX_CSTRING *name,
   Query_fragment pos(thd, sphead, start, end);
   Item_param *item= new (thd->mem_root) Item_param(thd, name,
                                                    pos.pos(), pos.length());
-  if (unlikely(!item) || unlikely(param_list.push_back(item, thd->mem_root)))
+  if (unlikely(!item) || unlikely(param_push_or_clone(thd, this, item)))
   {
     my_error(ER_OUT_OF_RESOURCES, MYF(0));
     return NULL;
diff --git a/sql/sql_lex.h b/sql/sql_lex.h
index 49e10421911..61063d06349 100644
--- a/sql/sql_lex.h
+++ b/sql/sql_lex.h
@@ -2803,6 +2803,11 @@ struct LEX: public Query_tables_list
      with clause in the current statement
   */
   With_clause **with_clauses_list_last_next;
+  /*
+    When a copy of a with element is parsed this is set to the offset of
+    the with element in the input string, otherwise it's set to 0
+  */
+  my_ptrdiff_t clone_spec_offset;
 
   Create_view_info *create_view;
 
diff --git a/sql/sql_parse.cc b/sql/sql_parse.cc
index c68e91bbef3..fa350379ee8 100644
--- a/sql/sql_parse.cc
+++ b/sql/sql_parse.cc
@@ -7877,7 +7877,9 @@ static void wsrep_mysql_parse(THD *thd, char *rawbuf, uint length,
                           "WAIT_FOR wsrep_retry_autocommit_continue";
                         DBUG_ASSERT(!debug_sync_set_action(thd, STRING_WITH_LEN(act)));
                       });
+      WSREP_DEBUG("Retry autocommit query: %s", thd->query());
     }
+
     mysql_parse(thd, rawbuf, length, parser_state, is_com_multi,
                 is_next_command);
 
diff --git a/sql/sql_partition.cc b/sql/sql_partition.cc
index ffe632d5409..151b94d1187 100644
--- a/sql/sql_partition.cc
+++ b/sql/sql_partition.cc
@@ -2575,6 +2575,18 @@ static int add_key_with_algorithm(String *str, partition_info *part_info)
   return err;
 }
 
+char *generate_partition_syntax_for_frm(THD *thd, partition_info *part_info,
+                                        uint *buf_length,
+                                        HA_CREATE_INFO *create_info,
+                                        Alter_info *alter_info)
+{
+  sql_mode_t old_mode= thd->variables.sql_mode;
+  thd->variables.sql_mode &= ~MODE_ANSI_QUOTES;
+  char *res= generate_partition_syntax(thd, part_info, buf_length,
+                                             true, create_info, alter_info);
+  thd->variables.sql_mode= old_mode;
+  return res;
+}
 
 /*
   Generate the partition syntax from the partition data structure.
diff --git a/sql/sql_partition.h b/sql/sql_partition.h
index 170ae8ccee1..83cac8f24ba 100644
--- a/sql/sql_partition.h
+++ b/sql/sql_partition.h
@@ -276,6 +276,10 @@ char *generate_partition_syntax(THD *thd, partition_info *part_info,
                                 bool show_partition_options,
                                 HA_CREATE_INFO *create_info,
                                 Alter_info *alter_info);
+char *generate_partition_syntax_for_frm(THD *thd, partition_info *part_info,
+                                        uint *buf_length,
+                                        HA_CREATE_INFO *create_info,
+                                        Alter_info *alter_info);
 bool verify_data_with_partition(TABLE *table, TABLE *part_table,
                                 uint32 part_id);
 bool compare_partition_options(HA_CREATE_INFO *table_create_info,
diff --git a/sql/sql_plugin_services.ic b/sql/sql_plugin_services.ic
index 47decf38973..8de53aa2161 100644
--- a/sql/sql_plugin_services.ic
+++ b/sql/sql_plugin_services.ic
@@ -181,7 +181,8 @@ static struct wsrep_service_st wsrep_handler = {
   wsrep_thd_ws_handle,
   wsrep_trx_is_aborting,
   wsrep_trx_order_before,
-  wsrep_unlock_rollback
+  wsrep_unlock_rollback,
+  wsrep_set_data_home_dir
 };
 
 static struct thd_specifics_service_st thd_specifics_handler=
diff --git a/sql/sql_prepare.cc b/sql/sql_prepare.cc
index 2ffecbab617..33e7e4dbf4e 100644
--- a/sql/sql_prepare.cc
+++ b/sql/sql_prepare.cc
@@ -893,6 +893,7 @@ static bool insert_params(Prepared_statement *stmt, uchar *null_array,
       DBUG_RETURN(1);
     if (param->convert_str_value(stmt->thd))
       DBUG_RETURN(1);                           /* out of memory */
+    param->sync_clones();
   }
   DBUG_RETURN(0);
 }
@@ -941,6 +942,7 @@ static bool insert_bulk_params(Prepared_statement *stmt,
     }
     else
       DBUG_RETURN(1); // long is not supported here
+    param->sync_clones();
   }
   DBUG_RETURN(0);
 }
@@ -969,6 +971,7 @@ static bool set_conversion_functions(Prepared_statement *stmt,
     read_pos+= 2;
     (**it).unsigned_flag= MY_TEST(typecode & signed_bit);
     (*it)->setup_conversion(thd, (uchar) (typecode & 0xff));
+    (*it)->sync_clones();
   }
   *data= read_pos;
   DBUG_RETURN(0);
@@ -1039,6 +1042,7 @@ static bool emb_insert_params(Prepared_statement *stmt, String *expanded_query)
         if (param->has_no_value())
           DBUG_RETURN(1);
       }
+      param->sync_clones();
     }
     if (param->convert_str_value(thd))
       DBUG_RETURN(1);                           /* out of memory */
@@ -1081,6 +1085,7 @@ static bool emb_insert_params_with_log(Prepared_statement *stmt, String *query)
 
     if (param->convert_str_value(thd))
       DBUG_RETURN(1);                           /* out of memory */
+    param->sync_clones();
   }
   if (acc.finalize())
     DBUG_RETURN(1);
@@ -1136,7 +1141,11 @@ swap_parameter_array(Item_param **param_array_dst,
   Item_param **end= param_array_dst + param_count;
 
   for (; dst < end; ++src, ++dst)
+  {
     (*dst)->set_param_type_and_swap_value(*src);
+    (*dst)->sync_clones();
+    (*src)->sync_clones();
+  }
 }
 
 
@@ -1167,6 +1176,7 @@ insert_params_from_actual_params(Prepared_statement *stmt,
     if (ps_param->save_in_param(stmt->thd, param) ||
         param->convert_str_value(stmt->thd))
       DBUG_RETURN(1);
+    param->sync_clones();
   }
   DBUG_RETURN(0);
 }
@@ -1209,6 +1219,8 @@ insert_params_from_actual_params_with_log(Prepared_statement *stmt,
 
     if (param->convert_str_value(thd))
       DBUG_RETURN(1);
+
+    param->sync_clones();
   }
   if (acc.finalize())
     DBUG_RETURN(1);
diff --git a/sql/sql_reload.cc b/sql/sql_reload.cc
index 9bcb9a30a4c..abdf9d76d15 100644
--- a/sql/sql_reload.cc
+++ b/sql/sql_reload.cc
@@ -289,9 +289,18 @@ bool reload_acl_and_cache(THD *thd, unsigned long long options,
         */
         if (tables)
         {
+          int err;
           for (TABLE_LIST *t= tables; t; t= t->next_local)
-            if (!find_table_for_mdl_upgrade(thd, t->db.str, t->table_name.str, false))
-              return 1;
+            if (!find_table_for_mdl_upgrade(thd, t->db.str, t->table_name.str, &err))
+            {
+              if (is_locked_view(thd, t))
+                t->next_local= t->next_global;
+              else
+              {
+                my_error(err, MYF(0), t->table_name.str);
+                return 1;
+              }
+            }
         }
         else
         {
@@ -616,4 +625,3 @@ static void disable_checkpoints(THD *thd)
       ha_checkpoint_state(1);                   // Disable checkpoints
   }
 }
-
diff --git a/sql/sql_select.cc b/sql/sql_select.cc
index e97ea680d87..cd8f3d07824 100644
--- a/sql/sql_select.cc
+++ b/sql/sql_select.cc
@@ -22126,11 +22126,30 @@ test_if_skip_sort_order(JOIN_TAB *tab,ORDER *order,ha_rows select_limit,
       tmp_map.clear_all();       // Force the creation of quick select
       tmp_map.set_bit(best_key); // only best_key.
       select->quick= 0;
+
+      bool cond_saved= false;
+      Item *saved_cond;
+
+      /*
+        Index Condition Pushdown may have removed parts of the condition for
+        this table. Temporarily put them back because we want the whole
+        condition for the range analysis.
+      */
+      if (select->pre_idx_push_select_cond)
+      {
+        saved_cond= select->cond;
+        select->cond= select->pre_idx_push_select_cond;
+        cond_saved= true;
+      }
+
       select->test_quick_select(join->thd, tmp_map, 0,
                                 join->select_options & OPTION_FOUND_ROWS ?
                                 HA_POS_ERROR :
                                 join->unit->select_limit_cnt,
                                 TRUE, FALSE, FALSE);
+
+      if (cond_saved)
+        select->cond= saved_cond;
     }
     order_direction= best_key_direction;
     /*
@@ -26071,7 +26090,7 @@ void TABLE_LIST::print(THD *thd, table_map eliminated_tables, String *str,
       LEX_CSTRING t_alias= alias;
 
       str->append(' ');
-      if (lower_case_table_names== 1)
+      if (lower_case_table_names == 1)
       {
         if (alias.str && alias.str[0])
         {
diff --git a/sql/sql_show.cc b/sql/sql_show.cc
index 33b28afd734..426f40d1729 100644
--- a/sql/sql_show.cc
+++ b/sql/sql_show.cc
@@ -5002,9 +5002,7 @@ public:
                         const char* msg,
                         Sql_condition ** cond_hdl)
   {
-    if (sql_errno == ER_PARSE_ERROR ||
-        sql_errno == ER_TRG_NO_DEFINER ||
-        sql_errno == ER_TRG_NO_CREATION_CTX)
+    if (sql_errno == ER_TRG_NO_DEFINER || sql_errno == ER_TRG_NO_CREATION_CTX)
       return true;
 
     if (*level != Sql_condition::WARN_LEVEL_ERROR)
diff --git a/sql/sql_statistics.cc b/sql/sql_statistics.cc
index ea18484f8a4..04806f07b3b 100644
--- a/sql/sql_statistics.cc
+++ b/sql/sql_statistics.cc
@@ -3272,6 +3272,9 @@ int read_statistics_for_tables_if_needed(THD *thd, TABLE_LIST *tables)
     if (!tl->is_view_or_derived() && !is_temporary_table(tl) && tl->table)
     { 
       TABLE_SHARE *table_share= tl->table->s;
+      if (table_share && !(table_share->table_category == TABLE_CATEGORY_USER))
+        continue;
+
       if (table_share && 
           table_share->stats_cb.stats_can_be_read &&
 	  !table_share->stats_cb.stats_is_read)
diff --git a/sql/sql_statistics.h b/sql/sql_statistics.h
index 714a9075f92..39cddf95188 100644
--- a/sql/sql_statistics.h
+++ b/sql/sql_statistics.h
@@ -333,12 +333,17 @@ private:
 public:
 
   Histogram histogram;
+
+  uint32 no_values_provided_bitmap()
+  {
+    return
+     ((1 << (COLUMN_STAT_HISTOGRAM-COLUMN_STAT_COLUMN_NAME))-1) <<
+      (COLUMN_STAT_COLUMN_NAME+1);
+  }
  
   void set_all_nulls()
   {
-    column_stat_nulls= 
-      ((1 << (COLUMN_STAT_HISTOGRAM-COLUMN_STAT_COLUMN_NAME))-1) <<
-      (COLUMN_STAT_COLUMN_NAME+1);
+    column_stat_nulls= no_values_provided_bitmap();
   }
 
   void set_not_null(uint stat_field_no)
@@ -384,8 +389,22 @@ public:
   bool min_max_values_are_provided()
   {
     return !is_null(COLUMN_STAT_MIN_VALUE) && 
-      !is_null(COLUMN_STAT_MIN_VALUE);
-  }          
+      !is_null(COLUMN_STAT_MAX_VALUE);
+  }
+  /*
+    This function checks whether the values for the fields of the statistical
+    tables that were NULL by DEFAULT for a column have changed or not.
+
+    @retval
+    TRUE: Statistics are not present for a column
+    FALSE: Statisitics are present for a column
+  */
+  bool no_stat_values_provided()
+  {
+    if (column_stat_nulls == no_values_provided_bitmap())
+      return true;
+    return false;
+  }
 };
 
 
diff --git a/sql/sql_table.cc b/sql/sql_table.cc
index daf8491a67f..bfeac0c3a57 100644
--- a/sql/sql_table.cc
+++ b/sql/sql_table.cc
@@ -29,7 +29,7 @@
 #include "lock.h"       // mysql_unlock_tables
 #include "strfunc.h"    // find_type2, find_set
 #include "sql_truncate.h"                       // regenerate_locked_table 
-#include "sql_partition.h"                   // generate_partition_syntax,
+#include "sql_partition.h"                      // mem_alloc_error,
                                                 // partition_info
                                                 // NOT_A_PARTITION_ID
 #include "sql_db.h"                             // load_db_opt_by_name
@@ -1838,13 +1838,10 @@ bool mysql_write_frm(ALTER_PARTITION_PARAM_TYPE *lpt, uint flags)
       partition_info *part_info= lpt->table->part_info;
       if (part_info)
       {
-        if (!(part_syntax_buf= generate_partition_syntax(lpt->thd, part_info,
-                                                         &syntax_len, TRUE,
-                                                         lpt->create_info,
-                                                         lpt->alter_info)))
-        {
+        part_syntax_buf= generate_partition_syntax_for_frm(lpt->thd, part_info,
+                               &syntax_len, lpt->create_info, lpt->alter_info);
+        if (!part_syntax_buf)
           DBUG_RETURN(TRUE);
-        }
         part_info->part_info_string= part_syntax_buf;
         part_info->part_info_len= syntax_len;
       }
@@ -1921,10 +1918,9 @@ bool mysql_write_frm(ALTER_PARTITION_PARAM_TYPE *lpt, uint flags)
     {
       TABLE_SHARE *share= lpt->table->s;
       char *tmp_part_syntax_str;
-      if (!(part_syntax_buf= generate_partition_syntax(lpt->thd, part_info,
-                                                       &syntax_len, TRUE,
-                                                       lpt->create_info,
-                                                       lpt->alter_info)))
+      part_syntax_buf= generate_partition_syntax_for_frm(lpt->thd,
+                   part_info, &syntax_len, lpt->create_info, lpt->alter_info);
+      if (!part_syntax_buf)
       {
         error= 1;
         goto err;
@@ -2113,7 +2109,7 @@ bool mysql_rm_table(THD *thd,TABLE_LIST *tables, bool if_exists,
             in its elements.
           */
           table->table= find_table_for_mdl_upgrade(thd, table->db.str,
-                                                   table->table_name.str, false);
+                                                   table->table_name.str, NULL);
           if (!table->table)
             DBUG_RETURN(true);
           table->mdl_request.ticket= table->table->mdl_ticket;
@@ -3456,6 +3452,10 @@ mysql_prepare_create_table(THD *thd, HA_CREATE_INFO *create_info,
         !(sql_field->charset= find_bin_collation(sql_field->charset)))
       DBUG_RETURN(true);
 
+    /* Virtual fields are always NULL */
+    if (sql_field->vcol_info)
+      sql_field->flags&= ~NOT_NULL_FLAG;
+
     if (sql_field->prepare_stage1(thd, thd->mem_root,
                                   file, file->ha_table_flags()))
       DBUG_RETURN(true);
@@ -4615,11 +4615,8 @@ handler *mysql_create_frm_image(THD *thd,
       We reverse the partitioning parser and generate a standard format
       for syntax stored in frm file.
     */
-    sql_mode_t old_mode= thd->variables.sql_mode;
-    thd->variables.sql_mode &= ~MODE_ANSI_QUOTES;
-    part_syntax_buf= generate_partition_syntax(thd, part_info, &syntax_len,
-                                               true, create_info, alter_info);
-    thd->variables.sql_mode= old_mode;
+    part_syntax_buf= generate_partition_syntax_for_frm(thd, part_info,
+                                      &syntax_len, create_info, alter_info);
     if (!part_syntax_buf)
       goto err;
     part_info->part_info_string= part_syntax_buf;
@@ -9799,8 +9796,10 @@ bool mysql_alter_table(THD *thd, const LEX_CSTRING *new_db,
       Table can be found in the list of open tables in THD::all_temp_tables
       list.
     */
-    tbl.table= thd->find_temporary_table(&tbl);
+    if ((tbl.table= thd->find_temporary_table(&tbl)) == NULL)
+      goto err_new_table_cleanup;
     new_table= tbl.table;
+    DBUG_ASSERT(new_table);
   }
   else
   {
@@ -9814,10 +9813,60 @@ bool mysql_alter_table(THD *thd, const LEX_CSTRING *new_db,
                                      alter_ctx.new_db.str,
                                      alter_ctx.tmp_name.str,
                                      true, true);
+    if (!new_table)
+      goto err_new_table_cleanup;
+
+    /*
+      Normally, an attempt to modify an FK parent table will cause
+      FK children to be prelocked, so the table-being-altered cannot
+      be modified by a cascade FK action, because ALTER holds a lock
+      and prelocking will wait.
+
+      But if a new FK is being added by this very ALTER, then the target
+      table is not locked yet (it's a temporary table). So, we have to
+      lock FK parents explicitly.
+    */
+    if (alter_info->flags & ALTER_ADD_FOREIGN_KEY)
+    {
+      List <FOREIGN_KEY_INFO> fk_list;
+      List_iterator<FOREIGN_KEY_INFO> fk_list_it(fk_list);
+      FOREIGN_KEY_INFO *fk;
+
+      /* tables_opened can be > 1 only for MERGE tables */
+      DBUG_ASSERT(tables_opened == 1);
+      DBUG_ASSERT(&table_list->next_global == thd->lex->query_tables_last);
+
+      new_table->file->get_foreign_key_list(thd, &fk_list);
+      while ((fk= fk_list_it++))
+      {
+        if (lower_case_table_names)
+        {
+         char buf[NAME_LEN];
+         size_t len;
+         strmake_buf(buf, fk->referenced_db->str);
+         len = my_casedn_str(files_charset_info, buf);
+         thd->make_lex_string(fk->referenced_db, buf, len);
+         strmake_buf(buf, fk->referenced_table->str);
+         len = my_casedn_str(files_charset_info, buf);
+         thd->make_lex_string(fk->referenced_table, buf, len);
+        }
+        if (table_already_fk_prelocked(table_list, fk->referenced_db,
+                                       fk->referenced_table, TL_READ_NO_INSERT))
+          continue;
+
+        TABLE_LIST *tl= (TABLE_LIST *) thd->alloc(sizeof(TABLE_LIST));
+        tl->init_one_table_for_prelocking(fk->referenced_db, fk->referenced_table,
+                           NULL, TL_READ_NO_INSERT, TABLE_LIST::PRELOCK_FK,
+                           NULL, 0, &thd->lex->query_tables_last);
+      }
+
+      if (open_tables(thd, &table_list->next_global, &tables_opened, 0,
+                      &alter_prelocking_strategy))
+        goto err_new_table_cleanup;
+    }
   }
-  if (!new_table)
-    goto err_new_table_cleanup;
   new_table->s->orig_table_name= table->s->table_name.str;
+
   /*
     Note: In case of MERGE table, we do not attach children. We do not
     copy data for MERGE tables. Only the children have data.
diff --git a/sql/sql_trigger.cc b/sql/sql_trigger.cc
index 755a9a9486d..60eff5ebc88 100644
--- a/sql/sql_trigger.cc
+++ b/sql/sql_trigger.cc
@@ -515,7 +515,11 @@ bool mysql_create_or_drop_trigger(THD *thd, TABLE_LIST *tables, bool create)
     if (err_status)
       goto end;
   }
-  WSREP_TO_ISOLATION_BEGIN(WSREP_MYSQL_DB, NULL, NULL);
+
+#ifdef WITH_WSREP
+  if (thd->wsrep_exec_mode == LOCAL_STATE)
+    WSREP_TO_ISOLATION_BEGIN(WSREP_MYSQL_DB, NULL, NULL);
+#endif
 
   /* We should have only one table in table list. */
   DBUG_ASSERT(tables->next_global == 0);
@@ -543,7 +547,7 @@ bool mysql_create_or_drop_trigger(THD *thd, TABLE_LIST *tables, bool create)
     /* Under LOCK TABLES we must only accept write locked tables. */
     if (!(tables->table= find_table_for_mdl_upgrade(thd, tables->db.str,
                                                     tables->table_name.str,
-                                                    FALSE)))
+                                                    NULL)))
       goto end;
   }
   else
diff --git a/sql/sql_truncate.cc b/sql/sql_truncate.cc
index 201825d4593..5a6417880b3 100644
--- a/sql/sql_truncate.cc
+++ b/sql/sql_truncate.cc
@@ -299,7 +299,7 @@ bool Sql_cmd_truncate_table::lock_table(THD *thd, TABLE_LIST *table_ref,
   if (thd->locked_tables_mode)
   {
     if (!(table= find_table_for_mdl_upgrade(thd, table_ref->db.str,
-                                            table_ref->table_name.str, FALSE)))
+                                            table_ref->table_name.str, NULL)))
       DBUG_RETURN(TRUE);
 
     *hton_can_recreate= ha_check_storage_engine_flag(table->file->ht,
diff --git a/sql/sql_type_int.h b/sql/sql_type_int.h
new file mode 100644
index 00000000000..1eda5651df5
--- /dev/null
+++ b/sql/sql_type_int.h
@@ -0,0 +1,44 @@
+/* Copyright (c) 2006, 2010, Oracle and/or its affiliates.
+   Copyright (c) 2011, 2016, MariaDB
+
+   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
+   the Free Software Foundation; version 2 of the License.
+
+   This program is distributed in the hope that it will be useful,
+   but WITHOUT ANY WARRANTY; without even the implied warranty of
+   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+   GNU General Public License for more details.
+
+   You should have received a copy of the GNU General Public License
+   along with this program; if not, write to the Free Software
+   Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA  02110-1301  USA */
+
+#ifndef SQL_TYPE_INT_INCLUDED
+#define SQL_TYPE_INT_INCLUDED
+
+
+// A longlong/ulonglong hybrid. Good to store results of val_int().
+class Longlong_hybrid
+{
+protected:
+  longlong m_value;
+  bool m_unsigned;
+public:
+  Longlong_hybrid(longlong nr, bool unsigned_flag)
+   :m_value(nr), m_unsigned(unsigned_flag)
+  { }
+  longlong value() const { return m_value; }
+  bool is_unsigned() const { return m_unsigned; }
+  bool neg() const { return m_value < 0 && !m_unsigned; }
+  ulonglong abs() const
+  {
+    if (m_unsigned)
+      return (ulonglong) m_value;
+    if (m_value == LONGLONG_MIN) // avoid undefined behavior
+      return ((ulonglong) LONGLONG_MAX) + 1;
+    return m_value < 0 ? -m_value : m_value;
+  }
+};
+
+#endif // SQL_TYPE_INT_INCLUDED
diff --git a/sql/sql_union.cc b/sql/sql_union.cc
index c8bf9bda57b..6368ed8afd8 100644
--- a/sql/sql_union.cc
+++ b/sql/sql_union.cc
@@ -1728,6 +1728,7 @@ bool st_select_lex_unit::exec_recursive()
       }
     }
     thd->lex->current_select= sl;
+    set_limit(sl);
     if (sl->tvc)
       sl->tvc->exec(sl);
     else
@@ -1796,6 +1797,37 @@ bool st_select_lex_unit::cleanup()
   {
     DBUG_RETURN(FALSE);
   }
+  /*
+    When processing a PS/SP or an EXPLAIN command cleanup of a unit can
+    be performed immediately when the unit is reached in the cleanup
+    traversal initiated by the cleanup of the main unit.
+  */
+  if (!thd->stmt_arena->is_stmt_prepare() && !thd->lex->describe &&
+      with_element && with_element->is_recursive && union_result)
+  {
+    select_union_recursive *result= with_element->rec_result;
+    if (++result->cleanup_count == with_element->rec_outer_references)
+    {
+      /*
+        Perform cleanup for with_element and for all with elements
+        mutually recursive with it.
+      */
+      cleaned= 1;
+      with_element->get_next_mutually_recursive()->spec->cleanup();
+    }
+    else
+    {
+      /*
+        Just increment by 1 cleanup_count for with_element and
+        for all with elements mutually recursive with it.
+      */
+      With_element *with_elem= with_element;
+      while ((with_elem= with_elem->get_next_mutually_recursive()) !=
+             with_element)
+        with_elem->rec_result->cleanup_count++;
+      DBUG_RETURN(FALSE);
+    }
+  }
   cleaned= 1;
 
   for (SELECT_LEX *sl= first_select(); sl; sl= sl->next_select())
@@ -1826,7 +1858,7 @@ bool st_select_lex_unit::cleanup()
 
   if (with_element && with_element->is_recursive)
   {
-    if (union_result )
+    if (union_result)
     {
       ((select_union_recursive *) union_result)->cleanup();
       delete union_result;
diff --git a/sql/sql_update.cc b/sql/sql_update.cc
index cb7bcdc33a1..7a496172aed 100644
--- a/sql/sql_update.cc
+++ b/sql/sql_update.cc
@@ -2266,7 +2266,7 @@ multi_update::~multi_update()
   TABLE_LIST *table;
   for (table= update_tables ; table; table= table->next_local)
   {
-    table->table->no_keyread= table->table->no_cache= 0;
+    table->table->no_keyread= 0;
     if (ignore)
       table->table->file->extra(HA_EXTRA_NO_IGNORE_DUP_KEY);
   }
diff --git a/sql/sql_yacc.yy b/sql/sql_yacc.yy
index 7ed44982dd6..f3f232493f3 100644
--- a/sql/sql_yacc.yy
+++ b/sql/sql_yacc.yy
@@ -1814,7 +1814,8 @@ bool my_yyoverflow(short **a, YYSTYPE **b, size_t *yystacksize);
         optionally_qualified_column_ident
 
 %type <simple_string>
-        remember_name remember_end remember_tok_start
+        remember_name remember_end
+        remember_tok_start remember_tok_end
         wild_and_where
 
 %type <const_simple_string>
@@ -9403,6 +9404,12 @@ remember_tok_start:
           }
         ;
 
+remember_tok_end:
+          {
+            $$= (char*) YYLIP->get_tok_end();
+          }
+        ;
+
 remember_name:
           {
             $$= (char*) YYLIP->get_cpp_tok_start();
@@ -14974,13 +14981,18 @@ with_list_element:
               MYSQL_YYABORT;
             Lex->with_column_list.empty();
           }
-          AS '(' remember_name subselect remember_end ')'
+          AS '(' remember_tok_start subselect remember_tok_end ')'
  	  {
+            LEX *lex= thd->lex;
+            const char *query_start= lex->sphead ? lex->sphead->m_tmp_query
+                                                 : thd->query();
+            char *spec_start= $6 + 1;
             With_element *elem= new With_element($1, *$2, $7->master_unit());
 	    if (unlikely(elem == NULL) ||
                 unlikely(Lex->curr_with_clause->add_with_element(elem)))
 	      MYSQL_YYABORT;
-	    if (unlikely(elem->set_unparsed_spec(thd, $6+1, $8)))
+            if (elem->set_unparsed_spec(thd, spec_start, $8,
+                                        spec_start - query_start))
               MYSQL_YYABORT;
 	  }
 	;
diff --git a/sql/sql_yacc_ora.yy b/sql/sql_yacc_ora.yy
index 8f98cfa3694..ecbf4f3aa28 100644
--- a/sql/sql_yacc_ora.yy
+++ b/sql/sql_yacc_ora.yy
@@ -1212,7 +1212,8 @@ bool my_yyoverflow(short **a, YYSTYPE **b, size_t *yystacksize);
         optionally_qualified_column_ident
 
 %type <simple_string>
-        remember_name remember_end remember_end_opt remember_tok_start
+        remember_name remember_end remember_end_opt
+        remember_tok_start remember_tok_end
         wild_and_where
         colon_with_pos
 
@@ -9466,6 +9467,12 @@ remember_tok_start:
           }
         ;
 
+remember_tok_end:
+          {
+            $$= (char*) YYLIP->get_tok_end();
+          }
+        ;
+
 remember_name:
           {
             $$= (char*) YYLIP->get_cpp_tok_start();
@@ -15148,13 +15155,18 @@ with_list_element:
               MYSQL_YYABORT;
             Lex->with_column_list.empty();
           }
-          AS '(' remember_name subselect remember_end ')'
+          AS '(' remember_tok_start subselect remember_tok_end ')'
  	  {
+            LEX *lex= thd->lex;
+            const char *query_start= lex->sphead ? lex->sphead->m_tmp_query
+                                                 : thd->query();
+            char *spec_start= $6 + 1;
             With_element *elem= new With_element($1, *$2, $7->master_unit());
 	    if (unlikely(elem == NULL) ||
                 unlikely(Lex->curr_with_clause->add_with_element(elem)))
 	      MYSQL_YYABORT;
-	    if (unlikely(elem->set_unparsed_spec(thd, $6+1, $8)))
+            if (elem->set_unparsed_spec(thd, spec_start, $8,
+                                        spec_start - query_start))
               MYSQL_YYABORT;
 	  }
 	;
diff --git a/sql/table.cc b/sql/table.cc
index 5eb789b779b..7ec2a22cbbb 100644
--- a/sql/table.cc
+++ b/sql/table.cc
@@ -431,6 +431,7 @@ void TABLE_SHARE::destroy()
     ha_share= NULL;                             // Safety
   }
 
+  delete_stat_values_for_table_share(this);
   delete sequence;
   free_root(&stats_cb.mem_root, MYF(0));
   stats_cb.stats_can_be_read= FALSE;
@@ -3122,6 +3123,7 @@ enum open_frm_error open_table_from_share(THD *thd, TABLE_SHARE *share,
 {
   enum open_frm_error error;
   uint records, i, bitmap_size, bitmap_count;
+  const char *tmp_alias;
   bool error_reported= FALSE;
   uchar *record, *bitmaps;
   Field **field_ptr;
@@ -3149,8 +3151,14 @@ enum open_frm_error open_table_from_share(THD *thd, TABLE_SHARE *share,
   init_sql_alloc(&outparam->mem_root, "table", TABLE_ALLOC_BLOCK_SIZE, 0,
                  MYF(0));
 
-  if (outparam->alias.copy(alias->str, alias->length, table_alias_charset))
+  /*
+    We have to store the original alias in mem_root as constraints and virtual
+    functions may store pointers to it
+  */
+  if (!(tmp_alias= strmake_root(&outparam->mem_root, alias->str, alias->length)))
     goto err;
+
+  outparam->alias.set(tmp_alias, alias->length, table_alias_charset);
   outparam->quick_keys.init();
   outparam->covering_keys.init();
   outparam->intersect_keys.init();
@@ -4585,7 +4593,7 @@ void TABLE::init(THD *thd, TABLE_LIST *tl)
                                    s->table_name.str,
                                    tl->alias.str);
   /* Fix alias if table name changes. */
-  if (strcmp(alias.c_ptr(), tl->alias.str))
+  if (!alias.alloced_length() || strcmp(alias.c_ptr(), tl->alias.str))
     alias.copy(tl->alias.str, tl->alias.length, alias.charset());
 
   tablenr= thd->current_tablenr++;
diff --git a/sql/table.h b/sql/table.h
index a14a82e5cd2..9f0825573cf 100644
--- a/sql/table.h
+++ b/sql/table.h
@@ -1945,6 +1945,14 @@ struct TABLE_LIST
                              const LEX_CSTRING *alias_arg,
                              enum thr_lock_type lock_type_arg)
   {
+    enum enum_mdl_type mdl_type;
+    if (lock_type_arg >= TL_WRITE_ALLOW_WRITE)
+      mdl_type= MDL_SHARED_WRITE;
+    else if (lock_type_arg == TL_READ_NO_INSERT)
+      mdl_type= MDL_SHARED_NO_WRITE;
+    else
+      mdl_type= MDL_SHARED_READ;
+
     bzero((char*) this, sizeof(*this));
     DBUG_ASSERT(!db_arg->str || strlen(db_arg->str) == db_arg->length);
     DBUG_ASSERT(!table_name_arg->str || strlen(table_name_arg->str) == table_name_arg->length);
@@ -1953,9 +1961,7 @@ struct TABLE_LIST
     table_name= *table_name_arg;
     alias= (alias_arg ? *alias_arg : *table_name_arg);
     lock_type= lock_type_arg;
-    mdl_request.init(MDL_key::TABLE, db.str, table_name.str,
-                     (lock_type >= TL_WRITE_ALLOW_WRITE) ?
-                     MDL_SHARED_WRITE : MDL_SHARED_READ,
+    mdl_request.init(MDL_key::TABLE, db.str, table_name.str, mdl_type,
                      MDL_TRANSACTION);
   }
 
@@ -1991,6 +1997,7 @@ struct TABLE_LIST
     *last_ptr= &next_global;
   }
 
+
   /*
     List of tables local to a subquery (used by SQL_I_List). Considers
     views as leaves (unlike 'next_leaf' below). Created at parse time
@@ -2605,6 +2612,16 @@ struct TABLE_LIST
   void check_pushable_cond_for_table(Item *cond);
   Item *build_pushable_cond_for_table(THD *thd, Item *cond); 
 
+  void remove_join_columns()
+  {
+    if (join_columns)
+    {
+      join_columns->empty();
+      join_columns= NULL;
+      is_join_columns_complete= FALSE;
+    }
+  }
+
 private:
   bool prep_check_option(THD *thd, uint8 check_opt_type);
   bool prep_where(THD *thd, Item **conds, bool no_where_clause);
diff --git a/sql/table_cache.cc b/sql/table_cache.cc
index 61e54c8f0f9..cb9583a2440 100644
--- a/sql/table_cache.cc
+++ b/sql/table_cache.cc
@@ -50,7 +50,6 @@
 #include "lf.h"
 #include "table.h"
 #include "sql_base.h"
-#include "sql_statistics.h"
 
 
 /** Configuration. */
@@ -464,6 +463,7 @@ void tc_release_table(TABLE *table)
   DBUG_ENTER("tc_release_table");
   DBUG_ASSERT(table->in_use);
   DBUG_ASSERT(table->file);
+  DBUG_ASSERT(!table->pos_in_locked_tables);
 
   mysql_mutex_lock(&tc[i].LOCK_table_cache);
   if (table->needs_reopen() || table->s->tdc->flushed ||
@@ -992,7 +992,6 @@ void tdc_release_share(TABLE_SHARE *share)
   }
   if (share->tdc->flushed || tdc_records() > tdc_size)
   {
-    delete_stat_values_for_table_share(share);
     mysql_mutex_unlock(&LOCK_unused_shares);
     tdc_delete_share_from_hash(share->tdc);
     DBUG_VOID_RETURN;
@@ -1142,11 +1141,10 @@ bool tdc_remove_table(THD *thd, enum_tdc_remove_table_type remove_type,
     All_share_tables_list::Iterator it(element->all_tables);
     while ((table= it++))
     {
-      my_refs++;
-      DBUG_ASSERT(table->in_use == thd);
+      if (table->in_use == thd)
+        my_refs++;
     }
   }
-  DBUG_ASSERT(element->all_tables.is_empty() || remove_type != TDC_RT_REMOVE_ALL);
   mysql_mutex_unlock(&element->LOCK_table_share);
 
   while ((table= purge_tables.pop_front()))
@@ -1178,6 +1176,17 @@ bool tdc_remove_table(THD *thd, enum_tdc_remove_table_type remove_type,
     mysql_mutex_lock(&element->LOCK_table_share);
     while (element->ref_count > my_refs)
       mysql_cond_wait(&element->COND_release, &element->LOCK_table_share);
+    DBUG_ASSERT(element->all_tables.is_empty() ||
+                remove_type != TDC_RT_REMOVE_ALL);
+#ifndef DBUG_OFF
+    if (remove_type == TDC_RT_REMOVE_NOT_OWN ||
+        remove_type == TDC_RT_REMOVE_NOT_OWN_KEEP_SHARE)
+    {
+      All_share_tables_list::Iterator it(element->all_tables);
+      while ((table= it++))
+        DBUG_ASSERT(table->in_use == thd);
+    }
+#endif
     mysql_mutex_unlock(&element->LOCK_table_share);
   }
 
diff --git a/sql/wsrep_dummy.cc b/sql/wsrep_dummy.cc
index 0b6e7e0d5bb..9a4bbd01bcd 100644
--- a/sql/wsrep_dummy.cc
+++ b/sql/wsrep_dummy.cc
@@ -139,3 +139,6 @@ int wsrep_trx_is_aborting(THD *)
 
 void wsrep_unlock_rollback()
 { }
+
+void wsrep_set_data_home_dir(const char *)
+{ }
diff --git a/sql/wsrep_hton.cc b/sql/wsrep_hton.cc
index d4d35d752be..8110faf7d11 100644
--- a/sql/wsrep_hton.cc
+++ b/sql/wsrep_hton.cc
@@ -25,6 +25,8 @@
 #include <cstdlib>
 #include "debug_sync.h"
 
+extern handlerton *binlog_hton;
+extern int binlog_close_connection(handlerton *hton, THD *thd);
 extern ulonglong thd_to_trx_id(THD *thd);
 
 extern "C" int thd_binlog_format(const MYSQL_THD thd);
@@ -173,7 +175,10 @@ wsrep_close_connection(handlerton*  hton, THD* thd)
   {
     DBUG_RETURN(0);
   }
-  DBUG_RETURN(wsrep_binlog_close_connection (thd));
+
+  if (wsrep_emulate_bin_log && thd_get_ha_data(thd, binlog_hton) != NULL)
+    binlog_hton->close_connection (binlog_hton, thd);
+  DBUG_RETURN(0);
 }
 
 /*
@@ -264,7 +269,7 @@ static int wsrep_rollback(handlerton *hton, THD *thd, bool all)
   }
 
   if ((all || !thd_test_options(thd, OPTION_NOT_AUTOCOMMIT | OPTION_BEGIN)) &&
-      (thd->variables.wsrep_on && thd->wsrep_conflict_state != MUST_REPLAY))
+      thd->variables.wsrep_on && thd->wsrep_conflict_state != MUST_REPLAY)
   {
     if (wsrep && wsrep->post_rollback(wsrep, &thd->wsrep_ws_handle))
     {
diff --git a/sql/wsrep_mysqld.cc b/sql/wsrep_mysqld.cc
index 76b27c91d01..886119a2327 100644
--- a/sql/wsrep_mysqld.cc
+++ b/sql/wsrep_mysqld.cc
@@ -73,10 +73,7 @@ const char *wsrep_start_position;
 const char *wsrep_data_home_dir;
 const char *wsrep_dbug_option;
 const char *wsrep_notify_cmd;
-const char *wsrep_sst_method;
-const char *wsrep_sst_receive_address;
-const char *wsrep_sst_donor;
-const char *wsrep_sst_auth;
+
 my_bool wsrep_debug;                            // Enable debug level logging
 my_bool wsrep_convert_LOCK_to_trx;              // Convert locking sessions to trx
 my_bool wsrep_auto_increment_control;           // Control auto increment variables
@@ -88,7 +85,6 @@ my_bool wsrep_log_conflicts;
 my_bool wsrep_load_data_splitting;              // Commit load data every 10K intervals
 my_bool wsrep_slave_UK_checks;                  // Slave thread does UK checks
 my_bool wsrep_slave_FK_checks;                  // Slave thread does FK checks
-my_bool wsrep_sst_donor_rejects_queries;
 my_bool wsrep_restart_slave;                    // Should mysql slave thread be
                                                 // restarted, when node joins back?
 my_bool wsrep_desync;                           // De(re)synchronize the node from the
@@ -653,6 +649,9 @@ int wsrep_init()
             wsrep->provider_vendor,  sizeof(provider_vendor) - 1);
   }
 
+  if (!wsrep_data_home_dir || strlen(wsrep_data_home_dir) == 0)
+    wsrep_data_home_dir = mysql_real_data_home;
+
   /* Initialize node address */
   char node_addr[512]= { 0, };
   size_t const node_addr_max= sizeof(node_addr) - 1;
@@ -1078,114 +1077,151 @@ static bool wsrep_prepare_key_for_isolation(const char* db,
                                             wsrep_buf_t* key,
                                             size_t* key_len)
 {
-    if (*key_len < 2) return false;
+  if (*key_len < 2) return false;
 
-    switch (wsrep_protocol_version)
-    {
-    case 0:
-        *key_len= 0;
-        break;
-    case 1:
-    case 2:
-    case 3:
+  switch (wsrep_protocol_version)
+  {
+  case 0:
+    *key_len= 0;
+    break;
+  case 1:
+  case 2:
+  case 3:
+  {
+    *key_len= 0;
+    if (db)
     {
-        *key_len= 0;
-        if (db)
-        {
-            // sql_print_information("%s.%s", db, table);
-            if (db)
-            {
-                key[*key_len].ptr= db;
-                key[*key_len].len= strlen(db);
-                ++(*key_len);
-                if (table)
-                {
-                    key[*key_len].ptr= table;
-                    key[*key_len].len= strlen(table);
-                    ++(*key_len);
-                }
-            }
-        }
-        break;
+      // sql_print_information("%s.%s", db, table);
+      key[*key_len].ptr= db;
+      key[*key_len].len= strlen(db);
+      ++(*key_len);
+      if (table)
+      {
+        key[*key_len].ptr= table;
+        key[*key_len].len= strlen(table);
+        ++(*key_len);
+      }
     }
-    default:
+    break;
+  }
+  default:
+    return false;
+  }
+  return true;
+}
+
+
+static bool wsrep_prepare_key_for_isolation(const char* db,
+                                            const char* table,
+                                            wsrep_key_arr_t* ka)
+{
+  wsrep_key_t* tmp;
+
+  if (!ka->keys)
+    tmp= (wsrep_key_t*)my_malloc((ka->keys_len + 1) * sizeof(wsrep_key_t),
+                                 MYF(0));
+  else
+    tmp= (wsrep_key_t*)my_realloc(ka->keys,
+                                 (ka->keys_len + 1) * sizeof(wsrep_key_t),
+                                 MYF(0));
+
+  if (!tmp)
+  {
+    WSREP_ERROR("Can't allocate memory for key_array");
+    return false;
+  }
+  ka->keys= tmp;
+  if (!(ka->keys[ka->keys_len].key_parts= (wsrep_buf_t*)
+        my_malloc(sizeof(wsrep_buf_t)*2, MYF(0))))
+  {
+    WSREP_ERROR("Can't allocate memory for key_parts");
+    return false;
+  }
+  ka->keys[ka->keys_len].key_parts_num= 2;
+  ++ka->keys_len;
+  if (!wsrep_prepare_key_for_isolation(db, table,
+                                       (wsrep_buf_t*)ka->keys[ka->keys_len - 1].key_parts,
+                                       &ka->keys[ka->keys_len - 1].key_parts_num))
+  {
+    WSREP_ERROR("Preparing keys for isolation failed");
+    return false;
+  }
+
+  return true;
+}
+
+
+static bool wsrep_prepare_keys_for_alter_add_fk(const char* child_table_db,
+                                                Alter_info* alter_info,
+                                                wsrep_key_arr_t* ka)
+{
+  Key *key;
+  List_iterator<Key> key_iterator(alter_info->key_list);
+  while ((key= key_iterator++))
+  {
+    if (key->type == Key::FOREIGN_KEY)
+    {
+      Foreign_key *fk_key= (Foreign_key *)key;
+      const char *db_name= fk_key->ref_db.str;
+      const char *table_name= fk_key->ref_table.str;
+      if (!db_name)
+      {
+        db_name= child_table_db;
+      }
+      if (!wsrep_prepare_key_for_isolation(db_name, table_name, ka))
+      {
         return false;
+      }
     }
-
-    return true;
+  }
+  return true;
 }
 
-/* Prepare key list from db/table and table_list */
-bool wsrep_prepare_keys_for_isolation(THD*              thd,
-                                      const char*       db,
-                                      const char*       table,
-                                      const TABLE_LIST* table_list,
-                                      wsrep_key_arr_t*  ka)
+
+static bool wsrep_prepare_keys_for_isolation(THD*              thd,
+                                             const char*       db,
+                                             const char*       table,
+                                             const TABLE_LIST* table_list,
+                                             Alter_info*       alter_info,
+                                             wsrep_key_arr_t*  ka)
 {
   ka->keys= 0;
   ka->keys_len= 0;
 
   if (db || table)
   {
-    if (!(ka->keys= (wsrep_key_t*)my_malloc(sizeof(wsrep_key_t), MYF(0))))
-    {
-      WSREP_ERROR("Can't allocate memory for key_array");
-      goto err;
-    }
-    ka->keys_len= 1;
-    if (!(ka->keys[0].key_parts= (wsrep_buf_t*)
-          my_malloc(sizeof(wsrep_buf_t)*2, MYF(0))))
-    {
-      WSREP_ERROR("Can't allocate memory for key_parts");
+    if (!wsrep_prepare_key_for_isolation(db, table, ka))
       goto err;
-     }
-    ka->keys[0].key_parts_num= 2;
-    if (!wsrep_prepare_key_for_isolation(
-                                         db, table,
-                                         (wsrep_buf_t*)ka->keys[0].key_parts,
-                                         &ka->keys[0].key_parts_num))
-    {
-      WSREP_ERROR("Preparing keys for isolation failed (1)");
-      goto err;
-    }
   }
 
   for (const TABLE_LIST* table= table_list; table; table= table->next_global)
   {
-    wsrep_key_t* tmp;
-    if (ka->keys)
-      tmp= (wsrep_key_t*)my_realloc(ka->keys,
-                                  (ka->keys_len + 1) * sizeof(wsrep_key_t),
-                                  MYF(0));
-    else
-      tmp= (wsrep_key_t*)my_malloc((ka->keys_len + 1) * sizeof(wsrep_key_t), MYF(0));
-
-    if (!tmp)
-    {
-      WSREP_ERROR("Can't allocate memory for key_array");
-      goto err;
-    }
-    ka->keys= tmp;
-    if (!(ka->keys[ka->keys_len].key_parts= (wsrep_buf_t*)
-          my_malloc(sizeof(wsrep_buf_t)*2, MYF(0))))
-    {
-      WSREP_ERROR("Can't allocate memory for key_parts");
+    if (!wsrep_prepare_key_for_isolation(table->db.str, table->table_name.str, ka))
       goto err;
-    }
-    ka->keys[ka->keys_len].key_parts_num= 2;
-    ++ka->keys_len;
-    if (!wsrep_prepare_key_for_isolation(table->db.str, table->table_name.str,
-                                         (wsrep_buf_t*)ka->keys[ka->keys_len - 1].key_parts,
-                                         &ka->keys[ka->keys_len - 1].key_parts_num))
-    {
-      WSREP_ERROR("Preparing keys for isolation failed (2)");
+  }
+
+  if (alter_info && (alter_info->flags & (ALTER_ADD_FOREIGN_KEY)))
+  {
+    if (!wsrep_prepare_keys_for_alter_add_fk(table_list->db.str, alter_info, ka))
       goto err;
-    }
   }
-    return 0;
+
+  return false;
+
 err:
-    wsrep_keys_free(ka);
-    return 1;
+  wsrep_keys_free(ka);
+  return true;
+}
+
+
+/* Prepare key list from db/table and table_list */
+bool wsrep_prepare_keys_for_isolation(THD*              thd,
+                                      const char*       db,
+                                      const char*       table,
+                                      const TABLE_LIST* table_list,
+                                      wsrep_key_arr_t*  ka)
+{
+  return wsrep_prepare_keys_for_isolation(thd, db, table, table_list, NULL, ka);
 }
 
 
@@ -1398,6 +1434,67 @@ create_view_query(THD *thd, uchar** buf, size_t* buf_len)
     return wsrep_to_buf_helper(thd, buff.ptr(), buff.length(), buf, buf_len);
 }
 
+/*
+  Rewrite DROP TABLE for TOI. Temporary tables are eliminated from
+  the query as they are visible only to client connection.
+
+  TODO: See comments for sql_base.cc:drop_temporary_table() and refine
+  the function to deal with transactional locked tables.
+ */
+static int wsrep_drop_table_query(THD* thd, uchar** buf, size_t* buf_len)
+{
+
+  LEX* lex= thd->lex;
+  SELECT_LEX* select_lex= &lex->select_lex;
+  TABLE_LIST* first_table= select_lex->table_list.first;
+  String buff;
+
+  DBUG_ASSERT(!lex->create_info.tmp_table());
+
+  bool found_temp_table= false;
+  for (TABLE_LIST* table= first_table; table; table= table->next_global)
+  {
+    if (thd->find_temporary_table(table->db.str, table->table_name.str))
+    {
+      found_temp_table= true;
+      break;
+    }
+  }
+
+  if (found_temp_table)
+  {
+    buff.append("DROP TABLE ");
+    if (lex->check_exists)
+      buff.append("IF EXISTS ");
+
+    for (TABLE_LIST* table= first_table; table; table= table->next_global)
+    {
+      if (!thd->find_temporary_table(table->db.str, table->table_name.str))
+      {
+        append_identifier(thd, &buff, table->db.str, table->db.length);
+        buff.append(".");
+        append_identifier(thd, &buff,
+                          table->table_name.str, table->table_name.length);
+        buff.append(",");
+      }
+    }
+
+    /* Chop the last comma */
+    buff.chop();
+    buff.append(" /* generated by wsrep */");
+
+    WSREP_DEBUG("Rewrote '%s' as '%s'", thd->query(), buff.ptr());
+
+    return wsrep_to_buf_helper(thd, buff.ptr(), buff.length(), buf, buf_len);
+  }
+  else
+  {
+    return wsrep_to_buf_helper(thd, thd->query(), thd->query_length(),
+                               buf, buf_len);
+  }
+}
+
+
 /* Forward declarations. */
 static int wsrep_create_sp(THD *thd, uchar** buf, size_t* buf_len);
 static int wsrep_create_trigger_query(THD *thd, uchar** buf, size_t* buf_len);
@@ -1506,7 +1603,8 @@ static const char* wsrep_get_query_or_msg(const THD* thd)
   -1: TOI replication failed 
  */
 static int wsrep_TOI_begin(THD *thd, const char *db_, const char *table_,
-                           const TABLE_LIST* table_list)
+                           const TABLE_LIST* table_list,
+                           Alter_info* alter_info)
 {
   wsrep_status_t ret(WSREP_WARNING);
   uchar* buf(0);
@@ -1541,6 +1639,9 @@ static int wsrep_TOI_begin(THD *thd, const char *db_, const char *table_,
   case SQLCOM_ALTER_EVENT:
     buf_err= wsrep_alter_event_query(thd, &buf, &buf_len);
     break;
+  case SQLCOM_DROP_TABLE:
+    buf_err= wsrep_drop_table_query(thd, &buf, &buf_len);
+    break;
   case SQLCOM_CREATE_ROLE:
     if (sp_process_definer(thd))
     {
@@ -1556,7 +1657,8 @@ static int wsrep_TOI_begin(THD *thd, const char *db_, const char *table_,
   wsrep_key_arr_t key_arr= {0, 0};
   struct wsrep_buf buff = { buf, buf_len };
   if (!buf_err                                                                  &&
-      !wsrep_prepare_keys_for_isolation(thd, db_, table_, table_list, &key_arr) &&
+      !wsrep_prepare_keys_for_isolation(thd, db_, table_,
+                                        table_list, alter_info, &key_arr)       &&
       key_arr.keys_len > 0                                                      &&
       WSREP_OK == (ret = wsrep->to_execute_start(wsrep, thd->thread_id,
 						 key_arr.keys, key_arr.keys_len,
@@ -1696,7 +1798,8 @@ static void wsrep_RSU_end(THD *thd)
 }
 
 int wsrep_to_isolation_begin(THD *thd, const char *db_, const char *table_,
-                             const TABLE_LIST* table_list)
+                             const TABLE_LIST* table_list,
+                             Alter_info* alter_info)
 {
   int ret= 0;
 
@@ -1750,10 +1853,10 @@ int wsrep_to_isolation_begin(THD *thd, const char *db_, const char *table_,
   {
     switch (thd->variables.wsrep_OSU_method) {
     case WSREP_OSU_TOI:
-      ret =  wsrep_TOI_begin(thd, db_, table_, table_list);
+      ret= wsrep_TOI_begin(thd, db_, table_, table_list, alter_info);
       break;
     case WSREP_OSU_RSU:
-      ret =  wsrep_RSU_begin(thd, db_, table_);
+      ret= wsrep_RSU_begin(thd, db_, table_);
       break;
     default:
       WSREP_ERROR("Unsupported OSU method: %lu",
diff --git a/sql/wsrep_mysqld.h b/sql/wsrep_mysqld.h
index 6aa8a68f222..819a56b9f23 100644
--- a/sql/wsrep_mysqld.h
+++ b/sql/wsrep_mysqld.h
@@ -159,7 +159,6 @@ extern "C" time_t wsrep_thd_query_start(THD *thd);
 extern "C" query_id_t wsrep_thd_query_id(THD *thd);
 extern "C" query_id_t wsrep_thd_wsrep_last_query_id(THD *thd);
 extern "C" void wsrep_thd_set_wsrep_last_query_id(THD *thd, query_id_t id);
-extern "C" void wsrep_set_data_home_dir(const char *data_dir);
 
 extern void wsrep_close_client_connections(my_bool wait_to_end);
 extern int  wsrep_wait_committing_connections_close(int wait_time);
@@ -286,8 +285,10 @@ extern PSI_mutex_key key_LOCK_wsrep_desync;
 extern PSI_file_key key_file_wsrep_gra_log;
 #endif /* HAVE_PSI_INTERFACE */
 struct TABLE_LIST;
+class Alter_info;
 int wsrep_to_isolation_begin(THD *thd, const char *db_, const char *table_,
-                             const TABLE_LIST* table_list);
+                             const TABLE_LIST* table_list,
+                             Alter_info* alter_info = NULL);
 void wsrep_to_isolation_end(THD *thd);
 void wsrep_cleanup_transaction(THD *thd);
 int wsrep_to_buf_helper(
diff --git a/sql/wsrep_sst.cc b/sql/wsrep_sst.cc
index 0a1d95f30b8..e648a7f4c69 100644
--- a/sql/wsrep_sst.cc
+++ b/sql/wsrep_sst.cc
@@ -15,6 +15,7 @@
 
 #include "mariadb.h"
 #include "wsrep_sst.h"
+#include <inttypes.h>
 #include <mysqld.h>
 #include <m_ctype.h>
 #include <strfunc.h>
@@ -36,8 +37,14 @@ static char wsrep_defaults_file[FN_REFLEN * 2 + 10 + 30 +
                                 sizeof(WSREP_SST_OPT_CONF_SUFFIX) +
                                 sizeof(WSREP_SST_OPT_CONF_EXTRA)] = {0};
 
+const char* wsrep_sst_method          = WSREP_SST_DEFAULT;
+const char* wsrep_sst_receive_address = WSREP_SST_ADDRESS_AUTO;
+const char* wsrep_sst_donor           = "";
+const char* wsrep_sst_auth            = NULL;
+
 // container for real auth string
 static const char* sst_auth_real      = NULL;
+my_bool wsrep_sst_donor_rejects_queries = FALSE;
 
 bool wsrep_sst_method_check (sys_var *self, THD* thd, set_var* var)
 {
@@ -60,7 +67,6 @@ bool wsrep_sst_method_update (sys_var *self, THD* thd, enum_var_type type)
 
 static const char* data_home_dir = NULL;
 
-extern "C"
 void wsrep_set_data_home_dir(const char *data_dir)
 {
   data_home_dir= (data_dir && *data_dir) ? data_dir : NULL;
@@ -156,7 +162,7 @@ void wsrep_sst_auth_free()
 
 bool wsrep_sst_auth_update (sys_var *self, THD* thd, enum_var_type type)
 {
-    return sst_auth_real_set (wsrep_sst_auth);
+  return sst_auth_real_set (wsrep_sst_auth);
 }
 
 void wsrep_sst_auth_init ()
@@ -171,7 +177,7 @@ bool  wsrep_sst_donor_check (sys_var *self, THD* thd, set_var* var)
 
 bool wsrep_sst_donor_update (sys_var *self, THD* thd, enum_var_type type)
 {
-    return 0;
+  return 0;
 }
 
 bool wsrep_before_SE()
@@ -217,7 +223,7 @@ bool wsrep_sst_wait ()
     if (!sst_complete)
     {
       total_wtime += difftime(end_time, start_time);
-      WSREP_DEBUG("Waiting for SST to complete. current seqno: %ld waited %f secs.", local_seqno, total_wtime);
+      WSREP_DEBUG("Waiting for SST to complete. current seqno: %" PRId64 " waited %f secs.", local_seqno, total_wtime);
       service_manager_extend_timeout(WSREP_EXTEND_TIMEOUT_INTERVAL,
         "WSREP state transfer ongoing, current seqno: %ld waited %f secs", local_seqno, total_wtime);
     }
@@ -303,7 +309,7 @@ bool wsrep_sst_received (wsrep_t*            const wsrep,
   }
 
   if (memcmp(&local_uuid, &uuid, sizeof(wsrep_uuid_t)) ||
-      local_seqno < seqno)
+      local_seqno < seqno || seqno < 0)
   {
     do_update= true;
   }
@@ -454,6 +460,22 @@ static int generate_binlog_opt_val(char** ret)
   return 0;
 }
 
+static int generate_binlog_index_opt_val(char** ret)
+{
+  DBUG_ASSERT(ret);
+  *ret= NULL;
+  if (opt_binlog_index_name) {
+    *ret= strcmp(opt_binlog_index_name, "0") ?
+      my_strdup(opt_binlog_index_name, MYF(0)) : my_strdup("", MYF(0));
+  }
+  else
+  {
+    *ret= my_strdup("", MYF(0));
+  }
+  if (!*ret) return -ENOMEM;
+  return 0;
+}
+
 static void* sst_joiner_thread (void* a)
 {
   sst_thread_arg* arg= (sst_thread_arg*) a;
@@ -641,7 +663,9 @@ static ssize_t sst_prepare_other (const char*  method,
   }
 
   const char* binlog_opt= "";
+  const char* binlog_index_opt= "";
   char* binlog_opt_val= NULL;
+  char* binlog_index_opt_val= NULL;
 
   int ret;
   if ((ret= generate_binlog_opt_val(&binlog_opt_val)))
@@ -650,7 +674,15 @@ static ssize_t sst_prepare_other (const char*  method,
                 ret);
     return ret;
   }
+
+  if ((ret= generate_binlog_index_opt_val(&binlog_index_opt_val)))
+  {
+    WSREP_ERROR("sst_prepare_other(): generate_binlog_index_opt_val() failed %d",
+                ret);
+  }
+
   if (strlen(binlog_opt_val)) binlog_opt= WSREP_SST_OPT_BINLOG;
+  if (strlen(binlog_index_opt_val)) binlog_index_opt= WSREP_SST_OPT_BINLOG_INDEX;
 
   make_wsrep_defaults_file();
 
@@ -661,11 +693,14 @@ static ssize_t sst_prepare_other (const char*  method,
                  WSREP_SST_OPT_DATA " '%s' "
                  " %s "
                  WSREP_SST_OPT_PARENT " '%d'"
-                 " %s '%s' ",
+                 " %s '%s'"
+	         " %s '%s'",
                  method, addr_in, mysql_real_data_home,
                  wsrep_defaults_file,
-                 (int)getpid(), binlog_opt, binlog_opt_val);
+                 (int)getpid(), binlog_opt, binlog_opt_val,
+                 binlog_index_opt, binlog_index_opt_val);
   my_free(binlog_opt_val);
+  my_free(binlog_index_opt_val);
 
   if (ret < 0 || ret >= cmd_len)
   {
@@ -1431,7 +1466,7 @@ void wsrep_SE_init_wait()
     if (!SE_initialized)
     {
       total_wtime += difftime(end_time, start_time);
-      WSREP_DEBUG("Waiting for SST to complete. current seqno: %ld waited %f secs.", local_seqno, total_wtime);
+      WSREP_DEBUG("Waiting for SST to complete. current seqno: %" PRId64 " waited %f secs.", local_seqno, total_wtime);
       service_manager_extend_timeout(WSREP_EXTEND_TIMEOUT_INTERVAL,
         "WSREP state transfer ongoing, current seqno: %ld waited %f secs", local_seqno, total_wtime);
     }
diff --git a/sql/wsrep_sst.h b/sql/wsrep_sst.h
index 8bf6dc31464..cc0f1f5389d 100644
--- a/sql/wsrep_sst.h
+++ b/sql/wsrep_sst.h
@@ -31,6 +31,7 @@
 #define WSREP_SST_OPT_CONF_EXTRA  "--defaults-extra-file"
 #define WSREP_SST_OPT_PARENT   "--parent"
 #define WSREP_SST_OPT_BINLOG   "--binlog"
+#define WSREP_SST_OPT_BINLOG_INDEX "--binlog-index"
 
 // mysqldump-specific options
 #define WSREP_SST_OPT_USER     "--user"
diff --git a/sql/wsrep_var.cc b/sql/wsrep_var.cc
index bc43b844a35..f8a494416e2 100644
--- a/sql/wsrep_var.cc
+++ b/sql/wsrep_var.cc
@@ -595,7 +595,7 @@ void wsrep_node_address_init (const char* value)
 static void wsrep_slave_count_change_update ()
 {
   wsrep_slave_count_change = (wsrep_slave_threads - wsrep_prev_slave_threads);
-  WSREP_DEBUG("Change on slave threads: New %lu old %lu difference %lu",
+  WSREP_DEBUG("Change on slave threads: New %lu old %lu difference %d",
 	  wsrep_slave_threads, wsrep_prev_slave_threads, wsrep_slave_count_change);
   wsrep_prev_slave_threads = wsrep_slave_threads;
 }
@@ -758,4 +758,3 @@ int wsrep_show_status (THD *thd, SHOW_VAR *var, char *buff,
   v->name= 0;                                   // terminator
   return 0;
 }
-