| /********************************************************************** |
| // @@@ START COPYRIGHT @@@ |
| // |
| // Licensed to the Apache Software Foundation (ASF) under one |
| // or more contributor license agreements. See the NOTICE file |
| // distributed with this work for additional information |
| // regarding copyright ownership. The ASF licenses this file |
| // to you under the Apache License, Version 2.0 (the |
| // "License"); you may not use this file except in compliance |
| // with the License. You may obtain a copy of the License at |
| // |
| // http://www.apache.org/licenses/LICENSE-2.0 |
| // |
| // Unless required by applicable law or agreed to in writing, |
| // software distributed under the License is distributed on an |
| // "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY |
| // KIND, either express or implied. See the License for the |
| // specific language governing permissions and limitations |
| // under the License. |
| // |
| // @@@ END COPYRIGHT @@@ |
| **********************************************************************/ |
| /* -*-C++-*- |
| ***************************************************************************** |
| * |
| * File: ex_stored_proc.cpp |
| * Description: |
| * |
| * |
| * Created: 3/15/1997 |
| * Language: C++ |
| * |
| * |
| * |
| * |
| ***************************************************************************** |
| */ |
| |
| #include "cli_stdh.h" |
| #include "ex_stdh.h" |
| #include "ComTdb.h" |
| #include "ex_tcb.h" |
| #include "ex_stored_proc.h" |
| #include "ex_exe_stmt_globals.h" |
| #include "ex_io_control.h" |
| #include "ex_expr.h" |
| #include "exp_tuple_desc.h" |
| #include "exp_attrs.h" |
| #include "exp_clause_derived.h" |
| #include "ExSqlComp.h" |
| |
| #include "ErrorMessage.h" |
| #include "CmpStatement.h" |
| #include "CmpStoredProc.h" |
| |
| ex_tcb * ExStoredProcTdb::build(ex_globals * glob) |
| { |
| ExStoredProcTcb * spTcb = new(glob->getSpace()) |
| ExStoredProcTcb(*this, glob); |
| |
| // add the stored_proc_tcb to the schedule |
| spTcb->registerSubtasks(); |
| |
| return (spTcb); |
| } |
| |
| |
| ExStoredProcTcb::ExStoredProcTcb(const ExStoredProcTdb & sp_tdb, |
| ex_globals * glob |
| ) |
| : ex_tcb( sp_tdb, 1, glob) |
| { |
| Space * space = (glob ? glob->getSpace() : 0); |
| |
| // Allocate the buffer pool |
| pool_ = new(space) sql_buffer_pool(sp_tdb.numBuffers_, |
| sp_tdb.bufferSize_, |
| space); |
| |
| // Allocate the queue to communicate with parent |
| qparent_.down = new(space) ex_queue(ex_queue::DOWN_QUEUE, |
| sp_tdb.queueSizeDown_, |
| sp_tdb.criDescDown_, |
| space); |
| |
| // Allocate the private state in each entry of the down queue |
| ExStoredProcPrivateState *p = new(space) ExStoredProcPrivateState(this); |
| qparent_.down->allocatePstate(p, this); |
| delete p; |
| |
| |
| qparent_.up = new(space) ex_queue(ex_queue::UP_QUEUE, |
| sp_tdb.queueSizeUp_, |
| sp_tdb.criDescUp_, |
| space); |
| |
| inputBuffer_ = pool_->get_free_buffer(0); |
| returnedBuffer_ = NULL; |
| |
| step_ = BUILD_INPUT_BUFFER_; |
| |
| workAtp_ = allocateAtp(sp_tdb.workCriDesc_, space); |
| |
| // create the arkcmp that will be used to process this |
| // SP request. |
| ExMasterStmtGlobals *master_glob = getGlobals()->castToExExeStmtGlobals() |
| ->castToExMasterStmtGlobals(); |
| ex_assert((master_glob != NULL), "Invalid reference in ExStoredProcTcb."); |
| |
| // Get the arkcmp_ that contains the query cache |
| if (spTdb().getUseExistingArkcmp()) { |
| arkcmp_ = master_glob->getCliGlobals()->getArkcmp(); |
| } |
| else { |
| arkcmp_ = |
| new(glob->getDefaultHeap()) ExSqlComp(0, |
| glob->getDefaultHeap(), |
| master_glob->getCliGlobals(), |
| this, |
| COM_VERS_COMPILER_VERSION, |
| NULL, |
| master_glob->getStatement()-> |
| getContext()->getEnvironment()); |
| } |
| |
| ioSubtask_ = NULL; |
| numInputRequests_ = 0; |
| |
| if (sp_tdb.inputExpr_) |
| (void) sp_tdb.inputExpr_->fixup(0, getExpressionMode(), this, space, glob->getDefaultHeap(), FALSE, glob); |
| |
| if (sp_tdb.outputExpr_) |
| (void) sp_tdb.outputExpr_->fixup(0, getExpressionMode(), this, space, glob->getDefaultHeap(), FALSE, glob); |
| |
| if (sp_tdb.predExpr_) |
| (void) sp_tdb.predExpr_->fixup(0, getExpressionMode(), this, space, glob->getDefaultHeap(), FALSE, glob); |
| |
| } |
| |
| ExStoredProcTcb::~ExStoredProcTcb() |
| { |
| delete qparent_.up; |
| delete qparent_.down; |
| |
| if (!spTdb().getUseExistingArkcmp()) { |
| delete arkcmp_; |
| } |
| |
| freeResources(); |
| } |
| |
| void ExStoredProcTcb::freeResources() |
| { |
| delete pool_; |
| pool_ = 0; |
| } |
| |
| void ExStoredProcTcb::registerSubtasks() |
| { |
| ex_tcb::registerSubtasks(); |
| |
| // register a non-queue event for the IPC with the send top node |
| ioSubtask_ = |
| getGlobals()->getScheduler()->registerNonQueueSubtask(sWork,this); |
| } |
| |
| short ExStoredProcTcb::work() |
| { |
| // if no parent request, return |
| if (qparent_.down->isEmpty()) |
| return WORK_OK; |
| |
| ExMasterStmtGlobals *master_glob = getGlobals()->castToExExeStmtGlobals() |
| ->castToExMasterStmtGlobals(); |
| |
| // If there is no arkcmp, return |
| if (!arkcmp_ && spTdb().getUseExistingArkcmp()) { |
| |
| // make sure we have space in the up queue |
| if (qparent_.up->isFull()) { |
| return WORK_OK; |
| } |
| |
| queue_index pindex = qparent_.down->getHeadIndex(); |
| ex_queue_entry *pentry_down = qparent_.down->getQueueEntry(pindex); |
| ex_queue_entry *up_entry = qparent_.up->getTailEntry(); |
| up_entry->upState.status = ex_queue::Q_NO_DATA; |
| up_entry->upState.parentIndex = pentry_down->downState.parentIndex; |
| ExStoredProcPrivateState & pstate |
| = *((ExStoredProcPrivateState *) pentry_down->pstate); |
| up_entry->upState.setMatchNo(pstate.matchCount_); |
| |
| // insert into parent |
| qparent_.up->insert(); |
| |
| pstate.matchCount_ = 0; |
| qparent_.down->removeHead(); |
| numInputRequests_--; |
| |
| return WORK_OK; |
| } |
| |
| while (1) |
| { |
| switch (step_) |
| { |
| case BUILD_INPUT_BUFFER_: |
| { |
| queue_index pindex = qparent_.down->getHeadIndex(); |
| NABoolean done = FALSE; |
| |
| while (NOT done) |
| { |
| if (pindex == qparent_.down->getTailIndex()) |
| done = TRUE; |
| else |
| { |
| ex_queue_entry *pentry_down = |
| qparent_.down->getQueueEntry(pindex); |
| |
| tupp_descriptor * dataDesc; |
| |
| if (inputBuffer_-> |
| moveInSendOrReplyData(TRUE, // send(input) data |
| FALSE, |
| TRUE, // move data |
| &(pentry_down->downState), |
| sizeof(ControlInfo), |
| 0, |
| spTdb().inputRowlen_, |
| &dataDesc, |
| 0, NULL) == SqlBuffer::BUFFER_FULL) |
| { |
| // no more space in input Buffer. |
| done = TRUE; |
| } |
| else |
| { |
| numInputRequests_++; |
| |
| // move input data |
| if (spTdb().inputExpr_) |
| { |
| workAtp_->getTupp(spTdb().workAtpIndex_) |
| = dataDesc; |
| spTdb().inputExpr_->eval(pentry_down->getAtp(), workAtp_); |
| workAtp_->getTupp(spTdb().workAtpIndex_).release(); |
| } |
| |
| // move to the next down entry |
| pindex++; |
| |
| } // move input data |
| } // more entries in down queue |
| } // while not done |
| //CmpCommon::getDefault(PROCESS_ISP_LOCALLY) == DF_ON |
| if (spTdb().isExecuteInLocalProcess()) |
| step_ = PROCESS_REQUEST_; |
| else |
| // Ship the request to ARKCMP. |
| step_ = SEND_INPUT_BUFFER_; |
| } |
| break; |
| case PROCESS_REQUEST_: |
| { |
| //construct request |
| ULng32 sz = inputBuffer_->get_buffer_size(); |
| inputBuffer_->drivePack(); |
| inputBuffer_->setBufferStatus(SqlBuffer::IN_USE); |
| // The request is dynamically allocated here instead of using stack variable, |
| // because it is owned by a CmpStatementISP inside compileDirect, |
| // and will be deleted in destructor of CmpStatementISP. |
| // refer to ExCmpMessage::actOnReceive() |
| CmpMessageISPRequest* ispRequest = new (getHeap()) CmpMessageISPRequest((char *)spTdb().spName_, |
| spTdb().extractInputExpr_, |
| spTdb().extractInputExpr_->getLength(), |
| spTdb().moveOutputExpr_, |
| spTdb().moveOutputExpr_->getLength(), |
| 0, 0, |
| (void *)inputBuffer_, sz, |
| spTdb().outputRowlen_, sz, |
| getHeap(), |
| master_glob->getStatement()->getUniqueStmtId(), |
| master_glob->getStatement()->getUniqueStmtIdLen()); |
| //save requestId in case there may be getNext |
| requestId_ = ispRequest->id(); |
| //allocate returnedBuffer |
| returnedBuffer_ = pool_->get_free_buffer(0); |
| if (returnedBuffer_ == NULL) |
| return WORK_POOL_BLOCKED; |
| ULng32 returnedBuflen = returnedBuffer_->get_buffer_size(); |
| UInt32 dummyDatalen = 0; //not used here |
| Int32 dummyCharSet = 0; //not used here |
| Int32 cpStatus; |
| ComDiagsArea *cpDiagsArea = ComDiagsArea::allocate(getHeap()); |
| //pass request and returnedBuffer_ to compileDirect |
| cpStatus = CmpCommon::context()->compileDirect( |
| (char*)ispRequest, dummyDatalen, |
| master_glob->getStatement()->getContext()->exHeap(), |
| dummyCharSet, |
| CmpMessageObj::INTERNALSP_REQUEST, |
| (char* &)returnedBuffer_, returnedBuflen, |
| master_glob->getStatement()->getContext()->getSqlParserFlags(), |
| |
| NULL, 0, cpDiagsArea); |
| if(cpStatus != 0)//FAILURE |
| { |
| //replied length exceeds the return buffer length, this will abort an dump the core |
| if(returnedBuflen > returnedBuffer_->get_buffer_size()) |
| ex_assert(FALSE, "Replied length exceeds receiving buffer length."); |
| |
| if(cpDiagsArea->mainSQLCODE() != 0) { |
| ComDiagsArea *mainDiags = getGlobals()->castToExExeStmtGlobals()->getDiagsArea(); |
| if (mainDiags == NULL) { |
| mainDiags = ComDiagsArea::allocate(getGlobals()->getDefaultHeap()); |
| getGlobals()->castToExExeStmtGlobals()->setGlobDiagsArea(mainDiags); |
| mainDiags->decrRefCount(); |
| } |
| mainDiags->mergeAfter(*cpDiagsArea); |
| } |
| cpDiagsArea->decrRefCount(); |
| return WORK_BAD_ERROR; |
| } |
| cpDiagsArea->decrRefCount(); |
| returnedBuffer_->driveUnpack(); |
| //on success, returned rows are copied to returnedBuffer_ in compileDirect |
| step_ = RETURN_ROWS_; |
| } |
| break; |
| case PROCESS_GETNEXT_: |
| { |
| //getNext request can be allocated in stack, because it's not own by CmpStatementISP |
| //refer to ExCmpMessage::actOnReceive() |
| CmpMessageISPGetNext ispRequestGetNext(returnedBuffer_->get_buffer_size(), requestId_, 0, 0); |
| returnedBuffer_ = pool_->get_free_buffer(0); |
| if (returnedBuffer_ == NULL) |
| return WORK_POOL_BLOCKED; |
| ULng32 returnedBuflen = returnedBuffer_->get_buffer_size(); |
| UInt32 dummyDatalen = 0; //not used here |
| Int32 dummyCharSet = 0; //not used here |
| Int32 cpStatus; |
| ComDiagsArea *cpDiagsArea = ComDiagsArea::allocate(getHeap()); |
| cpStatus = CmpCommon::context()->compileDirect( |
| (char*)&ispRequestGetNext, dummyDatalen, |
| master_glob->getStatement()->getContext()->exHeap(), |
| dummyCharSet, |
| CmpMessageObj::INTERNALSP_GETNEXT, |
| (char* &)returnedBuffer_, returnedBuflen, |
| master_glob->getStatement()->getContext()->getSqlParserFlags(), |
| NULL, 0, cpDiagsArea); |
| |
| if(cpStatus != 0)//FAILURE |
| { |
| //replied length exceeds the return buffer length, this will abort an dump the core |
| if(returnedBuflen > returnedBuffer_->get_buffer_size()) |
| ex_assert(FALSE, "Replied length exceeds receiving buffer length."); |
| |
| if(cpDiagsArea->mainSQLCODE() != 0) { |
| ComDiagsArea *mainDiags = getGlobals()->castToExExeStmtGlobals()->getDiagsArea(); |
| if (mainDiags == NULL) { |
| mainDiags = ComDiagsArea::allocate(getGlobals()->getDefaultHeap()); |
| getGlobals()->castToExExeStmtGlobals()->setGlobDiagsArea(mainDiags); |
| mainDiags->decrRefCount(); |
| } |
| mainDiags->mergeAfter(*cpDiagsArea); |
| } |
| cpDiagsArea->decrRefCount(); |
| return WORK_BAD_ERROR; |
| } |
| |
| cpDiagsArea->decrRefCount(); |
| returnedBuffer_->driveUnpack(); |
| //on success, returned rows are copied to returnedBuffer_ in compileDirect |
| step_ = RETURN_ROWS_; |
| } |
| break; |
| |
| case SEND_INPUT_BUFFER_: |
| { |
| ULng32 sz = inputBuffer_->get_buffer_size(); |
| inputBuffer_->drivePack(); |
| inputBuffer_->setBufferStatus(SqlBuffer::IN_USE); |
| // Send request to arkcomp and wait for reply. |
| if (arkcmp_->sendRequest( |
| spTdb().spName_, |
| spTdb().extractInputExpr_, |
| spTdb().extractInputExpr_->getLength(), |
| spTdb().moveOutputExpr_, |
| spTdb().moveOutputExpr_->getLength(), |
| 0, 0, |
| (void *)inputBuffer_, |
| sz, |
| spTdb().outputRowlen_, |
| sz, |
| spTdb().getUseExistingArkcmp(), |
| /* If true, then use a waited call. |
| Use no-waited otherwise*/ |
| &requestId_, |
| master_glob->getStatement()->getUniqueStmtId(), |
| master_glob->getStatement()->getUniqueStmtIdLen()) != |
| ExSqlComp::SUCCESS) |
| { |
| |
| if (arkcmp_->breakReceived()) |
| { |
| // Break signal was received. Return as quickly as possible |
| // to the scheduler, after setting a flag in the CLI globals |
| // that is used by mxci to determine whether a message |
| // should be returned to indicate that RECOVER needs to |
| // be run. |
| master_glob->getCliGlobals()->setSPBreakReceived(TRUE); |
| return WORK_BAD_ERROR; |
| } |
| else |
| master_glob->getCliGlobals()->setSPBreakReceived(FALSE); |
| |
| if (arkcmp_->badConnection()) |
| { |
| // Error reporting in the GET_REPLY_BUFFER_ step |
| // will be used, since the error is in the IPC |
| // connection. |
| } |
| else |
| { |
| // Get error from the arkcmp_. |
| ex_assert( |
| arkcmp_->getDiags(), |
| "ExSqlComp::getReply returns error, but no diagnostics"); |
| |
| ComDiagsArea *mainDiags = getGlobals()-> |
| castToExExeStmtGlobals()->getDiagsArea(); |
| |
| if (mainDiags == NULL) |
| { |
| mainDiags = ComDiagsArea::allocate( |
| getGlobals()->getDefaultHeap()); |
| getGlobals()->castToExExeStmtGlobals()-> |
| setGlobDiagsArea(mainDiags); |
| mainDiags->decrRefCount(); |
| } |
| |
| mainDiags->mergeAfter(*arkcmp_->getDiags()); |
| |
| return WORK_BAD_ERROR; |
| } |
| |
| } // end if !SUCCESS |
| |
| step_ = GET_REPLY_BUFFER_; |
| } |
| break; |
| |
| case GET_REPLY_BUFFER_: |
| { |
| if (arkcmp_->breakReceived()) |
| { |
| // Break signal was received. Return as quickly as possible |
| // to the scheduler, after setting a flag in the CLI globals |
| // that is used by mxci to determine whether a message |
| // should be returned to indicate that RECOVER needs to |
| // be run. |
| master_glob->getCliGlobals()->setSPBreakReceived(TRUE); |
| return WORK_BAD_ERROR; |
| } |
| else |
| master_glob->getCliGlobals()->setSPBreakReceived(FALSE); |
| |
| if (arkcmp_->badConnection()) |
| { |
| ComDiagsArea* da = getGlobals()-> |
| castToExExeStmtGlobals()->getDiagsArea(); |
| NABoolean alreadyHadGlobalDA = da ? TRUE : FALSE; |
| |
| arkcmp_->getServer()->getControlConnection()-> |
| populateDiagsArea( da, getHeap()); |
| |
| if (alreadyHadGlobalDA) |
| { |
| // Preceding populateDiagsArea call has |
| // added a ComCondition to existing DA, so |
| // there's nothing more to do. |
| } |
| else |
| { |
| // Preceding populateDiagsArea call has |
| // created a DiagsArea, added a ComCondition, |
| // and side-effected da so make this da |
| // the global DA. |
| getGlobals()->castToExExeStmtGlobals()-> |
| setGlobDiagsArea(da); |
| da->decrRefCount(); |
| } |
| return WORK_BAD_ERROR; |
| } |
| |
| if (arkcmp_->status() != ExSqlComp::FINISHED) |
| { |
| if (spTdb().getUseExistingArkcmp()) { |
| return WORK_CALL_AGAIN; |
| } |
| return WORK_OK; // come back later. |
| } |
| |
| // find an empty buffer where data will be returned. |
| returnedBuffer_ = pool_->get_free_buffer(0); |
| if (returnedBuffer_ == NULL) |
| return WORK_POOL_BLOCKED; |
| |
| ULng32 returnedBuflen_ = returnedBuffer_->get_buffer_size(); |
| char * rb = (char *)returnedBuffer_; |
| ExSqlComp::ReturnStatus status = |
| arkcmp_->getReply(rb, |
| returnedBuflen_, |
| returnedBuflen_, |
| requestId_); |
| if (status == ExSqlComp::MOREDATA) |
| { |
| // issue a getNext so arkcmp_ can prepare for |
| // the next buffer |
| // waited if sharing arkcmp (last argument) |
| status = arkcmp_->getNext(returnedBuflen_, requestId_, spTdb().getUseExistingArkcmp()); |
| } |
| |
| if (status == ExSqlComp::ERROR) |
| { |
| ex_assert( |
| arkcmp_->getDiags(), |
| "ExSqlComp::getReply returns error, but no diagnostics"); |
| |
| ComDiagsArea *mainDiags = getGlobals()-> |
| castToExExeStmtGlobals()->getDiagsArea(); |
| |
| if (mainDiags == NULL) |
| { |
| mainDiags = ComDiagsArea::allocate( |
| getGlobals()->getDefaultHeap()); |
| getGlobals()->castToExExeStmtGlobals()-> |
| setGlobDiagsArea(mainDiags); |
| mainDiags->decrRefCount(); |
| } |
| |
| mainDiags->mergeAfter(*arkcmp_->getDiags()); |
| |
| return WORK_BAD_ERROR; |
| } |
| |
| returnedBuffer_->driveUnpack(); |
| |
| step_ = RETURN_ROWS_; |
| } |
| break; |
| |
| case RETURN_ROWS_: |
| { |
| // make sure we have space in the up queue |
| if(qparent_.up->isFull()) |
| return WORK_OK; |
| |
| ex_queue_entry * up_entry = qparent_.up->getTailEntry(); |
| |
| tupp p; |
| ComDiagsArea* diags; |
| if (returnedBuffer_->moveOutSendOrReplyData(FALSE, // reply data |
| &(up_entry->upState), |
| p, NULL, &diags) |
| == TRUE) // no more rows in this buffer |
| { |
| // when we get a buffer from arkcmp, all tupp_descriptors in the |
| // buffer have a refcount of 0. To make sure that the buffer is not |
| // being reused it's status was IN_USE. Now we have put all |
| // the tupps in the buffer into the up queue, i.e., they have a |
| // refcount > 0. |
| // Mark the buffer as being FULL. |
| // It will be reused later after all tupp_descriptors have been |
| // consumed and reference count of all have become 0. |
| returnedBuffer_->bufferFull(); |
| |
| if (numInputRequests_ > 0) |
| { |
| // still more input requests. Get the |
| // next buffer from arkcmp. |
| if (spTdb().isExecuteInLocalProcess()) |
| step_ = PROCESS_GETNEXT_; |
| else |
| step_ = GET_REPLY_BUFFER_; |
| } |
| else |
| { |
| // done with all the inputs. |
| inputBuffer_->init(inputBuffer_->get_buffer_size()); |
| step_ = BUILD_INPUT_BUFFER_; |
| } |
| return WORK_CALL_AGAIN; |
| } |
| |
| ex_queue_entry *pentry_down = qparent_.down->getHeadEntry(); |
| ExStoredProcPrivateState & pstate |
| = *((ExStoredProcPrivateState *) pentry_down->pstate); |
| |
| ex_queue::up_status status = up_entry->upState.status; |
| |
| switch(status) |
| { |
| case ex_queue::Q_OK_MMORE: |
| { |
| if (pstate.errorHappened_) |
| { |
| // just ignore this row -- we've already returned an error. |
| } |
| else |
| { |
| // Copy the pointers to the input data |
| up_entry->copyAtp(pentry_down); |
| |
| // set the new pointer |
| up_entry->getTupp(spTdb().criDescUp_->noTuples()-1) =p; |
| |
| // Is there a selection predicate? |
| if (spTdb().predExpr_) { |
| ex_expr::exp_return_type retCode = |
| (spTdb().predExpr_->eval(up_entry->getAtp(), 0)); |
| if (retCode == ex_expr::EXPR_FALSE) { |
| break; |
| } |
| if (retCode == ex_expr::EXPR_ERROR) { |
| up_entry->upState.status = ex_queue::Q_SQLERROR; |
| up_entry->upState.parentIndex = pentry_down->downState.parentIndex; |
| up_entry->upState.downIndex = qparent_.down->getHeadIndex(); |
| up_entry->upState.setMatchNo(pstate.matchCount_); |
| qparent_.up->insert(); |
| pstate.errorHappened_ = TRUE; |
| break; |
| } |
| } |
| up_entry->upState.parentIndex = pentry_down->downState.parentIndex; |
| up_entry->upState.downIndex = qparent_.down->getHeadIndex(); |
| pstate.matchCount_++; |
| up_entry->upState.setMatchNo(pstate.matchCount_); |
| qparent_.up->insert(); |
| } |
| } |
| break; |
| |
| case ex_queue::Q_NO_DATA: |
| { |
| if (diags) |
| { |
| ComDiagsArea * da = |
| ComDiagsArea::allocate( |
| getGlobals()->getDefaultHeap()); |
| da->unpackObj(diags->getType(), |
| diags->getVersion(), |
| TRUE, // sameEndianness, |
| 0, // dummy for now... |
| (IpcConstMessageBufferPtr) diags); |
| up_entry->setDiagsArea(da); |
| } |
| |
| |
| // done with this input request. |
| up_entry->upState.status = ex_queue::Q_NO_DATA; |
| up_entry->upState.parentIndex = |
| pentry_down->downState.parentIndex; |
| up_entry->upState.setMatchNo(pstate.matchCount_); |
| |
| // insert into parent |
| qparent_.up->insert(); |
| |
| pstate.matchCount_ = 0; |
| pstate.errorHappened_ = FALSE; |
| |
| qparent_.down->removeHead(); |
| |
| numInputRequests_--; |
| } |
| break; |
| |
| case ex_queue::Q_SQLERROR: |
| { |
| if (pstate.errorHappened_) |
| { |
| // just ignore this row -- we've already returned an error. |
| } |
| else |
| { |
| ComDiagsArea * da = |
| ComDiagsArea::allocate( |
| getGlobals()->getDefaultHeap()); |
| da->unpackObj(diags->getType(), |
| diags->getVersion(), |
| TRUE, // sameEndianness, |
| 0, // dummy for now... |
| (IpcConstMessageBufferPtr) diags); |
| up_entry->setDiagsArea(da); |
| |
| // insert into parent |
| qparent_.up->insert(); |
| } |
| } |
| break; |
| |
| case ex_queue::Q_INVALID: |
| ex_assert(0, |
| "ExStoredProcTcb::work() Invalid state returned by arkcmp"); |
| break; |
| } // switch status |
| } |
| break; |
| |
| } // switch step_ |
| } // while |
| |
| return WORK_OK; |
| } |
| |
| |
| ///////////////////////////////////////////////////////////////////////////// |
| // Constructor and destructor for private state. |
| ///////////////////////////////////////////////////////////////////////////// |
| |
| ExStoredProcPrivateState::ExStoredProcPrivateState(const ExStoredProcTcb * |
| /*tcb*/) |
| { |
| matchCount_ = 0; |
| errorHappened_ = FALSE; |
| } |
| |
| ExStoredProcPrivateState::~ExStoredProcPrivateState() |
| { |
| }; |
| |
| ex_tcb_private_state * ExStoredProcPrivateState::allocate_new(const ex_tcb *tcb) |
| { |
| return new(((ex_tcb *)tcb)->getSpace()) ExStoredProcPrivateState((ExStoredProcTcb *) tcb); |
| }; |
| |
| ///////////////////////////////////////////////// |
| // class ExSPInputOutput |
| ///////////////////////////////////////////////// |
| |
| //////////////////////////////////////////////////////////////////// |
| // Moves data value from fieldNum'th position in inputRow to 'data. |
| // fieldNum is zero based, that is, the first field is number 0. |
| // Returns -1 in case of error. diagsArea contains the error number |
| // in this case (well, almost always). |
| //////////////////////////////////////////////////////////////////// |
| short ExSPInputOutput::inputValue(ULng32 fieldNum, char * inputRow, |
| char * data, ULng32 datalen, NABoolean casting, |
| ComDiagsArea * diagsArea) |
| { |
| Attributes * attr = tupleDesc_->getAttr((short)fieldNum); |
| short varcharlen = 2; |
| |
| // diagsArea must be present. And fieldNum must be within range. |
| if ((diagsArea == NULL) || |
| (fieldNum < 0) || |
| (fieldNum >= tupleDesc_->numAttrs())) |
| { |
| // return error |
| return -1; |
| } |
| |
| if (casting == FALSE) |
| convDoIt(&inputRow[attr->getOffset()], |
| attr->getLength(&inputRow[attr->getVCLenIndOffset()]), |
| attr->getDatatype(), |
| attr->getPrecision(), |
| attr->getScale(), |
| |
| // Pass in pointer to target data. |
| // if varchar, then data points to varchar length and |
| // 'data + varcharIndLen' points to the real data. |
| &data[((attr->getVCIndicatorLength() <= 0) ? 0 : |
| attr->getVCIndicatorLength())], |
| attr->getLength(), |
| attr->getDatatype(), |
| attr->getPrecision(), |
| attr->getScale(), |
| (attr->getVCIndicatorLength() > 0 ? data : NULL), |
| attr->getVCIndicatorLength(), |
| NULL, /*CollHeap *heap,*/ |
| &diagsArea, |
| CONV_UNKNOWN_LEFTPAD); |
| else |
| { |
| // TBD: |
| // if (datalen < (displayLength(attr) + |
| // ((attr->getVCIndicatorLength() > 0) ? |
| // attr->getVCIndicatorLength() : 0)) |
| // ; // return error. |
| |
| // target has to be formatted as a varchar. The first two bytes |
| // of target will contain the actual target length, followed |
| // by data bytes. |
| convDoIt(&inputRow[attr->getOffset()], |
| attr->getLength(&inputRow[attr->getVCLenIndOffset()]), |
| attr->getDatatype(), |
| attr->getPrecision(), |
| attr->getScale(), |
| &data[varcharlen], |
| datalen - varcharlen, |
| REC_BYTE_V_ASCII, |
| 0, |
| 0, |
| data, |
| varcharlen, |
| NULL, /*CollHeap *heap,*/ |
| &diagsArea, |
| CONV_UNKNOWN); |
| } |
| |
| return 0; |
| } |
| |
| //////////////////////////////////////////////////////////////////// |
| // Moves data value from 'data' to fieldNum'th position in |
| // outputRow. |
| // fieldNum is zero based, that is, the first field is number 0. |
| // Returns -1 in case of error. diagsArea contains the error number |
| // in this case (well, almost always). |
| //////////////////////////////////////////////////////////////////// |
| short ExSPInputOutput::outputValue(ULng32 fieldNum, |
| char * outputRow, |
| char * data, |
| ULng32 datalen, |
| NABoolean casting, |
| CollHeap * heap, |
| ComDiagsArea * diagsArea) |
| { |
| Attributes * attr = tupleDesc_->getAttr((short)fieldNum); |
| ex_expr::exp_return_type status = ex_expr::EXPR_OK; |
| |
| short nullIndicatorLen = attr->getNullIndicatorLength(), |
| varcharIndLen = attr->getVCIndicatorLength(), |
| scale = attr->getScale(), |
| dataType = attr->getDatatype(); |
| |
| Lng32 nullIndOffset = attr->getNullIndOffset(), |
| varcharIndOffset = attr->getVCLenIndOffset(), |
| precision = attr->getPrecision(); |
| |
| // diagsArea must be present. And fieldNum must be within range. |
| if ((diagsArea == NULL) || |
| (fieldNum < 0) || |
| (fieldNum >= tupleDesc_->numAttrs())) |
| { |
| // return error |
| ExRaiseSqlError(heap, &diagsArea, EXE_FIELD_NUM_OVERFLOW); |
| return ex_expr::EXPR_ERROR; |
| } |
| |
| if (data == NULL) |
| { |
| // a null value has been passed in. Move in a null value. |
| |
| if (attr->getNullFlag()) |
| { |
| // make the result null. |
| str_pad(&outputRow[nullIndOffset], |
| nullIndicatorLen, |
| '\377'); |
| return status; |
| } |
| else |
| { |
| // return error |
| ExRaiseSqlError(heap, &diagsArea, EXE_ASSIGNING_NULL_TO_NOT_NULL); |
| return ex_expr::EXPR_ERROR; |
| } |
| } // null value |
| |
| if (casting == FALSE) |
| status = convDoIt( |
| // Pass in pointer to real data. |
| // if varchar, then data points to varchar length and |
| // 'data + varcharIndLen' points to the real data. |
| &data[((varcharIndLen <= 0) ? 0 : varcharIndLen)], |
| attr->getLength(data), |
| dataType, |
| precision, |
| scale, |
| &outputRow[attr->getOffset()], |
| attr->getLength(), |
| dataType, |
| precision, |
| scale, |
| &outputRow[varcharIndOffset], |
| varcharIndLen, |
| heap, |
| &diagsArea, |
| CONV_UNKNOWN_LEFTPAD); |
| else |
| status = convDoIt(&data[0], |
| datalen, |
| REC_BYTE_F_ASCII, |
| precision, |
| scale, |
| &outputRow[attr->getOffset()], |
| attr->getLength(), |
| dataType, |
| precision, |
| scale, |
| (varcharIndLen > 0 ? |
| &outputRow[varcharIndOffset] : NULL), |
| varcharIndLen, |
| heap, |
| &diagsArea, |
| (getCaseIndexArray() ? getCaseIndexArray()[fieldNum] |
| : CONV_UNKNOWN_LEFTPAD)); |
| |
| if ((status == ex_expr::EXPR_OK) && attr->getNullFlag()) |
| { |
| // make the result not null. |
| str_pad(&outputRow[nullIndOffset], |
| nullIndicatorLen, |
| '\0'); |
| } |
| return status; |
| } |
| |
| // Moved to cli/StoredProcInterface.cpp |
| |
