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

:wqMerge branch 'develop' into hotfix/py-testcase-refine

This commit is contained in:
Shuduo Sang 2020-05-11 13:48:43 +08:00
parent 8fd8af6894 fead5b9038
commit 0ba839e18a
26 changed files with 485 additions and 490 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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 1. sudo apt install
build-essential cmake net-tools python-pip python-setuptools python3-pip 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 2. git clone <https://github.com/taosdata/TDengine>; cd TDengine

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) { static void last_function_f(SQLFunctionCtx *pCtx, int32_t index) {
if (pCtx->order == TSDB_ORDER_ASC) { assert(pCtx->order != TSDB_ORDER_ASC);
return;
}
void *pData = GET_INPUT_CHAR_INDEX(pCtx, index); void *pData = GET_INPUT_CHAR_INDEX(pCtx, index);
if (pCtx->hasNull && isNull(pData, pCtx->inputType)) { if (pCtx->hasNull && isNull(pData, pCtx->inputType)) {
return; return;
@ -2912,7 +2909,7 @@ static void leastsquares_finalizer(SQLFunctionCtx *pCtx) {
} }
static void date_col_output_function(SQLFunctionCtx *pCtx) { static void date_col_output_function(SQLFunctionCtx *pCtx) {
if (pCtx->scanFlag == SUPPLEMENTARY_SCAN) { if (pCtx->scanFlag == REVERSE_SCAN) {
return; return;
} }
@ -2969,11 +2966,12 @@ static void tag_project_function(SQLFunctionCtx *pCtx) {
char* output = pCtx->aOutputBuf; char* output = pCtx->aOutputBuf;
if (pCtx->tag.nType == TSDB_DATA_TYPE_BINARY || pCtx->tag.nType == TSDB_DATA_TYPE_NCHAR) { if (pCtx->tag.nType == TSDB_DATA_TYPE_BINARY || pCtx->tag.nType == TSDB_DATA_TYPE_NCHAR) {
*(int16_t*) output = pCtx->tag.nLen; varDataSetLen(output, pCtx->tag.nLen);
output += VARSTR_HEADER_SIZE; tVariantDump(&pCtx->tag, varDataVal(output), pCtx->outputType);
} else {
tVariantDump(&pCtx->tag, output, pCtx->outputType);
} }
tVariantDump(&pCtx->tag, output, pCtx->outputType);
pCtx->aOutputBuf += pCtx->outputBytes; 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->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 = calloc(1, sizeof(tExprNode));
(*pExpr)->nodeType = TSQL_NODE_VALUE; (*pExpr)->nodeType = TSQL_NODE_VALUE;
(*pExpr)->pVal = calloc(1, sizeof(tVariant)); (*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); pMsg += sizeof(pQueryInfo->type);
// todo valid the vgroupId at the client side // todo valid the vgroupId at the client side
if (UTIL_TABLE_IS_SUPERTABLE(pQueryInfo->pTableMetaInfo[0])) { STableMetaInfo* pTableMetaInfo = tscGetMetaInfo(pQueryInfo, 0);
SVgroupsInfo* pVgroupInfo = pQueryInfo->pTableMetaInfo[0]->vgroupList;
assert(pVgroupInfo->numOfVgroups == 1); // todo fix me
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 { } else {
STableMeta* pTableMeta = pQueryInfo->pTableMetaInfo[0]->pTableMeta; STableMeta* pTableMeta = pTableMetaInfo->pTableMeta;
pRetrieveMsg->header.vgId = htonl(pTableMeta->vgroupInfo.vgId); 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; return;
} else { // reach the maximum retry count, abort } else { // reach the maximum retry count, abort
atomic_val_compare_exchange_32(&pState->code, TSDB_CODE_SUCCESS, numOfRows); 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, tscError("%p sub:%p retrieve failed,code:%s,orderOfSub:%d failed.no more retry,set global code:%s", pPObj, pSql,
numOfRows, subqueryIndex, tstrerror(pState->code)); tstrerror(numOfRows), subqueryIndex, tstrerror(pState->code));
} }
} }
@ -1450,7 +1450,7 @@ void tscHandleSubqueryError(SRetrieveSupport *trsupport, SSqlObj *pSql, int numO
} }
// all subqueries are failed // 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; pPObj->res.code = pState->code;
// release allocated resource // 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; pTableMetaInfo->pTableMeta = pTableMeta;
if (vgroupList != NULL) { if (vgroupList != NULL) {
assert(vgroupList->numOfVgroups == 1); // todo fix me
size_t size = sizeof(SVgroupsInfo) + sizeof(SCMVgroupInfo) * vgroupList->numOfVgroups; size_t size = sizeof(SVgroupsInfo) + sizeof(SCMVgroupInfo) * vgroupList->numOfVgroups;
pTableMetaInfo->vgroupList = malloc(size); pTableMetaInfo->vgroupList = malloc(size);
memcpy(pTableMetaInfo->vgroupList, vgroupList, 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_BINARY:
case TSDB_DATA_TYPE_NCHAR: case TSDB_DATA_TYPE_NCHAR:
return POINTER_SHIFT(row, *(VarDataOffsetT *)POINTER_SHIFT(row, offset)); return POINTER_SHIFT(row, *(VarDataOffsetT *)POINTER_SHIFT(row, offset));
break;
default: default:
return POINTER_SHIFT(row, offset); return POINTER_SHIFT(row, offset);
break;
} }
} }

5
src/inc/taosdef.h
View File

@ -36,6 +36,11 @@ extern "C" {
typedef int32_t VarDataOffsetT; typedef int32_t VarDataOffsetT;
typedef int16_t VarDataLenT; typedef int16_t VarDataLenT;
typedef struct tstr {
VarDataLenT len;
char data[];
} tstr;
#define VARSTR_HEADER_SIZE sizeof(VarDataLenT) #define VARSTR_HEADER_SIZE sizeof(VarDataLenT)
#define varDataLen(v) ((VarDataLenT *)(v))[0] #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); void tsdbClearTableCfg(STableCfg *config);
int32_t tsdbGetTableTagVal(TsdbRepoT *repo, STableId* id, int32_t colId, int16_t *type, int16_t *bytes, char **val); 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 tsdbCreateTable(TsdbRepoT *repo, STableCfg *pCfg);
int tsdbDropTable(TsdbRepoT *pRepo, STableId tableId); int tsdbDropTable(TsdbRepoT *pRepo, STableId tableId);

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

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

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

@ -104,7 +104,7 @@ extern "C" {
enum { enum {
MASTER_SCAN = 0x0u, MASTER_SCAN = 0x0u,
SUPPLEMENTARY_SCAN = 0x1u, REVERSE_SCAN = 0x1u,
REPEAT_SCAN = 0x2u, //repeat scan belongs to the master scan REPEAT_SCAN = 0x2u, //repeat scan belongs to the master scan
FIRST_STAGE_MERGE = 0x10u, FIRST_STAGE_MERGE = 0x10u,
SECONDARY_STAGE_MERGE = 0x20u, 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) { if (cond.start != NULL) {
int32_t optr = cond.start->optr; int32_t optr = cond.start->optr;
if (optr == TSDB_RELATION_EQUAL) { if (optr == TSDB_RELATION_EQUAL) { // equals
while(tSkipListIterNext(iter)) { while(tSkipListIterNext(iter)) {
SSkipListNode* pNode = tSkipListIterGet(iter); SSkipListNode* pNode = tSkipListIterGet(iter);
int32_t ret = pQueryInfo->compare(SL_GET_NODE_KEY(pSkipList, pNode), cond.start->v); int32_t ret = pQueryInfo->compare(SL_GET_NODE_KEY(pSkipList, pNode), cond.start->v);
if (ret == 0) { if (ret != 0) {
taosArrayPush(result, SL_GET_NODE_DATA(pNode));
} else {
break; 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; bool comp = true;
int32_t ret = 0; int32_t ret = 0;
@ -582,8 +582,33 @@ static void tQueryIndexColumn(SSkipList* pSkipList, tQueryInfo* pQueryInfo, SArr
comp = false; comp = false;
} }
} }
} else if (optr == TSDB_RELATION_NOT_EQUAL) { } else if (optr == TSDB_RELATION_NOT_EQUAL) { // not equal
assert(0); 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 { } else {
assert(0); assert(0);
} }
@ -773,8 +798,6 @@ static void tSQLBinaryTraverseOnSkipList(tExprNode *pExpr, SArray *pResult, SSki
tSkipListDestroyIter(iter); tSkipListDestroyIter(iter);
} }
static void tQueryIndexlessColumn(SSkipList* pSkipList, tQueryInfo* pQueryInfo, SArray* result) { static void tQueryIndexlessColumn(SSkipList* pSkipList, tQueryInfo* pQueryInfo, SArray* result) {
SSkipListIterator* iter = tSkipListCreateIter(pSkipList); SSkipListIterator* iter = tSkipListCreateIter(pSkipList);
@ -785,7 +808,7 @@ static void tQueryIndexlessColumn(SSkipList* pSkipList, tQueryInfo* pQueryInfo,
char* pTable = SL_GET_NODE_DATA(pNode); char* pTable = SL_GET_NODE_DATA(pNode);
//todo refactor: //todo refactor:
char* name = (*(STable**) pTable)->name; tstr* name = ((STableIndexElem*) pTable)->pTable->name;
// char* name = NULL; // char* name = NULL;
// tsdbGetTableName(tsdb, pTable, &name); // tsdbGetTableName(tsdb, pTable, &name);

405
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 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_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_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))) #define GET_QINFO_ADDR(x) ((void *)((char *)(x)-offsetof(SQInfo, runtimeEnv)))
@ -96,11 +96,6 @@ typedef struct {
STSCursor cur; STSCursor cur;
} SQueryStatusInfo; } SQueryStatusInfo;
typedef struct SGroupItem {
STableId id;
STableQueryInfo* info;
} SGroupItem;
static void setQueryStatus(SQuery *pQuery, int8_t status); static void setQueryStatus(SQuery *pQuery, int8_t status);
static bool isIntervalQuery(SQuery *pQuery) { return pQuery->intervalTime > 0; } 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 createTableQueryInfo(SQInfo *pQInfo);
static void buildTagQueryResult(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); static int32_t flushFromResultBuf(SQInfo *pQInfo);
bool getNeighborPoints(SQInfo *pQInfo, void *pMeterObj, SPointInterpoSupporter *pPointInterpSupporter) { bool getNeighborPoints(SQInfo *pQInfo, void *pMeterObj, SPointInterpoSupporter *pPointInterpSupporter) {
@ -428,9 +423,10 @@ static bool hasNullValue(SQuery *pQuery, int32_t col, SDataBlockInfo *pDataBlock
return false; return false;
} }
*pColStatis = NULL;
if (pStatis != NULL) { if (pStatis != NULL) {
*pColStatis = getStatisInfo(pQuery, pStatis, pDataBlockInfo, col); *pColStatis = getStatisInfo(pQuery, pStatis, pDataBlockInfo, col);
} else {
*pColStatis = NULL;
} }
if ((*pColStatis) != NULL && (*pColStatis)->numOfNull == 0) { if ((*pColStatis) != NULL && (*pColStatis)->numOfNull == 0) {
@ -620,8 +616,8 @@ static void doCheckQueryCompleted(SQueryRuntimeEnv *pRuntimeEnv, TSKEY lastKey,
} }
// query completed // query completed
if ((lastKey >= pQuery->window.ekey && QUERY_IS_ASC_QUERY(pQuery)) || if ((lastKey >= pQuery->current->win.ekey && QUERY_IS_ASC_QUERY(pQuery)) ||
(lastKey <= pQuery->window.ekey && !QUERY_IS_ASC_QUERY(pQuery))) { (lastKey <= pQuery->current->win.ekey && !QUERY_IS_ASC_QUERY(pQuery))) {
closeAllTimeWindow(pWindowResInfo); closeAllTimeWindow(pWindowResInfo);
pWindowResInfo->curIndex = pWindowResInfo->size - 1; pWindowResInfo->curIndex = pWindowResInfo->size - 1;
@ -661,22 +657,22 @@ static void doCheckQueryCompleted(SQueryRuntimeEnv *pRuntimeEnv, TSKEY lastKey,
setQueryStatus(pQuery, QUERY_RESBUF_FULL); 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); assert(pWindowResInfo->prevSKey != 0);
} }
static int32_t getNumOfRowsInTimeWindow(SQuery *pQuery, SDataBlockInfo *pDataBlockInfo, TSKEY *pPrimaryColumn, static int32_t getNumOfRowsInTimeWindow(SQuery *pQuery, SDataBlockInfo *pDataBlockInfo, TSKEY *pPrimaryColumn,
int32_t startPos, TSKEY ekey, __block_search_fn_t searchFn, int32_t startPos, TSKEY ekey, __block_search_fn_t searchFn, bool updateLastKey) {
bool updateLastKey) {
assert(startPos >= 0 && startPos < pDataBlockInfo->rows); assert(startPos >= 0 && startPos < pDataBlockInfo->rows);
int32_t num = -1; int32_t num = -1;
int32_t order = pQuery->order.order; int32_t order = pQuery->order.order;
int32_t step = GET_FORWARD_DIRECTION_FACTOR(order); int32_t step = GET_FORWARD_DIRECTION_FACTOR(order);
STableQueryInfo* item = pQuery->current;
if (QUERY_IS_ASC_QUERY(pQuery)) { if (QUERY_IS_ASC_QUERY(pQuery)) {
if (ekey < pDataBlockInfo->window.ekey) { if (ekey < pDataBlockInfo->window.ekey) {
num = getForwardStepsInBlock(pDataBlockInfo->rows, searchFn, ekey, startPos, order, pPrimaryColumn); 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]); assert(ekey < pPrimaryColumn[startPos]);
} else { } else {
if (updateLastKey) { if (updateLastKey) {
pQuery->lastKey = pPrimaryColumn[startPos + (num - 1)] + step; item->lastKey = pPrimaryColumn[startPos + (num - 1)] + step;
} }
} }
} else { } else {
num = pDataBlockInfo->rows - startPos; num = pDataBlockInfo->rows - startPos;
if (updateLastKey) { if (updateLastKey) {
pQuery->lastKey = pDataBlockInfo->window.ekey + step; item->lastKey = pDataBlockInfo->window.ekey + step;
} }
} }
} else { // desc } else { // desc
@ -700,13 +696,13 @@ static int32_t getNumOfRowsInTimeWindow(SQuery *pQuery, SDataBlockInfo *pDataBlo
assert(ekey > pPrimaryColumn[startPos]); assert(ekey > pPrimaryColumn[startPos]);
} else { } else {
if (updateLastKey) { if (updateLastKey) {
pQuery->lastKey = pPrimaryColumn[startPos - (num - 1)] + step; item->lastKey = pPrimaryColumn[startPos - (num - 1)] + step;
} }
} }
} else { } else {
num = startPos + 1; num = startPos + 1;
if (updateLastKey) { 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; 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) { SArray *pDataBlock) {
char *dataBlock = NULL; char *dataBlock = NULL;
SQuery *pQuery = pRuntimeEnv->pQuery; SQuery *pQuery = pRuntimeEnv->pQuery;
@ -867,20 +877,7 @@ char *getDataBlocks(SQueryRuntimeEnv *pRuntimeEnv, SArithmeticSupport *sas, int3
if (TSDB_COL_IS_TAG(pCol->flag) || pDataBlock == NULL) { if (TSDB_COL_IS_TAG(pCol->flag) || pDataBlock == NULL) {
dataBlock = NULL; dataBlock = NULL;
} else { } else {
/* dataBlock = getDataBlockImpl(pDataBlock, pCol->colId);
* 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;
}
}
} }
} }
@ -904,8 +901,8 @@ static void blockwiseApplyFunctions(SQueryRuntimeEnv *pRuntimeEnv, SDataStatis *
SQuery * pQuery = pRuntimeEnv->pQuery; SQuery * pQuery = pRuntimeEnv->pQuery;
SColumnInfoData *pColInfo = NULL; SColumnInfoData *pColInfo = NULL;
TSKEY * primaryKeyCol = NULL;
TSKEY *primaryKeyCol = NULL;
if (pDataBlock != NULL) { if (pDataBlock != NULL) {
pColInfo = taosArrayGet(pDataBlock, 0); pColInfo = taosArrayGet(pDataBlock, 0);
primaryKeyCol = (TSKEY *)(pColInfo->pData); primaryKeyCol = (TSKEY *)(pColInfo->pData);
@ -919,7 +916,7 @@ static void blockwiseApplyFunctions(SQueryRuntimeEnv *pRuntimeEnv, SDataStatis *
SDataStatis *tpField = NULL; SDataStatis *tpField = NULL;
bool hasNull = hasNullValue(pQuery, k, pDataBlockInfo, pStatis, &tpField); 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, setExecParams(pQuery, &pCtx[k], dataBlock, primaryKeyCol, pDataBlockInfo->rows, functionId, tpField, hasNull,
&sasArray[k], pRuntimeEnv->scanFlag); &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) { static bool functionNeedToExecute(SQueryRuntimeEnv *pRuntimeEnv, SQLFunctionCtx *pCtx, int32_t functionId) {
SResultInfo *pResInfo = GET_RES_INFO(pCtx); SResultInfo *pResInfo = GET_RES_INFO(pCtx);
SQuery* pQuery = pRuntimeEnv->pQuery;
if (pResInfo->complete || functionId == TSDB_FUNC_TAG_DUMMY || functionId == TSDB_FUNC_TS_DUMMY) { if (pResInfo->complete || functionId == TSDB_FUNC_TAG_DUMMY || functionId == TSDB_FUNC_TS_DUMMY) {
return false; 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 // in the supplementary scan, only the following functions need to be executed
if (IS_REVERSE_SCAN(pRuntimeEnv) && if (IS_REVERSE_SCAN(pRuntimeEnv)) {// && (functionId == TSDB_FUNC_TAG || functionId == TSDB_FUNC_TS)) {
!(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)) {
return false; return false;
} }
@ -1115,8 +1117,9 @@ static void rowwiseApplyFunctions(SQueryRuntimeEnv *pRuntimeEnv, SDataStatis *pS
SQLFunctionCtx *pCtx = pRuntimeEnv->pCtx; SQLFunctionCtx *pCtx = pRuntimeEnv->pCtx;
SQuery *pQuery = pRuntimeEnv->pQuery; 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); bool groupbyStateValue = isGroupbyNormalCol(pQuery->pGroupbyExpr);
SArithmeticSupport *sasArray = calloc((size_t)pQuery->numOfOutput, sizeof(SArithmeticSupport)); SArithmeticSupport *sasArray = calloc((size_t)pQuery->numOfOutput, sizeof(SArithmeticSupport));
@ -1134,7 +1137,7 @@ static void rowwiseApplyFunctions(SQueryRuntimeEnv *pRuntimeEnv, SDataStatis *pS
SDataStatis *pColStatis = NULL; SDataStatis *pColStatis = NULL;
bool hasNull = hasNullValue(pQuery, k, pDataBlockInfo, pStatis, &pColStatis); 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, setExecParams(pQuery, &pCtx[k], dataBlock, primaryKeyCol, pDataBlockInfo->rows, functionId, pColStatis, hasNull,
&sasArray[k], pRuntimeEnv->scanFlag); &sasArray[k], pRuntimeEnv->scanFlag);
@ -1143,7 +1146,8 @@ static void rowwiseApplyFunctions(SQueryRuntimeEnv *pRuntimeEnv, SDataStatis *pS
// set the input column data // set the input column data
for (int32_t k = 0; k < pQuery->numOfFilterCols; ++k) { for (int32_t k = 0; k < pQuery->numOfFilterCols; ++k) {
SSingleColumnFilterInfo *pFilterInfo = &pQuery->pFilterInfo[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); 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; int32_t j = 0;
TSKEY lastKey = -1; int32_t offset = -1;
for (j = 0; j < pDataBlockInfo->rows; ++j) { 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) { if (pRuntimeEnv->pTSBuf != NULL) {
int32_t r = doTSJoinFilter(pRuntimeEnv, offset); int32_t r = doTSJoinFilter(pRuntimeEnv, offset);
@ -1194,7 +1198,6 @@ static void rowwiseApplyFunctions(SQueryRuntimeEnv *pRuntimeEnv, SDataStatis *pS
SWindowStatus *pStatus = getTimeWindowResStatus(pWindowResInfo, curTimeWindow(pWindowResInfo)); SWindowStatus *pStatus = getTimeWindowResStatus(pWindowResInfo, curTimeWindow(pWindowResInfo));
doRowwiseApplyFunctions(pRuntimeEnv, pStatus, &win, offset); doRowwiseApplyFunctions(pRuntimeEnv, pStatus, &win, offset);
lastKey = ts;
STimeWindow nextWin = win; STimeWindow nextWin = win;
int32_t index = pWindowResInfo->curIndex; int32_t index = pWindowResInfo->curIndex;
@ -1223,17 +1226,14 @@ static void rowwiseApplyFunctions(SQueryRuntimeEnv *pRuntimeEnv, SDataStatis *pS
} else { // other queries } else { // other queries
// decide which group this rows belongs to according to current state value // decide which group this rows belongs to according to current state value
if (groupbyStateValue) { 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 if (ret != TSDB_CODE_SUCCESS) { // null data, too many state code
continue; continue;
} }
} }
// update the lastKey
lastKey = primaryKeyCol[offset];
// all startOffset are identical // all startOffset are identical
offset -= pCtx[0].startOffset; 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 // todo refactor: extract method
for(int32_t i = 0; i < pQuery->numOfOutput; ++i) { 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, static int32_t tableApplyFunctionsOnBlock(SQueryRuntimeEnv *pRuntimeEnv, SDataBlockInfo *pDataBlockInfo,
SDataStatis *pStatis, __block_search_fn_t searchFn, SDataStatis *pStatis, __block_search_fn_t searchFn, SArray *pDataBlock) {
SWindowResInfo *pWindowResInfo, SArray *pDataBlock) {
SQuery *pQuery = pRuntimeEnv->pQuery; SQuery *pQuery = pRuntimeEnv->pQuery;
STableQueryInfo* pTableQInfo = pQuery->current;
SWindowResInfo* pWindowResInfo = &pRuntimeEnv->windowResInfo;
if (pQuery->numOfFilterCols > 0 || pRuntimeEnv->pTSBuf != NULL || isGroupbyNormalCol(pQuery->pGroupbyExpr)) { if (pQuery->numOfFilterCols > 0 || pRuntimeEnv->pTSBuf != NULL || isGroupbyNormalCol(pQuery->pGroupbyExpr)) {
rowwiseApplyFunctions(pRuntimeEnv, pStatis, pDataBlockInfo, pWindowResInfo, pDataBlock); rowwiseApplyFunctions(pRuntimeEnv, pStatis, pDataBlockInfo, pWindowResInfo, pDataBlock);
} else { } 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; 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); 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) { static void setScanLimitationByResultBuffer(SQuery *pQuery) {
if (isTopBottomQuery(pQuery)) { if (isTopBottomQuery(pQuery)) {
pQuery->checkBuffer = 0; pQuery->checkBuffer = 0;
@ -2327,7 +2262,7 @@ static void getNextTimeWindow(SQuery *pQuery, STimeWindow *pTimeWindow) {
pTimeWindow->ekey = pTimeWindow->skey + (pQuery->intervalTime - 1); 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; SQuery *pQuery = pRuntimeEnv->pQuery;
uint32_t r = 0; 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), qTrace("QInfo:%p data block ignored, brange:%" PRId64 "-%" PRId64 ", rows:%d", GET_QINFO_ADDR(pRuntimeEnv),
pBlockInfo->window.skey, pBlockInfo->window.ekey, pBlockInfo->rows); pBlockInfo->window.skey, pBlockInfo->window.ekey, pBlockInfo->rows);
} else if (r == BLK_DATA_FILEDS_NEEDED) { } 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; // return DISK_DATA_LOAD_FAILED;
} }
if (*pStatis == NULL) { if (*pStatis == NULL) {
pDataBlock = tsdbRetrieveDataBlock(pRuntimeEnv->pQueryHandle, NULL); pDataBlock = tsdbRetrieveDataBlock(pQueryHandle, NULL);
} }
} else { } else {
assert(r == BLK_DATA_ALL_NEEDED); 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; // return DISK_DATA_LOAD_FAILED;
} }
@ -2376,7 +2311,7 @@ SArray *loadDataBlockOnDemand(SQueryRuntimeEnv *pRuntimeEnv, SDataBlockInfo *pBl
// return DISK_DATA_DISCARDED; // return DISK_DATA_DISCARDED;
} }
pDataBlock = tsdbRetrieveDataBlock(pRuntimeEnv->pQueryHandle, NULL); pDataBlock = tsdbRetrieveDataBlock(pQueryHandle, NULL);
} }
return pDataBlock; return pDataBlock;
@ -2447,8 +2382,11 @@ int32_t binarySearchForKey(char *pValue, int num, TSKEY key, int order) {
static int64_t doScanAllDataBlocks(SQueryRuntimeEnv *pRuntimeEnv) { static int64_t doScanAllDataBlocks(SQueryRuntimeEnv *pRuntimeEnv) {
SQuery *pQuery = pRuntimeEnv->pQuery; SQuery *pQuery = pRuntimeEnv->pQuery;
STableQueryInfo* pTableQueryInfo = pQuery->current;
qTrace("QInfo:%p query start, qrange:%" PRId64 "-%" PRId64 ", lastkey:%" PRId64 ", order:%d", 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; TsdbQueryHandleT pQueryHandle = IS_MASTER_SCAN(pRuntimeEnv)? pRuntimeEnv->pQueryHandle : pRuntimeEnv->pSecQueryHandle;
while (tsdbNextDataBlock(pQueryHandle)) { 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)) { if (!isIntervalQuery(pQuery) && !isGroupbyNormalCol(pQuery->pGroupbyExpr) && !isFixedOutputQuery(pQuery)) {
SResultRec *pRec = &pQuery->rec; SResultRec *pRec = &pQuery->rec;
@ -2505,11 +2443,10 @@ static int64_t doScanAllDataBlocks(SQueryRuntimeEnv *pRuntimeEnv) {
} }
SDataStatis *pStatis = NULL; 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; pQuery->pos = QUERY_IS_ASC_QUERY(pQuery) ? 0 : blockInfo.rows - 1;
int32_t numOfRes = tableApplyFunctionsOnBlock(pRuntimeEnv, &blockInfo, pStatis, binarySearchForKey, int32_t numOfRes = tableApplyFunctionsOnBlock(pRuntimeEnv, &blockInfo, pStatis, binarySearchForKey, pDataBlock);
&pRuntimeEnv->windowResInfo, pDataBlock);
qTrace("QInfo:%p check data block, brange:%" PRId64 "-%" PRId64 ", rows:%d, res:%d", GET_QINFO_ADDR(pRuntimeEnv), 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); 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)) { if (!Q_STATUS_EQUAL(pQuery->status, QUERY_RESBUF_FULL)) {
setQueryStatus(pQuery, QUERY_COMPLETED); 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; int32_t step = QUERY_IS_ASC_QUERY(pQuery) ? QUERY_ASC_FORWARD_STEP : QUERY_DESC_FORWARD_STEP;
closeAllTimeWindow(&pRuntimeEnv->windowResInfo); closeAllTimeWindow(&pRuntimeEnv->windowResInfo);
removeRedundantWindow(&pRuntimeEnv->windowResInfo, pQuery->lastKey - step, step); removeRedundantWindow(&pRuntimeEnv->windowResInfo, pTableQueryInfo->lastKey - step, step);
pRuntimeEnv->windowResInfo.curIndex = pRuntimeEnv->windowResInfo.size - 1; pRuntimeEnv->windowResInfo.curIndex = pRuntimeEnv->windowResInfo.size - 1; // point to the last time window
} else { } else {
assert(Q_STATUS_EQUAL(pQuery->status, QUERY_RESBUF_FULL)); assert(Q_STATUS_EQUAL(pQuery->status, QUERY_RESBUF_FULL));
} }
@ -2540,8 +2477,6 @@ static int64_t doScanAllDataBlocks(SQueryRuntimeEnv *pRuntimeEnv) {
return 0; return 0;
} }
static void updatelastkey(SQuery *pQuery, STableQueryInfo *pTableQInfo) { pTableQInfo->lastKey = pQuery->lastKey; }
/* /*
* set tag value in SQLFunctionCtx * set tag value in SQLFunctionCtx
* e.g.,tag information into input buffer * 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; int16_t type = 0;
if (tagColId == TSDB_TBNAME_COLUMN_INDEX) { if (tagColId == TSDB_TBNAME_COLUMN_INDEX) {
tsdbGetTableName(tsdb, pTableId, &val); val = tsdbGetTableName(tsdb, pTableId, &bytes);
bytes = strnlen(val, TSDB_TABLE_NAME_LEN);
type = TSDB_DATA_TYPE_BINARY; type = TSDB_DATA_TYPE_BINARY;
tVariantCreateFromBinary(param, varDataVal(val), varDataLen(val), type);
} else { } else {
tsdbGetTableTagVal(tsdb, pTableId, tagColId, &type, &bytes, &val); 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); 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) { static void updateTableQueryInfoForReverseScan(SQuery *pQuery, STableQueryInfo *pTableQueryInfo) {
if (pTableQueryInfo == NULL) { if (pTableQueryInfo == NULL) {
return; 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) { 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) { void switchCtxOrder(SQueryRuntimeEnv *pRuntimeEnv) {
@ -3345,13 +3274,14 @@ bool needScanDataBlocksAgain(SQueryRuntimeEnv *pRuntimeEnv) {
static SQueryStatusInfo getQueryStatusInfo(SQueryRuntimeEnv *pRuntimeEnv) { static SQueryStatusInfo getQueryStatusInfo(SQueryRuntimeEnv *pRuntimeEnv) {
SQuery *pQuery = pRuntimeEnv->pQuery; SQuery *pQuery = pRuntimeEnv->pQuery;
STableQueryInfo* pTableQueryInfo = pQuery->current;
SQueryStatusInfo info = { SQueryStatusInfo info = {
.status = pQuery->status, .status = pQuery->status,
.windowIndex = pRuntimeEnv->windowResInfo.curIndex, .windowIndex = pRuntimeEnv->windowResInfo.curIndex,
.lastKey = pQuery->lastKey, .lastKey = pTableQueryInfo->lastKey,
.w = pQuery->window, .w = pQuery->window,
.curWindow = {.skey = pQuery->lastKey, .ekey = pQuery->window.ekey}, .curWindow = {.skey = pTableQueryInfo->lastKey, .ekey = pTableQueryInfo->win.ekey},
}; };
return info; return info;
@ -3395,6 +3325,7 @@ static void setEnvBeforeReverseScan(SQueryRuntimeEnv *pRuntimeEnv, SQueryStatusI
static void clearEnvAfterReverseScan(SQueryRuntimeEnv *pRuntimeEnv, SQueryStatusInfo *pStatus) { static void clearEnvAfterReverseScan(SQueryRuntimeEnv *pRuntimeEnv, SQueryStatusInfo *pStatus) {
SQuery *pQuery = pRuntimeEnv->pQuery; SQuery *pQuery = pRuntimeEnv->pQuery;
STableQueryInfo* pTableQueryInfo = pQuery->current;
SWITCH_ORDER(pQuery->order.order); SWITCH_ORDER(pQuery->order.order);
switchCtxOrder(pRuntimeEnv); 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 // update the pQuery->window.skey and pQuery->window.ekey to limit the scan scope of sliding query
// during reverse scan // during reverse scan
pQuery->lastKey = pStatus->lastKey; pTableQueryInfo->lastKey = pStatus->lastKey;
pQuery->status = pStatus->status; pQuery->status = pStatus->status;
pQuery->window = pStatus->w; pTableQueryInfo->win = pStatus->w;
} }
void scanAllDataBlocks(SQueryRuntimeEnv *pRuntimeEnv) { void scanAllDataBlocks(SQueryRuntimeEnv *pRuntimeEnv) {
SQInfo *pQInfo = (SQInfo *) GET_QINFO_ADDR(pRuntimeEnv);
SQuery *pQuery = pRuntimeEnv->pQuery; SQuery *pQuery = pRuntimeEnv->pQuery;
STableQueryInfo *pTableQueryInfo = pQuery->current;
setQueryStatus(pQuery, QUERY_NOT_COMPLETED); setQueryStatus(pQuery, QUERY_NOT_COMPLETED);
// store the start query position // store the start query position
SQInfo * pQInfo = (SQInfo *)GET_QINFO_ADDR(pRuntimeEnv);
SQueryStatusInfo qstatus = getQueryStatusInfo(pRuntimeEnv); SQueryStatusInfo qstatus = getQueryStatusInfo(pRuntimeEnv);
SET_MASTER_SCAN_FLAG(pRuntimeEnv); SET_MASTER_SCAN_FLAG(pRuntimeEnv);
@ -3429,7 +3362,7 @@ void scanAllDataBlocks(SQueryRuntimeEnv *pRuntimeEnv) {
if (pRuntimeEnv->scanFlag == MASTER_SCAN) { if (pRuntimeEnv->scanFlag == MASTER_SCAN) {
qstatus.status = pQuery->status; qstatus.status = pQuery->status;
qstatus.curWindow.ekey = pQuery->lastKey - step; qstatus.curWindow.ekey = pTableQueryInfo->lastKey - step;
} }
if (!needScanDataBlocksAgain(pRuntimeEnv)) { if (!needScanDataBlocksAgain(pRuntimeEnv)) {
@ -3549,12 +3482,10 @@ void destroyTableQueryInfo(STableQueryInfo *pTableQueryInfo, int32_t numOfCols)
void restoreIntervalQueryRange(SQueryRuntimeEnv *pRuntimeEnv, STableQueryInfo *pTableQueryInfo) { void restoreIntervalQueryRange(SQueryRuntimeEnv *pRuntimeEnv, STableQueryInfo *pTableQueryInfo) {
SQuery *pQuery = pRuntimeEnv->pQuery; SQuery *pQuery = pRuntimeEnv->pQuery;
pQuery->current = pTableQueryInfo;
pQuery->window = pTableQueryInfo->win; assert(((pTableQueryInfo->lastKey >= pTableQueryInfo->win.skey) && QUERY_IS_ASC_QUERY(pQuery)) ||
pQuery->lastKey = pTableQueryInfo->lastKey; ((pTableQueryInfo->lastKey <= pTableQueryInfo->win.skey) && !QUERY_IS_ASC_QUERY(pQuery)));
assert(((pQuery->lastKey >= pQuery->window.skey) && QUERY_IS_ASC_QUERY(pQuery)) ||
((pQuery->lastKey <= pQuery->window.skey) && !QUERY_IS_ASC_QUERY(pQuery)));
} }
/** /**
@ -3562,8 +3493,10 @@ void restoreIntervalQueryRange(SQueryRuntimeEnv *pRuntimeEnv, STableQueryInfo *p
* @param pRuntimeEnv * @param pRuntimeEnv
* @param pDataBlockInfo * @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; SQueryRuntimeEnv *pRuntimeEnv = &pQInfo->runtimeEnv;
STableQueryInfo *pTableQueryInfo = pRuntimeEnv->pQuery->current;
SWindowResInfo * pWindowResInfo = &pRuntimeEnv->windowResInfo; SWindowResInfo * pWindowResInfo = &pRuntimeEnv->windowResInfo;
int32_t GROUPRESULTID = 1; 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 * merged during merge stage. In this case, we need the pTableQueryInfo->lastResRows to decide if there
* is a previous result generated or not. * 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; SQueryRuntimeEnv *pRuntimeEnv = &pQInfo->runtimeEnv;
SQuery * pQuery = pRuntimeEnv->pQuery; SQuery * pQuery = pRuntimeEnv->pQuery;
STableQueryInfo *pTableQueryInfo = pQuery->current;
if (pTableQueryInfo->queryRangeSet) { if (pTableQueryInfo->queryRangeSet) {
pQuery->lastKey = key;
pTableQueryInfo->lastKey = key; pTableQueryInfo->lastKey = key;
} else { } else {
pQuery->window.skey = key; pQuery->window.skey = key;
@ -3689,8 +3622,6 @@ void setIntervalQueryRange(STableQueryInfo *pTableQueryInfo, SQInfo *pQInfo, TSK
pTableQueryInfo->queryRangeSet = 1; pTableQueryInfo->queryRangeSet = 1;
pTableQueryInfo->lastKey = pQuery->window.skey; pTableQueryInfo->lastKey = pQuery->window.skey;
pTableQueryInfo->win.skey = 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; * 2. if there are top/bottom, first_dst/last_dst functions, we need to load timestamp column in any cases;
*/ */
STimeWindow *w = &pDataBlockInfo->window; 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); (pQuery->window.ekey >= w->skey && pQuery->window.ekey <= w->ekey) || requireTimestamp(pQuery);
return loadPrimaryTS; return loadPrimaryTS;
@ -3847,7 +3780,6 @@ void stableApplyFunctionsOnBlock(SQueryRuntimeEnv *pRuntimeEnv, STableQueryInfo
} }
updateWindowResNumOfRes(pRuntimeEnv, pTableQueryInfo); updateWindowResNumOfRes(pRuntimeEnv, pTableQueryInfo);
updatelastkey(pQuery, pTableQueryInfo);
} }
bool vnodeHasRemainResults(void *handle) { bool vnodeHasRemainResults(void *handle) {
@ -4041,10 +3973,12 @@ void vnodePrintQueryStatistics(SQInfo *pQInfo) {
static void updateOffsetVal(SQueryRuntimeEnv *pRuntimeEnv, SDataBlockInfo *pBlockInfo) { static void updateOffsetVal(SQueryRuntimeEnv *pRuntimeEnv, SDataBlockInfo *pBlockInfo) {
SQuery *pQuery = pRuntimeEnv->pQuery; SQuery *pQuery = pRuntimeEnv->pQuery;
STableQueryInfo* pTableQueryInfo = pQuery->current;
int32_t step = GET_FORWARD_DIRECTION_FACTOR(pQuery->order.order); int32_t step = GET_FORWARD_DIRECTION_FACTOR(pQuery->order.order);
if (pQuery->limit.offset == pBlockInfo->rows) { // current block will ignore completed 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; pQuery->limit.offset = 0;
return; return;
} }
@ -4064,11 +3998,10 @@ static void updateOffsetVal(SQueryRuntimeEnv *pRuntimeEnv, SDataBlockInfo *pBloc
TSKEY *keys = (TSKEY *)pColInfoData->pData; TSKEY *keys = (TSKEY *)pColInfoData->pData;
// update the offset value // update the offset value
pQuery->lastKey = keys[pQuery->pos]; pTableQueryInfo->lastKey = keys[pQuery->pos];
pQuery->limit.offset = 0; pQuery->limit.offset = 0;
int32_t numOfRes = tableApplyFunctionsOnBlock(pRuntimeEnv, pBlockInfo, NULL, binarySearchForKey, int32_t numOfRes = tableApplyFunctionsOnBlock(pRuntimeEnv, pBlockInfo, NULL, binarySearchForKey, pDataBlock);
&pRuntimeEnv->windowResInfo, pDataBlock);
qTrace("QInfo:%p check data block, brange:%" PRId64 "-%" PRId64 ", rows:%d, res:%d", GET_QINFO_ADDR(pRuntimeEnv), 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); pBlockInfo->window.skey, pBlockInfo->window.ekey, pBlockInfo->rows, numOfRes);
@ -4084,6 +4017,7 @@ void skipBlocks(SQueryRuntimeEnv *pRuntimeEnv) {
pQuery->pos = 0; pQuery->pos = 0;
int32_t step = GET_FORWARD_DIRECTION_FACTOR(pQuery->order.order); int32_t step = GET_FORWARD_DIRECTION_FACTOR(pQuery->order.order);
STableQueryInfo* pTableQueryInfo = pQuery->current;
TsdbQueryHandleT pQueryHandle = pRuntimeEnv->pQueryHandle; TsdbQueryHandleT pQueryHandle = pRuntimeEnv->pQueryHandle;
while (tsdbNextDataBlock(pQueryHandle)) { while (tsdbNextDataBlock(pQueryHandle)) {
@ -4095,8 +4029,8 @@ void skipBlocks(SQueryRuntimeEnv *pRuntimeEnv) {
if (pQuery->limit.offset > blockInfo.rows) { if (pQuery->limit.offset > blockInfo.rows) {
pQuery->limit.offset -= blockInfo.rows; pQuery->limit.offset -= blockInfo.rows;
pQuery->lastKey = (QUERY_IS_ASC_QUERY(pQuery)) ? blockInfo.window.ekey : blockInfo.window.skey; pTableQueryInfo->lastKey = (QUERY_IS_ASC_QUERY(pQuery)) ? blockInfo.window.ekey : blockInfo.window.skey;
pQuery->lastKey += step; pTableQueryInfo->lastKey += step;
qTrace("QInfo:%p skip rows:%d, offset:%" PRId64 "", GET_QINFO_ADDR(pRuntimeEnv), blockInfo.rows, qTrace("QInfo:%p skip rows:%d, offset:%" PRId64 "", GET_QINFO_ADDR(pRuntimeEnv), blockInfo.rows,
pQuery->limit.offset); pQuery->limit.offset);
@ -4125,6 +4059,7 @@ static bool skipTimeInterval(SQueryRuntimeEnv *pRuntimeEnv) {
TSKEY skey1, ekey1; TSKEY skey1, ekey1;
STimeWindow w = {0}; STimeWindow w = {0};
SWindowResInfo *pWindowResInfo = &pRuntimeEnv->windowResInfo; SWindowResInfo *pWindowResInfo = &pRuntimeEnv->windowResInfo;
STableQueryInfo *pTableQueryInfo = pQuery->current;
while (tsdbNextDataBlock(pRuntimeEnv->pQueryHandle)) { while (tsdbNextDataBlock(pRuntimeEnv->pQueryHandle)) {
SDataBlockInfo blockInfo = tsdbRetrieveDataBlockInfo(pRuntimeEnv->pQueryHandle); SDataBlockInfo blockInfo = tsdbRetrieveDataBlockInfo(pRuntimeEnv->pQueryHandle);
@ -4170,11 +4105,10 @@ static bool skipTimeInterval(SQueryRuntimeEnv *pRuntimeEnv) {
// set the abort info // set the abort info
pQuery->pos = startPos; pQuery->pos = startPos;
pQuery->lastKey = ((TSKEY *)pColInfoData->pData)[startPos]; pTableQueryInfo->lastKey = ((TSKEY *)pColInfoData->pData)[startPos];
pWindowResInfo->prevSKey = tw.skey; pWindowResInfo->prevSKey = tw.skey;
int32_t numOfRes = tableApplyFunctionsOnBlock(pRuntimeEnv, &blockInfo, NULL, binarySearchForKey, int32_t numOfRes = tableApplyFunctionsOnBlock(pRuntimeEnv, &blockInfo, NULL, binarySearchForKey, pDataBlock);
&pRuntimeEnv->windowResInfo, pDataBlock);
qTrace("QInfo:%p check data block, brange:%" PRId64 "-%" PRId64 ", rows:%d, res:%d", 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); 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 // set the abort info
pQuery->pos = startPos; pQuery->pos = startPos;
pQuery->lastKey = ((TSKEY *)pColInfoData->pData)[startPos]; pTableQueryInfo->lastKey = ((TSKEY *)pColInfoData->pData)[startPos];
pWindowResInfo->prevSKey = tw.skey; pWindowResInfo->prevSKey = tw.skey;
win = tw; win = tw;
} else { } else {
@ -4220,9 +4154,6 @@ int32_t doInitQInfo(SQInfo *pQInfo, void *param, void *tsdb, int32_t vgId, bool
setScanLimitationByResultBuffer(pQuery); setScanLimitationByResultBuffer(pQuery);
changeExecuteScanOrder(pQuery, false); changeExecuteScanOrder(pQuery, false);
// dataInCache requires lastKey value
pQuery->lastKey = pQuery->window.skey;
STsdbQueryCond cond = { STsdbQueryCond cond = {
.twindow = pQuery->window, .twindow = pQuery->window,
.order = pQuery->order.order, .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)) { } else if (!isSTableQuery || isIntervalQuery(pQuery) || isFixedOutputQuery(pQuery)) {
pRuntimeEnv->pQueryHandle = tsdbQueryTables(tsdb, &cond, &pQInfo->tableIdGroupInfo); pRuntimeEnv->pQueryHandle = tsdbQueryTables(tsdb, &cond, &pQInfo->tableIdGroupInfo);
} }
// create the table query support structures
createTableQueryInfo(pQInfo);
} }
pQInfo->tsdb = tsdb; 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 // the pQuery->window.skey is changed during normalizedFirstQueryRange, so set the newest lastkey value
pQuery->lastKey = pQuery->window.skey;
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }
@ -4404,13 +4337,14 @@ static int64_t queryOnDataBlocks(SQInfo *pQInfo) {
restoreIntervalQueryRange(pRuntimeEnv, pTableQueryInfo); restoreIntervalQueryRange(pRuntimeEnv, pTableQueryInfo);
SDataStatis *pStatis = NULL; SDataStatis *pStatis = NULL;
SArray * pDataBlock = loadDataBlockOnDemand(pRuntimeEnv, &blockInfo, &pStatis);
SArray *pDataBlock = loadDataBlockOnDemand(pRuntimeEnv, pQueryHandle, &blockInfo, &pStatis);
TSKEY nextKey = blockInfo.window.skey; TSKEY nextKey = blockInfo.window.skey;
if (!isIntervalQuery(pQuery)) { if (!isIntervalQuery(pQuery)) {
setExecutionContext(pQInfo, pTableQueryInfo, &pTableQueryInfo->id, pTableQueryInfo->groupIdx, nextKey); setExecutionContext(pQInfo, &pTableQueryInfo->id, pTableQueryInfo->groupIdx, nextKey);
} else { // interval query } else { // interval query
setIntervalQueryRange(pTableQueryInfo, pQInfo, nextKey); setIntervalQueryRange(pQInfo, nextKey);
int32_t ret = setAdditionalInfo(pQInfo, &pTableQueryInfo->id, pTableQueryInfo); int32_t ret = setAdditionalInfo(pQInfo, &pTableQueryInfo->id, pTableQueryInfo);
if (ret != TSDB_CODE_SUCCESS) { if (ret != TSDB_CODE_SUCCESS) {
@ -4446,6 +4380,7 @@ static bool multiTableMultioutputHelper(SQInfo *pQInfo, int32_t index) {
.numOfCols = pQuery->numOfCols, .numOfCols = pQuery->numOfCols,
}; };
// todo refactor
SArray *g1 = taosArrayInit(1, POINTER_BYTES); SArray *g1 = taosArrayInit(1, POINTER_BYTES);
SArray *tx = taosArrayInit(1, sizeof(STableId)); SArray *tx = taosArrayInit(1, sizeof(STableId));
@ -4564,6 +4499,9 @@ static void sequentialTableProcess(SQInfo *pQInfo) {
initCtxOutputBuf(pRuntimeEnv); initCtxOutputBuf(pRuntimeEnv);
setTagVal(pRuntimeEnv, (STableId*) taosArrayGet(tx, 0), pQInfo->tsdb); 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); scanAllDataBlocks(pRuntimeEnv);
int64_t numOfRes = getNumOfResult(pRuntimeEnv); int64_t numOfRes = getNumOfResult(pRuntimeEnv);
@ -4580,8 +4518,6 @@ static void sequentialTableProcess(SQInfo *pQInfo) {
} }
} }
} else { } else {
createTableQueryInfo(pQInfo);
/* /*
* 1. super table projection query, 2. group-by on normal columns query, 3. ts-comp query * 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. * 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); SGroupItem *item = taosArrayGet(group, pQInfo->tableIndex);
pQuery->current = item->info;
STableQueryInfo *pInfo = item->info;
if (pInfo->lastKey > 0) {
pQuery->window.skey = pInfo->lastKey;
}
if (!multiTableMultioutputHelper(pQInfo, pQInfo->tableIndex)) { if (!multiTableMultioutputHelper(pQInfo, pQInfo->tableIndex)) {
pQInfo->tableIndex++; pQInfo->tableIndex++;
@ -4627,7 +4559,7 @@ static void sequentialTableProcess(SQInfo *pQInfo) {
// SPointInterpoSupporter pointInterpSupporter = {0}; // SPointInterpoSupporter pointInterpSupporter = {0};
// TODO handle the limit problem // TODO handle the limit offset problem
if (pQuery->numOfFilterCols == 0 && pQuery->limit.offset > 0) { if (pQuery->numOfFilterCols == 0 && pQuery->limit.offset > 0) {
// skipBlocks(pRuntimeEnv); // 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. * to ensure that, we can reset the query range once query on a meter is completed.
*/ */
pQInfo->tableIndex++; 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 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) /*|| if (Q_STATUS_EQUAL(pQuery->status, QUERY_RESBUF_FULL) /*||
isGroupbyEachTable(pQuery->pGroupbyExpr, pSupporter->pSidSet)*/) { isGroupbyEachTable(pQuery->pGroupbyExpr, pSupporter->pSidSet)*/) {
@ -4668,7 +4598,7 @@ static void sequentialTableProcess(SQInfo *pQInfo) {
} }
} else { // forward query range } 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 // all data in the result buffer are skipped due to the offset, continue to retrieve data from current meter
if (pQuery->rec.rows == 0) { 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, qTrace("QInfo:%p query start, qrange:%" PRId64 "-%" PRId64 ", order:%d, forward scan start", pQInfo,
pQuery->window.skey, pQuery->window.ekey, pQuery->order.order); pQuery->window.skey, pQuery->window.ekey, pQuery->order.order);
// create the query support structures
createTableQueryInfo(pQInfo);
// do check all qualified data blocks // do check all qualified data blocks
int64_t el = queryOnDataBlocks(pQInfo); 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 // query error occurred or query is killed, abort current execution
if (pQInfo->code != TSDB_CODE_SUCCESS || isQueryKilled(pQInfo)) { 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(*)/top(field,k)/avg(field name) from table_name [where ts>now-1a];
* select count(*) from table_name group by status_column; * 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; SQueryRuntimeEnv *pRuntimeEnv = &pQInfo->runtimeEnv;
SQuery * pQuery = pRuntimeEnv->pQuery;
SQuery *pQuery = pRuntimeEnv->pQuery;
pQuery->current = pTableInfo; // set current query table info
scanAllDataBlocks(pRuntimeEnv); scanAllDataBlocks(pRuntimeEnv);
finalizeQueryResult(pRuntimeEnv); finalizeQueryResult(pRuntimeEnv);
@ -4932,9 +4861,11 @@ static void tableFixedOutputProcess(SQInfo *pQInfo) {
limitResults(pQInfo); limitResults(pQInfo);
} }
static void tableMultiOutputProcess(SQInfo *pQInfo) { static void tableMultiOutputProcess(SQInfo *pQInfo, STableQueryInfo* pTableInfo) {
SQueryRuntimeEnv *pRuntimeEnv = &pQInfo->runtimeEnv; SQueryRuntimeEnv *pRuntimeEnv = &pQInfo->runtimeEnv;
SQuery * pQuery = pRuntimeEnv->pQuery;
SQuery *pQuery = pRuntimeEnv->pQuery;
pQuery->current = pTableInfo;
// for ts_comp query, re-initialized is not allowed // for ts_comp query, re-initialized is not allowed
if (!isTSCompQuery(pQuery)) { 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, 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); resetCtxOutputBuf(pRuntimeEnv);
} }
limitResults(pQInfo); limitResults(pQInfo);
if (Q_STATUS_EQUAL(pQuery->status, QUERY_RESBUF_FULL)) { 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, qTrace("QInfo:%p query paused due to output limitation, next qrange:%" PRId64 "-%" PRId64, pQInfo,
pQuery->window.ekey); pQuery->current->lastKey, pQuery->window.ekey);
} }
if (!isTSCompQuery(pQuery)) { if (!isTSCompQuery(pQuery)) {
@ -5018,11 +4949,12 @@ static void tableIntervalProcessImpl(SQueryRuntimeEnv *pRuntimeEnv) {
} }
// handle time interval query on table // handle time interval query on table
static void tableIntervalProcess(SQInfo *pQInfo) { static void tableIntervalProcess(SQInfo *pQInfo, STableQueryInfo* pTableInfo) {
SQueryRuntimeEnv *pRuntimeEnv = &(pQInfo->runtimeEnv); SQueryRuntimeEnv *pRuntimeEnv = &(pQInfo->runtimeEnv);
SQuery * pQuery = pRuntimeEnv->pQuery;
int32_t numOfInterpo = 0; 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 // skip blocks without load the actual data block from file if no filter condition present
skipTimeInterval(pRuntimeEnv); skipTimeInterval(pRuntimeEnv);
@ -5132,14 +5064,18 @@ static void tableQueryImpl(SQInfo *pQInfo) {
pQuery->rec.rows = 0; pQuery->rec.rows = 0;
int64_t st = taosGetTimestampUs(); 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 // 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) if (isIntervalQuery(pQuery) || isGroupbyNormalCol(pQuery->pGroupbyExpr)) { // interval (down sampling operation)
tableIntervalProcess(pQInfo); tableIntervalProcess(pQInfo, item->info);
} else if (isFixedOutputQuery(pQuery)) { } else if (isFixedOutputQuery(pQuery)) {
tableFixedOutputProcess(pQInfo); tableFixedOutputProcess(pQInfo, item->info);
} else { // diff/add/multiply/subtract/division } else { // diff/add/multiply/subtract/division
assert(pQuery->checkBuffer == 1); assert(pQuery->checkBuffer == 1);
tableMultiOutputProcess(pQInfo); tableMultiOutputProcess(pQInfo, item->info);
} }
// record the total elapsed time // record the total elapsed time
@ -5833,7 +5769,6 @@ static SQInfo *createQInfoImpl(SQueryTableMsg *pQueryMsg, SSqlGroupbyExpr *pGrou
pQuery->pos = -1; pQuery->pos = -1;
pQuery->window = pQueryMsg->window; pQuery->window = pQueryMsg->window;
pQuery->lastKey = pQuery->window.skey;
if (sem_init(&pQInfo->dataReady, 0, 0) != 0) { if (sem_init(&pQInfo->dataReady, 0, 0) != 0) {
qError("QInfo:%p init dataReady sem failed, reason:%s", pQInfo, strerror(errno)); 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; goto _error;
} }
// qTrace("QInfo:%p set query flag and prepare runtime environment completed, ref:%d, wait for schedule", pQInfo,
// pQInfo->refCount);
return code; return code;
_error: _error:
@ -6085,34 +6018,35 @@ int32_t qCreateQueryInfo(void *tsdb, int32_t vgId, SQueryTableMsg *pQueryMsg, qi
if (pQueryMsg->numOfTables <= 0) { if (pQueryMsg->numOfTables <= 0) {
qError("Invalid number of tables to query, numOfTables:%d", pQueryMsg->numOfTables); qError("Invalid number of tables to query, numOfTables:%d", pQueryMsg->numOfTables);
code = TSDB_CODE_INVALID_QUERY_MSG; code = TSDB_CODE_INVALID_QUERY_MSG;
goto _query_over; goto _over;
} }
if (pTableIdList == NULL || taosArrayGetSize(pTableIdList) == 0) { if (pTableIdList == NULL || taosArrayGetSize(pTableIdList) == 0) {
qError("qmsg:%p, SQueryTableMsg wrong format", pQueryMsg); qError("qmsg:%p, SQueryTableMsg wrong format", pQueryMsg);
code = TSDB_CODE_INVALID_QUERY_MSG; code = TSDB_CODE_INVALID_QUERY_MSG;
goto _query_over; goto _over;
} }
SExprInfo *pExprs = NULL; SExprInfo *pExprs = NULL;
if ((code = createSqlFunctionExprFromMsg(pQueryMsg, &pExprs, pExprMsg, pTagColumnInfo)) != TSDB_CODE_SUCCESS) { if ((code = createSqlFunctionExprFromMsg(pQueryMsg, &pExprs, pExprMsg, pTagColumnInfo)) != TSDB_CODE_SUCCESS) {
goto _query_over; goto _over;
} }
SSqlGroupbyExpr *pGroupbyExpr = createGroupbyExprFromMsg(pQueryMsg, pGroupColIndex, &code); SSqlGroupbyExpr *pGroupbyExpr = createGroupbyExprFromMsg(pQueryMsg, pGroupColIndex, &code);
if ((pGroupbyExpr == NULL && pQueryMsg->numOfGroupCols != 0) || code != TSDB_CODE_SUCCESS) { if ((pGroupbyExpr == NULL && pQueryMsg->numOfGroupCols != 0) || code != TSDB_CODE_SUCCESS) {
goto _query_over; goto _over;
} }
bool isSTableQuery = false; bool isSTableQuery = false;
STableGroupInfo groupInfo = {0}; STableGroupInfo groupInfo = {0};
//todo multitable_query??
if (TSDB_QUERY_HAS_TYPE(pQueryMsg->queryType, TSDB_QUERY_TYPE_MULTITABLE_QUERY|TSDB_QUERY_TYPE_TABLE_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); isSTableQuery = TSDB_QUERY_HAS_TYPE(pQueryMsg->queryType, TSDB_QUERY_TYPE_MULTITABLE_QUERY);
STableId *id = taosArrayGet(pTableIdList, 0); STableId *id = taosArrayGet(pTableIdList, 0);
if ((code = tsdbGetOneTableGroup(tsdb, id->uid, &groupInfo)) != TSDB_CODE_SUCCESS) { 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)) { } else if (TSDB_QUERY_HAS_TYPE(pQueryMsg->queryType, TSDB_QUERY_TYPE_STABLE_QUERY)) {
isSTableQuery = true; isSTableQuery = true;
@ -6129,7 +6063,7 @@ int32_t qCreateQueryInfo(void *tsdb, int32_t vgId, SQueryTableMsg *pQueryMsg, qi
numOfGroupByCols); numOfGroupByCols);
if (groupInfo.numOfTables == 0) { // no qualified tables no need to do query if (groupInfo.numOfTables == 0) { // no qualified tables no need to do query
code = TSDB_CODE_SUCCESS; code = TSDB_CODE_SUCCESS;
goto _query_over; goto _over;
} }
} else { } else {
assert(0); assert(0);
@ -6138,11 +6072,12 @@ int32_t qCreateQueryInfo(void *tsdb, int32_t vgId, SQueryTableMsg *pQueryMsg, qi
(*pQInfo) = createQInfoImpl(pQueryMsg, pGroupbyExpr, pExprs, &groupInfo, pTagColumnInfo); (*pQInfo) = createQInfoImpl(pQueryMsg, pGroupbyExpr, pExprs, &groupInfo, pTagColumnInfo);
if ((*pQInfo) == NULL) { if ((*pQInfo) == NULL) {
code = TSDB_CODE_SERV_OUT_OF_MEMORY; code = TSDB_CODE_SERV_OUT_OF_MEMORY;
goto _over;
} }
code = initQInfo(pQueryMsg, tsdb, vgId, *pQInfo, isSTableQuery); code = initQInfo(pQueryMsg, tsdb, vgId, *pQInfo, isSTableQuery);
_query_over: _over:
tfree(tagCond); tfree(tagCond);
tfree(tbnameCond); tfree(tbnameCond);
taosArrayDestroy(pTableIdList); taosArrayDestroy(pTableIdList);
@ -6318,12 +6253,10 @@ static void buildTagQueryResult(SQInfo* pQInfo) {
for(int32_t j = 0; j < pQuery->numOfOutput; ++j) { for(int32_t j = 0; j < pQuery->numOfOutput; ++j) {
// todo check the return value, refactor codes // todo check the return value, refactor codes
if (pExprInfo[j].base.colInfo.colId == TSDB_TBNAME_COLUMN_INDEX) { 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); 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); memcpy(dst, data, varDataTLen(data));
tfree(data);
} else {// todo refactor, return the true length of binary|nchar 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); tsdbGetTableTagVal(pQInfo->tsdb, &item->id, pExprInfo[j].base.colInfo.colId, &type, &bytes, &data);
assert(bytes == pExprInfo[j].bytes && type == pExprInfo[j].type); assert(bytes == pExprInfo[j].bytes && type == pExprInfo[j].type);

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

@ -81,7 +81,6 @@ typedef struct {
// ---------- TSDB TABLE DEFINITION // ---------- TSDB TABLE DEFINITION
typedef struct STable { typedef struct STable {
int8_t type; int8_t type;
char * name;
STableId tableId; STableId tableId;
int64_t superUid; // Super table UID int64_t superUid; // Super table UID
int32_t sversion; int32_t sversion;
@ -96,9 +95,10 @@ typedef struct STable {
TSKEY lastKey; // lastkey inserted in this table, initialized as 0, TODO: make a structure TSKEY lastKey; // lastkey inserted in this table, initialized as 0, TODO: make a structure
struct STable *next; // TODO: remove the next struct STable *next; // TODO: remove the next
struct STable *prev; struct STable *prev;
tstr * name; // NOTE: there a flexible string here
} STable; } 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); void * tsdbEncodeTable(STable *pTable, int *contLen);
STable *tsdbDecodeTable(void *cont, int contLen); STable *tsdbDecodeTable(void *cont, int contLen);
@ -121,6 +121,12 @@ typedef struct {
int maxCols; int maxCols;
} STsdbMeta; } STsdbMeta;
// element put in skiplist for each table
typedef struct STableIndexElem {
STsdbMeta* pMeta;
STable* pTable;
} STableIndexElem;
STsdbMeta *tsdbInitMeta(char *rootDir, int32_t maxTables); STsdbMeta *tsdbInitMeta(char *rootDir, int32_t maxTables);
int32_t tsdbFreeMeta(STsdbMeta *pMeta); int32_t tsdbFreeMeta(STsdbMeta *pMeta);
STSchema * tsdbGetTableSchema(STsdbMeta *pMeta, STable *pTable); STSchema * tsdbGetTableSchema(STsdbMeta *pMeta, STable *pTable);
@ -150,7 +156,7 @@ int32_t tsdbDropTableImpl(STsdbMeta *pMeta, STableId tableId);
STable *tsdbIsValidTableToInsert(STsdbMeta *pMeta, STableId tableId); STable *tsdbIsValidTableToInsert(STsdbMeta *pMeta, STableId tableId);
// int32_t tsdbInsertRowToTableImpl(SSkipListNode *pNode, STable *pTable); // int32_t tsdbInsertRowToTableImpl(SSkipListNode *pNode, STable *pTable);
STable *tsdbGetTableByUid(STsdbMeta *pMeta, int64_t uid); STable *tsdbGetTableByUid(STsdbMeta *pMeta, int64_t uid);
char * getTupleKey(const void *data); char *getTSTupleKey(const void * data);
typedef struct { typedef struct {
int blockId; 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) { if (pTable->mem == NULL) {
pTable->mem = (SMemTable *)calloc(1, sizeof(SMemTable)); pTable->mem = (SMemTable *)calloc(1, sizeof(SMemTable));
if (pTable->mem == NULL) return -1; 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->keyFirst = INT64_MAX;
pTable->mem->keyLast = 0; pTable->mem->keyLast = 0;
} }
@ -831,7 +831,7 @@ static int32_t tdInsertRowToTable(STsdbRepo *pRepo, SDataRow row, STable *pTable
if (pTable->mem == NULL) { if (pTable->mem == NULL) {
pTable->mem = (SMemTable *)calloc(1, sizeof(SMemTable)); pTable->mem = (SMemTable *)calloc(1, sizeof(SMemTable));
if (pTable->mem == NULL) return -1; 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->keyFirst = INT64_MAX;
pTable->mem->keyLast = 0; pTable->mem->keyLast = 0;
} }

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

@ -1,6 +1,4 @@
#include <stdlib.h> #include <stdlib.h>
// #include "taosdef.h"
#include "tskiplist.h" #include "tskiplist.h"
#include "tsdb.h" #include "tsdb.h"
#include "taosdef.h" #include "taosdef.h"
@ -10,6 +8,8 @@
#define TSDB_SUPER_TABLE_SL_LEVEL 5 // TODO: may change here #define TSDB_SUPER_TABLE_SL_LEVEL 5 // TODO: may change here
#define TSDB_META_FILE_NAME "META" #define TSDB_META_FILE_NAME "META"
const int32_t DEFAULT_TAG_INDEX_COLUMN = 0;
static int tsdbFreeTable(STable *pTable); static int tsdbFreeTable(STable *pTable);
static int32_t tsdbCheckTableCfg(STableCfg *pCfg); static int32_t tsdbCheckTableCfg(STableCfg *pCfg);
static int tsdbAddTableToMeta(STsdbMeta *pMeta, STable *pTable, bool addIdx); static int tsdbAddTableToMeta(STsdbMeta *pMeta, STable *pTable, bool addIdx);
@ -39,11 +39,12 @@ void *tsdbEncodeTable(STable *pTable, int *contLen) {
void *ptr = ret; void *ptr = ret;
T_APPEND_MEMBER(ptr, pTable, STable, type); T_APPEND_MEMBER(ptr, pTable, STable, type);
// Encode name // Encode name, todo refactor
*(int *)ptr = strlen(pTable->name); *(int *)ptr = varDataLen(pTable->name);
ptr = (char *)ptr + sizeof(int); ptr = (char *)ptr + sizeof(int);
memcpy(ptr, pTable->name, strlen(pTable->name)); memcpy(ptr, varDataVal(pTable->name), varDataLen(pTable->name));
ptr = (char *)ptr + strlen(pTable->name); ptr = (char *)ptr + varDataLen(pTable->name);
T_APPEND_MEMBER(ptr, &(pTable->tableId), STableId, uid); T_APPEND_MEMBER(ptr, &(pTable->tableId), STableId, uid);
T_APPEND_MEMBER(ptr, &(pTable->tableId), STableId, tid); T_APPEND_MEMBER(ptr, &(pTable->tableId), STableId, tid);
T_APPEND_MEMBER(ptr, pTable, STable, superUid); 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); T_READ_MEMBER(ptr, int8_t, pTable->type);
int len = *(int *)ptr; int len = *(int *)ptr;
ptr = (char *)ptr + sizeof(int); 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; 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; ptr = (char *)ptr + len;
T_READ_MEMBER(ptr, int64_t, pTable->tableId.uid); T_READ_MEMBER(ptr, int64_t, pTable->tableId.uid);
T_READ_MEMBER(ptr, int32_t, pTable->tableId.tid); T_READ_MEMBER(ptr, int32_t, pTable->tableId.tid);
@ -105,8 +109,13 @@ void tsdbFreeEncode(void *cont) {
} }
static char* getTagIndexKey(const void* pData) { static char* getTagIndexKey(const void* pData) {
STable* table = *(STable**) pData; STableIndexElem* elem = (STableIndexElem*) pData;
return getTupleKey(table->tagVal);
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) { 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) { int32_t tsdbGetTableTagVal(TsdbRepoT* repo, STableId* id, int32_t colId, int16_t* type, int16_t* bytes, char** val) {
STsdbMeta* pMeta = tsdbGetMeta(repo); STsdbMeta* pMeta = tsdbGetMeta(repo);
STable* pTable = tsdbGetTableByUid(pMeta, id->uid); STable* pTable = tsdbGetTableByUid(pMeta, id->uid);
STSchema* pSchema = tsdbGetTableTagSchema(pMeta, pTable); STSchema* pSchema = tsdbGetTableTagSchema(pMeta, pTable);
STColumn* pCol = NULL; STColumn* pCol = NULL;
int32_t offset = 0;
for(int32_t col = 0; col < schemaNCols(pSchema); ++col) { for(int32_t col = 0; col < schemaNCols(pSchema); ++col) {
STColumn* p = schemaColAt(pSchema, col); STColumn* p = schemaColAt(pSchema, col);
if (p->colId == colId) { if (p->colId == colId) {
pCol = p; pCol = p;
break; 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) { if (pCol == NULL) {
return -1; // No matched tags. Maybe the modification of tags has not been done yet. return -1; // No matched tags. Maybe the modification of tags has not been done yet.
} }
assert(pCol != NULL);
SDataRow row = (SDataRow)pTable->tagVal; 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; *val = d;
*type = pCol->type; *type = pCol->type;
@ -263,15 +264,22 @@ int32_t tsdbGetTableTagVal(TsdbRepoT* repo, STableId* id, int32_t colId, int16_t
return TSDB_CODE_SUCCESS; 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); STsdbMeta* pMeta = tsdbGetMeta(repo);
STable* pTable = tsdbGetTableByUid(pMeta, id->uid); STable* pTable = tsdbGetTableByUid(pMeta, id->uid);
*name = strndup(pTable->name, TSDB_TABLE_NAME_LEN); if (pTable == NULL) {
if (*name == NULL) { if (bytes != NULL) {
return -1; *bytes = 0;
}
return NULL;
} else { } 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->schema = tdDupSchema(pCfg->schema);
super->tagSchema = tdDupSchema(pCfg->tagSchema); super->tagSchema = tdDupSchema(pCfg->tagSchema);
super->tagVal = NULL; 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 // index the first tag column
STColumn* pColSchema = schemaColAt(super->tagSchema, 0); STColumn* pColSchema = schemaColAt(super->tagSchema, 0);
@ -322,7 +334,11 @@ int32_t tsdbCreateTableImpl(STsdbMeta *pMeta, STableCfg *pCfg) {
} }
table->tableId = pCfg->tableId; 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; table->lastKey = 0;
if (IS_CREATE_STABLE(pCfg)) { // TSDB_CHILD_TABLE if (IS_CREATE_STABLE(pCfg)) { // TSDB_CHILD_TABLE
table->type = 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 // 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 // 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; pNode->level = level;
SSkipList* list = pSTable->pIndex; 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); tSkipListPut(list, pNode);
return 0; return 0;
@ -539,7 +558,7 @@ static int tsdbRemoveTableFromIndex(STsdbMeta *pMeta, STable *pTable) {
static int tsdbEstimateTableEncodeSize(STable *pTable) { static int tsdbEstimateTableEncodeSize(STable *pTable) {
int size = 0; int size = 0;
size += T_MEMBER_SIZE(STable, type); 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, tableId);
size += T_MEMBER_SIZE(STable, superUid); size += T_MEMBER_SIZE(STable, superUid);
size += T_MEMBER_SIZE(STable, sversion); size += T_MEMBER_SIZE(STable, sversion);
@ -556,8 +575,7 @@ static int tsdbEstimateTableEncodeSize(STable *pTable) {
return size; return size;
} }
char *getTupleKey(const void * data) { char *getTSTupleKey(const void * data) {
SDataRow row = (SDataRow)data; SDataRow row = (SDataRow)data;
return POINTER_SHIFT(row, TD_DATA_ROW_HEAD_SIZE); 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) { 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); memmove(pCol->pData, src->pData + bytes * start, bytes * pQueryHandle->realNumOfRows);
} else { // handle the var-string } 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) { for(int32_t k = start; k < pQueryHandle->realNumOfRows + start; ++k) {
char* p = tdGetColDataOfRow(src, 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 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) { 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)); memcpy(pData, value, varDataTLen(value));
offset += sizeof(int32_t);
} else { } else {
memcpy(pData, dataRowTuple(row) + offset, pColInfo->info.bytes); memcpy(pData, value, pColInfo->info.bytes);
offset += pColInfo->info.bytes;
} }
} }
numOfRows++; numOfRows++;
@ -1225,8 +1225,8 @@ static int32_t getAllTableIdList(STable* pSuperTable, SArray* list) {
while (tSkipListIterNext(iter)) { while (tSkipListIterNext(iter)) {
SSkipListNode* pNode = tSkipListIterGet(iter); SSkipListNode* pNode = tSkipListIterGet(iter);
STable* t = *(STable**)SL_GET_NODE_DATA(pNode); STableIndexElem* elem = (STableIndexElem*)(SL_GET_NODE_DATA((SSkipListNode*) pNode));
taosArrayPush(list, &t->tableId); taosArrayPush(list, &elem->pTable->tableId);
} }
tSkipListDestroyIter(iter); 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 * 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 * @param pRes
*/ */
static void convertQueryResult(SArray* pRes, SArray* pTableList) { static void convertQueryResult(SArray* pRes, SArray* pTableList) {
@ -1244,8 +1245,8 @@ static void convertQueryResult(SArray* pRes, SArray* pTableList) {
size_t size = taosArrayGetSize(pTableList); size_t size = taosArrayGetSize(pTableList);
for (int32_t i = 0; i < size; ++i) { // todo speedup by using reserve space. for (int32_t i = 0; i < size; ++i) { // todo speedup by using reserve space.
STable* pTable = taosArrayGetP(pTableList, i); STableIndexElem* elem = taosArrayGet(pTableList, i);
taosArrayPush(pRes, &pTable->tableId); taosArrayPush(pRes, &elem->pTable->tableId);
} }
} }
@ -1309,8 +1310,13 @@ void filterPrepare(void* expr, void* param) {
pInfo->q = (char*) pCond->arr; pInfo->q = (char*) pCond->arr;
} else { } else {
pInfo->q = calloc(1, pSchema->bytes); 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); tVariantDump(pCond, pInfo->q, pSchema->type);
} }
}
} }
typedef struct STableGroupSupporter { typedef struct STableGroupSupporter {
@ -1341,16 +1347,16 @@ int32_t tableGroupComparFn(const void *p1, const void *p2, const void *param) {
int32_t bytes = 0; int32_t bytes = 0;
if (colIndex == TSDB_TBNAME_COLUMN_INDEX) { if (colIndex == TSDB_TBNAME_COLUMN_INDEX) {
f1 = pTable1->name; f1 = (char*) pTable1->name;
f2 = pTable2->name; f2 = (char*) pTable2->name;
type = TSDB_DATA_TYPE_BINARY; type = TSDB_DATA_TYPE_BINARY;
bytes = TSDB_TABLE_NAME_LEN; bytes = TSDB_TABLE_NAME_LEN + VARSTR_HEADER_SIZE;
} else { } else {
f1 = dataRowTuple(pTable1->tagVal); STColumn* pCol = schemaColAt(pTableGroupSupp->pTagSchema, colIndex);
f2 = dataRowTuple(pTable2->tagVal); bytes = pCol->bytes;
type = schemaColAt(pTableGroupSupp->pTagSchema, colIndex)->type; f1 = tdGetRowDataOfCol(pTable1->tagVal, pCol->type, TD_DATA_ROW_HEAD_SIZE + pCol->offset);
bytes = schemaColAt(pTableGroupSupp->pTagSchema, colIndex)->bytes; f2 = tdGetRowDataOfCol(pTable2->tagVal, pCol->type, TD_DATA_ROW_HEAD_SIZE + pCol->offset);
} }
int32_t ret = doCompare(f1, f2, type, bytes); 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) { bool tSkipListNodeFilterCallback(const void* pNode, void* param) {
tQueryInfo* pInfo = (tQueryInfo*) 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; char* val = NULL;
int8_t type = pInfo->sch.type; int8_t type = pInfo->sch.type;
if (pInfo->colIndex == TSDB_TBNAME_COLUMN_INDEX) { if (pInfo->colIndex == TSDB_TBNAME_COLUMN_INDEX) {
val = pTable->name; val = (char*) elem->pTable->name;
type = TSDB_DATA_TYPE_BINARY; type = TSDB_DATA_TYPE_BINARY;
} else { } else {
STSchema* pTSchema = (STSchema*) pInfo->param; // todo table schema is identical to stable schema?? STSchema* pTSchema = (STSchema*) pInfo->param; // todo table schema is identical to stable schema??
int32_t offset = pTSchema->columns[pInfo->colIndex].offset; 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(elem->pTable->tagVal, pInfo->sch.type, TD_DATA_ROW_HEAD_SIZE + offset);
val = tdGetRowDataOfCol(pTable->tagVal, pInfo->sch.type, TD_DATA_ROW_HEAD_SIZE + offset);
} else {
val = dataRowTuple(pTable->tagVal) + offset;
}
} }
int32_t ret = 0; int32_t ret = 0;
@ -1456,8 +1458,6 @@ bool tSkipListNodeFilterCallback(const void* pNode, void* param) {
ret = pInfo->compare(val, pInfo->q); ret = pInfo->compare(val, pInfo->q);
} }
} else { } else {
// tVariant t = {0};
// tVariantCreateFromBinary(&t, val, (uint32_t)pInfo->sch.bytes, type);
ret = pInfo->compare(val, pInfo->q); ret = pInfo->compare(val, pInfo->q);
} }
@ -1502,7 +1502,7 @@ static int32_t doQueryTableList(STable* pSTable, SArray* pRes, tExprNode* pExpr)
.pExtInfo = pSTable->tagSchema, .pExtInfo = pSTable->tagSchema,
}; };
SArray* pTableList = taosArrayInit(8, POINTER_BYTES); SArray* pTableList = taosArrayInit(8, sizeof(STableIndexElem));
tExprTreeTraverse(pExpr, pSTable->pIndex, pTableList, &supp); tExprTreeTraverse(pExpr, pSTable->pIndex, pTableList, &supp);
tExprTreeDestroy(&pExpr, destroyHelper); 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) { if (leftPartEnd > start) {
tqsortImpl(src, size, start, leftPartEnd, param, comparFn, buf); tqsortImpl(src, start, leftPartEnd, size, param, comparFn, buf);
} }
if (rightPartStart < end) { 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) { int32_t compareLenPrefixedStr(const void *pLeft, const void *pRight) {
return (int32_t)strcmp(pLeft, 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) { 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) { } else if (optr == TSDB_RELATION_IN) {
comparFn = compareFindStrInArray; comparFn = compareFindStrInArray;
} else { /* normal relational comparFn */ } else { /* normal relational comparFn */
comparFn = compareStrVal; comparFn = compareLenPrefixedStr;
} }
break; break;
@ -296,6 +303,7 @@ __compar_fn_t getKeyComparFunc(int32_t keyType) {
switch (keyType) { switch (keyType) {
case TSDB_DATA_TYPE_TINYINT: case TSDB_DATA_TYPE_TINYINT:
case TSDB_DATA_TYPE_BOOL:
comparFn = compareInt8Val; comparFn = compareInt8Val;
break; break;
case TSDB_DATA_TYPE_SMALLINT: case TSDB_DATA_TYPE_SMALLINT:
@ -308,17 +316,13 @@ __compar_fn_t getKeyComparFunc(int32_t keyType) {
case TSDB_DATA_TYPE_TIMESTAMP: case TSDB_DATA_TYPE_TIMESTAMP:
comparFn = compareInt64Val; comparFn = compareInt64Val;
break; break;
case TSDB_DATA_TYPE_BOOL:
comparFn = compareInt32Val;
break;
case TSDB_DATA_TYPE_FLOAT: case TSDB_DATA_TYPE_FLOAT:
case TSDB_DATA_TYPE_DOUBLE: case TSDB_DATA_TYPE_DOUBLE:
comparFn = compareDoubleVal; comparFn = compareDoubleVal;
break; break;
case TSDB_DATA_TYPE_BINARY: case TSDB_DATA_TYPE_BINARY:
comparFn = compareStrVal; comparFn = compareLenPrefixedStr;
break; break;
case TSDB_DATA_TYPE_NCHAR: 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; return (ret < 0) ? -1 : 1;
} }
default: { default: { // todo refactor
int32_t ret = strncmp(f1, f2, (size_t)size); tstr* t1 = (tstr*) f1;
if (ret == 0) { tstr* t2 = (tstr*) f2;
return ret;
}
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;
}
}
} }
} }
} }

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 insert into $tb values (now, 1)
sql select * from $tb sql select * from $tb
print ===>rows $rows, data $data01 #print ===>rows $rows, data $data01
if $rows != 1 then if $rows != 1 then
print expect 1, actual: $rows
return -1 return -1
endi endi
if $data01 != 1 then if $data01 != 1 then
print expect 1 actual: $data01
return -1 return -1
endi endi
@ -67,7 +71,9 @@ print ===>rows $rows, data $data01
if $rows != 1 then if $rows != 1 then
return -1 return -1
endi endi
if $data01 != 1 then if $data01 != 1 then
print expect 1, actual $data01
return -1 return -1
endi endi
@ -80,6 +86,7 @@ if $data01 != 1 then
return -1 return -1
endi endi
if $data02 != 3 then if $data02 != 3 then
print expect 3 actual: $data02
return -1 return -1
endi endi

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

@ -132,7 +132,9 @@ endi
print =============== step6 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 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 $data00 $data01 $data02 $data03 $data04 $data05 $data06
print $data10 $data11 $data12 $data13 $data14 $data15 $data16
if $data00 != 100 then if $data00 != 100 then
print expect 100, actual $data00
return -1 return -1
endi endi

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

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

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

@ -1,113 +1,97 @@
# run general/parser/alter.sim #run general/parser/alter.sim
# sleep 2000 #sleep 2000
# run general/parser/alter1.sim #run general/parser/alter1.sim
# sleep 2000 #sleep 2000
# run general/parser/alter_stable.sim #run general/parser/alter_stable.sim
# sleep 2000 #sleep 2000
# run general/parser/auto_create_tb.sim #run general/parser/auto_create_tb.sim
# sleep 2000 #sleep 2000
# run general/parser/auto_create_tb_drop_tb.sim #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 sleep 2000
run general/parser/col_arithmetic_operation.sim run general/parser/col_arithmetic_operation.sim
sleep 2000 sleep 2000
run general/parser/columnValue_bigint.sim run general/parser/columnValue.sim
sleep 2000 sleep 2000
run general/parser/columnValue_bool.sim run general/parser/commit.sim
sleep 2000 sleep 2000
run general/parser/columnValue_double.sim run general/parser/create_db.sim
sleep 2000 sleep 2000
run general/parser/columnValue_float.sim run general/parser/create_mt.sim
sleep 2000 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
# 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 run general/parser/select_from_cache_disk.sim
sleep 2000 sleep 2000
# run general/parser/set_tag_vals.sim run general/parser/set_tag_vals.sim
# sleep 2000 sleep 2000
# run general/parser/single_row_in_tb.sim run general/parser/single_row_in_tb.sim
# sleep 2000 sleep 2000
# run general/parser/slimit.sim run general/parser/slimit.sim
# sleep 2000 sleep 2000
run general/parser/slimit1.sim run general/parser/slimit1.sim
sleep 2000 sleep 2000
run general/parser/slimit1_query.sim run general/parser/slimit_alter_tags.sim
sleep 2000 sleep 2000
# run general/parser/slimit_alter_tags.sim run general/parser/stream_on_sys.sim
# sleep 2000 sleep 2000
# run general/parser/stream_on_sys.sim run general/parser/stream.sim
# sleep 2000 sleep 2000
# run general/parser/stream.sim run general/parser/tbnameIn.sim
# sleep 2000 sleep 2000
# run general/parser/tbnameIn.sim run general/parser/where.sim
# sleep 2000 sleep 2000
# run general/parser/where.sim #run general/parser/repeatAlter.sim
# sleep 2000 sleep 2000
# #run general/parser/repeatAlter.sim #run general/parser/repeatStream.sim
# sleep 2000 sleep 2000
# #run general/parser/repeatStream.sim run general/parser/join.sim
# sleep 2000 sleep 2000
# run general/parser/join.sim run general/parser/join_multivnode.sim
# run general/parser/join_multivnode.sim sleep 2000
# run general/parser/projection_limit_offset.sim run general/parser/projection_limit_offset.sim
# sleep 2000 sleep 2000
# run general/parser/select_with_tags.sim run general/parser/select_with_tags.sim
# run general/parser/groupby.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 endi
sql select * from $mt where tgcol = 1 sql select * from $mt where tgcol = 1
if $rows != 100 then if $rows != 100 then
print expect 100, actual:$rows
return -1 return -1
endi endi
sql select * from $mt where tgcol <> 1 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 sql select count(tbcol) as c from $mt where ts > 1000 group by tgcol
print $data00 $data01 $data02 $data03 $data04 $data05 $data06 print $data00 $data01 $data02 $data03 $data04 $data05 $data06
if $data00 != 100 then if $data00 != 100 then
print expect 100, actual $data00
return -1 return -1
endi endi

3
tests/script/test.sh
View File

@ -99,7 +99,8 @@ echo "cDebugFlag 135" >> $TAOS_CFG
echo "httpDebugFlag 135" >> $TAOS_CFG echo "httpDebugFlag 135" >> $TAOS_CFG
echo "monitorDebugFlag 135" >> $TAOS_CFG echo "monitorDebugFlag 135" >> $TAOS_CFG
echo "udebugFlag 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 "asyncLog 0" >> $TAOS_CFG
echo "locale en_US.UTF-8" >> $TAOS_CFG echo "locale en_US.UTF-8" >> $TAOS_CFG
echo " " >> $TAOS_CFG echo " " >> $TAOS_CFG