From 5d4e0a927e801d78d97e36938d6bf2261f8e9769 Mon Sep 17 00:00:00 2001 From: selvaganesang Date: Thu, 27 Aug 2015 00:51:58 +0000 Subject: [PATCH 1/2] [TRAFODION-1453] Improvements in Hive Scan The null byte is now skipped over to look for the column and record delimiter while scanning the hdfs row buffer. The null byte will be part of the column value. The behavior of null byte in the column value is unknown. Skipping over null byte is now in sync with other tool's behavior. Also, improved the code to scan the buffer for both record and column delimiter once. Earlier, the buffer was scanned twice; once for the record delimiter and another for column delimiter. No errors are returned when null byte is found. A sql error 7002 is now reported when a unspported row format type is accessed from Trafodion. *** ERROR[7002] The input row format of the hive table is not yet supported. --- core/sql/bin/SqlciErrors.txt | 2 +- core/sql/executor/ExHdfsScan.cpp | 110 +++++++++++++++++------------- core/sql/executor/ExHdfsScan.h | 44 ++++++++++++ core/sql/generator/GenRelScan.cpp | 6 +- 4 files changed, 112 insertions(+), 50 deletions(-) diff --git a/core/sql/bin/SqlciErrors.txt b/core/sql/bin/SqlciErrors.txt index 3df75da8b8..5dac5a5d53 100644 --- a/core/sql/bin/SqlciErrors.txt +++ b/core/sql/bin/SqlciErrors.txt @@ -1447,7 +1447,7 @@ $1~String1 -------------------------------- 6021 ZZZZZ 99999 BEGINNER MAJOR DBADMIN Compiler Optimizer Warning $0~String0. 7000 ZZZZZ 99999 ADVANCED MAJOR DIALOUT An internal error occurred in the code generator in file $0~string0 at line $1~int0: $2~string1. 7001 ZZZZZ 99999 BEGINNER MAJOR DBADMIN Default value $0~String0 is not valid for column $1~String1. -7002 ZZZZZ 99999 BEGINNER MAJOR DBADMIN --- available --- +7002 ZZZZZ 99999 BEGINNER MAJOR DBADMIN The input row format of the hive table is not yet supported. 7003 42000 99999 BEGINNER MAJOR DBADMIN A plan using cluster sampling could not be produced for this query. 7004 ZZZZZ 99999 BEGINNER MAJOR DBADMIN A parallel extract plan could not be produced. Possible causes include an incompatible Control Query Shape (CQS) specification, use of rowset expressions, or use of SQL features that cannot be parallelized such as [FIRST/LAST N], table-valued functions, stream access to tables, and embedded updates or deletes. 7005 ZZZZZ 99999 BEGINNER MAJOR DBADMIN Use of selection predicates in parallel extract consumer queries is not allowed. diff --git a/core/sql/executor/ExHdfsScan.cpp b/core/sql/executor/ExHdfsScan.cpp index e8b57ecdf6..d2eac33211 100644 --- a/core/sql/executor/ExHdfsScan.cpp +++ b/core/sql/executor/ExHdfsScan.cpp @@ -45,6 +45,7 @@ #include "ExpORCinterface.h" + ex_tcb * ExHdfsScanTdb::build(ex_globals * glob) { ExExeStmtGlobals * exe_glob = glob->castToExExeStmtGlobals(); @@ -716,8 +717,8 @@ ExWorkProcRetcode ExHdfsScanTcb::work() { // Position in the hdfsScanBuffer_ to the // first record delimiter. - hdfsBufNextRow_ = strchr(hdfsScanBuffer_, - hdfsScanTdb().recordDelimiter_); + hdfsBufNextRow_ = hdfs_strchr(hdfsScanBuffer_, + hdfsScanTdb().recordDelimiter_, hdfsScanBuffer_+trailingPrevRead_+ bytesRead_); // May be that the record is too long? Or data isn't ascii? // Or delimiter is incorrect. if (! hdfsBufNextRow_) @@ -1101,7 +1102,7 @@ ExWorkProcRetcode ExHdfsScanTcb::work() // extractSourceFields method to keep from processing // bytes left in the buffer from the previous read. if ((trailingPrevRead_ > 0) && - (hdfsBufNextRow_[0] == '\0')) + (hdfsBufNextRow_[0] == RANGE_DELIMITER)) { step_ = CLOSE_HDFS_CURSOR; break; @@ -1111,7 +1112,10 @@ ExWorkProcRetcode ExHdfsScanTcb::work() step_ = GET_HDFS_DATA; } else + { + trailingPrevRead_ = 0; step_ = CLOSE_HDFS_CURSOR; + } break; } case CLOSE_HDFS_CURSOR: @@ -1374,46 +1378,39 @@ char * ExHdfsScanTcb::extractAndTransformAsciiSourceToSqlRow(int &err, ComDiagsArea* &diagsArea) { err = 0; - char *sourceData = hdfsBufNextRow_; - char *sourceDataEnd = strchr(sourceData, - hdfsScanTdb().recordDelimiter_); - hdfsLoggingRowEnd_ = sourceDataEnd; - hdfsLoggingRow_ = hdfsBufNextRow_; + char *sourceRowEnd = NULL; char *sourceColEnd = NULL; - - if (!sourceDataEnd) - { - return NULL; - - } - else if (sourceDataEnd >= - (hdfsScanBuffer_ + trailingPrevRead_ + bytesRead_)) - { - return NULL; - } - NABoolean isTrailingMissingColumn = FALSE; - if ((endOfRequestedRange_) && - (sourceDataEnd >= endOfRequestedRange_)) - *(sourceDataEnd +1)= '\0'; - ExpTupleDesc * asciiSourceTD = hdfsScanTdb().workCriDesc_->getTupleDescriptor(hdfsScanTdb().asciiTuppIndex_); + + const char cd = hdfsScanTdb().columnDelimiter_; + const char rd = hdfsScanTdb().recordDelimiter_; + const char *sourceDataEnd = hdfsScanBuffer_+trailingPrevRead_+ bytesRead_; + + hdfsLoggingRow_ = hdfsBufNextRow_; if (asciiSourceTD->numAttrs() == 0) { + sourceRowEnd = hdfs_strchr(sourceData, rd, sourceDataEnd); + hdfsLoggingRowEnd_ = sourceRowEnd; + + if (!sourceRowEnd) + return NULL; + if ((endOfRequestedRange_) && + (sourceRowEnd >= endOfRequestedRange_)) + *(sourceRowEnd +1)= RANGE_DELIMITER; + // no columns need to be converted. For e.g. count(*) with no predicate - return sourceDataEnd+1; + return sourceRowEnd+1; } - const char delimiter = hdfsScanTdb().columnDelimiter_; - Lng32 neededColIndex = 0; Attributes * attr = NULL; + NABoolean rdSeen = FALSE; for (Lng32 i = 0; i < hdfsScanTdb().convertSkipListSize_; i++) { - // we have scanned for all needed columns from this row // all remainin columns wil be skip columns, don't bother // finding their column delimiters if (neededColIndex == asciiSourceTD->numAttrs()) @@ -1427,24 +1424,28 @@ char * ExHdfsScanTcb::extractAndTransformAsciiSourceToSqlRow(int &err, else attr = NULL; - if (!isTrailingMissingColumn) - sourceColEnd = strchr(sourceData, delimiter); - - if(!isTrailingMissingColumn) - { - short len = 0; - if (sourceColEnd && - sourceColEnd <= sourceDataEnd) - len = sourceColEnd - sourceData; - else - { - len = sourceDataEnd - sourceData; - if (i != hdfsScanTdb().convertSkipListSize_ - 1) - isTrailingMissingColumn = TRUE; - } - - if (attr) // this is a needed column. We need to convert - { + if (!isTrailingMissingColumn) { + sourceColEnd = hdfs_strchr(sourceData, rd, cd, sourceDataEnd, &rdSeen); + if (sourceColEnd == NULL) { + if (rdSeen || (sourceRowEnd == NULL)) + return NULL; + else + return sourceRowEnd+1; + } + short len = 0; + len = sourceColEnd - sourceData; + if (rdSeen) { + sourceRowEnd = sourceColEnd; + hdfsLoggingRowEnd_ = sourceRowEnd; + if ((endOfRequestedRange_) && + (sourceRowEnd >= endOfRequestedRange_)) + *(sourceRowEnd +1)= RANGE_DELIMITER; + if (i != hdfsScanTdb().convertSkipListSize_ - 1) + isTrailingMissingColumn = TRUE; + } + + if (attr) // this is a needed column. We need to convert + { *(short*)&hdfsAsciiSourceData_[attr->getVCLenIndOffset()] = len; if (attr->getNullFlag()) { @@ -1481,6 +1482,18 @@ char * ExHdfsScanTcb::extractAndTransformAsciiSourceToSqlRow(int &err, } sourceData = sourceColEnd + 1 ; } + // It is possible that the above loop came out before + // rowDelimiter is encountered + // So try to find the record delimiter + if (sourceRowEnd == NULL) { + sourceRowEnd = hdfs_strchr(sourceData, rd, sourceDataEnd); + if (sourceRowEnd) { + hdfsLoggingRowEnd_ = sourceRowEnd; + if ((endOfRequestedRange_) && + (sourceRowEnd >= endOfRequestedRange_)) + *(sourceRowEnd +1)= RANGE_DELIMITER; + } + } workAtp_->getTupp(hdfsScanTdb().workAtpIndex_) = hdfsSqlTupp_; workAtp_->getTupp(hdfsScanTdb().asciiTuppIndex_) = hdfsAsciiSourceTupp_; @@ -1496,8 +1509,9 @@ char * ExHdfsScanTcb::extractAndTransformAsciiSourceToSqlRow(int &err, else err = 0; } - - return sourceDataEnd+1; + if (sourceRowEnd) + return sourceRowEnd+1; + return NULL; } short ExHdfsScanTcb::moveRowToUpQueue(const char * row, Lng32 len, diff --git a/core/sql/executor/ExHdfsScan.h b/core/sql/executor/ExHdfsScan.h index 79efd2aeeb..62f41f72f0 100644 --- a/core/sql/executor/ExHdfsScan.h +++ b/core/sql/executor/ExHdfsScan.h @@ -426,4 +426,48 @@ class ExOrcFastAggrTcb : public ExOrcScanTcb char * aggrRow_; }; +#define RANGE_DELIMITER '\002' + +inline char *hdfs_strchr(const char *s, int c, const char *end) +{ + char *curr = (char *)s; + + while (curr < end) { + if (*curr == c) + return curr; + if (*curr == RANGE_DELIMITER) + return NULL; + curr++; + } + return NULL; +} + + +inline char *hdfs_strchr(const char *s, int rd, int cd, const char *end, NABoolean *rdSeen) +{ + char *curr = (char *)s; + + while (curr < end) { + if (*curr == rd) { + *rdSeen = TRUE; + return curr; + } + else + if (*curr == cd) { + *rdSeen = FALSE; + return curr; + } + else + if (*curr == RANGE_DELIMITER) { + *rdSeen = TRUE; + return NULL; + } + curr++; + } + *rdSeen = FALSE; + return NULL; +} + + + #endif diff --git a/core/sql/generator/GenRelScan.cpp b/core/sql/generator/GenRelScan.cpp index 260a4bd908..4a018e228b 100644 --- a/core/sql/generator/GenRelScan.cpp +++ b/core/sql/generator/GenRelScan.cpp @@ -1072,7 +1072,11 @@ short FileScan::codeGenForHive(Generator * generator) type = (short)ComTdbHdfsScan::SEQUENCE_; else if (hTabStats->isOrcFile()) type = (short)ComTdbHdfsScan::ORC_; - + else { + *CmpCommon::diags() << DgSqlCode(-7002); + GenExit(); + return -1; + } ULng32 buffersize = getDefault(GEN_DPSO_BUFFER_SIZE); queue_index upqueuelength = (queue_index)getDefault(GEN_DPSO_SIZE_UP); queue_index downqueuelength = (queue_index)getDefault(GEN_DPSO_SIZE_DOWN); From d547e1ec5a8e69011990eec09f0e829c843e2e76 Mon Sep 17 00:00:00 2001 From: selvaganesang Date: Thu, 27 Aug 2015 21:43:16 +0000 Subject: [PATCH 2/2] [TRAFODION-1453] Changes to avoid scan for a range delimiter in the buffer when not needed. It is now ok to encounter the range delimiter in the the column value, but its behavior is unknown. --- core/sql/executor/ExHdfsScan.cpp | 29 +++++++++++++++++++---------- core/sql/executor/ExHdfsScan.h | 9 +++++---- 2 files changed, 24 insertions(+), 14 deletions(-) diff --git a/core/sql/executor/ExHdfsScan.cpp b/core/sql/executor/ExHdfsScan.cpp index d2eac33211..71c0118501 100644 --- a/core/sql/executor/ExHdfsScan.cpp +++ b/core/sql/executor/ExHdfsScan.cpp @@ -114,6 +114,7 @@ ExHdfsScanTcb::ExHdfsScanTcb( , hdfo_(NULL) , numBytesProcessedInRange_(0) , exception_(FALSE) + , checkRangeDelimiter_(FALSE) { Space * space = (glob ? glob->getSpace() : 0); CollHeap * heap = (glob ? glob->getDefaultHeap() : 0); @@ -388,6 +389,7 @@ ExWorkProcRetcode ExHdfsScanTcb::work() beginRangeNum_ = -1; numRanges_ = -1; hdfsOffset_ = 0; + checkRangeDelimiter_ = FALSE; if (hdfsScanTdb().getHdfsFileInfoList()->isEmpty()) { @@ -718,7 +720,7 @@ ExWorkProcRetcode ExHdfsScanTcb::work() // Position in the hdfsScanBuffer_ to the // first record delimiter. hdfsBufNextRow_ = hdfs_strchr(hdfsScanBuffer_, - hdfsScanTdb().recordDelimiter_, hdfsScanBuffer_+trailingPrevRead_+ bytesRead_); + hdfsScanTdb().recordDelimiter_, hdfsScanBuffer_+trailingPrevRead_+ bytesRead_, checkRangeDelimiter_); // May be that the record is too long? Or data isn't ascii? // Or delimiter is incorrect. if (! hdfsBufNextRow_) @@ -1104,6 +1106,7 @@ ExWorkProcRetcode ExHdfsScanTcb::work() if ((trailingPrevRead_ > 0) && (hdfsBufNextRow_[0] == RANGE_DELIMITER)) { + checkRangeDelimiter_ = FALSE; step_ = CLOSE_HDFS_CURSOR; break; } @@ -1392,14 +1395,16 @@ char * ExHdfsScanTcb::extractAndTransformAsciiSourceToSqlRow(int &err, hdfsLoggingRow_ = hdfsBufNextRow_; if (asciiSourceTD->numAttrs() == 0) { - sourceRowEnd = hdfs_strchr(sourceData, rd, sourceDataEnd); + sourceRowEnd = hdfs_strchr(sourceData, rd, sourceDataEnd, checkRangeDelimiter_); hdfsLoggingRowEnd_ = sourceRowEnd; if (!sourceRowEnd) return NULL; if ((endOfRequestedRange_) && - (sourceRowEnd >= endOfRequestedRange_)) + (sourceRowEnd >= endOfRequestedRange_)) { + checkRangeDelimiter_ = TRUE; *(sourceRowEnd +1)= RANGE_DELIMITER; + } // no columns need to be converted. For e.g. count(*) with no predicate return sourceRowEnd+1; @@ -1425,7 +1430,7 @@ char * ExHdfsScanTcb::extractAndTransformAsciiSourceToSqlRow(int &err, attr = NULL; if (!isTrailingMissingColumn) { - sourceColEnd = hdfs_strchr(sourceData, rd, cd, sourceDataEnd, &rdSeen); + sourceColEnd = hdfs_strchr(sourceData, rd, cd, sourceDataEnd, checkRangeDelimiter_, &rdSeen); if (sourceColEnd == NULL) { if (rdSeen || (sourceRowEnd == NULL)) return NULL; @@ -1438,8 +1443,10 @@ char * ExHdfsScanTcb::extractAndTransformAsciiSourceToSqlRow(int &err, sourceRowEnd = sourceColEnd; hdfsLoggingRowEnd_ = sourceRowEnd; if ((endOfRequestedRange_) && - (sourceRowEnd >= endOfRequestedRange_)) - *(sourceRowEnd +1)= RANGE_DELIMITER; + (sourceRowEnd >= endOfRequestedRange_)) { + checkRangeDelimiter_ = TRUE; + *(sourceRowEnd +1)= RANGE_DELIMITER; + } if (i != hdfsScanTdb().convertSkipListSize_ - 1) isTrailingMissingColumn = TRUE; } @@ -1486,13 +1493,15 @@ char * ExHdfsScanTcb::extractAndTransformAsciiSourceToSqlRow(int &err, // rowDelimiter is encountered // So try to find the record delimiter if (sourceRowEnd == NULL) { - sourceRowEnd = hdfs_strchr(sourceData, rd, sourceDataEnd); + sourceRowEnd = hdfs_strchr(sourceData, rd, sourceDataEnd, checkRangeDelimiter_); if (sourceRowEnd) { hdfsLoggingRowEnd_ = sourceRowEnd; if ((endOfRequestedRange_) && - (sourceRowEnd >= endOfRequestedRange_)) - *(sourceRowEnd +1)= RANGE_DELIMITER; - } + (sourceRowEnd >= endOfRequestedRange_ )) { + checkRangeDelimiter_ = TRUE; + *(sourceRowEnd +1)= RANGE_DELIMITER; + } + } } workAtp_->getTupp(hdfsScanTdb().workAtpIndex_) = hdfsSqlTupp_; diff --git a/core/sql/executor/ExHdfsScan.h b/core/sql/executor/ExHdfsScan.h index 62f41f72f0..8ac16d32a4 100644 --- a/core/sql/executor/ExHdfsScan.h +++ b/core/sql/executor/ExHdfsScan.h @@ -282,6 +282,7 @@ class ExHdfsScanTcb : public ex_tcb NABoolean exception_; ComCondition * lastErrorCnd_; + NABoolean checkRangeDelimiter_; }; class ExOrcScanTcb : public ExHdfsScanTcb @@ -428,14 +429,14 @@ class ExOrcFastAggrTcb : public ExOrcScanTcb #define RANGE_DELIMITER '\002' -inline char *hdfs_strchr(const char *s, int c, const char *end) +inline char *hdfs_strchr(const char *s, int c, const char *end, NABoolean checkRangeDelimiter) { char *curr = (char *)s; while (curr < end) { if (*curr == c) return curr; - if (*curr == RANGE_DELIMITER) + if (checkRangeDelimiter &&*curr == RANGE_DELIMITER) return NULL; curr++; } @@ -443,7 +444,7 @@ inline char *hdfs_strchr(const char *s, int c, const char *end) } -inline char *hdfs_strchr(const char *s, int rd, int cd, const char *end, NABoolean *rdSeen) +inline char *hdfs_strchr(const char *s, int rd, int cd, const char *end, NABoolean checkRangeDelimiter, NABoolean *rdSeen) { char *curr = (char *)s; @@ -458,7 +459,7 @@ inline char *hdfs_strchr(const char *s, int rd, int cd, const char *end, NABoole return curr; } else - if (*curr == RANGE_DELIMITER) { + if (checkRangeDelimiter && *curr == RANGE_DELIMITER) { *rdSeen = TRUE; return NULL; }