// **********************************************************************
// @@@ 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 "ExHbaseAccess.h"
#include "ex_exe_stmt_globals.h"

#include "ExpLOBinterface.h"

#include "SQLTypeDefs.h"

#include "ExpHbaseInterface.h"

ExHbaseScanTaskTcb::ExHbaseScanTaskTcb(
				       ExHbaseAccessSelectTcb * tcb)
  :  ExHbaseTaskTcb(tcb)
  , step_(NOT_STARTED)
{
}

void ExHbaseScanTaskTcb::init() 
{
  step_ = NOT_STARTED;
}

ExWorkProcRetcode ExHbaseScanTaskTcb::work(short &rc)
{
  Lng32 retcode = 0;
  rc = 0;
  Lng32 remainingInBatch = batchSize_;

  while (1)
    {
      ex_queue_entry *pentry_down = tcb_->qparent_.down->getHeadEntry();
      
      switch (step_)
	{
	case NOT_STARTED:
	  {
	    step_ = SCAN_OPEN;
	  }
	  break;

	case SCAN_OPEN:
	  {

	    tcb_->table_.val = tcb_->hbaseAccessTdb().getTableName();
	    tcb_->table_.len = strlen(tcb_->hbaseAccessTdb().getTableName());

	    if (tcb_->setupHbaseFilterPreds())
	      {
		step_ = HANDLE_ERROR;
		break;
	      }
	    retcode = tcb_->ehi_->scanOpen(tcb_->table_, 
					   tcb_->beginRowId_, tcb_->endRowId_,
					   tcb_->columns_, -1,
					   tcb_->hbaseAccessTdb().readUncommittedScan(),
					   tcb_->hbaseAccessTdb().getHbasePerfAttributes()->cacheBlocks(),
					   tcb_->hbaseAccessTdb().getHbasePerfAttributes()->useSmallScanner(),
					   tcb_->hbaseAccessTdb().getHbasePerfAttributes()->numCacheRows(),
					   FALSE, 
					   (tcb_->hbaseFilterColumns_.entries() > 0 ?
					    &tcb_->hbaseFilterColumns_ : NULL),
					   (tcb_->hbaseFilterOps_.entries() > 0 ?
					    &tcb_->hbaseFilterOps_ : NULL),
					   (tcb_->hbaseFilterValues_.entries() > 0 ?
					    &tcb_->hbaseFilterValues_ : NULL),
                       tcb_->hbaseAccessTdb().getHbasePerfAttributes()->dopParallelScanner(),
					   tcb_->getSamplePercentage(),
                                           FALSE, 0, NULL, NULL, 0,
                                           (tcb_->hbaseAccessTdb().getHbaseAccessOptions() 
                                            ? tcb_->hbaseAccessTdb().getHbaseAccessOptions()->getNumVersions() : 0));
                                           
	    if (tcb_->setupError(retcode, "ExpHbaseInterface::scanOpen"))
	      step_ = HANDLE_ERROR;
	    else
	      step_ = NEXT_ROW;
	  }
	  break;

	case NEXT_ROW:
 	  {
            if (--remainingInBatch <= 0)
            {
              rc = WORK_CALL_AGAIN;
              return 1;
            }

	    retcode = tcb_->ehi_->nextRow();
	    if (retcode == HBASE_ACCESS_EOD || retcode == HBASE_ACCESS_EOR)
	    {
	       step_ = SCAN_CLOSE;
	       break;
	    }
	    if (tcb_->setupError(retcode, "ExpHbaseInterface::nextRow"))
	       step_ = HANDLE_ERROR;
	    else
	       step_ = NEXT_CELL;
	  }
	  break;
	case NEXT_CELL:
          {
 	    if (tcb_->colVal_.val == NULL)
	       tcb_->colVal_.val = new (tcb_->getHeap()) 
	          char[tcb_->hbaseAccessTdb().convertRowLen()];
	    tcb_->colVal_.len = tcb_->hbaseAccessTdb().convertRowLen();        
	    retcode = tcb_->ehi_->nextCell(tcb_->rowId_, tcb_->colFamName_, 
	       tcb_->colName_, tcb_->colVal_, tcb_->colTS_);
	    if (retcode == HBASE_ACCESS_EOD)
	    {
               step_ = NEXT_ROW;
               break;
	    }
	    if (tcb_->setupError(retcode, "ExpHbaseInterface::nextCell"))
               step_ = HANDLE_ERROR;
	    else
               step_ = CREATE_ROW;
	  }
	  break;

	case CREATE_ROW:
	  {
	    rc = tcb_->createColumnwiseRow();
	    if (rc == -1)
	      {
		step_ = HANDLE_ERROR;
		break;
	      }
            else if (tcb_->setupError(rc, "ExHbaseAccessTcb::createColumnwiseRow", "Not enough space in target buffer to move data"))
              {
                step_ = HANDLE_ERROR;
                break;
              }
            
	    step_ = APPLY_PRED;
	  }
	  break;

	  case APPLY_PRED:
	  {
	    rc = tcb_->applyPred(tcb_->scanExpr());
	    if (rc == 1)
	      step_ = RETURN_ROW;
	    else if (rc == -1)
	      step_ = HANDLE_ERROR;
	    else
	      step_ = NEXT_CELL;
	  }
	  break;

	case RETURN_ROW:
	  {
	    if (tcb_->moveRowToUpQueue(tcb_->convertRow_, 
                      tcb_->hbaseAccessTdb().convertRowLen(), &rc, FALSE))
	      return 1;
	    
	    if (tcb_->getHbaseAccessStats())
	      tcb_->getHbaseAccessStats()->incUsedRows();

	    if ((pentry_down->downState.request == ex_queue::GET_N) &&
		(pentry_down->downState.requestValue == tcb_->matches_))
	      {
		step_ = SCAN_CLOSE;
		break;
	      }

	    step_ = NEXT_CELL;
	  }
	  break;

	case SCAN_CLOSE:
	  {
	    retcode = tcb_->ehi_->scanClose();
	    if (tcb_->setupError(retcode, "ExpHbaseInterface::scanClose"))
	      step_ = HANDLE_ERROR;
	    else
	      step_ = DONE;
	  }
	  break;

	case HANDLE_ERROR:
	  {
	    step_ = NOT_STARTED;
	    return -1;
	  }
	  break;

	case DONE:
	  {
	    step_ = NOT_STARTED;
	    return 0;
	  }
	  break;

	}// switch

    } // while
}

ExHbaseScanRowwiseTaskTcb::ExHbaseScanRowwiseTaskTcb(ExHbaseAccessSelectTcb * tcb)
  :  ExHbaseTaskTcb(tcb)
  , step_(NOT_STARTED)
{
}

void ExHbaseScanRowwiseTaskTcb::init() 
{
  step_ = NOT_STARTED;
}

ExWorkProcRetcode ExHbaseScanRowwiseTaskTcb::work(short &rc)
{
  Lng32 retcode = 0;
  rc = 0;
  Lng32 remainingInBatch = batchSize_;

  while (1)
    {
      ex_queue_entry *pentry_down = tcb_->qparent_.down->getHeadEntry();
      
      switch (step_)
	{
	case NOT_STARTED:
	  {
	    step_ = SCAN_OPEN;
	  }
	  break;

	case SCAN_OPEN:
	  {
	    tcb_->table_.val = tcb_->hbaseAccessTdb().getTableName();
	    tcb_->table_.len = strlen(tcb_->hbaseAccessTdb().getTableName());

	    if (tcb_->setupHbaseFilterPreds())
	      {
		step_ = HANDLE_ERROR;
		break;
	      }

	    retcode = tcb_->ehi_->scanOpen(tcb_->table_, 
					   tcb_->beginRowId_, tcb_->endRowId_,
					   tcb_->columns_, -1,
					   tcb_->hbaseAccessTdb().readUncommittedScan(),
					   tcb_->hbaseAccessTdb().getHbasePerfAttributes()->cacheBlocks(),
					   tcb_->hbaseAccessTdb().getHbasePerfAttributes()->useSmallScanner(),
					   tcb_->hbaseAccessTdb().getHbasePerfAttributes()->numCacheRows(),
					   FALSE, 
					   (tcb_->hbaseFilterColumns_.entries() > 0 ?
					    &tcb_->hbaseFilterColumns_ : NULL),
					   (tcb_->hbaseFilterOps_.entries() > 0 ?
					    &tcb_->hbaseFilterOps_ : NULL),
					   (tcb_->hbaseFilterValues_.entries() > 0 ?
					    &tcb_->hbaseFilterValues_ : NULL),
                       tcb_->hbaseAccessTdb().getHbasePerfAttributes()->dopParallelScanner(),
					   tcb_->getSamplePercentage(),
                                           FALSE, 0, NULL, NULL, 0,
                                           (tcb_->hbaseAccessTdb().getHbaseAccessOptions() 
                                            ? tcb_->hbaseAccessTdb().getHbaseAccessOptions()->getNumVersions() : 0));
                                           
	    if (tcb_->setupError(retcode, "ExpHbaseInterface::scanOpen"))
	      step_ = HANDLE_ERROR;
	    else
	      step_ = NEXT_ROW;

	    tcb_->isEOD_ = FALSE;
	  }
	  break;

	case NEXT_ROW:
	  {
            if (--remainingInBatch <= 0)
            {
              rc = WORK_CALL_AGAIN;
              return 1;
            }

	    tcb_->rowwiseRowLen_ = 0;
	    retcode = tcb_->ehi_->nextRow();
	    if (retcode == HBASE_ACCESS_EOD || retcode == HBASE_ACCESS_EOR)
	    {
	       step_ = SCAN_CLOSE;
	       break;
	    }
	    if (tcb_->setupError(retcode, "ExpHbaseInterface::nextRow"))
	       step_ = HANDLE_ERROR;
	    else
	       step_ = NEXT_CELL;
	  }
	  break;

	case NEXT_CELL:
	  {
	    if (tcb_->colVal_.val == NULL)
	       tcb_->colVal_.val = new (tcb_->getHeap()) 
 	       char[tcb_->hbaseAccessTdb().convertRowLen()];
	    tcb_->colVal_.len = tcb_->hbaseAccessTdb().convertRowLen();        
	    retcode = tcb_->ehi_->nextCell(tcb_->rowId_, tcb_->colFamName_, 
					    tcb_->colName_, tcb_->colVal_,
					    tcb_->colTS_);
	    if (retcode == HBASE_ACCESS_EOD)
	    { 
	       step_ = CREATE_ROW;
	       break;
	    }
	    if (tcb_->setupError(retcode, "ExpHbaseInterface::nextCell"))
               step_ = HANDLE_ERROR;
	    else
	       step_ = APPEND_ROW;
	  }
	  break;

	case APPEND_ROW:
	  {
	    retcode = tcb_->copyCell();
            if (tcb_->setupError(retcode, "ExHbaseAccessTcb::copyCell", "Not enough space in target buffer to move data"))
              step_ = HANDLE_ERROR;
            else
              step_ = NEXT_CELL;
	  }
	  break;

	case CREATE_ROW:
	  {
	    rc = tcb_->createRowwiseRow();
	    if (rc == -1)
	      {
		step_ = HANDLE_ERROR;
		break;
	      }
	    
	    step_ = APPLY_PRED;
	  }
	  break;

	case APPLY_PRED:
	  {
	    rc = tcb_->applyPred(tcb_->scanExpr());

	    if (rc == 1)
	      {
		step_ = RETURN_ROW;
		break;
	      }
	    else if (rc == -1)
	      {
		step_ = HANDLE_ERROR;
		break;
	      }
	    else if (tcb_->isEOD_)
	      step_ = SCAN_CLOSE;
	    else
	      step_ = NEXT_ROW;
	  }
	  break;

	case RETURN_ROW:
	  {
	    rc = 0;
	    if (tcb_->moveRowToUpQueue(tcb_->convertRow_, 
                   tcb_->hbaseAccessTdb().convertRowLen(), &rc, FALSE))
	      return 1;

	    if (tcb_->getHbaseAccessStats())
	      tcb_->getHbaseAccessStats()->incUsedRows();

	    if ((pentry_down->downState.request == ex_queue::GET_N) &&
		(pentry_down->downState.requestValue == tcb_->matches_))
	      {
		step_ = SCAN_CLOSE;
		break;
	      }
             step_ = NEXT_ROW;
	  }
	  break;

	case SCAN_CLOSE:
	  {
	    retcode = tcb_->ehi_->scanClose();
	    if (tcb_->setupError(retcode, "ExpHbaseInterface::scanClose"))
	      step_ = HANDLE_ERROR;
	    else
	      step_ = DONE;
	  }
	  break;

	case HANDLE_ERROR:
	  {
	    step_ = NOT_STARTED;
	    return -1;
	  }
	  break;

	case DONE:
	  {
	    step_ = NOT_STARTED;
	    return 0;
	  }
	  break;

	}// switch

    } // while

  return 0; // WORK_OK
}

ExHbaseScanSQTaskTcb::ExHbaseScanSQTaskTcb(
				       ExHbaseAccessSelectTcb * tcb)
  :  ExHbaseTaskTcb(tcb)
  , step_(NOT_STARTED)
{
}

void ExHbaseScanSQTaskTcb::init() 
{
  step_ = NOT_STARTED;
}

ExWorkProcRetcode ExHbaseScanSQTaskTcb::work(short &rc)
{
  Lng32 retcode = 0;
  rc = 0;
  Lng32 remainingInBatch = batchSize_;
  NABoolean isFirstBatch = false;
  // isFirstInBatch is a stack variable for optimization reason. It is used for the mdam small scanner optimization heuristic that
  // is performed at runtime. Since this function is invoke intensively for all scan (mdam or regular scan), minimizing CPU/memory access
  // impact on runtime code to a strict minimum is attempted. Given that we are trying to detect if the actual scan is bellow the size
  // of an HBase block, having the runtime logic performing the detection only affect the first work invoke looks like the right idea.
  // and leveraging an existing counter (remainingInBatch) instead of creating a new one. The reasonable asumption to allow this is that
  // 1- batchSize_ being 8K, most likely times the row size, we are good in assuming that first hbase block will fit in batchSize
  // 2- parent buffer size will be large enough to deal with one HBAse_Block_size without having to rely on re-invoking work in the middle.
  // and anyway, if none of the reasonable assumption is true, then that's fine, the heuristic won't work, and we will use regular scanner,
  // meaning optimization is off for the scan part of MDAM (still on for the probe side of it).

  while (1)
    {
      ex_queue_entry *pentry_down = tcb_->qparent_.down->getHeadEntry();
      
      switch (step_)
	{
	case NOT_STARTED:
	  {
	    step_ = SCAN_OPEN;
	  }
	  break;

	case SCAN_OPEN:
	  {
	    tcb_->table_.val = tcb_->hbaseAccessTdb().getTableName();
	    tcb_->table_.len = strlen(tcb_->hbaseAccessTdb().getTableName());

	    if (tcb_->setupHbaseFilterPreds())
	      {
		step_ = HANDLE_ERROR;
		break;
	      }
	    isFirstBatch = true;
	    retcode = tcb_->ehi_->scanOpen(tcb_->table_, 
					   tcb_->beginRowId_, tcb_->endRowId_,
					   tcb_->columns_, -1,
					   tcb_->hbaseAccessTdb().readUncommittedScan(),
					   tcb_->hbaseAccessTdb().getHbasePerfAttributes()->cacheBlocks(),
					   tcb_->hbaseAccessTdb().getHbasePerfAttributes()->useSmallScanner(),
					   tcb_->hbaseAccessTdb().getHbasePerfAttributes()->numCacheRows(),
                                           TRUE,
					   (tcb_->hbaseFilterColumns_.entries() > 0 ?
					    &tcb_->hbaseFilterColumns_ : NULL),
					   (tcb_->hbaseFilterOps_.entries() > 0 ?
					    &tcb_->hbaseFilterOps_ : NULL),
					   (tcb_->hbaseFilterValues_.entries() > 0 ?
					    &tcb_->hbaseFilterValues_ : NULL),
                       tcb_->hbaseAccessTdb().getHbasePerfAttributes()->dopParallelScanner(),
                                           tcb_->getSamplePercentage(),
                                           tcb_->hbaseAccessTdb().getHbaseSnapshotScanAttributes()->getUseSnapshotScan(),
                                           tcb_->hbaseAccessTdb().getHbaseSnapshotScanAttributes()->getSnapshotScanTimeout(),
                                           tcb_->hbaseAccessTdb().getHbaseSnapshotScanAttributes()->getSnapshotName(),
                                           tcb_->hbaseAccessTdb().getHbaseSnapshotScanAttributes()->getSnapScanTmpLocation(),
                                           tcb_->getGlobals()->castToExExeStmtGlobals()->getMyInstanceNumber(),
                                           (tcb_->hbaseAccessTdb().multiVersions()
                                            ? tcb_->hbaseAccessTdb().getHbaseAccessOptions()->getNumVersions() : 0)
                                           );

	    if (tcb_->setupError(retcode, "ExpHbaseInterface::scanOpen"))
	      step_ = HANDLE_ERROR;
	    else
	      step_ = NEXT_ROW;
	  }
	  break;

	case NEXT_ROW:
	  {
            if (--remainingInBatch <= 0)
            {
              rc = WORK_CALL_AGAIN;
              return 1;
            }
	    retcode = tcb_->ehi_->nextRow();
	    if (retcode == HBASE_ACCESS_EOD || retcode == HBASE_ACCESS_EOR)
	      {
		step_ = SCAN_CLOSE;
		break;
	      }
	    if (tcb_->setupError(retcode, "ExpHbaseInterface::nextRow"))
               step_ = HANDLE_ERROR;
            else if (tcb_->hbaseAccessTdb().multiVersions())
              step_ = SETUP_MULTI_VERSION_ROW;
            else
	       step_ = CREATE_ROW;
	  }
	  break;

        case SETUP_MULTI_VERSION_ROW:
          {
	    retcode = tcb_->setupSQMultiVersionRow();
	    if (retcode == HBASE_ACCESS_NO_ROW)
              {
	        step_ = NEXT_ROW;
	        break;
              }

	    if (retcode < 0)
              {
	        rc = (short)retcode;
	        tcb_->setupError(rc, "setupSQMultiVersionRow");
	        step_ = HANDLE_ERROR;
	        break;
              }

            step_ = CREATE_ROW;
          }
          break;

	case CREATE_ROW:
	  {
	    retcode = tcb_->createSQRowDirect();
	    if (retcode == HBASE_ACCESS_NO_ROW)
	    {
	        step_ = NEXT_ROW;
	        break;
	    }
	    if (retcode < 0)
	    {
	        rc = (short)retcode;
	        tcb_->setupError(rc, "createSQRowDirect");
	        step_ = HANDLE_ERROR;
	        break;
	    }
	    if (retcode != HBASE_ACCESS_SUCCESS)
	    {
	        step_ = HANDLE_ERROR;
	        break;
	    }

	    step_ = APPLY_PRED;
	  }
	  break;

	  case APPLY_PRED:
	  {
	    rc = tcb_->applyPred(tcb_->scanExpr());
	    if (rc == 1)
	      step_ = RETURN_ROW;
	    else if (rc == -1)
	      step_ = HANDLE_ERROR;
	    else if (tcb_->hbaseAccessTdb().multiVersions())
              step_ = CREATE_ROW;
            else
              step_ = NEXT_ROW;
	  }
	  break;

	case RETURN_ROW:
	  {
	    if (tcb_->moveRowToUpQueue(tcb_->convertRow_, tcb_->convertRowLen_, 
				       &rc, FALSE))
	      return 1;
	    
	    if (tcb_->getHbaseAccessStats())
	      tcb_->getHbaseAccessStats()->incUsedRows();
	    
	    if ((pentry_down->downState.request == ex_queue::GET_N) &&
		(pentry_down->downState.requestValue == tcb_->matches_))
	      {
		step_ = SCAN_CLOSE;
		break;
	      }

            if (tcb_->hbaseAccessTdb().multiVersions())
              step_ = CREATE_ROW;
            else
              step_ = NEXT_ROW;
	  }
	  break;

	case SCAN_CLOSE:
	  {
	      if (isFirstBatch) //only if closed happen in a single batch, batchSize - remainingInBatch = nb rows retrieved
	          tcb_->hbaseAccessTdb().getHbasePerfAttributes()->setUseSmallScannerForMDAMifNeeded(batchSize_ - remainingInBatch); //calculate MDAM small scanner flag for next scan if it was MDAM
	    retcode = tcb_->ehi_->scanClose();
	    if (tcb_->setupError(retcode, "ExpHbaseInterface::scanClose"))
	      step_ = HANDLE_ERROR;
	    else
	      step_ = DONE;
	  }
	  break;

	case HANDLE_ERROR:
	  {
	    step_ = NOT_STARTED;
	    return -1;
	  }
	  break;

	case DONE:
	  {
	    step_ = NOT_STARTED;
	    return 0;
	  }
	  break;

	}// switch

    } // while
}

Lng32 ExHbaseScanSQTaskTcb::getProbeResult(char* &keyData)
{
  Lng32 retcode = 0;
  Lng32 rc = 0;
  Lng32 probeSize = 100; // using fewer rows results in intermittent wrong 
  //results. Using the hbase default scan size of 100 as a workarorund.
  if (tcb_->hbaseAccessTdb().getHbasePerfAttributes()->useMinMdamProbeSize())
    probeSize = 1; // if performance is vital, comp_bool_184 can be set to ON 
  // to choose this path.

  retcode = tcb_->ehi_->scanOpen(tcb_->table_, 
				 tcb_->beginRowId_, tcb_->endRowId_,
				 tcb_->columns_, -1,
				 tcb_->hbaseAccessTdb().readUncommittedScan(),
				 tcb_->hbaseAccessTdb().getHbasePerfAttributes()->cacheBlocks() ||
				 tcb_->hbaseAccessTdb().getHbasePerfAttributes()->useSmallScannerForProbes(), // when small scanner feature is ON or SYSTEM force cache ON
				 tcb_->hbaseAccessTdb().getHbasePerfAttributes()->useSmallScannerForProbes(),
				 probeSize,
				 TRUE, NULL, NULL, NULL);
  if (tcb_->setupError(retcode, "ExpHbaseInterface::scanOpen"))
    {
      rc = -1;
      goto label_return;
    }
  
  retcode = tcb_->ehi_->nextRow();
  if (retcode == HBASE_ACCESS_EOD || retcode == HBASE_ACCESS_EOR)
    {
      rc = 1; // no row found
      goto label_return;
    }
  if (tcb_->setupError(retcode, "ExpHbaseInterface::nextRow"))
    {
      rc = -1;
      goto label_return;
    }
  retcode = tcb_->createSQRowDirect();
  if (retcode == HBASE_ACCESS_NO_ROW)
  {
     rc = 1;
     goto label_return;
  }
  if (retcode < 0)
  {
      rc = retcode;
      tcb_->setupError(rc, "createSQRowDirect");
      rc = -1;
      goto label_return;
  }
  if (retcode != HBASE_ACCESS_SUCCESS)
  {
     rc = -1;
     goto label_return;
  }

  // extract the key from the fetched row, encode it and pass it back to mdam
  if (tcb_->evalEncodedKeyExpr() == -1)
    {
      rc = -1;
      goto label_return;
    }
  
 label_return:
  retcode = tcb_->ehi_->scanClose();
  if (tcb_->setupError(retcode, "ExpHbaseInterface::scanClose"))
    {
      rc = -1;
    }

  keyData = tcb_->encodedKeyRow_;

  return rc;
}

ExHbaseGetTaskTcb::ExHbaseGetTaskTcb(
				     ExHbaseAccessSelectTcb * tcb)
  :  ExHbaseTaskTcb(tcb)
  , step_(NOT_STARTED)
{
}

void ExHbaseGetTaskTcb::init() 
{
  step_ = NOT_STARTED;
}

ExWorkProcRetcode ExHbaseGetTaskTcb::work(short &rc)
{
  Lng32 retcode = 0;
  rc = 0;
  Lng32 remainingInBatch = batchSize_;

  while (1)
    {
      ex_queue_entry *pentry_down = tcb_->qparent_.down->getHeadEntry();
      
      switch (step_)
	{
	case NOT_STARTED:
	  {
	    step_ = GET_OPEN;
	  }
	  break;

	case GET_OPEN:
	  {
	    tcb_->table_.val = tcb_->hbaseAccessTdb().getTableName();
	    tcb_->table_.len = strlen(tcb_->hbaseAccessTdb().getTableName());

	    if (tcb_->evalRowIdExpr(TRUE) == -1)
	      {
		step_ = HANDLE_ERROR;
		break;
	      }

	    if (tcb_->rowIds_.entries() == 1)
	      {
		retcode = tcb_->ehi_->getRowOpen(tcb_->table_, tcb_->rowIds_[0],
					     tcb_->columns_, -1);
		if (tcb_->setupError(retcode, "ExpHbaseInterface::getRowOpen"))
		  step_ = HANDLE_ERROR;
		else
		  step_ = NEXT_ROW;
	      }
	    else
	      {
		retcode = tcb_->ehi_->getRowsOpen(tcb_->table_, &tcb_->rowIds_,
					     tcb_->columns_, -1);
		if (tcb_->setupError(retcode, "ExpHbaseInterface::getRowsOpen"))
		  step_ = HANDLE_ERROR;
		else
		  step_ = NEXT_ROW;
	      }
	      
	  }
	  break;

	case NEXT_ROW:
	  {       
            if (--remainingInBatch <= 0)
            {
              rc = WORK_CALL_AGAIN;
              return 1;
            }
	    retcode = tcb_->ehi_->nextRow();
	    if (retcode == HBASE_ACCESS_EOD || retcode == HBASE_ACCESS_EOR)
	    {
 	       step_ = GET_CLOSE;
	       break;
	    }
	    if (tcb_->setupError(retcode, "ExpHbaseInterface::nextRow"))
	       step_ = HANDLE_ERROR;
	    else
	       step_ = NEXT_CELL;
	  }
	  break;

	case NEXT_CELL:
	  {
	    if (tcb_->colVal_.val == NULL)
	        tcb_->colVal_.val = new (tcb_->getHeap()) 
	        char[tcb_->hbaseAccessTdb().convertRowLen()];
	    tcb_->hbaseAccessTdb().convertRowLen();
	    retcode = tcb_->ehi_->nextCell( tcb_->rowId_, tcb_->colFamName_, 
	       tcb_->colName_, tcb_->colVal_, tcb_->colTS_);
	    if (retcode == HBASE_ACCESS_EOD)
	    {
		step_ = NEXT_ROW;
		break;
	    }
	    if (tcb_->setupError(retcode, "ExpHbaseInterface::nextCell"))
	      step_ = HANDLE_ERROR;
	    else
	      step_ = CREATE_ROW;
	  }
	  break;

	case CREATE_ROW:
	  {
	    rc = tcb_->createColumnwiseRow();
	    if (rc == -1)
	      {
		step_ = HANDLE_ERROR;
		break;
	      }
            else if (tcb_->setupError(rc, "ExHbaseAccessTcb::createColumnwiseRow", "Not enough space in target buffer to move data"))
              {
                step_ = HANDLE_ERROR;
                break;
              }
	    
	    step_ = APPLY_PRED;
	  }
	  break;

	case APPLY_PRED:
	  {
	    rc = tcb_->applyPred(tcb_->scanExpr());

	    if (rc == 1)
	      step_ = RETURN_ROW;
	    else if (rc == -1)
	      step_ = HANDLE_ERROR;
	    else
	      step_ = NEXT_CELL;
	  }
	  break;

	case RETURN_ROW:
	  {
	    rc = 0;
	    if (tcb_->moveRowToUpQueue(tcb_->convertRow_, tcb_->hbaseAccessTdb().convertRowLen(), 
				       &rc, FALSE))
	      return 1;

	    if (tcb_->getHbaseAccessStats())
	      tcb_->getHbaseAccessStats()->incUsedRows();

	    if ((pentry_down->downState.request == ex_queue::GET_N) &&
		(pentry_down->downState.requestValue == tcb_->matches_))
	      {
		step_ = GET_CLOSE;
		break;
	      }

	    step_ = NEXT_CELL;
	  }
	  break;

	case GET_CLOSE:
	  {
	    retcode = tcb_->ehi_->getClose();
	    if (tcb_->setupError(retcode, "ExpHbaseInterface::getClose"))
	      step_ = HANDLE_ERROR;
	    else
	      step_ = DONE;
	  }
	  break;

	case HANDLE_ERROR:
	  {
	    step_ = NOT_STARTED;
	    return -1;
	  }
	  break;
	    
	case DONE:
	  {
	    step_ = NOT_STARTED;
	    return 0;
	  }
	  break;

	}// switch

    } // while
}

ExHbaseGetRowwiseTaskTcb::ExHbaseGetRowwiseTaskTcb(
				       ExHbaseAccessSelectTcb * tcb)
  :  ExHbaseTaskTcb(tcb)
  , step_(NOT_STARTED)
{
}

void ExHbaseGetRowwiseTaskTcb::init() 
{
  step_ = NOT_STARTED;
}

ExWorkProcRetcode ExHbaseGetRowwiseTaskTcb::work(short &rc)
{
  Lng32 retcode = 0;
  rc = 0;
  Lng32 remainingInBatch = batchSize_;

  while (1)
    {
      ex_queue_entry *pentry_down = tcb_->qparent_.down->getHeadEntry();
      
      switch (step_)
	{
	case NOT_STARTED:
	  {
	    step_ = GET_OPEN;
	  }
	  break;

	case GET_OPEN:
	  {
	    tcb_->table_.val = tcb_->hbaseAccessTdb().getTableName();
	    tcb_->table_.len = strlen(tcb_->hbaseAccessTdb().getTableName());

	    if (tcb_->evalRowIdExpr(TRUE) == -1)
	      {
		step_ = HANDLE_ERROR;
		break;
	      }

	    if (tcb_->rowIds_.entries() == 1)
	      {
		retcode = tcb_->ehi_->getRowOpen(tcb_->table_, tcb_->rowIds_[0],
					     tcb_->columns_, -1);
		if (tcb_->setupError(retcode, "ExpHbaseInterface::getRowOpen"))
		  step_ = HANDLE_ERROR;
		else
		  step_ = NEXT_ROW;
	      }
	    else
	      {
		retcode = tcb_->ehi_->getRowsOpen(tcb_->table_, &tcb_->rowIds_,
					     tcb_->columns_, -1);
		if (tcb_->setupError(retcode, "ExpHbaseInterface::getRowsOpen"))
		  step_ = HANDLE_ERROR;
		else
		  step_ = NEXT_ROW;
	      }

	  }
	  break;

	case NEXT_ROW:
	  {
            if (--remainingInBatch <= 0)
            {
              rc = WORK_CALL_AGAIN;
              return 1;
            }

	    retcode = tcb_->ehi_->nextRow();
	    if (retcode == HBASE_ACCESS_EOD || retcode == HBASE_ACCESS_EOR)
	    {
	       step_ = GET_CLOSE;
	       break;
	    }
	    if (tcb_->setupError(retcode, "ExpHbaseInterface::nextRow"))
	       step_ = HANDLE_ERROR;
	    else
	       step_ = NEXT_CELL;
	  }
	  break;

	case NEXT_CELL:
	  {
	     if (tcb_->colVal_.val == NULL)
	        tcb_->colVal_.val = new (tcb_->getHeap()) 
		   char[tcb_->hbaseAccessTdb().convertRowLen()];
	    tcb_->colVal_.len = tcb_->hbaseAccessTdb().convertRowLen();
	    retcode = tcb_->ehi_->nextCell(tcb_->rowId_, tcb_->colFamName_, 
	        tcb_->colName_, tcb_->colVal_, tcb_->colTS_);
	    if (retcode == HBASE_ACCESS_EOD)
	       step_ = CREATE_ROW;
	    else
	    if (tcb_->setupError(retcode, "ExpHbaseInterface::nextCell"))
	       step_ = HANDLE_ERROR;
	    else
	      step_ = APPEND_ROW;
	  }
	  break;

	case APPEND_ROW:
	  {
	    retcode = tcb_->copyCell();
            if (tcb_->setupError(retcode, "ExHbaseAccessTcb::copyCell", "Not enough space in target buffer to move data"))
              step_ = HANDLE_ERROR;
            else
              step_ = NEXT_CELL;
	  }
	  break;

	case CREATE_ROW:
	  {
	    rc = tcb_->createRowwiseRow();
	    if (rc == -1)
	      {
		step_ = HANDLE_ERROR;
		break;
	      }

	    step_ = APPLY_PRED;
	  }
	  break;

	case APPLY_PRED:
	  {
	    rc = tcb_->applyPred(tcb_->scanExpr());

	    if (rc == 1)
	      step_ = RETURN_ROW;
	    else if (rc == -1)
	      step_ = HANDLE_ERROR;
	    else
	      step_ = GET_CLOSE;
	  }
	  break;

	case RETURN_ROW:
	  {
	    rc = 0;
	    if (tcb_->moveRowToUpQueue(tcb_->convertRow_, tcb_->hbaseAccessTdb().convertRowLen(), 
				 &rc, FALSE))
	      return 1;
	    
	    if (tcb_->getHbaseAccessStats())
	      tcb_->getHbaseAccessStats()->incUsedRows();

	    step_ = GET_CLOSE;
	  }
	  break;

	case GET_CLOSE:
	  {
	    retcode = tcb_->ehi_->getClose();
	    if (tcb_->setupError(retcode, "ExpHbaseInterface::getClose"))
	      step_ = HANDLE_ERROR;
	    else
	      step_ = DONE;
	  }
	  break;

	case HANDLE_ERROR:
	  {
	    step_ = NOT_STARTED;
	    return -1;
	  }
	  break;
	    
	case DONE:
	  {
	    step_ = NOT_STARTED;
	    return 0;
	  }
	  break;

	}// switch

    } // while
}

ExHbaseGetSQTaskTcb::ExHbaseGetSQTaskTcb( ExHbaseAccessTcb * tcb, NABoolean rowsetTcb)
  :  ExHbaseTaskTcb(tcb, rowsetTcb)
  , step_(NOT_STARTED)
{
}

void ExHbaseGetSQTaskTcb::init() 
{
  step_ = NOT_STARTED;
}

ExWorkProcRetcode ExHbaseGetSQTaskTcb::work(short &rc)
{
  Lng32 retcode = 0;
  rc = 0;

  while (1)
    {
      
      ex_queue_entry *pentry_down = NULL;
      if (! tcb_->qparent_.down->isEmpty())
          pentry_down = tcb_->qparent_.down->getHeadEntry();
      
      switch (step_)
	{
	case NOT_STARTED:
	  {
	    step_ = GET_OPEN;
	  }
	  break;

	case GET_OPEN:
	  {
	    tcb_->table_.val = tcb_->hbaseAccessTdb().getTableName();
	    tcb_->table_.len = strlen(tcb_->hbaseAccessTdb().getTableName());
            remainingInBatch_ = tcb_->rowIds_.entries();
	    if (tcb_->rowIds_.entries() == 1)
	      {
		retcode = tcb_->ehi_->getRowOpen(tcb_->table_, tcb_->rowIds_[0],
					     tcb_->columns_, -1);
		if (tcb_->setupError(retcode, "ExpHbaseInterface::getRowOpen"))
		  step_ = HANDLE_ERROR;
		else
		  step_ = NEXT_ROW;
	      }
	    else
	      {
		retcode = tcb_->ehi_->getRowsOpen(tcb_->table_, &tcb_->rowIds_,
					     tcb_->columns_, -1);
		if (tcb_->setupError(retcode, "ExpHbaseInterface::getRowsOpen"))
		  step_ = HANDLE_ERROR;
		else
		  step_ = NEXT_ROW;
	      }
	  }
	  break;

	case NEXT_ROW:
	  {
            if (remainingInBatch_ <= 0) {
               step_ = GET_CLOSE;
               break;
            }
	    retcode = tcb_->ehi_->nextRow();
            remainingInBatch_--;
            // EOD is end of data, EOR is end of result set. 
            // for single get, EOD or EOR indicates DONE
            // for multi get, only EOR indicates DONE
	    if ( (retcode == HBASE_ACCESS_EOR) ||
                 ( (retcode == HBASE_ACCESS_EOD) &&
                   (tcb_->rowIds_.entries() == 1) ) )
	      {
                if (rowsetTcb_)
                   step_ = DONE;
                else
		   step_ = GET_CLOSE;
		break;
	      }
            
            // for multi get, do FETCH if retcode is EOD
            if ( (retcode == HBASE_ACCESS_EOD) && 
                 (tcb_->rowIds_.entries() > 1) )
            {
                if (rowsetTcb_)
                   step_ = DONE;
                else
                   step_ = NEXT_ROW;
                break;
            }
	    if (tcb_->setupError(retcode, "ExpHbaseInterface::nextRow"))
	      step_ = HANDLE_ERROR;
	    else
	      step_ = CREATE_ROW;
	  }
	  break;

	case CREATE_ROW:
	  {
	    retcode = tcb_->createSQRowDirect();
	    if (retcode == HBASE_ACCESS_NO_ROW)
	    {
               if (rowsetTcb_)
                  step_ = DONE;
               else
	          step_ = NEXT_ROW;
	       break;
	    }
	    if (retcode < 0)
	    {
	        rc = (short)retcode;
	        tcb_->setupError(rc, "createSQRowDirect");
	        step_ = HANDLE_ERROR;
	        break;
	    }  
	    if (retcode != HBASE_ACCESS_SUCCESS)
	    {
	        step_ = HANDLE_ERROR;
	        break;
	    }
	    step_ = APPLY_PRED;
	  }
	  break;

	case APPLY_PRED:
	  {
	    rc = tcb_->applyPred(tcb_->scanExpr());

	    if (rc == 1)
	      step_ = RETURN_ROW;
	    else if (rc == -1)
	      step_ = HANDLE_ERROR;
	    else
            {
              if (rowsetTcb_)
                 step_ = DONE;
              else
	         step_ = NEXT_ROW;
            }
	  }
	  break;
	case RETURN_ROW:
	  {
	    rc = 0;
	    if (tcb_->moveRowToUpQueue(tcb_->convertRow_, tcb_->hbaseAccessTdb().convertRowLen(), 
				       &rc, FALSE))
	      return 1;
	    if (tcb_->getHbaseAccessStats())
	      tcb_->getHbaseAccessStats()->incUsedRows();
            if (rowsetTcb_)
                step_ = DONE;
            else
            {
	      if ((pentry_down->downState.request == ex_queue::GET_N) &&
		(pentry_down->downState.requestValue == tcb_->matches_))
	      {
		step_ = GET_CLOSE;
		break;
	      }
	      step_ = NEXT_ROW;
            }
	  }
	  break;

	case GET_CLOSE:
	  {
	    retcode = tcb_->ehi_->getClose();
	    if (tcb_->setupError(retcode, "ExpHbaseInterface::getClose"))
	      step_ = HANDLE_ERROR;
	    else
	      step_ = ALL_DONE;
	  }
	  break;

	case HANDLE_ERROR:
	  {
	    step_ = NOT_STARTED;
	    return -1;
	  }
	  break;
	    
	case DONE:
           if (tcb_->handleDone(rc, 0))
               return 1;
           else
               step_ = NEXT_ROW;
           break;
        case ALL_DONE:
	   step_ = NOT_STARTED;
	   return 0;
	   break;
	}// switch

    } // while
    return 0;
}

ExHbaseAccessSelectTcb::ExHbaseAccessSelectTcb(
          const ExHbaseAccessTdb &hbaseAccessTdb, 
          ex_globals * glob ) :
  ExHbaseAccessTcb(hbaseAccessTdb, glob),
  step_(NOT_STARTED)
{
  scanRowwiseTaskTcb_ = NULL;
  scanTaskTcb_ = NULL;
  getRowwiseTaskTcb_ = NULL;
  getTaskTcb_ = NULL;
  scanSQTaskTcb_ = NULL;
  getSQTaskTcb_ = NULL;

  ExHbaseAccessTdb &hbaseTdb = (ExHbaseAccessTdb&)hbaseAccessTdb;
  samplePercentage_ = hbaseTdb.samplingRate_;

  if ((hbaseTdb.listOfScanRows()) ||
      ((hbaseTdb.keySubsetGen()) &&
       (NOT hbaseTdb.uniqueKeyInfo())) ||
      (hbaseTdb.keyMDAMGen()))
    {
      if (hbaseTdb.sqHbaseTable())
	scanSQTaskTcb_ = 
	  new(getGlobals()->getDefaultHeap()) ExHbaseScanSQTaskTcb(this);
      else if (hbaseTdb.rowwiseFormat())
	scanRowwiseTaskTcb_ = 
	  new(getGlobals()->getDefaultHeap()) ExHbaseScanRowwiseTaskTcb(this);
      else
	scanTaskTcb_ = 
	  new(getGlobals()->getDefaultHeap()) ExHbaseScanTaskTcb(this);
    }

  if ((hbaseTdb.listOfGetRows()) ||
      ((hbaseTdb.keySubsetGen()) &&
       (hbaseTdb.uniqueKeyInfo())) ||
      (hbaseTdb.keyMDAMGen()))
    {
      if (hbaseTdb.sqHbaseTable())
	getSQTaskTcb_ = 
	  new(getGlobals()->getDefaultHeap()) ExHbaseGetSQTaskTcb(this, FALSE);
      else if (hbaseTdb.rowwiseFormat())
	getRowwiseTaskTcb_ = 
	  new(getGlobals()->getDefaultHeap()) ExHbaseGetRowwiseTaskTcb(this);
      else
	getTaskTcb_ = 
	  new(getGlobals()->getDefaultHeap()) ExHbaseGetTaskTcb(this);
     }

  if (hbaseTdb.sqHbaseTable())
    {
      scanTask_ = scanSQTaskTcb_;
      getTask_   = getSQTaskTcb_;
    }
  else if (hbaseTdb.rowwiseFormat())
    {
      scanTask_ = scanRowwiseTaskTcb_;
      getTask_   = getRowwiseTaskTcb_;
    }
  else
    {
      scanTask_ = scanTaskTcb_;
      getTask_   = getTaskTcb_;
    }
}

ExWorkProcRetcode ExHbaseAccessSelectTcb::work()
{
  Lng32 retcode = 0;
  short rc = 0;

  while (!qparent_.down->isEmpty())
    {
      ex_queue_entry *pentry_down = qparent_.down->getHeadEntry();
      if ((pentry_down->downState.request == ex_queue::GET_NOMORE) &&
	  (step_ != DONE))
	{
	  step_ = SELECT_CLOSE_NO_ERROR; //DONE;
	}

      switch (step_)
	{
	case NOT_STARTED:
	  {
	    matches_ = 0;

	    if ((pentry_down->downState.request == ex_queue::GET_N) &&
		(pentry_down->downState.requestValue == matches_))
	      {
		step_ = DONE;
		break;
	      }

	    step_ = SELECT_INIT;
	  }
	  break;

	case SELECT_INIT:
	  {
	    retcode = ehi_->init(getHbaseAccessStats());
	    if (setupError(retcode, "ExpHbaseInterface::init"))
	      {
		step_ = HANDLE_ERROR;
		break;
	      }

	    if (hbaseAccessTdb().listOfScanRows())
	      hbaseAccessTdb().listOfScanRows()->position();

	    if (hbaseAccessTdb().listOfGetRows())
	      hbaseAccessTdb().listOfGetRows()->position();

	    if (scanTask_)
	      scanTask_->init();

	    if (getTask_)
	      getTask_->init();
	    
	    step_ = SETUP_SCAN;
	  }
	  break;

	case SETUP_SCAN:
	  {
	    if ((! scanTask_) || (! hbaseAccessTdb().listOfScanRows()))
	      {
		step_ = SETUP_GET;
		break;
	      }

	    hsr_ = 
	      (ComTdbHbaseAccess::HbaseScanRows*)hbaseAccessTdb().listOfScanRows()
	      ->getCurr();

	    retcode = setupSubsetRowIdsAndCols(hsr_);
	    if (retcode == -1)
	      {
		step_ = HANDLE_ERROR;
		break;
	      }

	    step_ = PROCESS_SCAN;
	  }
	  break;

	case PROCESS_SCAN:
	case PROCESS_SCAN_KEY:
	  {
	    rc = 0;
	    retcode = scanTask_->work(rc);

	    if (retcode == 1)
	      return rc;
	    else if (retcode < 0)
	      step_ = HANDLE_ERROR;
	    else if (step_ == PROCESS_SCAN_KEY)
	      step_ = SETUP_GET_KEY;
	    else
	      step_ = NEXT_SCAN;
	  }
	  break;

	case NEXT_SCAN:
	  {
	    hbaseAccessTdb().listOfScanRows()->advance();

	    if (! hbaseAccessTdb().listOfScanRows()->atEnd())
	      {
		step_ = SETUP_SCAN;
		break;
	      }

	    step_ = SETUP_GET;
	  }
	  break;

	case SETUP_GET:
	  {
	    if ((! getTask_) || (!hbaseAccessTdb().listOfGetRows()))
	      {
		step_ = SETUP_SCAN_KEY;
		break;
	      }

	    hgr_ = 
	      (ComTdbHbaseAccess::HbaseGetRows*)hbaseAccessTdb().listOfGetRows()
	      ->getCurr();

	    retcode = setupUniqueRowIdsAndCols(hgr_);
	    if (retcode == -1)
	      {
		step_ = HANDLE_ERROR;
		break;
	      }

	    step_ = PROCESS_GET;
	  }
	  break;

	case PROCESS_GET:
	case PROCESS_GET_KEY:
	  {
	    rc = 0;
	    retcode = getTask_->work(rc);
	    
	    if (retcode == 1)
	      return rc;
	    else if (retcode < 0)
	      step_ = HANDLE_ERROR;
	    else if (step_ == PROCESS_GET_KEY)
	      step_ = SELECT_CLOSE;
	    else
	      step_ = NEXT_GET;
	  }
	  break;

	case NEXT_GET:
	  {
	    hbaseAccessTdb().listOfGetRows()->advance();

	    if (! hbaseAccessTdb().listOfGetRows()->atEnd())
	      {
		step_ = SETUP_GET;
		break;
	      }

	    step_ = SETUP_SCAN_KEY;
	  }
	  break;

	case SETUP_SCAN_KEY:
	  {
	    if (! hbaseAccessTdb().keySubsetGen())
	      {
		step_ = SELECT_CLOSE;
		break;
	      }

	    if (hbaseAccessTdb().uniqueKeyInfo())
	      {
		step_ = SETUP_GET_KEY;
		break;
	      }
	    
	    retcode = setupSubsetKeysAndCols();
	    if (retcode == -1)
	      {
		step_ = HANDLE_ERROR;
		break;
	      }

	    step_ = PROCESS_SCAN_KEY;
	  }
	  break;

	case SETUP_GET_KEY:
	  {
	    if ((! getTask_) ||
		((hbaseAccessTdb().keySubsetGen()) &&
		 (NOT hbaseAccessTdb().uniqueKeyInfo())))
	      {
		step_ = SELECT_CLOSE;
		break;
	      }

	    retcode = setupUniqueKeyAndCols(TRUE);
	    if (retcode == -1)
	      {
		step_ = HANDLE_ERROR;
		break;
	      }

	    step_ = PROCESS_GET_KEY;
	  }
	  break;

	case SELECT_CLOSE:
	case SELECT_CLOSE_NO_ERROR:
	  {
	    retcode = ehi_->close();

	    if (step_ == SELECT_CLOSE)
	      {
		if (setupError(retcode, "ExpHbaseInterface::close"))
		  {
		    step_ = HANDLE_ERROR_NO_CLOSE;
		    break;
		  }
	      }

	    step_ = DONE;
	  }
	  break;

	case HANDLE_ERROR:
	case HANDLE_ERROR_NO_CLOSE:
	  {
	    if (handleError(rc))
	      return rc;

	    if (step_ == HANDLE_ERROR)
	      retcode = ehi_->close();

	    step_ = DONE;
	  }
	  break;

	case DONE:
	  {
	    if (handleDone(rc))
	      return rc;

	    if (scanTask_)
	      scanTask_->init();

	    if (getTask_)
	      getTask_->init();
	    
	    step_ = NOT_STARTED;
	  }
	  break;

	} // switch
    } // while
        
  return WORK_OK;
}

ExHbaseAccessMdamSelectTcb::ExHbaseAccessMdamSelectTcb(
          const ExHbaseAccessTdb &hbaseAccessTdb, 
          ex_globals * glob ) :
  ExHbaseAccessSelectTcb(hbaseAccessTdb, glob),
  step_(NOT_STARTED)
{
}

ExWorkProcRetcode ExHbaseAccessMdamSelectTcb::work()
{
  Lng32 retcode = 0;
  short rc = 0;

  while (!qparent_.down->isEmpty())
    {
      ex_queue_entry *pentry_down = qparent_.down->getHeadEntry();
      if ((pentry_down->downState.request == ex_queue::GET_NOMORE) &&
	  (step_ != DONE))
	{
	  step_ = SELECT_CLOSE_NO_ERROR;
	}

      switch (step_)
	{
	case NOT_STARTED:
	  {
	    matches_ = 0;
	    matchesBeforeFetch_ = 0;

	    if ((pentry_down->downState.request == ex_queue::GET_N) &&
		(pentry_down->downState.requestValue == matches_))
	      {
		step_ = DONE;
		break;
	      }
	    
	    step_ = SELECT_INIT;
	  }
	  break;

	case SELECT_INIT:
	  {
	    retcode = ehi_->init(getHbaseAccessStats());
	    if (setupError(retcode, "ExpHbaseInterface::init"))
	      {
		step_ = HANDLE_ERROR;
		break;
	      }

	    table_.val = hbaseAccessTdb().getTableName();
	    table_.len = strlen(hbaseAccessTdb().getTableName());

	    step_ = INIT_NEXT_KEY_RANGE;
	  }
	  break;

	case INIT_NEXT_KEY_RANGE:
	  {
	    retcode = initNextKeyRange(pool_, pentry_down->getAtp());
	    if (retcode == -1)
	      {
		step_ = HANDLE_ERROR;
		break;
	      }
	    
	    setupListOfColNames(hbaseAccessTdb().listOfFetchedColNames(),
				columns_);

	    fetchRangeHadRows_ = TRUE;

	    step_ = GET_NEXT_KEY_RANGE;
	  }
	  break;

	case GET_NEXT_KEY_RANGE:
	  {
	    keyRangeEx::getNextKeyRangeReturnType 
	      keyRangeStatus = setupSubsetKeys(fetchRangeHadRows_);

	    fetchRangeHadRows_ = FALSE;

	    if (keyRangeStatus == keyRangeEx::EXPRESSION_ERROR)
	      {
		step_ = HANDLE_ERROR;
		break;
	      }

	    if (keyRangeStatus == keyRangeEx::NO_MORE_RANGES)
	      {
		step_ = SELECT_CLOSE;
		break;
	      }

	    if (keyRangeStatus == keyRangeEx::PROBE_RANGE)
	      step_ = PROCESS_PROBE_RANGE;
	    else if (keyRangeStatus == keyRangeEx::FETCH_RANGE)
	      {
		matchesBeforeFetch_ = matches_;
		step_ = PROCESS_FETCH_RANGE;
	      }
	    else
	      step_ = HANDLE_ERROR;
	  }
	  break;

	case PROCESS_PROBE_RANGE:
	  {
	    char * keyData = NULL;
	    retcode = scanSQTaskTcb_->getProbeResult(keyData);
	    if (retcode == -1)
	      {
		step_ = HANDLE_ERROR;
		break;
	      }

	    if (retcode == 1) // no rows found
	      {
		keyExeExpr()->reportProbeResult(0);
		step_ = GET_NEXT_KEY_RANGE;
		break;
	      }

	    // pass the key value to the mdam generator
	    keyExeExpr()->reportProbeResult(keyData);
	    step_ = GET_NEXT_KEY_RANGE;
	  }
	  break;

	case PROCESS_FETCH_RANGE:
	  {
	    rc = 0;
	    retcode = scanSQTaskTcb_->work(rc);
	    
	    if (retcode == 1)
	      return rc;
	    else if (retcode < 0)
	      step_ = HANDLE_ERROR;
	    else
	      {
		if ((pentry_down->downState.request == ex_queue::GET_N) &&
		    (pentry_down->downState.requestValue == matches_))
		  {
		    step_ = SELECT_CLOSE;
		    break;
		  }

		if (matches_ > matchesBeforeFetch_)
		  fetchRangeHadRows_ = TRUE;

		step_ = GET_NEXT_KEY_RANGE;
	      }
	  }
	  break;

	case SELECT_CLOSE:
	case SELECT_CLOSE_NO_ERROR:
	  {
	    retcode = ehi_->close();
	    if (step_ == SELECT_CLOSE)
	      {
		if (setupError(retcode, "ExpHbaseInterface::close"))
		  {
		    step_ = HANDLE_ERROR_NO_CLOSE;
		    break;
		  }
	      }

	    step_ = DONE;
	  }
	  break;

	case HANDLE_ERROR:
	case HANDLE_ERROR_NO_CLOSE:
	  {
	    if (handleError(rc))
	      return rc;

	    if (step_ == HANDLE_ERROR)
	      retcode = ehi_->close();

	    step_ = DONE;
	  }
	  break;

	case DONE:
	  {
	    if (handleDone(rc))
	      return rc;

	    step_ = NOT_STARTED;
	  }
	  break;

	} // switch
    } // while

  return WORK_OK;
}

ExHbaseCoProcAggrTcb::ExHbaseCoProcAggrTcb(
          const ComTdbHbaseCoProcAggr &hbaseAccessTdb, 
          ex_globals * glob ) :
  ExHbaseAccessTcb(hbaseAccessTdb, glob),
  step_(NOT_STARTED)
{
}

ExWorkProcRetcode ExHbaseCoProcAggrTcb::work()
{
  Lng32 retcode = 0;
  short rc = 0;

  while (!qparent_.down->isEmpty())
    {
      ex_queue_entry *pentry_down = qparent_.down->getHeadEntry();
      if ((pentry_down->downState.request == ex_queue::GET_NOMORE) &&
	  (step_ != DONE))
	{
	  //	  step_ = SELECT_CLOSE_NO_ERROR;
	}

      switch (step_)
	{
	case NOT_STARTED:
	  {
	    matches_ = 0;

	    step_ = COPROC_INIT;
	  }
	  break;

	case COPROC_INIT:
	  {
	    retcode = ehi_->init(getHbaseAccessStats());
	    if (setupError(retcode, "ExpHbaseInterface::init"))
	      {
		step_ = HANDLE_ERROR;
		break;
	      }

	    table_.val = hbaseAccessTdb().getTableName();
	    table_.len = strlen(hbaseAccessTdb().getTableName());

	    hbaseAccessTdb().listOfFetchedColNames()->position();
	    hbaseAccessTdb().listOfAggrTypes()->position();

	    aggrIdx_ = 0;

	    step_ = COPROC_EVAL;
	  }
	  break;

	case COPROC_EVAL:
	  {
	    Lng32 aggrType = *(short*)hbaseAccessTdb().listOfAggrTypes()->getCurr();
	    char * col = (char*)hbaseAccessTdb().listOfFetchedColNames()->getCurr();

	    Text aggrVal;
	    Text colFam;
	    Text colName;
	    retcode = extractColFamilyAndName(col, colFam, colName);
	    if (retcode)
	      {
		step_ = HANDLE_ERROR;
		break;
	      }

	    retcode = ehi_->coProcAggr(table_, 
				       aggrType, 
				       "", // startRow
				       "", // stopRow
				       colFam, 
				       colName, 
				       FALSE, // cacheBlocks
				       100, //numCacheRows
				       aggrVal);
	    if (setupError(retcode, "ExpHbaseInterface::coProcAggr"))
	      {
		step_ = HANDLE_ERROR;
		break;
	      }

	    ExpTupleDesc * convertTuppTD =
	      hbaseAccessTdb().workCriDesc_->getTupleDescriptor
	      (hbaseAccessTdb().convertTuppIndex_);

	    Attributes * attr = convertTuppTD->getAttr(aggrIdx_);
	    if (! attr)
	      {
		step_ = HANDLE_ERROR;
		break;
	      }

	    if (attr->getNullFlag())
	      {
		*(short*)&convertRow_[attr->getNullIndOffset()] = 0;
	      }

	    str_cpy_all(&convertRow_[attr->getOffset()], aggrVal.data(), aggrVal.length());

	    hbaseAccessTdb().listOfAggrTypes()->advance();
	    hbaseAccessTdb().listOfFetchedColNames()->advance();
	    aggrIdx_++;

	    if (hbaseAccessTdb().listOfAggrTypes()->atEnd())
	      {
		step_ = RETURN_ROW;
		break;
	      }
	  }
	  break;

	case RETURN_ROW:
	  {
	    short rc = 0;
	    if (moveRowToUpQueue(convertRow_, hbaseAccessTdb().convertRowLen(), 
				 &rc, FALSE))
	      return 1;
	    
	    step_ = DONE;
	  }
	  break;

	case HANDLE_ERROR:
	  {
	    if (handleError(rc))
	      return rc;

	    retcode = ehi_->close();

	    step_ = DONE;
	  }
	  break;

	case DONE:
	  {
	    if (handleDone(rc))
	      return rc;

	    step_ = NOT_STARTED;
	  }
	  break;

	} // switch
    } // while

  return WORK_OK;
}