core/sql/executor/ExSample.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:         $File$
 * RCS:          $Id$
 * Description:
 * Created:
 * Language:     C++
 * Status:       $State$
 *
 *
 *
 *
 ******************************************************************************
 */

 #include "ex_stdh.h"
 #include "ComTdb.h"
 #include "ex_tcb.h"
 #include "ExSample.h"
 #include "ex_expr.h"
 #include "ExSimpleSqlBuffer.h"
 #include "ExStats.h"

 // build - construct the TCB subtree for this TDB.
 //
 ex_tcb * ExSampleTdb::build(ex_globals * glob)
  {
   // first build the child
   ex_tcb *  child_tcb;
   ExSampleTcb *sampTcb;

   child_tcb = tdbChild_->build(glob);

   sampTcb = new(glob->getSpace()) ExSampleTcb(*this, *child_tcb, glob);

   sampTcb->registerSubtasks();

   return (sampTcb);
 }


 // ExSampleTcb constructor
 //
 // 1. Allocate buffer pool.
 // 2. Allocate parent queues and initialize private state.
 // 3. Fixup expressions.
 //
 ExSampleTcb::ExSampleTcb
 (const ExSampleTdb &  myTdb,
  const ex_tcb &    child_tcb,
  ex_globals * glob
  ) :
   ex_tcb(myTdb, 1, glob)
 {
   Space * space = (glob ? glob->getSpace() : 0);
   CollHeap * heap = (glob ? glob->getDefaultHeap() : 0);

   childTcb_ = &child_tcb;

    // get the queue that child use to communicate with me
   qchild_  = child_tcb.getParentQueue();

   // Allocate the queue to communicate with parent
   qparent_.down = new(space) ex_queue(ex_queue::DOWN_QUEUE,
 							myTdb.queueSizeDown_,
 				     myTdb.criDescDown_,
 				     space);

   // Allocate the private state in each entry of the down queue
   ExSamplePrivateState *p
     = new(space) ExSamplePrivateState(this);
   qparent_.down->allocatePstate(p, this);
   delete p;


   qparent_.up = new(space) ex_queue(ex_queue::UP_QUEUE,
 				    myTdb.queueSizeUp_,
 				    myTdb.criDescUp_,
 				    space);

   // Intialized processedInputs_ to the next request to process
   processedInputs_ = qparent_.down->getTailIndex();

   // Fixup the sample expression
   //
   if (initExpr())
     initExpr()->fixup(0, getExpressionMode(), this, space, heap, FALSE, glob);

   if (balanceExpr())
     balanceExpr()->fixup(0, getExpressionMode(), this, space, heap, FALSE, glob);

   if (postPred())
     postPred()->fixup(0, getExpressionMode(), this, space, heap, FALSE, glob);
 }

 // Destructor
 //
 //
 ExSampleTcb::~ExSampleTcb()
 {
   delete qparent_.up;
   delete qparent_.down;
   freeResources();
 }

 // Free Resources
 //
 //
 void ExSampleTcb::freeResources()
 {
 }

 // work - doit...
 //
 //
 short ExSampleTcb::work()
 {
   // If there are no parent requests on the queue, then there cannot
   // be anything to do here.
   //
   if (qparent_.down->isEmpty())
     return WORK_OK;

   ex_queue_entry * pentry_down;
   ExSamplePrivateState * pstate;
   ex_queue::down_request request;

   // Take any new parent requests and pass them on to the child as long
   // as the child's queue is not full. processedInputs_ maintains the
   // Queue index of the last request that was passed on.
   //
   for(queue_index tail = qparent_.down->getTailIndex();
       (processedInputs_ != tail) && (!qchild_.down->isFull());
       processedInputs_++ )
     {
       pentry_down = qparent_.down->getQueueEntry(processedInputs_);
       pstate = (ExSamplePrivateState*) pentry_down->pstate;
       request = pentry_down->downState.request;

       // If the request has already been cancelled don't pass it to the
       // child. Instead, just mark the request as done. This will trigger
       // a EOD reply when this request gets worked on.
       //
       if (request == ex_queue::GET_NOMORE)
         {
           pstate->step_ = ExSamp_DONE;
         }
       else
         {
           pstate->step_ = ExSamp_PREWORK;

 	  // Pass the request to the child
 	  //
           ex_queue_entry * centry = qchild_.down->getTailEntry();
           centry->downState.request = ex_queue::GET_ALL;
           centry->downState.requestValue = 11;
           centry->downState.parentIndex = processedInputs_;
           centry->passAtp(pentry_down);
           qchild_.down->insert();

           // Copy the input atp to the work atp for this request
 	  //
           pstate->workAtp_->copyAtp(pentry_down->getAtp());
         }
     } // end for processedInputs_

   pentry_down = qparent_.down->getHeadEntry();
   pstate = (ExSamplePrivateState*) pentry_down->pstate;
   request = pentry_down->downState.request;

   // If the request has not been worked on yet, then
   // initialize the presistent expression variable data and
   // switch to the WORKING state.
   //
   if(pstate->step_ == ExSamp_PREWORK)
     {
       pstate->step_ = ExSamp_WORKING;
       if(balanceExpr()) balanceExpr()->initializePersistentData();
     }

   ExOperStats *statsEntry = getStatsEntry();
   // Take any child replies and process them. Return the processed
   // rows as long the parent queue has room.
   //
   while (1)
     {
       // If we have satisfied the parent request (or it was cancelled),
       // then stop processing rows, cancel any outstanding child
       // requests, and set this request to the CANCELLED state.
       //
       if((pstate->step_ == ExSamp_WORKING) ||
 	 (pstate->step_ == ExSamp_RETURNINGROWS))
 	{
 	  if ((request == ex_queue::GET_NOMORE) ||
 	      ((request == ex_queue::GET_N) &&
 	       (pentry_down->downState.requestValue
 		<= (Lng32)pstate->matchCount_)))
 	    {
 	      qchild_.down->cancelRequestWithParentIndex
 		(qparent_.down->getHeadIndex());
 	      pstate->step_ = ExSamp_CANCELLED;
 	    }
 	}

       switch (pstate->step_)
 	{
 	  // ExSamp_CANCELLED
 	  //
 	  // Transition to this state from ...
 	  // 1. ExSamp_Error - After the error has been processed.
 	  // 2. ExSamp_Working - If enough rows have been returned.
 	  // 3. ExSamp_Working - If the request was cancelled.
 	  //
 	  // Remain in this state until ..
 	  // 1. All rows from the child including EOD are consumed
 	  //
 	  // Transition from this state to ...
 	  // 1. ExSamp_DONE - In all cases.
 	  //
 	case ExSamp_CANCELLED:
 	  {
 	    // There are no extra rows to process from the child yet,
 	    // so try again later.
 	    //
 	    if (qchild_.up->isEmpty())
 	      {
 		return WORK_OK;
 	      }

 	    ex_queue_entry * centry = qchild_.up->getHeadEntry();
 	    ex_queue::up_status child_status = centry->upState.status;

 	    // If this is the EOD, transition to the ExSamp_DONE state.
 	    //
 	    if (child_status == ex_queue::Q_NO_DATA)
 		  pstate->step_ = ExSamp_DONE;

 	    // Discard the child row.
 	    qchild_.up->removeHead();
 	    break;
 	  }

 	// ExSamp_ERROR
 	//
 	// Transition to this state from ...
 	// 1. ExSamp_WORKING - a child reply with the type SQLERROR.
 	//
 	// Remain in this state until ..
 	// 1. The error row has been returned to the parent.
 	//
 	// Transition from this state to ...
 	// 1. ExSamp_CANCELLED - In all cases.
 	//
 	case ExSamp_ERROR:
 	  {
             // If there is no room in the parent queue for the reply,
 	    // try again later.
 	    //
 	    if (qparent_.up->isFull())
 	      return WORK_OK;

 	    ex_queue_entry *pentry_up = qparent_.up->getTailEntry();
 	    ex_queue_entry * centry = qchild_.up->getHeadEntry();

 	    // Cancel the child request - there must be a child request in
 	    // progress to get to the ExSamp_ERROR state.
 	    //
 	    qchild_.down->cancelRequestWithParentIndex
 	      (qparent_.down->getHeadIndex());

 	    // Construct and return the error row.
 	    //
             pentry_up->copyAtp(centry);
 	    pentry_up->upState.status = ex_queue::Q_SQLERROR;
 	    pentry_up->upState.parentIndex
 	      = pentry_down->downState.parentIndex;
 	    pentry_up->upState.downIndex = qparent_.down->getHeadIndex();
             pentry_up->upState.setMatchNo(pstate->matchCount_);
             qparent_.up->insert();

 	    // Transition to the ExSamp_CANCELLED state.
 	    //
 	    pstate->step_ = ExSamp_CANCELLED;
 	    break;
 	  }

 	// ExSamp_WORKING
 	//
 	// Transition to this state from ...
 	// 1. ExSamp_EMPTY - If a request is started.
 	//
 	// Remain in this state until ...
 	// 1. All child replies including EOD have been processed.
 	// 2. A SQLERROR row is received.
 	// 3. Enough rows have been returned.
 	// 4. The request is cancelled.
 	//
 	// Transition from this state to ...
 	// 1. ExSamp_DONE - If all the child rows including EOD have
 	//    been processed.
 	// 2. ExSamp_ERROR - If an SQLERROR rows is received.
 	// 3. ExSamp_CANCELLED - If enough rows have been returned.
 	// 3. ExSamp_CANCELLED - If the request is cancelled.
 	//
 	case ExSamp_WORKING:
 	  {
 	    // If there is not room in the parent Queue for the reply,
 	    // try again later.
 	    //
 	    if (qparent_.up->isFull())
 	      return WORK_OK;

 	    // If there are no replies, try again later.
 	    //
 	    if (qchild_.up->isEmpty())
 	      return WORK_OK;

 	    ex_queue_entry * centry = qchild_.up->getHeadEntry();
 	    switch (centry->upState.status)
 	      {

 		// A data row from the child.
 		//
 	      case ex_queue::Q_OK_MMORE:
 		{
 		  // Apply the sampling predicate if it exists and extract
 		  // the sampling factor.
 		  //
 		  ex_expr::exp_return_type retCode = ex_expr::EXPR_TRUE;
 		  Int32 samplingFactor = 1;
 		  if(balanceExpr())
 		    {
 		      retCode = balanceExpr()->eval
 			(centry->getAtp(), centry->getAtp());

 		      if(retCode == ex_expr::EXPR_OK)
 			{
 			  samplingFactor =
 			    *(Lng32*)balanceExpr()->getPersistentData
 			    (returnFactorOffset());

 			  // If the sampling factor is less than 0, then
 			  // we are done with this request. Mark the
 			  // request as get-no-more. Forces the child to be
                           // cancelled
 			  //
 			  if(samplingFactor < 0)
 			    {
                               // Cancel the rest of this
                               // request.
 			      pentry_down->downState.request
 				= ex_queue::GET_NOMORE;
 			      request = ex_queue::GET_NOMORE;

                               // Return no rows
 			      retCode = ex_expr::EXPR_FALSE;
 			    }
 			}
 		    }

 		  // If the row passed the sampling predicate, apply the
 		  // selection predicate if it exists.
 		  //
 		  if ((samplingFactor > 0) &&
 		      (retCode == ex_expr::EXPR_OK) &&
 		      postPred())
 		    retCode = postPred()->eval
 		      (centry->getAtp(), centry->getAtp());

 		  // Act on the result of the selection predicate.
 		  //
 		  switch(retCode) {
 		    // If the selection predicate returns TRUE,
 		    // return the row to the parent the number
 		    // of times indicated by the sampling factor as long
 		    // as there is room in the parent queue.
 		    //
 		  case ex_expr::EXPR_TRUE:
 		  case ex_expr::EXPR_OK:
 		    while(!qparent_.up->isFull() && (samplingFactor > 0))
 		      {
 			// Copy the child ATP to the parent ATP -- the
 			// row images are exactly the same.
 			//
 			ex_queue_entry * pentry_up
 			  = qparent_.up->getTailEntry();
 			pentry_up->copyAtp(centry);

 			// Fixup the up state.
 			//
 			pentry_up->upState.status = ex_queue::Q_OK_MMORE;
 			pentry_up->upState.parentIndex
 			  = pentry_down->downState.parentIndex;
 			pentry_up->upState.downIndex
 			  = qparent_.down->getHeadIndex();
 			pstate->matchCount_++;
 			pentry_up->upState.setMatchNo(pstate->matchCount_);

 			// Commit the entry.
 			//
 			qparent_.up->insert();
                         if (statsEntry)
                            statsEntry->incActualRowsReturned();
 			samplingFactor--;

 			// If we have satisfied a GET_N request, then
 			// break out of here so we can stop processing.
 			//
 			if((request == ex_queue::GET_N) &&
 			   (pentry_down->downState.requestValue
 			    <= (Lng32)pstate->matchCount_))
 			  {
 			    samplingFactor = 0;
 			    break;
 			  }
 		      }

 		    // If all of the rows are returned, then we are done
 		    // with this entry and can proceed.
 		    //
 		    if(samplingFactor == 0)
 		      {
 			qchild_.up->removeHead();
 		      }
 		    // Otherwise, the queue must have become full so we
 		    // have to switch to the RETURNINGROWS state to finish
 		    // the job.
 		    else
 		      {
 			pstate->rowsToReturn_ = samplingFactor;
 			pstate->step_ = ExSamp_RETURNINGROWS;
 			return WORK_OK;
 		      }

 		    break;

 		    // If the selection predicate returns FALSE,
 		    // do not return the child row.
 		    //
 		  case ex_expr::EXPR_FALSE:
 		    qchild_.up->removeHead();
 		    break;

 		    // If the selection predicate returns an ERROR,
 		    // go to the error processing state.
 		    //
 		  case ex_expr::EXPR_ERROR:
 		    pstate->step_ = ExSamp_ERROR;
 		    break;
 		  }
 		}
 		break;

 		// The EOD from the child. Transition to ExSamp_DONE.
 		//
 	      case ex_queue::Q_NO_DATA:
 		pstate->step_ = ExSamp_DONE;
 		qchild_.up->removeHead();
 		break;

 		// An SQLERROR from the child. Transition to ExSamp_ERROR.
 		//
 	      case ex_queue::Q_SQLERROR:
 		pstate->step_ = ExSamp_ERROR;
 		break;
 	      }
 	  }
 	break;

 	// ExSamp_RETURNINGROWS
 	//
 	// Transistion to this state from ...
 	// 1. ExSamp_WORKING - if up queue becomes full when returning
 	//                    multiple rows from oversampling.
 	//
 	// Remain in this state until ...
 	// 1. the multiple oversampled rows have been returned.
 	//
 	case ExSamp_RETURNINGROWS:
 	  {
 	    ex_queue_entry * centry = qchild_.up->getHeadEntry();
 	    while(!qparent_.up->isFull() && (pstate->rowsToReturn_ > 0))
 	      {
 		// Copy the child ATP to the parent ATP -- the row
 		// images are exactly the same.
 		//
 		ex_queue_entry * pentry_up = qparent_.up->getTailEntry();
 		pentry_up->copyAtp(centry);

 		// Fixup the up state.
 		//
 		pentry_up->upState.status = ex_queue::Q_OK_MMORE;
 		pentry_up->upState.parentIndex
 		  = pentry_down->downState.parentIndex;
 		pentry_up->upState.downIndex = qparent_.down->getHeadIndex();
 		pstate->matchCount_++;
 		pentry_up->upState.setMatchNo(pstate->matchCount_);

 		// Commit the entry.
 		//
 		qparent_.up->insert();
                 if (statsEntry)
                    statsEntry->incActualRowsReturned();

 		pstate->rowsToReturn_--;

 		// If we have satisfied a GET_N request, then
 		// break out of here so we can stop processing.
 		//
 		if((request == ex_queue::GET_N) &&
 		   (pentry_down->downState.requestValue
 		    <= (Lng32)pstate->matchCount_))
 		  {
 		    pstate->rowsToReturn_ = 0;
 		    break;
 		  }
 	      }

 	    // If all of the rows were returned, remove the child's reply
 	    // and go back to the WORKING state.
 	    //
 	    if(pstate->rowsToReturn_ == 0)
 	      {
 		qchild_.up->removeHead();
 		pstate->step_ = ExSamp_WORKING;
 	      }
 	    // Otherwise, we need to come back later to finish up.
 	    //
 	    else
 	      {
 		return WORK_OK;
 	      }
 	  }
 	break;

 	// ExSamp_DONE
 	//
 	// Transition to the state from ...
 	// 1. ExSamp_WORKING - if all child rows have been processed.
 	// 2. ExSamp_CANCELLED - if all child rows have been consumed.
 	// 3. ExSamp_EMPTY - if the request was DOA.
 	//
 	// Remain in this state until ...
 	// 1. The EOD is returned to the parent.
 	//
 	// Transition from this state to ...
 	// 1. ExSamp_EMPTY - In all cases.
 	//
 	case ExSamp_DONE:
 	  {
 	    // If there is not any room in the parent's queue,
 	    // try again later.
 	    //
 	    if (qparent_.up->isFull())
 	      return WORK_OK;

 	    ex_queue_entry * pentry_up = qparent_.up->getTailEntry();
 	    pentry_up->upState.status = ex_queue::Q_NO_DATA;
 	    pentry_up->upState.parentIndex
 	      = pentry_down->downState.parentIndex;
 	    pentry_up->upState.downIndex = qparent_.down->getHeadIndex();
 	    pentry_up->upState.setMatchNo(pstate->matchCount_);

 	    qparent_.down->removeHead();
 	    qparent_.up->insert();

             // Re-initialize pstate
 	    //
 	    pstate->step_ = ExSamp_EMPTY;
             pstate->matchCount_ = 0;
 	    pstate->rowsToReturn_ = 0;
             pstate->workAtp_->release();

 	    // If there are no more requests, simply return.
 	    //
             if (qparent_.down->isEmpty())
               return WORK_OK;

             // If we haven't given to our child the new head
             // index return and ask to be called again.
 	    //
             if (qparent_.down->getHeadIndex() == processedInputs_)
 	      return WORK_CALL_AGAIN;

             // Postion at the new head of the request queue.
 	    //
             pentry_down = qparent_.down->getHeadEntry();
             pstate = (ExSamplePrivateState*) pentry_down->pstate;
             request = pentry_down->downState.request;

 	    // If the request has not been worked on yet, then initialize the
 	    // presistent expression variable data and switch to the
 	    // WORKING state.
 	    //
 	    if(pstate->step_ == ExSamp_PREWORK)
 	      {
 		pstate->step_ = ExSamp_WORKING;
 		if(balanceExpr()) balanceExpr()->initializePersistentData();
 	      }
 	  }
 	break;
 	} // switch pstate->step_
     } // while
 }


 // Constructor and destructor private state
 //
 //
 ExSamplePrivateState::ExSamplePrivateState
 (const ExSampleTcb *  tcb)
 {
   matchCount_ = 0;
   step_ = ExSampleTcb::ExSamp_EMPTY;
   workAtp_ = allocateAtp(tcb->myTdb().criDescUp_, ((ex_tcb*)tcb)->getSpace());
 }

 ExSamplePrivateState::~ExSamplePrivateState()
 {
 };

 ex_tcb_private_state * ExSamplePrivateState::allocate_new
 (const ex_tcb *tcb)
 {
   return new(((ex_tcb*)tcb)->getSpace())
     ExSamplePrivateState((ExSampleTcb*) tcb);
 };
	/**********************************************************************
	// @@@ 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$
	* RCS: $Id$
	* Description:
	* Created:
	* Language: C++
	* Status: $State$
	*
	*
	*
	*
	******************************************************************************
	*/

	#include "ex_stdh.h"
	#include "ComTdb.h"
	#include "ex_tcb.h"
	#include "ExSample.h"
	#include "ex_expr.h"
	#include "ExSimpleSqlBuffer.h"
	#include "ExStats.h"

	// build - construct the TCB subtree for this TDB.
	//
	ex_tcb * ExSampleTdb::build(ex_globals * glob)
	{
	// first build the child
	ex_tcb * child_tcb;
	ExSampleTcb *sampTcb;

	child_tcb = tdbChild_->build(glob);

	sampTcb = new(glob->getSpace()) ExSampleTcb(this, child_tcb, glob);

	sampTcb->registerSubtasks();

	return (sampTcb);
	}


	// ExSampleTcb constructor
	//
	// 1. Allocate buffer pool.
	// 2. Allocate parent queues and initialize private state.
	// 3. Fixup expressions.
	//
	ExSampleTcb::ExSampleTcb
	(const ExSampleTdb & myTdb,
	const ex_tcb & child_tcb,
	ex_globals * glob
	) :
	ex_tcb(myTdb, 1, glob)
	{
	Space * space = (glob ? glob->getSpace() : 0);
	CollHeap * heap = (glob ? glob->getDefaultHeap() : 0);

	childTcb_ = &child_tcb;

	// get the queue that child use to communicate with me
	qchild_ = child_tcb.getParentQueue();

	// Allocate the queue to communicate with parent
	qparent_.down = new(space) ex_queue(ex_queue::DOWN_QUEUE,
	myTdb.queueSizeDown_,
	myTdb.criDescDown_,
	space);

	// Allocate the private state in each entry of the down queue
	ExSamplePrivateState *p
	= new(space) ExSamplePrivateState(this);
	qparent_.down->allocatePstate(p, this);
	delete p;


	qparent_.up = new(space) ex_queue(ex_queue::UP_QUEUE,
	myTdb.queueSizeUp_,
	myTdb.criDescUp_,
	space);

	// Intialized processedInputs_ to the next request to process
	processedInputs_ = qparent_.down->getTailIndex();

	// Fixup the sample expression
	//
	if (initExpr())
	initExpr()->fixup(0, getExpressionMode(), this, space, heap, FALSE, glob);

	if (balanceExpr())
	balanceExpr()->fixup(0, getExpressionMode(), this, space, heap, FALSE, glob);

	if (postPred())
	postPred()->fixup(0, getExpressionMode(), this, space, heap, FALSE, glob);
	}

	// Destructor
	//
	//
	ExSampleTcb::~ExSampleTcb()
	{
	delete qparent_.up;
	delete qparent_.down;
	freeResources();
	}

	// Free Resources
	//
	//
	void ExSampleTcb::freeResources()
	{
	}

	// work - doit...
	//
	//
	short ExSampleTcb::work()
	{
	// If there are no parent requests on the queue, then there cannot
	// be anything to do here.
	//
	if (qparent_.down->isEmpty())
	return WORK_OK;

	ex_queue_entry * pentry_down;
	ExSamplePrivateState * pstate;
	ex_queue::down_request request;

	// Take any new parent requests and pass them on to the child as long
	// as the child's queue is not full. processedInputs_ maintains the
	// Queue index of the last request that was passed on.
	//
	for(queue_index tail = qparent_.down->getTailIndex();
	(processedInputs_ != tail) && (!qchild_.down->isFull());
	processedInputs_++ )
	{
	pentry_down = qparent_.down->getQueueEntry(processedInputs_);
	pstate = (ExSamplePrivateState*) pentry_down->pstate;
	request = pentry_down->downState.request;

	// If the request has already been cancelled don't pass it to the
	// child. Instead, just mark the request as done. This will trigger
	// a EOD reply when this request gets worked on.
	//
	if (request == ex_queue::GET_NOMORE)
	{
	pstate->step_ = ExSamp_DONE;
	}
	else
	{
	pstate->step_ = ExSamp_PREWORK;

	// Pass the request to the child
	//
	ex_queue_entry * centry = qchild_.down->getTailEntry();
	centry->downState.request = ex_queue::GET_ALL;
	centry->downState.requestValue = 11;
	centry->downState.parentIndex = processedInputs_;
	centry->passAtp(pentry_down);
	qchild_.down->insert();

	// Copy the input atp to the work atp for this request
	//
	pstate->workAtp_->copyAtp(pentry_down->getAtp());
	}
	} // end for processedInputs_

	pentry_down = qparent_.down->getHeadEntry();
	pstate = (ExSamplePrivateState*) pentry_down->pstate;
	request = pentry_down->downState.request;

	// If the request has not been worked on yet, then
	// initialize the presistent expression variable data and
	// switch to the WORKING state.
	//
	if(pstate->step_ == ExSamp_PREWORK)
	{
	pstate->step_ = ExSamp_WORKING;
	if(balanceExpr()) balanceExpr()->initializePersistentData();
	}

	ExOperStats *statsEntry = getStatsEntry();
	// Take any child replies and process them. Return the processed
	// rows as long the parent queue has room.
	//
	while (1)
	{
	// If we have satisfied the parent request (or it was cancelled),
	// then stop processing rows, cancel any outstanding child
	// requests, and set this request to the CANCELLED state.
	//
	if((pstate->step_ == ExSamp_WORKING) \|\|
	(pstate->step_ == ExSamp_RETURNINGROWS))
	{
	if ((request == ex_queue::GET_NOMORE) \|\|
	((request == ex_queue::GET_N) &&
	(pentry_down->downState.requestValue
	<= (Lng32)pstate->matchCount_)))
	{
	qchild_.down->cancelRequestWithParentIndex
	(qparent_.down->getHeadIndex());
	pstate->step_ = ExSamp_CANCELLED;
	}
	}

	switch (pstate->step_)
	{
	// ExSamp_CANCELLED
	//
	// Transition to this state from ...
	// 1. ExSamp_Error - After the error has been processed.
	// 2. ExSamp_Working - If enough rows have been returned.
	// 3. ExSamp_Working - If the request was cancelled.
	//
	// Remain in this state until ..
	// 1. All rows from the child including EOD are consumed
	//
	// Transition from this state to ...
	// 1. ExSamp_DONE - In all cases.
	//
	case ExSamp_CANCELLED:
	{
	// There are no extra rows to process from the child yet,
	// so try again later.
	//
	if (qchild_.up->isEmpty())
	{
	return WORK_OK;
	}

	ex_queue_entry * centry = qchild_.up->getHeadEntry();
	ex_queue::up_status child_status = centry->upState.status;

	// If this is the EOD, transition to the ExSamp_DONE state.
	//
	if (child_status == ex_queue::Q_NO_DATA)
	pstate->step_ = ExSamp_DONE;

	// Discard the child row.
	qchild_.up->removeHead();
	break;
	}

	// ExSamp_ERROR
	//
	// Transition to this state from ...
	// 1. ExSamp_WORKING - a child reply with the type SQLERROR.
	//
	// Remain in this state until ..
	// 1. The error row has been returned to the parent.
	//
	// Transition from this state to ...
	// 1. ExSamp_CANCELLED - In all cases.
	//
	case ExSamp_ERROR:
	{
	// If there is no room in the parent queue for the reply,
	// try again later.
	//
	if (qparent_.up->isFull())
	return WORK_OK;

	ex_queue_entry *pentry_up = qparent_.up->getTailEntry();
	ex_queue_entry * centry = qchild_.up->getHeadEntry();

	// Cancel the child request - there must be a child request in
	// progress to get to the ExSamp_ERROR state.
	//
	qchild_.down->cancelRequestWithParentIndex
	(qparent_.down->getHeadIndex());

	// Construct and return the error row.
	//
	pentry_up->copyAtp(centry);
	pentry_up->upState.status = ex_queue::Q_SQLERROR;
	pentry_up->upState.parentIndex
	= pentry_down->downState.parentIndex;
	pentry_up->upState.downIndex = qparent_.down->getHeadIndex();
	pentry_up->upState.setMatchNo(pstate->matchCount_);
	qparent_.up->insert();

	// Transition to the ExSamp_CANCELLED state.
	//
	pstate->step_ = ExSamp_CANCELLED;
	break;
	}

	// ExSamp_WORKING
	//
	// Transition to this state from ...
	// 1. ExSamp_EMPTY - If a request is started.
	//
	// Remain in this state until ...
	// 1. All child replies including EOD have been processed.
	// 2. A SQLERROR row is received.
	// 3. Enough rows have been returned.
	// 4. The request is cancelled.
	//
	// Transition from this state to ...
	// 1. ExSamp_DONE - If all the child rows including EOD have
	// been processed.
	// 2. ExSamp_ERROR - If an SQLERROR rows is received.
	// 3. ExSamp_CANCELLED - If enough rows have been returned.
	// 3. ExSamp_CANCELLED - If the request is cancelled.
	//
	case ExSamp_WORKING:
	{
	// If there is not room in the parent Queue for the reply,
	// try again later.
	//
	if (qparent_.up->isFull())
	return WORK_OK;

	// If there are no replies, try again later.
	//
	if (qchild_.up->isEmpty())
	return WORK_OK;

	ex_queue_entry * centry = qchild_.up->getHeadEntry();
	switch (centry->upState.status)
	{

	// A data row from the child.
	//
	case ex_queue::Q_OK_MMORE:
	{
	// Apply the sampling predicate if it exists and extract
	// the sampling factor.
	//
	ex_expr::exp_return_type retCode = ex_expr::EXPR_TRUE;
	Int32 samplingFactor = 1;
	if(balanceExpr())
	{
	retCode = balanceExpr()->eval
	(centry->getAtp(), centry->getAtp());

	if(retCode == ex_expr::EXPR_OK)
	{
	samplingFactor =
	(Lng32)balanceExpr()->getPersistentData
	(returnFactorOffset());

	// If the sampling factor is less than 0, then
	// we are done with this request. Mark the
	// request as get-no-more. Forces the child to be
	// cancelled
	//
	if(samplingFactor < 0)
	{
	// Cancel the rest of this
	// request.
	pentry_down->downState.request
	= ex_queue::GET_NOMORE;
	request = ex_queue::GET_NOMORE;

	// Return no rows
	retCode = ex_expr::EXPR_FALSE;
	}
	}
	}

	// If the row passed the sampling predicate, apply the
	// selection predicate if it exists.
	//
	if ((samplingFactor > 0) &&
	(retCode == ex_expr::EXPR_OK) &&
	postPred())
	retCode = postPred()->eval
	(centry->getAtp(), centry->getAtp());

	// Act on the result of the selection predicate.
	//
	switch(retCode) {
	// If the selection predicate returns TRUE,
	// return the row to the parent the number
	// of times indicated by the sampling factor as long
	// as there is room in the parent queue.
	//
	case ex_expr::EXPR_TRUE:
	case ex_expr::EXPR_OK:
	while(!qparent_.up->isFull() && (samplingFactor > 0))
	{
	// Copy the child ATP to the parent ATP -- the
	// row images are exactly the same.
	//
	ex_queue_entry * pentry_up
	= qparent_.up->getTailEntry();
	pentry_up->copyAtp(centry);

	// Fixup the up state.
	//
	pentry_up->upState.status = ex_queue::Q_OK_MMORE;
	pentry_up->upState.parentIndex
	= pentry_down->downState.parentIndex;
	pentry_up->upState.downIndex
	= qparent_.down->getHeadIndex();
	pstate->matchCount_++;
	pentry_up->upState.setMatchNo(pstate->matchCount_);

	// Commit the entry.
	//
	qparent_.up->insert();
	if (statsEntry)
	statsEntry->incActualRowsReturned();
	samplingFactor--;

	// If we have satisfied a GET_N request, then
	// break out of here so we can stop processing.
	//
	if((request == ex_queue::GET_N) &&
	(pentry_down->downState.requestValue
	<= (Lng32)pstate->matchCount_))
	{
	samplingFactor = 0;
	break;
	}
	}

	// If all of the rows are returned, then we are done
	// with this entry and can proceed.
	//
	if(samplingFactor == 0)
	{
	qchild_.up->removeHead();
	}
	// Otherwise, the queue must have become full so we
	// have to switch to the RETURNINGROWS state to finish
	// the job.
	else
	{
	pstate->rowsToReturn_ = samplingFactor;
	pstate->step_ = ExSamp_RETURNINGROWS;
	return WORK_OK;
	}

	break;

	// If the selection predicate returns FALSE,
	// do not return the child row.
	//
	case ex_expr::EXPR_FALSE:
	qchild_.up->removeHead();
	break;

	// If the selection predicate returns an ERROR,
	// go to the error processing state.
	//
	case ex_expr::EXPR_ERROR:
	pstate->step_ = ExSamp_ERROR;
	break;
	}
	}
	break;

	// The EOD from the child. Transition to ExSamp_DONE.
	//
	case ex_queue::Q_NO_DATA:
	pstate->step_ = ExSamp_DONE;
	qchild_.up->removeHead();
	break;

	// An SQLERROR from the child. Transition to ExSamp_ERROR.
	//
	case ex_queue::Q_SQLERROR:
	pstate->step_ = ExSamp_ERROR;
	break;
	}
	}
	break;

	// ExSamp_RETURNINGROWS
	//
	// Transistion to this state from ...
	// 1. ExSamp_WORKING - if up queue becomes full when returning
	// multiple rows from oversampling.
	//
	// Remain in this state until ...
	// 1. the multiple oversampled rows have been returned.
	//
	case ExSamp_RETURNINGROWS:
	{
	ex_queue_entry * centry = qchild_.up->getHeadEntry();
	while(!qparent_.up->isFull() && (pstate->rowsToReturn_ > 0))
	{
	// Copy the child ATP to the parent ATP -- the row
	// images are exactly the same.
	//
	ex_queue_entry * pentry_up = qparent_.up->getTailEntry();
	pentry_up->copyAtp(centry);

	// Fixup the up state.
	//
	pentry_up->upState.status = ex_queue::Q_OK_MMORE;
	pentry_up->upState.parentIndex
	= pentry_down->downState.parentIndex;
	pentry_up->upState.downIndex = qparent_.down->getHeadIndex();
	pstate->matchCount_++;
	pentry_up->upState.setMatchNo(pstate->matchCount_);

	// Commit the entry.
	//
	qparent_.up->insert();
	if (statsEntry)
	statsEntry->incActualRowsReturned();

	pstate->rowsToReturn_--;

	// If we have satisfied a GET_N request, then
	// break out of here so we can stop processing.
	//
	if((request == ex_queue::GET_N) &&
	(pentry_down->downState.requestValue
	<= (Lng32)pstate->matchCount_))
	{
	pstate->rowsToReturn_ = 0;
	break;
	}
	}

	// If all of the rows were returned, remove the child's reply
	// and go back to the WORKING state.
	//
	if(pstate->rowsToReturn_ == 0)
	{
	qchild_.up->removeHead();
	pstate->step_ = ExSamp_WORKING;
	}
	// Otherwise, we need to come back later to finish up.
	//
	else
	{
	return WORK_OK;
	}
	}
	break;

	// ExSamp_DONE
	//
	// Transition to the state from ...
	// 1. ExSamp_WORKING - if all child rows have been processed.
	// 2. ExSamp_CANCELLED - if all child rows have been consumed.
	// 3. ExSamp_EMPTY - if the request was DOA.
	//
	// Remain in this state until ...
	// 1. The EOD is returned to the parent.
	//
	// Transition from this state to ...
	// 1. ExSamp_EMPTY - In all cases.
	//
	case ExSamp_DONE:
	{
	// If there is not any room in the parent's queue,
	// try again later.
	//
	if (qparent_.up->isFull())
	return WORK_OK;

	ex_queue_entry * pentry_up = qparent_.up->getTailEntry();
	pentry_up->upState.status = ex_queue::Q_NO_DATA;
	pentry_up->upState.parentIndex
	= pentry_down->downState.parentIndex;
	pentry_up->upState.downIndex = qparent_.down->getHeadIndex();
	pentry_up->upState.setMatchNo(pstate->matchCount_);

	qparent_.down->removeHead();
	qparent_.up->insert();

	// Re-initialize pstate
	//
	pstate->step_ = ExSamp_EMPTY;
	pstate->matchCount_ = 0;
	pstate->rowsToReturn_ = 0;
	pstate->workAtp_->release();

	// If there are no more requests, simply return.
	//
	if (qparent_.down->isEmpty())
	return WORK_OK;

	// If we haven't given to our child the new head
	// index return and ask to be called again.
	//
	if (qparent_.down->getHeadIndex() == processedInputs_)
	return WORK_CALL_AGAIN;

	// Postion at the new head of the request queue.
	//
	pentry_down = qparent_.down->getHeadEntry();
	pstate = (ExSamplePrivateState*) pentry_down->pstate;
	request = pentry_down->downState.request;

	// If the request has not been worked on yet, then initialize the
	// presistent expression variable data and switch to the
	// WORKING state.
	//
	if(pstate->step_ == ExSamp_PREWORK)
	{
	pstate->step_ = ExSamp_WORKING;
	if(balanceExpr()) balanceExpr()->initializePersistentData();
	}
	}
	break;
	} // switch pstate->step_
	} // while
	}


	// Constructor and destructor private state
	//
	//
	ExSamplePrivateState::ExSamplePrivateState
	(const ExSampleTcb * tcb)
	{
	matchCount_ = 0;
	step_ = ExSampleTcb::ExSamp_EMPTY;
	workAtp_ = allocateAtp(tcb->myTdb().criDescUp_, ((ex_tcb*)tcb)->getSpace());
	}

	ExSamplePrivateState::~ExSamplePrivateState()
	{
	};

	ex_tcb_private_state * ExSamplePrivateState::allocate_new
	(const ex_tcb *tcb)
	{
	return new(((ex_tcb*)tcb)->getSpace())
	ExSamplePrivateState((ExSampleTcb*) tcb);
	};