database/mariadb
1
0
Fork 0
mirror of https://github.com/MariaDB/server.git synced 2025-07-30 16:24:05 +03:00
Code Activity

ha_ndb blobs

Browse Source
This commit is contained in:
unknown
2004-07-22 12:33:14 +02:00
parent 149b70dc57
commit 9864327a61
19 changed files with 899 additions and 467 deletions
Download Patch File Download Diff File Expand all files Collapse all files

0
mysql-test/r/ndb_blob.result Normal file
View File

0
mysql-test/t/ndb_blob.test Normal file
View File

4
ndb/include/kernel/signaldata/DictTabInfo.hpp
View File

@ -311,7 +311,7 @@ public:
ExtDatetime = NdbSqlUtil::Type::Datetime,
ExtTimespec = NdbSqlUtil::Type::Timespec,
ExtBlob = NdbSqlUtil::Type::Blob,
ExtClob = NdbSqlUtil::Type::Clob
ExtText = NdbSqlUtil::Type::Text
};
// Attribute data interpretation
@ -435,7 +435,7 @@ public:
AttributeArraySize = 12 * AttributeExtLength;
return true;
case DictTabInfo::ExtBlob:
case DictTabInfo::ExtClob:
case DictTabInfo::ExtText:
AttributeType = DictTabInfo::StringType;
AttributeSize = DictTabInfo::an8Bit;
// head + inline part [ attr precision ]

82
ndb/include/ndbapi/NdbBlob.hpp
View File

@ -50,24 +50,33 @@ class NdbColumnImpl;
* - closed: after transaction commit
* - invalid: after rollback or transaction close
*
* NdbBlob supports 2 styles of data access:
* NdbBlob supports 3 styles of data access:
*
* - in prepare phase, NdbBlob methods getValue and setValue are used to
* prepare a read or write of a single blob value of known size
* prepare a read or write of a blob value of known size
*
* - in active phase, NdbBlob methods readData and writeData are used to
* read or write blob data of undetermined size
* - in prepare phase, setActiveHook is used to define a routine which
* is invoked as soon as the handle becomes active
*
* - in active phase, readData and writeData are used to read or write
* blob data of arbitrary size
*
* The styles can be applied in combination (in above order).
*
* Blob operations take effect at next transaction execute. In some
* cases NdbBlob is forced to do implicit executes. To avoid this,
* operate on complete blob parts.
*
* Use NdbConnection::executePendingBlobOps to flush your reads and
* writes. It avoids execute penalty if nothing is pending. It is not
* needed after execute (obviously) or after next scan result.
*
* NdbBlob methods return -1 on error and 0 on success, and use output
* parameters when necessary.
*
* Notes:
* - table and its blob part tables are not created atomically
* - blob data operations take effect at next transaction execute
* - NdbBlob may need to do implicit executes on the transaction
* - read and write of complete parts is much more efficient
* - scan must use the "new" interface NdbScanOperation
* - scan with blobs applies hold-read-lock (at minimum)
* - to update a blob in a read op requires exclusive tuple lock
* - update op in scan must do its own getBlobHandle
* - delete creates implicit, not-accessible blob handles
@ -78,12 +87,16 @@ class NdbColumnImpl;
* - scan must use exclusive locking for now
*
* Todo:
* - add scan method hold-read-lock-until-next + return-keyinfo
* - better check of keyinfo length when setting keys
* - better check of allowed blob op vs locking mode
* - add scan method hold-read-lock + return-keyinfo
* - check keyinfo length when setting keys
* - check allowed blob ops vs locking mode
* - overload control (too many pending ops)
*/
class NdbBlob {
public:
/**
* State.
*/
enum State {
Idle = 0,
Prepared = 1,
@ -92,9 +105,15 @@ public:
Invalid = 9
};
State getState();
/**
* Inline blob header.
*/
struct Head {
Uint64 length;
};
/**
* Prepare to read blob value. The value is available after execute.
* Use isNull to check for NULL and getLength to get the real length
* Use getNull to check for NULL and getLength to get the real length
* and to check for truncation. Sets current read/write position to
* after the data read.
*/
@ -106,6 +125,20 @@ public:
* data to null pointer (0) to create a NULL value.
*/
int setValue(const void* data, Uint32 bytes);
/**
* Callback for setActiveHook. Invoked immediately when the prepared
* operation has been executed (but not committed). Any getValue or
* setValue is done first. The blob handle is active so readData or
* writeData etc can be used to manipulate blob value. A user-defined
* argument is passed along. Returns non-zero on error.
*/
typedef int ActiveHook(NdbBlob* me, void* arg);
/**
* Define callback for blob handle activation. The queue of prepared
* operations will be executed in no commit mode up to this point and
* then the callback is invoked.
*/
int setActiveHook(ActiveHook* activeHook, void* arg);
/**
* Check if blob is null.
*/
@ -115,7 +148,7 @@ public:
*/
int setNull();
/**
* Get current length in bytes. Use isNull to distinguish between
* Get current length in bytes. Use getNull to distinguish between
* length 0 blob and NULL blob.
*/
int getLength(Uint64& length);
@ -180,6 +213,13 @@ public:
static const int ErrAbort = 4268;
// "Unknown blob error"
static const int ErrUnknown = 4269;
/**
* Return info about all blobs in this operation.
*/
// Get first blob in list
NdbBlob* blobsFirstBlob();
// Get next blob in list after this one
NdbBlob* blobsNextBlob();
private:
friend class Ndb;
@ -214,10 +254,11 @@ private:
bool theSetFlag;
const char* theSetBuf;
Uint32 theGetSetBytes;
// head
struct Head {
Uint64 length;
};
// pending ops
Uint8 thePendingBlobOps;
// activation callback
ActiveHook* theActiveHook;
void* theActiveHookArg;
// buffers
struct Buf {
char* data;
@ -235,7 +276,6 @@ private:
char* theInlineData;
NdbRecAttr* theHeadInlineRecAttr;
bool theHeadInlineUpdateFlag;
bool theNewPartFlag;
// length and read/write position
int theNullFlag;
Uint64 theLength;
@ -276,6 +316,11 @@ private:
int insertParts(const char* buf, Uint32 part, Uint32 count);
int updateParts(const char* buf, Uint32 part, Uint32 count);
int deleteParts(Uint32 part, Uint32 count);
// pending ops
int executePendingBlobReads();
int executePendingBlobWrites();
// callbacks
int invokeActiveHook();
// blob handle maintenance
int atPrepare(NdbConnection* aCon, NdbOperation* anOp, const NdbColumnImpl* aColumn);
int preExecute(ExecType anExecType, bool& batch);
@ -287,6 +332,7 @@ private:
void setErrorCode(NdbOperation* anOp, bool invalidFlag = true);
void setErrorCode(NdbConnection* aCon, bool invalidFlag = true);
#ifdef VM_TRACE
int getOperationType() const;
friend class NdbOut& operator<<(NdbOut&, const NdbBlob&);
#endif
};

27
ndb/include/ndbapi/NdbConnection.hpp
View File

@ -431,6 +431,15 @@ public:
/** @} *********************************************************************/
/**
* Execute the transaction in NoCommit mode if there are any not-yet
* executed blob part operations of given types. Otherwise do
* nothing. The flags argument is bitwise OR of (1 << optype) where
* optype comes from NdbOperation::OperationType. Only the basic PK
* ops are used (read, insert, update, delete).
*/
int executePendingBlobOps(Uint8 flags = 0xFF);
private:
/**
* Release completed operations
@ -642,6 +651,7 @@ private:
Uint32 theBuddyConPtr;
// optim: any blobs
bool theBlobFlag;
Uint8 thePendingBlobOps;
static void sendTC_COMMIT_ACK(NdbApiSignal *,
Uint32 transId1, Uint32 transId2,
@ -869,6 +879,21 @@ NdbConnection::OpSent()
theNoOfOpSent++;
}
/******************************************************************************
void executePendingBlobOps();
******************************************************************************/
#include <stdlib.h>
inline
int
NdbConnection::executePendingBlobOps(Uint8 flags)
{
if (thePendingBlobOps & flags) {
// not executeNoBlobs because there can be new ops with blobs
return execute(NoCommit);
}
return 0;
}
inline
Uint32
NdbConnection::ptr2int(){
@ -876,5 +901,3 @@ NdbConnection::ptr2int(){
}
#endif

7
ndb/include/ndbapi/NdbDictionary.hpp
View File

@ -183,7 +183,7 @@ public:
Datetime, ///< Precision down to 1 sec (sizeof(Datetime) == 8 bytes )
Timespec, ///< Precision down to 1 nsec(sizeof(Datetime) == 12 bytes )
Blob, ///< Binary large object (see NdbBlob)
Clob ///< Text blob
Text ///< Text blob
};
/**
@ -309,7 +309,8 @@ public:
/**
* For blob, set or get "part size" i.e. number of bytes to store in
* each tuple of the "blob table". Must be less than 64k.
* each tuple of the "blob table". Can be set to zero to omit parts
* and to allow only inline bytes ("tinyblob").
*/
void setPartSize(int size) { setScale(size); }
int getPartSize() const { return getScale(); }
@ -1060,6 +1061,6 @@ public:
};
};
class NdbOut& operator <<(class NdbOut& ndbout, const NdbDictionary::Column::Type type);
class NdbOut& operator <<(class NdbOut& out, const NdbDictionary::Column& col);
#endif

12
ndb/include/util/NdbSqlUtil.hpp
View File

@ -80,7 +80,7 @@ public:
Datetime, // Precision down to 1 sec (size 8 bytes)
Timespec, // Precision down to 1 nsec (size 12 bytes)
Blob, // Blob
Clob // Text blob
Text // Text blob
};
Enum m_typeId;
Cmp* m_cmp; // set to NULL if cmp not implemented
@ -125,7 +125,7 @@ private:
static Cmp cmpDatetime;
static Cmp cmpTimespec;
static Cmp cmpBlob;
static Cmp cmpClob;
static Cmp cmpText;
};
inline int
@ -344,17 +344,15 @@ NdbSqlUtil::cmp(Uint32 typeId, const Uint32* p1, const Uint32* p2, Uint32 full,
break;
case Type::Blob: // XXX fix
break;
case Type::Clob:
case Type::Text:
{
// skip blob head, the rest is varchar
// skip blob head, the rest is char
const unsigned skip = NDB_BLOB_HEAD_SIZE;
if (size >= skip + 1) {
union { const Uint32* p; const char* v; } u1, u2;
u1.p = p1 + skip;
u2.p = p2 + skip;
// length in first 2 bytes
int k = strncmp(u1.v + 2, u2.v + 2, ((size - skip) << 2) - 2);
return k < 0 ? -1 : k > 0 ? +1 : full == size ? 0 : CmpUnknown;
// TODO
}
return CmpUnknown;
}

8
ndb/src/common/util/NdbSqlUtil.cpp
View File

@ -161,8 +161,8 @@ NdbSqlUtil::m_typeList[] = {
NULL // cmpDatetime
},
{
Type::Clob,
cmpClob
Type::Text,
cmpText
}
};
@ -299,9 +299,9 @@ NdbSqlUtil::cmpBlob(const Uint32* p1, const Uint32* p2, Uint32 full, Uint32 size
}
int
NdbSqlUtil::cmpClob(const Uint32* p1, const Uint32* p2, Uint32 full, Uint32 size)
NdbSqlUtil::cmpText(const Uint32* p1, const Uint32* p2, Uint32 full, Uint32 size)
{
return cmp(Type::Clob, p1, p2, full, size);
return cmp(Type::Text, p1, p2, full, size);
}
#ifdef NDB_SQL_UTIL_TEST

252
ndb/src/ndbapi/NdbBlob.cpp
View File

@ -28,10 +28,11 @@
do { \
static const char* p = getenv("NDB_BLOB_DEBUG"); \
if (p == 0 || *p == 0 || *p == '0') break; \
const char* cname = theColumn == NULL ? "BLOB" : theColumn->m_name.c_str(); \
ndbout << cname << " " << __LINE__ << " " << x << " " << *this << endl; \
static char* prefix = "BLOB"; \
const char* cname = theColumn == NULL ? "-" : theColumn->m_name.c_str(); \
ndbout << prefix << " " << hex << (void*)this << " " << cname; \
ndbout << " " << dec << __LINE__ << " " << x << " " << *this << endl; \
} while (0)
#define EXE() assert(theNdbCon->executeNoBlobs(NoCommit) == 0)
#else
#define DBG(x)
#endif
@ -49,7 +50,7 @@ ndb_blob_debug(const Uint32* data, unsigned size)
/*
* Reading index table directly (as a table) is faster but there are
* bugs or limitations. Keep the code but make possible to choose.
* bugs or limitations. Keep the code and make possible to choose.
*/
static const bool g_ndb_blob_ok_to_read_index_table = false;
@ -116,7 +117,7 @@ NdbBlob::getBlobTable(NdbTableImpl& bt, const NdbTableImpl* t, const NdbColumnIm
case NdbDictionary::Column::Blob:
bc.setType(NdbDictionary::Column::Binary);
break;
case NdbDictionary::Column::Clob:
case NdbDictionary::Column::Text:
bc.setType(NdbDictionary::Column::Char);
break;
default:
@ -155,11 +156,13 @@ NdbBlob::init()
theSetFlag = false;
theSetBuf = NULL;
theGetSetBytes = 0;
thePendingBlobOps = 0;
theActiveHook = NULL;
theActiveHookArg = NULL;
theHead = NULL;
theInlineData = NULL;
theHeadInlineRecAttr = NULL;
theHeadInlineUpdateFlag = false;
theNewPartFlag = false;
theNullFlag = -1;
theLength = 0;
thePos = 0;
@ -270,7 +273,7 @@ NdbBlob::isScanOp()
inline Uint32
NdbBlob::getPartNumber(Uint64 pos)
{
assert(pos >= theInlineSize);
assert(thePartSize != 0 && pos >= theInlineSize);
return (pos - theInlineSize) / thePartSize;
}
@ -322,10 +325,10 @@ int
NdbBlob::setTableKeyValue(NdbOperation* anOp)
{
const Uint32* data = (const Uint32*)theKeyBuf.data;
DBG("setTableKeyValue key=" << ndb_blob_debug(data, theTable->m_sizeOfKeysInWords));
const unsigned columns = theTable->m_columns.size();
unsigned pos = 0;
const unsigned size = theTable->m_columns.size();
DBG("setTableKeyValue key=" << ndb_blob_debug(data, size));
for (unsigned i = 0; i < size; i++) {
for (unsigned i = 0; i < columns; i++) {
NdbColumnImpl* c = theTable->m_columns[i];
assert(c != NULL);
if (c->m_pk) {
@ -345,10 +348,10 @@ int
NdbBlob::setAccessKeyValue(NdbOperation* anOp)
{
const Uint32* data = (const Uint32*)theAccessKeyBuf.data;
DBG("setAccessKeyValue key=" << ndb_blob_debug(data, theAccessTable->m_sizeOfKeysInWords));
const unsigned columns = theAccessTable->m_columns.size();
unsigned pos = 0;
const unsigned size = theAccessTable->m_columns.size();
DBG("setAccessKeyValue key=" << ndb_blob_debug(data, size));
for (unsigned i = 0; i < size; i++) {
for (unsigned i = 0; i < columns; i++) {
NdbColumnImpl* c = theAccessTable->m_columns[i];
assert(c != NULL);
if (c->m_pk) {
@ -479,11 +482,27 @@ NdbBlob::setValue(const void* data, Uint32 bytes)
return 0;
}
// activation hook
int
NdbBlob::setActiveHook(ActiveHook activeHook, void* arg)
{
DBG("setActiveHook hook=" << hex << (void*)activeHook << " arg=" << hex << arg);
if (theState != Prepared) {
setErrorCode(ErrState);
return -1;
}
theActiveHook = activeHook;
theActiveHookArg = arg;
return 0;
}
// misc operations
int
NdbBlob::getNull(bool& isNull)
{
DBG("getNull");
if (theState == Prepared && theSetFlag) {
isNull = (theSetBuf == NULL);
return 0;
@ -520,6 +539,7 @@ NdbBlob::setNull()
int
NdbBlob::getLength(Uint64& len)
{
DBG("getLength");
if (theState == Prepared && theSetFlag) {
len = theGetSetBytes;
return 0;
@ -535,17 +555,17 @@ NdbBlob::getLength(Uint64& len)
int
NdbBlob::truncate(Uint64 length)
{
DBG("truncate kength=" << length);
DBG("truncate length=" << length);
if (theNullFlag == -1) {
setErrorCode(ErrState);
return -1;
}
if (theLength > length) {
if (length >= theInlineSize) {
Uint32 part1 = getPartNumber(length);
if (length > theInlineSize) {
Uint32 part1 = getPartNumber(length - 1);
Uint32 part2 = getPartNumber(theLength - 1);
assert(part2 >= part1);
if (deleteParts(part1, part2 - part1) == -1)
if (part2 > part1 && deleteParts(part1 + 1, part2 - part1) == -1)
return -1;
} else {
if (deleteParts(0, getPartCount()) == -1)
@ -560,6 +580,7 @@ NdbBlob::truncate(Uint64 length)
int
NdbBlob::getPos(Uint64& pos)
{
DBG("getPos");
if (theNullFlag == -1) {
setErrorCode(ErrState);
return -1;
@ -571,6 +592,7 @@ NdbBlob::getPos(Uint64& pos)
int
NdbBlob::setPos(Uint64 pos)
{
DBG("setPos pos=" << pos);
if (theNullFlag == -1) {
setErrorCode(ErrState);
return -1;
@ -629,6 +651,10 @@ NdbBlob::readDataPrivate(Uint64 pos, char* buf, Uint32& bytes)
len -= n;
}
}
if (len > 0 && thePartSize == 0) {
setErrorCode(ErrSeek);
return -1;
}
if (len > 0) {
assert(pos >= theInlineSize);
Uint32 off = (pos - theInlineSize) % thePartSize;
@ -638,11 +664,10 @@ NdbBlob::readDataPrivate(Uint64 pos, char* buf, Uint32& bytes)
Uint32 part = (pos - theInlineSize) / thePartSize;
if (readParts(thePartBuf.data, part, 1) == -1)
return -1;
DBG("force execute");
if (theNdbCon->executeNoBlobs(NoCommit) == -1) {
setErrorCode(theNdbOp);
// need result now
DBG("execute pending part reads");
if (executePendingBlobReads() == -1)
return -1;
}
Uint32 n = thePartSize - off;
if (n > len)
n = len;
@ -673,11 +698,10 @@ NdbBlob::readDataPrivate(Uint64 pos, char* buf, Uint32& bytes)
Uint32 part = (pos - theInlineSize) / thePartSize;
if (readParts(thePartBuf.data, part, 1) == -1)
return -1;
DBG("force execute");
if (theNdbCon->executeNoBlobs(NoCommit) == -1) {
setErrorCode(theNdbOp);
// need result now
DBG("execute pending part reads");
if (executePendingBlobReads() == -1)
return -1;
}
memcpy(buf, thePartBuf.data, len);
Uint32 n = len;
pos += n;
@ -736,29 +760,27 @@ NdbBlob::writeDataPrivate(Uint64 pos, const char* buf, Uint32 bytes)
len -= n;
}
}
if (len > 0 && thePartSize == 0) {
setErrorCode(ErrSeek);
return -1;
}
if (len > 0) {
assert(pos >= theInlineSize);
Uint32 off = (pos - theInlineSize) % thePartSize;
// partial first block
if (off != 0) {
DBG("partial first block pos=" << pos << " len=" << len);
if (theNewPartFlag) {
// must flush insert to guarantee read
DBG("force execute");
if (theNdbCon->executeNoBlobs(NoCommit) == -1) {
setErrorCode(theNdbOp);
return -1;
}
theNewPartFlag = false;
}
// flush writes to guarantee correct read
DBG("execute pending part writes");
if (executePendingBlobWrites() == -1)
return -1;
Uint32 part = (pos - theInlineSize) / thePartSize;
if (readParts(thePartBuf.data, part, 1) == -1)
return -1;
DBG("force execute");
if (theNdbCon->executeNoBlobs(NoCommit) == -1) {
setErrorCode(theNdbOp);
// need result now
DBG("execute pending part reafs");
if (executePendingBlobReads() == -1)
return -1;
}
Uint32 n = thePartSize - off;
if (n > len) {
memset(thePartBuf.data + off + len, theFillChar, n - len);
@ -799,22 +821,16 @@ NdbBlob::writeDataPrivate(Uint64 pos, const char* buf, Uint32 bytes)
assert((pos - theInlineSize) % thePartSize == 0 && len < thePartSize);
Uint32 part = (pos - theInlineSize) / thePartSize;
if (theLength > pos + len) {
if (theNewPartFlag) {
// must flush insert to guarantee read
DBG("force execute");
if (theNdbCon->executeNoBlobs(NoCommit) == -1) {
setErrorCode(theNdbOp);
return -1;
}
theNewPartFlag = false;
}
// flush writes to guarantee correct read
DBG("execute pending part writes");
if (executePendingBlobWrites() == -1)
return -1;
if (readParts(thePartBuf.data, part, 1) == -1)
return -1;
DBG("force execute");
if (theNdbCon->executeNoBlobs(NoCommit) == -1) {
setErrorCode(theNdbOp);
// need result now
DBG("execute pending part reads");
if (executePendingBlobReads() == -1)
return -1;
}
memcpy(thePartBuf.data, buf, len);
if (updateParts(thePartBuf.data, part, 1) == -1)
return -1;
@ -859,6 +875,8 @@ NdbBlob::readParts(char* buf, Uint32 part, Uint32 count)
}
buf += thePartSize;
n++;
thePendingBlobOps |= (1 << NdbOperation::ReadRequest);
theNdbCon->thePendingBlobOps |= (1 << NdbOperation::ReadRequest);
}
return 0;
}
@ -879,7 +897,8 @@ NdbBlob::insertParts(const char* buf, Uint32 part, Uint32 count)
}
buf += thePartSize;
n++;
theNewPartFlag = true;
thePendingBlobOps |= (1 << NdbOperation::InsertRequest);
theNdbCon->thePendingBlobOps |= (1 << NdbOperation::InsertRequest);
}
return 0;
}
@ -900,7 +919,8 @@ NdbBlob::updateParts(const char* buf, Uint32 part, Uint32 count)
}
buf += thePartSize;
n++;
theNewPartFlag = true;
thePendingBlobOps |= (1 << NdbOperation::UpdateRequest);
theNdbCon->thePendingBlobOps |= (1 << NdbOperation::UpdateRequest);
}
return 0;
}
@ -919,6 +939,52 @@ NdbBlob::deleteParts(Uint32 part, Uint32 count)
return -1;
}
n++;
thePendingBlobOps |= (1 << NdbOperation::DeleteRequest);
theNdbCon->thePendingBlobOps |= (1 << NdbOperation::DeleteRequest);
}
return 0;
}
// pending ops
int
NdbBlob::executePendingBlobReads()
{
Uint8 flags = (1 << NdbOperation::ReadRequest);
if (thePendingBlobOps & flags) {
if (theNdbCon->executeNoBlobs(NoCommit) == -1)
return -1;
thePendingBlobOps = 0;
theNdbCon->thePendingBlobOps = 0;
}
return 0;
}
int
NdbBlob::executePendingBlobWrites()
{
Uint8 flags = 0xFF & ~(1 << NdbOperation::ReadRequest);
if (thePendingBlobOps & flags) {
if (theNdbCon->executeNoBlobs(NoCommit) == -1)
return -1;
thePendingBlobOps = 0;
theNdbCon->thePendingBlobOps = 0;
}
return 0;
}
// callbacks
int
NdbBlob::invokeActiveHook()
{
DBG("invokeActiveHook");
assert(theState == Active && theActiveHook != NULL);
int ret = (*theActiveHook)(this, theActiveHookArg);
DBG("invokeActiveHook ret=" << ret);
if (ret != 0) {
// no error is set on blob level
return -1;
}
return 0;
}
@ -948,7 +1014,7 @@ NdbBlob::atPrepare(NdbConnection* aCon, NdbOperation* anOp, const NdbColumnImpl*
partType = NdbDictionary::Column::Binary;
theFillChar = 0x0;
break;
case NdbDictionary::Column::Clob:
case NdbDictionary::Column::Text:
partType = NdbDictionary::Column::Char;
theFillChar = 0x20;
break;
@ -960,22 +1026,21 @@ NdbBlob::atPrepare(NdbConnection* aCon, NdbOperation* anOp, const NdbColumnImpl*
theInlineSize = theColumn->getInlineSize();
thePartSize = theColumn->getPartSize();
theStripeSize = theColumn->getStripeSize();
// blob table sanity check
// sanity check
assert((NDB_BLOB_HEAD_SIZE << 2) == sizeof(Head));
assert(theColumn->m_attrSize * theColumn->m_arraySize == sizeof(Head) + theInlineSize);
getBlobTableName(theBlobTableName, theTable, theColumn);
const NdbDictionary::Table* bt;
const NdbDictionary::Column* bc;
if (theInlineSize >= (1 << 16) ||
thePartSize == 0 ||
thePartSize >= (1 << 16) ||
theStripeSize == 0 ||
(bt = theNdb->theDictionary->getTable(theBlobTableName)) == NULL ||
(bc = bt->getColumn("DATA")) == NULL ||
bc->getType() != partType ||
bc->getLength() != (int)thePartSize) {
setErrorCode(ErrTable);
return -1;
if (thePartSize > 0) {
if (theStripeSize == 0 ||
(bt = theNdb->theDictionary->getTable(theBlobTableName)) == NULL ||
(bc = bt->getColumn("DATA")) == NULL ||
bc->getType() != partType ||
bc->getLength() != (int)thePartSize) {
setErrorCode(ErrTable);
return -1;
}
}
// buffers
theKeyBuf.alloc(theTable->m_sizeOfKeysInWords << 2);
@ -1061,7 +1126,7 @@ NdbBlob::preExecute(ExecType anExecType, bool& batch)
Uint32 bytes = theGetSetBytes - theInlineSize;
if (writeDataPrivate(pos, buf, bytes) == -1)
return -1;
if (anExecType == Commit && theHeadInlineUpdateFlag) {
if (theHeadInlineUpdateFlag) {
// add an operation to update head+inline
NdbOperation* tOp = theNdbCon->getNdbOperation(theTable);
if (tOp == NULL ||
@ -1129,6 +1194,10 @@ NdbBlob::preExecute(ExecType anExecType, bool& batch)
batch = true;
}
}
if (theActiveHook != NULL) {
// need blob head for callback
batch = true;
}
DBG("preExecute out batch=" << batch);
return 0;
}
@ -1145,8 +1214,11 @@ NdbBlob::postExecute(ExecType anExecType)
DBG("postExecute type=" << anExecType);
if (theState == Invalid)
return -1;
if (theState == Active)
if (theState == Active) {
setState(anExecType == NoCommit ? Active : Closed);
DBG("postExecute skip");
return 0;
}
assert(theState == Prepared);
assert(isKeyOp());
if (isIndexOp()) {
@ -1200,8 +1272,12 @@ NdbBlob::postExecute(ExecType anExecType)
if (deleteParts(0, getPartCount()) == -1)
return -1;
}
theNewPartFlag = false;
setState(anExecType == NoCommit ? Active : Closed);
// activation callback
if (theActiveHook != NULL) {
if (invokeActiveHook() == -1)
return -1;
}
DBG("postExecute out");
return 0;
}
@ -1275,20 +1351,18 @@ NdbBlob::atNextResult()
Uint32 bytes = theGetSetBytes - theInlineSize;
if (readDataPrivate(pos, buf, bytes) == -1)
return -1;
// must also execute them
DBG("force execute");
if (theNdbCon->executeNoBlobs(NoCommit) == -1) {
setErrorCode((NdbOperation*)0);
return -1;
}
}
}
setState(Active);
// activation callback
if (theActiveHook != NULL) {
if (invokeActiveHook() == -1)
return -1;
}
DBG("atNextResult out");
return 0;
}
// misc
const NdbDictionary::Column*
@ -1304,6 +1378,9 @@ NdbBlob::setErrorCode(int anErrorCode, bool invalidFlag)
{
DBG("setErrorCode code=" << anErrorCode);
theError.code = anErrorCode;
// conditionally copy error to operation level
if (theNdbOp != NULL && theNdbOp->theError.code == 0)
theNdbOp->setErrorCode(theError.code);
if (invalidFlag)
setState(Invalid);
}
@ -1336,11 +1413,34 @@ NdbBlob::setErrorCode(NdbConnection* aCon, bool invalidFlag)
setErrorCode(code, invalidFlag);
}
// info about all blobs in this operation
NdbBlob*
NdbBlob::blobsFirstBlob()
{
return theNdbOp->theBlobList;
}
NdbBlob*
NdbBlob::blobsNextBlob()
{
return theNext;
}
// debug
#ifdef VM_TRACE
inline int
NdbBlob::getOperationType() const
{
return theNdbOp != NULL ? theNdbOp->theOperationType : -1;
}
NdbOut&
operator<<(NdbOut& out, const NdbBlob& blob)
{
ndbout << dec << "s=" << blob.theState;
ndbout << dec << "o=" << blob.getOperationType();
ndbout << dec << " s=" << blob.theState;
ndbout << dec << " n=" << blob.theNullFlag;;
ndbout << dec << " l=" << blob.theLength;
ndbout << dec << " p=" << blob.thePos;

67
ndb/src/ndbapi/NdbConnection.cpp
View File

@ -89,7 +89,8 @@ NdbConnection::NdbConnection( Ndb* aNdb ) :
// Scan operations
theScanningOp(NULL),
theBuddyConPtr(0xFFFFFFFF),
theBlobFlag(false)
theBlobFlag(false),
thePendingBlobOps(0)
{
theListState = NotInList;
theError.code = 0;
@ -150,6 +151,7 @@ NdbConnection::init()
theBuddyConPtr = 0xFFFFFFFF;
//
theBlobFlag = false;
thePendingBlobOps = 0;
}//NdbConnection::init()
/*****************************************************************************
@ -269,26 +271,34 @@ NdbConnection::execute(ExecType aTypeOfExec,
if (! theBlobFlag)
return executeNoBlobs(aTypeOfExec, abortOption, forceSend);
// execute prepared ops in batches, as requested by blobs
/*
* execute prepared ops in batches, as requested by blobs
* - blob error does not terminate execution
* - blob error sets error on operation
* - if error on operation skip blob calls
*/
ExecType tExecType;
NdbOperation* tPrepOp;
int ret = 0;
do {
tExecType = aTypeOfExec;
tPrepOp = theFirstOpInList;
while (tPrepOp != NULL) {
bool batch = false;
NdbBlob* tBlob = tPrepOp->theBlobList;
while (tBlob != NULL) {
if (tBlob->preExecute(tExecType, batch) == -1)
return -1;
tBlob = tBlob->theNext;
}
if (batch) {
// blob asked to execute all up to here now
tExecType = NoCommit;
break;
if (tPrepOp->theError.code == 0) {
bool batch = false;
NdbBlob* tBlob = tPrepOp->theBlobList;
while (tBlob != NULL) {
if (tBlob->preExecute(tExecType, batch) == -1)
ret = -1;
tBlob = tBlob->theNext;
}
if (batch) {
// blob asked to execute all up to here now
tExecType = NoCommit;
break;
}
}
tPrepOp = tPrepOp->next();
}
@ -304,26 +314,30 @@ NdbConnection::execute(ExecType aTypeOfExec,
if (tExecType == Commit) {
NdbOperation* tOp = theCompletedFirstOp;
while (tOp != NULL) {
NdbBlob* tBlob = tOp->theBlobList;
while (tBlob != NULL) {
if (tBlob->preCommit() == -1)
return -1;
tBlob = tBlob->theNext;
if (tOp->theError.code == 0) {
NdbBlob* tBlob = tOp->theBlobList;
while (tBlob != NULL) {
if (tBlob->preCommit() == -1)
ret = -1;
tBlob = tBlob->theNext;
}
}
tOp = tOp->next();
}
}
if (executeNoBlobs(tExecType, abortOption, forceSend) == -1)
return -1;
ret = -1;
{
NdbOperation* tOp = theCompletedFirstOp;
while (tOp != NULL) {
NdbBlob* tBlob = tOp->theBlobList;
while (tBlob != NULL) {
// may add new operations if batch
if (tBlob->postExecute(tExecType) == -1)
return -1;
tBlob = tBlob->theNext;
if (tOp->theError.code == 0) {
NdbBlob* tBlob = tOp->theBlobList;
while (tBlob != NULL) {
// may add new operations if batch
if (tBlob->postExecute(tExecType) == -1)
ret = -1;
tBlob = tBlob->theNext;
}
}
tOp = tOp->next();
}
@ -338,7 +352,7 @@ NdbConnection::execute(ExecType aTypeOfExec,
}
} while (theFirstOpInList != NULL || tExecType != aTypeOfExec);
return 0;
return ret;
}
int
@ -397,6 +411,7 @@ NdbConnection::executeNoBlobs(ExecType aTypeOfExec,
break;
}
}
thePendingBlobOps = 0;
return 0;
}//NdbConnection::execute()

107
ndb/src/ndbapi/NdbDictionary.cpp
View File

@ -806,73 +806,90 @@ NdbDictionary::Dictionary::getNdbError() const {
return m_impl.getNdbError();
}
NdbOut& operator <<(NdbOut& ndbout, const NdbDictionary::Column::Type type)
// printers
NdbOut&
operator<<(NdbOut& out, const NdbDictionary::Column& col)
{
switch(type){
case NdbDictionary::Column::Bigunsigned:
ndbout << "Bigunsigned";
break;
case NdbDictionary::Column::Unsigned:
ndbout << "Unsigned";
break;
case NdbDictionary::Column::Smallunsigned:
ndbout << "Smallunsigned";
out << col.getName() << " ";
switch (col.getType()) {
case NdbDictionary::Column::Tinyint:
out << "Tinyint";
break;
case NdbDictionary::Column::Tinyunsigned:
ndbout << "Tinyunsigned";
break;
case NdbDictionary::Column::Bigint:
ndbout << "Bigint";
break;
case NdbDictionary::Column::Int:
ndbout << "Int";
out << "Tinyunsigned";
break;
case NdbDictionary::Column::Smallint:
ndbout << "Smallint";
out << "Smallint";
break;
case NdbDictionary::Column::Tinyint:
ndbout << "Tinyint";
break;
case NdbDictionary::Column::Char:
ndbout << "Char";
break;
case NdbDictionary::Column::Varchar:
ndbout << "Varchar";
break;
case NdbDictionary::Column::Float:
ndbout << "Float";
break;
case NdbDictionary::Column::Double:
ndbout << "Double";
case NdbDictionary::Column::Smallunsigned:
out << "Smallunsigned";
break;
case NdbDictionary::Column::Mediumint:
ndbout << "Mediumint";
out << "Mediumint";
break;
case NdbDictionary::Column::Mediumunsigned:
ndbout << "Mediumunsigend";
out << "Mediumunsigned";
break;
case NdbDictionary::Column::Binary:
ndbout << "Binary";
case NdbDictionary::Column::Int:
out << "Int";
break;
case NdbDictionary::Column::Varbinary:
ndbout << "Varbinary";
case NdbDictionary::Column::Unsigned:
out << "Unsigned";
break;
case NdbDictionary::Column::Bigint:
out << "Bigint";
break;
case NdbDictionary::Column::Bigunsigned:
out << "Bigunsigned";
break;
case NdbDictionary::Column::Float:
out << "Float";
break;
case NdbDictionary::Column::Double:
out << "Double";
break;
case NdbDictionary::Column::Decimal:
ndbout << "Decimal";
out << "Decimal(" << col.getScale() << "," << col.getPrecision() << ")";
break;
case NdbDictionary::Column::Char:
out << "Char(" << col.getLength() << ")";
break;
case NdbDictionary::Column::Varchar:
out << "Varchar(" << col.getLength() << ")";
break;
case NdbDictionary::Column::Binary:
out << "Binary(" << col.getLength() << ")";
break;
case NdbDictionary::Column::Varbinary:
out << "Varbinary(" << col.getLength() << ")";
break;
case NdbDictionary::Column::Datetime:
out << "Datetime";
break;
case NdbDictionary::Column::Timespec:
ndbout << "Timespec";
out << "Timespec";
break;
case NdbDictionary::Column::Blob:
ndbout << "Blob";
out << "Blob(" << col.getInlineSize() << "," << col.getPartSize()
<< ";" << col.getStripeSize() << ")";
break;
case NdbDictionary::Column::Text:
out << "Text(" << col.getInlineSize() << "," << col.getPartSize()
<< ";" << col.getStripeSize() << ")";
break;
case NdbDictionary::Column::Undefined:
ndbout << "Undefined";
out << "Undefined";
break;
default:
ndbout << "Unknown type=" << (Uint32)type;
out << "Type" << (Uint32)col.getType();
break;
}
return ndbout;
if (col.getPrimaryKey())
out << " PRIMARY KEY";
else if (! col.getNullable())
out << " NOT NULL";
else
out << " NULL";
return out;
}

8
ndb/src/ndbapi/NdbDictionaryImpl.cpp
View File

@ -181,7 +181,7 @@ NdbColumnImpl::equal(const NdbColumnImpl& col) const
case NdbDictionary::Column::Timespec:
break;
case NdbDictionary::Column::Blob:
case NdbDictionary::Column::Clob:
case NdbDictionary::Column::Text:
if (m_precision != col.m_precision ||
m_scale != col.m_scale ||
m_length != col.m_length) {
@ -1088,7 +1088,7 @@ columnTypeMapping[] = {
{ DictTabInfo::ExtDatetime, NdbDictionary::Column::Datetime },
{ DictTabInfo::ExtTimespec, NdbDictionary::Column::Timespec },
{ DictTabInfo::ExtBlob, NdbDictionary::Column::Blob },
{ DictTabInfo::ExtClob, NdbDictionary::Column::Clob },
{ DictTabInfo::ExtText, NdbDictionary::Column::Text },
{ -1, -1 }
};
@ -1253,7 +1253,7 @@ NdbDictionaryImpl::createBlobTables(NdbTableImpl &t)
{
for (unsigned i = 0; i < t.m_columns.size(); i++) {
NdbColumnImpl & c = *t.m_columns[i];
if (! c.getBlobType())
if (! c.getBlobType() || c.getPartSize() == 0)
continue;
NdbTableImpl bt;
NdbBlob::getBlobTable(bt, &t, &c);
@ -1622,7 +1622,7 @@ NdbDictionaryImpl::dropBlobTables(NdbTableImpl & t)
{
for (unsigned i = 0; i < t.m_columns.size(); i++) {
NdbColumnImpl & c = *t.m_columns[i];
if (! c.getBlobType())
if (! c.getBlobType() || c.getPartSize() == 0)
continue;
char btname[NdbBlob::BlobTableNameSize];
NdbBlob::getBlobTableName(btname, &t, &c);

2
ndb/src/ndbapi/NdbDictionaryImpl.hpp
View File

@ -441,7 +441,7 @@ inline
bool
NdbColumnImpl::getBlobType() const {
return (m_type == NdbDictionary::Column::Blob ||
m_type == NdbDictionary::Column::Clob);
m_type == NdbDictionary::Column::Text);
}
inline

65
ndb/src/ndbapi/NdbRecAttr.cpp
View File

@ -29,6 +29,7 @@ Adjust: 971206 UABRONM First version
#include <ndb_global.h>
#include <NdbOut.hpp>
#include <NdbRecAttr.hpp>
#include <NdbBlob.hpp>
#include "NdbDictionaryImpl.hpp"
#include <NdbTCP.h>
@ -147,78 +148,100 @@ NdbRecAttr::receive_data(const Uint32 * data, Uint32 sz){
return false;
}
NdbOut& operator<<(NdbOut& ndbout, const NdbRecAttr &r)
NdbOut& operator<<(NdbOut& out, const NdbRecAttr &r)
{
if (r.isNULL())
{
ndbout << "[NULL]";
return ndbout;
out << "[NULL]";
return out;
}
if (r.arraySize() > 1)
ndbout << "[";
out << "[";
for (Uint32 j = 0; j < r.arraySize(); j++)
{
if (j > 0)
ndbout << " ";
out << " ";
switch(r.getType())
{
case NdbDictionary::Column::Bigunsigned:
ndbout << r.u_64_value();
out << r.u_64_value();
break;
case NdbDictionary::Column::Unsigned:
ndbout << r.u_32_value();
out << r.u_32_value();
break;
case NdbDictionary::Column::Smallunsigned:
ndbout << r.u_short_value();
out << r.u_short_value();
break;
case NdbDictionary::Column::Tinyunsigned:
ndbout << (unsigned) r.u_char_value();
out << (unsigned) r.u_char_value();
break;
case NdbDictionary::Column::Bigint:
ndbout << r.int64_value();
out << r.int64_value();
break;
case NdbDictionary::Column::Int:
ndbout << r.int32_value();
out << r.int32_value();
break;
case NdbDictionary::Column::Smallint:
ndbout << r.short_value();
out << r.short_value();
break;
case NdbDictionary::Column::Tinyint:
ndbout << (int) r.char_value();
out << (int) r.char_value();
break;
case NdbDictionary::Column::Char:
ndbout.print("%.*s", r.arraySize(), r.aRef());
out.print("%.*s", r.arraySize(), r.aRef());
j = r.arraySize();
break;
case NdbDictionary::Column::Varchar:
{
short len = ntohs(r.u_short_value());
ndbout.print("%.*s", len, r.aRef()+2);
out.print("%.*s", len, r.aRef()+2);
}
j = r.arraySize();
break;
case NdbDictionary::Column::Float:
ndbout << r.float_value();
out << r.float_value();
break;
case NdbDictionary::Column::Double:
ndbout << r.double_value();
out << r.double_value();
break;
case NdbDictionary::Column::Blob:
{
const NdbBlob::Head* h = (const NdbBlob::Head*)r.aRef();
out << h->length << ":";
const unsigned char* p = (const unsigned char*)(h + 1);
unsigned n = r.arraySize() - sizeof(*h);
for (unsigned k = 0; k < n && k < h->length; k++)
out.print("%02X", (int)p[k]);
j = r.arraySize();
}
break;
case NdbDictionary::Column::Text:
{
const NdbBlob::Head* h = (const NdbBlob::Head*)r.aRef();
out << h->length << ":";
const unsigned char* p = (const unsigned char*)(h + 1);
unsigned n = r.arraySize() - sizeof(*h);
for (unsigned k = 0; k < n && k < h->length; k++)
out.print("%c", (int)p[k]);
j = r.arraySize();
}
break;
default: /* no print functions for the rest, just print type */
ndbout << r.getType();
out << r.getType();
j = r.arraySize();
if (j > 1)
ndbout << " %u times" << j;
out << " " << j << " times";
break;
}
}
if (r.arraySize() > 1)
{
ndbout << "]";
out << "]";
}
return ndbout;
return out;
}

7
ndb/src/ndbapi/NdbResultSet.cpp
View File

@ -55,6 +55,13 @@ int NdbResultSet::nextResult(bool fetchAllowed)
return -1;
tBlob = tBlob->theNext;
}
/*
* Flush blob part ops on behalf of user because
* - nextResult is analogous to execute(NoCommit)
* - user is likely to want blob value before next execute
*/
if (m_operation->m_transConnection->executePendingBlobOps() == -1)
return -1;
return 0;
}
return res;

1
ndb/test/include/NDBT_Table.hpp
View File

@ -23,7 +23,6 @@
#include <NdbOut.hpp>
class NDBT_Attribute : public NdbDictionary::Column {
friend class NdbOut& operator <<(class NdbOut&, const NDBT_Attribute &);
public:
NDBT_Attribute(const char* _name,
Column::Type _type,

685
ndb/test/ndbapi/testBlobs.cpp
View File

File diff suppressed because it is too large Load Diff

29
ndb/test/src/NDBT_Table.cpp
View File

@ -18,35 +18,6 @@
#include <NdbTimer.hpp>
#include <NDBT.hpp>
class NdbOut&
operator <<(class NdbOut& ndbout, const NDBT_Attribute & attr){
NdbDictionary::Column::Type type = attr.getType();
ndbout << attr.getName() << " " << type;
switch(type){
case NdbDictionary::Column::Decimal:
ndbout << "(" << attr.getScale() << ", " << attr.getPrecision() << ")";
break;
default:
break;
}
if(attr.getLength() != 1)
ndbout << "[" << attr.getLength() << "]";
if(attr.getNullable())
ndbout << " NULL";
else
ndbout << " NOT NULL";
if(attr.getPrimaryKey())
ndbout << " PRIMARY KEY";
return ndbout;
}
class NdbOut&
operator <<(class NdbOut& ndbout, const NDBT_Table & tab)
{

3
ndb/test/src/NDBT_Test.cpp
View File

@ -830,7 +830,8 @@ void NDBT_TestSuite::execute(Ndb* ndb, const NdbDictionary::Table* pTab,
if(pTab2 == 0 && pDict->createTable(* pTab) != 0){
numTestsFail++;
numTestsExecuted++;
g_err << "ERROR1: Failed to create table " << pTab->getName() << endl;
g_err << "ERROR1: Failed to create table " << pTab->getName()
<< pDict->getNdbError() << endl;
tests[t]->saveTestResult(pTab, FAILED_TO_CREATE);
continue;
}