bk commit into 5.1-ndb tree (pekka:1.1865)

[pekka]

Below is the list of changes that have just been committed into a local
5.1-ndb repository of pekka. When pekka does a push these changes will
be propagated to the main repository and, within 24 hours after the
push, to the public repository.
For information on how to access the public repository
see http://dev.mysql.com/doc/mysql/en/installing-source-tree.html

ChangeSet
  1.1865 05/04/24 17:48:34 pekka@stripped +3 -0
  ndb - wl#2455 cont'd

  ndb/src/kernel/blocks/dbtup/DbtupScan.cpp
    1.4 05/04/24 17:38:12 pekka@stripped +24 -41
    tup scan, fixed MM: use Local_key + free bit in Tuple_header

  ndb/src/kernel/blocks/dbtup/DbtupFixAlloc.cpp
    1.13 05/04/24 17:38:12 pekka@stripped +3 -0
    tup scan, fixed MM: use Local_key + free bit in Tuple_header

  ndb/src/kernel/blocks/dbtup/Dbtup.hpp
    1.59 05/04/24 17:38:12 pekka@stripped +7 -6
    tup scan, fixed MM: use Local_key + free bit in Tuple_header

# This is a BitKeeper patch.  What follows are the unified diffs for the
# set of deltas contained in the patch.  The rest of the patch, the part
# that BitKeeper cares about, is below these diffs.
# User:	pekka
# Host:	tuna.ndb.mysql.com
# Root:	/orca/space/pekka/ndb/version/my51-dd

--- 1.58/ndb/src/kernel/blocks/dbtup/Dbtup.hpp	2005-04-15 09:14:17 +02:00
+++ 1.59/ndb/src/kernel/blocks/dbtup/Dbtup.hpp	2005-04-24 17:38:12 +02:00
@@ -335,15 +335,14 @@
 
   // Position for use by scan
   struct PagePos {
-    Uint32 m_fragId;            // "base" fragment id
-    Uint32 m_pageId;
-    Uint32 m_tupleNo;
+    Uint32 m_fragId;            // fragment id
+    Local_key m_key;
     bool m_match;
   };
 
   // Tup scan op (compare Dbtux::ScanOp)
   struct ScanOp {
-    enum {
+    enum {      // state
       Undef = 0,
       First = 1,                // before first entry
       Locked = 4,               // at current entry (no lock needed)
@@ -351,8 +350,9 @@
       Last = 6,                 // after last entry
       Invalid = 9               // cannot return REF to LQH currently
     };
-    Uint16 m_state;
-    Uint16 m_lockwait;          // unused
+    Uint8 m_state;
+    Uint8 m_scan_mm;            // MM or DD
+    Uint8 m_scan_vs;            // varsize pages
     Uint32 m_userPtr;           // scanptr.i in LQH
     Uint32 m_userRef;
     Uint32 m_tableId;
@@ -1058,6 +1058,7 @@
     STATIC_CONST( ALLOC       = 0x100000 ); // Is record allocated now
     STATIC_CONST( MM_SHRINK   = 0x200000 ); // Has MM part shrunk
     STATIC_CONST( MM_GROWN    = 0x400000 ); // Has MM part grown
+    STATIC_CONST( FREE        = 0x800000 ); // On free list of page
     
     Uint32 get_tuple_version() const { 
       return m_header_bits & TUP_VERSION_MASK;

--- 1.12/ndb/src/kernel/blocks/dbtup/DbtupFixAlloc.cpp	2005-04-05 22:20:06 +02:00
+++ 1.13/ndb/src/kernel/blocks/dbtup/DbtupFixAlloc.cpp	2005-04-24 17:38:12 +02:00
@@ -134,6 +134,7 @@
   do {
     endOfList = nextTuple - Tupheadsize;
     regPagePtr->m_data[endOfList] = ctpConstant + nextTuple;
+    regPagePtr->m_data[endOfList + 1] = Tuple_header::FREE;
     nextTuple += Tupheadsize;
     cnt ++;
   } while (nextTuple <= Fix_page::DATA_WORDS);
@@ -148,6 +149,7 @@
 			     Fix_page* const regPagePtr)
 {
   Uint32 idx= regPagePtr->alloc_record();
+  regPagePtr->m_data[idx + 1] = 0; //m_header_bits
   if(regPagePtr->free_space == 0)
   {
     jam();
@@ -175,6 +177,7 @@
 			 Fix_page* regPagePtr)
 {
   Uint32 free= regPagePtr->free_record(key->m_page_idx);
+  regPagePtr->m_data[key->m_page_idx + 1] = Tuple_header::FREE;
   
   if(free == 1)
   {

--- 1.3/ndb/src/kernel/blocks/dbtup/DbtupScan.cpp	2005-04-12 16:32:07 +02:00
+++ 1.4/ndb/src/kernel/blocks/dbtup/DbtupScan.cpp	2005-04-24 17:38:12 +02:00
@@ -51,6 +51,12 @@
     new (scanPtr.p) ScanOp();
     ScanOp& scan = *scanPtr.p;
     scan.m_state = ScanOp::First;
+    // TODO scan disk only if any scanned attribute on disk
+    scan.m_scan_mm =
+      tablePtr.p->m_attributes[DD].m_no_of_fixsize > 0 ||
+      tablePtr.p->m_attributes[DD].m_no_of_varsize > 0;
+    scan.m_scan_vs =
+      tablePtr.p->m_attributes[scan.m_scan_mm].m_no_of_varsize > 0;
     scan.m_userPtr = req->senderData;
     scan.m_userRef = req->senderRef;
     scan.m_tableId = tablePtr.i;
@@ -157,18 +163,12 @@
   }
   if (scan.m_state == ScanOp::Locked) {
     jam();
-    Uint32 tupheadsize;
-    { TablerecPtr tablePtr;
-      tablePtr.i = scan.m_tableId;
-      ptrCheckGuard(tablePtr, cnoOfTablerec, tablerec);
-      tupheadsize = tablePtr.p->m_offsets[MM].m_fix_header_size;
-    }
     const PagePos& pos = scan.m_scanPos;
     NextScanConf* const conf = (NextScanConf*)signal->getDataPtrSend();
+    conf->scanPtr = scan.m_userPtr; // fix also in 5.0
     conf->accOperationPtr = (Uint32)-1; // no lock returned
     conf->fragId = frag.fragmentId;
-    conf->localKey[0] = (pos.m_pageId << MAX_TUPLES_BITS) |
-                        (pos.m_tupleNo * tupheadsize);
+    conf->localKey[0] = pos.m_key.ref();
     conf->localKey[1] = 0;
     conf->localKeyLength = 1;
     unsigned signalLength = 6;
@@ -201,8 +201,8 @@
   // set to first fragment, first page, first tuple
   PagePos& pos = scan.m_scanPos;
   pos.m_fragId = scan.m_fragId;
-  pos.m_pageId = 0;
-  pos.m_tupleNo = 0;
+  pos.m_key.m_page_no = 0;
+  pos.m_key.m_page_idx = 0;
   // just before
   pos.m_match = false;
   // let scanNext() do the work
@@ -215,71 +215,54 @@
 {
   ScanOp& scan = *scanPtr.p;
   PagePos& pos = scan.m_scanPos;
+  Local_key& key = pos.m_key;
   TablerecPtr tablePtr;
   tablePtr.i = scan.m_tableId;
   ptrCheckGuard(tablePtr, cnoOfTablerec, tablerec);
-
   FragrecordPtr fragPtr;
   fragPtr.i = scan.m_fragPtrI;
   ptrCheckGuard(fragPtr, cnoOfFragrec, fragrecord);
   Fragrecord& frag = *fragPtr.p;
-
   while (true) {
     // TODO time-slice here after X loops
     jam();
     // get fragment
     // get page
     PagePtr pagePtr;
-    if (pos.m_pageId >= frag.noOfPages) {
+    if (key.m_page_no >= frag.noOfPages) {
       jam();
       scan.m_state = ScanOp::Last;
       break;
     }
-    Uint32 realPageId = getRealpid(fragPtr.p, pos.m_pageId);
+    Uint32 realPageId = getRealpid(fragPtr.p, key.m_page_no);
     pagePtr.i = realPageId;
     ptrCheckGuard(pagePtr, cnoOfPage, cpage);
     const Uint32 pageState = pagePtr.p->page_state;
     if (pageState != ZTH_MM_FREE &&
         pageState != ZTH_MM_FULL) {
       jam();
-      pos.m_pageId++;
-      pos.m_tupleNo = 0;
+      key.m_page_no++;
+      key.m_page_idx = 0;
       pos.m_match = false;
       continue;
     }
     // get next tuple
+    const Uint32 tupheadsize = tablePtr.p->m_offsets[MM].m_fix_header_size;
     if (pos.m_match)
-      pos.m_tupleNo++;
+      key.m_page_idx += tupheadsize;
     pos.m_match = true;
-    const Uint32 tupheadsize = tablePtr.p->m_offsets[MM].m_fix_header_size;
-    Uint32 pageOffset = pos.m_tupleNo * tupheadsize;
-    if (pageOffset + tupheadsize > Fix_page::DATA_WORDS) {
+    if (key.m_page_idx + tupheadsize > Fix_page::DATA_WORDS) {
       jam();
-      pos.m_pageId++;
-      pos.m_tupleNo = 0;
+      key.m_page_no++;
+      key.m_page_idx = 0;
       pos.m_match = false;
       continue;
     }
     // skip over free tuple
-    bool isFree = false;
-    if (pageState == ZTH_MM_FREE) {
-      jam();
-      if ((pagePtr.p->m_data[pageOffset] >> 16) == 0x8000) {
-        Uint32 nextTuple = pagePtr.p->next_free_index >> 16;
-        while (nextTuple != 0xffff) {
-          jam();
-          if (nextTuple == pageOffset) {
-            jam();
-            isFree = true;
-            break;
-          }
-          nextTuple = pagePtr.p->m_data[nextTuple] & 0xffff;
-        }
-      }
-    }
-    if (isFree) {
-      jam();
-      continue;
+    const Tuple_header* th = (Tuple_header*)&pagePtr.p->m_data[key.m_page_idx];
+    if (th->m_header_bits & Tuple_header::FREE) {
+        jam();
+        continue;
     }
     // TODO check for operation and return latest in own tx
     scan.m_state = ScanOp::Locked;