ndb - bug#27756

  testcase only
  new pseudo column NDB$COPY_ROWID

storage/ndb/include/kernel/AttributeHeader.hpp

@@ -46,6 +46,8 @@ public:
   STATIC_CONST( ROW_GCI      = 0xFFF5 );
   STATIC_CONST( FRAGMENT_VARSIZED_MEMORY = 0xFFF4 );
 
+  STATIC_CONST( COPY_ROWID   = 0xFFF1 );
+  
   // NOTE: in 5.1 ctors and init take size in bytes
 
   /** Initialize AttributeHeader at location aHeaderPtr */

storage/ndb/include/ndbapi/NdbDictionary.hpp

@@ -534,6 +534,7 @@ public:
     static const Column * RECORDS_IN_RANGE;
     static const Column * ROWID;
     static const Column * ROW_GCI;
+    static const Column * COPY_ROWID;
     
     int getSizeInBytes() const;
 #endif

storage/ndb/src/kernel/blocks/dbtup/DbtupRoutines.cpp

@@ -1198,6 +1198,10 @@ Dbtup::read_pseudo(Uint32 attrId,
       return 2;
     }
     return 0;
+  case AttributeHeader::COPY_ROWID:
+    outBuffer[0] = operPtr.p->m_copy_tuple_location.m_page_no;
+    outBuffer[0] = operPtr.p->m_copy_tuple_location.m_page_idx;
+    return 2;
   default:
     return 0;
   }

storage/ndb/src/ndbapi/NdbDictionaryImpl.cpp

@@ -378,6 +378,11 @@ NdbColumnImpl::create_pseudo(const char * name){
     col->m_impl.m_attrSize = 8;
     col->m_impl.m_arraySize = 1;
     col->m_impl.m_nullable = true;
+  } else if(!strcmp(name, "NDB$COPY_ROWID")){
+    col->setType(NdbDictionary::Column::Bigunsigned);
+    col->m_impl.m_attrId = AttributeHeader::COPY_ROWID;
+    col->m_impl.m_attrSize = 4;
+    col->m_impl.m_arraySize = 2;
   } else {
     abort();
   }
@@ -5089,3 +5094,4 @@ const NdbDictionary::Column * NdbDictionary::Column::DISK_REF = 0;
 const NdbDictionary::Column * NdbDictionary::Column::RECORDS_IN_RANGE = 0;
 const NdbDictionary::Column * NdbDictionary::Column::ROWID = 0;
 const NdbDictionary::Column * NdbDictionary::Column::ROW_GCI = 0;
+const NdbDictionary::Column * NdbDictionary::Column::COPY_ROWID = 0;

storage/ndb/src/ndbapi/ndb_cluster_connection.cpp

@@ -326,6 +326,8 @@ Ndb_cluster_connection_impl::Ndb_cluster_connection_impl(const char *
       NdbColumnImpl::create_pseudo("NDB$ROWID");
     NdbDictionary::Column::ROW_GCI= 
       NdbColumnImpl::create_pseudo("NDB$ROW_GCI");
+    NdbDictionary::Column::COPY_ROWID= 
+      NdbColumnImpl::create_pseudo("NDB$COPY_ROWID");
   }
   NdbMutex_Unlock(g_ndb_connection_mutex);
 
@@ -391,6 +393,9 @@ Ndb_cluster_connection_impl::~Ndb_cluster_connection_impl()
     NdbDictionary::Column::RECORDS_IN_RANGE= 0;
     NdbDictionary::Column::ROWID= 0;
     NdbDictionary::Column::ROW_GCI= 0;
+
+    delete NdbDictionary::Column::COPY_ROWID;
+    NdbDictionary::Column::COPY_ROWID = 0;
   }
   NdbMutex_Unlock(g_ndb_connection_mutex);
 

storage/ndb/test/ndbapi/testBasic.cpp

@@ -1272,6 +1272,64 @@ runBug25090(NDBT_Context* ctx, NDBT_Step* step){
   return NDBT_OK;
 }
 
+int
+runBug27756(NDBT_Context* ctx, NDBT_Step* step)
+{
+  
+  Ndb* pNdb = GETNDB(step);
+  NdbDictionary::Dictionary * dict = pNdb->getDictionary();
+  
+  HugoOperations ops(*ctx->getTab());
+
+  int loops = ctx->getNumLoops();
+  const int rows = ctx->getNumRecords();
+  
+  Vector<Uint64> copies;
+  while (loops--)
+  {
+    ops.startTransaction(pNdb);
+    ops.pkInsertRecord(pNdb, 1, 1);
+    ops.execute_NoCommit(pNdb);
+    
+    NdbTransaction* pTrans = ops.getTransaction();
+    NdbOperation* op = pTrans->getNdbOperation(ctx->getTab()->getName());
+    op->interpretedUpdateTuple();
+    ops.equalForRow(op, 1);
+    NdbRecAttr* attr = op->getValue(NdbDictionary::Column::COPY_ROWID);
+    ops.execute_NoCommit(pNdb);
+    
+    copies.push_back(attr->u_64_value());
+    ndbout_c("copy at: %llx", copies.back());
+    ops.execute_NoCommit(pNdb);
+    
+    ops.pkDeleteRecord(pNdb, 1, 1);
+    ops.execute_NoCommit(pNdb);
+    
+    if (loops & 1)
+    {
+      ops.execute_Rollback(pNdb);
+      ops.closeTransaction(pNdb);
+    }
+    else
+    {
+      ops.execute_Commit(pNdb);
+      ops.closeTransaction(pNdb);
+      ops.clearTable(pNdb, 100);
+    }
+  }
+  
+  for (Uint32 i = 0; i<copies.size(); i++)
+    if (copies[i] != copies.back())
+    {
+      ndbout_c("Memleak detected");
+      return NDBT_FAILED;
+    }
+  
+  return NDBT_OK;
+}
+
+template class Vector<Uint64>;
+
 NDBT_TESTSUITE(testBasic);
 TESTCASE("PkInsert", 
 	 "Verify that we can insert and delete from this table using PK"
@@ -1542,6 +1600,10 @@ TESTCASE("Bug25090",
 	 "Verify what happens when we fill the db" ){
   STEP(runBug25090);
 }
+TESTCASE("Bug27756", 
+	 "Verify what happens when we fill the db" ){
+  STEP(runBug27756);
+}
 NDBT_TESTSUITE_END(testBasic);
 
 #if 0