MDEV-15107 Add virtual Field::sp_prepare_and_store_item(), make sp_rcontext symmetric for scalar and ROW

After MDEV-14212, the Virtual_tmp_table instance that stores a ROW
variable elements is accessible from the underlying Field_row
(rather than Item_field_row).

This patch makes some further changes by moving the code from
sp_instr_xxx, sp_rcontext, Item_xxx to Virtual_tmp_table and Field_xxx.

The data type specific code (scalar vs ROW) now resides in
a new virtual method Field_xxx::sp_prepare_and_store_item().
The the code in sp_rcontext::set_variable() and sp_eval_expr()
is now symmetric for scalar and ROW values.
The code in sp_rcontext::set_variable_row_field(), sp_rcontext::set_variable_row_field(), sp_rcontext::set_variable_row()
is now symmetric for ROW elements (i.e. scalar and ROW elements inside a ROW).

Rationale:

Prepare the code to implement these tasks soon easier:

- MDEV-12252 ROW data type for stored function return values
- MDEV-12307 ROW data type for built-in function return values
- MDEV-6121 Data type: Array
- MDEV-10593 sql_mode=ORACLE: TYPE .. AS OBJECT: basic functionality
- ROW with ROW fields (no MDEV yet)

Details:

1. Moving the code in sp_eval_expr() responsible to backup/restore
   thd->count_cuted_fields, thd->abort_on_warning,
   thd->transaction.stmt.modified_non_trans_table
   into a new helper class Sp_eval_expr_state, to reuse it easier.
   Fixing sp_eval_expr() to use this new class.

2. Moving sp_eval_expr() and sp_prepare_func_item() from public functions
   to methods in THD, so they can be reused in *.cc files easier without
   a need to include "sp_head.h".

   Splitting sp_prepare_func_item() into two parts.
   Adding a new function sp_fix_func_item(), which fixes
   the underlying items, but does not do check_cols() for them.
   Reusing sp_fix_func_item() in Field_row::sp_prepare_and_store_item().

3. Moving the code to find ROW fields by name from Item to Virtual_tmp_table

   Moving the code searching for ROW fields by their names
   from Item_field_row::element_index_by_name() to a new method
   Item_field_row to Virtual_tmp_table::sp_find_field_by_name().

   Adding wrapper methods sp_rcontext::find_row_field_by_name() and
   find_row_field_by_name_or_error(), to search for a ROW variable
   fields by the variable offset and its field name.

   Changing Item_splocal_row_field_by_name::fix_fields() to do
   use sp_rcontext::find_row_field_by_name_or_error().
   Removing virtual Item::element_index_by_name().

4. Splitting sp_rcontext::set_variable()

   Adding a new virtual method Field::sp_prepare_and_store_item().
   Spliting the two branches of the code in sp_rcontext::set_variable()
   into two virtual implementations of Field::sp_prepare_and_store_item(),
   (for Field and for Field_row).

   Moving the former part of sp_rcontext::set_variable() with the loop
   doing set_null() for all ROW fields into a new method
   Virtual_tmp_table::set_all_fields_to_null() and using it in
   Field_row::sp_prepare_and_store_item().

   Moving the former part of sp_rcontext::set_variable() with the loop
   doing set_variable_row_field() into a new method
   Virtual_tmp_table::set_all_fields_from_item() and using it in
   Field_row::sp_prepare_and_store_item().
   The loop in the new method now uses sp_prepare_and_store_item()
   instead of set_variable_row_field(), because saving/restoring
   THD flags is now done on the upper level. No needs to save/restore
   on every iteration.

5. Fixing sp_eval_expr() to simply do two things:
   - backup/restore THD flags
   - call result_field->sp_prepare_and_store_item()
   So now sp_eval_expr() can be used for both scalar and ROW variables.
   Reusing it in sp_rcontext::set_variable*().

6. Moving the loop in sp_rcontext::set_variable_row() into a
   new method Virtual_tmp_table::sp_set_all_fields_from_item_list().

   Changing the loop body to call field->sp_prepare_and_store_item()
   instead of doing set_variable_row_field(). This removes
   saving/restoring of the THD flags from every interation.
   Instead, adding the code to save/restore the flags around
   the entire loop in set_variable_row(), using Sp_eval_expr_state.
   So now saving/restoring is done only once for the entire ROW
   (a slight performance improvement).

7. Removing the code in sp_instr_set::exec_core() that sets
   a variable to NULL if the value evaluation failed.
   sp_rcontext::set_variable() now makes sure to reset
   the variable properly by effectively calling sp_eval_expr(),
   which calls virtual Field::sp_prepare_and_store_item().

   Removing the similar code from sp_instr_set_row_field::exec_core()
   and sp_instr_set_row_field_by_name::exec_core().

   Removing the method sp_rcontext::set_variable_row_field_to_null(),
   as it's not used any more.

8. Removing the call for sp_prepare_func_item() from
   sp_rcontext::set_variable_row_field(), as it was duplicate:
   it was done inside sp_eval_expr(). Now it's done inside
   virtual Field::sp_prepare_and_store_item().

9. Moving the code from sp_instr_set_row_field_by_name::exec_core()
   into sp_rcontext::set_variable_row_field_by_name(), for symmetry
   with other sp_instr_set*::exec_core()/sp_rcontext::set_variable*() pairs.
   Now sp_instr_set_row_field_by_name::exec_core() calls
   sp_rcontext::set_variable_row_field_by_name().

10. Misc:
   - Adding a helper private method sp_rcontext::virtual_tmp_table_for_row(),
     reusing it in a new sp_rcontext methods.
   - Removing Item_field_row::get_row_field(), as it's not used any more.
   - Removing the "Item *result_item" from sp_eval_expr(),
     as it's not needed any more.

11 files changed, 317 insertions, 236 deletions

diff --git a/sql/field.cc b/sql/field.cc
index 7b8ec18cad6..3fd76591117 100644
--- a/sql/field.cc
+++ b/sql/field.cc
@@ -1267,6 +1267,45 @@ bool Field::load_data_set_null(THD *thd)
 }
 
 
+bool Field::sp_prepare_and_store_item(THD *thd, Item **value)
+{
+  DBUG_ENTER("Field::sp_prepare_and_store_item");
+  DBUG_ASSERT(value);
+
+  Item *expr_item;
+
+  if (!(expr_item= thd->sp_prepare_func_item(value, 1)))
+    goto error;
+
+  /*
+    expr_item is now fixed, it's safe to call cmp_type()
+  */
+  if (expr_item->cmp_type() == ROW_RESULT)
+  {
+    my_error(ER_OPERAND_COLUMNS, MYF(0), 1);
+    goto error;
+  }
+
+  /* Save the value in the field. Convert the value if needed. */
+
+  expr_item->save_in_field(this, 0);
+
+  if (!thd->is_error())
+    DBUG_RETURN(false);
+
+error:
+  /*
+    In case of error during evaluation, leave the result field set to NULL.
+    Sic: we can't do it in the beginning of the function because the 
+    result field might be needed for its own re-evaluation, e.g. case of 
+    set x = x + 1;
+  */
+  set_null();
+  DBUG_ASSERT(thd->is_error());
+  DBUG_RETURN(true);
+}
+
+
 /**
   Numeric fields base class constructor.
 */
@@ -2295,6 +2334,41 @@ Field_row::~Field_row()
 }
 
 
+bool Field_row::sp_prepare_and_store_item(THD *thd, Item **value)
+{
+  DBUG_ENTER("Field_row::sp_prepare_and_store_item");
+
+  if (value[0]->type() == Item::NULL_ITEM)
+  {
+    /*
+      We're in a auto-generated sp_inst_set, to assign
+      the explicit default NULL value to a ROW variable.
+    */
+    m_table->set_all_fields_to_null();
+    DBUG_RETURN(false);
+  }
+
+  /**
+    - In case if we're assigning a ROW variable from another ROW variable,
+      value[0] points to Item_splocal. sp_fix_func_item() will return the
+      fixed underlying Item_field pointing to Field_row.
+    - In case if we're assigning from a ROW() value, src and value[0] will
+      point to the same Item_row.
+  */
+  Item *src;
+  if (!(src= thd->sp_fix_func_item(value)) ||
+      src->cmp_type() != ROW_RESULT ||
+      src->cols() != m_table->s->fields)
+  {
+    my_error(ER_OPERAND_COLUMNS, MYF(0), m_table->s->fields);
+    m_table->set_all_fields_to_null();
+    DBUG_RETURN(true);
+  }
+
+  DBUG_RETURN(m_table->sp_set_all_fields_from_item(thd, src));
+}
+
+
 /****************************************************************************
   Functions for the Field_decimal class
   This is an number stored as a pre-space (or pre-zero) string
diff --git a/sql/field.h b/sql/field.h
index 0d7d8e42615..287837ee1e7 100644
--- a/sql/field.h
+++ b/sql/field.h
@@ -1521,6 +1521,7 @@ public:
   {
     return NULL;
   }
+  virtual bool sp_prepare_and_store_item(THD *thd, Item **value);
 
   friend int cre_myisam(char * name, register TABLE *form, uint options,
 			ulonglong auto_increment_value);
@@ -3834,6 +3835,7 @@ public:
     {}
   ~Field_row();
   Virtual_tmp_table **virtual_tmp_table_addr() { return &m_table; }
+  bool sp_prepare_and_store_item(THD *thd, Item **value);
 };
 
 
diff --git a/sql/item.cc b/sql/item.cc
index cc3543fe670..98d66df4682 100644
--- a/sql/item.cc
+++ b/sql/item.cc
@@ -1968,14 +1968,11 @@ bool Item_splocal_row_field::set_value(THD *thd, sp_rcontext *ctx, Item **it)
 bool Item_splocal_row_field_by_name::fix_fields(THD *thd, Item **it)
 {
   m_thd= thd;
-  Item *item, *row= m_thd->spcont->get_item(m_var_idx);
-  if (row->element_index_by_name(&m_field_idx, m_field_name))
-  {
-    my_error(ER_ROW_VARIABLE_DOES_NOT_HAVE_FIELD, MYF(0),
-             m_name.str, m_field_name.str);
+  if (thd->spcont->find_row_field_by_name_or_error(&m_field_idx,
+                                                   m_var_idx,
+                                                   m_field_name))
     return true;
-  }
-  item= row->element_index(m_field_idx);
+  Item *item= thd->spcont->get_item(m_var_idx)->element_index(m_field_idx);
   set_handler(item->type_handler());
   return fix_fields_from_item(thd, it, item);
 }
@@ -7424,26 +7421,6 @@ void Item_field::print(String *str, enum_query_type query_type)
 }
 
 
-bool Item_field_row::element_index_by_name(uint *idx,
-                                           const LEX_CSTRING &name) const
-{
-  Field *field;
-  for (uint i= 0; (field= get_row_field(i)); i++)
-  {
-    // Use the same comparison style with sp_context::find_variable()
-    if (!my_strnncoll(system_charset_info,
-                      (const uchar *) field->field_name.str,
-                      field->field_name.length,
-                      (const uchar *) name.str, name.length))
-    {
-      *idx= i;
-      return false;
-    }
-  }
-  return true;
-}
-
-
 void Item_temptable_field::print(String *str, enum_query_type query_type)
 {
   /*
@@ -9290,7 +9267,7 @@ void Item_trigger_field::set_required_privilege(bool rw)
 
 bool Item_trigger_field::set_value(THD *thd, sp_rcontext * /*ctx*/, Item **it)
 {
-  Item *item= sp_prepare_func_item(thd, it);
+  Item *item= thd->sp_prepare_func_item(it);
 
   if (!item)
     return true;
diff --git a/sql/item.h b/sql/item.h
index 46a65b31b38..f4fff4a0418 100644
--- a/sql/item.h
+++ b/sql/item.h
@@ -1710,10 +1710,6 @@ public:
   // Row emulation
   virtual uint cols() const { return 1; }
   virtual Item* element_index(uint i) { return this; }
-  virtual bool element_index_by_name(uint *idx, const LEX_CSTRING &name) const
-  {
-    return true; // Error
-  }
   virtual Item** addr(uint i) { return 0; }
   virtual bool check_cols(uint c);
   bool check_type_traditional_scalar(const char *opname) const;
@@ -2943,7 +2939,6 @@ public:
 
   const Type_handler *type_handler() const { return &type_handler_row; }
   uint cols() const { return arg_count; }
-  bool element_index_by_name(uint *idx, const LEX_CSTRING &name) const;
   Item* element_index(uint i) { return arg_count ? args[i] : this; }
   Item** addr(uint i) { return arg_count ? args + i : NULL; }
   bool check_cols(uint c)
@@ -2956,7 +2951,6 @@ public:
     return false;
   }
   bool row_create_items(THD *thd, List<Spvar_definition> *list);
-  Field *get_row_field(uint i) const;
 };
 
 
diff --git a/sql/sp_head.cc b/sql/sp_head.cc
index 7f5c22eab89..ead04368acd 100644
--- a/sql/sp_head.cc
+++ b/sql/sp_head.cc
@@ -27,6 +27,7 @@
 #include "sql_array.h"         // Dynamic_array
 #include "log_event.h"         // Query_log_event
 #include "sql_derived.h"       // mysql_handle_derived
+#include "sql_select.h"        // Virtual_tmp_table
 
 #ifdef USE_PRAGMA_IMPLEMENTATION
 #pragma implementation
@@ -337,8 +338,8 @@ sp_get_flags_for_command(LEX *lex)
 /**
   Prepare an Item for evaluation (call of fix_fields).
 
-  @param thd       thread handler
   @param it_addr   pointer on item refernce
+  @param cols      expected number of elements (1 for scalar, >=1 for ROWs)
 
   @retval
     NULL      error
@@ -346,21 +347,32 @@ sp_get_flags_for_command(LEX *lex)
     non-NULL  prepared item
 */
 
-Item *
-sp_prepare_func_item(THD* thd, Item **it_addr, uint cols)
+Item *THD::sp_prepare_func_item(Item **it_addr, uint cols)
 {
-  DBUG_ENTER("sp_prepare_func_item");
+  DBUG_ENTER("THD::sp_prepare_func_item");
+  Item *res= sp_fix_func_item(it_addr);
+  if (res && res->check_cols(cols))
+    DBUG_RETURN(NULL);
+  DBUG_RETURN(res);
+}
+
+
+/**
+  Fix an Item for evaluation for SP.
+*/
+Item *THD::sp_fix_func_item(Item **it_addr)
+{
+  DBUG_ENTER("THD::sp_fix_func_item");
   if (!(*it_addr)->fixed &&
-      (*it_addr)->fix_fields(thd, it_addr))
+      (*it_addr)->fix_fields(this, it_addr))
   {
     DBUG_PRINT("info", ("fix_fields() failed"));
     DBUG_RETURN(NULL);
   }
-  it_addr= (*it_addr)->this_item_addr(thd, it_addr);
+  it_addr= (*it_addr)->this_item_addr(this, it_addr);
 
-  if ((!(*it_addr)->fixed &&
-       (*it_addr)->fix_fields(thd, it_addr)) ||
-      (*it_addr)->check_cols(cols))
+  if (!(*it_addr)->fixed &&
+      (*it_addr)->fix_fields(this, it_addr))
   {
     DBUG_PRINT("info", ("fix_fields() failed"));
     DBUG_RETURN(NULL);
@@ -372,7 +384,6 @@ sp_prepare_func_item(THD* thd, Item **it_addr, uint cols)
 /**
   Evaluate an expression and store the result in the field.
 
-  @param thd                    current thread object
   @param result_field           the field to store the result
   @param expr_item_ptr          the root item of the expression
 
@@ -382,67 +393,13 @@ sp_prepare_func_item(THD* thd, Item **it_addr, uint cols)
     TRUE   on error
 */
 
-bool
-sp_eval_expr(THD *thd, Item *result_item, Field *result_field,
-             Item **expr_item_ptr)
+bool THD::sp_eval_expr(Field *result_field, Item **expr_item_ptr)
 {
-  Item *expr_item;
-  enum_check_fields save_count_cuted_fields= thd->count_cuted_fields;
-  bool save_abort_on_warning= thd->abort_on_warning;
-  bool save_stmt_modified_non_trans_table= 
-    thd->transaction.stmt.modified_non_trans_table;
-
-  DBUG_ENTER("sp_eval_expr");
-
-  if (!*expr_item_ptr)
-    goto error;
-
-  if (!(expr_item= sp_prepare_func_item(thd, expr_item_ptr,
-                                        result_item ? result_item->cols() : 1)))
-    goto error;
-
-  /*
-    expr_item is now fixed, it's safe to call cmp_type()
-    If result_item is NULL, then we're setting the RETURN value.
-  */
-  if ((!result_item || result_item->cmp_type() != ROW_RESULT) &&
-      expr_item->cmp_type() == ROW_RESULT)
-  {
-    my_error(ER_OPERAND_COLUMNS, MYF(0), 1);
-    goto error;
-  }
-
-  /*
-    Set THD flags to emit warnings/errors in case of overflow/type errors
-    during saving the item into the field.
-
-    Save original values and restore them after save.
-  */
-
-  thd->count_cuted_fields= CHECK_FIELD_ERROR_FOR_NULL;
-  thd->abort_on_warning= thd->is_strict_mode();
-  thd->transaction.stmt.modified_non_trans_table= FALSE;
-
+  DBUG_ENTER("THD::sp_eval_expr");
+  DBUG_ASSERT(*expr_item_ptr);
+  Sp_eval_expr_state state(this);
   /* Save the value in the field. Convert the value if needed. */
-
-  expr_item->save_in_field(result_field, 0);
-
-  thd->count_cuted_fields= save_count_cuted_fields;
-  thd->abort_on_warning= save_abort_on_warning;
-  thd->transaction.stmt.modified_non_trans_table= save_stmt_modified_non_trans_table;
-
-  if (!thd->is_error())
-    DBUG_RETURN(FALSE);
-
-error:
-  /*
-    In case of error during evaluation, leave the result field set to NULL.
-    Sic: we can't do it in the beginning of the function because the 
-    result field might be needed for its own re-evaluation, e.g. case of 
-    set x = x + 1;
-  */
-  result_field->set_null();
-  DBUG_RETURN (TRUE);
+  DBUG_RETURN(result_field->sp_prepare_and_store_item(this, expr_item_ptr));
 }
 
 
@@ -3371,19 +3328,7 @@ int
 sp_instr_set::exec_core(THD *thd, uint *nextp)
 {
   int res= thd->spcont->set_variable(thd, m_offset, &m_value);
-
-  if (res)
-  {
-    /* Failed to evaluate the value. Reset the variable to NULL. */
-
-    if (thd->spcont->set_variable(thd, m_offset, 0))
-    {
-      /* If this also failed, let's abort. */
-      my_error(ER_OUT_OF_RESOURCES, MYF(ME_FATALERROR));
-    }
-  }
   delete_explain_query(thd->lex);
-
   *nextp = m_ip+1;
   return res;
 }
@@ -3422,11 +3367,6 @@ sp_instr_set_row_field::exec_core(THD *thd, uint *nextp)
 {
   int res= thd->spcont->set_variable_row_field(thd, m_offset, m_field_offset,
                                                &m_value);
-  if (res)
-  {
-    /* Failed to evaluate the value. Reset the variable to NULL. */
-    thd->spcont->set_variable_row_field_to_null(thd, m_offset, m_field_offset);
-  }
   delete_explain_query(thd->lex);
   *nextp= m_ip + 1;
   return res;
@@ -3470,23 +3410,9 @@ sp_instr_set_row_field::print(String *str)
 int
 sp_instr_set_row_field_by_name::exec_core(THD *thd, uint *nextp)
 {
-  int res;
-  uint idx;
-  Item_field_row *row= (Item_field_row*) thd->spcont->get_item(m_offset);
-  if ((res= row->element_index_by_name(&idx, m_field_name)))
-  {
-    sp_variable *var= m_ctx->find_variable(m_offset);
-    my_error(ER_ROW_VARIABLE_DOES_NOT_HAVE_FIELD, MYF(0),
-             var->name.str, m_field_name.str);
-    goto error;
-  }
-  res= thd->spcont->set_variable_row_field(thd, m_offset, idx, &m_value);
-  if (res)
-  {
-    /* Failed to evaluate the value. Reset the variable to NULL. */
-    thd->spcont->set_variable_row_field_to_null(thd, m_offset, idx);
-  }
-error:
+  int res= thd->spcont->set_variable_row_field_by_name(thd, m_offset,
+                                                       m_field_name,
+                                                       &m_value);
   delete_explain_query(thd->lex);
   *nextp= m_ip + 1;
   return res;
@@ -3650,7 +3576,7 @@ sp_instr_jump_if_not::exec_core(THD *thd, uint *nextp)
   Item *it;
   int res;
 
-  it= sp_prepare_func_item(thd, &m_expr);
+  it= thd->sp_prepare_func_item(&m_expr);
   if (! it)
   {
     res= -1;
diff --git a/sql/sp_head.h b/sql/sp_head.h
index f8a819bbe94..9f6d4cd95e5 100644
--- a/sql/sp_head.h
+++ b/sql/sp_head.h
@@ -1910,13 +1910,6 @@ sp_add_to_query_tables(THD *thd, LEX *lex,
                        thr_lock_type locktype,
                        enum_mdl_type mdl_type);
 
-Item *
-sp_prepare_func_item(THD* thd, Item **it_addr, uint cols= 1);
-
-bool
-sp_eval_expr(THD *thd, Item *result_item, Field *result_field,
-             Item **expr_item_ptr);
-
 /**
   @} (end of group Stored_Routines)
 */
diff --git a/sql/sp_rcontext.cc b/sql/sp_rcontext.cc
index 62b5439d84d..43d033ad83e 100644
--- a/sql/sp_rcontext.cc
+++ b/sql/sp_rcontext.cc
@@ -388,22 +388,13 @@ bool Item_field_row::row_create_items(THD *thd, List<Spvar_definition> *list)
 }
 
 
-Field *Item_field_row::get_row_field(uint i) const
-{
-  DBUG_ASSERT(field);
-  Virtual_tmp_table **ptable= field->virtual_tmp_table_addr();
-  DBUG_ASSERT(ptable);
-  return ptable[0]->field[i];
-}
-
-
 bool sp_rcontext::set_return_value(THD *thd, Item **return_value_item)
 {
   DBUG_ASSERT(m_return_value_fld);
 
   m_return_value_set = true;
 
-  return sp_eval_expr(thd, NULL, m_return_value_fld, return_value_item);
+  return thd->sp_eval_expr(m_return_value_fld, return_value_item);
 }
 
 
@@ -612,84 +603,31 @@ uint sp_rcontext::exit_handler(Diagnostics_area *da)
 
 int sp_rcontext::set_variable(THD *thd, uint idx, Item **value)
 {
-  Field *field= m_var_table->field[idx];
-  if (!value)
-  {
-    field->set_null();
-    return 0;
-  }
-  Item *dst= m_var_items[idx];
-
-  if (dst->cmp_type() != ROW_RESULT)
-    return sp_eval_expr(thd, dst, m_var_table->field[idx], value);
-
-  DBUG_ASSERT(dst->type() == Item::FIELD_ITEM);
-  if (value[0]->type() == Item::NULL_ITEM)
-  {
-    /*
-      We're in a auto-generated sp_inst_set, to assign
-      the explicit default NULL value to a ROW variable.
-    */
-    for (uint i= 0; i < dst->cols(); i++)
-    {
-      Item_field_row *item_field_row= (Item_field_row*) dst;
-      item_field_row->get_row_field(i)->set_null();
-    }
-    return false;
-  }
-
-  /**
-    - In case if we're assigning a ROW variable from another ROW variable,
-      value[0] points to Item_splocal. sp_prepare_func_item() will return the
-      fixed underlying Item_field_spvar with ROW members in its aguments().
-    - In case if we're assigning from a ROW() value, src and value[0] will
-      point to the same Item_row.
-  */
-  Item *src;
-  if (!(src= sp_prepare_func_item(thd, value, dst->cols())) ||
-      src->cmp_type() != ROW_RESULT)
-  {
-    my_error(ER_OPERAND_COLUMNS, MYF(0), dst->cols());
-    return true;
-  }
-  DBUG_ASSERT(dst->cols() == src->cols());
-  for (uint i= 0; i < src->cols(); i++)
-    set_variable_row_field(thd, idx, i, src->addr(i));
-  return false;
-}
-
-
-void sp_rcontext::set_variable_row_field_to_null(THD *thd,
-                                                 uint var_idx,
-                                                 uint field_idx)
-{
-  Item *dst= get_item(var_idx);
-  DBUG_ASSERT(dst->type() == Item::FIELD_ITEM);
-  DBUG_ASSERT(dst->cmp_type() == ROW_RESULT);
-  Item_field_row *item_field_row= (Item_field_row*) dst;
-  item_field_row->get_row_field(field_idx)->set_null();
+  DBUG_ENTER("sp_rcontext::set_variable");
+  DBUG_ASSERT(value);
+  DBUG_RETURN(thd->sp_eval_expr(m_var_table->field[idx], value));
 }
 
 
 int sp_rcontext::set_variable_row_field(THD *thd, uint var_idx, uint field_idx,
                                         Item **value)
 {
+  DBUG_ENTER("sp_rcontext::set_variable_row_field");
   DBUG_ASSERT(value);
-  Item *dst= get_item(var_idx);
-  DBUG_ASSERT(dst->type() == Item::FIELD_ITEM);
-  DBUG_ASSERT(dst->cmp_type() == ROW_RESULT);
-  Item_field_row *item_field_row= (Item_field_row*) dst;
+  Virtual_tmp_table *vtable= virtual_tmp_table_for_row(var_idx);
+  DBUG_RETURN(thd->sp_eval_expr(vtable->field[field_idx], value));
+}
 
-  Item *expr_item= sp_prepare_func_item(thd, value);
-  if (!expr_item)
-  {
-    DBUG_ASSERT(thd->is_error());
-    return true;
-  }
-  return sp_eval_expr(thd,
-                      item_field_row->arguments()[field_idx],
-                      item_field_row->get_row_field(field_idx),
-                      value);
+
+int sp_rcontext::set_variable_row_field_by_name(THD *thd, uint var_idx,
+                                                const LEX_CSTRING &field_name,
+                                                Item **value)
+{
+  DBUG_ENTER("sp_rcontext::set_variable_row_field_by_name");
+  uint field_idx;
+  if (find_row_field_by_name_or_error(&field_idx, var_idx, field_name))
+    DBUG_RETURN(1);
+  DBUG_RETURN(set_variable_row_field(thd, var_idx, field_idx, value));
 }
 
 
@@ -697,15 +635,32 @@ int sp_rcontext::set_variable_row(THD *thd, uint var_idx, List<Item> &items)
 {
   DBUG_ENTER("sp_rcontext::set_variable_row");
   DBUG_ASSERT(get_item(var_idx)->cols() == items.elements);
-  List_iterator<Item> it(items);
-  Item *item;
-  for (uint i= 0 ; (item= it++) ; i++)
-  {
-    int rc;
-    if ((rc= set_variable_row_field(thd, var_idx, i, &item)))
-      DBUG_RETURN(rc);
-  }
-  DBUG_RETURN(0);
+  Virtual_tmp_table *vtable= virtual_tmp_table_for_row(var_idx);
+  Sp_eval_expr_state state(thd);
+  DBUG_RETURN(vtable->sp_set_all_fields_from_item_list(thd, items));
+}
+
+
+Virtual_tmp_table *sp_rcontext::virtual_tmp_table_for_row(uint var_idx)
+{
+  DBUG_ASSERT(get_item(var_idx)->type() == Item::FIELD_ITEM);
+  DBUG_ASSERT(get_item(var_idx)->cmp_type() == ROW_RESULT);
+  Field *field= m_var_table->field[var_idx];
+  Virtual_tmp_table **ptable= field->virtual_tmp_table_addr();
+  DBUG_ASSERT(ptable);
+  DBUG_ASSERT(ptable[0]);
+  return ptable[0];
+}
+
+
+bool sp_rcontext::find_row_field_by_name_or_error(uint *field_idx,
+                                                  uint var_idx,
+                                                  const LEX_CSTRING &field_name)
+{
+  Virtual_tmp_table *vtable= virtual_tmp_table_for_row(var_idx);
+  Field *row= m_var_table->field[var_idx];
+  return vtable->sp_find_field_by_name_or_error(field_idx,
+                                                row->field_name, field_name);
 }
 
 
@@ -728,7 +683,7 @@ Item_cache *sp_rcontext::create_case_expr_holder(THD *thd,
 bool sp_rcontext::set_case_expr(THD *thd, int case_expr_id,
                                 Item **case_expr_item_ptr)
 {
-  Item *case_expr_item= sp_prepare_func_item(thd, case_expr_item_ptr);
+  Item *case_expr_item= thd->sp_prepare_func_item(case_expr_item_ptr);
   if (!case_expr_item)
     return true;
 
diff --git a/sql/sp_rcontext.h b/sql/sp_rcontext.h
index 26c06512417..ca438107593 100644
--- a/sql/sp_rcontext.h
+++ b/sql/sp_rcontext.h
@@ -190,9 +190,11 @@ public:
   /////////////////////////////////////////////////////////////////////////
 
   int set_variable(THD *thd, uint var_idx, Item **value);
-  void set_variable_row_field_to_null(THD *thd, uint var_idx, uint field_idx);
   int set_variable_row_field(THD *thd, uint var_idx, uint field_idx,
                              Item **value);
+  int set_variable_row_field_by_name(THD *thd, uint var_idx,
+                                     const LEX_CSTRING &field_name,
+                                     Item **value);
   int set_variable_row(THD *thd, uint var_idx, List<Item> &items);
   Item *get_item(uint var_idx) const
   { return m_var_items[var_idx]; }
@@ -200,6 +202,9 @@ public:
   Item **get_item_addr(uint var_idx) const
   { return m_var_items.array() + var_idx; }
 
+  bool find_row_field_by_name_or_error(uint *field_idx, uint var_idx,
+                                       const LEX_CSTRING &field_name);
+
   bool set_return_value(THD *thd, Item **return_value_item);
 
   bool is_return_value_set() const
@@ -363,6 +368,8 @@ private:
   /// @return Pointer to valid object on success, or NULL in case of error.
   Item_cache *create_case_expr_holder(THD *thd, const Item *item) const;
 
+  Virtual_tmp_table *virtual_tmp_table_for_row(uint idx);
+
 private:
   /// Top-level (root) parsing context for this runtime context.
   const sp_pcontext *m_root_parsing_ctx;
diff --git a/sql/sql_class.h b/sql/sql_class.h
index a4a631ee9f0..70a57fa22f7 100644
--- a/sql/sql_class.h
+++ b/sql/sql_class.h
@@ -4552,6 +4552,10 @@ public:
     See also sp_head::merge_lex().
   */
   bool restore_from_local_lex_to_old_lex(LEX *oldlex);
+
+  Item *sp_fix_func_item(Item **it_addr);
+  Item *sp_prepare_func_item(Item **it_addr, uint cols= 1);
+  bool sp_eval_expr(Field *result_field, Item **expr_item_ptr);
 };
 
 inline void add_to_active_threads(THD *thd)
@@ -6162,6 +6166,49 @@ public:
 };
 
 
+/*
+  A helper class to set THD flags to emit warnings/errors in case of
+  overflow/type errors during assigning values into the SP variable fields.
+  Saves original flags values in constructor.
+  Restores original flags in destructor.
+*/
+class Sp_eval_expr_state
+{
+  THD *m_thd;
+  enum_check_fields m_count_cuted_fields;
+  bool m_abort_on_warning;
+  bool m_stmt_modified_non_trans_table;
+  void start()
+  {
+    m_thd->count_cuted_fields= CHECK_FIELD_ERROR_FOR_NULL;
+    m_thd->abort_on_warning= m_thd->is_strict_mode();
+    m_thd->transaction.stmt.modified_non_trans_table= false;
+  }
+  void stop()
+  {
+    m_thd->count_cuted_fields= m_count_cuted_fields;
+    m_thd->abort_on_warning= m_abort_on_warning;
+    m_thd->transaction.stmt.modified_non_trans_table=
+      m_stmt_modified_non_trans_table;
+  }
+public:
+  Sp_eval_expr_state(THD *thd)
+   :m_thd(thd),
+    m_count_cuted_fields(thd->count_cuted_fields),
+    m_abort_on_warning(thd->abort_on_warning),
+    m_stmt_modified_non_trans_table(thd->transaction.stmt.
+                                    modified_non_trans_table)
+  {
+    start();
+  }
+  ~Sp_eval_expr_state()
+  {
+    stop();
+  }
+};
+
+
+
 #endif /* MYSQL_SERVER */
 
 #endif /* SQL_CLASS_INCLUDED */
diff --git a/sql/sql_select.cc b/sql/sql_select.cc
index df03a0d30e7..2784938b654 100644
--- a/sql/sql_select.cc
+++ b/sql/sql_select.cc
@@ -17736,6 +17736,70 @@ bool Virtual_tmp_table::open()
 }
 
 
+bool Virtual_tmp_table::sp_find_field_by_name(uint *idx,
+                                              const LEX_CSTRING &name) const
+{
+  Field *f;
+  for (uint i= 0; (f= field[i]); i++)
+  {
+    // Use the same comparison style with sp_context::find_variable()
+    if (!my_strnncoll(system_charset_info,
+                      (const uchar *) f->field_name.str,
+                      f->field_name.length,
+                      (const uchar *) name.str, name.length))
+    {
+      *idx= i;
+      return false;
+    }
+  }
+  return true;
+}
+
+
+bool
+Virtual_tmp_table::sp_find_field_by_name_or_error(uint *idx,
+                                                  const LEX_CSTRING &var_name,
+                                                  const LEX_CSTRING &field_name)
+                                                  const
+{
+  if (sp_find_field_by_name(idx, field_name))
+  {
+    my_error(ER_ROW_VARIABLE_DOES_NOT_HAVE_FIELD, MYF(0),
+             var_name.str, field_name.str);
+    return true;
+  }
+  return false;
+}
+
+
+bool Virtual_tmp_table::sp_set_all_fields_from_item_list(THD *thd,
+                                                         List<Item> &items)
+{
+  DBUG_ASSERT(s->fields == items.elements);
+  List_iterator<Item> it(items);
+  Item *item;
+  for (uint i= 0 ; (item= it++) ; i++)
+  {
+    if (field[i]->sp_prepare_and_store_item(thd, &item))
+      return true;
+  }
+  return false;
+}
+
+
+bool Virtual_tmp_table::sp_set_all_fields_from_item(THD *thd, Item *value)
+{
+  DBUG_ASSERT(value->fixed);
+  DBUG_ASSERT(value->cols() == s->fields);
+  for (uint i= 0; i < value->cols(); i++)
+  {
+    if (field[i]->sp_prepare_and_store_item(thd, value->addr(i)))
+      return true;
+  }
+  return false;
+}
+
+
 bool open_tmp_table(TABLE *table)
 {
   int error;
diff --git a/sql/sql_select.h b/sql/sql_select.h
index 4e2206dd098..7cfc35a3b7d 100644
--- a/sql/sql_select.h
+++ b/sql/sql_select.h
@@ -2127,6 +2127,48 @@ public:
     @return true  - on error (e.g. could not allocate the record buffer).
   */
   bool open();
+
+  void set_all_fields_to_null()
+  {
+    for (uint i= 0; i < s->fields; i++)
+      field[i]->set_null();
+  }
+  /**
+    Set all fields from a compatible item list.
+    The number of fields in "this" must be equal to the number
+    of elements in "value".
+  */
+  bool sp_set_all_fields_from_item_list(THD *thd, List<Item> &items);
+
+  /**
+    Set all fields from a compatible item.
+    The number of fields in "this" must be the same with the number
+    of elements in "value".
+  */
+  bool sp_set_all_fields_from_item(THD *thd, Item *value);
+
+  /**
+    Find a ROW element index by its name
+    Assumes that "this" is used as a storage for a ROW-type SP variable.
+    @param [OUT] idx        - the index of the found field is returned here
+    @param [IN]  field_name - find a field with this name
+    @return      true       - on error (the field was not found)
+    @return      false      - on success (idx[0] was set to the field index)
+  */
+  bool sp_find_field_by_name(uint *idx, const LEX_CSTRING &name) const;
+
+  /**
+    Find a ROW element index by its name.
+    If the element is not found, and error is issued.
+    @param [OUT] idx        - the index of the found field is returned here
+    @param [IN]  var_name   - the name of the ROW variable (for error reporting)
+    @param [IN]  field_name - find a field with this name
+    @return      true       - on error (the field was not found)
+    @return      false      - on success (idx[0] was set to the field index)
+  */
+  bool sp_find_field_by_name_or_error(uint *idx,
+                                      const LEX_CSTRING &var_name,
+                                      const LEX_CSTRING &field_name) const;
 };