Merge pnousiainen@bk-internal.mysql.com:/home/bk/mysql-5.0-ndb

into mysql.com:/space/pekka/ndb/version/my50-cc


ndb/src/kernel/blocks/dbacc/DbaccMain.cpp:
  Auto merged
ndb/src/kernel/blocks/dbdict/Dbdict.cpp:
  Auto merged

ndb/src/kernel/blocks/dbacc/Dbacc.hpp

@@ -22,6 +22,9 @@
 #include <pc.hpp>
 #include <SimulatedBlock.hpp>
 
+// primary key is stored in TUP
+#include <Dbtup.hpp>
+
 #ifdef DBACC_C
 // Debug Macros
 #define dbgWord32(ptr, ind, val) 
@@ -661,9 +664,10 @@ struct Fragmentrec {
 //-----------------------------------------------------------------------------
 // elementLength: Length of element in bucket and overflow pages
 // keyLength: Length of key (== 0 if long key or variable key length)
+// wl-2066 always Length of key
 //-----------------------------------------------------------------------------
   Uint8 elementLength;
-  Uint8 keyLength;
+  Uint16 keyLength;
 
 //-----------------------------------------------------------------------------
 // This flag is used to avoid sending a big number of expand or shrink signals
@@ -783,6 +787,7 @@ struct Operationrec {
   Uint8 dirtyRead;
   Uint8 commitDeleteCheckFlag;
   Uint8 isAccLockReq;
+  Uint8 isUndoLogReq;
   Uint32 nextOpList;
 }; /* p2c: size = 168 bytes */
 
@@ -914,6 +919,9 @@ public:
   Dbacc(const class Configuration &);
   virtual ~Dbacc();
 
+  // pointer to TUP instance in this thread
+  Dbtup* c_tup;
+
 private:
   BLOCK_DEFINES(Dbacc);
 
@@ -1075,6 +1083,7 @@ private:
   void storeLongKeys(Signal* signal);
   void storeLongKeysAtPos(Signal* signal);
   void reorgLongPage(Signal* signal);
+  void readTablePk(Uint32 localkey1);
   void getElement(Signal* signal);
   void searchLongKey(Signal* signal);
   void getdirindex(Signal* signal);
@@ -1562,7 +1571,10 @@ private:
   Uint32 cexcPrevpageindex;
   Uint32 cexcPrevforward;
   Uint32 clocalkey[32];
+  union {
   Uint32 ckeys[2048];
+  Uint64 ckeys_align;
+  };
   
   Uint32 c_errorInsert3000_TableId;
   Uint32 cSrUndoRecords[5];

ndb/src/kernel/blocks/dbacc/DbaccInit.cpp

@@ -133,7 +133,8 @@ void Dbacc::initRecords()
 }//Dbacc::initRecords()
 
 Dbacc::Dbacc(const class Configuration & conf):
-  SimulatedBlock(DBACC, conf)
+  SimulatedBlock(DBACC, conf),
+  c_tup(0)
 {
   Uint32 log_page_size= 0;
   BLOCK_CONSTRUCTOR(Dbacc);

ndb/src/kernel/blocks/dbacc/Makefile.am

@@ -3,6 +3,8 @@ noinst_LIBRARIES = libdbacc.a
 
 libdbacc_a_SOURCES = DbaccInit.cpp DbaccMain.cpp
 
+INCLUDES_LOC = -I$(top_srcdir)/ndb/src/kernel/blocks/dbtup
+
 include $(top_srcdir)/ndb/config/common.mk.am
 include $(top_srcdir)/ndb/config/type_kernel.mk.am
 

ndb/src/kernel/blocks/dbdict/Dbdict.cpp

@@ -4103,7 +4103,7 @@ Dbdict::execADD_FRAGREQ(Signal* signal) {
     req->noOfPagesToPreAllocate = 0;
     req->schemaVersion = tabPtr.p->tableVersion;
     Uint32 keyLen = tabPtr.p->tupKeyLength;
-    req->keyLength = keyLen > 8 ? 0 : keyLen; // Put this into ACC instead
+    req->keyLength = keyLen; // wl-2066 no more "long keys"
     req->nextLCP = lcpNo;
 
     req->noOfKeyAttr = tabPtr.p->noOfPrimkey;

ndb/src/kernel/blocks/dbtup/Dbtup.hpp

@@ -1020,6 +1020,13 @@ public:
    */
   int tuxReadPk(Uint32 fragPtrI, Uint32 pageId, Uint32 pageOffset, Uint32* dataOut);
 
+  /*
+   * ACC reads primary key without headers into an array of words.  At
+   * this point in ACC deconstruction, ACC still uses logical references
+   * to fragment and tuple.
+   */
+  int accReadPk(Uint32 tableId, Uint32 fragId, Uint32 fragPageId, Uint32 pageIndex, Uint32* dataOut);
+
   /*
    * TUX checks if tuple is visible to scan.
    */

ndb/src/kernel/blocks/dbtup/DbtupIndex.cpp

@@ -200,8 +200,7 @@ Dbtup::tuxReadPk(Uint32 fragPtrI, Uint32 pageId, Uint32 pageOffset, Uint32* data
   operPtr.i = RNIL;
   operPtr.p = NULL;
   // do it
-  int ret = readAttributes(pagePtr.p, pageOffset, attrIds,
-			   numAttrs, dataOut, ZNIL, true);
+  int ret = readAttributes(pagePtr.p, pageOffset, attrIds, numAttrs, dataOut, ZNIL, true);
   // restore globals
   tabptr = tabptr_old;
   fragptr = fragptr_old;
@@ -229,6 +228,27 @@ Dbtup::tuxReadPk(Uint32 fragPtrI, Uint32 pageId, Uint32 pageOffset, Uint32* data
   return ret;
 }
 
+int
+Dbtup::accReadPk(Uint32 tableId, Uint32 fragId, Uint32 fragPageId, Uint32 pageIndex, Uint32* dataOut)
+{
+  ljamEntry();
+  // get table
+  TablerecPtr tablePtr;
+  tablePtr.i = tableId;
+  ptrCheckGuard(tablePtr, cnoOfTablerec, tablerec);
+  // get fragment
+  FragrecordPtr fragPtr;
+  getFragmentrec(fragPtr, fragId, tablePtr.p);
+  // get real page id and tuple offset
+  PagePtr pagePtr;
+  Uint32 pageId = getRealpid(fragPtr.p, fragPageId);
+  ndbrequire((pageIndex & 0x1) == 0);
+  Uint32 pageOffset = ZPAGE_HEADER_SIZE + (pageIndex >> 1) * tablePtr.p->tupheadsize;
+  // use TUX routine - optimize later
+  int ret = tuxReadPk(fragPtr.i, pageId, pageOffset, dataOut);
+  return ret;
+}
+
 bool
 Dbtup::tuxQueryTh(Uint32 fragPtrI, Uint32 tupAddr, Uint32 tupVersion, Uint32 transId1, Uint32 transId2, Uint32 savePointId)
 {