xxq250
/
homework-jianmu
1
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity

Merge branch 'develop' into feature/crash_gen

This commit is contained in:
Steven Li 2020-05-10 23:05:44 -07:00
parent b52b452972 fead5b9038
commit ebcbafeef6
43 changed files with 571 additions and 565 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

10
.travis.yml
View File

@ -59,12 +59,12 @@ matrix:
GREEN_UNDERLINE='\033[4;32m'
NC='\033[0m'
grep 'ERROR SUMMARY' mem-error-out.txt | uniq | tee uniq-mem-error-out.txt
grep 'ERROR SUMMARY' mem-error-out.txt|grep -v 'grep'|uniq|tee uniq-mem-error-out.txt
for memError in `cat uniq-mem-error-out.txt | awk '{print $4}'`
do
if [ -n "$memError" ]; then
if [ "$memError" -gt 16 ]; then
if [ "$memError" -gt 5 ]; then
echo -e "${RED} ## Memory errors number valgrind reports is $memError.\
More than our threshold! ## ${NC}"
travis_terminate $memError
@ -72,11 +72,11 @@ matrix:
fi
done
grep 'definitely lost:' mem-error-out.txt | uniq | tee uniq-definitely-lost-out.txt
for defiMemError in `cat uniq-definitely-lost-out.txt | awk '{print $4}'`
grep 'definitely lost:' mem-error-out.txt|grep -v 'grep'|uniq|tee uniq-definitely-lost-out.txt
for defiMemError in `cat uniq-definitely-lost-out.txt | awk '{print $7}'`
do
if [ -n "$defiMemError" ]; then
if [ "$defiMemError" -gt 16 ]; then
if [ "$defiMemError" -gt 3 ]; then
echo -e "${RED} ## Memory errors number valgrind reports \
Definitely lost is $defiMemError. More than our threshold! ## ${NC}"
travis_terminate $defiMemError

2
README.md
View File

@ -121,7 +121,7 @@ TDengine provides abundant developing tools for users to develop on TDengine. Fo
1. sudo apt install
build-essential cmake net-tools python-pip python-setuptools python3-pip
python3-setuptools valgrind
python3-setuptools valgrind psmisc curl
2. git clone <https://github.com/taosdata/TDengine>; cd TDengine

16
src/client/inc/tscLog.h
View File

@ -22,25 +22,25 @@ extern "C" {
#include "tlog.h"
extern int32_t cdebugFlag;
extern int32_t cDebugFlag;
#define tscError(...) \
if (cdebugFlag & DEBUG_ERROR) { \
if (cDebugFlag & DEBUG_ERROR) { \
taosPrintLog("ERROR TSC ", 255, __VA_ARGS__); \
}
#define tscWarn(...) \
if (cdebugFlag & DEBUG_WARN) { \
taosPrintLog("WARN TSC ", cdebugFlag, __VA_ARGS__); \
if (cDebugFlag & DEBUG_WARN) { \
taosPrintLog("WARN TSC ", cDebugFlag, __VA_ARGS__); \
}
#define tscTrace(...) \
if (cdebugFlag & DEBUG_TRACE) { \
taosPrintLog("TSC ", cdebugFlag, __VA_ARGS__); \
if (cDebugFlag & DEBUG_TRACE) { \
taosPrintLog("TSC ", cDebugFlag, __VA_ARGS__); \
}
#define tscPrint(...) \
{ taosPrintLog("TSC ", 255, __VA_ARGS__); }
#define tscDump(...) \
if (cdebugFlag & DEBUG_TRACE) { \
taosPrintLongString("TSC ", cdebugFlag, __VA_ARGS__); \
if (cDebugFlag & DEBUG_TRACE) { \
taosPrintLongString("TSC ", cDebugFlag, __VA_ARGS__); \
}
#ifdef __cplusplus

12
src/client/src/TSDBJNIConnector.c
View File

@ -23,16 +23,16 @@
#include "ttime.h"
#define jniError(...) \
if (jnidebugFlag & DEBUG_ERROR) { \
taosPrintLog("ERROR JNI ", jnidebugFlag, __VA_ARGS__); \
if (jniDebugFlag & DEBUG_ERROR) { \
taosPrintLog("ERROR JNI ", jniDebugFlag, __VA_ARGS__); \
}
#define jniWarn(...) \
if (jnidebugFlag & DEBUG_WARN) { \
taosPrintLog("WARN JNI ", jnidebugFlag, __VA_ARGS__); \
if (jniDebugFlag & DEBUG_WARN) { \
taosPrintLog("WARN JNI ", jniDebugFlag, __VA_ARGS__); \
}
#define jniTrace(...) \
if (jnidebugFlag & DEBUG_TRACE) { \
taosPrintLog("JNI ", jnidebugFlag, __VA_ARGS__); \
if (jniDebugFlag & DEBUG_TRACE) { \
taosPrintLog("JNI ", jniDebugFlag, __VA_ARGS__); \
}
#define jniPrint(...) \
{ taosPrintLog("JNI ", 255, __VA_ARGS__); }

14
src/client/src/tscFunctionImpl.c
View File

@ -1691,10 +1691,7 @@ static void last_function(SQLFunctionCtx *pCtx) {
}
static void last_function_f(SQLFunctionCtx *pCtx, int32_t index) {
if (pCtx->order == TSDB_ORDER_ASC) {
return;
}
assert(pCtx->order != TSDB_ORDER_ASC);
void *pData = GET_INPUT_CHAR_INDEX(pCtx, index);
if (pCtx->hasNull && isNull(pData, pCtx->inputType)) {
return;
@ -2912,7 +2909,7 @@ static void leastsquares_finalizer(SQLFunctionCtx *pCtx) {
}
static void date_col_output_function(SQLFunctionCtx *pCtx) {
if (pCtx->scanFlag == SUPPLEMENTARY_SCAN) {
if (pCtx->scanFlag == REVERSE_SCAN) {
return;
}
@ -2969,11 +2966,12 @@ static void tag_project_function(SQLFunctionCtx *pCtx) {
char* output = pCtx->aOutputBuf;
if (pCtx->tag.nType == TSDB_DATA_TYPE_BINARY || pCtx->tag.nType == TSDB_DATA_TYPE_NCHAR) {
*(int16_t*) output = pCtx->tag.nLen;
output += VARSTR_HEADER_SIZE;
varDataSetLen(output, pCtx->tag.nLen);
tVariantDump(&pCtx->tag, varDataVal(output), pCtx->outputType);
} else {
tVariantDump(&pCtx->tag, output, pCtx->outputType);
}
tVariantDump(&pCtx->tag, output, pCtx->outputType);
pCtx->aOutputBuf += pCtx->outputBytes;
}
}

2
src/client/src/tscSQLParser.c
View File

@ -5911,7 +5911,7 @@ int32_t exprTreeFromSqlExpr(tExprNode **pExpr, const tSQLExpr* pSqlExpr, SArray*
}
if (pSqlExpr->pLeft == NULL) {
if (pSqlExpr->nSQLOptr >= TK_TINYINT && pSqlExpr->nSQLOptr <= TK_DOUBLE) {
if (pSqlExpr->nSQLOptr >= TK_BOOL && pSqlExpr->nSQLOptr <= TK_STRING) {
*pExpr = calloc(1, sizeof(tExprNode));
(*pExpr)->nodeType = TSQL_NODE_VALUE;
(*pExpr)->pVal = calloc(1, sizeof(tVariant));

12
src/client/src/tscServer.c
View File

@ -492,13 +492,15 @@ int tscBuildFetchMsg(SSqlObj *pSql, SSqlInfo *pInfo) {
pMsg += sizeof(pQueryInfo->type);
// todo valid the vgroupId at the client side
if (UTIL_TABLE_IS_SUPERTABLE(pQueryInfo->pTableMetaInfo[0])) {
SVgroupsInfo* pVgroupInfo = pQueryInfo->pTableMetaInfo[0]->vgroupList;
assert(pVgroupInfo->numOfVgroups == 1); // todo fix me
STableMetaInfo* pTableMetaInfo = tscGetMetaInfo(pQueryInfo, 0);
pRetrieveMsg->header.vgId = htonl(pVgroupInfo->vgroups[0].vgId);
if (UTIL_TABLE_IS_SUPERTABLE(pTableMetaInfo)) {
int32_t vgIndex = pTableMetaInfo->vgroupIndex;
SVgroupsInfo* pVgroupInfo = pTableMetaInfo->vgroupList;
pRetrieveMsg->header.vgId = htonl(pVgroupInfo->vgroups[vgIndex].vgId);
} else {
STableMeta* pTableMeta = pQueryInfo->pTableMetaInfo[0]->pTableMeta;
STableMeta* pTableMeta = pTableMetaInfo->pTableMeta;
pRetrieveMsg->header.vgId = htonl(pTableMeta->vgroupInfo.vgId);
}

6
src/client/src/tscSubquery.c
View File

@ -1436,8 +1436,8 @@ void tscHandleSubqueryError(SRetrieveSupport *trsupport, SSqlObj *pSql, int numO
return;
} else { // reach the maximum retry count, abort
atomic_val_compare_exchange_32(&pState->code, TSDB_CODE_SUCCESS, numOfRows);
tscError("%p sub:%p retrieve failed,code:%d,orderOfSub:%d failed.no more retry,set global code:%d", pPObj, pSql,
numOfRows, subqueryIndex, tstrerror(pState->code));
tscError("%p sub:%p retrieve failed,code:%s,orderOfSub:%d failed.no more retry,set global code:%s", pPObj, pSql,
tstrerror(numOfRows), subqueryIndex, tstrerror(pState->code));
}
}
@ -1450,7 +1450,7 @@ void tscHandleSubqueryError(SRetrieveSupport *trsupport, SSqlObj *pSql, int numO
}
// all subqueries are failed
tscError("%p retrieve from %d vnode(s) completed,code:%d.FAILED.", pPObj, pState->numOfTotal, pState->code);
tscError("%p retrieve from %d vnode(s) completed,code:%s.FAILED.", pPObj, pState->numOfTotal, tstrerror(pState->code));
pPObj->res.code = pState->code;
// release allocated resource

3
src/client/src/tscUtil.c
View File

@ -1675,10 +1675,7 @@ STableMetaInfo* tscAddTableMetaInfo(SQueryInfo* pQueryInfo, const char* name, ST
pTableMetaInfo->pTableMeta = pTableMeta;
if (vgroupList != NULL) {
assert(vgroupList->numOfVgroups == 1); // todo fix me
size_t size = sizeof(SVgroupsInfo) + sizeof(SCMVgroupInfo) * vgroupList->numOfVgroups;
pTableMetaInfo->vgroupList = malloc(size);
memcpy(pTableMetaInfo->vgroupList, vgroupList, size);
}

2
src/common/inc/tdataformat.h
View File

@ -115,10 +115,8 @@ static FORCE_INLINE void *tdGetRowDataOfCol(SDataRow row, int8_t type, int32_t o
case TSDB_DATA_TYPE_BINARY:
case TSDB_DATA_TYPE_NCHAR:
return POINTER_SHIFT(row, *(VarDataOffsetT *)POINTER_SHIFT(row, offset));
break;
default:
return POINTER_SHIFT(row, offset);
break;
}
}

10
src/common/inc/tglobal.h
View File

@ -141,9 +141,9 @@ extern int32_t tsAsyncLog;
extern int32_t tsNumOfLogLines;
extern int32_t dDebugFlag;
extern int32_t vDebugFlag;
extern int32_t mdebugFlag;
extern int32_t cdebugFlag;
extern int32_t jnidebugFlag;
extern int32_t mDebugFlag;
extern int32_t cDebugFlag;
extern int32_t jniDebugFlag;
extern int32_t tmrDebugFlag;
extern int32_t sdbDebugFlag;
extern int32_t httpDebugFlag;
@ -151,8 +151,8 @@ extern int32_t monitorDebugFlag;
extern int32_t uDebugFlag;
extern int32_t rpcDebugFlag;
extern int32_t debugFlag;
extern int32_t odbcdebugFlag;
extern int32_t qdebugFlag;
extern int32_t odbcDebugFlag;
extern int32_t qDebugFlag;
extern uint32_t taosMaxTmrCtrl;

30
src/common/src/tglobal.c
View File

@ -127,16 +127,16 @@ int32_t tsRestRowLimit = 10240;
int32_t tsMaxSQLStringLen = TSDB_MAX_SQL_LEN;
int32_t tsNumOfLogLines = 10000000;
int32_t mdebugFlag = 135;
int32_t mDebugFlag = 135;
int32_t sdbDebugFlag = 135;
int32_t dDebugFlag = 135;
int32_t vDebugFlag = 135;
int32_t cdebugFlag = 135;
int32_t jnidebugFlag = 131;
int32_t odbcdebugFlag = 131;
int32_t cDebugFlag = 135;
int32_t jniDebugFlag = 131;
int32_t odbcDebugFlag = 131;
int32_t httpDebugFlag = 131;
int32_t monitorDebugFlag = 131;
int32_t qdebugFlag = 131;
int32_t qDebugFlag = 131;
int32_t rpcDebugFlag = 135;
int32_t uDebugFlag = 131;
int32_t debugFlag = 131;
@ -204,19 +204,19 @@ static pthread_once_t tsInitGlobalCfgOnce = PTHREAD_ONCE_INIT;
void taosSetAllDebugFlag() {
for (int32_t i = 0; i < tsGlobalConfigNum; ++i) {
mdebugFlag = debugFlag;
mDebugFlag = debugFlag;
sdbDebugFlag = debugFlag;
dDebugFlag = debugFlag;
vDebugFlag = debugFlag;
cdebugFlag = debugFlag;
jnidebugFlag = debugFlag;
odbcdebugFlag = debugFlag;
cDebugFlag = debugFlag;
jniDebugFlag = debugFlag;
odbcDebugFlag = debugFlag;
httpDebugFlag = debugFlag;
monitorDebugFlag = debugFlag;
rpcDebugFlag = debugFlag;
uDebugFlag = debugFlag;
sDebugFlag = debugFlag;
//qdebugFlag = debugFlag;
//qDebugFlag = debugFlag;
}
uPrint("all debug flag are set to %d", debugFlag);
}
@ -1002,7 +1002,7 @@ static void doInitGlobalConfig() {
taosInitConfigOption(cfg);
cfg.option = "mDebugFlag";
cfg.ptr = &mdebugFlag;
cfg.ptr = &mDebugFlag;
cfg.valType = TAOS_CFG_VTYPE_INT32;
cfg.cfgType = TSDB_CFG_CTYPE_B_CONFIG | TSDB_CFG_CTYPE_B_LOG;
cfg.minValue = 0;
@ -1062,7 +1062,7 @@ static void doInitGlobalConfig() {
taosInitConfigOption(cfg);
cfg.option = "cDebugFlag";
cfg.ptr = &cdebugFlag;
cfg.ptr = &cDebugFlag;
cfg.valType = TAOS_CFG_VTYPE_INT32;
cfg.cfgType = TSDB_CFG_CTYPE_B_CONFIG | TSDB_CFG_CTYPE_B_LOG | TSDB_CFG_CTYPE_B_CLIENT;
cfg.minValue = 0;
@ -1072,7 +1072,7 @@ static void doInitGlobalConfig() {
taosInitConfigOption(cfg);
cfg.option = "jniDebugFlag";
cfg.ptr = &jnidebugFlag;
cfg.ptr = &jniDebugFlag;
cfg.valType = TAOS_CFG_VTYPE_INT32;
cfg.cfgType = TSDB_CFG_CTYPE_B_CONFIG | TSDB_CFG_CTYPE_B_LOG | TSDB_CFG_CTYPE_B_CLIENT;
cfg.minValue = 0;
@ -1082,7 +1082,7 @@ static void doInitGlobalConfig() {
taosInitConfigOption(cfg);
cfg.option = "odbcDebugFlag";
cfg.ptr = &odbcdebugFlag;
cfg.ptr = &odbcDebugFlag;
cfg.valType = TAOS_CFG_VTYPE_INT32;
cfg.cfgType = TSDB_CFG_CTYPE_B_CONFIG | TSDB_CFG_CTYPE_B_LOG | TSDB_CFG_CTYPE_B_CLIENT;
cfg.minValue = 0;
@ -1122,7 +1122,7 @@ static void doInitGlobalConfig() {
taosInitConfigOption(cfg);
cfg.option = "qDebugFlag";
cfg.ptr = &qdebugFlag;
cfg.ptr = &qDebugFlag;
cfg.valType = TAOS_CFG_VTYPE_INT32;
cfg.cfgType = TSDB_CFG_CTYPE_B_CONFIG | TSDB_CFG_CTYPE_B_LOG | TSDB_CFG_CTYPE_B_CLIENT;
cfg.minValue = 0;

8
src/dnode/src/dnodeMain.c
View File

@ -167,9 +167,9 @@ static int32_t dnodeInitSystem() {
if (dnodeInitRead() != 0) return -1;
if (dnodeInitWrite() != 0) return -1;
if (dnodeInitClient() != 0) return -1;
if (dnodeInitModules() != 0) return -1;
if (dnodeInitServer() != 0) return -1;
if (dnodeInitMgmt() != 0) return -1;
if (dnodeInitModules() != 0) return -1;
if (dnodeInitShell() != 0) return -1;
dnodeStartModules();
@ -184,14 +184,14 @@ static void dnodeCleanUpSystem() {
if (dnodeGetRunStatus() != TSDB_DNODE_RUN_STATUS_STOPPED) {
dnodeSetRunStatus(TSDB_DNODE_RUN_STATUS_STOPPED);
dnodeCleanupShell();
dnodeCleanupServer();
dnodeCleanUpModules();
dnodeCleanupMgmt();
dnodeCleanupServer();
dnodeCleanupClient();
dnodeCleanupWrite();
dnodeCleanupRead();
dnodeCleanUpModules();
taos_cleanup();
dnodeCleanupStorage();
taos_cleanup();
taosCloseLog();
}
}

5
src/inc/taosdef.h
View File

@ -36,6 +36,11 @@ extern "C" {
typedef int32_t VarDataOffsetT;
typedef int16_t VarDataLenT;
typedef struct tstr {
VarDataLenT len;
char data[];
} tstr;
#define VARSTR_HEADER_SIZE sizeof(VarDataLenT)
#define varDataLen(v) ((VarDataLenT *)(v))[0]

2
src/inc/tsdb.h
View File

@ -102,7 +102,7 @@ int tsdbTableSetSName(STableCfg *config, char *sname, bool dup);
void tsdbClearTableCfg(STableCfg *config);
int32_t tsdbGetTableTagVal(TsdbRepoT *repo, STableId* id, int32_t colId, int16_t *type, int16_t *bytes, char **val);
int32_t tsdbGetTableName(TsdbRepoT *repo, STableId* id, char** name);
char* tsdbGetTableName(TsdbRepoT *repo, const STableId* id, int16_t* bytes);
int tsdbCreateTable(TsdbRepoT *repo, STableCfg *pCfg);
int tsdbDropTable(TsdbRepoT *pRepo, STableId tableId);

8
src/mnode/inc/mgmtInt.h
View File

@ -23,13 +23,13 @@ extern "C" {
#include "tlog.h"
#include "monitor.h"
extern int32_t mdebugFlag;
extern int32_t mDebugFlag;
extern int32_t sdbDebugFlag;
// mnode log function
#define mError(...) if (mdebugFlag & DEBUG_ERROR) { taosPrintLog("ERROR MND ", 255, __VA_ARGS__); }
#define mWarn(...) if (mdebugFlag & DEBUG_WARN) { taosPrintLog("WARN MND ", mdebugFlag, __VA_ARGS__); }
#define mTrace(...) if (mdebugFlag & DEBUG_TRACE) { taosPrintLog("MND ", mdebugFlag, __VA_ARGS__); }
#define mError(...) if (mDebugFlag & DEBUG_ERROR) { taosPrintLog("ERROR MND ", 255, __VA_ARGS__); }
#define mWarn(...) if (mDebugFlag & DEBUG_WARN) { taosPrintLog("WARN MND ", mDebugFlag, __VA_ARGS__); }
#define mTrace(...) if (mDebugFlag & DEBUG_TRACE) { taosPrintLog("MND ", mDebugFlag, __VA_ARGS__); }
#define mPrint(...) { taosPrintLog("MND ", 255, __VA_ARGS__); }
#define mLError(...) monitorSaveLog(2, __VA_ARGS__); mError(__VA_ARGS__)

4
src/mnode/src/mgmtServer.c
View File

@ -88,10 +88,12 @@ void mgmtProcessReqMsgFromDnode(SRpcMsg *rpcMsg) {
SRpcIpSet ipSet = {0};
dnodeGetMnodeDnodeIpSet(&ipSet);
for (int i = 0; i < ipSet.numOfIps; ++i)
ipSet.port[i] = htons(ipSet.port[i]);
mTrace("conn from dnode ip:%s user:%s redirect msg, inUse:%d", taosIpStr(connInfo.clientIp), connInfo.user, ipSet.inUse);
for (int32_t i = 0; i < ipSet.numOfIps; ++i) {
mTrace("index:%d %s:%d", i, ipSet.fqdn[i], ipSet.port[i]);
mTrace("mnode index:%d %s:%d", i, ipSet.fqdn[i], htons(ipSet.port[i]));
}
rpcSendRedirectRsp(rpcMsg->handle, &ipSet);
return;

2
src/mnode/src/mgmtShell.c
View File

@ -137,7 +137,7 @@ void mgmtProcessMsgFromShell(SRpcMsg *rpcMsg) {
mgmtGetMnodeIpSet(&ipSet);
mTrace("conn from shell ip:%s user:%s redirect msg, inUse:%d", taosIpStr(connInfo.clientIp), connInfo.user, ipSet.inUse);
for (int32_t i = 0; i < ipSet.numOfIps; ++i) {
mTrace("index:%d ip:%s:%d", i, ipSet.fqdn[i], ipSet.port[i]);
mTrace("mnode index:%d ip:%s:%d", i, ipSet.fqdn[i], htons(ipSet.port[i]));
}
rpcSendRedirectRsp(rpcMsg->handle, &ipSet);

14
src/query/inc/queryExecutor.h
View File

@ -110,6 +110,14 @@ typedef struct STableQueryInfo { // todo merge with the STableQueryInfo struct
SWindowResInfo windowResInfo;
} STableQueryInfo;
typedef struct SQueryCostSummary {
} SQueryCostSummary;
typedef struct SGroupItem {
STableId id;
STableQueryInfo* info;
} SGroupItem;
typedef struct SQuery {
int16_t numOfCols;
int16_t numOfTags;
@ -131,17 +139,15 @@ typedef struct SQuery {
SColumnInfo* tagColList;
int32_t numOfFilterCols;
int64_t* defaultVal;
TSKEY lastKey;
// TSKEY lastKey;
uint32_t status; // query status
SResultRec rec;
int32_t pos;
SData** sdata;
STableQueryInfo* current;
SSingleColumnFilterInfo* pFilterInfo;
} SQuery;
typedef struct SQueryCostSummary {
} SQueryCostSummary;
typedef struct SQueryRuntimeEnv {
SResultInfo* resultInfo; // todo refactor to merge with SWindowResInfo
SQuery* pQuery;

14
src/query/inc/queryLog.h
View File

@ -22,20 +22,20 @@ extern "C" {
#include "tlog.h"
extern int32_t qdebugFlag;
extern int32_t qDebugFlag;
#define qTrace(...) \
if (qdebugFlag & DEBUG_TRACE) { \
taosPrintLog("DND QRY ", qdebugFlag, __VA_ARGS__); \
if (qDebugFlag & DEBUG_TRACE) { \
taosPrintLog("DND QRY ", qDebugFlag, __VA_ARGS__); \
}
#define qError(...) \
if (qdebugFlag & DEBUG_ERROR) { \
taosPrintLog("ERROR QRY ", qdebugFlag, __VA_ARGS__); \
if (qDebugFlag & DEBUG_ERROR) { \
taosPrintLog("ERROR QRY ", qDebugFlag, __VA_ARGS__); \
}
#define qWarn(...) \
if (qdebugFlag & DEBUG_WARN) { \
taosPrintLog("WARN QRY ", qdebugFlag, __VA_ARGS__); \
if (qDebugFlag & DEBUG_WARN) { \
taosPrintLog("WARN QRY ", qDebugFlag, __VA_ARGS__); \
}
#ifdef __cplusplus

2
src/query/inc/tsqlfunction.h
View File

@ -104,7 +104,7 @@ extern "C" {
enum {
MASTER_SCAN = 0x0u,
SUPPLEMENTARY_SCAN = 0x1u,
REVERSE_SCAN = 0x1u,
REPEAT_SCAN = 0x2u, //repeat scan belongs to the master scan
FIRST_STAGE_MERGE = 0x10u,
SECONDARY_STAGE_MERGE = 0x20u,

43
src/query/src/qast.c
View File

@ -552,18 +552,18 @@ static void tQueryIndexColumn(SSkipList* pSkipList, tQueryInfo* pQueryInfo, SArr
if (cond.start != NULL) {
int32_t optr = cond.start->optr;
if (optr == TSDB_RELATION_EQUAL) {
if (optr == TSDB_RELATION_EQUAL) { // equals
while(tSkipListIterNext(iter)) {
SSkipListNode* pNode = tSkipListIterGet(iter);
int32_t ret = pQueryInfo->compare(SL_GET_NODE_KEY(pSkipList, pNode), cond.start->v);
if (ret == 0) {
taosArrayPush(result, SL_GET_NODE_DATA(pNode));
} else {
if (ret != 0) {
break;
}
taosArrayPush(result, SL_GET_NODE_DATA(pNode));
}
} else if (optr == TSDB_RELATION_GREATER || optr == TSDB_RELATION_GREATER_EQUAL) {
} else if (optr == TSDB_RELATION_GREATER || optr == TSDB_RELATION_GREATER_EQUAL) { // greater equal
bool comp = true;
int32_t ret = 0;
@ -582,8 +582,33 @@ static void tQueryIndexColumn(SSkipList* pSkipList, tQueryInfo* pQueryInfo, SArr
comp = false;
}
}
} else if (optr == TSDB_RELATION_NOT_EQUAL) {
assert(0);
} else if (optr == TSDB_RELATION_NOT_EQUAL) { // not equal
bool comp = true;
while(tSkipListIterNext(iter)) {
SSkipListNode* pNode = tSkipListIterGet(iter);
comp = comp && (pQueryInfo->compare(SL_GET_NODE_KEY(pSkipList, pNode), cond.start->v) == 0);
if (comp) {
continue;
}
taosArrayPush(result, SL_GET_NODE_DATA(pNode));
}
tSkipListDestroyIter(iter);
comp = true;
iter = tSkipListCreateIterFromVal(pSkipList, (char*) cond.start->v, pSkipList->keyInfo.type, TSDB_ORDER_DESC);
while(tSkipListIterNext(iter)) {
SSkipListNode* pNode = tSkipListIterGet(iter);
comp = comp && (pQueryInfo->compare(SL_GET_NODE_KEY(pSkipList, pNode), cond.start->v) == 0);
if (comp) {
continue;
}
taosArrayPush(result, SL_GET_NODE_DATA(pNode));
}
} else {
assert(0);
}
@ -773,8 +798,6 @@ static void tSQLBinaryTraverseOnSkipList(tExprNode *pExpr, SArray *pResult, SSki
tSkipListDestroyIter(iter);
}
static void tQueryIndexlessColumn(SSkipList* pSkipList, tQueryInfo* pQueryInfo, SArray* result) {
SSkipListIterator* iter = tSkipListCreateIter(pSkipList);
@ -785,7 +808,7 @@ static void tQueryIndexlessColumn(SSkipList* pSkipList, tQueryInfo* pQueryInfo,
char* pTable = SL_GET_NODE_DATA(pNode);
//todo refactor:
char* name = (*(STable**) pTable)->name;
tstr* name = ((STableIndexElem*) pTable)->pTable->name;
// char* name = NULL;
// tsdbGetTableName(tsdb, pTable, &name);

401
src/query/src/queryExecutor.c
View File

@ -40,9 +40,9 @@
#define QUERY_IS_ASC_QUERY(q) (GET_FORWARD_DIRECTION_FACTOR((q)->order.order) == QUERY_ASC_FORWARD_STEP)
#define IS_MASTER_SCAN(runtime) ((runtime)->scanFlag == MASTER_SCAN)
#define IS_REVERSE_SCAN(runtime) ((runtime)->scanFlag == SUPPLEMENTARY_SCAN)
#define IS_REVERSE_SCAN(runtime) ((runtime)->scanFlag == REVERSE_SCAN)
#define SET_MASTER_SCAN_FLAG(runtime) ((runtime)->scanFlag = MASTER_SCAN)
#define SET_REVERSE_SCAN_FLAG(runtime) ((runtime)->scanFlag = SUPPLEMENTARY_SCAN)
#define SET_REVERSE_SCAN_FLAG(runtime) ((runtime)->scanFlag = REVERSE_SCAN)
#define GET_QINFO_ADDR(x) ((void *)((char *)(x)-offsetof(SQInfo, runtimeEnv)))
@ -96,11 +96,6 @@ typedef struct {
STSCursor cur;
} SQueryStatusInfo;
typedef struct SGroupItem {
STableId id;
STableQueryInfo* info;
} SGroupItem;
static void setQueryStatus(SQuery *pQuery, int8_t status);
static bool isIntervalQuery(SQuery *pQuery) { return pQuery->intervalTime > 0; }
@ -121,7 +116,7 @@ static bool hasMainOutput(SQuery *pQuery);
static void createTableQueryInfo(SQInfo *pQInfo);
static void buildTagQueryResult(SQInfo *pQInfo);
static int32_t setAdditionalInfo(SQInfo *pQInfo, STableId *pTaleId, STableQueryInfo *pTableQueryInfo);
static int32_t setAdditionalInfo(SQInfo *pQInfo, STableId *pTableId, STableQueryInfo *pTableQueryInfo);
static int32_t flushFromResultBuf(SQInfo *pQInfo);
bool getNeighborPoints(SQInfo *pQInfo, void *pMeterObj, SPointInterpoSupporter *pPointInterpSupporter) {
@ -428,9 +423,10 @@ static bool hasNullValue(SQuery *pQuery, int32_t col, SDataBlockInfo *pDataBlock
return false;
}
*pColStatis = NULL;
if (pStatis != NULL) {
*pColStatis = getStatisInfo(pQuery, pStatis, pDataBlockInfo, col);
} else {
*pColStatis = NULL;
}
if ((*pColStatis) != NULL && (*pColStatis)->numOfNull == 0) {
@ -620,8 +616,8 @@ static void doCheckQueryCompleted(SQueryRuntimeEnv *pRuntimeEnv, TSKEY lastKey,
}
// query completed
if ((lastKey >= pQuery->window.ekey && QUERY_IS_ASC_QUERY(pQuery)) ||
(lastKey <= pQuery->window.ekey && !QUERY_IS_ASC_QUERY(pQuery))) {
if ((lastKey >= pQuery->current->win.ekey && QUERY_IS_ASC_QUERY(pQuery)) ||
(lastKey <= pQuery->current->win.ekey && !QUERY_IS_ASC_QUERY(pQuery))) {
closeAllTimeWindow(pWindowResInfo);
pWindowResInfo->curIndex = pWindowResInfo->size - 1;
@ -661,22 +657,22 @@ static void doCheckQueryCompleted(SQueryRuntimeEnv *pRuntimeEnv, TSKEY lastKey,
setQueryStatus(pQuery, QUERY_RESBUF_FULL);
}
qTrace("QInfo:%p total window:%d, closed:%d", GET_QINFO_ADDR(pQuery), pWindowResInfo->size, n);
qTrace("QInfo:%p total window:%d, closed:%d", GET_QINFO_ADDR(pRuntimeEnv), pWindowResInfo->size, n);
}
assert(pWindowResInfo->prevSKey != 0);
}
static int32_t getNumOfRowsInTimeWindow(SQuery *pQuery, SDataBlockInfo *pDataBlockInfo, TSKEY *pPrimaryColumn,
int32_t startPos, TSKEY ekey, __block_search_fn_t searchFn,
bool updateLastKey) {
int32_t startPos, TSKEY ekey, __block_search_fn_t searchFn, bool updateLastKey) {
assert(startPos >= 0 && startPos < pDataBlockInfo->rows);
int32_t num = -1;
int32_t order = pQuery->order.order;
int32_t step = GET_FORWARD_DIRECTION_FACTOR(order);
STableQueryInfo* item = pQuery->current;
if (QUERY_IS_ASC_QUERY(pQuery)) {
if (ekey < pDataBlockInfo->window.ekey) {
num = getForwardStepsInBlock(pDataBlockInfo->rows, searchFn, ekey, startPos, order, pPrimaryColumn);
@ -684,13 +680,13 @@ static int32_t getNumOfRowsInTimeWindow(SQuery *pQuery, SDataBlockInfo *pDataBlo
assert(ekey < pPrimaryColumn[startPos]);
} else {
if (updateLastKey) {
pQuery->lastKey = pPrimaryColumn[startPos + (num - 1)] + step;
item->lastKey = pPrimaryColumn[startPos + (num - 1)] + step;
}
}
} else {
num = pDataBlockInfo->rows - startPos;
if (updateLastKey) {
pQuery->lastKey = pDataBlockInfo->window.ekey + step;
item->lastKey = pDataBlockInfo->window.ekey + step;
}
}
} else { // desc
@ -700,13 +696,13 @@ static int32_t getNumOfRowsInTimeWindow(SQuery *pQuery, SDataBlockInfo *pDataBlo
assert(ekey > pPrimaryColumn[startPos]);
} else {
if (updateLastKey) {
pQuery->lastKey = pPrimaryColumn[startPos - (num - 1)] + step;
item->lastKey = pPrimaryColumn[startPos - (num - 1)] + step;
}
}
} else {
num = startPos + 1;
if (updateLastKey) {
pQuery->lastKey = pDataBlockInfo->window.skey + step;
item->lastKey = pDataBlockInfo->window.skey + step;
}
}
}
@ -820,7 +816,21 @@ static TSKEY reviseWindowEkey(SQuery *pQuery, STimeWindow *pWindow) {
return ekey;
}
char *getDataBlocks(SQueryRuntimeEnv *pRuntimeEnv, SArithmeticSupport *sas, int32_t col, int32_t size,
//todo binary search
static void* getDataBlockImpl(SArray* pDataBlock, int32_t colId) {
int32_t numOfCols = taosArrayGetSize(pDataBlock);
for (int32_t i = 0; i < numOfCols; ++i) {
SColumnInfoData *p = taosArrayGet(pDataBlock, i);
if (colId == p->info.colId) {
return p->pData;
}
}
return NULL;
}
static char *getDataBlock(SQueryRuntimeEnv *pRuntimeEnv, SArithmeticSupport *sas, int32_t col, int32_t size,
SArray *pDataBlock) {
char *dataBlock = NULL;
SQuery *pQuery = pRuntimeEnv->pQuery;
@ -867,20 +877,7 @@ char *getDataBlocks(SQueryRuntimeEnv *pRuntimeEnv, SArithmeticSupport *sas, int3
if (TSDB_COL_IS_TAG(pCol->flag) || pDataBlock == NULL) {
dataBlock = NULL;
} else {
/*
* the colIndex is acquired from the first meter of all qualified meters in this vnode during query prepare
* stage, the remain meter may not have the required column in cache actually. So, the validation of required
* column in cache with the corresponding meter schema is reinforced.
*/
int32_t numOfCols = taosArrayGetSize(pDataBlock);
for (int32_t i = 0; i < numOfCols; ++i) {
SColumnInfoData *p = taosArrayGet(pDataBlock, i);
if (pCol->colId == p->info.colId) {
dataBlock = p->pData;
break;
}
}
dataBlock = getDataBlockImpl(pDataBlock, pCol->colId);
}
}
@ -904,8 +901,8 @@ static void blockwiseApplyFunctions(SQueryRuntimeEnv *pRuntimeEnv, SDataStatis *
SQuery * pQuery = pRuntimeEnv->pQuery;
SColumnInfoData *pColInfo = NULL;
TSKEY * primaryKeyCol = NULL;
TSKEY *primaryKeyCol = NULL;
if (pDataBlock != NULL) {
pColInfo = taosArrayGet(pDataBlock, 0);
primaryKeyCol = (TSKEY *)(pColInfo->pData);
@ -919,7 +916,7 @@ static void blockwiseApplyFunctions(SQueryRuntimeEnv *pRuntimeEnv, SDataStatis *
SDataStatis *tpField = NULL;
bool hasNull = hasNullValue(pQuery, k, pDataBlockInfo, pStatis, &tpField);
char *dataBlock = getDataBlocks(pRuntimeEnv, &sasArray[k], k, pDataBlockInfo->rows, pDataBlock);
char *dataBlock = getDataBlock(pRuntimeEnv, &sasArray[k], k, pDataBlockInfo->rows, pDataBlock);
setExecParams(pQuery, &pCtx[k], dataBlock, primaryKeyCol, pDataBlockInfo->rows, functionId, tpField, hasNull,
&sasArray[k], pRuntimeEnv->scanFlag);
@ -1095,15 +1092,20 @@ static int32_t doTSJoinFilter(SQueryRuntimeEnv *pRuntimeEnv, int32_t offset) {
static bool functionNeedToExecute(SQueryRuntimeEnv *pRuntimeEnv, SQLFunctionCtx *pCtx, int32_t functionId) {
SResultInfo *pResInfo = GET_RES_INFO(pCtx);
SQuery* pQuery = pRuntimeEnv->pQuery;
if (pResInfo->complete || functionId == TSDB_FUNC_TAG_DUMMY || functionId == TSDB_FUNC_TS_DUMMY) {
return false;
}
if (functionId == TSDB_FUNC_LAST_DST || functionId == TSDB_FUNC_LAST) {
return !QUERY_IS_ASC_QUERY(pQuery);
} else if (functionId == TSDB_FUNC_FIRST_DST || functionId == TSDB_FUNC_FIRST) {
return QUERY_IS_ASC_QUERY(pQuery);
}
// in the supplementary scan, only the following functions need to be executed
if (IS_REVERSE_SCAN(pRuntimeEnv) &&
!(functionId == TSDB_FUNC_LAST_DST || functionId == TSDB_FUNC_FIRST_DST || functionId == TSDB_FUNC_FIRST ||
functionId == TSDB_FUNC_LAST || functionId == TSDB_FUNC_TAG || functionId == TSDB_FUNC_TS)) {
if (IS_REVERSE_SCAN(pRuntimeEnv)) {// && (functionId == TSDB_FUNC_TAG || functionId == TSDB_FUNC_TS)) {
return false;
}
@ -1115,8 +1117,9 @@ static void rowwiseApplyFunctions(SQueryRuntimeEnv *pRuntimeEnv, SDataStatis *pS
SQLFunctionCtx *pCtx = pRuntimeEnv->pCtx;
SQuery *pQuery = pRuntimeEnv->pQuery;
TSKEY *primaryKeyCol = (TSKEY*) ((SColumnInfoData *)taosArrayGet(pDataBlock, 0))->pData;
STableQueryInfo* item = pQuery->current;
TSKEY *primaryKeyCol = (TSKEY*) ((SColumnInfoData *)taosArrayGet(pDataBlock, 0))->pData;
bool groupbyStateValue = isGroupbyNormalCol(pQuery->pGroupbyExpr);
SArithmeticSupport *sasArray = calloc((size_t)pQuery->numOfOutput, sizeof(SArithmeticSupport));
@ -1134,7 +1137,7 @@ static void rowwiseApplyFunctions(SQueryRuntimeEnv *pRuntimeEnv, SDataStatis *pS
SDataStatis *pColStatis = NULL;
bool hasNull = hasNullValue(pQuery, k, pDataBlockInfo, pStatis, &pColStatis);
char *dataBlock = getDataBlocks(pRuntimeEnv, &sasArray[k], k, pDataBlockInfo->rows, pDataBlock);
char *dataBlock = getDataBlock(pRuntimeEnv, &sasArray[k], k, pDataBlockInfo->rows, pDataBlock);
setExecParams(pQuery, &pCtx[k], dataBlock, primaryKeyCol, pDataBlockInfo->rows, functionId, pColStatis, hasNull,
&sasArray[k], pRuntimeEnv->scanFlag);
@ -1143,7 +1146,8 @@ static void rowwiseApplyFunctions(SQueryRuntimeEnv *pRuntimeEnv, SDataStatis *pS
// set the input column data
for (int32_t k = 0; k < pQuery->numOfFilterCols; ++k) {
SSingleColumnFilterInfo *pFilterInfo = &pQuery->pFilterInfo[k];
pFilterInfo->pData = getDataBlocks(pRuntimeEnv, &sasArray[k], pFilterInfo->info.colId, pDataBlockInfo->rows, pDataBlock);
pFilterInfo->pData = getDataBlockImpl(pDataBlock, pFilterInfo->info.colId);
assert(pFilterInfo->pData != NULL);
}
int32_t step = GET_FORWARD_DIRECTION_FACTOR(pQuery->order.order);
@ -1157,10 +1161,10 @@ static void rowwiseApplyFunctions(SQueryRuntimeEnv *pRuntimeEnv, SDataStatis *pS
}
int32_t j = 0;
TSKEY lastKey = -1;
int32_t offset = -1;
for (j = 0; j < pDataBlockInfo->rows; ++j) {
int32_t offset = GET_COL_DATA_POS(pQuery, j, step);
offset = GET_COL_DATA_POS(pQuery, j, step);
if (pRuntimeEnv->pTSBuf != NULL) {
int32_t r = doTSJoinFilter(pRuntimeEnv, offset);
@ -1194,7 +1198,6 @@ static void rowwiseApplyFunctions(SQueryRuntimeEnv *pRuntimeEnv, SDataStatis *pS
SWindowStatus *pStatus = getTimeWindowResStatus(pWindowResInfo, curTimeWindow(pWindowResInfo));
doRowwiseApplyFunctions(pRuntimeEnv, pStatus, &win, offset);
lastKey = ts;
STimeWindow nextWin = win;
int32_t index = pWindowResInfo->curIndex;
@ -1223,17 +1226,14 @@ static void rowwiseApplyFunctions(SQueryRuntimeEnv *pRuntimeEnv, SDataStatis *pS
} else { // other queries
// decide which group this rows belongs to according to current state value
if (groupbyStateValue) {
char *stateVal = groupbyColumnData + bytes * offset;
char *val = groupbyColumnData + bytes * offset;
int32_t ret = setGroupResultOutputBuf(pRuntimeEnv, stateVal, type, bytes);
int32_t ret = setGroupResultOutputBuf(pRuntimeEnv, val, type, bytes);
if (ret != TSDB_CODE_SUCCESS) { // null data, too many state code
continue;
}
}
// update the lastKey
lastKey = primaryKeyCol[offset];
// all startOffset are identical
offset -= pCtx[0].startOffset;
@ -1254,7 +1254,7 @@ static void rowwiseApplyFunctions(SQueryRuntimeEnv *pRuntimeEnv, SDataStatis *pS
}
}
pQuery->lastKey = lastKey + step;
item->lastKey = primaryKeyCol[offset] + step;
// todo refactor: extract method
for(int32_t i = 0; i < pQuery->numOfOutput; ++i) {
@ -1269,10 +1269,12 @@ static void rowwiseApplyFunctions(SQueryRuntimeEnv *pRuntimeEnv, SDataStatis *pS
}
static int32_t tableApplyFunctionsOnBlock(SQueryRuntimeEnv *pRuntimeEnv, SDataBlockInfo *pDataBlockInfo,
SDataStatis *pStatis, __block_search_fn_t searchFn,
SWindowResInfo *pWindowResInfo, SArray *pDataBlock) {
SDataStatis *pStatis, __block_search_fn_t searchFn, SArray *pDataBlock) {
SQuery *pQuery = pRuntimeEnv->pQuery;
STableQueryInfo* pTableQInfo = pQuery->current;
SWindowResInfo* pWindowResInfo = &pRuntimeEnv->windowResInfo;
if (pQuery->numOfFilterCols > 0 || pRuntimeEnv->pTSBuf != NULL || isGroupbyNormalCol(pQuery->pGroupbyExpr)) {
rowwiseApplyFunctions(pRuntimeEnv, pStatis, pDataBlockInfo, pWindowResInfo, pDataBlock);
} else {
@ -1280,7 +1282,7 @@ static int32_t tableApplyFunctionsOnBlock(SQueryRuntimeEnv *pRuntimeEnv, SDataBl
}
TSKEY lastKey = QUERY_IS_ASC_QUERY(pQuery) ? pDataBlockInfo->window.ekey : pDataBlockInfo->window.skey;
pQuery->lastKey = lastKey + GET_FORWARD_DIRECTION_FACTOR(pQuery->order.order);
pTableQInfo->lastKey = lastKey + GET_FORWARD_DIRECTION_FACTOR(pQuery->order.order);
doCheckQueryCompleted(pRuntimeEnv, lastKey, pWindowResInfo);
@ -1746,73 +1748,6 @@ static UNUSED_FUNC bool doSetDataInfo(SQInfo *pQInfo, SPointInterpoSupporter *pP
}
}
// TODO refactor code, the best way to implement the last_row is utilizing the iterator
bool normalizeUnBoundLastRowQuery(SQInfo *pQInfo, SPointInterpoSupporter *pPointInterpSupporter) {
#if 0
SQueryRuntimeEnv *pRuntimeEnv = &pQInfo->runtimeEnv;
SQuery * pQuery = pRuntimeEnv->pQuery;
SMeterObj *pMeterObj = pRuntimeEnv->pTabObj;
assert(!QUERY_IS_ASC_QUERY(pQuery) && notHasQueryTimeRange(pQuery));
__block_search_fn_t searchFn = vnodeSearchKeyFunc[pMeterObj->searchAlgorithm];
TSKEY lastKey = -1;
pQuery->fileId = -1;
vnodeFreeFieldsEx(pRuntimeEnv);
// keep in-memory cache status in local variables in case that it may be changed by write operation
getBasicCacheInfoSnapshot(pQuery, pMeterObj->pCache, pMeterObj->vnode);
SCacheInfo *pCacheInfo = (SCacheInfo *)pMeterObj->pCache;
if (pCacheInfo != NULL && pCacheInfo->cacheBlocks != NULL && pQuery->numOfBlocks > 0) {
pQuery->fileId = -1;
TSKEY key = pMeterObj->lastKey;
pQuery->window.skey = key;
pQuery->window.ekey = key;
pQuery->lastKey = pQuery->window.skey;
/*
* cache block may have been flushed to disk, and no data in cache anymore.
* So, copy cache block to local buffer is required.
*/
lastKey = getQueryStartPositionInCache(pRuntimeEnv, &pQuery->slot, &pQuery->pos, false);
if (lastKey < 0) { // data has been flushed to disk, try again search in file
lastKey = getQueryPositionForCacheInvalid(pRuntimeEnv, searchFn);
if (Q_STATUS_EQUAL(pQuery->status, QUERY_NO_DATA_TO_CHECK | QUERY_COMPLETED)) {
return false;
}
}
} else { // no data in cache, try file
TSKEY key = pMeterObj->lastKeyOnFile;
pQuery->window.skey = key;
pQuery->window.ekey = key;
pQuery->lastKey = pQuery->window.skey;
bool ret = getQualifiedDataBlock(pMeterObj, pRuntimeEnv, QUERY_RANGE_LESS_EQUAL, searchFn);
if (!ret) { // no data in file, return false;
return false;
}
lastKey = getTimestampInDiskBlock(pRuntimeEnv, pQuery->pos);
}
assert(lastKey <= pQuery->window.skey);
pQuery->window.skey = lastKey;
pQuery->window.ekey = lastKey;
pQuery->lastKey = pQuery->window.skey;
return getNeighborPoints(pQInfo, pMeterObj, pPointInterpSupporter);
#endif
return true;
}
static void setScanLimitationByResultBuffer(SQuery *pQuery) {
if (isTopBottomQuery(pQuery)) {
pQuery->checkBuffer = 0;
@ -2327,7 +2262,7 @@ static void getNextTimeWindow(SQuery *pQuery, STimeWindow *pTimeWindow) {
pTimeWindow->ekey = pTimeWindow->skey + (pQuery->intervalTime - 1);
}
SArray *loadDataBlockOnDemand(SQueryRuntimeEnv *pRuntimeEnv, SDataBlockInfo *pBlockInfo, SDataStatis **pStatis) {
SArray *loadDataBlockOnDemand(SQueryRuntimeEnv *pRuntimeEnv, void* pQueryHandle, SDataBlockInfo* pBlockInfo, SDataStatis **pStatis) {
SQuery *pQuery = pRuntimeEnv->pQuery;
uint32_t r = 0;
@ -2351,16 +2286,16 @@ SArray *loadDataBlockOnDemand(SQueryRuntimeEnv *pRuntimeEnv, SDataBlockInfo *pBl
qTrace("QInfo:%p data block ignored, brange:%" PRId64 "-%" PRId64 ", rows:%d", GET_QINFO_ADDR(pRuntimeEnv),
pBlockInfo->window.skey, pBlockInfo->window.ekey, pBlockInfo->rows);
} else if (r == BLK_DATA_FILEDS_NEEDED) {
if (tsdbRetrieveDataBlockStatisInfo(pRuntimeEnv->pQueryHandle, pStatis) != TSDB_CODE_SUCCESS) {
if (tsdbRetrieveDataBlockStatisInfo(pQueryHandle, pStatis) != TSDB_CODE_SUCCESS) {
// return DISK_DATA_LOAD_FAILED;
}
if (*pStatis == NULL) {
pDataBlock = tsdbRetrieveDataBlock(pRuntimeEnv->pQueryHandle, NULL);
pDataBlock = tsdbRetrieveDataBlock(pQueryHandle, NULL);
}
} else {
assert(r == BLK_DATA_ALL_NEEDED);
if (tsdbRetrieveDataBlockStatisInfo(pRuntimeEnv->pQueryHandle, pStatis) != TSDB_CODE_SUCCESS) {
if (tsdbRetrieveDataBlockStatisInfo(pQueryHandle, pStatis) != TSDB_CODE_SUCCESS) {
// return DISK_DATA_LOAD_FAILED;
}
@ -2376,7 +2311,7 @@ SArray *loadDataBlockOnDemand(SQueryRuntimeEnv *pRuntimeEnv, SDataBlockInfo *pBl
// return DISK_DATA_DISCARDED;
}
pDataBlock = tsdbRetrieveDataBlock(pRuntimeEnv->pQueryHandle, NULL);
pDataBlock = tsdbRetrieveDataBlock(pQueryHandle, NULL);
}
return pDataBlock;
@ -2447,8 +2382,11 @@ int32_t binarySearchForKey(char *pValue, int num, TSKEY key, int order) {
static int64_t doScanAllDataBlocks(SQueryRuntimeEnv *pRuntimeEnv) {
SQuery *pQuery = pRuntimeEnv->pQuery;
STableQueryInfo* pTableQueryInfo = pQuery->current;
qTrace("QInfo:%p query start, qrange:%" PRId64 "-%" PRId64 ", lastkey:%" PRId64 ", order:%d",
GET_QINFO_ADDR(pRuntimeEnv), pQuery->window.skey, pQuery->window.ekey, pQuery->lastKey, pQuery->order.order);
GET_QINFO_ADDR(pRuntimeEnv), pTableQueryInfo->win.skey, pTableQueryInfo->win.ekey, pTableQueryInfo->lastKey,
pQuery->order.order);
TsdbQueryHandleT pQueryHandle = IS_MASTER_SCAN(pRuntimeEnv)? pRuntimeEnv->pQueryHandle : pRuntimeEnv->pSecQueryHandle;
while (tsdbNextDataBlock(pQueryHandle)) {
@ -2479,7 +2417,7 @@ static int64_t doScanAllDataBlocks(SQueryRuntimeEnv *pRuntimeEnv) {
}
}
// in case of prj/diff query, ensure the output buffer is sufficient to accomodate the results of current block
// in case of prj/diff query, ensure the output buffer is sufficient to accommodate the results of current block
if (!isIntervalQuery(pQuery) && !isGroupbyNormalCol(pQuery->pGroupbyExpr) && !isFixedOutputQuery(pQuery)) {
SResultRec *pRec = &pQuery->rec;
@ -2505,11 +2443,10 @@ static int64_t doScanAllDataBlocks(SQueryRuntimeEnv *pRuntimeEnv) {
}
SDataStatis *pStatis = NULL;
SArray * pDataBlock = loadDataBlockOnDemand(pRuntimeEnv, &blockInfo, &pStatis);
SArray * pDataBlock = loadDataBlockOnDemand(pRuntimeEnv, pQueryHandle, &blockInfo, &pStatis);
pQuery->pos = QUERY_IS_ASC_QUERY(pQuery) ? 0 : blockInfo.rows - 1;
int32_t numOfRes = tableApplyFunctionsOnBlock(pRuntimeEnv, &blockInfo, pStatis, binarySearchForKey,
&pRuntimeEnv->windowResInfo, pDataBlock);
int32_t numOfRes = tableApplyFunctionsOnBlock(pRuntimeEnv, &blockInfo, pStatis, binarySearchForKey, pDataBlock);
qTrace("QInfo:%p check data block, brange:%" PRId64 "-%" PRId64 ", rows:%d, res:%d", GET_QINFO_ADDR(pRuntimeEnv),
blockInfo.window.skey, blockInfo.window.ekey, blockInfo.rows, numOfRes);
@ -2520,7 +2457,7 @@ static int64_t doScanAllDataBlocks(SQueryRuntimeEnv *pRuntimeEnv) {
}
}
// if the result buffer is not full, set the query completed flag
// if the result buffer is not full, set the query complete
if (!Q_STATUS_EQUAL(pQuery->status, QUERY_RESBUF_FULL)) {
setQueryStatus(pQuery, QUERY_COMPLETED);
}
@ -2530,8 +2467,8 @@ static int64_t doScanAllDataBlocks(SQueryRuntimeEnv *pRuntimeEnv) {
int32_t step = QUERY_IS_ASC_QUERY(pQuery) ? QUERY_ASC_FORWARD_STEP : QUERY_DESC_FORWARD_STEP;
closeAllTimeWindow(&pRuntimeEnv->windowResInfo);
removeRedundantWindow(&pRuntimeEnv->windowResInfo, pQuery->lastKey - step, step);
pRuntimeEnv->windowResInfo.curIndex = pRuntimeEnv->windowResInfo.size - 1;
removeRedundantWindow(&pRuntimeEnv->windowResInfo, pTableQueryInfo->lastKey - step, step);
pRuntimeEnv->windowResInfo.curIndex = pRuntimeEnv->windowResInfo.size - 1; // point to the last time window
} else {
assert(Q_STATUS_EQUAL(pQuery->status, QUERY_RESBUF_FULL));
}
@ -2540,8 +2477,6 @@ static int64_t doScanAllDataBlocks(SQueryRuntimeEnv *pRuntimeEnv) {
return 0;
}
static void updatelastkey(SQuery *pQuery, STableQueryInfo *pTableQInfo) { pTableQInfo->lastKey = pQuery->lastKey; }
/*
* set tag value in SQLFunctionCtx
* e.g.,tag information into input buffer
@ -2554,17 +2489,17 @@ static void doSetTagValueInParam(void *tsdb, STableId* pTableId, int32_t tagColI
int16_t type = 0;
if (tagColId == TSDB_TBNAME_COLUMN_INDEX) {
tsdbGetTableName(tsdb, pTableId, &val);
bytes = strnlen(val, TSDB_TABLE_NAME_LEN);
val = tsdbGetTableName(tsdb, pTableId, &bytes);
type = TSDB_DATA_TYPE_BINARY;
tVariantCreateFromBinary(param, varDataVal(val), varDataLen(val), type);
} else {
tsdbGetTableTagVal(tsdb, pTableId, tagColId, &type, &bytes, &val);
}
if (type == TSDB_DATA_TYPE_BINARY || type == TSDB_DATA_TYPE_NCHAR) {
tVariantCreateFromBinary(param, varDataVal(val), varDataLen(val), type);
} else {
tVariantCreateFromBinary(param, val, bytes, type);
if (tagColId == TSDB_TBNAME_COLUMN_INDEX) {
tfree(val);
}
}
}
@ -3082,11 +3017,6 @@ void resetMergeResultBuf(SQuery *pQuery, SQLFunctionCtx *pCtx, SResultInfo *pRes
}
}
void setTableDataInfo(STableQueryInfo *pTableQueryInfo, int32_t tableIndex, int32_t groupId) {
pTableQueryInfo->groupIdx = groupId;
pTableQueryInfo->tableIndex = tableIndex;
}
static void updateTableQueryInfoForReverseScan(SQuery *pQuery, STableQueryInfo *pTableQueryInfo) {
if (pTableQueryInfo == NULL) {
return;
@ -3132,19 +3062,6 @@ static void disableFuncInReverseScanImpl(SQInfo* pQInfo, SWindowResInfo *pWindow
}
}
}
int32_t numOfGroups = taosArrayGetSize(pQInfo->groupInfo.pGroupList);
for(int32_t i = 0; i < numOfGroups; ++i) {
SArray *group = taosArrayGetP(pQInfo->groupInfo.pGroupList, i);
qTrace("QInfo:%p no result in group %d, continue", pQInfo, pQInfo->groupIndex - 1);
size_t t = taosArrayGetSize(group);
for (int32_t j = 0; j < t; ++j) {
SGroupItem *item = taosArrayGet(group, j);
updateTableQueryInfoForReverseScan(pQuery, item->info);
}
}
}
void disableFuncInReverseScan(SQInfo *pQInfo) {
@ -3170,6 +3087,18 @@ void disableFuncInReverseScan(SQInfo *pQInfo) {
}
}
}
int32_t numOfGroups = taosArrayGetSize(pQInfo->groupInfo.pGroupList);
for(int32_t i = 0; i < numOfGroups; ++i) {
SArray *group = taosArrayGetP(pQInfo->groupInfo.pGroupList, i);
size_t t = taosArrayGetSize(group);
for (int32_t j = 0; j < t; ++j) {
SGroupItem *item = taosArrayGet(group, j);
updateTableQueryInfoForReverseScan(pQuery, item->info);
}
}
}
void switchCtxOrder(SQueryRuntimeEnv *pRuntimeEnv) {
@ -3345,13 +3274,14 @@ bool needScanDataBlocksAgain(SQueryRuntimeEnv *pRuntimeEnv) {
static SQueryStatusInfo getQueryStatusInfo(SQueryRuntimeEnv *pRuntimeEnv) {
SQuery *pQuery = pRuntimeEnv->pQuery;
STableQueryInfo* pTableQueryInfo = pQuery->current;
SQueryStatusInfo info = {
.status = pQuery->status,
.windowIndex = pRuntimeEnv->windowResInfo.curIndex,
.lastKey = pQuery->lastKey,
.lastKey = pTableQueryInfo->lastKey,
.w = pQuery->window,
.curWindow = {.skey = pQuery->lastKey, .ekey = pQuery->window.ekey},
.curWindow = {.skey = pTableQueryInfo->lastKey, .ekey = pTableQueryInfo->win.ekey},
};
return info;
@ -3395,6 +3325,7 @@ static void setEnvBeforeReverseScan(SQueryRuntimeEnv *pRuntimeEnv, SQueryStatusI
static void clearEnvAfterReverseScan(SQueryRuntimeEnv *pRuntimeEnv, SQueryStatusInfo *pStatus) {
SQuery *pQuery = pRuntimeEnv->pQuery;
STableQueryInfo* pTableQueryInfo = pQuery->current;
SWITCH_ORDER(pQuery->order.order);
switchCtxOrder(pRuntimeEnv);
@ -3408,17 +3339,19 @@ static void clearEnvAfterReverseScan(SQueryRuntimeEnv *pRuntimeEnv, SQueryStatus
// update the pQuery->window.skey and pQuery->window.ekey to limit the scan scope of sliding query
// during reverse scan
pQuery->lastKey = pStatus->lastKey;
pTableQueryInfo->lastKey = pStatus->lastKey;
pQuery->status = pStatus->status;
pQuery->window = pStatus->w;
pTableQueryInfo->win = pStatus->w;
}
void scanAllDataBlocks(SQueryRuntimeEnv *pRuntimeEnv) {
SQInfo *pQInfo = (SQInfo *) GET_QINFO_ADDR(pRuntimeEnv);
SQuery *pQuery = pRuntimeEnv->pQuery;
STableQueryInfo *pTableQueryInfo = pQuery->current;
setQueryStatus(pQuery, QUERY_NOT_COMPLETED);
// store the start query position
SQInfo * pQInfo = (SQInfo *)GET_QINFO_ADDR(pRuntimeEnv);
SQueryStatusInfo qstatus = getQueryStatusInfo(pRuntimeEnv);
SET_MASTER_SCAN_FLAG(pRuntimeEnv);
@ -3429,7 +3362,7 @@ void scanAllDataBlocks(SQueryRuntimeEnv *pRuntimeEnv) {
if (pRuntimeEnv->scanFlag == MASTER_SCAN) {
qstatus.status = pQuery->status;
qstatus.curWindow.ekey = pQuery->lastKey - step;
qstatus.curWindow.ekey = pTableQueryInfo->lastKey - step;
}
if (!needScanDataBlocksAgain(pRuntimeEnv)) {
@ -3549,12 +3482,10 @@ void destroyTableQueryInfo(STableQueryInfo *pTableQueryInfo, int32_t numOfCols)
void restoreIntervalQueryRange(SQueryRuntimeEnv *pRuntimeEnv, STableQueryInfo *pTableQueryInfo) {
SQuery *pQuery = pRuntimeEnv->pQuery;
pQuery->current = pTableQueryInfo;
pQuery->window = pTableQueryInfo->win;
pQuery->lastKey = pTableQueryInfo->lastKey;
assert(((pQuery->lastKey >= pQuery->window.skey) && QUERY_IS_ASC_QUERY(pQuery)) ||
((pQuery->lastKey <= pQuery->window.skey) && !QUERY_IS_ASC_QUERY(pQuery)));
assert(((pTableQueryInfo->lastKey >= pTableQueryInfo->win.skey) && QUERY_IS_ASC_QUERY(pQuery)) ||
((pTableQueryInfo->lastKey <= pTableQueryInfo->win.skey) && !QUERY_IS_ASC_QUERY(pQuery)));
}
/**
@ -3562,8 +3493,10 @@ void restoreIntervalQueryRange(SQueryRuntimeEnv *pRuntimeEnv, STableQueryInfo *p
* @param pRuntimeEnv
* @param pDataBlockInfo
*/
void setExecutionContext(SQInfo *pQInfo, STableQueryInfo *pTableQueryInfo, STableId* pTableId, int32_t groupIdx, TSKEY nextKey) {
void setExecutionContext(SQInfo *pQInfo, STableId* pTableId, int32_t groupIdx, TSKEY nextKey) {
SQueryRuntimeEnv *pRuntimeEnv = &pQInfo->runtimeEnv;
STableQueryInfo *pTableQueryInfo = pRuntimeEnv->pQuery->current;
SWindowResInfo * pWindowResInfo = &pRuntimeEnv->windowResInfo;
int32_t GROUPRESULTID = 1;
@ -3647,12 +3580,12 @@ int32_t setAdditionalInfo(SQInfo *pQInfo, STableId* pTableId, STableQueryInfo *p
* merged during merge stage. In this case, we need the pTableQueryInfo->lastResRows to decide if there
* is a previous result generated or not.
*/
void setIntervalQueryRange(STableQueryInfo *pTableQueryInfo, SQInfo *pQInfo, TSKEY key) {
void setIntervalQueryRange(SQInfo *pQInfo, TSKEY key) {
SQueryRuntimeEnv *pRuntimeEnv = &pQInfo->runtimeEnv;
SQuery * pQuery = pRuntimeEnv->pQuery;
STableQueryInfo *pTableQueryInfo = pQuery->current;
if (pTableQueryInfo->queryRangeSet) {
pQuery->lastKey = key;
pTableQueryInfo->lastKey = key;
} else {
pQuery->window.skey = key;
@ -3689,8 +3622,6 @@ void setIntervalQueryRange(STableQueryInfo *pTableQueryInfo, SQInfo *pQInfo, TSK
pTableQueryInfo->queryRangeSet = 1;
pTableQueryInfo->lastKey = pQuery->window.skey;
pTableQueryInfo->win.skey = pQuery->window.skey;
pQuery->lastKey = pQuery->window.skey;
}
}
@ -3710,7 +3641,9 @@ bool needPrimaryTimestampCol(SQuery *pQuery, SDataBlockInfo *pDataBlockInfo) {
* 2. if there are top/bottom, first_dst/last_dst functions, we need to load timestamp column in any cases;
*/
STimeWindow *w = &pDataBlockInfo->window;
bool loadPrimaryTS = (pQuery->lastKey >= w->skey && pQuery->lastKey <= w->ekey) ||
STableQueryInfo* pTableQueryInfo = pQuery->current;
bool loadPrimaryTS = (pTableQueryInfo->lastKey >= w->skey && pTableQueryInfo->lastKey <= w->ekey) ||
(pQuery->window.ekey >= w->skey && pQuery->window.ekey <= w->ekey) || requireTimestamp(pQuery);
return loadPrimaryTS;
@ -3847,7 +3780,6 @@ void stableApplyFunctionsOnBlock(SQueryRuntimeEnv *pRuntimeEnv, STableQueryInfo
}
updateWindowResNumOfRes(pRuntimeEnv, pTableQueryInfo);
updatelastkey(pQuery, pTableQueryInfo);
}
bool vnodeHasRemainResults(void *handle) {
@ -4041,10 +3973,12 @@ void vnodePrintQueryStatistics(SQInfo *pQInfo) {
static void updateOffsetVal(SQueryRuntimeEnv *pRuntimeEnv, SDataBlockInfo *pBlockInfo) {
SQuery *pQuery = pRuntimeEnv->pQuery;
STableQueryInfo* pTableQueryInfo = pQuery->current;
int32_t step = GET_FORWARD_DIRECTION_FACTOR(pQuery->order.order);
if (pQuery->limit.offset == pBlockInfo->rows) { // current block will ignore completed
pQuery->lastKey = QUERY_IS_ASC_QUERY(pQuery) ? pBlockInfo->window.ekey + step : pBlockInfo->window.skey + step;
pTableQueryInfo->lastKey = QUERY_IS_ASC_QUERY(pQuery) ? pBlockInfo->window.ekey + step : pBlockInfo->window.skey + step;
pQuery->limit.offset = 0;
return;
}
@ -4064,11 +3998,10 @@ static void updateOffsetVal(SQueryRuntimeEnv *pRuntimeEnv, SDataBlockInfo *pBloc
TSKEY *keys = (TSKEY *)pColInfoData->pData;
// update the offset value
pQuery->lastKey = keys[pQuery->pos];
pTableQueryInfo->lastKey = keys[pQuery->pos];
pQuery->limit.offset = 0;
int32_t numOfRes = tableApplyFunctionsOnBlock(pRuntimeEnv, pBlockInfo, NULL, binarySearchForKey,
&pRuntimeEnv->windowResInfo, pDataBlock);
int32_t numOfRes = tableApplyFunctionsOnBlock(pRuntimeEnv, pBlockInfo, NULL, binarySearchForKey, pDataBlock);
qTrace("QInfo:%p check data block, brange:%" PRId64 "-%" PRId64 ", rows:%d, res:%d", GET_QINFO_ADDR(pRuntimeEnv),
pBlockInfo->window.skey, pBlockInfo->window.ekey, pBlockInfo->rows, numOfRes);
@ -4084,6 +4017,7 @@ void skipBlocks(SQueryRuntimeEnv *pRuntimeEnv) {
pQuery->pos = 0;
int32_t step = GET_FORWARD_DIRECTION_FACTOR(pQuery->order.order);
STableQueryInfo* pTableQueryInfo = pQuery->current;
TsdbQueryHandleT pQueryHandle = pRuntimeEnv->pQueryHandle;
while (tsdbNextDataBlock(pQueryHandle)) {
@ -4095,8 +4029,8 @@ void skipBlocks(SQueryRuntimeEnv *pRuntimeEnv) {
if (pQuery->limit.offset > blockInfo.rows) {
pQuery->limit.offset -= blockInfo.rows;
pQuery->lastKey = (QUERY_IS_ASC_QUERY(pQuery)) ? blockInfo.window.ekey : blockInfo.window.skey;
pQuery->lastKey += step;
pTableQueryInfo->lastKey = (QUERY_IS_ASC_QUERY(pQuery)) ? blockInfo.window.ekey : blockInfo.window.skey;
pTableQueryInfo->lastKey += step;
qTrace("QInfo:%p skip rows:%d, offset:%" PRId64 "", GET_QINFO_ADDR(pRuntimeEnv), blockInfo.rows,
pQuery->limit.offset);
@ -4125,6 +4059,7 @@ static bool skipTimeInterval(SQueryRuntimeEnv *pRuntimeEnv) {
TSKEY skey1, ekey1;
STimeWindow w = {0};
SWindowResInfo *pWindowResInfo = &pRuntimeEnv->windowResInfo;
STableQueryInfo *pTableQueryInfo = pQuery->current;
while (tsdbNextDataBlock(pRuntimeEnv->pQueryHandle)) {
SDataBlockInfo blockInfo = tsdbRetrieveDataBlockInfo(pRuntimeEnv->pQueryHandle);
@ -4170,11 +4105,10 @@ static bool skipTimeInterval(SQueryRuntimeEnv *pRuntimeEnv) {
// set the abort info
pQuery->pos = startPos;
pQuery->lastKey = ((TSKEY *)pColInfoData->pData)[startPos];
pTableQueryInfo->lastKey = ((TSKEY *)pColInfoData->pData)[startPos];
pWindowResInfo->prevSKey = tw.skey;
int32_t numOfRes = tableApplyFunctionsOnBlock(pRuntimeEnv, &blockInfo, NULL, binarySearchForKey,
&pRuntimeEnv->windowResInfo, pDataBlock);
int32_t numOfRes = tableApplyFunctionsOnBlock(pRuntimeEnv, &blockInfo, NULL, binarySearchForKey, pDataBlock);
qTrace("QInfo:%p check data block, brange:%" PRId64 "-%" PRId64 ", rows:%d, res:%d",
GET_QINFO_ADDR(pRuntimeEnv), blockInfo.window.skey, blockInfo.window.ekey, blockInfo.rows, numOfRes);
@ -4199,7 +4133,7 @@ static bool skipTimeInterval(SQueryRuntimeEnv *pRuntimeEnv) {
// set the abort info
pQuery->pos = startPos;
pQuery->lastKey = ((TSKEY *)pColInfoData->pData)[startPos];
pTableQueryInfo->lastKey = ((TSKEY *)pColInfoData->pData)[startPos];
pWindowResInfo->prevSKey = tw.skey;
win = tw;
} else {
@ -4220,9 +4154,6 @@ int32_t doInitQInfo(SQInfo *pQInfo, void *param, void *tsdb, int32_t vgId, bool
setScanLimitationByResultBuffer(pQuery);
changeExecuteScanOrder(pQuery, false);
// dataInCache requires lastKey value
pQuery->lastKey = pQuery->window.skey;
STsdbQueryCond cond = {
.twindow = pQuery->window,
.order = pQuery->order.order,
@ -4238,6 +4169,9 @@ int32_t doInitQInfo(SQInfo *pQInfo, void *param, void *tsdb, int32_t vgId, bool
} else if (!isSTableQuery || isIntervalQuery(pQuery) || isFixedOutputQuery(pQuery)) {
pRuntimeEnv->pQueryHandle = tsdbQueryTables(tsdb, &cond, &pQInfo->tableIdGroupInfo);
}
// create the table query support structures
createTableQueryInfo(pQInfo);
}
pQInfo->tsdb = tsdb;
@ -4331,7 +4265,6 @@ int32_t doInitQInfo(SQInfo *pQInfo, void *param, void *tsdb, int32_t vgId, bool
// }
// the pQuery->window.skey is changed during normalizedFirstQueryRange, so set the newest lastkey value
pQuery->lastKey = pQuery->window.skey;
return TSDB_CODE_SUCCESS;
}
@ -4404,13 +4337,14 @@ static int64_t queryOnDataBlocks(SQInfo *pQInfo) {
restoreIntervalQueryRange(pRuntimeEnv, pTableQueryInfo);
SDataStatis *pStatis = NULL;
SArray * pDataBlock = loadDataBlockOnDemand(pRuntimeEnv, &blockInfo, &pStatis);
SArray *pDataBlock = loadDataBlockOnDemand(pRuntimeEnv, pQueryHandle, &blockInfo, &pStatis);
TSKEY nextKey = blockInfo.window.skey;
if (!isIntervalQuery(pQuery)) {
setExecutionContext(pQInfo, pTableQueryInfo, &pTableQueryInfo->id, pTableQueryInfo->groupIdx, nextKey);
setExecutionContext(pQInfo, &pTableQueryInfo->id, pTableQueryInfo->groupIdx, nextKey);
} else { // interval query
setIntervalQueryRange(pTableQueryInfo, pQInfo, nextKey);
setIntervalQueryRange(pQInfo, nextKey);
int32_t ret = setAdditionalInfo(pQInfo, &pTableQueryInfo->id, pTableQueryInfo);
if (ret != TSDB_CODE_SUCCESS) {
@ -4446,6 +4380,7 @@ static bool multiTableMultioutputHelper(SQInfo *pQInfo, int32_t index) {
.numOfCols = pQuery->numOfCols,
};
// todo refactor
SArray *g1 = taosArrayInit(1, POINTER_BYTES);
SArray *tx = taosArrayInit(1, sizeof(STableId));
@ -4564,6 +4499,9 @@ static void sequentialTableProcess(SQInfo *pQInfo) {
initCtxOutputBuf(pRuntimeEnv);
setTagVal(pRuntimeEnv, (STableId*) taosArrayGet(tx, 0), pQInfo->tsdb);
// here we simply set the first table as current table
pRuntimeEnv->pQuery->current = ((SGroupItem*) taosArrayGet(group, 0))->info;
scanAllDataBlocks(pRuntimeEnv);
int64_t numOfRes = getNumOfResult(pRuntimeEnv);
@ -4580,8 +4518,6 @@ static void sequentialTableProcess(SQInfo *pQInfo) {
}
}
} else {
createTableQueryInfo(pQInfo);
/*
* 1. super table projection query, 2. group-by on normal columns query, 3. ts-comp query
* if the subgroup index is larger than 0, results generated by group by tbname,k is existed.
@ -4614,11 +4550,7 @@ static void sequentialTableProcess(SQInfo *pQInfo) {
}
SGroupItem *item = taosArrayGet(group, pQInfo->tableIndex);
STableQueryInfo *pInfo = item->info;
if (pInfo->lastKey > 0) {
pQuery->window.skey = pInfo->lastKey;
}
pQuery->current = item->info;
if (!multiTableMultioutputHelper(pQInfo, pQInfo->tableIndex)) {
pQInfo->tableIndex++;
@ -4627,7 +4559,7 @@ static void sequentialTableProcess(SQInfo *pQInfo) {
// SPointInterpoSupporter pointInterpSupporter = {0};
// TODO handle the limit problem
// TODO handle the limit offset problem
if (pQuery->numOfFilterCols == 0 && pQuery->limit.offset > 0) {
// skipBlocks(pRuntimeEnv);
@ -4659,8 +4591,6 @@ static void sequentialTableProcess(SQInfo *pQInfo) {
* to ensure that, we can reset the query range once query on a meter is completed.
*/
pQInfo->tableIndex++;
pInfo->lastKey = pQuery->lastKey;
// if the buffer is full or group by each table, we need to jump out of the loop
if (Q_STATUS_EQUAL(pQuery->status, QUERY_RESBUF_FULL) /*||
isGroupbyEachTable(pQuery->pGroupbyExpr, pSupporter->pSidSet)*/) {
@ -4668,7 +4598,7 @@ static void sequentialTableProcess(SQInfo *pQInfo) {
}
} else { // forward query range
pQuery->window.skey = pQuery->lastKey;
pQuery->window.skey = pQuery->current->lastKey;
// all data in the result buffer are skipped due to the offset, continue to retrieve data from current meter
if (pQuery->rec.rows == 0) {
@ -4853,12 +4783,9 @@ static void multiTableQueryProcess(SQInfo *pQInfo) {
qTrace("QInfo:%p query start, qrange:%" PRId64 "-%" PRId64 ", order:%d, forward scan start", pQInfo,
pQuery->window.skey, pQuery->window.ekey, pQuery->order.order);
// create the query support structures
createTableQueryInfo(pQInfo);
// do check all qualified data blocks
int64_t el = queryOnDataBlocks(pQInfo);
qTrace("QInfo:%p forward scan completed, elapsed time: %lldms, reversed scan start", pQInfo, el);
qTrace("QInfo:%p master scan completed, elapsed time: %lldms, reverse scan start", pQInfo, el);
// query error occurred or query is killed, abort current execution
if (pQInfo->code != TSDB_CODE_SUCCESS || isQueryKilled(pQInfo)) {
@ -4909,9 +4836,11 @@ static void multiTableQueryProcess(SQInfo *pQInfo) {
* select count(*)/top(field,k)/avg(field name) from table_name [where ts>now-1a];
* select count(*) from table_name group by status_column;
*/
static void tableFixedOutputProcess(SQInfo *pQInfo) {
static void tableFixedOutputProcess(SQInfo *pQInfo, STableQueryInfo* pTableInfo) {
SQueryRuntimeEnv *pRuntimeEnv = &pQInfo->runtimeEnv;
SQuery *pQuery = pRuntimeEnv->pQuery;
pQuery->current = pTableInfo; // set current query table info
scanAllDataBlocks(pRuntimeEnv);
finalizeQueryResult(pRuntimeEnv);
@ -4932,9 +4861,11 @@ static void tableFixedOutputProcess(SQInfo *pQInfo) {
limitResults(pQInfo);
}
static void tableMultiOutputProcess(SQInfo *pQInfo) {
static void tableMultiOutputProcess(SQInfo *pQInfo, STableQueryInfo* pTableInfo) {
SQueryRuntimeEnv *pRuntimeEnv = &pQInfo->runtimeEnv;
SQuery *pQuery = pRuntimeEnv->pQuery;
pQuery->current = pTableInfo;
// for ts_comp query, re-initialized is not allowed
if (!isTSCompQuery(pQuery)) {
@ -4970,15 +4901,15 @@ static void tableMultiOutputProcess(SQInfo *pQInfo) {
}
qTrace("QInfo:%p vid:%d sid:%d id:%s, skip current result, offset:%" PRId64 ", next qrange:%" PRId64 "-%" PRId64,
pQInfo, pQuery->limit.offset, pQuery->lastKey);
pQInfo, pQuery->limit.offset, pQuery->current->lastKey);
resetCtxOutputBuf(pRuntimeEnv);
}
limitResults(pQInfo);
if (Q_STATUS_EQUAL(pQuery->status, QUERY_RESBUF_FULL)) {
qTrace("QInfo:%p query paused due to output limitation, next qrange:%" PRId64 "-%" PRId64, pQInfo, pQuery->lastKey,
pQuery->window.ekey);
qTrace("QInfo:%p query paused due to output limitation, next qrange:%" PRId64 "-%" PRId64, pQInfo,
pQuery->current->lastKey, pQuery->window.ekey);
}
if (!isTSCompQuery(pQuery)) {
@ -5018,11 +4949,12 @@ static void tableIntervalProcessImpl(SQueryRuntimeEnv *pRuntimeEnv) {
}
// handle time interval query on table
static void tableIntervalProcess(SQInfo *pQInfo) {
static void tableIntervalProcess(SQInfo *pQInfo, STableQueryInfo* pTableInfo) {
SQueryRuntimeEnv *pRuntimeEnv = &(pQInfo->runtimeEnv);
SQuery * pQuery = pRuntimeEnv->pQuery;
int32_t numOfInterpo = 0;
SQuery *pQuery = pRuntimeEnv->pQuery;
pQuery->current = pTableInfo;
// skip blocks without load the actual data block from file if no filter condition present
skipTimeInterval(pRuntimeEnv);
@ -5132,14 +5064,18 @@ static void tableQueryImpl(SQInfo *pQInfo) {
pQuery->rec.rows = 0;
int64_t st = taosGetTimestampUs();
assert(pQInfo->groupInfo.numOfTables == 1);
SArray* g = taosArrayGetP(pQInfo->groupInfo.pGroupList, 0);
SGroupItem* item = taosArrayGet(g, 0);
// group by normal column, sliding window query, interval query are handled by interval query processor
if (isIntervalQuery(pQuery) || isGroupbyNormalCol(pQuery->pGroupbyExpr)) { // interval (down sampling operation)
tableIntervalProcess(pQInfo);
tableIntervalProcess(pQInfo, item->info);
} else if (isFixedOutputQuery(pQuery)) {
tableFixedOutputProcess(pQInfo);
tableFixedOutputProcess(pQInfo, item->info);
} else { // diff/add/multiply/subtract/division
assert(pQuery->checkBuffer == 1);
tableMultiOutputProcess(pQInfo);
tableMultiOutputProcess(pQInfo, item->info);
}
// record the total elapsed time
@ -5833,7 +5769,6 @@ static SQInfo *createQInfoImpl(SQueryTableMsg *pQueryMsg, SSqlGroupbyExpr *pGrou
pQuery->pos = -1;
pQuery->window = pQueryMsg->window;
pQuery->lastKey = pQuery->window.skey;
if (sem_init(&pQInfo->dataReady, 0, 0) != 0) {
qError("QInfo:%p init dataReady sem failed, reason:%s", pQInfo, strerror(errno));
@ -5911,8 +5846,6 @@ static int32_t initQInfo(SQueryTableMsg *pQueryMsg, void *tsdb, int32_t vgId, SQ
goto _error;
}
// qTrace("QInfo:%p set query flag and prepare runtime environment completed, ref:%d, wait for schedule", pQInfo,
// pQInfo->refCount);
return code;
_error:
@ -6085,34 +6018,35 @@ int32_t qCreateQueryInfo(void *tsdb, int32_t vgId, SQueryTableMsg *pQueryMsg, qi
if (pQueryMsg->numOfTables <= 0) {
qError("Invalid number of tables to query, numOfTables:%d", pQueryMsg->numOfTables);
code = TSDB_CODE_INVALID_QUERY_MSG;
goto _query_over;
goto _over;
}
if (pTableIdList == NULL || taosArrayGetSize(pTableIdList) == 0) {
qError("qmsg:%p, SQueryTableMsg wrong format", pQueryMsg);
code = TSDB_CODE_INVALID_QUERY_MSG;
goto _query_over;
goto _over;
}
SExprInfo *pExprs = NULL;
if ((code = createSqlFunctionExprFromMsg(pQueryMsg, &pExprs, pExprMsg, pTagColumnInfo)) != TSDB_CODE_SUCCESS) {
goto _query_over;
goto _over;
}
SSqlGroupbyExpr *pGroupbyExpr = createGroupbyExprFromMsg(pQueryMsg, pGroupColIndex, &code);
if ((pGroupbyExpr == NULL && pQueryMsg->numOfGroupCols != 0) || code != TSDB_CODE_SUCCESS) {
goto _query_over;
goto _over;
}
bool isSTableQuery = false;
STableGroupInfo groupInfo = {0};
//todo multitable_query??
if (TSDB_QUERY_HAS_TYPE(pQueryMsg->queryType, TSDB_QUERY_TYPE_MULTITABLE_QUERY|TSDB_QUERY_TYPE_TABLE_QUERY)) {
isSTableQuery = TSDB_QUERY_HAS_TYPE(pQueryMsg->queryType, TSDB_QUERY_TYPE_MULTITABLE_QUERY);
STableId *id = taosArrayGet(pTableIdList, 0);
if ((code = tsdbGetOneTableGroup(tsdb, id->uid, &groupInfo)) != TSDB_CODE_SUCCESS) {
goto _query_over;
goto _over;
}
} else if (TSDB_QUERY_HAS_TYPE(pQueryMsg->queryType, TSDB_QUERY_TYPE_STABLE_QUERY)) {
isSTableQuery = true;
@ -6129,7 +6063,7 @@ int32_t qCreateQueryInfo(void *tsdb, int32_t vgId, SQueryTableMsg *pQueryMsg, qi
numOfGroupByCols);
if (groupInfo.numOfTables == 0) { // no qualified tables no need to do query
code = TSDB_CODE_SUCCESS;
goto _query_over;
goto _over;
}
} else {
assert(0);
@ -6138,11 +6072,12 @@ int32_t qCreateQueryInfo(void *tsdb, int32_t vgId, SQueryTableMsg *pQueryMsg, qi
(*pQInfo) = createQInfoImpl(pQueryMsg, pGroupbyExpr, pExprs, &groupInfo, pTagColumnInfo);
if ((*pQInfo) == NULL) {
code = TSDB_CODE_SERV_OUT_OF_MEMORY;
goto _over;
}
code = initQInfo(pQueryMsg, tsdb, vgId, *pQInfo, isSTableQuery);
_query_over:
_over:
tfree(tagCond);
tfree(tbnameCond);
taosArrayDestroy(pTableIdList);
@ -6318,12 +6253,10 @@ static void buildTagQueryResult(SQInfo* pQInfo) {
for(int32_t j = 0; j < pQuery->numOfOutput; ++j) {
// todo check the return value, refactor codes
if (pExprInfo[j].base.colInfo.colId == TSDB_TBNAME_COLUMN_INDEX) {
tsdbGetTableName(pQInfo->tsdb, &item->id, &data);
data = tsdbGetTableName(pQInfo->tsdb, &item->id, &bytes);
char* dst = pQuery->sdata[j]->data + i * (TSDB_TABLE_NAME_LEN + VARSTR_HEADER_SIZE);
STR_WITH_MAXSIZE_TO_VARSTR(dst, data, TSDB_TABLE_NAME_LEN);
tfree(data);
memcpy(dst, data, varDataTLen(data));
} else {// todo refactor, return the true length of binary|nchar data
tsdbGetTableTagVal(pQInfo->tsdb, &item->id, pExprInfo[j].base.colInfo.colId, &type, &bytes, &data);
assert(bytes == pExprInfo[j].bytes && type == pExprInfo[j].type);

4
src/rpc/src/rpcMain.c
View File

@ -1094,8 +1094,8 @@ static void rpcSendMsgToPeer(SRpcConn *pConn, void *msg, int msgLen) {
} else {
if (pHead->code == 0) pConn->secured = 1; // for success response, set link as secured
if (pHead->msgType < TSDB_MSG_TYPE_CM_HEARTBEAT || (rpcDebugFlag & 16))
tTrace( "%s %p, %s is sent to %s:%hu, code:0x%x len:%d sig:0x%08x:0x%08x:%d",
pRpc->label, pConn, taosMsg[pHead->msgType], pConn->peerFqdn, pConn->peerPort,
tTrace( "%s %p, %s is sent to 0x%x:%hu, code:0x%x len:%d sig:0x%08x:0x%08x:%d",
pRpc->label, pConn, taosMsg[pHead->msgType], pConn->peerIp, pConn->peerPort,
htonl(pHead->code), msgLen, pHead->sourceId, pHead->destId, pHead->tranId);
}

12
src/tsdb/inc/tsdbMain.h
View File

@ -81,7 +81,6 @@ typedef struct {
// ---------- TSDB TABLE DEFINITION
typedef struct STable {
int8_t type;
char * name;
STableId tableId;
int64_t superUid; // Super table UID
int32_t sversion;
@ -96,9 +95,10 @@ typedef struct STable {
TSKEY lastKey; // lastkey inserted in this table, initialized as 0, TODO: make a structure
struct STable *next; // TODO: remove the next
struct STable *prev;
tstr * name; // NOTE: there a flexible string here
} STable;
#define TSDB_GET_TABLE_LAST_KEY(pTable) ((pTable)->lastKey)
#define TSDB_GET_TABLE_LAST_KEY(tb) ((tb)->lastKey)
void * tsdbEncodeTable(STable *pTable, int *contLen);
STable *tsdbDecodeTable(void *cont, int contLen);
@ -121,6 +121,12 @@ typedef struct {
int maxCols;
} STsdbMeta;
// element put in skiplist for each table
typedef struct STableIndexElem {
STsdbMeta* pMeta;
STable* pTable;
} STableIndexElem;
STsdbMeta *tsdbInitMeta(char *rootDir, int32_t maxTables);
int32_t tsdbFreeMeta(STsdbMeta *pMeta);
STSchema * tsdbGetTableSchema(STsdbMeta *pMeta, STable *pTable);
@ -150,7 +156,7 @@ int32_t tsdbDropTableImpl(STsdbMeta *pMeta, STableId tableId);
STable *tsdbIsValidTableToInsert(STsdbMeta *pMeta, STableId tableId);
// int32_t tsdbInsertRowToTableImpl(SSkipListNode *pNode, STable *pTable);
STable *tsdbGetTableByUid(STsdbMeta *pMeta, int64_t uid);
char * getTupleKey(const void *data);
char *getTSTupleKey(const void * data);
typedef struct {
int blockId;

4
src/tsdb/src/tsdbMain.c
View File

@ -808,7 +808,7 @@ static int32_t tdInsertRowToTable(STsdbRepo *pRepo, SDataRow row, STable *pTable
if (pTable->mem == NULL) {
pTable->mem = (SMemTable *)calloc(1, sizeof(SMemTable));
if (pTable->mem == NULL) return -1;
pTable->mem->pData = tSkipListCreate(5, TSDB_DATA_TYPE_TIMESTAMP, TYPE_BYTES[TSDB_DATA_TYPE_TIMESTAMP], 0, 0, 0, getTupleKey);
pTable->mem->pData = tSkipListCreate(5, TSDB_DATA_TYPE_TIMESTAMP, TYPE_BYTES[TSDB_DATA_TYPE_TIMESTAMP], 0, 0, 0, getTSTupleKey);
pTable->mem->keyFirst = INT64_MAX;
pTable->mem->keyLast = 0;
}
@ -831,7 +831,7 @@ static int32_t tdInsertRowToTable(STsdbRepo *pRepo, SDataRow row, STable *pTable
if (pTable->mem == NULL) {
pTable->mem = (SMemTable *)calloc(1, sizeof(SMemTable));
if (pTable->mem == NULL) return -1;
pTable->mem->pData = tSkipListCreate(5, TSDB_DATA_TYPE_TIMESTAMP, TYPE_BYTES[TSDB_DATA_TYPE_TIMESTAMP], 0, 0, 0, getTupleKey);
pTable->mem->pData = tSkipListCreate(5, TSDB_DATA_TYPE_TIMESTAMP, TYPE_BYTES[TSDB_DATA_TYPE_TIMESTAMP], 0, 0, 0, getTSTupleKey);
pTable->mem->keyFirst = INT64_MAX;
pTable->mem->keyLast = 0;
}

84
src/tsdb/src/tsdbMeta.c
View File

@ -1,6 +1,4 @@
#include <stdlib.h>
// #include "taosdef.h"
#include "tskiplist.h"
#include "tsdb.h"
#include "taosdef.h"
@ -10,6 +8,8 @@
#define TSDB_SUPER_TABLE_SL_LEVEL 5 // TODO: may change here
#define TSDB_META_FILE_NAME "META"
const int32_t DEFAULT_TAG_INDEX_COLUMN = 0;
static int tsdbFreeTable(STable *pTable);
static int32_t tsdbCheckTableCfg(STableCfg *pCfg);
static int tsdbAddTableToMeta(STsdbMeta *pMeta, STable *pTable, bool addIdx);
@ -39,11 +39,12 @@ void *tsdbEncodeTable(STable *pTable, int *contLen) {
void *ptr = ret;
T_APPEND_MEMBER(ptr, pTable, STable, type);
// Encode name
*(int *)ptr = strlen(pTable->name);
// Encode name, todo refactor
*(int *)ptr = varDataLen(pTable->name);
ptr = (char *)ptr + sizeof(int);
memcpy(ptr, pTable->name, strlen(pTable->name));
ptr = (char *)ptr + strlen(pTable->name);
memcpy(ptr, varDataVal(pTable->name), varDataLen(pTable->name));
ptr = (char *)ptr + varDataLen(pTable->name);
T_APPEND_MEMBER(ptr, &(pTable->tableId), STableId, uid);
T_APPEND_MEMBER(ptr, &(pTable->tableId), STableId, tid);
T_APPEND_MEMBER(ptr, pTable, STable, superUid);
@ -79,9 +80,12 @@ STable *tsdbDecodeTable(void *cont, int contLen) {
T_READ_MEMBER(ptr, int8_t, pTable->type);
int len = *(int *)ptr;
ptr = (char *)ptr + sizeof(int);
pTable->name = calloc(1, len + 1);
pTable->name = calloc(1, len + VARSTR_HEADER_SIZE);
if (pTable->name == NULL) return NULL;
memcpy(pTable->name, ptr, len);
varDataSetLen(pTable->name, len);
memcpy(pTable->name->data, ptr, len);
ptr = (char *)ptr + len;
T_READ_MEMBER(ptr, int64_t, pTable->tableId.uid);
T_READ_MEMBER(ptr, int32_t, pTable->tableId.tid);
@ -105,8 +109,13 @@ void tsdbFreeEncode(void *cont) {
}
static char* getTagIndexKey(const void* pData) {
STable* table = *(STable**) pData;
return getTupleKey(table->tagVal);
STableIndexElem* elem = (STableIndexElem*) pData;
SDataRow row = elem->pTable->tagVal;
STSchema* pSchema = tsdbGetTableTagSchema(elem->pMeta, elem->pTable);
STColumn* pCol = &pSchema->columns[DEFAULT_TAG_INDEX_COLUMN];
return tdGetRowDataOfCol(row, pCol->type, TD_DATA_ROW_HEAD_SIZE + pCol->offset);
}
int tsdbRestoreTable(void *pHandle, void *cont, int contLen) {
@ -225,36 +234,28 @@ STSchema * tsdbGetTableTagSchema(STsdbMeta *pMeta, STable *pTable) {
}
}
// todo refactor table name definition
int32_t tsdbGetTableTagVal(TsdbRepoT* repo, STableId* id, int32_t colId, int16_t* type, int16_t* bytes, char** val) {
STsdbMeta* pMeta = tsdbGetMeta(repo);
STable* pTable = tsdbGetTableByUid(pMeta, id->uid);
STSchema* pSchema = tsdbGetTableTagSchema(pMeta, pTable);
STColumn* pCol = NULL;
int32_t offset = 0;
for(int32_t col = 0; col < schemaNCols(pSchema); ++col) {
STColumn* p = schemaColAt(pSchema, col);
if (p->colId == colId) {
pCol = p;
break;
}
if (p->type == TSDB_DATA_TYPE_BINARY || p->type == TSDB_DATA_TYPE_NCHAR) {
offset += sizeof(int32_t);
} else {
offset += p->bytes;
}
}
if (pCol == NULL) {
return -1; // No matched tags. Maybe the modification of tags has not been done yet.
}
assert(pCol != NULL);
SDataRow row = (SDataRow)pTable->tagVal;
char* d = tdGetRowDataOfCol(row, pCol->type, TD_DATA_ROW_HEAD_SIZE + offset);
char* d = tdGetRowDataOfCol(row, pCol->type, TD_DATA_ROW_HEAD_SIZE + pCol->offset);
*val = d;
*type = pCol->type;
@ -263,15 +264,22 @@ int32_t tsdbGetTableTagVal(TsdbRepoT* repo, STableId* id, int32_t colId, int16_t
return TSDB_CODE_SUCCESS;
}
int32_t tsdbGetTableName(TsdbRepoT *repo, STableId* id, char** name) {
char* tsdbGetTableName(TsdbRepoT *repo, const STableId* id, int16_t* bytes) {
STsdbMeta* pMeta = tsdbGetMeta(repo);
STable* pTable = tsdbGetTableByUid(pMeta, id->uid);
*name = strndup(pTable->name, TSDB_TABLE_NAME_LEN);
if (*name == NULL) {
return -1;
if (pTable == NULL) {
if (bytes != NULL) {
*bytes = 0;
}
return NULL;
} else {
return 0;
if (bytes != NULL) {
*bytes = varDataLen(pTable->name);
}
return (char*) pTable->name;
}
}
@ -296,7 +304,11 @@ int32_t tsdbCreateTableImpl(STsdbMeta *pMeta, STableCfg *pCfg) {
super->schema = tdDupSchema(pCfg->schema);
super->tagSchema = tdDupSchema(pCfg->tagSchema);
super->tagVal = NULL;
super->name = strdup(pCfg->sname);
// todo refactor extract method
size_t size = strnlen(pCfg->sname, TSDB_TABLE_NAME_LEN);
super->name = malloc(size + VARSTR_HEADER_SIZE);
STR_WITH_SIZE_TO_VARSTR(super->name, pCfg->sname, size);
// index the first tag column
STColumn* pColSchema = schemaColAt(super->tagSchema, 0);
@ -322,7 +334,11 @@ int32_t tsdbCreateTableImpl(STsdbMeta *pMeta, STableCfg *pCfg) {
}
table->tableId = pCfg->tableId;
table->name = strdup(pCfg->name);
size_t size = strnlen(pCfg->name, TSDB_TABLE_NAME_LEN);
table->name = malloc(size + VARSTR_HEADER_SIZE);
STR_WITH_SIZE_TO_VARSTR(table->name, pCfg->name, size);
table->lastKey = 0;
if (IS_CREATE_STABLE(pCfg)) { // TSDB_CHILD_TABLE
table->type = TSDB_CHILD_TABLE;
@ -513,11 +529,14 @@ static int tsdbAddTableIntoIndex(STsdbMeta *pMeta, STable *pTable) {
// NOTE: do not allocate the space for key, since in each skip list node, only keep the pointer to pTable, not the
// actual key value, and the key value will be retrieved during query through the pTable and getTagIndexKey function
SSkipListNode* pNode = calloc(1, headSize + POINTER_BYTES);
SSkipListNode* pNode = calloc(1, headSize + sizeof(STableIndexElem));
pNode->level = level;
SSkipList* list = pSTable->pIndex;
memcpy(SL_GET_NODE_DATA(pNode), &pTable, POINTER_BYTES);
STableIndexElem* elem = (STableIndexElem*) (SL_GET_NODE_DATA(pNode));
elem->pTable = pTable;
elem->pMeta = pMeta;
tSkipListPut(list, pNode);
return 0;
@ -539,7 +558,7 @@ static int tsdbRemoveTableFromIndex(STsdbMeta *pMeta, STable *pTable) {
static int tsdbEstimateTableEncodeSize(STable *pTable) {
int size = 0;
size += T_MEMBER_SIZE(STable, type);
size += sizeof(int) + strlen(pTable->name);
size += sizeof(int) + varDataLen(pTable->name);
size += T_MEMBER_SIZE(STable, tableId);
size += T_MEMBER_SIZE(STable, superUid);
size += T_MEMBER_SIZE(STable, sversion);
@ -556,8 +575,7 @@ static int tsdbEstimateTableEncodeSize(STable *pTable) {
return size;
}
char *getTupleKey(const void * data) {
char *getTSTupleKey(const void * data) {
SDataRow row = (SDataRow)data;
return POINTER_SHIFT(row, TD_DATA_ROW_HEAD_SIZE);
}

56
src/tsdb/src/tsdbRead.c
View File

@ -620,9 +620,13 @@ static void filterDataInDataBlock(STsdbQueryHandle* pQueryHandle, STableCheckInf
if (pCol->info.type != TSDB_DATA_TYPE_BINARY && pCol->info.type != TSDB_DATA_TYPE_NCHAR) {
memmove(pCol->pData, src->pData + bytes * start, bytes * pQueryHandle->realNumOfRows);
} else { // handle the var-string
char* dst = pCol->pData;
// todo refactor, only copy one-by-one
for(int32_t k = start; k < pQueryHandle->realNumOfRows + start; ++k) {
char* p = tdGetColDataOfRow(src, k);
memcpy(pCol->pData + k * bytes, p, varDataTLen(p)); // todo refactor
memcpy(dst, p, varDataTLen(p));
dst += varDataTLen(p);
}
}
@ -1076,17 +1080,13 @@ static int tsdbReadRowsFromCache(SSkipListIterator* pIter, STable* pTable, TSKEY
}
assert(offset != -1); // todo handle error
void *value = tdGetRowDataOfCol(row, pColInfo->info.type, TD_DATA_ROW_HEAD_SIZE + offset);
if (pColInfo->info.type == TSDB_DATA_TYPE_BINARY || pColInfo->info.type == TSDB_DATA_TYPE_NCHAR) {
void *value = tdGetRowDataOfCol(row, pColInfo->info.type, TD_DATA_ROW_HEAD_SIZE + offset);
memcpy(pData, value, varDataTLen(value));
offset += sizeof(int32_t);
} else {
memcpy(pData, dataRowTuple(row) + offset, pColInfo->info.bytes);
offset += pColInfo->info.bytes;
memcpy(pData, value, pColInfo->info.bytes);
}
}
numOfRows++;
@ -1225,8 +1225,8 @@ static int32_t getAllTableIdList(STable* pSuperTable, SArray* list) {
while (tSkipListIterNext(iter)) {
SSkipListNode* pNode = tSkipListIterGet(iter);
STable* t = *(STable**)SL_GET_NODE_DATA(pNode);
taosArrayPush(list, &t->tableId);
STableIndexElem* elem = (STableIndexElem*)(SL_GET_NODE_DATA((SSkipListNode*) pNode));
taosArrayPush(list, &elem->pTable->tableId);
}
tSkipListDestroyIter(iter);
@ -1235,6 +1235,7 @@ static int32_t getAllTableIdList(STable* pSuperTable, SArray* list) {
/**
* convert the result pointer to table id instead of table object pointer
* todo remove it by using callback function to change the final result in-time.
* @param pRes
*/
static void convertQueryResult(SArray* pRes, SArray* pTableList) {
@ -1244,8 +1245,8 @@ static void convertQueryResult(SArray* pRes, SArray* pTableList) {
size_t size = taosArrayGetSize(pTableList);
for (int32_t i = 0; i < size; ++i) { // todo speedup by using reserve space.
STable* pTable = taosArrayGetP(pTableList, i);
taosArrayPush(pRes, &pTable->tableId);
STableIndexElem* elem = taosArrayGet(pTableList, i);
taosArrayPush(pRes, &elem->pTable->tableId);
}
}
@ -1309,9 +1310,14 @@ void filterPrepare(void* expr, void* param) {
pInfo->q = (char*) pCond->arr;
} else {
pInfo->q = calloc(1, pSchema->bytes);
if (pSchema->type == TSDB_DATA_TYPE_BINARY || pSchema->type == TSDB_DATA_TYPE_NCHAR) {
varDataSetLen(pInfo->q, pCond->nLen);
tVariantDump(pCond, varDataVal(pInfo->q), pSchema->type);
} else {
tVariantDump(pCond, pInfo->q, pSchema->type);
}
}
}
typedef struct STableGroupSupporter {
int32_t numOfCols;
@ -1341,16 +1347,16 @@ int32_t tableGroupComparFn(const void *p1, const void *p2, const void *param) {
int32_t bytes = 0;
if (colIndex == TSDB_TBNAME_COLUMN_INDEX) {
f1 = pTable1->name;
f2 = pTable2->name;
f1 = (char*) pTable1->name;
f2 = (char*) pTable2->name;
type = TSDB_DATA_TYPE_BINARY;
bytes = TSDB_TABLE_NAME_LEN;
bytes = TSDB_TABLE_NAME_LEN + VARSTR_HEADER_SIZE;
} else {
f1 = dataRowTuple(pTable1->tagVal);
f2 = dataRowTuple(pTable2->tagVal);
STColumn* pCol = schemaColAt(pTableGroupSupp->pTagSchema, colIndex);
bytes = pCol->bytes;
type = schemaColAt(pTableGroupSupp->pTagSchema, colIndex)->type;
bytes = schemaColAt(pTableGroupSupp->pTagSchema, colIndex)->bytes;
f1 = tdGetRowDataOfCol(pTable1->tagVal, pCol->type, TD_DATA_ROW_HEAD_SIZE + pCol->offset);
f2 = tdGetRowDataOfCol(pTable2->tagVal, pCol->type, TD_DATA_ROW_HEAD_SIZE + pCol->offset);
}
int32_t ret = doCompare(f1, f2, type, bytes);
@ -1429,23 +1435,19 @@ SArray* createTableGroup(SArray* pTableList, STSchema* pTagSchema, SColIndex* pC
bool tSkipListNodeFilterCallback(const void* pNode, void* param) {
tQueryInfo* pInfo = (tQueryInfo*) param;
STable* pTable = *(STable**)(SL_GET_NODE_DATA((SSkipListNode*)pNode));
STableIndexElem* elem = (STableIndexElem*)(SL_GET_NODE_DATA((SSkipListNode*)pNode));
char* val = NULL;
int8_t type = pInfo->sch.type;
if (pInfo->colIndex == TSDB_TBNAME_COLUMN_INDEX) {
val = pTable->name;
val = (char*) elem->pTable->name;
type = TSDB_DATA_TYPE_BINARY;
} else {
STSchema* pTSchema = (STSchema*) pInfo->param; // todo table schema is identical to stable schema??
int32_t offset = pTSchema->columns[pInfo->colIndex].offset;
if (pInfo->sch.type == TSDB_DATA_TYPE_BINARY || pInfo->sch.type == TSDB_DATA_TYPE_NCHAR) {
val = tdGetRowDataOfCol(pTable->tagVal, pInfo->sch.type, TD_DATA_ROW_HEAD_SIZE + offset);
} else {
val = dataRowTuple(pTable->tagVal) + offset;
}
val = tdGetRowDataOfCol(elem->pTable->tagVal, pInfo->sch.type, TD_DATA_ROW_HEAD_SIZE + offset);
}
int32_t ret = 0;
@ -1456,8 +1458,6 @@ bool tSkipListNodeFilterCallback(const void* pNode, void* param) {
ret = pInfo->compare(val, pInfo->q);
}
} else {
// tVariant t = {0};
// tVariantCreateFromBinary(&t, val, (uint32_t)pInfo->sch.bytes, type);
ret = pInfo->compare(val, pInfo->q);
}
@ -1502,7 +1502,7 @@ static int32_t doQueryTableList(STable* pSTable, SArray* pRes, tExprNode* pExpr)
.pExtInfo = pSTable->tagSchema,
};
SArray* pTableList = taosArrayInit(8, POINTER_BYTES);
SArray* pTableList = taosArrayInit(8, sizeof(STableIndexElem));
tExprTreeTraverse(pExpr, pSTable->pIndex, pTableList, &supp);
tExprTreeDestroy(&pExpr, destroyHelper);

4
src/util/src/talgo.c
View File

@ -144,11 +144,11 @@ static void tqsortImpl(void *src, int32_t start, int32_t end, size_t size, const
}
if (leftPartEnd > start) {
tqsortImpl(src, size, start, leftPartEnd, param, comparFn, buf);
tqsortImpl(src, start, leftPartEnd, size, param, comparFn, buf);
}
if (rightPartStart < end) {
tqsortImpl(src, size, rightPartStart, end, param, comparFn, buf);
tqsortImpl(src, rightPartStart, end, size, param, comparFn, buf);
}
}

39
src/util/src/tcompare.c
View File

@ -60,8 +60,15 @@ int32_t compareDoubleVal(const void *pLeft, const void *pRight) {
}
}
int32_t compareStrVal(const void *pLeft, const void *pRight) {
return (int32_t)strcmp(pLeft, pRight);
int32_t compareLenPrefixedStr(const void *pLeft, const void *pRight) {
int32_t len1 = varDataLen(pLeft);
int32_t len2 = varDataLen(pRight);
if (len1 != len2) {
return len1 > len2? 1:-1;
} else {
return (int32_t) strncmp(varDataVal(pLeft), varDataVal(pRight), len1);
}
}
int32_t compareWStrVal(const void *pLeft, const void *pRight) {
@ -267,7 +274,7 @@ __compar_fn_t getComparFunc(int32_t type, int32_t optr) {
} else if (optr == TSDB_RELATION_IN) {
comparFn = compareFindStrInArray;
} else { /* normal relational comparFn */
comparFn = compareStrVal;
comparFn = compareLenPrefixedStr;
}
break;
@ -296,6 +303,7 @@ __compar_fn_t getKeyComparFunc(int32_t keyType) {
switch (keyType) {
case TSDB_DATA_TYPE_TINYINT:
case TSDB_DATA_TYPE_BOOL:
comparFn = compareInt8Val;
break;
case TSDB_DATA_TYPE_SMALLINT:
@ -308,17 +316,13 @@ __compar_fn_t getKeyComparFunc(int32_t keyType) {
case TSDB_DATA_TYPE_TIMESTAMP:
comparFn = compareInt64Val;
break;
case TSDB_DATA_TYPE_BOOL:
comparFn = compareInt32Val;
break;
case TSDB_DATA_TYPE_FLOAT:
case TSDB_DATA_TYPE_DOUBLE:
comparFn = compareDoubleVal;
break;
case TSDB_DATA_TYPE_BINARY:
comparFn = compareStrVal;
comparFn = compareLenPrefixedStr;
break;
case TSDB_DATA_TYPE_NCHAR:
@ -349,13 +353,20 @@ int32_t doCompare(const char* f1, const char* f2, int32_t type, size_t size) {
}
return (ret < 0) ? -1 : 1;
}
default: {
int32_t ret = strncmp(f1, f2, (size_t)size);
if (ret == 0) {
return ret;
}
default: { // todo refactor
tstr* t1 = (tstr*) f1;
tstr* t2 = (tstr*) f2;
return (ret < 0) ? -1 : 1;
if (t1->len != t2->len) {
return t1->len > t2->len? 1:-1;
} else {
int32_t ret = strncmp(t1->data, t2->data, t1->len);
if (ret == 0) {
return 0;
} else {
return ret < 0? -1:1;
}
}
}
}
}

2
src/util/src/tconfig.c
View File

@ -236,7 +236,7 @@ void taosReadGlobalLogCfg() {
int olen, vlen;
char fileName[PATH_MAX] = {0};
mdebugFlag = 135;
mDebugFlag = 135;
sdbDebugFlag = 135;
wordexp_t full_path;

7
src/vnode/src/vnodeMain.c
View File

@ -267,7 +267,7 @@ void vnodeRelease(void *pVnodeRaw) {
assert(refCount >= 0);
if (refCount > 0) {
vTrace("vgId:%d, release vnode, refCount:%d", pVnode, vgId, refCount);
vTrace("vgId:%d, release vnode, refCount:%d", vgId, refCount);
return;
}
@ -693,8 +693,9 @@ static bool vnodeReadVersion(SVnodeObj *pVnode) {
sprintf(versionFile, "%s/vnode%d/version.json", tsVnodeDir, pVnode->vgId);
FILE *fp = fopen(versionFile, "r");
if (!fp) {
vTrace("vgId:%d, failed to open version file:%s error:%s", pVnode->vgId,
versionFile, strerror(errno));
if (errno != ENOENT) {
vError("vgId:%d, failed to open version file:%s error:%s", pVnode->vgId, versionFile, strerror(errno));
}
return false;
}

2
src/vnode/src/vnodeWrite.c
View File

@ -239,7 +239,7 @@ static int32_t vnodeProcessDropStableMsg(SVnodeObj *pVnode, void *pCont, SRspRet
code = tsdbDropTable(pVnode->tsdb, stableId);
vTrace("vgId:%d, stable:%s, drop stable result:%s", pVnode, pTable->tableId, tstrerror(code));
vTrace("vgId:%d, stable:%s, drop stable result:%s", pVnode->vgId, pTable->tableId, tstrerror(code));
return code;
}

9
tests/script/general/cache/restart_metrics.sim vendored
View File

@ -27,11 +27,15 @@ sql create table $tb using $mt tags( "1" )
sql insert into $tb values (now, 1)
sql select * from $tb
print ===>rows $rows, data $data01
#print ===>rows $rows, data $data01
if $rows != 1 then
print expect 1, actual: $rows
return -1
endi
if $data01 != 1 then
print expect 1 actual: $data01
return -1
endi
@ -67,7 +71,9 @@ print ===>rows $rows, data $data01
if $rows != 1 then
return -1
endi
if $data01 != 1 then
print expect 1, actual $data01
return -1
endi
@ -80,6 +86,7 @@ if $data01 != 1 then
return -1
endi
if $data02 != 3 then
print expect 3 actual: $data02
return -1
endi

6
tests/script/general/db/delete_reusevnode2.sim
View File

@ -1,7 +1,5 @@
system sh/stop_dnodes.sh
system sh/deploy.sh -n dnode1 -i 1
system sh/cfg.sh -n dnode1 -c numOfTotalVnodes -v 10
system sh/cfg.sh -n dnode2 -c numOfTotalVnodes -v 10
system sh/cfg.sh -n dnode3 -c numOfTotalVnodes -v 10
@ -36,7 +34,7 @@ while $i < 10
sql create table $tb1 using st tags(1)
sql insert into $tb1 values(now, 1);
$tb1 = $tb . 5
$tb1 = $tb . 4
sql create table $tb1 using st tags(1)
sql insert into $tb1 values(now, 1);
@ -58,7 +56,7 @@ while $i < 10
sql drop table st
sleep 1000
sleep 2000
print times $i
$i = $i + 1

2
tests/script/general/field/bigint.sim
View File

@ -132,7 +132,9 @@ endi
print =============== step6
sql select count(tbcol), avg(tbcol), sum(tbcol), min(tbcol), max(tbcol), first(tbcol), last(tbcol) from $mt where tbcol = 1 group by tgcol
print $data00 $data01 $data02 $data03 $data04 $data05 $data06
print $data10 $data11 $data12 $data13 $data14 $data15 $data16
if $data00 != 100 then
print expect 100, actual $data00
return -1
endi

2
tests/script/general/parser/lastrow_query.sim
View File

@ -43,9 +43,11 @@ if $data07 != 1 then
return -1
endi
if $data08 != BINARY then
print expect BINARY actual: $data08
return -1
endi
if $data09 != NCHAR then
print expect NCHAR actual: $data09
return -1
endi

164
tests/script/general/parser/testSuite.sim
View File

@ -7,107 +7,91 @@
#run general/parser/auto_create_tb.sim
#sleep 2000
#run general/parser/auto_create_tb_drop_tb.sim
# sleep 2000
run general/parser/binary_escapeCharacter.sim
sleep 2000
run general/parser/bug.sim
sleep 2000
run general/parser/col_arithmetic_operation.sim
sleep 2000
run general/parser/columnValue_bigint.sim
run general/parser/columnValue.sim
sleep 2000
run general/parser/columnValue_bool.sim
run general/parser/commit.sim
sleep 2000
run general/parser/columnValue_double.sim
run general/parser/create_db.sim
sleep 2000
run general/parser/columnValue_float.sim
run general/parser/create_mt.sim
sleep 2000
run general/parser/columnValue_int.sim
run general/parser/create_tb.sim
sleep 2000
run general/parser/dbtbnameValidate.sim
sleep 2000
run general/parser/import_commit1.sim
sleep 2000
run general/parser/import_commit2.sim
sleep 2000
run general/parser/import_commit3.sim
sleep 2000
run general/parser/insert_tb.sim
sleep 2000
run general/parser/first_last.sim
sleep 2000
run general/parser/import_file.sim
sleep 2000
run general/parser/fill.sim
sleep 2000
run general/parser/fill_stb.sim
sleep 2000
run general/parser/tags_dynamically_specifiy.sim
sleep 2000
run general/parser/interp.sim
run general/parser/lastrow.sim
sleep 2000
run general/parser/limit.sim
sleep 2000
run general/parser/limit1.sim
sleep 2000
run general/parser/limit1_tblocks100.sim
sleep 2000
run general/parser/limit2.sim
sleep 2000
run general/parser/mixed_blocks.sim
sleep 2000
run general/parser/nchar.sim
sleep 2000
run general/parser/null_char.sim
sleep 2000
run general/parser/selectResNum.sim
sleep 2000
run general/parser/select_across_vnodes.sim
sleep 2000
# sleep 2000
# run general/parser/col_arithmetic_operation.sim
# sleep 2000
# run general/parser/columnValue.sim
# sleep 2000
# run general/parser/commit.sim
# run general/parser/create_db.sim
# sleep 2000
# run general/parser/create_mt.sim
# sleep 2000
# run general/parser/create_tb.sim
# sleep 2000
# run general/parser/dbtbnameValidate.sim
# sleep 2000
# run general/parser/fill.sim
# sleep 2000
# run general/parser/fill_stb.sim
# sleep 2000
# run general/parser/first_last.sim
# sleep 2000
# run general/parser/import_commit1.sim
# sleep 2000
# run general/parser/import_commit2.sim
# sleep 2000
# run general/parser/import_commit3.sim
# sleep 2000
# run general/parser/import_file.sim
# sleep 2000
# run general/parser/insert_tb.sim
# sleep 2000
# run general/parser/tags_dynamically_specifiy.sim
# sleep 2000
# run general/parser/interp.sim
# run general/parser/lastrow.sim
# sleep 2000
# run general/parser/limit.sim
# sleep 2000
# run general/parser/limit1.sim
# sleep 2000
# run general/parser/limit1_tblocks100.sim
# sleep 2000
# run general/parser/limit2.sim
# sleep 2000
# run general/parser/mixed_blocks.sim
# sleep 2000
# run general/parser/nchar.sim
# sleep 2000
# run general/parser/null_char.sim
# sleep 2000
# run general/parser/selectResNum.sim
# sleep 2000
# run general/parser/select_across_vnodes.sim
# sleep 2000
run general/parser/select_from_cache_disk.sim
sleep 2000
# run general/parser/set_tag_vals.sim
# sleep 2000
# run general/parser/single_row_in_tb.sim
# sleep 2000
# run general/parser/slimit.sim
# sleep 2000
run general/parser/set_tag_vals.sim
sleep 2000
run general/parser/single_row_in_tb.sim
sleep 2000
run general/parser/slimit.sim
sleep 2000
run general/parser/slimit1.sim
sleep 2000
run general/parser/slimit1_query.sim
run general/parser/slimit_alter_tags.sim
sleep 2000
# run general/parser/slimit_alter_tags.sim
# sleep 2000
# run general/parser/stream_on_sys.sim
# sleep 2000
# run general/parser/stream.sim
# sleep 2000
# run general/parser/tbnameIn.sim
# sleep 2000
# run general/parser/where.sim
# sleep 2000
# #run general/parser/repeatAlter.sim
# sleep 2000
# #run general/parser/repeatStream.sim
# sleep 2000
# run general/parser/join.sim
# run general/parser/join_multivnode.sim
# run general/parser/projection_limit_offset.sim
# sleep 2000
# run general/parser/select_with_tags.sim
# run general/parser/groupby.sim
run general/parser/stream_on_sys.sim
sleep 2000
run general/parser/stream.sim
sleep 2000
run general/parser/tbnameIn.sim
sleep 2000
run general/parser/where.sim
sleep 2000
#run general/parser/repeatAlter.sim
sleep 2000
#run general/parser/repeatStream.sim
sleep 2000
run general/parser/join.sim
sleep 2000
run general/parser/join_multivnode.sim
sleep 2000
run general/parser/projection_limit_offset.sim
sleep 2000
run general/parser/select_with_tags.sim
sleep 2000
run general/parser/groupby.sim

1
tests/script/general/tag/bool.sim
View File

@ -123,6 +123,7 @@ if $rows != 100 then
endi
sql select * from $mt where tgcol = 1
if $rows != 100 then
print expect 100, actual:$rows
return -1
endi
sql select * from $mt where tgcol <> 1

1
tests/script/general/tag/filter.sim
View File

@ -123,6 +123,7 @@ print =============== step14
sql select count(tbcol) as c from $mt where ts > 1000 group by tgcol
print $data00 $data01 $data02 $data03 $data04 $data05 $data06
if $data00 != 100 then
print expect 100, actual $data00
return -1
endi

3
tests/script/test.sh
View File

@ -99,7 +99,8 @@ echo "cDebugFlag 135" >> $TAOS_CFG
echo "httpDebugFlag 135" >> $TAOS_CFG
echo "monitorDebugFlag 135" >> $TAOS_CFG
echo "udebugFlag 135" >> $TAOS_CFG
echo "clog 0" >> $TAOS_CFG
echo "tablemetakeeptimer 5" >> $TAOS_CFG
echo "wal 0" >> $TAOS_CFG
echo "asyncLog 0" >> $TAOS_CFG
echo "locale en_US.UTF-8" >> $TAOS_CFG
echo " " >> $TAOS_CFG

12
tests/script/unique/mnode/mgmt22.sim
View File

@ -8,7 +8,7 @@ system sh/cfg.sh -n dnode2 -c numOfMPeers -v 2
system sh/cfg.sh -n dnode3 -c numOfMPeers -v 2
print ============== step1
system sh/exec_up.sh -n dnode1 -s start -t
system sh/exec_up.sh -n dnode1 -s start
sleep 3000
sql connect
@ -20,7 +20,7 @@ if $data2_1 != master then
endi
print ============== step2
system sh/exec_up.sh -n dnode2 -s start -t
system sh/exec_up.sh -n dnode2 -s start
sql create dnode $hostname2
$x = 0
@ -45,10 +45,6 @@ print ============== step3
sql_error drop dnode $hostname1 -x error1
print should not drop master
system sh/exec_up.sh -n dnode1 -s stop -x SIGINT
system sh/exec_up.sh -n dnode2 -s stop -x SIGINT
return
print ============== step4
system sh/exec_up.sh -n dnode1 -s stop -x SIGINT
sql_error show mnodes
@ -83,7 +79,7 @@ if $data2_2 != slave then
endi
print ============== step7
system sh/exec_up.sh -n dnode3 -s start -t
system sh/exec_up.sh -n dnode3 -s start
sql create dnode $hostname3
sleep 5000
@ -105,7 +101,7 @@ endi
if $data2_2 != slave then
goto show7
endi
if $data3_3 != NULL then
if $data3_3 != null then
goto show7
endi

14
tests/script/unique/vnode/replica2_basic2.sim
View File

@ -27,6 +27,7 @@ system sh/cfg.sh -n dnode4 -c numOfTotalVnodes -v 4
print ========= start dnodes
system sh/exec_up.sh -n dnode1 -s start
sleep 3000
sql connect
sql create dnode $hostname2
system sh/exec_up.sh -n dnode2 -s start
@ -89,6 +90,19 @@ if $data2_3 != 4 then
return -1
endi
if $data4_1 != ready then
print dnode1 status should ready but is $data4_1
return -1
endi
if $data4_2 != ready then
return -1
endi
if $data4_3 != ready then
return -1
endi
print ========= step2
sql insert into d1.t1 values(now, 2)
sql insert into d2.t2 values(now, 2)