You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@trafodion.apache.org by se...@apache.org on 2018/03/02 06:57:18 UTC
[1/4] trafodion git commit: [TRAFODION-2917] Refactor Trafodion
implementation of hdfs scan for text formatted hive tables
Repository: trafodion
Updated Branches:
refs/heads/master d48a88741 -> e618aaf3d
[TRAFODION-2917] Refactor Trafodion implementation of hdfs scan for text formatted hive tables
Project: http://git-wip-us.apache.org/repos/asf/trafodion/repo
Commit: http://git-wip-us.apache.org/repos/asf/trafodion/commit/ac706607
Tree: http://git-wip-us.apache.org/repos/asf/trafodion/tree/ac706607
Diff: http://git-wip-us.apache.org/repos/asf/trafodion/diff/ac706607
Branch: refs/heads/master
Commit: ac7066074611a09da33adf88673c2f023e7dda7d
Parents: 75c7b39
Author: selvaganesang <se...@esgyn.com>
Authored: Wed Feb 28 19:27:40 2018 +0000
Committer: selvaganesang <se...@esgyn.com>
Committed: Wed Feb 28 21:57:59 2018 +0000
----------------------------------------------------------------------
core/sql/cli/Context.h | 1 +
core/sql/cli/Globals.cpp | 1 +
core/sql/cli/Globals.h | 1 +
core/sql/comexe/ComTdbHdfsScan.h | 5 +-
core/sql/common/ComRtUtils.cpp | 1 +
core/sql/executor/ExExeUtilGet.cpp | 2 -
core/sql/executor/ExFastTransport.cpp | 7 +--
core/sql/executor/ExHdfsScan.cpp | 58 ++++++++++++++++++--
core/sql/executor/ExHdfsScan.h | 23 ++++++++
core/sql/executor/HBaseClient_JNI.cpp | 1 -
core/sql/executor/HdfsClient_JNI.cpp | 28 +++++-----
core/sql/exp/ExpLOBaccess.cpp | 1 +
core/sql/optimizer/HDFSHook.cpp | 1 +
core/sql/optimizer/NATable.cpp | 1 +
core/sql/sqlcomp/DefaultConstants.h | 1 +
.../main/java/org/trafodion/sql/HDFSClient.java | 13 +++--
16 files changed, 112 insertions(+), 33 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/trafodion/blob/ac706607/core/sql/cli/Context.h
----------------------------------------------------------------------
diff --git a/core/sql/cli/Context.h b/core/sql/cli/Context.h
index 973a647..bcfd06b 100644
--- a/core/sql/cli/Context.h
+++ b/core/sql/cli/Context.h
@@ -512,6 +512,7 @@ private:
NAString jniErrorStr_;
HBaseClient_JNI *hbaseClientJNI_;
HiveClient_JNI *hiveClientJNI_;
+ HdfsClient *hdfsClientJNI_;
// this points to data used by trafSE (traf storage engine) that is context specific.
// It points to a list of 'black box' of data allocated by user and is returned
http://git-wip-us.apache.org/repos/asf/trafodion/blob/ac706607/core/sql/cli/Globals.cpp
----------------------------------------------------------------------
diff --git a/core/sql/cli/Globals.cpp b/core/sql/cli/Globals.cpp
index ed30502..fb09008 100644
--- a/core/sql/cli/Globals.cpp
+++ b/core/sql/cli/Globals.cpp
@@ -60,6 +60,7 @@
#include <semaphore.h>
#include <pthread.h>
#include "HBaseClient_JNI.h"
+#include "HdfsClient_JNI.h"
#include "LmLangManagerC.h"
#include "LmLangManagerJava.h"
#include "CliSemaphore.h"
http://git-wip-us.apache.org/repos/asf/trafodion/blob/ac706607/core/sql/cli/Globals.h
----------------------------------------------------------------------
diff --git a/core/sql/cli/Globals.h b/core/sql/cli/Globals.h
index 284d992..b8e04ea 100644
--- a/core/sql/cli/Globals.h
+++ b/core/sql/cli/Globals.h
@@ -88,6 +88,7 @@ class CliGlobals;
class CLISemaphore;
class HBaseClient_JNI;
class HiveClient_JNI;
+class HdfsClient;
class TransMode;
class ContextTidMap;
class LmLanguageManager;
http://git-wip-us.apache.org/repos/asf/trafodion/blob/ac706607/core/sql/comexe/ComTdbHdfsScan.h
----------------------------------------------------------------------
diff --git a/core/sql/comexe/ComTdbHdfsScan.h b/core/sql/comexe/ComTdbHdfsScan.h
index 86534be..46d7f2f 100755
--- a/core/sql/comexe/ComTdbHdfsScan.h
+++ b/core/sql/comexe/ComTdbHdfsScan.h
@@ -54,7 +54,8 @@ class ComTdbHdfsScan : public ComTdb
CONTINUE_ON_ERROR = 0x0020,
LOG_ERROR_ROWS = 0x0040,
ASSIGN_RANGES_AT_RUNTIME = 0x0080,
- USE_LIBHDFS_SCAN = 0x0100
+ TREAT_EMPTY_AS_NULL = 0x0100,
+ USE_LIBHDFS_SCAN = 0x0200
};
// Expression to filter rows.
@@ -288,7 +289,7 @@ public:
{(v ? flags_ |= USE_LIBHDFS_SCAN : flags_ &= ~USE_LIBHDFS_SCAN); }
NABoolean getUseLibhdfsScan() const
{ return (flags_ & USE_LIBHDFS_SCAN) != 0; }
-
+
UInt32 getMaxErrorRows() const { return maxErrorRows_;}
void setMaxErrorRows(UInt32 v ) { maxErrorRows_= v; }
http://git-wip-us.apache.org/repos/asf/trafodion/blob/ac706607/core/sql/common/ComRtUtils.cpp
----------------------------------------------------------------------
diff --git a/core/sql/common/ComRtUtils.cpp b/core/sql/common/ComRtUtils.cpp
index f2619f9..35f9ca7 100644
--- a/core/sql/common/ComRtUtils.cpp
+++ b/core/sql/common/ComRtUtils.cpp
@@ -81,6 +81,7 @@
#include "seabed/ms.h"
#include "seabed/fs.h"
+#include "HdfsClient_JNI.h"
struct ModName {
public:
const char * name;
http://git-wip-us.apache.org/repos/asf/trafodion/blob/ac706607/core/sql/executor/ExExeUtilGet.cpp
----------------------------------------------------------------------
diff --git a/core/sql/executor/ExExeUtilGet.cpp b/core/sql/executor/ExExeUtilGet.cpp
index 539a8cf..cd20fae 100644
--- a/core/sql/executor/ExExeUtilGet.cpp
+++ b/core/sql/executor/ExExeUtilGet.cpp
@@ -6103,7 +6103,6 @@ ExExeUtilRegionStatsTcb::ExExeUtilRegionStatsTcb(
ehi_ = ExpHbaseInterface::newInstance(glob->getDefaultHeap(),
(char*)"", //exe_util_tdb.server(),
(char*)""); //exe_util_tdb.zkPort(),
-
regionInfoList_ = NULL;
tableName_ = new(glob->getDefaultHeap()) char[2000];
@@ -6876,7 +6875,6 @@ ExExeUtilClusterStatsTcb::ExExeUtilClusterStatsTcb(
ehi_ = ExpHbaseInterface::newInstance(glob->getDefaultHeap(),
(char*)"", //exe_util_tdb.server(),
(char*)""); //exe_util_tdb.zkPort());
-
regionInfoList_ = NULL;
// get hbase rootdir location. Max linux pathlength is 1024.
http://git-wip-us.apache.org/repos/asf/trafodion/blob/ac706607/core/sql/executor/ExFastTransport.cpp
----------------------------------------------------------------------
diff --git a/core/sql/executor/ExFastTransport.cpp b/core/sql/executor/ExFastTransport.cpp
index 9d43545..ba4cfbb 100644
--- a/core/sql/executor/ExFastTransport.cpp
+++ b/core/sql/executor/ExFastTransport.cpp
@@ -827,12 +827,11 @@ ExWorkProcRetcode ExHdfsFastExtractTcb::work()
}
}
else
- {
+ {
updateWorkATPDiagsArea(__FILE__,__LINE__,"sockets are not supported");
pstate.step_ = EXTRACT_ERROR;
break;
- }
-
+ }
for (UInt32 i = 0; i < myTdb().getChildTuple()->numAttrs(); i++)
{
Attributes * attr = myTdb().getChildTableAttr(i);
@@ -1290,7 +1289,7 @@ void ExHdfsFastExtractTcb::createHdfsClientFileError(Int32 hdfsClientRetCode)
(ExeErrorCode)(8447),
NULL, NULL, NULL, NULL,
errorMsg,
- (char *)currContext->getJniErrorStr().data());
+ (char *)currContext->getJniErrorStr().data());
updateWorkATPDiagsArea(diagsArea);
}
http://git-wip-us.apache.org/repos/asf/trafodion/blob/ac706607/core/sql/executor/ExHdfsScan.cpp
----------------------------------------------------------------------
diff --git a/core/sql/executor/ExHdfsScan.cpp b/core/sql/executor/ExHdfsScan.cpp
index 2b73feb..3ff153e 100644
--- a/core/sql/executor/ExHdfsScan.cpp
+++ b/core/sql/executor/ExHdfsScan.cpp
@@ -317,7 +317,6 @@ void ExHdfsScanTcb::freeResources()
if (hdfsScan_ != NULL)
NADELETE(hdfsScan_, HdfsScan, getHeap());
}
-
NABoolean ExHdfsScanTcb::needStatsEntry()
{
// stats are collected for ALL and OPERATOR options.
@@ -1099,15 +1098,13 @@ ExWorkProcRetcode ExHdfsScanTcb::work()
if ((BYTE *)startOfNextRow > bufLogicalEnd_) {
step_ = TRAF_HDFS_READ;
hdfsBufNextRow_ = NULL;
- if (!exception_)
- break;
}
else
hdfsBufNextRow_ = startOfNextRow;
- }
+ }
}
-
+
if (exception_)
{
nextStep_ = step_;
@@ -2019,6 +2016,54 @@ void ExHdfsScanTcb::deallocateRuntimeRanges()
}
}
+void ExHdfsScanTcb::handleException(NAHeap *heap,
+ char *logErrorRow,
+ Lng32 logErrorRowLen,
+ ComCondition *errorCond)
+{
+ Lng32 errorMsgLen = 0;
+ charBuf *cBuf = NULL;
+ char *errorMsg;
+ HDFS_Client_RetCode hdfsClientRetcode;
+
+ if (loggingErrorDiags_ != NULL)
+ return;
+
+ if (!loggingFileCreated_) {
+ hdfsClient_ = HdfsClient::newInstance((NAHeap *)getHeap(), hdfsClientRetcode);
+ if (hdfsClientRetcode == HDFS_CLIENT_OK)
+ hdfsClientRetcode = hdfsClient_->hdfsCreate(loggingFileName_, FALSE);
+ if (hdfsClientRetcode == HDFS_CLIENT_OK)
+ loggingFileCreated_ = TRUE;
+ else
+ goto logErrorReturn;
+ }
+ hdfsClientRetcode = hdfsClient_->hdfsWrite(logErrorRow, logErrorRowLen);
+ if (hdfsClientRetcode != HDFS_CLIENT_OK)
+ goto logErrorReturn;
+ if (errorCond != NULL) {
+ errorMsgLen = errorCond->getMessageLength();
+ const NAWcharBuf wBuf((NAWchar*)errorCond->getMessageText(), errorMsgLen, heap);
+ cBuf = unicodeToISO88591(wBuf, heap, cBuf);
+ errorMsg = (char *)cBuf->data();
+ errorMsgLen = cBuf -> getStrLen();
+ errorMsg[errorMsgLen]='\n';
+ errorMsgLen++;
+ }
+ else {
+ errorMsg = (char *)"[UNKNOWN EXCEPTION]\n";
+ errorMsgLen = strlen(errorMsg);
+ }
+ hdfsClientRetcode = hdfsClient_->hdfsWrite(errorMsg, errorMsgLen);
+logErrorReturn:
+ if (hdfsClientRetcode != HDFS_CLIENT_OK) {
+ loggingErrorDiags_ = ComDiagsArea::allocate(heap);
+ *loggingErrorDiags_ << DgSqlCode(EXE_ERROR_WHILE_LOGGING)
+ << DgString0(loggingFileName_)
+ << DgString1((char *)GetCliGlobals()->currContext()->getJniErrorStr().data());
+ }
+}
+
short ExHdfsScanTcb::moveRowToUpQueue(const char * row, Lng32 len,
short * rc, NABoolean isVarchar)
{
@@ -2119,7 +2164,7 @@ short ExHdfsScanTcb::handleDone(ExWorkProcRetcode &rc)
return 0;
}
-
+/*
void ExHdfsScanTcb::handleException(NAHeap *heap,
char *logErrorRow,
Lng32 logErrorRowLen,
@@ -2167,6 +2212,7 @@ logErrorReturn:
<< DgString1((char *)GetCliGlobals()->currContext()->getJniErrorStr().data());
}
}
+*/
////////////////////////////////////////////////////////////////////////
// ORC files
http://git-wip-us.apache.org/repos/asf/trafodion/blob/ac706607/core/sql/executor/ExHdfsScan.h
----------------------------------------------------------------------
diff --git a/core/sql/executor/ExHdfsScan.h b/core/sql/executor/ExHdfsScan.h
index f4ad7e1..04b632e 100644
--- a/core/sql/executor/ExHdfsScan.h
+++ b/core/sql/executor/ExHdfsScan.h
@@ -135,6 +135,29 @@ class ExHdfsScanTcb : public ex_tcb
public:
enum
+/*
+ USE_LIBHDFS_SCAN - OFF enables hdfs access via java classes
+ org.trafodion.sql.HdfsScan and org.trafodion.sql.HdfsClient
+ Steps involved:
+ 1. Create a new HdfsScan object and set the scan ranges of the fragment instance in it
+ The scan range involves the following and it is determined either at runtime or compile time
+ a) filename
+ b) offset
+ c) len
+ Java layer always reads more than the len by rangeTailIOSize_ to accommdate the record split
+ 2. Two ByteBuffer objects are also passsed to HdfsScan object. These ByteBuffers are backed up by
+ 2 native buffers where the data is fetched. The buffer has a head room of size rangeTailIOSize_ and the
+ data is always read after the head room.
+ 3. HdfsScan returns an int array containing bytesRead, bufNo, rangeNo, isEOF and schedules either
+ the remaining bytes to be read or the next range using ByteBuffers alternatively.
+ 4. HdfsScan returns null array when there is no more data to be read.
+ 5. When the data is processed in one ByteBuffer in the native thread, the data is fetched into the other ByteBuffer by
+ another Java thread.
+ 6. Native layer after processing all the rows in one ByteBuffer, moves the last incomplete row to head room of the
+ other ByteBuffer. Then it requests to check if the read is complete. The native layer processes the buffer starting
+ from the copied incomplete row.
+*/
+
{
BYTES_COMPLETED,
BUF_NO,
http://git-wip-us.apache.org/repos/asf/trafodion/blob/ac706607/core/sql/executor/HBaseClient_JNI.cpp
----------------------------------------------------------------------
diff --git a/core/sql/executor/HBaseClient_JNI.cpp b/core/sql/executor/HBaseClient_JNI.cpp
index 6b400cd..8d12821 100644
--- a/core/sql/executor/HBaseClient_JNI.cpp
+++ b/core/sql/executor/HBaseClient_JNI.cpp
@@ -5166,4 +5166,3 @@ void deleteNAArray(CollHeap *heap, NAArray<HbaseStr> *array)
}
NADELETE(array, NAArray, heap);
}
-
http://git-wip-us.apache.org/repos/asf/trafodion/blob/ac706607/core/sql/executor/HdfsClient_JNI.cpp
----------------------------------------------------------------------
diff --git a/core/sql/executor/HdfsClient_JNI.cpp b/core/sql/executor/HdfsClient_JNI.cpp
index f08aa92..5f0f810 100644
--- a/core/sql/executor/HdfsClient_JNI.cpp
+++ b/core/sql/executor/HdfsClient_JNI.cpp
@@ -23,6 +23,7 @@
#include "QRLogger.h"
#include "Globals.h"
+#include "Context.h"
#include "jni.h"
#include "HdfsClient_JNI.h"
@@ -239,6 +240,7 @@ HDFS_Scan_RetCode HdfsScan::trafHdfsRead(NAHeap *heap, ExHdfsScanStats *hdfsStat
}
if (j_retArray == NULL)
return HDFS_SCAN_EOR;
+
short retArrayLen = jenv_->GetArrayLength(j_retArray);
ex_assert(retArrayLen == arrayLen, "HdfsScan::trafHdfsRead() InternalError: retArrayLen != arrayLen");
jenv_->GetIntArrayRegion(j_retArray, 0, 4, retArray);
@@ -285,23 +287,11 @@ HdfsClient::~HdfsClient()
deleteHdfsFileInfo();
}
-void HdfsClient::deleteHdfsFileInfo()
-{
- for (int i = 0; i < numFiles_ ; i ++) {
- NADELETEBASIC(hdfsFileInfo_[i].mName, getHeap());
- NADELETEBASIC(hdfsFileInfo_[i].mOwner, getHeap());
- NADELETEBASIC(hdfsFileInfo_[i].mGroup, getHeap());
- }
- NADELETEBASIC(hdfsFileInfo_, getHeap());
- numFiles_ = 0;
- hdfsFileInfo_ = NULL;
-}
-
HdfsClient *HdfsClient::newInstance(NAHeap *heap, HDFS_Client_RetCode &retCode)
{
QRLogger::log(CAT_SQL_HDFS, LL_DEBUG, "HdfsClient::newInstance() called.");
- if (initJNIEnv() != JOI_OK)
+ if (initJNIEnv() != JOI_OK)
return NULL;
retCode = HDFS_CLIENT_OK;
HdfsClient *hdfsClient = new (heap) HdfsClient(heap);
@@ -315,6 +305,18 @@ HdfsClient *HdfsClient::newInstance(NAHeap *heap, HDFS_Client_RetCode &retCode)
return hdfsClient;
}
+void HdfsClient::deleteHdfsFileInfo()
+{
+ for (int i = 0; i < numFiles_ ; i ++) {
+ NADELETEBASIC(hdfsFileInfo_[i].mName, getHeap());
+ NADELETEBASIC(hdfsFileInfo_[i].mOwner, getHeap());
+ NADELETEBASIC(hdfsFileInfo_[i].mGroup, getHeap());
+ }
+ NADELETEBASIC(hdfsFileInfo_, getHeap());
+ numFiles_ = 0;
+ hdfsFileInfo_ = NULL;
+}
+
HDFS_Client_RetCode HdfsClient::init()
{
static char className[]="org/trafodion/sql/HDFSClient";
http://git-wip-us.apache.org/repos/asf/trafodion/blob/ac706607/core/sql/exp/ExpLOBaccess.cpp
----------------------------------------------------------------------
diff --git a/core/sql/exp/ExpLOBaccess.cpp b/core/sql/exp/ExpLOBaccess.cpp
index b5a427b..481e960 100644
--- a/core/sql/exp/ExpLOBaccess.cpp
+++ b/core/sql/exp/ExpLOBaccess.cpp
@@ -63,6 +63,7 @@
#include "ComQueue.h"
#include "QRLogger.h"
#include "NAMemory.h"
+#include "HdfsClient_JNI.h"
#include <seabed/ms.h>
#include <seabed/fserr.h>
#include <curl/curl.h>
http://git-wip-us.apache.org/repos/asf/trafodion/blob/ac706607/core/sql/optimizer/HDFSHook.cpp
----------------------------------------------------------------------
diff --git a/core/sql/optimizer/HDFSHook.cpp b/core/sql/optimizer/HDFSHook.cpp
index dfc80ad..c8793d9 100644
--- a/core/sql/optimizer/HDFSHook.cpp
+++ b/core/sql/optimizer/HDFSHook.cpp
@@ -29,6 +29,7 @@
// for DNS name resolution
#include <netdb.h>
+#include "HdfsClient_JNI.h"
#include "Globals.h"
#include "Context.h"
// Initialize static variables
http://git-wip-us.apache.org/repos/asf/trafodion/blob/ac706607/core/sql/optimizer/NATable.cpp
----------------------------------------------------------------------
diff --git a/core/sql/optimizer/NATable.cpp b/core/sql/optimizer/NATable.cpp
index d657a48..cb0cc39 100644
--- a/core/sql/optimizer/NATable.cpp
+++ b/core/sql/optimizer/NATable.cpp
@@ -87,6 +87,7 @@
#define MAX_NODE_NAME 9
#include "SqlParserGlobals.h"
+#include "HdfsClient_JNI.h"
//#define __ROSETTA
//#include "rosetta_ddl_include.h"
http://git-wip-us.apache.org/repos/asf/trafodion/blob/ac706607/core/sql/sqlcomp/DefaultConstants.h
----------------------------------------------------------------------
diff --git a/core/sql/sqlcomp/DefaultConstants.h b/core/sql/sqlcomp/DefaultConstants.h
index 2671482..fd110de 100644
--- a/core/sql/sqlcomp/DefaultConstants.h
+++ b/core/sql/sqlcomp/DefaultConstants.h
@@ -3310,6 +3310,7 @@ enum DefaultConstants
// Use the earlier implementation of HdfsScan via libhdfs
USE_LIBHDFS_SCAN,
+
// This enum constant must be the LAST one in the list; it's a count,
// not an Attribute (it's not IN DefaultDefaults; it's the SIZE of it)!
__NUM_DEFAULT_ATTRIBUTES
http://git-wip-us.apache.org/repos/asf/trafodion/blob/ac706607/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 fe116d7..52453cc 100644
--- a/core/sql/src/main/java/org/trafodion/sql/HDFSClient.java
+++ b/core/sql/src/main/java/org/trafodion/sql/HDFSClient.java
@@ -21,19 +21,20 @@
package org.trafodion.sql;
+import java.io.IOException;
+import java.io.EOFException;
+import java.io.OutputStream;
+import java.nio.ByteBuffer;
import org.apache.log4j.PropertyConfigurator;
import org.apache.log4j.Logger;
+import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.FileUtil;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.fs.FileStatus;
-import org.apache.hadoop.conf.Configuration;
-import java.nio.ByteBuffer;
-import java.io.IOException;
import java.io.EOFException;
-import java.io.OutputStream;
import java.util.concurrent.Callable;
import java.util.concurrent.Future;
import java.util.concurrent.Executors;
@@ -168,7 +169,7 @@ public class HDFSClient
bytesRead = retObject.intValue();
fsdis_.close();
return bytesRead;
- }
+ }
public int getRangeNo()
{
@@ -392,3 +393,5 @@ public class HDFSClient
short permissions, long accessTime);
}
+
+
[2/4] trafodion git commit: [TRAFODION-2917] Refactor Trafodion
implementation of hdfs scan for text formatted hive tables
Posted by se...@apache.org.
http://git-wip-us.apache.org/repos/asf/trafodion/blob/e303b3a0/core/sql/nskgmake/tdm_sqlexp/Makefile
----------------------------------------------------------------------
diff --git a/core/sql/nskgmake/tdm_sqlexp/Makefile b/core/sql/nskgmake/tdm_sqlexp/Makefile
index 6b973af..7213449 100755
--- a/core/sql/nskgmake/tdm_sqlexp/Makefile
+++ b/core/sql/nskgmake/tdm_sqlexp/Makefile
@@ -77,7 +77,7 @@ CPPSRC += exp_ieee.cpp
CPPSRC += vers_libtdm_sqlexp.cpp
-DEFS := -D_IEEE_FLOAT -DHAVE_INTTYPES_H -DHAVE_NETINET_IN_H -D__STDC_CONSTANT_MACROS -D__STDC_FORMAT_MACROS -D__STDC_LIMIT_MACROS
+DEFS := -D_IEEE_FLOAT -DHAVE_INTTYPES_H -DHAVE_NETINET_IN_H -D__STDC_CONSTANT_MACROS
#
LLVM_OBJ_DIR := $(LLVM)/$(SQ_BUILD_TYPE)/lib
http://git-wip-us.apache.org/repos/asf/trafodion/blob/e303b3a0/core/sql/optimizer/HDFSHook.cpp
----------------------------------------------------------------------
diff --git a/core/sql/optimizer/HDFSHook.cpp b/core/sql/optimizer/HDFSHook.cpp
index c8793d9..4082679 100644
--- a/core/sql/optimizer/HDFSHook.cpp
+++ b/core/sql/optimizer/HDFSHook.cpp
@@ -895,7 +895,13 @@ NABoolean HHDFSTableStats::populate(struct hive_tbl_desc *htd)
// put back fully qualified URI
tableDir = hsd->location_;
- // visit the directory
+ // get the fine-resolution timestamp before visiting
+ // the tree, to avoid losing any updates while this
+ // method is executing
+ computeModificationTSmsec();
+
+ if (diags_.isSuccess())
+ // visit the directory
processDirectory(tableDir, hsd->buckets_,
hsd->isTrulyText(),
hsd->getRecordTerminator());
@@ -1149,6 +1155,32 @@ void HHDFSTableStats::disconnectHDFS()
// is dropped or the thread exits.
}
+void HHDFSTableStats::computeModificationTSmsec()
+{
+ if (modificationTSInMillisec_ <= 0)
+ {
+ HDFS_Client_RetCode rc;
+
+ // get a millisecond-resolution timestamp via JNI
+ rc = HdfsClient::getHiveTableMaxModificationTs(
+ modificationTSInMillisec_,
+ tableDir_.data(),
+ numOfPartCols_);
+ // check for errors and timestamp mismatches
+ if (rc != HDFS_CLIENT_OK || modificationTSInMillisec_ <= 0)
+ {
+ NAString errMsg;
+
+ errMsg.format("Error %d when reading msec timestamp for HDFS URL %s",
+ rc,
+ tableDir_.data());
+ diags_.recordError(errMsg, "HHDFSTableStats::computeModificationTSmsec");
+ modificationTSInMillisec_ = -1;
+ }
+ }
+
+ return;
+}
OsimHHDFSStatsBase* HHDFSTableStats::osimSnapShot(NAMemory * heap)
{
http://git-wip-us.apache.org/repos/asf/trafodion/blob/e303b3a0/core/sql/optimizer/HDFSHook.h
----------------------------------------------------------------------
diff --git a/core/sql/optimizer/HDFSHook.h b/core/sql/optimizer/HDFSHook.h
index f6f64fa..1aec212 100644
--- a/core/sql/optimizer/HDFSHook.h
+++ b/core/sql/optimizer/HDFSHook.h
@@ -319,7 +319,11 @@ public:
validationJTimestamp_(-1),
listPartitionStatsList_(heap),
hiveStatsSize_(0),
- heap_(heap) {}
+ heap_(heap),
+ type_(UNKNOWN_),
+ modificationTSInMillisec_(-1)
+ {}
+
~HHDFSTableStats();
const CollIndex entries() const { return listPartitionStatsList_.entries(); }
@@ -393,6 +397,12 @@ public:
const Lng32 numOfPartCols() const { return numOfPartCols_; }
const Lng32 totalNumPartitions() const { return totalNumPartitions_; }
+ // finer-resolution timestamp for entire table
+ // (can remove this once we use JNI to collect this info
+ // for all HDFS files)
+ Int64 getModificationTSmsec() const { return modificationTSInMillisec_; }
+ void computeModificationTSmsec();
+
private:
enum FileType
{
@@ -443,8 +453,10 @@ private:
HHDFSDiags diags_;
NAMemory *heap_;
-
+
FileType type_;
+
+ Int64 modificationTSInMillisec_;
};
#endif
http://git-wip-us.apache.org/repos/asf/trafodion/blob/e303b3a0/core/sql/optimizer/OptimizerSimulator.cpp
----------------------------------------------------------------------
diff --git a/core/sql/optimizer/OptimizerSimulator.cpp b/core/sql/optimizer/OptimizerSimulator.cpp
index 53af84d..d611098 100644
--- a/core/sql/optimizer/OptimizerSimulator.cpp
+++ b/core/sql/optimizer/OptimizerSimulator.cpp
@@ -55,7 +55,6 @@
#include "HBaseClient_JNI.h"
#include "vproc.h"
-#include "hdfs.h"
#include "CmpSeabaseDDL.h"
#include "ExExeUtilCli.h"
#include "ComUser.h"
http://git-wip-us.apache.org/repos/asf/trafodion/blob/e303b3a0/core/sql/optimizer/RelScan.h
----------------------------------------------------------------------
diff --git a/core/sql/optimizer/RelScan.h b/core/sql/optimizer/RelScan.h
index 58af7cd..cc36f79 100644
--- a/core/sql/optimizer/RelScan.h
+++ b/core/sql/optimizer/RelScan.h
@@ -882,7 +882,8 @@ public:
char* &hdfsHostName,
Int32 &hdfsPort,
NABoolean &doMultiCursor,
- NABoolean &doSplitFileOpt);
+ NABoolean &doSplitFileOpt,
+ NABoolean &isHdfsCompressed);
static short genForOrc(Generator * generator,
const HHDFSTableStats* hTabStats,
const PartitioningFunction * mypart,
http://git-wip-us.apache.org/repos/asf/trafodion/blob/e303b3a0/core/sql/regress/hive/EXPECTED007
----------------------------------------------------------------------
diff --git a/core/sql/regress/hive/EXPECTED007 b/core/sql/regress/hive/EXPECTED007
index 2162198..2b0fd34 100644
--- a/core/sql/regress/hive/EXPECTED007
+++ b/core/sql/regress/hive/EXPECTED007
@@ -956,7 +956,7 @@ create database hivesch0078;
*** WARNING[8597] Statement was automatically retried 1 time(s). Delay before each retry was 0 seconds. See next entry for the error that caused this retry.
-*** WARNING[8436] Mismatch detected between compiletime and runtime hive table definitions. DataModMismatchDetails: compiledModTS = 1508360788, failedModTS = -1, failedLoc = hdfs://localhost:25600/user/hive/warehouse/hivesch007.db/thive1
+*** WARNING[8577] Table, index, or view HIVE.HIVESCH007.THIVE1 was not found.
--- 0 row(s) selected.
>>
http://git-wip-us.apache.org/repos/asf/trafodion/blob/e303b3a0/core/sql/regress/hive/EXPECTED018
----------------------------------------------------------------------
diff --git a/core/sql/regress/hive/EXPECTED018 b/core/sql/regress/hive/EXPECTED018
index d693a56..6f100f1 100644
--- a/core/sql/regress/hive/EXPECTED018
+++ b/core/sql/regress/hive/EXPECTED018
@@ -2376,7 +2376,7 @@ C_FIRST_NAME
Task: UNLOAD Status: Started
Task: EXTRACT Status: Started Time: 2018-02-15 18:12:42.125129
-*** ERROR[8447] An error occurred during hdfs access. Error Detail: Java exception in hdfsCreate(). java.io.IOException: No FileSystem for scheme: null
+*** ERROR[8447] An error occurred during hdfs access. Error Detail: Java exception in HdfsClient::hdfsOpen(). java.io.IOException: No FileSystem for scheme: null
org.apache.hadoop.fs.FileSystem.getFileSystemClass(FileSystem.java:2584)
org.apache.hadoop.fs.FileSystem.createFileSystem(FileSystem.java:2591)
org.apache.hadoop.fs.FileSystem.access$200(FileSystem.java:91)
http://git-wip-us.apache.org/repos/asf/trafodion/blob/e303b3a0/core/sql/regress/hive/EXPECTED040
----------------------------------------------------------------------
diff --git a/core/sql/regress/hive/EXPECTED040 b/core/sql/regress/hive/EXPECTED040
index 8fe3212..bb75200 100644
--- a/core/sql/regress/hive/EXPECTED040
+++ b/core/sql/regress/hive/EXPECTED040
@@ -252,7 +252,7 @@ asciiTuppIndex_ = 4, asciiRowLen_ = 532
moveExprColsTuppIndex_ = 2, moveExprColsRowLength_ = 576
convertSkipListSize_ = 33, convertSkipList_ = 3
outputRowLength_ = 16
-Flag = 0xc
+Flag = 0x20c
Number of ranges to scan: 1
Number of esps to scan: 1
@@ -390,7 +390,7 @@ asciiTuppIndex_ = 4, asciiRowLen_ = 8
moveExprColsTuppIndex_ = 2, moveExprColsRowLength_ = 16
convertSkipListSize_ = 33, convertSkipList_ = 3
outputRowLength_ = 16
-Flag = 0xc
+Flag = 0x20c
Number of ranges to scan: 1
Number of esps to scan: 1
@@ -546,7 +546,7 @@ asciiTuppIndex_ = 4, asciiRowLen_ = 8
moveExprColsTuppIndex_ = 2, moveExprColsRowLength_ = 8
convertSkipListSize_ = 33, convertSkipList_ = 2
outputRowLength_ = 8
-Flag = 0xc
+Flag = 0x20c
Number of ranges to scan: 1
Number of esps to scan: 1
@@ -682,7 +682,7 @@ asciiTuppIndex_ = 4, asciiRowLen_ = 8
moveExprColsTuppIndex_ = 2, moveExprColsRowLength_ = 8
convertSkipListSize_ = 33, convertSkipList_ = 2
outputRowLength_ = 8
-Flag = 0xc
+Flag = 0x20c
Number of ranges to scan: 1
Number of esps to scan: 1
http://git-wip-us.apache.org/repos/asf/trafodion/blob/e303b3a0/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 52453cc..ff78d3d 100644
--- a/core/sql/src/main/java/org/trafodion/sql/HDFSClient.java
+++ b/core/sql/src/main/java/org/trafodion/sql/HDFSClient.java
@@ -22,17 +22,21 @@
package org.trafodion.sql;
import java.io.IOException;
+import java.io.FileNotFoundException;
import java.io.EOFException;
import java.io.OutputStream;
import java.nio.ByteBuffer;
+import java.util.ArrayList;
+import java.util.List;
import org.apache.log4j.PropertyConfigurator;
import org.apache.log4j.Logger;
import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.Path;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.FileUtil;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FSDataOutputStream;
-import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.fs.PathFilter;
import org.apache.hadoop.fs.FileStatus;
import java.io.EOFException;
import java.util.concurrent.Callable;
@@ -41,6 +45,7 @@ import java.util.concurrent.Executors;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
import org.apache.hadoop.io.compress.CodecPool;
import org.apache.hadoop.io.compress.CompressionCodec;
@@ -49,6 +54,17 @@ import org.apache.hadoop.io.compress.GzipCodec;
import org.apache.hadoop.io.SequenceFile.CompressionType;
import org.apache.hadoop.util.ReflectionUtils;
+//
+// To read a range in a Hdfs file, use the constructor
+// public HDFSClient(int bufNo, int rangeNo, String filename, ByteBuffer buffer, long position, int length) throws IOException
+//
+// For instance methods like hdfsListDirectory use the constructor
+// public HDFSClient()
+//
+// For all static methods use
+// HdfsClient::<static_method_name>
+//
+
public class HDFSClient
{
static Logger logger_ = Logger.getLogger(HDFSClient.class.getName());
@@ -71,6 +87,9 @@ public class HDFSClient
private int bytesRead_;
private Future future_ = null;
private int isEOF_ = 0;
+ private int totalBytesWritten_ = 0;
+ private Path filepath_ = null;
+ private boolean compression_;
static {
String confFile = System.getProperty("trafodion.log4j.configFile");
System.setProperty("trafodion.root", System.getenv("TRAF_HOME"));
@@ -89,6 +108,13 @@ public class HDFSClient
System.loadLibrary("executor");
}
+ // The object instance that runs in the threadpool to read
+ // the requested chunk in the range
+
+ // FSDataInputStream.read method may not read the requested length in one shot
+ // Loop to read the requested length or EOF is reached
+ // Requested length can never be larger than the buffer size
+
class HDFSRead implements Callable
{
HDFSRead()
@@ -136,15 +162,19 @@ public class HDFSClient
// This constructor enables the hdfs data to be read in another thread while the previously
// read buffer is being processed by the SQL engine
+ // Opens the file and hands over the needed info to HdfsRead instance to read
+ // The passed in length can never be more than the size of the buffer
+ // If the range has a length more than the buffer length, the range is chunked
+ // in HdfsScan
public HDFSClient(int bufNo, int rangeNo, String filename, ByteBuffer buffer, long position, int length) throws IOException
{
bufNo_ = bufNo;
rangeNo_ = rangeNo;
filename_ = filename;
- Path filepath = new Path(filename_);
- fs_ = FileSystem.get(filepath.toUri(),config_);
- fsdis_ = fs_.open(filepath);
- blockSize_ = (int)fs_.getDefaultBlockSize(filepath);
+ filepath_ = new Path(filename_);
+ fs_ = FileSystem.get(filepath_.toUri(),config_);
+ fsdis_ = fs_.open(filepath_);
+ blockSize_ = (int)fs_.getDefaultBlockSize(filepath_);
buf_ = buffer;
bufOffset_ = 0;
pos_ = position;
@@ -161,6 +191,10 @@ public class HDFSClient
}
}
+ // This method waits for the read to complete. Read can complete due to one of the following
+ // a) buffer is full
+ // b) EOF is reached
+ // c) An exception is encountered while reading the file
public int trafHdfsReadBuffer() throws IOException, InterruptedException, ExecutionException
{
Integer retObject = 0;
@@ -168,6 +202,7 @@ public class HDFSClient
retObject = (Integer)future_.get();
bytesRead = retObject.intValue();
fsdis_.close();
+ fsdis_ = null;
return bytesRead;
}
@@ -181,78 +216,127 @@ public class HDFSClient
return isEOF_;
}
- boolean hdfsCreate(String fname , boolean compress) throws IOException
+ boolean hdfsCreate(String fname , boolean overwrite, boolean compress) throws IOException
{
- if (logger_.isDebugEnabled())
+ if (logger_.isDebugEnabled())
logger_.debug("HDFSClient.hdfsCreate() - started" );
- Path filePath = null;
if (!compress || (compress && fname.endsWith(".gz")))
- filePath = new Path(fname);
+ filepath_ = new Path(fname);
else
- filePath = new Path(fname + ".gz");
+ filepath_ = new Path(fname + ".gz");
- FileSystem fs = FileSystem.get(filePath.toUri(),config_);
- FSDataOutputStream fsOut = fs.create(filePath, true);
-
- if (compress) {
- GzipCodec gzipCodec = (GzipCodec) ReflectionUtils.newInstance( GzipCodec.class, config_);
- Compressor gzipCompressor = CodecPool.getCompressor(gzipCodec);
- outStream_= gzipCodec.createOutputStream(fsOut, gzipCompressor);
+ fs_ = FileSystem.get(filepath_.toUri(),config_);
+ compression_ = compress;
+ fsdis_ = null;
+ FSDataOutputStream fsOut;
+ if (overwrite)
+ fsOut = fs_.create(filepath_);
+ else
+ if (fs_.exists(filepath_))
+ fsOut = fs_.append(filepath_);
+ else
+ fsOut = fs_.create(filepath_);
+
+ if (compression_) {
+ GzipCodec gzipCodec = (GzipCodec) ReflectionUtils.newInstance( GzipCodec.class, config_);
+ Compressor gzipCompressor = CodecPool.getCompressor(gzipCodec);
+ outStream_= gzipCodec.createOutputStream(fsOut, gzipCompressor);
}
else
- outStream_ = fsOut;
- if (logger_.isDebugEnabled())
- logger_.debug("HDFSClient.hdfsCreate() - compressed output stream created" );
+ outStream_ = fsOut;
return true;
- }
+ }
- boolean hdfsOpen(String fname , boolean compress) throws IOException
- {
+ boolean hdfsOpen(String fname , boolean compress) throws IOException
+ {
if (logger_.isDebugEnabled())
logger_.debug("HDFSClient.hdfsOpen() - started" );
- Path filePath = null;
if (!compress || (compress && fname.endsWith(".gz")))
- filePath = new Path(fname);
+ filepath_ = new Path(fname);
else
- filePath = new Path(fname + ".gz");
-
- FileSystem fs = FileSystem.get(filePath.toUri(),config_);
+ filepath_ = new Path(fname + ".gz");
+ fs_ = FileSystem.get(filepath_.toUri(),config_);
+ compression_ = compress;
+ outStream_ = null;
+ fsdis_ = null;
+ return true;
+ }
+
+ int hdfsWrite(byte[] buff) throws IOException
+ {
+ if (logger_.isDebugEnabled())
+ logger_.debug("HDFSClient.hdfsWrite() - started" );
+
FSDataOutputStream fsOut;
- if (fs.exists(filePath))
- fsOut = fs.append(filePath);
- else
- fsOut = fs.create(filePath);
+ if (outStream_ == null) {
+ if (fs_.exists(filepath_))
+ fsOut = fs_.append(filepath_);
+ else
+ fsOut = fs_.create(filepath_);
- if (compress) {
- GzipCodec gzipCodec = (GzipCodec) ReflectionUtils.newInstance( GzipCodec.class, config_);
- Compressor gzipCompressor = CodecPool.getCompressor(gzipCodec);
- outStream_= gzipCodec.createOutputStream(fsOut, gzipCompressor);
+ if (compression_) {
+ GzipCodec gzipCodec = (GzipCodec) ReflectionUtils.newInstance( GzipCodec.class, config_);
+ Compressor gzipCompressor = CodecPool.getCompressor(gzipCodec);
+ outStream_= gzipCodec.createOutputStream(fsOut, gzipCompressor);
+ }
+ else
+ outStream_ = fsOut;
+ if (logger_.isDebugEnabled())
+ logger_.debug("HDFSClient.hdfsWrite() - output stream created" );
}
+ outStream_.write(buff);
+ if (outStream_ instanceof FSDataOutputStream)
+ totalBytesWritten_ = ((FSDataOutputStream)outStream_).size();
else
- outStream_ = fsOut;
+ totalBytesWritten_ += buff.length;
if (logger_.isDebugEnabled())
- logger_.debug("HDFSClient.hdfsCreate() - compressed output stream created" );
- return true;
+ logger_.debug("HDFSClient.hdfsWrite() - bytes written " + totalBytesWritten_ );
+ return totalBytesWritten_;
}
-
- boolean hdfsWrite(byte[] buff, long len) throws IOException
- {
+ int hdfsRead(ByteBuffer buffer) throws IOException
+ {
if (logger_.isDebugEnabled())
- logger_.debug("HDFSClient.hdfsWrite() - started" );
- outStream_.write(buff);
- outStream_.flush();
- if (logger_.isDebugEnabled()) logger_.debug("HDFSClient.hdfsWrite() - bytes written and flushed:" + len );
- return true;
+ logger_.debug("HDFSClient.hdfsRead() - started" );
+ if (fsdis_ == null) {
+ fsdis_ = fs_.open(filepath_);
+ pos_ = 0;
+ }
+ int lenRemain;
+ int bytesRead;
+ int totalBytesRead = 0;
+ int bufLen;
+ int bufOffset = 0;
+ if (buffer.hasArray())
+ bufLen = buffer.array().length;
+ else
+ bufLen = buffer.capacity();
+ lenRemain = bufLen;
+ do
+ {
+ if (buffer.hasArray())
+ bytesRead = fsdis_.read(pos_, buffer.array(), bufOffset, lenRemain);
+ else
+ bytesRead = fsdis_.read(buffer);
+ if (bytesRead == -1 || bytesRead == 0)
+ break;
+ totalBytesRead += bytesRead;
+ pos_ += bytesRead;
+ lenRemain -= bytesRead;
+ } while (lenRemain > 0);
+ return totalBytesRead;
}
boolean hdfsClose() throws IOException
{
if (logger_.isDebugEnabled()) logger_.debug("HDFSClient.hdfsClose() - started" );
if (outStream_ != null) {
+ outStream_.flush();
outStream_.close();
outStream_ = null;
}
+ if (fsdis_ != null)
+ fsdis_.close();
return true;
}
@@ -380,6 +464,25 @@ public class HDFSClient
else
return 0;
}
+
+
+ public void stop() throws IOException
+ {
+ if (future_ != null) {
+ try {
+ future_.get(200, TimeUnit.MILLISECONDS);
+ } catch(TimeoutException e) {
+ logger_.error("Asynchronous Thread of HdfsScan is Cancelled (timeout), ", e);
+ future_.cancel(true);
+ } catch(InterruptedException e) {
+ logger_.error("Asynchronous Thread of HdfsScan is Cancelled (interrupt), ", e);
+ future_.cancel(true); // Interrupt the thread
+ } catch (ExecutionException ee)
+ {
+ }
+ future_ = null;
+ }
+ }
public static void shutdown() throws InterruptedException
{
@@ -387,6 +490,103 @@ public class HDFSClient
executorService_.shutdown();
}
+ private static FileSystem getFileSystem() throws IOException
+ {
+ return defaultFs_;
+ }
+
+ // if levelDeep = 0, return the max modification timestamp of the passed-in HDFS URIs
+ // (a tab-separated list of 0 or more paths)
+ // if levelDeep > 0, also check all directories "levelDeep" levels below. Exclude
+ // directories that start with a dot (hidden directories)
+ public static long getHiveTableMaxModificationTs( String stableDirPaths, int levelDeep) throws FileNotFoundException, IOException
+ {
+ long result = 0;
+ if (logger_.isDebugEnabled())
+ logger_.debug("HDFSClient:getHiveTableMaxModificationTs enter");
+
+ String[] tableDirPaths = stableDirPaths.split("\t");
+ // account for root dir
+ for (int i=0; i<tableDirPaths.length; i++) {
+ FileStatus r = getFileSystem().getFileStatus(new Path(tableDirPaths[i]));// super fast API, return in .2ms
+ if (r != null && r.getModificationTime() > result)
+ result = r.getModificationTime();
+ }
+
+ if (levelDeep>0)
+ {
+ Path[] paths = new Path[tableDirPaths.length];
+ for (int i=0; i<tableDirPaths.length; i++)
+ paths[i] = new Path(tableDirPaths[i]);
+ long l = getHiveTableMaxModificationTs2(paths,levelDeep);
+ if (l > result)
+ result = l;
+ }
+ if (logger_.isDebugEnabled())
+ logger_.debug("HDFSClient:getHiveTableMaxModificationTs "+stableDirPaths+" levelDeep"+levelDeep+":"+result);
+ return result;
+ }
+
+ private static long getHiveTableMaxModificationTs2(Path[] paths, int levelDeep)throws FileNotFoundException, IOException
+ {
+ long result = 0;
+ PathFilter filter = new PathFilter(){
+ public boolean accept(Path file){
+ return !file.getName().startsWith(".");//filter out hidden files and directories
+ }
+ };
+ FileStatus[] fileStatuss=null;
+ if (levelDeep == 1){ // stop condition on recursive function
+ //check parent level (important for deletes):
+ for (Path path : paths){
+ FileStatus r = getFileSystem().getFileStatus(path);// super fast API, return in .2ms
+ if (r != null && r.getModificationTime()>result)
+ result = r.getModificationTime();
+ }
+ if (paths.length==1)
+ fileStatuss = getFileSystem().listStatus(paths[0],filter);// minor optimization. avoid using list based API when not needed
+ else
+ fileStatuss = getFileSystem().listStatus(paths,filter);
+ for(int i=0;i<fileStatuss.length;i++)
+ if (fileStatuss[i].isDirectory() && fileStatuss[i].getModificationTime()>result)
+ result = fileStatuss[i].getModificationTime();
+ }else{//here levelDeep >1
+ List<Path> pathList = new ArrayList<Path>();
+ if (paths.length==1)
+ fileStatuss = getFileSystem().listStatus(paths[0],filter);// minor optimization. avoid using list based API when not needed
+ else
+ fileStatuss = getFileSystem().listStatus(paths,filter);
+ for(int i=0;i<fileStatuss.length;i++)
+ if (fileStatuss[i].isDirectory())
+ {
+ pathList.add(fileStatuss[i].getPath());
+ if (fileStatuss[i].getModificationTime()>result)
+ result = fileStatuss[i].getModificationTime();// make sure level n-1 is accounted for for delete partition case
+ }
+ long l = getHiveTableMaxModificationTs2(pathList.toArray(new Path[pathList.size()]),levelDeep-1);
+ if (l>result) result = l;
+
+ }
+ return result;
+ }
+
+ public static String getFsDefaultName()
+ {
+ String uri = config_.get("fs.defaultFS");
+ return uri;
+ }
+
+
+ public static boolean hdfsCreateDirectory(String pathStr) throws IOException
+ {
+ if (logger_.isDebugEnabled())
+ logger_.debug("HDFSClient.hdfsCreateDirectory()" + pathStr);
+ Path dirPath = new Path(pathStr );
+ FileSystem fs = FileSystem.get(dirPath.toUri(), config_);
+ fs.mkdirs(dirPath);
+ return true;
+ }
+
private native int sendFileStatus(long jniObj, int numFiles, int fileNo, boolean isDir,
String filename, long modTime, long len,
short numReplicas, long blockSize, String owner, String group,
http://git-wip-us.apache.org/repos/asf/trafodion/blob/e303b3a0/core/sql/src/main/java/org/trafodion/sql/HdfsScan.java
----------------------------------------------------------------------
diff --git a/core/sql/src/main/java/org/trafodion/sql/HdfsScan.java b/core/sql/src/main/java/org/trafodion/sql/HdfsScan.java
index e216555..99f021d 100644
--- a/core/sql/src/main/java/org/trafodion/sql/HdfsScan.java
+++ b/core/sql/src/main/java/org/trafodion/sql/HdfsScan.java
@@ -21,17 +21,6 @@
package org.trafodion.sql;
-// This class implements an efficient mechanism to read hdfs files
-// Trafodion ExHdfsScan operator provides a range of scans to be performed.
-// The range consists of a hdfs filename, offset and length to be read
-// This class takes in two ByteBuffers. These ByteBuffer can be either direct buffers
-// backed up native buffers or indirect buffer backed by java arrays.
-// All the ranges are read alternating between the two buffers using ExecutorService
-// using CachedThreadPool mechanism.
-// For a given HdfsScan instance, only one thread(IO thread) is scheduled to read
-// the next full or partial buffer while the main thread processes the previously
-// read information from the other buffer
-
import org.apache.log4j.PropertyConfigurator;
import org.apache.log4j.Logger;
import org.apache.hadoop.fs.FileSystem;
@@ -54,6 +43,24 @@ import org.apache.hadoop.hive.metastore.api.StorageDescriptor;
import org.apache.hadoop.fs.FileStatus;
import java.net.URI;
+// This class implements an efficient mechanism to read hdfs files
+// Trafodion ExHdfsScan operator provides a range of scans to be performed.
+// The range consists of a hdfs filename, offset and length to be read
+// This class takes in two ByteBuffers. These ByteBuffer can be either direct buffers
+// backed up native buffers or indirect buffer backed by java arrays.
+// All the ranges are read alternating between the two buffers using ExecutorService
+// using CachedThreadPool mechanism.
+
+// For a given HdfsScan instance, only one thread(IO thread) is scheduled to read
+// the next full or partial buffer while the main thread processes the previously
+// read information from the other buffer
+// HdfsScan picks up a range and schedules a read into a next available buffer.
+// If the range is more than the buffer size, then the range is read into multiple
+// chunks and schedules one chunk at a time alternatiing the buffers
+// Once the range is completed, the next range in the HdfsScanRange is picked up
+// for scheduling, till all the ranges assigned to the HdfsScan instance are read.
+
+
public class HdfsScan
{
static Logger logger_ = Logger.getLogger(HdfsScan.class.getName());
@@ -61,10 +68,13 @@ public class HdfsScan
private int bufLen_[];
private HDFSClient hdfsClient_[];
private int currRange_;
- private long currPos_;
- private long lenRemain_;
+ private long currRangePos_;
+ private long currRangeLenRemain_;
private int lastBufCompleted_ = -1;
private boolean scanCompleted_;
+
+ // Structure to hold the Scan ranges for this HdfsScan instance
+ //
class HdfsScanRange
{
@@ -95,6 +105,7 @@ public class HdfsScan
public void setScanRanges(ByteBuffer buf1, ByteBuffer buf2, String filename[], long pos[], long len[], int rangeNum[]) throws IOException
{
+ // Two buffers to hold the data read
buf_ = new ByteBuffer[2];
bufLen_ = new int[2];
@@ -114,39 +125,47 @@ public class HdfsScan
}
if (hdfsScanRanges_.length > 0) {
currRange_ = 0;
- currPos_ = hdfsScanRanges_[currRange_].pos_;
- lenRemain_ = hdfsScanRanges_[currRange_].len_;
- hdfsScanRange(0, 0);
+ currRangePos_ = hdfsScanRanges_[currRange_].pos_;
+ currRangeLenRemain_ = hdfsScanRanges_[currRange_].len_;
+ scheduleHdfsScanRange(0, 0);
}
scanCompleted_ = false;
}
- public void hdfsScanRange(int bufNo, int bytesCompleted) throws IOException
+ public void scheduleHdfsScanRange(int bufNo, int bytesCompleted) throws IOException
{
- lenRemain_ -= bytesCompleted;
- currPos_ += bytesCompleted;
+ currRangeLenRemain_ -= bytesCompleted;
+ currRangePos_ += bytesCompleted;
int readLength;
- if (lenRemain_ <= 0) {
+ if (currRangeLenRemain_ <= 0) {
if (currRange_ == (hdfsScanRanges_.length-1)) {
scanCompleted_ = true;
return;
}
else {
currRange_++;
- currPos_ = hdfsScanRanges_[currRange_].pos_;
- lenRemain_ = hdfsScanRanges_[currRange_].len_;
+ currRangePos_ = hdfsScanRanges_[currRange_].pos_;
+ currRangeLenRemain_ = hdfsScanRanges_[currRange_].len_;
}
}
- if (lenRemain_ > bufLen_[bufNo])
+ if (currRangeLenRemain_ > bufLen_[bufNo])
readLength = bufLen_[bufNo];
else
- readLength = (int)lenRemain_;
+ readLength = (int)currRangeLenRemain_;
if (! scanCompleted_) {
if (logger_.isDebugEnabled())
- logger_.debug(" CurrentRange " + hdfsScanRanges_[currRange_].tdbRangeNum_ + " LenRemain " + lenRemain_ + " BufNo " + bufNo);
- hdfsClient_[bufNo] = new HDFSClient(bufNo, hdfsScanRanges_[currRange_].tdbRangeNum_, hdfsScanRanges_[currRange_].filename_, buf_[bufNo], currPos_, readLength);
+ logger_.debug(" CurrentRange " + hdfsScanRanges_[currRange_].tdbRangeNum_ + " LenRemain " + currRangeLenRemain_ + " BufNo " + bufNo);
+ hdfsClient_[bufNo] = new HDFSClient(bufNo, hdfsScanRanges_[currRange_].tdbRangeNum_, hdfsScanRanges_[currRange_].filename_, buf_[bufNo], currRangePos_, readLength);
}
}
+
+/*
+ Method to wait for completion of the scheduled read of a chunk in a range
+ Returns 4 items, bytes read, buf no, range no, is EOF
+ If there are more chunks to be read in the range, schedules a read into the other buffer
+ If EOF is reached or the full range is read, the next range is picked up for
+ scheduling
+*/
public int[] trafHdfsRead() throws IOException, InterruptedException, ExecutionException
{
@@ -164,12 +183,14 @@ public class HdfsScan
switch (lastBufCompleted_) {
case -1:
case 1:
+ // Wait for the read to complete in buffer 0
bytesRead = hdfsClient_[0].trafHdfsReadBuffer();
bufNo = 0;
rangeNo = hdfsClient_[0].getRangeNo();
isEOF = hdfsClient_[0].isEOF();
break;
case 0:
+ // Wait for the read to complete in buffer 1
bytesRead = hdfsClient_[1].trafHdfsReadBuffer();
bufNo = 1;
rangeNo = hdfsClient_[1].getRangeNo();
@@ -194,18 +215,20 @@ public class HdfsScan
return retArray;
} else {
currRange_++;
- currPos_ = hdfsScanRanges_[currRange_].pos_;
- lenRemain_ = hdfsScanRanges_[currRange_].len_;
+ currRangePos_ = hdfsScanRanges_[currRange_].pos_;
+ currRangeLenRemain_ = hdfsScanRanges_[currRange_].len_;
bytesRead = 0;
}
}
switch (lastBufCompleted_)
{
case 0:
- hdfsScanRange(1, bytesRead);
+ // schedule the next chunk or next range to be read in buffer 1
+ scheduleHdfsScanRange(1, bytesRead);
break;
case 1:
- hdfsScanRange(0, bytesRead);
+ // schedule the next chunk or next range to be read in buffer 0
+ scheduleHdfsScanRange(0, bytesRead);
break;
default:
break;
@@ -213,10 +236,22 @@ public class HdfsScan
return retArray;
}
+ public void stop() throws IOException
+ {
+ if (hdfsClient_[0] != null)
+ hdfsClient_[0].stop();
+ if (hdfsClient_[1] != null)
+ hdfsClient_[1].stop();
+ hdfsClient_[0] = null;
+ hdfsClient_[1] = null;
+ return;
+ }
+
public static void shutdown() throws InterruptedException
{
HDFSClient.shutdown();
}
+
public static void main(String[] args) throws Exception
{
[3/4] trafodion git commit: [TRAFODION-2917] Refactor Trafodion
implementation of hdfs scan for text formatted hive tables
Posted by se...@apache.org.
[TRAFODION-2917] Refactor Trafodion implementation of hdfs scan for text formatted hive tables
Removed the dependency on libhdfs in Trafodion code in some more portions of the code.
Also, introduced HDFS Scan to use the java layer to read HDFS data providing the following features:
a) Prefetch the data using double buffering concept
b) Avoid unnecessary data copy
c) Ensure that the data read is initiated to be read into the other buffer in java layer, while it is being processed in one buffer in the native side, without any hiccups.
This HDFS Scan is still being tested and hence it is turned off by default.
Also changed the code to obtain millisecond-resolution modification timestamp for HDFS files
Project: http://git-wip-us.apache.org/repos/asf/trafodion/repo
Commit: http://git-wip-us.apache.org/repos/asf/trafodion/commit/e303b3a0
Tree: http://git-wip-us.apache.org/repos/asf/trafodion/tree/e303b3a0
Diff: http://git-wip-us.apache.org/repos/asf/trafodion/diff/e303b3a0
Branch: refs/heads/master
Commit: e303b3a083154779bcde8a84e1e2abff12d365e8
Parents: ac70660
Author: selvaganesang <se...@esgyn.com>
Authored: Wed Feb 28 22:25:43 2018 +0000
Committer: selvaganesang <se...@esgyn.com>
Committed: Thu Mar 1 22:30:05 2018 +0000
----------------------------------------------------------------------
core/sql/cli/Context.cpp | 2 +-
core/sql/cli/Context.h | 6 +
core/sql/comexe/ComTdbBlockingHdfsScan.h | 1 -
core/sql/comexe/ComTdbHdfsScan.h | 8 +-
core/sql/common/ComUser.cpp | 2 +-
core/sql/common/Platform.h | 2 +
core/sql/executor/ExExeUtil.h | 1 -
core/sql/executor/ExExeUtilGet.cpp | 5 +-
core/sql/executor/ExExeUtilGetStats.cpp | 1 -
core/sql/executor/ExExeUtilLoad.cpp | 25 +-
core/sql/executor/ExFastTransport.cpp | 47 +-
core/sql/executor/ExHbaseAccess.cpp | 11 +-
core/sql/executor/ExHbaseIUD.cpp | 2 +-
core/sql/executor/ExHdfsScan.cpp | 150 +++----
core/sql/executor/ExHdfsScan.h | 1 +
core/sql/executor/ExSMCommon.cpp | 1 +
core/sql/executor/ExSMCommon.h | 3 +-
core/sql/executor/ExSMGlobals.cpp | 1 +
core/sql/executor/ExSMQueue.cpp | 2 +-
core/sql/executor/ExSMReader.cpp | 1 +
core/sql/executor/ExSMShortMessage.cpp | 2 +-
core/sql/executor/ExSMTaskList.cpp | 2 +-
core/sql/executor/HdfsClient_JNI.cpp | 436 +++++++++++++++----
core/sql/executor/HdfsClient_JNI.h | 74 +++-
core/sql/executor/JavaObjectInterface.cpp | 3 +-
core/sql/executor/JavaObjectInterface.h | 12 +-
core/sql/executor/OrcFileReader.h | 2 +-
core/sql/executor/SequenceFileReader.h | 4 +-
core/sql/exp/ExpLOBaccess.cpp | 143 ++----
core/sql/exp/ExpLOBaccess.h | 8 -
core/sql/generator/GenFastTransport.cpp | 4 +-
core/sql/generator/GenRelScan.cpp | 17 +-
core/sql/nskgmake/tdm_sqlexp/Makefile | 2 +-
core/sql/optimizer/HDFSHook.cpp | 34 +-
core/sql/optimizer/HDFSHook.h | 16 +-
core/sql/optimizer/OptimizerSimulator.cpp | 1 -
core/sql/optimizer/RelScan.h | 3 +-
core/sql/regress/hive/EXPECTED007 | 2 +-
core/sql/regress/hive/EXPECTED018 | 2 +-
core/sql/regress/hive/EXPECTED040 | 8 +-
.../main/java/org/trafodion/sql/HDFSClient.java | 294 +++++++++++--
.../main/java/org/trafodion/sql/HdfsScan.java | 95 ++--
42 files changed, 971 insertions(+), 465 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/trafodion/blob/e303b3a0/core/sql/cli/Context.cpp
----------------------------------------------------------------------
diff --git a/core/sql/cli/Context.cpp b/core/sql/cli/Context.cpp
index 38622ae..cdf5909 100644
--- a/core/sql/cli/Context.cpp
+++ b/core/sql/cli/Context.cpp
@@ -60,7 +60,6 @@
#include "ComUser.h"
#include "CmpSeabaseDDLauth.h"
-#include "hdfs.h"
#include "StmtCompilationMode.h"
#include "ExCextdecs.h"
@@ -165,6 +164,7 @@ ContextCli::ContextCli(CliGlobals *cliGlobals)
jniErrorStr_(&exHeap_),
hbaseClientJNI_(NULL),
hiveClientJNI_(NULL),
+ hdfsClientJNI_(NULL),
arkcmpArray_(&exHeap_),
cmpContextInfo_(&exHeap_),
cmpContextInUse_(&exHeap_),
http://git-wip-us.apache.org/repos/asf/trafodion/blob/e303b3a0/core/sql/cli/Context.h
----------------------------------------------------------------------
diff --git a/core/sql/cli/Context.h b/core/sql/cli/Context.h
index bcfd06b..f2da6c8 100644
--- a/core/sql/cli/Context.h
+++ b/core/sql/cli/Context.h
@@ -62,6 +62,7 @@
#include "ExpSeqGen.h"
#include "ssmpipc.h"
#include "hdfs.h"
+#include "HdfsClient_JNI.h"
class CliGlobals;
class HashQueue;
@@ -200,6 +201,11 @@ public:
HiveClient_JNI *getHiveClient() { return hiveClientJNI_; }
void setHiveClient(HiveClient_JNI *hiveClientJNI)
{ hiveClientJNI_ = hiveClientJNI; }
+
+ HdfsClient *getHDFSClient() { return hdfsClientJNI_; }
+ void setHDFSClient(HdfsClient *hdfsClientJNI)
+ { hdfsClientJNI_ = hdfsClientJNI; }
+
//expose cmpContextInfo_ to get HQC info of different contexts
const NAArray<CmpContextInfo *> & getCmpContextInfo() const { return cmpContextInfo_; }
http://git-wip-us.apache.org/repos/asf/trafodion/blob/e303b3a0/core/sql/comexe/ComTdbBlockingHdfsScan.h
----------------------------------------------------------------------
diff --git a/core/sql/comexe/ComTdbBlockingHdfsScan.h b/core/sql/comexe/ComTdbBlockingHdfsScan.h
index 88a3b40..fe0b980 100755
--- a/core/sql/comexe/ComTdbBlockingHdfsScan.h
+++ b/core/sql/comexe/ComTdbBlockingHdfsScan.h
@@ -24,7 +24,6 @@
#define COM_HDFS_SCAN_H
#include "ComTdb.h"
-#include "hdfs.h" // tPort
//
// Task Definition Block
http://git-wip-us.apache.org/repos/asf/trafodion/blob/e303b3a0/core/sql/comexe/ComTdbHdfsScan.h
----------------------------------------------------------------------
diff --git a/core/sql/comexe/ComTdbHdfsScan.h b/core/sql/comexe/ComTdbHdfsScan.h
index 46d7f2f..ea995fb 100755
--- a/core/sql/comexe/ComTdbHdfsScan.h
+++ b/core/sql/comexe/ComTdbHdfsScan.h
@@ -55,7 +55,8 @@ class ComTdbHdfsScan : public ComTdb
LOG_ERROR_ROWS = 0x0040,
ASSIGN_RANGES_AT_RUNTIME = 0x0080,
TREAT_EMPTY_AS_NULL = 0x0100,
- USE_LIBHDFS_SCAN = 0x0200
+ USE_LIBHDFS_SCAN = 0x0200,
+ COMPRESSED_FILE = 0x0400
};
// Expression to filter rows.
@@ -290,6 +291,11 @@ public:
NABoolean getUseLibhdfsScan() const
{ return (flags_ & USE_LIBHDFS_SCAN) != 0; }
+ void setCompressedFile(NABoolean v)
+ {(v ? flags_ |= COMPRESSED_FILE : flags_ &= ~COMPRESSED_FILE); }
+ NABoolean isCompressedFile() const
+ { return (flags_ & COMPRESSED_FILE) != 0; }
+
UInt32 getMaxErrorRows() const { return maxErrorRows_;}
void setMaxErrorRows(UInt32 v ) { maxErrorRows_= v; }
http://git-wip-us.apache.org/repos/asf/trafodion/blob/e303b3a0/core/sql/common/ComUser.cpp
----------------------------------------------------------------------
diff --git a/core/sql/common/ComUser.cpp b/core/sql/common/ComUser.cpp
index 796d94b..2b6831f 100644
--- a/core/sql/common/ComUser.cpp
+++ b/core/sql/common/ComUser.cpp
@@ -72,7 +72,7 @@ Int32 ComUser::getCurrentUser(void)
NULL, 0, NULL);
assert(rc >= 0);
- assert(dbUserID >= SUPER_USER)
+ assert(dbUserID >= SUPER_USER);
return dbUserID;
http://git-wip-us.apache.org/repos/asf/trafodion/blob/e303b3a0/core/sql/common/Platform.h
----------------------------------------------------------------------
diff --git a/core/sql/common/Platform.h b/core/sql/common/Platform.h
index fe7598a..153f9b9 100644
--- a/core/sql/common/Platform.h
+++ b/core/sql/common/Platform.h
@@ -35,6 +35,8 @@
*
*****************************************************************************
*/
+#define __STDC_LIMIT_MACROS
+#define __STDC_FORMAT_MACROS
/*
// On Linux, either NA_BIG_ENDIAN or NA_LITTLE_ENDIAN may have already
http://git-wip-us.apache.org/repos/asf/trafodion/blob/e303b3a0/core/sql/executor/ExExeUtil.h
----------------------------------------------------------------------
diff --git a/core/sql/executor/ExExeUtil.h b/core/sql/executor/ExExeUtil.h
index e119ed1..85173c8 100644
--- a/core/sql/executor/ExExeUtil.h
+++ b/core/sql/executor/ExExeUtil.h
@@ -3893,7 +3893,6 @@ class ExExeUtilHBaseBulkUnLoadTcb : public ExExeUtilTcb
NABoolean emptyTarget_;
NABoolean oneFile_;
ExpHbaseInterface * ehi_;
- HdfsClient *hdfsClient_;
};
class ExExeUtilHbaseUnLoadPrivateState : public ex_tcb_private_state
http://git-wip-us.apache.org/repos/asf/trafodion/blob/e303b3a0/core/sql/executor/ExExeUtilGet.cpp
----------------------------------------------------------------------
diff --git a/core/sql/executor/ExExeUtilGet.cpp b/core/sql/executor/ExExeUtilGet.cpp
index cd20fae..ff40c40 100644
--- a/core/sql/executor/ExExeUtilGet.cpp
+++ b/core/sql/executor/ExExeUtilGet.cpp
@@ -69,7 +69,6 @@
#include "ExpHbaseInterface.h"
#include "sql_buffer_size.h"
-#include "hdfs.h"
#include "NAType.h"
@@ -6873,8 +6872,8 @@ ExExeUtilClusterStatsTcb::ExExeUtilClusterStatsTcb(
stats_ = (ComTdbClusterStatsVirtTableColumnStruct*)statsBuf_;
ehi_ = ExpHbaseInterface::newInstance(glob->getDefaultHeap(),
- (char*)"", //exe_util_tdb.server(),
- (char*)""); //exe_util_tdb.zkPort());
+ (char*)"",
+ (char*)"");
regionInfoList_ = NULL;
// get hbase rootdir location. Max linux pathlength is 1024.
http://git-wip-us.apache.org/repos/asf/trafodion/blob/e303b3a0/core/sql/executor/ExExeUtilGetStats.cpp
----------------------------------------------------------------------
diff --git a/core/sql/executor/ExExeUtilGetStats.cpp b/core/sql/executor/ExExeUtilGetStats.cpp
index 72bdd2b..9e99b5c 100644
--- a/core/sql/executor/ExExeUtilGetStats.cpp
+++ b/core/sql/executor/ExExeUtilGetStats.cpp
@@ -44,7 +44,6 @@
#include "ComTdb.h"
#include "ex_tcb.h"
#include "ComSqlId.h"
-
#include "ExExeUtil.h"
#include "ex_exe_stmt_globals.h"
#include "exp_expr.h"
http://git-wip-us.apache.org/repos/asf/trafodion/blob/e303b3a0/core/sql/executor/ExExeUtilLoad.cpp
----------------------------------------------------------------------
diff --git a/core/sql/executor/ExExeUtilLoad.cpp b/core/sql/executor/ExExeUtilLoad.cpp
index 115f2a4..97c185c 100644
--- a/core/sql/executor/ExExeUtilLoad.cpp
+++ b/core/sql/executor/ExExeUtilLoad.cpp
@@ -1960,7 +1960,7 @@ ex_tcb * ExExeUtilHBaseBulkUnLoadTdb::build(ex_globals * glob)
void ExExeUtilHBaseBulkUnLoadTcb::createHdfsFileError(Int32 hdfsClientRetCode)
{
ComDiagsArea * diagsArea = NULL;
- char* errorMsg = hdfsClient_->getErrorText((HDFS_Client_RetCode)hdfsClientRetCode);
+ char* errorMsg = HdfsClient::getErrorText((HDFS_Client_RetCode)hdfsClientRetCode);
ExRaiseSqlError(getHeap(), &diagsArea, (ExeErrorCode)(8447), NULL,
NULL, NULL, NULL, errorMsg, (char *)GetCliGlobals()->currContext()->getJniErrorStr().data());
ex_queue_entry *pentry_up = qparent_.up->getTailEntry();
@@ -1980,7 +1980,6 @@ ExExeUtilHBaseBulkUnLoadTcb::ExExeUtilHBaseBulkUnLoadTcb(
emptyTarget_(FALSE),
oneFile_(FALSE)
{
- hdfsClient_ = NULL;
ehi_ = ExpHbaseInterface::newInstance(getGlobals()->getDefaultHeap(),
(char*)"", //Later may need to change to hblTdb.server_,
(char*)""); //Later may need to change to hblTdb.zkPort_);
@@ -2005,12 +2004,6 @@ void ExExeUtilHBaseBulkUnLoadTcb::freeResources()
NADELETEBASIC (snapshotsList_, getMyHeap());
snapshotsList_ = NULL;
}
-
- if (hdfsClient_)
- {
- NADELETE(hdfsClient_, HdfsClient, getMyHeap());
- hdfsClient_ = NULL;
- }
NADELETE(ehi_, ExpHbaseInterface, getGlobals()->getDefaultHeap());
ehi_ = NULL;
}
@@ -2192,16 +2185,6 @@ short ExExeUtilHBaseBulkUnLoadTcb::work()
}
setEmptyTarget(hblTdb().getEmptyTarget());
setOneFile(hblTdb().getOneFile());
- if (!hdfsClient_)
- {
- hdfsClient_ = HdfsClient::newInstance((NAHeap *)getMyHeap(), hdfsClientRetCode);
- if (hdfsClientRetCode != HDFS_CLIENT_OK)
- {
- createHdfsFileError(hdfsClientRetCode);
- step_ = UNLOAD_END_ERROR_;
- break;
- }
- }
if ((retcode = ehi_->init(NULL)) != HBASE_ACCESS_SUCCESS)
{
ExHbaseAccessTcb::setupError((NAHeap *)getMyHeap(),qparent_, retcode,
@@ -2213,7 +2196,7 @@ short ExExeUtilHBaseBulkUnLoadTcb::work()
if (!hblTdb().getOverwriteMergeFile() && hblTdb().getMergePath() != NULL)
{
NABoolean exists = FALSE;
- hdfsClientRetCode = hdfsClient_->hdfsExists( hblTdb().getMergePath(), exists);
+ hdfsClientRetCode = HdfsClient::hdfsExists( hblTdb().getMergePath(), exists);
if (hdfsClientRetCode != HDFS_CLIENT_OK)
{
createHdfsFileError(hdfsClientRetCode);
@@ -2298,7 +2281,7 @@ short ExExeUtilHBaseBulkUnLoadTcb::work()
NAString uldPath ( hblTdb().getExtractLocation());
- hdfsClientRetCode = hdfsClient_->hdfsCleanUnloadPath( uldPath);
+ hdfsClientRetCode = HdfsClient::hdfsCleanUnloadPath( uldPath);
if (hdfsClientRetCode != HDFS_CLIENT_OK)
{
createHdfsFileError(hdfsClientRetCode);
@@ -2443,7 +2426,7 @@ short ExExeUtilHBaseBulkUnLoadTcb::work()
NAString srcPath ( hblTdb().getExtractLocation());
NAString dstPath ( hblTdb().getMergePath());
- hdfsClientRetCode = hdfsClient_->hdfsMergeFiles( srcPath, dstPath);
+ hdfsClientRetCode = HdfsClient::hdfsMergeFiles( srcPath, dstPath);
if (hdfsClientRetCode != HDFS_CLIENT_OK)
{
createHdfsFileError(hdfsClientRetCode);
http://git-wip-us.apache.org/repos/asf/trafodion/blob/e303b3a0/core/sql/executor/ExFastTransport.cpp
----------------------------------------------------------------------
diff --git a/core/sql/executor/ExFastTransport.cpp b/core/sql/executor/ExFastTransport.cpp
index ba4cfbb..d2cbd47 100644
--- a/core/sql/executor/ExFastTransport.cpp
+++ b/core/sql/executor/ExFastTransport.cpp
@@ -790,7 +790,7 @@ ExWorkProcRetcode ExHdfsFastExtractTcb::work()
}
else if (!isSequenceFile() && hdfsClient_ == NULL)
{
- hdfsClient_ = HdfsClient::newInstance((NAHeap *)getHeap(), hdfsClientRetCode);
+ hdfsClient_ = HdfsClient::newInstance((NAHeap *)getHeap(), NULL, hdfsClientRetCode);
if (hdfsClientRetCode != HDFS_CLIENT_OK)
{
createHdfsClientFileError(hdfsClientRetCode);
@@ -813,25 +813,30 @@ ExWorkProcRetcode ExHdfsFastExtractTcb::work()
}
else
{
- hdfsClientRetCode = hdfsClient_->hdfsCreate(targetLocation_, isHdfsCompressed());
- if (hdfsClientRetCode != HDFS_CLIENT_OK)
- {
- createHdfsClientFileError(hdfsClientRetCode);
- pstate.step_ = EXTRACT_ERROR;
- break;
- }
- }
- if (feStats)
- {
- feStats->setPartitionNumber(fileNum);
- }
- }
- else
- {
- updateWorkATPDiagsArea(__FILE__,__LINE__,"sockets are not supported");
- pstate.step_ = EXTRACT_ERROR;
- break;
- }
+ hdfsClientRetCode = hdfsClient_->hdfsOpen(targetLocation_, isHdfsCompressed());
+ if (hdfsClientRetCode != HDFS_CLIENT_OK)
+ {
+ createHdfsClientFileError(hdfsClientRetCode);
+ NADELETE(hdfsClient_,
+ HdfsClient,
+ heap_);
+ hdfsClient_ = NULL;
+ pstate.step_ = EXTRACT_ERROR;
+ break;
+ }
+ }
+ if (feStats)
+ {
+ feStats->setPartitionNumber(fileNum);
+ }
+ }
+ else
+ {
+ updateWorkATPDiagsArea(__FILE__,__LINE__,"sockets are not supported");
+ pstate.step_ = EXTRACT_ERROR;
+ break;
+ }
+
for (UInt32 i = 0; i < myTdb().getChildTuple()->numAttrs(); i++)
{
Attributes * attr = myTdb().getChildTableAttr(i);
@@ -1041,7 +1046,7 @@ ExWorkProcRetcode ExHdfsFastExtractTcb::work()
}
else
{
- hdfsClientRetCode = hdfsClient_->hdfsWrite(currBuffer_->data_, bytesToWrite);
+ hdfsClient_->hdfsWrite(currBuffer_->data_, bytesToWrite, hdfsClientRetCode);
if (hdfsClientRetCode != HDFS_CLIENT_OK)
{
createSequenceFileError(hdfsClientRetCode);
http://git-wip-us.apache.org/repos/asf/trafodion/blob/e303b3a0/core/sql/executor/ExHbaseAccess.cpp
----------------------------------------------------------------------
diff --git a/core/sql/executor/ExHbaseAccess.cpp b/core/sql/executor/ExHbaseAccess.cpp
index 2182f9a..8af9308 100644
--- a/core/sql/executor/ExHbaseAccess.cpp
+++ b/core/sql/executor/ExHbaseAccess.cpp
@@ -2991,7 +2991,8 @@ short ExHbaseAccessTcb::createDirectRowBuffer( UInt16 tuppIndex,
{
// Overwrite trailing delimiter with newline.
hiveBuff[hiveBuffInx-1] = '\n';
- sampleFileHdfsClient()->hdfsWrite(hiveBuff, hiveBuffInx);
+ HDFS_Client_RetCode hdfsClientRetcode;
+ sampleFileHdfsClient()->hdfsWrite(hiveBuff, hiveBuffInx, hdfsClientRetcode);
}
return 0;
}
@@ -3263,16 +3264,16 @@ void ExHbaseAccessTcb::handleException(NAHeap *heap,
return;
if (!loggingFileCreated_) {
- logFileHdfsClient_ = HdfsClient::newInstance((NAHeap *)getHeap(), hdfsClientRetcode);
+ logFileHdfsClient_ = HdfsClient::newInstance((NAHeap *)getHeap(), NULL, hdfsClientRetcode);
if (hdfsClientRetcode == HDFS_CLIENT_OK)
- hdfsClientRetcode = logFileHdfsClient_->hdfsCreate(loggingFileName_, FALSE);
+ hdfsClientRetcode = logFileHdfsClient_->hdfsCreate(loggingFileName_, TRUE, FALSE);
if (hdfsClientRetcode == HDFS_CLIENT_OK)
loggingFileCreated_ = TRUE;
else
goto logErrorReturn;
}
- hdfsClientRetcode = logFileHdfsClient_->hdfsWrite(logErrorRow, logErrorRowLen);
+ logFileHdfsClient_->hdfsWrite(logErrorRow, logErrorRowLen, hdfsClientRetcode);
if (hdfsClientRetcode != HDFS_CLIENT_OK)
goto logErrorReturn;
if (errorCond != NULL) {
@@ -3288,7 +3289,7 @@ void ExHbaseAccessTcb::handleException(NAHeap *heap,
errorMsg = (char *)"[UNKNOWN EXCEPTION]\n";
errorMsgLen = strlen(errorMsg);
}
- hdfsClientRetcode = logFileHdfsClient_->hdfsWrite(errorMsg, errorMsgLen);
+ logFileHdfsClient_->hdfsWrite(errorMsg, errorMsgLen, hdfsClientRetcode);
logErrorReturn:
if (hdfsClientRetcode != HDFS_CLIENT_OK) {
loggingErrorDiags_ = ComDiagsArea::allocate(heap);
http://git-wip-us.apache.org/repos/asf/trafodion/blob/e303b3a0/core/sql/executor/ExHbaseIUD.cpp
----------------------------------------------------------------------
diff --git a/core/sql/executor/ExHbaseIUD.cpp b/core/sql/executor/ExHbaseIUD.cpp
index e8aac8c..6a52457 100644
--- a/core/sql/executor/ExHbaseIUD.cpp
+++ b/core/sql/executor/ExHbaseIUD.cpp
@@ -1421,7 +1421,7 @@ ExWorkProcRetcode ExHbaseAccessBulkLoadPrepSQTcb::work()
HDFS_Client_RetCode hdfsClientRetcode;
samplePath.append(filePart);
if (sampleFileHdfsClient_ == NULL)
- sampleFileHdfsClient_ = HdfsClient::newInstance((NAHeap *)getHeap(), hdfsClientRetcode);
+ sampleFileHdfsClient_ = HdfsClient::newInstance((NAHeap *)getHeap(), NULL, hdfsClientRetcode);
if (hdfsClientRetcode == HDFS_CLIENT_OK) {
hdfsClientRetcode = sampleFileHdfsClient_->hdfsOpen(samplePath.data(), FALSE);
if (hdfsClientRetcode != HDFS_CLIENT_OK) {
http://git-wip-us.apache.org/repos/asf/trafodion/blob/e303b3a0/core/sql/executor/ExHdfsScan.cpp
----------------------------------------------------------------------
diff --git a/core/sql/executor/ExHdfsScan.cpp b/core/sql/executor/ExHdfsScan.cpp
index 3ff153e..a56cb06 100644
--- a/core/sql/executor/ExHdfsScan.cpp
+++ b/core/sql/executor/ExHdfsScan.cpp
@@ -124,11 +124,12 @@ ExHdfsScanTcb::ExHdfsScanTcb(
, hdfsScan_(NULL)
, hdfsStats_(NULL)
, hdfsFileInfoListAsArray_(glob->getDefaultHeap(), hdfsScanTdb.getHdfsFileInfoList()->numEntries())
+
{
Space * space = (glob ? glob->getSpace() : 0);
CollHeap * heap = (glob ? glob->getDefaultHeap() : 0);
useLibhdfsScan_ = hdfsScanTdb.getUseLibhdfsScan();
- if (isSequenceFile())
+ if (isSequenceFile() || hdfsScanTdb.isCompressedFile())
useLibhdfsScan_ = TRUE;
lobGlob_ = NULL;
hdfsScanBufMaxSize_ = hdfsScanTdb.hdfsBufSize_;
@@ -229,7 +230,7 @@ ExHdfsScanTcb::ExHdfsScanTcb(
(char*)"");
ex_assert(ehi_ != NULL, "Internal error: ehi_ is null in ExHdfsScan");
HDFS_Client_RetCode hdfsClientRetcode;
- hdfsClient_ = HdfsClient::newInstance((NAHeap *)getHeap(), hdfsClientRetcode);
+ hdfsClient_ = HdfsClient::newInstance((NAHeap *)getHeap(), NULL, hdfsClientRetcode);
ex_assert(hdfsClientRetcode == HDFS_CLIENT_OK, "Internal error: HdfsClient::newInstance returned an error");
// Populate the hdfsInfo list into an array to gain o(1) lookup access
Queue* hdfsInfoList = hdfsScanTdb.getHdfsFileInfoList();
@@ -419,19 +420,23 @@ ExWorkProcRetcode ExHdfsScanTcb::work()
while (!qparent_.down->isEmpty())
{
ex_queue_entry *pentry_down = qparent_.down->getHeadEntry();
+ if (pentry_down->downState.request == ex_queue::GET_NOMORE && step_ != DONE)
+ {
+ if (! useLibhdfsScan_)
+ step_ = STOP_HDFS_SCAN;
+ }
switch (step_)
{
case NOT_STARTED:
{
matches_ = 0;
-
beginRangeNum_ = -1;
numRanges_ = -1;
hdfsOffset_ = 0;
checkRangeDelimiter_ = FALSE;
-
+ if (getStatsEntry())
+ hdfsStats_ = getStatsEntry()->castToExHdfsScanStats();
dataModCheckDone_ = FALSE;
-
myInstNum_ = getGlobals()->getMyInstanceNumber();
hdfsScanBufMaxSize_ = hdfsScanTdb().hdfsBufSize_;
@@ -556,11 +561,14 @@ ExWorkProcRetcode ExHdfsScanTcb::work()
{
if (hdfsScan_ != NULL)
NADELETE(hdfsScan_, HdfsScan, getHeap());
+ if (hdfsFileInfoListAsArray_.entries() == 0) {
+ step_ = DONE;
+ break;
+ }
hdfsScan_ = HdfsScan::newInstance((NAHeap *)getHeap(), hdfsScanBuf_, hdfsScanBufMaxSize_,
&hdfsFileInfoListAsArray_, beginRangeNum_, numRanges_, hdfsScanTdb().rangeTailIOSize_,
hdfsStats_, hdfsScanRetCode);
- if (hdfsScanRetCode != HDFS_SCAN_OK)
- {
+ if (hdfsScanRetCode != HDFS_SCAN_OK) {
setupError(EXE_ERROR_HDFS_SCAN, hdfsScanRetCode, "SETUP_HDFS_SCAN",
currContext->getJniErrorStr(), NULL);
step_ = HANDLE_ERROR_AND_DONE;
@@ -579,7 +587,7 @@ ExWorkProcRetcode ExHdfsScanTcb::work()
break;
case TRAF_HDFS_READ:
{
- hdfsScanRetCode = hdfsScan_->trafHdfsRead((NAHeap *)getHeap(), hdfsStats_, retArray_, sizeof(retArray_)/sizeof(int));
+ hdfsScanRetCode = hdfsScan_->trafHdfsRead(retArray_, sizeof(retArray_)/sizeof(int));
if (hdfsScanRetCode == HDFS_SCAN_EOR) {
step_ = DONE;
break;
@@ -600,6 +608,11 @@ ExWorkProcRetcode ExHdfsScanTcb::work()
else
recordSkip_ = TRUE;
} else {
+ // Throw away the rest of the data when done with the current range
+ if (currRangeBytesRead_ > hdfo->getBytesToRead()) {
+ step_ = TRAF_HDFS_READ;
+ break;
+ }
currRangeBytesRead_ += retArray_[BYTES_COMPLETED];
bufBegin_ = hdfsScanBuf_[retArray_[BUF_NO]].buf_ - headRoomCopied_;
recordSkip_ = FALSE;
@@ -608,10 +621,14 @@ ExWorkProcRetcode ExHdfsScanTcb::work()
extraBytesRead_ = currRangeBytesRead_ - hdfo->getBytesToRead();
else
extraBytesRead_ = 0;
- // headRoom_ is the number of extra bytes read (rangeTailIOSize)
+ // headRoom_ is the number of extra bytes to be read (rangeTailIOSize)
// If EOF is reached while reading the range and the extraBytes read
- // is less than headRoom_, then process all the data till EOF
- if (retArray_[IS_EOF] && extraBytesRead_ < headRoom_)
+ // is less than headRoom_ then process all the data till EOF
+ // TODO: If the whole range fits in one buffer, it is need too to process rows till EOF for the last range alone
+ // No easy way to identify that last range read, but can identify that it is not the first range.
+ // The rows could be read more than once if there are more than 2 ranges.
+ // Fix optimizer not to have more than 2 ranges in that case
+ if (retArray_[IS_EOF] && extraBytesRead_ < headRoom_ && hdfo->getStartOffset() != 0)
extraBytesRead_ = 0;
bufLogicalEnd_ = hdfsScanBuf_[retArray_[BUF_NO]].buf_ + retArray_[BYTES_COMPLETED] - extraBytesRead_;
prevRangeNum_ = retArray_[RANGE_NO];
@@ -628,6 +645,8 @@ ExWorkProcRetcode ExHdfsScanTcb::work()
step_ = HANDLE_ERROR_AND_DONE;
break;
}
+ //add changedLen since hdfs_strchr will remove the pointer ahead to remove the \r
+ hdfsBufNextRow_ += 1 + changedLen; // point past record delimiter.
}
else
hdfsBufNextRow_ = (char *)bufBegin_;
@@ -646,6 +665,17 @@ ExWorkProcRetcode ExHdfsScanTcb::work()
step_ = TRAF_HDFS_READ;
}
break;
+ case STOP_HDFS_SCAN:
+ {
+ hdfsScanRetCode = hdfsScan_->stop();
+ if (hdfsScanRetCode != HDFS_SCAN_OK) {
+ setupError(EXE_ERROR_HDFS_SCAN, hdfsScanRetCode, "HdfsScan::stop",
+ currContext->getJniErrorStr(), NULL);
+ step_ = HANDLE_ERROR_AND_DONE;
+ }
+ step_ = DONE;
+ }
+ break;
case INIT_HDFS_CURSOR:
{
hdfo_ = getRange(currRangeNum_);
@@ -667,6 +697,7 @@ ExWorkProcRetcode ExHdfsScanTcb::work()
sprintf(cursorId_, "%d", currRangeNum_);
stopOffset_ = hdfsOffset_ + hdfo_->getBytesToRead();
+
step_ = OPEN_HDFS_CURSOR;
}
@@ -1076,16 +1107,19 @@ ExWorkProcRetcode ExHdfsScanTcb::work()
if (startOfNextRow == NULL)
{
- if (useLibhdfsScan_)
+ if (useLibhdfsScan_)
step_ = REPOS_HDFS_DATA;
else {
- if (retArray_[IS_EOF])
+ if (retArray_[IS_EOF]) {
+ headRoomCopied_ = 0;
step_ = TRAF_HDFS_READ;
+ }
else
step_ = COPY_TAIL_TO_HEAD;
}
+ // Looks like we can break always
if (!exception_)
- break;
+ break;
}
else
{
@@ -1096,13 +1130,11 @@ ExWorkProcRetcode ExHdfsScanTcb::work()
}
else {
if ((BYTE *)startOfNextRow > bufLogicalEnd_) {
- step_ = TRAF_HDFS_READ;
+ headRoomCopied_ = 0;
hdfsBufNextRow_ = NULL;
- }
- else
+ } else
hdfsBufNextRow_ = startOfNextRow;
- }
-
+ }
}
if (exception_)
@@ -1304,7 +1336,7 @@ ExWorkProcRetcode ExHdfsScanTcb::work()
if ((pentry_down->downState.request == ex_queue::GET_N) &&
(pentry_down->downState.requestValue == matches_)) {
if (useLibhdfsScan_)
- step_ = CLOSE_FILE;
+ step_ = CLOSE_HDFS_CURSOR;
else
step_ = DONE;
}
@@ -1361,8 +1393,7 @@ ExWorkProcRetcode ExHdfsScanTcb::work()
up_entry->getTupp(hdfsScanTdb().tuppIndex_).getDataPointer());
}
}
- }
-
+ }
up_entry->upState.setMatchNo(++matches_);
if (matches_ == matchBrkPoint_)
brkpoint();
@@ -1744,17 +1775,18 @@ char * ExHdfsScanTcb::extractAndTransformAsciiSourceToSqlRow(int &err,
{
sourceRowEnd = hdfs_strchr(sourceData, rd, sourceDataEnd, checkRangeDelimiter_, mode, &changedLen);
hdfsLoggingRowEnd_ = sourceRowEnd + changedLen;
-
+
if (sourceRowEnd == NULL)
return NULL;
- if ((endOfRequestedRange) &&
+ if (useLibhdfsScan_) {
+ if ((endOfRequestedRange) &&
(sourceRowEnd >= endOfRequestedRange)) {
- checkRangeDelimiter_ = TRUE;
- *(sourceRowEnd +1)= RANGE_DELIMITER;
+ checkRangeDelimiter_ = TRUE;
+ *(sourceRowEnd +1)= RANGE_DELIMITER;
+ }
}
-
- // no columns need to be converted. For e.g. count(*) with no predicate
- return sourceRowEnd+1;
+ // no columns need to be converted. For e.g. count(*) with no predicate
+ return sourceRowEnd+1;
}
Lng32 neededColIndex = 0;
@@ -2016,54 +2048,6 @@ void ExHdfsScanTcb::deallocateRuntimeRanges()
}
}
-void ExHdfsScanTcb::handleException(NAHeap *heap,
- char *logErrorRow,
- Lng32 logErrorRowLen,
- ComCondition *errorCond)
-{
- Lng32 errorMsgLen = 0;
- charBuf *cBuf = NULL;
- char *errorMsg;
- HDFS_Client_RetCode hdfsClientRetcode;
-
- if (loggingErrorDiags_ != NULL)
- return;
-
- if (!loggingFileCreated_) {
- hdfsClient_ = HdfsClient::newInstance((NAHeap *)getHeap(), hdfsClientRetcode);
- if (hdfsClientRetcode == HDFS_CLIENT_OK)
- hdfsClientRetcode = hdfsClient_->hdfsCreate(loggingFileName_, FALSE);
- if (hdfsClientRetcode == HDFS_CLIENT_OK)
- loggingFileCreated_ = TRUE;
- else
- goto logErrorReturn;
- }
- hdfsClientRetcode = hdfsClient_->hdfsWrite(logErrorRow, logErrorRowLen);
- if (hdfsClientRetcode != HDFS_CLIENT_OK)
- goto logErrorReturn;
- if (errorCond != NULL) {
- errorMsgLen = errorCond->getMessageLength();
- const NAWcharBuf wBuf((NAWchar*)errorCond->getMessageText(), errorMsgLen, heap);
- cBuf = unicodeToISO88591(wBuf, heap, cBuf);
- errorMsg = (char *)cBuf->data();
- errorMsgLen = cBuf -> getStrLen();
- errorMsg[errorMsgLen]='\n';
- errorMsgLen++;
- }
- else {
- errorMsg = (char *)"[UNKNOWN EXCEPTION]\n";
- errorMsgLen = strlen(errorMsg);
- }
- hdfsClientRetcode = hdfsClient_->hdfsWrite(errorMsg, errorMsgLen);
-logErrorReturn:
- if (hdfsClientRetcode != HDFS_CLIENT_OK) {
- loggingErrorDiags_ = ComDiagsArea::allocate(heap);
- *loggingErrorDiags_ << DgSqlCode(EXE_ERROR_WHILE_LOGGING)
- << DgString0(loggingFileName_)
- << DgString1((char *)GetCliGlobals()->currContext()->getJniErrorStr().data());
- }
-}
-
short ExHdfsScanTcb::moveRowToUpQueue(const char * row, Lng32 len,
short * rc, NABoolean isVarchar)
{
@@ -2164,7 +2148,7 @@ short ExHdfsScanTcb::handleDone(ExWorkProcRetcode &rc)
return 0;
}
-/*
+
void ExHdfsScanTcb::handleException(NAHeap *heap,
char *logErrorRow,
Lng32 logErrorRowLen,
@@ -2179,15 +2163,15 @@ void ExHdfsScanTcb::handleException(NAHeap *heap,
return;
if (!loggingFileCreated_) {
- logFileHdfsClient_ = HdfsClient::newInstance((NAHeap *)getHeap(), hdfsClientRetcode);
+ logFileHdfsClient_ = HdfsClient::newInstance((NAHeap *)getHeap(), NULL, hdfsClientRetcode);
if (hdfsClientRetcode == HDFS_CLIENT_OK)
- hdfsClientRetcode = logFileHdfsClient_->hdfsCreate(loggingFileName_, FALSE);
+ hdfsClientRetcode = logFileHdfsClient_->hdfsCreate(loggingFileName_, TRUE, FALSE);
if (hdfsClientRetcode == HDFS_CLIENT_OK)
loggingFileCreated_ = TRUE;
else
goto logErrorReturn;
}
- hdfsClientRetcode = logFileHdfsClient_->hdfsWrite(logErrorRow, logErrorRowLen);
+ logFileHdfsClient_->hdfsWrite(logErrorRow, logErrorRowLen, hdfsClientRetcode);
if (hdfsClientRetcode != HDFS_CLIENT_OK)
goto logErrorReturn;
if (errorCond != NULL) {
@@ -2203,7 +2187,7 @@ void ExHdfsScanTcb::handleException(NAHeap *heap,
errorMsg = (char *)"[UNKNOWN EXCEPTION]\n";
errorMsgLen = strlen(errorMsg);
}
- hdfsClientRetcode = logFileHdfsClient_->hdfsWrite(errorMsg, errorMsgLen);
+ logFileHdfsClient_->hdfsWrite(errorMsg, errorMsgLen, hdfsClientRetcode);
logErrorReturn:
if (hdfsClientRetcode != HDFS_CLIENT_OK) {
loggingErrorDiags_ = ComDiagsArea::allocate(heap);
@@ -2212,7 +2196,7 @@ logErrorReturn:
<< DgString1((char *)GetCliGlobals()->currContext()->getJniErrorStr().data());
}
}
-*/
+
////////////////////////////////////////////////////////////////////////
// ORC files
http://git-wip-us.apache.org/repos/asf/trafodion/blob/e303b3a0/core/sql/executor/ExHdfsScan.h
----------------------------------------------------------------------
diff --git a/core/sql/executor/ExHdfsScan.h b/core/sql/executor/ExHdfsScan.h
index 04b632e..aa01b6e 100644
--- a/core/sql/executor/ExHdfsScan.h
+++ b/core/sql/executor/ExHdfsScan.h
@@ -227,6 +227,7 @@ protected:
, SETUP_HDFS_SCAN
, TRAF_HDFS_READ
, COPY_TAIL_TO_HEAD
+ , STOP_HDFS_SCAN
} step_,nextStep_;
/////////////////////////////////////////////////////
http://git-wip-us.apache.org/repos/asf/trafodion/blob/e303b3a0/core/sql/executor/ExSMCommon.cpp
----------------------------------------------------------------------
diff --git a/core/sql/executor/ExSMCommon.cpp b/core/sql/executor/ExSMCommon.cpp
index fde847f..d946732 100644
--- a/core/sql/executor/ExSMCommon.cpp
+++ b/core/sql/executor/ExSMCommon.cpp
@@ -29,6 +29,7 @@
#include <sys/syscall.h>
#include <unistd.h>
#include <pthread.h>
+#include "Platform.h"
#include "ExSMCommon.h"
#include "ExSMGlobals.h"
#include "ExSMEvent.h"
http://git-wip-us.apache.org/repos/asf/trafodion/blob/e303b3a0/core/sql/executor/ExSMCommon.h
----------------------------------------------------------------------
diff --git a/core/sql/executor/ExSMCommon.h b/core/sql/executor/ExSMCommon.h
index 1cfe3e3..0a23d68 100644
--- a/core/sql/executor/ExSMCommon.h
+++ b/core/sql/executor/ExSMCommon.h
@@ -23,11 +23,10 @@
#ifndef EXSM_COMMON_H
#define EXSM_COMMON_H
-#define __STDC_LIMIT_MACROS
-#define __STDC_FORMAT_MACROS
#include <stdint.h>
#include <inttypes.h>
#include <stdio.h>
+#include "Platform.h"
#include "sm.h"
#include "NAAssert.h"
http://git-wip-us.apache.org/repos/asf/trafodion/blob/e303b3a0/core/sql/executor/ExSMGlobals.cpp
----------------------------------------------------------------------
diff --git a/core/sql/executor/ExSMGlobals.cpp b/core/sql/executor/ExSMGlobals.cpp
index 45dda94..fb205dc 100644
--- a/core/sql/executor/ExSMGlobals.cpp
+++ b/core/sql/executor/ExSMGlobals.cpp
@@ -26,6 +26,7 @@
#include <signal.h>
#include <errno.h>
#include "seabed/pctl.h"
+#include "Platform.h"
#include "ExSMGlobals.h"
#include "ExSMTask.h"
#include "ExSMReader.h"
http://git-wip-us.apache.org/repos/asf/trafodion/blob/e303b3a0/core/sql/executor/ExSMQueue.cpp
----------------------------------------------------------------------
diff --git a/core/sql/executor/ExSMQueue.cpp b/core/sql/executor/ExSMQueue.cpp
index ca1f857..7eac848 100644
--- a/core/sql/executor/ExSMQueue.cpp
+++ b/core/sql/executor/ExSMQueue.cpp
@@ -20,7 +20,7 @@
//
// @@@ END COPYRIGHT @@@
**********************************************************************/
-
+#include "Platform.h"
#include "ExSMQueue.h"
#include "NAMemory.h"
http://git-wip-us.apache.org/repos/asf/trafodion/blob/e303b3a0/core/sql/executor/ExSMReader.cpp
----------------------------------------------------------------------
diff --git a/core/sql/executor/ExSMReader.cpp b/core/sql/executor/ExSMReader.cpp
index f59cbc7..315be70 100644
--- a/core/sql/executor/ExSMReader.cpp
+++ b/core/sql/executor/ExSMReader.cpp
@@ -23,6 +23,7 @@
#include <unistd.h>
#include <time.h>
+#include "Platform.h"
#include "seabed/pctl.h"
#include "seabed/pevents.h"
#include "ExSMReader.h"
http://git-wip-us.apache.org/repos/asf/trafodion/blob/e303b3a0/core/sql/executor/ExSMShortMessage.cpp
----------------------------------------------------------------------
diff --git a/core/sql/executor/ExSMShortMessage.cpp b/core/sql/executor/ExSMShortMessage.cpp
index b564605..c2a24ef 100644
--- a/core/sql/executor/ExSMShortMessage.cpp
+++ b/core/sql/executor/ExSMShortMessage.cpp
@@ -20,7 +20,7 @@
//
// @@@ END COPYRIGHT @@@
**********************************************************************/
-
+#include "Platform.h"
#include "ExSMShortMessage.h"
#include "ExSMGlobals.h"
http://git-wip-us.apache.org/repos/asf/trafodion/blob/e303b3a0/core/sql/executor/ExSMTaskList.cpp
----------------------------------------------------------------------
diff --git a/core/sql/executor/ExSMTaskList.cpp b/core/sql/executor/ExSMTaskList.cpp
index 698c63b..58e3c7c 100644
--- a/core/sql/executor/ExSMTaskList.cpp
+++ b/core/sql/executor/ExSMTaskList.cpp
@@ -20,7 +20,7 @@
//
// @@@ END COPYRIGHT @@@
**********************************************************************/
-
+#include "Platform.h"
#include "ExSMTaskList.h"
#include "ExSMTask.h"
#include "ExSMTrace.h"
http://git-wip-us.apache.org/repos/asf/trafodion/blob/e303b3a0/core/sql/executor/HdfsClient_JNI.cpp
----------------------------------------------------------------------
diff --git a/core/sql/executor/HdfsClient_JNI.cpp b/core/sql/executor/HdfsClient_JNI.cpp
index 5f0f810..d5eb4ec 100644
--- a/core/sql/executor/HdfsClient_JNI.cpp
+++ b/core/sql/executor/HdfsClient_JNI.cpp
@@ -39,16 +39,30 @@ pthread_mutex_t HdfsScan::javaMethodsInitMutex_ = PTHREAD_MUTEX_INITIALIZER;
static const char* const hdfsScanErrorEnumStr[] =
{
"Error in HdfsScan::setScanRanges"
- ,"Java Exception in HdfsScan::setScanRanges"
+ ,"Java exception in HdfsScan::setScanRanges"
,"Error in HdfsScan::trafHdfsRead"
- ,"Java Exceptiokn in HdfsScan::trafHdfsRead"
+ ,"Java exception in HdfsScan::trafHdfsRead"
, "Hdfs scan End of Ranges"
+ ,"Error in HdfsScan::stop"
+ ,"Java exception in HdfsScan::stop"
};
//////////////////////////////////////////////////////////////////////////////
//
//////////////////////////////////////////////////////////////////////////////
+HdfsScan::~HdfsScan()
+{
+ if (j_buf1_ != NULL) {
+ jenv_->DeleteGlobalRef(j_buf1_);
+ j_buf1_ = NULL;
+ }
+ if (j_buf2_ != NULL) {
+ jenv_->DeleteGlobalRef(j_buf2_);
+ j_buf2_ = NULL;
+ }
+}
+
HDFS_Scan_RetCode HdfsScan::init()
{
static char className[]="org/trafodion/sql/HdfsScan";
@@ -72,6 +86,8 @@ HDFS_Scan_RetCode HdfsScan::init()
JavaMethods_[JM_SET_SCAN_RANGES].jm_signature = "(Ljava/nio/ByteBuffer;Ljava/nio/ByteBuffer;[Ljava/lang/String;[J[J[I)V";
JavaMethods_[JM_TRAF_HDFS_READ].jm_name = "trafHdfsRead";
JavaMethods_[JM_TRAF_HDFS_READ].jm_signature = "()[I";
+ JavaMethods_[JM_STOP].jm_name = "stop";
+ JavaMethods_[JM_STOP].jm_signature = "()V";
rc = (HDFS_Scan_RetCode)JavaObjectInterface::init(className, javaClass_, JavaMethods_, (Int32)JM_LAST, javaMethodsInitialized_);
if (rc == HDFS_SCAN_OK)
@@ -90,9 +106,8 @@ char* HdfsScan::getErrorText(HDFS_Scan_RetCode errEnum)
}
/////////////////////////////////////////////////////////////////////////////
-HDFS_Scan_RetCode HdfsScan::setScanRanges(NAHeap *heap, ExHdfsScanTcb::HDFS_SCAN_BUF *hdfsScanBuf, int scanBufSize,
- HdfsFileInfoArray *hdfsFileInfoArray, Int32 beginRangeNum, Int32 numRanges, int rangeTailIOSize,
- ExHdfsScanStats *hdfsStats)
+HDFS_Scan_RetCode HdfsScan::setScanRanges(ExHdfsScanTcb::HDFS_SCAN_BUF *hdfsScanBuf, int scanBufSize,
+ HdfsFileInfoArray *hdfsFileInfoArray, Int32 beginRangeNum, Int32 numRanges, int rangeTailIOSize)
{
QRLogger::log(CAT_SQL_HDFS, LL_DEBUG, "HdfsScan::setScanRanges() called.");
@@ -105,13 +120,24 @@ HDFS_Scan_RetCode HdfsScan::setScanRanges(NAHeap *heap, ExHdfsScanTcb::HDFS_SCAN
jenv_->PopLocalFrame(NULL);
return HDFS_SCAN_ERROR_SET_SCAN_RANGES_PARAM;
}
-
+ j_buf1_ = jenv_->NewGlobalRef(j_buf1);
+ if (j_buf1_ == NULL) {
+ GetCliGlobals()->setJniErrorStr(getErrorText(HDFS_SCAN_ERROR_SET_SCAN_RANGES_PARAM));
+ jenv_->PopLocalFrame(NULL);
+ return HDFS_SCAN_ERROR_SET_SCAN_RANGES_PARAM;
+ }
jobject j_buf2 = jenv_->NewDirectByteBuffer(hdfsScanBuf[1].buf_, scanBufSize);
if (j_buf2 == NULL) {
GetCliGlobals()->setJniErrorStr(getErrorText(HDFS_SCAN_ERROR_SET_SCAN_RANGES_PARAM));
jenv_->PopLocalFrame(NULL);
return HDFS_SCAN_ERROR_SET_SCAN_RANGES_PARAM;
}
+ j_buf2_ = jenv_->NewGlobalRef(j_buf2);
+ if (j_buf2_ == NULL) {
+ GetCliGlobals()->setJniErrorStr(getErrorText(HDFS_SCAN_ERROR_SET_SCAN_RANGES_PARAM));
+ jenv_->PopLocalFrame(NULL);
+ return HDFS_SCAN_ERROR_SET_SCAN_RANGES_PARAM;
+ }
jobjectArray j_filenames = NULL;
jlongArray j_offsets = NULL;
jlongArray j_lens = NULL;
@@ -172,13 +198,13 @@ HDFS_Scan_RetCode HdfsScan::setScanRanges(NAHeap *heap, ExHdfsScanTcb::HDFS_SCAN
jenv_->SetIntArrayRegion(j_rangenums, rangeCount, 1, &tdbRangeNum);
}
- if (hdfsStats)
- hdfsStats->getHdfsTimer().start();
+ if (hdfsStats_ != NULL)
+ hdfsStats_->getHdfsTimer().start();
tsRecentJMFromJNI = JavaMethods_[JM_SET_SCAN_RANGES].jm_full_name;
jenv_->CallVoidMethod(javaObj_, JavaMethods_[JM_SET_SCAN_RANGES].methodID, j_buf1, j_buf2, j_filenames, j_offsets, j_lens, j_rangenums);
- if (hdfsStats) {
- hdfsStats->incMaxHdfsIOTime(hdfsStats->getHdfsTimer().stop());
- hdfsStats->incHdfsCalls();
+ if (hdfsStats_ != NULL) {
+ hdfsStats_->incMaxHdfsIOTime(hdfsStats_->getHdfsTimer().stop());
+ hdfsStats_->incHdfsCalls();
}
if (jenv_->ExceptionCheck()) {
@@ -204,9 +230,11 @@ HdfsScan *HdfsScan::newInstance(NAHeap *heap, ExHdfsScanTcb::HDFS_SCAN_BUF *hdfs
if (hdfsScan != NULL) {
hdfsScanRetCode = hdfsScan->init();
if (hdfsScanRetCode == HDFS_SCAN_OK)
- hdfsScanRetCode = hdfsScan->setScanRanges(heap, hdfsScanBuf, scanBufSize,
- hdfsFileInfoArray, beginRangeNum, numRanges, rangeTailIOSize, hdfsStats);
- if (hdfsScanRetCode != HDFS_SCAN_OK) {
+ hdfsScanRetCode = hdfsScan->setScanRanges(hdfsScanBuf, scanBufSize,
+ hdfsFileInfoArray, beginRangeNum, numRanges, rangeTailIOSize);
+ if (hdfsScanRetCode == HDFS_SCAN_OK)
+ hdfsScan->setHdfsStats(hdfsStats);
+ else {
NADELETE(hdfsScan, HdfsScan, heap);
hdfsScan = NULL;
}
@@ -215,20 +243,20 @@ HdfsScan *HdfsScan::newInstance(NAHeap *heap, ExHdfsScanTcb::HDFS_SCAN_BUF *hdfs
}
-HDFS_Scan_RetCode HdfsScan::trafHdfsRead(NAHeap *heap, ExHdfsScanStats *hdfsStats, int retArray[], short arrayLen)
+HDFS_Scan_RetCode HdfsScan::trafHdfsRead(int retArray[], short arrayLen)
{
QRLogger::log(CAT_SQL_HDFS, LL_DEBUG, "HdfsScan::trafHdfsRead() called.");
if (initJNIEnv() != JOI_OK)
return HDFS_SCAN_ERROR_TRAF_HDFS_READ_PARAM;
- if (hdfsStats)
- hdfsStats->getHdfsTimer().start();
+ if (hdfsStats_ != NULL)
+ hdfsStats_->getHdfsTimer().start();
tsRecentJMFromJNI = JavaMethods_[JM_TRAF_HDFS_READ].jm_full_name;
jintArray j_retArray = (jintArray)jenv_->CallObjectMethod(javaObj_, JavaMethods_[JM_TRAF_HDFS_READ].methodID);
- if (hdfsStats) {
- hdfsStats->incMaxHdfsIOTime(hdfsStats->getHdfsTimer().stop());
- hdfsStats->incHdfsCalls();
+ if (hdfsStats_ != NULL) {
+ hdfsStats_->incMaxHdfsIOTime(hdfsStats_->getHdfsTimer().stop());
+ hdfsStats_->incHdfsCalls();
}
if (jenv_->ExceptionCheck()) {
@@ -247,6 +275,32 @@ HDFS_Scan_RetCode HdfsScan::trafHdfsRead(NAHeap *heap, ExHdfsScanStats *hdfsStat
return HDFS_SCAN_OK;
}
+HDFS_Scan_RetCode HdfsScan::stop()
+{
+ QRLogger::log(CAT_SQL_HDFS, LL_DEBUG, "HdfsScan::stop() called.");
+
+ if (initJNIEnv() != JOI_OK)
+ return HDFS_SCAN_ERROR_STOP_PARAM;
+
+ if (hdfsStats_ != NULL)
+ hdfsStats_->getHdfsTimer().start();
+ tsRecentJMFromJNI = JavaMethods_[JM_STOP].jm_full_name;
+ jenv_->CallVoidMethod(javaObj_, JavaMethods_[JM_STOP].methodID);
+ if (hdfsStats_ != NULL) {
+ hdfsStats_->incMaxHdfsIOTime(hdfsStats_->getHdfsTimer().stop());
+ hdfsStats_->incHdfsCalls();
+ }
+
+ if (jenv_->ExceptionCheck()) {
+ getExceptionDetails();
+ logError(CAT_SQL_HDFS, __FILE__, __LINE__);
+ logError(CAT_SQL_HDFS, "HdfsScan::stop()", getLastError());
+ jenv_->PopLocalFrame(NULL);
+ return HDFS_SCAN_ERROR_STOP_EXCEPTION;
+ }
+ return HDFS_SCAN_OK;
+}
+
// ===========================================================================
// ===== Class HdfsClient
// ===========================================================================
@@ -258,24 +312,33 @@ pthread_mutex_t HdfsClient::javaMethodsInitMutex_ = PTHREAD_MUTEX_INITIALIZER;
static const char* const hdfsClientErrorEnumStr[] =
{
- "JNI NewStringUTF() in hdfsCreate()."
- ,"Java exception in hdfsCreate()."
- ,"JNI NewStringUTF() in hdfsOpen()."
- ,"Java exception in hdfsOpen()."
- ,"JNI NewStringUTF() in hdfsWrite()."
- ,"Java exception in hdfsWrite()."
- ,"Java exception in hdfsClose()."
- ,"JNI NewStringUTF() in hdfsMergeFiles()."
- ,"Java exception in hdfsMergeFiles()."
- ,"JNI NewStringUTF() in hdfsCleanUnloadPath()."
- ,"Java exception in hdfsCleanUnloadPath()."
- ,"JNI NewStringUTF() in hdfsExists()."
- ,"Java exception in hdfsExists()."
- ,"JNI NewStringUTF() in hdfsDeletePath()."
- ,"Java exception in hdfsDeletePath()."
- ,"Error in setHdfsFileInfo()."
- ,"Error in hdfsListDirectory()."
- ,"Java exception in hdfsListDirectory()."
+ "JNI NewStringUTF() in HdfsClient::hdfsCreate()."
+ ,"Java exception in HdfsClient::hdfsCreate()."
+ ,"JNI NewStringUTF() in HdfsClient::hdfsOpen()."
+ ,"Java exception in HdfsClient::hdfsOpen()."
+ ,"JNI NewStringUTF() in HdfsClient::hdfsWrite()."
+ ,"Java exception in HdfsClient::hdfsWrite()."
+ ,"Error in HdfsClient::hdfsRead()."
+ ,"Java exception in HdfsClient::hdfsRead()."
+ ,"Java exception in HdfsClient::hdfsClose()."
+ ,"JNI NewStringUTF() in HdfsClient::hdfsMergeFiles()."
+ ,"Java exception in HdfsClient::hdfsMergeFiles()."
+ ,"JNI NewStringUTF() in HdfsClient::hdfsCleanUnloadPath()."
+ ,"Java exception in HdfsClient::hdfsCleanUnloadPath()."
+ ,"JNI NewStringUTF() in HdfsClient::hdfsExists()."
+ ,"Java exception in HdfsClient::hdfsExists()."
+ ,"JNI NewStringUTF() in HdfsClient::hdfsDeletePath()."
+ ,"Java exception in HdfsClient::hdfsDeletePath()."
+ ,"Error in HdfsClient::setHdfsFileInfo()."
+ ,"Error in HdfsClient::hdfsListDirectory()."
+ ,"Java exception in HdfsClient::hdfsListDirectory()."
+ ,"preparing parameters for HdfsClient::getHiveTableMaxModificationTs()."
+ ,"java exception in HdfsClient::getHiveTableMaxModificationTs()."
+ ,"Error in HdfsClient::getFsDefaultName()."
+ ,"Java exception in HdfsClient::getFsDefaultName()."
+ ,"Buffer is small in HdfsClient::getFsDefaultName()."
+ ,"Error in HdfsClient::hdfsCreateDirectory()."
+ ,"Java exception in HdfsClient::hdfsCreateDirectory()."
};
//////////////////////////////////////////////////////////////////////////////
@@ -285,9 +348,24 @@ HdfsClient::~HdfsClient()
{
QRLogger::log(CAT_SQL_HDFS, LL_DEBUG, "HdfsClient::~HdfsClient() called.");
deleteHdfsFileInfo();
+ if (path_ != NULL)
+ NADELETEBASIC(path_, getHeap());
+}
+
+void HdfsClient::deleteHdfsFileInfo()
+{
+ for (int i = 0; i < numFiles_ ; i ++) {
+ NADELETEBASIC(hdfsFileInfo_[i].mName, getHeap());
+ NADELETEBASIC(hdfsFileInfo_[i].mOwner, getHeap());
+ NADELETEBASIC(hdfsFileInfo_[i].mGroup, getHeap());
+ }
+ if (hdfsFileInfo_ != NULL)
+ NADELETEBASICARRAY(hdfsFileInfo_, getHeap());
+ numFiles_ = 0;
+ hdfsFileInfo_ = NULL;
}
-HdfsClient *HdfsClient::newInstance(NAHeap *heap, HDFS_Client_RetCode &retCode)
+HdfsClient *HdfsClient::newInstance(NAHeap *heap, ExHdfsScanStats *hdfsStats, HDFS_Client_RetCode &retCode)
{
QRLogger::log(CAT_SQL_HDFS, LL_DEBUG, "HdfsClient::newInstance() called.");
@@ -297,7 +375,9 @@ HdfsClient *HdfsClient::newInstance(NAHeap *heap, HDFS_Client_RetCode &retCode)
HdfsClient *hdfsClient = new (heap) HdfsClient(heap);
if (hdfsClient != NULL) {
retCode = hdfsClient->init();
- if (retCode != HDFS_CLIENT_OK) {
+ if (retCode == HDFS_CLIENT_OK)
+ hdfsClient->setHdfsStats(hdfsStats);
+ else {
NADELETE(hdfsClient, HdfsClient, heap);
hdfsClient = NULL;
}
@@ -305,16 +385,30 @@ HdfsClient *HdfsClient::newInstance(NAHeap *heap, HDFS_Client_RetCode &retCode)
return hdfsClient;
}
-void HdfsClient::deleteHdfsFileInfo()
+HdfsClient* HdfsClient::getInstance()
{
- for (int i = 0; i < numFiles_ ; i ++) {
- NADELETEBASIC(hdfsFileInfo_[i].mName, getHeap());
- NADELETEBASIC(hdfsFileInfo_[i].mOwner, getHeap());
- NADELETEBASIC(hdfsFileInfo_[i].mGroup, getHeap());
+ ContextCli *currContext = GetCliGlobals()->currContext();
+ HdfsClient *hdfsClient = currContext->getHDFSClient();
+ HDFS_Client_RetCode retcode;
+ if (hdfsClient == NULL) {
+ NAHeap *heap = currContext->exHeap();
+ hdfsClient = newInstance(heap, NULL, retcode);
+ if (retcode != HDFS_CLIENT_OK)
+ return NULL;
+ currContext->setHDFSClient(hdfsClient);
+ }
+ return hdfsClient;
+}
+
+void HdfsClient::deleteInstance()
+{
+ ContextCli *currContext = GetCliGlobals()->currContext();
+ HdfsClient *hdfsClient = currContext->getHDFSClient();
+ if (hdfsClient != NULL) {
+ NAHeap *heap = currContext->exHeap();
+ NADELETE(hdfsClient, HdfsClient, heap);
+ currContext->setHDFSClient(NULL);
}
- NADELETEBASIC(hdfsFileInfo_, getHeap());
- numFiles_ = 0;
- hdfsFileInfo_ = NULL;
}
HDFS_Client_RetCode HdfsClient::init()
@@ -337,11 +431,13 @@ HDFS_Client_RetCode HdfsClient::init()
JavaMethods_[JM_CTOR ].jm_name = "<init>";
JavaMethods_[JM_CTOR ].jm_signature = "()V";
JavaMethods_[JM_HDFS_CREATE ].jm_name = "hdfsCreate";
- JavaMethods_[JM_HDFS_CREATE ].jm_signature = "(Ljava/lang/String;Z)Z";
+ JavaMethods_[JM_HDFS_CREATE ].jm_signature = "(Ljava/lang/String;ZZ)Z";
JavaMethods_[JM_HDFS_OPEN ].jm_name = "hdfsOpen";
JavaMethods_[JM_HDFS_OPEN ].jm_signature = "(Ljava/lang/String;Z)Z";
JavaMethods_[JM_HDFS_WRITE ].jm_name = "hdfsWrite";
- JavaMethods_[JM_HDFS_WRITE ].jm_signature = "([BJ)Z";
+ JavaMethods_[JM_HDFS_WRITE ].jm_signature = "([B)I";
+ JavaMethods_[JM_HDFS_READ ].jm_name = "hdfsRead";
+ JavaMethods_[JM_HDFS_READ ].jm_signature = "(Ljava/nio/ByteBuffer;)I";
JavaMethods_[JM_HDFS_CLOSE ].jm_name = "hdfsClose";
JavaMethods_[JM_HDFS_CLOSE ].jm_signature = "()Z";
JavaMethods_[JM_HDFS_MERGE_FILES].jm_name = "hdfsMergeFiles";
@@ -354,6 +450,12 @@ HDFS_Client_RetCode HdfsClient::init()
JavaMethods_[JM_HDFS_DELETE_PATH].jm_signature = "(Ljava/lang/String;)Z";
JavaMethods_[JM_HDFS_LIST_DIRECTORY].jm_name = "hdfsListDirectory";
JavaMethods_[JM_HDFS_LIST_DIRECTORY].jm_signature = "(Ljava/lang/String;J)I";
+ JavaMethods_[JM_HIVE_TBL_MAX_MODIFICATION_TS].jm_name = "getHiveTableMaxModificationTs";
+ JavaMethods_[JM_HIVE_TBL_MAX_MODIFICATION_TS ].jm_signature = "(Ljava/lang/String;I)J";
+ JavaMethods_[JM_GET_FS_DEFAULT_NAME].jm_name = "getFsDefaultName";
+ JavaMethods_[JM_GET_FS_DEFAULT_NAME].jm_signature = "()Ljava/lang/String;";
+ JavaMethods_[JM_HDFS_CREATE_DIRECTORY].jm_name = "hdfsCreateDirectory";
+ JavaMethods_[JM_HDFS_CREATE_DIRECTORY].jm_signature = "(Ljava/lang/String;)Z";
rc = (HDFS_Client_RetCode)JavaObjectInterface::init(className, javaClass_, JavaMethods_, (Int32)JM_LAST, javaMethodsInitialized_);
if (rc == HDFS_CLIENT_OK)
javaMethodsInitialized_ = TRUE;
@@ -373,13 +475,22 @@ char* HdfsClient::getErrorText(HDFS_Client_RetCode errEnum)
return (char*)hdfsClientErrorEnumStr[errEnum-HDFS_CLIENT_FIRST];
}
-HDFS_Client_RetCode HdfsClient::hdfsCreate(const char* path, NABoolean compress)
+void HdfsClient::setPath(const char *path)
+{
+ if (path_ != NULL)
+ NADELETEBASIC(path_, getHeap());
+ short len = strlen(path);
+ path_ = new (getHeap()) char[len+1];
+ strcpy(path_, path);
+}
+
+HDFS_Client_RetCode HdfsClient::hdfsCreate(const char* path, NABoolean overwrite, NABoolean compress)
{
QRLogger::log(CAT_SQL_HDFS, LL_DEBUG, "HdfsClient::hdfsCreate(%s) called.", path);
if (initJNIEnv() != JOI_OK)
return HDFS_CLIENT_ERROR_HDFS_CREATE_PARAM;
-
+ setPath(path);
jstring js_path = jenv_->NewStringUTF(path);
if (js_path == NULL) {
GetCliGlobals()->setJniErrorStr(getErrorText(HDFS_CLIENT_ERROR_HDFS_CREATE_PARAM));
@@ -388,9 +499,17 @@ HDFS_Client_RetCode HdfsClient::hdfsCreate(const char* path, NABoolean compress)
}
jboolean j_compress = compress;
+ jboolean j_overwrite = overwrite;
+
+ if (hdfsStats_ != NULL)
+ hdfsStats_->getHdfsTimer().start();
tsRecentJMFromJNI = JavaMethods_[JM_HDFS_CREATE].jm_full_name;
- jboolean jresult = jenv_->CallBooleanMethod(javaObj_, JavaMethods_[JM_HDFS_CREATE].methodID, js_path, j_compress);
+ jboolean jresult = jenv_->CallBooleanMethod(javaObj_, JavaMethods_[JM_HDFS_CREATE].methodID, js_path, j_overwrite, j_compress);
+ if (hdfsStats_ != NULL) {
+ hdfsStats_->incMaxHdfsIOTime(hdfsStats_->getHdfsTimer().stop());
+ hdfsStats_->incHdfsCalls();
+ }
if (jenv_->ExceptionCheck())
{
@@ -427,9 +546,14 @@ HDFS_Client_RetCode HdfsClient::hdfsOpen(const char* path, NABoolean compress)
}
jboolean j_compress = compress;
-
+ if (hdfsStats_ != NULL)
+ hdfsStats_->getHdfsTimer().start();
tsRecentJMFromJNI = JavaMethods_[JM_HDFS_OPEN].jm_full_name;
jboolean jresult = jenv_->CallBooleanMethod(javaObj_, JavaMethods_[JM_HDFS_OPEN].methodID, js_path, j_compress);
+ if (hdfsStats_ != NULL) {
+ hdfsStats_->incMaxHdfsIOTime(hdfsStats_->getHdfsTimer().stop());
+ hdfsStats_->incHdfsCalls();
+ }
if (jenv_->ExceptionCheck())
{
@@ -452,50 +576,92 @@ HDFS_Client_RetCode HdfsClient::hdfsOpen(const char* path, NABoolean compress)
}
-HDFS_Client_RetCode HdfsClient::hdfsWrite(const char* data, Int64 len)
+Int32 HdfsClient::hdfsWrite(const char* data, Int64 len, HDFS_Client_RetCode &hdfsClientRetcode)
{
QRLogger::log(CAT_SQL_HDFS, LL_DEBUG, "HdfsClient::hdfsWrite(%ld) called.", len);
- if (initJNIEnv() != JOI_OK)
- return HDFS_CLIENT_ERROR_HDFS_WRITE_EXCEPTION;
+ if (initJNIEnv() != JOI_OK) {
+ hdfsClientRetcode = HDFS_CLIENT_ERROR_HDFS_WRITE_EXCEPTION;
+ return 0;
+ }
//Write the requisite bytes into the file
jbyteArray jbArray = jenv_->NewByteArray( len);
if (!jbArray) {
GetCliGlobals()->setJniErrorStr(getErrorText(HDFS_CLIENT_ERROR_HDFS_WRITE_PARAM));
jenv_->PopLocalFrame(NULL);
- return HDFS_CLIENT_ERROR_HDFS_WRITE_PARAM;
+ hdfsClientRetcode = HDFS_CLIENT_ERROR_HDFS_WRITE_PARAM;
+ return 0;
}
jenv_->SetByteArrayRegion(jbArray, 0, len, (const jbyte*)data);
- jlong j_len = len;
+ if (hdfsStats_ != NULL)
+ hdfsStats_->getHdfsTimer().start();
+
tsRecentJMFromJNI = JavaMethods_[JM_HDFS_WRITE].jm_full_name;
- jboolean jresult = jenv_->CallBooleanMethod(javaObj_, JavaMethods_[JM_HDFS_WRITE].methodID,jbArray , j_len);
+ // Java method returns the cumulative bytes written
+ jint totalBytesWritten = jenv_->CallIntMethod(javaObj_, JavaMethods_[JM_HDFS_WRITE].methodID, jbArray);
+ if (hdfsStats_ != NULL) {
+ hdfsStats_->incMaxHdfsIOTime(hdfsStats_->getHdfsTimer().stop());
+ hdfsStats_->incHdfsCalls();
+ }
if (jenv_->ExceptionCheck())
{
getExceptionDetails();
logError(CAT_SQL_HDFS, __FILE__, __LINE__);
logError(CAT_SQL_HDFS, "HdfsClient::hdfsWrite()", getLastError());
jenv_->PopLocalFrame(NULL);
- return HDFS_CLIENT_ERROR_HDFS_WRITE_EXCEPTION;
+ hdfsClientRetcode = HDFS_CLIENT_ERROR_HDFS_WRITE_EXCEPTION;
+ return 0;
}
- if (jresult == false)
+ jenv_->PopLocalFrame(NULL);
+ hdfsClientRetcode = HDFS_CLIENT_OK;
+ Int32 bytesWritten = totalBytesWritten - totalBytesWritten_;
+ totalBytesWritten_ = totalBytesWritten;
+ return bytesWritten;
+}
+
+Int32 HdfsClient::hdfsRead(const char* data, Int64 len, HDFS_Client_RetCode &hdfsClientRetcode)
+{
+ QRLogger::log(CAT_SQL_HDFS, LL_DEBUG, "HdfsClient::hdfsWrite(%ld) called.", len);
+
+ if (initJNIEnv() != JOI_OK) {
+ hdfsClientRetcode = HDFS_CLIENT_ERROR_HDFS_READ_EXCEPTION;
+ return 0;
+ }
+ jobject j_buf = jenv_->NewDirectByteBuffer((BYTE *)data, len);
+ if (j_buf == NULL) {
+ GetCliGlobals()->setJniErrorStr(getErrorText(HDFS_CLIENT_ERROR_HDFS_READ_PARAM));
+ jenv_->PopLocalFrame(NULL);
+ return HDFS_CLIENT_ERROR_HDFS_READ_PARAM;
+ }
+ if (hdfsStats_ != NULL)
+ hdfsStats_->getHdfsTimer().start();
+
+ tsRecentJMFromJNI = JavaMethods_[JM_HDFS_READ].jm_full_name;
+ jint bytesRead = 0;
+ bytesRead = jenv_->CallIntMethod(javaObj_, JavaMethods_[JM_HDFS_READ].methodID, j_buf);
+
+ if (hdfsStats_ != NULL) {
+ hdfsStats_->incMaxHdfsIOTime(hdfsStats_->getHdfsTimer().stop());
+ hdfsStats_->incHdfsCalls();
+ }
+ if (jenv_->ExceptionCheck())
{
- logError(CAT_SQL_HDFS, "HdfsClient::hdfsWrite()", getLastError());
+ getExceptionDetails();
+ logError(CAT_SQL_HDFS, __FILE__, __LINE__);
+ logError(CAT_SQL_HDFS, "HdfsClient::hdfsRead()", getLastError());
jenv_->PopLocalFrame(NULL);
- return HDFS_CLIENT_ERROR_HDFS_WRITE_EXCEPTION;
+ hdfsClientRetcode = HDFS_CLIENT_ERROR_HDFS_READ_EXCEPTION;
+ return 0;
}
-
-
jenv_->PopLocalFrame(NULL);
- return HDFS_CLIENT_OK;
+ hdfsClientRetcode = HDFS_CLIENT_OK;
+ return bytesRead;
}
-//////////////////////////////////////////////////////////////////////////////
-//
-//////////////////////////////////////////////////////////////////////////////
HDFS_Client_RetCode HdfsClient::hdfsClose()
{
QRLogger::log(CAT_SQL_HDFS, LL_DEBUG, "HdfsClient::close() called.");
@@ -504,9 +670,15 @@ HDFS_Client_RetCode HdfsClient::hdfsClose()
return HDFS_CLIENT_ERROR_HDFS_CLOSE_EXCEPTION;
// String close();
+ if (hdfsStats_ != NULL)
+ hdfsStats_->getHdfsTimer().start();
tsRecentJMFromJNI = JavaMethods_[JM_HDFS_CLOSE].jm_full_name;
jboolean jresult = jenv_->CallBooleanMethod(javaObj_, JavaMethods_[JM_HDFS_CLOSE].methodID);
+ if (hdfsStats_ != NULL) {
+ hdfsStats_->incMaxHdfsIOTime(hdfsStats_->getHdfsTimer().stop());
+ hdfsStats_->incHdfsCalls();
+ }
if (jenv_->ExceptionCheck())
{
getExceptionDetails();
@@ -533,6 +705,8 @@ HDFS_Client_RetCode HdfsClient::hdfsCleanUnloadPath( const NAString& uldPath)
uldPath.data());
if (initJNIEnv() != JOI_OK)
return HDFS_CLIENT_ERROR_HDFS_CLEANUP_PARAM;
+ if (getInstance() == NULL)
+ return HDFS_CLIENT_ERROR_HDFS_CLEANUP_PARAM;
jstring js_UldPath = jenv_->NewStringUTF(uldPath.data());
if (js_UldPath == NULL) {
GetCliGlobals()->setJniErrorStr(getErrorText(HDFS_CLIENT_ERROR_HDFS_CLEANUP_PARAM));
@@ -563,8 +737,9 @@ HDFS_Client_RetCode HdfsClient::hdfsMergeFiles( const NAString& srcPath,
srcPath.data(), dstPath.data());
if (initJNIEnv() != JOI_OK)
- return HDFS_CLIENT_ERROR_HDFS_MERGE_FILES_EXCEPTION;
-
+ return HDFS_CLIENT_ERROR_HDFS_MERGE_FILES_PARAM;
+ if (getInstance() == NULL)
+ return HDFS_CLIENT_ERROR_HDFS_MERGE_FILES_PARAM;
jstring js_SrcPath = jenv_->NewStringUTF(srcPath.data());
if (js_SrcPath == NULL) {
@@ -608,7 +783,9 @@ HDFS_Client_RetCode HdfsClient::hdfsDeletePath( const NAString& delPath)
QRLogger::log(CAT_SQL_HDFS, LL_DEBUG, "HdfsClient::hdfsDeletePath(%s called.",
delPath.data());
if (initJNIEnv() != JOI_OK)
- return HDFS_CLIENT_ERROR_HDFS_DELETE_PATH_EXCEPTION;
+ return HDFS_CLIENT_ERROR_HDFS_DELETE_PATH_PARAM;
+ if (getInstance() == NULL)
+ return HDFS_CLIENT_ERROR_HDFS_DELETE_PATH_PARAM;
jstring js_delPath = jenv_->NewStringUTF(delPath.data());
if (js_delPath == NULL) {
@@ -617,7 +794,6 @@ HDFS_Client_RetCode HdfsClient::hdfsDeletePath( const NAString& delPath)
return HDFS_CLIENT_ERROR_HDFS_DELETE_PATH_PARAM;
}
-
tsRecentJMFromJNI = JavaMethods_[JM_HDFS_DELETE_PATH].jm_full_name;
jboolean jresult = jenv_->CallStaticBooleanMethod(javaClass_, JavaMethods_[JM_HDFS_DELETE_PATH].methodID, js_delPath);
@@ -676,9 +852,10 @@ HDFS_Client_RetCode HdfsClient::hdfsExists( const NAString& uldPath, NABoolean &
{
QRLogger::log(CAT_SQL_HDFS, LL_DEBUG, "HdfsClient::hdfsExists(%s) called.",
uldPath.data());
-
if (initJNIEnv() != JOI_OK)
- return HDFS_CLIENT_ERROR_HDFS_EXISTS_EXCEPTION;
+ return HDFS_CLIENT_ERROR_HDFS_EXISTS_PARAM;
+ if (getInstance() == NULL)
+ return HDFS_CLIENT_ERROR_HDFS_EXISTS_PARAM;
jstring js_UldPath = jenv_->NewStringUTF(uldPath.data());
if (js_UldPath == NULL) {
@@ -701,6 +878,113 @@ HDFS_Client_RetCode HdfsClient::hdfsExists( const NAString& uldPath, NABoolean &
return HDFS_CLIENT_OK;
}
+HDFS_Client_RetCode HdfsClient::getHiveTableMaxModificationTs( Int64& maxModificationTs, const char * tableDirPaths, int levelDeep)
+{
+ QRLogger::log(CAT_SQL_HBASE, LL_DEBUG, "Enter HDFSClient_JNI::getHiveTableMaxModificationTs(%s) called.",tableDirPaths);
+ if (initJNIEnv() != JOI_OK)
+ return HDFS_CLIENT_ERROR_HIVE_TBL_MAX_MODIFICATION_TS_PARAM;
+ if (getInstance() == NULL)
+ return HDFS_CLIENT_ERROR_HIVE_TBL_MAX_MODIFICATION_TS_PARAM;
+ jstring js_tableDirPaths = jenv_->NewStringUTF(tableDirPaths);
+ if (js_tableDirPaths == NULL)
+ {
+ GetCliGlobals()->setJniErrorStr(getErrorText(HDFS_CLIENT_ERROR_HIVE_TBL_MAX_MODIFICATION_TS_PARAM));
+ jenv_->PopLocalFrame(NULL);
+ return HDFS_CLIENT_ERROR_HIVE_TBL_MAX_MODIFICATION_TS_PARAM;
+ }
+
+ jint jlevelDeep = levelDeep;
+ tsRecentJMFromJNI = JavaMethods_[JM_HIVE_TBL_MAX_MODIFICATION_TS].jm_full_name;
+ jlong jresult = jenv_->CallStaticLongMethod(javaClass_,
+ JavaMethods_[JM_HIVE_TBL_MAX_MODIFICATION_TS].methodID,
+ js_tableDirPaths, jlevelDeep);
+ jenv_->DeleteLocalRef(js_tableDirPaths);
+ if (jenv_->ExceptionCheck())
+ {
+ getExceptionDetails(jenv_);
+ logError(CAT_SQL_HBASE, __FILE__, __LINE__);
+ logError(CAT_SQL_HBASE, "HDFSClientI::getHiveTableMaxModificationTs()", getLastError());
+ jenv_->PopLocalFrame(NULL);
+ return HDFS_CLIENT_ERROR_HIVE_TBL_MAX_MODIFICATION_TS_EXCEPTION;
+ }
+ QRLogger::log(CAT_SQL_HBASE, LL_DEBUG,
+ "Exit HDFSClient_JNI::getHiveTableMaxModificationTs() called.");
+ maxModificationTs = jresult;
+ jenv_->PopLocalFrame(NULL);
+
+ return HDFS_CLIENT_OK;
+}
+
+HDFS_Client_RetCode HdfsClient::getFsDefaultName(char* buf, int buf_len)
+{
+ QRLogger::log(CAT_SQL_HBASE, LL_DEBUG, "Enter HDFSClient_JNI::getFsDefaultName() called.");
+ if (initJNIEnv() != JOI_OK)
+ return HDFS_CLIENT_ERROR_GET_FS_DEFAULT_NAME_PARAM;
+ if (getInstance() == NULL)
+ return HDFS_CLIENT_ERROR_GET_FS_DEFAULT_NAME_PARAM;
+
+ tsRecentJMFromJNI = JavaMethods_[JM_GET_FS_DEFAULT_NAME].jm_full_name;
+ jstring jresult =
+ (jstring)jenv_->CallStaticObjectMethod(javaClass_,
+ JavaMethods_[JM_GET_FS_DEFAULT_NAME].methodID);
+ if (jenv_->ExceptionCheck())
+ {
+ getExceptionDetails(jenv_);
+ logError(CAT_SQL_HBASE, __FILE__, __LINE__);
+ logError(CAT_SQL_HBASE, "HDFSClient_JNI::getFsDefaultName()", getLastError());
+ jenv_->PopLocalFrame(NULL);
+ return HDFS_CLIENT_ERROR_GET_FS_DEFAULT_NAME_EXCEPTION;
+ }
+ const char* char_result = jenv_->GetStringUTFChars(jresult, 0);
+
+ HDFS_Client_RetCode retcode = HDFS_CLIENT_OK;
+ if ( buf_len >= strlen(char_result) ) {
+ strcpy(buf, char_result);
+ } else
+ retcode = HDFS_CLIENT_ERROR_GET_FS_DEFAULT_NAME_BUFFER_TOO_SMALL;
+
+ jenv_->ReleaseStringUTFChars(jresult, char_result);
+ jenv_->PopLocalFrame(NULL);
+
+ return retcode;
+}
+
+HDFS_Client_RetCode HdfsClient::hdfsCreateDirectory(const NAString &dirName)
+{
+ QRLogger::log(CAT_SQL_HBASE, LL_DEBUG, "Enter HDFSClient_JNI::createDirectory() called.");
+ if (initJNIEnv() != JOI_OK)
+ return HDFS_CLIENT_ERROR_CREATE_DIRECTORY_PARAM;
+ if (getInstance() == NULL)
+ return HDFS_CLIENT_ERROR_CREATE_DIRECTORY_PARAM;
+
+ jstring js_dirName = jenv_->NewStringUTF(dirName.data());
+ if (js_dirName == NULL) {
+ jenv_->PopLocalFrame(NULL);
+ return HDFS_CLIENT_ERROR_CREATE_DIRECTORY_PARAM;
+ }
+
+ tsRecentJMFromJNI = JavaMethods_[JM_HDFS_CREATE_DIRECTORY].jm_full_name;
+ jstring jresult =
+ (jstring)jenv_->CallStaticObjectMethod(javaClass_,
+ JavaMethods_[JM_HDFS_CREATE_DIRECTORY].methodID, js_dirName);
+ if (jenv_->ExceptionCheck())
+ {
+ getExceptionDetails(jenv_);
+ logError(CAT_SQL_HBASE, __FILE__, __LINE__);
+ logError(CAT_SQL_HBASE, "HDFSClient_JNI::hdfsCreateDirectory()", getLastError());
+ jenv_->PopLocalFrame(NULL);
+ return HDFS_CLIENT_ERROR_CREATE_DIRECTORY_EXCEPTION;
+ }
+ if (jresult == false)
+ {
+ logError(CAT_SQL_HDFS, "HdfsClient::hdfsCreateDirectory()", getLastError());
+ jenv_->PopLocalFrame(NULL);
+ return HDFS_CLIENT_ERROR_HDFS_DELETE_PATH_EXCEPTION;
+ }
+ jenv_->PopLocalFrame(NULL);
+ return HDFS_CLIENT_OK;
+}
+
HDFS_Client_RetCode HdfsClient::setHdfsFileInfo(JNIEnv *jenv, jint numFiles, jint fileNo, jboolean isDir,
jstring filename, jlong modTime, jlong len, jshort numReplicas, jlong blockSize,
jstring owner, jstring group, jshort permissions, jlong accessTime)
http://git-wip-us.apache.org/repos/asf/trafodion/blob/e303b3a0/core/sql/executor/HdfsClient_JNI.h
----------------------------------------------------------------------
diff --git a/core/sql/executor/HdfsClient_JNI.h b/core/sql/executor/HdfsClient_JNI.h
index c45d226..6f68f4d 100644
--- a/core/sql/executor/HdfsClient_JNI.h
+++ b/core/sql/executor/HdfsClient_JNI.h
@@ -39,6 +39,8 @@ typedef enum {
,HDFS_SCAN_ERROR_TRAF_HDFS_READ_PARAM
,HDFS_SCAN_ERROR_TRAF_HDFS_READ_EXCEPTION
,HDFS_SCAN_EOR
+ ,HDFS_SCAN_ERROR_STOP_PARAM
+ ,HDFS_SCAN_ERROR_STOP_EXCEPTION
,HDFS_SCAN_LAST
} HDFS_Scan_RetCode;
@@ -48,11 +50,18 @@ public:
// Default constructor - for creating a new JVM
HdfsScan(NAHeap *heap)
: JavaObjectInterface(heap)
+ , hdfsStats_(NULL)
+ , j_buf1_(NULL)
+ , j_buf2_(NULL)
{}
+ ~HdfsScan();
+
// Initialize JVM and all the JNI configuration.
// Must be called.
HDFS_Scan_RetCode init();
+ void setHdfsStats(ExHdfsScanStats *hdfsStats)
+ { hdfsStats_ = hdfsStats; }
// Get the error description.
static char* getErrorText(HDFS_Scan_RetCode errEnum);
@@ -61,19 +70,25 @@ public:
HdfsFileInfoArray *hdfsFileInfoArray, Int32 beginRangeNum, Int32 numRanges, int rangeTailIOSize,
ExHdfsScanStats *hdfsStats, HDFS_Scan_RetCode &hdfsScanRetCode);
- HDFS_Scan_RetCode setScanRanges(NAHeap *heap, ExHdfsScanTcb::HDFS_SCAN_BUF *hdfsScanBuf, int scanBufSize,
+ HDFS_Scan_RetCode setScanRanges(ExHdfsScanTcb::HDFS_SCAN_BUF *hdfsScanBuf, int scanBufSize,
HdfsFileInfoArray *hdfsFileInfoArray, Int32 beginRangeNum, Int32 numRanges,
- int rangeTailIOSize, ExHdfsScanStats *hdfsStats);
+ int rangeTailIOSize);
+
+ HDFS_Scan_RetCode trafHdfsRead(int retArray[], short arrayLen);
- HDFS_Scan_RetCode trafHdfsRead(NAHeap *heap, ExHdfsScanStats *hdfsStats, int retArray[], short arrayLen);
+ HDFS_Scan_RetCode stop();
private:
enum JAVA_METHODS {
JM_CTOR = 0,
JM_SET_SCAN_RANGES,
JM_TRAF_HDFS_READ,
+ JM_STOP,
JM_LAST
};
+ jobject j_buf1_;
+ jobject j_buf2_;
+ ExHdfsScanStats *hdfsStats_;
static jclass javaClass_;
static JavaMethodInit* JavaMethods_;
static bool javaMethodsInitialized_;
@@ -114,6 +129,8 @@ typedef enum {
,HDFS_CLIENT_ERROR_HDFS_OPEN_EXCEPTION
,HDFS_CLIENT_ERROR_HDFS_WRITE_PARAM
,HDFS_CLIENT_ERROR_HDFS_WRITE_EXCEPTION
+ ,HDFS_CLIENT_ERROR_HDFS_READ_PARAM
+ ,HDFS_CLIENT_ERROR_HDFS_READ_EXCEPTION
,HDFS_CLIENT_ERROR_HDFS_CLOSE_EXCEPTION
,HDFS_CLIENT_ERROR_HDFS_MERGE_FILES_PARAM
,HDFS_CLIENT_ERROR_HDFS_MERGE_FILES_EXCEPTION
@@ -127,6 +144,13 @@ typedef enum {
,HDFS_CLIENT_ERROR_SET_HDFSFILEINFO
,HDFS_CLIENT_ERROR_HDFS_LIST_DIR_PARAM
,HDFS_CLIENT_ERROR_HDFS_LIST_DIR_EXCEPTION
+ ,HDFS_CLIENT_ERROR_HIVE_TBL_MAX_MODIFICATION_TS_PARAM
+ ,HDFS_CLIENT_ERROR_HIVE_TBL_MAX_MODIFICATION_TS_EXCEPTION
+ ,HDFS_CLIENT_ERROR_GET_FS_DEFAULT_NAME_PARAM
+ ,HDFS_CLIENT_ERROR_GET_FS_DEFAULT_NAME_EXCEPTION
+ ,HDFS_CLIENT_ERROR_GET_FS_DEFAULT_NAME_BUFFER_TOO_SMALL
+ ,HDFS_CLIENT_ERROR_CREATE_DIRECTORY_PARAM
+ ,HDFS_CLIENT_ERROR_CREATE_DIRECTORY_EXCEPTION
,HDFS_CLIENT_LAST
} HDFS_Client_RetCode;
@@ -136,34 +160,43 @@ public:
// Default constructor - for creating a new JVM
HdfsClient(NAHeap *heap)
: JavaObjectInterface(heap)
+ , path_(NULL)
, hdfsFileInfo_(NULL)
, numFiles_(0)
+ , totalBytesWritten_(0)
+ , hdfsStats_(NULL)
{
}
~HdfsClient();
- static HdfsClient *newInstance(NAHeap *heap, HDFS_Client_RetCode &retCode);
+ static HdfsClient *newInstance(NAHeap *heap, ExHdfsScanStats *hdfsStats, HDFS_Client_RetCode &retCode);
+ static HdfsClient *getInstance();
+ static void deleteInstance();
// Get the error description.
static char* getErrorText(HDFS_Client_RetCode errEnum);
-
- // Initialize JVM and all the JNI configuration.
- // Must be called.
+ void setHdfsStats(ExHdfsScanStats *hdfsStats)
+ { hdfsStats_ = hdfsStats; }
HDFS_Client_RetCode init();
- HDFS_Client_RetCode hdfsCreate(const char* path, NABoolean compress);
+ HDFS_Client_RetCode hdfsCreate(const char* path, NABoolean overwrite, NABoolean compress);
HDFS_Client_RetCode hdfsOpen(const char* path, NABoolean compress);
- HDFS_Client_RetCode hdfsWrite(const char* data, Int64 size);
+ Int32 hdfsWrite(const char* data, Int64 size, HDFS_Client_RetCode &hdfsClientRetcode);
+ Int32 hdfsRead(const char* data, Int64 size, HDFS_Client_RetCode &hdfsClientRetcode);
HDFS_Client_RetCode hdfsClose();
- HDFS_Client_RetCode hdfsMergeFiles(const NAString& srcPath,
- const NAString& dstPath);
- HDFS_Client_RetCode hdfsCleanUnloadPath(const NAString& uldPath );
- HDFS_Client_RetCode hdfsExists(const NAString& uldPath, NABoolean & exists );
- HDFS_Client_RetCode hdfsDeletePath(const NAString& delPath);
HDFS_Client_RetCode setHdfsFileInfo(JNIEnv *jenv, jint numFiles, jint fileNo, jboolean isDir,
jstring filename, jlong modTime, jlong len, jshort numReplicas, jlong blockSize,
jstring owner, jstring group, jshort permissions, jlong accessTime);
HDFS_Client_RetCode hdfsListDirectory(const char *pathStr, HDFS_FileInfo **hdfsFileInfo, int *numFiles);
- void deleteHdfsFileInfo();
+ static HDFS_Client_RetCode hdfsMergeFiles(const NAString& srcPath, const NAString& dstPath);
+ static HDFS_Client_RetCode hdfsCleanUnloadPath(const NAString& uldPath );
+ static HDFS_Client_RetCode hdfsExists(const NAString& uldPath, NABoolean & exists );
+ static HDFS_Client_RetCode hdfsDeletePath(const NAString& delPath);
+ static HDFS_Client_RetCode getHiveTableMaxModificationTs(Int64& maxModificationTs, const char * tableDirPaths, int levelDeep);
+ // Get the hdfs URL.
+ // buffer is the buffer pre-allocated to hold the result
+ // buf_len is the length of the buffer in bytes
+ static HDFS_Client_RetCode getFsDefaultName(char* buffer, Int32 buf_len);
+ static HDFS_Client_RetCode hdfsCreateDirectory(const NAString& path);
private:
enum JAVA_METHODS {
@@ -171,16 +204,27 @@ private:
JM_HDFS_CREATE,
JM_HDFS_OPEN,
JM_HDFS_WRITE,
+ JM_HDFS_READ,
JM_HDFS_CLOSE,
JM_HDFS_MERGE_FILES,
JM_HDFS_CLEAN_UNLOAD_PATH,
JM_HDFS_EXISTS,
JM_HDFS_DELETE_PATH,
JM_HDFS_LIST_DIRECTORY,
+ JM_HIVE_TBL_MAX_MODIFICATION_TS,
+ JM_GET_FS_DEFAULT_NAME,
+ JM_HDFS_CREATE_DIRECTORY,
JM_LAST
};
+
+ void deleteHdfsFileInfo();
+ void setPath(const char *path);
+
HDFS_FileInfo *hdfsFileInfo_;
int numFiles_;
+ char *path_;
+ Int64 totalBytesWritten_;
+ ExHdfsScanStats *hdfsStats_;
static jclass javaClass_;
static JavaMethodInit* JavaMethods_;
static bool javaMethodsInitialized_;
http://git-wip-us.apache.org/repos/asf/trafodion/blob/e303b3a0/core/sql/executor/JavaObjectInterface.cpp
----------------------------------------------------------------------
diff --git a/core/sql/executor/JavaObjectInterface.cpp b/core/sql/executor/JavaObjectInterface.cpp
index 6919866..2260715 100644
--- a/core/sql/executor/JavaObjectInterface.cpp
+++ b/core/sql/executor/JavaObjectInterface.cpp
@@ -24,6 +24,7 @@
#include "JavaObjectInterface.h"
#include "QRLogger.h"
#include "Globals.h"
+#include "Context.h"
#include "ComUser.h"
#include "LmJavaOptions.h"
#include "ex_ex.h"
@@ -546,7 +547,7 @@ NABoolean JavaObjectInterface::getExceptionDetails(JNIEnv *jenv)
if (jenv == NULL)
jenv = jenv_;
CliGlobals *cliGlobals = GetCliGlobals();
- NAString error_msg(heap_);
+ NAString error_msg(cliGlobals->currContext()->exHeap());
if (jenv == NULL)
{
error_msg = "Internal Error - Unable to obtain jenv";
http://git-wip-us.apache.org/repos/asf/trafodion/blob/e303b3a0/core/sql/executor/JavaObjectInterface.h
----------------------------------------------------------------------
diff --git a/core/sql/executor/JavaObjectInterface.h b/core/sql/executor/JavaObjectInterface.h
index b167420..d07b6f0 100644
--- a/core/sql/executor/JavaObjectInterface.h
+++ b/core/sql/executor/JavaObjectInterface.h
@@ -119,12 +119,12 @@ protected:
// Get the error description.
static char* getErrorText(JOI_RetCode errEnum);
- NAString getLastError();
+ static NAString getLastError();
// Write the description of a Java error to the log file.
- void logError(std::string &cat, const char* methodName, const char *result);
- void logError(std::string &cat, const char* methodName, jstring jresult);
- void logError(std::string &cat, const char* file, int line);
+ static void logError(std::string &cat, const char* methodName, const char *result);
+ static void logError(std::string &cat, const char* methodName, jstring jresult);
+ static void logError(std::string &cat, const char* file, int line);
static JOI_RetCode initJNIEnv();
static char* buildClassPath();
@@ -145,9 +145,9 @@ public:
}
// Pass in jenv if the thread where the object is created is different than
// the thread where exception occurred
- NABoolean getExceptionDetails(JNIEnv *jenv = NULL);
+ static NABoolean getExceptionDetails(JNIEnv *jenv = NULL);
- void appendExceptionMessages(JNIEnv *jenv, jthrowable a_exception, NAString &error_msg);
+ static void appendExceptionMessages(JNIEnv *jenv, jthrowable a_exception, NAString &error_msg);
NAHeap *getHeap() { return heap_; }
protected:
http://git-wip-us.apache.org/repos/asf/trafodion/blob/e303b3a0/core/sql/executor/OrcFileReader.h
----------------------------------------------------------------------
diff --git a/core/sql/executor/OrcFileReader.h b/core/sql/executor/OrcFileReader.h
index 925456b..536235a 100644
--- a/core/sql/executor/OrcFileReader.h
+++ b/core/sql/executor/OrcFileReader.h
@@ -90,7 +90,7 @@ public:
OFR_RetCode getRowCount(Int64& count);
- virtual char* getErrorText(OFR_RetCode errEnum);
+ static char* getErrorText(OFR_RetCode errEnum);
protected:
jstring getLastError();
http://git-wip-us.apache.org/repos/asf/trafodion/blob/e303b3a0/core/sql/executor/SequenceFileReader.h
----------------------------------------------------------------------
diff --git a/core/sql/executor/SequenceFileReader.h b/core/sql/executor/SequenceFileReader.h
index 12a68c2..a679eba 100644
--- a/core/sql/executor/SequenceFileReader.h
+++ b/core/sql/executor/SequenceFileReader.h
@@ -99,7 +99,7 @@ public:
SFR_RetCode fetchRowsIntoBuffer(Int64 stopOffset, char* buffer, Int64 buffSize, Int64& bytesRead, char rowDelimiter);
- virtual char* getErrorText(SFR_RetCode errEnum);
+ static char* getErrorText(SFR_RetCode errEnum);
protected:
jstring getLastError();
@@ -187,7 +187,7 @@ public:
SFW_RetCode close();
SFW_RetCode release();
- virtual char* getErrorText(SFW_RetCode errEnum);
+ static char* getErrorText(SFW_RetCode errEnum);
http://git-wip-us.apache.org/repos/asf/trafodion/blob/e303b3a0/core/sql/exp/ExpLOBaccess.cpp
----------------------------------------------------------------------
diff --git a/core/sql/exp/ExpLOBaccess.cpp b/core/sql/exp/ExpLOBaccess.cpp
index 481e960..4b3a785 100644
--- a/core/sql/exp/ExpLOBaccess.cpp
+++ b/core/sql/exp/ExpLOBaccess.cpp
@@ -367,143 +367,58 @@ Ex_Lob_Error ExLob::writeDataSimple(char *data, Int64 size, LobsSubOper subOpera
return LOB_OPER_OK;
}
-Ex_Lob_Error ExLob::dataModCheck2(
+// numOfPartLevels: 0, if not partitioned
+// N, number of partitioning cols
+// failedModTS: timestamp value that caused the mismatch
+Ex_Lob_Error ExLob::dataModCheck(
char * dirPath,
Int64 inputModTS,
Lng32 numOfPartLevels,
+ ExLobGlobals *lobGlobals,
Int64 &failedModTS,
char *failedLocBuf,
Int32 *failedLocBufLen)
{
- if (numOfPartLevels == 0)
+ if (inputModTS <= 0)
return LOB_OPER_OK;
- Lng32 currNumFilesInDir = 0;
- hdfsFileInfo * fileInfos =
- hdfsListDirectory(fs_, dirPath, &currNumFilesInDir);
- if ((currNumFilesInDir > 0) && (fileInfos == NULL))
- {
- return LOB_DATA_FILE_NOT_FOUND_ERROR;
- }
-
- NABoolean failed = FALSE;
- for (Lng32 i = 0; ((NOT failed) && (i < currNumFilesInDir)); i++)
- {
- hdfsFileInfo &fileInfo = fileInfos[i];
- if (fileInfo.mKind == kObjectKindDirectory)
- {
- Int64 currModTS = fileInfo.mLastMod;
- if ((inputModTS > 0) &&
- (currModTS > inputModTS) &&
- (!strstr(fileInfo.mName, ".hive-staging_hive_")))
- {
- failed = TRUE;
- failedModTS = currModTS;
+ Ex_Lob_Error result = LOB_OPER_OK;
+ HDFS_Client_RetCode rc;
+ Int64 currModTS;
- if (failedLocBuf && failedLocBufLen)
- {
- Lng32 failedFileLen = strlen(fileInfo.mName);
- Lng32 copyLen = (failedFileLen > (*failedLocBufLen-1)
- ? (*failedLocBufLen-1) : failedFileLen);
-
- str_cpy_and_null(failedLocBuf, fileInfo.mName, copyLen,
- '\0', ' ', TRUE);
- *failedLocBufLen = copyLen;
- }
- }
- }
- }
- hdfsFreeFileInfo(fileInfos, currNumFilesInDir);
- if (failed)
- return LOB_DATA_MOD_CHECK_ERROR;
+ failedModTS = -1;
- numOfPartLevels--;
- Ex_Lob_Error err = LOB_OPER_OK;
- if (numOfPartLevels > 0)
+ // libhdfs returns a second-resolution timestamp,
+ // get a millisecond-resolution timestamp via JNI
+ rc = HdfsClient::getHiveTableMaxModificationTs(currModTS,
+ dirPath,
+ numOfPartLevels);
+ // check for errors and timestamp mismatches
+ if (rc != HDFS_CLIENT_OK || currModTS <= 0)
{
- for (Lng32 i = 0; ((NOT failed) && (i < currNumFilesInDir)); i++)
- {
- hdfsFileInfo &fileInfo = fileInfos[i];
- err = dataModCheck2(fileInfo.mName, inputModTS, numOfPartLevels,
- failedModTS, failedLocBuf, failedLocBufLen);
- if (err != LOB_OPER_OK)
- return err;
- }
+ result = LOB_DATA_READ_ERROR;
+ }
+ else if (currModTS > inputModTS)
+ {
+ result = LOB_DATA_MOD_CHECK_ERROR;
+ failedModTS = currModTS;
}
- return LOB_OPER_OK;
-}
-
-// numOfPartLevels: 0, if not partitioned
-// N, number of partitioning cols
-// failedModTS: timestamp value that caused the mismatch
-Ex_Lob_Error ExLob::dataModCheck(
- char * dirPath,
- Int64 inputModTS,
- Lng32 numOfPartLevels,
- ExLobGlobals *lobGlobals,
- Int64 &failedModTS,
- char *failedLocBuf,
- Int32 *failedLocBufLen)
-{
- failedModTS = -1;
-
- // find mod time of root dir
- hdfsFileInfo *fileInfos = hdfsGetPathInfo(fs_, dirPath);
- if (fileInfos == NULL)
- {
+ if (result != LOB_OPER_OK && failedLocBuf && failedLocBufLen)
+ {
+ // sorry, we lost the exact location for partitioned
+ // files, user needs to search for him/herself
Lng32 failedFileLen = strlen(dirPath);
Lng32 copyLen = (failedFileLen > (*failedLocBufLen-1)
? (*failedLocBufLen-1) : failedFileLen);
- Int32 hdfserror = errno;
- char hdfsErrStr[20];
- snprintf(hdfsErrStr,sizeof(hdfsErrStr),"(errno %d)",errno);
+
str_cpy_and_null(failedLocBuf, dirPath, copyLen,
'\0', ' ', TRUE);
- str_cat_c(failedLocBuf,hdfsErrStr);
*failedLocBufLen = copyLen;
- if (errno)
- {
- // Allow for hdfs error. AQR will find the new hive mapped files
- // if the hive table has been remapped to new data files
- return LOB_DATA_MOD_CHECK_ERROR;
- }
- else
- return LOB_DATA_READ_ERROR;
}
-
- Int64 currModTS = fileInfos[0].mLastMod;
- if ((inputModTS > 0) &&
- (currModTS > inputModTS))
- {
- hdfsFileInfo &fileInfo = fileInfos[0];
- failedModTS = currModTS;
-
- if (failedLocBuf && failedLocBufLen)
- {
- Lng32 failedFileLen = strlen(fileInfo.mName);
- Lng32 copyLen = (failedFileLen > (*failedLocBufLen-1)
- ? (*failedLocBufLen-1) : failedFileLen);
-
- str_cpy_and_null(failedLocBuf, fileInfo.mName, copyLen,
- '\0', ' ', TRUE);
- *failedLocBufLen = copyLen;
- }
-
- hdfsFreeFileInfo(fileInfos, 1);
- return LOB_DATA_MOD_CHECK_ERROR;
- }
-
- hdfsFreeFileInfo(fileInfos, 1);
- if (numOfPartLevels > 0)
- {
- return dataModCheck2(dirPath, inputModTS, numOfPartLevels,
- failedModTS, failedLocBuf, failedLocBufLen);
- }
-
- return LOB_OPER_OK;
+ return result;
}
Ex_Lob_Error ExLob::emptyDirectory(char *dirPath,
http://git-wip-us.apache.org/repos/asf/trafodion/blob/e303b3a0/core/sql/exp/ExpLOBaccess.h
----------------------------------------------------------------------
diff --git a/core/sql/exp/ExpLOBaccess.h b/core/sql/exp/ExpLOBaccess.h
index 47f06df..3c5e93e 100644
--- a/core/sql/exp/ExpLOBaccess.h
+++ b/core/sql/exp/ExpLOBaccess.h
@@ -484,14 +484,6 @@ class ExLob : public NABasicObject
char *failedLocBuf,
Int32 *failedLocBufLen);
- Ex_Lob_Error dataModCheck2(
- char * dirPath,
- Int64 modTS,
- Lng32 numOfPartLevels,
- Int64 &failedModTS,
- char *failedLocBuf,
- Int32 *failedLocBufLen);
-
Ex_Lob_Error emptyDirectory(char* dirPath, ExLobGlobals* lobGlobals);
ExLobStats *getStats() { return &stats_; }
http://git-wip-us.apache.org/repos/asf/trafodion/blob/e303b3a0/core/sql/generator/GenFastTransport.cpp
----------------------------------------------------------------------
diff --git a/core/sql/generator/GenFastTransport.cpp b/core/sql/generator/GenFastTransport.cpp
index bc788d9..75c1e0e 100644
--- a/core/sql/generator/GenFastTransport.cpp
+++ b/core/sql/generator/GenFastTransport.cpp
@@ -646,7 +646,7 @@ PhysicalFastExtract::codeGen(Generator *generator)
(char*)getHiveTableName().data(),
TRUE, // isHive
(char*)getTargetName().data(), // root dir
- hTabStats->getModificationTS(),
+ hTabStats->getModificationTSmsec(),
0,
NULL,
(char*)getHdfsHostName().data(),
@@ -657,7 +657,7 @@ PhysicalFastExtract::codeGen(Generator *generator)
else
{
// sim check at leaf
- modTS = hTabStats->getModificationTS();
+ modTS = hTabStats->getModificationTSmsec();
}
} // do sim check
http://git-wip-us.apache.org/repos/asf/trafodion/blob/e303b3a0/core/sql/generator/GenRelScan.cpp
----------------------------------------------------------------------
diff --git a/core/sql/generator/GenRelScan.cpp b/core/sql/generator/GenRelScan.cpp
index 3d01223..df4b303 100644
--- a/core/sql/generator/GenRelScan.cpp
+++ b/core/sql/generator/GenRelScan.cpp
@@ -393,7 +393,8 @@ short FileScan::genForTextAndSeq(Generator * generator,
char* &hdfsHostName,
Int32 &hdfsPort,
NABoolean &useCursorMulti,
- NABoolean &doSplitFileOpt)
+ NABoolean &doSplitFileOpt,
+ NABoolean &isCompressedFile)
{
Space * space = generator->getSpace();
@@ -480,6 +481,9 @@ short FileScan::genForTextAndSeq(Generator * generator,
hfi.bytesToRead_ = span;
hfi.fileName_ = fnameInList;
+ isCompressedFile = FALSE;
+ //if (file->getCompressionInfo().getCompressionMethod() != ComCompressionInfo::UNCOMPRESSED)
+ // isCompressedFile = TRUE;
char * hfiInList = space->allocateAndCopyToAlignedSpace
((char*)&hfi, sizeof(HdfsFileInfo));
@@ -1144,13 +1148,14 @@ short FileScan::codeGenForHive(Generator * generator)
}
NABoolean useCursorMulti = FALSE;
NABoolean doSplitFileOpt = FALSE;
+ NABoolean isCompressedFile = FALSE;
if ((hTabStats->isTextFile()) || (hTabStats->isSequenceFile()))
{
genForTextAndSeq(generator,
hdfsFileInfoList, hdfsFileRangeBeginList, hdfsFileRangeNumList,
hdfsHostName, hdfsPort,
- useCursorMulti, doSplitFileOpt);
+ useCursorMulti, doSplitFileOpt, isCompressedFile);
}
else if (hTabStats->isOrcFile())
{
@@ -1277,7 +1282,7 @@ if (hTabStats->isOrcFile())
(hTabStats->numOfPartCols() <= 0) &&
(!getCommonSubExpr()))
{
- modTS = hTabStats->getModificationTS();
+ modTS = hTabStats->getModificationTSmsec();
numOfPartLevels = hTabStats->numOfPartCols();
// if specific directories are to checked based on the query struct
@@ -1296,7 +1301,7 @@ if (hTabStats->isOrcFile())
tiName,
TRUE, // isHive
(char*)hTabStats->tableDir().data(), // root dir
- hTabStats->getModificationTS(),
+ modTS,
numOfPartLevels,
hdfsDirsToCheck,
hdfsHostName, hdfsPort);
@@ -1310,8 +1315,6 @@ if (hTabStats->isOrcFile())
space->allocateAndCopyToAlignedSpace(hTabStats->tableDir().data(),
hTabStats->tableDir().length(),
0);
- modTS = hTabStats->getModificationTS();
- numOfPartLevels = hTabStats->numOfPartCols();
}
}
@@ -1394,6 +1397,8 @@ if (hTabStats->isOrcFile())
if (CmpCommon::getDefault(USE_LIBHDFS_SCAN) == DF_ON)
hdfsscan_tdb->setUseLibhdfsScan(TRUE);
+ hdfsscan_tdb->setCompressedFile(isCompressedFile);
+
if(!generator->explainDisabled()) {
generator->setExplainTuple(
addExplainInfo(hdfsscan_tdb, 0, 0, generator));
[4/4] trafodion git commit: Merge PR 1461 TRA[TRAFODION-2917]
Refactor Trafodion implementation of hdfs scan for text formatted hive tables
Posted by se...@apache.org.
Merge PR 1461 TRA[TRAFODION-2917] Refactor Trafodion implementation of hdfs scan for text formatted hive tables
Project: http://git-wip-us.apache.org/repos/asf/trafodion/repo
Commit: http://git-wip-us.apache.org/repos/asf/trafodion/commit/e618aaf3
Tree: http://git-wip-us.apache.org/repos/asf/trafodion/tree/e618aaf3
Diff: http://git-wip-us.apache.org/repos/asf/trafodion/diff/e618aaf3
Branch: refs/heads/master
Commit: e618aaf3dfd011739654d1a62e321539ace4da00
Parents: d48a887 e303b3a
Author: selvaganesang <se...@apache.org>
Authored: Fri Mar 2 06:56:07 2018 +0000
Committer: selvaganesang <se...@apache.org>
Committed: Fri Mar 2 06:56:07 2018 +0000
----------------------------------------------------------------------
core/sql/cli/Context.cpp | 2 +-
core/sql/cli/Context.h | 7 +
core/sql/cli/Globals.cpp | 1 +
core/sql/cli/Globals.h | 1 +
core/sql/comexe/ComTdbBlockingHdfsScan.h | 1 -
core/sql/comexe/ComTdbHdfsScan.h | 11 +-
core/sql/common/ComRtUtils.cpp | 1 +
core/sql/common/ComUser.cpp | 2 +-
core/sql/common/Platform.h | 2 +
core/sql/executor/ExExeUtil.h | 1 -
core/sql/executor/ExExeUtilGet.cpp | 7 +-
core/sql/executor/ExExeUtilGetStats.cpp | 1 -
core/sql/executor/ExExeUtilLoad.cpp | 25 +-
core/sql/executor/ExFastTransport.cpp | 48 ++-
core/sql/executor/ExHbaseAccess.cpp | 11 +-
core/sql/executor/ExHbaseIUD.cpp | 2 +-
core/sql/executor/ExHdfsScan.cpp | 102 +++--
core/sql/executor/ExHdfsScan.h | 24 ++
core/sql/executor/ExSMCommon.cpp | 1 +
core/sql/executor/ExSMCommon.h | 3 +-
core/sql/executor/ExSMGlobals.cpp | 1 +
core/sql/executor/ExSMQueue.cpp | 2 +-
core/sql/executor/ExSMReader.cpp | 1 +
core/sql/executor/ExSMShortMessage.cpp | 2 +-
core/sql/executor/ExSMTaskList.cpp | 2 +-
core/sql/executor/HBaseClient_JNI.cpp | 1 -
core/sql/executor/HdfsClient_JNI.cpp | 426 ++++++++++++++++---
core/sql/executor/HdfsClient_JNI.h | 74 +++-
core/sql/executor/JavaObjectInterface.cpp | 3 +-
core/sql/executor/JavaObjectInterface.h | 12 +-
core/sql/executor/OrcFileReader.h | 2 +-
core/sql/executor/SequenceFileReader.h | 4 +-
core/sql/exp/ExpLOBaccess.cpp | 144 ++-----
core/sql/exp/ExpLOBaccess.h | 8 -
core/sql/generator/GenFastTransport.cpp | 4 +-
core/sql/generator/GenRelScan.cpp | 17 +-
core/sql/nskgmake/tdm_sqlexp/Makefile | 2 +-
core/sql/optimizer/HDFSHook.cpp | 35 +-
core/sql/optimizer/HDFSHook.h | 16 +-
core/sql/optimizer/NATable.cpp | 1 +
core/sql/optimizer/OptimizerSimulator.cpp | 1 -
core/sql/optimizer/RelScan.h | 3 +-
core/sql/regress/hive/EXPECTED007 | 2 +-
core/sql/regress/hive/EXPECTED018 | 2 +-
core/sql/regress/hive/EXPECTED040 | 8 +-
core/sql/sqlcomp/DefaultConstants.h | 1 +
.../main/java/org/trafodion/sql/HDFSClient.java | 307 ++++++++++---
.../main/java/org/trafodion/sql/HdfsScan.java | 95 +++--
48 files changed, 1007 insertions(+), 422 deletions(-)
----------------------------------------------------------------------