ndb - csc#4847 release scan op early to save memory

ndb/include/ndbapi/NdbConnection.hpp:
  release scan op of hupped trans at scan close to save memory
ndb/include/ndbapi/NdbScanOperation.hpp:
  release scan op of hupped trans at scan close to save memory
ndb/src/ndbapi/NdbConnection.cpp:
  release scan op of hupped trans at scan close to save memory
ndb/src/ndbapi/NdbResultSet.cpp:
  release scan op of hupped trans at scan close to save memory
ndb/src/ndbapi/NdbScanOperation.cpp:
  release scan op of hupped trans at scan close to save memory
ndb/tools/desc.cpp:
  release scan op of hupped trans at scan close to save memory

ndb/include/ndbapi/NdbConnection.hpp

@@ -542,6 +542,7 @@ private:
   // Release all cursor operations in connection
   void releaseOps(NdbOperation*);	
   void releaseScanOperations(NdbIndexScanOperation*);	
+  void releaseExecutedScanOperation(NdbIndexScanOperation*);
 
   // Set the transaction identity of the transaction
   void		setTransactionId(Uint64 aTransactionId);

ndb/include/ndbapi/NdbScanOperation.hpp

@@ -93,7 +93,7 @@ protected:
   int nextResult(bool fetchAllowed = true, bool forceSend = false);
   virtual void release();
   
-  void closeScan(bool forceSend = false);
+  void closeScan(bool forceSend = false, bool releaseOp = false);
   int close_impl(class TransporterFacade*, bool forceSend = false);
 
   // Overloaded methods from NdbCursorOperation

ndb/src/ndbapi/NdbConnection.cpp

@@ -963,6 +963,37 @@ NdbConnection::releaseScanOperations(NdbIndexScanOperation* cursorOp)
   }
 }//NdbConnection::releaseScanOperations()
 
+/*****************************************************************************
+void releaseExecutedScanOperation();
+
+Remark:         Release scan op when hupp'ed trans closed (save memory)
+******************************************************************************/
+void 
+NdbConnection::releaseExecutedScanOperation(NdbIndexScanOperation* cursorOp)
+{
+  DBUG_ENTER("NdbConnection::releaseExecutedScanOperation");
+  DBUG_PRINT("enter", ("this=0x%x op=0x%x", (UintPtr)this, (UintPtr)cursorOp))
+
+  // here is one reason to make op lists doubly linked
+  if (m_firstExecutedScanOp == cursorOp) {
+    m_firstExecutedScanOp = (NdbIndexScanOperation*)cursorOp->theNext;
+    cursorOp->release();
+    theNdb->releaseScanOperation(cursorOp);
+  } else if (m_firstExecutedScanOp != NULL) {
+    NdbIndexScanOperation* tOp = m_firstExecutedScanOp;
+    while (tOp->theNext != NULL) {
+      if (tOp->theNext == cursorOp) {
+        tOp->theNext = cursorOp->theNext;
+        cursorOp->release();
+        theNdb->releaseScanOperation(cursorOp);
+        break;
+      }
+      tOp = (NdbIndexScanOperation*)tOp->theNext;
+    }
+  }
+  DBUG_VOID_RETURN;
+}//NdbConnection::releaseExecutedScanOperation()
+
 /*****************************************************************************
 NdbOperation* getNdbOperation(const char* aTableName);
 

ndb/src/ndbapi/NdbResultSet.cpp

@@ -69,7 +69,7 @@ int NdbResultSet::nextResult(bool fetchAllowed, bool forceSend)
 
 void NdbResultSet::close(bool forceSend)
 {
-  m_operation->closeScan(forceSend);
+  m_operation->closeScan(forceSend, true);
 }
 
 NdbOperation* 

ndb/src/ndbapi/NdbScanOperation.cpp

@@ -674,7 +674,7 @@ NdbScanOperation::doSend(int ProcessorId)
   return 0;
 }
 
-void NdbScanOperation::closeScan(bool forceSend)
+void NdbScanOperation::closeScan(bool forceSend, bool releaseOp)
 {
   if(m_transConnection){
     if(DEBUG_NEXT_RESULT)
@@ -691,13 +691,20 @@ void NdbScanOperation::closeScan(bool forceSend)
     Guard guard(tp->theMutexPtr);
     close_impl(tp, forceSend);
     
-  } while(0);
-  
-  theNdbCon->theScanningOp = 0;
-  theNdb->closeTransaction(theNdbCon);
+  }
 
-  theNdbCon = 0;
+  NdbConnection* tCon = theNdbCon;
+  NdbConnection* tTransCon = m_transConnection;
+  theNdbCon = NULL;
   m_transConnection = NULL;
+
+  if (releaseOp && tTransCon) {
+    NdbIndexScanOperation* tOp = (NdbIndexScanOperation*)this;
+    tTransCon->releaseExecutedScanOperation(tOp);
+  }
+  
+  tCon->theScanningOp = 0;
+  theNdb->closeTransaction(tCon);
 }
 
 void

ndb/tools/desc.cpp

@@ -89,7 +89,7 @@ int main(int argc, char** argv){
       unsigned j;
       for (j= 0; (int)j < pTab->getNoOfPrimaryKeys(); j++)
       {
-	const NdbDictionary::Column * col = pTab->getColumn(j);
+	const NdbDictionary::Column * col = pTab->getColumn(pTab->getPrimaryKey(j));
 	ndbout << col->getName();
 	if ((int)j < pTab->getNoOfPrimaryKeys()-1)
 	  ndbout << ", ";