Bug fix for lp:732124 union + limit returns wrong result

mysql-test/r/union.result:
  Added test for lp:732124
mysql-test/t/union.test:
  Added test for lp:732124
sql/sp_rcontext.cc:
  Updated function definition for ::send_data()
sql/sp_rcontext.h:
  Updated function definition for ::send_data()
sql/sql_analyse.cc:
  Test if send_data() returned an error
sql/sql_class.cc:
  Updated function definition for ::send_data()
sql/sql_class.h:
  Changed select_result::send_data(List<Item> &items) to return -1 in case of duplicate row that should not be counted as part of LIMIT
sql/sql_cursor.cc:
  Check if send_data returned error
sql/sql_delete.cc:
  Updated function definition for ::send_data()
sql/sql_insert.cc:
  Updated function definition for ::send_data()
sql/sql_select.cc:
  Don't count rows which send_data() tells you to ignore
sql/sql_union.cc:
  Inform caller that the row should be ignored. This is the real bug fix for lp:732124
sql/sql_update.cc:
  Updated function definition for ::send_data()

mysql-test/r/union.result

@@ -414,7 +414,7 @@ a
 5
 select found_rows();
 found_rows()
-6
+5
 SELECT SQL_CALC_FOUND_ROWS * FROM t1 LIMIT 100 UNION SELECT * FROM t2;
 a
 1
@@ -447,7 +447,7 @@ a
 4
 select found_rows();
 found_rows()
-6
+5
 SELECT SQL_CALC_FOUND_ROWS * FROM t1 limit 2,2 UNION SELECT * FROM t2;
 a
 3
@@ -1208,9 +1208,12 @@ a	b
 select * from ((select * from t1 limit 1) union (select * from t1 limit 1)) a;
 a	b
 1	a
+2	b
 select * from ((select * from t1 limit 1) union (select * from t1 limit 1) union (select * from t1 limit 1)) a;
 a	b
 1	a
+2	b
+3	c
 select * from ((((select * from t1))) union (select * from t1) union (select * from t1)) a;
 a	b
 1	a
@@ -1647,4 +1650,17 @@ b
 1
 2
 DROP TABLE t1,t2;
+create table t1 (a int);
+insert into t1 values (10),(10),(10),(2),(3),(4),(5),(6),(7),(8),(9),(1),(10);
+select a from t1 where false UNION select a from t1 limit 8;
+a
+10
+2
+3
+4
+5
+6
+7
+8
+drop table t1;
 End of 5.1 tests

mysql-test/t/union.test

@@ -1155,5 +1155,13 @@ SELECT * FROM t2 UNION SELECT * FROM t2
 
 DROP TABLE t1,t2;
 
+#
+# lp:732124 union + limit returns wrong result
+#
+create table t1 (a int);
+insert into t1 values (10),(10),(10),(2),(3),(4),(5),(6),(7),(8),(9),(1),(10);
+--sorted_result
+select a from t1 where false UNION select a from t1 limit 8;
+drop table t1;
 
 --echo End of 5.1 tests

sql/sp_rcontext.cc

@@ -651,7 +651,7 @@ int Select_fetch_into_spvars::prepare(List<Item> &fields, SELECT_LEX_UNIT *u)
 }
 
 
-bool Select_fetch_into_spvars::send_data(List<Item> &items)
+int Select_fetch_into_spvars::send_data(List<Item> &items)
 {
   List_iterator_fast<struct sp_variable> spvar_iter(*spvar_list);
   List_iterator_fast<Item> item_iter(items);
@@ -668,7 +668,7 @@ bool Select_fetch_into_spvars::send_data(List<Item> &items)
   for (; spvar= spvar_iter++, item= item_iter++; )
   {
     if (thd->spcont->set_variable(thd, spvar->offset, &item))
-      return TRUE;
+      return 1;
   }
-  return FALSE;
+  return 0;
 }

sql/sp_rcontext.h

@@ -254,7 +254,7 @@ class Select_fetch_into_spvars: public select_result_interceptor
   void set_spvar_list(List<struct sp_variable> *vars) { spvar_list= vars; }
 
   virtual bool send_eof() { return FALSE; }
-  virtual bool send_data(List<Item> &items);
+  virtual int send_data(List<Item> &items);
   virtual int prepare(List<Item> &list, SELECT_LEX_UNIT *u);
 };
 

sql/sql_analyse.cc

@@ -753,7 +753,7 @@ int analyse::end_of_records()
 	tmp_str.append(STRING_WITH_LEN(" NOT NULL"));
       output_str_length = tmp_str.length();
       func_items[9]->set(tmp_str.ptr(), tmp_str.length(), tmp_str.charset());
-      if (result->send_data(result_fields))
+      if (result->send_data(result_fields) > 0)
 	return -1;
       continue;
     }
@@ -798,7 +798,7 @@ int analyse::end_of_records()
     if (!(*f)->nulls)
       ans.append(STRING_WITH_LEN(" NOT NULL"));
     func_items[9]->set(ans.ptr(), ans.length(), ans.charset());
-    if (result->send_data(result_fields))
+    if (result->send_data(result_fields) > 0)
       return -1;
   }
   return 0;

sql/sql_class.cc

@@ -1760,7 +1760,7 @@ void select_send::cleanup()
 
 /* Send data to client. Returns 0 if ok */
 
-bool select_send::send_data(List<Item> &items)
+int select_send::send_data(List<Item> &items)
 {
   if (unit->offset_limit_cnt)
   {						// using limit offset,count
@@ -2069,7 +2069,7 @@ select_export::prepare(List<Item> &list, SELECT_LEX_UNIT *u)
                           (int) (uchar) (x) == line_sep_char  || \
                           !(x))
 
-bool select_export::send_data(List<Item> &items)
+int select_export::send_data(List<Item> &items)
 {
 
   DBUG_ENTER("select_export::send_data");
@@ -2327,7 +2327,7 @@ select_dump::prepare(List<Item> &list __attribute__((unused)),
 }
 
 
-bool select_dump::send_data(List<Item> &items)
+int select_dump::send_data(List<Item> &items)
 {
   List_iterator_fast<Item> li(items);
   char buff[MAX_FIELD_WIDTH];
@@ -2372,7 +2372,7 @@ select_subselect::select_subselect(Item_subselect *item_arg)
 }
 
 
-bool select_singlerow_subselect::send_data(List<Item> &items)
+int select_singlerow_subselect::send_data(List<Item> &items)
 {
   DBUG_ENTER("select_singlerow_subselect::send_data");
   Item_singlerow_subselect *it= (Item_singlerow_subselect *)item;
@@ -2403,7 +2403,7 @@ void select_max_min_finder_subselect::cleanup()
 }
 
 
-bool select_max_min_finder_subselect::send_data(List<Item> &items)
+int select_max_min_finder_subselect::send_data(List<Item> &items)
 {
   DBUG_ENTER("select_max_min_finder_subselect::send_data");
   Item_maxmin_subselect *it= (Item_maxmin_subselect *)item;
@@ -2507,7 +2507,7 @@ bool select_max_min_finder_subselect::cmp_str()
      sortcmp(val1, val2, cache->collation.collation) < 0);
 }
 
-bool select_exists_subselect::send_data(List<Item> &items)
+int select_exists_subselect::send_data(List<Item> &items)
 {
   DBUG_ENTER("select_exists_subselect::send_data");
   Item_exists_subselect *it= (Item_exists_subselect *)item;
@@ -2859,7 +2859,7 @@ Statement_map::~Statement_map()
   hash_free(&st_hash);
 }
 
-bool select_dumpvar::send_data(List<Item> &items)
+int select_dumpvar::send_data(List<Item> &items)
 {
   List_iterator_fast<my_var> var_li(var_list);
   List_iterator<Item> it(items);

sql/sql_class.h

@@ -2514,7 +2514,11 @@ class select_result :public Sql_alloc {
   virtual uint field_count(List<Item> &fields) const
   { return fields.elements; }
   virtual bool send_fields(List<Item> &list, uint flags)=0;
-  virtual bool send_data(List<Item> &items)=0;
+  /*
+    send_data returns 0 on ok, 1 on error and -1 if data was ignored, for
+    example for a duplicate row entry written to a temp table.
+  */
+  virtual int send_data(List<Item> &items)=0;
   virtual bool initialize_tables (JOIN *join=0) { return 0; }
   virtual void send_error(uint errcode,const char *err);
   virtual bool send_eof()=0;
@@ -2572,7 +2576,7 @@ class select_send :public select_result {
 public:
   select_send() :is_result_set_started(FALSE) {}
   bool send_fields(List<Item> &list, uint flags);
-  bool send_data(List<Item> &items);
+  int send_data(List<Item> &items);
   bool send_eof();
   virtual bool check_simple_select() const { return FALSE; }
   void abort();
@@ -2643,7 +2647,7 @@ class select_export :public select_to_file {
   }
   ~select_export();
   int prepare(List<Item> &list, SELECT_LEX_UNIT *u);
-  bool send_data(List<Item> &items);
+  int send_data(List<Item> &items);
 };
 
 
@@ -2660,7 +2664,7 @@ class select_dump :public select_to_file {
     nest_level= nest_level_arg;
   }
   int prepare(List<Item> &list, SELECT_LEX_UNIT *u);
-  bool send_data(List<Item> &items);
+  int send_data(List<Item> &items);
 };
 
 
@@ -2681,7 +2685,7 @@ class select_insert :public select_result_interceptor {
   ~select_insert();
   int prepare(List<Item> &list, SELECT_LEX_UNIT *u);
   virtual int prepare2(void);
-  bool send_data(List<Item> &items);
+  virtual int send_data(List<Item> &items);
   virtual void store_values(List<Item> &values);
   virtual bool can_rollback_data() { return 0; }
   void send_error(uint errcode,const char *err);
@@ -2836,7 +2840,7 @@ class select_union :public select_result_interceptor
 
   select_union() :table(0) {}
   int prepare(List<Item> &list, SELECT_LEX_UNIT *u);
-  bool send_data(List<Item> &items);
+  int send_data(List<Item> &items);
   bool send_eof();
   bool flush();
 
@@ -2852,7 +2856,7 @@ class select_subselect :public select_result_interceptor
   Item_subselect *item;
 public:
   select_subselect(Item_subselect *item);
-  bool send_data(List<Item> &items)=0;
+  int send_data(List<Item> &items)=0;
   bool send_eof() { return 0; };
 };
 
@@ -2863,7 +2867,7 @@ class select_singlerow_subselect :public select_subselect
   select_singlerow_subselect(Item_subselect *item_arg)
     :select_subselect(item_arg)
   {}
-  bool send_data(List<Item> &items);
+  int send_data(List<Item> &items);
 };
 
 /* used in independent ALL/ANY optimisation */
@@ -2877,7 +2881,7 @@ class select_max_min_finder_subselect :public select_subselect
     :select_subselect(item_arg), cache(0), fmax(mx)
   {}
   void cleanup();
-  bool send_data(List<Item> &items);
+  int send_data(List<Item> &items);
   bool cmp_real();
   bool cmp_int();
   bool cmp_decimal();
@@ -2890,7 +2894,7 @@ class select_exists_subselect :public select_subselect
 public:
   select_exists_subselect(Item_subselect *item_arg)
     :select_subselect(item_arg){}
-  bool send_data(List<Item> &items);
+  int send_data(List<Item> &items);
 };
 
 /* Structs used when sorting */
@@ -3055,7 +3059,7 @@ class multi_delete :public select_result_interceptor
   multi_delete(TABLE_LIST *dt, uint num_of_tables);
   ~multi_delete();
   int prepare(List<Item> &list, SELECT_LEX_UNIT *u);
-  bool send_data(List<Item> &items);
+  int send_data(List<Item> &items);
   bool initialize_tables (JOIN *join);
   void send_error(uint errcode,const char *err);
   int do_deletes();
@@ -3099,7 +3103,7 @@ class multi_update :public select_result_interceptor
 	       enum_duplicates handle_duplicates, bool ignore);
   ~multi_update();
   int prepare(List<Item> &list, SELECT_LEX_UNIT *u);
-  bool send_data(List<Item> &items);
+  int send_data(List<Item> &items);
   bool initialize_tables (JOIN *join);
   void send_error(uint errcode,const char *err);
   int  do_updates();
@@ -3143,7 +3147,7 @@ class select_dumpvar :public select_result_interceptor {
   }
   ~select_dumpvar() {}
   int prepare(List<Item> &list, SELECT_LEX_UNIT *u);
-  bool send_data(List<Item> &items);
+  int send_data(List<Item> &items);
   bool send_eof();
   virtual bool check_simple_select() const;
   void cleanup();

sql/sql_cursor.cc

@@ -662,7 +662,7 @@ void Materialized_cursor::fetch(ulong num_rows)
       If network write failed (i.e. due to a closed socked),
       the error has already been set. Just return.
     */
-    if (result->send_data(item_list))
+    if (result->send_data(item_list) > 0)
       return;
   }
 

sql/sql_delete.cc

@@ -750,7 +750,7 @@ multi_delete::~multi_delete()
 }
 
 
-bool multi_delete::send_data(List<Item> &values)
+int multi_delete::send_data(List<Item> &values)
 {
   int secure_counter= delete_while_scanning ? -1 : 0;
   TABLE_LIST *del_table;

sql/sql_insert.cc

@@ -3178,7 +3178,7 @@ select_insert::~select_insert()
 }
 
 
-bool select_insert::send_data(List<Item> &values)
+int select_insert::send_data(List<Item> &values)
 {
   DBUG_ENTER("select_insert::send_data");
   bool error=0;

sql/sql_prepare.cc

@@ -106,7 +106,7 @@ class Select_fetch_protocol_binary: public select_send
 public:
   Select_fetch_protocol_binary(THD *thd);
   virtual bool send_fields(List<Item> &list, uint flags);
-  virtual bool send_data(List<Item> &items);
+  virtual int send_data(List<Item> &items);
   virtual bool send_eof();
 #ifdef EMBEDDED_LIBRARY
   void begin_dataset()
@@ -2839,11 +2839,11 @@ bool Select_fetch_protocol_binary::send_eof()
 }
 
 
-bool
+int
 Select_fetch_protocol_binary::send_data(List<Item> &fields)
 {
   Protocol *save_protocol= thd->protocol;
-  bool rc;
+  int rc;
 
   thd->protocol= &protocol;
   rc= select_send::send_data(fields);

sql/sql_select.cc

@@ -1811,7 +1811,7 @@ JOIN::exec()
       {
 	if (do_send_rows &&
             (procedure ? (procedure->send_row(procedure_fields_list) ||
-             procedure->end_of_records()) : result->send_data(fields_list)))
+             procedure->end_of_records()) : result->send_data(fields_list)> 0))
 	  error= 1;
 	else
 	{
@@ -7422,7 +7422,7 @@ return_zero_rows(JOIN *join, select_result *result,TABLE_LIST *tables,
       Item *item;
       while ((item= it++))
 	item->no_rows_in_result();
-      send_error= result->send_data(fields);
+      send_error= result->send_data(fields) > 0;
     }
     if (!send_error)
       result->send_eof();				// Should be safe
@@ -11456,7 +11456,7 @@ do_select(JOIN *join,List<Item> *fields,TABLE *table,Procedure *procedure)
     {
       List<Item> *columns_list= (procedure ? &join->procedure_fields_list :
                                  fields);
-      rc= join->result->send_data(*columns_list);
+      rc= join->result->send_data(*columns_list) > 0;
     }
   }
   else
@@ -12667,7 +12667,13 @@ end_send(JOIN *join, JOIN_TAB *join_tab __attribute__((unused)),
     if (join->procedure)
       error=join->procedure->send_row(join->procedure_fields_list);
     else if (join->do_send_rows)
-      error=join->result->send_data(*join->fields);
+    {
+      if ((error= join->result->send_data(*join->fields)) < 0)
+      {
+        /* row was not accepted. Don't count it */
+        DBUG_RETURN(NESTED_LOOP_OK);
+      }
+    }
     if (error)
       DBUG_RETURN(NESTED_LOOP_ERROR); /* purecov: inspected */
     if (++join->send_records >= join->unit->select_limit_cnt &&
@@ -12779,7 +12785,15 @@ end_send_group(JOIN *join, JOIN_TAB *join_tab __attribute__((unused)),
 	  else
 	  {
 	    if (join->do_send_rows)
-	      error=join->result->send_data(*join->fields) ? 1 : 0;
+            {
+	      error= join->result->send_data(*join->fields);
+              if (error < 0)
+              {
+                /* Duplicate row, don't count */
+                join->send_records--;
+                error= 0;
+              }
+            }
 	    join->send_records++;
 	  }
 	  if (join->rollup.state != ROLLUP::STATE_NONE && error <= 0)
@@ -16653,6 +16667,7 @@ int JOIN::rollup_send_data(uint idx)
   uint i;
   for (i= send_group_parts ; i-- > idx ; )
   {
+    int res= 0;
     /* Get reference pointers to sum functions in place */
     memcpy((char*) ref_pointer_array,
 	   (char*) rollup.ref_pointer_arrays[i],
@@ -16660,9 +16675,10 @@ int JOIN::rollup_send_data(uint idx)
     if ((!having || having->val_int()))
     {
       if (send_records < unit->select_limit_cnt && do_send_rows &&
-	  result->send_data(rollup.fields[i]))
+	  (res= result->send_data(rollup.fields[i])) > 0)
 	return 1;
-      send_records++;
+      if (!res)
+        send_records++;
     }
   }
   /* Restore ref_pointer_array */

sql/sql_union.cc

@@ -49,7 +49,7 @@ int select_union::prepare(List<Item> &list, SELECT_LEX_UNIT *u)
 }
 
 
-bool select_union::send_data(List<Item> &values)
+int select_union::send_data(List<Item> &values)
 {
   int error= 0;
   if (unit->offset_limit_cnt)
@@ -63,6 +63,14 @@ bool select_union::send_data(List<Item> &values)
 
   if ((error= table->file->ha_write_row(table->record[0])))
   {
+    if (error == HA_ERR_FOUND_DUPP_KEY)
+    {
+      /*
+        Inform upper level that we found a duplicate key, that should not
+        be counted as part of limit
+      */
+      return -1;
+    }
     /* create_internal_tmp_table_from_heap will generate error if needed */
     if (table->file->is_fatal_error(error, HA_CHECK_DUP) &&
         create_internal_tmp_table_from_heap(thd, table, &tmp_table_param, error, 1))

sql/sql_update.cc

@@ -1682,7 +1682,7 @@ multi_update::~multi_update()
 }
 
 
-bool multi_update::send_data(List<Item> &not_used_values)
+int multi_update::send_data(List<Item> &not_used_values)
 {
   TABLE_LIST *cur_table;
   DBUG_ENTER("multi_update::send_data");