Implementation of the MWL#247: Make the optimizer use extended keys.

The main patch.
.

8 files changed, 283 insertions, 49 deletions

diff --git a/sql/mysql_priv.h b/sql/mysql_priv.h
index bccd5b189ff..697a14f3a38 100644
--- a/sql/mysql_priv.h
+++ b/sql/mysql_priv.h
@@ -590,7 +590,8 @@ protected:
 #define OPTIMIZER_SWITCH_JOIN_CACHE_BKA            (1ULL << 23)
 #define OPTIMIZER_SWITCH_OPTIMIZE_JOIN_BUFFER_SIZE (1ULL << 24)
 #define OPTIMIZER_SWITCH_TABLE_ELIMINATION         (1ULL << 25)
-#define OPTIMIZER_SWITCH_LAST                      (1ULL << 26)
+#define OPTIMIZER_SWITCH_EXTENDED_KEYS             (1ULL << 26)
+#define OPTIMIZER_SWITCH_LAST                      (1ULL << 27)
 
 /* The following must be kept in sync with optimizer_switch_str in mysqld.cc */
 /*
diff --git a/sql/mysqld.cc b/sql/mysqld.cc
index be66aa957e9..d6d164fdea9 100644
--- a/sql/mysqld.cc
+++ b/sql/mysqld.cc
@@ -351,6 +351,7 @@ static const char *optimizer_switch_names[]=
   "join_cache_bka",
   "optimize_join_buffer_size",
   "table_elimination",
+  "extended_keys",
   "default", NullS
 };
 
@@ -383,6 +384,7 @@ static const unsigned int optimizer_switch_names_len[]=
   sizeof("join_cache_bka") - 1,
   sizeof("optimize_join_buffer_size") - 1,
   sizeof("table_elimination") - 1,
+  sizeof("extended_keys") - 1,
   sizeof("default") - 1
 };
 TYPELIB optimizer_switch_typelib= { array_elements(optimizer_switch_names)-1,"",
@@ -499,8 +501,8 @@ static const char *optimizer_switch_str="index_merge=on,index_merge_union=on,"
                                         "join_cache_hashed=on,"
                                         "join_cache_bka=on,"
                                         "optimize_join_buffer_size=off,"
-                                        "table_elimination=on";
-                                        ;
+                                        "table_elimination=on,"
+                                        "extended_keys=off";
 #ifdef SAFEMALLOC
 my_bool sf_malloc_trough_check= 0;
 #endif
@@ -7565,7 +7567,7 @@ each time the SQL thread starts.",
    "subquery_cache, outer_join_with_cache, semijoin_with_cache, "
    "join_cache_incremental, join_cache_hashed, join_cache_bka, "
    "optimize_join_buffer_size"
-   ", table_elimination"
+   ", table_elimination, extended_keys"
    "} and val={on, off, default}.",
    &optimizer_switch_str, &optimizer_switch_str, 0, GET_STR, REQUIRED_ARG, 
    /*OPTIMIZER_SWITCH_DEFAULT*/0, 0, 0, 0, 0, 0},
diff --git a/sql/opt_range.cc b/sql/opt_range.cc
index 1226144f5ad..1c45f8fdd14 100644
--- a/sql/opt_range.cc
+++ b/sql/opt_range.cc
@@ -2977,9 +2977,10 @@ int SQL_SELECT::test_quick_select(THD *thd, key_map keys_to_use,
 
     thd->no_errors=1;				// Don't warn about NULL
     init_sql_alloc(&alloc, thd->variables.range_alloc_block_size, 0);
-    if (!(param.key_parts= (KEY_PART*) alloc_root(&alloc,
-                                                  sizeof(KEY_PART)*
-                                                  head->s->key_parts)) ||
+    if (!(param.key_parts=
+           (KEY_PART*) alloc_root(&alloc,
+                                  sizeof(KEY_PART)*
+	                          head->s->actual_n_key_parts(thd))) ||
         fill_used_fields_bitmap(&param))
     {
       thd->no_errors=0;
@@ -2997,6 +2998,8 @@ int SQL_SELECT::test_quick_select(THD *thd, key_map keys_to_use,
     for (idx=0 ; idx < head->s->keys ; idx++, key_info++)
     {
       KEY_PART_INFO *key_part_info;
+      uint n_key_parts= head->actual_n_key_parts(key_info);
+
       if (!keys_to_use.is_set(idx))
 	continue;
       if (key_info->flags & HA_FULLTEXT)
@@ -3004,9 +3007,9 @@ int SQL_SELECT::test_quick_select(THD *thd, key_map keys_to_use,
 
       param.key[param.keys]=key_parts;
       key_part_info= key_info->key_part;
-      for (uint part=0 ; part < key_info->key_parts ;
-	   part++, key_parts++, key_part_info++)
-      {
+      for (uint part= 0 ; part < n_key_parts ; 
+           part++, key_parts++, key_part_info++)
+     {
 	key_parts->key=		 param.keys;
 	key_parts->part=	 part;
 	key_parts->length=       key_part_info->length;
@@ -10017,13 +10020,14 @@ get_quick_select(PARAM *param,uint idx,SEL_ARG *key_tree, uint mrr_flags,
     }
     else
     {
+      KEY *keyinfo= param->table->key_info+param->real_keynr[idx];
       quick->mrr_flags= mrr_flags;
       quick->mrr_buf_size= mrr_buf_size;
       quick->key_parts=(KEY_PART*)
         memdup_root(parent_alloc? parent_alloc : &quick->alloc,
                     (char*) param->key[idx],
                     sizeof(KEY_PART)*
-                    param->table->key_info[param->real_keynr[idx]].key_parts);
+                    param->table->actual_n_key_parts(keyinfo));
     }
   }
   DBUG_RETURN(quick);
@@ -11681,6 +11685,7 @@ get_best_group_min_max(PARAM *param, SEL_TREE *tree)
     KEY_PART_INFO *last_part;
     KEY_PART_INFO *first_non_group_part;
     KEY_PART_INFO *first_non_infix_part;
+    uint key_parts;
     uint key_infix_parts;
     uint cur_group_key_parts= 0;
     uint cur_group_prefix_len= 0;
@@ -11695,7 +11700,8 @@ get_best_group_min_max(PARAM *param, SEL_TREE *tree)
     if (!table->covering_keys.is_set(cur_index))
       goto next_index;
 
-    /*
+     /*
+      Unless extended keys can be used for cur_index:
       If the current storage manager is such that it appends the primary key to
       each index, then the above condition is insufficient to check if the
       index is covering. In such cases it may happen that some fields are
@@ -11704,7 +11710,8 @@ get_best_group_min_max(PARAM *param, SEL_TREE *tree)
       does not qualify as covering in our case. If this is the case, below
       we check that all query fields are indeed covered by 'cur_index'.
     */
-    if (pk < MAX_KEY && cur_index != pk &&
+    if (cur_index_info->key_parts==table->actual_n_key_parts(cur_index_info) &&
+        pk < MAX_KEY && cur_index != pk &&
         (table->file->ha_table_flags() & HA_PRIMARY_KEY_IN_READ_INDEX))
     {
       /* For each table field */
@@ -11721,13 +11728,14 @@ get_best_group_min_max(PARAM *param, SEL_TREE *tree)
       }
     }
 
+
     /*
       Check (GA1) for GROUP BY queries.
     */
     if (join->group_list)
     {
       cur_part= cur_index_info->key_part;
-      end_part= cur_part + cur_index_info->key_parts;
+      end_part= cur_part + table->actual_n_key_parts(cur_index_info);
       /* Iterate in parallel over the GROUP list and the index parts. */
       for (tmp_group= join->group_list; tmp_group && (cur_part != end_part);
            tmp_group= tmp_group->next, cur_part++)
@@ -11820,10 +11828,11 @@ get_best_group_min_max(PARAM *param, SEL_TREE *tree)
       If there is no MIN/MAX, the keyparts after the last group part can be
       referenced only in equalities with constants, and the referenced keyparts
       must form a sequence without any gaps that starts immediately after the
-      last group keypart.
+      last group keyparyt.
     */
-    last_part= cur_index_info->key_part + cur_index_info->key_parts;
-    first_non_group_part= (cur_group_key_parts < cur_index_info->key_parts) ?
+    key_parts= table->actual_n_key_parts(cur_index_info);
+    last_part= cur_index_info->key_part + key_parts;
+    first_non_group_part= (cur_group_key_parts < key_parts) ?
                           cur_index_info->key_part + cur_group_key_parts :
                           NULL;
     first_non_infix_part= min_max_arg_part ?
@@ -12273,7 +12282,8 @@ get_field_keypart(KEY *index, Field *field)
 {
   KEY_PART_INFO *part, *end;
 
-  for (part= index->key_part, end= part + index->key_parts; part < end; part++)
+  for (part= index->key_part, end= part+field->table->actual_n_key_parts(index);
+       part < end; part++)
   {
     if (field->eq(part->field))
       return part - index->key_part + 1;
diff --git a/sql/opt_sum.cc b/sql/opt_sum.cc
index 7b4c48497fa..e6fb28f73b3 100644
--- a/sql/opt_sum.cc
+++ b/sql/opt_sum.cc
@@ -844,7 +844,6 @@ static bool matching_cond(bool max_fl, TABLE_REF *ref, KEY *keyinfo,
     1   Can use key to optimize MIN()/MAX().
     In this case ref, range_fl and prefix_len are updated
 */
-
       
 static bool find_key_for_maxmin(bool max_fl, TABLE_REF *ref,
                                 Field* field, COND *cond,
@@ -873,7 +872,8 @@ static bool find_key_for_maxmin(bool max_fl, TABLE_REF *ref,
       continue;
     uint jdx= 0;
     *prefix_len= 0;
-    for (part= keyinfo->key_part, part_end= part+keyinfo->key_parts ;
+    part_end= keyinfo->key_part+table->actual_n_key_parts(keyinfo);
+    for (part= keyinfo->key_part ;
          part != part_end ;
          part++, jdx++, key_part_to_use= (key_part_to_use << 1) | 1)
     {
diff --git a/sql/sql_select.cc b/sql/sql_select.cc
index 0252a00dbe3..9284804432f 100644
--- a/sql/sql_select.cc
+++ b/sql/sql_select.cc
@@ -3339,11 +3339,13 @@ make_join_statistics(JOIN *join, List<TABLE_LIST> &tables_list,
             The effect of this is that we don't do const substitution for
             such tables.
           */
-	  if (eq_part.is_prefix(table->key_info[key].key_parts) &&
+          KEY *keyinfo= table->key_info+key;
+          uint  key_parts= table->actual_n_key_parts(keyinfo);
+          if (eq_part.is_prefix(key_parts) &&
               !table->fulltext_searched && 
               (!embedding || (embedding->sj_on_expr && !embedding->embedding)))
 	  {
-            if (table->key_info[key].flags & HA_NOSAME)
+            if (table->actual_key_flags(keyinfo) & HA_NOSAME)
             {
 	      if (const_ref == eq_part &&
                   !has_expensive_keyparts &&
@@ -4367,7 +4369,8 @@ add_key_part(DYNAMIC_ARRAY *keyuse_array, KEY_FIELD *key_field)
       if (form->key_info[key].flags & (HA_FULLTEXT | HA_SPATIAL))
 	continue;    // ToDo: ft-keys in non-ft queries.   SerG
 
-      uint key_parts= (uint) form->key_info[key].key_parts;
+      KEY *keyinfo= form->key_info+key;
+      uint key_parts= form->actual_n_key_parts(keyinfo);
       for (uint part=0 ; part <  key_parts ; part++)
       {
 	if (field->eq(form->key_info[key].key_part[part].field))
@@ -4670,7 +4673,7 @@ update_ref_and_keys(THD *thd, DYNAMIC_ARRAY *keyuse,JOIN_TAB *join_tab,
   /* fill keyuse with found key parts */
   for ( ; field != end ; field++)
   {
-    if (add_key_part(keyuse,field))
+    if (add_key_part(keyuse, field))
       return TRUE;
   }
 
@@ -4997,6 +5000,8 @@ best_access_path(JOIN      *join,
     for (keyuse=s->keyuse ; keyuse->table == table ;)
     {
       KEY *keyinfo;
+      ulong key_flags;
+      uint key_parts;
       key_part_map found_part= 0;
       table_map found_ref= 0;
       uint key= keyuse->key;
@@ -5023,6 +5028,8 @@ best_access_path(JOIN      *join,
       }
 
       keyinfo= table->key_info+key;
+      key_parts= table->actual_n_key_parts(keyinfo);
+      key_flags= table->actual_key_flags(keyinfo);
 
       /* Calculate how many key segments of the current key we can use */
       start_key= keyuse;
@@ -5101,11 +5108,11 @@ best_access_path(JOIN      *join,
         loose_scan_opt.check_ref_access_part1(s, key, start_key, found_part);
 
         /* Check if we found full key */
-        if (found_part == PREV_BITS(uint,keyinfo->key_parts) &&
+        if (found_part == PREV_BITS(uint, key_parts) &&
             !ref_or_null_part)
         {                                         /* use eq key */
           max_key_part= (uint) ~0;
-          if ((keyinfo->flags & (HA_NOSAME | HA_NULL_PART_KEY)) == HA_NOSAME)
+          if ((key_flags & (HA_NOSAME | HA_NULL_PART_KEY)) == HA_NOSAME)
           {
             tmp = prev_record_reads(join->positions, idx, found_ref);
             records=1.0;
@@ -7196,6 +7203,7 @@ static bool create_hj_key_for_table(JOIN *join, JOIN_TAB *join_tab,
                                                      key_parts)))
     DBUG_RETURN(TRUE);
   keyinfo->usable_key_parts= keyinfo->key_parts = key_parts;
+  keyinfo->ext_key_parts= keyinfo->key_parts;
   keyinfo->key_part= key_part_info;
   keyinfo->key_length=0;
   keyinfo->algorithm= HA_KEY_ALG_UNDEF;
@@ -7403,8 +7411,11 @@ static bool create_ref_for_key(JOIN *join, JOIN_TAB *j,
     DBUG_RETURN(0);
   if (j->type == JT_CONST)
     j->table->const_table= 1;
-  else if (((keyinfo->flags & (HA_NOSAME | HA_NULL_PART_KEY)) != HA_NOSAME) ||
-	   keyparts != keyinfo->key_parts || null_ref_key)
+  else if (((j->table->actual_key_flags(keyinfo) &
+            (HA_NOSAME | HA_NULL_PART_KEY))
+           != HA_NOSAME) ||
+	   keyparts != j->table->actual_n_key_parts(keyinfo) ||
+           null_ref_key)
   {
     /* Must read with repeat */
     j->type= null_ref_key ? JT_REF_OR_NULL : JT_REF;
diff --git a/sql/structs.h b/sql/structs.h
index 18f90d5b8b1..be53a88a0cc 100644
--- a/sql/structs.h
+++ b/sql/structs.h
@@ -77,6 +77,9 @@ typedef struct st_key {
   ulong flags;                          /* dupp key and pack flags */
   uint	key_parts;			/* How many key_parts */
   uint	usable_key_parts;		/* Should normally be = key_parts */
+  uint ext_key_parts;              /* Number of key parts in extended key */
+  ulong ext_key_flags;             /* Flags for extended key              */
+  key_part_map ext_key_part_map;   /* Bitmap of pk key parts in extension */ 
   uint  block_size;
   uint  name_length;
   enum  ha_key_alg algorithm;
diff --git a/sql/table.cc b/sql/table.cc
index d54135b5620..8216ccf980d 100644
--- a/sql/table.cc
+++ b/sql/table.cc
@@ -705,6 +705,13 @@ static int open_binary_frm(THD *thd, TABLE_SHARE *share, uchar *head,
   enum legacy_db_type legacy_db_type;
   my_bitmap_map *bitmaps;
   bool null_bits_are_used;
+  KEY first_keyinfo;
+  uint len;
+  KEY_PART_INFO *first_key_part= NULL;
+  uint ext_key_parts= 0;
+  uint first_key_parts= 0;
+  keyinfo= &first_keyinfo;
+  share->ext_key_parts= 0;
   DBUG_ENTER("open_binary_frm");
 
   LINT_INIT(options);
@@ -798,18 +805,28 @@ static int open_binary_frm(THD *thd, TABLE_SHARE *share, uchar *head,
   share->keys_for_keyread.init(0);
   share->keys_in_use.init(keys);
 
-  n_length=keys*sizeof(KEY)+key_parts*sizeof(KEY_PART_INFO);
-  if (!(keyinfo = (KEY*) alloc_root(&share->mem_root,
-				    n_length + uint2korr(disk_buff+4))))
-    goto err;                                   /* purecov: inspected */
-  bzero((char*) keyinfo,n_length);
-  share->key_info= keyinfo;
-  key_part= my_reinterpret_cast(KEY_PART_INFO*) (keyinfo+keys);
-  strpos=disk_buff+6;
+  /* Currently only InnoDB can use extended keys */
+  share->set_use_ext_keys_flag(legacy_db_type == DB_TYPE_INNODB);
 
-  if (!(rec_per_key= (ulong*) alloc_root(&share->mem_root,
-                                         sizeof(ulong)*key_parts)))
-    goto err;
+  len= (uint) uint2korr(disk_buff+4);
+  if (!keys)
+  {  
+    if (!(keyinfo = (KEY*) alloc_root(&share->mem_root, len)))
+      goto err;
+    bzero((char*) keyinfo, len);
+    key_part= my_reinterpret_cast(KEY_PART_INFO*) (keyinfo+keys);
+  }
+  strpos= disk_buff+6;
+
+  /*
+    If share->use_ext_keys is set to TRUE we assume that any key
+    can be extended by the components of the primary key whose
+    definition is read first from the frm file.
+    For each key only those fields of the assumed primary key are
+    added that are not included in the proper key definition. 
+    If after all it turns out that there is no primary key the
+    added components are removed from each key.
+  */   
 
   for (i=0 ; i < keys ; i++, keyinfo++)
   {
@@ -831,6 +848,32 @@ static int open_binary_frm(THD *thd, TABLE_SHARE *share, uchar *head,
       strpos+=4;
     }
 
+    if (i == 0)
+    {
+      ext_key_parts= key_parts +
+	             (share->use_ext_keys ? first_keyinfo.key_parts*(keys-1) : 0); 
+    
+      n_length=keys*sizeof(KEY)+ext_key_parts*sizeof(KEY_PART_INFO);
+      if (!(keyinfo= (KEY*) alloc_root(&share->mem_root,
+				       n_length + len)))
+        goto err;                                   /* purecov: inspected */
+      bzero((char*) keyinfo,n_length);
+      share->key_info= keyinfo;
+      key_part= my_reinterpret_cast(KEY_PART_INFO*) (keyinfo+keys);
+
+      if (!(rec_per_key= (ulong*) alloc_root(&share->mem_root,
+                                             sizeof(ulong)*ext_key_parts)))
+        goto err;
+      first_key_part= key_part;
+      first_key_parts= first_keyinfo.key_parts;
+      keyinfo->flags= first_keyinfo.flags;
+      keyinfo->key_length= first_keyinfo.key_length;
+      keyinfo->key_parts= first_keyinfo.key_parts;
+      keyinfo->algorithm= first_keyinfo.algorithm;
+      if (new_frm_ver >= 3)
+        keyinfo->block_size= first_keyinfo.block_size;
+    }
+
     keyinfo->key_part=	 key_part;
     keyinfo->rec_per_key= rec_per_key;
     for (j=keyinfo->key_parts ; j-- ; key_part++)
@@ -859,6 +902,31 @@ static int open_binary_frm(THD *thd, TABLE_SHARE *share, uchar *head,
       }
       key_part->store_length=key_part->length;
     }
+    keyinfo->ext_key_parts= keyinfo->key_parts;
+    if (share->use_ext_keys && i)
+    {
+      keyinfo->ext_key_flags= keyinfo->flags | HA_NOSAME;
+      keyinfo->ext_key_part_map= 0;
+      for (j= 0; j<first_key_parts && keyinfo->ext_key_parts<MAX_REF_PARTS; j++)
+      {
+        uint key_parts= keyinfo->key_parts;
+        KEY_PART_INFO* curr_key_part= keyinfo->key_part;
+        KEY_PART_INFO* curr_key_part_end= curr_key_part+key_parts;
+        for ( ; curr_key_part < curr_key_part_end; curr_key_part++)
+        {
+          if (curr_key_part->fieldnr == first_key_part[j].fieldnr)
+            break;
+        }
+        if (curr_key_part == curr_key_part_end)
+        {
+          *key_part++= first_key_part[j];
+          *rec_per_key++= 0;
+          keyinfo->ext_key_parts++;
+          keyinfo->ext_key_part_map|= 1 << j;
+        }
+      } 
+    }
+    share->ext_key_parts+= keyinfo->ext_key_parts;  
   }
   keynames=(char*) key_part;
   strpos+= (strmov(keynames, (char *) strpos) - keynames)+1;
@@ -1421,11 +1489,38 @@ static int open_binary_frm(THD *thd, TABLE_SHARE *share, uchar *head,
   /* Fix key->name and key_part->field */
   if (key_parts)
   {
+    uint add_first_key_parts= 0;
     uint primary_key=(uint) (find_type((char*) primary_key_name,
 				       &share->keynames, 3) - 1);
     longlong ha_option= handler_file->ha_table_flags();
     keyinfo= share->key_info;
-    key_part= keyinfo->key_part;
+
+    if (share->use_ext_keys)
+    { 
+      if (primary_key >= MAX_KEY)
+      {
+        add_first_key_parts= 0;
+        share->set_use_ext_keys_flag(FALSE);
+      }
+      else
+      {
+        add_first_key_parts= first_key_parts;
+        /* 
+          Do not add components of the primary key starting from
+          the major component defined over the beginning of a field.
+	*/
+	for (i= 0; i < first_key_parts; i++)
+	{
+          uint fieldnr= keyinfo[0].key_part[i].fieldnr;
+          if (share->field[fieldnr-1]->key_length() !=
+              keyinfo[0].key_part[i].length)
+	  {
+            add_first_key_parts= i;
+            break;
+          }
+        }
+      }   
+    }
 
     for (uint key=0 ; key < share->keys ; key++,keyinfo++)
     {
@@ -1444,6 +1539,51 @@ static int open_binary_frm(THD *thd, TABLE_SHARE *share, uchar *head,
                keyinfo->name_length+1);
       }
 
+      if (ext_key_parts > share->key_parts && key)
+      {
+	KEY_PART_INFO *new_key_part= (keyinfo-1)->key_part+
+                                     (keyinfo-1)->ext_key_parts;
+
+        /* 
+          Do not extend the key that contains a component
+          defined over the beginning of a field.
+	*/ 
+        for (i= 0; i < keyinfo->key_parts; i++)
+	{
+          uint fieldnr= keyinfo->key_part[i].fieldnr;
+          if (share->field[fieldnr-1]->key_length() !=
+              keyinfo->key_part[i].length)
+	  {
+            add_first_key_parts= 0;
+            break;
+          }
+        }
+
+        if (add_first_key_parts < keyinfo->ext_key_parts-keyinfo->key_parts)
+	{
+          share->ext_key_parts-= keyinfo->ext_key_parts;
+          key_part_map ext_key_part_map= keyinfo->ext_key_part_map;
+          keyinfo->ext_key_parts= keyinfo->key_parts;
+          keyinfo->ext_key_flags= keyinfo->flags;
+	  keyinfo->ext_key_part_map= 0; 
+          for (i= 0; i < add_first_key_parts; i++)
+	  {
+            if (ext_key_part_map & 1<<i)
+	    {
+              keyinfo->ext_key_part_map|= 1<<i;
+	      keyinfo->ext_key_parts++;
+            }
+          }
+          share->ext_key_parts+= keyinfo->ext_key_parts;
+        }
+        if (new_key_part != keyinfo->key_part)
+	{
+          memmove(new_key_part, keyinfo->key_part,
+                  sizeof(KEY_PART_INFO) * keyinfo->ext_key_parts);
+          keyinfo->key_part= new_key_part;
+        }
+      }
+    
       /* Fix fulltext keys for old .frm files */
       if (share->key_info[key].flags & HA_FULLTEXT)
 	share->key_info[key].algorithm= HA_KEY_ALG_FULLTEXT;
@@ -1455,6 +1595,7 @@ static int open_binary_frm(THD *thd, TABLE_SHARE *share, uchar *head,
 	  declare this as a primary key.
 	*/
 	primary_key=key;
+        key_part= keyinfo->key_part;
 	for (i=0 ; i < keyinfo->key_parts ;i++)
 	{
 	  uint fieldnr= key_part[i].fieldnr;
@@ -1469,7 +1610,10 @@ static int open_binary_frm(THD *thd, TABLE_SHARE *share, uchar *head,
 	}
       }
 
-      for (i=0 ; i < keyinfo->key_parts ; key_part++,i++)
+      key_part= keyinfo->key_part;
+      uint key_parts= share->use_ext_keys ? keyinfo->ext_key_parts :
+	                                    keyinfo->key_parts;
+      for (i=0; i < key_parts; key_part++, i++)
       {
         Field *field;
 	if (new_field_pack_flag <= 1)
@@ -1521,7 +1665,8 @@ static int open_binary_frm(THD *thd, TABLE_SHARE *share, uchar *head,
           {
             share->keys_for_keyread.set_bit(key);
             field->part_of_key.set_bit(key);
-            field->part_of_key_not_clustered.set_bit(key);
+            if (i < keyinfo->key_parts)
+              field->part_of_key_not_clustered.set_bit(key);
           }
           if (handler_file->index_flags(key, i, 1) & HA_READ_ORDER)
             field->part_of_sortkey.set_bit(key);
@@ -2196,15 +2341,15 @@ int open_table_from_share(THD *thd, TABLE_SHARE *share, const char *alias,
     KEY	*key_info, *key_info_end;
     KEY_PART_INFO *key_part;
     uint n_length;
-    n_length= share->keys*sizeof(KEY) + share->key_parts*sizeof(KEY_PART_INFO);
+    n_length= share->keys*sizeof(KEY) + share->ext_key_parts*sizeof(KEY_PART_INFO);
     if (!(key_info= (KEY*) alloc_root(&outparam->mem_root, n_length)))
       goto err;
     outparam->key_info= key_info;
     key_part= (my_reinterpret_cast(KEY_PART_INFO*) (key_info+share->keys));
-    
+
     memcpy(key_info, share->key_info, sizeof(*key_info)*share->keys);
     memcpy(key_part, share->key_info[0].key_part, (sizeof(*key_part) *
-                                                   share->key_parts));
+                                                   share->ext_key_parts));
 
     for (key_info_end= key_info + share->keys ;
          key_info < key_info_end ;
@@ -2215,9 +2360,9 @@ int open_table_from_share(THD *thd, TABLE_SHARE *share, const char *alias,
       key_info->table= outparam;
       key_info->key_part= key_part;
 
-      for (key_part_end= key_part+ key_info->key_parts ;
-           key_part < key_part_end ;
-           key_part++)
+      key_part_end= key_part + (share->use_ext_keys ? key_info->ext_key_parts :
+			                              key_info->key_parts) ;      
+      for ( ; key_part < key_part_end; key_part++)
       {
         Field *field= key_part->field= outparam->field[key_part->fieldnr-1];
 
@@ -2233,6 +2378,8 @@ int open_table_from_share(THD *thd, TABLE_SHARE *share, const char *alias,
           field->field_length= key_part->length;
         }
       }
+      if (!share->use_ext_keys)
+	key_part+= key_info->ext_key_parts-key_info->key_parts;
     }
   }
 
@@ -3138,7 +3285,7 @@ uint calculate_key_len(TABLE *table, uint key, const uchar *buf,
 
   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;
+  KEY_PART_INFO *end_key_part= key_part + table->actual_n_key_parts(key_info);
   uint length= 0;
 
   while (key_part < end_key_part && keypart_map)
@@ -5414,6 +5561,47 @@ bool st_table::is_filled_at_execution()
 }
 
 
+/**
+  @brief
+  Get actual number of key components
+
+  @param keyinfo
+
+  @details
+  The function calculates actual number of key components, possibly including
+  components of extended keys, taken into consideration by the optimizer for the
+  key described by the parameter keyinfo.
+
+  @return number of considered key components
+*/ 
+
+inline uint st_table::actual_n_key_parts(KEY *keyinfo)
+{
+  return optimizer_flag(in_use, OPTIMIZER_SWITCH_EXTENDED_KEYS) ?
+           keyinfo->ext_key_parts : keyinfo->key_parts;
+}
+
+ 
+/**
+  @brief
+  Get actual key flags for a table key 
+
+  @param keyinfo
+
+  @details
+  The function finds out actual key flags taken into consideration by the
+  optimizer for the key described by the parameter keyinfo.
+
+  @return actual key flags
+*/ 
+
+ulong st_table::actual_key_flags(KEY *keyinfo)
+{
+  return optimizer_flag(in_use, OPTIMIZER_SWITCH_EXTENDED_KEYS) ?
+           keyinfo->ext_key_flags : keyinfo->flags;
+} 
+
+
 /*
   Cleanup this table for re-execution.
 
@@ -6007,6 +6195,14 @@ bool TABLE_LIST::change_refs_to_fields()
 }
 
 
+uint TABLE_SHARE::actual_n_key_parts(THD *thd)
+{
+  return use_ext_keys &&
+         optimizer_flag(thd, OPTIMIZER_SWITCH_EXTENDED_KEYS) ?
+           ext_key_parts : key_parts;
+}  
+
+
 /*****************************************************************************
 ** Instansiate templates
 *****************************************************************************/
diff --git a/sql/table.h b/sql/table.h
index 5dd464ac876..1f3369a4ea5 100644
--- a/sql/table.h
+++ b/sql/table.h
@@ -418,6 +418,7 @@ typedef struct st_table_share
   uint stored_fields;                   
   uint rec_buff_length;                 /* Size of table->record[] buffer */
   uint keys, key_parts;
+  uint ext_key_parts;       /* Total number of key parts in extended keys */
   uint max_key_length, max_unique_length, total_key_length;
   uint uniques;                         /* Number of UNIQUE index */
   uint null_fields;			/* number of null fields */
@@ -438,6 +439,7 @@ typedef struct st_table_share
   uint column_bitmap_size;
   uchar frm_version;
   uint vfields;                         /* Number of computed (virtual) fields */
+  bool use_ext_keys;                    /* Extended keys can be used */
   bool null_field_first;
   bool system;                          /* Set if system table (one record) */
   bool crypted;                         /* If .frm file is crypted */
@@ -652,6 +654,13 @@ typedef struct st_table_share
     return (tmp_table == SYSTEM_TMP_TABLE || is_view) ? 0 : table_map_id;
   }
 
+  void set_use_ext_keys_flag(bool fl) 
+  {
+    use_ext_keys= fl;
+  }
+  
+  uint actual_n_key_parts(THD *thd);
+
 } TABLE_SHARE;
 
 
@@ -991,6 +1000,8 @@ struct st_table {
     }
     DBUG_VOID_RETURN;
   }
+  uint actual_n_key_parts(KEY *keyinfo);
+  ulong actual_key_flags(KEY *keyinfo);
 };
 
 enum enum_schema_table_state