Merge mysql.com:/home/marty/MySQL/mysql-5.0

into  mysql.com:/home/marty/MySQL/mysql-5.1

mysql-test/r/ndb_truncate.result

-DROP TABLE IF EXISTS t2;
-CREATE TABLE t2 (
-a bigint unsigned NOT NULL PRIMARY KEY,
+DROP TABLE IF EXISTS t1, t2;
+CREATE TABLE t1 (
+a bigint unsigned NOT NULL AUTO_INCREMENT PRIMARY KEY,
 b int unsigned not null,
 c int unsigned
 ) engine=ndbcluster;
-select count(*) from t2;
+select count(*) from t1;
 count(*)
 5000
-truncate table t2;
-select count(*) from t2;
+select * from t1 order by a limit 2;
+a	b	c
+1	509	2500
+2	510	7
+truncate table t1;
+select count(*) from t1;
 count(*)
 0
-drop table t2;
+insert into t1 values(NULL,1,1),(NULL,2,2);
+select * from t1 order by a;
+a	b	c
+1	1	1
+2	2	2
+drop table t1;

mysql-test/t/ndb_lock.test

@@ -69,6 +69,119 @@ insert into t1 values (1,1,1);
 
 drop table t1;
 
+# Lock for update
+
+create table t1 (x integer not null primary key, y varchar(32), z integer, key(z)) engine = ndb;
+
+insert into t1 values (1,'one',1), (2,'two',2),(3,"three",3); 
+
+# PK access
+connection con1;
+begin;
+select * from t1 where x = 1 for update;
+
+connection con2;
+begin;
+select * from t1 where x = 2 for update;
+--error 1205
+select * from t1 where x = 1 for update;
+rollback;
+
+connection con1;
+commit;
+
+# table scan
+connection con1;
+begin;
+select * from t1 where y = 'one' or y = 'three' for update;
+
+connection con2;
+begin;
+# Have to check with pk access here since scans take locks on
+# all rows and then release them in chunks
+# Bug #20390 SELECT FOR UPDATE does not release locks of untouched rows in full table scans
+#select * from t1 where x = 2 for update;
+--error 1205
+select * from t1 where x = 1 for update;
+rollback;
+
+connection con1;
+commit;
+
+# index scan
+connection con1;
+begin;
+select * from t1 where z > 1 and z < 3 for update;
+
+connection con2;
+begin;
+# Have to check with pk access here since scans take locks on
+# all rows and then release them in chunks
+select * from t1 where x = 1 for update;
+--error 1205
+select * from t1 where x = 2 for update;
+rollback;
+
+connection con1;
+commit;
+
+# share locking
+
+# PK access
+connection con1;
+begin;
+select * from t1 where x = 1 lock in share mode;
+
+connection con2;
+begin;
+select * from t1 where x = 1 lock in share mode;
+select * from t1 where x = 2 for update;
+--error 1205
+select * from t1 where x = 1 for update;
+rollback;
+
+connection con1;
+commit;
+
+# table scan
+connection con1;
+begin;
+select * from t1 where y = 'one' or y = 'three' lock in share mode;
+
+connection con2;
+begin;
+select * from t1 where y = 'one' lock in share mode;
+# Have to check with pk access here since scans take locks on
+# all rows and then release them in chunks
+# Bug #20390 SELECT FOR UPDATE does not release locks of untouched rows in full table scans
+#select * from t1 where x = 2 for update;
+--error 1205
+select * from t1 where x = 1 for update;
+rollback;
+
+connection con1;
+commit;
+
+# index scan
+connection con1;
+begin;
+select * from t1 where z > 1 and z < 3 lock in share mode;
+
+connection con2;
+begin;
+select * from t1 where z = 1 lock in share mode;
+# Have to check with pk access here since scans take locks on
+# all rows and then release them in chunks
+select * from t1 where x = 1 for update;
+--error 1205
+select * from t1 where x = 2 for update;
+rollback;
+
+connection con1;
+commit;
+
+drop table t1;
+
 # End of 4.1 tests
 
 #

mysql-test/t/ndb_truncate.test

@@ -2,12 +2,11 @@
 -- source include/not_embedded.inc
 
 --disable_warnings
-DROP TABLE IF EXISTS t2;
+DROP TABLE IF EXISTS t1, t2;
 --enable_warnings
 
-
-CREATE TABLE t2 (
-  a bigint unsigned NOT NULL PRIMARY KEY,
+CREATE TABLE t1 (
+  a bigint unsigned NOT NULL AUTO_INCREMENT PRIMARY KEY,
   b int unsigned not null,
   c int unsigned
 ) engine=ndbcluster;
@@ -20,17 +19,23 @@ let $1=500;
 disable_query_log;
 while ($1)
 {
- eval insert into t2 values($1*10, $1+9, 5*$1), ($1*10+1, $1+10, 7),($1*10+2, $1+10, 7*$1), ($1*10+3, $1+10, 10+$1), ($1*10+4, $1+10, 70*$1), ($1*10+5, $1+10, 7), ($1*10+6, $1+10, 9), ($1*10+7, $1+299, 899), ($1*10+8, $1+10, 12), ($1*10+9, $1+10, 14*$1);
+ eval insert into t1 values(NULL, $1+9, 5*$1), (NULL, $1+10, 7),(NULL, $1+10, 7*$1), (NULL, $1+10, 10+$1), (NULL, $1+10, 70*$1), (NULL, $1+10, 7), (NULL, $1+10, 9), (NULL, $1+299, 899), (NULL, $1+10, 12), (NULL, $1+10, 14*$1);
  dec $1;
 }
 enable_query_log;
 
-select count(*) from t2;
+select count(*) from t1;
+
+select * from t1 order by a limit 2;
+
+truncate table t1;
+
+select count(*) from t1;
 
-truncate table t2;
+insert into t1 values(NULL,1,1),(NULL,2,2);
 
-select count(*) from t2;
+select * from t1 order by a;
 
-drop table t2;
+drop table t1;
 
 # End of 4.1 tests

sql/ha_ndbcluster.h

@@ -628,6 +628,7 @@ class ha_ndbcluster: public handler
   int extra_opt(enum ha_extra_function operation, ulong cache_size);
   int reset();
   int external_lock(THD *thd, int lock_type);
+  void unlock_row();
   int start_stmt(THD *thd, thr_lock_type lock_type);
   void print_error(int error, myf errflag);
   const char * table_type() const;
@@ -855,6 +856,7 @@ private:
   char m_tabname[FN_HEADLEN];
   ulong m_table_flags;
   THR_LOCK_DATA m_lock;
+  bool m_lock_tuple;
   NDB_SHARE *m_share;
   NDB_INDEX_DATA  m_index[MAX_KEY];
   THD_NDB_SHARE *m_thd_ndb_share;

storage/ndb/include/ndbapi/NdbIndexScanOperation.hpp

@@ -62,11 +62,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

storage/ndb/include/ndbapi/NdbScanOperation.hpp

@@ -45,7 +45,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
   };
 
   /**
@@ -62,15 +63,14 @@ public:
 #ifndef DOXYGEN_SHOULD_SKIP_DEPRECATED
   /**
    * readTuples
-   * 
    * @param lock_mode Lock mode
    * @param batch No of rows to fetch from each fragment at a time
    * @param parallel No of fragments to scan in parallell
-   * @note specifying 0 for batch and parallall means max performance
+   * @note specifying 0 for batch and parallell means max performance
    */ 
 #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){
@@ -142,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
    *
@@ -251,6 +265,19 @@ protected:
   bool m_executed; // Marker if operation should be released at close
 };
 
+inline
+NdbOperation* 
+NdbScanOperation::lockCurrentTuple(){
+  return lockCurrentTuple(m_transConnection);
+}
+
+inline
+NdbOperation* 
+NdbScanOperation::lockCurrentTuple(NdbTransaction* takeOverTrans){
+  return takeOverScanOp(NdbOperation::ReadRequest, 
+			takeOverTrans);
+}
+
 inline
 NdbOperation* 
 NdbScanOperation::updateCurrentTuple(){

storage/ndb/src/ndbapi/NdbDictionaryImpl.hpp

@@ -581,7 +581,7 @@ public:
   int createIndex(NdbIndexImpl &ix, NdbTableImpl & tab);
   int dropIndex(const char * indexName, 
 		const char * tableName);
-  int dropIndex(NdbIndexImpl &, const char * tableName);
+  int dropIndex(NdbIndexImpl &);
   NdbTableImpl * getIndexTable(NdbIndexImpl * index, 
 			       NdbTableImpl * table);
 

storage/ndb/src/ndbapi/NdbScanOperation.cpp

@@ -162,7 +162,7 @@ NdbScanOperation::readTuples(NdbScanOperation::LockMode lm,
     return -1;
   }
 
-  m_keyInfo = lockExcl ? 1 : 0;
+  m_keyInfo = ((scan_flags & SF_KeyInfo) || lockExcl) ? 1 : 0;
   bool tupScan = (scan_flags & SF_TupScan);
 
 #if 1 // XXX temp for testing
@@ -942,6 +942,12 @@ NdbScanOperation::takeOverScanOp(OperationType opType, NdbTransaction* pTrans)
     if (newOp == NULL){
       return NULL;
     }
+    if (!m_keyInfo)
+    {
+      // Cannot take over lock if no keyinfo was requested
+      setErrorCodeAbort(4604);
+      return NULL;
+    }
     pTrans->theSimpleState = 0;
     
     assert(tRecAttr->get_size_in_bytes() > 0);
@@ -950,12 +956,16 @@ NdbScanOperation::takeOverScanOp(OperationType opType, NdbTransaction* pTrans)
     
     newOp->theTupKeyLen = len;
     newOp->theOperationType = opType;
-    if (opType == DeleteRequest) {
+    switch (opType) {
+    case (ReadRequest):
+      newOp->theLockMode = theLockMode;
+      // Fall through
+    case (DeleteRequest):
       newOp->theStatus = GetValue;
-    } else {
+      break;
+    default:
       newOp->theStatus = SetValue;
     }
-    
     const Uint32 * src = (Uint32*)tRecAttr->aRef();
     const Uint32 tScanInfo = src[len] & 0x3FFFF;
     const Uint32 tTakeOverFragment = src[len] >> 20;

storage/ndb/src/ndbapi/ndberror.c

@@ -308,7 +308,7 @@ ErrorBundle ErrorCodes[] = {
   { 4601, DMEC, AE, "Transaction is not started"},
   { 4602, DMEC, AE, "You must call getNdbOperation before executeScan" },
   { 4603, DMEC, AE, "There can only be ONE operation in a scan transaction" },
-  { 4604, DMEC, AE, "takeOverScanOp, opType must be UpdateRequest or DeleteRequest" },
+  { 4604, DMEC, AE, "takeOverScanOp, to take over a scanned row one must explicitly request keyinfo on readTuples call" },
   { 4605, DMEC, AE, "You may only call openScanRead or openScanExclusive once for each operation"},
   { 4607, DMEC, AE, "There may only be one operation in a scan transaction"},
   { 4608, DMEC, AE, "You can not takeOverScan unless you have used openScanExclusive"},