Skip to content

M
mariadb
Commit cae0b6f9 authored by unknown's avatar unknown
Browse files
Options

ndb - wl#2972 rbr blobs ndb api support 


storage/ndb/include/ndbapi/NdbBlob.hpp:
  rbr blobs ndb api support
storage/ndb/include/ndbapi/NdbDictionary.hpp:
  rbr blobs ndb api support
storage/ndb/include/ndbapi/NdbEventOperation.hpp:
  rbr blobs ndb api support
storage/ndb/ndbapi-examples/ndbapi_event/Makefile:
  rbr blobs ndb api support
storage/ndb/ndbapi-examples/ndbapi_event/ndbapi_event.cpp:
  rbr blobs ndb api support
storage/ndb/src/ndbapi/NdbBlob.cpp:
  rbr blobs ndb api support
storage/ndb/src/ndbapi/NdbDictionary.cpp:
  rbr blobs ndb api support
storage/ndb/src/ndbapi/NdbDictionaryImpl.cpp:
  rbr blobs ndb api support
storage/ndb/src/ndbapi/NdbDictionaryImpl.hpp:
  rbr blobs ndb api support
storage/ndb/src/ndbapi/NdbEventOperation.cpp:
  rbr blobs ndb api support
storage/ndb/src/ndbapi/NdbEventOperationImpl.cpp:
  rbr blobs ndb api support
storage/ndb/src/ndbapi/NdbEventOperationImpl.hpp:
  rbr blobs ndb api support
storage/ndb/test/ndbapi/test_event_merge.cpp:
  rbr blobs ndb api support
parent 1e0f64fb
Expand all Hide whitespace changes
Inline Side-by-side
Showing with 1057 additions and 224 deletions
storage/ndb/include/ndbapi/NdbBlob.hpp
View file @ cae0b6f9
...@@ -28,6 +28,7 @@ class NdbOperation; ...@@ -28,6 +28,7 @@ class NdbOperation;
class NdbRecAttr; class NdbRecAttr;
class NdbTableImpl; class NdbTableImpl;
class NdbColumnImpl; class NdbColumnImpl;
class NdbEventOperationImpl;
/** /**
* @class NdbBlob * @class NdbBlob
...@@ -71,6 +72,10 @@ class NdbColumnImpl; ...@@ -71,6 +72,10 @@ class NdbColumnImpl;
* writes. It avoids execute penalty if nothing is pending. It is not * writes. It avoids execute penalty if nothing is pending. It is not
* needed after execute (obviously) or after next scan result. * 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 * NdbBlob methods return -1 on error and 0 on success, and use output
* parameters when necessary. * parameters when necessary.
* *
...@@ -145,6 +150,12 @@ public: ...@@ -145,6 +150,12 @@ public:
* then the callback is invoked. * then the callback is invoked.
*/ */
int setActiveHook(ActiveHook* activeHook, void* arg); 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. * Check if blob is null.
*/ */
...@@ -191,6 +202,11 @@ public: ...@@ -191,6 +202,11 @@ public:
* Get blob parts table name. Useful only to test programs. * Get blob parts table name. Useful only to test programs.
*/ */
static int getBlobTableName(char* btname, Ndb* anNdb, const char* tableName, const char* columnName); 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 * Return error object. The error may be blob specific (below) or may
* be copied from a failed implicit operation. * be copied from a failed implicit operation.
...@@ -217,17 +233,29 @@ private: ...@@ -217,17 +233,29 @@ private:
friend class NdbScanOperation; friend class NdbScanOperation;
friend class NdbDictionaryImpl; friend class NdbDictionaryImpl;
friend class NdbResultSet; // atNextResult friend class NdbResultSet; // atNextResult
friend class NdbEventBuffer;
friend class NdbEventOperationImpl;
#endif #endif
// state // state
State theState; State theState;
void setState(State newState); 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 // define blob table
static void getBlobTableName(char* btname, const NdbTableImpl* t, const NdbColumnImpl* c); static void getBlobTableName(char* btname, const NdbTableImpl* t, const NdbColumnImpl* c);
static void getBlobTable(NdbTableImpl& bt, 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 api stuff
Ndb* theNdb; Ndb* theNdb;
NdbTransaction* theNdbCon; NdbTransaction* theNdbCon;
NdbOperation* theNdbOp; NdbOperation* theNdbOp;
NdbEventOperationImpl* theEventOp;
NdbEventOperationImpl* theBlobEventOp;
NdbRecAttr* theBlobEventPkRecAttr;
NdbRecAttr* theBlobEventDistRecAttr;
NdbRecAttr* theBlobEventPartRecAttr;
NdbRecAttr* theBlobEventDataRecAttr;
const NdbTableImpl* theTable; const NdbTableImpl* theTable;
const NdbTableImpl* theAccessTable; const NdbTableImpl* theAccessTable;
const NdbTableImpl* theBlobTable; const NdbTableImpl* theBlobTable;
...@@ -263,6 +291,8 @@ private: ...@@ -263,6 +291,8 @@ private:
Buf theHeadInlineBuf; Buf theHeadInlineBuf;
Buf theHeadInlineCopyBuf; // for writeTuple Buf theHeadInlineCopyBuf; // for writeTuple
Buf thePartBuf; Buf thePartBuf;
Buf theBlobEventDataBuf;
Uint32 thePartNumber; // for event
Head* theHead; Head* theHead;
char* theInlineData; char* theInlineData;
NdbRecAttr* theHeadInlineRecAttr; NdbRecAttr* theHeadInlineRecAttr;
...@@ -306,6 +336,8 @@ private: ...@@ -306,6 +336,8 @@ private:
int readDataPrivate(char* buf, Uint32& bytes); int readDataPrivate(char* buf, Uint32& bytes);
int writeDataPrivate(const char* buf, Uint32 bytes); int writeDataPrivate(const char* buf, Uint32 bytes);
int readParts(char* buf, Uint32 part, Uint32 count); 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 insertParts(const char* buf, Uint32 part, Uint32 count);
int updateParts(const char* buf, Uint32 part, Uint32 count); int updateParts(const char* buf, Uint32 part, Uint32 count);
int deleteParts(Uint32 part, Uint32 count); int deleteParts(Uint32 part, Uint32 count);
...@@ -317,19 +349,23 @@ private: ...@@ -317,19 +349,23 @@ private:
int invokeActiveHook(); int invokeActiveHook();
// blob handle maintenance // blob handle maintenance
int atPrepare(NdbTransaction* aCon, NdbOperation* anOp, const NdbColumnImpl* aColumn); 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 preExecute(NdbTransaction::ExecType anExecType, bool& batch);
int postExecute(NdbTransaction::ExecType anExecType); int postExecute(NdbTransaction::ExecType anExecType);
int preCommit(); int preCommit();
int atNextResult(); int atNextResult();
int atNextEvent();
// errors // errors
void setErrorCode(int anErrorCode, bool invalidFlag = true); void setErrorCode(int anErrorCode, bool invalidFlag = true);
void setErrorCode(NdbOperation* anOp, bool invalidFlag = true); void setErrorCode(NdbOperation* anOp, bool invalidFlag = true);
void setErrorCode(NdbTransaction* aCon, bool invalidFlag = true); void setErrorCode(NdbTransaction* aCon, bool invalidFlag = true);
void setErrorCode(NdbEventOperationImpl* anOp, bool invalidFlag = true);
#ifdef VM_TRACE #ifdef VM_TRACE
int getOperationType() const; int getOperationType() const;
friend class NdbOut& operator<<(NdbOut&, const NdbBlob&); friend class NdbOut& operator<<(NdbOut&, const NdbBlob&);
#endif #endif
// list stuff
void next(NdbBlob* obj) { theNext= obj;} void next(NdbBlob* obj) { theNext= obj;}
NdbBlob* next() { return theNext;} NdbBlob* next() { return theNext;}
friend struct Ndb_free_list_t<NdbBlob>; friend struct Ndb_free_list_t<NdbBlob>;
......
storage/ndb/include/ndbapi/NdbDictionary.hpp
View file @ cae0b6f9
...@@ -1124,7 +1124,7 @@ public: ...@@ -1124,7 +1124,7 @@ public:
_TE_NODE_FAILURE=10, _TE_NODE_FAILURE=10,
_TE_SUBSCRIBE=11, _TE_SUBSCRIBE=11,
_TE_UNSUBSCRIBE=12, _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 #endif
/** /**
...@@ -1261,6 +1261,24 @@ public: ...@@ -1261,6 +1261,24 @@ public:
*/ */
int getNoOfEventColumns() const; 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 * Get object status
*/ */
......
storage/ndb/include/ndbapi/NdbEventOperation.hpp
View file @ cae0b6f9
...@@ -150,6 +150,14 @@ public: ...@@ -150,6 +150,14 @@ public:
*/ */
NdbRecAttr *getPreValue(const char *anAttrName, char *aValue = 0); 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; int isOverrun() const;
/** /**
......
storage/ndb/ndbapi-examples/ndbapi_event/Makefile
View file @ cae0b6f9
...@@ -4,7 +4,7 @@ OBJS = ndbapi_event.o ...@@ -4,7 +4,7 @@ OBJS = ndbapi_event.o
CXX = g++ -g CXX = g++ -g
CFLAGS = -c -Wall -fno-rtti -fno-exceptions CFLAGS = -c -Wall -fno-rtti -fno-exceptions
CXXFLAGS = CXXFLAGS =
DEBUG = DEBUG =# -DVM_TRACE
LFLAGS = -Wall LFLAGS = -Wall
TOP_SRCDIR = ../../../.. TOP_SRCDIR = ../../../..
INCLUDE_DIR = $(TOP_SRCDIR)/storage/ndb/include INCLUDE_DIR = $(TOP_SRCDIR)/storage/ndb/include
...@@ -16,8 +16,8 @@ SYS_LIB = ...@@ -16,8 +16,8 @@ SYS_LIB =
$(TARGET): $(OBJS) $(TARGET): $(OBJS)
$(CXX) $(CXXFLAGS) $(LFLAGS) $(LIB_DIR) $(OBJS) -lndbclient -lmysqlclient_r -lmysys -lmystrings -lz $(SYS_LIB) -o $(TARGET) $(CXX) $(CXXFLAGS) $(LFLAGS) $(LIB_DIR) $(OBJS) -lndbclient -lmysqlclient_r -lmysys -lmystrings -lz $(SYS_LIB) -o $(TARGET)
$(TARGET).o: $(SRCS) $(TARGET).o: $(SRCS) Makefile
$(CXX) $(CFLAGS) -I$(INCLUDE_DIR) -I$(INCLUDE_DIR)/ndbapi -I$(TOP_SRCDIR)/include $(SRCS) $(CXX) $(CFLAGS) $(DEBUG) -I$(INCLUDE_DIR) -I$(INCLUDE_DIR)/ndbapi -I$(TOP_SRCDIR)/include $(SRCS)
clean: clean:
rm -f *.o $(TARGET) rm -f *.o $(TARGET)
storage/ndb/ndbapi-examples/ndbapi_event/ndbapi_event.cpp
View file @ cae0b6f9
...@@ -54,26 +54,32 @@ ...@@ -54,26 +54,32 @@
#include <stdio.h> #include <stdio.h>
#include <iostream> #include <iostream>
#include <unistd.h> #include <unistd.h>
#ifdef VM_TRACE
#include <my_global.h>
#endif
#ifndef assert
#include <assert.h>
#endif
/** /**
* * Assume that there is a table which is being updated by
* Assume that there is a table t0 which is being updated by
* another process (e.g. flexBench -l 0 -stdtables). * 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 * shell> mysql -u root
* mysql> create database TEST_DB; * mysql> create database TEST_DB;
* mysql> use 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; * primary key(c0, c2)) engine ndb charset latin1;
* *
* In another window start ndbapi_event, wait until properly started * In another window start ndbapi_event, wait until properly started
*
insert into t0 values (1, 2, 'a', 'b'); insert into t0 values (1, 2, 'a', 'b', null);
insert into t0 values (3, 4, 'c', 'd'); 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 = 'e' where c0 = 1 and c2 = 'a'; -- use pk
update t0 set c3 = 'f'; -- use scan update t0 set c3 = 'f'; -- use scan
update t0 set c3 = 'F'; -- use scan update to 'same' update t0 set c3 = 'F'; -- use scan update to 'same'
...@@ -81,7 +87,18 @@ ...@@ -81,7 +87,18 @@
update t0 set c2 = 'G' where c0 = 1; -- update pk part to 'same' 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 update t0 set c0 = 5, c2 = 'H' where c0 = 3; -- update full PK
delete from t0; 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 * you should see the data popping up in the example window
* *
*/ */
...@@ -95,12 +112,18 @@ int myCreateEvent(Ndb* myNdb, ...@@ -95,12 +112,18 @@ int myCreateEvent(Ndb* myNdb,
const char *eventName, const char *eventName,
const char *eventTableName, const char *eventTableName,
const char **eventColumnName, const char **eventColumnName,
const int noEventColumnName); const int noEventColumnName,
bool merge_events);
int main(int argc, char** argv) int main(int argc, char** argv)
{ {
ndb_init(); 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= Ndb_cluster_connection *cluster_connection=
new Ndb_cluster_connection(); // Object representing the cluster new Ndb_cluster_connection(); // Object representing the cluster
...@@ -134,12 +157,13 @@ int main(int argc, char** argv) ...@@ -134,12 +157,13 @@ int main(int argc, char** argv)
const char *eventName= "CHNG_IN_t0"; const char *eventName= "CHNG_IN_t0";
const char *eventTableName= "t0"; const char *eventTableName= "t0";
const int noEventColumnName= 4; const int noEventColumnName= 5;
const char *eventColumnName[noEventColumnName]= const char *eventColumnName[noEventColumnName]=
{"c0", {"c0",
"c1", "c1",
"c2", "c2",
"c3" "c3",
"c4"
}; };
// Create events // Create events
...@@ -147,9 +171,14 @@ int main(int argc, char** argv) ...@@ -147,9 +171,14 @@ int main(int argc, char** argv)
eventName, eventName,
eventTableName, eventTableName,
eventColumnName, eventColumnName,
noEventColumnName); noEventColumnName,
merge_events);
// Normal values and blobs are unfortunately handled differently..
typedef union { NdbRecAttr* ra; NdbBlob* bh; } RA_BH;
int j= 0; int i, j, k, l;
j = 0;
while (j < 99) { while (j < 99) {
// Start "transaction" for handling events // Start "transaction" for handling events
...@@ -160,12 +189,17 @@ int main(int argc, char** argv) ...@@ -160,12 +189,17 @@ int main(int argc, char** argv)
op->mergeEvents(merge_events); op->mergeEvents(merge_events);
printf("get values\n"); printf("get values\n");
NdbRecAttr* recAttr[noEventColumnName]; RA_BH recAttr[noEventColumnName];
NdbRecAttr* recAttrPre[noEventColumnName]; RA_BH recAttrPre[noEventColumnName];
// primary keys should always be a part of the result // primary keys should always be a part of the result
for (int i = 0; i < noEventColumnName; i++) { for (i = 0; i < noEventColumnName; i++) {
recAttr[i] = op->getValue(eventColumnName[i]); if (i < 4) {
recAttrPre[i] = op->getPreValue(eventColumnName[i]); 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 // set up the callbacks
...@@ -174,13 +208,16 @@ int main(int argc, char** argv) ...@@ -174,13 +208,16 @@ int main(int argc, char** argv)
if (op->execute()) if (op->execute())
APIERROR(op->getNdbError()); APIERROR(op->getNdbError());
int i= 0; NdbEventOperation* the_op = op;
while(i < 40) {
i= 0;
while (i < 40) {
// printf("now waiting for event...\n"); // 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) { if (r > 0) {
// printf("got data! %d\n", r); // printf("got data! %d\n", r);
while ((op= myNdb->nextEvent())) { while ((op= myNdb->nextEvent())) {
assert(the_op == op);
i++; i++;
switch (op->getEventType()) { switch (op->getEventType()) {
case NdbDictionary::Event::TE_INSERT: case NdbDictionary::Event::TE_INSERT:
...@@ -195,40 +232,66 @@ int main(int argc, char** argv) ...@@ -195,40 +232,66 @@ int main(int argc, char** argv)
default: default:
abort(); // should not happen abort(); // should not happen
} }
printf(" gci=%d\n", op->getGCI()); printf(" gci=%d\n", (int)op->getGCI());
printf("post: "); for (k = 0; k <= 1; k++) {
for (int i = 0; i < noEventColumnName; i++) { printf(k == 0 ? "post: " : "pre : ");
if (recAttr[i]->isNULL() >= 0) { // we have a value for (l = 0; l < noEventColumnName; l++) {
if (recAttr[i]->isNULL() == 0) { // we have a non-null value if (l < 4) {
if (i < 2) NdbRecAttr* ra = k == 0 ? recAttr[l].ra : recAttrPre[l].ra;
printf("%-5u", recAttr[i]->u_32_value()); if (ra->isNULL() >= 0) { // we have a value
else if (ra->isNULL() == 0) { // we have a non-null value
printf("%-5.4s", recAttr[i]->aRef()); if (l < 2)
} else // we have a null value printf("%-5u", ra->u_32_value());
printf("%-5s", "NULL"); else
} else printf("%-5.4s", ra->aRef());
printf("%-5s", "-"); } else
printf("%-5s", "NULL");
} else
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("[%u]", n);
delete [] buf;
} else
printf("%-5s", "NULL");
} else
printf("%-5s", "-"); // no value
}
}
printf("\n");
} }
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("%-5s", "NULL");
} else
printf("%-5s", "-");
}
printf("\n");
} }
} else } else
;//printf("timed out\n"); ;//printf("timed out\n");
} }
// don't want to listen to events anymore // 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++; j++;
} }
...@@ -250,7 +313,8 @@ int myCreateEvent(Ndb* myNdb, ...@@ -250,7 +313,8 @@ int myCreateEvent(Ndb* myNdb,
const char *eventName, const char *eventName,
const char *eventTableName, const char *eventTableName,
const char **eventColumnNames, const char **eventColumnNames,
const int noEventColumnNames) const int noEventColumnNames,
bool merge_events)
{ {
NdbDictionary::Dictionary *myDict= myNdb->getDictionary(); NdbDictionary::Dictionary *myDict= myNdb->getDictionary();
if (!myDict) APIERROR(myNdb->getNdbError()); if (!myDict) APIERROR(myNdb->getNdbError());
...@@ -265,6 +329,7 @@ int myCreateEvent(Ndb* myNdb, ...@@ -265,6 +329,7 @@ int myCreateEvent(Ndb* myNdb,
// myEvent.addTableEvent(NdbDictionary::Event::TE_DELETE); // myEvent.addTableEvent(NdbDictionary::Event::TE_DELETE);
myEvent.addEventColumns(noEventColumnNames, eventColumnNames); myEvent.addEventColumns(noEventColumnNames, eventColumnNames);
myEvent.mergeEvents(merge_events);
// Add event to database // Add event to database
if (myDict->createEvent(myEvent) == 0) if (myDict->createEvent(myEvent) == 0)
......
storage/ndb/src/ndbapi/NdbBlob.cpp
View file @ cae0b6f9
This diff is collapsed.
storage/ndb/src/ndbapi/NdbDictionary.cpp
View file @ cae0b6f9
...@@ -901,6 +901,11 @@ int NdbDictionary::Event::getNoOfEventColumns() const ...@@ -901,6 +901,11 @@ int NdbDictionary::Event::getNoOfEventColumns() const
return m_impl.getNoOfEventColumns(); return m_impl.getNoOfEventColumns();
} }
void NdbDictionary::Event::mergeEvents(bool flag)
{
m_impl.m_mergeEvents = flag;
}
NdbDictionary::Object::Status NdbDictionary::Object::Status
NdbDictionary::Event::getObjectStatus() const NdbDictionary::Event::getObjectStatus() const
{ {
......
storage/ndb/src/ndbapi/NdbDictionaryImpl.cpp
View file @ cae0b6f9
...@@ -1072,6 +1072,7 @@ void NdbEventImpl::init() ...@@ -1072,6 +1072,7 @@ void NdbEventImpl::init()
m_tableId= RNIL; m_tableId= RNIL;
mi_type= 0; mi_type= 0;
m_dur= NdbDictionary::Event::ED_UNDEFINED; m_dur= NdbDictionary::Event::ED_UNDEFINED;
m_mergeEvents = false;
m_tableImpl= NULL; m_tableImpl= NULL;
m_rep= NdbDictionary::Event::ER_UPDATED; m_rep= NdbDictionary::Event::ER_UPDATED;
} }
...@@ -2036,7 +2037,7 @@ int ...@@ -2036,7 +2037,7 @@ int
NdbDictionaryImpl::addBlobTables(NdbTableImpl &t) NdbDictionaryImpl::addBlobTables(NdbTableImpl &t)
{ {
unsigned n= t.m_noOfBlobs; unsigned n= t.m_noOfBlobs;
DBUG_ENTER("NdbDictioanryImpl::addBlobTables"); DBUG_ENTER("NdbDictionaryImpl::addBlobTables");
// optimized for blob column being the last one // optimized for blob column being the last one
// and not looking for more than one if not neccessary // and not looking for more than one if not neccessary
for (unsigned i = t.m_columns.size(); i > 0 && n > 0;) { for (unsigned i = t.m_columns.size(); i > 0 && n > 0;) {
...@@ -3151,7 +3152,37 @@ NdbDictionaryImpl::createEvent(NdbEventImpl & evnt) ...@@ -3151,7 +3152,37 @@ NdbDictionaryImpl::createEvent(NdbEventImpl & evnt)
#endif #endif
// NdbDictInterface m_receiver; // 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 int
...@@ -3400,6 +3431,7 @@ NdbDictionaryImpl::getEvent(const char * eventName) ...@@ -3400,6 +3431,7 @@ NdbDictionaryImpl::getEvent(const char * eventName)
if ( attributeList_sz > table.getNoOfColumns() ) if ( attributeList_sz > table.getNoOfColumns() )
{ {
m_error.code = 241;
DBUG_PRINT("error",("Invalid version, too many columns")); DBUG_PRINT("error",("Invalid version, too many columns"));
delete ev; delete ev;
DBUG_RETURN(NULL); DBUG_RETURN(NULL);
...@@ -3409,6 +3441,7 @@ NdbDictionaryImpl::getEvent(const char * eventName) ...@@ -3409,6 +3441,7 @@ NdbDictionaryImpl::getEvent(const char * eventName)
for(unsigned id= 0; ev->m_columns.size() < attributeList_sz; id++) { for(unsigned id= 0; ev->m_columns.size() < attributeList_sz; id++) {
if ( id >= table.getNoOfColumns()) if ( id >= table.getNoOfColumns())
{ {
m_error.code = 241;
DBUG_PRINT("error",("Invalid version, column %d out of range", id)); DBUG_PRINT("error",("Invalid version, column %d out of range", id));
delete ev; delete ev;
DBUG_RETURN(NULL); DBUG_RETURN(NULL);
...@@ -3566,13 +3599,54 @@ NdbDictInterface::execSUB_START_REF(NdbApiSignal * signal, ...@@ -3566,13 +3599,54 @@ NdbDictInterface::execSUB_START_REF(NdbApiSignal * signal,
int int
NdbDictionaryImpl::dropEvent(const char * eventName) NdbDictionaryImpl::dropEvent(const char * eventName)
{ {
NdbEventImpl *ev= new NdbEventImpl(); DBUG_ENTER("NdbDictionaryImpl::dropEvent");
ev->setName(eventName); DBUG_PRINT("info", ("name=%s", eventName));
int ret= m_receiver.dropEvent(*ev);
delete ev;
// printf("__________________RET %u\n", ret); NdbEventImpl *evnt = getEvent(eventName); // allocated
return ret; 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 int
......
storage/ndb/src/ndbapi/NdbDictionaryImpl.hpp
View file @ cae0b6f9
...@@ -277,7 +277,6 @@ public: ...@@ -277,7 +277,6 @@ public:
NdbDictionary::Event::EventDurability getDurability() const; NdbDictionary::Event::EventDurability getDurability() const;
void setReport(NdbDictionary::Event::EventReport r); void setReport(NdbDictionary::Event::EventReport r);
NdbDictionary::Event::EventReport getReport() const; NdbDictionary::Event::EventReport getReport() const;
void addEventColumn(const NdbColumnImpl &c);
int getNoOfEventColumns() const; int getNoOfEventColumns() const;
void print() { void print() {
...@@ -295,6 +294,7 @@ public: ...@@ -295,6 +294,7 @@ public:
Uint32 mi_type; Uint32 mi_type;
NdbDictionary::Event::EventDurability m_dur; NdbDictionary::Event::EventDurability m_dur;
NdbDictionary::Event::EventReport m_rep; NdbDictionary::Event::EventReport m_rep;
bool m_mergeEvents;
NdbTableImpl *m_tableImpl; NdbTableImpl *m_tableImpl;
BaseString m_tableName; BaseString m_tableName;
...@@ -547,7 +547,10 @@ public: ...@@ -547,7 +547,10 @@ public:
NdbTableImpl * table); NdbTableImpl * table);
int createEvent(NdbEventImpl &); int createEvent(NdbEventImpl &);
int createBlobEvents(NdbEventImpl &);
int dropEvent(const char * eventName); int dropEvent(const char * eventName);
int dropEvent(const NdbEventImpl &);
int dropBlobEvents(const NdbEventImpl &);
int executeSubscribeEvent(NdbEventOperationImpl &); int executeSubscribeEvent(NdbEventOperationImpl &);
int stopSubscribeEvent(NdbEventOperationImpl &); int stopSubscribeEvent(NdbEventOperationImpl &);
......
storage/ndb/src/ndbapi/NdbEventOperation.cpp
View file @ cae0b6f9
...@@ -55,6 +55,18 @@ NdbEventOperation::getPreValue(const char *colName, char *aValue) ...@@ -55,6 +55,18 @@ NdbEventOperation::getPreValue(const char *colName, char *aValue)
return m_impl.getValue(colName, aValue, 1); 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 int
NdbEventOperation::execute() NdbEventOperation::execute()
{ {
......
storage/ndb/src/ndbapi/NdbEventOperationImpl.hpp
View file @ cae0b6f9
...@@ -21,6 +21,7 @@ ...@@ -21,6 +21,7 @@
#include <signaldata/SumaImpl.hpp> #include <signaldata/SumaImpl.hpp>
#include <transporter/TransporterDefinitions.hpp> #include <transporter/TransporterDefinitions.hpp>
#include <NdbRecAttr.hpp> #include <NdbRecAttr.hpp>
#include <AttributeHeader.hpp>
#define NDB_EVENT_OP_MAGIC_NUMBER 0xA9F301B4 #define NDB_EVENT_OP_MAGIC_NUMBER 0xA9F301B4
...@@ -35,9 +36,28 @@ struct EventBufData ...@@ -35,9 +36,28 @@ struct EventBufData
LinearSectionPtr ptr[3]; LinearSectionPtr ptr[3];
unsigned sz; unsigned sz;
NdbEventOperationImpl *m_event_op; 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 EventBufData *m_next_hash; // Next in per-GCI hash
Uint32 m_pkhash; // PK hash (without op) for fast compare 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 class EventBufData_list
...@@ -70,7 +90,6 @@ EventBufData_list::~EventBufData_list() ...@@ -70,7 +90,6 @@ EventBufData_list::~EventBufData_list()
{ {
} }
inline inline
int EventBufData_list::is_empty() int EventBufData_list::is_empty()
{ {
...@@ -173,9 +192,13 @@ public: ...@@ -173,9 +192,13 @@ public:
NdbEventOperation::State getState(); NdbEventOperation::State getState();
int execute(); int execute();
int execute_nolock();
int stop(); int stop();
NdbRecAttr *getValue(const char *colName, char *aValue, int n); NdbRecAttr *getValue(const char *colName, char *aValue, int n);
NdbRecAttr *getValue(const NdbColumnImpl *, 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(); int receive_event();
Uint64 getGCI(); Uint64 getGCI();
Uint64 getLatestGCI(); Uint64 getLatestGCI();
...@@ -199,6 +222,13 @@ public: ...@@ -199,6 +222,13 @@ public:
NdbRecAttr *theFirstDataAttrs[2]; NdbRecAttr *theFirstDataAttrs[2];
NdbRecAttr *theCurrentDataAttrs[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 */ NdbEventOperation::State m_state; /* note connection to mi_type */
Uint32 mi_type; /* should be == 0 if m_state != EO_EXECUTING Uint32 mi_type; /* should be == 0 if m_state != EO_EXECUTING
* else same as in EventImpl * else same as in EventImpl
...@@ -275,6 +305,11 @@ public: ...@@ -275,6 +305,11 @@ public:
int merge_data(const SubTableData * const sdata, int merge_data(const SubTableData * const sdata,
LinearSectionPtr ptr[3], LinearSectionPtr ptr[3],
EventBufData* data); 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); void free_list(EventBufData_list &list);
......
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment
GitLab Nexedi Edition | About GitLab | About Nexedi | 沪ICP备2021021310号-2 | 沪ICP备2021021310号-7