Merge mysql.com:/usr/local/home/marty/MySQL/mysql-5.1-new-ndb_improved_on-line_discover

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

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

@@ -992,7 +992,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

@@ -230,60 +230,13 @@ static void run_query(THD *thd, char *buf, char *end,
   }
 }
 
-/*
-  Initialize the binlog part of the NDB_SHARE
-*/
-void ndbcluster_binlog_init_share(NDB_SHARE *share, TABLE *_table)
+int
+ndbcluster_binlog_open_table(THD *thd, NDB_SHARE *share,
+                             TABLE_SHARE *table_share, TABLE *table)
 {
-  THD *thd= current_thd;
-  MEM_ROOT *mem_root= &share->mem_root;
-  int do_event_op= ndb_binlog_running;
-
-  share->op= 0;
-  share->table= 0;
-
-  if (!schema_share &&
-      strcmp(share->db, NDB_REP_DB) == 0 &&
-      strcmp(share->table_name, NDB_SCHEMA_TABLE) == 0)
-    do_event_op= 1;
-
-  {
-    int i, no_nodes= g_ndb_cluster_connection->no_db_nodes();
-    share->subscriber_bitmap= (MY_BITMAP*)
-      alloc_root(mem_root, no_nodes * sizeof(MY_BITMAP));
-    for (i= 0; i < no_nodes; i++)
-    {
-      bitmap_init(&share->subscriber_bitmap[i],
-                  (Uint32*)alloc_root(mem_root, max_ndb_nodes/8),
-                  max_ndb_nodes, false);
-      bitmap_clear_all(&share->subscriber_bitmap[i]);
-    }
-    bitmap_init(&share->slock_bitmap, share->slock,
-                sizeof(share->slock)*8, false);
-    bitmap_clear_all(&share->slock_bitmap);
-  }
-
-  if (!do_event_op)
-  {
-    if (_table)
-    {
-      if (_table->s->primary_key == MAX_KEY)
-        share->flags|= NSF_HIDDEN_PK;
-      if (_table->s->blob_fields != 0)
-        share->flags|= NSF_BLOB_FLAG;
-    }
-    else
-    {
-      share->flags|= NSF_NO_BINLOG;
-    }
-    return;
-  }
-  while (1) 
-  {
-    TABLE_SHARE *table_share= 
-      (TABLE_SHARE *) my_malloc(sizeof(*table_share), MYF(MY_WME));
-    TABLE *table= (TABLE*) my_malloc(sizeof(*table), MYF(MY_WME));
   int error;
+  MEM_ROOT *mem_root= &share->mem_root;
+  DBUG_ENTER("ndbcluster_binlog_open_table");
   
   init_tmp_table_share(table_share, share->db, 0, share->table_name, 
                        share->key);
@@ -296,7 +249,7 @@ void ndbcluster_binlog_init_share(NDB_SHARE *share, TABLE *_table)
     table_share= 0;
     my_free((gptr) table, MYF(0));
     table= 0;
-      break;
+    DBUG_RETURN(error);
   }
   if ((error= open_table_from_share(thd, table_share, "", 0, 
                                     (uint) READ_ALL, 0, table, FALSE)))
@@ -308,7 +261,7 @@ void ndbcluster_binlog_init_share(NDB_SHARE *share, TABLE *_table)
     table_share= 0;
     my_free((gptr) table, MYF(0));
     table= 0;
-      break;
+    DBUG_RETURN(error);
   }
   assign_new_table_id(table);
   if (!table->record[1] || table->record[1] == table->record[0])
@@ -338,6 +291,67 @@ void ndbcluster_binlog_init_share(NDB_SHARE *share, TABLE *_table)
   share->ndb_value[1]= (NdbValue*)
     alloc_root(mem_root, sizeof(NdbValue) * table->s->fields
                +1 /*extra for hidden key*/);
+
+  DBUG_RETURN(0);
+}
+
+
+/*
+  Initialize the binlog part of the NDB_SHARE
+*/
+void ndbcluster_binlog_init_share(NDB_SHARE *share, TABLE *_table)
+{
+  THD *thd= current_thd;
+  MEM_ROOT *mem_root= &share->mem_root;
+  int do_event_op= ndb_binlog_running;
+
+  share->op= 0;
+  share->table= 0;
+
+  if (!schema_share &&
+      strcmp(share->db, NDB_REP_DB) == 0 &&
+      strcmp(share->table_name, NDB_SCHEMA_TABLE) == 0)
+    do_event_op= 1;
+
+  {
+    int i, no_nodes= g_ndb_cluster_connection->no_db_nodes();
+    share->subscriber_bitmap= (MY_BITMAP*)
+      alloc_root(mem_root, no_nodes * sizeof(MY_BITMAP));
+    for (i= 0; i < no_nodes; i++)
+    {
+      bitmap_init(&share->subscriber_bitmap[i],
+                  (Uint32*)alloc_root(mem_root, max_ndb_nodes/8),
+                  max_ndb_nodes, false);
+      bitmap_clear_all(&share->subscriber_bitmap[i]);
+    }
+    bitmap_init(&share->slock_bitmap, share->slock,
+                sizeof(share->slock)*8, false);
+    bitmap_clear_all(&share->slock_bitmap);
+  }
+
+  if (!do_event_op)
+  {
+    if (_table)
+    {
+      if (_table->s->primary_key == MAX_KEY)
+        share->flags|= NSF_HIDDEN_PK;
+      if (_table->s->blob_fields != 0)
+        share->flags|= NSF_BLOB_FLAG;
+    }
+    else
+    {
+      share->flags|= NSF_NO_BINLOG;
+    }
+    return;
+  }
+  while (1) 
+  {
+    int error;
+    TABLE_SHARE *table_share= 
+      (TABLE_SHARE *) my_malloc(sizeof(*table_share), MYF(MY_WME));
+    TABLE *table= (TABLE*) my_malloc(sizeof(*table), MYF(MY_WME));
+    if ((error= ndbcluster_binlog_open_table(thd, share, table_share, table)))
+      break;
     if (table->s->primary_key == MAX_KEY)
       share->flags|= NSF_HIDDEN_PK;
     if (table->s->blob_fields != 0)
@@ -1289,10 +1303,15 @@ 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())
+      (uint) pOp->getReqNodeId() != g_ndb_cluster_connection->node_id())
   {
     TABLE_SHARE *table_share= share->table->s;
     TABLE* table= share->table;
@@ -1300,15 +1319,65 @@ ndb_handle_schema_change(THD *thd, Ndb *ndb, NdbEventOperation *pOp,
     /* 
        Invalidate table and all it's indexes
     */
+    ndb->setDatabaseName(share->table->s->db.str);
     ha_ndbcluster table_handler(table_share);
     table_handler.set_dbname(share->key);
     table_handler.set_tabname(share->key);
     table_handler.open_indexes(ndb, table, TRUE);
     table_handler.invalidate_dictionary_cache(TRUE);
     
+    if (online_alter_table)
+    {  
+      char key[FN_REFLEN];
+      const void *data= 0, *pack_data= 0;
+      uint length, pack_length;
+      int error;
+      NDBDICT *dict= ndb->getDictionary();
+      const NDBTAB *altered_table= pOp->getTable();
+
+      DBUG_PRINT("info", ("Detected frm change of table %s.%s",
+                          dbname, tabname));
+      build_table_filename(key, FN_LEN-1, 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 (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);
+        /*
+        if ((error= ndbcluster_binlog_open_table(thd, share, 
+                                                 table_share, table)))
+          sql_print_information("NDB: Failed to re-open table %s.%s",
+                                dbname, tabname);
+        */
+      }
+    }
     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)
@@ -1487,11 +1556,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:
@@ -1772,6 +1846,7 @@ int ndbcluster_create_binlog_setup(Ndb *ndb, const char *key,
   /* Handle any trailing share */
   NDB_SHARE *share= (NDB_SHARE*) hash_search(&ndbcluster_open_tables,
                                              (byte*) key, key_len);
+
   if (share && share_may_exist)
   {
     if (share->flags & NSF_NO_BINLOG ||
@@ -2378,6 +2453,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.",
@@ -2389,6 +2468,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

@@ -123,6 +123,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/kernel/ndb_limits.h

@@ -66,6 +66,7 @@
 #define MAX_FRAGMENT_DATA_BYTES (4+(2 * 8 * MAX_REPLICAS * MAX_NDB_NODES))
 #define MAX_NDB_PARTITIONS 1024
 #define MAX_RANGE_DATA (131072+MAX_NDB_PARTITIONS) //0.5 MByte of list data
+#define MAX_WORDS_META_FILE 16382
 
 #define MIN_ATTRBUF ((MAX_ATTRIBUTES_IN_TABLE/24) + 1)
 /*

storage/ndb/include/kernel/signaldata/AlterTable.hpp

@@ -39,6 +39,7 @@ class AlterTableReq {
   friend class NdbEventOperationImpl;
   friend class NdbDictInterface;
   friend class Dbdict;
+  friend class Suma;
 
   /**
    * For printing

storage/ndb/include/ndbapi/NdbDictionary.hpp

@@ -1598,6 +1598,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/kernel/blocks/backup/Backup.hpp

@@ -530,7 +530,7 @@ public:
   Config c_defaults;
   Uint32 m_diskless;
 
-  STATIC_CONST(NO_OF_PAGES_META_FILE = 2);
+  STATIC_CONST(NO_OF_PAGES_META_FILE = MAX_WORDS_META_FILE/BACKUP_WORDS_PER_PAGE);
 
   /**
    * Pools

storage/ndb/src/kernel/blocks/suma/Suma.cpp

@@ -42,6 +42,7 @@
 #include <signaldata/GCPSave.hpp>
 #include <signaldata/CreateTab.hpp>
 #include <signaldata/DropTab.hpp>
+#include <signaldata/AlterTable.hpp>
 #include <signaldata/AlterTab.hpp>
 #include <signaldata/DihFragCount.hpp>
 #include <signaldata/SystemError.hpp>
@@ -3440,7 +3441,7 @@ Suma::execDROP_TAB_CONF(Signal *signal)
   DBUG_VOID_RETURN;
 }
 
-static Uint32 b_dti_buf[10000];
+static Uint32 b_dti_buf[MAX_WORDS_META_FILE];
 
 void
 Suma::execALTER_TAB_REQ(Signal *signal)
@@ -3462,7 +3463,7 @@ Suma::execALTER_TAB_REQ(Signal *signal)
   }
 
   DBUG_PRINT("info",("alter table id: %d[i=%u]", tableId, tabPtr.i));
-
+  Table::State old_state = tabPtr.p->m_state;
   tabPtr.p->m_state = Table::ALTERED;
   // triggers must be removed, waiting for sub stop req for that
 
@@ -3520,6 +3521,11 @@ Suma::execALTER_TAB_REQ(Signal *signal)
       DBUG_PRINT("info",("sent to subscriber %d", subbPtr.i));
     }
   }
+  if (AlterTableReq::getFrmFlag(changeMask))
+  {
+    // Frm changes only are handled on-line
+    tabPtr.p->m_state = old_state;
+  }
   DBUG_VOID_RETURN;
 }
 

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=
@@ -1313,6 +1320,8 @@ NdbDictionaryImpl::putTable(NdbTableImpl *impl)
   
   m_localHash.put(impl->m_internalName.c_str(), info);
   
+  addBlobTables(*impl);
+
   m_ndb.theFirstTupleId[impl->getTableId()] = ~0;
   m_ndb.theLastTupleId[impl->getTableId()]  = ~0;
 }

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;