Bug#19194 (Right recursion in parser for CASE causes excessive stack usage,

  limitation)

Note to the reviewer
====================

Warning: reviewing this patch is somewhat involved.
Due to the nature of several issues all affecting the same area,
fixing separately each issue is not practical, since each fix can not be
implemented and tested independently.
In particular, the issues with
- rule recursion
- nested case statements
- forward jump resolution (backpatch list)
are tightly coupled (see below).

Definitions
===========

The expression
  CASE expr
  WHEN expr THEN expr
  WHEN expr THEN expr
  ...
  END
is a "Simple Case Expression".

The expression
  CASE
  WHEN expr THEN expr
  WHEN expr THEN expr
  ...
  END
is a "Searched Case Expression".

The statement
  CASE expr
  WHEN expr THEN stmts
  WHEN expr THEN stmts
  ...
  END CASE
is a "Simple Case Statement".

The statement
  CASE
  WHEN expr THEN stmts
  WHEN expr THEN stmts
  ...
  END CASE
is a "Searched Case Statement".

A "Left Recursive" rule is like
  list:
      element
    | list element
    ;

A "Right Recursive" rule is like
  list:
      element
    | element list
    ;

Left and right recursion produces the same language, the difference only
affects the *order* in which the text is parsed.

In a descendant parser (usually written manually), right recursion works
very well, and is typically implemented with a while loop.
In an ascendant parser (yacc/bison) left recursion works very well,
and is implemented naturally by the parser stack.
In both cases, using the wrong type or recursion is very bad and should be
avoided, as it causes technical issues with the parser implementation.

Before this change
==================

The "Simple Case Expression" and "Searched Case Expression" were both
implemented by the "when_list" and "when_list2" rules, which are left
recursive (ok).

These rules, however, used lex->when_list instead of using the parser stack,
which is more complex that necessary, and potentially dangerous because
of other rules using THD::reset_lex.

The "Simple Case Statement" and "Searched Case Statements" were implemented
by the "sp_case", "sp_whens" and in part by "sp_proc_stmt" rules.
Both cases were right recursive (bad).

The grammar involved was convoluted, and is assumed to be the results of
tweaks to get the code generation to work, but is not what someone would
naturally write.

In addition, using a common rule for both "Simple" and "Searched" case
statements was implemented with sp_head::m_flags |= IN_SIMPLE_CASE,
which is a flag and not a stack, and therefore does not take into account
*nested* case statements. This leads to incorrect generated code, and either
a server crash or an incorrect result.

With regards to the backpatch mechanism, a *different* backpatch list was
created for each jump from "WHEN expr THEN stmt" to "END CASE", which
relied on the grammar to be right recursive.
This is a mis-use of the backpatch list, since this list can resolve
multiple references to the same target at once.

The optimizer algorithm used to detect dead code in the "assembly" SQL
instructions, implemented by sp_head::opt_mark(uint ip), was recursive
in some cases (a conditional jump pointing forward to another conditional
jump).
In case of specially crafted code, like
- a long list of "IF expr THEN stmt END IF"
- a long CASE statement
this would actually cause a server crash with a stack overflow.
In general, having a stack that grows proportionally with user data (the
SQL code given by the client in a CREATE PROCEDURE) is to be avoided.

In debug builds only, creating a SP / SF / Trigger which had a significant
amount of code would spend --literally-- several minutes in sp_head::create,
because of the debug code involved with DBUG_PRINT("info", ("Code %s ...
There are several issues with this code:
- in a CASE with 5 000 WHEN, there are 15 000 instructions generated,
  which create a sting representation of the code which is 500 000 bytes
  long,
- using a String instead of an io stream causes performances to degrade
  to a total server freeze, as time is spent doing realloc of a buffer
  always too short,
- Printing a 500 000 long string in the debug log is too verbose,
- Generating this string even when DBUG_PRINT is off is useless,
- Having code that potentially can affect the server behavior, used with
  #ifdef / #endif is useful in some cases, but is also a bad practice.

After this change
=================

"Case Expressions" (both simple and searched) have been simplified to
not use LEX::when_list, which has been removed.

Considering all the issues affecting case statements, the grammar for these
has been totally re written.

The existing actions, used to generate "assembly" sp_inst* code, have been
preserved but moved in the new grammar, with the following changes:

a) Bison rules are no longer shared between "Simple" and "Searched" case
statements, because a stack instead of a flag is required to handle them.
Nested statements are handled naturally by the parser stack, which by
definition uses the correct rule in the correct context.
Nested statements of the opposite type (simple vs searched) works correctly.
The flag sp_head::IN_SIMPLE_CASE is no longer used.
This is a step towards resolution of WL#2999, which correctly identified
that temporary parsing flags do not belong to sp_head.
The code in the action is shared by mean of the case_stmt_action_xxx()
helpers.

b) The backpatch mechanism, used to resolve forward jumps in the generated
code, has been changed to:
- create a label for the instruction following 'END CASE',
- register each jump at the end of a "WHEN expr THEN stmt" in a *unique*
  backpatch list associated with the 'END CASE' label
- resolve all the forward jumps for this label at once.

In addition, the code involving backpatch has been commented, so that a
reader can now understand by reading matching "Registering" and "Resolving"
comments how the forward jumps are resolved and what target they resolve to,
as this is far from evident when reading the code alone.

The implementation of sp_head::opt_mark() has been revised to avoid
recursive calls from jump instructions, and instead add the jump location
to the list of paths to explore during the flow analysis of the instruction
graph, with a call to sp_head::add_mark_lead().
In addition, the flow analysis will stop if an instruction has already
been marked as reachable, which the previous code failed to do in the
recursive case.
sp_head::opt_mark() is now private, to prevent new calls to this method from
being introduced.

The debug code present in sp_head::create() has been removed.
Considering that SHOW PROCEDURE CODE is also available in debug builds,
and can be used anytime regardless of the trace level, as opposed to
"CREATE PROCEDURE" time and only if the trace was on,
removing the code actually makes debugging easier (usable trace).

Tests have been written to cover the parser overflow (big CASE),
and to cover nested CASE statements.


mysql-test/r/sp-code.result:
  Test cases for nested CASE statements.
mysql-test/t/sp-code.test:
  Test cases for nested CASE statements.
sql/sp_head.cc:
  Re factored opt_mark() to avoid recursion, clean up.
sql/sp_head.h:
  Re factored opt_mark() to avoid recursion, clean up.
sql/sql_lex.cc:
  Removed when_list.
sql/sql_lex.h:
  Removed when_list.
sql/sql_yacc.yy:
  Minor clean up for case expressions,
  Major re write for case statements (Bug#19194).
mysql-test/r/sp_stress_case.result:
  New test for massive CASE statements.
mysql-test/t/sp_stress_case.sh:
  New test for massive CASE statements.
mysql-test/t/sp_stress_case.test:
  New test for massive CASE statements.

5 files changed, 340 insertions, 168 deletions

diff --git a/sql/sp_head.cc b/sql/sp_head.cc
index 9761de625be..689922cfa37 100644
--- a/sql/sp_head.cc
+++ b/sql/sp_head.cc
@@ -603,27 +603,6 @@ sp_head::create(THD *thd)
   DBUG_PRINT("info", ("type: %d name: %s params: %s body: %s",
 		      m_type, m_name.str, m_params.str, m_body.str));
 
-#ifndef DBUG_OFF
-  optimize();
-  {
-    String s;
-    sp_instr *i;
-    uint ip= 0;
-    while ((i = get_instr(ip)))
-    {
-      char buf[8];
-
-      sprintf(buf, "%4u: ", ip);
-      s.append(buf);
-      i->print(&s);
-      s.append('\n');
-      ip+= 1;
-    }
-    s.append('\0');
-    DBUG_PRINT("info", ("Code %s\n%s", m_qname.str, s.ptr()));
-  }
-#endif
-
   if (m_type == TYPE_ENUM_FUNCTION)
     ret= sp_create_function(thd, this);
   else
@@ -2172,7 +2151,7 @@ sp_head::show_create_function(THD *thd)
   This is the main mark and move loop; it relies on the following methods
   in sp_instr and its subclasses:
 
-  opt_mark()           Mark instruction as reachable (will recurse for jumps)
+  opt_mark()           Mark instruction as reachable
   opt_shortcut_jump()  Shortcut jumps to the final destination;
                        used by opt_mark().
   opt_move()           Update moved instruction
@@ -2185,7 +2164,7 @@ void sp_head::optimize()
   sp_instr *i;
   uint src, dst;
 
-  opt_mark(0);
+  opt_mark();
 
   bp.empty();
   src= dst= 0;
@@ -2219,13 +2198,50 @@ void sp_head::optimize()
   bp.empty();
 }
 
+void sp_head::add_mark_lead(uint ip, List<sp_instr> *leads)
+{
+  sp_instr *i= get_instr(ip);
+
+  if (i && ! i->marked)
+    leads->push_front(i);
+}
+
 void
-sp_head::opt_mark(uint ip)
+sp_head::opt_mark()
 {
+  uint ip;
   sp_instr *i;
+  List<sp_instr> leads;
 
-  while ((i= get_instr(ip)) && !i->marked)
-    ip= i->opt_mark(this);
+  /*
+    Forward flow analysis algorithm in the instruction graph:
+    - first, add the entry point in the graph (the first instruction) to the
+      'leads' list of paths to explore.
+    - while there are still leads to explore:
+      - pick one lead, and follow the path forward. Mark instruction reached.
+        Stop only if the end of the routine is reached, or the path converge
+        to code already explored (marked).
+      - while following a path, collect in the 'leads' list any fork to
+        another path (caused by conditional jumps instructions), so that these
+        paths can be explored as well.
+  */
+
+  /* Add the entry point */
+  i= get_instr(0);
+  leads.push_front(i);
+
+  /* For each path of code ... */
+  while (leads.elements != 0)
+  {
+    i= leads.pop();
+
+    /* Mark the entire path, collecting new leads. */
+    while (i && ! i->marked)
+    {
+      ip= i->opt_mark(this, & leads);
+      i= get_instr(ip);
+    }
+  }
 }
 
 
@@ -2618,7 +2634,7 @@ sp_instr_jump::print(String *str)
 }
 
 uint
-sp_instr_jump::opt_mark(sp_head *sp)
+sp_instr_jump::opt_mark(sp_head *sp, List<sp_instr> *leads)
 {
   m_dest= opt_shortcut_jump(sp, this);
   if (m_dest != m_ip+1)		/* Jumping to following instruction? */
@@ -2712,7 +2728,7 @@ sp_instr_jump_if_not::print(String *str)
 
 
 uint
-sp_instr_jump_if_not::opt_mark(sp_head *sp)
+sp_instr_jump_if_not::opt_mark(sp_head *sp, List<sp_instr> *leads)
 {
   sp_instr *i;
 
@@ -2722,13 +2738,13 @@ sp_instr_jump_if_not::opt_mark(sp_head *sp)
     m_dest= i->opt_shortcut_jump(sp, this);
     m_optdest= sp->get_instr(m_dest);
   }
-  sp->opt_mark(m_dest);
+  sp->add_mark_lead(m_dest, leads);
   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);
+  sp->add_mark_lead(m_cont_dest, leads);
   return m_ip+1;
 }
 
@@ -2849,7 +2865,7 @@ sp_instr_hpush_jump::print(String *str)
 
 
 uint
-sp_instr_hpush_jump::opt_mark(sp_head *sp)
+sp_instr_hpush_jump::opt_mark(sp_head *sp, List<sp_instr> *leads)
 {
   sp_instr *i;
 
@@ -2859,7 +2875,7 @@ sp_instr_hpush_jump::opt_mark(sp_head *sp)
     m_dest= i->opt_shortcut_jump(sp, this);
     m_optdest= sp->get_instr(m_dest);
   }
-  sp->opt_mark(m_dest);
+  sp->add_mark_lead(m_dest, leads);
   return m_ip+1;
 }
 
@@ -2924,15 +2940,13 @@ sp_instr_hreturn::print(String *str)
 
 
 uint
-sp_instr_hreturn::opt_mark(sp_head *sp)
+sp_instr_hreturn::opt_mark(sp_head *sp, List<sp_instr> *leads)
 {
   if (m_dest)
-    return sp_instr_jump::opt_mark(sp);
-  else
-  {
-    marked= 1;
-    return UINT_MAX;
-  }
+    return sp_instr_jump::opt_mark(sp, leads);
+
+  marked= 1;
+  return UINT_MAX;
 }
 
 
@@ -3275,7 +3289,7 @@ sp_instr_set_case_expr::print(String *str)
 }
 
 uint
-sp_instr_set_case_expr::opt_mark(sp_head *sp)
+sp_instr_set_case_expr::opt_mark(sp_head *sp, List<sp_instr> *leads)
 {
   sp_instr *i;
 
@@ -3285,7 +3299,7 @@ sp_instr_set_case_expr::opt_mark(sp_head *sp)
     m_cont_dest= i->opt_shortcut_jump(sp, this);
     m_cont_optdest= sp->get_instr(m_cont_dest);
   }
-  sp->opt_mark(m_cont_dest);
+  sp->add_mark_lead(m_cont_dest, leads);
   return m_ip+1;
 }
 
diff --git a/sql/sp_head.h b/sql/sp_head.h
index 7f2da69aa0c..6a377e6f188 100644
--- a/sql/sp_head.h
+++ b/sql/sp_head.h
@@ -302,8 +302,19 @@ public:
 
   void restore_thd_mem_root(THD *thd);
 
+  /**
+    Optimize the code.
+  */
   void optimize();
-  void opt_mark(uint ip);
+
+  /**
+    Helper used during flow analysis during code optimization.
+    See the implementation of <code>opt_mark()</code>.
+    @param ip the instruction to add to the leads list
+    @param leads the list of remaining paths to explore in the graph that
+    represents the code, during flow analysis.
+  */
+  void add_mark_lead(uint ip, List<sp_instr> *leads);
 
   void recursion_level_error(THD *thd);
 
@@ -393,6 +404,12 @@ private:
   bool
   execute(THD *thd);
 
+  /**
+    Perform a forward flow analysis in the generated code.
+    Mark reachable instructions, for the optimizer.
+  */
+  void opt_mark();
+
   /*
     Merge the list of tables used by query into the multi-set of tables used
     by routine.
@@ -460,10 +477,10 @@ public:
 
   /*
     Mark this instruction as reachable during optimization and return the
-    index to the next instruction. Jump instruction will mark their
-    destination too recursively.
+    index to the next instruction. Jump instruction will add their
+    destination to the leads list.
   */
-  virtual uint opt_mark(sp_head *sp)
+  virtual uint opt_mark(sp_head *sp, List<sp_instr> *leads)
   {
     marked= 1;
     return m_ip+1;
@@ -735,7 +752,7 @@ public:
 
   virtual void print(String *str);
 
-  virtual uint opt_mark(sp_head *sp);
+  virtual uint opt_mark(sp_head *sp, List<sp_instr> *leads);
 
   virtual uint opt_shortcut_jump(sp_head *sp, sp_instr *start);
 
@@ -785,7 +802,7 @@ public:
 
   virtual void print(String *str);
 
-  virtual uint opt_mark(sp_head *sp);
+  virtual uint opt_mark(sp_head *sp, List<sp_instr> *leads);
 
   /* Override sp_instr_jump's shortcut; we stop here */
   virtual uint opt_shortcut_jump(sp_head *sp, sp_instr *start)
@@ -831,7 +848,7 @@ public:
 
   virtual void print(String *str);
 
-  virtual uint opt_mark(sp_head *sp)
+  virtual uint opt_mark(sp_head *sp, List<sp_instr> *leads)
   {
     marked= 1;
     return UINT_MAX;
@@ -868,7 +885,7 @@ public:
 
   virtual void print(String *str);
 
-  virtual uint opt_mark(sp_head *sp);
+  virtual uint opt_mark(sp_head *sp, List<sp_instr> *leads);
 
   /* Override sp_instr_jump's shortcut; we stop here. */
   virtual uint opt_shortcut_jump(sp_head *sp, sp_instr *start)
@@ -933,7 +950,7 @@ public:
 
   virtual void print(String *str);
 
-  virtual uint opt_mark(sp_head *sp);
+  virtual uint opt_mark(sp_head *sp, List<sp_instr> *leads);
 
 private:
 
@@ -1103,7 +1120,7 @@ public:
 
   virtual void print(String *str);
 
-  virtual uint opt_mark(sp_head *sp)
+  virtual uint opt_mark(sp_head *sp, List<sp_instr> *leads)
   {
     marked= 1;
     return UINT_MAX;
@@ -1136,7 +1153,7 @@ public:
 
   virtual void print(String *str);
 
-  virtual uint opt_mark(sp_head *sp);
+  virtual uint opt_mark(sp_head *sp, List<sp_instr> *leads);
 
   virtual void opt_move(uint dst, List<sp_instr> *ibp);
 
diff --git a/sql/sql_lex.cc b/sql/sql_lex.cc
index 405f576ac04..71e7d1cf86f 100644
--- a/sql/sql_lex.cc
+++ b/sql/sql_lex.cc
@@ -1163,7 +1163,6 @@ void st_select_lex::init_select()
   options= 0;
   sql_cache= SQL_CACHE_UNSPECIFIED;
   braces= 0;
-  when_list.empty();
   expr_list.empty();
   interval_list.empty();
   use_index.empty();
diff --git a/sql/sql_lex.h b/sql/sql_lex.h
index 378f968118e..8f4b021ca00 100644
--- a/sql/sql_lex.h
+++ b/sql/sql_lex.h
@@ -518,7 +518,6 @@ public:
 
   SQL_LIST order_list;                /* ORDER clause */
   List<List_item>     expr_list;
-  List<List_item>     when_list;      /* WHEN clause (expression) */
   SQL_LIST *gorder_list;
   Item *select_limit, *offset_limit;  /* LIMIT clause parameters */
   // Arrays of pointers to top elements of all_fields list
diff --git a/sql/sql_yacc.yy b/sql/sql_yacc.yy
index 6f24a42c07c..57b6201d4a9 100644
--- a/sql/sql_yacc.yy
+++ b/sql/sql_yacc.yy
@@ -96,6 +96,146 @@ void turn_parser_debug_on()
 }
 #endif
 
+
+/**
+  Helper action for a case statement (entering the CASE).
+  This helper is used for both 'simple' and 'searched' cases.
+  @param lex the parser lex context
+*/
+
+void case_stmt_action_case(LEX *lex)
+{
+  lex->sphead->new_cont_backpatch(NULL);
+
+  /*
+    BACKPATCH: Creating target label for the jump to
+    "case_stmt_action_end_case"
+  */
+
+  lex->spcont->push_label((char *)"", lex->sphead->instructions());
+}
+
+/**
+  Helper action for a case expression statement (the expr in 'CASE expr').
+  This helper is used for 'searched' cases only.
+  @param lex the parser lex context
+  @param expr the parsed expression
+  @return 0 on success
+*/
+
+int case_stmt_action_expr(LEX *lex, Item* expr)
+{
+  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))
+    return 1;
+
+  i= new sp_instr_set_case_expr(sp->instructions(),
+                                parsing_ctx, case_expr_id, expr, lex);
+
+  sp->add_cont_backpatch(i);
+  sp->add_instr(i);
+
+  return 0;
+}
+
+/**
+  Helper action for a case when condition.
+  This helper is used for both 'simple' and 'searched' cases.
+  @param lex the parser lex context
+  @param when the parsed expression for the WHEN clause
+  @param simple true for simple cases, false for searched cases
+*/
+
+void case_stmt_action_when(LEX *lex, Item *when, bool simple)
+{
+  sp_head *sp= lex->sphead;
+  sp_pcontext *ctx= lex->spcont;
+  uint ip= sp->instructions();
+  sp_instr_jump_if_not *i;
+  Item_case_expr *var;
+  Item *expr;
+
+  if (simple)
+  {
+    var= new Item_case_expr(ctx->get_current_case_expr_id());
+
+#ifndef DBUG_OFF
+    if (var)
+    {
+      var->m_sp= sp;
+    }
+#endif
+
+    expr= new Item_func_eq(var, when);
+    i= new sp_instr_jump_if_not(ip, ctx, expr, lex);
+  }
+  else
+    i= new sp_instr_jump_if_not(ip, ctx, when, lex);
+
+  /*
+    BACKPATCH: Registering forward jump from
+    "case_stmt_action_when" to "case_stmt_action_then"
+  */
+
+  sp->push_backpatch(i, ctx->push_label((char *)"", 0));
+  sp->add_cont_backpatch(i);
+  sp->add_instr(i);
+}
+
+/**
+  Helper action for a case then statements.
+  This helper is used for both 'simple' and 'searched' cases.
+  @param lex the parser lex context
+*/
+
+void case_stmt_action_then(LEX *lex)
+{
+  sp_head *sp= lex->sphead;
+  sp_pcontext *ctx= lex->spcont;
+  uint ip= sp->instructions();
+  sp_instr_jump *i = new sp_instr_jump(ip, ctx);
+  sp->add_instr(i);
+
+  /*
+    BACKPATCH: Resolving forward jump from
+    "case_stmt_action_when" to "case_stmt_action_then"
+  */
+
+  sp->backpatch(ctx->pop_label());
+
+  /*
+    BACKPATCH: Registering forward jump from
+    "case_stmt_action_when" to "case_stmt_action_end_case"
+  */
+
+  sp->push_backpatch(i, ctx->last_label());
+}
+
+/**
+  Helper action for an end case.
+  This helper is used for both 'simple' and 'searched' cases.
+  @param lex the parser lex context
+  @param simple true for simple cases, false for searched cases
+*/
+
+void case_stmt_action_end_case(LEX *lex, bool simple)
+{
+  /*
+    BACKPATCH: Resolving forward jump from
+    "case_stmt_action_then" to "case_stmt_action_end_case"
+  */
+  lex->sphead->backpatch(lex->spcont->pop_label());
+
+  if (simple)
+    lex->spcont->pop_case_expr_id();
+
+  lex->sphead->do_cont_backpatch();
+}
+
 %}
 %union {
   int  num;
@@ -832,7 +972,7 @@ bool my_yyoverflow(short **a, YYSTYPE **b, ulong *yystacksize);
 	select_item_list select_item values_list no_braces
 	opt_limit_clause delete_limit_clause fields opt_values values
 	procedure_list procedure_list2 procedure_item
-	when_list2 expr_list2 udf_expr_list3 handler
+	expr_list2 udf_expr_list3 handler
 	opt_precision opt_ignore opt_column opt_restrict
 	grant revoke set lock unlock string_list field_options field_option
 	field_opt_list opt_binary table_lock_list table_lock
@@ -860,6 +1000,7 @@ bool my_yyoverflow(short **a, YYSTYPE **b, ulong *yystacksize);
         view_algorithm view_or_trigger_or_sp view_or_trigger_or_sp_tail
 	view_suid view_tail view_list_opt view_list view_select
 	view_check_option trigger_tail sp_tail
+        case_stmt_specification simple_case_stmt searched_case_stmt
 END_OF_INPUT
 
 %type <NONE> call sp_proc_stmts sp_proc_stmts1 sp_proc_stmt
@@ -1995,43 +2136,7 @@ sp_proc_stmt:
           { Lex->sphead->new_cont_backpatch(NULL); }
           sp_if END IF
           { Lex->sphead->do_cont_backpatch(); }
-	| CASE_SYM WHEN_SYM
-	  {
-	    Lex->sphead->m_flags&= ~sp_head::IN_SIMPLE_CASE;
-            Lex->sphead->new_cont_backpatch(NULL);
-	  }
-          sp_case END CASE_SYM { Lex->sphead->do_cont_backpatch(); }
-        | CASE_SYM
-          {
-            Lex->sphead->reset_lex(YYTHD);
-            Lex->sphead->new_cont_backpatch(NULL);
-          }
-          expr WHEN_SYM
-	  {
-	    LEX *lex= Lex;
-	    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;
-
-            i= new sp_instr_set_case_expr(sp->instructions(),
-                                          parsing_ctx,
-                                          case_expr_id,
-                                          $3,
-                                          lex);
-            sp->add_cont_backpatch(i);
-            sp->add_instr(i);
-	    sp->m_flags|= sp_head::IN_SIMPLE_CASE;
-	    sp->restore_lex(YYTHD);
-	  }
-	  sp_case END CASE_SYM
-	  {
-	    Lex->spcont->pop_case_expr_id();
-            Lex->sphead->do_cont_backpatch();
-	  }
+        | case_stmt_specification
 	| sp_labeled_control
 	  {}
 	| { /* Unlabeled controls get a secret label. */
@@ -2242,72 +2347,114 @@ sp_elseifs:
 	| ELSE sp_proc_stmts1
 	;
 
-sp_case:
-	  { Lex->sphead->reset_lex(YYTHD); }
-          expr THEN_SYM
-	  {
-            LEX *lex= Lex;
-	    sp_head *sp= lex->sphead;
-	    sp_pcontext *ctx= Lex->spcont;
-	    uint ip= sp->instructions();
-	    sp_instr_jump_if_not *i;
-
-	    if (! (sp->m_flags & sp_head::IN_SIMPLE_CASE))
-	      i= new sp_instr_jump_if_not(ip, ctx, $2, lex);
-	    else
-	    { /* Simple case: <caseval> = <whenval> */
+case_stmt_specification:
+          simple_case_stmt
+        | searched_case_stmt
+        ;
 
-	      Item_case_expr *var;
-              Item *expr;
+simple_case_stmt:
+          CASE_SYM
+          {
+            LEX *lex= Lex;
+            case_stmt_action_case(lex);
+            lex->sphead->reset_lex(YYTHD); /* For expr $3 */
+          }
+          expr
+          {
+            LEX *lex= Lex;
+            if (case_stmt_action_expr(lex, $3))
+              YYABORT;
 
-              var= new Item_case_expr(ctx->get_current_case_expr_id());
+            lex->sphead->restore_lex(YYTHD); /* For expr $3 */
+          }
+          simple_when_clause_list
+          else_clause_opt
+          END
+          CASE_SYM
+          {
+            LEX *lex= Lex;
+            case_stmt_action_end_case(lex, true);
+          }
+        ;
 
-#ifndef DBUG_OFF
-              if (var)
-                var->m_sp= sp;
-#endif
+searched_case_stmt:
+          CASE_SYM
+          {
+            LEX *lex= Lex;
+            case_stmt_action_case(lex);
+          }
+          searched_when_clause_list
+          else_clause_opt
+          END
+          CASE_SYM
+          {
+            LEX *lex= Lex;
+            case_stmt_action_end_case(lex, false);
+          }
+        ;
 
-	      expr= new Item_func_eq(var, $2);
+simple_when_clause_list:
+          simple_when_clause
+        | simple_when_clause_list simple_when_clause
+        ;
 
-	      i= new sp_instr_jump_if_not(ip, ctx, expr, lex);
-	    }
-	    sp->push_backpatch(i, ctx->push_label((char *)"", 0));
-            sp->add_cont_backpatch(i);
-            sp->add_instr(i);
-            sp->restore_lex(YYTHD);
-	  }
-	  sp_proc_stmts1
-	  {
-	    sp_head *sp= Lex->sphead;
-	    sp_pcontext *ctx= Lex->spcont;
-	    uint ip= sp->instructions();
-	    sp_instr_jump *i = new sp_instr_jump(ip, ctx);
+searched_when_clause_list:
+          searched_when_clause
+        | searched_when_clause_list searched_when_clause
+        ;
 
-	    sp->add_instr(i);
-	    sp->backpatch(ctx->pop_label());
-	    sp->push_backpatch(i, ctx->push_label((char *)"", 0));
-	  }
-	  sp_whens
-	  {
-	    LEX *lex= Lex;
+simple_when_clause:
+          WHEN_SYM
+          {
+            Lex->sphead->reset_lex(YYTHD); /* For expr $3 */
+          }
+          expr
+          {
+            /* Simple case: <caseval> = <whenval> */
 
-	    lex->sphead->backpatch(lex->spcont->pop_label());
-	  }
-	;
+            LEX *lex= Lex;
+            case_stmt_action_when(lex, $3, true);
+            lex->sphead->restore_lex(YYTHD); /* For expr $3 */
+          }
+          THEN_SYM
+          sp_proc_stmts1
+          {
+            LEX *lex= Lex;
+            case_stmt_action_then(lex);
+          }
+        ;
 
-sp_whens:
-	  /* Empty */
-	  {
-	    sp_head *sp= Lex->sphead;
-	    uint ip= sp->instructions();
-	    sp_instr_error *i= new sp_instr_error(ip, Lex->spcont,
-						  ER_SP_CASE_NOT_FOUND);
+searched_when_clause:
+          WHEN_SYM
+          {
+            Lex->sphead->reset_lex(YYTHD); /* For expr $3 */
+          }
+          expr
+          {
+            LEX *lex= Lex;
+            case_stmt_action_when(lex, $3, false);
+            lex->sphead->restore_lex(YYTHD); /* For expr $3 */
+          }
+          THEN_SYM
+          sp_proc_stmts1
+          {
+            LEX *lex= Lex;
+            case_stmt_action_then(lex);
+          }
+        ;
 
-	    sp->add_instr(i);
-	  }
-	| ELSE sp_proc_stmts1 {}
-	| WHEN_SYM sp_case {}
-	;
+else_clause_opt:
+          /* empty */
+          {
+            LEX *lex= Lex;
+            sp_head *sp= lex->sphead;
+            uint ip= sp->instructions();
+            sp_instr_error *i= new sp_instr_error(ip, lex->spcont,
+                                                  ER_SP_CASE_NOT_FOUND);
+            sp->add_instr(i);
+          }
+        | ELSE sp_proc_stmts1
+        ;
 
 sp_labeled_control:
 	  label_ident ':'
@@ -4374,8 +4521,8 @@ simple_expr:
             if (!$$)
               YYABORT;
 	  }
-	| CASE_SYM opt_expr WHEN_SYM when_list opt_else END
-	  { $$= new Item_func_case(* $4, $2, $5 ); }
+	| CASE_SYM opt_expr when_list opt_else END
+	  { $$= new Item_func_case(* $3, $2, $4 ); }
 	| CONVERT_SYM '(' expr ',' cast_type ')'
 	  {
 	    $$= create_func_cast($3, $5,
@@ -5162,23 +5309,19 @@ opt_else:
 	| ELSE expr    { $$= $2; };
 
 when_list:
-        { Select->when_list.push_front(new List<Item>); }
-	when_list2
-	{ $$= Select->when_list.pop(); };
-
-when_list2:
-	expr THEN_SYM expr
-	  {
-	    SELECT_LEX *sel=Select;
-	    sel->when_list.head()->push_back($1);
-	    sel->when_list.head()->push_back($3);
-	}
-	| when_list2 WHEN_SYM expr THEN_SYM expr
-	  {
-	    SELECT_LEX *sel=Select;
-	    sel->when_list.head()->push_back($3);
-	    sel->when_list.head()->push_back($5);
-	  };
+          WHEN_SYM expr THEN_SYM expr
+          {
+            $$= new List<Item>;
+            $$->push_back($2);
+            $$->push_back($4);
+          }
+        | when_list WHEN_SYM expr THEN_SYM expr
+          {
+            $1->push_back($3);
+            $1->push_back($5);
+            $$= $1;
+          }
+        ;
 
 /* Warning - may return NULL in case of incomplete SELECT */
 table_ref: