Merge mskold@bk-internal.mysql.com:/home/bk/mysql-5.1-new

into  mysql.com:/usr/local/home/marty/MySQL/mysql-5.1-new


sql/ha_ndbcluster.cc:
  Auto merged

mysql-test/r/ndb_alter_table_row.result

@@ -3,10 +3,19 @@ create table t1 ( a int primary key, b varchar(10), c varchar(10), index (b) )
 engine=ndb;
 insert into t1 values (1,'one','one'), (2,'two','two'), (3,'three','three');
 create index c on t1(c);
+show indexes from t1;
+Table	Non_unique	Key_name	Seq_in_index	Column_name	Collation	Cardinality	Sub_part	Packed	Null	Index_type	Comment
+t1	0	PRIMARY	1	a	A	3	NULL	NULL		BTREE	
+t1	1	b	1	b	A	3	NULL	NULL	YES	BTREE	
+t1	1	c	1	c	A	3	NULL	NULL	YES	BTREE	
 select * from t1 where c = 'two';
 a	b	c
 2	two	two
 alter table t1 drop index c;
+show indexes from t1;
+Table	Non_unique	Key_name	Seq_in_index	Column_name	Collation	Cardinality	Sub_part	Packed	Null	Index_type	Comment
+t1	0	PRIMARY	1	a	A	3	NULL	NULL		BTREE	
+t1	1	b	1	b	A	3	NULL	NULL	YES	BTREE	
 select * from t1 where c = 'two';
 a	b	c
 2	two	two

mysql-test/t/ndb_alter_table_row.test

@@ -13,10 +13,12 @@ engine=ndb;
 insert into t1 values (1,'one','one'), (2,'two','two'), (3,'three','three');
 create index c on t1(c); 
 connection server2;
+show indexes from t1;
 select * from t1 where c = 'two';
 connection server1;
 alter table t1 drop index c;
 connection server2;
+show indexes from t1;
 select * from t1 where c = 'two';
 connection server1;
 drop table t1;

sql/ha_ndbcluster.cc

@@ -991,7 +991,7 @@ bool ha_ndbcluster::uses_blob_value()
     -2   Meta data has changed; Re-read data and try again
 */
 
-static int cmp_frm(const NDBTAB *ndbtab, const void *pack_data,
+int cmp_frm(const NDBTAB *ndbtab, const void *pack_data,
             uint pack_length)
 {
   DBUG_ENTER("cmp_frm");

sql/ha_ndbcluster_binlog.cc

@@ -1291,20 +1291,87 @@ static int
 ndb_handle_schema_change(THD *thd, Ndb *ndb, NdbEventOperation *pOp,
                          NDB_SHARE *share)
 {
+  DBUG_ENTER("ndb_handle_schema_change");
   int remote_drop_table= 0, do_close_cached_tables= 0;
+  const char *dbname= share->table->s->db.str;
+  const char *tabname= share->table->s->table_name.str;
+  bool online_alter_table= (pOp->getEventType() == NDBEVENT::TE_ALTER &&
+                            pOp->tableFrmChanged());
 
   if (pOp->getEventType() != NDBEVENT::TE_CLUSTER_FAILURE &&
       pOp->getReqNodeId() != g_ndb_cluster_connection->node_id())
   {
-    ndb->setDatabaseName(share->table->s->db.str);
+    NDBDICT *dict= ndb->getDictionary();
+    NdbDictionary::Dictionary::List index_list;
+
+    ndb->setDatabaseName(dbname);
+    // Invalidating indexes
+    if (! dict->listIndexes(index_list, tabname))
+    {
+      for (unsigned i = 0; i < index_list.count; i++) {
+        NdbDictionary::Dictionary::List::Element& index=
+          index_list.elements[i];
+        DBUG_PRINT("info", ("Invalidating index %s.%s",
+                            index.database, index.name));
+        dict->invalidateIndex(index.name, tabname);
+      }
+    }
+    // Invalidate table
     ha_ndbcluster::invalidate_dictionary_cache(share->table->s,
                                                ndb,
-                                               share->table->s->db.str,
-                                               share->table->s->table_name.str,
+                                               dbname,
+                                               tabname,
                                                TRUE);
+
+    if (online_alter_table)
+    {  
+      char key[FN_REFLEN];
+      const void *data= 0, *pack_data= 0;
+      uint length, pack_length;
+      int error;
+
+      DBUG_PRINT("info", ("Detected frm change of table %s.%s",
+                          dbname, tabname));
+      const NDBTAB *altered_table= pOp->getEvent()->getTable();
+      bool remote_event=
+        pOp->getReqNodeId() != g_ndb_cluster_connection->node_id();
+      strxnmov(key, FN_LEN-1, mysql_data_home, "/",
+               dbname, "/", tabname, NullS);
+      /*
+        If the frm of the altered table is different than the one on
+        disk then overwrite it with the new table definition
+      */
+      if (remote_event &&
+          readfrm(key, &data, &length) == 0 &&
+          packfrm(data, length, &pack_data, &pack_length) == 0 &&
+          cmp_frm(altered_table, pack_data, pack_length))
+      {
+        DBUG_DUMP("frm", (char*)altered_table->getFrmData(), 
+                  altered_table->getFrmLength());
+        pthread_mutex_lock(&LOCK_open);
+        dict->putTable(altered_table);
+        
+        if ((error= unpackfrm(&data, &length, altered_table->getFrmData())) ||
+            (error= writefrm(key, data, length)))
+        {
+          sql_print_information("NDB: Failed write frm for %s.%s, error %d",
+                                dbname, tabname, error);
+        }
+        pthread_mutex_unlock(&LOCK_open);
+        close_cached_tables((THD*) 0, 0, (TABLE_LIST*) 0);
+      }
+    }
     remote_drop_table= 1;
   }
 
+  // If only frm was changed continue replicating
+  if (online_alter_table)
+  {
+    /* Signal ha_ndbcluster::alter_table that drop is done */
+    (void) pthread_cond_signal(&injector_cond);
+    DBUG_RETURN(0);
+  }
+
   (void) pthread_mutex_lock(&share->mutex);
   DBUG_ASSERT(share->op == pOp || share->op_old == pOp);
   if (share->op_old == pOp)
@@ -1385,7 +1452,7 @@ ndb_binlog_thread_handle_schema_event(THD *thd, Ndb *ndb,
           /* fall through */
         case SOT_ALTER_TABLE:
           /* fall through */
-          if (!ndb_binlog_running)
+          if (ndb_binlog_running)
           {
             log_query= 1;
             break; /* discovery will be handled by binlog */
@@ -1482,11 +1549,16 @@ ndb_binlog_thread_handle_schema_event(THD *thd, Ndb *ndb,
       // skip
       break;
     case NDBEVENT::TE_ALTER:
+      if (pOp->tableNameChanged())
+      {  
+        DBUG_PRINT("info", ("Detected name change of table %s.%s",
+                            share->db, share->table_name));
         /* do the rename of the table in the share */
         share->table->s->db.str= share->db;
         share->table->s->db.length= strlen(share->db);
         share->table->s->table_name.str= share->table_name;
         share->table->s->table_name.length= strlen(share->table_name);
+      }
       ndb_handle_schema_change(thd, ndb, pOp, share);
       break;
     case NDBEVENT::TE_CLUSTER_FAILURE:
@@ -2357,6 +2429,10 @@ ndb_binlog_thread_handle_non_data_event(Ndb *ndb, NdbEventOperation *pOp,
                        share->key, share, pOp, share->op, share->op_old));
     break;
   case NDBEVENT::TE_ALTER:
+    if (pOp->tableNameChanged())
+    {
+      DBUG_PRINT("info", ("Detected name change of table %s.%s",
+                          share->db, share->table_name));
       /* ToDo: remove printout */
       if (ndb_extra_logging)
         sql_print_information("NDB Binlog: rename table %s%s/%s -> %s.",
@@ -2368,6 +2444,7 @@ ndb_binlog_thread_handle_non_data_event(Ndb *ndb, NdbEventOperation *pOp,
       share->table->s->db.length= strlen(share->db);
       share->table->s->table_name.str= share->table_name;
       share->table->s->table_name.length= strlen(share->table_name);
+    }
     goto drop_alter_common;
   case NDBEVENT::TE_DROP:
     if (apply_status_share == share)

sql/ha_ndbcluster_binlog.h

@@ -122,6 +122,8 @@ ndbcluster_show_status_binlog(THD* thd, stat_print_fn *stat_print,
   prototypes for ndb handler utility function also needed by
   the ndb binlog code
 */
+int cmp_frm(const NDBTAB *ndbtab, const void *pack_data,
+            uint pack_length);
 int ndbcluster_find_all_files(THD *thd);
 #endif /* HAVE_NDB_BINLOG */
 

storage/ndb/include/ndbapi/NdbDictionary.hpp

@@ -1597,6 +1597,14 @@ public:
      */
     const Table * getTable(const char * name) const;
 
+#ifndef DOXYGEN_SHOULD_SKIP_INTERNAL
+    /*
+     * Save a table definition in dictionary cache
+     * @param table Object to put into cache
+     */
+    void putTable(const Table * table);
+#endif
+
     /**
      * Get index with given name, NULL if undefined
      * @param indexName  Name of index to get.

storage/ndb/include/ndbapi/NdbEventOperation.hpp

@@ -220,6 +220,7 @@ public:
 
 #ifndef DOXYGEN_SHOULD_SKIP_INTERNAL
   /** these are subject to change at any time */
+  const NdbDictionary::Table* getTable() const;
   const NdbDictionary::Event *getEvent() const;
   const NdbRecAttr *getFirstPkAttr() const;
   const NdbRecAttr *getFirstPkPreAttr() const;

storage/ndb/src/ndbapi/NdbDictionary.cpp

@@ -1349,6 +1349,13 @@ NdbDictionary::Dictionary::getTable(const char * name, void **data) const
   return 0;
 }
 
+void NdbDictionary::Dictionary::putTable(const NdbDictionary::Table * table)
+{
+ NdbDictionary::Table  *copy_table = new NdbDictionary::Table;
+  *copy_table = *table;
+  m_impl.putTable(&NdbTableImpl::getImpl(*copy_table));
+}
+
 void NdbDictionary::Dictionary::set_local_table_data_size(unsigned sz)
 {
   m_impl.m_local_table_data_size= sz;

storage/ndb/src/ndbapi/NdbDictionaryImpl.cpp

@@ -1305,7 +1305,14 @@ NdbDictionaryImpl::fetchGlobalTableImpl(const BaseString& internalTableName)
 void
 NdbDictionaryImpl::putTable(NdbTableImpl *impl)
 {
+  NdbTableImpl *old;
+
   m_globalHash->lock();
+  if ((old= m_globalHash->get(impl->m_internalName.c_str())))
+  {
+    old->m_status = NdbDictionary::Object::Invalid;
+    m_globalHash->drop(old);
+  }
   m_globalHash->put(impl->m_internalName.c_str(), impl);
   m_globalHash->unlock();
   Ndb_local_table_info *info=

storage/ndb/src/ndbapi/NdbEventOperation.cpp

@@ -144,6 +144,10 @@ NdbEventOperation::print()
 /*
  * Internal for the mysql server
  */
+const NdbDictionary::Table *NdbEventOperation::getTable() const
+{
+  return m_impl.m_eventImpl->m_tableImpl->m_facade;
+}
 const NdbDictionary::Event *NdbEventOperation::getEvent() const
 {
   return m_impl.m_eventImpl->m_facade;

storage/ndb/src/ndbapi/NdbEventOperationImpl.cpp

@@ -645,6 +645,14 @@ NdbEventOperationImpl::receive_event()
                                     m_buffer.length() / 4, 
                                     true);
     m_buffer.clear();
+    if (at)
+      at->buildColumnHash();
+    else
+    {
+      DBUG_PRINT_EVENT("info", ("Failed to parse DictTabInfo error %u", 
+                                error.code));
+      DBUG_RETURN_EVENT(1);
+    }
     if ( m_eventImpl->m_tableImpl) 
       delete m_eventImpl->m_tableImpl;
     m_eventImpl->m_tableImpl = at;