summaryrefslogtreecommitdiff
path: root/ndb/src/ndbapi
diff options
context:
space:
mode:
Diffstat (limited to 'ndb/src/ndbapi')
-rw-r--r--ndb/src/ndbapi/NdbBlob.cpp359
-rw-r--r--ndb/src/ndbapi/NdbConnection.cpp15
-rw-r--r--ndb/src/ndbapi/NdbDictionaryImpl.cpp9
-rw-r--r--ndb/src/ndbapi/NdbIndexOperation.cpp26
-rw-r--r--ndb/src/ndbapi/NdbOperation.cpp4
-rw-r--r--ndb/src/ndbapi/NdbOperationExec.cpp14
6 files changed, 292 insertions, 135 deletions
diff --git a/ndb/src/ndbapi/NdbBlob.cpp b/ndb/src/ndbapi/NdbBlob.cpp
index feab95d8ca5..5b193870558 100644
--- a/ndb/src/ndbapi/NdbBlob.cpp
+++ b/ndb/src/ndbapi/NdbBlob.cpp
@@ -33,21 +33,24 @@
ndbout << prefix << " " << hex << (void*)this << " " << cname; \
ndbout << " " << dec << __LINE__ << " " << x << " " << *this << endl; \
} while (0)
-#else
-#define DBG(x)
-#endif
static char*
ndb_blob_debug(const Uint32* data, unsigned size)
{
- static char buf[128 + 1]; // MT irrelevant
+ static char buf[200]; // MT irrelevant
buf[0] = 0;
- for (unsigned i = 0; i < size && i < 128 / 4; i++) {
- sprintf(buf + strlen(buf), "%*s%08x", i != 0, "", data[i]);
+ for (unsigned i = 0; i < size; i++) {
+ unsigned n = strlen(buf);
+ if (n + 10 < sizeof(buf))
+ sprintf(buf + n, "%*s%08x", i != 0, "", data[i]);
}
return buf;
}
+#else
+#define DBG(x)
+#endif
+
/*
* Reading index table directly (as a table) is faster but there are
* bugs or limitations. Keep the code and make possible to choose.
@@ -162,6 +165,7 @@ NdbBlob::init()
theHead = NULL;
theInlineData = NULL;
theHeadInlineRecAttr = NULL;
+ theHeadInlineReadOp = NULL;
theHeadInlineUpdateFlag = false;
theNullFlag = -1;
theLength = 0;
@@ -206,6 +210,13 @@ NdbBlob::Buf::alloc(unsigned n)
#endif
}
+void
+NdbBlob::Buf::copyfrom(const NdbBlob::Buf& src)
+{
+ assert(size == src.size);
+ memcpy(data, src.data, size);
+}
+
// classify operations (inline)
inline bool
@@ -226,6 +237,7 @@ NdbBlob::isKeyOp()
return
theNdbOp->theOperationType == NdbOperation::InsertRequest ||
theNdbOp->theOperationType == NdbOperation::UpdateRequest ||
+ theNdbOp->theOperationType == NdbOperation::WriteRequest ||
theNdbOp->theOperationType == NdbOperation::ReadRequest ||
theNdbOp->theOperationType == NdbOperation::ReadExclusive ||
theNdbOp->theOperationType == NdbOperation::DeleteRequest;
@@ -254,6 +266,13 @@ NdbBlob::isUpdateOp()
}
inline bool
+NdbBlob::isWriteOp()
+{
+ return
+ theNdbOp->theOperationType == NdbOperation::WriteRequest;
+}
+
+inline bool
NdbBlob::isDeleteOp()
{
return
@@ -289,7 +308,7 @@ inline Uint32
NdbBlob::getDistKey(Uint32 part)
{
assert(theStripeSize != 0);
- return (part / theStripeSize) % theStripeSize;
+ return part / theStripeSize;
}
// getters and setters
@@ -401,7 +420,7 @@ NdbBlob::getHeadFromRecAttr()
theNullFlag = theHeadInlineRecAttr->isNULL();
assert(theNullFlag != -1);
theLength = ! theNullFlag ? theHead->length : 0;
- DBG("getHeadFromRecAttr out");
+ DBG("getHeadFromRecAttr [out]");
}
int
@@ -453,7 +472,7 @@ NdbBlob::setValue(const void* data, Uint32 bytes)
setErrorCode(ErrState);
return -1;
}
- if (! isInsertOp() && ! isUpdateOp()) {
+ if (! isInsertOp() && ! isUpdateOp() && ! isWriteOp()) {
setErrorCode(ErrUsage);
return -1;
}
@@ -466,11 +485,12 @@ NdbBlob::setValue(const void* data, Uint32 bytes)
theGetSetBytes = bytes;
if (isInsertOp()) {
// write inline part now
- if (theSetBuf != 0) {
- unsigned n = theGetSetBytes;
+ if (theSetBuf != NULL) {
+ Uint32 n = theGetSetBytes;
if (n > theInlineSize)
n = theInlineSize;
- if (writeDataPrivate(0, theSetBuf, n) == -1)
+ assert(thePos == 0);
+ if (writeDataPrivate(theSetBuf, n) == -1)
return -1;
} else {
theNullFlag = true;
@@ -555,7 +575,7 @@ NdbBlob::getLength(Uint64& len)
int
NdbBlob::truncate(Uint64 length)
{
- DBG("truncate length=" << length);
+ DBG("truncate [in] length=" << length);
if (theNullFlag == -1) {
setErrorCode(ErrState);
return -1;
@@ -573,7 +593,10 @@ NdbBlob::truncate(Uint64 length)
}
theLength = length;
theHeadInlineUpdateFlag = true;
+ if (thePos > length)
+ thePos = length;
}
+ DBG("truncate [out]");
return 0;
}
@@ -610,32 +633,20 @@ NdbBlob::setPos(Uint64 pos)
int
NdbBlob::readData(void* data, Uint32& bytes)
{
- if (readData(thePos, data, bytes) == -1)
- return -1;
- thePos += bytes;
- assert(thePos <= theLength);
- return 0;
-}
-
-int
-NdbBlob::readData(Uint64 pos, void* data, Uint32& bytes)
-{
if (theState != Active) {
setErrorCode(ErrState);
return -1;
}
char* buf = static_cast<char*>(data);
- return readDataPrivate(pos, buf, bytes);
+ return readDataPrivate(buf, bytes);
}
int
-NdbBlob::readDataPrivate(Uint64 pos, char* buf, Uint32& bytes)
+NdbBlob::readDataPrivate(char* buf, Uint32& bytes)
{
- DBG("readData pos=" << pos << " bytes=" << bytes);
- if (pos > theLength) {
- setErrorCode(ErrSeek);
- return -1;
- }
+ DBG("readData [in] bytes=" << bytes);
+ assert(thePos <= theLength);
+ Uint64 pos = thePos;
if (bytes > theLength - pos)
bytes = theLength - pos;
Uint32 len = bytes;
@@ -709,38 +720,29 @@ NdbBlob::readDataPrivate(Uint64 pos, char* buf, Uint32& bytes)
len -= n;
}
assert(len == 0);
- return 0;
-}
-
-int
-NdbBlob::writeData(const void* data, Uint32 bytes)
-{
- if (writeData(thePos, data, bytes) == -1)
- return -1;
- thePos += bytes;
+ thePos = pos;
assert(thePos <= theLength);
+ DBG("readData [out]");
return 0;
}
int
-NdbBlob::writeData(Uint64 pos, const void* data, Uint32 bytes)
+NdbBlob::writeData(const void* data, Uint32 bytes)
{
if (theState != Active) {
setErrorCode(ErrState);
return -1;
}
const char* buf = static_cast<const char*>(data);
- return writeDataPrivate(pos, buf, bytes);
+ return writeDataPrivate(buf, bytes);
}
int
-NdbBlob::writeDataPrivate(Uint64 pos, const char* buf, Uint32 bytes)
+NdbBlob::writeDataPrivate(const char* buf, Uint32 bytes)
{
- DBG("writeData pos=" << pos << " bytes=" << bytes);
- if (pos > theLength) {
- setErrorCode(ErrSeek);
- return -1;
- }
+ DBG("writeData [in] bytes=" << bytes);
+ assert(thePos <= theLength);
+ Uint64 pos = thePos;
Uint32 len = bytes;
// any write makes blob not NULL
if (theNullFlag) {
@@ -778,7 +780,7 @@ NdbBlob::writeDataPrivate(Uint64 pos, const char* buf, Uint32 bytes)
if (readParts(thePartBuf.data, part, 1) == -1)
return -1;
// need result now
- DBG("execute pending part reafs");
+ DBG("execute pending part reads");
if (executePendingBlobReads() == -1)
return -1;
Uint32 n = thePartSize - off;
@@ -855,14 +857,16 @@ NdbBlob::writeDataPrivate(Uint64 pos, const char* buf, Uint32 bytes)
theLength = pos;
theHeadInlineUpdateFlag = true;
}
- DBG("writeData out");
+ thePos = pos;
+ assert(thePos <= theLength);
+ DBG("writeData [out]");
return 0;
}
int
NdbBlob::readParts(char* buf, Uint32 part, Uint32 count)
{
- DBG("readParts part=" << part << " count=" << count);
+ DBG("readParts [in] part=" << part << " count=" << count);
Uint32 n = 0;
while (n < count) {
NdbOperation* tOp = theNdbCon->getNdbOperation(theBlobTable);
@@ -873,6 +877,7 @@ NdbBlob::readParts(char* buf, Uint32 part, Uint32 count)
setErrorCode(tOp);
return -1;
}
+ tOp->m_abortOption = AbortOnError;
buf += thePartSize;
n++;
thePendingBlobOps |= (1 << NdbOperation::ReadRequest);
@@ -884,7 +889,7 @@ NdbBlob::readParts(char* buf, Uint32 part, Uint32 count)
int
NdbBlob::insertParts(const char* buf, Uint32 part, Uint32 count)
{
- DBG("insertParts part=" << part << " count=" << count);
+ DBG("insertParts [in] part=" << part << " count=" << count);
Uint32 n = 0;
while (n < count) {
NdbOperation* tOp = theNdbCon->getNdbOperation(theBlobTable);
@@ -895,6 +900,7 @@ NdbBlob::insertParts(const char* buf, Uint32 part, Uint32 count)
setErrorCode(tOp);
return -1;
}
+ tOp->m_abortOption = AbortOnError;
buf += thePartSize;
n++;
thePendingBlobOps |= (1 << NdbOperation::InsertRequest);
@@ -906,7 +912,7 @@ NdbBlob::insertParts(const char* buf, Uint32 part, Uint32 count)
int
NdbBlob::updateParts(const char* buf, Uint32 part, Uint32 count)
{
- DBG("updateParts part=" << part << " count=" << count);
+ DBG("updateParts [in] part=" << part << " count=" << count);
Uint32 n = 0;
while (n < count) {
NdbOperation* tOp = theNdbCon->getNdbOperation(theBlobTable);
@@ -917,6 +923,7 @@ NdbBlob::updateParts(const char* buf, Uint32 part, Uint32 count)
setErrorCode(tOp);
return -1;
}
+ tOp->m_abortOption = AbortOnError;
buf += thePartSize;
n++;
thePendingBlobOps |= (1 << NdbOperation::UpdateRequest);
@@ -928,7 +935,7 @@ NdbBlob::updateParts(const char* buf, Uint32 part, Uint32 count)
int
NdbBlob::deleteParts(Uint32 part, Uint32 count)
{
- DBG("deleteParts part=" << part << " count=" << count);
+ DBG("deleteParts [in] part=" << part << " count=" << count);
Uint32 n = 0;
while (n < count) {
NdbOperation* tOp = theNdbCon->getNdbOperation(theBlobTable);
@@ -938,6 +945,7 @@ NdbBlob::deleteParts(Uint32 part, Uint32 count)
setErrorCode(tOp);
return -1;
}
+ tOp->m_abortOption = AbortOnError;
n++;
thePendingBlobOps |= (1 << NdbOperation::DeleteRequest);
theNdbCon->thePendingBlobOps |= (1 << NdbOperation::DeleteRequest);
@@ -945,6 +953,57 @@ NdbBlob::deleteParts(Uint32 part, Uint32 count)
return 0;
}
+/*
+ * Number of blob parts not known. Used to check for race condition
+ * when writeTuple is used for insert. Deletes all parts found.
+ */
+int
+NdbBlob::deletePartsUnknown(Uint32 part)
+{
+ DBG("deletePartsUnknown [in] part=" << part << " count=all");
+ static const unsigned maxbat = 256;
+ static const unsigned minbat = 1;
+ unsigned bat = minbat;
+ NdbOperation* tOpList[maxbat];
+ Uint32 count = 0;
+ while (true) {
+ Uint32 n;
+ n = 0;
+ while (n < bat) {
+ NdbOperation*& tOp = tOpList[n]; // ref
+ tOp = theNdbCon->getNdbOperation(theBlobTable);
+ if (tOp == NULL ||
+ tOp->deleteTuple() == -1 ||
+ setPartKeyValue(tOp, part + count + n) == -1) {
+ setErrorCode(tOp);
+ return -1;
+ }
+ tOp->m_abortOption = IgnoreError;
+ n++;
+ if (theNdbCon->executeNoBlobs(NoCommit) == -1)
+ return -1;
+ }
+ n = 0;
+ while (n < bat) {
+ NdbOperation* tOp = tOpList[n];
+ if (tOp->theError.code != 0) {
+ if (tOp->theError.code != 626) {
+ setErrorCode(tOp);
+ return -1;
+ }
+ // first non-existent part
+ DBG("deletePartsUnknown [out] count=" << count);
+ return 0;
+ }
+ n++;
+ count++;
+ }
+ bat *= 4;
+ if (bat > maxbat)
+ bat = maxbat;
+ }
+}
+
// pending ops
int
@@ -1007,7 +1066,7 @@ NdbBlob::atPrepare(NdbConnection* aCon, NdbOperation* anOp, const NdbColumnImpl*
theTable = anOp->m_currentTable;
theAccessTable = anOp->m_accessTable;
theColumn = aColumn;
- DBG("atPrepare");
+ DBG("atPrepare [in]");
NdbDictionary::Column::Type partType = NdbDictionary::Column::Undefined;
switch (theColumn->getType()) {
case NdbDictionary::Column::Blob:
@@ -1046,6 +1105,7 @@ NdbBlob::atPrepare(NdbConnection* aCon, NdbOperation* anOp, const NdbColumnImpl*
theKeyBuf.alloc(theTable->m_sizeOfKeysInWords << 2);
theAccessKeyBuf.alloc(theAccessTable->m_sizeOfKeysInWords << 2);
theHeadInlineBuf.alloc(sizeof(Head) + theInlineSize);
+ theHeadInlineCopyBuf.alloc(sizeof(Head) + theInlineSize);
thePartBuf.alloc(thePartSize);
theHead = (Head*)theHeadInlineBuf.data;
theInlineData = theHeadInlineBuf.data + sizeof(Head);
@@ -1080,6 +1140,12 @@ NdbBlob::atPrepare(NdbConnection* aCon, NdbOperation* anOp, const NdbColumnImpl*
theNullFlag = true;
theLength = 0;
}
+ if (isWriteOp()) {
+ // becomes NULL unless set before execute
+ theNullFlag = true;
+ theLength = 0;
+ theHeadInlineUpdateFlag = true;
+ }
supportedOp = true;
}
if (isScanOp()) {
@@ -1093,19 +1159,21 @@ NdbBlob::atPrepare(NdbConnection* aCon, NdbOperation* anOp, const NdbColumnImpl*
return -1;
}
setState(Prepared);
- DBG("atPrepare out");
+ DBG("atPrepare [out]");
return 0;
}
/*
* Before execute of prepared operation. May add new operations before
* this one. May ask that this operation and all before it (a "batch")
- * is executed immediately in no-commit mode.
+ * is executed immediately in no-commit mode. In this case remaining
+ * prepared operations are saved in a separate list. They are added
+ * back after postExecute.
*/
int
NdbBlob::preExecute(ExecType anExecType, bool& batch)
{
- DBG("preExecute");
+ DBG("preExecute [in]");
if (theState == Invalid)
return -1;
assert(theState == Prepared);
@@ -1120,11 +1188,11 @@ NdbBlob::preExecute(ExecType anExecType, bool& batch)
if (isInsertOp()) {
if (theSetFlag && theGetSetBytes > theInlineSize) {
// add ops to write rest of a setValue
- assert(theSetBuf != 0);
- Uint64 pos = theInlineSize;
+ assert(theSetBuf != NULL);
const char* buf = theSetBuf + theInlineSize;
Uint32 bytes = theGetSetBytes - theInlineSize;
- if (writeDataPrivate(pos, buf, bytes) == -1)
+ assert(thePos == theInlineSize);
+ if (writeDataPrivate(buf, bytes) == -1)
return -1;
if (theHeadInlineUpdateFlag) {
// add an operation to update head+inline
@@ -1136,11 +1204,12 @@ NdbBlob::preExecute(ExecType anExecType, bool& batch)
setErrorCode(ErrAbort);
return -1;
}
+ DBG("add op to update head+inline");
}
}
}
if (isTableOp()) {
- if (isUpdateOp() || isDeleteOp()) {
+ if (isUpdateOp() || isWriteOp() || isDeleteOp()) {
// add operation before this one to read head+inline
NdbOperation* tOp = theNdbCon->getNdbOperation(theTable, theNdbOp);
if (tOp == NULL ||
@@ -1150,8 +1219,13 @@ NdbBlob::preExecute(ExecType anExecType, bool& batch)
setErrorCode(tOp);
return -1;
}
+ if (isWriteOp()) {
+ tOp->m_abortOption = IgnoreError;
+ }
+ theHeadInlineReadOp = tOp;
// execute immediately
batch = true;
+ DBG("add op before to read head+inline");
}
}
if (isIndexOp()) {
@@ -1180,6 +1254,7 @@ NdbBlob::preExecute(ExecType anExecType, bool& batch)
}
}
}
+ DBG("added op before to read table key");
if (isUpdateOp() || isDeleteOp()) {
// add op before this one to read head+inline via index
NdbIndexOperation* tOp = theNdbCon->getNdbIndexOperation(theAccessTable->m_index, theTable, theNdbOp);
@@ -1190,15 +1265,43 @@ NdbBlob::preExecute(ExecType anExecType, bool& batch)
setErrorCode(tOp);
return -1;
}
+ if (isWriteOp()) {
+ tOp->m_abortOption = IgnoreError;
+ }
+ theHeadInlineReadOp = tOp;
// execute immediately
batch = true;
+ DBG("added index op before to read head+inline");
+ }
+ if (isWriteOp()) {
+ // XXX until IgnoreError fixed for index op
+ batch = true;
+ }
+ }
+ if (isWriteOp()) {
+ if (theSetFlag) {
+ // write head+inline now
+ theNullFlag = true;
+ theLength = 0;
+ if (theSetBuf != NULL) {
+ Uint32 n = theGetSetBytes;
+ if (n > theInlineSize)
+ n = theInlineSize;
+ assert(thePos == 0);
+ if (writeDataPrivate(theSetBuf, n) == -1)
+ return -1;
+ }
+ if (setHeadInlineValue(theNdbOp) == -1)
+ return -1;
+ // the read op before us may overwrite
+ theHeadInlineCopyBuf.copyfrom(theHeadInlineBuf);
}
}
if (theActiveHook != NULL) {
// need blob head for callback
batch = true;
}
- DBG("preExecute out batch=" << batch);
+ DBG("preExecute [out] batch=" << batch);
return 0;
}
@@ -1211,15 +1314,16 @@ NdbBlob::preExecute(ExecType anExecType, bool& batch)
int
NdbBlob::postExecute(ExecType anExecType)
{
- DBG("postExecute type=" << anExecType);
+ DBG("postExecute [in] type=" << anExecType);
if (theState == Invalid)
return -1;
if (theState == Active) {
setState(anExecType == NoCommit ? Active : Closed);
- DBG("postExecute skip");
+ DBG("postExecute [skip]");
return 0;
}
assert(theState == Prepared);
+ setState(anExecType == NoCommit ? Active : Closed);
assert(isKeyOp());
if (isIndexOp()) {
NdbBlob* tFirstBlob = theNdbOp->theBlobList;
@@ -1231,22 +1335,13 @@ NdbBlob::postExecute(ExecType anExecType)
}
if (isReadOp()) {
getHeadFromRecAttr();
- if (theGetFlag && theGetSetBytes > 0) {
- // copy inline bytes to user buffer
- assert(theGetBuf != NULL);
- unsigned n = theGetSetBytes;
- if (n > theInlineSize)
- n = theInlineSize;
- memcpy(theGetBuf, theInlineData, n);
- }
- if (theGetFlag && theGetSetBytes > theInlineSize) {
- // add ops to read rest of a getValue
- assert(anExecType == NoCommit);
- assert(theGetBuf != 0);
- Uint64 pos = theInlineSize;
- char* buf = theGetBuf + theInlineSize;
- Uint32 bytes = theGetSetBytes - theInlineSize;
- if (readDataPrivate(pos, buf, bytes) == -1)
+ if (setPos(0) == -1)
+ return -1;
+ if (theGetFlag) {
+ assert(theGetSetBytes == 0 || theGetBuf != 0);
+ assert(theGetSetBytes <= theInlineSize || anExecType == NoCommit);
+ Uint32 bytes = theGetSetBytes;
+ if (readDataPrivate(theGetBuf, bytes) == -1)
return -1;
}
}
@@ -1255,10 +1350,11 @@ NdbBlob::postExecute(ExecType anExecType)
getHeadFromRecAttr();
if (theSetFlag) {
// setValue overwrites everything
- if (theSetBuf != 0) {
+ if (theSetBuf != NULL) {
if (truncate(0) == -1)
return -1;
- if (writeDataPrivate(0, theSetBuf, theGetSetBytes) == -1)
+ assert(thePos == 0);
+ if (writeDataPrivate(theSetBuf, theGetSetBytes) == -1)
return -1;
} else {
if (setNull() == -1)
@@ -1266,6 +1362,57 @@ NdbBlob::postExecute(ExecType anExecType)
}
}
}
+ if (isWriteOp() && isTableOp()) {
+ assert(anExecType == NoCommit);
+ if (theHeadInlineReadOp->theError.code == 0) {
+ int tNullFlag = theNullFlag;
+ Uint64 tLength = theLength;
+ Uint64 tPos = thePos;
+ getHeadFromRecAttr();
+ DBG("tuple found");
+ if (truncate(0) == -1)
+ return -1;
+ // restore previous head+inline
+ theHeadInlineBuf.copyfrom(theHeadInlineCopyBuf);
+ theNullFlag = tNullFlag;
+ theLength = tLength;
+ thePos = tPos;
+ } else {
+ if (theHeadInlineReadOp->theError.code != 626) {
+ setErrorCode(theHeadInlineReadOp);
+ return -1;
+ }
+ DBG("tuple not found");
+ /*
+ * Read found no tuple but it is possible that a tuple was
+ * created after the read by another transaction. Delete all
+ * blob parts which may exist.
+ */
+ if (deletePartsUnknown(0) == -1)
+ return -1;
+ }
+ if (theSetFlag && theGetSetBytes > theInlineSize) {
+ assert(theSetBuf != NULL);
+ const char* buf = theSetBuf + theInlineSize;
+ Uint32 bytes = theGetSetBytes - theInlineSize;
+ assert(thePos == theInlineSize);
+ if (writeDataPrivate(buf, bytes) == -1)
+ return -1;
+ }
+ }
+ if (isWriteOp() && isIndexOp()) {
+ // XXX until IgnoreError fixed for index op
+ if (deletePartsUnknown(0) == -1)
+ return -1;
+ if (theSetFlag && theGetSetBytes > theInlineSize) {
+ assert(theSetBuf != NULL);
+ const char* buf = theSetBuf + theInlineSize;
+ Uint32 bytes = theGetSetBytes - theInlineSize;
+ assert(thePos == theInlineSize);
+ if (writeDataPrivate(buf, bytes) == -1)
+ return -1;
+ }
+ }
if (isDeleteOp()) {
assert(anExecType == NoCommit);
getHeadFromRecAttr();
@@ -1278,7 +1425,7 @@ NdbBlob::postExecute(ExecType anExecType)
if (invokeActiveHook() == -1)
return -1;
}
- DBG("postExecute out");
+ DBG("postExecute [out]");
return 0;
}
@@ -1289,12 +1436,12 @@ NdbBlob::postExecute(ExecType anExecType)
int
NdbBlob::preCommit()
{
- DBG("preCommit");
+ DBG("preCommit [in]");
if (theState == Invalid)
return -1;
assert(theState == Active);
assert(isKeyOp());
- if (isInsertOp() || isUpdateOp()) {
+ if (isInsertOp() || isUpdateOp() || isWriteOp()) {
if (theHeadInlineUpdateFlag) {
// add an operation to update head+inline
NdbOperation* tOp = theNdbCon->getNdbOperation(theTable);
@@ -1305,9 +1452,11 @@ NdbBlob::preCommit()
setErrorCode(ErrAbort);
return -1;
}
+ tOp->m_abortOption = AbortOnError;
+ DBG("added op to update head+inline");
}
}
- DBG("preCommit out");
+ DBG("preCommit [out]");
return 0;
}
@@ -1317,13 +1466,10 @@ NdbBlob::preCommit()
int
NdbBlob::atNextResult()
{
- DBG("atNextResult");
+ DBG("atNextResult [in]");
if (theState == Invalid)
return -1;
assert(isScanOp());
- getHeadFromRecAttr();
- // reset position
- thePos = 0;
// get primary key
{ Uint32* data = (Uint32*)theKeyBuf.data;
unsigned size = theTable->m_sizeOfKeysInWords;
@@ -1332,26 +1478,14 @@ NdbBlob::atNextResult()
return -1;
}
}
- if (! theNullFlag) {
- if (theGetFlag && theGetSetBytes > 0) {
- // copy inline bytes to user buffer
- assert(theGetBuf != NULL);
- unsigned n = theGetSetBytes;
- if (n > theLength)
- n = theLength;
- if (n > theInlineSize)
- n = theInlineSize;
- memcpy(theGetBuf, theInlineData, n);
- }
- if (theGetFlag && theGetSetBytes > theInlineSize && theLength > theInlineSize) {
- // add ops to read rest of a getValue
- assert(theGetBuf != 0);
- Uint64 pos = theInlineSize;
- char* buf = theGetBuf + theInlineSize;
- Uint32 bytes = theGetSetBytes - theInlineSize;
- if (readDataPrivate(pos, buf, bytes) == -1)
- return -1;
- }
+ getHeadFromRecAttr();
+ if (setPos(0) == -1)
+ return -1;
+ if (theGetFlag) {
+ assert(theGetSetBytes == 0 || theGetBuf != 0);
+ Uint32 bytes = theGetSetBytes;
+ if (readDataPrivate(theGetBuf, bytes) == -1)
+ return -1;
}
setState(Active);
// activation callback
@@ -1359,7 +1493,7 @@ NdbBlob::atNextResult()
if (invokeActiveHook() == -1)
return -1;
}
- DBG("atNextResult out");
+ DBG("atNextResult [out]");
return 0;
}
@@ -1444,7 +1578,8 @@ operator<<(NdbOut& out, const NdbBlob& blob)
ndbout << dec << " n=" << blob.theNullFlag;;
ndbout << dec << " l=" << blob.theLength;
ndbout << dec << " p=" << blob.thePos;
- ndbout << dec << " u=" << (Uint32) blob.theHeadInlineUpdateFlag;
+ ndbout << dec << " u=" << (Uint32)blob.theHeadInlineUpdateFlag;
+ ndbout << dec << " g=" << (Uint32)blob.theGetSetBytes;
return out;
}
#endif
diff --git a/ndb/src/ndbapi/NdbConnection.cpp b/ndb/src/ndbapi/NdbConnection.cpp
index 1457792cf28..c21a85fd24d 100644
--- a/ndb/src/ndbapi/NdbConnection.cpp
+++ b/ndb/src/ndbapi/NdbConnection.cpp
@@ -170,12 +170,14 @@ Remark: Sets an error code on the connection object from an
operation object.
*****************************************************************************/
void
-NdbConnection::setOperationErrorCodeAbort(int error)
+NdbConnection::setOperationErrorCodeAbort(int error, int abortOption)
{
DBUG_ENTER("NdbConnection::setOperationErrorCodeAbort");
+ if (abortOption == -1)
+ abortOption = m_abortOption;
if (theTransactionIsStarted == false) {
theCommitStatus = Aborted;
- } else if ((m_abortOption == AbortOnError) &&
+ } else if ((abortOption == AbortOnError) &&
(theCommitStatus != Committed) &&
(theCommitStatus != Aborted)) {
theCommitStatus = NeedAbort;
@@ -335,8 +337,11 @@ NdbConnection::execute(ExecType aTypeOfExec,
tOp = tOp->next();
}
}
+
if (executeNoBlobs(tExecType, abortOption, forceSend) == -1)
ret = -1;
+ assert(theFirstOpInList == NULL && theLastOpInList == NULL);
+
{
NdbOperation* tOp = theCompletedFirstOp;
while (tOp != NULL) {
@@ -360,6 +365,7 @@ NdbConnection::execute(ExecType aTypeOfExec,
theLastOpInList->next(tRestOp);
theLastOpInList = tLastOp;
}
+ assert(theFirstOpInList == NULL || tExecType == NoCommit);
} while (theFirstOpInList != NULL || tExecType != aTypeOfExec);
DBUG_RETURN(ret);
@@ -1806,11 +1812,12 @@ Parameters: aErrorCode: The error code.
Remark: An operation was completed with failure.
*******************************************************************************/
int
-NdbConnection::OpCompleteFailure(Uint8 abortOption)
+NdbConnection::OpCompleteFailure(Uint8 abortOption, bool setFailure)
{
Uint32 tNoComp = theNoOfOpCompleted;
Uint32 tNoSent = theNoOfOpSent;
- theCompletionStatus = NdbConnection::CompletedFailure;
+ if (setFailure)
+ theCompletionStatus = NdbConnection::CompletedFailure;
tNoComp++;
theNoOfOpCompleted = tNoComp;
if (tNoComp == tNoSent) {
diff --git a/ndb/src/ndbapi/NdbDictionaryImpl.cpp b/ndb/src/ndbapi/NdbDictionaryImpl.cpp
index cf51a30fe0b..76854cabcd7 100644
--- a/ndb/src/ndbapi/NdbDictionaryImpl.cpp
+++ b/ndb/src/ndbapi/NdbDictionaryImpl.cpp
@@ -47,13 +47,15 @@
* Column
*/
NdbColumnImpl::NdbColumnImpl()
- : NdbDictionary::Column(* this), m_facade(this)
+ : NdbDictionary::Column(* this), m_facade(this),
+ m_attrId(-1)
{
init();
}
NdbColumnImpl::NdbColumnImpl(NdbDictionary::Column & f)
- : NdbDictionary::Column(* this), m_facade(&f)
+ : NdbDictionary::Column(* this), m_facade(&f),
+ m_attrId(-1)
{
init();
}
@@ -93,8 +95,7 @@ NdbColumnImpl::init(Type t)
{
// do not use default_charset_info as it may not be initialized yet
// use binary collation until NDB tests can handle charsets
- CHARSET_INFO* default_cs = &my_charset_latin1_bin;
- m_attrId = -1;
+ CHARSET_INFO* default_cs = &my_charset_bin;
m_type = t;
switch (m_type) {
case Tinyint:
diff --git a/ndb/src/ndbapi/NdbIndexOperation.cpp b/ndb/src/ndbapi/NdbIndexOperation.cpp
index 9abde639914..3f174a61b64 100644
--- a/ndb/src/ndbapi/NdbIndexOperation.cpp
+++ b/ndb/src/ndbapi/NdbIndexOperation.cpp
@@ -71,6 +71,7 @@ NdbIndexOperation::indxInit(const NdbIndexImpl * anIndex,
return -1;
}
m_theIndex = anIndex;
+ m_thePrimaryTable = aTable;
m_accessTable = anIndex->m_table;
m_theIndexLen = 0;
m_theNoOfIndexDefined = 0;
@@ -102,6 +103,12 @@ int NdbIndexOperation::readTuple(NdbOperation::LockMode lm)
};
}
+int NdbIndexOperation::insertTuple()
+{
+ setErrorCode(4200);
+ return -1;
+}
+
int NdbIndexOperation::readTuple()
{
// First check that index is unique
@@ -341,12 +348,11 @@ int NdbIndexOperation::equal_impl(const NdbColumnImpl* tAttrInfo,
theDistrGroupIndicator = 1;
}//if
/**************************************************************************
- * If the operation is an insert request and the attribute is stored then
+ * If the operation is a write request and the attribute is stored then
* we also set the value in the stored part through putting the
* information in the INDXATTRINFO signals.
*************************************************************************/
- if ((tOpType == InsertRequest) ||
- (tOpType == WriteRequest)) {
+ if ((tOpType == WriteRequest)) {
if (!tAttrInfo->m_indexOnly){
// invalid data can crash kernel
if (cs != NULL &&
@@ -357,7 +363,13 @@ int NdbIndexOperation::equal_impl(const NdbColumnImpl* tAttrInfo,
goto equal_error4;
Uint32 ahValue;
Uint32 sz = totalSizeInWords;
- AttributeHeader::init(&ahValue, tAttrId, sz);
+ /*
+ * XXX should be linked in metadata but cannot now because
+ * things can be defined in arbitrary order
+ */
+ const NdbColumnImpl* primaryCol = m_thePrimaryTable->getColumn(tAttrInfo->m_name.c_str());
+ assert(primaryCol != NULL);
+ AttributeHeader::init(&ahValue, primaryCol->m_attrId, sz);
insertATTRINFO( ahValue );
insertATTRINFOloop((Uint32*)aValueToWrite, sizeInWords);
if (bitsInLastWord != 0) {
@@ -369,7 +381,6 @@ int NdbIndexOperation::equal_impl(const NdbColumnImpl* tAttrInfo,
}//if
}//if
}//if
-
/**************************************************************************
* Store the Key information in the TCINDXREQ and INDXKEYINFO signals.
*************************************************************************/
@@ -734,13 +745,10 @@ NdbIndexOperation::receiveTCINDXREF( NdbApiSignal* aSignal)
}//if
theStatus = Finished;
-
+
theNdbCon->theReturnStatus = NdbConnection::ReturnFailure;
Uint32 errorCode = tcIndxRef->errorCode;
theError.code = errorCode;
theNdbCon->setOperationErrorCodeAbort(errorCode);
return theNdbCon->OpCompleteFailure(theNdbCon->m_abortOption);
}//NdbIndexOperation::receiveTCINDXREF()
-
-
-
diff --git a/ndb/src/ndbapi/NdbOperation.cpp b/ndb/src/ndbapi/NdbOperation.cpp
index b0b95d0ff43..88d8a000d50 100644
--- a/ndb/src/ndbapi/NdbOperation.cpp
+++ b/ndb/src/ndbapi/NdbOperation.cpp
@@ -78,7 +78,8 @@ NdbOperation::NdbOperation(Ndb* aNdb) :
m_tcReqGSN(GSN_TCKEYREQ),
m_keyInfoGSN(GSN_KEYINFO),
m_attrInfoGSN(GSN_ATTRINFO),
- theBlobList(NULL)
+ theBlobList(NULL),
+ m_abortOption(-1)
{
theReceiver.init(NdbReceiver::NDB_OPERATION, this);
theError.code = 0;
@@ -167,6 +168,7 @@ NdbOperation::init(const NdbTableImpl* tab, NdbConnection* myConnection){
theTotalNrOfKeyWordInSignal = 8;
theMagicNumber = 0xABCDEF01;
theBlobList = NULL;
+ m_abortOption = -1;
tSignal = theNdb->getSignal();
if (tSignal == NULL)
diff --git a/ndb/src/ndbapi/NdbOperationExec.cpp b/ndb/src/ndbapi/NdbOperationExec.cpp
index f1338ae01e4..fa46e93a57f 100644
--- a/ndb/src/ndbapi/NdbOperationExec.cpp
+++ b/ndb/src/ndbapi/NdbOperationExec.cpp
@@ -191,7 +191,8 @@ NdbOperation::prepareSend(Uint32 aTC_ConnectPtr, Uint64 aTransId)
Uint8 tDirtyIndicator = theDirtyIndicator;
OperationType tOperationType = theOperationType;
Uint32 tTupKeyLen = theTupKeyLen;
- Uint8 abortOption = theNdbCon->m_abortOption;
+ Uint8 abortOption =
+ m_abortOption != -1 ? m_abortOption : theNdbCon->m_abortOption;
tcKeyReq->setDirtyFlag(tReqInfo, tDirtyIndicator);
tcKeyReq->setOperationType(tReqInfo, tOperationType);
@@ -541,17 +542,20 @@ NdbOperation::receiveTCKEYREF( NdbApiSignal* aSignal)
return -1;
}//if
- AbortOption ao = (AbortOption)theNdbCon->m_abortOption;
+ AbortOption ao = (AbortOption)
+ (m_abortOption != -1 ? m_abortOption : theNdbCon->m_abortOption);
theReceiver.m_received_result_length = ~0;
theStatus = Finished;
- theNdbCon->theReturnStatus = NdbConnection::ReturnFailure;
+ // blobs want this
+ if (m_abortOption != IgnoreError)
+ theNdbCon->theReturnStatus = NdbConnection::ReturnFailure;
theError.code = aSignal->readData(4);
- theNdbCon->setOperationErrorCodeAbort(aSignal->readData(4));
+ theNdbCon->setOperationErrorCodeAbort(aSignal->readData(4), ao);
if(theOperationType != ReadRequest || !theSimpleIndicator) // not simple read
- return theNdbCon->OpCompleteFailure(ao);
+ return theNdbCon->OpCompleteFailure(ao, m_abortOption != IgnoreError);
/**
* If TCKEYCONF has arrived
View Lorry Controller Status