MySQL数据库

开发平台：

Visual C++

Dbdict.cpp：源码内容

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_global.h>
#include <my_sys.h>
#define DBDICT_C
#include "Dbdict.hpp"
#include <ndb_limits.h>
#include <NdbOut.hpp>
#include <Properties.hpp>
#include <Configuration.hpp>
#include <SectionReader.hpp>
#include <SimpleProperties.hpp>
#include <AttributeHeader.hpp>
#include <signaldata/DictSchemaInfo.hpp>
#include <signaldata/DictTabInfo.hpp>
#include <signaldata/DropTabFile.hpp>
#include <signaldata/EventReport.hpp>
#include <signaldata/FsCloseReq.hpp>
#include <signaldata/FsConf.hpp>
#include <signaldata/FsOpenReq.hpp>
#include <signaldata/FsReadWriteReq.hpp>
#include <signaldata/FsRef.hpp>
#include <signaldata/GetTabInfo.hpp>
#include <signaldata/GetTableId.hpp>
#include <signaldata/HotSpareRep.hpp>
#include <signaldata/NFCompleteRep.hpp>
#include <signaldata/NodeFailRep.hpp>
#include <signaldata/ReadNodesConf.hpp>
#include <signaldata/RelTabMem.hpp>
#include <signaldata/WaitGCP.hpp>
#include <signaldata/ListTables.hpp>
#include <signaldata/CreateTrig.hpp>
#include <signaldata/AlterTrig.hpp>
#include <signaldata/DropTrig.hpp>
#include <signaldata/CreateIndx.hpp>
#include <signaldata/DropIndx.hpp>
#include <signaldata/BuildIndx.hpp>
#include <signaldata/CreateEvnt.hpp>
#include <signaldata/UtilPrepare.hpp>
#include <signaldata/UtilExecute.hpp>
#include <signaldata/UtilRelease.hpp>
#include <signaldata/SumaImpl.hpp>
#include <GrepError.hpp>
//#include <signaldata/DropEvnt.hpp>
#include <signaldata/LqhFrag.hpp>
#include <signaldata/DiAddTab.hpp>
#include <signaldata/DihStartTab.hpp>
#include <signaldata/DropTable.hpp>
#include <signaldata/DropTab.hpp>
#include <signaldata/PrepDropTab.hpp>
#include <signaldata/CreateTable.hpp>
#include <signaldata/AlterTable.hpp>
#include <signaldata/AlterTab.hpp>
#include <signaldata/CreateFragmentation.hpp>
#include <signaldata/CreateTab.hpp>
#include <NdbSleep.h>
#include <signaldata/ApiBroadcast.hpp>
#define ZNOT_FOUND 626
#define ZALREADYEXIST 630
//#define EVENT_PH2_DEBUG
//#define EVENT_PH3_DEBUG
//#define EVENT_DEBUG
#define EVENT_TRACE
// ndbout_c("Event debug trace: File: %s Line: %u", __FILE__, __LINE__)
#define DIV(x,y) (((x)+(y)-1)/(y))
#include <ndb_version.h>
static
Uint32
alter_table_inc_schema_version(Uint32 old)
{
return (old & 0x00FFFFFF) + ((old + 0x1000000) & 0xFF000000);
}
static
Uint32
alter_table_dec_schema_version(Uint32 old)
{
return (old & 0x00FFFFFF) + ((old - 0x1000000) & 0xFF000000);
}
static
Uint32
create_table_inc_schema_version(Uint32 old)
{
return (old + 0x00000001) & 0x00FFFFFF;
}
/* **************************************************************** */
/* ---------------------------------------------------------------- */
/* MODULE: GENERAL MODULE -------------------------------- */
/* ---------------------------------------------------------------- */
/* */
/* This module contains general stuff. Mostly debug signals and */
/* general signals that go into a specific module after checking a */
/* state variable. Also general subroutines used by many. */
/* ---------------------------------------------------------------- */
/* **************************************************************** */
/* ---------------------------------------------------------------- */
// This signal is used to dump states of various variables in the
// block by command.
/* ---------------------------------------------------------------- */
void
Dbdict::execDUMP_STATE_ORD(Signal* signal)
{
jamEntry();
#ifdef VM_TRACE
if(signal->theData[0] == 1222){
const Uint32 tab = signal->theData[1];
PrepDropTabReq* req = (PrepDropTabReq*)signal->getDataPtr();
req->senderRef = reference();
req->senderData = 1222;
req->tableId = tab;
sendSignal(DBLQH_REF, GSN_PREP_DROP_TAB_REQ, signal,
PrepDropTabReq::SignalLength, JBB);
}
if(signal->theData[0] == 1223){
const Uint32 tab = signal->theData[1];
PrepDropTabReq* req = (PrepDropTabReq*)signal->getDataPtr();
req->senderRef = reference();
req->senderData = 1222;
req->tableId = tab;
sendSignal(DBTC_REF, GSN_PREP_DROP_TAB_REQ, signal,
PrepDropTabReq::SignalLength, JBB);
}
if(signal->theData[0] == 1224){
const Uint32 tab = signal->theData[1];
PrepDropTabReq* req = (PrepDropTabReq*)signal->getDataPtr();
req->senderRef = reference();
req->senderData = 1222;
req->tableId = tab;
sendSignal(DBDIH_REF, GSN_PREP_DROP_TAB_REQ, signal,
PrepDropTabReq::SignalLength, JBB);
}
if(signal->theData[0] == 1225){
const Uint32 tab = signal->theData[1];
const Uint32 ver = signal->theData[2];
TableRecordPtr tabRecPtr;
c_tableRecordPool.getPtr(tabRecPtr, tab);
DropTableReq * req = (DropTableReq*)signal->getDataPtr();
req->senderData = 1225;
req->senderRef = numberToRef(1,1);
req->tableId = tab;
req->tableVersion = tabRecPtr.p->tableVersion + ver;
sendSignal(DBDICT_REF, GSN_DROP_TABLE_REQ, signal,
DropTableReq::SignalLength, JBB);
}
#endif
return;
}//Dbdict::execDUMP_STATE_ORD()
/* ---------------------------------------------------------------- */
/* ---------------------------------------------------------------- */
// CONTINUEB is used when a real-time break is needed for long
// processes.
/* ---------------------------------------------------------------- */
/* ---------------------------------------------------------------- */
void Dbdict::execCONTINUEB(Signal* signal)
{
jamEntry();
switch (signal->theData[0]) {
case ZPACK_TABLE_INTO_PAGES :
jam();
packTableIntoPages(signal, signal->theData[1], signal->theData[2]);
break;
case ZSEND_GET_TAB_RESPONSE :
jam();
sendGetTabResponse(signal);
break;
default :
ndbrequire(false);
break;
}//switch
return;
}//execCONTINUEB()
/* ---------------------------------------------------------------- */
/* ---------------------------------------------------------------- */
// Routine to handle pack table into pages.
/* ---------------------------------------------------------------- */
/* ---------------------------------------------------------------- */
void Dbdict::packTableIntoPages(Signal* signal, Uint32 tableId, Uint32 pageId)
{
PageRecordPtr pagePtr;
TableRecordPtr tablePtr;
c_pageRecordArray.getPtr(pagePtr, pageId);
memset(&pagePtr.p->word[0], 0, 4 * ZPAGE_HEADER_SIZE);
c_tableRecordPool.getPtr(tablePtr, tableId);
LinearWriter w(&pagePtr.p->word[ZPAGE_HEADER_SIZE],
8 * ZSIZE_OF_PAGES_IN_WORDS);
w.first();
packTableIntoPagesImpl(w, tablePtr);
Uint32 wordsOfTable = w.getWordsUsed();
Uint32 pagesUsed =
DIV(wordsOfTable + ZPAGE_HEADER_SIZE, ZSIZE_OF_PAGES_IN_WORDS);
pagePtr.p->word[ZPOS_CHECKSUM] =
computeChecksum(&pagePtr.p->word[0], pagesUsed * ZSIZE_OF_PAGES_IN_WORDS);
switch (c_packTable.m_state) {
case PackTable::PTS_IDLE:
case PackTable::PTS_ADD_TABLE_MASTER:
case PackTable::PTS_ADD_TABLE_SLAVE:
case PackTable::PTS_RESTART:
ndbrequire(false);
break;
case PackTable::PTS_GET_TAB:
jam();
c_retrieveRecord.retrievedNoOfPages = pagesUsed;
c_retrieveRecord.retrievedNoOfWords = wordsOfTable;
sendGetTabResponse(signal);
return;
break;
}//switch
ndbrequire(false);
return;
}//packTableIntoPages()
void
Dbdict::packTableIntoPagesImpl(SimpleProperties::Writer & w,
TableRecordPtr tablePtr){
w.add(DictTabInfo::TableName, tablePtr.p->tableName);
w.add(DictTabInfo::TableId, tablePtr.i);
#ifdef HAVE_TABLE_REORG
w.add(DictTabInfo::SecondTableId, tablePtr.p->secondTable);
#else
w.add(DictTabInfo::SecondTableId, (Uint32)0);
#endif
w.add(DictTabInfo::TableVersion, tablePtr.p->tableVersion);
w.add(DictTabInfo::NoOfKeyAttr, tablePtr.p->noOfPrimkey);
w.add(DictTabInfo::NoOfAttributes, tablePtr.p->noOfAttributes);
w.add(DictTabInfo::NoOfNullable, tablePtr.p->noOfNullAttr);
w.add(DictTabInfo::NoOfVariable, (Uint32)0);
w.add(DictTabInfo::KeyLength, tablePtr.p->tupKeyLength);
w.add(DictTabInfo::TableLoggedFlag, tablePtr.p->storedTable);
w.add(DictTabInfo::MinLoadFactor, tablePtr.p->minLoadFactor);
w.add(DictTabInfo::MaxLoadFactor, tablePtr.p->maxLoadFactor);
w.add(DictTabInfo::TableKValue, tablePtr.p->kValue);
w.add(DictTabInfo::FragmentTypeVal, tablePtr.p->fragmentType);
w.add(DictTabInfo::FragmentKeyTypeVal, tablePtr.p->fragmentKeyType);
w.add(DictTabInfo::TableTypeVal, tablePtr.p->tableType);
w.add(DictTabInfo::FragmentCount, tablePtr.p->fragmentCount);
if (tablePtr.p->primaryTableId != RNIL){
TableRecordPtr primTab;
c_tableRecordPool.getPtr(primTab, tablePtr.p->primaryTableId);
w.add(DictTabInfo::PrimaryTable, primTab.p->tableName);
w.add(DictTabInfo::PrimaryTableId, tablePtr.p->primaryTableId);
w.add(DictTabInfo::IndexState, tablePtr.p->indexState);
w.add(DictTabInfo::InsertTriggerId, tablePtr.p->insertTriggerId);
w.add(DictTabInfo::UpdateTriggerId, tablePtr.p->updateTriggerId);
w.add(DictTabInfo::DeleteTriggerId, tablePtr.p->deleteTriggerId);
w.add(DictTabInfo::CustomTriggerId, tablePtr.p->customTriggerId);
}
w.add(DictTabInfo::FrmLen, tablePtr.p->frmLen);
w.add(DictTabInfo::FrmData, tablePtr.p->frmData, tablePtr.p->frmLen);
Uint32 nextAttribute = tablePtr.p->firstAttribute;
AttributeRecordPtr attrPtr;
do {
jam();
c_attributeRecordPool.getPtr(attrPtr, nextAttribute);
w.add(DictTabInfo::AttributeName, attrPtr.p->attributeName);
w.add(DictTabInfo::AttributeId, attrPtr.p->attributeId);
w.add(DictTabInfo::AttributeKeyFlag, attrPtr.p->tupleKey > 0);
const Uint32 desc = attrPtr.p->attributeDescriptor;
const Uint32 attrType = AttributeDescriptor::getType(desc);
const Uint32 attrSize = AttributeDescriptor::getSize(desc);
const Uint32 arraySize = AttributeDescriptor::getArraySize(desc);
const Uint32 nullable = AttributeDescriptor::getNullable(desc);
const Uint32 DGroup = AttributeDescriptor::getDGroup(desc);
const Uint32 DKey = AttributeDescriptor::getDKey(desc);
const Uint32 attrStoredInd = AttributeDescriptor::getStoredInTup(desc);
w.add(DictTabInfo::AttributeType, attrType);
w.add(DictTabInfo::AttributeSize, attrSize);
w.add(DictTabInfo::AttributeArraySize, arraySize);
w.add(DictTabInfo::AttributeNullableFlag, nullable);
w.add(DictTabInfo::AttributeDGroup, DGroup);
w.add(DictTabInfo::AttributeDKey, DKey);
w.add(DictTabInfo::AttributeStoredInd, attrStoredInd);
w.add(DictTabInfo::AttributeExtType, attrPtr.p->extType);
w.add(DictTabInfo::AttributeExtPrecision, attrPtr.p->extPrecision);
w.add(DictTabInfo::AttributeExtScale, attrPtr.p->extScale);
w.add(DictTabInfo::AttributeExtLength, attrPtr.p->extLength);
w.add(DictTabInfo::AttributeAutoIncrement,
(Uint32)attrPtr.p->autoIncrement);
w.add(DictTabInfo::AttributeDefaultValue, attrPtr.p->defaultValue);
w.add(DictTabInfo::AttributeEnd, 1);
nextAttribute = attrPtr.p->nextAttrInTable;
} while (nextAttribute != RNIL);
w.add(DictTabInfo::TableEnd, 1);
}
/* ---------------------------------------------------------------- */
/* ---------------------------------------------------------------- */
// The routines to handle responses from file system.
/* ---------------------------------------------------------------- */
/* ---------------------------------------------------------------- */
/* ---------------------------------------------------------------- */
// A file was successfully closed.
/* ---------------------------------------------------------------- */
void Dbdict::execFSCLOSECONF(Signal* signal)
{
FsConnectRecordPtr fsPtr;
FsConf * const fsConf = (FsConf *)&signal->theData[0];
jamEntry();
c_fsConnectRecordPool.getPtr(fsPtr, fsConf->userPointer);
switch (fsPtr.p->fsState) {
case FsConnectRecord::CLOSE_WRITE_SCHEMA:
jam();
closeWriteSchemaConf(signal, fsPtr);
break;
case FsConnectRecord::CLOSE_READ_SCHEMA:
jam();
closeReadSchemaConf(signal, fsPtr);
break;
case FsConnectRecord::CLOSE_READ_TAB_FILE:
jam();
closeReadTableConf(signal, fsPtr);
break;
case FsConnectRecord::CLOSE_WRITE_TAB_FILE:
jam();
closeWriteTableConf(signal, fsPtr);
break;
case FsConnectRecord::OPEN_READ_SCHEMA2:
openSchemaFile(signal, 1, fsPtr.i, false);
break;
default:
jamLine((fsPtr.p->fsState & 0xFFF));
ndbrequire(false);
break;
}//switch
}//execFSCLOSECONF()
/* ---------------------------------------------------------------- */
// A file was successfully opened.
/* ---------------------------------------------------------------- */
void Dbdict::execFSOPENCONF(Signal* signal)
{
FsConnectRecordPtr fsPtr;
jamEntry();
FsConf * const fsConf = (FsConf *)&signal->theData[0];
c_fsConnectRecordPool.getPtr(fsPtr, fsConf->userPointer);
Uint32 filePointer = fsConf->filePointer;
fsPtr.p->filePtr = filePointer;
switch (fsPtr.p->fsState) {
case FsConnectRecord::OPEN_WRITE_SCHEMA:
jam();
fsPtr.p->fsState = FsConnectRecord::WRITE_SCHEMA;
writeSchemaFile(signal, filePointer, fsPtr.i);
break;
case FsConnectRecord::OPEN_READ_SCHEMA1:
jam();
fsPtr.p->fsState = FsConnectRecord::READ_SCHEMA1;
readSchemaFile(signal, filePointer, fsPtr.i);
break;
case FsConnectRecord::OPEN_READ_SCHEMA2:
jam();
fsPtr.p->fsState = FsConnectRecord::READ_SCHEMA2;
readSchemaFile(signal, filePointer, fsPtr.i);
break;
case FsConnectRecord::OPEN_READ_TAB_FILE1:
jam();
fsPtr.p->fsState = FsConnectRecord::READ_TAB_FILE1;
readTableFile(signal, filePointer, fsPtr.i);
break;
case FsConnectRecord::OPEN_READ_TAB_FILE2:
jam();
fsPtr.p->fsState = FsConnectRecord::READ_TAB_FILE2;
readTableFile(signal, filePointer, fsPtr.i);
break;
case FsConnectRecord::OPEN_WRITE_TAB_FILE:
jam();
fsPtr.p->fsState = FsConnectRecord::WRITE_TAB_FILE;
writeTableFile(signal, filePointer, fsPtr.i);
break;
default:
jamLine((fsPtr.p->fsState & 0xFFF));
ndbrequire(false);
break;
}//switch
}//execFSOPENCONF()
/* ---------------------------------------------------------------- */
// An open file was refused.
/* ---------------------------------------------------------------- */
void Dbdict::execFSOPENREF(Signal* signal)
{
jamEntry();
FsRef * const fsRef = (FsRef *)&signal->theData[0];
FsConnectRecordPtr fsPtr;
c_fsConnectRecordPool.getPtr(fsPtr, fsRef->userPointer);
switch (fsPtr.p->fsState) {
case FsConnectRecord::OPEN_READ_SCHEMA1:
jam();
openReadSchemaRef(signal, fsPtr);
return;
case FsConnectRecord::OPEN_READ_TAB_FILE1:
jam();
openReadTableRef(signal, fsPtr);
return;
default:
break;
}//switch
{
char msg[100];
sprintf(msg, "File system open failed during FsConnectRecord state %d", (Uint32)fsPtr.p->fsState);
fsRefError(signal,__LINE__,msg);
}
}//execFSOPENREF()
/* ---------------------------------------------------------------- */
// A file was successfully read.
/* ---------------------------------------------------------------- */
void Dbdict::execFSREADCONF(Signal* signal)
{
jamEntry();
FsConf * const fsConf = (FsConf *)&signal->theData[0];
FsConnectRecordPtr fsPtr;
c_fsConnectRecordPool.getPtr(fsPtr, fsConf->userPointer);
switch (fsPtr.p->fsState) {
case FsConnectRecord::READ_SCHEMA1:
case FsConnectRecord::READ_SCHEMA2:
readSchemaConf(signal ,fsPtr);
break;
case FsConnectRecord::READ_TAB_FILE1:
case FsConnectRecord::READ_TAB_FILE2:
jam();
readTableConf(signal ,fsPtr);
break;
default:
jamLine((fsPtr.p->fsState & 0xFFF));
ndbrequire(false);
break;
}//switch
}//execFSREADCONF()
/* ---------------------------------------------------------------- */
// A read file was refused.
/* ---------------------------------------------------------------- */
void Dbdict::execFSREADREF(Signal* signal)
{
jamEntry();
FsRef * const fsRef = (FsRef *)&signal->theData[0];
FsConnectRecordPtr fsPtr;
c_fsConnectRecordPool.getPtr(fsPtr, fsRef->userPointer);
switch (fsPtr.p->fsState) {
case FsConnectRecord::READ_SCHEMA1:
jam();
readSchemaRef(signal, fsPtr);
return;
case FsConnectRecord::READ_TAB_FILE1:
jam();
readTableRef(signal, fsPtr);
return;
default:
break;
}//switch
{
char msg[100];
sprintf(msg, "File system read failed during FsConnectRecord state %d", (Uint32)fsPtr.p->fsState);
fsRefError(signal,__LINE__,msg);
}
}//execFSREADREF()
/* ---------------------------------------------------------------- */
// A file was successfully written.
/* ---------------------------------------------------------------- */
void Dbdict::execFSWRITECONF(Signal* signal)
{
FsConf * const fsConf = (FsConf *)&signal->theData[0];
FsConnectRecordPtr fsPtr;
jamEntry();
c_fsConnectRecordPool.getPtr(fsPtr, fsConf->userPointer);
switch (fsPtr.p->fsState) {
case FsConnectRecord::WRITE_TAB_FILE:
writeTableConf(signal, fsPtr);
break;
case FsConnectRecord::WRITE_SCHEMA:
jam();
writeSchemaConf(signal, fsPtr);
break;
default:
jamLine((fsPtr.p->fsState & 0xFFF));
ndbrequire(false);
break;
}//switch
}//execFSWRITECONF()
/* ---------------------------------------------------------------- */
// Routines to handle Read/Write of Table Files
/* ---------------------------------------------------------------- */
void
Dbdict::writeTableFile(Signal* signal, Uint32 tableId,
SegmentedSectionPtr tabInfoPtr, Callback* callback){
ndbrequire(c_writeTableRecord.tableWriteState == WriteTableRecord::IDLE);
Uint32 sz = tabInfoPtr.sz + ZPAGE_HEADER_SIZE;
c_writeTableRecord.noOfPages = DIV(sz, ZSIZE_OF_PAGES_IN_WORDS);
c_writeTableRecord.tableWriteState = WriteTableRecord::TWR_CALLBACK;
c_writeTableRecord.m_callback = * callback;
c_writeTableRecord.pageId = 0;
ndbrequire(c_writeTableRecord.noOfPages < 8);
PageRecordPtr pageRecPtr;
c_pageRecordArray.getPtr(pageRecPtr, c_writeTableRecord.pageId);
copy(&pageRecPtr.p->word[ZPAGE_HEADER_SIZE], tabInfoPtr);
memset(&pageRecPtr.p->word[0], 0, 4 * ZPAGE_HEADER_SIZE);
pageRecPtr.p->word[ZPOS_CHECKSUM] =
computeChecksum(&pageRecPtr.p->word[0],
c_writeTableRecord.noOfPages * ZSIZE_OF_PAGES_IN_WORDS);
startWriteTableFile(signal, tableId);
}
void Dbdict::startWriteTableFile(Signal* signal, Uint32 tableId)
{
FsConnectRecordPtr fsPtr;
c_writeTableRecord.tableId = tableId;
c_fsConnectRecordPool.getPtr(fsPtr, getFsConnRecord());
fsPtr.p->fsState = FsConnectRecord::OPEN_WRITE_TAB_FILE;
openTableFile(signal, 0, fsPtr.i, tableId, true);
c_writeTableRecord.noOfTableFilesHandled = 0;
}//Dbdict::startWriteTableFile()
void Dbdict::openTableFile(Signal* signal,
Uint32 fileNo,
Uint32 fsConPtr,
Uint32 tableId,
bool writeFlag)
{
TableRecordPtr tablePtr;
FsOpenReq * const fsOpenReq = (FsOpenReq *)&signal->theData[0];
c_tableRecordPool.getPtr(tablePtr, tableId);
fsOpenReq->userReference = reference();
fsOpenReq->userPointer = fsConPtr;
if (writeFlag) {
jam();
fsOpenReq->fileFlags =
FsOpenReq::OM_WRITEONLY |
FsOpenReq::OM_TRUNCATE |
FsOpenReq::OM_CREATE |
FsOpenReq::OM_SYNC;
} else {
jam();
fsOpenReq->fileFlags = FsOpenReq::OM_READONLY;
}//if
fsOpenReq->fileNumber[3] = 0; // Initialise before byte changes
FsOpenReq::setVersion(fsOpenReq->fileNumber, 1);
FsOpenReq::setSuffix(fsOpenReq->fileNumber, FsOpenReq::S_TABLELIST);
FsOpenReq::v1_setDisk(fsOpenReq->fileNumber, (fileNo + 1));
FsOpenReq::v1_setTable(fsOpenReq->fileNumber, tableId);
FsOpenReq::v1_setFragment(fsOpenReq->fileNumber, (Uint32)-1);
FsOpenReq::v1_setS(fsOpenReq->fileNumber, tablePtr.p->tableVersion);
FsOpenReq::v1_setP(fsOpenReq->fileNumber, 255);
/* ---------------------------------------------------------------- */
// File name : D1/DBDICT/T0/S1.TableList
// D1 means Disk 1 (set by fileNo + 1)
// T0 means table id = 0
// S1 means tableVersion 1
// TableList indicates that this is a file for a table description.
/* ---------------------------------------------------------------- */
sendSignal(NDBFS_REF, GSN_FSOPENREQ, signal, FsOpenReq::SignalLength, JBA);
}//openTableFile()
void Dbdict::writeTableFile(Signal* signal, Uint32 filePtr, Uint32 fsConPtr)
{
FsReadWriteReq * const fsRWReq = (FsReadWriteReq *)&signal->theData[0];
fsRWReq->filePointer = filePtr;
fsRWReq->userReference = reference();
fsRWReq->userPointer = fsConPtr;
fsRWReq->operationFlag = 0; // Initialise before bit changes
FsReadWriteReq::setSyncFlag(fsRWReq->operationFlag, 1);
FsReadWriteReq::setFormatFlag(fsRWReq->operationFlag,
FsReadWriteReq::fsFormatArrayOfPages);
fsRWReq->varIndex = ZALLOCATE;
fsRWReq->numberOfPages = c_writeTableRecord.noOfPages;
fsRWReq->data.arrayOfPages.varIndex = c_writeTableRecord.pageId;
fsRWReq->data.arrayOfPages.fileOffset = 0; // Write to file page 0
sendSignal(NDBFS_REF, GSN_FSWRITEREQ, signal, 8, JBA);
}//writeTableFile()
void Dbdict::writeTableConf(Signal* signal,
FsConnectRecordPtr fsPtr)
{
fsPtr.p->fsState = FsConnectRecord::CLOSE_WRITE_TAB_FILE;
closeFile(signal, fsPtr.p->filePtr, fsPtr.i);
return;
}//Dbdict::writeTableConf()
void Dbdict::closeWriteTableConf(Signal* signal,
FsConnectRecordPtr fsPtr)
{
c_writeTableRecord.noOfTableFilesHandled++;
if (c_writeTableRecord.noOfTableFilesHandled < 2) {
jam();
fsPtr.p->fsState = FsConnectRecord::OPEN_WRITE_TAB_FILE;
openTableFile(signal, 1, fsPtr.i, c_writeTableRecord.tableId, true);
return;
}
ndbrequire(c_writeTableRecord.noOfTableFilesHandled == 2);
c_fsConnectRecordPool.release(fsPtr);
WriteTableRecord::TableWriteState state = c_writeTableRecord.tableWriteState;
c_writeTableRecord.tableWriteState = WriteTableRecord::IDLE;
switch (state) {
case WriteTableRecord::IDLE:
case WriteTableRecord::WRITE_ADD_TABLE_MASTER :
case WriteTableRecord::WRITE_ADD_TABLE_SLAVE :
case WriteTableRecord::WRITE_RESTART_FROM_MASTER :
case WriteTableRecord::WRITE_RESTART_FROM_OWN :
ndbrequire(false);
break;
case WriteTableRecord::TWR_CALLBACK:
jam();
execute(signal, c_writeTableRecord.m_callback, 0);
return;
}
ndbrequire(false);
}//Dbdict::closeWriteTableConf()
void Dbdict::startReadTableFile(Signal* signal, Uint32 tableId)
{
//globalSignalLoggers.log(number(), "startReadTableFile");
ndbrequire(!c_readTableRecord.inUse);
FsConnectRecordPtr fsPtr;
c_fsConnectRecordPool.getPtr(fsPtr, getFsConnRecord());
c_readTableRecord.inUse = true;
c_readTableRecord.tableId = tableId;
fsPtr.p->fsState = FsConnectRecord::OPEN_READ_TAB_FILE1;
openTableFile(signal, 0, fsPtr.i, tableId, false);
}//Dbdict::startReadTableFile()
void Dbdict::openReadTableRef(Signal* signal,
FsConnectRecordPtr fsPtr)
{
fsPtr.p->fsState = FsConnectRecord::OPEN_READ_TAB_FILE2;
openTableFile(signal, 1, fsPtr.i, c_readTableRecord.tableId, false);
return;
}//Dbdict::openReadTableConf()
void Dbdict::readTableFile(Signal* signal, Uint32 filePtr, Uint32 fsConPtr)
{
FsReadWriteReq * const fsRWReq = (FsReadWriteReq *)&signal->theData[0];
fsRWReq->filePointer = filePtr;
fsRWReq->userReference = reference();
fsRWReq->userPointer = fsConPtr;
fsRWReq->operationFlag = 0; // Initialise before bit changes
FsReadWriteReq::setSyncFlag(fsRWReq->operationFlag, 0);
FsReadWriteReq::setFormatFlag(fsRWReq->operationFlag,
FsReadWriteReq::fsFormatArrayOfPages);
fsRWReq->varIndex = ZALLOCATE;
fsRWReq->numberOfPages = c_readTableRecord.noOfPages;
fsRWReq->data.arrayOfPages.varIndex = c_readTableRecord.pageId;
fsRWReq->data.arrayOfPages.fileOffset = 0; // Write to file page 0
sendSignal(NDBFS_REF, GSN_FSREADREQ, signal, 8, JBA);
}//readTableFile()
void Dbdict::readTableConf(Signal* signal,
FsConnectRecordPtr fsPtr)
{
/* ---------------------------------------------------------------- */
// Verify the data read from disk
/* ---------------------------------------------------------------- */
bool crashInd;
if (fsPtr.p->fsState == FsConnectRecord::READ_TAB_FILE1) {
jam();
crashInd = false;
} else {
jam();
crashInd = true;
}//if
PageRecordPtr tmpPagePtr;
c_pageRecordArray.getPtr(tmpPagePtr, c_readTableRecord.pageId);
Uint32 sz = c_readTableRecord.noOfPages * ZSIZE_OF_PAGES_IN_WORDS;
Uint32 chk = computeChecksum((const Uint32*)tmpPagePtr.p, sz);
ndbrequire((chk == 0) || !crashInd);
if(chk != 0){
jam();
ndbrequire(fsPtr.p->fsState == FsConnectRecord::READ_TAB_FILE1);
readTableRef(signal, fsPtr);
return;
}//if
fsPtr.p->fsState = FsConnectRecord::CLOSE_READ_TAB_FILE;
closeFile(signal, fsPtr.p->filePtr, fsPtr.i);
return;
}//Dbdict::readTableConf()
void Dbdict::readTableRef(Signal* signal,
FsConnectRecordPtr fsPtr)
{
fsPtr.p->fsState = FsConnectRecord::OPEN_READ_TAB_FILE2;
openTableFile(signal, 1, fsPtr.i, c_readTableRecord.tableId, false);
return;
}//Dbdict::readTableRef()
void Dbdict::closeReadTableConf(Signal* signal,
FsConnectRecordPtr fsPtr)
{
c_fsConnectRecordPool.release(fsPtr);
c_readTableRecord.inUse = false;
execute(signal, c_readTableRecord.m_callback, 0);
return;
}//Dbdict::closeReadTableConf()
/* ---------------------------------------------------------------- */
// Routines to handle Read/Write of Schema Files
/* ---------------------------------------------------------------- */
void
Dbdict::updateSchemaState(Signal* signal, Uint32 tableId,
SchemaFile::TableEntry* te, Callback* callback){
jam();
PageRecordPtr pagePtr;
c_pageRecordArray.getPtr(pagePtr, c_schemaRecord.schemaPage);
ndbrequire(tableId < c_tableRecordPool.getSize());
SchemaFile::TableEntry * tableEntry = getTableEntry(pagePtr.p, tableId);
SchemaFile::TableState newState =
(SchemaFile::TableState)te->m_tableState;
SchemaFile::TableState oldState =
(SchemaFile::TableState)tableEntry->m_tableState;
Uint32 newVersion = te->m_tableVersion;
Uint32 oldVersion = tableEntry->m_tableVersion;
bool ok = false;
switch(newState){
case SchemaFile::ADD_STARTED:
jam();
ok = true;
ndbrequire(create_table_inc_schema_version(oldVersion) == newVersion);
ndbrequire(oldState == SchemaFile::INIT ||
oldState == SchemaFile::DROP_TABLE_COMMITTED);
break;
case SchemaFile::TABLE_ADD_COMMITTED:
jam();
ok = true;
ndbrequire(newVersion == oldVersion);
ndbrequire(oldState == SchemaFile::ADD_STARTED);
break;
case SchemaFile::ALTER_TABLE_COMMITTED:
jam();
ok = true;
ndbrequire(alter_table_inc_schema_version(oldVersion) == newVersion);
ndbrequire(oldState == SchemaFile::TABLE_ADD_COMMITTED ||
oldState == SchemaFile::ALTER_TABLE_COMMITTED);
break;
case SchemaFile::DROP_TABLE_STARTED:
jam();
case SchemaFile::DROP_TABLE_COMMITTED:
jam();
ok = true;
ndbrequire(false);
break;
case SchemaFile::INIT:
jam();
ok = true;
ndbrequire((oldState == SchemaFile::ADD_STARTED));
}//if
ndbrequire(ok);
* tableEntry = * te;
computeChecksum((SchemaFile*)pagePtr.p);
ndbrequire(c_writeSchemaRecord.inUse == false);
c_writeSchemaRecord.inUse = true;
c_writeSchemaRecord.pageId = c_schemaRecord.schemaPage;
c_writeSchemaRecord.m_callback = * callback;
startWriteSchemaFile(signal);
}
void Dbdict::startWriteSchemaFile(Signal* signal)
{
FsConnectRecordPtr fsPtr;
c_fsConnectRecordPool.getPtr(fsPtr, getFsConnRecord());
fsPtr.p->fsState = FsConnectRecord::OPEN_WRITE_SCHEMA;
openSchemaFile(signal, 0, fsPtr.i, true);
c_writeSchemaRecord.noOfSchemaFilesHandled = 0;
}//Dbdict::startWriteSchemaFile()
void Dbdict::openSchemaFile(Signal* signal,
Uint32 fileNo,
Uint32 fsConPtr,
bool writeFlag)
{
FsOpenReq * const fsOpenReq = (FsOpenReq *)&signal->theData[0];
fsOpenReq->userReference = reference();
fsOpenReq->userPointer = fsConPtr;
if (writeFlag) {
jam();
fsOpenReq->fileFlags =
FsOpenReq::OM_WRITEONLY |
FsOpenReq::OM_TRUNCATE |
FsOpenReq::OM_CREATE |
FsOpenReq::OM_SYNC;
} else {
jam();
fsOpenReq->fileFlags = FsOpenReq::OM_READONLY;
}//if
fsOpenReq->fileNumber[3] = 0; // Initialise before byte changes
FsOpenReq::setVersion(fsOpenReq->fileNumber, 1);
FsOpenReq::setSuffix(fsOpenReq->fileNumber, FsOpenReq::S_SCHEMALOG);
FsOpenReq::v1_setDisk(fsOpenReq->fileNumber, (fileNo + 1));
FsOpenReq::v1_setTable(fsOpenReq->fileNumber, (Uint32)-1);
FsOpenReq::v1_setFragment(fsOpenReq->fileNumber, (Uint32)-1);
FsOpenReq::v1_setS(fsOpenReq->fileNumber, (Uint32)-1);
FsOpenReq::v1_setP(fsOpenReq->fileNumber, 0);
/* ---------------------------------------------------------------- */
// File name : D1/DBDICT/P0.SchemaLog
// D1 means Disk 1 (set by fileNo + 1). Writes to both D1 and D2
// SchemaLog indicates that this is a file giving a list of current tables.
/* ---------------------------------------------------------------- */
sendSignal(NDBFS_REF, GSN_FSOPENREQ, signal, FsOpenReq::SignalLength, JBA);
}//openSchemaFile()
void Dbdict::writeSchemaFile(Signal* signal, Uint32 filePtr, Uint32 fsConPtr)
{
FsReadWriteReq * const fsRWReq = (FsReadWriteReq *)&signal->theData[0];
fsRWReq->filePointer = filePtr;
fsRWReq->userReference = reference();
fsRWReq->userPointer = fsConPtr;
fsRWReq->operationFlag = 0; // Initialise before bit changes
FsReadWriteReq::setSyncFlag(fsRWReq->operationFlag, 1);
FsReadWriteReq::setFormatFlag(fsRWReq->operationFlag,
FsReadWriteReq::fsFormatArrayOfPages);
fsRWReq->varIndex = ZALLOCATE;
fsRWReq->numberOfPages = 1;
// Write from memory page
fsRWReq->data.arrayOfPages.varIndex = c_writeSchemaRecord.pageId;
fsRWReq->data.arrayOfPages.fileOffset = 0; // Write to file page 0
sendSignal(NDBFS_REF, GSN_FSWRITEREQ, signal, 8, JBA);
}//writeSchemaFile()
void Dbdict::writeSchemaConf(Signal* signal,
FsConnectRecordPtr fsPtr)
{
fsPtr.p->fsState = FsConnectRecord::CLOSE_WRITE_SCHEMA;
closeFile(signal, fsPtr.p->filePtr, fsPtr.i);
return;
}//Dbdict::writeSchemaConf()
void Dbdict::closeFile(Signal* signal, Uint32 filePtr, Uint32 fsConPtr)
{
FsCloseReq * const fsCloseReq = (FsCloseReq *)&signal->theData[0];
fsCloseReq->filePointer = filePtr;
fsCloseReq->userReference = reference();
fsCloseReq->userPointer = fsConPtr;
FsCloseReq::setRemoveFileFlag(fsCloseReq->fileFlag, false);
sendSignal(NDBFS_REF, GSN_FSCLOSEREQ, signal, FsCloseReq::SignalLength, JBA);
return;
}//closeFile()
void Dbdict::closeWriteSchemaConf(Signal* signal,
FsConnectRecordPtr fsPtr)
{
c_writeSchemaRecord.noOfSchemaFilesHandled++;
if (c_writeSchemaRecord.noOfSchemaFilesHandled < 2) {
jam();
fsPtr.p->fsState = FsConnectRecord::OPEN_WRITE_SCHEMA;
openSchemaFile(signal, 1, fsPtr.i, true);
return;
}
ndbrequire(c_writeSchemaRecord.noOfSchemaFilesHandled == 2);
c_fsConnectRecordPool.release(fsPtr);
c_writeSchemaRecord.inUse = false;
execute(signal, c_writeSchemaRecord.m_callback, 0);
return;
}//Dbdict::closeWriteSchemaConf()
void Dbdict::startReadSchemaFile(Signal* signal)
{
//globalSignalLoggers.log(number(), "startReadSchemaFile");
FsConnectRecordPtr fsPtr;
c_fsConnectRecordPool.getPtr(fsPtr, getFsConnRecord());
fsPtr.p->fsState = FsConnectRecord::OPEN_READ_SCHEMA1;
openSchemaFile(signal, 0, fsPtr.i, false);
}//Dbdict::startReadSchemaFile()
void Dbdict::openReadSchemaRef(Signal* signal,
FsConnectRecordPtr fsPtr)
{
fsPtr.p->fsState = FsConnectRecord::OPEN_READ_SCHEMA2;
openSchemaFile(signal, 1, fsPtr.i, false);
}//Dbdict::openReadSchemaRef()
void Dbdict::readSchemaFile(Signal* signal, Uint32 filePtr, Uint32 fsConPtr)
{
FsReadWriteReq * const fsRWReq = (FsReadWriteReq *)&signal->theData[0];
fsRWReq->filePointer = filePtr;
fsRWReq->userReference = reference();
fsRWReq->userPointer = fsConPtr;
fsRWReq->operationFlag = 0; // Initialise before bit changes
FsReadWriteReq::setSyncFlag(fsRWReq->operationFlag, 0);
FsReadWriteReq::setFormatFlag(fsRWReq->operationFlag,
FsReadWriteReq::fsFormatArrayOfPages);
fsRWReq->varIndex = ZALLOCATE;
fsRWReq->numberOfPages = 1;
fsRWReq->data.arrayOfPages.varIndex = c_readSchemaRecord.pageId;
fsRWReq->data.arrayOfPages.fileOffset = 0;
sendSignal(NDBFS_REF, GSN_FSREADREQ, signal, 8, JBA);
}//readSchemaFile()
void Dbdict::readSchemaConf(Signal* signal,
FsConnectRecordPtr fsPtr)
{
/* ---------------------------------------------------------------- */
// Verify the data read from disk
/* ---------------------------------------------------------------- */
bool crashInd;
if (fsPtr.p->fsState == FsConnectRecord::READ_SCHEMA1) {
jam();
crashInd = false;
} else {
jam();
crashInd = true;
}//if
PageRecordPtr tmpPagePtr;
c_pageRecordArray.getPtr(tmpPagePtr, c_readSchemaRecord.pageId);
Uint32 sz = ZSIZE_OF_PAGES_IN_WORDS;
Uint32 chk = computeChecksum((const Uint32*)tmpPagePtr.p, sz);
ndbrequire((chk == 0) || !crashInd);
if (chk != 0){
jam();
ndbrequire(fsPtr.p->fsState == FsConnectRecord::READ_SCHEMA1);
readSchemaRef(signal, fsPtr);
return;
}//if
fsPtr.p->fsState = FsConnectRecord::CLOSE_READ_SCHEMA;
closeFile(signal, fsPtr.p->filePtr, fsPtr.i);
return;
}//Dbdict::readSchemaConf()
void Dbdict::readSchemaRef(Signal* signal,
FsConnectRecordPtr fsPtr)
{
/**
* First close corrupt file
*/
fsPtr.p->fsState = FsConnectRecord::OPEN_READ_SCHEMA2;
closeFile(signal, fsPtr.p->filePtr, fsPtr.i);
return;
}
void Dbdict::closeReadSchemaConf(Signal* signal,
FsConnectRecordPtr fsPtr)
{
c_fsConnectRecordPool.release(fsPtr);
ReadSchemaRecord::SchemaReadState state = c_readSchemaRecord.schemaReadState;
c_readSchemaRecord.schemaReadState = ReadSchemaRecord::IDLE;
switch(state) {
case ReadSchemaRecord::INITIAL_READ :
jam();
sendNDB_STTORRY(signal);
break;
default :
ndbrequire(false);
break;
}//switch
}//Dbdict::closeReadSchemaConf()
/* **************************************************************** */
/* ---------------------------------------------------------------- */
/* MODULE: INITIALISATION MODULE ------------------------- */
/* ---------------------------------------------------------------- */
/* */
/* This module contains initialisation of data at start/restart. */
/* ---------------------------------------------------------------- */
/* **************************************************************** */
Dbdict::Dbdict(const class Configuration & conf):
SimulatedBlock(DBDICT, conf),
c_tableRecordHash(c_tableRecordPool),
c_attributeRecordHash(c_attributeRecordPool),
c_triggerRecordHash(c_triggerRecordPool),
c_opCreateTable(c_opRecordPool),
c_opDropTable(c_opRecordPool),
c_opCreateIndex(c_opRecordPool),
c_opDropIndex(c_opRecordPool),
c_opAlterIndex(c_opRecordPool),
c_opBuildIndex(c_opRecordPool),
c_opCreateEvent(c_opRecordPool),
c_opSubEvent(c_opRecordPool),
c_opDropEvent(c_opRecordPool),
c_opSignalUtil(c_opRecordPool),
c_opCreateTrigger(c_opRecordPool),
c_opDropTrigger(c_opRecordPool),
c_opAlterTrigger(c_opRecordPool),
c_opRecordSequence(0)
{
BLOCK_CONSTRUCTOR(Dbdict);
const ndb_mgm_configuration_iterator * p = conf.getOwnConfigIterator();
ndbrequire(p != 0);
ndb_mgm_get_int_parameter(p, CFG_DB_NO_TRIGGERS, &c_maxNoOfTriggers);
// Transit signals
addRecSignal(GSN_DUMP_STATE_ORD, &Dbdict::execDUMP_STATE_ORD);
addRecSignal(GSN_GET_TABINFOREQ, &Dbdict::execGET_TABINFOREQ);
addRecSignal(GSN_GET_TABLEID_REQ, &Dbdict::execGET_TABLEDID_REQ);
addRecSignal(GSN_GET_TABINFO_CONF, &Dbdict::execGET_TABINFO_CONF);
addRecSignal(GSN_CONTINUEB, &Dbdict::execCONTINUEB);
addRecSignal(GSN_CREATE_TABLE_REQ, &Dbdict::execCREATE_TABLE_REQ);
addRecSignal(GSN_CREATE_TAB_REQ, &Dbdict::execCREATE_TAB_REQ);
addRecSignal(GSN_CREATE_TAB_REF, &Dbdict::execCREATE_TAB_REF);
addRecSignal(GSN_CREATE_TAB_CONF, &Dbdict::execCREATE_TAB_CONF);
addRecSignal(GSN_CREATE_FRAGMENTATION_REF, &Dbdict::execCREATE_FRAGMENTATION_REF);
addRecSignal(GSN_CREATE_FRAGMENTATION_CONF, &Dbdict::execCREATE_FRAGMENTATION_CONF);
addRecSignal(GSN_DIADDTABCONF, &Dbdict::execDIADDTABCONF);
addRecSignal(GSN_DIADDTABREF, &Dbdict::execDIADDTABREF);
addRecSignal(GSN_ADD_FRAGREQ, &Dbdict::execADD_FRAGREQ);
addRecSignal(GSN_TAB_COMMITCONF, &Dbdict::execTAB_COMMITCONF);
addRecSignal(GSN_TAB_COMMITREF, &Dbdict::execTAB_COMMITREF);
addRecSignal(GSN_ALTER_TABLE_REQ, &Dbdict::execALTER_TABLE_REQ);
addRecSignal(GSN_ALTER_TAB_REQ, &Dbdict::execALTER_TAB_REQ);
addRecSignal(GSN_ALTER_TAB_REF, &Dbdict::execALTER_TAB_REF);
addRecSignal(GSN_ALTER_TAB_CONF, &Dbdict::execALTER_TAB_CONF);
// Index signals
addRecSignal(GSN_CREATE_INDX_REQ, &Dbdict::execCREATE_INDX_REQ);
addRecSignal(GSN_CREATE_INDX_CONF, &Dbdict::execCREATE_INDX_CONF);
addRecSignal(GSN_CREATE_INDX_REF, &Dbdict::execCREATE_INDX_REF);
addRecSignal(GSN_ALTER_INDX_REQ, &Dbdict::execALTER_INDX_REQ);
addRecSignal(GSN_ALTER_INDX_CONF, &Dbdict::execALTER_INDX_CONF);
addRecSignal(GSN_ALTER_INDX_REF, &Dbdict::execALTER_INDX_REF);
addRecSignal(GSN_CREATE_TABLE_CONF, &Dbdict::execCREATE_TABLE_CONF);
addRecSignal(GSN_CREATE_TABLE_REF, &Dbdict::execCREATE_TABLE_REF);
addRecSignal(GSN_DROP_INDX_REQ, &Dbdict::execDROP_INDX_REQ);
addRecSignal(GSN_DROP_INDX_CONF, &Dbdict::execDROP_INDX_CONF);
addRecSignal(GSN_DROP_INDX_REF, &Dbdict::execDROP_INDX_REF);
addRecSignal(GSN_DROP_TABLE_CONF, &Dbdict::execDROP_TABLE_CONF);
addRecSignal(GSN_DROP_TABLE_REF, &Dbdict::execDROP_TABLE_REF);
addRecSignal(GSN_BUILDINDXREQ, &Dbdict::execBUILDINDXREQ);
addRecSignal(GSN_BUILDINDXCONF, &Dbdict::execBUILDINDXCONF);
addRecSignal(GSN_BUILDINDXREF, &Dbdict::execBUILDINDXREF);
// Util signals
addRecSignal(GSN_UTIL_PREPARE_CONF, &Dbdict::execUTIL_PREPARE_CONF);
addRecSignal(GSN_UTIL_PREPARE_REF, &Dbdict::execUTIL_PREPARE_REF);
addRecSignal(GSN_UTIL_EXECUTE_CONF, &Dbdict::execUTIL_EXECUTE_CONF);
addRecSignal(GSN_UTIL_EXECUTE_REF, &Dbdict::execUTIL_EXECUTE_REF);
addRecSignal(GSN_UTIL_RELEASE_CONF, &Dbdict::execUTIL_RELEASE_CONF);
addRecSignal(GSN_UTIL_RELEASE_REF, &Dbdict::execUTIL_RELEASE_REF);
// Event signals
addRecSignal(GSN_CREATE_EVNT_REQ, &Dbdict::execCREATE_EVNT_REQ);
addRecSignal(GSN_CREATE_EVNT_CONF, &Dbdict::execCREATE_EVNT_CONF);
addRecSignal(GSN_CREATE_EVNT_REF, &Dbdict::execCREATE_EVNT_REF);
addRecSignal(GSN_CREATE_SUBID_CONF, &Dbdict::execCREATE_SUBID_CONF);
addRecSignal(GSN_CREATE_SUBID_REF, &Dbdict::execCREATE_SUBID_REF);
addRecSignal(GSN_SUB_CREATE_CONF, &Dbdict::execSUB_CREATE_CONF);
addRecSignal(GSN_SUB_CREATE_REF, &Dbdict::execSUB_CREATE_REF);
addRecSignal(GSN_SUB_START_REQ, &Dbdict::execSUB_START_REQ);
addRecSignal(GSN_SUB_START_CONF, &Dbdict::execSUB_START_CONF);
addRecSignal(GSN_SUB_START_REF, &Dbdict::execSUB_START_REF);
addRecSignal(GSN_SUB_STOP_REQ, &Dbdict::execSUB_STOP_REQ);
addRecSignal(GSN_SUB_STOP_CONF, &Dbdict::execSUB_STOP_CONF);
addRecSignal(GSN_SUB_STOP_REF, &Dbdict::execSUB_STOP_REF);
addRecSignal(GSN_SUB_SYNC_CONF, &Dbdict::execSUB_SYNC_CONF);
addRecSignal(GSN_SUB_SYNC_REF, &Dbdict::execSUB_SYNC_REF);
addRecSignal(GSN_DROP_EVNT_REQ, &Dbdict::execDROP_EVNT_REQ);
addRecSignal(GSN_SUB_REMOVE_REQ, &Dbdict::execSUB_REMOVE_REQ);
addRecSignal(GSN_SUB_REMOVE_CONF, &Dbdict::execSUB_REMOVE_CONF);
addRecSignal(GSN_SUB_REMOVE_REF, &Dbdict::execSUB_REMOVE_REF);
// Trigger signals
addRecSignal(GSN_CREATE_TRIG_REQ, &Dbdict::execCREATE_TRIG_REQ);
addRecSignal(GSN_CREATE_TRIG_CONF, &Dbdict::execCREATE_TRIG_CONF);
addRecSignal(GSN_CREATE_TRIG_REF, &Dbdict::execCREATE_TRIG_REF);
addRecSignal(GSN_ALTER_TRIG_REQ, &Dbdict::execALTER_TRIG_REQ);
addRecSignal(GSN_ALTER_TRIG_CONF, &Dbdict::execALTER_TRIG_CONF);
addRecSignal(GSN_ALTER_TRIG_REF, &Dbdict::execALTER_TRIG_REF);
addRecSignal(GSN_DROP_TRIG_REQ, &Dbdict::execDROP_TRIG_REQ);
addRecSignal(GSN_DROP_TRIG_CONF, &Dbdict::execDROP_TRIG_CONF);
addRecSignal(GSN_DROP_TRIG_REF, &Dbdict::execDROP_TRIG_REF);
// Received signals
addRecSignal(GSN_HOT_SPAREREP, &Dbdict::execHOT_SPAREREP);
addRecSignal(GSN_GET_SCHEMA_INFOREQ, &Dbdict::execGET_SCHEMA_INFOREQ);
addRecSignal(GSN_SCHEMA_INFO, &Dbdict::execSCHEMA_INFO);
addRecSignal(GSN_SCHEMA_INFOCONF, &Dbdict::execSCHEMA_INFOCONF);
addRecSignal(GSN_DICTSTARTREQ, &Dbdict::execDICTSTARTREQ);
addRecSignal(GSN_READ_NODESCONF, &Dbdict::execREAD_NODESCONF);
addRecSignal(GSN_FSOPENCONF, &Dbdict::execFSOPENCONF);
addRecSignal(GSN_FSOPENREF, &Dbdict::execFSOPENREF, true);
addRecSignal(GSN_FSCLOSECONF, &Dbdict::execFSCLOSECONF);
addRecSignal(GSN_FSWRITECONF, &Dbdict::execFSWRITECONF);
addRecSignal(GSN_FSREADCONF, &Dbdict::execFSREADCONF);
addRecSignal(GSN_FSREADREF, &Dbdict::execFSREADREF, true);
addRecSignal(GSN_LQHFRAGCONF, &Dbdict::execLQHFRAGCONF);
addRecSignal(GSN_LQHADDATTCONF, &Dbdict::execLQHADDATTCONF);
addRecSignal(GSN_LQHADDATTREF, &Dbdict::execLQHADDATTREF);
addRecSignal(GSN_LQHFRAGREF, &Dbdict::execLQHFRAGREF);
addRecSignal(GSN_NDB_STTOR, &Dbdict::execNDB_STTOR);
addRecSignal(GSN_READ_CONFIG_REQ, &Dbdict::execREAD_CONFIG_REQ, true);
addRecSignal(GSN_STTOR, &Dbdict::execSTTOR);
addRecSignal(GSN_TC_SCHVERCONF, &Dbdict::execTC_SCHVERCONF);
addRecSignal(GSN_NODE_FAILREP, &Dbdict::execNODE_FAILREP);
addRecSignal(GSN_INCL_NODEREQ, &Dbdict::execINCL_NODEREQ);
addRecSignal(GSN_API_FAILREQ, &Dbdict::execAPI_FAILREQ);
addRecSignal(GSN_WAIT_GCP_REF, &Dbdict::execWAIT_GCP_REF);
addRecSignal(GSN_WAIT_GCP_CONF, &Dbdict::execWAIT_GCP_CONF);
addRecSignal(GSN_LIST_TABLES_REQ, &Dbdict::execLIST_TABLES_REQ);
addRecSignal(GSN_DROP_TABLE_REQ, &Dbdict::execDROP_TABLE_REQ);
addRecSignal(GSN_PREP_DROP_TAB_REQ, &Dbdict::execPREP_DROP_TAB_REQ);
addRecSignal(GSN_PREP_DROP_TAB_REF, &Dbdict::execPREP_DROP_TAB_REF);
addRecSignal(GSN_PREP_DROP_TAB_CONF, &Dbdict::execPREP_DROP_TAB_CONF);
addRecSignal(GSN_DROP_TAB_REQ, &Dbdict::execDROP_TAB_REQ);
addRecSignal(GSN_DROP_TAB_REF, &Dbdict::execDROP_TAB_REF);
addRecSignal(GSN_DROP_TAB_CONF, &Dbdict::execDROP_TAB_CONF);
addRecSignal(GSN_BACKUP_FRAGMENT_REQ, &Dbdict::execBACKUP_FRAGMENT_REQ);
}//Dbdict::Dbdict()
Dbdict::~Dbdict()
{
}//Dbdict::~Dbdict()
BLOCK_FUNCTIONS(Dbdict)
void Dbdict::initCommonData()
{
/* ---------------------------------------------------------------- */
// Initialise all common variables.
/* ---------------------------------------------------------------- */
initRetrieveRecord(0, 0, 0);
initSchemaRecord();
initRestartRecord();
initSendSchemaRecord();
initReadTableRecord();
initWriteTableRecord();
initReadSchemaRecord();
initWriteSchemaRecord();
c_masterNodeId = ZNIL;
c_numberNode = 0;
c_noNodesFailed = 0;
c_failureNr = 0;
c_blockState = BS_IDLE;
c_packTable.m_state = PackTable::PTS_IDLE;
c_startPhase = 0;
c_restartType = 255; //Ensure not used restartType
c_tabinfoReceived = 0;
c_initialStart = false;
c_systemRestart = false;
c_initialNodeRestart = false;
c_nodeRestart = false;
}//Dbdict::initCommonData()
void Dbdict::initRecords()
{
initNodeRecords();
initPageRecords();
initTableRecords();
initTriggerRecords();
}//Dbdict::initRecords()
void Dbdict::initSendSchemaRecord()
{
c_sendSchemaRecord.noOfWords = (Uint32)-1;
c_sendSchemaRecord.pageId = RNIL;
c_sendSchemaRecord.noOfWordsCurrentlySent = 0;
c_sendSchemaRecord.noOfSignalsSentSinceDelay = 0;
c_sendSchemaRecord.inUse = false;
//c_sendSchemaRecord.sendSchemaState = SendSchemaRecord::IDLE;
}//initSendSchemaRecord()
void Dbdict::initReadTableRecord()
{
c_readTableRecord.noOfPages = (Uint32)-1;
c_readTableRecord.pageId = RNIL;
c_readTableRecord.tableId = ZNIL;
c_readTableRecord.inUse = false;
}//initReadTableRecord()
void Dbdict::initWriteTableRecord()
{
c_writeTableRecord.noOfPages = (Uint32)-1;
c_writeTableRecord.pageId = RNIL;
c_writeTableRecord.noOfTableFilesHandled = 3;
c_writeTableRecord.tableId = ZNIL;
c_writeTableRecord.tableWriteState = WriteTableRecord::IDLE;
}//initWriteTableRecord()
void Dbdict::initReadSchemaRecord()
{
c_readSchemaRecord.pageId = RNIL;
c_readSchemaRecord.schemaReadState = ReadSchemaRecord::IDLE;
}//initReadSchemaRecord()
void Dbdict::initWriteSchemaRecord()
{
c_writeSchemaRecord.inUse = false;
c_writeSchemaRecord.pageId = RNIL;
c_writeSchemaRecord.noOfSchemaFilesHandled = 3;
}//initWriteSchemaRecord()
void Dbdict::initRetrieveRecord(Signal* signal, Uint32 i, Uint32 returnCode)
{
c_retrieveRecord.busyState = false;
c_retrieveRecord.blockRef = 0;
c_retrieveRecord.m_senderData = RNIL;
c_retrieveRecord.tableId = RNIL;
c_retrieveRecord.currentSent = 0;
c_retrieveRecord.retrievedNoOfPages = 0;
c_retrieveRecord.retrievedNoOfWords = 0;
c_retrieveRecord.m_useLongSig = false;
}//initRetrieveRecord()
void Dbdict::initSchemaRecord()
{
c_schemaRecord.schemaPage = RNIL;
}//Dbdict::initSchemaRecord()
void Dbdict::initRestartRecord()
{
c_restartRecord.gciToRestart = 0;
c_restartRecord.activeTable = ZNIL;
}//Dbdict::initRestartRecord()
void Dbdict::initNodeRecords()
{
jam();
for (unsigned i = 1; i < MAX_NODES; i++) {
NodeRecordPtr nodePtr;
c_nodes.getPtr(nodePtr, i);
nodePtr.p->hotSpare = false;
nodePtr.p->nodeState = NodeRecord::API_NODE;
}//for
}//Dbdict::initNodeRecords()
void Dbdict::initPageRecords()
{
c_schemaRecord.schemaPage = ZMAX_PAGES_OF_TABLE_DEFINITION;
c_schemaRecord.oldSchemaPage = ZMAX_PAGES_OF_TABLE_DEFINITION + 1;
c_retrieveRecord.retrievePage = ZMAX_PAGES_OF_TABLE_DEFINITION + 2;
ndbrequire(ZNUMBER_OF_PAGES >= (2 * ZMAX_PAGES_OF_TABLE_DEFINITION + 2));
}//Dbdict::initPageRecords()
void Dbdict::initTableRecords()
{
TableRecordPtr tablePtr;
while (1) {
jam();
refresh_watch_dog();
c_tableRecordPool.seize(tablePtr);
if (tablePtr.i == RNIL) {
jam();
break;
}//if
initialiseTableRecord(tablePtr);
}//while
}//Dbdict::initTableRecords()
void Dbdict::initialiseTableRecord(TableRecordPtr tablePtr)
{
tablePtr.p->activePage = RNIL;
tablePtr.p->filePtr[0] = RNIL;
tablePtr.p->filePtr[1] = RNIL;
tablePtr.p->firstAttribute = RNIL;
tablePtr.p->firstPage = RNIL;
tablePtr.p->lastAttribute = RNIL;
tablePtr.p->tableId = tablePtr.i;
tablePtr.p->tableVersion = (Uint32)-1;
tablePtr.p->tabState = TableRecord::NOT_DEFINED;
tablePtr.p->tabReturnState = TableRecord::TRS_IDLE;
tablePtr.p->storageType = DictTabInfo::MainMemory;
tablePtr.p->fragmentType = DictTabInfo::AllNodesSmallTable;
tablePtr.p->fragmentKeyType = DictTabInfo::PrimaryKey;
memset(tablePtr.p->tableName, 0, sizeof(tablePtr.p->tableName));
tablePtr.p->gciTableCreated = 0;
tablePtr.p->noOfAttributes = ZNIL;
tablePtr.p->noOfNullAttr = 0;
tablePtr.p->frmLen = 0;
memset(tablePtr.p->frmData, 0, sizeof(tablePtr.p->frmData));
/*
tablePtr.p->lh3PageIndexBits = 0;
tablePtr.p->lh3DistrBits = 0;
tablePtr.p->lh3PageBits = 6;
*/
tablePtr.p->kValue = 6;
tablePtr.p->localKeyLen = 1;
tablePtr.p->maxLoadFactor = 80;
tablePtr.p->minLoadFactor = 70;
tablePtr.p->noOfPrimkey = 1;
tablePtr.p->tupKeyLength = 1;
tablePtr.p->storedTable = true;
tablePtr.p->tableType = DictTabInfo::UserTable;
tablePtr.p->primaryTableId = RNIL;
// volatile elements
tablePtr.p->indexState = TableRecord::IS_UNDEFINED;
tablePtr.p->insertTriggerId = RNIL;
tablePtr.p->updateTriggerId = RNIL;
tablePtr.p->deleteTriggerId = RNIL;
tablePtr.p->customTriggerId = RNIL;
tablePtr.p->buildTriggerId = RNIL;
tablePtr.p->indexLocal = 0;
}//Dbdict::initialiseTableRecord()
void Dbdict::initTriggerRecords()
{
TriggerRecordPtr triggerPtr;
while (1) {
jam();
refresh_watch_dog();
c_triggerRecordPool.seize(triggerPtr);
if (triggerPtr.i == RNIL) {
jam();
break;
}//if
initialiseTriggerRecord(triggerPtr);
}//while
}
void Dbdict::initialiseTriggerRecord(TriggerRecordPtr triggerPtr)
{
triggerPtr.p->triggerState = TriggerRecord::TS_NOT_DEFINED;
triggerPtr.p->triggerLocal = 0;
memset(triggerPtr.p->triggerName, 0, sizeof(triggerPtr.p->triggerName));
triggerPtr.p->triggerId = RNIL;
triggerPtr.p->tableId = RNIL;
triggerPtr.p->triggerType = (TriggerType::Value)~0;
triggerPtr.p->triggerActionTime = (TriggerActionTime::Value)~0;
triggerPtr.p->triggerEvent = (TriggerEvent::Value)~0;
triggerPtr.p->monitorReplicas = false;
triggerPtr.p->monitorAllAttributes = false;
triggerPtr.p->attributeMask.clear();
triggerPtr.p->indexId = RNIL;
}
Uint32 Dbdict::getFsConnRecord()
{
FsConnectRecordPtr fsPtr;
c_fsConnectRecordPool.seize(fsPtr);
ndbrequire(fsPtr.i != RNIL);
fsPtr.p->filePtr = (Uint32)-1;
fsPtr.p->ownerPtr = RNIL;
fsPtr.p->fsState = FsConnectRecord::IDLE;
return fsPtr.i;
}//Dbdict::getFsConnRecord()
Uint32 Dbdict::getFreeTableRecord(Uint32 primaryTableId)
{
Uint32 minId = (primaryTableId == RNIL ? 0 : primaryTableId + 1);
TableRecordPtr tablePtr;
TableRecordPtr firstTablePtr;
bool firstFound = false;
Uint32 tabSize = c_tableRecordPool.getSize();
for (tablePtr.i = minId; tablePtr.i < tabSize ; tablePtr.i++) {
jam();
c_tableRecordPool.getPtr(tablePtr);
if (tablePtr.p->tabState == TableRecord::NOT_DEFINED) {
jam();
initialiseTableRecord(tablePtr);
tablePtr.p->tabState = TableRecord::DEFINING;
firstFound = true;
firstTablePtr.i = tablePtr.i;
firstTablePtr.p = tablePtr.p;
break;
}//if
}//for
if (!firstFound) {
jam();
return RNIL;
}//if
#ifdef HAVE_TABLE_REORG
bool secondFound = false;
for (tablePtr.i = firstTablePtr.i + 1; tablePtr.i < tabSize ; tablePtr.i++) {
jam();
c_tableRecordPool.getPtr(tablePtr);
if (tablePtr.p->tabState == TableRecord::NOT_DEFINED) {
jam();
initialiseTableRecord(tablePtr);
tablePtr.p->tabState = TableRecord::REORG_TABLE_PREPARED;
tablePtr.p->secondTable = firstTablePtr.i;
firstTablePtr.p->secondTable = tablePtr.i;
secondFound = true;
break;
}//if
}//for
if (!secondFound) {
jam();
firstTablePtr.p->tabState = TableRecord::NOT_DEFINED;
return RNIL;
}//if
#endif
return firstTablePtr.i;
}//Dbdict::getFreeTableRecord()
Uint32 Dbdict::getFreeTriggerRecord()
{
const Uint32 size = c_triggerRecordPool.getSize();
TriggerRecordPtr triggerPtr;
for (triggerPtr.i = 0; triggerPtr.i < size; triggerPtr.i++) {
jam();
c_triggerRecordPool.getPtr(triggerPtr);
if (triggerPtr.p->triggerState == TriggerRecord::TS_NOT_DEFINED) {
jam();
initialiseTriggerRecord(triggerPtr);
return triggerPtr.i;
}
}
return RNIL;
}
bool
Dbdict::getNewAttributeRecord(TableRecordPtr tablePtr,
AttributeRecordPtr & attrPtr)
{
c_attributeRecordPool.seize(attrPtr);
if(attrPtr.i == RNIL){
return false;
}
memset(attrPtr.p->attributeName, 0, sizeof(attrPtr.p->attributeName));
attrPtr.p->attributeDescriptor = 0x00012255; //Default value
attrPtr.p->attributeId = ZNIL;
attrPtr.p->nextAttrInTable = RNIL;
attrPtr.p->tupleKey = 0;
memset(attrPtr.p->defaultValue, 0, sizeof(attrPtr.p->defaultValue));
/* ---------------------------------------------------------------- */
// A free attribute record has been acquired. We will now link it
// to the table record.
/* ---------------------------------------------------------------- */
if (tablePtr.p->lastAttribute == RNIL) {
jam();
tablePtr.p->firstAttribute = attrPtr.i;
} else {
jam();
AttributeRecordPtr lastAttrPtr;
c_attributeRecordPool.getPtr(lastAttrPtr, tablePtr.p->lastAttribute);
lastAttrPtr.p->nextAttrInTable = attrPtr.i;
}//if
tablePtr.p->lastAttribute = attrPtr.i;
return true;
}//Dbdict::getNewAttributeRecord()
/* **************************************************************** */
/* ---------------------------------------------------------------- */
/* MODULE: START/RESTART HANDLING ------------------------ */
/* ---------------------------------------------------------------- */
/* */
/* This module contains the code that is common for all */
/* start/restart types. */
/* ---------------------------------------------------------------- */
/* **************************************************************** */
/* ---------------------------------------------------------------- */
// This is sent as the first signal during start/restart.
/* ---------------------------------------------------------------- */
void Dbdict::execSTTOR(Signal* signal)
{
jamEntry();
c_startPhase = signal->theData[1];
switch (c_startPhase) {
case 1:
break;
case 3:
c_restartType = signal->theData[7]; /* valid if 3 */
ndbrequire(c_restartType == NodeState::ST_INITIAL_START ||
c_restartType == NodeState::ST_SYSTEM_RESTART ||
c_restartType == NodeState::ST_INITIAL_NODE_RESTART ||
c_restartType == NodeState::ST_NODE_RESTART);
break;
}
sendSTTORRY(signal);
}//execSTTOR()
void Dbdict::sendSTTORRY(Signal* signal)
{
signal->theData[0] = 0; /* garbage SIGNAL KEY */
signal->theData[1] = 0; /* garbage SIGNAL VERSION NUMBER */
signal->theData[2] = 0; /* garbage */
signal->theData[3] = 1; /* first wanted start phase */
signal->theData[4] = 3; /* get type of start */
signal->theData[5] = ZNOMOREPHASES;
sendSignal(NDBCNTR_REF, GSN_STTORRY, signal, 6, JBB);
}
/* ---------------------------------------------------------------- */
// We receive information about sizes of records.
/* ---------------------------------------------------------------- */
void Dbdict::execREAD_CONFIG_REQ(Signal* signal)
{
const ReadConfigReq * req = (ReadConfigReq*)signal->getDataPtr();
Uint32 ref = req->senderRef;
Uint32 senderData = req->senderData;
ndbrequire(req->noOfParameters == 0);
jamEntry();
const ndb_mgm_configuration_iterator * p =
theConfiguration.getOwnConfigIterator();
ndbrequire(p != 0);
Uint32 attributesize, tablerecSize;
ndbrequire(!ndb_mgm_get_int_parameter(p, CFG_DICT_ATTRIBUTE,&attributesize));
ndbrequire(!ndb_mgm_get_int_parameter(p, CFG_DICT_TABLE, &tablerecSize));
c_attributeRecordPool.setSize(attributesize);
c_attributeRecordHash.setSize(64);
c_fsConnectRecordPool.setSize(ZFS_CONNECT_SIZE);
c_nodes.setSize(MAX_NODES);
c_pageRecordArray.setSize(ZNUMBER_OF_PAGES);
c_tableRecordPool.setSize(tablerecSize);
c_tableRecordHash.setSize(tablerecSize);
c_triggerRecordPool.setSize(c_maxNoOfTriggers);
c_triggerRecordHash.setSize(c_maxNoOfTriggers);
c_opRecordPool.setSize(256); // XXX need config params
c_opCreateTable.setSize(8);
c_opDropTable.setSize(8);
c_opCreateIndex.setSize(8);
c_opCreateEvent.setSize(8);
c_opSubEvent.setSize(8);
c_opDropEvent.setSize(8);
c_opSignalUtil.setSize(8);
c_opDropIndex.setSize(8);
c_opAlterIndex.setSize(8);
c_opBuildIndex.setSize(8);
c_opCreateTrigger.setSize(8);
c_opDropTrigger.setSize(8);
c_opAlterTrigger.setSize(8);
// Initialize BAT for interface to file system
PageRecordPtr pageRecPtr;
c_pageRecordArray.getPtr(pageRecPtr, 0);
NewVARIABLE* bat = allocateBat(2);
bat[1].WA = &pageRecPtr.p->word[0];
bat[1].nrr = ZNUMBER_OF_PAGES;
bat[1].ClusterSize = ZSIZE_OF_PAGES_IN_WORDS * 4;
bat[1].bits.q = ZLOG_SIZE_OF_PAGES_IN_WORDS; // 2**13 = 8192 elements
bat[1].bits.v = 5; // 32 bits per element
initCommonData();
initRecords();
ReadConfigConf * conf = (ReadConfigConf*)signal->getDataPtrSend();
conf->senderRef = reference();
conf->senderData = senderData;
sendSignal(ref, GSN_READ_CONFIG_CONF, signal,
ReadConfigConf::SignalLength, JBB);
}//execSIZEALT_REP()
/* ---------------------------------------------------------------- */
// Start phase signals sent by CNTR. We reply with NDB_STTORRY when
// we completed this phase.
/* ---------------------------------------------------------------- */
void Dbdict::execNDB_STTOR(Signal* signal)
{
jamEntry();
c_startPhase = signal->theData[2];
const Uint32 restartType = signal->theData[3];
if (restartType == NodeState::ST_INITIAL_START) {
jam();
c_initialStart = true;
} else if (restartType == NodeState::ST_SYSTEM_RESTART) {
jam();
c_systemRestart = true;
} else if (restartType == NodeState::ST_INITIAL_NODE_RESTART) {
jam();
c_initialNodeRestart = true;
} else if (restartType == NodeState::ST_NODE_RESTART) {
jam();
c_nodeRestart = true;
} else {
ndbrequire(false);
}//if
switch (c_startPhase) {
case 1:
jam();
initSchemaFile(signal);
break;
case 3:
jam();
signal->theData[0] = reference();
sendSignal(NDBCNTR_REF, GSN_READ_NODESREQ, signal, 1, JBB);
break;
case 6:
jam();
c_initialStart = false;
c_systemRestart = false;
c_initialNodeRestart = false;
c_nodeRestart = false;
sendNDB_STTORRY(signal);
break;
case 7:
// uses c_restartType
if(restartType == NodeState::ST_SYSTEM_RESTART &&
c_masterNodeId == getOwnNodeId()){
rebuildIndexes(signal, 0);
return;
}
sendNDB_STTORRY(signal);
break;
default:
jam();
sendNDB_STTORRY(signal);
break;
}//switch
}//execNDB_STTOR()
void Dbdict::sendNDB_STTORRY(Signal* signal)
{
signal->theData[0] = reference();
sendSignal(NDBCNTR_REF, GSN_NDB_STTORRY, signal, 1, JBB);
return;
}//sendNDB_STTORRY()
/* ---------------------------------------------------------------- */
// We receive the information about which nodes that are up and down.
/* ---------------------------------------------------------------- */
void Dbdict::execREAD_NODESCONF(Signal* signal)
{
jamEntry();
ReadNodesConf * const readNodes = (ReadNodesConf *)&signal->theData[0];
c_numberNode = readNodes->noOfNodes;
c_masterNodeId = readNodes->masterNodeId;
c_noNodesFailed = 0;
c_aliveNodes.clear();
for (unsigned i = 1; i < MAX_NDB_NODES; i++) {
jam();
NodeRecordPtr nodePtr;
c_nodes.getPtr(nodePtr, i);
if (NodeBitmask::get(readNodes->allNodes, i)) {
jam();
nodePtr.p->nodeState = NodeRecord::NDB_NODE_ALIVE;
if (NodeBitmask::get(readNodes->inactiveNodes, i)) {
jam();
/**-------------------------------------------------------------------
*
* THIS NODE IS DEFINED IN THE CLUSTER BUT IS NOT ALIVE CURRENTLY.
* WE ADD THE NODE TO THE SET OF FAILED NODES AND ALSO SET THE
* BLOCKSTATE TO BUSY TO AVOID ADDING TABLES WHILE NOT ALL NODES ARE
* ALIVE.
*------------------------------------------------------------------*/
nodePtr.p->nodeState = NodeRecord::NDB_NODE_DEAD;
c_noNodesFailed++;
} else {
c_aliveNodes.set(i);
}
}//if
}//for
sendNDB_STTORRY(signal);
}//execREAD_NODESCONF()
/* ---------------------------------------------------------------- */
// HOT_SPAREREP informs DBDICT about which nodes that have become
// hot spare nodes.
/* ---------------------------------------------------------------- */
void Dbdict::execHOT_SPAREREP(Signal* signal)
{
Uint32 hotSpareNodes = 0;
jamEntry();
HotSpareRep * const hotSpare = (HotSpareRep*)&signal->theData[0];
for (unsigned i = 1; i < MAX_NDB_NODES; i++) {
if (NodeBitmask::get(hotSpare->theHotSpareNodes, i)) {
NodeRecordPtr nodePtr;
c_nodes.getPtr(nodePtr, i);
nodePtr.p->hotSpare = true;
hotSpareNodes++;
}//if
}//for
ndbrequire(hotSpareNodes == hotSpare->noHotSpareNodes);
c_noHotSpareNodes = hotSpareNodes;
return;
}//execHOT_SPAREREP()
void Dbdict::initSchemaFile(Signal* signal)
{
PageRecordPtr pagePtr;
c_pageRecordArray.getPtr(pagePtr, c_schemaRecord.schemaPage);
SchemaFile * schemaFile = (SchemaFile *)pagePtr.p;
initSchemaFile(schemaFile, 4 * ZSIZE_OF_PAGES_IN_WORDS);
if (c_initialStart || c_initialNodeRestart) {
jam();
ndbrequire(c_writeSchemaRecord.inUse == false);
c_writeSchemaRecord.inUse = true;
c_writeSchemaRecord.pageId = c_schemaRecord.schemaPage;
c_writeSchemaRecord.m_callback.m_callbackFunction =
safe_cast(&Dbdict::initSchemaFile_conf);
startWriteSchemaFile(signal);
} else if (c_systemRestart || c_nodeRestart) {
jam();
ndbrequire(c_readSchemaRecord.schemaReadState == ReadSchemaRecord::IDLE);
c_readSchemaRecord.pageId = c_schemaRecord.oldSchemaPage;
c_readSchemaRecord.schemaReadState = ReadSchemaRecord::INITIAL_READ;
startReadSchemaFile(signal);
} else {
ndbrequire(false);
}//if
}//Dbdict::initSchemaFile()
void
Dbdict::initSchemaFile_conf(Signal* signal, Uint32 callbackData, Uint32 rv){
jam();
sendNDB_STTORRY(signal);
}
void
Dbdict::activateIndexes(Signal* signal, Uint32 i)
{
AlterIndxReq* req = (AlterIndxReq*)signal->getDataPtrSend();
TableRecordPtr tablePtr;
for (; i < c_tableRecordPool.getSize(); i++) {
tablePtr.i = i;
c_tableRecordPool.getPtr(tablePtr);
if (tablePtr.p->tabState != TableRecord::DEFINED)
continue;
if (! tablePtr.p->isIndex())
continue;
jam();
req->setUserRef(reference());
req->setConnectionPtr(i);
req->setTableId(tablePtr.p->primaryTableId);
req->setIndexId(tablePtr.i);
req->setIndexVersion(tablePtr.p->tableVersion);
req->setOnline(true);
if (c_restartType == NodeState::ST_SYSTEM_RESTART) {
if (c_masterNodeId != getOwnNodeId())
continue;
// from file index state is not defined currently
req->setRequestType(AlterIndxReq::RT_SYSTEMRESTART);
req->addRequestFlag((Uint32)RequestFlag::RF_NOBUILD);
}
else if (
c_restartType == NodeState::ST_NODE_RESTART ||
c_restartType == NodeState::ST_INITIAL_NODE_RESTART) {
// from master index must be online
if (tablePtr.p->indexState != TableRecord::IS_ONLINE)
continue;
req->setRequestType(AlterIndxReq::RT_NODERESTART);
// activate locally, rebuild not needed
req->addRequestFlag((Uint32)RequestFlag::RF_LOCAL);
req->addRequestFlag((Uint32)RequestFlag::RF_NOBUILD);
} else {
ndbrequire(false);
}
sendSignal(reference(), GSN_ALTER_INDX_REQ,
signal, AlterIndxReq::SignalLength, JBB);
return;
}
signal->theData[0] = reference();
sendSignal(c_restartRecord.returnBlockRef, GSN_DICTSTARTCONF,
signal, 1, JBB);
}
void
Dbdict::rebuildIndexes(Signal* signal, Uint32 i){
BuildIndxReq* const req = (BuildIndxReq*)signal->getDataPtrSend();
TableRecordPtr indexPtr;
for (; i < c_tableRecordPool.getSize(); i++) {
indexPtr.i = i;
c_tableRecordPool.getPtr(indexPtr);
if (indexPtr.p->tabState != TableRecord::DEFINED)
continue;
if (! indexPtr.p->isIndex())
continue;
jam();
req->setUserRef(reference());
req->setConnectionPtr(i);
req->setRequestType(BuildIndxReq::RT_SYSTEMRESTART);
req->setBuildId(0); // not used
req->setBuildKey(0); // not used
req->setIndexType(indexPtr.p->tableType);
req->setIndexId(indexPtr.i);
req->setTableId(indexPtr.p->primaryTableId);
req->setParallelism(16);
// from file index state is not defined currently
if (indexPtr.p->storedTable) {
// rebuild not needed
req->addRequestFlag((Uint32)RequestFlag::RF_NOBUILD);
}
// send
sendSignal(reference(), GSN_BUILDINDXREQ,
signal, BuildIndxReq::SignalLength, JBB);
return;
}
sendNDB_STTORRY(signal);
}
/* **************************************************************** */
/* ---------------------------------------------------------------- */
/* MODULE: SYSTEM RESTART MODULE ------------------------- */
/* ---------------------------------------------------------------- */
/* */
/* This module contains code specific for system restart */
/* ---------------------------------------------------------------- */
/* **************************************************************** */
/* ---------------------------------------------------------------- */
// DIH asks DICT to read in table data from disk during system
// restart. DIH also asks DICT to send information about which
// tables that should be started as part of this system restart.
// DICT will also activate the tables in TC as part of this process.
/* ---------------------------------------------------------------- */
void Dbdict::execDICTSTARTREQ(Signal* signal)
{
jamEntry();
c_restartRecord.gciToRestart = signal->theData[0];
c_restartRecord.returnBlockRef = signal->theData[1];
if (c_nodeRestart || c_initialNodeRestart) {
jam();
CRASH_INSERTION(6000);
BlockReference dictRef = calcDictBlockRef(c_masterNodeId);
signal->theData[0] = getOwnNodeId();
sendSignal(dictRef, GSN_GET_SCHEMA_INFOREQ, signal, 1, JBB);
return;
}
ndbrequire(c_systemRestart);
ndbrequire(c_masterNodeId == getOwnNodeId());
c_schemaRecord.m_callback.m_callbackData = 0;
c_schemaRecord.m_callback.m_callbackFunction =
safe_cast(&Dbdict::masterRestart_checkSchemaStatusComplete);
c_restartRecord.activeTable = 0;
c_schemaRecord.schemaPage = c_schemaRecord.oldSchemaPage;
checkSchemaStatus(signal);
}//execDICTSTARTREQ()
void
Dbdict::masterRestart_checkSchemaStatusComplete(Signal* signal,
Uint32 callbackData,
Uint32 returnCode){
c_schemaRecord.schemaPage = ZMAX_PAGES_OF_TABLE_DEFINITION;
LinearSectionPtr ptr[3];
PageRecordPtr pagePtr;
c_pageRecordArray.getPtr(pagePtr, c_schemaRecord.oldSchemaPage);
ptr[0].p = &pagePtr.p->word[0];
ptr[0].sz = ZSIZE_OF_PAGES_IN_WORDS;
c_sendSchemaRecord.m_SCHEMAINFO_Counter = c_aliveNodes;
NodeReceiverGroup rg(DBDICT, c_aliveNodes);
rg.m_nodes.clear(getOwnNodeId());
Callback c = { 0, 0 };
sendFragmentedSignal(rg,
GSN_SCHEMA_INFO,
signal,
1, //SchemaInfo::SignalLength,
JBB,
ptr,
1,
c);
PageRecordPtr newPagePtr;
c_pageRecordArray.getPtr(newPagePtr, c_schemaRecord.schemaPage);
memcpy(&newPagePtr.p->word[0], &pagePtr.p->word[0],
4 * ZSIZE_OF_PAGES_IN_WORDS);
signal->theData[0] = getOwnNodeId();
sendSignal(reference(), GSN_SCHEMA_INFOCONF, signal, 1, JBB);
}
void
Dbdict::execGET_SCHEMA_INFOREQ(Signal* signal){
const Uint32 ref = signal->getSendersBlockRef();
//const Uint32 senderData = signal->theData[0];
ndbrequire(c_sendSchemaRecord.inUse == false);
c_sendSchemaRecord.inUse = true;
LinearSectionPtr ptr[3];
PageRecordPtr pagePtr;
c_pageRecordArray.getPtr(pagePtr, c_schemaRecord.schemaPage);
ptr[0].p = &pagePtr.p->word[0];
ptr[0].sz = ZSIZE_OF_PAGES_IN_WORDS;
Callback c = { safe_cast(&Dbdict::sendSchemaComplete), 0 };
sendFragmentedSignal(ref,
GSN_SCHEMA_INFO,
signal,
1, //GetSchemaInfoConf::SignalLength,
JBB,
ptr,
1,
c);
}//Dbdict::execGET_SCHEMA_INFOREQ()
void
Dbdict::sendSchemaComplete(Signal * signal,
Uint32 callbackData,
Uint32 returnCode){
ndbrequire(c_sendSchemaRecord.inUse == true);
c_sendSchemaRecord.inUse = false;
}
/* ---------------------------------------------------------------- */
// We receive the schema info from master as part of all restarts
// except the initial start where no tables exists.
/* ---------------------------------------------------------------- */
void Dbdict::execSCHEMA_INFO(Signal* signal)
{
jamEntry();
if(!assembleFragments(signal)){
jam();
return;
}
if(getNodeState().getNodeRestartInProgress()){
CRASH_INSERTION(6001);
}
SegmentedSectionPtr schemaDataPtr;
signal->getSection(schemaDataPtr, 0);
PageRecordPtr pagePtr;
c_pageRecordArray.getPtr(pagePtr, c_schemaRecord.schemaPage);
copy(&pagePtr.p->word[0], schemaDataPtr);
releaseSections(signal);
validateChecksum((SchemaFile*)pagePtr.p);
ndbrequire(signal->getSendersBlockRef() != reference());
/* ---------------------------------------------------------------- */
// Synchronise our view on data with other nodes in the cluster.
// This is an important part of restart handling where we will handle
// cases where the table have been added but only partially, where
// tables have been deleted but not completed the deletion yet and
// other scenarios needing synchronisation.
/* ---------------------------------------------------------------- */
c_schemaRecord.m_callback.m_callbackData = 0;
c_schemaRecord.m_callback.m_callbackFunction =
safe_cast(&Dbdict::restart_checkSchemaStatusComplete);
c_restartRecord.activeTable = 0;
checkSchemaStatus(signal);
}//execSCHEMA_INFO()
void
Dbdict::restart_checkSchemaStatusComplete(Signal * signal,
Uint32 callbackData,
Uint32 returnCode){
ndbrequire(c_writeSchemaRecord.inUse == false);
c_writeSchemaRecord.inUse = true;
c_writeSchemaRecord.pageId = c_schemaRecord.schemaPage;
c_writeSchemaRecord.m_callback.m_callbackData = 0;
c_writeSchemaRecord.m_callback.m_callbackFunction =
safe_cast(&Dbdict::restart_writeSchemaConf);
startWriteSchemaFile(signal);
}
void
Dbdict::restart_writeSchemaConf(Signal * signal,
Uint32 callbackData,
Uint32 returnCode){
if(c_systemRestart){
jam();
signal->theData[0] = getOwnNodeId();
sendSignal(calcDictBlockRef(c_masterNodeId), GSN_SCHEMA_INFOCONF,
signal, 1, JBB);
return;
}
ndbrequire(c_nodeRestart || c_initialNodeRestart);
c_blockState = BS_IDLE;
activateIndexes(signal, 0);
return;
}
void Dbdict::execSCHEMA_INFOCONF(Signal* signal)
{
jamEntry();
ndbrequire(signal->getNoOfSections() == 0);
/* ---------------------------------------------------------------- */
// This signal is received in the master as part of system restart
// from all nodes (including the master) after they have synchronised
// their data with the master node's schema information.
/* ---------------------------------------------------------------- */
const Uint32 nodeId = signal->theData[0];
c_sendSchemaRecord.m_SCHEMAINFO_Counter.clearWaitingFor(nodeId);
if (!c_sendSchemaRecord.m_SCHEMAINFO_Counter.done()){
jam();
return;
}//if
activateIndexes(signal, 0);
}//execSCHEMA_INFOCONF()
void Dbdict::checkSchemaStatus(Signal* signal)
{
PageRecordPtr pagePtr;
c_pageRecordArray.getPtr(pagePtr, c_schemaRecord.schemaPage);
PageRecordPtr oldPagePtr;
c_pageRecordArray.getPtr(oldPagePtr, c_schemaRecord.oldSchemaPage);
for (; c_restartRecord.activeTable < MAX_TABLES;
c_restartRecord.activeTable++) {
jam();
Uint32 tableId = c_restartRecord.activeTable;
SchemaFile::TableEntry *newEntry = getTableEntry(pagePtr.p, tableId);
SchemaFile::TableEntry *oldEntry = getTableEntry(oldPagePtr.p, tableId,
true);
SchemaFile::TableState schemaState =
(SchemaFile::TableState)newEntry->m_tableState;
SchemaFile::TableState oldSchemaState =
(SchemaFile::TableState)oldEntry->m_tableState;
if (c_restartRecord.activeTable >= c_tableRecordPool.getSize()) {
jam();
ndbrequire(schemaState == SchemaFile::INIT);
ndbrequire(oldSchemaState == SchemaFile::INIT);
continue;
}//if
switch(schemaState){
case SchemaFile::INIT:{
jam();
bool ok = false;
switch(oldSchemaState) {
case SchemaFile::INIT:
jam();
case SchemaFile::DROP_TABLE_COMMITTED:
jam();
ok = true;
jam();
break;
case SchemaFile::ADD_STARTED:
jam();
case SchemaFile::TABLE_ADD_COMMITTED:
jam();
case SchemaFile::DROP_TABLE_STARTED:
jam();
case SchemaFile::ALTER_TABLE_COMMITTED:
jam();
ok = true;
jam();
newEntry->m_tableState = SchemaFile::INIT;
restartDropTab(signal, tableId);
return;
}//switch
ndbrequire(ok);
break;
}
case SchemaFile::ADD_STARTED:{
jam();
bool ok = false;
switch(oldSchemaState) {
case SchemaFile::INIT:
jam();
case SchemaFile::DROP_TABLE_COMMITTED:
jam();
ok = true;
break;
case SchemaFile::ADD_STARTED:
jam();
case SchemaFile::DROP_TABLE_STARTED:
jam();
case SchemaFile::TABLE_ADD_COMMITTED:
jam();
case SchemaFile::ALTER_TABLE_COMMITTED:
jam();
ok = true;
//------------------------------------------------------------------
// Add Table was started but not completed. Will be dropped in all
// nodes. Update schema information (restore table version).
//------------------------------------------------------------------
newEntry->m_tableState = SchemaFile::INIT;
restartDropTab(signal, tableId);
return;
}
ndbrequire(ok);
break;
}
case SchemaFile::TABLE_ADD_COMMITTED:{
jam();
bool ok = false;
switch(oldSchemaState) {
case SchemaFile::INIT:
jam();
case SchemaFile::ADD_STARTED:
jam();
case SchemaFile::DROP_TABLE_STARTED:
jam();
case SchemaFile::DROP_TABLE_COMMITTED:
jam();
ok = true;
//------------------------------------------------------------------
// Table was added in the master node but not in our node. We can
// retrieve the table definition from the master.
//------------------------------------------------------------------
restartCreateTab(signal, tableId, oldEntry, false);
return;
break;
case SchemaFile::TABLE_ADD_COMMITTED:
jam();
case SchemaFile::ALTER_TABLE_COMMITTED:
jam();
ok = true;
//------------------------------------------------------------------
// Table was added in both our node and the master node. We can
// retrieve the table definition from our own disk.
//------------------------------------------------------------------
if(* newEntry == * oldEntry){
jam();
TableRecordPtr tablePtr;
c_tableRecordPool.getPtr(tablePtr, tableId);
tablePtr.p->tableVersion = oldEntry->m_tableVersion;
tablePtr.p->tableType = (DictTabInfo::TableType)oldEntry->m_tableType;
// On NR get index from master because index state is not on file
const bool file = c_systemRestart || tablePtr.p->isTable();
restartCreateTab(signal, tableId, oldEntry, file);
return;
} else {
//------------------------------------------------------------------
// Must be a new version of the table if anything differs. Both table
// version and global checkpoint must be different.
// This should not happen for the master node. This can happen after
// drop table followed by add table or after change table.
// Not supported in this version.
//------------------------------------------------------------------
ndbrequire(c_masterNodeId != getOwnNodeId());
ndbrequire(newEntry->m_tableVersion != oldEntry->m_tableVersion);
jam();
restartCreateTab(signal, tableId, oldEntry, false);
return;
}//if
}
ndbrequire(ok);
break;
}
case SchemaFile::DROP_TABLE_STARTED:
jam();
case SchemaFile::DROP_TABLE_COMMITTED:{
jam();
bool ok = false;
switch(oldSchemaState){
case SchemaFile::INIT:
jam();
case SchemaFile::DROP_TABLE_COMMITTED:
jam();
ok = true;
break;
case SchemaFile::ADD_STARTED:
jam();
case SchemaFile::TABLE_ADD_COMMITTED:
jam();
case SchemaFile::DROP_TABLE_STARTED:
jam();
case SchemaFile::ALTER_TABLE_COMMITTED:
jam();
newEntry->m_tableState = SchemaFile::INIT;
restartDropTab(signal, tableId);
return;
}
ndbrequire(ok);
break;
}
case SchemaFile::ALTER_TABLE_COMMITTED: {
jam();
bool ok = false;
switch(oldSchemaState) {
case SchemaFile::INIT:
jam();
case SchemaFile::ADD_STARTED:
jam();
case SchemaFile::DROP_TABLE_STARTED:
jam();
case SchemaFile::DROP_TABLE_COMMITTED:
jam();
case SchemaFile::TABLE_ADD_COMMITTED:
jam();
ok = true;
//------------------------------------------------------------------
// Table was altered in the master node but not in our node. We can
// retrieve the altered table definition from the master.
//------------------------------------------------------------------
restartCreateTab(signal, tableId, oldEntry, false);
return;
break;
case SchemaFile::ALTER_TABLE_COMMITTED:
jam();
ok = true;
//------------------------------------------------------------------
// Table was altered in both our node and the master node. We can
// retrieve the table definition from our own disk.
//------------------------------------------------------------------
TableRecordPtr tablePtr;
c_tableRecordPool.getPtr(tablePtr, tableId);
tablePtr.p->tableVersion = oldEntry->m_tableVersion;
tablePtr.p->tableType = (DictTabInfo::TableType)oldEntry->m_tableType;
// On NR get index from master because index state is not on file
const bool file = c_systemRestart || tablePtr.p->isTable();
restartCreateTab(signal, tableId, oldEntry, file);
return;
}
ndbrequire(ok);
break;
}
}
}
execute(signal, c_schemaRecord.m_callback, 0);
}//checkSchemaStatus()
void
Dbdict::restartCreateTab(Signal* signal, Uint32 tableId,
const SchemaFile::TableEntry * te, bool file){
jam();
CreateTableRecordPtr createTabPtr;
c_opCreateTable.seize(createTabPtr);
ndbrequire(!createTabPtr.isNull());
createTabPtr.p->key = ++c_opRecordSequence;
c_opCreateTable.add(createTabPtr);
createTabPtr.p->m_errorCode = 0;
createTabPtr.p->m_tablePtrI = tableId;
createTabPtr.p->m_coordinatorRef = reference();
createTabPtr.p->m_senderRef = 0;
createTabPtr.p->m_senderData = RNIL;
createTabPtr.p->m_tabInfoPtrI = RNIL;
createTabPtr.p->m_dihAddFragPtr = RNIL;
if(file && !ERROR_INSERTED(6002)){
jam();
c_readTableRecord.noOfPages = te->m_noOfPages;
c_readTableRecord.pageId = 0;
c_readTableRecord.m_callback.m_callbackData = createTabPtr.p->key;
c_readTableRecord.m_callback.m_callbackFunction =
safe_cast(&Dbdict::restartCreateTab_readTableConf);
startReadTableFile(signal, tableId);
return;
} else {
ndbrequire(c_masterNodeId != getOwnNodeId());
/**
* Get from master
*/
GetTabInfoReq * const req = (GetTabInfoReq *)&signal->theData[0];
req->senderRef = reference();
req->senderData = createTabPtr.p->key;
req->requestType = GetTabInfoReq::RequestById |
GetTabInfoReq::LongSignalConf;
req->tableId = tableId;
sendSignal(calcDictBlockRef(c_masterNodeId), GSN_GET_TABINFOREQ, signal,
GetTabInfoReq::SignalLength, JBB);
if(ERROR_INSERTED(6002)){
NdbSleep_MilliSleep(10);
CRASH_INSERTION(6002);
}
}
}
void
Dbdict::restartCreateTab_readTableConf(Signal* signal,
Uint32 callbackData,
Uint32 returnCode){
jam();
PageRecordPtr pageRecPtr;
c_pageRecordArray.getPtr(pageRecPtr, c_readTableRecord.pageId);
ParseDictTabInfoRecord parseRecord;
parseRecord.requestType = DictTabInfo::GetTabInfoConf;
parseRecord.errorCode = 0;
Uint32 sz = c_readTableRecord.noOfPages * ZSIZE_OF_PAGES_IN_WORDS;
SimplePropertiesLinearReader r(&pageRecPtr.p->word[0], sz);
handleTabInfoInit(r, &parseRecord);
ndbrequire(parseRecord.errorCode == 0);
/* ---------------------------------------------------------------- */
// We have read the table description from disk as part of system restart.
// We will also write it back again to ensure that both copies are ok.
/* ---------------------------------------------------------------- */
ndbrequire(c_writeTableRecord.tableWriteState == WriteTableRecord::IDLE);
c_writeTableRecord.noOfPages = c_readTableRecord.noOfPages;
c_writeTableRecord.pageId = c_readTableRecord.pageId;
c_writeTableRecord.tableWriteState = WriteTableRecord::TWR_CALLBACK;
c_writeTableRecord.m_callback.m_callbackData = callbackData;
c_writeTableRecord.m_callback.m_callbackFunction =
safe_cast(&Dbdict::restartCreateTab_writeTableConf);
startWriteTableFile(signal, c_readTableRecord.tableId);
}
void
Dbdict::execGET_TABINFO_CONF(Signal* signal){
jamEntry();
if(!assembleFragments(signal)){
jam();
return;
}
GetTabInfoConf * const conf = (GetTabInfoConf*)signal->getDataPtr();
const Uint32 tableId = conf->tableId;
const Uint32 senderData = conf->senderData;
SegmentedSectionPtr tabInfoPtr;
signal->getSection(tabInfoPtr, GetTabInfoConf::DICT_TAB_INFO);
CreateTableRecordPtr createTabPtr;
ndbrequire(c_opCreateTable.find(createTabPtr, senderData));
ndbrequire(!createTabPtr.isNull());
ndbrequire(createTabPtr.p->m_tablePtrI == tableId);
/**
* Put data into table record
*/
ParseDictTabInfoRecord parseRecord;
parseRecord.requestType = DictTabInfo::GetTabInfoConf;
parseRecord.errorCode = 0;
SimplePropertiesSectionReader r(tabInfoPtr, getSectionSegmentPool());
handleTabInfoInit(r, &parseRecord);
ndbrequire(parseRecord.errorCode == 0);
Callback callback;
callback.m_callbackData = createTabPtr.p->key;
callback.m_callbackFunction =
safe_cast(&Dbdict::restartCreateTab_writeTableConf);
signal->header.m_noOfSections = 0;
writeTableFile(signal, createTabPtr.p->m_tablePtrI, tabInfoPtr, &callback);
signal->setSection(tabInfoPtr, 0);
releaseSections(signal);
}
void
Dbdict::restartCreateTab_writeTableConf(Signal* signal,
Uint32 callbackData,
Uint32 returnCode){
jam();
CreateTableRecordPtr createTabPtr;
ndbrequire(c_opCreateTable.find(createTabPtr, callbackData));
Callback callback;
callback.m_callbackData = callbackData;
callback.m_callbackFunction =
safe_cast(&Dbdict::restartCreateTab_dihComplete);
SegmentedSectionPtr fragDataPtr;
fragDataPtr.sz = 0;
fragDataPtr.setNull();
createTab_dih(signal, createTabPtr, fragDataPtr, &callback);
}
void
Dbdict::restartCreateTab_dihComplete(Signal* signal,
Uint32 callbackData,
Uint32 returnCode){
jam();
CreateTableRecordPtr createTabPtr;
ndbrequire(c_opCreateTable.find(createTabPtr, callbackData));
//@todo check error
ndbrequire(createTabPtr.p->m_errorCode == 0);
Callback callback;
callback.m_callbackData = callbackData;
callback.m_callbackFunction =
safe_cast(&Dbdict::restartCreateTab_activateComplete);
alterTab_activate(signal, createTabPtr, &callback);
}
void
Dbdict::restartCreateTab_activateComplete(Signal* signal,
Uint32 callbackData,
Uint32 returnCode){
jam();
CreateTableRecordPtr createTabPtr;
ndbrequire(c_opCreateTable.find(createTabPtr, callbackData));
TableRecordPtr tabPtr;
c_tableRecordPool.getPtr(tabPtr, createTabPtr.p->m_tablePtrI);
tabPtr.p->tabState = TableRecord::DEFINED;
c_opCreateTable.release(createTabPtr);
c_restartRecord.activeTable++;
checkSchemaStatus(signal);
}
void
Dbdict::restartDropTab(Signal* signal, Uint32 tableId){
const Uint32 key = ++c_opRecordSequence;
DropTableRecordPtr dropTabPtr;
ndbrequire(c_opDropTable.seize(dropTabPtr));
dropTabPtr.p->key = key;
c_opDropTable.add(dropTabPtr);
dropTabPtr.p->m_errorCode = 0;
dropTabPtr.p->m_request.tableId = tableId;
dropTabPtr.p->m_coordinatorRef = 0;
dropTabPtr.p->m_requestType = DropTabReq::RestartDropTab;
dropTabPtr.p->m_participantData.m_gsn = GSN_DROP_TAB_REQ;
dropTabPtr.p->m_participantData.m_block = 0;
dropTabPtr.p->m_participantData.m_callback.m_callbackData = key;
dropTabPtr.p->m_participantData.m_callback.m_callbackFunction =
safe_cast(&Dbdict::restartDropTab_complete);
dropTab_nextStep(signal, dropTabPtr);
}
void
Dbdict::restartDropTab_complete(Signal* signal,
Uint32 callbackData,
Uint32 returnCode){
jam();
DropTableRecordPtr dropTabPtr;
ndbrequire(c_opDropTable.find(dropTabPtr, callbackData));
//@todo check error
c_opDropTable.release(dropTabPtr);
c_restartRecord.activeTable++;
checkSchemaStatus(signal);
}
/* **************************************************************** */
/* ---------------------------------------------------------------- */
/* MODULE: NODE FAILURE HANDLING ------------------------- */
/* ---------------------------------------------------------------- */
/* */
/* This module contains the code that is used when nodes */
/* (kernel/api) fails. */
/* ---------------------------------------------------------------- */
/* **************************************************************** */
/* ---------------------------------------------------------------- */
// We receive a report of an API that failed.
/* ---------------------------------------------------------------- */
void Dbdict::execAPI_FAILREQ(Signal* signal)
{
jamEntry();
Uint32 failedApiNode = signal->theData[0];
BlockReference retRef = signal->theData[1];
#if 0
Uint32 userNode = refToNode(c_connRecord.userBlockRef);
if (userNode == failedApiNode) {
jam();
c_connRecord.userBlockRef = (Uint32)-1;
}//if
#endif
signal->theData[0] = failedApiNode;
signal->theData[1] = reference();
sendSignal(retRef, GSN_API_FAILCONF, signal, 2, JBB);
}//execAPI_FAILREQ()
/* ---------------------------------------------------------------- */
// We receive a report of one or more node failures of kernel nodes.
/* ---------------------------------------------------------------- */
void Dbdict::execNODE_FAILREP(Signal* signal)
{
jamEntry();
NodeFailRep * const nodeFail = (NodeFailRep *)&signal->theData[0];
c_failureNr = nodeFail->failNo;
const Uint32 numberOfFailedNodes = nodeFail->noOfNodes;
const bool masterFailed = (c_masterNodeId != nodeFail->masterNodeId);
c_masterNodeId = nodeFail->masterNodeId;
c_noNodesFailed += numberOfFailedNodes;
Uint32 theFailedNodes[NodeBitmask::Size];
memcpy(theFailedNodes, nodeFail->theNodes, sizeof(theFailedNodes));
c_counterMgr.execNODE_FAILREP(signal);
bool ok = false;
switch(c_blockState){
case BS_IDLE:
jam();
ok = true;
if(c_opRecordPool.getSize() != c_opRecordPool.getNoOfFree()){
jam();
c_blockState = BS_NODE_FAILURE;
}
break;
case BS_CREATE_TAB:
jam();
ok = true;
if(!masterFailed)
break;
// fall through
case BS_BUSY:
case BS_NODE_FAILURE:
jam();
c_blockState = BS_NODE_FAILURE;
ok = true;
break;
}
ndbrequire(ok);
for(unsigned i = 1; i < MAX_NDB_NODES; i++) {
jam();
if(NodeBitmask::get(theFailedNodes, i)) {
jam();
NodeRecordPtr nodePtr;
c_nodes.getPtr(nodePtr, i);
nodePtr.p->nodeState = NodeRecord::NDB_NODE_DEAD;
NFCompleteRep * const nfCompRep = (NFCompleteRep *)&signal->theData[0];
nfCompRep->blockNo = DBDICT;
nfCompRep->nodeId = getOwnNodeId();
nfCompRep->failedNodeId = nodePtr.i;
sendSignal(DBDIH_REF, GSN_NF_COMPLETEREP, signal,
NFCompleteRep::SignalLength, JBB);
c_aliveNodes.clear(i);
}//if
}//for
}//execNODE_FAILREP()
/* **************************************************************** */
/* ---------------------------------------------------------------- */
/* MODULE: NODE START HANDLING --------------------------- */
/* ---------------------------------------------------------------- */
/* */
/* This module contains the code that is used when kernel nodes */
/* starts. */
/* ---------------------------------------------------------------- */
/* **************************************************************** */
/* ---------------------------------------------------------------- */
// Include a starting node in list of nodes to be part of adding
// and dropping tables.
/* ---------------------------------------------------------------- */
void Dbdict::execINCL_NODEREQ(Signal* signal)
{
jamEntry();
NodeRecordPtr nodePtr;
BlockReference retRef = signal->theData[0];
nodePtr.i = signal->theData[1];
ndbrequire(c_noNodesFailed > 0);
c_noNodesFailed--;
c_nodes.getPtr(nodePtr);
ndbrequire(nodePtr.p->nodeState == NodeRecord::NDB_NODE_DEAD);
nodePtr.p->nodeState = NodeRecord::NDB_NODE_ALIVE;
signal->theData[0] = reference();
sendSignal(retRef, GSN_INCL_NODECONF, signal, 1, JBB);
c_aliveNodes.set(nodePtr.i);
}//execINCL_NODEREQ()
/* **************************************************************** */
/* ---------------------------------------------------------------- */
/* MODULE: ADD TABLE HANDLING ---------------------------- */
/* ---------------------------------------------------------------- */
/* */
/* This module contains the code that is used when adding a table. */
/* ---------------------------------------------------------------- */
/* **************************************************************** */
/* ---------------------------------------------------------------- */
// This signal receives information about a table from either:
// API, Ndbcntr or from other DICT.
/* ---------------------------------------------------------------- */
void
Dbdict::execCREATE_TABLE_REQ(Signal* signal){
jamEntry();
if(!assembleFragments(signal)){
return;
}
CreateTableReq* const req = (CreateTableReq*)signal->getDataPtr();
const Uint32 senderRef = req->senderRef;
const Uint32 senderData = req->senderData;
ParseDictTabInfoRecord parseRecord;
do {
if(getOwnNodeId() != c_masterNodeId){
jam();
parseRecord.errorCode = CreateTableRef::NotMaster;
break;
}
if (c_blockState != BS_IDLE){
jam();
parseRecord.errorCode = CreateTableRef::Busy;
break;
}
CreateTableRecordPtr createTabPtr;
c_opCreateTable.seize(createTabPtr);
if(createTabPtr.isNull()){
jam();
parseRecord.errorCode = CreateTableRef::Busy;
break;
}
parseRecord.requestType = DictTabInfo::CreateTableFromAPI;
parseRecord.errorCode = 0;
SegmentedSectionPtr ptr;
signal->getSection(ptr, CreateTableReq::DICT_TAB_INFO);
SimplePropertiesSectionReader r(ptr, getSectionSegmentPool());
handleTabInfoInit(r, &parseRecord);
releaseSections(signal);
if(parseRecord.errorCode != 0){
jam();
c_opCreateTable.release(createTabPtr);
break;
}
createTabPtr.p->key = ++c_opRecordSequence;
c_opCreateTable.add(createTabPtr);
createTabPtr.p->m_errorCode = 0;
createTabPtr.p->m_senderRef = senderRef;
createTabPtr.p->m_senderData = senderData;
createTabPtr.p->m_tablePtrI = parseRecord.tablePtr.i;
createTabPtr.p->m_coordinatorRef = reference();
createTabPtr.p->m_fragmentsPtrI = RNIL;
createTabPtr.p->m_dihAddFragPtr = RNIL;
Uint32 * theData = signal->getDataPtrSend();
CreateFragmentationReq * const req = (CreateFragmentationReq*)theData;
req->senderRef = reference();
req->senderData = createTabPtr.p->key;
req->fragmentationType = parseRecord.tablePtr.p->fragmentType;
req->noOfFragments = 0;
req->fragmentNode = 0;
req->primaryTableId = RNIL;
if (parseRecord.tablePtr.p->isOrderedIndex()) {
// ordered index has same fragmentation as the table
const Uint32 primaryTableId = parseRecord.tablePtr.p->primaryTableId;
TableRecordPtr primaryTablePtr;
c_tableRecordPool.getPtr(primaryTablePtr, primaryTableId);
// fragmentationType must be consistent
req->fragmentationType = primaryTablePtr.p->fragmentType;
req->primaryTableId = primaryTableId;
}
sendSignal(DBDIH_REF, GSN_CREATE_FRAGMENTATION_REQ, signal,
CreateFragmentationReq::SignalLength, JBB);
c_blockState = BS_CREATE_TAB;
return;
} while(0);
/**
* Something went wrong
*/
releaseSections(signal);
CreateTableRef * ref = (CreateTableRef*)signal->getDataPtrSend();
ref->senderData = senderData;
ref->senderRef = reference();
ref->masterNodeId = c_masterNodeId;
ref->errorCode = parseRecord.errorCode;
ref->errorLine = parseRecord.errorLine;
ref->errorKey = parseRecord.errorKey;
ref->status = parseRecord.status;
sendSignal(senderRef, GSN_CREATE_TABLE_REF, signal,
CreateTableRef::SignalLength, JBB);
}
void
Dbdict::execBACKUP_FRAGMENT_REQ(Signal* signal)
{
jamEntry();
Uint32 tableId = signal->theData[0];
Uint32 lock = signal->theData[1];
TableRecordPtr tablePtr;
c_tableRecordPool.getPtr(tablePtr, tableId, true);
if(lock)
{
ndbrequire(tablePtr.p->tabState == TableRecord::DEFINED);
tablePtr.p->tabState = TableRecord::BACKUP_ONGOING;
}
else if(tablePtr.p->tabState == TableRecord::BACKUP_ONGOING)
{
tablePtr.p->tabState = TableRecord::DEFINED;
}
}
bool
Dbdict::check_ndb_versions() const
{
Uint32 node = 0;
Uint32 version = getNodeInfo(getOwnNodeId()).m_version;
while((node = c_aliveNodes.find(node + 1)) != BitmaskImpl::NotFound)
{
if(getNodeInfo(node).m_version != version)
{
return false;
}
}
return true;
}
void
Dbdict::execALTER_TABLE_REQ(Signal* signal)
{
// Received by master
jamEntry();
if(!assembleFragments(signal)){
return;
}
AlterTableReq* const req = (AlterTableReq*)signal->getDataPtr();
const Uint32 senderRef = req->senderRef;
const Uint32 senderData = req->senderData;
const Uint32 changeMask = req->changeMask;
const Uint32 tableId = req->tableId;
const Uint32 tableVersion = req->tableVersion;
ParseDictTabInfoRecord* aParseRecord;
// Get table definition
TableRecordPtr tablePtr;
c_tableRecordPool.getPtr(tablePtr, tableId, false);
if(tablePtr.isNull()){
jam();
alterTableRef(signal, req, AlterTableRef::NoSuchTable);
return;
}
if(getOwnNodeId() != c_masterNodeId){
jam();
alterTableRef(signal, req, AlterTableRef::NotMaster);
return;
}
if(c_blockState != BS_IDLE){
jam();
alterTableRef(signal, req, AlterTableRef::Busy);
return;
}
if (!check_ndb_versions())
{
jam();
alterTableRef(signal, req, AlterTableRef::IncompatibleVersions);
return;
}
const TableRecord::TabState tabState = tablePtr.p->tabState;
bool ok = false;
switch(tabState){
case TableRecord::NOT_DEFINED:
case TableRecord::REORG_TABLE_PREPARED:
case TableRecord::DEFINING:
case TableRecord::CHECKED:
jam();
alterTableRef(signal, req, AlterTableRef::NoSuchTable);
return;
case TableRecord::DEFINED:
ok = true;
jam();
break;
case TableRecord::BACKUP_ONGOING:
jam();
alterTableRef(signal, req, AlterTableRef::BackupInProgress);
return;
case TableRecord::PREPARE_DROPPING:
case TableRecord::DROPPING:
jam();
alterTableRef(signal, req, AlterTableRef::DropInProgress);
return;
}
ndbrequire(ok);
if(tablePtr.p->tableVersion != tableVersion){
jam();
alterTableRef(signal, req, AlterTableRef::InvalidTableVersion);
return;
}
// Parse new table defintion
ParseDictTabInfoRecord parseRecord;
aParseRecord = &parseRecord;
CreateTableRecordPtr alterTabPtr; // Reuse create table records
c_opCreateTable.seize(alterTabPtr);
if(alterTabPtr.isNull()){
jam();
alterTableRef(signal, req, AlterTableRef::Busy);
return;
}
alterTabPtr.p->m_changeMask = changeMask;
parseRecord.requestType = DictTabInfo::AlterTableFromAPI;
parseRecord.errorCode = 0;
SegmentedSectionPtr ptr;
signal->getSection(ptr, AlterTableReq::DICT_TAB_INFO);
SimplePropertiesSectionReader r(ptr, getSectionSegmentPool());
handleTabInfoInit(r, &parseRecord, false); // Will not save info
if(parseRecord.errorCode != 0){
jam();
c_opCreateTable.release(alterTabPtr);
alterTableRef(signal, req,
(AlterTableRef::ErrorCode) parseRecord.errorCode,
aParseRecord);
return;
}
releaseSections(signal);
alterTabPtr.p->key = ++c_opRecordSequence;
c_opCreateTable.add(alterTabPtr);
ndbrequire(c_opCreateTable.find(alterTabPtr, alterTabPtr.p->key));
alterTabPtr.p->m_errorCode = 0;
alterTabPtr.p->m_senderRef = senderRef;
alterTabPtr.p->m_senderData = senderData;
alterTabPtr.p->m_tablePtrI = parseRecord.tablePtr.i;
alterTabPtr.p->m_alterTableFailed = false;
alterTabPtr.p->m_coordinatorRef = reference();
alterTabPtr.p->m_fragmentsPtrI = RNIL;
alterTabPtr.p->m_dihAddFragPtr = RNIL;
alterTabPtr.p->m_alterTableId = tablePtr.p->tableId;
// Send prepare request to all alive nodes
SimplePropertiesSectionWriter w(getSectionSegmentPool());
packTableIntoPagesImpl(w, parseRecord.tablePtr);
SegmentedSectionPtr tabInfoPtr;
w.getPtr(tabInfoPtr);
alterTabPtr.p->m_tabInfoPtrI = tabInfoPtr.i;
// Alter table on all nodes
c_blockState = BS_BUSY;
Mutex mutex(signal, c_mutexMgr, alterTabPtr.p->m_startLcpMutex);
Callback c = { safe_cast(&Dbdict::alterTable_backup_mutex_locked),
alterTabPtr.p->key };
ndbrequire(mutex.lock(c));
}
void
Dbdict::alterTable_backup_mutex_locked(Signal* signal,
Uint32 callbackData,
Uint32 retValue)
{
jamEntry();
ndbrequire(retValue == 0);
CreateTableRecordPtr alterTabPtr;
ndbrequire(c_opCreateTable.find(alterTabPtr, callbackData));
TableRecordPtr tablePtr;
c_tableRecordPool.getPtr(tablePtr, alterTabPtr.p->m_alterTableId, true);
Mutex mutex(signal, c_mutexMgr, alterTabPtr.p->m_startLcpMutex);
mutex.unlock(); // ignore response
SegmentedSectionPtr tabInfoPtr;
getSection(tabInfoPtr, alterTabPtr.p->m_tabInfoPtrI);
signal->setSection(tabInfoPtr, AlterTabReq::DICT_TAB_INFO);
alterTabPtr.p->m_tabInfoPtrI = RNIL;
if(tablePtr.p->tabState == TableRecord::BACKUP_ONGOING)
{
jam();
AlterTableReq* req = (AlterTableReq*)signal->getDataPtr();
req->senderData = alterTabPtr.p->m_senderData;
req->senderRef = alterTabPtr.p->m_senderRef;
alterTableRef(signal, req, AlterTableRef::BackupInProgress);
c_tableRecordPool.getPtr(tablePtr, alterTabPtr.p->m_tablePtrI);
releaseTableObject(tablePtr.i, false);
c_opCreateTable.release(alterTabPtr);
c_blockState = BS_IDLE;
return;
}
NodeReceiverGroup rg(DBDICT, c_aliveNodes);
alterTabPtr.p->m_coordinatorData.m_gsn = GSN_ALTER_TAB_REQ;
SafeCounter safeCounter(c_counterMgr,
alterTabPtr.p->m_coordinatorData.m_counter);
safeCounter.init<AlterTabRef>(rg, alterTabPtr.p->key);
AlterTabReq * const lreq = (AlterTabReq*)signal->getDataPtrSend();
lreq->senderRef = reference();
lreq->senderData = alterTabPtr.p->key;
lreq->clientRef = alterTabPtr.p->m_senderRef;
lreq->clientData = alterTabPtr.p->m_senderData;
lreq->changeMask = alterTabPtr.p->m_changeMask;
lreq->tableId = tablePtr.p->tableId;
lreq->tableVersion = alter_table_inc_schema_version(tablePtr.p->tableVersion);
lreq->gci = tablePtr.p->gciTableCreated;
lreq->requestType = AlterTabReq::AlterTablePrepare;
sendFragmentedSignal(rg, GSN_ALTER_TAB_REQ, signal,
AlterTabReq::SignalLength, JBB);
}
void Dbdict::alterTableRef(Signal * signal,
AlterTableReq * req,
AlterTableRef::ErrorCode errCode,
ParseDictTabInfoRecord* parseRecord)
{
jam();
releaseSections(signal);
AlterTableRef * ref = (AlterTableRef*)signal->getDataPtrSend();
Uint32 senderRef = req->senderRef;
ref->senderData = req->senderData;
ref->senderRef = reference();
ref->masterNodeId = c_masterNodeId;
if (parseRecord) {
ref->errorCode = parseRecord->errorCode;
ref->errorLine = parseRecord->errorLine;
ref->errorKey = parseRecord->errorKey;
ref->status = parseRecord->status;
}
else {
ref->errorCode = errCode;
ref->errorLine = 0;
ref->errorKey = 0;
ref->status = 0;
}
sendSignal(senderRef, GSN_ALTER_TABLE_REF, signal,
AlterTableRef::SignalLength, JBB);
}
void
Dbdict::execALTER_TAB_REQ(Signal * signal)
{
// Received in all nodes to handle change locally
jamEntry();
if(!assembleFragments(signal)){
return;
}
AlterTabReq* const req = (AlterTabReq*)signal->getDataPtr();
const Uint32 senderRef = req->senderRef;
const Uint32 senderData = req->senderData;
const Uint32 changeMask = req->changeMask;
const Uint32 tableId = req->tableId;
const Uint32 tableVersion = req->tableVersion;
const Uint32 gci = req->gci;
AlterTabReq::RequestType requestType =
(AlterTabReq::RequestType) req->requestType;
SegmentedSectionPtr tabInfoPtr;
signal->getSection(tabInfoPtr, AlterTabReq::DICT_TAB_INFO);
CreateTableRecordPtr alterTabPtr; // Reuse create table records
if (senderRef != reference()) {
jam();
c_blockState = BS_BUSY;
}
if ((requestType == AlterTabReq::AlterTablePrepare)
&& (senderRef != reference())) {
jam();
c_opCreateTable.seize(alterTabPtr);
if(!alterTabPtr.isNull())
alterTabPtr.p->m_changeMask = changeMask;
}
else {
jam();
ndbrequire(c_opCreateTable.find(alterTabPtr, senderData));
}
if(alterTabPtr.isNull()){
jam();
alterTabRef(signal, req, AlterTableRef::Busy);
return;
}
if (!check_ndb_versions())
{
jam();
alterTabRef(signal, req, AlterTableRef::IncompatibleVersions);
return;
}
alterTabPtr.p->m_alterTableId = tableId;
alterTabPtr.p->m_coordinatorRef = senderRef;
// Get table definition
TableRecordPtr tablePtr;
c_tableRecordPool.getPtr(tablePtr, tableId, false);
if(tablePtr.isNull()){
jam();
alterTabRef(signal, req, AlterTableRef::NoSuchTable);
return;
}
switch(requestType) {
case(AlterTabReq::AlterTablePrepare): {
ParseDictTabInfoRecord* aParseRecord;
const TableRecord::TabState tabState = tablePtr.p->tabState;
bool ok = false;
switch(tabState){
case TableRecord::NOT_DEFINED:
case TableRecord::REORG_TABLE_PREPARED:
case TableRecord::DEFINING:
case TableRecord::CHECKED:
jam();
alterTabRef(signal, req, AlterTableRef::NoSuchTable);
return;
case TableRecord::DEFINED:
ok = true;
jam();
break;
case TableRecord::PREPARE_DROPPING:
case TableRecord::DROPPING:
jam();
alterTabRef(signal, req, AlterTableRef::DropInProgress);
return;
case TableRecord::BACKUP_ONGOING:
jam();
alterTabRef(signal, req, AlterTableRef::BackupInProgress);
return;
}
ndbrequire(ok);
if(alter_table_inc_schema_version(tablePtr.p->tableVersion) != tableVersion){
jam();
alterTabRef(signal, req, AlterTableRef::InvalidTableVersion);
return;
}
TableRecordPtr newTablePtr;
if (senderRef != reference()) {
jam();
// Parse altered table defintion
ParseDictTabInfoRecord parseRecord;
aParseRecord = &parseRecord;
parseRecord.requestType = DictTabInfo::AlterTableFromAPI;
parseRecord.errorCode = 0;
SimplePropertiesSectionReader r(tabInfoPtr, getSectionSegmentPool());
handleTabInfoInit(r, &parseRecord, false); // Will not save info
if(parseRecord.errorCode != 0){
jam();
c_opCreateTable.release(alterTabPtr);
alterTabRef(signal, req,
(AlterTableRef::ErrorCode) parseRecord.errorCode,
aParseRecord);
return;
}
alterTabPtr.p->key = senderData;
c_opCreateTable.add(alterTabPtr);
alterTabPtr.p->m_errorCode = 0;
alterTabPtr.p->m_senderRef = senderRef;
alterTabPtr.p->m_senderData = senderData;
alterTabPtr.p->m_tablePtrI = parseRecord.tablePtr.i;
alterTabPtr.p->m_fragmentsPtrI = RNIL;
alterTabPtr.p->m_dihAddFragPtr = RNIL;
newTablePtr = parseRecord.tablePtr;
newTablePtr.p->tableVersion = tableVersion;
}
else { // (req->senderRef == reference())
jam();
c_tableRecordPool.getPtr(newTablePtr, alterTabPtr.p->m_tablePtrI);
newTablePtr.p->tableVersion = tableVersion;
}
if (handleAlterTab(req, alterTabPtr.p, tablePtr, newTablePtr) == -1) {
jam();
c_opCreateTable.release(alterTabPtr);
alterTabRef(signal, req, AlterTableRef::UnsupportedChange);
return;
}
releaseSections(signal);
// Propagate alter table to other local blocks
AlterTabReq * req = (AlterTabReq*)signal->getDataPtrSend();
req->senderRef = reference();
req->senderData = senderData;
req->changeMask = changeMask;
req->tableId = tableId;
req->tableVersion = tableVersion;
req->gci = gci;
req->requestType = requestType;
sendSignal(DBLQH_REF, GSN_ALTER_TAB_REQ, signal,
AlterTabReq::SignalLength, JBB);
return;
}
case(AlterTabReq::AlterTableCommit): {
jam();
// Write schema for altered table to disk
SegmentedSectionPtr tabInfoPtr;
signal->getSection(tabInfoPtr, AlterTabReq::DICT_TAB_INFO);
alterTabPtr.p->m_tabInfoPtrI = tabInfoPtr.i;
signal->header.m_noOfSections = 0;
// Update table record
tablePtr.p->packedSize = tabInfoPtr.sz;
tablePtr.p->tableVersion = tableVersion;
tablePtr.p->gciTableCreated = gci;
SchemaFile::TableEntry tabEntry;
tabEntry.m_tableVersion = tableVersion;
tabEntry.m_tableType = tablePtr.p->tableType;
tabEntry.m_tableState = SchemaFile::ALTER_TABLE_COMMITTED;
tabEntry.m_gcp = gci;
tabEntry.m_noOfPages =
DIV(tabInfoPtr.sz + ZPAGE_HEADER_SIZE, ZSIZE_OF_PAGES_IN_WORDS);
Callback callback;
callback.m_callbackData = senderData;
callback.m_callbackFunction =
safe_cast(&Dbdict::alterTab_writeSchemaConf);
updateSchemaState(signal, tableId, &tabEntry, &callback);
break;
}
case(AlterTabReq::AlterTableRevert): {
jam();
// Revert failed alter table
revertAlterTable(signal, changeMask, tableId, alterTabPtr.p);
// Acknowledge the reverted alter table
AlterTabConf * conf = (AlterTabConf*)signal->getDataPtrSend();
conf->senderRef = reference();
conf->senderData = senderData;
conf->changeMask = changeMask;
conf->tableId = tableId;
conf->tableVersion = tableVersion;
conf->gci = gci;
conf->requestType = requestType;
sendSignal(senderRef, GSN_ALTER_TAB_CONF, signal,
AlterTabConf::SignalLength, JBB);
break;
}
default: ndbrequire(false);
}
}
void Dbdict::alterTabRef(Signal * signal,
AlterTabReq * req,
AlterTableRef::ErrorCode errCode,
ParseDictTabInfoRecord* parseRecord)
{
jam();
releaseSections(signal);
AlterTabRef * ref = (AlterTabRef*)signal->getDataPtrSend();
Uint32 senderRef = req->senderRef;
ref->senderData = req->senderData;
ref->senderRef = reference();
if (parseRecord) {
jam();
ref->errorCode = parseRecord->errorCode;
ref->errorLine = parseRecord->errorLine;
ref->errorKey = parseRecord->errorKey;
ref->errorStatus = parseRecord->status;
}
else {
jam();
ref->errorCode = errCode;
ref->errorLine = 0;
ref->errorKey = 0;
ref->errorStatus = 0;
}
sendSignal(senderRef, GSN_ALTER_TAB_REF, signal,
AlterTabRef::SignalLength, JBB);
c_blockState = BS_IDLE;
}
void Dbdict::execALTER_TAB_REF(Signal * signal){
jamEntry();
AlterTabRef * ref = (AlterTabRef*)signal->getDataPtr();
Uint32 senderRef = ref->senderRef;
Uint32 senderData = ref->senderData;
Uint32 errorCode = ref->errorCode;
Uint32 errorLine = ref->errorLine;
Uint32 errorKey = ref->errorKey;
Uint32 errorStatus = ref->errorStatus;
AlterTabReq::RequestType requestType =
(AlterTabReq::RequestType) ref->requestType;
CreateTableRecordPtr alterTabPtr;
ndbrequire(c_opCreateTable.find(alterTabPtr, senderData));
Uint32 changeMask = alterTabPtr.p->m_changeMask;
SafeCounter safeCounter(c_counterMgr, alterTabPtr.p->m_coordinatorData.m_counter);
safeCounter.clearWaitingFor(refToNode(senderRef));
switch (requestType) {
case(AlterTabReq::AlterTablePrepare): {
if (safeCounter.done()) {
jam();
// Send revert request to all alive nodes
TableRecordPtr tablePtr;
c_tableRecordPool.getPtr(tablePtr, alterTabPtr.p->m_alterTableId);
Uint32 tableId = tablePtr.p->tableId;
Uint32 tableVersion = tablePtr.p->tableVersion;
Uint32 gci = tablePtr.p->gciTableCreated;
SimplePropertiesSectionWriter w(getSectionSegmentPool());
packTableIntoPagesImpl(w, tablePtr);
SegmentedSectionPtr spDataPtr;
w.getPtr(spDataPtr);
signal->setSection(spDataPtr, AlterTabReq::DICT_TAB_INFO);
NodeReceiverGroup rg(DBDICT, c_aliveNodes);
alterTabPtr.p->m_coordinatorData.m_gsn = GSN_ALTER_TAB_REQ;
safeCounter.init<AlterTabRef>(rg, alterTabPtr.p->key);
AlterTabReq * const lreq = (AlterTabReq*)signal->getDataPtrSend();
lreq->senderRef = reference();
lreq->senderData = alterTabPtr.p->key;
lreq->clientRef = alterTabPtr.p->m_senderRef;
lreq->clientData = alterTabPtr.p->m_senderData;
lreq->changeMask = changeMask;
lreq->tableId = tableId;
lreq->tableVersion = tableVersion;
lreq->gci = gci;
lreq->requestType = AlterTabReq::AlterTableRevert;
sendSignal(rg, GSN_ALTER_TAB_REQ, signal,
AlterTabReq::SignalLength, JBB);
}
else {
jam();
alterTabPtr.p->m_alterTableFailed = true;
}
break;
}
case(AlterTabReq::AlterTableCommit):
jam();
case(AlterTabReq::AlterTableRevert): {
AlterTableRef * apiRef = (AlterTableRef*)signal->getDataPtrSend();
apiRef->senderData = senderData;
apiRef->senderRef = reference();
apiRef->masterNodeId = c_masterNodeId;
apiRef->errorCode = errorCode;
apiRef->errorLine = errorLine;
apiRef->errorKey = errorKey;
apiRef->status = errorStatus;
if (safeCounter.done()) {
jam();
sendSignal(senderRef, GSN_ALTER_TABLE_REF, signal,
AlterTableRef::SignalLength, JBB);
c_blockState = BS_IDLE;
}
else {
jam();
alterTabPtr.p->m_alterTableFailed = true;
alterTabPtr.p->m_alterTableRef = *apiRef;
}
break;
}
default: ndbrequire(false);
}
}
void
Dbdict::execALTER_TAB_CONF(Signal * signal){
jamEntry();
AlterTabConf * const conf = (AlterTabConf*)signal->getDataPtr();
Uint32 senderRef = conf->senderRef;
Uint32 senderData = conf->senderData;
Uint32 changeMask = conf->changeMask;
Uint32 tableId = conf->tableId;
Uint32 tableVersion = conf->tableVersion;
Uint32 gci = conf->gci;
AlterTabReq::RequestType requestType =
(AlterTabReq::RequestType) conf->requestType;
CreateTableRecordPtr alterTabPtr;
ndbrequire(c_opCreateTable.find(alterTabPtr, senderData));
switch (requestType) {
case(AlterTabReq::AlterTablePrepare): {
switch(refToBlock(signal->getSendersBlockRef())) {
case DBLQH: {
jam();
AlterTabReq * req = (AlterTabReq*)signal->getDataPtrSend();
req->senderRef = reference();
req->senderData = senderData;
req->changeMask = changeMask;
req->tableId = tableId;
req->tableVersion = tableVersion;
req->gci = gci;
req->requestType = requestType;
sendSignal(DBDIH_REF, GSN_ALTER_TAB_REQ, signal,
AlterTabReq::SignalLength, JBB);
return;
}
case DBDIH: {
jam();
AlterTabReq * req = (AlterTabReq*)signal->getDataPtrSend();
req->senderRef = reference();
req->senderData = senderData;
req->changeMask = changeMask;
req->tableId = tableId;
req->tableVersion = tableVersion;
req->gci = gci;
req->requestType = requestType;
sendSignal(DBTC_REF, GSN_ALTER_TAB_REQ, signal,
AlterTabReq::SignalLength, JBB);
return;
}
case DBTC: {
jam();
// Participant is done with prepare phase, send conf to coordinator
AlterTabConf * conf = (AlterTabConf*)signal->getDataPtrSend();
conf->senderRef = reference();
conf->senderData = senderData;
conf->changeMask = changeMask;
conf->tableId = tableId;
conf->tableVersion = tableVersion;
conf->gci = gci;
conf->requestType = requestType;
sendSignal(alterTabPtr.p->m_coordinatorRef, GSN_ALTER_TAB_CONF, signal,
AlterTabConf::SignalLength, JBB);
return;
}
default :break;
}
// Coordinator only
SafeCounter safeCounter(c_counterMgr, alterTabPtr.p->m_coordinatorData.m_counter);
safeCounter.clearWaitingFor(refToNode(senderRef));
if (safeCounter.done()) {
jam();
// We have received all local confirmations
if (alterTabPtr.p->m_alterTableFailed) {
jam();
// Send revert request to all alive nodes
TableRecordPtr tablePtr;
c_tableRecordPool.getPtr(tablePtr, alterTabPtr.p->m_alterTableId);
Uint32 tableId = tablePtr.p->tableId;
Uint32 tableVersion = tablePtr.p->tableVersion;
Uint32 gci = tablePtr.p->gciTableCreated;
SimplePropertiesSectionWriter w(getSectionSegmentPool());
packTableIntoPagesImpl(w, tablePtr);
SegmentedSectionPtr spDataPtr;
w.getPtr(spDataPtr);
signal->setSection(spDataPtr, AlterTabReq::DICT_TAB_INFO);
NodeReceiverGroup rg(DBDICT, c_aliveNodes);
alterTabPtr.p->m_coordinatorData.m_gsn = GSN_ALTER_TAB_REQ;
safeCounter.init<AlterTabRef>(rg, alterTabPtr.p->key);
AlterTabReq * const lreq = (AlterTabReq*)signal->getDataPtrSend();
lreq->senderRef = reference();
lreq->senderData = alterTabPtr.p->key;
lreq->clientRef = alterTabPtr.p->m_senderRef;
lreq->clientData = alterTabPtr.p->m_senderData;
lreq->changeMask = changeMask;
lreq->tableId = tableId;
lreq->tableVersion = tableVersion;
lreq->gci = gci;
lreq->requestType = AlterTabReq::AlterTableRevert;
sendSignal(rg, GSN_ALTER_TAB_REQ, signal,
AlterTabReq::SignalLength, JBB);
}
else {
jam();
// Send commit request to all alive nodes
TableRecordPtr tablePtr;
c_tableRecordPool.getPtr(tablePtr, tableId);
SimplePropertiesSectionWriter w(getSectionSegmentPool());
packTableIntoPagesImpl(w, tablePtr);
SegmentedSectionPtr spDataPtr;
w.getPtr(spDataPtr);
signal->setSection(spDataPtr, AlterTabReq::DICT_TAB_INFO);
NodeReceiverGroup rg(DBDICT, c_aliveNodes);
alterTabPtr.p->m_coordinatorData.m_gsn = GSN_ALTER_TAB_REQ;
safeCounter.init<AlterTabRef>(rg, alterTabPtr.p->key);
AlterTabReq * const lreq = (AlterTabReq*)signal->getDataPtrSend();
lreq->senderRef = reference();
lreq->senderData = alterTabPtr.p->key;
lreq->clientRef = alterTabPtr.p->m_senderRef;
lreq->clientData = alterTabPtr.p->m_senderData;
lreq->changeMask = changeMask;
lreq->tableId = tableId;
lreq->tableVersion = tableVersion;
lreq->gci = gci;
lreq->requestType = AlterTabReq::AlterTableCommit;
sendFragmentedSignal(rg, GSN_ALTER_TAB_REQ, signal,
AlterTabReq::SignalLength, JBB);
}
}
else {
// (!safeCounter.done())
jam();
}
break;
}
case(AlterTabReq::AlterTableRevert):
jam();
case(AlterTabReq::AlterTableCommit): {
SafeCounter safeCounter(c_counterMgr, alterTabPtr.p->m_coordinatorData.m_counter);
safeCounter.clearWaitingFor(refToNode(senderRef));
if (safeCounter.done()) {
jam();
// We have received all local confirmations
releaseSections(signal);
if (alterTabPtr.p->m_alterTableFailed) {
jam();
AlterTableRef * apiRef =
(AlterTableRef*)signal->getDataPtrSend();
*apiRef = alterTabPtr.p->m_alterTableRef;
sendSignal(alterTabPtr.p->m_senderRef, GSN_ALTER_TABLE_REF, signal,
AlterTableRef::SignalLength, JBB);
}
else {
jam();
// Alter table completed, inform API
AlterTableConf * const apiConf =
(AlterTableConf*)signal->getDataPtrSend();
apiConf->senderRef = reference();
apiConf->senderData = alterTabPtr.p->m_senderData;
apiConf->tableId = tableId;
apiConf->tableVersion = tableVersion;
//@todo check api failed
sendSignal(alterTabPtr.p->m_senderRef, GSN_ALTER_TABLE_CONF, signal,
AlterTableConf::SignalLength, JBB);
}
// Release resources
TableRecordPtr tabPtr;
c_tableRecordPool.getPtr(tabPtr, alterTabPtr.p->m_tablePtrI);
releaseTableObject(tabPtr.i, false);
c_opCreateTable.release(alterTabPtr);
c_blockState = BS_IDLE;
}
else {
// (!safeCounter.done())
jam();
}
break;
}
default: ndbrequire(false);
}
}
// For debugging
inline
void Dbdict::printTables()
{
DLHashTable<TableRecord>::Iterator iter;
bool moreTables = c_tableRecordHash.first(iter);
printf("TABLES IN DICT:n");
while (moreTables) {
TableRecordPtr tablePtr = iter.curr;
printf("%s ", tablePtr.p->tableName);
moreTables = c_tableRecordHash.next(iter);
}
printf("n");
}
int Dbdict::handleAlterTab(AlterTabReq * req,
CreateTableRecord * alterTabPtrP,
TableRecordPtr origTablePtr,
TableRecordPtr newTablePtr)
{
Uint32 changeMask = req->changeMask;
if (AlterTableReq::getNameFlag(changeMask)) {
jam();
// Table rename
// Remove from hashtable
#ifdef VM_TRACE
TableRecordPtr tmp;
ndbrequire(c_tableRecordHash.find(tmp, *origTablePtr.p));
#endif
c_tableRecordHash.remove(origTablePtr);
strcpy(alterTabPtrP->previousTableName, origTablePtr.p->tableName);
strcpy(origTablePtr.p->tableName, newTablePtr.p->tableName);
// Set new schema version
origTablePtr.p->tableVersion = newTablePtr.p->tableVersion;
// Put it back
#ifdef VM_TRACE
ndbrequire(!c_tableRecordHash.find(tmp, *origTablePtr.p));
#endif
c_tableRecordHash.add(origTablePtr);
return 0;
}
jam();
return -1;
}
void Dbdict::revertAlterTable(Signal * signal,
Uint32 changeMask,
Uint32 tableId,
CreateTableRecord * alterTabPtrP)
{
if (AlterTableReq::getNameFlag(changeMask)) {
jam();
// Table rename
// Restore previous name
TableRecordPtr tablePtr;
c_tableRecordPool.getPtr(tablePtr, tableId);
// Remove from hashtable
#ifdef VM_TRACE
TableRecordPtr tmp;
ndbrequire(c_tableRecordHash.find(tmp, * tablePtr.p));
#endif
c_tableRecordHash.remove(tablePtr);
// Restore name
strcpy(tablePtr.p->tableName, alterTabPtrP->previousTableName);
// Revert schema version
tablePtr.p->tableVersion = alter_table_dec_schema_version(tablePtr.p->tableVersion);
// Put it back
#ifdef VM_TRACE
ndbrequire(!c_tableRecordHash.find(tmp, * tablePtr.p));
#endif
c_tableRecordHash.add(tablePtr);
return;
}
ndbrequire(false);
}
void
Dbdict::alterTab_writeSchemaConf(Signal* signal,
Uint32 callbackData,
Uint32 returnCode)
{
jam();
Uint32 key = callbackData;
CreateTableRecordPtr alterTabPtr;
ndbrequire(c_opCreateTable.find(alterTabPtr, key));
Uint32 tableId = alterTabPtr.p->m_alterTableId;
Callback callback;
callback.m_callbackData = alterTabPtr.p->key;
callback.m_callbackFunction =
safe_cast(&Dbdict::alterTab_writeTableConf);
SegmentedSectionPtr tabInfoPtr;
getSection(tabInfoPtr, alterTabPtr.p->m_tabInfoPtrI);
writeTableFile(signal, tableId, tabInfoPtr, &callback);
signal->setSection(tabInfoPtr, 0);
releaseSections(signal);
}
void
Dbdict::alterTab_writeTableConf(Signal* signal,
Uint32 callbackData,
Uint32 returnCode)
{
jam();
CreateTableRecordPtr alterTabPtr;
ndbrequire(c_opCreateTable.find(alterTabPtr, callbackData));
Uint32 coordinatorRef = alterTabPtr.p->m_coordinatorRef;
TableRecordPtr tabPtr;
c_tableRecordPool.getPtr(tabPtr, alterTabPtr.p->m_alterTableId);
// Alter table commit request handled successfully
AlterTabConf * conf = (AlterTabConf*)signal->getDataPtrSend();
conf->senderRef = reference();
conf->senderData = callbackData;
conf->tableId = tabPtr.p->tableId;
conf->tableVersion = tabPtr.p->tableVersion;
conf->gci = tabPtr.p->gciTableCreated;
conf->requestType = AlterTabReq::AlterTableCommit;
sendSignal(coordinatorRef, GSN_ALTER_TAB_CONF, signal,
AlterTabConf::SignalLength, JBB);
{
ApiBroadcastRep* api= (ApiBroadcastRep*)signal->getDataPtrSend();
api->gsn = GSN_ALTER_TABLE_REP;
api->minVersion = MAKE_VERSION(4,1,15);
AlterTableRep* rep = (AlterTableRep*)api->theData;
rep->tableId = tabPtr.p->tableId;
rep->tableVersion = alter_table_dec_schema_version(tabPtr.p->tableVersion);
rep->changeType = AlterTableRep::CT_ALTERED;
LinearSectionPtr ptr[3];
ptr[0].p = (Uint32*)alterTabPtr.p->previousTableName;
ptr[0].sz = (sizeof(alterTabPtr.p->previousTableName) + 3) >> 2;
sendSignal(QMGR_REF, GSN_API_BROADCAST_REP, signal,
ApiBroadcastRep::SignalLength + AlterTableRep::SignalLength,
JBB, ptr,1);
}
if(coordinatorRef != reference()) {
jam();
// Release resources
c_tableRecordPool.getPtr(tabPtr, alterTabPtr.p->m_tablePtrI);
releaseTableObject(tabPtr.i, false);
c_opCreateTable.release(alterTabPtr);
c_blockState = BS_IDLE;
}
}
void
Dbdict::execCREATE_FRAGMENTATION_REF(Signal * signal){
jamEntry();
const Uint32 * theData = signal->getDataPtr();
CreateFragmentationRef * const ref = (CreateFragmentationRef*)theData;
(void)ref;
ndbrequire(false);
}
void
Dbdict::execCREATE_FRAGMENTATION_CONF(Signal* signal){
jamEntry();
const Uint32 * theData = signal->getDataPtr();
CreateFragmentationConf * const conf = (CreateFragmentationConf*)theData;
CreateTableRecordPtr createTabPtr;
ndbrequire(c_opCreateTable.find(createTabPtr, conf->senderData));
ndbrequire(signal->getNoOfSections() == 1);
SegmentedSectionPtr fragDataPtr;
signal->getSection(fragDataPtr, CreateFragmentationConf::FRAGMENTS);
signal->header.m_noOfSections = 0;
/**
* Get table
*/
TableRecordPtr tabPtr;
c_tableRecordPool.getPtr(tabPtr, createTabPtr.p->m_tablePtrI);
/**
* Save fragment count
*/
tabPtr.p->fragmentCount = conf->noOfFragments;
/**
* Update table version
*/
PageRecordPtr pagePtr;
c_pageRecordArray.getPtr(pagePtr, c_schemaRecord.schemaPage);
SchemaFile::TableEntry * tabEntry = getTableEntry(pagePtr.p, tabPtr.i);
tabPtr.p->tableVersion = create_table_inc_schema_version(tabEntry->m_tableVersion);
/**
* Pack
*/
SimplePropertiesSectionWriter w(getSectionSegmentPool());
packTableIntoPagesImpl(w, tabPtr);
SegmentedSectionPtr spDataPtr;
w.getPtr(spDataPtr);
signal->setSection(spDataPtr, CreateTabReq::DICT_TAB_INFO);
signal->setSection(fragDataPtr, CreateTabReq::FRAGMENTATION);
NodeReceiverGroup rg(DBDICT, c_aliveNodes);
SafeCounter tmp(c_counterMgr, createTabPtr.p->m_coordinatorData.m_counter);
createTabPtr.p->m_coordinatorData.m_gsn = GSN_CREATE_TAB_REQ;
createTabPtr.p->m_coordinatorData.m_requestType = CreateTabReq::CreateTablePrepare;
tmp.init<CreateTabRef>(rg, GSN_CREATE_TAB_REF, createTabPtr.p->key);
CreateTabReq * const req = (CreateTabReq*)theData;
req->senderRef = reference();
req->senderData = createTabPtr.p->key;
req->clientRef = createTabPtr.p->m_senderRef;
req->clientData = createTabPtr.p->m_senderData;
req->requestType = CreateTabReq::CreateTablePrepare;
req->gci = 0;
req->tableId = tabPtr.i;
req->tableVersion = create_table_inc_schema_version(tabEntry->m_tableVersion);
sendFragmentedSignal(rg, GSN_CREATE_TAB_REQ, signal,
CreateTabReq::SignalLength, JBB);
return;
}
void
Dbdict::execCREATE_TAB_REF(Signal* signal){
jamEntry();
CreateTabRef * const ref = (CreateTabRef*)signal->getDataPtr();
CreateTableRecordPtr createTabPtr;
ndbrequire(c_opCreateTable.find(createTabPtr, ref->senderData));
ndbrequire(createTabPtr.p->m_coordinatorRef == reference());
ndbrequire(createTabPtr.p->m_coordinatorData.m_gsn == GSN_CREATE_TAB_REQ);
if(ref->errorCode != CreateTabRef::NF_FakeErrorREF){
createTabPtr.p->setErrorCode(ref->errorCode);
}
createTab_reply(signal, createTabPtr, refToNode(ref->senderRef));
}
void
Dbdict::execCREATE_TAB_CONF(Signal* signal){
jamEntry();
ndbrequire(signal->getNoOfSections() == 0);
CreateTabConf * const conf = (CreateTabConf*)signal->getDataPtr();
CreateTableRecordPtr createTabPtr;
ndbrequire(c_opCreateTable.find(createTabPtr, conf->senderData));
ndbrequire(createTabPtr.p->m_coordinatorRef == reference());
ndbrequire(createTabPtr.p->m_coordinatorData.m_gsn == GSN_CREATE_TAB_REQ);
createTab_reply(signal, createTabPtr, refToNode(conf->senderRef));
}
void
Dbdict::createTab_reply(Signal* signal,
CreateTableRecordPtr createTabPtr,
Uint32 nodeId)
{
SafeCounter tmp(c_counterMgr, createTabPtr.p->m_coordinatorData.m_counter);
if(!tmp.clearWaitingFor(nodeId)){
jam();
return;
}
switch(createTabPtr.p->m_coordinatorData.m_requestType){
case CreateTabReq::CreateTablePrepare:{
if(createTabPtr.p->m_errorCode != 0){
jam();
/**
* Failed to prepare on atleast one node -> abort on all
*/
NodeReceiverGroup rg(DBDICT, c_aliveNodes);
createTabPtr.p->m_coordinatorData.m_gsn = GSN_CREATE_TAB_REQ;
createTabPtr.p->m_coordinatorData.m_requestType = CreateTabReq::CreateTableDrop;
ndbrequire(tmp.init<CreateTabRef>(rg, createTabPtr.p->key));
CreateTabReq * const req = (CreateTabReq*)signal->getDataPtrSend();
req->senderRef = reference();
req->senderData = createTabPtr.p->key;
req->requestType = CreateTabReq::CreateTableDrop;
sendSignal(rg, GSN_CREATE_TAB_REQ, signal,
CreateTabReq::SignalLength, JBB);
return;
}
/**
* Lock mutex before commiting table
*/
Mutex mutex(signal, c_mutexMgr, createTabPtr.p->m_startLcpMutex);
Callback c = { safe_cast(&Dbdict::createTab_startLcpMutex_locked),
createTabPtr.p->key};
ndbrequire(mutex.lock(c));
return;
}
case CreateTabReq::CreateTableCommit:{
jam();
ndbrequire(createTabPtr.p->m_errorCode == 0);
/**
* Unlock mutex before commiting table
*/
Mutex mutex(signal, c_mutexMgr, createTabPtr.p->m_startLcpMutex);
Callback c = { safe_cast(&Dbdict::createTab_startLcpMutex_unlocked),
createTabPtr.p->key};
mutex.unlock(c);
return;
}
case CreateTabReq::CreateTableDrop:{
jam();
CreateTableRef * const ref = (CreateTableRef*)signal->getDataPtr();
ref->senderRef = reference();
ref->senderData = createTabPtr.p->m_senderData;
ref->errorCode = createTabPtr.p->m_errorCode;
ref->masterNodeId = c_masterNodeId;
ref->status = 0;
ref->errorKey = 0;
ref->errorLine = 0;
//@todo check api failed
sendSignal(createTabPtr.p->m_senderRef, GSN_CREATE_TABLE_REF, signal,
CreateTableRef::SignalLength, JBB);
c_opCreateTable.release(createTabPtr);
c_blockState = BS_IDLE;
return;
}
}
ndbrequire(false);
}
void
Dbdict::createTab_startLcpMutex_locked(Signal* signal,
Uint32 callbackData,
Uint32 retValue){
jamEntry();
ndbrequire(retValue == 0);
CreateTableRecordPtr createTabPtr;
ndbrequire(c_opCreateTable.find(createTabPtr, callbackData));
NodeReceiverGroup rg(DBDICT, c_aliveNodes);
createTabPtr.p->m_coordinatorData.m_gsn = GSN_CREATE_TAB_REQ;
createTabPtr.p->m_coordinatorData.m_requestType = CreateTabReq::CreateTableCommit;
SafeCounter tmp(c_counterMgr, createTabPtr.p->m_coordinatorData.m_counter);
tmp.init<CreateTabRef>(rg, GSN_CREATE_TAB_REF, createTabPtr.p->key);
CreateTabReq * const req = (CreateTabReq*)signal->getDataPtrSend();
req->senderRef = reference();
req->senderData = createTabPtr.p->key;
req->requestType = CreateTabReq::CreateTableCommit;
sendSignal(rg, GSN_CREATE_TAB_REQ, signal,
CreateTabReq::SignalLength, JBB);
}
void
Dbdict::createTab_startLcpMutex_unlocked(Signal* signal,
Uint32 callbackData,
Uint32 retValue){
jamEntry();
ndbrequire(retValue == 0);
CreateTableRecordPtr createTabPtr;
ndbrequire(c_opCreateTable.find(createTabPtr, callbackData));
createTabPtr.p->m_startLcpMutex.release(c_mutexMgr);
TableRecordPtr tabPtr;
c_tableRecordPool.getPtr(tabPtr, createTabPtr.p->m_tablePtrI);
CreateTableConf * const conf = (CreateTableConf*)signal->getDataPtr();
conf->senderRef = reference();
conf->senderData = createTabPtr.p->m_senderData;
conf->tableId = createTabPtr.p->m_tablePtrI;
conf->tableVersion = tabPtr.p->tableVersion;
//@todo check api failed
sendSignal(createTabPtr.p->m_senderRef, GSN_CREATE_TABLE_CONF, signal,
CreateTableConf::SignalLength, JBB);
c_opCreateTable.release(createTabPtr);
c_blockState = BS_IDLE;
return;
}
/***********************************************************
* CreateTable participant code
**********************************************************/
void
Dbdict::execCREATE_TAB_REQ(Signal* signal){
jamEntry();
if(!assembleFragments(signal)){
jam();
return;
}
CreateTabReq * const req = (CreateTabReq*)signal->getDataPtr();
CreateTabReq::RequestType rt = (CreateTabReq::RequestType)req->requestType;
switch(rt){
case CreateTabReq::CreateTablePrepare:
CRASH_INSERTION2(6003, getOwnNodeId() != c_masterNodeId);
createTab_prepare(signal, req);
return;
case CreateTabReq::CreateTableCommit:
CRASH_INSERTION2(6004, getOwnNodeId() != c_masterNodeId);
createTab_commit(signal, req);
return;
case CreateTabReq::CreateTableDrop:
CRASH_INSERTION2(6005, getOwnNodeId() != c_masterNodeId);
createTab_drop(signal, req);
return;
}
ndbrequire(false);
}