Merge joreland@bk-internal.mysql.com:/home/bk/mysql-5.1-new-ndb

into  perch.ndb.mysql.com:/home/jonas/src/mysql-5.1-new-ndb

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;

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

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:

sql/ha_ndbcluster.h

@@ -808,7 +808,7 @@ static void set_tabname(const char *pathname, char *tabname);
   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,

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*");

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
@@ -339,6 +340,18 @@ private:
   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

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 << "]";

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.

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) {
+  if (! scan.m_accLockOps.isEmpty()) {
     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;
-    }
+    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;
   }
+#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);
   }
-  if (! ok) {
-    unsigned i = scan.m_maxAccLockOps;
-    if (i < MaxAccLockOps) {
-      list[i] = accLockOp;
-      ok = true;
-      scan.m_maxAccLockOps = i + 1;
+}
+
+void
+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);
+  lockPtr.p->m_accLockOp = accLockOp;
 }
 
 void
-Dbtux::removeAccLockOp(ScanOp& scan, Uint32 accLockOp)
+Dbtux::removeAccLockOp(ScanOpPtr scanPtr, Uint32 accLockOp)
 {
-  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;
+  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);
 }
 
 /*