Additional fixes for WL#1892

sql/ha_ndbcluster.cc:
  Additional fixes for WL#1892, disable autodiscover during add/drop index
sql/ha_ndbcluster.h:
  Additional fixes for WL#1892, disable autodiscover during add/drop index
sql/sql_table.cc:
  Additional fixes for WL#1892, removed double commit + added call to create_handler_files after create/drop index
mysql-test/r/ndb_alter_table_row.result:
  Additional fixes for WL#1892, changed test case
mysql-test/t/ndb_alter_table_row.test:
  Additional fixes for WL#1892, changed test case

mysql-test/r/ndb_alter_table_row.result

@@ -8,6 +8,8 @@ a	b	c
 2	two	two
 alter table t1 drop index c;
 select * from t1 where c = 'two';
+ERROR HY000: Table definition has changed, please retry transaction
+select * from t1 where c = 'two';
 a	b	c
 2	two	two
 drop table t1;

mysql-test/t/ndb_alter_table_row.test

@@ -17,6 +17,8 @@ select * from t1 where c = 'two';
 connection server1;
 alter table t1 drop index c;
 connection server2;
+--error 1412
+select * from t1 where c = 'two';
 select * from t1 where c = 'two';
 connection server1;
 drop table t1;

sql/ha_ndbcluster.cc

@@ -974,22 +974,25 @@ int ha_ndbcluster::get_metadata(const char *path)
     
     if (cmp_frm(tab, pack_data, pack_length))
     {
-      if (!invalidating_ndb_table)
+      if (m_share->state != NSS_ALTERED)
       {
-        DBUG_PRINT("info", ("Invalidating table"));
-        invalidate_dictionary_cache(TRUE);
-        invalidating_ndb_table= TRUE;
-      }
-      else
-      {
-        DBUG_PRINT("error", 
-                   ("metadata, pack_length: %d  getFrmLength: %d  memcmp: %d",
-                    pack_length, tab->getFrmLength(),
-                    memcmp(pack_data, tab->getFrmData(), pack_length)));
-        DBUG_DUMP("pack_data", (char*)pack_data, pack_length);
-        DBUG_DUMP("frm", (char*)tab->getFrmData(), tab->getFrmLength());
-        error= HA_ERR_TABLE_DEF_CHANGED;
-        invalidating_ndb_table= FALSE;
+        if (!invalidating_ndb_table)
+        {
+          DBUG_PRINT("info", ("Invalidating table"));
+          invalidate_dictionary_cache(TRUE);
+          invalidating_ndb_table= TRUE;
+        }
+        else
+        {
+          DBUG_PRINT("error", 
+                     ("metadata, pack_length: %d  getFrmLength: %d  memcmp: %d",
+                      pack_length, tab->getFrmLength(),
+                      memcmp(pack_data, tab->getFrmData(), pack_length)));
+          DBUG_DUMP("pack_data", (char*)pack_data, pack_length);
+          DBUG_DUMP("frm", (char*)tab->getFrmData(), tab->getFrmLength());
+          error= HA_ERR_TABLE_DEF_CHANGED;
+          invalidating_ndb_table= FALSE;
+        }
       }
     }
     else
@@ -1044,6 +1047,36 @@ static int fix_unique_index_attr_order(NDB_INDEX_DATA &data,
   DBUG_RETURN(0);
 }
 
+int ha_ndbcluster::table_changed(const void *pack_frm_data, uint pack_frm_len)
+{
+  Ndb *ndb;
+  NDBDICT *dict;
+  const NDBTAB *orig_tab;
+  NdbDictionary::Table new_tab;
+  int result;
+  DBUG_ENTER("ha_ndbcluster::table_changed");
+  DBUG_PRINT("info", ("Modifying frm for table %s", m_tabname));
+  if (check_ndb_connection())
+    DBUG_RETURN(my_errno= HA_ERR_NO_CONNECTION);
+                                                                             
+  ndb= get_ndb();
+  dict= ndb->getDictionary();
+  if (!(orig_tab= dict->getTable(m_tabname)))
+    ERR_RETURN(dict->getNdbError());
+  // Check if thread has stale local cache
+  if (orig_tab->getObjectStatus() == NdbDictionary::Object::Invalid)
+  {
+    dict->removeCachedTable(m_tabname);
+    if (!(orig_tab= dict->getTable(m_tabname)))
+      ERR_RETURN(dict->getNdbError());
+  }
+  new_tab= *orig_tab;
+  new_tab.setFrm(pack_frm_data, pack_frm_len);
+  if (dict->alterTable(new_tab) != 0)
+    ERR_RETURN(dict->getNdbError());
+  DBUG_RETURN(0);
+}
+
 /*
   Create all the indexes for a table.
   If any index should fail to be created,
@@ -4280,6 +4313,46 @@ int ha_ndbcluster::create(const char *name,
   DBUG_RETURN(my_errno);
 }
 
+int ha_ndbcluster::create_handler_files(const char *file) 
+{ 
+  const char *name;
+  Ndb* ndb;
+  const NDBTAB *tab;
+  const void *data, *pack_data;
+  uint length, pack_length;
+  int error= 0;
+
+  DBUG_ENTER("create_handler_files");
+
+  if (!(ndb= get_ndb()))
+    DBUG_RETURN(HA_ERR_NO_CONNECTION);
+
+  NDBDICT *dict= ndb->getDictionary();
+  if (!(tab= dict->getTable(m_tabname)))
+    DBUG_RETURN(0); // Must be a create, ignore since frm is saved in create
+
+  name= table->s->normalized_path.str;
+  DBUG_PRINT("enter", ("m_tabname: %s, path: %s", m_tabname, name));
+  if (readfrm(name, &data, &length) ||
+      packfrm(data, length, &pack_data, &pack_length))
+  {
+    DBUG_PRINT("info", ("Missing frm for %s", m_tabname));
+    my_free((char*)data, MYF(MY_ALLOW_ZERO_PTR));
+    my_free((char*)pack_data, MYF(MY_ALLOW_ZERO_PTR));
+    DBUG_RETURN(1);
+  }
+  if (cmp_frm(tab, pack_data, pack_length))
+  {  
+    DBUG_PRINT("info", ("Table %s has changed, altering frm in ndb", 
+                        m_tabname));
+    error= table_changed(pack_data, pack_length);
+  }
+  my_free((char*)data, MYF(MY_ALLOW_ZERO_PTR));
+  my_free((char*)pack_data, MYF(MY_ALLOW_ZERO_PTR));
+
+  DBUG_RETURN(error);
+}
+
 int ha_ndbcluster::create_index(const char *name, KEY *key_info, 
                                 NDB_INDEX_TYPE idx_type, uint idx_no)
 {
@@ -4407,7 +4480,7 @@ int ha_ndbcluster::add_index(TABLE *table_arg,
     if((error= create_index(key_info[idx].name, key, idx_type, idx)))
       break;
   }
-
+  m_share->state= NSS_ALTERED;
   DBUG_RETURN(error);  
 }
 
@@ -4442,6 +4515,7 @@ int ha_ndbcluster::prepare_drop_index(TABLE *table_arg,
   THD *thd= current_thd;
   Thd_ndb *thd_ndb= get_thd_ndb(thd);
   Ndb *ndb= thd_ndb->ndb;
+  m_share->state= NSS_ALTERED;
   DBUG_RETURN(renumber_indexes(ndb, table_arg));
 }
  
@@ -4452,14 +4526,11 @@ 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= drop_indexes(ndb, table_arg);
-
-  DBUG_RETURN(error);  
+  DBUG_RETURN(drop_indexes(ndb, table_arg));
 }
 
 /*
@@ -5282,9 +5353,13 @@ int ndbcluster_find_all_files(THD *thd)
       }
       else if (cmp_frm(ndbtab, pack_data, pack_length))
       {
-        discover= 1;
-        sql_print_information("NDB: mismatch in frm for %s.%s, discovering...",
-                              elmt.database, elmt.name);
+        NDB_SHARE *share= get_share(key, 0, false);
+        if (!share || share->state != NSS_ALTERED)
+        {
+          discover= 1;
+          sql_print_information("NDB: mismatch in frm for %s.%s, discovering...",
+                                elmt.database, elmt.name);
+        }
       }
       my_free((char*) data, MYF(MY_ALLOW_ZERO_PTR));
       my_free((char*) pack_data, MYF(MY_ALLOW_ZERO_PTR));
@@ -6451,7 +6526,7 @@ NDB_SHARE *ndbcluster_get_share(const char *key, TABLE *table,
       MEM_ROOT *old_root= *root_ptr;
       init_sql_alloc(&share->mem_root, 1024, 0);
       *root_ptr= &share->mem_root; // remember to reset before return
-
+      share->state= NSS_INITIAL;
       /* enough space for key, db, and table_name */
       share->key= alloc_root(*root_ptr, 2 * (length + 1));
       share->key_length= length;
@@ -9003,13 +9078,6 @@ static void ndb_set_fragmentation(NDBTAB &tab, TABLE *form, uint pk_length)
 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;
   

sql/ha_ndbcluster.h

@@ -80,10 +80,12 @@ typedef union { const NdbRecAttr *rec; NdbBlob *blob; void *ptr; } NdbValue;
 
 typedef enum {
   NSS_INITIAL= 0,
-  NSS_DROPPED
+  NSS_DROPPED,
+  NSS_ALTERED 
 } NDB_SHARE_STATE;
 
 typedef struct st_ndbcluster_share {
+  NDB_SHARE_STATE state;
   MEM_ROOT mem_root;
   THR_LOCK lock;
   pthread_mutex_t mutex;
@@ -97,7 +99,6 @@ typedef struct st_ndbcluster_share {
   char *table_name;
 #ifdef HAVE_NDB_BINLOG
   uint32 flags;
-  NDB_SHARE_STATE state;
   NdbEventOperation *op;
   NdbEventOperation *op_old; // for rename table
   char *old_names; // for rename table
@@ -587,6 +588,7 @@ class ha_ndbcluster: public handler
   int rename_table(const char *from, const char *to);
   int delete_table(const char *name);
   int create(const char *name, TABLE *form, HA_CREATE_INFO *info);
+  int create_handler_files(const char *file);
   int get_default_no_partitions(ulonglong max_rows);
   THR_LOCK_DATA **store_lock(THD *thd,
                              THR_LOCK_DATA **to,
@@ -674,6 +676,7 @@ static void set_tabname(const char *pathname, char *tabname);
   int create_index(const char *name, KEY *key_info, 
                    NDB_INDEX_TYPE idx_type, uint idx_no);
   int drop_ndb_index(const char *name);
+  int table_changed(const void *pack_frm_data, uint pack_frm_len);
 // Index list management
   int create_indexes(Ndb *ndb, TABLE *tab);
   void clear_index(int i);

sql/sql_table.cc

@@ -3562,6 +3562,7 @@ bool mysql_alter_table(THD *thd,char *new_db, char *new_name,
   uint *index_drop_buffer;
   uint index_add_count;
   uint *index_add_buffer;
+  bool committed= 0;
   DBUG_ENTER("mysql_alter_table");
 
   thd->proc_info="init";
@@ -4968,6 +4969,7 @@ bool mysql_alter_table(THD *thd,char *new_db, char *new_name,
       DBUG_PRINT("info", ("Committing after add/drop index"));
       if (ha_commit_stmt(thd) || ha_commit(thd))
         goto err;
+      committed= 1;
     }
   }
   /*end of if (! new_table) for add/drop index*/
@@ -5099,7 +5101,6 @@ bool mysql_alter_table(THD *thd,char *new_db, char *new_name,
     VOID(pthread_mutex_unlock(&LOCK_open));
     goto err;
   }
-#ifdef XXX_TO_BE_DONE_LATER_BY_WL1892
   if (! need_copy_table)
   {
     if (! table)
@@ -5116,7 +5117,6 @@ bool mysql_alter_table(THD *thd,char *new_db, char *new_name,
       goto err;
     }
   }
-#endif
   if (thd->lock || new_name != table_name)	// True if WIN32
   {
     /*
@@ -5166,11 +5166,14 @@ bool mysql_alter_table(THD *thd,char *new_db, char *new_name,
     wait_if_global_read_lock(), which could create a deadlock if called
     with LOCK_open.
   */
-  error = ha_commit_stmt(thd);
-  if (ha_commit(thd))
-    error=1;
-  if (error)
-    goto err;
+  if (!committed)
+  {
+    error = ha_commit_stmt(thd);
+    if (ha_commit(thd))
+      error=1;
+    if (error)
+      goto err;
+  }
   thd->proc_info="end";
 
   DBUG_ASSERT(!(mysql_bin_log.is_open() && binlog_row_based &&