bk commit into 5.1 tree (tomas:1.2631) BUG#31082

[tomas]

Below is the list of changes that have just been committed into a local
5.1 repository of tomas. When tomas does a push these changes will
be propagated to the main repository and, within 24 hours after the
push, to the public repository.
For information on how to access the public repository
see http://dev.mysql.com/doc/mysql/en/installing-source-tree.html

ChangeSet@stripped, 2007-09-18 17:10:20+02:00, tomas@stripped +4 -0
  Bug #31082 Online add column under load hits assert during replication
  - make sure we have both versions of the table object available to be used for binlog
injection
  - move table object to event_data struct, hence we have one per ongoing event
subscription

  sql/ha_ndbcluster.cc@stripped, 2007-09-18 17:10:16+02:00, tomas@stripped +25
-30
    Bug #31082 Online add column under load hits assert during replication
    - make sure we have both versions of the table object available to be used for binlog
injection
    - move table object to event_data struct, hence we have one per ongoing event
subscription

  sql/ha_ndbcluster.h@stripped, 2007-09-18 17:10:16+02:00, tomas@stripped +2
-3
    Bug #31082 Online add column under load hits assert during replication
    - make sure we have both versions of the table object available to be used for binlog
injection
    - move table object to event_data struct, hence we have one per ongoing event
subscription

  sql/ha_ndbcluster_binlog.cc@stripped, 2007-09-18 17:10:16+02:00,
tomas@stripped +103 -109
    Bug #31082 Online add column under load hits assert during replication
    - make sure we have both versions of the table object available to be used for binlog
injection
    - move table object to event_data struct, hence we have one per ongoing event
subscription

  sql/ha_ndbcluster_binlog.h@stripped, 2007-09-18 17:10:16+02:00,
tomas@stripped +13 -8
    Bug #31082 Online add column under load hits assert during replication
    - make sure we have both versions of the table object available to be used for binlog
injection
    - move table object to event_data struct, hence we have one per ongoing event
subscription

diff -Nrup a/sql/ha_ndbcluster.cc b/sql/ha_ndbcluster.cc
--- a/sql/ha_ndbcluster.cc	2007-09-12 14:04:58 +02:00
+++ b/sql/ha_ndbcluster.cc	2007-09-18 17:10:16 +02:00
@@ -8704,11 +8704,18 @@ static void print_share(const char* wher
           share->key, share->key_length);
 
 #ifdef HAVE_NDB_BINLOG
-  if (share->table)
+  Ndb_event_data *event_data= 0;
+  if (share->event_data)
+    event_data= share->event_data;
+  else if (share->op)
+    event_data= (Ndb_event_data *) share->op->getCustomData();
+  if (event_data)
+  {
     fprintf(DBUG_FILE,
-            "  - share->table: %p %s.%s\n",
-            share->table, share->table->s->db.str,
-            share->table->s->table_name.str);
+            "  - event_data->table: %p %s.%s\n",
+            event_data->table, event_data->table->s->db.str,
+            event_data->table->s->table_name.str);
+  }
 #endif
 }
 
@@ -8935,12 +8942,17 @@ int ndbcluster_rename_share(NDB_SHARE *s
   ha_ndbcluster::set_tabname(share->new_key, share->table_name);
 
   dbug_print_share("ndbcluster_rename_share:", share);
-  if (share->table)
+  Ndb_event_data *event_data= 0;
+  if (share->event_data)
+    event_data= share->event_data;
+  else if (share->op)
+    event_data= (Ndb_event_data *) share->op->getCustomData();
+  if (event_data && event_data->table)
   {
-    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);
+    event_data->table->s->db.str= share->db;
+    event_data->table->s->db.length= strlen(share->db);
+    event_data->table->s->table_name.str= share->table_name;
+    event_data->table->s->table_name.length= strlen(share->table_name);
   }
   /* else rename will be handled when the ALTER event comes */
   share->old_names= old_key;
@@ -9086,28 +9098,11 @@ void ndbcluster_real_free_share(NDB_SHAR
   pthread_mutex_destroy(&(*share)->mutex);
 
 #ifdef HAVE_NDB_BINLOG
-  if ((*share)->table)
+  (*share)->new_op= 0;
+  if ((*share)->event_data)
   {
-    // (*share)->table->mem_root is freed by closefrm
-    (*share)->new_op= 0;
-    if ((*share)->op)
-    {
-      Ndb_event_data *event_data= (Ndb_event_data *) (*share)->op->getCustomData();
-      if (event_data)
-      {
-        delete event_data;
-        (*share)->op->setCustomData(NULL);
-      }
-    }
-    closefrm((*share)->table, 0);
-    // (*share)->table_share->mem_root is freed by free_table_share
-    free_table_share((*share)->table_share);
-#ifndef DBUG_OFF
-    bzero((uchar*)(*share)->table_share, sizeof(*(*share)->table_share));
-    bzero((uchar*)(*share)->table, sizeof(*(*share)->table));
-    (*share)->table_share= 0;
-    (*share)->table= 0;
-#endif
+    delete (*share)->event_data;
+    (*share)->event_data= 0;
   }
 #endif
   free_root(&(*share)->mem_root, MYF(0));
diff -Nrup a/sql/ha_ndbcluster.h b/sql/ha_ndbcluster.h
--- a/sql/ha_ndbcluster.h	2007-09-06 15:04:32 +02:00
+++ b/sql/ha_ndbcluster.h	2007-09-18 17:10:16 +02:00
@@ -42,6 +42,7 @@ class NdbBlob;
 class NdbIndexStat;
 class NdbEventOperation;
 class ha_ndbcluster_cond;
+class Ndb_event_data;
 
 // connectstring to cluster if given by mysqld
 extern const char *ndbcluster_connectstring;
@@ -137,11 +138,9 @@ typedef struct st_ndbcluster_share {
 #ifdef HAVE_NDB_BINLOG
   uint32 connect_count;
   uint32 flags;
+  Ndb_event_data *event_data; // Place holder before NdbEventOperation is created
   NdbEventOperation *op;
-  Uint64 op_gci;
   char *old_names; // for rename table
-  TABLE_SHARE *table_share;
-  TABLE *table;
   MY_BITMAP *subscriber_bitmap;
   NdbEventOperation *new_op;
 #endif
diff -Nrup a/sql/ha_ndbcluster_binlog.cc b/sql/ha_ndbcluster_binlog.cc
--- a/sql/ha_ndbcluster_binlog.cc	2007-09-07 11:15:41 +02:00
+++ b/sql/ha_ndbcluster_binlog.cc	2007-09-18 17:10:16 +02:00
@@ -327,13 +327,12 @@ static void
 ndbcluster_binlog_close_table(THD *thd, NDB_SHARE *share)
 {
   DBUG_ENTER("ndbcluster_binlog_close_table");
-  if (share->table_share)
+  Ndb_event_data *event_data= share->event_data;
+  if (event_data)
   {
-    closefrm(share->table, 1);
-    share->table_share= 0;
-    share->table= 0;
+    delete event_data;
+    share->event_data= 0;
   }
-  DBUG_ASSERT(share->table == 0);
   DBUG_VOID_RETURN;
 }
 
@@ -346,26 +345,40 @@ ndbcluster_binlog_close_table(THD *thd, 
 */
 
 static int
-ndbcluster_binlog_open_table(THD *thd, NDB_SHARE *share,
-                             TABLE_SHARE *table_share, TABLE *table)
+ndbcluster_binlog_open_table(THD *thd, NDB_SHARE *share)
 {
   int error;
+  DBUG_ASSERT(share->event_data == 0);
+  Ndb_event_data *event_data= share->event_data= new Ndb_event_data(share);
   DBUG_ENTER("ndbcluster_binlog_open_table");
-  
+
+  MEM_ROOT **root_ptr=
+    my_pthread_getspecific_ptr(MEM_ROOT**, THR_MALLOC);
+  MEM_ROOT *old_root= *root_ptr;
+  init_sql_alloc(&event_data->mem_root, 1024, 0);
+  *root_ptr= &event_data->mem_root;
+
+  TABLE_SHARE *table_share= event_data->table_share= 
+    (TABLE_SHARE*)alloc_root(&event_data->mem_root, sizeof(TABLE_SHARE));
+  TABLE *table= event_data->table= 
+    (TABLE*)alloc_root(&event_data->mem_root, sizeof(TABLE));
+
   safe_mutex_assert_owner(&LOCK_open);
   init_tmp_table_share(table_share, share->db, 0, share->table_name, 
                        share->key);
-  if ((error= open_table_def(thd, table_share, 0)))
+  if ((error= open_table_def(thd, table_share, 0)) ||
+      (error= open_table_from_share(thd, table_share, "", 0, 
+                                    (uint) (OPEN_FRM_FILE_ONLY | DELAYED_OPEN |
READ_ALL),
+                                    0, table, OTM_OPEN)))
   {
-    DBUG_PRINT("error", ("open_table_def failed: %d my_errno: %d", error, my_errno));
-    free_table_share(table_share);
-    DBUG_RETURN(error);
-  }
-  if ((error= open_table_from_share(thd, table_share, "", 0 /* don't allocate buffers */,

-                                    (uint) READ_ALL, 0, table, OTM_OPEN)))
-  {
-    DBUG_PRINT("error", ("open_table_from_share failed %d my_errno: %d", error,
my_errno));
+    DBUG_PRINT("error", ("open_table_def/open_table_from_share failed: %d my_errno: %d",
+                         error, my_errno));
     free_table_share(table_share);
+    event_data->table= 0;
+    event_data->table_share= 0;
+    delete event_data;
+    share->event_data= 0;
+    *root_ptr= old_root;
     DBUG_RETURN(error);
   }
   assign_new_table_id(table_share);
@@ -376,16 +389,12 @@ ndbcluster_binlog_open_table(THD *thd, N
   table->s->db.length= strlen(share->db);
   table->s->table_name.str= share->table_name;
   table->s->table_name.length= strlen(share->table_name);
-  
-  DBUG_ASSERT(share->table_share == 0);
-  share->table_share= table_share;
-  DBUG_ASSERT(share->table == 0);
-  share->table= table;
   /* We can't use 'use_all_columns()' as the file object is not setup yet */
   table->column_bitmaps_set_no_signal(&table->s->all_set,
&table->s->all_set);
 #ifndef DBUG_OFF
   dbug_print_table("table", table);
 #endif
+  *root_ptr= old_root;
   DBUG_RETURN(0);
 }
 
@@ -404,8 +413,8 @@ int ndbcluster_binlog_init_share(NDB_SHA
   share->connect_count= g_ndb_cluster_connection->get_connect_count();
 
   share->op= 0;
-  share->table= 0;
   share->new_op= 0;
+  share->event_data= 0;
 
   if (!ndb_schema_share &&
       strcmp(share->db, NDB_REP_DB) == 0 &&
@@ -446,15 +455,11 @@ int ndbcluster_binlog_init_share(NDB_SHA
   }
   while (1) 
   {
-    int error;
-    TABLE_SHARE *table_share= (TABLE_SHARE *) alloc_root(mem_root, sizeof(*table_share));
-    TABLE *table= (TABLE*) alloc_root(mem_root, sizeof(*table));
-    if ((error= ndbcluster_binlog_open_table(thd, share, table_share, table)))
+    if ((error= ndbcluster_binlog_open_table(thd, share)))
       break;
-
-    if (table->s->primary_key == MAX_KEY)
+    if (share->event_data->table->s->primary_key == MAX_KEY)
       share->flags|= NSF_HIDDEN_PK;
-    if (table->s->blob_fields != 0)
+    if (share->event_data->table->s->blob_fields != 0)
       share->flags|= NSF_BLOB_FLAG;
     break;
   }
@@ -937,8 +942,7 @@ struct Cluster_schema
 static void ndbcluster_get_schema(Ndb_event_data *event_data,
                                   Cluster_schema *s)
 {
-  NDB_SHARE *share= event_data->share;
-  TABLE *table= share->table;
+  TABLE *table= event_data->table;
   Field **field;
   /* unpack blob values */
   uchar* blobs_buffer= 0;
@@ -1622,7 +1626,7 @@ ndb_handle_schema_change(THD *thd, Ndb *
 {
   DBUG_ENTER("ndb_handle_schema_change");
   NDB_SHARE *share= event_data->share;
-  TABLE_SHARE *table_share= share->table_share;
+  TABLE_SHARE *table_share= event_data->table_share;
   const char *tabname= table_share->table_name.str;
   const char *dbname= table_share->db.str;
   bool do_close_cached_tables= FALSE;
@@ -2143,7 +2147,7 @@ ndb_binlog_thread_handle_schema_event_po
         }
         {
           if (ndb_extra_logging > 9)
-            sql_print_information("NDB: renaming files start");
+            sql_print_information("NDB Binlog: renaming files start");
           pthread_mutex_lock(&LOCK_open);
           char from[FN_REFLEN];
           char to[FN_REFLEN];
@@ -2154,7 +2158,7 @@ ndb_binlog_thread_handle_schema_event_po
           rename_file_ext(from, to, ".frm");
           pthread_mutex_unlock(&LOCK_open);
           if (ndb_extra_logging > 9)
-            sql_print_information("NDB: renaming files done");
+            sql_print_information("NDB Binlog: renaming files done");
         }
         break;
       case SOT_RENAME_TABLE_PREPARE:
@@ -2288,8 +2292,8 @@ ndb_binlog_thread_handle_schema_event_po
                                   (const uchar*) ndbtab->getFrmData())) ||
                 (error= writefrm(key, data, length)))
             {
-              sql_print_information("NDB: Failed write frm for %s.%s, error %d",
-                                    schema->db, schema->name, error);
+              sql_print_error("NDB: Failed write frm for %s.%s, error %d",
+                              schema->db, schema->name, error);
             }
           }
           my_free((char*)data, MYF(MY_ALLOW_ZERO_PTR));
@@ -2300,28 +2304,26 @@ ndb_binlog_thread_handle_schema_event_po
         else
         {
           if (ndb_extra_logging > 9)
-            sql_print_information("NDB: handeling online alter/rename");
+            sql_print_information("NDB Binlog: handeling online alter/rename");
 
           (void) pthread_mutex_lock(&share->mutex);
           ndbcluster_binlog_close_table(thd, share);
 
-          TABLE_SHARE *new_table_share=
-            (share->table_share) ?
-            share->table_share // allready allocated
-            : (TABLE_SHARE *) alloc_root(&share->mem_root,
sizeof(*new_table_share));
-          TABLE *new_table= (share->table) ?
-            share->table // allready allocated
-            : (TABLE*) alloc_root(&share->mem_root, sizeof(*new_table));
-          if ((error= ndbcluster_binlog_open_table(thd, share,
-                                                   new_table_share, new_table)))
-            sql_print_information("NDB: Failed to re-open table %s.%s",
-                                  schema->db, schema->name);
+          if ((error= ndbcluster_binlog_open_table(thd, share)))
+            sql_print_error("NDB Binlog: Failed to re-open table %s.%s",
+                            schema->db, schema->name);
           pthread_mutex_unlock(&LOCK_open);
-
+          if (error)
+            (void) pthread_mutex_unlock(&share->mutex);
+        }
+        if (!error && share)
+        {
+          if (share->event_data->table->s->primary_key == MAX_KEY)
+            share->flags|= NSF_HIDDEN_PK;
           /*
             Refresh share->flags to handle added BLOB columns
           */
-          if (share->table->s->blob_fields != 0)
+          if (share->event_data->table->s->blob_fields != 0)
             share->flags|= NSF_BLOB_FLAG;
 
           /*
@@ -2332,6 +2334,7 @@ ndb_binlog_thread_handle_schema_event_po
           NdbEventOperation *tmp_op= share->op;
           share->new_op= 0;
           share->op= 0;
+
           if (ndbcluster_create_event_ops(thd, share, ndbtab, event_name.c_ptr()))
           {
             sql_print_error("NDB Binlog:"
@@ -2346,7 +2349,7 @@ ndb_binlog_thread_handle_schema_event_po
           (void) pthread_mutex_unlock(&share->mutex);
 
           if (ndb_extra_logging > 9)
-            sql_print_information("NDB: handeling online alter/rename done");
+            sql_print_information("NDB Binlog: handeling online alter/rename done");
         }
         break;
       }
@@ -3062,6 +3065,7 @@ ndbcluster_create_event_ops(THD *thd, ND
     DBUG_RETURN(0);
   }
 
+  Ndb_event_data *event_data= share->event_data;
   int do_ndb_schema_share= 0, do_ndb_apply_status_share= 0;
   if (!ndb_schema_share && strcmp(share->db, NDB_REP_DB) == 0 &&
       strcmp(share->table_name, NDB_SCHEMA_TABLE) == 0)
@@ -3077,8 +3081,9 @@ ndbcluster_create_event_ops(THD *thd, ND
 
   if (share->op)
   {
-    Ndb_event_data *event_data= (Ndb_event_data *) share->op->getCustomData();
+    event_data= (Ndb_event_data *) share->op->getCustomData();
     assert(event_data->share == share);
+    assert(share->event_data == 0);
 
     DBUG_ASSERT(share->use_count > 1);
     sql_print_error("NDB Binlog: discover reusing old ev op");
@@ -3089,7 +3094,8 @@ ndbcluster_create_event_ops(THD *thd, ND
     DBUG_RETURN(0);
   }
 
-  TABLE *table= share->table;
+  DBUG_ASSERT(event_data != 0);
+  TABLE *table= event_data->table;
 
   int retries= 100;
   /*
@@ -3139,35 +3145,18 @@ ndbcluster_create_event_ops(THD *thd, ND
       op->mergeEvents(TRUE); // currently not inherited from event
 
     uint n_columns= ndbtab->getNoOfColumns();
-    uint n_fields= table ? table->s->fields : 0; // XXX ???
-    Ndb_event_data *event_data= new Ndb_event_data(share, n_fields);
+    uint n_fields= table->s->fields;
     uint val_length= sizeof(NdbValue) * n_columns;
-    uint rec_length= table->s->rec_buff_length;
+
     /*
        Allocate memory globally so it can be reused after online alter table
     */
     if (my_multi_malloc(MYF(MY_WME),
                         &event_data->ndb_value[0],
                         val_length,
-	                &event_data->ndb_value[1],
+                        &event_data->ndb_value[1],
                         val_length,
-                        &event_data->record[0],
-                        rec_length,
-                        &event_data->record[1],
-                        rec_length,
-                        NULL) != 0)
-    {
-      /*
-         Set table record pointers to point to new records 
-      */
-      my_ptrdiff_t row_offset= event_data->record[0] - table->record[0];
-      Field **p_field;
-      for (p_field= table->field; *p_field; p_field++)
-        (*p_field)->move_field_offset(row_offset);
-      table->record[0]= event_data->record[0];
-      table->record[1]= event_data->record[1];
-    }
-    else
+                        NULL) == 0)
     {
       DBUG_PRINT("info", ("Failed to allocate records for event operation"));
       DBUG_RETURN(-1);
@@ -3179,14 +3168,14 @@ ndbcluster_create_event_ops(THD *thd, ND
       NdbValue attr0, attr1;
       if (j < n_fields)
       {
-        Field *f= share->table->field[j];
+        Field *f= table->field[j];
         if (is_ndb_compatible_type(f))
         {
           DBUG_PRINT("info", ("%s compatible", col_name));
           attr0.rec= op->getValue(col_name, (char*) f->ptr);
           attr1.rec= op->getPreValue(col_name,
-                                     (f->ptr - share->table->record[0]) +
-                                     (char*) share->table->record[1]);
+                                     (f->ptr - table->record[0]) +
+                                     (char*) table->record[1]);
         }
         else if (! (f->flags & BLOB_FLAG))
         {
@@ -3210,8 +3199,6 @@ ndbcluster_create_event_ops(THD *thd, ND
                                 op->getNdbError().code,
                                 op->getNdbError().message,
                                 "NDB");
-            delete event_data;
-            op->setCustomData(NULL);
             ndb->dropEventOperation(op);
             pthread_mutex_unlock(&injector_mutex);
             DBUG_RETURN(-1);
@@ -3236,8 +3223,9 @@ ndbcluster_create_event_ops(THD *thd, ND
                           j, (long) attr1.ptr));
     }
     op->setCustomData((void *) event_data); // set before execute
+    share->event_data= 0;                   // take over event data
     share->op= op; // assign op in NDB_SHARE
-    share->op_gci= op->getLatestGCI();
+
     if (op->execute())
     {
       share->op= NULL;
@@ -3255,12 +3243,8 @@ ndbcluster_create_event_ops(THD *thd, ND
                         event_name,
                         op->getNdbError().code, op->getNdbError().message);
       }
-      Ndb_event_data *event_data= (Ndb_event_data *) op->getCustomData();
-      if (event_data)
-      {
-        delete event_data;
-        op->setCustomData(NULL);
-      }
+      share->event_data= event_data;
+      op->setCustomData(NULL);
       ndb->dropEventOperation(op);
       pthread_mutex_unlock(&injector_mutex);
       if (retries)
@@ -3781,6 +3765,7 @@ ndb_binlog_thread_handle_data_event(Ndb 
                                     injector::transaction &trans)
 {
   Ndb_event_data *event_data= (Ndb_event_data *) pOp->getCustomData();
+  TABLE *table= event_data->table;
   NDB_SHARE *share= event_data->share;
   if (pOp != share->op)
   {
@@ -3797,13 +3782,12 @@ ndb_binlog_thread_handle_data_event(Ndb 
       case NDBEVENT::TE_UPDATE:
       {
         /* unpack data to fetch orig_server_id and orig_epoch */
-        TABLE *table= share->table;
-        uint n_fields= event_data->no_fields;
+        uint n_fields= table->s->fields;
         MY_BITMAP b;
         uint32 bitbuf[128 / (sizeof(uint32) * 8)];
         bitmap_init(&b, bitbuf, n_fields, FALSE);
         bitmap_set_all(&b);
-        ndb_unpack_record(table, event_data->ndb_value[0], &b,
event_data->record[0]);
+        ndb_unpack_record(table, event_data->ndb_value[0], &b,
table->record[0]);
         /* store */
         ndb_binlog_index_row *row= ndb_find_binlog_index_row
           (rows, ((Field_long *)table->field[0])->val_int(), 1);
@@ -3840,13 +3824,12 @@ ndb_binlog_thread_handle_data_event(Ndb 
     return 0;
   }
 
-  TABLE *table= share->table;
   DBUG_ASSERT(trans.good());
   DBUG_ASSERT(table != 0);
 
   dbug_print_table("table", table);
 
-  uint n_fields= event_data->no_fields;
+  uint n_fields= table->s->fields;
   DBUG_PRINT("info", ("Assuming %u columns for table %s",
                       n_fields, table->s->table_name.str));
   MY_BITMAP b;
@@ -3888,11 +3871,11 @@ ndb_binlog_thread_handle_data_event(Ndb 
                                                ptrdiff);
         DBUG_ASSERT(ret == 0);
       }
-      ndb_unpack_record(table, event_data->ndb_value[0], &b,
event_data->record[0]);
+      ndb_unpack_record(table, event_data->ndb_value[0], &b, table->record[0]);
       IF_DBUG(int ret=) trans.write_row(originating_server_id,
                                         injector::transaction::table(table,
                                                                      TRUE),
-                                        &b, n_fields, event_data->record[0]);
+                                        &b, n_fields, table->record[0]);
       DBUG_ASSERT(ret == 0);
     }
     break;
@@ -3920,19 +3903,19 @@ ndb_binlog_thread_handle_data_event(Ndb 
 
       if (share->flags & NSF_BLOB_FLAG)
       {
-        my_ptrdiff_t ptrdiff= event_data->record[n] - table->record[0];
+        my_ptrdiff_t ptrdiff= table->record[n] - table->record[0];
         IF_DBUG(int ret =) get_ndb_blobs_value(table, event_data->ndb_value[n],
                                                blobs_buffer[n],
                                                blobs_buffer_size[n],
                                                ptrdiff);
         DBUG_ASSERT(ret == 0);
       }
-      ndb_unpack_record(table, event_data->ndb_value[n], &b,
event_data->record[n]);
-      DBUG_EXECUTE("info", print_records(table, event_data->record[n]););
+      ndb_unpack_record(table, event_data->ndb_value[n], &b, table->record[n]);
+      DBUG_EXECUTE("info", print_records(table, table->record[n]););
       IF_DBUG(int ret =) trans.delete_row(originating_server_id,
                                           injector::transaction::table(table,
                                                                        TRUE),
-                                          &b, n_fields, event_data->record[n]);
+                                          &b, n_fields, table->record[n]);
       DBUG_ASSERT(ret == 0);
     }
     break;
@@ -3951,8 +3934,8 @@ ndb_binlog_thread_handle_data_event(Ndb 
         DBUG_ASSERT(ret == 0);
       }
       ndb_unpack_record(table, event_data->ndb_value[0],
-                        &b, event_data->record[0]);
-      DBUG_EXECUTE("info", print_records(table, event_data->record[0]););
+                        &b, table->record[0]);
+      DBUG_EXECUTE("info", print_records(table, table->record[0]););
       if (table->s->primary_key != MAX_KEY) 
       {
         /*
@@ -3961,7 +3944,7 @@ ndb_binlog_thread_handle_data_event(Ndb 
         */
         IF_DBUG(int ret =) trans.write_row(originating_server_id,
                                            injector::transaction::table(table, TRUE),
-                                           &b, n_fields, event_data->record[0]);//
after values
+                                           &b, n_fields, table->record[0]);//
after values
         DBUG_ASSERT(ret == 0);
       }
       else
@@ -3972,21 +3955,21 @@ ndb_binlog_thread_handle_data_event(Ndb 
         */
         if (share->flags & NSF_BLOB_FLAG)
         {
-          my_ptrdiff_t ptrdiff= event_data->record[1] - table->record[0];
+          my_ptrdiff_t ptrdiff= table->record[1] - table->record[0];
           IF_DBUG(int ret =) get_ndb_blobs_value(table, event_data->ndb_value[1],
                                                  blobs_buffer[1],
                                                  blobs_buffer_size[1],
                                                  ptrdiff);
           DBUG_ASSERT(ret == 0);
         }
-        ndb_unpack_record(table, event_data->ndb_value[1], &b,
event_data->record[1]);
-        DBUG_EXECUTE("info", print_records(table, event_data->record[1]););
+        ndb_unpack_record(table, event_data->ndb_value[1], &b,
table->record[1]);
+        DBUG_EXECUTE("info", print_records(table, table->record[1]););
         IF_DBUG(int ret =) trans.update_row(originating_server_id,
                                             injector::transaction::table(table,
                                                                          TRUE),
                                             &b, n_fields,
-                                            event_data->record[1], // before values
-                                            event_data->record[0]);// after values
+                                            table->record[1], // before values
+                                            table->record[0]);// after values
         DBUG_ASSERT(ret == 0);
       }
     }
@@ -4534,7 +4517,7 @@ restart:
               continue;
             }
             // this should not happen
-            if (share == NULL || share->table == NULL)
+            if (share == NULL || event_data->table == NULL)
             {
               DBUG_PRINT("info", ("no share or table %s!",
                                   gci_op->getEvent()->getTable()->getName()));
@@ -4545,7 +4528,7 @@ restart:
               // skip this table, it is handled specially
               continue;
             }
-            TABLE *table= share->table;
+            TABLE *table= event_data->table;
 #ifndef DBUG_OFF
             const LEX_STRING &name= table->s->table_name;
 #endif
@@ -4575,7 +4558,18 @@ restart:
         {
           if (ndb_apply_status_share)
           {
-            TABLE *table= ndb_apply_status_share->table;
+            Ndb_event_data *event_data= 0;
+            if (ndb_apply_status_share->event_data)
+            {
+              event_data= ndb_apply_status_share->event_data;
+            }
+            else if (ndb_apply_status_share->op)
+            {
+              event_data= 
+                (Ndb_event_data *) ndb_apply_status_share->op->getCustomData();
+            }
+            DBUG_ASSERT(event_data);
+            TABLE *table= event_data->table;
 
 #ifndef DBUG_OFF
             const LEX_STRING& name= table->s->table_name;
diff -Nrup a/sql/ha_ndbcluster_binlog.h b/sql/ha_ndbcluster_binlog.h
--- a/sql/ha_ndbcluster_binlog.h	2007-09-05 15:21:30 +02:00
+++ b/sql/ha_ndbcluster_binlog.h	2007-09-18 17:10:16 +02:00
@@ -36,29 +36,34 @@ extern ulong ndb_extra_logging;
 
 extern handlerton *ndbcluster_hton;
 
-class Ndb_event_data {
+class Ndb_event_data
+{
 public:
-  Ndb_event_data(NDB_SHARE *the_share, uint n_fields) : 
-  share(the_share), no_fields(n_fields)
+  Ndb_event_data(NDB_SHARE *the_share) : 
+    share(the_share), table_share((TABLE_SHARE *)0), table((TABLE *)0)
   {
     ndb_value[0]= 0;
     ndb_value[1]= 0;
-    record[0]= 0;
-    record[1]= 0;
   }
   ~Ndb_event_data()
   {
+    if (table)
+      closefrm(table, 1);
+    free_root(&mem_root, MYF(0));
+    table_share= 0;
+    table= 0;
     share= 0;
     /*
-       ndbvalue[] and record[] are allocated with my_multi_malloc
+       ndbvalue[] allocated with my_multi_malloc
        so only first pointer should be freed  
     */
     my_free(ndb_value[0], MYF(MY_WME|MY_ALLOW_ZERO_PTR));
   }
+  MEM_ROOT mem_root;
+  TABLE_SHARE *table_share;
+  TABLE *table;
   NDB_SHARE *share;
-  uint no_fields;
   NdbValue *ndb_value[2];
-  uchar *record[2]; // pointer to allocated records for receiving data
 };
 
 /*