6 files changed, 86 insertions, 56 deletions

diff --git a/sql/examples/ha_archive.cc b/sql/examples/ha_archive.cc
index 070469819ee..98ed65fc786 100644
--- a/sql/examples/ha_archive.cc
+++ b/sql/examples/ha_archive.cc
@@ -752,7 +752,7 @@ int ha_archive::rnd_next(byte *buf)
 }
 
 
-/* 
+/*
   Thanks to the table flag HA_REC_NOT_IN_SEQ this will be called after
   each call to ha_archive::rnd_next() if an ordering of the rows is
   needed.
@@ -761,7 +761,7 @@ int ha_archive::rnd_next(byte *buf)
 void ha_archive::position(const byte *record)
 {
   DBUG_ENTER("ha_archive::position");
-  ha_store_ptr(ref, ref_length, current_position);
+  my_store_ptr(ref, ref_length, current_position);
   DBUG_VOID_RETURN;
 }
 
@@ -778,7 +778,7 @@ int ha_archive::rnd_pos(byte * buf, byte *pos)
   DBUG_ENTER("ha_archive::rnd_pos");
   statistic_increment(table->in_use->status_var.ha_read_rnd_next_count,
 		      &LOCK_status);
-  current_position= ha_get_ptr(pos, ref_length);
+  current_position= my_get_ptr(pos, ref_length);
   z_off_t seek= gzseek(archive, current_position, SEEK_SET);
 
   DBUG_RETURN(get_row(archive, buf));
diff --git a/sql/examples/ha_archive.h b/sql/examples/ha_archive.h
index a511b5c67d6..e2c40c880a4 100644
--- a/sql/examples/ha_archive.h
+++ b/sql/examples/ha_archive.h
@@ -106,6 +106,6 @@ public:
                              enum thr_lock_type lock_type);
 };
 
-bool archive_db_init(void);
+handlerton *archive_db_init(void);
 bool archive_db_end(void);
 
diff --git a/sql/ha_ndbcluster.cc b/sql/ha_ndbcluster.cc
index f62a65cb109..8360c60073c 100644
--- a/sql/ha_ndbcluster.cc
+++ b/sql/ha_ndbcluster.cc
@@ -12,7 +12,7 @@
 
   You should have received a copy of the GNU General Public License
   along with this program; if not, write to the Free Software
-  Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA 
+  Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
 */
 
 /*
@@ -45,6 +45,23 @@ static const int max_transactions= 256;
 
 static const char *ha_ndb_ext=".ndb";
 
+static int ndbcluster_close_connection(THD *thd);
+static int ndbcluster_commit(THD *thd, bool all);
+static int ndbcluster_rollback(THD *thd, bool all);
+
+static handlerton ndbcluster_hton = {
+  0, /* slot */
+  0, /* savepoint size */
+  ndbcluster_close_connection,
+  NULL, /* savepoint_set */
+  NULL, /* savepoint_rollback */
+  NULL, /* savepoint_release */
+  ndbcluster_commit,
+  ndbcluster_rollback,
+  NULL, /* prepare */
+  NULL  /* recover */
+};
+
 #define NDB_HIDDEN_PRIMARY_KEY_LENGTH 8
 
 
@@ -258,6 +275,8 @@ Thd_ndb::Thd_ndb()
   ndb= new Ndb(g_ndb_cluster_connection, "");
   lock_count= 0;
   count= 0;
+  all= NULL;
+  stmt= NULL;
   error= 0;
 }
 
@@ -269,9 +288,17 @@ Thd_ndb::~Thd_ndb()
 }
 
 inline
+Thd_ndb *
+get_thd_ndb(THD *thd) { return (Thd_ndb *) thd->ha_data[ndbcluster_hton.slot]; }
+
+inline
+void
+set_thd_ndb(THD *thd, Thd_ndb *thd_ndb) { thd->ha_data[ndbcluster_hton.slot]= thd_ndb; }
+
+inline
 Ndb *ha_ndbcluster::get_ndb()
 {
-  return ((Thd_ndb*)current_thd->transaction.thd_ndb)->ndb;
+  return get_thd_ndb(table->in_use)->ndb;
 }
 
 /*
@@ -315,7 +342,7 @@ void ha_ndbcluster::records_update()
   }
   {
     THD *thd= current_thd;
-    if (((Thd_ndb*)(thd->transaction.thd_ndb))->error)
+    if (get_thd_ndb(thd)->error)
       info->no_uncommitted_rows_count= 0;
   }
   records= info->records+ info->no_uncommitted_rows_count;
@@ -327,8 +354,7 @@ void ha_ndbcluster::no_uncommitted_rows_execute_failure()
   if (m_ha_not_exact_count)
     return;
   DBUG_ENTER("ha_ndbcluster::no_uncommitted_rows_execute_failure");
-  THD *thd= current_thd;
-  ((Thd_ndb*)(thd->transaction.thd_ndb))->error= 1;
+  get_thd_ndb(table->in_use)->error= 1;
   DBUG_VOID_RETURN;
 }
 
@@ -338,7 +364,7 @@ void ha_ndbcluster::no_uncommitted_rows_init(THD *thd)
     return;
   DBUG_ENTER("ha_ndbcluster::no_uncommitted_rows_init");
   struct Ndb_table_local_info *info= (struct Ndb_table_local_info *)m_table_info;
-  Thd_ndb *thd_ndb= (Thd_ndb *)thd->transaction.thd_ndb;
+  Thd_ndb *thd_ndb= get_thd_ndb(thd);
   if (info->last_count != thd_ndb->count)
   {
     info->last_count= thd_ndb->count;
@@ -370,14 +396,15 @@ void ha_ndbcluster::no_uncommitted_rows_reset(THD *thd)
   if (m_ha_not_exact_count)
     return;
   DBUG_ENTER("ha_ndbcluster::no_uncommitted_rows_reset");
-  ((Thd_ndb*)(thd->transaction.thd_ndb))->count++;
-  ((Thd_ndb*)(thd->transaction.thd_ndb))->error= 0;
+  Thd_ndb *thd_ndb= get_thd_ndb(thd);
+  thd_ndb->count++;
+  thd_ndb->error= 0;
   DBUG_VOID_RETURN;
 }
 
 /*
   Take care of the error that occured in NDB
-  
+
   RETURN
     0	No error
     #   The mapped error code
@@ -3136,9 +3163,9 @@ THR_LOCK_DATA **ha_ndbcluster::store_lock(THD *thd,
   As MySQL will execute an external lock for every new table it uses
   we can use this to start the transactions.
   If we are in auto_commit mode we just need to start a transaction
-  for the statement, this will be stored in transaction.stmt.
+  for the statement, this will be stored in thd_ndb.stmt.
   If not, we have to start a master transaction if there doesn't exist
-  one from before, this will be stored in transaction.all
+  one from before, this will be stored in thd_ndb.all
  
   When a table lock is held one transaction will be started which holds
   the table lock and for each statement a hupp transaction will be started  
@@ -3157,7 +3184,7 @@ int ha_ndbcluster::external_lock(THD *thd, int lock_type)
   if (check_ndb_connection())
     DBUG_RETURN(1);
  
-  Thd_ndb *thd_ndb= (Thd_ndb*)thd->transaction.thd_ndb;
+  Thd_ndb *thd_ndb= get_thd_ndb(thd);
   Ndb *ndb= thd_ndb->ndb;
 
   DBUG_PRINT("enter", ("transaction.thd_ndb->lock_count: %d", 
@@ -3173,18 +3200,19 @@ int ha_ndbcluster::external_lock(THD *thd, int lock_type)
       if (!(thd->options & (OPTION_NOT_AUTOCOMMIT | OPTION_BEGIN | OPTION_TABLE_LOCK))) 
       {
         // Autocommit transaction
-        DBUG_ASSERT(!thd->transaction.stmt.ndb_tid);
+        DBUG_ASSERT(!thd_ndb->stmt);
         DBUG_PRINT("trans",("Starting transaction stmt"));      
 
         trans= ndb->startTransaction();
         if (trans == NULL)
           ERR_RETURN(ndb->getNdbError());
 	no_uncommitted_rows_reset(thd);
-        thd->transaction.stmt.ndb_tid= trans;
+        thd_ndb->stmt= trans;
+        trans_register_ha(thd, FALSE, &ndbcluster_hton);
       } 
       else 
       { 
-        if (!thd->transaction.all.ndb_tid)
+        if (!thd_ndb->all)
 	{
           // Not autocommit transaction
           // A "master" transaction ha not been started yet
@@ -3194,6 +3222,8 @@ int ha_ndbcluster::external_lock(THD *thd, int lock_type)
           if (trans == NULL)
             ERR_RETURN(ndb->getNdbError());
 	  no_uncommitted_rows_reset(thd);
+          thd_ndb->all= trans; 
+          trans_register_ha(thd, TRUE, &ndbcluster_hton);
 
           /*
             If this is the start of a LOCK TABLE, a table look 
@@ -3207,7 +3237,6 @@ int ha_ndbcluster::external_lock(THD *thd, int lock_type)
             DBUG_PRINT("info", ("Locking the table..." ));
           }
 
-          thd->transaction.all.ndb_tid= trans; 
         }
       }
     }
@@ -3232,9 +3261,7 @@ int ha_ndbcluster::external_lock(THD *thd, int lock_type)
     else
       m_transaction_on= thd->variables.ndb_use_transactions;
 
-    m_active_trans= thd->transaction.all.ndb_tid ? 
-      (NdbTransaction*)thd->transaction.all.ndb_tid:
-      (NdbTransaction*)thd->transaction.stmt.ndb_tid;
+    m_active_trans= thd_ndb->all ? thd_ndb->all : thd_ndb->stmt;
     DBUG_ASSERT(m_active_trans);
     // Start of transaction
     m_retrieve_all_fields= FALSE;
@@ -3260,7 +3287,7 @@ int ha_ndbcluster::external_lock(THD *thd, int lock_type)
       DBUG_PRINT("trans", ("Last external_lock"));
       PRINT_OPTION_FLAGS(thd);
 
-      if (thd->transaction.stmt.ndb_tid)
+      if (thd_ndb->stmt)
       {
         /*
           Unlock is done without a transaction commit / rollback.
@@ -3269,7 +3296,7 @@ int ha_ndbcluster::external_lock(THD *thd, int lock_type)
         */
         DBUG_PRINT("trans",("ending non-updating transaction"));
         ndb->closeTransaction(m_active_trans);
-        thd->transaction.stmt.ndb_tid= 0;
+        thd_ndb->stmt= NULL;
       }
     }
     m_table= NULL;
@@ -3316,14 +3343,14 @@ int ha_ndbcluster::start_stmt(THD *thd)
   DBUG_ENTER("start_stmt");
   PRINT_OPTION_FLAGS(thd);
 
-  NdbTransaction *trans= (NdbTransaction*)thd->transaction.stmt.ndb_tid;
+  Thd_ndb *thd_ndb= get_thd_ndb(thd);
+  NdbTransaction *trans= thd_ndb->stmt;
   if (!trans){
-    Ndb *ndb= ((Thd_ndb*)thd->transaction.thd_ndb)->ndb;
+    Ndb *ndb= thd_ndb->ndb;
     DBUG_PRINT("trans",("Starting transaction stmt"));  
 
 #if 0    
-    NdbTransaction *tablock_trans= 
-      (NdbTransaction*)thd->transaction.all.ndb_tid;
+    NdbTransaction *tablock_trans= thd_ndb->all;
     DBUG_PRINT("info", ("tablock_trans: %x", (UintPtr)tablock_trans));
     DBUG_ASSERT(tablock_trans);
 //    trans= ndb->hupp(tablock_trans);
@@ -3332,7 +3359,8 @@ int ha_ndbcluster::start_stmt(THD *thd)
     if (trans == NULL)
       ERR_RETURN(ndb->getNdbError());
     no_uncommitted_rows_reset(thd);
-    thd->transaction.stmt.ndb_tid= trans;
+    thd_ndb->stmt= trans;
+    trans_register_ha(thd, FALSE, &ndbcluster_hton);
   }
   m_active_trans= trans;
 
@@ -3349,15 +3377,16 @@ int ha_ndbcluster::start_stmt(THD *thd)
   Commit a transaction started in NDB 
  */
 
-int ndbcluster_commit(THD *thd, void *ndb_transaction)
+int ndbcluster_commit(THD *thd, bool all)
 {
   int res= 0;
-  Ndb *ndb= ((Thd_ndb*)thd->transaction.thd_ndb)->ndb;
-  NdbTransaction *trans= (NdbTransaction*)ndb_transaction;
+  Thd_ndb *thd_ndb= get_thd_ndb(thd);
+  Ndb *ndb= thd_ndb->ndb;
+  NdbTransaction *trans= all ? thd_ndb->all : thd_ndb->stmt;
 
   DBUG_ENTER("ndbcluster_commit");
   DBUG_PRINT("transaction",("%s",
-                            trans == thd->transaction.stmt.ndb_tid ? 
+                            trans == thd_ndb->stmt ? 
                             "stmt" : "all"));
   DBUG_ASSERT(ndb && trans);
 
@@ -3371,6 +3400,7 @@ int ndbcluster_commit(THD *thd, void *ndb_transaction)
       ndbcluster_print_error(res, error_op);
   }
   ndb->closeTransaction(trans);
+  thd_ndb->all= thd_ndb->stmt= NULL;
   DBUG_RETURN(res);
 }
 
@@ -3379,15 +3409,16 @@ int ndbcluster_commit(THD *thd, void *ndb_transaction)
   Rollback a transaction started in NDB
  */
 
-int ndbcluster_rollback(THD *thd, void *ndb_transaction)
+int ndbcluster_rollback(THD *thd, bool all)
 {
   int res= 0;
-  Ndb *ndb= ((Thd_ndb*)thd->transaction.thd_ndb)->ndb;
-  NdbTransaction *trans= (NdbTransaction*)ndb_transaction;
+  Thd_ndb *thd_ndb= get_thd_ndb(thd);
+  Ndb *ndb= thd_ndb->ndb;
+  NdbTransaction *trans= all ? thd_ndb->all : thd_ndb->stmt;
 
   DBUG_ENTER("ndbcluster_rollback");
   DBUG_PRINT("transaction",("%s",
-                            trans == thd->transaction.stmt.ndb_tid ? 
+                            trans == thd_ndb->stmt ? 
                             "stmt" : "all"));
   DBUG_ASSERT(ndb && trans);
 
@@ -3401,7 +3432,8 @@ int ndbcluster_rollback(THD *thd, void *ndb_transaction)
       ndbcluster_print_error(res, error_op);
   }
   ndb->closeTransaction(trans);
-  DBUG_RETURN(0);
+  thd_ndb->all= thd_ndb->stmt= NULL;
+  DBUG_RETURN(res);
 }
 
 
@@ -4257,12 +4289,12 @@ void ha_ndbcluster::release_thd_ndb(Thd_ndb* thd_ndb)
 
 Ndb* check_ndb_in_thd(THD* thd)
 {
-  Thd_ndb *thd_ndb= (Thd_ndb*)thd->transaction.thd_ndb;  
+  Thd_ndb *thd_ndb= get_thd_ndb(thd);
   if (!thd_ndb)
   {
     if (!(thd_ndb= ha_ndbcluster::seize_thd_ndb()))
       return NULL;
-    thd->transaction.thd_ndb= thd_ndb;
+    set_thd_ndb(thd, thd_ndb);
   }
   return thd_ndb->ndb;
 }
@@ -4282,16 +4314,16 @@ int ha_ndbcluster::check_ndb_connection()
 }
 
 
-void ndbcluster_close_connection(THD *thd)
+int ndbcluster_close_connection(THD *thd)
 {
-  Thd_ndb *thd_ndb= (Thd_ndb*)thd->transaction.thd_ndb;
+  Thd_ndb *thd_ndb= get_thd_ndb(thd);
   DBUG_ENTER("ndbcluster_close_connection");
   if (thd_ndb)
   {
     ha_ndbcluster::release_thd_ndb(thd_ndb);
-    thd->transaction.thd_ndb= NULL;
+    set_thd_ndb(thd, NULL); // not strictly required but does not hurt either
   }
-  DBUG_VOID_RETURN;
+  DBUG_RETURN(0);
 }
 
 
@@ -4554,7 +4586,8 @@ static int connect_callback()
   return 0;
 }
 
-bool ndbcluster_init()
+handlerton *
+ndbcluster_init()
 {
   int res;
   DBUG_ENTER("ndbcluster_init");
@@ -4630,11 +4663,11 @@ bool ndbcluster_init()
   }
   
   ndbcluster_inited= 1;
-  DBUG_RETURN(FALSE);
+  DBUG_RETURN(&ndbcluster_hton);
 
  ndbcluster_init_error:
   ndbcluster_end();
-  DBUG_RETURN(TRUE);
+  DBUG_RETURN(NULL);
 }
 
 
diff --git a/sql/ha_ndbcluster.h b/sql/ha_ndbcluster.h
index b27908c20ea..10ee568df69 100644
--- a/sql/ha_ndbcluster.h
+++ b/sql/ha_ndbcluster.h
@@ -74,6 +74,8 @@ class Thd_ndb {
   Ndb *ndb;
   ulong count;
   uint lock_count;
+  NdbTransaction *all;
+  NdbTransaction *stmt;
   int error;
 };
 
@@ -286,14 +288,9 @@ private:
 
 extern struct show_var_st ndb_status_variables[];
 
-bool ndbcluster_init(void);
+handlerton *ndbcluster_init(void);
 bool ndbcluster_end(void);
 
-int ndbcluster_commit(THD *thd, void* ndb_transaction);
-int ndbcluster_rollback(THD *thd, void* ndb_transaction);
-
-void ndbcluster_close_connection(THD *thd);
-
 int ndbcluster_discover(THD* thd, const char* dbname, const char* name,
 			const void** frmblob, uint* frmlen);
 int ndbcluster_find_files(THD *thd,const char *db,const char *path,
diff --git a/sql/item_func.h b/sql/item_func.h
index 20f70fce575..70ad1a0375f 100644
--- a/sql/item_func.h
+++ b/sql/item_func.h
@@ -348,7 +348,7 @@ class Item_func_div :public Item_num_op
 {
 public:
   Item_func_div(Item *a,Item *b) :Item_num_op(a,b) {}
-  longlong int_op() { DBUG_ASSERT(0); }
+  longlong int_op() { DBUG_ASSERT(0); return 0; }
   double real_op();
   my_decimal *decimal_op(my_decimal *);
   const char *func_name() const { return "/"; }
diff --git a/sql/sql_yacc.yy b/sql/sql_yacc.yy
index 765f1ae0e73..5bc7f0e2bb9 100644
--- a/sql/sql_yacc.yy
+++ b/sql/sql_yacc.yy
@@ -1924,7 +1924,7 @@ sp_proc_stmt:
 	    sp_instr_set *i = new sp_instr_set(lex->sphead->instructions(),
 					       lex->spcont,
 	                                       offset, $2, MYSQL_TYPE_STRING);
-	    LEX_STRING dummy={"", 0};
+	    LEX_STRING dummy={(char*)"", 0};
 
 	    lex->spcont->push_pvar(&dummy, MYSQL_TYPE_STRING, sp_param_in);
 	    i->tables= lex->query_tables;