Commit dc86ad6d authored by unknown's avatar unknown
Browse files

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


sql/ha_ndbcluster.cc:
  Auto merged
sql/ha_ndbcluster.h:
  Auto merged
sql/ha_ndbcluster_binlog.h:
  Auto merged
storage/ndb/include/ndbapi/NdbDictionary.hpp:
  Auto merged
storage/ndb/src/kernel/blocks/backup/Backup.hpp:
  Auto merged
storage/ndb/src/kernel/blocks/suma/Suma.cpp:
  Auto merged
storage/ndb/src/ndbapi/NdbDictionary.cpp:
  Auto merged
storage/ndb/src/ndbapi/NdbDictionaryImpl.cpp:
  Auto merged
storage/ndb/src/ndbapi/NdbEventOperationImpl.cpp:
  Auto merged
sql/ha_ndbcluster_binlog.cc:
  Merge
parents c1473ba8 12e660e8

mysql-test/r/ndb_alter_table_row.result

+9 −0
Original line number Diff line number Diff line
@@ -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

+2 −0
Original line number Diff line number Diff line
@@ -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

+2 −2
Original line number Diff line number Diff line
@@ -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

+152 −72
Original line number Diff line number Diff line
@@ -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)

{

  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

ndbcluster_binlog_open_table(THD *thd, NDB_SHARE *share,

                             TABLE_SHARE *table_share, TABLE *table)

{

    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

+2 −0
Original line number Diff line number Diff line
@@ -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 */