bk commit into 5.0 tree (mskold:1.2171) BUG#18184

[Martin Skold]

Below is the list of changes that have just been committed into a local
5.0 repository of marty. When marty 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.2171 06/06/12 09:37:19 mskold@stripped +4 -0
  Fix for Bug #18184  SELECT ... FOR UPDATE does not work..: Adapted to 5.0 code changes

  sql/ha_ndbcluster.cc
    1.253 06/06/12 09:37:05 mskold@stripped +81 -10
    Fix for Bug #18184  SELECT ... FOR UPDATE does not work..: Adapted to 5.0 code changes

  ndb/src/ndbapi/NdbScanOperation.cpp
    1.71 06/06/12 09:37:05 mskold@stripped +1 -1
    Fix for Bug #18184  SELECT ... FOR UPDATE does not work..: Adapted to 5.0 code changes

  ndb/include/ndbapi/NdbScanOperation.hpp
    1.35 06/06/12 09:37:05 mskold@stripped +30 -2
    Fix for Bug #18184  SELECT ... FOR UPDATE does not work..: Adapted to 5.0 code changes

  ndb/include/ndbapi/NdbIndexScanOperation.hpp
    1.20 06/06/12 09:37:05 mskold@stripped +5 -2
    Fix for Bug #18184  SELECT ... FOR UPDATE does not work..: Adapted to 5.0 code changes

# 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:	mskold
# Host:	linux.site
# Root:	/home/marty/MySQL/mysql-5.0

--- 1.19/ndb/include/ndbapi/NdbIndexScanOperation.hpp	2006-06-09 12:06:59 +02:00
+++ 1.20/ndb/include/ndbapi/NdbIndexScanOperation.hpp	2006-06-12 09:37:05 +02:00
@@ -61,11 +61,14 @@ public:
                         Uint32 parallel,
                         bool order_by,
                         bool order_desc = false,
-                        bool read_range_no = false) {
+                        bool read_range_no = false,
+			bool keyinfo = false) {
     Uint32 scan_flags =
       (SF_OrderBy & -(Int32)order_by) |
       (SF_Descending & -(Int32)order_desc) |
-      (SF_ReadRangeNo & -(Int32)read_range_no);
+      (SF_ReadRangeNo & -(Int32)read_range_no) | 
+      (SF_KeyInfo & -(Int32)keyinfo);
+    
     return readTuples(lock_mode, scan_flags, parallel);
   }
 #endif

--- 1.34/ndb/include/ndbapi/NdbScanOperation.hpp	2006-06-09 12:06:59 +02:00
+++ 1.35/ndb/include/ndbapi/NdbScanOperation.hpp	2006-06-12 09:37:05 +02:00
@@ -44,7 +44,8 @@ public:
     SF_TupScan = (1 << 16),     // scan TUP - only LM_CommittedRead
     SF_OrderBy = (1 << 24),     // index scan in order
     SF_Descending = (2 << 24),  // index scan in descending order
-    SF_ReadRangeNo = (4 << 24)  // enable @ref get_range_no
+    SF_ReadRangeNo = (4 << 24), // enable @ref get_range_no
+    SF_KeyInfo = 1              // request KeyInfo to be sent back
   };
 
   /**
@@ -68,7 +69,7 @@ public:
    */ 
 #ifdef ndb_readtuples_impossible_overload
   int readTuples(LockMode lock_mode = LM_Read, 
-		 Uint32 batch = 0, Uint32 parallel = 0);
+		 Uint32 batch = 0, Uint32 parallel = 0, bool keyinfo = false);
 #endif
   
   inline int readTuples(int parallell){
@@ -141,6 +142,20 @@ public:
   void close(bool forceSend = false, bool releaseOp = false);
 
   /**
+   * Lock current tuple
+   *
+   * @return an NdbOperation or NULL.
+   */
+  NdbOperation* lockCurrentTuple();
+  /**
+   * Lock current tuple
+   *
+   * @param lockTrans Transaction that should perform the lock
+   *
+   * @return an NdbOperation or NULL.
+   */
+  NdbOperation*	lockCurrentTuple(NdbTransaction* lockTrans);
+  /**
    * Update current tuple
    *
    * @return an NdbOperation or NULL.
@@ -247,6 +262,19 @@ protected:
   Uint32 m_read_range_no;
   NdbRecAttr *m_curr_row; // Pointer to last returned row
 };
+
+inline
+NdbOperation* 
+NdbScanOperation::lockCurrentTuple(){
+  return lockCurrentTuple(m_transConnection);
+}
+
+inline
+NdbOperation* 
+NdbScanOperation::lockCurrentTuple(NdbTransaction* takeOverTrans){
+  return takeOverScanOp(NdbOperation::ReadRequest, 
+			takeOverTrans);
+}
 
 inline
 NdbOperation* 

--- 1.70/ndb/src/ndbapi/NdbScanOperation.cpp	2006-06-09 12:06:59 +02:00
+++ 1.71/ndb/src/ndbapi/NdbScanOperation.cpp	2006-06-12 09:37:05 +02:00
@@ -160,7 +160,7 @@ NdbScanOperation::readTuples(NdbScanOper
     return -1;
   }
 
-  m_keyInfo = (keyinfo || lockExcl) ? 1 : 0;
+  m_keyInfo = ((scan_flags & SF_KeyInfo) || lockExcl) ? 1 : 0;
 
   bool rangeScan = false;
   if (m_accessTable->m_indexType == NdbDictionary::Index::OrderedIndex)

--- 1.252/sql/ha_ndbcluster.cc	2006-06-09 12:06:59 +02:00
+++ 1.253/sql/ha_ndbcluster.cc	2006-06-12 09:37:05 +02:00
@@ -1174,12 +1174,23 @@ void ha_ndbcluster::release_metadata()
 
 int ha_ndbcluster::get_ndb_lock_type(enum thr_lock_type type)
 {
+  DBUG_ENTER("ha_ndbcluster::get_ndb_lock_type");
   if (type >= TL_WRITE_ALLOW_WRITE)
-    return NdbOperation::LM_Exclusive;
-  else if (uses_blob_value(m_retrieve_all_fields))
-    return NdbOperation::LM_Read;
+  {
+    DBUG_PRINT("info", ("Using exclusive lock"));
+    DBUG_RETURN(NdbOperation::LM_Exclusive);
+  }
+  else if (type ==  TL_READ_WITH_SHARED_LOCKS ||
+	   uses_blob_value(m_retrieve_all_fields))
+  {
+    DBUG_PRINT("info", ("Using read lock"));
+    DBUG_RETURN(NdbOperation::LM_Read);
+  }
   else
-    return NdbOperation::LM_CommittedRead;
+  {
+    DBUG_PRINT("info", ("Using committed read"));
+    DBUG_RETURN(NdbOperation::LM_CommittedRead);
+  }
 }
 
 static const ulong index_type_flags[]=
@@ -1679,7 +1690,30 @@ inline int ha_ndbcluster::fetch_next(Ndb
   int check;
   NdbTransaction *trans= m_active_trans;
   
-  bool contact_ndb= m_lock.type < TL_WRITE_ALLOW_WRITE;
+    if (m_lock_tuple)
+  {
+    /*
+      Lock level m_lock.type either TL_WRITE_ALLOW_WRITE
+      (SELECT FOR UPDATE) or TL_READ_WITH_SHARED_LOCKS (SELECT
+      LOCK WITH SHARE MODE) and row was not explictly unlocked 
+      with unlock_row() call
+    */
+      NdbConnection *trans= m_active_trans;
+      NdbOperation *op;
+      // Lock row
+      DBUG_PRINT("info", ("Keeping lock on scanned row"));
+      
+      if (!(op= m_active_cursor->lockCurrentTuple()))
+      {
+	m_lock_tuple= false;
+	ERR_RETURN(trans->getNdbError());
+      }
+      m_ops_pending++;
+  }
+  m_lock_tuple= false;
+
+  bool contact_ndb= m_lock.type < TL_WRITE_ALLOW_WRITE &&
+                    m_lock.type != TL_READ_WITH_SHARED_LOCKS;
   do {
     DBUG_PRINT("info", ("Call nextResult, contact_ndb: %d", contact_ndb));
     /*
@@ -1695,6 +1729,13 @@ inline int ha_ndbcluster::fetch_next(Ndb
     
     if ((check= cursor->nextResult(contact_ndb, m_force_send)) == 0)
     {
+      /*
+	Explicitly lock tuple if "select for update" or
+	"select lock in share mode"
+      */
+      m_lock_tuple= (m_lock.type == TL_WRITE_ALLOW_WRITE
+		     || 
+		     m_lock.type == TL_READ_WITH_SHARED_LOCKS);
       DBUG_RETURN(0);
     } 
     else if (check == 1 || check == 2)
@@ -1983,10 +2024,11 @@ int ha_ndbcluster::ordered_index_scan(co
     restart= FALSE;
     NdbOperation::LockMode lm=
       (NdbOperation::LockMode)get_ndb_lock_type(m_lock.type);
+    bool need_pk = (lm == NdbOperation::LM_Read);
     if (!(op= trans->getNdbIndexScanOperation((NDBINDEX *)
                                               m_index[active_index].index, 
                                               (const NDBTAB *) m_table)) ||
-        op->readTuples(lm, 0, parallelism, sorted, descending))
+        op->readTuples(lm, 0, parallelism, sorted, descending, need_pk))
       ERR_RETURN(trans->getNdbError());
     m_active_cursor= op;
   } else {
@@ -2036,8 +2078,11 @@ int ha_ndbcluster::full_table_scan(byte 
 
   NdbOperation::LockMode lm=
     (NdbOperation::LockMode)get_ndb_lock_type(m_lock.type);
+  bool need_pk = (lm == NdbOperation::LM_Read);
   if (!(op=trans->getNdbScanOperation((const NDBTAB *) m_table)) ||
-      op->readTuples(lm, 0, parallelism))
+      op->readTuples(lm, 
+		     (need_pk)?NdbScanOperation::SF_KeyInfo:0, 
+		     parallelism))
     ERR_RETURN(trans->getNdbError());
   m_active_cursor= op;
   if (generate_scan_filter(m_cond_stack, op))
@@ -2327,6 +2372,7 @@ int ha_ndbcluster::update_row(const byte
     DBUG_PRINT("info", ("Calling updateTuple on cursor"));
     if (!(op= cursor->updateCurrentTuple()))
       ERR_RETURN(trans->getNdbError());
+    m_lock_tuple= false;
     m_ops_pending++;
     if (uses_blob_value(FALSE))
       m_blobs_pending= TRUE;
@@ -2406,6 +2452,7 @@ int ha_ndbcluster::delete_row(const byte
     DBUG_PRINT("info", ("Calling deleteTuple on cursor"));
     if (cursor->deleteCurrentTuple() != 0)
       ERR_RETURN(trans->getNdbError());     
+    m_lock_tuple= false;
     m_ops_pending++;
 
     no_uncommitted_rows_update(-1);
@@ -2529,9 +2576,9 @@ void ha_ndbcluster::unpack_record(byte* 
     const NdbRecAttr* rec= m_value[hidden_no].rec;
     DBUG_ASSERT(rec);
     DBUG_PRINT("hidden", ("%d: %s \"%llu\"", hidden_no, 
-                          hidden_col->getName(), rec->u_64_value()));
+			  hidden_col->getName(), rec->u_64_value()));
   } 
-  //print_results();
+  print_results();
 #endif
   DBUG_VOID_RETURN;
 }
@@ -2605,6 +2652,12 @@ int ha_ndbcluster::index_init(uint index
 {
   DBUG_ENTER("ha_ndbcluster::index_init");
   DBUG_PRINT("enter", ("index: %u", index));
+ /*
+    Locks are are explicitly released in scan
+    unless m_lock.type == TL_READ_HIGH_PRIORITY
+    and no sub-sequent call to unlock_row()
+   */
+  m_lock_tuple= false;
   DBUG_RETURN(handler::index_init(index));
 }
 
@@ -3614,6 +3667,22 @@ int ha_ndbcluster::external_lock(THD *th
 }
 
 /*
+  Unlock the last row read in an open scan.
+  Rows are unlocked by default in ndb, but
+  for SELECT FOR UPDATE and SELECT LOCK WIT SHARE MODE
+  locks are kept if unlock_row() is not called.
+*/
+
+void ha_ndbcluster::unlock_row() 
+{
+  DBUG_ENTER("unlock_row");
+
+  DBUG_PRINT("info", ("Unlocking row"));
+  m_lock_tuple= false;
+  DBUG_VOID_RETURN;
+}
+
+/*
   Start a transaction for running a statement if one is not
   already running in a transaction. This will be the case in
   a BEGIN; COMMIT; block
@@ -5897,6 +5966,7 @@ ha_ndbcluster::read_multi_range_first(KE
   byte *end_of_buffer= (byte*)buffer->buffer_end;
   NdbOperation::LockMode lm= 
     (NdbOperation::LockMode)get_ndb_lock_type(m_lock.type);
+  bool need_pk = (lm == NdbOperation::LM_Read);
   const NDBTAB *tab= (const NDBTAB *) m_table;
   const NDBINDEX *unique_idx= (NDBINDEX *) m_index[active_index].unique_index;
   const NDBINDEX *idx= (NDBINDEX *) m_index[active_index].index; 
@@ -5963,7 +6033,8 @@ ha_ndbcluster::read_multi_range_first(KE
           end_of_buffer -= reclength;
         }
         else if ((scanOp= m_active_trans->getNdbIndexScanOperation(idx, tab)) 
-                 &&!scanOp->readTuples(lm, 0, parallelism, sorted, FALSE,
TRUE)
+                 &&!scanOp->readTuples(lm, 0, parallelism, sorted, 
+				       FALSE, TRUE, need_pk)
                  &&!generate_scan_filter(m_cond_stack, scanOp)
                  &&!define_read_attrs(end_of_buffer-reclength, scanOp))
         {