summaryrefslogtreecommitdiff
diff options
context:
space:
mode:
Diffstat
-rw-r--r--mysql-test/r/ndb_index_unique.result3
-rw-r--r--mysql-test/t/ndb_index_unique.test8
-rw-r--r--sql/ha_ndbcluster.cc5
-rw-r--r--sql/ha_ndbcluster.h2
-rw-r--r--sql/handler.cc1
-rw-r--r--storage/ndb/src/kernel/blocks/dbtux/Dbtux.hpp39
-rw-r--r--storage/ndb/src/kernel/blocks/dbtux/DbtuxDebug.cpp16
-rw-r--r--storage/ndb/src/kernel/blocks/dbtux/DbtuxGen.cpp4
-rw-r--r--storage/ndb/src/kernel/blocks/dbtux/DbtuxScan.cpp115
9 files changed, 121 insertions, 72 deletions
diff --git a/mysql-test/r/ndb_index_unique.result b/mysql-test/r/ndb_index_unique.result
index bdca141df61..563024a8800 100644
--- a/mysql-test/r/ndb_index_unique.result
+++ b/mysql-test/r/ndb_index_unique.result
@@ -639,3 +639,6 @@ select a, sha1(b) from t1;
a sha1(b)
1 08f5d02c8b8bc244f275bdfc22c42c5cab0d9d7d
drop table t1;
+create table t1(id int not null) engine = NDB;
+alter table t1 add constraint uk_test unique (id) using hash;
+drop table t1;
diff --git a/mysql-test/t/ndb_index_unique.test b/mysql-test/t/ndb_index_unique.test
index a40efc8a40c..94bba98662c 100644
--- a/mysql-test/t/ndb_index_unique.test
+++ b/mysql-test/t/ndb_index_unique.test
@@ -329,4 +329,12 @@ select a, sha1(b) from t1;
drop table t1;
+# bug#21873 MySQLD Crash on ALTER...ADD..UNIQUE..USING HASH statement for NDB backed table
+
+create table t1(id int not null) engine = NDB;
+
+alter table t1 add constraint uk_test unique (id) using hash;
+
+drop table t1;
+
# End of 4.1 tests
diff --git a/sql/ha_ndbcluster.cc b/sql/ha_ndbcluster.cc
index f91cbe2a403..4ebebd853ef 100644
--- a/sql/ha_ndbcluster.cc
+++ b/sql/ha_ndbcluster.cc
@@ -1399,9 +1399,8 @@ NDB_INDEX_TYPE ha_ndbcluster::get_index_type_from_key(uint inx,
ORDERED_INDEX);
}
-int ha_ndbcluster::check_index_fields_not_null(uint inx)
+int ha_ndbcluster::check_index_fields_not_null(KEY* key_info)
{
- KEY* key_info= table->key_info + inx;
KEY_PART_INFO* key_part= key_info->key_part;
KEY_PART_INFO* end= key_part+key_info->key_parts;
DBUG_ENTER("ha_ndbcluster::check_index_fields_not_null");
@@ -5017,7 +5016,7 @@ int ha_ndbcluster::create_index(const char *name, KEY *key_info,
error= create_unique_index(unique_name, key_info);
break;
case UNIQUE_INDEX:
- if (!(error= check_index_fields_not_null(idx_no)))
+ if (!(error= check_index_fields_not_null(key_info)))
error= create_unique_index(unique_name, key_info);
break;
case ORDERED_INDEX:
diff --git a/sql/ha_ndbcluster.h b/sql/ha_ndbcluster.h
index cffc7dcdd47..f0903d3a233 100644
--- a/sql/ha_ndbcluster.h
+++ b/sql/ha_ndbcluster.h
@@ -808,7 +808,7 @@ private:
NDB_INDEX_TYPE get_index_type_from_table(uint index_no) const;
NDB_INDEX_TYPE get_index_type_from_key(uint index_no, KEY *key_info,
bool primary) const;
- int check_index_fields_not_null(uint index_no);
+ int check_index_fields_not_null(KEY *key_info);
uint set_up_partition_info(partition_info *part_info,
TABLE *table,
diff --git a/sql/handler.cc b/sql/handler.cc
index 7716d0d0c02..d7ae6f9e098 100644
--- a/sql/handler.cc
+++ b/sql/handler.cc
@@ -2036,6 +2036,7 @@ void handler::print_keydup_error(uint key_nr, const char *msg)
{
/* Key is unknown */
str.length(0);
+ str.append(STRING_WITH_LEN(""));
key_nr= -1;
my_printf_error(ER_DUP_ENTRY, msg,
MYF(0), str.c_ptr(), "*UNKNOWN*");
diff --git a/storage/ndb/src/kernel/blocks/dbtux/Dbtux.hpp b/storage/ndb/src/kernel/blocks/dbtux/Dbtux.hpp
index 8882b1f8093..d943fabd5c9 100644
--- a/storage/ndb/src/kernel/blocks/dbtux/Dbtux.hpp
+++ b/storage/ndb/src/kernel/blocks/dbtux/Dbtux.hpp
@@ -23,6 +23,7 @@
#include <AttributeHeader.hpp>
#include <ArrayPool.hpp>
#include <DataBuffer.hpp>
+#include <DLFifoList.hpp>
#include <md5_hash.hpp>
// big brother
@@ -338,6 +339,18 @@ private:
typedef DataBuffer<ScanBoundSegmentSize>::ConstDataBufferIterator ScanBoundIterator;
typedef DataBuffer<ScanBoundSegmentSize>::DataBufferPool ScanBoundPool;
ScanBoundPool c_scanBoundPool;
+
+ // ScanLock
+ struct ScanLock {
+ Uint32 m_accLockOp;
+ union {
+ Uint32 nextPool;
+ Uint32 nextList;
+ };
+ Uint32 prevList;
+ };
+ typedef Ptr<ScanLock> ScanLockPtr;
+ ArrayPool<ScanLock> c_scanLockPool;
/*
* Scan operation.
@@ -384,6 +397,8 @@ private:
Uint32 m_savePointId;
// lock waited for or obtained and not yet passed to LQH
Uint32 m_accLockOp;
+ // locks obtained and passed to LQH but not yet returned by LQH
+ DLFifoList<ScanLock>::Head m_accLockOps;
Uint8 m_readCommitted; // no locking
Uint8 m_lockMode;
Uint8 m_descending;
@@ -399,13 +414,6 @@ private:
Uint32 nextList;
};
Uint32 prevList;
- /*
- * Locks obtained and passed to LQH but not yet returned by LQH.
- * The max was increased from 16 to 992 (default 64). Record max
- * ever used in this scan. TODO fix quadratic behaviour
- */
- Uint32 m_maxAccLockOps;
- Uint32 m_accLockOps[MaxAccLockOps];
ScanOp(ScanBoundPool& scanBoundPool);
};
typedef Ptr<ScanOp> ScanOpPtr;
@@ -625,8 +633,9 @@ private:
bool scanCheck(ScanOpPtr scanPtr, TreeEnt ent);
bool scanVisible(ScanOpPtr scanPtr, TreeEnt ent);
void scanClose(Signal* signal, ScanOpPtr scanPtr);
- void addAccLockOp(ScanOp& scan, Uint32 accLockOp);
- void removeAccLockOp(ScanOp& scan, Uint32 accLockOp);
+ void abortAccLockOps(Signal* signal, ScanOpPtr scanPtr);
+ void addAccLockOp(ScanOpPtr scanPtr, Uint32 accLockOp);
+ void removeAccLockOp(ScanOpPtr scanPtr, Uint32 accLockOp);
void releaseScanOp(ScanOpPtr& scanPtr);
/*
@@ -687,7 +696,8 @@ private:
DebugMeta = 1, // log create and drop index
DebugMaint = 2, // log maintenance ops
DebugTree = 4, // log and check tree after each op
- DebugScan = 8 // log scans
+ DebugScan = 8, // log scans
+ DebugLock = 16 // log ACC locks
};
STATIC_CONST( DataFillByte = 0xa2 );
STATIC_CONST( NodeFillByte = 0xa4 );
@@ -944,6 +954,7 @@ Dbtux::ScanOp::ScanOp(ScanBoundPool& scanBoundPool) :
m_transId2(0),
m_savePointId(0),
m_accLockOp(RNIL),
+ m_accLockOps(),
m_readCommitted(0),
m_lockMode(0),
m_descending(0),
@@ -951,18 +962,12 @@ Dbtux::ScanOp::ScanOp(ScanBoundPool& scanBoundPool) :
m_boundMax(scanBoundPool),
m_scanPos(),
m_scanEnt(),
- m_nodeScan(RNIL),
- m_maxAccLockOps(0)
+ m_nodeScan(RNIL)
{
m_bound[0] = &m_boundMin;
m_bound[1] = &m_boundMax;
m_boundCnt[0] = 0;
m_boundCnt[1] = 0;
-#ifdef VM_TRACE
- for (unsigned i = 0; i < MaxAccLockOps; i++) {
- m_accLockOps[i] = 0x1f1f1f1f;
- }
-#endif
}
// Dbtux::Index
diff --git a/storage/ndb/src/kernel/blocks/dbtux/DbtuxDebug.cpp b/storage/ndb/src/kernel/blocks/dbtux/DbtuxDebug.cpp
index e1e858622cc..61aae83ca27 100644
--- a/storage/ndb/src/kernel/blocks/dbtux/DbtuxDebug.cpp
+++ b/storage/ndb/src/kernel/blocks/dbtux/DbtuxDebug.cpp
@@ -299,6 +299,7 @@ operator<<(NdbOut& out, const Dbtux::DescAttr& descAttr)
NdbOut&
operator<<(NdbOut& out, const Dbtux::ScanOp& scan)
{
+ Dbtux* tux = (Dbtux*)globalData.getBlock(DBTUX);
out << "[ScanOp " << hex << &scan;
out << " [state " << dec << scan.m_state << "]";
out << " [lockwait " << dec << scan.m_lockwait << "]";
@@ -308,9 +309,15 @@ operator<<(NdbOut& out, const Dbtux::ScanOp& scan)
out << " [savePointId " << dec << scan.m_savePointId << "]";
out << " [accLockOp " << hex << scan.m_accLockOp << "]";
out << " [accLockOps";
- for (unsigned i = 0; i < scan.m_maxAccLockOps; i++) {
- if (scan.m_accLockOps[i] != RNIL)
- out << " " << hex << scan.m_accLockOps[i];
+ {
+ DLFifoList<Dbtux::ScanLock>::Head head = scan.m_accLockOps;
+ LocalDLFifoList<Dbtux::ScanLock> list(tux->c_scanLockPool, head);
+ Dbtux::ScanLockPtr lockPtr;
+ list.first(lockPtr);
+ while (lockPtr.i != RNIL) {
+ out << " " << hex << lockPtr.p->m_accLockOp;
+ list.next(lockPtr);
+ }
}
out << "]";
out << " [readCommitted " << dec << scan.m_readCommitted << "]";
@@ -336,13 +343,12 @@ operator<<(NdbOut& out, const Dbtux::ScanOp& scan)
NdbOut&
operator<<(NdbOut& out, const Dbtux::Index& index)
{
+ Dbtux* tux = (Dbtux*)globalData.getBlock(DBTUX);
out << "[Index " << hex << &index;
out << " [tableId " << dec << index.m_tableId << "]";
out << " [numFrags " << dec << index.m_numFrags << "]";
for (unsigned i = 0; i < index.m_numFrags; i++) {
out << " [frag " << dec << i << " ";
- // dangerous and wrong
- Dbtux* tux = (Dbtux*)globalData.getBlock(DBTUX);
const Dbtux::Frag& frag = *tux->c_fragPool.getPtr(index.m_fragPtrI[i]);
out << frag;
out << "]";
diff --git a/storage/ndb/src/kernel/blocks/dbtux/DbtuxGen.cpp b/storage/ndb/src/kernel/blocks/dbtux/DbtuxGen.cpp
index 8aeb4934600..beed787f45f 100644
--- a/storage/ndb/src/kernel/blocks/dbtux/DbtuxGen.cpp
+++ b/storage/ndb/src/kernel/blocks/dbtux/DbtuxGen.cpp
@@ -163,6 +163,7 @@ Dbtux::execREAD_CONFIG_REQ(Signal* signal)
Uint32 nFragment;
Uint32 nAttribute;
Uint32 nScanOp;
+ Uint32 nScanBatch;
const ndb_mgm_configuration_iterator * p =
m_ctx.m_config.getOwnConfigIterator();
@@ -172,9 +173,11 @@ Dbtux::execREAD_CONFIG_REQ(Signal* signal)
ndbrequire(!ndb_mgm_get_int_parameter(p, CFG_TUX_FRAGMENT, &nFragment));
ndbrequire(!ndb_mgm_get_int_parameter(p, CFG_TUX_ATTRIBUTE, &nAttribute));
ndbrequire(!ndb_mgm_get_int_parameter(p, CFG_TUX_SCAN_OP, &nScanOp));
+ ndbrequire(!ndb_mgm_get_int_parameter(p, CFG_DB_BATCH_SIZE, &nScanBatch));
const Uint32 nDescPage = (nIndex * DescHeadSize + nAttribute * DescAttrSize + DescPageSize - 1) / DescPageSize;
const Uint32 nScanBoundWords = nScanOp * ScanBoundSegmentSize * 4;
+ const Uint32 nScanLock = nScanOp * nScanBatch;
c_indexPool.setSize(nIndex);
c_fragPool.setSize(nFragment);
@@ -182,6 +185,7 @@ Dbtux::execREAD_CONFIG_REQ(Signal* signal)
c_fragOpPool.setSize(MaxIndexFragments);
c_scanOpPool.setSize(nScanOp);
c_scanBoundPool.setSize(nScanBoundWords);
+ c_scanLockPool.setSize(nScanLock);
/*
* Index id is physical array index. We seize and initialize all
* index records now.
diff --git a/storage/ndb/src/kernel/blocks/dbtux/DbtuxScan.cpp b/storage/ndb/src/kernel/blocks/dbtux/DbtuxScan.cpp
index 0ba9bd15b53..7d10e4c9cb2 100644
--- a/storage/ndb/src/kernel/blocks/dbtux/DbtuxScan.cpp
+++ b/storage/ndb/src/kernel/blocks/dbtux/DbtuxScan.cpp
@@ -313,7 +313,7 @@ Dbtux::execNEXT_SCANREQ(Signal* signal)
EXECUTE_DIRECT(DBACC, GSN_ACC_LOCKREQ, signal, AccLockReq::UndoSignalLength);
jamEntry();
ndbrequire(lockReq->returnCode == AccLockReq::Success);
- removeAccLockOp(scan, req->accOperationPtr);
+ removeAccLockOp(scanPtr, req->accOperationPtr);
}
if (req->scanFlag == NextScanReq::ZSCAN_COMMIT) {
jam();
@@ -469,7 +469,7 @@ Dbtux::execACC_CHECK_SCAN(Signal* signal)
scan.m_state = ScanOp::Locked;
scan.m_accLockOp = lockReq->accOpPtr;
#ifdef VM_TRACE
- if (debugFlags & DebugScan) {
+ if (debugFlags & (DebugScan | DebugLock)) {
debugOut << "Lock immediate scan " << scanPtr.i << " " << scan << endl;
}
#endif
@@ -481,7 +481,7 @@ Dbtux::execACC_CHECK_SCAN(Signal* signal)
scan.m_lockwait = true;
scan.m_accLockOp = lockReq->accOpPtr;
#ifdef VM_TRACE
- if (debugFlags & DebugScan) {
+ if (debugFlags & (DebugScan | DebugLock)) {
debugOut << "Lock wait scan " << scanPtr.i << " " << scan << endl;
}
#endif
@@ -537,7 +537,7 @@ Dbtux::execACC_CHECK_SCAN(Signal* signal)
if (accLockOp != RNIL) {
scan.m_accLockOp = RNIL;
// remember it until LQH unlocks it
- addAccLockOp(scan, accLockOp);
+ addAccLockOp(scanPtr, accLockOp);
} else {
ndbrequire(scan.m_readCommitted);
// operation RNIL in LQH would signal no tuple returned
@@ -592,7 +592,7 @@ Dbtux::execACCKEYCONF(Signal* signal)
c_scanOpPool.getPtr(scanPtr);
ScanOp& scan = *scanPtr.p;
#ifdef VM_TRACE
- if (debugFlags & DebugScan) {
+ if (debugFlags & (DebugScan | DebugLock)) {
debugOut << "Lock obtained scan " << scanPtr.i << " " << scan << endl;
}
#endif
@@ -637,7 +637,7 @@ Dbtux::execACCKEYREF(Signal* signal)
c_scanOpPool.getPtr(scanPtr);
ScanOp& scan = *scanPtr.p;
#ifdef VM_TRACE
- if (debugFlags & DebugScan) {
+ if (debugFlags & (DebugScan | DebugLock)) {
debugOut << "Lock refused scan " << scanPtr.i << " " << scan << endl;
}
#endif
@@ -681,7 +681,7 @@ Dbtux::execACC_ABORTCONF(Signal* signal)
c_scanOpPool.getPtr(scanPtr);
ScanOp& scan = *scanPtr.p;
#ifdef VM_TRACE
- if (debugFlags & DebugScan) {
+ if (debugFlags & (DebugScan | DebugLock)) {
debugOut << "ACC_ABORTCONF scan " << scanPtr.i << " " << scan << endl;
}
#endif
@@ -1016,18 +1016,9 @@ Dbtux::scanClose(Signal* signal, ScanOpPtr scanPtr)
ScanOp& scan = *scanPtr.p;
ndbrequire(! scan.m_lockwait && scan.m_accLockOp == RNIL);
// unlock all not unlocked by LQH
- for (unsigned i = 0; i < scan.m_maxAccLockOps; i++) {
- if (scan.m_accLockOps[i] != RNIL) {
- jam();
- AccLockReq* const lockReq = (AccLockReq*)signal->getDataPtrSend();
- lockReq->returnCode = RNIL;
- lockReq->requestInfo = AccLockReq::Abort;
- lockReq->accOpPtr = scan.m_accLockOps[i];
- EXECUTE_DIRECT(DBACC, GSN_ACC_LOCKREQ, signal, AccLockReq::UndoSignalLength);
- jamEntry();
- ndbrequire(lockReq->returnCode == AccLockReq::Success);
- scan.m_accLockOps[i] = RNIL;
- }
+ if (! scan.m_accLockOps.isEmpty()) {
+ jam();
+ abortAccLockOps(signal, scanPtr);
}
// send conf
NextScanConf* const conf = (NextScanConf*)signal->getDataPtrSend();
@@ -1041,44 +1032,76 @@ Dbtux::scanClose(Signal* signal, ScanOpPtr scanPtr)
}
void
-Dbtux::addAccLockOp(ScanOp& scan, Uint32 accLockOp)
+Dbtux::abortAccLockOps(Signal* signal, ScanOpPtr scanPtr)
{
- ndbrequire(accLockOp != RNIL);
- Uint32* list = scan.m_accLockOps;
- bool ok = false;
- for (unsigned i = 0; i < scan.m_maxAccLockOps; i++) {
- ndbrequire(list[i] != accLockOp);
- if (! ok && list[i] == RNIL) {
- list[i] = accLockOp;
- ok = true;
- // continue check for duplicates
- }
+ ScanOp& scan = *scanPtr.p;
+#ifdef VM_TRACE
+ if (debugFlags & (DebugScan | DebugLock)) {
+ debugOut << "Abort locks in scan " << scanPtr.i << " " << scan << endl;
}
- if (! ok) {
- unsigned i = scan.m_maxAccLockOps;
- if (i < MaxAccLockOps) {
- list[i] = accLockOp;
- ok = true;
- scan.m_maxAccLockOps = i + 1;
- }
+#endif
+ LocalDLFifoList<ScanLock> list(c_scanLockPool, scan.m_accLockOps);
+ ScanLockPtr lockPtr;
+ while (list.first(lockPtr)) {
+ jam();
+ AccLockReq* const lockReq = (AccLockReq*)signal->getDataPtrSend();
+ lockReq->returnCode = RNIL;
+ lockReq->requestInfo = AccLockReq::Abort;
+ lockReq->accOpPtr = lockPtr.p->m_accLockOp;
+ EXECUTE_DIRECT(DBACC, GSN_ACC_LOCKREQ, signal, AccLockReq::UndoSignalLength);
+ jamEntry();
+ ndbrequire(lockReq->returnCode == AccLockReq::Success);
+ list.release(lockPtr);
}
- ndbrequire(ok);
}
void
-Dbtux::removeAccLockOp(ScanOp& scan, Uint32 accLockOp)
+Dbtux::addAccLockOp(ScanOpPtr scanPtr, Uint32 accLockOp)
{
+ ScanOp& scan = *scanPtr.p;
+#ifdef VM_TRACE
+ if (debugFlags & (DebugScan | DebugLock)) {
+ debugOut << "Add lock " << hex << accLockOp << dec
+ << " to scan " << scanPtr.i << " " << scan << endl;
+ }
+#endif
+ LocalDLFifoList<ScanLock> list(c_scanLockPool, scan.m_accLockOps);
+ ScanLockPtr lockPtr;
+#ifdef VM_TRACE
+ list.first(lockPtr);
+ while (lockPtr.i != RNIL) {
+ ndbrequire(lockPtr.p->m_accLockOp != accLockOp);
+ list.next(lockPtr);
+ }
+#endif
+ bool ok = list.seize(lockPtr);
+ ndbrequire(ok);
ndbrequire(accLockOp != RNIL);
- Uint32* list = scan.m_accLockOps;
- bool ok = false;
- for (unsigned i = 0; i < scan.m_maxAccLockOps; i++) {
- if (list[i] == accLockOp) {
- list[i] = RNIL;
- ok = true;
+ lockPtr.p->m_accLockOp = accLockOp;
+}
+
+void
+Dbtux::removeAccLockOp(ScanOpPtr scanPtr, Uint32 accLockOp)
+{
+ ScanOp& scan = *scanPtr.p;
+#ifdef VM_TRACE
+ if (debugFlags & (DebugScan | DebugLock)) {
+ debugOut << "Remove lock " << hex << accLockOp << dec
+ << " from scan " << scanPtr.i << " " << scan << endl;
+ }
+#endif
+ LocalDLFifoList<ScanLock> list(c_scanLockPool, scan.m_accLockOps);
+ ScanLockPtr lockPtr;
+ list.first(lockPtr);
+ while (lockPtr.i != RNIL) {
+ if (lockPtr.p->m_accLockOp == accLockOp) {
+ jam();
break;
}
+ list.next(lockPtr);
}
- ndbrequire(ok);
+ ndbrequire(lockPtr.i != RNIL);
+ list.release(lockPtr);
}
/*
View Lorry Controller Status