bzr push into mysql-5.1-telco-6.2 branch (pekka:3022 to 3023) Bug#47832

[Pekka Nousiainen]

 3023 Pekka Nousiainen	2009-10-20
      bug#47832 01_lcp.diff
      Fix un-initialized scan position of LCP scan of empty fragment and
      make the scan state a bit more explicit.

    modified:
      storage/ndb/src/kernel/blocks/dbtup/Dbtup.hpp
      storage/ndb/src/kernel/blocks/dbtup/DbtupCommit.cpp
      storage/ndb/src/kernel/blocks/dbtup/DbtupScan.cpp
 3022 Jonas Oreland	2009-10-20
      ndb - fix newly introduced compiler warnings

    modified:
      storage/ndb/test/ndbapi/testBasic.cpp
=== modified file 'storage/ndb/src/kernel/blocks/dbtup/Dbtup.hpp'
--- a/storage/ndb/src/kernel/blocks/dbtup/Dbtup.hpp	2009-10-20 12:10:49 +0000
+++ b/storage/ndb/src/kernel/blocks/dbtup/Dbtup.hpp	2009-10-20 15:11:38 +0000
@@ -468,6 +468,13 @@ typedef Ptr<Fragoperrec> FragoperrecPtr;
     Local_key m_key_mm;         // MM local key returned
     Uint32 m_realpid_mm;        // MM real page id
     Uint32 m_extent_info_ptr_i;
+    ScanPos() {
+      /*
+       * Position is Null until scanFirst().  In particular in LCP scan
+       * it is Null between LCP_FRAG_ORD and ACC_SCANREQ.
+       */
+      m_key.setNull();
+    }
   };
 
   // Scan Lock
@@ -3203,6 +3210,8 @@ private:
 #endif
   
   void findFirstOp(OperationrecPtr&);
+  bool is_rowid_lcp_scanned(const Local_key& key1,
+                           const Dbtup::ScanOp& op);
   void commit_operation(Signal*, Uint32, Tuple_header*, PagePtr,
 			Operationrec*, Fragrecord*, Tablerec*);
   

=== modified file 'storage/ndb/src/kernel/blocks/dbtup/DbtupCommit.cpp'
--- a/storage/ndb/src/kernel/blocks/dbtup/DbtupCommit.cpp	2009-05-26 18:53:34 +0000
+++ b/storage/ndb/src/kernel/blocks/dbtup/DbtupCommit.cpp	2009-10-20 15:11:38 +0000
@@ -119,34 +119,39 @@ void Dbtup::initOpConnection(Operationre
   regOperPtr->m_undo_buffer_space= 0;
 }
 
-static
 bool
-operator>(const Local_key& key1, const Dbtup::ScanOp& op)
+Dbtup::is_rowid_lcp_scanned(const Local_key& key1,
+                            const Dbtup::ScanOp& op)
 {
   Local_key key2 = op.m_scanPos.m_key;
-  int gth0 = int(key1.m_page_no) - int(key2.m_page_no);
-  int gth1 = int(key1.m_page_idx) - int(key2.m_page_idx);
-  if (gth0 > 0 || (gth0 == 0 && gth1 > 0))
-  {
-    return true;
-  }
-  if (gth0 < 0 || (gth0 == 0 && gth1 < 0))
-  {
+  switch (op.m_state) {
+  case Dbtup::ScanOp::First:
+    ndbrequire(key2.isNull());
     return false;
-  }
-
-  /**
-   * key are equal...need to look at scan state
-   */
-  switch(op.m_state){
+  case Dbtup::ScanOp::Current:
   case Dbtup::ScanOp::Next:
+    ndbrequire(!key2.isNull());
+    if (key1.m_page_no < key2.m_page_no)
+      return true;
+    if (key1.m_page_no > key2.m_page_no)
+      return false;
+    if (key1.m_page_idx < key2.m_page_idx)
+      return true;
+    if (key1.m_page_idx > key2.m_page_idx)
+      return false;
     /**
-     * This row-id has already been scanned
+     * key are equal...need to look at scan state
      */
+    if (op.m_state == Dbtup::ScanOp::Next)
+      return true;
     return false;
-  default:
+  case Dbtup::ScanOp::Last:
     return true;
+  default:
+    ndbrequire(false);
+    break;
   }
+  return false;
 }
 
 void
@@ -183,7 +188,7 @@ Dbtup::dealloc_tuple(Signal* signal,
     c_scanOpPool.getPtr(scanOp, lcpScan_ptr_i);
     Local_key rowid = regOperPtr->m_tuple_location;
     rowid.m_page_no = page->frag_page_id;
-    if (rowid > *scanOp.p)
+    if (!is_rowid_lcp_scanned(rowid, *scanOp.p))
     {
       jam();
       extra_bits = Tuple_header::LCP_KEEP; // Note REMOVE FREE
@@ -368,7 +373,7 @@ Dbtup::commit_operation(Signal* signal,
     c_scanOpPool.getPtr(scanOp, lcpScan_ptr_i);
     Local_key rowid = regOperPtr->m_tuple_location;
     rowid.m_page_no = pagePtr.p->frag_page_id;
-    if (rowid > *scanOp.p)
+    if (!is_rowid_lcp_scanned(rowid, *scanOp.p))
     {
       jam();
        copy_bits |= Tuple_header::LCP_SKIP;

=== modified file 'storage/ndb/src/kernel/blocks/dbtup/DbtupScan.cpp'
--- a/storage/ndb/src/kernel/blocks/dbtup/DbtupScan.cpp	2009-10-06 13:10:55 +0000
+++ b/storage/ndb/src/kernel/blocks/dbtup/DbtupScan.cpp	2009-10-20 15:11:38 +0000
@@ -57,12 +57,13 @@ Dbtup::execACC_SCANREQ(Signal* signal)
     // flags
     Uint32 bits = 0;
     
-
     if (AccScanReq::getLcpScanFlag(req->requestInfo))
     {
       jam();
       bits |= ScanOp::SCAN_LCP;
       c_scanOpPool.getPtr(scanPtr, c_lcp_scan_op);
+      ndbrequire(scanPtr.p->m_fragPtrI == fragPtr.i);
+      ndbrequire(scanPtr.p->m_state == ScanOp::First);
     }
     else
     {
@@ -72,6 +73,7 @@ Dbtup::execACC_SCANREQ(Signal* signal)
 	jam();
 	break;
       }
+      new (scanPtr.p) ScanOp;
     }
 
     if (!AccScanReq::getNoDiskScanFlag(req->requestInfo)
@@ -123,7 +125,6 @@ Dbtup::execACC_SCANREQ(Signal* signal)
     }
     
     // set up scan op
-    new (scanPtr.p) ScanOp();
     ScanOp& scan = *scanPtr.p;
     scan.m_state = ScanOp::First;
     scan.m_bits = bits;
@@ -1254,8 +1255,7 @@ Dbtup::execLCP_FRAG_ORD(Signal* signal)
   ScanOpPtr scanPtr;
   c_scanOpPool.getPtr(scanPtr, frag.m_lcp_scan_op);
   ndbrequire(scanPtr.p->m_fragPtrI == RNIL);
+  new (scanPtr.p) ScanOp;
   scanPtr.p->m_fragPtrI = fragPtr.i;
-  
-  scanFirst(signal, scanPtr);
   scanPtr.p->m_state = ScanOp::First;
 }


Attachment: [text/bzr-bundle] bzr/[e-mail]