MDEV-10914 ROW data type for stored routine variables

24 files changed, 1458 insertions, 295 deletions

diff --git a/sql/field.cc b/sql/field.cc
index 40bd6a8a2dd..06cf9fbee58 100644
--- a/sql/field.cc
+++ b/sql/field.cc
@@ -10203,6 +10203,7 @@ bool Column_definition::check(THD *thd)
       ((length > max_field_charlength &&
         sql_type != MYSQL_TYPE_VARCHAR) ||
        (length == 0 &&
+        sql_type != MYSQL_TYPE_NULL /* e.g. a ROW variable */ &&
         sql_type != MYSQL_TYPE_ENUM && sql_type != MYSQL_TYPE_SET &&
         sql_type != MYSQL_TYPE_STRING && sql_type != MYSQL_TYPE_VARCHAR &&
         sql_type != MYSQL_TYPE_GEOMETRY)))
diff --git a/sql/field.h b/sql/field.h
index 359101530f3..68327159a22 100644
--- a/sql/field.h
+++ b/sql/field.h
@@ -3958,11 +3958,47 @@ public:
 
 
 /**
+  List of ROW element definitions, e.g.:
+    DECLARE a ROW(a INT,b VARCHAR(10))
+*/
+class Row_definition_list: public List<class Spvar_definition>
+{
+public:
+  inline bool eq_name(const Spvar_definition *def, const char *name) const;
+  /**
+    Find a ROW field by name.
+    @param [IN]  name   - the name
+    @param [OUT] offset - if the ROW field found, its offset it returned here
+    @retval NULL        - the ROW field was not found
+    @retval !NULL       - the pointer to the found ROW field
+  */
+  Spvar_definition *find_row_field_by_name(const char *name, uint *offset) const
+  {
+    // Cast-off the "const" qualifier
+    List_iterator<Spvar_definition> it(*((List<Spvar_definition>*)this));
+    Spvar_definition *def;
+    for (*offset= 0; (def= it++); (*offset)++)
+    {
+      if (eq_name(def, name))
+        return def;
+    }
+    return 0;
+  }
+};
+
+
+/**
   This class is used during a stored routine or a trigger execution,
   at sp_rcontext::create() time.
   Currently it can represent:
   - variables with explicit data types:   DECLARE a INT;
   - variables with data type references:  DECLARE a t1.a%TYPE;
+  - ROW type variables
+
+  Notes:
+  - Scalar variables have m_field_definitions==NULL.
+  - ROW variables are defined as having MYSQL_TYPE_NULL,
+    with a non-empty m_field_definitions.
 
   Data type references to other object types will be added soon, e.g.:
   - DECLARE a table_name%ROWTYPE;
@@ -3973,9 +4009,11 @@ public:
 class Spvar_definition: public Column_definition
 {
   class Qualified_column_ident *m_column_type_ref; // for %TYPE
+  Row_definition_list *m_row_field_definitions;    // for ROW
 public:
   Spvar_definition()
-   :m_column_type_ref(NULL) { }
+   :m_column_type_ref(NULL),
+    m_row_field_definitions(NULL) { }
   bool is_column_type_ref() const { return m_column_type_ref != 0; }
   class Qualified_column_ident *column_type_ref() const
   {
@@ -3985,9 +4023,39 @@ public:
   {
     m_column_type_ref= ref;
   }
+
+  /*
+    Find a ROW field by name.
+    See Row_field_list::find_row_field_by_name() for details.
+  */
+  Spvar_definition *find_row_field_by_name(const char *name, uint *offset) const
+  {
+    DBUG_ASSERT(m_row_field_definitions);
+    return m_row_field_definitions->find_row_field_by_name(name, offset);
+  }
+  uint is_row() const
+  {
+    return m_row_field_definitions != NULL;
+  }
+  Row_definition_list *row_field_definitions() const
+  {
+    return m_row_field_definitions;
+  }
+  void set_row_field_definitions(Row_definition_list *list)
+  {
+    m_row_field_definitions= list;
+  }
+
 };
 
 
+inline bool Row_definition_list::eq_name(const Spvar_definition *def,
+                                         const char *name) const
+{
+  return my_strcasecmp(system_charset_info, def->field_name, name) == 0;
+}
+
+
 class Create_field :public Column_definition
 {
 public:
diff --git a/sql/item.cc b/sql/item.cc
index 9ad54cfad70..56428fb39bb 100644
--- a/sql/item.cc
+++ b/sql/item.cc
@@ -1653,6 +1653,126 @@ bool Item_splocal::set_value(THD *thd, sp_rcontext *ctx, Item **it)
 }
 
 
+/**
+  These two declarations are different:
+    x INT;
+    ROW(x INT);
+  A ROW with one elements should not be comparable to scalar value.
+
+  TODO: Currently we don't support one argument with the function ROW(), so
+  this query returns a syntax error, meaning that more arguments are expected:
+    SELECT ROW(1);
+
+  Therefore, all around the code we assume that cols()==1 means a scalar value
+  and cols()>1 means a ROW value. With adding ROW SP variables this
+  assumption is not true any more. ROW variables with one element are
+  now possible.
+
+  To implement Item::check_cols() correctly, we now should extend it to
+  know if a ROW or a scalar value is being tested. For example,
+  these new prototypes should work:
+    virtual bool check_cols(Item_result result, uint c);
+  or
+    virtual bool check_cols(const Type_handler *type, uint c);
+
+  The current implementation of Item_splocal::check_cols() is a compromise
+  that should be more or less fine until we extend check_cols().
+  It disallows ROW variables to appear in a scalar context.
+  The "|| n == 1" part of the conditon is responsible for this.
+  For example, it disallows ROW variables to appear in SELECT list:
+
+DELIMITER $$;
+CREATE PROCEDURE p1()
+AS
+  a ROW (a INT);
+BEGIN
+  SELECT a;
+END;
+$$
+DELIMITER ;$$
+--error ER_OPERAND_COLUMNS
+CALL p1();
+
+  But is produces false negatives with ROW variables consisting of one element.
+  For example, this script fails:
+
+SET sql_mode=ORACLE;
+DROP PROCEDURE IF EXISTS p1;
+DELIMITER $$
+CREATE PROCEDURE p1
+AS
+  a ROW(a INT);
+  b ROW(a INT);
+BEGIN
+  SELECT a=b;
+END;
+$$
+DELIMITER ;
+CALL p1();
+
+  and returns "ERROR 1241 (21000): Operand should contain 1 column(s)".
+  This will be fixed that we change check_cols().
+*/
+
+bool Item_splocal::check_cols(uint n)
+{
+  DBUG_ASSERT(m_thd->spcont);
+  if (cmp_type() != ROW_RESULT)
+    return Item::check_cols(n);
+
+  if (n != this_item()->cols() || n == 1)
+  {
+    my_error(ER_OPERAND_COLUMNS, MYF(0), n);
+    return true;
+  }
+  return false;
+}
+
+
+Item *
+Item_splocal_row_field::this_item()
+{
+  DBUG_ASSERT(m_sp == m_thd->spcont->sp);
+  return m_thd->spcont->get_item(m_var_idx)->element_index(m_field_idx);
+}
+
+
+const Item *
+Item_splocal_row_field::this_item() const
+{
+  DBUG_ASSERT(m_sp == m_thd->spcont->sp);
+  return m_thd->spcont->get_item(m_var_idx)->element_index(m_field_idx);
+}
+
+
+Item **
+Item_splocal_row_field::this_item_addr(THD *thd, Item **)
+{
+  DBUG_ASSERT(m_sp == thd->spcont->sp);
+  return thd->spcont->get_item(m_var_idx)->addr(m_field_idx);
+}
+
+
+void Item_splocal_row_field::print(String *str, enum_query_type)
+{
+  str->reserve(m_name.length + m_field_name.length + 8);
+  str->append(m_name.str, m_name.length);
+  str->append('.');
+  str->append(m_field_name.str, m_field_name.length);
+  str->append('@');
+  str->qs_append(m_var_idx);
+  str->append('[');
+  str->qs_append(m_field_idx);
+  str->append(']');
+}
+
+
+bool Item_splocal_row_field::set_value(THD *thd, sp_rcontext *ctx, Item **it)
+{
+  return ctx->set_variable_row_field(thd, m_var_idx, m_field_idx, it);
+}
+
+
 /*****************************************************************************
   Item_case_expr methods
 *****************************************************************************/
diff --git a/sql/item.h b/sql/item.h
index 4e737d09952..01573c9e6ea 100644
--- a/sql/item.h
+++ b/sql/item.h
@@ -1843,6 +1843,94 @@ inline Item* get_item_copy (THD *thd, MEM_ROOT *mem_root, T* item)
 bool cmp_items(Item *a, Item *b);
 
 
+/**
+  Array of items, e.g. function or aggerate function arguments.
+*/
+class Item_args
+{
+protected:
+  Item **args, *tmp_arg[2];
+  uint arg_count;
+  bool alloc_arguments(THD *thd, uint count);
+  void set_arguments(THD *thd, List<Item> &list);
+  bool walk_args(Item_processor processor, bool walk_subquery, void *arg)
+  {
+    for (uint i= 0; i < arg_count; i++)
+    {
+      if (args[i]->walk(processor, walk_subquery, arg))
+        return true;
+    }
+    return false;
+  }
+  bool transform_args(THD *thd, Item_transformer transformer, uchar *arg);
+  void propagate_equal_fields(THD *, const Item::Context &, COND_EQUAL *);
+public:
+  Item_args(void)
+    :args(NULL), arg_count(0)
+  { }
+  Item_args(Item *a)
+    :args(tmp_arg), arg_count(1)
+  {
+    args[0]= a;
+  }
+  Item_args(Item *a, Item *b)
+    :args(tmp_arg), arg_count(2)
+  {
+    args[0]= a; args[1]= b;
+  }
+  Item_args(THD *thd, Item *a, Item *b, Item *c)
+  {
+    arg_count= 0;
+    if ((args= (Item**) thd_alloc(thd, sizeof(Item*) * 3)))
+    {
+      arg_count= 3;
+      args[0]= a; args[1]= b; args[2]= c;
+    }
+  }
+  Item_args(THD *thd, Item *a, Item *b, Item *c, Item *d)
+  {
+    arg_count= 0;
+    if ((args= (Item**) thd_alloc(thd, sizeof(Item*) * 4)))
+    {
+      arg_count= 4;
+      args[0]= a; args[1]= b; args[2]= c; args[3]= d;
+    }
+  }
+  Item_args(THD *thd, Item *a, Item *b, Item *c, Item *d, Item* e)
+  {
+    arg_count= 5;
+    if ((args= (Item**) thd_alloc(thd, sizeof(Item*) * 5)))
+    {
+      arg_count= 5;
+      args[0]= a; args[1]= b; args[2]= c; args[3]= d; args[4]= e;
+    }
+  }
+  Item_args(THD *thd, List<Item> &list)
+  {
+    set_arguments(thd, list);
+  }
+  Item_args(THD *thd, const Item_args *other);
+  inline Item **arguments() const { return args; }
+  inline uint argument_count() const { return arg_count; }
+  inline void remove_arguments() { arg_count=0; }
+};
+
+
+class Item_spvar_args: public Item_args
+{
+  TABLE *m_table;
+public:
+  Item_spvar_args():Item_args(), m_table(NULL) { }
+  ~Item_spvar_args();
+  bool row_create_items(THD *thd, List<Spvar_definition> *list);
+  Field *get_row_field(uint i) const
+  {
+    DBUG_ASSERT(m_table);
+    return m_table->field[i];
+  }
+};
+
+
 /*
   Class to be used to enumerate all field references in an item tree. This
   includes references to outside but not fields of the tables within a
@@ -2076,9 +2164,12 @@ class Item_splocal :public Item_sp_variable,
                     public Rewritable_query_parameter,
                     public Type_handler_hybrid_field_type
 {
+protected:
   uint m_var_idx;
 
   Type m_type;
+
+  bool append_value_for_log(THD *thd, String *str);
 public:
   Item_splocal(THD *thd, const LEX_STRING &sp_var_name, uint sp_var_idx,
                enum_field_types sp_var_type,
@@ -2104,6 +2195,10 @@ public:
   { return Type_handler_hybrid_field_type::result_type(); }
   enum Item_result cmp_type () const
   { return Type_handler_hybrid_field_type::cmp_type(); }
+  uint cols() { return this_item()->cols(); }
+  Item* element_index(uint i) { return this_item()->element_index(i); }
+  Item** addr(uint i) { return this_item()->addr(i); }
+  bool check_cols(uint c);
 
 private:
   bool set_value(THD *thd, sp_rcontext *ctx, Item **it);
@@ -2123,6 +2218,20 @@ public:
 };
 
 
+class Item_splocal_row: public Item_splocal
+{
+public:
+  Item_splocal_row(THD *thd, const LEX_STRING &sp_var_name,
+                   uint sp_var_idx, uint pos_in_q, uint len_in_q)
+   :Item_splocal(thd, sp_var_name, sp_var_idx, MYSQL_TYPE_NULL,
+                 pos_in_q, len_in_q)
+  {
+    set_handler(&type_handler_row);
+  }
+  enum Type type() const { return ROW_ITEM; }
+};
+
+
 /**
   An Item_splocal variant whose data type becomes known only at
   sp_rcontext creation time, e.g. "DECLARE var1 t1.col1%TYPE".
@@ -2156,6 +2265,37 @@ public:
   { return tmp_table_field_from_field_type(table, false, true); }
 };
 
+
+/**
+  SP variables that are fields of a ROW.
+  DELCARE r ROW(a INT,b INT);
+  SELECT r.a; -- This is handled by Item_splocal_row_field
+*/
+class Item_splocal_row_field :public Item_splocal
+{
+  LEX_STRING m_field_name;
+  uint m_field_idx;
+  bool set_value(THD *thd, sp_rcontext *ctx, Item **it);
+public:
+  Item_splocal_row_field(THD *thd,
+                         const LEX_STRING &sp_var_name,
+                         const LEX_STRING &sp_field_name,
+                         uint sp_var_idx, uint sp_field_idx,
+                         enum_field_types sp_var_type,
+                         uint pos_in_q= 0, uint len_in_q= 0)
+   :Item_splocal(thd, sp_var_name, sp_var_idx, sp_var_type,
+                 pos_in_q, len_in_q),
+    m_field_name(sp_field_name),
+    m_field_idx(sp_field_idx)
+  { }
+  Item *this_item();
+  const Item *this_item() const;
+  Item **this_item_addr(THD *thd, Item **);
+  bool append_for_log(THD *thd, String *str);
+  void print(String *str, enum_query_type query_type);
+};
+
+
 /*****************************************************************************
   Item_splocal inline implementation.
 *****************************************************************************/
@@ -2599,6 +2739,39 @@ public:
 };
 
 
+/**
+  Item_field for the ROW data type
+*/
+class Item_field_row: public Item_field,
+                      public Item_spvar_args
+{
+public:
+  Item_field_row(THD *thd, Field *field)
+   :Item_field(thd, field),
+    Item_spvar_args()
+  { }
+
+  Item *get_copy(THD *thd, MEM_ROOT *mem_root)
+  { return get_item_copy<Item_field_row>(thd, mem_root, this); }
+
+  const Type_handler *type_handler() const { return &type_handler_row; }
+  Item_result result_type() const{ return ROW_RESULT ; }
+  Item_result cmp_type() const { return ROW_RESULT; }
+  uint cols() { return arg_count; }
+  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)
+  {
+    if (cols() != c)
+    {
+      my_error(ER_OPERAND_COLUMNS, MYF(0), c);
+      return true;
+    }
+    return false;
+  }
+};
+
+
 /*
   @brief 
     Item_temptable_field is the same as Item_field, except that print() 
@@ -3785,78 +3958,6 @@ public:
 };
 
 
-/**
-  Array of items, e.g. function or aggerate function arguments.
-*/
-class Item_args
-{
-protected:
-  Item **args, *tmp_arg[2];
-  uint arg_count;
-  void set_arguments(THD *thd, List<Item> &list);
-  bool walk_args(Item_processor processor, bool walk_subquery, void *arg)
-  {
-    for (uint i= 0; i < arg_count; i++)
-    {
-      if (args[i]->walk(processor, walk_subquery, arg))
-        return true;
-    }
-    return false;
-  }
-  bool transform_args(THD *thd, Item_transformer transformer, uchar *arg);
-  void propagate_equal_fields(THD *, const Item::Context &, COND_EQUAL *);
-public:
-  Item_args(void)
-    :args(NULL), arg_count(0)
-  { }
-  Item_args(Item *a)
-    :args(tmp_arg), arg_count(1)
-  {
-    args[0]= a;
-  }
-  Item_args(Item *a, Item *b)
-    :args(tmp_arg), arg_count(2)
-  {
-    args[0]= a; args[1]= b;
-  }
-  Item_args(THD *thd, Item *a, Item *b, Item *c)
-  {
-    arg_count= 0;
-    if ((args= (Item**) thd_alloc(thd, sizeof(Item*) * 3)))
-    {
-      arg_count= 3;
-      args[0]= a; args[1]= b; args[2]= c;
-    }
-  }
-  Item_args(THD *thd, Item *a, Item *b, Item *c, Item *d)
-  {
-    arg_count= 0;
-    if ((args= (Item**) thd_alloc(thd, sizeof(Item*) * 4)))
-    {
-      arg_count= 4;
-      args[0]= a; args[1]= b; args[2]= c; args[3]= d;
-    }
-  }
-  Item_args(THD *thd, Item *a, Item *b, Item *c, Item *d, Item* e)
-  {
-    arg_count= 5;
-    if ((args= (Item**) thd_alloc(thd, sizeof(Item*) * 5)))
-    {
-      arg_count= 5;
-      args[0]= a; args[1]= b; args[2]= c; args[3]= d; args[4]= e;
-    }
-  }
-  Item_args(THD *thd, List<Item> &list)
-  {
-    set_arguments(thd, list);
-  }
-  Item_args(THD *thd, const Item_args *other);
-  inline Item **arguments() const { return args; }
-  inline uint argument_count() const { return arg_count; }
-  inline void remove_arguments() { arg_count=0; }
-};
-
-
 class Used_tables_and_const_cache
 {
 public:
diff --git a/sql/item_func.cc b/sql/item_func.cc
index e981722a123..7eeb9547c89 100644
--- a/sql/item_func.cc
+++ b/sql/item_func.cc
@@ -89,23 +89,35 @@ static inline bool test_if_sum_overflows_ull(ulonglong arg1, ulonglong arg2)
 }
 
 
-void Item_args::set_arguments(THD *thd, List<Item> &list)
+/**
+  Allocate memory for arguments using tmp_args or thd->alloc().
+  @retval false  - success
+  @retval true   - error (arg_count is set to 0 for conveniece)
+*/
+bool Item_args::alloc_arguments(THD *thd, uint count)
 {
-  arg_count= list.elements;
-  if (arg_count <= 2)
+  if (count <= 2)
   {
     args= tmp_arg;
+    return false;
   }
-  else if (!(args= (Item**) thd->alloc(sizeof(Item*) * arg_count)))
+  if ((args= (Item**) thd->alloc(sizeof(Item*) * count)) == NULL)
   {
     arg_count= 0;
-    return;
+    return true;
   }
-  uint i= 0;
+  return false;
+}
+
+
+void Item_args::set_arguments(THD *thd, List<Item> &list)
+{
+  if (alloc_arguments(thd, list.elements))
+    return;
   List_iterator_fast<Item> li(list);
   Item *item;
-  while ((item= li++))
-    args[i++]= item;
+  for (arg_count= 0; (item= li++); )
+    args[arg_count++]= item;
 }
 
 
@@ -138,6 +150,19 @@ void Item_func::sync_with_sum_func_and_with_field(List<Item> &list)
 }
 
 
+bool Item_func::check_allowed_arg_cols(uint n)
+{
+  if (allowed_arg_cols)
+    return args[n]->check_cols(allowed_arg_cols);
+
+  /*  we have to fetch allowed_arg_cols from first argument */
+  DBUG_ASSERT(n == 0); // it is first argument
+  allowed_arg_cols= args[n]->cols();
+  DBUG_ASSERT(allowed_arg_cols); // Can't be 0 any more
+  return false;
+}
+
+
 /*
   Resolve references to table column for a function and its argument
 
@@ -210,18 +235,8 @@ Item_func::fix_fields(THD *thd, Item **ref)
 	return TRUE;				/* purecov: inspected */
       item= *arg;
 
-      if (allowed_arg_cols)
-      {
-        if (item->check_cols(allowed_arg_cols))
-          return 1;
-      }
-      else
-      {
-        /*  we have to fetch allowed_arg_cols from first argument */
-        DBUG_ASSERT(arg == args); // it is first argument
-        allowed_arg_cols= item->cols();
-        DBUG_ASSERT(allowed_arg_cols); // Can't be 0 any more
-      }
+      if (check_allowed_arg_cols(arg - args))
+        return true;
 
       if (item->maybe_null)
 	maybe_null=1;
diff --git a/sql/item_func.h b/sql/item_func.h
index 88fa78cd4a4..5b39c2ef0d8 100644
--- a/sql/item_func.h
+++ b/sql/item_func.h
@@ -53,6 +53,7 @@ protected:
       set_if_bigger(res, item[i]->decimals);
     return res;
   }
+  virtual bool check_allowed_arg_cols(uint argno);
 public:
   void aggregate_attributes_int(Item **items, uint nitems)
   {
@@ -2547,7 +2548,12 @@ private:
 protected:
   bool is_expensive_processor(void *arg)
   { return is_expensive(); }
-  
+
+  bool check_allowed_arg_cols(uint n)
+  {
+    // sp_prepare_func_item() checks that the number of columns is correct
+    return false;
+  } 
 public:
 
   Item_func_sp(THD *thd, Name_resolution_context *context_arg, sp_name *name);
diff --git a/sql/share/errmsg-utf8.txt b/sql/share/errmsg-utf8.txt
index 3f4299e2166..8b0bd18e95a 100644
--- a/sql/share/errmsg-utf8.txt
+++ b/sql/share/errmsg-utf8.txt
@@ -7460,3 +7460,5 @@ ER_ILLEGAL_PARAMETER_DATA_TYPE_FOR_OPERATION
 ER_WRONG_PARAMCOUNT_TO_CURSOR 42000
         eng "Incorrect parameter count to cursor '%-.192s'"
         rus "Некорректное количество параметров для курсора '%-.192s'"
+ER_UNKNOWN_STRUCTURED_VARIABLE
+        eng "Unknown structured system variable or ROW routine variable '%-.*s'"
diff --git a/sql/sp_head.cc b/sql/sp_head.cc
index 7fd2624ab7a..471e6d46828 100644
--- a/sql/sp_head.cc
+++ b/sql/sp_head.cc
@@ -100,19 +100,51 @@ bool Item_splocal::append_for_log(THD *thd, String *str)
   if (limit_clause_param)
     return str->append_ulonglong(val_uint());
 
+  /*
+    ROW variables are currently not allowed in select_list, e.g.:
+      SELECT row_variable;
+    ROW variables can appear in query parts where name is not important, e.g.:
+      SELECT ROW(1,2)=row_variable FROM t1;
+    So we can skip using NAME_CONST() and use ROW() constants directly.
+  */
+  if (type_handler() == &type_handler_row)
+    return append_value_for_log(thd, str);
+
   if (str->append(STRING_WITH_LEN(" NAME_CONST('")) ||
       str->append(&m_name) ||
       str->append(STRING_WITH_LEN("',")))
     return true;
+  return append_value_for_log(thd, str) || str->append(')');
+}
+
 
+bool Item_splocal::append_value_for_log(THD *thd, String *str)
+{
   StringBuffer<STRING_BUFFER_USUAL_SIZE> str_value_holder(&my_charset_latin1);
   Item *item= this_item();
   String *str_value= item->type_handler()->print_item_value(thd, item,
                                                             &str_value_holder);
-  if (str_value)
-    return str->append(*str_value) || str->append(')');
-  else
-    return str->append(STRING_WITH_LEN("NULL)"));
+  return str_value ?
+         str->append(*str_value) :
+         str->append(STRING_WITH_LEN("NULL"));
+}
+
+
+bool Item_splocal_row_field::append_for_log(THD *thd, String *str)
+{
+  if (fix_fields(thd, NULL))
+    return true;
+
+  if (limit_clause_param)
+    return str->append_ulonglong(val_uint());
+
+  if (str->append(STRING_WITH_LEN(" NAME_CONST('")) ||
+      str->append(&m_name) ||
+      str->append(".") ||
+      str->append(&m_field_name) ||
+      str->append(STRING_WITH_LEN("',")))
+    return true;
+  return append_value_for_log(thd, str) || str->append(')');
 }
 
 
@@ -308,14 +340,14 @@ sp_get_flags_for_command(LEX *lex)
 */
 
 Item *
-sp_prepare_func_item(THD* thd, Item **it_addr)
+sp_prepare_func_item(THD* thd, Item **it_addr, uint cols)
 {
   DBUG_ENTER("sp_prepare_func_item");
   it_addr= (*it_addr)->this_item_addr(thd, it_addr);
 
-  if (!(*it_addr)->fixed &&
-      ((*it_addr)->fix_fields(thd, it_addr) ||
-       (*it_addr)->check_cols(1)))
+  if ((!(*it_addr)->fixed &&
+       (*it_addr)->fix_fields(thd, it_addr)) ||
+      (*it_addr)->check_cols(cols))
   {
     DBUG_PRINT("info", ("fix_fields() failed"));
     DBUG_RETURN(NULL);
@@ -338,7 +370,8 @@ sp_prepare_func_item(THD* thd, Item **it_addr)
 */
 
 bool
-sp_eval_expr(THD *thd, Field *result_field, Item **expr_item_ptr)
+sp_eval_expr(THD *thd, Item *result_item, Field *result_field,
+             Item **expr_item_ptr)
 {
   Item *expr_item;
   enum_check_fields save_count_cuted_fields= thd->count_cuted_fields;
@@ -351,10 +384,21 @@ sp_eval_expr(THD *thd, Field *result_field, Item **expr_item_ptr)
   if (!*expr_item_ptr)
     goto error;
 
-  if (!(expr_item= sp_prepare_func_item(thd, expr_item_ptr)))
+  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 && 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.
 
@@ -3185,6 +3229,56 @@ sp_instr_set::print(String *str)
 
 
 /*
+  sp_instr_set_field class functions
+*/
+
+int
+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;
+}
+
+
+void
+sp_instr_set_row_field::print(String *str)
+{
+  /* set name@offset[field_offset] ... */
+  int rsrv= SP_INSTR_UINT_MAXLEN + 6 + 6 + 3;
+  sp_variable *var= m_ctx->find_variable(m_offset);
+  DBUG_ASSERT(var);
+  DBUG_ASSERT(var->field_def.is_row());
+  const Column_definition *def=
+    var->field_def.row_field_definitions()->elem(m_field_offset);
+  DBUG_ASSERT(def);
+
+  rsrv+= var->name.length + strlen(def->field_name);
+  if (str->reserve(rsrv))
+    return;
+  str->qs_append(STRING_WITH_LEN("set "));
+    str->qs_append(var->name.str, var->name.length);
+  str->qs_append('.');
+  str->qs_append(def->field_name);
+  str->qs_append('@');
+  str->qs_append(m_offset);
+  str->qs_append('[');
+  str->qs_append(m_field_offset);
+  str->qs_append(']');
+  str->qs_append(' ');
+  m_value->print(str, enum_query_type(QT_ORDINARY |
+                                      QT_ITEM_ORIGINAL_FUNC_NULLIF));
+}
+
+
+/*
   sp_instr_set_trigger_field class functions
 */
 
@@ -4209,6 +4303,11 @@ sp_add_to_query_tables(THD *thd, LEX *lex,
 }
 
 
+Item *sp_head::adjust_assignment_source(THD *thd, Item *val, Item *val2)
+{
+  return val ? val : val2 ? val2 : new (thd->mem_root) Item_null(thd);
+}
+
 /**
   Helper action for a SET statement.
   Used to push a SP local variable into the assignment list.
@@ -4223,28 +4322,40 @@ bool
 sp_head::set_local_variable(THD *thd, sp_pcontext *spcont,
                             sp_variable *spv, Item *val, LEX *lex)
 {
-  Item *it;
-
-  if (val)
-    it= val;
-  else if (spv->default_value)
-    it= spv->default_value;
-  else
-  {
-    it= new (thd->mem_root) Item_null(thd);
-    if (it == NULL)
-      return TRUE;
-  }
+  if (!(val= adjust_assignment_source(thd, val, spv->default_value)))
+    return true;
 
   sp_instr_set *sp_set= new (thd->mem_root)
                         sp_instr_set(instructions(), spcont,
-                                     spv->offset, it, spv->sql_type(),
+                                     spv->offset, val, spv->sql_type(),
                                      lex, true);
 
   return sp_set == NULL || add_instr(sp_set);
 }
 
 
+/**
+  Similar to set_local_variable(), but for ROW variable fields.
+*/
+bool
+sp_head::set_local_variable_row_field(THD *thd, sp_pcontext *spcont,
+                                      sp_variable *spv, uint field_idx,
+                                      Item *val, LEX *lex)
+{
+  if (!(val= adjust_assignment_source(thd, val, NULL)))
+    return true;
+
+  sp_instr_set_row_field *sp_set= new (thd->mem_root)
+                                  sp_instr_set_row_field(instructions(),
+                                                         spcont,
+                                                         spv->offset,
+                                                         field_idx, val,
+                                                         spv->sql_type(),
+                                                         lex, true);
+  return sp_set == NULL || add_instr(sp_set);
+}
+
+
 bool sp_head::add_open_cursor(THD *thd, sp_pcontext *spcont, uint offset,
                               sp_pcontext *param_spcont,
                               List<sp_assignment_lex> *parameters)
diff --git a/sql/sp_head.h b/sql/sp_head.h
index ec19058a7f1..fc3dfa49f63 100644
--- a/sql/sp_head.h
+++ b/sql/sp_head.h
@@ -360,9 +360,12 @@ public:
                                                  spcont->last_label());
   }
 
+  Item *adjust_assignment_source(THD *thd, Item *val, Item *val2);
   bool set_local_variable(THD *thd, sp_pcontext *spcont,
                           sp_variable *spv, Item *val, LEX *lex);
-
+  bool set_local_variable_row_field(THD *thd, sp_pcontext *spcont,
+                                    sp_variable *spv, uint field_idx,
+                                    Item *val, LEX *lex);
 private:
   /**
     Generate a code to set a single cursor parameter variable.
@@ -992,7 +995,7 @@ public:
 
   virtual void print(String *str);
 
-private:
+protected:
 
   uint m_offset;		///< Frame offset
   Item *m_value;
@@ -1002,6 +1005,36 @@ private:
 }; // class sp_instr_set : public sp_instr
 
 
+/*
+  This class handles assignments of a ROW fields:
+    DECLARE rec ROW (a INT,b INT);
+    SET rec.a= 10;
+*/
+class sp_instr_set_row_field : public sp_instr_set
+{
+  sp_instr_set_row_field(const sp_instr_set_row_field &); // Prevent use of this
+  void operator=(sp_instr_set_row_field &);
+  uint m_field_offset;
+
+public:
+
+  sp_instr_set_row_field(uint ip, sp_pcontext *ctx,
+                         uint offset, uint field_offset,
+                         Item *val, enum enum_field_types type_arg,
+                         LEX *lex, bool lex_resp)
+    : sp_instr_set(ip, ctx, offset, val, type_arg, lex, lex_resp),
+      m_field_offset(field_offset)
+  {}
+
+  virtual ~sp_instr_set_row_field()
+  {}
+
+  virtual int exec_core(THD *thd, uint *nextp);
+
+  virtual void print(String *str);
+}; // class sp_instr_set_field : public sp_instr_set
+
+
 /**
   Set NEW/OLD row field value instruction. Used in triggers.
 */
@@ -1603,10 +1636,11 @@ sp_add_to_query_tables(THD *thd, LEX *lex,
                        enum_mdl_type mdl_type);
 
 Item *
-sp_prepare_func_item(THD* thd, Item **it_addr);
+sp_prepare_func_item(THD* thd, Item **it_addr, uint cols= 1);
 
 bool
-sp_eval_expr(THD *thd, Field *result_field, Item **expr_item_ptr);
+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_pcontext.cc b/sql/sp_pcontext.cc
index 642f1f16d29..06c51642d68 100644
--- a/sql/sp_pcontext.cc
+++ b/sql/sp_pcontext.cc
@@ -563,3 +563,24 @@ const sp_pcursor *sp_pcontext::find_cursor(uint offset) const
          m_parent->find_cursor(offset) :  // Some previous frame
          NULL;                            // Index out of bounds
 }
+
+
+const Spvar_definition *
+sp_variable::find_row_field(const LEX_STRING &var_name,
+                            const LEX_STRING &field_name,
+                            uint *row_field_offset)
+{
+  if (!field_def.is_row())
+  {
+    my_printf_error(ER_UNKNOWN_ERROR,
+                    "'%s' is not a row variable", MYF(0), var_name.str);
+    return NULL;
+  }
+  const Spvar_definition *def;
+  if ((def= field_def.find_row_field_by_name(field_name.str, row_field_offset)))
+    return def;
+  my_printf_error(ER_UNKNOWN_ERROR,
+                  "Row variable '%s' does not have a field '%s'",
+                  MYF(0), var_name.str, field_name.str);
+  return NULL;
+}
diff --git a/sql/sp_pcontext.h b/sql/sp_pcontext.h
index 414a3cbc7cc..00cc81964eb 100644
--- a/sql/sp_pcontext.h
+++ b/sql/sp_pcontext.h
@@ -69,6 +69,20 @@ public:
     offset(_offset),
     default_value(NULL)
   { }
+  /*
+    Find a ROW field by its qualified name.
+    @param      var_name - the name of the variable
+    @param      field_name - the name of the variable field
+    @param[OUT] row_field_offset - the index of the field
+
+    @retval  NULL if the variable with the given name was not found,
+             or it is not a row variable, or it does not have a field
+             with the given name, or a non-null pointer otherwise.
+             row_field_offset[0] is set only when the method returns !NULL.
+  */
+  const Spvar_definition *find_row_field(const LEX_STRING &var_name,
+                                         const LEX_STRING &field_name,
+                                         uint *row_field_offset);
 };
 
 ///////////////////////////////////////////////////////////////////////////
diff --git a/sql/sp_rcontext.cc b/sql/sp_rcontext.cc
index 5fcdbf42313..17c91b09727 100644
--- a/sql/sp_rcontext.cc
+++ b/sql/sp_rcontext.cc
@@ -222,23 +222,65 @@ bool sp_rcontext::init_var_items(THD *thd,
   if (!m_var_items.array())
     return true;
 
-  for (uint idx = 0; idx < num_vars; ++idx)
+  DBUG_ASSERT(field_def_lst.elements == num_vars);
+  List_iterator<Spvar_definition> it(field_def_lst);
+  Spvar_definition *def= it++;
+
+  for (uint idx= 0; idx < num_vars; ++idx, def= it++)
   {
-    if (!(m_var_items[idx]= new (thd->mem_root) Item_field(thd, m_var_table->field[idx])))
-      return true;
+    Field *field= m_var_table->field[idx];
+    if (def->is_row())
+    {
+      Item_field_row *item= new (thd->mem_root) Item_field_row(thd, field);
+      if (!(m_var_items[idx]= item) ||
+          item->row_create_items(thd, def->row_field_definitions()))
+        return true;
+    }
+    else
+    {
+      if (!(m_var_items[idx]= new (thd->mem_root) Item_field(thd, field)))
+        return true;
+    }
   }
+  return false;
+}
+
+
+bool Item_spvar_args::row_create_items(THD *thd, List<Spvar_definition> *list)
+{
+  DBUG_ASSERT(list);
+  if (!(m_table= create_virtual_tmp_table(thd, *list)))
+    return true;
+
+  if (alloc_arguments(thd, list->elements))
+    return true;
 
+  List_iterator<Spvar_definition> it(*list);
+  Spvar_definition *def;
+  for (arg_count= 0; (def= it++); arg_count++)
+  {
+    if (!(args[arg_count]= new (thd->mem_root)
+                           Item_field(thd, m_table->field[arg_count])))
+      return true;
+  }
   return false;
 }
 
 
+Item_spvar_args::~Item_spvar_args()
+{
+  if (m_table)
+    free_blobs(m_table);
+}
+
+
 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, m_return_value_fld, return_value_item);
+  return sp_eval_expr(thd, NULL, m_return_value_fld, return_value_item);
 }
 
 
@@ -455,8 +497,78 @@ int sp_rcontext::set_variable(THD *thd, uint idx, Item **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();
+}
+
 
-  return sp_eval_expr(thd, field, value);
+int sp_rcontext::set_variable_row_field(THD *thd, uint var_idx, uint field_idx,
+                                        Item **value)
+{
+  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;
+
+  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);
 }
 
 
diff --git a/sql/sp_rcontext.h b/sql/sp_rcontext.h
index 061f736f6e9..78cbfd9ec26 100644
--- a/sql/sp_rcontext.h
+++ b/sql/sp_rcontext.h
@@ -187,7 +187,9 @@ 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);
   Item *get_item(uint var_idx) const
   { return m_var_items[var_idx]; }
 
diff --git a/sql/sql_class.cc b/sql/sql_class.cc
index 65a4dabb85b..5ddb0a86592 100644
--- a/sql/sql_class.cc
+++ b/sql/sql_class.cc
@@ -3839,6 +3839,11 @@ bool my_var_sp::set(THD *thd, Item *item)
   return thd->spcont->set_variable(thd, offset, &item);
 }
 
+bool my_var_sp_row_field::set(THD *thd, Item *item)
+{
+  return thd->spcont->set_variable_row_field(thd, offset, m_field_offset, &item);
+}
+
 int select_dumpvar::send_data(List<Item> &items)
 {
   List_iterator_fast<my_var> var_li(var_list);
diff --git a/sql/sql_class.h b/sql/sql_class.h
index cac1cdd8fd3..1cc2e22bb17 100644
--- a/sql/sql_class.h
+++ b/sql/sql_class.h
@@ -5522,6 +5522,22 @@ public:
   bool set(THD *thd, Item *val);
 };
 
+/*
+  This class handles fields of a ROW SP variable when it's used as a OUT
+  parameter in a stored procedure.
+*/
+class my_var_sp_row_field: public my_var_sp
+{
+  uint m_field_offset;
+public:
+  my_var_sp_row_field(const LEX_STRING &varname, const LEX_STRING &fieldname,
+                      uint var_idx, uint field_idx, sp_head *s)
+   :my_var_sp(varname, var_idx, MYSQL_TYPE_DOUBLE/*Not really used*/, s),
+    m_field_offset(field_idx)
+  { }
+  bool set(THD *thd, Item *val);
+};
+
 class my_var_user: public my_var {
 public:
   my_var_user(const LEX_STRING& j)
diff --git a/sql/sql_lex.cc b/sql/sql_lex.cc
index 83983a1a357..b0647f6b404 100644
--- a/sql/sql_lex.cc
+++ b/sql/sql_lex.cc
@@ -5153,7 +5153,8 @@ bool LEX::init_internal_variable(struct sys_var_with_base *variable,
 {
   if (check_reserved_words(&dbname))
   {
-    thd->parse_error();
+    my_error(ER_UNKNOWN_STRUCTURED_VARIABLE, MYF(0),
+             (int) dbname.length, dbname.str);
     return true;
   }
   if (is_trigger_new_or_old_reference(dbname))
@@ -5179,9 +5180,13 @@ bool LEX::init_internal_variable(struct sys_var_with_base *variable,
     return false;
   }
 
-  sys_var *tmp= find_sys_var(thd, name.str, name.length);
+  sys_var *tmp= find_sys_var_ex(thd, name.str, name.length, true, false);
   if (!tmp)
+  {
+    my_error(ER_UNKNOWN_STRUCTURED_VARIABLE, MYF(0),
+             (int) dbname.length, dbname.str);
     return true;
+  }
   if (!tmp->is_struct())
     my_error(ER_VARIABLE_IS_NOT_STRUCT, MYF(0), name.str);
   variable->var= tmp;
@@ -5219,6 +5224,7 @@ void LEX::sp_variable_declarations_init(THD *thd, int nvars)
 
 bool LEX::sp_variable_declarations_finalize(THD *thd, int nvars,
                                             const Column_definition *cdef,
+                                            Row_definition_list *row,
                                             Item *dflt_value_item)
 {
   uint num_vars= spcont->context_var_count();
@@ -5226,7 +5232,31 @@ bool LEX::sp_variable_declarations_finalize(THD *thd, int nvars,
   if (!dflt_value_item &&
       !(dflt_value_item= new (thd->mem_root) Item_null(thd)))
     return true;
-  /* QQ Set to the var_type with null_value? */
+
+  if (row)
+  {
+    /*
+      Prepare all row fields. This will (among other things)
+      - convert VARCHAR lengths from character length to octet length
+      - calculate interval lengths for SET and ENUM
+      Note, we do it only one time outside of the below loop.
+      The converted list in "row" is further reused by all variable
+      declarations processed by the current call.
+      Example:
+        DECLARE
+          a, b, c ROW(x VARCHAR(10) CHARACTER SET utf8);
+        BEGIN
+          ...
+        END;
+    */
+    List_iterator<Spvar_definition> it(*row);
+    for (Spvar_definition *def= it++; def; def= it++)
+    {
+      def->pack_flag|= FIELDFLAG_MAYBE_NULL;
+      if (sphead->fill_field_definition(thd, def))
+        return true;
+    }
+  }
 
   for (uint i= num_vars - nvars ; i < num_vars ; i++)
   {
@@ -5246,6 +5276,7 @@ bool LEX::sp_variable_declarations_finalize(THD *thd, int nvars,
       if (sphead->fill_spvar_definition(thd, &spvar->field_def, spvar->name.str))
         return true;
     }
+    spvar->field_def.set_row_field_definitions(row);
 
     /* The last instruction is responsible for freeing LEX. */
     sp_instr_set *is= new (this->thd->mem_root)
@@ -5274,7 +5305,7 @@ LEX::sp_variable_declarations_with_ref_finalize(THD *thd, int nvars,
     spvar->field_def.field_name= spvar->name.str;
   }
   sphead->m_flags|= sp_head::HAS_COLUMN_TYPE_REFS;
-  return sp_variable_declarations_finalize(thd, nvars, NULL, def);
+  return sp_variable_declarations_finalize(thd, nvars, NULL, NULL, def);
 }
 
 
@@ -5315,7 +5346,7 @@ sp_variable *LEX::sp_add_for_loop_variable(THD *thd, const LEX_STRING name,
   */
   spvar->field_def.pack_flag= (FIELDFLAG_NUMBER |
                                f_settype((uint) MYSQL_TYPE_LONGLONG));
-  if (sp_variable_declarations_finalize(thd, 1, NULL, value))
+  if (sp_variable_declarations_finalize(thd, 1, NULL, NULL, value))
     return NULL;
   return spvar;
 }
@@ -5953,11 +5984,9 @@ Item_param *LEX::add_placeholder(THD *thd, char *name,
 }
 
 
-Item_param *LEX::add_placeholder(THD *thd, char *name,
-                                 const char *pos, const char *end)
+const char *LEX::substatement_query(THD *thd) const
 {
-  const char *query_start= sphead ? sphead->m_tmp_query : thd->query();
-  return add_placeholder(thd, name, pos - query_start, end - pos);
+  return sphead ? sphead->m_tmp_query : thd->query();
 }
 
 
@@ -5979,6 +6008,160 @@ bool LEX::add_resignal_statement(THD *thd, const sp_condition_value *v)
 }
 
 
+Item *LEX::create_item_ident_nospvar(THD *thd,
+                                     const LEX_STRING &a,
+                                     const LEX_STRING &b)
+{
+  DBUG_ASSERT(this == thd->lex);
+  /*
+    FIXME This will work ok in simple_ident_nospvar case because
+    we can't meet simple_ident_nospvar in trigger now. But it
+    should be changed in future.
+  */
+  if (is_trigger_new_or_old_reference(a))
+  {
+    bool new_row= (a.str[0]=='N' || a.str[0]=='n');
+
+    return create_and_link_Item_trigger_field(thd, b.str, new_row);
+  }
+
+  if (current_select->no_table_names_allowed)
+  {
+    my_error(ER_TABLENAME_NOT_ALLOWED_HERE, MYF(0), a.str, thd->where);
+    return NULL;
+  }
+  if ((current_select->parsing_place != IN_HAVING) ||
+      (current_select->get_in_sum_expr() > 0))
+    return new (thd->mem_root) Item_field(thd, current_context(),
+                                          NullS, a.str, b.str);
+  return new (thd->mem_root) Item_ref(thd, current_context(),
+                                      NullS, a.str, b.str);
+}
+
+
+Item_splocal_row_field *LEX::create_item_spvar_row_field(THD *thd,
+                                                         const LEX_STRING &a,
+                                                         const LEX_STRING &b,
+                                                         sp_variable *spv,
+                                                         uint pos_in_q,
+                                                         uint length_in_q)
+{
+  if (!parsing_options.allows_variable)
+  {
+    my_error(ER_VIEW_SELECT_VARIABLE, MYF(0));
+    return NULL;
+  }
+
+  uint row_field_offset;
+  const Spvar_definition *def;
+  if (!(def= spv->find_row_field(a, b, &row_field_offset)))
+    return NULL;
+
+  Item_splocal_row_field *item;
+  if (!(item= new (thd->mem_root)
+              Item_splocal_row_field(thd, a, b,
+                                     spv->offset, row_field_offset,
+                                     def->sql_type, pos_in_q, length_in_q)))
+    return NULL;
+#ifndef DBUG_OFF
+  item->m_sp= sphead;
+#endif
+  safe_to_cache_query=0;
+  return item;
+}
+
+
+my_var *LEX::create_outvar(THD *thd,
+                           const LEX_STRING &a,
+                           const LEX_STRING &b)
+{
+  sp_variable *t;
+  if (!spcont || !(t= spcont->find_variable(a, false)))
+  {
+    my_error(ER_SP_UNDECLARED_VAR, MYF(0), a.str);
+    return NULL;
+  }
+  uint row_field_offset;
+  if (!t->find_row_field(a, b, &row_field_offset))
+    return NULL;
+  return result ?
+    new (thd->mem_root) my_var_sp_row_field(a, b, t->offset,
+                                            row_field_offset, sphead) :
+    NULL;
+}
+
+
+Item *LEX::create_item_ident(THD *thd,
+                             const LEX_STRING &a,
+                             const LEX_STRING &b,
+                             uint pos_in_q, uint length_in_q)
+{
+  sp_variable *spv;
+  if (spcont && (spv= spcont->find_variable(a, false)))
+    return create_item_spvar_row_field(thd, a, b, spv, pos_in_q, length_in_q);
+  return create_item_ident_nospvar(thd, a, b);
+}
+
+
+
+Item *LEX::create_item_limit(THD *thd,
+                             const LEX_STRING &a,
+                             uint pos_in_q, uint length_in_q)
+{
+  sp_variable *spv;
+  if (!spcont || !(spv= spcont->find_variable(a, false)))
+  {
+    my_error(ER_SP_UNDECLARED_VAR, MYF(0), a.str);
+    return NULL;
+  }
+
+  Item_splocal *item;
+  if (!(item= new (thd->mem_root) Item_splocal(thd, a,
+                                               spv->offset, spv->sql_type(),
+                                               pos_in_q, length_in_q)))
+    return NULL;
+#ifndef DBUG_OFF
+  item->m_sp= sphead;
+#endif
+  safe_to_cache_query= 0;
+
+  if (item->type() != Item::INT_ITEM)
+  {
+    my_error(ER_WRONG_SPVAR_TYPE_IN_LIMIT, MYF(0));
+    return NULL;
+  }
+  item->limit_clause_param= true;
+  return item;
+}
+
+
+Item *LEX::create_item_limit(THD *thd,
+                             const LEX_STRING &a,
+                             const LEX_STRING &b,
+                             uint pos_in_q, uint length_in_q)
+{
+  sp_variable *spv;
+  if (!spcont || !(spv= spcont->find_variable(a, false)))
+  {
+    my_error(ER_SP_UNDECLARED_VAR, MYF(0), a.str);
+    return NULL;
+  }
+  // Qualified %TYPE variables are not possible
+  DBUG_ASSERT(!spv->field_def.column_type_ref());
+  Item_splocal *item;
+  if (!(item= create_item_spvar_row_field(thd, a, b, spv,
+                                          pos_in_q, length_in_q)))
+    return NULL;
+  if (item->type() != Item::INT_ITEM)
+  {
+    my_error(ER_WRONG_SPVAR_TYPE_IN_LIMIT, MYF(0));
+    return NULL;
+  }
+  item->limit_clause_param= true;
+  return item;
+}
+
+
 /*
   Perform assignment for a trigger, a system variable, or an SP variable.
   "variable" be previously set by init_internal_variable(variable, name).
@@ -6040,6 +6223,9 @@ Item *LEX::create_item_ident_sp(THD *thd, LEX_STRING name,
                                                               spv->offset,
                                                               start_in_q,
                                                               length_in_q) :
+      spv->field_def.is_row() ?
+      new (thd->mem_root) Item_splocal_row(thd, name, spv->offset,
+                                           start_in_q, length_in_q) :
       new (thd->mem_root) Item_splocal(thd, name,
                                        spv->offset, spv->sql_type(),
                                        start_in_q, length_in_q);
@@ -6073,6 +6259,32 @@ Item *LEX::create_item_ident_sp(THD *thd, LEX_STRING name,
 }
 
 
+/**
+  Generate instructions for:
+    SET x.y= expr;
+*/
+bool LEX::set_variable(const LEX_STRING &name1,
+                       const LEX_STRING &name2,
+                       Item *item)
+{
+  sp_variable *spv;
+  if (spcont && (spv= spcont->find_variable(name1, false)))
+  {
+    // A field of a ROW variable
+    uint row_field_offset;
+    return !spv->find_row_field(name1, name2, &row_field_offset) ||
+           sphead->set_local_variable_row_field(thd, spcont,
+                                                spv, row_field_offset,
+                                                item, this);
+  }
+
+  // A trigger field or a system variable
+  sys_var_with_base sysvar;
+  return init_internal_variable(&sysvar, name1, name2) ||
+         set_variable(&sysvar, item);
+}
+
+
 #ifdef MYSQL_SERVER
 uint binlog_unsafe_map[256];
 
diff --git a/sql/sql_lex.h b/sql/sql_lex.h
index c3ef0710112..913478b9589 100644
--- a/sql/sql_lex.h
+++ b/sql/sql_lex.h
@@ -96,6 +96,7 @@ class Key_part_spec;
 class Item_window_func;
 struct sql_digest_state;
 class With_clause;
+class my_var;
 
 
 #define ALLOC_ROOT_SET 1024
@@ -2226,6 +2227,20 @@ public:
     return m_cpp_tok_end;
   }
 
+  /**
+    Get the token end position in the pre-processed buffer,
+    with trailing spaces removed.
+  */
+  const char *get_cpp_tok_end_rtrim()
+  {
+    const char *p;
+    for (p= m_cpp_tok_end;
+         p > m_cpp_buf && my_isspace(system_charset_info, p[-1]);
+         p--)
+    { }
+    return p;
+  }
+
   /** Get the current stream pointer, in the pre-processed buffer. */
   const char *get_cpp_ptr()
   {
@@ -2923,6 +2938,8 @@ public:
 
   void start(THD *thd);
 
+  const char *substatement_query(THD *thd) const;
+
   inline bool is_ps_or_view_context_analysis()
   {
     return (context_analysis_only &
@@ -3108,10 +3125,25 @@ public:
   bool init_default_internal_variable(struct sys_var_with_base *variable,
                                       LEX_STRING name);
   bool set_variable(struct sys_var_with_base *variable, Item *item);
+  bool set_variable(const LEX_STRING &name1, const LEX_STRING &name2,
+                    Item *item);
   void sp_variable_declarations_init(THD *thd, int nvars);
   bool sp_variable_declarations_finalize(THD *thd, int nvars,
                                          const Column_definition *cdef,
+                                         Row_definition_list *row,
                                          Item *def);
+  bool sp_variable_declarations_finalize(THD *thd, int nvars,
+                                         const Column_definition *cdef,
+                                         Item *def)
+  {
+    return sp_variable_declarations_finalize(thd, nvars, cdef, NULL, def);
+  }
+  bool sp_variable_declarations_row_finalize(THD *thd, int nvars,
+                                             Row_definition_list *row,
+                                             Item *def)
+  {
+    return sp_variable_declarations_finalize(thd, nvars, NULL, row, def);
+  }
   bool sp_variable_declarations_with_ref_finalize(THD *thd, int nvars,
                                                   Qualified_column_ident *col,
                                                   Item *def);
@@ -3141,6 +3173,98 @@ public:
            create_item_ident_sp(thd, name, start_in_q, end_in_q) :
            create_item_ident_nosp(thd, name);
   }
+
+  /*
+    Create an Item corresponding to a qualified name: a.b
+    when the parser is out of an SP context.
+      @param THD        - THD, for mem_root
+      @param a          - the first name
+      @param b          - the second name
+      @retval           - a pointer to a created item, or NULL on error.
+
+    Possible Item types that can be created:
+    - Item_trigger_field
+    - Item_field
+    - Item_ref
+  */
+  Item *create_item_ident_nospvar(THD *thd,
+                                  const LEX_STRING &a,
+                                  const LEX_STRING &b);
+  /*
+    Create an Item corresponding to a ROW field valiable:  var.field
+      @param THD        - THD, for mem_root
+      @param var        - the ROW variable name
+      @param field      - the ROW variable field name
+      @param spvar      - the variable that was previously found by name
+                          using "var_name".
+      @pos_in_q         - position in the query (for binary log)
+      @length_in_q      - length in the query (for binary log)
+  */
+  Item_splocal_row_field *create_item_spvar_row_field(THD *thd,
+                                                      const LEX_STRING &var,
+                                                      const LEX_STRING &field,
+                                                      sp_variable *spvar,
+                                                      uint pos_in_q,
+                                                      uint length_in_q);
+  /*
+    Create an item from its qualified name.
+    Depending on context, it can be either a ROW variable field,
+    or trigger, table field, table field reference.
+    See comments to create_item_spvar_row_field() and
+    create_item_ident_nospvar().
+      @param thd         - THD, for mem_root
+      @param a           - the first name
+      @param b           - the second name
+      @param pos_in_q    - position in the query (for binary log)
+      @param length_in_q - length in the query (for binary log)
+      @retval            - NULL on error, or a pointer to a new Item.
+  */
+  Item *create_item_ident(THD *thd,
+                          const LEX_STRING &a,
+                          const LEX_STRING &b,
+                          uint pos_in_q, uint length_in_q);
+  /*
+    Create an item for a name in LIMIT clause: LIMIT var
+      @param THD         - THD, for mem_root
+      @param var_name    - the variable name
+      @param pos_in_q    - position in the query (for binary log)
+      @param length_in_q - length in the query (for binary log)
+      @retval            - a new Item corresponding to the SP variable,
+                           or NULL on error
+                           (non in SP, unknown variable, wrong data type).
+  */
+  Item *create_item_limit(THD *thd,
+                          const LEX_STRING &var_name,
+                          uint pos_in_q, uint length_in_q);
+
+  /*
+    Create an item for a qualified name in LIMIT clause: LIMIT var.field
+      @param THD         - THD, for mem_root
+      @param var_name    - the variable name
+      @param field_name  - the variable field name
+      @param pos_in_q    - position in the query (for binary log)
+      @param length_in_q - length in the query (for binary log)
+      @retval            - a new Item corresponding to the SP variable,
+                           or NULL on error
+                           (non in SP, unknown variable, unknown ROW field,
+                            wrong data type).
+  */
+  Item *create_item_limit(THD *thd,
+                          const LEX_STRING &var_name,
+                          const LEX_STRING &field_name,
+                          uint pos_in_q, uint length_in_q);
+
+  /*
+    Create a my_var instance for a ROW field variable that was used
+    as an OUT SP parameter: CALL p1(var.field);
+      @param THD        - THD, for mem_root
+      @param var_name   - the variable name
+      @param field_name - the variable field name
+  */
+  my_var *create_outvar(THD *thd,
+                        const LEX_STRING &var_name,
+                        const LEX_STRING &field_name);
+
   bool is_trigger_new_or_old_reference(const LEX_STRING name);
 
   Item *create_and_link_Item_trigger_field(THD *thd, const char *name,
@@ -3209,8 +3333,10 @@ public:
   Item_param *add_placeholder(THD *thd, char *name,
                               uint pos_in_query, uint len_in_query);
   Item_param *add_placeholder(THD *thd, char *name,
-                              const char *pos, const char *end);
-
+                              const char *pos, const char *end)
+  {
+    return add_placeholder(thd, name, pos - substatement_query(thd), end - pos);
+  }
   sp_variable *sp_add_for_loop_variable(THD *thd, const LEX_STRING name,
                                         Item *value);
   sp_variable *sp_add_for_loop_upper_bound(THD *thd, Item *value)
diff --git a/sql/sql_list.h b/sql/sql_list.h
index 8f6f7337f1c..217cdad5abb 100644
--- a/sql/sql_list.h
+++ b/sql/sql_list.h
@@ -334,16 +334,15 @@ public:
   friend class error_list;
   friend class error_list_iterator;
 
-#ifndef DBUG_OFF
   /*
-    Debugging help: return N-th element in the list, or NULL if the list has
+    Return N-th element in the list, or NULL if the list has
     less than N elements.
   */
-  void *elem(int n)
+  void *elem(uint n)
   {
     list_node *node= first;
     void *data= NULL;
-    for (int i=0; i <= n; i++)
+    for (uint i= 0; i <= n; i++)
     {
       if (node == &end_of_list)
       {
@@ -355,7 +354,6 @@ public:
     }
     return data;
   }
-#endif
 
 #ifdef LIST_EXTRA_DEBUG
   /*
@@ -546,9 +544,7 @@ public:
     }
     empty();
   }
-#ifndef DBUG_OFF
-  T *elem(int n) { return (T*)base_list::elem(n); }
-#endif
+  T *elem(uint n) { return (T*) base_list::elem(n); }
 };
 
 
diff --git a/sql/sql_select.cc b/sql/sql_select.cc
index ed6a412fdc5..d3f2a674664 100644
--- a/sql/sql_select.cc
+++ b/sql/sql_select.cc
@@ -12955,6 +12955,14 @@ static bool check_row_equality(THD *thd, const Arg_comparator *comparators,
     if (left_item->type() == Item::ROW_ITEM &&
         right_item->type() == Item::ROW_ITEM)
     {
+      /*
+        Item_splocal for ROW SP variables return Item::ROW_ITEM.
+        Here we know that left_item and right_item are not Item_splocal,
+        because ROW SP variables with nested ROWs are not supported yet.
+        It's safe to cast left_item and right_item to Item_row.
+      */
+      DBUG_ASSERT(!left_item->get_item_splocal());
+      DBUG_ASSERT(!right_item->get_item_splocal());
       is_converted= check_row_equality(thd,
                                        comparators[i].subcomparators(),
                                        (Item_row *) left_item,
@@ -13025,6 +13033,15 @@ bool Item_func_eq::check_equality(THD *thd, COND_EQUAL *cond_equal,
   if (left_item->type() == Item::ROW_ITEM &&
       right_item->type() == Item::ROW_ITEM)
   {
+    /*
+      Item_splocal::type() for ROW variables returns Item::ROW_ITEM.
+      Distinguish ROW-type Item_splocal from Item_row.
+      Example query:
+        SELECT 1 FROM DUAL WHERE row_sp_variable=ROW(100,200);
+    */
+    if (left_item->get_item_splocal() ||
+        right_item->get_item_splocal())
+      return false;
     return check_row_equality(thd,
                               cmp.subcomparators(),
                               (Item_row *) left_item,
diff --git a/sql/sql_test.cc b/sql/sql_test.cc
index e70872af6c5..804a6b5c493 100644
--- a/sql/sql_test.cc
+++ b/sql/sql_test.cc
@@ -385,10 +385,10 @@ void print_sjm(SJ_MATERIALIZATION_INFO *sjm)
 /*
   Debugging help: force List<...>::elem function not be removed as unused.
 */
-Item* (List<Item>:: *dbug_list_item_elem_ptr)(int)= &List<Item>::elem;
-Item_equal* (List<Item_equal>:: *dbug_list_item_equal_elem_ptr)(int)=
+Item* (List<Item>:: *dbug_list_item_elem_ptr)(uint)= &List<Item>::elem;
+Item_equal* (List<Item_equal>:: *dbug_list_item_equal_elem_ptr)(uint)=
   &List<Item_equal>::elem;
-TABLE_LIST* (List<TABLE_LIST>:: *dbug_list_table_list_elem_ptr)(int) =
+TABLE_LIST* (List<TABLE_LIST>:: *dbug_list_table_list_elem_ptr)(uint) =
   &List<TABLE_LIST>::elem;
 
 #endif
diff --git a/sql/sql_type.cc b/sql/sql_type.cc
index 2679795eaa5..055a8969787 100644
--- a/sql/sql_type.cc
+++ b/sql/sql_type.cc
@@ -2225,8 +2225,22 @@ bool Type_handler_temporal_result::
 String *Type_handler_row::
           print_item_value(THD *thd, Item *item, String *str) const
 {
-  DBUG_ASSERT(0);
-  return NULL;
+  CHARSET_INFO *cs= thd->variables.character_set_client;
+  StringBuffer<STRING_BUFFER_USUAL_SIZE> val(cs);
+  str->append(C_STRING_WITH_LEN("ROW("));
+  for (uint i= 0 ; i < item->cols(); i++)
+  {
+    if (i > 0)
+      str->append(',');
+    Item *elem= item->element_index(i);
+    String *tmp= elem->type_handler()->print_item_value(thd, elem, &val);
+    if (tmp)
+      str->append(*tmp);
+    else
+      str->append(STRING_WITH_LEN("NULL"));
+  }
+  str->append(C_STRING_WITH_LEN(")"));
+  return str;
 }
 
 
diff --git a/sql/sql_yacc.yy b/sql/sql_yacc.yy
index 321e4e58d61..3115f9603e3 100644
--- a/sql/sql_yacc.yy
+++ b/sql/sql_yacc.yy
@@ -761,6 +761,7 @@ Virtual_column_info *add_virtual_expression(THD *thd, Item *expr)
   LEX_SYMBOL symbol;
   Lex_string_with_metadata_st lex_string_with_metadata;
   struct sys_var_with_base variable;
+  Lex_string_with_pos_st lex_string_with_pos;
   Lex_spblock_st spblock;
   Lex_spblock_handlers_st spblock_handlers;
   Lex_length_and_dec_st Lex_length_and_dec;
@@ -771,6 +772,8 @@ Virtual_column_info *add_virtual_expression(THD *thd, Item *expr)
 
   /* pointers */
   Create_field *create_field;
+  Spvar_definition *spvar_definition;
+  Row_definition_list *spvar_definition_list;
   CHARSET_INFO *charset;
   Condition_information_item *cond_info_item;
   DYNCALL_CREATE_DEF *dyncol_def;
@@ -1579,7 +1582,7 @@ bool my_yyoverflow(short **a, YYSTYPE **b, ulong *yystacksize);
 %type <lex_str>
         IDENT IDENT_QUOTED DECIMAL_NUM FLOAT_NUM NUM LONG_NUM
         HEX_NUM HEX_STRING
-        LEX_HOSTNAME ULONGLONG_NUM field_ident select_alias ident ident_or_text
+        LEX_HOSTNAME ULONGLONG_NUM field_ident select_alias ident_or_text
         IDENT_sys TEXT_STRING_sys TEXT_STRING_literal
         opt_component key_cache_name
         sp_opt_label BIN_NUM label_ident TEXT_STRING_filesystem ident_or_empty
@@ -1593,6 +1596,9 @@ bool my_yyoverflow(short **a, YYSTYPE **b, ulong *yystacksize);
 %type <lex_str_ptr>
         opt_table_alias
 
+%type <lex_string_with_pos>
+        ident ident_with_tok_start
+
 %type <table>
         table_ident table_ident_nodb references xid
         table_ident_opt_wild create_like
@@ -1679,7 +1685,7 @@ bool my_yyoverflow(short **a, YYSTYPE **b, ulong *yystacksize);
         geometry_function signed_literal expr_or_literal
         opt_escape
         sp_opt_default
-        simple_ident_nospvar simple_ident_q
+        simple_ident_nospvar simple_ident_q simple_ident_q2
         field_or_var limit_option
         part_func_expr
         window_func_expr
@@ -1886,6 +1892,9 @@ END_OF_INPUT
 %type <cond_info_item_name> condition_information_item_name;
 %type <cond_info_list> condition_information;
 
+%type <spvar_definition> row_field_name row_field_definition
+%type <spvar_definition_list> row_field_definition_list field_type_row
+
 %type <NONE> opt_window_clause window_def_list window_def window_spec
 %type <lex_str_ptr> window_name
 %type <NONE> opt_window_ref opt_window_frame_clause
@@ -2853,6 +2862,12 @@ sp_param_name_and_type:
             if (Lex->sp_param_fill_definition($$= $1))
               MYSQL_YYABORT;
           }
+        | sp_param_name field_type_row
+          {
+            $$= $1;
+            $$->field_def.field_name= $$->name.str;
+            $$->field_def.set_row_field_definitions($2);
+          }
         ;
 
 /* Stored PROCEDURE parameter declaration list */
@@ -2931,6 +2946,44 @@ sp_decl:
           DECLARE_SYM sp_decl_body { $$= $2; }
         ;
 
+row_field_name:
+          ident
+          {
+            if (check_string_char_length(&$1, 0, NAME_CHAR_LEN,
+                                         system_charset_info, 1))
+              my_yyabort_error((ER_TOO_LONG_IDENT, MYF(0), $1.str));
+            if (!($$= new (thd->mem_root) Spvar_definition()))
+              MYSQL_YYABORT;
+            Lex->init_last_field($$, $1.str, thd->variables.collation_database);
+          }
+        ;
+
+row_field_definition:
+          row_field_name type_with_opt_collate
+        ;
+
+row_field_definition_list:
+          row_field_definition
+          {
+            if (!($$= new (thd->mem_root) Row_definition_list()))
+              MYSQL_YYABORT;
+            $$->push_back($1, thd->mem_root);
+          }
+        | row_field_definition_list ',' row_field_definition
+          {
+            uint unused;
+            if ($1->find_row_field_by_name($3->field_name, &unused))
+              my_yyabort_error((ER_DUP_FIELDNAME, MYF(0), $3->field_name));
+            $$= $1;
+            $$->push_back($3, thd->mem_root);
+          }
+        ;
+
+field_type_row:
+          ROW_SYM '(' row_field_definition_list ')' { $$= $3; }
+        ;
+
+
 sp_decl_body:
           sp_decl_idents
           {
@@ -2942,8 +2995,18 @@ sp_decl_body:
             if (Lex->sp_variable_declarations_finalize(thd, $1,
                                                        &Lex->last_field[0], $4))
               MYSQL_YYABORT;
-            $$.vars= $1;
-            $$.conds= $$.hndlrs= $$.curs= 0;
+            $$.init_using_vars($1);
+          }
+        | sp_decl_idents
+          {
+            Lex->sp_variable_declarations_init(thd, $1);
+          }
+          field_type_row
+          sp_opt_default
+          {
+            if (Lex->sp_variable_declarations_row_finalize(thd, $1, $3, $4))
+              MYSQL_YYABORT;
+            $$.init_using_vars($1);
           }
         | ident CONDITION_SYM FOR_SYM sp_cond
           {
@@ -8357,7 +8420,7 @@ remember_name:
 
 remember_end:
           {
-            $$= (char*) YYLIP->get_cpp_tok_end();
+            $$= (char*) YYLIP->get_cpp_tok_end_rtrim();
           }
         ;
 
@@ -11421,32 +11484,25 @@ limit_options:
         ;
 
 limit_option:
-        ident
+        ident_with_tok_start
         {
-          Item_splocal *splocal;
           LEX *lex= thd->lex;
           Lex_input_stream *lip= & thd->m_parser_state->m_lip;
-          sp_variable *spv;
-          sp_pcontext *spc = lex->spcont;
-          if (spc && (spv = spc->find_variable($1, false)))
-          {
-            splocal= new (thd->mem_root)
-              Item_splocal(thd, $1, spv->offset, spv->sql_type(),
-                  lip->get_tok_start() - lex->sphead->m_tmp_query,
-                  lip->get_ptr() - lip->get_tok_start());
-            if (splocal == NULL)
-              MYSQL_YYABORT;
-#ifndef DBUG_OFF
-            splocal->m_sp= lex->sphead;
-#endif
-            lex->safe_to_cache_query=0;
-          }
-          else
-            my_yyabort_error((ER_SP_UNDECLARED_VAR, MYF(0), $1.str));
-          if (splocal->type() != Item::INT_ITEM)
-            my_yyabort_error((ER_WRONG_SPVAR_TYPE_IN_LIMIT, MYF(0)));
-          splocal->limit_clause_param= TRUE;
-          $$= splocal;
+          if (!($$= lex->create_item_limit(thd, $1,
+                                           $1.m_pos -
+                                           lex->substatement_query(thd),
+                                           lip->get_tok_end() - $1.m_pos)))
+            MYSQL_YYABORT;
+        }
+        | ident_with_tok_start '.' ident
+        {
+          LEX *lex= thd->lex;
+          Lex_input_stream *lip= & thd->m_parser_state->m_lip;
+          if (!($$= lex->create_item_limit(thd, $1, $3,
+                                           $1.m_pos -
+                                           lex->substatement_query(thd),
+                                           lip->get_ptr() - $1.m_pos)))
+            MYSQL_YYABORT;
         }
         | param_marker
         {
@@ -11655,6 +11711,11 @@ select_outvar:
                                           Lex->sphead)) :
                                 NULL;
           }
+        | ident '.' ident
+          {
+            if (!($$= Lex->create_outvar(thd, $1, $3)))
+              MYSQL_YYABORT;
+          }
         ;
 
 into:
@@ -13706,7 +13767,16 @@ simple_ident:
                                              lip->get_tok_end())))
               MYSQL_YYABORT;
           }
-        | simple_ident_q { $$= $1; }
+        | simple_ident_q2 { $$= $1; }
+        | ident '.' ident
+          {
+            LEX *lex= thd->lex;
+            if (!($$= lex->create_item_ident(thd, $1, $3,
+                                             $1.m_pos -
+                                             lex->substatement_query(thd),
+                                             YYLIP->get_tok_end() - $1.m_pos)))
+              MYSQL_YYABORT;
+          }
         ;
 
 simple_ident_nospvar:
@@ -13721,45 +13791,14 @@ simple_ident_nospvar:
 simple_ident_q:
           ident '.' ident
           {
-            LEX *lex= thd->lex;
-
-            /*
-              FIXME This will work ok in simple_ident_nospvar case because
-              we can't meet simple_ident_nospvar in trigger now. But it
-              should be changed in future.
-            */
-            if (lex->is_trigger_new_or_old_reference($1))
-            {
-              bool new_row= ($1.str[0]=='N' || $1.str[0]=='n');
-
-              if (!($$= lex->create_and_link_Item_trigger_field(thd, $3.str,
-                                                                new_row)))
-                MYSQL_YYABORT;
-            }
-            else
-            {
-              SELECT_LEX *sel= lex->current_select;
-              if (sel->no_table_names_allowed)
-              {
-                my_error(ER_TABLENAME_NOT_ALLOWED_HERE,
-                         MYF(0), $1.str, thd->where);
-              }
-              if ((sel->parsing_place != IN_HAVING) ||
-                  (sel->get_in_sum_expr() > 0))
-              {
-                $$= new (thd->mem_root) Item_field(thd, Lex->current_context(),
-                                                   NullS, $1.str, $3.str);
-              }
-              else
-              {
-                $$= new (thd->mem_root) Item_ref(thd, Lex->current_context(),
-                                                 NullS, $1.str, $3.str);
-              }
-              if ($$ == NULL)
-                MYSQL_YYABORT;
-            }
+            if (!($$= Lex->create_item_ident_nospvar(thd, $1, $3)))
+              MYSQL_YYABORT;
           }
-        | '.' ident '.' ident
+        | simple_ident_q2
+        ;
+
+simple_ident_q2:
+          '.' ident '.' ident
           {
             LEX *lex= thd->lex;
             SELECT_LEX *sel= lex->current_select;
@@ -13951,13 +13990,33 @@ TEXT_STRING_filesystem:
         ;
 
 ident:
-          IDENT_sys    { $$=$1; }
+          IDENT_sys
+          {
+            (LEX_STRING &)$$= $1;
+            $$.m_pos= (char *) YYLIP->get_tok_start_prev();
+          }
         | keyword
           {
             $$.str= thd->strmake($1.str, $1.length);
             if ($$.str == NULL)
               MYSQL_YYABORT;
             $$.length= $1.length;
+            $$.m_pos= (char *) YYLIP->get_tok_start_prev();
+          }
+        ;
+
+ident_with_tok_start:
+          IDENT_sys
+          {
+            (LEX_STRING &)$$= $1;
+            $$.m_pos= (char *) YYLIP->get_tok_start();
+          }
+        | keyword
+          {
+            if (!($$.str= thd->strmake($1.str, $1.length)))
+              MYSQL_YYABORT;
+            $$.length= $1.length;
+            $$.m_pos= (char *) YYLIP->get_tok_start();
           }
         ;
 
@@ -14616,9 +14675,24 @@ option_value_following_option_type:
 
 // Option values without preceding option_type.
 option_value_no_option_type:
-          internal_variable_name equal set_expr_or_default
+          ident equal set_expr_or_default
+          {
+            struct sys_var_with_base var;
+            if (Lex->init_internal_variable(&var, $1) ||
+                Lex->set_variable(&var, $3))
+              MYSQL_YYABORT;
+          }
+        | ident '.' ident equal set_expr_or_default
+          {
+            DBUG_ASSERT(Lex->var_list.is_empty());
+            if (Lex->set_variable($1, $3, $5))
+              MYSQL_YYABORT;
+           }
+        | DEFAULT '.' ident equal set_expr_or_default
           {
-            if (Lex->set_variable(&$1, $3))
+            struct sys_var_with_base var;
+            if (Lex->init_default_internal_variable(&var, $3) ||
+                Lex->set_variable(&var, $5))
               MYSQL_YYABORT;
           }
         | '@' ident_or_text equal expr
diff --git a/sql/sql_yacc_ora.yy b/sql/sql_yacc_ora.yy
index de712e33bc3..c652192622e 100644
--- a/sql/sql_yacc_ora.yy
+++ b/sql/sql_yacc_ora.yy
@@ -171,6 +171,7 @@ void ORAerror(THD *thd, const char *s)
   LEX_SYMBOL symbol;
   Lex_string_with_metadata_st lex_string_with_metadata;
   struct sys_var_with_base variable;
+  Lex_string_with_pos_st lex_string_with_pos;
   Lex_spblock_st spblock;
   Lex_spblock_handlers_st spblock_handlers;
   Lex_length_and_dec_st Lex_length_and_dec;
@@ -186,6 +187,8 @@ void ORAerror(THD *thd, const char *s)
 
   /* pointers */
   Create_field *create_field;
+  Spvar_definition *spvar_definition;
+  Row_definition_list *spvar_definition_list;
   CHARSET_INFO *charset;
   Condition_information_item *cond_info_item;
   DYNCALL_CREATE_DEF *dyncol_def;
@@ -997,7 +1000,7 @@ bool my_yyoverflow(short **a, YYSTYPE **b, ulong *yystacksize);
 %type <lex_str>
         IDENT IDENT_QUOTED DECIMAL_NUM FLOAT_NUM NUM LONG_NUM
         HEX_NUM HEX_STRING
-        LEX_HOSTNAME ULONGLONG_NUM field_ident select_alias ident ident_or_text
+        LEX_HOSTNAME ULONGLONG_NUM field_ident select_alias ident_or_text
         IDENT_sys TEXT_STRING_sys TEXT_STRING_literal
         opt_component key_cache_name
         sp_opt_label BIN_NUM label_ident TEXT_STRING_filesystem ident_or_empty
@@ -1013,6 +1016,9 @@ bool my_yyoverflow(short **a, YYSTYPE **b, ulong *yystacksize);
 %type <lex_str_ptr>
         opt_table_alias
 
+%type <lex_string_with_pos>
+        ident ident_with_tok_start
+
 %type <table>
         table_ident table_ident_nodb references xid
         table_ident_opt_wild create_like
@@ -1113,7 +1119,7 @@ bool my_yyoverflow(short **a, YYSTYPE **b, ulong *yystacksize);
         geometry_function signed_literal expr_or_literal
         opt_escape
         sp_opt_default
-        simple_ident_nospvar simple_ident_q
+        simple_ident_nospvar simple_ident_q simple_ident_q2
         field_or_var limit_option
         part_func_expr
         window_func_expr
@@ -1346,6 +1352,9 @@ END_OF_INPUT
 %type <cond_info_item_name> condition_information_item_name;
 %type <cond_info_list> condition_information;
 
+%type <spvar_definition> row_field_name row_field_definition
+%type <spvar_definition_list> row_field_definition_list field_type_row
+
 %type <NONE> opt_window_clause window_def_list window_def window_spec
 %type <lex_str_ptr> window_name
 %type <NONE> opt_window_ref opt_window_frame_clause
@@ -2318,6 +2327,12 @@ sp_param_name_and_type:
           {
             Lex->sphead->fill_spvar_using_type_reference($$= $1, $2);
           }
+        | sp_param_name field_type_row
+          {
+            $$= $1;
+            $$->field_def.field_name= $$->name.str;
+            $$->field_def.set_row_field_definitions($2);
+          }
         ;
 
 /* Stored PROCEDURE parameter declaration list */
@@ -2342,6 +2357,12 @@ sp_pdparam:
           {
             Lex->sphead->fill_spvar_using_type_reference($1, $3);
           }
+        | sp_param_name sp_opt_inout field_type_row
+          {
+            $1->mode= $2;
+            $1->field_def.field_name= $1->name.str;
+            $1->field_def.set_row_field_definitions($3);
+          }
         ;
 
 sp_opt_inout:
@@ -2435,6 +2456,44 @@ qualified_column_ident:
           }
         ;
 
+row_field_name:
+          ident_directly_assignable
+          {
+            if (check_string_char_length(&$1, 0, NAME_CHAR_LEN,
+                                         system_charset_info, 1))
+              my_yyabort_error((ER_TOO_LONG_IDENT, MYF(0), $1.str));
+            if (!($$= new (thd->mem_root) Spvar_definition()))
+              MYSQL_YYABORT;
+            Lex->init_last_field($$, $1.str, thd->variables.collation_database);
+          }
+        ;
+
+row_field_definition:
+          row_field_name type_with_opt_collate
+        ;
+
+row_field_definition_list:
+          row_field_definition
+          {
+            if (!($$= new (thd->mem_root) Row_definition_list()))
+              MYSQL_YYABORT;
+            $$->push_back($1, thd->mem_root);
+          }
+        | row_field_definition_list ',' row_field_definition
+          {
+            uint unused;
+            if ($1->find_row_field_by_name($3->field_name, &unused))
+              my_yyabort_error((ER_DUP_FIELDNAME, MYF(0), $3->field_name));
+            $$= $1;
+            $$->push_back($3, thd->mem_root);
+          }
+        ;
+
+field_type_row:
+          ROW_SYM '(' row_field_definition_list ')' { $$= $3; }
+        ;
+
+
 sp_decl_body:
           sp_decl_idents
           {
@@ -2459,6 +2518,17 @@ sp_decl_body:
               MYSQL_YYABORT;
             $$.init_using_vars($1);
           }
+        | sp_decl_idents
+          {
+            Lex->sp_variable_declarations_init(thd, $1);
+          }
+          field_type_row
+          sp_opt_default
+          {
+            if (Lex->sp_variable_declarations_row_finalize(thd, $1, $3, $4))
+              MYSQL_YYABORT;
+            $$.init_using_vars($1);
+          }
         | ident_directly_assignable CONDITION_SYM FOR_SYM sp_cond
           {
             if (Lex->spcont->declare_condition(thd, $1, $4))
@@ -8360,7 +8430,7 @@ remember_name:
 
 remember_end:
           {
-            $$= (char*) YYLIP->get_cpp_tok_end();
+            $$= (char*) YYLIP->get_cpp_tok_end_rtrim();
           }
         ;
 
@@ -11516,32 +11586,25 @@ limit_options:
         ;
 
 limit_option:
-        ident
+        ident_with_tok_start
         {
-          Item_splocal *splocal;
           LEX *lex= thd->lex;
           Lex_input_stream *lip= & thd->m_parser_state->m_lip;
-          sp_variable *spv;
-          sp_pcontext *spc = lex->spcont;
-          if (spc && (spv = spc->find_variable($1, false)))
-          {
-            splocal= new (thd->mem_root)
-              Item_splocal(thd, $1, spv->offset, spv->sql_type(),
-                  lip->get_tok_start() - lex->sphead->m_tmp_query,
-                  lip->get_ptr() - lip->get_tok_start());
-            if (splocal == NULL)
-              MYSQL_YYABORT;
-#ifndef DBUG_OFF
-            splocal->m_sp= lex->sphead;
-#endif
-            lex->safe_to_cache_query=0;
-          }
-          else
-            my_yyabort_error((ER_SP_UNDECLARED_VAR, MYF(0), $1.str));
-          if (splocal->type() != Item::INT_ITEM)
-            my_yyabort_error((ER_WRONG_SPVAR_TYPE_IN_LIMIT, MYF(0)));
-          splocal->limit_clause_param= TRUE;
-          $$= splocal;
+          if (!($$= lex->create_item_limit(thd, $1,
+                                           $1.m_pos -
+                                           lex->substatement_query(thd),
+                                           lip->get_tok_end() - $1.m_pos)))
+            MYSQL_YYABORT;
+        }
+        | ident_with_tok_start '.' ident
+        {
+          LEX *lex= thd->lex;
+          Lex_input_stream *lip= & thd->m_parser_state->m_lip;
+          if (!($$= lex->create_item_limit(thd, $1, $3,
+                                           $1.m_pos -
+                                           lex->substatement_query(thd),
+                                           lip->get_ptr() - $1.m_pos)))
+            MYSQL_YYABORT;
         }
         | param_marker
         {
@@ -11750,6 +11813,11 @@ select_outvar:
                                           Lex->sphead)) :
                                 NULL;
           }
+        | ident '.' ident
+          {
+            if (!($$= Lex->create_outvar(thd, $1, $3)))
+              MYSQL_YYABORT;
+          }
         ;
 
 into:
@@ -13820,7 +13888,16 @@ simple_ident:
                                              lip->get_tok_end())))
               MYSQL_YYABORT;
           }
-        | simple_ident_q { $$= $1; }
+        | simple_ident_q2
+        | ident '.' ident
+          {
+            LEX *lex= thd->lex;
+            if (!($$= lex->create_item_ident(thd, $1, $3,
+                                             $1.m_pos -
+                                             lex->substatement_query(thd),
+                                             YYLIP->get_tok_end() - $1.m_pos)))
+              MYSQL_YYABORT;
+          }
         ;
 
 simple_ident_nospvar:
@@ -13835,45 +13912,14 @@ simple_ident_nospvar:
 simple_ident_q:
           ident '.' ident
           {
-            LEX *lex= thd->lex;
-
-            /*
-              FIXME This will work ok in simple_ident_nospvar case because
-              we can't meet simple_ident_nospvar in trigger now. But it
-              should be changed in future.
-            */
-            if (lex->is_trigger_new_or_old_reference($1))
-            {
-              bool new_row= ($1.str[0]=='N' || $1.str[0]=='n');
-
-              if (!($$= lex->create_and_link_Item_trigger_field(thd, $3.str,
-                                                                new_row)))
-                MYSQL_YYABORT;
-            }
-            else
-            {
-              SELECT_LEX *sel= lex->current_select;
-              if (sel->no_table_names_allowed)
-              {
-                my_error(ER_TABLENAME_NOT_ALLOWED_HERE,
-                         MYF(0), $1.str, thd->where);
-              }
-              if ((sel->parsing_place != IN_HAVING) ||
-                  (sel->get_in_sum_expr() > 0))
-              {
-                $$= new (thd->mem_root) Item_field(thd, Lex->current_context(),
-                                                   NullS, $1.str, $3.str);
-              }
-              else
-              {
-                $$= new (thd->mem_root) Item_ref(thd, Lex->current_context(),
-                                                 NullS, $1.str, $3.str);
-              }
-              if ($$ == NULL)
-                MYSQL_YYABORT;
-            }
+            if (!($$= Lex->create_item_ident_nospvar(thd, $1, $3)))
+              MYSQL_YYABORT;
           }
-        | colon_with_pos ident '.' ident
+        | simple_ident_q2
+        ;
+
+simple_ident_q2:
+          colon_with_pos ident '.' ident
           {
             LEX *lex= Lex;
             if (lex->is_trigger_new_or_old_reference($2))
@@ -14082,16 +14128,35 @@ TEXT_STRING_filesystem:
         ;
 
 ident:
-          IDENT_sys    { $$=$1; }
+          IDENT_sys
+          {
+            (LEX_STRING &)$$= $1;
+            $$.m_pos= (char *) YYLIP->get_tok_start_prev();
+          }
         | keyword
           {
             $$.str= thd->strmake($1.str, $1.length);
             if ($$.str == NULL)
               MYSQL_YYABORT;
             $$.length= $1.length;
+            $$.m_pos= (char *) YYLIP->get_tok_start_prev();
           }
         ;
 
+ident_with_tok_start:
+          IDENT_sys
+          {
+            (LEX_STRING &)$$= $1;
+            $$.m_pos= (char *) YYLIP->get_tok_start();
+          }
+        | keyword
+          {
+            if (!($$.str= thd->strmake($1.str, $1.length)))
+              MYSQL_YYABORT;
+            $$.length= $1.length;
+            $$.m_pos= (char *) YYLIP->get_tok_start();
+          }
+        ;
 
 ident_directly_assignable:
           IDENT_sys    { $$=$1; }
@@ -14358,6 +14423,7 @@ keyword_sp_data_type:
         | POINT_SYM                {}
         | POLYGON                  {}
         | RAW                      {} /* Oracle-R */
+        | ROW_SYM                  {}
         | SERIAL_SYM               {}
         | TEXT_SYM                 {}
         | TIMESTAMP                {}
@@ -14593,7 +14659,6 @@ keyword_sp_not_data_type:
         | ROWCOUNT_SYM             {}
         | ROW_COUNT_SYM            {}
         | ROW_FORMAT_SYM           {}
-        | ROW_SYM                  {}
         | RTREE_SYM                {}
         | SCHEDULE_SYM             {}
         | SCHEMA_NAME_SYM          {}
@@ -14711,6 +14776,21 @@ set_assign:
                 sp_create_assignment_instr(thd, yychar == YYEMPTY))
               MYSQL_YYABORT;
           }
+        | ident_directly_assignable '.' ident SET_VAR
+          {
+            LEX *lex=Lex;
+            lex->set_stmt_init();
+            lex->var_list.empty();
+            sp_create_assignment_lex(thd, yychar == YYEMPTY);
+          }
+          set_expr_or_default
+          {
+            LEX *lex= Lex;
+            DBUG_ASSERT(lex->var_list.is_empty());
+            if (lex->set_variable($1, $3, $6) ||
+                lex->sphead->restore_lex(thd))
+              MYSQL_YYABORT;
+          }
         ;
 
 set_stmt_option_value_following_option_type_list:
@@ -14845,11 +14925,26 @@ option_value_following_option_type:
 
 // Option values without preceding option_type.
 option_value_no_option_type:
-          internal_variable_name equal set_expr_or_default
+          ident equal set_expr_or_default
+          {
+            struct sys_var_with_base var;
+            if (Lex->init_internal_variable(&var, $1) ||
+                Lex->set_variable(&var, $3))
+              MYSQL_YYABORT;
+          }
+        | ident '.' ident equal set_expr_or_default
           {
-            if (Lex->set_variable(&$1, $3))
+            DBUG_ASSERT(Lex->var_list.is_empty());
+            if (Lex->set_variable($1, $3, $5))
               MYSQL_YYABORT;
           }
+        | DEFAULT '.' ident equal set_expr_or_default
+          {
+            struct sys_var_with_base var;
+            if (Lex->init_default_internal_variable(&var, $3) ||
+                Lex->set_variable(&var, $5))
+              MYSQL_YYABORT;
+           }
         | '@' ident_or_text equal expr
           {
             Item_func_set_user_var *item;
@@ -14998,11 +15093,6 @@ internal_variable_name_directly_assignable:
             if (Lex->init_internal_variable(&$$, $1))
               MYSQL_YYABORT;
           }
-        | ident_directly_assignable '.' ident
-          {
-            if (Lex->init_internal_variable(&$$, $1, $3))
-              MYSQL_YYABORT;
-          }
         | DEFAULT '.' ident
           {
             if (Lex->init_default_internal_variable(&$$, $3))
diff --git a/sql/structs.h b/sql/structs.h
index d3ad16592ef..e0fb05f05da 100644
--- a/sql/structs.h
+++ b/sql/structs.h
@@ -693,4 +693,10 @@ public:
 };
 
 
+struct Lex_string_with_pos_st: public LEX_STRING
+{
+  const char *m_pos;
+};
+
+
 #endif /* STRUCTS_INCLUDED */