// **********************************************************************
// @@@ 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 @@@
// **********************************************************************

#include "Platform.h"

#include "ex_stdh.h"
#include "ComTdb.h"
#include "ex_tcb.h"
#include "ExProbeCache.h"
#include "ex_exe_stmt_globals.h"
#include "ex_expr.h"


ex_tcb * ExProbeCacheTdb::build(ex_globals * glob)
{
  ExExeStmtGlobals * exe_glob = glob->castToExExeStmtGlobals();
  
  ex_assert(exe_glob,"Probe Cache operator can't be in DP2");

  // first build the child
  ex_tcb *  child_tcb;
  child_tcb = tdbChild_->build(glob);

  ExProbeCacheTcb *pc_tcb = new(exe_glob->getSpace()) 
        ExProbeCacheTcb(
	 *this,
         *child_tcb,
	 exe_glob);

  ex_assert(pc_tcb, "Error building ExProbeCacheTcb.");

  // Add subtasks to the scheduler.
  pc_tcb->registerSubtasks();

  // This operator does use dynamic queue resizing.
  pc_tcb->registerResizeSubtasks();

  return (pc_tcb);
}


////////////////////////////////////////////////////////////////
// Constructor and initialization.
////////////////////////////////////////////////////////////////

ExProbeCacheTcb::ExProbeCacheTcb(const ExProbeCacheTdb &probeCacheTdb, 
				   const ex_tcb &child_tcb,
				   ex_globals * glob
				    ) : 
    ex_tcb( probeCacheTdb, 1, glob),
    childTcb_(NULL),
    workAtp_(NULL),
    probeBytes_(NULL),
    pool_(NULL),
    pcm_(NULL),
    workUpTask_(NULL)
{
  Space * space = (glob ? glob->getSpace() : 0);
  CollHeap * heap = (glob ? glob->getDefaultHeap() : 0);

  qparent_.up = qparent_.down = NULL;
  
  childTcb_ = &child_tcb;

  // Allocate the buffer pool
  pool_ = new(space) ExSimpleSQLBuffer( probeCacheTdb.numInnerTuples_,
                                        probeCacheTdb.recLen_,
                                        space);

  // get the child queue pair
  qchild_  = child_tcb.getParentQueue(); 

  // Allocate the queue to communicate with parent
  allocateParentQueues(qparent_);

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

  // Allocate buffer for probeBytes_.
  probeBytes_ = new(space) char[ probeCacheTdb.probeLen_ ];

  // allocate work atp and initialize the two tupps.
  workAtp_ = allocateAtp(probeCacheTdb.criDescUp_,getSpace());

  probeHashTupp_.init(sizeof(probeHashVal_),
		    NULL,
		    (char *) (&probeHashVal_));
  workAtp_->getTupp(probeCacheTdb.hashValIdx_) = &probeHashTupp_;

  hashProbeExpr()->fixup(0, getExpressionMode(), this, space, heap,
                glob->computeSpace(), glob);

  probeEncodeTupp_.init(probeCacheTdb.probeLen_,
		    NULL,
		    (char *) (probeBytes_));
  workAtp_->getTupp(probeCacheTdb.encodedProbeDataIdx_) = &probeEncodeTupp_;

  encodeProbeExpr()->fixup(0, getExpressionMode(), this, space, heap,
                glob->computeSpace(), glob);

  if (moveInnerExpr())
    moveInnerExpr()->fixup(0, getExpressionMode(), this, space, heap,
                glob->computeSpace(), glob);

  if (selectPred())
    selectPred()->fixup(0, getExpressionMode(), this, space, heap,
                glob->computeSpace(), glob);

  pcm_ = new(space) ExPCMgr(space, 
                probeCacheTdb.cacheSize_, probeCacheTdb.probeLen_, this);

}

NABoolean ExProbeCacheTcb::needStatsEntry()
{
  // stats are collected for ALL and OPERATOR options.
  if ((getGlobals()->getStatsArea()->getCollectStatsType() == ComTdb::ALL_STATS) ||
      (getGlobals()->getStatsArea()->getCollectStatsType() == ComTdb::OPERATOR_STATS))
    return TRUE;
  else
    return FALSE;
}

ExOperStats * ExProbeCacheTcb::doAllocateStatsEntry(CollHeap *heap,
                                                        ComTdb *tdb)
{
  ExOperStats * stat = NULL;
  ComTdb::CollectStatsType statsType = getGlobals()->getStatsArea()->getCollectStatsType();
  if (statsType == ComTdb::OPERATOR_STATS)
  {
    return ex_tcb::doAllocateStatsEntry(heap, tdb);
  }
  else
  {
    ExProbeCacheTdb * pcTdb = (ExProbeCacheTdb*) tdb;

    return new(heap) ExProbeCacheStats(heap,
                      this,
                      tdb,
                      pcTdb->bufferSize_,
                      pcTdb->cacheSize_);
  }
}

void ExProbeCacheTcb::registerSubtasks()
{
  ExScheduler *sched = getGlobals()->getScheduler();

  sched->registerInsertSubtask(sWorkDown,   this, qparent_.down,"PD");
  sched->registerUnblockSubtask(sWorkDown,  this, qchild_.down, "PD");

  sched->registerInsertSubtask(sWorkUp,     this, qchild_.up,   "PU");
  sched->registerUnblockSubtask(sWorkUp,    this, qparent_.up,  "PU");

  sched->registerCancelSubtask(sCancel,     this, qparent_.down,"CN");

  // We need to schedule workUp from workDown if a call to workDown 
  // has changed any request to either CACHE_HIT or CANCELED_NOT_STARTED 
  // and if it has not changed any request to CACHE_MISS.
  workUpTask_ = sched->registerNonQueueSubtask(sWorkUp, this, "PU");

}

ex_tcb_private_state * ExProbeCacheTcb::allocatePstates(
     Lng32 &numElems,      // inout, desired/actual elements
     Lng32 &pstateLength)  // out, length of one element
{
  PstateAllocator<ExProbeCachePrivateState> pa;

  return pa.allocatePstates(this, numElems, pstateLength);
}

////////////////////////////////////////////////////////////////
// Destructor and cleanup.
////////////////////////////////////////////////////////////////

ExProbeCacheTcb::~ExProbeCacheTcb()
{
  freeResources();
}
  
void ExProbeCacheTcb::freeResources()
{
  if (pool_) 
    {
      delete pool_;
      pool_ = NULL;
    }

  if (qparent_.up)
    {
      delete qparent_.up;
      qparent_.up = NULL;
    }
  
  if (qparent_.down)
    {
      delete qparent_.down;
      qparent_.down = NULL;
    }

  if (probeBytes_)  
    {
      NADELETEBASIC(probeBytes_, getSpace());
      probeBytes_ = NULL;
    }

  if (pcm_)
    {
      delete pcm_;
      pcm_ = NULL;
    }
}
  
///////////////////////////////////////////////////////////////////
// The various work subtask methods.
///////////////////////////////////////////////////////////////////
ExWorkProcRetcode ExProbeCacheTcb::work()
{
  ex_assert(0, "ExProbeCache has separate workUp and workDown subtasks.");
  return WORK_OK;
}

///////////////////////////////////////////////////////////////////

ExWorkProcRetcode ExProbeCacheTcb::workDown()
{
  NABoolean anyMiss = FALSE;
  NABoolean anyHitOrCanceled = FALSE;
  queue_index tlindex = qparent_.down->getTailIndex();

  while (nextRequest_ != tlindex && !qchild_.down->isFull()) 
    { 
      ex_queue_entry *pentry = qparent_.down->getQueueEntry(nextRequest_);
      ExProbeCachePrivateState & pstate
	= *((ExProbeCachePrivateState *) pentry->pstate);

      ex_assert(pstate.step_ == NOT_STARTED, "deja vu in ExProbeCacheTcb")
      
      switch (pentry->downState.request)
      {
        case ex_queue::GET_N:
          {
            // As long as probe cache restricts probes to no more than
            // one reply, nothing special is need for GET_N.  Just fall
            // thru to the GET_ALL case.
          }
        case ex_queue::GET_ALL:
          {
            // Evaluate the hash expression on the probe.
            if ((hashProbeExpr()->eval(pentry->getAtp(),workAtp_)) != 
                ex_expr::EXPR_OK)
              ex_assert(0, "Unexpected result from hashProbeExpr");

            // Evaluate the encode expression on the probe.
            if ((encodeProbeExpr()->eval(pentry->getAtp(),workAtp_)) != 
                ex_expr::EXPR_OK)
              ex_assert(0, "Unexpected result from hashEncodeExpr");

            // Look for a match in the cache.  If it is not found
            // this same method will add this probe.
            if (pcm_->addOrFindEntry(
                  probeHashVal_, probeBytes_, nextRequest_, pstate.pcEntry_
                                    ) == ExPCMgr::FOUND)
              {
                // Had a probe already in cache, so workDown is finished
                // with this request.  Change step_ and move on to next one.
                pstate.step_ = CACHE_HIT;
                anyHitOrCanceled = TRUE;
              }
            else
              {
                // No probe was found in in cache; the probe cache manager has 
                // created one, but workDown must pass this request to
                // the child so that the inner table reply will be ready when 
                // this request (and any subsequent request that match this
                // request's probe data) comes to the head of the parent down
                // queue and is processed by work up.
                pstate.step_ = CACHE_MISS;
                anyMiss = TRUE;
                ex_queue_entry * centry = qchild_.down->getTailEntry();
                // pass same request down.
                centry->downState.request = pentry->downState.request;
                centry->downState.requestValue = 
                          pentry->downState.requestValue;
                centry->downState.numGetNextsIssued = 
                          pentry->downState.numGetNextsIssued;
                // remember the parent q entry from whence this request came.
                centry->downState.parentIndex = nextRequest_; 
                // set the atp to the parent atp.
                centry->passAtp(pentry);
                qchild_.down->insert();
              }
            break;
          }
        case ex_queue::GET_NOMORE:
          {
	    pstate.step_ = CANCELED_NOT_STARTED;
            anyHitOrCanceled = TRUE;
	    break;
          }
        default:
          {
            ex_assert(0, "ExProbeCacheTcb cannot handle this request");
            break;
          }
      }  // end switch on pentry->downState.request
    nextRequest_++;
    }

  if (anyHitOrCanceled && !anyMiss)
    {
      // Must be sure that workUp will be dispatched and since we 
      // cannot rely on activity in the ex_queues given that this
      // call to workDown didn't put anything in the child down queue,
      // we must scheduler workUp here.
      workUpTask_->schedule();
    }

  return WORK_OK;
}

///////////////////////////////////////////////////////////////////

ExWorkProcRetcode ExProbeCacheTcb::workUp()
{

  ExProbeCacheStats *stats = getProbeCacheStats();
  // Work on requests from the head of parent down, until
  // either we reach nextRequest_ (which work down hasn't seen yet)
  // or until there is not room in the up queue.  Note that 
  // there are "return" statements coded in this loop.
  while ((qparent_.down->getHeadIndex() != nextRequest_) &&
         !qparent_.up->isFull())
    {
      ex_queue_entry *pentry_down = qparent_.down->getHeadEntry();
      ExProbeCachePrivateState & pstate
	= *((ExProbeCachePrivateState *) pentry_down->pstate);

      switch (pstate.step_)
        {
          case CACHE_MISS: 
            { 
              if (qchild_.up->isEmpty())
                return WORK_OK;
              ex_queue_entry *reply = qchild_.up->getHeadEntry();
              switch( reply->upState.status )
                {
                  case ex_queue::Q_OK_MMORE:
                    {
                      MoveStatus moveRetCode = 
                        moveReplyToCache(*reply, *pstate.pcEntry_);

                      if (moveRetCode == MOVE_BLOCKED)
                        {
                          return WORK_POOL_BLOCKED;
                        }
                      else if (moveRetCode == MOVE_OK)
                        {
                          pstate.matchCount_ = 1;

                          makeReplyToParentUp(pentry_down, pstate, 
                               ex_queue::Q_OK_MMORE);

                          // Cancel here, b/c semi-join and anti-semi-join
                          // will return more than one Q_OK_MMORE.  Tbd -
                          // perhaps the tdb should pass a flag for this.  

                          qchild_.down->cancelRequestWithParentIndex(
                                    pstate.pcEntry_->probeQueueIndex_);

                          break;
                        }
                      else
                        {
                          ex_assert(moveRetCode == MOVE_ERROR, 
                                    "bad retcode from moveReplyToCache");
                          // Don't break from this Q_OK_MMORE case, but
                          // instead flow down as if Q_SQLERROR.  The 
                          // diagsArea should have been init'd in the 
                          // moveInnerExpr()->eval. 
                        }
                    }
                  case ex_queue::Q_SQLERROR:
                    {
                      // Initialize ExPCE members
                      pstate.pcEntry_->upstateStatus_ = ex_queue::Q_SQLERROR;
  
                      ComDiagsArea *da = reply->getAtp()->getDiagsArea();
                      ex_assert(da, "Q_SQLERROR without a diags area");
                      pstate.pcEntry_->diagsArea_ =  da;
                      da->incrRefCount();

                      makeReplyToParentUp(pentry_down, pstate,
                                          ex_queue::Q_SQLERROR);

                      // No need to cancel, since we expect no more than
                      // one reply from our child.

                      break;
                    }
                  case ex_queue::Q_NO_DATA:
                    {
                      // Initialize ExPCE members
                      pstate.pcEntry_->upstateStatus_ = ex_queue::Q_NO_DATA;
  
                      ComDiagsArea *da = reply->getAtp()->getDiagsArea();
                      if (da)
                        {
                          pstate.pcEntry_->diagsArea_ =  da;
                          da->incrRefCount();
                        }
                      break;
                      // A Q_NO_DATA will be inserted into the parent up queue 
                      // in the DONE step_.  
                    }
                  default:
                    {
                      ex_assert(0, "Unknown upstate.status in child up queue");
                      break;
                    }
                }
              if (stats)
                stats->incMiss();
              pstate.pcEntry_->release(); //Request no longer references PCE.
              pstate.step_ = DONE_MISS;
              break;
            }
          case CACHE_HIT:
            {
              switch(pstate.pcEntry_->upstateStatus_)
                {
                  case ex_queue::Q_OK_MMORE:
                    {
                      pstate.matchCount_ = 1;

                      makeReplyToParentUp(pentry_down, pstate,
                            pstate.pcEntry_->upstateStatus_);

                      break;
                    }
                  case ex_queue::Q_SQLERROR:
                    {

                      makeReplyToParentUp(pentry_down, pstate,
                            pstate.pcEntry_->upstateStatus_);

                      // No need to cancel, since we expect no more than
                      // one reply from our child.

                      break;
                    }
                  case ex_queue::Q_NO_DATA:
                    {
                      // The DONE step will handle this.
                      break;
                    }
                  case ex_queue::Q_INVALID:
                    {
                      // Should not happen.
                      ex_assert(0, "CACHE_HIT saw Q_INVALID");
                    }
                  default:
                    {
                      // Should not happen.
                      ex_assert(0, "CACHE_HIT saw unknown upstateStatus");
                    }
                }
              if (stats)
                stats->incHit();
              pstate.step_ = DONE;
              pstate.pcEntry_->release(); //Request no longer references PCE.
              break;
            }
          case CANCELED_MISS:
            { 
              if (qchild_.up->isEmpty())
                return WORK_OK;
              if (pstate.pcEntry_->refCnt_ != 0)
                {
                  // There are other requests that are interested in this
                  // reply, so put it into the Probe Cache, according to
                  // its upState.status.  However, do not reply with Q_OK_MMORE
                  // or Q_SQLERROR to this request.
                  ex_queue_entry *reply = qchild_.up->getHeadEntry();
                  switch( reply->upState.status )
                    {
                      case ex_queue::Q_OK_MMORE:
                        {
                          MoveStatus moveRetCode2 = 
                            moveReplyToCache(*reply, *pstate.pcEntry_);
                          if (moveRetCode2 == MOVE_BLOCKED)
                            {
                              return WORK_POOL_BLOCKED;
                            }
                          else if (moveRetCode2 == MOVE_OK)
                            {
                              // Now that we have the reply, we can propagate 
                              // the cancel.  We do this b/c (anti-)semi-join 
                              // will return more than one Q_OK_MMORE.  Tbd -
                              // perhaps the tdb should pass a flag for this.  

                              qchild_.down->cancelRequestWithParentIndex(
                                        pstate.pcEntry_->probeQueueIndex_);
                              break;
                            }
                          else
                            {
                              ex_assert(moveRetCode2 == MOVE_ERROR, 
                                        "bad retcode from moveReplyToCache");
                              // Don't break from this Q_OK_MMORE case, but
                              // instead flow down as if Q_SQLERROR.  The 
                              // diagsArea should have been init'd in the 
                              // moveInnerExpr()->eval. 
                            }
                        }
                      case ex_queue::Q_SQLERROR:
                        {
                          // Initialize ExPCE members
                          pstate.pcEntry_->upstateStatus_ = ex_queue::Q_SQLERROR;
  
                          ComDiagsArea *da = reply->getAtp()->getDiagsArea();
                          ex_assert(da, "Q_SQLERROR without a diags area");
                          pstate.pcEntry_->diagsArea_ =  da;
                          da->incrRefCount();

                          break;
                        }
                      case ex_queue::Q_NO_DATA:
                        {
                          // Initialize ExPCE members
                          pstate.pcEntry_->upstateStatus_ = ex_queue::Q_NO_DATA;
  
                          ComDiagsArea *da = reply->getAtp()->getDiagsArea();
                          if (da)
                            {
                              pstate.pcEntry_->diagsArea_ =  da;
                              da->incrRefCount();
                            }
                          break;
                        }
                      default:
                        {
                          ex_assert(0, "Unknown upstate.status in child up queue");
                          break;
                        }
                    }
                }
              else
                {
                  // There are no uncanceled requests interested 
                  // in this  PCE.  
                }
              if (stats)
                stats->incCanceledMiss();
              pstate.step_ = DONE_MISS;
              break;
            }
          case CANCELED_HIT:
            {
              if (stats)
                stats->incCanceledHit();
              pstate.step_ = DONE;
              break;
            }
          case CANCELED_NOT_STARTED:
            {
              if (stats)
                stats->incCanceledNotStarted();
              pstate.step_ = DONE;
              break;
            }
          case DONE_MISS:
            {
              // In this step we discard the original 1st reply to the 
              // CACHE_MISS or CANCELED_MISS, as well as any other replies.
              // We can get multiple replies for a semi-join or
              // anti-semi-join.  
              // We also discard the Q_NO_DATA.
              NABoolean finishedDoneMiss = FALSE;
              while (!finishedDoneMiss)
                {
                  if (qchild_.up->isEmpty())
                    return WORK_OK;
                  ex_queue_entry *reply2 = qchild_.up->getHeadEntry();
                  if (reply2->upState.status == ex_queue::Q_NO_DATA)
                    finishedDoneMiss = TRUE;
                  qchild_.up->removeHead();
                }
              pstate.step_ = DONE;  
              break;
            }
          case DONE:
            {
              makeReplyToParentUp(pentry_down, pstate, ex_queue::Q_NO_DATA);
                 
              pstate.init();
              qparent_.down->removeHead();
              break;
            }
          case NOT_STARTED:
          default:  
            {
              ex_assert(0, "workUp saw unexpected pstate.step_");
              break;
            }
        }
    }
  return WORK_OK;
}

///////////////////////////////////////////////////////////////////

ExWorkProcRetcode ExProbeCacheTcb::workCancel()
{
  queue_index qindex = qparent_.down->getHeadIndex();

  while (qindex != nextRequest_) 
    { 
      ex_queue_entry *pentry = qparent_.down->getQueueEntry(qindex);
      ExProbeCachePrivateState & pstate
	= *((ExProbeCachePrivateState *) pentry->pstate);
      if (pentry->downState.request == ex_queue::GET_NOMORE)
        {
          switch (pstate.step_)
            {
              case CACHE_MISS:
                {
                  cancelInterest(pstate.pcEntry_);
                  pstate.step_ = CANCELED_MISS;
                  break;
                }
              case CACHE_HIT:
                {
                  cancelInterest(pstate.pcEntry_);
                  pstate.step_ = CANCELED_HIT;
                  break;
                }
              case DONE:
              case DONE_MISS:
                {
                  // This can happen, but no further canceling is needed.
                  break;
                }
              case CANCELED_MISS:
              case CANCELED_HIT:
              case CANCELED_NOT_STARTED:
                {
                  // I think this can happen, but no further canceling needed.
                  break;
                }
              case NOT_STARTED:
                {
                  // The comparison of qindex to nextRequest_ should prevent.
                  ex_assert(0, 
                  "canceling & parent down queue processing out of sync");
                }
              default:
                {
                  ex_assert(0, 
                  "canceling found bad pstate.step_");
                }
            }
        }
      qindex++;
    }
  return WORK_OK;
}

///////////////////////////////////////////////////////////////////
// Some helper methods called by the work methods.
///////////////////////////////////////////////////////////////////

void ExProbeCacheTcb::makeReplyToParentUp(ex_queue_entry *pentry_down, 
                                      ExProbeCachePrivateState &pstate, 
                                      ex_queue::up_status reply_status)
{
  ExOperStats *stats;
  ex_queue_entry * up_entry = qparent_.up->getTailEntry();
  Int32 rowQualifies = 1;
  
  // Copy the pointers to the input data from parent
  up_entry->copyAtp(pentry_down);

  if ((reply_status == ex_queue::Q_OK_MMORE) &&
      (pstate.pcEntry_->innerRowTupp_.isAllocated()))
    {
      // Use the pcEntry_ to set the returned tuple from 
      // the pool. 
      up_entry->getAtp()->getTupp(probeCacheTdb().tuppIndex_) = 
            pstate.pcEntry_->innerRowTupp_;

      if (selectPred())
        {
          ex_expr::exp_return_type evalRetCode =
            selectPred()->eval(up_entry->getAtp(), 0);

          if (evalRetCode == ex_expr::EXPR_FALSE)
            rowQualifies = 0;
          else if (evalRetCode != ex_expr::EXPR_TRUE)
            {
              ex_assert(evalRetCode == ex_expr::EXPR_ERROR,
                        "invalid return code from expr eval");
              // diags area should have been generated
              reply_status = ex_queue::Q_SQLERROR;
            }
        }
    }

  if (rowQualifies)
    {
      // Initialize the upState.
      up_entry->upState.parentIndex = 
        pentry_down->downState.parentIndex;
      up_entry->upState.downIndex = qparent_.down->getHeadIndex();
      up_entry->upState.setMatchNo(pstate.matchCount_);
      up_entry->upState.status = reply_status;

      // Give the reply any diagsArea.  Test pcEntry_ before using it
      // because a request that went from CANCELED_NOT_STARTED to 
      // DONE will never be hooked up with a pcEntry_.  
      if (pstate.pcEntry_ &&
          pstate.pcEntry_->diagsArea_) 
        {	
          ComDiagsArea *accumulatedDiagsArea = 
            up_entry->getDiagsArea();
          if (accumulatedDiagsArea)
            accumulatedDiagsArea->mergeAfter(*pstate.pcEntry_->diagsArea_);
          else
            {
              up_entry->setDiagsArea(pstate.pcEntry_->diagsArea_);
              pstate.pcEntry_->diagsArea_->incrRefCount();
            }
        }

      // Insert the reply.
      qparent_.up->insert();
      if ((stats = getStatsEntry()) != NULL && reply_status == ex_queue::Q_OK_MMORE)
        stats->incActualRowsReturned();
    }
  return;
}

///////////////////////////////////////////////////////////////////

ExProbeCacheTcb::MoveStatus 
ExProbeCacheTcb::moveReplyToCache(ex_queue_entry &reply, 
                                             ExPCE &pcEntry)
{
  if (moveInnerExpr())
    {
      ex_assert(!pcEntry.innerRowTupp_.isAllocated(),
                "reusing an allocated innerRowTupp");

      if (pool_->getFreeTuple(pcEntry.innerRowTupp_))
        return MOVE_BLOCKED;

      workAtp_->getTupp(probeCacheTdb().innerRowDataIdx_) = 
            pcEntry.innerRowTupp_;

      // Evaluate the move expression on the reply.
      ex_expr::exp_return_type innerMoveRtn = 
        moveInnerExpr()->eval(reply.getAtp(),workAtp_);
      
      if (innerMoveRtn == ex_expr::EXPR_ERROR)
        return MOVE_ERROR;
    }
  else
    {
      ex_assert(pcEntry.innerRowTupp_.isAllocated() == FALSE, 
               "Incorrectly initialized inneRowTupp_");
    }

  // Initialize ExPCE members
  pcEntry.upstateStatus_ = ex_queue::Q_OK_MMORE;
  
  ComDiagsArea *da = reply.getAtp()->getDiagsArea();
  if (da)
    {
      pcEntry.diagsArea_ =  da;
      da->incrRefCount();
    }

return MOVE_OK;
}

///////////////////////////////////////////////////////////////////

void ExProbeCacheTcb::cancelInterest(ExPCE *pcEntry)
{
  pcEntry->release();

  if ((pcEntry->refCnt_ == 0) &&
      (pcEntry->upstateStatus_ == ex_queue::Q_INVALID))
    {
      // No request is interested in this probe's result and 
      // our workup method has still not processed any reply
      // from the child, so cancel the probe.  (The reason for
      // the Q_INVALID test is that it lets use distinguish
      // a probe that does not have its result from a probe
      // that was able to get a result before being canceled.)
      qchild_.down->cancelRequestWithParentIndex(
                                        pcEntry->probeQueueIndex_);

      // This ExPCE will never get any reply except
      // Q_NO_DATA, and hence is no longer a valid 
      // probe. Mark it so no future request from our parent 
      // attempts to use this probe.
      pcEntry->flags_.canceledPending_ = 1;
    }
}

///////////////////////////////////////////////////////////////////
// Methods for the ExProbeCachePrivateState
///////////////////////////////////////////////////////////////////

ExProbeCachePrivateState::ExProbeCachePrivateState()
{
  init();
}

ExProbeCachePrivateState::~ExProbeCachePrivateState() {}

void ExProbeCachePrivateState::init()
{
  step_ = ExProbeCacheTcb::NOT_STARTED;
  pcEntry_ = NULL;
  matchCount_ = 0;
}

///////////////////////////////////////////////////////////////////
// Methods for the Probe Cache Manager
///////////////////////////////////////////////////////////////////
ExPCMgr::ExPCMgr(Space *space,
                 ULng32 numEntries, 
                 ULng32 probeLength,
                 ExProbeCacheTcb *tcb) :
    space_(space),
    numBuckets_(numEntries),
    probeLen_(probeLength),
    tcb_(tcb),
    buckets_(NULL),
    entries_(NULL),
    nextVictim_(0)
{
    buckets_ = new(space_) ExPCE *[numBuckets_];

    // Initialize all the buckets to "empty".    
    memset((char *)buckets_, 0, numBuckets_ * sizeof(ExPCE *));

    // Calculate the real size for each ExPCE -- the probeData_ 
    // array is one byte so subtract that from probeLength.
    sizeofExPCE_ = ROUND8(sizeof(ExPCE) + (probeLength - 1));

    // Get the size in bytes of the ExPCE array.
    const Int32 totalExPCEsizeInBytes = numEntries * sizeofExPCE_;

    entries_ = new(space_) char[totalExPCEsizeInBytes];

    memset(entries_, 0, totalExPCEsizeInBytes);
};

///////////////////////////////////////////////////////////////////
ExPCMgr::~ExPCMgr()
{
  if (buckets_ != NULL)
    NADELETEBASIC(buckets_, space_);
  buckets_ = NULL;

  if (entries_ != NULL)
    NADELETEBASIC(entries_, space_);
  entries_ = NULL;
}

///////////////////////////////////////////////////////////////////
ExPCMgr::AddedOrFound 
ExPCMgr::addOrFindEntry( ULng32 probeHashVal, 
                         char * probeBytes, 
                         queue_index qIdxForCancel, 
                         ExPCE * &pcEntry )
{
  AddedOrFound retcode;
  const Int32 bucketNum = probeHashVal % numBuckets_;
  UInt32 chainLength = 1;
  ExProbeCacheStats *stats = tcb_->getProbeCacheStats();

  pcEntry = buckets_[bucketNum];

  while (pcEntry)
    {
      if ((pcEntry->probeHashVal_ == probeHashVal) &&
          (memcmp((char *) pcEntry->probeData_, probeBytes, probeLen_) == 0))
        {
          if (pcEntry->flags_.canceledPending_)
            {
              // Found the correct entry for this probe but 
              // but we cannot use it because the corresponding child down 
              // queue request has been canceled and we are awaiting
              // the Q_NO_DATA reply.  Keep looking in case a duplicate
              // probe added an etry after this entry was canceled.
            }
          else
            {
              // This is a hit.
              break;
            }
        }
      pcEntry = pcEntry->nextHashVal_;
      chainLength++;
    }

  if (stats)
    stats->updateLongChain(chainLength);

  if (pcEntry == NULL)
    {
      // Not found (or it was found but a cancel was pending).  So add it.
      pcEntry = addEntry(bucketNum,   // add it to this collision chain.
                     probeHashVal, 
                     probeBytes, 
                     qIdxForCancel);
      retcode = ADDED;
    }
  else 
    {
      pcEntry->refCnt_++;
      pcEntry->flags_.useBit_ = 1;
      retcode = FOUND;
      if (stats)
        stats->updateUseCount(pcEntry->refCnt_);
    }

  return retcode;
  
}

///////////////////////////////////////////////////////////////////
ExPCE *ExPCMgr::addEntry(Int32 buckNum, 
                ULng32 probeHashVal, 
                char * probeBytes, 
                queue_index qIdxForCancel)
{
  ExProbeCacheStats *stats = tcb_->getProbeCacheStats();
  bool foundVictim = false;
  ExPCE *pce;

  while (foundVictim == false)
    {
      pce = getPossibleVictim();
      if (pce->refCnt_ == 0)
        if (pce->flags_.useBit_ == 1)
          pce->flags_.useBit_ = 0;   // give it a second chance.
        else
          foundVictim = true;
    }

  
  if (pce->flags_.everUsed_ == FALSE)
    {
      // This entry has never been added.
    }
  else
    {
      // Unlink from old collision chain
      const Int32 bucketNum = pce->probeHashVal_ % numBuckets_;
      ExPCE *pcEntryToUnlink = buckets_[bucketNum];
      ExPCE *prevPCE = NULL;
      UInt32 chainLength = 1;

      for (;;)
        {
          ex_assert(pcEntryToUnlink != NULL, "corrupt hash table.");

          // Use ptr values to make a match.
          if (pcEntryToUnlink == pce)
            {
              if (prevPCE == NULL)
                {
                  // This was the first entry in the chain.
                  buckets_[bucketNum] = pcEntryToUnlink->nextHashVal_;
                  
                  if ((stats) &&
                      (buckets_[bucketNum] == NULL))
                    stats->freeChain();
                }
              else
                {
                  prevPCE->nextHashVal_ = pcEntryToUnlink->nextHashVal_;
                }
              break;
            }
          prevPCE = pcEntryToUnlink;
          pcEntryToUnlink = pcEntryToUnlink->nextHashVal_;
          chainLength++;
        }
      if (stats)
        stats->updateLongChain(chainLength);
    }

  // Link this entry to head of its collision chain.
  if ((pce->nextHashVal_ = buckets_[buckNum]) == NULL)
    {
      // chain was empty, but now it is not.  Tell stats.
      if (stats)
          stats->newChain();
    }
  buckets_[buckNum] = pce;


  // Now initialize other members:
  pce->flags_.useBit_ = 1;  
  pce->flags_.canceledPending_ = 0;  
  pce->flags_.everUsed_ = 1;
  pce->probeHashVal_ = probeHashVal;
  pce->refCnt_ = 1;
  if (stats)
    stats->updateUseCount(1) ;
  pce->upstateStatus_ = ex_queue::Q_INVALID;
  pce->probeQueueIndex_ = qIdxForCancel;
  if (pce->diagsArea_)
    {
      pce->diagsArea_->decrRefCount();
      pce->diagsArea_ = NULL;
    }
  pce->innerRowTupp_.release();
  memcpy(pce->probeData_, probeBytes, probeLen_);

  return pce;
}

///////////////////////////////////////////////////////////////////
ExPCE *ExPCMgr::getPossibleVictim()
{
  // This method accesses the probe cache entries as an array,
  // to choose a possible victim.  The caller, addEntry, decides
  // whether the entry will be reused.  
  // Besides managing the nextVictim_ index, this method also encapsulates
  // access the array of entries, performing its own pointer arithmetic,
  // which is necessary because of the probeData_ array, the size of which
  // is not know when this C++ is compiled.
 
  // Note assumption that # buckets == # cache entries.
  if (nextVictim_ == numBuckets_)
    nextVictim_ = 0;

  ULng32 offsetToEntry = nextVictim_++ * sizeofExPCE_;
  
  return (ExPCE *) &entries_[offsetToEntry]; 
}