A fix for Bug#11918 "SP does not accept variables in LIMIT clause"

Allow stored procedure variables in LIMIT clause.
Only allow variables of INTEGER types. 
Handle negative values by means of an implicit cast to UNSIGNED 
(similarly to prepared statement placeholders).
Add tests.
Make sure replication works by not doing NAME_CONST substitution
for variables in LIMIT clause.
Add replication tests.

mysql-test/r/sp.result:
  Update results (Bug#11918).
mysql-test/suite/rpl/r/rpl_sp.result:
  Update results (Bug#11918).
mysql-test/suite/rpl/t/rpl_sp.test:
  Add a test case for Bug#11918.
mysql-test/t/sp.test:
  Add a test case for Bug#11918.
sql/item.cc:
  Mark sp variables in LIMIT clause (a hack for replication).
sql/item.h:
  Mark sp variables in LIMIT clause (a hack for replication).
sql/share/errmsg-utf8.txt:
  Add a new error message (a type mismatch for LIMIT
  clause parameter).
sql/sp_head.cc:
  Binlog rewrite sp variables in LIMIT clause without NAME_CONST
  substitution.
sql/sql_string.cc:
  Implement append_ulonglong method.
sql/sql_string.h:
  Declare append_ulonglong().
sql/sql_yacc.yy:
  Support stored procedure variables in LIMIT.

11 files changed, 357 insertions, 3 deletions

diff --git a/mysql-test/r/sp.result b/mysql-test/r/sp.result
index 5a746b330bc..7d24e6db920 100644
--- a/mysql-test/r/sp.result
+++ b/mysql-test/r/sp.result
@@ -7192,3 +7192,106 @@ f1
 7
 DROP TEMPORARY TABLE t1;
 DROP PROCEDURE p1;
+#
+# Bug #11918 Can't use a declared variable in LIMIT clause
+#
+drop table if exists t1;
+drop procedure if exists p1;
+create table t1 (c1 int);
+insert into t1 (c1) values (1), (2), (3), (4), (5);
+create procedure p1()
+begin
+declare a integer;
+declare b integer;
+select * from t1 limit a, b;
+end|
+# How do we handle NULL limit values?
+call p1();
+c1
+drop table t1;
+create table t1 (a int);
+insert into t1 (a) values (1), (2), (3), (4), (5);
+#
+# Do we correctly resolve identifiers in LIMIT?
+# Since DROP and CREATE did not invalidate
+# the SP cache, we can't test until
+# we drop and re-create the procedure.
+#
+call p1();
+ERROR 42S22: Unknown column 'test.t1.c1' in 'field list'
+# 
+# Drop and recreate the procedure, then repeat
+#
+drop procedure p1;
+create procedure p1()
+begin
+declare a integer;
+declare b integer;
+select * from t1 limit a, b;
+end|
+# Stored procedure variables are resolved correctly in the LIMIT
+call p1();
+a
+drop table t1;
+create table t1 (c1 int);
+insert into t1 (c1) values (1), (2), (3), (4), (5);
+drop procedure p1;
+# Try to create a procedure that
+# refers to non-existing variables.
+create procedure p1(p1 integer, p2 integer)
+select * from t1 limit a, b;
+ERROR 42000: Undeclared variable: a
+#
+# Try to use data types not allowed in LIMIT 
+#
+create procedure p1(p1 date, p2 date) select * from t1 limit p1, p2;
+ERROR HY000: A variable of a non-integer type in LIMIT clause
+create procedure p1(p1 integer, p2 float) select * from t1 limit p1, p2;
+ERROR HY000: A variable of a non-integer type in LIMIT clause
+create procedure p1(p1 integer, p2 char(1)) select * from t1 limit p1, p2;
+ERROR HY000: A variable of a non-integer type in LIMIT clause
+create procedure p1(p1 varchar(5), p2 char(1)) select * from t1 limit p1, p2;
+ERROR HY000: A variable of a non-integer type in LIMIT clause
+create procedure p1(p1 decimal, p2 decimal) select * from t1 limit p1, p2;
+ERROR HY000: A variable of a non-integer type in LIMIT clause
+create procedure p1(p1 double, p2 double) select * from t1 limit p1, p2;
+ERROR HY000: A variable of a non-integer type in LIMIT clause
+#
+# Finally, test the valid case.
+#
+create procedure p1(p1 integer, p2 integer)
+select * from t1 limit p1, p2;
+call p1(NULL, NULL);
+c1
+call p1(0, 0);
+c1
+call p1(0, -1);
+c1
+1
+2
+3
+4
+5
+call p1(-1, 0);
+c1
+call p1(-1, -1);
+c1
+call p1(0, 1);
+c1
+1
+call p1(1, 0);
+c1
+call p1(1, 5);
+c1
+2
+3
+4
+5
+call p1(3, 2);
+c1
+4
+5
+# Cleanup
+drop table t1;
+drop procedure p1;
+# End of 5.5 test
diff --git a/mysql-test/suite/rpl/r/rpl_sp.result b/mysql-test/suite/rpl/r/rpl_sp.result
index 7f810c5c128..e054ad09e10 100644
--- a/mysql-test/suite/rpl/r/rpl_sp.result
+++ b/mysql-test/suite/rpl/r/rpl_sp.result
@@ -1230,4 +1230,49 @@ a
 drop table t1, t2;
 drop function f1;
 ERROR 42000: FUNCTION test.f1 does not exist
+#
+# Bug #11918 Can't use a declared variable in LIMIT clause
+#
+drop table if exists t1;
+drop procedure if exists p1;
+stop slave;
+drop table if exists t1,t2,t3,t4,t5,t6,t7,t8,t9;
+reset master;
+reset slave;
+drop table if exists t1,t2,t3,t4,t5,t6,t7,t8,t9;
+start slave;
+create table t1 (c1 int);
+insert into t1 (c1) values
+(1), (2), (3), (4), (5), (6), (7), (8), (9), (10);
+call mtr.add_suppression("Unsafe statement binlogged in statement format since BINLOG_FORMAT = STATEMENT");
+create procedure p1(p1 integer)
+delete from t1 limit p1;
+set @save_binlog_format=@@session.binlog_format;
+set @@session.binlog_format=STATEMENT;
+call p1(NULL);
+call p1(0);
+call p1(1);
+call p1(2);
+call p1(3);
+select * from t1;
+c1
+7
+8
+9
+10
+select * from t1;
+c1
+7
+8
+9
+10
+call p1(-1);
+select * from t1;
+c1
+select * from t1;
+c1
+# Cleanup
+set @@session.binlog_format=@save_binlog_format;
+drop table t1;
+drop procedure p1;
 # End of 5.5 tests.
diff --git a/mysql-test/suite/rpl/t/rpl_sp.test b/mysql-test/suite/rpl/t/rpl_sp.test
index 362bfa17ef3..342ae258a3f 100644
--- a/mysql-test/suite/rpl/t/rpl_sp.test
+++ b/mysql-test/suite/rpl/t/rpl_sp.test
@@ -679,6 +679,56 @@ drop table t1, t2;
 --error ER_SP_DOES_NOT_EXIST
 drop function f1;
 
+
+--echo #
+--echo # Bug #11918 Can't use a declared variable in LIMIT clause
+--echo #
+--disable_warnings
+drop table if exists t1;
+drop procedure if exists p1;
+--enable_warnings
+connection master;
+-- source include/master-slave-reset.inc
+connection default;
+create table t1 (c1 int);
+insert into t1 (c1) values
+(1), (2), (3), (4), (5), (6), (7), (8), (9), (10);
+
+call mtr.add_suppression("Unsafe statement binlogged in statement format since BINLOG_FORMAT = STATEMENT");
+
+create procedure p1(p1 integer)
+  delete from t1 limit p1; 
+
+set @save_binlog_format=@@session.binlog_format;
+set @@session.binlog_format=STATEMENT;
+
+--disable_warnings
+call p1(NULL);
+call p1(0);
+call p1(1);
+call p1(2);
+call p1(3);
+--enable_warnings
+
+select * from t1;
+sync_slave_with_master;
+connection slave;
+select * from t1;
+connection default;
+--disable_warnings
+call p1(-1);
+--enable_warnings
+select * from t1;
+sync_slave_with_master;
+connection slave;
+select * from t1;
+connection default;
+
+--echo # Cleanup
+set @@session.binlog_format=@save_binlog_format;
+drop table t1;
+drop procedure p1;
+
 --echo # End of 5.5 tests.
 
 
diff --git a/mysql-test/t/sp.test b/mysql-test/t/sp.test
index 310803531d9..da949016a03 100644
--- a/mysql-test/t/sp.test
+++ b/mysql-test/t/sp.test
@@ -8599,4 +8599,95 @@ SELECT * FROM t1;
 DROP TEMPORARY TABLE t1;
 DROP PROCEDURE p1;
 
+--echo #
+--echo # Bug #11918 Can't use a declared variable in LIMIT clause
+--echo #
+--disable_warnings
+drop table if exists t1;
+drop procedure if exists p1;
+--enable_warnings
+create table t1 (c1 int);
+insert into t1 (c1) values (1), (2), (3), (4), (5);
+
+delimiter |;
+create procedure p1()
+begin
+  declare a integer;
+  declare b integer;
+  select * from t1 limit a, b;
+end|
+delimiter ;|
+--echo # How do we handle NULL limit values?
+call p1();
+drop table t1;
+create table t1 (a int);
+insert into t1 (a) values (1), (2), (3), (4), (5);
+--echo #
+--echo # Do we correctly resolve identifiers in LIMIT?
+--echo # Since DROP and CREATE did not invalidate
+--echo # the SP cache, we can't test until
+--echo # we drop and re-create the procedure.
+--echo #
+--error ER_BAD_FIELD_ERROR
+call p1();
+--echo # 
+--echo # Drop and recreate the procedure, then repeat
+--echo #
+drop procedure p1;
+delimiter |;
+create procedure p1()
+begin
+  declare a integer;
+  declare b integer;
+  select * from t1 limit a, b;
+end|
+delimiter ;|
+--echo # Stored procedure variables are resolved correctly in the LIMIT
+call p1();
+drop table t1;
+create table t1 (c1 int);
+insert into t1 (c1) values (1), (2), (3), (4), (5);
+drop procedure p1;
+--echo # Try to create a procedure that
+--echo # refers to non-existing variables.
+--error ER_SP_UNDECLARED_VAR
+create procedure p1(p1 integer, p2 integer)
+  select * from t1 limit a, b;
+--echo #
+--echo # Try to use data types not allowed in LIMIT 
+--echo #
+--error ER_WRONG_SPVAR_TYPE_IN_LIMIT
+create procedure p1(p1 date, p2 date) select * from t1 limit p1, p2;
+--error ER_WRONG_SPVAR_TYPE_IN_LIMIT
+create procedure p1(p1 integer, p2 float) select * from t1 limit p1, p2;
+--error ER_WRONG_SPVAR_TYPE_IN_LIMIT
+create procedure p1(p1 integer, p2 char(1)) select * from t1 limit p1, p2;
+--error ER_WRONG_SPVAR_TYPE_IN_LIMIT
+create procedure p1(p1 varchar(5), p2 char(1)) select * from t1 limit p1, p2;
+--error ER_WRONG_SPVAR_TYPE_IN_LIMIT
+create procedure p1(p1 decimal, p2 decimal) select * from t1 limit p1, p2;
+--error ER_WRONG_SPVAR_TYPE_IN_LIMIT
+create procedure p1(p1 double, p2 double) select * from t1 limit p1, p2;
+
+--echo #
+--echo # Finally, test the valid case.
+--echo #
+create procedure p1(p1 integer, p2 integer)
+  select * from t1 limit p1, p2;
+
+call p1(NULL, NULL);
+call p1(0, 0);
+call p1(0, -1);
+call p1(-1, 0);
+call p1(-1, -1);
+call p1(0, 1);
+call p1(1, 0);
+call p1(1, 5);
+call p1(3, 2);
+
+
+--echo # Cleanup
+drop table t1;
+drop procedure p1;
 
+--echo # End of 5.5 test
diff --git a/sql/item.cc b/sql/item.cc
index 209d5aa0197..4199f8a409a 100644
--- a/sql/item.cc
+++ b/sql/item.cc
@@ -1186,7 +1186,9 @@ Item_splocal::Item_splocal(const LEX_STRING &sp_var_name,
                            enum_field_types sp_var_type,
                            uint pos_in_q, uint len_in_q)
   :Item_sp_variable(sp_var_name.str, sp_var_name.length),
-   m_var_idx(sp_var_idx), pos_in_query(pos_in_q), len_in_query(len_in_q)
+   m_var_idx(sp_var_idx),
+   limit_clause_param(FALSE),
+   pos_in_query(pos_in_q), len_in_query(len_in_q)
 {
   maybe_null= TRUE;
 
diff --git a/sql/item.h b/sql/item.h
index 4241074c659..11b4199da2c 100644
--- a/sql/item.h
+++ b/sql/item.h
@@ -1302,6 +1302,13 @@ class Item_splocal :public Item_sp_variable,
   Item_result m_result_type;
   enum_field_types m_field_type;
 public:
+  /*
+    Is this variable a parameter in LIMIT clause. 
+    Used only during NAME_CONST substitution, to not append
+    NAME_CONST to the resulting query and thus not break
+    the slave.
+  */
+  bool limit_clause_param;
   /* 
     Position of this reference to SP variable in the statement (the
     statement itself is in sp_instr_stmt::m_query).
diff --git a/sql/share/errmsg-utf8.txt b/sql/share/errmsg-utf8.txt
index 517782cb0b4..3acf990e00d 100644
--- a/sql/share/errmsg-utf8.txt
+++ b/sql/share/errmsg-utf8.txt
@@ -6327,3 +6327,6 @@ ER_LOCK_ABORTED
 
 ER_DATA_OUT_OF_RANGE 22003 
   eng "%s value is out of range in '%s'"
+
+ER_WRONG_SPVAR_TYPE_IN_LIMIT
+  eng "A variable of a non-integer type in LIMIT clause"
diff --git a/sql/sp_head.cc b/sql/sp_head.cc
index 62d53888149..c91ba2a68b4 100644
--- a/sql/sp_head.cc
+++ b/sql/sp_head.cc
@@ -978,11 +978,20 @@ subst_spvars(THD *thd, sp_instr *instr, LEX_STRING *query_str)
     res|= qbuf.append(cur + prev_pos, (*splocal)->pos_in_query - prev_pos);
     prev_pos= (*splocal)->pos_in_query + (*splocal)->len_in_query;
     
+    res|= (*splocal)->fix_fields(thd, (Item **) splocal);
+    if (res) 
+      break;
+
+    if ((*splocal)->limit_clause_param)
+    {
+      res|= qbuf.append_ulonglong((*splocal)->val_uint());
+      continue;
+    }
+
     /* append the spvar substitute */
     res|= qbuf.append(STRING_WITH_LEN(" NAME_CONST('"));
     res|= qbuf.append((*splocal)->m_name.str, (*splocal)->m_name.length);
     res|= qbuf.append(STRING_WITH_LEN("',"));
-    res|= (*splocal)->fix_fields(thd, (Item **) splocal);
 
     if (res)
       break;
diff --git a/sql/sql_string.cc b/sql/sql_string.cc
index 75e8ca30cf0..9fbc06b7529 100644
--- a/sql/sql_string.cc
+++ b/sql/sql_string.cc
@@ -405,6 +405,16 @@ bool String::append(const char *s)
 }
 
 
+
+bool String::append_ulonglong(ulonglong val)
+{
+  if (realloc(str_length+MAX_BIGINT_WIDTH+2))
+    return TRUE;
+  char *end= (char*) longlong10_to_str(val, (char*) Ptr + str_length, 10);
+  str_length= end - Ptr;
+  return FALSE;
+}
+
 /*
   Append a string in the given charset to the string
   with character set recoding
diff --git a/sql/sql_string.h b/sql/sql_string.h
index f67d6de9a0f..debfb7aa9c6 100644
--- a/sql/sql_string.h
+++ b/sql/sql_string.h
@@ -267,6 +267,7 @@ public:
   bool append(const char *s);
   bool append(const char *s,uint32 arg_length);
   bool append(const char *s,uint32 arg_length, CHARSET_INFO *cs);
+  bool append_ulonglong(ulonglong val);
   bool append(IO_CACHE* file, uint32 arg_length);
   bool append_with_prefill(const char *s, uint32 arg_length, 
 			   uint32 full_length, char fill_char);
diff --git a/sql/sql_yacc.yy b/sql/sql_yacc.yy
index 207e72404ef..a0d64e6a378 100644
--- a/sql/sql_yacc.yy
+++ b/sql/sql_yacc.yy
@@ -9830,7 +9830,40 @@ limit_options:
         ;
 
 limit_option:
-        param_marker
+        ident
+        {
+          Item_splocal *splocal;
+          THD *thd= YYTHD;
+          LEX *lex= thd->lex;
+          Lex_input_stream *lip= & thd->m_parser_state->m_lip;
+          sp_variable_t *spv;
+          sp_pcontext *spc = lex->spcont;
+          if (spc && (spv = spc->find_variable(&$1)))
+          {
+            splocal= new (thd->mem_root)
+              Item_splocal($1, spv->offset, spv->type,
+                  lip->get_tok_start() - lex->sphead->m_tmp_query,
+                  lip->get_ptr() - lip->get_tok_start());
+            if (splocal == NULL)
+              MYSQL_YYABORT;
+#ifndef DBUG_OFF
+            splocal->m_sp= lex->sphead;
+#endif
+            lex->safe_to_cache_query=0;
+          }
+          else
+          {
+            my_error(ER_SP_UNDECLARED_VAR, MYF(0), $1.str);
+            MYSQL_YYABORT;
+          }
+          if (splocal->type() != Item::INT_ITEM)
+          {
+            my_error(ER_WRONG_SPVAR_TYPE_IN_LIMIT, MYF(0));
+            MYSQL_YYABORT;
+          }
+          splocal->limit_clause_param= TRUE;
+          $$= splocal;
+        } | param_marker
         {
           ((Item_param *) $1)->limit_clause_param= TRUE;
         }