Merge mskold@bk-internal.mysql.com:/home/bk/mysql-5.1-ndb

into  mysql.com:/windows/Linux_space/MySQL/mysql-5.1-new-ndb

mysql-test/r/ndb_read_multi_range.result

@@ -459,3 +459,17 @@ INSERT INTO t1 VALUES (1,1),(2,2),(3,3);
 UPDATE t1 SET var2 = 9 WHERE var1 IN(1,2,3);
 DROP TRIGGER testtrigger;
 DROP TABLE t1, t2;
+create table t1 (a int, b int, primary key (a), key ab (a,b)) engine=ndbcluster;
+insert into t1 values (1,1), (10,10);
+select * from t1 use index (ab) where a in(1,10) order by a;
+a	b
+1	1
+10	10
+create table t2 (a int, b int, primary key (a,b)) engine=ndbcluster
+partition by key(a);
+insert into t2 values (1,1), (10,10);
+select * from t2 where a in (1,10) order by a;
+a	b
+1	1
+10	10
+drop table t1, t2;

mysql-test/t/ndb_read_multi_range.test

@@ -301,3 +301,19 @@ UPDATE t1 SET var2 = 9 WHERE var1 IN(1,2,3);
 DROP TRIGGER testtrigger;
 
 DROP TABLE t1, t2;
+
+#bug#25821
+create table t1 (a int, b int, primary key (a), key ab (a,b)) engine=ndbcluster;
+
+insert into t1 values (1,1), (10,10);
+
+select * from t1 use index (ab) where a in(1,10) order by a;
+
+create table t2 (a int, b int, primary key (a,b)) engine=ndbcluster
+partition by key(a);
+
+insert into t2 values (1,1), (10,10);
+
+select * from t2 where a in (1,10) order by a;
+
+drop table t1, t2;

sql/ha_ndbcluster.cc

@@ -8231,7 +8231,7 @@ ha_ndbcluster::read_multi_range_first(KEY_MULTI_RANGE **found_range_p,
         }
         else if ((scanOp= m_active_trans->getNdbIndexScanOperation(idx, tab)) 
                  &&!scanOp->readTuples(lm, 0, parallelism, sorted, 
-				       FALSE, TRUE, need_pk)
+				       FALSE, TRUE, need_pk, TRUE)
                  &&!generate_scan_filter(m_cond_stack, scanOp)
                  &&!define_read_attrs(end_of_buffer-reclength, scanOp))
         {

storage/ndb/include/ndbapi/NdbIndexScanOperation.hpp

@@ -64,12 +64,14 @@ public:
                         bool order_by,
                         bool order_desc = false,
                         bool read_range_no = false,
-			bool keyinfo = false) {
+			bool keyinfo = false,
+			bool multi_range = false) {
     Uint32 scan_flags =
       (SF_OrderBy & -(Int32)order_by) |
       (SF_Descending & -(Int32)order_desc) |
       (SF_ReadRangeNo & -(Int32)read_range_no) | 
-      (SF_KeyInfo & -(Int32)keyinfo);
+      (SF_KeyInfo & -(Int32)keyinfo) |
+      (SF_MultiRange & -(Int32)multi_range);
     
     return readTuples(lock_mode, scan_flags, parallel, batch);
   }

storage/ndb/include/ndbapi/NdbScanOperation.hpp

@@ -46,6 +46,7 @@ public:
     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_MultiRange = (8 << 24),  // scan is part of multi-range scan
     SF_KeyInfo = 1              // request KeyInfo to be sent back
   };
 
@@ -72,7 +73,8 @@ public:
    */ 
 #ifdef ndb_readtuples_impossible_overload
   int readTuples(LockMode lock_mode = LM_Read, 
-		 Uint32 batch = 0, Uint32 parallel = 0, bool keyinfo = false);
+		 Uint32 batch = 0, Uint32 parallel = 0, 
+                 bool keyinfo = false, bool multi_range = false);
 #endif
   
   inline int readTuples(int parallell){
@@ -264,6 +266,7 @@ protected:
   bool m_descending;
   Uint32 m_read_range_no;
   NdbRecAttr *m_curr_row; // Pointer to last returned row
+  bool m_multi_range; // Mark if operation is part of multi-range scan
   bool m_executed; // Marker if operation should be released at close
 };
 

storage/ndb/src/ndbapi/NdbScanOperation.cpp

@@ -1225,7 +1225,7 @@ NdbIndexScanOperation::setBound(const NdbColumnImpl* tAttrInfo,
      * so it's safe to use [tIndexAttrId] 
      * (instead of looping as is NdbOperation::equal_impl)
      */
-    if(type == BoundEQ && tDistrKey)
+    if(type == BoundEQ && tDistrKey && !m_multi_range)
     {
       theNoOfTupKeyLeft--;
       return handle_distribution_key(valPtr, sizeInWords);
@@ -1311,6 +1311,7 @@ NdbIndexScanOperation::readTuples(LockMode lm,
   const bool order_by = scan_flags & SF_OrderBy;
   const bool order_desc = scan_flags & SF_Descending;
   const bool read_range_no = scan_flags & SF_ReadRangeNo;
+  m_multi_range = scan_flags & SF_MultiRange;
 
   int res = NdbScanOperation::readTuples(lm, scan_flags, parallel, batch);
   if(!res && read_range_no)