Fix for bug#59308: Incorrect result for SELECT DISTINCT <col>... ORDER BY <col> DESC.

      
Also fix bug#59110: Memory leak of QUICK_SELECT_I allocated memory.
Includes Jørgen Lølands review comments.
      
Root cause of these bugs are that test_if_skip_sort_order() decided to
revert the 'skip_sort_order' descision (and use filesort) after the
query plan has been updated to reflect a 'skip' of the sort order.
      
This might happen in 'check_reverse_order:' if we have a 
select->quick which could not be made descending by appending 
a QUICK_SELECT_DESC. ().
      
The original 'save_quick' was then restored after the QEP has been modified,
which caused:
      
  - An incorrect 'precomputed_group_by= TRUE' may have been set, 
    and not reverted, as part of the already modifified QEP (Bug#59308)
  - A 'select->quick' might have been created which we fail to delete (bug#59110).
      
This fix is a refactorication of test_if_skip_sort_order() where all logic
related to modification of QEP (controlled by argument 'bool no_changes'), is
moved to the end of test_if_skip_sort_order(), and done after *all* 'test_if_skip'
checks has been performed - including the 'check_reverse_order:' checks.
      
The refactorication above contains now intentional changes to the logic which 
has been moved to the end of the function.
      
Furthermore, a smaller part of the fix address the handling of the 
select->quick objects which may already exists when we call 
'test_if_skip_sort_order()' (save_quick) -and
new select->quick's created during test_if_skip_sort_order():
      
  - Before new select->quick may be created by calling ::test_quick_select(), we
    set 'select->quick= 0' to avoid that ::test_quick_select() prematurely
    delete the save_quick's. (After this call we may have both a 'save_quick' 
    and 'select->quick')
      
  - All returns from ::test_if_skip_sort_order() where we may have both a
    'save_quick' and a 'select->quick' has been changed to goto's to the
    exit points 'skiped_sort_order:' or 'need_filesort:' where we
    decide which of the QUICK_SELECT's to keep, and delete the other.

mysql-test/r/order_by.result

@@ -1638,4 +1638,29 @@ id	select_type	table	type	possible_keys	key	key_len	ref	rows	Extra
 1	SIMPLE	t1	index	NULL	a	8	NULL	10	Using index; Using temporary; Using filesort
 1	SIMPLE	t2	eq_ref	PRIMARY	PRIMARY	4	test.t1.b	1	Using where
 DROP TABLE t1, t2;
+#
+# Bug #59110: Memory leak of QUICK_SELECT_I allocated memory 
+#  and
+# Bug #59308: Incorrect result for 
+SELECT DISTINCT <col>... ORDER BY <col> DESC 
+
+# Use Valgrind to detect #59110!
+#
+CREATE TABLE t1 (a INT,KEY (a));
+INSERT INTO t1 VALUES (1),(2),(3),(4),(5),(6),(7),(8),(9),(10);
+EXPLAIN SELECT DISTINCT a,1 FROM t1 WHERE a <> 1 ORDER BY a DESC;
+id	select_type	table	type	possible_keys	key	key_len	ref	rows	Extra
+1	SIMPLE	t1	index	a	a	5	NULL	10	Using where; Using index; Using filesort
+SELECT DISTINCT a,1 FROM t1 WHERE a <> 1 ORDER BY a DESC;
+a	1
+10	1
+9	1
+8	1
+7	1
+6	1
+5	1
+4	1
+3	1
+2	1
+DROP TABLE t1;
 End of 5.1 tests

mysql-test/t/order_by.test

@@ -1492,4 +1492,21 @@ LIMIT 2;
 DROP TABLE t1, t2;
 
 
+--echo #
+--echo # Bug #59110: Memory leak of QUICK_SELECT_I allocated memory 
+--echo #  and
+--echo # Bug #59308: Incorrect result for 
+--echo               SELECT DISTINCT <col>... ORDER BY <col> DESC 
+--echo
+--echo # Use Valgrind to detect #59110!
+--echo #
+
+CREATE TABLE t1 (a INT,KEY (a));
+INSERT INTO t1 VALUES (1),(2),(3),(4),(5),(6),(7),(8),(9),(10);
+
+EXPLAIN SELECT DISTINCT a,1 FROM t1 WHERE a <> 1 ORDER BY a DESC;
+SELECT DISTINCT a,1 FROM t1 WHERE a <> 1 ORDER BY a DESC;
+
+DROP TABLE t1;
+
 --echo End of 5.1 tests

sql/sql_select.cc

@@ -13364,12 +13364,14 @@ test_if_skip_sort_order(JOIN_TAB *tab,ORDER *order,ha_rows select_limit,
 {
   int ref_key;
   uint ref_key_parts;
-  int order_direction;
+  int order_direction= 0;
   uint used_key_parts;
   TABLE *table=tab->table;
   SQL_SELECT *select=tab->select;
   key_map usable_keys;
   QUICK_SELECT_I *save_quick= 0;
+  int best_key= -1;
+
   DBUG_ENTER("test_if_skip_sort_order");
   LINT_INIT(ref_key_parts);
 
@@ -13473,13 +13475,14 @@ test_if_skip_sort_order(JOIN_TAB *tab,ORDER *order,ha_rows select_limit,
           new_ref_key_map.clear_all();  // Force the creation of quick select
           new_ref_key_map.set_bit(new_ref_key); // only for new_ref_key.
 
+          select->quick= 0;
           if (select->test_quick_select(tab->join->thd, new_ref_key_map, 0,
                                         (tab->join->select_options &
                                          OPTION_FOUND_ROWS) ?
                                         HA_POS_ERROR :
                                         tab->join->unit->select_limit_cnt,0) <=
               0)
-            DBUG_RETURN(0);
+            goto use_filesort;
 	}
         ref_key= new_ref_key;
       }
@@ -13504,7 +13507,6 @@ test_if_skip_sort_order(JOIN_TAB *tab,ORDER *order,ha_rows select_limit,
     int best_key_direction= 0;
     ha_rows best_records= 0;
     double read_time;
-    int best_key= -1;
     bool is_best_covering= FALSE;
     double fanout= 1;
     JOIN *join= tab->join;
@@ -13681,25 +13683,76 @@ test_if_skip_sort_order(JOIN_TAB *tab,ORDER *order,ha_rows select_limit,
          tab->join->tables > tab->join->const_tables + 1) &&
          ((unsigned) best_key != table->s->primary_key ||
           !table->file->primary_key_is_clustered()))
-      DBUG_RETURN(0);
+      goto use_filesort;
 
     if (best_key >= 0)
     {
-      bool quick_created= FALSE;
       if (table->quick_keys.is_set(best_key) && best_key != ref_key)
       {
         key_map map;
         map.clear_all();       // Force the creation of quick select
         map.set_bit(best_key); // only best_key.
-        quick_created=         
+        select->quick= 0;
         select->test_quick_select(join->thd, map, 0,
                                   join->select_options & OPTION_FOUND_ROWS ?
                                   HA_POS_ERROR :
                                   join->unit->select_limit_cnt,
-                                    0) > 0;
+                                  0);
+      }
+      order_direction= best_key_direction;
+      /*
+        saved_best_key_parts is actual number of used keyparts found by the
+        test_if_order_by_key function. It could differ from keyinfo->key_parts,
+        thus we have to restore it in case of desc order as it affects
+        QUICK_SELECT_DESC behaviour.
+      */
+      used_key_parts= (order_direction == -1) ?
+        saved_best_key_parts :  best_key_parts;
     }
-      if (!no_changes)
+    else
+      goto use_filesort;
+  } 
+
+check_reverse_order:                  
+  DBUG_ASSERT(order_direction != 0);
+
+  if (order_direction == -1)		// If ORDER BY ... DESC
+  {
+    if (select && select->quick)
     {
+      /*
+	Don't reverse the sort order, if it's already done.
+        (In some cases test_if_order_by_key() can be called multiple times
+      */
+      if (select->quick->reverse_sorted())
+        goto skipped_filesort;
+      else
+      {
+        int quick_type= select->quick->get_type();
+        if (quick_type == QUICK_SELECT_I::QS_TYPE_INDEX_MERGE ||
+            quick_type == QUICK_SELECT_I::QS_TYPE_ROR_INTERSECT ||
+            quick_type == QUICK_SELECT_I::QS_TYPE_ROR_UNION ||
+            quick_type == QUICK_SELECT_I::QS_TYPE_GROUP_MIN_MAX)
+        {
+          tab->limit= 0;
+          goto use_filesort;               // Use filesort
+        }
+      }
+    }
+  }
+
+  /*
+    Update query plan with access pattern for doing 
+    ordered access according to what we have decided
+    above.
+  */
+  if (!no_changes) // We are allowed to update QEP
+  {
+    if (best_key >= 0)
+    {
+      bool quick_created= 
+        (select && select->quick && select->quick!=save_quick);
+
       /* 
          If ref_key used index tree reading only ('Using index' in EXPLAIN),
          and best_key doesn't, then revert the decision.
@@ -13708,23 +13761,22 @@ test_if_skip_sort_order(JOIN_TAB *tab,ORDER *order,ha_rows select_limit,
         table->set_keyread(FALSE);
       if (!quick_created)
       {
+        if (select)                  // Throw any existing quick select
+          select->quick= 0;          // Cleanup either reset to save_quick,
+                                     // or 'delete save_quick'
         tab->index= best_key;
-          tab->read_first_record= best_key_direction > 0 ?
+        tab->read_first_record= order_direction > 0 ?
                                 join_read_first:join_read_last;
         tab->type=JT_NEXT;           // Read with index_first(), index_next()
-          if (select && select->quick)
-          {
-            delete select->quick;
-            select->quick= 0;
-          }
+
         if (table->covering_keys.is_set(best_key))
           table->set_keyread(TRUE);
         table->file->ha_index_or_rnd_end();
-          if (join->select_options & SELECT_DESCRIBE)
+        if (tab->join->select_options & SELECT_DESCRIBE)
         {
           tab->ref.key= -1;
           tab->ref.key_parts= 0;
-            if (select_limit < table_records) 
+          if (select_limit < table->file->stats.records) 
             tab->limit= select_limit;
         }
       }
@@ -13742,61 +13794,31 @@ test_if_skip_sort_order(JOIN_TAB *tab,ORDER *order,ha_rows select_limit,
         tab->ref.key_parts=0;		// Don't use ref key.
         tab->read_first_record= join_init_read_record;
         if (tab->is_using_loose_index_scan())
-            join->tmp_table_param.precomputed_group_by= TRUE;
+          tab->join->tmp_table_param.precomputed_group_by= TRUE;
         /*
           TODO: update the number of records in join->best_positions[tablenr]
         */
       }
-      }
-      order_direction= best_key_direction;
-      /*
-        saved_best_key_parts is actual number of used keyparts found by the
-        test_if_order_by_key function. It could differ from keyinfo->key_parts,
-        thus we have to restore it in case of desc order as it affects
-        QUICK_SELECT_DESC behaviour.
-      */
-      used_key_parts= (order_direction == -1) ?
-        saved_best_key_parts :  best_key_parts;
-    }
-    else
-      DBUG_RETURN(0); 
-  } 
+    } // best_key >= 0
 
-check_reverse_order:                  
     if (order_direction == -1)		// If ORDER BY ... DESC
     {
       if (select && select->quick)
-    {
-      /*
-	Don't reverse the sort order, if it's already done.
-        (In some cases test_if_order_by_key() can be called multiple times
-      */
-      if (!select->quick->reverse_sorted())
       {
         QUICK_SELECT_DESC *tmp;
-        int quick_type= select->quick->get_type();
-        if (quick_type == QUICK_SELECT_I::QS_TYPE_INDEX_MERGE ||
-            quick_type == QUICK_SELECT_I::QS_TYPE_ROR_INTERSECT ||
-            quick_type == QUICK_SELECT_I::QS_TYPE_ROR_UNION ||
-            quick_type == QUICK_SELECT_I::QS_TYPE_GROUP_MIN_MAX)
-        {
-          tab->limit= 0;
-          select->quick= save_quick;
-          DBUG_RETURN(0);                   // Use filesort
-        }
-            
         /* ORDER BY range_key DESC */
         tmp= new QUICK_SELECT_DESC((QUICK_RANGE_SELECT*)(select->quick),
                                     used_key_parts);
+        if (tmp && select->quick == save_quick)
+          save_quick= 0;    // ::QUICK_SELECT_DESC consumed it
+
         if (!tmp || tmp->error)
         {
           delete tmp;
-          select->quick= save_quick;
           tab->limit= 0;
-	  DBUG_RETURN(0);		// Reverse sort not supported
-	}
-	select->quick=tmp;
+          goto use_filesort;           // Reverse sort failed -> filesort
         }
+        select->quick= tmp;
       }
       else if (tab->type != JT_NEXT && tab->type != JT_REF_OR_NULL &&
                tab->ref.key >= 0 && tab->ref.key_parts <= used_key_parts)
@@ -13813,7 +13835,32 @@ check_reverse_order:
     }
     else if (select && select->quick)
       select->quick->sorted= 1;
+
+  } // QEP has been modified
+
+  /*
+    Cleanup:
+    We may have both a 'select->quick' and 'save_quick' (original)
+    at this point. Delete the one that we wan't use.
+  */
+
+skipped_filesort:
+  // Keep current (ordered) select->quick 
+  if (select && save_quick != select->quick)
+  {
+    delete save_quick;
+    save_quick= NULL;
+  }
   DBUG_RETURN(1);
+
+use_filesort:
+  // Restore original save_quick
+  if (select && select->quick != save_quick)
+  {
+    delete select->quick;
+    select->quick= save_quick;
+  }
+  DBUG_RETURN(0);
 }