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

[td-32] fix bugs in returning data to client

This commit is contained in:
hjxilinx 2020-03-24 01:01:55 +08:00
parent 09962c0039
commit e49bd8a723
9 changed files with 497 additions and 491 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; numOfElem += 1;
} }
} else { } else {
/*
* when counting on the primary time stamp column and no statistics data is provided,
* simple use the size value
*/
numOfElem = pCtx->size; numOfElem = pCtx->size;
} }
} }

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

@ -662,12 +662,16 @@ int tscBuildQueryMsg(SSqlObj *pSql, SSqlInfo *pInfo) {
SQueryInfo *pQueryInfo = tscGetQueryInfoDetail(pCmd, pCmd->clauseIndex); SQueryInfo *pQueryInfo = tscGetQueryInfoDetail(pCmd, pCmd->clauseIndex);
STableMetaInfo *pTableMetaInfo = tscGetMetaInfo(pQueryInfo, 0); STableMetaInfo *pTableMetaInfo = tscGetMetaInfo(pQueryInfo, 0);
char *pStart = pCmd->payload + tsRpcHeadSize;
STableMeta * pTableMeta = pTableMetaInfo->pTableMeta; STableMeta * pTableMeta = pTableMetaInfo->pTableMeta;
SSuperTableMeta *pMetricMeta = pTableMetaInfo->pMetricMeta; 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; SQueryTableMsg *pQueryMsg = (SQueryTableMsg *)pStart;
int32_t msgLen = 0; int32_t msgLen = 0;
@ -675,11 +679,7 @@ int tscBuildQueryMsg(SSqlObj *pSql, SSqlInfo *pInfo) {
if (UTIL_TABLE_IS_NOMRAL_TABLE(pTableMetaInfo)) { if (UTIL_TABLE_IS_NOMRAL_TABLE(pTableMetaInfo)) {
numOfTables = 1; numOfTables = 1;
pQueryMsg->head.vgId = htonl(pTableMeta->vgId);
pQueryMsg->uid = pTableMeta->uid;
pQueryMsg->numOfTagsCols = 0;
pQueryMsg->vgId = htonl(pTableMeta->vgId);
tscTrace("%p queried tables:%d, table id: %s", pSql, 1, pTableMetaInfo->name); tscTrace("%p queried tables:%d, table id: %s", pSql, 1, pTableMetaInfo->name);
} else { // query on super table } else { // query on super table
if (pTableMetaInfo->vnodeIndex < 0) { 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); 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->numOfTables = htonl(numOfTables);
pQueryMsg->numOfTagsCols = htons(pTableMetaInfo->numOfTags);
if (pQueryInfo->order.order == TSQL_SO_ASC) { if (pQueryInfo->order.order == TSQL_SO_ASC) {
pQueryMsg->window.skey = htobe64(pQueryInfo->stime); pQueryMsg->window.skey = htobe64(pQueryInfo->stime);
@ -713,22 +712,13 @@ int tscBuildQueryMsg(SSqlObj *pSql, SSqlInfo *pInfo) {
pQueryMsg->order = htons(pQueryInfo->order.order); pQueryMsg->order = htons(pQueryInfo->order.order);
pQueryMsg->orderColId = htons(pQueryInfo->order.orderColId); pQueryMsg->orderColId = htons(pQueryInfo->order.orderColId);
pQueryMsg->interpoType = htons(pQueryInfo->interpoType); pQueryMsg->interpoType = htons(pQueryInfo->interpoType);
pQueryMsg->limit = htobe64(pQueryInfo->limit.limit); pQueryMsg->limit = htobe64(pQueryInfo->limit.limit);
pQueryMsg->offset = htobe64(pQueryInfo->limit.offset); pQueryMsg->offset = htobe64(pQueryInfo->limit.offset);
pQueryMsg->numOfCols = htons(pQueryInfo->colList.numOfCols); 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->intervalTime = htobe64(pQueryInfo->intervalTime);
pQueryMsg->slidingTimeUnit = pQueryInfo->slidingTimeUnit;
pQueryMsg->slidingTime = htobe64(pQueryInfo->slidingTime); pQueryMsg->slidingTime = htobe64(pQueryInfo->slidingTime);
pQueryMsg->slidingTimeUnit = pQueryInfo->slidingTimeUnit;
if (pQueryInfo->intervalTime < 0) { if (pQueryInfo->intervalTime < 0) {
tscError("%p illegal value of aggregation time interval in query msg: %ld", pSql, pQueryInfo->intervalTime); 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); pQueryMsg->colNameLen = htonl(len);
// serialize the table info (sid, uid, tags) // serialize the table info (sid, uid, tags)
pMsg = doSerializeTableInfo(pSql, numOfTables, htons(pQueryMsg->vgId), pMsg); pMsg = doSerializeTableInfo(pSql, numOfTables, htons(pQueryMsg->head.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);
}
}
SSqlGroupbyExpr *pGroupbyExpr = &pQueryInfo->groupbyExpr; SSqlGroupbyExpr *pGroupbyExpr = &pQueryInfo->groupbyExpr;
if (pGroupbyExpr->numOfGroupCols != 0) { if (pGroupbyExpr->numOfGroupCols != 0) {
@ -948,8 +920,7 @@ int tscBuildQueryMsg(SSqlObj *pSql, SSqlInfo *pInfo) {
pCmd->payloadLen = msgLen; pCmd->payloadLen = msgLen;
pSql->cmd.msgType = TSDB_MSG_TYPE_QUERY; pSql->cmd.msgType = TSDB_MSG_TYPE_QUERY;
pQueryMsg->contLen = htonl(msgLen); pQueryMsg->head.contLen = htonl(msgLen);
assert(msgLen + minMsgSize() <= size); assert(msgLen + minMsgSize() <= size);
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
@ -2358,7 +2329,6 @@ int tscProcessRetrieveRspFromVnode(SSqlObj *pSql) {
} }
pRes->row = 0; pRes->row = 0;
tscTrace("%p numOfRows:%d, offset:%d", pSql, pRes->numOfRows, pRes->offset); tscTrace("%p numOfRows:%d, offset:%d", pSql, pRes->numOfRows, pRes->offset);
return 0; return 0;

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

@ -284,16 +284,10 @@ static void dnodeProcessRetrieveMsg(SReadMsg *pMsg) {
//todo free qinfo //todo free qinfo
} else { } else {
contLen = 100; SRetrieveTableRsp* pRsp = NULL;
SRetrieveTableRsp *pRsp = (SRetrieveTableRsp *)rpcMallocCont(contLen); int32_t code = qDumpRetrieveResult(pQInfo, &pRsp, &contLen);
pRsp->numOfRows = htonl(1); //todo check code
pRsp->precision = htons(0);
pRsp->offset = htobe64(0);
pRsp->useconds = htobe64(0);
// todo set the data
*(int64_t*) pRsp->data = 1000;
rpcRsp = (SRpcMsg) { rpcRsp = (SRpcMsg) {
.handle = pMsg->rpcMsg.handle, .handle = pMsg->rpcMsg.handle,

27
src/inc/taosmsg.h
View File

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

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

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

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

@ -219,7 +219,7 @@ typedef struct SQLAggFuncElem {
void (*distSecondaryMergeFunc)(SQLFunctionCtx *pCtx); 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; } SQLAggFuncElem;
typedef struct SPatternCompareInfo { typedef struct SPatternCompareInfo {

342
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 bool functionNeedToExecute(SQueryRuntimeEnv *pRuntimeEnv, SQLFunctionCtx *pCtx, int32_t functionId);
static void getNextTimeWindow(SQuery *pQuery, STimeWindow *pTimeWindow); 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); int32_t functionId, SDataStatis *pStatis, bool hasNull, void *param, int32_t scanFlag);
static void initCtxOutputBuf(SQueryRuntimeEnv *pRuntimeEnv); static void initCtxOutputBuf(SQueryRuntimeEnv *pRuntimeEnv);
static void destroyMeterQueryInfo(STableQueryInfo *pTableQueryInfo, int32_t numOfCols); static void destroyMeterQueryInfo(STableQueryInfo *pTableQueryInfo, int32_t numOfCols);
@ -427,15 +427,9 @@ static bool isTopBottomQuery(SQuery *pQuery) {
return false; return false;
} }
static SDataStatis *getStatisInfo(SQuery *pQuery, SDataStatis *pStatis, SDataBlockInfo *pDataBlockInfo, static SDataStatis *getStatisInfo(SQuery *pQuery, SDataStatis *pStatis, SDataBlockInfo *pDataBlockInfo, int32_t index) {
int32_t columnIndex) {
// no SField info exist, or column index larger than the output column, no result.
if (pStatis == NULL) {
return NULL;
}
// for a tag column, no corresponding field info // 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)) { if (TSDB_COL_IS_TAG(pColIndexEx->flag)) {
return NULL; return NULL;
} }
@ -453,13 +447,31 @@ static SDataStatis *getStatisInfo(SQuery *pQuery, SDataStatis *pStatis, SDataBlo
return NULL; 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, static bool hasNullValue(SQuery *pQuery, int32_t col, SDataBlockInfo *pDataBlockInfo, SDataStatis *pStatis,
SDataStatis **pColStatis) { 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; 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); *pColStatis = getStatisInfo(pQuery, pStatis, pDataBlockInfo, col);
}
if ((*pColStatis) != NULL && (*pColStatis)->numOfNull == 0) { if ((*pColStatis) != NULL && (*pColStatis)->numOfNull == 0) {
return false; 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. * 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. * So, the validation of required column in cache with the corresponding meter schema is reinforced.
*/ */
if (pDataBlock == NULL) { if (pDataBlock == NULL) {
return NULL; return NULL;
} }
@ -939,10 +950,11 @@ static int32_t blockwiseApplyAllFunctions(SQueryRuntimeEnv *pRuntimeEnv, SDataSt
int32_t functionId = pQuery->pSelectExpr[k].pBase.functionId; int32_t functionId = pQuery->pSelectExpr[k].pBase.functionId;
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->size, pDataBlock); 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); hasNull, &sasArray[k], pRuntimeEnv->scanFlag);
} }
@ -1381,7 +1393,7 @@ static int32_t tableApplyFunctionsOnBlock(SQueryRuntimeEnv *pRuntimeEnv, SDataBl
return 0; 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) { int32_t functionId, SDataStatis *pStatis, bool hasNull, void *param, int32_t scanFlag) {
pCtx->scanFlag = scanFlag; pCtx->scanFlag = scanFlag;
@ -1396,15 +1408,15 @@ void setExecParams(SQuery *pQuery, SQLFunctionCtx *pCtx, void *inputData, char *
pCtx->preAggVals.isSet = false; pCtx->preAggVals.isSet = false;
} }
if ((aAggs[functionId].nStatus & TSDB_FUNCSTATE_SELECTIVITY) != 0 && (primaryColumnData != NULL)) { if ((aAggs[functionId].nStatus & TSDB_FUNCSTATE_SELECTIVITY) != 0 && (tsCol != NULL)) {
pCtx->ptsList = (int64_t *)(primaryColumnData); pCtx->ptsList = tsCol;
} }
if (functionId >= TSDB_FUNC_FIRST_DST && functionId <= TSDB_FUNC_LAST_DST) { if (functionId >= TSDB_FUNC_FIRST_DST && functionId <= TSDB_FUNC_LAST_DST) {
// last_dist or first_dist function // last_dist or first_dist function
// store the first&last timestamp into the intermediate buffer [1], the true // store the first&last timestamp into the intermediate buffer [1], the true
// value may be null but timestamp will never be null // 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 || } 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)) { 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; pTWAInfo->EKey = pQuery->window.ekey;
} }
pCtx->ptsList = (int64_t *)(primaryColumnData); pCtx->ptsList = tsCol;
} else if (functionId == TSDB_FUNC_ARITHM) { } else if (functionId == TSDB_FUNC_ARITHM) {
pCtx->param[1].pz = param; pCtx->param[1].pz = param;
@ -2525,21 +2537,17 @@ static void getNextTimeWindow(SQuery *pQuery, STimeWindow *pTimeWindow) {
SArray *loadDataBlockOnDemand(SQueryRuntimeEnv *pRuntimeEnv, SDataBlockInfo *pBlockInfo, SDataStatis **pStatis) { SArray *loadDataBlockOnDemand(SQueryRuntimeEnv *pRuntimeEnv, SDataBlockInfo *pBlockInfo, SDataStatis **pStatis) {
SQuery *pQuery = pRuntimeEnv->pQuery; SQuery *pQuery = pRuntimeEnv->pQuery;
tsdb_query_handle_t pQueryHandle = pRuntimeEnv->pQueryHandle;
uint32_t r = 0; uint32_t r = 0;
SArray * pDataBlock = NULL; SArray * pDataBlock = NULL;
// STimeWindow *w = &pQueryHandle->window;
if (pQuery->numOfFilterCols > 0) { if (pQuery->numOfFilterCols > 0) {
r = BLK_DATA_ALL_NEEDED; r = BLK_DATA_ALL_NEEDED;
} else { } else {
for (int32_t i = 0; i < pQuery->numOfOutputCols; ++i) { for (int32_t i = 0; i < pQuery->numOfOutputCols; ++i) {
int32_t functionId = pQuery->pSelectExpr[i].pBase.functionId; int32_t functionId = pQuery->pSelectExpr[i].pBase.functionId;
int32_t colId = pQuery->pSelectExpr[i].pBase.colInfo.colId; int32_t colId = pQuery->pSelectExpr[i].pBase.colInfo.colId;
r |= aAggs[functionId].dataReqFunc(&pRuntimeEnv->pCtx[i], pQuery->window.skey, pQuery->window.ekey, colId);
// r |= aAggs[functionId].dataReqFunc(&pRuntimeEnv->pCtx[i], w->skey, w->ekey, colId);
} }
if (pRuntimeEnv->pTSBuf > 0 || isIntervalQuery(pQuery)) { if (pRuntimeEnv->pTSBuf > 0 || isIntervalQuery(pQuery)) {
@ -3036,7 +3044,7 @@ void copyResToQueryResultBuf(SQInfo *pQInfo, SQuery *pQuery) {
total += pData->numOfElems; total += pData->numOfElems;
} }
pQuery->sdata[0]->num = total; int32_t rows = total;
int32_t offset = 0; int32_t offset = 0;
for (int32_t num = 0; num < list.size; ++num) { for (int32_t num = 0; num < list.size; ++num) {
@ -3044,7 +3052,7 @@ void copyResToQueryResultBuf(SQInfo *pQInfo, SQuery *pQuery) {
for (int32_t i = 0; i < pQuery->numOfOutputCols; ++i) { for (int32_t i = 0; i < pQuery->numOfOutputCols; ++i) {
int32_t bytes = pRuntimeEnv->pCtx[i].outputBytes; 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, memcpy(pDest + offset * bytes, pData->data + pRuntimeEnv->offset[i] * pData->numOfElems,
bytes * pData->numOfElems); bytes * pData->numOfElems);
@ -3055,7 +3063,7 @@ void copyResToQueryResultBuf(SQInfo *pQInfo, SQuery *pQuery) {
assert(pQuery->rec.pointsRead == 0); assert(pQuery->rec.pointsRead == 0);
pQuery->rec.pointsRead += pQuery->sdata[0]->num; pQuery->rec.pointsRead += rows;
pQInfo->offset += 1; pQInfo->offset += 1;
} }
@ -3222,7 +3230,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. // 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 pageId = -1;
int32_t remain = pQuery->sdata[0]->num; assert(0);
int32_t remain = 0;//pQuery->sdata[0]->num;
int32_t offset = 0; int32_t offset = 0;
while (remain > 0) { while (remain > 0) {
@ -3253,12 +3262,12 @@ int32_t flushFromResultBuf(SQInfo *pQInfo) {
void resetMergeResultBuf(SQuery *pQuery, SQLFunctionCtx *pCtx, SResultInfo *pResultInfo) { void resetMergeResultBuf(SQuery *pQuery, SQLFunctionCtx *pCtx, SResultInfo *pResultInfo) {
for (int32_t k = 0; k < pQuery->numOfOutputCols; ++k) { 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].size = 1;
pCtx[k].startOffset = 0; pCtx[k].startOffset = 0;
pCtx[k].resultInfo = &pResultInfo[k]; pCtx[k].resultInfo = &pResultInfo[k];
pQuery->sdata[k]->num = 0; pQuery->sdata[k] = 0;
} }
} }
@ -4080,11 +4089,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) { bool vnodeHasRemainResults(void *handle) {
SQInfo *pQInfo = (SQInfo *)handle; SQInfo *pQInfo = (SQInfo *)handle;
@ -4162,62 +4166,13 @@ static int32_t resultInterpolate(SQInfo *pQInfo, tFilePage **data, tFilePage **p
} }
static void doCopyQueryResultToMsg(SQInfo *pQInfo, int32_t numOfRows, char *data) { static void doCopyQueryResultToMsg(SQInfo *pQInfo, int32_t numOfRows, char *data) {
#if 0 SQuery *pQuery = pQInfo->runtimeEnv.pQuery;
SMeterObj *pObj = pQInfo->pObj; for (int32_t col = 0; col < pQuery->numOfOutputCols; ++col) {
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
int32_t bytes = pQuery->pSelectExpr[col].resBytes; int32_t bytes = pQuery->pSelectExpr[col].resBytes;
memmove(data, pQuery->sdata[col]->data, bytes * numOfRows); memmove(data, pQuery->sdata[col]->data, bytes * numOfRows);
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, int32_t vnodeQueryResultInterpolate(SQInfo *pQInfo, tFilePage **pDst, tFilePage **pDataSrc, int32_t numOfRows,
@ -4308,7 +4263,7 @@ void vnodePrintQueryStatistics(SQInfo *pQInfo) {
#endif #endif
} }
int32_t vnodeQueryTablePrepare(SQInfo *pQInfo, void *param, void* tsdb) { int32_t initQInfo(SQInfo *pQInfo, void *param, void* tsdb) {
SQuery *pQuery = pQInfo->runtimeEnv.pQuery; SQuery *pQuery = pQInfo->runtimeEnv.pQuery;
int32_t code = TSDB_CODE_SUCCESS; int32_t code = TSDB_CODE_SUCCESS;
@ -4342,7 +4297,6 @@ int32_t vnodeQueryTablePrepare(SQInfo *pQInfo, void *param, void* tsdb) {
taosArrayPush(cols, &pQuery->colList[i]); taosArrayPush(cols, &pQuery->colList[i]);
} }
pQInfo->runtimeEnv.pQueryHandle = tsdbQueryByTableId(tsdb, &cond, pQInfo->pTableIdList, cols); pQInfo->runtimeEnv.pQueryHandle = tsdbQueryByTableId(tsdb, &cond, pQInfo->pTableIdList, cols);
SQueryRuntimeEnv *pRuntimeEnv = &pQInfo->runtimeEnv; SQueryRuntimeEnv *pRuntimeEnv = &pQInfo->runtimeEnv;
@ -4993,8 +4947,7 @@ static void vnodeMultiMeterQueryProcessor(SQInfo *pQInfo) {
// vnodePrintQueryStatistics(pSupporter); // vnodePrintQueryStatistics(pSupporter);
} }
dTrace("QInfo:%p points returned:%d, totalRead:%d totalReturn:%d", pQInfo, pQuery->rec.pointsRead, dTrace("QInfo:%p current:%lldd, total:%lldd", pQInfo, pQuery->rec.pointsRead, pQInfo->rec.pointsTotal);
pQInfo->rec.pointsRead, pQInfo->pointsReturned);
return; return;
} }
#if 0 #if 0
@ -5294,7 +5247,7 @@ void qTableQuery(SQInfo *pQInfo) {
if (Q_STATUS_EQUAL(pQuery->status, QUERY_NO_DATA_TO_CHECK | QUERY_COMPLETED)) { if (Q_STATUS_EQUAL(pQuery->status, QUERY_NO_DATA_TO_CHECK | QUERY_COMPLETED)) {
// continue to get push data from the group result // continue to get push data from the group result
if (isGroupbyNormalCol(pQuery->pGroupbyExpr) || 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? // todo limit the output for interval query?
pQuery->rec.pointsRead = 0; pQuery->rec.pointsRead = 0;
pQInfo->subgroupIdx = 0; // always start from 0 pQInfo->subgroupIdx = 0; // always start from 0
@ -5436,11 +5389,6 @@ static int32_t validateQueryMeterMsg(SQueryTableMsg *pQueryTableMsg) {
return -1; 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) { if (pQueryTableMsg->numOfCols <= 0 || pQueryTableMsg->numOfCols > TSDB_MAX_COLUMNS) {
dError("qmsg:%p illegal value of numOfCols %d", pQueryTableMsg, pQueryTableMsg->numOfCols); dError("qmsg:%p illegal value of numOfCols %d", pQueryTableMsg, pQueryTableMsg->numOfCols);
return -1; return -1;
@ -5496,8 +5444,15 @@ static char* createTableIdList(SQueryTableMsg* pQueryTableMsg, char* pMsg, SArra
return pMsg; return pMsg;
} }
static int32_t convertQueryMsg(SQueryTableMsg *pQueryTableMsg, SArray **pTableIdList) { /**
pQueryTableMsg->vgId = htons(pQueryTableMsg->vgId); * 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->numOfTables = htonl(pQueryTableMsg->numOfTables);
pQueryTableMsg->window.skey = htobe64(pQueryTableMsg->window.skey); pQueryTableMsg->window.skey = htobe64(pQueryTableMsg->window.skey);
@ -5513,7 +5468,6 @@ static int32_t convertQueryMsg(SQueryTableMsg *pQueryTableMsg, SArray **pTableId
pQueryTableMsg->queryType = htons(pQueryTableMsg->queryType); pQueryTableMsg->queryType = htons(pQueryTableMsg->queryType);
pQueryTableMsg->numOfTagsCols = htons(pQueryTableMsg->numOfTagsCols);
pQueryTableMsg->numOfCols = htons(pQueryTableMsg->numOfCols); pQueryTableMsg->numOfCols = htons(pQueryTableMsg->numOfCols);
pQueryTableMsg->numOfOutputCols = htons(pQueryTableMsg->numOfOutputCols); pQueryTableMsg->numOfOutputCols = htons(pQueryTableMsg->numOfOutputCols);
pQueryTableMsg->numOfGroupCols = htons(pQueryTableMsg->numOfGroupCols); pQueryTableMsg->numOfGroupCols = htons(pQueryTableMsg->numOfGroupCols);
@ -5572,20 +5526,16 @@ static int32_t convertQueryMsg(SQueryTableMsg *pQueryTableMsg, SArray **pTableId
bool hasArithmeticFunction = false; bool hasArithmeticFunction = false;
/* *pExpr = calloc(pQueryTableMsg->numOfOutputCols, POINTER_BYTES);
* 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);
SSqlFuncExprMsg *pExprMsg = (SSqlFuncExprMsg *)pMsg; SSqlFuncExprMsg *pExprMsg = (SSqlFuncExprMsg *)pMsg;
for (int32_t i = 0; i < pQueryTableMsg->numOfOutputCols; ++i) { 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.colIdx = htons(pExprMsg->colInfo.colIdx);
pExprMsg->colInfo.colId = htons(pExprMsg->colInfo.colId); pExprMsg->colInfo.colId = htons(pExprMsg->colInfo.colId);
pExprMsg->colInfo.flag = htons(pExprMsg->colInfo.flag); pExprMsg->colInfo.flag = htons(pExprMsg->colInfo.flag);
pExprMsg->functionId = htons(pExprMsg->functionId); pExprMsg->functionId = htons(pExprMsg->functionId);
pExprMsg->numOfParams = htons(pExprMsg->numOfParams); pExprMsg->numOfParams = htons(pExprMsg->numOfParams);
@ -5628,22 +5578,17 @@ static int32_t convertQueryMsg(SQueryTableMsg *pQueryTableMsg, SArray **pTableId
pMsg = createTableIdList(pQueryTableMsg, pMsg, pTableIdList); pMsg = createTableIdList(pQueryTableMsg, pMsg, pTableIdList);
if (pQueryTableMsg->numOfGroupCols > 0 || pQueryTableMsg->numOfTagsCols > 0) { // group by tag columns if (pQueryTableMsg->numOfGroupCols > 0) { // group by tag columns
pQueryTableMsg->pTagSchema = (uint64_t)pMsg; // if (pQueryTableMsg->numOfGroupCols > 0) {
SSchema *pTagSchema = (SSchema *)pQueryTableMsg->pTagSchema; // pQueryTableMsg->groupbyTagIds = (uint64_t) & (pTagSchema[pQueryTableMsg->numOfTagsCols]);
pMsg += sizeof(SSchema) * pQueryTableMsg->numOfTagsCols; // } else {
// pQueryTableMsg->groupbyTagIds = 0;
if (pQueryTableMsg->numOfGroupCols > 0) { // }
pQueryTableMsg->groupbyTagIds = (uint64_t) & (pTagSchema[pQueryTableMsg->numOfTagsCols]);
} else {
pQueryTableMsg->groupbyTagIds = 0;
}
pQueryTableMsg->orderByIdx = htons(pQueryTableMsg->orderByIdx); pQueryTableMsg->orderByIdx = htons(pQueryTableMsg->orderByIdx);
pQueryTableMsg->orderType = htons(pQueryTableMsg->orderType); pQueryTableMsg->orderType = htons(pQueryTableMsg->orderType);
pMsg += sizeof(SColIndexEx) * pQueryTableMsg->numOfGroupCols; pMsg += sizeof(SColIndexEx) * pQueryTableMsg->numOfGroupCols;
} else { } else {
pQueryTableMsg->pTagSchema = 0;
pQueryTableMsg->groupbyTagIds = 0; pQueryTableMsg->groupbyTagIds = 0;
} }
@ -5657,13 +5602,12 @@ static int32_t convertQueryMsg(SQueryTableMsg *pQueryTableMsg, SArray **pTableId
} }
} }
dTrace("qmsg:%p query on %d meter(s), qrange:%" PRId64 "-%" PRId64 dTrace("qmsg:%p query on %d meter(s), qrange:%" PRId64 "-%" PRId64 ", numOfGroupbyTagCols:%d, numOfTagCols:%d, "
", numOfGroupbyTagCols:%d, numOfTagCols:%d, timestamp order:%d, " "timestamp order:%d, tags order:%d, tags order col:%d, numOfOutputCols:%d, numOfCols:%d, interval:%" PRId64
"tags order:%d, tags order col:%d, numOfOutputCols:%d, numOfCols:%d, interval:%" PRId64
", fillType:%d, comptslen:%d, limit:%" PRId64 ", offset:%" PRId64, ", fillType:%d, comptslen:%d, limit:%" PRId64 ", offset:%" PRId64,
pQueryTableMsg, pQueryTableMsg->numOfTables, pQueryTableMsg->window.skey, pQueryTableMsg->window.ekey, pQueryTableMsg, pQueryTableMsg->numOfTables, pQueryTableMsg->window.skey, pQueryTableMsg->window.ekey,
pQueryTableMsg->numOfGroupCols, pQueryTableMsg->numOfTagsCols, pQueryTableMsg->order, pQueryTableMsg->numOfGroupCols, pQueryTableMsg->order, pQueryTableMsg->orderType,
pQueryTableMsg->orderType, pQueryTableMsg->orderByIdx, pQueryTableMsg->numOfOutputCols, pQueryTableMsg->orderByIdx, pQueryTableMsg->numOfOutputCols,
pQueryTableMsg->numOfCols, pQueryTableMsg->intervalTime, pQueryTableMsg->interpoType, pQueryTableMsg->tsLen, pQueryTableMsg->numOfCols, pQueryTableMsg->intervalTime, pQueryTableMsg->interpoType, pQueryTableMsg->tsLen,
pQueryTableMsg->limit, pQueryTableMsg->offset); pQueryTableMsg->limit, pQueryTableMsg->offset);
@ -5721,22 +5665,20 @@ static int32_t buildAirthmeticExprFromMsg(SSqlFunctionExpr *pExpr, SQueryTableMs
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }
static int32_t createSqlFunctionExprFromMsg(SQueryTableMsg *pQueryMsg, SSqlFunctionExpr **pSqlFuncExpr) { static int32_t createSqlFunctionExprFromMsg(SQueryTableMsg *pQueryMsg, SSqlFunctionExpr **pSqlFuncExpr, SSqlFuncExprMsg** pExprMsg) {
*pSqlFuncExpr = NULL; *pSqlFuncExpr = NULL;
int32_t code = TSDB_CODE_SUCCESS; int32_t code = TSDB_CODE_SUCCESS;
SSqlFunctionExpr *pExprs = (SSqlFunctionExpr *)calloc(1, sizeof(SSqlFunctionExpr) * pQueryMsg->numOfOutputCols); SSqlFunctionExpr *pExprs = (SSqlFunctionExpr *)calloc(1, sizeof(SSqlFunctionExpr) * pQueryMsg->numOfOutputCols);
if (pExprs == NULL) { if (pExprs == NULL) {
tfree(pQueryMsg->pSqlFuncExprs);
return TSDB_CODE_SERV_OUT_OF_MEMORY; return TSDB_CODE_SERV_OUT_OF_MEMORY;
} }
bool isSuperTable = QUERY_IS_STABLE_QUERY(pQueryMsg->queryType); bool isSuperTable = QUERY_IS_STABLE_QUERY(pQueryMsg->queryType);
int16_t tagLen = 0; int16_t tagLen = 0;
SSchema *pTagSchema = (SSchema *)pQueryMsg->pTagSchema;
for (int32_t i = 0; i < pQueryMsg->numOfOutputCols; ++i) { 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; pExprs[i].resBytes = 0;
int16_t type = 0; int16_t type = 0;
@ -5744,18 +5686,7 @@ static int32_t createSqlFunctionExprFromMsg(SQueryTableMsg *pQueryMsg, SSqlFunct
SColIndexEx *pColumnIndexExInfo = &pExprs[i].pBase.colInfo; SColIndexEx *pColumnIndexExInfo = &pExprs[i].pBase.colInfo;
// tag column schema is kept in pQueryMsg->pColumnModel // parse the arithmetic expression
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
if (pExprs[i].pBase.functionId == TSDB_FUNC_ARITHM) { if (pExprs[i].pBase.functionId == TSDB_FUNC_ARITHM) {
code = buildAirthmeticExprFromMsg(&pExprs[i], pQueryMsg); code = buildAirthmeticExprFromMsg(&pExprs[i], pQueryMsg);
@ -5774,7 +5705,6 @@ static int32_t createSqlFunctionExprFromMsg(SQueryTableMsg *pQueryMsg, SSqlFunct
type = pCol->type; type = pCol->type;
bytes = pCol->bytes; bytes = pCol->bytes;
} }
}
int32_t param = pExprs[i].pBase.arg[0].argValue.i64; 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, if (getResultDataInfo(type, bytes, pExprs[i].pBase.functionId, param, &pExprs[i].resType, &pExprs[i].resBytes,
@ -5793,7 +5723,7 @@ static int32_t createSqlFunctionExprFromMsg(SQueryTableMsg *pQueryMsg, SSqlFunct
// TODO refactor // TODO refactor
for (int32_t i = 0; i < pQueryMsg->numOfOutputCols; ++i) { 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; int16_t functId = pExprs[i].pBase.functionId;
if (functId == TSDB_FUNC_TOP || functId == TSDB_FUNC_BOTTOM) { if (functId == TSDB_FUNC_TOP || functId == TSDB_FUNC_BOTTOM) {
int32_t j = getColumnIndexInSource(pQueryMsg, &pExprs[i].pBase); int32_t j = getColumnIndexInSource(pQueryMsg, &pExprs[i].pBase);
@ -5810,7 +5740,7 @@ static int32_t createSqlFunctionExprFromMsg(SQueryTableMsg *pQueryMsg, SSqlFunct
} }
} }
tfree(pQueryMsg->pSqlFuncExprs); tfree(pExprMsg);
*pSqlFuncExpr = pExprs; *pSqlFuncExpr = pExprs;
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
@ -5962,8 +5892,9 @@ static SQInfo *createQInfoImpl(SQueryTableMsg *pQueryMsg, SSqlGroupbyExpr *pGrou
for (int16_t i = 0; i < numOfCols; ++i) { for (int16_t i = 0; i < numOfCols; ++i) {
pQuery->colList[i].info = pQueryMsg->colList[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) { // if (colList[i].numOfFilters > 0) {
// pColInfo->filters = calloc(1, colList[i].numOfFilters * sizeof(SColumnFilterInfo)); // pColInfo->filters = calloc(1, colList[i].numOfFilters * sizeof(SColumnFilterInfo));
// //
@ -5987,7 +5918,7 @@ static SQInfo *createQInfoImpl(SQueryTableMsg *pQueryMsg, SSqlGroupbyExpr *pGrou
} }
// prepare the result buffer // prepare the result buffer
pQuery->sdata = (SData **)calloc(pQuery->numOfOutputCols, sizeof(SData *)); pQuery->sdata = (SData **)calloc(pQuery->numOfOutputCols, POINTER_BYTES);
if (pQuery->sdata == NULL) { if (pQuery->sdata == NULL) {
goto _clean_memory; goto _clean_memory;
} }
@ -6142,12 +6073,10 @@ static int32_t createQInfo(SQueryTableMsg *pQueryMsg, SSqlGroupbyExpr *pGroupbyE
pQuery->window.skey = pQueryMsg->window.skey; pQuery->window.skey = pQueryMsg->window.skey;
pQuery->window.ekey = pQueryMsg->window.ekey; pQuery->window.ekey = pQueryMsg->window.ekey;
pQuery->lastKey = pQuery->window.skey; pQuery->lastKey = pQuery->window.skey;
if (sem_init(&(*pQInfo)->dataReady, 0, 0) != 0) { 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, dError("QInfo:%p init dataReady sem failed, reason:%s", pQInfo, strerror(errno));
// pMeterObj->sid, pMeterObj->meterId, strerror(errno));
code = TSDB_CODE_APP_ERROR; code = TSDB_CODE_APP_ERROR;
goto _error; goto _error;
} }
@ -6163,7 +6092,7 @@ static int32_t createQInfo(SQueryTableMsg *pQueryMsg, SSqlGroupbyExpr *pGroupbyE
tsBufNextPos(pTSBuf); tsBufNextPos(pTSBuf);
} }
if ((code = vnodeQueryTablePrepare(*pQInfo, pTSBuf, tsdb)) != TSDB_CODE_SUCCESS) { if ((code = initQInfo(*pQInfo, pTSBuf, tsdb)) != TSDB_CODE_SUCCESS) {
goto _error; goto _error;
} }
@ -6188,7 +6117,8 @@ int32_t qCreateQueryInfo(void* tsdb, SQueryTableMsg *pQueryTableMsg, SQInfo **pQ
int32_t code = TSDB_CODE_SUCCESS; int32_t code = TSDB_CODE_SUCCESS;
SArray *pTableIdList = NULL; SArray *pTableIdList = NULL;
if ((code = convertQueryMsg(pQueryTableMsg, &pTableIdList)) != TSDB_CODE_SUCCESS) { SSqlFuncExprMsg** pExprMsg = NULL;
if ((code = convertQueryMsg(pQueryTableMsg, &pTableIdList, &pExprMsg)) != TSDB_CODE_SUCCESS) {
return code; return code;
} }
@ -6206,7 +6136,7 @@ int32_t qCreateQueryInfo(void* tsdb, SQueryTableMsg *pQueryTableMsg, SQInfo **pQ
} }
SSqlFunctionExpr *pExprs = NULL; SSqlFunctionExpr *pExprs = NULL;
if ((code = createSqlFunctionExprFromMsg(pQueryTableMsg, &pExprs)) != TSDB_CODE_SUCCESS) { if ((code = createSqlFunctionExprFromMsg(pQueryTableMsg, &pExprs, pExprMsg)) != TSDB_CODE_SUCCESS) {
goto _query_over; goto _query_over;
} }
@ -6261,15 +6191,119 @@ int32_t qRetrieveQueryResultInfo(SQInfo *pQInfo, int32_t *numOfRows, int32_t *ro
sem_wait(&pQInfo->dataReady); sem_wait(&pQInfo->dataReady);
SQuery *pQuery = pQInfo->runtimeEnv.pQuery; SQuery *pQuery = pQInfo->runtimeEnv.pQuery;
*numOfRows = pQInfo->rec.pointsRead;
*rowsize = pQuery->rowSize;
// *numOfRows = pQInfo->rec.pointsRead; dTrace("QInfo:%p, retrieve res info, rowsize:%d, rows:%d, code:%d", pQInfo, *rowsize, *numOfRows, pQInfo->code);
// *rowsize = pQuery->rowSize;
*numOfRows = 1;
// dTrace("QInfo:%p, retrieve data info completed, precision:%d, rowsize:%d, rows:%d, code:%d", pQInfo, *timePrec,
// *rowsize, *numOfRows, pQInfo->code);
if (pQInfo->code < 0) { // less than 0 means there are error existed. if (pQInfo->code < 0) { // less than 0 means there are error existed.
return -pQInfo->code; 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 (pObj == NULL || pObj->maxSessions == 0) return;
if (string == NULL || string[0] == 0) return; if (string == NULL || string[0] == 0) return;
return;
hash = (*(pObj->hashFp))(pObj, string); hash = (*(pObj->hashFp))(pObj, string);
pthread_mutex_lock(&pObj->mutex); pthread_mutex_lock(&pObj->mutex);

40
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 uint16_t flag; // denotes reversed scan of data or not
int16_t order; int16_t order;
STimeWindow window; // the primary query time window that applies to all queries STimeWindow window; // the primary query time window that applies to all queries
TSKEY lastKey;
int32_t blockBufferSize; int32_t blockBufferSize;
SCompBlock *pBlock; SCompBlock *pBlock;
int32_t numOfBlocks; int32_t numOfBlocks;
@ -265,7 +264,19 @@ tsdb_query_handle_t *tsdbQueryByTableId(tsdb_repo_t* tsdb, STsdbQueryCond *pCond
pQueryHandle->loadDataAfterSeek = false; pQueryHandle->loadDataAfterSeek = false;
pQueryHandle->isFirstSlot = true; 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 // malloc buffer in order to load data from file
int32_t numOfCols = taosArrayGetSize(pColumnInfo); 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; return (tsdb_query_handle_t)pQueryHandle;
} }
static int32_t next = 1;
bool tsdbNextDataBlock(tsdb_query_handle_t *pQueryHandle) { 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; 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, static int tsdbReadRowsFromCache(SSkipListIterator *pIter, TSKEY maxKey, int maxRowsToRead,
@ -344,9 +362,7 @@ SDataBlockInfo tsdbRetrieveDataBlockInfo(tsdb_query_handle_t *pQueryHandle) {
STsdbQueryHandle* pHandle = (STsdbQueryHandle*) pQueryHandle; STsdbQueryHandle* pHandle = (STsdbQueryHandle*) pQueryHandle;
STableIdInfo* idInfo = taosArrayGet(pHandle->pTableIdList, 0); STableIdInfo* idInfo = taosArrayGet(pHandle->pTableIdList, 0);
STableId tableId = {.uid = idInfo->uid, .tid = idInfo->sid}; STable *pTable = pHandle->pTableQueryInfo->pTableObj;
STable *pTable = tsdbIsValidTableToInsert(tsdbGetMeta(pHandle->pTsdb), tableId);
assert(pTable != NULL);
TSKEY skey = 0, ekey = 0; TSKEY skey = 0, ekey = 0;
int32_t rows = 0; int32_t rows = 0;
@ -357,8 +373,8 @@ SDataBlockInfo tsdbRetrieveDataBlockInfo(tsdb_query_handle_t *pQueryHandle) {
} }
SDataBlockInfo blockInfo = { SDataBlockInfo blockInfo = {
.uid = tableId.uid, .uid = idInfo->uid,
.sid = tableId.tid, .sid = idInfo->sid,
.size = rows, .size = rows,
.window = {.skey = skey, .ekey = ekey} .window = {.skey = skey, .ekey = ekey}
}; };