| /********************************************************************** |
| // @@@ 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: <file> |
| * Description: |
| * |
| * |
| * Created: 7/10/95 |
| * Language: C++ |
| * |
| * |
| * |
| * |
| ***************************************************************************** |
| */ |
| |
| #include "ex_stdh.h" |
| #include "ComTdb.h" |
| #include "ex_tcb.h" |
| #include "ex_tuple_flow.h" |
| #include "ex_exe_stmt_globals.h" |
| #include "ex_expr.h" |
| #include "str.h" |
| #include "ExStats.h" |
| #include "ExpError.h" |
| #include "cli_stdh.h" |
| |
| ///////////////////////////////////////////////////////////////////////// |
| // |
| // TDB procedures |
| // |
| ///////////////////////////////////////////////////////////////////////// |
| |
| ex_tcb * ExTupleFlowTdb::build(ex_globals * glob) |
| { |
| // first build the children |
| ex_tcb * src_tcb = tdbSrc_->build(glob); |
| ex_tcb * tgt_tcb = tdbTgt_->build(glob); |
| |
| ExTupleFlowTcb * tuple_flow_tcb = |
| new(glob->getSpace()) ExTupleFlowTcb(*this, |
| *src_tcb, *tgt_tcb, glob); |
| |
| // add the tuple_flow_tcb to the schedule |
| tuple_flow_tcb->registerSubtasks(); |
| |
| return (tuple_flow_tcb); |
| } |
| |
| |
| ///////////////////////////////////////////////////////////////////////// |
| // |
| // TCB procedures |
| // |
| ///////////////////////////////////////////////////////////////////////// |
| ExTupleFlowTcb::ExTupleFlowTcb(const ExTupleFlowTdb & tuple_flow_tdb, |
| const ex_tcb & src_tcb, // src queue pair |
| const ex_tcb & tgt_tcb, // tgt queue pair |
| ex_globals *glob |
| ) : ex_tcb( tuple_flow_tdb, 1, glob) |
| { |
| CollHeap * space = glob->getSpace(); |
| |
| // Allocate the buffer pool, if tgtExpr_ is present |
| if (tuple_flow_tdb.tgtExpr_) |
| pool_ = new(glob->getSpace()) sql_buffer_pool(tuple_flow_tdb.numBuffers_, |
| tuple_flow_tdb.bufferSize_, |
| glob->getSpace()); |
| |
| tcbSrc_ = &src_tcb; |
| tcbTgt_ = &tgt_tcb; |
| |
| // get the queues that src and tgt use to communicate with me |
| qSrc_ = src_tcb.getParentQueue(); |
| qTgt_ = tgt_tcb.getParentQueue(); |
| |
| ex_cri_desc * from_parent_cri = tuple_flow_tdb.criDescDown_; |
| ex_cri_desc * to_parent_cri = tuple_flow_tdb.criDescUp_; |
| |
| // Allocate the queue to communicate with parent |
| qParent_.down = new(space) ex_queue(ex_queue::DOWN_QUEUE, |
| tuple_flow_tdb.queueSizeDown_, |
| from_parent_cri, |
| space); |
| |
| // source atps will be copied to target atps. Allocate |
| // target atps. |
| qTgt_.down->allocateAtps(space); |
| |
| // Allocate the private state in each entry of the down queue |
| ExTupleFlowPrivateState *p = |
| new(space) ExTupleFlowPrivateState(this); |
| qParent_.down->allocatePstate(p, this); |
| delete p; |
| |
| qParent_.up = new(space) ex_queue(ex_queue::UP_QUEUE, |
| tuple_flow_tdb.queueSizeUp_, |
| to_parent_cri, |
| space); |
| |
| // fixup expressions |
| if (tflowTdb().tgtExpr_) |
| (void) tflowTdb().tgtExpr_->fixup(0, getExpressionMode(), this, |
| glob->getSpace(), glob->getDefaultHeap(), FALSE, glob); |
| |
| } |
| |
| ExTupleFlowTcb::~ExTupleFlowTcb() |
| { |
| delete qParent_.up; |
| delete qParent_.down; |
| freeResources(); |
| } |
| |
| void ExTupleFlowTcb::freeResources() |
| { |
| delete pool_; |
| pool_ = 0; |
| } |
| |
| short ExTupleFlowTcb::work() |
| { |
| // This is some sort of a hack to fix the problems with the number of rows |
| // inserted returned to the user for packed tables. For these tables, rows |
| // are packed (by the Pack node which is the left child of this tuple flow) |
| // before they are sent off to DP2. DP2 has no idea that it's actually |
| // inserting multiple logical rows (as perceived by the user). However, |
| // there is actually a hidden count of logical rows stored as the first 4 |
| // bytes of the packed row. This counter is supposed to keep track of a sum |
| // of this number in each packed row it gets from the left. When all |
| // insertions are done, this sum is used to override what's stored by the |
| // PA node in the executor global area the number of rows inserted. This is |
| // not a very good place to have this fix, but since this is a low-priority |
| // problem at this time, here we are. |
| // |
| // |
| // NB: The code introduced for this fix |
| // could be safely removed if desired. Also, all changes are within |
| // this file. |
| // |
| |
| if (qParent_.down->isEmpty()) |
| return WORK_OK; |
| |
| ex_queue_entry * pentry_down = qParent_.down->getHeadEntry(); |
| ExTupleFlowPrivateState & pstate = |
| *((ExTupleFlowPrivateState*) pentry_down->pstate); |
| |
| if ((tflowTdb().userSidetreeInsert()) && |
| (pentry_down->downState.request == ex_queue::GET_EOD) && |
| (NOT pstate.parentEOD_)) |
| { |
| pstate.step_ = MOVE_EOD_TO_TGT_; |
| } |
| else if ((pstate.step_ != DONE_) && |
| (pstate.step_ != CANCELLED_) && |
| (pentry_down->downState.request == ex_queue::GET_NOMORE)) |
| { |
| if (pstate.step_ == EMPTY_) |
| pstate.step_ = DONE_; |
| else |
| pstate.step_ = CANCELLED_; |
| } |
| |
| while (1) |
| { |
| switch (pstate.step_) |
| { |
| case EMPTY_: |
| { |
| if (qSrc_.down->isFull()) |
| return WORK_OK; |
| |
| ex_queue_entry * src_entry = qSrc_.down->getTailEntry(); |
| |
| src_entry->downState.request = pentry_down->downState.request; |
| src_entry->downState.requestValue = |
| pentry_down->downState.requestValue; |
| |
| if ((tflowTdb().firstNRows() >= 0) && |
| (pentry_down->downState.request != ex_queue::GET_N)) |
| { |
| src_entry->downState.request = ex_queue::GET_N; |
| src_entry->downState.requestValue = tflowTdb().firstNRows(); |
| } |
| |
| src_entry->downState.parentIndex = |
| qParent_.down->getHeadIndex(); |
| |
| src_entry->passAtp(pentry_down); |
| |
| qSrc_.down->insert(); |
| |
| // just checking to make sure we got a diags area from the CLI if we are |
| // executing a non-tomic insert. This is done now so that we don't have to do it in multiple |
| // places later. |
| if (tflowTdb().isNonFatalErrorTolerated()) { |
| ComDiagsArea *cliDiagsArea = pentry_down->getDiagsArea(); |
| ex_assert(cliDiagsArea, "In Tupleflow : Non-Atomic insert received no diags area from the CLI"); |
| } |
| |
| pstate.parentEOD_ = FALSE; |
| pstate.srcEOD_ = FALSE; |
| pstate.matchCount_ = 0; |
| pstate.tgtRequests_ = 0; |
| pstate.tgtRowsSent_ = FALSE; |
| pstate.noOfUnPackedRows_ = 0; |
| pstate.srcRequestCount_ = -1; |
| pstate.nonFatalErrorSeen_ = FALSE; |
| // Set startRightIndex_ so that CancelReques doesn't do anything. |
| pstate.startRightIndex_ = pstate.srcRequestCount_; |
| pstate.step_ = MOVE_SRC_TO_TGT_; |
| } |
| break; |
| |
| case MOVE_SRC_TO_TGT_: |
| { |
| // if there are some rows in source up queue, move them to target. |
| while ((! qSrc_.up->isEmpty()) && (! qTgt_.down->isFull()) |
| && (pstate.step_ != HANDLE_ERROR_)) |
| { |
| ex_queue_entry * src_entry = qSrc_.up->getHeadEntry(); |
| ex_queue_entry * tgt_entry = qTgt_.down->getTailEntry(); |
| |
| switch (src_entry->upState.status) |
| { |
| case ex_queue::Q_OK_MMORE: |
| { |
| // move this source row to target. |
| |
| // BEGIN: - Read note at beginning of work(). |
| // |
| if (tcbSrc_->getNodeType() == ComTdb::ex_PACKROWS) |
| { |
| char* packTuppPtr = |
| src_entry->getTupp(src_entry->numTuples()-1) |
| .getDataPointer(); |
| Int32 noOfRows = *((Int32 *)packTuppPtr); |
| pstate.noOfUnPackedRows_ += (noOfRows - 1); |
| |
| } |
| // |
| // END:- Read note at beginning of work(). |
| |
| pstate.srcRequestCount_++; |
| tgt_entry->downState.request = |
| pentry_down->downState.request; |
| tgt_entry->downState.requestValue = |
| pentry_down->downState.requestValue; |
| tgt_entry->downState.parentIndex = |
| (Lng32) pstate.srcRequestCount_; |
| tgt_entry->copyAtp(src_entry); |
| qTgt_.down->insert(); |
| pstate.tgtRequests_++; |
| pstate.tgtRowsSent_ = TRUE; |
| qSrc_.up->removeHead(); |
| } |
| break; |
| |
| case ex_queue::Q_NO_DATA: |
| { |
| if ((tflowTdb().vsbbInsertOn()) && |
| (pstate.tgtRowsSent_ == TRUE)) |
| { |
| if (tflowTdb().userSidetreeInsert()) |
| { |
| tgt_entry->downState.request = |
| ex_queue::GET_EOD_NO_ST_COMMIT; |
| } |
| else |
| { |
| tgt_entry->downState.request = |
| ex_queue::GET_EOD; |
| } |
| |
| tgt_entry->downState.requestValue = |
| pentry_down->downState.requestValue; |
| tgt_entry->downState.parentIndex = |
| (Lng32) pstate.srcRequestCount_; |
| tgt_entry->copyAtp(src_entry); |
| |
| qTgt_.down->insert(); |
| pstate.tgtRequests_++; |
| } |
| |
| if ((pstate.tgtRowsSent_ == FALSE) && |
| (src_entry->getDiagsArea())) |
| { |
| // a warning is returned with EOD and |
| // nothing else was returned from source. |
| // Move warning to parent's up queue. |
| if (qParent_.up->isFull()) |
| return WORK_OK; |
| |
| ex_queue_entry * up_entry = |
| qParent_.up->getTailEntry(); |
| up_entry->setDiagsArea(src_entry->getDiagsArea()); |
| } |
| |
| qSrc_.up->removeHead(); |
| |
| pstate.srcEOD_ = TRUE; |
| |
| if (tflowTdb().sendEODtoTgt()) |
| pstate.step_ = MOVE_EOD_TO_TGT_; |
| } |
| break; |
| |
| case ex_queue::Q_SQLERROR: |
| { |
| if (qParent_.up->isFull()) |
| return WORK_OK; |
| |
| ex_queue_entry * pentry = qParent_.up->getTailEntry(); |
| ComDiagsArea * da = src_entry->getDiagsArea(); |
| ex_assert(da, "We have a Q_SQLERROR in Tupleflow but no diags area"); |
| |
| if (tflowTdb().isNonFatalErrorTolerated() && |
| (da->getNextRowNumber(ComCondition::NONFATAL_ERROR) == |
| ComCondition::NONFATAL_ERROR)) |
| { |
| pstate.nonFatalErrorSeen_ = TRUE; |
| } |
| else |
| { |
| pstate.step_ = HANDLE_ERROR_; |
| pstate.nonFatalErrorSeen_ = FALSE; |
| } |
| |
| pstate.srcRequestCount_++; |
| if(tflowTdb().isRowsetIterator()) |
| da->setAllRowNumber((Lng32) pstate.srcRequestCount_); |
| |
| ComDiagsArea *accumulatedDiagsArea = pentry->getDiagsArea(); |
| if (accumulatedDiagsArea) |
| { |
| accumulatedDiagsArea->mergeAfter(*da); |
| if (!(accumulatedDiagsArea->canAcceptMoreErrors()) && |
| tflowTdb().isNonFatalErrorTolerated()) |
| { |
| pstate.nonFatalErrorSeen_ = FALSE; |
| pstate.step_ = HANDLE_ERROR_; |
| } |
| } |
| else |
| { |
| pentry->setDiagsArea(da); |
| da->incrRefCount(); |
| accumulatedDiagsArea = da ; |
| if (tflowTdb().isNonFatalErrorTolerated()) |
| { |
| ComDiagsArea *cliDiagsArea = pentry_down->getDiagsArea(); |
| da->setLengthLimit(cliDiagsArea->getLengthLimit()); |
| } |
| } |
| |
| // For Non-Fatal errors we will remove this Q_SQLERROR reply from the |
| // left child right below as we will continue to stay in this state (MOVE_SRC_TO_TGT_). |
| // For fatal errors this Q_SQLERROR reply is removed in HANDLE_ERROR step to which |
| // we will transition immediately. |
| if (pstate.nonFatalErrorSeen_ == TRUE) |
| qSrc_.up->removeHead(); |
| } |
| break; |
| |
| case ex_queue::Q_REC_SKIPPED: |
| { |
| pstate.srcRequestCount_++; |
| ComDiagsArea * da = src_entry->getDiagsArea(); |
| if (da) |
| pstate.nonFatalErrorSeen_ = TRUE; |
| qSrc_.up->removeHead(); |
| } |
| break; |
| |
| default: |
| { |
| ex_assert(0, "ExTupleFlowTcb::work() Error returned from src"); |
| } |
| break; |
| } // switch |
| |
| } // while |
| |
| // if the child reply is not an Q_SQLERROR, then process target |
| if ((pstate.step_ != HANDLE_ERROR_) && |
| (pstate.step_ != MOVE_EOD_TO_TGT_)) |
| pstate.step_ = PROCESS_TGT_; |
| } // MOVE_SRC_TO_TGT |
| break; |
| |
| case MOVE_EOD_TO_TGT_: |
| { |
| pstate.parentEOD_ = TRUE; |
| |
| if (qTgt_.down->isFull()) |
| return WORK_OK; |
| |
| ex_queue_entry * tgt_entry = qTgt_.down->getTailEntry(); |
| tgt_entry->downState.request = ex_queue::GET_EOD; |
| |
| tgt_entry->downState.requestValue = |
| pentry_down->downState.requestValue; |
| tgt_entry->downState.parentIndex = |
| qParent_.down->getHeadIndex(); |
| //tgt_entry->passAtp(pentry_down); |
| |
| qTgt_.down->insert(); |
| |
| pstate.tgtRequests_++; |
| |
| if (tflowTdb().sendEODtoTgt()) |
| pstate.srcEOD_ = TRUE; |
| else |
| pstate.srcEOD_ = FALSE; |
| |
| pstate.step_ = PROCESS_TGT_; |
| } |
| break; |
| |
| case PROCESS_TGT_: |
| { |
| while (! qTgt_.up->isEmpty() && pstate.step_ != HANDLE_ERROR_) |
| { |
| ex_queue_entry * tgt_entry = qTgt_.up->getHeadEntry(); |
| switch (tgt_entry->upState.status) |
| { |
| case ex_queue::Q_OK_MMORE: |
| { |
| if (!tflowTdb().isNonFatalErrorTolerated()) |
| { |
| // ex_assert(0, "ExTupleFlowTcb::work() OK_MMORE from tgt"); |
| if (qParent_.up->isFull()) |
| return WORK_OK; |
| |
| ex_queue_entry * pentry = qParent_.up->getTailEntry(); |
| |
| pentry->upState.status = ex_queue::Q_OK_MMORE; |
| pentry->upState.downIndex = qParent_.down->getHeadIndex(); |
| pentry->upState.parentIndex = pentry_down->downState.parentIndex; |
| pentry->upState.setMatchNo(pstate.matchCount_); |
| |
| // copy input tupps from parent request |
| pentry->copyAtp(pentry_down); |
| |
| // copy child's atp to |
| // the output atp (to parent's up queue) |
| pentry->copyAtp(tgt_entry); |
| |
| // insert into parent up queue |
| qParent_.up->insert(); |
| } |
| else |
| { |
| ComDiagsArea * da = tgt_entry->getDiagsArea(); |
| ex_assert(da, "We have a Q_OK_MMORE in Tupleflow but no diags area"); |
| if (da->mainSQLCODE() != 0) { |
| // Non-atomic Rowsets sends OK_MMORE with non-empty diags from child |
| // empty diags (mainsqlcode == 0) implies OK_MMORE sent by ignoreDupKey code |
| // when NAR is on, for -8102 error. Just consume the OK_MMORE. |
| |
| if(tflowTdb().isRowsetIterator()) |
| { |
| da->setAllRowNumber(Lng32 (tgt_entry->upState.parentIndex)); |
| } |
| |
| pstate.nonFatalErrorSeen_ = TRUE; |
| ex_queue_entry * pentry = qParent_.up->getTailEntry(); |
| ComDiagsArea *accumulatedDiagsArea = pentry->getDiagsArea(); |
| if (accumulatedDiagsArea) |
| { |
| accumulatedDiagsArea->mergeAfter(*da); |
| if (!(accumulatedDiagsArea->canAcceptMoreErrors()) && |
| tflowTdb().isNonFatalErrorTolerated()) |
| { |
| pstate.nonFatalErrorSeen_ = FALSE; |
| pstate.step_ = HANDLE_ERROR_; |
| } |
| } |
| else |
| { |
| pentry->setDiagsArea(da); |
| da->incrRefCount(); |
| if (tflowTdb().isNonFatalErrorTolerated()) { |
| ComDiagsArea *cliDiagsArea = pentry_down->getDiagsArea(); |
| da->setLengthLimit(cliDiagsArea->getLengthLimit()); |
| } |
| } |
| } |
| } |
| |
| qTgt_.up->removeHead(); |
| } |
| break; |
| |
| case ex_queue::Q_NO_DATA: |
| { |
| ComDiagsArea * da = tgt_entry->getDiagsArea(); |
| if (da) |
| { |
| ex_queue_entry * pentry = qParent_.up->getTailEntry(); |
| ComDiagsArea *accumulatedDiagsArea = pentry->getDiagsArea(); |
| if (accumulatedDiagsArea) |
| accumulatedDiagsArea->mergeAfter(*da); |
| else |
| { |
| pentry->setDiagsArea(da); |
| da->incrRefCount(); |
| if (tflowTdb().isNonFatalErrorTolerated()) { |
| ComDiagsArea *cliDiagsArea = pentry_down->getDiagsArea(); |
| da->setLengthLimit(cliDiagsArea->getLengthLimit()); |
| } |
| } |
| } |
| pstate.matchCount_ += tgt_entry->upState.getMatchNo(); |
| qTgt_.up->removeHead(); |
| pstate.tgtRequests_--; |
| pstate.startRightIndex_++; |
| } |
| break; |
| |
| case ex_queue::Q_SQLERROR: |
| { |
| if (qParent_.up->isFull()) |
| return WORK_OK; |
| |
| ex_queue_entry * pentry = qParent_.up->getTailEntry(); |
| pentry->copyAtp(tgt_entry); |
| pstate.nonFatalErrorSeen_ = FALSE; |
| pstate.step_ = HANDLE_ERROR_; |
| |
| if(tflowTdb().isRowsetIterator()) |
| { |
| ex_queue_entry * pentry = qParent_.up->getTailEntry(); |
| ComDiagsArea *da = pentry->getDiagsArea(); |
| ex_assert(da, "To set RowNumber, an error condition must be present in the diags area"); |
| da->setAllRowNumber(Lng32 (tgt_entry->upState.parentIndex)); |
| } |
| |
| } |
| break; |
| |
| default: |
| { |
| ex_assert(0, "ExTupleFlowTcb::work() Error returned from tgt"); |
| } |
| break; |
| |
| } // switch |
| |
| } // while |
| |
| if (pstate.step_ == HANDLE_ERROR_) |
| break; |
| |
| // if source has returned EOD, |
| // and there are no pending requests in target's down |
| // queue, then we are done with this parent request. |
| if (((pstate.srcEOD_ == TRUE) || |
| (pstate.parentEOD_ == TRUE)) && |
| (qTgt_.down->isEmpty())) |
| pstate.step_ = DONE_; |
| else |
| { |
| if (NOT pstate.parentEOD_) |
| pstate.step_ = MOVE_SRC_TO_TGT_; |
| |
| if (qSrc_.up->isEmpty() || qTgt_.down->isFull()) |
| return WORK_OK; |
| else |
| return WORK_CALL_AGAIN; |
| } |
| } |
| break; |
| |
| case HANDLE_ERROR_: |
| { |
| ex_queue_entry * pentry = qParent_.up->getTailEntry(); |
| |
| pentry->upState.status = ex_queue::Q_SQLERROR; |
| pentry->upState.downIndex = qParent_.down->getHeadIndex(); |
| pentry->upState.parentIndex = pentry_down->downState.parentIndex; |
| pentry->upState.setMatchNo(pstate.matchCount_); |
| |
| ComDiagsArea *da = pentry->getDiagsArea(); |
| if (tflowTdb().isNonFatalErrorTolerated() && |
| !(da->canAcceptMoreErrors())) { |
| ComDiagsArea *cliDiagsArea = pentry_down->getDiagsArea(); |
| da->removeLastErrorCondition(); |
| *da << DgSqlCode(-EXE_NONATOMIC_FAILURE_LIMIT_EXCEEDED) |
| << DgInt0(cliDiagsArea->getLengthLimit()); |
| } |
| |
| // insert into parent up queue |
| qParent_.up->insert(); |
| |
| pstate.step_ = CANCELLED_; |
| } |
| break; |
| |
| case CANCELLED_: |
| { |
| qSrc_.down->cancelRequestWithParentIndex(qParent_.down->getHeadIndex()); |
| |
| // Cancel all the outstanding requests that have been sent to the target. |
| // Cancel all requests within given range (inclusive) |
| qTgt_.down->cancelRequestWithParentIndexRange((queue_index)pstate.startRightIndex_+1, |
| (queue_index)pstate.srcRequestCount_); |
| pstate.startRightIndex_ = pstate.srcRequestCount_; |
| |
| //ignore all rows from source child, till Q_NO_DATA is reached |
| while ((pstate.srcEOD_ != TRUE) && (!qSrc_.up->isEmpty())) |
| { |
| ex_queue_entry * src_entry = qSrc_.up->getHeadEntry(); |
| switch(src_entry->upState.status) |
| { |
| case ex_queue::Q_OK_MMORE: |
| case ex_queue::Q_SQLERROR: |
| case ex_queue::Q_REC_SKIPPED: |
| { |
| qSrc_.up->removeHead(); |
| } |
| break; |
| |
| case ex_queue::Q_NO_DATA: |
| { |
| pstate.srcEOD_ = TRUE; |
| qSrc_.up->removeHead(); |
| } |
| break; |
| |
| default: |
| { |
| ex_assert(0, "ExTupleFlowTcb::work() Error returned from src"); |
| } |
| break; |
| } |
| } |
| |
| //ignore all rows from target child, till Q_NO_DATA is reached |
| while (pstate.tgtRequests_ && !qTgt_.up->isEmpty()) |
| { |
| ex_queue_entry * tgt_entry = qTgt_.up->getHeadEntry(); |
| switch(tgt_entry->upState.status) |
| { |
| case ex_queue::Q_OK_MMORE: |
| case ex_queue::Q_SQLERROR: |
| { |
| qTgt_.up->removeHead(); |
| } |
| break; |
| |
| case ex_queue::Q_NO_DATA: |
| { |
| qTgt_.up->removeHead(); |
| pstate.tgtRequests_--; |
| } |
| break; |
| |
| default: |
| { |
| ex_assert(0, "ExTupleFlowTcb::work() Error returned from tgt"); |
| } |
| break; |
| } |
| } |
| |
| // if both source and target returned all the rows, |
| // insert Q_SQLERROR into the parent up queue |
| if ((pstate.srcEOD_ == TRUE) && !pstate.tgtRequests_) |
| { |
| pstate.step_ = DONE_; |
| } |
| else |
| return WORK_OK; |
| } |
| break; |
| |
| case DONE_: |
| { |
| if (qParent_.up->isFull()) |
| return WORK_OK; |
| |
| ex_queue_entry * pentry = qParent_.up->getTailEntry(); |
| |
| if (pstate.nonFatalErrorSeen_) { |
| ComDiagsArea *da = pentry->getDiagsArea(); |
| ComDiagsArea *cliDiagsArea = pentry_down->getDiagsArea(); |
| ex_assert((da || cliDiagsArea), "We have non-fatal errors in Tupleflow but no diags area"); |
| if (cliDiagsArea) { |
| if (da) |
| da->mergeAfter(*cliDiagsArea); |
| else |
| { |
| pentry->setDiagsArea(cliDiagsArea); |
| cliDiagsArea->incrRefCount(); |
| } |
| } |
| |
| if (cliDiagsArea->canAcceptMoreErrors()) { |
| ComDiagsArea *mergedDiagsArea = pentry->getDiagsArea(); |
| // used to make mainSQLCODE() return 30022 or 30035. |
| mergedDiagsArea->setNonFatalErrorSeen(TRUE); |
| NABoolean anyRowsAffected = FALSE; |
| |
| |
| // This tupleflow should be in the master for |
| // non-atomic rowsets. |
| ExMasterStmtGlobals *g = getGlobals()-> |
| castToExExeStmtGlobals()->castToExMasterStmtGlobals(); |
| ex_assert(g, "Rowset insert has a flow node that is not in the master executor"); |
| if (g->getRowsAffected() > 0) |
| anyRowsAffected = TRUE; |
| |
| if (anyRowsAffected) |
| *mergedDiagsArea << DgSqlCode(EXE_NONFATAL_ERROR_SEEN); |
| else |
| *mergedDiagsArea << DgSqlCode(EXE_NONFATAL_ERROR_ON_ALL_ROWS); |
| |
| } // we exceeded the Nonfatal error limit when merging with the CLI diags area |
| else { |
| pstate.step_ = HANDLE_ERROR_; |
| // will prevent us from merging the diags areas again |
| pstate.nonFatalErrorSeen_ = FALSE ; |
| break ; |
| } |
| } |
| |
| pentry->upState.status = ex_queue::Q_NO_DATA; |
| pentry->upState.downIndex = qParent_.down->getHeadIndex(); |
| pentry->upState.parentIndex = pentry_down->downState.parentIndex; |
| pentry->upState.setMatchNo(pstate.matchCount_); |
| |
| // BEGIN: Read note at beginning of work(). |
| // |
| if(pstate.noOfUnPackedRows_ != 0) |
| { |
| ComDiagsArea *da = pentry->getDiagsArea(); |
| if (da == NULL) |
| { |
| da = ComDiagsArea::allocate(getGlobals()->getDefaultHeap()); |
| pentry->setDiagsArea(da); |
| } |
| da->addRowCount(pstate.noOfUnPackedRows_); |
| pstate.noOfUnPackedRows_ = 0; |
| } |
| // |
| // END: - Read note at beginning of work(). |
| |
| // if stats are to be collected, collect them. |
| if (getStatsEntry()) |
| { |
| // nothing yet returned from right child or returned |
| // to parent. |
| getStatsEntry()->setActualRowsReturned(0); |
| } |
| |
| // insert into parent up queue |
| qParent_.up->insert(); |
| |
| pstate.step_ = EMPTY_; |
| qParent_.down->removeHead(); |
| |
| return WORK_CALL_AGAIN; // check for more down requests |
| |
| } |
| break; |
| |
| } // switch pstate.step_ |
| } // while |
| |
| return 0; |
| } |
| |
| |
| ///////////////////////////////////////////////////////////////////////// |
| // |
| // Private state procedures |
| // |
| ///////////////////////////////////////////////////////////////////////// |
| |
| ExTupleFlowPrivateState::ExTupleFlowPrivateState(const ExTupleFlowTcb * /*tcb*/) |
| { |
| init(); |
| } |
| |
| void ExTupleFlowPrivateState::init() |
| { |
| matchCount_ = 0; |
| tgtRequests_ = 0; |
| tgtRowsSent_ = FALSE; |
| srcEOD_ = FALSE; |
| parentEOD_ = FALSE; |
| noOfUnPackedRows_ = 0; |
| step_ = ExTupleFlowTcb::EMPTY_; |
| srcRequestCount_ = -1; |
| nonFatalErrorSeen_ = FALSE; |
| // parent index for target Q entries are eqaul to value of srcRequestCount_. |
| // therefore startRightIndex_ is initialized to the same value as srcRequestCount_ |
| startRightIndex_ = -1; |
| } |
| |
| ExTupleFlowPrivateState::~ExTupleFlowPrivateState() |
| { |
| } |
| |
| ex_tcb_private_state * ExTupleFlowPrivateState::allocate_new(const ex_tcb *tcb) |
| { |
| return new(((ex_tcb *)tcb)->getSpace()) ExTupleFlowPrivateState((ExTupleFlowTcb *) tcb); |
| } |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
