Merge branch '10.7' into 10.8

45 files changed, 936 insertions, 315 deletions

diff --git a/sql/field.cc b/sql/field.cc
index 6b5e8251822..530a8c09c59 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/ha_partition.cc b/sql/ha_partition.cc
index 3a9d1eccaf5..0ecbd6b3fc1 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
   {
@@ -4197,13 +4199,7 @@ int ha_partition::external_lock(THD *thd, int lock_type)
   {
     if (m_part_info->part_expr)
       m_part_info->part_expr->walk(&Item::register_field_in_read_map, 1, 0);
-    if (m_part_info->part_type == VERSIONING_PARTITION &&
-      /* TODO: MDEV-20345 exclude more inapproriate commands like INSERT
-         These commands may be excluded because working history partition is needed
-         only for versioned DML. */
-      thd->lex->sql_command != SQLCOM_SELECT &&
-      thd->lex->sql_command != SQLCOM_INSERT_SELECT &&
-      (error= m_part_info->vers_set_hist_part(thd)))
+    if ((error= m_part_info->vers_set_hist_part(thd)))
       goto err_handler;
   }
   DBUG_RETURN(0);
@@ -4349,11 +4345,7 @@ int ha_partition::start_stmt(THD *thd, thr_lock_type lock_type)
   {
     if (m_part_info->part_expr)
       m_part_info->part_expr->walk(&Item::register_field_in_read_map, 1, 0);
-    if (m_part_info->part_type == VERSIONING_PARTITION &&
-      // TODO: MDEV-20345 (see above)
-      thd->lex->sql_command != SQLCOM_SELECT &&
-      thd->lex->sql_command != SQLCOM_INSERT_SELECT)
-      error= m_part_info->vers_set_hist_part(thd);
+    error= m_part_info->vers_set_hist_part(thd);
   }
   DBUG_RETURN(error);
 }
@@ -10591,7 +10583,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/handler.cc b/sql/handler.cc
index c5bd441eaf0..9ba96244557 100644
--- a/sql/handler.cc
+++ b/sql/handler.cc
@@ -4364,6 +4364,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;
       }
     }
@@ -5809,6 +5810,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_mode && f->is_real_null(0))
       {
diff --git a/sql/item.cc b/sql/item.cc
index 7c9aa662609..1f6d585efd7 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;
     }
@@ -7983,8 +7982,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. */
@@ -7997,8 +7995,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;
       }
 
@@ -9485,6 +9482,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;
@@ -9526,7 +9529,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();
@@ -9555,8 +9557,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();
 }
 
@@ -9631,6 +9632,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();
@@ -9690,6 +9697,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 fbe26c597a9..f6c123e79d4 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
 
@@ -6626,7 +6632,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),
@@ -6643,6 +6648,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)
+  { return Item::val_datetime_packed(thd); }
+  longlong val_time_packed(THD *thd)
+  { return Item::val_time_packed(thd); }
 
   /* Result variants */
   double val_result() override;
@@ -6656,6 +6665,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)"
@@ -6671,11 +6681,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)) ||
@@ -6954,7 +6965,7 @@ public:
   for any value.
 */
 
-class Item_cache: public Item,
+class Item_cache: public Item_fixed_hybrid,
                   public Type_handler_hybrid_field_type
 {
 protected:
@@ -6985,7 +6996,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),
@@ -6994,10 +7005,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),
@@ -7006,6 +7018,7 @@ protected:
     set_maybe_null();
     null_value= 1;
     null_value_inside= true;
+    quick_fix_field();
   }
 
 public:
@@ -7058,10 +7071,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 034833a0ba2..9123bd96d5e 100644
--- a/sql/item_cmpfunc.cc
+++ b/sql/item_cmpfunc.cc
@@ -7668,7 +7668,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 94b14fa70ac..dae2be5aa4e 100644
--- a/sql/item_func.cc
+++ b/sql/item_func.cc
@@ -2496,7 +2496,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_subselect.cc b/sql/item_subselect.cc
index e4bbe475212..b1f913b64ef 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,
@@ -1355,11 +1356,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
   {
@@ -1395,6 +1403,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 2b8aae57ece..ca757b6b94a 100644
--- a/sql/item_sum.cc
+++ b/sql/item_sum.cc
@@ -1551,6 +1551,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);
@@ -1962,12 +1964,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);
 }
 
diff --git a/sql/mysqld.cc b/sql/mysqld.cc
index 75d8b4af72e..fce7d32549e 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 */
@@ -5352,6 +5380,7 @@ static int init_server_components()
   if (ha_recover(0))
     unireg_abort(1);
 
+  start_handle_manager();
   if (opt_bin_log)
   {
     int error;
@@ -5809,8 +5838,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 548a9ed6f67..294834b84f0 100644
--- a/sql/partition_info.cc
+++ b/sql/partition_info.cc
@@ -822,6 +822,9 @@ bool partition_info::has_unique_name(partition_element *element)
 */
 int partition_info::vers_set_hist_part(THD *thd)
 {
+  if (!vers_require_hist_part(thd))
+    return 0;
+
   if (table->pos_in_table_list &&
       table->pos_in_table_list->partition_names)
   {
@@ -830,12 +833,10 @@ int partition_info::vers_set_hist_part(THD *thd)
   if (vers_info->limit)
   {
     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));
@@ -845,17 +846,8 @@ int partition_info::vers_set_hist_part(THD *thd)
       vers_info->hist_part= next;
       records= next_records;
     }
-    if (records >= vers_info->limit)
-    {
-      if (next == vers_info->now_part)
-      {
-        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;
-    }
+    if (records >= vers_info->limit && next != vers_info->now_part)
+      vers_info->hist_part= next;
     return 0;
   }
 
@@ -880,6 +872,45 @@ int partition_info::vers_set_hist_part(THD *thd)
 }
 
 
+/**
+  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->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);
+  for (; part_id < part_id_end; ++part_id)
+    bitmap_set_bit(&read_partitions, part_id);
+
+  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 f2b54d14a49..46e17044b85 100644
--- a/sql/partition_info.h
+++ b/sql/partition_info.h
@@ -400,7 +400,13 @@ public:
     vers_info->limit= limit;
     return !limit;
   }
+  bool vers_require_hist_part(THD *thd) const
+  {
+    return part_type == VERSIONING_PARTITION &&
+      thd->lex->vers_history_generating();
+  }
   int vers_set_hist_part(THD *thd);
+  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.h b/sql/rpl_gtid.h
index 759e4e4eea3..fa05148a39b 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 799a62b83bc..c5965540136 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 23a35c5805f..18262c71e87 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)
@@ -7142,11 +7240,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,12 +7256,39 @@ 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;
 
-      /* Reset the domain_id_filter flag. */
-      mi->domain_id_filter.reset_filter();
+      if (unlikely(gtid_skip_enqueue))
+      {
+        error= ER_SLAVE_RELAY_LOG_WRITE_FAILURE;
+        sql_print_error("Recieved a group closing %s event "
+                        "at %llu position in the group while there are "
+                        "still %llu events to skip upon reconnecting; "
+                        "the last seen GTID is %u-%u-%llu",
+                        Log_event::get_type_str((Log_event_type) (uchar)
+                                                buf[EVENT_TYPE_OFFSET]),
+                        (mi->events_queued_since_last_gtid -
+                         mi->gtid_reconnect_event_skip_count),
+                        mi->events_queued_since_last_gtid,
+                        mi->last_queued_gtid);
+        goto err;
+      }
+      else
+      {
+        /*
+          The whole of the current event group is queued. So in case of
+          reconnect we can start from after the current GTID.
+        */
+        mi->gtid_current_pos.update(&mi->last_queued_gtid);
+        mi->events_queued_since_last_gtid= 0;
+
+        /* Reset the domain_id_filter flag. */
+        mi->domain_id_filter.reset_filter();
+      }
     }
 
 skip_relay_logging:
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 baf4d7d1d82..ec69730a702 100644
--- a/sql/sp_head.cc
+++ b/sql/sp_head.cc
@@ -829,7 +829,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_base.cc b/sql/sql_base.cc
index bc16af59d8d..4f041c061c3 100644
--- a/sql/sql_base.cc
+++ b/sql/sql_base.cc
@@ -808,7 +808,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 ||
@@ -8399,9 +8406,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));
@@ -8454,7 +8463,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 27875e3837b..3eab44c4179 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
@@ -679,7 +679,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 +862,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 +1435,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 +1562,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 +6731,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
diff --git a/sql/sql_class.h b/sql/sql_class.h
index f7f646e8310..5d56d3a048d 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
@@ -1183,7 +1183,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) :
@@ -2361,6 +2361,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);
 
 
@@ -2556,7 +2589,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
@@ -3590,10 +3624,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;
@@ -4391,8 +4421,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)
@@ -4982,18 +5011,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; }
@@ -5243,6 +5272,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);
diff --git a/sql/sql_expression_cache.cc b/sql/sql_expression_cache.cc
index 351ec258ddb..bc44ecd79fa 100644
--- a/sql/sql_expression_cache.cc
+++ b/sql/sql_expression_cache.cc
@@ -63,7 +63,7 @@ void Expression_cache_tmptable::disable_cache()
   cache_table= NULL;
   update_tracker();
   if (tracker)
-    tracker->cache= NULL;
+    tracker->detach_from_cache();
 }
 
 
@@ -189,6 +189,8 @@ Expression_cache_tmptable::~Expression_cache_tmptable()
   else
   {
     update_tracker();
+    if (tracker)
+      tracker->detach_from_cache();
     tracker= NULL;
   }
 }
diff --git a/sql/sql_expression_cache.h b/sql/sql_expression_cache.h
index 031773adb9f..9c618a5ae9b 100644
--- a/sql/sql_expression_cache.h
+++ b/sql/sql_expression_cache.h
@@ -83,7 +83,11 @@ public:
     cache(c), hit(0), miss(0), state(UNINITED)
   {}
 
+private:
+  // This can be NULL if the cache is already deleted
   Expression_cache *cache;
+
+public:
   ulong hit, miss;
   enum expr_cache_state state;
 
@@ -91,6 +95,7 @@ public:
   void set(ulong h, ulong m, enum expr_cache_state s)
   {hit= h; miss= m; state= s;}
 
+  void detach_from_cache() { cache= NULL; }
   void fetch_current_stats()
   {
     if (cache)
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 aecd70f1810..d22a5670495 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 e513b350d0d..0313aa2e65b 100644
--- a/sql/sql_lex.cc
+++ b/sql/sql_lex.cc
@@ -745,7 +745,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;
@@ -1238,6 +1237,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;
@@ -5322,17 +5322,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);
@@ -5359,8 +5363,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;
diff --git a/sql/sql_lex.h b/sql/sql_lex.h
index b692f33bc66..4625e60703b 100644
--- a/sql/sql_lex.h
+++ b/sql/sql_lex.h
@@ -3311,6 +3311,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;
@@ -3754,6 +3760,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();
@@ -4532,6 +4549,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 b097f84132b..cdafc3f720e 100644
--- a/sql/sql_parse.cc
+++ b/sql/sql_parse.cc
@@ -8886,6 +8886,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));
 }
 
@@ -8913,6 +8914,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));
 }
 
@@ -9056,9 +9058,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);
 }
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 0f4f5eda83b..d8b57e18f86 100644
--- a/sql/sql_select.cc
+++ b/sql/sql_select.cc
@@ -18403,14 +18403,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
@@ -18847,6 +18850,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);
@@ -25059,8 +25063,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;
@@ -25071,20 +25075,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;
 }
@@ -26034,15 +26036,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();
@@ -27893,7 +27897,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 a8081b8d6e3..e854792b073 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
@@ -2469,6 +2469,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 dc8a8c53223..2ebb6222fb4 100644
--- a/sql/sql_show.cc
+++ b/sql/sql_show.cc
@@ -5157,6 +5157,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;
@@ -5174,7 +5175,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));
 
@@ -5350,6 +5373,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);
 }
 
@@ -9057,7 +9087,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),
@@ -9094,8 +9124,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()
 };
@@ -9103,10 +9133,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()
 };
 
@@ -9250,7 +9280,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),
@@ -9262,46 +9292,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()
 };
 
@@ -9442,7 +9472,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 44fa8709b1d..8fb3fae97ba 100644
--- a/sql/sql_table.cc
+++ b/sql/sql_table.cc
@@ -7943,6 +7943,8 @@ mysql_prepare_alter_table(THD *thd, TABLE *table,
 {
   /* 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);
@@ -8169,7 +8171,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
@@ -8220,6 +8222,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);
+    }
   }
 
   /*
@@ -8382,6 +8390,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),
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 adfe02e633b..8e150de7556 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),
diff --git a/sql/sql_window.cc b/sql/sql_window.cc
index 17920519b41..963535e2417 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 7e273fe9b54..13196a1ce8d 100644
--- a/sql/sql_yacc.yy
+++ b/sql/sql_yacc.yy
@@ -350,12 +350,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
@@ -383,6 +382,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 */
@@ -396,7 +396,6 @@ bool my_yyoverflow(short **a, YYSTYPE **b, size_t *yystacksize);
 %token  <NONE> WITH_ROLLUP_SYM        /* INTERNAL */
 %token  <NONE> WITH_SYSTEM_SYM        /* INTERNAL */
 
-
 /*
   Identifiers
 */
@@ -2703,6 +2702,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
           {
@@ -2904,7 +2906,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))
@@ -12912,6 +12914,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;
               }
               ;
@@ -17704,8 +17707,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;
@@ -17746,7 +17749,6 @@ sf_return_type:
           }
         ;
 
-
 /*************************************************************************/
 
 xa:
@@ -18084,7 +18086,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;
@@ -18105,7 +18107,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;
@@ -18956,8 +18958,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/table.cc b/sql/table.cc
index b9f0e7596aa..3bf85449f6c 100644
--- a/sql/table.cc
+++ b/sql/table.cc
@@ -50,25 +50,14 @@
 #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;
 }
 
@@ -3659,8 +3648,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 +3658,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 +3669,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 +3677,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 +3687,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 +4088,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 +6938,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");