Implement ORDER BY DESC optimization, which reads values in descending

order directly from the index instead of using a filesort.

.bzrignore

@@ -368,3 +368,5 @@ libmysqld/hash_filo.cc
 libmysqld/sql_unions.cc
 libmysqld/stacktrace.c
 sql/share/mysql
+.gdbinit
+.vimrc

BitKeeper/etc/logging_ok

@@ -20,3 +20,4 @@ tim@threads.polyesthetic.msg
 tim@work.mysql.com
 tonu@hundin.mysql.fi
 tonu@x3.internalnet
+tim@white.box

mysql-test/mysql-test-run.sh

@@ -250,7 +250,7 @@ fi
 
 [ -z "$COLUMNS" ] && COLUMNS=80
 E=`$EXPR $COLUMNS - 8`
-#DASH72=`expr substr '------------------------------------------------------------------------' 1 $E`
+#DASH72=`$EXPR substr '------------------------------------------------------------------------' 1 $E`
 DASH72=`$ECHO '------------------------------------------------------------------------'|$CUT -c 1-$E`
 
 # on source dist, we pick up freshly build executables
@@ -667,7 +667,7 @@ run_testcase ()
  slave_init_script=$TESTDIR/$tname-slave.sh
  slave_master_info_file=$TESTDIR/$tname-slave-master-info.opt
  SKIP_SLAVE=`$EXPR \( $tname : rpl \) = 0`
- if [ -n $SKIP_TEST ] ; then 
+ if [ -n "$SKIP_TEST" ] ; then 
    SKIP_THIS_TEST=`$EXPR \( $tname : "$SKIP_TEST" \) != 0`
    if [ x$SKIP_THIS_TEST = x1 ] ;
    then

mysql-test/r/order_by.result

@@ -111,3 +111,59 @@ DateOfAction	TransactionID
 member_id	nickname	voornaam
 1		
 2		
+table	type	possible_keys	key	key_len	ref	rows	Extra
+t1	index	NULL	a	20	NULL	10	Using index
+a	b	c
+1	NULL	NULL
+1	NULL	b
+1	1	NULL
+1	1	b
+1	1	b
+2	0	a
+2	0	b
+2	1	a
+2	1	b
+2	1	c
+table	type	possible_keys	key	key_len	ref	rows	Extra
+t1	index	NULL	a	20	NULL	10	Using index
+a	b	c
+2	1	c
+2	1	b
+2	1	a
+2	0	b
+2	0	a
+1	1	b
+1	1	b
+1	1	NULL
+1	NULL	b
+1	NULL	NULL
+table	type	possible_keys	key	key_len	ref	rows	Extra
+t1	range	a	a	20	NULL	2	where used; Using index
+a	b	c
+1	1	b
+1	1	b
+table	type	possible_keys	key	key_len	ref	rows	Extra
+t1	range	a	a	4	NULL	5	where used; Using index
+a	b	c
+1	1	b
+1	1	b
+1	1	NULL
+table	type	possible_keys	key	key_len	ref	rows	Extra
+t1	range	a	a	9	NULL	7	where used; Using index
+a	b	c
+2	1	c
+2	1	b
+2	1	a
+2	0	b
+2	0	a
+1	1	b
+1	1	b
+1	1	NULL
+table	type	possible_keys	key	key_len	ref	rows	Extra
+t1	range	a	a	4	NULL	4	where used; Using index
+a	b	c
+1	1	b
+1	1	b
+1	1	NULL
+1	NULL	b
+1	NULL	NULL

mysql-test/t/order_by.test

@@ -205,3 +205,25 @@ select member_id, nickname, voornaam FROM members
 ORDER by lastchange_datum DESC LIMIT 2;
 drop table members;
 
+
+create table t1 (a int not null, b int, c varchar(10), key (a, b, c));
+insert into t1 values (1, NULL, NULL), (1, NULL, 'b'), (1, 1, NULL), (1, 1, 'b'), (1, 1, 'b'), (2, 0, 'a'), (2, 0, 'b'), (2, 1, 'a'), (2, 1, 'b'), (2, 1, 'c');
+explain select * from t1 order by a, b, c;
+select * from t1 order by a, b, c;
+explain select * from t1 order by a desc, b desc, c desc;
+select * from t1 order by a desc, b desc, c desc;
+# test multiple ranges, NO_MAX_RANGE and EQ_RANGE
+explain select * from t1 where (a = 1 and b is null and c = 'b') or (a > 2) order by a desc;
+select * from t1 where (a = 1 and b = 1 and c = 'b') or (a > 2) order by a desc;
+# test NEAR_MAX, NO_MIN_RANGE
+explain select * from t1 where a < 2 and b <= 1 order by a desc, b desc;
+select * from t1 where a < 2 and b <= 1 order by a desc, b desc;
+# test HA_READ_AFTER_KEY (at the end of the file), NEAR_MIN
+explain select * from t1 where a between 1 and 3 and b <= 1 order by a desc, b desc;
+select * from t1 where a between 1 and 3 and b <= 1 order by a desc, b desc;
+# test HA_READ_AFTER_KEY (in the middle of the file)
+explain select * from t1 where a between 0 and 1 order by a desc, b desc;
+select * from t1 where a between 0 and 1 order by a desc, b desc;
+drop table t1;
+
+/* vim:set ft=sql sw=2 noet: */

sql/opt_range.cc

@@ -33,6 +33,7 @@
 #include <m_ctype.h>
 #include <nisam.h>
 #include "sql_select.h"
+#include <assert.h>
 
 
 #ifndef EXTRA_DEBUG
@@ -289,7 +290,6 @@ typedef struct st_qsel_param {
     max_key[MAX_KEY_LENGTH+MAX_FIELD_WIDTH];
 } PARAM;
 
-
 static SEL_TREE * get_mm_parts(PARAM *param,Field *field,
 			       Item_func::Functype type,Item *value,
 			       Item_result cmp_type);
@@ -2455,8 +2455,8 @@ int QUICK_SELECT::get_next()
       if ((error=file->index_first(record)))
 	DBUG_RETURN(error);			// Empty table
       if (cmp_next(range) == 0)
-	DBUG_RETURN(0);				// No matching records
-      range=0;					// To next range
+	DBUG_RETURN(0);
+      range=0;			// No matching records; go to next range
       continue;
     }
     if ((result = file->index_read(record,(byte*) range->min_key,
@@ -2516,6 +2516,158 @@ int QUICK_SELECT::cmp_next(QUICK_RANGE *range)
   return (range->flag & NEAR_MAX) ? 1 : 0;		// Exact match
 }
 
+/*
+ * This is a hack: we inherit from QUICK_SELECT so that we can use the
+ * get_next() interface, but we have to hold a pointer to the original
+ * QUICK_SELECT because its data are used all over the place.  What
+ * should be done is to factor out the data that is needed into a base
+ * class (QUICK_SELECT), and then have two subclasses (_ASC and _DESC)
+ * which handle the ranges and implement the get_next() function.  But
+ * for now, this seems to work right at least.
+ */
+QUICK_SELECT_DESC::QUICK_SELECT_DESC(QUICK_SELECT *q)
+  : QUICK_SELECT(*q), quick(q), rev_it(rev_ranges)
+{
+  bool not_read_after_key = file->option_flag() & HA_NOT_READ_AFTER_KEY;
+  for (QUICK_RANGE *r = it++; r; r = it++)
+  {
+    rev_ranges.push_front(r);
+    if (not_read_after_key && range_reads_after_key(r))
+    {
+      error = HA_ERR_UNSUPPORTED;
+      break;
+    }
+  }
+}
+
+int QUICK_SELECT_DESC::get_next()
+{
+  DBUG_ENTER("QUICK_SELECT_DESC::get_next");
+
+  /* The max key is handled as follows:
+   *   - if there is NO_MAX_RANGE, start at the end and move backwards
+   *   - if it is an EQ_RANGE and max key covers the entire key, go directly
+   *     to the key and read through it (sorting backwards is
+   *     same as sorting forwards)
+   *   - if it is NEAR_MAX, go to the key or next, step back once, and
+   *     move backwards
+   *   - otherwise (not NEAR_MAX == include the key), go after the key,
+   *     step back once, and move backwards
+   */
+
+  for (;;)
+  {
+    int result;
+    if (range)
+    {						// Already read through key
+      result = ((range->flag & EQ_RANGE) ?
+		file->index_next_same(record, (byte*) range->min_key,
+				      range->min_length) :
+		file->index_prev(record));
+      if (!result)
+      {
+	if (cmp_prev(*rev_it.ref()) == 0)
+	  DBUG_RETURN(0);
+      }
+      else if (result != HA_ERR_END_OF_FILE)
+	DBUG_RETURN(result);
+    }
+
+    if (!(range=rev_it++))
+      DBUG_RETURN(HA_ERR_END_OF_FILE);		// All ranges used
+
+    if (range->flag & NO_MAX_RANGE)		// Read last record
+    {
+      int error;
+      if ((error=file->index_last(record)))
+	DBUG_RETURN(error);			// Empty table
+      if (cmp_prev(range) == 0)
+	DBUG_RETURN(0);
+      range=0;			// No matching records; go to next range
+      continue;
+    }
+
+    if (range->flag & EQ_RANGE &&
+	head->key_info[index].key_length == range->max_length)
+    {
+      result = file->index_read(record, (byte*) range->max_key,
+				range->max_length, HA_READ_KEY_EXACT);
+    }
+    else
+    {
+      dbug_assert(range->flag & NEAR_MAX || range_reads_after_key(range));
+      /* Note: even if max_key is only a prefix, HA_READ_AFTER_KEY will
+       * do the right thing - go past all keys which match the prefix */
+      file->index_read(record, (byte*) range->max_key, range->max_length,
+		       ((range->flag & NEAR_MAX) ?
+			HA_READ_KEY_OR_PREV : HA_READ_AFTER_KEY));
+      result = file->index_prev(record);
+    }
+    if (result)
+    {
+      if (result != HA_ERR_KEY_NOT_FOUND)
+	DBUG_RETURN(result);
+      range=0;					// Not found, to next range
+      continue;
+    }
+    if (cmp_prev(range) == 0)
+    {
+      if (range->flag == (UNIQUE_RANGE | EQ_RANGE))
+	range = 0;				// Stop searching
+      DBUG_RETURN(0);				// Found key is in range
+    }
+    range = 0;					// To next range
+  }
+  DBUG_RETURN(HA_ERR_END_OF_FILE);
+}
+
+/*
+ * Returns 0 if found key is inside range (found key >= range->min_key).
+ */
+int QUICK_SELECT_DESC::cmp_prev(QUICK_RANGE *range)
+{
+  if (range->flag & NO_MIN_RANGE)
+    return (0);					/* key can't be to small */
+
+  KEY_PART *key_part = quick->key_parts;
+  for (char *key = range->min_key, *end = key + range->min_length;
+       key < end;
+       key += key_part++->part_length)
+  {
+    int cmp;
+    if (key_part->null_bit)
+    {
+      // this key part allows null values; NULL is lower than everything else
+      if (*key++)
+      {
+	// the range is expecting a null value
+	if (!key_part->field->is_null())
+	  return 0;	// not null -- still inside the range
+	continue;	// null -- exact match, go to next key part
+      }
+      else if (key_part->field->is_null())
+	return 1;	// null -- outside the range
+    }
+    if ((cmp = key_part->field->key_cmp((byte*) key,
+					key_part->part_length)) > 0)
+      return 0;
+    if (cmp < 0)
+      return 1;
+  }
+  return (range->flag & NEAR_MIN) ? 1 : 0;		// Exact match
+}
+
+/*
+ * True if this range will require using HA_READ_AFTER_KEY
+ */
+bool QUICK_SELECT_DESC::range_reads_after_key(QUICK_RANGE *range)
+{
+  // See comment in get_next()
+  return range->flag & (NO_MAX_RANGE | NEAR_MAX) ? 1 :
+    (range->flag & EQ_RANGE &&
+     head->key_info[index].key_length == range->max_length) ? 0 : 1;
+}
+
 /*****************************************************************************
 ** Print a quick range for debugging
 ** TODO:

sql/opt_range.h

@@ -80,6 +80,20 @@ public:
   bool unique_key_range();
 };
 
+class QUICK_SELECT_DESC: public QUICK_SELECT
+{
+public:
+  QUICK_SELECT_DESC(QUICK_SELECT *q);
+  int get_next();
+private:
+  int cmp_prev(QUICK_RANGE *range);
+  bool range_reads_after_key(QUICK_RANGE *range);
+
+  QUICK_SELECT *quick;
+  List<QUICK_RANGE> rev_ranges;
+  List_iterator<QUICK_RANGE> rev_it;
+};
+
 class SQL_SELECT :public Sql_alloc {
  public:
   QUICK_SELECT *quick;		// If quick-select used

sql/sql_select.cc

@@ -5248,10 +5248,20 @@ test_if_skip_sort_order(JOIN_TAB *tab,ORDER *order,ha_rows select_limit,
 
   if (ref_key >= 0)
   {
+    int order_direction;
     /* Check if we get the rows in requested sorted order by using the key */
     if ((usable_keys & ((key_map) 1 << ref_key)) &&
-	test_if_order_by_key(order,table,ref_key) == 1)
+	(order_direction = test_if_order_by_key(order,table,ref_key)))
+    {
+      if (order_direction == -1 && select && select->quick)
+      {
+	// ORDER BY ref_key DESC
+	select->quick = new QUICK_SELECT_DESC(select->quick);
+	if (select->quick->error)
+	  DBUG_RETURN(0);
+      }
       DBUG_RETURN(1);			/* No need to sort */
+    }
   }
   else
   {