=== modified file 'sql/ha_ndbcluster.cc' --- sql/ha_ndbcluster.cc 2009-01-29 16:24:04 +0000 +++ sql/ha_ndbcluster.cc 2009-02-05 12:53:16 +0000 @@ -2594,7 +2594,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; @@ -9317,7 +9317,7 @@ ha_ndbcluster::read_multi_range_first(KE 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; 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:52:14 +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 06:52:21 +0000 +++ storage/ndb/src/ndbapi/NdbScanOperation.cpp 2009-02-09 08:32:24 +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,34 @@ 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) { 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(4341); + return -1; + } + } + if (!(key_record->flags & NdbRecord::RecIsIndex)) { setErrorCodeAbort(4283); @@ -832,8 +850,6 @@ NdbIndexScanOperation::scanIndexImpl(con res= processIndexScanDefs(lock_mode, scan_flags, parallel, batch); if (res==-1) return -1; - - result_record->copyMask(m_read_mask, result_mask); /* Fix theStatus as set in processIndexScanDefs(). */ theStatus= NdbOperation::UseNdbRecord; === modified file 'storage/ndb/src/ndbapi/ndberror.c' --- storage/ndb/src/ndbapi/ndberror.c 2009-02-05 13:03:37 +0000 +++ storage/ndb/src/ndbapi/ndberror.c 2009-02-09 08:33:23 +0000 @@ -678,6 +678,7 @@ ErrorBundle ErrorCodes[] = { { 4290, DMEC, AE, "Missing column specification in NdbDictionary::RecordSpecification" }, { 4291, DMEC, AE, "Duplicate column specification in NdbDictionary::RecordSpecification" }, { 4292, DMEC, AE, "NdbRecord for tuple access is not an index key NdbRecord" }, + { 4341, DMEC, AE, "Not all keys read when using option SF_OrderBy" }, { 4293, DMEC, AE, "Error returned from application scanIndex() callback" }, { 4294, DMEC, AE, "Scan filter is too large, discarded" }, { 4295, DMEC, AE, "Column is NULL in Get/SetValueSpec structure" },