WL #12788: Iterator executor analytics queries [7/7, duplicate removal]

Implement final duplicate removal by means of filesort.

Normally, deduplication (SELECT DISTINCT) happens by adding a unique index to
the final temporary table. However, in the cases where we do aggregation
directly into a temporary table, we cannot use such an index, since rows change
during query execution and thus cannot be deduplicated on-the-fly.
(E.g., consider SELECT DISTINCT COUNT(*) FROM t1 GROUP BY f1.) The old executor solves
this by adding a post-pass that actually deletes rows from the temporary table;
for small tables, it uses a hash table to deduplicate, but for larger ones, it
uses an O(n^2) algorithm based on pairwise comparison, which is extremely slow.
Neither fits very well in an iterator design, and thus, we replace this
step by filesort, which is consistently O(n log n) with a small constant
factor. Filesort needs to be extended with support for deduplicating rows,
which is done as part of this work.

Note that this removes a determinism-by-accident that was in filesort earlier
(if the algorithm decided to sort by row ID, the row ID would be part of the
key). This also requires new functionality in the test framework, so that we
can test for partially ordered results (--partial_sorted_result).

Change-Id: I985f6d1f30630d6e8e20767c67ba8b4382144df6

Author: Steinar H. Gunderson

client/mysqltest.cc

@@ -37,7 +37,6 @@
 #include <algorithm>
 #include <chrono>
 #include <cmath>  // std::isinf
-#include <functional>
 #include <limits>
 #include <new>
 #include <regex>
@@ -211,6 +210,7 @@ static bool testcase_disabled = 0;
 static bool display_result_vertically = false, display_result_lower = false,
             display_metadata = false, display_result_sorted = false,
             display_session_track_info = false;
+static int start_sort_column = 0;
 static bool disable_query_log = 0, disable_result_log = 0;
 static bool disable_connect_log = 1;
 static bool disable_warnings = 0;
@@ -572,6 +572,7 @@ enum enum_commands {
   Q_QUERY_VERTICAL,
   Q_QUERY_HORIZONTAL,
   Q_SORTED_RESULT,
+  Q_PARTIALLY_SORTED_RESULT,
   Q_LOWERCASE,
   Q_START_TIMER,
   Q_END_TIMER,
@@ -638,11 +639,11 @@ const char *command_names[] = {
     "disable_async_client", "exec", "execw", "exec_in_background", "delimiter",
     "disable_abort_on_error", "enable_abort_on_error", "vertical_results",
     "horizontal_results", "query_vertical", "query_horizontal", "sorted_result",
-    "lowercase_result", "start_timer", "end_timer", "character_set",
-    "disable_ps_protocol", "enable_ps_protocol", "disable_reconnect",
-    "enable_reconnect", "if", "disable_testcase", "enable_testcase",
-    "replace_regex", "replace_numeric_round", "remove_file", "file_exists",
-    "write_file", "copy_file", "perl", "die",
+    "partially_sorted_result", "lowercase_result", "start_timer", "end_timer",
+    "character_set", "disable_ps_protocol", "enable_ps_protocol",
+    "disable_reconnect", "enable_reconnect", "if", "disable_testcase",
+    "enable_testcase", "replace_regex", "replace_numeric_round", "remove_file",
+    "file_exists", "write_file", "copy_file", "perl", "die",
 
     /* Don't execute any more commands, compare result */
     "exit", "skip", "chmod", "append_file", "cat_file", "diff_files",
@@ -1110,7 +1111,8 @@ static void mysql_free_result_wrapper(MYSQL_RES *result) {
 void replace_dynstr_append_mem(DYNAMIC_STRING *ds, const char *val, size_t len);
 void replace_dynstr_append(DYNAMIC_STRING *ds, const char *val);
 void replace_dynstr_append_uint(DYNAMIC_STRING *ds, uint val);
-void dynstr_append_sorted(DYNAMIC_STRING *ds, DYNAMIC_STRING *ds_input);
+void dynstr_append_sorted(DYNAMIC_STRING *ds, DYNAMIC_STRING *ds_input,
+                          int start_sort_column);
 
 void revert_properties();
 
@@ -2653,7 +2655,7 @@ static void var_query_set(VAR *var, const char *query, const char **query_end) {
           Call the replace_numeric_round function with the specified
           precision. It may be used along with replace_result, so use the
           output from replace_result as the input for replace_numeric_round.
-       */
+*/
         if (glob_replace_numeric_round >= 0) {
           /* Copy the result from replace_result if it was used, into buffer */
           if (ds_temp.length > 0) {
@@ -8881,7 +8883,7 @@ static void run_query(struct st_connection *cn, struct st_command *command,
 
   if (display_result_sorted) {
     /* Sort the result set and append it to result */
-    dynstr_append_sorted(save_ds, &ds_sorted);
+    dynstr_append_sorted(save_ds, &ds_sorted, start_sort_column);
     ds = save_ds;
     dynstr_free(&ds_sorted);
   }
@@ -9714,6 +9716,16 @@ int main(int argc, char **argv) {
             command
           */
           display_result_sorted = true;
+          start_sort_column = 0;
+          break;
+        case Q_PARTIALLY_SORTED_RESULT:
+          /*
+            Turn on sorting of result set, will be reset after next
+            command
+          */
+          display_result_sorted = true;
+          start_sort_column = atoi(command->first_argument);
+          command->last_argument = command->end;
           break;
         case Q_LOWERCASE:
           /*
@@ -11315,18 +11327,33 @@ void replace_dynstr_append_uint(DYNAMIC_STRING *ds, uint val) {
   dynstr_append_sorted
   ds - string where the sorted output will be appended
   ds_input - string to be sorted
-
+  start_sort_column - column to start sorting from (0 for sorting
+    the entire line); a stable sort will be used
 */
 
-class Comp_lines
-    : public std::binary_function<const char *, const char *, bool> {
+class Comp_lines {
  public:
   bool operator()(const char *a, const char *b) {
     return std::strcmp(a, b) < 0;
   }
 };
 
-void dynstr_append_sorted(DYNAMIC_STRING *ds, DYNAMIC_STRING *ds_input) {
+static size_t length_of_n_first_columns(const char *str,
+                                        int start_sort_column) {
+  const char *ptr = str;
+  for (int i = 0; i < start_sort_column; ++i) {
+    const char *first_tab = strchr(ptr, '\t');
+    if (first_tab == nullptr) {
+      return strlen(str);
+    } else {
+      ptr = first_tab + 1;
+    }
+  }
+  return ptr - str;
+}
+
+void dynstr_append_sorted(DYNAMIC_STRING *ds, DYNAMIC_STRING *ds_input,
+                          int start_sort_column) {
   char *start = ds_input->str;
   Prealloced_array<const char *, 32> lines(PSI_NOT_INSTRUMENTED);
   DBUG_ENTER("dynstr_append_sorted");
@@ -11339,21 +11366,40 @@ void dynstr_append_sorted(DYNAMIC_STRING *ds, DYNAMIC_STRING *ds_input) {
   dynstr_append_mem(ds, ds_input->str, start - ds_input->str);
 
   /* Insert line(s) in array */
+  size_t first_unsorted_row = 0;
   while (*start) {
     char *line_end = (char *)start;
 
     /* Find end of line */
     while (*line_end && *line_end != '\n') line_end++;
     *line_end = 0;
 
+    if (!lines.empty() && start_sort_column > 0) {
+      /*
+        If doing partial sorting, and the prefix is different from that of the
+        previous line, the group is done. Sort it and start another one.
+       */
+      size_t prev_line_prefix_len =
+          length_of_n_first_columns(lines.back(), start_sort_column);
+      size_t this_line_prefix_len =
+          length_of_n_first_columns(start, start_sort_column);
+      if (this_line_prefix_len != prev_line_prefix_len ||
+          memcmp(lines.back(), start, prev_line_prefix_len) != 0) {
+        std::sort(lines.begin() + first_unsorted_row, lines.end(),
+                  Comp_lines());
+        first_unsorted_row = lines.size();
+      }
+    }
+
     /* Insert pointer to the line in array */
     if (lines.push_back(start)) die("Out of memory inserting lines to sort");
 
     start = line_end + 1;
   }
 
   /* Sort array */
-  std::sort(lines.begin(), lines.end(), Comp_lines());
+  std::stable_sort(lines.begin() + first_unsorted_row, lines.end(),
+                   Comp_lines());
 
   /* Create new result */
   for (const char **line = lines.begin(); line != lines.end(); ++line) {

mysql-test/include/bug13581713.inc

@@ -4,8 +4,10 @@ insert into t1 values(100,1,2),(200,1,1),(300,2,1),(400,2,2);
 analyze table t1;
 
 eval select distinct b from $source order by c;
+--sorted_result
 eval select distinct min(b) from $source group by a order by min(c);
 # just to see that if source is a view, it is merged
+--sorted_result
 --replace_column 9 #
 eval explain select distinct min(b) from $source group by a order by min(c);
 
@@ -23,6 +25,7 @@ analyze table t1;
 eval select distinct b from $source order by c;
 
 --echo Random order too (same reason):
+--sorted_result
 eval select distinct min(b) from $source group by a order by min(c);
 
 --source include/restore_sql_mode_after_turn_off_only_full_group_by.inc
@@ -35,6 +38,7 @@ eval select distinct b from $source order by c;
 eval select distinct b from $source order by b-1,b+1,c;
 
 --echo and this one too:
+--sorted_result
 --error ER_AGGREGATE_IN_ORDER_NOT_SELECT
 eval select distinct min(b) from $source group by a order by min(c);
 

mysql-test/include/varchar.inc

@@ -92,7 +92,7 @@ explain select * from t1 where v='a';
 #
 # GROUP BY
 # Note that we have a number of keys here that are equivalent under
-# PAD collations, e.g. "a" equals "a ", and there's no knowing which one
+# NO PAD collations, e.g. "a" equals "a ", and there's no knowing which one
 # we get when we just say GROUP BY <field>. Thus, we add TRIM() around it
 # to get a deterministic result.
 #
@@ -105,9 +105,9 @@ select sql_big_result TRIM(v),count(c) from t1 group by TRIM(v) order by TRIM(v)
 select c,count(*) from t1 group by c order by c limit 10;
 select c,count(t) from t1 group by c order by c limit 10;
 select sql_big_result c,count(t) from t1 group by c limit 10;
-select t,count(*) from t1 group by t order by t limit 10;
-select t,count(t) from t1 group by t order by t limit 10;
-select sql_big_result t,count(t) from t1 group by t order by t limit 10;
+select TRIM(t),count(*) from t1 group by TRIM(t) order by TRIM(t) limit 10;
+select TRIM(t),count(t) from t1 group by TRIM(t) order by TRIM(t) limit 10;
+select sql_big_result TRIM(t),count(t) from t1 group by TRIM(t) order by TRIM(t) limit 10;
 
 #
 # Test varchar > 255 bytes

mysql-test/mysql-test-run.dox

@@ -7610,6 +7610,21 @@
       server bugs that result in true problems with result order.
     </li>
 
+    <li>
+      <tt>partially_sorted_result <b>start_column</b></tt>
+
+      Similarly to <b>sorted_result</b>, but assumes that the first
+      <b>start_column</b> columns are already sorted, sorting only
+      on the remaining ones. This is useful if you wish to verify
+      a partial ordering property, e.g. that <b>SELECT a,b,c ORDER BY
+      a</b>. In this case, you could use <b>partially_sorted_result 1</b>
+      to sort only on b and c, giving deterministic results without
+      masking bugs where sorting on a column would not work correctly.
+
+      <tt>sorted_result</tt> is functionally equal to
+      <tt>partially_sorted_result 0</tt>.
+    </li>
+
     <li>
       <tt>source <b>file_name</b></tt>
 

mysql-test/r/compress.result

@@ -596,16 +596,16 @@ companynr	fld3	count(*)
 SET BIG_TABLES=0;
 select distinct fld3,repeat("a",length(fld3)),count(*) from t2 group by companynr,fld3 limit 100,10;
 fld3	repeat("a",length(fld3))	count(*)
-bivalves	aaaaaaaa	1
-incurring	aaaaaaaaa	1
-Adolph	aaaaaa	1
-pithed	aaaaaa	1
-emergency	aaaaaaaaa	1
-Miles	aaaaa	1
-trimmings	aaaaaaaaa	1
-tragedies	aaaaaaaaa	1
-skulking	aaaaaaaa	1
-flint	aaaaa	1
+babysitting	aaaaaaaaaaa	1
+Baird	aaaaa	1
+balled	aaaaaa	1
+ballgown	aaaaaaaa	1
+Baltimorean	aaaaaaaaaaa	1
+bankruptcies	aaaaaaaaaaaa	1
+Barry	aaaaa	1
+batting	aaaaaaa	1
+beaner	aaaaaa	1
+beasts	aaaaaa	1
 select distinct companynr,rtrim(space(512+companynr)) from t3 order by 1,2;
 companynr	rtrim(space(512+companynr))
 37	

mysql-test/r/distinct.result

@@ -929,6 +929,14 @@ c1	c2	COUNT(*)
 2	2	1
 3	1	2
 4	4	1
+EXPLAIN FORMAT=tree SELECT DISTINCT c2 FROM t1 GROUP BY c1 HAVING COUNT(*) > 1;
+EXPLAIN
+-> Sort with duplicate removal: <temporary>.c2
+    -> Filter: (count(0) > 1)
+        -> Table scan on <temporary>
+            -> Aggregate using temporary table
+                -> Table scan on t1
+
 SELECT DISTINCT c2 FROM t1 GROUP BY c1 HAVING COUNT(*) > 1;
 c2
 1
@@ -1096,13 +1104,13 @@ b
 1
 select distinct min(b) from t1 group by a order by min(c);
 min(b)
-2
 1
+2
 explain select distinct min(b) from t1 group by a order by min(c);
 id	select_type	table	partitions	type	possible_keys	key	key_len	ref	rows	filtered	Extra
 1	SIMPLE	t1	NULL	ALL	NULL	NULL	NULL	#	4	100.00	Using temporary; Using filesort
-Warnings:
 Note	1003	/* select#1 */ select distinct min(`test`.`t1`.`b`) AS `min(b)` from `test`.`t1` group by `test`.`t1`.`a` order by min(`test`.`t1`.`c`)
+Warnings:
 Insert rows in different order:
 delete from t1;
 insert into t1 values(200,1,1),(100,1,2),(400,2,2),(300,2,1);
@@ -1324,14 +1332,14 @@ b
 1
 select distinct min(b) from v1 group by a order by min(c);
 min(b)
-2
 1
+2
 explain select distinct min(b) from v1 group by a order by min(c);
 id	select_type	table	partitions	type	possible_keys	key	key_len	ref	rows	filtered	Extra
 1	SIMPLE	t1	NULL	ALL	NULL	NULL	NULL	#	4	100.00	Using temporary; Using filesort
 1	SIMPLE	t2	NULL	ALL	NULL	NULL	NULL	#	1	100.00	Using where; Using join buffer (Block Nested Loop)
-Warnings:
 Note	1003	/* select#1 */ select distinct min(`test`.`t1`.`b`) AS `min(b)` from `test`.`t1` left join `test`.`t2` on(true) where true group by `test`.`t1`.`a` order by min(`test`.`t1`.`c`)
+Warnings:
 Insert rows in different order:
 delete from t1;
 insert into t1 values(200,1,1),(100,1,2),(400,2,2),(300,2,1);
@@ -1554,13 +1562,13 @@ b
 2
 select distinct min(b) from v1 group by a order by min(c);
 min(b)
-4
 2
+4
 explain select distinct min(b) from v1 group by a order by min(c);
 id	select_type	table	partitions	type	possible_keys	key	key_len	ref	rows	filtered	Extra
 1	SIMPLE	t1	NULL	ALL	NULL	NULL	NULL	#	4	100.00	Using temporary; Using filesort
-Warnings:
 Note	1003	/* select#1 */ select distinct min((`test`.`t1`.`b` * 2)) AS `min(b)` from `test`.`t1` group by (`test`.`t1`.`a` * 2) order by min((`test`.`t1`.`c` * 2))
+Warnings:
 Insert rows in different order:
 delete from t1;
 insert into t1 values(200,1,1),(100,1,2),(400,2,2),(300,2,1);
@@ -1782,14 +1790,14 @@ b
 1
 select distinct min(b) from (SELECT t1.* FROM t1 left join t2 on 1) AS derived group by a order by min(c);
 min(b)
-2
 1
+2
 explain select distinct min(b) from (SELECT t1.* FROM t1 left join t2 on 1) AS derived group by a order by min(c);
 id	select_type	table	partitions	type	possible_keys	key	key_len	ref	rows	filtered	Extra
 1	SIMPLE	t1	NULL	ALL	NULL	NULL	NULL	#	4	100.00	Using temporary; Using filesort
 1	SIMPLE	t2	NULL	ALL	NULL	NULL	NULL	#	1	100.00	Using where; Using join buffer (Block Nested Loop)
-Warnings:
 Note	1003	/* select#1 */ select distinct min(`test`.`t1`.`b`) AS `min(b)` from `test`.`t1` left join `test`.`t2` on(true) where true group by `test`.`t1`.`a` order by min(`test`.`t1`.`c`)
+Warnings:
 Insert rows in different order:
 delete from t1;
 insert into t1 values(200,1,1),(100,1,2),(400,2,2),(300,2,1);
@@ -2106,7 +2114,37 @@ pk
 12
 EXPLAIN FORMAT=tree SELECT /*+JOIN_ORDER(t2,t3,t1) */ DISTINCT t2.pk FROM t1 LEFT JOIN t2 RIGHT OUTER JOIN t3 ON t2.f1 = t3.f3 ON t1.pk = t3.f2 WHERE t3.pk <> t2.pk;
 EXPLAIN
-<not executable by iterator executor>
+-> Limit: 1 row(s)
+    -> Table scan on <temporary>
+        -> Temporary table
+            -> Filter: ((t3.f2 = '3') and (t3.f3 = '4') and (t3.pk <> '12'))
+                -> Table scan on t3
 
 DROP TABLE t1, t2, t3;
 SET optimizer_switch=@old_optimizer_switch;
+#
+# Test that DISTINCT-by-filesort manages to deduplicate across sort chunks.
+#
+SET @old_sort_buffer_size = @@sort_buffer_size;
+SET @@sort_buffer_size = 32768;
+CREATE TABLE t1 ( f FLOAT );
+INSERT INTO t1 VALUES (0.0);
+INSERT INTO t1 SELECT RAND() FROM t1 AS t1, t1 AS t2;
+INSERT INTO t1 SELECT RAND() FROM t1 AS t1, t1 AS t2;
+INSERT INTO t1 SELECT RAND() FROM t1 AS t1, t1 AS t2;
+INSERT INTO t1 SELECT RAND() FROM t1 AS t1, t1 AS t2;
+INSERT INTO t1 SELECT RAND() FROM t1;
+INSERT INTO t1 SELECT RAND() FROM t1;
+EXPLAIN FORMAT=tree SELECT DISTINCT COUNT(*) AS num FROM t1 GROUP BY f HAVING num=1;
+EXPLAIN
+-> Sort with duplicate removal: <temporary>.num
+    -> Filter: (num = 1)
+        -> Table scan on <temporary>
+            -> Aggregate using temporary table
+                -> Table scan on t1
+
+SELECT DISTINCT COUNT(*) AS num FROM t1 GROUP BY f HAVING num=1;
+num
+1
+DROP TABLE t1;
+SET @@sort_buffer_size = @old_sort_buffer_size;

mysql-test/r/explain_json_all.result

@@ -2584,7 +2584,7 @@ s
 SHOW SESSION STATUS WHERE (Variable_name LIKE 'Sort_%' OR Variable_name LIKE 'Created_%_tables') AND Value > 0;
 Variable_name	Value
 Created_tmp_tables	1
-Sort_rows	2
+Sort_rows	3
 Sort_scan	1
 DROP TABLE t1;
 # "buffer_result" node