bk commit into 5.1 tree (stewart:1.1797)

[Stewart Smith]

Below is the list of changes that have just been committed into a local
5.1 repository of stewart. When stewart 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
  1.1797 05/04/15 11:47:15 stewart@stripped +1 -0
  Merge ssmith@stripped:/home/bk/mysql-5.1-wl2325
  into mysql.com:/home/stewart/Documents/MySQL/5.1/wl2325

  sql/ha_ndbcluster.cc
    1.206 05/04/15 11:47:12 stewart@stripped +0 -0
    Auto merged

# This is a BitKeeper patch.  What follows are the unified diffs for the
# set of deltas contained in the patch.  The rest of the patch, the part
# that BitKeeper cares about, is below these diffs.
# User:	stewart
# Host:	kennedy.(none)
# Root:	/home/stewart/Documents/MySQL/5.1/wl2325/RESYNC

--- 1.205/sql/ha_ndbcluster.cc	2005-04-14 22:19:42 +10:00
+++ 1.206/sql/ha_ndbcluster.cc	2005-04-15 11:47:12 +10:00
@@ -100,9 +100,10 @@
 
 static byte *ndbcluster_get_key(NDB_SHARE *share,uint *length,
                                 my_bool not_used __attribute__((unused)));
-static NDB_SHARE *get_share(const char *table_name,
+static NDB_SHARE *get_share(const char *key,
 			    bool create_if_not_exists= TRUE,
 			    TABLE *table= 0);
+static int rename_share(NDB_SHARE *share, const char *new_key);
 static void free_share(NDB_SHARE **share);
 
 static int packfrm(const void *data, uint len, const void **pack_data, uint *pack_len);
@@ -5595,6 +5596,55 @@
   return (byte*) share->key;
 }
 
+static int rename_share(NDB_SHARE *share, const char *new_key)
+{
+  NDB_SHARE *tmp;
+  pthread_mutex_lock(&ndbcluster_mutex);
+  uint new_length=(uint) strlen(new_key);
+  if ((tmp=(NDB_SHARE*) hash_search(&ndbcluster_open_tables,
+				     (byte*) new_key,
+				    new_length)))
+  {
+    DBUG_PRINT("error", ("rename_share: %s already exists", new_key));
+    pthread_mutex_unlock(&ndbcluster_mutex);
+    return -1;
+  }
+
+  // save old stuff
+  uint old_length= share->key_length;
+  char *old_key= share->key;
+
+  // temporary set to insert into hash
+  share->key_length=new_length;
+  share->key=(char*)new_key;
+  if (my_hash_insert(&ndbcluster_open_tables, (byte*) share))
+  {
+    share->key_length=old_length;
+    share->key=old_key;
+    pthread_mutex_unlock(&ndbcluster_mutex);
+    return -1;
+  }
+
+  // remove the share (with old name from hash
+  share->key_length=old_length;
+  share->key=old_key;
+  hash_delete(&ndbcluster_open_tables, (byte*) share);
+
+  // not allocate and set the new key, db etc
+  // enough space for key, db, and table_name
+  share->key= alloc_root(&share->mem_root,2*(new_length+1));
+  strmov(share->key, new_key);
+  share->db= share->key+new_length+1;
+  set_dbname(new_key, share->db);
+  share->table_name= share->db+strlen(share->db)+1;
+  set_tabname(new_key, share->table_name);
+
+  // ToDo free old_key...
+
+  pthread_mutex_lock(&ndbcluster_mutex);
+  return 0;
+}
+
 static NDB_SHARE* get_share(const char *key, bool create_if_not_exists, TABLE *table)
 {
   NDB_SHARE *share;
@@ -5610,23 +5660,23 @@
       pthread_mutex_unlock(&ndbcluster_mutex);
       return 0;
     }
-    if ((share=(NDB_SHARE *) my_malloc(sizeof(*share)+length+1+length+1,
-                                       MYF(MY_WME | MY_ZEROFILL))))
+    if ((share=(NDB_SHARE *) my_malloc(sizeof(*share), MYF(MY_WME | MY_ZEROFILL))))
     {
       MEM_ROOT **root_ptr= my_pthread_getspecific_ptr(MEM_ROOT**, THR_MALLOC);
       MEM_ROOT *old_root= *root_ptr;
       init_sql_alloc(&share->mem_root, 1024, 0);
       *root_ptr= &share->mem_root; // remember to reset before return
 
+      // enough space for key, db, and table_name
+      share->key= alloc_root(*root_ptr, 2*(length+1));
       share->key_length=length;
-      share->key=(char*) (share+1);
       strmov(share->key,key);
       if (my_hash_insert(&ndbcluster_open_tables, (byte*) share))
       {
-        pthread_mutex_unlock(&ndbcluster_mutex);
 	free_root(&share->mem_root, MYF(0));
         my_free((gptr) share,0);
 	*root_ptr= old_root;
+        pthread_mutex_unlock(&ndbcluster_mutex);
         return 0;
       }
       thr_lock_init(&share->lock);
@@ -5665,6 +5715,9 @@
 					 table->s->rec_buff_length);
 	  }
 	  table->in_use= injector_thd;
+
+	  table->s->db= share->db;
+	  table->s->table_name= share->table_name;
 	  break;
 	}
 	share->table= table;
@@ -7538,26 +7591,32 @@
 TABLE *binlog_index= 0;
 TABLE_LIST tables;
 
+static int open_binlog_index(THD *thd)
+{
+  bzero((char*)&tables,sizeof(tables));
+  tables.db= "cluster_replication";
+  tables.alias= tables.table_name= "binlog_index";
+  tables.lock_type= TL_WRITE;
+  thd->proc_info= "Opening cluster_replication.binlog_index table";
+  tables.required_type= FRMTYPE_TABLE;
+  uint counter;
+  TABLE_LIST *ttt= &tables;
+  if (open_tables(thd,&ttt,&counter))
+  {
+    sql_print_error("open_tables");
+    return -1;
+  }
+  binlog_index= tables.table;
+  return 0;
+}
+
 static int add_binlog_index(THD *thd, Binlog_index_row *row)
 {
   int error= 0;
-  if (!binlog_index)
+  if (!binlog_index && open_binlog_index(thd))
   {
-    bzero((char*)&tables,sizeof(tables));
-    tables.db= "cluster_replication";
-    tables.alias= tables.table_name= "binlog_index";
-    tables.lock_type= TL_WRITE;
-    thd->proc_info= "Opening cluster_replication.binlog_index table";
-    tables.required_type= FRMTYPE_TABLE;
-    uint counter;
-    TABLE_LIST *ttt= &tables;
-    if (open_tables(thd,&ttt,&counter))
-    {
-      sql_print_error("open_tables");
-      error= -1;
-      goto add_binlog_index_err;
-    }
-    binlog_index= tables.table;
+    error= -1;
+    goto add_binlog_index_err;
   }
 
   if (lock_tables(thd, &tables, 1))
@@ -7575,12 +7634,18 @@
   binlog_index->field[4]->store(row->n_updates);
   binlog_index->field[5]->store(row->n_deletes);
   binlog_index->field[6]->store(row->n_schemaops);
+
+  opt_binlog_row_level= FALSE;
   if (binlog_index->file->ha_write_row(binlog_index->record[0]))
   {
+    opt_binlog_row_level= TRUE;
     sql_print_error("write_row");
     error= -1;
     goto add_binlog_index_err;
   }
+  opt_binlog_row_level= TRUE;
+
+  binlog_index->file->extra(HA_EXTRA_FLUSH);
   mysql_unlock_tables(thd, thd->lock);
   thd->lock=0;
   return 0;
@@ -7836,6 +7901,7 @@
     DBUG_RETURN(0);
   }
 
+#ifdef SYNC_DROP_
   (void) pthread_mutex_lock(&share->mutex);
   int max_timeout= 10;
   while (share->op && max_timeout)
@@ -7857,6 +7923,8 @@
     sql_print_error("NDB delete table: timed out. Ignoring...");
     // ToDo: handle possible mem leak if cluster goes down and share is not freed...
   }
+#endif
+
   DBUG_RETURN(0);
 }
 
@@ -7874,7 +7942,8 @@
 
 static int
 ndb_injector_thread_handle_non_data_event(Ndb *ndb, NdbEventOperation *pOp,
-					  Binlog_index_row &row)
+					  Binlog_index_row &row,
+					  injector::transaction &trans)
 {
   NDB_SHARE *share= (NDB_SHARE *)pOp->getCustomData();
   const NDBTAB *ndbtab= pOp->getTable();
@@ -7918,6 +7987,8 @@
     (void) pthread_mutex_unlock(&share->mutex);
     (void) pthread_cond_signal(&injector_cond);
       
+    // ToDo Flush before releasing table object
+
     // release previous lock
     free_share(&share);
     // ToDo: remove printout
@@ -8009,6 +8080,8 @@
   Uint64 waitGCI= latestGCI;
   ndb->setWaitGCI(waitGCI);
   thd->query_id= 0; // to keep valgrind quiet
+  thd->db="";
+  open_binlog_index(thd);
   for (;;)
   {
     /**
@@ -8022,9 +8095,13 @@
 
     if (binlog_index && binlog_index->s->version < refresh_version)
     {
-      close_thread_tables(thd);
-      binlog_index= 0;
+      if (binlog_index->s->version < refresh_version)
+      {
+	close_thread_tables(thd);
+	binlog_index= 0;
+      }
     }
+
     if ((latestGCI = ndb->getLatestGCI()) < waitGCI)
     {
       /**
@@ -8114,7 +8191,7 @@
 	if ( (unsigned)type >= (unsigned)NDBEVENT::TE_FIRST_NON_DATA_EVENT )
 	{
 	  // a non-data event
-	  ndb_injector_thread_handle_non_data_event(ndb,pOp,row);
+	  ndb_injector_thread_handle_non_data_event(ndb,pOp,row,trans);
 	  continue;
 	}