Added support of null keys in HEAP tables

Added ORDER BY optimization


Docs/manual.texi:
  Added ORDER BY optimisation section
heap/_check.c:
  Cleanup
heap/heapdef.h:
  Added support of null keys in HEAP tables
heap/hp_hash.c:
  Added support of null keys in HEAP tables
heap/hp_open.c:
  Added support of null keys in HEAP tables
heap/hp_test2.c:
  Added support of null keys in HEAP tables
heap/hp_write.c:
  Added support of null keys in HEAP tables
include/heap.h:
  Added support of null keys in HEAP tables
include/my_base.h:
  Support for hash algoritm
isam/static.c:
  Cleanup
myisam/mi_static.c:
  Cleanup
sql/Makefile.am:
  Rename innobase -> innodb
sql/ha_berkeley.cc:
  Added ORDER BY optimization
sql/ha_berkeley.h:
  Added ORDER BY optimization
sql/ha_heap.cc:
  Added support for NULL keys
sql/ha_heap.h:
  Added support for NULL keys
sql/ha_isam.cc:
  Added ORDER BY optimization
sql/ha_isam.h:
  Added ORDER BY optimization
sql/ha_isammrg.h:
  Added ORDER BY optimization
sql/ha_myisam.cc:
  Added ORDER BY optimization
sql/ha_myisam.h:
  Added ORDER BY optimization
sql/ha_myisammrg.cc:
  Added ORDER BY optimization
sql/ha_myisammrg.h:
  Added ORDER BY optimization
sql/handler.cc:
  Rename innobase -> innodb
sql/handler.h:
  Rename innobase -> innodb
sql/ha_innodb.cc:
  Rename innobase -> innodb
sql/ha_innodb.h:
  Rename innobase -> innodb
sql/mysqld.cc:
  Rename innobase -> innodb
sql/sql_delete.cc:
  Rename innobase -> innodb
sql/sql_select.cc:
  Added ORDER BY optimization
sql/sql_select.h:
  Added ORDER BY optimization

heap/_check.c

@@ -79,9 +79,11 @@ static int check_one_key(HP_KEYDEF *keydef, uint keynr, ulong records,
   }
   DBUG_PRINT("info",
 	     ("records: %ld   seeks: %d   max links: %d   hitrate: %.2f",
-	      records,seek,max_links,(float) seek / (float) (records ? records : 1)));
+	      records,seek,max_links,
+	      (float) seek / (float) (records ? records : 1)));
   if (print_status)
     printf("Key: %d  records: %ld   seeks: %d   max links: %d   hitrate: %.2f\n",
-	   keynr, records, seek, max_links, (float) seek / (float) records);
+	   keynr, records, seek, max_links,
+	   (float) seek / (float) (records ? records : 1));
   return error;
 }

heap/heapdef.h

@@ -70,6 +70,7 @@ extern int _hp_rec_key_cmp(HP_KEYDEF *keydef,const byte *rec1,
 extern int _hp_key_cmp(HP_KEYDEF *keydef,const byte *rec,
 			   const byte *key);
 extern void _hp_make_key(HP_KEYDEF *keydef,byte *key,const byte *rec);
+extern my_bool hp_if_null_in_key(HP_KEYDEF *keyinfo, const byte *record);
 extern int _hp_close(register HP_INFO *info);
 extern void _hp_clear(HP_SHARE *info);
 

heap/hp_hash.c

@@ -158,11 +158,22 @@ ulong _hp_hashnr(register HP_KEYDEF *keydef, register const byte *key)
   {
     uchar *pos=(uchar*) key;
     key+=seg->length;
+    if (seg->null_bit)
+    {
+      key++;					/* Skipp null byte */
+      if (*pos)					/* Found null */
+      {
+	nr^= (nr << 1) | 1;
+	continue;
+      }
+      pos++;
+    }
     if (seg->type == HA_KEYTYPE_TEXT)
     {
       for (; pos < (uchar*) key ; pos++)
       {
-	nr^=(ulong) ((((uint) nr & 63)+nr2)*((uint) my_sort_order[(uint) *pos]))+ (nr << 8);
+	nr^=(ulong) ((((uint) nr & 63)+nr2) * 
+		     ((uint) my_sort_order[(uint) *pos])) + (nr << 8);
 	nr2+=3;
       }
     }
@@ -170,7 +181,7 @@ ulong _hp_hashnr(register HP_KEYDEF *keydef, register const byte *key)
     {
       for (; pos < (uchar*) key ; pos++)
       {
-	nr^=(ulong) ((((uint) nr & 63)+nr2)*((uint) *pos))+ (nr << 8);
+	nr^=(ulong) ((((uint) nr & 63)+nr2)*((uint) *pos)) + (nr << 8);
 	nr2+=3;
       }
     }
@@ -188,11 +199,20 @@ ulong _hp_rec_hashnr(register HP_KEYDEF *keydef, register const byte *rec)
   for (seg=keydef->seg,endseg=seg+keydef->keysegs ; seg < endseg ; seg++)
   {
     uchar *pos=(uchar*) rec+seg->start,*end=pos+seg->length;
+    if (seg->null_bit)
+    {
+      if (rec[seg->null_pos] & seg->null_bit)
+      {
+	nr^= (nr << 1) | 1;
+	continue;
+      }
+    }
     if (seg->type == HA_KEYTYPE_TEXT)
     {
       for (; pos < end ; pos++)
       {
-	nr^=(ulong) ((((uint) nr & 63)+nr2)*((uint) my_sort_order[(uint) *pos]))+ (nr << 8);
+	nr^=(ulong) ((((uint) nr & 63)+nr2)*
+		     ((uint) my_sort_order[(uint) *pos]))+ (nr << 8);
 	nr2+=3;
       }
     }
@@ -234,6 +254,16 @@ ulong _hp_hashnr(register HP_KEYDEF *keydef, register const byte *key)
   {
     uchar *pos=(uchar*) key;
     key+=seg->length;
+    if (seg->null_bit)
+    {
+      key++;
+      if (*pos)
+      {
+	nr^= (nr << 1) | 1;
+	continue;
+      }
+      pos++;
+    }
     if (seg->type == HA_KEYTYPE_TEXT)
     {
       for (; pos < (uchar*) key ; pos++)
@@ -264,6 +294,14 @@ ulong _hp_rec_hashnr(register HP_KEYDEF *keydef, register const byte *rec)
   for (seg=keydef->seg,endseg=seg+keydef->keysegs ; seg < endseg ; seg++)
   {
     uchar *pos=(uchar*) rec+seg->start,*end=pos+seg->length;
+    if (seg->null_bit)
+    {
+      if (rec[seg->null_pos] & seg->null_bit)
+      {
+	nr^= (nr << 1) | 1;
+	continue;
+      }
+    }
     if (seg->type == HA_KEYTYPE_TEXT)
     {
       for ( ; pos < end ; pos++)
@@ -295,6 +333,14 @@ int _hp_rec_key_cmp(HP_KEYDEF *keydef, const byte *rec1, const byte *rec2)
 
   for (seg=keydef->seg,endseg=seg+keydef->keysegs ; seg < endseg ; seg++)
   {
+    if (seg->null_bit)
+    {
+      if ((rec1[seg->null_pos] & seg->null_bit) !=
+	  (rec2[seg->null_pos] & seg->null_bit))
+	return 1;
+      if (rec1[seg->null_pos] & seg->null_bit)
+	continue;
+    }
     if (seg->type == HA_KEYTYPE_TEXT)
     {
       if (my_sortcmp(rec1+seg->start,rec2+seg->start,seg->length))
@@ -309,14 +355,24 @@ int _hp_rec_key_cmp(HP_KEYDEF *keydef, const byte *rec1, const byte *rec2)
   return 0;
 }
 
-	/* Compare a key in a record to a hole key */
+	/* Compare a key in a record to a whole key */
 
 int _hp_key_cmp(HP_KEYDEF *keydef, const byte *rec, const byte *key)
 {
   HP_KEYSEG *seg,*endseg;
 
-  for (seg=keydef->seg,endseg=seg+keydef->keysegs ; seg < endseg ; seg++)
+  for (seg=keydef->seg,endseg=seg+keydef->keysegs ;
+       seg < endseg ;
+       key+= (seg++)->length)
+  {
+    if (seg->null_bit)
     {
+      int found_null=test(rec[seg->null_pos] & seg->null_bit);
+      if (found_null != (int) *key++)
+	return 1;
+      if (found_null)
+	continue;
+    }
     if (seg->type == HA_KEYTYPE_TEXT)
     {
       if (my_sortcmp(rec+seg->start,key,seg->length))
@@ -327,7 +383,6 @@ int _hp_key_cmp(HP_KEYDEF *keydef, const byte *rec, const byte *key)
       if (bcmp(rec+seg->start,key,seg->length))
 	return 1;
     }
-    key+=seg->length;
   }
   return 0;
 }
@@ -341,7 +396,28 @@ void _hp_make_key(HP_KEYDEF *keydef, byte *key, const byte *rec)
 
   for (seg=keydef->seg,endseg=seg+keydef->keysegs ; seg < endseg ; seg++)
   {
+    if (seg->null_bit)
+      *key++= test(rec[seg->null_pos] & seg->null_bit);
     memcpy(key,rec+seg->start,(size_t) seg->length);
     key+=seg->length;
   }
 }
+
+
+/*
+  Test if any of the key parts are NULL.
+  Return:
+    1 if any of the key parts was NULL
+    0 otherwise
+*/
+
+my_bool hp_if_null_in_key(HP_KEYDEF *keydef, const byte *record)
+{
+  HP_KEYSEG *seg,*endseg;
+  for (seg=keydef->seg,endseg=seg+keydef->keysegs ; seg < endseg ; seg++)
+  {
+    if (seg->null_bit && (record[seg->null_pos] & seg->null_bit))
+      return 1;
+  }
+  return 0;
+}

heap/hp_open.c

@@ -44,7 +44,11 @@ HP_INFO *heap_open(const char *name, int mode, uint keys, HP_KEYDEF *keydef,
       key_segs+= keydef[i].keysegs;
       bzero((char*) &keydef[i].block,sizeof(keydef[i].block));
       for (j=length=0 ; j < keydef[i].keysegs; j++)
+      {
 	length+=keydef[i].seg[j].length;
+	if (keydef[i].seg[j].null_bit)
+	  keydef[i].flag |= HA_NULL_PART_KEY;
+      }
       keydef[i].length=length;
       if (length > max_length)
 	max_length=length;

heap/hp_test2.c

@@ -30,7 +30,7 @@
 #include <signal.h>
 
 #define MAX_RECORDS 100000
-#define MAX_KEYS 3
+#define MAX_KEYS 4
 
 static int get_options(int argc, char *argv[]);
 static int rnd(int max_value);
@@ -40,16 +40,20 @@ static uint flag=0,verbose=0,testflag=0,recant=10000,silent=0;
 static uint keys=MAX_KEYS;
 static uint16 key1[1001];
 static my_bool key3[MAX_RECORDS];
+static int reclength=39;
+
 
 static int calc_check(byte *buf,uint length);
+static void make_record(char *record, uint n1, uint n2, uint n3,
+			const char *mark, uint count);
 
-		/* Huvudprogrammet */
+/* Main program */
 
 int main(int argc, char *argv[])
 {
   register uint i,j;
   uint ant,n1,n2,n3;
-  uint reclength,write_count,update,opt_delete,check2,dupp_keys,found_key;
+  uint write_count,update,opt_delete,check2,dupp_keys,found_key;
   int error;
   ulong pos;
   unsigned long key_check;
@@ -66,7 +70,6 @@ int main(int argc, char *argv[])
   filename2= "test2_2";
   file=file2=0;
   get_options(argc,argv);
-  reclength=37;
 
   write_count=update=opt_delete=0;
   key_check=0;
@@ -77,21 +80,33 @@ int main(int argc, char *argv[])
   keyinfo[0].seg[0].type=HA_KEYTYPE_BINARY;
   keyinfo[0].seg[0].start=0;
   keyinfo[0].seg[0].length=6;
+  keyinfo[0].seg[0].null_bit=0;
   keyinfo[1].seg=keyseg+1;
   keyinfo[1].keysegs=2;
   keyinfo[1].flag=0;
   keyinfo[1].seg[0].type=HA_KEYTYPE_BINARY;
   keyinfo[1].seg[0].start=7;
   keyinfo[1].seg[0].length=6;
+  keyinfo[1].seg[0].null_bit=0;
   keyinfo[1].seg[1].type=HA_KEYTYPE_TEXT;
-  keyinfo[1].seg[1].start=0;			/* Tv}delad nyckel */
+  keyinfo[1].seg[1].start=0;			/* key in two parts */
   keyinfo[1].seg[1].length=6;
+  keyinfo[1].seg[1].null_bit=0;
   keyinfo[2].seg=keyseg+3;
   keyinfo[2].keysegs=1;
   keyinfo[2].flag=HA_NOSAME;
   keyinfo[2].seg[0].type=HA_KEYTYPE_BINARY;
   keyinfo[2].seg[0].start=12;
   keyinfo[2].seg[0].length=8;
+  keyinfo[2].seg[0].null_bit=0;
+  keyinfo[3].keysegs=1;
+  keyinfo[3].flag=HA_NOSAME;
+  keyinfo[3].seg=keyseg+4;
+  keyinfo[3].seg[0].type=HA_KEYTYPE_BINARY;
+  keyinfo[3].seg[0].start=37;
+  keyinfo[3].seg[0].length=1;
+  keyinfo[3].seg[0].null_bit=1;
+  keyinfo[3].seg[0].null_pos=38;
 
   bzero((char*) key1,sizeof(key1));
   bzero((char*) key3,sizeof(key3));
@@ -110,7 +125,7 @@ int main(int argc, char *argv[])
   for (i=0 ; i < recant ; i++)
   {
     n1=rnd(1000); n2=rnd(100); n3=rnd(min(recant*5,MAX_RECORDS));
-    sprintf(record,"%6d:%4d:%8d:Pos: %4d          ",n1,n2,n3,write_count);
+    make_record(record,n1,n2,n3,"Pos",write_count);
 
     if (heap_write(file,record))
     {
@@ -191,7 +206,7 @@ int main(int argc, char *argv[])
   for (i=0 ; i < write_count/10 ; i++)
   {
     n1=rnd(1000); n2=rnd(100); n3=rnd(min(recant*2,MAX_RECORDS));
-    sprintf(record2,"%6d:%4d:%8d:XXX: %4d          ",n1,n2,n3,update);
+    make_record(record2, n1, n2, n3, "XXX", update);
     if (rnd(2) == 1)
     {
       if (heap_scan_init(file))
@@ -654,3 +669,13 @@ static int calc_check(byte *buf, uint length)
     check+= (int) (uchar) *(buf++);
   return check;
 }
+
+static void make_record(char *record, uint n1, uint n2, uint n3,
+			const char *mark, uint count)
+{
+  bfill(record,reclength,' ');
+  sprintf(record,"%6d:%4d:%8d:%3.3s: %4d",
+	  n1,n2,n3,mark,count);
+  record[37]='A';				/* Store A in null key */
+  record[38]=1;					/* set as null */
+}

heap/hp_write.c

@@ -239,7 +239,9 @@ int _hp_write_key(register HP_SHARE *info, HP_KEYDEF *keyinfo,
     }
 
     /* Check if dupplicated keys */
-    if ((keyinfo->flag & HA_NOSAME) && pos == gpos)
+    if ((keyinfo->flag & HA_NOSAME) && pos == gpos &&
+	(!(keyinfo->flag & HA_NULL_PART_KEY) ||
+	 !hp_if_null_in_key(keyinfo, record)))
     {
       pos=empty;
       do

include/heap.h

@@ -78,11 +78,13 @@ typedef struct st_hp_keyseg		/* Key-portion */
   uint start;				/* Start of key in record (from 0) */
   uint length;				/* Keylength */
   uint type;
+  uint null_bit;			/* bit set in row+null_pos */
+  uint null_pos;
 } HP_KEYSEG;
 
 typedef struct st_hp_keydef		/* Key definition with open */
 {
-  uint flag;				/* NOSAME */
+  uint flag;				/* HA_NOSAME |HA_NULL_PART_KEY */
   uint keysegs;				/* Number of key-segment */
   uint length;				/* Length of key (automatic) */
   HP_KEYSEG *seg;

include/my_base.h

@@ -68,7 +68,8 @@ enum ha_rkey_function {
 
 enum ha_key_alg {			
   HA_KEY_ALG_BTREE=0,			/* B-tree, default one          */
-  HA_KEY_ALG_RTREE=1			/* R-tree, for spatial searches */
+  HA_KEY_ALG_RTREE=1,			/* R-tree, for spatial searches */
+  HA_KEY_ALG_HASH=2			/* HASH keys (HEAP tables) */
 };
 
 	/* The following is parameter to ha_extra() */

isam/static.c

@@ -15,7 +15,7 @@
    Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA */
 
 /*
-  Static variables for pisam library. All definied here for easy making of
+  Static variables for ISAM library. All definied here for easy making of
   a shared library
 */
 

myisam/mi_static.c

@@ -15,7 +15,7 @@
    Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA */
 
 /*
-  Static variables for pisam library. All definied here for easy making of
+  Static variables for MyISAM library. All definied here for easy making of
   a shared library
 */
 

sql/Makefile.am

@@ -51,7 +51,7 @@ noinst_HEADERS =	item.h item_func.h item_sum.h item_cmpfunc.h \
 			sql_manager.h sql_map.h sql_string.h unireg.h \
 			field.h handler.h \
 			ha_isammrg.h ha_isam.h ha_myisammrg.h\
-			ha_heap.h ha_myisam.h ha_berkeley.h ha_innobase.h \
+			ha_heap.h ha_myisam.h ha_berkeley.h ha_innodb.h \
 			opt_range.h opt_ft.h \
 			sql_select.h structs.h table.h sql_udf.h hash_filo.h\
 			lex.h lex_symbol.h sql_acl.h sql_crypt.h  \
@@ -74,7 +74,7 @@ mysqld_SOURCES =	sql_lex.cc sql_handler.cc \
 			time.cc opt_range.cc opt_sum.cc opt_ft.cc \
 		   	records.cc filesort.cc handler.cc \
 		        ha_heap.cc ha_myisam.cc ha_myisammrg.cc \
-	                ha_berkeley.cc ha_innobase.cc \
+	                ha_berkeley.cc ha_innodb.cc \
 			ha_isam.cc ha_isammrg.cc \
 			sql_db.cc sql_table.cc sql_rename.cc sql_crypt.cc \
 			sql_load.cc mf_iocache.cc field_conv.cc sql_show.cc \

sql/ha_berkeley.cc

@@ -1454,6 +1454,37 @@ int ha_berkeley::index_read(byte * buf, const byte * key,
   DBUG_RETURN(error);
 }
 
+/*
+  Read last key is solved by reading the next key and then reading
+  the previous key
+*/
+
+int ha_berkeley::index_read_last(byte * buf, const byte * key, uint key_len)
+{
+  DBT row;
+  int error;
+  KEY *key_info= &table->key_info[active_index];
+  DBUG_ENTER("ha_berkeley::index_read");
+
+  statistic_increment(ha_read_key_count,&LOCK_status);
+  bzero((char*) &row,sizeof(row));
+
+  /* read of partial key */
+  pack_key(&last_key, active_index, key_buff, key, key_len);
+  /* Store for compare */
+  memcpy(key_buff2, key_buff, (key_len=last_key.size));
+  key_info->handler.bdb_return_if_eq= 1;
+  error=read_row(cursor->c_get(cursor, &last_key, &row, DB_SET_RANGE),
+		 (char*) buf, active_index, &row, (DBT*) 0, 0);
+  key_info->handler.bdb_return_if_eq= 0;
+  bzero((char*) &row,sizeof(row));
+  if (read_row(cursor->c_get(cursor, &last_key, &row, DB_PREV),
+	       (char*) buf, active_index, &row, &last_key, 1) ||
+      berkeley_key_cmp(table, key_info, key_buff2, key_len))
+    error=HA_ERR_KEY_NOT_FOUND;
+  DBUG_RETURN(error);
+}
+
 
 int ha_berkeley::index_next(byte * buf)
 {

sql/ha_berkeley.h

@@ -89,7 +89,7 @@ class ha_berkeley: public handler
     int_option_flag(HA_READ_NEXT | HA_READ_PREV |
 		    HA_REC_NOT_IN_SEQ |
 		    HA_KEYPOS_TO_RNDPOS | HA_READ_ORDER | HA_LASTKEY_ORDER |
-		    HA_LONGLONG_KEYS | HA_NULL_KEY | HA_HAVE_KEY_READ_ONLY |
+		    HA_NULL_KEY | HA_HAVE_KEY_READ_ONLY |
 		    HA_BLOB_KEY | HA_NOT_EXACT_COUNT |
 		    HA_PRIMARY_KEY_IN_READ_INDEX | HA_DROP_BEFORE_CREATE |
 		    HA_AUTO_PART_KEY),
@@ -123,6 +123,7 @@ class ha_berkeley: public handler
 		 uint key_len, enum ha_rkey_function find_flag);
   int index_read_idx(byte * buf, uint index, const byte * key,
 		     uint key_len, enum ha_rkey_function find_flag);
+  int index_read_last(byte * buf, const byte * key, uint key_len);
   int index_next(byte * buf);
   int index_next_same(byte * buf, const byte *key, uint keylen);
   int index_prev(byte * buf);

sql/ha_heap.cc

@@ -57,6 +57,7 @@ int ha_heap::open(const char *name, int mode, uint test_if_locked)
     for (part=0 ; part < pos->key_parts ; part++)
     {
       uint flag=pos->key_part[part].key_type;
+      Field *field=pos->key_part[part].field;
       if (!f_is_packed(flag) &&
 	  f_packtype(flag) == (int) FIELD_TYPE_DECIMAL &&
 	  !(flag & FIELDFLAG_BINARY))
@@ -65,7 +66,17 @@ int ha_heap::open(const char *name, int mode, uint test_if_locked)
 	seg->type= (int) HA_KEYTYPE_BINARY;
       seg->start=(uint) pos->key_part[part].offset;
       seg->length=(uint) pos->key_part[part].length;
-      seg++;
+      if (field->null_ptr)
+      {
+	seg->null_bit=field->null_bit;
+	seg->null_pos= (uint) (field->null_ptr-
+			       (uchar*) table->record[0]);
+      }
+      else
+      {
+	seg->null_bit=0;
+	seg->null_pos=0;
+      }
     }
   }
   mem_per_row += MY_ALIGN(table->reclength+1, sizeof(char*));

sql/ha_heap.h

@@ -36,7 +36,7 @@ class ha_heap: public handler
   ulong option_flag() const
   { return (HA_READ_RND_SAME | HA_NO_INDEX | HA_ONLY_WHOLE_INDEX |
 	    HA_WRONG_ASCII_ORDER | HA_KEYPOS_TO_RNDPOS | HA_NO_BLOBS |
-	    HA_REC_NOT_IN_SEQ); }
+	    HA_NULL_KEY | HA_REC_NOT_IN_SEQ | HA_NOT_READ_PREFIX_LAST); }
   uint max_record_length() const { return HA_MAX_REC_LENGTH; }
   uint max_keys()          const { return MAX_KEY; }
   uint max_key_parts()     const { return MAX_REF_PARTS; }

sql/ha_innobase.cc → sql/ha_innodb.cc

@@ -35,7 +35,7 @@ InnoDB */
 
 #define MAX_ULONG_BIT ((ulong) 1 << (sizeof(ulong)*8-1))
 
-#include "ha_innobase.h"
+#include "ha_innodb.h"
 
 /* We must declare this here because we undef SAFE_MUTEX below */
 pthread_mutex_t innobase_mutex;
@@ -2008,6 +2008,24 @@ ha_innobase::index_read(
 	DBUG_RETURN(error);
 }
 
+
+/*
+  The following functions works like index_read, but it find the last
+  row with the current index prefix.
+  This code is disabled until Heikki has verified that InnoDB support the
+  HA_READ_PREFIX_LAST flag and removed the HA_NOT_READ_PREFIX_LAST
+  flag from ha_innodb.h
+*/
+
+int
+ha_innobase::index_read_last(mysql_byte *buf,
+			     const mysql_byte *key_ptr,
+			     uint key_len)
+{
+  return index_read(buf, key_ptr, key_len, HA_READ_PREFIX_LAST);
+}
+
+
 /************************************************************************
 Changes the active index of a handle. */
 

sql/ha_innobase.h → sql/ha_innodb.h

-/* Copyright (C) 2000 MySQL AB & MySQL Finland AB & TCX DataKonsult AB
-   && Innobase Oy
-
-   -This file is modified from ha_berkeley.h of MySQL distribution-
+/* Copyright (C) 2000 MySQL AB && Innobase Oy
 
    This program is free software; you can redistribute it and/or modify
    it under the terms of the GNU General Public License as published by
@@ -17,13 +14,17 @@
    along with this program; if not, write to the Free Software
    Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA */
 
+/*
+  This file is based on ha_berkeley.h of MySQL distribution
+
+  This file defines the Innodb handler: the interface between MySQL and
+  Innodb
+*/
+
 #ifdef __GNUC__
 #pragma interface			/* gcc class implementation */
 #endif
 
-/* This file defines the Innobase handler: the interface between MySQL and
-Innobase */
-
 typedef struct st_innobase_share {
   THR_LOCK lock;
   pthread_mutex_t mutex;
@@ -32,11 +33,11 @@ typedef struct st_innobase_share {
 } INNOBASE_SHARE;
 
 
-/* The class defining a handle to an Innobase table */
+/* The class defining a handle to an Innodb table */
 class ha_innobase: public handler
 {
 	void*	innobase_prebuilt;	/* (row_prebuilt_t*) prebuilt
-					struct in Innobase, used to save
+					struct in Innodb, used to save
 					CPU */
 	THD*		user_thd;	/* the thread handle of the user
 					currently using the handle; this is
@@ -50,7 +51,7 @@ class ha_innobase: public handler
   	byte*		upd_buff;	/* buffer used in updates */
   	byte*		key_val_buff;	/* buffer used in converting
   					search key values from MySQL format
-  					to Innobase format */
+  					to Innodb format */
 	uint		ref_stored_len;	/* length of the key value stored to
 					'ref' buffer of the handle, if any */
   	ulong 		int_option_flag;
@@ -78,11 +79,11 @@ class ha_innobase: public handler
 			  HA_REC_NOT_IN_SEQ |
 			  HA_KEYPOS_TO_RNDPOS | HA_LASTKEY_ORDER |
 			  HA_HAVE_KEY_READ_ONLY | HA_READ_NOT_EXACT_KEY |
-			  HA_LONGLONG_KEYS | HA_NULL_KEY |
+			  HA_NULL_KEY |
 			  HA_NOT_EXACT_COUNT |
 			  HA_NO_WRITE_DELAYED |
 			  HA_PRIMARY_KEY_IN_READ_INDEX |
-			  HA_DROP_BEFORE_CREATE |
+			  HA_DROP_BEFORE_CREATE | HA_NOT_READ_PREFIX_LAST |
 			  HA_NO_PREFIX_CHAR_KEYS),
 	  last_dup_key((uint) -1),
 	  start_of_scan(0)
@@ -125,6 +126,7 @@ class ha_innobase: public handler
 		       uint key_len, enum ha_rkey_function find_flag);
   	int index_read_idx(byte * buf, uint index, const byte * key,
 			   uint key_len, enum ha_rkey_function find_flag);
+	int index_read_last(byte * buf, const byte * key, uint key_len);
   	int index_next(byte * buf);
   	int index_next_same(byte * buf, const byte *key, uint keylen);
   	int index_prev(byte * buf);

sql/ha_isam.cc

@@ -109,6 +109,15 @@ int ha_isam::index_read_idx(byte * buf, uint index, const byte * key,
   return !error ? 0 : my_errno ? my_errno : -1;
 }
 
+int ha_isam::index_read_last(byte * buf, const byte * key, uint key_len)
+{
+  statistic_increment(ha_read_key_count,&LOCK_status);
+  int error=nisam_rkey(file, buf, active_index, key, key_len,
+		       HA_READ_PREFIX_LAST);
+  table->status=error ? STATUS_NOT_FOUND: 0;
+  return !error ? 0 : my_errno ? my_errno : -1;
+}
+
 int ha_isam::index_next(byte * buf)
 {
   statistic_increment(ha_read_next_count,&LOCK_status);

sql/ha_isam.h

@@ -33,7 +33,7 @@ class ha_isam: public handler
     int_option_flag(HA_READ_NEXT | HA_READ_PREV | HA_READ_RND_SAME |
 		    HA_KEYPOS_TO_RNDPOS | HA_READ_ORDER | HA_LASTKEY_ORDER |
 		    HA_HAVE_KEY_READ_ONLY | HA_READ_NOT_EXACT_KEY |
-		    HA_LONGLONG_KEYS | HA_KEY_READ_WRONG_STR | HA_DUPP_POS |
+		    HA_KEY_READ_WRONG_STR | HA_DUPP_POS |
 		    HA_NOT_DELETE_WITH_CACHE)
     {}
   ~ha_isam() {}
@@ -57,6 +57,7 @@ class ha_isam: public handler
 		 uint key_len, enum ha_rkey_function find_flag);
   int index_read_idx(byte * buf, uint idx, const byte * key,
 		     uint key_len, enum ha_rkey_function find_flag);
+  int index_read_last(byte * buf, const byte * key, uint key_len);
   int index_next(byte * buf);
   int index_prev(byte * buf);
   int index_first(byte * buf);

sql/ha_isammrg.h

@@ -32,8 +32,9 @@ class ha_isammrg: public handler
   ~ha_isammrg() {}
   const char *table_type() const { return "MRG_ISAM"; }
   const char **bas_ext() const;
-  ulong option_flag() const { return HA_READ_RND_SAME | HA_KEYPOS_TO_RNDPOS
-                                                      | HA_REC_NOT_IN_SEQ;}
+  ulong option_flag() const { return (HA_READ_RND_SAME | HA_KEYPOS_TO_RNDPOS |
+				      HA_NOT_READ_PREFIX_LAST |
+				      HA_REC_NOT_IN_SEQ); }
   uint max_record_length() const { return HA_MAX_REC_LENGTH; }
   uint max_keys()          const { return 0; }
   uint max_key_parts()     const { return 0; }

sql/ha_myisam.cc

@@ -777,6 +777,14 @@ int ha_myisam::index_read_idx(byte * buf, uint index, const byte * key,
   return error;
 }
 
+int ha_myisam::index_read_last(byte * buf, const byte * key, uint key_len)
+{
+  statistic_increment(ha_read_key_count,&LOCK_status);
+  int error=mi_rkey(file,buf,active_index, key, key_len, HA_READ_PREFIX_LAST);
+  table->status=error ? STATUS_NOT_FOUND: 0;
+  return error;
+}
+
 int ha_myisam::index_next(byte * buf)
 {
   statistic_increment(ha_read_next_count,&LOCK_status);
@@ -973,7 +981,7 @@ void ha_myisam::update_create_info(HA_CREATE_INFO *create_info)
 }
 
 
-int ha_myisam::create(const char *name, register TABLE *form,
+int ha_myisam::create(const char *name, register TABLE *table,
 		      HA_CREATE_INFO *info)
 {
   int error;
@@ -985,20 +993,20 @@ int ha_myisam::create(const char *name, register TABLE *form,
   MI_KEYDEF *keydef;
   MI_COLUMNDEF *recinfo,*recinfo_pos;
   MI_KEYSEG *keyseg;
-  uint options=form->db_options_in_use;
+  uint options=table->db_options_in_use;
   DBUG_ENTER("ha_myisam::create");
 
   type=HA_KEYTYPE_BINARY;				// Keep compiler happy
   if (!(my_multi_malloc(MYF(MY_WME),
-			&recinfo,(form->fields*2+2)*sizeof(MI_COLUMNDEF),
-			&keydef, form->keys*sizeof(MI_KEYDEF),
+			&recinfo,(table->fields*2+2)*sizeof(MI_COLUMNDEF),
+			&keydef, table->keys*sizeof(MI_KEYDEF),
 			&keyseg,
-			((form->key_parts + form->keys) * sizeof(MI_KEYSEG)),
+			((table->key_parts + table->keys) * sizeof(MI_KEYSEG)),
 			0)))
     DBUG_RETURN(1);
 
-  pos=form->key_info;
-  for (i=0; i < form->keys ; i++, pos++)
+  pos=table->key_info;
+  for (i=0; i < table->keys ; i++, pos++)
   {
     keydef[i].flag= (pos->flags & (HA_NOSAME | HA_FULLTEXT));
     keydef[i].seg=keyseg;
@@ -1041,7 +1049,7 @@ int ha_myisam::create(const char *name, register TABLE *form,
       {
 	keydef[i].seg[j].null_bit=field->null_bit;
 	keydef[i].seg[j].null_pos= (uint) (field->null_ptr-
-					   (uchar*) form->record[0]);
+					   (uchar*) table->record[0]);
       }
       else
       {
@@ -1059,19 +1067,19 @@ int ha_myisam::create(const char *name, register TABLE *form,
 	keydef[i].seg[j].flag|=HA_BLOB_PART;
 	/* save number of bytes used to pack length */
 	keydef[i].seg[j].bit_start= (uint) (field->pack_length() -
-					    form->blob_ptr_size);
+					    table->blob_ptr_size);
       }
     }
     keyseg+=pos->key_parts;
   }
 
   recpos=0; recinfo_pos=recinfo;
-  while (recpos < (uint) form->reclength)
+  while (recpos < (uint) table->reclength)
   {
     Field **field,*found=0;
-    minpos=form->reclength; length=0;
+    minpos=table->reclength; length=0;
 
-    for (field=form->field ; *field ; field++)
+    for (field=table->field ; *field ; field++)
     {
       if ((fieldpos=(*field)->offset()) >= recpos &&
 	  fieldpos <= minpos)
@@ -1117,7 +1125,7 @@ int ha_myisam::create(const char *name, register TABLE *form,
     {
       recinfo_pos->null_bit=found->null_bit;
       recinfo_pos->null_pos= (uint) (found->null_ptr-
-				  (uchar*) form->record[0]);
+				  (uchar*) table->record[0]);
     }
     else
     {
@@ -1132,20 +1140,23 @@ int ha_myisam::create(const char *name, register TABLE *form,
   }
   MI_CREATE_INFO create_info;
   bzero((char*) &create_info,sizeof(create_info));
-  create_info.max_rows=form->max_rows;
-  create_info.reloc_rows=form->min_rows;
+  create_info.max_rows=table->max_rows;
+  create_info.reloc_rows=table->min_rows;
   create_info.auto_increment=(info->auto_increment_value ?
 			      info->auto_increment_value -1 :
 			      (ulonglong) 0);
-  create_info.data_file_length=(ulonglong) form->max_rows*form->avg_row_length;
+  create_info.data_file_length= ((ulonglong) table->max_rows *
+				 table->avg_row_length);
   create_info.raid_type=info->raid_type;
-  create_info.raid_chunks=info->raid_chunks ? info->raid_chunks : RAID_DEFAULT_CHUNKS;
-  create_info.raid_chunksize=info->raid_chunksize ? info->raid_chunksize : RAID_DEFAULT_CHUNKSIZE;
+  create_info.raid_chunks= (info->raid_chunks ? info->raid_chunks :
+			    RAID_DEFAULT_CHUNKS);
+  create_info.raid_chunksize=(info->raid_chunksize ? info->raid_chunksize :
+			      RAID_DEFAULT_CHUNKSIZE);
   create_info.data_file_name= info->data_file_name;
   create_info.index_file_name=info->index_file_name;
 
   error=mi_create(fn_format(buff,name,"","",2+4),
-		  form->keys,keydef,
+		  table->keys,keydef,
 		  (uint) (recinfo_pos-recinfo), recinfo,
 		  0, (MI_UNIQUEDEF*) 0,
 		  &create_info,

sql/ha_myisam.h

@@ -47,7 +47,7 @@ class ha_myisam: public handler
     int_option_flag(HA_READ_NEXT | HA_READ_PREV | HA_READ_RND_SAME |
 		    HA_KEYPOS_TO_RNDPOS | HA_READ_ORDER | HA_LASTKEY_ORDER |
 		    HA_HAVE_KEY_READ_ONLY | HA_READ_NOT_EXACT_KEY |
-		    HA_LONGLONG_KEYS | HA_NULL_KEY |
+		    HA_NULL_KEY |
                     HA_CAN_FULLTEXT | HA_CAN_SQL_HANDLER |
 		    HA_DUPP_POS | HA_BLOB_KEY | HA_AUTO_PART_KEY),
     enable_activate_all_index(1)
@@ -71,6 +71,7 @@ class ha_myisam: public handler
 		 uint key_len, enum ha_rkey_function find_flag);
   int index_read_idx(byte * buf, uint idx, const byte * key,
 		     uint key_len, enum ha_rkey_function find_flag);
+  int index_read_last(byte * buf, const byte * key, uint key_len);
   int index_next(byte * buf);
   int index_prev(byte * buf);
   int index_first(byte * buf);
@@ -78,8 +79,14 @@ class ha_myisam: public handler
   int index_next_same(byte *buf, const byte *key, uint keylen);
   int index_end() { ft_handler=NULL; return 0; }
   int ft_init()
-         { if(!ft_handler) return 1; ft_handler->please->reinit_search(ft_handler); return 0; }
-  FT_INFO *ft_init_ext(uint mode, uint inx,const byte *key, uint keylen, bool presort)
+  {
+    if (!ft_handler)
+      return 1;
+    ft_handler->please->reinit_search(ft_handler);
+    return 0;
+  }
+  FT_INFO *ft_init_ext(uint mode, uint inx,const byte *key, uint keylen,
+		       bool presort)
   { return ft_init_search(mode, file,inx,(byte*) key,keylen,presort); }
   int ft_read(byte *buf);
   int rnd_init(bool scan=1);

sql/ha_myisammrg.cc

@@ -112,6 +112,15 @@ int ha_myisammrg::index_read_idx(byte * buf, uint index, const byte * key,
   return error;
 }
 
+int ha_myisammrg::index_read_last(byte * buf, const byte * key, uint key_len)
+{
+  statistic_increment(ha_read_key_count,&LOCK_status);
+  int error=myrg_rkey(file,buf,active_index, key, key_len,
+		      HA_READ_PREFIX_LAST);
+  table->status=error ? STATUS_NOT_FOUND: 0;
+  return error;
+}
+
 int ha_myisammrg::index_next(byte * buf)
 {
   statistic_increment(ha_read_next_count,&LOCK_status);

sql/ha_myisammrg.h

@@ -38,7 +38,7 @@ class ha_myisammrg: public handler
 	      HA_HAVE_KEY_READ_ONLY |
 	      HA_KEYPOS_TO_RNDPOS | HA_READ_ORDER |
 	      HA_LASTKEY_ORDER | HA_READ_NOT_EXACT_KEY |
-	      HA_LONGLONG_KEYS | HA_NULL_KEY | HA_BLOB_KEY); }
+	      HA_NULL_KEY | HA_BLOB_KEY); }
   uint max_record_length() const { return HA_MAX_REC_LENGTH; }
   uint max_keys()          const { return MI_MAX_KEY; }
   uint max_key_parts()     const { return MAX_REF_PARTS; }
@@ -55,6 +55,7 @@ class ha_myisammrg: public handler
 		 uint key_len, enum ha_rkey_function find_flag);
   int index_read_idx(byte * buf, uint idx, const byte * key,
 		     uint key_len, enum ha_rkey_function find_flag);
+  int index_read_last(byte * buf, const byte * key, uint key_len);
   int index_next(byte * buf);
   int index_prev(byte * buf);
   int index_first(byte * buf);

sql/handler.cc

@@ -33,7 +33,7 @@
 #include "ha_berkeley.h"
 #endif
 #ifdef HAVE_INNOBASE_DB
-#include "ha_innobase.h"
+#include "ha_innodb.h"
 #endif
 #include <myisampack.h>
 #include <errno.h>

sql/handler.h

@@ -55,12 +55,11 @@
 #define HA_REC_NOT_IN_SEQ	64	/* ha_info don't return recnumber;
 					   It returns a position to ha_r_rnd */
 #define HA_ONLY_WHOLE_INDEX	128	/* Can't use part key searches */
-#define HA_RSAME_NO_INDEX	256	/* RSAME can't restore index */
+#define HA_NOT_READ_PREFIX_LAST	256	/* RSAME can't restore index */
 #define HA_WRONG_ASCII_ORDER	512	/* Can't use sorting through key */
 #define HA_HAVE_KEY_READ_ONLY	1024	/* Can read only keys (no record) */
 #define HA_READ_NOT_EXACT_KEY	2048	/* Can read record after/before key */
 #define HA_NO_INDEX		4096	/* No index needed for next/prev */
-#define HA_LONGLONG_KEYS	8192	/* Can have longlong as key */
 #define HA_KEY_READ_WRONG_STR	16384	/* keyread returns converted strings */
 #define HA_NULL_KEY		32768	/* One can have keys with NULL */
 #define HA_DUPP_POS		65536	/* ha_position() gives dupp row */
@@ -256,6 +255,10 @@ public:
   virtual int index_first(byte * buf)=0;
   virtual int index_last(byte * buf)=0;
   virtual int index_next_same(byte *buf, const byte *key, uint keylen);
+  virtual int index_read_last(byte * buf, const byte * key, uint key_len)
+  {
+    return (my_errno=HA_ERR_WRONG_COMMAND);
+  }
   virtual int ft_init()
     { return -1; }
   virtual FT_INFO *ft_init_ext(uint mode,uint inx,const byte *key, uint keylen,

sql/mysqld.cc

@@ -27,7 +27,7 @@
 #include "ha_berkeley.h"
 #endif
 #ifdef HAVE_INNOBASE_DB
-#include "ha_innobase.h"
+#include "ha_innodb.h"
 #endif
 #include "ha_myisam.h"
 #include <nisam.h>

sql/sql_delete.cc

@@ -24,7 +24,7 @@
 
 
 #include "mysql_priv.h"
-#include "ha_innobase.h"
+#include "ha_innodb.h"
 #include "sql_select.h"
 
 int mysql_delete(THD *thd, TABLE_LIST *table_list, COND *conds, ORDER *order,

sql/sql_select.cc

@@ -89,15 +89,18 @@ static int join_read_system(JOIN_TAB *tab);
 static int join_read_const(JOIN_TAB *tab);
 static int join_read_key(JOIN_TAB *tab);
 static int join_read_always_key(JOIN_TAB *tab);
+static int join_read_last_key(JOIN_TAB *tab);
 static int join_no_more_records(READ_RECORD *info);
 static int join_read_next(READ_RECORD *info);
 static int join_init_quick_read_record(JOIN_TAB *tab);
 static int test_if_quick_select(JOIN_TAB *tab);
 static int join_init_read_record(JOIN_TAB *tab);
-static int join_init_read_first_with_key(JOIN_TAB *tab);
-static int join_init_read_next_with_key(READ_RECORD *info);
-static int join_init_read_last_with_key(JOIN_TAB *tab);
-static int join_init_read_prev_with_key(READ_RECORD *info);
+static int join_read_first(JOIN_TAB *tab);
+static int join_read_next(READ_RECORD *info);
+static int join_read_next_same(READ_RECORD *info);
+static int join_read_last(JOIN_TAB *tab);
+static int join_read_prev_same(READ_RECORD *info);
+static int join_read_prev(READ_RECORD *info);
 static int join_ft_read_first(JOIN_TAB *tab);
 static int join_ft_read_next(READ_RECORD *info);
 static COND *make_cond_for_table(COND *cond,table_map table,
@@ -2510,7 +2513,7 @@ make_join_readinfo(JOIN *join,uint options)
       tab->quick=0;
       table->file->index_init(tab->ref.key);
       tab->read_first_record= join_read_always_key;
-      tab->read_record.read_record= join_read_next;
+      tab->read_record.read_record= join_read_next_same;
       if (table->used_keys & ((key_map) 1 << tab->ref.key) &&
 	  !table->no_keyread)
       {
@@ -2585,7 +2588,7 @@ make_join_readinfo(JOIN *join,uint options)
 	  {					// Only read index tree
 	    tab->index=find_shortest_key(table, table->used_keys);
 	    tab->table->file->index_init(tab->index);
-	    tab->read_first_record= join_init_read_first_with_key;
+	    tab->read_first_record= join_read_first;
 	    tab->type=JT_NEXT;		// Read with index_first / index_next
 	  }
 	}
@@ -3641,6 +3644,10 @@ create_tmp_table(THD *thd,TMP_TABLE_PARAM *param,List<Item> &fields,
     {
       if (field->flags & GROUP_FLAG && !using_unique_constraint)
       {
+	/*
+	  We have to reserve one byte here for NULL bits,
+	  as this is updated by 'end_update()'
+	*/
 	*pos++=0;				// Null is stored here
 	recinfo->length=1;
 	recinfo->type=FIELD_NORMAL;
@@ -3730,11 +3737,11 @@ create_tmp_table(THD *thd,TMP_TABLE_PARAM *param,List<Item> &fields,
 	{
 	  /*
 	    To be able to group on NULL, we move the null bit to be
-	     just before the column and extend the key to cover the null bit
+	    just before the column.
+	    The null byte is updated by 'end_update()' 
 	  */
-	  *group_buff= 0;			// Init null byte
-	  key_part_info->offset--;
-	  key_part_info->length++;
+	  key_part_info->null_bit=1;
+	  key_part_info->null_offset= key_part_info->offset-1;
 	  group->field->move_field((char*) group_buff+1, (uchar*) group_buff,
 				   1);
 	}
@@ -4497,6 +4504,35 @@ join_read_always_key(JOIN_TAB *tab)
   return 0;
 }
 
+/*
+  This function is used when optimizing away ORDER BY in 
+  SELECT * FROM t1 WHERE a=1 ORDER BY a DESC,b DESC
+*/
+  
+static int
+join_read_last_key(JOIN_TAB *tab)
+{
+  int error;
+  TABLE *table= tab->table;
+
+  if (cp_buffer_from_ref(&tab->ref))
+    return -1;
+  if ((error=table->file->index_read_last(table->record[0],
+					  tab->ref.key_buff,
+					  tab->ref.key_length)))
+  {
+    if (error != HA_ERR_KEY_NOT_FOUND)
+    {
+      sql_print_error("read_const: Got error %d when reading table %s",error,
+		      table->path);
+      table->file->print_error(error,MYF(0));
+      return 1;
+    }
+    return -1; /* purecov: inspected */
+  }
+  return 0;
+}
+
 
 	/* ARGSUSED */
 static int
@@ -4507,7 +4543,7 @@ join_no_more_records(READ_RECORD *info __attribute__((unused)))
 
 
 static int
-join_read_next(READ_RECORD *info)
+join_read_next_same(READ_RECORD *info)
 {
   int error;
   TABLE *table= info->table;
@@ -4530,6 +4566,37 @@ join_read_next(READ_RECORD *info)
   return 0;
 }
 
+static int
+join_read_prev_same(READ_RECORD *info)
+{
+  int error;
+  TABLE *table= info->table;
+  JOIN_TAB *tab=table->reginfo.join_tab;
+
+  if ((error=table->file->index_prev(table->record[0])))
+  {
+    if (error != HA_ERR_END_OF_FILE)
+    {
+      sql_print_error("read_next: Got error %d when reading table %s",error,
+		      table->path);
+      table->file->print_error(error,MYF(0));
+      error= 1;
+    }
+    else
+    {
+      table->status= STATUS_GARBAGE;
+      error= -1;
+    }
+  }
+  else if (key_cmp(table, tab->ref.key_buff, tab->ref.key,
+		   tab->ref.key_length))
+  {
+    table->status=STATUS_NOT_FOUND;
+    error= 1;
+  }
+  return error;
+}
+
 
 static int
 join_init_quick_read_record(JOIN_TAB *tab)
@@ -4560,7 +4627,7 @@ join_init_read_record(JOIN_TAB *tab)
 }
 
 static int
-join_init_read_first_with_key(JOIN_TAB *tab)
+join_read_first(JOIN_TAB *tab)
 {
   int error;
   TABLE *table=tab->table;
@@ -4571,7 +4638,7 @@ join_init_read_first_with_key(JOIN_TAB *tab)
     table->file->extra(HA_EXTRA_KEYREAD);
   }
   tab->table->status=0;
-  tab->read_record.read_record=join_init_read_next_with_key;
+  tab->read_record.read_record=join_read_next;
   tab->read_record.table=table;
   tab->read_record.file=table->file;
   tab->read_record.index=tab->index;
@@ -4591,8 +4658,9 @@ join_init_read_first_with_key(JOIN_TAB *tab)
   return 0;
 }
 
+
 static int
-join_init_read_next_with_key(READ_RECORD *info)
+join_read_next(READ_RECORD *info)
 {
   int error=info->file->index_next(info->record);
   if (error)
@@ -4609,9 +4677,8 @@ join_init_read_next_with_key(READ_RECORD *info)
   return 0;
 }
 
-
 static int
-join_init_read_last_with_key(JOIN_TAB *tab)
+join_read_last(JOIN_TAB *tab)
 {
   TABLE *table=tab->table;
   int error;
@@ -4621,7 +4688,7 @@ join_init_read_last_with_key(JOIN_TAB *tab)
     table->file->extra(HA_EXTRA_KEYREAD);
   }
   tab->table->status=0;
-  tab->read_record.read_record=join_init_read_prev_with_key;
+  tab->read_record.read_record=join_read_prev;
   tab->read_record.table=table;
   tab->read_record.file=table->file;
   tab->read_record.index=tab->index;
@@ -4641,8 +4708,9 @@ join_init_read_last_with_key(JOIN_TAB *tab)
   return 0;
 }
 
+
 static int
-join_init_read_prev_with_key(READ_RECORD *info)
+join_read_prev(READ_RECORD *info)
 {
   int error=info->file->index_prev(info->record);
   if (error)
@@ -4659,6 +4727,7 @@ join_init_read_prev_with_key(READ_RECORD *info)
   return 0;
 }
 
+
 static int
 join_ft_read_first(JOIN_TAB *tab)
 {
@@ -4734,7 +4803,8 @@ end_send(JOIN *join, JOIN_TAB *join_tab __attribute__((unused)),
       if (join->select_options & OPTION_FOUND_ROWS)
       {
 	JOIN_TAB *jt=join->join_tab;
-	if ((join->tables == 1) && !join->tmp_table && !join->sort_and_group && !join->send_group_parts && !join->having && !jt->select_cond )
+	if ((join->tables == 1) && !join->tmp_table && !join->sort_and_group
+	    && !join->send_group_parts && !join->having && !jt->select_cond)
 	{
 	  join->select_options ^= OPTION_FOUND_ROWS;
 	  join->send_records = jt->records;
@@ -5315,6 +5385,9 @@ test_if_skip_sort_order(JOIN_TAB *tab,ORDER *order,ha_rows select_limit,
 
   if (ref_key >= 0)
   {
+    /*
+      We come here when there is a REF key.
+    */
     int order_direction;
     uint used_key_parts;
     /* Check if we get the rows in requested sorted order by using the key */
@@ -5322,11 +5395,11 @@ test_if_skip_sort_order(JOIN_TAB *tab,ORDER *order,ha_rows select_limit,
 	(order_direction = test_if_order_by_key(order,table,ref_key,
 						&used_key_parts)))
     {
-      if (order_direction == -1)
+      if (order_direction == -1)		// If ORDER BY ... DESC
       {
 	if (select && select->quick)
 	{
-	  // ORDER BY ref_key DESC
+	  // ORDER BY range_key DESC
 	  QUICK_SELECT_DESC *tmp=new QUICK_SELECT_DESC(select->quick,
 						       used_key_parts);
 	  if (!tmp || tmp->error)
@@ -5341,11 +5414,15 @@ test_if_skip_sort_order(JOIN_TAB *tab,ORDER *order,ha_rows select_limit,
 	{
 	  /*
 	    SELECT * FROM t1 WHERE a=1 ORDER BY a DESC,b DESC
-	    TODO:
-	    Add a new traversal function to read last matching row and
-	    traverse backwards.
+
+	    Use a traversal function that starts by reading the last row
+	    with key part (A) and then traverse the index backwards.
 	  */
-	  DBUG_RETURN(0);
+	  if (table->file->option_flag() & HA_NOT_READ_PREFIX_LAST)
+	    DBUG_RETURN(1);
+	  tab->read_first_record=       join_read_last_key;
+	  tab->read_record.read_record= join_read_prev_same;
+	  /* fall through */
 	}
       }
       DBUG_RETURN(1);			/* No need to sort */
@@ -5377,8 +5454,8 @@ test_if_skip_sort_order(JOIN_TAB *tab,ORDER *order,ha_rows select_limit,
 	  if (!no_changes)
 	  {
 	    tab->index=nr;
-	    tab->read_first_record=  (flag > 0 ? join_init_read_first_with_key:
-				      join_init_read_last_with_key);
+	    tab->read_first_record=  (flag > 0 ? join_read_first:
+				      join_read_last);
 	    table->file->index_init(nr);
 	    tab->type=JT_NEXT;	// Read with index_first(), index_next()
 	    if (table->used_keys & ((key_map) 1 << nr))
@@ -6369,7 +6446,8 @@ get_sort_by_table(ORDER *a,ORDER *b,TABLE_LIST *tables)
 static void
 calc_group_buffer(JOIN *join,ORDER *group)
 {
-  uint key_length=0,parts=0;
+  uint key_length=0, parts=0, null_parts=0;
+
   if (group)
     join->group= 1;
   for (; group ; group=group->next)
@@ -6390,10 +6468,11 @@ calc_group_buffer(JOIN *join,ORDER *group)
       key_length+=(*group->item)->max_length;
     parts++;
     if ((*group->item)->maybe_null)
-      key_length++;
+      null_parts++;
   }
-  join->tmp_table_param.group_length=key_length;
+  join->tmp_table_param.group_length=key_length+null_parts;
   join->tmp_table_param.group_parts=parts;
+  join->tmp_table_param.group_null_parts=null_parts;
 }
 
 

sql/sql_select.h

@@ -127,12 +127,13 @@ class TMP_TABLE_PARAM {
   ha_rows end_write_records;
   uint	field_count,sum_func_count,func_count;
   uint  hidden_field_count;
-  uint	group_parts,group_length;
+  uint	group_parts,group_length,group_null_parts;
   uint	quick_group;
   bool  using_indirect_summary_function;
 
   TMP_TABLE_PARAM()
-    :copy_funcs_it(copy_funcs), copy_field(0), group_parts(0), group_length(0)
+    :copy_funcs_it(copy_funcs), copy_field(0), group_parts(0),
+    group_length(0), group_null_parts(0)
   {}
   ~TMP_TABLE_PARAM()
   {