bk commit into 5.1 tree (mskold:1.2039)

[Martin Skold]

Below is the list of changes that have just been committed into a local
5.1 repository of marty. When marty does a push these changes will
be propagated to the main repository and, within 24 hours after the
push, to the public repository.
For information on how to access the public repository
see http://dev.mysql.com/doc/mysql/en/installing-source-tree.html

ChangeSet
  1.2039 05/10/12 13:26:53 mskold@stripped +7 -0
  Merge mysql.com:/usr/local/home/marty/MySQL/mysql-5.1-opt
  into  mysql.com:/usr/local/home/marty/MySQL/mysql-5.1-wl2604

  storage/ndb/src/ndbapi/NdbDictionaryImpl.hpp
    1.53 05/10/12 13:26:45 mskold@stripped +0 -2
    merge

  storage/ndb/src/ndbapi/NdbDictionaryImpl.cpp
    1.111 05/10/12 13:26:45 mskold@stripped +32 -0
    merge

  sql/ha_ndbcluster.h
    1.86 05/10/12 13:26:44 mskold@stripped +0 -4
    merge

  sql/ha_ndbcluster.cc
    1.165 05/10/12 13:26:44 mskold@stripped +207 -1
    merge

  storage/ndb/src/ndbapi/NdbDictionary.cpp
    1.58 05/10/12 13:15:56 mskold@stripped +0 -0
    Auto merged

  storage/ndb/src/kernel/blocks/dbdict/Dbdict.hpp
    1.46 05/10/12 13:15:56 mskold@stripped +0 -0
    Auto merged

  storage/ndb/src/kernel/blocks/dbdict/Dbdict.cpp
    1.109 05/10/12 13:15:56 mskold@stripped +0 -0
    Auto merged

# This is a BitKeeper patch.  What follows are the unified diffs for the
# set of deltas contained in the patch.  The rest of the patch, the part
# that BitKeeper cares about, is below these diffs.
# User:	mskold
# Host:	blowfish.ndb.mysql.com
# Root:	/usr/local/home/marty/MySQL/mysql-5.1-wl2604/RESYNC

--- 1.110/storage/ndb/src/ndbapi/NdbDictionaryImpl.cpp	2005-10-06 05:47:38 +02:00
+++ 1.111/storage/ndb/src/ndbapi/NdbDictionaryImpl.cpp	2005-10-12 13:26:45 +02:00
@@ -506,6 +506,37 @@
 }
 
 void
+NdbTableImpl::updateMysqlName()
+{
+  Vector<BaseString> v;
+  if (m_internalName.split(v,"/") == 3)
+  {
+    m_mysqlName.assfmt("%s/%s",v[0].c_str(),v[2].c_str());
+    return;
+  }
+  m_mysqlName.assign("");
+}
+void NdbTableImpl::setFrm(const void* data, Uint32 len)
+{
+  m_newFrm.assign(data, len);
+}
+
+const void * 
+NdbTableImpl::getFrmData() const
+{
+  if (m_newFrm.empty())
+    return m_frm.get_data();
+  else
+    return m_newFrm.get_data();
+}
+
+Uint32
+NdbTableImpl::getFrmLength() const 
+{
+  if (m_newFrm.empty())
+    return m_frm.length();
+  else
+    return m_newFrm.length();
 NdbTableImpl::computeAggregates()
 {
   m_noOfKeys = 0;

--- 1.164/sql/ha_ndbcluster.cc	2005-10-07 20:32:56 +02:00
+++ 1.165/sql/ha_ndbcluster.cc	2005-10-12 13:26:44 +02:00
@@ -1005,11 +1005,9 @@
 
 int ha_ndbcluster::build_index_list(Ndb *ndb, TABLE *tab, enum ILBP phase)
 {
-  uint i;
+  uint i, free;
   int error= 0;
   const char *index_name;
-  char unique_index_name[FN_LEN];
-  static const char* unique_suffix= "$unique";
   KEY* key_info= tab->key_info;
   const char **key_name= tab->s->keynames.type_names;
   NDBDICT *dict= ndb->getDictionary();
@@ -1021,49 +1019,76 @@
     index_name= *key_name;
     NDB_INDEX_TYPE idx_type= get_index_type_from_table(i);
     m_index[i].type= idx_type;
-    if (idx_type == UNIQUE_ORDERED_INDEX || idx_type == UNIQUE_INDEX)
-    {
-      strxnmov(unique_index_name, FN_LEN, index_name, unique_suffix, NullS);
-      DBUG_PRINT("info", ("Created unique index name \'%s\' for index %d",
-                          unique_index_name, i));
-    }
     // Create secondary indexes if in create phase
     if (phase == ILBP_CREATE)
     {
-      DBUG_PRINT("info", ("Creating index %u: %s", i, index_name));      
-      switch (idx_type){
-        
-      case PRIMARY_KEY_INDEX:
-        // Do nothing, already created
-        break;
-      case PRIMARY_KEY_ORDERED_INDEX:
-        error= create_ordered_index(index_name, key_info);
-        break;
-      case UNIQUE_ORDERED_INDEX:
-        if (!(error= create_ordered_index(index_name, key_info)))
-          error= create_unique_index(unique_index_name, key_info);
-        break;
-      case UNIQUE_INDEX:
-        if (!(error= check_index_fields_not_null(i)))
-          error= create_unique_index(unique_index_name, key_info);
-        break;
-      case ORDERED_INDEX:
-        error= create_ordered_index(index_name, key_info);
-        break;
-      default:
-        DBUG_ASSERT(FALSE);
-        break;
-      }
+      error= create_index(index_name, key_info, idx_type, i);
       if (error)
       {
         DBUG_PRINT("error", ("Failed to create index %u", i));
         drop_table();
         break;
       }
+      m_index[i].status= CREATED;
+    }
+    // Drop references to dropped indexes
+    if (phase == ILBP_DROP && m_index[i].status == TO_BE_DROPPED)
+    {
+      NdbDictionary::Index *index= 
+        (NdbDictionary::Index *) m_index[i].index;
+      NdbDictionary::Index *unique_index= 
+        (NdbDictionary::Index *) m_index[i].unique_index;
+      
+      if (index)
+      {
+        index_name= index->getName();
+        DBUG_PRINT("info", ("Dropping index %u: %s", i, index_name));  
+        // Drop ordered index from ndb
+        error= drop_index(index_name);
+      }
+      if (!error)
+        m_index[i].index= NULL;
+      if (!error && unique_index)
+      {
+        index_name= index->getName();
+        DBUG_PRINT("info", ("Dropping index %u: %s", i, index_name));
+        // Drop unique index from ndb
+        error= drop_index(index_name);
+      }
+      if (error)
+        DBUG_RETURN(error);
+      m_index[i].type= UNDEFINED_INDEX;
+      m_index[i].status= UNDEFINED;
+      m_index[i].unique_index= NULL;
+      m_index[i].unique_index_attrid_map= NULL;
+      m_index[i].index_stat=NULL;
+      m_index[i].index_stat_cache_entries=0;
+      m_index[i].index_stat_update_freq=0;
+      m_index[i].index_stat_query_count=0;
+      continue;
+    }
+    else if (phase == ILBP_RENUMBER && m_index[i].status == TO_BE_DROPPED) 
+    {
+      DBUG_PRINT("info", ("Shifting index %s out of the list", index_name));
+      NDB_INDEX_DATA tmp;
+      uint j= i + 1;
+      // Shift index out of list
+      while(j != MAX_KEY && m_index[j].status != UNDEFINED)
+      {
+        tmp=  m_index[j - 1];
+        m_index[j - 1]= m_index[j];
+        m_index[j]= tmp;
+        j++;
+      }
+      continue;
     }
+    // else if (phase == ILBP_OPEN || phase == ILBP_ADD)
     // Add handles to index objects
     if (idx_type != PRIMARY_KEY_INDEX && idx_type != UNIQUE_INDEX)
     {
+      if (phase == ILBP_ADD && m_index[i].index)
+        // Index already added
+        continue;
       DBUG_PRINT("info", ("Get handle to index %s", index_name));
       const NDBINDEX *index= dict->getIndex(index_name, m_tabname);
       if (!index) DBUG_RETURN(1);
@@ -1091,12 +1116,20 @@
     }
     if (idx_type == UNIQUE_ORDERED_INDEX || idx_type == UNIQUE_INDEX)
     {
+      char unique_index_name[FN_LEN];
+      static const char* unique_suffix= "$unique";
+      if (phase == ILBP_ADD && m_index[i].unique_index)
+        // Index already added
+        continue;
+      strxnmov(unique_index_name, FN_LEN, index_name, unique_suffix, NullS);
       DBUG_PRINT("info", ("Get handle to unique_index %s", unique_index_name));
       const NDBINDEX *index= dict->getIndex(unique_index_name, m_tabname);
       if (!index) DBUG_RETURN(1);
       m_index[i].unique_index= (void *) index;
       error= fix_unique_index_attr_order(m_index[i], index, key_info);
     }
+    if (!error)
+      m_index[i].status= ACTIVE;
   }
   
   DBUG_RETURN(error);
@@ -1109,11 +1142,17 @@
 */
 NDB_INDEX_TYPE ha_ndbcluster::get_index_type_from_table(uint inx) const
 {
-  bool is_hash_index=  (table->key_info[inx].algorithm == HA_KEY_ALG_HASH);
+  return get_index_type_from_key(inx, table->key_info);
+}
+
+NDB_INDEX_TYPE ha_ndbcluster::get_index_type_from_key(uint inx,
+                                                      KEY *key_info) const
+{
+  bool is_hash_index=  (key_info[inx].algorithm == HA_KEY_ALG_HASH);
   if (inx == table->s->primary_key)
     return is_hash_index ? PRIMARY_KEY_INDEX : PRIMARY_KEY_ORDERED_INDEX;
-
-  return ((table->key_info[inx].flags & HA_NOSAME) ? 
+  
+  return ((key_info[inx].flags & HA_NOSAME) ? 
           (is_hash_index ? UNIQUE_INDEX : UNIQUE_ORDERED_INDEX) :
           ORDERED_INDEX);
 } 
@@ -3943,12 +3982,53 @@
   DBUG_RETURN(my_errno);
 }
 
+int ha_ndbcluster::create_index(const char *name, KEY *key_info, 
+                                NDB_INDEX_TYPE idx_type, uint idx_no)
+{
+  int error= 0;
+  char unique_name[FN_LEN];
+  static const char* unique_suffix= "$unique";
+  DBUG_ENTER("ha_ndbcluster::create_ordered_index");
+  DBUG_PRINT("info", ("Creating index %u: %s", idx_no, name));  
+
+  if (idx_type == UNIQUE_ORDERED_INDEX || idx_type == UNIQUE_INDEX)
+  {
+    strxnmov(unique_name, FN_LEN, name, unique_suffix, NullS);
+    DBUG_PRINT("info", ("Created unique index name \'%s\' for index %d",
+                        unique_name, idx_no));
+  }
+    
+  switch (idx_type){
+  case PRIMARY_KEY_INDEX:
+    // Do nothing, already created
+    break;
+  case PRIMARY_KEY_ORDERED_INDEX:
+    error= create_ordered_index(name, key_info);
+    break;
+  case UNIQUE_ORDERED_INDEX:
+    if (!(error= create_ordered_index(name, key_info)))
+      error= create_unique_index(unique_name, key_info);
+    break;
+  case UNIQUE_INDEX:
+    if (!(error= check_index_fields_not_null(idx_no)))
+      error= create_unique_index(unique_name, key_info);
+    break;
+  case ORDERED_INDEX:
+    error= create_ordered_index(name, key_info);
+    break;
+  default:
+    DBUG_ASSERT(FALSE);
+    break;
+  }
+  
+  DBUG_RETURN(error);
+}
 
 int ha_ndbcluster::create_ordered_index(const char *name, 
                                         KEY *key_info)
 {
   DBUG_ENTER("ha_ndbcluster::create_ordered_index");
-  DBUG_RETURN(create_index(name, key_info, FALSE));
+  DBUG_RETURN(create_ndb_index(name, key_info, FALSE));
 }
 
 int ha_ndbcluster::create_unique_index(const char *name, 
@@ -3956,7 +4036,7 @@
 {
 
   DBUG_ENTER("ha_ndbcluster::create_unique_index");
-  DBUG_RETURN(create_index(name, key_info, TRUE));
+  DBUG_RETURN(create_ndb_index(name, key_info, TRUE));
 }
 
 
@@ -3964,7 +4044,7 @@
   Create an index in NDB Cluster
  */
 
-int ha_ndbcluster::create_index(const char *name, 
+int ha_ndbcluster::create_ndb_index(const char *name, 
                                 KEY *key_info,
                                 bool unique)
 {
@@ -3973,7 +4053,7 @@
   KEY_PART_INFO *key_part= key_info->key_part;
   KEY_PART_INFO *end= key_part + key_info->key_parts;
   
-  DBUG_ENTER("ha_ndbcluster::create_index");
+  DBUG_ENTER("ha_ndbcluster::create_ndb_index");
   DBUG_PRINT("enter", ("name: %s ", name));
 
   NdbDictionary::Index ndb_index(name);
@@ -4003,6 +4083,82 @@
 }
 
 
+int ha_ndbcluster::add_index(TABLE *table_arg, 
+                             KEY *key_info, uint num_of_keys)
+{
+  DBUG_ENTER("ha_ndbcluster::add_index");
+  DBUG_PRINT("info", ("ha_ndbcluster::add_index to table %s", 
+                      table_arg->s->table_name));
+  int error= 0;
+  uint idx;
+
+  for (idx= 0; idx < num_of_keys; idx++)
+  {
+    KEY *key= key_info + idx;
+    KEY_PART_INFO *key_part= key->key_part;
+    KEY_PART_INFO *end= key_part + key->key_parts;
+    NDB_INDEX_TYPE idx_type= get_index_type_from_key(idx, key);
+    DBUG_PRINT("info", ("Adding index: '%s'", key_info[idx].name));
+    // Add fields to key_part struct
+    for (; key_part != end; key_part++)
+      key_part->field= table->field[key_part->fieldnr];
+    // Check index type
+    // Create index in ndb
+    error= create_index(key_info[idx].name, key, idx_type, idx);
+  }
+  if (!error)
+  {
+    THD *thd= current_thd;
+    Thd_ndb *thd_ndb= get_thd_ndb(thd);
+    Ndb *ndb= thd_ndb->ndb;
+    // Rebuild index list
+    error= build_index_list(ndb, table_arg, ILBP_ADD);
+  }
+
+  DBUG_RETURN(error);  
+}
+
+int ha_ndbcluster::drop_index(const char *name)
+{
+  DBUG_ENTER("ha_ndbcluster::drop_index");
+  DBUG_PRINT("enter", ("name: %s ", name));
+  Ndb *ndb= get_ndb();
+  NdbDictionary::Dictionary *dict= ndb->getDictionary();
+  DBUG_RETURN(dict->dropIndex(name, m_tabname));
+} 
+
+int ha_ndbcluster::prepare_drop_index(TABLE *table_arg, 
+                                      uint *key_num, uint num_of_keys)
+{
+  DBUG_ENTER("ha_ndbcluster::prepare_drop_index");
+  // Mark indexes for deletion
+  uint idx;
+  for (idx= 0; idx < num_of_keys; idx++)
+  {
+    DBUG_PRINT("info", ("ha_ndbcluster::prepare_drop_index %u", *key_num));
+    m_index[*key_num++].status= TO_BE_DROPPED;
+  }
+  // Renumber indexes
+  THD *thd= current_thd;
+  Thd_ndb *thd_ndb= get_thd_ndb(thd);
+  Ndb *ndb= thd_ndb->ndb;
+  DBUG_RETURN(build_index_list(ndb, table_arg, ILBP_RENUMBER));
+}
+ 
+int ha_ndbcluster::final_drop_index(TABLE *table_arg)
+{
+  DBUG_ENTER("ha_ndbcluster::final_drop_index");
+  DBUG_PRINT("info", ("ha_ndbcluster::final_drop_index"));
+  int error= 0;
+  // Really drop indexes
+  THD *thd= current_thd;
+  Thd_ndb *thd_ndb= get_thd_ndb(thd);
+  Ndb *ndb= thd_ndb->ndb;
+  error= build_index_list(ndb, table_arg, ILBP_DROP);
+
+  DBUG_RETURN(error);  
+}
+
 /*
   Rename a table in NDB Cluster
 */
@@ -4201,6 +4357,7 @@
   for (i= 0; i < MAX_KEY; i++)
   {
     m_index[i].type= UNDEFINED_INDEX;
+    m_index[i].status= UNDEFINED;
     m_index[i].unique_index= NULL;
     m_index[i].index= NULL;
     m_index[i].unique_index_attrid_map= NULL;
@@ -7377,6 +7534,213 @@
   }
 
   DBUG_RETURN(0);
+}
+
+
+/*
+  Create a table in NDB Cluster
+ */
+static uint get_no_fragments(ulonglong max_rows)
+{
+#if MYSQL_VERSION_ID >= 50000
+  uint acc_row_size= 25 + /*safety margin*/ 2;
+#else
+  uint acc_row_size= pk_length*4;
+  /* add acc overhead */
+  if (pk_length <= 8)  /* main page will set the limit */
+    acc_row_size+= 25 + /*safety margin*/ 2;
+  else                /* overflow page will set the limit */
+    acc_row_size+= 4 + /*safety margin*/ 4;
+#endif
+  ulonglong acc_fragment_size= 512*1024*1024;
+#if MYSQL_VERSION_ID >= 50100
+  return (max_rows*acc_row_size)/acc_fragment_size+1;
+#else
+  return ((max_rows*acc_row_size)/acc_fragment_size+1
+	  +1/*correct rounding*/)/2;
+#endif
+}
+
+
+/*
+  Routine to adjust default number of partitions to always be a multiple
+  of number of nodes and never more than 4 times the number of nodes.
+
+*/
+static bool adjusted_frag_count(uint no_fragments, uint no_nodes,
+                                uint &reported_frags)
+{
+  uint i= 0;
+  reported_frags= no_nodes;
+  while (reported_frags < no_fragments && ++i < 4 &&
+         (reported_frags + no_nodes) < MAX_PARTITIONS) 
+    reported_frags+= no_nodes;
+  return (reported_frags < no_fragments);
+}
+
+int ha_ndbcluster::get_default_no_partitions(ulonglong max_rows)
+{
+  uint reported_frags;
+  uint no_fragments= get_no_fragments(max_rows);
+  uint no_nodes= g_ndb_cluster_connection->no_db_nodes();
+  adjusted_frag_count(no_fragments, no_nodes, reported_frags);
+  return (int)reported_frags;
+}
+
+
+/*
+  User defined partitioning set-up. We need to check how many fragments the
+  user wants defined and which node groups to put those into. Later we also
+  want to attach those partitions to a tablespace.
+
+  All the functionality of the partition function, partition limits and so
+  forth are entirely handled by the MySQL Server. There is one exception to
+  this rule for PARTITION BY KEY where NDB handles the hash function and
+  this type can thus be handled transparently also by NDB API program.
+  For RANGE, HASH and LIST and subpartitioning the NDB API programs must
+  implement the function to map to a partition.
+*/
+
+uint ha_ndbcluster::set_up_partition_info(partition_info *part_info,
+                                          TABLE *table,
+                                          void *tab_par)
+{
+  DBUG_ENTER("ha_ndbcluster::set_up_partition_info");
+  ushort node_group[MAX_PARTITIONS];
+  ulong ng_index= 0, i, j;
+  NDBTAB *tab= (NDBTAB*)tab_par;
+  NDBTAB::FragmentType ftype= NDBTAB::UserDefined;
+  partition_element *part_elem;
+
+  if (part_info->part_type == HASH_PARTITION &&
+      part_info->list_of_part_fields == TRUE)
+  {
+    Field **fields= part_info->part_field_array;
+
+    if (part_info->linear_hash_ind)
+      ftype= NDBTAB::DistrKeyLin;
+    else
+      ftype= NDBTAB::DistrKeyHash;
+
+    for (i= 0; i < part_info->part_field_list.elements; i++)
+    {
+      NDBCOL *col= tab->getColumn(fields[i]->fieldnr - 1);
+      DBUG_PRINT("info",("setting dist key on %s", col->getName()));
+      col->setPartitionKey(TRUE);
+    }
+  }
+  List_iterator<partition_element> part_it(part_info->partitions);
+  for (i= 0; i < part_info->no_parts; i++)
+  {
+    part_elem= part_it++;
+    if (!is_sub_partitioned(part_info))
+    {
+      node_group[ng_index++]= part_elem->nodegroup_id;
+      //Here we should insert tablespace id based on tablespace name
+    }
+    else
+    {
+      List_iterator<partition_element> sub_it(part_elem->subpartitions);
+      for (j= 0; j < part_info->no_subparts; j++)
+      {
+        part_elem= sub_it++;
+        node_group[ng_index++]= part_elem->nodegroup_id;
+        //Here we should insert tablespace id based on tablespace name
+      }
+    }
+  }
+  {
+    uint no_nodes= g_ndb_cluster_connection->no_db_nodes();
+    if (ng_index > 4 * no_nodes)
+    {
+      DBUG_RETURN(1300);
+    }
+  }
+  tab->setNodeGroupIds(&node_group, ng_index);
+  tab->setFragmentType(ftype);
+  DBUG_RETURN(0);
+}
+
+
+/*
+  This routine is used to set-up fragmentation when the user has only specified
+  ENGINE = NDB and no user defined partitioning what so ever. Thus all values
+  will be based on default values. We will choose Linear Hash or Hash with
+  perfect spread dependent on a session variable defined in MySQL.
+*/
+
+static void ndb_set_fragmentation(NDBTAB &tab, TABLE *form, uint pk_length)
+{
+  NDBTAB::FragmentType ftype= NDBTAB::DistrKeyHash;
+  ushort node_group[MAX_PARTITIONS];
+  uint no_nodes= g_ndb_cluster_connection->no_db_nodes(), no_fragments, i;
+  DBUG_ENTER("ndb_set_fragmentation");
+
+  if (form->s->max_rows == (ha_rows) 0)
+  {
+    no_fragments= no_nodes;
+  }
+  else
+  {
+    /*
+      Ensure that we get enough fragments to handle all rows and ensure that
+      the table is fully distributed by keeping the number of fragments a
+      multiple of the number of nodes.
+    */
+    uint fragments= get_no_fragments(form->s->max_rows);
+    if (adjusted_frag_count(fragments, no_nodes, no_fragments))
+    {
+      push_warning(current_thd,
+                   MYSQL_ERROR::WARN_LEVEL_WARN, ER_UNKNOWN_ERROR,
+      "Ndb might have problems storing the max amount of rows specified");
+    }
+  }
+  /*
+    Always start with node group 0 and continue with next node group from
+    there
+  */
+  node_group[0]= 0;
+  for (i= 1; i < no_fragments; i++)
+    node_group[i]= UNDEF_NODEGROUP;
+  switch (opt_ndb_distribution_id)
+  {
+  case ND_KEYHASH:
+    ftype= NDBTAB::DistrKeyHash;
+    break;
+  case ND_LINHASH:
+    ftype= NDBTAB::DistrKeyLin;
+    break;
+  }
+  tab.setFragmentType(ftype);
+  tab.setNodeGroupIds(&node_group, no_fragments);
+  DBUG_VOID_RETURN;
+}
+
+
+bool ha_ndbcluster::check_if_incompatible_data(HA_CREATE_INFO *info,
+					       uint table_changes)
+{
+  /*
+    TODO: Remove the dummy return below, when cluster gets
+    signal from alter table when only .frm is changed. Cluster
+    needs it to manage the copies.
+  */
+  return COMPATIBLE_DATA_NO;
+
+  if (table_changes != IS_EQUAL_YES)
+    return COMPATIBLE_DATA_NO;
+  
+  /* Check that auto_increment value was not changed */
+  if ((info->used_fields & HA_CREATE_USED_AUTO) &&
+      info->auto_increment_value != 0)
+    return COMPATIBLE_DATA_NO;
+  
+  /* Check that row format didn't change */
+  if ((info->used_fields & HA_CREATE_USED_AUTO) &&
+      get_row_type() != info->row_type)
+    return COMPATIBLE_DATA_NO;
+
+  return COMPATIBLE_DATA_YES;
 }
 
 bool set_up_tablespace(st_alter_tablespace *info,

--- 1.85/sql/ha_ndbcluster.h	2005-10-07 20:32:56 +02:00
+++ 1.86/sql/ha_ndbcluster.h	2005-10-12 13:26:44 +02:00
@@ -50,8 +50,16 @@
   ORDERED_INDEX = 5
 } NDB_INDEX_TYPE;
 
+typedef enum ndb_index_status {
+  UNDEFINED = 0,
+  CREATED = 1,
+  ACTIVE = 2,
+  TO_BE_DROPPED = 3
+} NDB_INDEX_STATUS;
+
 typedef struct ndb_index_data {
   NDB_INDEX_TYPE type;
+  NDB_INDEX_STATUS status;
   void *index;
   void *unique_index;
   unsigned char *unique_index_attrid_map;
@@ -61,6 +69,7 @@
   // Simple counter mechanism to decide when to connect to db
   uint index_stat_update_freq;
   uint index_stat_query_count;
+  // Index marked for deletion, for DROP INDEX
 } NDB_INDEX_DATA;
 
 typedef struct st_ndbcluster_share {
@@ -516,9 +525,13 @@
   const char ** bas_ext() const;
   ulong table_flags(void) const;
   ulong alter_table_flags(void) const
-  {
-    return 0;
+  { 
+    return (HA_ONLINE_ADD_INDEX | HA_ONLINE_DROP_INDEX |
+            HA_ONLINE_ADD_UNIQUE_INDEX | HA_ONLINE_DROP_UNIQUE_INDEX);
   }
+  int add_index(TABLE *table_arg, KEY *key_info, uint num_of_keys);
+  int prepare_drop_index(TABLE *table_arg, uint *key_num, uint num_of_keys);
+  int final_drop_index(TABLE *table_arg);
   ulong partition_flags(void) const
   {
     return (HA_CAN_PARTITION | HA_CAN_UPDATE_PARTITION_KEY |
@@ -621,16 +634,22 @@
 private:
   int alter_table_name(const char *to);
   int drop_table();
-  int create_index(const char *name, KEY *key_info, bool unique);
+  int create_ndb_index(const char *name, KEY *key_info, bool unique);
   int create_ordered_index(const char *name, KEY *key_info);
   int create_unique_index(const char *name, KEY *key_info);
+  int create_index(const char *name, KEY *key_info, 
+                   NDB_INDEX_TYPE idx_type, uint idx_no);
+  int drop_index(const char *name);
   int initialize_autoincrement(const void *table);
-  enum ILBP {ILBP_CREATE = 0, ILBP_OPEN = 1}; // Index List Build Phase
+  // Index List Build Phases
+  enum ILBP {ILBP_CREATE = 0, ILBP_OPEN = 1, 
+             ILBP_ADD = 2, ILBP_RENUMBER = 3, ILBP_DROP = 4}; 
   int build_index_list(Ndb *ndb, TABLE *tab, enum ILBP phase);
   int get_metadata(const char* path);
   void release_metadata();
   NDB_INDEX_TYPE get_index_type(uint idx_no) const;
   NDB_INDEX_TYPE get_index_type_from_table(uint index_no) const;
+  NDB_INDEX_TYPE get_index_type_from_key(uint index_no, KEY *key_info) const;
   int check_index_fields_not_null(uint index_no);
 
   uint set_up_partition_info(partition_info *part_info,