Implement ORDER BY DESC optimization, which reads values in descending

order directly from the index instead of using a filesort.


mysql-test/mysql-test-run.sh:
  [ -n $SKIP_TEST ] --> [ -n "$SKIP_TEST" ]; portability fix
mysql-test/r/order_by.result:
  Added test for ORDER BY DESC optimization
mysql-test/t/order_by.test:
  Added test for ORDER BY DESC optimization
sql/opt_range.cc:
  Added QUICK_SELECT_DESC class which implements ORDER BY DESC optimization.
sql/opt_range.h:
  Added QUICK_SELECT_DESC class which implements ORDER BY DESC optimization.
sql/sql_select.cc:
  Added QUICK_SELECT_DESC class which implements ORDER BY DESC optimization.
BitKeeper/etc/ignore:
  Added .gdbinit .vimrc to the ignore list
BitKeeper/etc/logging_ok:
  Logging to logging@openlogging.org accepted

8 files changed, 263 insertions, 6 deletions

diff --git a/.bzrignore b/.bzrignore
index 2c5833b8912..e8383bf4f64 100644
--- a/.bzrignore
+++ b/.bzrignore
@@ -368,3 +368,5 @@ libmysqld/hash_filo.cc
 libmysqld/sql_unions.cc
 libmysqld/stacktrace.c
 sql/share/mysql
+.gdbinit
+.vimrc
diff --git a/BitKeeper/etc/logging_ok b/BitKeeper/etc/logging_ok
index 88c78fef7b7..93b5d236970 100644
--- a/BitKeeper/etc/logging_ok
+++ b/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
diff --git a/mysql-test/mysql-test-run.sh b/mysql-test/mysql-test-run.sh
index c1f00f00fc2..5029601adb6 100644
--- a/mysql-test/mysql-test-run.sh
+++ b/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
diff --git a/mysql-test/r/order_by.result b/mysql-test/r/order_by.result
index 74c8bd53af2..5c9e20c35a3 100644
--- a/mysql-test/r/order_by.result
+++ b/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
diff --git a/mysql-test/t/order_by.test b/mysql-test/t/order_by.test
index 4e5cee0d0ff..7d1c6346aa1 100644
--- a/mysql-test/t/order_by.test
+++ b/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: */
diff --git a/sql/opt_range.cc b/sql/opt_range.cc
index b95b97d670f..cd7c8196c4d 100644
--- a/sql/opt_range.cc
+++ b/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:
diff --git a/sql/opt_range.h b/sql/opt_range.h
index 247dd260817..0c8dcf7fed3 100644
--- a/sql/opt_range.h
+++ b/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
diff --git a/sql/sql_select.cc b/sql/sql_select.cc
index bb1e72d943f..6a5d3fede15 100644
--- a/sql/sql_select.cc
+++ b/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
   {