14 files changed, 536 insertions, 198 deletions

diff --git a/sql/event_data_objects.cc b/sql/event_data_objects.cc
index eef45f93b7a..54b043bd916 100644
--- a/sql/event_data_objects.cc
+++ b/sql/event_data_objects.cc
@@ -394,7 +394,14 @@ Event_parse_data::init_starts(THD *thd)
   if ((not_used= item_starts->get_date(&ltime, TIME_NO_ZERO_DATE)))
     goto wrong_value;
 
-  /* Let's check whether time is in the past */
+  /*
+    Let's check whether time is in the past.
+    Note: This call is not post year 2038 safe. That's because
+          thd->query_start() is of time_t, while gmt_sec_to_TIME()
+          wants my_time_t. In the case time_t is larger than my_time_t
+          an overflow might happen and events subsystem will not work as
+          expected.
+  */
   thd->variables.time_zone->gmt_sec_to_TIME(&time_tmp,
                                             (my_time_t) thd->query_start());
 
@@ -406,12 +413,12 @@ Event_parse_data::init_starts(THD *thd)
     goto wrong_value;
 
   /*
-    This may result in a 1970-01-01 date if ltime is > 2037-xx-xx.
-    CONVERT_TZ has similar problem.
-    mysql_priv.h currently lists 
+    Again, after 2038 this code won't work. As
+    mysql_priv.h currently lists
       #define TIMESTAMP_MAX_YEAR 2038 (see TIME_to_timestamp())
   */
-  my_tz_UTC->gmt_sec_to_TIME(&ltime,t=TIME_to_timestamp(thd, &ltime, &not_used));
+  my_tz_UTC->gmt_sec_to_TIME(&ltime,t=TIME_to_timestamp(thd, &ltime,
+                                                        &not_used));
   if (!t)
     goto wrong_value;
 
@@ -464,13 +471,13 @@ Event_parse_data::init_ends(THD *thd)
     goto error_bad_params;
 
   /*
-    This may result in a 1970-01-01 date if ltime is > 2037-xx-xx.
-    CONVERT_TZ has similar problem.
-    mysql_priv.h currently lists 
+    Again, after 2038 this code won't work. As
+    mysql_priv.h currently lists
       #define TIMESTAMP_MAX_YEAR 2038 (see TIME_to_timestamp())
   */
   DBUG_PRINT("info", ("get the UTC time"));
-  my_tz_UTC->gmt_sec_to_TIME(&ltime,t=TIME_to_timestamp(thd, &ltime, &not_used));
+  my_tz_UTC->gmt_sec_to_TIME(&ltime,t=TIME_to_timestamp(thd, &ltime,
+                                                        &not_used));
   if (!t)
     goto error_bad_params;
 
diff --git a/sql/event_queue.cc b/sql/event_queue.cc
index 6ff5fe55cd6..9a740114193 100644
--- a/sql/event_queue.cc
+++ b/sql/event_queue.cc
@@ -159,7 +159,6 @@ Event_queue::init_queue(THD *thd, Event_db_repository *db_repo)
   {
     sql_print_error("SCHEDULER: sizeof(my_time_t) != sizeof(time_t) ."
                     "The scheduler may not work correctly. Stopping");
-    DBUG_ASSERT(0);
     goto err;
   }
 
diff --git a/sql/handler.cc b/sql/handler.cc
index e0955132998..3d47a6a2eaf 100644
--- a/sql/handler.cc
+++ b/sql/handler.cc
@@ -691,6 +691,19 @@ int ha_commit_trans(THD *thd, bool all)
       ha_rollback_trans(thd, all);
       DBUG_RETURN(1);
     }
+
+    if (   is_real_trans
+        && opt_readonly
+        && ! (thd->security_ctx->master_access & SUPER_ACL)
+        && ! thd->slave_thread
+       )
+    {
+      my_error(ER_OPTION_PREVENTS_STATEMENT, MYF(0), "--read-only");
+      ha_rollback_trans(thd, all);
+      error= 1;
+      goto end;
+    }
+
     DBUG_EXECUTE_IF("crash_commit_before", abort(););
 
     /* Close all cursors that can not survive COMMIT */
diff --git a/sql/lock.cc b/sql/lock.cc
index 2bcee1e4baa..533307c6b85 100644
--- a/sql/lock.cc
+++ b/sql/lock.cc
@@ -150,6 +150,23 @@ MYSQL_LOCK *mysql_lock_tables(THD *thd, TABLE **tables, uint count,
       }
     }
 
+    if (   write_lock_used
+        && opt_readonly
+        && ! (thd->security_ctx->master_access & SUPER_ACL)
+        && ! thd->slave_thread
+       )
+    {
+      /*
+	Someone has issued SET GLOBAL READ_ONLY=1 and we want a write lock.
+        We do not wait for READ_ONLY=0, and fail.
+      */
+      reset_lock_data(sql_lock);
+      my_free((gptr) sql_lock, MYF(0));
+      sql_lock=0;
+      my_error(ER_OPTION_PREVENTS_STATEMENT, MYF(0), "--read-only");
+      break;
+    }
+
     thd->proc_info="System lock";
     DBUG_PRINT("info", ("thd->proc_info %s", thd->proc_info));
     if (lock_external(thd, tables, count))
diff --git a/sql/set_var.cc b/sql/set_var.cc
index b998548a6ab..4d2195b2306 100644
--- a/sql/set_var.cc
+++ b/sql/set_var.cc
@@ -369,7 +369,7 @@ sys_var_thd_ulong       sys_preload_buff_size("preload_buffer_size",
                                               &SV::preload_buff_size);
 sys_var_thd_ulong	sys_read_buff_size("read_buffer_size",
 					   &SV::read_buff_size);
-sys_var_bool_ptr	sys_readonly("read_only", &opt_readonly);
+sys_var_opt_readonly    sys_readonly("read_only", &opt_readonly);
 sys_var_thd_ulong	sys_read_rnd_buff_size("read_rnd_buffer_size",
 					       &SV::read_rnd_buff_size);
 sys_var_thd_ulong	sys_div_precincrement("div_precision_increment",
@@ -3857,6 +3857,70 @@ bool sys_var_trust_routine_creators::update(THD *thd, set_var *var)
   return sys_var_bool_ptr::update(thd, var);
 }
 
+bool sys_var_opt_readonly::update(THD *thd, set_var *var)
+{
+  bool result;
+
+  DBUG_ENTER("sys_var_opt_readonly::update");
+
+  /* Prevent self dead-lock */
+  if (thd->locked_tables || thd->active_transaction())
+  {
+    my_error(ER_LOCK_OR_ACTIVE_TRANSACTION, MYF(0));
+    DBUG_RETURN(true);
+  }
+
+  if (thd->global_read_lock)
+  {
+    /*
+      This connection already holds the global read lock.
+      This can be the case with:
+      - FLUSH TABLES WITH READ LOCK
+      - SET GLOBAL READ_ONLY = 1
+    */
+    result= sys_var_bool_ptr::update(thd, var);
+    DBUG_RETURN(result);
+  }
+
+  /*
+    Perform a 'FLUSH TABLES WITH READ LOCK'.
+    This is a 3 step process:
+    - [1] lock_global_read_lock()
+    - [2] close_cached_tables()
+    - [3] make_global_read_lock_block_commit()
+    [1] prevents new connections from obtaining tables locked for write.
+    [2] waits until all existing connections close their tables.
+    [3] prevents transactions from being committed.
+  */
+
+  if (lock_global_read_lock(thd))
+    DBUG_RETURN(true);
+
+  /*
+    This call will be blocked by any connection holding a READ or WRITE lock.
+    Ideally, we want to wait only for pending WRITE locks, but since:
+    con 1> LOCK TABLE T FOR READ;
+    con 2> LOCK TABLE T FOR WRITE; (blocked by con 1)
+    con 3> SET GLOBAL READ ONLY=1; (blocked by con 2)
+    can cause to wait on a read lock, it's required for the client application
+    to unlock everything, and acceptable for the server to wait on all locks.
+  */
+  if (close_cached_tables(thd, true, NULL, false))
+    goto end_with_read_lock;
+
+  if (result= make_global_read_lock_block_commit(thd))
+    goto end_with_read_lock;
+
+  /* Change the opt_readonly system variable, safe because the lock is held */
+  result= sys_var_bool_ptr::update(thd, var);
+
+end_with_read_lock:
+  /* Release the lock */
+  unlock_global_read_lock(thd);
+  DBUG_RETURN(result);
+}
+
+
 /* even session variable here requires SUPER, because of -#o,file */
 bool sys_var_thd_dbug::check(THD *thd, set_var *var)
 {
diff --git a/sql/set_var.h b/sql/set_var.h
index 3b0aa6cde9c..f957ec931d6 100644
--- a/sql/set_var.h
+++ b/sql/set_var.h
@@ -904,6 +904,20 @@ public:
 };
 
 
+/**
+  Handler for setting the system variable --read-only.
+*/
+
+class sys_var_opt_readonly :public sys_var_bool_ptr
+{
+public:
+  sys_var_opt_readonly(const char *name_arg, my_bool *value_arg) :
+    sys_var_bool_ptr(name_arg, value_arg) {};
+  ~sys_var_opt_readonly() {};
+  bool update(THD *thd, set_var *var);
+};
+
+
 class sys_var_thd_lc_time_names :public sys_var_thd
 {
 public:
diff --git a/sql/sp_head.cc b/sql/sp_head.cc
index 714202b0864..2b65e2b345f 100644
--- a/sql/sp_head.cc
+++ b/sql/sp_head.cc
@@ -628,27 +628,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
@@ -2229,7 +2208,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
@@ -2242,7 +2221,7 @@ void sp_head::optimize()
   sp_instr *i;
   uint src, dst;
 
-  opt_mark(0);
+  opt_mark();
 
   bp.empty();
   src= dst= 0;
@@ -2276,13 +2255,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);
+    }
+  }
 }
 
 
@@ -2684,7 +2700,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? */
@@ -2778,7 +2794,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;
 
@@ -2788,13 +2804,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;
 }
 
@@ -2915,7 +2931,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;
 
@@ -2925,7 +2941,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;
 }
 
@@ -2990,15 +3006,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;
 }
 
 
@@ -3341,7 +3355,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;
 
@@ -3351,7 +3365,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 2c554d50bd8..0085608ae40 100644
--- a/sql/sp_head.h
+++ b/sql/sp_head.h
@@ -303,8 +303,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);
 
@@ -413,6 +424,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.
@@ -480,10 +497,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;
@@ -755,7 +772,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);
 
@@ -805,7 +822,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)
@@ -851,7 +868,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;
@@ -888,7 +905,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)
@@ -953,7 +970,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:
 
@@ -1123,7 +1140,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;
@@ -1156,7 +1173,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_class.cc b/sql/sql_class.cc
index af66fd2d3de..9e3c2442f57 100644
--- a/sql/sql_class.cc
+++ b/sql/sql_class.cc
@@ -984,6 +984,13 @@ void select_result::cleanup()
   /* do nothing */
 }
 
+bool select_result::check_simple_select() const
+{
+  my_error(ER_SP_BAD_CURSOR_QUERY, MYF(0));
+  return TRUE;
+}
+
+
 static String default_line_term("\n",default_charset_info);
 static String default_escaped("\\",default_charset_info);
 static String default_field_term("\t",default_charset_info);
@@ -1624,6 +1631,13 @@ int select_dumpvar::prepare(List<Item> &list, SELECT_LEX_UNIT *u)
 }
 
 
+bool select_dumpvar::check_simple_select() const
+{
+  my_error(ER_SP_BAD_CURSOR_SELECT, MYF(0));
+  return TRUE;
+}
+
+
 void select_dumpvar::cleanup()
 {
   row_count= 0;
diff --git a/sql/sql_class.h b/sql/sql_class.h
index 94535d6f57b..8c86a66392a 100644
--- a/sql/sql_class.h
+++ b/sql/sql_class.h
@@ -1702,7 +1702,14 @@ public:
   virtual bool initialize_tables (JOIN *join=0) { return 0; }
   virtual void send_error(uint errcode,const char *err);
   virtual bool send_eof()=0;
-  virtual bool simple_select() { return 0; }
+  /**
+    Check if this query returns a result set and therefore is allowed in
+    cursors and set an error message if it is not the case.
+
+    @retval FALSE     success
+    @retval TRUE      error, an error message is set
+  */
+  virtual bool check_simple_select() const;
   virtual void abort() {}
   /*
     Cleanup instance of this class for next execution of a prepared
@@ -1740,7 +1747,7 @@ public:
   bool send_fields(List<Item> &list, uint flags);
   bool send_data(List<Item> &items);
   bool send_eof();
-  bool simple_select() { return 1; }
+  virtual bool check_simple_select() const { return FALSE; }
   void abort();
 };
 
@@ -2186,6 +2193,7 @@ public:
   int prepare(List<Item> &list, SELECT_LEX_UNIT *u);
   bool send_data(List<Item> &items);
   bool send_eof();
+  virtual bool check_simple_select() const;
   void cleanup();
 };
 
diff --git a/sql/sql_lex.cc b/sql/sql_lex.cc
index 18d30494701..1c399503498 100644
--- a/sql/sql_lex.cc
+++ b/sql/sql_lex.cc
@@ -1193,7 +1193,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 12e41d12899..fd76ec7ac51 100644
--- a/sql/sql_lex.h
+++ b/sql/sql_lex.h
@@ -552,7 +552,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_prepare.cc b/sql/sql_prepare.cc
index 2a162216add..d95c434390c 100644
--- a/sql/sql_prepare.cc
+++ b/sql/sql_prepare.cc
@@ -2945,10 +2945,9 @@ bool Prepared_statement::execute(String *expanded_query, bool open_cursor)
     in INSERT ... SELECT and similar commands.
   */
 
-  if (open_cursor && lex->result && !lex->result->simple_select())
+  if (open_cursor && lex->result && lex->result->check_simple_select())
   {
     DBUG_PRINT("info",("Cursor asked for not SELECT stmt"));
-    my_error(ER_SP_BAD_CURSOR_QUERY, MYF(0));
     return TRUE;
   }
 
diff --git a/sql/sql_yacc.yy b/sql/sql_yacc.yy
index 4d8eda06faf..6938a7b10d1 100644
--- a/sql/sql_yacc.yy
+++ b/sql/sql_yacc.yy
@@ -107,6 +107,187 @@ static bool is_native_function(THD *thd, const LEX_STRING *name)
   return false;
 }
 
+
+/**
+  Helper action for a case statement (entering the CASE).
+  This helper is used for both 'simple' and 'searched' cases.
+  This helper, with the other case_stmt_action_..., is executed when
+  the following SQL code is parsed:
+<pre>
+CREATE PROCEDURE proc_19194_simple(i int)
+BEGIN
+  DECLARE str CHAR(10);
+
+  CASE i
+    WHEN 1 THEN SET str="1";
+    WHEN 2 THEN SET str="2";
+    WHEN 3 THEN SET str="3";
+    ELSE SET str="unknown";
+  END CASE;
+
+  SELECT str;
+END
+</pre>
+  The actions are used to generate the following code:
+<pre>
+SHOW PROCEDURE CODE proc_19194_simple;
+Pos     Instruction
+0       set str@1 NULL
+1       set_case_expr (12) 0 i@0
+2       jump_if_not 5(12) (case_expr@0 = 1)
+3       set str@1 _latin1'1'
+4       jump 12
+5       jump_if_not 8(12) (case_expr@0 = 2)
+6       set str@1 _latin1'2'
+7       jump 12
+8       jump_if_not 11(12) (case_expr@0 = 3)
+9       set str@1 _latin1'3'
+10      jump 12
+11      set str@1 _latin1'unknown'
+12      stmt 0 "SELECT str"
+</pre>
+
+  @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"
+    (Instruction 12 in the example)
+  */
+
+  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"
+    (jump_if_not from instruction 2 to 5, 5 to 8 ... in the example)
+  */
+
+  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"
+    (jump_if_not from instruction 2 to 5, 5 to 8 ... in the example)
+  */
+
+  sp->backpatch(ctx->pop_label());
+
+  /*
+    BACKPATCH: Registering forward jump from
+    "case_stmt_action_then" to "case_stmt_action_end_case"
+    (jump from instruction 4 to 12, 7 to 12 ... in the example)
+  */
+
+  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"
+    (jump from instruction 4 to 12, 7 to 12 ... in the example)
+  */
+  lex->sphead->backpatch(lex->spcont->pop_label());
+
+  if (simple)
+    lex->spcont->pop_case_expr_id();
+
+  lex->sphead->do_cont_backpatch();
+}
+
 %}
 %union {
   int  num;
@@ -886,7 +1067,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
@@ -922,10 +1103,11 @@ END_OF_INPUT
 
 %type <NONE> call sp_proc_stmts sp_proc_stmts1 sp_proc_stmt
 %type <NONE> sp_proc_stmt_statement sp_proc_stmt_return
-%type <NONE> sp_proc_stmt_if sp_proc_stmt_case_simple sp_proc_stmt_case
+%type <NONE> sp_proc_stmt_if
 %type <NONE> sp_labeled_control sp_proc_stmt_unlabeled sp_proc_stmt_leave
 %type <NONE> sp_proc_stmt_iterate
 %type <NONE> sp_proc_stmt_open sp_proc_stmt_fetch sp_proc_stmt_close
+%type <NONE> case_stmt_specification simple_case_stmt searched_case_stmt
 
 %type <num>  sp_decl_idents sp_opt_inout sp_handler_type sp_hcond_list
 %type <spcondtype> sp_cond sp_hcond
@@ -1546,8 +1728,7 @@ ev_sql_stmt_inner:
           sp_proc_stmt_statement
         | sp_proc_stmt_return
         | sp_proc_stmt_if
-        | sp_proc_stmt_case_simple
-        | sp_proc_stmt_case
+        | case_stmt_specification
         | sp_labeled_control
         | sp_proc_stmt_unlabeled
         | sp_proc_stmt_leave
@@ -2311,8 +2492,7 @@ sp_proc_stmt:
 	  sp_proc_stmt_statement
         | sp_proc_stmt_return
 	| sp_proc_stmt_if
-	| sp_proc_stmt_case_simple
-        | sp_proc_stmt_case
+	| case_stmt_specification
 	| sp_labeled_control
 	| sp_proc_stmt_unlabeled
 	| sp_proc_stmt_leave
@@ -2402,49 +2582,6 @@ sp_proc_stmt_return:
 	  }
         ;
 
-sp_proc_stmt_case_simple:
-	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(); }
-        ;
-        
-sp_proc_stmt_case:
-          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();
-	  }
-        ;
-
 sp_proc_stmt_unlabeled:
 	  { /* Unlabeled controls get a secret label. */
 	    LEX *lex= Lex;
@@ -2669,72 +2806,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 ':'
@@ -4922,7 +5101,6 @@ opt_ev_sql_stmt: /* empty*/ { $$= 0;}
         | DO_SYM ev_sql_stmt { $$= 1; }
         ;
 
-
 ident_or_empty:
 	/* empty */  { $$.str= 0; $$.length= 0; }
 	| ident      { $$= $1; };
@@ -6130,8 +6308,8 @@ simple_expr:
             if (!$$)
               YYABORT;
 	  }
-	| CASE_SYM opt_expr WHEN_SYM when_list opt_else END
-	  { $$= new (YYTHD->mem_root) Item_func_case(* $4, $2, $5 ); }
+	| CASE_SYM opt_expr when_list opt_else END
+	  { $$= new (YYTHD->mem_root) Item_func_case(* $3, $2, $4 ); }
 	| CONVERT_SYM '(' expr ',' cast_type ')'
 	  {
 	    $$= create_func_cast(YYTHD, $3, $5,
@@ -6349,7 +6527,7 @@ function_call_nonkeyword:
         ;
 
 /*
-  Functions calls using a non reserved keywork, and using a regular syntax.
+  Functions calls using a non reserved keyword, and using a regular syntax.
   Because the non reserved keyword is used in another part of the grammar,
   a dedicated rule is needed here.
 */
@@ -6564,7 +6742,7 @@ function_call_generic:
             parser and the implementation in item_create.cc clean,
             since this will change with WL#2128 (SQL PATH):
             - INFORMATION_SCHEMA.version() is the SQL 99 syntax for the native
-            funtion version(),
+            function version(),
             - MySQL.version() is the SQL 2003 syntax for the native function
             version() (a vendor can specify any schema).
           */
@@ -6666,7 +6844,7 @@ sum_expr:
 	  { $$=new Item_sum_min($3); }
 /*
    According to ANSI SQL, DISTINCT is allowed and has
-   no sence inside MIN and MAX grouping functions; so MIN|MAX(DISTINCT ...)
+   no sense inside MIN and MAX grouping functions; so MIN|MAX(DISTINCT ...)
    is processed like an ordinary MIN | MAX()
  */
 	| MIN_SYM '(' DISTINCT in_sum_expr ')'
@@ -6830,23 +7008,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: