Skip to content

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

ndb charsets: TUX must use Dbtup::readAttributes

parent 82c0e3b4
Hide whitespace changes
Inline Side-by-side
Showing with 161 additions and 261 deletions
ndb/src/kernel/blocks/dbtup/Dbtup.hpp
View file @ 23dd2e15
......@@ -1004,17 +1004,20 @@ public:
void tuxGetNode(Uint32 fragPtrI, Uint32 pageId, Uint32 pageOffset, Uint32*& node);
/*
* TUX reads primary table attributes for index keys. Input is
* attribute ids in AttributeHeader format. Output is pointers to
* attribute data within tuple or 0 for NULL value.
* TUX reads primary table attributes for index keys. Tuple is
* specified by location of original tuple and version number. Input
* is attribute ids in AttributeHeader format. Output is attribute
* data with headers. Uses readAttributes with xfrm option set.
* Returns number of words or negative (-terrorCode) on error.
*/
void tuxReadAttrs(Uint32 fragPtrI, Uint32 pageId, Uint32 pageOffset, Uint32 tupVersion, Uint32 numAttrs, const Uint32* attrIds, const Uint32** attrData);
int tuxReadAttrs(Uint32 fragPtrI, Uint32 pageId, Uint32 pageOffset, Uint32 tupVersion, const Uint32* attrIds, Uint32 numAttrs, Uint32* dataOut);
/*
* TUX reads primary key without headers into an array of words. Used
* for md5 summing and when returning keyinfo.
* for md5 summing and when returning keyinfo. Returns number of
* words or negative (-terrorCode) on error.
*/
void tuxReadKeys(Uint32 fragPtrI, Uint32 pageId, Uint32 pageOffset, Uint32* pkSize, Uint32* pkData);
int tuxReadPk(Uint32 fragPtrI, Uint32 pageId, Uint32 pageOffset, Uint32* dataOut);
/*
* TUX checks if tuple is visible to scan.
......@@ -1368,7 +1371,7 @@ private:
//------------------------------------------------------------------
int readAttributes(Page* const pagePtr,
Uint32 TupHeadOffset,
Uint32* inBuffer,
const Uint32* inBuffer,
Uint32 inBufLen,
Uint32* outBuffer,
Uint32 TmaxRead);
......
ndb/src/kernel/blocks/dbtup/DbtupIndex.cpp
View file @ 23dd2e15
......@@ -112,10 +112,11 @@ Dbtup::tuxGetNode(Uint32 fragPtrI, Uint32 pageId, Uint32 pageOffset, Uint32*& no
node = &pagePtr.p->pageWord[pageOffset] + attrDataOffset;
}
void
Dbtup::tuxReadAttrs(Uint32 fragPtrI, Uint32 pageId, Uint32 pageOffset, Uint32 tupVersion, Uint32 numAttrs, const Uint32* attrIds, const Uint32** attrData)
int
Dbtup::tuxReadAttrs(Uint32 fragPtrI, Uint32 pageId, Uint32 pageOffset, Uint32 tupVersion, const Uint32* attrIds, Uint32 numAttrs, Uint32* dataOut)
{
ljamEntry();
// use own variables instead of globals
FragrecordPtr fragPtr;
fragPtr.i = fragPtrI;
ptrCheckGuard(fragPtr, cnoOfFragrec, fragrecord);
......@@ -134,6 +135,7 @@ Dbtup::tuxReadAttrs(Uint32 fragPtrI, Uint32 pageId, Uint32 pageOffset, Uint32 tu
while (true) {
ptrCheckGuard(opPtr, cnoOfOprec, operationrec);
if (opPtr.p->realPageIdC != RNIL) {
// update page and offset
pagePtr.i = opPtr.p->realPageIdC;
pageOffset = opPtr.p->pageOffsetC;
ptrCheckGuard(pagePtr, cnoOfPage, page);
......@@ -147,33 +149,34 @@ Dbtup::tuxReadAttrs(Uint32 fragPtrI, Uint32 pageId, Uint32 pageOffset, Uint32 tu
ndbrequire(++loopGuard < (1 << ZTUP_VERSION_BITS));
}
}
const Uint32 tabDescriptor = tablePtr.p->tabDescriptor;
const Uint32* tupleHeader = &pagePtr.p->pageWord[pageOffset];
for (Uint32 i = 0; i < numAttrs; i++) {
AttributeHeader ah(attrIds[i]);
const Uint32 attrId = ah.getAttributeId();
const Uint32 index = tabDescriptor + (attrId << ZAD_LOG_SIZE);
const Uint32 desc1 = tableDescriptor[index].tabDescr;
const Uint32 desc2 = tableDescriptor[index + 1].tabDescr;
if (AttributeDescriptor::getNullable(desc1)) {
Uint32 offset = AttributeOffset::getNullFlagOffset(desc2);
ndbrequire(offset < tablePtr.p->tupNullWords);
offset += tablePtr.p->tupNullIndex;
ndbrequire(offset < tablePtr.p->tupheadsize);
if (AttributeOffset::isNULL(tupleHeader[offset], desc2)) {
ljam();
attrData[i] = 0;
continue;
}
}
attrData[i] = tupleHeader + AttributeOffset::getOffset(desc2);
// read key attributes from found tuple version
// save globals
TablerecPtr tabptr_old = tabptr;
FragrecordPtr fragptr_old = fragptr;
OperationrecPtr operPtr_old = operPtr;
// new globals
tabptr = tablePtr;
fragptr = fragPtr;
operPtr.i = RNIL;
operPtr.p = NULL;
// do it
int ret = readAttributes(pagePtr.p, pageOffset, attrIds, numAttrs, dataOut, ZNIL);
// restore globals
tabptr = tabptr_old;
fragptr = fragptr_old;
operPtr = operPtr_old;
// done
if (ret == (Uint32)-1) {
ret = terrorCode ? (-(int)terrorCode) : -1;
}
return ret;
}
void
Dbtup::tuxReadKeys(Uint32 fragPtrI, Uint32 pageId, Uint32 pageOffset, Uint32* pkSize, Uint32* pkData)
int
Dbtup::tuxReadPk(Uint32 fragPtrI, Uint32 pageId, Uint32 pageOffset, Uint32* dataOut)
{
ljamEntry();
// use own variables instead of globals
FragrecordPtr fragPtr;
fragPtr.i = fragPtrI;
ptrCheckGuard(fragPtr, cnoOfFragrec, fragrecord);
......@@ -184,25 +187,45 @@ Dbtup::tuxReadKeys(Uint32 fragPtrI, Uint32 pageId, Uint32 pageOffset, Uint32* pk
pagePtr.i = pageId;
ptrCheckGuard(pagePtr, cnoOfPage, page);
const Uint32 tabDescriptor = tablePtr.p->tabDescriptor;
const Uint32 numAttrs = tablePtr.p->noOfKeyAttr;
const Uint32* attrIds = &tableDescriptor[tablePtr.p->readKeyArray].tabDescr;
const Uint32* tupleHeader = &pagePtr.p->pageWord[pageOffset];
Uint32 size = 0;
for (Uint32 i = 0; i < numAttrs; i++) {
AttributeHeader ah(attrIds[i]);
const Uint32 attrId = ah.getAttributeId();
const Uint32 index = tabDescriptor + (attrId << ZAD_LOG_SIZE);
const Uint32 desc1 = tableDescriptor[index].tabDescr;
const Uint32 desc2 = tableDescriptor[index + 1].tabDescr;
ndbrequire(! AttributeDescriptor::getNullable(desc1));
const Uint32 attrSize = AttributeDescriptor::getSizeInWords(desc1);
const Uint32* attrData = tupleHeader + AttributeOffset::getOffset(desc2);
for (Uint32 j = 0; j < attrSize; j++) {
pkData[size + j] = attrData[j];
const Uint32 numAttrs = tablePtr.p->noOfKeyAttr;
// read pk attributes from original tuple
// save globals
TablerecPtr tabptr_old = tabptr;
FragrecordPtr fragptr_old = fragptr;
OperationrecPtr operPtr_old = operPtr;
// new globals
tabptr = tablePtr;
fragptr = fragPtr;
operPtr.i = RNIL;
operPtr.p = NULL;
// do it
int ret = readAttributes(pagePtr.p, pageOffset, attrIds, numAttrs, dataOut, ZNIL);
// restore globals
tabptr = tabptr_old;
fragptr = fragptr_old;
operPtr = operPtr_old;
// done
if (ret != (Uint32)-1) {
// remove headers
Uint32 n = 0;
Uint32 i = 0;
while (n < numAttrs) {
const AttributeHeader ah(dataOut[i]);
Uint32 size = ah.getDataSize();
ndbrequire(size != 0);
for (Uint32 j = 0; j < size; j++) {
dataOut[i + j - n] = dataOut[i + j + 1];
}
n += 1;
i += 1 + size;
}
size += attrSize;
ndbrequire(i == ret);
ret -= numAttrs;
} else {
ret = terrorCode ? (-(int)terrorCode) : -1;
}
*pkSize = size;
return ret;
}
bool
......
ndb/src/kernel/blocks/dbtup/DbtupRoutines.cpp
View file @ 23dd2e15
......@@ -146,7 +146,7 @@ Dbtup::setUpQueryRoutines(Tablerec* const regTabPtr)
/* ---------------------------------------------------------------- */
int Dbtup::readAttributes(Page* const pagePtr,
Uint32 tupHeadOffset,
Uint32* inBuffer,
const Uint32* inBuffer,
Uint32 inBufLen,
Uint32* outBuffer,
Uint32 maxRead)
......
ndb/src/kernel/blocks/dbtux/Dbtux.hpp
View file @ 23dd2e15
......@@ -162,11 +162,6 @@ private:
// AttributeHeader size is assumed to be 1 word
static const unsigned AttributeHeaderSize = 1;
/*
* Array of pointers to TUP table attributes. Always read-on|y.
*/
typedef const Uint32** TableData;
/*
* Logical tuple address, "local key". Identifies table tuples.
*/
......@@ -557,9 +552,9 @@ private:
void execREAD_CONFIG_REQ(Signal* signal);
// utils
void setKeyAttrs(const Frag& frag);
void readKeyAttrs(const Frag& frag, TreeEnt ent, unsigned start, TableData keyData);
void readTablePk(const Frag& frag, TreeEnt ent, unsigned& pkSize, Data pkData);
void copyAttrs(const Frag& frag, TableData data1, Data data2, unsigned maxlen2 = MaxAttrDataSize);
void readKeyAttrs(const Frag& frag, TreeEnt ent, unsigned start, Data keyData);
void readTablePk(const Frag& frag, TreeEnt ent, Data pkData, unsigned& pkSize);
void copyAttrs(const Frag& frag, ConstData data1, Data data2, unsigned maxlen2 = MaxAttrDataSize);
/*
* DbtuxMeta.cpp
......@@ -626,17 +621,15 @@ private:
/*
* DbtuxSearch.cpp
*/
void searchToAdd(Signal* signal, Frag& frag, TableData searchKey, TreeEnt searchEnt, TreePos& treePos);
void searchToRemove(Signal* signal, Frag& frag, TableData searchKey, TreeEnt searchEnt, TreePos& treePos);
void searchToAdd(Signal* signal, Frag& frag, ConstData searchKey, TreeEnt searchEnt, TreePos& treePos);
void searchToRemove(Signal* signal, Frag& frag, ConstData searchKey, TreeEnt searchEnt, TreePos& treePos);
void searchToScan(Signal* signal, Frag& frag, ConstData boundInfo, unsigned boundCount, TreePos& treePos);
/*
* DbtuxCmp.cpp
*/
int cmpSearchKey(const Frag& frag, unsigned& start, TableData searchKey, ConstData entryData, unsigned maxlen = MaxAttrDataSize);
int cmpSearchKey(const Frag& frag, unsigned& start, TableData searchKey, TableData entryKey);
int cmpSearchKey(const Frag& frag, unsigned& start, ConstData searchKey, ConstData entryData, unsigned maxlen = MaxAttrDataSize);
int cmpScanBound(const Frag& frag, unsigned dir, ConstData boundInfo, unsigned boundCount, ConstData entryData, unsigned maxlen = MaxAttrDataSize);
int cmpScanBound(const Frag& frag, unsigned dir, ConstData boundInfo, unsigned boundCount, TableData entryKey);
/*
* DbtuxDebug.cpp
......@@ -683,17 +676,27 @@ private:
Uint32 c_typeOfStart;
/*
* Array of index key attribute ids in AttributeHeader format.
* Includes fixed attribute sizes. This is global data set at
* operation start and is not passed as a parameter.
* Global data set at operation start. Unpacked from index metadata.
* Not passed as parameter to methods. Invalid across timeslices.
*
* TODO inline all into index metadata
*/
// index key attr ids with sizes in AttributeHeader format
Data c_keyAttrs;
// buffer for search key data as pointers to TUP storage
TableData c_searchKey;
// pointers to index key comparison functions
NdbSqlUtil::Cmp** c_sqlCmp;
/*
* Other buffers used during the operation.
*/
// buffer for search key data with headers
Data c_searchKey;
// buffer for current entry key data as pointers to TUP storage
TableData c_entryKey;
// buffer for current entry key data with headers
Data c_entryKey;
// buffer for scan bounds and keyinfo (primary key)
Data c_dataBuffer;
......
ndb/src/kernel/blocks/dbtux/DbtuxCmp.cpp
View file @ 23dd2e15
......@@ -18,21 +18,24 @@
#include "Dbtux.hpp"
/*
* Search key vs node prefix.
* Search key vs node prefix or entry
*
* The comparison starts at given attribute position (in fact 0). The
* position is updated by number of equal initial attributes found. The
* prefix may be partial in which case CmpUnknown may be returned.
* The comparison starts at given attribute position. The position is
* updated by number of equal initial attributes found. The entry data
* may be partial in which case CmpUnknown may be returned.
*/
int
Dbtux::cmpSearchKey(const Frag& frag, unsigned& start, TableData searchKey, ConstData entryData, unsigned maxlen)
Dbtux::cmpSearchKey(const Frag& frag, unsigned& start, ConstData searchKey, ConstData entryData, unsigned maxlen)
{
const unsigned numAttrs = frag.m_numAttrs;
const DescEnt& descEnt = getDescEnt(frag.m_descPage, frag.m_descOff);
// number of words of attribute data left
unsigned len2 = maxlen;
// skip to right position in search key
searchKey += start;
// skip to right position in search key only
for (unsigned i = 0; i < start; i++) {
jam();
searchKey += AttributeHeaderSize + searchKey.ah().getDataSize();
}
int ret = 0;
while (start < numAttrs) {
if (len2 <= AttributeHeaderSize) {
......@@ -41,22 +44,21 @@ Dbtux::cmpSearchKey(const Frag& frag, unsigned& start, TableData searchKey, Cons
break;
}
len2 -= AttributeHeaderSize;
if (*searchKey != 0) {
if (! searchKey.ah().isNULL()) {
if (! entryData.ah().isNULL()) {
jam();
// current attribute
const DescAttr& descAttr = descEnt.m_descAttr[start];
const NdbSqlUtil::Type& type = NdbSqlUtil::getType(descAttr.m_typeId);
ndbassert(type.m_typeId != NdbSqlUtil::Type::Undefined);
// full data size
const unsigned size1 = AttributeDescriptor::getSizeInWords(descAttr.m_attrDesc);
ndbrequire(size1 != 0 && size1 == entryData.ah().getDataSize());
const unsigned size2 = min(size1, len2);
len2 -= size2;
// compare
const Uint32* const p1 = *searchKey;
NdbSqlUtil::Cmp* const cmp = c_sqlCmp[start];
const Uint32* const p1 = &searchKey[AttributeHeaderSize];
const Uint32* const p2 = &entryData[AttributeHeaderSize];
ret = (*type.m_cmp)(p1, p2, size1, size2);
ret = (*cmp)(p1, p2, size1, size2);
if (ret != 0) {
jam();
break;
......@@ -75,7 +77,7 @@ Dbtux::cmpSearchKey(const Frag& frag, unsigned& start, TableData searchKey, Cons
break;
}
}
searchKey += 1;
searchKey += AttributeHeaderSize + searchKey.ah().getDataSize();
entryData += AttributeHeaderSize + entryData.ah().getDataSize();
start++;
}
......@@ -83,60 +85,7 @@ Dbtux::cmpSearchKey(const Frag& frag, unsigned& start, TableData searchKey, Cons
}
/*
* Search key vs tree entry.
*
* Start position is updated as in previous routine.
*/
int
Dbtux::cmpSearchKey(const Frag& frag, unsigned& start, TableData searchKey, TableData entryKey)
{
const unsigned numAttrs = frag.m_numAttrs;
const DescEnt& descEnt = getDescEnt(frag.m_descPage, frag.m_descOff);
// skip to right position
searchKey += start;
entryKey += start;
int ret = 0;
while (start < numAttrs) {
if (*searchKey != 0) {
if (*entryKey != 0) {
jam();
// current attribute
const DescAttr& descAttr = descEnt.m_descAttr[start];
const NdbSqlUtil::Type& type = NdbSqlUtil::getType(descAttr.m_typeId);
ndbassert(type.m_typeId != NdbSqlUtil::Type::Undefined);
// full data size
const unsigned size1 = AttributeDescriptor::getSizeInWords(descAttr.m_attrDesc);
// compare
const Uint32* const p1 = *searchKey;
const Uint32* const p2 = *entryKey;
ret = (*type.m_cmp)(p1, p2, size1, size1);
if (ret != 0) {
jam();
break;
}
} else {
jam();
// not NULL > NULL
ret = +1;
break;
}
} else {
if (*entryKey != 0) {
jam();
// NULL < not NULL
ret = -1;
break;
}
}
searchKey += 1;
entryKey += 1;
start++;
}
return ret;
}
/*
* Scan bound vs node prefix.
* Scan bound vs node prefix or entry.
*
* Compare lower or upper bound and index attribute data. The attribute
* data may be partial in which case CmpUnknown may be returned.
......@@ -184,8 +133,6 @@ Dbtux::cmpScanBound(const Frag& frag, unsigned dir, ConstData boundInfo, unsigne
// current attribute
const unsigned index = boundInfo.ah().getAttributeId();
const DescAttr& descAttr = descEnt.m_descAttr[index];
const NdbSqlUtil::Type& type = NdbSqlUtil::getType(descAttr.m_typeId);
ndbassert(type.m_typeId != NdbSqlUtil::Type::Undefined);
ndbrequire(entryData.ah().getAttributeId() == descAttr.m_primaryAttrId);
// full data size
const unsigned size1 = boundInfo.ah().getDataSize();
......@@ -193,9 +140,10 @@ Dbtux::cmpScanBound(const Frag& frag, unsigned dir, ConstData boundInfo, unsigne
const unsigned size2 = min(size1, len2);
len2 -= size2;
// compare
NdbSqlUtil::Cmp* const cmp = c_sqlCmp[index];
const Uint32* const p1 = &boundInfo[AttributeHeaderSize];
const Uint32* const p2 = &entryData[AttributeHeaderSize];
int ret = (*type.m_cmp)(p1, p2, size1, size2);
int ret = (*cmp)(p1, p2, size1, size2);
if (ret != 0) {
jam();
return ret;
......@@ -244,72 +192,3 @@ Dbtux::cmpScanBound(const Frag& frag, unsigned dir, ConstData boundInfo, unsigne
return +1;
}
}
/*
* Scan bound vs tree entry.
*/
int
Dbtux::cmpScanBound(const Frag& frag, unsigned dir, ConstData boundInfo, unsigned boundCount, TableData entryKey)
{
const DescEnt& descEnt = getDescEnt(frag.m_descPage, frag.m_descOff);
// direction 0-lower 1-upper
ndbrequire(dir <= 1);
// initialize type to equality
unsigned type = 4;
while (boundCount != 0) {
// get and skip bound type
type = boundInfo[0];
boundInfo += 1;
if (! boundInfo.ah().isNULL()) {
if (*entryKey != 0) {
jam();
// current attribute
const unsigned index = boundInfo.ah().getAttributeId();
const DescAttr& descAttr = descEnt.m_descAttr[index];
const NdbSqlUtil::Type& type = NdbSqlUtil::getType(descAttr.m_typeId);
ndbassert(type.m_typeId != NdbSqlUtil::Type::Undefined);
// full data size
const unsigned size1 = AttributeDescriptor::getSizeInWords(descAttr.m_attrDesc);
// compare
const Uint32* const p1 = &boundInfo[AttributeHeaderSize];
const Uint32* const p2 = *entryKey;
int ret = (*type.m_cmp)(p1, p2, size1, size1);
if (ret != 0) {
jam();
return ret;
}
} else {
jam();
// not NULL > NULL
return +1;
}
} else {
jam();
if (*entryKey != 0) {
jam();
// NULL < not NULL
return -1;
}
}
boundInfo += AttributeHeaderSize + boundInfo.ah().getDataSize();
entryKey += 1;
boundCount -= 1;
}
if (dir == 0) {
// lower bound
jam();
if (type == 1) {
jam();
return +1;
}
return -1;
} else {
// upper bound
jam();
if (type == 3) {
jam();
return -1;
}
return +1;
}
}
ndb/src/kernel/blocks/dbtux/DbtuxDebug.cpp
View file @ 23dd2e15
......@@ -207,14 +207,10 @@ Dbtux::printNode(Signal* signal, Frag& frag, NdbOut& out, TupLoc loc, PrintPar&
}
// check ordering within node
for (unsigned j = 1; j < node.getOccup(); j++) {
unsigned start = 0;
const TreeEnt ent1 = node.getEnt(j - 1);
const TreeEnt ent2 = node.getEnt(j);
if (j == 1) {
readKeyAttrs(frag, ent1, start, c_searchKey);
} else {
memcpy(c_searchKey, c_entryKey, frag.m_numAttrs << 2);
}
unsigned start = 0;
readKeyAttrs(frag, ent1, start, c_searchKey);
readKeyAttrs(frag, ent2, start, c_entryKey);
int ret = cmpSearchKey(frag, start, c_searchKey, c_entryKey);
if (ret == 0)
......
ndb/src/kernel/blocks/dbtux/DbtuxGen.cpp
View file @ 23dd2e15
......@@ -16,8 +16,6 @@
#define DBTUX_GEN_CPP
#include "Dbtux.hpp"
#include <signaldata/TuxContinueB.hpp>
#include <signaldata/TuxContinueB.hpp>
Dbtux::Dbtux(const Configuration& conf) :
SimulatedBlock(DBTUX, conf),
......@@ -202,8 +200,9 @@ Dbtux::execREAD_CONFIG_REQ(Signal* signal)
}
// allocate buffers
c_keyAttrs = (Uint32*)allocRecord("c_keyAttrs", sizeof(Uint32), MaxIndexAttributes);
c_searchKey = (TableData)allocRecord("c_searchKey", sizeof(Uint32*), MaxIndexAttributes);
c_entryKey = (TableData)allocRecord("c_entryKey", sizeof(Uint32*), MaxIndexAttributes);
c_sqlCmp = (NdbSqlUtil::Cmp**)allocRecord("c_sqlCmp", sizeof(NdbSqlUtil::Cmp*), MaxIndexAttributes);
c_searchKey = (Uint32*)allocRecord("c_searchKey", sizeof(Uint32*), MaxIndexAttributes);
c_entryKey = (Uint32*)allocRecord("c_entryKey", sizeof(Uint32*), MaxIndexAttributes);
c_dataBuffer = (Uint32*)allocRecord("c_dataBuffer", sizeof(Uint64), (MaxAttrDataSize + 1) >> 1);
// ack
ReadConfigConf * conf = (ReadConfigConf*)signal->getDataPtrSend();
......@@ -218,7 +217,8 @@ Dbtux::execREAD_CONFIG_REQ(Signal* signal)
void
Dbtux::setKeyAttrs(const Frag& frag)
{
Data keyAttrs = c_keyAttrs; // global
Data keyAttrs = c_keyAttrs; // global
NdbSqlUtil::Cmp** sqlCmp = c_sqlCmp; // global
const unsigned numAttrs = frag.m_numAttrs;
const DescEnt& descEnt = getDescEnt(frag.m_descPage, frag.m_descOff);
for (unsigned i = 0; i < numAttrs; i++) {
......@@ -227,75 +227,71 @@ Dbtux::setKeyAttrs(const Frag& frag)
// set attr id and fixed size
keyAttrs.ah() = AttributeHeader(descAttr.m_primaryAttrId, size);
keyAttrs += 1;
// set comparison method pointer
const NdbSqlUtil::Type& sqlType = NdbSqlUtil::getTypeBinary(descAttr.m_typeId);
ndbrequire(sqlType.m_cmp != 0);
*(sqlCmp++) = sqlType.m_cmp;
}
}
void
Dbtux::readKeyAttrs(const Frag& frag, TreeEnt ent, unsigned start, TableData keyData)
Dbtux::readKeyAttrs(const Frag& frag, TreeEnt ent, unsigned start, Data keyData)
{
ConstData keyAttrs = c_keyAttrs; // global
const Uint32 tableFragPtrI = frag.m_tupTableFragPtrI[ent.m_fragBit];
const TupLoc tupLoc = ent.m_tupLoc;
const Uint32 tupVersion = ent.m_tupVersion;
ndbrequire(start < frag.m_numAttrs);
const unsigned numAttrs = frag.m_numAttrs - start;
// start applies to both keys and output data
const Uint32 numAttrs = frag.m_numAttrs - start;
// skip to start position in keyAttrs only
keyAttrs += start;
keyData += start;
c_tup->tuxReadAttrs(tableFragPtrI, tupLoc.m_pageId, tupLoc.m_pageOffset, tupVersion, numAttrs, keyAttrs, keyData);
int ret = c_tup->tuxReadAttrs(tableFragPtrI, tupLoc.m_pageId, tupLoc.m_pageOffset, tupVersion, keyAttrs, numAttrs, keyData);
jamEntry();
// TODO handle error
ndbrequire(ret > 0);
}
void
Dbtux::readTablePk(const Frag& frag, TreeEnt ent, unsigned& pkSize, Data pkData)
Dbtux::readTablePk(const Frag& frag, TreeEnt ent, Data pkData, unsigned& pkSize)
{
const Uint32 tableFragPtrI = frag.m_tupTableFragPtrI[ent.m_fragBit];
const TupLoc tupLoc = ent.m_tupLoc;
Uint32 size = 0;
c_tup->tuxReadKeys(tableFragPtrI, tupLoc.m_pageId, tupLoc.m_pageOffset, &size, pkData);
ndbrequire(size != 0);
pkSize = size;
int ret = c_tup->tuxReadPk(tableFragPtrI, tupLoc.m_pageId, tupLoc.m_pageOffset, pkData);
jamEntry();
// TODO handle error
ndbrequire(ret > 0);
pkSize = ret;
}
/*
* Input is pointers to table attributes. Output is array of attribute
* data with headers. Copies whatever fits.
* Copy attribute data with headers. Input is all index key data.
* Copies whatever fits.
*/
void
Dbtux::copyAttrs(const Frag& frag, TableData data1, Data data2, unsigned maxlen2)
Dbtux::copyAttrs(const Frag& frag, ConstData data1, Data data2, unsigned maxlen2)
{
ConstData keyAttrs = c_keyAttrs; // global
const unsigned numAttrs = frag.m_numAttrs;
unsigned n = frag.m_numAttrs;
unsigned len2 = maxlen2;
for (unsigned n = 0; n < numAttrs; n++) {
while (n != 0) {
jam();
const unsigned attrId = keyAttrs.ah().getAttributeId();
const unsigned dataSize = keyAttrs.ah().getDataSize();
const Uint32* const p1 = *data1;
if (p1 != 0) {
if (len2 == 0)
return;
data2.ah() = AttributeHeader(attrId, dataSize);
data2 += 1;
len2 -= 1;
unsigned n = dataSize;
for (unsigned i = 0; i < dataSize; i++) {
if (len2 == 0)
return;
*data2 = p1[i];
data2 += 1;
len2 -= 1;
}
} else {
const unsigned dataSize = data1.ah().getDataSize();
// copy header
if (len2 == 0)
return;
data2[0] = data1[0];
data1 += 1;
data2 += 1;
len2 -= 1;
// copy data
for (unsigned i = 0; i < dataSize; i++) {
if (len2 == 0)
return;
data2.ah() = AttributeHeader(attrId, 0);
data2.ah().setNULL();
data2 += 1;
data2[i] = data1[i];
len2 -= 1;
}
keyAttrs += 1;
data1 += 1;
data1 += dataSize;
data2 += dataSize;
n -= 1;
}
#ifdef VM_TRACE
memset(data2, DataFillByte, len2 << 2);
......
ndb/src/kernel/blocks/dbtux/DbtuxScan.cpp
View file @ 23dd2e15
......@@ -389,7 +389,7 @@ Dbtux::execACC_CHECK_SCAN(Signal* signal)
jam();
const TreeEnt ent = scan.m_scanPos.m_ent;
// read tuple key
readTablePk(frag, ent, pkSize, pkData);
readTablePk(frag, ent, pkData, pkSize);
// get read lock or exclusive lock
AccLockReq* const lockReq = (AccLockReq*)signal->getDataPtrSend();
lockReq->returnCode = RNIL;
......@@ -480,7 +480,7 @@ Dbtux::execACC_CHECK_SCAN(Signal* signal)
jam();
if (pkSize == 0) {
jam();
readTablePk(frag, ent, pkSize, pkData);
readTablePk(frag, ent, pkData, pkSize);
}
}
// conf signal
......
ndb/src/kernel/blocks/dbtux/DbtuxSearch.cpp
View file @ 23dd2e15
......@@ -25,7 +25,7 @@
* TODO optimize for initial equal attrs in node min/max
*/
void
Dbtux::searchToAdd(Signal* signal, Frag& frag, TableData searchKey, TreeEnt searchEnt, TreePos& treePos)
Dbtux::searchToAdd(Signal* signal, Frag& frag, ConstData searchKey, TreeEnt searchEnt, TreePos& treePos)
{
const TreeHead& tree = frag.m_tree;
const unsigned numAttrs = frag.m_numAttrs;
......@@ -144,7 +144,7 @@ Dbtux::searchToAdd(Signal* signal, Frag& frag, TableData searchKey, TreeEnt sear
* to it.
*/
void
Dbtux::searchToRemove(Signal* signal, Frag& frag, TableData searchKey, TreeEnt searchEnt, TreePos& treePos)
Dbtux::searchToRemove(Signal* signal, Frag& frag, ConstData searchKey, TreeEnt searchEnt, TreePos& treePos)
{
const TreeHead& tree = frag.m_tree;
const unsigned numAttrs = frag.m_numAttrs;
......
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