database/mariadb
1
0
Fork 0
mirror of https://github.com/MariaDB/server.git synced 2025-08-01 03:47:19 +03:00
Code Activity

wl2126 - ndb - fix last unhandled part of read_multi_range

Browse Source 
reading sorted multi_range with several range-scans


BitKeeper/deleted/.del-NdbResultSet.hpp~e0042f4c18088ff6:
  Delete: ndb/include/ndbapi/NdbResultSet.hpp
BitKeeper/deleted/.del-NdbResultSet.cpp~84d192cf3f42600d:
  Delete: ndb/src/ndbapi/NdbResultSet.cpp
ndb/include/ndbapi/NdbApi.hpp:
  Remove NdbResultSet
ndb/include/ndbapi/NdbIndexScanOperation.hpp:
  Remove NdbResultSet
ndb/include/ndbapi/NdbScanOperation.hpp:
  Remove NdbResultSet
ndb/src/ndbapi/Makefile.am:
  Remove NdbResultSet
ndb/src/ndbapi/NdbConnection.cpp:
  Remove NdbResultSet
ndb/src/ndbapi/NdbIndexOperation.cpp:
  Remove NdbResultSet
ndb/src/ndbapi/NdbScanOperation.cpp:
  Remove NdbResultSet
ndb/test/include/HugoOperations.hpp:
  Remove NdbResultSet
ndb/test/ndbapi/ScanFunctions.hpp:
  Remove NdbResultSet
ndb/test/ndbapi/ScanInterpretTest.hpp:
  Remove NdbResultSet
ndb/test/ndbapi/bank/Bank.cpp:
  Remove NdbResultSet
ndb/test/ndbapi/bank/BankLoad.cpp:
  Remove NdbResultSet
ndb/test/ndbapi/slow_select.cpp:
  Remove NdbResultSet
ndb/test/ndbapi/testBlobs.cpp:
  Remove NdbResultSet
ndb/test/ndbapi/testDataBuffers.cpp:
  Remove NdbResultSet
ndb/test/ndbapi/testDeadlock.cpp:
  Remove NdbResultSet
ndb/test/ndbapi/testIndex.cpp:
  Remove NdbResultSet
ndb/test/ndbapi/testOIBasic.cpp:
  Remove NdbResultSet
ndb/test/ndbapi/testReadPerf.cpp:
  Remove NdbResultSet
ndb/test/ndbapi/testScan.cpp:
  Remove NdbResultSet
ndb/test/ndbapi/testScanPerf.cpp:
  Remove NdbResultSet
ndb/test/src/HugoOperations.cpp:
  Remove NdbResultSet
ndb/test/src/HugoTransactions.cpp:
  Remove NdbResultSet
ndb/test/src/UtilTransactions.cpp:
  Remove NdbResultSet
ndb/tools/delete_all.cpp:
  Remove NdbResultSet
ndb/tools/select_all.cpp:
  Remove NdbResultSet
ndb/tools/select_count.cpp:
  Remove NdbResultSet
sql/ha_ndbcluster.cc:
  1) Fix a = 3 -> a= 3
  2) Impl. last part of wl2126
     ordered "multi ordered scan"
This commit is contained in:
unknown
2004-12-07 15:15:49 +01:00
parent a01f06864b
commit 2cae258f75
30 changed files with 498 additions and 709 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
ndb/include/ndbapi/NdbApi.hpp
View File

@ -26,7 +26,6 @@
#include "NdbIndexScanOperation.hpp" #include "NdbIndexScanOperation.hpp"
#include "NdbScanFilter.hpp" #include "NdbScanFilter.hpp"
#include "NdbRecAttr.hpp" #include "NdbRecAttr.hpp"
#include "NdbResultSet.hpp"
#include "NdbDictionary.hpp" #include "NdbDictionary.hpp"
#include "NdbEventOperation.hpp" #include "NdbEventOperation.hpp"
#include "NdbPool.hpp" #include "NdbPool.hpp"

8
ndb/include/ndbapi/NdbIndexScanOperation.hpp
View File

@ -42,17 +42,17 @@ public:
* @returns NdbResultSet. * @returns NdbResultSet.
* @see NdbScanOperation::readTuples * @see NdbScanOperation::readTuples
*/ */
NdbResultSet* readTuples(LockMode = LM_Read, int readTuples(LockMode = LM_Read,
Uint32 batch = 0, Uint32 batch = 0,
Uint32 parallel = 0, Uint32 parallel = 0,
bool order_by = false, bool order_by = false,
bool read_range_no = false); bool read_range_no = false);
inline NdbResultSet* readTuples(int parallell){ inline int readTuples(int parallell){
return readTuples(LM_Read, 0, parallell, false); return readTuples(LM_Read, 0, parallell, false);
} }
inline NdbResultSet* readTuplesExclusive(int parallell = 0){ inline int readTuplesExclusive(int parallell = 0){
return readTuples(LM_Exclusive, 0, parallell, false); return readTuples(LM_Exclusive, 0, parallell, false);
} }
@ -125,7 +125,7 @@ public:
/** /**
* Return range no for current row * Return range no for current row
*/ */
Uint32 get_range_no(); int get_range_no();
bool getSorted() const { return m_ordered; } bool getSorted() const { return m_ordered; }
private: private:

162
ndb/include/ndbapi/NdbResultSet.hpp
View File

@ -1,162 +0,0 @@
/* Copyright (C) 2003 MySQL AB
This program is free software; you can redistribute it and/or modify
it under the terms of the GNU General Public License as published by
the Free Software Foundation; either version 2 of the License, or
(at your option) any later version.
This program is distributed in the hope that it will be useful,
but WITHOUT ANY WARRANTY; without even the implied warranty of
MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
GNU General Public License for more details.
You should have received a copy of the GNU General Public License
along with this program; if not, write to the Free Software
Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA */
/*****************************************************************************
* Name: NdbResultSet.hpp
* Include:
* Link:
* Author: Martin Sk<53>ld
* Date: 2002-04-01
* Version: 0.1
* Description: Cursor class
* Documentation:
* Adjust: 2002-04-01 Martin Sk<53>ld First version.
****************************************************************************/
#ifndef NdbResultSet_H
#define NdbResultSet_H
#include <NdbScanOperation.hpp>
/**
* @class NdbResultSet
* @brief NdbResultSet contains a NdbScanOperation.
*/
class NdbResultSet
{
friend class NdbScanOperation;
public:
/**
* Get the next tuple in a scan transaction.
*
* After each call to NdbResult::nextResult
* the buffers and NdbRecAttr objects defined in
* NdbOperation::getValue are updated with values
* from the scanned tuple.
*
* @param fetchAllowed If set to false, then fetching is disabled
*
* The NDB API will contact the NDB Kernel for more tuples
* when necessary to do so unless you set the fetchAllowed
* to false.
* This will force NDB to process any records it
* already has in it's caches. When there are no more cached
* records it will return 2. You must then call nextResult
* with fetchAllowed = true in order to contact NDB for more
* records.
*
* fetchAllowed = false is useful when you want to update or
* delete all the records fetched in one transaction(This will save a
* lot of round trip time and make updates or deletes of scanned
* records a lot faster).
* While nextResult(false)
* returns 0 take over the record to another transaction. When
* nextResult(false) returns 2 you must execute and commit the other
* transaction. This will cause the locks to be transferred to the
* other transaction, updates or deletes will be made and then the
* locks will be released.
* After that, call nextResult(true) which will fetch new records and
* cache them in the NdbApi.
*
* @note If you don't take over the records to another transaction the
* locks on those records will be released the next time NDB Kernel
* is contacted for more records.
*
* @note Please contact for examples of efficient scan
* updates and deletes.
*
* @note See ndb/examples/ndbapi_scan_example for usage.
*
* @return
* - -1: if unsuccessful,<br>
* - 0: if another tuple was received, and<br>
* - 1: if there are no more tuples to scan.
* - 2: if there are no more cached records in NdbApi
*/
int nextResult(bool fetchAllowed = true, bool forceSend = false);
/**
* Close result set (scan)
*/
void close(bool forceSend = false);
/**
* Restart
*/
int restart(bool forceSend = false);
/**
* Transfer scan operation to an updating transaction. Use this function
* when a scan has found a record that you want to update.
* 1. Start a new transaction.
* 2. Call the function takeOverForUpdate using your new transaction
* as parameter, all the properties of the found record will be copied
* to the new transaction.
* 3. When you execute the new transaction, the lock held by the scan will
* be transferred to the new transaction(it's taken over).
*
* @note You must have started the scan with openScanExclusive
* to be able to update the found tuple.
*
* @param updateTrans the update transaction connection.
* @return an NdbOperation or NULL.
*/
NdbOperation* updateTuple();
NdbOperation* updateTuple(NdbConnection* updateTrans);
/**
* Transfer scan operation to a deleting transaction. Use this function
* when a scan has found a record that you want to delete.
* 1. Start a new transaction.
* 2. Call the function takeOverForDelete using your new transaction
* as parameter, all the properties of the found record will be copied
* to the new transaction.
* 3. When you execute the new transaction, the lock held by the scan will
* be transferred to the new transaction(its taken over).
*
* @note You must have started the scan with openScanExclusive
* to be able to delete the found tuple.
*
* @param deleteTrans the delete transaction connection.
* @return an NdbOperation or NULL.
*/
int deleteTuple();
int deleteTuple(NdbConnection* takeOverTransaction);
/**
* Get underlying operation
*/
NdbOperation* getOperation();
private:
NdbResultSet(NdbScanOperation*);
~NdbResultSet();
void init();
NdbScanOperation* m_operation;
};
inline
NdbOperation*
NdbResultSet::getOperation(){
return m_operation;
}
#endif

156
ndb/include/ndbapi/NdbScanOperation.hpp
View File

@ -45,20 +45,6 @@ class NdbScanOperation : public NdbOperation {
friend class NdbOperation; friend class NdbOperation;
friend class NdbBlob; friend class NdbBlob;
public: public:
/**
* Type of cursor
*/
enum CursorType {
NoCursor = 0,
ScanCursor = 1,
IndexCursor = 2
};
/**
* Type of cursor
*/
CursorType get_cursor_type() const;
/** /**
* readTuples returns a NdbResultSet where tuples are stored. * readTuples returns a NdbResultSet where tuples are stored.
* Tuples are not stored in NdbResultSet until execute(NoCommit) * Tuples are not stored in NdbResultSet until execute(NoCommit)
@ -67,33 +53,126 @@ public:
* @param parallel Scan parallelism * @param parallel Scan parallelism
* @param batch No of rows to fetch from each fragment at a time * @param batch No of rows to fetch from each fragment at a time
* @param LockMode Scan lock handling * @param LockMode Scan lock handling
* @returns NdbResultSet.
* @note specifying 0 for batch and parallall means max performance * @note specifying 0 for batch and parallall means max performance
*/ */
NdbResultSet* readTuples(LockMode = LM_Read, int readTuples(LockMode = LM_Read,
Uint32 batch = 0, Uint32 parallel = 0); Uint32 batch = 0, Uint32 parallel = 0);
inline NdbResultSet* readTuples(int parallell){ inline int readTuples(int parallell){
return readTuples(LM_Read, 0, parallell); return readTuples(LM_Read, 0, parallell);
} }
inline NdbResultSet* readTuplesExclusive(int parallell = 0){ inline int readTuplesExclusive(int parallell = 0){
return readTuples(LM_Exclusive, 0, parallell); return readTuples(LM_Exclusive, 0, parallell);
} }
NdbBlob* getBlobHandle(const char* anAttrName); NdbBlob* getBlobHandle(const char* anAttrName);
NdbBlob* getBlobHandle(Uint32 anAttrId); NdbBlob* getBlobHandle(Uint32 anAttrId);
protected: /**
CursorType m_cursor_type; * Get the next tuple in a scan transaction.
*
* After each call to NdbResult::nextResult
* the buffers and NdbRecAttr objects defined in
* NdbOperation::getValue are updated with values
* from the scanned tuple.
*
* @param fetchAllowed If set to false, then fetching is disabled
*
* The NDB API will contact the NDB Kernel for more tuples
* when necessary to do so unless you set the fetchAllowed
* to false.
* This will force NDB to process any records it
* already has in it's caches. When there are no more cached
* records it will return 2. You must then call nextResult
* with fetchAllowed = true in order to contact NDB for more
* records.
*
* fetchAllowed = false is useful when you want to update or
* delete all the records fetched in one transaction(This will save a
* lot of round trip time and make updates or deletes of scanned
* records a lot faster).
* While nextResult(false)
* returns 0 take over the record to another transaction. When
* nextResult(false) returns 2 you must execute and commit the other
* transaction. This will cause the locks to be transferred to the
* other transaction, updates or deletes will be made and then the
* locks will be released.
* After that, call nextResult(true) which will fetch new records and
* cache them in the NdbApi.
*
* @note If you don't take over the records to another transaction the
* locks on those records will be released the next time NDB Kernel
* is contacted for more records.
*
* @note Please contact for examples of efficient scan
* updates and deletes.
*
* @note See ndb/examples/ndbapi_scan_example for usage.
*
* @return
* - -1: if unsuccessful,<br>
* - 0: if another tuple was received, and<br>
* - 1: if there are no more tuples to scan.
* - 2: if there are no more cached records in NdbApi
*/
int nextResult(bool fetchAllowed = true, bool forceSend = false);
/**
* Close result set (scan)
*/
void close(bool forceSend = false);
/**
* Restart
*/
int restart(bool forceSend = false);
/**
* Transfer scan operation to an updating transaction. Use this function
* when a scan has found a record that you want to update.
* 1. Start a new transaction.
* 2. Call the function takeOverForUpdate using your new transaction
* as parameter, all the properties of the found record will be copied
* to the new transaction.
* 3. When you execute the new transaction, the lock held by the scan will
* be transferred to the new transaction(it's taken over).
*
* @note You must have started the scan with openScanExclusive
* to be able to update the found tuple.
*
* @param updateTrans the update transaction connection.
* @return an NdbOperation or NULL.
*/
NdbOperation* updateCurrentTuple();
NdbOperation* updateCurrentTuple(NdbConnection* updateTrans);
/**
* Transfer scan operation to a deleting transaction. Use this function
* when a scan has found a record that you want to delete.
* 1. Start a new transaction.
* 2. Call the function takeOverForDelete using your new transaction
* as parameter, all the properties of the found record will be copied
* to the new transaction.
* 3. When you execute the new transaction, the lock held by the scan will
* be transferred to the new transaction(its taken over).
*
* @note You must have started the scan with openScanExclusive
* to be able to delete the found tuple.
*
* @param deleteTrans the delete transaction connection.
* @return an NdbOperation or NULL.
*/
int deleteCurrentTuple();
int deleteCurrentTuple(NdbConnection* takeOverTransaction);
protected:
NdbScanOperation(Ndb* aNdb); NdbScanOperation(Ndb* aNdb);
virtual ~NdbScanOperation(); virtual ~NdbScanOperation();
int nextResult(bool fetchAllowed = true, bool forceSend = false); int nextResultImpl(bool fetchAllowed = true, bool forceSend = false);
virtual void release(); virtual void release();
void closeScan(bool forceSend = false);
int close_impl(class TransporterFacade*, bool forceSend = false); int close_impl(class TransporterFacade*, bool forceSend = false);
// Overloaded methods from NdbCursorOperation // Overloaded methods from NdbCursorOperation
@ -108,8 +187,6 @@ protected:
virtual void setErrorCode(int aErrorCode); virtual void setErrorCode(int aErrorCode);
virtual void setErrorCodeAbort(int aErrorCode); virtual void setErrorCodeAbort(int aErrorCode);
NdbResultSet * m_resultSet;
NdbResultSet* getResultSet();
NdbConnection *m_transConnection; NdbConnection *m_transConnection;
// Scan related variables // Scan related variables
@ -157,14 +234,35 @@ protected:
Uint32 m_ordered; Uint32 m_ordered;
Uint32 m_read_range_no; Uint32 m_read_range_no;
int restart(bool forceSend = false);
}; };
inline inline
NdbScanOperation::CursorType NdbOperation*
NdbScanOperation::get_cursor_type() const { NdbScanOperation::updateCurrentTuple(){
return m_cursor_type; return updateCurrentTuple(m_transConnection);
}
inline
NdbOperation*
NdbScanOperation::updateCurrentTuple(NdbConnection* takeOverTrans){
return takeOverScanOp(NdbOperation::UpdateRequest,
takeOverTrans);
}
inline
int
NdbScanOperation::deleteCurrentTuple(){
return deleteCurrentTuple(m_transConnection);
}
inline
int
NdbScanOperation::deleteCurrentTuple(NdbConnection * takeOverTrans){
void * res = takeOverScanOp(NdbOperation::DeleteRequest,
takeOverTrans);
if(res == 0)
return -1;
return 0;
} }
#endif #endif

1
ndb/src/ndbapi/Makefile.am
View File

@ -22,7 +22,6 @@ libndbapi_la_SOURCES = \
NdbOperationInt.cpp \ NdbOperationInt.cpp \
NdbOperationDefine.cpp \ NdbOperationDefine.cpp \
NdbOperationExec.cpp \ NdbOperationExec.cpp \
NdbResultSet.cpp \
NdbScanOperation.cpp NdbScanFilter.cpp \ NdbScanOperation.cpp NdbScanFilter.cpp \
NdbIndexOperation.cpp \ NdbIndexOperation.cpp \
NdbEventOperation.cpp \ NdbEventOperation.cpp \

1
ndb/src/ndbapi/NdbConnection.cpp
View File

@ -1109,7 +1109,6 @@ NdbConnection::getNdbIndexScanOperation(const NdbIndexImpl* index,
if(tOp) if(tOp)
{ {
tOp->m_currentTable = table; tOp->m_currentTable = table;
tOp->m_cursor_type = NdbScanOperation::IndexCursor;
} }
return tOp; return tOp;
} else { } else {

1
ndb/src/ndbapi/NdbIndexOperation.cpp
View File

@ -16,7 +16,6 @@
#include <ndb_global.h> #include <ndb_global.h>
#include <NdbIndexOperation.hpp> #include <NdbIndexOperation.hpp>
#include <NdbResultSet.hpp>
#include <Ndb.hpp> #include <Ndb.hpp>
#include <NdbConnection.hpp> #include <NdbConnection.hpp>
#include "NdbApiSignal.hpp" #include "NdbApiSignal.hpp"

103
ndb/src/ndbapi/NdbResultSet.cpp
View File

@ -1,103 +0,0 @@
/* Copyright (C) 2003 MySQL AB
This program is free software; you can redistribute it and/or modify
it under the terms of the GNU General Public License as published by
the Free Software Foundation; either version 2 of the License, or
(at your option) any later version.
This program is distributed in the hope that it will be useful,
but WITHOUT ANY WARRANTY; without even the implied warranty of
MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
GNU General Public License for more details.
You should have received a copy of the GNU General Public License
along with this program; if not, write to the Free Software
Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA */
/*****************************************************************************
* Name: NdbResultSet.cpp
* Include:
* Link:
* Author: UABMASD Martin Sk<53>ld INN/V Alzato
* Date: 2002-04-01
* Version: 0.1
* Description: Cursor class
* Documentation:
* Adjust: 2002-04-01 UABMASD First version.
****************************************************************************/
#include <Ndb.hpp>
#include <NdbConnection.hpp>
#include <NdbResultSet.hpp>
#include <NdbBlob.hpp>
NdbResultSet::NdbResultSet(NdbScanOperation *owner)
: m_operation(owner)
{
}
NdbResultSet::~NdbResultSet()
{
}
void NdbResultSet::init()
{
}
int NdbResultSet::nextResult(bool fetchAllowed, bool forceSend)
{
int res;
if ((res = m_operation->nextResult(fetchAllowed, forceSend)) == 0) {
// handle blobs
NdbBlob* tBlob = m_operation->theBlobList;
while (tBlob != 0) {
if (tBlob->atNextResult() == -1)
return -1;
tBlob = tBlob->theNext;
}
/*
* Flush blob part ops on behalf of user because
* - nextResult is analogous to execute(NoCommit)
* - user is likely to want blob value before next execute
*/
if (m_operation->m_transConnection->executePendingBlobOps() == -1)
return -1;
return 0;
}
return res;
}
void NdbResultSet::close(bool forceSend)
{
m_operation->closeScan(forceSend);
}
NdbOperation*
NdbResultSet::updateTuple(){
return updateTuple(m_operation->m_transConnection);
}
NdbOperation*
NdbResultSet::updateTuple(NdbConnection* takeOverTrans){
return m_operation->takeOverScanOp(NdbOperation::UpdateRequest,
takeOverTrans);
}
int
NdbResultSet::deleteTuple(){
return deleteTuple(m_operation->m_transConnection);
}
int
NdbResultSet::deleteTuple(NdbConnection * takeOverTrans){
void * res = m_operation->takeOverScanOp(NdbOperation::DeleteRequest,
takeOverTrans);
if(res == 0)
return -1;
return 0;
}
int
NdbResultSet::restart(bool forceSend){
return m_operation->restart(forceSend);
}

72
ndb/src/ndbapi/NdbScanOperation.cpp
View File

@ -19,10 +19,10 @@
#include <NdbScanOperation.hpp> #include <NdbScanOperation.hpp>
#include <NdbIndexScanOperation.hpp> #include <NdbIndexScanOperation.hpp>
#include <NdbConnection.hpp> #include <NdbConnection.hpp>
#include <NdbResultSet.hpp>
#include "NdbApiSignal.hpp" #include "NdbApiSignal.hpp"
#include <NdbOut.hpp> #include <NdbOut.hpp>
#include "NdbDictionaryImpl.hpp" #include "NdbDictionaryImpl.hpp"
#include <NdbBlob.hpp>
#include <NdbRecAttr.hpp> #include <NdbRecAttr.hpp>
#include <NdbReceiver.hpp> #include <NdbReceiver.hpp>
@ -39,7 +39,6 @@
NdbScanOperation::NdbScanOperation(Ndb* aNdb) : NdbScanOperation::NdbScanOperation(Ndb* aNdb) :
NdbOperation(aNdb), NdbOperation(aNdb),
m_resultSet(0),
m_transConnection(NULL) m_transConnection(NULL)
{ {
theParallelism = 0; theParallelism = 0;
@ -60,21 +59,8 @@ NdbScanOperation::~NdbScanOperation()
theNdb->releaseNdbScanRec(m_receivers[i]); theNdb->releaseNdbScanRec(m_receivers[i]);
} }
delete[] m_array; delete[] m_array;
if (m_resultSet)
delete m_resultSet;
} }
NdbResultSet*
NdbScanOperation::getResultSet()
{
if (!m_resultSet)
m_resultSet = new NdbResultSet(this);
return m_resultSet;
}
void void
NdbScanOperation::setErrorCode(int aErrorCode){ NdbScanOperation::setErrorCode(int aErrorCode){
NdbConnection* tmp = theNdbCon; NdbConnection* tmp = theNdbCon;
@ -125,7 +111,8 @@ NdbScanOperation::init(const NdbTableImpl* tab, NdbConnection* myConnection)
return 0; return 0;
} }
NdbResultSet* NdbScanOperation::readTuples(NdbScanOperation::LockMode lm, int
NdbScanOperation::readTuples(NdbScanOperation::LockMode lm,
Uint32 batch, Uint32 batch,
Uint32 parallel) Uint32 parallel)
{ {
@ -142,7 +129,7 @@ NdbResultSet* NdbScanOperation::readTuples(NdbScanOperation::LockMode lm,
// 3. theScanOp contains a NdbScanOperation // 3. theScanOp contains a NdbScanOperation
if (theNdbCon->theScanningOp != NULL){ if (theNdbCon->theScanningOp != NULL){
setErrorCode(4605); setErrorCode(4605);
return 0; return -1;
} }
theNdbCon->theScanningOp = this; theNdbCon->theScanningOp = this;
@ -167,7 +154,7 @@ NdbResultSet* NdbScanOperation::readTuples(NdbScanOperation::LockMode lm,
break; break;
default: default:
setErrorCode(4003); setErrorCode(4003);
return 0; return -1;
} }
m_keyInfo = lockExcl ? 1 : 0; m_keyInfo = lockExcl ? 1 : 0;
@ -192,13 +179,13 @@ NdbResultSet* NdbScanOperation::readTuples(NdbScanOperation::LockMode lm,
if(fix_receivers(parallel) == -1){ if(fix_receivers(parallel) == -1){
setErrorCodeAbort(4000); setErrorCodeAbort(4000);
return 0; return -1;
} }
theSCAN_TABREQ = (!theSCAN_TABREQ ? theNdb->getSignal() : theSCAN_TABREQ); theSCAN_TABREQ = (!theSCAN_TABREQ ? theNdb->getSignal() : theSCAN_TABREQ);
if (theSCAN_TABREQ == NULL) { if (theSCAN_TABREQ == NULL) {
setErrorCodeAbort(4000); setErrorCodeAbort(4000);
return 0; return -1;
}//if }//if
theSCAN_TABREQ->setSignal(GSN_SCAN_TABREQ); theSCAN_TABREQ->setSignal(GSN_SCAN_TABREQ);
@ -234,7 +221,7 @@ NdbResultSet* NdbScanOperation::readTuples(NdbScanOperation::LockMode lm,
theTotalNrOfKeyWordInSignal= 0; theTotalNrOfKeyWordInSignal= 0;
getFirstATTRINFOScan(); getFirstATTRINFOScan();
return getResultSet(); return 0;
} }
int int
@ -403,6 +390,29 @@ NdbScanOperation::executeCursor(int nodeId){
int NdbScanOperation::nextResult(bool fetchAllowed, bool forceSend) int NdbScanOperation::nextResult(bool fetchAllowed, bool forceSend)
{
int res;
if ((res = nextResultImpl(fetchAllowed, forceSend)) == 0) {
// handle blobs
NdbBlob* tBlob = theBlobList;
while (tBlob != 0) {
if (tBlob->atNextResult() == -1)
return -1;
tBlob = tBlob->theNext;
}
/*
* Flush blob part ops on behalf of user because
* - nextResult is analogous to execute(NoCommit)
* - user is likely to want blob value before next execute
*/
if (m_transConnection->executePendingBlobOps() == -1)
return -1;
return 0;
}
return res;
}
int NdbScanOperation::nextResultImpl(bool fetchAllowed, bool forceSend)
{ {
if(m_ordered) if(m_ordered)
return ((NdbIndexScanOperation*)this)->next_result_ordered(fetchAllowed, return ((NdbIndexScanOperation*)this)->next_result_ordered(fetchAllowed,
@ -622,11 +632,11 @@ NdbScanOperation::doSend(int ProcessorId)
return 0; return 0;
} }
void NdbScanOperation::closeScan(bool forceSend) void NdbScanOperation::close(bool forceSend)
{ {
if(m_transConnection){ if(m_transConnection){
if(DEBUG_NEXT_RESULT) if(DEBUG_NEXT_RESULT)
ndbout_c("closeScan() theError.code = %d " ndbout_c("close() theError.code = %d "
"m_api_receivers_count = %d " "m_api_receivers_count = %d "
"m_conf_receivers_count = %d " "m_conf_receivers_count = %d "
"m_sent_receivers_count = %d", "m_sent_receivers_count = %d",
@ -657,7 +667,7 @@ NdbScanOperation::execCLOSE_SCAN_REP(){
void NdbScanOperation::release() void NdbScanOperation::release()
{ {
if(theNdbCon != 0 || m_transConnection != 0){ if(theNdbCon != 0 || m_transConnection != 0){
closeScan(); close();
} }
for(Uint32 i = 0; i<m_allocated_receivers; i++){ for(Uint32 i = 0; i<m_allocated_receivers; i++){
m_receivers[i]->release(); m_receivers[i]->release();
@ -1196,22 +1206,22 @@ error:
return -1; return -1;
} }
NdbResultSet* int
NdbIndexScanOperation::readTuples(LockMode lm, NdbIndexScanOperation::readTuples(LockMode lm,
Uint32 batch, Uint32 batch,
Uint32 parallel, Uint32 parallel,
bool order_by, bool order_by,
bool read_range_no){ bool read_range_no){
NdbResultSet * rs = NdbScanOperation::readTuples(lm, batch, 0); int res = NdbScanOperation::readTuples(lm, batch, 0);
if(read_range_no) if(read_range_no)
{ {
m_read_range_no = 1; m_read_range_no = 1;
Uint32 word = 0; Uint32 word = 0;
AttributeHeader::init(&word, AttributeHeader::RANGE_NO, 0); AttributeHeader::init(&word, AttributeHeader::RANGE_NO, 0);
if(insertATTRINFO(word) == -1) if(insertATTRINFO(word) == -1)
rs = 0; res = -1;
} }
if(rs && order_by){ if(!res && order_by){
m_ordered = 1; m_ordered = 1;
Uint32 cnt = m_accessTable->getNoOfColumns() - 1; Uint32 cnt = m_accessTable->getNoOfColumns() - 1;
m_sort_columns = cnt; // -1 for NDB$NODE m_sort_columns = cnt; // -1 for NDB$NODE
@ -1234,7 +1244,7 @@ NdbIndexScanOperation::readTuples(LockMode lm,
m_this_bound_start = 0; m_this_bound_start = 0;
m_first_bound_word = theKEYINFOptr; m_first_bound_word = theKEYINFOptr;
return rs; return res;
} }
void void
@ -1666,7 +1676,7 @@ NdbIndexScanOperation::end_of_bound(Uint32 no)
return -1; return -1;
} }
Uint32 int
NdbIndexScanOperation::get_range_no() NdbIndexScanOperation::get_range_no()
{ {
if(m_read_range_no) if(m_read_range_no)
@ -1687,5 +1697,5 @@ NdbIndexScanOperation::get_range_no()
return ret; return ret;
} }
} }
return 0; return -1;
} }

2
ndb/test/include/HugoOperations.hpp
View File

@ -95,7 +95,7 @@ protected:
Vector<BaseString> savedRecords; Vector<BaseString> savedRecords;
struct RsPair { NdbResultSet* m_result_set; int records; }; struct RsPair { NdbScanOperation* m_result_set; int records; };
Vector<RsPair> m_result_sets; Vector<RsPair> m_result_sets;
Vector<RsPair> m_executed_result_sets; Vector<RsPair> m_executed_result_sets;
private: private:

21
ndb/test/ndbapi/ScanFunctions.hpp
View File

@ -81,7 +81,6 @@ ScanFunctions::scanReadFunctions(Ndb* pNdb,
int check; int check;
NdbConnection *pTrans = 0; NdbConnection *pTrans = 0;
NdbScanOperation *pOp = 0; NdbScanOperation *pOp = 0;
NdbResultSet *rs = 0;
while (true){ while (true){
if (retryAttempt >= retryMax){ if (retryAttempt >= retryMax){
@ -111,12 +110,9 @@ ScanFunctions::scanReadFunctions(Ndb* pNdb,
return NDBT_FAILED; return NDBT_FAILED;
} }
if( pOp->readTuples(exclusive ?
rs = pOp->readTuples(exclusive ?
NdbScanOperation::LM_Exclusive : NdbScanOperation::LM_Exclusive :
NdbScanOperation::LM_Read); NdbScanOperation::LM_Read) ) {
if( rs == 0 ) {
ERR(pTrans->getNdbError()); ERR(pTrans->getNdbError());
pNdb->closeTransaction(pTrans); pNdb->closeTransaction(pTrans);
return NDBT_FAILED; return NDBT_FAILED;
@ -125,8 +121,7 @@ ScanFunctions::scanReadFunctions(Ndb* pNdb,
if (action == OnlyOpenScanOnce){ if (action == OnlyOpenScanOnce){
// Call openScan one more time when it's already defined // Call openScan one more time when it's already defined
NdbResultSet* rs2 = pOp->readTuples(NdbScanOperation::LM_Read); if( pOp->readTuples(NdbScanOperation::LM_Read) ) {
if( rs2 == 0 ) {
ERR(pTrans->getNdbError()); ERR(pTrans->getNdbError());
pNdb->closeTransaction(pTrans); pNdb->closeTransaction(pTrans);
return NDBT_FAILED; return NDBT_FAILED;
@ -168,7 +163,7 @@ ScanFunctions::scanReadFunctions(Ndb* pNdb,
bool abortTrans = (action==CloseWithoutStop); bool abortTrans = (action==CloseWithoutStop);
int eof; int eof;
int rows = 0; int rows = 0;
eof = rs->nextResult(); eof = pOp->nextResult();
while(eof == 0){ while(eof == 0){
rows++; rows++;
@ -178,7 +173,7 @@ ScanFunctions::scanReadFunctions(Ndb* pNdb,
if (action != CloseWithoutStop){ if (action != CloseWithoutStop){
// Test that we can closeTrans without stopScan // Test that we can closeTrans without stopScan
rs->close(); pOp->close();
if( check == -1 ) { if( check == -1 ) {
ERR(pTrans->getNdbError()); ERR(pTrans->getNdbError());
pNdb->closeTransaction(pTrans); pNdb->closeTransaction(pTrans);
@ -201,7 +196,7 @@ ScanFunctions::scanReadFunctions(Ndb* pNdb,
} }
} }
eof = rs->nextResult(); eof = pOp->nextResult();
} }
if (eof == -1) { if (eof == -1) {
const NdbError err = pTrans->getNdbError(); const NdbError err = pTrans->getNdbError();
@ -211,7 +206,7 @@ ScanFunctions::scanReadFunctions(Ndb* pNdb,
// Be cruel, call nextScanResult after error // Be cruel, call nextScanResult after error
for(int i=0; i<10; i++){ for(int i=0; i<10; i++){
eof = rs->nextResult(); eof = pOp->nextResult();
if(eof == 0){ if(eof == 0){
g_err << "nextScanResult returned eof = " << eof << endl g_err << "nextScanResult returned eof = " << eof << endl
<< " That is an error when there are no more records" << endl; << " That is an error when there are no more records" << endl;
@ -241,7 +236,7 @@ ScanFunctions::scanReadFunctions(Ndb* pNdb,
if (action == NextScanWhenNoMore){ if (action == NextScanWhenNoMore){
g_info << "Calling nextScanresult when there are no more records" << endl; g_info << "Calling nextScanresult when there are no more records" << endl;
for(int i=0; i<10; i++){ for(int i=0; i<10; i++){
eof = rs->nextResult(); eof = pOp->nextResult();
if(eof == 0){ if(eof == 0){
g_err << "nextScanResult returned eof = " << eof << endl g_err << "nextScanResult returned eof = " << eof << endl
<< " That is an error when there are no more records" << endl; << " That is an error when there are no more records" << endl;

17
ndb/test/ndbapi/ScanInterpretTest.hpp
View File

@ -227,10 +227,7 @@ ScanInterpretTest::scanRead(Ndb* pNdb,
return NDBT_FAILED; return NDBT_FAILED;
} }
NdbResultSet * rs = pOp->readTuples(NdbScanOperation::LM_Read, if( pOp->readTuples(NdbScanOperation::LM_Read, 0, parallelism) ) {
0, parallelism);
if( rs == 0 ) {
ERR(pTrans->getNdbError()); ERR(pTrans->getNdbError());
pNdb->closeTransaction(pTrans); pNdb->closeTransaction(pTrans);
return NDBT_FAILED; return NDBT_FAILED;
@ -262,14 +259,14 @@ ScanInterpretTest::scanRead(Ndb* pNdb,
int rows = 0; int rows = 0;
NdbConnection* pInsTrans; NdbConnection* pInsTrans;
while((eof = rs->nextResult(true)) == 0){ while((eof = pOp->nextResult(true)) == 0){
do { do {
rows++; rows++;
if (addRowToInsert(pNdb, pTrans) != 0){ if (addRowToInsert(pNdb, pTrans) != 0){
pNdb->closeTransaction(pTrans); pNdb->closeTransaction(pTrans);
return NDBT_FAILED; return NDBT_FAILED;
} }
} while((eof = rs->nextResult(false)) == 0); } while((eof = pOp->nextResult(false)) == 0);
check = pTrans->execute(Commit); check = pTrans->execute(Commit);
if( check == -1 ) { if( check == -1 ) {
@ -349,9 +346,7 @@ ScanInterpretTest::scanReadVerify(Ndb* pNdb,
return NDBT_FAILED; return NDBT_FAILED;
} }
NdbResultSet * rs = pOp->readTuples(NdbScanOperation::LM_Read, if( pOp->readTuples(NdbScanOperation::LM_Read, 0, parallelism) ) {
0, parallelism);
if( rs == 0 ) {
ERR(pTrans->getNdbError()); ERR(pTrans->getNdbError());
pNdb->closeTransaction(pTrans); pNdb->closeTransaction(pTrans);
return NDBT_FAILED; return NDBT_FAILED;
@ -392,7 +387,7 @@ ScanInterpretTest::scanReadVerify(Ndb* pNdb,
NdbConnection* pExistTrans; NdbConnection* pExistTrans;
NdbConnection* pNoExistTrans; NdbConnection* pNoExistTrans;
while((eof = rs->nextResult(true)) == 0){ while((eof = pOp->nextResult(true)) == 0){
pExistTrans = pNdb->startTransaction(); pExistTrans = pNdb->startTransaction();
if (pExistTrans == NULL) { if (pExistTrans == NULL) {
const NdbError err = pNdb->getNdbError(); const NdbError err = pNdb->getNdbError();
@ -424,7 +419,7 @@ ScanInterpretTest::scanReadVerify(Ndb* pNdb,
return NDBT_FAILED; return NDBT_FAILED;
} }
} }
} while((eof = rs->nextResult(false)) == 0); } while((eof = pOp->nextResult(false)) == 0);
// Execute the transaction containing reads of // Execute the transaction containing reads of

60
ndb/test/ndbapi/bank/Bank.cpp
View File

@ -669,8 +669,7 @@ int Bank::findLastGL(Uint64 &lastTime){
return NDBT_FAILED; return NDBT_FAILED;
} }
NdbResultSet * rs = pOp->readTuples(); if( pOp->readTuples() ) {
if( rs == 0 ) {
ERR(pScanTrans->getNdbError()); ERR(pScanTrans->getNdbError());
m_ndb.closeTransaction(pScanTrans); m_ndb.closeTransaction(pScanTrans);
return NDBT_FAILED; return NDBT_FAILED;
@ -699,7 +698,7 @@ int Bank::findLastGL(Uint64 &lastTime){
int eof; int eof;
int rows = 0; int rows = 0;
eof = rs->nextResult(); eof = pOp->nextResult();
lastTime = 0; lastTime = 0;
while(eof == 0){ while(eof == 0){
@ -709,7 +708,7 @@ int Bank::findLastGL(Uint64 &lastTime){
if (t > lastTime) if (t > lastTime)
lastTime = t; lastTime = t;
eof = rs->nextResult(); eof = pOp->nextResult();
} }
if (eof == -1) { if (eof == -1) {
ERR(pScanTrans->getNdbError()); ERR(pScanTrans->getNdbError());
@ -1001,8 +1000,7 @@ int Bank::sumTransactionsForGL(const Uint64 glTime,
return NDBT_FAILED; return NDBT_FAILED;
} }
NdbResultSet * rs = pOp->readTuplesExclusive(); if( pOp->readTuplesExclusive()) {
if( rs == 0 ) {
ERR(pScanTrans->getNdbError()); ERR(pScanTrans->getNdbError());
m_ndb.closeTransaction(pScanTrans); m_ndb.closeTransaction(pScanTrans);
return NDBT_FAILED; return NDBT_FAILED;
@ -1053,7 +1051,7 @@ int Bank::sumTransactionsForGL(const Uint64 glTime,
int eof; int eof;
int rows = 0; int rows = 0;
int rowsFound = 0; int rowsFound = 0;
eof = rs->nextResult(); eof = pOp->nextResult();
while(eof == 0){ while(eof == 0){
rows++; rows++;
@ -1077,7 +1075,7 @@ int Bank::sumTransactionsForGL(const Uint64 glTime,
} }
} }
eof = rs->nextResult(); eof = pOp->nextResult();
if ((rows % 100) == 0){ if ((rows % 100) == 0){
// "refresh" ownner transaction every 100th row // "refresh" ownner transaction every 100th row
@ -1161,8 +1159,7 @@ int Bank::performValidateGL(Uint64 glTime){
return NDBT_FAILED; return NDBT_FAILED;
} }
NdbResultSet * rs = pOp->readTuples(); if( pOp->readTuples() ) {
if( rs == 0 ) {
ERR(pScanTrans->getNdbError()); ERR(pScanTrans->getNdbError());
m_ndb.closeTransaction(pScanTrans); m_ndb.closeTransaction(pScanTrans);
return NDBT_FAILED; return NDBT_FAILED;
@ -1241,7 +1238,7 @@ int Bank::performValidateGL(Uint64 glTime){
int rows = 0; int rows = 0;
int countGlRecords = 0; int countGlRecords = 0;
int result = NDBT_OK; int result = NDBT_OK;
eof = rs->nextResult(); eof = pOp->nextResult();
while(eof == 0){ while(eof == 0){
rows++; rows++;
@ -1328,7 +1325,7 @@ int Bank::performValidateGL(Uint64 glTime){
} }
} }
eof = rs->nextResult(); eof = pOp->nextResult();
} }
if (eof == -1) { if (eof == -1) {
ERR(pScanTrans->getNdbError()); ERR(pScanTrans->getNdbError());
@ -1425,8 +1422,7 @@ int Bank::getOldestPurgedGL(const Uint32 accountType,
return NDBT_FAILED; return NDBT_FAILED;
} }
NdbResultSet * rs = pOp->readTuples(); if( pOp->readTuples() ) {
if( rs == 0 ) {
ERR(pScanTrans->getNdbError()); ERR(pScanTrans->getNdbError());
m_ndb.closeTransaction(pScanTrans); m_ndb.closeTransaction(pScanTrans);
return NDBT_FAILED; return NDBT_FAILED;
@ -1469,7 +1465,7 @@ int Bank::getOldestPurgedGL(const Uint32 accountType,
int eof; int eof;
int rows = 0; int rows = 0;
eof = rs->nextResult(); eof = pOp->nextResult();
oldest = 0; oldest = 0;
while(eof == 0){ while(eof == 0){
@ -1483,7 +1479,7 @@ int Bank::getOldestPurgedGL(const Uint32 accountType,
if (t > oldest) if (t > oldest)
oldest = t; oldest = t;
} }
eof = rs->nextResult(); eof = pOp->nextResult();
} }
if (eof == -1) { if (eof == -1) {
ERR(pScanTrans->getNdbError()); ERR(pScanTrans->getNdbError());
@ -1517,8 +1513,7 @@ int Bank::getOldestNotPurgedGL(Uint64 &oldest,
return NDBT_FAILED; return NDBT_FAILED;
} }
NdbResultSet * rs = pOp->readTuples(); if( pOp->readTuples() ) {
if( rs == 0 ) {
ERR(pScanTrans->getNdbError()); ERR(pScanTrans->getNdbError());
m_ndb.closeTransaction(pScanTrans); m_ndb.closeTransaction(pScanTrans);
return NDBT_FAILED; return NDBT_FAILED;
@ -1561,7 +1556,7 @@ int Bank::getOldestNotPurgedGL(Uint64 &oldest,
int eof; int eof;
int rows = 0; int rows = 0;
eof = rs->nextResult(); eof = pOp->nextResult();
oldest = (Uint64)-1; oldest = (Uint64)-1;
found = false; found = false;
@ -1578,7 +1573,7 @@ int Bank::getOldestNotPurgedGL(Uint64 &oldest,
accountTypeId = a; accountTypeId = a;
} }
} }
eof = rs->nextResult(); eof = pOp->nextResult();
} }
if (eof == -1) { if (eof == -1) {
ERR(pScanTrans->getNdbError()); ERR(pScanTrans->getNdbError());
@ -1614,8 +1609,7 @@ int Bank::checkNoTransactionsOlderThan(const Uint32 accountType,
return NDBT_FAILED; return NDBT_FAILED;
} }
NdbResultSet * rs = pOp->readTuples(); if( pOp->readTuples() ) {
if( rs == 0 ) {
ERR(pScanTrans->getNdbError()); ERR(pScanTrans->getNdbError());
m_ndb.closeTransaction(pScanTrans); m_ndb.closeTransaction(pScanTrans);
return NDBT_FAILED; return NDBT_FAILED;
@ -1659,7 +1653,7 @@ int Bank::checkNoTransactionsOlderThan(const Uint32 accountType,
int eof; int eof;
int rows = 0; int rows = 0;
int found = 0; int found = 0;
eof = rs->nextResult(); eof = pOp->nextResult();
while(eof == 0){ while(eof == 0){
rows++; rows++;
@ -1675,7 +1669,7 @@ int Bank::checkNoTransactionsOlderThan(const Uint32 accountType,
<< " ti = " << ti << endl; << " ti = " << ti << endl;
found++; found++;
} }
eof = rs->nextResult(); eof = pOp->nextResult();
} }
if (eof == -1) { if (eof == -1) {
ERR(pScanTrans->getNdbError()); ERR(pScanTrans->getNdbError());
@ -1858,8 +1852,7 @@ int Bank::findTransactionsToPurge(const Uint64 glTime,
return NDBT_FAILED; return NDBT_FAILED;
} }
NdbResultSet * rs = pOp->readTuplesExclusive(); if( pOp->readTuplesExclusive() ) {
if( rs == 0 ) {
ERR(pScanTrans->getNdbError()); ERR(pScanTrans->getNdbError());
m_ndb.closeTransaction(pScanTrans); m_ndb.closeTransaction(pScanTrans);
return NDBT_FAILED; return NDBT_FAILED;
@ -1896,7 +1889,7 @@ int Bank::findTransactionsToPurge(const Uint64 glTime,
int eof; int eof;
int rows = 0; int rows = 0;
int rowsFound = 0; int rowsFound = 0;
eof = rs->nextResult(); eof = pOp->nextResult();
while(eof == 0){ while(eof == 0){
rows++; rows++;
@ -1906,7 +1899,7 @@ int Bank::findTransactionsToPurge(const Uint64 glTime,
if (a == accountType && t == glTime){ if (a == accountType && t == glTime){
rowsFound++; rowsFound++;
// One record found // One record found
check = rs->deleteTuple(pTrans); check = pOp->deleteCurrentTuple(pTrans);
if (check == -1){ if (check == -1){
ERR(m_ndb.getNdbError()); ERR(m_ndb.getNdbError());
m_ndb.closeTransaction(pScanTrans); m_ndb.closeTransaction(pScanTrans);
@ -1921,7 +1914,7 @@ int Bank::findTransactionsToPurge(const Uint64 glTime,
return NDBT_FAILED; return NDBT_FAILED;
} }
} }
eof = rs->nextResult(); eof = pOp->nextResult();
} }
if (eof == -1) { if (eof == -1) {
ERR(pScanTrans->getNdbError()); ERR(pScanTrans->getNdbError());
@ -2367,8 +2360,7 @@ int Bank::getSumAccounts(Uint32 &sumAccounts,
return NDBT_FAILED; return NDBT_FAILED;
} }
NdbResultSet * rs = pOp->readTuplesExclusive(); if( pOp->readTuplesExclusive() ) {
if( rs == 0 ) {
ERR(pScanTrans->getNdbError()); ERR(pScanTrans->getNdbError());
m_ndb.closeTransaction(pScanTrans); m_ndb.closeTransaction(pScanTrans);
return NDBT_FAILED; return NDBT_FAILED;
@ -2403,7 +2395,7 @@ int Bank::getSumAccounts(Uint32 &sumAccounts,
} }
int eof; int eof;
eof = rs->nextResult(); eof = pOp->nextResult();
while(eof == 0){ while(eof == 0){
Uint32 b = balanceRec->u_32_value(); Uint32 b = balanceRec->u_32_value();
@ -2415,7 +2407,7 @@ int Bank::getSumAccounts(Uint32 &sumAccounts,
// << ", sum="<< sumAccounts << endl; // << ", sum="<< sumAccounts << endl;
// Take over the operation so that the lock is kept in db // Take over the operation so that the lock is kept in db
NdbOperation* pLockOp = rs->updateTuple(pTrans); NdbOperation* pLockOp = pOp->updateCurrentTuple(pTrans);
if (pLockOp == NULL){ if (pLockOp == NULL){
ERR(m_ndb.getNdbError()); ERR(m_ndb.getNdbError());
m_ndb.closeTransaction(pScanTrans); m_ndb.closeTransaction(pScanTrans);
@ -2441,7 +2433,7 @@ int Bank::getSumAccounts(Uint32 &sumAccounts,
return NDBT_FAILED; return NDBT_FAILED;
} }
eof = rs->nextResult(); eof = pOp->nextResult();
} }
if (eof == -1) { if (eof == -1) {
ERR(pScanTrans->getNdbError()); ERR(pScanTrans->getNdbError());

7
ndb/test/ndbapi/bank/BankLoad.cpp
View File

@ -342,8 +342,7 @@ int Bank::getBalanceForAccountType(const Uint32 accountType,
return NDBT_FAILED; return NDBT_FAILED;
} }
NdbResultSet* rs = pOp->readTuples(); if( pOp->readTuples() ) {
if( rs == 0 ) {
ERR(pScanTrans->getNdbError()); ERR(pScanTrans->getNdbError());
m_ndb.closeTransaction(pScanTrans); m_ndb.closeTransaction(pScanTrans);
return NDBT_FAILED; return NDBT_FAILED;
@ -379,7 +378,7 @@ int Bank::getBalanceForAccountType(const Uint32 accountType,
int eof; int eof;
int rows = 0; int rows = 0;
eof = rs->nextResult(); eof = pOp->nextResult();
while(eof == 0){ while(eof == 0){
rows++; rows++;
@ -391,7 +390,7 @@ int Bank::getBalanceForAccountType(const Uint32 accountType,
balance += b; balance += b;
} }
eof = rs->nextResult(); eof = pOp->nextResult();
} }
if (eof == -1) { if (eof == -1) {
ERR(pScanTrans->getNdbError()); ERR(pScanTrans->getNdbError());

18
ndb/test/ndbapi/slow_select.cpp
View File

@ -8,18 +8,17 @@ S_Scan {
const char * m_table; const char * m_table;
const char * m_index; const char * m_index;
NdbIndexScanOperation * m_scan; NdbIndexScanOperation * m_scan;
NdbResultSet * m_result;
Uint32 metaid; Uint32 metaid;
Uint32 match_count; Uint32 match_count;
Uint32 row_count; Uint32 row_count;
}; };
static S_Scan g_scans[] = { static S_Scan g_scans[] = {
{ "affiliatestometa", "ind_affiliatestometa", 0, 0, 0, 0, 0 }, { "affiliatestometa", "ind_affiliatestometa", 0, 0, 0, 0 },
{ "media", "metaid", 0, 0, 0, 0, 0 }, { "media", "metaid", 0, 0, 0, 0 },
{ "meta", "PRIMARY", 0, 0, 0, 0, 0 }, { "meta", "PRIMARY", 0, 0, 0, 0 },
{ "artiststometamap", "PRIMARY", 0, 0, 0, 0, 0 }, { "artiststometamap", "PRIMARY", 0, 0, 0, 0 },
{ "subgenrestometamap", "metaid", 0, 0, 0, 0, 0 } { "subgenrestometamap", "metaid", 0, 0, 0, 0 }
}; };
#define require(x) if(!(x)) { ndbout << "LINE: " << __LINE__ << endl;abort(); } #define require(x) if(!(x)) { ndbout << "LINE: " << __LINE__ << endl;abort(); }
@ -58,9 +57,8 @@ main(void){
g_scans[i].m_table); g_scans[i].m_table);
NdbIndexScanOperation* scan = g_scans[i].m_scan; NdbIndexScanOperation* scan = g_scans[i].m_scan;
require(scan); require(scan);
g_scans[i].m_result = scan->readTuples(NdbScanOperation::LM_CommittedRead, require(scan->readTuples(NdbScanOperation::LM_CommittedRead,
0, 0, true); 0, 0, true) == 0);
require(g_scans[i].m_result);
} }
require(!g_scans[0].m_scan->setBound((Uint32)0, require(!g_scans[0].m_scan->setBound((Uint32)0,
@ -125,7 +123,7 @@ main(void){
//ndbout_c("%s - %d", g_scans[i].m_table, g_scans[i].metaid); //ndbout_c("%s - %d", g_scans[i].m_table, g_scans[i].metaid);
for(i = 0; i<prev_F_sz; i++){ for(i = 0; i<prev_F_sz; i++){
int res = F[i]->m_result->nextResult(); int res = F[i]->m_scan->nextResult();
if(res == -1) if(res == -1)
abort(); abort();

34
ndb/test/ndbapi/testBlobs.cpp
View File

@ -739,10 +739,9 @@ verifyBlobTable(const Bcol& b, const Bval& v, Uint32 pk1, bool exists)
NdbRecAttr* ra_pk; NdbRecAttr* ra_pk;
NdbRecAttr* ra_part; NdbRecAttr* ra_part;
NdbRecAttr* ra_data; NdbRecAttr* ra_data;
NdbResultSet* rs;
CHK((g_con = g_ndb->startTransaction()) != 0); CHK((g_con = g_ndb->startTransaction()) != 0);
CHK((g_ops = g_con->getNdbScanOperation(b.m_btname)) != 0); CHK((g_ops = g_con->getNdbScanOperation(b.m_btname)) != 0);
CHK((rs = g_ops->readTuples()) != 0); CHK(g_ops->readTuples() == 0);
CHK((ra_pk = g_ops->getValue("PK")) != 0); CHK((ra_pk = g_ops->getValue("PK")) != 0);
CHK((ra_part = g_ops->getValue("PART")) != 0); CHK((ra_part = g_ops->getValue("PART")) != 0);
CHK((ra_data = g_ops->getValue("DATA")) != 0); CHK((ra_data = g_ops->getValue("DATA")) != 0);
@ -756,7 +755,7 @@ verifyBlobTable(const Bcol& b, const Bval& v, Uint32 pk1, bool exists)
memset(seen, 0, partcount); memset(seen, 0, partcount);
while (1) { while (1) {
int ret; int ret;
CHK((ret = rs->nextResult()) == 0 || ret == 1); CHK((ret = g_ops->nextResult()) == 0 || ret == 1);
if (ret == 1) if (ret == 1)
break; break;
if (pk1 != ra_pk->u_32_value()) if (pk1 != ra_pk->u_32_value())
@ -1104,14 +1103,13 @@ readScan(int style, bool idx)
DBG("--- " << "readScan" << (idx ? "Idx" : "") << " " << stylename[style] << " ---"); DBG("--- " << "readScan" << (idx ? "Idx" : "") << " " << stylename[style] << " ---");
Tup tup; Tup tup;
tup.alloc(); // allocate buffers tup.alloc(); // allocate buffers
NdbResultSet* rs;
CHK((g_con = g_ndb->startTransaction()) != 0); CHK((g_con = g_ndb->startTransaction()) != 0);
if (! idx) { if (! idx) {
CHK((g_ops = g_con->getNdbScanOperation(g_opt.m_tname)) != 0); CHK((g_ops = g_con->getNdbScanOperation(g_opt.m_tname)) != 0);
} else { } else {
CHK((g_ops = g_con->getNdbIndexScanOperation(g_opt.m_x2name, g_opt.m_tname)) != 0); CHK((g_ops = g_con->getNdbIndexScanOperation(g_opt.m_x2name, g_opt.m_tname)) != 0);
} }
CHK((rs = g_ops->readTuples(NdbScanOperation::LM_Read)) != 0); CHK(g_ops->readTuples(NdbScanOperation::LM_Read) == 0);
CHK(g_ops->getValue("PK1", (char*)&tup.m_pk1) != 0); CHK(g_ops->getValue("PK1", (char*)&tup.m_pk1) != 0);
if (g_opt.m_pk2len != 0) if (g_opt.m_pk2len != 0)
CHK(g_ops->getValue("PK2", tup.m_pk2) != 0); CHK(g_ops->getValue("PK2", tup.m_pk2) != 0);
@ -1127,7 +1125,7 @@ readScan(int style, bool idx)
int ret; int ret;
tup.m_pk1 = (Uint32)-1; tup.m_pk1 = (Uint32)-1;
memset(tup.m_pk2, 'x', g_opt.m_pk2len); memset(tup.m_pk2, 'x', g_opt.m_pk2len);
CHK((ret = rs->nextResult(true)) == 0 || ret == 1); CHK((ret = g_ops->nextResult(true)) == 0 || ret == 1);
if (ret == 1) if (ret == 1)
break; break;
DBG("readScan" << (idx ? "Idx" : "") << " pk1=" << hex << tup.m_pk1); DBG("readScan" << (idx ? "Idx" : "") << " pk1=" << hex << tup.m_pk1);
@ -1157,14 +1155,13 @@ updateScan(int style, bool idx)
DBG("--- " << "updateScan" << (idx ? "Idx" : "") << " " << stylename[style] << " ---"); DBG("--- " << "updateScan" << (idx ? "Idx" : "") << " " << stylename[style] << " ---");
Tup tup; Tup tup;
tup.alloc(); // allocate buffers tup.alloc(); // allocate buffers
NdbResultSet* rs;
CHK((g_con = g_ndb->startTransaction()) != 0); CHK((g_con = g_ndb->startTransaction()) != 0);
if (! idx) { if (! idx) {
CHK((g_ops = g_con->getNdbScanOperation(g_opt.m_tname)) != 0); CHK((g_ops = g_con->getNdbScanOperation(g_opt.m_tname)) != 0);
} else { } else {
CHK((g_ops = g_con->getNdbIndexScanOperation(g_opt.m_x2name, g_opt.m_tname)) != 0); CHK((g_ops = g_con->getNdbIndexScanOperation(g_opt.m_x2name, g_opt.m_tname)) != 0);
} }
CHK((rs = g_ops->readTuples(NdbScanOperation::LM_Exclusive)) != 0); CHK(g_ops->readTuples(NdbScanOperation::LM_Exclusive) == 0);
CHK(g_ops->getValue("PK1", (char*)&tup.m_pk1) != 0); CHK(g_ops->getValue("PK1", (char*)&tup.m_pk1) != 0);
if (g_opt.m_pk2len != 0) if (g_opt.m_pk2len != 0)
CHK(g_ops->getValue("PK2", tup.m_pk2) != 0); CHK(g_ops->getValue("PK2", tup.m_pk2) != 0);
@ -1174,7 +1171,7 @@ updateScan(int style, bool idx)
int ret; int ret;
tup.m_pk1 = (Uint32)-1; tup.m_pk1 = (Uint32)-1;
memset(tup.m_pk2, 'x', g_opt.m_pk2len); memset(tup.m_pk2, 'x', g_opt.m_pk2len);
CHK((ret = rs->nextResult(true)) == 0 || ret == 1); CHK((ret = g_ops->nextResult(true)) == 0 || ret == 1);
if (ret == 1) if (ret == 1)
break; break;
DBG("updateScan" << (idx ? "Idx" : "") << " pk1=" << hex << tup.m_pk1); DBG("updateScan" << (idx ? "Idx" : "") << " pk1=" << hex << tup.m_pk1);
@ -1183,7 +1180,7 @@ updateScan(int style, bool idx)
// calculate new blob values // calculate new blob values
calcBval(g_tups[k], false); calcBval(g_tups[k], false);
tup.copyfrom(g_tups[k]); tup.copyfrom(g_tups[k]);
CHK((g_opr = rs->updateTuple()) != 0); CHK((g_opr = g_ops->updateCurrentTuple()) != 0);
CHK(getBlobHandles(g_opr) == 0); CHK(getBlobHandles(g_opr) == 0);
if (style == 0) { if (style == 0) {
CHK(setBlobValue(tup) == 0); CHK(setBlobValue(tup) == 0);
@ -1210,14 +1207,13 @@ deleteScan(bool idx)
{ {
DBG("--- " << "deleteScan" << (idx ? "Idx" : "") << " ---"); DBG("--- " << "deleteScan" << (idx ? "Idx" : "") << " ---");
Tup tup; Tup tup;
NdbResultSet* rs;
CHK((g_con = g_ndb->startTransaction()) != 0); CHK((g_con = g_ndb->startTransaction()) != 0);
if (! idx) { if (! idx) {
CHK((g_ops = g_con->getNdbScanOperation(g_opt.m_tname)) != 0); CHK((g_ops = g_con->getNdbScanOperation(g_opt.m_tname)) != 0);
} else { } else {
CHK((g_ops = g_con->getNdbIndexScanOperation(g_opt.m_x2name, g_opt.m_tname)) != 0); CHK((g_ops = g_con->getNdbIndexScanOperation(g_opt.m_x2name, g_opt.m_tname)) != 0);
} }
CHK((rs = g_ops->readTuples(NdbScanOperation::LM_Exclusive)) != 0); CHK(g_ops->readTuples(NdbScanOperation::LM_Exclusive) == 0);
CHK(g_ops->getValue("PK1", (char*)&tup.m_pk1) != 0); CHK(g_ops->getValue("PK1", (char*)&tup.m_pk1) != 0);
if (g_opt.m_pk2len != 0) if (g_opt.m_pk2len != 0)
CHK(g_ops->getValue("PK2", tup.m_pk2) != 0); CHK(g_ops->getValue("PK2", tup.m_pk2) != 0);
@ -1227,11 +1223,11 @@ deleteScan(bool idx)
int ret; int ret;
tup.m_pk1 = (Uint32)-1; tup.m_pk1 = (Uint32)-1;
memset(tup.m_pk2, 'x', g_opt.m_pk2len); memset(tup.m_pk2, 'x', g_opt.m_pk2len);
CHK((ret = rs->nextResult()) == 0 || ret == 1); CHK((ret = g_ops->nextResult()) == 0 || ret == 1);
if (ret == 1) if (ret == 1)
break; break;
DBG("deleteScan" << (idx ? "Idx" : "") << " pk1=" << hex << tup.m_pk1); DBG("deleteScan" << (idx ? "Idx" : "") << " pk1=" << hex << tup.m_pk1);
CHK(rs->deleteTuple() == 0); CHK(g_ops->deleteCurrentTuple() == 0);
CHK(g_con->execute(NoCommit) == 0); CHK(g_con->execute(NoCommit) == 0);
Uint32 k = tup.m_pk1 - g_opt.m_pk1off; Uint32 k = tup.m_pk1 - g_opt.m_pk1off;
CHK(k < g_opt.m_rows && g_tups[k].m_exists); CHK(k < g_opt.m_rows && g_tups[k].m_exists);
@ -1608,12 +1604,11 @@ testperf()
// scan read char // scan read char
{ {
DBG("--- scan read char ---"); DBG("--- scan read char ---");
NdbResultSet* rs;
Uint32 a; Uint32 a;
char b[20]; char b[20];
CHK((g_con = g_ndb->startTransaction()) != 0); CHK((g_con = g_ndb->startTransaction()) != 0);
CHK((g_ops = g_con->getNdbScanOperation(tab.getName())) != 0); CHK((g_ops = g_con->getNdbScanOperation(tab.getName())) != 0);
CHK((rs = g_ops->readTuples(NdbScanOperation::LM_Read)) != 0); CHK(g_ops->readTuples(NdbScanOperation::LM_Read) == 0);
CHK(g_ops->getValue(cA, (char*)&a) != 0); CHK(g_ops->getValue(cA, (char*)&a) != 0);
CHK(g_ops->getValue(cB, b) != 0); CHK(g_ops->getValue(cB, b) != 0);
CHK(g_con->execute(NoCommit) == 0); CHK(g_con->execute(NoCommit) == 0);
@ -1623,7 +1618,7 @@ testperf()
a = (Uint32)-1; a = (Uint32)-1;
b[0] = 0; b[0] = 0;
int ret; int ret;
CHK((ret = rs->nextResult(true)) == 0 || ret == 1); CHK((ret = g_ops->nextResult(true)) == 0 || ret == 1);
if (ret == 1) if (ret == 1)
break; break;
CHK(a < g_opt.m_rowsperf && strcmp(b, "b") == 0); CHK(a < g_opt.m_rowsperf && strcmp(b, "b") == 0);
@ -1638,12 +1633,11 @@ testperf()
// scan read text // scan read text
{ {
DBG("--- read text ---"); DBG("--- read text ---");
NdbResultSet* rs;
Uint32 a; Uint32 a;
char c[20]; char c[20];
CHK((g_con = g_ndb->startTransaction()) != 0); CHK((g_con = g_ndb->startTransaction()) != 0);
CHK((g_ops = g_con->getNdbScanOperation(tab.getName())) != 0); CHK((g_ops = g_con->getNdbScanOperation(tab.getName())) != 0);
CHK((rs = g_ops->readTuples(NdbScanOperation::LM_Read)) != 0); CHK(g_ops->readTuples(NdbScanOperation::LM_Read) == 0);
CHK(g_ops->getValue(cA, (char*)&a) != 0); CHK(g_ops->getValue(cA, (char*)&a) != 0);
CHK((g_bh1 = g_ops->getBlobHandle(cC)) != 0); CHK((g_bh1 = g_ops->getBlobHandle(cC)) != 0);
CHK(g_con->execute(NoCommit) == 0); CHK(g_con->execute(NoCommit) == 0);
@ -1653,7 +1647,7 @@ testperf()
a = (Uint32)-1; a = (Uint32)-1;
c[0] = 0; c[0] = 0;
int ret; int ret;
CHK((ret = rs->nextResult(true)) == 0 || ret == 1); CHK((ret = g_ops->nextResult(true)) == 0 || ret == 1);
if (ret == 1) if (ret == 1)
break; break;
Uint32 m = 20; Uint32 m = 20;

4
ndb/test/ndbapi/testDataBuffers.cpp
View File

@ -444,7 +444,7 @@ testcase(int flag)
return ndberror("startTransaction key=%d", key); return ndberror("startTransaction key=%d", key);
if ((op = sop = con->getNdbScanOperation(tab)) == 0) if ((op = sop = con->getNdbScanOperation(tab)) == 0)
return ndberror("getNdbOperation key=%d", key); return ndberror("getNdbOperation key=%d", key);
if ((rs = sop->readTuples(1)) == 0) if (sop->readTuples(1))
return ndberror("openScanRead key=%d", key); return ndberror("openScanRead key=%d", key);
{ {
col& c = ccol[0]; col& c = ccol[0];
@ -488,7 +488,7 @@ testcase(int flag)
if (con->execute(NoCommit) < 0) if (con->execute(NoCommit) < 0)
return ndberror("executeScan key=%d", key); return ndberror("executeScan key=%d", key);
int ret, cnt = 0; int ret, cnt = 0;
while ((ret = rs->nextResult()) == 0) { while ((ret = sop->nextResult()) == 0) {
if (key != newkey) if (key != newkey)
return ndberror("unexpected key=%d newkey=%d", key, newkey); return ndberror("unexpected key=%d newkey=%d", key, newkey);
for (i = 1; i < attrcnt; i++) { for (i = 1; i < attrcnt; i++) {

11
ndb/test/ndbapi/testDeadlock.cpp
View File

@ -91,7 +91,6 @@ struct Thr {
NdbConnection* m_con; NdbConnection* m_con;
NdbScanOperation* m_scanop; NdbScanOperation* m_scanop;
NdbIndexScanOperation* m_indexscanop; NdbIndexScanOperation* m_indexscanop;
NdbResultSet* m_rs;
// //
Thr(int no); Thr(int no);
~Thr(); ~Thr();
@ -136,7 +135,6 @@ Thr::Thr(int no)
m_con = 0; m_con = 0;
m_scanop = 0; m_scanop = 0;
m_indexscanop = 0; m_indexscanop = 0;
m_rs = 0;
} }
Thr::~Thr() Thr::~Thr()
@ -374,7 +372,7 @@ wl1822_tx2_scanXY(Thr& thr)
CHN(con, (scanop = thr.m_scanop = indexscanop = thr.m_indexscanop = con->getNdbIndexScanOperation(g_opt.m_xname, g_opt.m_tname)) != 0); CHN(con, (scanop = thr.m_scanop = indexscanop = thr.m_indexscanop = con->getNdbIndexScanOperation(g_opt.m_xname, g_opt.m_tname)) != 0);
DBG("tx2 scan exclusive " << g_opt.m_xname); DBG("tx2 scan exclusive " << g_opt.m_xname);
} }
CHN(scanop, (rs = thr.m_rs = scanop->readTuplesExclusive(16)) != 0); CHN(scanop, scanop->readTuplesExclusive(16) == 0);
CHN(scanop, scanop->getValue("A", (char*)&wl1822_bufA) != 0); CHN(scanop, scanop->getValue("A", (char*)&wl1822_bufA) != 0);
CHN(scanop, scanop->getValue("B", (char*)&wl1822_bufB) != 0); CHN(scanop, scanop->getValue("B", (char*)&wl1822_bufB) != 0);
CHN(con, con->execute(NoCommit) == 0); CHN(con, con->execute(NoCommit) == 0);
@ -383,7 +381,7 @@ wl1822_tx2_scanXY(Thr& thr)
DBG("before row " << row); DBG("before row " << row);
int ret; int ret;
wl1822_bufA = wl1822_bufB = ~0; wl1822_bufA = wl1822_bufB = ~0;
CHN(con, (ret = rs->nextResult(true)) == 0); CHN(con, (ret = scanop->nextResult(true)) == 0);
DBG("got row " << row << " a=" << wl1822_bufA << " b=" << wl1822_bufB); DBG("got row " << row << " a=" << wl1822_bufA << " b=" << wl1822_bufB);
CHK(wl1822_bufA == wl1822_valA[wl1822_r2k[row]]); CHK(wl1822_bufA == wl1822_valA[wl1822_r2k[row]]);
CHK(wl1822_bufB == wl1822_valB[wl1822_r2k[row]]); CHK(wl1822_bufB == wl1822_valB[wl1822_r2k[row]]);
@ -419,14 +417,13 @@ wl1822_tx2_scanZ_close(Thr& thr)
Ndb* ndb = thr.m_ndb; Ndb* ndb = thr.m_ndb;
NdbConnection* con = thr.m_con; NdbConnection* con = thr.m_con;
NdbScanOperation* scanop = thr.m_scanop; NdbScanOperation* scanop = thr.m_scanop;
NdbResultSet* rs = thr.m_rs; assert(ndb != 0 && con != 0 && scanop != 0);
assert(ndb != 0 && con != 0 && scanop != 0 && rs != 0);
unsigned row = 2; unsigned row = 2;
while (true) { while (true) {
DBG("before row " << row); DBG("before row " << row);
int ret; int ret;
wl1822_bufA = wl1822_bufB = ~0; wl1822_bufA = wl1822_bufB = ~0;
CHN(con, (ret = rs->nextResult(true)) == 0 || ret == 1); CHN(con, (ret = scanop->nextResult(true)) == 0 || ret == 1);
if (ret == 1) if (ret == 1)
break; break;
DBG("got row " << row << " a=" << wl1822_bufA << " b=" << wl1822_bufB); DBG("got row " << row << " a=" << wl1822_bufA << " b=" << wl1822_bufB);

10
ndb/test/ndbapi/testIndex.cpp
View File

@ -1145,20 +1145,18 @@ runUniqueNullTransactions(NDBT_Context* ctx, NDBT_Step* step){
pTrans = pNdb->startTransaction(); pTrans = pNdb->startTransaction();
NdbScanOperation * sOp; NdbScanOperation * sOp;
NdbOperation * uOp; NdbOperation * uOp;
NdbResultSet * rs;
int eof; int eof;
if(!pTrans) goto done; if(!pTrans) goto done;
sOp = pTrans->getNdbScanOperation(pTab->getName()); sOp = pTrans->getNdbScanOperation(pTab->getName());
if(!sOp) goto done; if(!sOp) goto done;
rs = sOp->readTuples(NdbScanOperation::LM_Exclusive); if(sOp->readTuples(NdbScanOperation::LM_Exclusive)) goto done;
if(!rs) goto done;
if(pTrans->execute(NoCommit) == -1) goto done; if(pTrans->execute(NoCommit) == -1) goto done;
while((eof = rs->nextResult(true)) == 0){ while((eof = sOp->nextResult(true)) == 0){
do { do {
NdbOperation * uOp = rs->updateTuple(); NdbOperation * uOp = sOp->updateCurrentTuple();
if(uOp == 0) goto done; if(uOp == 0) goto done;
uOp->setValue(colId, 0); uOp->setValue(colId, 0);
} while((eof = rs->nextResult(false)) == 0); } while((eof = sOp->nextResult(false)) == 0);
eof = pTrans->execute(Commit); eof = pTrans->execute(Commit);
if(eof == -1) goto done; if(eof == -1) goto done;
} }

23
ndb/test/ndbapi/testOIBasic.cpp
View File

@ -679,14 +679,13 @@ struct Con {
NdbOperation* m_op; NdbOperation* m_op;
NdbScanOperation* m_scanop; NdbScanOperation* m_scanop;
NdbIndexScanOperation* m_indexscanop; NdbIndexScanOperation* m_indexscanop;
NdbResultSet* m_resultset;
enum ScanMode { ScanNo = 0, Committed, Latest, Exclusive }; enum ScanMode { ScanNo = 0, Committed, Latest, Exclusive };
ScanMode m_scanmode; ScanMode m_scanmode;
enum ErrType { ErrNone = 0, ErrDeadlock, ErrOther }; enum ErrType { ErrNone = 0, ErrDeadlock, ErrOther };
ErrType m_errtype; ErrType m_errtype;
Con() : Con() :
m_ndb(0), m_dic(0), m_tx(0), m_op(0), m_ndb(0), m_dic(0), m_tx(0), m_op(0),
m_scanop(0), m_indexscanop(0), m_resultset(0), m_scanmode(ScanNo), m_errtype(ErrNone) {} m_scanop(0), m_indexscanop(0), m_scanmode(ScanNo), m_errtype(ErrNone) {}
~Con() { ~Con() {
if (m_tx != 0) if (m_tx != 0)
closeTransaction(); closeTransaction();
@ -836,7 +835,7 @@ Con::openScanRead(unsigned scanbat, unsigned scanpar)
{ {
assert(m_tx != 0 && m_op != 0); assert(m_tx != 0 && m_op != 0);
NdbOperation::LockMode lm = NdbOperation::LM_Read; NdbOperation::LockMode lm = NdbOperation::LM_Read;
CHKCON((m_resultset = m_scanop->readTuples(lm, scanbat, scanpar)) != 0, *this); CHKCON(m_scanop->readTuples(lm, scanbat, scanpar) == 0, *this);
return 0; return 0;
} }
@ -845,7 +844,7 @@ Con::openScanExclusive(unsigned scanbat, unsigned scanpar)
{ {
assert(m_tx != 0 && m_op != 0); assert(m_tx != 0 && m_op != 0);
NdbOperation::LockMode lm = NdbOperation::LM_Exclusive; NdbOperation::LockMode lm = NdbOperation::LM_Exclusive;
CHKCON((m_resultset = m_scanop->readTuples(lm, scanbat, scanpar)) != 0, *this); CHKCON(m_scanop->readTuples(lm, scanbat, scanpar) == 0, *this);
return 0; return 0;
} }
@ -860,8 +859,8 @@ int
Con::nextScanResult(bool fetchAllowed) Con::nextScanResult(bool fetchAllowed)
{ {
int ret; int ret;
assert(m_resultset != 0); assert(m_scanop != 0);
CHKCON((ret = m_resultset->nextResult(fetchAllowed)) != -1, *this); CHKCON((ret = m_scanop->nextResult(fetchAllowed)) != -1, *this);
assert(ret == 0 || ret == 1 || (! fetchAllowed && ret == 2)); assert(ret == 0 || ret == 1 || (! fetchAllowed && ret == 2));
return ret; return ret;
} }
@ -886,7 +885,7 @@ int
Con::updateScanTuple(Con& con2) Con::updateScanTuple(Con& con2)
{ {
assert(con2.m_tx != 0); assert(con2.m_tx != 0);
CHKCON((con2.m_op = m_resultset->updateTuple(con2.m_tx)) != 0, *this); CHKCON((con2.m_op = m_scanop->updateCurrentTuple(con2.m_tx)) != 0, *this);
return 0; return 0;
} }
@ -894,16 +893,16 @@ int
Con::deleteScanTuple(Con& con2) Con::deleteScanTuple(Con& con2)
{ {
assert(con2.m_tx != 0); assert(con2.m_tx != 0);
CHKCON(m_resultset->deleteTuple(con2.m_tx) == 0, *this); CHKCON(m_scanop->deleteCurrentTuple(con2.m_tx) == 0, *this);
return 0; return 0;
} }
void void
Con::closeScan() Con::closeScan()
{ {
assert(m_resultset != 0); assert(m_scanop != 0);
m_resultset->close(); m_scanop->close();
m_scanop = 0, m_indexscanop = 0, m_resultset = 0; m_scanop = 0, m_indexscanop = 0;
} }
@ -913,7 +912,7 @@ Con::closeTransaction()
assert(m_ndb != 0 && m_tx != 0); assert(m_ndb != 0 && m_tx != 0);
m_ndb->closeTransaction(m_tx); m_ndb->closeTransaction(m_tx);
m_tx = 0, m_op = 0; m_tx = 0, m_op = 0;
m_scanop = 0, m_indexscanop = 0, m_resultset = 0; m_scanop = 0, m_indexscanop = 0;
} }
void void

20
ndb/test/ndbapi/testReadPerf.cpp
View File

@ -266,7 +266,6 @@ run_read(){
NdbScanOperation * pSp; NdbScanOperation * pSp;
NdbIndexOperation * pUp; NdbIndexOperation * pUp;
NdbIndexScanOperation * pIp; NdbIndexScanOperation * pIp;
NdbResultSet * rs = (NdbResultSet*)~0;
Uint32 start_row = rand() % (rows - range); Uint32 start_row = rand() % (rows - range);
Uint32 stop_row = start_row + range; Uint32 stop_row = start_row + range;
@ -319,27 +318,27 @@ run_read(){
} }
break; break;
case 4: case 4:
pOp = pIp = pTrans->getNdbIndexScanOperation(g_ordered,g_table); pOp = pSp = pIp = pTrans->getNdbIndexScanOperation(g_ordered,g_table);
rs = pIp->readTuples(NdbScanOperation::LM_CommittedRead, 0, 0); pIp->readTuples(NdbScanOperation::LM_CommittedRead, 0, 0);
check = pIp->setBound(pk, NdbIndexScanOperation::BoundEQ, &start_row); check = pIp->setBound(pk, NdbIndexScanOperation::BoundEQ, &start_row);
break; break;
case 5: case 5:
pOp = pIp = pTrans->getNdbIndexScanOperation(g_ordered,g_table); pOp = pSp = pIp = pTrans->getNdbIndexScanOperation(g_ordered,g_table);
rs = pIp->readTuples(NdbScanOperation::LM_CommittedRead, 0, 0); pIp->readTuples(NdbScanOperation::LM_CommittedRead, 0, 0);
check = pIp->setBound(pk, NdbIndexScanOperation::BoundLE, &start_row); check = pIp->setBound(pk, NdbIndexScanOperation::BoundLE, &start_row);
check = pIp->setBound(pk, NdbIndexScanOperation::BoundGT, &stop_row); check = pIp->setBound(pk, NdbIndexScanOperation::BoundGT, &stop_row);
start_row = stop_row; start_row = stop_row;
break; break;
case 6: case 6:
pOp = pIp = pTrans->getNdbIndexScanOperation(g_ordered,g_table); pOp = pSp = pIp = pTrans->getNdbIndexScanOperation(g_ordered,g_table);
rs = pIp->readTuples(NdbScanOperation::LM_CommittedRead, 0, 0, true); pIp->readTuples(NdbScanOperation::LM_CommittedRead, 0, 0, true);
check = pIp->setBound(pk, NdbIndexScanOperation::BoundLE, &start_row); check = pIp->setBound(pk, NdbIndexScanOperation::BoundLE, &start_row);
check = pIp->setBound(pk, NdbIndexScanOperation::BoundGT, &stop_row); check = pIp->setBound(pk, NdbIndexScanOperation::BoundGT, &stop_row);
start_row = stop_row; start_row = stop_row;
break; break;
case 7: case 7:
pOp = pSp = pTrans->getNdbScanOperation(g_table); pOp = pSp = pTrans->getNdbScanOperation(g_table);
rs = pSp->readTuples(NdbScanOperation::LM_CommittedRead, 0, 0); pSp->readTuples(NdbScanOperation::LM_CommittedRead, 0, 0);
NdbScanFilter filter(pOp) ; NdbScanFilter filter(pOp) ;
filter.begin(NdbScanFilter::AND); filter.begin(NdbScanFilter::AND);
filter.ge(pk, start_row); filter.ge(pk, start_row);
@ -355,7 +354,6 @@ run_read(){
ndbout << pTrans->getNdbError() << endl; ndbout << pTrans->getNdbError() << endl;
} }
assert(check == 0); assert(check == 0);
assert(rs);
for(int j = 0; j<g_tab->getNoOfColumns(); j++){ for(int j = 0; j<g_tab->getNoOfColumns(); j++){
res = pOp->getValue(j); res = pOp->getValue(j);
@ -368,7 +366,7 @@ run_read(){
} }
assert(check == 0); assert(check == 0);
if(g_paramters[P_OPER].value >= 4){ if(g_paramters[P_OPER].value >= 4){
while((check = rs->nextResult(true)) == 0){ while((check = pSp->nextResult(true)) == 0){
cnt++; cnt++;
} }
@ -377,7 +375,7 @@ run_read(){
return -1; return -1;
} }
assert(check == 1); assert(check == 1);
rs->close(); pSp->close();
} }
} }
assert(g_paramters[P_OPER].value < 4 || (cnt == range)); assert(g_paramters[P_OPER].value < 4 || (cnt == range));

9
ndb/test/ndbapi/testScan.cpp
View File

@ -999,8 +999,7 @@ int runScanRestart(NDBT_Context* ctx, NDBT_Step* step){
return NDBT_FAILED; return NDBT_FAILED;
} }
NdbResultSet* rs = pOp->readTuples(); if( pOp->readTuples() ) {
if( rs == 0 ) {
ERR(pCon->getNdbError()); ERR(pCon->getNdbError());
return NDBT_FAILED; return NDBT_FAILED;
} }
@ -1028,7 +1027,7 @@ int runScanRestart(NDBT_Context* ctx, NDBT_Step* step){
int res; int res;
int row = 0; int row = 0;
while(row < record && (res = rs->nextResult()) == 0) { while(row < record && (res = pOp->nextResult()) == 0) {
if(calc.verifyRowValues(&tmpRow) != 0){ if(calc.verifyRowValues(&tmpRow) != 0){
abort(); abort();
return NDBT_FAILED; return NDBT_FAILED;
@ -1041,14 +1040,14 @@ int runScanRestart(NDBT_Context* ctx, NDBT_Step* step){
return NDBT_FAILED; return NDBT_FAILED;
} }
g_info << " restarting" << endl; g_info << " restarting" << endl;
if((res = rs->restart()) != 0){ if((res = pOp->restart()) != 0){
ERR(pCon->getNdbError()); ERR(pCon->getNdbError());
abort(); abort();
return NDBT_FAILED; return NDBT_FAILED;
} }
row = 0; row = 0;
while((res = rs->nextResult()) == 0) { while((res = pOp->nextResult()) == 0) {
if(calc.verifyRowValues(&tmpRow) != 0){ if(calc.verifyRowValues(&tmpRow) != 0){
abort(); abort();
return NDBT_FAILED; return NDBT_FAILED;

12
ndb/test/ndbapi/testScanPerf.cpp
View File

@ -198,7 +198,6 @@ run_scan(){
NdbScanOperation * pOp = 0; NdbScanOperation * pOp = 0;
NdbIndexScanOperation * pIOp = 0; NdbIndexScanOperation * pIOp = 0;
NdbConnection * pTrans = 0; NdbConnection * pTrans = 0;
NdbResultSet * rs = 0;
int check = 0; int check = 0;
for(int i = 0; i<iter; i++){ for(int i = 0; i<iter; i++){
@ -230,13 +229,13 @@ run_scan(){
if(g_paramters[P_ACCESS].value == 0){ if(g_paramters[P_ACCESS].value == 0){
pOp = pTrans->getNdbScanOperation(g_tablename); pOp = pTrans->getNdbScanOperation(g_tablename);
assert(pOp); assert(pOp);
rs = pOp->readTuples(lm, bat, par); pOp->readTuples(lm, bat, par);
} else { } else {
if(g_paramters[P_RESET].value == 0 || pIOp == 0) if(g_paramters[P_RESET].value == 0 || pIOp == 0)
{ {
pOp= pIOp= pTrans->getNdbIndexScanOperation(g_indexname, g_tablename); pOp= pIOp= pTrans->getNdbIndexScanOperation(g_indexname, g_tablename);
bool ord = g_paramters[P_ACCESS].value == 2; bool ord = g_paramters[P_ACCESS].value == 2;
rs = pIOp->readTuples(lm, bat, par, ord); pIOp->readTuples(lm, bat, par, ord);
} }
else else
{ {
@ -270,7 +269,7 @@ run_scan(){
{ {
int row = rand() % tot; int row = rand() % tot;
pIOp->setBound((Uint32)0, NdbIndexScanOperation::BoundEQ, &row); pIOp->setBound((Uint32)0, NdbIndexScanOperation::BoundEQ, &row);
pIOp->end_of_bound(); pIOp->end_of_bound(i);
} }
if(g_paramters[P_RESET].value == 2) if(g_paramters[P_RESET].value == 2)
goto execute; goto execute;
@ -279,7 +278,6 @@ run_scan(){
} }
} }
assert(pOp); assert(pOp);
assert(rs);
switch(g_paramters[P_FILT].value){ switch(g_paramters[P_FILT].value){
case 0: // All case 0: // All
@ -323,10 +321,10 @@ execute:
check = pTrans->execute(NoCommit); check = pTrans->execute(NoCommit);
assert(check == 0); assert(check == 0);
int fetch = g_paramters[P_FETCH].value; int fetch = g_paramters[P_FETCH].value;
while((check = rs->nextResult(true)) == 0){ while((check = pOp->nextResult(true)) == 0){
do { do {
rows++; rows++;
} while(!fetch && ((check = rs->nextResult(false)) == 0)); } while(!fetch && ((check = pOp->nextResult(false)) == 0));
if(check == -1){ if(check == -1){
err(pTrans->getNdbError()); err(pTrans->getNdbError());
return -1; return -1;

10
ndb/test/src/HugoOperations.cpp
View File

@ -305,7 +305,7 @@ int HugoOperations::execute_Commit(Ndb* pNdb,
m_executed_result_sets.push_back(m_result_sets[i]); m_executed_result_sets.push_back(m_result_sets[i]);
int rows = m_result_sets[i].records; int rows = m_result_sets[i].records;
NdbResultSet* rs = m_result_sets[i].m_result_set; NdbScanOperation* rs = m_result_sets[i].m_result_set;
int res = rs->nextResult(); int res = rs->nextResult();
switch(res){ switch(res){
case 1: case 1:
@ -354,7 +354,7 @@ int HugoOperations::execute_NoCommit(Ndb* pNdb, AbortOption eao){
m_executed_result_sets.push_back(m_result_sets[i]); m_executed_result_sets.push_back(m_result_sets[i]);
int rows = m_result_sets[i].records; int rows = m_result_sets[i].records;
NdbResultSet* rs = m_result_sets[i].m_result_set; NdbScanOperation* rs = m_result_sets[i].m_result_set;
int res = rs->nextResult(); int res = rs->nextResult();
switch(res){ switch(res){
case 1: case 1:
@ -700,9 +700,7 @@ HugoOperations::scanReadRecords(Ndb* pNdb, NdbScanOperation::LockMode lm,
if(!pOp) if(!pOp)
return -1; return -1;
NdbResultSet * rs = pOp->readTuples(lm, 1, 1); if(pOp->readTuples(lm, 1, 1)){
if(!rs){
return -1; return -1;
} }
@ -714,7 +712,7 @@ HugoOperations::scanReadRecords(Ndb* pNdb, NdbScanOperation::LockMode lm,
} }
} }
RsPair p = {rs, records}; RsPair p = {pOp, records};
m_result_sets.push_back(p); m_result_sets.push_back(p);
return 0; return 0;

41
ndb/test/src/HugoTransactions.cpp
View File

@ -72,10 +72,7 @@ HugoTransactions::scanReadRecords(Ndb* pNdb,
return NDBT_FAILED; return NDBT_FAILED;
} }
NdbResultSet * rs; if( pOp ->readTuples(lm) ) {
rs = pOp ->readTuples(lm);
if( rs == 0 ) {
ERR(pTrans->getNdbError()); ERR(pTrans->getNdbError());
pNdb->closeTransaction(pTrans); pNdb->closeTransaction(pTrans);
return NDBT_FAILED; return NDBT_FAILED;
@ -124,7 +121,7 @@ HugoTransactions::scanReadRecords(Ndb* pNdb,
int eof; int eof;
int rows = 0; int rows = 0;
while((eof = rs->nextResult(true)) == 0){ while((eof = pOp->nextResult(true)) == 0){
rows++; rows++;
if (calc.verifyRowValues(&row) != 0){ if (calc.verifyRowValues(&row) != 0){
pNdb->closeTransaction(pTrans); pNdb->closeTransaction(pTrans);
@ -134,7 +131,7 @@ HugoTransactions::scanReadRecords(Ndb* pNdb,
if (abortCount == rows && abortTrans == true){ if (abortCount == rows && abortTrans == true){
ndbout << "Scan is aborted" << endl; ndbout << "Scan is aborted" << endl;
g_info << "Scan is aborted" << endl; g_info << "Scan is aborted" << endl;
rs->close(); pOp->close();
if( check == -1 ) { if( check == -1 ) {
ERR(pTrans->getNdbError()); ERR(pTrans->getNdbError());
pNdb->closeTransaction(pTrans); pNdb->closeTransaction(pTrans);
@ -228,10 +225,7 @@ HugoTransactions::scanReadRecords(Ndb* pNdb,
return NDBT_FAILED; return NDBT_FAILED;
} }
NdbResultSet * rs; if( pOp ->readTuples(lm, 0, parallelism, sorted) ) {
rs = pOp ->readTuples(lm, 0, parallelism, sorted);
if( rs == 0 ) {
ERR(pTrans->getNdbError()); ERR(pTrans->getNdbError());
pNdb->closeTransaction(pTrans); pNdb->closeTransaction(pTrans);
return NDBT_FAILED; return NDBT_FAILED;
@ -280,7 +274,7 @@ HugoTransactions::scanReadRecords(Ndb* pNdb,
int eof; int eof;
int rows = 0; int rows = 0;
while((eof = rs->nextResult(true)) == 0){ while((eof = pOp->nextResult(true)) == 0){
rows++; rows++;
if (calc.verifyRowValues(&row) != 0){ if (calc.verifyRowValues(&row) != 0){
pNdb->closeTransaction(pTrans); pNdb->closeTransaction(pTrans);
@ -290,7 +284,7 @@ HugoTransactions::scanReadRecords(Ndb* pNdb,
if (abortCount == rows && abortTrans == true){ if (abortCount == rows && abortTrans == true){
ndbout << "Scan is aborted" << endl; ndbout << "Scan is aborted" << endl;
g_info << "Scan is aborted" << endl; g_info << "Scan is aborted" << endl;
rs->close(); pOp->close();
if( check == -1 ) { if( check == -1 ) {
ERR(pTrans->getNdbError()); ERR(pTrans->getNdbError());
pNdb->closeTransaction(pTrans); pNdb->closeTransaction(pTrans);
@ -732,8 +726,7 @@ HugoTransactions::scanUpdateRecords3(Ndb* pNdb,
return NDBT_FAILED; return NDBT_FAILED;
} }
NdbResultSet *rs = pOp->readTuplesExclusive(parallelism); if( pOp->readTuplesExclusive(parallelism) ) {
if( rs == 0 ) {
ERR(pTrans->getNdbError()); ERR(pTrans->getNdbError());
pNdb->closeTransaction(pTrans); pNdb->closeTransaction(pTrans);
return NDBT_FAILED; return NDBT_FAILED;
@ -771,10 +764,10 @@ HugoTransactions::scanUpdateRecords3(Ndb* pNdb,
} }
int rows = 0; int rows = 0;
while((check = rs->nextResult(true)) == 0){ while((check = pOp->nextResult(true)) == 0){
do { do {
rows++; rows++;
NdbOperation* pUp = rs->updateTuple(); NdbOperation* pUp = pOp->updateCurrentTuple();
if(pUp == 0){ if(pUp == 0){
ERR(pTrans->getNdbError()); ERR(pTrans->getNdbError());
pNdb->closeTransaction(pTrans); pNdb->closeTransaction(pTrans);
@ -798,7 +791,7 @@ HugoTransactions::scanUpdateRecords3(Ndb* pNdb,
pNdb->closeTransaction(pTrans); pNdb->closeTransaction(pTrans);
return NDBT_OK; return NDBT_OK;
} }
} while((check = rs->nextResult(false)) == 0); } while((check = pOp->nextResult(false)) == 0);
if(check != -1){ if(check != -1){
check = pTrans->execute(Commit); check = pTrans->execute(Commit);
@ -2133,7 +2126,6 @@ HugoTransactions::indexReadRecords(Ndb* pNdb,
NdbConnection *pTrans; NdbConnection *pTrans;
NdbOperation *pOp; NdbOperation *pOp;
NdbIndexScanOperation *sOp; NdbIndexScanOperation *sOp;
NdbResultSet * rs;
const NdbDictionary::Index* pIndex const NdbDictionary::Index* pIndex
= pNdb->getDictionary()->getIndex(idxName, tab.getName()); = pNdb->getDictionary()->getIndex(idxName, tab.getName());
@ -2189,9 +2181,7 @@ HugoTransactions::indexReadRecords(Ndb* pNdb,
pNdb->closeTransaction(pTrans); pNdb->closeTransaction(pTrans);
return NDBT_FAILED; return NDBT_FAILED;
} }
check = 0; check = 0;
rs = sOp->readTuples();
} }
if( check == -1 ) { if( check == -1 ) {
@ -2223,7 +2213,7 @@ HugoTransactions::indexReadRecords(Ndb* pNdb,
} }
check = pTrans->execute(Commit); check = pTrans->execute(Commit);
check = (check == -1 ? -1 : !ordered ? check : rs->nextResult(true)); check = (check == -1 ? -1 : !ordered ? check : sOp->nextResult(true));
if( check == -1 ) { if( check == -1 ) {
const NdbError err = pTrans->getNdbError(); const NdbError err = pTrans->getNdbError();
@ -2254,7 +2244,7 @@ HugoTransactions::indexReadRecords(Ndb* pNdb,
reads++; reads++;
r++; r++;
} }
if(ordered && rs->nextResult(true) == 0){ if(ordered && sOp->nextResult(true) == 0){
ndbout << "Error when comparing records " ndbout << "Error when comparing records "
<< " - index op next_result to many" << endl; << " - index op next_result to many" << endl;
pNdb->closeTransaction(pTrans); pNdb->closeTransaction(pTrans);
@ -2284,7 +2274,6 @@ HugoTransactions::indexUpdateRecords(Ndb* pNdb,
NdbConnection *pTrans; NdbConnection *pTrans;
NdbOperation *pOp; NdbOperation *pOp;
NdbScanOperation * sOp; NdbScanOperation * sOp;
NdbResultSet * rs;
const NdbDictionary::Index* pIndex const NdbDictionary::Index* pIndex
= pNdb->getDictionary()->getIndex(idxName, tab.getName()); = pNdb->getDictionary()->getIndex(idxName, tab.getName());
@ -2341,7 +2330,7 @@ HugoTransactions::indexUpdateRecords(Ndb* pNdb,
} }
check = 0; check = 0;
rs = sOp->readTuplesExclusive(); sOp->readTuplesExclusive();
} }
// Define primary keys // Define primary keys
@ -2367,7 +2356,7 @@ HugoTransactions::indexUpdateRecords(Ndb* pNdb,
} }
check = pTrans->execute(NoCommit); check = pTrans->execute(NoCommit);
check = (check == -1 ? -1 : !ordered ? check : rs->nextResult(true)); check = (check == -1 ? -1 : !ordered ? check : sOp->nextResult(true));
if( check == -1 ) { if( check == -1 ) {
const NdbError err = pTrans->getNdbError(); const NdbError err = pTrans->getNdbError();
ERR(err); ERR(err);
@ -2400,7 +2389,7 @@ HugoTransactions::indexUpdateRecords(Ndb* pNdb,
pUpdOp = pTrans->getNdbIndexOperation(idxName, tab.getName()); pUpdOp = pTrans->getNdbIndexOperation(idxName, tab.getName());
check = (pUpdOp == 0 ? -1 : pUpdOp->updateTuple()); check = (pUpdOp == 0 ? -1 : pUpdOp->updateTuple());
} else { } else {
pUpdOp = rs->updateTuple(); pUpdOp = sOp->updateCurrentTuple();
} }
if (pUpdOp == NULL) { if (pUpdOp == NULL) {

61
ndb/test/src/UtilTransactions.cpp
View File

@ -394,8 +394,7 @@ UtilTransactions::clearTable3(Ndb* pNdb,
goto failed; goto failed;
} }
NdbResultSet * rs = pOp->readTuplesExclusive(par); if( pOp->readTuplesExclusive(par) ) {
if( rs == 0 ) {
err = pTrans->getNdbError(); err = pTrans->getNdbError();
goto failed; goto failed;
} }
@ -411,13 +410,13 @@ UtilTransactions::clearTable3(Ndb* pNdb,
goto failed; goto failed;
} }
while((check = rs->nextResult(true)) == 0){ while((check = pOp->nextResult(true)) == 0){
do { do {
if (rs->deleteTuple() != 0){ if (pOp->deleteCurrentTuple() != 0){
goto failed; goto failed;
} }
deletedRows++; deletedRows++;
} while((check = rs->nextResult(false)) == 0); } while((check = pOp->nextResult(false)) == 0);
if(check != -1){ if(check != -1){
check = pTrans->execute(Commit); check = pTrans->execute(Commit);
@ -502,9 +501,7 @@ UtilTransactions::copyTableData(Ndb* pNdb,
return NDBT_FAILED; return NDBT_FAILED;
} }
NdbResultSet* rs = pOp->readTuples(NdbScanOperation::LM_Read, if( pOp->readTuples(NdbScanOperation::LM_Read, parallelism) ) {
parallelism);
if( check == -1 ) {
ERR(pTrans->getNdbError()); ERR(pTrans->getNdbError());
pNdb->closeTransaction(pTrans); pNdb->closeTransaction(pTrans);
return NDBT_FAILED; return NDBT_FAILED;
@ -535,14 +532,14 @@ UtilTransactions::copyTableData(Ndb* pNdb,
} }
int eof; int eof;
while((eof = rs->nextResult(true)) == 0){ while((eof = pOp->nextResult(true)) == 0){
do { do {
insertedRows++; insertedRows++;
if (addRowToInsert(pNdb, pTrans, row, destName) != 0){ if (addRowToInsert(pNdb, pTrans, row, destName) != 0){
pNdb->closeTransaction(pTrans); pNdb->closeTransaction(pTrans);
return NDBT_FAILED; return NDBT_FAILED;
} }
} while((eof = rs->nextResult(false)) == 0); } while((eof = pOp->nextResult(false)) == 0);
check = pTrans->execute(Commit); check = pTrans->execute(Commit);
pTrans->restart(); pTrans->restart();
@ -669,8 +666,7 @@ UtilTransactions::scanReadRecords(Ndb* pNdb,
return NDBT_FAILED; return NDBT_FAILED;
} }
NdbResultSet * rs = pOp->readTuples(lm, 0, parallelism); if( pOp->readTuples(lm, 0, parallelism) ) {
if( rs == 0 ) {
ERR(pTrans->getNdbError()); ERR(pTrans->getNdbError());
pNdb->closeTransaction(pTrans); pNdb->closeTransaction(pTrans);
return NDBT_FAILED; return NDBT_FAILED;
@ -718,7 +714,7 @@ UtilTransactions::scanReadRecords(Ndb* pNdb,
int rows = 0; int rows = 0;
while((eof = rs->nextResult()) == 0){ while((eof = pOp->nextResult()) == 0){
rows++; rows++;
// Call callback for each record returned // Call callback for each record returned
@ -782,8 +778,7 @@ UtilTransactions::selectCount(Ndb* pNdb,
return NDBT_FAILED; return NDBT_FAILED;
} }
NdbResultSet * rs = pOp->readTuples(lm); if( pOp->readTuples(lm) ) {
if( rs == 0) {
ERR(pTrans->getNdbError()); ERR(pTrans->getNdbError());
pNdb->closeTransaction(pTrans); pNdb->closeTransaction(pTrans);
return NDBT_FAILED; return NDBT_FAILED;
@ -815,7 +810,7 @@ UtilTransactions::selectCount(Ndb* pNdb,
int rows = 0; int rows = 0;
while((eof = rs->nextResult()) == 0){ while((eof = pOp->nextResult()) == 0){
rows++; rows++;
} }
if (eof == -1) { if (eof == -1) {
@ -948,14 +943,14 @@ UtilTransactions::scanAndCompareUniqueIndex(Ndb* pNdb,
return NDBT_FAILED; return NDBT_FAILED;
} }
NdbResultSet* rs; int rs;
if(transactional){ if(transactional){
rs = pOp->readTuples(NdbScanOperation::LM_Read, 0, parallelism); rs = pOp->readTuples(NdbScanOperation::LM_Read, 0, parallelism);
} else { } else {
rs = pOp->readTuples(NdbScanOperation::LM_CommittedRead, 0, parallelism); rs = pOp->readTuples(NdbScanOperation::LM_CommittedRead, 0, parallelism);
} }
if( rs == 0 ) { if( rs != 0 ) {
ERR(pTrans->getNdbError()); ERR(pTrans->getNdbError());
pNdb->closeTransaction(pTrans); pNdb->closeTransaction(pTrans);
return NDBT_FAILED; return NDBT_FAILED;
@ -998,7 +993,7 @@ UtilTransactions::scanAndCompareUniqueIndex(Ndb* pNdb,
int rows = 0; int rows = 0;
while((eof = rs->nextResult()) == 0){ while((eof = pOp->nextResult()) == 0){
rows++; rows++;
// ndbout << row.c_str().c_str() << endl; // ndbout << row.c_str().c_str() << endl;
@ -1046,7 +1041,6 @@ UtilTransactions::readRowFromTableAndIndex(Ndb* pNdb,
const int retryMax = 100; const int retryMax = 100;
int check, a; int check, a;
NdbConnection *pTrans1=NULL; NdbConnection *pTrans1=NULL;
NdbResultSet *cursor= NULL;
NdbOperation *pOp; NdbOperation *pOp;
int return_code= NDBT_FAILED; int return_code= NDBT_FAILED;
@ -1174,7 +1168,7 @@ UtilTransactions::readRowFromTableAndIndex(Ndb* pNdb,
if (pIndexOp) { if (pIndexOp) {
not_ok = pIndexOp->readTuple() == -1; not_ok = pIndexOp->readTuple() == -1;
} else { } else {
not_ok = (cursor= pScanOp->readTuples()) == 0; not_ok = pScanOp->readTuples();
} }
if( not_ok ) { if( not_ok ) {
@ -1251,7 +1245,7 @@ UtilTransactions::readRowFromTableAndIndex(Ndb* pNdb,
*/ */
if(!null_found){ if(!null_found){
if (pScanOp) { if (pScanOp) {
if (cursor->nextResult() != 0){ if (pScanOp->nextResult() != 0){
const NdbError err = pTrans1->getNdbError(); const NdbError err = pTrans1->getNdbError();
ERR(err); ERR(err);
ndbout << "Error when comparing records - index op next_result missing" << endl; ndbout << "Error when comparing records - index op next_result missing" << endl;
@ -1266,7 +1260,7 @@ UtilTransactions::readRowFromTableAndIndex(Ndb* pNdb,
goto close_all; goto close_all;
} }
if (pScanOp) { if (pScanOp) {
if (cursor->nextResult() == 0){ if (pScanOp->nextResult() == 0){
ndbout << "Error when comparing records - index op next_result to many" << endl; ndbout << "Error when comparing records - index op next_result to many" << endl;
ndbout << "row: " << row.c_str().c_str() << endl; ndbout << "row: " << row.c_str().c_str() << endl;
goto close_all; goto close_all;
@ -1278,8 +1272,6 @@ UtilTransactions::readRowFromTableAndIndex(Ndb* pNdb,
} }
close_all: close_all:
if (cursor)
cursor->close();
if (pTrans1) if (pTrans1)
pNdb->closeTransaction(pTrans1); pNdb->closeTransaction(pTrans1);
@ -1298,7 +1290,6 @@ UtilTransactions::verifyOrderedIndex(Ndb* pNdb,
NdbConnection *pTrans; NdbConnection *pTrans;
NdbScanOperation *pOp; NdbScanOperation *pOp;
NdbIndexScanOperation * iop = 0; NdbIndexScanOperation * iop = 0;
NdbResultSet* cursor= 0;
NDBT_ResultRow scanRow(tab); NDBT_ResultRow scanRow(tab);
NDBT_ResultRow pkRow(tab); NDBT_ResultRow pkRow(tab);
@ -1337,10 +1328,7 @@ UtilTransactions::verifyOrderedIndex(Ndb* pNdb,
return NDBT_FAILED; return NDBT_FAILED;
} }
NdbResultSet* if( pOp->readTuples(NdbScanOperation::LM_Read, 0, parallelism) ) {
rs = pOp->readTuples(NdbScanOperation::LM_Read, 0, parallelism);
if( rs == 0 ) {
ERR(pTrans->getNdbError()); ERR(pTrans->getNdbError());
pNdb->closeTransaction(pTrans); pNdb->closeTransaction(pTrans);
return NDBT_FAILED; return NDBT_FAILED;
@ -1376,7 +1364,7 @@ UtilTransactions::verifyOrderedIndex(Ndb* pNdb,
int eof; int eof;
int rows = 0; int rows = 0;
while(check == 0 && (eof = rs->nextResult()) == 0){ while(check == 0 && (eof = pOp->nextResult()) == 0){
rows++; rows++;
bool null_found= false; bool null_found= false;
@ -1401,10 +1389,11 @@ UtilTransactions::verifyOrderedIndex(Ndb* pNdb,
if(!iop && (iop= pTrans->getNdbIndexScanOperation(indexName, if(!iop && (iop= pTrans->getNdbIndexScanOperation(indexName,
tab.getName()))) tab.getName())))
{ {
cursor= iop->readTuples(NdbScanOperation::LM_CommittedRead, if(iop->readTuples(NdbScanOperation::LM_CommittedRead,
parallelism); parallelism))
goto error;
iop->interpret_exit_ok(); iop->interpret_exit_ok();
if(!cursor || get_values(iop, indexRow)) if(get_values(iop, indexRow))
goto error; goto error;
} }
else if(!iop || iop->reset_bounds()) else if(!iop || iop->reset_bounds())
@ -1431,7 +1420,7 @@ UtilTransactions::verifyOrderedIndex(Ndb* pNdb,
if(!null_found) if(!null_found)
{ {
if((res= cursor->nextResult()) != 0){ if((res= iop->nextResult()) != 0){
g_err << "Failed to find row using index: " << res << endl; g_err << "Failed to find row using index: " << res << endl;
ERR(pTrans->getNdbError()); ERR(pTrans->getNdbError());
pNdb->closeTransaction(pTrans); pNdb->closeTransaction(pTrans);
@ -1446,7 +1435,7 @@ UtilTransactions::verifyOrderedIndex(Ndb* pNdb,
return NDBT_FAILED; return NDBT_FAILED;
} }
if(cursor->nextResult() == 0){ if(iop->nextResult() == 0){
g_err << "Found extra row!!" << endl; g_err << "Found extra row!!" << endl;
g_err << " indexRow: \n" << indexRow.c_str().c_str() << endl; g_err << " indexRow: \n" << indexRow.c_str().c_str() << endl;
pNdb->closeTransaction(pTrans); pNdb->closeTransaction(pTrans);

9
ndb/tools/delete_all.cpp
View File

@ -141,8 +141,7 @@ int clear_table(Ndb* pNdb, const NdbDictionary::Table* pTab, int parallelism)
goto failed; goto failed;
} }
NdbResultSet * rs = pOp->readTuplesExclusive(par); if( pOp->readTuplesExclusive(par) ) {
if( rs == 0 ) {
goto failed; goto failed;
} }
@ -157,13 +156,13 @@ int clear_table(Ndb* pNdb, const NdbDictionary::Table* pTab, int parallelism)
goto failed; goto failed;
} }
while((check = rs->nextResult(true)) == 0){ while((check = pOp->nextResult(true)) == 0){
do { do {
if (rs->deleteTuple() != 0){ if (pOp->deleteCurrentTuple() != 0){
goto failed; goto failed;
} }
deletedRows++; deletedRows++;
} while((check = rs->nextResult(false)) == 0); } while((check = pOp->nextResult(false)) == 0);
if(check != -1){ if(check != -1){
check = pTrans->execute(Commit); check = pTrans->execute(Commit);

8
ndb/tools/select_all.cpp
View File

@ -215,7 +215,7 @@ int scanReadRecords(Ndb* pNdb,
return -1; return -1;
} }
NdbResultSet * rs; int rs;
switch(_lock + (3 * order)){ switch(_lock + (3 * order)){
case 1: case 1:
rs = pOp->readTuples(NdbScanOperation::LM_Read, 0, parallel); rs = pOp->readTuples(NdbScanOperation::LM_Read, 0, parallel);
@ -238,7 +238,7 @@ int scanReadRecords(Ndb* pNdb,
rs = pOp->readTuples(NdbScanOperation::LM_CommittedRead, 0, parallel); rs = pOp->readTuples(NdbScanOperation::LM_CommittedRead, 0, parallel);
break; break;
} }
if( rs == 0 ){ if( rs != 0 ){
ERR(pTrans->getNdbError()); ERR(pTrans->getNdbError());
pNdb->closeTransaction(pTrans); pNdb->closeTransaction(pTrans);
return -1; return -1;
@ -324,7 +324,7 @@ int scanReadRecords(Ndb* pNdb,
int eof; int eof;
int rows = 0; int rows = 0;
eof = rs->nextResult(); eof = pOp->nextResult();
while(eof == 0){ while(eof == 0){
rows++; rows++;
@ -335,7 +335,7 @@ int scanReadRecords(Ndb* pNdb,
ndbout << (*row) << endl; ndbout << (*row) << endl;
} }
eof = rs->nextResult(); eof = pOp->nextResult();
} }
if (eof == -1) { if (eof == -1) {
const NdbError err = pTrans->getNdbError(); const NdbError err = pTrans->getNdbError();

5
ndb/tools/select_count.cpp
View File

@ -164,8 +164,7 @@ select_count(Ndb* pNdb, const NdbDictionary::Table* pTab,
return NDBT_FAILED; return NDBT_FAILED;
} }
NdbResultSet * rs = pOp->readTuples(NdbScanOperation::LM_Dirty); if( pOp->readTuples(NdbScanOperation::LM_Dirty) ) {
if( rs == 0 ) {
ERR(pTrans->getNdbError()); ERR(pTrans->getNdbError());
pNdb->closeTransaction(pTrans); pNdb->closeTransaction(pTrans);
return NDBT_FAILED; return NDBT_FAILED;
@ -191,7 +190,7 @@ select_count(Ndb* pNdb, const NdbDictionary::Table* pTab,
Uint64 row_count = 0; Uint64 row_count = 0;
int eof; int eof;
while((eof = rs->nextResult(true)) == 0){ while((eof = pOp->nextResult(true)) == 0){
row_count += tmp; row_count += tmp;
} }

252
sql/ha_ndbcluster.cc
View File

@ -278,7 +278,7 @@ void ha_ndbcluster::no_uncommitted_rows_init(THD *thd)
Thd_ndb *thd_ndb= (Thd_ndb *)thd->transaction.thd_ndb; Thd_ndb *thd_ndb= (Thd_ndb *)thd->transaction.thd_ndb;
if (info->last_count != thd_ndb->count) if (info->last_count != thd_ndb->count)
{ {
info->last_count = thd_ndb->count; info->last_count= thd_ndb->count;
info->no_uncommitted_rows_count= 0; info->no_uncommitted_rows_count= 0;
info->records= ~(ha_rows)0; info->records= ~(ha_rows)0;
DBUG_PRINT("info", ("id=%d, no_uncommitted_rows_count=%d", DBUG_PRINT("info", ("id=%d, no_uncommitted_rows_count=%d",
@ -1221,31 +1221,13 @@ int ha_ndbcluster::unique_index_read(const byte *key,
DBUG_RETURN(0); DBUG_RETURN(0);
} }
/* inline int ha_ndbcluster::fetch_next()
Get the next record of a started scan. Try to fetch
it locally from NdbApi cached records if possible,
otherwise ask NDB for more.
NOTE
If this is a update/delete make sure to not contact
NDB before any pending ops have been sent to NDB.
*/
inline int ha_ndbcluster::next_result(byte *buf)
{ {
DBUG_ENTER("fetch_next");
int check; int check;
NdbConnection *trans= m_active_trans; NdbConnection *trans= m_active_trans;
NdbResultSet *cursor= m_active_cursor; NdbScanOperation *cursor= m_active_cursor;
DBUG_ENTER("next_result");
if (!cursor)
DBUG_RETURN(HA_ERR_END_OF_FILE);
/*
If this an update or delete, call nextResult with false
to process any records already cached in NdbApi
*/
bool contact_ndb= m_lock.type < TL_WRITE_ALLOW_WRITE; bool contact_ndb= m_lock.type < TL_WRITE_ALLOW_WRITE;
do { do {
DBUG_PRINT("info", ("Call nextResult, contact_ndb: %d", contact_ndb)); DBUG_PRINT("info", ("Call nextResult, contact_ndb: %d", contact_ndb));
@ -1259,14 +1241,9 @@ inline int ha_ndbcluster::next_result(byte *buf)
m_ops_pending= 0; m_ops_pending= 0;
m_blobs_pending= FALSE; m_blobs_pending= FALSE;
} }
check= cursor->nextResult(contact_ndb, m_force_send);
if (check == 0)
{
// One more record found
DBUG_PRINT("info", ("One more record found"));
unpack_record(buf); if ((check= cursor->nextResult(contact_ndb, m_force_send)) == 0)
table->status= 0; {
DBUG_RETURN(0); DBUG_RETURN(0);
} }
else if (check == 1 || check == 2) else if (check == 1 || check == 2)
@ -1282,33 +1259,63 @@ inline int ha_ndbcluster::next_result(byte *buf)
DBUG_PRINT("info", ("ops_pending: %d", m_ops_pending)); DBUG_PRINT("info", ("ops_pending: %d", m_ops_pending));
if (m_ops_pending) if (m_ops_pending)
{ {
// if (current_thd->transaction.on)
if (m_transaction_on) if (m_transaction_on)
{ {
if (execute_no_commit(this,trans) != 0) if (execute_no_commit(this,trans) != 0)
DBUG_RETURN(ndb_err(trans)); return -1;
} }
else else
{ {
if (execute_commit(this,trans) != 0) if (execute_commit(this,trans) != 0)
DBUG_RETURN(ndb_err(trans)); return -1;
int res= trans->restart(); int res= trans->restart();
DBUG_ASSERT(res == 0); DBUG_ASSERT(res == 0);
} }
m_ops_pending= 0; m_ops_pending= 0;
} }
contact_ndb= (check == 2); contact_ndb= (check == 2);
} }
} while (check == 2); } while (check == 2);
table->status= STATUS_NOT_FOUND; DBUG_RETURN(1);
if (check == -1) }
DBUG_RETURN(ndb_err(trans));
/*
Get the next record of a started scan. Try to fetch
it locally from NdbApi cached records if possible,
otherwise ask NDB for more.
NOTE
If this is a update/delete make sure to not contact
NDB before any pending ops have been sent to NDB.
*/
inline int ha_ndbcluster::next_result(byte *buf)
{
int res;
DBUG_ENTER("next_result");
if((res= fetch_next()) == 0)
{
DBUG_PRINT("info", ("One more record found"));
unpack_record(buf);
table->status= 0;
DBUG_RETURN(0);
}
else if(res == 1)
{
// No more records // No more records
table->status= STATUS_NOT_FOUND;
DBUG_PRINT("info", ("No more records")); DBUG_PRINT("info", ("No more records"));
DBUG_RETURN(HA_ERR_END_OF_FILE); DBUG_RETURN(HA_ERR_END_OF_FILE);
}
else
{
DBUG_RETURN(ndb_err(m_active_trans));
}
} }
/* /*
@ -1365,7 +1372,7 @@ int ha_ndbcluster::set_bounds(NdbIndexScanOperation *op,
for (j= 0; j <= 1; j++) for (j= 0; j <= 1; j++)
{ {
struct part_st &p = part[j]; struct part_st &p= part[j];
p.key= NULL; p.key= NULL;
p.bound_type= -1; p.bound_type= -1;
if (tot_len < key_tot_len[j]) if (tot_len < key_tot_len[j])
@ -1445,7 +1452,7 @@ int ha_ndbcluster::set_bounds(NdbIndexScanOperation *op,
for (j= 0; j <= 1; j++) for (j= 0; j <= 1; j++)
{ {
struct part_st &p = part[j]; struct part_st &p= part[j];
// Set bound if not done with this key // Set bound if not done with this key
if (p.key != NULL) if (p.key != NULL)
{ {
@ -1524,7 +1531,6 @@ int ha_ndbcluster::ordered_index_scan(const key_range *start_key,
int res; int res;
bool restart; bool restart;
NdbConnection *trans= m_active_trans; NdbConnection *trans= m_active_trans;
NdbResultSet *cursor;
NdbIndexScanOperation *op; NdbIndexScanOperation *op;
DBUG_ENTER("ordered_index_scan"); DBUG_ENTER("ordered_index_scan");
@ -1542,12 +1548,12 @@ int ha_ndbcluster::ordered_index_scan(const key_range *start_key,
if (!(op= trans->getNdbIndexScanOperation((NDBINDEX *) if (!(op= trans->getNdbIndexScanOperation((NDBINDEX *)
m_index[active_index].index, m_index[active_index].index,
(const NDBTAB *) m_table)) || (const NDBTAB *) m_table)) ||
!(cursor= op->readTuples(lm, 0, parallelism, sorted))) op->readTuples(lm, 0, parallelism, sorted))
ERR_RETURN(trans->getNdbError()); ERR_RETURN(trans->getNdbError());
m_active_cursor= cursor; m_active_cursor= op;
} else { } else {
restart= true; restart= true;
op= (NdbIndexScanOperation*)m_active_cursor->getOperation(); op= (NdbIndexScanOperation*)m_active_cursor;
DBUG_ASSERT(op->getSorted() == sorted); DBUG_ASSERT(op->getSorted() == sorted);
DBUG_ASSERT(op->getLockMode() == DBUG_ASSERT(op->getLockMode() ==
@ -1594,7 +1600,6 @@ int ha_ndbcluster::filtered_scan(const byte *key, uint key_len,
{ {
int res; int res;
NdbConnection *trans= m_active_trans; NdbConnection *trans= m_active_trans;
NdbResultSet *cursor;
NdbScanOperation *op; NdbScanOperation *op;
DBUG_ENTER("filtered_scan"); DBUG_ENTER("filtered_scan");
@ -1607,9 +1612,9 @@ int ha_ndbcluster::filtered_scan(const byte *key, uint key_len,
NdbOperation::LockMode lm= NdbOperation::LockMode lm=
(NdbOperation::LockMode)get_ndb_lock_type(m_lock.type); (NdbOperation::LockMode)get_ndb_lock_type(m_lock.type);
if (!(op= trans->getNdbScanOperation((const NDBTAB *) m_table)) || if (!(op= trans->getNdbScanOperation((const NDBTAB *) m_table)) ||
!(cursor= op->readTuples(lm, 0, parallelism))) op->readTuples(lm, 0, parallelism))
ERR_RETURN(trans->getNdbError()); ERR_RETURN(trans->getNdbError());
m_active_cursor= cursor; m_active_cursor= op;
{ {
// Start scan filter // Start scan filter
@ -1673,7 +1678,6 @@ int ha_ndbcluster::full_table_scan(byte *buf)
{ {
uint i; uint i;
int res; int res;
NdbResultSet *cursor;
NdbScanOperation *op; NdbScanOperation *op;
NdbConnection *trans= m_active_trans; NdbConnection *trans= m_active_trans;
@ -1683,9 +1687,9 @@ int ha_ndbcluster::full_table_scan(byte *buf)
NdbOperation::LockMode lm= NdbOperation::LockMode lm=
(NdbOperation::LockMode)get_ndb_lock_type(m_lock.type); (NdbOperation::LockMode)get_ndb_lock_type(m_lock.type);
if (!(op=trans->getNdbScanOperation((const NDBTAB *) m_table)) || if (!(op=trans->getNdbScanOperation((const NDBTAB *) m_table)) ||
!(cursor= op->readTuples(lm, 0, parallelism))) op->readTuples(lm, 0, parallelism))
ERR_RETURN(trans->getNdbError()); ERR_RETURN(trans->getNdbError());
m_active_cursor= cursor; m_active_cursor= op;
if((res= define_read_attrs(buf, op))) if((res= define_read_attrs(buf, op)))
DBUG_RETURN(res); DBUG_RETURN(res);
@ -1871,7 +1875,7 @@ int ha_ndbcluster::update_row(const byte *old_data, byte *new_data)
{ {
THD *thd= current_thd; THD *thd= current_thd;
NdbConnection *trans= m_active_trans; NdbConnection *trans= m_active_trans;
NdbResultSet* cursor= m_active_cursor; NdbScanOperation* cursor= m_active_cursor;
NdbOperation *op; NdbOperation *op;
uint i; uint i;
DBUG_ENTER("update_row"); DBUG_ENTER("update_row");
@ -1926,7 +1930,7 @@ int ha_ndbcluster::update_row(const byte *old_data, byte *new_data)
the active record in cursor the active record in cursor
*/ */
DBUG_PRINT("info", ("Calling updateTuple on cursor")); DBUG_PRINT("info", ("Calling updateTuple on cursor"));
if (!(op= cursor->updateTuple())) if (!(op= cursor->updateCurrentTuple()))
ERR_RETURN(trans->getNdbError()); ERR_RETURN(trans->getNdbError());
m_ops_pending++; m_ops_pending++;
if (uses_blob_value(FALSE)) if (uses_blob_value(FALSE))
@ -1989,7 +1993,7 @@ int ha_ndbcluster::delete_row(const byte *record)
{ {
THD *thd= current_thd; THD *thd= current_thd;
NdbConnection *trans= m_active_trans; NdbConnection *trans= m_active_trans;
NdbResultSet* cursor= m_active_cursor; NdbScanOperation* cursor= m_active_cursor;
NdbOperation *op; NdbOperation *op;
DBUG_ENTER("delete_row"); DBUG_ENTER("delete_row");
@ -2005,7 +2009,7 @@ int ha_ndbcluster::delete_row(const byte *record)
the active record in cursor the active record in cursor
*/ */
DBUG_PRINT("info", ("Calling deleteTuple on cursor")); DBUG_PRINT("info", ("Calling deleteTuple on cursor"));
if (cursor->deleteTuple() != 0) if (cursor->deleteCurrentTuple() != 0)
ERR_RETURN(trans->getNdbError()); ERR_RETURN(trans->getNdbError());
m_ops_pending++; m_ops_pending++;
@ -2073,7 +2077,7 @@ void ha_ndbcluster::unpack_record(byte* buf)
NdbValue *value= m_value; NdbValue *value= m_value;
DBUG_ENTER("unpack_record"); DBUG_ENTER("unpack_record");
end = table->field + table->fields; end= table->field + table->fields;
// Set null flag(s) // Set null flag(s)
bzero(buf, table->null_bytes); bzero(buf, table->null_bytes);
@ -2303,7 +2307,7 @@ int
check_null_in_key(const KEY* key_info, const byte *key, uint key_len) check_null_in_key(const KEY* key_info, const byte *key, uint key_len)
{ {
KEY_PART_INFO *curr_part, *end_part; KEY_PART_INFO *curr_part, *end_part;
const byte* end_ptr = key + key_len; const byte* end_ptr= key + key_len;
curr_part= key_info->key_part; curr_part= key_info->key_part;
end_part= curr_part + key_info->key_parts; end_part= curr_part + key_info->key_parts;
@ -2327,8 +2331,8 @@ int ha_ndbcluster::index_read(byte *buf,
active_index, key_len, find_flag)); active_index, key_len, find_flag));
int error; int error;
ndb_index_type type = get_index_type(active_index); ndb_index_type type= get_index_type(active_index);
const KEY* key_info = table->key_info+active_index; const KEY* key_info= table->key_info+active_index;
switch (type){ switch (type){
case PRIMARY_KEY_ORDERED_INDEX: case PRIMARY_KEY_ORDERED_INDEX:
case PRIMARY_KEY_INDEX: case PRIMARY_KEY_INDEX:
@ -2367,9 +2371,9 @@ int ha_ndbcluster::index_read(byte *buf,
} }
key_range start_key; key_range start_key;
start_key.key = key; start_key.key= key;
start_key.length = key_len; start_key.length= key_len;
start_key.flag = find_flag; start_key.flag= find_flag;
error= ordered_index_scan(&start_key, 0, TRUE, buf); error= ordered_index_scan(&start_key, 0, TRUE, buf);
DBUG_RETURN(error == HA_ERR_END_OF_FILE ? HA_ERR_KEY_NOT_FOUND : error); DBUG_RETURN(error == HA_ERR_END_OF_FILE ? HA_ERR_KEY_NOT_FOUND : error);
} }
@ -2425,7 +2429,7 @@ int ha_ndbcluster::index_last(byte *buf)
statistic_increment(current_thd->status_var.ha_read_last_count,&LOCK_status); statistic_increment(current_thd->status_var.ha_read_last_count,&LOCK_status);
int res; int res;
if((res= ordered_index_scan(0, 0, TRUE, buf)) == 0){ if((res= ordered_index_scan(0, 0, TRUE, buf)) == 0){
NdbResultSet *cursor= m_active_cursor; NdbScanOperation *cursor= m_active_cursor;
while((res= cursor->nextResult(TRUE, m_force_send)) == 0); while((res= cursor->nextResult(TRUE, m_force_send)) == 0);
if(res == 1){ if(res == 1){
unpack_record(buf); unpack_record(buf);
@ -2508,7 +2512,7 @@ int ha_ndbcluster::read_range_next()
int ha_ndbcluster::rnd_init(bool scan) int ha_ndbcluster::rnd_init(bool scan)
{ {
NdbResultSet *cursor= m_active_cursor; NdbScanOperation *cursor= m_active_cursor;
DBUG_ENTER("rnd_init"); DBUG_ENTER("rnd_init");
DBUG_PRINT("enter", ("scan: %d", scan)); DBUG_PRINT("enter", ("scan: %d", scan));
// Check if scan is to be restarted // Check if scan is to be restarted
@ -2525,7 +2529,7 @@ int ha_ndbcluster::rnd_init(bool scan)
int ha_ndbcluster::close_scan() int ha_ndbcluster::close_scan()
{ {
NdbResultSet *cursor= m_active_cursor; NdbScanOperation *cursor= m_active_cursor;
NdbConnection *trans= m_active_trans; NdbConnection *trans= m_active_trans;
DBUG_ENTER("close_scan"); DBUG_ENTER("close_scan");
@ -3528,12 +3532,12 @@ int ha_ndbcluster::create(const char *name,
case MYSQL_TYPE_MEDIUM_BLOB: case MYSQL_TYPE_MEDIUM_BLOB:
case MYSQL_TYPE_LONG_BLOB: case MYSQL_TYPE_LONG_BLOB:
{ {
NdbDictionary::Column * col = tab.getColumn(i); NdbDictionary::Column * col= tab.getColumn(i);
int size = pk_length + (col->getPartSize()+3)/4 + 7; int size= pk_length + (col->getPartSize()+3)/4 + 7;
if(size > NDB_MAX_TUPLE_SIZE_IN_WORDS && if(size > NDB_MAX_TUPLE_SIZE_IN_WORDS &&
(pk_length+7) < NDB_MAX_TUPLE_SIZE_IN_WORDS) (pk_length+7) < NDB_MAX_TUPLE_SIZE_IN_WORDS)
{ {
size = NDB_MAX_TUPLE_SIZE_IN_WORDS - pk_length - 7; size= NDB_MAX_TUPLE_SIZE_IN_WORDS - pk_length - 7;
col->setPartSize(4*size); col->setPartSize(4*size);
} }
/** /**
@ -4696,8 +4700,7 @@ ndb_get_table_statistics(Ndb* ndb, const char * table,
if (pOp == NULL) if (pOp == NULL)
break; break;
NdbResultSet* rs= pOp->readTuples(NdbOperation::LM_CommittedRead); if (pOp->readTuples(NdbOperation::LM_CommittedRead))
if (rs == 0)
break; break;
int check= pOp->interpret_exit_last_row(); int check= pOp->interpret_exit_last_row();
@ -4714,7 +4717,7 @@ ndb_get_table_statistics(Ndb* ndb, const char * table,
Uint64 sum_rows= 0; Uint64 sum_rows= 0;
Uint64 sum_commits= 0; Uint64 sum_commits= 0;
while((check= rs->nextResult(TRUE, TRUE)) == 0) while((check= pOp->nextResult(TRUE, TRUE)) == 0)
{ {
sum_rows+= rows; sum_rows+= rows;
sum_commits+= commits; sum_commits+= commits;
@ -4723,7 +4726,7 @@ ndb_get_table_statistics(Ndb* ndb, const char * table,
if (check == -1) if (check == -1)
break; break;
rs->close(TRUE); pOp->close(TRUE);
ndb->closeTransaction(pTrans); ndb->closeTransaction(pTrans);
if(row_count) if(row_count)
@ -4777,8 +4780,8 @@ ha_ndbcluster::read_multi_range_first(key_multi_range **found_range_p,
int res; int res;
uint i; uint i;
KEY* key_info= table->key_info + active_index; KEY* key_info= table->key_info + active_index;
NDB_INDEX_TYPE index_type = get_index_type(active_index); NDB_INDEX_TYPE index_type= get_index_type(active_index);
ulong reclength = table->reclength; ulong reclength= table->reclength;
NdbOperation* op; NdbOperation* op;
if (uses_blob_value(m_retrieve_all_fields)) if (uses_blob_value(m_retrieve_all_fields))
@ -4877,14 +4880,14 @@ ha_ndbcluster::read_multi_range_first(key_multi_range **found_range_p,
if (scanOp == 0) if (scanOp == 0)
{ {
if ((scanOp= m_active_trans->getNdbIndexScanOperation(idx, tab)) && if ((scanOp= m_active_trans->getNdbIndexScanOperation(idx, tab)) &&
(m_active_cursor= scanOp->readTuples(lm, 0, !scanOp->readTuples(lm, 0, parallelism, sorted, true) &&
parallelism, sorted, true))&&
!define_read_attrs(curr, scanOp)) !define_read_attrs(curr, scanOp))
curr += reclength; curr += reclength;
else else
ERR_RETURN(scanOp ? ERR_RETURN(scanOp ? scanOp->getNdbError() :
scanOp->getNdbError() :
m_active_trans->getNdbError()); m_active_trans->getNdbError());
m_multi_cursor= scanOp;
m_multi_range_cursor_result_ptr= curr;
} }
const key_range *keys[2]= { &ranges[i].start_key, &ranges[i].end_key }; const key_range *keys[2]= { &ranges[i].start_key, &ranges[i].end_key };
if ((res= set_bounds(scanOp, keys, i))) if ((res= set_bounds(scanOp, keys, i)))
@ -4933,62 +4936,58 @@ ha_ndbcluster::read_multi_range_next(key_multi_range ** multi_range_found_p)
} }
int res; int res;
int range_no;
ulong reclength= table->reclength; ulong reclength= table->reclength;
const NdbOperation* op= m_current_multi_operation; const NdbOperation* op= m_current_multi_operation;
for(;multi_range_curr < m_multi_range_defined_count; multi_range_curr++) for(;multi_range_curr < m_multi_range_defined_count; multi_range_curr++)
{ {
if(multi_ranges[multi_range_curr].range_flag & UNIQUE_RANGE) if (multi_ranges[multi_range_curr].range_flag & UNIQUE_RANGE)
{ {
if(op->getNdbError().code == 0) if (op->getNdbError().code == 0)
goto found_next; goto found_next;
op= m_active_trans->getNextCompletedOperation(op); op= m_active_trans->getNextCompletedOperation(op);
m_multi_range_result_ptr += reclength; m_multi_range_result_ptr += reclength;
continue;
} }
else if(m_active_cursor) else if (m_multi_cursor && !multi_range_sorted)
{ {
int check; if ((res= fetch_next() == 0))
bool contact_ndb= m_lock.type < TL_WRITE_ALLOW_WRITE; {
do { range_no= m_multi_cursor->get_range_no();
check= m_active_cursor->nextResult(contact_ndb, m_force_send);
if (check == 0)
goto found; goto found;
else if (check == 1 || check == 2)
{
// 1: No more records
// 2: No more cached records
/*
Before fetching more rows and releasing lock(s),
all pending update or delete operations should
be sent to NDB
*/
DBUG_PRINT("info", ("ops_pending: %d", m_ops_pending));
if (m_ops_pending)
{
// if (current_thd->transaction.on)
if (m_transaction_on)
{
if (execute_no_commit(this, m_active_trans) != 0)
DBUG_RETURN(ndb_err(m_active_trans));
} }
else else
{ {
if (execute_commit(this, m_active_trans) != 0) goto close_scan;
DBUG_RETURN(ndb_err(m_active_trans));
int res= m_active_trans->restart();
DBUG_ASSERT(res == 0);
} }
m_ops_pending= 0;
} }
contact_ndb= (check == 2); else if (multi_range_sorted)
} {
} while (check == 2); DBUG_ASSERT(m_multi_cursor);
if (m_active_cursor && (res= fetch_next()))
goto close_scan;
m_multi_range_result_ptr += reclength; range_no= m_multi_cursor->get_range_no();
if (range_no == multi_range_curr)
m_active_cursor->close(); {
// return current row
goto found;
}
else if (range_no > multi_range_curr)
{
// wait with current row
m_active_cursor= 0; m_active_cursor= 0;
continue;
}
else
{
// First fetch from cursor
DBUG_ASSERT(range_no == -1);
m_multi_cursor->nextResult(true);
multi_range_curr--; // Will be increased in for-loop
continue;
}
} }
else /** m_active_cursor == 0 */ else /** m_active_cursor == 0 */
{ {
@ -4996,6 +4995,22 @@ ha_ndbcluster::read_multi_range_next(key_multi_range ** multi_range_found_p)
* Corresponds to range 5 in example in read_multi_range_first * Corresponds to range 5 in example in read_multi_range_first
*/ */
(void)1; (void)1;
continue;
}
DBUG_ASSERT(false); // Should only get here via goto's
close_scan:
if (res == 1)
{
m_multi_range_result_ptr += reclength;
m_active_cursor->close();
m_active_cursor= m_multi_cursor= 0;
continue;
}
else
{
DBUG_RETURN(ndb_err(m_active_trans));
} }
} }
@ -5016,11 +5031,10 @@ found:
/** /**
* Found a record belonging to a scan * Found a record belonging to a scan
*/ */
Uint32 range_no = ((NdbIndexScanOperation*)m_active_cursor->getOperation()) m_active_cursor= m_multi_cursor;
->get_range_no();
* multi_range_found_p= multi_ranges + range_no; * multi_range_found_p= multi_ranges + range_no;
memcpy(table->record[0], m_multi_range_result_ptr, reclength); memcpy(table->record[0], m_multi_range_cursor_result_ptr, reclength);
setup_recattr(m_active_cursor->getOperation()->getFirstRecAttr()); setup_recattr(m_active_cursor->getFirstRecAttr());
unpack_record(table->record[0]); unpack_record(table->record[0]);
table->status= 0; table->status= 0;
DBUG_RETURN(0); DBUG_RETURN(0);
@ -5050,15 +5064,15 @@ ha_ndbcluster::setup_recattr(const NdbRecAttr* curr)
Field **field, **end; Field **field, **end;
NdbValue *value= m_value; NdbValue *value= m_value;
end = table->field + table->fields; end= table->field + table->fields;
for (field= table->field; field < end; field++, value++) for (field= table->field; field < end; field++, value++)
{ {
if ((* value).ptr) if ((* value).ptr)
{ {
DBUG_ASSERT(curr != 0); DBUG_ASSERT(curr != 0);
(* value).rec = curr; (* value).rec= curr;
curr = curr->next(); curr= curr->next();
} }
} }