| /********************************************************************** |
| // @@@ 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: ExPackedRows.cpp |
| * Description: Methods for the tdb and tcb of a UNPACKROWS operation |
| * |
| * |
| * Created: 6/19/97 |
| * Language: C++ |
| * |
| * |
| * |
| * |
| ****************************************************************************** |
| */ |
| |
| // |
| // This file contains all the executor methods associated |
| // with a unPackRows operator |
| // |
| |
| #include "ExPackedRows.h" |
| #include "ExSimpleSqlBuffer.h" |
| #include "ex_error.h" |
| #include "ExpError.h" |
| |
| |
| ////////////////////////////////////////////////////////////////////////////// |
| // |
| // TDB procedures |
| // |
| ////////////////////////////////////////////////////////////////////////////// |
| |
| |
| // ExUnPackRowsTdb::build() -------------------------------------------- |
| // Build the TCB tree for this node and all nodes below. |
| // |
| // Parameters |
| // |
| // ex_globals *glob - contains references to global executor information, |
| // notably the space object used to allocate objects. |
| // |
| // Returns - returns the TCB tree build for this node and all nodes below. |
| // |
| ex_tcb * |
| ExUnPackRowsTdb::build(ex_globals *glob) |
| { |
| |
| // Build the Tcb tree below the unPackRows node. |
| // |
| ex_tcb *childTcb = NULL; |
| if (childTdb_) |
| childTcb = childTdb_->build(glob); |
| |
| // Build the unPackRows Tcb, given the TDB. |
| // |
| ex_tcb *unPackRowsTcb = NULL; |
| |
| if (rowwiseRowset()) |
| unPackRowsTcb = |
| new(glob->getSpace()) ExUnPackRowwiseRowsTcb(*this, glob); |
| else |
| unPackRowsTcb = |
| new(glob->getSpace()) ExUnPackRowsTcb(*this, *childTcb, glob); |
| |
| // add the unPackRowsTcb and all its work procedures to the schedule |
| // |
| unPackRowsTcb->registerSubtasks(); |
| |
| // Return the result to my parent. |
| // |
| return unPackRowsTcb; |
| } |
| |
| |
| ///////////////////////////////////////////////////////////////////////// |
| // |
| // TCB procedures |
| // |
| ///////////////////////////////////////////////////////////////////////// |
| |
| // ExUnPackRowsTcb::ExUnPackRowsTcb() ------------------------------------ |
| // Constructor for unPackRows TCB |
| // |
| ExUnPackRowsTcb::ExUnPackRowsTcb(const ExUnPackRowsTdb &unPackRowsTdbLocal, |
| const ex_tcb &childTcb, |
| ex_globals *glob) : |
| ex_tcb(unPackRowsTdbLocal, 1, glob) |
| { |
| |
| childTcb_ = &childTcb; |
| CollHeap *space = glob->getSpace(); |
| |
| // Allocate the buffer pool. Allocate a buffer big enough to hold |
| // twice the size of my upqueue. |
| // |
| pool_ = new(space) ExSimpleSQLBuffer(unPackRowsTdb().queueSizeUp_ * 2, |
| unPackRowsTdb().unPackColsTupleLen_, |
| space); |
| |
| // get the queue used by my child to communicate with me |
| // |
| childQueue_ = childTcb_->getParentQueue(); |
| |
| // Allocate the queue to communicate with parent |
| // (Child allocates queue to communicate with Child) |
| // |
| qParent_.down = new(space) ex_queue(ex_queue::DOWN_QUEUE, |
| unPackRowsTdb().queueSizeDown_, |
| unPackRowsTdb().criDescDown_, |
| space); |
| |
| // Allocate the private state in each entry of the down queue |
| // |
| ExUnPackRowsPrivateState privateState(this); |
| qParent_.down->allocatePstate(&privateState, this); |
| |
| |
| // Initialize processedInputs_ to refer to the queue entry |
| // which will be used next. |
| // |
| processedInputs_ = qParent_.down->getHeadIndex(); |
| |
| // Allocate a queue to communicate with the parent node. |
| // |
| qParent_.up = new(space) ex_queue(ex_queue::UP_QUEUE, |
| unPackRowsTdb().queueSizeUp_, |
| unPackRowsTdb().criDescUp_, |
| space); |
| |
| // fixup all expressions expressions |
| // |
| |
| if(unPackColsExpr()) |
| unPackColsExpr()->fixup(0, |
| getExpressionMode(), this, |
| glob->getSpace(), |
| glob->getDefaultHeap(), FALSE, glob); |
| |
| // Initialize the workAtp_ and the Tupp used to hold the |
| // index value. This index value is an integer local to the |
| // workUp method. |
| // |
| indexValueTuppDesc_.init(4, 0, (char *)&indexValue_); |
| indexValueTupp_.init(); |
| indexValueTupp_ = &indexValueTuppDesc_; |
| workAtp_ = allocateAtp(unPackRowsTdb().workCriDesc_, |
| glob->getSpace()); |
| |
| // Must always have a packingFactor expression |
| // |
| ex_assert(packingFactor(), |
| "ExUnPackRowsTcb::ExUnPackRowsTcb(): Internal Error"); |
| |
| packingFactor()->fixup(0, |
| getExpressionMode(), this, |
| glob->getSpace(), |
| glob->getDefaultHeap(), FALSE, glob); |
| |
| // Initialize the numRowsAtp_ and the numRowsTupp. These are used |
| // to hold the result of the move expression (packingFactor()) used |
| // to extract the packing Factor from one of the packed rows. |
| // |
| numRowsTuppDesc_.init(4, 0, (char *)&numRows_); |
| numRowsTupp_.init(); |
| numRowsTupp_ = &numRowsTuppDesc_; |
| numRowsAtp_ = allocateAtp(unPackRowsTdb().criDescUp_, |
| glob->getSpace()); |
| |
| } |
| |
| // ExUnPackRowsTcb::~ExUnPackRowsTcb() --------------------------------------- |
| // Destructor for ExUnPackRowsTcb |
| // Free up the queues and the sql buffer and the child TCB tree. |
| // |
| ExUnPackRowsTcb::~ExUnPackRowsTcb() |
| { |
| delete qParent_.up; |
| delete qParent_.down; |
| freeResources(); |
| } |
| |
| // ExUnPackRowsTcb::freeResources() ---------------------------------------- |
| // Free any run time resources. |
| // For unPackRows, this means freeing up the buffer pool. |
| // The queues are not freed here. (Should they be?). |
| // This method is call by the TCB destructor. |
| // |
| void ExUnPackRowsTcb::freeResources() |
| { |
| // Add this when space_ is added; |
| // if (workAtp_) { |
| // deallocateAtp(workAtp_, space_); |
| // workAtp_ = NULL; |
| // } |
| |
| // if (numRowsAtp_) { |
| // deallocateAtp(numRowsAtp_, space_); |
| // numRowsAtp_ = NULL; |
| // } |
| |
| if (pool_) { |
| delete pool_; |
| pool_ = NULL; |
| } |
| } |
| |
| // ExUnPackRowsTcb::registerSubtasks() ------------------------------------- |
| // Register all the unPackRows subtasks with the scheduler. |
| // |
| void ExUnPackRowsTcb::registerSubtasks() |
| { |
| ExScheduler *sched = getGlobals()->getScheduler(); |
| |
| // down queues are handled by workDown() |
| // |
| |
| // Schedule this routine if a new entry is inserted into the |
| // parents down queue and we are waiting on an event. |
| // (had returned WORK_OK). |
| // |
| sched->registerInsertSubtask(sWorkDown, this, qParent_.down, "DN"); |
| |
| // Schedule this routine if the child's down queue changes from being |
| // full to being not full and we are waiting on an event. |
| // (had returned WORK_OK). |
| // |
| sched->registerUnblockSubtask(sWorkDown, this, childQueue_.down, "DN"); |
| |
| // Schedule this routine if a cancel request occurs on the |
| // parents down queue. |
| // |
| sched->registerCancelSubtask(sCancel, this, qParent_.down, "CN"); |
| |
| // up queues are handled by workUp() |
| // |
| |
| // Schedule this routine if the parent's up queue changes from being |
| // full to being not full and we are waiting on an event. |
| // (had returned WORK_OK). |
| // |
| sched->registerUnblockSubtask(sWorkUp, this, qParent_.up,"UP"); |
| |
| // Schedule this routine if a new entry is inserted into the |
| // child's up queue and we are waiting on an event. |
| // (had returned WORK_OK). |
| // |
| sched->registerInsertSubtask(sWorkUp, this, childQueue_.up); |
| } |
| |
| // ExUnPackRowsTcb::start() ------------------------------------------ |
| // Starts up the next requested operation by sending |
| // requests to children. Child queue must have room! |
| // Called by workDown(). |
| // |
| void |
| ExUnPackRowsTcb::start() |
| { |
| // Advance processedInputs_ by one, all checks need to be done |
| // by the caller. No down request is sent for a cancelled message. |
| // |
| ex_queue_entry *childEntry = childQueue_.down->getTailEntry(); |
| |
| ex_queue_entry *pEntryDown = qParent_.down->getQueueEntry(processedInputs_); |
| const ex_queue::down_request request = pEntryDown->downState.request; |
| |
| ExUnPackRowsPrivateState &pState = |
| *((ExUnPackRowsPrivateState*)pEntryDown->pstate); |
| |
| // pass request to children |
| // |
| childEntry->downState.request = ex_queue::GET_ALL; |
| childEntry->downState.requestValue = pEntryDown->downState.requestValue; |
| childEntry->downState.parentIndex = processedInputs_; |
| childEntry->passAtp(pEntryDown); |
| |
| // no matches yet for this parent row |
| // |
| pState.matchCount_ = 0; |
| |
| // Insert request in child queue. |
| // |
| childQueue_.down->insert(); |
| |
| // Record the state of this request. |
| // |
| pState.childState_ = STARTED_; |
| |
| // If this is a CANCEL request, Cancel the request we just gave to |
| // our child. |
| // |
| if (request == ex_queue::GET_NOMORE) { |
| // immediately cancel the request (requests are already in |
| // cancelled state but the cancel callback isn't activated yet) |
| // |
| childQueue_.down->cancelRequestWithParentIndex(processedInputs_); |
| pState.childState_ = CANCELLED_; |
| |
| // $$$$ should find a smarter way to do this |
| } |
| |
| // We are now ready to look at the next request. |
| // |
| processedInputs_++; |
| } |
| |
| // ExUnPackRowsTcb::stop() ---------------------------------------------- |
| // All child rows have been returned. |
| // Now return an EOD indication to parent. |
| // |
| void ExUnPackRowsTcb::stop() |
| { |
| // Remove the head entries of the parent down queue and the |
| // child up queue. All checks are done by the caller. |
| // |
| ex_queue_entry *pEntryDown = qParent_.down->getHeadEntry(); |
| |
| ExUnPackRowsPrivateState &pState = |
| *((ExUnPackRowsPrivateState*) pEntryDown->pstate); |
| |
| ex_queue_entry *pEntry = qParent_.up->getTailEntry(); |
| |
| pEntry->upState.status = ex_queue::Q_NO_DATA; |
| pEntry->upState.parentIndex = pEntryDown->downState.parentIndex; |
| pEntry->upState.downIndex = qParent_.down->getHeadIndex(); |
| pEntry->upState.setMatchNo(pState.matchCount_); |
| |
| // insert EOD into parent |
| // |
| qParent_.up->insert(); |
| |
| // consume the child row |
| // |
| childQueue_.up->removeHead(); |
| |
| // Reinitialize the state of this processed request. |
| // Pstate will be initialized for when this entry gets used again. |
| // |
| pState.init(); |
| |
| // this parent request has been processed. |
| // |
| qParent_.down->removeHead(); |
| } |
| |
| ExWorkProcRetcode |
| ExUnPackRowsTcb::processCancel() |
| { |
| // Check the down queue from the parent for cancellations. Propagate |
| // cancel requests and remove all requests that are completely cancelled. |
| |
| // loop over all requests that have been sent down |
| // (the others are handled by start() later) |
| queue_index ix = qParent_.down->getHeadIndex(); |
| |
| while (ix != processedInputs_) { |
| ex_queue_entry *pEntryDown = qParent_.down->getQueueEntry(ix); |
| |
| // check whether the current down request is cancelled |
| // |
| if (pEntryDown->downState.request == ex_queue::GET_NOMORE) { |
| |
| // yes, cancel this down request if not already done so |
| // |
| ExUnPackRowsPrivateState &pState = |
| *((ExUnPackRowsPrivateState*) pEntryDown->pstate); |
| |
| // cancel child requests if not already done so and if they |
| // aren't already finished |
| // |
| if (pState.childState_ == ExUnPackRowsTcb::STARTED_) { |
| |
| // cancel request to this child |
| // |
| childQueue_.down->cancelRequestWithParentIndex(ix); |
| pState.childState_ = ExUnPackRowsTcb::CANCELLED_; |
| } |
| |
| } // cancelled rows request |
| |
| ix++; // may wrap around |
| } // parent queue is not empty |
| |
| // checked all active entries nothing more to do |
| return WORK_OK; |
| } |
| |
| // ----------------------------------------------------------------------- |
| // Generic work procedure should never be called |
| // ----------------------------------------------------------------------- |
| ExWorkProcRetcode ExUnPackRowsTcb::work() |
| { |
| ex_assert(0,"Should never reach ExUnPackRowsTcb::work()"); |
| return WORK_BAD_ERROR; |
| } |
| |
| // ----------------------------------------------------------------------- |
| // Work procedure to send requests down |
| // ----------------------------------------------------------------------- |
| ExWorkProcRetcode ExUnPackRowsTcb::workDown() |
| { |
| // while we have unprocessed down requests and while |
| // child's down ComQueue.has room, start more child requests |
| // |
| while (qParent_.down->entryExists(processedInputs_) && |
| ! childQueue_.down->isFull()) |
| start(); |
| |
| return WORK_OK; |
| } |
| |
| // ----------------------------------------------------------------------- |
| // generic up-processing for unPackRows |
| // ----------------------------------------------------------------------- |
| |
| ExWorkProcRetcode |
| ExUnPackRowsTcb::workUp() |
| { |
| |
| // while there is a chance that we have work (may exit via return) |
| // (If processedInputs_ equals the HeadIndex(), then there are no |
| // entries in the parent down queue.) |
| // |
| const NABoolean rowsetIterator = unPackRowsTdb().isRowsetIterator(); |
| const NABoolean tolerateNonFatalError = unPackRowsTdb().isNonFatalErrorTolerated(); |
| |
| while (qParent_.down->getHeadIndex() != processedInputs_) { |
| |
| // get head entry and pState |
| // |
| ex_queue_entry *pEntryDown = qParent_.down->getHeadEntry(); |
| |
| ExUnPackRowsPrivateState &pState = |
| *((ExUnPackRowsPrivateState*)pEntryDown->pstate); |
| |
| |
| // Set up the numRowsAtp_ to recieve the packing factor. |
| // |
| numRowsAtp_->copyAtp(pEntryDown->getAtp()); |
| |
| numRowsAtp_-> |
| getTupp(unPackRowsTdb().criDescUp_->noTuples() - 1) |
| = numRowsTupp_; |
| |
| // We are not returning any of the childs returned values. |
| // So we copy from the parents request. |
| // |
| workAtp_->copyAtp(pEntryDown->getAtp()); |
| |
| // Set up the Tupp for the index value. The index value is |
| // a data member of this tcb and is input to the unPackExpr. |
| // |
| workAtp_->getTupp(unPackRowsTdb().indexValueAtpIndex_) |
| = indexValueTupp_; |
| |
| // Get the current value of the index for this request. |
| // |
| indexValue_ = pState.unPackCount_; |
| |
| // for non-atomic rowsets store the next rownumber where a CLI |
| // error was raised. This row will be skipped later. |
| if (tolerateNonFatalError) { |
| ComDiagsArea * da = workAtp_->getDiagsArea(); |
| if (da) { |
| pState.nextCLIErrorRowNum_ = da->getNextRowNumber(indexValue_); |
| } |
| } |
| |
| // while we have room in the up queue and rows to process |
| // |
| while (!qParent_.up->isFull() && !childQueue_.up->isEmpty()) { |
| |
| ex_queue_entry *cEntry = childQueue_.up->getHeadEntry(); |
| atp_struct * childAtp = cEntry->getAtp(); |
| |
| if(cEntry->upState.status == ex_queue::Q_NO_DATA) { |
| |
| // Send EOD to parent and clean up the current request. |
| // (will loop back to to of routine and try the next |
| // request if there is one) |
| // |
| stop(); |
| |
| // Must break here (rather than continue, so that we will |
| // reset the pEntryDown and pState.) |
| // |
| break; |
| } else if(pState.childState_ == CANCELLED_) { |
| } else if(cEntry->upState.status == ex_queue::Q_SQLERROR) { |
| processError(childAtp,FALSE,0); |
| continue; |
| } else if (pEntryDown->downState.request == ex_queue::GET_N && |
| pEntryDown->downState.requestValue <= |
| (Lng32)pState.matchCount_) { |
| |
| qParent_.down->cancelRequest(qParent_.down->getHeadIndex()); |
| processCancel(); |
| continue; |
| |
| } else { |
| |
| ex_expr::exp_return_type retCode = ex_expr::EXPR_TRUE; |
| |
| // If this is the first row to be unpacked from this packed row. |
| // |
| if(indexValue_ == 0) { |
| |
| // Extract the packing factor (numRows) from the packed column. |
| // |
| retCode = packingFactor()->eval(numRowsAtp_, childAtp); |
| |
| if(retCode == ex_expr::EXPR_ERROR) { |
| processError(workAtp_,FALSE,0); |
| continue; |
| } |
| |
| // Register the numRows in the private state, in case we |
| // start working on an other request. |
| // |
| pState.numRows_ = numRows_; |
| |
| // If the number of entries in the rowset is negative, we |
| // signal an error |
| if (numRows_ < 0) { |
| ComDiagsArea * da = workAtp_->getDiagsArea(); |
| if (!da) |
| { |
| da = ComDiagsArea::allocate(getHeap()); |
| workAtp_->setDiagsArea(da); |
| } |
| *da << DgSqlCode(-EXE_ROWSET_INDEX_OUTOF_RANGE); |
| processError(workAtp_,FALSE,0); |
| continue; |
| } |
| |
| } |
| |
| // Only allocate a new tuple if needed. If a tuple has |
| // been allocated, but not inserted into the up queue |
| // because the predicate did not pass, reuse the tuple. |
| // |
| Int32 needTuple = (unPackColsExpr() ? TRUE : FALSE); |
| |
| // While we still have expressions to apply to the current |
| // child entry and the parent can accept rows. |
| // Each expression that is applied to the child entry |
| // produces one row to be put onto the parents up queue. |
| // |
| while(indexValue_ < pState.numRows_ && !qParent_.up->isFull()) { |
| |
| // Handle GET_N processing |
| // |
| if (pEntryDown->downState.request == ex_queue::GET_N && |
| pEntryDown->downState.requestValue <= (Lng32)pState.matchCount_) { |
| qParent_.down->cancelRequest(qParent_.down->getHeadIndex()); |
| processCancel(); |
| indexValue_ = pState.numRows_; |
| continue; |
| } |
| |
| // For Non-atomic rowsets skips over rows that raised an error |
| // in the CLI |
| if (pState.nextCLIErrorRowNum_ == indexValue_) { |
| processSkippedRow(workAtp_); |
| indexValue_++; |
| ComDiagsArea * da = workAtp_->getDiagsArea(); |
| pState.nextCLIErrorRowNum_ = da->getNextRowNumber(indexValue_); |
| continue ; |
| } |
| |
| // UnPackRows also adds one TP for the generated columns. A |
| // tuple must be allocated if one isn't already available. |
| // |
| if(needTuple) { |
| if (pool_-> |
| getFreeTuple(workAtp_-> |
| getTupp(unPackRowsTdb().unPackColsAtpIndex_))) { |
| // Return. Will be called again when some space frees up. |
| // |
| pState.unPackCount_ = indexValue_; |
| return WORK_POOL_BLOCKED; |
| } |
| |
| // Set flag to indicate that we have allocated a tuple. |
| // |
| needTuple = FALSE; |
| } |
| |
| // Apply any selection predicate. |
| // |
| // The unPackColsExpr() will be NULL when there are no |
| // columns to unPack, but we still need to produce the |
| // proper number of rows. For example count(*). |
| // |
| // Apply the unPack expression to this row. This will |
| // generate the proper values for the generated columns. |
| // |
| |
| Lng32 markValue = 0; |
| ComDiagsArea * da = workAtp_->getDiagsArea(); |
| if (da) |
| markValue = da->mark(); |
| |
| retCode = (unPackColsExpr() ? |
| unPackColsExpr()->eval(workAtp_, childAtp) : |
| ex_expr::EXPR_TRUE); |
| |
| switch(retCode) { |
| case ex_expr::EXPR_TRUE: |
| { |
| // Get the entry on the parent up queue. |
| // |
| ex_queue_entry *pEntry = qParent_.up->getTailEntry(); |
| |
| pEntry->copyAtp(workAtp_); |
| |
| // for non-atomic rowsets do NOT flow digas area which we get from parent |
| // this diags area contains errors from the CLI. We do not want to flow them up |
| // with OK_MMORE replies as this will lead to duplication. |
| // We will flow them up with the Q_REC_SKIPPED replies. |
| if (tolerateNonFatalError) |
| pEntry->setDiagsArea(NULL); |
| |
| // Release the reference to the tupp in the workAtp_ |
| // |
| workAtp_->getTupp(unPackRowsTdb().unPackColsAtpIndex_).release(); |
| |
| // Set the flag to indicate that the tuple has been used and |
| // a new tuple needs to be allocated. |
| // |
| needTuple = (unPackColsExpr() ? TRUE : FALSE); |
| |
| // Finialize the queue entry, then insert it |
| // |
| pEntry->upState.status = cEntry->upState.status; |
| pEntry->upState.parentIndex = pEntryDown->downState.parentIndex; |
| pEntry->upState.setMatchNo(++pState.matchCount_); |
| qParent_.up->insert(); |
| break; |
| } |
| case ex_expr::EXPR_FALSE: |
| if (rowsetIterator) |
| processSkippedRow(NULL); |
| else { |
| // Do nothing |
| } |
| break; |
| case ex_expr::EXPR_ERROR: |
| processError(workAtp_,tolerateNonFatalError,markValue); |
| |
| // Release the reference to the tupp in the workAtp_ |
| // |
| workAtp_->getTupp(unPackRowsTdb().unPackColsAtpIndex_).release(); |
| |
| // Set the flag to indicate that the tuple has been used and |
| // a new tuple needs to be allocated. |
| // |
| needTuple = (unPackColsExpr() ? TRUE : FALSE); |
| break; |
| default: |
| break; |
| } |
| |
| if(retCode == ex_expr::EXPR_ERROR && !tolerateNonFatalError) |
| break; |
| |
| // Advance the expression counter. |
| // |
| indexValue_++; |
| |
| } |
| |
| // This will cause the EOD to be sent after error processing. |
| // |
| if(retCode == ex_expr::EXPR_ERROR && !tolerateNonFatalError) |
| continue; |
| |
| if(indexValue_ < pState.numRows_) { |
| // We still have more expressions to apply. We must have |
| // reached this point because the parent up queue is full. |
| // Return, will be called again when the queue becomes non-full. |
| // |
| pState.unPackCount_ = indexValue_; |
| return WORK_OK; |
| } |
| } |
| // We have applied all the expressions on this child entry, |
| // reset the expression counter, remove the child entry |
| // and then see if there is another entry to be processed. |
| // |
| indexValue_ = 0; |
| childQueue_.up->removeHead(); |
| |
| } // while parent up ComQueue.has room and child up ComQueue.has replies |
| |
| // If we got here because we finished a request and stop()ed, |
| // then try again on a new request. But if we got here because |
| // the parent queue is full or the child queue is empty, then |
| // we must return. |
| // |
| if(qParent_.up->isFull() || childQueue_.up->isEmpty()) { |
| pState.unPackCount_ = indexValue_; |
| return WORK_OK; |
| } |
| |
| pState.unPackCount_ = indexValue_; |
| |
| } // while parent ComQueue.has entries |
| |
| // parent down queue is empty |
| // |
| return WORK_OK; |
| } |
| |
| void |
| ExUnPackRowsTcb::processError(atp_struct *atp, NABoolean isNonFatalError, Lng32 markValue) |
| { |
| ex_queue_entry * pEntryDown = qParent_.down->getHeadEntry(); |
| ex_queue_entry * pEntry = qParent_.up->getTailEntry(); |
| ExUnPackRowsPrivateState &pState = |
| *((ExUnPackRowsPrivateState*) pEntryDown->pstate); |
| |
| pEntry->upState.status = ex_queue::Q_SQLERROR; |
| pEntry->upState.parentIndex = pEntryDown->downState.parentIndex; |
| pEntry->upState.downIndex = qParent_.down->getHeadIndex(); |
| pEntry->upState.setMatchNo(pState.matchCount_); |
| |
| if (!isNonFatalError) { |
| pEntry->copyAtp(atp); |
| |
| // cancel this request and cancel all the children. |
| qParent_.down->cancelRequest(qParent_.down->getHeadIndex()); |
| pState.childState_ = CANCELLED_; |
| |
| while(!childQueue_.up->isEmpty() && |
| (childQueue_.up->getHeadEntry()->upState.status |
| != ex_queue::Q_NO_DATA)) |
| childQueue_.up->removeHead(); |
| } |
| else { |
| ComDiagsArea * fromDa = atp->getDiagsArea(); |
| ex_assert(fromDa, "We have an expression error in UnPack but no diags area"); |
| ComDiagsArea* toDa = ComDiagsArea::allocate(getHeap()); |
| fromDa->rewindAndMergeIfDifferent(markValue, toDa); |
| pEntry->setDiagsArea(toDa); |
| toDa->setAllRowNumber(ComCondition::NONFATAL_ERROR) ; |
| } |
| qParent_.up->insert(); |
| } |
| |
| void |
| ExUnPackRowsTcb::processSkippedRow(atp_struct *atp) |
| { |
| ex_queue_entry * pEntryDown = qParent_.down->getHeadEntry(); |
| ex_queue_entry * pEntry = qParent_.up->getTailEntry(); |
| ExUnPackRowsPrivateState &pState = |
| *((ExUnPackRowsPrivateState*) pEntryDown->pstate); |
| |
| pEntry->upState.status = ex_queue::Q_REC_SKIPPED; |
| pEntry->upState.parentIndex = pEntryDown->downState.parentIndex; |
| pEntry->upState.downIndex = qParent_.down->getHeadIndex(); |
| pEntry->upState.setMatchNo(pState.matchCount_); |
| if (atp) |
| pEntry->copyAtp(atp); |
| qParent_.up->insert(); |
| } |
| |
| /////////////////////////////////////////////////////////////// |
| // |
| // Private state procedures |
| // |
| /////////////////////////////////////////////////////////////// |
| |
| // Constructor and destructor for ExUnPackRowsPrivateState |
| // |
| ExUnPackRowsPrivateState::ExUnPackRowsPrivateState(const ExUnPackRowsTcb *) |
| { |
| init(); |
| } |
| |
| |
| // Initialize the data members of the private state. |
| // Called by the constructor when the private state is |
| // initially allocated and when we are now with a parent |
| // request, so that the next time the queue entry is used, |
| // it's private state is already initialized. |
| // |
| void |
| ExUnPackRowsPrivateState::init() |
| { |
| matchCount_ = 0; |
| unPackCount_ = 0; |
| numRows_ = 0; |
| childState_ = ExUnPackRowsTcb::EMPTY_; |
| // row number is a zero based index, so initialize to -1. |
| nextCLIErrorRowNum_ = -1; |
| } |
| |
| ExUnPackRowsPrivateState::~ExUnPackRowsPrivateState() |
| { |
| } |
| |
| ex_tcb_private_state * |
| ExUnPackRowsPrivateState::allocate_new(const ex_tcb *tcb) |
| { |
| return new(((ex_tcb *)tcb)->getSpace()) |
| ExUnPackRowsPrivateState((ExUnPackRowsTcb *)tcb); |
| } |
| |
| |
| // ExUnPackRowwiseRowsTcb::ExUnPackRowwiseRowsTcb() ------------------------- |
| // |
| ExUnPackRowwiseRowsTcb::ExUnPackRowwiseRowsTcb( |
| const ExUnPackRowsTdb &unPackRowsTdbLocal, |
| ex_globals *glob) : |
| ex_tcb(unPackRowsTdbLocal, 1, glob) |
| { |
| CollHeap *space = glob->getSpace(); |
| |
| pool_ = new(space) sql_buffer_pool(uprTdb().numBuffers_, |
| uprTdb().bufferSize_, |
| space); |
| |
| // Allocate the queue to communicate with parent |
| // (Child allocates queue to communicate with Child) |
| // |
| qParent_.down = new(space) ex_queue(ex_queue::DOWN_QUEUE, |
| uprTdb().queueSizeDown_, |
| uprTdb().criDescDown_, |
| space); |
| |
| // Allocate the private state in each entry of the down queue |
| // |
| ExUnPackRowwiseRowsPrivateState privateState(this); |
| qParent_.down->allocatePstate(&privateState, this); |
| |
| // Allocate a queue to communicate with the parent node. |
| // |
| qParent_.up = new(space) ex_queue(ex_queue::UP_QUEUE, |
| uprTdb().queueSizeUp_, |
| uprTdb().criDescUp_, |
| space); |
| |
| // fixup all expressions |
| if (uprTdb().rwrsInputSizeExpr()) |
| uprTdb().rwrsInputSizeExpr()->fixup(0, |
| getExpressionMode(), this, |
| glob->getSpace(), |
| glob->getDefaultHeap(), FALSE, glob); |
| |
| if (uprTdb().rwrsMaxInputRowlenExpr()) |
| uprTdb().rwrsMaxInputRowlenExpr()->fixup(0, |
| getExpressionMode(), this, |
| glob->getSpace(), |
| glob->getDefaultHeap(), FALSE, glob); |
| |
| if (uprTdb().rwrsBufferAddrExpr()) |
| uprTdb().rwrsBufferAddrExpr()->fixup(0, |
| getExpressionMode(), this, |
| glob->getSpace(), |
| glob->getDefaultHeap(), FALSE, glob); |
| |
| rwrsInputValuesTuppDesc_.init(); |
| //rwrsInputValuesTupp_.init(); |
| workAtp_ = allocateAtp(uprTdb().workCriDesc_, |
| glob->getSpace()); |
| workAtp_->getTupp(uprTdb().rwrsWorkIndex()) = |
| &rwrsInputValuesTuppDesc_; |
| |
| rwrsNumRows_ = 0; |
| rwrsMaxInputRowlen_ = -1; |
| rwrsBufferAddr_ = NULL; |
| |
| step_ = INITIAL_; |
| } |
| |
| ExUnPackRowwiseRowsTcb::~ExUnPackRowwiseRowsTcb() |
| { |
| delete qParent_.up; |
| delete qParent_.down; |
| } |
| |
| ExWorkProcRetcode |
| ExUnPackRowwiseRowsTcb::work() |
| { |
| ex_expr::exp_return_type retCode; |
| |
| // if no parent request, return |
| if (qParent_.down->isEmpty()) |
| return WORK_OK; |
| |
| ex_queue_entry * pentry_down = qParent_.down->getHeadEntry(); |
| ExUnPackRowwiseRowsPrivateState * pstate |
| = (ExUnPackRowwiseRowsPrivateState*)pentry_down->pstate; |
| ex_queue::down_request request = pentry_down->downState.request; |
| |
| while (TRUE) // exit via return |
| { |
| switch (step_) |
| { |
| case INITIAL_: |
| { |
| rwrsNumRows_ = 0; |
| rwrsMaxInputRowlen_ = -1; |
| rwrsBufferAddr_ = NULL; |
| |
| currentRowNum_ = 0; |
| |
| step_ = GET_INPUT_VALUES_; |
| } |
| break; |
| |
| case GET_INPUT_VALUES_: |
| { |
| // get number of rows in the input rwrs buffer |
| workAtp_->getTupp(uprTdb().rwrsWorkIndex()). |
| setDataPointer((char *)&rwrsNumRows_); |
| retCode = |
| uprTdb().rwrsInputSizeExpr()->eval(pentry_down->getAtp(), |
| workAtp_); |
| if (retCode == ex_expr::EXPR_ERROR) |
| { |
| step_ = ERROR_; |
| break; |
| } |
| |
| if (rwrsNumRows_ == 0) |
| { |
| step_ = DONE_; |
| break; |
| } |
| |
| if (rwrsNumRows_ < 0) |
| { |
| ExHandleErrors(qParent_, |
| pentry_down, |
| 0, |
| getGlobals(), |
| NULL, |
| EXE_NUMERIC_OVERFLOW, |
| NULL, |
| NULL |
| ); |
| step_ = DONE_; |
| break; |
| } |
| |
| // get max length of each row |
| workAtp_->getTupp(uprTdb().rwrsWorkIndex()). |
| setDataPointer((char *)&rwrsMaxInputRowlen_); |
| retCode = |
| uprTdb().rwrsMaxInputRowlenExpr()->eval(pentry_down->getAtp(), |
| workAtp_); |
| if (retCode == ex_expr::EXPR_ERROR) |
| { |
| step_ = ERROR_; |
| break; |
| } |
| |
| if (rwrsMaxInputRowlen_ <= 0) |
| { |
| ExHandleErrors(qParent_, |
| pentry_down, |
| 0, |
| getGlobals(), |
| NULL, |
| EXE_NUMERIC_OVERFLOW, |
| NULL, |
| NULL |
| ); |
| step_ = DONE_; |
| break; |
| } |
| |
| // get address of rwrs buffer |
| workAtp_->getTupp(uprTdb().rwrsWorkIndex()). |
| setDataPointer((char *)&rwrsBufferAddr_); |
| retCode = |
| uprTdb().rwrsBufferAddrExpr()->eval(pentry_down->getAtp(), |
| workAtp_); |
| if (retCode == ex_expr::EXPR_ERROR) |
| { |
| step_ = ERROR_; |
| break; |
| } |
| |
| step_ = RETURN_ROW_; |
| } |
| break; |
| |
| case RETURN_ROW_: |
| { |
| if (qParent_.up->isFull()) |
| return WORK_OK; |
| |
| ex_queue_entry * up_entry = qParent_.up->getTailEntry(); |
| |
| // get pointer to current row |
| char * currRow = (char*)(rwrsBufferAddr_ + |
| currentRowNum_ * rwrsMaxInputRowlen_); |
| |
| // allocate an empty tupp descriptor |
| tupp_descriptor * td = pool_->get_free_tupp_descriptor(0); |
| if (! td) |
| return WORK_POOL_BLOCKED; // couldn't allocate, try again later. |
| |
| // initialize it with the addr of the row to be returned |
| td->init(rwrsMaxInputRowlen_, NULL, currRow); |
| |
| up_entry->copyAtp(pentry_down); |
| |
| // move this row to up queue |
| up_entry->getTupp(uprTdb().unPackColsAtpIndex_) = td; |
| up_entry->upState.status = ex_queue::Q_OK_MMORE; |
| up_entry->upState.parentIndex |
| = pentry_down->downState.parentIndex; |
| up_entry->upState.downIndex = qParent_.down->getHeadIndex(); |
| up_entry->upState.setMatchNo(1); |
| |
| qParent_.up->insert(); |
| |
| currentRowNum_++; |
| if (currentRowNum_ == rwrsNumRows_) |
| { |
| step_ = DONE_; |
| break; |
| } |
| |
| } |
| break; |
| |
| case ERROR_: |
| { |
| if (qParent_.up->isFull()) |
| return WORK_OK; |
| |
| // Return EOF. |
| ex_queue_entry * up_entry = qParent_.up->getTailEntry(); |
| |
| up_entry->upState.parentIndex = |
| pentry_down->downState.parentIndex; |
| |
| up_entry->upState.setMatchNo(0); |
| up_entry->upState.status = ex_queue::Q_SQLERROR; |
| |
| ComDiagsArea *diagsArea = pentry_down->getDiagsArea(); |
| |
| if (diagsArea != up_entry->getDiagsArea()) |
| { |
| up_entry->setDiagsArea (diagsArea); |
| if (diagsArea != NULL) |
| diagsArea->incrRefCount(); |
| } |
| // insert into parent |
| qParent_.up->insert(); |
| |
| step_ = DONE_; |
| } |
| break; |
| |
| case DONE_: |
| { |
| if (qParent_.up->isFull()) |
| return WORK_OK; |
| |
| // Return EOF. |
| ex_queue_entry * up_entry = qParent_.up->getTailEntry(); |
| |
| up_entry->upState.parentIndex = |
| pentry_down->downState.parentIndex; |
| |
| up_entry->upState.setMatchNo(0); |
| up_entry->upState.status = ex_queue::Q_NO_DATA; |
| |
| // insert into parent |
| qParent_.up->insert(); |
| |
| step_ = INITIAL_; |
| qParent_.down->removeHead(); |
| |
| return WORK_OK; |
| } |
| break; |
| |
| } |
| } // while |
| |
| return WORK_OK; |
| } |
| |
| |
| /////////////////////////////////////////////////////////////// |
| // |
| // Private state procedures |
| // |
| /////////////////////////////////////////////////////////////// |
| |
| // Constructor and destructor for ExUnPackRowsPrivateState |
| // |
| ExUnPackRowwiseRowsPrivateState::ExUnPackRowwiseRowsPrivateState(const ExUnPackRowwiseRowsTcb *) |
| { |
| init(); |
| } |
| |
| |
| // Initialize the data members of the private state. |
| // Called by the constructor when the private state is |
| // initially allocated and when we are now with a parent |
| // request, so that the next time the queue entry is used, |
| // it's private state is already initialized. |
| // |
| void |
| ExUnPackRowwiseRowsPrivateState::init() |
| { |
| |
| } |
| |
| ExUnPackRowwiseRowsPrivateState::~ExUnPackRowwiseRowsPrivateState() |
| { |
| } |
| |
| ex_tcb_private_state * |
| ExUnPackRowwiseRowsPrivateState::allocate_new(const ex_tcb *tcb) |
| { |
| return new(((ex_tcb *)tcb)->getSpace()) |
| ExUnPackRowwiseRowsPrivateState((ExUnPackRowwiseRowsTcb *)tcb); |
| } |
| |
| |
| |
