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

@@ -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

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.

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.
 
 

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

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;
 

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).

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"

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;

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

sql/sql_string.h

@@ -267,6 +267,7 @@ class String
   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);

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;
         }