Merge 10.3 into 10.4

17 files changed, 502 insertions, 75 deletions

diff --git a/sql/create_options.cc b/sql/create_options.cc
index 5adcb2f1e9e..a8d997efaf4 100644
--- a/sql/create_options.cc
+++ b/sql/create_options.cc
@@ -1,4 +1,4 @@
-/* Copyright (C) 2010, 2017, MariaDB Corporation Ab
+/* Copyright (C) 2010, 2019, 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
@@ -132,7 +132,8 @@ static bool set_one_value(ha_create_table_option *opt,
   switch (opt->type)
   {
   case HA_OPTION_TYPE_SYSVAR:
-    DBUG_ASSERT(0); // HA_OPTION_TYPE_SYSVAR's are replaced in resolve_sysvars()
+    // HA_OPTION_TYPE_SYSVAR's are replaced in resolve_sysvars()
+    break; // to DBUG_ASSERT(0)
   case HA_OPTION_TYPE_ULL:
     {
       ulonglong *val= (ulonglong*)value_ptr(base, opt);
diff --git a/sql/field.cc b/sql/field.cc
index 969c32a5180..750e5bce651 100644
--- a/sql/field.cc
+++ b/sql/field.cc
@@ -7315,6 +7315,28 @@ void Field_string::sql_type(String &res) const
     res.append(STRING_WITH_LEN(" binary"));
 }
 
+/**
+   For fields which are associated with character sets their length is provided
+   in octets and their character set information is also provided as part of
+   type information.
+
+   @param   res       String which contains filed type and length.
+*/
+void Field_string::sql_rpl_type(String *res) const
+{
+  CHARSET_INFO *cs=charset();
+  if (Field_string::has_charset())
+  {
+    size_t length= cs->cset->snprintf(cs, (char*) res->ptr(),
+                                      res->alloced_length(),
+                                      "char(%u octets) character set %s",
+                                      field_length,
+                                      charset()->csname);
+    res->length(length);
+  }
+  else
+    Field_string::sql_type(*res);
+ }
 
 uchar *Field_string::pack(uchar *to, const uchar *from, uint max_length)
 {
@@ -7735,6 +7757,29 @@ void Field_varstring::sql_type(String &res) const
     res.append(STRING_WITH_LEN(" binary"));
 }
 
+/**
+   For fields which are associated with character sets their length is provided
+   in octets and their character set information is also provided as part of
+   type information.
+
+   @param   res       String which contains filed type and length.
+*/
+void Field_varstring::sql_rpl_type(String *res) const
+{
+  CHARSET_INFO *cs=charset();
+  if (Field_varstring::has_charset())
+  {
+    size_t length= cs->cset->snprintf(cs, (char*) res->ptr(),
+                                      res->alloced_length(),
+                                      "varchar(%u octets) character set %s",
+                                      field_length,
+                                      charset()->csname);
+    res->length(length);
+  }
+  else
+    Field_varstring::sql_type(*res);
+}
+
 
 uint32 Field_varstring::data_length()
 {
diff --git a/sql/field.h b/sql/field.h
index 79bc55d2323..8bbcaed7e39 100644
--- a/sql/field.h
+++ b/sql/field.h
@@ -1,7 +1,7 @@
 #ifndef FIELD_INCLUDED
 #define FIELD_INCLUDED
 /* Copyright (c) 2000, 2015, Oracle and/or its affiliates.
-   Copyright (c) 2008, 2017, MariaDB Corporation.
+   Copyright (c) 2008, 2019, 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
@@ -540,6 +540,7 @@ public:
   /* Flag indicating  that the field is physically stored in the database */
   bool stored_in_db;
   bool utf8;                                    /* Already in utf8 */
+  bool automatic_name;
   Item *expr;
   Lex_ident name;                               /* Name of constraint */
   /* see VCOL_* (VCOL_FIELD_REF, ...) */
@@ -549,7 +550,7 @@ public:
    :Type_handler_hybrid_field_type(&type_handler_null),
     vcol_type((enum_vcol_info_type)VCOL_TYPE_NONE),
     in_partitioning_expr(FALSE), stored_in_db(FALSE),
-    utf8(TRUE), expr(NULL), flags(0)
+    utf8(TRUE), automatic_name(FALSE), expr(NULL), flags(0)
   {
     name.str= NULL;
     name.length= 0;
@@ -1148,6 +1149,7 @@ public:
     in str and restore it with set() if needed
   */
   virtual void sql_type(String &str) const =0;
+  virtual void sql_rpl_type(String *str) const { sql_type(*str); }
   virtual uint size_of() const =0;		// For new field
   inline bool is_null(my_ptrdiff_t row_offset= 0) const
   {
@@ -3576,6 +3578,7 @@ public:
   int cmp(const uchar *,const uchar *);
   void sort_string(uchar *buff,uint length);
   void sql_type(String &str) const;
+  void sql_rpl_type(String*) const;
   bool is_equal(const Column_definition &new_field) const;
   bool can_be_converted_by_engine(const Column_definition &new_type) const
   {
@@ -3692,6 +3695,7 @@ public:
   uint get_key_image(uchar *buff,uint length, imagetype type);
   void set_key_image(const uchar *buff,uint length);
   void sql_type(String &str) const;
+  void sql_rpl_type(String*) const;
   virtual uchar *pack(uchar *to, const uchar *from, uint max_length);
   virtual const uchar *unpack(uchar* to, const uchar *from,
                               const uchar *from_end, uint param_data);
diff --git a/sql/item.cc b/sql/item.cc
index eaec0a2a737..24747d1aae1 100644
--- a/sql/item.cc
+++ b/sql/item.cc
@@ -3768,6 +3768,20 @@ my_decimal *Item_null::val_decimal(my_decimal *decimal_value)
 }
 
 
+longlong Item_null::val_datetime_packed(THD *)
+{
+  null_value= true;
+  return 0;
+}
+
+
+longlong Item_null::val_time_packed(THD *)
+{
+  null_value= true;
+  return 0;
+}
+
+
 bool Item_null::get_date(THD *thd, MYSQL_TIME *ltime, date_mode_t fuzzydate)
 {
   set_zero_time(ltime, MYSQL_TIMESTAMP_NONE);
@@ -8212,6 +8226,24 @@ bool Item_ref::val_native(THD *thd, Native *to)
 }
 
 
+longlong Item_ref::val_datetime_packed(THD *thd)
+{
+  DBUG_ASSERT(fixed);
+  longlong tmp= (*ref)->val_datetime_packed(thd);
+  null_value= (*ref)->null_value;
+  return tmp;
+}
+
+
+longlong Item_ref::val_time_packed(THD *thd)
+{
+  DBUG_ASSERT(fixed);
+  longlong tmp= (*ref)->val_time_packed(thd);
+  null_value= (*ref)->null_value;
+  return tmp;
+}
+
+
 my_decimal *Item_ref::val_decimal(my_decimal *decimal_value)
 {
   my_decimal *val= (*ref)->val_decimal_result(decimal_value);
diff --git a/sql/item.h b/sql/item.h
index 5cda78d42fa..75c0fbc67fe 100644
--- a/sql/item.h
+++ b/sql/item.h
@@ -3557,6 +3557,8 @@ public:
   String *val_str(String *str);
   my_decimal *val_decimal(my_decimal *);
   bool get_date(THD *thd, MYSQL_TIME *ltime, date_mode_t fuzzydate);
+  longlong val_datetime_packed(THD *);
+  longlong val_time_packed(THD *);
   int save_in_field(Field *field, bool no_conversions);
   int save_safe_in_field(Field *field);
   bool send(Protocol *protocol, st_value *buffer);
@@ -5179,6 +5181,8 @@ public:
   bool val_native(THD *thd, Native *to);
   bool is_null();
   bool get_date(THD *thd, MYSQL_TIME *ltime, date_mode_t fuzzydate);
+  longlong val_datetime_packed(THD *);
+  longlong val_time_packed(THD *);
   double val_result();
   longlong val_int_result();
   String *str_result(String* tmp);
diff --git a/sql/mysqld.cc b/sql/mysqld.cc
index fc7e82936ea..a4118ced910 100644
--- a/sql/mysqld.cc
+++ b/sql/mysqld.cc
@@ -6719,6 +6719,10 @@ struct my_option my_long_options[]=
    0, GET_INT, REQUIRED_ARG, 0, 0, 0, 0, 0, 0},
 #endif /* HAVE_REPLICATION */
 #ifndef DBUG_OFF
+  {"debug-assert", 0,
+   "Allow DBUG_ASSERT() to invoke assert()",
+   &my_assert, &my_assert,
+   0, GET_BOOL, OPT_ARG, 1, 0, 0, 0, 0, 0},
   {"debug-assert-on-error", 0,
    "Do an assert in various functions if we get a fatal error",
    &my_assert_on_error, &my_assert_on_error,
diff --git a/sql/opt_subselect.cc b/sql/opt_subselect.cc
index 9205380bd19..a10c8de417f 100644
--- a/sql/opt_subselect.cc
+++ b/sql/opt_subselect.cc
@@ -2460,7 +2460,7 @@ bool optimize_semijoin_nests(JOIN *join, table_map all_table_map)
                                          &subjoin_out_rows);
 
         sjm->materialization_cost.convert_from_cost(subjoin_read_time);
-        sjm->rows= subjoin_out_rows;
+        sjm->rows_with_duplicates= sjm->rows= subjoin_out_rows;
         
         // Don't use the following list because it has "stale" items. use
         // ref_pointer_array instead:
@@ -3123,11 +3123,14 @@ bool Sj_materialization_picker::check_qep(JOIN *join,
                        disable_jbuf, prefix_rec_count, &curpos, &dummy);
       prefix_rec_count= COST_MULT(prefix_rec_count, curpos.records_read);
       prefix_cost= COST_ADD(prefix_cost, curpos.read_time);
+      prefix_cost= COST_ADD(prefix_cost,
+                            prefix_rec_count / (double) TIME_FOR_COMPARE);
+      //TODO: take into account join condition selectivity here
     }
 
     *strategy= SJ_OPT_MATERIALIZE_SCAN;
     *read_time=    prefix_cost;
-    *record_count= prefix_rec_count;
+    *record_count= prefix_rec_count / mat_info->rows_with_duplicates;
     *handled_fanout= mat_nest->sj_inner_tables;
     if (unlikely(join->thd->trace_started()))
     {
diff --git a/sql/rpl_utility.cc b/sql/rpl_utility.cc
index b36838b9b22..437d58d772f 100644
--- a/sql/rpl_utility.cc
+++ b/sql/rpl_utility.cc
@@ -352,7 +352,8 @@ uint32 table_def::calc_field_size(uint col, uchar *master_data) const
 #if !defined(MYSQL_CLIENT) && defined(HAVE_REPLICATION)
 /**
  */
-void show_sql_type(enum_field_types type, uint16 metadata, String *str, CHARSET_INFO *field_cs)
+void show_sql_type(enum_field_types type, uint16 metadata, String *str,
+                   bool char_with_octets)
 {
   DBUG_ENTER("show_sql_type");
   DBUG_PRINT("enter", ("type: %d, metadata: 0x%x", type, metadata));
@@ -420,11 +421,13 @@ void show_sql_type(enum_field_types type, uint16 metadata, String *str, CHARSET_
   case MYSQL_TYPE_VARCHAR_COMPRESSED:
     {
       CHARSET_INFO *cs= str->charset();
-      size_t length=
-        cs->cset->snprintf(cs, (char*) str->ptr(), str->alloced_length(),
-                           "varchar(%u)%s", metadata,
-                           type == MYSQL_TYPE_VARCHAR_COMPRESSED ? " compressed"
-                                                                 : "");
+      size_t length=0;
+      if (char_with_octets)
+        length= cs->cset->snprintf(cs, (char*) str->ptr(), str->alloced_length(),
+                                   "varchar(%u octets)", metadata);
+      else
+        length= cs->cset->snprintf(cs, (char*) str->ptr(), str->alloced_length(),
+                                   "varbinary(%u)", metadata);
       str->length(length);
     }
     break;
@@ -475,22 +478,22 @@ void show_sql_type(enum_field_types type, uint16 metadata, String *str, CHARSET_
       it is necessary to check the pack length to figure out what kind
       of blob it really is.
      */
-    switch (get_blob_type_from_length(metadata))
+    switch (metadata)
     {
-    case MYSQL_TYPE_TINY_BLOB:
+    case 1:
       str->set_ascii(STRING_WITH_LEN("tinyblob"));
       break;
 
-    case MYSQL_TYPE_MEDIUM_BLOB:
-      str->set_ascii(STRING_WITH_LEN("mediumblob"));
+    case 2:
+      str->set_ascii(STRING_WITH_LEN("blob"));
       break;
 
-    case MYSQL_TYPE_LONG_BLOB:
-      str->set_ascii(STRING_WITH_LEN("longblob"));
+    case 3:
+      str->set_ascii(STRING_WITH_LEN("mediumblob"));
       break;
 
-    case MYSQL_TYPE_BLOB:
-      str->set_ascii(STRING_WITH_LEN("blob"));
+    case 4:
+      str->set_ascii(STRING_WITH_LEN("longblob"));
       break;
 
     default:
@@ -509,9 +512,13 @@ void show_sql_type(enum_field_types type, uint16 metadata, String *str, CHARSET_
       */
       CHARSET_INFO *cs= str->charset();
       uint bytes= (((metadata >> 4) & 0x300) ^ 0x300) + (metadata & 0x00ff);
-      size_t length=
-        cs->cset->snprintf(cs, (char*) str->ptr(), str->alloced_length(),
-                           "char(%d)", bytes / field_cs->mbmaxlen);
+      size_t length=0;
+      if (char_with_octets)
+        length= cs->cset->snprintf(cs, (char*) str->ptr(), str->alloced_length(),
+                                   "char(%u octets)", bytes);
+      else
+        length= cs->cset->snprintf(cs, (char*) str->ptr(), str->alloced_length(),
+                                   "binary(%u)", bytes);
       str->length(length);
     }
     break;
@@ -948,9 +955,13 @@ table_def::compatible_with(THD *thd, rpl_group_info *rgi,
       String source_type(source_buf, sizeof(source_buf), &my_charset_latin1);
       String target_type(target_buf, sizeof(target_buf), &my_charset_latin1);
       THD *thd= table->in_use;
+      bool char_with_octets= field->cmp_type() == STRING_RESULT ?
+                             field->has_charset() : true;
+
+      show_sql_type(type(col), field_metadata(col), &source_type,
+                    char_with_octets);
+      field->sql_rpl_type(&target_type);
 
-      show_sql_type(type(col), field_metadata(col), &source_type, field->charset());
-      field->sql_type(target_type);
       rli->report(ERROR_LEVEL, ER_SLAVE_CONVERSION_FAILED, rgi->gtid_info(),
                   ER_THD(thd, ER_SLAVE_CONVERSION_FAILED),
                   col, db_name, tbl_name,
diff --git a/sql/share/errmsg-utf8.txt b/sql/share/errmsg-utf8.txt
index c64f60f3562..c91132c1eae 100644
--- a/sql/share/errmsg-utf8.txt
+++ b/sql/share/errmsg-utf8.txt
@@ -6461,7 +6461,7 @@ ER_MESSAGE_AND_STATEMENT
   eng "%s Statement: %s"
 
 ER_SLAVE_CONVERSION_FAILED
-  eng "Column %d of table '%-.192s.%-.192s' cannot be converted from type '%-.32s' to type '%-.32s'"
+  eng "Column %d of table '%-.192s.%-.192s' cannot be converted from type '%-.50s' to type '%-.50s'"
 ER_SLAVE_CANT_CREATE_CONVERSION
   eng "Can't create conversion table for table '%-.192s.%-.192s'"
 ER_INSIDE_TRANSACTION_PREVENTS_SWITCH_BINLOG_FORMAT
diff --git a/sql/sql_class.h b/sql/sql_class.h
index 2b46fc69365..5ffc8bde269 100644
--- a/sql/sql_class.h
+++ b/sql/sql_class.h
@@ -5995,7 +5995,10 @@ public:
 
   uint tables; /* Number of tables in the sj-nest */
 
-  /* Expected #rows in the materialized table */
+  /* Number of rows in the materialized table, before the de-duplication */
+  double rows_with_duplicates;
+
+  /* Expected #rows in the materialized table, after de-duplication */
   double rows;
 
   /* 
diff --git a/sql/sql_lex.cc b/sql/sql_lex.cc
index a1ee99f29ec..21473c28b84 100644
--- a/sql/sql_lex.cc
+++ b/sql/sql_lex.cc
@@ -2993,6 +2993,7 @@ void st_select_lex_unit::print(String *str, enum_query_type query_type)
       {
       default:
         DBUG_ASSERT(0);
+        /* fall through */
       case UNION_TYPE:
         str->append(STRING_WITH_LEN(" union "));
         if (union_all)
diff --git a/sql/sql_parse.cc b/sql/sql_parse.cc
index 1f612ff2266..ab915bf3fcc 100644
--- a/sql/sql_parse.cc
+++ b/sql/sql_parse.cc
@@ -3623,6 +3623,7 @@ mysql_execute_command(THD *thd)
         case GET_NO_ARG:
         case GET_DISABLED:
           DBUG_ASSERT(0);
+          /* fall through */
         case 0:
         case GET_FLAGSET:
         case GET_ENUM:
diff --git a/sql/sql_select.cc b/sql/sql_select.cc
index 150e8008096..bd6d9852f40 100644
--- a/sql/sql_select.cc
+++ b/sql/sql_select.cc
@@ -8788,6 +8788,7 @@ void JOIN::get_prefix_cost_and_fanout(uint n_tables,
       record_count= COST_MULT(record_count, best_positions[i].records_read);
       read_time= COST_ADD(read_time, best_positions[i].read_time);
     }
+    /* TODO: Take into account condition selectivities here */
   }
   *read_time_arg= read_time;// + record_count / TIME_FOR_COMPARE;
   *record_count_arg= record_count;
@@ -16678,10 +16679,20 @@ void optimize_wo_join_buffering(JOIN *join, uint first_tab, uint last_tab,
     reopt_remaining_tables &= ~rs->table->map;
     rec_count= COST_MULT(rec_count, pos.records_read);
     cost= COST_ADD(cost, pos.read_time);
-
-
+    cost= COST_ADD(cost, rec_count / (double) TIME_FOR_COMPARE);
+    //TODO: take into account join condition selectivity here
+    double pushdown_cond_selectivity= 1.0;
+    table_map real_table_bit= rs->table->map;
+    if (join->thd->variables.optimizer_use_condition_selectivity > 1)
+    {
+      pushdown_cond_selectivity= table_cond_selectivity(join, i, rs,
+                                                        reopt_remaining_tables &
+                                                        ~real_table_bit);
+    }
+    (*outer_rec_count) *= pushdown_cond_selectivity;
     if (!rs->emb_sj_nest)
       *outer_rec_count= COST_MULT(*outer_rec_count, pos.records_read);
+
   }
   join->cur_sj_inner_tables= save_cur_sj_inner_tables;
 
diff --git a/sql/sql_show.cc b/sql/sql_show.cc
index 2b9a5a6ea29..e94bafedb83 100644
--- a/sql/sql_show.cc
+++ b/sql/sql_show.cc
@@ -2593,6 +2593,7 @@ static const LEX_CSTRING *view_algorithm(TABLE_LIST *table)
     return &merge;
   default:
     DBUG_ASSERT(0); // never should happen
+    /* fall through */
   case VIEW_ALGORITHM_UNDEFINED:
     return &undefined;
   }
diff --git a/sql/sql_table.cc b/sql/sql_table.cc
index 36b2174c8fd..d6d6c083432 100644
--- a/sql/sql_table.cc
+++ b/sql/sql_table.cc
@@ -64,7 +64,7 @@ const char *primary_key_name="PRIMARY";
 
 static int check_if_keyname_exists(const char *name,KEY *start, KEY *end);
 static char *make_unique_key_name(THD *, const char *, KEY *, KEY *);
-static void make_unique_constraint_name(THD *, LEX_CSTRING *, const char *,
+static bool make_unique_constraint_name(THD *, LEX_CSTRING *, const char *,
                                         List<Virtual_column_info> *, uint *);
 static const char *make_unique_invisible_field_name(THD *, const char *,
                                                     List<Create_field> *);
@@ -76,6 +76,9 @@ static int copy_data_between_tables(THD *, TABLE *,TABLE *,
 static int mysql_prepare_create_table(THD *, HA_CREATE_INFO *, Alter_info *,
                                       uint *, handler *, KEY **, uint *, int);
 static uint blob_length_by_type(enum_field_types type);
+static bool fix_constraints_names(THD *thd, List<Virtual_column_info>
+                                  *check_constraint_list,
+                                  const HA_CREATE_INFO *create_info);
 
 /**
   @brief Helper function for explain_filename
@@ -4311,20 +4314,13 @@ mysql_prepare_create_table(THD *thd, HA_CREATE_INFO *create_info,
   /* Check table level constraints */
   create_info->check_constraint_list= &alter_info->check_constraint_list;
   {
-    uint nr= 1;
     List_iterator_fast<Virtual_column_info> c_it(alter_info->check_constraint_list);
     Virtual_column_info *check;
     while ((check= c_it++))
     {
-      if (!check->name.length)
-      {
-        const char *own_name_base= create_info->period_info.constr == check
-                                   ? create_info->period_info.name.str : NULL;
+      if (!check->name.length || check->automatic_name)
+        continue;
 
-        make_unique_constraint_name(thd, &check->name, own_name_base,
-                                    &alter_info->check_constraint_list,
-                                    &nr);
-      }
       {
         /* Check that there's no repeating constraint names. */
         List_iterator_fast<Virtual_column_info>
@@ -4869,6 +4865,10 @@ int create_table_impl(THD *thd, const LEX_CSTRING &orig_db,
   DBUG_PRINT("enter", ("db: '%s'  table: '%s'  tmp: %d  path: %s",
                        db.str, table_name.str, internal_tmp_table, path));
 
+  if (fix_constraints_names(thd, &alter_info->check_constraint_list,
+                            create_info))
+    DBUG_RETURN(1);
+
   if (thd->variables.sql_mode & MODE_NO_DIR_IN_CREATE)
   {
     if (create_info->data_file_name)
@@ -5367,7 +5367,7 @@ make_unique_key_name(THD *thd, const char *field_name,KEY *start,KEY *end)
    Make an unique name for constraints without a name
 */
 
-static void make_unique_constraint_name(THD *thd, LEX_CSTRING *name,
+static bool make_unique_constraint_name(THD *thd, LEX_CSTRING *name,
                                         const char *own_name_base,
                                         List<Virtual_column_info> *vcol,
                                         uint *nr)
@@ -5395,9 +5395,10 @@ static void make_unique_constraint_name(THD *thd, LEX_CSTRING *name,
     {
       name->length= (size_t) (real_end - buff);
       name->str= thd->strmake(buff, name->length);
-      return;
+      return (name->str == NULL);
     }
   }
+  return FALSE;
 }
 
 /**
@@ -6020,10 +6021,11 @@ static bool is_candidate_key(KEY *key)
      from the list if existing found.
 
    RETURN VALUES
-     NONE
+     TRUE error
+     FALSE OK
 */
 
-static void
+static bool
 handle_if_exists_options(THD *thd, TABLE *table, Alter_info *alter_info,
                          Table_period_info *period_info)
 {
@@ -6469,6 +6471,7 @@ remove_key:
     Virtual_column_info *check;
     TABLE_SHARE *share= table->s;
     uint c;
+
     while ((check=it++))
     {
       if (!(check->flags & Alter_info::CHECK_CONSTRAINT_IF_NOT_EXISTS) &&
@@ -6516,7 +6519,48 @@ remove_key:
     *period_info= {};
   }
 
-  DBUG_VOID_RETURN;
+  DBUG_RETURN(false);
+}
+
+
+static bool fix_constraints_names(THD *thd, List<Virtual_column_info>
+                                  *check_constraint_list,
+                                  const HA_CREATE_INFO *create_info)
+{
+  List_iterator<Virtual_column_info> it((*check_constraint_list));
+  Virtual_column_info *check;
+  uint nr= 1;
+  DBUG_ENTER("fix_constraints_names");
+  if (!check_constraint_list)
+    DBUG_RETURN(FALSE);
+  // Prevent accessing freed memory during generating unique names
+  while ((check=it++))
+  {
+    if (check->automatic_name)
+    {
+      check->name.str= NULL;
+      check->name.length= 0;
+    }
+  }
+  it.rewind();
+  // Generate unique names if needed
+  while ((check=it++))
+  {
+    if (!check->name.length)
+    {
+      check->automatic_name= TRUE;
+
+      const char *own_name_base= create_info->period_info.constr == check
+        ? create_info->period_info.name.str : NULL;
+
+      if (make_unique_constraint_name(thd, &check->name,
+                                      own_name_base,
+                                      check_constraint_list,
+                                      &nr))
+        DBUG_RETURN(TRUE);
+    }
+  }
+  DBUG_RETURN(FALSE);
 }
 
 
@@ -9657,7 +9701,11 @@ do_continue:;
     }
   }
 
-  handle_if_exists_options(thd, table, alter_info, &create_info->period_info);
+  if (handle_if_exists_options(thd, table, alter_info,
+                               &create_info->period_info) ||
+      fix_constraints_names(thd, &alter_info->check_constraint_list,
+                            create_info))
+    DBUG_RETURN(true);
 
   /*
     Look if we have to do anything at all.
diff --git a/sql/table.cc b/sql/table.cc
index da0c4be41b0..4f6399d84ed 100644
--- a/sql/table.cc
+++ b/sql/table.cc
@@ -9444,8 +9444,7 @@ bool vers_select_conds_t::eq(const vers_select_conds_t &conds) const
   case SYSTEM_TIME_ALL:
     return true;
   case SYSTEM_TIME_BEFORE:
-    DBUG_ASSERT(0);
-    return false;
+    break;
   case SYSTEM_TIME_AS_OF:
     return start.eq(conds.start);
   case SYSTEM_TIME_FROM_TO:
diff --git a/sql/wsrep_sst.cc b/sql/wsrep_sst.cc
index 62b30c0d67d..74a8b9dff05 100644
--- a/sql/wsrep_sst.cc
+++ b/sql/wsrep_sst.cc
@@ -16,6 +16,7 @@
 #include "mariadb.h"
 #include "wsrep_sst.h"
 #include <inttypes.h>
+#include <ctype.h>
 #include <mysqld.h>
 #include <m_ctype.h>
 #include <strfunc.h>
@@ -309,7 +310,31 @@ static char* my_fgets (char* buf, size_t buf_len, FILE* stream)
 }
 
 /*
-  Generate opt_binlog_opt_val for sst_donate_other(), sst_prepare_other().
+  Generate "name 'value'" string.
+*/
+static char* generate_name_value(const char* name, const char* value)
+{
+  size_t name_len= strlen(name);
+  size_t value_len= strlen(value);
+  char* buf=
+    (char*) my_malloc((name_len + value_len + 5) * sizeof(char), MYF(0));
+  if (buf)
+  {
+    char* ref= buf;
+    *ref++ = ' ';
+    memcpy(ref, name, name_len * sizeof(char));
+    ref += name_len;
+    *ref++ = ' ';
+    *ref++ = '\'';
+    memcpy(ref, value, value_len * sizeof(char));
+    ref += value_len;
+    *ref++ = '\'';
+    *ref = 0;
+  }
+  return buf;
+}
+/*
+  Generate binlog option string for sst_donate_other(), sst_prepare_other().
 
   Returns zero on success, negative error code otherwise.
 
@@ -325,7 +350,9 @@ static int generate_binlog_opt_val(char** ret)
   {
     assert(opt_bin_logname);
     *ret= strcmp(opt_bin_logname, "0") ?
-      my_strdup(opt_bin_logname, MYF(0)) : my_strdup("", MYF(0));
+      generate_name_value(WSREP_SST_OPT_BINLOG,
+                          opt_bin_logname) :
+      my_strdup("", MYF(0));
   }
   else
   {
@@ -342,7 +369,9 @@ static int generate_binlog_index_opt_val(char** ret)
   if (opt_binlog_index_name)
   {
     *ret= strcmp(opt_binlog_index_name, "0") ?
-      my_strdup(opt_binlog_index_name, MYF(0)) : my_strdup("", MYF(0));
+      generate_name_value(WSREP_SST_OPT_BINLOG_INDEX,
+                          opt_binlog_index_name) :
+      my_strdup("", MYF(0));
   }
   else
   {
@@ -595,7 +624,86 @@ static size_t estimate_cmd_len (bool* extra_args)
   {
     for (int i = 1; i < orig_argc; i++)
     {
-      cmd_len += strlen(orig_argv[i]);
+      const char* arg= orig_argv[i];
+      size_t n= strlen(arg);
+      if (n == 0) continue;
+      cmd_len += n;
+      bool quotation= false;
+      char c;
+      while ((c = *arg++) != 0)
+      {
+        /* A whitespace or a single quote requires double quotation marks: */
+        if (isspace(c) || c == '\'')
+        {
+          quotation= true;
+        }
+        /*
+          If the equals symbol is encountered, then we need to separately
+          process the right side:
+        */
+        else if (c == '=')
+        {
+          /* Perhaps we need to quote the left part of the argument: */
+          if (quotation)
+          {
+            cmd_len += 2;
+            /*
+              Reset the quotation flag, since now the status for
+              the right side of the expression will be saved here:
+            */
+            quotation= false;
+          }
+          while ((c = *arg++) != 0)
+          {
+            /*
+              A whitespace or a single quote requires double
+              quotation marks:
+            */
+            if (isspace(c) || c == '\'')
+            {
+              quotation= true;
+            }
+            /*
+              Double quotation mark or backslash symbol requires backslash
+              prefixing:
+            */
+#ifdef __WIN__
+            else if (c == '"' || c == '\\')
+#else
+            /*
+              The dollar symbol is used to substitute a variable, therefore
+              it also requires escaping:
+            */
+            else if (c == '"' || c == '\\' || c == '$')
+#endif
+            {
+              cmd_len++;
+            }
+          }
+          break;
+        }
+        /*
+          Double quotation mark or backslash symbol requires backslash
+          prefixing:
+        */
+#ifdef __WIN__
+        else if (c == '"' || c == '\\')
+#else
+        /*
+          The dollar symbol is used to substitute a variable, therefore
+          it also requires escaping:
+        */
+        else if (c == '"' || c == '\\' || c == '$')
+#endif
+        {
+          cmd_len++;
+        }
+      }
+      /* Perhaps we need to quote the entire argument or its right part: */
+      if (quotation)
+      {
+        cmd_len += 2;
+      }
     }
     extra = true;
     cmd_len += strlen(WSREP_SST_OPT_MYSQLD);
@@ -623,10 +731,171 @@ static void copy_orig_argv (char* cmd_str)
     for (int i = 1; i < orig_argc; i++)
     {
       char* arg= orig_argv[i];
-      *cmd_str++ = ' ';
       n = strlen(arg);
-      memcpy(cmd_str, arg, n * sizeof(char));
-      cmd_str += n;
+      if (n == 0) continue;
+      *cmd_str++ = ' ';
+      bool quotation= false;
+      bool plain= true;
+      char *arg_scan= arg;
+      char c;
+      while ((c = *arg_scan++) != 0)
+      {
+        /* A whitespace or a single quote requires double quotation marks: */
+        if (isspace(c) || c == '\'')
+        {
+          quotation= true;
+        }
+        /*
+          If the equals symbol is encountered, then we need to separately
+          process the right side:
+        */
+        else if (c == '=')
+        {
+          /* Calculate length of the Left part of the argument: */
+          size_t m = (size_t) (arg_scan - arg) - 1;
+          if (m)
+          {
+            /* Perhaps we need to quote the left part of the argument: */
+            if (quotation)
+            {
+              *cmd_str++ = '"';
+            }
+            /*
+              If there were special characters inside, then we can use
+              the fast memcpy function:
+            */
+            if (plain)
+            {
+              memcpy(cmd_str, arg, m * sizeof(char));
+              cmd_str += m;
+              /* Left part of the argument has already been processed: */
+              n -= m;
+              arg += m;
+            }
+            /* Otherwise we need to prefix individual characters: */
+            else
+            {
+              n -= m;
+              while (m)
+              {
+                c = *arg++;
+#ifdef __WIN__
+                if (c == '"' || c == '\\')
+#else
+                if (c == '"' || c == '\\' || c == '$')
+#endif
+                {
+                  *cmd_str++ = '\\';
+                }
+                *cmd_str++ = c;
+                m--;
+              }
+              /*
+                Reset the plain string flag, since now the status for
+                the right side of the expression will be saved here:
+              */
+              plain= true;
+            }
+            /* Perhaps we need to quote the left part of the argument: */
+            if (quotation)
+            {
+              *cmd_str++ = '"';
+              /*
+                Reset the quotation flag, since now the status for
+                the right side of the expression will be saved here:
+              */
+              quotation= false;
+            }
+          }
+          /* Copy equals symbol: */
+          *cmd_str++ = '=';
+          arg++;
+          n--;
+          /* Let's deal with the left side of the expression: */
+          while ((c = *arg_scan++) != 0)
+          {
+            /*
+              A whitespace or a single quote requires double
+              quotation marks:
+            */
+            if (isspace(c) || c == '\'')
+            {
+              quotation= true;
+            }
+            /*
+              Double quotation mark or backslash symbol requires backslash
+              prefixing:
+            */
+#ifdef __WIN__
+            else if (c == '"' || c == '\\')
+#else
+            /*
+              The dollar symbol is used to substitute a variable, therefore
+              it also requires escaping:
+            */
+            else if (c == '"' || c == '\\' || c == '$')
+#endif
+            {
+              plain= false;
+            }
+          }
+          break;
+        }
+        /*
+          Double quotation mark or backslash symbol requires backslash
+          prefixing:
+        */
+#ifdef __WIN__
+        else if (c == '"' || c == '\\')
+#else
+        /*
+          The dollar symbol is used to substitute a variable, therefore
+          it also requires escaping:
+        */
+        else if (c == '"' || c == '\\' || c == '$')
+#endif
+        {
+          plain= false;
+        }
+      }
+      if (n)
+      {
+        /* Perhaps we need to quote the entire argument or its right part: */
+        if (quotation)
+        {
+          *cmd_str++ = '"';
+        }
+        /*
+          If there were no special characters inside, then we can use
+          the fast memcpy function:
+        */
+        if (plain)
+        {
+          memcpy(cmd_str, arg, n * sizeof(char));
+          cmd_str += n;
+        }
+        /* Otherwise we need to prefix individual characters: */
+        else
+        {
+          while ((c = *arg++) != 0)
+          {
+#ifdef __WIN__
+            if (c == '"' || c == '\\')
+#else
+            if (c == '"' || c == '\\' || c == '$')
+#endif
+            {
+              *cmd_str++ = '\\';
+            }
+            *cmd_str++ = c;
+          }
+        }
+        /* Perhaps we need to quote the entire argument or its right part: */
+        if (quotation)
+        {
+          *cmd_str++ = '"';
+        }
+      }
     }
     /*
       Add a terminating null character (not counted in the length,
@@ -653,8 +922,6 @@ static ssize_t sst_prepare_other (const char*  method,
     return -ENOMEM;
   }
 
-  const char* binlog_opt= "";
-  const char* binlog_index_opt= "";
   char* binlog_opt_val= NULL;
   char* binlog_index_opt_val= NULL;
 
@@ -672,9 +939,6 @@ static ssize_t sst_prepare_other (const char*  method,
                 ret);
   }
 
-  if (strlen(binlog_opt_val)) binlog_opt= WSREP_SST_OPT_BINLOG;
-  if (strlen(binlog_index_opt_val)) binlog_index_opt= WSREP_SST_OPT_BINLOG_INDEX;
-
   make_wsrep_defaults_file();
 
   ret= snprintf (cmd_str(), cmd_len,
@@ -682,14 +946,14 @@ static ssize_t sst_prepare_other (const char*  method,
                  WSREP_SST_OPT_ROLE " 'joiner' "
                  WSREP_SST_OPT_ADDR " '%s' "
                  WSREP_SST_OPT_DATA " '%s' "
-                 " %s "
+                 "%s"
                  WSREP_SST_OPT_PARENT " '%d'"
-                 " %s '%s'"
-                 " %s '%s'",
+                 "%s"
+                 "%s",
                  method, addr_in, mysql_real_data_home,
                  wsrep_defaults_file,
-                 (int)getpid(), binlog_opt, binlog_opt_val,
-                 binlog_index_opt, binlog_index_opt_val);
+                 (int)getpid(),
+                 binlog_opt_val, binlog_index_opt_val);
   my_free(binlog_opt_val);
   my_free(binlog_index_opt_val);
 
@@ -983,7 +1247,7 @@ static int sst_donate_mysqldump (const char*         addr,
                      WSREP_SST_OPT_PORT " '%u' "
                      WSREP_SST_OPT_LPORT " '%u' "
                      WSREP_SST_OPT_SOCKET " '%s' "
-                     " %s "
+                     "%s"
                      WSREP_SST_OPT_GTID " '%s:%lld' "
                      WSREP_SST_OPT_GTID_DOMAIN_ID " '%d'"
                      "%s",
@@ -1363,8 +1627,6 @@ static int sst_donate_other (const char*        method,
     return -ENOMEM;
   }
 
-  const char* binlog_opt= "";
-  const char* binlog_index_opt= "";
   char* binlog_opt_val= NULL;
   char* binlog_index_opt_val= NULL;
 
@@ -1381,9 +1643,6 @@ static int sst_donate_other (const char*        method,
                 ret);
   }
 
-  if (strlen(binlog_opt_val)) binlog_opt= WSREP_SST_OPT_BINLOG;
-  if (strlen(binlog_index_opt_val)) binlog_index_opt= WSREP_SST_OPT_BINLOG_INDEX;
-
   make_wsrep_defaults_file();
 
   std::ostringstream uuid_oss;
@@ -1394,18 +1653,18 @@ static int sst_donate_other (const char*        method,
                  WSREP_SST_OPT_ADDR " '%s' "
                  WSREP_SST_OPT_SOCKET " '%s' "
                  WSREP_SST_OPT_DATA " '%s' "
-                 " %s "
-                 " %s '%s' "
-                 " %s '%s' "
+                 "%s"
                  WSREP_SST_OPT_GTID " '%s:%lld' "
                  WSREP_SST_OPT_GTID_DOMAIN_ID " '%d'"
+                 "%s"
+                 "%s"
                  "%s",
                  method, addr, mysqld_unix_port, mysql_real_data_home,
                  wsrep_defaults_file,
-                 binlog_opt, binlog_opt_val,
-                 binlog_index_opt, binlog_index_opt_val,
                  uuid_oss.str().c_str(), gtid.seqno().get(), wsrep_gtid_domain_id,
+                 binlog_opt_val, binlog_index_opt_val,
                  bypass ? " " WSREP_SST_OPT_BYPASS : "");
+
   my_free(binlog_opt_val);
   my_free(binlog_index_opt_val);