Fixed on BUG#16568: Continue handler with simple CASE not working correctly

After trying multiple inheritance (to messy and hard make it work) and
  sublassing jump_if_not (worked, but ugly), decided to on this solution
  instead:
  Inserting an abstract sp_instr_opt_meta class as parent for all instructions
  with destinations makes it possible to handle a continuation pointer for
  sp_instr_set_case_expr too.
  Note: No special test case; the fix is captured by the changed behaviour of
  bug14643_2, and bug14498_4 (formerly disabled), in sp.test.

mysql-test/r/sp.result

@@ -4089,8 +4089,6 @@ NULL	1
 call bug14643_2()|
 Handler
 boo
-2
-2
 Handler
 boo
 drop procedure bug14643_1|
@@ -4432,6 +4430,11 @@ Handler
 error
 End
 done
+call bug14498_4()|
+Handler
+error
+End
+done
 call bug14498_5()|
 Handler
 error

mysql-test/t/sp.test

@@ -5210,10 +5210,7 @@ end|
 call bug14498_1()|
 call bug14498_2()|
 call bug14498_3()|
-# We couldn't call this before, due to a known bug (BUG#14643)
+call bug14498_4()|
-# QQ We still can't since the new set_case_expr instruction breaks
-#    the semantics of case; it won't crash, but will get the wrong result.
-#call bug14498_4()|
 call bug14498_5()|
 
 drop procedure bug14498_1|

sql/sp_head.cc

@@ -1761,7 +1761,7 @@ sp_head::fill_field_definition(THD *thd, LEX *lex,
 
 
 void
-sp_head::new_cont_backpatch(sp_instr_jump_if_not *i)
+sp_head::new_cont_backpatch(sp_instr_opt_meta *i)
 {
   m_cont_level+= 1;
   if (i)
@@ -1773,7 +1773,7 @@ sp_head::new_cont_backpatch(sp_instr_jump_if_not *i)
 }
 
 void
-sp_head::add_cont_backpatch(sp_instr_jump_if_not *i)
+sp_head::add_cont_backpatch(sp_instr_opt_meta *i)
 {
   i->m_cont_dest= m_cont_level;
   (void)m_cont_backpatch.push_front(i);
@@ -1784,7 +1784,7 @@ sp_head::do_cont_backpatch()
 {
   uint dest= instructions();
   uint lev= m_cont_level--;
-  sp_instr_jump_if_not *i;
+  sp_instr_opt_meta *i;
 
   while ((i= m_cont_backpatch.head()) && i->m_cont_dest == lev)
   {
@@ -2040,8 +2040,8 @@ void sp_head::optimize()
 	set_dynamic(&m_instr, (gptr)&i, dst);
 	while ((ibp= li++))
         {
-	  sp_instr_jump *ji= static_cast<sp_instr_jump *>(ibp);
+          sp_instr_opt_meta *im= static_cast<sp_instr_opt_meta *>(ibp);
-          ji->set_destination(src, dst);
+          im->set_destination(src, dst);
         }
       }
       i->opt_move(dst, &bp);
@@ -2064,6 +2064,10 @@ sp_head::opt_mark(uint ip)
 
 
 #ifndef DBUG_OFF
+/*
+  Return the routine instructions as a result set.
+  Returns 0 if ok, !=0 on error.
+*/
 int
 sp_head::show_routine_code(THD *thd)
 {
@@ -2091,6 +2095,22 @@ sp_head::show_routine_code(THD *thd)
 
   for (ip= 0; (i = get_instr(ip)) ; ip++)
   {
+    /* 
+      Consistency check. If these are different something went wrong
+      during optimization.
+    */
+    if (ip != i->m_ip)
+    {
+      const char *format= "Instruction at position %u has m_ip=%u";
+      char tmp[sizeof(format) + 2*SP_INSTR_UINT_MAXLEN + 1];
+
+      sprintf(tmp, format, ip, i->m_ip);
+      /*
+        Since this is for debugging purposes only, we don't bother to
+        introduce a special error code for it.
+      */
+      push_warning(thd, MYSQL_ERROR::WARN_LEVEL_WARN, ER_UNKNOWN_ERROR, tmp);
+    }
     protocol->prepare_for_resend();
     protocol->store((longlong)ip);
 
@@ -2515,14 +2535,14 @@ sp_instr_jump_if_not::exec_core(THD *thd, uint *nextp)
 void
 sp_instr_jump_if_not::print(String *str)
 {
-  /* jump_if_not dest ... */
+  /* jump_if_not dest(cont) ... */
   if (str->reserve(2*SP_INSTR_UINT_MAXLEN+14+32)) // Add some for the expr. too
     return;
   str->qs_append(STRING_WITH_LEN("jump_if_not "));
   str->qs_append(m_dest);
-  str->append('(');
+  str->qs_append('(');
   str->qs_append(m_cont_dest);
-  str->append(") ");
+  str->qs_append(STRING_WITH_LEN(") "));
   m_expr->print(str);
 }
 
@@ -3080,30 +3100,53 @@ sp_instr_set_case_expr::exec_core(THD *thd, uint *nextp)
       spcont->clear_handler();
       thd->spcont= spcont;
     }
+    *nextp= m_cont_dest;        /* For continue handler */
   }
+  else
+    *nextp= m_ip+1;
 
-  *nextp = m_ip+1;
+  return res;
-
-  return res; /* no error */
 }
 
 
 void
 sp_instr_set_case_expr::print(String *str)
 {
-  const char CASE_EXPR_TAG[]= "set_case_expr ";
+  /* set_case_expr (cont) id ... */
-  const int CASE_EXPR_TAG_LEN= sizeof(CASE_EXPR_TAG) - 1;
+  str->reserve(2*SP_INSTR_UINT_MAXLEN+18+32); // Add some extra for expr too
-  const int INT_STRING_MAX_LEN= 10;
+  str->qs_append(STRING_WITH_LEN("set_case_expr ("));
-
+  str->qs_append(m_cont_dest);
-  /* We must call reserve(), because qs_append() doesn't care about memory. */
+  str->qs_append(STRING_WITH_LEN(") "));
-  str->reserve(CASE_EXPR_TAG_LEN + INT_STRING_MAX_LEN + 2);
-
-  str->qs_append(CASE_EXPR_TAG, CASE_EXPR_TAG_LEN);
   str->qs_append(m_case_expr_id);
   str->qs_append(' ');
   m_case_expr->print(str);
 }
 
+uint
+sp_instr_set_case_expr::opt_mark(sp_head *sp)
+{
+  sp_instr *i;
+
+  marked= 1;
+  if ((i= sp->get_instr(m_cont_dest)))
+  {
+    m_cont_dest= i->opt_shortcut_jump(sp, this);
+    m_cont_optdest= sp->get_instr(m_cont_dest);
+  }
+  sp->opt_mark(m_cont_dest);
+  return m_ip+1;
+}
+
+void
+sp_instr_set_case_expr::opt_move(uint dst, List<sp_instr> *bp)
+{
+  if (m_cont_dest > m_ip)
+    bp->push_back(this);        // Forward
+  else if (m_cont_optdest)
+    m_cont_dest= m_cont_optdest->m_ip; // Backward
+  m_ip= dst;
+}
+
 
 /* ------------------------------------------------------------------ */
 

sql/sp_head.h

@@ -41,6 +41,7 @@ sp_get_flags_for_command(LEX *lex);
 
 struct sp_label;
 class sp_instr;
+class sp_instr_opt_meta;
 class sp_instr_jump_if_not;
 struct sp_cond_type;
 struct sp_pvar;
@@ -271,11 +272,11 @@ public:
 
   // Start a new cont. backpatch level. If 'i' is NULL, the level is just incr.
   void
-  new_cont_backpatch(sp_instr_jump_if_not *i);
+  new_cont_backpatch(sp_instr_opt_meta *i);
 
   // Add an instruction to the current level
   void
-  add_cont_backpatch(sp_instr_jump_if_not *i);
+  add_cont_backpatch(sp_instr_opt_meta *i);
 
   // Backpatch (and pop) the current level to the current position.
   void
@@ -372,15 +373,15 @@ private:
   } bp_t;
   List<bp_t> m_backpatch;	// Instructions needing backpatching
   /*
-    We need a special list for backpatching of conditional jump's continue
+    We need a special list for backpatching of instructions with a continue
     destination (in the case of a continue handler catching an error in
     the test), since it would otherwise interfere with the normal backpatch
-    mechanism - jump_if_not instructions have two different destination
+    mechanism - e.g. jump_if_not instructions have two different destinations
     which are to be patched differently.
     Since these occur in a more restricted way (always the same "level" in
     the code), we don't need the label.
    */
-  List<sp_instr_jump_if_not> m_cont_backpatch;
+  List<sp_instr_opt_meta> m_cont_backpatch;
   uint m_cont_level;            // The current cont. backpatch level
 
   /*
@@ -660,21 +661,55 @@ private:
 }; // class sp_instr_trigger_field : public sp_instr
 
 
-class sp_instr_jump : public sp_instr
+/*
+  An abstract class for all instructions with destinations that
+  needs to be updated by the optimizer.
+  Even if not all subclasses will use both the normal destination and
+  the continuation destination, we put them both here for simplicity.
+ */
+class sp_instr_opt_meta : public sp_instr
+{
+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), m_cont_dest(0), m_optdest(0), m_cont_optdest(0)
+  {}
+
+  virtual ~sp_instr_opt_meta()
+  {}
+
+  virtual void set_destination(uint old_dest, uint new_dest)
+    = 0;
+
+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_jump : public sp_instr_opt_meta
 {
   sp_instr_jump(const sp_instr_jump &);	/* Prevent use of these */
   void operator=(sp_instr_jump &);
 
 public:
 
-  uint m_dest;			// Where we will go
-
   sp_instr_jump(uint ip, sp_pcontext *ctx)
-    : sp_instr(ip, ctx), m_dest(0), m_optdest(0)
+    : sp_instr_opt_meta(ip, ctx)
   {}
 
   sp_instr_jump(uint ip, sp_pcontext *ctx, uint dest)
-    : sp_instr(ip, ctx), m_dest(dest), m_optdest(0)
+    : sp_instr_opt_meta(ip, ctx, dest)
   {}
 
   virtual ~sp_instr_jump()
@@ -702,11 +737,7 @@ public:
       m_dest= new_dest;
   }
 
-protected:
+}; // class sp_instr_jump : public sp_instr_opt_meta
-
-  sp_instr *m_optdest;		// Used during optimization
-
-}; // class sp_instr_jump : public sp_instr
 
 
 class sp_instr_jump_if_not : public sp_instr_jump
@@ -716,16 +747,14 @@ class sp_instr_jump_if_not : public sp_instr_jump
 
 public:
 
-  uint m_cont_dest;             // Where continue handlers will go
-
   sp_instr_jump_if_not(uint ip, sp_pcontext *ctx, Item *i, LEX *lex)
-    : sp_instr_jump(ip, ctx), m_cont_dest(0), m_expr(i),
+    : sp_instr_jump(ip, ctx), m_expr(i),
-      m_lex_keeper(lex, TRUE), m_cont_optdest(0)
+      m_lex_keeper(lex, TRUE)
   {}
 
   sp_instr_jump_if_not(uint ip, sp_pcontext *ctx, Item *i, uint dest, LEX *lex)
-    : sp_instr_jump(ip, ctx, dest), m_cont_dest(0), m_expr(i),
+    : sp_instr_jump(ip, ctx, dest), m_expr(i),
-      m_lex_keeper(lex, TRUE), m_cont_optdest(0)
+      m_lex_keeper(lex, TRUE)
   {}
 
   virtual ~sp_instr_jump_if_not()
@@ -757,7 +786,6 @@ private:
 
   Item *m_expr;			// The condition
   sp_lex_keeper m_lex_keeper;
-  sp_instr *m_cont_optdest;     // Used during optimization
 
 }; // class sp_instr_jump_if_not : public sp_instr_jump
 
@@ -899,7 +927,7 @@ private:
 
   uint m_frame;
 
-}; // class sp_instr_hreturn : public sp_instr
+}; // class sp_instr_hreturn : public sp_instr_jump
 
 
 /* This is DECLARE CURSOR */
@@ -1085,29 +1113,43 @@ private:
 }; // class sp_instr_error : public sp_instr
 
 
-class sp_instr_set_case_expr :public sp_instr
+class sp_instr_set_case_expr : 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(ip, ctx), m_case_expr_id(case_expr_id), m_case_expr(case_expr),
+    : sp_instr_opt_meta(ip, ctx),
+      m_case_expr_id(case_expr_id), m_case_expr(case_expr),
       m_lex_keeper(lex, TRUE)
   {}
 
+  virtual ~sp_instr_set_case_expr()
+  {}
+
   virtual int execute(THD *thd, uint *nextp);
 
   virtual int exec_core(THD *thd, uint *nextp);
 
   virtual void print(String *str);
 
+  virtual uint opt_mark(sp_head *sp);
+
+  virtual void opt_move(uint dst, List<sp_instr> *ibp);
+
+  virtual void set_destination(uint old_dest, uint new_dest)
+  {
+    if (m_cont_dest == old_dest)
+      m_cont_dest= new_dest;
+  }
+
 private:
 
   uint m_case_expr_id;
   Item *m_case_expr;
   sp_lex_keeper m_lex_keeper;
 
-}; // class sp_instr_set_case_expr : public sp_instr
+}; // class sp_instr_set_case_expr : public sp_instr_opt_meta
 
 
 #ifndef NO_EMBEDDED_ACCESS_CHECKS

sql/sql_parse.cc

@@ -4599,7 +4599,7 @@ end_with_restore_list:
       else
         sp= sp_find_routine(thd, TYPE_ENUM_FUNCTION, lex->spname,
                             &thd->sp_func_cache, FALSE);
-      if (!sp || !sp->show_routine_code(thd))
+      if (!sp || sp->show_routine_code(thd))
       {
         /* We don't distinguish between errors for now */
         my_error(ER_SP_DOES_NOT_EXIST, MYF(0),

sql/sql_yacc.yy

@@ -2018,17 +2018,18 @@ sp_proc_stmt:
 	    sp_head *sp= lex->sphead;
 	    sp_pcontext *parsing_ctx= lex->spcont;
 	    int case_expr_id= parsing_ctx->register_case_expr();
+            sp_instr_set_case_expr *i;
 	    
 	    if (parsing_ctx->push_case_expr_id(case_expr_id))
               YYABORT;
 
-	    sp->add_instr(
+            i= new sp_instr_set_case_expr(sp->instructions(),
-	      new sp_instr_set_case_expr(sp->instructions(),
                                           parsing_ctx,
                                           case_expr_id,
                                           $3,
-	                                 lex));
+                                          lex);
-	    
+            sp->add_cont_backpatch(i);
+            sp->add_instr(i);
 	    sp->m_flags|= sp_head::IN_SIMPLE_CASE;
 	    sp->restore_lex(YYTHD);
 	  }