database/mariadb
1
0
Fork 0
mirror of https://github.com/MariaDB/server.git synced 2025-09-02 09:41:40 +03:00
Code Activity

bug#12220 - ndb - node recovery with charsets

Browse Source 
LQH computes incorrect hash values during NR (as it doesn't concider charsets)

  Solution: make LQH compute correct hash :-)
           1) move xfrm_key into SimulatedBlock so that there's _one_ impl.
           2) make TC, ACC, LQH use same impl.


ndb/include/kernel/AttributeDescriptor.hpp:
  Make SimulatedBlock use AttributeDescriptor (to xfrm)
ndb/src/kernel/blocks/dbacc/Dbacc.hpp:
  Move xfrm handling into SimulatedBlock
ndb/src/kernel/blocks/dbacc/DbaccInit.cpp:
  Move xfrm handling into SimulatedBlock
ndb/src/kernel/blocks/dbacc/DbaccMain.cpp:
  Move xfrm handling into SimulatedBlock
ndb/src/kernel/blocks/dbdict/Dbdict.cpp:
  Move xfrm handling into SimulatedBlock
ndb/src/kernel/blocks/dblqh/Dblqh.hpp:
  Move xfrm handling into SimulatedBlock
ndb/src/kernel/blocks/dblqh/DblqhMain.cpp:
  Move xfrm handling into SimulatedBlock
ndb/src/kernel/blocks/dbtc/Dbtc.hpp:
  Move xfrm handling into SimulatedBlock
ndb/src/kernel/blocks/dbtc/DbtcMain.cpp:
  Move xfrm handling into SimulatedBlock
ndb/src/kernel/vm/SimulatedBlock.cpp:
  Move xfrm handling into SimulatedBlock
ndb/src/kernel/vm/SimulatedBlock.hpp:
  Move xfrm handling into SimulatedBlock
This commit is contained in:
unknown
2005-08-22 10:59:25 +02:00
parent ad56f835d7
commit 9349800946
11 changed files with 237 additions and 318 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
ndb/include/kernel/AttributeDescriptor.hpp
View File

@@ -23,6 +23,7 @@ class AttributeDescriptor {
friend class Dbacc;
friend class Dbtup;
friend class Dbtux;
friend class SimulatedBlock;
private:
static void setType(Uint32 &, Uint32 type);

8
ndb/src/kernel/blocks/dbacc/Dbacc.hpp
View File

@@ -851,13 +851,6 @@ struct Tabrec {
Uint32 fragptrholder[MAX_FRAG_PER_NODE];
Uint32 tabUserPtr;
BlockReference tabUserRef;
Uint8 noOfKeyAttr;
Uint8 hasCharAttr;
struct KeyAttr {
Uint32 attributeDescriptor;
CHARSET_INFO* charsetInfo;
} keyAttr[MAX_ATTRIBUTES_IN_INDEX];
};
typedef Ptr<Tabrec> TabrecPtr;
@@ -903,7 +896,6 @@ private:
void execACCKEYREQ(Signal* signal);
void execACCSEIZEREQ(Signal* signal);
void execACCFRAGREQ(Signal* signal);
void execTC_SCHVERREQ(Signal* signal);
void execACC_SRREQ(Signal* signal);
void execNEXT_SCANREQ(Signal* signal);
void execACC_ABORTREQ(Signal* signal);

1
ndb/src/kernel/blocks/dbacc/DbaccInit.cpp
View File

@@ -179,7 +179,6 @@ Dbacc::Dbacc(const class Configuration & conf):
addRecSignal(GSN_ACCKEYREQ, &Dbacc::execACCKEYREQ);
addRecSignal(GSN_ACCSEIZEREQ, &Dbacc::execACCSEIZEREQ);
addRecSignal(GSN_ACCFRAGREQ, &Dbacc::execACCFRAGREQ);
addRecSignal(GSN_TC_SCHVERREQ, &Dbacc::execTC_SCHVERREQ);
addRecSignal(GSN_ACC_SRREQ, &Dbacc::execACC_SRREQ);
addRecSignal(GSN_NEXT_SCANREQ, &Dbacc::execNEXT_SCANREQ);
addRecSignal(GSN_ACC_ABORTREQ, &Dbacc::execACC_ABORTREQ);

129
ndb/src/kernel/blocks/dbacc/DbaccMain.cpp
View File

@@ -28,7 +28,8 @@
#include <signaldata/FsRemoveReq.hpp>
#include <signaldata/DropTab.hpp>
#include <signaldata/DumpStateOrd.hpp>
#include <SectionReader.hpp>
#include <KeyDescriptor.hpp>
// TO_DO_RONM is a label for comments on what needs to be improved in future versions
// when more time is given.
@@ -1037,12 +1038,6 @@ void Dbacc::initialiseTableRec(Signal* signal)
tabptr.p->fragholder[i] = RNIL;
tabptr.p->fragptrholder[i] = RNIL;
}//for
tabptr.p->noOfKeyAttr = 0;
tabptr.p->hasCharAttr = 0;
for (Uint32 k = 0; k < MAX_ATTRIBUTES_IN_INDEX; k++) {
tabptr.p->keyAttr[k].attributeDescriptor = 0;
tabptr.p->keyAttr[k].charsetInfo = 0;
}
}//for
}//Dbacc::initialiseTableRec()
@@ -1172,8 +1167,8 @@ void Dbacc::execACCFRAGREQ(Signal* signal)
Uint32 userPtr = req->userPtr;
BlockReference retRef = req->userRef;
rootfragrecptr.p->rootState = ACTIVEROOT;
AccFragConf * const conf = (AccFragConf*)&signal->theData[0];
AccFragConf * const conf = (AccFragConf*)&signal->theData[0];
conf->userPtr = userPtr;
conf->rootFragPtr = rootfragrecptr.i;
conf->fragId[0] = rootfragrecptr.p->fragmentid[0];
@@ -1197,65 +1192,6 @@ void Dbacc::addFragRefuse(Signal* signal, Uint32 errorCode)
return;
}//Dbacc::addFragRefuseEarly()
void
Dbacc::execTC_SCHVERREQ(Signal* signal)
{
jamEntry();
if (! assembleFragments(signal)) {
jam();
return;
}
tabptr.i = signal->theData[0];
ptrCheckGuard(tabptr, ctablesize, tabrec);
Uint32 noOfKeyAttr = signal->theData[6];
ndbrequire(noOfKeyAttr <= MAX_ATTRIBUTES_IN_INDEX);
Uint32 hasCharAttr = 0;
SegmentedSectionPtr s0Ptr;
signal->getSection(s0Ptr, 0);
SectionReader r0(s0Ptr, getSectionSegmentPool());
Uint32 i = 0;
while (i < noOfKeyAttr) {
jam();
Uint32 attributeDescriptor = ~0;
Uint32 csNumber = ~0;
if (! r0.getWord(&attributeDescriptor) ||
! r0.getWord(&csNumber)) {
jam();
break;
}
CHARSET_INFO* cs = 0;
if (csNumber != 0) {
cs = all_charsets[csNumber];
ndbrequire(cs != 0);
hasCharAttr = 1;
}
tabptr.p->keyAttr[i].attributeDescriptor = attributeDescriptor;
tabptr.p->keyAttr[i].charsetInfo = cs;
i++;
}
ndbrequire(i == noOfKeyAttr);
releaseSections(signal);
tabptr.p->noOfKeyAttr = noOfKeyAttr;
tabptr.p->hasCharAttr = hasCharAttr;
// copy char attr flag to each fragment
for (Uint32 i1 = 0; i1 < MAX_FRAG_PER_NODE; i1++) {
jam();
if (tabptr.p->fragptrholder[i1] != RNIL) {
rootfragrecptr.i = tabptr.p->fragptrholder[i1];
ptrCheckGuard(rootfragrecptr, crootfragmentsize, rootfragmentrec);
for (Uint32 i2 = 0; i2 < 2; i2++) {
fragrecptr.i = rootfragrecptr.p->fragmentptr[i2];
ptrCheckGuard(fragrecptr, cfragmentsize, fragmentrec);
fragrecptr.p->hasCharAttr = hasCharAttr;
}
}
}
// no reply to DICT
}
void
Dbacc::execDROP_TAB_REQ(Signal* signal){
@@ -1841,55 +1777,14 @@ void Dbacc::execACCKEYREQ(Signal* signal)
void
Dbacc::xfrmKeyData(Signal* signal)
{
tabptr.i = fragrecptr.p->myTableId;
ptrCheckGuard(tabptr, ctablesize, tabrec);
Uint32 dst[1024 * MAX_XFRM_MULTIPLY];
Uint32 dstSize = (sizeof(dst) >> 2);
Uint32 table = fragrecptr.p->myTableId;
Uint32 dst[MAX_KEY_SIZE_IN_WORDS * MAX_XFRM_MULTIPLY];
Uint32 keyPartLen[MAX_ATTRIBUTES_IN_INDEX];
Uint32* src = &signal->theData[7];
const Uint32 noOfKeyAttr = tabptr.p->noOfKeyAttr;
Uint32 dstPos = 0;
Uint32 srcPos = 0;
Uint32 i = 0;
while (i < noOfKeyAttr) {
const Tabrec::KeyAttr& keyAttr = tabptr.p->keyAttr[i];
Uint32 srcBytes = AttributeDescriptor::getSizeInBytes(keyAttr.attributeDescriptor);
Uint32 srcWords = (srcBytes + 3) / 4;
Uint32 dstWords = ~0;
uchar* dstPtr = (uchar*)&dst[dstPos];
const uchar* srcPtr = (const uchar*)&src[srcPos];
CHARSET_INFO* cs = keyAttr.charsetInfo;
if (cs == 0) {
jam();
memcpy(dstPtr, srcPtr, srcWords << 2);
dstWords = srcWords;
} else {
jam();
Uint32 typeId = AttributeDescriptor::getType(keyAttr.attributeDescriptor);
Uint32 lb, len;
bool ok = NdbSqlUtil::get_var_length(typeId, srcPtr, srcBytes, lb, len);
ndbrequire(ok);
Uint32 xmul = cs->strxfrm_multiply;
if (xmul == 0)
xmul = 1;
// see comment in DbtcMain.cpp
Uint32 dstLen = xmul * (srcBytes - lb);
ndbrequire(dstLen <= ((dstSize - dstPos) << 2));
int n = NdbSqlUtil::strnxfrm_bug7284(cs, dstPtr, dstLen, srcPtr + lb, len);
ndbrequire(n != -1);
while ((n & 3) != 0)
dstPtr[n++] = 0;
dstWords = (n >> 2);
}
dstPos += dstWords;
srcPos += srcWords;
i++;
}
memcpy(src, dst, dstPos << 2);
operationRecPtr.p->xfrmtupkeylen = dstPos;
Uint32 len = xfrm_key(table, src, dst, sizeof(dst) >> 2, keyPartLen);
ndbrequire(len); // 0 means error
memcpy(src, dst, len << 2);
operationRecPtr.p->xfrmtupkeylen = len;
}
void Dbacc::accIsLockedLab(Signal* signal)
@@ -8024,6 +7919,10 @@ void Dbacc::initFragAdd(Signal* signal,
Uint32 Tmp2 = regFragPtr.p->maxloadfactor - regFragPtr.p->minloadfactor;
Tmp2 = Tmp1 * Tmp2;
regFragPtr.p->slackCheck = Tmp2;
Uint32 hasCharAttr = g_key_descriptor_pool.getPtr(req->tableId)->hasCharAttr;
regFragPtr.p->hasCharAttr = hasCharAttr;
}//Dbacc::initFragAdd()
void Dbacc::initFragGeneral(FragmentrecPtr regFragPtr)

64
ndb/src/kernel/blocks/dbdict/Dbdict.cpp
View File

@@ -27,6 +27,7 @@
#include <SectionReader.hpp>
#include <SimpleProperties.hpp>
#include <AttributeHeader.hpp>
#include <KeyDescriptor.hpp>
#include <signaldata/DictSchemaInfo.hpp>
#include <signaldata/DictTabInfo.hpp>
#include <signaldata/DropTabFile.hpp>
@@ -1750,6 +1751,7 @@ void Dbdict::execREAD_CONFIG_REQ(Signal* signal)
c_schemaPageRecordArray.setSize(2 * NDB_SF_MAX_PAGES);
c_tableRecordPool.setSize(tablerecSize);
c_tableRecordHash.setSize(tablerecSize);
g_key_descriptor_pool.setSize(tablerecSize);
c_triggerRecordPool.setSize(c_maxNoOfTriggers);
c_triggerRecordHash.setSize(c_maxNoOfTriggers);
c_opRecordPool.setSize(256); // XXX need config params
@@ -4450,6 +4452,44 @@ Dbdict::execADD_FRAGREQ(Signal* signal) {
sendSignal(DBLQH_REF, GSN_LQHFRAGREQ, signal,
LqhFragReq::SignalLength, JBB);
}
/**
* Create KeyDescriptor
*/
KeyDescriptor* desc= g_key_descriptor_pool.getPtr(tabPtr.i);
new (desc) KeyDescriptor();
Uint32 key = 0;
Uint32 tAttr = tabPtr.p->firstAttribute;
while (tAttr != RNIL)
{
jam();
AttributeRecord* aRec = c_attributeRecordPool.getPtr(tAttr);
if (aRec->tupleKey)
{
desc->noOfKeyAttr ++;
desc->keyAttr[key].attributeDescriptor = aRec->attributeDescriptor;
Uint32 csNumber = (aRec->extPrecision >> 16);
if(csNumber)
{
desc->keyAttr[key].charsetInfo = all_charsets[csNumber];
ndbrequire(all_charsets[csNumber]);
desc->hasCharAttr = 1;
}
else
{
desc->keyAttr[key].charsetInfo = 0;
}
if(AttributeDescriptor::getDKey(aRec->attributeDescriptor))
{
desc->noOfDistrKeys ++;
}
key++;
}
tAttr = aRec->nextAttrInTable;
}
ndbrequire(key == tabPtr.p->noOfPrimkey);
}
void
@@ -4645,27 +4685,7 @@ Dbdict::execTAB_COMMITCONF(Signal* signal){
signal->theData[5] = createTabPtr.p->key;
signal->theData[6] = (Uint32)tabPtr.p->noOfPrimkey;
Uint32 buf[2 * MAX_ATTRIBUTES_IN_INDEX];
Uint32 sz = 0;
Uint32 tAttr = tabPtr.p->firstAttribute;
while (tAttr != RNIL) {
jam();
AttributeRecord* aRec = c_attributeRecordPool.getPtr(tAttr);
if (aRec->tupleKey) {
buf[sz++] = aRec->attributeDescriptor;
buf[sz++] = (aRec->extPrecision >> 16); // charset number
}
tAttr = aRec->nextAttrInTable;
}
ndbrequire((int)sz == 2 * tabPtr.p->noOfPrimkey);
LinearSectionPtr lsPtr[3];
lsPtr[0].p = buf;
lsPtr[0].sz = sz;
// note: ACC does not reply
if (tabPtr.p->isTable() || tabPtr.p->isHashIndex())
sendSignal(DBACC_REF, GSN_TC_SCHVERREQ, signal, 7, JBB, lsPtr, 1);
sendSignal(DBTC_REF, GSN_TC_SCHVERREQ, signal, 7, JBB, lsPtr, 1);
sendSignal(DBTC_REF, GSN_TC_SCHVERREQ, signal, 7, JBB);
return;
}
@@ -12342,3 +12362,5 @@ Dbdict::getMetaAttribute(MetaData::Attribute& attr, const MetaData::Table& table
new (&attr) MetaData::Attribute(*attrPtr.p);
return 0;
}
CArray<KeyDescriptor> g_key_descriptor_pool;

2
ndb/src/kernel/blocks/dblqh/Dblqh.hpp
View File

@@ -2233,7 +2233,7 @@ private:
void LQHKEY_abort(Signal* signal, int errortype);
void LQHKEY_error(Signal* signal, int errortype);
void nextRecordCopy(Signal* signal);
void calculateHash(Signal* signal);
Uint32 calculateHash(Uint32 tableId, const Uint32* src);
void continueAfterCheckLcpStopBlocked(Signal* signal);
void checkLcpStopBlockedLab(Signal* signal);
void sendCommittedTc(Signal* signal, BlockReference atcBlockref);

55
ndb/src/kernel/blocks/dblqh/DblqhMain.cpp
View File

@@ -55,6 +55,7 @@
#include <signaldata/AlterTab.hpp>
#include <signaldata/LCP.hpp>
#include <KeyDescriptor.hpp>
// Use DEBUG to print messages that should be
// seen only when we debug the product
@@ -9013,44 +9014,17 @@ void Dblqh::sendScanFragConf(Signal* signal, Uint32 scanCompleted)
/* FRAGMENT TO A NEW REPLICA OF THE FRAGMENT. IT DOES ALSO SHUT DOWN ALL */
/* CONNECTIONS TO THE FAILED NODE. */
/*---------------------------------------------------------------------------*/
void Dblqh::calculateHash(Signal* signal)
Uint32
Dblqh::calculateHash(Uint32 tableId, const Uint32* src)
{
DatabufPtr locDatabufptr;
UintR Ti;
UintR Tdata0;
UintR Tdata1;
UintR Tdata2;
UintR Tdata3;
UintR* Tdata32;
Uint64 Tdata[512];
jam();
Uint64 Tmp[(MAX_KEY_SIZE_IN_WORDS*MAX_XFRM_MULTIPLY) >> 1];
Uint32 keyPartLen[MAX_ATTRIBUTES_IN_INDEX];
Uint32 keyLen = xfrm_key(tableId, src, (Uint32*)Tmp, sizeof(Tmp) >> 2,
keyPartLen);
ndbrequire(keyLen);
Tdata32 = (UintR*)&Tdata[0];
Tdata0 = tcConnectptr.p->tupkeyData[0];
Tdata1 = tcConnectptr.p->tupkeyData[1];
Tdata2 = tcConnectptr.p->tupkeyData[2];
Tdata3 = tcConnectptr.p->tupkeyData[3];
Tdata32[0] = Tdata0;
Tdata32[1] = Tdata1;
Tdata32[2] = Tdata2;
Tdata32[3] = Tdata3;
locDatabufptr.i = tcConnectptr.p->firstTupkeybuf;
Ti = 4;
while (locDatabufptr.i != RNIL) {
ptrCheckGuard(locDatabufptr, cdatabufFileSize, databuf);
Tdata0 = locDatabufptr.p->data[0];
Tdata1 = locDatabufptr.p->data[1];
Tdata2 = locDatabufptr.p->data[2];
Tdata3 = locDatabufptr.p->data[3];
Tdata32[Ti ] = Tdata0;
Tdata32[Ti + 1] = Tdata1;
Tdata32[Ti + 2] = Tdata2;
Tdata32[Ti + 3] = Tdata3;
locDatabufptr.i = locDatabufptr.p->nextDatabuf;
Ti += 4;
}//while
tcConnectptr.p->hashValue =
md5_hash((Uint64*)&Tdata32[0], (UintR)tcConnectptr.p->primKeyLen);
return md5_hash(Tmp, keyLen);
}//Dblqh::calculateHash()
/* *************************************** */
@@ -9384,7 +9358,7 @@ void Dblqh::copyTupkeyConfLab(Signal* signal)
const TupKeyConf * const tupKeyConf = (TupKeyConf *)signal->getDataPtr();
UintR readLength = tupKeyConf->readLength;
Uint32 tableId = tcConnectptr.p->tableref;
scanptr.i = tcConnectptr.p->tcScanRec;
c_scanRecordPool.getPtr(scanptr);
ScanRecord* scanP = scanptr.p;
@@ -9411,7 +9385,14 @@ void Dblqh::copyTupkeyConfLab(Signal* signal)
Uint32 len= tcConnectptr.p->primKeyLen = readPrimaryKeys(scanP, tcConP, tmp);
// Calculate hash (no need to linearies key)
if (g_key_descriptor_pool.getPtr(tableId)->hasCharAttr)
{
tcConnectptr.p->hashValue = calculateHash(tableId, tmp);
}
else
{
tcConnectptr.p->hashValue = md5_hash((Uint64*)tmp, len);
}
// Move into databuffer to make packLqhkeyreqLab happy
memcpy(tcConP->tupkeyData, tmp, 4*4);

5
ndb/src/kernel/blocks/dbtc/Dbtc.hpp
View File

@@ -962,11 +962,6 @@ public:
Uint8 hasCharAttr;
Uint8 noOfDistrKeys;
struct KeyAttr {
Uint32 attributeDescriptor;
CHARSET_INFO* charsetInfo;
} keyAttr[MAX_ATTRIBUTES_IN_INDEX];
bool checkTable(Uint32 schemaVersion) const {
return enabled && !dropping &&
(table_version_major(schemaVersion) == table_version_major(currentSchemaVersion));

143
ndb/src/kernel/blocks/dbtc/DbtcMain.cpp
View File

@@ -66,6 +66,7 @@
#include <signaldata/DictTabInfo.hpp>
#include <AttributeDescriptor.hpp>
#include <SectionReader.hpp>
#include <KeyDescriptor.hpp>
#include <NdbOut.hpp>
#include <DebuggerNames.hpp>
@@ -329,42 +330,16 @@ void Dbtc::execTC_SCHVERREQ(Signal* signal)
BlockReference retPtr = signal->theData[5];
Uint32 noOfKeyAttr = signal->theData[6];
ndbrequire(noOfKeyAttr <= MAX_ATTRIBUTES_IN_INDEX);
Uint32 hasCharAttr = 0;
Uint32 noOfDistrKeys = 0;
SegmentedSectionPtr s0Ptr;
signal->getSection(s0Ptr, 0);
SectionReader r0(s0Ptr, getSectionSegmentPool());
Uint32 i = 0;
while (i < noOfKeyAttr) {
jam();
Uint32 attributeDescriptor = ~0;
Uint32 csNumber = ~0;
if (! r0.getWord(&attributeDescriptor) ||
! r0.getWord(&csNumber)) {
jam();
break;
}
CHARSET_INFO* cs = 0;
if (csNumber != 0) {
cs = all_charsets[csNumber];
ndbrequire(cs != 0);
hasCharAttr = 1;
}
noOfDistrKeys += AttributeDescriptor::getDKey(attributeDescriptor);
tabptr.p->keyAttr[i].attributeDescriptor = attributeDescriptor;
tabptr.p->keyAttr[i].charsetInfo = cs;
i++;
}
ndbrequire(i == noOfKeyAttr);
releaseSections(signal);
const KeyDescriptor* desc = g_key_descriptor_pool.getPtr(tabptr.i);
ndbrequire(noOfKeyAttr == desc->noOfKeyAttr);
ndbrequire(tabptr.p->enabled == false);
tabptr.p->enabled = true;
tabptr.p->dropping = false;
tabptr.p->noOfKeyAttr = noOfKeyAttr;
tabptr.p->hasCharAttr = hasCharAttr;
tabptr.p->noOfDistrKeys = noOfDistrKeys;
tabptr.p->noOfKeyAttr = desc->noOfKeyAttr;
tabptr.p->hasCharAttr = desc->hasCharAttr;
tabptr.p->noOfDistrKeys = desc->noOfDistrKeys;
signal->theData[0] = tabptr.i;
signal->theData[1] = retPtr;
@@ -2323,113 +2298,37 @@ Dbtc::handle_special_hash(Uint32 dstHash[4], Uint32* src, Uint32 srcLen,
Uint32 tabPtrI,
bool distr)
{
Uint64 Tmp[MAX_KEY_SIZE_IN_WORDS * 4 * MAX_XFRM_MULTIPLY];
const Uint32 dstSize = sizeof(Tmp) / 4;
Uint64 Tmp[MAX_KEY_SIZE_IN_WORDS * MAX_XFRM_MULTIPLY];
const TableRecord* tabPtrP = &tableRecord[tabPtrI];
const Uint32 noOfKeyAttr = tabPtrP->noOfKeyAttr;
Uint32 noOfDistrKeys = tabPtrP->noOfDistrKeys;
const bool hasCharAttr = tabPtrP->hasCharAttr;
const bool hasDistKeys = tabPtrP->noOfDistrKeys > 0;
Uint32 *dst = (Uint32*)Tmp;
Uint32 dstPos = 0;
Uint32 srcPos = 0;
Uint32 keyPartLen[MAX_ATTRIBUTES_IN_INDEX];
if(hasCharAttr){
Uint32 i = 0;
while (i < noOfKeyAttr) {
const TableRecord::KeyAttr& keyAttr = tabPtrP->keyAttr[i];
Uint32 srcBytes =
AttributeDescriptor::getSizeInBytes(keyAttr.attributeDescriptor);
Uint32 srcWords = (srcBytes + 3) / 4;
Uint32 dstWords = ~0;
uchar* dstPtr = (uchar*)&dst[dstPos];
const uchar* srcPtr = (const uchar*)&src[srcPos];
CHARSET_INFO* cs = keyAttr.charsetInfo;
if (cs == NULL) {
jam();
memcpy(dstPtr, srcPtr, srcWords << 2);
dstWords = srcWords;
} else {
jam();
Uint32 typeId =
AttributeDescriptor::getType(keyAttr.attributeDescriptor);
Uint32 lb, len;
bool ok = NdbSqlUtil::get_var_length(typeId, srcPtr, srcBytes, lb, len);
ndbrequire(ok);
Uint32 xmul = cs->strxfrm_multiply;
if (xmul == 0)
xmul = 1;
/*
* Varchar is really Char. End spaces do not matter. To get
* same hash we blank-pad to maximum length via strnxfrm.
* TODO use MySQL charset-aware hash function instead
*/
Uint32 dstLen = xmul * (srcBytes - lb);
ndbrequire(dstLen <= ((dstSize - dstPos) << 2));
int n = NdbSqlUtil::strnxfrm_bug7284(cs, dstPtr, dstLen, srcPtr + lb, len);
ndbrequire(n != -1);
while ((n & 3) != 0) {
dstPtr[n++] = 0;
}
dstWords = (n >> 2);
}
dstPos += dstWords;
srcPos += srcWords;
keyPartLen[i++] = dstWords;
}
Uint32 * keyPartLenPtr;
if(hasCharAttr)
{
keyPartLenPtr = keyPartLen;
dstPos = xfrm_key(tabPtrI, src, dst, sizeof(Tmp) >> 2, keyPartLenPtr);
ndbrequire(dstPos);
}
else
{
dst = src;
dstPos = srcLen;
keyPartLenPtr = 0;
}
md5_hash(dstHash, (Uint64*)dst, dstPos);
if(distr && noOfDistrKeys)
if(distr && hasDistKeys)
{
jam();
src = dst;
dstPos = 0;
Uint32 i = 0;
if(hasCharAttr)
{
while (i < noOfKeyAttr && noOfDistrKeys)
{
const TableRecord::KeyAttr& keyAttr = tabPtrP->keyAttr[i];
Uint32 len = keyPartLen[i];
if(AttributeDescriptor::getDKey(keyAttr.attributeDescriptor))
{
noOfDistrKeys--;
memmove(dst+dstPos, src, len << 2);
dstPos += len;
}
src += len;
i++;
}
}
else
{
while (i < noOfKeyAttr && noOfDistrKeys)
{
const TableRecord::KeyAttr& keyAttr = tabPtrP->keyAttr[i];
Uint32 len =
AttributeDescriptor::getSizeInBytes(keyAttr.attributeDescriptor);
len = (len + 3) / 4;
if(AttributeDescriptor::getDKey(keyAttr.attributeDescriptor))
{
noOfDistrKeys--;
memmove(dst+dstPos, src, len << 2);
dstPos += len;
}
src += len;
i++;
}
}
Uint32 tmp[4];
md5_hash(tmp, (Uint64*)dst, dstPos);
Uint32 len = create_distr_key(tabPtrI, dst, keyPartLenPtr);
md5_hash(tmp, (Uint64*)dst, len);
dstHash[1] = tmp[1];
}
return true; // success
@@ -10204,10 +10103,6 @@ void Dbtc::initTable(Signal* signal)
tabptr.p->noOfKeyAttr = 0;
tabptr.p->hasCharAttr = 0;
tabptr.p->noOfDistrKeys = 0;
for (unsigned k = 0; k < MAX_ATTRIBUTES_IN_INDEX; k++) {
tabptr.p->keyAttr[k].attributeDescriptor = 0;
tabptr.p->keyAttr[k].charsetInfo = 0;
}
}//for
}//Dbtc::initTable()

115
ndb/src/kernel/vm/SimulatedBlock.cpp
View File

@@ -1802,3 +1802,118 @@ SimulatedBlock::init_globals_list(void ** tmp, size_t cnt){
}
#endif
#include "KeyDescriptor.hpp"
Uint32
SimulatedBlock::xfrm_key(Uint32 tab, const Uint32* src,
Uint32 *dst, Uint32 dstSize,
Uint32 keyPartLen[MAX_ATTRIBUTES_IN_INDEX]) const
{
const KeyDescriptor * desc = g_key_descriptor_pool.getPtr(tab);
const Uint32 noOfKeyAttr = desc->noOfKeyAttr;
Uint32 i = 0;
Uint32 srcPos = 0;
Uint32 dstPos = 0;
while (i < noOfKeyAttr)
{
const KeyDescriptor::KeyAttr& keyAttr = desc->keyAttr[i];
Uint32 srcBytes =
AttributeDescriptor::getSizeInBytes(keyAttr.attributeDescriptor);
Uint32 srcWords = (srcBytes + 3) / 4;
Uint32 dstWords = ~0;
uchar* dstPtr = (uchar*)&dst[dstPos];
const uchar* srcPtr = (const uchar*)&src[srcPos];
CHARSET_INFO* cs = keyAttr.charsetInfo;
if (cs == NULL)
{
jam();
memcpy(dstPtr, srcPtr, srcWords << 2);
dstWords = srcWords;
}
else
{
jam();
Uint32 typeId =
AttributeDescriptor::getType(keyAttr.attributeDescriptor);
Uint32 lb, len;
bool ok = NdbSqlUtil::get_var_length(typeId, srcPtr, srcBytes, lb, len);
ndbrequire(ok);
Uint32 xmul = cs->strxfrm_multiply;
if (xmul == 0)
xmul = 1;
/*
* Varchar is really Char. End spaces do not matter. To get
* same hash we blank-pad to maximum length via strnxfrm.
* TODO use MySQL charset-aware hash function instead
*/
Uint32 dstLen = xmul * (srcBytes - lb);
ndbrequire(dstLen <= ((dstSize - dstPos) << 2));
int n = NdbSqlUtil::strnxfrm_bug7284(cs, dstPtr, dstLen, srcPtr + lb, len);
ndbrequire(n != -1);
while ((n & 3) != 0)
{
dstPtr[n++] = 0;
}
dstWords = (n >> 2);
}
dstPos += dstWords;
srcPos += srcWords;
keyPartLen[i++] = dstWords;
}
return dstPos;
}
Uint32
SimulatedBlock::create_distr_key(Uint32 tableId,
Uint32 *data,
const Uint32
keyPartLen[MAX_ATTRIBUTES_IN_INDEX]) const
{
const KeyDescriptor* desc = g_key_descriptor_pool.getPtr(tableId);
const Uint32 noOfKeyAttr = desc->noOfKeyAttr;
Uint32 noOfDistrKeys = desc->noOfDistrKeys;
Uint32 *src = data;
Uint32 *dst = data;
Uint32 i = 0;
Uint32 dstPos = 0;
if(keyPartLen)
{
while (i < noOfKeyAttr && noOfDistrKeys)
{
Uint32 attr = desc->keyAttr[i].attributeDescriptor;
Uint32 len = keyPartLen[i];
if(AttributeDescriptor::getDKey(attr))
{
noOfDistrKeys--;
memmove(dst+dstPos, src, len << 2);
dstPos += len;
}
src += len;
i++;
}
}
else
{
while (i < noOfKeyAttr && noOfDistrKeys)
{
Uint32 attr = desc->keyAttr[i].attributeDescriptor;
Uint32 len = AttributeDescriptor::getSizeInWords(attr);
if(AttributeDescriptor::getDKey(attr))
{
noOfDistrKeys--;
memmove(dst+dstPos, src, len << 2);
dstPos += len;
}
src += len;
i++;
}
}
return dstPos;
}

20
ndb/src/kernel/vm/SimulatedBlock.hpp
View File

@@ -20,11 +20,13 @@
#include <NdbTick.h>
#include <kernel_types.h>
#include <ndb_version.h>
#include <ndb_limits.h>
#include "VMSignal.hpp"
#include <RefConvert.hpp>
#include <BlockNumbers.h>
#include <GlobalSignalNumbers.h>
#include "pc.hpp"
#include <NodeInfo.hpp>
#include <NodeState.hpp>
@@ -386,6 +388,24 @@ protected:
const NodeInfo & getNodeInfo(NodeId nodeId) const;
NodeInfo & setNodeInfo(NodeId);
/**********************
* Xfrm stuff
*/
/**
* @return length
*/
Uint32 xfrm_key(Uint32 tab, const Uint32* src,
Uint32 *dst, Uint32 dstLen,
Uint32 keyPartLen[MAX_ATTRIBUTES_IN_INDEX]) const;
/**
*
*/
Uint32 create_distr_key(Uint32 tableId,
Uint32 *data,
const Uint32 keyPaLen[MAX_ATTRIBUTES_IN_INDEX])const;
private:
NewVARIABLE* NewVarRef; /* New Base Address Table for block */
Uint16 theBATSize; /* # entries in BAT */