MDEV-5816: Stored programs: validation of stored program statements

This is the prerequisite patch introducing the class sp_lex_instr
that encapsulates access to an instance of the class sp_lex_keeper.
Every SP instruction that need to get access to a LEX object on its
processing should inherit this class and implement two abstract methods:
  is_invalid(),
  invalidate().

These methods will be used in subsequent patches to implement recompilation of
SP instructions on failure.

Currently, the following instructions are derived from the class sp_lex_instr:
  sp_instr_stmt,
  sp_instr_set,
  sp_instr_set_trigger_field,
  sp_instr_jump_if_not,
  sp_instr_freturn,
  sp_instr_cpush,
  sp_instr_cursor_copy_struct,
  sp_instr_set_case_expr

Additionally, this patch converts the class sp_instr_opt_meta
to base abstract class (that is, not inherited from the class
sp_instr). Since this class originally was designed to provide a way
for opimizer to update a destination address for jump SP-instructions,
the only useful method at the interface of this class is set_destination
and therefore inheritance from the class sp_instr is meaningless.

Every jump SP instruction now must be inherited directly from
the class sp_instr_opt_meta and additionally from either the class
sp_lex_instr or sp_instr depending on whether this SP instruction
need to get access to a LEX object or not.

Moreover, the class sp_cursor doesn't own a data member of
the class sp_lex_keeper any more. Instead, the virtual method
get_lex_keeper() has been added to the class sp_cursor() that
returns nullptr and this method is overridden in the derived class
sp_instr_cpush to provide a pointer to a real instance of
the class sp_lex_keeper. Doing this way we exclude duplication
of a data member of the type sp_lex_keeper at the class sp_instr_cpush
since it is derived both from sp_lex_instr and sp_cursor, and sp_lex_instr
already encapsulates a data member of the class sp_lex_keeper.

3 files changed, 208 insertions, 91 deletions

diff --git a/sql/sp_instr.cc b/sql/sp_instr.cc
index 94926d6227f..bc1394ac942 100644
--- a/sql/sp_instr.cc
+++ b/sql/sp_instr.cc
@@ -747,15 +747,6 @@ sp_instr_set_trigger_field::print(String *str)
                                     QT_ITEM_ORIGINAL_FUNC_NULLIF));
 }
 
-/*
-  sp_instr_opt_meta
-*/
-
-uint sp_instr_opt_meta::get_cont_dest() const
-{
-  return m_cont_dest;
-}
-
 
 /*
  sp_instr_jump class functions
@@ -918,8 +909,15 @@ sp_instr_jump_if_not::opt_move(uint dst, List<sp_instr_opt_meta> *bp)
   }
   else if (m_cont_optdest)
     m_cont_dest= m_cont_optdest->m_ip; // Backward
-  /* This will take care of m_dest and m_ip */
-  sp_instr_jump::opt_move(dst, bp);
+
+  /*
+    Take care about m_dest and m_ip
+  */
+  if (m_dest > m_ip)
+    bp->push_back(this);      // Forward
+  else if (m_optdest)
+    m_dest= m_optdest->m_ip;  // Backward
+  m_ip= dst;
 }
 
 
@@ -1195,7 +1193,7 @@ sp_instr_cpush::execute(THD *thd, uint *nextp)
 {
   DBUG_ENTER("sp_instr_cpush::execute");
 
-  sp_cursor::reset(thd, &m_lex_keeper);
+  sp_cursor::reset(thd);
   m_lex_keeper.disable_query_cache();
   thd->spcont->push_cursor(this);
 
@@ -1492,14 +1490,14 @@ sp_instr_cursor_copy_struct::exec_core(THD *thd, uint *nextp)
 
   /*
     Copy structure only once. If the cursor%ROWTYPE variable is declared
-    inside a LOOP block, it gets its structure on the first loop interation
+    inside a LOOP block, it gets its structure on the first loop iteration
     and remembers the structure for all consequent loop iterations.
     It we recreated the structure on every iteration, we would get
     potential memory leaks, and it would be less efficient.
   */
   if (!row->arguments())
   {
-    sp_cursor tmp(thd, &m_lex_keeper, true);
+    sp_cursor tmp(thd, true);
     // Open the cursor without copying data
     if (!(ret= tmp.open(thd)))
     {
diff --git a/sql/sp_instr.h b/sql/sp_instr.h
index 90f427d64ce..6bffd6efa6b 100644
--- a/sql/sp_instr.h
+++ b/sql/sp_instr.h
@@ -280,20 +280,48 @@ private:
 
 
 /**
-  Call out to some prepared SQL statement.
+  The base class for any stored program instruction that need to get access
+  to a LEX object on execution.
 */
 
-class sp_instr_stmt : public sp_instr
+class sp_lex_instr : public sp_instr
+{
+public:
+  sp_lex_instr(uint ip, sp_pcontext *ctx, LEX *lex, bool is_lex_owner)
+  : sp_instr(ip, ctx),
+    m_lex_keeper(lex, is_lex_owner)
+  {}
+
+  virtual bool is_invalid() const = 0;
+
+  virtual void invalidate() = 0;
+
+protected:
+  sp_lex_keeper m_lex_keeper;
+};
+
+
+/**
+  The class sp_instr_stmt represents almost all conventional SQL-statements.
+*/
+
+class sp_instr_stmt : public sp_lex_instr
 {
   sp_instr_stmt(const sp_instr_stmt &);	/**< Prevent use of these */
   void operator=(sp_instr_stmt &);
 
+  /**
+    Flag to tell whether a metadata this instruction depends on
+    has been changed and a LEX object should be reinitialized.
+  */
+  bool m_valid;
+
 public:
   LEX_STRING m_query;		///< For thd->query
 
   sp_instr_stmt(uint ip, sp_pcontext *ctx, LEX *lex)
-    : sp_instr(ip, ctx),
-      m_lex_keeper(lex, true)
+    : sp_lex_instr(ip, ctx, lex, true),
+      m_valid(true)
   {
     m_query.str= 0;
     m_query.length= 0;
@@ -307,16 +335,23 @@ public:
 
   void print(String *str) override;
 
-private:
-  sp_lex_keeper m_lex_keeper;
+  bool is_invalid() const override
+  {
+    return !m_valid;
+  }
+
+  void invalidate() override
+  {
+    m_valid= false;
+  }
 
 public:
   PSI_statement_info* get_psi_info() override { return & psi_info; }
   static PSI_statement_info psi_info;
-}; // class sp_instr_stmt : public sp_instr
+}; // class sp_instr_stmt : public sp_lex_instr
 
 
-class sp_instr_set : public sp_instr
+class sp_instr_set : public sp_lex_instr
 {
   sp_instr_set(const sp_instr_set &);	/**< Prevent use of these */
   void operator=(sp_instr_set &);
@@ -326,11 +361,10 @@ public:
                const Sp_rcontext_handler *rh,
 	       uint offset, Item *val,
                LEX *lex, bool lex_resp)
-    : sp_instr(ip, ctx),
+    : sp_lex_instr(ip, ctx, lex, lex_resp),
       m_rcontext_handler(rh),
       m_offset(offset),
-      m_value(val),
-      m_lex_keeper(lex, lex_resp)
+      m_value(val)
   {}
 
   virtual ~sp_instr_set() = default;
@@ -341,17 +375,26 @@ public:
 
   void print(String *str) override;
 
+  bool is_invalid() const override
+  {
+    return m_value == nullptr;
+  }
+
+  void invalidate() override
+  {
+    m_value= nullptr;
+  }
+
 protected:
   sp_rcontext *get_rcontext(THD *thd) const;
   const Sp_rcontext_handler *m_rcontext_handler;
   uint m_offset;		///< Frame offset
   Item *m_value;
-  sp_lex_keeper m_lex_keeper;
 
 public:
   PSI_statement_info* get_psi_info() override { return & psi_info; }
   static PSI_statement_info psi_info;
-}; // class sp_instr_set : public sp_instr
+}; // class sp_instr_set : public sp_lex_instr
 
 
 /*
@@ -431,7 +474,7 @@ public:
   Set NEW/OLD row field value instruction. Used in triggers.
 */
 
-class sp_instr_set_trigger_field : public sp_instr
+class sp_instr_set_trigger_field : public sp_lex_instr
 {
   sp_instr_set_trigger_field(const sp_instr_set_trigger_field &);
   void operator=(sp_instr_set_trigger_field &);
@@ -440,10 +483,9 @@ public:
   sp_instr_set_trigger_field(uint ip, sp_pcontext *ctx,
                              Item_trigger_field *trg_fld,
                              Item *val, LEX *lex)
-    : sp_instr(ip, ctx),
+    : sp_lex_instr(ip, ctx, lex, true),
       trigger_field(trg_fld),
-      value(val),
-      m_lex_keeper(lex, true)
+      value(val)
   {}
 
   virtual ~sp_instr_set_trigger_field() = default;
@@ -454,15 +496,24 @@ public:
 
   void print(String *str) override;
 
+  bool is_invalid() const override
+  {
+    return value == nullptr;
+  }
+
+  void invalidate() override
+  {
+    value= nullptr;
+  }
+
 private:
   Item_trigger_field *trigger_field;
   Item *value;
-  sp_lex_keeper m_lex_keeper;
 
 public:
   PSI_statement_info* get_psi_info() override { return & psi_info; }
   static PSI_statement_info psi_info;
-}; // class sp_instr_trigger_field : public sp_instr
+}; // class sp_instr_trigger_field : public sp_lex_instr
 
 
 /**
@@ -473,23 +524,14 @@ public:
   the continuation destination, we put them both here for simplicity.
 */
 
-class sp_instr_opt_meta : public sp_instr
+class sp_instr_opt_meta
 {
 public:
   uint m_dest;			///< Where we will go
   uint m_cont_dest;             ///< Where continue handlers will go
 
-  sp_instr_opt_meta(uint ip, sp_pcontext *ctx)
-    : sp_instr(ip, ctx),
-      m_dest(0),
-      m_cont_dest(0),
-      m_optdest(0),
-      m_cont_optdest(0)
-  {}
-
-  sp_instr_opt_meta(uint ip, sp_pcontext *ctx, uint dest)
-    : sp_instr(ip, ctx),
-      m_dest(dest),
+  explicit sp_instr_opt_meta(uint dest)
+    : m_dest(dest),
       m_cont_dest(0),
       m_optdest(0),
       m_cont_optdest(0)
@@ -499,26 +541,26 @@ public:
 
   virtual void set_destination(uint old_dest, uint new_dest) = 0;
 
-  uint get_cont_dest() const override;
-
 protected:
   sp_instr *m_optdest;		///< Used during optimization
   sp_instr *m_cont_optdest;     ///< Used during optimization
-}; // class sp_instr_opt_meta : public sp_instr
+}; // class sp_instr_opt_meta
 
 
-class sp_instr_jump : public sp_instr_opt_meta
+class sp_instr_jump : public sp_instr, public sp_instr_opt_meta
 {
   sp_instr_jump(const sp_instr_jump &);	/**< Prevent use of these */
   void operator=(sp_instr_jump &);
 
 public:
   sp_instr_jump(uint ip, sp_pcontext *ctx)
-    : sp_instr_opt_meta(ip, ctx)
+    : sp_instr(ip, ctx),
+      sp_instr_opt_meta(0)
   {}
 
   sp_instr_jump(uint ip, sp_pcontext *ctx, uint dest)
-    : sp_instr_opt_meta(ip, ctx, dest)
+    : sp_instr(ip, ctx),
+      sp_instr_opt_meta(dest)
   {}
 
   virtual ~sp_instr_jump() = default;
@@ -540,6 +582,11 @@ public:
     m_dest= dest;
   }
 
+  uint get_cont_dest() const override
+  {
+    return m_cont_dest;
+  }
+
   /**
     Update the destination; used by the optimizer.
   */
@@ -552,10 +599,10 @@ public:
 public:
   PSI_statement_info* get_psi_info() override { return & psi_info; }
   static PSI_statement_info psi_info;
-}; // class sp_instr_jump : public sp_instr_opt_meta
+}; // class sp_instr_jump : public sp_instr, public sp_instr_opt_meta
 
 
-class sp_instr_jump_if_not : public sp_instr_jump
+class sp_instr_jump_if_not : public sp_lex_instr, public sp_instr_opt_meta
 {
   /**< Prevent use of these */
   sp_instr_jump_if_not(const sp_instr_jump_if_not &);
@@ -563,15 +610,15 @@ class sp_instr_jump_if_not : public sp_instr_jump
 
 public:
   sp_instr_jump_if_not(uint ip, sp_pcontext *ctx, Item *i, LEX *lex)
-    : sp_instr_jump(ip, ctx),
-      m_expr(i),
-      m_lex_keeper(lex, true)
+    : sp_lex_instr(ip, ctx, lex, true),
+      sp_instr_opt_meta(0),
+      m_expr(i)
   {}
 
   sp_instr_jump_if_not(uint ip, sp_pcontext *ctx, Item *i, uint dest, LEX *lex)
-    : sp_instr_jump(ip, ctx, dest),
-      m_expr(i),
-      m_lex_keeper(lex, true)
+    : sp_lex_instr(ip, ctx, lex, true),
+      sp_instr_opt_meta(dest),
+      m_expr(i)
   {}
 
   virtual ~sp_instr_jump_if_not() = default;
@@ -592,21 +639,43 @@ public:
 
   void opt_move(uint dst, List<sp_instr_opt_meta> *ibp) override;
 
+  uint get_cont_dest() const override
+  {
+    return m_cont_dest;
+  }
+
   void set_destination(uint old_dest, uint new_dest) override
   {
-    sp_instr_jump::set_destination(old_dest, new_dest);
+    if (m_dest == old_dest)
+      m_dest= new_dest;
     if (m_cont_dest == old_dest)
       m_cont_dest= new_dest;
   }
 
+  void backpatch(uint dest, sp_pcontext *dst_ctx) override
+  {
+    /* Calling backpatch twice is a logic flaw in jump resolution. */
+    DBUG_ASSERT(m_dest == 0);
+    m_dest= dest;
+  }
+
+  bool is_invalid() const override
+  {
+    return m_expr == nullptr;
+  }
+
+  void invalidate() override
+  {
+    m_expr= nullptr;
+  }
+
 private:
   Item *m_expr;			///< The condition
-  sp_lex_keeper m_lex_keeper;
 
 public:
   PSI_statement_info* get_psi_info() override { return & psi_info; }
   static PSI_statement_info psi_info;
-}; // class sp_instr_jump_if_not : public sp_instr_jump
+}; // class sp_instr_jump_if_not
 
 
 class sp_instr_preturn : public sp_instr
@@ -637,7 +706,7 @@ public:
 }; // class sp_instr_preturn : public sp_instr
 
 
-class sp_instr_freturn : public sp_instr
+class sp_instr_freturn : public sp_lex_instr
 {
   sp_instr_freturn(const sp_instr_freturn &);	/**< Prevent use of these */
   void operator=(sp_instr_freturn &);
@@ -645,10 +714,9 @@ class sp_instr_freturn : public sp_instr
 public:
   sp_instr_freturn(uint ip, sp_pcontext *ctx,
 		   Item *val, const Type_handler *handler, LEX *lex)
-    : sp_instr(ip, ctx),
+    : sp_lex_instr(ip, ctx, lex, true),
       m_value(val),
-      m_type_handler(handler),
-      m_lex_keeper(lex, true)
+      m_type_handler(handler)
   {}
 
   virtual ~sp_instr_freturn() = default;
@@ -665,15 +733,24 @@ public:
     return UINT_MAX;
   }
 
+  bool is_invalid() const override
+  {
+    return m_value == nullptr;
+  }
+
+  void invalidate() override
+  {
+    m_value= nullptr;
+  }
+
 protected:
   Item *m_value;
   const Type_handler *m_type_handler;
-  sp_lex_keeper m_lex_keeper;
 
 public:
   PSI_statement_info* get_psi_info() override { return & psi_info; }
   static PSI_statement_info psi_info;
-}; // class sp_instr_freturn : public sp_instr
+}; // class sp_instr_freturn : public sp_lex_instr
 
 
 class sp_instr_hpush_jump : public sp_instr_jump
@@ -816,16 +893,16 @@ public:
   This is DECLARE CURSOR
 */
 
-class sp_instr_cpush : public sp_instr, public sp_cursor
+class sp_instr_cpush : public sp_lex_instr, public sp_cursor
 {
   sp_instr_cpush(const sp_instr_cpush &); /**< Prevent use of these */
   void operator=(sp_instr_cpush &);
 
 public:
   sp_instr_cpush(uint ip, sp_pcontext *ctx, LEX *lex, uint offset)
-    : sp_instr(ip, ctx),
-      m_lex_keeper(lex, true),
-      m_cursor(offset)
+    : sp_lex_instr(ip, ctx, lex, true),
+      m_cursor(offset),
+      m_metadata_changed(false)
   {}
 
   virtual ~sp_instr_cpush() = default;
@@ -844,9 +921,28 @@ public:
     return false;
   }
 
+  bool is_invalid() const override
+  {
+    return m_metadata_changed;
+  }
+
+  void invalidate() override
+  {
+    m_metadata_changed= true;
+  }
+
+  sp_lex_keeper *get_lex_keeper() override
+  {
+    return &m_lex_keeper;
+  }
+
 private:
-  sp_lex_keeper m_lex_keeper;
   uint m_cursor;                /**< Frame offset (for debugging) */
+  /**
+    Flag if a statement's metadata has been changed in result of running DDL
+    on depending database objects used in the statement.
+  */
+  bool m_metadata_changed;
 
 public:
   PSI_statement_info* get_psi_info() override { return & psi_info; }
@@ -918,27 +1014,40 @@ public:
   from the LEX containing the cursor SELECT statement.
 */
 
-class sp_instr_cursor_copy_struct: public sp_instr
+class sp_instr_cursor_copy_struct: public sp_lex_instr
 {
   /**< Prevent use of these */
   sp_instr_cursor_copy_struct(const sp_instr_cursor_copy_struct &);
   void operator=(sp_instr_cursor_copy_struct &);
-  sp_lex_keeper m_lex_keeper;
   uint m_cursor;
   uint m_var;
+  /**
+    Flag to tell whether metadata has been changed and the LEX object should
+    be reinitialized.
+  */
+  bool m_valid;
 
 public:
   sp_instr_cursor_copy_struct(uint ip, sp_pcontext *ctx, uint coffs,
                               sp_lex_cursor *lex, uint voffs)
-    : sp_instr(ip, ctx),
-      m_lex_keeper(lex, false),
+    : sp_lex_instr(ip, ctx, lex, false),
       m_cursor(coffs),
-      m_var(voffs)
+      m_var(voffs),
+      m_valid(true)
   {}
   virtual ~sp_instr_cursor_copy_struct() = default;
   int execute(THD *thd, uint *nextp) override;
   int exec_core(THD *thd, uint *nextp) override;
   void print(String *str) override;
+  bool is_invalid() const override
+  {
+    return !m_valid;
+  }
+
+  void invalidate() override
+  {
+    m_valid= false;
+  }
 
 public:
   PSI_statement_info* get_psi_info() override { return & psi_info; }
@@ -1068,15 +1177,15 @@ public:
 }; // class sp_instr_error : public sp_instr
 
 
-class sp_instr_set_case_expr : public sp_instr_opt_meta
+class sp_instr_set_case_expr : public sp_lex_instr, public sp_instr_opt_meta
 {
 public:
   sp_instr_set_case_expr(uint ip, sp_pcontext *ctx, uint case_expr_id,
                          Item *case_expr, LEX *lex)
-    : sp_instr_opt_meta(ip, ctx),
+    : sp_lex_instr(ip, ctx, lex, true),
+      sp_instr_opt_meta(0),
       m_case_expr_id(case_expr_id),
-      m_case_expr(case_expr),
-      m_lex_keeper(lex, true)
+      m_case_expr(case_expr)
   {}
 
   virtual ~sp_instr_set_case_expr() = default;
@@ -1091,20 +1200,34 @@ public:
 
   void opt_move(uint dst, List<sp_instr_opt_meta> *ibp) override;
 
+  uint get_cont_dest() const override
+  {
+    return m_cont_dest;
+  }
+
   void set_destination(uint old_dest, uint new_dest) override
   {
     if (m_cont_dest == old_dest)
       m_cont_dest= new_dest;
   }
 
+  bool is_invalid() const override
+  {
+    return m_case_expr == nullptr;
+  }
+
+  void invalidate() override
+  {
+    m_case_expr= nullptr;
+  }
+
 private:
   uint m_case_expr_id;
   Item *m_case_expr;
-  sp_lex_keeper m_lex_keeper;
 
 public:
   PSI_statement_info* get_psi_info() override { return & psi_info; }
   static PSI_statement_info psi_info;
-}; // class sp_instr_set_case_expr : public sp_instr_opt_meta
+}; // class sp_instr_set_case_expr : public sp_lex_instr,
 
 #endif
diff --git a/sql/sql_class.h b/sql/sql_class.h
index 089b9ce5dbe..567be82eb38 100644
--- a/sql/sql_class.h
+++ b/sql/sql_class.h
@@ -6011,19 +6011,17 @@ private:
 public:
   sp_cursor()
    :result(NULL, false),
-    m_lex_keeper(NULL),
     server_side_cursor(NULL)
   { }
-  sp_cursor(THD *thd_arg, sp_lex_keeper *lex_keeper, bool view_structure_only)
+  sp_cursor(THD *thd_arg, bool view_structure_only)
    :result(thd_arg, view_structure_only),
-    m_lex_keeper(lex_keeper),
     server_side_cursor(NULL)
   {}
 
   virtual ~sp_cursor()
   { destroy(); }
 
-  sp_lex_keeper *get_lex_keeper() { return m_lex_keeper; }
+  virtual sp_lex_keeper *get_lex_keeper() { return nullptr; }
 
   int open(THD *thd);
 
@@ -6036,17 +6034,15 @@ public:
 
   bool export_structure(THD *thd, Row_definition_list *list);
 
-  void reset(THD *thd_arg, sp_lex_keeper *lex_keeper)
+  void reset(THD *thd_arg)
   {
     sp_cursor_statistics::reset();
     result.reset(thd_arg);
-    m_lex_keeper= lex_keeper;
     server_side_cursor= NULL;
   }
 
 private:
   Select_fetch_into_spvars result;
-  sp_lex_keeper *m_lex_keeper;
   Server_side_cursor *server_side_cursor;
   void destroy();
 };