Merge bk-internal.mysql.com:/home/bk/mysql-5.1

into  janus.mylan:/usr/home/serg/Abk/mysql-5.1


include/my_global.h:
  Auto merged
sql/ha_ndbcluster.cc:
  Auto merged
sql/ha_partition.cc:
  Auto merged
sql/handler.cc:
  Auto merged
sql/item_subselect.cc:
  Auto merged
sql/log.cc:
  Auto merged
sql/log_event.cc:
  Auto merged
sql/mysql_priv.h:
  Auto merged
sql/opt_range.cc:
  Auto merged
sql/slave.cc:
  Auto merged
sql/slave.h:
  Auto merged
sql/sp.cc:
  Auto merged
sql/sql_acl.cc:
  Auto merged
sql/sql_insert.cc:
  Auto merged
sql/sql_select.cc:
  Auto merged
sql/sql_select.h:
  Auto merged
sql/sql_servers.cc:
  Auto merged
sql/table.cc:
  Auto merged
sql/table.h:
  Auto merged
sql/tztime.cc:
  Auto merged
storage/heap/ha_heap.cc:
  Auto merged
storage/innobase/handler/ha_innodb.cc:
  Auto merged
storage/myisam/ha_myisam.cc:
  Auto merged
storage/myisammrg/ha_myisammrg.h:
  Auto merged

29 files changed, 349 insertions, 261 deletions

diff --git a/sql/event_db_repository.cc b/sql/event_db_repository.cc
index 940930ec4c6..860cb54e27f 100644
--- a/sql/event_db_repository.cc
+++ b/sql/event_db_repository.cc
@@ -288,7 +288,7 @@ Event_db_repository::index_read_for_db_for_i_s(THD *thd, TABLE *schema_table,
   {
     key_copy(key_buf, event_table->record[0], key_info, key_len);
     if (!(ret= event_table->file->index_read(event_table->record[0], key_buf,
-                                             key_len, HA_READ_PREFIX)))
+                                             (ulonglong)1, HA_READ_PREFIX)))
     {
       DBUG_PRINT("info",("Found rows. Let's retrieve them. ret=%d", ret));
       do
@@ -843,8 +843,7 @@ Event_db_repository::find_named_event(THD *thd, LEX_STRING db, LEX_STRING name,
 
   key_copy(key, table->record[0], table->key_info, table->key_info->key_length);
 
-  if (table->file->index_read_idx(table->record[0], 0, key,
-                                  table->key_info->key_length,
+  if (table->file->index_read_idx(table->record[0], 0, key, ~ULL(0),
                                   HA_READ_KEY_EXACT))
   {
     DBUG_PRINT("info", ("Row not found"));
diff --git a/sql/ha_ndbcluster.cc b/sql/ha_ndbcluster.cc
index dfdcffc4e71..bc1c3c8e324 100644
--- a/sql/ha_ndbcluster.cc
+++ b/sql/ha_ndbcluster.cc
@@ -1104,7 +1104,7 @@ int ha_ndbcluster::create_indexes(Ndb *ndb, TABLE *tab)
   KEY* key_info= tab->key_info;
   const char **key_name= tab->s->keynames.type_names;
   DBUG_ENTER("ha_ndbcluster::create_indexes");
-  
+
   for (i= 0; i < tab->s->keys; i++, key_info++, key_name++)
   {
     index_name= *key_name;
@@ -3370,19 +3370,6 @@ int ha_ndbcluster::index_read(byte *buf,
 }
 
 
-int ha_ndbcluster::index_read_idx(byte *buf, uint index_no, 
-                              const byte *key, uint key_len, 
-                              enum ha_rkey_function find_flag)
-{
-  statistic_increment(current_thd->status_var.ha_read_key_count, &LOCK_status);
-  DBUG_ENTER("ha_ndbcluster::index_read_idx");
-  DBUG_PRINT("enter", ("index_no: %u, key_len: %u", index_no, key_len));  
-  close_scan();
-  index_init(index_no, 0);  
-  DBUG_RETURN(index_read(buf, key, key_len, find_flag));
-}
-
-
 int ha_ndbcluster::index_next(byte *buf)
 {
   DBUG_ENTER("ha_ndbcluster::index_next");
@@ -3549,10 +3536,10 @@ int ha_ndbcluster::close_scan()
 
   m_multi_cursor= 0;
   if (!m_active_cursor && !m_multi_cursor)
-    DBUG_RETURN(1);
+    DBUG_RETURN(0);
 
   NdbScanOperation *cursor= m_active_cursor ? m_active_cursor : m_multi_cursor;
-  
+
   if (m_lock_tuple)
   {
     /*
diff --git a/sql/ha_ndbcluster.h b/sql/ha_ndbcluster.h
index 63665fde0f8..6cc0e423f2f 100644
--- a/sql/ha_ndbcluster.h
+++ b/sql/ha_ndbcluster.h
@@ -642,8 +642,6 @@ class ha_ndbcluster: public handler
   int index_end();
   int index_read(byte *buf, const byte *key, 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_next(byte *buf);
   int index_prev(byte *buf);
   int index_first(byte *buf);
diff --git a/sql/ha_partition.cc b/sql/ha_partition.cc
index db0d118c2e0..4ed602be54d 100644
--- a/sql/ha_partition.cc
+++ b/sql/ha_partition.cc
@@ -3336,13 +3336,14 @@ int ha_partition::index_end()
 */
 
 int ha_partition::index_read(byte * buf, const byte * key,
-			     uint key_len, enum ha_rkey_function find_flag)
+			     ulonglong keypart_map,
+                             enum ha_rkey_function find_flag)
 {
   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));
+  DBUG_RETURN(common_index_read(buf, key, keypart_map, find_flag));
 }
 
 
@@ -3355,14 +3356,17 @@ int ha_partition::index_read(byte * buf, const byte * key,
   see index_read for rest
 */
 
-int ha_partition::common_index_read(byte *buf, const byte *key, uint key_len,
+int ha_partition::common_index_read(byte *buf, const byte *key,
+                                    ulonglong keypart_map,
 				    enum ha_rkey_function find_flag)
 {
   int error;
   bool reverse_order= FALSE;
+  uint key_len= calculate_key_len(table, active_index, key, keypart_map);
   DBUG_ENTER("ha_partition::common_index_read");
 
   memcpy((void*)m_start_key.key, key, key_len);
+  m_start_key.keypart_map= keypart_map;
   m_start_key.length= key_len;
   m_start_key.flag= find_flag;
 
@@ -3491,33 +3495,6 @@ int ha_partition::common_first_last(byte *buf)
 
 
 /*
-  Perform index read using index where always only one row is returned
-
-  SYNOPSIS
-    index_read_idx()
-    see index_read for rest of parameters and return values
-
-  DESCRIPTION
-    Positions an index cursor to the index specified in key. Fetches the
-    row if any.  This is only used to read whole keys.
-    TODO: Optimise this code to avoid index_init and index_end
-*/
-
-int ha_partition::index_read_idx(byte * buf, uint index, const byte * key,
-				 uint key_len,
-                                 enum ha_rkey_function find_flag)
-{
-  int res;
-  DBUG_ENTER("ha_partition::index_read_idx");
-
-  index_init(index, 0);
-  res= index_read(buf, key, key_len, find_flag);
-  index_end();
-  DBUG_RETURN(res);
-}
-
-
-/*
   Read last using key
 
   SYNOPSIS
@@ -3535,14 +3512,15 @@ int ha_partition::index_read_idx(byte * buf, uint index, const byte * key,
     Can only be used on indexes supporting HA_READ_ORDER
 */
 
-int ha_partition::index_read_last(byte *buf, const byte *key, uint keylen)
+int ha_partition::index_read_last(byte *buf, const byte *key,
+                                      ulonglong keypart_map)
 {
   DBUG_ENTER("ha_partition::index_read_last");
 
   m_ordered= TRUE;				// Safety measure
   end_range= 0;
   m_index_scan_type= partition_index_read_last;
-  DBUG_RETURN(common_index_read(buf, key, keylen, HA_READ_PREFIX_LAST));
+  DBUG_RETURN(common_index_read(buf, key, keypart_map, HA_READ_PREFIX_LAST));
 }
 
 
@@ -3688,7 +3666,7 @@ int ha_partition::read_range_first(const key_range *start_key,
     m_index_scan_type= partition_index_read;
     error= common_index_read(m_rec0,
 			     start_key->key,
-			     start_key->length, start_key->flag);
+                             start_key->keypart_map, start_key->flag);
   }
   DBUG_RETURN(error);
 }
@@ -3887,7 +3865,7 @@ int ha_partition::handle_unordered_scan_next_partition(byte * buf)
     case partition_index_read:
       DBUG_PRINT("info", ("index_read on partition %d", i));
       error= file->index_read(buf, m_start_key.key,
-                              m_start_key.length,
+                              m_start_key.keypart_map,
                               m_start_key.flag);
       break;
     case partition_index_first:
@@ -3979,7 +3957,7 @@ int ha_partition::handle_ordered_index_scan(byte *buf, bool reverse_order)
     case partition_index_read:
       error= file->index_read(rec_buf_ptr,
                               m_start_key.key,
-                              m_start_key.length,
+                              m_start_key.keypart_map,
                               m_start_key.flag);
       break;
     case partition_index_first:
@@ -3993,7 +3971,7 @@ int ha_partition::handle_ordered_index_scan(byte *buf, bool reverse_order)
     case partition_index_read_last:
       error= file->index_read_last(rec_buf_ptr,
                                    m_start_key.key,
-                                   m_start_key.length);
+                                   m_start_key.keypart_map);
       reverse_order= TRUE;
       break;
     default:
diff --git a/sql/ha_partition.h b/sql/ha_partition.h
index 1069dd058d5..2d43e2b0df2 100644
--- a/sql/ha_partition.h
+++ b/sql/ha_partition.h
@@ -384,9 +384,8 @@ public:
     any end processing needed.
   */
   virtual int index_read(byte * buf, const byte * key,
-			 uint key_len, enum ha_rkey_function find_flag);
-  virtual int index_read_idx(byte * buf, uint idx, const byte * key,
-			     uint key_len, enum ha_rkey_function find_flag);
+                             ulonglong keypart_map,
+                             enum ha_rkey_function find_flag);
   virtual int index_init(uint idx, bool sorted);
   virtual int index_end();
 
@@ -399,7 +398,8 @@ public:
   virtual int index_first(byte * buf);
   virtual int index_last(byte * buf);
   virtual int index_next_same(byte * buf, const byte * key, uint keylen);
-  virtual int index_read_last(byte * buf, const byte * key, uint keylen);
+  virtual int index_read_last(byte * buf, const byte * key,
+                                  ulonglong keypart_map);
 
   /*
     read_first_row is virtual method but is only implemented by
@@ -425,7 +425,7 @@ public:
 
 private:
   int common_index_read(byte * buf, const byte * key,
-			uint key_len, enum ha_rkey_function find_flag);
+                        ulonglong keypart_map, enum ha_rkey_function find_flag);
   int common_first_last(byte * buf);
   int partition_scan_set_up(byte * buf, bool idx_read_flag);
   int handle_unordered_next(byte * buf, bool next_same);
diff --git a/sql/handler.cc b/sql/handler.cc
index 23c3103493e..7d3edf36162 100644
--- a/sql/handler.cc
+++ b/sql/handler.cc
@@ -1843,7 +1843,7 @@ int handler::update_auto_increment()
       nr= compute_next_insert_id(nr-1, variables);
     }
     
-    if (table->s->next_number_key_offset == 0)
+    if (table->s->next_number_keypart == 0)
     {
       /* We must defer the appending until "nr" has been possibly truncated */
       append= TRUE;
@@ -1963,7 +1963,7 @@ void handler::get_auto_increment(ulonglong offset, ulonglong increment,
                                         table->read_set);
   column_bitmaps_signal();
   index_init(table->s->next_number_index, 1);
-  if (!table->s->next_number_key_offset)
+  if (table->s->next_number_keypart == 0)
   {						// Autoincrement at key-start
     error=index_last(table->record[1]);
     /*
@@ -1979,7 +1979,8 @@ void handler::get_auto_increment(ulonglong offset, ulonglong increment,
     key_copy(key, table->record[0],
              table->key_info + table->s->next_number_index,
              table->s->next_number_key_offset);
-    error= index_read(table->record[1], key, table->s->next_number_key_offset,
+    error= index_read(table->record[1], key,
+                      make_prev_keypart_map(table->s->next_number_keypart),
                       HA_READ_PREFIX_LAST);
     /*
       MySQL needs to call us for next row: assume we are inserting ("a",null)
@@ -3079,9 +3080,9 @@ int handler::read_multi_range_first(KEY_MULTI_RANGE **found_range_p,
        multi_range_curr < multi_range_end;
        multi_range_curr++)
   {
-    result= read_range_first(multi_range_curr->start_key.length ?
+    result= read_range_first(multi_range_curr->start_key.keypart_map ?
                              &multi_range_curr->start_key : 0,
-                             multi_range_curr->end_key.length ?
+                             multi_range_curr->end_key.keypart_map ?
                              &multi_range_curr->end_key : 0,
                              test(multi_range_curr->range_flag & EQ_RANGE),
                              multi_range_sorted);
@@ -3146,9 +3147,9 @@ int handler::read_multi_range_next(KEY_MULTI_RANGE **found_range_p)
          multi_range_curr < multi_range_end;
          multi_range_curr++)
     {
-      result= read_range_first(multi_range_curr->start_key.length ?
+      result= read_range_first(multi_range_curr->start_key.keypart_map ?
                                &multi_range_curr->start_key : 0,
-                               multi_range_curr->end_key.length ?
+                               multi_range_curr->end_key.keypart_map ?
                                &multi_range_curr->end_key : 0,
                                test(multi_range_curr->range_flag & EQ_RANGE),
                                multi_range_sorted);
@@ -3207,7 +3208,7 @@ int handler::read_range_first(const key_range *start_key,
   else
     result= index_read(table->record[0],
 		       start_key->key,
-		       start_key->length,
+                       start_key->keypart_map,
 		       start_key->flag);
   if (result)
     DBUG_RETURN((result == HA_ERR_KEY_NOT_FOUND) 
@@ -3279,15 +3280,19 @@ int handler::compare_key(key_range *range)
   return cmp;
 }
 
+
 int handler::index_read_idx(byte * buf, uint index, const byte * key,
-			     uint key_len, enum ha_rkey_function find_flag)
+                             ulonglong keypart_map,
+                             enum ha_rkey_function find_flag)
 {
-  int error= ha_index_init(index, 0);
-  if (!error)
-    error= index_read(buf, key, key_len, find_flag);
+  int error, error1;
+  error= index_init(index, 0);
   if (!error)
-    error= ha_index_end();
-  return error;
+  {
+    error= index_read(buf, key, keypart_map, find_flag);
+    error1= index_end();
+  }
+  return error ?  error : error1;
 }
 
 
diff --git a/sql/handler.h b/sql/handler.h
index 82970cc1ac6..1d02d6a7ee3 100644
--- a/sql/handler.h
+++ b/sql/handler.h
@@ -867,6 +867,18 @@ public:
   {}
 };
 
+uint calculate_key_len(TABLE *, uint, const byte *, ulonglong);
+/*
+  bitmap with first N+1 bits set
+  (keypart_map for a key prefix of [0..N] keyparts)
+*/
+#define make_keypart_map(N) (((ulonglong)2 << (N)) - 1)
+/*
+  bitmap with first N bits set
+  (keypart_map for a key prefix of [0..N-1] keyparts)
+*/
+#define make_prev_keypart_map(N) (((ulonglong)1 << (N)) - 1)
+
 /*
   The handler class is the interface for dynamically loadable
   storage engines. Do not add ifdefs and take care when adding or
@@ -1202,11 +1214,20 @@ public:
     DBUG_ASSERT(FALSE);
     return HA_ERR_WRONG_COMMAND;
   }
-  virtual int index_read(byte * buf, const byte * key,
-			 uint key_len, enum ha_rkey_function find_flag)
+  private:
+  virtual int index_read(byte * buf, const byte * key, uint key_len,
+                         enum ha_rkey_function find_flag)
    { return  HA_ERR_WRONG_COMMAND; }
+  public:
+  virtual int index_read(byte * buf, const byte * key, ulonglong keypart_map,
+                         enum ha_rkey_function find_flag)
+   {
+     uint key_len= calculate_key_len(table, active_index, key, keypart_map);
+     return  index_read(buf, key, key_len, find_flag);
+   }
   virtual int index_read_idx(byte * buf, uint index, const byte * key,
-			     uint key_len, enum ha_rkey_function find_flag);
+                             ulonglong keypart_map,
+                             enum ha_rkey_function find_flag);
   virtual int index_next(byte * buf)
    { return  HA_ERR_WRONG_COMMAND; }
   virtual int index_prev(byte * buf)
@@ -1216,8 +1237,16 @@ public:
   virtual int index_last(byte * buf)
    { return  HA_ERR_WRONG_COMMAND; }
   virtual int index_next_same(byte *buf, const byte *key, uint keylen);
+  private:
   virtual int index_read_last(byte * buf, const byte * key, uint key_len)
    { return (my_errno=HA_ERR_WRONG_COMMAND); }
+  public:
+  virtual int index_read_last(byte * buf, const byte * key,
+                              ulonglong keypart_map)
+   {
+     uint key_len= calculate_key_len(table, active_index, key, keypart_map);
+     return  index_read_last(buf, key, key_len);
+   }
   virtual int read_multi_range_first(KEY_MULTI_RANGE **found_range_p,
                                      KEY_MULTI_RANGE *ranges, uint range_count,
                                      bool sorted, HANDLER_BUFFER *buffer);
@@ -1243,8 +1272,7 @@ public:
     { return HA_ERR_WRONG_COMMAND; }
   virtual int rnd_same(byte *buf, uint inx)
     { return HA_ERR_WRONG_COMMAND; }
-  virtual ha_rows records_in_range(uint inx, key_range *min_key,
-                                   key_range *max_key)
+  virtual ha_rows records_in_range(uint inx, key_range *min_key, key_range *max_key)
     { return (ha_rows) 10; }
   virtual void position(const byte *record)=0;
   virtual int info(uint)=0; // see my_base.h for full description
diff --git a/sql/item_subselect.cc b/sql/item_subselect.cc
index 135ed33feec..d9141cc0740 100644
--- a/sql/item_subselect.cc
+++ b/sql/item_subselect.cc
@@ -2045,7 +2045,8 @@ int subselect_uniquesubquery_engine::exec()
     table->file->ha_index_init(tab->ref.key, 0);
   error= table->file->index_read(table->record[0],
                                  tab->ref.key_buff,
-                                 tab->ref.key_length,HA_READ_KEY_EXACT);
+                                 tab_to_keypart_map(tab),
+                                 HA_READ_KEY_EXACT);
   if (error &&
       error != HA_ERR_KEY_NOT_FOUND && error != HA_ERR_END_OF_FILE)
     error= report_error(table, error);
@@ -2154,7 +2155,8 @@ int subselect_indexsubquery_engine::exec()
     table->file->ha_index_init(tab->ref.key, 1);
   error= table->file->index_read(table->record[0],
                                  tab->ref.key_buff,
-                                 tab->ref.key_length,HA_READ_KEY_EXACT);
+                                 tab_to_keypart_map(tab),
+                                 HA_READ_KEY_EXACT);
   if (error &&
       error != HA_ERR_KEY_NOT_FOUND && error != HA_ERR_END_OF_FILE)
     error= report_error(table, error);
diff --git a/sql/key.cc b/sql/key.cc
index bd614b10a70..faa7bf1f04b 100644
--- a/sql/key.cc
+++ b/sql/key.cc
@@ -29,6 +29,7 @@
     field		Field to search after
     key_length		On partial match, contains length of fields before
 			field
+    keypart             key part # of a field
 
   NOTES
    Used when calculating key for NEXT_NUMBER
@@ -45,7 +46,7 @@
 */
 
 int find_ref_key(KEY *key, uint key_count, byte *record, Field *field,
-                 uint *key_length)
+                 uint *key_length, uint *keypart)
 {
   reg2 int i;
   reg3 KEY *key_info;
@@ -60,8 +61,8 @@ int find_ref_key(KEY *key, uint key_count, byte *record, Field *field,
   {
     if (key_info->key_part[0].offset == fieldpos)
     {                                  		/* Found key. Calc keylength */
-      *key_length=0;
-      return(i);                                /* Use this key */
+      *key_length= *keypart= 0;
+      return i;                                 /* Use this key */
     }
   }
 
@@ -78,8 +79,11 @@ int find_ref_key(KEY *key, uint key_count, byte *record, Field *field,
 	 j++, key_part++)
     {
       if (key_part->offset == fieldpos)
-	return(i);                              /* Use this key */
-      *key_length+=key_part->store_length;
+      {
+        *keypart= j;
+        return i;                               /* Use this key */
+      }
+      *key_length+= key_part->store_length;
     }
   }
   return(-1);					/* No key is ok */
diff --git a/sql/log.cc b/sql/log.cc
index deb77890f35..1994660bf5b 100644
--- a/sql/log.cc
+++ b/sql/log.cc
@@ -3773,7 +3773,7 @@ bool MYSQL_BIN_LOG::write(Log_event *event_info)
                              nb_elements()));
           /*
             If the auto_increment was second in a table's index (possible with
-            MyISAM or BDB) (table->next_number_key_offset != 0), such event is
+            MyISAM or BDB) (table->next_number_keypart != 0), such event is
             in fact not necessary. We could avoid logging it.
           */
           Intvar_log_event e(thd, (uchar) INSERT_ID_EVENT,
diff --git a/sql/log_event.cc b/sql/log_event.cc
index f8d3c43bfba..5ff23a49252 100644
--- a/sql/log_event.cc
+++ b/sql/log_event.cc
@@ -6854,9 +6854,8 @@ replace_record(THD *thd, TABLE *table,
       }
 
       key_copy((byte*)key.get(), table->record[0], table->key_info + keynum, 0);
-      error= table->file->index_read_idx(table->record[1], keynum, 
-                                         (const byte*)key.get(),
-                                         table->key_info[keynum].key_length,
+      error= table->file->index_read_idx(table->record[1], keynum,
+                                         (const byte*)key.get(), ~ULL(0),
                                          HA_READ_KEY_EXACT);
       if (error)
         DBUG_RETURN(error);
@@ -7041,8 +7040,7 @@ static int find_and_fetch_row(TABLE *table, byte *key)
       table->s->null_bytes > 0 ? table->s->null_bytes - 1 : 0;
     table->record[1][pos]= 0xFF;
     if ((error= table->file->index_read(table->record[1], key,
-                                        table->key_info->key_length,
-                                        HA_READ_KEY_EXACT)))
+                                        ~(ulonglong)0, HA_READ_KEY_EXACT)))
     {
       table->file->print_error(error, MYF(0));
       table->file->ha_index_end();
diff --git a/sql/mysql_priv.h b/sql/mysql_priv.h
index efcbdf968bf..d2e37e059a7 100644
--- a/sql/mysql_priv.h
+++ b/sql/mysql_priv.h
@@ -1492,7 +1492,7 @@ void print_plan(JOIN* join,uint idx, double record_count, double read_time,
 void mysql_print_status();
 /* key.cc */
 int find_ref_key(KEY *key, uint key_count, byte *record, Field *field,
-                 uint *key_length);
+                 uint *key_length, uint *keypart);
 void key_copy(byte *to_key, byte *from_record, KEY *key_info, uint key_length);
 void key_restore(byte *to_record, byte *from_key, KEY *key_info,
                  uint key_length);
diff --git a/sql/opt_range.cc b/sql/opt_range.cc
index 05b22cee935..d5fc3af9ceb 100644
--- a/sql/opt_range.cc
+++ b/sql/opt_range.cc
@@ -312,7 +312,7 @@ public:
     min_value=arg->max_value;
     min_flag=arg->max_flag & NEAR_MAX ? 0 : NEAR_MIN;
   }
-  void store_min(uint length,char **min_key,uint min_key_flag)
+  int store_min(uint length,char **min_key,uint min_key_flag)
   {
     if ((min_flag & GEOM_FLAG) ||
         (!(min_flag & NO_MIN_RANGE) &&
@@ -326,12 +326,12 @@ public:
       else
 	memcpy(*min_key,min_value,length);
       (*min_key)+= length;
+      return 1;
     }
+    return 0;
   }
-  void store(uint length,char **min_key,uint min_key_flag,
-	     char **max_key, uint max_key_flag)
+  int store_max(uint length,char **max_key, uint max_key_flag)
   {
-    store_min(length, min_key, min_key_flag);
     if (!(max_flag & NO_MAX_RANGE) &&
 	!(max_key_flag & (NO_MAX_RANGE | NEAR_MAX)))
     {
@@ -343,33 +343,45 @@ public:
       else
 	memcpy(*max_key,max_value,length);
       (*max_key)+= length;
+      return 1;
     }
+    return 0;
   }
+  /*void store(uint length,char **min_key,uint min_key_flag,
+	     char **max_key, uint max_key_flag)
+  {
+    store_min(length, min_key, min_key_flag);
+    store_max(length, max_key, max_key_flag);
+  }*/
 
-  void store_min_key(KEY_PART *key,char **range_key, uint *range_key_flag)
+  int store_min_key(KEY_PART *key,char **range_key, uint *range_key_flag)
   {
     SEL_ARG *key_tree= first();
-    key_tree->store(key[key_tree->part].store_length,
-		    range_key,*range_key_flag,range_key,NO_MAX_RANGE);
+    uint res= key_tree->store_min(key[key_tree->part].store_length,
+                                  range_key, *range_key_flag);
     *range_key_flag|= key_tree->min_flag;
     if (key_tree->next_key_part &&
 	key_tree->next_key_part->part == key_tree->part+1 &&
 	!(*range_key_flag & (NO_MIN_RANGE | NEAR_MIN)) &&
 	key_tree->next_key_part->type == SEL_ARG::KEY_RANGE)
-      key_tree->next_key_part->store_min_key(key,range_key, range_key_flag);
+      res+= key_tree->next_key_part->store_min_key(key, range_key,
+                                                   range_key_flag);
+    return res;
   }
 
-  void store_max_key(KEY_PART *key,char **range_key, uint *range_key_flag)
+  int store_max_key(KEY_PART *key,char **range_key, uint *range_key_flag)
   {
     SEL_ARG *key_tree= last();
-    key_tree->store(key[key_tree->part].store_length,
-		    range_key, NO_MIN_RANGE, range_key,*range_key_flag);
+    uint res=key_tree->store_max(key[key_tree->part].store_length,
+                                 range_key, *range_key_flag);
     (*range_key_flag)|= key_tree->max_flag;
     if (key_tree->next_key_part &&
 	key_tree->next_key_part->part == key_tree->part+1 &&
 	!(*range_key_flag & (NO_MAX_RANGE | NEAR_MAX)) &&
 	key_tree->next_key_part->type == SEL_ARG::KEY_RANGE)
-      key_tree->next_key_part->store_max_key(key,range_key, range_key_flag);
+      res+= key_tree->next_key_part->store_max_key(key, range_key,
+                                                   range_key_flag);
+    return res;
   }
 
   SEL_ARG *insert(SEL_ARG *key);
@@ -586,8 +598,8 @@ static bool is_key_scan_ror(PARAM *param, uint keynr, uint8 nparts);
 static ha_rows check_quick_select(PARAM *param,uint index,SEL_ARG *key_tree, 
                                   bool update_tbl_stats);
 static ha_rows check_quick_keys(PARAM *param,uint index,SEL_ARG *key_tree,
-				char *min_key,uint min_key_flag,
-				char *max_key, uint max_key_flag);
+                                char *min_key, uint min_key_flag, int,
+                                char *max_key, uint max_key_flag, int);
 
 QUICK_RANGE_SELECT *get_quick_select(PARAM *param,uint index,
                                      SEL_ARG *key_tree,
@@ -1453,6 +1465,7 @@ QUICK_ROR_UNION_SELECT::~QUICK_ROR_UNION_SELECT()
 
 QUICK_RANGE::QUICK_RANGE()
   :min_key(0),max_key(0),min_length(0),max_length(0),
+  min_keypart_map(0), max_keypart_map(0),
    flag(NO_MIN_RANGE | NO_MAX_RANGE)
 {}
 
@@ -4031,9 +4044,9 @@ void ror_intersect_cpy(ROR_INTERSECT_INFO *dst, const ROR_INTERSECT_INFO *src)
     The calculation is conducted as follows:
     Lets denote #records(keypart1, ... keypartK) as n_k. We need to calculate
 
-     n_{k1}      n_{k_2}
+     n_{k1}      n_{k2}
     --------- * ---------  * .... (3)
-     n_{k1-1}    n_{k2_1}
+     n_{k1-1}    n_{k2-1}
 
     where k1,k2,... are key parts which fields were not yet marked as fixed
     ( this is result of application of option b) of the recursion step for
@@ -4041,9 +4054,9 @@ void ror_intersect_cpy(ROR_INTERSECT_INFO *dst, const ROR_INTERSECT_INFO *src)
     Since it is reasonable to expect that most of the fields are not marked
     as fixed, we calculate (3) as
 
-                                  n_{i1}      n_{i_2}
+                                  n_{i1}      n_{i2}
     (3) = n_{max_key_part}  / (   --------- * ---------  * ....  )
-                                  n_{i1-1}    n_{i2_1}
+                                  n_{i1-1}    n_{i2-1}
 
     where i1,i2, .. are key parts that were already marked as fixed.
 
@@ -4052,7 +4065,6 @@ void ror_intersect_cpy(ROR_INTERSECT_INFO *dst, const ROR_INTERSECT_INFO *src)
 
   RETURN
     Selectivity of given ROR scan.
-    
 */
 
 static double ror_scan_selectivity(const ROR_INTERSECT_INFO *info, 
@@ -4063,6 +4075,7 @@ static double ror_scan_selectivity(const ROR_INTERSECT_INFO *info,
   byte key_val[MAX_KEY_LENGTH+MAX_FIELD_WIDTH]; /* key values tuple */
   char *key_ptr= (char*) key_val;
   SEL_ARG *sel_arg, *tuple_arg= NULL;
+  ulonglong keypart_map= 0;
   bool cur_covered;
   bool prev_covered= test(bitmap_is_set(&info->covered_fields,
                                         key_part->fieldnr-1));
@@ -4073,7 +4086,7 @@ static double ror_scan_selectivity(const ROR_INTERSECT_INFO *info,
   max_range.key= (byte*) key_val;
   max_range.flag= HA_READ_AFTER_KEY;
   ha_rows prev_records= info->param->table->file->stats.records;
-  DBUG_ENTER("ror_intersect_selectivity");
+  DBUG_ENTER("ror_scan_selectivity");
 
   for (sel_arg= scan->sel_arg; sel_arg;
        sel_arg= sel_arg->next_key_part)
@@ -4090,13 +4103,17 @@ static double ror_scan_selectivity(const ROR_INTERSECT_INFO *info,
         tuple_arg= scan->sel_arg;
         /* Here we use the length of the first key part */
         tuple_arg->store_min(key_part->store_length, &key_ptr, 0);
+        keypart_map= 1;
       }
       while (tuple_arg->next_key_part != sel_arg)
       {
         tuple_arg= tuple_arg->next_key_part;
-        tuple_arg->store_min(key_part[tuple_arg->part].store_length, &key_ptr, 0);
+        tuple_arg->store_min(key_part[tuple_arg->part].store_length,
+                             &key_ptr, 0);
+        keypart_map= (keypart_map << 1) | 1;
       }
       min_range.length= max_range.length= ((char*) key_ptr - (char*) key_val);
+      min_range.keypart_map= max_range.keypart_map= keypart_map;
       records= (info->param->table->file->
                 records_in_range(scan->keynr, &min_range, &max_range));
       if (cur_covered)
@@ -5305,12 +5322,11 @@ static SEL_TREE *get_mm_tree(RANGE_OPT_PARAM *param,COND *cond)
     */
     for (uint i= 1 ; i < cond_func->arg_count ; i++)
     {
-     
       if (cond_func->arguments()[i]->real_item()->type() == Item::FIELD_ITEM)
       {
         field_item= (Item_field*) (cond_func->arguments()[i]->real_item());
         SEL_TREE *tmp= get_full_func_mm_tree(param, cond_func, 
-                                    field_item, (Item*) i, inv);
+                                    field_item, (Item*)(intptr)i, inv);
         if (inv)
           tree= !tree ? tmp : tree_or(param, tree, tmp);
         else 
@@ -7057,7 +7073,9 @@ check_quick_select(PARAM *param,uint idx,SEL_ARG *tree, bool update_tbl_stats)
   }
   param->n_ranges= 0;
 
-  records=check_quick_keys(param,idx,tree,param->min_key,0,param->max_key,0);
+  records= check_quick_keys(param, idx, tree,
+                            param->min_key, 0, -1,
+                            param->max_key, 0, -1);
   if (records != HA_POS_ERROR)
   {
     if (update_tbl_stats)
@@ -7120,12 +7138,13 @@ check_quick_select(PARAM *param,uint idx,SEL_ARG *tree, bool update_tbl_stats)
 */
 
 static ha_rows
-check_quick_keys(PARAM *param,uint idx,SEL_ARG *key_tree,
-		 char *min_key,uint min_key_flag, char *max_key,
-		 uint max_key_flag)
+check_quick_keys(PARAM *param, uint idx, SEL_ARG *key_tree,
+		 char *min_key, uint min_key_flag, int min_keypart,
+                 char *max_key, uint max_key_flag, int max_keypart)
 {
   ha_rows records=0, tmp;
   uint tmp_min_flag, tmp_max_flag, keynr, min_key_length, max_key_length;
+  uint tmp_min_keypart= min_keypart, tmp_max_keypart= max_keypart;
   char *tmp_min_key, *tmp_max_key;
   uint8 save_first_null_comp= param->first_null_comp;
 
@@ -7139,18 +7158,21 @@ check_quick_keys(PARAM *param,uint idx,SEL_ARG *key_tree,
       This is not a ROR scan if the key is not Clustered Primary Key.
     */
     param->is_ror_scan= FALSE;
-    records=check_quick_keys(param,idx,key_tree->left,min_key,min_key_flag,
-			     max_key,max_key_flag);
+    records=check_quick_keys(param, idx, key_tree->left,
+                             min_key, min_key_flag, min_keypart,
+			     max_key, max_key_flag, max_keypart);
     if (records == HA_POS_ERROR)			// Impossible
       return records;
   }
 
   tmp_min_key= min_key;
   tmp_max_key= max_key;
-  key_tree->store(param->key[idx][key_tree->part].store_length,
-		  &tmp_min_key,min_key_flag,&tmp_max_key,max_key_flag);
-  min_key_length= (uint) (tmp_min_key- param->min_key);
-  max_key_length= (uint) (tmp_max_key- param->max_key);
+  tmp_min_keypart+= key_tree->store_min(param->key[idx][key_tree->part].store_length,
+                                        &tmp_min_key, min_key_flag);
+  tmp_max_keypart+= key_tree->store_max(param->key[idx][key_tree->part].store_length,
+                                        &tmp_max_key, max_key_flag);
+  min_key_length= (uint) (tmp_min_key - param->min_key);
+  max_key_length= (uint) (tmp_max_key - param->max_key);
 
   if (param->is_ror_scan)
   {
@@ -7173,12 +7195,13 @@ check_quick_keys(PARAM *param,uint idx,SEL_ARG *key_tree,
       key_tree->next_key_part->type == SEL_ARG::KEY_RANGE)
   {						// const key as prefix
     if (min_key_length == max_key_length &&
-	!memcmp(min_key,max_key, (uint) (tmp_max_key - max_key)) &&
+	!memcmp(min_key, max_key, (uint) (tmp_max_key - max_key)) &&
 	!key_tree->min_flag && !key_tree->max_flag)
     {
-      tmp=check_quick_keys(param,idx,key_tree->next_key_part,
-			   tmp_min_key, min_key_flag | key_tree->min_flag,
-			   tmp_max_key, max_key_flag | key_tree->max_flag);
+      tmp=check_quick_keys(param,idx,key_tree->next_key_part, tmp_min_key,
+                           min_key_flag | key_tree->min_flag, tmp_min_keypart,
+                           tmp_max_key, max_key_flag | key_tree->max_flag,
+                           tmp_max_keypart);
       goto end;					// Ugly, but efficient
     }
     else
@@ -7190,18 +7213,20 @@ check_quick_keys(PARAM *param,uint idx,SEL_ARG *key_tree,
     tmp_min_flag=key_tree->min_flag;
     tmp_max_flag=key_tree->max_flag;
     if (!tmp_min_flag)
+      tmp_min_keypart+=
       key_tree->next_key_part->store_min_key(param->key[idx], &tmp_min_key,
 					     &tmp_min_flag);
     if (!tmp_max_flag)
+      tmp_max_keypart+=
       key_tree->next_key_part->store_max_key(param->key[idx], &tmp_max_key,
 					     &tmp_max_flag);
-    min_key_length= (uint) (tmp_min_key- param->min_key);
-    max_key_length= (uint) (tmp_max_key- param->max_key);
+    min_key_length= (uint) (tmp_min_key - param->min_key);
+    max_key_length= (uint) (tmp_max_key - param->max_key);
   }
   else
   {
-    tmp_min_flag=min_key_flag | key_tree->min_flag;
-    tmp_max_flag=max_key_flag | key_tree->max_flag;
+    tmp_min_flag= min_key_flag | key_tree->min_flag;
+    tmp_max_flag= max_key_flag | key_tree->max_flag;
   }
 
   keynr=param->real_keynr[idx];
@@ -7209,9 +7234,8 @@ check_quick_keys(PARAM *param,uint idx,SEL_ARG *key_tree,
   if (!tmp_min_flag && ! tmp_max_flag &&
       (uint) key_tree->part+1 == param->table->key_info[keynr].key_parts &&
       (param->table->key_info[keynr].flags & (HA_NOSAME | HA_END_SPACE_KEY)) ==
-      HA_NOSAME &&
-      min_key_length == max_key_length &&
-      !memcmp(param->min_key,param->max_key,min_key_length) &&
+      HA_NOSAME && min_key_length == max_key_length &&
+      !memcmp(param->min_key, param->max_key, min_key_length) &&
       !param->first_null_comp)
   {
     tmp=1;					// Max one record
@@ -7231,7 +7255,7 @@ check_quick_keys(PARAM *param,uint idx,SEL_ARG *key_tree,
           first members of clustered primary key.
       */
       if (!(min_key_length == max_key_length &&
-            !memcmp(min_key,max_key, (uint) (tmp_max_key - max_key)) &&
+            !memcmp(min_key, max_key, (uint) (tmp_max_key - max_key)) &&
             !key_tree->min_flag && !key_tree->max_flag &&
             is_key_scan_ror(param, keynr, key_tree->part + 1)))
         param->is_ror_scan= FALSE;
@@ -7243,11 +7267,12 @@ check_quick_keys(PARAM *param,uint idx,SEL_ARG *key_tree,
       key_range min_range;
       min_range.key=    (byte*) param->min_key;
       min_range.length= min_key_length;
+      min_range.keypart_map= make_keypart_map(tmp_min_keypart);
       /* In this case tmp_min_flag contains the handler-read-function */
       min_range.flag=   (ha_rkey_function) (tmp_min_flag ^ GEOM_FLAG);
 
-      tmp= param->table->file->records_in_range(keynr, &min_range,
-                                                (key_range*) 0);
+      tmp= param->table->file->records_in_range(keynr,
+                                                &min_range, (key_range*) 0);
     }
     else
     {
@@ -7257,10 +7282,12 @@ check_quick_keys(PARAM *param,uint idx,SEL_ARG *key_tree,
       min_range.length= min_key_length;
       min_range.flag=   (tmp_min_flag & NEAR_MIN ? HA_READ_AFTER_KEY :
                          HA_READ_KEY_EXACT);
+      min_range.keypart_map= make_keypart_map(tmp_min_keypart);
       max_range.key=    (byte*) param->max_key;
       max_range.length= max_key_length;
       max_range.flag=   (tmp_max_flag & NEAR_MAX ?
                          HA_READ_BEFORE_KEY : HA_READ_AFTER_KEY);
+      max_range.keypart_map= make_keypart_map(tmp_max_keypart);
       tmp=param->table->file->records_in_range(keynr,
                                                (min_key_length ? &min_range :
                                                 (key_range*) 0),
@@ -7281,8 +7308,9 @@ check_quick_keys(PARAM *param,uint idx,SEL_ARG *key_tree,
       This is not a ROR scan if the key is not Clustered Primary Key.
     */
     param->is_ror_scan= FALSE;
-    tmp=check_quick_keys(param,idx,key_tree->right,min_key,min_key_flag,
-			 max_key,max_key_flag);
+    tmp=check_quick_keys(param, idx, key_tree->right,
+                         min_key, min_key_flag, min_keypart,
+                         max_key, max_key_flag, max_keypart);
     if (tmp == HA_POS_ERROR)
       return tmp;
     records+=tmp;
@@ -7429,6 +7457,7 @@ get_quick_keys(PARAM *param,QUICK_RANGE_SELECT *quick,KEY_PART *key,
 {
   QUICK_RANGE *range;
   uint flag;
+  int min_part= key_tree->part-1, max_part=key_tree->part-1;
 
   if (key_tree->left != &null_element)
   {
@@ -7437,16 +7466,18 @@ get_quick_keys(PARAM *param,QUICK_RANGE_SELECT *quick,KEY_PART *key,
       return 1;
   }
   char *tmp_min_key=min_key,*tmp_max_key=max_key;
-  key_tree->store(key[key_tree->part].store_length,
-		  &tmp_min_key,min_key_flag,&tmp_max_key,max_key_flag);
+  min_part+= key_tree->store_min(key[key_tree->part].store_length,
+                                 &tmp_min_key,min_key_flag);
+  max_part+= key_tree->store_max(key[key_tree->part].store_length,
+                                 &tmp_max_key,max_key_flag);
 
   if (key_tree->next_key_part &&
       key_tree->next_key_part->part == key_tree->part+1 &&
       key_tree->next_key_part->type == SEL_ARG::KEY_RANGE)
   {						  // const key as prefix
-    if (!((tmp_min_key - min_key) != (tmp_max_key - max_key) ||
-	  memcmp(min_key,max_key, (uint) (tmp_max_key - max_key)) ||
-	  key_tree->min_flag || key_tree->max_flag))
+    if ((tmp_min_key - min_key) == (tmp_max_key - max_key) &&
+         memcmp(min_key, max_key, (uint)(tmp_max_key - max_key))==0 &&
+	 key_tree->min_flag==0 && key_tree->max_flag==0)
     {
       if (get_quick_keys(param,quick,key,key_tree->next_key_part,
 			 tmp_min_key, min_key_flag | key_tree->min_flag,
@@ -7457,11 +7488,15 @@ get_quick_keys(PARAM *param,QUICK_RANGE_SELECT *quick,KEY_PART *key,
     {
       uint tmp_min_flag=key_tree->min_flag,tmp_max_flag=key_tree->max_flag;
       if (!tmp_min_flag)
-	key_tree->next_key_part->store_min_key(key, &tmp_min_key,
+      {
+        min_part+= key_tree->next_key_part->store_min_key(key, &tmp_min_key,
 					       &tmp_min_flag);
+      }
       if (!tmp_max_flag)
-	key_tree->next_key_part->store_max_key(key, &tmp_max_key,
+      {
+        max_part+= key_tree->next_key_part->store_max_key(key, &tmp_max_key,
 					       &tmp_max_flag);
+      }
       flag=tmp_min_flag | tmp_max_flag;
     }
   }
@@ -7511,13 +7546,15 @@ get_quick_keys(PARAM *param,QUICK_RANGE_SELECT *quick,KEY_PART *key,
   /* Get range for retrieving rows in QUICK_SELECT::get_next */
   if (!(range= new QUICK_RANGE((const char *) param->min_key,
 			       (uint) (tmp_min_key - param->min_key),
+                               min_part >=0 ? make_keypart_map(min_part) : 0,
 			       (const char *) param->max_key,
 			       (uint) (tmp_max_key - param->max_key),
+                               max_part >=0 ? make_keypart_map(max_part) : 0,
 			       flag)))
     return 1;			// out of memory
 
-  set_if_bigger(quick->max_used_key_length,range->min_length);
-  set_if_bigger(quick->max_used_key_length,range->max_length);
+  set_if_bigger(quick->max_used_key_length, range->min_length);
+  set_if_bigger(quick->max_used_key_length, range->max_length);
   set_if_bigger(quick->used_key_parts, (uint) key_tree->part+1);
   if (insert_dynamic(&quick->ranges, (gptr)&range))
     return 1;
@@ -7659,6 +7696,7 @@ QUICK_RANGE_SELECT *get_quick_select_for_ref(THD *thd, TABLE *table,
 
   range->min_key=range->max_key=(char*) ref->key_buff;
   range->min_length=range->max_length=ref->key_length;
+  range->min_keypart_map= range->max_keypart_map= (1 << ref->key_parts) - 1;
   range->flag= ((ref->key_length == key_info->key_length &&
 		 (key_info->flags & (HA_NOSAME | HA_END_SPACE_KEY)) ==
 		 HA_NOSAME) ? EQ_RANGE : 0);
@@ -7692,8 +7730,10 @@ QUICK_RANGE_SELECT *get_quick_select_for_ref(THD *thd, TABLE *table,
     *ref->null_ref_key= 1;		// Set null byte then create a range
     if (!(null_range= new (alloc) QUICK_RANGE((char*)ref->key_buff,
                                               ref->key_length,
+                                              (1 << ref->key_parts) - 1,
                                               (char*)ref->key_buff,
                                               ref->key_length,
+                                              (1 << ref->key_parts) - 1,
                                               EQ_RANGE)))
       goto err;
     *ref->null_ref_key= 0;		// Clear null byte
@@ -8146,6 +8186,7 @@ int QUICK_RANGE_SELECT::get_next()
       start_key->flag=   ((last_range->flag & NEAR_MIN) ? HA_READ_AFTER_KEY :
                           (last_range->flag & EQ_RANGE) ?
                           HA_READ_KEY_EXACT : HA_READ_KEY_OR_NEXT);
+      start_key->keypart_map= last_range->min_keypart_map;
       end_key->key=      (const byte*) last_range->max_key;
       end_key->length=   last_range->max_length;
       /*
@@ -8154,6 +8195,7 @@ int QUICK_RANGE_SELECT::get_next()
       */
       end_key->flag=     (last_range->flag & NEAR_MAX ? HA_READ_BEFORE_KEY :
                           HA_READ_AFTER_KEY);
+      end_key->keypart_map= last_range->max_keypart_map;
 
       mrange_slot->range_flag= last_range->flag;
     }
@@ -8203,7 +8245,9 @@ end:
     other              if some error occurred
 */
 
-int QUICK_RANGE_SELECT::get_next_prefix(uint prefix_length, byte *cur_prefix)
+int QUICK_RANGE_SELECT::get_next_prefix(uint prefix_length,
+                                        ulonglong keypart_map,
+                                        byte *cur_prefix)
 {
   DBUG_ENTER("QUICK_RANGE_SELECT::get_next_prefix");
 
@@ -8215,8 +8259,7 @@ int QUICK_RANGE_SELECT::get_next_prefix(uint prefix_length, byte *cur_prefix)
     {
       /* Read the next record in the same range with prefix after cur_prefix. */
       DBUG_ASSERT(cur_prefix != 0);
-      result= file->index_read(record, cur_prefix, prefix_length,
-                               HA_READ_AFTER_KEY);
+      result= file->index_read(record, cur_prefix, keypart_map, HA_READ_AFTER_KEY);
       if (result || (file->compare_key(file->end_range) <= 0))
         DBUG_RETURN(result);
     }
@@ -8232,11 +8275,13 @@ int QUICK_RANGE_SELECT::get_next_prefix(uint prefix_length, byte *cur_prefix)
 
     start_key.key=    (const byte*) last_range->min_key;
     start_key.length= min(last_range->min_length, prefix_length);
+    start_key.keypart_map= last_range->min_keypart_map & keypart_map;
     start_key.flag=   ((last_range->flag & NEAR_MIN) ? HA_READ_AFTER_KEY :
 		       (last_range->flag & EQ_RANGE) ?
 		       HA_READ_KEY_EXACT : HA_READ_KEY_OR_NEXT);
     end_key.key=      (const byte*) last_range->max_key;
     end_key.length=   min(last_range->max_length, prefix_length);
+    end_key.keypart_map= last_range->max_keypart_map & keypart_map;
     /*
       We use READ_AFTER_KEY here because if we are reading on a key
       prefix we want to find all keys with this prefix
@@ -8244,8 +8289,8 @@ int QUICK_RANGE_SELECT::get_next_prefix(uint prefix_length, byte *cur_prefix)
     end_key.flag=     (last_range->flag & NEAR_MAX ? HA_READ_BEFORE_KEY :
 		       HA_READ_AFTER_KEY);
 
-    result= file->read_range_first(last_range->min_length ? &start_key : 0,
-				   last_range->max_length ? &end_key : 0,
+    result= file->read_range_first(last_range->min_keypart_map ? &start_key : 0,
+				   last_range->max_keypart_map ? &end_key : 0,
                                    test(last_range->flag & EQ_RANGE),
 				   sorted);
     if (last_range->flag == (UNIQUE_RANGE | EQ_RANGE))
@@ -8285,9 +8330,8 @@ int QUICK_RANGE_SELECT_GEOM::get_next()
     }
     last_range= *(cur_range++);
 
-    result= file->index_read(record,
-			     (byte*) last_range->min_key,
-			     last_range->min_length,
+    result= file->index_read(record, (byte*) last_range->min_key,
+			     last_range->min_keypart_map,
 			     (ha_rkey_function)(last_range->flag ^ GEOM_FLAG));
     if (result != HA_ERR_KEY_NOT_FOUND && result != HA_ERR_END_OF_FILE)
       DBUG_RETURN(result);
@@ -8419,15 +8463,15 @@ int QUICK_SELECT_DESC::get_next()
 
     if (last_range->flag & EQ_RANGE)
     {
-      result= file->index_read(record, (byte*) last_range->max_key,
-                               last_range->max_length, HA_READ_KEY_EXACT);
+      result = file->index_read(record, (byte*) last_range->max_key,
+				last_range->max_keypart_map, HA_READ_KEY_EXACT);
     }
     else
     {
       DBUG_ASSERT(last_range->flag & NEAR_MAX ||
                   range_reads_after_key(last_range));
       result=file->index_read(record, (byte*) last_range->max_key,
-			      last_range->max_length,
+			      last_range->max_keypart_map,
 			      ((last_range->flag & NEAR_MAX) ?
 			       HA_READ_BEFORE_KEY :
                                HA_READ_PREFIX_LAST_OR_PREV));
@@ -8747,8 +8791,7 @@ void QUICK_ROR_UNION_SELECT::add_keys_and_lengths(String *key_names,
 static inline uint get_field_keypart(KEY *index, Field *field);
 static inline SEL_ARG * get_index_range_tree(uint index, SEL_TREE* range_tree,
                                              PARAM *param, uint *param_idx);
-static bool
-get_constant_key_infix(KEY *index_info, SEL_ARG *index_range_tree,
+static bool get_constant_key_infix(KEY *index_info, SEL_ARG *index_range_tree,
                        KEY_PART_INFO *first_non_group_part,
                        KEY_PART_INFO *min_max_arg_part,
                        KEY_PART_INFO *last_part, THD *thd,
@@ -9146,7 +9189,7 @@ get_best_group_min_max(PARAM *param, SEL_TREE *tree)
                           NULL;
     first_non_infix_part= min_max_arg_part ?
                           (min_max_arg_part < last_part) ?
-                             min_max_arg_part + 1 :
+                             min_max_arg_part :
                              NULL :
                            NULL;
     if (first_non_group_part &&
@@ -9203,7 +9246,9 @@ get_best_group_min_max(PARAM *param, SEL_TREE *tree)
     */
     if (first_non_infix_part)
     {
-      for (cur_part= first_non_infix_part; cur_part != last_part; cur_part++)
+      cur_part= first_non_infix_part +
+                (min_max_arg_part && (min_max_arg_part < last_part));
+      for (; cur_part != last_part; cur_part++)
       {
         if (bitmap_is_set(table->read_set, cur_part->field->field_index))
           goto next_index;
@@ -9749,7 +9794,7 @@ void cost_group_min_max(TABLE* table, KEY *index_info, uint used_key_parts,
 
   RETURN
     New QUICK_GROUP_MIN_MAX_SELECT object if successfully created,
-    NULL o/w.
+    NULL otherwise.
 */
 
 QUICK_SELECT_I *
@@ -9762,10 +9807,10 @@ TRP_GROUP_MIN_MAX::make_quick(PARAM *param, bool retrieve_full_rows,
   quick= new QUICK_GROUP_MIN_MAX_SELECT(param->table,
                                         param->thd->lex->current_select->join,
                                         have_min, have_max, min_max_arg_part,
-                                        group_prefix_len, used_key_parts,
-                                        index_info, index, read_cost, records,
-                                        key_infix_len, key_infix,
-                                        parent_alloc);
+                                        group_prefix_len, group_key_parts,
+                                        used_key_parts, index_info, index,
+                                        read_cost, records, key_infix_len,
+                                        key_infix, parent_alloc);
   if (!quick)
     DBUG_RETURN(NULL);
 
@@ -9854,7 +9899,7 @@ QUICK_GROUP_MIN_MAX_SELECT::
 QUICK_GROUP_MIN_MAX_SELECT(TABLE *table, JOIN *join_arg, bool have_min_arg,
                            bool have_max_arg,
                            KEY_PART_INFO *min_max_arg_part_arg,
-                           uint group_prefix_len_arg,
+                           uint group_prefix_len_arg, uint group_key_parts_arg,
                            uint used_key_parts_arg, KEY *index_info_arg,
                            uint use_index, double read_cost_arg,
                            ha_rows records_arg, uint key_infix_len_arg,
@@ -9864,7 +9909,7 @@ QUICK_GROUP_MIN_MAX_SELECT(TABLE *table, JOIN *join_arg, bool have_min_arg,
    have_max(have_max_arg), seen_first_key(FALSE),
    min_max_arg_part(min_max_arg_part_arg), key_infix(key_infix_arg),
    key_infix_len(key_infix_len_arg), min_functions_it(NULL),
-   max_functions_it(NULL)
+   max_functions_it(NULL), group_key_parts(group_key_parts_arg)
 {
   head=       table;
   file=       head->file;
@@ -9874,6 +9919,7 @@ QUICK_GROUP_MIN_MAX_SELECT(TABLE *table, JOIN *join_arg, bool have_min_arg,
   read_time= read_cost_arg;
   records= records_arg;
   used_key_parts= used_key_parts_arg;
+  real_key_parts= used_key_parts_arg;
   real_prefix_len= group_prefix_len + key_infix_len;
   group_prefix= NULL;
   min_max_arg_len= min_max_arg_part ? min_max_arg_part->store_length : 0;
@@ -10040,7 +10086,9 @@ bool QUICK_GROUP_MIN_MAX_SELECT::add_range(SEL_ARG *sel_range)
       range_flag|= EQ_RANGE;  /* equality condition */
   }
   range= new QUICK_RANGE(sel_range->min_value, min_max_arg_len,
+                         make_keypart_map(sel_range->part),
                          sel_range->max_value, min_max_arg_len,
+                         make_keypart_map(sel_range->part),
                          range_flag);
   if (!range)
     return TRUE;
@@ -10279,7 +10327,8 @@ int QUICK_GROUP_MIN_MAX_SELECT::get_next()
       first sub-group with the extended prefix.
     */
     if (!have_min && !have_max && key_infix_len > 0)
-      result= file->index_read(record, group_prefix, real_prefix_len,
+      result= file->index_read(record, group_prefix,
+                               make_prev_keypart_map(real_key_parts),
                                HA_READ_KEY_EXACT);
 
     result= have_min ? min_res : have_max ? max_res : result;
@@ -10342,7 +10391,8 @@ int QUICK_GROUP_MIN_MAX_SELECT::next_min()
     /* Apply the constant equality conditions to the non-group select fields */
     if (key_infix_len > 0)
     {
-      if ((result= file->index_read(record, group_prefix, real_prefix_len,
+      if ((result= file->index_read(record, group_prefix,
+                                    make_prev_keypart_map(real_key_parts),
                                     HA_READ_KEY_EXACT)))
         DBUG_RETURN(result);
     }
@@ -10359,7 +10409,7 @@ int QUICK_GROUP_MIN_MAX_SELECT::next_min()
       /* Find the first subsequent record without NULL in the MIN/MAX field. */
       key_copy(tmp_record, record, index_info, 0);
       result= file->index_read(record, tmp_record,
-                               real_prefix_len + min_max_arg_len,
+                               make_keypart_map(real_key_parts),
                                HA_READ_AFTER_KEY);
       /*
         Check if the new record belongs to the current group by comparing its
@@ -10415,7 +10465,8 @@ int QUICK_GROUP_MIN_MAX_SELECT::next_max()
   if (min_max_ranges.elements > 0)
     result= next_max_in_range();
   else
-    result= file->index_read(record, group_prefix, real_prefix_len,
+    result= file->index_read(record, group_prefix,
+                             make_prev_keypart_map(real_key_parts),
                              HA_READ_PREFIX_LAST);
   DBUG_RETURN(result);
 }
@@ -10451,7 +10502,7 @@ int QUICK_GROUP_MIN_MAX_SELECT::next_prefix()
   {
     byte *cur_prefix= seen_first_key ? group_prefix : NULL;
     if ((result= quick_prefix_select->get_next_prefix(group_prefix_len,
-                                                      cur_prefix)))
+                         (ULL(1) << group_key_parts) - 1, cur_prefix)))
       DBUG_RETURN(result);
     seen_first_key= TRUE;
   }
@@ -10467,7 +10518,8 @@ int QUICK_GROUP_MIN_MAX_SELECT::next_prefix()
     else
     {
       /* Load the first key in this group into record. */
-      result= file->index_read(record, group_prefix, group_prefix_len,
+      result= file->index_read(record, group_prefix,
+                               make_prev_keypart_map(group_key_parts),
                                HA_READ_AFTER_KEY);
       if (result)
         DBUG_RETURN(result);
@@ -10510,6 +10562,7 @@ int QUICK_GROUP_MIN_MAX_SELECT::next_min_in_range()
 {
   ha_rkey_function find_flag;
   uint search_prefix_len;
+  ulonglong keypart_map;
   QUICK_RANGE *cur_range;
   bool found_null= FALSE;
   int result= HA_ERR_KEY_NOT_FOUND;
@@ -10531,8 +10584,9 @@ int QUICK_GROUP_MIN_MAX_SELECT::next_min_in_range()
 
     if (cur_range->flag & NO_MIN_RANGE)
     {
-      find_flag= HA_READ_KEY_EXACT;
       search_prefix_len= real_prefix_len;
+      keypart_map= (ULL(1) << real_key_parts) - 1;
+      find_flag= HA_READ_KEY_EXACT;
     }
     else
     {
@@ -10540,13 +10594,13 @@ int QUICK_GROUP_MIN_MAX_SELECT::next_min_in_range()
       memcpy(group_prefix + real_prefix_len, cur_range->min_key,
              cur_range->min_length);
       search_prefix_len= real_prefix_len + min_max_arg_len;
+      keypart_map= (ULL(2) << real_key_parts) - 1;
       find_flag= (cur_range->flag & (EQ_RANGE | NULL_RANGE)) ?
                  HA_READ_KEY_EXACT : (cur_range->flag & NEAR_MIN) ?
                  HA_READ_AFTER_KEY : HA_READ_KEY_OR_NEXT;
     }
 
-    result= file->index_read(record, group_prefix, search_prefix_len,
-                             find_flag);
+    result= file->index_read(record, group_prefix, keypart_map, find_flag);
     if (result)
     {
       if ((result == HA_ERR_KEY_NOT_FOUND || result == HA_ERR_END_OF_FILE) &&
@@ -10644,6 +10698,7 @@ int QUICK_GROUP_MIN_MAX_SELECT::next_max_in_range()
 {
   ha_rkey_function find_flag;
   uint search_prefix_len;
+  ulonglong keypart_map;
   QUICK_RANGE *cur_range;
   int result;
 
@@ -10665,8 +10720,9 @@ int QUICK_GROUP_MIN_MAX_SELECT::next_max_in_range()
 
     if (cur_range->flag & NO_MAX_RANGE)
     {
-      find_flag= HA_READ_PREFIX_LAST;
       search_prefix_len= real_prefix_len;
+      keypart_map= (ULL(1) << real_key_parts) - 1;
+      find_flag= HA_READ_PREFIX_LAST;
     }
     else
     {
@@ -10674,13 +10730,13 @@ int QUICK_GROUP_MIN_MAX_SELECT::next_max_in_range()
       memcpy(group_prefix + real_prefix_len, cur_range->max_key,
              cur_range->max_length);
       search_prefix_len= real_prefix_len + min_max_arg_len;
+      keypart_map= (ULL(2) << real_key_parts) - 1;
       find_flag= (cur_range->flag & EQ_RANGE) ?
                  HA_READ_KEY_EXACT : (cur_range->flag & NEAR_MAX) ?
                  HA_READ_BEFORE_KEY : HA_READ_PREFIX_LAST_OR_PREV;
     }
 
-    result= file->index_read(record, group_prefix, search_prefix_len,
-                             find_flag);
+    result= file->index_read(record, group_prefix, keypart_map, find_flag);
 
     if (result)
     {
diff --git a/sql/opt_range.h b/sql/opt_range.h
index d82e1dc459e..ce98b4609d1 100644
--- a/sql/opt_range.h
+++ b/sql/opt_range.h
@@ -37,17 +37,22 @@ class QUICK_RANGE :public Sql_alloc {
  public:
   char *min_key,*max_key;
   uint16 min_length,max_length,flag;
+  ulonglong min_keypart_map, max_keypart_map;
 #ifdef HAVE_purify
   uint16 dummy;					/* Avoid warnings on 'flag' */
 #endif
   QUICK_RANGE();				/* Full range */
-  QUICK_RANGE(const char *min_key_arg,uint min_length_arg,
-	      const char *max_key_arg,uint max_length_arg,
+  QUICK_RANGE(const char *min_key_arg, uint min_length_arg,
+              ulonglong min_keypart_map_arg,
+	      const char *max_key_arg, uint max_length_arg,
+              ulonglong max_keypart_map_arg,
 	      uint flag_arg)
     : min_key((char*) sql_memdup(min_key_arg,min_length_arg+1)),
       max_key((char*) sql_memdup(max_key_arg,max_length_arg+1)),
       min_length((uint16) min_length_arg),
       max_length((uint16) max_length_arg),
+      min_keypart_map(min_keypart_map_arg),
+      max_keypart_map(max_keypart_map_arg),
       flag((uint16) flag_arg)
     {
 #ifdef HAVE_purify
@@ -318,7 +323,8 @@ public:
   int reset(void);
   int get_next();
   void range_end();
-  int get_next_prefix(uint prefix_length, byte *cur_prefix);
+  int get_next_prefix(uint prefix_length, ulonglong keypart_map,
+                      byte *cur_prefix);
   bool reverse_sorted() { return 0; }
   bool unique_key_range();
   int init_ror_merged_scan(bool reuse_handler);
@@ -605,6 +611,7 @@ private:
   byte *tmp_record;      /* Temporary storage for next_min(), next_max(). */
   byte *group_prefix;    /* Key prefix consisting of the GROUP fields. */
   uint group_prefix_len; /* Length of the group prefix. */
+  uint group_key_parts;
   byte *last_prefix;     /* Prefix of the last group for detecting EOF. */
   bool have_min;         /* Specify whether we are computing */
   bool have_max;         /*   a MIN, a MAX, or both.         */
@@ -616,6 +623,7 @@ private:
   uint key_infix_len;
   DYNAMIC_ARRAY min_max_ranges; /* Array of range ptrs for the MIN/MAX field. */
   uint real_prefix_len; /* Length of key prefix extended with key_infix. */
+  uint real_key_parts;
   List<Item_sum> *min_functions;
   List<Item_sum> *max_functions;
   List_iterator<Item_sum> *min_functions_it;
@@ -638,10 +646,11 @@ private:
 public:
   QUICK_GROUP_MIN_MAX_SELECT(TABLE *table, JOIN *join, bool have_min,
                              bool have_max, KEY_PART_INFO *min_max_arg_part,
-                             uint group_prefix_len, uint used_key_parts,
-                             KEY *index_info, uint use_index, double read_cost,
-                             ha_rows records, uint key_infix_len,
-                             byte *key_infix, MEM_ROOT *parent_alloc);
+                             uint group_prefix_len, uint group_key_parts,
+                             uint used_key_parts, KEY *index_info, uint
+                             use_index, double read_cost, ha_rows records, uint
+                             key_infix_len, byte *key_infix, MEM_ROOT
+                             *parent_alloc);
   ~QUICK_GROUP_MIN_MAX_SELECT();
   bool add_range(SEL_ARG *sel_range);
   void update_key_stat();
diff --git a/sql/opt_sum.cc b/sql/opt_sum.cc
index 5bd5ec4b42d..f9a06f3fb6e 100644
--- a/sql/opt_sum.cc
+++ b/sql/opt_sum.cc
@@ -251,7 +251,7 @@ int opt_sum_query(TABLE_LIST *tables, List<Item> &all_fields,COND *conds)
             error= table->file->index_first(table->record[0]);
           else
 	    error= table->file->index_read(table->record[0],key_buff,
-					   ref.key_length,
+                                           make_prev_keypart_map(ref.key_parts),
 					   range_fl & NEAR_MIN ?
 					   HA_READ_AFTER_KEY :
 					   HA_READ_KEY_OR_NEXT);
@@ -338,11 +338,11 @@ int opt_sum_query(TABLE_LIST *tables, List<Item> &all_fields,COND *conds)
             error= table->file->index_last(table->record[0]);
           else
 	    error= table->file->index_read(table->record[0], key_buff,
-					   ref.key_length,
+                                           make_prev_keypart_map(ref.key_parts),
 					   range_fl & NEAR_MAX ?
 					   HA_READ_BEFORE_KEY :
 					   HA_READ_PREFIX_LAST_OR_PREV);
-	  if (!error && reckey_in_range(1, &ref, item_field->field, 
+	  if (!error && reckey_in_range(1, &ref, item_field->field,
 			                conds, range_fl, prefix_len))
 	    error= HA_ERR_KEY_NOT_FOUND;
           if (table->key_read)
@@ -605,15 +605,13 @@ static bool matching_cond(bool max_fl, TABLE_REF *ref, KEY *keyinfo,
   /* Check if field is part of the tested partial key */
   byte *key_ptr= ref->key_buff;
   KEY_PART_INFO *part;
-  for (part= keyinfo->key_part;
-       ;
-       key_ptr+= part++->store_length)
+  for (part= keyinfo->key_part; ; key_ptr+= part++->store_length)
 
   {
     if (part > field_part)
       return 0;                     // Field is beyond the tested parts
     if (part->field->eq(((Item_field*) args[0])->field))
-      break;                        // Found a part od the key for the field
+      break;                        // Found a part of the key for the field
   }
 
   bool is_field_part= part == field_part;
@@ -625,8 +623,11 @@ static bool matching_cond(bool max_fl, TABLE_REF *ref, KEY *keyinfo,
   {
     uint length= (key_ptr-ref->key_buff)+part->store_length;
     if (ref->key_length < length)
+    {
     /* Ultimately ref->key_length will contain the length of the search key */
       ref->key_length= length;      
+      ref->key_parts= (part - keyinfo->key_part) + 1;
+    }
     if (!*prefix_len && part+1 == field_part)       
       *prefix_len= length;
     if (is_field_part && eq_type)
@@ -773,6 +774,7 @@ static bool find_key_for_maxmin(bool max_fl, TABLE_REF *ref,
       {
         ref->key= idx;
         ref->key_length= 0;
+        ref->key_parts= 0;
         key_part_map key_part_used= 0;
         *range_fl= NO_MIN_RANGE | NO_MAX_RANGE;
         if (matching_cond(max_fl, ref, keyinfo, part, cond,
@@ -788,6 +790,8 @@ static bool find_key_for_maxmin(bool max_fl, TABLE_REF *ref,
             */
             ref->key_buff[ref->key_length]= 1;
             ref->key_length+= part->store_length;
+            ref->key_parts++;
+            DBUG_ASSERT(ref->key_parts == jdx+1);
             *range_fl&= ~NO_MIN_RANGE;
             *range_fl|= NEAR_MIN;                // > NULL
           }
diff --git a/sql/slave.cc b/sql/slave.cc
index 6f62f74647a..6bd5fd3da21 100644
--- a/sql/slave.cc
+++ b/sql/slave.cc
@@ -31,6 +31,8 @@
 #include "rpl_tblmap.h"
 
 int queue_event(MASTER_INFO* mi,const char* buf,ulong event_len);
+static Log_event* next_event(RELAY_LOG_INFO* rli);
+
 
 #define FLAGSTR(V,F) ((V)&(F)?#F" ":"")
 
diff --git a/sql/sp.cc b/sql/sp.cc
index 0c0bc8e8869..27b3b2532c3 100644
--- a/sql/sp.cc
+++ b/sql/sp.cc
@@ -218,8 +218,7 @@ db_find_routine_aux(THD *thd, int type, sp_name *name, TABLE *table)
   key_copy(key, table->record[0], table->key_info,
            table->key_info->key_length);
 
-  if (table->file->index_read_idx(table->record[0], 0,
-				  key, table->key_info->key_length,
+  if (table->file->index_read_idx(table->record[0], 0, key, ~ULL(0),
 				  HA_READ_KEY_EXACT))
     DBUG_RETURN(SP_KEY_NOT_FOUND);
 
@@ -925,7 +924,7 @@ sp_drop_db_routines(THD *thd, char *db)
   table->file->ha_index_init(0, 1);
   if (! table->file->index_read(table->record[0],
                                 (byte *)table->field[MYSQL_PROC_FIELD_DB]->ptr,
-				key_len, HA_READ_KEY_EXACT))
+                                (ulonglong)1, HA_READ_KEY_EXACT))
   {
     int nxtres;
     bool deleted= FALSE;
diff --git a/sql/sql_acl.cc b/sql/sql_acl.cc
index ee7127fcd8d..599f739d0c3 100644
--- a/sql/sql_acl.cc
+++ b/sql/sql_acl.cc
@@ -1813,8 +1813,7 @@ static bool update_user_table(THD *thd, TABLE *table,
            table->key_info->key_length);
 
   if (table->file->index_read_idx(table->record[0], 0,
-				  (byte *) user_key,
-                                  table->key_info->key_length,
+				  (byte *) user_key, ~ULL(0),
 				  HA_READ_KEY_EXACT))
   {
     my_message(ER_PASSWORD_NO_MATCH, ER(ER_PASSWORD_NO_MATCH),
@@ -1905,8 +1904,7 @@ static int replace_user_table(THD *thd, TABLE *table, const LEX_USER &combo,
   key_copy(user_key, table->record[0], table->key_info,
            table->key_info->key_length);
 
-  if (table->file->index_read_idx(table->record[0], 0,
-                                  user_key, table->key_info->key_length,
+  if (table->file->index_read_idx(table->record[0], 0, user_key, ~ULL(0),
                                   HA_READ_KEY_EXACT))
   {
     /* what == 'N' means revoke */
@@ -2123,8 +2121,7 @@ static int replace_db_table(TABLE *table, const char *db,
   key_copy(user_key, table->record[0], table->key_info,
            table->key_info->key_length);
 
-  if (table->file->index_read_idx(table->record[0],0,
-                                  user_key, table->key_info->key_length,
+  if (table->file->index_read_idx(table->record[0],0, user_key, ~ULL(0),
                                   HA_READ_KEY_EXACT))
   {
     if (what == 'N')
@@ -2341,9 +2338,8 @@ GRANT_TABLE::GRANT_TABLE(TABLE *form, TABLE *col_privs)
     col_privs->field[4]->store("",0, &my_charset_latin1);
 
     col_privs->file->ha_index_init(0, 1);
-    if (col_privs->file->index_read(col_privs->record[0],
-                                    (byte*) key,
-                                    key_prefix_len, HA_READ_KEY_EXACT))
+    if (col_privs->file->index_read(col_privs->record[0], (byte*) key,
+                                    (ulonglong)15, HA_READ_KEY_EXACT))
     {
       cols = 0; /* purecov: deadcode */
       col_privs->file->ha_index_end();
@@ -2479,7 +2475,7 @@ static int replace_column_table(GRANT_TABLE *g_t,
   table->field[3]->store(table_name,(uint) strlen(table_name),
                          system_charset_info);
 
-  /* Get length of 3 first key parts */
+  /* Get length of 4 first key parts */
   key_prefix_length= (key_part[0].store_length + key_part[1].store_length +
                       key_part[2].store_length + key_part[3].store_length);
   key_copy(key, table->record[0], table->key_info, key_prefix_length);
@@ -2505,8 +2501,7 @@ static int replace_column_table(GRANT_TABLE *g_t,
     key_copy(user_key, table->record[0], table->key_info,
              table->key_info->key_length);
 
-    if (table->file->index_read(table->record[0], user_key,
-				table->key_info->key_length,
+    if (table->file->index_read(table->record[0], user_key, ~(ulonglong)0,
                                 HA_READ_KEY_EXACT))
     {
       if (revoke_grant)
@@ -2582,8 +2577,7 @@ static int replace_column_table(GRANT_TABLE *g_t,
     key_copy(user_key, table->record[0], table->key_info,
              key_prefix_length);
 
-    if (table->file->index_read(table->record[0], user_key,
-				key_prefix_length,
+    if (table->file->index_read(table->record[0], user_key, (ulonglong)15,
                                 HA_READ_KEY_EXACT))
       goto end;
 
@@ -2684,8 +2678,7 @@ static int replace_table_table(THD *thd, GRANT_TABLE *grant_table,
   key_copy(user_key, table->record[0], table->key_info,
            table->key_info->key_length);
 
-  if (table->file->index_read_idx(table->record[0], 0,
-                                  user_key, table->key_info->key_length,
+  if (table->file->index_read_idx(table->record[0], 0, user_key, ~ULL(0),
 				  HA_READ_KEY_EXACT))
   {
     /*
@@ -2808,8 +2801,8 @@ static int replace_routine_table(THD *thd, GRANT_NAME *grant_name,
                          TRUE);
   store_record(table,record[1]);			// store at pos 1
 
-  if (table->file->index_read_idx(table->record[0],0,
-				  (byte*) table->field[0]->ptr,0,
+  if (table->file->index_read_idx(table->record[0], 0,
+				  (byte*) table->field[0]->ptr, ~ULL(0),
 				  HA_READ_KEY_EXACT))
   {
     /*
@@ -5008,7 +5001,7 @@ static int handle_grant_table(TABLE_LIST *tables, uint table_no, bool drop,
     key_copy(user_key, table->record[0], table->key_info, key_prefix_length);
 
     if ((error= table->file->index_read_idx(table->record[0], 0,
-                                            user_key, key_prefix_length,
+                                            user_key, ULL(3),
                                             HA_READ_KEY_EXACT)))
     {
       if (error != HA_ERR_KEY_NOT_FOUND && error != HA_ERR_END_OF_FILE)
diff --git a/sql/sql_handler.cc b/sql/sql_handler.cc
index 91e61be0478..9c6c98cb151 100644
--- a/sql/sql_handler.cc
+++ b/sql/sql_handler.cc
@@ -515,7 +515,8 @@ bool mysql_ha_read(THD *thd, TABLE_LIST *tables,
       }
       List_iterator<Item> it_ke(*key_expr);
       Item *item;
-      for (key_len=0 ; (item=it_ke++) ; key_part++)
+      ulonglong keypart_map;
+      for (keypart_map= key_len=0 ; (item=it_ke++) ; key_part++)
       {
         my_bitmap_map *old_map;
 	// 'item' can be changed by fix_fields() call
@@ -532,6 +533,7 @@ bool mysql_ha_read(THD *thd, TABLE_LIST *tables,
 	(void) item->save_in_field(key_part->field, 1);
         dbug_tmp_restore_column_map(table->write_set, old_map);
 	key_len+=key_part->store_length;
+        keypart_map= (keypart_map << 1) | 1;
       }
 
       if (!(key= (byte*) thd->calloc(ALIGN_SIZE(key_len))))
@@ -540,7 +542,7 @@ bool mysql_ha_read(THD *thd, TABLE_LIST *tables,
       table->file->ha_index_init(keyno, 1);
       key_copy(key, table->record[0], table->key_info + keyno, key_len);
       error= table->file->index_read(table->record[0],
-                                     key,key_len,ha_rkey_mode);
+                                     key, keypart_map, ha_rkey_mode);
       mode=rkey_to_rnext[(int)ha_rkey_mode];
       break;
     }
diff --git a/sql/sql_help.cc b/sql/sql_help.cc
index b8a904b0c6e..2130e632571 100644
--- a/sql/sql_help.cc
+++ b/sql/sql_help.cc
@@ -295,8 +295,7 @@ int get_topics_for_keyword(THD *thd, TABLE *topics, TABLE *relations,
   rkey_id->store((longlong) key_id, TRUE);
   rkey_id->get_key_image(buff, rkey_id->pack_length(), Field::itRAW);
   int key_res= relations->file->index_read(relations->record[0],
-                                           (byte *) buff,
-                                           rkey_id->pack_length(),
+                                           (byte *) buff, (ulonglong)1,
                                            HA_READ_KEY_EXACT);
 
   for ( ;
@@ -310,7 +309,7 @@ int get_topics_for_keyword(THD *thd, TABLE *topics, TABLE *relations,
     field->get_key_image(topic_id_buff, field->pack_length(), Field::itRAW);
 
     if (!topics->file->index_read(topics->record[0], (byte *)topic_id_buff,
-				  field->pack_length(), HA_READ_KEY_EXACT))
+				  (ulonglong)1, HA_READ_KEY_EXACT))
     {
       memorize_variant_topic(thd,topics,count,find_fields,
 			     names,name,description,example);
diff --git a/sql/sql_insert.cc b/sql/sql_insert.cc
index 479154cbed2..01450a26a62 100644
--- a/sql/sql_insert.cc
+++ b/sql/sql_insert.cc
@@ -1201,9 +1201,7 @@ int write_record(THD *thd, TABLE *table,COPY_INFO *info)
 	}
 	key_copy((byte*) key,table->record[0],table->key_info+key_nr,0);
 	if ((error=(table->file->index_read_idx(table->record[1],key_nr,
-						(byte*) key,
-						table->key_info[key_nr].
-						key_length,
+						(byte*) key, ~ULL(0),
 						HA_READ_KEY_EXACT))))
 	  goto err;
       }
diff --git a/sql/sql_plugin.cc b/sql/sql_plugin.cc
index e3e24c1f375..d7ced748394 100644
--- a/sql/sql_plugin.cc
+++ b/sql/sql_plugin.cc
@@ -943,8 +943,7 @@ my_bool mysql_uninstall_plugin(THD *thd, const LEX_STRING *name)
   table->use_all_columns();
   table->field[0]->store(name->str, name->length, system_charset_info);
   if (! table->file->index_read_idx(table->record[0], 0,
-                                    (byte *)table->field[0]->ptr,
-                                    table->key_info[0].key_length,
+                                    (byte *)table->field[0]->ptr, ~ULL(0),
                                     HA_READ_KEY_EXACT))
   {
     int error;
diff --git a/sql/sql_select.cc b/sql/sql_select.cc
index c901655f2d3..e6b495b7cea 100644
--- a/sql/sql_select.cc
+++ b/sql/sql_select.cc
@@ -10973,7 +10973,8 @@ int safe_index_read(JOIN_TAB *tab)
   TABLE *table= tab->table;
   if ((error=table->file->index_read(table->record[0],
 				     tab->ref.key_buff,
-				     tab->ref.key_length, HA_READ_KEY_EXACT)))
+                                     tab_to_keypart_map(tab),
+                                     HA_READ_KEY_EXACT)))
     return report_error(table, error);
   return 0;
 }
@@ -11111,7 +11112,8 @@ join_read_const(JOIN_TAB *tab)
     {
       error=table->file->index_read_idx(table->record[0],tab->ref.key,
 					(byte*) tab->ref.key_buff,
-					tab->ref.key_length,HA_READ_KEY_EXACT);
+					tab_to_keypart_map(tab),
+                                        HA_READ_KEY_EXACT);
     }
     if (error)
     {
@@ -11154,7 +11156,8 @@ join_read_key(JOIN_TAB *tab)
     }
     error=table->file->index_read(table->record[0],
 				  tab->ref.key_buff,
-				  tab->ref.key_length,HA_READ_KEY_EXACT);
+                                  tab_to_keypart_map(tab),
+                                  HA_READ_KEY_EXACT);
     if (error && error != HA_ERR_KEY_NOT_FOUND && error != HA_ERR_END_OF_FILE)
       return report_error(table, error);
   }
@@ -11182,7 +11185,8 @@ join_read_always_key(JOIN_TAB *tab)
     return -1;
   if ((error=table->file->index_read(table->record[0],
 				     tab->ref.key_buff,
-				     tab->ref.key_length,HA_READ_KEY_EXACT)))
+				     tab_to_keypart_map(tab),
+                                     HA_READ_KEY_EXACT)))
   {
     if (error != HA_ERR_KEY_NOT_FOUND && error != HA_ERR_END_OF_FILE)
       return report_error(table, error);
@@ -11209,7 +11213,7 @@ join_read_last_key(JOIN_TAB *tab)
     return -1;
   if ((error=table->file->index_read_last(table->record[0],
 					  tab->ref.key_buff,
-					  tab->ref.key_length)))
+					  tab_to_keypart_map(tab))))
   {
     if (error != HA_ERR_KEY_NOT_FOUND && error != HA_ERR_END_OF_FILE)
       return report_error(table, error);
@@ -11750,7 +11754,7 @@ end_update(JOIN *join, JOIN_TAB *join_tab __attribute__((unused)),
       group->buff[-1]= (char) group->field->is_null();
   }
   if (!table->file->index_read(table->record[1],
-			       join->tmp_table_param.group_buff,0,
+			       join->tmp_table_param.group_buff, ~(ulonglong)0,
 			       HA_READ_KEY_EXACT))
   {						/* Update old record */
     restore_record(table,record[1]);
diff --git a/sql/sql_select.h b/sql/sql_select.h
index 2e01e54e4ca..e31fe143573 100644
--- a/sql/sql_select.h
+++ b/sql/sql_select.h
@@ -202,6 +202,11 @@ typedef struct st_join_table {
   }
 } JOIN_TAB;
 
+static inline ulonglong tab_to_keypart_map(JOIN_TAB *tab)
+{
+  return (ULL(1) << tab->ref.key_parts) - 1;
+}
+
 enum_nested_loop_state sub_select_cache(JOIN *join, JOIN_TAB *join_tab, bool
                                         end_of_records);
 enum_nested_loop_state sub_select(JOIN *join,JOIN_TAB *join_tab, bool
diff --git a/sql/sql_servers.cc b/sql/sql_servers.cc
index 9a5176d9fe3..1d7c8d5272d 100644
--- a/sql/sql_servers.cc
+++ b/sql/sql_servers.cc
@@ -25,6 +25,7 @@
 #include "sp_head.h"
 #include "sp.h"
 
+static my_bool servers_load(THD *thd, TABLE_LIST *tables);
 HASH servers_cache;
 pthread_mutex_t servers_cache_mutex;                // To init the hash
 uint servers_cache_initialised=FALSE;
@@ -353,8 +354,7 @@ my_bool server_exists_in_table(THD *thd, LEX_SERVER_OPTIONS *server_options)
                          system_charset_info);
 
   if ((error= table->file->index_read_idx(table->record[0], 0,
-                                   (byte *)table->field[0]->ptr,
-                                   table->key_info[0].key_length,
+                                   (byte *)table->field[0]->ptr, ~(ulonglong)0,
                                    HA_READ_KEY_EXACT)))
   {
     if (error != HA_ERR_KEY_NOT_FOUND && error != HA_ERR_END_OF_FILE)
@@ -554,8 +554,7 @@ int insert_server_record(TABLE *table, FOREIGN_SERVER *server)
 
   /* read index until record is that specified in server_name */
   if ((error= table->file->index_read_idx(table->record[0], 0,
-                                   (byte *)table->field[0]->ptr,
-                                   table->key_info[0].key_length,
+                                   (byte *)table->field[0]->ptr, ~(longlong)0,
                                    HA_READ_KEY_EXACT)))
   {
     /* if not found, err */
@@ -873,8 +872,7 @@ int update_server_record(TABLE *table, FOREIGN_SERVER *server)
                          system_charset_info);
 
   if ((error= table->file->index_read_idx(table->record[0], 0,
-                                   (byte *)table->field[0]->ptr,
-                                   table->key_info[0].key_length,
+                                   (byte *)table->field[0]->ptr, ~(longlong)0,
                                    HA_READ_KEY_EXACT)))
   {
     if (error != HA_ERR_KEY_NOT_FOUND && error != HA_ERR_END_OF_FILE)
@@ -928,8 +926,7 @@ int delete_server_record(TABLE *table,
   table->field[0]->store(server_name, server_name_length, system_charset_info);
 
   if ((error= table->file->index_read_idx(table->record[0], 0,
-                                   (byte *)table->field[0]->ptr,
-                                   table->key_info[0].key_length,
+                                   (byte *)table->field[0]->ptr, ~(ulonglong)0,
                                    HA_READ_KEY_EXACT)))
   {
     if (error != HA_ERR_KEY_NOT_FOUND && error != HA_ERR_END_OF_FILE)
diff --git a/sql/sql_udf.cc b/sql/sql_udf.cc
index 7dec58d9b6e..b0e7831465a 100644
--- a/sql/sql_udf.cc
+++ b/sql/sql_udf.cc
@@ -514,8 +514,7 @@ int mysql_drop_function(THD *thd,const LEX_STRING *udf_name)
   table->use_all_columns();
   table->field[0]->store(exact_name_str, exact_name_len, &my_charset_bin);
   if (!table->file->index_read_idx(table->record[0], 0,
-				   (byte*) table->field[0]->ptr,
-				   table->key_info[0].key_length,
+				   (byte*) table->field[0]->ptr, ~ULL(0),
 				   HA_READ_KEY_EXACT))
   {
     int error;
diff --git a/sql/table.cc b/sql/table.cc
index 560f53bae26..5a5d471bce9 100644
--- a/sql/table.cc
+++ b/sql/table.cc
@@ -1222,12 +1222,12 @@ static int open_binary_frm(THD *thd, TABLE_SHARE *share, uchar *head,
     if ((int) (share->next_number_index= (uint)
 	       find_ref_key(share->key_info, share->keys,
                             share->default_values, reg_field,
-			    &share->next_number_key_offset)) < 0)
+			    &share->next_number_key_offset,
+                            &share->next_number_keypart)) < 0)
     {
       /* Wrong field definition */
-      DBUG_ASSERT(0);
-      reg_field->unireg_check= Field::NONE;	/* purecov: inspected */
-      share->found_next_number_field= 0;
+      error= 4;
+      goto err;
     }
     else
       reg_field->flags |= AUTO_INCREMENT_FLAG;
@@ -2244,6 +2244,30 @@ char *get_field(MEM_ROOT *mem, Field *field)
   return to;
 }
 
+/*
+  DESCRIPTION
+    given a buffer with a key value, and a map of keyparts
+    that are present in this value, returns the length of the value
+*/
+uint calculate_key_len(TABLE *table, uint key, const byte *buf,
+                       ulonglong keypart_map)
+{
+  /* works only with key prefixes */
+  DBUG_ASSERT(((keypart_map + 1) & keypart_map) == 0);
+
+  KEY *key_info= table->s->key_info+key;
+  KEY_PART_INFO *key_part= key_info->key_part;
+  KEY_PART_INFO *end_key_part= key_part + key_info->key_parts;
+  uint length= 0;
+
+  while (key_part < end_key_part && keypart_map)
+  {
+    length+= key_part->store_length;
+    keypart_map >>= 1;
+    key_part++;
+  }
+  return length;
+}
 
 /*
   Check if database name is valid
@@ -3938,7 +3962,7 @@ void st_table::mark_auto_increment_column()
   */
   bitmap_set_bit(read_set, found_next_number_field->field_index);
   bitmap_set_bit(write_set, found_next_number_field->field_index);
-  if (s->next_number_key_offset)
+  if (s->next_number_keypart)
     mark_columns_used_by_index_no_reset(s->next_number_index, read_set);
   file->column_bitmaps_signal();
 }
diff --git a/sql/table.h b/sql/table.h
index 54c820d391c..6b8fc84d835 100644
--- a/sql/table.h
+++ b/sql/table.h
@@ -199,6 +199,7 @@ typedef struct st_table_share
   uint primary_key;
   uint next_number_index;
   uint next_number_key_offset;
+  uint next_number_keypart;
   uint error, open_errno, errarg;       /* error from open_table_def() */
   uint column_bitmap_size;
   uchar frm_version;
diff --git a/sql/tztime.cc b/sql/tztime.cc
index 914c7be46de..b9f757faeb8 100644
--- a/sql/tztime.cc
+++ b/sql/tztime.cc
@@ -1917,9 +1917,9 @@ tz_load_from_open_tables(const String *tz_name, TABLE_LIST *tz_tables)
   (void)table->file->ha_index_init(0, 1);
 
   if (table->file->index_read(table->record[0], (byte*)table->field[0]->ptr,
-                              0, HA_READ_KEY_EXACT))
+                              ~(ulonglong)0, HA_READ_KEY_EXACT))
   {
-#ifdef EXTRA_DEBUG    
+#ifdef EXTRA_DEBUG
     /*
       Most probably user has mistyped time zone name, so no need to bark here
       unless we need it for debugging.
@@ -1945,7 +1945,7 @@ tz_load_from_open_tables(const String *tz_name, TABLE_LIST *tz_tables)
   (void)table->file->ha_index_init(0, 1);
 
   if (table->file->index_read(table->record[0], (byte*)table->field[0]->ptr,
-                              0, HA_READ_KEY_EXACT))
+                              ~(ulonglong)0, HA_READ_KEY_EXACT))
   {
     sql_print_error("Can't find description of time zone '%u'", tzid);
     goto end;
@@ -1972,9 +1972,8 @@ tz_load_from_open_tables(const String *tz_name, TABLE_LIST *tz_tables)
   table->field[0]->store((longlong) tzid, TRUE);
   (void)table->file->ha_index_init(0, 1);
 
-  // FIXME Is there any better approach than explicitly specifying 4 ???
   res= table->file->index_read(table->record[0], (byte*)table->field[0]->ptr,
-                               4, HA_READ_KEY_EXACT);
+                               (ulonglong)1, HA_READ_KEY_EXACT);
   while (!res)
   {
     ttid= (uint)table->field[1]->val_int();
@@ -2045,9 +2044,8 @@ tz_load_from_open_tables(const String *tz_name, TABLE_LIST *tz_tables)
   table->field[0]->store((longlong) tzid, TRUE);
   (void)table->file->ha_index_init(0, 1);
 
-  // FIXME Is there any better approach than explicitly specifying 4 ???
   res= table->file->index_read(table->record[0], (byte*)table->field[0]->ptr,
-                               4, HA_READ_KEY_EXACT);
+                               (ulonglong)1, HA_READ_KEY_EXACT);
   while (!res)
   {
     ttime= (my_time_t)table->field[1]->val_int();