Merge 10.9 into 10.10

64 files changed, 2420 insertions, 892 deletions

diff --git a/sql/field.cc b/sql/field.cc
index 2fe66a96b15..4435e8c7f38 100644
--- a/sql/field.cc
+++ b/sql/field.cc
@@ -3345,11 +3345,12 @@ Field_new_decimal::Field_new_decimal(uchar *ptr_arg,
                                      decimal_digits_t dec_arg,bool zero_arg,
                                      bool unsigned_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)
+             unireg_check_arg, field_name_arg,
+             MY_MIN(dec_arg, DECIMAL_MAX_SCALE), zero_arg, unsigned_arg)
 {
   precision= get_decimal_precision(len_arg, dec_arg, unsigned_arg);
-  DBUG_ASSERT((precision <= DECIMAL_MAX_PRECISION) &&
-              (dec <= DECIMAL_MAX_SCALE));
+  DBUG_ASSERT(precision <= DECIMAL_MAX_PRECISION);
+  DBUG_ASSERT(dec <= DECIMAL_MAX_SCALE);
   bin_size= my_decimal_get_binary_size(precision, dec);
 }
 
diff --git a/sql/field.h b/sql/field.h
index 7b279ffb41e..cb78bdccd4b 100644
--- a/sql/field.h
+++ b/sql/field.h
@@ -5501,22 +5501,22 @@ public:
 
   bool check_vcol_for_key(THD *thd) const;
 
-  void set_lex_charset_collation(const Lex_charset_collation_st &lc)
+  void set_charset_collation_attrs(const
+                                   Lex_column_charset_collation_attrs_st &lc)
   {
-    charset= lc.charset_collation();
+    charset= lc.charset_info();
     if (lc.is_contextually_typed_collation())
       flags|= CONTEXT_COLLATION_FLAG;
     else
       flags&= ~CONTEXT_COLLATION_FLAG;
   }
-  Lex_charset_collation lex_charset_collation() const
+  Lex_column_charset_collation_attrs charset_collation_attrs() const
   {
-    return Lex_charset_collation(
-             charset,
-             !charset ? Lex_charset_collation_st::TYPE_EMPTY :
-             flags & CONTEXT_COLLATION_FLAG ?
-             Lex_charset_collation_st::TYPE_COLLATE_CONTEXTUALLY_TYPED :
-             Lex_charset_collation_st::TYPE_CHARACTER_SET);
+    if (!charset)
+      return Lex_column_charset_collation_attrs();
+    if (flags & CONTEXT_COLLATION_FLAG)
+      return Lex_column_charset_collation_attrs(Lex_context_collation(charset));
+    return Lex_column_charset_collation_attrs(Lex_exact_collation(charset));
   }
 };
 
diff --git a/sql/gcalc_tools.cc b/sql/gcalc_tools.cc
index 307f063fb43..25c80a7a796 100644
--- a/sql/gcalc_tools.cc
+++ b/sql/gcalc_tools.cc
@@ -132,7 +132,7 @@ int Gcalc_function::count_internal(const char *cur_func, uint set_type,
   int mask= (c_op & op_not) ? 1:0;
   uint n_ops= c_op & ~(op_any | op_not | v_mask);
   uint n_shape= c_op & ~(op_any | op_not | v_mask); /* same as n_ops */
-  value v_state= (value) (c_op & v_mask);
+  op_type v_state= (op_type) (c_op & v_mask);
   int result= 0;
   const char *sav_cur_func= cur_func;
 
diff --git a/sql/gcalc_tools.h b/sql/gcalc_tools.h
index e625b355d95..bb1f473e180 100644
--- a/sql/gcalc_tools.h
+++ b/sql/gcalc_tools.h
@@ -52,17 +52,15 @@ private:
   int count_internal(const char *cur_func, uint set_type,
                      const char **end);
 public:
-  enum value
-  {
-    v_empty=   0x0000000,
-    v_find_t=  0x1000000,
-    v_find_f=  0x2000000,
-    v_t_found= 0x3000000,
-    v_f_found= 0x4000000,
-    v_mask=    0x7000000
-  };
   enum op_type
   {
+    v_empty=          0x00000000,
+    v_find_t=         0x01000000,
+    v_find_f=         0x02000000,
+    v_t_found=        0x03000000,
+    v_f_found=        0x04000000,
+    v_mask=           0x07000000,
+
     op_not=           0x80000000,
     op_shape=         0x00000000,
     op_union=         0x10000000,
diff --git a/sql/ha_partition.cc b/sql/ha_partition.cc
index 34c65f2e45d..e85ce02b7e6 100644
--- a/sql/ha_partition.cc
+++ b/sql/ha_partition.cc
@@ -1,6 +1,6 @@
 /*
   Copyright (c) 2005, 2019, Oracle and/or its affiliates.
-  Copyright (c) 2009, 2021, MariaDB
+  Copyright (c) 2009, 2022, 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
@@ -4177,6 +4177,8 @@ int ha_partition::external_lock(THD *thd, int lock_type)
   if (lock_type == F_UNLCK)
   {
     bitmap_clear_all(used_partitions);
+    if (m_lock_type == F_WRLCK && m_part_info->vers_require_hist_part(thd))
+      m_part_info->vers_check_limit(thd);
   }
   else
   {
@@ -10576,7 +10578,16 @@ bool ha_partition::commit_inplace_alter_table(TABLE *altered_table,
         Loop over all other partitions as to follow the protocol!
       */
       uint i;
-      DBUG_ASSERT(0);
+      /*
+        InnoDB does not set ha_alter_info->group_commit_ctx to NULL in the
+        case if autoincrement attribute is necessary to reset for all
+        partitions for INNOBASE_INPLACE_IGNORE handler flags. It does not
+        affect durability, because it is solely about updating the InnoDB data
+        dictionary caches (one InnoDB dict_table_t per partition or
+        sub-partition).
+      */
+      DBUG_ASSERT(table->found_next_number_field
+          && !altered_table->found_next_number_field);
       for (i= 1; i < m_tot_parts; i++)
       {
         ha_alter_info->handler_ctx= part_inplace_ctx->handler_ctx_array[i];
diff --git a/sql/ha_partition.h b/sql/ha_partition.h
index fbf34efe39f..9cbb6c928ee 100644
--- a/sql/ha_partition.h
+++ b/sql/ha_partition.h
@@ -1608,9 +1608,6 @@ public:
 
   /**
      Get the number of records in part_elem and its subpartitions, if any.
-     Also sets read_partitions bit for each partition id it uses (that is needed
-     for vers_set_hist_part() because it is called before read_partitions bitmap
-     is initialized).
   */
   ha_rows part_records(partition_element *part_elem)
   {
@@ -1623,7 +1620,6 @@ public:
     for (; part_id < part_id_end; ++part_id)
     {
       handler *file= m_file[part_id];
-      bitmap_set_bit(&(m_part_info->read_partitions), part_id);
       file->info(HA_STATUS_VARIABLE | HA_STATUS_NO_LOCK | HA_STATUS_OPEN);
       part_recs+= file->stats.records;
     }
diff --git a/sql/handler.cc b/sql/handler.cc
index fb1f9ba8a76..2f2c7064e6a 100644
--- a/sql/handler.cc
+++ b/sql/handler.cc
@@ -4365,6 +4365,7 @@ void handler::print_error(int error, myf errflag)
       if ((int) key_nr >= 0 && key_nr < table->s->keys)
       {
         print_keydup_error(table, &table->key_info[key_nr], errflag);
+        table->file->lookup_errkey= -1;
         DBUG_VOID_RETURN;
       }
     }
@@ -5810,6 +5811,9 @@ int handler::calculate_checksum()
     for (uint i= 0; i < table->s->fields; i++ )
     {
       Field *f= table->field[i];
+      if (!f->stored_in_db())
+        continue;
+
 
       if (! (thd->variables.old_behavior & OLD_MODE_COMPAT_5_1_CHECKSUM) &&
             f->is_real_null(0))
@@ -7917,24 +7921,6 @@ int ha_abort_transaction(THD *bf_thd, THD *victim_thd, my_bool signal)
 #endif /* WITH_WSREP */
 
 
-bool HA_CREATE_INFO::check_conflicting_charset_declarations(CHARSET_INFO *cs)
-{
-  if ((used_fields & HA_CREATE_USED_DEFAULT_CHARSET) &&
-      /* DEFAULT vs explicit, or explicit vs DEFAULT */
-      (((default_table_charset == NULL) != (cs == NULL)) ||
-      /* Two different explicit character sets */
-       (default_table_charset && cs &&
-        !my_charset_same(default_table_charset, cs))))
-  {
-    my_error(ER_CONFLICTING_DECLARATIONS, MYF(0),
-             "CHARACTER SET ", default_table_charset ?
-                               default_table_charset->cs_name.str : "DEFAULT",
-             "CHARACTER SET ", cs ? cs->cs_name.str : "DEFAULT");
-    return true;
-  }
-  return false;
-}
-
 /* Remove all indexes for a given table from global index statistics */
 
 static
diff --git a/sql/handler.h b/sql/handler.h
index cdab18cecbc..60c6195e68e 100644
--- a/sql/handler.h
+++ b/sql/handler.h
@@ -967,6 +967,7 @@ typedef struct xid_t XID;
 */
 typedef uint Binlog_file_id;
 const Binlog_file_id MAX_binlog_id= UINT_MAX;
+const my_off_t       MAX_off_t    = (~(my_off_t) 0);
 /*
   Compound binlog-id and byte offset of transaction's first event
   in a sequence (e.g the recovery sequence) of binlog files.
@@ -981,14 +982,22 @@ struct xid_recovery_member
   my_xid xid;
   uint in_engine_prepare;  // number of engines that have xid prepared
   bool decided_to_commit;
-  Binlog_offset binlog_coord; // semisync recovery binlog offset
+  /*
+    Semisync recovery binlog offset. It's initialized with the maximum
+    unreachable offset. The max value will remain for any transaction
+    not found in binlog to yield its rollback decision as it's guaranteed
+    to be within a truncated tail part of the binlog.
+  */
+  Binlog_offset binlog_coord;
   XID *full_xid;           // needed by wsrep or past it recovery
   decltype(::server_id) server_id;         // server id of orginal server
 
   xid_recovery_member(my_xid xid_arg, uint prepare_arg, bool decided_arg,
                       XID *full_xid_arg, decltype(::server_id) server_id_arg)
     : xid(xid_arg), in_engine_prepare(prepare_arg),
-      decided_to_commit(decided_arg), full_xid(full_xid_arg) , server_id(server_id_arg) {};
+      decided_to_commit(decided_arg),
+      binlog_coord(Binlog_offset(MAX_binlog_id, MAX_off_t)),
+      full_xid(full_xid_arg), server_id(server_id_arg) {};
 };
 
 /* for recover() handlerton call */
@@ -2280,33 +2289,6 @@ struct HA_CREATE_INFO: public Table_scope_and_contents_source_st,
     Schema_specification_st::init();
     alter_info= NULL;
   }
-  bool check_conflicting_charset_declarations(CHARSET_INFO *cs);
-  bool add_table_option_default_charset(CHARSET_INFO *cs)
-  {
-    // cs can be NULL, e.g.:  CREATE TABLE t1 (..) CHARACTER SET DEFAULT;
-    if (check_conflicting_charset_declarations(cs))
-      return true;
-    default_table_charset= cs;
-    used_fields|= HA_CREATE_USED_DEFAULT_CHARSET;
-    return false;
-  }
-  bool add_alter_list_item_convert_to_charset(CHARSET_INFO *cs)
-  {
-    /* 
-      cs cannot be NULL, as sql_yacc.yy translates
-         CONVERT TO CHARACTER SET DEFAULT
-      to
-         CONVERT TO CHARACTER SET <character-set-of-the-current-database>
-      TODO: Shouldn't we postpone resolution of DEFAULT until the
-      character set of the table owner database is loaded from its db.opt?
-    */
-    DBUG_ASSERT(cs);
-    if (check_conflicting_charset_declarations(cs))
-      return true;
-    alter_table_convert_to_charset= default_table_charset= cs;
-    used_fields|= (HA_CREATE_USED_CHARSET | HA_CREATE_USED_DEFAULT_CHARSET);  
-    return false;
-  }
   ulong table_options_with_row_type()
   {
     if (row_type == ROW_TYPE_DYNAMIC || row_type == ROW_TYPE_PAGE)
@@ -2314,6 +2296,10 @@ struct HA_CREATE_INFO: public Table_scope_and_contents_source_st,
     else
       return table_options;
   }
+  bool resolve_to_charset_collation_context(THD *thd,
+                  const Lex_table_charset_collation_attrs_st &default_cscl,
+                  const Lex_table_charset_collation_attrs_st &convert_cscl,
+                  const Charset_collation_context &ctx);
 };
 
 
@@ -2324,16 +2310,23 @@ struct HA_CREATE_INFO: public Table_scope_and_contents_source_st,
 struct Table_specification_st: public HA_CREATE_INFO,
                                public DDL_options_st
 {
+  Lex_table_charset_collation_attrs_st default_charset_collation;
+  Lex_table_charset_collation_attrs_st convert_charset_collation;
+
   // Deep initialization
   void init()
   {
     HA_CREATE_INFO::init();
     DDL_options_st::init();
+    default_charset_collation.init();
+    convert_charset_collation.init();
   }
   void init(DDL_options_st::Options options_arg)
   {
     HA_CREATE_INFO::init();
     DDL_options_st::init(options_arg);
+    default_charset_collation.init();
+    convert_charset_collation.init();
   }
   /*
     Quick initialization, for parser.
@@ -2345,6 +2338,46 @@ struct Table_specification_st: public HA_CREATE_INFO,
   {
     HA_CREATE_INFO::options= 0;
     DDL_options_st::init();
+    default_charset_collation.init();
+    convert_charset_collation.init();
+  }
+
+  bool add_table_option_convert_charset(CHARSET_INFO *cs)
+  {
+    // cs can be NULL, e.g.: ALTER TABLE t1 CONVERT TO CHARACTER SET DEFAULT;
+    used_fields|= (HA_CREATE_USED_CHARSET | HA_CREATE_USED_DEFAULT_CHARSET);
+    return cs ?
+      convert_charset_collation.merge_exact_charset(Lex_exact_charset(cs)) :
+      convert_charset_collation.merge_charset_default();
+  }
+  bool add_table_option_convert_collation(const Lex_extended_collation_st &cl)
+  {
+    used_fields|= (HA_CREATE_USED_CHARSET | HA_CREATE_USED_DEFAULT_CHARSET);
+    return convert_charset_collation.merge_collation(cl);
+  }
+
+  bool add_table_option_default_charset(CHARSET_INFO *cs)
+  {
+    // cs can be NULL, e.g.:  CREATE TABLE t1 (..) CHARACTER SET DEFAULT;
+    used_fields|= HA_CREATE_USED_DEFAULT_CHARSET;
+    return cs ?
+      default_charset_collation.merge_exact_charset(Lex_exact_charset(cs)) :
+      default_charset_collation.merge_charset_default();
+  }
+  bool add_table_option_default_collation(const Lex_extended_collation_st &cl)
+  {
+    used_fields|= HA_CREATE_USED_DEFAULT_CHARSET;
+    return default_charset_collation.merge_collation(cl);
+  }
+
+  bool resolve_to_charset_collation_context(THD *thd,
+                                         const Charset_collation_context &ctx)
+  {
+    return HA_CREATE_INFO::
+             resolve_to_charset_collation_context(thd,
+                                                  default_charset_collation,
+                                                  convert_charset_collation,
+                                                  ctx);
   }
 };
 
diff --git a/sql/item.cc b/sql/item.cc
index 3b8345f4416..6c5a892fc96 100644
--- a/sql/item.cc
+++ b/sql/item.cc
@@ -959,8 +959,7 @@ bool Item_field::check_field_expression_processor(void *arg)
           (!field->vcol_info && !org_field->vcol_info)) &&
          field->field_index >= org_field->field_index))
     {
-      my_error(ER_EXPRESSION_REFERS_TO_UNINIT_FIELD,
-               MYF(0),
+      my_error(ER_EXPRESSION_REFERS_TO_UNINIT_FIELD, MYF(0),
                org_field->field_name.str, field->field_name.str);
       return 1;
     }
@@ -7998,8 +7997,7 @@ bool Item_ref::fix_fields(THD *thd, Item **reference)
   else if (!ref || ref == not_found_item)
   {
     DBUG_ASSERT(reference_trough_name != 0);
-    if (!(ref= resolve_ref_in_select_and_group(thd, this,
-                                               context->select_lex)))
+    if (!(ref= resolve_ref_in_select_and_group(thd, this, context->select_lex)))
       goto error;             /* Some error occurred (e.g. ambiguous names). */
 
     if (ref == not_found_item) /* This reference was not resolved. */
@@ -8012,8 +8010,7 @@ bool Item_ref::fix_fields(THD *thd, Item **reference)
       if (unlikely(!outer_context))
       {
         /* The current reference cannot be resolved in this query. */
-        my_error(ER_BAD_FIELD_ERROR,MYF(0),
-                 this->full_name(), thd->where);
+        my_error(ER_BAD_FIELD_ERROR,MYF(0), full_name(), thd->where);
         goto error;
       }
 
@@ -9505,6 +9502,12 @@ bool Item_default_value::eq(const Item *item, bool binary_cmp) const
 }
 
 
+bool Item_default_value::check_field_expression_processor(void *)
+{
+  field->default_value= ((Item_field *)(arg->real_item()))->field->default_value;
+  return 0;
+}
+
 bool Item_default_value::fix_fields(THD *thd, Item **items)
 {
   Item *real_arg;
@@ -9546,7 +9549,6 @@ bool Item_default_value::fix_fields(THD *thd, Item **items)
   }
   if (!(def_field= (Field*) thd->alloc(field_arg->field->size_of())))
     goto error;
-  cached_field= def_field;
   memcpy((void *)def_field, (void *)field_arg->field,
          field_arg->field->size_of());
   def_field->reset_fields();
@@ -9575,8 +9577,7 @@ error:
 
 void Item_default_value::cleanup()
 {
-  delete cached_field;                        // Free cached blob data
-  cached_field= 0;
+  delete field;                        // Free cached blob data
   Item_field::cleanup();
 }
 
@@ -9651,6 +9652,12 @@ int Item_default_value::save_in_field(Field *field_arg, bool no_conversions)
   return Item_field::save_in_field(field_arg, no_conversions);
 }
 
+void Item_default_value::save_in_result_field(bool no_conversions)
+{
+  calculate();
+  Item_field::save_in_result_field(no_conversions);
+}
+
 double Item_default_value::val_result()
 {
   calculate();
@@ -9710,6 +9717,23 @@ table_map Item_default_value::used_tables() const
   return field->default_value->expr->used_tables();
 }
 
+bool Item_default_value::register_field_in_read_map(void *arg)
+{
+  TABLE *table= (TABLE *) arg;
+  int res= 0;
+  if (!table || (table && table == field->table))
+  {
+    if (field->default_value && field->default_value->expr)
+      res= field->default_value->expr->walk(&Item::register_field_in_read_map,1,arg);
+  }
+  else if (result_field && table == result_field->table)
+  {
+    bitmap_set_bit(table->read_set, result_field->field_index);
+  }
+
+  return res;
+}
+
 /**
   This method like the walk method traverses the item tree, but at the
   same time it can replace some nodes in the tree.
diff --git a/sql/item.h b/sql/item.h
index 4ef0e1b93b2..d1d19157b83 100644
--- a/sql/item.h
+++ b/sql/item.h
@@ -2177,6 +2177,12 @@ public:
     return 0;
   }
 
+  virtual bool set_extraction_flag_processor(void *arg)
+  {
+    set_extraction_flag(*(int16*)arg);
+    return 0;
+  }
+
   /**
     Check db/table_name if they defined in item and match arg values
 
@@ -6613,7 +6619,6 @@ class Item_default_value : public Item_field
   void calculate();
 public:
   Item *arg= nullptr;
-  Field *cached_field= nullptr;
   Item_default_value(THD *thd, Name_resolution_context *context_arg, Item *a,
                      bool vcol_assignment_arg)
     : Item_field(thd, context_arg),
@@ -6630,6 +6635,10 @@ public:
   bool get_date(THD *thd, MYSQL_TIME *ltime,date_mode_t fuzzydate) override;
   bool val_native(THD *thd, Native *to) override;
   bool val_native_result(THD *thd, Native *to) override;
+  longlong val_datetime_packed(THD *thd) override
+  { return Item::val_datetime_packed(thd); }
+  longlong val_time_packed(THD *thd) override
+  { return Item::val_time_packed(thd); }
 
   /* Result variants */
   double val_result() override;
@@ -6643,6 +6652,7 @@ public:
 
   bool send(Protocol *protocol, st_value *buffer) override;
   int save_in_field(Field *field_arg, bool no_conversions) override;
+  void save_in_result_field(bool no_conversions) override;
   bool save_in_param(THD *, Item_param *param) override
   {
     // It should not be possible to have "EXECUTE .. USING DEFAULT(a)"
@@ -6658,11 +6668,12 @@ public:
   }
   bool vcol_assignment_allowed_value() const override
   { return vcol_assignment_ok; }
-  Field *get_tmp_table_field() override { return nullptr; }
   Item *get_tmp_table_item(THD *) override { return this; }
   Item_field *field_for_view_update() override { return nullptr; }
   bool update_vcol_processor(void *) override { return false; }
+  bool check_field_expression_processor(void *arg) override;
   bool check_func_default_processor(void *) override { return true; }
+  bool register_field_in_read_map(void *arg) override;
   bool walk(Item_processor processor, bool walk_subquery, void *args) override
   {
     return (arg && arg->walk(processor, walk_subquery, args)) ||
@@ -6941,7 +6952,7 @@ public:
   for any value.
 */
 
-class Item_cache: public Item,
+class Item_cache: public Item_fixed_hybrid,
                   public Type_handler_hybrid_field_type
 {
 protected:
@@ -6972,7 +6983,7 @@ public:
   bool null_value_inside;
 
   Item_cache(THD *thd):
-    Item(thd),
+    Item_fixed_hybrid(thd),
     Type_handler_hybrid_field_type(&type_handler_string),
     example(0), cached_field(0),
     value_cached(0),
@@ -6981,10 +6992,11 @@ public:
     set_maybe_null();
     null_value= 1;
     null_value_inside= true;
+    quick_fix_field();
   }
 protected:
   Item_cache(THD *thd, const Type_handler *handler):
-    Item(thd),
+    Item_fixed_hybrid(thd),
     Type_handler_hybrid_field_type(handler),
     example(0), cached_field(0),
     value_cached(0),
@@ -6993,6 +7005,7 @@ protected:
     set_maybe_null();
     null_value= 1;
     null_value_inside= true;
+    quick_fix_field();
   }
 
 public:
@@ -7045,10 +7058,17 @@ public:
     }
     return mark_unsupported_function("cache", arg, VCOL_IMPOSSIBLE);
   }
+  bool fix_fields(THD *thd, Item **ref) override
+  {
+    quick_fix_field();
+    if (example && !example->fixed())
+      return example->fix_fields(thd, ref);
+    return 0;
+  }
   void cleanup() override
   {
     clear();
-    Item::cleanup();
+    Item_fixed_hybrid::cleanup();
   }
   /**
      Check if saved item has a non-NULL value.
diff --git a/sql/item_cmpfunc.cc b/sql/item_cmpfunc.cc
index dbfbc07a600..a53bb2e53b6 100644
--- a/sql/item_cmpfunc.cc
+++ b/sql/item_cmpfunc.cc
@@ -7663,7 +7663,17 @@ bool Item_equal::create_pushable_equalities(THD *thd,
     if (!eq ||  equalities->push_back(eq, thd->mem_root))
       return true;
     if (!clone_const)
-      right_item->set_extraction_flag(MARKER_IMMUTABLE);
+    {
+      /*
+        Also set IMMUTABLE_FL for any sub-items of the right_item.
+        This is needed to prevent Item::cleanup_excluding_immutables_processor
+        from peforming cleanup of the sub-items and so creating an item tree
+        where a fixed item has non-fixed items inside it.
+      */
+      int16 new_flag= MARKER_IMMUTABLE;
+      right_item->walk(&Item::set_extraction_flag_processor, false,
+                       (void*)&new_flag);
+    }
   }
 
   while ((item=it++))
diff --git a/sql/item_func.cc b/sql/item_func.cc
index f25dffdfd02..e53b89a9291 100644
--- a/sql/item_func.cc
+++ b/sql/item_func.cc
@@ -2501,7 +2501,7 @@ void Item_func_round::fix_arg_decimal()
     set_handler(&type_handler_newdecimal);
     unsigned_flag= args[0]->unsigned_flag;
     decimals= args[0]->decimals;
-    max_length= float_length(args[0]->decimals) + 1;
+    max_length= args[0]->max_length;
   }
 }
 
diff --git a/sql/item_func.h b/sql/item_func.h
index 185242b7901..a4122f78e87 100644
--- a/sql/item_func.h
+++ b/sql/item_func.h
@@ -3773,7 +3773,7 @@ public:
   }
   bool set(const Type_handler *handler,
            const Lex_length_and_dec_st & length_and_dec,
-           const Lex_charset_collation_st &cscl,
+           const Lex_column_charset_collation_attrs_st &cscl,
            CHARSET_INFO *defcs)
   {
     CHARSET_INFO *tmp= cscl.resolved_to_character_set(defcs);
diff --git a/sql/item_subselect.cc b/sql/item_subselect.cc
index de31c82abf4..2169ac874e5 100644
--- a/sql/item_subselect.cc
+++ b/sql/item_subselect.cc
@@ -826,6 +826,7 @@ bool Item_subselect::exec()
   DBUG_ENTER("Item_subselect::exec");
   DBUG_ASSERT(fixed());
   DBUG_ASSERT(thd);
+  DBUG_ASSERT(!eliminated);
 
   DBUG_EXECUTE_IF("Item_subselect",
     Item::Print print(this,
@@ -1361,11 +1362,18 @@ bool Item_singlerow_subselect::fix_length_and_dec()
   }
   unsigned_flag= value->unsigned_flag;
   /*
-    If there are not tables in subquery then ability to have NULL value
-    depends on SELECT list (if single row subquery have tables then it
-    always can be NULL if there are not records fetched).
+    If the subquery has no tables (1) and is not a UNION (2), like:
+
+      (SELECT subq_value)
+
+    then its NULLability is the same as subq_value's NULLability.
+
+    (1): A subquery that uses a table will return NULL when the table is empty.
+    (2): A UNION subquery will return NULL if it produces a "Subquery returns
+         more than one row" error.
   */
-  if (engine->no_tables())
+  if (engine->no_tables() &&
+      engine->engine_type() != subselect_engine::UNION_ENGINE)
     set_maybe_null(engine->may_be_null());
   else
   {
@@ -1401,6 +1409,16 @@ Item* Item_singlerow_subselect::expr_cache_insert_transformer(THD *tmp_thd,
 
   DBUG_ASSERT(thd == tmp_thd);
 
+  /*
+    Do not create subquery cache if the subquery was eliminated.
+    The optimizer may eliminate subquery items (see
+    eliminate_subselect_processor). However it does not update
+    all query's data structures, so the eliminated item may be
+    still reachable.
+  */
+  if (eliminated)
+    DBUG_RETURN(this);
+
   if (expr_cache)
     DBUG_RETURN(expr_cache);
 
diff --git a/sql/item_sum.cc b/sql/item_sum.cc
index 109a0f883e8..4f79c8f647d 100644
--- a/sql/item_sum.cc
+++ b/sql/item_sum.cc
@@ -1559,6 +1559,8 @@ void Item_sum_sum::fix_length_and_dec_decimal()
   decimals= args[0]->decimals;
   /* SUM result can't be longer than length(arg) + length(MAX_ROWS) */
   int precision= args[0]->decimal_precision() + DECIMAL_LONGLONG_DIGITS;
+  decimals= MY_MIN(decimals, DECIMAL_MAX_SCALE);
+  precision= MY_MIN(precision, DECIMAL_MAX_PRECISION);
   max_length= my_decimal_precision_to_length_no_truncation(precision,
                                                            decimals,
                                                            unsigned_flag);
@@ -1970,12 +1972,12 @@ void Item_sum_avg::fix_length_and_dec_decimal()
 {
   Item_sum_sum::fix_length_and_dec_decimal();
   int precision= args[0]->decimal_precision() + prec_increment;
-  decimals= MY_MIN(args[0]->decimals + prec_increment, DECIMAL_MAX_SCALE);
+  decimals= MY_MIN(args[0]->decimal_scale() + prec_increment, DECIMAL_MAX_SCALE);
   max_length= my_decimal_precision_to_length_no_truncation(precision,
                                                            decimals,
                                                            unsigned_flag);
   f_precision= MY_MIN(precision+DECIMAL_LONGLONG_DIGITS, DECIMAL_MAX_PRECISION);
-  f_scale=  args[0]->decimals;
+  f_scale= args[0]->decimal_scale();
   dec_bin_size= my_decimal_get_binary_size(f_precision, f_scale);
 }
 
@@ -4261,9 +4263,9 @@ Item_func_group_concat::fix_fields(THD *thd, Item **ref)
   result.set_charset(collation.collation);
   result_field= 0;
   null_value= 1;
-  max_length= (uint32)MY_MIN(thd->variables.group_concat_max_len
-                             / collation.collation->mbminlen
-                             * collation.collation->mbmaxlen, UINT_MAX32);
+  max_length= (uint32) MY_MIN((ulonglong) thd->variables.group_concat_max_len
+                              / collation.collation->mbminlen
+                              * collation.collation->mbmaxlen, UINT_MAX32);
 
   uint32 offset;
   if (separator->needs_conversion(separator->length(), separator->charset(),
diff --git a/sql/json_table.cc b/sql/json_table.cc
index e2f90aa2a9a..a9e882291b6 100644
--- a/sql/json_table.cc
+++ b/sql/json_table.cc
@@ -874,7 +874,7 @@ int Json_table_column::set(THD *thd, enum_type ctype, const LEX_CSTRING &path,
 
 
 int Json_table_column::set(THD *thd, enum_type ctype, const LEX_CSTRING &path,
-                           const Lex_charset_collation_st &cl)
+                           const Lex_column_charset_collation_attrs_st &cl)
 {
   if (cl.is_empty() || cl.is_contextually_typed_collate_default())
     return set(thd, ctype, path, nullptr);
diff --git a/sql/json_table.h b/sql/json_table.h
index 2cadb07961e..6398b061889 100644
--- a/sql/json_table.h
+++ b/sql/json_table.h
@@ -161,7 +161,7 @@ public:
   }
   int set(THD *thd, enum_type ctype, const LEX_CSTRING &path, CHARSET_INFO *cs);
   int set(THD *thd, enum_type ctype, const LEX_CSTRING &path,
-          const Lex_charset_collation_st &cl);
+          const Lex_column_charset_collation_attrs_st &cl);
   Json_table_column(Create_field *f, Json_table_nested_path *nest) :
     m_field(f), m_nest(nest), m_explicit_cs(NULL)
   {
diff --git a/sql/lex_charset.cc b/sql/lex_charset.cc
index 7570abbdfb9..c36e01498c5 100644
--- a/sql/lex_charset.cc
+++ b/sql/lex_charset.cc
@@ -21,47 +21,333 @@
 #include "mysqld_error.h"
 
 
+static void
+raise_ER_CONFLICTING_DECLARATIONS(const char *clause1,
+                                  const char *name1,
+                                  const char *clause2,
+                                  const char *name2,
+                                  bool reverse_order)
+{
+  if (!reverse_order)
+    my_error(ER_CONFLICTING_DECLARATIONS, MYF(0),
+             clause1, name1, clause2, name2);
+  else
+    my_error(ER_CONFLICTING_DECLARATIONS, MYF(0),
+             clause2, name2, clause1, name1);
+}
+
+
+static void
+raise_ER_CONFLICTING_DECLARATIONS(const char *clause1,
+                                  const char *name1,
+                                  const char *name1_part2,
+                                  const char *clause2,
+                                  const char *name2,
+                                  bool reverse_order)
+{
+  char def[MY_CS_NAME_SIZE * 2];
+  my_snprintf(def, sizeof(def), "%s (%s)", name1, name1_part2);
+  raise_ER_CONFLICTING_DECLARATIONS(clause1, def,
+                                    clause2, name2,
+                                    reverse_order);
+}
+
+
+bool Lex_exact_charset::raise_if_not_equal(const Lex_exact_charset &rhs) const
+{
+  if (m_ci == rhs.m_ci)
+    return false;
+  my_error(ER_CONFLICTING_DECLARATIONS, MYF(0),
+           "CHARACTER SET ", m_ci->cs_name.str,
+           "CHARACTER SET ", rhs.m_ci->cs_name.str);
+  return true;
+}
+
+
+bool Lex_exact_charset::
+  raise_if_not_applicable(const Lex_exact_collation &cl) const
+{
+  return Lex_exact_charset_opt_extended_collate(m_ci, false).
+           raise_if_not_applicable(cl);
+}
+
+
+bool Lex_exact_charset_opt_extended_collate::
+  raise_if_not_applicable(const Lex_exact_collation &cl) const
+{
+  if (!my_charset_same(m_ci, cl.charset_info()))
+  {
+    my_error(ER_COLLATION_CHARSET_MISMATCH, MYF(0),
+             cl.charset_info()->coll_name.str, m_ci->cs_name.str);
+    return true;
+  }
+  return false;
+}
+
+
+bool
+Lex_exact_collation::raise_if_not_equal(const Lex_exact_collation &cl) const
+{
+  if (m_ci != cl.m_ci)
+  {
+    my_error(ER_CONFLICTING_DECLARATIONS, MYF(0),
+             "COLLATE ", m_ci->coll_name.str,
+             "COLLATE ", cl.m_ci->coll_name.str);
+    return true;
+  }
+  return false;
+}
+
+
+/*
+  Merge an exact collation and a contexual collation.
+  @param cl            - The contextual collation to merge to "this".
+  @param reverse_order - If the contextual collation is on the left side
+
+  Use reverse_order as follows:
+  false:  COLLATE latin1_swedish_ci COLLATE DEFAULT
+  true:   COLLATE DEFAULT COLLATE latin1_swedish_ci
+*/
+bool
+Lex_exact_collation::
+  raise_if_conflicts_with_context_collation(const Lex_context_collation &cl,
+                                            bool reverse_order) const
+{
+  if (cl.is_contextually_typed_collate_default() &&
+      !(m_ci->state & MY_CS_PRIMARY))
+  {
+    raise_ER_CONFLICTING_DECLARATIONS("COLLATE ", m_ci->coll_name.str,
+                                      "COLLATE ", "DEFAULT", reverse_order);
+    return true;
+  }
+
+  if (cl.is_contextually_typed_binary_style() &&
+      !(m_ci->state & MY_CS_BINSORT))
+  {
+    raise_ER_CONFLICTING_DECLARATIONS("COLLATE ", m_ci->coll_name.str,
+                                      "", "BINARY", reverse_order);
+    return true;
+  }
+  return false;
+}
+
+
+bool
+Lex_context_collation::raise_if_not_equal(const Lex_context_collation &cl) const
+{
+  /*
+    Only equal context collations are possible here so far:
+    - Column grammar only supports BINARY, but does not support COLLATE DEFAULT
+    - DB/Table grammar only support COLLATE DEFAULT
+    But we'll have different collations here - uca140 is coming soon.
+  */
+  DBUG_ASSERT(m_ci == cl.m_ci);
+  return false;
+}
+
+
+/*
+  Resolve a context collation to the character set (when the former gets known):
+    CREATE TABLE t1 (a CHAR(10) BINARY) CHARACTER SET latin1;
+    CREATE DATABASE db1 COLLATE DEFAULT CHARACTER SET latin1;
+*/
+bool Lex_exact_charset_opt_extended_collate::
+  merge_context_collation_override(const Lex_context_collation &cl)
+{
+  DBUG_ASSERT(m_ci);
+
+  // CHAR(10) BINARY
+  if (cl.is_contextually_typed_binary_style())
+  {
+    CHARSET_INFO *ci= find_bin_collation();
+    if (!ci)
+      return true;
+    m_ci= ci;
+    m_with_collate= true;
+    return false;
+  }
+
+  // COLLATE DEFAULT
+  if (cl.is_contextually_typed_collate_default())
+  {
+    CHARSET_INFO *ci= find_default_collation();
+    DBUG_ASSERT(ci);
+    if (!ci)
+      return true;
+    m_ci= ci;
+    m_with_collate= true;
+    return false;
+  }
+
+  /*
+    A non-binary and non-default contextually typed collation.
+    We don't have such yet - the parser cannot produce this.
+    But we have "uca1400_as_ci" coming soon.
+  */
+  DBUG_ASSERT(0);
+  return false;
+}
+
+
+bool Lex_extended_collation_st::merge_exact_charset(const Lex_exact_charset &cs)
+{
+  switch (m_type) {
+  case TYPE_EXACT:
+    {
+      // COLLATE latin1_swedish_ci .. CHARACTER SET latin1
+      return cs.raise_if_not_applicable(Lex_exact_collation(m_ci));
+    }
+  case TYPE_CONTEXTUALLY_TYPED:
+    {
+      // COLLATE DEFAULT .. CHARACTER SET latin1
+      Lex_exact_charset_opt_extended_collate tmp(cs);
+      if (tmp.merge_context_collation(Lex_context_collation(m_ci)))
+        return true;
+      *this= Lex_extended_collation(tmp.collation());
+      return false;
+    }
+  }
+  DBUG_ASSERT(0);
+  return false;
+}
+
+
+bool Lex_extended_collation_st::
+       merge_exact_collation(const Lex_exact_collation &rhs)
+{
+  switch (m_type) {
+
+  case TYPE_EXACT:
+    /*
+      EXACT + EXACT
+      COLLATE latin1_bin .. COLLATE latin1_bin
+    */
+    return Lex_exact_collation(m_ci).raise_if_not_equal(rhs);
+
+  case TYPE_CONTEXTUALLY_TYPED:
+    {
+      /*
+        CONTEXT + EXACT
+        CHAR(10) COLLATE DEFAULT       .. COLLATE latin1_swedish_ci
+        CHAR(10) BINARY                .. COLLATE latin1_bin
+        CHAR(10) COLLATE uca1400_as_ci .. COLLATE latin1_bin - coming soon
+      */
+      if (rhs.raise_if_conflicts_with_context_collation(
+                Lex_context_collation(m_ci), true))
+        return true;
+      *this= Lex_extended_collation(rhs);
+      return false;
+    }
+  }
+  DBUG_ASSERT(0);
+  return false;
+}
+
+
+bool Lex_extended_collation_st::
+  raise_if_conflicts_with_context_collation(const Lex_context_collation &rhs)
+                                                                        const
+{
+  switch (m_type) {
+
+  case TYPE_EXACT:
+    /*
+      EXACT + CONTEXT
+      COLLATE latin1_swedish_ci .. COLLATE DEFAULT
+    */
+    return Lex_exact_collation(m_ci).
+             raise_if_conflicts_with_context_collation(rhs, false);
+
+  case TYPE_CONTEXTUALLY_TYPED:
+    {
+      /*
+        CONTEXT + CONTEXT:
+        CHAR(10) BINARY .. COLLATE DEFAULT - not supported by the parser
+        CREATE DATABASE db1 COLLATE DEFAULT COLLATE DEFAULT;
+      */
+      return Lex_context_collation(m_ci).raise_if_not_equal(rhs);
+    }
+  }
+  DBUG_ASSERT(0);
+  return false;
+}
+
+
+/*
+  Merge two non-empty COLLATE clauses.
+*/
+bool Lex_extended_collation_st::merge(const Lex_extended_collation_st &rhs)
+{
+  switch (rhs.type()) {
+  case TYPE_EXACT:
+    /*
+      EXACT + EXACT
+      COLLATE latin1_swedish_ci .. COLLATE latin1_swedish_ci
+
+      CONTEXT + EXACT
+      COLLATE DEFAULT           .. COLLATE latin1_swedish_ci
+      CHAR(10) BINARY           .. COLLATE latin1_bin
+    */
+    return merge_exact_collation(Lex_exact_collation(rhs.m_ci));
+  case TYPE_CONTEXTUALLY_TYPED:
+    /*
+       EXACT + CONTEXT
+       COLLATE latin1_swedish_ci .. COLLATE DEFAULT
+
+       CONTEXT + CONTEXT
+       COLLATE DEFAULT           .. COLLATE DEFAULT
+       CHAR(10) BINARY           .. COLLATE DEFAULT
+    */
+    return raise_if_conflicts_with_context_collation(
+             Lex_context_collation(rhs.m_ci));
+  }
+  DBUG_ASSERT(0);
+  return false;
+}
+
+
 /** find a collation with binary comparison rules
 */
-CHARSET_INFO *Lex_charset_collation_st::find_bin_collation(CHARSET_INFO *cs)
+CHARSET_INFO *Lex_exact_charset_opt_extended_collate::find_bin_collation() const
 {
   /*
     We don't need to handle old_mode=UTF8_IS_UTF8MB3 here,
-    because "cs" points to a real character set name.
+    because "m_ci" points to a real character set name.
     It can be either "utf8mb3" or "utf8mb4". It cannot be "utf8".
     No thd->get_utf8_flag() flag passed to get_charset_by_csname().
   */
-  DBUG_ASSERT(cs->cs_name.length !=4 || memcmp(cs->cs_name.str, "utf8", 4));
+  DBUG_ASSERT(m_ci->cs_name.length !=4 || memcmp(m_ci->cs_name.str, "utf8", 4));
   /*
     CREATE TABLE t1 (a CHAR(10) BINARY)
       CHARACTER SET utf8mb4 COLLATE utf8mb4_bin;
     Nothing to do, we have the binary collation already.
   */
-  if (cs->state & MY_CS_BINSORT)
-    return cs;
+  if (m_ci->state & MY_CS_BINSORT)
+    return m_ci;
 
   // CREATE TABLE t1 (a CHAR(10) BINARY) CHARACTER SET utf8mb4;
-  const LEX_CSTRING &cs_name= cs->cs_name;
-  if (!(cs= get_charset_by_csname(cs->cs_name.str, MY_CS_BINSORT, MYF(0))))
+  CHARSET_INFO *cs;
+  if (!(cs= get_charset_by_csname(m_ci->cs_name.str, MY_CS_BINSORT, MYF(0))))
   {
     char tmp[65];
-    strxnmov(tmp, sizeof(tmp)-1, cs_name.str, "_bin", NULL);
+    strxnmov(tmp, sizeof(tmp)-1, m_ci->cs_name.str, "_bin", NULL);
     my_error(ER_UNKNOWN_COLLATION, MYF(0), tmp);
   }
   return cs;
 }
 
 
-CHARSET_INFO *Lex_charset_collation_st::find_default_collation(CHARSET_INFO *cs)
+CHARSET_INFO *
+Lex_exact_charset_opt_extended_collate::find_default_collation() const
 {
   // See comments in find_bin_collation()
-  DBUG_ASSERT(cs->cs_name.length !=4 || memcmp(cs->cs_name.str, "utf8", 4));
+  DBUG_ASSERT(m_ci->cs_name.length !=4 || memcmp(m_ci->cs_name.str, "utf8", 4));
   /*
     CREATE TABLE t1 (a CHAR(10) COLLATE DEFAULT) CHARACTER SET utf8mb4;
     Nothing to do, we have the default collation already.
   */
-  if (cs->state & MY_CS_PRIMARY)
-    return cs;
+  if (m_ci->state & MY_CS_PRIMARY)
+    return m_ci;
   /*
     CREATE TABLE t1 (a CHAR(10) COLLATE DEFAULT)
       CHARACTER SET utf8mb4 COLLATE utf8mb4_bin;
@@ -69,7 +355,8 @@ CHARSET_INFO *Lex_charset_collation_st::find_default_collation(CHARSET_INFO *cs)
     Don't need to handle old_mode=UTF8_IS_UTF8MB3 here.
     See comments in find_bin_collation.
   */
-  cs= get_charset_by_csname(cs->cs_name.str, MY_CS_PRIMARY, MYF(MY_WME));
+  CHARSET_INFO *cs= get_charset_by_csname(m_ci->cs_name.str,
+                                          MY_CS_PRIMARY, MYF(MY_WME));
   /*
     The above should never fail, as we have default collations for
     all character sets.
@@ -79,21 +366,6 @@ CHARSET_INFO *Lex_charset_collation_st::find_default_collation(CHARSET_INFO *cs)
 }
 
 
-bool Lex_charset_collation_st::set_charset_collate_exact(CHARSET_INFO *cs,
-                                                         CHARSET_INFO *cl)
-{
-  DBUG_ASSERT(cs != nullptr && cl != nullptr);
-  if (!my_charset_same(cl, cs))
-  {
-    my_error(ER_COLLATION_CHARSET_MISMATCH, MYF(0),
-             cl->coll_name.str, cs->cs_name.str);
-    return true;
-  }
-  set_collate_exact(cl);
-  return false;
-}
-
-
 /*
   Resolve an empty or a contextually typed collation according to the
   upper level default character set (and optionally a collation), e.g.:
@@ -105,8 +377,8 @@ bool Lex_charset_collation_st::set_charset_collate_exact(CHARSET_INFO *cs,
   "this" is the COLLATE clause                  (e.g. of a column)
   "def" is the upper level CHARACTER SET clause (e.g. of a table)
 */
-CHARSET_INFO *
-Lex_charset_collation_st::resolved_to_character_set(CHARSET_INFO *def) const
+CHARSET_INFO *Lex_exact_charset_extended_collation_attrs_st::
+                resolved_to_character_set(CHARSET_INFO *def) const
 {
   DBUG_ASSERT(def);
 
@@ -120,103 +392,84 @@ Lex_charset_collation_st::resolved_to_character_set(CHARSET_INFO *def) const
     DBUG_ASSERT(m_ci);
     return m_ci;
   case TYPE_COLLATE_CONTEXTUALLY_TYPED:
-    break;
+  {
+    Lex_exact_charset_opt_extended_collate tmp(def, true);
+    if (tmp.merge_context_collation_override(Lex_context_collation(m_ci)))
+      return NULL;
+    return tmp.collation().charset_info();
+  }
   }
-
-  // Contextually typed
-  DBUG_ASSERT(m_ci);
-
-  if (is_contextually_typed_binary_style())    // CHAR(10) BINARY
-    return find_bin_collation(def);
-
-  if (is_contextually_typed_collate_default()) // CHAR(10) COLLATE DEFAULT
-    return find_default_collation(def);
-
-  /*
-    Non-binary and non-default contextually typed collation.
-    We don't have such yet - the parser cannot produce this.
-    But will have soon, e.g. "uca1400_as_ci".
-  */
   DBUG_ASSERT(0);
   return NULL;
 }
 
 
-/*
-  Merge the CHARACTER SET clause to:
-  - an empty COLLATE clause
-  - an explicitly typed collation name
-  - a contextually typed collation
-
-  "this" corresponds to `CHARACTER SET xxx [BINARY]`
-  "cl" corresponds to the COLLATE clause
-*/
-bool
-Lex_charset_collation_st::
-  merge_charset_clause_and_collate_clause(const Lex_charset_collation_st &cl)
+bool Lex_exact_charset_extended_collation_attrs_st::
+       merge_exact_collation(const Lex_exact_collation &cl)
 {
-  if (cl.is_empty()) // No COLLATE clause
-    return false;
-
   switch (m_type) {
   case TYPE_EMPTY:
     /*
       No CHARACTER SET clause
       CHAR(10) NOT NULL COLLATE latin1_bin
-      CHAR(10) NOT NULL COLLATE DEFAULT
     */
-    *this= cl;
+    *this= Lex_exact_charset_extended_collation_attrs(cl);
     return false;
   case TYPE_CHARACTER_SET:
-  case TYPE_COLLATE_EXACT:
     {
-      Lex_explicit_charset_opt_collate ecs(m_ci, m_type == TYPE_COLLATE_EXACT);
-      if (ecs.merge_collate_or_error(cl))
+      // CHARACTER SET latin1 .. COLLATE latin1_swedish_ci
+      if (Lex_exact_charset(m_ci).raise_if_not_applicable(cl))
         return true;
-      set_collate_exact(ecs.charset_and_collation());
+      *this= Lex_exact_charset_extended_collation_attrs(cl);
       return false;
     }
+  case TYPE_COLLATE_EXACT:
+    {
+      // [CHARACTER SET latin1] COLLATE latin1_bin .. COLLATE latin1_bin
+      return Lex_exact_collation(m_ci).raise_if_not_equal(cl);
+    }
   case TYPE_COLLATE_CONTEXTUALLY_TYPED:
-    break;
-  }
-
-  if (is_contextually_typed_collation())
-  {
-    if (cl.is_contextually_typed_collation())
     {
-      /*
-        CONTEXT + CONTEXT:
-        CHAR(10) BINARY .. COLLATE DEFAULT - not supported by the parser
-        CHAR(10) BINARY .. COLLATE uca1400_as_ci - not supported yet
-      */
-      DBUG_ASSERT(0); // Not possible yet
+      // COLLATE DEFAULT .. COLLATE latin1_swedish_ci
+      if (cl.raise_if_conflicts_with_context_collation(
+               Lex_context_collation(m_ci), true))
+        return true;
+      *this= Lex_exact_charset_extended_collation_attrs(cl);
       return false;
     }
+  }
+  DBUG_ASSERT(0);
+  return false;
+}
+
 
+bool Lex_exact_charset_extended_collation_attrs_st::
+       merge_context_collation(const Lex_context_collation &cl)
+{
+  switch (m_type) {
+  case TYPE_EMPTY:
     /*
-      CONTEXT + EXPLICIT
-      CHAR(10) COLLATE DEFAULT       .. COLLATE latin1_swedish_ci
-      CHAR(10) BINARY                .. COLLATE latin1_bin
-      CHAR(10) COLLATE uca1400_as_ci .. COLLATE latin1_bin
+      No CHARACTER SET clause
+      CHAR(10) NOT NULL .. COLLATE DEFAULT
     */
-    if (is_contextually_typed_collate_default() &&
-        !(cl.charset_collation()->state & MY_CS_PRIMARY))
-    {
-      my_error(ER_CONFLICTING_DECLARATIONS, MYF(0),
-               "COLLATE ", "DEFAULT", "COLLATE ",
-               cl.charset_collation()->coll_name.str);
-      return true;
-    }
-
-    if (is_contextually_typed_binary_style() &&
-        !(cl.charset_collation()->state & MY_CS_BINSORT))
+    *this= Lex_exact_charset_extended_collation_attrs(cl);
+    return false;
+  case TYPE_CHARACTER_SET:
     {
-      my_error(ER_CONFLICTING_DECLARATIONS, MYF(0),
-               "", "BINARY", "COLLATE ", cl.charset_collation()->coll_name.str);
-      return true;
+      // CHARACTER SET latin1 .. COLLATE DEFAULT
+      Lex_exact_charset_opt_extended_collate tmp(m_ci, false);
+      if (tmp.merge_context_collation(cl))
+        return true;
+      *this= Lex_exact_charset_extended_collation_attrs(tmp.collation());
+      return false;
     }
-    *this= cl;
-    return false;
+  case TYPE_COLLATE_EXACT:
+    // [CHARACTER SET latin1] COLLATE latin1_swedish_ci .. COLLATE DEFAULT
+    return Lex_exact_collation(m_ci).
+             raise_if_conflicts_with_context_collation(cl, false);
+  case TYPE_COLLATE_CONTEXTUALLY_TYPED:
+    // COLLATE DEFAULT .. COLLATE DEFAULT
+    return Lex_context_collation(m_ci).raise_if_not_equal(cl);
   }
 
   DBUG_ASSERT(0);
@@ -224,69 +477,38 @@ Lex_charset_collation_st::
 }
 
 
-bool
-Lex_explicit_charset_opt_collate::
-  merge_collate_or_error(const Lex_charset_collation_st &cl)
+bool Lex_exact_charset_opt_extended_collate::
+       merge_exact_collation(const Lex_exact_collation &cl)
 {
-  DBUG_ASSERT(cl.type() != Lex_charset_collation_st::TYPE_CHARACTER_SET);
+ // CHARACTER SET latin1 [COLLATE latin1_bin] .. COLLATE latin1_bin
+  if (m_with_collate)
+    return Lex_exact_collation(m_ci).raise_if_not_equal(cl);
+  if (raise_if_not_applicable(cl))
+    return true;
+  *this= Lex_exact_charset_opt_extended_collate(cl);
+  return false;
+}
 
-  switch (cl.type()) {
-  case Lex_charset_collation_st::TYPE_EMPTY:
-    return false;
-  case Lex_charset_collation_st::TYPE_CHARACTER_SET:
-    DBUG_ASSERT(0);
-    return false;
-  case Lex_charset_collation_st::TYPE_COLLATE_EXACT:
-    /*
-      EXPLICIT + EXPLICIT
-      CHAR(10) CHARACTER SET latin1                    .. COLLATE latin1_bin
-      CHAR(10) CHARACTER SET latin1 COLLATE latin1_bin .. COLLATE latin1_bin
-      CHAR(10) COLLATE latin1_bin                      .. COLLATE latin1_bin
-      CHAR(10) COLLATE latin1_bin                      .. COLLATE latin1_bin
-      CHAR(10) CHARACTER SET latin1 BINARY             .. COLLATE latin1_bin
-    */
-    if (m_with_collate && m_ci != cl.charset_collation())
-    {
-      my_error(ER_CONFLICTING_DECLARATIONS, MYF(0),
-               "COLLATE ", m_ci->coll_name.str,
-               "COLLATE ", cl.charset_collation()->coll_name.str);
-      return true;
-    }
-    if (!my_charset_same(m_ci, cl.charset_collation()))
-    {
-      my_error(ER_COLLATION_CHARSET_MISMATCH, MYF(0),
-               cl.charset_collation()->coll_name.str, m_ci->cs_name.str);
-      return true;
-    }
-    m_ci= cl.charset_collation();
-    m_with_collate= true;
-    return false;
 
-  case Lex_charset_collation_st::TYPE_COLLATE_CONTEXTUALLY_TYPED:
-    if (cl.is_contextually_typed_collate_default())
-    {
-      /*
-        SET NAMES latin1 COLLATE DEFAULT;
-        ALTER TABLE t1 CONVERT TO CHARACTER SET latin1 COLLATE DEFAULT;
-      */
-      CHARSET_INFO *tmp= Lex_charset_collation_st::find_default_collation(m_ci);
-      if (!tmp)
-        return true;
-      m_ci= tmp;
-      m_with_collate= true;
-      return false;
-    }
-    else
-    {
-      /*
-        EXPLICIT + CONTEXT
-        CHAR(10) COLLATE latin1_bin .. COLLATE DEFAULT not possible yet
-        CHAR(10) COLLATE latin1_bin .. COLLATE uca1400_as_ci
-      */
+bool Lex_exact_charset_opt_extended_collate::
+       merge_context_collation(const Lex_context_collation &cl)
+{
+  // CHARACTER SET latin1 [COLLATE latin1_bin] .. COLLATE DEFAULT
+  if (m_with_collate)
+    return Lex_exact_collation(m_ci).
+             raise_if_conflicts_with_context_collation(cl, false);
+  return merge_context_collation_override(cl);
+}
 
-      DBUG_ASSERT(0); // Not possible yet
-      return false;
-    }
+
+bool Lex_exact_charset_extended_collation_attrs_st::
+       merge_collation(const Lex_extended_collation_st &cl)
+{
+  switch (cl.type()) {
+  case Lex_extended_collation_st::TYPE_EXACT:
+    return merge_exact_collation(Lex_exact_collation(cl.charset_info()));
+  case Lex_extended_collation_st::TYPE_CONTEXTUALLY_TYPED:
+    return merge_context_collation(Lex_context_collation(cl.charset_info()));
   }
   DBUG_ASSERT(0);
   return false;
@@ -294,47 +516,161 @@ Lex_explicit_charset_opt_collate::
 
 
 /*
-  This method is used in the "attribute_list" rule to merge two independent
-  COLLATE clauses (not belonging to a CHARACTER SET clause).
+  Mix an unordered combination of CHARACTER SET and COLLATE clauses
+  (i.e. COLLATE can come before CHARACTER SET).
+  Merge a CHARACTER SET clause.
+  @param cs         - The "CHARACTER SET exact_charset_name".
 */
-bool
-Lex_charset_collation_st::
-  merge_collate_clause_and_collate_clause(const Lex_charset_collation_st &cl)
+bool Lex_exact_charset_extended_collation_attrs_st::
+       merge_exact_charset(const Lex_exact_charset &cs)
 {
-  /*
-    "BINARY" and "COLLATE DEFAULT" are not possible
-    in an independent COLLATE clause in a column attribute.
-  */
-  DBUG_ASSERT(!is_contextually_typed_collation());
-  DBUG_ASSERT(!cl.is_contextually_typed_collation());
-
-  if (cl.is_empty())
-    return false;
+  DBUG_ASSERT(cs.charset_info());
 
   switch (m_type) {
   case TYPE_EMPTY:
-    *this= cl;
+    // CHARACTER SET cs
+    *this= Lex_exact_charset_extended_collation_attrs(cs);
     return false;
+
   case TYPE_CHARACTER_SET:
-    DBUG_ASSERT(0);
-    return false;
+    // CHARACTER SET cs1 .. CHARACTER SET cs2
+    return Lex_exact_charset(m_ci).raise_if_not_equal(cs);
+
   case TYPE_COLLATE_EXACT:
+    // COLLATE latin1_bin .. CHARACTER SET cs
+    return cs.raise_if_not_applicable(Lex_exact_collation(m_ci));
+
   case TYPE_COLLATE_CONTEXTUALLY_TYPED:
-    break;
+    // COLLATE DEFAULT .. CHARACTER SET cs
+    {
+      Lex_exact_charset_opt_extended_collate tmp(cs);
+      if (tmp.merge_context_collation(Lex_context_collation(m_ci)))
+        return true;
+      *this= Lex_exact_charset_extended_collation_attrs(tmp.collation());
+      return false;
+    }
   }
+  DBUG_ASSERT(0);
+  return false;
+}
 
-  /*
-    Two independent explicit collations:
-      CHAR(10) NOT NULL COLLATE latin1_bin DEFAULT 'a' COLLATE latin1_bin
-    Note, we should perhaps eventually disallow double COLLATE clauses.
-    But for now let's just disallow only conflicting ones.
-  */
-  if (charset_collation() != cl.charset_collation())
+
+bool Lex_extended_charset_extended_collation_attrs_st::merge_charset_default()
+{
+  if (m_charset_order == CHARSET_TYPE_EMPTY)
+    m_charset_order= CHARSET_TYPE_CONTEXT;
+  Lex_opt_context_charset_st::merge_charset_default();
+  return false;
+}
+
+
+bool Lex_extended_charset_extended_collation_attrs_st::
+       merge_exact_charset(const Lex_exact_charset &cs)
+{
+  m_had_charset_exact= true;
+  if (m_charset_order == CHARSET_TYPE_EMPTY)
+    m_charset_order= CHARSET_TYPE_EXACT;
+  return Lex_exact_charset_extended_collation_attrs_st::merge_exact_charset(cs);
+}
+
+
+bool Lex_extended_charset_extended_collation_attrs_st::
+       raise_if_charset_conflicts_with_default(
+         const Lex_exact_charset_opt_extended_collate &def) const
+{
+  DBUG_ASSERT(m_charset_order != CHARSET_TYPE_EMPTY || is_empty());
+  if (!my_charset_same(def.collation().charset_info(), m_ci))
   {
-    my_error(ER_CONFLICTING_DECLARATIONS, MYF(0),
-             "COLLATE ", charset_collation()->coll_name.str,
-             "COLLATE ", cl.charset_collation()->coll_name.str);
+    raise_ER_CONFLICTING_DECLARATIONS("CHARACTER SET ", "DEFAULT",
+                                  def.collation().charset_info()->cs_name.str,
+                                  "CHARACTER SET ", m_ci->cs_name.str,
+                                  m_charset_order == CHARSET_TYPE_EXACT);
     return true;
   }
   return false;
 }
+
+
+CHARSET_INFO *
+Lex_extended_charset_extended_collation_attrs_st::
+  resolved_to_context(const Charset_collation_context &ctx) const
+{
+  if (Lex_opt_context_charset_st::is_empty())
+  {
+    // Without CHARACTER SET DEFAULT
+    return Lex_exact_charset_extended_collation_attrs_st::
+             resolved_to_character_set(ctx.collate_default().charset_info());
+  }
+
+  // With CHARACTER SET DEFAULT
+  switch (type()) {
+  case TYPE_EMPTY:
+    // CHARACTER SET DEFAULT;
+    return ctx.charset_default().charset().charset_info();
+
+  case TYPE_CHARACTER_SET:
+    // CHARACTER SET DEFAULT CHARACTER SET cs_exact
+    if (raise_if_charset_conflicts_with_default(ctx.charset_default()))
+    {
+      /*
+        A possible scenario:
+          SET character_set_server=utf8mb4;
+          CREATE DATABASE db1  CHARACTER SET latin1  CHARACTER SET DEFAULT;
+      */
+      return NULL;
+    }
+    return m_ci;
+
+  case TYPE_COLLATE_EXACT:
+  {
+    /*
+      CREATE DATABASE db1
+        COLLATE cl_exact
+        [ CHARACTER SET cs_exact ]
+        CHARACTER SET DEFAULT;
+    */
+    if (m_had_charset_exact &&
+        raise_if_charset_conflicts_with_default(ctx.charset_default()))
+    {
+      /*
+        A possible scenario:
+          SET character_set_server=utf8mb4;
+          CREATE DATABASE db1
+            COLLATE latin1_bin
+            CHARACTER SET latin1
+            CHARACTER SET DEFAULT;
+      */
+      return NULL;
+    }
+    /*
+      Now check that "COLLATE cl_exact" does not conflict with
+      CHARACTER SET DEFAULT.
+    */
+    if (ctx.charset_default().
+          raise_if_not_applicable(Lex_exact_collation(m_ci)))
+    {
+      /*
+        A possible scenario:
+          SET character_set_server=utf8mb4;
+          CREATE DATABASE db1
+            COLLATE latin1_bin
+            CHARACTER SET DEFAULT;
+      */
+      return NULL;
+    }
+    return m_ci;
+  }
+
+  case TYPE_COLLATE_CONTEXTUALLY_TYPED:
+    /*
+      Both CHARACTER SET and COLLATE are contextual:
+        ALTER DATABASE db1 CHARACTER SET DEFAULT COLLATE DEFAULT;
+        ALTER DATABASE db1 COLLATE DEFAULT CHARACTER SET DEFAULT;
+    */
+    return Lex_exact_charset_extended_collation_attrs_st::
+             resolved_to_character_set(ctx.charset_default().
+                                             collation().charset_info());
+  }
+  DBUG_ASSERT(0);
+  return NULL;
+}
diff --git a/sql/lex_charset.h b/sql/lex_charset.h
index abbe761df36..d195b405569 100644
--- a/sql/lex_charset.h
+++ b/sql/lex_charset.h
@@ -16,8 +16,237 @@
 #ifndef LEX_CHARSET_INCLUDED
 #define LEX_CHARSET_INCLUDED
 
+
+/*
+  An exact character set, e.g:
+    CHARACTER SET latin1
+*/
+class Lex_exact_charset
+{
+  CHARSET_INFO *m_ci;
+public:
+  explicit Lex_exact_charset(CHARSET_INFO *ci)
+   :m_ci(ci)
+  {
+    DBUG_ASSERT(m_ci);
+    DBUG_ASSERT(m_ci->state & MY_CS_PRIMARY);
+  }
+  CHARSET_INFO *charset_info() const { return m_ci; }
+  bool raise_if_not_equal(const Lex_exact_charset &rhs) const;
+  bool raise_if_not_applicable(const class Lex_exact_collation &cl) const;
+};
+
+
+/*
+  An optional contextually typed character set:
+    [ CHARACTER SET DEFAULT ]
+*/
+class Lex_opt_context_charset_st
+{
+  /*
+    Currently we support only DEFAULT as a possible value.
+    So "bool" is enough.
+  */
+  bool m_had_charset_default;
+public:
+  void init()
+  {
+    m_had_charset_default= false;
+  }
+  void merge_charset_default()
+  {
+    /*
+      Ok to specify CHARACTER SET DEFAULT multiple times.
+      No error raised here.
+    */
+    m_had_charset_default= true;
+  }
+  bool is_empty() const
+  {
+    return !m_had_charset_default;
+  }
+  bool is_contextually_typed_charset_default() const
+  {
+    return m_had_charset_default;
+  }
+};
+
+
+/*
+  A contextually typed collation, e.g.:
+    COLLATE DEFAULT
+    CHAR(10) BINARY
+*/
+class Lex_context_collation
+{
+  CHARSET_INFO *m_ci;
+public:
+  explicit Lex_context_collation(CHARSET_INFO *ci)
+   :m_ci(ci)
+  {
+    DBUG_ASSERT(ci);
+  }
+  CHARSET_INFO *charset_info() const { return m_ci; }
+  bool is_contextually_typed_collate_default() const
+  {
+    return m_ci == &my_collation_contextually_typed_default;
+  }
+  bool is_contextually_typed_binary_style() const
+  {
+    return m_ci == &my_collation_contextually_typed_binary;
+  }
+  bool raise_if_not_equal(const Lex_context_collation &cl) const;
+};
+
+
+/*
+  An exact collation, e.g.
+    COLLATE latin1_swedish_ci
+*/
+class Lex_exact_collation
+{
+  CHARSET_INFO *m_ci;
+public:
+  explicit Lex_exact_collation(CHARSET_INFO *ci)
+   :m_ci(ci)
+  {
+    DBUG_ASSERT(ci);
+  }
+  CHARSET_INFO *charset_info() const { return m_ci; }
+  // EXACT + EXACT
+  bool raise_if_not_equal(const Lex_exact_collation &cl) const;
+  // EXACT + CONTEXT
+  // CONTEXT + EXACT
+  bool raise_if_conflicts_with_context_collation(const Lex_context_collation &,
+                                                 bool reverse_order) const;
+};
+
+
+/*
+  Parse time COLLATE clause:
+    COLLATE colation_name
+  The collation can be either exact or contextual:
+    COLLATE latin1_bin
+    COLLATE DEFAULT
+*/
+class Lex_extended_collation_st
+{
+public:
+  enum Type
+  {
+    TYPE_EXACT,
+    TYPE_CONTEXTUALLY_TYPED
+  };
+protected:
+  CHARSET_INFO *m_ci;
+  Type m_type;
+public:
+  void init(CHARSET_INFO *ci, Type type)
+  {
+    m_ci= ci;
+    m_type= type;
+  }
+  CHARSET_INFO *charset_info() const { return m_ci; }
+  Type type() const { return m_type; }
+  void set_collate_default()
+  {
+    m_ci= &my_collation_contextually_typed_default;
+    m_type= TYPE_CONTEXTUALLY_TYPED;
+  }
+  bool raise_if_conflicts_with_context_collation(const Lex_context_collation &)
+                                                 const;
+  bool merge_exact_charset(const Lex_exact_charset &rhs);
+  bool merge_exact_collation(const Lex_exact_collation &rhs);
+  bool merge(const Lex_extended_collation_st &rhs);
+};
+
+
+class Lex_extended_collation: public Lex_extended_collation_st
+{
+public:
+  Lex_extended_collation(CHARSET_INFO *ci, Type type)
+  {
+    init(ci, type);
+  }
+  Lex_extended_collation(const Lex_exact_collation &rhs)
+  {
+    init(rhs.charset_info(), TYPE_EXACT);
+  }
+};
+
+
+/*
+  CHARACTER SET cs_exact [COLLATE cl_exact_or_context]
+*/
+class Lex_exact_charset_opt_extended_collate
+{
+  CHARSET_INFO *m_ci;
+  bool m_with_collate;
+public:
+  Lex_exact_charset_opt_extended_collate(CHARSET_INFO *ci, bool with_collate)
+   :m_ci(ci), m_with_collate(with_collate)
+  {
+    DBUG_ASSERT(m_ci);
+    DBUG_ASSERT((m_ci->state & MY_CS_PRIMARY) || m_with_collate);
+  }
+  Lex_exact_charset_opt_extended_collate(const Lex_exact_charset &cs)
+   :m_ci(cs.charset_info()), m_with_collate(false)
+  {
+    DBUG_ASSERT(m_ci);
+    DBUG_ASSERT(m_ci->state & MY_CS_PRIMARY);
+  }
+  Lex_exact_charset_opt_extended_collate(const Lex_exact_collation &cl)
+   :m_ci(cl.charset_info()), m_with_collate(true)
+  {
+    DBUG_ASSERT(m_ci);
+  }
+  bool with_collate() const { return m_with_collate; }
+  CHARSET_INFO *find_bin_collation() const;
+  CHARSET_INFO *find_default_collation() const;
+  bool raise_if_not_applicable(const Lex_exact_collation &cl) const;
+  /*
+    Add another COLLATE clause (exact or context).
+    So the full syntax looks like:
+      CHARACTER SET cs [COLLATE cl] ... COLLATE cl2
+  */
+  bool merge_collation(const Lex_extended_collation_st &cl)
+  {
+    switch (cl.type()) {
+    case Lex_extended_collation_st::TYPE_EXACT:
+      return merge_exact_collation(Lex_exact_collation(cl.charset_info()));
+    case Lex_extended_collation_st::TYPE_CONTEXTUALLY_TYPED:
+      return merge_context_collation(Lex_context_collation(cl.charset_info()));
+    }
+    DBUG_ASSERT(0);
+    return false;
+  }
+  /*
+    Add a context collation:
+      CHARACTER SET cs [COLLATE cl] ... COLLATE DEFAULT
+  */
+  bool merge_context_collation(const Lex_context_collation &cl);
+  bool merge_context_collation_override(const Lex_context_collation &cl);
+  /*
+    Add an exact collation:
+      CHARACTER SET cs [COLLATE cl] ... COLLATE latin1_bin
+  */
+  bool merge_exact_collation(const Lex_exact_collation &cl);
+  Lex_exact_collation collation() const
+  {
+    return Lex_exact_collation(m_ci);
+  }
+  Lex_exact_charset charset() const
+  {
+    if ((m_ci->state & MY_CS_PRIMARY))
+      return Lex_exact_charset(m_ci);
+    return Lex_exact_charset(find_default_collation());
+  }
+};
+
+
 /*
-  Parse time character set and collation.
+  Parse time character set and collation for:
+    [CHARACTER SET cs_exact] [COLLATE cl_exact_or_context]
 
   Can be:
 
@@ -44,7 +273,7 @@
 
   Resolution happens in Type_handler::Column_definition_prepare_stage1().
 */
-struct Lex_charset_collation_st
+struct Lex_exact_charset_extended_collation_attrs_st
 {
 public:
   enum Type
@@ -60,20 +289,50 @@ public:
 #define LEX_CHARSET_COLLATION_TYPE_BITS 2
   static_assert(((1<<LEX_CHARSET_COLLATION_TYPE_BITS)-1) >=
                  TYPE_COLLATE_CONTEXTUALLY_TYPED,
-                 "Lex_charset_collation_st::Type bits check");
+                 "Lex_exact_charset_extended_collation_attrs_st::Type bits");
 
 protected:
   CHARSET_INFO *m_ci;
   Type m_type;
-public:
-  static CHARSET_INFO *find_bin_collation(CHARSET_INFO *cs);
-  static CHARSET_INFO *find_default_collation(CHARSET_INFO *cs);
+protected:
+  static Type type_from_lex_collation_type(Lex_extended_collation_st::Type type)
+  {
+    switch (type) {
+    case Lex_extended_collation_st::TYPE_EXACT:
+      return TYPE_COLLATE_EXACT;
+    case Lex_extended_collation_st::TYPE_CONTEXTUALLY_TYPED:
+      return TYPE_COLLATE_CONTEXTUALLY_TYPED;
+    }
+    DBUG_ASSERT(0);
+    return TYPE_COLLATE_EXACT;
+  }
 public:
   void init()
   {
     m_ci= NULL;
     m_type= TYPE_EMPTY;
   }
+  void init(CHARSET_INFO *cs, Type type)
+  {
+    DBUG_ASSERT(cs || type == TYPE_EMPTY);
+    m_ci= cs;
+    m_type= type;
+  }
+  void init(const Lex_exact_charset &cs)
+  {
+    m_ci= cs.charset_info();
+    m_type= TYPE_CHARACTER_SET;
+  }
+  void init(const Lex_exact_collation &cs)
+  {
+    m_ci= cs.charset_info();
+    m_type= TYPE_COLLATE_EXACT;
+  }
+  void init(const Lex_exact_charset_opt_extended_collate &cscl)
+  {
+    cscl.with_collate() ? init(cscl.collation()) :
+                          init(cscl.charset());
+  }
   bool is_empty() const
   {
     return m_type == TYPE_EMPTY;
@@ -84,23 +343,26 @@ public:
     m_ci= cs;
     m_type= TYPE_CHARACTER_SET;
   }
-  void set_charset_collate_default(CHARSET_INFO *cs)
+  bool set_charset_collate_default(CHARSET_INFO *cs)
   {
     DBUG_ASSERT(cs);
+    if (!(cs= Lex_exact_charset_opt_extended_collate(cs, true).
+                find_default_collation()))
+      return true;
     m_ci= cs;
     m_type= TYPE_COLLATE_EXACT;
+    return false;
   }
   bool set_charset_collate_binary(CHARSET_INFO *cs)
   {
     DBUG_ASSERT(cs);
-    if (!(cs= find_bin_collation(cs)))
+    if (!(cs= Lex_exact_charset_opt_extended_collate(cs, true).
+                find_bin_collation()))
       return true;
     m_ci= cs;
     m_type= TYPE_COLLATE_EXACT;
     return false;
   }
-  bool set_charset_collate_exact(CHARSET_INFO *cs,
-                                 CHARSET_INFO *cl);
   void set_collate_default()
   {
     m_ci= &my_collation_contextually_typed_default;
@@ -113,19 +375,9 @@ public:
   }
   bool is_contextually_typed_collate_default() const
   {
-    return m_ci == &my_collation_contextually_typed_default;
+    return Lex_context_collation(m_ci).is_contextually_typed_collate_default();
   }
-  bool is_contextually_typed_binary_style() const
-  {
-    return m_ci == &my_collation_contextually_typed_binary;
-  }
-  void set_collate_exact(CHARSET_INFO *cl)
-  {
-    DBUG_ASSERT(cl);
-    m_ci= cl;
-    m_type= TYPE_COLLATE_EXACT;
-  }
-  CHARSET_INFO *charset_collation() const
+  CHARSET_INFO *charset_info() const
   {
     return m_ci;
   }
@@ -138,62 +390,262 @@ public:
     return m_type == TYPE_COLLATE_CONTEXTUALLY_TYPED;
   }
   CHARSET_INFO *resolved_to_character_set(CHARSET_INFO *cs) const;
-  bool merge_charset_clause_and_collate_clause(const Lex_charset_collation_st &cl);
-  bool merge_collate_clause_and_collate_clause(const Lex_charset_collation_st &cl);
+  /*
+    Merge the column CHARACTER SET clause to:
+    - an exact collation name
+    - a contextually typed collation
+    "this" corresponds to `CHARACTER SET xxx [BINARY]`
+    "cl" corresponds to the COLLATE clause
+  */
+  bool merge_column_charset_clause_and_collate_clause(
+                    const Lex_exact_charset_extended_collation_attrs_st &cl)
+  {
+    switch (cl.type()) {
+    case TYPE_EMPTY:
+      return false;
+    case TYPE_COLLATE_EXACT:
+      return merge_exact_collation(Lex_exact_collation(cl.charset_info()));
+    case TYPE_COLLATE_CONTEXTUALLY_TYPED:
+    case TYPE_CHARACTER_SET:
+      break;
+    }
+    DBUG_ASSERT(0);
+    return false;
+  }
+  /*
+    This method is used in the "attribute_list" rule to merge two independent
+    COLLATE clauses (not belonging to a CHARACTER SET clause).
+    "BINARY" and "COLLATE DEFAULT" are not possible
+    in an independent COLLATE clause in a column attribute.
+  */
+  bool merge_column_collate_clause_and_collate_clause(
+                    const Lex_exact_charset_extended_collation_attrs_st &cl)
+  {
+    DBUG_ASSERT(m_type != TYPE_COLLATE_CONTEXTUALLY_TYPED);
+    DBUG_ASSERT(m_type != TYPE_CHARACTER_SET);
+    switch (cl.type()) {
+    case TYPE_EMPTY:
+      return false;
+    case TYPE_COLLATE_EXACT:
+      return merge_exact_collation(Lex_exact_collation(cl.charset_info()));
+    case TYPE_COLLATE_CONTEXTUALLY_TYPED:
+    case TYPE_CHARACTER_SET:
+      break;
+    }
+    DBUG_ASSERT(0);
+    return false;
+  }
+  bool merge_exact_charset(const Lex_exact_charset &cs);
+  bool merge_exact_collation(const Lex_exact_collation &cl);
+  bool merge_context_collation(const Lex_context_collation &cl);
+  bool merge_collation(const Lex_extended_collation_st &cl);
 };
 
 
-/*
-  CHARACTER SET cs [COLLATE cl]
-*/
-class Lex_explicit_charset_opt_collate
+class Charset_collation_context
 {
-  CHARSET_INFO *m_ci;
-  bool m_with_collate;
+  /*
+    Although the goal of m_charset_default is to store the meaning
+    of CHARACTER SET DEFAULT, it does not necessarily point to a
+    default collation of CHARACTER SET DEFAULT. It can point to its any
+    arbitrary collation.
+    For performance purposes we don't need to find the default
+    collation at the instantiation time of "this", because:
+    - m_charset_default may not be even needed during the resolution
+    - when it's needed, in many cases it's passed to my_charset_same(),
+      which does not need the default collation again.
+
+    Note, m_charset_default and m_collate_default are not necessarily equal.
+
+    - The default value for CHARACTER SET is taken from the upper level:
+        CREATE DATABASE db1 CHARACTER SET DEFAULT; <-- @@character_set_server
+        ALTER DATABASE db1 CHARACTER SET DEFAULT;  <-- @@character_set_server
+
+    - The default value for COLLATE is taken from the upper level for CREATE:
+        CREATE DATABASE db1 COLLATE DEFAULT; <-- @@collation_server
+        CREATE TABLE db1.t1 COLLATE DEFAULT; <-- character set of "db1"
+
+    - The default value for COLLATE is taken from the same level for ALTER:
+        ALTER DATABASE db1 COLLATE DEFAULT; <-- the default collation of the
+                                                current db1 character set
+        ALTER TABLE db1.t1 COLLATE DEFAULT; <-- the default collation of the
+                                                current db1.t1 character set
+  */
+
+  // comes from the upper level
+  Lex_exact_charset_opt_extended_collate m_charset_default;
+
+  // comes from the upper or the current level
+  Lex_exact_collation m_collate_default;
 public:
-  Lex_explicit_charset_opt_collate(CHARSET_INFO *ci, bool with_collate)
-   :m_ci(ci), m_with_collate(with_collate)
+  Charset_collation_context(CHARSET_INFO *charset_default,
+                            CHARSET_INFO *collate_default)
+   :m_charset_default(charset_default,
+                      !(charset_default->state & MY_CS_PRIMARY)),
+    m_collate_default(collate_default)
+  { }
+  const Lex_exact_charset_opt_extended_collate charset_default() const
   {
-    DBUG_ASSERT(m_ci);
-    // Item_func_set_collation uses non-default collations in "ci"
-    //DBUG_ASSERT(m_ci->default_flag() || m_with_collate);
+    return m_charset_default;
+  }
+  const Lex_exact_collation collate_default() const
+  {
+    return m_collate_default;
   }
+};
+
+
+/*
+  A universal container. It can store at the same time:
+  - CHARACTER SET DEFAULT
+  - CHARACTER SET cs_exact
+  - COLLATE {cl_exact|cl_context}
+  All three parts can co-exist.
+  All three parts are optional.
+  Parts can come in any arbitrary order, e.g:
+
+    CHARACTER SET DEFAULT [CHARACTER SET latin1] COLLATE latin1_bin
+    CHARACTER SET latin1 CHARACTER SET DEFAULT COLLATE latin1_bin
+    COLLATE latin1_bin [CHARACTER SET latin1] CHARACTER SET DEFAULT
+    COLLATE latin1_bin CHARACTER SET DEFAULT [CHARACTER SET latin1]
+*/
+class Lex_extended_charset_extended_collation_attrs_st:
+                        public Lex_opt_context_charset_st,
+                        public Lex_exact_charset_extended_collation_attrs_st
+{
+  enum charset_type_t
+  {
+    CHARSET_TYPE_EMPTY,
+    CHARSET_TYPE_CONTEXT,
+    CHARSET_TYPE_EXACT
+  };
   /*
-    Merge to another COLLATE clause. So the full syntax looks like:
-      CHARACTER SET cs [COLLATE cl] ... COLLATE cl2
+    Which part came first:
+    - CHARACTER SET DEFAULT or
+    - CHARACTER SET cs_exact
+    e.g. to produce error messages preserving the user typed
+    order of CHARACTER SET clauses in case of conflicts.
   */
-  bool merge_collate_or_error(const Lex_charset_collation_st &cl);
-  bool merge_opt_collate_or_error(const Lex_charset_collation_st &cl)
+  charset_type_t m_charset_order;
+  /*
+    The parent class Lex_exact_charset_extended_collation_attrs_st
+    does not let know if a "COLLATE cl_exact" was used in combination with
+    "CHARACTER SET cs_exact" or just alone.
+    Here we need to distinguish:
+    - CHARACTER SET cs_exact COLLATE cl_exact, or
+    - COLLATE cl_exact CHARACTER SET cs_exact
+    versus just:
+    - COLLATE cl_exact
+    to produce better error messages in case of conflicts.
+    So let's add a flag member:
+  */
+  bool m_had_charset_exact;
+public:
+  void init()
   {
-    if (cl.is_empty())
-      return false;
-    return merge_collate_or_error(cl);
+    Lex_opt_context_charset_st::init();
+    Lex_exact_charset_extended_collation_attrs_st::init();
+    m_charset_order= CHARSET_TYPE_EMPTY;
+    m_had_charset_exact= false;
   }
-  CHARSET_INFO *charset_and_collation() const { return m_ci; }
-  bool with_collate() const { return m_with_collate; }
+  void init(const Lex_exact_charset_opt_extended_collate &c)
+  {
+    Lex_opt_context_charset_st::init();
+    Lex_exact_charset_extended_collation_attrs_st::init(c);
+    m_charset_order= CHARSET_TYPE_EXACT;
+    m_had_charset_exact= true;
+  }
+  bool is_empty() const
+  {
+    return Lex_opt_context_charset_st::is_empty() &&
+           Lex_exact_charset_extended_collation_attrs_st::is_empty();
+  }
+  bool raise_if_charset_conflicts_with_default(
+                        const Lex_exact_charset_opt_extended_collate &def) const;
+  CHARSET_INFO *resolved_to_context(const Charset_collation_context &ctx) const;
+  bool merge_charset_default();
+  bool merge_exact_charset(const Lex_exact_charset &cs);
 };
 
 
-class Lex_charset_collation: public Lex_charset_collation_st
+class Lex_exact_charset_extended_collation_attrs:
+                        public Lex_exact_charset_extended_collation_attrs_st
 {
 public:
-  Lex_charset_collation()
+  Lex_exact_charset_extended_collation_attrs()
   {
     init();
   }
-  Lex_charset_collation(CHARSET_INFO *collation, Type type)
+  Lex_exact_charset_extended_collation_attrs(CHARSET_INFO *collation, Type type)
   {
-    DBUG_ASSERT(collation || type == TYPE_EMPTY);
-    m_ci= collation;
-    m_type= type;
+    init(collation, type);
+  }
+  explicit
+  Lex_exact_charset_extended_collation_attrs(const Lex_exact_charset &cs)
+  {
+    init(cs.charset_info(), TYPE_CHARACTER_SET);
+  }
+  explicit
+  Lex_exact_charset_extended_collation_attrs(const Lex_exact_collation &cl)
+  {
+    init(cl.charset_info(), TYPE_COLLATE_EXACT);
+  }
+  explicit
+  Lex_exact_charset_extended_collation_attrs(const Lex_context_collation &cl)
+  {
+    init(cl.charset_info(), TYPE_COLLATE_CONTEXTUALLY_TYPED);
   }
-  static Lex_charset_collation national(bool bin_mod)
+  explicit
+  Lex_exact_charset_extended_collation_attrs(
+                    const Lex_exact_charset_opt_extended_collate &cscl)
+  {
+    init(cscl);
+  }
+  explicit
+  Lex_exact_charset_extended_collation_attrs(const Lex_extended_collation_st &cl)
+  {
+    init(cl.charset_info(), type_from_lex_collation_type(cl.type()));
+  }
+  static Lex_exact_charset_extended_collation_attrs national(bool bin_mod)
   {
     return bin_mod ?
-      Lex_charset_collation(&my_charset_utf8mb3_bin, TYPE_COLLATE_EXACT) :
-      Lex_charset_collation(&my_charset_utf8mb3_general_ci, TYPE_CHARACTER_SET);
+      Lex_exact_charset_extended_collation_attrs(&my_charset_utf8mb3_bin,
+                                                 TYPE_COLLATE_EXACT) :
+      Lex_exact_charset_extended_collation_attrs(&my_charset_utf8mb3_general_ci,
+                                                 TYPE_CHARACTER_SET);
+  }
+};
+
+
+class Lex_extended_charset_extended_collation_attrs:
+  public Lex_extended_charset_extended_collation_attrs_st
+{
+public:
+  Lex_extended_charset_extended_collation_attrs()
+  {
+    init();
+  }
+  explicit Lex_extended_charset_extended_collation_attrs(
+    const Lex_exact_charset_opt_extended_collate &c)
+  {
+    init(c);
   }
 };
 
 
+
+using Lex_column_charset_collation_attrs_st =
+        Lex_exact_charset_extended_collation_attrs_st;
+
+using Lex_column_charset_collation_attrs =
+        Lex_exact_charset_extended_collation_attrs;
+
+
+using Lex_table_charset_collation_attrs_st =
+        Lex_extended_charset_extended_collation_attrs_st;
+
+using Lex_table_charset_collation_attrs =
+        Lex_extended_charset_extended_collation_attrs;
+
+
 #endif // LEX_CHARSET_INCLUDED
diff --git a/sql/mysql_install_db.cc b/sql/mysql_install_db.cc
index 710f05784d1..934226685a9 100644
--- a/sql/mysql_install_db.cc
+++ b/sql/mysql_install_db.cc
@@ -335,7 +335,7 @@ static char *init_bootstrap_command_line(char *cmdline, size_t size)
     " %s"
     " --bootstrap"
     " --datadir=."
-    " --loose-innodb-buffer-pool-size=10M"
+    " --loose-innodb-buffer-pool-size=20M"
     "\""
     , mysqld_path, opt_verbose_bootstrap ? "--console" : "");
   return cmdline;
diff --git a/sql/mysqld.cc b/sql/mysqld.cc
index 00a7b23b085..9a64fbf2c52 100644
--- a/sql/mysqld.cc
+++ b/sql/mysqld.cc
@@ -1559,11 +1559,15 @@ static void kill_thread(THD *thd)
 /**
   First shutdown everything but slave threads and binlog dump connections
 */
-static my_bool kill_thread_phase_1(THD *thd, void *)
+static my_bool kill_thread_phase_1(THD *thd, int *n_threads_awaiting_ack)
 {
   DBUG_PRINT("quit", ("Informing thread %ld that it's time to die",
                       (ulong) thd->thread_id));
-  if (thd->slave_thread || thd->is_binlog_dump_thread())
+
+  if (thd->slave_thread || thd->is_binlog_dump_thread() ||
+      (shutdown_wait_for_slaves &&
+       repl_semisync_master.is_thd_awaiting_semisync_ack(thd) &&
+       ++(*n_threads_awaiting_ack)))
     return 0;
 
   if (DBUG_IF("only_kill_system_threads") ? !thd->system_thread : 0)
@@ -1578,7 +1582,7 @@ static my_bool kill_thread_phase_1(THD *thd, void *)
 */
 static my_bool kill_thread_phase_2(THD *thd, void *)
 {
-  if (shutdown_wait_for_slaves)
+  if (shutdown_wait_for_slaves && thd->is_binlog_dump_thread())
   {
     thd->set_killed(KILL_SERVER);
   }
@@ -1751,7 +1755,29 @@ static void close_connections(void)
     This will give the threads some time to gracefully abort their
     statements and inform their clients that the server is about to die.
   */
-  server_threads.iterate(kill_thread_phase_1);
+  DBUG_EXECUTE_IF("mysqld_delay_kill_threads_phase_1", my_sleep(200000););
+  int n_threads_awaiting_ack= 0;
+  server_threads.iterate(kill_thread_phase_1, &n_threads_awaiting_ack);
+
+  /*
+    If we are waiting on any ACKs, delay killing the thread until either an ACK
+    is received or the timeout is hit.
+
+    Allow at max the number of sessions to await a timeout; however, if all
+    ACKs have been received in less iterations, then quit early
+  */
+  if (shutdown_wait_for_slaves && repl_semisync_master.get_master_enabled())
+  {
+    int waiting_threads= repl_semisync_master.sync_get_master_wait_sessions();
+    if (waiting_threads)
+      sql_print_information("Delaying shutdown to await semi-sync ACK");
+
+    while (waiting_threads-- > 0)
+      repl_semisync_master.await_slave_reply();
+  }
+
+  DBUG_EXECUTE_IF("delay_shutdown_phase_2_after_semisync_wait",
+                  my_sleep(500000););
 
   Events::deinit();
   slave_prepare_for_shutdown();
@@ -1774,7 +1800,8 @@ static void close_connections(void)
   */
   DBUG_PRINT("info", ("THD_count: %u", THD_count::value()));
 
-  for (int i= 0; (THD_count::connection_thd_count()) && i < 1000; i++)
+  for (int i= 0; THD_count::connection_thd_count() - n_threads_awaiting_ack
+                 && i < 1000; i++)
     my_sleep(20000);
 
   if (global_system_variables.log_warnings)
@@ -1788,9 +1815,10 @@ static void close_connections(void)
   wsrep_sst_auth_free();
 #endif
   /* All threads has now been aborted */
-  DBUG_PRINT("quit", ("Waiting for threads to die (count=%u)", THD_count::value()));
+  DBUG_PRINT("quit", ("Waiting for threads to die (count=%u)",
+                  THD_count::connection_thd_count() - n_threads_awaiting_ack));
 
-  while (THD_count::connection_thd_count())
+  while (THD_count::connection_thd_count() - n_threads_awaiting_ack)
     my_sleep(1000);
 
   /* Kill phase 2 */
@@ -5353,6 +5381,9 @@ static int init_server_components()
   if (ha_recover(0))
     unireg_abort(1);
 
+#ifndef EMBEDDED_LIBRARY
+  start_handle_manager();
+#endif
   if (opt_bin_log)
   {
     int error;
@@ -5810,8 +5841,6 @@ int mysqld_main(int argc, char **argv)
     }
   }
 
-  start_handle_manager();
-
   /* Copy default global rpl_filter to global_rpl_filter */
   copy_filter_setting(global_rpl_filter, get_or_create_rpl_filter("", 0));
 
diff --git a/sql/partition_info.cc b/sql/partition_info.cc
index 5c2237fb3d2..19f077136cb 100644
--- a/sql/partition_info.cc
+++ b/sql/partition_info.cc
@@ -829,12 +829,10 @@ bool partition_info::vers_set_hist_part(THD *thd, uint *create_count)
   {
     DBUG_ASSERT(!vers_info->interval.is_set());
     ha_partition *hp= (ha_partition*)(table->file);
-    partition_element *next= NULL;
+    partition_element *next;
     List_iterator<partition_element> it(partitions);
-    while (next != vers_info->hist_part)
-      next= it++;
-    DBUG_ASSERT(bitmap_is_set(&read_partitions, next->id));
-    ha_rows records= hp->part_records(next);
+    ha_rows records= 0;
+    vers_info->hist_part= partitions.head();
     while ((next= it++) != vers_info->now_part)
     {
       DBUG_ASSERT(bitmap_is_set(&read_partitions, next->id));
@@ -850,14 +848,11 @@ bool partition_info::vers_set_hist_part(THD *thd, uint *create_count)
       {
         if (auto_hist)
           *create_count= 1;
-        else
-          my_error(WARN_VERS_PART_FULL, MYF(ME_WARNING|ME_ERROR_LOG),
-                  table->s->db.str, table->s->table_name.str,
-                  vers_info->hist_part->partition_name, "LIMIT");
       }
       else
         vers_info->hist_part= next;
     }
+    return 0;
   }
   else if (vers_info->interval.is_set() &&
            vers_info->hist_part->range_value <= thd->query_start())
@@ -927,7 +922,7 @@ bool partition_info::vers_set_hist_part(THD *thd, uint *create_count)
 bool vers_create_partitions(THD *thd, TABLE_LIST* tl, uint num_parts)
 {
   bool result= true;
-  HA_CREATE_INFO create_info;
+  Table_specification_st create_info;
   Alter_info alter_info;
   partition_info *save_part_info= thd->work_part_info;
   Query_tables_list save_query_tables;
@@ -1039,6 +1034,43 @@ exit:
 }
 
 
+/**
+  Warn at the end of DML command if the last history partition is out of LIMIT.
+*/
+void partition_info::vers_check_limit(THD *thd)
+{
+  if (vers_info->auto_hist || !vers_info->limit ||
+      vers_info->hist_part->id + 1 < vers_info->now_part->id)
+    return;
+
+  /*
+    NOTE: at this point read_partitions bitmap is already pruned by DML code,
+    we have to set read bits for working history partition. We could use
+    bitmap_set_all(), but this is not optimal since there can be quite a number
+    of partitions.
+  */
+  const uint32 sub_factor= num_subparts ? num_subparts : 1;
+  uint32 part_id= vers_info->hist_part->id * sub_factor;
+  const uint32 part_id_end= part_id + sub_factor;
+  DBUG_ASSERT(part_id_end <= num_parts * sub_factor);
+
+  ha_partition *hp= (ha_partition*)(table->file);
+  ha_rows hist_rows= hp->part_records(vers_info->hist_part);
+  if (hist_rows >= vers_info->limit)
+  {
+    push_warning_printf(thd, Sql_condition::WARN_LEVEL_WARN,
+                        WARN_VERS_PART_FULL,
+                        ER_THD(thd, WARN_VERS_PART_FULL),
+                        table->s->db.str, table->s->table_name.str,
+                        vers_info->hist_part->partition_name, "LIMIT");
+
+    sql_print_warning(ER_THD(thd, WARN_VERS_PART_FULL),
+                      table->s->db.str, table->s->table_name.str,
+                      vers_info->hist_part->partition_name, "LIMIT");
+  }
+}
+
+
 /*
   Check that the partition/subpartition is setup to use the correct
   storage engine
diff --git a/sql/partition_info.h b/sql/partition_info.h
index a3e36a64ffa..93247b939e4 100644
--- a/sql/partition_info.h
+++ b/sql/partition_info.h
@@ -407,6 +407,12 @@ public:
                          bool auto_part, const char *table_name);
   bool vers_set_limit(ulonglong limit, bool auto_part, const char *table_name);
   bool vers_set_hist_part(THD* thd, uint *create_count);
+  bool vers_require_hist_part(THD *thd) const
+  {
+    return part_type == VERSIONING_PARTITION &&
+      thd->lex->vers_history_generating();
+  }
+  void vers_check_limit(THD *thd);
   bool vers_fix_field_list(THD *thd);
   void vers_update_el_ids();
   partition_element *get_partition(uint part_id)
diff --git a/sql/rpl_gtid.cc b/sql/rpl_gtid.cc
index eb9c0758d02..516c9da31d9 100644
--- a/sql/rpl_gtid.cc
+++ b/sql/rpl_gtid.cc
@@ -3677,9 +3677,9 @@ int Id_delegating_gtid_event_filter<T>::set_id_restrictions(
     {
       DBUG_ASSERT(map_element->filter->get_filter_type() ==
                   (mode ==
-                   Gtid_event_filter::id_restriction_mode::WHITELIST_MODE)
-                  ? Gtid_event_filter::ACCEPT_ALL_GTID_FILTER_TYPE
-                  : Gtid_event_filter::REJECT_ALL_GTID_FILTER_TYPE);
+                   Gtid_event_filter::id_restriction_mode::WHITELIST_MODE
+                   ? Gtid_event_filter::ACCEPT_ALL_GTID_FILTER_TYPE
+                   : Gtid_event_filter::REJECT_ALL_GTID_FILTER_TYPE));
     }
   }
 
diff --git a/sql/rpl_gtid.h b/sql/rpl_gtid.h
index 8bbb87c0e81..925e271e68a 100644
--- a/sql/rpl_gtid.h
+++ b/sql/rpl_gtid.h
@@ -26,6 +26,7 @@
 extern const LEX_CSTRING rpl_gtid_slave_state_table_name;
 
 class String;
+#define PARAM_GTID(G) G.domain_id, G.server_id, G.seq_no
 
 #define GTID_MAX_STR_LENGTH (10+1+10+1+20)
 #define PARAM_GTID(G) G.domain_id, G.server_id, G.seq_no
diff --git a/sql/semisync_master.cc b/sql/semisync_master.cc
index cfe29824328..0e72ff441c4 100644
--- a/sql/semisync_master.cc
+++ b/sql/semisync_master.cc
@@ -463,6 +463,37 @@ void Repl_semi_sync_master::cleanup()
   delete m_active_tranxs;
 }
 
+int Repl_semi_sync_master::sync_get_master_wait_sessions()
+{
+  int wait_sessions;
+  lock();
+  wait_sessions= rpl_semi_sync_master_wait_sessions;
+  unlock();
+  return wait_sessions;
+}
+
+void Repl_semi_sync_master::create_timeout(struct timespec *out,
+                                           struct timespec *start_arg)
+{
+  struct timespec *start_ts;
+  struct timespec now_ts;
+  if (!start_arg)
+  {
+    set_timespec(now_ts, 0);
+    start_ts= &now_ts;
+  }
+  else
+  {
+    start_ts= start_arg;
+  }
+
+  long diff_secs= (long) (m_wait_timeout / TIME_THOUSAND);
+  long diff_nsecs= (long) ((m_wait_timeout % TIME_THOUSAND) * TIME_MILLION);
+  long nsecs= start_ts->tv_nsec + diff_nsecs;
+  out->tv_sec= start_ts->tv_sec + diff_secs + nsecs / TIME_BILLION;
+  out->tv_nsec= nsecs % TIME_BILLION;
+}
+
 void Repl_semi_sync_master::lock()
 {
   mysql_mutex_lock(&LOCK_binlog);
@@ -862,13 +893,6 @@ int Repl_semi_sync_master::commit_trx(const char* trx_wait_binlog_name,
                                 m_wait_file_name, (ulong)m_wait_file_pos));
       }
 
-      /* Calcuate the waiting period. */
-      long diff_secs = (long) (m_wait_timeout / TIME_THOUSAND);
-      long diff_nsecs = (long) ((m_wait_timeout % TIME_THOUSAND) * TIME_MILLION);
-      long nsecs = start_ts.tv_nsec + diff_nsecs;
-      abstime.tv_sec = start_ts.tv_sec + diff_secs + nsecs/TIME_BILLION;
-      abstime.tv_nsec = nsecs % TIME_BILLION;
-
       /* In semi-synchronous replication, we wait until the binlog-dump
        * thread has received the reply on the relevant binlog segment from the
        * replication slave.
@@ -879,12 +903,20 @@ int Repl_semi_sync_master::commit_trx(const char* trx_wait_binlog_name,
        */
       rpl_semi_sync_master_wait_sessions++;
 
+      /* We keep track of when this thread is awaiting an ack to ensure it is
+       * not killed while awaiting an ACK if a shutdown is issued.
+       */
+      set_thd_awaiting_semisync_ack(thd, TRUE);
+
       DBUG_PRINT("semisync", ("%s: wait %lu ms for binlog sent (%s, %lu)",
                               "Repl_semi_sync_master::commit_trx",
                               m_wait_timeout,
                               m_wait_file_name, (ulong)m_wait_file_pos));
 
+      create_timeout(&abstime, &start_ts);
       wait_result = cond_timewait(&abstime);
+
+      set_thd_awaiting_semisync_ack(thd, FALSE);
       rpl_semi_sync_master_wait_sessions--;
 
       if (wait_result != 0)
@@ -1320,6 +1352,25 @@ void Repl_semi_sync_master::set_export_stats()
   unlock();
 }
 
+void Repl_semi_sync_master::await_slave_reply()
+{
+  struct timespec abstime;
+
+  DBUG_ENTER("Repl_semi_sync_master::::await_slave_reply");
+  lock();
+
+  /* Just return if there is nothing to wait for */
+  if (!rpl_semi_sync_master_wait_sessions)
+    goto end;
+
+  create_timeout(&abstime, NULL);
+  cond_timewait(&abstime);
+
+end:
+  unlock();
+  DBUG_VOID_RETURN;
+}
+
 /* Get the waiting time given the wait's staring time.
  *
  * Return:
diff --git a/sql/semisync_master.h b/sql/semisync_master.h
index 9f0acf57a60..04fc0e5ce50 100644
--- a/sql/semisync_master.h
+++ b/sql/semisync_master.h
@@ -472,6 +472,21 @@ class Repl_semi_sync_master
     m_wait_timeout = wait_timeout;
   }
 
+  int sync_get_master_wait_sessions();
+
+  /*
+    Calculates a timeout that is m_wait_timeout after start_arg and saves it
+    in out. If start_arg is NULL, the timeout is m_wait_timeout after the
+    current system time.
+  */
+  void create_timeout(struct timespec *out, struct timespec *start_arg);
+
+  /*
+    Blocks the calling thread until the ack_receiver either receives an ACK
+    or times out (from rpl_semi_sync_master_timeout)
+  */
+  void await_slave_reply();
+
   /*set the ACK point, after binlog sync or after transaction commit*/
   void set_wait_point(unsigned long ack_point)
   {
@@ -620,6 +635,30 @@ class Repl_semi_sync_master
 
   void check_and_switch();
 
+  /*
+    Determines if the given thread is currently awaiting a semisync_ack. Note
+    that the thread's value is protected by this class's LOCK_binlog, so this
+    function (indirectly) provides safe access.
+  */
+  my_bool is_thd_awaiting_semisync_ack(THD *thd)
+  {
+    lock();
+    my_bool ret= thd->is_awaiting_semisync_ack;
+    unlock();
+    return ret;
+  }
+
+  /*
+    Update the thread's value for is_awaiting_semisync_ack. LOCK_binlog (from
+    this class) should be acquired before calling this function.
+  */
+  void set_thd_awaiting_semisync_ack(THD *thd,
+                                     my_bool _is_awaiting_semisync_ack)
+  {
+    mysql_mutex_assert_owner(&LOCK_binlog);
+    thd->is_awaiting_semisync_ack= _is_awaiting_semisync_ack;
+  }
+
   mysql_mutex_t LOCK_rpl_semi_sync_master_enabled;
 };
 
diff --git a/sql/semisync_slave.cc b/sql/semisync_slave.cc
index 684199031ee..788aab78911 100644
--- a/sql/semisync_slave.cc
+++ b/sql/semisync_slave.cc
@@ -114,10 +114,12 @@ int Repl_semi_sync_slave::slave_start(Master_info *mi)
 
 int Repl_semi_sync_slave::slave_stop(Master_info *mi)
 {
-  if (rpl_semi_sync_slave_status)
-    rpl_semi_sync_slave_status= 0;
   if (get_slave_enabled())
     kill_connection(mi->mysql);
+
+  if (rpl_semi_sync_slave_status)
+    rpl_semi_sync_slave_status= 0;
+
   return 0;
 }
 
@@ -133,6 +135,8 @@ void Repl_semi_sync_slave::kill_connection(MYSQL *mysql)
 
   char kill_buffer[30];
   MYSQL *kill_mysql = NULL;
+  size_t kill_buffer_length;
+
   kill_mysql = mysql_init(kill_mysql);
   mysql_options(kill_mysql, MYSQL_OPT_CONNECT_TIMEOUT, &m_kill_conn_timeout);
   mysql_options(kill_mysql, MYSQL_OPT_READ_TIMEOUT, &m_kill_conn_timeout);
@@ -144,13 +148,35 @@ void Repl_semi_sync_slave::kill_connection(MYSQL *mysql)
   {
     sql_print_information("cannot connect to master to kill slave io_thread's "
                           "connection");
-    mysql_close(kill_mysql);
-    return;
+    goto failed_graceful_kill;
   }
-  size_t kill_buffer_length = my_snprintf(kill_buffer, 30, "KILL %lu",
-                                        mysql->thread_id);
-  mysql_real_query(kill_mysql, kill_buffer, (ulong)kill_buffer_length);
+
+  DBUG_EXECUTE_IF("slave_delay_killing_semisync_connection", my_sleep(400000););
+
+  kill_buffer_length= my_snprintf(kill_buffer, 30, "KILL %lu",
+                                mysql->thread_id);
+  if (mysql_real_query(kill_mysql, kill_buffer, (ulong)kill_buffer_length))
+  {
+    sql_print_information(
+        "Failed to gracefully kill our active semi-sync connection with "
+        "primary. Silently closing the connection.");
+    goto failed_graceful_kill;
+  }
+
+end:
   mysql_close(kill_mysql);
+  return;
+
+failed_graceful_kill:
+  /*
+    If we fail to issue `KILL` on the primary to kill the active semi-sync
+    connection; we need to locally clean up our side of the connection. This
+    is because mysql_close will send COM_QUIT on the active semi-sync
+    connection, causing the primary to error.
+  */
+  net_clear(&(mysql->net), 0);
+  end_server(mysql);
+  goto end;
 }
 
 int Repl_semi_sync_slave::request_transmit(Master_info *mi)
diff --git a/sql/semisync_slave.h b/sql/semisync_slave.h
index e7ccd952130..f0b8eceeebf 100644
--- a/sql/semisync_slave.h
+++ b/sql/semisync_slave.h
@@ -23,6 +23,7 @@
 #include "sql_priv.h"
 #include "rpl_mi.h"
 #include "mysql.h"
+#include <sql_common.h>
 
 class Master_info;
 
diff --git a/sql/share/CMakeLists.txt b/sql/share/CMakeLists.txt
index 55f39e5e22b..ddb813057ab 100644
--- a/sql/share/CMakeLists.txt
+++ b/sql/share/CMakeLists.txt
@@ -15,30 +15,32 @@
 # Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA  02110-1335 USA
 
 SET (dirs 
+bulgarian
+chinese
+czech
 danish
-german 
-slovak
 dutch
-greek
-norwegian
-spanish
 english
+estonian
+french
+german
+greek
+hindi
 hungarian
-norwegian-ny
-swedish
 italian
-polish
-ukrainian
 japanese
+korean
+norwegian-ny
+norwegian
+polish
 portuguese
 romanian
-estonian 
-korean
 russian
-czech
-french
 serbian
-hindi
+slovak
+spanish
+swedish
+ukrainian
 )
 
 SET(files 
diff --git a/sql/share/errmsg-utf8.txt b/sql/share/errmsg-utf8.txt
index 388a3aa74c1..e2cacdae420 100644
--- a/sql/share/errmsg-utf8.txt
+++ b/sql/share/errmsg-utf8.txt
@@ -1,4 +1,4 @@
-languages chinese=chi gbk, czech=cze latin2, danish=dan latin1, dutch=nla latin1, english=eng latin1, estonian=est latin7, french=fre latin1, german=ger latin1, greek=greek greek, hungarian=hun latin2, italian=ita latin1, japanese=jpn ujis, korean=kor euckr, norwegian-ny=norwegian-ny latin1, norwegian=nor latin1, polish=pol latin2, portuguese=por latin1, romanian=rum latin2, russian=rus koi8r, serbian=serbian cp1250, slovak=slo latin2, spanish=spa latin1, swedish=swe latin1, ukrainian=ukr koi8u, bulgarian=bgn cp1251, hindi=hindi utf8mb3;
+languages bulgarian=bgn cp1251, chinese=chi gbk, czech=cze latin2, danish=dan latin1, dutch=nla latin1, english=eng latin1, estonian=est latin7, french=fre latin1, german=ger latin1, greek=greek greek, hindi=hindi utf8mb3, hungarian=hun latin2, italian=ita latin1, japanese=jpn ujis, korean=kor euckr, norwegian-ny=norwegian-ny latin1, norwegian=nor latin1, polish=pol latin2, portuguese=por latin1, romanian=rum latin2, russian=rus koi8r, serbian=serbian cp1250, slovak=slo latin2, spanish=spa latin1, swedish=swe latin1, ukrainian=ukr koi8u;
 
 default-language eng
 
@@ -6450,10 +6450,21 @@ ER_AMBIGUOUS_FIELD_TERM
         ger "Das erste Zeichen der Zeichenkette FIELDS TERMINATED ist mehrdeutig; bitte benutzen Sie nicht optionale und nicht leere FIELDS ENCLOSED BY"
         spa "El primer carácter de la cadena de los FIELDS TERMINATED es ambiguo; por favor, use FIELDS ENCLOSED BY no opcionales y no vacíos"
 ER_FOREIGN_SERVER_EXISTS
-        chi "您正在尝试创建的外来服务器%s已存在"
-        eng "The foreign server, %s, you are trying to create already exists"
-        ger "Der entfernte Server %s, den Sie versuchen zu erzeugen, existiert schon"
-        spa "El servidor foráneo %s que intenta crear ya existe"
+        chi "无法创建外部服务器'%s'，因为它已经存在"
+        eng "Cannot create foreign server '%s' as it already exists"
+        fin "Vieraata palvelinta '%s' ei voida luoda, koska se on jo olemassa"
+        fre "Impossible de créer le serveur étranger '%s' car il existe déjà"
+        ger "Der auswärtige Server '%s' kann nicht erstellt werden, da er bereits vorhanden ist"
+        greek "Δεν είναι δυνατή η δημιουργία ξένου διακομιστή '%s' επειδή υπάρχει ήδη"
+        ita "Impossibile creare il server esterno '%s' poiché esiste già"
+        jpn "外部サーバー '%s'は既に存在するため、作成できません"
+        nla "Kan geen externe server '%s' maken omdat deze al bestaat"
+        nor "Kan ikke opprette utenlandsk server '%s' fordi den allerede eksisterer"
+        pol "Nie można utworzyć obcego serwera '%s', ponieważ już istnieje"
+        por "Não foi possível criar o servidor externo '%s' porque ele já existe"
+        rus "Невозможно создать сторонний сервер '%s', так как он уже существует"
+        spa "No se puede crear el servidor externo '%s' porque ya existe"
+        swe "Det gick inte att skapa främmande server '%s' eftersom den redan finns"
 ER_FOREIGN_SERVER_DOESNT_EXIST
         chi "您尝试引用的外部服务器名称不存在。数据源错误：%-.64s"
         eng "The foreign server name you are trying to reference does not exist. Data source error:  %-.64s"
@@ -8200,12 +8211,12 @@ ER_TABLE_IN_SYSTEM_TABLESPACE
         spa "Tabla %-.192s en espacio de tablas del sitema"
 
 ER_IO_READ_ERROR
-        chi "IO读取错误:（%lu，%s）%s"
+        chi "IO读取错误:(%lu，%s)%s"
         eng "IO Read error: (%lu, %s) %s"
         spa "Error de Lectura de E/S: (%lu, %s) %s"
 
 ER_IO_WRITE_ERROR
-        chi "IO写错错误:（%lu，%s）%s"
+        chi "IO写错错误:(%lu，%s)%s"
         eng "IO Write error: (%lu, %s) %s"
         spa "Error de Escritura de E/S: (%lu, %s) %s"
 
@@ -9300,7 +9311,7 @@ ER_REF_TO_RECURSIVE_WITH_TABLE_IN_DERIVED
         eng "Reference to recursive WITH table '%s' in materialized derived"
         spa "Referencia recursiva con WITH tabla '%s' en derivada materializada"
 ER_NOT_STANDARD_COMPLIANT_RECURSIVE
-        chi "表'%s'违反了递归定义的限制"
+        chi "表'%s'R_WRONG_WINDOW_SPEC_NAME违反了递归定义的限制"
         eng "Restrictions imposed on recursive definitions are violated for table '%s'"
 ER_WRONG_WINDOW_SPEC_NAME
         chi "没有定义名称'%s'的窗口规范"
diff --git a/sql/slave.cc b/sql/slave.cc
index f6f243d7f03..19037525422 100644
--- a/sql/slave.cc
+++ b/sql/slave.cc
@@ -3307,9 +3307,9 @@ static bool send_show_master_info_data(THD *thd, Master_info *mi, bool full,
     protocol->store((uint32) mi->master_id);
     // SQL_Delay
     // Master_Ssl_Crl
-    protocol->store_string_or_null(mi->ssl_ca, &my_charset_bin);
+    protocol->store_string_or_null(mi->ssl_crl, &my_charset_bin);
     // Master_Ssl_Crlpath
-    protocol->store_string_or_null(mi->ssl_capath, &my_charset_bin);
+    protocol->store_string_or_null(mi->ssl_crlpath, &my_charset_bin);
     // Using_Gtid
     protocol->store_string_or_null(mi->using_gtid_astext(mi->using_gtid),
                                    &my_charset_bin);
@@ -3402,7 +3402,7 @@ bool show_all_master_info(THD* thd)
   String gtid_pos;
   Master_info **tmp;
   List<Item> field_list;
-  DBUG_ENTER("show_master_info");
+  DBUG_ENTER("show_all_master_info");
   mysql_mutex_assert_owner(&LOCK_active_mi);
 
   gtid_pos.length(0);
@@ -4991,6 +4991,7 @@ Stopping slave I/O thread due to out-of-memory error from master");
           not cause the slave IO thread to stop, and the error messages are
           already reported.
         */
+        DBUG_EXECUTE_IF("simulate_delay_semisync_slave_reply", my_sleep(800000););
         (void)repl_semisync_slave.slave_reply(mi);
       }
 
@@ -6777,17 +6778,75 @@ static int queue_event(Master_info* mi, const uchar *buf, ulong event_len)
     mi->gtid_event_seen= true;
 
     /*
-      We have successfully queued to relay log everything before this GTID, so
+      Unless the previous group is malformed,
+      we have successfully queued to relay log everything before this GTID, so
       in case of reconnect we can start from after any previous GTID.
-      (Normally we would have updated gtid_current_pos earlier at the end of
-      the previous event group, but better leave an extra check here for
-      safety).
+      (We must have updated gtid_current_pos earlier at the end of
+      the previous event group. Unless ...)
     */
-    if (mi->events_queued_since_last_gtid)
+    if (unlikely(mi->events_queued_since_last_gtid > 0))
     {
-      mi->gtid_current_pos.update(&mi->last_queued_gtid);
-      mi->events_queued_since_last_gtid= 0;
+      /*
+        ...unless the last group has not been completed. An assert below
+        can be satisfied only with the strict mode that ensures
+        against "genuine" gtid duplicates.
+      */
+      rpl_gtid *gtid_in_slave_state=
+        mi->gtid_current_pos.find(mi->last_queued_gtid.domain_id);
+
+      // Slave gtid state must not have updated yet to the last received gtid.
+      DBUG_ASSERT((mi->using_gtid == Master_info::USE_GTID_NO ||
+                   !opt_gtid_strict_mode) ||
+                  (!gtid_in_slave_state ||
+                   !(*gtid_in_slave_state == mi->last_queued_gtid)));
+
+      DBUG_EXECUTE_IF("slave_discard_xid_for_gtid_0_x_1000",
+      {
+        /* Inject an event group that is missing its XID commit event. */
+        if ((mi->last_queued_gtid.domain_id == 0 &&
+             mi->last_queued_gtid.seq_no == 1000) ||
+            (mi->last_queued_gtid.domain_id == 1 &&
+             mi->last_queued_gtid.seq_no == 32))
+        {
+          sql_print_warning(
+            "Unexpected break of being relay-logged GTID %u-%u-%llu "
+            "event group by the current GTID event %u-%u-%llu",
+            PARAM_GTID(mi->last_queued_gtid),PARAM_GTID(event_gtid));
+          DBUG_SET("-d,slave_discard_xid_for_gtid_0_x_1000");
+          goto dbug_gtid_accept;
+        }
+      });
+      error= ER_SLAVE_RELAY_LOG_WRITE_FAILURE;
+      sql_print_error("Unexpected break of being relay-logged GTID %u-%u-%llu "
+                      "event group by the current GTID event %u-%u-%llu",
+                      PARAM_GTID(mi->last_queued_gtid),PARAM_GTID(event_gtid));
+      goto err;
     }
+    else if (unlikely(mi->gtid_reconnect_event_skip_count > 0))
+    {
+      if (mi->gtid_reconnect_event_skip_count ==
+          mi->events_queued_since_last_gtid)
+      {
+        DBUG_ASSERT(event_gtid == mi->last_queued_gtid);
+
+        goto default_action;
+      }
+
+      DBUG_ASSERT(0);
+    }
+    // else_likely{...
+#ifndef DBUG_OFF
+dbug_gtid_accept:
+    DBUG_EXECUTE_IF("slave_discard_gtid_0_x_1002",
+    {
+      if (mi->last_queued_gtid.server_id == 27697 &&
+          mi->last_queued_gtid.seq_no == 1002)
+      {
+        DBUG_SET("-d,slave_discard_gtid_0_x_1002");
+        goto skip_relay_logging;
+      }
+    });
+#endif
     mi->last_queued_gtid= event_gtid;
     mi->last_queued_gtid_standalone=
       (gtid_flag & Gtid_log_event::FL_STANDALONE) != 0;
@@ -6797,7 +6856,7 @@ static int queue_event(Master_info* mi, const uchar *buf, ulong event_len)
 
     ++mi->events_queued_since_last_gtid;
     inc_pos= event_len;
-
+    // ...} eof else_likely
   }
   break;
   /*
@@ -6877,6 +6936,12 @@ static int queue_event(Master_info* mi, const uchar *buf, ulong event_len)
   case XID_EVENT:
     DBUG_EXECUTE_IF("slave_discard_xid_for_gtid_0_x_1000",
     {
+      if (mi->last_queued_gtid.server_id == 27697 &&
+          mi->last_queued_gtid.seq_no == 1000)
+      {
+        DBUG_SET("-d,slave_discard_xid_for_gtid_0_x_1000");
+        goto skip_relay_logging;
+      }
       /* Inject an event group that is missing its XID commit event. */
       if (mi->last_queued_gtid.domain_id == 0 &&
           mi->last_queued_gtid.seq_no == 1000)
@@ -6923,15 +6988,48 @@ static int queue_event(Master_info* mi, const uchar *buf, ulong event_len)
                       }
                     };);
 
-    if (mi->using_gtid != Master_info::USE_GTID_NO && mi->gtid_event_seen)
+    if (mi->using_gtid != Master_info::USE_GTID_NO)
     {
-      if (unlikely(mi->gtid_reconnect_event_skip_count))
+      if (likely(mi->gtid_event_seen))
       {
-        --mi->gtid_reconnect_event_skip_count;
-        gtid_skip_enqueue= true;
+        if (unlikely(mi->gtid_reconnect_event_skip_count))
+        {
+          if (!got_gtid_event &&
+              mi->gtid_reconnect_event_skip_count ==
+              mi->events_queued_since_last_gtid)
+            goto gtid_not_start; // the 1st re-sent must be gtid
+
+          --mi->gtid_reconnect_event_skip_count;
+          gtid_skip_enqueue= true;
+        }
+        else if (likely(mi->events_queued_since_last_gtid))
+        {
+          DBUG_ASSERT(!got_gtid_event);
+
+          ++mi->events_queued_since_last_gtid;
+        }
+        else if (Log_event::is_group_event((Log_event_type) (uchar)
+                                           buf[EVENT_TYPE_OFFSET]))
+        {
+          goto gtid_not_start; // no first gtid event in this group
+        }
+      }
+      else if (Log_event::is_group_event((Log_event_type) (uchar)
+                                           buf[EVENT_TYPE_OFFSET]))
+      {
+      gtid_not_start:
+
+        DBUG_ASSERT(!got_gtid_event);
+
+        error= ER_SLAVE_RELAY_LOG_WRITE_FAILURE;
+        sql_print_error("The current group of events starts with "
+                        "a non-GTID %s event; "
+                        "the last seen GTID is %u-%u-%llu",
+                        Log_event::get_type_str((Log_event_type) (uchar)
+                                                buf[EVENT_TYPE_OFFSET]),
+                        mi->last_queued_gtid);
+        goto err;
       }
-      else if (mi->events_queued_since_last_gtid)
-        ++mi->events_queued_since_last_gtid;
     }
 
     if (!is_compress_event)
@@ -7129,8 +7227,9 @@ static int queue_event(Master_info* mi, const uchar *buf, ulong event_len)
       mi->using_gtid != Master_info::USE_GTID_NO &&
       mi->events_queued_since_last_gtid > 0 &&
       ( (mi->last_queued_gtid_standalone &&
-         !Log_event::is_part_of_group((Log_event_type)(uchar)
-                                      buf[EVENT_TYPE_OFFSET])) ||
+         (LOG_EVENT_IS_QUERY((Log_event_type)(uchar)
+                             buf[EVENT_TYPE_OFFSET]) ||
+          (uchar)buf[EVENT_TYPE_OFFSET] == INCIDENT_EVENT)) ||
         (!mi->last_queued_gtid_standalone &&
          ((uchar)buf[EVENT_TYPE_OFFSET] == XID_EVENT ||
           (uchar)buf[EVENT_TYPE_OFFSET] == XA_PREPARE_LOG_EVENT ||
@@ -7142,11 +7241,13 @@ static int queue_event(Master_info* mi, const uchar *buf, ulong event_len)
         The whole of the current event group is queued. So in case of
         reconnect we can start from after the current GTID.
       */
-      if (mi->gtid_reconnect_event_skip_count)
+      if (gtid_skip_enqueue)
       {
         bool first= true;
         StringBuffer<1024> gtid_text;
 
+        DBUG_ASSERT(mi->events_queued_since_last_gtid > 1);
+
         rpl_slave_state_tostring_helper(&gtid_text, &mi->last_queued_gtid,
                                         &first);
         sql_print_error("Slave IO thread received a terminal event from "
@@ -7156,6 +7257,8 @@ static int queue_event(Master_info* mi, const uchar *buf, ulong event_len)
                         gtid_text.ptr(),
                         mi->gtid_reconnect_event_skip_count,
                         mi->events_queued_since_last_gtid);
+        error= ER_SLAVE_RELAY_LOG_WRITE_FAILURE;
+        goto err;
       }
       mi->gtid_current_pos.update(&mi->last_queued_gtid);
       mi->events_queued_since_last_gtid= 0;
diff --git a/sql/sp_cache.cc b/sql/sp_cache.cc
index b3949a94751..36ad37104bd 100644
--- a/sql/sp_cache.cc
+++ b/sql/sp_cache.cc
@@ -312,3 +312,12 @@ sp_cache::cleanup()
 {
   my_hash_free(&m_hashtable);
 }
+
+
+void Sp_caches::sp_caches_clear()
+{
+  sp_cache_clear(&sp_proc_cache);
+  sp_cache_clear(&sp_func_cache);
+  sp_cache_clear(&sp_package_spec_cache);
+  sp_cache_clear(&sp_package_body_cache);
+}
diff --git a/sql/sp_head.cc b/sql/sp_head.cc
index fab72fd0c6f..bdd94fd6d0e 100644
--- a/sql/sp_head.cc
+++ b/sql/sp_head.cc
@@ -830,7 +830,7 @@ void
 sp_head::set_stmt_end(THD *thd)
 {
   Lex_input_stream *lip= & thd->m_parser_state->m_lip; /* shortcut */
-  const char *end_ptr= lip->get_cpp_ptr(); /* shortcut */
+  const char *end_ptr= lip->get_cpp_tok_start(); /* shortcut */
 
   /* Make the string of parameters. */
 
diff --git a/sql/sql_alter.cc b/sql/sql_alter.cc
index 86c6e9a27f8..1d5875733ac 100644
--- a/sql/sql_alter.cc
+++ b/sql/sql_alter.cc
@@ -411,7 +411,7 @@ bool Sql_cmd_alter_table::execute(THD *thd)
     referenced from this structure will be modified.
     @todo move these into constructor...
   */
-  HA_CREATE_INFO create_info(lex->create_info);
+  Table_specification_st create_info(lex->create_info);
   Alter_info alter_info(lex->alter_info, thd->mem_root);
   create_info.alter_info= &alter_info;
   privilege_t priv(NO_ACL);
diff --git a/sql/sql_base.cc b/sql/sql_base.cc
index 50371b3e722..18ffdc9e174 100644
--- a/sql/sql_base.cc
+++ b/sql/sql_base.cc
@@ -819,7 +819,14 @@ int close_thread_tables(THD *thd)
                           table->s->table_name.str, (ulong) table->query_id));
 
     if (thd->locked_tables_mode)
+    {
+#ifdef WITH_PARTITION_STORAGE_ENGINE
+      if (table->part_info && table->part_info->vers_require_hist_part(thd) &&
+          !thd->stmt_arena->is_stmt_prepare())
+        table->part_info->vers_check_limit(thd);
+#endif
       table->vcol_cleanup_expr(thd);
+    }
 
     /* Detach MERGE children after every statement. Even under LOCK TABLES. */
     if (thd->locked_tables_mode <= LTM_LOCK_TABLES ||
@@ -8664,9 +8671,11 @@ int setup_conds(THD *thd, TABLE_LIST *tables, List<TABLE_LIST> &leaves,
     thd->lex->which_check_option_applicable();
   bool save_is_item_list_lookup= select_lex->is_item_list_lookup;
   TABLE_LIST *derived= select_lex->master_unit()->derived;
+  bool save_resolve_in_select_list= select_lex->context.resolve_in_select_list;
   DBUG_ENTER("setup_conds");
 
   select_lex->is_item_list_lookup= 0;
+  select_lex->context.resolve_in_select_list= false;
 
   thd->column_usage= MARK_COLUMNS_READ;
   DBUG_PRINT("info", ("thd->column_usage: %d", thd->column_usage));
@@ -8719,7 +8728,8 @@ int setup_conds(THD *thd, TABLE_LIST *tables, List<TABLE_LIST> &leaves,
     select_lex->where= *conds;
   }
   thd->lex->current_select->is_item_list_lookup= save_is_item_list_lookup;
-  DBUG_RETURN(MY_TEST(thd->is_error()));
+  select_lex->context.resolve_in_select_list= save_resolve_in_select_list;
+  DBUG_RETURN(thd->is_error());
 
 err_no_arena:
   select_lex->is_item_list_lookup= save_is_item_list_lookup;
diff --git a/sql/sql_class.cc b/sql/sql_class.cc
index f902f12481e..a0ba1d7e5a3 100644
--- a/sql/sql_class.cc
+++ b/sql/sql_class.cc
@@ -1,6 +1,6 @@
 /*
    Copyright (c) 2000, 2015, Oracle and/or its affiliates.
-   Copyright (c) 2008, 2021, MariaDB Corporation.
+   Copyright (c) 2008, 2022, MariaDB Corporation.
 
    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
@@ -56,6 +56,7 @@
 #include "sp_rcontext.h"
 #include "sp_cache.h"
 #include "sql_show.h"                           // append_identifier
+#include "sql_db.h"                             // get_default_db_collation
 #include "transaction.h"
 #include "sql_select.h" /* declares create_tmp_table() */
 #include "debug_sync.h"
@@ -679,7 +680,8 @@ THD::THD(my_thread_id id, bool is_wsrep_applier)
 #ifdef HAVE_REPLICATION
    ,
    current_linfo(0),
-   slave_info(0)
+   slave_info(0),
+   is_awaiting_semisync_ack(0)
 #endif
 #ifdef WITH_WSREP
    ,
@@ -861,11 +863,6 @@ THD::THD(my_thread_id id, bool is_wsrep_applier)
                get_sequence_last_key, (my_hash_free_key) free_sequence_last,
                HASH_THREAD_SPECIFIC);
 
-  sp_proc_cache= NULL;
-  sp_func_cache= NULL;
-  sp_package_spec_cache= NULL;
-  sp_package_body_cache= NULL;
-
   /* For user vars replication*/
   if (opt_bin_log)
     my_init_dynamic_array(key_memory_user_var_entry, &user_var_events,
@@ -1439,10 +1436,7 @@ void THD::change_user(void)
                SEQUENCES_HASH_SIZE, 0, 0, (my_hash_get_key)
                get_sequence_last_key, (my_hash_free_key) free_sequence_last,
                HASH_THREAD_SPECIFIC);
-  sp_cache_clear(&sp_proc_cache);
-  sp_cache_clear(&sp_func_cache);
-  sp_cache_clear(&sp_package_spec_cache);
-  sp_cache_clear(&sp_package_body_cache);
+  sp_caches_clear();
   opt_trace.delete_traces();
 }
 
@@ -1569,10 +1563,7 @@ void THD::cleanup(void)
 
   my_hash_free(&user_vars);
   my_hash_free(&sequences);
-  sp_cache_clear(&sp_proc_cache);
-  sp_cache_clear(&sp_func_cache);
-  sp_cache_clear(&sp_package_spec_cache);
-  sp_cache_clear(&sp_package_body_cache);
+  sp_caches_clear();
   auto_inc_intervals_forced.empty();
   auto_inc_intervals_in_cur_stmt_for_binlog.empty();
 
@@ -6741,49 +6732,86 @@ int THD::decide_logging_format(TABLE_LIST *tables)
   DBUG_RETURN(0);
 }
 
-int THD::decide_logging_format_low(TABLE *table)
+
+/*
+  Reconsider logging format in case of INSERT...ON DUPLICATE KEY UPDATE
+  for tables with more than one unique keys in case of MIXED binlog format.
+
+  Unsafe means that a master could execute the statement differently than
+  the slave.
+  This could can happen in the following cases:
+  - The unique check are done in different order on master or slave
+    (different engine or different key order).
+  - There is a conflict on another key than the first and before the
+    statement is committed, another connection commits a row that conflicts
+    on an earlier unique key. Example follows:
+
+    Below a and b are unique keys, the table has a row (1,1,0)
+    connection 1:
+    INSERT INTO t1 set a=2,b=1,c=0 ON DUPLICATE KEY UPDATE c=1;
+    connection 2:
+    INSERT INTO t1 set a=2,b=2,c=0;
+
+    If 2 commits after 1 has been executed but before 1 has committed
+    (and are thus put before the other in the binary log), one will
+    get different data on the slave:
+    (1,1,1),(2,2,1) instead of (1,1,1),(2,2,0)
+*/
+
+void THD::reconsider_logging_format_for_iodup(TABLE *table)
 {
-  DBUG_ENTER("decide_logging_format_low");
-  /*
-    INSERT...ON DUPLICATE KEY UPDATE on a table with more than one unique keys
-    can be unsafe.
-  */
-  if (wsrep_binlog_format() <= BINLOG_FORMAT_STMT &&
-      !is_current_stmt_binlog_format_row() &&
-      !lex->is_stmt_unsafe() &&
-      lex->duplicates == DUP_UPDATE)
+  DBUG_ENTER("reconsider_logging_format_for_iodup");
+  enum_binlog_format bf= (enum_binlog_format) wsrep_binlog_format();
+
+  DBUG_ASSERT(lex->duplicates == DUP_UPDATE);
+
+  if (bf <= BINLOG_FORMAT_STMT &&
+      !is_current_stmt_binlog_format_row())
   {
+    KEY *end= table->s->key_info + table->s->keys;
     uint unique_keys= 0;
-    uint keys= table->s->keys, i= 0;
-    Field *field;
-    for (KEY* keyinfo= table->s->key_info;
-             i < keys && unique_keys <= 1; i++, keyinfo++)
-      if (keyinfo->flags & HA_NOSAME &&
-         !(keyinfo->key_part->field->flags & AUTO_INCREMENT_FLAG &&
-             //User given auto inc can be unsafe
-             !keyinfo->key_part->field->val_int()))
+
+    for (KEY *keyinfo= table->s->key_info; keyinfo < end ; keyinfo++)
+    {
+      if (keyinfo->flags & HA_NOSAME)
       {
+        /*
+          We assume that the following cases will guarantee that the
+          key is unique if a key part is not set:
+          - The key part is an autoincrement (autogenerated)
+          - The key part has a default value that is null and it not
+            a virtual field that will be calculated later.
+        */
         for (uint j= 0; j < keyinfo->user_defined_key_parts; j++)
         {
-          field= keyinfo->key_part[j].field;
-          if(!bitmap_is_set(table->write_set,field->field_index))
-            goto exit;
+          Field *field= keyinfo->key_part[j].field;
+          if (!bitmap_is_set(table->write_set, field->field_index))
+          {
+            /* Check auto_increment */
+            if (field == table->next_number_field)
+              goto exit;
+            if (field->is_real_null() && !field->default_value)
+              goto exit;
+          }
         }
-        unique_keys++;
+        if (unique_keys++)
+          break;
 exit:;
       }
-
+    }
     if (unique_keys > 1)
     {
-      lex->set_stmt_unsafe(LEX::BINLOG_STMT_UNSAFE_INSERT_TWO_KEYS);
-      binlog_unsafe_warning_flags|= lex->get_stmt_unsafe_flags();
+      if (bf == BINLOG_FORMAT_STMT && !lex->is_stmt_unsafe())
+      {
+        lex->set_stmt_unsafe(LEX::BINLOG_STMT_UNSAFE_INSERT_TWO_KEYS);
+        binlog_unsafe_warning_flags|= lex->get_stmt_unsafe_flags();
+      }
       set_current_stmt_binlog_format_row_if_mixed();
       if (is_current_stmt_binlog_format_row())
         binlog_prepare_for_row_logging();
-      DBUG_RETURN(1);
     }
   }
-  DBUG_RETURN(0);
+  DBUG_VOID_RETURN;
 }
 
 #ifndef MYSQL_CLIENT
@@ -8269,3 +8297,27 @@ THD_list_iterator *THD_list_iterator::iterator()
 {
   return &server_threads;
 }
+
+
+Charset_collation_context
+THD::charset_collation_context_alter_db(const char *db)
+{
+  return Charset_collation_context(variables.collation_server,
+                                   get_default_db_collation(this, db));
+}
+
+
+Charset_collation_context
+THD::charset_collation_context_create_table_in_db(const char *db)
+{
+  CHARSET_INFO *cs= get_default_db_collation(this, db);
+  return Charset_collation_context(cs, cs);
+}
+
+
+Charset_collation_context
+THD::charset_collation_context_alter_table(const TABLE_SHARE *s)
+{
+  return Charset_collation_context(get_default_db_collation(this, s->db.str),
+                                   s->table_charset);
+}
diff --git a/sql/sql_class.h b/sql/sql_class.h
index 47d4f9fee6a..806f77c1ba2 100644
--- a/sql/sql_class.h
+++ b/sql/sql_class.h
@@ -1,6 +1,6 @@
 /*
    Copyright (c) 2000, 2016, Oracle and/or its affiliates.
-   Copyright (c) 2009, 2021, MariaDB Corporation.
+   Copyright (c) 2009, 2022, MariaDB Corporation.
 
    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
@@ -1185,7 +1185,7 @@ public:
   /* We build without RTTI, so dynamic_cast can't be used. */
   enum Type
   {
-    STATEMENT, PREPARED_STATEMENT, STORED_PROCEDURE, TABLE_ARENA
+    STATEMENT, PREPARED_STATEMENT, STORED_PROCEDURE
   };
 
   Query_arena(MEM_ROOT *mem_root_arg, enum enum_state state_arg) :
@@ -2367,6 +2367,39 @@ struct wait_for_commit
   void reinit();
 };
 
+
+class Sp_caches
+{
+public:
+  sp_cache *sp_proc_cache;
+  sp_cache *sp_func_cache;
+  sp_cache *sp_package_spec_cache;
+  sp_cache *sp_package_body_cache;
+  Sp_caches()
+   :sp_proc_cache(NULL),
+    sp_func_cache(NULL),
+    sp_package_spec_cache(NULL),
+    sp_package_body_cache(NULL)
+  { }
+  ~Sp_caches()
+  {
+    // All caches must be freed by the caller explicitly
+    DBUG_ASSERT(sp_proc_cache == NULL);
+    DBUG_ASSERT(sp_func_cache == NULL);
+    DBUG_ASSERT(sp_package_spec_cache == NULL);
+    DBUG_ASSERT(sp_package_body_cache == NULL);
+  }
+  void sp_caches_swap(Sp_caches &rhs)
+  {
+    swap_variables(sp_cache*, sp_proc_cache, rhs.sp_proc_cache);
+    swap_variables(sp_cache*, sp_func_cache, rhs.sp_func_cache);
+    swap_variables(sp_cache*, sp_package_spec_cache, rhs.sp_package_spec_cache);
+    swap_variables(sp_cache*, sp_package_body_cache, rhs.sp_package_body_cache);
+  }
+  void sp_caches_clear();
+};
+
+
 extern "C" void my_message_sql(uint error, const char *str, myf MyFlags);
 
 
@@ -2562,7 +2595,8 @@ class THD: public THD_count, /* this must be first */
            */
            public Item_change_list,
            public MDL_context_owner,
-           public Open_tables_state
+           public Open_tables_state,
+           public Sp_caches
 {
 private:
   inline bool is_stmt_prepare() const
@@ -3606,10 +3640,6 @@ public:
   enum_sql_command last_sql_command;  // Last sql_command exceuted in mysql_execute_command()
 
   sp_rcontext *spcont;		// SP runtime context
-  sp_cache   *sp_proc_cache;
-  sp_cache   *sp_func_cache;
-  sp_cache   *sp_package_spec_cache;
-  sp_cache   *sp_package_body_cache;
 
   /** number of name_const() substitutions, see sp_head.cc:subst_spvars() */
   uint       query_name_consts;
@@ -4407,8 +4437,7 @@ public:
 
   bool is_item_tree_change_register_required()
   {
-    return !stmt_arena->is_conventional()
-           || stmt_arena->type() == Query_arena::TABLE_ARENA;
+    return !stmt_arena->is_conventional();
   }
 
   void change_item_tree(Item **place, Item *new_value)
@@ -4998,18 +5027,18 @@ public:
       mdl_context.release_transactional_locks(this);
   }
   int decide_logging_format(TABLE_LIST *tables);
+
   /*
-   In Some cases when decide_logging_format is called it does not have all
-   information to decide the logging format. So that cases we call decide_logging_format_2
-   at later stages in execution.
-   One example would be binlog format for IODKU but column with unique key is not inserted.
-   We don't have inserted columns info when we call decide_logging_format so on later stage we call
-   decide_logging_format_low
+   In Some cases when decide_logging_format is called it does not have
+   all information to decide the logging format. So that cases we call
+   decide_logging_format_2 at later stages in execution.
 
-   @returns 0 if no format is changed
-            1 if there is change in binlog format
+   One example would be binlog format for insert on duplicate key
+   (IODKU) but column with unique key is not inserted.  We do not have
+   inserted columns info when we call decide_logging_format so on
+   later stage we call reconsider_logging_format_for_iodup()
   */
-  int decide_logging_format_low(TABLE *table);
+  void reconsider_logging_format_for_iodup(TABLE *table);
 
   enum need_invoker { INVOKER_NONE=0, INVOKER_USER, INVOKER_ROLE};
   void binlog_invoker(bool role) { m_binlog_invoker= role ? INVOKER_ROLE : INVOKER_USER; }
@@ -5259,6 +5288,14 @@ public:
   bool is_binlog_dump_thread();
 #endif
 
+  /*
+    Indicates if this thread is suspended due to awaiting an ACK from a
+    replica. True if suspended, false otherwise.
+
+    Note that this variable is protected by Repl_semi_sync_master::LOCK_binlog
+  */
+  bool is_awaiting_semisync_ack;
+
   inline ulong wsrep_binlog_format() const
   {
     return WSREP_BINLOG_FORMAT(variables.binlog_format);
@@ -5484,6 +5521,19 @@ public:
             MY_UTF8_IS_UTF8MB3 : 0);
   }
 
+  Charset_collation_context
+    charset_collation_context_create_db() const
+  {
+    return Charset_collation_context(variables.collation_server,
+                                     variables.collation_server);
+  }
+  Charset_collation_context
+    charset_collation_context_alter_db(const char *db);
+  Charset_collation_context
+    charset_collation_context_create_table_in_db(const char *db);
+  Charset_collation_context
+    charset_collation_context_alter_table(const TABLE_SHARE *s);
+
   /**
     Save current lex to the output parameter and reset it to point to
     main_lex. This method is called from mysql_client_binlog_statement()
@@ -5581,8 +5631,8 @@ my_eof(THD *thd)
 
 inline date_conv_mode_t sql_mode_for_dates(THD *thd)
 {
-  static_assert((date_conv_mode_t::KNOWN_MODES &
-                time_round_mode_t::KNOWN_MODES) == 0,
+  static_assert((ulonglong(date_conv_mode_t::KNOWN_MODES) &
+                 ulonglong(time_round_mode_t::KNOWN_MODES)) == 0,
                 "date_conv_mode_t and time_round_mode_t must use different "
                 "bit values");
   static_assert(MODE_NO_ZERO_DATE    == date_mode_t::NO_ZERO_DATE &&
@@ -6103,6 +6153,7 @@ public:
     m_plock(NULL), exit_done(0),
     saved_tmp_table_share(0)
     {
+      DBUG_ASSERT(create_info->default_table_charset);
       bzero(&ddl_log_state_create, sizeof(ddl_log_state_create));
       bzero(&ddl_log_state_rm, sizeof(ddl_log_state_rm));
     }
diff --git a/sql/sql_db.cc b/sql/sql_db.cc
index c5defc1959c..9da1ac5ca77 100644
--- a/sql/sql_db.cc
+++ b/sql/sql_db.cc
@@ -943,6 +943,7 @@ exit:
 int mysql_create_db(THD *thd, const LEX_CSTRING *db, DDL_options_st options,
                     const Schema_specification_st *create_info)
 {
+  DBUG_ASSERT(create_info->default_table_charset);
   /*
     As mysql_create_db_internal() may modify Db_create_info structure passed
     to it, we need to use a copy to make execution prepared statement- safe.
@@ -958,6 +959,7 @@ int mysql_create_db(THD *thd, const LEX_CSTRING *db, DDL_options_st options,
 bool mysql_alter_db(THD *thd, const LEX_CSTRING *db,
                     const Schema_specification_st *create_info)
 {
+  DBUG_ASSERT(create_info->default_table_charset);
   /*
     As mysql_alter_db_internal() may modify Db_create_info structure passed
     to it, we need to use a copy to make execution prepared statement- safe.
diff --git a/sql/sql_handler.cc b/sql/sql_handler.cc
index 833492d3d9f..7235dc6472d 100644
--- a/sql/sql_handler.cc
+++ b/sql/sql_handler.cc
@@ -712,8 +712,10 @@ mysql_ha_fix_cond_and_key(SQL_HANDLER *handler,
         if (!in_prepare)
         {
           MY_BITMAP *old_map= dbug_tmp_use_all_columns(table, &table->write_set);
-          (void) item->save_in_field(key_part->field, 1);
+          int res= item->save_in_field(key_part->field, 1);
           dbug_tmp_restore_column_map(&table->write_set, old_map);
+          if (res)
+            return 1;
         }
         key_len+= key_part->store_length;
         keypart_map= (keypart_map << 1) | 1;
diff --git a/sql/sql_i_s.h b/sql/sql_i_s.h
index bed2e886718..a3614d889c9 100644
--- a/sql/sql_i_s.h
+++ b/sql/sql_i_s.h
@@ -158,10 +158,10 @@ public:
 };
 
 
-class Yesno: public Varchar
+class Yes_or_empty: public Varchar
 {
 public:
-  Yesno(): Varchar(3) { }
+  Yes_or_empty(): Varchar(3) { }
 };
 
 
diff --git a/sql/sql_insert.cc b/sql/sql_insert.cc
index e498daca83c..d9d22dbe514 100644
--- a/sql/sql_insert.cc
+++ b/sql/sql_insert.cc
@@ -1,6 +1,6 @@
 /*
    Copyright (c) 2000, 2016, Oracle and/or its affiliates.
-   Copyright (c) 2010, 2021, MariaDB
+   Copyright (c) 2010, 2022, MariaDB Corporation
 
    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
@@ -995,7 +995,12 @@ bool mysql_insert(THD *thd, TABLE_LIST *table_list,
     goto values_loop_end;
 
   THD_STAGE_INFO(thd, stage_update);
-  thd->decide_logging_format_low(table);
+
+  if  (duplic == DUP_UPDATE)
+  {
+    restore_record(table,s->default_values);	// Get empty record
+    thd->reconsider_logging_format_for_iodup(table);
+  }
   fix_rownum_pointers(thd, thd->lex->current_select, &info.accepted_rows);
   if (returning)
     fix_rownum_pointers(thd, thd->lex->returning(), &info.accepted_rows);
diff --git a/sql/sql_lex.cc b/sql/sql_lex.cc
index 18d54bf0d69..5f2f072b348 100644
--- a/sql/sql_lex.cc
+++ b/sql/sql_lex.cc
@@ -719,7 +719,6 @@ init_lex_with_single_table(THD *thd, TABLE *table, LEX *lex)
     return TRUE;
   context->resolve_in_table_list_only(table_list);
   lex->use_only_table_context= TRUE;
-  lex->context_analysis_only|= CONTEXT_ANALYSIS_ONLY_VCOL_EXPR;
   select_lex->cur_pos_in_select_list= UNDEF_POS;
   table->map= 1; //To ensure correct calculation of const item
   table_list->table= table;
@@ -1212,6 +1211,7 @@ void LEX::start(THD *thd_arg)
   context_stack.empty();
   //empty select_stack
   select_stack_top= 0;
+  select_stack_outer_barrier= 0;
   unit.init_query();
   current_select_number= 0;
   curr_with_clause= 0;
@@ -5296,17 +5296,21 @@ void SELECT_LEX::update_used_tables()
   while ((tl= ti++))
   {
     TABLE_LIST *embedding= tl;
-    do
+    if (!is_eliminated_table(join->eliminated_tables, tl))
     {
-      bool maybe_null;
-      if ((maybe_null= MY_TEST(embedding->outer_join)))
+      do
       {
-        tl->table->maybe_null= maybe_null;
-        break;
+        bool maybe_null;
+        if ((maybe_null= MY_TEST(embedding->outer_join)))
+        {
+          tl->table->maybe_null= maybe_null;
+          break;
+        }
       }
+      while ((embedding= embedding->embedding));
     }
-    while ((embedding= embedding->embedding));
-    if (tl->on_expr)
+
+    if (tl->on_expr && !is_eliminated_table(join->eliminated_tables, tl))
     {
       tl->on_expr->update_used_tables();
       tl->on_expr->walk(&Item::eval_not_null_tables, 0, NULL);
@@ -5333,8 +5337,11 @@ void SELECT_LEX::update_used_tables()
       if (embedding->on_expr && 
           embedding->nested_join->join_list.head() == tl)
       {
-        embedding->on_expr->update_used_tables();
-        embedding->on_expr->walk(&Item::eval_not_null_tables, 0, NULL);
+        if (!is_eliminated_table(join->eliminated_tables, embedding))
+        {
+          embedding->on_expr->update_used_tables();
+          embedding->on_expr->walk(&Item::eval_not_null_tables, 0, NULL);
+        }
       }
       tl= embedding;
       embedding= tl->embedding;
@@ -7800,7 +7807,7 @@ bool LEX::maybe_start_compound_statement(THD *thd)
     if (!make_sp_head(thd, NULL, &sp_handler_procedure, DEFAULT_AGGREGATE))
       return true;
     sphead->set_suid(SP_IS_NOT_SUID);
-    sphead->set_body_start(thd, thd->m_parser_state->m_lip.get_cpp_ptr());
+    sphead->set_body_start(thd, thd->m_parser_state->m_lip.get_cpp_tok_start());
   }
   return false;
 }
@@ -11867,6 +11874,21 @@ bool LEX::sp_create_set_password_instr(THD *thd,
 }
 
 
+bool LEX::set_names(const char *pos,
+                    const Lex_exact_charset_opt_extended_collate &cscl,
+                    bool no_lookahead)
+{
+  if (sp_create_assignment_lex(thd, pos))
+    return true;
+  CHARSET_INFO *ci= cscl.collation().charset_info();
+  set_var_collation_client *var;
+  var= new (thd->mem_root) set_var_collation_client(ci, ci, ci);
+  return unlikely(var == NULL) ||
+         unlikely(thd->lex->var_list.push_back(var, thd->mem_root)) ||
+         unlikely(sp_create_assignment_instr(thd, no_lookahead));
+}
+
+
 bool LEX::map_data_type(const Lex_ident_sys_st &schema_name,
                         Lex_field_type_st *type) const
 {
diff --git a/sql/sql_lex.h b/sql/sql_lex.h
index a6e7546d38c..4f2e775a7fc 100644
--- a/sql/sql_lex.h
+++ b/sql/sql_lex.h
@@ -2047,8 +2047,7 @@ public:
     @retval nonzero if the statement is a row injection
   */
   inline bool is_stmt_row_injection() const {
-    return binlog_stmt_flags &
-      (1U << (BINLOG_STMT_UNSAFE_COUNT + BINLOG_STMT_TYPE_ROW_INJECTION));
+    return binlog_stmt_flags & (1U << BINLOG_STMT_TYPE_ROW_INJECTION);
   }
 
   /**
@@ -2058,8 +2057,7 @@ public:
   */
   inline void set_stmt_row_injection() {
     DBUG_ENTER("set_stmt_row_injection");
-    binlog_stmt_flags|=
-      (1U << (BINLOG_STMT_UNSAFE_COUNT + BINLOG_STMT_TYPE_ROW_INJECTION));
+    binlog_stmt_flags|= (1U << BINLOG_STMT_TYPE_ROW_INJECTION);
     DBUG_VOID_RETURN;
   }
 
@@ -2335,7 +2333,7 @@ private:
       The statement is a row injection (i.e., either a BINLOG
       statement or a row event executed by the slave SQL thread).
     */
-    BINLOG_STMT_TYPE_ROW_INJECTION = 0,
+    BINLOG_STMT_TYPE_ROW_INJECTION = BINLOG_STMT_UNSAFE_COUNT,
 
     /** The last element of this enumeration type. */
     BINLOG_STMT_TYPE_COUNT
@@ -2349,8 +2347,8 @@ private:
     - The low BINLOG_STMT_UNSAFE_COUNT bits indicate the types of
       unsafeness that the current statement has.
 
-    - The next BINLOG_STMT_TYPE_COUNT bits indicate if the statement
-      is of some special type.
+      - The next BINLOG_STMT_TYPE_COUNT-BINLOG_STMT_TYPE_COUNT bits indicate if
+      the statement is of some special type.
 
     This must be a member of LEX, not of THD: each stored procedure
     needs to remember its unsafeness state between calls and each
@@ -3309,6 +3307,12 @@ public:
   List<Name_resolution_context> context_stack;
   SELECT_LEX *select_stack[MAX_SELECT_NESTING + 1];
   uint select_stack_top;
+  /*
+    Usually this is set to 0, but for INSERT/REPLACE SELECT it is set to 1.
+    When parsing such statements the pointer to the most outer select is placed
+    into the second element of select_stack rather than into the first.
+  */
+  uint select_stack_outer_barrier;
 
   SQL_I_List<ORDER> proc_list;
   SQL_I_List<TABLE_LIST> auxiliary_table_list, save_list;
@@ -3752,6 +3756,17 @@ public:
 
   bool copy_db_to(LEX_CSTRING *to);
 
+  void inc_select_stack_outer_barrier()
+  {
+    select_stack_outer_barrier++;
+  }
+
+  SELECT_LEX *parser_current_outer_select()
+  {
+    return select_stack_top - 1 == select_stack_outer_barrier ?
+             0 : select_stack[select_stack_top - 2];
+  }
+
   Name_resolution_context *current_context()
   {
     return context_stack.head();
@@ -3818,6 +3833,9 @@ public:
 
   int case_stmt_action_then();
   bool setup_select_in_parentheses();
+  bool set_names(const char *pos,
+                 const Lex_exact_charset_opt_extended_collate &cs,
+                 bool no_lookahead);
   bool set_trigger_new_row(const LEX_CSTRING *name, Item *val);
   bool set_trigger_field(const LEX_CSTRING *name1, const LEX_CSTRING *name2,
                          Item *val);
@@ -4390,6 +4408,23 @@ public:
   bool add_alter_list(LEX_CSTRING par_name, Virtual_column_info *expr,
                       bool par_exists);
   bool add_alter_list(LEX_CSTRING name, LEX_CSTRING new_name, bool exists);
+  bool add_alter_list_item_convert_to_charset(CHARSET_INFO *cs)
+  {
+    if (create_info.add_table_option_convert_charset(cs))
+      return true;
+    alter_info.flags|= ALTER_CONVERT_TO;
+    return false;
+  }
+  bool
+  add_alter_list_item_convert_to_charset(CHARSET_INFO *cs,
+                                         const Lex_extended_collation_st &cl)
+  {
+    if (create_info.add_table_option_convert_charset(cs) ||
+        create_info.add_table_option_convert_collation(cl))
+      return true;
+    alter_info.flags|= ALTER_CONVERT_TO;
+    return false;
+  }
   void set_command(enum_sql_command command,
                    DDL_options_st options)
   {
@@ -4529,6 +4564,29 @@ public:
     return create_info.vers_info;
   }
 
+  /* The list of history-generating DML commands */
+  bool vers_history_generating() const
+  {
+    switch (sql_command)
+    {
+      case SQLCOM_DELETE:
+        return !vers_conditions.delete_history;
+      case SQLCOM_UPDATE:
+      case SQLCOM_UPDATE_MULTI:
+      case SQLCOM_DELETE_MULTI:
+      case SQLCOM_REPLACE:
+      case SQLCOM_REPLACE_SELECT:
+        return true;
+      case SQLCOM_INSERT:
+      case SQLCOM_INSERT_SELECT:
+        return duplicates == DUP_UPDATE;
+      case SQLCOM_LOAD:
+        return duplicates == DUP_REPLACE;
+      default:
+        return false;
+    }
+  }
+
   int add_period(Lex_ident name, Lex_ident_sys_st start, Lex_ident_sys_st end)
   {
     if (lex_string_cmp(system_charset_info, &start, &end) == 0)
diff --git a/sql/sql_manager.cc b/sql/sql_manager.cc
index df6847f2ba1..3d3728b9e00 100644
--- a/sql/sql_manager.cc
+++ b/sql/sql_manager.cc
@@ -134,6 +134,7 @@ void start_handle_manager()
   {
     pthread_t hThread;
     int err;
+    DBUG_EXECUTE_IF("delay_start_handle_manager", my_sleep(1000););
     manager_thread_in_use = 1;
     mysql_cond_init(key_COND_manager, &COND_manager,NULL);
     mysql_mutex_init(key_LOCK_manager, &LOCK_manager, NULL);
diff --git a/sql/sql_parse.cc b/sql/sql_parse.cc
index 0ac082b1812..0597b086b7b 100644
--- a/sql/sql_parse.cc
+++ b/sql/sql_parse.cc
@@ -43,9 +43,7 @@
                               // mysql_alter_db,
                               // check_db_dir_existence,
                               // my_dbopt_cleanup
-#include "sql_table.h"        // mysql_create_like_table,
-                              // mysql_create_table,
-                              // mysql_alter_table,
+#include "sql_table.h"        // mysql_alter_table,
                               // mysql_backup_table,
                               // mysql_restore_table
 #include "sql_reload.h"       // reload_acl_and_cache
@@ -1507,22 +1505,6 @@ static bool deny_updates_if_read_only_option(THD *thd, TABLE_LIST *all_tables)
 }
 
 #ifdef WITH_WSREP
-static my_bool wsrep_read_only_option(THD *thd, TABLE_LIST *all_tables)
-{
-  int opt_readonly_saved = opt_readonly;
-  privilege_t flag_saved= thd->security_ctx->master_access & PRIV_IGNORE_READ_ONLY;
-
-  opt_readonly = 0;
-  thd->security_ctx->master_access &= ~PRIV_IGNORE_READ_ONLY;
-
-  my_bool ret = !deny_updates_if_read_only_option(thd, all_tables);
-
-  opt_readonly = opt_readonly_saved;
-  thd->security_ctx->master_access |= flag_saved;
-
-  return ret;
-}
-
 static void wsrep_copy_query(THD *thd)
 {
   thd->wsrep_retry_command   = thd->get_command();
@@ -4206,7 +4188,7 @@ mysql_execute_command(THD *thd, bool is_called_from_prepared_stmt)
   */
   {
     /* Prepare stack copies to be re-execution safe */
-    HA_CREATE_INFO create_info;
+    Table_specification_st create_info;
     Alter_info alter_info(lex->alter_info, thd->mem_root);
 
     if (unlikely(thd->is_fatal_error)) /* out of memory creating alter_info */
@@ -4216,10 +4198,9 @@ mysql_execute_command(THD *thd, bool is_called_from_prepared_stmt)
     if (check_one_table_access(thd, INDEX_ACL, all_tables))
       goto error; /* purecov: inspected */
 
-    bzero((char*) &create_info, sizeof(create_info));
+    create_info.init();
     create_info.db_type= 0;
     create_info.row_type= ROW_TYPE_NOT_USED;
-    create_info.default_table_charset= thd->variables.collation_database;
     create_info.alter_info= &alter_info;
 
     WSREP_TO_ISOLATION_BEGIN(first_table->db.str, first_table->table_name.str, NULL);
@@ -5178,6 +5159,10 @@ mysql_execute_command(THD *thd, bool is_called_from_prepared_stmt)
                           &lex->name))
       break;
 
+    if ((res= lex->create_info.resolve_to_charset_collation_context(thd,
+                                 thd->charset_collation_context_create_db())))
+      break;
+
     WSREP_TO_ISOLATION_BEGIN(lex->name.str, NULL, NULL);
 
     res= mysql_create_db(thd, &lex->name,
@@ -5239,6 +5224,10 @@ mysql_execute_command(THD *thd, bool is_called_from_prepared_stmt)
     if (prepare_db_action(thd, ALTER_ACL, db))
       break;
 
+    if ((res= lex->create_info.resolve_to_charset_collation_context(thd,
+                     thd->charset_collation_context_alter_db(lex->name.str))))
+      break;
+
     WSREP_TO_ISOLATION_BEGIN(db->str, NULL, NULL);
 
     res= mysql_alter_db(thd, db, &lex->create_info);
@@ -7852,7 +7841,7 @@ static bool wsrep_mysql_parse(THD *thd, char *rawbuf, uint length,
 {
   bool is_autocommit=
     !thd->in_multi_stmt_transaction_mode()                  &&
-    wsrep_read_only_option(thd, thd->lex->query_tables);
+    !thd->wsrep_applier;
   bool retry_autocommit;
   do
   {
@@ -8890,6 +8879,7 @@ bool st_select_lex::add_window_def(THD *thd,
     fields_in_window_functions+= win_part_list_ptr->elements +
                                  win_order_list_ptr->elements;
   }
+  win_def->win_spec_number= window_specs.elements;
   return (win_def == NULL || window_specs.push_back(win_def));
 }
 
@@ -8917,6 +8907,7 @@ bool st_select_lex::add_window_spec(THD *thd,
                                  win_order_list_ptr->elements;
   }
   thd->lex->win_spec= win_spec;
+  win_spec->win_spec_number= window_specs.elements;
   return (win_spec == NULL || window_specs.push_back(win_spec));
 }
 
@@ -9060,9 +9051,7 @@ push_new_name_resolution_context(THD *thd,
     right_op->last_leaf_for_name_resolution();
   LEX *lex= thd->lex;
   on_context->select_lex = lex->current_select;
-  st_select_lex *curr_select= lex->pop_select();
-  st_select_lex *outer_sel= lex->select_stack_head();
-  lex->push_select(curr_select);
+  st_select_lex *outer_sel= lex->parser_current_outer_select();
   on_context->outer_context = outer_sel ? &outer_sel->context : 0;
   return lex->push_context(on_context);
 }
@@ -10488,40 +10477,6 @@ bool parse_sql(THD *thd, Parser_state *parser_state,
 */
 
 
-
-/**
-  Check and merge "CHARACTER SET cs [ COLLATE cl ]" clause
-
-  @param cs character set pointer.
-  @param cl collation pointer.
-
-  Check if collation "cl" is applicable to character set "cs".
-
-  If "cl" is NULL (e.g. when COLLATE clause is not specified),
-  then simply "cs" is returned.
-  
-  @return Error status.
-    @retval NULL, if "cl" is not applicable to "cs".
-    @retval pointer to merged CHARSET_INFO on success.
-*/
-
-
-CHARSET_INFO*
-merge_charset_and_collation(CHARSET_INFO *cs, CHARSET_INFO *cl)
-{
-  if (cl)
-  {
-    if (!my_charset_same(cs, cl))
-    {
-      my_error(ER_COLLATION_CHARSET_MISMATCH, MYF(0), cl->coll_name.str,
-               cs->cs_name.str);
-      return NULL;
-    }
-    return cl;
-  }
-  return cs;
-}
-
 void LEX::mark_first_table_as_inserting()
 {
   TABLE_LIST *t= first_select_lex()->table_list.first;
diff --git a/sql/sql_parse.h b/sql/sql_parse.h
index 9e1ec6fabbc..d3cf83b6e08 100644
--- a/sql/sql_parse.h
+++ b/sql/sql_parse.h
@@ -78,7 +78,6 @@ bool check_string_char_length(const LEX_CSTRING *str, uint err_msg,
                               size_t max_char_length, CHARSET_INFO *cs,
                               bool no_error);
 bool check_ident_length(const LEX_CSTRING *ident);
-CHARSET_INFO* merge_charset_and_collation(CHARSET_INFO *cs, CHARSET_INFO *cl);
 bool check_host_name(LEX_CSTRING *str);
 bool check_identifier_name(LEX_CSTRING *str, uint max_char_length,
                            uint err_code, const char *param_for_err_msg);
diff --git a/sql/sql_partition.cc b/sql/sql_partition.cc
index d2cdbdaf122..f42b9955b90 100644
--- a/sql/sql_partition.cc
+++ b/sql/sql_partition.cc
@@ -5984,11 +5984,12 @@ the generated partition syntax in a correct manner.
           tab_part_info != part_info && part_info->part_type == VERSIONING_PARTITION &&
           part_info->num_parts == 0)
       {
-        if (part_info->vers_info->interval.is_set() &&
+        if (part_info->vers_info->interval.is_set() && (
+            !tab_part_info->vers_info->interval.is_set() ||
             /* TODO: equivalent intervals like 1 hour and 60 mins should be considered equal */
             memcmp(&part_info->vers_info->interval,
                    &tab_part_info->vers_info->interval,
-                   sizeof(Vers_part_info::interval)))
+                   sizeof(Vers_part_info::interval))))
         {
           /* If interval is changed we can not do fast alter */
           tab_part_info= tab_part_info->get_clone(thd);
diff --git a/sql/sql_partition_admin.cc b/sql/sql_partition_admin.cc
index fb1ae0d5fc7..4edd6e7f9a0 100644
--- a/sql/sql_partition_admin.cc
+++ b/sql/sql_partition_admin.cc
@@ -195,7 +195,8 @@ static bool check_exchange_partition(TABLE *table, TABLE *part_table)
 bool compare_table_with_partition(THD *thd, TABLE *table, TABLE *part_table,
                                   partition_element *part_elem, uint part_id)
 {
-  HA_CREATE_INFO table_create_info, part_create_info;
+  HA_CREATE_INFO table_create_info;
+  Table_specification_st part_create_info;
   Alter_info part_alter_info;
   Alter_table_ctx part_alter_ctx; // Not used
   DBUG_ENTER("compare_table_with_partition");
@@ -258,8 +259,13 @@ bool compare_table_with_partition(THD *thd, TABLE *table, TABLE *part_table,
     my_error(ER_TABLES_DIFFERENT_METADATA, MYF(0));
     DBUG_RETURN(TRUE);
   }
-  DBUG_ASSERT(table->s->db_create_options ==
-              part_table->s->db_create_options);
+
+  if (table->s->db_create_options != part_table->s->db_create_options)
+  {
+    my_error(ER_TABLES_DIFFERENT_METADATA, MYF(0));
+    DBUG_RETURN(TRUE);
+  }
+
   DBUG_ASSERT(table->s->db_options_in_use ==
               part_table->s->db_options_in_use);
 
diff --git a/sql/sql_plugin.cc b/sql/sql_plugin.cc
index 2e647712e64..66781caac3e 100644
--- a/sql/sql_plugin.cc
+++ b/sql/sql_plugin.cc
@@ -374,7 +374,8 @@ bool check_valid_path(const char *path, size_t len)
 static void fix_dl_name(MEM_ROOT *root, LEX_CSTRING *dl)
 {
   const size_t so_ext_len= sizeof(SO_EXT) - 1;
-  if (my_strcasecmp(&my_charset_latin1, dl->str + dl->length - so_ext_len,
+  if (dl->length < so_ext_len ||
+      my_strcasecmp(&my_charset_latin1, dl->str + dl->length - so_ext_len,
                     SO_EXT))
   {
     char *s= (char*)alloc_root(root, dl->length + so_ext_len + 1);
diff --git a/sql/sql_select.cc b/sql/sql_select.cc
index 766cb28dfd1..019adceacc5 100644
--- a/sql/sql_select.cc
+++ b/sql/sql_select.cc
@@ -111,12 +111,19 @@ static void optimize_straight_join(JOIN *join, table_map join_tables);
 static bool greedy_search(JOIN *join, table_map remaining_tables,
                           uint depth, uint prune_level,
                           uint use_cond_selectivity);
-static bool best_extension_by_limited_search(JOIN *join,
-                                             table_map remaining_tables,
-                                             uint idx, double record_count,
-                                             double read_time, uint depth,
-                                             uint prune_level,
-                                             uint use_cond_selectivity);
+enum enum_best_search {
+  SEARCH_ABORT= -2,
+  SEARCH_ERROR= -1,
+  SEARCH_OK= 0,
+  SEARCH_FOUND_EDGE=1
+};
+static enum_best_search
+best_extension_by_limited_search(JOIN *join,
+                                 table_map remaining_tables,
+                                 uint idx, double record_count,
+                                 double read_time, uint depth,
+                                 uint prune_level,
+                                 uint use_cond_selectivity);
 static uint determine_search_depth(JOIN* join);
 C_MODE_START
 static int join_tab_cmp(const void *dummy, const void* ptr1, const void* ptr2);
@@ -409,6 +416,7 @@ POSITION::POSITION()
   range_rowid_filter_info= 0;
   ref_depend_map= dups_producing_tables= 0;
   inner_tables_handled_with_other_sjs= 0;
+  type= JT_UNKNOWN;
   dups_weedout_picker.set_empty();
   firstmatch_picker.set_empty();
   loosescan_picker.set_empty();
@@ -8468,6 +8476,7 @@ best_access_path(JOIN      *join,
   pos->records_read= records;
   pos->read_time=    best;
   pos->key=          best_key;
+  pos->type=         best_type;
   pos->table=        s;
   pos->ref_depend_map= best_ref_depends_map;
   pos->loosescan_picker.loosescan_key= MAX_KEY;
@@ -9123,9 +9132,12 @@ greedy_search(JOIN      *join,
   do {
     /* Find the extension of the current QEP with the lowest cost */
     join->best_read= DBL_MAX;
-    if (best_extension_by_limited_search(join, remaining_tables, idx, record_count,
-                                         read_time, search_depth, prune_level,
-                                         use_cond_selectivity))
+    if ((int) best_extension_by_limited_search(join, remaining_tables, idx,
+                                               record_count,
+                                               read_time, search_depth,
+                                               prune_level,
+                                               use_cond_selectivity) <
+        (int) SEARCH_OK)
       DBUG_RETURN(TRUE);
     /*
       'best_read < DBL_MAX' means that optimizer managed to find
@@ -9761,6 +9773,28 @@ exit:
 }
 
 
+/*
+  Check if the table is an EQ_REF or similar table and there is no cost
+  to gain by moveing it to a later stage.
+  We call such a table a edge table (or hanging leaf) as it will read at
+  most one row and will not add to the number of row combinations in the join.
+*/
+
+static inline enum_best_search
+check_if_edge_table(POSITION *pos,
+                    double pushdown_cond_selectivity)
+{
+
+  if ((pos->type == JT_EQ_REF ||
+       (pos->type == JT_REF &&
+        pos->records_read == 1 &&
+        !pos->range_rowid_filter_info)) &&
+      pushdown_cond_selectivity >= 0.999)
+    return SEARCH_FOUND_EDGE;
+  return SEARCH_OK;
+}
+
+
 /**
   Find a good, possibly optimal, query execution plan (QEP) by a possibly
   exhaustive search.
@@ -9875,12 +9909,17 @@ exit:
                           pushed to a table should be taken into account
 
   @retval
-    FALSE       ok
+    enum_best_search::SEARCH_OK          All fine
   @retval
-    TRUE        Fatal error
+    enum_best_search::SEARCH_FOUND_EDGE  All remaning tables are edge tables
+  @retval
+    enum_best_search::SEARCH_ABORT       Killed by user
+  @retval
+    enum_best_search::SEARCH_ERROR       Fatal error
 */
 
-static bool
+
+static enum_best_search
 best_extension_by_limited_search(JOIN      *join,
                                  table_map remaining_tables,
                                  uint      idx,
@@ -9890,9 +9929,17 @@ best_extension_by_limited_search(JOIN      *join,
                                  uint      prune_level,
                                  uint      use_cond_selectivity)
 {
-  DBUG_ENTER("best_extension_by_limited_search");
-
   THD *thd= join->thd;
+  /*
+     'join' is a partial plan with lower cost than the best plan so far,
+     so continue expanding it further with the tables in 'remaining_tables'.
+  */
+  JOIN_TAB *s;
+  double best_record_count= DBL_MAX;
+  double best_read_time=    DBL_MAX;
+  bool disable_jbuf= join->thd->variables.join_cache_level == 0;
+  enum_best_search best_res;
+  DBUG_ENTER("best_extension_by_limited_search");
 
   DBUG_EXECUTE_IF("show_explain_probe_best_ext_lim_search", 
                   if (dbug_user_var_equals_int(thd, 
@@ -9902,19 +9949,7 @@ best_extension_by_limited_search(JOIN      *join,
                  );
 
   if (unlikely(thd->check_killed()))  // Abort
-    DBUG_RETURN(TRUE);
-
-  DBUG_EXECUTE("opt", print_plan(join, idx, read_time, record_count, idx,
-                                 "SOFAR:"););
-
-  /* 
-     'join' is a partial plan with lower cost than the best plan so far,
-     so continue expanding it further with the tables in 'remaining_tables'.
-  */
-  JOIN_TAB *s;
-  double best_record_count= DBL_MAX;
-  double best_read_time=    DBL_MAX;
-  bool disable_jbuf= join->thd->variables.join_cache_level == 0;
+    DBUG_RETURN(SEARCH_ABORT);
 
   DBUG_EXECUTE("opt", print_plan(join, idx, record_count, read_time, read_time,
                                 "part_plan"););
@@ -9930,15 +9965,18 @@ best_extension_by_limited_search(JOIN      *join,
   for (JOIN_TAB **pos= join->best_ref + idx ; (s= *pos) ; pos++)
   {
     table_map real_table_bit= s->table->map;
-    if ((remaining_tables & real_table_bit) &&
-        (allowed_tables & real_table_bit) &&
+    DBUG_ASSERT(remaining_tables & real_table_bit);
+
+    if ((allowed_tables & real_table_bit) &&
         !(remaining_tables & s->dependent) &&
-        (!idx || !check_interleaving_with_nj(s)))
+        !check_interleaving_with_nj(s))
     {
       double current_record_count, current_read_time;
+      double partial_join_cardinality;
       POSITION *position= join->positions + idx;
-
+      POSITION loose_scan_pos;
       Json_writer_object trace_one_table(thd);
+
       if (unlikely(thd->trace_started()))
       {
         trace_plan_prefix(join, idx, remaining_tables);
@@ -9946,7 +9984,6 @@ best_extension_by_limited_search(JOIN      *join,
       }
 
       /* Find the best access method from 's' to the current partial plan */
-      POSITION loose_scan_pos;
       best_access_path(join, s, remaining_tables, join->positions, idx,
                        disable_jbuf, record_count, position, &loose_scan_pos);
 
@@ -10021,32 +10058,51 @@ best_extension_by_limited_search(JOIN      *join,
       double pushdown_cond_selectivity= 1.0;
       if (use_cond_selectivity > 1)
         pushdown_cond_selectivity= table_cond_selectivity(join, idx, s,
-				                          remaining_tables &
+                                                          remaining_tables &
                                                           ~real_table_bit);
       join->positions[idx].cond_selectivity= pushdown_cond_selectivity;
 
-      if (unlikely(thd->trace_started()) && pushdown_cond_selectivity < 1.0)
-        trace_one_table.add("selectivity", pushdown_cond_selectivity);
+      partial_join_cardinality= (current_record_count *
+                                 pushdown_cond_selectivity);
 
-      double partial_join_cardinality= current_record_count *
-                                        pushdown_cond_selectivity;
-      if ( (search_depth > 1) && (remaining_tables & ~real_table_bit) & allowed_tables )
-      { /* Recursively expand the current partial plan */
+      if (unlikely(thd->trace_started()))
+      {
+        if (pushdown_cond_selectivity < 1.0)
+        {
+          trace_one_table.add("selectivity", pushdown_cond_selectivity);
+          trace_one_table.add("estimated_join_cardinality",
+                              partial_join_cardinality);
+        }
+      }
+
+      if ((search_depth > 1) && (remaining_tables & ~real_table_bit) &
+          allowed_tables)
+      {
+        /* Recursively expand the current partial plan */
         swap_variables(JOIN_TAB*, join->best_ref[idx], *pos);
         Json_writer_array trace_rest(thd, "rest_of_plan");
-        if (best_extension_by_limited_search(join,
-                                             remaining_tables & ~real_table_bit,
-                                             idx + 1,
-                                             partial_join_cardinality,
-                                             current_read_time,
-                                             search_depth - 1,
-                                             prune_level,
-                                             use_cond_selectivity))
-          DBUG_RETURN(TRUE);
+        best_res=
+          best_extension_by_limited_search(join,
+                                           remaining_tables &
+                                           ~real_table_bit,
+                                           idx + 1,
+                                           partial_join_cardinality,
+                                           current_read_time,
+                                           search_depth - 1,
+                                           prune_level,
+                                           use_cond_selectivity);
+        if ((int) best_res < (int) SEARCH_OK)
+          DBUG_RETURN(best_res);                // Abort
         swap_variables(JOIN_TAB*, join->best_ref[idx], *pos);
+        if (best_res == SEARCH_FOUND_EDGE &&
+            check_if_edge_table(join->positions+ idx,
+                                pushdown_cond_selectivity) !=
+            SEARCH_FOUND_EDGE)
+          best_res= SEARCH_OK;
       }
       else
-      { /*
+      {
+        /*
           'join' is either the best partial QEP with 'search_depth' relations,
           or the best complete QEP so far, whichever is smaller.
         */
@@ -10055,15 +10111,13 @@ best_extension_by_limited_search(JOIN      *join,
             join->positions[join->const_tables].table->table)
         {
           /*
-             We may have to make a temp table, note that this is only a
-             heuristic since we cannot know for sure at this point.
-             Hence it may be wrong.
+            We may have to make a temp table, note that this is only a
+            heuristic since we cannot know for sure at this point.
+            Hence it may be wrong.
           */
           trace_one_table.add("cost_for_sorting", current_record_count);
           current_read_time= COST_ADD(current_read_time, current_record_count);
         }
-        trace_one_table.add("estimated_join_cardinality",
-                            partial_join_cardinality);
         if (current_read_time < join->best_read)
         {
           memcpy((uchar*) join->best_positions, (uchar*) join->positions,
@@ -10076,12 +10130,19 @@ best_extension_by_limited_search(JOIN      *join,
                                        read_time,
                                        current_read_time,
                                        "full_plan"););
+        best_res= check_if_edge_table(join->positions + idx,
+                                      pushdown_cond_selectivity);
       }
       restore_prev_nj_state(s);
       restore_prev_sj_state(remaining_tables, s, idx);
+      if (best_res == SEARCH_FOUND_EDGE)
+      {
+        trace_one_table.add("pruned_by_hanging_leaf", true);
+        DBUG_RETURN(best_res);
+      }
     }
   }
-  DBUG_RETURN(FALSE);
+  DBUG_RETURN(SEARCH_OK);
 }
 
 
@@ -17233,7 +17294,6 @@ static uint reset_nj_counters(JOIN *join, List<TABLE_LIST> *join_list)
 
 static bool check_interleaving_with_nj(JOIN_TAB *next_tab)
 {
-  TABLE_LIST *next_emb= next_tab->table->pos_in_table_list->embedding;
   JOIN *join= next_tab->join;
 
   if (join->cur_embedding_map & ~next_tab->embedding_map)
@@ -17245,6 +17305,7 @@ static bool check_interleaving_with_nj(JOIN_TAB *next_tab)
     return TRUE;
   }
    
+  TABLE_LIST *next_emb= next_tab->table->pos_in_table_list->embedding;
   /*
     Do update counters for "pairs of brackets" that we've left (marked as
     X,Y,Z in the above picture)
@@ -18430,14 +18491,17 @@ Field *Item_default_value::create_tmp_field_ex(MEM_ROOT *root, TABLE *table,
                                                Tmp_field_src *src,
                                                const Tmp_field_param *param)
 {
-  if (field->default_value && (field->flags & BLOB_FLAG))
+  if (field->default_value || (field->flags & BLOB_FLAG))
   {
     /*
       We have to use a copy function when using a blob with default value
       as the we have to calculate the default value before we can use it.
     */
      get_tmp_field_src(src, param);
-     return tmp_table_field_from_field_type(root, table);
+     Field *result= tmp_table_field_from_field_type(root, table);
+     if (result && param->modify_item())
+       result_field= result;
+     return result;
   }
   /*
     Same code as in Item_field::create_tmp_field_ex, except no default field
@@ -18874,6 +18938,7 @@ TABLE *Create_tmp_table::start(THD *thd,
   table->copy_blobs= 1;
   table->in_use= thd;
   table->no_rows_with_nulls= param->force_not_null_cols;
+  table->expr_arena= thd;
 
   table->s= share;
   init_tmp_table_share(thd, share, "", 0, "(temporary)", tmpname);
@@ -25087,8 +25152,8 @@ int setup_order(THD *thd, Ref_ptr_array ref_pointer_array, TABLE_LIST *tables,
     if (find_order_in_list(thd, ref_pointer_array, tables, order, fields,
                            all_fields, false, true, from_window_spec))
       return 1;
-    if ((*order->item)->with_window_func() &&
-        context_analysis_place != IN_ORDER_BY)
+    Item * const item= *order->item;
+    if (item->with_window_func() && context_analysis_place != IN_ORDER_BY)
     {
       my_error(ER_WINDOW_FUNCTION_IN_WINDOW_SPEC, MYF(0));
       return 1;
@@ -25099,20 +25164,18 @@ int setup_order(THD *thd, Ref_ptr_array ref_pointer_array, TABLE_LIST *tables,
       an ORDER BY clause
     */
 
-    if (for_union &&
-        ((*order->item)->with_sum_func() ||
-         (*order->item)->with_window_func()))
+    if (for_union && (item->with_sum_func() || item->with_window_func()))
     {
       my_error(ER_AGGREGATE_ORDER_FOR_UNION, MYF(0), number);
       return 1;
     }
 
-    if (!(*order->item)->with_sum_func())
-      continue;
-
-    if (from_window_spec && (*order->item)->type() != Item::SUM_FUNC_ITEM)
-      (*order->item)->split_sum_func(thd, ref_pointer_array,
-                                     all_fields, SPLIT_SUM_SELECT);
+    if ((from_window_spec && item->with_sum_func() &&
+         item->type() != Item::SUM_FUNC_ITEM) || item->with_window_func())
+    {
+      item->split_sum_func(thd, ref_pointer_array,
+                           all_fields, SPLIT_SUM_SELECT);
+    }
   }
   return 0;
 }
@@ -26062,15 +26125,17 @@ change_to_use_tmp_fields(THD *thd, Ref_ptr_array ref_pointer_array,
   for (uint i= 0; (item= it++); i++)
   {
     Field *field;
-    if ((item->with_sum_func() && item->type() != Item::SUM_FUNC_ITEM) ||
-        item->with_window_func())
+    enum Item::Type item_type= item->type();
+    if ((item->with_sum_func() && item_type != Item::SUM_FUNC_ITEM) ||
+       item->with_window_func())
       item_field= item;
-    else if (item->type() == Item::FIELD_ITEM)
+    else if (item_type == Item::FIELD_ITEM ||
+             item_type == Item::DEFAULT_VALUE_ITEM)
     {
       if (!(item_field= item->get_tmp_table_item(thd)))
         DBUG_RETURN(true);
     }
-    else if (item->type() == Item::FUNC_ITEM &&
+    else if (item_type == Item::FUNC_ITEM &&
              ((Item_func*)item)->functype() == Item_func::SUSERVAR_FUNC)
     {
       field= item->get_tmp_table_field();
@@ -27929,7 +27994,7 @@ static void print_table_array(THD *thd,
      too)
 */
 
-static bool is_eliminated_table(table_map eliminated_tables, TABLE_LIST *tbl)
+bool is_eliminated_table(table_map eliminated_tables, TABLE_LIST *tbl)
 {
   return eliminated_tables &&
     ((tbl->table && (tbl->table->map & eliminated_tables)) ||
diff --git a/sql/sql_select.h b/sql/sql_select.h
index e6f63b465f6..34d689cf698 100644
--- a/sql/sql_select.h
+++ b/sql/sql_select.h
@@ -2,7 +2,7 @@
 #define SQL_SELECT_INCLUDED
 
 /* Copyright (c) 2000, 2013, Oracle and/or its affiliates.
-   Copyright (c) 2008, 2020, MariaDB Corporation.
+   Copyright (c) 2008, 2022, MariaDB Corporation.
 
    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
@@ -989,6 +989,8 @@ public:
   */
   enum sj_strategy_enum sj_strategy;
 
+  /* Type of join (EQ_REF, REF etc) */
+  enum join_type type;
   /*
     Valid only after fix_semijoin_strategies_for_picked_join_order() call:
     if sj_strategy!=SJ_OPT_NONE, this is the number of subsequent tables that
@@ -2469,6 +2471,8 @@ int create_sort_index(THD *thd, JOIN *join, JOIN_TAB *tab, Filesort *fsort);
 JOIN_TAB *first_explain_order_tab(JOIN* join);
 JOIN_TAB *next_explain_order_tab(JOIN* join, JOIN_TAB* tab);
 
+bool is_eliminated_table(table_map eliminated_tables, TABLE_LIST *tbl);
+
 bool check_simple_equality(THD *thd, const Item::Context &ctx,
                            Item *left_item, Item *right_item,
                            COND_EQUAL *cond_equal);
diff --git a/sql/sql_show.cc b/sql/sql_show.cc
index 2134bb04f6e..04bec87ca13 100644
--- a/sql/sql_show.cc
+++ b/sql/sql_show.cc
@@ -5193,6 +5193,7 @@ public:
 
 int get_all_tables(THD *thd, TABLE_LIST *tables, COND *cond)
 {
+  DBUG_ENTER("get_all_tables");
   LEX *lex= thd->lex;
   TABLE *table= tables->table;
   TABLE_LIST table_acl_check;
@@ -5210,7 +5211,29 @@ int get_all_tables(THD *thd, TABLE_LIST *tables, COND *cond)
   uint table_open_method= tables->table_open_method;
   bool can_deadlock;
   MEM_ROOT tmp_mem_root;
-  DBUG_ENTER("get_all_tables");
+  /*
+    We're going to open FRM files for tables.
+    In case of VIEWs that contain stored function calls,
+    these stored functions will be parsed and put to the SP cache.
+
+    Suppose we have a view containing a stored function call:
+      CREATE VIEW v1 AS SELECT f1() AS c1;
+    and now we're running:
+      SELECT * FROM INFORMATION_SCHEMA.TABLES WHERE TABLE_NAME=f1();
+    If a parallel thread invalidates the cache,
+    e.g. by creating or dropping some stored routine,
+    the SELECT query will re-parse f1() when processing "v1"
+    and replace the outdated cached version of f1() to a new one.
+    But the old version of f1() is referenced from the m_sp member
+    of the Item_func_sp instances used in the WHERE condition.
+    We cannot destroy it. To avoid such clashes, let's remember
+    all old routines into a temporary SP cache collection
+    and process tables with a new empty temporary SP cache collection.
+    Then restore to the old SP cache collection at the end.
+  */
+  Sp_caches old_sp_caches;
+
+  old_sp_caches.sp_caches_swap(*thd);
 
   bzero(&tmp_mem_root, sizeof(tmp_mem_root));
 
@@ -5386,6 +5409,13 @@ err:
   thd->restore_backup_open_tables_state(&open_tables_state_backup);
   free_root(&tmp_mem_root, 0);
 
+  /*
+    Now restore to the saved SP cache collection
+    and clear the temporary SP cache collection.
+  */
+  old_sp_caches.sp_caches_swap(*thd);
+  old_sp_caches.sp_caches_clear();
+
   DBUG_RETURN(error);
 }
 
@@ -9093,7 +9123,7 @@ ST_FIELD_INFO columns_fields_info[]=
   Column("ORDINAL_POSITION",        ULonglong(), NOT_NULL,          OPEN_FRM_ONLY),
   Column("COLUMN_DEFAULT", Longtext(MAX_FIELD_VARCHARLENGTH),
                                                  NULLABLE, "Default",OPEN_FRM_ONLY),
-  Column("IS_NULLABLE",             Yesno(),     NOT_NULL, "Null",  OPEN_FRM_ONLY),
+  Column("IS_NULLABLE",          Yes_or_empty(), NOT_NULL, "Null",  OPEN_FRM_ONLY),
   Column("DATA_TYPE",               Name(),      NOT_NULL,          OPEN_FRM_ONLY),
   Column("CHARACTER_MAXIMUM_LENGTH",ULonglong(), NULLABLE,          OPEN_FRM_ONLY),
   Column("CHARACTER_OCTET_LENGTH",  ULonglong(), NULLABLE,          OPEN_FRM_ONLY),
@@ -9130,8 +9160,8 @@ ST_FIELD_INFO collation_fields_info[]=
   Column("COLLATION_NAME",               CSName(),     NOT_NULL, "Collation"),
   Column("CHARACTER_SET_NAME",           CSName(),     NOT_NULL, "Charset"),
   Column("ID", SLonglong(MY_INT32_NUM_DECIMAL_DIGITS), NOT_NULL, "Id"),
-  Column("IS_DEFAULT",                   Yesno(),      NOT_NULL, "Default"),
-  Column("IS_COMPILED",                  Yesno(),      NOT_NULL, "Compiled"),
+  Column("IS_DEFAULT",                 Yes_or_empty(), NOT_NULL, "Default"),
+  Column("IS_COMPILED",                Yes_or_empty(), NOT_NULL, "Compiled"),
   Column("SORTLEN",                      SLonglong(3), NOT_NULL, "Sortlen"),
   CEnd()
 };
@@ -9139,10 +9169,10 @@ ST_FIELD_INFO collation_fields_info[]=
 
 ST_FIELD_INFO applicable_roles_fields_info[]=
 {
-  Column("GRANTEE",                  Userhost(), NOT_NULL),
+  Column("GRANTEE",                  Userhost(),     NOT_NULL),
   Column("ROLE_NAME", Varchar(USERNAME_CHAR_LENGTH), NOT_NULL),
-  Column("IS_GRANTABLE",             Yesno(),    NOT_NULL),
-  Column("IS_DEFAULT",               Yesno(),    NULLABLE),
+  Column("IS_GRANTABLE",             Yes_or_empty(), NOT_NULL),
+  Column("IS_DEFAULT",               Yes_or_empty(), NULLABLE),
   CEnd()
 };
 
@@ -9286,7 +9316,7 @@ ST_FIELD_INFO view_fields_info[]=
   Column("TABLE_NAME",           Name(),     NOT_NULL, OPEN_FRM_ONLY),
   Column("VIEW_DEFINITION", Longtext(65535), NOT_NULL, OPEN_FRM_ONLY),
   Column("CHECK_OPTION",         Varchar(8), NOT_NULL, OPEN_FRM_ONLY),
-  Column("IS_UPDATABLE",         Yesno(),    NOT_NULL, OPEN_FULL_TABLE),
+  Column("IS_UPDATABLE",     Yes_or_empty(), NOT_NULL, OPEN_FULL_TABLE),
   Column("DEFINER",              Definer(),  NOT_NULL, OPEN_FRM_ONLY),
   Column("SECURITY_TYPE",        Varchar(7), NOT_NULL, OPEN_FRM_ONLY),
   Column("CHARACTER_SET_CLIENT", CSName(),   NOT_NULL, OPEN_FRM_ONLY),
@@ -9298,46 +9328,46 @@ ST_FIELD_INFO view_fields_info[]=
 
 ST_FIELD_INFO user_privileges_fields_info[]=
 {
-  Column("GRANTEE",        Userhost(), NOT_NULL),
-  Column("TABLE_CATALOG",  Catalog(),  NOT_NULL),
-  Column("PRIVILEGE_TYPE", Name(),     NOT_NULL),
-  Column("IS_GRANTABLE",   Yesno(),    NOT_NULL),
+  Column("GRANTEE",        Userhost(),     NOT_NULL),
+  Column("TABLE_CATALOG",  Catalog(),      NOT_NULL),
+  Column("PRIVILEGE_TYPE", Name(),         NOT_NULL),
+  Column("IS_GRANTABLE",   Yes_or_empty(), NOT_NULL),
   CEnd()
 };
 
 
 ST_FIELD_INFO schema_privileges_fields_info[]=
 {
-  Column("GRANTEE",        Userhost(), NOT_NULL),
-  Column("TABLE_CATALOG",  Catalog(),  NOT_NULL),
-  Column("TABLE_SCHEMA",   Name(),     NOT_NULL),
-  Column("PRIVILEGE_TYPE", Name(),     NOT_NULL),
-  Column("IS_GRANTABLE",   Yesno(),    NOT_NULL),
+  Column("GRANTEE",        Userhost(),     NOT_NULL),
+  Column("TABLE_CATALOG",  Catalog(),      NOT_NULL),
+  Column("TABLE_SCHEMA",   Name(),         NOT_NULL),
+  Column("PRIVILEGE_TYPE", Name(),         NOT_NULL),
+  Column("IS_GRANTABLE",   Yes_or_empty(), NOT_NULL),
   CEnd()
 };
 
 
 ST_FIELD_INFO table_privileges_fields_info[]=
 {
-  Column("GRANTEE",        Userhost(), NOT_NULL),
-  Column("TABLE_CATALOG",  Catalog(),  NOT_NULL),
-  Column("TABLE_SCHEMA",   Name(),     NOT_NULL),
-  Column("TABLE_NAME",     Name(),     NOT_NULL),
-  Column("PRIVILEGE_TYPE", Name(),     NOT_NULL),
-  Column("IS_GRANTABLE",   Yesno(),    NOT_NULL),
+  Column("GRANTEE",        Userhost(),     NOT_NULL),
+  Column("TABLE_CATALOG",  Catalog(),      NOT_NULL),
+  Column("TABLE_SCHEMA",   Name(),         NOT_NULL),
+  Column("TABLE_NAME",     Name(),         NOT_NULL),
+  Column("PRIVILEGE_TYPE", Name(),         NOT_NULL),
+  Column("IS_GRANTABLE",   Yes_or_empty(), NOT_NULL),
   CEnd()
 };
 
 
 ST_FIELD_INFO column_privileges_fields_info[]=
 {
-  Column("GRANTEE",        Userhost(), NOT_NULL),
-  Column("TABLE_CATALOG",  Catalog(),  NOT_NULL),
-  Column("TABLE_SCHEMA",   Name(),     NOT_NULL),
-  Column("TABLE_NAME",     Name(),     NOT_NULL),
-  Column("COLUMN_NAME",    Name(),     NOT_NULL),
-  Column("PRIVILEGE_TYPE", Name(),     NOT_NULL),
-  Column("IS_GRANTABLE",   Yesno(),    NOT_NULL),
+  Column("GRANTEE",        Userhost(),     NOT_NULL),
+  Column("TABLE_CATALOG",  Catalog(),      NOT_NULL),
+  Column("TABLE_SCHEMA",   Name(),         NOT_NULL),
+  Column("TABLE_NAME",     Name(),         NOT_NULL),
+  Column("COLUMN_NAME",    Name(),         NOT_NULL),
+  Column("PRIVILEGE_TYPE", Name(),         NOT_NULL),
+  Column("IS_GRANTABLE",   Yes_or_empty(), NOT_NULL),
   CEnd()
 };
 
@@ -9478,7 +9508,7 @@ ST_FIELD_INFO sysvars_fields_info[]=
   Column("NUMERIC_MAX_VALUE",    Varchar(MY_INT64_NUM_DECIMAL_DIGITS), NULLABLE),
   Column("NUMERIC_BLOCK_SIZE",   Varchar(MY_INT64_NUM_DECIMAL_DIGITS), NULLABLE),
   Column("ENUM_VALUE_LIST",      Longtext(65535),                  NULLABLE),
-  Column("READ_ONLY",            Yesno(),                          NOT_NULL),
+  Column("READ_ONLY",            Yes_or_empty(),                   NOT_NULL),
   Column("COMMAND_LINE_ARGUMENT",Name(),                           NULLABLE),
   Column("GLOBAL_VALUE_PATH",    Varchar(2048),                    NULLABLE),
   CEnd()
diff --git a/sql/sql_table.cc b/sql/sql_table.cc
index 43fa10fab3c..449478a7fa4 100644
--- a/sql/sql_table.cc
+++ b/sql/sql_table.cc
@@ -2202,7 +2202,7 @@ bool check_duplicates_in_interval(const char *set_or_name,
 bool Column_definition::
        prepare_charset_for_string(const Column_derived_attributes *dattr)
 {
-  CHARSET_INFO *tmp= lex_charset_collation().
+  CHARSET_INFO *tmp= charset_collation_attrs().
                        resolved_to_character_set(dattr->charset());
   if (!tmp)
     return true;
@@ -3875,38 +3875,6 @@ bool validate_comment_length(THD *thd, LEX_CSTRING *comment, size_t max_len,
 
 
 /*
-  Set table default charset, if not set
-
-  SYNOPSIS
-    set_table_default_charset()
-    create_info        Table create information
-
-  DESCRIPTION
-    If the table character set was not given explicitly,
-    let's fetch the database default character set and
-    apply it to the table.
-*/
-
-static void set_table_default_charset(THD *thd, HA_CREATE_INFO *create_info,
-                                      const LEX_CSTRING &db)
-{
-  /*
-    If the table character set was not given explicitly,
-    let's fetch the database default character set and
-    apply it to the table.
-  */
-  if (!create_info->default_table_charset)
-  {
-    Schema_specification_st db_info;
-
-    load_db_opt_by_name(thd, db.str, &db_info);
-
-    create_info->default_table_charset= db_info.default_table_charset;
-  }
-}
-
-
-/*
   Extend long VARCHAR fields to blob & prepare field if it's a blob
 
   SYNOPSIS
@@ -4064,14 +4032,14 @@ handler *mysql_create_frm_image(THD *thd, const LEX_CSTRING &db,
   handler       *file;
   DBUG_ENTER("mysql_create_frm_image");
 
+  DBUG_ASSERT(create_info->default_table_charset);
+
   if (!alter_info->create_list.elements)
   {
     my_error(ER_TABLE_MUST_HAVE_COLUMNS, MYF(0));
     DBUG_RETURN(NULL);
   }
 
-  set_table_default_charset(thd, create_info, db);
-
   db_options= create_info->table_options_with_row_type();
 
   if (unlikely(!(file= get_new_handler((TABLE_SHARE*) 0, thd->mem_root,
@@ -4345,6 +4313,7 @@ err:
   @retval -1 table existed but IF NOT EXISTS was used
 */
 
+static
 int create_table_impl(THD *thd,
                       DDL_LOG_STATE *ddl_log_state_create,
                       DDL_LOG_STATE *ddl_log_state_rm,
@@ -4365,6 +4334,8 @@ int create_table_impl(THD *thd,
   DBUG_PRINT("enter", ("db: '%s'  table: '%s'  tmp: %d  path: %s",
                        db.str, table_name.str, internal_tmp_table, path.str));
 
+  DBUG_ASSERT(create_info->default_table_charset);
+
   /* Easy check for ddl logging if we are creating a temporary table */
   if (create_info->tmp_table())
   {
@@ -4703,6 +4674,8 @@ int mysql_create_table_no_lock(THD *thd,
   LEX_CSTRING cpath;
   LEX_CUSTRING frm= {0,0};
 
+  DBUG_ASSERT(create_info->default_table_charset);
+
   if (create_info->tmp_table())
     path_length= build_tmptable_filename(thd, path, sizeof(path));
   else
@@ -4771,6 +4744,8 @@ int mysql_create_table_no_lock(THD *thd,
   close of thread tables.
 */
 
+
+static
 bool mysql_create_table(THD *thd, TABLE_LIST *create_table,
                         Table_specification_st *create_info,
                         Alter_info *alter_info)
@@ -4784,6 +4759,8 @@ bool mysql_create_table(THD *thd, TABLE_LIST *create_table,
   bool result;
   DBUG_ENTER("mysql_create_table");
 
+  DBUG_ASSERT(create_info->default_table_charset);
+
   DBUG_ASSERT(create_table == thd->lex->query_tables);
 
   bzero(&ddl_log_state_create, sizeof(ddl_log_state_create));
@@ -5222,6 +5199,7 @@ mysql_rename_table(handlerton *base, const LEX_CSTRING *old_db,
     TRUE  error
 */
 
+static
 bool mysql_create_like_table(THD* thd, TABLE_LIST* table,
                              TABLE_LIST* src_table,
                              Table_specification_st *create_info)
@@ -5296,6 +5274,14 @@ bool mysql_create_like_table(THD* thd, TABLE_LIST* table,
   local_create_info.db_type= src_table->table->s->db_type();
   local_create_info.row_type= src_table->table->s->row_type;
   local_create_info.alter_info= &local_alter_info;
+  /*
+    This statement:
+      CREATE TABLE t1 LIKE t2
+    does not support table charset/collation clauses.
+    No needs to copy. Assert they are empty.
+  */
+  DBUG_ASSERT(create_info->default_charset_collation.is_empty());
+  DBUG_ASSERT(create_info->convert_charset_collation.is_empty());
   if (mysql_prepare_alter_table(thd, src_table->table, &local_create_info,
                                 &local_alter_info, &local_alter_ctx))
     goto err;
@@ -7949,12 +7935,14 @@ void append_drop_column(THD *thd, String *str, Field *field)
 
 bool
 mysql_prepare_alter_table(THD *thd, TABLE *table,
-                          HA_CREATE_INFO *create_info,
+                          Table_specification_st *create_info,
                           Alter_info *alter_info,
                           Alter_table_ctx *alter_ctx)
 {
   /* New column definitions are added here */
   List<Create_field> new_create_list;
+  /* System-invisible fields must be added last */
+  List<Create_field> new_create_tail;
   /* New key definitions are added here */
   List<Key> new_key_list;
   List<Alter_rename_key> rename_key_list(alter_info->alter_rename_key_list);
@@ -8013,8 +8001,11 @@ mysql_prepare_alter_table(THD *thd, TABLE *table,
     create_info->max_rows= table->s->max_rows;
   if (!(used_fields & HA_CREATE_USED_AVG_ROW_LENGTH))
     create_info->avg_row_length= table->s->avg_row_length;
-  if (!(used_fields & HA_CREATE_USED_DEFAULT_CHARSET))
-    create_info->default_table_charset= table->s->table_charset;
+
+  if (create_info->resolve_to_charset_collation_context(thd,
+        thd->charset_collation_context_alter_table(table->s)))
+    DBUG_RETURN(true);
+
   if (!(used_fields & HA_CREATE_USED_AUTO) && table->found_next_number_field)
   {
     /* Table has an autoincrement, copy value to new table */
@@ -8181,7 +8172,7 @@ mysql_prepare_alter_table(THD *thd, TABLE *table,
       dropped_sys_vers_fields|= field->flags;
       drop_it.remove();
     }
-    else
+    else if (field->invisible < INVISIBLE_SYSTEM)
     {
       /*
         This field was not dropped and not changed, add it to the list
@@ -8232,6 +8223,12 @@ mysql_prepare_alter_table(THD *thd, TABLE *table,
 	alter_it.remove();
       }
     }
+    else
+    {
+      DBUG_ASSERT(field->invisible == INVISIBLE_SYSTEM);
+      def= new (thd->mem_root) Create_field(thd, field, field);
+      new_create_tail.push_back(def, thd->mem_root);
+    }
   }
 
   /*
@@ -8394,6 +8391,9 @@ mysql_prepare_alter_table(THD *thd, TABLE *table,
       alter_it.remove();
     }
   }
+
+  new_create_list.append(&new_create_tail);
+
   if (unlikely(alter_info->alter_list.elements))
   {
     my_error(ER_BAD_FIELD_ERROR, MYF(0),
@@ -9763,7 +9763,7 @@ static uint64 get_start_alter_id(THD *thd)
 
 bool mysql_alter_table(THD *thd, const LEX_CSTRING *new_db,
                        const LEX_CSTRING *new_name,
-                       HA_CREATE_INFO *create_info,
+                       Table_specification_st *create_info,
                        TABLE_LIST *table_list,
                        Alter_info *alter_info,
                        uint order_num, ORDER *order, bool ignore,
@@ -10356,7 +10356,7 @@ do_continue:;
     DBUG_RETURN(true);
   }
 
-  set_table_default_charset(thd, create_info, alter_ctx.db);
+  DBUG_ASSERT(create_info->default_table_charset);
 
   if (create_info->check_fields(thd, alter_info,
                                 table_list->table_name, table_list->db) ||
@@ -11776,7 +11776,7 @@ copy_data_between_tables(THD *thd, TABLE *from, TABLE *to,
 
 bool mysql_recreate_table(THD *thd, TABLE_LIST *table_list, bool table_copy)
 {
-  HA_CREATE_INFO create_info;
+  Table_specification_st create_info;
   Alter_info alter_info;
   TABLE_LIST *next_table= table_list->next_global;
   DBUG_ENTER("mysql_recreate_table");
@@ -11788,9 +11788,8 @@ bool mysql_recreate_table(THD *thd, TABLE_LIST *table_list, bool table_copy)
   /* hide following tables from open_tables() */
   table_list->next_global= NULL;
 
-  bzero((char*) &create_info, sizeof(create_info));
+  create_info.init();
   create_info.row_type=ROW_TYPE_NOT_USED;
-  create_info.default_table_charset=default_charset_info;
   create_info.alter_info= &alter_info;
   /* Force alter table to recreate table */
   alter_info.flags= (ALTER_CHANGE_COLUMN | ALTER_RECREATE);
@@ -12047,6 +12046,11 @@ bool Sql_cmd_create_table_like::execute(THD *thd)
 
   const bool used_engine= lex->create_info.used_fields & HA_CREATE_USED_ENGINE;
   DBUG_ASSERT((m_storage_engine_name.str != NULL) == used_engine);
+
+  if (lex->create_info.resolve_to_charset_collation_context(thd,
+        thd->charset_collation_context_create_table_in_db(first_table->db.str)))
+    DBUG_RETURN(true);
+
   if (used_engine)
   {
     if (resolve_storage_engine_with_error(thd, &lex->create_info.db_type,
@@ -12119,19 +12123,9 @@ bool Sql_cmd_create_table_like::execute(THD *thd)
   */
   if (!(create_info.used_fields & HA_CREATE_USED_ENGINE))
     create_info.use_default_db_type(thd);
-  /*
-    If we are using SET CHARSET without DEFAULT, add an implicit
-    DEFAULT to not confuse old users. (This may change).
-  */
-  if ((create_info.used_fields &
-       (HA_CREATE_USED_DEFAULT_CHARSET | HA_CREATE_USED_CHARSET)) ==
-      HA_CREATE_USED_CHARSET)
-  {
-    create_info.used_fields&= ~HA_CREATE_USED_CHARSET;
-    create_info.used_fields|= HA_CREATE_USED_DEFAULT_CHARSET;
-    create_info.default_table_charset= create_info.alter_table_convert_to_charset;
-    create_info.alter_table_convert_to_charset= 0;
-  }
+
+  DBUG_ASSERT(!(create_info.used_fields & HA_CREATE_USED_CHARSET));
+  DBUG_ASSERT(create_info.convert_charset_collation.is_empty());
 
   /*
     If we are a slave, we should add OR REPLACE if we don't have
@@ -12357,3 +12351,46 @@ bool Sql_cmd_create_table_like::execute(THD *thd)
 end_with_restore_list:
   DBUG_RETURN(res);
 }
+
+
+bool HA_CREATE_INFO::
+       resolve_to_charset_collation_context(THD *thd,
+             const Lex_table_charset_collation_attrs_st &default_cscl_arg,
+             const Lex_table_charset_collation_attrs_st &convert_cscl,
+             const Charset_collation_context &ctx)
+{
+  /*
+    If CONVERT TO clauses are specified only (without table default clauses),
+    then we copy CONVERT TO clauses to default clauses, so e.g:
+      CONVERT TO CHARACTER SET utf8mb4
+    means
+      CONVERT TO CHARACTER SET utf8mb4, DEFAULT CHARACTER SET utf8mb4
+  */
+  Lex_table_charset_collation_attrs_st default_cscl=
+    !convert_cscl.is_empty() && default_cscl_arg.is_empty() ?
+    convert_cscl : default_cscl_arg;
+
+  if (default_cscl.is_empty())
+    default_table_charset= ctx.collate_default().charset_info();
+  else
+  {
+    // Make sure we don't do double resolution in direct SQL execution
+    DBUG_ASSERT(!default_table_charset || thd->stmt_arena->is_stmt_execute());
+    if (!(default_table_charset=
+            default_cscl.resolved_to_context(ctx)))
+      return true;
+  }
+
+  if (convert_cscl.is_empty())
+    alter_table_convert_to_charset= NULL;
+  else
+  {
+    // Make sure we don't do double resolution in direct SQL execution
+    DBUG_ASSERT(!alter_table_convert_to_charset ||
+                thd->stmt_arena->is_stmt_execute());
+    if (!(alter_table_convert_to_charset=
+            convert_cscl.resolved_to_context(ctx)))
+      return true;
+  }
+  return false;
+}
diff --git a/sql/sql_table.h b/sql/sql_table.h
index e3cfec9ca93..c9e4d969482 100644
--- a/sql/sql_table.h
+++ b/sql/sql_table.h
@@ -19,6 +19,7 @@
 
 #include <my_sys.h>                             // pthread_mutex_t
 #include "m_string.h"                           // LEX_CUSTRING
+#include "lex_charset.h"
 
 #define ERROR_INJECT(code) \
   ((DBUG_IF("crash_" code) && (DBUG_SUICIDE(), 0)) || \
@@ -90,9 +91,6 @@ void build_lower_case_table_filename(char *buff, size_t bufflen,
                                      const LEX_CSTRING *table,
                                      uint flags);
 uint build_tmptable_filename(THD* thd, char *buff, size_t bufflen);
-bool mysql_create_table(THD *thd, TABLE_LIST *create_table,
-                        Table_specification_st *create_info,
-                        Alter_info *alter_info);
 bool add_keyword_to_query(THD *thd, String *result, const LEX_CSTRING *keyword,
                           const LEX_CSTRING *add);
 
@@ -156,14 +154,14 @@ int mysql_discard_or_import_tablespace(THD *thd,
                                        bool discard);
 
 bool mysql_prepare_alter_table(THD *thd, TABLE *table,
-                               HA_CREATE_INFO *create_info,
+                               Table_specification_st *create_info,
                                Alter_info *alter_info,
                                Alter_table_ctx *alter_ctx);
 bool mysql_trans_prepare_alter_copy_data(THD *thd);
 bool mysql_trans_commit_alter_copy_data(THD *thd);
 bool mysql_alter_table(THD *thd, const LEX_CSTRING *new_db,
                        const LEX_CSTRING *new_name,
-                       HA_CREATE_INFO *create_info,
+                       Table_specification_st *create_info,
                        TABLE_LIST *table_list,
                        Alter_info *alter_info,
                        uint order_num, ORDER *order, bool ignore,
@@ -173,9 +171,6 @@ bool mysql_compare_tables(TABLE *table,
                           HA_CREATE_INFO *create_info,
                           bool *metadata_equal);
 bool mysql_recreate_table(THD *thd, TABLE_LIST *table_list, bool table_copy);
-bool mysql_create_like_table(THD *thd, TABLE_LIST *table,
-                             TABLE_LIST *src_table,
-                             Table_specification_st *create_info);
 bool mysql_rename_table(handlerton *base, const LEX_CSTRING *old_db,
                         const LEX_CSTRING *old_name, const LEX_CSTRING *new_db,
                         const LEX_CSTRING *new_name, LEX_CUSTRING *id,
diff --git a/sql/sql_tvc.cc b/sql/sql_tvc.cc
index be6e120716d..d80342a8395 100644
--- a/sql/sql_tvc.cc
+++ b/sql/sql_tvc.cc
@@ -1209,4 +1209,3 @@ bool JOIN::transform_in_predicates_into_in_subq(THD *thd)
   thd->lex->current_select= save_current_select;
   DBUG_RETURN(false);
 }
-
diff --git a/sql/sql_type.cc b/sql/sql_type.cc
index 69ca474eee1..0ede2b3fee2 100644
--- a/sql/sql_type.cc
+++ b/sql/sql_type.cc
@@ -2295,7 +2295,6 @@ Type_handler_decimal_result::make_num_distinct_aggregator_field(
                                                             const Item *item)
                                                             const
 {
-  DBUG_ASSERT(item->decimals <= DECIMAL_MAX_SCALE);
   return new (mem_root)
          Field_new_decimal(NULL, item->max_length,
                            (uchar *) (item->maybe_null() ? "" : 0),
@@ -2716,7 +2715,7 @@ Type_handler::Column_definition_set_attributes(THD *thd,
                                                column_definition_type_t type)
                                                const
 {
-  def->set_lex_charset_collation(attr.lex_charset_collation());
+  def->set_charset_collation_attrs(attr.charset_collation_attrs());
   def->set_length_and_dec(attr);
   return false;
 }
diff --git a/sql/sql_update.cc b/sql/sql_update.cc
index 82ef47608b9..74ed078e936 100644
--- a/sql/sql_update.cc
+++ b/sql/sql_update.cc
@@ -2130,10 +2130,9 @@ int multi_update::prepare(List<Item> &not_used_values,
       if (!tl)
 	DBUG_RETURN(1);
       update.link_in_list(tl, &tl->next_local);
-      tl->shared= table_count++;
+      table_ref->shared= tl->shared= table_count++;
       table->no_keyread=1;
       table->covering_keys.clear_all();
-      table->pos_in_table_list= tl;
       table->prepare_triggers_for_update_stmt_or_event();
       table->reset_default_fields();
     }
diff --git a/sql/sql_window.cc b/sql/sql_window.cc
index e50a17333e8..c0acecd138f 100644
--- a/sql/sql_window.cc
+++ b/sql/sql_window.cc
@@ -1,5 +1,5 @@
 /*
-   Copyright (c) 2016, 2017 MariaDB
+   Copyright (c) 2016, 2022 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
@@ -424,16 +424,49 @@ ORDER *st_select_lex::find_common_window_func_partition_fields(THD *thd)
 #define CMP_GT         2    // Greater then
 
 static
-int compare_order_elements(ORDER *ord1, ORDER *ord2)
+int compare_order_elements(ORDER *ord1, int weight1,
+                           ORDER *ord2, int weight2)
 {
   if (*ord1->item == *ord2->item && ord1->direction == ord2->direction)
     return CMP_EQ;
   Item *item1= (*ord1->item)->real_item();
   Item *item2= (*ord2->item)->real_item();
-  DBUG_ASSERT(item1->type() == Item::FIELD_ITEM &&
-              item2->type() == Item::FIELD_ITEM); 
-  int cmp= ((Item_field *) item1)->field->field_index -
-           ((Item_field *) item2)->field->field_index;
+
+  bool item1_field= (item1->type() == Item::FIELD_ITEM);
+  bool item2_field= (item2->type() == Item::FIELD_ITEM);
+
+  ptrdiff_t cmp;
+  if (item1_field && item2_field)
+  {
+    DBUG_ASSERT(((Item_field *) item1)->field->table ==
+                ((Item_field *) item2)->field->table);
+    cmp= ((Item_field *) item1)->field->field_index -
+         ((Item_field *) item2)->field->field_index;
+  }
+  else if (item1_field && !item2_field)
+    return CMP_LT;
+  else if (!item1_field && item2_field)
+    return CMP_LT;
+  else
+  {
+    /*
+      Ok, item1_field==NULL and item2_field==NULL.
+      We're not able to compare Item expressions. Order them according to
+      their passed "weight" (which comes from Window_spec::win_spec_number):
+    */
+    if (weight1 != weight2)
+      cmp= weight1 - weight2;
+    else
+    {
+      /*
+        The weight is the same. That is, the elements come from the same
+        window specification... This shouldn't happen.
+      */
+      DBUG_ASSERT(0);
+      cmp= item1 - item2;
+    }
+  }
+
   if (cmp == 0)
   {
     if (ord1->direction == ord2->direction)
@@ -446,7 +479,9 @@ int compare_order_elements(ORDER *ord1, ORDER *ord2)
 
 static
 int compare_order_lists(SQL_I_List<ORDER> *part_list1,
-                        SQL_I_List<ORDER> *part_list2)
+                        int spec_number1,
+                        SQL_I_List<ORDER> *part_list2,
+                        int spec_number2)
 {
   if (part_list1 == part_list2)
     return CMP_EQ;
@@ -471,7 +506,8 @@ int compare_order_lists(SQL_I_List<ORDER> *part_list1,
     if (!elem1 || !elem2)
       break;
 
-    if ((cmp= compare_order_elements(elem1, elem2)))
+    if ((cmp= compare_order_elements(elem1, spec_number1,
+                                     elem2, spec_number2)))
       return cmp;
   }
   if (elem1)
@@ -566,7 +602,9 @@ int compare_window_spec_joined_lists(Window_spec *win_spec1,
   win_spec1->join_partition_and_order_lists();
   win_spec2->join_partition_and_order_lists();
   int cmp= compare_order_lists(win_spec1->partition_list, 
-                               win_spec2->partition_list);
+                               win_spec1->win_spec_number,
+                               win_spec2->partition_list,
+                               win_spec2->win_spec_number);
   win_spec1->disjoin_partition_and_order_lists();
   win_spec2->disjoin_partition_and_order_lists();
   return cmp;
@@ -584,7 +622,9 @@ int compare_window_funcs_by_window_specs(Item_window_func *win_func1,
   if (win_spec1 == win_spec2)
     return CMP_EQ;
   cmp= compare_order_lists(win_spec1->partition_list, 
-                           win_spec2->partition_list);
+                           win_spec1->win_spec_number,
+                           win_spec2->partition_list,
+                           win_spec2->win_spec_number);
   if (cmp == CMP_EQ)
   {
     /* 
@@ -603,7 +643,9 @@ int compare_window_funcs_by_window_specs(Item_window_func *win_func1,
     }
 
     cmp= compare_order_lists(win_spec1->order_list,
-                             win_spec2->order_list);
+                             win_spec1->win_spec_number,
+                             win_spec2->order_list,
+                             win_spec2->win_spec_number);
 
     if (cmp != CMP_EQ)
       return cmp;
@@ -696,7 +738,9 @@ void order_window_funcs_by_window_specs(List<Item_window_func> *win_func_list)
       int cmp;
       if (win_spec_prev->partition_list == win_spec_curr->partition_list)
         cmp= compare_order_lists(win_spec_prev->order_list,
-                                 win_spec_curr->order_list);
+                                 win_spec_prev->win_spec_number,
+                                 win_spec_curr->order_list,
+                                 win_spec_curr->win_spec_number);
       else
         cmp= compare_window_spec_joined_lists(win_spec_prev, win_spec_curr);
       if (!(CMP_LT_C <= cmp && cmp <= CMP_GT_C))
diff --git a/sql/sql_window.h b/sql/sql_window.h
index 5e76a33dcd0..66ea8c7dd4d 100644
--- a/sql/sql_window.h
+++ b/sql/sql_window.h
@@ -1,5 +1,5 @@
 /*
-   Copyright (c) 2016, 2017 MariaDB
+   Copyright (c) 2016, 2022 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
@@ -120,10 +120,15 @@ class Window_spec : public Sql_alloc
 
   Window_spec *referenced_win_spec;
 
-  Window_spec(LEX_CSTRING *win_ref, 
-              SQL_I_List<ORDER> *part_list,
-              SQL_I_List<ORDER> *ord_list,
-              Window_frame *win_frame)
+  /*
+    Window_spec objects are numbered by the number of their appearance in the
+    query. This is used by compare_order_elements() to provide a predictable
+    ordering of PARTITION/ORDER BY clauses.
+  */
+  int win_spec_number;
+
+  Window_spec(LEX_CSTRING *win_ref, SQL_I_List<ORDER> *part_list,
+              SQL_I_List<ORDER> *ord_list, Window_frame *win_frame)
     : window_names_are_checked(false), window_ref(win_ref),
       partition_list(part_list), save_partition_list(NULL),
       order_list(ord_list), save_order_list(NULL),
diff --git a/sql/sql_yacc.yy b/sql/sql_yacc.yy
index 826c470511f..80d6c12e4e5 100644
--- a/sql/sql_yacc.yy
+++ b/sql/sql_yacc.yy
@@ -213,7 +213,9 @@ void _CONCAT_UNDERSCORED(turn_parser_debug_on,yyparse)()
   Lex_length_and_dec_st Lex_length_and_dec;
   Lex_cast_type_st Lex_cast_type;
   Lex_field_type_st Lex_field_type;
-  Lex_charset_collation_st Lex_charset_collation;
+  Lex_exact_charset_extended_collation_attrs_st
+                    Lex_exact_charset_extended_collation_attrs;
+  Lex_extended_collation_st Lex_extended_collation;
   Lex_dyncol_type_st Lex_dyncol_type;
   Lex_for_loop_st for_loop;
   Lex_for_loop_bounds_st for_loop_bounds;
@@ -343,12 +345,11 @@ bool my_yyoverflow(short **a, YYSTYPE **b, size_t *yystacksize);
 */
 
 %ifdef MARIADB
-%expect 67
+%expect 71
 %else
-%expect 69
+%expect 72
 %endif
 
-
 /*
    Comments for TOKENS.
    For each token, please include in the same line a comment that contains
@@ -376,6 +377,7 @@ bool my_yyoverflow(short **a, YYSTYPE **b, size_t *yystacksize);
 */
 %token  <NONE> ABORT_SYM              /* INTERNAL (used in lex) */
 %token  <NONE> IMPOSSIBLE_ACTION      /* To avoid warning for yyerrlab1 */
+%token  <NONE> FORCE_LOOKAHEAD        /* INTERNAL never returned by the lexer */
 %token  <NONE> END_OF_INPUT           /* INTERNAL */
 %token  <kwd>  COLON_ORACLE_SYM       /* INTERNAL */
 %token  <kwd>  PARAM_MARKER           /* INTERNAL */
@@ -389,7 +391,6 @@ bool my_yyoverflow(short **a, YYSTYPE **b, size_t *yystacksize);
 %token  <NONE> WITH_ROLLUP_SYM        /* INTERNAL */
 %token  <NONE> WITH_SYSTEM_SYM        /* INTERNAL */
 
-
 /*
   Identifiers
 */
@@ -1379,7 +1380,7 @@ bool my_yyoverflow(short **a, YYSTYPE **b, size_t *yystacksize);
         field_type_misc
         json_table_field_type
 
-%type <Lex_charset_collation>
+%type <Lex_exact_charset_extended_collation_attrs>
         binary
         opt_binary
         opt_binary_and_compression
@@ -1387,6 +1388,10 @@ bool my_yyoverflow(short **a, YYSTYPE **b, size_t *yystacksize);
         attribute_list
         field_def
 
+%type <Lex_extended_collation>
+        collation_name
+        collation_name_or_default
+
 
 %type <Lex_dyncol_type> opt_dyncol_type dyncol_type
         numeric_dyncol_type temporal_dyncol_type string_dyncol_type
@@ -1579,14 +1584,11 @@ bool my_yyoverflow(short **a, YYSTYPE **b, size_t *yystacksize);
                   text_or_password
 
 %type <charset>
-        opt_collate_or_default
         charset_name
         charset_or_alias
         charset_name_or_default
         old_or_new_charset_name
         old_or_new_charset_name_or_default
-        collation_name
-        collation_name_or_default
         opt_load_data_charset
         UNDERSCORE_CHARSET
 
@@ -2367,9 +2369,9 @@ create:
               If the table exists, we should either not create it or replace it
             */
             lex->query_tables->open_strategy= TABLE_LIST::OPEN_STUB;
-            lex->create_info.default_table_charset= NULL;
             lex->name= null_clex_str;
             lex->create_last_non_select_table= lex->last_table();
+            lex->inc_select_stack_outer_barrier();
           }
           create_body
           {
@@ -2509,9 +2511,7 @@ create:
           }
         | create_or_replace DATABASE opt_if_not_exists ident
           {
-            Lex->create_info.default_table_charset= NULL;
-            Lex->create_info.schema_comment= NULL;
-            Lex->create_info.used_fields= 0;
+            Lex->create_info.init();
           }
           opt_create_database_options
           {
@@ -2709,6 +2709,9 @@ sequence_def:
           }
         ;
 
+/* this rule is used to force look-ahead in the parser */
+force_lookahead: {} | FORCE_LOOKAHEAD {} ;
+
 server_def:
           SERVER_SYM opt_if_not_exists ident_or_text
           {
@@ -2910,7 +2913,7 @@ ev_sql_stmt:
 
             lex->sphead->set_body_start(thd, lip->get_cpp_ptr());
           }
-          sp_proc_stmt
+          sp_proc_stmt force_lookahead
           {
             /* return back to the original memory root ASAP */
             if (Lex->sp_body_finalize_event(thd))
@@ -5527,15 +5530,9 @@ default_charset:
 default_collation:
           opt_default COLLATE_SYM opt_equal collation_name_or_default
           {
-            HA_CREATE_INFO *cinfo= &Lex->create_info;
-            if (unlikely((cinfo->used_fields & HA_CREATE_USED_DEFAULT_CHARSET) &&
-                         cinfo->default_table_charset && $4 &&
-                         !($4= merge_charset_and_collation(cinfo->default_table_charset,
-                                                           $4))))
+            Table_specification_st *cinfo= &Lex->create_info;
+            if (unlikely(cinfo->add_table_option_default_collation($4)))
               MYSQL_YYABORT;
-
-            Lex->create_info.default_table_charset= $4;
-            Lex->create_info.used_fields|= HA_CREATE_USED_DEFAULT_CHARSET;
           }
         ;
 
@@ -5786,10 +5783,10 @@ field_type_or_serial:
           }
           field_def
           {
-            Lex_charset_collation tmp= $1.lex_charset_collation();
-            if (tmp.merge_charset_clause_and_collate_clause($3))
+            auto tmp= $1.charset_collation_attrs();
+            if (tmp.merge_column_charset_clause_and_collate_clause($3))
               MYSQL_YYABORT;
-            Lex->last_field->set_lex_charset_collation(tmp);
+            Lex->last_field->set_charset_collation_attrs(tmp);
           }
         | SERIAL_SYM
           {
@@ -5827,7 +5824,7 @@ field_def:
         | attribute_list compressed_deprecated_column_attribute { $$= $1; }
         | attribute_list compressed_deprecated_column_attribute attribute_list
           {
-            if (($$= $1).merge_collate_clause_and_collate_clause($3))
+            if (($$= $1).merge_column_collate_clause_and_collate_clause($3))
               MYSQL_YYABORT;
           }
         | opt_generated_always AS virtual_column_func
@@ -6076,7 +6073,7 @@ field_type_string:
         | nchar opt_field_length opt_bin_mod
           {
             $$.set(&type_handler_string, $2,
-                   Lex_charset_collation::national($3));
+                   Lex_exact_charset_extended_collation_attrs::national($3));
           }
         | BINARY opt_field_length
           {
@@ -6093,7 +6090,7 @@ field_type_string:
         | nvarchar opt_field_length opt_compressed opt_bin_mod
           {
             $$.set(&type_handler_varchar, $2,
-                   Lex_charset_collation::national($4));
+                   Lex_exact_charset_extended_collation_attrs::national($4));
           }
         | VARBINARY opt_field_length opt_compressed
           {
@@ -6308,7 +6305,7 @@ opt_precision:
 attribute_list:
           attribute_list attribute
           {
-             if (($$= $1).merge_collate_clause_and_collate_clause($2))
+             if (($$= $1).merge_column_collate_clause_and_collate_clause($2))
                MYSQL_YYABORT;
           }
         | attribute
@@ -6335,7 +6332,7 @@ attribute:
           }
         | COLLATE_SYM collation_name
           {
-            $$.set_collate_exact($2);
+            $$= Lex_exact_charset_extended_collation_attrs($2);
           }
         | serial_attribute { $$.init(); }
         ;
@@ -6475,20 +6472,17 @@ old_or_new_charset_name_or_default:
 collation_name:
           ident_or_text
           {
-            if (unlikely(!($$= mysqld_collation_get_by_name($1.str,
+            CHARSET_INFO *cs;
+            if (unlikely(!(cs= mysqld_collation_get_by_name($1.str,
                                                             thd->get_utf8_flag()))))
               MYSQL_YYABORT;
+            $$= Lex_extended_collation(Lex_exact_collation(cs));
           }
         ;
 
-opt_collate_or_default:
-          /* empty */ { $$=NULL; }
-        | COLLATE_SYM collation_name_or_default { $$=$2; }
-        ;
-
 collation_name_or_default:
           collation_name { $$=$1; }
-        | DEFAULT    { $$=NULL; }
+        | DEFAULT        { $$.set_collate_default(); }
         ;
 
 opt_default:
@@ -6531,11 +6525,18 @@ binary:
           }
         | charset_or_alias COLLATE_SYM collation_name
           {
-            if ($$.set_charset_collate_exact($1, $3))
+            if ($3.merge_exact_charset(Lex_exact_charset($1)))
               MYSQL_YYABORT;
+            $$= Lex_exact_charset_extended_collation_attrs($3);
+          }
+        | COLLATE_SYM collation_name
+          {
+            $$= Lex_exact_charset_extended_collation_attrs($2);
+          }
+        | COLLATE_SYM DEFAULT
+          {
+            $$.set_collate_default();
           }
-        | COLLATE_SYM collation_name  { $$.set_collate_exact($2); }
-        | COLLATE_SYM DEFAULT         { $$.set_collate_default(); }
         ;
 
 opt_bin_mod:
@@ -6943,9 +6944,7 @@ alter:
           }
         | ALTER DATABASE ident_or_empty
           {
-            Lex->create_info.default_table_charset= NULL;
-            Lex->create_info.schema_comment= NULL;
-            Lex->create_info.used_fields= 0;
+            Lex->create_info.init();
             if (Lex->main_select_push(true))
               MYSQL_YYABORT;
           }
@@ -6961,8 +6960,7 @@ alter:
           }
         | ALTER DATABASE COMMENT_SYM opt_equal TEXT_STRING_sys
           {
-            Lex->create_info.default_table_charset= NULL;
-            Lex->create_info.used_fields= 0;
+            Lex->create_info.init();
             Lex->create_info.schema_comment= thd->make_clex_string($5);
             Lex->create_info.used_fields|= HA_CREATE_USED_COMMENT;
           }
@@ -7606,19 +7604,15 @@ alter_list_item:
             lex->alter_info.flags|= ALTER_RENAME_INDEX;
           }
         | CONVERT_SYM TO_SYM charset charset_name_or_default
-                             opt_collate_or_default
           {
-            if (!$4)
-            {
-              $4= thd->variables.collation_database;
-            }
-            $5= $5 ? $5 : $4;
-            if (unlikely(!my_charset_same($4,$5)))
-              my_yyabort_error((ER_COLLATION_CHARSET_MISMATCH, MYF(0),
-                                $5->coll_name.str, $4->cs_name.str));
-            if (unlikely(Lex->create_info.add_alter_list_item_convert_to_charset($5)))
+            if (Lex->add_alter_list_item_convert_to_charset($4))
+              MYSQL_YYABORT;
+          }
+        | CONVERT_SYM TO_SYM charset charset_name_or_default
+                             COLLATE_SYM collation_name_or_default
+          {
+            if (Lex->add_alter_list_item_convert_to_charset($4, $6))
               MYSQL_YYABORT;
-            Lex->alter_info.flags|= ALTER_CONVERT_TO;
           }
         | create_table_options_space_separated
           {
@@ -9735,7 +9729,9 @@ string_factor_expr:
           primary_expr
         | string_factor_expr COLLATE_SYM collation_name
           {
-            if (unlikely(!($$= new (thd->mem_root) Item_func_set_collation(thd, $1, $3))))
+            if (unlikely(!($$= new (thd->mem_root)
+                               Item_func_set_collation(thd, $1,
+                                                       $3.charset_info()))))
               MYSQL_YYABORT;
           }
         ;
@@ -11355,7 +11351,7 @@ json_table_column_type:
                                             COLUMN_DEFINITION_TABLE_FIELD);
             if (Lex->json_table->m_cur_json_table_column->
                   set(thd, Json_table_column::PATH, $3,
-                      $1.lex_charset_collation()))
+                      $1.charset_collation_attrs()))
             {
               MYSQL_YYABORT;
             }
@@ -11366,7 +11362,7 @@ json_table_column_type:
                                             COLUMN_DEFINITION_TABLE_FIELD);
             if (Lex->json_table->m_cur_json_table_column->
                   set(thd, Json_table_column::EXISTS_PATH, $4,
-                      $1.lex_charset_collation()))
+                      $1.charset_collation_attrs()))
                MYSQL_YYABORT;
           }
         ;
@@ -12946,6 +12942,7 @@ insert_start: {
                 if (Lex->main_select_push())
                   MYSQL_YYABORT;
                 mysql_init_select(Lex);
+                Lex->inc_select_stack_outer_barrier();
                 Lex->current_select->parsing_place= BEFORE_OPT_LIST;
               }
               ;
@@ -16576,26 +16573,20 @@ option_value_no_option_type:
               thd->parse_error();
             MYSQL_YYABORT;
           }
-        | NAMES_SYM charset_name_or_default opt_collate_or_default
+        | NAMES_SYM charset_name_or_default
           {
-            if (sp_create_assignment_lex(thd, $1.pos()))
+            CHARSET_INFO *def= global_system_variables.character_set_client;
+            Lex_exact_charset_opt_extended_collate tmp($2 ? $2 : def, false);
+            if (Lex->set_names($1.pos(), tmp, yychar == YYEMPTY))
               MYSQL_YYABORT;
-            LEX *lex= Lex;
-            CHARSET_INFO *cs2;
-            CHARSET_INFO *cs3;
-            cs2= $2 ? $2 : global_system_variables.character_set_client;
-            cs3= $3 ? $3 : cs2;
-            if (unlikely(!my_charset_same(cs2, cs3)))
-            {
-              my_error(ER_COLLATION_CHARSET_MISMATCH, MYF(0),
-                       cs3->coll_name.str, cs2->cs_name.str);
-              MYSQL_YYABORT;
-            }
-            set_var_collation_client *var;
-            var= new (thd->mem_root) set_var_collation_client(cs3, cs3, cs3);
-            if (unlikely(var == NULL) ||
-                unlikely(lex->var_list.push_back(var, thd->mem_root)) ||
-                unlikely(sp_create_assignment_instr(thd, yychar == YYEMPTY)))
+          }
+        | NAMES_SYM charset_name_or_default
+                    COLLATE_SYM collation_name_or_default
+          {
+            CHARSET_INFO *def= global_system_variables.character_set_client;
+            Lex_exact_charset_opt_extended_collate tmp($2 ? $2 : def, false);
+            if (tmp.merge_collation($4) ||
+                Lex->set_names($1.pos(), tmp, yychar == YYEMPTY))
               MYSQL_YYABORT;
           }
         | DEFAULT ROLE_SYM grant_role
@@ -17776,8 +17767,8 @@ trigger_tail:
 
             lex->sphead->set_body_start(thd, lip->get_cpp_tok_start());
           }
-          sp_proc_stmt /* $19 */
-          { /* $20 */
+          sp_proc_stmt /* $19 */ force_lookahead /* $20 */
+          { /* $21 */
             LEX *lex= Lex;
 
             lex->sql_command= SQLCOM_CREATE_TRIGGER;
@@ -17817,7 +17808,6 @@ sf_return_type:
           }
         ;
 
-
 /*************************************************************************/
 
 xa:
@@ -18155,7 +18145,7 @@ sf_c_chistics_and_body_standalone:
             lex->sphead->set_c_chistics(lex->sp_chistics);
             lex->sphead->set_body_start(thd, YYLIP->get_cpp_tok_start());
           }
-          sp_proc_stmt_in_returns_clause
+          sp_proc_stmt_in_returns_clause force_lookahead
           {
             if (unlikely(Lex->sp_body_finalize_function(thd)))
               MYSQL_YYABORT;
@@ -18176,7 +18166,7 @@ sp_tail_standalone:
             Lex->sphead->set_c_chistics(Lex->sp_chistics);
             Lex->sphead->set_body_start(thd, YYLIP->get_cpp_tok_start());
           }
-          sp_proc_stmt
+          sp_proc_stmt force_lookahead
           {
             if (unlikely(Lex->sp_body_finalize_procedure(thd)))
               MYSQL_YYABORT;
@@ -19027,8 +19017,7 @@ sf_c_chistics_and_body_standalone:
             lex->sphead->set_c_chistics(lex->sp_chistics);
             lex->sphead->set_body_start(thd, YYLIP->get_cpp_tok_start());
           }
-          sp_tail_is
-          sp_body
+          sp_tail_is sp_body force_lookahead
           {
             if (unlikely(Lex->sp_body_finalize_function(thd)))
               MYSQL_YYABORT;
diff --git a/sql/structs.h b/sql/structs.h
index eab15c4d92b..52e47d5d7ee 100644
--- a/sql/structs.h
+++ b/sql/structs.h
@@ -699,17 +699,18 @@ public:
   }
   void set(const Type_handler *handler,
            const Lex_length_and_dec_st &length_and_dec,
-           const Lex_charset_collation_st &coll)
+           const Lex_column_charset_collation_attrs_st &coll)
   {
     m_handler= handler;
-    m_ci= coll.charset_collation();
+    m_ci= coll.charset_info();
     Lex_length_and_dec_st::operator=(length_and_dec);
     m_collation_type= ((uint8) coll.type()) & 0x3;
   }
-  void set(const Type_handler *handler, const Lex_charset_collation_st &coll)
+  void set(const Type_handler *handler,
+           const Lex_column_charset_collation_attrs_st &coll)
   {
     m_handler= handler;
-    m_ci= coll.charset_collation();
+    m_ci= coll.charset_info();
     Lex_length_and_dec_st::reset();
     m_collation_type= ((uint8) coll.type()) & 0x3;
   }
@@ -734,10 +735,10 @@ public:
   }
   const Type_handler *type_handler() const { return m_handler; }
   CHARSET_INFO *charset_collation() const { return m_ci; }
-  Lex_charset_collation lex_charset_collation() const
+  Lex_column_charset_collation_attrs charset_collation_attrs() const
   {
-    return Lex_charset_collation(m_ci,
-                                 (Lex_charset_collation_st::Type)
+    return Lex_column_charset_collation_attrs(m_ci,
+                                 (Lex_column_charset_collation_attrs_st::Type)
                                  m_collation_type);
   }
 };
@@ -768,7 +769,7 @@ public:
     m_ci= cs;
     Lex_length_and_dec_st::reset();
   }
-  bool set(int type, const Lex_charset_collation_st &collation,
+  bool set(int type, const Lex_column_charset_collation_attrs_st &collation,
            CHARSET_INFO *charset)
   {
     CHARSET_INFO *tmp= collation.resolved_to_character_set(charset);
diff --git a/sql/table.cc b/sql/table.cc
index efa7ae3a5b6..f0d51495953 100644
--- a/sql/table.cc
+++ b/sql/table.cc
@@ -45,30 +45,20 @@
 #include "ha_sequence.h"
 #include "sql_show.h"
 #include "opt_trace.h"
+#include "sql_db.h"              // get_default_db_collation
 
 /* For MySQL 5.7 virtual fields */
 #define MYSQL57_GENERATED_FIELD 128
 #define MYSQL57_GCOL_HEADER_SIZE 4
 
-class Table_arena: public Query_arena
-{
-public:
-  Table_arena(MEM_ROOT *mem_root, enum enum_state state_arg) :
-          Query_arena(mem_root, state_arg){}
-  virtual Type type() const
-  {
-    return TABLE_ARENA;
-  }
-};
-
 bool TABLE::init_expr_arena(MEM_ROOT *mem_root)
 {
   /*
     We need to use CONVENTIONAL_EXECUTION here to ensure that
     any new items created by fix_fields() are not reverted.
   */
-  expr_arena= new (alloc_root(mem_root, sizeof(Table_arena)))
-                Table_arena(mem_root, Query_arena::STMT_CONVENTIONAL_EXECUTION);
+  expr_arena= new (alloc_root(mem_root, sizeof(Query_arena)))
+                Query_arena(mem_root, Query_arena::STMT_CONVENTIONAL_EXECUTION);
   return expr_arena == NULL;
 }
 
@@ -3514,6 +3504,19 @@ int TABLE_SHARE::init_from_sql_statement_string(THD *thd, bool write,
   else
     thd->set_n_backup_active_arena(arena, &backup);
 
+  /*
+    THD::reset_db() does not set THD::db_charset,
+    so it keeps pointing to the character set and collation
+    of the current database, rather than the database of the
+    new initialized table. After reset_db() the result of
+    get_default_db_collation() can be wrong. The latter is
+    used inside charset_collation_context_create_table_in_db().
+    Let's initialize ctx before calling reset_db().
+    This makes sure the db.opt file to be loaded properly when needed.
+  */
+  Charset_collation_context
+    ctx(thd->charset_collation_context_create_table_in_db(db.str));
+
   thd->reset_db(&db);
   lex_start(thd);
 
@@ -3521,6 +3524,9 @@ int TABLE_SHARE::init_from_sql_statement_string(THD *thd, bool write,
                 sql_unusable_for_discovery(thd, hton, sql_copy))))
     goto ret;
 
+  if (thd->lex->create_info.resolve_to_charset_collation_context(thd, ctx))
+    DBUG_RETURN(true);
+
   thd->lex->create_info.db_type= hton;
 #ifdef WITH_PARTITION_STORAGE_ENGINE
   thd->work_part_info= 0;                       // For partitioning
@@ -3659,8 +3665,7 @@ class Vcol_expr_context
   bool inited;
   THD *thd;
   TABLE *table;
-  LEX *old_lex;
-  LEX lex;
+  Query_arena backup_arena;
   table_map old_map;
   Security_context *save_security_ctx;
   sql_mode_t save_sql_mode;
@@ -3670,7 +3675,6 @@ public:
     inited(false),
     thd(_thd),
     table(_table),
-    old_lex(thd->lex),
     old_map(table->map),
     save_security_ctx(thd->security_ctx),
     save_sql_mode(thd->variables.sql_mode) {}
@@ -3682,18 +3686,6 @@ public:
 
 bool Vcol_expr_context::init()
 {
-  /*
-      As this is vcol expression we must narrow down name resolution to
-      single table.
-  */
-  if (init_lex_with_single_table(thd, table, &lex))
-  {
-    my_error(ER_OUT_OF_RESOURCES, MYF(0));
-    table->map= old_map;
-    return true;
-  }
-
-  lex.sql_command= old_lex->sql_command;
   thd->variables.sql_mode= 0;
 
   TABLE_LIST const *tl= table->pos_in_table_list;
@@ -3702,6 +3694,8 @@ bool Vcol_expr_context::init()
   if (table->pos_in_table_list->security_ctx)
     thd->security_ctx= tl->security_ctx;
 
+  thd->set_n_backup_active_arena(table->expr_arena, &backup_arena);
+
   inited= true;
   return false;
 }
@@ -3710,9 +3704,9 @@ Vcol_expr_context::~Vcol_expr_context()
 {
   if (!inited)
     return;
-  end_lex_with_single_table(thd, table, old_lex);
   table->map= old_map;
   thd->security_ctx= save_security_ctx;
+  thd->restore_active_arena(table->expr_arena, &backup_arena);
   thd->variables.sql_mode= save_sql_mode;
 }
 
@@ -4111,6 +4105,7 @@ enum open_frm_error open_table_from_share(THD *thd, TABLE_SHARE *share,
   outparam->s= share;
   outparam->db_stat= db_stat;
   outparam->write_row_record= NULL;
+  outparam->status= STATUS_NO_RECORD;
 
   if (share->incompatible_version &&
       !(ha_open_flags & (HA_OPEN_FOR_ALTER | HA_OPEN_FOR_REPAIR |
@@ -6960,7 +6955,7 @@ Item *Field_iterator_view::create_item(THD *thd)
 Item *create_view_field(THD *thd, TABLE_LIST *view, Item **field_ref,
                         LEX_CSTRING *name)
 {
-  bool save_wrapper= thd->lex->first_select_lex()->no_wrap_view_item;
+  bool save_wrapper= thd->lex->current_select->no_wrap_view_item;
   Item *field= *field_ref;
   DBUG_ENTER("create_view_field");
 
diff --git a/sql/wsrep_trans_observer.h b/sql/wsrep_trans_observer.h
index 70759e381a5..f99d33094df 100644
--- a/sql/wsrep_trans_observer.h
+++ b/sql/wsrep_trans_observer.h
@@ -232,6 +232,10 @@ static inline int wsrep_before_prepare(THD* thd, bool all)
   WSREP_DEBUG("wsrep_before_prepare: %d", wsrep_is_real(thd, all));
   int ret= 0;
   DBUG_ASSERT(wsrep_run_commit_hook(thd, all));
+  if ((ret= thd->wsrep_parallel_slave_wait_for_prior_commit()))
+  {
+    DBUG_RETURN(ret);
+  }
   if ((ret= thd->wsrep_cs().before_prepare()) == 0)
   {
     DBUG_ASSERT(!thd->wsrep_trx().ws_meta().gtid().is_undefined());