bug#21072 Duplicate key error in NDB references wrong key: Return correct key...

bug#21072 Duplicate key error in NDB references wrong key: Return correct key for non-batching inserts

mysql-test/r/ndb_charset.result

@@ -112,9 +112,9 @@ unique key(a)
 ) engine=ndb;
 insert into t1 values(1, 'aAa');
 insert into t1 values(2, 'aaa');
-ERROR 23000: Duplicate entry '' for key 0
+ERROR 23000: Duplicate entry 'aaa' for key 2
 insert into t1 values(3, 'AAA');
-ERROR 23000: Duplicate entry '' for key 0
+ERROR 23000: Duplicate entry 'AAA' for key 2
 select * from t1 order by p;
 p	a
 1	aAa
@@ -138,9 +138,9 @@ unique key(a)
 ) engine=ndb;
 insert into t1 values (1,'A'),(2,'b '),(3,'C  '),(4,'d      '),(5,'E'),(6,'f');
 insert into t1 values(99,'b');
-ERROR 23000: Duplicate entry '' for key 0
+ERROR 23000: Duplicate entry 'b' for key 2
 insert into t1 values(99,'a         ');
-ERROR 23000: Duplicate entry '' for key 0
+ERROR 23000: Duplicate entry 'a         ' for key 2
 select a,length(a) from t1 order by a;
 a	length(a)
 A	1

mysql-test/r/ndb_index_unique.result

@@ -22,7 +22,7 @@ select * from t1 where b = 4 order by a;
 a	b	c
 3	4	6
 insert into t1 values(8, 2, 3);
-ERROR 23000: Duplicate entry '' for key 0
+ERROR 23000: Duplicate entry '2' for key 2
 select * from t1 order by a;
 a	b	c
 1	2	3
@@ -89,7 +89,7 @@ a	b	c
 1	1	1
 4	4	NULL
 insert into t1 values(5,1,1);
-ERROR 23000: Duplicate entry '' for key 0
+ERROR 23000: Duplicate entry '1-1' for key 2
 drop table t1;
 CREATE TABLE t2 (
 a int unsigned NOT NULL PRIMARY KEY,
@@ -112,7 +112,7 @@ select * from t2 where b = 4 order by a;
 a	b	c
 3	4	6
 insert into t2 values(8, 2, 3);
-ERROR 23000: Duplicate entry '' for key 0
+ERROR 23000: Duplicate entry '2-3' for key 2
 select * from t2 order by a;
 a	b	c
 1	2	3
@@ -135,7 +135,7 @@ a	b	c
 8	2	3
 create unique index bi using hash on t2(b);
 insert into t2 values(9, 3, 1);
-ERROR 23000: Duplicate entry '' for key 0
+ERROR 23000: Duplicate entry '3' for key 3
 alter table t2 drop index bi;
 insert into t2 values(9, 3, 1);
 select * from t2 order by a;
@@ -225,7 +225,7 @@ pk	a
 3	NULL
 4	4
 insert into t1 values (5,0);
-ERROR 23000: Duplicate entry '' for key 0
+ERROR 23000: Duplicate entry '0' for key 2
 select * from t1 order by pk;
 pk	a
 -1	NULL
@@ -258,7 +258,7 @@ pk	a	b	c
 0	NULL	18	NULL
 1	3	19	abc
 insert into t2 values(2,3,19,'abc');
-ERROR 23000: Duplicate entry '' for key 0
+ERROR 23000: Duplicate entry '3-abc' for key 2
 select * from t2 order by pk;
 pk	a	b	c
 -1	1	17	NULL
@@ -678,7 +678,7 @@ create table t1 (a int primary key, b varchar(1000) not null, unique key (b))
 engine=ndb charset=utf8;
 insert into t1 values (1, repeat(_utf8 0xe288ab6474, 200));
 insert into t1 values (2, repeat(_utf8 0xe288ab6474, 200));
-ERROR 23000: Duplicate entry '' for key 0
+ERROR 23000: Duplicate entry '∫dt∫dt∫dt∫dt∫dt∫dt∫dt∫dt∫dt∫dt∫dt∫dt∫d' for key 2
 select a, sha1(b) from t1;
 a	sha1(b)
 1	08f5d02c8b8bc244f275bdfc22c42c5cab0d9d7d

mysql-test/r/ndb_update.result

@@ -26,7 +26,7 @@ pk1	b	c
 2	2	2
 4	1	1
 UPDATE t1 set pk1 = 1, c = 2 where pk1 = 4;
-ERROR 23000: Duplicate entry '' for key 0
+ERROR 23000: Duplicate entry '2' for key 2
 UPDATE IGNORE t1 set pk1 = 1, c = 2 where pk1 = 4;
 select * from t1 order by pk1;
 pk1	b	c

ndb/include/kernel/signaldata/TcKeyRef.hpp

@@ -40,12 +40,13 @@ class TcKeyRef {
   friend bool printTCKEYREF(FILE *, const Uint32 *, Uint32, Uint16);
   
 public:
-  STATIC_CONST( SignalLength = 4 );
+  STATIC_CONST( SignalLength = 5 );
 
 private:
   Uint32 connectPtr;
   Uint32 transId[2];
   Uint32 errorCode;
+  Uint32 errorData;
 };
 
 #endif

ndb/include/kernel/signaldata/TcRollbackRep.hpp

@@ -38,12 +38,13 @@ class TcRollbackRep {
   friend bool printTCROLBACKREP(FILE *, const Uint32 *, Uint32, Uint16);
   
 public:
-  STATIC_CONST( SignalLength = 4 );
+  STATIC_CONST( SignalLength = 5 );
 
 private:
   Uint32 connectPtr;
   Uint32 transId[2];
   Uint32 returnCode;
+  Uint32 errorData;
 };
 
 #endif

ndb/include/ndbapi/NdbDictionary.hpp

@@ -792,7 +792,12 @@ public:
      * Get the name of the table being indexed
      */
     const char * getTable() const;
-    
+
+    /**
+     * Get the table representing the index
+     */
+    const Table * getIndexTable() const;
+
     /**
      * Get the number of columns in the index
      */

ndb/src/kernel/blocks/dbtc/Dbtc.hpp

@@ -727,6 +727,7 @@ public:
     // Index op return context
     UintR indexOp;
     UintR clientData;
+    Uint32 errorData;
     UintR attrInfoLen;
     
     UintR accumulatingIndexOp;

ndb/src/kernel/blocks/dbtc/DbtcMain.cpp

@@ -5107,6 +5107,7 @@ void Dbtc::releaseDirtyWrite(Signal* signal)
 void Dbtc::execLQHKEYREF(Signal* signal) 
 {
   const LqhKeyRef * const lqhKeyRef = (LqhKeyRef *)signal->getDataPtr();
+  Uint32 indexId = 0;
   jamEntry();
   
   UintR compare_transid1, compare_transid2;
@@ -5158,6 +5159,10 @@ void Dbtc::execLQHKEYREF(Signal* signal)
 	ptrCheckGuard(opPtr, ctcConnectFilesize, localTcConnectRecord);
 	
 	// The operation executed an index trigger
+        TcIndexData* indexData = c_theIndexes.getPtr(currentIndexId);
+        indexId = indexData->indexId;
+        regApiPtr->errorData = indexId;
+        ndbout_c("LQHKEYREF, found index %u", indexId);
 	const Uint32 opType = regTcPtr->operation;
 	if (errCode == ZALREADYEXIST)
 	  errCode = terrorCode = ZNOTUNIQUE;
@@ -5170,7 +5175,6 @@ void Dbtc::execLQHKEYREF(Signal* signal)
 	} else {
 	  jam();
 	  /** ZDELETE && NOT_FOUND */
-	  TcIndexData* indexData = c_theIndexes.getPtr(currentIndexId);
 	  if(indexData->indexState == IS_BUILDING && state != CS_ABORTING){
 	    jam();
 	    /**
@@ -5242,12 +5246,16 @@ void Dbtc::execLQHKEYREF(Signal* signal)
         jam();
 	regApiPtr->lqhkeyreqrec--; // Compensate for extra during read
 	tcKeyRef->connectPtr = indexOp;
+        ndbout_c("TCKEYREF, sending index %u", indexId);
+        tcKeyRef->errorData = indexId;
 	EXECUTE_DIRECT(DBTC, GSN_TCKEYREF, signal, TcKeyRef::SignalLength);
 	apiConnectptr.i = save;
 	apiConnectptr.p = regApiPtr;
       } else {
         jam();
 	tcKeyRef->connectPtr = clientData;
+        ndbout_c("TCKEYREF, sending index %u", indexId);
+        tcKeyRef->errorData = indexId;
 	sendSignal(regApiPtr->ndbapiBlockref, 
 		   GSN_TCKEYREF, signal, TcKeyRef::SignalLength, JBB);
       }//if
@@ -10548,6 +10556,7 @@ void Dbtc::releaseAbortResources(Signal* signal)
       tcRollbackRep->transId[0] = apiConnectptr.p->transid[0];
       tcRollbackRep->transId[1] = apiConnectptr.p->transid[1];
       tcRollbackRep->returnCode = apiConnectptr.p->returncode;
+      tcRollbackRep->errorData = apiConnectptr.p->errorData;
       sendSignal(blockRef, GSN_TCROLLBACKREP, signal, 
 		 TcRollbackRep::SignalLength, JBB);
     }
@@ -11972,6 +11981,7 @@ void Dbtc::execTCKEYCONF(Signal* signal)
     tcIndxRef->transId[0] = regApiPtr->transid[0];
     tcIndxRef->transId[1] = regApiPtr->transid[1];
     tcIndxRef->errorCode = 4349;    
+    tcIndxRef->errorData = 0;
     sendSignal(regApiPtr->ndbapiBlockref, GSN_TCINDXREF, signal, 
 	       TcKeyRef::SignalLength, JBB);
     return;
@@ -11991,6 +12001,7 @@ void Dbtc::execTCKEYCONF(Signal* signal)
     tcIndxRef->transId[0] = regApiPtr->transid[0];
     tcIndxRef->transId[1] = regApiPtr->transid[1];
     tcIndxRef->errorCode = 4349;    
+    tcIndxRef->errorData = 0;
     sendSignal(regApiPtr->ndbapiBlockref, GSN_TCINDXREF, signal, 
 	       TcKeyRef::SignalLength, JBB);
     return;
@@ -12074,6 +12085,7 @@ void Dbtc::execTCKEYREF(Signal* signal)
     tcIndxRef->transId[0] = tcKeyRef->transId[0];
     tcIndxRef->transId[1] = tcKeyRef->transId[1];
     tcIndxRef->errorCode = tcKeyRef->errorCode;
+    tcIndxRef->errorData = 0;
 
     releaseIndexOperation(regApiPtr, indexOp);
 
@@ -12151,6 +12163,7 @@ void Dbtc::execTRANSID_AI(Signal* signal)
     tcIndxRef->transId[0] = regApiPtr->transid[0];
     tcIndxRef->transId[1] = regApiPtr->transid[1];
     tcIndxRef->errorCode = 4000;
+    tcIndxRef->errorData = 0;
     sendSignal(regApiPtr->ndbapiBlockref, GSN_TCINDXREF, signal, 
 	       TcKeyRef::SignalLength, JBB);
     return;
@@ -12166,6 +12179,7 @@ void Dbtc::execTRANSID_AI(Signal* signal)
     tcIndxRef->transId[0] = regApiPtr->transid[0];
     tcIndxRef->transId[1] = regApiPtr->transid[1];
     tcIndxRef->errorCode = 4349;
+    tcIndxRef->errorData = 0;
     sendSignal(regApiPtr->ndbapiBlockref, GSN_TCINDXREF, signal, 
 	       TcKeyRef::SignalLength, JBB);
     return;
@@ -12194,6 +12208,7 @@ void Dbtc::execTRANSID_AI(Signal* signal)
     tcIndxRef->transId[0] = regApiPtr->transid[0];
     tcIndxRef->transId[1] = regApiPtr->transid[1];
     tcIndxRef->errorCode = 4349;
+    tcIndxRef->errorData = 0;
     sendSignal(regApiPtr->ndbapiBlockref, GSN_TCINDXREF, signal, 
                TcKeyRef::SignalLength, JBB);
     */
@@ -12219,6 +12234,7 @@ void Dbtc::execTRANSID_AI(Signal* signal)
     tcIndxRef->transId[0] = regApiPtr->transid[0];
     tcIndxRef->transId[1] = regApiPtr->transid[1];
     tcIndxRef->errorCode = 4349;
+    // tcIndxRef->errorData = ??; Where to find indexId
     sendSignal(regApiPtr->ndbapiBlockref, GSN_TCINDXREF, signal, 
 	       TcKeyRef::SignalLength, JBB);
     return;
@@ -12272,6 +12288,7 @@ void Dbtc::readIndexTable(Signal* signal,
     tcIndxRef->transId[0] = regApiPtr->transid[0];
     tcIndxRef->transId[1] = regApiPtr->transid[1];
     tcIndxRef->errorCode = 4000;    
+    // tcIndxRef->errorData = ??; Where to find indexId
     sendSignal(regApiPtr->ndbapiBlockref, GSN_TCINDXREF, signal, 
 	       TcKeyRef::SignalLength, JBB);
     return;
@@ -12414,6 +12431,7 @@ void Dbtc::executeIndexOperation(Signal* signal,
     tcIndxRef->transId[0] = regApiPtr->transid[0];
     tcIndxRef->transId[1] = regApiPtr->transid[1];
     tcIndxRef->errorCode = 4349;    
+    tcIndxRef->errorData = 0;
     sendSignal(regApiPtr->ndbapiBlockref, GSN_TCINDXREF, signal, 
 	       TcKeyRef::SignalLength, JBB);
     return;
@@ -13012,6 +13030,7 @@ void Dbtc::insertIntoIndexTable(Signal* signal,
   }
 
   regApiPtr->currSavePointId = currSavePointId;
+  ndbout_c("TCKEYREQ, saving index %u", indexData->indexId);
   tcConnectptr.p->currentIndexId = indexData->indexId;
 
   // *********** KEYINFO ***********

ndb/src/ndbapi/NdbDictionary.cpp

@@ -542,6 +542,15 @@ NdbDictionary::Index::getTable() const {
   return m_impl.getTable();
 }
 
+const NdbDictionary::Table *
+NdbDictionary::Index::getIndexTable() const {
+  NdbTableImpl * t = m_impl.m_table;
+  if (t) {
+    return t->m_facade;
+  }
+  return 0;
+}
+
 unsigned
 NdbDictionary::Index::getNoOfColumns() const {
   return m_impl.m_columns.size();

ndb/src/ndbapi/NdbOperationExec.cpp

@@ -24,6 +24,7 @@
 #include "Interpreter.hpp"
 #include <AttributeHeader.hpp>
 #include <signaldata/TcKeyReq.hpp>
+#include <signaldata/TcKeyRef.hpp>
 #include <signaldata/KeyInfo.hpp>
 #include <signaldata/AttrInfo.hpp>
 #include <signaldata/ScanTab.hpp>
@@ -550,6 +551,12 @@ NdbOperation::receiveTCKEYREF( NdbApiSignal* aSignal)
     theNdbCon->theReturnStatus = NdbTransaction::ReturnFailure;
   }
   theError.code = aSignal->readData(4);
+  if (aSignal->getLength() == TcKeyRef::SignalLength)
+  {
+    // Signal may contain additional error data
+    theError.details = (char *) aSignal->readData(5);
+  }
+
   theNdbCon->setOperationErrorCodeAbort(aSignal->readData(4), ao);
 
   if(theOperationType != ReadRequest || !theSimpleIndicator) // not simple read

ndb/src/ndbapi/NdbTransaction.cpp

@@ -30,6 +30,7 @@
 #include <signaldata/TcCommit.hpp>
 #include <signaldata/TcKeyFailConf.hpp>
 #include <signaldata/TcHbRep.hpp>
+#include <signaldata/TcRollbackRep.hpp>
 
 /*****************************************************************************
 NdbTransaction( Ndb* aNdb );
@@ -1757,6 +1758,8 @@ Remark:         Handles the reception of the ROLLBACKREP signal.
 int
 NdbTransaction::receiveTCROLLBACKREP( NdbApiSignal* aSignal)
 {
+  DBUG_ENTER("NdbTransaction::receiveTCROLLBACKREP");
+
   /****************************************************************************
 Check that we are expecting signals from this transaction and that it doesn't
 belong to a transaction already completed. Simply ignore messages from other 
@@ -1764,6 +1767,12 @@ transactions.
   ****************************************************************************/
   if(checkState_TransId(aSignal->getDataPtr() + 1)){
     theError.code = aSignal->readData(4);// Override any previous errors
+    if (aSignal->getLength() == TcRollbackRep::SignalLength)
+    {
+      DBUG_PRINT("info", ("Found error data %u", aSignal->readData(5)));
+      // Signal may contain additional error data
+      theError.details = (char *) aSignal->readData(5);
+    }
 
     /**********************************************************************/
     /*	A serious error has occured. This could be due to deadlock or */
@@ -1775,14 +1784,14 @@ transactions.
     theCompletionStatus = CompletedFailure;
     theCommitStatus = Aborted;
     theReturnStatus = ReturnFailure;
-    return 0;
+    DBUG_RETURN(0);
   } else {
 #ifdef NDB_NO_DROPPED_SIGNAL
     abort();
 #endif
   }
 
-  return -1;
+  DBUG_RETURN(-1);
 }//NdbTransaction::receiveTCROLLBACKREP()
 
 /*******************************************************************************

ndb/src/ndbapi/ndberror.c

@@ -640,8 +640,6 @@ ndberror_update(ndberror_struct * error){
   if(!found){
     error->status = ST_U;
   }
-
-  error->details = 0;
 }
 
 int

sql/ha_ndbcluster.cc

@@ -540,6 +540,27 @@ int ha_ndbcluster::ndb_err(NdbTransaction *trans)
                       err.code, res));
   if (res == HA_ERR_FOUND_DUPP_KEY)
   {
+    uint error_data= (uint) err.details;
+    uint dupkey= MAX_KEY;
+
+    DBUG_PRINT("info", ("HA_ERR_FOUND_DUPP_KEY, index table %u", error_data));
+    for (uint i= 0; i < MAX_KEY; i++)
+    {
+      if (m_index[i].type == UNIQUE_INDEX || 
+          m_index[i].type == UNIQUE_ORDERED_INDEX)
+      {
+        const NDBINDEX *unique_index=
+          (const NDBINDEX *) m_index[i].unique_index;
+        if (unique_index &&
+            unique_index->getIndexTable() &&
+            (uint) unique_index->getIndexTable()->getTableId() == error_data)
+        {
+          DBUG_PRINT("info", ("Found violated key %u", i));
+          dupkey= i;
+          break;
+        }
+      }
+    }
     if (m_rows_to_insert == 1)
     {
       /*
@@ -547,7 +568,7 @@ int ha_ndbcluster::ndb_err(NdbTransaction *trans)
 	violations here, so we need to return MAX_KEY for non-primary
 	to signal that key is unknown
       */
-      m_dupkey= err.code == 630 ? table->s->primary_key : MAX_KEY; 
+      m_dupkey= err.code == 630 ? table->s->primary_key : dupkey; 
     }
     else
     {