Skip to content

M
mariadb
Commit 9a862183 authored by pekka@mysql.com's avatar pekka@mysql.com
Browse files
Options

Merge pnousiainen@bk-internal.mysql.com:/home/bk/mysql-4.1-ndb 

into mysql.com:/space/pekka/ndb/version/my41-blobs
parents 44c339a1 f099d1d3
Hide whitespace changes
Inline Side-by-side
Showing with 3391 additions and 176 deletions
ndb/include/kernel/ndb_limits.h
View file @ 9a862183
......@@ -91,4 +91,9 @@
#define MAX_TTREE_PREF_SIZE 4 // words in min/max prefix each
#define MAX_TTREE_NODE_SLACK 3 // diff between max and min occupancy
/*
* Blobs.
*/
#define NDB_BLOB_HEAD_SIZE 2 // sizeof(NdbBlob::Head) >> 2
#endif
ndb/include/kernel/signaldata/DictTabInfo.hpp
View file @ 9a862183
......@@ -307,7 +307,9 @@ public:
ExtBinary = NdbSqlUtil::Type::Binary,
ExtVarbinary = NdbSqlUtil::Type::Varbinary,
ExtDatetime = NdbSqlUtil::Type::Datetime,
ExtTimespec = NdbSqlUtil::Type::Timespec
ExtTimespec = NdbSqlUtil::Type::Timespec,
ExtBlob = NdbSqlUtil::Type::Blob,
ExtClob = NdbSqlUtil::Type::Clob
};
// Attribute data interpretation
......@@ -430,6 +432,13 @@ public:
AttributeSize = DictTabInfo::an8Bit;
AttributeArraySize = 12 * AttributeExtLength;
return true;
case DictTabInfo::ExtBlob:
case DictTabInfo::ExtClob:
AttributeType = DictTabInfo::StringType;
AttributeSize = DictTabInfo::an8Bit;
// head + inline part [ attr precision ]
AttributeArraySize = (NDB_BLOB_HEAD_SIZE << 2) + AttributeExtPrecision;
return true;
};
return false;
}
......
ndb/include/ndbapi/Ndb.hpp
View file @ 9a862183
......@@ -879,6 +879,7 @@ class NdbScanReceiver;
class Table;
class BaseString;
class NdbEventOperation;
class NdbBlob;
typedef void (* NdbEventCallback)(NdbEventOperation*, Ndb*, void*);
......@@ -964,6 +965,7 @@ class Ndb
friend class NdbIndexOperation;
friend class NdbDictionaryImpl;
friend class NdbDictInterface;
friend class NdbBlob;
public:
/**
......@@ -1452,6 +1454,7 @@ private:
NdbIndexOperation* getIndexOperation();// Get an index operation from idle
class NdbGlobalEventBufferHandle* getGlobalEventBufferHandle();
NdbBlob* getNdbBlob();// Get a blob handle etc
void releaseSignal(NdbApiSignal* anApiSignal);
void releaseSignalsInList(NdbApiSignal** pList);
......@@ -1463,6 +1466,7 @@ private:
void releaseRecAttr (NdbRecAttr* aRecAttr);
void releaseOperation(NdbOperation* anOperation);
void releaseScanOperation(NdbScanOperation* aScanOperation);
void releaseNdbBlob(NdbBlob* aBlob);
void check_send_timeout();
void remove_sent_list(Uint32);
......@@ -1505,6 +1509,7 @@ private:
void freeNdbSubroutine();// Free the first idle NdbSubroutine obj
void freeNdbCall(); // Free the first idle NdbCall obj
void freeNdbScanRec(); // Free the first idle NdbScanRec obj
void freeNdbBlob(); // Free the first etc
NdbConnection* getNdbCon(); // Get a connection from idle list
......@@ -1613,6 +1618,7 @@ private:
NdbSubroutine* theSubroutineList; // First subroutine descriptor in
NdbCall* theCallList; // First call descriptor in list
NdbScanReceiver* theScanList;
NdbBlob* theNdbBlobIdleList;
Uint32 theMyRef; // My block reference
Uint32 theNode; // The node number of our node
......
ndb/include/ndbapi/NdbApi.hpp
View file @ 9a862183
......@@ -28,4 +28,5 @@
#include "NdbDictionary.hpp"
#include "NdbEventOperation.hpp"
#include "NdbPool.hpp"
#include "NdbBlob.hpp"
#endif
ndb/include/ndbapi/NdbBlob.hpp 0 → 100644
View file @ 9a862183
/* Copyright (C) 2003 MySQL AB
This program is free software; you can redistribute it and/or modify
it under the terms of the GNU General Public License as published by
the Free Software Foundation; either version 2 of the License, or
(at your option) any later version.
This program is distributed in the hope that it will be useful,
but WITHOUT ANY WARRANTY; without even the implied warranty of
MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
GNU General Public License for more details.
You should have received a copy of the GNU General Public License
along with this program; if not, write to the Free Software
Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA */
#ifndef NdbBlob_H
#define NdbBlob_H
#include <ndb_types.h>
#include <NdbDictionary.hpp>
#include <NdbConnection.hpp>
#include <NdbError.hpp>
class Ndb;
class NdbConnection;
class NdbOperation;
class NdbRecAttr;
class NdbTableImpl;
class NdbColumnImpl;
/**
* @class NdbBlob
* @brief Blob handle
*
* Blob data is stored in 2 places:
*
* - "header" and "inline bytes" stored in the blob attribute
* - "blob parts" stored in a separate table NDB$BLOB_<t>_<v>_<c>
*
* Inline and part sizes can be set via NdbDictionary::Column methods
* when the table is created.
*
* NdbBlob is a blob handle. To access blob data, the handle must be
* created using NdbOperation::getBlobHandle in operation prepare phase.
* The handle has following states:
*
* - prepared: before the operation is executed
* - active: after execute or next result but before transaction commit
* - closed: after transaction commit
* - invalid: after rollback or transaction close
*
* NdbBlob supports 2 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
*
* - in active phase, NdbBlob methods readData and writeData are used to
* read or write blob data of undetermined size
*
* 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
* - NdbOperation::writeTuple does not support blobs
* - there is no support for an asynchronous interface
*
* Bugs / limitations:
* - 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
*/
class NdbBlob {
public:
enum State {
Idle = 0,
Prepared = 1,
Active = 2,
Closed = 3,
Invalid = 9
};
State getState();
/**
* Prepare to read blob value. The value is available after execute.
* Use isNull 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.
*/
int getValue(void* data, Uint32 bytes);
/**
* Prepare to insert or update blob value. An existing longer blob
* value will be truncated. The data buffer must remain valid until
* execute. Sets current read/write position to after the data. Set
* data to null pointer (0) to create a NULL value.
*/
int setValue(const void* data, Uint32 bytes);
/**
* Check if blob is null.
*/
int getNull(bool& isNull);
/**
* Set blob to NULL.
*/
int setNull();
/**
* Get current length in bytes. Use isNull to distinguish between
* length 0 blob and NULL blob.
*/
int getLength(Uint64& length);
/**
* Truncate blob to given length. Has no effect if the length is
* larger than current length.
*/
int truncate(Uint64 length = 0);
/**
* Get current read/write position.
*/
int getPos(Uint64& pos);
/**
* Set read/write position. Must be between 0 and current length.
* "Sparse blobs" are not supported.
*/
int setPos(Uint64 pos);
/**
* Read at current position and set new position to first byte after
* the data read. A read past blob end returns actual number of bytes
* read in the in/out bytes parameter.
*/
int readData(void* data, Uint32& bytes);
/**
* Read at given position. Does not use or update current position.
*/
int readData(Uint64 pos, void* data, Uint32& bytes);
/**
* Write at current position and set new position to first byte after
* the data written. A write past blob end extends the blob value.
*/
int writeData(const void* data, Uint32 bytes);
/**
* Write at given position. Does not use or update current position.
*/
int writeData(Uint64 pos, const void* data, Uint32 bytes);
/**
* Return the blob column.
*/
const NdbDictionary::Column* getColumn();
/**
* Get blob parts table name. Useful only to test programs.
*/
static const unsigned BlobTableNameSize = 40;
static int getBlobTableName(char* btname, Ndb* anNdb, const char* tableName, const char* columnName);
/**
* Return error object. The error may be blob specific (below) or may
* be copied from a failed implicit operation.
*/
const NdbError& getNdbError() const;
// "Invalid blob attributes or invalid blob parts table"
static const int ErrTable = 4263;
// "Invalid usage of blob attribute"
static const int ErrUsage = 4264;
// "Method is not valid in current blob state"
static const int ErrState = 4265;
// "Invalid blob seek position"
static const int ErrSeek = 4266;
// "Corrupted blob value"
static const int ErrCorrupt = 4267;
// "Error in blob head update forced rollback of transaction"
static const int ErrAbort = 4268;
// "Unknown blob error"
static const int ErrUnknown = 4269;
private:
friend class Ndb;
friend class NdbConnection;
friend class NdbOperation;
friend class NdbScanOperation;
friend class NdbDictionaryImpl;
// state
State theState;
void setState(State newState);
// define blob table
static void getBlobTableName(char* btname, const NdbTableImpl* t, const NdbColumnImpl* c);
static void getBlobTable(NdbTableImpl& bt, const NdbTableImpl* t, const NdbColumnImpl* c);
// table name
char theBlobTableName[BlobTableNameSize];
// ndb api stuff
Ndb* theNdb;
NdbConnection* theNdbCon;
NdbOperation* theNdbOp;
NdbTableImpl* theTable;
NdbTableImpl* theAccessTable;
const NdbColumnImpl* theColumn;
char theFillChar;
// sizes
Uint32 theInlineSize;
Uint32 thePartSize;
Uint32 theStripeSize;
// getValue/setValue
bool theGetFlag;
char* theGetBuf;
bool theSetFlag;
const char* theSetBuf;
Uint32 theGetSetBytes;
// head
struct Head {
Uint64 length;
};
// buffers
struct Buf {
char* data;
unsigned size;
unsigned maxsize;
Buf();
~Buf();
void alloc(unsigned n);
};
Buf theKeyBuf;
Buf theAccessKeyBuf;
Buf theHeadInlineBuf;
Buf thePartBuf;
Head* theHead;
char* theInlineData;
NdbRecAttr* theHeadInlineRecAttr;
bool theHeadInlineUpdateFlag;
bool theNewPartFlag;
// length and read/write position
int theNullFlag;
Uint64 theLength;
Uint64 thePos;
// errors
NdbError theError;
// for keeping in lists
NdbBlob* theNext;
// initialization
NdbBlob();
void init();
void release();
// classify operations
bool isTableOp();
bool isIndexOp();
bool isKeyOp();
bool isReadOp();
bool isInsertOp();
bool isUpdateOp();
bool isDeleteOp();
bool isScanOp();
// computations
Uint32 getPartNumber(Uint64 pos);
Uint32 getPartCount();
Uint32 getDistKey(Uint32 part);
// getters and setters
int getTableKeyValue(NdbOperation* anOp);
int setTableKeyValue(NdbOperation* anOp);
int setAccessKeyValue(NdbOperation* anOp);
int setPartKeyValue(NdbOperation* anOp, Uint32 part);
int getHeadInlineValue(NdbOperation* anOp);
void getHeadFromRecAttr();
int setHeadInlineValue(NdbOperation* anOp);
// data operations
int readDataPrivate(Uint64 pos, char* buf, Uint32& bytes);
int writeDataPrivate(Uint64 pos, const char* buf, Uint32 bytes);
int readParts(char* buf, Uint32 part, Uint32 count);
int insertParts(const char* buf, Uint32 part, Uint32 count);
int updateParts(const char* buf, Uint32 part, Uint32 count);
int deleteParts(Uint32 part, Uint32 count);
// blob handle maintenance
int atPrepare(NdbConnection* aCon, NdbOperation* anOp, const NdbColumnImpl* aColumn);
int preExecute(ExecType anExecType, bool& batch);
int postExecute(ExecType anExecType);
int preCommit();
int atNextResult();
// errors
void setErrorCode(int anErrorCode, bool invalidFlag = true);
void setErrorCode(NdbOperation* anOp, bool invalidFlag = true);
void setErrorCode(NdbConnection* aCon, bool invalidFlag = true);
#ifdef VM_TRACE
friend class NdbOut& operator<<(NdbOut&, const NdbBlob&);
#endif
};
#endif
ndb/include/ndbapi/NdbConnection.hpp
View file @ 9a862183
......@@ -28,6 +28,7 @@ class NdbIndexOperation;
class NdbApiSignal;
class Ndb;
class NdbScanReceiver;
class NdbBlob;
/**
......@@ -160,6 +161,7 @@ class NdbConnection
friend class NdbScanOperation;
friend class NdbIndexOperation;
friend class NdbScanReceiver;
friend class NdbBlob;
public:
......@@ -536,6 +538,10 @@ private:
~NdbConnection();
void init(); // Initialize connection object for new transaction
int executeNoBlobs(ExecType execType,
AbortOption abortOption = AbortOnError,
int force = 0 );
/**
* Set Connected node id
......@@ -625,10 +631,12 @@ private:
void setOperationErrorCodeAbort(int anErrorCode);
int checkMagicNumber(); // Verify correct object
NdbOperation* getNdbOperation(class NdbTableImpl* aTable);
NdbOperation* getNdbOperation(class NdbTableImpl* aTable,
NdbOperation* aNextOp = 0);
NdbScanOperation* getNdbScanOperation(class NdbTableImpl* aTable);
NdbIndexOperation* getNdbIndexOperation(class NdbIndexImpl* anIndex,
class NdbTableImpl* aTable);
class NdbTableImpl* aTable,
NdbOperation* aNextOp = 0);
void handleExecuteCompletion();
......@@ -730,6 +738,8 @@ private:
// nextScanResult.
NdbOperation* theScanningOp; // The operation actually performing the scan
Uint32 theBuddyConPtr;
// optim: any blobs
bool theBlobFlag;
static void sendTC_COMMIT_ACK(NdbApiSignal *,
Uint32 transId1, Uint32 transId2,
......
ndb/include/ndbapi/NdbDictionary.hpp
View file @ 9a862183
......@@ -182,7 +182,8 @@ public:
Varbinary, ///< Max len
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)
Blob, ///< Binary large object (see NdbBlob)
Clob ///< Text blob
};
/**
......@@ -297,7 +298,29 @@ public:
* Array length for column or max length for variable length arrays.
*/
int getLength() const;
/**
* For blob, set or get "inline size" i.e. number of initial bytes
* to store in table's blob attribute. This part is normally in
* main memory and can be indexed and interpreted.
*/
void setInlineSize(int size) { setPrecision(size); }
int getInlineSize() const { return getPrecision(); }
/**
* 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.
*/
void setPartSize(int size) { setScale(size); }
int getPartSize() const { return getScale(); }
/**
* For blob, set or get "stripe size" i.e. number of consecutive
* <em>parts</em> to store in each node group.
*/
void setStripeSize(int size) { setLength(size); }
int getStripeSize() const { return getLength(); }
/**
* Get size of element
*/
......@@ -1029,6 +1052,7 @@ public:
private:
friend class NdbDictionaryImpl;
friend class UtilTransactions;
friend class NdbBlob;
class NdbDictionaryImpl & m_impl;
Dictionary(NdbDictionaryImpl&);
const Table * getIndexTable(const char * indexName,
......
ndb/include/ndbapi/NdbOperation.hpp
View file @ 9a862183
......@@ -28,6 +28,7 @@ class NdbRecAttr;
class NdbOperation;
class NdbConnection;
class NdbColumnImpl;
class NdbBlob;
/**
* @class NdbOperation
......@@ -41,7 +42,8 @@ class NdbOperation
friend class NdbScanReceiver;
friend class NdbScanFilter;
friend class NdbScanFilterImpl;
friend class NdbBlob;
public:
/**
* @name Define Standard Operation Type
......@@ -525,6 +527,17 @@ public:
virtual int setValue(Uint32 anAttrId, Int64 aValue);
virtual int setValue(Uint32 anAttrId, float aValue);
virtual int setValue(Uint32 anAttrId, double aValue);
/**
* This method replaces getValue/setValue for blobs. It creates
* a blob handle NdbBlob. A second call with same argument returns
* the previously created handle. The handle is linked to the
* operation and is maintained automatically.
*
* See NdbBlob for details.
*/
virtual NdbBlob* getBlobHandle(const char* anAttrName);
virtual NdbBlob* getBlobHandle(Uint32 anAttrId);
/** @} *********************************************************************/
/**
......@@ -832,6 +845,11 @@ public:
*/
int getNdbErrorLine();
/**
* Get table name of this operation.
*/
const char* getTableName() const;
/** @} *********************************************************************/
/**
......@@ -953,6 +971,7 @@ protected:
Uint32 len);
NdbRecAttr* getValue(const NdbColumnImpl* anAttrObject, char* aValue = 0);
int setValue(const NdbColumnImpl* anAttrObject, const char* aValue, Uint32 len);
NdbBlob* getBlobHandle(NdbConnection* aCon, const NdbColumnImpl* anAttrObject);
int incValue(const NdbColumnImpl* anAttrObject, Uint32 aValue);
int incValue(const NdbColumnImpl* anAttrObject, Uint64 aValue);
int subValue(const NdbColumnImpl* anAttrObject, Uint32 aValue);
......@@ -997,6 +1016,10 @@ protected:
NdbOperation*
takeOverScanOp(OperationType opType, NdbConnection* updateTrans);
// get table or index key from prepared signals
int getKeyFromTCREQ(Uint32* data, unsigned size);
int getKeyFromKEYINFO20(Uint32* data, unsigned size);
/******************************************************************************
* These are the private variables that are defined in the operation objects.
*****************************************************************************/
......@@ -1095,6 +1118,8 @@ protected:
// saveBoundATTRINFO() moves ATTRINFO here when setBound() is ready
NdbApiSignal* theBoundATTRINFO;
Uint32 theTotalBoundAI_Len;
// Blobs in this operation
NdbBlob* theBlobList;
};
......
ndb/include/ndbapi/NdbScanOperation.hpp
View file @ 9a862183
......@@ -33,6 +33,8 @@
#include <NdbOperation.hpp>
#include <NdbCursorOperation.hpp>
class NdbBlob;
/**
* @class NdbScanOperation
* @brief Class of scan operations for use in transactions.
......@@ -82,6 +84,10 @@ public:
int setValue(Uint32 anAttrId, float aValue);
int setValue(Uint32 anAttrId, double aValue);
#endif
NdbBlob* getBlobHandle(const char* anAttrName);
NdbBlob* getBlobHandle(Uint32 anAttrId);
private:
NdbScanOperation(Ndb* aNdb);
......
ndb/include/util/NdbSqlUtil.hpp
View file @ 9a862183
......@@ -19,6 +19,7 @@
#include <string.h>
#include <ndb_types.h>
#include <kernel/ndb_limits.h>
class NdbSqlUtil {
public:
......@@ -77,7 +78,9 @@ public:
Binary, // Len
Varbinary, // Max len
Datetime, // Precision down to 1 sec (size 8 bytes)
Timespec // Precision down to 1 nsec (size 12 bytes)
Timespec, // Precision down to 1 nsec (size 12 bytes)
Blob, // Blob
Clob // Text blob
};
Enum m_typeId;
Cmp* m_cmp; // set to NULL if cmp not implemented
......@@ -121,6 +124,8 @@ private:
static Cmp cmpVarbinary;
static Cmp cmpDatetime;
static Cmp cmpTimespec;
static Cmp cmpBlob;
static Cmp cmpClob;
};
inline int
......@@ -350,6 +355,23 @@ NdbSqlUtil::cmp(Uint32 typeId, const Uint32* p1, const Uint32* p2, Uint32 full,
break;
case Type::Timespec: // XXX fix this
break;
case Type::Blob: // XXX fix
break;
case Type::Clob:
{
// skip blob head, the rest is varchar
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;
}
return CmpUnknown;
}
break;
}
return CmpError;
}
......
ndb/src/common/util/NdbSqlUtil.cpp
View file @ 9a862183
......@@ -155,6 +155,14 @@ NdbSqlUtil::m_typeList[] = {
{
Type::Timespec,
NULL // cmpTimespec
},
{
Type::Blob,
NULL // cmpDatetime
},
{
Type::Clob,
cmpClob
}
};
......@@ -284,6 +292,18 @@ NdbSqlUtil::cmpTimespec(const Uint32* p1, const Uint32* p2, Uint32 full, Uint32
return cmp(Type::Timespec, p1, p2, full, size);
}
int
NdbSqlUtil::cmpBlob(const Uint32* p1, const Uint32* p2, Uint32 full, Uint32 size)
{
return cmp(Type::Blob, p1, p2, full, size);
}
int
NdbSqlUtil::cmpClob(const Uint32* p1, const Uint32* p2, Uint32 full, Uint32 size)
{
return cmp(Type::Clob, p1, p2, full, size);
}
#ifdef NDB_SQL_UTIL_TEST
#include <NdbTick.h>
......
ndb/src/kernel/blocks/dbtc/DbtcMain.cpp
View file @ 9a862183
......@@ -2736,8 +2736,8 @@ void Dbtc::execTCKEYREQ(Signal* signal)
case ZUPDATE:
jam();
if (Tattrlength == 0) {
TCKEY_abort(signal, 5);
return;
//TCKEY_abort(signal, 5);
//return;
}//if
/*---------------------------------------------------------------------*/
// The missing break is intentional since we also want to set the opLock
......
ndb/src/ndbapi/Makefile.am
View file @ 9a862183
......@@ -35,7 +35,8 @@ libndbapi_la_SOURCES = \
NdbReceiver.cpp \
NdbDictionary.cpp \
NdbDictionaryImpl.cpp \
DictCache.cpp
DictCache.cpp \
NdbBlob.cpp
INCLUDES_LOC = -I$(top_srcdir)/ndb/src/mgmapi
......
ndb/src/ndbapi/Makefile_old
View file @ 9a862183
......@@ -58,7 +58,8 @@ SOURCES = \
NdbSchemaOp.cpp \
NdbUtil.cpp \
NdbReceiver.cpp \
NdbDictionary.cpp NdbDictionaryImpl.cpp DictCache.cpp
NdbDictionary.cpp NdbDictionaryImpl.cpp DictCache.cpp \
NdbBlob.cpp
include $(NDB_TOP)/Epilogue.mk
......
ndb/src/ndbapi/NdbBlob.cpp 0 → 100644
View file @ 9a862183
/* Copyright (C) 2003 MySQL AB
This program is free software; you can redistribute it and/or modify
it under the terms of the GNU General Public License as published by
the Free Software Foundation; either version 2 of the License, or
(at your option) any later version.
This program is distributed in the hope that it will be useful,
but WITHOUT ANY WARRANTY; without even the implied warranty of
MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
GNU General Public License for more details.
You should have received a copy of the GNU General Public License
along with this program; if not, write to the Free Software
Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA */
#include "Ndb.hpp"
#include "NdbDictionaryImpl.hpp"
#include "NdbConnection.hpp"
#include "NdbOperation.hpp"
#include "NdbIndexOperation.hpp"
#include "NdbRecAttr.hpp"
#include "NdbBlob.hpp"
#ifdef NDB_BLOB_DEBUG
#define DBG(x) \
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; \
} while (0)
#define EXE() assert(theNdbCon->executeNoBlobs(NoCommit) == 0)
#else
#define DBG(x)
#endif
/*
* Reading index table directly (as a table) is faster but there are
* bugs or limitations. Keep the code but make possible to choose.
*/
static const bool g_ndb_blob_ok_to_read_index_table = false;
// state (inline)
inline void
NdbBlob::setState(State newState)
{
DBG("setState " << newState);
theState = newState;
}
// define blob table
int
NdbBlob::getBlobTableName(char* btname, Ndb* anNdb, const char* tableName, const char* columnName)
{
NdbTableImpl* t = anNdb->theDictionary->m_impl.getTable(tableName);
if (t == NULL)
return -1;
NdbColumnImpl* c = t->getColumn(columnName);
if (c == NULL)
return -1;
getBlobTableName(btname, t, c);
return 0;
}
void
NdbBlob::getBlobTableName(char* btname, const NdbTableImpl* t, const NdbColumnImpl* c)
{
assert(t != 0 && c != 0 && c->getBlobType());
memset(btname, 0, BlobTableNameSize);
sprintf(btname, "NDB$BLOB_%d_%d_%d", (int)t->m_tableId, (int)t->m_version, (int)c->m_attrId);
}
void
NdbBlob::getBlobTable(NdbTableImpl& bt, const NdbTableImpl* t, const NdbColumnImpl* c)
{
char btname[BlobTableNameSize];
getBlobTableName(btname, t, c);
bt.setName(btname);
bt.setLogging(t->getLogging());
bt.setFragmentType(t->getFragmentType());
{ NdbDictionary::Column bc("DIST");
bc.setType(NdbDictionary::Column::Unsigned);
bc.setPrimaryKey(true);
bc.setDistributionKey(true);
bt.addColumn(bc);
}
{ NdbDictionary::Column bc("PART");
bc.setType(NdbDictionary::Column::Unsigned);
bc.setPrimaryKey(true);
bt.addColumn(bc);
}
{ NdbDictionary::Column bc("PK");
bc.setType(NdbDictionary::Column::Unsigned);
assert(t->m_sizeOfKeysInWords != 0);
bc.setLength(t->m_sizeOfKeysInWords);
bc.setPrimaryKey(true);
bt.addColumn(bc);
}
{ NdbDictionary::Column bc("DATA");
switch (c->m_type) {
case NdbDictionary::Column::Blob:
bc.setType(NdbDictionary::Column::Binary);
break;
case NdbDictionary::Column::Clob:
bc.setType(NdbDictionary::Column::Char);
break;
default:
assert(false);
break;
}
bc.setLength(c->getPartSize());
bt.addColumn(bc);
}
}
// initialization
NdbBlob::NdbBlob()
{
init();
}
void
NdbBlob::init()
{
theState = Idle;
theBlobTableName[0] = 0;
theNdb = NULL;
theNdbCon = NULL;
theNdbOp = NULL;
theTable = NULL;
theAccessTable = NULL;
theColumn = NULL;
theFillChar = 0;
theInlineSize = 0;
thePartSize = 0;
theStripeSize = 0;
theGetFlag = false;
theGetBuf = NULL;
theSetFlag = false;
theSetBuf = NULL;
theGetSetBytes = 0;
theHead = NULL;
theInlineData = NULL;
theHeadInlineRecAttr = NULL;
theHeadInlineUpdateFlag = false;
theNewPartFlag = false;
theNullFlag = -1;
theLength = 0;
thePos = 0;
theNext = NULL;
}
void
NdbBlob::release()
{
setState(Idle);
}
// buffers
NdbBlob::Buf::Buf() :
data(NULL),
size(0),
maxsize(0)
{
}
NdbBlob::Buf::~Buf()
{
delete [] data;
}
void
NdbBlob::Buf::alloc(unsigned n)
{
size = n;
if (maxsize < n) {
delete [] data;
// align to Uint64
if (n % 8 != 0)
n += 8 - n % 8;
data = new char [n];
maxsize = n;
}
#ifdef VM_TRACE
memset(data, 'X', maxsize);
#endif
}
// classify operations (inline)
inline bool
NdbBlob::isTableOp()
{
return theTable == theAccessTable;
}
inline bool
NdbBlob::isIndexOp()
{
return theTable != theAccessTable;
}
inline bool
NdbBlob::isKeyOp()
{
return
theNdbOp->theOperationType == NdbOperation::InsertRequest ||
theNdbOp->theOperationType == NdbOperation::UpdateRequest ||
theNdbOp->theOperationType == NdbOperation::ReadRequest ||
theNdbOp->theOperationType == NdbOperation::ReadExclusive ||
theNdbOp->theOperationType == NdbOperation::DeleteRequest;
}
inline bool
NdbBlob::isReadOp()
{
return
theNdbOp->theOperationType == NdbOperation::ReadRequest ||
theNdbOp->theOperationType == NdbOperation::ReadExclusive;
}
inline bool
NdbBlob::isInsertOp()
{
return
theNdbOp->theOperationType == NdbOperation::InsertRequest;
}
inline bool
NdbBlob::isUpdateOp()
{
return
theNdbOp->theOperationType == NdbOperation::UpdateRequest;
}
inline bool
NdbBlob::isDeleteOp()
{
return
theNdbOp->theOperationType == NdbOperation::DeleteRequest;
}
inline bool
NdbBlob::isScanOp()
{
return
theNdbOp->theOperationType == NdbOperation::OpenScanRequest ||
theNdbOp->theOperationType == NdbOperation::OpenRangeScanRequest;
}
// computations (inline)
inline Uint32
NdbBlob::getPartNumber(Uint64 pos)
{
assert(pos >= theInlineSize);
return (pos - theInlineSize) / thePartSize;
}
inline Uint32
NdbBlob::getPartCount()
{
if (theLength <= theInlineSize)
return 0;
return 1 + getPartNumber(theLength - 1);
}
inline Uint32
NdbBlob::getDistKey(Uint32 part)
{
assert(theStripeSize != 0);
return (part / theStripeSize) % theStripeSize;
}
// getters and setters
int
NdbBlob::getTableKeyValue(NdbOperation* anOp)
{
Uint32* data = (Uint32*)theKeyBuf.data;
unsigned pos = 0;
DBG("getTableKeyValue");
for (unsigned i = 0; i < theTable->m_columns.size(); i++) {
NdbColumnImpl* c = theTable->m_columns[i];
assert(c != NULL);
if (c->m_pk) {
unsigned len = c->m_attrSize * c->m_arraySize;
if (anOp->getValue(c, (char*)&data[pos]) == NULL) {
setErrorCode(anOp);
return -1;
}
// odd bytes receive no data and must be zeroed
while (len % 4 != 0) {
char* p = (char*)data + len++;
*p = 0;
}
pos += len / 4;
}
}
assert(pos == theKeyBuf.size / 4);
return 0;
}
int
NdbBlob::setTableKeyValue(NdbOperation* anOp)
{
const Uint32* data = (const Uint32*)theKeyBuf.data;
unsigned pos = 0;
DBG("setTableKeyValue key0=" << data[0]);
for (unsigned i = 0; i < theTable->m_columns.size(); i++) {
NdbColumnImpl* c = theTable->m_columns[i];
assert(c != NULL);
if (c->m_pk) {
unsigned len = c->m_attrSize * c->m_arraySize;
if (anOp->equal_impl(c, (const char*)&data[pos], len) == -1) {
setErrorCode(anOp);
return -1;
}
pos += (len + 3) / 4;
}
}
assert(pos == theKeyBuf.size / 4);
return 0;
}
int
NdbBlob::setAccessKeyValue(NdbOperation* anOp)
{
const Uint32* data = (const Uint32*)theAccessKeyBuf.data;
unsigned pos = 0;
DBG("setAccessKeyValue key0=" << data[0]);
for (unsigned i = 0; i < theAccessTable->m_columns.size(); i++) {
NdbColumnImpl* c = theAccessTable->m_columns[i];
assert(c != NULL);
if (c->m_pk) {
unsigned len = c->m_attrSize * c->m_arraySize;
if (anOp->equal_impl(c, (const char*)&data[pos], len) == -1) {
setErrorCode(anOp);
return -1;
}
pos += (len + 3) / 4;
}
}
assert(pos == theAccessKeyBuf.size / 4);
return 0;
}
int
NdbBlob::setPartKeyValue(NdbOperation* anOp, Uint32 part)
{
DBG("setPartKeyValue dist=" << getDistKey(part) << " part=" << part << " key0=" << *(Uint32*)theKeyBuf.data);
if (anOp->equal((Uint32)0, getDistKey(part)) == -1 ||
anOp->equal((Uint32)1, part) == -1 ||
anOp->equal((Uint32)2, theKeyBuf.data) == -1) {
setErrorCode(anOp);
return -1;
}
return 0;
}
int
NdbBlob::getHeadInlineValue(NdbOperation* anOp)
{
DBG("getHeadInlineValue");
theHeadInlineRecAttr = anOp->getValue(theColumn, theHeadInlineBuf.data);
if (theHeadInlineRecAttr == NULL) {
setErrorCode(anOp);
return -1;
}
return 0;
}
void
NdbBlob::getHeadFromRecAttr()
{
assert(theHeadInlineRecAttr != NULL);
theNullFlag = theHeadInlineRecAttr->isNULL();
assert(theNullFlag != -1);
theLength = ! theNullFlag ? theHead->length : 0;
DBG("getHeadFromRecAttr out");
}
int
NdbBlob::setHeadInlineValue(NdbOperation* anOp)
{
DBG("setHeadInlineValue");
theHead->length = theLength;
if (theLength < theInlineSize)
memset(theInlineData + theLength, 0, theInlineSize - theLength);
assert(theNullFlag != -1);
const char* aValue = theNullFlag ? 0 : theHeadInlineBuf.data;
if (anOp->setValue(theColumn, aValue, theHeadInlineBuf.size) == -1) {
setErrorCode(anOp);
return -1;
}
theHeadInlineUpdateFlag = false;
return 0;
}
// getValue/setValue
int
NdbBlob::getValue(void* data, Uint32 bytes)
{
DBG("getValue data=" << hex << data << " bytes=" << dec << bytes);
if (theGetFlag || theState != Prepared) {
setErrorCode(ErrState);
return -1;
}
if (! isReadOp() && ! isScanOp()) {
setErrorCode(ErrUsage);
return -1;
}
if (data == NULL && bytes != 0) {
setErrorCode(ErrUsage);
return -1;
}
theGetFlag = true;
theGetBuf = static_cast<char*>(data);
theGetSetBytes = bytes;
return 0;
}
int
NdbBlob::setValue(const void* data, Uint32 bytes)
{
DBG("setValue data=" << hex << data << " bytes=" << dec << bytes);
if (theSetFlag || theState != Prepared) {
setErrorCode(ErrState);
return -1;
}
if (! isInsertOp() && ! isUpdateOp()) {
setErrorCode(ErrUsage);
return -1;
}
if (data == NULL && bytes != 0) {
setErrorCode(ErrUsage);
return -1;
}
theSetFlag = true;
theSetBuf = static_cast<const char*>(data);
theGetSetBytes = bytes;
if (isInsertOp()) {
// write inline part now
if (theSetBuf != 0) {
unsigned n = theGetSetBytes;
if (n > theInlineSize)
n = theInlineSize;
if (writeDataPrivate(0, theSetBuf, n) == -1)
return -1;
} else {
theNullFlag = true;
theLength = 0;
}
if (setHeadInlineValue(theNdbOp) == -1)
return -1;
}
return 0;
}
// misc operations
int
NdbBlob::getNull(bool& isNull)
{
if (theState == Prepared && theSetFlag) {
isNull = (theSetBuf == NULL);
return 0;
}
if (theNullFlag == -1) {
setErrorCode(ErrState);
return -1;
}
isNull = theNullFlag;
return 0;
}
int
NdbBlob::setNull()
{
DBG("setNull");
if (theNullFlag == -1) {
if (theState == Prepared) {
return setValue(0, 0);
}
setErrorCode(ErrState);
return -1;
}
if (theNullFlag)
return 0;
if (deleteParts(0, getPartCount()) == -1)
return -1;
theNullFlag = true;
theLength = 0;
theHeadInlineUpdateFlag = true;
return 0;
}
int
NdbBlob::getLength(Uint64& len)
{
if (theState == Prepared && theSetFlag) {
len = theGetSetBytes;
return 0;
}
if (theNullFlag == -1) {
setErrorCode(ErrState);
return -1;
}
len = theLength;
return 0;
}
int
NdbBlob::truncate(Uint64 length)
{
DBG("truncate kength=" << length);
if (theNullFlag == -1) {
setErrorCode(ErrState);
return -1;
}
if (theLength > length) {
if (length >= theInlineSize) {
Uint32 part1 = getPartNumber(length);
Uint32 part2 = getPartNumber(theLength - 1);
assert(part2 >= part1);
if (deleteParts(part1, part2 - part1) == -1)
return -1;
} else {
if (deleteParts(0, getPartCount()) == -1)
return -1;
}
theLength = length;
theHeadInlineUpdateFlag = true;
}
return 0;
}
int
NdbBlob::getPos(Uint64& pos)
{
if (theNullFlag == -1) {
setErrorCode(ErrState);
return -1;
}
pos = thePos;
return 0;
}
int
NdbBlob::setPos(Uint64 pos)
{
if (theNullFlag == -1) {
setErrorCode(ErrState);
return -1;
}
if (pos > theLength) {
setErrorCode(ErrSeek);
return -1;
}
thePos = pos;
return 0;
}
// read/write
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);
}
int
NdbBlob::readDataPrivate(Uint64 pos, char* buf, Uint32& bytes)
{
DBG("readData pos=" << pos << " bytes=" << bytes);
if (pos > theLength) {
setErrorCode(ErrSeek);
return -1;
}
if (bytes > theLength - pos)
bytes = theLength - pos;
Uint32 len = bytes;
if (len > 0) {
// inline part
if (pos < theInlineSize) {
Uint32 n = theInlineSize - pos;
if (n > len)
n = len;
memcpy(buf, theInlineData + pos, n);
pos += n;
buf += n;
len -= n;
}
}
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);
Uint32 part = (pos - theInlineSize) / thePartSize;
if (readParts(thePartBuf.data, part, 1) == -1)
return -1;
DBG("force execute");
if (theNdbCon->executeNoBlobs(NoCommit) == -1) {
setErrorCode(theNdbOp);
return -1;
}
Uint32 n = thePartSize - off;
if (n > len)
n = len;
memcpy(buf, thePartBuf.data + off, n);
pos += n;
buf += n;
len -= n;
}
}
if (len > 0) {
assert((pos - theInlineSize) % thePartSize == 0);
// complete blocks in the middle
if (len >= thePartSize) {
Uint32 part = (pos - theInlineSize) / thePartSize;
Uint32 count = len / thePartSize;
if (readParts(buf, part, count) == -1)
return -1;
Uint32 n = thePartSize * count;
pos += n;
buf += n;
len -= n;
}
}
if (len > 0) {
// partial last block
DBG("partial last block pos=" << pos << " len=" << len);
assert((pos - theInlineSize) % thePartSize == 0 && len < thePartSize);
Uint32 part = (pos - theInlineSize) / thePartSize;
if (readParts(thePartBuf.data, part, 1) == -1)
return -1;
DBG("force execute");
if (theNdbCon->executeNoBlobs(NoCommit) == -1) {
setErrorCode(theNdbOp);
return -1;
}
memcpy(buf, thePartBuf.data, len);
Uint32 n = len;
pos += n;
buf += n;
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;
assert(thePos <= theLength);
return 0;
}
int
NdbBlob::writeData(Uint64 pos, 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);
}
int
NdbBlob::writeDataPrivate(Uint64 pos, const char* buf, Uint32 bytes)
{
DBG("writeData pos=" << pos << " bytes=" << bytes);
if (pos > theLength) {
setErrorCode(ErrSeek);
return -1;
}
Uint32 len = bytes;
// any write makes blob not NULL
if (theNullFlag) {
theNullFlag = false;
theHeadInlineUpdateFlag = true;
}
if (len > 0) {
// inline part
if (pos < theInlineSize) {
Uint32 n = theInlineSize - pos;
if (n > len)
n = len;
memcpy(theInlineData + pos, buf, n);
theHeadInlineUpdateFlag = true;
pos += n;
buf += n;
len -= n;
}
}
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;
}
Uint32 part = (pos - theInlineSize) / thePartSize;
if (readParts(thePartBuf.data, part, 1) == -1)
return -1;
DBG("force execute");
if (theNdbCon->executeNoBlobs(NoCommit) == -1) {
setErrorCode(theNdbOp);
return -1;
}
Uint32 n = thePartSize - off;
if (n > len) {
memset(thePartBuf.data + off + len, theFillChar, n - len);
n = len;
}
memcpy(thePartBuf.data + off, buf, n);
if (updateParts(thePartBuf.data, part, 1) == -1)
return -1;
pos += n;
buf += n;
len -= n;
}
}
if (len > 0) {
assert((pos - theInlineSize) % thePartSize == 0);
// complete blocks in the middle
if (len >= thePartSize) {
Uint32 part = (pos - theInlineSize) / thePartSize;
Uint32 count = len / thePartSize;
for (unsigned i = 0; i < count; i++) {
if (part + i < getPartCount()) {
if (updateParts(buf, part + i, 1) == -1)
return -1;
} else {
if (insertParts(buf, part + i, 1) == -1)
return -1;
}
Uint32 n = thePartSize;
pos += n;
buf += n;
len -= n;
}
}
}
if (len > 0) {
// partial last block
DBG("partial last block pos=" << pos << " len=" << len);
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;
}
if (readParts(thePartBuf.data, part, 1) == -1)
return -1;
DBG("force execute");
if (theNdbCon->executeNoBlobs(NoCommit) == -1) {
setErrorCode(theNdbOp);
return -1;
}
memcpy(thePartBuf.data, buf, len);
if (updateParts(thePartBuf.data, part, 1) == -1)
return -1;
} else {
memcpy(thePartBuf.data, buf, len);
memset(thePartBuf.data + len, theFillChar, thePartSize - len);
if (part < getPartCount()) {
if (updateParts(thePartBuf.data, part, 1) == -1)
return -1;
} else {
if (insertParts(thePartBuf.data, part, 1) == -1)
return -1;
}
}
Uint32 n = len;
pos += n;
buf += n;
len -= n;
}
assert(len == 0);
if (theLength < pos) {
theLength = pos;
theHeadInlineUpdateFlag = true;
}
DBG("writeData out");
return 0;
}
int
NdbBlob::readParts(char* buf, Uint32 part, Uint32 count)
{
DBG("readParts part=" << part << " count=" << count);
Uint32 n = 0;
while (n < count) {
NdbOperation* tOp = theNdbCon->getNdbOperation(theBlobTableName);
if (tOp == NULL ||
tOp->readTuple() == -1 ||
setPartKeyValue(tOp, part + n) == -1 ||
tOp->getValue((Uint32)3, buf) == NULL) {
setErrorCode(tOp);
return -1;
}
buf += thePartSize;
n++;
}
return 0;
}
int
NdbBlob::insertParts(const char* buf, Uint32 part, Uint32 count)
{
DBG("insertParts part=" << part << " count=" << count);
Uint32 n = 0;
while (n < count) {
NdbOperation* tOp = theNdbCon->getNdbOperation(theBlobTableName);
if (tOp == NULL ||
tOp->insertTuple() == -1 ||
setPartKeyValue(tOp, part + n) == -1 ||
tOp->setValue((Uint32)3, buf) == -1) {
setErrorCode(tOp);
return -1;
}
buf += thePartSize;
n++;
theNewPartFlag = true;
}
return 0;
}
int
NdbBlob::updateParts(const char* buf, Uint32 part, Uint32 count)
{
DBG("updateParts part=" << part << " count=" << count);
Uint32 n = 0;
while (n < count) {
NdbOperation* tOp = theNdbCon->getNdbOperation(theBlobTableName);
if (tOp == NULL ||
tOp->updateTuple() == -1 ||
setPartKeyValue(tOp, part + n) == -1 ||
tOp->setValue((Uint32)3, buf) == -1) {
setErrorCode(tOp);
return -1;
}
buf += thePartSize;
n++;
theNewPartFlag = true;
}
return 0;
}
int
NdbBlob::deleteParts(Uint32 part, Uint32 count)
{
DBG("deleteParts part=" << part << " count=" << count);
Uint32 n = 0;
while (n < count) {
NdbOperation* tOp = theNdbCon->getNdbOperation(theBlobTableName);
if (tOp == NULL ||
tOp->deleteTuple() == -1 ||
setPartKeyValue(tOp, part + n) == -1) {
setErrorCode(tOp);
return -1;
}
n++;
}
return 0;
}
// blob handle maintenance
/*
* Prepare blob handle linked to an operation. Checks blob table.
* Allocates buffers. For key operation fetches key data from signal
* data. For read operation adds read of head+inline.
*/
int
NdbBlob::atPrepare(NdbConnection* aCon, NdbOperation* anOp, const NdbColumnImpl* aColumn)
{
assert(theState == Idle);
// ndb api stuff
theNdb = anOp->theNdb;
theNdbCon = aCon; // for scan, this is the real transaction (m_transConnection)
theNdbOp = anOp;
theTable = anOp->m_currentTable;
theAccessTable = anOp->m_accessTable;
theColumn = aColumn;
DBG("atPrepare");
NdbDictionary::Column::Type partType = NdbDictionary::Column::Undefined;
switch (theColumn->getType()) {
case NdbDictionary::Column::Blob:
partType = NdbDictionary::Column::Binary;
theFillChar = 0x0;
break;
case NdbDictionary::Column::Clob:
partType = NdbDictionary::Column::Char;
theFillChar = 0x20;
break;
default:
setErrorCode(ErrUsage);
return -1;
}
// sizes
theInlineSize = theColumn->getInlineSize();
thePartSize = theColumn->getPartSize();
theStripeSize = theColumn->getStripeSize();
// blob table 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;
}
// buffers
theKeyBuf.alloc(theTable->m_sizeOfKeysInWords << 2);
theAccessKeyBuf.alloc(theAccessTable->m_sizeOfKeysInWords << 2);
theHeadInlineBuf.alloc(sizeof(Head) + theInlineSize);
thePartBuf.alloc(thePartSize);
theHead = (Head*)theHeadInlineBuf.data;
theInlineData = theHeadInlineBuf.data + sizeof(Head);
// handle different operation types
bool supportedOp = false;
if (isKeyOp()) {
if (isTableOp()) {
// get table key
Uint32* data = (Uint32*)theKeyBuf.data;
unsigned size = theTable->m_sizeOfKeysInWords;
if (theNdbOp->getKeyFromTCREQ(data, size) == -1) {
setErrorCode(ErrUsage);
return -1;
}
}
if (isIndexOp()) {
// get index key
Uint32* data = (Uint32*)theAccessKeyBuf.data;
unsigned size = theAccessTable->m_sizeOfKeysInWords;
if (theNdbOp->getKeyFromTCREQ(data, size) == -1) {
setErrorCode(ErrUsage);
return -1;
}
}
if (isReadOp()) {
// add read of head+inline in this op
if (getHeadInlineValue(theNdbOp) == -1)
return -1;
}
if (isInsertOp()) {
// becomes NULL unless set before execute
theNullFlag = true;
theLength = 0;
}
supportedOp = true;
}
if (isScanOp()) {
// add read of head+inline in this op
if (getHeadInlineValue(theNdbOp) == -1)
return -1;
supportedOp = true;
}
if (! supportedOp) {
setErrorCode(ErrUsage);
return -1;
}
setState(Prepared);
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.
*/
int
NdbBlob::preExecute(ExecType anExecType, bool& batch)
{
DBG("preExecute");
if (theState == Invalid)
return -1;
assert(theState == Prepared);
// handle different operation types
assert(isKeyOp());
if (isReadOp()) {
if (theGetFlag && theGetSetBytes > theInlineSize) {
// need blob head before proceeding
batch = true;
}
}
if (isInsertOp()) {
if (theSetFlag && theGetSetBytes > theInlineSize) {
// add ops to write rest of a setValue
assert(theSetBuf != 0);
Uint64 pos = theInlineSize;
const char* buf = theSetBuf + theInlineSize;
Uint32 bytes = theGetSetBytes - theInlineSize;
if (writeDataPrivate(pos, buf, bytes) == -1)
return -1;
if (anExecType == Commit && theHeadInlineUpdateFlag) {
// add an operation to update head+inline
NdbOperation* tOp = theNdbCon->getNdbOperation(theTable);
if (tOp == NULL ||
tOp->updateTuple() == -1 ||
setTableKeyValue(tOp) == -1 ||
setHeadInlineValue(tOp) == -1) {
setErrorCode(ErrAbort);
return -1;
}
}
}
}
if (isTableOp()) {
if (isUpdateOp() || isDeleteOp()) {
// add operation before this one to read head+inline
NdbOperation* tOp = theNdbCon->getNdbOperation(theTable, theNdbOp);
if (tOp == NULL ||
tOp->readTuple() == -1 ||
setTableKeyValue(tOp) == -1 ||
getHeadInlineValue(tOp) == -1) {
setErrorCode(tOp);
return -1;
}
// execute immediately
batch = true;
}
}
if (isIndexOp()) {
// add op before this one to read table key
NdbBlob* tFirstBlob = theNdbOp->theBlobList;
if (this == tFirstBlob) {
// first blob does it for all
if (g_ndb_blob_ok_to_read_index_table) {
Uint32 pkAttrId = theAccessTable->getNoOfColumns() - 1;
NdbOperation* tOp = theNdbCon->getNdbOperation(theAccessTable, theNdbOp);
if (tOp == NULL ||
tOp->readTuple() == -1 ||
setAccessKeyValue(tOp) == -1 ||
tOp->getValue(pkAttrId, theKeyBuf.data) == NULL) {
setErrorCode(tOp);
return -1;
}
} else {
NdbOperation* tOp = theNdbCon->getNdbIndexOperation(theAccessTable->m_index, theTable, theNdbOp);
if (tOp == NULL ||
tOp->readTuple() == -1 ||
setAccessKeyValue(tOp) == -1 ||
getTableKeyValue(tOp) == -1) {
setErrorCode(tOp);
return -1;
}
}
}
if (isUpdateOp() || isDeleteOp()) {
// add op before this one to read head+inline via index
NdbIndexOperation* tOp = theNdbCon->getNdbIndexOperation(theAccessTable->m_index, theTable, theNdbOp);
if (tOp == NULL ||
tOp->readTuple() == -1 ||
setAccessKeyValue(tOp) == -1 ||
getHeadInlineValue(tOp) == -1) {
setErrorCode(tOp);
return -1;
}
// execute immediately
batch = true;
}
}
DBG("preExecute out batch=" << batch);
return 0;
}
/*
* After execute, for any operation. If already Active, this routine
* has been done previously. Operations which requested a no-commit
* batch can add new operations after this one. They are added before
* any remaining prepared operations.
*/
int
NdbBlob::postExecute(ExecType anExecType)
{
DBG("postExecute type=" << anExecType);
if (theState == Invalid)
return -1;
if (theState == Active)
return 0;
assert(theState == Prepared);
assert(isKeyOp());
if (isIndexOp()) {
NdbBlob* tFirstBlob = theNdbOp->theBlobList;
if (this != tFirstBlob) {
// copy key from first blob
assert(theKeyBuf.size == tFirstBlob->theKeyBuf.size);
memcpy(theKeyBuf.data, tFirstBlob->theKeyBuf.data, tFirstBlob->theKeyBuf.size);
}
}
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)
return -1;
}
}
if (isUpdateOp()) {
assert(anExecType == NoCommit);
getHeadFromRecAttr();
if (theSetFlag) {
// setValue overwrites everything
if (theSetBuf != 0) {
if (truncate(0) == -1)
return -1;
if (writeDataPrivate(0, theSetBuf, theGetSetBytes) == -1)
return -1;
} else {
if (setNull() == -1)
return -1;
}
}
}
if (isDeleteOp()) {
assert(anExecType == NoCommit);
getHeadFromRecAttr();
if (deleteParts(0, getPartCount()) == -1)
return -1;
}
theNewPartFlag = false;
setState(anExecType == NoCommit ? Active : Closed);
DBG("postExecute out");
return 0;
}
/*
* Before commit of completed operation. For write add operation to
* update head+inline.
*/
int
NdbBlob::preCommit()
{
DBG("preCommit");
if (theState == Invalid)
return -1;
assert(theState == Active);
assert(isKeyOp());
if (isInsertOp() || isUpdateOp()) {
if (theHeadInlineUpdateFlag) {
// add an operation to update head+inline
NdbOperation* tOp = theNdbCon->getNdbOperation(theTable);
if (tOp == NULL ||
tOp->updateTuple() == -1 ||
setTableKeyValue(tOp) == -1 ||
setHeadInlineValue(tOp) == -1) {
setErrorCode(ErrAbort);
return -1;
}
}
}
DBG("preCommit out");
return 0;
}
/*
* After next scan result. Handle like read op above.
*/
int
NdbBlob::atNextResult()
{
DBG("atNextResult");
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;
if (theNdbOp->getKeyFromKEYINFO20(data, size) == -1) {
setErrorCode(ErrUsage);
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;
// must also execute them
DBG("force execute");
if (theNdbCon->executeNoBlobs(NoCommit) == -1) {
setErrorCode((NdbOperation*)0);
return -1;
}
}
}
setState(Active);
DBG("atNextResult out");
return 0;
}
// misc
const NdbDictionary::Column*
NdbBlob::getColumn()
{
return theColumn;
}
// errors
void
NdbBlob::setErrorCode(int anErrorCode, bool invalidFlag)
{
DBG("setErrorCode code=" << anErrorCode);
theError.code = anErrorCode;
if (invalidFlag)
setState(Invalid);
}
void
NdbBlob::setErrorCode(NdbOperation* anOp, bool invalidFlag)
{
int code = 0;
if (anOp != NULL && (code = anOp->theError.code) != 0)
;
else if ((code = theNdbCon->theError.code) != 0)
;
else if ((code = theNdb->theError.code) != 0)
;
else
code = ErrUnknown;
setErrorCode(code, invalidFlag);
}
void
NdbBlob::setErrorCode(NdbConnection* aCon, bool invalidFlag)
{
int code = 0;
if (theNdbCon != NULL && (code = theNdbCon->theError.code) != 0)
;
else if ((code = theNdb->theError.code) != 0)
;
else
code = ErrUnknown;
setErrorCode(code, invalidFlag);
}
#ifdef VM_TRACE
NdbOut&
operator<<(NdbOut& out, const NdbBlob& blob)
{
ndbout << dec << "s=" << blob.theState;
ndbout << dec << " n=" << blob.theNullFlag;;
ndbout << dec << " l=" << blob.theLength;
ndbout << dec << " p=" << blob.thePos;
ndbout << dec << " u=" << blob.theHeadInlineUpdateFlag;
return out;
}
#endif
ndb/src/ndbapi/NdbConnection.cpp
View file @ 9a862183
......@@ -35,6 +35,7 @@ Adjust: 971022 UABMNST First version.
#include "NdbApiSignal.hpp"
#include "TransporterFacade.hpp"
#include "API.hpp"
#include "NdbBlob.hpp"
#include <ndb_limits.h>
#include <signaldata/TcKeyConf.hpp>
......@@ -89,7 +90,8 @@ NdbConnection::NdbConnection( Ndb* aNdb ) :
theCurrentScanRec(NULL),
thePreviousScanRec(NULL),
theScanningOp(NULL),
theBuddyConPtr(0xFFFFFFFF)
theBuddyConPtr(0xFFFFFFFF),
theBlobFlag(false)
{
theListState = NotInList;
theError.code = 0;
......@@ -152,6 +154,8 @@ NdbConnection::init()
m_theLastCursorOperation = NULL;
m_firstExecutedCursorOp = 0;
theBuddyConPtr = 0xFFFFFFFF;
//
theBlobFlag = false;
}//NdbConnection::init()
/*****************************************************************************
......@@ -250,6 +254,86 @@ int
NdbConnection::execute(ExecType aTypeOfExec,
AbortOption abortOption,
int forceSend)
{
if (! theBlobFlag)
return executeNoBlobs(aTypeOfExec, abortOption, forceSend);
// execute prepared ops in batches, as requested by blobs
ExecType tExecType;
NdbOperation* tPrepOp;
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;
}
tPrepOp = tPrepOp->next();
}
// save rest of prepared ops if batch
NdbOperation* tRestOp;
NdbOperation* tLastOp;
if (tPrepOp != NULL) {
tRestOp = tPrepOp->next();
tPrepOp->next(NULL);
tLastOp = theLastOpInList;
theLastOpInList = tPrepOp;
}
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;
}
tOp = tOp->next();
}
}
if (executeNoBlobs(tExecType, abortOption, forceSend) == -1)
return -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;
}
tOp = tOp->next();
}
}
// add saved prepared ops if batch
if (tPrepOp != NULL && tRestOp != NULL) {
if (theFirstOpInList == NULL)
theFirstOpInList = tRestOp;
else
theLastOpInList->next(tRestOp);
theLastOpInList = tLastOp;
}
} while (theFirstOpInList != NULL || tExecType != aTypeOfExec);
return 0;
}
int
NdbConnection::executeNoBlobs(ExecType aTypeOfExec,
AbortOption abortOption,
int forceSend)
{
//------------------------------------------------------------------------
// We will start by preparing all operations in the transaction defined
......@@ -330,7 +414,6 @@ NdbConnection::executeAsynchPrepare( ExecType aTypeOfExec,
* Reset error.code on execute
*/
theError.code = 0;
NdbCursorOperation* tcOp = m_theFirstCursorOperation;
if (tcOp != 0){
// Execute any cursor operations
......@@ -885,7 +968,7 @@ Remark: Get an operation from NdbOperation object idlelist and
object, synchronous.
*****************************************************************************/
NdbOperation*
NdbConnection::getNdbOperation(NdbTableImpl * tab)
NdbConnection::getNdbOperation(NdbTableImpl * tab, NdbOperation* aNextOp)
{
NdbOperation* tOp;
......@@ -897,14 +980,28 @@ NdbConnection::getNdbOperation(NdbTableImpl * tab)
tOp = theNdb->getOperation();
if (tOp == NULL)
goto getNdbOp_error1;
if (theLastOpInList != NULL) {
theLastOpInList->next(tOp);
theLastOpInList = tOp;
if (aNextOp == NULL) {
if (theLastOpInList != NULL) {
theLastOpInList->next(tOp);
theLastOpInList = tOp;
} else {
theLastOpInList = tOp;
theFirstOpInList = tOp;
}//if
tOp->next(NULL);
} else {
theLastOpInList = tOp;
theFirstOpInList = tOp;
}//if
tOp->next(NULL);
// add before the given op
if (theFirstOpInList == aNextOp) {
theFirstOpInList = tOp;
} else {
NdbOperation* aLoopOp = theFirstOpInList;
while (aLoopOp != NULL && aLoopOp->next() != aNextOp)
aLoopOp = aLoopOp->next();
assert(aLoopOp != NULL);
aLoopOp->next(tOp);
}
tOp->next(aNextOp);
}
if (tOp->init(tab, this) != -1) {
return tOp;
} else {
......@@ -1068,21 +1165,36 @@ Remark: Get an operation from NdbIndexOperation object idlelist and get
*****************************************************************************/
NdbIndexOperation*
NdbConnection::getNdbIndexOperation(NdbIndexImpl * anIndex,
NdbTableImpl * aTable)
NdbTableImpl * aTable,
NdbOperation* aNextOp)
{
NdbIndexOperation* tOp;
tOp = theNdb->getIndexOperation();
if (tOp == NULL)
goto getNdbOp_error1;
if (theLastOpInList != NULL) {
theLastOpInList->next(tOp);
theLastOpInList = tOp;
if (aNextOp == NULL) {
if (theLastOpInList != NULL) {
theLastOpInList->next(tOp);
theLastOpInList = tOp;
} else {
theLastOpInList = tOp;
theFirstOpInList = tOp;
}//if
tOp->next(NULL);
} else {
theLastOpInList = tOp;
theFirstOpInList = tOp;
}//if
tOp->next(NULL);
// add before the given op
if (theFirstOpInList == aNextOp) {
theFirstOpInList = tOp;
} else {
NdbOperation* aLoopOp = theFirstOpInList;
while (aLoopOp != NULL && aLoopOp->next() != aNextOp)
aLoopOp = aLoopOp->next();
assert(aLoopOp != NULL);
aLoopOp->next(tOp);
}
tOp->next(aNextOp);
}
if (tOp->indxInit(anIndex, aTable, this)!= -1) {
return tOp;
} else {
......
ndb/src/ndbapi/NdbDictionary.cpp
View file @ 9a862183
......@@ -273,6 +273,9 @@ NdbDictionary::Table::addColumn(const Column & c){
if(c.getPrimaryKey()){
m_impl.m_noOfKeys++;
}
if (col->getBlobType()) {
m_impl.m_noOfBlobs++;
}
m_impl.buildColumnHash();
}
......
ndb/src/ndbapi/NdbDictionaryImpl.cpp
View file @ 9a862183
......@@ -34,6 +34,7 @@
#include <AttributeList.hpp>
#include <NdbEventOperation.hpp>
#include "NdbEventOperationImpl.hpp"
#include "NdbBlob.hpp"
#define DEBUG_PRINT 0
#define INCOMPATIBLE_VERSION -2
......@@ -178,7 +179,14 @@ NdbColumnImpl::equal(const NdbColumnImpl& col) const
case NdbDictionary::Column::Double:
case NdbDictionary::Column::Datetime:
case NdbDictionary::Column::Timespec:
break;
case NdbDictionary::Column::Blob:
case NdbDictionary::Column::Clob:
if (m_precision != col.m_precision ||
m_scale != col.m_scale ||
m_length != col.m_length) {
return false;
}
break;
}
if (m_autoIncrement != col.m_autoIncrement){
......@@ -223,6 +231,8 @@ NdbTableImpl::NdbTableImpl()
: NdbDictionary::Table(* this), m_facade(this)
{
m_noOfKeys = 0;
m_sizeOfKeysInWords = 0;
m_noOfBlobs = 0;
m_index = 0;
init();
}
......@@ -257,6 +267,8 @@ NdbTableImpl::init(){
m_indexType = NdbDictionary::Index::Undefined;
m_noOfKeys = 0;
m_sizeOfKeysInWords = 0;
m_noOfBlobs = 0;
}
bool
......@@ -336,6 +348,8 @@ NdbTableImpl::assign(const NdbTableImpl& org)
m_index = org.m_index;
m_noOfKeys = org.m_noOfKeys;
m_sizeOfKeysInWords = org.m_sizeOfKeysInWords;
m_noOfBlobs = org.m_noOfBlobs;
m_version = org.m_version;
m_status = org.m_status;
......@@ -1076,6 +1090,8 @@ columnTypeMapping[] = {
{ DictTabInfo::ExtVarbinary, NdbDictionary::Column::Varbinary },
{ DictTabInfo::ExtDatetime, NdbDictionary::Column::Datetime },
{ DictTabInfo::ExtTimespec, NdbDictionary::Column::Timespec },
{ DictTabInfo::ExtBlob, NdbDictionary::Column::Blob },
{ DictTabInfo::ExtClob, NdbDictionary::Column::Clob },
{ -1, -1 }
};
......@@ -1131,6 +1147,7 @@ NdbDictInterface::parseTableInfo(NdbTableImpl ** ret,
Uint32 keyInfoPos = 0;
Uint32 keyCount = 0;
Uint32 blobCount;
for(Uint32 i = 0; i < tableDesc.NoOfAttributes; i++) {
DictTabInfo::Attribute attrDesc; attrDesc.init();
......@@ -1187,6 +1204,8 @@ NdbDictInterface::parseTableInfo(NdbTableImpl ** ret,
} else {
col->m_keyInfoPos = 0;
}
if (col->getBlobType())
blobCount++;
NdbColumnImpl * null = 0;
impl->m_columns.fill(attrDesc.AttributeId, null);
......@@ -1199,6 +1218,8 @@ NdbDictInterface::parseTableInfo(NdbTableImpl ** ret,
it.next();
}
impl->m_noOfKeys = keyCount;
impl->m_sizeOfKeysInWords = keyInfoPos;
impl->m_noOfBlobs = blobCount;
* ret = impl;
return 0;
}
......@@ -1206,6 +1227,43 @@ NdbDictInterface::parseTableInfo(NdbTableImpl ** ret,
/*****************************************************************
* Create table and alter table
*/
int
NdbDictionaryImpl::createTable(NdbTableImpl &t)
{
if (m_receiver.createTable(m_ndb, t) != 0)
return -1;
if (t.m_noOfBlobs == 0)
return 0;
// update table def from DICT
NdbTableImpl * tp = getTable(t.m_externalName.c_str());
if (tp == NULL) {
m_error.code = 709;
return -1;
}
if (createBlobTables(* tp) != 0) {
int save_code = m_error.code;
(void)dropTable(t);
m_error.code = save_code;
return -1;
}
return 0;
}
int
NdbDictionaryImpl::createBlobTables(NdbTableImpl &t)
{
for (unsigned i = 0; i < t.m_columns.size(); i++) {
NdbColumnImpl & c = *t.m_columns[i];
if (! c.getBlobType())
continue;
NdbTableImpl bt;
NdbBlob::getBlobTable(bt, &t, &c);
if (createTable(bt) != 0)
return -1;
}
return 0;
}
int
NdbDictInterface::createTable(Ndb & ndb,
NdbTableImpl & impl)
......@@ -1540,6 +1598,12 @@ NdbDictionaryImpl::dropTable(NdbTableImpl & impl)
if (dropIndex(element.name, name) == -1)
return -1;
}
if (impl.m_noOfBlobs != 0) {
if (dropBlobTables(impl) != 0)
return -1;
}
int ret = m_receiver.dropTable(impl);
if(ret == 0){
const char * internalTableName = impl.m_internalName.c_str();
......@@ -1554,6 +1618,23 @@ NdbDictionaryImpl::dropTable(NdbTableImpl & impl)
return ret;
}
int
NdbDictionaryImpl::dropBlobTables(NdbTableImpl & t)
{
for (unsigned i = 0; i < t.m_columns.size(); i++) {
NdbColumnImpl & c = *t.m_columns[i];
if (! c.getBlobType())
continue;
char btname[NdbBlob::BlobTableNameSize];
NdbBlob::getBlobTableName(btname, &t, &c);
if (dropTable(btname) != 0) {
if (m_error.code != 709)
return -1;
}
}
return 0;
}
int
NdbDictInterface::dropTable(const NdbTableImpl & impl)
{
......
ndb/src/ndbapi/NdbDictionaryImpl.hpp
View file @ 9a862183
......@@ -81,6 +81,7 @@ public:
Uint32 m_keyInfoPos;
Uint32 m_extType; // used by restore (kernel type in versin v2x)
bool getInterpretableType() const ;
bool getBlobType() const;
/**
* Equality/assign
......@@ -141,6 +142,8 @@ public:
* Aggregates
*/
Uint32 m_noOfKeys;
unsigned short m_sizeOfKeysInWords;
unsigned short m_noOfBlobs;
/**
* Equality/assign
......@@ -353,13 +356,12 @@ public:
bool setTransporter(class Ndb * ndb, class TransporterFacade * tf);
bool setTransporter(class TransporterFacade * tf);
int createTable(NdbTableImpl &t)
{
return m_receiver.createTable(m_ndb, t);
}
int createTable(NdbTableImpl &t);
int createBlobTables(NdbTableImpl &);
int alterTable(NdbTableImpl &t);
int dropTable(const char * name);
int dropTable(NdbTableImpl &);
int dropBlobTables(NdbTableImpl &);
int invalidateObject(NdbTableImpl &);
int removeCachedObject(NdbTableImpl &);
......@@ -432,6 +434,13 @@ NdbColumnImpl::getInterpretableType() const {
m_type == NdbDictionary::Column::Bigunsigned);
}
inline
bool
NdbColumnImpl::getBlobType() const {
return (m_type == NdbDictionary::Column::Blob ||
m_type == NdbDictionary::Column::Clob);
}
inline
NdbTableImpl &
NdbTableImpl::getImpl(NdbDictionary::Table & t){
......
ndb/src/ndbapi/NdbIndexOperation.cpp
View file @ 9a862183
......@@ -372,6 +372,17 @@ int NdbIndexOperation::equal_impl(const NdbColumnImpl* tAttrInfo,
} else if ((tOpType == ReadRequest) || (tOpType == DeleteRequest) ||
(tOpType == ReadExclusive)) {
theStatus = GetValue;
// create blob handles automatically
if (tOpType == DeleteRequest && m_currentTable->m_noOfBlobs != 0) {
for (unsigned i = 0; i < m_currentTable->m_columns.size(); i++) {
NdbColumnImpl* c = m_currentTable->m_columns[i];
assert(c != 0);
if (c->getBlobType()) {
if (getBlobHandle(theNdbCon, c) == NULL)
return -1;
}
}
}
return 0;
} else if ((tOpType == InsertRequest) || (tOpType == WriteRequest)) {
theStatus = SetValue;
......
ndb/src/ndbapi/NdbOperation.cpp
View file @ 9a862183
......@@ -31,6 +31,7 @@
#include "NdbApiSignal.hpp"
#include "NdbRecAttr.hpp"
#include "NdbUtil.hpp"
#include "NdbBlob.hpp"
#include "ndbapi_limits.h"
#include <signaldata/TcKeyReq.hpp>
#include "NdbDictionaryImpl.hpp"
......@@ -103,7 +104,8 @@ NdbOperation::NdbOperation(Ndb* aNdb) :
theFirstSCAN_TABINFO_Recv(NULL),
theLastSCAN_TABINFO_Recv(NULL),
theSCAN_TABCONF_Recv(NULL),
theBoundATTRINFO(NULL)
theBoundATTRINFO(NULL),
theBlobList(NULL)
{
theReceiver.init(NdbReceiver::NDB_OPERATION, this);
theError.code = 0;
......@@ -197,6 +199,7 @@ NdbOperation::init(NdbTableImpl* tab, NdbConnection* myConnection){
theTotalNrOfKeyWordInSignal = 8;
theMagicNumber = 0xABCDEF01;
theBoundATTRINFO = NULL;
theBlobList = NULL;
tSignal = theNdb->getSignal();
if (tSignal == NULL)
......@@ -236,6 +239,8 @@ NdbOperation::release()
NdbCall* tSaveCall;
NdbSubroutine* tSubroutine;
NdbSubroutine* tSaveSubroutine;
NdbBlob* tBlob;
NdbBlob* tSaveBlob;
if (theTCREQ != NULL)
{
......@@ -308,6 +313,14 @@ NdbOperation::release()
}
theBoundATTRINFO = NULL;
}
tBlob = theBlobList;
while (tBlob != NULL)
{
tSaveBlob = tBlob;
tBlob = tBlob->theNext;
theNdb->releaseNdbBlob(tSaveBlob);
}
theBlobList = NULL;
releaseScan();
}
......@@ -356,6 +369,18 @@ NdbOperation::setValue( Uint32 anAttrId,
return setValue(m_currentTable->getColumn(anAttrId), aValuePassed, len);
}
NdbBlob*
NdbOperation::getBlobHandle(const char* anAttrName)
{
return getBlobHandle(theNdbCon, m_currentTable->getColumn(anAttrName));
}
NdbBlob*
NdbOperation::getBlobHandle(Uint32 anAttrId)
{
return getBlobHandle(theNdbCon, m_currentTable->getColumn(anAttrId));
}
int
NdbOperation::incValue(const char* anAttrName, Uint32 aValue)
{
......@@ -428,4 +453,8 @@ NdbOperation::setBound(Uint32 anAttrId, int type, const void* aValue, Uint32 len
return setBound(m_accessTable->getColumn(anAttrId), type, aValue, len);
}
const char*
NdbOperation::getTableName() const
{
return m_currentTable->m_externalName.c_str();
}
ndb/src/ndbapi/NdbOperationDefine.cpp
View file @ 9a862183
......@@ -34,6 +34,7 @@
#include "NdbUtil.hpp"
#include "NdbOut.hpp"
#include "NdbImpl.hpp"
#include "NdbBlob.hpp"
#include <Interpreter.hpp>
......@@ -604,6 +605,33 @@ NdbOperation::setValue( const NdbColumnImpl* tAttrInfo,
return 0;
}//NdbOperation::setValue()
NdbBlob*
NdbOperation::getBlobHandle(NdbConnection* aCon, const NdbColumnImpl* tAttrInfo)
{
NdbBlob* tBlob = theBlobList;
NdbBlob* tLastBlob = NULL;
while (tBlob != NULL) {
if (tBlob->theColumn == tAttrInfo)
return tBlob;
tLastBlob = tBlob;
tBlob = tBlob->theNext;
}
tBlob = theNdb->getNdbBlob();
if (tBlob == NULL)
return NULL;
if (tBlob->atPrepare(aCon, this, tAttrInfo) == -1) {
theNdb->releaseNdbBlob(tBlob);
return NULL;
}
if (tLastBlob == NULL)
theBlobList = tBlob;
else
tLastBlob->theNext = tBlob;
tBlob->theNext = NULL;
theNdbCon->theBlobFlag = true;
return tBlob;
}
/*
* Define bound on index column in range scan.
*/
......
ndb/src/ndbapi/NdbOperationScan.cpp
View file @ 9a862183
......@@ -569,8 +569,35 @@ NdbOperation::takeOverScanOp(OperationType opType, NdbConnection* updateTrans)
}
}
// create blob handles automatically
if (opType == DeleteRequest && m_currentTable->m_noOfBlobs != 0) {
for (unsigned i = 0; i < m_currentTable->m_columns.size(); i++) {
NdbColumnImpl* c = m_currentTable->m_columns[i];
assert(c != 0);
if (c->getBlobType()) {
if (newOp->getBlobHandle(updateTrans, c) == NULL)
return NULL;
}
}
}
return newOp;
}
int
NdbOperation::getKeyFromKEYINFO20(Uint32* data, unsigned size)
{
const NdbScanReceiver* tScanRec = theNdbCon->thePreviousScanRec;
NdbApiSignal* tSignal = tScanRec->theFirstKEYINFO20_Recv;
unsigned pos = 0;
unsigned n = 0;
while (pos < size) {
if (n == 20) {
tSignal = tSignal->next();
n = 0;
}
const unsigned h = KeyInfo20::HeaderLength;
data[pos++] = tSignal->getDataPtrSend()[h + n++];
}
return 0;
}
ndb/src/ndbapi/NdbOperationSearch.cpp
View file @ 9a862183
......@@ -251,6 +251,17 @@ NdbOperation::equal_impl(const NdbColumnImpl* tAttrInfo,
} else if ((tOpType == ReadRequest) || (tOpType == DeleteRequest) ||
(tOpType == ReadExclusive)) {
theStatus = GetValue;
// create blob handles automatically
if (tOpType == DeleteRequest && m_currentTable->m_noOfBlobs != 0) {
for (unsigned i = 0; i < m_currentTable->m_columns.size(); i++) {
NdbColumnImpl* c = m_currentTable->m_columns[i];
assert(c != 0);
if (c->getBlobType()) {
if (getBlobHandle(theNdbCon, c) == NULL)
return -1;
}
}
}
return 0;
} else if ((tOpType == InsertRequest) || (tOpType == WriteRequest)) {
theStatus = SetValue;
......@@ -497,3 +508,24 @@ LastWordLabel:
return 0;
}
int
NdbOperation::getKeyFromTCREQ(Uint32* data, unsigned size)
{
assert(m_accessTable != 0 && m_accessTable->m_sizeOfKeysInWords != 0);
assert(m_accessTable->m_sizeOfKeysInWords == size);
unsigned pos = 0;
while (pos < 8 && pos < size) {
data[pos++] = theKEYINFOptr[pos];
}
NdbApiSignal* tSignal = theFirstKEYINFO;
unsigned n = 0;
while (pos < size) {
if (n == 20) {
tSignal = tSignal->next();
n = 0;
}
data[pos++] = tSignal->getDataPtrSend()[3 + n++];
}
return 0;
}
ndb/src/ndbapi/NdbScanOperation.cpp
View file @ 9a862183
......@@ -34,6 +34,7 @@
#include "NdbApiSignal.hpp"
#include <NdbOut.hpp>
#include "NdbDictionaryImpl.hpp"
#include "NdbBlob.hpp"
NdbScanOperation::NdbScanOperation(Ndb* aNdb) :
NdbCursorOperation(aNdb),
......@@ -294,6 +295,18 @@ int NdbScanOperation::setValue(Uint32 anAttrId, double aValue)
return 0;
}
NdbBlob*
NdbScanOperation::getBlobHandle(const char* anAttrName)
{
return NdbOperation::getBlobHandle(m_transConnection, m_currentTable->getColumn(anAttrName));
}
NdbBlob*
NdbScanOperation::getBlobHandle(Uint32 anAttrId)
{
return NdbOperation::getBlobHandle(m_transConnection, m_currentTable->getColumn(anAttrId));
}
// Private methods
int NdbScanOperation::executeCursor(int ProcessorId)
......@@ -344,6 +357,15 @@ int NdbScanOperation::nextResult(bool fetchAllowed)
const NdbError err = theNdbCon->getNdbError();
m_transConnection->setOperationErrorCode(err.code);
}
if (result == 0) {
// handle blobs
NdbBlob* tBlob = theBlobList;
while (tBlob != NULL) {
if (tBlob->atNextResult() == -1)
return -1;
tBlob = tBlob->theNext;
}
}
return result;
}
......
ndb/src/ndbapi/Ndberr.cpp
View file @ 9a862183
......@@ -20,6 +20,7 @@
#include "NdbDictionaryImpl.hpp"
#include <NdbOperation.hpp>
#include <NdbConnection.hpp>
#include <NdbBlob.hpp>
static void
......@@ -65,3 +66,10 @@ NdbOperation::getNdbError() const {
update(theError);
return theError;
}
const
NdbError &
NdbBlob::getNdbError() const {
update(theError);
return theError;
}
ndb/src/ndbapi/Ndbinit.cpp
View file @ 9a862183
......@@ -85,6 +85,7 @@ Ndb::Ndb( const char* aDataBase , const char* aDataBaseSchema) :
theSubroutineList(NULL),
theCallList(NULL),
theScanList(NULL),
theNdbBlobIdleList(NULL),
theNoOfDBnodes(0),
theDBnodes(NULL),
the_release_ind(NULL),
......@@ -235,6 +236,8 @@ Ndb::~Ndb()
freeNdbCall();
while (theScanList != NULL)
freeNdbScanRec();
while (theNdbBlobIdleList != NULL)
freeNdbBlob();
releaseTransactionArrays();
startTransactionNodeSelectionData.release();
......
ndb/src/ndbapi/Ndblist.cpp
View file @ 9a862183
......@@ -27,6 +27,7 @@
#include "NdbScanReceiver.hpp"
#include "NdbUtil.hpp"
#include "API.hpp"
#include "NdbBlob.hpp"
void
Ndb::checkFailedNode()
......@@ -435,6 +436,19 @@ Ndb::getSignal()
return tSignal;
}
NdbBlob*
Ndb::getNdbBlob()
{
NdbBlob* tBlob = theNdbBlobIdleList;
if (tBlob != NULL) {
theNdbBlobIdleList = tBlob->theNext;
tBlob->init();
} else {
tBlob = new NdbBlob;
}
return tBlob;
}
/***************************************************************************
void releaseNdbBranch(NdbBranch* aNdbBranch);
......@@ -601,6 +615,14 @@ Ndb::releaseSignalsInList(NdbApiSignal** pList){
}
}
void
Ndb::releaseNdbBlob(NdbBlob* aBlob)
{
aBlob->release();
aBlob->theNext = theNdbBlobIdleList;
theNdbBlobIdleList = aBlob;
}
/***************************************************************************
void freeOperation();
......@@ -745,6 +767,14 @@ Ndb::freeSignal()
cfreeSignals++;
}
void
Ndb::freeNdbBlob()
{
NdbBlob* tBlob = theNdbBlobIdleList;
theNdbBlobIdleList = tBlob->theNext;
delete tBlob;
}
/****************************************************************************
int releaseConnectToNdb(NdbConnection* aConnectConnection);
......
ndb/src/ndbapi/ndberror.c
View file @ 9a862183
......@@ -418,8 +418,14 @@ ErrorBundle ErrorCodes[] = {
{ 4259, AE, "Invalid set of range scan bounds" },
{ 4260, UD, "NdbScanFilter: Operator is not defined in NdbScanFilter::Group"},
{ 4261, UD, "NdbScanFilter: Column is NULL"},
{ 4262, UD, "NdbScanFilter: Condition is out of bounds"}
{ 4262, UD, "NdbScanFilter: Condition is out of bounds"},
{ 4263, IE, "Invalid blob attributes or invalid blob parts table" },
{ 4264, AE, "Invalid usage of blob attribute" },
{ 4265, AE, "Method is not valid in current blob state" },
{ 4266, AE, "Invalid blob seek position" },
{ 4267, IE, "Corrupted blob value" },
{ 4268, IE, "Error in blob head update forced rollback of transaction" },
{ 4268, IE, "Unknown blob error" }
};
static
......
ndb/src/old_files/client/odbc/codegen/SimpleGram.ypp
View file @ 9a862183
......@@ -88,6 +88,7 @@ struct PhysAttr { int storage; int logging; };
struct PhysAttr* m_phys_attr;
NdbDictionary::Object::FragmentType m_storage_attr;
bool m_logging_attr;
SqlType::Type m_sql_type;
}
/* keywords */
......@@ -100,6 +101,7 @@ struct PhysAttr { int storage; int logging; };
T_BLOB
T_BY
T_CHAR
T_CLOB
T_CONSTRAINT
T_CREATE
T_DATETIME
......@@ -128,6 +130,7 @@ struct PhysAttr { int storage; int logging; };
T_LIMIT
T_LOGGING
T_LONGBLOB
T_LONGCLOB
T_MEDIUM
T_NOLOGGING
T_NOT
......@@ -248,6 +251,7 @@ struct PhysAttr { int storage; int logging; };
%type <m_phys_attr> phys_attr2
%type <m_storage_attr> storage_attr
%type <m_logging_attr> logging_attr
%type <m_sql_type> blob_type
%%
......@@ -606,10 +610,10 @@ data_type:
$$ = dataType;
}
|
blob_keyword
blob_type
{
Plan_root* root = simpleParser.root();
SqlType sqlType(SqlType::Blob, true);
SqlType sqlType($1, true);
Plan_data_type* dataType = new Plan_data_type(root, sqlType);
root->saveNode(dataType);
$$ = dataType;
......@@ -620,10 +624,26 @@ dummy_binary:
|
T_BINARY
;
blob_keyword:
blob_type:
T_BLOB
{
$$ = SqlType::Blob;
}
|
T_LONGBLOB
{
$$ = SqlType::Blob;
}
|
T_CLOB
{
$$ = SqlType::Clob;
}
|
T_LONGCLOB
{
$$ = SqlType::Clob;
}
;
create_column_rest:
/* empty */
......
ndb/src/old_files/client/odbc/codegen/SimpleScan.lpp
View file @ 9a862183
......@@ -149,6 +149,7 @@ static const SqlKeyword sqlKeyword[] = {
{ "BLOB", T_BLOB, StateType },
{ "BY", T_BY, -1 },
{ "CHAR", T_CHAR, StateType },
{ "CLOB", T_CLOB, StateType },
{ "CONSTRAINT", T_CONSTRAINT, -1 },
{ "CREATE", T_CREATE, -1 },
{ "DATETIME", T_DATETIME, StateType },
......@@ -177,6 +178,7 @@ static const SqlKeyword sqlKeyword[] = {
{ "LIMIT", T_LIMIT, -1 },
{ "LOGGING", T_LOGGING, StatePhys },
{ "LONGBLOB", T_LONGBLOB, StateType },
{ "LONGCLOB", T_LONGCLOB, StateType },
{ "MEDIUM", T_MEDIUM, StatePhys },
{ "NOLOGGING", T_NOLOGGING, StatePhys },
{ "NOT", T_NOT, -1 },
......
ndb/src/old_files/client/odbc/common/DataType.cpp
View file @ 9a862183
......@@ -78,6 +78,9 @@ SqlType::setType(Ctx& ctx, Type type, bool nullable)
case Blob:
setType(ctx, Varbinary, FAKE_BLOB_SIZE, nullable); // XXX BLOB hack
return;
case Clob:
setType(ctx, Varchar, FAKE_BLOB_SIZE, nullable); // XXX BLOB hack
return;
case Null:
case Unbound:
break;
......@@ -193,6 +196,9 @@ SqlType::setType(Ctx& ctx, const NdbDictionary::Column* ndbColumn)
case NdbDictionary::Column::Blob:
setType(ctx, Blob, nullable);
return;
case NdbDictionary::Column::Clob:
setType(ctx, Clob, nullable);
return;
default:
break;
}
......
ndb/src/old_files/client/odbc/common/DataType.hpp
View file @ 9a862183
......@@ -74,6 +74,7 @@ public:
Date = SQL_DATE,
Datetime = SQL_TYPE_TIMESTAMP,
Blob = SQL_BLOB,
Clob = SQL_CLOB,
Null = NullDataType, // not an ODBC SQL type
Unbound = UnboundDataType // special for placeholders
};
......
ndb/test/ndbapi/Makefile.am
View file @ 9a862183
......@@ -14,6 +14,7 @@ flexTT \
testBackup \
testBasic \
testBasicAsynch \
testBlobs \
testDataBuffers \
testDict \
testIndex \
......@@ -47,7 +48,7 @@ flexTT_SOURCES = flexTT.cpp
testBackup_SOURCES = testBackup.cpp
testBasic_SOURCES = testBasic.cpp
testBasicAsynch_SOURCES = testBasicAsynch.cpp
#testBlobs_SOURCES = testBlobs.cpp
testBlobs_SOURCES = testBlobs.cpp
testDataBuffers_SOURCES = testDataBuffers.cpp
testDict_SOURCES = testDict.cpp
testIndex_SOURCES = testIndex.cpp
......@@ -77,3 +78,4 @@ testBackup_LDADD = $(LDADD) bank/libbank.a
# Don't update the files from bitkeeper
%::SCCS/s.%
ndb/test/ndbapi/Makefile_old
View file @ 9a862183
......@@ -37,7 +37,8 @@ BIN_DIRS = \
indexTest \
test_event \
indexTest2 \
testGrep
testGrep \
testBlobs
ifeq ($(NDB_OS), SOLARIS)
ifeq ($(NDB_COMPILER), FORTE6)
......
ndb/test/ndbapi/testBlobs.cpp
View file @ 9a862183
......@@ -19,176 +19,1193 @@
*/
#include <ndb_global.h>
#include <NdbMain.h>
#include <NdbOut.hpp>
#include <NdbThread.h>
#include <NdbMutex.h>
#include <NdbCondition.h>
#include <NdbTest.hpp>
#include <NdbTick.h>
struct Bcol {
bool m_nullable;
unsigned m_inline;
unsigned m_partsize;
unsigned m_stripe;
char m_btname[NdbBlob::BlobTableNameSize];
Bcol(bool a, unsigned b, unsigned c, unsigned d) :
m_nullable(a),
m_inline(b),
m_partsize(c),
m_stripe(d)
{}
};
struct Opt {
bool m_core;
const char* m_table;
Opt() :
m_core(false),
m_table("TB1")
{
}
bool m_core;
bool m_dbg;
bool m_dbgall;
bool m_full;
unsigned m_loop;
unsigned m_parts;
unsigned m_rows;
unsigned m_seed;
char m_skip[255];
// metadata
const char* m_tname;
const char* m_x1name; // hash index
const char* m_x2name; // ordered index
unsigned m_pk1off;
unsigned m_pk2len;
bool m_oneblob;
Bcol m_blob1;
Bcol m_blob2;
// bugs
int m_bug;
int (*m_bugtest)();
Opt() :
m_core(false),
m_dbg(false),
m_dbgall(false),
m_full(false),
m_loop(1),
m_parts(10),
m_rows(100),
m_seed(0),
// metadata
m_tname("TBLOB1"),
m_x1name("TBLOB1X1"),
m_x2name("TBLOB1X2"),
m_pk1off(999000000),
m_pk2len(55),
m_oneblob(false),
m_blob1(false, 7, 1137, 10),
m_blob2(true, 99, 55, 1),
// bugs
m_bug(0),
m_bugtest(0) {
memset(m_skip, false, sizeof(m_skip));
}
};
static Opt opt;
static const unsigned g_max_pk2len = 256;
static void printusage()
static void
printusage()
{
Opt d;
ndbout
<< "usage: testBlobs [options]" << endl
<< "-core dump core on error - default " << d.m_core << endl
;
Opt d;
ndbout
<< "usage: testBlobs options [default/max]" << endl
<< " -core dump core on error" << endl
<< " -dbg print debug" << endl
<< " -dbgall print also NDB API debug (if compiled in)" << endl
<< " -full read/write only full blob values" << endl
<< " -inline read/write only blobs which fit inline" << endl
<< " -loop N loop N times 0=forever [" << d.m_loop << "]" << endl
<< " -parts N max parts in blob value [" << d.m_parts << "]" << endl
<< " -rows N number of rows [" << d.m_rows << "]" << endl
<< " -seed N random seed 0=loop number [" << d.m_seed << "]" << endl
<< " -skip xxx skip these tests (see list)" << endl
<< "metadata" << endl
<< " -pk2len N length of PK2 [" << d.m_pk2len << "/" << g_max_pk2len <<"]" << endl
<< " -oneblob only 1 blob attribute [default 2]" << endl
<< "testcases for -skip" << endl
<< " k primary key ops" << endl
<< " i hash index ops" << endl
<< " s table scans" << endl
<< " r ordered index scans" << endl
<< " u update blob value" << endl
<< " v getValue / setValue" << endl
<< " w readData / writeData" << endl
<< "bug tests (no blob test)" << endl
<< " -bug 4088 ndb api hang with mixed ops on index table" << endl
<< " -bug 2222 delete + write gives 626" << endl
<< " -bug 3333 acc crash on delete and long key" << endl
;
}
static Ndb* myNdb = 0;
static NdbDictionary::Dictionary* myDic = 0;
static NdbConnection* myCon = 0;
static NdbOperation* myOp = 0;
static NdbBlob* myBlob = 0;
static Opt g_opt;
static void
fatal(const char* fmt, ...)
{
va_list ap;
char buf[200];
va_start(ap, fmt);
vsnprintf(buf, sizeof(buf), fmt, ap);
va_end(ap);
ndbout << "fatal: " << buf << endl;
if (myNdb != 0 && myNdb->getNdbError().code != 0)
ndbout << "ndb - " << myNdb->getNdbError() << endl;
if (myDic != 0 && myDic->getNdbError().code != 0)
ndbout << "dic - " << myDic->getNdbError() << endl;
if (opt.m_core)
abort();
NDBT_ProgramExit(NDBT_FAILED);
exit(1);
static char&
skip(unsigned x)
{
assert(x < sizeof(g_opt.m_skip));
return g_opt.m_skip[x];
}
static Ndb* g_ndb = 0;
static NdbDictionary::Dictionary* g_dic = 0;
static NdbConnection* g_con = 0;
static NdbOperation* g_opr = 0;
static NdbIndexOperation* g_opx = 0;
static NdbScanOperation* g_ops = 0;
static NdbBlob* g_bh1 = 0;
static NdbBlob* g_bh2 = 0;
static bool g_printerror = true;
static void
dropBlobsTable()
printerror(int line, const char* msg)
{
NdbDictionary::Table tab(NDB_BLOB_TABLE_NAME);
if (myDic->dropTable(tab) == -1)
if (myDic->getNdbError().code != 709)
fatal("dropTable");
ndbout << "line " << line << ": " << msg << " failed" << endl;
if (! g_printerror) {
return;
}
if (g_ndb != 0 && g_ndb->getNdbError().code != 0) {
ndbout << "ndb: " << g_ndb->getNdbError() << endl;
}
if (g_dic != 0 && g_dic->getNdbError().code != 0) {
ndbout << "dic: " << g_dic->getNdbError() << endl;
}
if (g_con != 0 && g_con->getNdbError().code != 0) {
ndbout << "con: " << g_con->getNdbError() << endl;
if (g_opr != 0 && g_opr->getNdbError().code != 0) {
ndbout << "opr: table=" << g_opr->getTableName() << " " << g_opr->getNdbError() << endl;
}
if (g_opx != 0 && g_opx->getNdbError().code != 0) {
ndbout << "opx: table=" << g_opx->getTableName() << " " << g_opx->getNdbError() << endl;
}
if (g_ops != 0 && g_ops->getNdbError().code != 0) {
ndbout << "ops: table=" << g_ops->getTableName() << " " << g_ops->getNdbError() << endl;
}
NdbOperation* ope = g_con->getNdbErrorOperation();
if (ope != 0 && ope->getNdbError().code != 0) {
if (ope != g_opr && ope != g_opx && ope != g_ops)
ndbout << "ope: table=" << ope->getTableName() << " " << ope->getNdbError() << endl;
}
}
if (g_bh1 != 0 && g_bh1->getNdbError().code != 0) {
ndbout << "bh1: " << g_bh1->getNdbError() << endl;
}
if (g_bh2 != 0 && g_bh2->getNdbError().code != 0) {
ndbout << "bh2: " << g_bh2->getNdbError() << endl;
}
if (g_opt.m_core) {
abort();
}
g_printerror = false;
}
static void
createBlobsTable()
{
NdbDictionary::Table tab(NDB_BLOB_TABLE_NAME);
// col 0
NdbDictionary::Column col0("BLOBID");
col0.setPrimaryKey(true);
col0.setType(NdbDictionary::Column::Bigunsigned);
tab.addColumn(col0);
// col 1
NdbDictionary::Column col1("DATA");
col1.setPrimaryKey(false);
col1.setType(NdbDictionary::Column::Binary);
col1.setLength(NDB_BLOB_PIECE_SIZE);
tab.addColumn(col1);
// create
if (myDic->createTable(tab) == -1)
fatal("createTable");
}
#define CHK(x) \
do { \
if (x) break; \
printerror(__LINE__, #x); return -1; \
} while (0)
#define DBG(x) \
do { \
if (! g_opt.m_dbg) break; \
ndbout << "line " << __LINE__ << " " << x << endl; \
} while (0)
static void
static int
dropTable()
{
NdbDictionary::Table tab(opt.m_table);
if (myDic->dropTable(tab) == -1)
if (myDic->getNdbError().code != 709)
fatal("dropTable");
NdbDictionary::Table tab(g_opt.m_tname);
if (g_dic->getTable(g_opt.m_tname) != 0)
CHK(g_dic->dropTable(tab) == 0);
return 0;
}
static void
static int
createTable()
{
NdbDictionary::Table tab(opt.m_table);
// col 0
NdbDictionary::Column col0("A");
col0.setPrimaryKey(true);
col0.setType(NdbDictionary::Column::Unsigned);
tab.addColumn(col0);
// col 1
NdbDictionary::Column col1("B");
col1.setPrimaryKey(false);
col1.setType(NdbDictionary::Column::Blob);
tab.addColumn(col1);
// create
if (myDic->createTable(tab) == -1)
fatal("createTable");
NdbDictionary::Table tab(g_opt.m_tname);
// col PK1 - Uint32
{ NdbDictionary::Column col("PK1");
col.setType(NdbDictionary::Column::Unsigned);
col.setPrimaryKey(true);
tab.addColumn(col);
}
// col BL1 - Blob not-nullable
{ NdbDictionary::Column col("BL1");
const Bcol& b = g_opt.m_blob1;
col.setType(NdbDictionary::Column::Blob);
col.setInlineSize(b.m_inline);
col.setPartSize(b.m_partsize);
col.setStripeSize(b.m_stripe);
tab.addColumn(col);
}
// col PK2 - Char[55]
if (g_opt.m_pk2len != 0)
{ NdbDictionary::Column col("PK2");
col.setType(NdbDictionary::Column::Char);
col.setLength(g_opt.m_pk2len);
col.setPrimaryKey(true);
tab.addColumn(col);
}
// col BL2 - Clob nullable
if (! g_opt.m_oneblob)
{ NdbDictionary::Column col("BL2");
const Bcol& b = g_opt.m_blob2;
col.setType(NdbDictionary::Column::Clob);
col.setNullable(true);
col.setInlineSize(b.m_inline);
col.setPartSize(b.m_partsize);
col.setStripeSize(b.m_stripe);
tab.addColumn(col);
}
// create table
CHK(g_dic->createTable(tab) == 0);
// unique hash index on PK2
if (g_opt.m_pk2len != 0)
{ NdbDictionary::Index idx(g_opt.m_x1name);
idx.setType(NdbDictionary::Index::UniqueHashIndex);
idx.setTable(g_opt.m_tname);
idx.addColumnName("PK2");
CHK(g_dic->createIndex(idx) == 0);
}
// ordered index on PK2
if (g_opt.m_pk2len != 0)
{ NdbDictionary::Index idx(g_opt.m_x2name);
idx.setType(NdbDictionary::Index::OrderedIndex);
idx.setLogging(false);
idx.setTable(g_opt.m_tname);
idx.addColumnName("PK2");
CHK(g_dic->createIndex(idx) == 0);
}
return 0;
}
// tuples
struct Bval {
char* m_val;
unsigned m_len;
char* m_buf;
unsigned m_buflen;
Bval() :
m_val(0),
m_len(0),
m_buf(0), // read/write buffer
m_buflen(0)
{}
~Bval() { delete [] m_val; delete [] m_buf; }
void alloc(unsigned buflen) {
m_buflen = buflen;
delete [] m_buf;
m_buf = new char [m_buflen];
trash();
}
void copy(const Bval& v) {
m_len = v.m_len;
delete [] m_val;
if (v.m_val == 0)
m_val = 0;
else
m_val = (char*)memcpy(new char [m_len], v.m_val, m_len);
}
void trash() const {
assert(m_buf != 0);
memset(m_buf, 'x', m_buflen);
}
private:
Bval(const Bval&);
Bval& operator=(const Bval&);
};
struct Tup {
bool m_exists; // exists in table
Uint32 m_pk1; // primary keys concatenated like keyinfo
char m_pk2[g_max_pk2len + 1];
Bval m_blob1;
Bval m_blob2;
Tup() :
m_exists(false)
{}
~Tup() { }
// alloc buffers of max size
void alloc() {
m_blob1.alloc(g_opt.m_blob1.m_inline + g_opt.m_blob1.m_partsize * g_opt.m_parts);
m_blob2.alloc(g_opt.m_blob2.m_inline + g_opt.m_blob2.m_partsize * g_opt.m_parts);
}
void copy(const Tup& tup) {
assert(m_pk1 == tup.m_pk1);
m_blob1.copy(tup.m_blob1);
m_blob2.copy(tup.m_blob2);
}
private:
Tup(const Tup&);
Tup& operator=(const Tup&);
};
static Tup* g_tups;
static unsigned
urandom(unsigned n)
{
return n == 0 ? 0 : random() % n;
}
static void
insertData(Uint32 key)
calcBval(const Bcol& b, Bval& v, bool keepsize)
{
if (b.m_nullable && urandom(10) == 0) {
v.m_len = 0;
delete v.m_val;
v.m_val = 0;
v.m_buf = new char [1];
} else {
if (keepsize && v.m_val != 0)
;
else if (urandom(10) == 0)
v.m_len = urandom(b.m_inline);
else
v.m_len = urandom(b.m_inline + g_opt.m_parts * b.m_partsize + 1);
delete v.m_val;
v.m_val = new char [v.m_len + 1];
for (unsigned i = 0; i < v.m_len; i++)
v.m_val[i] = 'a' + urandom(25);
v.m_val[v.m_len] = 0;
v.m_buf = new char [v.m_len];
}
v.m_buflen = v.m_len;
v.trash();
}
static void
insertTuples()
{
for (Uint32 key = 0; key <= 99; key++) {
if ((myCon = myNdb->startTransaction()) == 0)
fatal("startTransaction");
if ((myOp = myCon->getNdbOperation(opt.m_table)) == 0)
fatal("getNdbOperation");
if (myOp->insertTuple() == -1)
fatal("insertTuple");
if (myOp->setValue((unsigned)0, key) == -1)
fatal("setValue %u", (unsigned)key);
if ((myBlob = myOp->setBlob(1)) == 0)
fatal("setBlob");
if (myCon->execute(NoCommit) == -1)
fatal("execute NoCommit");
insertData(key);
if (myCon->execute(Commit) == -1)
fatal("execute Commit");
myNdb->closeTransaction(myCon);
myOp = 0;
myBlob = 0;
myCon = 0;
calcTups(bool keepsize)
{
for (unsigned k = 0; k < g_opt.m_rows; k++) {
Tup& tup = g_tups[k];
tup.m_pk1 = g_opt.m_pk1off + k;
for (unsigned i = 0, n = k; i < g_opt.m_pk2len; i++) {
if (n != 0) {
tup.m_pk2[i] = '0' + n % 10;
n = n / 10;
} else {
tup.m_pk2[i] = 'a' + i % 26;
}
}
calcBval(g_opt.m_blob1, tup.m_blob1, keepsize);
if (! g_opt.m_oneblob)
calcBval(g_opt.m_blob2, tup.m_blob2, keepsize);
}
}
static void
testMain()
// blob handle ops
static int
getBlobLength(NdbBlob* h, unsigned& len)
{
Uint64 len2 = (unsigned)-1;
CHK(h->getLength(len2) == 0);
len = (unsigned)len2;
assert(len == len2);
return 0;
}
static int
setBlobValue(NdbBlob* h, const Bval& v)
{
bool null = (v.m_val == 0);
bool isNull;
unsigned len;
DBG("set " << h->getColumn()->getName() << " len=" << v.m_len << " null=" << null);
if (null) {
CHK(h->setNull() == 0);
isNull = false;
CHK(h->getNull(isNull) == 0 && isNull == true);
CHK(getBlobLength(h, len) == 0 && len == 0);
} else {
CHK(h->setValue(v.m_val, v.m_len) == 0);
CHK(h->getNull(isNull) == 0 && isNull == false);
CHK(getBlobLength(h, len) == 0 && len == v.m_len);
}
return 0;
}
static int
getBlobValue(NdbBlob* h, const Bval& v)
{
bool null = (v.m_val == 0);
DBG("get " << h->getColumn()->getName() << " len=" << v.m_len << " null=" << null);
CHK(h->getValue(v.m_buf, v.m_buflen) == 0);
return 0;
}
static int
getBlobValue(const Tup& tup)
{
CHK(getBlobValue(g_bh1, tup.m_blob1) == 0);
if (! g_opt.m_oneblob)
CHK(getBlobValue(g_bh2, tup.m_blob2) == 0);
return 0;
}
static int
verifyBlobValue(NdbBlob* h, const Bval& v)
{
bool null = (v.m_val == 0);
bool isNull;
unsigned len;
if (null) {
isNull = false;
CHK(h->getNull(isNull) == 0 && isNull == true);
CHK(getBlobLength(h, len) == 0 && len == 0);
} else {
isNull = true;
CHK(h->getNull(isNull) == 0 && isNull == false);
CHK(getBlobLength(h, len) == 0 && len == v.m_len);
for (unsigned i = 0; i < v.m_len; i++)
CHK(v.m_val[i] == v.m_buf[i]);
}
return 0;
}
static int
verifyBlobValue(const Tup& tup)
{
CHK(verifyBlobValue(g_bh1, tup.m_blob1) == 0);
if (! g_opt.m_oneblob)
CHK(verifyBlobValue(g_bh2, tup.m_blob2) == 0);
return 0;
}
static int
writeBlobData(NdbBlob* h, const Bval& v)
{
bool null = (v.m_val == 0);
bool isNull;
unsigned len;
DBG("write " << h->getColumn()->getName() << " len=" << v.m_len << " null=" << null);
if (null) {
CHK(h->setNull() == 0);
isNull = false;
CHK(h->getNull(isNull) == 0 && isNull == true);
CHK(getBlobLength(h, len) == 0 && len == 0);
} else {
unsigned n = 0;
do {
unsigned m = g_opt.m_full ? v.m_len : urandom(v.m_len + 1);
if (m > v.m_len - n)
m = v.m_len - n;
DBG("write pos=" << n << " cnt=" << m);
CHK(h->writeData(v.m_val + n, m) == 0);
n += m;
} while (n < v.m_len);
assert(n == v.m_len);
isNull = true;
CHK(h->getNull(isNull) == 0 && isNull == false);
CHK(getBlobLength(h, len) == 0 && len == v.m_len);
}
return 0;
}
static int
readBlobData(NdbBlob* h, const Bval& v)
{
bool null = (v.m_val == 0);
bool isNull;
unsigned len;
DBG("read " << h->getColumn()->getName() << " len=" << v.m_len << " null=" << null);
if (null) {
isNull = false;
CHK(h->getNull(isNull) == 0 && isNull == true);
CHK(getBlobLength(h, len) == 0 && len == 0);
} else {
isNull = true;
CHK(h->getNull(isNull) == 0 && isNull == false);
CHK(getBlobLength(h, len) == 0 && len == v.m_len);
v.trash();
unsigned n = 0;
while (n < v.m_len) {
unsigned m = g_opt.m_full ? v.m_len : urandom(v.m_len + 1);
if (m > v.m_len - n)
m = v.m_len - n;
DBG("read pos=" << n << " cnt=" << m);
const unsigned m2 = m;
CHK(h->readData(v.m_buf + n, m) == 0);
CHK(m2 == m);
n += m;
}
assert(n == v.m_len);
// need to execute to see the data
CHK(g_con->execute(NoCommit) == 0);
for (unsigned i = 0; i < v.m_len; i++)
CHK(v.m_val[i] == v.m_buf[i]);
}
return 0;
}
static int
readBlobData(const Tup& tup)
{
CHK(readBlobData(g_bh1, tup.m_blob1) == 0);
if (! g_opt.m_oneblob)
CHK(readBlobData(g_bh2, tup.m_blob2) == 0);
return 0;
}
// verify blob data
static int
verifyHeadInline(const Bcol& c, const Bval& v, NdbRecAttr* ra)
{
if (v.m_val == 0) {
CHK(ra->isNULL() == 1);
} else {
CHK(ra->isNULL() == 0);
CHK(ra->u_64_value() == v.m_len);
}
return 0;
}
static int
verifyHeadInline(const Tup& tup)
{
DBG("verifyHeadInline pk1=" << tup.m_pk1);
CHK((g_con = g_ndb->startTransaction()) != 0);
CHK((g_opr = g_con->getNdbOperation(g_opt.m_tname)) != 0);
CHK(g_opr->readTuple() == 0);
CHK(g_opr->equal("PK1", tup.m_pk1) == 0);
if (g_opt.m_pk2len != 0)
CHK(g_opr->equal("PK2", tup.m_pk2) == 0);
NdbRecAttr* ra1;
NdbRecAttr* ra2;
CHK((ra1 = g_opr->getValue("BL1")) != 0);
if (! g_opt.m_oneblob)
CHK((ra2 = g_opr->getValue("BL2")) != 0);
if (tup.m_exists) {
CHK(g_con->execute(Commit) == 0);
DBG("verifyHeadInline BL1");
CHK(verifyHeadInline(g_opt.m_blob1, tup.m_blob1, ra1) == 0);
if (! g_opt.m_oneblob) {
DBG("verifyHeadInline BL2");
CHK(verifyHeadInline(g_opt.m_blob2, tup.m_blob2, ra2) == 0);
}
} else {
CHK(g_con->execute(Commit) == -1 && g_con->getNdbError().code == 626);
}
g_ndb->closeTransaction(g_con);
g_opr = 0;
g_con = 0;
return 0;
}
static int
verifyBlobTable(const Bcol& b, const Bval& v, Uint32 pk1, bool exists)
{
DBG("verify " << b.m_btname << " pk1=" << pk1);
NdbRecAttr* ra_pk;
NdbRecAttr* ra_part;
NdbRecAttr* ra_data;
CHK((g_con = g_ndb->startTransaction()) != 0);
CHK((g_opr = g_con->getNdbOperation(b.m_btname)) != 0);
CHK(g_opr->openScanRead() == 0);
CHK((ra_pk = g_opr->getValue("PK")) != 0);
CHK((ra_part = g_opr->getValue("PART")) != 0);
CHK((ra_data = g_opr->getValue("DATA")) != 0);
CHK(g_con->executeScan() == 0);
unsigned partcount;
if (! exists || v.m_len <= b.m_inline)
partcount = 0;
else
partcount = (v.m_len - b.m_inline + b.m_partsize - 1) / b.m_partsize;
char* seen = new char [partcount];
memset(seen, 0, partcount);
while (1) {
int ret;
CHK((ret = g_con->nextScanResult()) == 0 || ret == 1);
if (ret == 1)
break;
if (pk1 != ra_pk->u_32_value())
continue;
Uint32 part = ra_part->u_32_value();
DBG("part " << part << " of " << partcount);
const char* data = ra_data->aRef();
CHK(part < partcount && ! seen[part]);
seen[part] = 1;
unsigned n = b.m_inline + part * b.m_partsize;
assert(exists && v.m_val != 0 && n < v.m_len);
unsigned m = v.m_len - n;
if (m > b.m_partsize)
m = b.m_partsize;
CHK(memcmp(data, v.m_val + n, m) == 0);
}
for (unsigned i = 0; i < partcount; i++)
CHK(seen[i] == 1);
g_ndb->closeTransaction(g_con);
g_opr = 0;
g_con = 0;
return 0;
}
static int
verifyBlobTable(const Tup& tup)
{
CHK(verifyBlobTable(g_opt.m_blob1, tup.m_blob1, tup.m_pk1, tup.m_exists) == 0);
if (! g_opt.m_oneblob)
CHK(verifyBlobTable(g_opt.m_blob2, tup.m_blob2, tup.m_pk1, tup.m_exists) == 0);
return 0;
}
static int
verifyBlob()
{
myNdb = new Ndb("TEST_DB");
if (myNdb->init() != 0)
fatal("init");
if (myNdb->waitUntilReady() < 0)
fatal("waitUntilReady");
myDic = myNdb->getDictionary();
dropBlobsTable();
createBlobsTable(); // until moved to Ndbcntr
dropTable();
createTable();
insertTuples();
for (unsigned k = 0; k < g_opt.m_rows; k++) {
const Tup& tup = g_tups[k];
DBG("verifyBlob pk1=" << tup.m_pk1);
CHK(verifyHeadInline(tup) == 0);
CHK(verifyBlobTable(tup) == 0);
}
return 0;
}
// operations
static int
insertPk(bool rw)
{
DBG("--- insertPk ---");
for (unsigned k = 0; k < g_opt.m_rows; k++) {
Tup& tup = g_tups[k];
DBG("insertPk pk1=" << tup.m_pk1);
CHK((g_con = g_ndb->startTransaction()) != 0);
CHK((g_opr = g_con->getNdbOperation(g_opt.m_tname)) != 0);
CHK(g_opr->insertTuple() == 0);
CHK(g_opr->equal("PK1", tup.m_pk1) == 0);
if (g_opt.m_pk2len != 0)
CHK(g_opr->equal("PK2", tup.m_pk2) == 0);
CHK((g_bh1 = g_opr->getBlobHandle("BL1")) != 0);
if (! g_opt.m_oneblob)
CHK((g_bh2 = g_opr->getBlobHandle("BL2")) != 0);
if (! rw) {
CHK(setBlobValue(g_bh1, tup.m_blob1) == 0);
if (! g_opt.m_oneblob)
CHK(setBlobValue(g_bh2, tup.m_blob2) == 0);
} else {
// non-nullable must be set
CHK(g_bh1->setValue("", 0) == 0);
CHK(g_con->execute(NoCommit) == 0);
CHK(writeBlobData(g_bh1, tup.m_blob1) == 0);
if (! g_opt.m_oneblob)
CHK(writeBlobData(g_bh2, tup.m_blob2) == 0);
}
CHK(g_con->execute(Commit) == 0);
g_ndb->closeTransaction(g_con);
g_opr = 0;
g_con = 0;
tup.m_exists = true;
}
return 0;
}
static int
updatePk(bool rw)
{
DBG("--- updatePk ---");
for (unsigned k = 0; k < g_opt.m_rows; k++) {
Tup& tup = g_tups[k];
DBG("updatePk pk1=" << tup.m_pk1);
CHK((g_con = g_ndb->startTransaction()) != 0);
CHK((g_opr = g_con->getNdbOperation(g_opt.m_tname)) != 0);
CHK(g_opr->updateTuple() == 0);
CHK(g_opr->equal("PK1", tup.m_pk1) == 0);
if (g_opt.m_pk2len != 0)
CHK(g_opr->equal("PK2", tup.m_pk2) == 0);
CHK((g_bh1 = g_opr->getBlobHandle("BL1")) != 0);
if (! g_opt.m_oneblob)
CHK((g_bh2 = g_opr->getBlobHandle("BL2")) != 0);
if (! rw) {
CHK(setBlobValue(g_bh1, tup.m_blob1) == 0);
if (! g_opt.m_oneblob)
CHK(setBlobValue(g_bh2, tup.m_blob2) == 0);
} else {
CHK(g_con->execute(NoCommit) == 0);
CHK(writeBlobData(g_bh1, tup.m_blob1) == 0);
if (! g_opt.m_oneblob)
CHK(writeBlobData(g_bh2, tup.m_blob2) == 0);
}
CHK(g_con->execute(Commit) == 0);
g_ndb->closeTransaction(g_con);
g_opr = 0;
g_con = 0;
tup.m_exists = true;
}
return 0;
}
static int
updateIdx(bool rw)
{
DBG("--- updateIdx ---");
for (unsigned k = 0; k < g_opt.m_rows; k++) {
Tup& tup = g_tups[k];
DBG("updateIdx pk1=" << tup.m_pk1);
CHK((g_con = g_ndb->startTransaction()) != 0);
CHK((g_opx = g_con->getNdbIndexOperation(g_opt.m_x1name, g_opt.m_tname)) != 0);
CHK(g_opx->updateTuple() == 0);
CHK(g_opx->equal("PK2", tup.m_pk2) == 0);
CHK((g_bh1 = g_opx->getBlobHandle("BL1")) != 0);
if (! g_opt.m_oneblob)
CHK((g_bh2 = g_opx->getBlobHandle("BL2")) != 0);
if (! rw) {
CHK(setBlobValue(g_bh1, tup.m_blob1) == 0);
if (! g_opt.m_oneblob)
CHK(setBlobValue(g_bh2, tup.m_blob2) == 0);
} else {
CHK(g_con->execute(NoCommit) == 0);
CHK(writeBlobData(g_bh1, tup.m_blob1) == 0);
if (! g_opt.m_oneblob)
CHK(writeBlobData(g_bh2, tup.m_blob2) == 0);
}
CHK(g_con->execute(Commit) == 0);
g_ndb->closeTransaction(g_con);
g_opx = 0;
g_con = 0;
tup.m_exists = true;
}
return 0;
}
static int
readPk(bool rw)
{
DBG("--- readPk ---");
for (unsigned k = 0; k < g_opt.m_rows; k++) {
Tup& tup = g_tups[k];
DBG("readPk pk1=" << tup.m_pk1);
CHK((g_con = g_ndb->startTransaction()) != 0);
CHK((g_opr = g_con->getNdbOperation(g_opt.m_tname)) != 0);
CHK(g_opr->readTuple() == 0);
CHK(g_opr->equal("PK1", tup.m_pk1) == 0);
if (g_opt.m_pk2len != 0)
CHK(g_opr->equal("PK2", tup.m_pk2) == 0);
CHK((g_bh1 = g_opr->getBlobHandle("BL1")) != 0);
if (! g_opt.m_oneblob)
CHK((g_bh2 = g_opr->getBlobHandle("BL2")) != 0);
if (! rw) {
CHK(getBlobValue(tup) == 0);
} else {
CHK(g_con->execute(NoCommit) == 0);
CHK(readBlobData(tup) == 0);
}
CHK(g_con->execute(Commit) == 0);
if (! rw) {
CHK(verifyBlobValue(tup) == 0);
}
g_ndb->closeTransaction(g_con);
g_opr = 0;
g_con = 0;
}
return 0;
}
static int
readIdx(bool rw)
{
DBG("--- readIdx ---");
for (unsigned k = 0; k < g_opt.m_rows; k++) {
Tup& tup = g_tups[k];
DBG("readIdx pk1=" << tup.m_pk1);
CHK((g_con = g_ndb->startTransaction()) != 0);
CHK((g_opx = g_con->getNdbIndexOperation(g_opt.m_x1name, g_opt.m_tname)) != 0);
CHK(g_opx->readTuple() == 0);
CHK(g_opx->equal("PK2", tup.m_pk2) == 0);
CHK((g_bh1 = g_opx->getBlobHandle("BL1")) != 0);
if (! g_opt.m_oneblob)
CHK((g_bh2 = g_opx->getBlobHandle("BL2")) != 0);
if (! rw) {
CHK(getBlobValue(tup) == 0);
} else {
CHK(g_con->execute(NoCommit) == 0);
CHK(readBlobData(tup) == 0);
}
CHK(g_con->execute(Commit) == 0);
if (! rw) {
CHK(verifyBlobValue(tup) == 0);
}
g_ndb->closeTransaction(g_con);
g_opx = 0;
g_con = 0;
}
return 0;
}
static int
readScan(bool rw, bool idx)
{
const char* func = ! idx ? "scan read table" : "scan read index";
DBG("--- " << func << " ---");
Tup tup;
tup.alloc(); // allocate buffers
NdbResultSet* rs;
CHK((g_con = g_ndb->startTransaction()) != 0);
if (! idx) {
CHK((g_ops = g_con->getNdbScanOperation(g_opt.m_tname)) != 0);
} else {
CHK((g_ops = g_con->getNdbScanOperation(g_opt.m_x2name, g_opt.m_tname)) != 0);
}
CHK((rs = g_ops->readTuples(240, NdbScanOperation::LM_Exclusive)) != 0);
CHK(g_ops->getValue("PK1", (char*)&tup.m_pk1) != 0);
if (g_opt.m_pk2len != 0)
CHK(g_ops->getValue("PK2", tup.m_pk2) != 0);
CHK((g_bh1 = g_ops->getBlobHandle("BL1")) != 0);
if (! g_opt.m_oneblob)
CHK((g_bh2 = g_ops->getBlobHandle("BL2")) != 0);
if (! rw) {
CHK(getBlobValue(tup) == 0);
}
CHK(g_con->execute(NoCommit) == 0);
unsigned rows = 0;
while (1) {
int ret;
tup.m_pk1 = (Uint32)-1;
memset(tup.m_pk2, 'x', g_opt.m_pk2len);
CHK((ret = rs->nextResult(true)) == 0 || ret == 1);
if (ret == 1)
break;
DBG(func << " pk1=" << tup.m_pk1);
Uint32 k = tup.m_pk1 - g_opt.m_pk1off;
CHK(k < g_opt.m_rows && g_tups[k].m_exists);
tup.copy(g_tups[k]);
if (! rw) {
CHK(verifyBlobValue(tup) == 0);
} else {
CHK(readBlobData(tup) == 0);
}
rows++;
}
g_ndb->closeTransaction(g_con);
g_con = 0;
g_ops = 0;
CHK(g_opt.m_rows == rows);
return 0;
}
static int
deletePk()
{
DBG("--- deletePk ---");
for (unsigned k = 0; k < g_opt.m_rows; k++) {
Tup& tup = g_tups[k];
DBG("deletePk pk1=" << tup.m_pk1);
CHK((g_con = g_ndb->startTransaction()) != 0);
CHK((g_opr = g_con->getNdbOperation(g_opt.m_tname)) != 0);
CHK(g_opr->deleteTuple() == 0);
CHK(g_opr->equal("PK1", tup.m_pk1) == 0);
if (g_opt.m_pk2len != 0)
CHK(g_opr->equal("PK2", tup.m_pk2) == 0);
CHK(g_con->execute(Commit) == 0);
g_ndb->closeTransaction(g_con);
g_opr = 0;
g_con = 0;
tup.m_exists = false;
}
return 0;
}
static int
deleteIdx()
{
DBG("--- deleteIdx ---");
for (unsigned k = 0; k < g_opt.m_rows; k++) {
Tup& tup = g_tups[k];
DBG("deleteIdx pk1=" << tup.m_pk1);
CHK((g_con = g_ndb->startTransaction()) != 0);
CHK((g_opx = g_con->getNdbIndexOperation(g_opt.m_x1name, g_opt.m_tname)) != 0);
CHK(g_opx->deleteTuple() == 0);
CHK(g_opx->equal("PK2", tup.m_pk2) == 0);
CHK(g_con->execute(Commit) == 0);
g_ndb->closeTransaction(g_con);
g_opx = 0;
g_con = 0;
tup.m_exists = false;
}
return 0;
}
static int
deleteScan(bool idx)
{
const char* func = ! idx ? "scan delete table" : "scan delete index";
DBG("--- " << func << " ---");
Tup tup;
NdbResultSet* rs;
CHK((g_con = g_ndb->startTransaction()) != 0);
if (! idx) {
CHK((g_ops = g_con->getNdbScanOperation(g_opt.m_tname)) != 0);
} else {
CHK((g_ops = g_con->getNdbScanOperation(g_opt.m_x2name, g_opt.m_tname)) != 0);
}
CHK((rs = g_ops->readTuples(240, NdbScanOperation::LM_Exclusive)) != 0);
CHK(g_ops->getValue("PK1", (char*)&tup.m_pk1) != 0);
if (g_opt.m_pk2len != 0)
CHK(g_ops->getValue("PK2", tup.m_pk2) != 0);
CHK(g_con->execute(NoCommit) == 0);
unsigned rows = 0;
while (1) {
int ret;
tup.m_pk1 = (Uint32)-1;
memset(tup.m_pk2, 'x', g_opt.m_pk2len);
CHK((ret = rs->nextResult()) == 0 || ret == 1);
if (ret == 1)
break;
DBG(func << " pk1=" << tup.m_pk1);
CHK(rs->deleteTuple() == 0);
CHK(g_con->execute(NoCommit) == 0);
Uint32 k = tup.m_pk1 - g_opt.m_pk1off;
CHK(k < g_opt.m_rows && g_tups[k].m_exists);
g_tups[k].m_exists = false;
rows++;
}
CHK(g_con->execute(Commit) == 0);
g_ndb->closeTransaction(g_con);
g_con = 0;
g_opr = 0;
g_ops = 0;
CHK(g_opt.m_rows == rows);
return 0;
}
// main
static int
testmain()
{
g_ndb = new Ndb("TEST_DB");
CHK(g_ndb->init() == 0);
CHK(g_ndb->waitUntilReady() == 0);
g_dic = g_ndb->getDictionary();
g_tups = new Tup [g_opt.m_rows];
CHK(dropTable() == 0);
CHK(createTable() == 0);
if (g_opt.m_bugtest != 0) {
// test a general bug instead of blobs
CHK((*g_opt.m_bugtest)() == 0);
return 0;
}
Bcol& b1 = g_opt.m_blob1;
CHK(NdbBlob::getBlobTableName(b1.m_btname, g_ndb, g_opt.m_tname, "BL1") == 0);
DBG("BL1: inline=" << b1.m_inline << " part=" << b1.m_partsize << " table=" << b1.m_btname);
if (! g_opt.m_oneblob) {
Bcol& b2 = g_opt.m_blob2;
CHK(NdbBlob::getBlobTableName(b2.m_btname, g_ndb, g_opt.m_tname, "BL2") == 0);
DBG("BL2: inline=" << b2.m_inline << " part=" << b2.m_partsize << " table=" << b2.m_btname);
}
if (g_opt.m_seed != 0)
srandom(g_opt.m_seed);
for (unsigned loop = 0; g_opt.m_loop == 0 || loop < g_opt.m_loop; loop++) {
DBG("=== loop " << loop << " ===");
if (g_opt.m_seed == 0)
srandom(loop);
bool llim = skip('v') ? true : false;
bool ulim = skip('w') ? false : true;
// pk
for (int rw = llim; rw <= ulim; rw++) {
DBG("--- pk ops " << (! rw ? "get/set" : "read/write") << " ---");
calcTups(false);
CHK(insertPk(rw) == 0);
CHK(verifyBlob() == 0);
CHK(readPk(rw) == 0);
if (! skip('u')) {
calcTups(rw);
CHK(updatePk(rw) == 0);
CHK(verifyBlob() == 0);
}
CHK(readPk(rw) == 0);
CHK(deletePk() == 0);
CHK(verifyBlob() == 0);
}
// hash index
for (int rw = llim; rw <= ulim; rw++) {
if (skip('i'))
continue;
DBG("--- idx ops " << (! rw ? "get/set" : "read/write") << " ---");
calcTups(false);
CHK(insertPk(rw) == 0);
CHK(verifyBlob() == 0);
CHK(readIdx(rw) == 0);
calcTups(rw);
if (! skip('u')) {
CHK(updateIdx(rw) == 0);
CHK(verifyBlob() == 0);
CHK(readIdx(rw) == 0);
}
CHK(deleteIdx() == 0);
CHK(verifyBlob() == 0);
}
// scan table
for (int rw = llim; rw <= ulim; rw++) {
if (skip('s'))
continue;
DBG("--- table scan " << (! rw ? "get/set" : "read/write") << " ---");
calcTups(false);
CHK(insertPk(rw) == 0);
CHK(verifyBlob() == 0);
CHK(readScan(rw, false) == 0);
CHK(deleteScan(false) == 0);
CHK(verifyBlob() == 0);
}
// scan index
for (int rw = llim; rw <= ulim; rw++) {
if (skip('r'))
continue;
DBG("--- index scan " << (! rw ? "get/set" : "read/write") << " ---");
calcTups(false);
CHK(insertPk(rw) == 0);
CHK(verifyBlob() == 0);
CHK(readScan(rw, true) == 0);
CHK(deleteScan(true) == 0);
CHK(verifyBlob() == 0);
}
}
delete g_ndb;
return 0;
}
// bug tests
static int
bugtest_4088()
{
DBG("bug test 4088 - ndb api hang with mixed ops on index table");
// insert rows
calcTups(false);
CHK(insertPk(false) == 0);
// new trans
CHK((g_con = g_ndb->startTransaction()) != 0);
for (unsigned k = 0; k < g_opt.m_rows; k++) {
Tup& tup = g_tups[k];
// read table pk via index as a table
const unsigned pkcnt = 2;
Tup pktup[pkcnt];
for (unsigned i = 0; i < pkcnt; i++) {
char name[20];
// XXX guess table id
sprintf(name, "%d/%s", 4, g_opt.m_x1name);
CHK((g_opr = g_con->getNdbOperation(name)) != 0);
CHK(g_opr->readTuple() == 0);
CHK(g_opr->equal("PK2", tup.m_pk2) == 0);
CHK(g_opr->getValue("NDB$PK", (char*)&pktup[i].m_pk1) != 0);
}
// read blob inline via index as an index
CHK((g_opx = g_con->getNdbIndexOperation(g_opt.m_x1name, g_opt.m_tname)) != 0);
CHK(g_opx->readTuple() == 0);
CHK(g_opx->equal("PK2", tup.m_pk2) == 0);
assert(tup.m_blob1.m_buf != 0);
CHK(g_opx->getValue("BL1", (char*)tup.m_blob1.m_buf) != 0);
// execute
// BUG 4088: gets 1 tckeyconf, 1 tcindxconf, then hangs
CHK(g_con->execute(Commit) == 0);
// verify
for (unsigned i = 0; i < pkcnt; i++) {
CHK(pktup[i].m_pk1 == tup.m_pk1);
CHK(memcmp(pktup[i].m_pk2, tup.m_pk2, g_opt.m_pk2len) == 0);
}
CHK(memcmp(tup.m_blob1.m_val, tup.m_blob1.m_buf, 8 + g_opt.m_blob1.m_inline) == 0);
}
return 0;
}
static int
bugtest_2222()
{
return 0;
}
static int
bugtest_3333()
{
return 0;
}
static struct {
int m_bug;
int (*m_test)();
} g_bugtest[] = {
{ 4088, bugtest_4088 },
{ 2222, bugtest_2222 },
{ 3333, bugtest_3333 }
};
NDB_COMMAND(testOdbcDriver, "testBlobs", "testBlobs", "testBlobs", 65535)
{
while (++argv, --argc > 0) {
const char* arg = argv[0];
if (strcmp(arg, "-core") == 0) {
opt.m_core = true;
continue;
}
while (++argv, --argc > 0) {
const char* arg = argv[0];
if (strcmp(arg, "-core") == 0) {
g_opt.m_core = true;
continue;
}
if (strcmp(arg, "-dbg") == 0) {
g_opt.m_dbg = true;
continue;
}
if (strcmp(arg, "-dbgall") == 0) {
g_opt.m_dbg = true;
g_opt.m_dbgall = true;
putenv("NDB_BLOB_DEBUG=1");
continue;
}
if (strcmp(arg, "-full") == 0) {
g_opt.m_full = true;
continue;
}
if (strcmp(arg, "-loop") == 0) {
if (++argv, --argc > 0) {
g_opt.m_loop = atoi(argv[0]);
continue;
}
}
if (strcmp(arg, "-parts") == 0) {
if (++argv, --argc > 0) {
g_opt.m_parts = atoi(argv[0]);
continue;
}
}
if (strcmp(arg, "-rows") == 0) {
if (++argv, --argc > 0) {
g_opt.m_rows = atoi(argv[0]);
continue;
}
}
if (strcmp(arg, "-seed") == 0) {
if (++argv, --argc > 0) {
g_opt.m_seed = atoi(argv[0]);
continue;
}
}
if (strcmp(arg, "-skip") == 0) {
if (++argv, --argc > 0) {
for (const char* p = argv[0]; *p != 0; p++) {
skip(*p) = true;
}
continue;
}
}
// metadata
if (strcmp(arg, "-pk2len") == 0) {
if (++argv, --argc > 0) {
g_opt.m_pk2len = atoi(argv[0]);
if (g_opt.m_pk2len == 0) {
skip('i') = true;
skip('r') = true;
}
if (g_opt.m_pk2len <= g_max_pk2len)
continue;
}
}
if (strcmp(arg, "-oneblob") == 0) {
g_opt.m_oneblob = true;
continue;
}
// bugs
if (strcmp(arg, "-bug") == 0) {
if (++argv, --argc > 0) {
g_opt.m_bug = atoi(argv[0]);
for (unsigned i = 0; i < sizeof(g_bugtest)/sizeof(g_bugtest[0]); i++) {
if (g_opt.m_bug == g_bugtest[i].m_bug) {
g_opt.m_bugtest = g_bugtest[i].m_test;
break;
}
}
if (g_opt.m_bugtest != 0)
continue;
}
}
testMain();
return NDBT_ProgramExit(NDBT_OK);
ndbout << "testOIBasic: unknown option " << arg << endl;
printusage();
return NDBT_ProgramExit(NDBT_WRONGARGS);
}
if (testmain() == -1) {
return NDBT_ProgramExit(NDBT_FAILED);
}
return NDBT_ProgramExit(NDBT_OK);
}
// vim: set sw=4:
// vim: set sw=2 et:
ndb/tools/ndbsql.cpp
View file @ 9a862183
......@@ -33,6 +33,12 @@
#include <ctype.h>
#include <wctype.h>
#ifndef SQL_BLOB
#define SQL_BLOB 30
#endif
#ifndef SQL_CLOB
#define SQL_CLOB 40
#endif
/**************************************************************************
* ------------------------------------------------------------------------
......@@ -211,8 +217,10 @@ SQLINTEGER display_length(SQLSMALLINT coltype, SQLINTEGER collen,
switch (coltype) {
case SQL_VARCHAR:
case SQL_CHAR:
//case SQL_BLOB:
//case SQL_CLOB:
case SQL_VARBINARY:
case SQL_BINARY:
case SQL_BLOB:
case SQL_CLOB:
case SQL_BIT:
//case SQL_REF:
//case SQL_BIT_VARYING:
......
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