mirror/mariadb
1
0
Fork 0
mirror of https://github.com/MariaDB/server.git synced 2025-01-16 20:12:31 +01:00
Code Issues Projects Releases Packages Wiki Activity Actions

Merge whalegate.ndb.mysql.com:/home/tomas/mysql-5.1

Browse source 
into  whalegate.ndb.mysql.com:/home/tomas/mysql-5.1-new-ndb-merge


sql/ha_ndbcluster_binlog.cc:
  Auto merged
This commit is contained in:
unknown 2007-12-10 10:26:01 +01:00
commit c853fa4fe2
27 changed files with 666 additions and 299 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

32
mysql-test/include/have_multi_ndb.inc
View file

@ -5,10 +5,6 @@ connect (server2,127.0.0.1,root,,test,$MASTER_MYPORT1,);
# Check that server1 has NDB support
connection server1;
disable_query_log;
--disable_warnings
drop table if exists t1, t2;
--enable_warnings
flush tables;
--require r/true.require
select (support = 'YES' or support = 'DEFAULT') as `TRUE` from information_schema.engines where engine = 'ndbcluster';
--source include/ndb_not_readonly.inc
@ -17,14 +13,32 @@ enable_query_log;
# Check that server2 has NDB support
connection server2;
disable_query_log;
--disable_warnings
drop table if exists t1, t2;
--enable_warnings
flush tables;
--require r/true.require
select (support = 'YES' or support = 'DEFAULT') as `TRUE` from information_schema.engines where engine = 'ndbcluster';
--source include/ndb_not_readonly.inc
enable_query_log;
# Set the default connection to 'server1'
# cleanup
connection server1;
disable_query_log;
disable_warnings;
--error 0,1051
drop table if exists t1,t2,t3,t4,t5,t6,t7,t8,t9,t10;
flush tables;
flush status;
enable_warnings;
enable_query_log;
connection server2;
disable_query_log;
disable_warnings;
--error 0,1051
drop table if exists t1,t2,t3,t4,t5,t6,t7,t8,t9,t10;
flush tables;
flush status;
enable_warnings;
enable_query_log;
# Set the default connection
connection server1;

24
mysql-test/suite/ndb/r/ndb_basic.result
View file

@ -869,6 +869,30 @@ a b
3 30
4 1
drop table t1,t2;
create table t1 (a varchar(100) primary key, b varchar(100)) engine = NDB;
insert into t1 values
('a', 'a'),('b','b'),('c', 'c'),('aa', 'aa'),('bb', 'bb'),('cc', 'cc');
replace into t1 values ('a', '-a');
replace into t1 values ('b', '-b');
replace into t1 values ('c', '-c');
replace into t1 values ('aa', '-aa');
replace into t1 values ('bb', '-bb');
replace into t1 values ('cc', '-cc');
replace into t1 values ('aaa', '-aaa');
replace into t1 values ('bbb', '-bbb');
replace into t1 values ('ccc', '-ccc');
select * from t1 order by 1,2;
a b
a -a
aa -aa
aaa -aaa
b -b
bb -bb
bbb -bbb
c -c
cc -cc
ccc -ccc
drop table t1;
End of 5.0 tests
CREATE TABLE t1 (a VARCHAR(255) NOT NULL,
CONSTRAINT pk_a PRIMARY KEY (a))engine=ndb;

1
mysql-test/suite/ndb/r/ndb_multi.result
View file

@ -1,4 +1,5 @@
drop table if exists t1, t2, t3, t4;
flush status;
drop table if exists t1, t2, t3, t4;
flush status;
create table t1 (a int) engine=ndbcluster;

1
mysql-test/suite/ndb/r/ndb_multi_row.result
View file

@ -1,4 +1,5 @@
drop table if exists t1, t2, t3, t4;
flush status;
drop table if exists t1, t2, t3, t4;
flush status;
create table t1 (a int) engine=ndbcluster;

22
mysql-test/suite/ndb/t/ndb_basic.test
View file

@ -800,9 +800,27 @@ update ignore t1,t2 set a = 1, c = 1 where a = 3 and c = 3;
select * from t1 order by a;
drop table t1,t2;
# End of 5.0 tests
--echo End of 5.0 tests
#
# Bug#31635
#
create table t1 (a varchar(100) primary key, b varchar(100)) engine = NDB;
insert into t1 values
('a', 'a'),('b','b'),('c', 'c'),('aa', 'aa'),('bb', 'bb'),('cc', 'cc');
replace into t1 values ('a', '-a');
replace into t1 values ('b', '-b');
replace into t1 values ('c', '-c');
replace into t1 values ('aa', '-aa');
replace into t1 values ('bb', '-bb');
replace into t1 values ('cc', '-cc');
replace into t1 values ('aaa', '-aaa');
replace into t1 values ('bbb', '-bbb');
replace into t1 values ('ccc', '-ccc');
select * from t1 order by 1,2;
drop table t1;
--echo End of 5.0 tests
#
# Bug #18483 Cannot create table with FK constraint

4
mysql-test/suite/ndb/t/ndb_multi.test
View file

@ -4,11 +4,11 @@
--disable_warnings
connection server2;
drop table if exists t1, t2, t3, t4;
flush status;
connection server1;
drop table if exists t1, t2, t3, t4;
--enable_warnings
flush status;
--enable_warnings
# Create test tables on server1
create table t1 (a int) engine=ndbcluster;

3
mysql-test/suite/ndb/t/ndb_multi_row.test
View file

@ -6,11 +6,12 @@
--disable_warnings
connection server2;
drop table if exists t1, t2, t3, t4;
flush status;
connection server1;
drop table if exists t1, t2, t3, t4;
flush status;
--enable_warnings
flush status;
# Create test tables on server1
create table t1 (a int) engine=ndbcluster;

2
sql/ha_ndbcluster.cc
View file

@ -618,7 +618,7 @@ bool ha_ndbcluster::get_error_message(int error,
DBUG_ENTER("ha_ndbcluster::get_error_message");
DBUG_PRINT("enter", ("error: %d", error));
Ndb *ndb= get_ndb();
Ndb *ndb= check_ndb_in_thd(current_thd);
if (!ndb)
DBUG_RETURN(FALSE);

84
sql/ha_ndbcluster_binlog.cc
View file

@ -241,18 +241,22 @@ static void dbug_print_table(const char *info, TABLE *table)
static void run_query(THD *thd, char *buf, char *end,
const int *no_print_error, my_bool disable_binlog)
{
ulong save_query_length= thd->query_length;
char *save_query= thd->query;
ulong save_thread_id= thd->variables.pseudo_thread_id;
ulong save_thd_query_length= thd->query_length;
char *save_thd_query= thd->query;
struct system_variables save_thd_variables= thd->variables;
struct system_status_var save_thd_status_var= thd->status_var;
THD_TRANS save_thd_transaction_all= thd->transaction.all;
THD_TRANS save_thd_transaction_stmt= thd->transaction.stmt;
ulonglong save_thd_options= thd->options;
DBUG_ASSERT(sizeof(save_thd_options) == sizeof(thd->options));
NET save_net= thd->net;
NET save_thd_net= thd->net;
const char* found_semicolon= NULL;
bzero((char*) &thd->net, sizeof(NET));
thd->query_length= end - buf;
thd->query= buf;
thd->variables.pseudo_thread_id= thread_id;
thd->transaction.stmt.modified_non_trans_table= FALSE;
if (disable_binlog)
thd->options&= ~OPTION_BIN_LOG;
@ -275,10 +279,13 @@ static void run_query(THD *thd, char *buf, char *end,
}
thd->options= save_thd_options;
thd->query_length= save_query_length;
thd->query= save_query;
thd->variables.pseudo_thread_id= save_thread_id;
thd->net= save_net;
thd->query_length= save_thd_query_length;
thd->query= save_thd_query;
thd->variables= save_thd_variables;
thd->status_var= save_thd_status_var;
thd->transaction.all= save_thd_transaction_all;
thd->transaction.stmt= save_thd_transaction_stmt;
thd->net= save_thd_net;
if (thd == injector_thd)
{
@ -777,8 +784,9 @@ static int ndbcluster_create_ndb_apply_status_table(THD *thd)
" end_pos BIGINT UNSIGNED NOT NULL, "
" PRIMARY KEY USING HASH (server_id) ) ENGINE=NDB");
const int no_print_error[4]= {ER_TABLE_EXISTS_ERROR,
const int no_print_error[5]= {ER_TABLE_EXISTS_ERROR,
701,
702,
4009,
0}; // do not print error 701 etc
run_query(thd, buf, end, no_print_error, TRUE);
@ -837,8 +845,9 @@ static int ndbcluster_create_schema_table(THD *thd)
" type INT UNSIGNED NOT NULL,"
" PRIMARY KEY USING HASH (db,name) ) ENGINE=NDB");
const int no_print_error[4]= {ER_TABLE_EXISTS_ERROR,
const int no_print_error[5]= {ER_TABLE_EXISTS_ERROR,
701,
702,
4009,
0}; // do not print error 701 etc
run_query(thd, buf, end, no_print_error, TRUE);
@ -3587,6 +3596,7 @@ pthread_handler_t ndb_binlog_thread_func(void *arg)
Thd_ndb *thd_ndb=0;
int ndb_update_ndb_binlog_index= 1;
injector *inj= injector::instance();
uint incident_id= 0;
#ifdef RUN_NDB_BINLOG_TIMER
Timer main_timer;
@ -3693,18 +3703,64 @@ pthread_handler_t ndb_binlog_thread_func(void *arg)
pthread_mutex_unlock(&injector_mutex);
pthread_cond_signal(&injector_cond);
/*
wait for mysql server to start (so that the binlog is started
and thus can receive the first GAP event)
*/
pthread_mutex_lock(&LOCK_server_started);
while (!mysqld_server_started)
{
struct timespec abstime;
set_timespec(abstime, 1);
pthread_cond_timedwait(&COND_server_started, &LOCK_server_started,
&abstime);
if (ndbcluster_terminating)
{
pthread_mutex_unlock(&LOCK_server_started);
pthread_mutex_lock(&LOCK_ndb_util_thread);
goto err;
}
}
pthread_mutex_unlock(&LOCK_server_started);
restart:
/*
Main NDB Injector loop
*/
while (ndb_binlog_running)
{
/*
Always insert a GAP event as we cannot know what has happened in the cluster
while not being connected.
check if it is the first log, if so we do not insert a GAP event
as there is really no log to have a GAP in
*/
LEX_STRING const msg= { C_STRING_WITH_LEN("Cluster connect") };
inj->record_incident(thd, INCIDENT_LOST_EVENTS, msg);
if (incident_id == 0)
{
LOG_INFO log_info;
mysql_bin_log.get_current_log(&log_info);
int len= strlen(log_info.log_file_name);
uint no= 0;
if ((sscanf(log_info.log_file_name + len - 6, "%u", &no) == 1) &&
no == 1)
{
/* this is the fist log, so skip GAP event */
break;
}
}
/*
Always insert a GAP event as we cannot know what has happened
in the cluster while not being connected.
*/
LEX_STRING const msg[2]=
{
{ C_STRING_WITH_LEN("mysqld startup") },
{ C_STRING_WITH_LEN("cluster disconnect")}
};
IF_DBUG(int error=)
inj->record_incident(thd, INCIDENT_LOST_EVENTS, msg[incident_id]);
DBUG_ASSERT(!error);
break;
}
incident_id= 1;
{
thd->proc_info= "Waiting for ndbcluster to start";

4
storage/ndb/src/common/transporter/Transporter.cpp
View file

@ -70,7 +70,7 @@ Transporter::Transporter(TransporterRegistry &t_reg,
signalIdUsed = _signalId;
m_connected = false;
m_timeOutMillis = 1000;
m_timeOutMillis = 30000;
m_connect_address.s_addr= 0;
if(s_port<0)
@ -101,7 +101,7 @@ Transporter::connect_server(NDB_SOCKET_TYPE sockfd) {
if(m_connected)
{
DBUG_RETURN(true); // TODO assert(0);
DBUG_RETURN(false); // TODO assert(0);
}
{

3
storage/ndb/src/common/transporter/TransporterRegistry.cpp
View file

@ -758,7 +758,8 @@ TransporterRegistry::poll_TCP(Uint32 timeOutMillis)
TCP_Transporter * t = theTCPTransporters[i];
// If the transporter is connected
if (t->isConnected()) {
NodeId nodeId = t->getRemoteNodeId();
if (is_connected(nodeId) && t->isConnected()) {
const NDB_SOCKET_TYPE socket = t->getSocket();
// Find the highest socket value. It will be used by select

7
storage/ndb/src/kernel/blocks/ERROR_codes.txt
View file

@ -5,7 +5,7 @@ Next DBACC 3002
Next DBTUP 4029
Next DBLQH 5047
Next DBDICT 6008
Next DBDIH 7193
Next DBDIH 7195
Next DBTC 8054
Next CMVMI 9000
Next BACKUP 10038
@ -81,6 +81,11 @@ Delay GCP_SAVEREQ by 10 secs
7185: Dont reply to COPY_GCI_REQ where reason == GCP
7193: Dont send LCP_FRAG_ORD to self, and crash when sending first
LCP_FRAG_ORD(last)
7194: Force removeNodeFromStored to complete in the middle of MASTER_LCPCONF
ERROR CODES FOR TESTING NODE FAILURE, LOCAL CHECKPOINT HANDLING:
-----------------------------------------------------------------

10
storage/ndb/src/kernel/blocks/dbdih/Dbdih.hpp
View file

@ -1310,7 +1310,17 @@ private:
LcpStatus lcpStatus;
Uint32 lcpStatusUpdatedPlace;
struct Save {
LcpStatus m_status;
Uint32 m_place;
} m_saveState[10];
void setLcpStatus(LcpStatus status, Uint32 line){
for (Uint32 i = 9; i > 0; i--)
m_saveState[i] = m_saveState[i-1];
m_saveState[0].m_status = lcpStatus;
m_saveState[0].m_place = lcpStatusUpdatedPlace;
lcpStatus = status;
lcpStatusUpdatedPlace = line;
}

60
storage/ndb/src/kernel/blocks/dbdih/DbdihMain.cpp
View file

@ -5181,11 +5181,19 @@ void Dbdih::startRemoveFailedNode(Signal* signal, NodeRecordPtr failedNodePtr)
}
jam();
signal->theData[0] = DihContinueB::ZREMOVE_NODE_FROM_TABLE;
signal->theData[1] = failedNodePtr.i;
signal->theData[2] = 0; // Tab id
sendSignal(reference(), GSN_CONTINUEB, signal, 3, JBB);
if (!ERROR_INSERTED(7194))
{
signal->theData[0] = DihContinueB::ZREMOVE_NODE_FROM_TABLE;
signal->theData[1] = failedNodePtr.i;
signal->theData[2] = 0; // Tab id
sendSignal(reference(), GSN_CONTINUEB, signal, 3, JBB);
}
else
{
ndbout_c("7194 Not starting ZREMOVE_NODE_FROM_TABLE");
}
setLocalNodefailHandling(signal, failedNodePtr.i, NF_REMOVE_NODE_FROM_TABLE);
}//Dbdih::startRemoveFailedNode()
@ -6114,12 +6122,22 @@ Dbdih::checkEmptyLcpComplete(Signal *signal){
signal->theData[0] = 7012;
execDUMP_STATE_ORD(signal);
if (ERROR_INSERTED(7194))
{
ndbout_c("7194 starting ZREMOVE_NODE_FROM_TABLE");
signal->theData[0] = DihContinueB::ZREMOVE_NODE_FROM_TABLE;
signal->theData[1] = c_lcpMasterTakeOverState.failedNodeId;
signal->theData[2] = 0; // Tab id
sendSignal(reference(), GSN_CONTINUEB, signal, 3, JBB);
}
c_lcpMasterTakeOverState.set(LMTOS_INITIAL, __LINE__);
MasterLCPReq * const req = (MasterLCPReq *)&signal->theData[0];
req->masterRef = reference();
req->failedNodeId = c_lcpMasterTakeOverState.failedNodeId;
sendLoopMacro(MASTER_LCPREQ, sendMASTER_LCPREQ);
} else {
sendMASTER_LCPCONF(signal);
}
@ -6432,6 +6450,15 @@ void Dbdih::execMASTER_LCPCONF(Signal* signal)
{
const MasterLCPConf * const conf = (MasterLCPConf *)&signal->theData[0];
jamEntry();
if (ERROR_INSERTED(7194))
{
ndbout_c("delaying MASTER_LCPCONF due to error 7194");
sendSignalWithDelay(reference(), GSN_MASTER_LCPCONF, signal,
300, signal->getLength());
return;
}
Uint32 senderNodeId = conf->senderNodeId;
MasterLCPConf::State lcpState = (MasterLCPConf::State)conf->lcpState;
const Uint32 failedNodeId = conf->failedNodeId;
@ -6566,7 +6593,6 @@ void Dbdih::MASTER_LCPhandling(Signal* signal, Uint32 failedNodeId)
#endif
c_lcpState.keepGci = SYSFILE->keepGCI;
c_lcpState.setLcpStatus(LCP_START_LCP_ROUND, __LINE__);
startLcpRoundLoopLab(signal, 0, 0);
break;
}
@ -10538,6 +10564,8 @@ void Dbdih::sendLastLCP_FRAG_ORD(Signal* signal)
if(ERROR_INSERTED(7075)){
continue;
}
CRASH_INSERTION(7193);
BlockReference ref = calcLqhBlockRef(nodePtr.i);
sendSignal(ref, GSN_LCP_FRAG_ORD, signal,LcpFragOrd::SignalLength, JBB);
}
@ -10765,6 +10793,13 @@ Dbdih::checkLcpAllTablesDoneInLqh(){
CRASH_INSERTION2(7017, !isMaster());
c_lcpState.setLcpStatus(LCP_TAB_COMPLETED, __LINE__);
if (ERROR_INSERTED(7194))
{
ndbout_c("CLEARING 7194");
CLEAR_ERROR_INSERT_VALUE;
}
return true;
}
@ -10954,6 +10989,11 @@ Dbdih::sendLCP_FRAG_ORD(Signal* signal,
BlockReference ref = calcLqhBlockRef(replicaPtr.p->procNode);
if (ERROR_INSERTED(7193) && replicaPtr.p->procNode == getOwnNodeId())
{
return;
}
LcpFragOrd * const lcpFragOrd = (LcpFragOrd *)&signal->theData[0];
lcpFragOrd->tableId = info.tableId;
lcpFragOrd->fragmentId = info.fragId;
@ -14500,6 +14540,14 @@ Dbdih::execDUMP_STATE_ORD(Signal* signal)
("immediateLcpStart = %d masterLcpNodeId = %d",
c_lcpState.immediateLcpStart,
refToNode(c_lcpState.m_masterLcpDihRef));
for (Uint32 i = 0; i<10; i++)
{
infoEvent("%u : status: %u place: %u", i,
c_lcpState.m_saveState[i].m_status,
c_lcpState.m_saveState[i].m_place);
}
infoEvent("-- Node %d LCP STATE --", getOwnNodeId());
}

19
storage/ndb/src/kernel/blocks/dblqh/Dblqh.hpp
View file

@ -585,7 +585,6 @@ public:
enum ExecSrStatus {
IDLE = 0,
ACTIVE_REMOVE_AFTER = 1,
ACTIVE = 2
};
/**
@ -869,11 +868,6 @@ public:
* heard of.
*/
Uint8 fragDistributionKey;
/**
* The identity of the next local checkpoint this fragment
* should perform.
*/
Uint8 nextLcp;
/**
* How many local checkpoints does the fragment contain
*/
@ -2097,10 +2091,6 @@ private:
void execEXEC_SRCONF(Signal* signal);
void execREAD_PSEUDO_REQ(Signal* signal);
void build_acc(Signal*, Uint32 fragPtrI);
void execBUILDINDXREF(Signal*signal);
void execBUILDINDXCONF(Signal*signal);
void execDUMP_STATE_ORD(Signal* signal);
void execACC_ABORTCONF(Signal* signal);
void execNODE_FAILREP(Signal* signal);
@ -2780,7 +2770,13 @@ private:
/*THIS VARIABLE KEEPS TRACK OF HOW MANY FRAGMENTS THAT PARTICIPATE IN */
/*EXECUTING THE LOG. IF ZERO WE DON'T NEED TO EXECUTE THE LOG AT ALL. */
/* ------------------------------------------------------------------------- */
UintR cnoFragmentsExecSr;
Uint32 cnoFragmentsExecSr;
/**
* This is no of sent GSN_EXEC_FRAGREQ during this log phase
*/
Uint32 cnoOutstandingExecFragReq;
/* ------------------------------------------------------------------------- */
/*THIS VARIABLE KEEPS TRACK OF WHICH OF THE FIRST TWO RESTART PHASES THAT */
/*HAVE COMPLETED. */
@ -2801,7 +2797,6 @@ private:
DLFifoList<Fragrecord> c_lcp_waiting_fragments; // StartFragReq'ed
DLFifoList<Fragrecord> c_lcp_restoring_fragments; // Restoring as we speek
DLFifoList<Fragrecord> c_lcp_complete_fragments; // Restored
DLFifoList<Fragrecord> c_redo_complete_fragments; // Redo'ed
/* ------------------------------------------------------------------------- */
/*USED DURING SYSTEM RESTART, INDICATES THE OLDEST GCI THAT CAN BE RESTARTED */

4
storage/ndb/src/kernel/blocks/dblqh/DblqhInit.cpp
View file

@ -168,7 +168,6 @@ Dblqh::Dblqh(Block_context& ctx):
c_lcp_waiting_fragments(c_fragment_pool),
c_lcp_restoring_fragments(c_fragment_pool),
c_lcp_complete_fragments(c_fragment_pool),
c_redo_complete_fragments(c_fragment_pool),
m_commitAckMarkerHash(m_commitAckMarkerPool),
c_scanTakeOverHash(c_scanRecordPool)
{
@ -295,9 +294,6 @@ Dblqh::Dblqh(Block_context& ctx):
addRecSignal(GSN_READ_PSEUDO_REQ, &Dblqh::execREAD_PSEUDO_REQ);
addRecSignal(GSN_BUILDINDXREF, &Dblqh::execBUILDINDXREF);
addRecSignal(GSN_BUILDINDXCONF, &Dblqh::execBUILDINDXCONF);
addRecSignal(GSN_DEFINE_BACKUP_REF, &Dblqh::execDEFINE_BACKUP_REF);
addRecSignal(GSN_DEFINE_BACKUP_CONF, &Dblqh::execDEFINE_BACKUP_CONF);

214
storage/ndb/src/kernel/blocks/dblqh/DblqhMain.cpp
View file

@ -356,7 +356,6 @@ void Dblqh::execCONTINUEB(Signal* signal)
break;
case ZSR_PHASE3_START:
jam();
signal->theData[0] = data0;
srPhase3Start(signal);
return;
break;
@ -428,25 +427,25 @@ void Dblqh::execCONTINUEB(Signal* signal)
if (fragptr.i != RNIL)
{
jam();
c_redo_complete_fragments.getPtr(fragptr);
c_lcp_complete_fragments.getPtr(fragptr);
signal->theData[0] = fragptr.p->tabRef;
signal->theData[1] = fragptr.p->fragId;
sendSignal(DBACC_REF, GSN_EXPANDCHECK2, signal, 2, JBB);
Ptr<Fragrecord> save = fragptr;
c_redo_complete_fragments.next(fragptr);
c_lcp_complete_fragments.next(fragptr);
signal->theData[0] = ZENABLE_EXPAND_CHECK;
signal->theData[1] = fragptr.i;
sendSignal(DBLQH_REF, GSN_CONTINUEB, signal, 2, JBB);
c_redo_complete_fragments.remove(save);
c_lcp_complete_fragments.remove(save);
return;
}
else
{
jam();
cstartRecReq = 2;
ndbrequire(c_redo_complete_fragments.isEmpty());
ndbrequire(c_lcp_complete_fragments.isEmpty());
StartRecConf * conf = (StartRecConf*)signal->getDataPtrSend();
conf->startingNodeId = getOwnNodeId();
sendSignal(cmasterDihBlockref, GSN_START_RECCONF, signal,
@ -1121,7 +1120,6 @@ void Dblqh::execLQHFRAGREQ(Signal* signal)
Uint32 minRowsHigh = req->minRowsHigh;
Uint32 tschemaVersion = req->schemaVersion;
Uint32 ttupKeyLength = req->keyLength;
Uint32 nextLcp = req->nextLCP;
Uint32 noOfKeyAttr = req->noOfKeyAttr;
Uint32 noOfCharsets = req->noOfCharsets;
Uint32 checksumIndicator = req->checksumIndicator;
@ -1214,7 +1212,6 @@ void Dblqh::execLQHFRAGREQ(Signal* signal)
fragptr.p->lcpFlag = Fragrecord::LCP_STATE_FALSE;
}//if
fragptr.p->nextLcp = nextLcp;
//----------------------------------------------
// For node restarts it is not necessarily zero
//----------------------------------------------
@ -8939,6 +8936,9 @@ void Dblqh::storedProcConfScanLab(Signal* signal)
case Fragrecord::REMOVING:
jam();
default:
jamLine(fragptr.p->fragStatus);
ndbout_c("fragptr.p->fragStatus: %u",
fragptr.p->fragStatus);
ndbrequire(false);
break;
}//switch
@ -14141,15 +14141,12 @@ void Dblqh::execSTART_FRAGREQ(Signal* signal)
if (lcpNo == (MAX_LCP_STORED - 1)) {
jam();
fragptr.p->lcpId[lcpNo] = lcpId;
fragptr.p->nextLcp = 0;
} else if (lcpNo < (MAX_LCP_STORED - 1)) {
jam();
fragptr.p->lcpId[lcpNo] = lcpId;
fragptr.p->nextLcp = lcpNo + 1;
} else {
ndbrequire(lcpNo == ZNIL);
jam();
fragptr.p->nextLcp = 0;
}//if
fragptr.p->srNoLognodes = noOfLogNodes;
fragptr.p->logFlag = Fragrecord::STATE_FALSE;
@ -14181,19 +14178,9 @@ void Dblqh::execSTART_FRAGREQ(Signal* signal)
*/
c_lcp_complete_fragments.add(fragptr);
if(lcpNo == ZNIL)
{
signal->theData[0] = tabptr.i;
signal->theData[1] = fragId;
sendSignal(DBACC_REF, GSN_EXPANDCHECK2, signal, 2, JBB);
}
if (getNodeState().getNodeRestartInProgress())
{
jam();
fragptr.p->fragStatus = Fragrecord::ACTIVE_CREATION;
}
signal->theData[0] = tabptr.i;
signal->theData[1] = fragId;
sendSignal(DBACC_REF, GSN_EXPANDCHECK2, signal, 2, JBB);
c_tup->disk_restart_lcp_id(tabptr.i, fragId, RNIL);
jamEntry();
return;
@ -14395,65 +14382,9 @@ void Dblqh::execSTART_RECCONF(Signal* signal)
return;
}
c_lcp_complete_fragments.first(fragptr);
build_acc(signal, fragptr.i);
return;
}//Dblqh::execSTART_RECCONF()
void
Dblqh::build_acc(Signal* signal, Uint32 fragPtrI)
{
fragptr.i = fragPtrI;
while(fragptr.i != RNIL)
{
c_lcp_complete_fragments.getPtr(fragptr);
tabptr.i = fragptr.p->tabRef;
ptrCheckGuard(tabptr, ctabrecFileSize, tablerec);
if(true || fragptr.i != tabptr.p->fragrec[0])
{
// Only need to send 1 build per table, TUP will rebuild all
fragptr.i = fragptr.p->nextList;
continue;
}
BuildIndxReq* const req = (BuildIndxReq*)signal->getDataPtrSend();
req->setUserRef(reference());
req->setConnectionPtr(fragptr.i);
req->setRequestType(BuildIndxReq::RT_SYSTEMRESTART);
req->setBuildId(0); // not used
req->setBuildKey(0); // not used
req->setIndexType(RNIL);
req->setIndexId(RNIL);
req->setTableId(tabptr.i);
req->setParallelism(0);
sendSignal(DBTUP_REF, GSN_BUILDINDXREQ, signal,
BuildIndxReq::SignalLength, JBB);
return;
}
startExecSr(signal);
}
void
Dblqh::execBUILDINDXREF(Signal* signal)
{
ndbrequire(false);
}
void
Dblqh::execBUILDINDXCONF(Signal* signal)
{
BuildIndxConf* conf = (BuildIndxConf*)signal->getDataPtrSend();
Uint32 fragPtrI = conf->getConnectionPtr();
fragptr.i = fragPtrI;
c_fragment_pool.getPtr(fragptr);
infoEvent("LQH: primary key index %u rebuild done", fragptr.p->tabRef);
build_acc(signal, fragptr.p->nextList);
}
/* ***************> */
/* START_RECREF > */
/* ***************> */
@ -14472,9 +14403,9 @@ void Dblqh::execSTART_EXEC_SR(Signal* signal)
fragptr.i = signal->theData[0];
Uint32 next = RNIL;
if (fragptr.i == RNIL) {
if (fragptr.i == RNIL)
{
jam();
ndbrequire(cnoOfNodes < MAX_NDB_NODES);
/* ----------------------------------------------------------------------
* NO MORE FRAGMENTS TO START EXECUTING THE LOG ON.
* SEND EXEC_SRREQ TO ALL LQH TO INDICATE THAT THIS NODE WILL
@ -14490,10 +14421,15 @@ void Dblqh::execSTART_EXEC_SR(Signal* signal)
} else {
jam();
c_lcp_complete_fragments.getPtr(fragptr);
if (fragptr.p->srNoLognodes > csrPhasesCompleted) {
next = fragptr.p->nextList;
if (fragptr.p->srNoLognodes > csrPhasesCompleted)
{
jam();
cnoOutstandingExecFragReq++;
Uint32 index = csrPhasesCompleted;
arrGuard(index, 4);
arrGuard(index, MAX_LOG_EXEC);
BlockReference ref = calcLqhBlockRef(fragptr.p->srLqhLognode[index]);
fragptr.p->srStatus = Fragrecord::SS_STARTED;
@ -14512,34 +14448,7 @@ void Dblqh::execSTART_EXEC_SR(Signal* signal)
sendSignal(ref, GSN_EXEC_FRAGREQ, signal,
ExecFragReq::SignalLength, JBB);
next = fragptr.p->nextList;
} else {
jam();
/* --------------------------------------------------------------------
* THIS FRAGMENT IS NOW FINISHED WITH THE SYSTEM RESTART. IT DOES
* NOT NEED TO PARTICIPATE IN ANY MORE PHASES. REMOVE IT FROM THE
* LIST OF COMPLETED FRAGMENTS TO EXECUTE THE LOG ON.
* ALSO SEND START_FRAGCONF TO DIH AND SET THE STATE TO ACTIVE ON THE
* FRAGMENT.
* ------------------------------------------------------------------- */
next = fragptr.p->nextList;
c_lcp_complete_fragments.remove(fragptr);
c_redo_complete_fragments.add(fragptr);
if (!getNodeState().getNodeRestartInProgress())
{
fragptr.p->logFlag = Fragrecord::STATE_TRUE;
fragptr.p->fragStatus = Fragrecord::FSACTIVE;
}
else
{
fragptr.p->fragStatus = Fragrecord::ACTIVE_CREATION;
}
signal->theData[0] = fragptr.p->srUserptr;
signal->theData[1] = cownNodeid;
sendSignal(fragptr.p->srBlockref, GSN_START_FRAGCONF, signal, 2, JBB);
} //if
}
signal->theData[0] = next;
sendSignal(cownref, GSN_START_EXEC_SR, signal, 1, JBB);
}//if
@ -14560,24 +14469,8 @@ void Dblqh::execEXEC_FRAGREQ(Signal* signal)
tabptr.i = execFragReq->tableId;
Uint32 fragId = execFragReq->fragId;
ptrCheckGuard(tabptr, ctabrecFileSize, tablerec);
if (!getFragmentrec(signal, fragId)) {
jam();
if (!insertFragrec(signal, fragId)) {
jam();
sendExecFragRefLab(signal);
return;
}//if
initFragrec(signal, tabptr.i, fragId, ZLOG_NODE);
fragptr.p->execSrStatus = Fragrecord::ACTIVE_REMOVE_AFTER;
} else {
jam();
if (fragptr.p->execSrStatus == Fragrecord::ACTIVE_REMOVE_AFTER) {
jam();
fragptr.p->execSrStatus = Fragrecord::ACTIVE_REMOVE_AFTER;
} else {
jam();
}//if
}//if
ndbrequire(getFragmentrec(signal, fragId));
ndbrequire(fragptr.p->execSrNoReplicas < 4);
fragptr.p->execSrBlockref[fragptr.p->execSrNoReplicas] = execFragReq->userRef;
fragptr.p->execSrUserptr[fragptr.p->execSrNoReplicas] = execFragReq->userPtr;
@ -14610,6 +14503,21 @@ void Dblqh::execEXEC_FRAGCONF(Signal* signal)
fragptr.i = signal->theData[0];
c_fragment_pool.getPtr(fragptr);
fragptr.p->srStatus = Fragrecord::SS_COMPLETED;
ndbrequire(cnoOutstandingExecFragReq);
cnoOutstandingExecFragReq--;
if (fragptr.p->srNoLognodes == csrPhasesCompleted + 1)
{
jam();
fragptr.p->logFlag = Fragrecord::STATE_TRUE;
fragptr.p->fragStatus = Fragrecord::FSACTIVE;
signal->theData[0] = fragptr.p->srUserptr;
signal->theData[1] = cownNodeid;
sendSignal(fragptr.p->srBlockref, GSN_START_FRAGCONF, signal, 2, JBB);
}
return;
}//Dblqh::execEXEC_FRAGCONF()
@ -14633,6 +14541,7 @@ void Dblqh::execEXEC_SRCONF(Signal* signal)
Uint32 nodeId = signal->theData[0];
arrGuard(nodeId, MAX_NDB_NODES);
m_sr_exec_sr_conf.set(nodeId);
if (!m_sr_nodes.equal(m_sr_exec_sr_conf))
{
jam();
@ -14653,16 +14562,8 @@ void Dblqh::execEXEC_SRCONF(Signal* signal)
* NOW CHECK IF ALL FRAGMENTS IN THIS PHASE HAVE COMPLETED. IF SO START THE
* NEXT PHASE.
* ----------------------------------------------------------------------- */
c_lcp_complete_fragments.first(fragptr);
while (fragptr.i != RNIL)
{
jam();
if(fragptr.p->srStatus != Fragrecord::SS_COMPLETED)
{
return;
}
c_lcp_complete_fragments.next(fragptr);
}
ndbrequire(cnoOutstandingExecFragReq == 0);
execSrCompletedLab(signal);
return;
}//Dblqh::execEXEC_SRCONF()
@ -14718,6 +14619,7 @@ void Dblqh::execSrCompletedLab(Signal* signal)
* THERE ARE YET MORE PHASES TO RESTART.
* WE MUST INITIALISE DATA FOR NEXT PHASE AND SEND START SIGNAL.
* --------------------------------------------------------------------- */
csrPhaseStarted = ZSR_PHASE1_COMPLETED; // Set correct state first...
startExecSr(signal);
}//if
return;
@ -14791,7 +14693,8 @@ void Dblqh::srPhase3Start(Signal* signal)
UintR tsrPhaseStarted;
jamEntry();
tsrPhaseStarted = signal->theData[0];
tsrPhaseStarted = signal->theData[1];
if (csrPhaseStarted == ZSR_NO_PHASE_STARTED) {
jam();
csrPhaseStarted = tsrPhaseStarted;
@ -15968,18 +15871,6 @@ void Dblqh::sendExecConf(Signal* signal)
sendSignal(fragptr.p->execSrBlockref[i], GSN_EXEC_FRAGCONF,
signal, 1, JBB);
}//for
if (fragptr.p->execSrStatus == Fragrecord::ACTIVE) {
jam();
fragptr.p->execSrStatus = Fragrecord::IDLE;
} else {
ndbrequire(fragptr.p->execSrStatus == Fragrecord::ACTIVE_REMOVE_AFTER);
jam();
Uint32 fragId = fragptr.p->fragId;
tabptr.i = fragptr.p->tabRef;
ptrCheckGuard(tabptr, ctabrecFileSize, tablerec);
c_lcp_complete_fragments.remove(fragptr);
deleteFragrec(fragId);
}//if
fragptr.p->execSrNoReplicas = 0;
}//if
loopCount++;
@ -16007,17 +15898,10 @@ void Dblqh::sendExecConf(Signal* signal)
void Dblqh::srPhase3Comp(Signal* signal)
{
jamEntry();
ndbrequire(cnoOfNodes < MAX_NDB_NODES);
for (Uint32 i = 0; i < cnoOfNodes; i++) {
jam();
if (cnodeStatus[i] == ZNODE_UP) {
jam();
ndbrequire(cnodeData[i] < MAX_NDB_NODES);
BlockReference ref = calcLqhBlockRef(cnodeData[i]);
signal->theData[0] = cownNodeid;
sendSignal(ref, GSN_EXEC_SRCONF, signal, 1, JBB);
}//if
}//for
signal->theData[0] = cownNodeid;
NodeReceiverGroup rg(DBLQH, m_sr_nodes);
sendSignal(rg, GSN_EXEC_SRCONF, signal, 1, JBB);
return;
}//Dblqh::srPhase3Comp()
@ -16259,7 +16143,7 @@ void Dblqh::srFourthComp(Signal* signal)
if(cstartType == NodeState::ST_SYSTEM_RESTART)
{
jam();
if (c_redo_complete_fragments.first(fragptr))
if (c_lcp_complete_fragments.first(fragptr))
{
jam();
signal->theData[0] = ZENABLE_EXPAND_CHECK;
@ -17367,7 +17251,6 @@ void Dblqh::initFragrec(Signal* signal,
fragptr.p->maxGciInLcp = 0;
fragptr.p->copyFragState = ZIDLE;
fragptr.p->newestGci = cnewestGci;
fragptr.p->nextLcp = 0;
fragptr.p->tabRef = tableId;
fragptr.p->fragId = fragId;
fragptr.p->srStatus = Fragrecord::SS_IDLE;
@ -18456,6 +18339,7 @@ void Dblqh::sendLqhTransconf(Signal* signal, LqhTransConf::OperationStatus stat)
void Dblqh::startExecSr(Signal* signal)
{
cnoFragmentsExecSr = 0;
cnoOutstandingExecFragReq = 0;
c_lcp_complete_fragments.first(fragptr);
signal->theData[0] = fragptr.i;
sendSignal(cownref, GSN_START_EXEC_SR, signal, 1, JBB);

48
storage/ndb/src/kernel/blocks/suma/Suma.cpp
View file

@ -974,6 +974,54 @@ Suma::execDUMP_STATE_ORD(Signal* signal){
}
return;
}
if (tCase == 8011)
{
jam();
Uint32 bucket = signal->theData[1];
KeyTable<Table>::Iterator it;
if (signal->getLength() == 1)
{
jam();
bucket = 0;
infoEvent("-- Starting dump of subscribers --");
}
c_tables.next(bucket, it);
const Uint32 RT_BREAK = 16;
for(Uint32 i = 0; i<RT_BREAK || it.bucket == bucket; i++)
{
jam();
if(it.curr.i == RNIL)
{
jam();
infoEvent("-- Ending dump of subscribers --");
return;
}
infoEvent("Table: %u ver: %u #n: %u (ref,data,subscritopn)",
it.curr.p->m_tableId,
it.curr.p->m_schemaVersion,
it.curr.p->n_subscribers);
Ptr<Subscriber> ptr;
LocalDLList<Subscriber> list(c_subscriberPool, it.curr.p->c_subscribers);
for (list.first(ptr); !ptr.isNull(); list.next(ptr), i++)
{
jam();
infoEvent(" [ %x %u %u ]",
ptr.p->m_senderRef,
ptr.p->m_senderData,
ptr.p->m_subPtrI);
}
c_tables.next(it);
}
signal->theData[0] = tCase;
signal->theData[1] = it.bucket;
sendSignalWithDelay(reference(), GSN_DUMP_STATE_ORD, signal, 100, 2);
return;
}
}
/*************************************************************

2
storage/ndb/src/ndbapi/TransporterFacade.cpp
View file

@ -1501,9 +1501,9 @@ void PollGuard::unlock_and_signal()
if (t_signal_cond_waiter)
t_signal_cond_waiter->set_poll_owner(true);
}
m_tp->unlock_mutex();
if (t_signal_cond_waiter)
t_signal_cond_waiter->cond_signal();
m_tp->unlock_mutex();
m_locked=false;
}

10
storage/ndb/test/ndbapi/testDict.cpp
View file

@ -2776,9 +2776,13 @@ runDropDDObjects(NDBT_Context* ctx, NDBT_Step* step){
case NdbDictionary::Object::UserTable:
tableFound = list.elements[i].name;
if(tableFound != 0){
if(pDict->dropTable(tableFound) != 0){
g_err << "Failed to drop table: " << pDict->getNdbError() << endl;
return NDBT_FAILED;
if(strcmp(tableFound, "ndb_apply_status") != 0 &&
strcmp(tableFound, "NDB$BLOB_2_3") != 0 &&
strcmp(tableFound, "ndb_schema") != 0){
if(pDict->dropTable(tableFound) != 0){
g_err << "Failed to drop table: " << tableFound << pDict->getNdbError() << endl;
return NDBT_FAILED;
}
}
}
tableFound = 0;

48
storage/ndb/test/ndbapi/testNodeRestart.cpp
View file

@ -1830,6 +1830,51 @@ runBug31525(NDBT_Context* ctx, NDBT_Step* step)
if (res.restartOneDbNode(nodes[1], false, false, true))
return NDBT_FAILED;
if (res.waitClusterStarted())
return NDBT_FAILED;
return NDBT_OK;
}
int
runBug32160(NDBT_Context* ctx, NDBT_Step* step)
{
int result = NDBT_OK;
int loops = ctx->getNumLoops();
int records = ctx->getNumRecords();
Ndb* pNdb = GETNDB(step);
NdbRestarter res;
if (res.getNumDbNodes() < 2)
{
return NDBT_OK;
}
int master = res.getMasterNodeId();
int next = res.getNextMasterNodeId(master);
if (res.insertErrorInNode(next, 7194))
{
return NDBT_FAILED;
}
int val2[] = { DumpStateOrd::CmvmiSetRestartOnErrorInsert, 1 };
if (res.dumpStateOneNode(master, val2, 2))
return NDBT_FAILED;
if (res.insertErrorInNode(master, 7193))
return NDBT_FAILED;
int val3[] = { 7099 };
if (res.dumpStateOneNode(master, val3, 1))
return NDBT_FAILED;
if (res.waitNodesNoStart(&master, 1))
return NDBT_FAILED;
if (res.startNodes(&master, 1))
return NDBT_FAILED;
if (res.waitClusterStarted())
return NDBT_FAILED;
@ -2205,6 +2250,9 @@ TESTCASE("Bug28717", ""){
TESTCASE("Bug29364", ""){
INITIALIZER(runBug29364);
}
TESTCASE("Bug32160", ""){
INITIALIZER(runBug32160);
}
NDBT_TESTSUITE_END(testNodeRestart);
int main(int argc, const char** argv){

41
storage/ndb/test/ndbapi/testSystemRestart.cpp
View file

@ -1501,6 +1501,38 @@ int runSR_DD_2(NDBT_Context* ctx, NDBT_Step* step)
return result;
}
int runBug22696(NDBT_Context* ctx, NDBT_Step* step)
{
Ndb* pNdb = GETNDB(step);
int result = NDBT_OK;
Uint32 loops = ctx->getNumLoops();
Uint32 rows = ctx->getNumRecords();
NdbRestarter restarter;
HugoTransactions hugoTrans(*ctx->getTab());
Uint32 i = 0;
while(i<=loops && result != NDBT_FAILED)
{
for (Uint32 j = 0; j<10 && result != NDBT_FAILED; j++)
CHECK(hugoTrans.scanUpdateRecords(pNdb, rows) == 0);
CHECK(restarter.restartAll(false, true, i > 0 ? true : false) == 0);
CHECK(restarter.waitClusterNoStart() == 0);
CHECK(restarter.insertErrorInAllNodes(7072) == 0);
CHECK(restarter.startAll() == 0);
CHECK(restarter.waitClusterStarted() == 0);
i++;
if (i < loops)
{
NdbSleep_SecSleep(5); // Wait for a few gcp
}
}
ctx->stopTest();
return result;
}
int
runBug27434(NDBT_Context* ctx, NDBT_Step* step)
{
@ -1813,8 +1845,13 @@ TESTCASE("Bug28770",
STEP(runBug28770);
FINALIZER(runClearTable);
}
TESTCASE("Bug22696", "")
{
INITIALIZER(runWaitStarted);
INITIALIZER(runLoadTable);
INITIALIZER(runBug22696);
FINALIZER(runClearTable);
}
NDBT_TESTSUITE_END(testSystemRestart);
int main(int argc, const char** argv){

7
storage/ndb/test/run-test/daily-basic-tests.txt
View file

@ -581,6 +581,10 @@ max-time: 1000
cmd: testNodeRestart
args: -n Bug29364 T1
max-time: 300
cmd: testNodeRestart
args: -n Bug32160 T1
#
# DICT TESTS
max-time: 500
@ -1038,4 +1042,7 @@ max-time: 300
cmd: test_event
args: -n Bug31701 T1
max-time: 300
cmd: testSystemRestart
args: -n Bug22696 T1

3
storage/ndb/test/tools/Makefile.am
View file

@ -13,7 +13,7 @@
# along with this program; if not, write to the Free Software
# Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA
ndbtest_PROGRAMS = hugoLoad hugoFill hugoLockRecords hugoPkDelete hugoPkRead hugoPkReadRecord hugoPkUpdate hugoScanRead hugoScanUpdate restart verify_index copy_tab create_index ndb_cpcc listen_event eventlog rep_latency
ndbtest_PROGRAMS = hugoLoad hugoFill hugoLockRecords hugoPkDelete hugoPkRead hugoPkReadRecord hugoPkUpdate hugoScanRead hugoScanUpdate restart verify_index copy_tab create_index ndb_cpcc listen_event eventlog rep_latency ndb_connect
# transproxy
@ -35,6 +35,7 @@ ndb_cpcc_SOURCES = cpcc.cpp
listen_event_SOURCES = listen.cpp
eventlog_SOURCES = log_listner.cpp
rep_latency_SOURCES = rep_latency.cpp
ndb_connect_SOURCES = connect.cpp
include $(top_srcdir)/storage/ndb/config/common.mk.am
include $(top_srcdir)/storage/ndb/config/type_ndbapitest.mk.am

152
storage/ndb/test/tools/connect.cpp Normal file
View file

@ -0,0 +1,152 @@
/* Copyright (C) 2003 MySQL AB
This program is free software; you can redistribute it and/or modify
it under the terms of the GNU General Public License as published by
the Free Software Foundation; either version 2 of the License, or
(at your option) any later version.
This program is distributed in the hope that it will be useful,
but WITHOUT ANY WARRANTY; without even the implied warranty of
MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
GNU General Public License for more details.
You should have received a copy of the GNU General Public License
along with this program; if not, write to the Free Software
Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA */
#include <ndb_global.h>
#include <ndb_opts.h>
#include <NDBT.hpp>
#include <NdbApi.hpp>
#include <NdbSleep.h>
NDB_STD_OPTS_VARS;
static int _loop = 25;
static int _sleep = 25;
static int _drop = 1;
typedef uchar* gptr;
static struct my_option my_long_options[] =
{
NDB_STD_OPTS("ndb_desc"),
{ "loop", 'l', "loops",
(gptr*) &_loop, (gptr*) &_loop, 0,
GET_INT, REQUIRED_ARG, _loop, 0, 0, 0, 0, 0 },
{ "sleep", 's', "Sleep (ms) between connection attempt",
(gptr*) &_sleep, (gptr*) &_sleep, 0,
GET_INT, REQUIRED_ARG, _sleep, 0, 0, 0, 0, 0 },
{ "drop", 'd',
"Drop event operations before disconnect (0 = no, 1 = yes, else rand",
(gptr*) &_drop, (gptr*) &_drop, 0,
GET_INT, REQUIRED_ARG, _drop, 0, 0, 0, 0, 0 },
{ 0, 0, 0, 0, 0, 0, GET_NO_ARG, NO_ARG, 0, 0, 0, 0, 0, 0}
};
static void usage()
{
char desc[] = "This program connects to ndbd, and then disconnects\n";
ndb_std_print_version();
my_print_help(my_long_options);
my_print_variables(my_long_options);
}
int main(int argc, char** argv){
NDB_INIT(argv[0]);
const char *load_default_groups[]= { "mysql_cluster",0 };
load_defaults("my",load_default_groups,&argc,&argv);
int ho_error;
#ifndef DBUG_OFF
opt_debug= "d:t:O,/tmp/ndb_desc.trace";
#endif
if ((ho_error=handle_options(&argc, &argv, my_long_options,
ndb_std_get_one_option)))
return NDBT_ProgramExit(NDBT_WRONGARGS);
for (int i = 0; i<_loop; i++)
{
Ndb_cluster_connection con(opt_connect_str);
if(con.connect(12, 5, 1) != 0)
{
ndbout << "Unable to connect to management server." << endl;
return NDBT_ProgramExit(NDBT_FAILED);
}
if (con.wait_until_ready(30,30) != 0)
{
ndbout << "Cluster nodes not ready in 30 seconds." << endl;
return NDBT_ProgramExit(NDBT_FAILED);
}
Ndb MyNdb(&con, "TEST_DB");
if(MyNdb.init() != 0){
ERR(MyNdb.getNdbError());
return NDBT_ProgramExit(NDBT_FAILED);
}
Vector<NdbEventOperation*> ops;
const NdbDictionary::Dictionary * dict= MyNdb.getDictionary();
for (int j = 0; j < argc; j++)
{
const NdbDictionary::Table * pTab = dict->getTable(argv[j]);
if (pTab == 0)
{
ndbout_c("Failed to retreive table: \"%s\"", argv[j]);
}
BaseString tmp;
tmp.appfmt("EV-%s", argv[j]);
NdbEventOperation* pOp = MyNdb.createEventOperation(tmp.c_str());
if ( pOp == NULL )
{
ndbout << "Event operation creation failed: " <<
MyNdb.getNdbError() << endl;
return NDBT_ProgramExit(NDBT_FAILED);
}
for (int a = 0; a < pTab->getNoOfColumns(); a++)
{
pOp->getValue(pTab->getColumn(a)->getName());
pOp->getPreValue(pTab->getColumn(a)->getName());
}
if (pOp->execute())
{
ndbout << "operation execution failed: " << pOp->getNdbError()
<< endl;
return NDBT_ProgramExit(NDBT_FAILED);
}
ops.push_back(pOp);
}
if (_sleep)
{
NdbSleep_MilliSleep(10 + rand() % _sleep);
}
for (Uint32 i = 0; i<ops.size(); i++)
{
switch(_drop){
case 0:
break;
do_drop:
case 1:
if (MyNdb.dropEventOperation(ops[i]))
{
ndbout << "drop event operation failed "
<< MyNdb.getNdbError() << endl;
return NDBT_ProgramExit(NDBT_FAILED);
}
break;
default:
if ((rand() % 100) > 50)
goto do_drop;
}
}
}
return NDBT_ProgramExit(NDBT_OK);
}
template class Vector<NdbEventOperation*>;

156
storage/ndb/tools/restore/Restore.cpp
View file

@ -534,6 +534,88 @@ TupleS::prepareRecord(TableS & tab){
return true;
}
int
RestoreDataIterator::readTupleData(Uint32 *buf_ptr, Uint32 *ptr,
Uint32 dataLength)
{
while (ptr + 2 < buf_ptr + dataLength)
{
typedef BackupFormat::DataFile::VariableData VarData;
VarData * data = (VarData *)ptr;
Uint32 sz = ntohl(data->Sz);
Uint32 attrId = ntohl(data->Id); // column_no
AttributeData * attr_data = m_tuple.getData(attrId);
const AttributeDesc * attr_desc = m_tuple.getDesc(attrId);
// just a reminder - remove when backwards compat implemented
if (m_currentTable->backupVersion < MAKE_VERSION(5,1,3) &&
attr_desc->m_column->getNullable())
{
const Uint32 ind = attr_desc->m_nullBitIndex;
if(BitmaskImpl::get(m_currentTable->m_nullBitmaskSize,
buf_ptr,ind))
{
attr_data->null = true;
attr_data->void_value = NULL;
continue;
}
}
if (m_currentTable->backupVersion < MAKE_VERSION(5,1,3))
{
sz *= 4;
}
attr_data->null = false;
attr_data->void_value = &data->Data[0];
attr_data->size = sz;
//if (m_currentTable->getTableId() >= 2) { ndbout << "var off=" << ptr-buf_ptr << " attrId=" << attrId << endl; }
/**
* Compute array size
*/
const Uint32 arraySize = sz / (attr_desc->size / 8);
assert(arraySize <= attr_desc->arraySize);
//convert the length of blob(v1) and text(v1)
if(!m_hostByteOrder
&& (attr_desc->m_column->getType() == NdbDictionary::Column::Blob
|| attr_desc->m_column->getType() == NdbDictionary::Column::Text)
&& attr_desc->m_column->getArrayType() == NdbDictionary::Column::ArrayTypeFixed)
{
char* p = (char*)&attr_data->u_int64_value[0];
Uint64 x;
memcpy(&x, p, sizeof(Uint64));
x = Twiddle64(x);
memcpy(p, &x, sizeof(Uint64));
}
//convert datetime type
if(!m_hostByteOrder
&& attr_desc->m_column->getType() == NdbDictionary::Column::Datetime)
{
char* p = (char*)&attr_data->u_int64_value[0];
Uint64 x;
memcpy(&x, p, sizeof(Uint64));
x = Twiddle64(x);
memcpy(p, &x, sizeof(Uint64));
}
if(!Twiddle(attr_desc, attr_data, attr_desc->arraySize))
{
return -1;
}
ptr += ((sz + 3) >> 2) + 2;
}
assert(ptr == buf_ptr + dataLength);
return 0;
}
const TupleS *
RestoreDataIterator::getNextTuple(int & res)
{
@ -630,78 +712,8 @@ RestoreDataIterator::getNextTuple(int & res)
attr_data->void_value = NULL;
}
while (ptr + 2 < buf_ptr + dataLength) {
typedef BackupFormat::DataFile::VariableData VarData;
VarData * data = (VarData *)ptr;
Uint32 sz = ntohl(data->Sz);
Uint32 attrId = ntohl(data->Id); // column_no
AttributeData * attr_data = m_tuple.getData(attrId);
const AttributeDesc * attr_desc = m_tuple.getDesc(attrId);
// just a reminder - remove when backwards compat implemented
if(m_currentTable->backupVersion < MAKE_VERSION(5,1,3) &&
attr_desc->m_column->getNullable()){
const Uint32 ind = attr_desc->m_nullBitIndex;
if(BitmaskImpl::get(m_currentTable->m_nullBitmaskSize,
buf_ptr,ind)){
attr_data->null = true;
attr_data->void_value = NULL;
continue;
}
}
if (m_currentTable->backupVersion < MAKE_VERSION(5,1,3))
{
sz *= 4;
}
attr_data->null = false;
attr_data->void_value = &data->Data[0];
attr_data->size = sz;
//if (m_currentTable->getTableId() >= 2) { ndbout << "var off=" << ptr-buf_ptr << " attrId=" << attrId << endl; }
/**
* Compute array size
*/
const Uint32 arraySize = sz / (attr_desc->size / 8);
assert(arraySize <= attr_desc->arraySize);
//convert the length of blob(v1) and text(v1)
if(!m_hostByteOrder
&& (attr_desc->m_column->getType() == NdbDictionary::Column::Blob
|| attr_desc->m_column->getType() == NdbDictionary::Column::Text)
&& attr_desc->m_column->getArrayType() == NdbDictionary::Column::ArrayTypeFixed)
{
char* p = (char*)&attr_data->u_int64_value[0];
Uint64 x;
memcpy(&x, p, sizeof(Uint64));
x = Twiddle64(x);
memcpy(p, &x, sizeof(Uint64));
}
//convert datetime type
if(!m_hostByteOrder
&& attr_desc->m_column->getType() == NdbDictionary::Column::Datetime)
{
char* p = (char*)&attr_data->u_int64_value[0];
Uint64 x;
memcpy(&x, p, sizeof(Uint64));
x = Twiddle64(x);
memcpy(p, &x, sizeof(Uint64));
}
if(!Twiddle(attr_desc, attr_data, attr_desc->arraySize))
{
res = -1;
return NULL;
}
ptr += ((sz + 3) >> 2) + 2;
}
assert(ptr == buf_ptr + dataLength);
if ((res = readTupleData(buf_ptr, ptr, dataLength)))
return NULL;
m_count ++;
res = 0;

4
storage/ndb/tools/restore/Restore.hpp
View file

@ -355,6 +355,10 @@ public:
bool validateFragmentFooter();
const TupleS *getNextTuple(int & res);
private:
int readTupleData(Uint32 *buf_ptr, Uint32 *ptr, Uint32 dataLength);
};
class LogEntry {