ndb - blob tables vs dict cache : patch 2 - bug#17761 (maybe)

storage/ndb/src/ndbapi/NdbDictionaryImpl.cpp:
  by-pass dict cache when creating blob event op
  does get event using table stored under main event op
  in preparation to remove blob tables from dict cache
storage/ndb/src/ndbapi/NdbDictionaryImpl.hpp:
  by-pass dict cache when creating blob event op
  does get event using table stored under main event op
  in preparation to remove blob tables from dict cache
storage/ndb/src/ndbapi/NdbEventOperationImpl.cpp:
  by-pass dict cache when creating blob event op
  does get event using table stored under main event op
  in preparation to remove blob tables from dict cache
storage/ndb/src/ndbapi/NdbEventOperationImpl.hpp:
  by-pass dict cache when creating blob event op
  does get event using table stored under main event op
  in preparation to remove blob tables from dict cache

storage/ndb/src/ndbapi/NdbDictionaryImpl.cpp

@@ -3480,7 +3480,7 @@ NdbDictInterface::stopSubscribeEvent(class Ndb & ndb,
 }
 
 NdbEventImpl * 
-NdbDictionaryImpl::getEvent(const char * eventName)
+NdbDictionaryImpl::getEvent(const char * eventName, NdbTableImpl* tab)
 {
   DBUG_ENTER("NdbDictionaryImpl::getEvent");
   DBUG_PRINT("enter",("eventName= %s", eventName));
@@ -3499,6 +3499,7 @@ NdbDictionaryImpl::getEvent(const char * eventName)
     DBUG_RETURN(NULL);
   }
 
+  if (tab == NULL) {
     // We only have the table name with internal name
     DBUG_PRINT("info",("table %s", ev->getTableName()));
     Ndb_local_table_info *info;
@@ -3520,7 +3521,10 @@ NdbDictionaryImpl::getEvent(const char * eventName)
         DBUG_RETURN(NULL);
       }
     }
-  ev->setTable(info->m_table_impl);
+    tab = info->m_table_impl;
+  }
+
+  ev->setTable(tab);
   ev->setTable(m_ndb.externalizeTableName(ev->getTableName()));  
   // get the columns from the attrListBitmask
   NdbTableImpl &table = *ev->m_tableImpl;
@@ -3568,6 +3572,26 @@ NdbDictionaryImpl::getEvent(const char * eventName)
   DBUG_RETURN(ev);
 }
 
+// ev is main event and has been retrieved previously
+NdbEventImpl *
+NdbDictionaryImpl::getBlobEvent(const NdbEventImpl& ev, uint col_no)
+{
+  DBUG_ENTER("NdbDictionaryImpl::getBlobEvent");
+  DBUG_PRINT("enter", ("ev=%s col=%u", ev.m_name.c_str(), col_no));
+
+  NdbTableImpl* tab = ev.m_tableImpl;
+  assert(tab != NULL && col_no < tab->m_columns.size());
+  NdbColumnImpl* col = tab->m_columns[col_no];
+  assert(col != NULL && col->getBlobType() && col->getPartSize() != 0);
+  NdbTableImpl* blob_tab = col->m_blobTable;
+  assert(blob_tab != NULL);
+  char bename[MAX_TAB_NAME_SIZE];
+  NdbBlob::getBlobEventName(bename, &ev, col);
+
+  NdbEventImpl* blob_ev = getEvent(bename, blob_tab);
+  DBUG_RETURN(blob_ev);
+}
+
 void
 NdbDictInterface::execCREATE_EVNT_CONF(NdbApiSignal * signal,
 				       LinearSectionPtr ptr[3])

storage/ndb/src/ndbapi/NdbDictionaryImpl.hpp

@@ -577,7 +577,8 @@ public:
     const BaseString& internalTableName, bool do_add_blob_tables);
   NdbIndexImpl * getIndex(const char * indexName,
 			  const char * tableName);
-  NdbEventImpl * getEvent(const char * eventName);
+  NdbEventImpl * getEvent(const char * eventName, NdbTableImpl* = NULL);
+  NdbEventImpl * getBlobEvent(const NdbEventImpl& ev, uint col_no);
   NdbEventImpl * getEventImpl(const char * internalName);
 
   int createDatafile(const NdbDatafileImpl &, bool force = false);

storage/ndb/src/ndbapi/NdbEventOperationImpl.cpp

@@ -88,18 +88,55 @@ print_std(const SubTableData * sdata, LinearSectionPtr ptr[3])
 // todo handle several ndb objects
 // todo free allocated data when closing NdbEventBuffer
 
-NdbEventOperationImpl::NdbEventOperationImpl(NdbEventOperation &N,
+NdbEventOperationImpl::NdbEventOperationImpl(NdbEventOperation &f,
 					     Ndb *theNdb, 
-					     const char* eventName) 
-  : NdbEventOperation(*this), m_facade(&N), m_magic_number(0),
-    m_ndb(theNdb), m_state(EO_ERROR), mi_type(0), m_oid(~(Uint32)0),
-    m_change_mask(0),
-#ifdef VM_TRACE
-    m_data_done_count(0), m_data_count(0),
-#endif
-    m_next(0), m_prev(0)
+					     const char* eventName) :
+  NdbEventOperation(*this),
+  m_facade(&f),
+  m_ndb(theNdb),
+  m_state(EO_ERROR)
 {
   DBUG_ENTER("NdbEventOperationImpl::NdbEventOperationImpl");
+
+  assert(m_ndb != NULL);
+  NdbDictionary::Dictionary *myDict = m_ndb->getDictionary();
+  assert(myDict != NULL);
+
+  const NdbDictionary::Event *myEvnt = myDict->getEvent(eventName);
+  if (!myEvnt) { m_error.code= myDict->getNdbError().code; DBUG_VOID_RETURN; }
+
+  init(myEvnt->m_impl);
+  DBUG_VOID_RETURN;
+}
+
+NdbEventOperationImpl::NdbEventOperationImpl(Ndb *theNdb,
+                                             NdbEventImpl& evnt) :
+  NdbEventOperation(*this),
+  m_facade(this),
+  m_ndb(theNdb),
+  m_state(EO_ERROR)
+{
+  DBUG_ENTER("NdbEventOperationImpl::NdbEventOperationImpl [evnt]");
+  init(evnt);
+  DBUG_VOID_RETURN;
+}
+
+void
+NdbEventOperationImpl::init(NdbEventImpl& evnt)
+{
+  DBUG_ENTER("NdbEventOperationImpl::init");
+
+  m_magic_number = 0;
+  mi_type = 0;
+  m_oid = ~(Uint32)0;
+  m_change_mask = 0;
+#ifdef VM_TRACE
+  m_data_done_count = 0;
+  m_data_count = 0;
+#endif
+  m_next = 0;
+  m_prev = 0;
+
   m_eventId = 0;
   theFirstPkAttrs[0] = NULL;
   theCurrentPkAttrs[0] = NULL;
@@ -123,15 +160,7 @@ NdbEventOperationImpl::NdbEventOperationImpl(NdbEventOperation &N,
   // we should lookup id in Dictionary, TODO
   // also make sure we only have one listener on each event
 
-  if (!m_ndb) abort();
-
-  NdbDictionary::Dictionary *myDict = m_ndb->getDictionary();
-  if (!myDict) { m_error.code= m_ndb->getNdbError().code; DBUG_VOID_RETURN; }
-
-  const NdbDictionary::Event *myEvnt = myDict->getEvent(eventName);
-  if (!myEvnt) { m_error.code= myDict->getNdbError().code; DBUG_VOID_RETURN; }
-
-  m_eventImpl = &myEvnt->m_impl;
+  m_eventImpl = &evnt;
 
   m_eventId = m_eventImpl->m_eventId;
 
@@ -347,18 +376,27 @@ NdbEventOperationImpl::getBlobHandle(const NdbColumnImpl *tAttrInfo, int n)
 
     // create blob event op if not found
     if (tBlobOp == NULL) {
-      // to hide blob op it is linked under main op, not under m_ndb
-      NdbEventOperation* tmp =
-        m_ndb->theEventBuffer->createEventOperation(bename, m_error);
-      if (tmp == NULL)
+      // get blob event
+      NdbDictionaryImpl& dict =
+        NdbDictionaryImpl::getImpl(*m_ndb->getDictionary());
+      NdbEventImpl* blobEvnt =
+        dict.getBlobEvent(*this->m_eventImpl, tAttrInfo->m_column_no);
+      if (blobEvnt == NULL) {
+        m_error.code = dict.m_error.code;
+        DBUG_RETURN(NULL);
+      }
+
+      // create blob event operation
+      tBlobOp =
+        m_ndb->theEventBuffer->createEventOperation(*blobEvnt, m_error);
+      if (tBlobOp == NULL)
         DBUG_RETURN(NULL);
-      tBlobOp = &tmp->m_impl;
 
       // pointer to main table op
       tBlobOp->theMainOp = this;
       tBlobOp->m_mergeEvents = m_mergeEvents;
 
-      // add to list end
+      // to hide blob op it is linked under main op, not under m_ndb
       if (tLastBlopOp == NULL)
         theBlobOpList = tBlobOp;
       else
@@ -2131,6 +2169,25 @@ NdbEventBuffer::createEventOperation(const char* eventName,
   DBUG_RETURN(tOp);
 }
 
+NdbEventOperationImpl*
+NdbEventBuffer::createEventOperation(NdbEventImpl& evnt,
+				     NdbError &theError)
+{
+  DBUG_ENTER("NdbEventBuffer::createEventOperation [evnt]");
+  NdbEventOperationImpl* tOp= new NdbEventOperationImpl(m_ndb, evnt);
+  if (tOp == 0)
+  {
+    theError.code= 4000;
+    DBUG_RETURN(NULL);
+  }
+  if (tOp->getState() != NdbEventOperation::EO_CREATED) {
+    theError.code= tOp->getNdbError().code;
+    delete tOp;
+    DBUG_RETURN(NULL);
+  }
+  DBUG_RETURN(tOp);
+}
+
 void
 NdbEventBuffer::dropEventOperation(NdbEventOperation* tOp)
 {

storage/ndb/src/ndbapi/NdbEventOperationImpl.hpp

@@ -202,9 +202,12 @@ struct Gci_container
 
 class NdbEventOperationImpl : public NdbEventOperation {
 public:
-  NdbEventOperationImpl(NdbEventOperation &N,
+  NdbEventOperationImpl(NdbEventOperation &f,
 			Ndb *theNdb, 
 			const char* eventName);
+  NdbEventOperationImpl(Ndb *theNdb, 
+			NdbEventImpl& evnt);
+  void init(NdbEventImpl& evnt);
   NdbEventOperationImpl(NdbEventOperationImpl&); //unimplemented
   NdbEventOperationImpl& operator=(const NdbEventOperationImpl&); //unimplemented
   ~NdbEventOperationImpl();
@@ -293,6 +296,8 @@ public:
   Vector<Gci_container> m_active_gci;
   NdbEventOperation *createEventOperation(const char* eventName,
 					  NdbError &);
+  NdbEventOperationImpl *createEventOperation(NdbEventImpl& evnt,
+					  NdbError &);
   void dropEventOperation(NdbEventOperation *);
   static NdbEventOperationImpl* getEventOperationImpl(NdbEventOperation* tOp);