bzr commit into mysql-5.1 branch (DonKehndon.kehn:2749) Bug#33626

[DonKehndon.kehn]

#At file:///u02/devl/bzr-tree/6.3-patch3/ based on
revid:don.kehn@stripped

 2749 Don Kehn	2008-11-14
      BUG#33626 Check that the ndb object in thd_ndb has the same connect count
modified:
  sql/ha_ndbcluster.cc
  sql/ha_ndbcluster.h
  sql/ha_ndbcluster_binlog.h

=== modified file 'sql/ha_ndbcluster.cc'
--- a/sql/ha_ndbcluster.cc	2008-11-08 21:22:57 +0000
+++ b/sql/ha_ndbcluster.cc	2008-11-14 19:32:11 +0000
@@ -611,6 +611,7 @@ uchar *thd_ndb_share_get_key(THD_NDB_SHA
 Thd_ndb::Thd_ndb()
 {
   connection= ndb_get_cluster_connection();
+  m_connect_count= connection->get_connect_count();
   ndb= new Ndb(connection, "");
   lock_count= 0;
   start_stmt_count= 0;
@@ -8042,6 +8043,44 @@ void ha_ndbcluster::release_thd_ndb(Thd_
 }
 
 
+bool Thd_ndb::recycle_ndb(THD* thd)
+{
+  DBUG_ENTER("recycle_ndb");
+  DBUG_PRINT("enter", ("ndb: 0x%lx", (long)ndb));
+
+  delete ndb;
+  if ((ndb= new Ndb(connection, "")) == NULL)
+  {
+    DBUG_PRINT("error",("failed to allocate Ndb object"));
+    DBUG_RETURN(false);
+  }
+
+  if (ndb->init(max_transactions) != 0)
+  {
+    delete ndb;
+    ndb= NULL;
+    DBUG_PRINT("error", ("Ndb::init failed, %d  message: %s",
+                         ndb->getNdbError().code,
+                         ndb->getNdbError().message));
+    DBUG_RETURN(false);
+  }
+
+  trans= NULL;
+
+  DBUG_RETURN(true);
+}
+
+
+bool
+Thd_ndb::valid_ndb(void)
+{
+  if (unlikely(m_connect_count != connection->get_connect_count()))
+    return false;
+  return true;
+}
+
+
+
 /**
   If this thread already has a Thd_ndb object allocated
   in current THD, reuse it. Otherwise
@@ -8049,7 +8088,7 @@ void ha_ndbcluster::release_thd_ndb(Thd_
  
 */
 
-Ndb* check_ndb_in_thd(THD* thd)
+Ndb* check_ndb_in_thd(THD* thd, bool validate_ndb)
 {
   Thd_ndb *thd_ndb= get_thd_ndb(thd);
   if (!thd_ndb)
@@ -8058,6 +8097,11 @@ Ndb* check_ndb_in_thd(THD* thd)
       return NULL;
     set_thd_ndb(thd, thd_ndb);
   }
+  else if (validate_ndb && !thd_ndb->valid_ndb())
+  {
+    if (!thd_ndb->recycle_ndb(thd))
+      return NULL;
+  }
   return thd_ndb->ndb;
 }
 
@@ -8068,7 +8112,7 @@ int ha_ndbcluster::check_ndb_connection(
   Ndb *ndb;
   DBUG_ENTER("check_ndb_connection");
   
-  if (!(ndb= check_ndb_in_thd(thd)))
+  if (!(ndb= check_ndb_in_thd(thd, true)))
     DBUG_RETURN(HA_ERR_NO_CONNECTION);
   if (ndb->setDatabaseName(m_dbname))
   {

=== modified file 'sql/ha_ndbcluster.h'
--- a/sql/ha_ndbcluster.h	2008-10-15 14:27:13 +0000
+++ b/sql/ha_ndbcluster.h	2008-11-14 19:32:11 +0000
@@ -331,6 +331,10 @@ class Thd_ndb 
   NdbTransaction *global_schema_lock_trans;
   uint global_schema_lock_count;
   uint global_schema_lock_error;
+
+  unsigned m_connect_count;
+  bool valid_ndb(void);
+  bool recycle_ndb(THD* thd);
 };
 
 int ndbcluster_commit(handlerton *hton, THD *thd, bool all);

=== modified file 'sql/ha_ndbcluster_binlog.h'
--- a/sql/ha_ndbcluster_binlog.h	2008-10-29 14:12:28 +0000
+++ b/sql/ha_ndbcluster_binlog.h	2008-11-14 19:32:11 +0000
@@ -293,7 +293,7 @@ void
 set_thd_ndb(THD *thd, Thd_ndb *thd_ndb)
 { thd_set_ha_data(thd, ndbcluster_hton, thd_ndb); }
 
-Ndb* check_ndb_in_thd(THD* thd);
+Ndb* check_ndb_in_thd(THD* thd, bool validate_ndb= false);
 
 int ndbcluster_has_global_schema_lock(Thd_ndb *thd_ndb);
 int ndbcluster_no_global_schema_lock_abort(THD *thd, const char *msg);