ndb - wl#2972 rbr blobs ndb api support

storage/ndb/include/ndbapi/NdbBlob.hpp

@@ -28,6 +28,7 @@ class NdbOperation;
 class NdbRecAttr;
 class NdbTableImpl;
 class NdbColumnImpl;
+class NdbEventOperationImpl;
 
 /**
  * @class NdbBlob
@@ -71,6 +72,10 @@ class NdbColumnImpl;
  * writes.  It avoids execute penalty if nothing is pending.  It is not
  * needed after execute (obviously) or after next scan result.
  *
+ * NdbBlob also supports reading post or pre blob data from events.  The
+ * handle can be read after next event on main table has been retrieved.
+ * The data is available immediately.  See NdbEventOperation.
+ *
  * NdbBlob methods return -1 on error and 0 on success, and use output
  * parameters when necessary.
  *
@@ -145,6 +150,12 @@ public:
    * then the callback is invoked.
    */
   int setActiveHook(ActiveHook* activeHook, void* arg);
+  /**
+   * Check if blob value is defined (NULL or not).  Used as first call
+   * on event based blob.  The argument is set to -1 for not defined.
+   * Unlike getNull() this does not cause error on the handle.
+   */
+  int getDefined(int& isNull);
   /**
    * Check if blob is null.
    */
@@ -191,6 +202,11 @@ public:
    * Get blob parts table name.  Useful only to test programs.
    */
   static int getBlobTableName(char* btname, Ndb* anNdb, const char* tableName, const char* columnName);
+  /**
+   * Get blob event name.  The blob event is created if the main event
+   * monitors the blob column.  The name includes main event name.
+   */
+  static int getBlobEventName(char* bename, Ndb* anNdb, const char* eventName, const char* columnName);
   /**
    * Return error object.  The error may be blob specific (below) or may
    * be copied from a failed implicit operation.
@@ -217,17 +233,29 @@ private:
   friend class NdbScanOperation;
   friend class NdbDictionaryImpl;
   friend class NdbResultSet; // atNextResult
+  friend class NdbEventBuffer;
+  friend class NdbEventOperationImpl;
 #endif
   // state
   State theState;
   void setState(State newState);
+  // quick and dirty support for events (consider subclassing)
+  int theEventBlobVersion; // -1=normal blob 0=post event 1=pre event
   // define blob table
   static void getBlobTableName(char* btname, const NdbTableImpl* t, const NdbColumnImpl* c);
   static void getBlobTable(NdbTableImpl& bt, const NdbTableImpl* t, const NdbColumnImpl* c);
+  static void getBlobEventName(char* bename, const NdbEventImpl* e, const NdbColumnImpl* c);
+  static void getBlobEvent(NdbEventImpl& be, const NdbEventImpl* e, const NdbColumnImpl* c);
   // ndb api stuff
   Ndb* theNdb;
   NdbTransaction* theNdbCon;
   NdbOperation* theNdbOp;
+  NdbEventOperationImpl* theEventOp;
+  NdbEventOperationImpl* theBlobEventOp;
+  NdbRecAttr* theBlobEventPkRecAttr;
+  NdbRecAttr* theBlobEventDistRecAttr;
+  NdbRecAttr* theBlobEventPartRecAttr;
+  NdbRecAttr* theBlobEventDataRecAttr;
   const NdbTableImpl* theTable;
   const NdbTableImpl* theAccessTable;
   const NdbTableImpl* theBlobTable;
@@ -263,6 +291,8 @@ private:
   Buf theHeadInlineBuf;
   Buf theHeadInlineCopyBuf;     // for writeTuple
   Buf thePartBuf;
+  Buf theBlobEventDataBuf;
+  Uint32 thePartNumber;         // for event
   Head* theHead;
   char* theInlineData;
   NdbRecAttr* theHeadInlineRecAttr;
@@ -306,6 +336,8 @@ private:
   int readDataPrivate(char* buf, Uint32& bytes);
   int writeDataPrivate(const char* buf, Uint32 bytes);
   int readParts(char* buf, Uint32 part, Uint32 count);
+  int readTableParts(char* buf, Uint32 part, Uint32 count);
+  int readEventParts(char* buf, Uint32 part, Uint32 count);
   int insertParts(const char* buf, Uint32 part, Uint32 count);
   int updateParts(const char* buf, Uint32 part, Uint32 count);
   int deleteParts(Uint32 part, Uint32 count);
@@ -317,19 +349,23 @@ private:
   int invokeActiveHook();
   // blob handle maintenance
   int atPrepare(NdbTransaction* aCon, NdbOperation* anOp, const NdbColumnImpl* aColumn);
+  int atPrepare(NdbEventOperationImpl* anOp, NdbEventOperationImpl* aBlobOp, const NdbColumnImpl* aColumn, int version);
+  int prepareColumn();
   int preExecute(NdbTransaction::ExecType anExecType, bool& batch);
   int postExecute(NdbTransaction::ExecType anExecType);
   int preCommit();
   int atNextResult();
+  int atNextEvent();
   // errors
   void setErrorCode(int anErrorCode, bool invalidFlag = true);
   void setErrorCode(NdbOperation* anOp, bool invalidFlag = true);
   void setErrorCode(NdbTransaction* aCon, bool invalidFlag = true);
+  void setErrorCode(NdbEventOperationImpl* anOp, bool invalidFlag = true);
 #ifdef VM_TRACE
   int getOperationType() const;
   friend class NdbOut& operator<<(NdbOut&, const NdbBlob&);
 #endif
-
+  // list stuff
   void next(NdbBlob* obj) { theNext= obj;}
   NdbBlob* next() { return theNext;}
   friend struct Ndb_free_list_t<NdbBlob>;

storage/ndb/include/ndbapi/NdbDictionary.hpp

@@ -1124,7 +1124,7 @@ public:
       _TE_NODE_FAILURE=10,
       _TE_SUBSCRIBE=11,
       _TE_UNSUBSCRIBE=12,
-      _TE_NUL=13 // internal (INS o DEL within same GCI)
+      _TE_NUL=13 // internal (e.g. INS o DEL within same GCI)
     };
 #endif
     /**
@@ -1261,6 +1261,24 @@ public:
      */
     int getNoOfEventColumns() const;
 
+    /**
+     * The merge events flag is false by default.  Setting it true
+     * implies that events are merged in following ways:
+     *
+     * - for given NdbEventOperation associated with this event,
+     *   events on same PK within same GCI are merged into single event
+     *
+     * - a blob table event is created for each blob attribute
+     *   and blob events are handled as part of main table events
+     *
+     * - blob post/pre data from the blob part events can be read
+     *   via NdbBlob methods as a single value
+     *
+     * NOTE: Currently this flag is not inherited by NdbEventOperation
+     * and must be set on NdbEventOperation explicitly.
+     */
+    void mergeEvents(bool flag);
+
     /**
      * Get object status
      */

storage/ndb/include/ndbapi/NdbEventOperation.hpp

@@ -150,6 +150,14 @@ public:
    */
   NdbRecAttr *getPreValue(const char *anAttrName, char *aValue = 0);
 
+  /**
+   * These methods replace getValue/getPreValue for blobs.  Each
+   * method creates a blob handle NdbBlob.  The handle supports only
+   * read operations.  See NdbBlob.
+   */
+  NdbBlob* getBlobHandle(const char *anAttrName);
+  NdbBlob* getPreBlobHandle(const char *anAttrName);
+
   int isOverrun() const;
 
   /**

storage/ndb/ndbapi-examples/ndbapi_event/Makefile

@@ -4,7 +4,7 @@ OBJS = ndbapi_event.o
 CXX = g++ -g
 CFLAGS = -c -Wall -fno-rtti -fno-exceptions
 CXXFLAGS = 
-DEBUG = 
+DEBUG =# -DVM_TRACE
 LFLAGS = -Wall
 TOP_SRCDIR = ../../../..
 INCLUDE_DIR = $(TOP_SRCDIR)/storage/ndb/include
@@ -16,8 +16,8 @@ SYS_LIB =
 $(TARGET): $(OBJS)
 	$(CXX) $(CXXFLAGS) $(LFLAGS) $(LIB_DIR) $(OBJS) -lndbclient -lmysqlclient_r -lmysys -lmystrings -lz $(SYS_LIB) -o $(TARGET)
 
-$(TARGET).o: $(SRCS)
-	$(CXX) $(CFLAGS) -I$(INCLUDE_DIR) -I$(INCLUDE_DIR)/ndbapi -I$(TOP_SRCDIR)/include $(SRCS)
+$(TARGET).o: $(SRCS) Makefile
+	$(CXX) $(CFLAGS) $(DEBUG) -I$(INCLUDE_DIR) -I$(INCLUDE_DIR)/ndbapi -I$(TOP_SRCDIR)/include $(SRCS)
 
 clean:
 	rm -f *.o $(TARGET)

storage/ndb/ndbapi-examples/ndbapi_event/ndbapi_event.cpp

@@ -54,26 +54,32 @@
 #include <stdio.h>
 #include <iostream>
 #include <unistd.h>
+#ifdef VM_TRACE
+#include <my_global.h>
+#endif
+#ifndef assert
+#include <assert.h>
+#endif
 
 
 /**
- *
- * Assume that there is a table t0 which is being updated by 
+ * Assume that there is a table which is being updated by 
  * another process (e.g. flexBench -l 0 -stdtables).
- * We want to monitor what happens with columns c0,c1,c2,c3.
+ * We want to monitor what happens with column values.
  *
- * or together with the mysql client;
+ * Or using the mysql client:
  *
  * shell> mysql -u root
  * mysql> create database TEST_DB;
  * mysql> use TEST_DB;
- * mysql> create table t0 (c0 int, c1 int, c2 char(4), c3 char(4),
+ * mysql> create table t0
+ *        (c0 int, c1 int, c2 char(4), c3 char(4), c4 text,
  *        primary key(c0, c2)) engine ndb charset latin1;
  *
  * In another window start ndbapi_event, wait until properly started
- *
-   insert into t0 values (1, 2, 'a', 'b');
-   insert into t0 values (3, 4, 'c', 'd');
+ 
+   insert into t0 values (1, 2, 'a', 'b', null);
+   insert into t0 values (3, 4, 'c', 'd', null);
    update t0 set c3 = 'e' where c0 = 1 and c2 = 'a'; -- use pk
    update t0 set c3 = 'f'; -- use scan
    update t0 set c3 = 'F'; -- use scan update to 'same'
@@ -81,7 +87,18 @@
    update t0 set c2 = 'G' where c0 = 1; -- update pk part to 'same'
    update t0 set c0 = 5, c2 = 'H' where c0 = 3; -- update full PK
    delete from t0;
- *
+
+   insert ...; update ...; -- see events w/ same pk merged (if -m option)
+   delete ...; insert ...; -- there are 5 combinations ID IU DI UD UU
+   update ...; update ...;
+
+   -- text requires -m flag
+   set @a = repeat('a',256); -- inline size
+   set @b = repeat('b',2000); -- part size
+   set @c = repeat('c',2000*30); -- 30 parts
+
+   -- update the text field using combinations of @a, @b, @c ...
+ 
  * you should see the data popping up in the example window
  *
  */
@@ -95,12 +112,18 @@ int myCreateEvent(Ndb* myNdb,
 		  const char *eventName,
 		  const char *eventTableName,
 		  const char **eventColumnName,
-		  const int noEventColumnName);
+		  const int noEventColumnName,
+                  bool merge_events);
 
 int main(int argc, char** argv)
 {
   ndb_init();
-  bool merge_events = argc > 1 && strcmp(argv[1], "-m") == 0;
+  bool merge_events = argc > 1 && strchr(argv[1], 'm') != 0;
+#ifdef VM_TRACE
+  bool dbug = argc > 1 && strchr(argv[1], 'd') != 0;
+  if (dbug) DBUG_PUSH("d:t:");
+  if (dbug) putenv("API_SIGNAL_LOG=-");
+#endif
 
   Ndb_cluster_connection *cluster_connection=
     new Ndb_cluster_connection(); // Object representing the cluster
@@ -134,12 +157,13 @@ int main(int argc, char** argv)
 
   const char *eventName= "CHNG_IN_t0";
   const char *eventTableName= "t0";
-  const int noEventColumnName= 4;
+  const int noEventColumnName= 5;
   const char *eventColumnName[noEventColumnName]=
     {"c0",
      "c1",
      "c2",
-     "c3"
+     "c3",
+     "c4"
     };
   
   // Create events
@@ -147,9 +171,14 @@ int main(int argc, char** argv)
 		eventName,
 		eventTableName,
 		eventColumnName,
-		noEventColumnName);
+		noEventColumnName,
+                merge_events);
 
-  int j= 0;
+  // Normal values and blobs are unfortunately handled differently..
+  typedef union { NdbRecAttr* ra; NdbBlob* bh; } RA_BH;
+
+  int i, j, k, l;
+  j = 0;
   while (j < 99) {
 
     // Start "transaction" for handling events
@@ -160,12 +189,17 @@ int main(int argc, char** argv)
     op->mergeEvents(merge_events);
 
     printf("get values\n");
-    NdbRecAttr* recAttr[noEventColumnName];
-    NdbRecAttr* recAttrPre[noEventColumnName];
+    RA_BH recAttr[noEventColumnName];
+    RA_BH recAttrPre[noEventColumnName];
     // primary keys should always be a part of the result
-    for (int i = 0; i < noEventColumnName; i++) {
-      recAttr[i]    = op->getValue(eventColumnName[i]);
-      recAttrPre[i] = op->getPreValue(eventColumnName[i]);
+    for (i = 0; i < noEventColumnName; i++) {
+      if (i < 4) {
+        recAttr[i].ra    = op->getValue(eventColumnName[i]);
+        recAttrPre[i].ra = op->getPreValue(eventColumnName[i]);
+      } else if (merge_events) {
+        recAttr[i].bh    = op->getBlobHandle(eventColumnName[i]);
+        recAttrPre[i].bh = op->getPreBlobHandle(eventColumnName[i]);
+      }
     }
 
     // set up the callbacks
@@ -174,13 +208,16 @@ int main(int argc, char** argv)
     if (op->execute())
       APIERROR(op->getNdbError());
 
-    int i= 0;
-    while(i < 40) {
+    NdbEventOperation* the_op = op;
+
+    i= 0;
+    while (i < 40) {
       // printf("now waiting for event...\n");
-      int r= myNdb->pollEvents(1000); // wait for event or 1000 ms
+      int r = myNdb->pollEvents(1000); // wait for event or 1000 ms
       if (r > 0) {
 	// printf("got data! %d\n", r);
 	while ((op= myNdb->nextEvent())) {
+          assert(the_op == op);
 	  i++;
 	  switch (op->getEventType()) {
 	  case NdbDictionary::Event::TE_INSERT:
@@ -195,40 +232,66 @@ int main(int argc, char** argv)
 	  default:
 	    abort(); // should not happen
 	  }
-          printf(" gci=%d\n", op->getGCI());
-          printf("post:  ");
-	  for (int i = 0; i < noEventColumnName; i++) {
-	    if (recAttr[i]->isNULL() >= 0) { // we have a value
-	      if (recAttr[i]->isNULL() == 0) { // we have a non-null value
-                if (i < 2)
-                  printf("%-5u", recAttr[i]->u_32_value());
+          printf(" gci=%d\n", (int)op->getGCI());
+          for (k = 0; k <= 1; k++) {
+            printf(k == 0 ? "post: " : "pre : ");
+            for (l = 0; l < noEventColumnName; l++) {
+              if (l < 4) {
+                NdbRecAttr* ra = k == 0 ? recAttr[l].ra : recAttrPre[l].ra;
+                if (ra->isNULL() >= 0) { // we have a value
+                  if (ra->isNULL() == 0) { // we have a non-null value
+                    if (l < 2)
+                      printf("%-5u", ra->u_32_value());
                     else
-                  printf("%-5.4s", recAttr[i]->aRef());
-              } else                           // we have a null value
+                      printf("%-5.4s", ra->aRef());
+                  } else
                     printf("%-5s", "NULL");
                 } else
-              printf("%-5s", "-");
+                  printf("%-5s", "-"); // no value
+              } else if (merge_events) {
+                int isNull;
+                NdbBlob* bh = k == 0 ? recAttr[l].bh : recAttrPre[l].bh;
+                bh->getDefined(isNull);
+                if (isNull >= 0) { // we have a value
+                  if (! isNull) { // we have a non-null value
+                    Uint64 length = 0;
+                    bh->getLength(length);
+                    // read into buffer
+                    unsigned char* buf = new unsigned char [length];
+                    memset(buf, 'X', length);
+                    Uint32 n = length;
+                    bh->readData(buf, n); // n is in/out
+                    assert(n == length);
+                    // pretty-print
+                    bool first = true;
+                    Uint32 i = 0;
+                    while (i < n) {
+                      unsigned char c = buf[i++];
+                      Uint32 m = 1;
+                      while (i < n && buf[i] == c)
+                        i++, m++;
+                      if (! first)
+                        printf("+");
+                      printf("%u%c", m, c);
+                      first = false;
                     }
-          printf("\npre :  ");
-	  for (int i = 0; i < noEventColumnName; i++) {
-	    if (recAttrPre[i]->isNULL() >= 0) { // we have a value
-	      if (recAttrPre[i]->isNULL() == 0) { // we have a non-null value
-                if (i < 2)
-                  printf("%-5u", recAttrPre[i]->u_32_value());
-                else
-                  printf("%-5.4s", recAttrPre[i]->aRef());
-              } else                              // we have a null value
+                    printf("[%u]", n);
+                    delete [] buf;
+                  } else
                     printf("%-5s", "NULL");
                 } else
-              printf("%-5s", "-");
+                  printf("%-5s", "-"); // no value
+              }
             }
             printf("\n");
           }
+	}
       } else
 	;//printf("timed out\n");
     }
     // don't want to listen to events anymore
-    if (myNdb->dropEventOperation(op)) APIERROR(myNdb->getNdbError());
+    if (myNdb->dropEventOperation(the_op)) APIERROR(myNdb->getNdbError());
+    the_op = 0;
 
     j++;
   }
@@ -250,7 +313,8 @@ int myCreateEvent(Ndb* myNdb,
 		  const char *eventName,
 		  const char *eventTableName,
 		  const char **eventColumnNames,
-		  const int noEventColumnNames)
+		  const int noEventColumnNames,
+                  bool merge_events)
 {
   NdbDictionary::Dictionary *myDict= myNdb->getDictionary();
   if (!myDict) APIERROR(myNdb->getNdbError());
@@ -265,6 +329,7 @@ int myCreateEvent(Ndb* myNdb,
   //  myEvent.addTableEvent(NdbDictionary::Event::TE_DELETE);
 
   myEvent.addEventColumns(noEventColumnNames, eventColumnNames);
+  myEvent.mergeEvents(merge_events);
 
   // Add event to database
   if (myDict->createEvent(myEvent) == 0)

storage/ndb/src/ndbapi/NdbDictionary.cpp

@@ -901,6 +901,11 @@ int NdbDictionary::Event::getNoOfEventColumns() const
   return m_impl.getNoOfEventColumns();
 }
 
+void NdbDictionary::Event::mergeEvents(bool flag)
+{
+  m_impl.m_mergeEvents = flag;
+}
+
 NdbDictionary::Object::Status
 NdbDictionary::Event::getObjectStatus() const
 {

storage/ndb/src/ndbapi/NdbDictionaryImpl.cpp

@@ -1072,6 +1072,7 @@ void NdbEventImpl::init()
   m_tableId= RNIL;
   mi_type= 0;
   m_dur= NdbDictionary::Event::ED_UNDEFINED;
+  m_mergeEvents = false;
   m_tableImpl= NULL;
   m_rep= NdbDictionary::Event::ER_UPDATED;
 }
@@ -2036,7 +2037,7 @@ int
 NdbDictionaryImpl::addBlobTables(NdbTableImpl &t)
 {
   unsigned n= t.m_noOfBlobs;
-  DBUG_ENTER("NdbDictioanryImpl::addBlobTables");
+  DBUG_ENTER("NdbDictionaryImpl::addBlobTables");
   // optimized for blob column being the last one
   // and not looking for more than one if not neccessary
   for (unsigned i = t.m_columns.size(); i > 0 && n > 0;) {
@@ -3151,7 +3152,37 @@ NdbDictionaryImpl::createEvent(NdbEventImpl & evnt)
 #endif
 
   // NdbDictInterface m_receiver;
-  DBUG_RETURN(m_receiver.createEvent(m_ndb, evnt, 0 /* getFlag unset */));
+  if (m_receiver.createEvent(m_ndb, evnt, 0 /* getFlag unset */) != 0)
+    DBUG_RETURN(-1);
+
+  // Create blob events
+  if (evnt.m_mergeEvents && createBlobEvents(evnt) != 0) {
+    int save_code = m_error.code;
+    (void)dropEvent(evnt.m_name.c_str());
+    m_error.code = save_code;
+    DBUG_RETURN(-1);
+  }
+  DBUG_RETURN(0);
+}
+
+int
+NdbDictionaryImpl::createBlobEvents(NdbEventImpl& evnt)
+{
+  DBUG_ENTER("NdbDictionaryImpl::createBlobEvents");
+  NdbTableImpl& t = *evnt.m_tableImpl;
+  Uint32 n = t.m_noOfBlobs;
+  Uint32 i;
+  for (i = 0; i < evnt.m_columns.size() && n > 0; i++) {
+    NdbColumnImpl & c = *evnt.m_columns[i];
+    if (! c.getBlobType() || c.getPartSize() == 0)
+      continue;
+    n--;
+    NdbEventImpl blob_evnt;
+    NdbBlob::getBlobEvent(blob_evnt, &evnt, &c);
+    if (createEvent(blob_evnt) != 0)
+      DBUG_RETURN(-1);
+  }
+  DBUG_RETURN(0);
 }
 
 int
@@ -3400,6 +3431,7 @@ NdbDictionaryImpl::getEvent(const char * eventName)
   
   if ( attributeList_sz > table.getNoOfColumns() )
   {
+    m_error.code = 241;
     DBUG_PRINT("error",("Invalid version, too many columns"));
     delete ev;
     DBUG_RETURN(NULL);
@@ -3409,6 +3441,7 @@ NdbDictionaryImpl::getEvent(const char * eventName)
   for(unsigned id= 0; ev->m_columns.size() < attributeList_sz; id++) {
     if ( id >= table.getNoOfColumns())
     {
+      m_error.code = 241;
       DBUG_PRINT("error",("Invalid version, column %d out of range", id));
       delete ev;
       DBUG_RETURN(NULL);
@@ -3566,13 +3599,54 @@ NdbDictInterface::execSUB_START_REF(NdbApiSignal * signal,
 int 
 NdbDictionaryImpl::dropEvent(const char * eventName)
 {
-  NdbEventImpl *ev= new NdbEventImpl();
-  ev->setName(eventName);
-  int ret= m_receiver.dropEvent(*ev);
-  delete ev;  
+  DBUG_ENTER("NdbDictionaryImpl::dropEvent");
+  DBUG_PRINT("info", ("name=%s", eventName));
 
-  //  printf("__________________RET %u\n", ret);
-  return ret;
+  NdbEventImpl *evnt = getEvent(eventName); // allocated
+  if (evnt == NULL) {
+    if (m_error.code != 723 && // no such table
+        m_error.code != 241)   // invalid table
+      DBUG_RETURN(-1);
+    DBUG_PRINT("info", ("no table, drop by name alone"));
+    evnt = new NdbEventImpl();
+    evnt->setName(eventName);
+  }
+  int ret = dropEvent(*evnt);
+  delete evnt;  
+  DBUG_RETURN(ret);
+}
+
+int
+NdbDictionaryImpl::dropEvent(const NdbEventImpl& evnt)
+{
+  if (dropBlobEvents(evnt) != 0)
+    return -1;
+  if (m_receiver.dropEvent(evnt) != 0)
+    return -1;
+  return 0;
+}
+
+int
+NdbDictionaryImpl::dropBlobEvents(const NdbEventImpl& evnt)
+{
+  DBUG_ENTER("NdbDictionaryImpl::dropBlobEvents");
+  if (evnt.m_tableImpl != 0) {
+    const NdbTableImpl& t = *evnt.m_tableImpl;
+    Uint32 n = t.m_noOfBlobs;
+    Uint32 i;
+    for (i = 0; i < evnt.m_columns.size() && n > 0; i++) {
+      const NdbColumnImpl& c = *evnt.m_columns[i];
+      if (! c.getBlobType() || c.getPartSize() == 0)
+        continue;
+      n--;
+      char bename[MAX_TAB_NAME_SIZE];
+      NdbBlob::getBlobEventName(bename, &evnt, &c);
+      (void)dropEvent(bename);
+    }
+  } else {
+    // could loop over MAX_ATTRIBUTES_IN_TABLE ...
+  }
+  DBUG_RETURN(0);
 }
 
 int

storage/ndb/src/ndbapi/NdbDictionaryImpl.hpp

@@ -277,7 +277,6 @@ public:
   NdbDictionary::Event::EventDurability  getDurability() const;
   void setReport(NdbDictionary::Event::EventReport r);
   NdbDictionary::Event::EventReport  getReport() const;
-  void addEventColumn(const NdbColumnImpl &c);
   int getNoOfEventColumns() const;
 
   void print() {
@@ -295,6 +294,7 @@ public:
   Uint32 mi_type;
   NdbDictionary::Event::EventDurability m_dur;
   NdbDictionary::Event::EventReport m_rep;
+  bool m_mergeEvents;
 
   NdbTableImpl *m_tableImpl;
   BaseString m_tableName;
@@ -547,7 +547,10 @@ public:
 			       NdbTableImpl * table);
 
   int createEvent(NdbEventImpl &);
+  int createBlobEvents(NdbEventImpl &);
   int dropEvent(const char * eventName);
+  int dropEvent(const NdbEventImpl &);
+  int dropBlobEvents(const NdbEventImpl &);
 
   int executeSubscribeEvent(NdbEventOperationImpl &);
   int stopSubscribeEvent(NdbEventOperationImpl &);

storage/ndb/src/ndbapi/NdbEventOperation.cpp

@@ -55,6 +55,18 @@ NdbEventOperation::getPreValue(const char *colName, char *aValue)
   return m_impl.getValue(colName, aValue, 1);
 }
 
+NdbBlob *
+NdbEventOperation::getBlobHandle(const char *colName)
+{
+  return m_impl.getBlobHandle(colName, 0);
+}
+
+NdbBlob *
+NdbEventOperation::getPreBlobHandle(const char *colName)
+{
+  return m_impl.getBlobHandle(colName, 1);
+}
+
 int
 NdbEventOperation::execute()
 {

storage/ndb/src/ndbapi/NdbEventOperationImpl.hpp

@@ -21,6 +21,7 @@
 #include <signaldata/SumaImpl.hpp>
 #include <transporter/TransporterDefinitions.hpp>
 #include <NdbRecAttr.hpp>
+#include <AttributeHeader.hpp>
 
 #define NDB_EVENT_OP_MAGIC_NUMBER 0xA9F301B4
 
@@ -35,9 +36,28 @@ struct EventBufData
   LinearSectionPtr ptr[3];
   unsigned sz;
   NdbEventOperationImpl *m_event_op;
-  EventBufData *m_next; // Next wrt to global order
+
+  /*
+   * Blobs are stored in blob list (m_next_blob) where each entry
+   * is list of parts (m_next) in part number order.
+   *
+   * TODO order by part no and link for fast read and free_list
+   */
+
+  EventBufData *m_next; // Next wrt to global order or Next blob part
+  EventBufData *m_next_blob; // First part in next blob
+
   EventBufData *m_next_hash; // Next in per-GCI hash
   Uint32 m_pkhash; // PK hash (without op) for fast compare
+
+  // Get blob part number from blob data
+  Uint32 get_blob_part_no() {
+    assert(ptr[0].sz > 2);
+    Uint32 pos = AttributeHeader(ptr[0].p[0]).getDataSize() +
+                AttributeHeader(ptr[0].p[1]).getDataSize();
+    Uint32 no = ptr[1].p[pos];
+    return no;
+  }
 };
 
 class EventBufData_list
@@ -70,7 +90,6 @@ EventBufData_list::~EventBufData_list()
 {
 }
 
-
 inline
 int EventBufData_list::is_empty()
 {
@@ -173,9 +192,13 @@ public:
   NdbEventOperation::State getState();
 
   int execute();
+  int execute_nolock();
   int stop();
   NdbRecAttr *getValue(const char *colName, char *aValue, int n);
   NdbRecAttr *getValue(const NdbColumnImpl *, char *aValue, int n);
+  NdbBlob *getBlobHandle(const char *colName, int n);
+  NdbBlob *getBlobHandle(const NdbColumnImpl *, int n);
+  int readBlobParts(char* buf, NdbBlob* blob, Uint32 part, Uint32 count);
   int receive_event();
   Uint64 getGCI();
   Uint64 getLatestGCI();
@@ -199,6 +222,13 @@ public:
   NdbRecAttr *theFirstDataAttrs[2];
   NdbRecAttr *theCurrentDataAttrs[2];
 
+  NdbBlob* theBlobList;
+  union {
+  NdbEventOperationImpl* theBlobOpList;
+  NdbEventOperationImpl* theNextBlobOp;
+  };
+  NdbEventOperationImpl* theMainOp; // blob op pointer to main op
+
   NdbEventOperation::State m_state; /* note connection to mi_type */
   Uint32 mi_type; /* should be == 0 if m_state != EO_EXECUTING
 		   * else same as in EventImpl
@@ -275,6 +305,11 @@ public:
   int merge_data(const SubTableData * const sdata,
                  LinearSectionPtr ptr[3],
                  EventBufData* data);
+  int get_main_data(Gci_container* bucket,
+                    EventBufData_hash::Pos& hpos,
+                    EventBufData* blob_data);
+  void add_blob_data(EventBufData* main_data,
+                     EventBufData* blob_data);
 
   void free_list(EventBufData_list &list);