core/sql/executor/ExPack.cpp - trafodion - Git at Google

 /**********************************************************************
 // @@@ 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:         ExPack.cpp
 * Description:  Methods for ExPackTdb and ExPackTcb
 *
 * Created:      6/16/97
 * Language:     C++
 *
 *
 *
 *
 ******************************************************************************
 */

 #include "ExPack.h"
 #include "ExSimpleSqlBuffer.h"
 #include "ex_exe_stmt_globals.h"
 #include "ComDefs.h"

 //////////////////////////////////////////////////////////////////////////////
 //
 //  TDB procedures
 //
 //////////////////////////////////////////////////////////////////////////////


 // Build the TCB tree for this Pack node and all of its descendents.
 ex_tcb* ExPackRowsTdb::build(ex_globals* glob)
 {
   ex_tcb* childTcb = childTdb_->build(glob);
   ExPackRowsTcb* packTcb = new(glob->getSpace()) ExPackRowsTcb(*this,*childTcb,glob);
   packTcb->registerSubtasks();
   return packTcb;
 }


 //////////////////////////////////////////////////////////////////////////////
 //
 //  TCB procedures
 //
 //////////////////////////////////////////////////////////////////////////////

 // Constructor.
 ExPackRowsTcb::ExPackRowsTcb(const ExPackRowsTdb& packTdb,
                      const ex_tcb& childTcb,
                      ex_globals* glob)
  : ex_tcb(packTdb,1,glob)
 {
   childTcb_ = &childTcb;
   Space* space = (glob ? glob->getSpace() : 0);
   CollHeap* heap = (glob ? glob->getDefaultHeap() : 0);

   // Allocate the buffer pool.
   pool_ = new (space) ExSimpleSQLBuffer(packTdb.noOfBuffers_,
 					packTdb.packTuppLen_,
 					space);

   // Get the queue used by my child to communicate with me.
   qChild_ = childTcb_->getParentQueue();

   // Allocate an extra input tupp the Pack node adds to its down queues.
   qChild_.down->allocateAtps(glob->getSpace());

   // Allocate the queue to communicate with parent.
   qParent_.down = new (space) ex_queue(ex_queue::DOWN_QUEUE,
                                        packTdb.fromParent_,
                                        packTdb.givenCriDesc_,
                                        space);

   // Allocate the private state in each entry of the down queue.
   ExPackPrivateState privateState(this);
   qParent_.down->allocatePstate(&privateState,this);

   // Initialize nextRqst_ to refer to the queue entry used next.
   nextRqst_ = qParent_.down->getHeadIndex();

   // Allocate a queue to communicate with the parent node.
   qParent_.up = new(space) ex_queue(ex_queue::UP_QUEUE,
                                     packTdb.toParent_,
                                     packTdb.returnedCriDesc_,
                                     space);
   // Fix up the expressions.
   packExpr()->fixup(0,getExpressionMode(),this,space,heap, FALSE, glob);
   if(predExpr()) predExpr()->fixup(0,getExpressionMode(),this,space,heap, FALSE, glob);
 }

 // Destructor.
 ExPackRowsTcb::~ExPackRowsTcb()
 {
   delete qParent_.up;
   delete qParent_.down;
   freeResources();
 }

 void ExPackRowsTcb::freeResources()
 {
   if(pool_) delete pool_;
   pool_ = 0;
 }

 // -----------------------------------------------------------------------
 // Register all the pack subtasks with the scheduler.
 // -----------------------------------------------------------------------
 void ExPackRowsTcb::registerSubtasks()
 {
   ExScheduler* sched = getGlobals()->getScheduler();

   // Schedule this routine if a new entry is inserted into the parent's
   // down queue and we are waiting on an event.
   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.
   sched->registerUnblockSubtask(sWorkDown,this,qChild_.down,"DN");

   // Schedule this routine if a cancel request occurs on the parent's
   // down queue.
   sched->registerCancelSubtask(sCancel,this,qParent_.down,"CN");

   // Schedule this routine if the parent's up queue changes from being
   // full to being not full and we are waiting on an event.
   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.
   sched->registerInsertSubtask(sWorkUp,this,qChild_.up);

   // BertBert VV
   // Schedule this routine if a GET_NEXT_N request is updated in the parent's
   // down queue and we are waiting on an event.
   sched->registerNextSubtask(sWorkDown,this,qParent_.down,"DN");
   // BertBert ^^
 }

 // -----------------------------------------------------------------------
 // Send next request down to children. Called by workDown().
 // -----------------------------------------------------------------------
 void ExPackRowsTcb::start()
 {
   // Caller needs to make sure we have room in the child's down queue.
   ex_queue_entry* cEntryDown = qChild_.down->getTailEntry();

   // Caller needs to make sure we have a new request in parent's down queue.
   ex_queue_entry* pEntryDown = qParent_.down->getQueueEntry(nextRqst_);

   const ex_queue::down_request request = pEntryDown->downState.request;
   ExPackPrivateState& pState = *((ExPackPrivateState*)pEntryDown->pstate);

   // don't forward cancelled request.
   // if(request == ex_queue::GET_NOMORE)
   // {
   //   pState.childState_ = CANCELLED_;
   //   nextRqst_++;
   //   return;
   // }

   // Pass request to children.
   cEntryDown->downState.request = request;
   cEntryDown->downState.requestValue = pEntryDown->downState.requestValue;
   // BertBert VV
   // Note, there is no complete support for the GET_NEXT_N protocol in this operator.
   cEntryDown->downState.numGetNextsIssued = pEntryDown->downState.numGetNextsIssued;
   // BertBert ^^
   cEntryDown->downState.parentIndex = nextRqst_;
   cEntryDown->copyAtp(pEntryDown->getAtp());
   pState.matchCount_ = 0;
   qChild_.down->insert();
   pState.childState_ = STARTED_;

   // If this is a CANCEL request, cancel 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)
     qChild_.down->cancelRequestWithParentIndex(nextRqst_);
     pState.childState_ = CANCELLED_;
   }
   nextRqst_++;
 }

 // -----------------------------------------------------------------------
 // All child rows have been returned. Return an EOD indication to parent.
 // -----------------------------------------------------------------------
 void ExPackRowsTcb::stop()
 {
   // Caller needs to make sure there is an outstanding parent request.
   ex_queue_entry* pEntryDown = qParent_.down->getHeadEntry();
   ExPackPrivateState& pState = *((ExPackPrivateState*)pEntryDown->pstate);

   // Caller needs to make sure we have room in the parent's up queue.
   ex_queue_entry* pEntryUp = qParent_.up->getTailEntry();

   // Insert EOD to parent's up queue.
   pEntryUp->upState.status = ex_queue::Q_NO_DATA;
   pEntryUp->upState.parentIndex = pEntryDown->downState.parentIndex;
   qParent_.up->insert();

   // child row removed outside of stop().
   // Consume the child row.
   qChild_.up->removeHead();

   // Reinitialize the state of this processed request.
   pState.init();

   // This parent request has been processed.
   qParent_.down->removeHead();
 }

 // -----------------------------------------------------------------------
 // For those requests my parent cancel, cancel the corresponding requests
 // that have already been forwarded to my child.
 // -----------------------------------------------------------------------
 ExWorkProcRetcode ExPackRowsTcb::processCancel()
 {
   queue_index ix = qParent_.down->getHeadIndex();

   // Loop over all requests that have been sent down.
   while(ix != nextRqst_)
   {
     ex_queue_entry* pEntryDown = qParent_.down->getQueueEntry(ix);

     // Check whether this is a request cancelled by the parent.
     if(pEntryDown->downState.request == ex_queue::GET_NOMORE)
     {
       ExPackPrivateState& pState =
                                *((ExPackPrivateState*)pEntryDown->pstate);

       // Cancel corresponding request for my child if it's been forwarded.
       if(pState.childState_ == ExPackRowsTcb::STARTED_)
       {
 	qChild_.down->cancelRequestWithParentIndex(ix);
 	pState.childState_ = ExPackRowsTcb::CANCELLED_;
       }
     }
     ix++;
   }
   return WORK_OK;
 }

 // -----------------------------------------------------------------------
 // Generic work procedure should never be called
 // -----------------------------------------------------------------------
 ExWorkProcRetcode ExPackRowsTcb::work()
 {
   ex_assert(0,"Should never reach ExPackRowsTcb::work()");
   return WORK_BAD_ERROR;
 }

 // -----------------------------------------------------------------------
 // Work procedure to send requests down
 // -----------------------------------------------------------------------
 ExWorkProcRetcode ExPackRowsTcb::workDown()
 {
   // While we have unprocessed down requests and the child's down queue
   // has room, start more child requests.
   while(
      qParent_.down->entryExists(nextRqst_) && !qChild_.down->isFull())
      start();

   return WORK_OK;
 }

 // -----------------------------------------------------------------------
 // Work procedure to process child rows.
 // -----------------------------------------------------------------------
 ExWorkProcRetcode ExPackRowsTcb::workUp()
 {
   // While there is an outstanding parent request not finished.
   while(qParent_.down->getHeadIndex() != nextRqst_)
   {
     // Get the request and retrieve its private state.
     ex_queue_entry* pEntryDown = qParent_.down->getHeadEntry();
     ExPackPrivateState& pState = *((ExPackPrivateState*)pEntryDown->pstate);

     // While we have room in the up queue and rows to process.
     while(!qParent_.up->isFull() && !qChild_.up->isEmpty())
     {
       // New row produced by child.
       ex_queue_entry* cEntryUp = qChild_.up->getHeadEntry();

       // The current request has already been cancelled.
       if(pState.childState_ == ExPackRowsTcb::CANCELLED_)
       {
         if(cEntryUp->upState.status == ex_queue::Q_NO_DATA)
         {
           // No more rows for the cancelled request.
           // stop() returns Q_NO_DATA and remove the cancelled request.
           stop();

           // Break out of the inner while-loop and check for next request.
           break;
         }
         else
         {
           // Child is returning data for a request already cancelled.
           // Just ignore them and continue to check whether child's up
           // ComQueue.has any more entries.
           qChild_.up->removeHead();
           continue;
         }
       }

       if(cEntryUp->upState.status == ex_queue::Q_NO_DATA)
       {
         // Apply selection predicates to and return row being partially
         // packed in buffer if there is one.
         workReturnRow();

 	// Send EOD to parent and clean up the current request.
 	stop();

 	// Break out of the inner while-loop and check for next request.
 	break;
       }
       else if(cEntryUp->upState.status == ex_queue::Q_SQLERROR)
       {
         // Child has produced an error. Copy diagnostic area to parent.
         ex_queue_entry* pEntryUp = qParent_.up->getTailEntry();
         pEntryUp->copyAtp(cEntryUp);
         pEntryUp->upState.status = ex_queue::Q_SQLERROR;
         pEntryUp->upState.parentIndex = pEntryDown->downState.parentIndex;
         pEntryUp->upState.downIndex = qParent_.down->getHeadIndex();
         pEntryUp->upState.setMatchNo(pState.matchCount_); // (+1?)
         qParent_.up->insert();

         // Cancel the whole request on child. (may be optional)
         qChild_.down->cancelRequestWithParentIndex(
                                            qParent_.down->getHeadIndex());
         // Mark the corresponding request cancelled.
         pState.childState_ = ExPackRowsTcb::CANCELLED_;
         qChild_.up->removeHead();
         continue;
       }

       //
       // Now, we got a new row to be packed into the current packed row.
       // Here's the core stuff.

       // Check if the packTupp has been allocated. If not, allocate it.
       if(pState.packTupp_.getDataPointer() == NULL)
       {
         // Cannot allocate space. Call again when some space free up.
         if(pool_->getFreeTuple(pState.packTupp_))
           return WORK_POOL_BLOCKED;

         // A cheat to initialize contents of the packed row. We need this
         // since the logic of packExpr evaluation relies on the fact that
         // the no of rows already packed in a packed row (which is 0 to
         // begin with) is stored in the packed row itself, even initially.
         //
         char* packTuppPtr = pState.packTupp_.getDataPointer();
         for(Int32 i=0; i < packTdb().packTuppLen_; i++) packTuppPtr[i] = 0;

         // $$$ could potentially make it faster by initializing only the
         // $$$ first four bytes (which stores the no of rows) to 0.
       }

       // Put the pack tupp directly in child's returned ATP.
       atp_struct * cEntryUpAtp = cEntryUp->getAtp();
       cEntryUpAtp->getTupp(packTdb().packTuppIndex_) = pState.packTupp_;

       // Apply the pack expression to pack child row into packed row.
       ex_expr::exp_return_type retCode = packExpr()->eval(cEntryUpAtp,0);
       if(!cEntryUpAtp->getDiagsArea());
       else if(!pEntryDown->getDiagsArea())
 	{
 	  pEntryDown->setDiagsArea(cEntryUpAtp->getDiagsArea());
 	  pEntryDown->getDiagsArea()->incrRefCount();
 	}
       else
 	pEntryDown->getDiagsArea()->mergeAfter(*cEntryUpAtp->getDiagsArea());

       if(retCode == ex_expr::EXPR_TRUE)
       {
         // for debugging.
         // pState.printPackTupp();

         // eval() returns TRUE if the packed record is full after the
         // child row is packed into it. Return the full packed record.

          workReturnRow();
       }
       else
       {
         // for debugging.
         // pState.printPackTupp();

         // Error occurs during expression evaluation.
         if(retCode == ex_expr::EXPR_ERROR)
         {
           // There must be room in the up queue to have reached here.
           ex_queue_entry* pEntryUp = qParent_.up->getTailEntry();

           // cEntryUp's atp has diagnostic area set up. copy it to parent.
           pEntryUp->copyAtp(cEntryUp);
           pEntryUp->upState.status = ex_queue::Q_SQLERROR;
           pEntryUp->upState.downIndex = qParent_.down->getHeadIndex();
           pEntryUp->upState.parentIndex = pEntryDown->downState.parentIndex;
           pEntryUp->upState.setMatchNo(pState.matchCount_); //(+1?)
           qParent_.up->insert();

           // mark this request cancelled in the down queue to child.
           qChild_.down->cancelRequestWithParentIndex(
                                            qParent_.down->getHeadIndex());
           pState.childState_ = ExPackRowsTcb::CANCELLED_;
         }
       }

       // Child row is no longer useful after the eval, get rid of it.
       qChild_.up->removeHead();

     }  // while parent's up queue is not full and child's up queue not empty.

     // If we got here because we finished a request and stop()ed, then
     // try again on a new request. If we got here because the parent
     // queue is full or the child queue is empty, then we must return.
     //
     if(qParent_.up->isFull() || qChild_.up->isEmpty()) {
       return WORK_OK;
     }

   } // while parent's down ComQueue.has entries.

   // Parent down queue is empty.
   return WORK_OK;

 }

 // -----------------------------------------------------------------------
 // Work procedure to apply the selection predicates on the current packed
 // record and return it.
 // -----------------------------------------------------------------------
 ExWorkProcRetcode ExPackRowsTcb::workReturnRow()
 {
   // Caller needs to make sure there is an outerstanding request.
   ex_queue_entry* pEntryDown = qParent_.down->getHeadEntry();
   ExPackPrivateState& pState = *((ExPackPrivateState*)pEntryDown->pstate);

   // Nothing in packed record. Nothing to return.
   if(pState.packTupp_.getDataPointer() == NULL) return WORK_OK;

   // Copy parent's ATP from down queue to its up queue and add last tupp.
   ex_queue_entry* pEntryUp = qParent_.up->getTailEntry();
   atp_struct * pEntryUpAtp = pEntryUp->getAtp();
   pEntryUpAtp->copyAtp(pEntryDown->getAtp());
   pEntryUpAtp->getTupp(packTdb().packTuppIndex_) = pState.packTupp_;

   // Release the tupp in the private state so that it stops referencing
   // the packed record.
   pState.packTupp_.release();

   // Evaluate the selection predicates.
   ex_expr::exp_return_type predVal = ex_expr::EXPR_TRUE;
   if(predExpr()) predVal = predExpr()->eval(pEntryUpAtp,0);
   if(predVal)
   {
     // Formally insert the ATP into the parent's up queue.
     pEntryUp->upState.status = ex_queue::Q_OK_MMORE;
     pEntryUp->upState.parentIndex = pEntryDown->downState.parentIndex;
     pEntryUp->upState.setMatchNo(pState.matchCount_ + 1);
     pState.matchCount_++;
     qParent_.up->insert();
   }
   else
   {
     // Release the ATP formed in the parent's up queue entry.
     pEntryUpAtp->release();
   }
   return WORK_OK;
 }

 //////////////////////////////////////////////////////////////////////////////
 //
 //  Private state procedures
 //
 //////////////////////////////////////////////////////////////////////////////

 // Constructor.
 ExPackPrivateState::ExPackPrivateState(const ExPackRowsTcb *)
 {
   packTupp_.init();
   init();
 }

 // Destructor.
 ExPackPrivateState::~ExPackPrivateState()
 {}

 // Initialization.
 void ExPackPrivateState::init()
 {
   matchCount_ = 0;
   childState_ = ExPackRowsTcb::EMPTY_;
   packTupp_.release();
 }

 // Allocation.
 ex_tcb_private_state* ExPackPrivateState::allocate_new(const ex_tcb *tcb)
 {
   return new(((ex_tcb *)tcb)->getSpace()) ExPackPrivateState((ExPackRowsTcb *)tcb);
 }

 void ExPackPrivateState::printPackTupp()
 {
   /*
   char* env = getenv("DEBUG_PACKING");
   if(env != NULL)
   {
     char* data = packTupp_.getDataPointer();
     printf("Pack tupp in HEX:\n");
     if(data == NULL) return;
     int len = packTupp_.getAllocatedSize();
     for(int i=0;i<len;i++) printf("%x ",data[i]);
     printf("\n");
   }
   */
 }
	/**********************************************************************
	// @@@ 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: ExPack.cpp
	* Description: Methods for ExPackTdb and ExPackTcb
	*
	* Created: 6/16/97
	* Language: C++
	*
	*
	*
	*
	******************************************************************************
	*/

	#include "ExPack.h"
	#include "ExSimpleSqlBuffer.h"
	#include "ex_exe_stmt_globals.h"
	#include "ComDefs.h"

	//////////////////////////////////////////////////////////////////////////////
	//
	// TDB procedures
	//
	//////////////////////////////////////////////////////////////////////////////


	// Build the TCB tree for this Pack node and all of its descendents.
	ex_tcb* ExPackRowsTdb::build(ex_globals* glob)
	{
	ex_tcb* childTcb = childTdb_->build(glob);
	ExPackRowsTcb* packTcb = new(glob->getSpace()) ExPackRowsTcb(this,childTcb,glob);
	packTcb->registerSubtasks();
	return packTcb;
	}


	//////////////////////////////////////////////////////////////////////////////
	//
	// TCB procedures
	//
	//////////////////////////////////////////////////////////////////////////////

	// Constructor.
	ExPackRowsTcb::ExPackRowsTcb(const ExPackRowsTdb& packTdb,
	const ex_tcb& childTcb,
	ex_globals* glob)
	: ex_tcb(packTdb,1,glob)
	{
	childTcb_ = &childTcb;
	Space* space = (glob ? glob->getSpace() : 0);
	CollHeap* heap = (glob ? glob->getDefaultHeap() : 0);

	// Allocate the buffer pool.
	pool_ = new (space) ExSimpleSQLBuffer(packTdb.noOfBuffers_,
	packTdb.packTuppLen_,
	space);

	// Get the queue used by my child to communicate with me.
	qChild_ = childTcb_->getParentQueue();

	// Allocate an extra input tupp the Pack node adds to its down queues.
	qChild_.down->allocateAtps(glob->getSpace());

	// Allocate the queue to communicate with parent.
	qParent_.down = new (space) ex_queue(ex_queue::DOWN_QUEUE,
	packTdb.fromParent_,
	packTdb.givenCriDesc_,
	space);

	// Allocate the private state in each entry of the down queue.
	ExPackPrivateState privateState(this);
	qParent_.down->allocatePstate(&privateState,this);

	// Initialize nextRqst_ to refer to the queue entry used next.
	nextRqst_ = qParent_.down->getHeadIndex();

	// Allocate a queue to communicate with the parent node.
	qParent_.up = new(space) ex_queue(ex_queue::UP_QUEUE,
	packTdb.toParent_,
	packTdb.returnedCriDesc_,
	space);
	// Fix up the expressions.
	packExpr()->fixup(0,getExpressionMode(),this,space,heap, FALSE, glob);
	if(predExpr()) predExpr()->fixup(0,getExpressionMode(),this,space,heap, FALSE, glob);
	}

	// Destructor.
	ExPackRowsTcb::~ExPackRowsTcb()
	{
	delete qParent_.up;
	delete qParent_.down;
	freeResources();
	}

	void ExPackRowsTcb::freeResources()
	{
	if(pool_) delete pool_;
	pool_ = 0;
	}

	// -----------------------------------------------------------------------
	// Register all the pack subtasks with the scheduler.
	// -----------------------------------------------------------------------
	void ExPackRowsTcb::registerSubtasks()
	{
	ExScheduler* sched = getGlobals()->getScheduler();

	// Schedule this routine if a new entry is inserted into the parent's
	// down queue and we are waiting on an event.
	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.
	sched->registerUnblockSubtask(sWorkDown,this,qChild_.down,"DN");

	// Schedule this routine if a cancel request occurs on the parent's
	// down queue.
	sched->registerCancelSubtask(sCancel,this,qParent_.down,"CN");

	// Schedule this routine if the parent's up queue changes from being
	// full to being not full and we are waiting on an event.
	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.
	sched->registerInsertSubtask(sWorkUp,this,qChild_.up);

	// BertBert VV
	// Schedule this routine if a GET_NEXT_N request is updated in the parent's
	// down queue and we are waiting on an event.
	sched->registerNextSubtask(sWorkDown,this,qParent_.down,"DN");
	// BertBert ^^
	}

	// -----------------------------------------------------------------------
	// Send next request down to children. Called by workDown().
	// -----------------------------------------------------------------------
	void ExPackRowsTcb::start()
	{
	// Caller needs to make sure we have room in the child's down queue.
	ex_queue_entry* cEntryDown = qChild_.down->getTailEntry();

	// Caller needs to make sure we have a new request in parent's down queue.
	ex_queue_entry* pEntryDown = qParent_.down->getQueueEntry(nextRqst_);

	const ex_queue::down_request request = pEntryDown->downState.request;
	ExPackPrivateState& pState = ((ExPackPrivateState)pEntryDown->pstate);

	// don't forward cancelled request.
	// if(request == ex_queue::GET_NOMORE)
	// {
	// pState.childState_ = CANCELLED_;
	// nextRqst_++;
	// return;
	// }

	// Pass request to children.
	cEntryDown->downState.request = request;
	cEntryDown->downState.requestValue = pEntryDown->downState.requestValue;
	// BertBert VV
	// Note, there is no complete support for the GET_NEXT_N protocol in this operator.
	cEntryDown->downState.numGetNextsIssued = pEntryDown->downState.numGetNextsIssued;
	// BertBert ^^
	cEntryDown->downState.parentIndex = nextRqst_;
	cEntryDown->copyAtp(pEntryDown->getAtp());
	pState.matchCount_ = 0;
	qChild_.down->insert();
	pState.childState_ = STARTED_;

	// If this is a CANCEL request, cancel 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)
	qChild_.down->cancelRequestWithParentIndex(nextRqst_);
	pState.childState_ = CANCELLED_;
	}
	nextRqst_++;
	}

	// -----------------------------------------------------------------------
	// All child rows have been returned. Return an EOD indication to parent.
	// -----------------------------------------------------------------------
	void ExPackRowsTcb::stop()
	{
	// Caller needs to make sure there is an outstanding parent request.
	ex_queue_entry* pEntryDown = qParent_.down->getHeadEntry();
	ExPackPrivateState& pState = ((ExPackPrivateState)pEntryDown->pstate);

	// Caller needs to make sure we have room in the parent's up queue.
	ex_queue_entry* pEntryUp = qParent_.up->getTailEntry();

	// Insert EOD to parent's up queue.
	pEntryUp->upState.status = ex_queue::Q_NO_DATA;
	pEntryUp->upState.parentIndex = pEntryDown->downState.parentIndex;
	qParent_.up->insert();

	// child row removed outside of stop().
	// Consume the child row.
	qChild_.up->removeHead();

	// Reinitialize the state of this processed request.
	pState.init();

	// This parent request has been processed.
	qParent_.down->removeHead();
	}

	// -----------------------------------------------------------------------
	// For those requests my parent cancel, cancel the corresponding requests
	// that have already been forwarded to my child.
	// -----------------------------------------------------------------------
	ExWorkProcRetcode ExPackRowsTcb::processCancel()
	{
	queue_index ix = qParent_.down->getHeadIndex();

	// Loop over all requests that have been sent down.
	while(ix != nextRqst_)
	{
	ex_queue_entry* pEntryDown = qParent_.down->getQueueEntry(ix);

	// Check whether this is a request cancelled by the parent.
	if(pEntryDown->downState.request == ex_queue::GET_NOMORE)
	{
	ExPackPrivateState& pState =
	((ExPackPrivateState)pEntryDown->pstate);

	// Cancel corresponding request for my child if it's been forwarded.
	if(pState.childState_ == ExPackRowsTcb::STARTED_)
	{
	qChild_.down->cancelRequestWithParentIndex(ix);
	pState.childState_ = ExPackRowsTcb::CANCELLED_;
	}
	}
	ix++;
	}
	return WORK_OK;
	}

	// -----------------------------------------------------------------------
	// Generic work procedure should never be called
	// -----------------------------------------------------------------------
	ExWorkProcRetcode ExPackRowsTcb::work()
	{
	ex_assert(0,"Should never reach ExPackRowsTcb::work()");
	return WORK_BAD_ERROR;
	}

	// -----------------------------------------------------------------------
	// Work procedure to send requests down
	// -----------------------------------------------------------------------
	ExWorkProcRetcode ExPackRowsTcb::workDown()
	{
	// While we have unprocessed down requests and the child's down queue
	// has room, start more child requests.
	while(
	qParent_.down->entryExists(nextRqst_) && !qChild_.down->isFull())
	start();

	return WORK_OK;
	}

	// -----------------------------------------------------------------------
	// Work procedure to process child rows.
	// -----------------------------------------------------------------------
	ExWorkProcRetcode ExPackRowsTcb::workUp()
	{
	// While there is an outstanding parent request not finished.
	while(qParent_.down->getHeadIndex() != nextRqst_)
	{
	// Get the request and retrieve its private state.
	ex_queue_entry* pEntryDown = qParent_.down->getHeadEntry();
	ExPackPrivateState& pState = ((ExPackPrivateState)pEntryDown->pstate);

	// While we have room in the up queue and rows to process.
	while(!qParent_.up->isFull() && !qChild_.up->isEmpty())
	{
	// New row produced by child.
	ex_queue_entry* cEntryUp = qChild_.up->getHeadEntry();

	// The current request has already been cancelled.
	if(pState.childState_ == ExPackRowsTcb::CANCELLED_)
	{
	if(cEntryUp->upState.status == ex_queue::Q_NO_DATA)
	{
	// No more rows for the cancelled request.
	// stop() returns Q_NO_DATA and remove the cancelled request.
	stop();

	// Break out of the inner while-loop and check for next request.
	break;
	}
	else
	{
	// Child is returning data for a request already cancelled.
	// Just ignore them and continue to check whether child's up
	// ComQueue.has any more entries.
	qChild_.up->removeHead();
	continue;
	}
	}

	if(cEntryUp->upState.status == ex_queue::Q_NO_DATA)
	{
	// Apply selection predicates to and return row being partially
	// packed in buffer if there is one.
	workReturnRow();

	// Send EOD to parent and clean up the current request.
	stop();

	// Break out of the inner while-loop and check for next request.
	break;
	}
	else if(cEntryUp->upState.status == ex_queue::Q_SQLERROR)
	{
	// Child has produced an error. Copy diagnostic area to parent.
	ex_queue_entry* pEntryUp = qParent_.up->getTailEntry();
	pEntryUp->copyAtp(cEntryUp);
	pEntryUp->upState.status = ex_queue::Q_SQLERROR;
	pEntryUp->upState.parentIndex = pEntryDown->downState.parentIndex;
	pEntryUp->upState.downIndex = qParent_.down->getHeadIndex();
	pEntryUp->upState.setMatchNo(pState.matchCount_); // (+1?)
	qParent_.up->insert();

	// Cancel the whole request on child. (may be optional)
	qChild_.down->cancelRequestWithParentIndex(
	qParent_.down->getHeadIndex());
	// Mark the corresponding request cancelled.
	pState.childState_ = ExPackRowsTcb::CANCELLED_;
	qChild_.up->removeHead();
	continue;
	}

	//
	// Now, we got a new row to be packed into the current packed row.
	// Here's the core stuff.

	// Check if the packTupp has been allocated. If not, allocate it.
	if(pState.packTupp_.getDataPointer() == NULL)
	{
	// Cannot allocate space. Call again when some space free up.
	if(pool_->getFreeTuple(pState.packTupp_))
	return WORK_POOL_BLOCKED;

	// A cheat to initialize contents of the packed row. We need this
	// since the logic of packExpr evaluation relies on the fact that
	// the no of rows already packed in a packed row (which is 0 to
	// begin with) is stored in the packed row itself, even initially.
	//
	char* packTuppPtr = pState.packTupp_.getDataPointer();
	for(Int32 i=0; i < packTdb().packTuppLen_; i++) packTuppPtr[i] = 0;

	// $$$ could potentially make it faster by initializing only the
	// $$$ first four bytes (which stores the no of rows) to 0.
	}

	// Put the pack tupp directly in child's returned ATP.
	atp_struct * cEntryUpAtp = cEntryUp->getAtp();
	cEntryUpAtp->getTupp(packTdb().packTuppIndex_) = pState.packTupp_;

	// Apply the pack expression to pack child row into packed row.
	ex_expr::exp_return_type retCode = packExpr()->eval(cEntryUpAtp,0);
	if(!cEntryUpAtp->getDiagsArea());
	else if(!pEntryDown->getDiagsArea())
	{
	pEntryDown->setDiagsArea(cEntryUpAtp->getDiagsArea());
	pEntryDown->getDiagsArea()->incrRefCount();
	}
	else
	pEntryDown->getDiagsArea()->mergeAfter(*cEntryUpAtp->getDiagsArea());

	if(retCode == ex_expr::EXPR_TRUE)
	{
	// for debugging.
	// pState.printPackTupp();

	// eval() returns TRUE if the packed record is full after the
	// child row is packed into it. Return the full packed record.

	workReturnRow();
	}
	else
	{
	// for debugging.
	// pState.printPackTupp();

	// Error occurs during expression evaluation.
	if(retCode == ex_expr::EXPR_ERROR)
	{
	// There must be room in the up queue to have reached here.
	ex_queue_entry* pEntryUp = qParent_.up->getTailEntry();

	// cEntryUp's atp has diagnostic area set up. copy it to parent.
	pEntryUp->copyAtp(cEntryUp);
	pEntryUp->upState.status = ex_queue::Q_SQLERROR;
	pEntryUp->upState.downIndex = qParent_.down->getHeadIndex();
	pEntryUp->upState.parentIndex = pEntryDown->downState.parentIndex;
	pEntryUp->upState.setMatchNo(pState.matchCount_); //(+1?)
	qParent_.up->insert();

	// mark this request cancelled in the down queue to child.
	qChild_.down->cancelRequestWithParentIndex(
	qParent_.down->getHeadIndex());
	pState.childState_ = ExPackRowsTcb::CANCELLED_;
	}
	}

	// Child row is no longer useful after the eval, get rid of it.
	qChild_.up->removeHead();

	} // while parent's up queue is not full and child's up queue not empty.

	// If we got here because we finished a request and stop()ed, then
	// try again on a new request. If we got here because the parent
	// queue is full or the child queue is empty, then we must return.
	//
	if(qParent_.up->isFull() \|\| qChild_.up->isEmpty()) {
	return WORK_OK;
	}

	} // while parent's down ComQueue.has entries.

	// Parent down queue is empty.
	return WORK_OK;

	}

	// -----------------------------------------------------------------------
	// Work procedure to apply the selection predicates on the current packed
	// record and return it.
	// -----------------------------------------------------------------------
	ExWorkProcRetcode ExPackRowsTcb::workReturnRow()
	{
	// Caller needs to make sure there is an outerstanding request.
	ex_queue_entry* pEntryDown = qParent_.down->getHeadEntry();
	ExPackPrivateState& pState = ((ExPackPrivateState)pEntryDown->pstate);

	// Nothing in packed record. Nothing to return.
	if(pState.packTupp_.getDataPointer() == NULL) return WORK_OK;

	// Copy parent's ATP from down queue to its up queue and add last tupp.
	ex_queue_entry* pEntryUp = qParent_.up->getTailEntry();
	atp_struct * pEntryUpAtp = pEntryUp->getAtp();
	pEntryUpAtp->copyAtp(pEntryDown->getAtp());
	pEntryUpAtp->getTupp(packTdb().packTuppIndex_) = pState.packTupp_;

	// Release the tupp in the private state so that it stops referencing
	// the packed record.
	pState.packTupp_.release();

	// Evaluate the selection predicates.
	ex_expr::exp_return_type predVal = ex_expr::EXPR_TRUE;
	if(predExpr()) predVal = predExpr()->eval(pEntryUpAtp,0);
	if(predVal)
	{
	// Formally insert the ATP into the parent's up queue.
	pEntryUp->upState.status = ex_queue::Q_OK_MMORE;
	pEntryUp->upState.parentIndex = pEntryDown->downState.parentIndex;
	pEntryUp->upState.setMatchNo(pState.matchCount_ + 1);
	pState.matchCount_++;
	qParent_.up->insert();
	}
	else
	{
	// Release the ATP formed in the parent's up queue entry.
	pEntryUpAtp->release();
	}
	return WORK_OK;
	}

	//////////////////////////////////////////////////////////////////////////////
	//
	// Private state procedures
	//
	//////////////////////////////////////////////////////////////////////////////

	// Constructor.
	ExPackPrivateState::ExPackPrivateState(const ExPackRowsTcb *)
	{
	packTupp_.init();
	init();
	}

	// Destructor.
	ExPackPrivateState::~ExPackPrivateState()
	{}

	// Initialization.
	void ExPackPrivateState::init()
	{
	matchCount_ = 0;
	childState_ = ExPackRowsTcb::EMPTY_;
	packTupp_.release();
	}

	// Allocation.
	ex_tcb_private_state* ExPackPrivateState::allocate_new(const ex_tcb *tcb)
	{
	return new(((ex_tcb )tcb)->getSpace()) ExPackPrivateState((ExPackRowsTcb )tcb);
	}

	void ExPackPrivateState::printPackTupp()
	{
	/*
	char* env = getenv("DEBUG_PACKING");
	if(env != NULL)
	{
	char* data = packTupp_.getDataPointer();
	printf("Pack tupp in HEX:\n");
	if(data == NULL) return;
	int len = packTupp_.getAllocatedSize();
	for(int i=0;i<len;i++) printf("%x ",data[i]);
	printf("\n");
	}
	*/
	}