bk commit into 5.1 tree (jonas:1.2371) BUG#19929

[jonas]

Below is the list of changes that have just been committed into a local
5.1 repository of jonas. When jonas 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.2371 06/05/19 09:38:59 jonas@stripped +1 -0
  ndb -
    bug#19928 and bug#19929
    fix to critical bugs in tup scan
    that affected lcp,backup and opt. nr

  storage/ndb/src/kernel/blocks/dbtup/DbtupScan.cpp
    1.13 06/05/19 09:38:56 jonas@stripped +24 -16
    1) dont let dirty read scan find uncommitted inserts
    2) force opt. nr scan to wait for locked rows
    3) when finding LCP keep record, use accOpPtr -1, so that it will not be committed
towards ACC

# 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:	jonas
# Host:	perch.ndb.mysql.com
# Root:	/home/jonas/src/51-jonas

--- 1.12/storage/ndb/src/kernel/blocks/dbtup/DbtupScan.cpp	2006-03-13 13:37:39 +01:00
+++ 1.13/storage/ndb/src/kernel/blocks/dbtup/DbtupScan.cpp	2006-05-19 09:38:56 +02:00
@@ -582,12 +582,15 @@
   Fragrecord& frag = *fragPtr.p;
   // tuple found
   Tuple_header* th = 0;
+  Uint32 thbits = 0;
   Uint32 loop_count = 0;
   Uint32 scanGCI = scanPtr.p->m_scanGCI;
   Uint32 foundGCI;
  
-  bool mm = (bits & ScanOp::SCAN_DD);
-  bool lcp = (bits & ScanOp::SCAN_LCP);
+  const bool mm = (bits & ScanOp::SCAN_DD);
+  const bool lcp = (bits & ScanOp::SCAN_LCP);
+  const bool dirty = (bits & ScanOp::SCAN_LOCK) == 0;
+  
   Uint32 lcp_list = fragPtr.p->m_lcp_keep_list;
   Uint32 size = table.m_offsets[mm].m_fix_header_size +
     (bits & ScanOp::SCAN_VS ? Tuple_header::HeaderSize + 1: 0);
@@ -750,22 +753,22 @@
 	{
 	  pos.m_get = ScanPos::Get_next_tuple_fs;
           th = (Tuple_header*)&page->m_data[key.m_page_idx];
+	  thbits = th->m_header_bits;
+	  
 	  if (likely(! (bits & ScanOp::SCAN_NR)))
 	  {
-	    if (! (th->m_header_bits & Tuple_header::FREE)) {
-	      goto found_tuple;
-	    } 
-	    else 
+	    jam();
+	    if (! (thbits & Tuple_header::FREE))
 	    {
-	      jam();
-	      // skip free tuple
-	    }
+	      if (! ((thbits & Tuple_header::ALLOC) && dirty))
+		goto found_tuple;
+	    } 
 	  }
 	  else
 	  {
 	    if ((foundGCI = *th->get_mm_gci(tablePtr.p)) > scanGCI)
 	    {
-	      if (! (th->m_header_bits & Tuple_header::FREE))
+	      if (! (thbits & Tuple_header::FREE))
 	      {
 		jam();
 		goto found_tuple;
@@ -775,9 +778,11 @@
 		goto found_deleted_rowid;
 	      }
 	    }
-	    else
+	    else if (thbits != Fix_page::FREE_RECORD && 
+		     th->m_operation_ptr_i != RNIL)
 	    {
 	      jam();
+	      goto found_tuple; // Locked tuple...
 	      // skip free tuple
 	    }
 	  }
@@ -793,8 +798,7 @@
       jam();
       {
         // caller has already set pos.m_get to next tuple
-        if (! (bits & ScanOp::SCAN_LCP && 
-           th->m_header_bits & Tuple_header::LCP_SKIP)) {
+        if (! (bits & ScanOp::SCAN_LCP && thbits &
Tuple_header::LCP_SKIP)) {
           Local_key& key_mm = pos.m_key_mm;
           if (! (bits & ScanOp::SCAN_DD)) {
             key_mm = pos.m_key;
@@ -810,7 +814,11 @@
         } else {
           jam();
           // clear it so that it will show up in next LCP
-          th->m_header_bits &= ~(Uint32)Tuple_header::LCP_SKIP;
+          th->m_header_bits = thbits & ~(Uint32)Tuple_header::LCP_SKIP;
+	  if (tablePtr.p->m_bits & Tablerec::TR_Checksum) {
+	    jam();
+	    setChecksum(th, tablePtr.p);
+	  }
         }
       }
       break;
@@ -833,7 +841,7 @@
 	  th = (Tuple_header*)(mmpage->m_data + key_mm.m_page_idx);
 	  if ((foundGCI = *th->get_mm_gci(tablePtr.p)) > scanGCI)
 	  {
-	    if (! (th->m_header_bits & Tuple_header::FREE))
+	    if (! (thbits & Tuple_header::FREE))
 	      break;
 	  }
 	}
@@ -893,7 +901,7 @@
   
   NextScanConf* const conf = (NextScanConf*)signal->getDataPtrSend();
   conf->scanPtr = scan.m_userPtr;
-  conf->accOperationPtr = RNIL + 1;
+  conf->accOperationPtr = (Uint32)-1;
   conf->fragId = frag.fragmentId;
   conf->localKey[0] = lcp_list;
   conf->localKey[1] = 0;