mirror of
https://github.com/MariaDB/server.git
synced 2025-01-18 21:12:26 +01:00
wl2126 - ndb - fix last unhandled part of read_multi_range
reading sorted multi_range with several range-scans
This commit is contained in:
parent
cd79b299a6
commit
20fa339cb1
30 changed files with 498 additions and 709 deletions
|
@ -26,7 +26,6 @@
|
|||
#include "NdbIndexScanOperation.hpp"
|
||||
#include "NdbScanFilter.hpp"
|
||||
#include "NdbRecAttr.hpp"
|
||||
#include "NdbResultSet.hpp"
|
||||
#include "NdbDictionary.hpp"
|
||||
#include "NdbEventOperation.hpp"
|
||||
#include "NdbPool.hpp"
|
||||
|
|
|
@ -42,20 +42,20 @@ public:
|
|||
* @returns NdbResultSet.
|
||||
* @see NdbScanOperation::readTuples
|
||||
*/
|
||||
NdbResultSet* readTuples(LockMode = LM_Read,
|
||||
Uint32 batch = 0,
|
||||
Uint32 parallel = 0,
|
||||
bool order_by = false,
|
||||
bool read_range_no = false);
|
||||
int readTuples(LockMode = LM_Read,
|
||||
Uint32 batch = 0,
|
||||
Uint32 parallel = 0,
|
||||
bool order_by = false,
|
||||
bool read_range_no = false);
|
||||
|
||||
inline NdbResultSet* readTuples(int parallell){
|
||||
inline int readTuples(int parallell){
|
||||
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);
|
||||
}
|
||||
|
||||
|
||||
/**
|
||||
* Type of ordered index key bound. The values (0-4) will not change
|
||||
* and can be used explicitly (e.g. they could be computed).
|
||||
|
@ -125,7 +125,7 @@ public:
|
|||
/**
|
||||
* Return range no for current row
|
||||
*/
|
||||
Uint32 get_range_no();
|
||||
int get_range_no();
|
||||
|
||||
bool getSorted() const { return m_ordered; }
|
||||
private:
|
||||
|
|
|
@ -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öld
|
||||
* Date: 2002-04-01
|
||||
* Version: 0.1
|
||||
* Description: Cursor class
|
||||
* Documentation:
|
||||
* Adjust: 2002-04-01 Martin Skö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
|
|
@ -45,20 +45,6 @@ class NdbScanOperation : public NdbOperation {
|
|||
friend class NdbOperation;
|
||||
friend class NdbBlob;
|
||||
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.
|
||||
* Tuples are not stored in NdbResultSet until execute(NoCommit)
|
||||
|
@ -67,33 +53,126 @@ public:
|
|||
* @param parallel Scan parallelism
|
||||
* @param batch No of rows to fetch from each fragment at a time
|
||||
* @param LockMode Scan lock handling
|
||||
* @returns NdbResultSet.
|
||||
* @note specifying 0 for batch and parallall means max performance
|
||||
*/
|
||||
NdbResultSet* readTuples(LockMode = LM_Read,
|
||||
Uint32 batch = 0, Uint32 parallel = 0);
|
||||
int readTuples(LockMode = LM_Read,
|
||||
Uint32 batch = 0, Uint32 parallel = 0);
|
||||
|
||||
inline NdbResultSet* readTuples(int parallell){
|
||||
inline int readTuples(int 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);
|
||||
}
|
||||
|
||||
NdbBlob* getBlobHandle(const char* anAttrName);
|
||||
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);
|
||||
virtual ~NdbScanOperation();
|
||||
|
||||
int nextResult(bool fetchAllowed = true, bool forceSend = false);
|
||||
int nextResultImpl(bool fetchAllowed = true, bool forceSend = false);
|
||||
virtual void release();
|
||||
|
||||
void closeScan(bool forceSend = false);
|
||||
int close_impl(class TransporterFacade*, bool forceSend = false);
|
||||
|
||||
// Overloaded methods from NdbCursorOperation
|
||||
|
@ -108,8 +187,6 @@ protected:
|
|||
virtual void setErrorCode(int aErrorCode);
|
||||
virtual void setErrorCodeAbort(int aErrorCode);
|
||||
|
||||
NdbResultSet * m_resultSet;
|
||||
NdbResultSet* getResultSet();
|
||||
NdbConnection *m_transConnection;
|
||||
|
||||
// Scan related variables
|
||||
|
@ -157,14 +234,35 @@ protected:
|
|||
|
||||
Uint32 m_ordered;
|
||||
Uint32 m_read_range_no;
|
||||
|
||||
int restart(bool forceSend = false);
|
||||
};
|
||||
|
||||
inline
|
||||
NdbScanOperation::CursorType
|
||||
NdbScanOperation::get_cursor_type() const {
|
||||
return m_cursor_type;
|
||||
NdbOperation*
|
||||
NdbScanOperation::updateCurrentTuple(){
|
||||
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
|
||||
|
|
|
@ -22,7 +22,6 @@ libndbapi_la_SOURCES = \
|
|||
NdbOperationInt.cpp \
|
||||
NdbOperationDefine.cpp \
|
||||
NdbOperationExec.cpp \
|
||||
NdbResultSet.cpp \
|
||||
NdbScanOperation.cpp NdbScanFilter.cpp \
|
||||
NdbIndexOperation.cpp \
|
||||
NdbEventOperation.cpp \
|
||||
|
|
|
@ -1109,7 +1109,6 @@ NdbConnection::getNdbIndexScanOperation(const NdbIndexImpl* index,
|
|||
if(tOp)
|
||||
{
|
||||
tOp->m_currentTable = table;
|
||||
tOp->m_cursor_type = NdbScanOperation::IndexCursor;
|
||||
}
|
||||
return tOp;
|
||||
} else {
|
||||
|
|
|
@ -16,7 +16,6 @@
|
|||
|
||||
#include <ndb_global.h>
|
||||
#include <NdbIndexOperation.hpp>
|
||||
#include <NdbResultSet.hpp>
|
||||
#include <Ndb.hpp>
|
||||
#include <NdbConnection.hpp>
|
||||
#include "NdbApiSignal.hpp"
|
||||
|
|
|
@ -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ö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);
|
||||
}
|
|
@ -19,10 +19,10 @@
|
|||
#include <NdbScanOperation.hpp>
|
||||
#include <NdbIndexScanOperation.hpp>
|
||||
#include <NdbConnection.hpp>
|
||||
#include <NdbResultSet.hpp>
|
||||
#include "NdbApiSignal.hpp"
|
||||
#include <NdbOut.hpp>
|
||||
#include "NdbDictionaryImpl.hpp"
|
||||
#include <NdbBlob.hpp>
|
||||
|
||||
#include <NdbRecAttr.hpp>
|
||||
#include <NdbReceiver.hpp>
|
||||
|
@ -39,7 +39,6 @@
|
|||
|
||||
NdbScanOperation::NdbScanOperation(Ndb* aNdb) :
|
||||
NdbOperation(aNdb),
|
||||
m_resultSet(0),
|
||||
m_transConnection(NULL)
|
||||
{
|
||||
theParallelism = 0;
|
||||
|
@ -60,21 +59,8 @@ NdbScanOperation::~NdbScanOperation()
|
|||
theNdb->releaseNdbScanRec(m_receivers[i]);
|
||||
}
|
||||
delete[] m_array;
|
||||
if (m_resultSet)
|
||||
delete m_resultSet;
|
||||
}
|
||||
|
||||
NdbResultSet*
|
||||
NdbScanOperation::getResultSet()
|
||||
{
|
||||
if (!m_resultSet)
|
||||
m_resultSet = new NdbResultSet(this);
|
||||
|
||||
return m_resultSet;
|
||||
}
|
||||
|
||||
|
||||
|
||||
void
|
||||
NdbScanOperation::setErrorCode(int aErrorCode){
|
||||
NdbConnection* tmp = theNdbCon;
|
||||
|
@ -125,9 +111,10 @@ NdbScanOperation::init(const NdbTableImpl* tab, NdbConnection* myConnection)
|
|||
return 0;
|
||||
}
|
||||
|
||||
NdbResultSet* NdbScanOperation::readTuples(NdbScanOperation::LockMode lm,
|
||||
Uint32 batch,
|
||||
Uint32 parallel)
|
||||
int
|
||||
NdbScanOperation::readTuples(NdbScanOperation::LockMode lm,
|
||||
Uint32 batch,
|
||||
Uint32 parallel)
|
||||
{
|
||||
m_ordered = 0;
|
||||
Uint32 fragCount = m_currentTable->m_fragmentCount;
|
||||
|
@ -142,7 +129,7 @@ NdbResultSet* NdbScanOperation::readTuples(NdbScanOperation::LockMode lm,
|
|||
// 3. theScanOp contains a NdbScanOperation
|
||||
if (theNdbCon->theScanningOp != NULL){
|
||||
setErrorCode(4605);
|
||||
return 0;
|
||||
return -1;
|
||||
}
|
||||
|
||||
theNdbCon->theScanningOp = this;
|
||||
|
@ -167,7 +154,7 @@ NdbResultSet* NdbScanOperation::readTuples(NdbScanOperation::LockMode lm,
|
|||
break;
|
||||
default:
|
||||
setErrorCode(4003);
|
||||
return 0;
|
||||
return -1;
|
||||
}
|
||||
|
||||
m_keyInfo = lockExcl ? 1 : 0;
|
||||
|
@ -192,13 +179,13 @@ NdbResultSet* NdbScanOperation::readTuples(NdbScanOperation::LockMode lm,
|
|||
|
||||
if(fix_receivers(parallel) == -1){
|
||||
setErrorCodeAbort(4000);
|
||||
return 0;
|
||||
return -1;
|
||||
}
|
||||
|
||||
theSCAN_TABREQ = (!theSCAN_TABREQ ? theNdb->getSignal() : theSCAN_TABREQ);
|
||||
if (theSCAN_TABREQ == NULL) {
|
||||
setErrorCodeAbort(4000);
|
||||
return 0;
|
||||
return -1;
|
||||
}//if
|
||||
|
||||
theSCAN_TABREQ->setSignal(GSN_SCAN_TABREQ);
|
||||
|
@ -234,7 +221,7 @@ NdbResultSet* NdbScanOperation::readTuples(NdbScanOperation::LockMode lm,
|
|||
theTotalNrOfKeyWordInSignal= 0;
|
||||
|
||||
getFirstATTRINFOScan();
|
||||
return getResultSet();
|
||||
return 0;
|
||||
}
|
||||
|
||||
int
|
||||
|
@ -403,6 +390,29 @@ NdbScanOperation::executeCursor(int nodeId){
|
|||
|
||||
|
||||
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)
|
||||
return ((NdbIndexScanOperation*)this)->next_result_ordered(fetchAllowed,
|
||||
|
@ -622,11 +632,11 @@ NdbScanOperation::doSend(int ProcessorId)
|
|||
return 0;
|
||||
}
|
||||
|
||||
void NdbScanOperation::closeScan(bool forceSend)
|
||||
void NdbScanOperation::close(bool forceSend)
|
||||
{
|
||||
if(m_transConnection){
|
||||
if(DEBUG_NEXT_RESULT)
|
||||
ndbout_c("closeScan() theError.code = %d "
|
||||
ndbout_c("close() theError.code = %d "
|
||||
"m_api_receivers_count = %d "
|
||||
"m_conf_receivers_count = %d "
|
||||
"m_sent_receivers_count = %d",
|
||||
|
@ -657,7 +667,7 @@ NdbScanOperation::execCLOSE_SCAN_REP(){
|
|||
void NdbScanOperation::release()
|
||||
{
|
||||
if(theNdbCon != 0 || m_transConnection != 0){
|
||||
closeScan();
|
||||
close();
|
||||
}
|
||||
for(Uint32 i = 0; i<m_allocated_receivers; i++){
|
||||
m_receivers[i]->release();
|
||||
|
@ -1196,22 +1206,22 @@ error:
|
|||
return -1;
|
||||
}
|
||||
|
||||
NdbResultSet*
|
||||
int
|
||||
NdbIndexScanOperation::readTuples(LockMode lm,
|
||||
Uint32 batch,
|
||||
Uint32 parallel,
|
||||
bool order_by,
|
||||
bool read_range_no){
|
||||
NdbResultSet * rs = NdbScanOperation::readTuples(lm, batch, 0);
|
||||
int res = NdbScanOperation::readTuples(lm, batch, 0);
|
||||
if(read_range_no)
|
||||
{
|
||||
m_read_range_no = 1;
|
||||
Uint32 word = 0;
|
||||
AttributeHeader::init(&word, AttributeHeader::RANGE_NO, 0);
|
||||
if(insertATTRINFO(word) == -1)
|
||||
rs = 0;
|
||||
res = -1;
|
||||
}
|
||||
if(rs && order_by){
|
||||
if(!res && order_by){
|
||||
m_ordered = 1;
|
||||
Uint32 cnt = m_accessTable->getNoOfColumns() - 1;
|
||||
m_sort_columns = cnt; // -1 for NDB$NODE
|
||||
|
@ -1234,7 +1244,7 @@ NdbIndexScanOperation::readTuples(LockMode lm,
|
|||
m_this_bound_start = 0;
|
||||
m_first_bound_word = theKEYINFOptr;
|
||||
|
||||
return rs;
|
||||
return res;
|
||||
}
|
||||
|
||||
void
|
||||
|
@ -1666,7 +1676,7 @@ NdbIndexScanOperation::end_of_bound(Uint32 no)
|
|||
return -1;
|
||||
}
|
||||
|
||||
Uint32
|
||||
int
|
||||
NdbIndexScanOperation::get_range_no()
|
||||
{
|
||||
if(m_read_range_no)
|
||||
|
@ -1687,5 +1697,5 @@ NdbIndexScanOperation::get_range_no()
|
|||
return ret;
|
||||
}
|
||||
}
|
||||
return 0;
|
||||
return -1;
|
||||
}
|
||||
|
|
|
@ -95,7 +95,7 @@ protected:
|
|||
|
||||
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_executed_result_sets;
|
||||
private:
|
||||
|
|
|
@ -81,7 +81,6 @@ ScanFunctions::scanReadFunctions(Ndb* pNdb,
|
|||
int check;
|
||||
NdbConnection *pTrans = 0;
|
||||
NdbScanOperation *pOp = 0;
|
||||
NdbResultSet *rs = 0;
|
||||
|
||||
while (true){
|
||||
if (retryAttempt >= retryMax){
|
||||
|
@ -111,12 +110,9 @@ ScanFunctions::scanReadFunctions(Ndb* pNdb,
|
|||
return NDBT_FAILED;
|
||||
}
|
||||
|
||||
|
||||
rs = pOp->readTuples(exclusive ?
|
||||
NdbScanOperation::LM_Exclusive :
|
||||
NdbScanOperation::LM_Read);
|
||||
|
||||
if( rs == 0 ) {
|
||||
if( pOp->readTuples(exclusive ?
|
||||
NdbScanOperation::LM_Exclusive :
|
||||
NdbScanOperation::LM_Read) ) {
|
||||
ERR(pTrans->getNdbError());
|
||||
pNdb->closeTransaction(pTrans);
|
||||
return NDBT_FAILED;
|
||||
|
@ -125,8 +121,7 @@ ScanFunctions::scanReadFunctions(Ndb* pNdb,
|
|||
|
||||
if (action == OnlyOpenScanOnce){
|
||||
// Call openScan one more time when it's already defined
|
||||
NdbResultSet* rs2 = pOp->readTuples(NdbScanOperation::LM_Read);
|
||||
if( rs2 == 0 ) {
|
||||
if( pOp->readTuples(NdbScanOperation::LM_Read) ) {
|
||||
ERR(pTrans->getNdbError());
|
||||
pNdb->closeTransaction(pTrans);
|
||||
return NDBT_FAILED;
|
||||
|
@ -168,7 +163,7 @@ ScanFunctions::scanReadFunctions(Ndb* pNdb,
|
|||
bool abortTrans = (action==CloseWithoutStop);
|
||||
int eof;
|
||||
int rows = 0;
|
||||
eof = rs->nextResult();
|
||||
eof = pOp->nextResult();
|
||||
|
||||
while(eof == 0){
|
||||
rows++;
|
||||
|
@ -178,7 +173,7 @@ ScanFunctions::scanReadFunctions(Ndb* pNdb,
|
|||
|
||||
if (action != CloseWithoutStop){
|
||||
// Test that we can closeTrans without stopScan
|
||||
rs->close();
|
||||
pOp->close();
|
||||
if( check == -1 ) {
|
||||
ERR(pTrans->getNdbError());
|
||||
pNdb->closeTransaction(pTrans);
|
||||
|
@ -201,7 +196,7 @@ ScanFunctions::scanReadFunctions(Ndb* pNdb,
|
|||
}
|
||||
}
|
||||
|
||||
eof = rs->nextResult();
|
||||
eof = pOp->nextResult();
|
||||
}
|
||||
if (eof == -1) {
|
||||
const NdbError err = pTrans->getNdbError();
|
||||
|
@ -211,7 +206,7 @@ ScanFunctions::scanReadFunctions(Ndb* pNdb,
|
|||
|
||||
// Be cruel, call nextScanResult after error
|
||||
for(int i=0; i<10; i++){
|
||||
eof = rs->nextResult();
|
||||
eof = pOp->nextResult();
|
||||
if(eof == 0){
|
||||
g_err << "nextScanResult returned eof = " << eof << endl
|
||||
<< " That is an error when there are no more records" << endl;
|
||||
|
@ -241,7 +236,7 @@ ScanFunctions::scanReadFunctions(Ndb* pNdb,
|
|||
if (action == NextScanWhenNoMore){
|
||||
g_info << "Calling nextScanresult when there are no more records" << endl;
|
||||
for(int i=0; i<10; i++){
|
||||
eof = rs->nextResult();
|
||||
eof = pOp->nextResult();
|
||||
if(eof == 0){
|
||||
g_err << "nextScanResult returned eof = " << eof << endl
|
||||
<< " That is an error when there are no more records" << endl;
|
||||
|
|
|
@ -227,10 +227,7 @@ ScanInterpretTest::scanRead(Ndb* pNdb,
|
|||
return NDBT_FAILED;
|
||||
}
|
||||
|
||||
NdbResultSet * rs = pOp->readTuples(NdbScanOperation::LM_Read,
|
||||
0, parallelism);
|
||||
|
||||
if( rs == 0 ) {
|
||||
if( pOp->readTuples(NdbScanOperation::LM_Read, 0, parallelism) ) {
|
||||
ERR(pTrans->getNdbError());
|
||||
pNdb->closeTransaction(pTrans);
|
||||
return NDBT_FAILED;
|
||||
|
@ -262,14 +259,14 @@ ScanInterpretTest::scanRead(Ndb* pNdb,
|
|||
int rows = 0;
|
||||
NdbConnection* pInsTrans;
|
||||
|
||||
while((eof = rs->nextResult(true)) == 0){
|
||||
while((eof = pOp->nextResult(true)) == 0){
|
||||
do {
|
||||
rows++;
|
||||
if (addRowToInsert(pNdb, pTrans) != 0){
|
||||
pNdb->closeTransaction(pTrans);
|
||||
return NDBT_FAILED;
|
||||
}
|
||||
} while((eof = rs->nextResult(false)) == 0);
|
||||
} while((eof = pOp->nextResult(false)) == 0);
|
||||
|
||||
check = pTrans->execute(Commit);
|
||||
if( check == -1 ) {
|
||||
|
@ -349,9 +346,7 @@ ScanInterpretTest::scanReadVerify(Ndb* pNdb,
|
|||
return NDBT_FAILED;
|
||||
}
|
||||
|
||||
NdbResultSet * rs = pOp->readTuples(NdbScanOperation::LM_Read,
|
||||
0, parallelism);
|
||||
if( rs == 0 ) {
|
||||
if( pOp->readTuples(NdbScanOperation::LM_Read, 0, parallelism) ) {
|
||||
ERR(pTrans->getNdbError());
|
||||
pNdb->closeTransaction(pTrans);
|
||||
return NDBT_FAILED;
|
||||
|
@ -392,7 +387,7 @@ ScanInterpretTest::scanReadVerify(Ndb* pNdb,
|
|||
NdbConnection* pExistTrans;
|
||||
NdbConnection* pNoExistTrans;
|
||||
|
||||
while((eof = rs->nextResult(true)) == 0){
|
||||
while((eof = pOp->nextResult(true)) == 0){
|
||||
pExistTrans = pNdb->startTransaction();
|
||||
if (pExistTrans == NULL) {
|
||||
const NdbError err = pNdb->getNdbError();
|
||||
|
@ -424,7 +419,7 @@ ScanInterpretTest::scanReadVerify(Ndb* pNdb,
|
|||
return NDBT_FAILED;
|
||||
}
|
||||
}
|
||||
} while((eof = rs->nextResult(false)) == 0);
|
||||
} while((eof = pOp->nextResult(false)) == 0);
|
||||
|
||||
|
||||
// Execute the transaction containing reads of
|
||||
|
|
|
@ -669,8 +669,7 @@ int Bank::findLastGL(Uint64 &lastTime){
|
|||
return NDBT_FAILED;
|
||||
}
|
||||
|
||||
NdbResultSet * rs = pOp->readTuples();
|
||||
if( rs == 0 ) {
|
||||
if( pOp->readTuples() ) {
|
||||
ERR(pScanTrans->getNdbError());
|
||||
m_ndb.closeTransaction(pScanTrans);
|
||||
return NDBT_FAILED;
|
||||
|
@ -699,7 +698,7 @@ int Bank::findLastGL(Uint64 &lastTime){
|
|||
|
||||
int eof;
|
||||
int rows = 0;
|
||||
eof = rs->nextResult();
|
||||
eof = pOp->nextResult();
|
||||
lastTime = 0;
|
||||
|
||||
while(eof == 0){
|
||||
|
@ -709,7 +708,7 @@ int Bank::findLastGL(Uint64 &lastTime){
|
|||
if (t > lastTime)
|
||||
lastTime = t;
|
||||
|
||||
eof = rs->nextResult();
|
||||
eof = pOp->nextResult();
|
||||
}
|
||||
if (eof == -1) {
|
||||
ERR(pScanTrans->getNdbError());
|
||||
|
@ -1001,8 +1000,7 @@ int Bank::sumTransactionsForGL(const Uint64 glTime,
|
|||
return NDBT_FAILED;
|
||||
}
|
||||
|
||||
NdbResultSet * rs = pOp->readTuplesExclusive();
|
||||
if( rs == 0 ) {
|
||||
if( pOp->readTuplesExclusive()) {
|
||||
ERR(pScanTrans->getNdbError());
|
||||
m_ndb.closeTransaction(pScanTrans);
|
||||
return NDBT_FAILED;
|
||||
|
@ -1053,7 +1051,7 @@ int Bank::sumTransactionsForGL(const Uint64 glTime,
|
|||
int eof;
|
||||
int rows = 0;
|
||||
int rowsFound = 0;
|
||||
eof = rs->nextResult();
|
||||
eof = pOp->nextResult();
|
||||
|
||||
while(eof == 0){
|
||||
rows++;
|
||||
|
@ -1077,7 +1075,7 @@ int Bank::sumTransactionsForGL(const Uint64 glTime,
|
|||
}
|
||||
}
|
||||
|
||||
eof = rs->nextResult();
|
||||
eof = pOp->nextResult();
|
||||
|
||||
if ((rows % 100) == 0){
|
||||
// "refresh" ownner transaction every 100th row
|
||||
|
@ -1161,8 +1159,7 @@ int Bank::performValidateGL(Uint64 glTime){
|
|||
return NDBT_FAILED;
|
||||
}
|
||||
|
||||
NdbResultSet * rs = pOp->readTuples();
|
||||
if( rs == 0 ) {
|
||||
if( pOp->readTuples() ) {
|
||||
ERR(pScanTrans->getNdbError());
|
||||
m_ndb.closeTransaction(pScanTrans);
|
||||
return NDBT_FAILED;
|
||||
|
@ -1241,7 +1238,7 @@ int Bank::performValidateGL(Uint64 glTime){
|
|||
int rows = 0;
|
||||
int countGlRecords = 0;
|
||||
int result = NDBT_OK;
|
||||
eof = rs->nextResult();
|
||||
eof = pOp->nextResult();
|
||||
|
||||
while(eof == 0){
|
||||
rows++;
|
||||
|
@ -1328,7 +1325,7 @@ int Bank::performValidateGL(Uint64 glTime){
|
|||
}
|
||||
|
||||
}
|
||||
eof = rs->nextResult();
|
||||
eof = pOp->nextResult();
|
||||
}
|
||||
if (eof == -1) {
|
||||
ERR(pScanTrans->getNdbError());
|
||||
|
@ -1425,8 +1422,7 @@ int Bank::getOldestPurgedGL(const Uint32 accountType,
|
|||
return NDBT_FAILED;
|
||||
}
|
||||
|
||||
NdbResultSet * rs = pOp->readTuples();
|
||||
if( rs == 0 ) {
|
||||
if( pOp->readTuples() ) {
|
||||
ERR(pScanTrans->getNdbError());
|
||||
m_ndb.closeTransaction(pScanTrans);
|
||||
return NDBT_FAILED;
|
||||
|
@ -1469,7 +1465,7 @@ int Bank::getOldestPurgedGL(const Uint32 accountType,
|
|||
|
||||
int eof;
|
||||
int rows = 0;
|
||||
eof = rs->nextResult();
|
||||
eof = pOp->nextResult();
|
||||
oldest = 0;
|
||||
|
||||
while(eof == 0){
|
||||
|
@ -1483,7 +1479,7 @@ int Bank::getOldestPurgedGL(const Uint32 accountType,
|
|||
if (t > oldest)
|
||||
oldest = t;
|
||||
}
|
||||
eof = rs->nextResult();
|
||||
eof = pOp->nextResult();
|
||||
}
|
||||
if (eof == -1) {
|
||||
ERR(pScanTrans->getNdbError());
|
||||
|
@ -1517,8 +1513,7 @@ int Bank::getOldestNotPurgedGL(Uint64 &oldest,
|
|||
return NDBT_FAILED;
|
||||
}
|
||||
|
||||
NdbResultSet * rs = pOp->readTuples();
|
||||
if( rs == 0 ) {
|
||||
if( pOp->readTuples() ) {
|
||||
ERR(pScanTrans->getNdbError());
|
||||
m_ndb.closeTransaction(pScanTrans);
|
||||
return NDBT_FAILED;
|
||||
|
@ -1561,7 +1556,7 @@ int Bank::getOldestNotPurgedGL(Uint64 &oldest,
|
|||
|
||||
int eof;
|
||||
int rows = 0;
|
||||
eof = rs->nextResult();
|
||||
eof = pOp->nextResult();
|
||||
oldest = (Uint64)-1;
|
||||
found = false;
|
||||
|
||||
|
@ -1578,7 +1573,7 @@ int Bank::getOldestNotPurgedGL(Uint64 &oldest,
|
|||
accountTypeId = a;
|
||||
}
|
||||
}
|
||||
eof = rs->nextResult();
|
||||
eof = pOp->nextResult();
|
||||
}
|
||||
if (eof == -1) {
|
||||
ERR(pScanTrans->getNdbError());
|
||||
|
@ -1614,8 +1609,7 @@ int Bank::checkNoTransactionsOlderThan(const Uint32 accountType,
|
|||
return NDBT_FAILED;
|
||||
}
|
||||
|
||||
NdbResultSet * rs = pOp->readTuples();
|
||||
if( rs == 0 ) {
|
||||
if( pOp->readTuples() ) {
|
||||
ERR(pScanTrans->getNdbError());
|
||||
m_ndb.closeTransaction(pScanTrans);
|
||||
return NDBT_FAILED;
|
||||
|
@ -1659,7 +1653,7 @@ int Bank::checkNoTransactionsOlderThan(const Uint32 accountType,
|
|||
int eof;
|
||||
int rows = 0;
|
||||
int found = 0;
|
||||
eof = rs->nextResult();
|
||||
eof = pOp->nextResult();
|
||||
|
||||
while(eof == 0){
|
||||
rows++;
|
||||
|
@ -1675,7 +1669,7 @@ int Bank::checkNoTransactionsOlderThan(const Uint32 accountType,
|
|||
<< " ti = " << ti << endl;
|
||||
found++;
|
||||
}
|
||||
eof = rs->nextResult();
|
||||
eof = pOp->nextResult();
|
||||
}
|
||||
if (eof == -1) {
|
||||
ERR(pScanTrans->getNdbError());
|
||||
|
@ -1858,8 +1852,7 @@ int Bank::findTransactionsToPurge(const Uint64 glTime,
|
|||
return NDBT_FAILED;
|
||||
}
|
||||
|
||||
NdbResultSet * rs = pOp->readTuplesExclusive();
|
||||
if( rs == 0 ) {
|
||||
if( pOp->readTuplesExclusive() ) {
|
||||
ERR(pScanTrans->getNdbError());
|
||||
m_ndb.closeTransaction(pScanTrans);
|
||||
return NDBT_FAILED;
|
||||
|
@ -1896,7 +1889,7 @@ int Bank::findTransactionsToPurge(const Uint64 glTime,
|
|||
int eof;
|
||||
int rows = 0;
|
||||
int rowsFound = 0;
|
||||
eof = rs->nextResult();
|
||||
eof = pOp->nextResult();
|
||||
|
||||
while(eof == 0){
|
||||
rows++;
|
||||
|
@ -1906,7 +1899,7 @@ int Bank::findTransactionsToPurge(const Uint64 glTime,
|
|||
if (a == accountType && t == glTime){
|
||||
rowsFound++;
|
||||
// One record found
|
||||
check = rs->deleteTuple(pTrans);
|
||||
check = pOp->deleteCurrentTuple(pTrans);
|
||||
if (check == -1){
|
||||
ERR(m_ndb.getNdbError());
|
||||
m_ndb.closeTransaction(pScanTrans);
|
||||
|
@ -1921,7 +1914,7 @@ int Bank::findTransactionsToPurge(const Uint64 glTime,
|
|||
return NDBT_FAILED;
|
||||
}
|
||||
}
|
||||
eof = rs->nextResult();
|
||||
eof = pOp->nextResult();
|
||||
}
|
||||
if (eof == -1) {
|
||||
ERR(pScanTrans->getNdbError());
|
||||
|
@ -2367,8 +2360,7 @@ int Bank::getSumAccounts(Uint32 &sumAccounts,
|
|||
return NDBT_FAILED;
|
||||
}
|
||||
|
||||
NdbResultSet * rs = pOp->readTuplesExclusive();
|
||||
if( rs == 0 ) {
|
||||
if( pOp->readTuplesExclusive() ) {
|
||||
ERR(pScanTrans->getNdbError());
|
||||
m_ndb.closeTransaction(pScanTrans);
|
||||
return NDBT_FAILED;
|
||||
|
@ -2403,7 +2395,7 @@ int Bank::getSumAccounts(Uint32 &sumAccounts,
|
|||
}
|
||||
|
||||
int eof;
|
||||
eof = rs->nextResult();
|
||||
eof = pOp->nextResult();
|
||||
|
||||
while(eof == 0){
|
||||
Uint32 b = balanceRec->u_32_value();
|
||||
|
@ -2415,7 +2407,7 @@ int Bank::getSumAccounts(Uint32 &sumAccounts,
|
|||
// << ", sum="<< sumAccounts << endl;
|
||||
|
||||
// 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){
|
||||
ERR(m_ndb.getNdbError());
|
||||
m_ndb.closeTransaction(pScanTrans);
|
||||
|
@ -2441,7 +2433,7 @@ int Bank::getSumAccounts(Uint32 &sumAccounts,
|
|||
return NDBT_FAILED;
|
||||
}
|
||||
|
||||
eof = rs->nextResult();
|
||||
eof = pOp->nextResult();
|
||||
}
|
||||
if (eof == -1) {
|
||||
ERR(pScanTrans->getNdbError());
|
||||
|
|
|
@ -342,8 +342,7 @@ int Bank::getBalanceForAccountType(const Uint32 accountType,
|
|||
return NDBT_FAILED;
|
||||
}
|
||||
|
||||
NdbResultSet* rs = pOp->readTuples();
|
||||
if( rs == 0 ) {
|
||||
if( pOp->readTuples() ) {
|
||||
ERR(pScanTrans->getNdbError());
|
||||
m_ndb.closeTransaction(pScanTrans);
|
||||
return NDBT_FAILED;
|
||||
|
@ -379,7 +378,7 @@ int Bank::getBalanceForAccountType(const Uint32 accountType,
|
|||
|
||||
int eof;
|
||||
int rows = 0;
|
||||
eof = rs->nextResult();
|
||||
eof = pOp->nextResult();
|
||||
|
||||
while(eof == 0){
|
||||
rows++;
|
||||
|
@ -391,7 +390,7 @@ int Bank::getBalanceForAccountType(const Uint32 accountType,
|
|||
balance += b;
|
||||
}
|
||||
|
||||
eof = rs->nextResult();
|
||||
eof = pOp->nextResult();
|
||||
}
|
||||
if (eof == -1) {
|
||||
ERR(pScanTrans->getNdbError());
|
||||
|
|
|
@ -8,18 +8,17 @@ S_Scan {
|
|||
const char * m_table;
|
||||
const char * m_index;
|
||||
NdbIndexScanOperation * m_scan;
|
||||
NdbResultSet * m_result;
|
||||
Uint32 metaid;
|
||||
Uint32 match_count;
|
||||
Uint32 row_count;
|
||||
};
|
||||
|
||||
static S_Scan g_scans[] = {
|
||||
{ "affiliatestometa", "ind_affiliatestometa", 0, 0, 0, 0, 0 },
|
||||
{ "media", "metaid", 0, 0, 0, 0, 0 },
|
||||
{ "meta", "PRIMARY", 0, 0, 0, 0, 0 },
|
||||
{ "artiststometamap", "PRIMARY", 0, 0, 0, 0, 0 },
|
||||
{ "subgenrestometamap", "metaid", 0, 0, 0, 0, 0 }
|
||||
{ "affiliatestometa", "ind_affiliatestometa", 0, 0, 0, 0 },
|
||||
{ "media", "metaid", 0, 0, 0, 0 },
|
||||
{ "meta", "PRIMARY", 0, 0, 0, 0 },
|
||||
{ "artiststometamap", "PRIMARY", 0, 0, 0, 0 },
|
||||
{ "subgenrestometamap", "metaid", 0, 0, 0, 0 }
|
||||
};
|
||||
|
||||
#define require(x) if(!(x)) { ndbout << "LINE: " << __LINE__ << endl;abort(); }
|
||||
|
@ -58,9 +57,8 @@ main(void){
|
|||
g_scans[i].m_table);
|
||||
NdbIndexScanOperation* scan = g_scans[i].m_scan;
|
||||
require(scan);
|
||||
g_scans[i].m_result = scan->readTuples(NdbScanOperation::LM_CommittedRead,
|
||||
0, 0, true);
|
||||
require(g_scans[i].m_result);
|
||||
require(scan->readTuples(NdbScanOperation::LM_CommittedRead,
|
||||
0, 0, true) == 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);
|
||||
|
||||
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)
|
||||
abort();
|
||||
|
||||
|
|
|
@ -739,10 +739,9 @@ verifyBlobTable(const Bcol& b, const Bval& v, Uint32 pk1, bool exists)
|
|||
NdbRecAttr* ra_pk;
|
||||
NdbRecAttr* ra_part;
|
||||
NdbRecAttr* ra_data;
|
||||
NdbResultSet* rs;
|
||||
CHK((g_con = g_ndb->startTransaction()) != 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_part = g_ops->getValue("PART")) != 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);
|
||||
while (1) {
|
||||
int ret;
|
||||
CHK((ret = rs->nextResult()) == 0 || ret == 1);
|
||||
CHK((ret = g_ops->nextResult()) == 0 || ret == 1);
|
||||
if (ret == 1)
|
||||
break;
|
||||
if (pk1 != ra_pk->u_32_value())
|
||||
|
@ -1104,14 +1103,13 @@ readScan(int style, bool idx)
|
|||
DBG("--- " << "readScan" << (idx ? "Idx" : "") << " " << stylename[style] << " ---");
|
||||
Tup tup;
|
||||
tup.alloc(); // allocate buffers
|
||||
NdbResultSet* rs;
|
||||
CHK((g_con = g_ndb->startTransaction()) != 0);
|
||||
if (! idx) {
|
||||
CHK((g_ops = g_con->getNdbScanOperation(g_opt.m_tname)) != 0);
|
||||
} else {
|
||||
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);
|
||||
if (g_opt.m_pk2len != 0)
|
||||
CHK(g_ops->getValue("PK2", tup.m_pk2) != 0);
|
||||
|
@ -1127,7 +1125,7 @@ readScan(int style, bool idx)
|
|||
int ret;
|
||||
tup.m_pk1 = (Uint32)-1;
|
||||
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)
|
||||
break;
|
||||
DBG("readScan" << (idx ? "Idx" : "") << " pk1=" << hex << tup.m_pk1);
|
||||
|
@ -1157,14 +1155,13 @@ updateScan(int style, bool idx)
|
|||
DBG("--- " << "updateScan" << (idx ? "Idx" : "") << " " << stylename[style] << " ---");
|
||||
Tup tup;
|
||||
tup.alloc(); // allocate buffers
|
||||
NdbResultSet* rs;
|
||||
CHK((g_con = g_ndb->startTransaction()) != 0);
|
||||
if (! idx) {
|
||||
CHK((g_ops = g_con->getNdbScanOperation(g_opt.m_tname)) != 0);
|
||||
} else {
|
||||
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);
|
||||
if (g_opt.m_pk2len != 0)
|
||||
CHK(g_ops->getValue("PK2", tup.m_pk2) != 0);
|
||||
|
@ -1174,7 +1171,7 @@ updateScan(int style, bool idx)
|
|||
int ret;
|
||||
tup.m_pk1 = (Uint32)-1;
|
||||
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)
|
||||
break;
|
||||
DBG("updateScan" << (idx ? "Idx" : "") << " pk1=" << hex << tup.m_pk1);
|
||||
|
@ -1183,7 +1180,7 @@ updateScan(int style, bool idx)
|
|||
// calculate new blob values
|
||||
calcBval(g_tups[k], false);
|
||||
tup.copyfrom(g_tups[k]);
|
||||
CHK((g_opr = rs->updateTuple()) != 0);
|
||||
CHK((g_opr = g_ops->updateCurrentTuple()) != 0);
|
||||
CHK(getBlobHandles(g_opr) == 0);
|
||||
if (style == 0) {
|
||||
CHK(setBlobValue(tup) == 0);
|
||||
|
@ -1210,14 +1207,13 @@ deleteScan(bool idx)
|
|||
{
|
||||
DBG("--- " << "deleteScan" << (idx ? "Idx" : "") << " ---");
|
||||
Tup tup;
|
||||
NdbResultSet* rs;
|
||||
CHK((g_con = g_ndb->startTransaction()) != 0);
|
||||
if (! idx) {
|
||||
CHK((g_ops = g_con->getNdbScanOperation(g_opt.m_tname)) != 0);
|
||||
} else {
|
||||
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);
|
||||
if (g_opt.m_pk2len != 0)
|
||||
CHK(g_ops->getValue("PK2", tup.m_pk2) != 0);
|
||||
|
@ -1227,11 +1223,11 @@ deleteScan(bool idx)
|
|||
int ret;
|
||||
tup.m_pk1 = (Uint32)-1;
|
||||
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)
|
||||
break;
|
||||
DBG("deleteScan" << (idx ? "Idx" : "") << " pk1=" << hex << tup.m_pk1);
|
||||
CHK(rs->deleteTuple() == 0);
|
||||
CHK(g_ops->deleteCurrentTuple() == 0);
|
||||
CHK(g_con->execute(NoCommit) == 0);
|
||||
Uint32 k = tup.m_pk1 - g_opt.m_pk1off;
|
||||
CHK(k < g_opt.m_rows && g_tups[k].m_exists);
|
||||
|
@ -1608,12 +1604,11 @@ testperf()
|
|||
// scan read char
|
||||
{
|
||||
DBG("--- scan read char ---");
|
||||
NdbResultSet* rs;
|
||||
Uint32 a;
|
||||
char b[20];
|
||||
CHK((g_con = g_ndb->startTransaction()) != 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(cB, b) != 0);
|
||||
CHK(g_con->execute(NoCommit) == 0);
|
||||
|
@ -1623,7 +1618,7 @@ testperf()
|
|||
a = (Uint32)-1;
|
||||
b[0] = 0;
|
||||
int ret;
|
||||
CHK((ret = rs->nextResult(true)) == 0 || ret == 1);
|
||||
CHK((ret = g_ops->nextResult(true)) == 0 || ret == 1);
|
||||
if (ret == 1)
|
||||
break;
|
||||
CHK(a < g_opt.m_rowsperf && strcmp(b, "b") == 0);
|
||||
|
@ -1638,12 +1633,11 @@ testperf()
|
|||
// scan read text
|
||||
{
|
||||
DBG("--- read text ---");
|
||||
NdbResultSet* rs;
|
||||
Uint32 a;
|
||||
char c[20];
|
||||
CHK((g_con = g_ndb->startTransaction()) != 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_bh1 = g_ops->getBlobHandle(cC)) != 0);
|
||||
CHK(g_con->execute(NoCommit) == 0);
|
||||
|
@ -1653,7 +1647,7 @@ testperf()
|
|||
a = (Uint32)-1;
|
||||
c[0] = 0;
|
||||
int ret;
|
||||
CHK((ret = rs->nextResult(true)) == 0 || ret == 1);
|
||||
CHK((ret = g_ops->nextResult(true)) == 0 || ret == 1);
|
||||
if (ret == 1)
|
||||
break;
|
||||
Uint32 m = 20;
|
||||
|
|
|
@ -443,9 +443,9 @@ testcase(int flag)
|
|||
if ((con = ndb->startTransaction()) == 0)
|
||||
return ndberror("startTransaction key=%d", key);
|
||||
if ((op = sop = con->getNdbScanOperation(tab)) == 0)
|
||||
return ndberror("getNdbOperation key=%d", key);
|
||||
if ((rs = sop->readTuples(1)) == 0)
|
||||
return ndberror("openScanRead key=%d", key);
|
||||
return ndberror("getNdbOperation key=%d", key);
|
||||
if (sop->readTuples(1))
|
||||
return ndberror("openScanRead key=%d", key);
|
||||
{
|
||||
col& c = ccol[0];
|
||||
if (op->load_const_u32(1, key) < 0)
|
||||
|
@ -488,7 +488,7 @@ testcase(int flag)
|
|||
if (con->execute(NoCommit) < 0)
|
||||
return ndberror("executeScan key=%d", key);
|
||||
int ret, cnt = 0;
|
||||
while ((ret = rs->nextResult()) == 0) {
|
||||
while ((ret = sop->nextResult()) == 0) {
|
||||
if (key != newkey)
|
||||
return ndberror("unexpected key=%d newkey=%d", key, newkey);
|
||||
for (i = 1; i < attrcnt; i++) {
|
||||
|
|
|
@ -91,7 +91,6 @@ struct Thr {
|
|||
NdbConnection* m_con;
|
||||
NdbScanOperation* m_scanop;
|
||||
NdbIndexScanOperation* m_indexscanop;
|
||||
NdbResultSet* m_rs;
|
||||
//
|
||||
Thr(int no);
|
||||
~Thr();
|
||||
|
@ -136,7 +135,6 @@ Thr::Thr(int no)
|
|||
m_con = 0;
|
||||
m_scanop = 0;
|
||||
m_indexscanop = 0;
|
||||
m_rs = 0;
|
||||
}
|
||||
|
||||
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);
|
||||
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("B", (char*)&wl1822_bufB) != 0);
|
||||
CHN(con, con->execute(NoCommit) == 0);
|
||||
|
@ -383,7 +381,7 @@ wl1822_tx2_scanXY(Thr& thr)
|
|||
DBG("before row " << row);
|
||||
int ret;
|
||||
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);
|
||||
CHK(wl1822_bufA == wl1822_valA[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;
|
||||
NdbConnection* con = thr.m_con;
|
||||
NdbScanOperation* scanop = thr.m_scanop;
|
||||
NdbResultSet* rs = thr.m_rs;
|
||||
assert(ndb != 0 && con != 0 && scanop != 0 && rs != 0);
|
||||
assert(ndb != 0 && con != 0 && scanop != 0);
|
||||
unsigned row = 2;
|
||||
while (true) {
|
||||
DBG("before row " << row);
|
||||
int ret;
|
||||
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)
|
||||
break;
|
||||
DBG("got row " << row << " a=" << wl1822_bufA << " b=" << wl1822_bufB);
|
||||
|
|
|
@ -1145,20 +1145,18 @@ runUniqueNullTransactions(NDBT_Context* ctx, NDBT_Step* step){
|
|||
pTrans = pNdb->startTransaction();
|
||||
NdbScanOperation * sOp;
|
||||
NdbOperation * uOp;
|
||||
NdbResultSet * rs;
|
||||
int eof;
|
||||
if(!pTrans) goto done;
|
||||
sOp = pTrans->getNdbScanOperation(pTab->getName());
|
||||
if(!sOp) goto done;
|
||||
rs = sOp->readTuples(NdbScanOperation::LM_Exclusive);
|
||||
if(!rs) goto done;
|
||||
if(sOp->readTuples(NdbScanOperation::LM_Exclusive)) goto done;
|
||||
if(pTrans->execute(NoCommit) == -1) goto done;
|
||||
while((eof = rs->nextResult(true)) == 0){
|
||||
while((eof = sOp->nextResult(true)) == 0){
|
||||
do {
|
||||
NdbOperation * uOp = rs->updateTuple();
|
||||
NdbOperation * uOp = sOp->updateCurrentTuple();
|
||||
if(uOp == 0) goto done;
|
||||
uOp->setValue(colId, 0);
|
||||
} while((eof = rs->nextResult(false)) == 0);
|
||||
} while((eof = sOp->nextResult(false)) == 0);
|
||||
eof = pTrans->execute(Commit);
|
||||
if(eof == -1) goto done;
|
||||
}
|
||||
|
|
|
@ -679,14 +679,13 @@ struct Con {
|
|||
NdbOperation* m_op;
|
||||
NdbScanOperation* m_scanop;
|
||||
NdbIndexScanOperation* m_indexscanop;
|
||||
NdbResultSet* m_resultset;
|
||||
enum ScanMode { ScanNo = 0, Committed, Latest, Exclusive };
|
||||
ScanMode m_scanmode;
|
||||
enum ErrType { ErrNone = 0, ErrDeadlock, ErrOther };
|
||||
ErrType m_errtype;
|
||||
Con() :
|
||||
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() {
|
||||
if (m_tx != 0)
|
||||
closeTransaction();
|
||||
|
@ -836,7 +835,7 @@ Con::openScanRead(unsigned scanbat, unsigned scanpar)
|
|||
{
|
||||
assert(m_tx != 0 && m_op != 0);
|
||||
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;
|
||||
}
|
||||
|
||||
|
@ -845,7 +844,7 @@ Con::openScanExclusive(unsigned scanbat, unsigned scanpar)
|
|||
{
|
||||
assert(m_tx != 0 && m_op != 0);
|
||||
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;
|
||||
}
|
||||
|
||||
|
@ -860,8 +859,8 @@ int
|
|||
Con::nextScanResult(bool fetchAllowed)
|
||||
{
|
||||
int ret;
|
||||
assert(m_resultset != 0);
|
||||
CHKCON((ret = m_resultset->nextResult(fetchAllowed)) != -1, *this);
|
||||
assert(m_scanop != 0);
|
||||
CHKCON((ret = m_scanop->nextResult(fetchAllowed)) != -1, *this);
|
||||
assert(ret == 0 || ret == 1 || (! fetchAllowed && ret == 2));
|
||||
return ret;
|
||||
}
|
||||
|
@ -886,7 +885,7 @@ int
|
|||
Con::updateScanTuple(Con& con2)
|
||||
{
|
||||
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;
|
||||
}
|
||||
|
||||
|
@ -894,16 +893,16 @@ int
|
|||
Con::deleteScanTuple(Con& con2)
|
||||
{
|
||||
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;
|
||||
}
|
||||
|
||||
void
|
||||
Con::closeScan()
|
||||
{
|
||||
assert(m_resultset != 0);
|
||||
m_resultset->close();
|
||||
m_scanop = 0, m_indexscanop = 0, m_resultset = 0;
|
||||
assert(m_scanop != 0);
|
||||
m_scanop->close();
|
||||
m_scanop = 0, m_indexscanop = 0;
|
||||
|
||||
}
|
||||
|
||||
|
@ -913,7 +912,7 @@ Con::closeTransaction()
|
|||
assert(m_ndb != 0 && m_tx != 0);
|
||||
m_ndb->closeTransaction(m_tx);
|
||||
m_tx = 0, m_op = 0;
|
||||
m_scanop = 0, m_indexscanop = 0, m_resultset = 0;
|
||||
m_scanop = 0, m_indexscanop = 0;
|
||||
}
|
||||
|
||||
void
|
||||
|
|
|
@ -266,7 +266,6 @@ run_read(){
|
|||
NdbScanOperation * pSp;
|
||||
NdbIndexOperation * pUp;
|
||||
NdbIndexScanOperation * pIp;
|
||||
NdbResultSet * rs = (NdbResultSet*)~0;
|
||||
|
||||
Uint32 start_row = rand() % (rows - range);
|
||||
Uint32 stop_row = start_row + range;
|
||||
|
@ -319,27 +318,27 @@ run_read(){
|
|||
}
|
||||
break;
|
||||
case 4:
|
||||
pOp = pIp = pTrans->getNdbIndexScanOperation(g_ordered,g_table);
|
||||
rs = pIp->readTuples(NdbScanOperation::LM_CommittedRead, 0, 0);
|
||||
pOp = pSp = pIp = pTrans->getNdbIndexScanOperation(g_ordered,g_table);
|
||||
pIp->readTuples(NdbScanOperation::LM_CommittedRead, 0, 0);
|
||||
check = pIp->setBound(pk, NdbIndexScanOperation::BoundEQ, &start_row);
|
||||
break;
|
||||
case 5:
|
||||
pOp = pIp = pTrans->getNdbIndexScanOperation(g_ordered,g_table);
|
||||
rs = pIp->readTuples(NdbScanOperation::LM_CommittedRead, 0, 0);
|
||||
pOp = pSp = pIp = pTrans->getNdbIndexScanOperation(g_ordered,g_table);
|
||||
pIp->readTuples(NdbScanOperation::LM_CommittedRead, 0, 0);
|
||||
check = pIp->setBound(pk, NdbIndexScanOperation::BoundLE, &start_row);
|
||||
check = pIp->setBound(pk, NdbIndexScanOperation::BoundGT, &stop_row);
|
||||
start_row = stop_row;
|
||||
break;
|
||||
case 6:
|
||||
pOp = pIp = pTrans->getNdbIndexScanOperation(g_ordered,g_table);
|
||||
rs = pIp->readTuples(NdbScanOperation::LM_CommittedRead, 0, 0, true);
|
||||
pOp = pSp = pIp = pTrans->getNdbIndexScanOperation(g_ordered,g_table);
|
||||
pIp->readTuples(NdbScanOperation::LM_CommittedRead, 0, 0, true);
|
||||
check = pIp->setBound(pk, NdbIndexScanOperation::BoundLE, &start_row);
|
||||
check = pIp->setBound(pk, NdbIndexScanOperation::BoundGT, &stop_row);
|
||||
start_row = stop_row;
|
||||
break;
|
||||
case 7:
|
||||
pOp = pSp = pTrans->getNdbScanOperation(g_table);
|
||||
rs = pSp->readTuples(NdbScanOperation::LM_CommittedRead, 0, 0);
|
||||
pSp->readTuples(NdbScanOperation::LM_CommittedRead, 0, 0);
|
||||
NdbScanFilter filter(pOp) ;
|
||||
filter.begin(NdbScanFilter::AND);
|
||||
filter.ge(pk, start_row);
|
||||
|
@ -355,7 +354,6 @@ run_read(){
|
|||
ndbout << pTrans->getNdbError() << endl;
|
||||
}
|
||||
assert(check == 0);
|
||||
assert(rs);
|
||||
|
||||
for(int j = 0; j<g_tab->getNoOfColumns(); j++){
|
||||
res = pOp->getValue(j);
|
||||
|
@ -368,7 +366,7 @@ run_read(){
|
|||
}
|
||||
assert(check == 0);
|
||||
if(g_paramters[P_OPER].value >= 4){
|
||||
while((check = rs->nextResult(true)) == 0){
|
||||
while((check = pSp->nextResult(true)) == 0){
|
||||
cnt++;
|
||||
}
|
||||
|
||||
|
@ -377,13 +375,13 @@ run_read(){
|
|||
return -1;
|
||||
}
|
||||
assert(check == 1);
|
||||
rs->close();
|
||||
pSp->close();
|
||||
}
|
||||
}
|
||||
assert(g_paramters[P_OPER].value < 4 || (cnt == range));
|
||||
|
||||
|
||||
pTrans->close();
|
||||
|
||||
|
||||
stop = NdbTick_CurrentMillisecond();
|
||||
g_times[g_paramters[P_OPER].value] += (stop - start1);
|
||||
return 0;
|
||||
|
|
|
@ -999,8 +999,7 @@ int runScanRestart(NDBT_Context* ctx, NDBT_Step* step){
|
|||
return NDBT_FAILED;
|
||||
}
|
||||
|
||||
NdbResultSet* rs = pOp->readTuples();
|
||||
if( rs == 0 ) {
|
||||
if( pOp->readTuples() ) {
|
||||
ERR(pCon->getNdbError());
|
||||
return NDBT_FAILED;
|
||||
}
|
||||
|
@ -1028,7 +1027,7 @@ int runScanRestart(NDBT_Context* ctx, NDBT_Step* step){
|
|||
|
||||
int res;
|
||||
int row = 0;
|
||||
while(row < record && (res = rs->nextResult()) == 0) {
|
||||
while(row < record && (res = pOp->nextResult()) == 0) {
|
||||
if(calc.verifyRowValues(&tmpRow) != 0){
|
||||
abort();
|
||||
return NDBT_FAILED;
|
||||
|
@ -1041,14 +1040,14 @@ int runScanRestart(NDBT_Context* ctx, NDBT_Step* step){
|
|||
return NDBT_FAILED;
|
||||
}
|
||||
g_info << " restarting" << endl;
|
||||
if((res = rs->restart()) != 0){
|
||||
if((res = pOp->restart()) != 0){
|
||||
ERR(pCon->getNdbError());
|
||||
abort();
|
||||
return NDBT_FAILED;
|
||||
}
|
||||
|
||||
row = 0;
|
||||
while((res = rs->nextResult()) == 0) {
|
||||
while((res = pOp->nextResult()) == 0) {
|
||||
if(calc.verifyRowValues(&tmpRow) != 0){
|
||||
abort();
|
||||
return NDBT_FAILED;
|
||||
|
|
|
@ -198,7 +198,6 @@ run_scan(){
|
|||
NdbScanOperation * pOp = 0;
|
||||
NdbIndexScanOperation * pIOp = 0;
|
||||
NdbConnection * pTrans = 0;
|
||||
NdbResultSet * rs = 0;
|
||||
int check = 0;
|
||||
|
||||
for(int i = 0; i<iter; i++){
|
||||
|
@ -230,13 +229,13 @@ run_scan(){
|
|||
if(g_paramters[P_ACCESS].value == 0){
|
||||
pOp = pTrans->getNdbScanOperation(g_tablename);
|
||||
assert(pOp);
|
||||
rs = pOp->readTuples(lm, bat, par);
|
||||
pOp->readTuples(lm, bat, par);
|
||||
} else {
|
||||
if(g_paramters[P_RESET].value == 0 || pIOp == 0)
|
||||
{
|
||||
pOp= pIOp= pTrans->getNdbIndexScanOperation(g_indexname, g_tablename);
|
||||
bool ord = g_paramters[P_ACCESS].value == 2;
|
||||
rs = pIOp->readTuples(lm, bat, par, ord);
|
||||
pIOp->readTuples(lm, bat, par, ord);
|
||||
}
|
||||
else
|
||||
{
|
||||
|
@ -270,7 +269,7 @@ run_scan(){
|
|||
{
|
||||
int row = rand() % tot;
|
||||
pIOp->setBound((Uint32)0, NdbIndexScanOperation::BoundEQ, &row);
|
||||
pIOp->end_of_bound();
|
||||
pIOp->end_of_bound(i);
|
||||
}
|
||||
if(g_paramters[P_RESET].value == 2)
|
||||
goto execute;
|
||||
|
@ -279,7 +278,6 @@ run_scan(){
|
|||
}
|
||||
}
|
||||
assert(pOp);
|
||||
assert(rs);
|
||||
|
||||
switch(g_paramters[P_FILT].value){
|
||||
case 0: // All
|
||||
|
@ -323,10 +321,10 @@ execute:
|
|||
check = pTrans->execute(NoCommit);
|
||||
assert(check == 0);
|
||||
int fetch = g_paramters[P_FETCH].value;
|
||||
while((check = rs->nextResult(true)) == 0){
|
||||
while((check = pOp->nextResult(true)) == 0){
|
||||
do {
|
||||
rows++;
|
||||
} while(!fetch && ((check = rs->nextResult(false)) == 0));
|
||||
} while(!fetch && ((check = pOp->nextResult(false)) == 0));
|
||||
if(check == -1){
|
||||
err(pTrans->getNdbError());
|
||||
return -1;
|
||||
|
|
|
@ -305,7 +305,7 @@ int HugoOperations::execute_Commit(Ndb* pNdb,
|
|||
m_executed_result_sets.push_back(m_result_sets[i]);
|
||||
|
||||
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();
|
||||
switch(res){
|
||||
case 1:
|
||||
|
@ -354,7 +354,7 @@ int HugoOperations::execute_NoCommit(Ndb* pNdb, AbortOption eao){
|
|||
m_executed_result_sets.push_back(m_result_sets[i]);
|
||||
|
||||
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();
|
||||
switch(res){
|
||||
case 1:
|
||||
|
@ -700,12 +700,10 @@ HugoOperations::scanReadRecords(Ndb* pNdb, NdbScanOperation::LockMode lm,
|
|||
if(!pOp)
|
||||
return -1;
|
||||
|
||||
NdbResultSet * rs = pOp->readTuples(lm, 1, 1);
|
||||
|
||||
if(!rs){
|
||||
if(pOp->readTuples(lm, 1, 1)){
|
||||
return -1;
|
||||
}
|
||||
|
||||
|
||||
for(int a = 0; a<tab.getNoOfColumns(); a++){
|
||||
if((rows[0]->attributeStore(a) =
|
||||
pOp->getValue(tab.getColumn(a)->getName())) == 0) {
|
||||
|
@ -713,8 +711,8 @@ HugoOperations::scanReadRecords(Ndb* pNdb, NdbScanOperation::LockMode lm,
|
|||
return NDBT_FAILED;
|
||||
}
|
||||
}
|
||||
|
||||
RsPair p = {rs, records};
|
||||
|
||||
RsPair p = {pOp, records};
|
||||
m_result_sets.push_back(p);
|
||||
|
||||
return 0;
|
||||
|
|
|
@ -72,10 +72,7 @@ HugoTransactions::scanReadRecords(Ndb* pNdb,
|
|||
return NDBT_FAILED;
|
||||
}
|
||||
|
||||
NdbResultSet * rs;
|
||||
rs = pOp ->readTuples(lm);
|
||||
|
||||
if( rs == 0 ) {
|
||||
if( pOp ->readTuples(lm) ) {
|
||||
ERR(pTrans->getNdbError());
|
||||
pNdb->closeTransaction(pTrans);
|
||||
return NDBT_FAILED;
|
||||
|
@ -124,7 +121,7 @@ HugoTransactions::scanReadRecords(Ndb* pNdb,
|
|||
|
||||
int eof;
|
||||
int rows = 0;
|
||||
while((eof = rs->nextResult(true)) == 0){
|
||||
while((eof = pOp->nextResult(true)) == 0){
|
||||
rows++;
|
||||
if (calc.verifyRowValues(&row) != 0){
|
||||
pNdb->closeTransaction(pTrans);
|
||||
|
@ -134,7 +131,7 @@ HugoTransactions::scanReadRecords(Ndb* pNdb,
|
|||
if (abortCount == rows && abortTrans == true){
|
||||
ndbout << "Scan is aborted" << endl;
|
||||
g_info << "Scan is aborted" << endl;
|
||||
rs->close();
|
||||
pOp->close();
|
||||
if( check == -1 ) {
|
||||
ERR(pTrans->getNdbError());
|
||||
pNdb->closeTransaction(pTrans);
|
||||
|
@ -228,10 +225,7 @@ HugoTransactions::scanReadRecords(Ndb* pNdb,
|
|||
return NDBT_FAILED;
|
||||
}
|
||||
|
||||
NdbResultSet * rs;
|
||||
rs = pOp ->readTuples(lm, 0, parallelism, sorted);
|
||||
|
||||
if( rs == 0 ) {
|
||||
if( pOp ->readTuples(lm, 0, parallelism, sorted) ) {
|
||||
ERR(pTrans->getNdbError());
|
||||
pNdb->closeTransaction(pTrans);
|
||||
return NDBT_FAILED;
|
||||
|
@ -280,7 +274,7 @@ HugoTransactions::scanReadRecords(Ndb* pNdb,
|
|||
|
||||
int eof;
|
||||
int rows = 0;
|
||||
while((eof = rs->nextResult(true)) == 0){
|
||||
while((eof = pOp->nextResult(true)) == 0){
|
||||
rows++;
|
||||
if (calc.verifyRowValues(&row) != 0){
|
||||
pNdb->closeTransaction(pTrans);
|
||||
|
@ -290,7 +284,7 @@ HugoTransactions::scanReadRecords(Ndb* pNdb,
|
|||
if (abortCount == rows && abortTrans == true){
|
||||
ndbout << "Scan is aborted" << endl;
|
||||
g_info << "Scan is aborted" << endl;
|
||||
rs->close();
|
||||
pOp->close();
|
||||
if( check == -1 ) {
|
||||
ERR(pTrans->getNdbError());
|
||||
pNdb->closeTransaction(pTrans);
|
||||
|
@ -732,8 +726,7 @@ HugoTransactions::scanUpdateRecords3(Ndb* pNdb,
|
|||
return NDBT_FAILED;
|
||||
}
|
||||
|
||||
NdbResultSet *rs = pOp->readTuplesExclusive(parallelism);
|
||||
if( rs == 0 ) {
|
||||
if( pOp->readTuplesExclusive(parallelism) ) {
|
||||
ERR(pTrans->getNdbError());
|
||||
pNdb->closeTransaction(pTrans);
|
||||
return NDBT_FAILED;
|
||||
|
@ -771,10 +764,10 @@ HugoTransactions::scanUpdateRecords3(Ndb* pNdb,
|
|||
}
|
||||
|
||||
int rows = 0;
|
||||
while((check = rs->nextResult(true)) == 0){
|
||||
while((check = pOp->nextResult(true)) == 0){
|
||||
do {
|
||||
rows++;
|
||||
NdbOperation* pUp = rs->updateTuple();
|
||||
NdbOperation* pUp = pOp->updateCurrentTuple();
|
||||
if(pUp == 0){
|
||||
ERR(pTrans->getNdbError());
|
||||
pNdb->closeTransaction(pTrans);
|
||||
|
@ -798,7 +791,7 @@ HugoTransactions::scanUpdateRecords3(Ndb* pNdb,
|
|||
pNdb->closeTransaction(pTrans);
|
||||
return NDBT_OK;
|
||||
}
|
||||
} while((check = rs->nextResult(false)) == 0);
|
||||
} while((check = pOp->nextResult(false)) == 0);
|
||||
|
||||
if(check != -1){
|
||||
check = pTrans->execute(Commit);
|
||||
|
@ -2133,7 +2126,6 @@ HugoTransactions::indexReadRecords(Ndb* pNdb,
|
|||
NdbConnection *pTrans;
|
||||
NdbOperation *pOp;
|
||||
NdbIndexScanOperation *sOp;
|
||||
NdbResultSet * rs;
|
||||
|
||||
const NdbDictionary::Index* pIndex
|
||||
= pNdb->getDictionary()->getIndex(idxName, tab.getName());
|
||||
|
@ -2189,9 +2181,7 @@ HugoTransactions::indexReadRecords(Ndb* pNdb,
|
|||
pNdb->closeTransaction(pTrans);
|
||||
return NDBT_FAILED;
|
||||
}
|
||||
|
||||
check = 0;
|
||||
rs = sOp->readTuples();
|
||||
}
|
||||
|
||||
if( check == -1 ) {
|
||||
|
@ -2223,7 +2213,7 @@ HugoTransactions::indexReadRecords(Ndb* pNdb,
|
|||
}
|
||||
|
||||
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 ) {
|
||||
const NdbError err = pTrans->getNdbError();
|
||||
|
||||
|
@ -2254,7 +2244,7 @@ HugoTransactions::indexReadRecords(Ndb* pNdb,
|
|||
reads++;
|
||||
r++;
|
||||
}
|
||||
if(ordered && rs->nextResult(true) == 0){
|
||||
if(ordered && sOp->nextResult(true) == 0){
|
||||
ndbout << "Error when comparing records "
|
||||
<< " - index op next_result to many" << endl;
|
||||
pNdb->closeTransaction(pTrans);
|
||||
|
@ -2284,7 +2274,6 @@ HugoTransactions::indexUpdateRecords(Ndb* pNdb,
|
|||
NdbConnection *pTrans;
|
||||
NdbOperation *pOp;
|
||||
NdbScanOperation * sOp;
|
||||
NdbResultSet * rs;
|
||||
|
||||
const NdbDictionary::Index* pIndex
|
||||
= pNdb->getDictionary()->getIndex(idxName, tab.getName());
|
||||
|
@ -2341,7 +2330,7 @@ HugoTransactions::indexUpdateRecords(Ndb* pNdb,
|
|||
}
|
||||
|
||||
check = 0;
|
||||
rs = sOp->readTuplesExclusive();
|
||||
sOp->readTuplesExclusive();
|
||||
}
|
||||
|
||||
// Define primary keys
|
||||
|
@ -2367,7 +2356,7 @@ HugoTransactions::indexUpdateRecords(Ndb* pNdb,
|
|||
}
|
||||
|
||||
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 ) {
|
||||
const NdbError err = pTrans->getNdbError();
|
||||
ERR(err);
|
||||
|
@ -2400,7 +2389,7 @@ HugoTransactions::indexUpdateRecords(Ndb* pNdb,
|
|||
pUpdOp = pTrans->getNdbIndexOperation(idxName, tab.getName());
|
||||
check = (pUpdOp == 0 ? -1 : pUpdOp->updateTuple());
|
||||
} else {
|
||||
pUpdOp = rs->updateTuple();
|
||||
pUpdOp = sOp->updateCurrentTuple();
|
||||
}
|
||||
|
||||
if (pUpdOp == NULL) {
|
||||
|
|
|
@ -394,8 +394,7 @@ UtilTransactions::clearTable3(Ndb* pNdb,
|
|||
goto failed;
|
||||
}
|
||||
|
||||
NdbResultSet * rs = pOp->readTuplesExclusive(par);
|
||||
if( rs == 0 ) {
|
||||
if( pOp->readTuplesExclusive(par) ) {
|
||||
err = pTrans->getNdbError();
|
||||
goto failed;
|
||||
}
|
||||
|
@ -411,13 +410,13 @@ UtilTransactions::clearTable3(Ndb* pNdb,
|
|||
goto failed;
|
||||
}
|
||||
|
||||
while((check = rs->nextResult(true)) == 0){
|
||||
while((check = pOp->nextResult(true)) == 0){
|
||||
do {
|
||||
if (rs->deleteTuple() != 0){
|
||||
if (pOp->deleteCurrentTuple() != 0){
|
||||
goto failed;
|
||||
}
|
||||
deletedRows++;
|
||||
} while((check = rs->nextResult(false)) == 0);
|
||||
} while((check = pOp->nextResult(false)) == 0);
|
||||
|
||||
if(check != -1){
|
||||
check = pTrans->execute(Commit);
|
||||
|
@ -502,9 +501,7 @@ UtilTransactions::copyTableData(Ndb* pNdb,
|
|||
return NDBT_FAILED;
|
||||
}
|
||||
|
||||
NdbResultSet* rs = pOp->readTuples(NdbScanOperation::LM_Read,
|
||||
parallelism);
|
||||
if( check == -1 ) {
|
||||
if( pOp->readTuples(NdbScanOperation::LM_Read, parallelism) ) {
|
||||
ERR(pTrans->getNdbError());
|
||||
pNdb->closeTransaction(pTrans);
|
||||
return NDBT_FAILED;
|
||||
|
@ -535,14 +532,14 @@ UtilTransactions::copyTableData(Ndb* pNdb,
|
|||
}
|
||||
|
||||
int eof;
|
||||
while((eof = rs->nextResult(true)) == 0){
|
||||
while((eof = pOp->nextResult(true)) == 0){
|
||||
do {
|
||||
insertedRows++;
|
||||
if (addRowToInsert(pNdb, pTrans, row, destName) != 0){
|
||||
pNdb->closeTransaction(pTrans);
|
||||
return NDBT_FAILED;
|
||||
}
|
||||
} while((eof = rs->nextResult(false)) == 0);
|
||||
} while((eof = pOp->nextResult(false)) == 0);
|
||||
|
||||
check = pTrans->execute(Commit);
|
||||
pTrans->restart();
|
||||
|
@ -669,8 +666,7 @@ UtilTransactions::scanReadRecords(Ndb* pNdb,
|
|||
return NDBT_FAILED;
|
||||
}
|
||||
|
||||
NdbResultSet * rs = pOp->readTuples(lm, 0, parallelism);
|
||||
if( rs == 0 ) {
|
||||
if( pOp->readTuples(lm, 0, parallelism) ) {
|
||||
ERR(pTrans->getNdbError());
|
||||
pNdb->closeTransaction(pTrans);
|
||||
return NDBT_FAILED;
|
||||
|
@ -718,7 +714,7 @@ UtilTransactions::scanReadRecords(Ndb* pNdb,
|
|||
int rows = 0;
|
||||
|
||||
|
||||
while((eof = rs->nextResult()) == 0){
|
||||
while((eof = pOp->nextResult()) == 0){
|
||||
rows++;
|
||||
|
||||
// Call callback for each record returned
|
||||
|
@ -782,8 +778,7 @@ UtilTransactions::selectCount(Ndb* pNdb,
|
|||
return NDBT_FAILED;
|
||||
}
|
||||
|
||||
NdbResultSet * rs = pOp->readTuples(lm);
|
||||
if( rs == 0) {
|
||||
if( pOp->readTuples(lm) ) {
|
||||
ERR(pTrans->getNdbError());
|
||||
pNdb->closeTransaction(pTrans);
|
||||
return NDBT_FAILED;
|
||||
|
@ -815,7 +810,7 @@ UtilTransactions::selectCount(Ndb* pNdb,
|
|||
int rows = 0;
|
||||
|
||||
|
||||
while((eof = rs->nextResult()) == 0){
|
||||
while((eof = pOp->nextResult()) == 0){
|
||||
rows++;
|
||||
}
|
||||
if (eof == -1) {
|
||||
|
@ -948,14 +943,14 @@ UtilTransactions::scanAndCompareUniqueIndex(Ndb* pNdb,
|
|||
return NDBT_FAILED;
|
||||
}
|
||||
|
||||
NdbResultSet* rs;
|
||||
int rs;
|
||||
if(transactional){
|
||||
rs = pOp->readTuples(NdbScanOperation::LM_Read, 0, parallelism);
|
||||
} else {
|
||||
rs = pOp->readTuples(NdbScanOperation::LM_CommittedRead, 0, parallelism);
|
||||
}
|
||||
|
||||
if( rs == 0 ) {
|
||||
if( rs != 0 ) {
|
||||
ERR(pTrans->getNdbError());
|
||||
pNdb->closeTransaction(pTrans);
|
||||
return NDBT_FAILED;
|
||||
|
@ -998,7 +993,7 @@ UtilTransactions::scanAndCompareUniqueIndex(Ndb* pNdb,
|
|||
int rows = 0;
|
||||
|
||||
|
||||
while((eof = rs->nextResult()) == 0){
|
||||
while((eof = pOp->nextResult()) == 0){
|
||||
rows++;
|
||||
|
||||
// ndbout << row.c_str().c_str() << endl;
|
||||
|
@ -1046,7 +1041,6 @@ UtilTransactions::readRowFromTableAndIndex(Ndb* pNdb,
|
|||
const int retryMax = 100;
|
||||
int check, a;
|
||||
NdbConnection *pTrans1=NULL;
|
||||
NdbResultSet *cursor= NULL;
|
||||
NdbOperation *pOp;
|
||||
|
||||
int return_code= NDBT_FAILED;
|
||||
|
@ -1174,7 +1168,7 @@ UtilTransactions::readRowFromTableAndIndex(Ndb* pNdb,
|
|||
if (pIndexOp) {
|
||||
not_ok = pIndexOp->readTuple() == -1;
|
||||
} else {
|
||||
not_ok = (cursor= pScanOp->readTuples()) == 0;
|
||||
not_ok = pScanOp->readTuples();
|
||||
}
|
||||
|
||||
if( not_ok ) {
|
||||
|
@ -1251,7 +1245,7 @@ UtilTransactions::readRowFromTableAndIndex(Ndb* pNdb,
|
|||
*/
|
||||
if(!null_found){
|
||||
if (pScanOp) {
|
||||
if (cursor->nextResult() != 0){
|
||||
if (pScanOp->nextResult() != 0){
|
||||
const NdbError err = pTrans1->getNdbError();
|
||||
ERR(err);
|
||||
ndbout << "Error when comparing records - index op next_result missing" << endl;
|
||||
|
@ -1266,7 +1260,7 @@ UtilTransactions::readRowFromTableAndIndex(Ndb* pNdb,
|
|||
goto close_all;
|
||||
}
|
||||
if (pScanOp) {
|
||||
if (cursor->nextResult() == 0){
|
||||
if (pScanOp->nextResult() == 0){
|
||||
ndbout << "Error when comparing records - index op next_result to many" << endl;
|
||||
ndbout << "row: " << row.c_str().c_str() << endl;
|
||||
goto close_all;
|
||||
|
@ -1278,8 +1272,6 @@ UtilTransactions::readRowFromTableAndIndex(Ndb* pNdb,
|
|||
}
|
||||
|
||||
close_all:
|
||||
if (cursor)
|
||||
cursor->close();
|
||||
if (pTrans1)
|
||||
pNdb->closeTransaction(pTrans1);
|
||||
|
||||
|
@ -1298,7 +1290,6 @@ UtilTransactions::verifyOrderedIndex(Ndb* pNdb,
|
|||
NdbConnection *pTrans;
|
||||
NdbScanOperation *pOp;
|
||||
NdbIndexScanOperation * iop = 0;
|
||||
NdbResultSet* cursor= 0;
|
||||
|
||||
NDBT_ResultRow scanRow(tab);
|
||||
NDBT_ResultRow pkRow(tab);
|
||||
|
@ -1337,10 +1328,7 @@ UtilTransactions::verifyOrderedIndex(Ndb* pNdb,
|
|||
return NDBT_FAILED;
|
||||
}
|
||||
|
||||
NdbResultSet*
|
||||
rs = pOp->readTuples(NdbScanOperation::LM_Read, 0, parallelism);
|
||||
|
||||
if( rs == 0 ) {
|
||||
if( pOp->readTuples(NdbScanOperation::LM_Read, 0, parallelism) ) {
|
||||
ERR(pTrans->getNdbError());
|
||||
pNdb->closeTransaction(pTrans);
|
||||
return NDBT_FAILED;
|
||||
|
@ -1376,7 +1364,7 @@ UtilTransactions::verifyOrderedIndex(Ndb* pNdb,
|
|||
|
||||
int eof;
|
||||
int rows = 0;
|
||||
while(check == 0 && (eof = rs->nextResult()) == 0){
|
||||
while(check == 0 && (eof = pOp->nextResult()) == 0){
|
||||
rows++;
|
||||
|
||||
bool null_found= false;
|
||||
|
@ -1401,10 +1389,11 @@ UtilTransactions::verifyOrderedIndex(Ndb* pNdb,
|
|||
if(!iop && (iop= pTrans->getNdbIndexScanOperation(indexName,
|
||||
tab.getName())))
|
||||
{
|
||||
cursor= iop->readTuples(NdbScanOperation::LM_CommittedRead,
|
||||
parallelism);
|
||||
if(iop->readTuples(NdbScanOperation::LM_CommittedRead,
|
||||
parallelism))
|
||||
goto error;
|
||||
iop->interpret_exit_ok();
|
||||
if(!cursor || get_values(iop, indexRow))
|
||||
if(get_values(iop, indexRow))
|
||||
goto error;
|
||||
}
|
||||
else if(!iop || iop->reset_bounds())
|
||||
|
@ -1431,7 +1420,7 @@ UtilTransactions::verifyOrderedIndex(Ndb* pNdb,
|
|||
if(!null_found)
|
||||
{
|
||||
|
||||
if((res= cursor->nextResult()) != 0){
|
||||
if((res= iop->nextResult()) != 0){
|
||||
g_err << "Failed to find row using index: " << res << endl;
|
||||
ERR(pTrans->getNdbError());
|
||||
pNdb->closeTransaction(pTrans);
|
||||
|
@ -1446,7 +1435,7 @@ UtilTransactions::verifyOrderedIndex(Ndb* pNdb,
|
|||
return NDBT_FAILED;
|
||||
}
|
||||
|
||||
if(cursor->nextResult() == 0){
|
||||
if(iop->nextResult() == 0){
|
||||
g_err << "Found extra row!!" << endl;
|
||||
g_err << " indexRow: \n" << indexRow.c_str().c_str() << endl;
|
||||
pNdb->closeTransaction(pTrans);
|
||||
|
|
|
@ -141,8 +141,7 @@ int clear_table(Ndb* pNdb, const NdbDictionary::Table* pTab, int parallelism)
|
|||
goto failed;
|
||||
}
|
||||
|
||||
NdbResultSet * rs = pOp->readTuplesExclusive(par);
|
||||
if( rs == 0 ) {
|
||||
if( pOp->readTuplesExclusive(par) ) {
|
||||
goto failed;
|
||||
}
|
||||
|
||||
|
@ -157,13 +156,13 @@ int clear_table(Ndb* pNdb, const NdbDictionary::Table* pTab, int parallelism)
|
|||
goto failed;
|
||||
}
|
||||
|
||||
while((check = rs->nextResult(true)) == 0){
|
||||
while((check = pOp->nextResult(true)) == 0){
|
||||
do {
|
||||
if (rs->deleteTuple() != 0){
|
||||
if (pOp->deleteCurrentTuple() != 0){
|
||||
goto failed;
|
||||
}
|
||||
deletedRows++;
|
||||
} while((check = rs->nextResult(false)) == 0);
|
||||
} while((check = pOp->nextResult(false)) == 0);
|
||||
|
||||
if(check != -1){
|
||||
check = pTrans->execute(Commit);
|
||||
|
|
|
@ -215,7 +215,7 @@ int scanReadRecords(Ndb* pNdb,
|
|||
return -1;
|
||||
}
|
||||
|
||||
NdbResultSet * rs;
|
||||
int rs;
|
||||
switch(_lock + (3 * order)){
|
||||
case 1:
|
||||
rs = pOp->readTuples(NdbScanOperation::LM_Read, 0, parallel);
|
||||
|
@ -238,7 +238,7 @@ int scanReadRecords(Ndb* pNdb,
|
|||
rs = pOp->readTuples(NdbScanOperation::LM_CommittedRead, 0, parallel);
|
||||
break;
|
||||
}
|
||||
if( rs == 0 ){
|
||||
if( rs != 0 ){
|
||||
ERR(pTrans->getNdbError());
|
||||
pNdb->closeTransaction(pTrans);
|
||||
return -1;
|
||||
|
@ -324,7 +324,7 @@ int scanReadRecords(Ndb* pNdb,
|
|||
|
||||
int eof;
|
||||
int rows = 0;
|
||||
eof = rs->nextResult();
|
||||
eof = pOp->nextResult();
|
||||
|
||||
while(eof == 0){
|
||||
rows++;
|
||||
|
@ -335,7 +335,7 @@ int scanReadRecords(Ndb* pNdb,
|
|||
ndbout << (*row) << endl;
|
||||
}
|
||||
|
||||
eof = rs->nextResult();
|
||||
eof = pOp->nextResult();
|
||||
}
|
||||
if (eof == -1) {
|
||||
const NdbError err = pTrans->getNdbError();
|
||||
|
|
|
@ -164,8 +164,7 @@ select_count(Ndb* pNdb, const NdbDictionary::Table* pTab,
|
|||
return NDBT_FAILED;
|
||||
}
|
||||
|
||||
NdbResultSet * rs = pOp->readTuples(NdbScanOperation::LM_Dirty);
|
||||
if( rs == 0 ) {
|
||||
if( pOp->readTuples(NdbScanOperation::LM_Dirty) ) {
|
||||
ERR(pTrans->getNdbError());
|
||||
pNdb->closeTransaction(pTrans);
|
||||
return NDBT_FAILED;
|
||||
|
@ -191,7 +190,7 @@ select_count(Ndb* pNdb, const NdbDictionary::Table* pTab,
|
|||
|
||||
Uint64 row_count = 0;
|
||||
int eof;
|
||||
while((eof = rs->nextResult(true)) == 0){
|
||||
while((eof = pOp->nextResult(true)) == 0){
|
||||
row_count += tmp;
|
||||
}
|
||||
|
||||
|
|
|
@ -278,7 +278,7 @@ void ha_ndbcluster::no_uncommitted_rows_init(THD *thd)
|
|||
Thd_ndb *thd_ndb= (Thd_ndb *)thd->transaction.thd_ndb;
|
||||
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->records= ~(ha_rows)0;
|
||||
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);
|
||||
}
|
||||
|
||||
/*
|
||||
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)
|
||||
{
|
||||
inline int ha_ndbcluster::fetch_next()
|
||||
{
|
||||
DBUG_ENTER("fetch_next");
|
||||
int check;
|
||||
NdbConnection *trans= m_active_trans;
|
||||
NdbResultSet *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
|
||||
*/
|
||||
NdbScanOperation *cursor= m_active_cursor;
|
||||
|
||||
bool contact_ndb= m_lock.type < TL_WRITE_ALLOW_WRITE;
|
||||
do {
|
||||
DBUG_PRINT("info", ("Call nextResult, contact_ndb: %d", contact_ndb));
|
||||
|
@ -1259,21 +1241,16 @@ inline int ha_ndbcluster::next_result(byte *buf)
|
|||
m_ops_pending= 0;
|
||||
m_blobs_pending= FALSE;
|
||||
}
|
||||
check= cursor->nextResult(contact_ndb, m_force_send);
|
||||
if (check == 0)
|
||||
|
||||
if ((check= cursor->nextResult(contact_ndb, m_force_send)) == 0)
|
||||
{
|
||||
// One more record found
|
||||
DBUG_PRINT("info", ("One more record found"));
|
||||
|
||||
unpack_record(buf);
|
||||
table->status= 0;
|
||||
DBUG_RETURN(0);
|
||||
}
|
||||
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
|
||||
|
@ -1282,33 +1259,63 @@ inline int ha_ndbcluster::next_result(byte *buf)
|
|||
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,trans) != 0)
|
||||
DBUG_RETURN(ndb_err(trans));
|
||||
return -1;
|
||||
}
|
||||
else
|
||||
{
|
||||
if (execute_commit(this,trans) != 0)
|
||||
DBUG_RETURN(ndb_err(trans));
|
||||
return -1;
|
||||
int res= trans->restart();
|
||||
DBUG_ASSERT(res == 0);
|
||||
}
|
||||
m_ops_pending= 0;
|
||||
}
|
||||
|
||||
contact_ndb= (check == 2);
|
||||
}
|
||||
} while (check == 2);
|
||||
|
||||
table->status= STATUS_NOT_FOUND;
|
||||
if (check == -1)
|
||||
DBUG_RETURN(ndb_err(trans));
|
||||
|
||||
DBUG_RETURN(1);
|
||||
}
|
||||
|
||||
// No more records
|
||||
DBUG_PRINT("info", ("No more records"));
|
||||
DBUG_RETURN(HA_ERR_END_OF_FILE);
|
||||
/*
|
||||
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
|
||||
table->status= STATUS_NOT_FOUND;
|
||||
|
||||
DBUG_PRINT("info", ("No more records"));
|
||||
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++)
|
||||
{
|
||||
struct part_st &p = part[j];
|
||||
struct part_st &p= part[j];
|
||||
p.key= NULL;
|
||||
p.bound_type= -1;
|
||||
if (tot_len < key_tot_len[j])
|
||||
|
@ -1445,7 +1452,7 @@ int ha_ndbcluster::set_bounds(NdbIndexScanOperation *op,
|
|||
|
||||
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
|
||||
if (p.key != NULL)
|
||||
{
|
||||
|
@ -1524,7 +1531,6 @@ int ha_ndbcluster::ordered_index_scan(const key_range *start_key,
|
|||
int res;
|
||||
bool restart;
|
||||
NdbConnection *trans= m_active_trans;
|
||||
NdbResultSet *cursor;
|
||||
NdbIndexScanOperation *op;
|
||||
|
||||
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 *)
|
||||
m_index[active_index].index,
|
||||
(const NDBTAB *) m_table)) ||
|
||||
!(cursor= op->readTuples(lm, 0, parallelism, sorted)))
|
||||
op->readTuples(lm, 0, parallelism, sorted))
|
||||
ERR_RETURN(trans->getNdbError());
|
||||
m_active_cursor= cursor;
|
||||
m_active_cursor= op;
|
||||
} else {
|
||||
restart= true;
|
||||
op= (NdbIndexScanOperation*)m_active_cursor->getOperation();
|
||||
op= (NdbIndexScanOperation*)m_active_cursor;
|
||||
|
||||
DBUG_ASSERT(op->getSorted() == sorted);
|
||||
DBUG_ASSERT(op->getLockMode() ==
|
||||
|
@ -1555,14 +1561,14 @@ int ha_ndbcluster::ordered_index_scan(const key_range *start_key,
|
|||
if(op->reset_bounds(m_force_send))
|
||||
DBUG_RETURN(ndb_err(m_active_trans));
|
||||
}
|
||||
|
||||
|
||||
{
|
||||
const key_range *keys[2]= { start_key, end_key };
|
||||
res= set_bounds(op, keys);
|
||||
if (res)
|
||||
DBUG_RETURN(res);
|
||||
}
|
||||
|
||||
|
||||
if (!restart && (res= define_read_attrs(buf, op)))
|
||||
{
|
||||
DBUG_RETURN(res);
|
||||
|
@ -1594,7 +1600,6 @@ int ha_ndbcluster::filtered_scan(const byte *key, uint key_len,
|
|||
{
|
||||
int res;
|
||||
NdbConnection *trans= m_active_trans;
|
||||
NdbResultSet *cursor;
|
||||
NdbScanOperation *op;
|
||||
|
||||
DBUG_ENTER("filtered_scan");
|
||||
|
@ -1607,9 +1612,9 @@ int ha_ndbcluster::filtered_scan(const byte *key, uint key_len,
|
|||
NdbOperation::LockMode lm=
|
||||
(NdbOperation::LockMode)get_ndb_lock_type(m_lock.type);
|
||||
if (!(op= trans->getNdbScanOperation((const NDBTAB *) m_table)) ||
|
||||
!(cursor= op->readTuples(lm, 0, parallelism)))
|
||||
op->readTuples(lm, 0, parallelism))
|
||||
ERR_RETURN(trans->getNdbError());
|
||||
m_active_cursor= cursor;
|
||||
m_active_cursor= op;
|
||||
|
||||
{
|
||||
// Start scan filter
|
||||
|
@ -1673,7 +1678,6 @@ int ha_ndbcluster::full_table_scan(byte *buf)
|
|||
{
|
||||
uint i;
|
||||
int res;
|
||||
NdbResultSet *cursor;
|
||||
NdbScanOperation *op;
|
||||
NdbConnection *trans= m_active_trans;
|
||||
|
||||
|
@ -1683,9 +1687,9 @@ int ha_ndbcluster::full_table_scan(byte *buf)
|
|||
NdbOperation::LockMode lm=
|
||||
(NdbOperation::LockMode)get_ndb_lock_type(m_lock.type);
|
||||
if (!(op=trans->getNdbScanOperation((const NDBTAB *) m_table)) ||
|
||||
!(cursor= op->readTuples(lm, 0, parallelism)))
|
||||
op->readTuples(lm, 0, parallelism))
|
||||
ERR_RETURN(trans->getNdbError());
|
||||
m_active_cursor= cursor;
|
||||
m_active_cursor= op;
|
||||
|
||||
if((res= define_read_attrs(buf, op)))
|
||||
DBUG_RETURN(res);
|
||||
|
@ -1871,7 +1875,7 @@ int ha_ndbcluster::update_row(const byte *old_data, byte *new_data)
|
|||
{
|
||||
THD *thd= current_thd;
|
||||
NdbConnection *trans= m_active_trans;
|
||||
NdbResultSet* cursor= m_active_cursor;
|
||||
NdbScanOperation* cursor= m_active_cursor;
|
||||
NdbOperation *op;
|
||||
uint i;
|
||||
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
|
||||
*/
|
||||
DBUG_PRINT("info", ("Calling updateTuple on cursor"));
|
||||
if (!(op= cursor->updateTuple()))
|
||||
if (!(op= cursor->updateCurrentTuple()))
|
||||
ERR_RETURN(trans->getNdbError());
|
||||
m_ops_pending++;
|
||||
if (uses_blob_value(FALSE))
|
||||
|
@ -1989,7 +1993,7 @@ int ha_ndbcluster::delete_row(const byte *record)
|
|||
{
|
||||
THD *thd= current_thd;
|
||||
NdbConnection *trans= m_active_trans;
|
||||
NdbResultSet* cursor= m_active_cursor;
|
||||
NdbScanOperation* cursor= m_active_cursor;
|
||||
NdbOperation *op;
|
||||
DBUG_ENTER("delete_row");
|
||||
|
||||
|
@ -2005,7 +2009,7 @@ int ha_ndbcluster::delete_row(const byte *record)
|
|||
the active record in cursor
|
||||
*/
|
||||
DBUG_PRINT("info", ("Calling deleteTuple on cursor"));
|
||||
if (cursor->deleteTuple() != 0)
|
||||
if (cursor->deleteCurrentTuple() != 0)
|
||||
ERR_RETURN(trans->getNdbError());
|
||||
m_ops_pending++;
|
||||
|
||||
|
@ -2073,7 +2077,7 @@ void ha_ndbcluster::unpack_record(byte* buf)
|
|||
NdbValue *value= m_value;
|
||||
DBUG_ENTER("unpack_record");
|
||||
|
||||
end = table->field + table->fields;
|
||||
end= table->field + table->fields;
|
||||
|
||||
// Set null flag(s)
|
||||
bzero(buf, table->null_bytes);
|
||||
|
@ -2303,7 +2307,7 @@ int
|
|||
check_null_in_key(const KEY* key_info, const byte *key, uint key_len)
|
||||
{
|
||||
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;
|
||||
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));
|
||||
|
||||
int error;
|
||||
ndb_index_type type = get_index_type(active_index);
|
||||
const KEY* key_info = table->key_info+active_index;
|
||||
ndb_index_type type= get_index_type(active_index);
|
||||
const KEY* key_info= table->key_info+active_index;
|
||||
switch (type){
|
||||
case PRIMARY_KEY_ORDERED_INDEX:
|
||||
case PRIMARY_KEY_INDEX:
|
||||
|
@ -2367,9 +2371,9 @@ int ha_ndbcluster::index_read(byte *buf,
|
|||
}
|
||||
|
||||
key_range start_key;
|
||||
start_key.key = key;
|
||||
start_key.length = key_len;
|
||||
start_key.flag = find_flag;
|
||||
start_key.key= key;
|
||||
start_key.length= key_len;
|
||||
start_key.flag= find_flag;
|
||||
error= ordered_index_scan(&start_key, 0, TRUE, buf);
|
||||
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);
|
||||
int res;
|
||||
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);
|
||||
if(res == 1){
|
||||
unpack_record(buf);
|
||||
|
@ -2508,7 +2512,7 @@ int ha_ndbcluster::read_range_next()
|
|||
|
||||
int ha_ndbcluster::rnd_init(bool scan)
|
||||
{
|
||||
NdbResultSet *cursor= m_active_cursor;
|
||||
NdbScanOperation *cursor= m_active_cursor;
|
||||
DBUG_ENTER("rnd_init");
|
||||
DBUG_PRINT("enter", ("scan: %d", scan));
|
||||
// Check if scan is to be restarted
|
||||
|
@ -2525,7 +2529,7 @@ int ha_ndbcluster::rnd_init(bool scan)
|
|||
|
||||
int ha_ndbcluster::close_scan()
|
||||
{
|
||||
NdbResultSet *cursor= m_active_cursor;
|
||||
NdbScanOperation *cursor= m_active_cursor;
|
||||
NdbConnection *trans= m_active_trans;
|
||||
DBUG_ENTER("close_scan");
|
||||
|
||||
|
@ -3528,12 +3532,12 @@ int ha_ndbcluster::create(const char *name,
|
|||
case MYSQL_TYPE_MEDIUM_BLOB:
|
||||
case MYSQL_TYPE_LONG_BLOB:
|
||||
{
|
||||
NdbDictionary::Column * col = tab.getColumn(i);
|
||||
int size = pk_length + (col->getPartSize()+3)/4 + 7;
|
||||
NdbDictionary::Column * col= tab.getColumn(i);
|
||||
int size= pk_length + (col->getPartSize()+3)/4 + 7;
|
||||
if(size > 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);
|
||||
}
|
||||
/**
|
||||
|
@ -4696,8 +4700,7 @@ ndb_get_table_statistics(Ndb* ndb, const char * table,
|
|||
if (pOp == NULL)
|
||||
break;
|
||||
|
||||
NdbResultSet* rs= pOp->readTuples(NdbOperation::LM_CommittedRead);
|
||||
if (rs == 0)
|
||||
if (pOp->readTuples(NdbOperation::LM_CommittedRead))
|
||||
break;
|
||||
|
||||
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_commits= 0;
|
||||
while((check= rs->nextResult(TRUE, TRUE)) == 0)
|
||||
while((check= pOp->nextResult(TRUE, TRUE)) == 0)
|
||||
{
|
||||
sum_rows+= rows;
|
||||
sum_commits+= commits;
|
||||
|
@ -4723,7 +4726,7 @@ ndb_get_table_statistics(Ndb* ndb, const char * table,
|
|||
if (check == -1)
|
||||
break;
|
||||
|
||||
rs->close(TRUE);
|
||||
pOp->close(TRUE);
|
||||
|
||||
ndb->closeTransaction(pTrans);
|
||||
if(row_count)
|
||||
|
@ -4777,8 +4780,8 @@ ha_ndbcluster::read_multi_range_first(key_multi_range **found_range_p,
|
|||
int res;
|
||||
uint i;
|
||||
KEY* key_info= table->key_info + active_index;
|
||||
NDB_INDEX_TYPE index_type = get_index_type(active_index);
|
||||
ulong reclength = table->reclength;
|
||||
NDB_INDEX_TYPE index_type= get_index_type(active_index);
|
||||
ulong reclength= table->reclength;
|
||||
NdbOperation* op;
|
||||
|
||||
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= m_active_trans->getNdbIndexScanOperation(idx, tab)) &&
|
||||
(m_active_cursor= scanOp->readTuples(lm, 0,
|
||||
parallelism, sorted, true))&&
|
||||
!scanOp->readTuples(lm, 0, parallelism, sorted, true) &&
|
||||
!define_read_attrs(curr, scanOp))
|
||||
curr += reclength;
|
||||
else
|
||||
ERR_RETURN(scanOp ?
|
||||
scanOp->getNdbError() :
|
||||
ERR_RETURN(scanOp ? scanOp->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 };
|
||||
if ((res= set_bounds(scanOp, keys, i)))
|
||||
|
@ -4931,64 +4934,60 @@ ha_ndbcluster::read_multi_range_next(key_multi_range ** multi_range_found_p)
|
|||
{
|
||||
DBUG_RETURN(handler::read_multi_range_next(multi_range_found_p));
|
||||
}
|
||||
|
||||
|
||||
int res;
|
||||
int range_no;
|
||||
ulong reclength= table->reclength;
|
||||
const NdbOperation* op= m_current_multi_operation;
|
||||
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;
|
||||
|
||||
op= m_active_trans->getNextCompletedOperation(op);
|
||||
m_multi_range_result_ptr += reclength;
|
||||
continue;
|
||||
}
|
||||
else if(m_active_cursor)
|
||||
else if (m_multi_cursor && !multi_range_sorted)
|
||||
{
|
||||
int check;
|
||||
bool contact_ndb= m_lock.type < TL_WRITE_ALLOW_WRITE;
|
||||
do {
|
||||
check= m_active_cursor->nextResult(contact_ndb, m_force_send);
|
||||
if (check == 0)
|
||||
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
|
||||
{
|
||||
if (execute_commit(this, m_active_trans) != 0)
|
||||
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);
|
||||
}
|
||||
} while (check == 2);
|
||||
if ((res= fetch_next() == 0))
|
||||
{
|
||||
range_no= m_multi_cursor->get_range_no();
|
||||
goto found;
|
||||
}
|
||||
else
|
||||
{
|
||||
goto close_scan;
|
||||
}
|
||||
}
|
||||
else if (multi_range_sorted)
|
||||
{
|
||||
DBUG_ASSERT(m_multi_cursor);
|
||||
if (m_active_cursor && (res= fetch_next()))
|
||||
goto close_scan;
|
||||
|
||||
m_multi_range_result_ptr += reclength;
|
||||
|
||||
m_active_cursor->close();
|
||||
m_active_cursor= 0;
|
||||
range_no= m_multi_cursor->get_range_no();
|
||||
if (range_no == multi_range_curr)
|
||||
{
|
||||
// return current row
|
||||
goto found;
|
||||
}
|
||||
else if (range_no > multi_range_curr)
|
||||
{
|
||||
// wait with current row
|
||||
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 */
|
||||
{
|
||||
|
@ -4996,8 +4995,24 @@ ha_ndbcluster::read_multi_range_next(key_multi_range ** multi_range_found_p)
|
|||
* Corresponds to range 5 in example in read_multi_range_first
|
||||
*/
|
||||
(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));
|
||||
}
|
||||
}
|
||||
|
||||
if (multi_range_curr == multi_range_count)
|
||||
DBUG_RETURN(HA_ERR_END_OF_FILE);
|
||||
|
@ -5016,15 +5031,14 @@ found:
|
|||
/**
|
||||
* Found a record belonging to a scan
|
||||
*/
|
||||
Uint32 range_no = ((NdbIndexScanOperation*)m_active_cursor->getOperation())
|
||||
->get_range_no();
|
||||
m_active_cursor= m_multi_cursor;
|
||||
* multi_range_found_p= multi_ranges + range_no;
|
||||
memcpy(table->record[0], m_multi_range_result_ptr, reclength);
|
||||
setup_recattr(m_active_cursor->getOperation()->getFirstRecAttr());
|
||||
memcpy(table->record[0], m_multi_range_cursor_result_ptr, reclength);
|
||||
setup_recattr(m_active_cursor->getFirstRecAttr());
|
||||
unpack_record(table->record[0]);
|
||||
table->status= 0;
|
||||
DBUG_RETURN(0);
|
||||
|
||||
|
||||
found_next:
|
||||
/**
|
||||
* Found a record belonging to a pk/index op,
|
||||
|
@ -5050,15 +5064,15 @@ ha_ndbcluster::setup_recattr(const NdbRecAttr* curr)
|
|||
Field **field, **end;
|
||||
NdbValue *value= m_value;
|
||||
|
||||
end = table->field + table->fields;
|
||||
end= table->field + table->fields;
|
||||
|
||||
for (field= table->field; field < end; field++, value++)
|
||||
{
|
||||
if ((* value).ptr)
|
||||
{
|
||||
DBUG_ASSERT(curr != 0);
|
||||
(* value).rec = curr;
|
||||
curr = curr->next();
|
||||
(* value).rec= curr;
|
||||
curr= curr->next();
|
||||
}
|
||||
}
|
||||
|
||||
|
|
Loading…
Reference in a new issue