MDEV-10577 sql_mode=ORACLE: %TYPE in variable declarations

22 files changed, 505 insertions, 111 deletions

diff --git a/sql/field.h b/sql/field.h
index 89b209379dd..359101530f3 100644
--- a/sql/field.h
+++ b/sql/field.h
@@ -3853,7 +3853,7 @@ public:
   Column_definition(const char *name, enum_field_types type):
     field_name(name),
     comment(null_lex_str),
-    on_update(0), sql_type(type), length(0), decimals(0),
+    on_update(NULL), sql_type(type), length(0), decimals(0),
     flags(0), pack_length(0), key_length(0), unireg_check(Field::NONE),
     interval(0), charset(&my_charset_bin),
     srid(0), geom_type(Field::GEOM_GEOMETRY),
@@ -3948,6 +3948,43 @@ public:
     return unireg_check == Field::TIMESTAMP_DN_FIELD
         || unireg_check == Field::TIMESTAMP_DNUN_FIELD;
   }
+
+  // Replace the entire value by another definition
+  void set_column_definition(const Column_definition *def)
+  {
+    *this= *def;
+  }
+};
+
+
+/**
+  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;
+
+  Data type references to other object types will be added soon, e.g.:
+  - DECLARE a table_name%ROWTYPE;
+  - DECLARE a cursor_name%ROWTYPE;
+  - DECLARE a record_name%TYPE;
+  - DECLARE a variable_name%TYPE;
+*/
+class Spvar_definition: public Column_definition
+{
+  class Qualified_column_ident *m_column_type_ref; // for %TYPE
+public:
+  Spvar_definition()
+   :m_column_type_ref(NULL) { }
+  bool is_column_type_ref() const { return m_column_type_ref != 0; }
+  class Qualified_column_ident *column_type_ref() const
+  {
+    return m_column_type_ref;
+  }
+  void set_column_type_ref(class Qualified_column_ident *ref)
+  {
+    m_column_type_ref= ref;
+  }
 };
 
 
diff --git a/sql/item.cc b/sql/item.cc
index c77a488961c..9ad54cfad70 100644
--- a/sql/item.cc
+++ b/sql/item.cc
@@ -6100,8 +6100,11 @@ Field *Item::make_string_field(TABLE *table)
 /**
   Create a field based on field_type of argument.
 
-  For now, this is only used to create a field for
-  IFNULL(x,something) and time functions
+  This is used to create a field for
+  - IFNULL(x,something)
+  - time functions
+  - prepared statement placeholders
+  - SP variables with data type references: DECLARE a t1.a%TYPE;
 
   @retval
     NULL  error
diff --git a/sql/item.h b/sql/item.h
index 1cde032636c..4e737d09952 100644
--- a/sql/item.h
+++ b/sql/item.h
@@ -2122,6 +2122,40 @@ public:
   Item *get_copy(THD *thd, MEM_ROOT *mem_root) { return 0; }
 };
 
+
+/**
+  An Item_splocal variant whose data type becomes known only at
+  sp_rcontext creation time, e.g. "DECLARE var1 t1.col1%TYPE".
+*/
+class Item_splocal_with_delayed_data_type: public Item_splocal
+{
+public:
+  Item_splocal_with_delayed_data_type(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)
+  { }
+  bool fix_fields(THD *thd, Item **it)
+  {
+    if (Item_splocal::fix_fields(thd, it))
+      return true;
+    set_handler(this_item()->type_handler());
+    return false;
+  }
+  /*
+    Override the inherited create_field_for_create_select(),
+    because we want to preserve the exact data type for:
+      DECLARE a t1.a%TYPE;
+      CREATE TABLE t1 AS SELECT a;
+    The inherited implementation would create a column
+    based on result_type(), which is less exact.
+  */
+  Field *create_field_for_create_select(TABLE *table)
+  { return tmp_table_field_from_field_type(table, false, true); }
+};
+
 /*****************************************************************************
   Item_splocal inline implementation.
 *****************************************************************************/
diff --git a/sql/sp_head.cc b/sql/sp_head.cc
index d63b5700a28..7fd2624ab7a 100644
--- a/sql/sp_head.cc
+++ b/sql/sp_head.cc
@@ -1397,6 +1397,41 @@ set_routine_security_ctx(THD *thd, sp_head *sp, bool is_proc,
 
 
 /**
+  Create rcontext using the routine security.
+  This is important for sql_mode=ORACLE to make sure that the invoker has
+  access to the tables mentioned in the %TYPE references.
+
+  In non-Oracle sql_modes we do not need access to any tables,
+  so we can omit the security context switch for performance purposes.
+
+  @param thd
+  @param sphead
+  @param is_proc
+  @param root_pctx
+  @param ret_value
+  @retval           NULL - error (access denided or EOM)
+  @retval          !NULL - success (the invoker has rights to all %TYPE tables)
+*/
+sp_rcontext *sp_head::rcontext_create(THD *thd, bool is_proc, Field *ret_value)
+{
+  bool has_column_type_refs= m_flags & HAS_COLUMN_TYPE_REFS;
+#ifndef NO_EMBEDDED_ACCESS_CHECKS
+  Security_context *save_security_ctx;
+  if (has_column_type_refs &&
+      set_routine_security_ctx(thd, this, is_proc, &save_security_ctx))
+    return NULL;
+#endif
+  sp_rcontext *res= sp_rcontext::create(thd, m_pcont, ret_value,
+                                        has_column_type_refs);
+#ifndef NO_EMBEDDED_ACCESS_CHECKS
+  if (has_column_type_refs)
+    m_security_ctx.restore_security_context(thd, save_security_ctx);
+#endif
+  return res;
+}
+
+
+/**
   Execute trigger stored program.
 
   - changes security context for triggers
@@ -1493,7 +1528,8 @@ sp_head::execute_trigger(THD *thd,
   init_sql_alloc(&call_mem_root, MEM_ROOT_BLOCK_SIZE, 0, MYF(0));
   thd->set_n_backup_active_arena(&call_arena, &backup_arena);
 
-  if (!(nctx= sp_rcontext::create(thd, m_pcont, NULL)))
+  if (!(nctx= sp_rcontext::create(thd, m_pcont, NULL,
+                                  m_flags & HAS_COLUMN_TYPE_REFS)))
   {
     err_status= TRUE;
     goto err_with_cleanup;
@@ -1608,7 +1644,7 @@ sp_head::execute_function(THD *thd, Item **argp, uint argcount,
   init_sql_alloc(&call_mem_root, MEM_ROOT_BLOCK_SIZE, 0, MYF(0));
   thd->set_n_backup_active_arena(&call_arena, &backup_arena);
 
-  if (!(nctx= sp_rcontext::create(thd, m_pcont, return_value_fld)))
+  if (!(nctx= rcontext_create(thd, false, return_value_fld)))
   {
     thd->restore_active_arena(&call_arena, &backup_arena);
     err_status= TRUE;
@@ -1823,7 +1859,7 @@ sp_head::execute_procedure(THD *thd, List<Item> *args)
   if (! octx)
   {
     /* Create a temporary old context. */
-    if (!(octx= sp_rcontext::create(thd, m_pcont, NULL)))
+    if (!(octx= rcontext_create(thd, true, NULL)))
     {
       DBUG_PRINT("error", ("Could not create octx"));
       DBUG_RETURN(TRUE);
@@ -1838,7 +1874,7 @@ sp_head::execute_procedure(THD *thd, List<Item> *args)
     thd->spcont->callers_arena= thd;
   }
 
-  if (!(nctx= sp_rcontext::create(thd, m_pcont, NULL)))
+  if (!(nctx= rcontext_create(thd, true, NULL)))
   {
     delete nctx; /* Delete nctx if it was init() that failed. */
     thd->spcont= save_spcont;
diff --git a/sql/sp_head.h b/sql/sp_head.h
index cdf2dd94f3a..ec19058a7f1 100644
--- a/sql/sp_head.h
+++ b/sql/sp_head.h
@@ -165,7 +165,11 @@ public:
       b) because in CONTAINS SQL case they don't provide enough
       information anyway.
      */
-    MODIFIES_DATA= 4096
+    MODIFIES_DATA= 4096,
+    /*
+      Marks routines that have column type references: DECLARE a t1.a%TYPE;
+    */
+    HAS_COLUMN_TYPE_REFS= 8192
   };
 
   stored_procedure_type m_type;
@@ -201,6 +205,9 @@ public:
   {
     m_sp_cache_version= version_arg;
   }
+
+  sp_rcontext *rcontext_create(THD *thd, bool is_proc, Field *ret_value);
+
 private:
   /**
     Version of the stored routine cache at the moment when the
@@ -539,7 +546,8 @@ public:
 
   /**
     Check and prepare an instance of Column_definition for field creation
-    (fill all necessary attributes).
+    (fill all necessary attributes), for variables, parameters and
+    function return values.
 
     @param[in]  thd          Thread handle
     @param[in]  lex          Yacc parsing context
@@ -553,6 +561,32 @@ public:
     return field_def->check(thd) ||
            field_def->sp_prepare_create_field(thd, mem_root);
   }
+  /**
+    Check and prepare a Column_definition for a variable or a parameter.
+  */
+  bool fill_spvar_definition(THD *thd, Column_definition *def)
+  {
+    if (fill_field_definition(thd, def))
+      return true;
+    def->pack_flag|= FIELDFLAG_MAYBE_NULL;
+    return false;
+  }
+  bool fill_spvar_definition(THD *thd, Column_definition *def, const char *name)
+  {
+    def->field_name= name;
+    return fill_spvar_definition(thd, def);
+  }
+  /**
+    Set a column type reference for a parameter definition
+  */
+  void fill_spvar_using_type_reference(sp_variable *spvar,
+                                       Qualified_column_ident *ref)
+  {
+    spvar->field_def.set_column_type_ref(ref);
+    spvar->field_def.field_name= spvar->name.str;
+    m_flags|= sp_head::HAS_COLUMN_TYPE_REFS;
+  }
+
   void set_info(longlong created, longlong modified,
 		st_sp_chistics *chistics, sql_mode_t sql_mode);
 
diff --git a/sql/sp_pcontext.cc b/sql/sp_pcontext.cc
index fb162501ebc..642f1f16d29 100644
--- a/sql/sp_pcontext.cc
+++ b/sql/sp_pcontext.cc
@@ -533,7 +533,7 @@ const sp_pcursor *sp_pcontext::find_cursor(const LEX_STRING name,
 
 
 void sp_pcontext::retrieve_field_definitions(
-  List<Column_definition> *field_def_lst) const
+  List<Spvar_definition> *field_def_lst) const
 {
   /* Put local/context fields in the result list. */
 
diff --git a/sql/sp_pcontext.h b/sql/sp_pcontext.h
index 33a7bffca7c..414a3cbc7cc 100644
--- a/sql/sp_pcontext.h
+++ b/sql/sp_pcontext.h
@@ -57,7 +57,7 @@ public:
   Item *default_value;
 
   /// Full type information (field meta-data) of the SP-variable.
-  Column_definition field_def;
+  Spvar_definition field_def;
 
   /// Field-type of the SP-variable.
   enum_field_types sql_type() const { return field_def.sql_type; }
@@ -436,7 +436,7 @@ public:
   /// context and its children.
   ///
   /// @param field_def_lst[out] Container to store type information.
-  void retrieve_field_definitions(List<Column_definition> *field_def_lst) const;
+  void retrieve_field_definitions(List<Spvar_definition> *field_def_lst) const;
 
   /// Find SP-variable by name.
   ///
diff --git a/sql/sp_rcontext.cc b/sql/sp_rcontext.cc
index 488fb85a69f..5fcdbf42313 100644
--- a/sql/sp_rcontext.cc
+++ b/sql/sp_rcontext.cc
@@ -26,6 +26,9 @@
 #include "sp_rcontext.h"
 #include "sp_pcontext.h"
 #include "sql_select.h"                     // create_virtual_tmp_table
+#include "sql_base.h"                       // open_tables_only_view_structure
+#include "sql_acl.h"                        // SELECT_ACL
+#include "sql_parse.h"                      // check_table_access
 
 ///////////////////////////////////////////////////////////////////////////
 // sp_rcontext implementation.
@@ -59,7 +62,8 @@ sp_rcontext::~sp_rcontext()
 
 sp_rcontext *sp_rcontext::create(THD *thd,
                                  const sp_pcontext *root_parsing_ctx,
-                                 Field *return_value_fld)
+                                 Field *return_value_fld,
+                                 bool resolve_type_refs)
 {
   sp_rcontext *ctx= new (thd->mem_root) sp_rcontext(root_parsing_ctx,
                                                     return_value_fld,
@@ -68,9 +72,13 @@ sp_rcontext *sp_rcontext::create(THD *thd,
   if (!ctx)
     return NULL;
 
+  List<Spvar_definition> field_def_lst;
+  ctx->m_root_parsing_ctx->retrieve_field_definitions(&field_def_lst);
+
   if (ctx->alloc_arrays(thd) ||
-      ctx->init_var_table(thd) ||
-      ctx->init_var_items(thd))
+      (resolve_type_refs && ctx->resolve_type_refs(thd, field_def_lst)) ||
+      ctx->init_var_table(thd, field_def_lst) ||
+      ctx->init_var_items(thd, field_def_lst))
   {
     delete ctx;
     return NULL;
@@ -102,28 +110,107 @@ bool sp_rcontext::alloc_arrays(THD *thd)
 }
 
 
-bool sp_rcontext::init_var_table(THD *thd)
+bool sp_rcontext::init_var_table(THD *thd,
+                                 List<Spvar_definition> &field_def_lst)
 {
-  List<Column_definition> field_def_lst;
-
   if (!m_root_parsing_ctx->max_var_index())
     return false;
 
-  m_root_parsing_ctx->retrieve_field_definitions(&field_def_lst);
-
   DBUG_ASSERT(field_def_lst.elements == m_root_parsing_ctx->max_var_index());
 
   if (!(m_var_table= create_virtual_tmp_table(thd, field_def_lst)))
     return true;
 
-  m_var_table->copy_blobs= true;
-  m_var_table->alias.set("", 0, m_var_table->alias.charset());
+  return false;
+}
+
 
+/**
+  Check if we have access to use a column as a %TYPE reference.
+  @return false - OK
+  @return true  - access denied
+*/
+static inline bool
+check_column_grant_for_type_ref(THD *thd, TABLE_LIST *table_list,
+                                const Qualified_column_ident *col)
+{
+#ifndef NO_EMBEDDED_ACCESS_CHECKS
+  table_list->table->grant.want_privilege= SELECT_ACL;
+  return check_column_grant_in_table_ref(thd, table_list,
+                                         col->m_column.str,
+                                         col->m_column.length);
+#else
   return false;
+#endif
+}
+
+
+/**
+  This method implementation is very close to fill_schema_table_by_open().
+*/
+bool sp_rcontext::resolve_type_ref(THD *thd, Column_definition *def,
+                                   Qualified_column_ident *ref)
+{
+  Open_tables_backup open_tables_state_backup;
+  thd->reset_n_backup_open_tables_state(&open_tables_state_backup);
+
+  TABLE_LIST *table_list;
+  Field *src;
+  LEX *save_lex= thd->lex;
+  bool rc= true;
+
+  sp_lex_local lex(thd, thd->lex);
+  thd->lex= &lex;
+
+  lex.context_analysis_only= CONTEXT_ANALYSIS_ONLY_VIEW;
+  // Make %TYPE variables see temporary tables that shadow permanent tables
+  thd->temporary_tables= open_tables_state_backup.temporary_tables;
+
+  if ((table_list= lex.select_lex.add_table_to_list(thd, ref, NULL, 0,
+                                                    TL_READ_NO_INSERT,
+                                                    MDL_SHARED_READ)) &&
+      !check_table_access(thd, SELECT_ACL, table_list, TRUE, UINT_MAX, FALSE) &&
+      !open_tables_only_view_structure(thd, table_list,
+                                       thd->mdl_context.has_locks()))
+  {
+    if ((src= lex.query_tables->table->find_field_by_name(ref->m_column.str)))
+    {
+      if (!(rc= check_column_grant_for_type_ref(thd, table_list, ref)))
+      {
+        *def= Column_definition(thd, src, NULL/*No defaults,no constraints*/);
+        def->flags&= (uint) ~NOT_NULL_FLAG;
+        rc= def->sp_prepare_create_field(thd, thd->mem_root);
+      }
+    }
+    else
+      my_error(ER_BAD_FIELD_ERROR, MYF(0), ref->m_column.str, ref->table.str);
+  }
+
+  lex.unit.cleanup();
+  thd->temporary_tables= NULL; // Avoid closing temporary tables
+  close_thread_tables(thd);
+  thd->lex= save_lex;
+  thd->restore_backup_open_tables_state(&open_tables_state_backup);
+  return rc;
 }
 
 
-bool sp_rcontext::init_var_items(THD *thd)
+bool sp_rcontext::resolve_type_refs(THD *thd, List<Spvar_definition> &defs)
+{
+  List_iterator<Spvar_definition> it(defs);
+  Spvar_definition *def;
+  while ((def= it++))
+  {
+    if (def->is_column_type_ref() &&
+        resolve_type_ref(thd, def, def->column_type_ref()))
+      return true;
+  }
+  return false;
+};
+
+
+bool sp_rcontext::init_var_items(THD *thd,
+                                 List<Spvar_definition> &field_def_lst)
 {
   uint num_vars= m_root_parsing_ctx->max_var_index();
 
@@ -360,8 +447,9 @@ uint sp_rcontext::exit_handler(Diagnostics_area *da)
 }
 
 
-int sp_rcontext::set_variable(THD *thd, Field *field, Item **value)
+int sp_rcontext::set_variable(THD *thd, uint idx, Item **value)
 {
+  Field *field= m_var_table->field[idx];
   if (!value)
   {
     field->set_null();
diff --git a/sql/sp_rcontext.h b/sql/sp_rcontext.h
index d040186ec34..061f736f6e9 100644
--- a/sql/sp_rcontext.h
+++ b/sql/sp_rcontext.h
@@ -70,7 +70,8 @@ public:
   /// @return valid sp_rcontext object or NULL in case of OOM-error.
   static sp_rcontext *create(THD *thd,
                              const sp_pcontext *root_parsing_ctx,
-                             Field *return_value_fld);
+                             Field *return_value_fld,
+                             bool resolve_type_refs);
 
   ~sp_rcontext();
 
@@ -185,8 +186,7 @@ public:
   // SP-variables.
   /////////////////////////////////////////////////////////////////////////
 
-  int set_variable(THD *thd, uint var_idx, Item **value)
-  { return set_variable(thd, m_var_table->field[var_idx], value); }
+  int set_variable(THD *thd, uint var_idx, Item **value);
 
   Item *get_item(uint var_idx) const
   { return m_var_items[var_idx]; }
@@ -334,7 +334,11 @@ private:
   /// @return error flag.
   /// @retval false on success.
   /// @retval true on error.
-  bool init_var_table(THD *thd);
+  bool init_var_table(THD *thd, List<Spvar_definition> &defs);
+
+  bool resolve_type_refs(THD *, List<Spvar_definition> &defs);
+  bool resolve_type_ref(THD *thd, Column_definition *def,
+                                  Qualified_column_ident *ref);
 
   /// Create and initialize an Item-adapter (Item_field) for each SP-var field.
   ///
@@ -343,7 +347,7 @@ private:
   /// @return error flag.
   /// @retval false on success.
   /// @retval true on error.
-  bool init_var_items(THD *thd);
+  bool init_var_items(THD *thd, List<Spvar_definition> &defs);
 
   /// Create an instance of appropriate Item_cache class depending on the
   /// specified type in the callers arena.
@@ -357,8 +361,6 @@ 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;
 
-  int set_variable(THD *thd, Field *field, Item **value);
-
 private:
   /// Top-level (root) parsing context for this runtime context.
   const sp_pcontext *m_root_parsing_ctx;
diff --git a/sql/sql_base.cc b/sql/sql_base.cc
index 93511bb9188..5a51e2b34af 100644
--- a/sql/sql_base.cc
+++ b/sql/sql_base.cc
@@ -4763,6 +4763,51 @@ end:
 }
 
 
+/**
+  Open a table to read its structure, e.g. for:
+  - SHOW FIELDS
+  - delayed SP variable data type definition: DECLARE a t1.a%TYPE
+
+  The flag MYSQL_OPEN_GET_NEW_TABLE is passed to make %TYPE work
+  in stored functions, as during a stored function call
+  (e.g. in a SELECT query) the tables referenced in %TYPE can already be locked,
+  and attempt to open it again would return an error in open_table().
+
+  The flag MYSQL_OPEN_GET_NEW_TABLE is not really needed for
+  SHOW FIELDS or for a "CALL sp()" statement, but it's not harmful,
+  so let's pass it unconditionally.
+*/
+
+bool open_tables_only_view_structure(THD *thd, TABLE_LIST *table_list,
+                                     bool can_deadlock)
+{
+  DBUG_ENTER("open_tables_only_view_structure");
+  /*
+    Let us set fake sql_command so views won't try to merge
+    themselves into main statement. If we don't do this,
+    SELECT * from information_schema.xxxx will cause problems.
+    SQLCOM_SHOW_FIELDS is used because it satisfies
+    'LEX::only_view_structure()'.
+  */
+  enum_sql_command save_sql_command= thd->lex->sql_command;
+  thd->lex->sql_command= SQLCOM_SHOW_FIELDS;
+  bool rc= (thd->open_temporary_tables(table_list) ||
+           open_normal_and_derived_tables(thd, table_list,
+                                          (MYSQL_OPEN_IGNORE_FLUSH |
+                                           MYSQL_OPEN_FORCE_SHARED_HIGH_PRIO_MDL |
+                                           MYSQL_OPEN_GET_NEW_TABLE |
+                                           (can_deadlock ?
+                                            MYSQL_OPEN_FAIL_ON_MDL_CONFLICT : 0)),
+                                          DT_PREPARE | DT_CREATE));
+  /*
+    Restore old value of sql_command back as it is being looked at in
+    process_table() function.
+  */
+  thd->lex->sql_command= save_sql_command;
+  DBUG_RETURN(rc);
+}
+
+
 /*
   Mark all real tables in the list as free for reuse.
 
diff --git a/sql/sql_base.h b/sql/sql_base.h
index 374ac56c3d8..96bb1fb24cd 100644
--- a/sql/sql_base.h
+++ b/sql/sql_base.h
@@ -254,6 +254,8 @@ TABLE *open_n_lock_single_table(THD *thd, TABLE_LIST *table_l,
                                 Prelocking_strategy *prelocking_strategy);
 bool open_normal_and_derived_tables(THD *thd, TABLE_LIST *tables, uint flags,
                                     uint dt_phases);
+bool open_tables_only_view_structure(THD *thd, TABLE_LIST *tables,
+                                     bool can_deadlock);
 bool lock_tables(THD *thd, TABLE_LIST *tables, uint counter, uint flags);
 int decide_logging_format(THD *thd, TABLE_LIST *tables);
 void close_thread_table(THD *thd, TABLE **table_ptr);
diff --git a/sql/sql_class.h b/sql/sql_class.h
index 1b8d8ff0b37..cac1cdd8fd3 100644
--- a/sql/sql_class.h
+++ b/sql/sql_class.h
@@ -5370,6 +5370,26 @@ public:
   }
 };
 
+
+class Qualified_column_ident: public Table_ident
+{
+public:
+  LEX_STRING m_column;
+public:
+  Qualified_column_ident(const LEX_STRING table, const LEX_STRING column)
+   :Table_ident(table),
+    m_column(column)
+  { }
+  Qualified_column_ident(THD *thd,
+                         const LEX_STRING db,
+                         const LEX_STRING table,
+                         const LEX_STRING column)
+   :Table_ident(thd, db, table, false),
+    m_column(column)
+  { }
+};
+
+
 // this is needed for user_vars hash
 class user_var_entry
 {
diff --git a/sql/sql_lex.cc b/sql/sql_lex.cc
index 3a34ca9e64e..0b861817bb4 100644
--- a/sql/sql_lex.cc
+++ b/sql/sql_lex.cc
@@ -5098,11 +5098,7 @@ sp_variable *LEX::sp_param_init(LEX_STRING name)
 
 bool LEX::sp_param_fill_definition(sp_variable *spvar)
 {
-  if (sphead->fill_field_definition(thd, last_field))
-    return true;
-  spvar->field_def.field_name= spvar->name.str;
-  spvar->field_def.pack_flag |= FIELDFLAG_MAYBE_NULL;
-  return false;
+  return sphead->fill_spvar_definition(thd, last_field, spvar->name.str);
 }
 
 
@@ -5222,7 +5218,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,
+                                            const Column_definition *cdef,
                                             Item *dflt_value_item)
 {
   uint num_vars= spcont->context_var_count();
@@ -5241,16 +5237,15 @@ bool LEX::sp_variable_declarations_finalize(THD *thd, int nvars,
     if (!spvar)
       return true;
 
-    if (!last)
-      spvar->field_def= cdef;
-
     spvar->default_value= dflt_value_item;
-    spvar->field_def.field_name= spvar->name.str;
 
-    if (sphead->fill_field_definition(thd, &spvar->field_def))
-      return true;
-
-    spvar->field_def.pack_flag |= FIELDFLAG_MAYBE_NULL;
+    if (cdef)
+    {
+      if (!last)
+        spvar->field_def.set_column_definition(cdef);
+      if (sphead->fill_spvar_definition(thd, &spvar->field_def, spvar->name.str))
+        return true;
+    }
 
     /* The last instruction is responsible for freeing LEX. */
     sp_instr_set *is= new (this->thd->mem_root)
@@ -5266,6 +5261,23 @@ bool LEX::sp_variable_declarations_finalize(THD *thd, int nvars,
 }
 
 
+bool
+LEX::sp_variable_declarations_with_ref_finalize(THD *thd, int nvars,
+                                                Qualified_column_ident *ref,
+                                                Item *def)
+{
+  uint num_vars= spcont->context_var_count();
+  for (uint i= num_vars - nvars; i < num_vars; i++)
+  {
+    sp_variable *spvar= spcont->find_context_variable(i);
+    spvar->field_def.set_column_type_ref(ref);
+    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);
+}
+
+
 /**********************************************************************
   The FOR LOOP statement
 
@@ -5295,8 +5307,9 @@ sp_variable *LEX::sp_add_for_loop_variable(THD *thd, const LEX_STRING name,
 {
   sp_variable *spvar= spcont->add_variable(thd, name);
   spcont->declare_var_boundary(1);
-  spvar->field_def= Column_definition(spvar->name.str, MYSQL_TYPE_LONGLONG);
-  if (sp_variable_declarations_finalize(thd, 1, spvar->field_def, value))
+  spvar->field_def.field_name= spvar->name.str;
+  spvar->field_def.sql_type= MYSQL_TYPE_LONGLONG;
+  if (sp_variable_declarations_finalize(thd, 1, NULL, value))
     return NULL;
   return spvar;
 }
@@ -6016,10 +6029,14 @@ Item *LEX::create_item_ident_sp(THD *thd, LEX_STRING name,
       return NULL;
     }
 
-    Item_splocal *splocal;
-    splocal= new (thd->mem_root) Item_splocal(thd, name,
-                                              spv->offset, spv->sql_type(),
-                                              start_in_q, length_in_q);
+    Item_splocal *splocal= spv->field_def.is_column_type_ref() ?
+      new (thd->mem_root) Item_splocal_with_delayed_data_type(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);
     if (splocal == NULL)
       return NULL;
 #ifndef DBUG_OFF
diff --git a/sql/sql_lex.h b/sql/sql_lex.h
index 7a531fb761f..c3ef0710112 100644
--- a/sql/sql_lex.h
+++ b/sql/sql_lex.h
@@ -3110,8 +3110,11 @@ public:
   bool set_variable(struct sys_var_with_base *variable, Item *item);
   void sp_variable_declarations_init(THD *thd, int nvars);
   bool sp_variable_declarations_finalize(THD *thd, int nvars,
-                                         const Column_definition &cdef,
+                                         const Column_definition *cdef,
                                          Item *def);
+  bool sp_variable_declarations_with_ref_finalize(THD *thd, int nvars,
+                                                  Qualified_column_ident *col,
+                                                  Item *def);
   bool sp_handler_declaration_init(THD *thd, int type);
   bool sp_handler_declaration_finalize(THD *thd, int type);
 
diff --git a/sql/sql_select.cc b/sql/sql_select.cc
index 18c1a06cf60..ed6a412fdc5 100644
--- a/sql/sql_select.cc
+++ b/sql/sql_select.cc
@@ -17086,11 +17086,11 @@ bool Virtual_tmp_table::init(uint field_count)
 };
 
 
-bool Virtual_tmp_table::add(List<Column_definition> &field_list)
+bool Virtual_tmp_table::add(List<Spvar_definition> &field_list)
 {
   /* Create all fields and calculate the total length of record */
-  Column_definition *cdef;            /* column definition */
-  List_iterator_fast<Column_definition> it(field_list);
+  Spvar_definition *cdef;            /* column definition */
+  List_iterator_fast<Spvar_definition> it(field_list);
   for ( ; (cdef= it++); )
   {
     Field *tmp;
diff --git a/sql/sql_select.h b/sql/sql_select.h
index 4327646cdee..77cf73d785f 100644
--- a/sql/sql_select.h
+++ b/sql/sql_select.h
@@ -2032,6 +2032,8 @@ public:
     bzero(this, sizeof(*this));
     temp_pool_slot= MY_BIT_NONE;
     in_use= thd;
+    copy_blobs= true;
+    alias.set("", 0, &my_charset_bin);
   }
 
   ~Virtual_tmp_table()
@@ -2075,11 +2077,11 @@ public:
   }
 
   /**
-    Add fields from a Column_definition list
+    Add fields from a Spvar_definition list
     @returns false - on success.
     @returns true  - on error.
   */
-  bool add(List<Column_definition> &field_list);
+  bool add(List<Spvar_definition> &field_list);
 
   /**
     Open a virtual table for read/write:
@@ -2117,7 +2119,7 @@ public:
 */
 
 inline TABLE *
-create_virtual_tmp_table(THD *thd, List<Column_definition> &field_list)
+create_virtual_tmp_table(THD *thd, List<Spvar_definition> &field_list)
 {
   Virtual_tmp_table *table;
   if (!(table= new(thd) Virtual_tmp_table(thd)))
diff --git a/sql/sql_show.cc b/sql/sql_show.cc
index d1d1410d51a..4fb1b2e8c84 100644
--- a/sql/sql_show.cc
+++ b/sql/sql_show.cc
@@ -4207,6 +4207,7 @@ fill_schema_table_by_open(THD *thd, bool is_show_fields_or_keys,
   /* Prepare temporary LEX. */
   thd->lex= lex= &temp_lex;
   lex_start(thd);
+  lex->sql_command= old_lex->sql_command;
 
   /* Disable constant subquery evaluation as we won't be locking tables. */
   lex->context_analysis_only= CONTEXT_ANALYSIS_ONLY_VIEW;
@@ -4259,26 +4260,8 @@ fill_schema_table_by_open(THD *thd, bool is_show_fields_or_keys,
     table_list->i_s_requested_object= schema_table->i_s_requested_object;
   }
 
-  /*
-    Let us set fake sql_command so views won't try to merge
-    themselves into main statement. If we don't do this,
-    SELECT * from information_schema.xxxx will cause problems.
-    SQLCOM_SHOW_FIELDS is used because it satisfies
-    'only_view_structure()'.
-  */
-  lex->sql_command= SQLCOM_SHOW_FIELDS;
-  result= (thd->open_temporary_tables(table_list) ||
-           open_normal_and_derived_tables(thd, table_list,
-                                          (MYSQL_OPEN_IGNORE_FLUSH |
-                                           MYSQL_OPEN_FORCE_SHARED_HIGH_PRIO_MDL |
-                                           (can_deadlock ?
-                                            MYSQL_OPEN_FAIL_ON_MDL_CONFLICT : 0)),
-                                          DT_PREPARE | DT_CREATE));
-  /*
-    Restore old value of sql_command back as it is being looked at in
-    process_table() function.
-  */
-  lex->sql_command= old_lex->sql_command;
+  DBUG_ASSERT(thd->lex == lex);
+  result= open_tables_only_view_structure(thd, table_list, can_deadlock);
 
   DEBUG_SYNC(thd, "after_open_table_ignore_flush");
 
diff --git a/sql/sql_yacc.yy b/sql/sql_yacc.yy
index de5c07c6c6a..321e4e58d61 100644
--- a/sql/sql_yacc.yy
+++ b/sql/sql_yacc.yy
@@ -2940,7 +2940,7 @@ sp_decl_body:
           sp_opt_default
           {
             if (Lex->sp_variable_declarations_finalize(thd, $1,
-                                                       Lex->last_field[0], $4))
+                                                       &Lex->last_field[0], $4))
               MYSQL_YYABORT;
             $$.vars= $1;
             $$.conds= $$.hndlrs= $$.curs= 0;
diff --git a/sql/sql_yacc_ora.yy b/sql/sql_yacc_ora.yy
index 46de67036ff..192144013ac 100644
--- a/sql/sql_yacc_ora.yy
+++ b/sql/sql_yacc_ora.yy
@@ -209,6 +209,7 @@ void ORAerror(THD *thd, const char *s)
   String *string;
   TABLE_LIST *table_list;
   Table_ident *table;
+  Qualified_column_ident *qualified_column_ident;
   char *simple_string;
   const char *const_simple_string;
   chooser_compare_func_creator boolfunc2creator;
@@ -1002,6 +1003,7 @@ bool my_yyoverflow(short **a, YYSTYPE **b, ulong *yystacksize);
         sp_opt_label BIN_NUM label_ident TEXT_STRING_filesystem ident_or_empty
         opt_constraint constraint opt_ident
         label_declaration_oracle ident_directly_assignable
+        sp_decl_ident
         sp_block_label
 
 %type <lex_string_with_metadata>
@@ -1015,6 +1017,9 @@ bool my_yyoverflow(short **a, YYSTYPE **b, ulong *yystacksize);
         table_ident table_ident_nodb references xid
         table_ident_opt_wild create_like
 
+%type <qualified_column_ident>
+        qualified_column_ident
+
 %type <simple_string>
         remember_name remember_end opt_db remember_tok_start
         wild_and_where
@@ -1184,6 +1189,9 @@ bool my_yyoverflow(short **a, YYSTYPE **b, ulong *yystacksize);
 %type <symbol> keyword keyword_sp
                keyword_directly_assignable
                keyword_directly_not_assignable
+               sp_decl_ident_keyword
+               keyword_sp_data_type
+               keyword_sp_not_data_type
 
 %type <lex_user> user grant_user grant_role user_or_role current_role
                  admin_option_for_role user_maybe_role
@@ -2307,6 +2315,10 @@ sp_param_name_and_type:
             if (Lex->sp_param_fill_definition($$= $1))
               MYSQL_YYABORT;
           }
+        | sp_param_name qualified_column_ident '%' TYPE_SYM
+          {
+            Lex->sphead->fill_spvar_using_type_reference($$= $1, $2);
+          }
         ;
 
 /* Stored PROCEDURE parameter declaration list */
@@ -2327,6 +2339,10 @@ sp_pdparam:
             if (Lex->sp_param_fill_definition($1))
               MYSQL_YYABORT;
           }
+        | sp_param_name sp_opt_inout qualified_column_ident '%' TYPE_SYM
+          {
+            Lex->sphead->fill_spvar_using_type_reference($1, $3);
+          }
         ;
 
 sp_opt_inout:
@@ -2406,6 +2422,20 @@ sp_decl_body_list:
           }
         ;
 
+qualified_column_ident:
+          sp_decl_ident '.' ident
+          {
+            if (!($$= new (thd->mem_root) Qualified_column_ident($1, $3)))
+              MYSQL_YYABORT;
+          }
+        | sp_decl_ident '.' ident '.' ident
+          {
+            if (!($$= new (thd->mem_root) Qualified_column_ident(thd,
+                                                                 $1, $3, $5)))
+              MYSQL_YYABORT;
+          }
+        ;
+
 sp_decl_body:
           sp_decl_idents
           {
@@ -2415,10 +2445,20 @@ sp_decl_body:
           sp_opt_default
           {
             if (Lex->sp_variable_declarations_finalize(thd, $1,
-                                                       Lex->last_field[0], $4))
+                                                       &Lex->last_field[0], $4))
+              MYSQL_YYABORT;
+            $$.init_using_vars($1);
+          }
+        | sp_decl_idents
+          {
+            Lex->sp_variable_declarations_init(thd, $1);
+          }
+          qualified_column_ident '%' TYPE_SYM
+          sp_opt_default
+          {
+            if (Lex->sp_variable_declarations_with_ref_finalize(thd, $1, $3, $6))
               MYSQL_YYABORT;
-            $$.vars= $1;
-            $$.conds= $$.hndlrs= $$.curs= 0;
+            $$.init_using_vars($1);
           }
         | ident_directly_assignable CONDITION_SYM FOR_SYM sp_cond
           {
@@ -2875,8 +2915,25 @@ condition_information_item_name:
           { $$= Condition_information_item::RETURNED_SQLSTATE; }
         ;
 
+sp_decl_ident:
+          IDENT_sys
+        | sp_decl_ident_keyword
+          {
+            $$.str= thd->strmake($1.str, $1.length);
+            if ($$.str == NULL)
+              MYSQL_YYABORT;
+            $$.length= $1.length;
+          }
+        ;
+
+sp_decl_ident_keyword:
+          keyword_directly_assignable
+        | keyword_sp_not_data_type
+        ;
+
+
 sp_decl_idents:
-          ident_directly_assignable
+          sp_decl_ident
           {
             /* NOTE: field definition is filled in sp_decl section. */
 
@@ -14276,6 +14333,48 @@ keyword_directly_not_assignable:
  * conflicts.
  */
 keyword_sp:
+          keyword_sp_data_type
+        | keyword_sp_not_data_type
+        ;
+
+
+/*
+  These keywords are generally allowed as identifiers,
+  but not allowed as non-delimited SP variable names.
+*/
+keyword_sp_data_type:
+          BIT_SYM                  {}
+        | BOOLEAN_SYM              {} /* PLSQL-R */
+        | BOOL_SYM                 {}
+        | CLOB                     {}
+        | DATE_SYM                 {} /* Oracle-R, PLSQL-R */
+        | DATETIME                 {}
+        | ENUM                     {}
+        | FIXED_SYM                {}
+        | GEOMETRYCOLLECTION       {}
+        | GEOMETRY_SYM             {}
+        | LINESTRING               {}
+        | MEDIUM_SYM               {}
+        | MULTILINESTRING          {}
+        | MULTIPOINT               {}
+        | MULTIPOLYGON             {}
+        | NATIONAL_SYM             {}
+        | NCHAR_SYM                {}
+        | NUMBER_SYM               {} /* Oracle-R, PLSQL-R */
+        | NVARCHAR_SYM             {}
+        | POINT_SYM                {}
+        | POLYGON                  {}
+        | RAW                      {} /* Oracle-R */
+        | SERIAL_SYM               {}
+        | TEXT_SYM                 {}
+        | TIMESTAMP                {}
+        | TIME_SYM                 {} /* Oracle-R */
+        | VARCHAR2                 {} /* Oracle-R, PLSQL-R */
+        | YEAR_SYM                 {}
+        ;
+
+
+keyword_sp_not_data_type:
           ACTION                   {}
         | ADDDATE_SYM              {}
         | ADMIN_SYM                {}
@@ -14293,10 +14392,7 @@ keyword_sp:
         | AUTO_SYM                 {}
         | AVG_ROW_LENGTH           {}
         | AVG_SYM                  {}
-        | BIT_SYM                  {}
         | BLOCK_SYM                {}
-        | BOOL_SYM                 {}
-        | BOOLEAN_SYM              {}
         | BTREE_SYM                {}
         | CASCADED                 {}
         | CATALOG_NAME_SYM         {}
@@ -14305,7 +14401,6 @@ keyword_sp:
         | CIPHER_SYM               {}
         | CLIENT_SYM               {}
         | CLASS_ORIGIN_SYM         {}
-        | CLOB                     {}
         | COALESCE                 {}
         | CODE_SYM                 {}
         | COLLATION_SYM            {}
@@ -14334,8 +14429,6 @@ keyword_sp:
         | CURSOR_NAME_SYM          {}
         | DATA_SYM                 {}
         | DATAFILE_SYM             {}
-        | DATETIME                 {}
-        | DATE_SYM                 {}
         | DAY_SYM                  {}
         | DECODE_SYM               {}
         | DEFINER_SYM              {}
@@ -14350,7 +14443,6 @@ keyword_sp:
         | DUPLICATE_SYM            {}
         | DYNAMIC_SYM              {}
         | ENDS_SYM                 {}
-        | ENUM                     {}
         | ENGINE_SYM               {}
         | ENGINES_SYM              {}
         | ERROR_SYM                {}
@@ -14371,11 +14463,8 @@ keyword_sp:
         | FULL                     {}
         | FILE_SYM                 {}
         | FIRST_SYM                {}
-        | FIXED_SYM                {}
         | GENERAL                  {}
         | GENERATED_SYM            {}
-        | GEOMETRY_SYM             {}
-        | GEOMETRYCOLLECTION       {}
         | GET_FORMAT               {}
         | GRANTS                   {}
         | GLOBAL_SYM               {}
@@ -14404,7 +14493,6 @@ keyword_sp:
         | LEAVES                   {}
         | LESS_SYM                 {}
         | LEVEL_SYM                {}
-        | LINESTRING               {}
         | LIST_SYM                 {}
         | LOCAL_SYM                {}
         | LOCKS_SYM                {}
@@ -14438,7 +14526,6 @@ keyword_sp:
         | MAX_STATEMENT_TIME_SYM   {}
         | MAX_UPDATES_PER_HOUR     {}
         | MAX_USER_CONNECTIONS_SYM {}
-        | MEDIUM_SYM               {}
         | MEMORY_SYM               {}
         | MERGE_SYM                {}
         | MESSAGE_TEXT_SYM         {}
@@ -14449,24 +14536,17 @@ keyword_sp:
         | MODIFY_SYM               {}
         | MODE_SYM                 {}
         | MONTH_SYM                {}
-        | MULTILINESTRING          {}
-        | MULTIPOINT               {}
-        | MULTIPOLYGON             {}
         | MUTEX_SYM                {}
         | MYSQL_SYM                {}
         | MYSQL_ERRNO_SYM          {}
         | NAME_SYM                 {}
         | NAMES_SYM                {}
-        | NATIONAL_SYM             {}
-        | NCHAR_SYM                {}
         | NEXT_SYM                 {}
         | NEW_SYM                  {}
         | NO_WAIT_SYM              {}
         | NODEGROUP_SYM            {}
         | NONE_SYM                 {}
         | NOTFOUND_SYM             {}
-        | NUMBER_SYM               {}
-        | NVARCHAR_SYM             {}
         | OFFSET_SYM               {}
         | OLD_PASSWORD_SYM         {}
         | ONE_SYM                  {}
@@ -14482,8 +14562,6 @@ keyword_sp:
         | PHASE_SYM                {}
         | PLUGIN_SYM               {}
         | PLUGINS_SYM              {}
-        | POINT_SYM                {}
-        | POLYGON                  {}
         | PRESERVE_SYM             {}
         | PREV_SYM                 {}
         | PRIVILEGES               {}
@@ -14495,7 +14573,6 @@ keyword_sp:
         | QUARTER_SYM              {}
         | QUERY_SYM                {}
         | QUICK                    {}
-        | RAW                      {}
         | READ_ONLY_SYM            {}
         | REBUILD_SYM              {}
         | RECOVER_SYM              {}
@@ -14528,7 +14605,6 @@ keyword_sp:
         | SCHEDULE_SYM             {}
         | SCHEMA_NAME_SYM          {}
         | SECOND_SYM               {}
-        | SERIAL_SYM               {}
         | SERIALIZABLE_SYM         {}
         | SESSION_SYM              {}
         | SIMPLE_SYM               {}
@@ -14563,15 +14639,12 @@ keyword_sp:
         | TABLESPACE               {}
         | TEMPORARY                {}
         | TEMPTABLE_SYM            {}
-        | TEXT_SYM                 {}
         | THAN_SYM                 {}
         | TRANSACTION_SYM          {}
         | TRANSACTIONAL_SYM        {}
         | TRIGGERS_SYM             {}
-        | TIMESTAMP                {}
         | TIMESTAMP_ADD            {}
         | TIMESTAMP_DIFF           {}
-        | TIME_SYM                 {}
         | TYPES_SYM                {}
         | TYPE_SYM                 {}
         | UDF_RETURNS_SYM          {}
@@ -14584,7 +14657,6 @@ keyword_sp:
         | UNTIL_SYM                {}
         | USER_SYM                 {}
         | USE_FRM                  {}
-        | VARCHAR2                 {}
         | VARIABLES                {}
         | VIEW_SYM                 {}
         | VIRTUAL_SYM              {}
@@ -14596,7 +14668,6 @@ keyword_sp:
         | WORK_SYM                 {}
         | X509_SYM                 {}
         | XML_SYM                  {}
-        | YEAR_SYM                 {}
         | VIA_SYM                  {}
         ;
 
diff --git a/sql/structs.h b/sql/structs.h
index 10dc1f40fa9..d3ad16592ef 100644
--- a/sql/structs.h
+++ b/sql/structs.h
@@ -647,6 +647,11 @@ public:
   {
     vars= conds= hndlrs= curs= 0;
   }
+  void init_using_vars(uint nvars)
+  {
+    vars= nvars;
+    conds= hndlrs= curs= 0;
+  }
   void join(const Lex_spblock_st &b1, const Lex_spblock_st &b2)
   {
     vars= b1.vars + b2.vars;
diff --git a/sql/table.cc b/sql/table.cc
index d09a29ea64f..38cda5cc209 100644
--- a/sql/table.cc
+++ b/sql/table.cc
@@ -8202,3 +8202,14 @@ LEX_CSTRING *fk_option_name(enum_fk_option opt)
   };
   return names + opt;
 }
+
+
+Field *TABLE::find_field_by_name(const char *str) const
+{
+  for (Field **tmp= field; *tmp; tmp++)
+  {
+    if (!my_strcasecmp(system_charset_info, (*tmp)->field_name, str))
+      return *tmp;
+  }
+  return NULL;
+}
diff --git a/sql/table.h b/sql/table.h
index a94fcef2316..22dfeaaa596 100644
--- a/sql/table.h
+++ b/sql/table.h
@@ -1440,6 +1440,7 @@ public:
                                       TABLE *tmp_table,
                                       TMP_TABLE_PARAM *tmp_table_param,
                                       bool with_cleanup);
+  Field *find_field_by_name(const char *str) const;
 };