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

Merge branch '2.0' into refact/slguan

This commit is contained in:
slguan 2020-03-24 15:39:52 +08:00
parent c541d35462 a863cd8675
commit 930f4f169a
9 changed files with 580 additions and 580 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -417,6 +417,10 @@ static void count_function(SQLFunctionCtx *pCtx) {
numOfElem += 1;
}
} else {
/*
* when counting on the primary time stamp column and no statistics data is provided,
* simple use the size value
*/
numOfElem = pCtx->size;
}
}

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

@ -273,8 +273,8 @@ void tscProcessMsgFromServer(SRpcMsg *rpcMsg) {
}
pSql->retry = 0;
pRes->rspLen = 0;
if (pRes->code != TSDB_CODE_QUERY_CANCELLED) {
pRes->code = (rpcMsg->code != TSDB_CODE_SUCCESS) ? rpcMsg->code : TSDB_CODE_NETWORK_UNAVAIL;
} else {
@ -283,7 +283,7 @@ void tscProcessMsgFromServer(SRpcMsg *rpcMsg) {
if (pRes->code != TSDB_CODE_QUERY_CANCELLED) {
assert(rpcMsg->msgType == pCmd->msgType + 1);
pRes->code = (int32_t)rpcMsg->code;
pRes->code = rpcMsg->code;
pRes->rspType = rpcMsg->msgType;
pRes->rspLen = rpcMsg->contLen;
@ -662,12 +662,16 @@ int tscBuildQueryMsg(SSqlObj *pSql, SSqlInfo *pInfo) {
SQueryInfo *pQueryInfo = tscGetQueryInfoDetail(pCmd, pCmd->clauseIndex);
STableMetaInfo *pTableMetaInfo = tscGetMetaInfo(pQueryInfo, 0);
char *pStart = pCmd->payload + tsRpcHeadSize;
STableMeta * pTableMeta = pTableMetaInfo->pTableMeta;
SSuperTableMeta *pMetricMeta = pTableMetaInfo->pMetricMeta;
if (pQueryInfo->colList.numOfCols <= 0) {
tscError("%p illegal value of numOfCols in query msg: %d", pSql, tscGetNumOfColumns(pTableMeta));
return -1;
}
char *pStart = pCmd->payload + tsRpcHeadSize;
SQueryTableMsg *pQueryMsg = (SQueryTableMsg *)pStart;
int32_t msgLen = 0;
@ -675,11 +679,7 @@ int tscBuildQueryMsg(SSqlObj *pSql, SSqlInfo *pInfo) {
if (UTIL_TABLE_IS_NOMRAL_TABLE(pTableMetaInfo)) {
numOfTables = 1;
pQueryMsg->uid = pTableMeta->uid;
pQueryMsg->numOfTagsCols = 0;
pQueryMsg->vgId = htonl(pTableMeta->vgId);
pQueryMsg->head.vgId = htonl(pTableMeta->vgId);
tscTrace("%p queried tables:%d, table id: %s", pSql, 1, pTableMetaInfo->name);
} else { // query on super table
if (pTableMetaInfo->vnodeIndex < 0) {
@ -697,11 +697,10 @@ int tscBuildQueryMsg(SSqlObj *pSql, SSqlInfo *pInfo) {
}
tscTrace("%p query on vid:%d, number of tables:%d", pSql, vnodeId, numOfTables);
pQueryMsg->vgId = htons(vnodeId);
pQueryMsg->head.vgId = htons(vnodeId);
}
pQueryMsg->numOfTables = htonl(numOfTables);
pQueryMsg->numOfTagsCols = htons(pTableMetaInfo->numOfTags);
if (pQueryInfo->order.order == TSQL_SO_ASC) {
pQueryMsg->window.skey = htobe64(pQueryInfo->stime);
@ -713,22 +712,13 @@ int tscBuildQueryMsg(SSqlObj *pSql, SSqlInfo *pInfo) {
pQueryMsg->order = htons(pQueryInfo->order.order);
pQueryMsg->orderColId = htons(pQueryInfo->order.orderColId);
pQueryMsg->interpoType = htons(pQueryInfo->interpoType);
pQueryMsg->limit = htobe64(pQueryInfo->limit.limit);
pQueryMsg->offset = htobe64(pQueryInfo->limit.offset);
pQueryMsg->numOfCols = htons(pQueryInfo->colList.numOfCols);
if (pQueryInfo->colList.numOfCols <= 0) {
tscError("%p illegal value of numOfCols in query msg: %d", pSql, tscGetNumOfColumns(pTableMeta));
return -1;
}
pQueryMsg->intervalTime = htobe64(pQueryInfo->intervalTime);
pQueryMsg->slidingTimeUnit = pQueryInfo->slidingTimeUnit;
pQueryMsg->slidingTime = htobe64(pQueryInfo->slidingTime);
pQueryMsg->slidingTimeUnit = pQueryInfo->slidingTimeUnit;
if (pQueryInfo->intervalTime < 0) {
tscError("%p illegal value of aggregation time interval in query msg: %ld", pSql, pQueryInfo->intervalTime);
@ -866,25 +856,7 @@ int tscBuildQueryMsg(SSqlObj *pSql, SSqlInfo *pInfo) {
pQueryMsg->colNameLen = htonl(len);
// serialize the table info (sid, uid, tags)
pMsg = doSerializeTableInfo(pSql, numOfTables, htons(pQueryMsg->vgId), pMsg);
// only include the required tag column schema. If a tag is not required, it won't be sent to vnode
if (pTableMetaInfo->numOfTags > 0) {
// always transfer tag schema to vnode if exists
SSchema *pTagSchema = tscGetTableTagSchema(pTableMeta);
for (int32_t j = 0; j < pTableMetaInfo->numOfTags; ++j) {
if (pTableMetaInfo->tagColumnIndex[j] == TSDB_TBNAME_COLUMN_INDEX) {
SSchema tbSchema = {
.bytes = TSDB_TABLE_NAME_LEN, .colId = TSDB_TBNAME_COLUMN_INDEX, .type = TSDB_DATA_TYPE_BINARY};
memcpy(pMsg, &tbSchema, sizeof(SSchema));
} else {
memcpy(pMsg, &pTagSchema[pTableMetaInfo->tagColumnIndex[j]], sizeof(SSchema));
}
pMsg += sizeof(SSchema);
}
}
pMsg = doSerializeTableInfo(pSql, numOfTables, htons(pQueryMsg->head.vgId), pMsg);
SSqlGroupbyExpr *pGroupbyExpr = &pQueryInfo->groupbyExpr;
if (pGroupbyExpr->numOfGroupCols != 0) {
@ -948,8 +920,7 @@ int tscBuildQueryMsg(SSqlObj *pSql, SSqlInfo *pInfo) {
pCmd->payloadLen = msgLen;
pSql->cmd.msgType = TSDB_MSG_TYPE_QUERY;
pQueryMsg->contLen = htonl(msgLen);
pQueryMsg->head.contLen = htonl(msgLen);
assert(msgLen + minMsgSize() <= size);
return TSDB_CODE_SUCCESS;
@ -2358,7 +2329,6 @@ int tscProcessRetrieveRspFromVnode(SSqlObj *pSql) {
}
pRes->row = 0;
tscTrace("%p numOfRows:%d, offset:%d", pSql, pRes->numOfRows, pRes->offset);
return 0;

34
src/dnode/src/dnodeRead.c
View File

@ -254,7 +254,7 @@ static void dnodeProcessQueryMsg(SReadMsg *pMsg) {
static void dnodeProcessRetrieveMsg(SReadMsg *pMsg) {
SRetrieveTableMsg *pRetrieve = pMsg->pCont;
void *pQInfo = htobe64(pRetrieve->qhandle);
void *pQInfo = (void*) htobe64(pRetrieve->qhandle);
dTrace("QInfo:%p vgId:%d, retrieve msg is received", pQInfo, pRetrieve->header.vgId);
@ -263,37 +263,18 @@ static void dnodeProcessRetrieveMsg(SReadMsg *pMsg) {
int32_t contLen = 0;
SRpcMsg rpcRsp = {0};
SRetrieveTableRsp *pRsp = NULL;
int32_t code = qRetrieveQueryResultInfo(pQInfo, &numOfRows, &rowSize);
if (code != TSDB_CODE_SUCCESS) {
contLen = sizeof(SRetrieveTableRsp);
SRetrieveTableRsp *pRsp = (SRetrieveTableRsp *)rpcMallocCont(contLen);
pRsp->numOfRows = 0;
pRsp->precision = 0;
pRsp->offset = 0;
pRsp->useconds = 0;
rpcRsp = (SRpcMsg) {
.handle = pMsg->rpcMsg.handle,
.pCont = pRsp,
.contLen = contLen,
.code = code,
.msgType = 0
};
//todo free qinfo
pRsp = (SRetrieveTableRsp *)rpcMallocCont(contLen);
memset(pRsp, 0, sizeof(SRetrieveTableRsp));
} else {
contLen = 100;
SRetrieveTableRsp *pRsp = (SRetrieveTableRsp *)rpcMallocCont(contLen);
pRsp->numOfRows = htonl(1);
pRsp->precision = htons(0);
pRsp->offset = htobe64(0);
pRsp->useconds = htobe64(0);
// todo set the data
*(int64_t*) pRsp->data = 1000;
// todo check code and handle error in build result set
code = qDumpRetrieveResult(pQInfo, &pRsp, &contLen);
}
rpcRsp = (SRpcMsg) {
.handle = pMsg->rpcMsg.handle,
@ -302,7 +283,6 @@ static void dnodeProcessRetrieveMsg(SReadMsg *pMsg) {
.code = code,
.msgType = 0
};
}
rpcSendResponse(&rpcRsp);
}

27
src/inc/taosmsg.h
View File

@ -20,8 +20,8 @@
extern "C" {
#endif
#include <stdint.h>
#include <stdbool.h>
#include <stdint.h>
#include "taosdef.h"
#include "taoserror.h"
@ -472,47 +472,30 @@ typedef struct STimeWindow {
* the outputCols will be 3 while the numOfCols is 1.
*/
typedef struct {
int32_t contLen; // msg header
int16_t vgId;
int32_t numOfTables;
uint64_t uid;
SMsgHead head;
STimeWindow window;
int32_t numOfTables;
int16_t order;
int16_t orderColId;
int16_t numOfCols; // the number of columns will be load from vnode
char slidingTimeUnit; // time interval type, for revisement of interval(1d)
int64_t intervalTime; // time interval for aggregation, in million second
int64_t intervalOffset; // start offset for interval query
int64_t slidingTime; // value for sliding window
// tag schema, used to parse tag information in pSidExtInfo
uint64_t pTagSchema;
int16_t numOfTagsCols; // required number of tags
char slidingTimeUnit; // time interval type, for revisement of interval(1d)
int16_t tagLength; // tag length in current query
int16_t numOfGroupCols; // num of group by columns
int16_t orderByIdx;
int16_t orderType; // used in group by xx order by xxx
uint64_t groupbyTagIds;
int64_t limit;
int64_t offset;
int16_t queryType; // denote another query process
int16_t numOfOutputCols; // final output columns numbers
int16_t interpoType; // interpolate type
uint64_t defaultVal; // default value array list
int32_t colNameLen;
int64_t colNameList;
int64_t pSqlFuncExprs;
int32_t tsOffset; // offset value in current msg body, NOTE: ts list is compressed
int32_t tsLen; // total length of ts comp block
int32_t tsNumOfBlocks; // ts comp block numbers

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

@ -175,7 +175,6 @@ typedef struct SQInfo {
TSKEY startTime;
int64_t elapsedTime;
SResultRec rec;
int32_t pointsReturned;
int32_t pointsInterpo;
int32_t code; // error code to returned to client
int32_t killed; // denotes if current query is killed
@ -184,7 +183,6 @@ typedef struct SQInfo {
SQueryRuntimeEnv runtimeEnv;
int32_t subgroupIdx;
int32_t offset; /* offset in group result set of subgroup */
// tSidSet* pSidSet;
T_REF_DECLARE()
/*
@ -226,7 +224,12 @@ void qSuperTableQuery(void* pReadMsg);
*/
int32_t qRetrieveQueryResultInfo(SQInfo* pQInfo, int32_t *numOfRows, int32_t* rowsize);
//int32_t qBuildQueryResult(SQInfo* pQInfo, void* pBuf);
/**
*
* @param pQInfo
* @param pRsp
* @return
*/
int32_t qDumpRetrieveResult(SQInfo *pQInfo, SRetrieveTableRsp** pRsp, int32_t* contLen);
#endif // TDENGINE_QUERYEXECUTOR_H

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

@ -219,7 +219,7 @@ typedef struct SQLAggFuncElem {
void (*distSecondaryMergeFunc)(SQLFunctionCtx *pCtx);
int32_t (*dataReqFunc)(SQLFunctionCtx *pCtx, TSKEY start, TSKEY end, int32_t colId, int32_t blockStatus);
int32_t (*dataReqFunc)(SQLFunctionCtx *pCtx, TSKEY start, TSKEY end, int32_t colId);
} SQLAggFuncElem;
typedef struct SPatternCompareInfo {

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

@ -119,7 +119,7 @@ static int32_t flushFromResultBuf(SQInfo *pQInfo);
static bool functionNeedToExecute(SQueryRuntimeEnv *pRuntimeEnv, SQLFunctionCtx *pCtx, int32_t functionId);
static void getNextTimeWindow(SQuery *pQuery, STimeWindow *pTimeWindow);
static void setExecParams(SQuery *pQuery, SQLFunctionCtx *pCtx, void *inputData, char *primaryColumnData, int32_t size,
static void setExecParams(SQuery *pQuery, SQLFunctionCtx *pCtx, void *inputData, TSKEY *tsCol, int32_t size,
int32_t functionId, SDataStatis *pStatis, bool hasNull, void *param, int32_t scanFlag);
static void initCtxOutputBuf(SQueryRuntimeEnv *pRuntimeEnv);
static void destroyMeterQueryInfo(STableQueryInfo *pTableQueryInfo, int32_t numOfCols);
@ -427,15 +427,9 @@ static bool isTopBottomQuery(SQuery *pQuery) {
return false;
}
static SDataStatis *getStatisInfo(SQuery *pQuery, SDataStatis *pStatis, SDataBlockInfo *pDataBlockInfo,
int32_t columnIndex) {
// no SField info exist, or column index larger than the output column, no result.
if (pStatis == NULL) {
return NULL;
}
static SDataStatis *getStatisInfo(SQuery *pQuery, SDataStatis *pStatis, SDataBlockInfo *pDataBlockInfo, int32_t index) {
// for a tag column, no corresponding field info
SColIndexEx *pColIndexEx = &pQuery->pSelectExpr[columnIndex].pBase.colInfo;
SColIndexEx *pColIndexEx = &pQuery->pSelectExpr[index].pBase.colInfo;
if (TSDB_COL_IS_TAG(pColIndexEx->flag)) {
return NULL;
}
@ -453,13 +447,31 @@ static SDataStatis *getStatisInfo(SQuery *pQuery, SDataStatis *pStatis, SDataBlo
return NULL;
}
/**
* @param pQuery
* @param col
* @param pDataBlockInfo
* @param pStatis
* @param pColStatis
* @return
*/
static bool hasNullValue(SQuery *pQuery, int32_t col, SDataBlockInfo *pDataBlockInfo, SDataStatis *pStatis,
SDataStatis **pColStatis) {
if (TSDB_COL_IS_TAG(pQuery->pSelectExpr[col].pBase.colInfo.flag) || pStatis == NULL) {
SColIndexEx* pColIndex = &pQuery->pSelectExpr[col].pBase.colInfo;
if (TSDB_COL_IS_TAG(pColIndex->flag)) {
return false;
}
// query on primary timestamp column, not null value at all
if (pColIndex->colId == PRIMARYKEY_TIMESTAMP_COL_INDEX) {
return false;
}
*pColStatis = NULL;
if (pStatis != NULL) {
*pColStatis = getStatisInfo(pQuery, pStatis, pDataBlockInfo, col);
}
if ((*pColStatis) != NULL && (*pColStatis)->numOfNull == 0) {
return false;
}
@ -887,7 +899,6 @@ char *getDataBlocks(SQueryRuntimeEnv *pRuntimeEnv, SArithmeticSupport *sas, int3
* 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.
*/
if (pDataBlock == NULL) {
return NULL;
}
@ -939,10 +950,11 @@ static int32_t blockwiseApplyAllFunctions(SQueryRuntimeEnv *pRuntimeEnv, SDataSt
int32_t functionId = pQuery->pSelectExpr[k].pBase.functionId;
SDataStatis *tpField = NULL;
bool hasNull = hasNullValue(pQuery, k, pDataBlockInfo, pStatis, &tpField);
char *dataBlock = getDataBlocks(pRuntimeEnv, &sasArray[k], k, pDataBlockInfo->size, pDataBlock);
setExecParams(pQuery, &pCtx[k], dataBlock, (char *)primaryKeyCol, pDataBlockInfo->size, functionId, tpField,
setExecParams(pQuery, &pCtx[k], dataBlock, primaryKeyCol, pDataBlockInfo->size, functionId, tpField,
hasNull, &sasArray[k], pRuntimeEnv->scanFlag);
}
@ -1381,7 +1393,7 @@ static int32_t tableApplyFunctionsOnBlock(SQueryRuntimeEnv *pRuntimeEnv, SDataBl
return 0;
}
void setExecParams(SQuery *pQuery, SQLFunctionCtx *pCtx, void *inputData, char *primaryColumnData, int32_t size,
void setExecParams(SQuery *pQuery, SQLFunctionCtx *pCtx, void *inputData, TSKEY *tsCol, int32_t size,
int32_t functionId, SDataStatis *pStatis, bool hasNull, void *param, int32_t scanFlag) {
pCtx->scanFlag = scanFlag;
@ -1396,15 +1408,15 @@ void setExecParams(SQuery *pQuery, SQLFunctionCtx *pCtx, void *inputData, char *
pCtx->preAggVals.isSet = false;
}
if ((aAggs[functionId].nStatus & TSDB_FUNCSTATE_SELECTIVITY) != 0 && (primaryColumnData != NULL)) {
pCtx->ptsList = (int64_t *)(primaryColumnData);
if ((aAggs[functionId].nStatus & TSDB_FUNCSTATE_SELECTIVITY) != 0 && (tsCol != NULL)) {
pCtx->ptsList = tsCol;
}
if (functionId >= TSDB_FUNC_FIRST_DST && functionId <= TSDB_FUNC_LAST_DST) {
// last_dist or first_dist function
// store the first&last timestamp into the intermediate buffer [1], the true
// value may be null but timestamp will never be null
pCtx->ptsList = (int64_t *)(primaryColumnData);
pCtx->ptsList = tsCol;
} else if (functionId == TSDB_FUNC_TOP || functionId == TSDB_FUNC_BOTTOM || functionId == TSDB_FUNC_TWA ||
functionId == TSDB_FUNC_DIFF || (functionId >= TSDB_FUNC_RATE && functionId <= TSDB_FUNC_AVG_IRATE)) {
/*
@ -1420,7 +1432,7 @@ void setExecParams(SQuery *pQuery, SQLFunctionCtx *pCtx, void *inputData, char *
pTWAInfo->EKey = pQuery->window.ekey;
}
pCtx->ptsList = (int64_t *)(primaryColumnData);
pCtx->ptsList = tsCol;
} else if (functionId == TSDB_FUNC_ARITHM) {
pCtx->param[1].pz = param;
@ -2525,21 +2537,17 @@ static void getNextTimeWindow(SQuery *pQuery, STimeWindow *pTimeWindow) {
SArray *loadDataBlockOnDemand(SQueryRuntimeEnv *pRuntimeEnv, SDataBlockInfo *pBlockInfo, SDataStatis **pStatis) {
SQuery *pQuery = pRuntimeEnv->pQuery;
tsdb_query_handle_t pQueryHandle = pRuntimeEnv->pQueryHandle;
uint32_t r = 0;
SArray * pDataBlock = NULL;
// STimeWindow *w = &pQueryHandle->window;
if (pQuery->numOfFilterCols > 0) {
r = BLK_DATA_ALL_NEEDED;
} else {
for (int32_t i = 0; i < pQuery->numOfOutputCols; ++i) {
int32_t functionId = pQuery->pSelectExpr[i].pBase.functionId;
int32_t colId = pQuery->pSelectExpr[i].pBase.colInfo.colId;
// r |= aAggs[functionId].dataReqFunc(&pRuntimeEnv->pCtx[i], w->skey, w->ekey, colId);
r |= aAggs[functionId].dataReqFunc(&pRuntimeEnv->pCtx[i], pQuery->window.skey, pQuery->window.ekey, colId);
}
if (pRuntimeEnv->pTSBuf > 0 || isIntervalQuery(pQuery)) {
@ -2690,10 +2698,8 @@ static int64_t doScanAllDataBlocks(SQueryRuntimeEnv *pRuntimeEnv) {
int32_t forwardStep = tableApplyFunctionsOnBlock(pRuntimeEnv, &blockInfo, pStatis, binarySearchForKey, &numOfRes,
&pRuntimeEnv->windowResInfo, pDataBlock);
// dTrace("QInfo:%p check data block, brange:%" PRId64 "-%" PRId64 ", fileId:%d, slot:%d, pos:%d, rows:%d,
// checked:%d",
// GET_QINFO_ADDR(pQuery), blockInfo.window.skey, blockInfo.window.ekey, pQueryHandle->cur.fileId,
// pQueryHandle->cur.slot, pQuery->pos, blockInfo.size, forwardStep);
dTrace("QInfo:%p check data block, brange:%" PRId64 "-%" PRId64 ", rows:%d",
GET_QINFO_ADDR(pRuntimeEnv), blockInfo.window.skey, blockInfo.window.ekey, blockInfo.size);
// save last access position
cnt += forwardStep;
@ -3036,7 +3042,7 @@ void copyResToQueryResultBuf(SQInfo *pQInfo, SQuery *pQuery) {
total += pData->numOfElems;
}
pQuery->sdata[0]->num = total;
int32_t rows = total;
int32_t offset = 0;
for (int32_t num = 0; num < list.size; ++num) {
@ -3044,7 +3050,7 @@ void copyResToQueryResultBuf(SQInfo *pQInfo, SQuery *pQuery) {
for (int32_t i = 0; i < pQuery->numOfOutputCols; ++i) {
int32_t bytes = pRuntimeEnv->pCtx[i].outputBytes;
char * pDest = pQuery->sdata[i]->data;
char * pDest = pQuery->sdata[i];
memcpy(pDest + offset * bytes, pData->data + pRuntimeEnv->offset[i] * pData->numOfElems,
bytes * pData->numOfElems);
@ -3055,7 +3061,7 @@ void copyResToQueryResultBuf(SQInfo *pQInfo, SQuery *pQuery) {
assert(pQuery->rec.pointsRead == 0);
pQuery->rec.pointsRead += pQuery->sdata[0]->num;
pQuery->rec.pointsRead += rows;
pQInfo->offset += 1;
}
@ -3222,7 +3228,8 @@ int32_t flushFromResultBuf(SQInfo *pQInfo) {
// the base value for group result, since the maximum number of table for each vnode will not exceed 100,000.
int32_t pageId = -1;
int32_t remain = pQuery->sdata[0]->num;
assert(0);
int32_t remain = 0;//pQuery->sdata[0]->num;
int32_t offset = 0;
while (remain > 0) {
@ -3253,12 +3260,12 @@ int32_t flushFromResultBuf(SQInfo *pQInfo) {
void resetMergeResultBuf(SQuery *pQuery, SQLFunctionCtx *pCtx, SResultInfo *pResultInfo) {
for (int32_t k = 0; k < pQuery->numOfOutputCols; ++k) {
pCtx[k].aOutputBuf = pQuery->sdata[k]->data - pCtx[k].outputBytes;
pCtx[k].aOutputBuf = pQuery->sdata[k] - pCtx[k].outputBytes;
pCtx[k].size = 1;
pCtx[k].startOffset = 0;
pCtx[k].resultInfo = &pResultInfo[k];
pQuery->sdata[k]->num = 0;
pQuery->sdata[k] = 0;
}
}
@ -4080,11 +4087,6 @@ int32_t vnodeGetResultSize(void *thandle, int32_t *numOfRows) {
}
}
int64_t vnodeGetOffsetVal(void *thandle) {
SQInfo *pQInfo = (SQInfo *)thandle;
return pQInfo->runtimeEnv.pQuery->limit.offset;
}
bool vnodeHasRemainResults(void *handle) {
SQInfo *pQInfo = (SQInfo *)handle;
@ -4162,62 +4164,13 @@ static int32_t resultInterpolate(SQInfo *pQInfo, tFilePage **data, tFilePage **p
}
static void doCopyQueryResultToMsg(SQInfo *pQInfo, int32_t numOfRows, char *data) {
#if 0
SMeterObj *pObj = pQInfo->pObj;
SQuery * pQuery = &pQInfo->query;
int tnumOfRows = vnodeList[pObj->vnode].cfg.rowsInFileBlock;
// for metric query, bufIndex always be 0.
for (int32_t col = 0; col < pQuery->numOfOutputCols; ++col) { // pQInfo->bufIndex == 0
SQuery *pQuery = pQInfo->runtimeEnv.pQuery;
for (int32_t col = 0; col < pQuery->numOfOutputCols; ++col) {
int32_t bytes = pQuery->pSelectExpr[col].resBytes;
memmove(data, pQuery->sdata[col]->data, bytes * numOfRows);
data += bytes * numOfRows;
}
#endif
}
/**
* Copy the result data/file to output message buffer.
* If the result is in file format, read file from disk and copy to output buffer, compression is not involved since
* data in file is already compressed.
* In case of other result in buffer, compress the result before copy once the tsComressMsg is set.
*
* @param handle
* @param data
* @param numOfRows the number of rows that are not returned in current retrieve
* @return
*/
int32_t vnodeCopyQueryResultToMsg(void *handle, char *data, int32_t numOfRows) {
SQInfo *pQInfo = (SQInfo *)handle;
SQuery *pQuery = pQInfo->runtimeEnv.pQuery;
assert(pQuery->pSelectExpr != NULL && pQuery->numOfOutputCols > 0);
// load data from file to msg buffer
if (isTSCompQuery(pQuery)) {
int32_t fd = open(pQuery->sdata[0]->data, O_RDONLY, 0666);
// make sure file exist
if (FD_VALID(fd)) {
size_t s = lseek(fd, 0, SEEK_END);
dTrace("QInfo:%p ts comp data return, file:%s, size:%zu", pQInfo, pQuery->sdata[0]->data, s);
lseek(fd, 0, SEEK_SET);
read(fd, data, s);
close(fd);
unlink(pQuery->sdata[0]->data);
} else {
dError("QInfo:%p failed to open tmp file to send ts-comp data to client, path:%s, reason:%s", pQInfo,
pQuery->sdata[0]->data, strerror(errno));
}
} else {
doCopyQueryResultToMsg(pQInfo, numOfRows, data);
}
return numOfRows;
}
int32_t vnodeQueryResultInterpolate(SQInfo *pQInfo, tFilePage **pDst, tFilePage **pDataSrc, int32_t numOfRows,
@ -4308,7 +4261,7 @@ void vnodePrintQueryStatistics(SQInfo *pQInfo) {
#endif
}
int32_t vnodeQueryTablePrepare(SQInfo *pQInfo, void *param, void* tsdb) {
int32_t initQInfo(SQInfo *pQInfo, void *param, void* tsdb) {
SQuery *pQuery = pQInfo->runtimeEnv.pQuery;
int32_t code = TSDB_CODE_SUCCESS;
@ -4342,7 +4295,6 @@ int32_t vnodeQueryTablePrepare(SQInfo *pQInfo, void *param, void* tsdb) {
taosArrayPush(cols, &pQuery->colList[i]);
}
pQInfo->runtimeEnv.pQueryHandle = tsdbQueryByTableId(tsdb, &cond, pQInfo->pTableIdList, cols);
SQueryRuntimeEnv *pRuntimeEnv = &pQInfo->runtimeEnv;
@ -4993,8 +4945,7 @@ static void vnodeMultiMeterQueryProcessor(SQInfo *pQInfo) {
// vnodePrintQueryStatistics(pSupporter);
}
dTrace("QInfo:%p points returned:%d, totalRead:%d totalReturn:%d", pQInfo, pQuery->rec.pointsRead,
pQInfo->rec.pointsRead, pQInfo->pointsReturned);
dTrace("QInfo:%p current:%lldd, total:%lldd", pQInfo, pQuery->rec.pointsRead, pQInfo->rec.pointsTotal);
return;
}
#if 0
@ -5294,7 +5245,7 @@ void qTableQuery(SQInfo *pQInfo) {
if (Q_STATUS_EQUAL(pQuery->status, QUERY_NO_DATA_TO_CHECK | QUERY_COMPLETED)) {
// continue to get push data from the group result
if (isGroupbyNormalCol(pQuery->pGroupbyExpr) ||
(pQuery->intervalTime > 0 && pQInfo->pointsReturned < pQuery->limit.limit)) {
(pQuery->intervalTime > 0 && pQInfo->rec.pointsTotal < pQuery->limit.limit)) {
// todo limit the output for interval query?
pQuery->rec.pointsRead = 0;
pQInfo->subgroupIdx = 0; // always start from 0
@ -5436,11 +5387,6 @@ static int32_t validateQueryMeterMsg(SQueryTableMsg *pQueryTableMsg) {
return -1;
}
if (pQueryTableMsg->numOfTagsCols < 0 || pQueryTableMsg->numOfTagsCols > TSDB_MAX_TAGS + 1) {
dError("qmsg:%p illegal value of numOfTagsCols %d", pQueryTableMsg, pQueryTableMsg->numOfTagsCols);
return -1;
}
if (pQueryTableMsg->numOfCols <= 0 || pQueryTableMsg->numOfCols > TSDB_MAX_COLUMNS) {
dError("qmsg:%p illegal value of numOfCols %d", pQueryTableMsg, pQueryTableMsg->numOfCols);
return -1;
@ -5469,30 +5415,62 @@ static int32_t validateQueryMeterMsg(SQueryTableMsg *pQueryTableMsg) {
return 0;
}
static int32_t convertQueryTableMsg(SQueryTableMsg *pQueryTableMsg, SArray **pTableIdList) {
pQueryTableMsg->vgId = htons(pQueryTableMsg->vgId);
static char* createTableIdList(SQueryTableMsg* pQueryTableMsg, char* pMsg, SArray** pTableIdList) {
assert(pQueryTableMsg->numOfTables > 0);
*pTableIdList = taosArrayInit(pQueryTableMsg->numOfTables, sizeof(STableIdInfo));
STableIdInfo *pTableIdInfo = (STableIdInfo *)pMsg;
pTableIdInfo->sid = htonl(pTableIdInfo->sid);
pTableIdInfo->uid = htobe64(pTableIdInfo->uid);
pTableIdInfo->key = htobe64(pTableIdInfo->key);
taosArrayPush(*pTableIdList, pTableIdInfo);
pMsg += sizeof(STableIdInfo);
for (int32_t j = 1; j < pQueryTableMsg->numOfTables; ++j) {
pTableIdInfo = (STableIdInfo *)pMsg;
pTableIdInfo->sid = htonl(pTableIdInfo->sid);
pTableIdInfo->uid = htobe64(pTableIdInfo->uid);
pTableIdInfo->key = htobe64(pTableIdInfo->key);
taosArrayPush(*pTableIdList, pTableIdInfo);
pMsg += sizeof(STableIdInfo);
}
return pMsg;
}
/**
* pQueryTableMsg->head has been converted before this function is called.
*
* @param pQueryTableMsg
* @param pTableIdList
* @param pExpr
* @return
*/
static int32_t convertQueryMsg(SQueryTableMsg *pQueryTableMsg, SArray **pTableIdList, SSqlFuncExprMsg ***pExpr) {
pQueryTableMsg->numOfTables = htonl(pQueryTableMsg->numOfTables);
pQueryTableMsg->window.skey = htobe64(pQueryTableMsg->window.skey);
pQueryTableMsg->window.ekey = htobe64(pQueryTableMsg->window.ekey);
pQueryTableMsg->intervalTime = htobe64(pQueryTableMsg->intervalTime);
pQueryTableMsg->slidingTime = htobe64(pQueryTableMsg->slidingTime);
pQueryTableMsg->limit = htobe64(pQueryTableMsg->limit);
pQueryTableMsg->offset = htobe64(pQueryTableMsg->offset);
pQueryTableMsg->order = htons(pQueryTableMsg->order);
pQueryTableMsg->orderColId = htons(pQueryTableMsg->orderColId);
pQueryTableMsg->queryType = htons(pQueryTableMsg->queryType);
pQueryTableMsg->intervalTime = htobe64(pQueryTableMsg->intervalTime);
pQueryTableMsg->slidingTime = htobe64(pQueryTableMsg->slidingTime);
pQueryTableMsg->numOfTagsCols = htons(pQueryTableMsg->numOfTagsCols);
pQueryTableMsg->numOfCols = htons(pQueryTableMsg->numOfCols);
pQueryTableMsg->numOfOutputCols = htons(pQueryTableMsg->numOfOutputCols);
pQueryTableMsg->numOfGroupCols = htons(pQueryTableMsg->numOfGroupCols);
pQueryTableMsg->tagLength = htons(pQueryTableMsg->tagLength);
pQueryTableMsg->limit = htobe64(pQueryTableMsg->limit);
pQueryTableMsg->offset = htobe64(pQueryTableMsg->offset);
pQueryTableMsg->tsOffset = htonl(pQueryTableMsg->tsOffset);
pQueryTableMsg->tsLen = htonl(pQueryTableMsg->tsLen);
pQueryTableMsg->tsNumOfBlocks = htonl(pQueryTableMsg->tsNumOfBlocks);
@ -5506,23 +5484,23 @@ static int32_t convertQueryTableMsg(SQueryTableMsg *pQueryTableMsg, SArray **pTa
char *pMsg = (char *)(pQueryTableMsg->colList) + sizeof(SColumnInfo) * pQueryTableMsg->numOfCols;
for (int32_t col = 0; col < pQueryTableMsg->numOfCols; ++col) {
pQueryTableMsg->colList[col].colId = htons(pQueryTableMsg->colList[col].colId);
pQueryTableMsg->colList[col].type = htons(pQueryTableMsg->colList[col].type);
pQueryTableMsg->colList[col].bytes = htons(pQueryTableMsg->colList[col].bytes);
pQueryTableMsg->colList[col].numOfFilters = htons(pQueryTableMsg->colList[col].numOfFilters);
SColumnInfo* pColInfo = &pQueryTableMsg->colList[col];
assert(pQueryTableMsg->colList[col].type >= TSDB_DATA_TYPE_BOOL &&
pQueryTableMsg->colList[col].type <= TSDB_DATA_TYPE_NCHAR);
pColInfo->colId = htons(pColInfo->colId);
pColInfo->type = htons(pColInfo->type);
pColInfo->bytes = htons(pColInfo->bytes);
pColInfo->numOfFilters = htons(pColInfo->numOfFilters);
int32_t numOfFilters = pQueryTableMsg->colList[col].numOfFilters;
assert(pColInfo->type >= TSDB_DATA_TYPE_BOOL && pColInfo->type <= TSDB_DATA_TYPE_NCHAR);
int32_t numOfFilters = pColInfo->numOfFilters;
if (numOfFilters > 0) {
pQueryTableMsg->colList[col].filters = calloc(numOfFilters, sizeof(SColumnFilterInfo));
pColInfo->filters = calloc(numOfFilters, sizeof(SColumnFilterInfo));
}
for (int32_t f = 0; f < numOfFilters; ++f) {
SColumnFilterInfo *pFilterInfo = (SColumnFilterInfo *)pMsg;
SColumnFilterInfo *pDestFilterInfo = &pQueryTableMsg->colList[col].filters[f];
SColumnFilterInfo *pDestFilterInfo = &pColInfo->filters[f];
pDestFilterInfo->filterOnBinary = htons(pFilterInfo->filterOnBinary);
@ -5546,20 +5524,16 @@ static int32_t convertQueryTableMsg(SQueryTableMsg *pQueryTableMsg, SArray **pTa
bool hasArithmeticFunction = false;
/*
* 1. simple projection query on meters, we only record the pSqlFuncExprs[i].colIdx value
* 2. for complex queries, whole SqlExprs object is required.
*/
pQueryTableMsg->pSqlFuncExprs = (int64_t)malloc(POINTER_BYTES * pQueryTableMsg->numOfOutputCols);
*pExpr = calloc(pQueryTableMsg->numOfOutputCols, POINTER_BYTES);
SSqlFuncExprMsg *pExprMsg = (SSqlFuncExprMsg *)pMsg;
for (int32_t i = 0; i < pQueryTableMsg->numOfOutputCols; ++i) {
((SSqlFuncExprMsg **)pQueryTableMsg->pSqlFuncExprs)[i] = pExprMsg;
(*pExpr)[i] = pExprMsg;
pExprMsg->colInfo.colIdx = htons(pExprMsg->colInfo.colIdx);
pExprMsg->colInfo.colId = htons(pExprMsg->colInfo.colId);
pExprMsg->colInfo.flag = htons(pExprMsg->colInfo.flag);
pExprMsg->functionId = htons(pExprMsg->functionId);
pExprMsg->numOfParams = htons(pExprMsg->numOfParams);
@ -5600,43 +5574,19 @@ static int32_t convertQueryTableMsg(SQueryTableMsg *pQueryTableMsg, SArray **pTa
pMsg += pQueryTableMsg->colNameLen;
}
*pTableIdList = taosArrayInit(pQueryTableMsg->numOfTables, sizeof(STableIdInfo));
pMsg = createTableIdList(pQueryTableMsg, pMsg, pTableIdList);
STableIdInfo *pTableIdInfo = (STableIdInfo *)pMsg;
pTableIdInfo->sid = htonl(pTableIdInfo->sid);
pTableIdInfo->uid = htobe64(pTableIdInfo->uid);
pTableIdInfo->key = htobe64(pTableIdInfo->key);
taosArrayPush(*pTableIdList, pTableIdInfo);
pMsg += sizeof(STableIdInfo);
for (int32_t j = 1; j < pQueryTableMsg->numOfTables; ++j) {
pTableIdInfo = (STableIdInfo *)pMsg;
pTableIdInfo->sid = htonl(pTableIdInfo->sid);
pTableIdInfo->uid = htobe64(pTableIdInfo->uid);
pTableIdInfo->key = htobe64(pTableIdInfo->key);
taosArrayPush(*pTableIdList, pTableIdInfo);
pMsg += sizeof(STableIdInfo);
}
if (pQueryTableMsg->numOfGroupCols > 0 || pQueryTableMsg->numOfTagsCols > 0) { // group by tag columns
pQueryTableMsg->pTagSchema = (uint64_t)pMsg;
SSchema *pTagSchema = (SSchema *)pQueryTableMsg->pTagSchema;
pMsg += sizeof(SSchema) * pQueryTableMsg->numOfTagsCols;
if (pQueryTableMsg->numOfGroupCols > 0) {
pQueryTableMsg->groupbyTagIds = (uint64_t) & (pTagSchema[pQueryTableMsg->numOfTagsCols]);
} else {
pQueryTableMsg->groupbyTagIds = 0;
}
if (pQueryTableMsg->numOfGroupCols > 0) { // group by tag columns
// if (pQueryTableMsg->numOfGroupCols > 0) {
// pQueryTableMsg->groupbyTagIds = (uint64_t) & (pTagSchema[pQueryTableMsg->numOfTagsCols]);
// } else {
// pQueryTableMsg->groupbyTagIds = 0;
// }
pQueryTableMsg->orderByIdx = htons(pQueryTableMsg->orderByIdx);
pQueryTableMsg->orderType = htons(pQueryTableMsg->orderType);
pMsg += sizeof(SColIndexEx) * pQueryTableMsg->numOfGroupCols;
} else {
pQueryTableMsg->pTagSchema = 0;
pQueryTableMsg->groupbyTagIds = 0;
}
@ -5650,15 +5600,12 @@ static int32_t convertQueryTableMsg(SQueryTableMsg *pQueryTableMsg, SArray **pTa
}
}
dTrace("qmsg:%p query on %d meter(s), qrange:%" PRId64 "-%" PRId64
", numOfGroupbyTagCols:%d, numOfTagCols:%d, timestamp order:%d, "
"tags order:%d, tags order col:%d, numOfOutputCols:%d, numOfCols:%d, interval:%" PRId64
", fillType:%d, comptslen:%d, limit:%" PRId64
", "
"offset:%" PRId64,
dTrace("qmsg:%p query on %d meter(s), qrange:%" PRId64 "-%" PRId64 ", numOfGroupbyTagCols:%d, numOfTagCols:%d, "
"timestamp order:%d, tags order:%d, tags order col:%d, numOfOutputCols:%d, numOfCols:%d, interval:%" PRId64
", fillType:%d, comptslen:%d, limit:%" PRId64 ", offset:%" PRId64,
pQueryTableMsg, pQueryTableMsg->numOfTables, pQueryTableMsg->window.skey, pQueryTableMsg->window.ekey,
pQueryTableMsg->numOfGroupCols, pQueryTableMsg->numOfTagsCols, pQueryTableMsg->order,
pQueryTableMsg->orderType, pQueryTableMsg->orderByIdx, pQueryTableMsg->numOfOutputCols,
pQueryTableMsg->numOfGroupCols, pQueryTableMsg->order, pQueryTableMsg->orderType,
pQueryTableMsg->orderByIdx, pQueryTableMsg->numOfOutputCols,
pQueryTableMsg->numOfCols, pQueryTableMsg->intervalTime, pQueryTableMsg->interpoType, pQueryTableMsg->tsLen,
pQueryTableMsg->limit, pQueryTableMsg->offset);
@ -5716,22 +5663,20 @@ static int32_t buildAirthmeticExprFromMsg(SSqlFunctionExpr *pExpr, SQueryTableMs
return TSDB_CODE_SUCCESS;
}
static int32_t createSqlFunctionExprFromMsg(SQueryTableMsg *pQueryMsg, SSqlFunctionExpr **pSqlFuncExpr) {
static int32_t createSqlFunctionExprFromMsg(SQueryTableMsg *pQueryMsg, SSqlFunctionExpr **pSqlFuncExpr, SSqlFuncExprMsg** pExprMsg) {
*pSqlFuncExpr = NULL;
int32_t code = TSDB_CODE_SUCCESS;
SSqlFunctionExpr *pExprs = (SSqlFunctionExpr *)calloc(1, sizeof(SSqlFunctionExpr) * pQueryMsg->numOfOutputCols);
if (pExprs == NULL) {
tfree(pQueryMsg->pSqlFuncExprs);
return TSDB_CODE_SERV_OUT_OF_MEMORY;
}
bool isSuperTable = QUERY_IS_STABLE_QUERY(pQueryMsg->queryType);
int16_t tagLen = 0;
SSchema *pTagSchema = (SSchema *)pQueryMsg->pTagSchema;
for (int32_t i = 0; i < pQueryMsg->numOfOutputCols; ++i) {
pExprs[i].pBase = *((SSqlFuncExprMsg **)pQueryMsg->pSqlFuncExprs)[i];
pExprs[i].pBase = *pExprMsg[i];
pExprs[i].resBytes = 0;
int16_t type = 0;
@ -5739,18 +5684,7 @@ static int32_t createSqlFunctionExprFromMsg(SQueryTableMsg *pQueryMsg, SSqlFunct
SColIndexEx *pColumnIndexExInfo = &pExprs[i].pBase.colInfo;
// tag column schema is kept in pQueryMsg->pColumnModel
if (TSDB_COL_IS_TAG(pColumnIndexExInfo->flag)) {
if (pColumnIndexExInfo->colIdx >= pQueryMsg->numOfTagsCols) {
tfree(pExprs);
return TSDB_CODE_INVALID_QUERY_MSG;
}
type = pTagSchema[pColumnIndexExInfo->colIdx].type;
bytes = pTagSchema[pColumnIndexExInfo->colIdx].bytes;
} else { // parse the arithmetic expression
// parse the arithmetic expression
if (pExprs[i].pBase.functionId == TSDB_FUNC_ARITHM) {
code = buildAirthmeticExprFromMsg(&pExprs[i], pQueryMsg);
@ -5769,7 +5703,6 @@ static int32_t createSqlFunctionExprFromMsg(SQueryTableMsg *pQueryMsg, SSqlFunct
type = pCol->type;
bytes = pCol->bytes;
}
}
int32_t param = pExprs[i].pBase.arg[0].argValue.i64;
if (getResultDataInfo(type, bytes, pExprs[i].pBase.functionId, param, &pExprs[i].resType, &pExprs[i].resBytes,
@ -5788,7 +5721,7 @@ static int32_t createSqlFunctionExprFromMsg(SQueryTableMsg *pQueryMsg, SSqlFunct
// TODO refactor
for (int32_t i = 0; i < pQueryMsg->numOfOutputCols; ++i) {
pExprs[i].pBase = *((SSqlFuncExprMsg **)pQueryMsg->pSqlFuncExprs)[i];
pExprs[i].pBase = *pExprMsg[i];
int16_t functId = pExprs[i].pBase.functionId;
if (functId == TSDB_FUNC_TOP || functId == TSDB_FUNC_BOTTOM) {
int32_t j = getColumnIndexInSource(pQueryMsg, &pExprs[i].pBase);
@ -5805,7 +5738,7 @@ static int32_t createSqlFunctionExprFromMsg(SQueryTableMsg *pQueryMsg, SSqlFunct
}
}
tfree(pQueryMsg->pSqlFuncExprs);
tfree(pExprMsg);
*pSqlFuncExpr = pExprs;
return TSDB_CODE_SUCCESS;
@ -5957,8 +5890,9 @@ static SQInfo *createQInfoImpl(SQueryTableMsg *pQueryMsg, SSqlGroupbyExpr *pGrou
for (int16_t i = 0; i < numOfCols; ++i) {
pQuery->colList[i].info = pQueryMsg->colList[i];
// SColumnInfo *pColInfo = &pQuery->colList[i].data;
// pColInfo->filters = NULL;
SColumnInfo *pColInfo = &pQuery->colList[i].info;
pColInfo->filters = NULL;
// if (colList[i].numOfFilters > 0) {
// pColInfo->filters = calloc(1, colList[i].numOfFilters * sizeof(SColumnFilterInfo));
//
@ -5982,7 +5916,7 @@ static SQInfo *createQInfoImpl(SQueryTableMsg *pQueryMsg, SSqlGroupbyExpr *pGrou
}
// prepare the result buffer
pQuery->sdata = (SData **)calloc(pQuery->numOfOutputCols, sizeof(SData *));
pQuery->sdata = (SData **)calloc(pQuery->numOfOutputCols, POINTER_BYTES);
if (pQuery->sdata == NULL) {
goto _clean_memory;
}
@ -6137,12 +6071,10 @@ static int32_t createQInfo(SQueryTableMsg *pQueryMsg, SSqlGroupbyExpr *pGroupbyE
pQuery->window.skey = pQueryMsg->window.skey;
pQuery->window.ekey = pQueryMsg->window.ekey;
pQuery->lastKey = pQuery->window.skey;
if (sem_init(&(*pQInfo)->dataReady, 0, 0) != 0) {
// dError("QInfo:%p vid:%d sid:%d meterId:%s, init dataReady sem failed, reason:%s", pQInfo, pMeterObj->vnode,
// pMeterObj->sid, pMeterObj->meterId, strerror(errno));
dError("QInfo:%p init dataReady sem failed, reason:%s", pQInfo, strerror(errno));
code = TSDB_CODE_APP_ERROR;
goto _error;
}
@ -6158,7 +6090,7 @@ static int32_t createQInfo(SQueryTableMsg *pQueryMsg, SSqlGroupbyExpr *pGroupbyE
tsBufNextPos(pTSBuf);
}
if ((code = vnodeQueryTablePrepare(*pQInfo, pTSBuf, tsdb)) != TSDB_CODE_SUCCESS) {
if ((code = initQInfo(*pQInfo, pTSBuf, tsdb)) != TSDB_CODE_SUCCESS) {
goto _error;
}
@ -6181,8 +6113,10 @@ int32_t qCreateQueryInfo(void* tsdb, SQueryTableMsg *pQueryTableMsg, SQInfo **pQ
assert(pQueryTableMsg != NULL);
int32_t code = TSDB_CODE_SUCCESS;
SArray *pTableIdList = NULL;
if ((code = convertQueryTableMsg(pQueryTableMsg, &pTableIdList)) != TSDB_CODE_SUCCESS) {
SSqlFuncExprMsg** pExprMsg = NULL;
if ((code = convertQueryMsg(pQueryTableMsg, &pTableIdList, &pExprMsg)) != TSDB_CODE_SUCCESS) {
return code;
}
@ -6200,7 +6134,7 @@ int32_t qCreateQueryInfo(void* tsdb, SQueryTableMsg *pQueryTableMsg, SQInfo **pQ
}
SSqlFunctionExpr *pExprs = NULL;
if ((code = createSqlFunctionExprFromMsg(pQueryTableMsg, &pExprs)) != TSDB_CODE_SUCCESS) {
if ((code = createSqlFunctionExprFromMsg(pQueryTableMsg, &pExprs, pExprMsg)) != TSDB_CODE_SUCCESS) {
goto _query_over;
}
@ -6256,14 +6190,119 @@ int32_t qRetrieveQueryResultInfo(SQInfo *pQInfo, int32_t *numOfRows, int32_t *ro
SQuery *pQuery = pQInfo->runtimeEnv.pQuery;
// *numOfRows = pQInfo->rec.pointsRead;
// *rowsize = pQuery->rowSize;
*numOfRows = 1;
*numOfRows = pQInfo->rec.pointsRead;
*rowsize = pQuery->rowSize;
// dTrace("QInfo:%p, retrieve data info completed, precision:%d, rowsize:%d, rows:%d, code:%d", pQInfo, *timePrec,
// *rowsize, *numOfRows, pQInfo->code);
dTrace("QInfo:%p retrieve result info, rowsize:%d, rows:%d, code:%d", pQInfo, *rowsize, *numOfRows, pQInfo->code);
if (pQInfo->code < 0) { // less than 0 means there are error existed.
return -pQInfo->code;
}
}
static size_t getResultSize(SQInfo *pQInfo, int64_t *numOfRows) {
SQuery* pQuery = pQInfo->runtimeEnv.pQuery;
/*
* get the file size and set the numOfRows to be the file size, since for tsComp query,
* the returned row size is equalled to 1
* TODO handle the case that the file is too large to send back one time
*/
if (isTSCompQuery(pQuery) && (*numOfRows) > 0) {
struct stat fstat;
if (stat(pQuery->sdata[0]->data, &fstat) == 0) {
*numOfRows = fstat.st_size;
return fstat.st_size;
} else {
dError("QInfo:%p failed to get file info, path:%s, reason:%s", pQInfo, pQuery->sdata[0]->data, strerror(errno));
return 0;
}
} else {
return pQuery->rowSize * (*numOfRows);
}
}
static int32_t doDumpQueryResult(SQInfo *pQInfo, char *data, int32_t *size) {
// the remained number of retrieved rows, not the interpolated result
SQuery *pQuery = pQInfo->runtimeEnv.pQuery;
// load data from file to msg buffer
if (isTSCompQuery(pQuery)) {
int32_t fd = open(pQuery->sdata[0]->data, O_RDONLY, 0666);
// make sure file exist
if (FD_VALID(fd)) {
size_t s = lseek(fd, 0, SEEK_END);
dTrace("QInfo:%p ts comp data return, file:%s, size:%zu", pQInfo, pQuery->sdata[0]->data, s);
lseek(fd, 0, SEEK_SET);
read(fd, data, s);
close(fd);
unlink(pQuery->sdata[0]->data);
} else {
dError("QInfo:%p failed to open tmp file to send ts-comp data to client, path:%s, reason:%s", pQInfo,
pQuery->sdata[0]->data, strerror(errno));
}
} else {
doCopyQueryResultToMsg(pQInfo, pQInfo->rec.pointsRead, data);
}
pQInfo->rec.pointsTotal += pQInfo->rec.pointsRead;
dTrace("QInfo:%p current:%d, total:%d", pQInfo, pQInfo->rec.pointsRead, pQInfo->rec.pointsTotal);
}
static void addToTaskQueue(SQInfo* pQInfo) {
// no error occurred, continue retrieving data
if (pQInfo->code == TSDB_CODE_SUCCESS) {
#ifdef _TD_ARM_
dTrace("QInfo:%p set query flag, sig:%" PRIu64 ", func:doDumpQueryResult", pQInfo, pQInfo->signature);
#else
dTrace("QInfo:%p set query flag, sig:%" PRIu64 ", func:%s", pQInfo, pQInfo->signature, __FUNCTION__);
#endif
if (pQInfo->killed == 1) {
dTrace("%p freed or killed, abort query", pQInfo);
} else {
// todo add to task queue
}
}
}
int32_t qDumpRetrieveResult(SQInfo *pQInfo, SRetrieveTableRsp** pRsp, int32_t* contLen) {
if (pQInfo == NULL || !isQInfoValid(pQInfo)) {
return TSDB_CODE_INVALID_QHANDLE;
}
SQuery* pQuery = pQInfo->runtimeEnv.pQuery;
size_t size = getResultSize(pQInfo, &pQInfo->rec.pointsRead);
*contLen = size + sizeof(SRetrieveTableRsp);
// todo handle failed to allocate memory
*pRsp = (SRetrieveTableRsp *)rpcMallocCont(*contLen);
(*pRsp)->numOfRows = htonl(pQInfo->rec.pointsRead);
int32_t code = pQInfo->code;
if (code == TSDB_CODE_SUCCESS) {
(*pRsp)->offset = htobe64(pQuery->limit.offset);
(*pRsp)->useconds = htobe64(pQInfo->elapsedTime);
} else {
(*pRsp)->offset = 0;
(*pRsp)->useconds = 0;
}
if (pQInfo->rec.pointsRead > 0 && code == TSDB_CODE_SUCCESS) {
doDumpQueryResult(pQInfo, (*pRsp)->data, NULL);
addToTaskQueue(pQInfo);
return TSDB_CODE_SUCCESS;
}
assert(code != TSDB_CODE_ACTION_IN_PROGRESS);
// if (numOfRows == 0 && (pRetrieve->qhandle == (uint64_t)pObj->qhandle) && (code != TSDB_CODE_ACTION_IN_PROGRESS)) {
// dTrace("QInfo:%p %s free qhandle code:%d", pObj->qhandle, __FUNCTION__, code);
// vnodeDecRefCount(pObj->qhandle);
// pObj->qhandle = NULL;
// }
}

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

@ -162,6 +162,8 @@ void taosDeleteStrHash(void *handle, char *string) {
if (pObj == NULL || pObj->maxSessions == 0) return;
if (string == NULL || string[0] == 0) return;
return;
hash = (*(pObj->hashFp))(pObj, string);
pthread_mutex_lock(&pObj->mutex);

43
src/vnode/tsdb/src/tsdbRead.c
View File

@ -109,7 +109,6 @@ typedef struct STsdbQueryHandle {
uint16_t flag; // denotes reversed scan of data or not
int16_t order;
STimeWindow window; // the primary query time window that applies to all queries
TSKEY lastKey;
int32_t blockBufferSize;
SCompBlock *pBlock;
int32_t numOfBlocks;
@ -265,7 +264,19 @@ tsdb_query_handle_t *tsdbQueryByTableId(tsdb_repo_t* tsdb, STsdbQueryCond *pCond
pQueryHandle->loadDataAfterSeek = false;
pQueryHandle->isFirstSlot = true;
pQueryHandle->lastKey = pQueryHandle->window.skey; // ascending query
// only support table query
assert(taosArrayGetSize(idList) == 1);
pQueryHandle->pTableQueryInfo = calloc(1, sizeof(STableQueryRec));
STableQueryRec* pTableQRec = pQueryHandle->pTableQueryInfo;
pTableQRec->lastKey = pQueryHandle->window.skey;
STableIdInfo* idInfo = taosArrayGet(pQueryHandle->pTableIdList, 0);
STableId tableId = {.uid = idInfo->uid, .tid = idInfo->sid};
STable *pTable = tsdbIsValidTableToInsert(tsdbGetMeta(pQueryHandle->pTsdb), tableId);
pTableQRec->pTableObj = pTable;
// malloc buffer in order to load data from file
int32_t numOfCols = taosArrayGetSize(pColumnInfo);
@ -295,14 +306,21 @@ tsdb_query_handle_t *tsdbQueryByTableId(tsdb_repo_t* tsdb, STsdbQueryCond *pCond
return (tsdb_query_handle_t)pQueryHandle;
}
static int32_t next = 1;
bool tsdbNextDataBlock(tsdb_query_handle_t *pQueryHandle) {
if (next == 0) {
STsdbQueryHandle* pHandle = (STsdbQueryHandle*) pQueryHandle;
STable *pTable = pHandle->pTableQueryInfo->pTableObj;
// no data in cache, abort
if (pTable->mem == NULL && pTable->imem == NULL) {
return false;
} else {
next = 0;
return true;
}
// all data in mem are checked already.
if (pHandle->pTableQueryInfo->lastKey > pTable->mem->keyLast) {
return false;
}
return true;
}
static int tsdbReadRowsFromCache(SSkipListIterator *pIter, TSKEY maxKey, int maxRowsToRead,
@ -344,9 +362,7 @@ SDataBlockInfo tsdbRetrieveDataBlockInfo(tsdb_query_handle_t *pQueryHandle) {
STsdbQueryHandle* pHandle = (STsdbQueryHandle*) pQueryHandle;
STableIdInfo* idInfo = taosArrayGet(pHandle->pTableIdList, 0);
STableId tableId = {.uid = idInfo->uid, .tid = idInfo->sid};
STable *pTable = tsdbIsValidTableToInsert(tsdbGetMeta(pHandle->pTsdb), tableId);
assert(pTable != NULL);
STable *pTable = pHandle->pTableQueryInfo->pTableObj;
TSKEY skey = 0, ekey = 0;
int32_t rows = 0;
@ -357,12 +373,15 @@ SDataBlockInfo tsdbRetrieveDataBlockInfo(tsdb_query_handle_t *pQueryHandle) {
}
SDataBlockInfo blockInfo = {
.uid = tableId.uid,
.sid = tableId.tid,
.uid = idInfo->uid,
.sid = idInfo->sid,
.size = rows,
.window = {.skey = skey, .ekey = ekey}
};
// update the last key value
pHandle->pTableQueryInfo->lastKey = ekey + 1;
return blockInfo;
}