bk commit into 5.1 tree (mskold:1.2280)

[Martin Skold]

Below is the list of changes that have just been committed into a local
5.1 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@stripped, 2006-08-15 14:39:48+02:00, mskold@stripped +5 -0
  Merge mysql.com:/windows/Linux_space/MySQL/mysql-5.0
  into  mysql.com:/windows/Linux_space/MySQL/mysql-5.1
  MERGE: 1.1810.1977.2

  mysql-test/r/ndb_lock.result@stripped, 2006-08-15 14:34:52+02:00, mskold@stripped +0 -12
    Merge (using local)
    MERGE: 1.5.5.1

  mysql-test/t/ndb_lock.test@stripped, 2006-08-15 14:34:41+02:00, mskold@stripped +0 -13
    Auto merged
    MERGE: 1.7.5.1

  sql/ha_ndbcluster.cc@stripped, 2006-08-15 14:39:46+02:00, mskold@stripped +2 -4
    Merge
    MERGE: 1.175.45.2

  sql/ha_ndbcluster.h@stripped, 2006-08-15 14:39:46+02:00, mskold@stripped +0 -0
    Merge
    MERGE: 1.82.3.2

  storage/ndb/include/ndbapi/NdbTransaction.hpp@stripped, 2006-08-15 14:34:41+02:00,
mskold@stripped +0 -0
    Auto merged
    MERGE: 1.41.3.2

  storage/ndb/include/ndbapi/NdbTransaction.hpp@stripped, 2006-08-15 14:34:41+02:00,
mskold@stripped +0 -0
    Merge rename: ndb/include/ndbapi/NdbTransaction.hpp ->
storage/ndb/include/ndbapi/NdbTransaction.hpp

# 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:	/windows/Linux_space/MySQL/mysql-5.1/RESYNC

--- 1.41.3.1/ndb/include/ndbapi/NdbTransaction.hpp	2006-08-15 14:39:55 +02:00
+++ 1.48/storage/ndb/include/ndbapi/NdbTransaction.hpp	2006-08-15 14:39:55 +02:00
@@ -658,7 +658,7 @@ private:						
   // Release all cursor operations in connection
   void releaseOps(NdbOperation*);	
   void releaseScanOperations(NdbIndexScanOperation*);	
-  void releaseExecutedScanOperation(NdbIndexScanOperation*);
+  void releaseScanOperation(NdbIndexScanOperation*);
 
   // Set the transaction identity of the transaction
   void		setTransactionId(Uint64 aTransactionId);
@@ -794,7 +794,7 @@ private:						
   Uint8 thePendingBlobOps;
   inline bool hasBlobOperation() { return theBlobFlag; }
 
-  static void sendTC_COMMIT_ACK(NdbApiSignal *,
+  static void sendTC_COMMIT_ACK(class TransporterFacade *, NdbApiSignal *,
 				Uint32 transId1, Uint32 transId2, 
 				Uint32 aBlockRef);
 

--- 1.348/sql/ha_ndbcluster.cc	2006-08-15 14:39:55 +02:00
+++ 1.349/sql/ha_ndbcluster.cc	2006-08-15 14:39:55 +02:00
@@ -256,13 +256,15 @@ int execute_no_commit_ignore_no_key(ha_n
 }
 
 inline
-int execute_no_commit(ha_ndbcluster *h, NdbTransaction *trans)
+int execute_no_commit(ha_ndbcluster *h, NdbTransaction *trans,
+		      bool force_release)
 {
 #ifdef NOT_USED
   int m_batch_execute= 0;
   if (m_batch_execute)
     return 0;
 #endif
+  h->release_completed_operations(trans, force_release);
   return h->m_ignore_no_key ?
     execute_no_commit_ignore_no_key(h,trans) :
     trans->execute(NdbTransaction::NoCommit,
@@ -297,13 +299,15 @@ int execute_commit(THD *thd, NdbTransact
 }
 
 inline
-int execute_no_commit_ie(ha_ndbcluster *h, NdbTransaction *trans)
+int execute_no_commit_ie(ha_ndbcluster *h, NdbTransaction *trans,
+			 bool force_release)
 {
 #ifdef NOT_USED
   int m_batch_execute= 0;
   if (m_batch_execute)
     return 0;
 #endif
+  h->release_completed_operations(trans, force_release);
   return trans->execute(NdbTransaction::NoCommit,
                         NdbTransaction::AO_IgnoreError,
                         h->m_force_send);
@@ -328,6 +332,7 @@ Thd_ndb::Thd_ndb()
   all= NULL;
   stmt= NULL;
   error= 0;
+  query_state&= NDB_QUERY_NORMAL;
   options= 0;
   (void) hash_init(&open_tables, &my_charset_bin, 5, 0, 0,
                    (hash_get_key)thd_ndb_share_get_key, 0, 0);
@@ -1696,7 +1701,7 @@ int ha_ndbcluster::pk_read(const byte *k
       ERR_RETURN(trans->getNdbError());
   }
 
-  if (execute_no_commit_ie(this,trans) != 0) 
+  if (execute_no_commit_ie(this,trans,false) != 0) 
   {
     table->status= STATUS_NOT_FOUND;
     DBUG_RETURN(ndb_err(trans));
@@ -1761,7 +1766,7 @@ int ha_ndbcluster::complemented_read(con
     }
   }
   
-  if (execute_no_commit(this,trans) != 0) 
+  if (execute_no_commit(this,trans,false) != 0) 
   {
     table->status= STATUS_NOT_FOUND;
     DBUG_RETURN(ndb_err(trans));
@@ -1914,7 +1919,7 @@ int ha_ndbcluster::peek_indexed_rows(con
   }
   last= trans->getLastDefinedOperation();
   if (first)
-    res= execute_no_commit_ie(this,trans);
+    res= execute_no_commit_ie(this,trans,false);
   else
   {
     // Table has no keys
@@ -1963,7 +1968,7 @@ int ha_ndbcluster::unique_index_read(con
   if ((res= define_read_attrs(buf, op)))
     DBUG_RETURN(res);
 
-  if (execute_no_commit_ie(this,trans) != 0) 
+  if (execute_no_commit_ie(this,trans,false) != 0) 
   {
     table->status= STATUS_NOT_FOUND;
     DBUG_RETURN(ndb_err(trans));
@@ -2011,7 +2016,7 @@ inline int ha_ndbcluster::fetch_next(Ndb
     */
     if (m_ops_pending && m_blobs_pending)
     {
-      if (execute_no_commit(this,trans) != 0)
+      if (execute_no_commit(this,trans,false) != 0)
         DBUG_RETURN(ndb_err(trans));
       m_ops_pending= 0;
       m_blobs_pending= FALSE;
@@ -2043,7 +2048,7 @@ inline int ha_ndbcluster::fetch_next(Ndb
       {
         if (m_transaction_on)
         {
-          if (execute_no_commit(this,trans) != 0)
+          if (execute_no_commit(this,trans,false) != 0)
             DBUG_RETURN(-1);
         }
         else
@@ -2370,7 +2375,7 @@ int ha_ndbcluster::ordered_index_scan(co
       ERR_RETURN(trans->getNdbError());
   }
 
-  if (execute_no_commit(this,trans) != 0)
+  if (execute_no_commit(this,trans,false) != 0)
     DBUG_RETURN(ndb_err(trans));
   
   DBUG_RETURN(next_result(buf));
@@ -2440,7 +2445,7 @@ int ha_ndbcluster::full_table_scan(byte 
   if ((res= define_read_attrs(buf, op)))
     DBUG_RETURN(res);
 
-  if (execute_no_commit(this,trans) != 0)
+  if (execute_no_commit(this,trans,false) != 0)
     DBUG_RETURN(ndb_err(trans));
   DBUG_PRINT("exit", ("Scan started successfully"));
   DBUG_RETURN(next_result(buf));
@@ -2603,7 +2608,7 @@ int ha_ndbcluster::write_row(byte *recor
     m_bulk_insert_not_flushed= FALSE;
     if (m_transaction_on)
     {
-      if (execute_no_commit(this,trans) != 0)
+      if (execute_no_commit(this,trans,false) != 0)
       {
         m_skip_auto_increment= TRUE;
         no_uncommitted_rows_execute_failure();
@@ -2840,7 +2845,7 @@ int ha_ndbcluster::update_row(const byte
     op->setValue(no_fields, part_func_value);
   }
   // Execute update operation
-  if (!cursor && execute_no_commit(this,trans) != 0) {
+  if (!cursor && execute_no_commit(this,trans,false) != 0) {
     no_uncommitted_rows_execute_failure();
     DBUG_RETURN(ndb_err(trans));
   }
@@ -2926,7 +2931,7 @@ int ha_ndbcluster::delete_row(const byte
   }
 
   // Execute delete operation
-  if (execute_no_commit(this,trans) != 0) {
+  if (execute_no_commit(this,trans,false) != 0) {
     no_uncommitted_rows_execute_failure();
     DBUG_RETURN(ndb_err(trans));
   }
@@ -3392,6 +3397,26 @@ int ha_ndbcluster::close_scan()
 
   NdbScanOperation *cursor= m_active_cursor ? m_active_cursor : m_multi_cursor;
   
+  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
+    */
+      NdbOperation *op;
+      // Lock row
+      DBUG_PRINT("info", ("Keeping lock on scanned row"));
+      
+      if (!(op= cursor->lockCurrentTuple()))
+      {
+	m_lock_tuple= false;
+	ERR_RETURN(trans->getNdbError());
+      }
+      m_ops_pending++;      
+  }
+  m_lock_tuple= false;
   if (m_ops_pending)
   {
     /*
@@ -3399,7 +3424,7 @@ int ha_ndbcluster::close_scan()
       deleteing/updating transaction before closing the scan    
     */
     DBUG_PRINT("info", ("ops_pending: %d", m_ops_pending));    
-    if (execute_no_commit(this,trans) != 0) {
+    if (execute_no_commit(this,trans,false) != 0) {
       no_uncommitted_rows_execute_failure();
       DBUG_RETURN(ndb_err(trans));
     }
@@ -3793,7 +3818,7 @@ int ha_ndbcluster::end_bulk_insert()
     m_bulk_insert_not_flushed= FALSE;
     if (m_transaction_on)
     {
-      if (execute_no_commit(this, trans) != 0)
+      if (execute_no_commit(this, trans,false) != 0)
       {
         no_uncommitted_rows_execute_failure();
         my_errno= error= ndb_err(trans);
@@ -3968,6 +3993,7 @@ int ha_ndbcluster::external_lock(THD *th
           ERR_RETURN(ndb->getNdbError());
         thd_ndb->init_open_tables();
         thd_ndb->stmt= trans;
+	thd_ndb->query_state&= NDB_QUERY_NORMAL;
         trans_register_ha(thd, FALSE, &ndbcluster_hton);
       } 
       else 
@@ -3983,6 +4009,7 @@ int ha_ndbcluster::external_lock(THD *th
             ERR_RETURN(ndb->getNdbError());
           thd_ndb->init_open_tables();
           thd_ndb->all= trans; 
+	  thd_ndb->query_state&= NDB_QUERY_NORMAL;
           trans_register_ha(thd, TRUE, &ndbcluster_hton);
 
           /*
@@ -4139,6 +4166,7 @@ int ha_ndbcluster::start_stmt(THD *thd, 
     thd_ndb->stmt= trans;
     trans_register_ha(thd, FALSE, &ndbcluster_hton);
   }
+  thd_ndb->query_state&= NDB_QUERY_NORMAL;
   m_active_trans= trans;
 
   // Start of statement
@@ -7557,6 +7585,30 @@ int ha_ndbcluster::write_ndb_file(const 
   DBUG_RETURN(error);
 }
 
+void 
+ha_ndbcluster::release_completed_operations(NdbTransaction *trans,
+					    bool force_release)
+{
+  if (trans->hasBlobOperation())
+  {
+    /* We are reading/writing BLOB fields, 
+       releasing operation records is unsafe
+    */
+    return;
+  }
+  if (!force_release)
+  {
+    if (get_thd_ndb(current_thd)->query_state & NDB_QUERY_MULTI_READ_RANGE)
+    {
+      /* We are batching reads and have not consumed all fetched
+	 rows yet, releasing operation records is unsafe 
+      */
+      return;
+    }
+  }
+  trans->releaseCompletedOperations();
+}
+
 int
 ha_ndbcluster::read_multi_range_first(KEY_MULTI_RANGE **found_range_p,
                                       KEY_MULTI_RANGE *ranges, 
@@ -7572,6 +7624,7 @@ ha_ndbcluster::read_multi_range_first(KE
   NDB_INDEX_TYPE index_type= get_index_type(active_index);
   ulong reclength= table_share->reclength;
   NdbOperation* op;
+  Thd_ndb *thd_ndb= get_thd_ndb(current_thd);
 
   if (uses_blob_value())
   {
@@ -7585,7 +7638,7 @@ ha_ndbcluster::read_multi_range_first(KE
                                                 sorted, 
                                                 buffer));
   }
-
+  thd_ndb->query_state|= NDB_QUERY_MULTI_READ_RANGE;
   m_disable_multi_read= FALSE;
 
   /**
@@ -7757,7 +7810,7 @@ ha_ndbcluster::read_multi_range_first(KE
    */
   m_current_multi_operation= 
     lastOp ? lastOp->next() : m_active_trans->getFirstDefinedOperation();
-  if (!(res= execute_no_commit_ie(this, m_active_trans)))
+  if (!(res= execute_no_commit_ie(this, m_active_trans, true)))
   {
     m_multi_range_defined= multi_range_curr;
     multi_range_curr= ranges;

--- 1.145/sql/ha_ndbcluster.h	2006-08-15 14:39:55 +02:00
+++ 1.146/sql/ha_ndbcluster.h	2006-08-15 14:39:55 +02:00
@@ -534,6 +534,11 @@ class Ndb_cond_traverse_context 
   Ndb_rewrite_context *rewrite_stack;
 };
 
+typedef enum ndb_query_state_bits {
+  NDB_QUERY_NORMAL = 0,
+  NDB_QUERY_MULTI_READ_RANGE = 1
+} NDB_QUERY_STATE_BITS;
+
 /*
   Place holder for ha_ndbcluster thread specific data
 */
@@ -571,6 +576,7 @@ class Thd_ndb 
   int error;
   uint32 options;
   List<NDB_SHARE> changed_tables;
+  uint query_state;
   HASH open_tables;
 };
 
@@ -849,8 +855,8 @@ private:
 
   friend int execute_commit(ha_ndbcluster*, NdbTransaction*);
   friend int execute_no_commit_ignore_no_key(ha_ndbcluster*, NdbTransaction*);
-  friend int execute_no_commit(ha_ndbcluster*, NdbTransaction*);
-  friend int execute_no_commit_ie(ha_ndbcluster*, NdbTransaction*);
+  friend int execute_no_commit(ha_ndbcluster*, NdbTransaction*, bool);
+  friend int execute_no_commit_ie(ha_ndbcluster*, NdbTransaction*, bool);
 
   NdbTransaction *m_active_trans;
   NdbScanOperation *m_active_cursor;
@@ -898,6 +904,8 @@ private:
   bool m_force_send;
   ha_rows m_autoincrement_prefetch;
   bool m_transaction_on;
+  void release_completed_operations(NdbTransaction*, bool);
+
   Ndb_cond_stack *m_cond_stack;
   bool m_disable_multi_read;
   byte *m_multi_range_result_ptr;