=== modified file 'sql/ha_ndbcluster.cc'
--- sql/ha_ndbcluster.cc	2009-02-02 15:58:48 +0000
+++ sql/ha_ndbcluster.cc	2009-02-05 12:35:42 +0000
@@ -2602,7 +2602,7 @@ int ha_ndbcluster::ordered_index_scan(co
   if (lm == NdbOperation::LM_Read)
     options.scan_flags|= NdbScanOperation::SF_KeyInfo;
   if (sorted)
-    options.scan_flags|= NdbScanOperation::SF_OrderBy;
+    options.scan_flags|= NdbScanOperation::SF_OrderByFull;
   if (descending)
     options.scan_flags|= NdbScanOperation::SF_Descending;
   const NdbRecord *key_rec= m_index[active_index].ndb_record_key;
@@ -9697,7 +9697,7 @@ int ha_ndbcluster::multi_range_start_ret
         if (lm == NdbOperation::LM_Read)
           options.scan_flags|= NdbScanOperation::SF_KeyInfo;
         if (mrr_is_output_sorted)
-          options.scan_flags|= NdbScanOperation::SF_OrderBy;
+          options.scan_flags|= NdbScanOperation::SF_OrderByFull;
 
         options.parallel=parallelism;
 

=== modified file 'storage/ndb/include/ndbapi/NdbScanOperation.hpp'
--- storage/ndb/include/ndbapi/NdbScanOperation.hpp	2008-11-08 20:40:15 +0000
+++ storage/ndb/include/ndbapi/NdbScanOperation.hpp	2009-02-05 12:25:27 +0000
@@ -54,6 +54,12 @@ public:
       each fragment, to get a single sorted result set.
     */
     SF_OrderBy = (1 << 24),
+    /**
+     * Same as order by, except that it will automatically 
+     *   add all key columns into the read-mask
+     */
+    SF_OrderByFull = (16 << 24),
+
     /* Index scan in descending order, instead of default ascending. */
     SF_Descending = (2 << 24),
     /*

=== modified file 'storage/ndb/src/ndbapi/NdbScanOperation.cpp'
--- storage/ndb/src/ndbapi/NdbScanOperation.cpp	2009-02-04 13:08:05 +0000
+++ storage/ndb/src/ndbapi/NdbScanOperation.cpp	2009-02-05 12:35:15 +0000
@@ -788,7 +788,10 @@ NdbIndexScanOperation::scanIndexImpl(con
     return -1;
   }
 
-  if (scan_flags & NdbScanOperation::SF_OrderBy)
+  result_record->copyMask(m_read_mask, result_mask);
+
+  if (scan_flags & (NdbScanOperation::SF_OrderBy | 
+                    NdbScanOperation::SF_OrderByFull))
   {
     /**
      * For ordering, we need all keys in the result row.
@@ -796,19 +799,35 @@ NdbIndexScanOperation::scanIndexImpl(con
      * So for each key column, check that it is included in the result
      * NdbRecord.
      */
+#define MASKSZ ((NDB_MAX_ATTRIBUTES_IN_TABLE+31)>>5)
+    Uint32 keymask[MASKSZ];
+    BitmaskImpl::clear(MASKSZ, keymask);
+
     for (i = 0; i < key_record->key_index_length; i++)
     {
-      const NdbRecord::Attr *key_col =
-        &key_record->columns[key_record->key_indexes[i]];
-      if (key_col->attrId >= result_record->m_attrId_indexes_length ||
-          result_record->m_attrId_indexes[key_col->attrId] < 0)
+      Uint32 attrId = key_record->columns[key_record->key_indexes[i]].attrId;
+      if (attrId >= result_record->m_attrId_indexes_length ||
+          result_record->m_attrId_indexes[attrId] < 0)
       {
+        abort();
         setErrorCodeAbort(4292);
         return -1;
       }
+
+      BitmaskImpl::set(MASKSZ, keymask);
     }
-  }
 
+    if (scan_flags & NdbScanOperation::SF_OrderByFull)
+    {
+      BitmaskImpl::bitOR(MASKSZ, m_read_mask, keymask);
+    }
+    else if (!BitmaskImpl::contains(MASKSZ, m_read_mask, keymask))
+    {
+      setErrorCodeAbort(4292);
+      return;
+    }
+  }
+  
   if (!(key_record->flags & NdbRecord::RecIsIndex))
   {
     setErrorCodeAbort(4283);
@@ -833,8 +852,6 @@ NdbIndexScanOperation::scanIndexImpl(con
   if (res==-1)
     return -1;
 
-  result_record->copyMask(m_read_mask, result_mask);
-
   /* Fix theStatus as set in processIndexScanDefs(). */
   theStatus= NdbOperation::UseNdbRecord;