trafodion-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From se...@apache.org
Subject [1/3] trafodion git commit: [TRAFODION-3009] Streamline error handling in Executor utility commands
Date Thu, 19 Apr 2018 01:26:25 GMT
Repository: trafodion
Updated Branches:
  refs/heads/master 05d92ae7b -> c29ebc6e5


[TRAFODION-3009] Streamline error handling in Executor utility commands

Changes to avoid allocation of ComDiagsArea in some more places unless it is
needed to pass errors or warnings.

[TRAFODION-2917] Refactor Trafodion implementation of hdfs scan for text formatted hive tables

Disabling USE_LIBHDFS_SCAN by default. The new implementation of Hdfs Scan is now used to
scan the text type hive files.

Hdfs Scan is now stopped gracefully when the hive scan is cancelled. This avoids the random
core seen with new implementation.


Project: http://git-wip-us.apache.org/repos/asf/trafodion/repo
Commit: http://git-wip-us.apache.org/repos/asf/trafodion/commit/060bfc6c
Tree: http://git-wip-us.apache.org/repos/asf/trafodion/tree/060bfc6c
Diff: http://git-wip-us.apache.org/repos/asf/trafodion/diff/060bfc6c

Branch: refs/heads/master
Commit: 060bfc6c41277907af523d7bb62e02eb91b5bc86
Parents: 6e39af2
Author: selvaganesang <selva.govindarajan@esgyn.com>
Authored: Tue Apr 17 05:09:04 2018 +0000
Committer: selvaganesang <selva.govindarajan@esgyn.com>
Committed: Tue Apr 17 05:09:04 2018 +0000

----------------------------------------------------------------------
 core/sql/executor/ExExeUtilCli.cpp              |   4 +-
 core/sql/executor/ExExeUtilLoad.cpp             | 202 ++-----------------
 core/sql/executor/ExHdfsScan.cpp                |   2 +
 core/sql/sqlcomp/nadefaults.cpp                 |   2 +-
 .../main/java/org/trafodion/sql/HDFSClient.java |   2 +-
 5 files changed, 27 insertions(+), 185 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/trafodion/blob/060bfc6c/core/sql/executor/ExExeUtilCli.cpp
----------------------------------------------------------------------
diff --git a/core/sql/executor/ExExeUtilCli.cpp b/core/sql/executor/ExExeUtilCli.cpp
index aa1553b..eb7ae04 100644
--- a/core/sql/executor/ExExeUtilCli.cpp
+++ b/core/sql/executor/ExExeUtilCli.cpp
@@ -1094,7 +1094,7 @@ Lng32 ExeCliInterface::executeImmediate(const char * stmtStr,
   Lng32 retcode = 0;
 
   ComDiagsArea * tempDiags = NULL;
-  if (globalDiags != NULL && *globalDiags != NULL)
+  if (globalDiags != NULL && *globalDiags != NULL && (*globalDiags)->getNumber()
> 0)
   {
      tempDiags = ComDiagsArea::allocate(heap_);
      tempDiags->mergeAfter(**globalDiags);
@@ -1119,7 +1119,7 @@ ExecuteImmediateReturn:
     {
       // Allocate the diagnostics area if needed
       // and populate the diagnostics conditions
-      if (*globalDiags == NULL && retcode != 0) {
+      if (*globalDiags == NULL && retcode != 0 && retcode != 100) {
          *globalDiags = ComDiagsArea::allocate(getHeap());
          SQL_EXEC_MergeDiagnostics_Internal(**globalDiags);
       }

http://git-wip-us.apache.org/repos/asf/trafodion/blob/060bfc6c/core/sql/executor/ExExeUtilLoad.cpp
----------------------------------------------------------------------
diff --git a/core/sql/executor/ExExeUtilLoad.cpp b/core/sql/executor/ExExeUtilLoad.cpp
index 5f33b72..5024f8c 100644
--- a/core/sql/executor/ExExeUtilLoad.cpp
+++ b/core/sql/executor/ExExeUtilLoad.cpp
@@ -124,9 +124,7 @@ short ExExeUtilCreateTableAsTcb::work()
   ExExeStmtGlobals *exeGlob = getGlobals()->castToExExeStmtGlobals();
   ExMasterStmtGlobals *masterGlob = exeGlob->castToExMasterStmtGlobals();
   ContextCli *currContext = masterGlob->getStatement()->getContext();
-  
-  ExTransaction *ta = getGlobals()->castToExExeStmtGlobals()->
-    castToExMasterStmtGlobals()->getStatement()->getContext()->getTransaction();
+  ExTransaction *ta = currContext->getTransaction();
   
   while (1)
     {
@@ -535,69 +533,24 @@ short ExExeUtilCreateTableAsTcb::work()
 
 	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();
-	    
+	    retcode = handleDone();
+	    if (retcode == 1)
+	       return WORK_OK;
 	    step_ = INITIAL_;
-	    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;
-
-	    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();
-	    
-	    pstate.matches_ = 0;
-
+	    retcode = handleError();
+	    if (retcode == 1)
+	       return WORK_OK;
 	    step_ = DONE_;
 	  }
 	break;
-
-
 	} // switch
     } // while
-
-  
   
   return WORK_OK;
 
@@ -1679,74 +1632,20 @@ short ExExeUtilHBaseBulkLoadTcb::work()
 
     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;
-
-      ComDiagsArea *diagsArea = up_entry->getDiagsArea();
-
-      if (diagsArea == NULL)
-        diagsArea = ComDiagsArea::allocate(getMyHeap());
-      else
-        diagsArea->incrRefCount(); // setDiagsArea call below will decr ref count
-
-      if (getDiagsArea())
-        diagsArea->mergeAfter(*getDiagsArea());
-
+      retcode = handleDone();
+      if (retcode == 1)
+         return WORK_OK;
       masterGlob->setRowsAffected(rowsAffected_);
-
-      up_entry->setDiagsArea(diagsArea);
-
-      // insert into parent
-      qparent_.up->insert();
       step_ = INITIAL_;
-      qparent_.down->removeHead();
       return WORK_OK;
     }
     break;
 
     case LOAD_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;
-
-      ComDiagsArea *diagsArea = up_entry->getDiagsArea();
-
-      if (diagsArea == NULL)
-        diagsArea = ComDiagsArea::allocate(getMyHeap());
-      else
-        diagsArea->incrRefCount(); // setDiagsArea call below will decr ref count
-
-      if (getDiagsArea())
-        {
-        diagsArea->mergeAfter(*getDiagsArea());
-          diagsArea->setRowCount(rowsAffected_);
-        }
-
-      up_entry->setDiagsArea(diagsArea);
-
-      // insert into parent
-      qparent_.up->insert();
-
-      pstate.matches_ = 0;
-
-
-
+      retcode = handleError();
+      if (retcode == 1)
+         return WORK_OK;
       step_ = DONE_;
     }
     break;
@@ -2153,9 +2052,8 @@ short ExExeUtilHBaseBulkUnLoadTcb::work()
 
   ex_queue_entry * pentry_down = qparent_.down->getHeadEntry();
   ExExeUtilPrivateState & pstate = *((ExExeUtilPrivateState*) pentry_down->pstate);
-
-  ExTransaction *ta = getGlobals()->castToExExeStmtGlobals()->
-      castToExMasterStmtGlobals()->getStatement()->getContext()->getTransaction();
+  ExMasterStmtGlobals *masterGlob = getGlobals()->castToExExeStmtGlobals()->castToExMasterStmtGlobals();
+  ExTransaction *ta = masterGlob->getStatement()->getContext()->getTransaction();
 
   while (1)
   {
@@ -2177,7 +2075,6 @@ short ExExeUtilHBaseBulkUnLoadTcb::work()
       {
          ExHbaseAccessTcb::setupError((NAHeap *)getMyHeap(),qparent_, retcode, 
                 "ExpHbaseInterface_JNI::init"); 
-         handleError();
          step_ = UNLOAD_END_ERROR_;
          break;
       }
@@ -2327,7 +2224,6 @@ short ExExeUtilHBaseBulkUnLoadTcb::work()
           ExHbaseAccessTcb::setupError((NAHeap *)getMyHeap(),qparent_, hbcRetCode, 
                 "HBaseClient_JNI::createSnapshot/verifySnapshot", 
                 snapshotsList_->at(i)->snapshotName->data() );
-          handleError();
           step_ = UNLOAD_END_ERROR_;
           break;
         }
@@ -2389,7 +2285,6 @@ short ExExeUtilHBaseBulkUnLoadTcb::work()
           ExHbaseAccessTcb::setupError((NAHeap *)getMyHeap(),qparent_, hbcRetCode, 
                 "HBaseClient_JNI::createSnapshot/verifySnapshot", 
                 snapshotsList_->at(i)->snapshotName->data() );
-          handleError();
           step_ = UNLOAD_END_ERROR_;
           break;
         }
@@ -2477,36 +2372,11 @@ short ExExeUtilHBaseBulkUnLoadTcb::work()
 
     case DONE_:
     {
-      if (qparent_.up->isFull())
+      retcode = handleDone();
+      if (retcode == 1)
         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;
-
-      ComDiagsArea *diagsArea = up_entry->getDiagsArea();
-
-      if (diagsArea == NULL)
-        diagsArea = ComDiagsArea::allocate(getMyHeap());
-      else
-        diagsArea->incrRefCount(); // setDiagsArea call below will decr ref count
-
-      diagsArea->setRowCount(rowsAffected_);
-
-      if (getDiagsArea())
-        diagsArea->mergeAfter(*getDiagsArea());
-
-      up_entry->setDiagsArea(diagsArea);
-
-      // insert into parent
-      qparent_.up->insert();
+      masterGlob->setRowsAffected(rowsAffected_);
       step_ = INITIAL_;
-      qparent_.down->removeHead();
-
       freeResources();
       return WORK_OK;
     }
@@ -2514,36 +2384,9 @@ short ExExeUtilHBaseBulkUnLoadTcb::work()
 
     case UNLOAD_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;
-
-      ComDiagsArea *diagsArea = up_entry->getDiagsArea();
-
-      if (diagsArea == NULL)
-        diagsArea = ComDiagsArea::allocate(getMyHeap());
-      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();
-
-      pstate.matches_ = 0;
-
-
-
+      retcode = handleError();
+      if (retcode == 1)
+         return WORK_OK;
       step_ = DONE_;
     }
     break;
@@ -2552,7 +2395,6 @@ short ExExeUtilHBaseBulkUnLoadTcb::work()
   } // while
 
   return WORK_OK;
-
 }
 
 short ExExeUtilHBaseBulkUnLoadTcb::moveRowToUpQueue(const char * row, Lng32 len,
@@ -3447,7 +3289,6 @@ short ExExeUtilLobExtractTcb::work()
 	    retcode = handleError();
 	    if (retcode == 1)
 	      return WORK_OK;
-
 	    step_ = DONE_;
 	  }
 	  break;
@@ -3457,7 +3298,6 @@ short ExExeUtilLobExtractTcb::work()
 	    retcode = handleDone();
 	    if (retcode == 1)
 	      return WORK_OK;
-
 	    step_ = EMPTY_;
 	    return WORK_OK;
 	  }

http://git-wip-us.apache.org/repos/asf/trafodion/blob/060bfc6c/core/sql/executor/ExHdfsScan.cpp
----------------------------------------------------------------------
diff --git a/core/sql/executor/ExHdfsScan.cpp b/core/sql/executor/ExHdfsScan.cpp
index f77d714..d09f6cd 100644
--- a/core/sql/executor/ExHdfsScan.cpp
+++ b/core/sql/executor/ExHdfsScan.cpp
@@ -255,6 +255,8 @@ ExHdfsScanTcb::~ExHdfsScanTcb()
 
 void ExHdfsScanTcb::freeResources()
 {
+  if (hdfsScan_ != NULL)
+     hdfsScan_->stop();
   if (loggingFileName_ != NULL) {
      NADELETEBASIC(loggingFileName_, getHeap());
      loggingFileName_ = NULL;

http://git-wip-us.apache.org/repos/asf/trafodion/blob/060bfc6c/core/sql/sqlcomp/nadefaults.cpp
----------------------------------------------------------------------
diff --git a/core/sql/sqlcomp/nadefaults.cpp b/core/sql/sqlcomp/nadefaults.cpp
index 244a2bc..0d7b41b 100644
--- a/core/sql/sqlcomp/nadefaults.cpp
+++ b/core/sql/sqlcomp/nadefaults.cpp
@@ -3033,7 +3033,7 @@ XDDkwd__(SUBQUERY_UNNESTING,			"ON"),
   // Use large queues on RHS of Flow/Nested Join when appropriate
   DDkwd__(USE_LARGE_QUEUES,                     "ON"),
 
-  DDkwd__(USE_LIBHDFS_SCAN,                     "ON"),
+  DDkwd__(USE_LIBHDFS_SCAN,                     "OFF"),
 
   DDkwd__(USE_MAINTAIN_CONTROL_TABLE,          "OFF"),
 

http://git-wip-us.apache.org/repos/asf/trafodion/blob/060bfc6c/core/sql/src/main/java/org/trafodion/sql/HDFSClient.java
----------------------------------------------------------------------
diff --git a/core/sql/src/main/java/org/trafodion/sql/HDFSClient.java b/core/sql/src/main/java/org/trafodion/sql/HDFSClient.java
index ff78d3d..0346bef 100644
--- a/core/sql/src/main/java/org/trafodion/sql/HDFSClient.java
+++ b/core/sql/src/main/java/org/trafodion/sql/HDFSClient.java
@@ -470,7 +470,7 @@ public class HDFSClient
    {
       if (future_ != null) {
          try {
-           future_.get(200, TimeUnit.MILLISECONDS);
+           future_.get(30, TimeUnit.SECONDS);
          } catch(TimeoutException e) {
             logger_.error("Asynchronous Thread of HdfsScan is Cancelled (timeout), ", e);
             future_.cancel(true);


Mime
View raw message