Merge tulin@bk-internal.mysql.com:/home/bk/mysql-5.1-new

into  poseidon.ndb.mysql.com:/home/tomas/mysql-5.1-new

sql/ha_ndbcluster.cc

@@ -101,8 +101,6 @@ static uint ndbcluster_alter_table_flags(uint flags)
 #define NDB_FAILED_AUTO_INCREMENT ~(Uint64)0
 #define NDB_AUTO_INCREMENT_RETRIES 10
 
-#define NDB_INVALID_SCHEMA_OBJECT 241
-
 #define ERR_PRINT(err) \
   DBUG_PRINT("error", ("%d  message: %s", err.code, err.message))
 

sql/ha_ndbcluster_binlog.cc

@@ -2226,10 +2226,20 @@ ndbcluster_create_event(Ndb *ndb, const NDBTAB *ndbtab,
       DBUG_RETURN(-1);
     }
 
+    /*
+      try retrieving the event, if table version/id matches, we will get
+      a valid event.  Otherwise we have a trailing event from before
+    */
+    if (dict->getEvent(event_name))
+    {
+      DBUG_RETURN(0);
+    }
+
     /*
       trailing event from before; an error, but try to correct it
     */
-    if (dict->dropEvent(my_event.getName()))
+    if (dict->getNdbError().code == NDB_INVALID_SCHEMA_OBJECT &&
+        dict->dropEvent(my_event.getName()))
     {
       if (push_warning)
         push_warning_printf(current_thd, MYSQL_ERROR::WARN_LEVEL_ERROR,

sql/ha_ndbcluster_binlog.h

@@ -29,6 +29,8 @@ extern ulong ndb_extra_logging;
 
 #define INJECTOR_EVENT_LEN 200
 
+#define NDB_INVALID_SCHEMA_OBJECT 241
+
 /*
   The numbers below must not change as they
   are passed between mysql servers, and if changed

storage/ndb/src/ndbapi/NdbDictionaryImpl.cpp

@@ -3435,6 +3435,11 @@ NdbDictInterface::createEvent(class Ndb & ndb,
   
   //  NdbEventImpl *evntImpl = (NdbEventImpl *)evntConf->getUserData();
 
+  evnt.m_eventId = evntConf->getEventId();
+  evnt.m_eventKey = evntConf->getEventKey();
+  evnt.m_table_id = evntConf->getTableId();
+  evnt.m_table_version = evntConf->getTableVersion();
+
   if (getFlag) {
     evnt.m_attrListBitmask = evntConf->getAttrListBitmask();
     evnt.mi_type           = evntConf->getEventType();
@@ -3449,9 +3454,6 @@ NdbDictInterface::createEvent(class Ndb & ndb,
     }
   }
 
-  evnt.m_eventId         = evntConf->getEventId();
-  evnt.m_eventKey        = evntConf->getEventKey();
-
   DBUG_RETURN(0);
 }
 
@@ -3560,7 +3562,10 @@ NdbDictionaryImpl::getEvent(const char * eventName, NdbTableImpl* tab)
       delete ev;
       DBUG_RETURN(NULL);
     }
-    if (info->m_table_impl->m_status != NdbDictionary::Object::Retrieved)
+    if ((info->m_table_impl->m_status != NdbDictionary::Object::Retrieved) ||
+        (info->m_table_impl->m_id != ev->m_table_id) ||
+        (table_version_major(info->m_table_impl->m_version) !=
+         table_version_major(ev->m_table_version)))
     {
       removeCachedObject(*info->m_table_impl);
       info= get_local_table_info(ev->getTableName());
@@ -3584,6 +3589,14 @@ NdbDictionaryImpl::getEvent(const char * eventName, NdbTableImpl* tab)
   DBUG_PRINT("info",("Table: id: %d version: %d", 
                      table.m_id, table.m_version));
 
+  if (table.m_id != ev->m_table_id ||
+      table_version_major(table.m_version) !=
+      table_version_major(ev->m_table_version))
+  {
+    m_error.code = 241;
+    delete ev;
+    DBUG_RETURN(NULL);
+  }
 #ifndef DBUG_OFF
   char buf[128] = {0};
   mask.getText(buf);

storage/ndb/src/ndbapi/NdbDictionaryImpl.hpp

@@ -303,6 +303,8 @@ public:
   Uint32 m_eventId;
   Uint32 m_eventKey;
   AttributeMask m_attrListBitmask;
+  Uint32 m_table_id;
+  Uint32 m_table_version;
   BaseString m_name;
   Uint32 mi_type;
   NdbDictionary::Event::EventDurability m_dur;