simba.cc

Go to the documentation of this file.

// =================================================================================================
// =================================================================================================
#include <stdio.h>

#include "boost/shared_ptr.hpp"
#include "boost/thread.hpp"
#include "thrift/protocol/TBinaryProtocol.h"
//#include "thrift/transport/sasltransport/TSaslClientTransport.h"
#include "thrift/transport/TTransportUtils.h"
#include "thrift/transport/TSocket.h"
#include "gen-cpp/TCLIService.h"

using namespace apache::hive::service::cli::thrift;
using namespace apache::thrift;
using namespace apache::thrift::transport;
using namespace apache::thrift::protocol;
using namespace boost;


void CloseOperation(apache::hive::service::cli::thrift::TFetchResultsReq& fetchResultsReq, TCLIServiceClient& impalaClient)
{
  try
  {
    TCloseOperationReq closeOperationReq;
    TCloseOperationResp closeOperationResp;

    closeOperationReq.operationHandle = fetchResultsReq.operationHandle;

    impalaClient.CloseOperation(closeOperationResp, closeOperationReq);
  }
  catch (TException& ex)
  {
    printf("\nClose Operation failed from thread\n\n");
  }
}

void FetchFunction(apache::hive::service::cli::thrift::TOperationHandle opHandle, shared_ptr<TCLIServiceClient> impalaClient,
    shared_ptr<mutex> m){

    try
    {
      size_t numRows = 0;
      bool hasMoreRows = true;

      TFetchResultsResp fetchResultsResp;

      apache::hive::service::cli::thrift::TFetchResultsReq fetchResultsReq;
      fetchResultsReq.orientation = TFetchOrientation::FETCH_NEXT;
      fetchResultsReq.__set_operationHandle(opHandle);
      fetchResultsReq.maxRows = 1000;
      {
        unique_lock<mutex> l(*m);
        impalaClient->FetchResults(fetchResultsResp, fetchResultsReq);
      }

      if ((TStatusCode::SUCCESS_STATUS == fetchResultsResp.status.statusCode) ||
          (TStatusCode::SUCCESS_WITH_INFO_STATUS == fetchResultsResp.status.statusCode) )
      {
        unique_lock<mutex> l(*m);
        numRows = fetchResultsResp.results.rows.size();
        hasMoreRows = 0 < numRows;

        if (!hasMoreRows)
        {
          printf("\nNo More Rows, closing operation\n\n");
          CloseOperation(fetchResultsReq, *impalaClient);
        }
      }
      else
      {
        unique_lock<mutex> l(*m);
        printf("\nNo Success\n\n");
        CloseOperation(fetchResultsReq, *impalaClient);
      }
    }
    catch (...)
    {
        printf("\nTransport exception while fetching\n\n");
    }
}

void ExecuteAndCancelOperationTest()
{
    shared_ptr<TTransport> transport;
    shared_ptr<mutex> mutex(new mutex);

    apache::hive::service::cli::thrift::TOpenSessionResp openSessionResp;
    apache::hive::service::cli::thrift::TExecuteStatementResp executeResponse;

    try
    {
      shared_ptr<TSocket> socket(new TSocket("localhost", 21050));

      transport.reset(new TBufferedTransport(socket));
      transport->open();
    }
    catch (const TException& e)
    {
      printf("\n Error in the transport open \n\n");
    }

    shared_ptr<TProtocol> protocol(new TBinaryProtocol(transport));
    shared_ptr<TCLIServiceClient> client(new TCLIServiceClient(protocol));

    TOpenSessionReq request;

    try
    {
      request.__set_username("");
      request.__set_password("");

      client->OpenSession(openSessionResp, request);
    }
    catch (TException& ex)
    {
      printf("\nOpenSession Failed\n\n");
    }

    TExecuteStatementReq queryRequest;

    try
    {
      queryRequest.__set_sessionHandle(openSessionResp.sessionHandle);
      queryRequest.__set_statement("SELECT COUNT(*) FROM tpch.lineitem"); //BIM_REPOSITORY.BIM_InstanceSettings WHERE Setting='BIM_REPOSITORY_VERSION' AND SettingValue='C'");

      client->ExecuteStatement(executeResponse, queryRequest);
    }

    catch (const TTransportException& e)
    {
      printf("\nTransport error\n\n");
    }

    thread workerThread(FetchFunction, executeResponse.operationHandle, client, mutex);

    try
    {
      unique_lock<mutex> l(*mutex);
      TGetOperationStatusReq statusReq;
      TGetOperationStatusResp statusResp;

      statusReq.__set_operationHandle(executeResponse.operationHandle);
      printf("GET OPERATION STATUS\n");
      client->GetOperationStatus(statusResp, statusReq);
      printf("OPERATION STATUS RETURNED\n");
      if ((TOperationState::RUNNING_STATE == statusResp.operationState) ||
          (TOperationState::INITIALIZED_STATE == statusResp.operationState))
      {
        TCancelOperationReq cancelReq;
        TCancelOperationResp cancelResp;

        cancelReq.__set_operationHandle(executeResponse.operationHandle);
        client->CancelOperation(cancelResp, cancelReq);
      }
    }
    catch(...)
    {
      printf("Cancellation error\n");
    }
}



int main( int argc, const char* argv[] )
{
    // Loop to run test behavior repeatedly until the driver experiences a "hang"
    // This loop of 100 iterations should take < 1 minute; otherwise the driver is hung up
    for (int testLoop = 0; testLoop < 100; testLoop++)
    {
        ExecuteAndCancelOperationTest();
    }
}