BUG#20389: Crash when using index scan in reverse order

mysql-test/r/partition_order.result:
  Changed a test case to handle ordered index scan reverse order as well
mysql-test/t/partition_order.test:
  Changed a test case to handle ordered index scan reverse order as well
sql/ha_myisam.cc:
  More debug info
sql/ha_partition.cc:
  Introduced partition_index_read_last to ensure we use index_read_last in those
  cases towards underlying handler.
  Ensured that index_read with HA_READ_PREFIX_LAST, HA_READ_PREFIX_LAST_OR_PREV and
  HA_READ_BEFORE_KEY uses ordered index scan in reverse order.
sql/ha_partition.h:
  Introduced partition_index_read_last to ensure we use index_read_last in those
  cases towards underlying handler.
  Ensured that index_read with HA_READ_PREFIX_LAST, HA_READ_PREFIX_LAST_OR_PREV and
  HA_READ_BEFORE_KEY uses ordered index scan in reverse order.

mysql-test/r/partition_order.result

@@ -718,7 +718,11 @@ partitions 2
 partition x2 values less than (100));
 INSERT into t1 values (1, 1);
 INSERT into t1 values (5, NULL);
-INSERT into t1 values (2, 5);
+INSERT into t1 values (2, 4);
+INSERT into t1 values (3, 3);
+INSERT into t1 values (4, 5);
+INSERT into t1 values (7, 1);
+INSERT into t1 values (6, 6);
 INSERT into t1 values (30, 4);
 INSERT into t1 values (35, 2);
 INSERT into t1 values (40, NULL);
@@ -727,7 +731,55 @@ a	b
 5	NULL
 40	NULL
 1	1
+7	1
 35	2
+3	3
+2	4
 30	4
-2	5
+4	5
+6	6
+select * from t1 force index (b) where b < 10 ORDER BY b;
+a	b
+1	1
+7	1
+35	2
+3	3
+2	4
+30	4
+4	5
+6	6
+select * from t1 force index (b) where b < 10 ORDER BY b DESC;
+a	b
+6	6
+4	5
+2	4
+30	4
+3	3
+35	2
+7	1
+1	1
+drop table t1;
+create table t1 (a int not null, b int, c varchar(20), key (a,b,c))
+partition by range (b)
+(partition p0 values less than (5),
+partition p1 values less than (10));
+INSERT into t1 values (1,1,'1'),(2,2,'2'),(1,3,'3'),(2,4,'4'),(1,5,'5');
+INSERT into t1 values (2,6,'6'),(1,7,'7'),(2,8,'8'),(1,9,'9');
+INSERT into t1 values (1, NULL, NULL), (2, NULL, '10');
+select * from t1 where a = 1 order by a desc, b desc;
+a	b	c
+1	9	9
+1	7	7
+1	5	5
+1	3	3
+1	1	1
+1	NULL	NULL
+select * from t1 where a = 1 order by b desc;
+a	b	c
+1	9	9
+1	7	7
+1	5	5
+1	3	3
+1	1	1
+1	NULL	NULL
 drop table t1;

mysql-test/t/partition_order.test

@@ -818,11 +818,27 @@ partitions 2
 # Insert a couple of tuples
 INSERT into t1 values (1, 1);
 INSERT into t1 values (5, NULL);
-INSERT into t1 values (2, 5);
+INSERT into t1 values (2, 4);
+INSERT into t1 values (3, 3);
+INSERT into t1 values (4, 5);
+INSERT into t1 values (7, 1);
+INSERT into t1 values (6, 6);
 INSERT into t1 values (30, 4);
 INSERT into t1 values (35, 2);
 INSERT into t1 values (40, NULL);
 
 select * from t1 force index (b) where b < 10 OR b IS NULL order by b;
-
+select * from t1 force index (b) where b < 10 ORDER BY b;
+select * from t1 force index (b) where b < 10 ORDER BY b DESC;
+drop table t1;
+
+create table t1 (a int not null, b int, c varchar(20), key (a,b,c))
+partition by range (b)
+(partition p0 values less than (5),
+ partition p1 values less than (10));
+INSERT into t1 values (1,1,'1'),(2,2,'2'),(1,3,'3'),(2,4,'4'),(1,5,'5');
+INSERT into t1 values (2,6,'6'),(1,7,'7'),(2,8,'8'),(1,9,'9');
+INSERT into t1 values (1, NULL, NULL), (2, NULL, '10');
+select * from t1 where a = 1 order by a desc, b desc;
+select * from t1 where a = 1 order by b desc;
 drop table t1;

sql/ha_myisam.cc

@@ -1202,12 +1202,13 @@ int ha_myisam::index_read_idx(byte * buf, uint index, const byte * key,
 
 int ha_myisam::index_read_last(byte * buf, const byte * key, uint key_len)
 {
+  DBUG_ENTER("ha_myisam::index_read_last");
   DBUG_ASSERT(inited==INDEX);
   statistic_increment(table->in_use->status_var.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;
+  DBUG_RETURN(error);
 }
 
 int ha_myisam::index_next(byte * buf)

sql/ha_partition.cc

@@ -3265,6 +3265,7 @@ int ha_partition::index_read(byte * buf, const byte * key,
   DBUG_ENTER("ha_partition::index_read");
 
   end_range= 0;
+  m_index_scan_type= partition_index_read;
   DBUG_RETURN(common_index_read(buf, key, key_len, find_flag));
 }
 
@@ -3282,18 +3283,24 @@ int ha_partition::common_index_read(byte *buf, const byte *key, uint key_len,
 				    enum ha_rkey_function find_flag)
 {
   int error;
+  bool reverse_order= FALSE;
   DBUG_ENTER("ha_partition::common_index_read");
 
   memcpy((void*)m_start_key.key, key, key_len);
   m_start_key.length= key_len;
   m_start_key.flag= find_flag;
-  m_index_scan_type= partition_index_read;
 
   if ((error= partition_scan_set_up(buf, TRUE)))
   {
     DBUG_RETURN(error);
   }
-
+  if (find_flag == HA_READ_PREFIX_LAST ||
+      find_flag == HA_READ_PREFIX_LAST_OR_PREV ||
+      find_flag == HA_READ_BEFORE_KEY)
+  {
+    reverse_order= TRUE;
+    m_ordered_scan_ongoing= TRUE;
+  }
   if (!m_ordered_scan_ongoing ||
       (find_flag == HA_READ_KEY_EXACT &&
        (key_len >= m_curr_key_info->key_length ||
@@ -3319,7 +3326,7 @@ int ha_partition::common_index_read(byte *buf, const byte *key, uint key_len,
       In all other cases we will use the ordered index scan. This will use
       the partition set created by the get_partition_set method.
     */
-    error= handle_ordered_index_scan(buf);
+    error= handle_ordered_index_scan(buf, reverse_order);
   }
   DBUG_RETURN(error);
 }
@@ -3403,7 +3410,7 @@ int ha_partition::common_first_last(byte *buf)
   if (!m_ordered_scan_ongoing &&
       m_index_scan_type != partition_index_last)
     return handle_unordered_scan_next_partition(buf);
-  return handle_ordered_index_scan(buf);
+  return handle_ordered_index_scan(buf, FALSE);
 }
 
 
@@ -3457,7 +3464,9 @@ int ha_partition::index_read_last(byte *buf, const byte *key, uint keylen)
   DBUG_ENTER("ha_partition::index_read_last");
 
   m_ordered= TRUE;				// Safety measure
-  DBUG_RETURN(index_read(buf, key, keylen, HA_READ_PREFIX_LAST));
+  end_range= 0;
+  m_index_scan_type= partition_index_read_last;
+  DBUG_RETURN(common_index_read(buf, key, keylen, HA_READ_PREFIX_LAST));
 }
 
 
@@ -3597,6 +3606,7 @@ int ha_partition::read_range_first(const key_range *start_key,
   }
   else
   {
+    m_index_scan_type= partition_index_read;
     error= common_index_read(m_rec0,
 			     start_key->key,
 			     start_key->length, start_key->flag);
@@ -3855,12 +3865,11 @@ int ha_partition::handle_unordered_scan_next_partition(byte * buf)
     entries.
 */
 
-int ha_partition::handle_ordered_index_scan(byte *buf)
+int ha_partition::handle_ordered_index_scan(byte *buf, bool reverse_order)
 {
   uint i;
   uint j= 0;
   bool found= FALSE;
-  bool reverse_order= FALSE;
   DBUG_ENTER("ha_partition::handle_ordered_index_scan");
 
   m_top_entry= NO_CURRENT_PART_ID;
@@ -3881,7 +3890,6 @@ int ha_partition::handle_ordered_index_scan(byte *buf)
                               m_start_key.key,
                               m_start_key.length,
                               m_start_key.flag);
-      reverse_order= FALSE;
       break;
     case partition_index_first:
       error= file->index_first(rec_buf_ptr);
@@ -3891,6 +3899,12 @@ int ha_partition::handle_ordered_index_scan(byte *buf)
       error= file->index_last(rec_buf_ptr);
       reverse_order= TRUE;
       break;
+    case partition_index_read_last:
+      error= file->index_read_last(rec_buf_ptr,
+                                   m_start_key.key,
+                                   m_start_key.length);
+      reverse_order= TRUE;
+      break;
     default:
       DBUG_ASSERT(FALSE);
       DBUG_RETURN(HA_ERR_END_OF_FILE);

sql/ha_partition.h

@@ -46,7 +46,8 @@ class ha_partition :public handler
     partition_index_read= 0,
     partition_index_first= 1,
     partition_index_last= 2,
-    partition_no_index_scan= 3
+    partition_index_read_last= 3,
+    partition_no_index_scan= 4
   };
   /* Data for the partition handler */
   int  m_mode;                          // Open mode
@@ -429,7 +430,7 @@ class ha_partition :public handler
       return (queue_buf(part_id) +
               PARTITION_BYTES_IN_POS);
     }
-  int handle_ordered_index_scan(byte * buf);
+  int handle_ordered_index_scan(byte * buf, bool reverse_order);
   int handle_ordered_next(byte * buf, bool next_same);
   int handle_ordered_prev(byte * buf);
   void return_top_record(byte * buf);