Commit c86dcfd9 authored by unknown's avatar unknown
Browse files

Invalidate table including all it's indexes from binlog thread

parent 42f8c2d2

sql/ha_ndbcluster.cc

+8 −4
Original line number Diff line number Diff line
@@ -519,6 +519,7 @@ void ha_ndbcluster::invalidate_dictionary_cache(bool global) 
  {

    NDBINDEX *index = (NDBINDEX *) m_index[i].index;

    NDBINDEX *unique_index = (NDBINDEX *) m_index[i].unique_index;

    if (!index && !unique_index) continue;

    NDB_INDEX_TYPE idx_type= m_index[i].type;



    switch (idx_type) {
@@ -1076,7 +1077,7 @@ int ha_ndbcluster::get_metadata(const char *path) 
  m_table= (void *)tab; 

  m_table_info= NULL; // Set in external lock

  

  DBUG_RETURN(open_indexes(ndb, table));

  DBUG_RETURN(open_indexes(ndb, table, FALSE));

}



static int fix_unique_index_attr_order(NDB_INDEX_DATA &data,
@@ -1249,7 +1250,7 @@ int ha_ndbcluster::add_index_handle(THD *thd, NDBDICT *dict, KEY *key_info, 
/*

  Associate index handles for each index of a table

*/

int ha_ndbcluster::open_indexes(Ndb *ndb, TABLE *tab)

int ha_ndbcluster::open_indexes(Ndb *ndb, TABLE *tab, bool ignore_error)

{

  uint i;

  int error= 0;
@@ -1263,6 +1264,9 @@ int ha_ndbcluster::open_indexes(Ndb *ndb, TABLE *tab) 
  for (i= 0; i < tab->s->keys; i++, key_info++, key_name++)

  {

    if ((error= add_index_handle(thd, dict, key_info, *key_name, i)))

      if (ignore_error)

        m_index[i].index= m_index[i].unique_index= NULL;

      else

        break;

  }
@@ -3739,7 +3743,7 @@ int ha_ndbcluster::external_lock(THD *thd, int lock_type) 
      {

        m_table= (void *)tab;

        m_table_version = tab->getObjectVersion();

        if (!(my_errno= open_indexes(ndb, table)))

        if (!(my_errno= open_indexes(ndb, table, FALSE)))

          DBUG_RETURN(my_errno);

      }

      m_table_info= tab_info;

sql/ha_ndbcluster.h

+5 −1
Original line number Diff line number Diff line
@@ -691,6 +691,10 @@ static void set_tabname(const char *pathname, char *tabname); 


private:

  friend int ndbcluster_drop_database_impl(const char *path);

  friend int ndb_handle_schema_change(THD *thd, 

                                      Ndb *ndb, NdbEventOperation *pOp,

                                      NDB_SHARE *share);



  int alter_table_name(const char *to);

  static int delete_table(ha_ndbcluster *h, Ndb *ndb,

			  const char *path,
@@ -708,7 +712,7 @@ static void set_tabname(const char *pathname, char *tabname); 
  int create_indexes(Ndb *ndb, TABLE *tab);

  void clear_index(int i);

  void clear_indexes();

  int open_indexes(Ndb *ndb, TABLE *tab);

  int open_indexes(Ndb *ndb, TABLE *tab, bool ignore_error);

  void renumber_indexes(Ndb *ndb, TABLE *tab);

  int drop_indexes(Ndb *ndb, TABLE *tab);

  int add_index_handle(THD *thd, NdbDictionary::Dictionary *dict,

sql/ha_ndbcluster_binlog.cc

+13 −7
Original line number Diff line number Diff line
@@ -1285,7 +1285,7 @@ ndbcluster_update_slock(THD *thd, 
/*

  Handle _non_ data events from the storage nodes

*/

static int

int

ndb_handle_schema_change(THD *thd, Ndb *ndb, NdbEventOperation *pOp,

                         NDB_SHARE *share)

{
@@ -1294,12 +1294,18 @@ ndb_handle_schema_change(THD *thd, Ndb *ndb, NdbEventOperation *pOp, 
  if (pOp->getEventType() != NDBEVENT::TE_CLUSTER_FAILURE &&

      pOp->getReqNodeId() != g_ndb_cluster_connection->node_id())

  {

    ndb->setDatabaseName(share->table->s->db.str);

    ha_ndbcluster::invalidate_dictionary_cache(share->table->s,

                                               ndb,

                                               share->table->s->db.str,

                                               share->table->s->table_name.str,

                                               TRUE);

    TABLE_SHARE *table_share= share->table->s;

    TABLE* table= share->table;

    

    /* 

       Invalidate table and all it's indexes

    */

    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);



    remote_drop_table= 1;

  }