/**********************************************************************
// @@@ 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:         ExExeUtilMisc.cpp
 * Description:  
 *               
 *               
 * Language:     C++
 *
 *
 *
 *
 *****************************************************************************
 */

#include "ComCextdecs.h"
#include  "cli_stdh.h"
#include  "ex_stdh.h"
#include  "sql_id.h"
#include  "ex_transaction.h"
#include  "ComTdb.h"
#include  "ex_tcb.h"
#include  "ComSqlId.h"

#include  "ExExeUtil.h"
#include  "ex_exe_stmt_globals.h"
#include  "exp_expr.h"
#include  "exp_clause_derived.h"
#include  "ExpLOB.h"
#include  "ComRtUtils.h"
#include  "ExStats.h"
#include  "ComSmallDefs.h"
#include <unistd.h>

//////////////////////////////////////////////////////////
// classes defined in this file:
//
//////////////////////////////////////////////////////////

///////////////////////////////////////////////////////////////////
ex_tcb * ExExeUtilLongRunningTdb::build(ex_globals * glob)
{
  ExExeUtilLongRunningTcb * exe_util_tcb;

  exe_util_tcb = 
    new(glob->getSpace()) ExExeUtilLongRunningTcb(*this, glob);

  exe_util_tcb->registerSubtasks();

  return (exe_util_tcb);
}

////////////////////////////////////////////////////////////////
// Constructor for class ExExeUtilLongRunningTcb
///////////////////////////////////////////////////////////////
ExExeUtilLongRunningTcb::ExExeUtilLongRunningTcb(
     const ComTdbExeUtilLongRunning & exe_util_tdb,
     ex_globals * glob)
  : ExExeUtilTcb( exe_util_tdb, NULL, glob) 
{
  // Allocate the private state in each entry of the down queue
  qparent_.down->allocatePstate(this);

  step_ = INITIAL_;
  rowsDeleted_ = 0;
  transactions_ = 0;
  initial_ = 1;
  initialOutputVarPtrList_ = NULL;
  continuingOutputVarPtrList_ = NULL;
  lruStmtAndPartInfo_= NULL;
  lruStmtWithCKAndPartInfo_= NULL;
  currTransaction_ = NULL;
}

ExExeUtilLongRunningTcb::~ExExeUtilLongRunningTcb()
{
}

Int32 ExExeUtilLongRunningTcb::fixup()
{
    return ex_tcb::fixup();
}


short ExExeUtilLongRunningTcb::doLongRunning()
{
  Lng32 cliRC =0;
  short retcode = 0;
  NABoolean xnAlreadyStarted = FALSE;
     
  // Get the globals stucture of the master executor.
  ExExeStmtGlobals *exeGlob = getGlobals()->castToExExeStmtGlobals();
  ExMasterStmtGlobals *masterGlob = exeGlob->castToExMasterStmtGlobals();

  ex_queue_entry * pentry_down = qparent_.down->getHeadEntry();
  ExExeUtilPrivateState & pstate =
    *((ExExeUtilPrivateState*) pentry_down->pstate);


  CliGlobals *cliGlobals = 0;
  cliGlobals = GetCliGlobals();

  ex_assert(cliGlobals != NULL, "Cli globals is NULL - should have been allocated already");

  if (cliGlobals->isESPProcess())
  {
     if (!currTransaction_)
        currTransaction_ = new (getHeap()) ExTransaction (cliGlobals, getHeap());
  }
  else  // in master executor
  {
     currTransaction_ = masterGlob->getStatement()->getContext()->getTransaction();
  }


  if (currTransaction_->xnInProgress())
  {
     xnAlreadyStarted = TRUE;
  }

  // cannot run LRU when a user transaction is in progress
  if (xnAlreadyStarted)
  {
     ExHandleErrors(qparent_,
         pentry_down,
         0,
         getGlobals(),
         NULL,
         (ExeErrorCode)(-8603),
         NULL,
         exeUtilTdb().getTableName()
         );
     return (-8603);
  }

  SQL_EXEC_ClearDiagnostics(NULL);

  // no Xn in progress. Start one.
  cliRC = currTransaction_->beginTransaction ();

  if (cliRC < 0)
  {
      ExHandleErrors(qparent_,
        pentry_down,
        0,
        getGlobals(),
        NULL,
        (ExeErrorCode)(cliRC),
        NULL,
        exeUtilTdb().getTableName()
        );
       return (short) cliRC;
  }

  retcode = executeLongRunningQuery();

  // Rollback the transaction, if there is an error.
  if (retcode < 0)
  {
     // rollback the transaction
     cliRC = currTransaction_->rollbackTransaction ();

     return retcode;
  }
  else 
  {
    // commit the transaction
    cliRC = currTransaction_->commitTransaction ();

    if (cliRC < 0)
    {
       ExHandleErrors(qparent_,
           pentry_down,
           0,
           getGlobals(),
           NULL,
          (ExeErrorCode)(cliRC),
           NULL,
           exeUtilTdb().getTableName()
           );

       return short(cliRC);
    }

    addTransactionCount();
  }

  return retcode;
}


//////////////////////////////////////////////////////
// work() for ExExeUtilLongRunningTcb
//////////////////////////////////////////////////////
short ExExeUtilLongRunningTcb::work()
{
  short rc = 0;
  Lng32 cliRC = 0;
  Int64 rowsDeleted = 0;
  Int64 transactions = 0;

  // if no parent request, return
  if (qparent_.down->isEmpty())
    return WORK_OK;
  
  // if no room in up queue, won't be able to return data/status.
  // Come back later.
  if (qparent_.up->isFull())
    return WORK_OK;
  
  ex_queue_entry * pentry_down = qparent_.down->getHeadEntry();
  ExExeUtilPrivateState & pstate =
    *((ExExeUtilPrivateState*) pentry_down->pstate);

  // Get the globals stucture of the ESP if this is an ESP
  ExExeStmtGlobals *exeGlob = getGlobals()->castToExExeStmtGlobals();
  ExEspStmtGlobals *espGlob = exeGlob->castToExEspStmtGlobals();


  Int32 espNum = 1;
 
  // this is an ESP?
  if (espGlob != NULL)
  {
     espNum = (Int32) espGlob->getMyInstanceNumber();
  }

  while (1)
    {
      switch (step_)
	{
	case INITIAL_:
	  {
	    step_ = LONG_RUNNING_;
	  }
	break;

	
	case LONG_RUNNING_:
	  {
	    rc = doLongRunning();
	    if ((rc < 0) || (rc == 100)) 
	      {
		finalizeDoLongRunning();
		if (rc <0)
		  step_ = ERROR_;
		else                       // rc == 100 - done with all the transactions.
		  step_ = DONE_;
	      }

            // continue in LONG_RUNNING_ state if (rc >= 0) - success and warning.
	  }
	break;

	
	case DONE_:
	  {
 
	    if (qparent_.up->isFull())
	      return WORK_OK;

	    // Return EOF.
	    ex_queue_entry * up_entry = qparent_.up->getTailEntry();
	    
	    up_entry->upState.parentIndex = 
	      pentry_down->downState.parentIndex;
	    
	    up_entry->upState.setMatchNo(0);
	    up_entry->upState.status = ex_queue::Q_NO_DATA;

	    // before sending the Q_NO_DATA, send the rowcount as well thro'
	    // the diagsArea.
	    
	    getDiagsArea()->setRowCount(getRowsDeleted());

            ComDiagsArea *diagsArea = getDiagAreaFromUpQueueTail();
            if (lrTdb().longRunningQueryPlan())
            {
               (*diagsArea) << DgSqlCode(8450)
                            << DgString0((char*)exeUtilTdb().getTableName())
                            << DgInt0(espNum)
                            << DgInt1((Lng32)getTransactionCount());
            }
	    
	    // insert into parent
	    qparent_.up->insert();
	    
	    //pstate.matches_ = 0;
	    // reset the parameters.
	    step_ = INITIAL_;
	    transactions_ = 0;
	    rowsDeleted_ = 0;
	    initial_ = 1;
	    
	    // clear diags if any
            if (getDiagsArea())
            {
                getDiagsArea()->clear();
            }
    
	    qparent_.down->removeHead();
 
	    return WORK_OK;
	  }
	break;

	case ERROR_:
	  {
	    if (qparent_.up->isFull())
	      return WORK_OK;

	    // Return EOF.
	    ex_queue_entry * up_entry = qparent_.up->getTailEntry();
	    
	    up_entry->upState.parentIndex = 
	      pentry_down->downState.parentIndex;
	    
	    up_entry->upState.setMatchNo(0);
	    up_entry->upState.status = ex_queue::Q_SQLERROR;
	    // get rows deleted so far.
	    getDiagsArea()->setRowCount(getRowsDeleted());
	    ComDiagsArea *diagsArea = up_entry->getDiagsArea();
	    
	    if (diagsArea == NULL)
	      diagsArea = 
		ComDiagsArea::allocate(this->getGlobals()->getDefaultHeap());
            else
              diagsArea->incrRefCount (); // setDiagsArea call below will decr ref count
	    
	    if (getDiagsArea())
	      diagsArea->mergeAfter(*getDiagsArea());
	    
	    up_entry->setDiagsArea (diagsArea);
	    
	    // insert into parent
	    qparent_.up->insert();

	    // clear diags if any, since we already sent the information
	    // up and don't want to send it again as part of DONE_
            if (getDiagsArea())
            {
	      rowsDeleted_ = 0;
	      getDiagsArea()->clear();
            }
	    step_ = DONE_;
	  }
	break;

	} // switch
    } // while

}


void ExExeUtilLongRunningTcb::registerSubtasks()
{

  // register events for parent queue
  ex_tcb::registerSubtasks();
}

ComDiagsArea *ExExeUtilLongRunningTcb::getDiagAreaFromUpQueueTail()
{
  ex_queue_entry * up_entry = qparent_.up->getTailEntry();
  ComDiagsArea *diagsArea = up_entry->getDiagsArea();

  if (diagsArea == NULL)
     diagsArea = ComDiagsArea::allocate(this->getGlobals()->getDefaultHeap());
  else
    diagsArea->incrRefCount (); // setDiagsArea call below will decr ref count

  // this is the side-effect of this function. Merge in this object's
  // diagsarea.
  if (getDiagsArea())
    diagsArea->mergeAfter(*getDiagsArea());

  up_entry->setDiagsArea (diagsArea);

  return diagsArea;
}

short ExExeUtilLongRunningTcb::executeLongRunningQuery()
{
  Lng32 rc = 0;

  if (getInitial())
    {
      Lng32 cliRC = 
      cliInterface()->executeImmediate(
        "control query default HIST_ON_DEMAND_STATS_SIZE '0'");
      if (cliRC < 0) 
        {
          cliInterface()->allocAndRetrieveSQLDiagnostics(diagsArea_);
	  return cliRC;
        }
      

      // Perform the initial processing
      short rtc = processInitial(rc);
      if ((rc != 0) && (rc != 100))
        {
          cliInterface()->allocAndRetrieveSQLDiagnostics(diagsArea_);
        }

      cliRC = 
        cliInterface()->executeImmediate(
            "control query default HIST_ON_DEMAND_STATS_SIZE 'RESET'");
      if (cliRC < 0) 
        {
          cliInterface()->allocAndRetrieveSQLDiagnostics(diagsArea_);
	  return cliRC;
        }
      

#ifdef _DEBUG
      if (lrTdb().longRunningQueryPlan()) {

          Int32 bufSize = 100 + 
                strlen(lruStmtAndPartInfo_)+ strlen(lruStmtWithCKAndPartInfo_);
          char* lruQPInfo = new (getHeap()) char[bufSize];

          // str_printf() does not support %ld. Use %d instead.
          str_sprintf(lruQPInfo, 
	             "Queries to be processed: \n\n%s\n\n%s\n\n\n Initial rows deleted: %ld",
                      lruStmtAndPartInfo_,
                      lruStmtWithCKAndPartInfo_,
                      getRowsDeleted());

          ComDiagsArea * diagsArea = getDiagAreaFromUpQueueTail();
          (*diagsArea) << DgSqlCode(8427) << DgString0(lruQPInfo);

          NADELETEBASIC(lruQPInfo, getHeap());

      }
#endif

      setInitial(0);
    
    }
  else
    {
      // Perform the continuing processing
      short rtc = processContinuing(rc);
      if ((rc != 0) && (rc != 100))
        {
          cliInterface()->allocAndRetrieveSQLDiagnostics(diagsArea_);
        }

#ifdef _DEBUG
      if ((rc == 100 || rc > 0) && lrTdb().longRunningQueryPlan()) {

        char lruQPInfo [100];
        str_sprintf(lruQPInfo, "Total rows deleted: %ld\n\n", getRowsDeleted());

        ComDiagsArea * diagsArea = getDiagAreaFromUpQueueTail();
        (*diagsArea) << DgSqlCode(8427) << DgString0(lruQPInfo);
      }
#endif
    }
  

  return (short )rc;
}

short ExExeUtilLongRunningTcb::finalizeDoLongRunning() 
{

  // Close the statements
  Lng32 rc = cliInterface()->prepareAndExecRowsEpilogue();

  NADELETE(initialOutputVarPtrList_, Queue, getHeap());
  NADELETE(continuingOutputVarPtrList_, Queue, getHeap());
  NADELETEBASIC(lruStmtAndPartInfo_, getHeap());
  NADELETEBASIC(lruStmtWithCKAndPartInfo_, getHeap());

  // if this is an ESP, deallocate transaction
  CliGlobals *cliGlobals = GetCliGlobals();
  if (cliGlobals->isESPProcess())
  {
     NADELETEBASIC(currTransaction_, getHeap());
     currTransaction_ = NULL;
  }

  return 0;
}

short ExExeUtilLongRunningTcb::processInitial(Lng32 &rc)
{

  Int64 rowsAffected = 0;

  setInitialOutputVarPtrList(new(getHeap()) Queue(getHeap()));
  setContinuingOutputVarPtrList(new(getHeap()) Queue(getHeap()));

  lruStmtAndPartInfo_ = new(getHeap()) char[(UInt32)(lrTdb().getLruStmtLen() + lrTdb().getPredicateLen() + 10)];

  ExExeStmtGlobals *exeGlob = getGlobals()->castToExExeStmtGlobals();
  ExEspStmtGlobals *espGlob = exeGlob->castToExEspStmtGlobals();

  ContextCli *currContext = NULL;
  CliGlobals *cliGlobals = NULL;

  if (espGlob)
  {
    cliGlobals = GetCliGlobals();
    currContext = cliGlobals->currContext();
  }
  else
  {
     currContext = exeGlob->castToExMasterStmtGlobals()->getStatement()->getContext();
  }

  Int32 espNum = 1;
  
  // we are executing inside an ESP go ahead and set the partition number to the 
  // ESP instance number. The added one is because ESP instances are 0 based
  if (espGlob)
  {
     espNum = (Int32) espGlob->getMyInstanceNumber() + 1;
  }

  if (lrTdb().getPredicate() != NULL)
      str_sprintf(lruStmtAndPartInfo_, 
	      lrTdb().getLruStmt(), 
	      espNum,
              lrTdb().getPredicate());
  else
      str_sprintf(lruStmtAndPartInfo_, 
	      lrTdb().getLruStmt(), 
	      espNum);

  lruStmtWithCKAndPartInfo_ = new(getHeap()) char[(UInt32)(lrTdb().getLruStmtWithCKLen() + lrTdb().getPredicateLen() + 10)];

  if (lrTdb().getPredicate() != NULL)
     str_sprintf(lruStmtWithCKAndPartInfo_, 
	      lrTdb().getLruStmtWithCK(), 
	      espNum,
              lrTdb().getPredicate());
  else
     str_sprintf(lruStmtWithCKAndPartInfo_, 
	      lrTdb().getLruStmtWithCK(), 
	      espNum);

  // All internal queries issued from CliInterface assume that
  // they are in ISO_MAPPING.
  // For LongRunning we need to send the actual ISO_MAPPING.
  // Save it and restore after the prepare

  Int32 savedIsoMapping = 
    cliInterface()->getIsoMapping();

  cliInterface()->setIsoMapping
    (currContext->getSessionDefaults()->getIsoMappingEnum());

  // If the table we are deleting from is an IUD log table,
  // we need to set the parserflags to accept the special
  // table type and the quoted column names
  if (lrTdb().useParserflags())
    currContext->setSqlParserFlags(0x3);

  rc = cliInterface()->prepareAndExecRowsPrologue(lruStmtAndPartInfo_,
						  lruStmtWithCKAndPartInfo_, 
                                                  getInitialOutputVarPtrList(),
                                                  getContinuingOutputVarPtrList(),
                                                  rowsAffected);

  cliInterface()->setIsoMapping(savedIsoMapping);

  if (rc < 0)
    {
      return -1;
    }

  if (rc >= 0 && rowsAffected > 0)
    addRowsDeleted(rowsAffected);
 
  return 0;
}

short ExExeUtilLongRunningTcb::processContinuing(Lng32 &rc) 
{

  Int64 rowsAffected = 0;

  rc = cliInterface()->execContinuingRows(getContinuingOutputVarPtrList(),
                                          rowsAffected);

  if (rc < 0)
    {
      return -1;
    }

  if (rowsAffected > 0)
    addRowsDeleted(rowsAffected);

#ifdef _DEBUG

   if ((rowsAffected > 0) && lrTdb().longRunningQueryPlan()) {

        char lruQPInfo[100];

        str_sprintf(lruQPInfo, "Continuing rows deleted: %ld\n\n", 
                                rowsAffected);

        ComDiagsArea * diagsArea = getDiagAreaFromUpQueueTail();
        (*diagsArea) << DgSqlCode(8427) << DgString0(lruQPInfo);
   }
#endif
 
  return 0;
}


////////////////////////////////////////////////////////////////////////
// Redefine virtual method allocatePstates, to be used by dynamic queue
// resizing, as well as the initial queue construction.
////////////////////////////////////////////////////////////////////////
ex_tcb_private_state * ExExeUtilLongRunningTcb::allocatePstates(
     Lng32 &numElems,      // inout, desired/actual elements
     Lng32 &pstateLength)  // out, length of one element
{
  PstateAllocator<ExExeUtilLongRunningPrivateState> pa;

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

/////////////////////////////////////////////////////////////////////////////
// Constructor and destructor for ExeUtil_private_state
/////////////////////////////////////////////////////////////////////////////
ExExeUtilLongRunningPrivateState::ExExeUtilLongRunningPrivateState()
{
}

ExExeUtilLongRunningPrivateState::~ExExeUtilLongRunningPrivateState()
{
};

///////////////////////////////////////////////////////////////////
// class ExExeUtilAQRTdb
///////////////////////////////////////////////////////////////
ex_tcb * ExExeUtilAQRTdb::build(ex_globals * glob)
{
  ExExeUtilAQRTcb * exe_util_tcb;

  exe_util_tcb = new(glob->getSpace()) ExExeUtilAQRTcb(*this, glob);

  exe_util_tcb->registerSubtasks();

  return (exe_util_tcb);
}

ExExeUtilAQRTcb::ExExeUtilAQRTcb(const ComTdbExeUtilAQR & exe_util_tdb,
				 ex_globals * glob)
     : ExExeUtilTcb(exe_util_tdb, NULL, glob),
       step_(EMPTY_)
{
}

short ExExeUtilAQRTcb::work()
{
  // if no parent request, return
  if (qparent_.down->isEmpty())
    return WORK_OK;

  ex_queue_entry * pentry_down = qparent_.down->getHeadEntry();
  ExExeUtilPrivateState & pstate =
    *((ExExeUtilPrivateState*) pentry_down->pstate);

  ContextCli *currContext =
      getGlobals()->castToExExeStmtGlobals()->castToExMasterStmtGlobals()->
        getStatement()->getContext();
    
  AQRInfo * aqr = currContext->aqrInfo();

  while (1)
    {
      switch (step_)
	{
	case EMPTY_:
	  {
	    if (aqrTdb().getTask() == ComTdbExeUtilAQR::GET_)
	      {
		// GET aqr entries and display them
		aqr->position();
		
		step_ = RETURN_HEADER_;
	      }
	    else
	      {
		step_ = SET_ENTRY_;
	      }
	  }
	break;

	case SET_ENTRY_:
	  {
	    if (aqr->setAQREntry(aqrTdb().getTask(), 
				 aqrTdb().sqlcode_, aqrTdb().nskcode_,
				 aqrTdb().retries_, aqrTdb().delay_,
				 aqrTdb().type_, 
				 0, NULL, 0, 0))
	      {
		// error
		step_ = HANDLE_ERROR_;
		break;
	      }

	    step_ = DONE_;
	  }
	break;

	case RETURN_HEADER_:
	  {
	    // if no room in up queue for 4 display rows, 
	    // won't be able to return data/status.
	    // Come back later.
	    if ((qparent_.up->getSize() - qparent_.up->getLength()) < 5)
	      return -1;
	    
	    moveRowToUpQueue("  ");
	    moveRowToUpQueue("  SQLCODE    PLATFORMERROR    RETRIES      DELAY       TYPE ");
	    moveRowToUpQueue("============================================================");
	    moveRowToUpQueue("  ");
	    
	    step_ = RETURNING_ENTRY_;
	  }
	break;

	case RETURNING_ENTRY_:
	  {
	    // if no room in up queue, won't be able to return data/status.
	    // Come back later.
	    if (qparent_.up->isFull())
	      return WORK_OK;

	    Lng32 sqlcode, nskcode, retries, delay, type, intAQR;
	    Lng32 eof = 0;
	    eof = 
	      aqr->getNextAQREntry(sqlcode, nskcode, retries, delay, type,
				   intAQR);
	    if (eof)
	      {
		step_ = DONE_;
		break;
	      }

	    // internal aqr entry, no need to display
	    if (intAQR)
	      break;

	    char formattedStr[400];
	    str_sprintf(formattedStr, "%9d  %13d  %9d  %9d  %9d",
			sqlcode, nskcode, retries, delay, type);

	    moveRowToUpQueue(formattedStr);
	  }
	break;

	case HANDLE_ERROR_:
	  {
	    if (qparent_.up->isFull())
	      return WORK_OK;

	    ExHandleErrors(qparent_,
			   pentry_down,
			   0,
			   getGlobals(),
			   NULL,
			   (ExeErrorCode)8931,
			   NULL,
			   NULL
			   );

	    step_ = DONE_;
	  }
	break;

	case DONE_:
	  {
	    if (qparent_.up->isFull())
	      return WORK_OK;

	    // all ok. Return EOF.
	    ex_queue_entry * up_entry = qparent_.up->getTailEntry();
	    
	    up_entry->upState.parentIndex = 
	      pentry_down->downState.parentIndex;
	    
	    up_entry->upState.setMatchNo(0);
	    up_entry->upState.status = ex_queue::Q_NO_DATA;
	    
	    // insert into parent
	    qparent_.up->insert();
	    
	    step_ = EMPTY_;
	    
	    qparent_.down->removeHead();
	    
	    return WORK_OK;
	  }
	  break;

	} // switch
    }

  return 0;
}

///////////////////////////////////////////////////////////////////
ex_tcb * ExExeUtilPopulateInMemStatsTdb::build(ex_globals * glob)
{
  ex_tcb * exe_util_tcb;

  exe_util_tcb = new(glob->getSpace()) ExExeUtilPopulateInMemStatsTcb(*this, glob);

  exe_util_tcb->registerSubtasks();

  return (exe_util_tcb);
}

////////////////////////////////////////////////////////////////
// Constructor for class ExExeUtilPopulateInMemStatsTcb
///////////////////////////////////////////////////////////////
ExExeUtilPopulateInMemStatsTcb::ExExeUtilPopulateInMemStatsTcb(
     const ComTdbExeUtilPopulateInMemStats & exe_util_tdb,
     ex_globals * glob)
     : ExExeUtilTcb( exe_util_tdb, NULL, glob)
{
  // Allocate the private state in each entry of the down queue
  qparent_.down->allocatePstate(this);

  step_ = INITIAL_;
}

ExExeUtilPopulateInMemStatsTcb::~ExExeUtilPopulateInMemStatsTcb()
{
}

//////////////////////////////////////////////////////
// work() for ExExeUtilPopulateInMemStatsTcb
//////////////////////////////////////////////////////
static const QueryString deleteStatsQuery[] =
{
  {" delete from %s "},
  {"  where table_uid = %ld"}
};

static const QueryString populateHistogramsStatsQuery[] =
{
  {" insert into %s "},
  {" select  "},
  {"   %ld, "},
  {"   histogram_id, "},
  {"   col_position, column_number, colcount, interval_count, rowcount, "},
  {"   total_uec, stats_time, low_value, high_value, read_time, read_count, "},
  {"   sample_secs, col_secs, sample_percent, "},
  {"   cv, reason, v1, v2, v3, v4, v5, v6 "},
  {" from "},
  {"    %s "},
  {"    where table_uid =  "},
  {"     ( "},
  {"       select object_uid from  "},
  {"       HP_SYSTEM_CATALOG.system_schema.catsys C, "},
  {"       HP_SYSTEM_CATALOG.system_schema.schemata S, "},
  {"       %s.HP_DEFINITION_SCHEMA.objects O "},
  {"       where "},
  {"             C.cat_name = '%s' "},
  {"       and S.schema_name = '%s' "},
  {"       and C.cat_uid = S.cat_uid "},
  {"       and S.schema_uid = O.schema_uid "},
  {"       and O.object_name = '%s' "},
  {"     ) "}
};

static const QueryString populateHistintsStatsQuery[] =
{
  {" insert into %s "},
  {" select  "},
  {"   %ld, "},
  {"   histogram_id, "},
  {"   interval_number, interval_rowcount, interval_uec, interval_boundary, "},
  {"   std_dev_of_freq, "},
  {"   v1, v2, v3, v4, v5, v6 "},
  {" from "},
  {"    %s "},
  {"    where table_uid =  "},
  {"     ( "},
  {"       select object_uid from  "},
  {"       HP_SYSTEM_CATALOG.system_schema.catsys C, "},
  {"       HP_SYSTEM_CATALOG.system_schema.schemata S, "},
  {"       %s.HP_DEFINITION_SCHEMA.objects O "},
  {"       where "},
  {"             C.cat_name = '%s' "},
  {"       and S.schema_name = '%s' "},
  {"       and C.cat_uid = S.cat_uid "},
  {"       and S.schema_uid = O.schema_uid "},
  {"       and O.object_name = '%s' "},
  {"     ) "}
};

short ExExeUtilPopulateInMemStatsTcb::work()
{
  //  short rc = 0;
  Lng32 cliRC = 0;

  // if no parent request, return
  if (qparent_.down->isEmpty())
    return WORK_OK;
  
  // if no room in up queue, won't be able to return data/status.
  // Come back later.
  if (qparent_.up->isFull())
    return WORK_OK;
  
  ex_queue_entry * pentry_down = qparent_.down->getHeadEntry();
  ExExeUtilPrivateState & pstate =
    *((ExExeUtilPrivateState*) pentry_down->pstate);

  // Get the globals stucture of the master executor.
  ExExeStmtGlobals *exeGlob = getGlobals()->castToExExeStmtGlobals();
  ExMasterStmtGlobals *masterGlob = exeGlob->castToExMasterStmtGlobals();
  ContextCli * currContext = masterGlob->getStatement()->getContext();

  while (1)
    {
      switch (step_)
	{
	case INITIAL_:
	  {
	    if (getDiagsArea())
	      {
		getDiagsArea()->clear();
		getDiagsArea()->deAllocate();
	      }
	    
	    setDiagsArea(ComDiagsArea::allocate(getHeap()));

	    step_ = PROLOGUE_;
	  }
	break;

	case PROLOGUE_:
	  {
	    if (disableCQS())
	      {
		step_ = ERROR_;
		break;
	      }

	    if (setSchemaVersion(pimsTdb().sourceTableCatName_))
	      {
		step_ = ERROR_;
		break;
	      }

	    // set sqlparserflags to allow use of volatile schema in queries.
	    masterGlob->getStatement()->getContext()->
	      setSqlParserFlags(0x10000);//ALLOW_VOLATILE_SCHEMA_IN_TABLE_NAME

	    step_ = DELETE_STATS_;
	  }
	break;

	case DELETE_STATS_:
	  {
	    Int32 qry_array_size = sizeof(deleteStatsQuery) 
	      / sizeof(QueryString);
	    
	    const QueryString * queryString = deleteStatsQuery;;
	    
	    char * gluedQuery;
	    Lng32 gluedQuerySize;
	    glueQueryFragments(qry_array_size, queryString,
			       gluedQuery, gluedQuerySize);
	    
	    Lng32 extraSpace = 
	      ComMAX_3_PART_EXTERNAL_UTF8_NAME_LEN_IN_BYTES /* fullyQualTableName */
	      + 20 /* UID */
	      + 200 /* overhead */;

	    char * query = new(getHeap()) char[gluedQuerySize + extraSpace];
	    str_sprintf(query, gluedQuery, 
			(char*)pimsTdb().inMemHistogramsTableName_,
			pimsTdb().uid_);
	    
	    cliRC = 
	      cliInterface()->executeImmediate(query);
	    
	    if (cliRC >= 0)
	      {
		str_sprintf(query, gluedQuery, 
			    (char*)pimsTdb().inMemHistintsTableName_,
			    pimsTdb().uid_);
		
		cliRC = 
		  cliInterface()->executeImmediate(query);
	      }

	    // Delete new'd string
	    NADELETEBASIC(gluedQuery, getHeap());
	    gluedQuery = NULL;
	    
	    NADELETEBASIC(query, getHeap());
	    query = NULL;
	    
	    if (cliRC < 0)
	      {
                cliInterface()->allocAndRetrieveSQLDiagnostics(diagsArea_);
		step_ = ERROR_;
	      }
	    else
	      step_ = POPULATE_HISTOGRAMS_STATS_;
	  }
	break;

	case POPULATE_HISTOGRAMS_STATS_:
	  {
	    Int32 qry_array_size = sizeof(populateHistogramsStatsQuery) 
	      / sizeof(QueryString);
	    
	    const QueryString * queryString = populateHistogramsStatsQuery;;
	    
	    char * gluedQuery;
	    Lng32 gluedQuerySize;
	    glueQueryFragments(qry_array_size, queryString,
			       gluedQuery, gluedQuerySize);
	    
	    Lng32 extraSpace =
	        ComMAX_3_PART_EXTERNAL_UTF8_NAME_LEN_IN_BYTES /* fullyQualInMemHistTableName */
	      + 20 /* UID */
	      + ComMAX_3_PART_EXTERNAL_UTF8_NAME_LEN_IN_BYTES /* fullyQualSourceHistTableName */
	      + 2 * 10 /*segment name*/
	      + ComMAX_1_PART_EXTERNAL_UTF8_NAME_LEN_IN_BYTES /*cat name*/
	      + 10  /*version*/ 
	      + ComMAX_1_PART_EXTERNAL_UTF8_NAME_LEN_IN_BYTES /*cat name*/
	      + ComMAX_1_PART_EXTERNAL_UTF8_NAME_LEN_IN_BYTES /*sch name*/
	      + ComMAX_1_PART_EXTERNAL_UTF8_NAME_LEN_IN_BYTES /*obj name*/
	      + 200 /* overhead */;

	    char * query = new(getHeap()) char[gluedQuerySize + extraSpace];
	    
	    str_sprintf(query, gluedQuery, 
			(char*)pimsTdb().inMemHistogramsTableName_,
			pimsTdb().uid_,
			(char*)pimsTdb().sourceHistogramsTableName_,
			(char*)pimsTdb().sourceTableCatName_,
			(char*)pimsTdb().sourceTableCatName_,
			(char*)pimsTdb().sourceTableSchName_,
			(char*)pimsTdb().sourceTableObjName_
			);
	    
	    cliRC = 
	      cliInterface()->executeImmediate(query);
	    
	    // Delete new'd string
	    NADELETEBASIC(gluedQuery, getHeap());
	    gluedQuery = NULL;
	    
	    NADELETEBASIC(query, getHeap());
	    query = NULL;
	    
	    if (cliRC < 0)
	      {
                cliInterface()->allocAndRetrieveSQLDiagnostics(diagsArea_);
		step_ = ERROR_;
	      }
	    else
	      step_ = POPULATE_HISTINTS_STATS_;
	  }
	break;
	  
	case POPULATE_HISTINTS_STATS_:
	  {
	    Int32 qry_array_size = sizeof(populateHistintsStatsQuery) 
	      / sizeof(QueryString);
	    
	    const QueryString * queryString = populateHistintsStatsQuery;;
	    
	    char * gluedQuery;
	    Lng32 gluedQuerySize;
	    glueQueryFragments(qry_array_size, queryString,
			       gluedQuery, gluedQuerySize);
	    
	    Lng32 extraSpace =
	        ComMAX_3_PART_EXTERNAL_UTF8_NAME_LEN_IN_BYTES /* fullyQualInMemHistTableName */
	      + 20 /* UID */
	      + ComMAX_3_PART_EXTERNAL_UTF8_NAME_LEN_IN_BYTES /* fullyQualSourceHistTableName */
	      + 2 * 10 /*segment name*/
	      + ComMAX_1_PART_EXTERNAL_UTF8_NAME_LEN_IN_BYTES /*cat name*/
	      + 10  /*version*/ 
	      + ComMAX_1_PART_EXTERNAL_UTF8_NAME_LEN_IN_BYTES /*cat name*/
	      + ComMAX_1_PART_EXTERNAL_UTF8_NAME_LEN_IN_BYTES /*sch name*/
	      + ComMAX_1_PART_EXTERNAL_UTF8_NAME_LEN_IN_BYTES /*obj name*/
	      + 200 /* overhead */;

	    char * query = new(getHeap()) char[gluedQuerySize + extraSpace];
	    
	    str_sprintf(query, gluedQuery, 
			(char*)pimsTdb().inMemHistintsTableName_,
			pimsTdb().uid_,
			(char*)pimsTdb().sourceHistintsTableName_,
			(char*)pimsTdb().sourceTableCatName_,
			(char*)pimsTdb().sourceTableCatName_,
			(char*)pimsTdb().sourceTableSchName_,
			(char*)pimsTdb().sourceTableObjName_
			);
	    
	    cliRC = 
	      cliInterface()->executeImmediate(query);
	    
	    // Delete new'd string
	    NADELETEBASIC(gluedQuery, getHeap());
	    gluedQuery = NULL;
	    
	    NADELETEBASIC(query, getHeap());
	    query = NULL;
	    
	    if (cliRC < 0)
	      {
                cliInterface()->allocAndRetrieveSQLDiagnostics(diagsArea_);
		step_ = ERROR_;
	      }
	    else
	      step_ = EPILOGUE_;
	  }
	break;

	case EPILOGUE_:
	case EPILOGUE_AND_ERROR_RETURN_:
	  {
	    // reset sqlparserflags
	    masterGlob->getStatement()->getContext()->
	      resetSqlParserFlags(0x10000);//ALLOW_VOLATILE_SCHEMA_IN_TABLE_NAME

	    restoreCQS();

	    if (step_ == EPILOGUE_AND_ERROR_RETURN_)
	      step_ = ERROR_RETURN_;
	    else
	      step_ = DONE_;
	  }
	  break;

	case ERROR_:
	  {
	    step_ = EPILOGUE_AND_ERROR_RETURN_;
	  }
	  break;

	case ERROR_RETURN_:
	  {
	    if (qparent_.up->isFull())
	      return WORK_OK;

	    // Return EOF.
	    ex_queue_entry * up_entry = qparent_.up->getTailEntry();
	    
	    up_entry->upState.parentIndex = 
	      pentry_down->downState.parentIndex;
	    
	    up_entry->upState.setMatchNo(0);
	    up_entry->upState.status = ex_queue::Q_SQLERROR;

	    ComDiagsArea *diagsArea = up_entry->getDiagsArea();
	    
	    if (diagsArea == NULL)
	      diagsArea = 
		ComDiagsArea::allocate(this->getGlobals()->getDefaultHeap());
	    
	    if (getDiagsArea())
	      diagsArea->mergeAfter(*getDiagsArea());
	    
	    up_entry->setDiagsArea (diagsArea);
	    
	    // insert into parent
	    qparent_.up->insert();
	    
	    step_ = DONE_;
	  }
	  break;

	case DONE_:
	  {
	    if (qparent_.up->isFull())
	      return WORK_OK;
	    
	    // Return EOF.
	    ex_queue_entry * up_entry = qparent_.up->getTailEntry();
	    
	    up_entry->upState.parentIndex = 
	      pentry_down->downState.parentIndex;
	    
	    up_entry->upState.setMatchNo(0);
	    up_entry->upState.status = ex_queue::Q_NO_DATA;
	    
	    // insert into parent
	    qparent_.up->insert();
	    
	    qparent_.down->removeHead();
	    
	    step_ = INITIAL_;
	    return WORK_OK;
	  }
	
	break;
	
	default:
	  break;

	}

    }

  return 0;
}


////////////////////////////////////////////////////////////////////////
// Redefine virtual method allocatePstates, to be used by dynamic queue
// resizing, as well as the initial queue construction.
////////////////////////////////////////////////////////////////////////
ex_tcb_private_state * ExExeUtilPopulateInMemStatsTcb::allocatePstates(
     Lng32 &numElems,      // inout, desired/actual elements
     Lng32 &pstateLength)  // out, length of one element
{
  PstateAllocator<ExExeUtilPopulateInMemStatsPrivateState> pa;

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

/////////////////////////////////////////////////////////////////////////////
// Constructor and destructor for ExeUtil_private_state
/////////////////////////////////////////////////////////////////////////////
ExExeUtilPopulateInMemStatsPrivateState::ExExeUtilPopulateInMemStatsPrivateState()
{
}

ExExeUtilPopulateInMemStatsPrivateState::~ExExeUtilPopulateInMemStatsPrivateState()
{
};


////////////////////////////////////////////////////////////////
// Constructor for class ExExeUtilHiveTruncateTdb
///////////////////////////////////////////////////////////////
ex_tcb * ExExeUtilHiveTruncateTdb::build(ex_globals * glob)
{
  ExExeUtilTcb * exe_util_tcb;

  if (getIsLegacy())
    exe_util_tcb = new(glob->getSpace()) ExExeUtilHiveTruncateLegacyTcb(*this, glob);
  else
    exe_util_tcb = new(glob->getSpace()) ExExeUtilHiveTruncateTcb(*this, glob);
    
  exe_util_tcb->registerSubtasks();

  return (exe_util_tcb);
}


////////////////////////////////////////////////////////////////
// Constructor for class ExExeUtilHiveTruncateLegacyTcb
///////////////////////////////////////////////////////////////
ExExeUtilHiveTruncateLegacyTcb::ExExeUtilHiveTruncateLegacyTcb(
     const ComTdbExeUtilHiveTruncate & exe_util_tdb,
     ex_globals * glob)
     : ExExeUtilTcb( exe_util_tdb, NULL, glob)
{
  // Allocate the private state in each entry of the down queue
  qparent_.down->allocatePstate(this);

  numExistingFiles_ = 0;

  step_ = INITIAL_;
}

ExExeUtilHiveTruncateLegacyTcb::~ExExeUtilHiveTruncateLegacyTcb()
{
  freeResources();
}

void ExExeUtilHiveTruncateLegacyTcb::freeResources()
{
  if (htTdb().getDropOnDealloc())
  {
      NAString hiveDropDDL("drop table ");
      hiveDropDDL += htTdb().getHiveTableName();

      // TODO: is it ok to ignore the error 
      HiveClient_JNI::executeHiveSQL(hiveDropDDL);
  }
  if (lobGlob_) {
    ExpLOBinterfaceCleanup(lobGlob_);
    lobGlob_ = NULL;
  }
}

Int32 ExExeUtilHiveTruncateLegacyTcb::fixup()
{
  lobGlob_ = NULL;

  ExpLOBinterfaceInit
    (lobGlob_, (NAHeap *)getGlobals()->getDefaultHeap(),
     getGlobals()->castToExExeStmtGlobals()->getContext(),FALSE, 
     htTdb().getHdfsHost(),
     htTdb().getHdfsPort());

  return 0;
}
//////////////////////////////////////////////////////
// work() for ExExeUtilHiveTruncateLegacyTcb
//////////////////////////////////////////////////////
short ExExeUtilHiveTruncateLegacyTcb::work()
{
  short rc = 0;
  Lng32 cliRC = 0;

  // if no parent request, return
  if (qparent_.down->isEmpty())
    return WORK_OK;

  // if no room in up queue, won't be able to return data/status.
  // Come back later.
  if (qparent_.up->isFull())
    return WORK_OK;

  ex_queue_entry * pentry_down = qparent_.down->getHeadEntry();
  ExExeUtilPrivateState & pstate = *((ExExeUtilPrivateState*) pentry_down->pstate);

  while (1)
  {
    switch (step_)
    {
      case INITIAL_:
      {
         step_ = EMPTY_DIRECTORY_;
      }
      break;
      case EMPTY_DIRECTORY_:
      {
        cliRC = ExpLOBinterfaceEmptyDirectory(
             lobGlob_,
             (char*)"",                  //name is empty
             (htTdb().getPartnLocation() ? 
              htTdb().getPartnLocation() : 
              htTdb().getTableLocation()),
             Lob_HDFS_File,
             htTdb().getHdfsHost(),
             htTdb().getHdfsPort(),
             0 ,
             1 ,
             0);
        if (cliRC != 0)
        {
          Lng32 cliError = 0;
          
          Lng32 intParam1 = -cliRC;
          ExRaiseSqlError(getHeap(), &diagsArea_, 
                          (ExeErrorCode)(EXE_ERROR_FROM_LOB_INTERFACE),
                          NULL, &intParam1, 
                          &cliError, 
                          NULL, 
                          "HDFS",
                          (char*)"ExpLOBInterfaceEmptyDirectory",
                          getLobErrStr(intParam1));

          char reason[200];
          
          strcpy(reason, " ");
          if (intParam1 == LOB_DIR_NAME_ERROR)
            {
              if (htTdb().getPartnLocation())
                strcpy(reason, "Reason: specified partition does not exist");
              else
                strcpy(reason, "Reason: specified table location does not exist");
            }
          else if (intParam1 == LOB_DATA_FILE_DELETE_ERROR)
            {
              strcpy(reason, "Reason: error occurred during deletion of one or more files at the specified location");
            }
          
          ExRaiseSqlError(getHeap(), &diagsArea_, 
                          (ExeErrorCode)(EXE_HIVE_TRUNCATE_ERROR), NULL,
                          NULL, NULL, NULL,
                          reason,
                          NULL, NULL);
          step_ = ERROR_;
        }
        else
        {
          step_= DONE_;
        }
      }
      break;

      case ERROR_:
      {
        if (handleError())
           return WORK_OK;
        step_ = DONE_;
      }
      break;

      case DONE_:
      {
        if (handleDone())
           return WORK_OK;
        step_ = INITIAL_;

        return WORK_OK;
      }
      break;

    } // switch
  } // while

}


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

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

/////////////////////////////////////////////////////////////////////////////
// Constructor and destructor for ExeUtil_private_state
/////////////////////////////////////////////////////////////////////////////
ExExeUtilHiveTruncateLegacyPrivateState::ExExeUtilHiveTruncateLegacyPrivateState()
{
}

ExExeUtilHiveTruncateLegacyPrivateState::~ExExeUtilHiveTruncateLegacyPrivateState()
{
};

////////////////////////////////////////////////////////////////
// Constructor for class ExExeUtilHiveTruncateTcb
///////////////////////////////////////////////////////////////
ExExeUtilHiveTruncateTcb::ExExeUtilHiveTruncateTcb(
     const ComTdbExeUtilHiveTruncate & exe_util_tdb,
     ex_globals * glob)
     : ExExeUtilTcb( exe_util_tdb, NULL, glob)
{
  // Allocate the private state in each entry of the down queue
  qparent_.down->allocatePstate(this);

  step_ = INITIAL_;
}

ExExeUtilHiveTruncateTcb::~ExExeUtilHiveTruncateTcb()
{
  freeResources();
}

void ExExeUtilHiveTruncateTcb::freeResources()
{
  if (htTdb().getDropOnDealloc())
  {
    NAString hiveDropDDL("drop table ");
    hiveDropDDL += htTdb().getHiveTableName();

    // TODO: is it ok to ignore the error 
    HiveClient_JNI::executeHiveSQL(hiveDropDDL);
  }
}

//////////////////////////////////////////////////////
// work() for ExExeUtilHiveTruncateTcb
//////////////////////////////////////////////////////
short ExExeUtilHiveTruncateTcb::work()
{
  short rc = 0;
  Lng32 cliRC = 0;

  // if no parent request, return
  if (qparent_.down->isEmpty())
    return WORK_OK;

  // if no room in up queue, won't be able to return data/status.
  // Come back later.
  if (qparent_.up->isFull())
    return WORK_OK;

  ex_queue_entry * pentry_down = qparent_.down->getHeadEntry();
  ExExeUtilPrivateState & pstate = *((ExExeUtilPrivateState*) pentry_down->pstate);

  while (1)
    {
      switch (step_)
        {
        case INITIAL_:
          {
            // if 'if exists' clause was specified and table does not exist
            // during compile phase, return.
            // If table was missing during compile and was created before
            // execute, then QI/AQR/Timestamp check will recompile.
            if (htTdb().getIfExists() && htTdb().getTableNotExists())
              {
                step_ = DONE_;
                break;
              }

            if (htTdb().getIsExternal())
              step_ = ALTER_TO_MANAGED_;
            else
              step_ = TRUNCATE_TABLE_;
          }
          break;
          
        case ALTER_TO_MANAGED_:
          {
            // A Hive table can be an External or Managed table.
            // Currently, an External Hive table cannot be truncated.
            // Maybe some future Hive version will allow that.
            // Temporarily change the table attribute to be Managed,
            // truncate the table and then change it back to be External.
            NAString alterStmt("alter table ");
            alterStmt += htTdb().getHiveTableName(); 
            alterStmt += " set tblproperties ('EXTERNAL'='False')";
            if (HiveClient_JNI::executeHiveSQL(alterStmt.data()) != HVC_OK)
              {
                // alter failed
                ExRaiseSqlError(getHeap(), &diagsArea_, -1214,
                                NULL, NULL, NULL,
                                getSqlJniErrorStr(), htTdb().getHiveTruncQuery()); 
                step_ = ERROR_;
                break;
              }
            
            step_ = TRUNCATE_TABLE_;
          }
          break;

        case TRUNCATE_TABLE_:
          {
            if (HiveClient_JNI::executeHiveSQL(htTdb().getHiveTruncQuery()) != HVC_OK)
              {
                ExRaiseSqlError(getHeap(), &diagsArea_, -1214,
                                NULL, NULL, NULL,
                                getSqlJniErrorStr(), htTdb().getHiveTruncQuery());

                if (htTdb().getIsExternal())
                  {
                    step_ = ALTER_TO_EXTERNAL_AND_ERROR_;
                    break;
                  }
                
                step_ = ERROR_;
                break;
              }
 
            if (htTdb().getIsExternal())
              step_ = ALTER_TO_EXTERNAL_;
            else
              step_= DONE_;
          }
          break;
          
         case ALTER_TO_EXTERNAL_:
         case ALTER_TO_EXTERNAL_AND_ERROR_:
          {
            // table was altered to Managed. Alter it back to External.
            NAString alterStmt("alter table ");
            alterStmt += htTdb().getHiveTableName(); 
            alterStmt += " set tblproperties ('EXTERNAL'='TRUE')";
             if (HiveClient_JNI::executeHiveSQL(alterStmt.data()) != HVC_OK)
              {
                // alter failed
                ExRaiseSqlError(getHeap(), &diagsArea_, -1214,
                                NULL, NULL, NULL,
                                getSqlJniErrorStr(), alterStmt.data());
                step_ = ERROR_;
                break;
              }                    

            if (step_ == ALTER_TO_EXTERNAL_AND_ERROR_)
              step_ = ERROR_;
            else
              step_ = DONE_;
          }
          break;

        case ERROR_:
          {
            if (handleError())
              return WORK_OK;
            step_ = DONE_;
          }
          break;
          
        case DONE_:
          {
            if (handleDone())
              return WORK_OK;
            step_ = INITIAL_;
            
            return WORK_OK;
          }
          break;
          
        } // switch
    } // while
  
}

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

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

/////////////////////////////////////////////////////////////////////////////
// Constructor and destructor for ExeUtil_private_state
/////////////////////////////////////////////////////////////////////////////
ExExeUtilHiveTruncatePrivateState::ExExeUtilHiveTruncatePrivateState()
{
}

ExExeUtilHiveTruncatePrivateState::~ExExeUtilHiveTruncatePrivateState()
{
};

///////////////////////////////////////////////////////////////////////
// ExExeUtilHiveQueryTdb
///////////////////////////////////////////////////////////////////////
ex_tcb * ExExeUtilHiveQueryTdb::build(ex_globals * glob)
{
  ExExeUtilTcb * exe_util_tcb;
  exe_util_tcb = new(glob->getSpace()) ExExeUtilHiveQueryTcb(*this, glob);
  exe_util_tcb->registerSubtasks();
  return (exe_util_tcb);
}
ExExeUtilHiveQueryTcb::ExExeUtilHiveQueryTcb(
     const ComTdbExeUtilHiveQuery & exe_util_tdb,
     ex_globals * glob)
     : ExExeUtilTcb( exe_util_tdb, NULL, glob)
{
  qparent_.down->allocatePstate(this);
  step_ = INITIAL_;
}
ExExeUtilHiveQueryTcb::~ExExeUtilHiveQueryTcb()
{
}
short ExExeUtilHiveQueryTcb::work()
{
  short rc = 0;
  Lng32 cliRC = 0;
  if (qparent_.down->isEmpty())
    return WORK_OK;
  if (qparent_.up->isFull())
    return WORK_OK;
  ex_queue_entry * pentry_down = qparent_.down->getHeadEntry();
  ExExeUtilPrivateState & pstate = *((ExExeUtilPrivateState*) pentry_down->pstate);
  while (1)
    {
      switch (step_)
        {
        case INITIAL_:
          {
            step_ = PROCESS_QUERY_;
          }
          break;
        case PROCESS_QUERY_:
          {
            if (HiveClient_JNI::executeHiveSQL(htTdb().getHiveQuery()) != HVC_OK)
            {
                ExRaiseSqlError(getHeap(), &diagsArea_, -1214,
                        NULL, NULL, NULL,
                        getSqlJniErrorStr(), htTdb().getHiveQuery()); 
                step_ = ERROR_;
                break;
            }
            step_ = DONE_;
          }
          break;
        case ERROR_:
          {
            if (handleError())
              return WORK_OK;
            step_ = DONE_;
          }
          break;
        case DONE_:
          {
            if (handleDone())
              return WORK_OK;
            step_ = INITIAL_;
            return WORK_OK;
          }
          break;
        } // switch
    } // while
}
ex_tcb_private_state * ExExeUtilHiveQueryTcb::allocatePstates(
     Lng32 &numElems,      // inout, desired/actual elements
     Lng32 &pstateLength)  // out, length of one element
{
  PstateAllocator<ExExeUtilHiveQueryPrivateState> pa;
  return pa.allocatePstates(this, numElems, pstateLength);
}
ExExeUtilHiveQueryPrivateState::ExExeUtilHiveQueryPrivateState()
{
}
ExExeUtilHiveQueryPrivateState::~ExExeUtilHiveQueryPrivateState()
{
};