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

[td-225]

This commit is contained in:
Haojun Liao 2021-03-25 11:59:55 +08:00
parent 0ee8b40f93
commit 8796999068
23 changed files with 431 additions and 368 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
src/client/inc/tscUtil.h
View File

@ -294,6 +294,8 @@ uint32_t tscGetTableMetaMaxSize();
int32_t tscCreateTableMetaFromCChildMeta(STableMeta* pChild, const char* name);
STableMeta* tscTableMetaDup(STableMeta* pTableMeta);
void tsCreateSQLFunctionCtx(SQueryInfo* pQueryInfo, SQLFunctionCtx* pCtx);
void* malloc_throw(size_t size);
void* calloc_throw(size_t nmemb, size_t size);

18
src/client/inc/tsclient.h
View File

@ -35,6 +35,7 @@ extern "C" {
#include "qExecutor.h"
#include "qSqlparser.h"
#include "qTsbuf.h"
#include "qUtil.h"
#include "tcmdtype.h"
// forward declaration
@ -96,21 +97,6 @@ typedef struct STableMetaInfo {
SArray *tagColList; // SArray<SColumn*>, involved tag columns
} STableMetaInfo;
/* the structure for sql function in select clause */
typedef struct SSqlExpr {
char aliasName[TSDB_COL_NAME_LEN]; // as aliasName
SColIndex colInfo;
uint64_t uid; // refactor use the pointer
int16_t functionId; // function id in aAgg array
int16_t resType; // return value type
int16_t resBytes; // length of return value
int32_t interBytes; // inter result buffer size
int16_t numOfParams; // argument value of each function
tVariant param[3]; // parameters are not more than 3
int32_t offset; // sub result column value of arithmetic expression.
int16_t resColId; // result column id
} SSqlExpr;
typedef struct SColumnIndex {
int16_t tableIndex;
int16_t columnIndex;
@ -290,7 +276,7 @@ typedef struct {
char * pRsp;
int32_t rspType;
int32_t rspLen;
uint64_t qhandle;
uint64_t qid;
int64_t useconds;
int64_t offset; // offset value from vnode during projection query of stable
int32_t row;

9
src/client/src/tscAsync.c
View File

@ -161,8 +161,8 @@ static void tscProcessAsyncRetrieveImpl(void *param, TAOS_RES *tres, int numOfRo
SSqlCmd *pCmd = &pSql->cmd;
SSqlRes *pRes = &pSql->res;
if ((pRes->qhandle == 0 || numOfRows != 0) && pCmd->command < TSDB_SQL_LOCAL) {
if (pRes->qhandle == 0 && numOfRows != 0) {
if ((pRes->qid == 0 || numOfRows != 0) && pCmd->command < TSDB_SQL_LOCAL) {
if (pRes->qid == 0 && numOfRows != 0) {
tscError("qhandle is NULL");
} else {
pRes->code = numOfRows;
@ -210,10 +210,9 @@ void taos_fetch_rows_a(TAOS_RES *tres, __async_cb_func_t fp, void *param) {
pSql->fp = tscAsyncFetchRowsProxy;
pSql->param = param;
if (pRes->qhandle == 0) {
tscError("qhandle is NULL");
if (pRes->qid == 0) {
tscError("qhandle is invalid");
pRes->code = TSDB_CODE_TSC_INVALID_QHANDLE;
tscAsyncResultOnError(pSql);
return;
}

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

@ -309,7 +309,7 @@ TAOS_ROW tscFetchRow(void *param) {
SSqlCmd *pCmd = &pSql->cmd;
SSqlRes *pRes = &pSql->res;
if (pRes->qhandle == 0 ||
if (pRes->qid == 0 ||
pCmd->command == TSDB_SQL_RETRIEVE_EMPTY_RESULT ||
pCmd->command == TSDB_SQL_INSERT) {
return NULL;
@ -905,7 +905,7 @@ int tscProcessLocalCmd(SSqlObj *pSql) {
* set the qhandle to be 1 in order to pass the qhandle check, and to call partial release function to
* free allocated resources and remove the SqlObj from sql query linked list
*/
pRes->qhandle = 0x1;
pRes->qid = 0x1;
pRes->numOfRows = 0;
} else if (pCmd->command == TSDB_SQL_SHOW_CREATE_TABLE) {
pRes->code = tscProcessShowCreateTable(pSql);

93
src/client/src/tscLocalMerge.c
View File

@ -56,82 +56,83 @@ int32_t treeComparator(const void *pLeft, const void *pRight, void *param) {
}
}
static void tscInitSqlContext(SSqlCmd *pCmd, SLocalMerger *pReducer, tOrderDescriptor *pDesc) {
/*
* the fields and offset attributes in pCmd and pModel may be different due to
* merge requirement. So, the final result in pRes structure is formatted in accordance with the pCmd object.
*/
SQueryInfo *pQueryInfo = tscGetQueryInfo(pCmd, pCmd->clauseIndex);
// todo merge with vnode side function
void tsCreateSQLFunctionCtx(SQueryInfo* pQueryInfo, SQLFunctionCtx* pCtx) {
size_t size = tscSqlExprNumOfExprs(pQueryInfo);
for (int32_t i = 0; i < size; ++i) {
SQLFunctionCtx *pCtx = &pReducer->pCtx[i];
SSqlExpr * pExpr = tscSqlExprGet(pQueryInfo, i);
SSqlExpr *pExpr = tscSqlExprGet(pQueryInfo, i);
pCtx->pOutput = pReducer->pResultBuf->data + pExpr->offset * pReducer->resColModel->capacity;
pCtx->order = pQueryInfo->order.order;
pCtx->functionId = pExpr->functionId;
pCtx[i].order = pQueryInfo->order.order;
pCtx[i].functionId = pExpr->functionId;
// input buffer hold only one point data
int16_t offset = getColumnModelOffset(pDesc->pColumnModel, i);
SSchema *pSchema = getColumnModelSchema(pDesc->pColumnModel, i);
pCtx[i].inputType = pExpr->colType;
pCtx[i].inputBytes = pExpr->colBytes;
pCtx->pInput = pReducer->pTempBuffer->data + offset;
pCtx[i].outputBytes = pExpr->resBytes;
pCtx[i].outputType = pExpr->resType;
// input data format comes from pModel
pCtx->inputType = pSchema->type;
pCtx->inputBytes = pSchema->bytes;
// output data format yet comes from pCmd.
pCtx->outputBytes = pExpr->resBytes;
pCtx->outputType = pExpr->resType;
pCtx->size = 1;
pCtx->hasNull = true;
pCtx->currentStage = MERGE_STAGE;
pCtx[i].size = 1;
pCtx[i].hasNull = true;
pCtx[i].currentStage = MERGE_STAGE;
// for top/bottom function, the output of timestamp is the first column
int32_t functionId = pExpr->functionId;
if (functionId == TSDB_FUNC_TOP || functionId == TSDB_FUNC_BOTTOM || functionId == TSDB_FUNC_DIFF) {
pCtx->ptsOutputBuf = pReducer->pCtx[0].pOutput;
pCtx->param[2].i64 = pQueryInfo->order.order;
pCtx->param[2].nType = TSDB_DATA_TYPE_BIGINT;
pCtx->param[1].i64 = pQueryInfo->order.orderColId;
pCtx[i].ptsOutputBuf = pCtx[0].pOutput;
pCtx[i].param[2].i64 = pQueryInfo->order.order;
pCtx[i].param[2].nType = TSDB_DATA_TYPE_BIGINT;
pCtx[i].param[1].i64 = pQueryInfo->order.orderColId;
} else if (functionId == TSDB_FUNC_APERCT) {
pCtx->param[0].i64 = pExpr->param[0].i64;
pCtx->param[0].nType = pExpr->param[0].nType;
pCtx[i].param[0].i64 = pExpr->param[0].i64;
pCtx[i].param[0].nType = pExpr->param[0].nType;
} else if (functionId == TSDB_FUNC_BLKINFO) {
pCtx->param[0].i64 = pExpr->param[0].i64;
pCtx->param[0].nType = pExpr->param[0].nType;
pCtx->numOfParams = 1;
pCtx[i].param[0].i64 = pExpr->param[0].i64;
pCtx[i].param[0].nType = pExpr->param[0].nType;
pCtx[i].numOfParams = 1;
}
pCtx->interBufBytes = pExpr->interBytes;
pCtx->resultInfo = calloc(1, pCtx->interBufBytes + sizeof(SResultRowCellInfo));
pCtx->stableQuery = true;
pCtx[i].interBufBytes = pExpr->interBytes;
pCtx[i].resultInfo = calloc(1, pCtx[i].interBufBytes + sizeof(SResultRowCellInfo));
pCtx[i].stableQuery = true;
}
int16_t n = 0;
int16_t tagLen = 0;
SQLFunctionCtx **pTagCtx = calloc(pQueryInfo->fieldsInfo.numOfOutput, POINTER_BYTES);
SQLFunctionCtx *pCtx = NULL;
SQLFunctionCtx *pCtx1 = NULL;
for (int32_t i = 0; i < pQueryInfo->fieldsInfo.numOfOutput; ++i) {
SSqlExpr *pExpr = tscSqlExprGet(pQueryInfo, i);
if (pExpr->functionId == TSDB_FUNC_TAG_DUMMY || pExpr->functionId == TSDB_FUNC_TS_DUMMY) {
tagLen += pExpr->resBytes;
pTagCtx[n++] = &pReducer->pCtx[i];
pTagCtx[n++] = &pCtx[i];
} else if ((aAggs[pExpr->functionId].status & TSDB_FUNCSTATE_SELECTIVITY) != 0) {
pCtx = &pReducer->pCtx[i];
pCtx1 = &pCtx[i];
}
}
if (n == 0 || pCtx == NULL) {
free(pTagCtx);
} else {
pCtx->tagInfo.pTagCtxList = pTagCtx;
pCtx->tagInfo.numOfTagCols = n;
pCtx->tagInfo.tagsLen = tagLen;
pCtx1->tagInfo.pTagCtxList = pTagCtx;
pCtx1->tagInfo.numOfTagCols = n;
pCtx1->tagInfo.tagsLen = tagLen;
}
}
static void setCtxInputOutputBuffer(SQueryInfo* pQueryInfo, SQLFunctionCtx *pCtx, SLocalMerger *pReducer,
tOrderDescriptor *pDesc) {
size_t size = tscSqlExprNumOfExprs(pQueryInfo);
for (int32_t i = 0; i < size; ++i) {
SSqlExpr *pExpr = tscSqlExprGet(pQueryInfo, i);
pCtx[i].pOutput = pReducer->pResultBuf->data + pExpr->offset * pReducer->resColModel->capacity;
// input buffer hold only one point data
int16_t offset = getColumnModelOffset(pDesc->pColumnModel, i);
pCtx[i].pInput = pReducer->pTempBuffer->data + offset;
}
}
@ -362,8 +363,8 @@ void tscCreateLocalMerger(tExtMemBuffer **pMemBuffer, int32_t numOfBuffer, tOrde
pReducer->pTempBuffer->num = 0;
tscCreateResPointerInfo(pRes, pQueryInfo);
tscInitSqlContext(pCmd, pReducer, pDesc);
tsCreateSQLFunctionCtx(pQueryInfo, pReducer->pCtx);
setCtxInputOutputBuffer(pQueryInfo, pReducer->pCtx, pReducer, pDesc);
// we change the capacity of schema to denote that there is only one row in temp buffer
pReducer->pDesc->pColumnModel->capacity = 1;
@ -1607,7 +1608,7 @@ void tscInitResObjForLocalQuery(SSqlObj *pObj, int32_t numOfRes, int32_t rowLen)
tscDestroyLocalMerger(pObj);
}
pRes->qhandle = 1; // hack to pass the safety check in fetch_row function
pRes->qid = 1; // hack to pass the safety check in fetch_row function
pRes->numOfRows = 0;
pRes->row = 0;

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

@ -903,7 +903,7 @@ int taos_stmt_prepare(TAOS_STMT* stmt, const char* sql, unsigned long length) {
return TSDB_CODE_TSC_OUT_OF_MEMORY;
}
pRes->qhandle = 0;
pRes->qid = 0;
pRes->numOfRows = 1;
strtolower(pSql->sqlstr, sql);

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

@ -250,7 +250,7 @@ int tscBuildQueryStreamDesc(void *pMsg, STscObj *pObj) {
pQdesc->queryId = htonl(pSql->queryId);
//pQdesc->useconds = htobe64(pSql->res.useconds);
pQdesc->useconds = htobe64(now - pSql->stime);
pQdesc->qHandle = htobe64(pSql->res.qhandle);
pQdesc->qid = htobe64(pSql->res.qid);
pHeartbeat->numOfQueries++;
pQdesc++;

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

@ -1490,9 +1490,9 @@ static int32_t handleArithmeticExpr(SSqlCmd* pCmd, int32_t clauseIndex, int32_t
SExprInfo* pArithExprInfo = calloc(1, sizeof(SExprInfo));
// arithmetic expression always return result in the format of double float
pArithExprInfo->bytes = sizeof(double);
pArithExprInfo->interBytes = sizeof(double);
pArithExprInfo->type = TSDB_DATA_TYPE_DOUBLE;
pArithExprInfo->base.resBytes = sizeof(double);
pArithExprInfo->base.interBytes = sizeof(double);
pArithExprInfo->base.resType = TSDB_DATA_TYPE_DOUBLE;
pArithExprInfo->base.functionId = TSDB_FUNC_ARITHM;
pArithExprInfo->base.numOfParams = 1;
@ -1517,10 +1517,10 @@ static int32_t handleArithmeticExpr(SSqlCmd* pCmd, int32_t clauseIndex, int32_t
// TODO: other error handling
} END_TRY
SSqlFuncMsg* pFuncMsg = &pInfo->pArithExprInfo->base;
pFuncMsg->arg[0].argBytes = (int16_t) tbufTell(&bw);
pFuncMsg->arg[0].argValue.pz = tbufGetData(&bw, true);
pFuncMsg->arg[0].argType = TSDB_DATA_TYPE_BINARY;
SSqlExpr* pFuncMsg = &pInfo->pArithExprInfo->base;
pFuncMsg->param[0].nLen = (int16_t) tbufTell(&bw);
pFuncMsg->param[0].pz = tbufGetData(&bw, true);
pFuncMsg->param[0].nType = TSDB_DATA_TYPE_BINARY;
// tbufCloseWriter(&bw); // TODO there is a memory leak
}
@ -6631,6 +6631,10 @@ static STableMeta* extractTempTableMetaFromNestQuery(SQueryInfo* pUpstream) {
return meta;
}
//static SColumnInfo* getColumnInfoFromSchema(SQueryInfo* pUpstream) {
//
//}
int32_t validateSqlNode(SSqlObj* pSql, SQuerySqlNode* pQuerySqlNode, int32_t index) {
assert(pQuerySqlNode != NULL && (pQuerySqlNode->from == NULL || taosArrayGetSize(pQuerySqlNode->from->tableList) > 0));

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

@ -510,8 +510,9 @@ int tscBuildFetchMsg(SSqlObj *pSql, SSqlInfo *pInfo) {
SRetrieveTableMsg *pRetrieveMsg = (SRetrieveTableMsg *) pSql->cmd.payload;
SQueryInfo *pQueryInfo = tscGetActiveQueryInfo(&pSql->cmd);
pRetrieveMsg->free = htons(pQueryInfo->type);
pRetrieveMsg->qhandle = htobe64(pSql->res.qhandle);
pRetrieveMsg->free = htons(pQueryInfo->type);
pRetrieveMsg->qid = htobe64(pSql->res.qid);
// todo valid the vgroupId at the client side
STableMetaInfo* pTableMetaInfo = tscGetMetaInfo(pQueryInfo, 0);
@ -523,7 +524,7 @@ int tscBuildFetchMsg(SSqlObj *pSql, SSqlInfo *pInfo) {
assert(pVgroupInfo->vgroups[vgIndex].vgId > 0 && vgIndex < pTableMetaInfo->vgroupList->numOfVgroups);
pRetrieveMsg->header.vgId = htonl(pVgroupInfo->vgroups[vgIndex].vgId);
tscDebug("%p build fetch msg from vgId:%d, vgIndex:%d, qhandle:%" PRIX64, pSql, pVgroupInfo->vgroups[vgIndex].vgId, vgIndex, pSql->res.qhandle);
tscDebug("%p build fetch msg from vgId:%d, vgIndex:%d, qid:%" PRIu64, pSql, pVgroupInfo->vgroups[vgIndex].vgId, vgIndex, pSql->res.qid);
} else {
int32_t numOfVgroups = (int32_t)taosArrayGetSize(pTableMetaInfo->pVgroupTables);
assert(vgIndex >= 0 && vgIndex < numOfVgroups);
@ -531,12 +532,12 @@ int tscBuildFetchMsg(SSqlObj *pSql, SSqlInfo *pInfo) {
SVgroupTableInfo* pTableIdList = taosArrayGet(pTableMetaInfo->pVgroupTables, vgIndex);
pRetrieveMsg->header.vgId = htonl(pTableIdList->vgInfo.vgId);
tscDebug("%p build fetch msg from vgId:%d, vgIndex:%d, qhandle:%" PRIX64, pSql, pTableIdList->vgInfo.vgId, vgIndex, pSql->res.qhandle);
tscDebug("%p build fetch msg from vgId:%d, vgIndex:%d, qid:%" PRIu64, pSql, pTableIdList->vgInfo.vgId, vgIndex, pSql->res.qid);
}
} else {
STableMeta* pTableMeta = pTableMetaInfo->pTableMeta;
pRetrieveMsg->header.vgId = htonl(pTableMeta->vgId);
tscDebug("%p build fetch msg from only one vgroup, vgId:%d, qhandle:%" PRIX64, pSql, pTableMeta->vgId, pSql->res.qhandle);
tscDebug("%p build fetch msg from only one vgroup, vgId:%d, qid:%" PRIu64, pSql, pTableMeta->vgId, pSql->res.qid);
}
pSql->cmd.payloadLen = sizeof(SRetrieveTableMsg);
@ -592,7 +593,7 @@ static int32_t tscEstimateQueryMsgSize(SSqlObj *pSql, int32_t clauseIndex) {
int32_t srcColListSize = (int32_t)(taosArrayGetSize(pQueryInfo->colList) * sizeof(SColumnInfo));
size_t numOfExprs = tscSqlExprNumOfExprs(pQueryInfo);
int32_t exprSize = (int32_t)(sizeof(SSqlFuncMsg) * numOfExprs * 2);
int32_t exprSize = (int32_t)(sizeof(SSqlExpr) * numOfExprs * 2);
int32_t tsBufSize = (pQueryInfo->tsBuf != NULL) ? pQueryInfo->tsBuf->fileSize : 0;
int32_t sqlLen = (int32_t) strlen(pSql->sqlstr) + 1;
@ -818,7 +819,8 @@ int tscBuildQueryMsg(SSqlObj *pSql, SSqlInfo *pInfo) {
}
}
SSqlFuncMsg *pSqlFuncExpr = (SSqlFuncMsg *)pMsg;
SSqlExpr *pSqlFuncExpr = (SSqlExpr *)pMsg;
for (int32_t i = 0; i < tscSqlExprNumOfExprs(pQueryInfo); ++i) {
SSqlExpr *pExpr = tscSqlExprGet(pQueryInfo, i);
@ -840,43 +842,41 @@ int tscBuildQueryMsg(SSqlObj *pSql, SSqlInfo *pInfo) {
pSqlFuncExpr->colInfo.colIndex = htons(pExpr->colInfo.colIndex);
pSqlFuncExpr->colInfo.flag = htons(pExpr->colInfo.flag);
if (TSDB_COL_IS_UD_COL(pExpr->colInfo.flag)) {
pSqlFuncExpr->colType = htons(pExpr->resType);
pSqlFuncExpr->colBytes = htons(pExpr->resBytes);
} else if (pExpr->colInfo.colId == TSDB_TBNAME_COLUMN_INDEX) {
SSchema *s = tGetTbnameColumnSchema();
pSqlFuncExpr->colType = htons(pExpr->colType);
pSqlFuncExpr->colBytes = htons(pExpr->colBytes);
pSqlFuncExpr->colType = htons(s->type);
pSqlFuncExpr->colBytes = htons(s->bytes);
if (TSDB_COL_IS_UD_COL(pExpr->colInfo.flag) || pExpr->colInfo.colId == TSDB_TBNAME_COLUMN_INDEX) {
pSqlFuncExpr->resType = htons(pExpr->resType);
pSqlFuncExpr->resBytes = htons(pExpr->resBytes);
} else if (pExpr->colInfo.colId == TSDB_BLOCK_DIST_COLUMN_INDEX) {
SSchema s = tGetBlockDistColumnSchema();
pSqlFuncExpr->colType = htons(s.type);
pSqlFuncExpr->colBytes = htons(s.bytes);
pSqlFuncExpr->resType = htons(s.type);
pSqlFuncExpr->resBytes = htons(s.bytes);
} else {
SSchema* s = tscGetColumnSchemaById(pTableMeta, pExpr->colInfo.colId);
pSqlFuncExpr->colType = htons(s->type);
pSqlFuncExpr->colBytes = htons(s->bytes);
pSqlFuncExpr->resType = htons(s->type);
pSqlFuncExpr->resBytes = htons(s->bytes);
}
pSqlFuncExpr->functionId = htons(pExpr->functionId);
pSqlFuncExpr->numOfParams = htons(pExpr->numOfParams);
pSqlFuncExpr->resColId = htons(pExpr->resColId);
pMsg += sizeof(SSqlFuncMsg);
pMsg += sizeof(SSqlExpr);
for (int32_t j = 0; j < pExpr->numOfParams; ++j) { // todo add log
pSqlFuncExpr->arg[j].argType = htons((uint16_t)pExpr->param[j].nType);
pSqlFuncExpr->arg[j].argBytes = htons(pExpr->param[j].nLen);
pSqlFuncExpr->param[j].nType = htons((uint16_t)pExpr->param[j].nType);
pSqlFuncExpr->param[j].nLen = htons(pExpr->param[j].nLen);
if (pExpr->param[j].nType == TSDB_DATA_TYPE_BINARY) {
memcpy(pMsg, pExpr->param[j].pz, pExpr->param[j].nLen);
pMsg += pExpr->param[j].nLen;
} else {
pSqlFuncExpr->arg[j].argValue.i64 = htobe64(pExpr->param[j].i64);
pSqlFuncExpr->param[j].i64 = htobe64(pExpr->param[j].i64);
}
}
pSqlFuncExpr = (SSqlFuncMsg *)pMsg;
pSqlFuncExpr = (SSqlExpr *)pMsg;
}
size_t output = tscNumOfFields(pQueryInfo);
@ -884,7 +884,7 @@ int tscBuildQueryMsg(SSqlObj *pSql, SSqlInfo *pInfo) {
if (tscIsSecondStageQuery(pQueryInfo)) {
pQueryMsg->secondStageOutput = htonl((int32_t) output);
SSqlFuncMsg *pSqlFuncExpr1 = (SSqlFuncMsg *)pMsg;
SSqlExpr *pExpr1 = (SSqlExpr *)pMsg;
for (int32_t i = 0; i < output; ++i) {
SInternalField* pField = tscFieldInfoGetInternalField(&pQueryInfo->fieldsInfo, i);
@ -904,49 +904,56 @@ int tscBuildQueryMsg(SSqlObj *pSql, SSqlInfo *pInfo) {
return TSDB_CODE_TSC_INVALID_SQL;
}
pSqlFuncExpr1->numOfParams = 0; // no params for projection query
pSqlFuncExpr1->functionId = htons(TSDB_FUNC_PRJ);
pSqlFuncExpr1->colInfo.colId = htons(pExpr->resColId);
pSqlFuncExpr1->colInfo.flag = htons(TSDB_COL_NORMAL);
pExpr1->numOfParams = 0; // no params for projection query
pExpr1->functionId = htons(TSDB_FUNC_PRJ);
pExpr1->colInfo.colId = htons(pExpr->resColId);
pExpr1->colInfo.flag = htons(TSDB_COL_NORMAL);
pExpr1->colType = htons(pExpr->resType);
pExpr1->colBytes = htons(pExpr->resBytes);
bool assign = false;
for (int32_t f = 0; f < tscSqlExprNumOfExprs(pQueryInfo); ++f) {
SSqlExpr *pe = tscSqlExprGet(pQueryInfo, f);
if (pe == pExpr) {
pSqlFuncExpr1->colInfo.colIndex = htons(f);
pSqlFuncExpr1->colType = htons(pe->resType);
pSqlFuncExpr1->colBytes = htons(pe->resBytes);
pExpr1->colInfo.colIndex = htons(f);
pExpr1->resType = htons(pe->resType);
pExpr1->resBytes = htons(pe->resBytes);
assign = true;
break;
}
}
assert(assign);
pMsg += sizeof(SSqlFuncMsg);
pSqlFuncExpr1 = (SSqlFuncMsg *)pMsg;
pMsg += sizeof(SSqlExpr);
pExpr1 = (SSqlExpr *)pMsg;
} else {
assert(pField->pArithExprInfo != NULL);
SExprInfo* pExprInfo = pField->pArithExprInfo;
pSqlFuncExpr1->colInfo.colId = htons(pExprInfo->base.colInfo.colId);
pSqlFuncExpr1->functionId = htons(pExprInfo->base.functionId);
pSqlFuncExpr1->numOfParams = htons(pExprInfo->base.numOfParams);
pMsg += sizeof(SSqlFuncMsg);
pExpr1->colInfo.colId = htons(pExprInfo->base.colInfo.colId);
pExpr1->colType = htons(pExprInfo->base.colType);
pExpr1->colBytes = htons(pExprInfo->base.colBytes);
pExpr1->resBytes = htons(sizeof(double));
pExpr1->resType = htons(TSDB_DATA_TYPE_DOUBLE);
pExpr1->functionId = htons(pExprInfo->base.functionId);
pExpr1->numOfParams = htons(pExprInfo->base.numOfParams);
pMsg += sizeof(SSqlExpr);
for (int32_t j = 0; j < pExprInfo->base.numOfParams; ++j) {
// todo add log
pSqlFuncExpr1->arg[j].argType = htons((uint16_t)pExprInfo->base.arg[j].argType);
pSqlFuncExpr1->arg[j].argBytes = htons(pExprInfo->base.arg[j].argBytes);
pExpr1->param[j].nType = htons((uint16_t)pExprInfo->base.param[j].nType);
pExpr1->param[j].nLen = htons(pExprInfo->base.param[j].nLen);
if (pExprInfo->base.arg[j].argType == TSDB_DATA_TYPE_BINARY) {
memcpy(pMsg, pExprInfo->base.arg[j].argValue.pz, pExprInfo->base.arg[j].argBytes);
pMsg += pExprInfo->base.arg[j].argBytes;
if (pExprInfo->base.param[j].nType == TSDB_DATA_TYPE_BINARY) {
memcpy(pMsg, pExprInfo->base.param[j].pz, pExprInfo->base.param[j].nLen);
pMsg += pExprInfo->base.param[j].nLen;
} else {
pSqlFuncExpr1->arg[j].argValue.i64 = htobe64(pExprInfo->base.arg[j].argValue.i64);
pExpr1->param[j].i64 = htobe64(pExprInfo->base.param[j].i64);
}
}
pSqlFuncExpr1 = (SSqlFuncMsg *)pMsg;
pExpr1 = (SSqlExpr *)pMsg;
}
}
} else {
@ -1561,7 +1568,7 @@ int tscBuildRetrieveFromMgmtMsg(SSqlObj *pSql, SSqlInfo *pInfo) {
SQueryInfo *pQueryInfo = tscGetQueryInfo(pCmd, 0);
SRetrieveTableMsg *pRetrieveMsg = (SRetrieveTableMsg*)pCmd->payload;
pRetrieveMsg->qhandle = htobe64(pSql->res.qhandle);
pRetrieveMsg->qid = htobe64(pSql->res.qid);
pRetrieveMsg->free = htons(pQueryInfo->type);
return TSDB_CODE_SUCCESS;
@ -2107,7 +2114,7 @@ int tscProcessShowRsp(SSqlObj *pSql) {
pShow = (SShowRsp *)pRes->pRsp;
pShow->qhandle = htobe64(pShow->qhandle);
pRes->qhandle = pShow->qhandle;
pRes->qid = pShow->qhandle;
tscResetForNextRetrieve(pRes);
pMetaMsg = &(pShow->tableMeta);
@ -2289,10 +2296,11 @@ int tscProcessQueryRsp(SSqlObj *pSql) {
SSqlRes *pRes = &pSql->res;
SQueryTableRsp *pQuery = (SQueryTableRsp *)pRes->pRsp;
pQuery->qhandle = htobe64(pQuery->qhandle);
pRes->qhandle = pQuery->qhandle;
pQuery->qid = htobe64(pQuery->qid);
pRes->qid = pQuery->qid;
pRes->data = NULL;
tscResetForNextRetrieve(pRes);
return 0;
}

8
src/client/src/tscSql.c
View File

@ -476,7 +476,7 @@ TAOS_ROW taos_fetch_row(TAOS_RES *res) {
SSqlCmd *pCmd = &pSql->cmd;
SSqlRes *pRes = &pSql->res;
if (pRes->qhandle == 0 ||
if (pRes->qid == 0 ||
pRes->code == TSDB_CODE_TSC_QUERY_CANCELLED ||
pCmd->command == TSDB_SQL_RETRIEVE_EMPTY_RESULT ||
pCmd->command == TSDB_SQL_INSERT) {
@ -508,7 +508,7 @@ int taos_fetch_block(TAOS_RES *res, TAOS_ROW *rows) {
SSqlCmd *pCmd = &pSql->cmd;
SSqlRes *pRes = &pSql->res;
if (pRes->qhandle == 0 ||
if (pRes->qid == 0 ||
pRes->code == TSDB_CODE_TSC_QUERY_CANCELLED ||
pCmd->command == TSDB_SQL_RETRIEVE_EMPTY_RESULT ||
pCmd->command == TSDB_SQL_INSERT) {
@ -554,7 +554,7 @@ static bool tscKillQueryInDnode(SSqlObj* pSql) {
SSqlCmd* pCmd = &pSql->cmd;
SSqlRes* pRes = &pSql->res;
if (pRes == NULL || pRes->qhandle == 0) {
if (pRes == NULL || pRes->qid == 0) {
return true;
}
@ -1050,7 +1050,7 @@ int taos_load_table_info(TAOS *taos, const char *tableNameList) {
* If qhandle is NOT set 0, the function of taos_free_result() will send message to server by calling tscProcessSql()
* to free connection, which may cause segment fault, when the parse phrase is not even successfully executed.
*/
pRes->qhandle = 0;
pRes->qid = 0;
free(str);
if (code != TSDB_CODE_SUCCESS) {

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

@ -149,7 +149,7 @@ static SSub* tscCreateSubscription(STscObj* pObj, const char* topic, const char*
}
strtolower(pSql->sqlstr, pSql->sqlstr);
pRes->qhandle = 0;
pRes->qid = 0;
pRes->numOfRows = 1;
code = tscAllocPayload(pCmd, TSDB_DEFAULT_PAYLOAD_SIZE);
@ -448,7 +448,7 @@ SSqlObj* recreateSqlObj(SSub* pSub) {
return NULL;
}
pRes->qhandle = 0;
pRes->qid = 0;
pRes->numOfRows = 1;
int code = tscAllocPayload(pCmd, TSDB_DEFAULT_PAYLOAD_SIZE);
@ -546,7 +546,7 @@ TAOS_RES *taos_consume(TAOS_SUB *tsub) {
uint32_t type = pQueryInfo->type;
tscFreeSqlResult(pSql);
pRes->numOfRows = 1;
pRes->qhandle = 0;
pRes->qid = 0;
pSql->cmd.command = TSDB_SQL_SELECT;
pQueryInfo->type = type;

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

@ -1584,7 +1584,7 @@ int32_t tscCreateJoinSubquery(SSqlObj *pSql, int16_t tableIndex, SJoinSupporter
SSqlCmd * pCmd = &pSql->cmd;
SQueryInfo *pQueryInfo = tscGetQueryInfo(pCmd, pCmd->clauseIndex);
pSql->res.qhandle = 0x1;
pSql->res.qid = 0x1;
assert(pSql->res.numOfRows == 0);
if (pSql->pSubs == NULL) {
@ -2179,7 +2179,7 @@ int32_t tscHandleMasterSTableQuery(SSqlObj *pSql) {
SColumnModel *pModel = NULL;
SColumnModel *pFinalModel = NULL;
pRes->qhandle = 0x1; // hack the qhandle check
pRes->qid = 0x1; // hack the qhandle check
const uint32_t nBufferSize = (1u << 16u); // 64KB
@ -2727,7 +2727,7 @@ void tscRetrieveDataRes(void *param, TAOS_RES *tres, int code) {
tscDebug("%p sub:%p query complete, ep:%s, vgId:%d, orderOfSub:%d, retrieve data", trsupport->pParentSql, pSql,
pVgroup->epAddr[0].fqdn, pVgroup->vgId, trsupport->subqueryIndex);
if (pSql->res.qhandle == 0) { // qhandle is NULL, code is TSDB_CODE_SUCCESS means no results generated from this vnode
if (pSql->res.qid == 0) { // qhandle is 0, code is TSDB_CODE_SUCCESS means no results generated from this vnode
tscRetrieveFromDnodeCallBack(param, pSql, 0);
} else {
taos_fetch_rows_a(tres, tscRetrieveFromDnodeCallBack, param);

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

@ -371,11 +371,39 @@ void tscSetResRawPtr(SSqlRes* pRes, SQueryInfo* pQueryInfo) {
}
}
static SColumnInfo* extractColumnInfoFromResult(STableMeta* pTableMeta, SArray* pTableCols) {
int32_t numOfCols = taosArrayGetSize(pTableCols);
SColumnInfo* pColInfo = calloc(numOfCols, sizeof(SColumnInfo));
SSchema *pSchema = pTableMeta->schema;
for(int32_t i = 0; i < numOfCols; ++i) {
SColumn* pCol = taosArrayGetP(pTableCols, i);
int32_t index = pCol->colIndex.columnIndex;
pColInfo[i].type = pSchema[index].type;
pColInfo[i].bytes = pSchema[index].bytes;
pColInfo[i].colId = pSchema[index].colId;
}
return pColInfo;
}
void prepareInputDataFromUpstream(SSqlRes* pRes, SQueryInfo* pQueryInfo) {
if (pQueryInfo->pDownstream != NULL && taosArrayGetSize(pQueryInfo->pDownstream) > 0) {
// handle the following query process
SQueryInfo* px = taosArrayGetP(pQueryInfo->pDownstream, 0);
printf("%d\n", px->type);
SColumnInfo* colInfo = extractColumnInfoFromResult(px->pTableMetaInfo[0]->pTableMeta, px->colList);
int32_t numOfOutput = tscSqlExprNumOfExprs(px);
SExprInfo *exprInfo = NULL;
SQLFunctionCtx *pCtx = calloc(numOfOutput, sizeof(SQLFunctionCtx));
int32_t numOfCols = taosArrayGetSize(px->colList);
SQueriedTableInfo info = {.colList = colInfo, .numOfCols = numOfCols,};
/*int32_t code = */createQueryFunc(&info, numOfOutput, &exprInfo, px->exprList->pData, NULL, px->type, NULL);
tsCreateSQLFunctionCtx(px, pCtx);
}
}
@ -1095,11 +1123,9 @@ void tscFieldInfoClear(SFieldInfo* pFieldInfo) {
if (pInfo->pArithExprInfo != NULL) {
tExprTreeDestroy(pInfo->pArithExprInfo->pExpr, NULL);
SSqlFuncMsg* pFuncMsg = &pInfo->pArithExprInfo->base;
SSqlExpr* pFuncMsg = &pInfo->pArithExprInfo->base;
for(int32_t j = 0; j < pFuncMsg->numOfParams; ++j) {
if (pFuncMsg->arg[j].argType == TSDB_DATA_TYPE_BINARY) {
tfree(pFuncMsg->arg[j].argValue.pz);
}
tVariantDestroy(&pFuncMsg->param[j]);
}
tfree(pInfo->pArithExprInfo);
@ -1125,20 +1151,33 @@ static SSqlExpr* doCreateSqlExpr(SQueryInfo* pQueryInfo, int16_t functionId, SCo
// set the correct columnIndex index
if (pColIndex->columnIndex == TSDB_TBNAME_COLUMN_INDEX) {
SSchema* s = tGetTbnameColumnSchema();
pExpr->colInfo.colId = TSDB_TBNAME_COLUMN_INDEX;
pExpr->colBytes = s->bytes;
pExpr->colType = s->type;
} else if (pColIndex->columnIndex == TSDB_BLOCK_DIST_COLUMN_INDEX) {
SSchema s = tGetBlockDistColumnSchema();
pExpr->colInfo.colId = TSDB_BLOCK_DIST_COLUMN_INDEX;
pExpr->colBytes = s.bytes;
pExpr->colType = s.type;
} else if (pColIndex->columnIndex <= TSDB_UD_COLUMN_INDEX) {
pExpr->colInfo.colId = pColIndex->columnIndex;
pExpr->colBytes = size;
pExpr->colType = type;
} else {
if (TSDB_COL_IS_TAG(colType)) {
SSchema* pSchema = tscGetTableTagSchema(pTableMetaInfo->pTableMeta);
pExpr->colInfo.colId = pSchema[pColIndex->columnIndex].colId;
pExpr->colBytes = pSchema[pColIndex->columnIndex].bytes;
pExpr->colType = pSchema[pColIndex->columnIndex].type;
tstrncpy(pExpr->colInfo.name, pSchema[pColIndex->columnIndex].name, sizeof(pExpr->colInfo.name));
} else if (pTableMetaInfo->pTableMeta != NULL) {
// in handling select database/version/server_status(), the pTableMeta is NULL
SSchema* pSchema = tscGetTableColumnSchema(pTableMetaInfo->pTableMeta, pColIndex->columnIndex);
pExpr->colInfo.colId = pSchema->colId;
pExpr->colBytes = pSchema->bytes;
pExpr->colType = pSchema->type;
tstrncpy(pExpr->colInfo.name, pSchema->name, sizeof(pExpr->colInfo.name));
}
}
@ -1769,7 +1808,7 @@ void tscInitQueryInfo(SQueryInfo* pQueryInfo) {
pQueryInfo->exprList = taosArrayInit(4, POINTER_BYTES);
pQueryInfo->colList = taosArrayInit(4, POINTER_BYTES);
pQueryInfo->udColumnId = TSDB_UD_COLUMN_INDEX;
pQueryInfo->resColumnId = -1000;
pQueryInfo->resColumnId = TSDB_RES_COL_ID;
pQueryInfo->limit.limit = -1;
pQueryInfo->limit.offset = 0;
@ -2128,6 +2167,8 @@ SSqlObj* createSubqueryObj(SSqlObj* pSql, int16_t tableIndex, __async_cb_func_t
SQueryInfo* pQueryInfo = tscGetQueryInfo(pCmd, pCmd->clauseIndex);
pNewQueryInfo->command = pQueryInfo->command;
pnCmd->active = pNewQueryInfo;
memcpy(&pNewQueryInfo->interval, &pQueryInfo->interval, sizeof(pNewQueryInfo->interval));
pNewQueryInfo->type = pQueryInfo->type;
pNewQueryInfo->window = pQueryInfo->window;

27
src/common/inc/tname.h
View File

@ -41,6 +41,33 @@ typedef struct SResPair {
double avg;
} SResPair;
/* the structure for sql function in select clause */
typedef struct SSqlExpr {
char aliasName[TSDB_COL_NAME_LEN]; // as aliasName
SColIndex colInfo;
uint64_t uid; // refactor use the pointer
int16_t functionId; // function id in aAgg array
int16_t resType; // return value type
int16_t resBytes; // length of return value
int32_t interBytes; // inter result buffer size
int16_t colType; // table column type
int16_t colBytes; // table column bytes
int16_t numOfParams; // argument value of each function
tVariant param[3]; // parameters are not more than 3
int32_t offset; // sub result column value of arithmetic expression.
int16_t resColId; // result column id
} SSqlExpr;
typedef struct SExprInfo {
SSqlExpr base;
int64_t uid;
struct tExprNode* pExpr;
} SExprInfo;
#define TSDB_DB_NAME_T 1
#define TSDB_TABLE_NAME_T 2

5
src/inc/taosdef.h
View File

@ -243,8 +243,9 @@ do { \
#define TSDB_MAX_REPLICA 5
#define TSDB_TBNAME_COLUMN_INDEX (-1)
#define TSDB_BLOCK_DIST_COLUMN_INDEX (-2)
#define TSDB_UD_COLUMN_INDEX (-100)
#define TSDB_BLOCK_DIST_COLUMN_INDEX (-2)
#define TSDB_UD_COLUMN_INDEX (-1000)
#define TSDB_RES_COL_ID (-5000)
#define TSDB_MULTI_TABLEMETA_MAX_NUM 100000 // maximum batch size allowed to load table meta

55
src/inc/taosmsg.h
View File

@ -398,35 +398,26 @@ typedef struct SColIndex {
} SColIndex;
/* sql function msg, to describe the message to vnode about sql function
* operations in select clause */
typedef struct SSqlFuncMsg {
int16_t functionId;
int16_t numOfParams;
int16_t resColId; // result column id, id of the current output column
int16_t colType;
int16_t colBytes;
SColIndex colInfo;
struct ArgElem {
int16_t argType;
int16_t argBytes;
union {
double d;
int64_t i64;
char * pz;
} argValue;
} arg[3];
} SSqlFuncMsg;
typedef struct SExprInfo {
SSqlFuncMsg base;
struct tExprNode* pExpr;
int16_t bytes;
int16_t type;
int32_t interBytes;
int64_t uid;
} SExprInfo;
* operations in select */
//typedef struct SSqlFuncMsg {
// int16_t functionId;
// int16_t numOfParams;
//
// int16_t resColId; // result column id, id of the current output column
// int16_t colType;
// int16_t colBytes;
//
// SColIndex colInfo;
// struct ArgElem {
// int16_t argType;
// int16_t argBytes;
// union {
// double d;
// int64_t i64;
// char * pz;
// } argValue;
// } arg[3];
//} SSqlFuncMsg;
typedef struct SColumnFilterInfo {
int16_t lowerRelOptr;
@ -513,12 +504,12 @@ typedef struct {
typedef struct {
int32_t code;
uint64_t qhandle; // query handle
uint64_t qid; // dnode generated query id
} SQueryTableRsp;
typedef struct {
SMsgHead header;
uint64_t qhandle;
union{uint64_t qid; uint64_t qhandle;};
uint16_t free;
} SRetrieveTableMsg;
@ -818,7 +809,7 @@ typedef struct {
uint32_t queryId;
int64_t useconds;
int64_t stime;
uint64_t qHandle;
uint64_t qid;
} SQueryDesc;
typedef struct {

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

@ -344,7 +344,7 @@ static int32_t mnodeGetQueryMeta(STableMetaMsg *pMeta, SShowObj *pShow, void *pC
pShow->bytes[cols] = 24;
pSchema[cols].type = TSDB_DATA_TYPE_BINARY;
strcpy(pSchema[cols].name, "qhandle");
strcpy(pSchema[cols].name, "qid");
pSchema[cols].bytes = htons(pShow->bytes[cols]);
cols++;
@ -420,7 +420,7 @@ static int32_t mnodeRetrieveQueries(SShowObj *pShow, char *data, int32_t rows, v
cols++;
char handleBuf[24] = {0};
snprintf(handleBuf, tListLen(handleBuf), "%p", (void*)htobe64(pDesc->qHandle));
snprintf(handleBuf, tListLen(handleBuf), "%p", (void*)htobe64(pDesc->qid));
pWrite = data + pShow->offset[cols] * rows + pShow->bytes[cols] * numOfRows;
STR_WITH_MAXSIZE_TO_VARSTR(pWrite, handleBuf, pShow->bytes[cols]);

17
src/query/inc/qExecutor.h
View File

@ -306,6 +306,12 @@ enum {
QUERY_RESULT_READY = 2,
};
typedef struct {
int32_t numOfTags;
int32_t numOfCols;
SColumnInfo *colList;
} SQueriedTableInfo;
typedef struct SQInfo {
void* signature;
uint64_t qId;
@ -331,8 +337,8 @@ typedef struct SQueryParam {
char *tbnameCond;
char *prevResult;
SArray *pTableIdList;
SSqlFuncMsg **pExprMsg;
SSqlFuncMsg **pSecExprMsg;
SSqlExpr **pExpr;
SSqlExpr **pSecExpr;
SExprInfo *pExprs;
SExprInfo *pSecExprs;
@ -423,10 +429,11 @@ typedef struct SSWindowOperatorInfo {
void freeParam(SQueryParam *param);
int32_t convertQueryMsg(SQueryTableMsg *pQueryMsg, SQueryParam* param);
int32_t createQueryFuncExprFromMsg(SQueryTableMsg *pQueryMsg, int32_t numOfOutput, SExprInfo **pExprInfo, SSqlFuncMsg **pExprMsg,
SColumnInfo* pTagCols);
int32_t createQueryFunc(SQueriedTableInfo* pTableInfo, int32_t numOfOutput, SExprInfo** pExprInfo,
SSqlExpr** pExprMsg, SColumnInfo* pTagCols, int32_t queryType, void* pMsg);
int32_t createIndirectQueryFuncExprFromMsg(SQueryTableMsg *pQueryMsg, int32_t numOfOutput, SExprInfo **pExprInfo,
SSqlFuncMsg **pExprMsg, SExprInfo *prevExpr);
SSqlExpr **pExpr, SExprInfo *prevExpr);
SSqlGroupbyExpr *createGroupbyExprFromMsg(SQueryTableMsg *pQueryMsg, SColIndex *pColIndex, int32_t *code);
SQInfo *createQInfoImpl(SQueryTableMsg *pQueryMsg, SSqlGroupbyExpr *pGroupbyExpr, SExprInfo *pExprs,

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

@ -27,7 +27,7 @@
#define GET_RES_WINDOW_KEY_LEN(_l) ((_l) + sizeof(uint64_t))
#define curTimeWindowIndex(_winres) ((_winres)->curIndex)
#define GET_ROW_PARAM_FOR_MULTIOUTPUT(_q, tbq, sq) (((tbq) && (!(sq)))? (_q)->pExpr1[1].base.arg->argValue.i64:1)
#define GET_ROW_PARAM_FOR_MULTIOUTPUT(_q, tbq, sq) (((tbq) && (!(sq)))? (_q)->pExpr1[1].base.param[0].i64:1)
int32_t getOutputInterResultBufSize(SQuery* pQuery);

323
src/query/src/qExecutor.c
View File

@ -214,8 +214,8 @@ static SSDataBlock* createOutputBuf(SExprInfo* pExpr, int32_t numOfOutput, int32
res->pDataBlock = taosArrayInit(numOfOutput, sizeof(SColumnInfoData));
for (int32_t i = 0; i < numOfOutput; ++i) {
SColumnInfoData idata = {{0}};
idata.info.type = pExpr[i].type;
idata.info.bytes = pExpr[i].bytes;
idata.info.type = pExpr[i].base.resType;
idata.info.bytes = pExpr[i].base.resBytes;
idata.info.colId = pExpr[i].base.resColId;
idata.pData = calloc(1, MAX(idata.info.bytes * numOfRows, minSize)); // at least to hold a pointer on x64 platform
@ -1607,10 +1607,10 @@ static SQLFunctionCtx* createSQLFunctionCtx(SQueryRuntimeEnv* pRuntimeEnv, SExpr
}
for (int32_t i = 0; i < numOfOutput; ++i) {
SSqlFuncMsg *pSqlFuncMsg = &pExpr[i].base;
SSqlExpr *pSqlExpr = &pExpr[i].base;
SQLFunctionCtx* pCtx = &pFuncCtx[i];
SColIndex *pIndex = &pSqlFuncMsg->colInfo;
SColIndex *pIndex = &pSqlExpr->colInfo;
if (TSDB_COL_REQ_NULL(pIndex->flag)) {
pCtx->requireNull = true;
@ -1619,33 +1619,33 @@ static SQLFunctionCtx* createSQLFunctionCtx(SQueryRuntimeEnv* pRuntimeEnv, SExpr
pCtx->requireNull = false;
}
pCtx->inputBytes = pSqlFuncMsg->colBytes;
pCtx->inputType = pSqlFuncMsg->colType;
pCtx->inputBytes = pSqlExpr->colBytes;
pCtx->inputType = pSqlExpr->colType;
pCtx->ptsOutputBuf = NULL;
pCtx->outputBytes = pExpr[i].bytes;
pCtx->outputType = pExpr[i].type;
pCtx->outputBytes = pSqlExpr->resBytes;
pCtx->outputType = pSqlExpr->resType;
pCtx->order = pQuery->order.order;
pCtx->functionId = pSqlFuncMsg->functionId;
pCtx->functionId = pSqlExpr->functionId;
pCtx->stableQuery = pQuery->stableQuery;
pCtx->interBufBytes = pExpr[i].interBytes;
pCtx->interBufBytes = pSqlExpr->interBytes;
pCtx->start.key = INT64_MIN;
pCtx->end.key = INT64_MIN;
pCtx->numOfParams = pSqlFuncMsg->numOfParams;
pCtx->numOfParams = pSqlExpr->numOfParams;
for (int32_t j = 0; j < pCtx->numOfParams; ++j) {
int16_t type = pSqlFuncMsg->arg[j].argType;
int16_t bytes = pSqlFuncMsg->arg[j].argBytes;
if (pSqlFuncMsg->functionId == TSDB_FUNC_STDDEV_DST) {
int16_t type = pSqlExpr->param[j].nType;
int16_t bytes = pSqlExpr->param[j].nLen;
if (pSqlExpr->functionId == TSDB_FUNC_STDDEV_DST) {
continue;
}
if (type == TSDB_DATA_TYPE_BINARY || type == TSDB_DATA_TYPE_NCHAR) {
tVariantCreateFromBinary(&pCtx->param[j], pSqlFuncMsg->arg[j].argValue.pz, bytes, type);
tVariantCreateFromBinary(&pCtx->param[j], pSqlExpr->param[j].pz, bytes, type);
} else {
tVariantCreateFromBinary(&pCtx->param[j], (char *)&pSqlFuncMsg->arg[j].argValue.i64, bytes, type);
tVariantCreateFromBinary(&pCtx->param[j], (char *)&pSqlExpr->param[j].i64, bytes, type);
}
}
@ -1688,7 +1688,7 @@ static SQLFunctionCtx* createSQLFunctionCtx(SQueryRuntimeEnv* pRuntimeEnv, SExpr
for(int32_t i = 1; i < numOfOutput; ++i) {
(*rowCellInfoOffset)[i] = (int32_t)((*rowCellInfoOffset)[i - 1] + sizeof(SResultRowCellInfo) +
pExpr[i - 1].interBytes * GET_ROW_PARAM_FOR_MULTIOUTPUT(pQuery, pQuery->topBotQuery, pQuery->stableQuery));
pExpr[i - 1].base.interBytes * GET_ROW_PARAM_FOR_MULTIOUTPUT(pQuery, pQuery->topBotQuery, pQuery->stableQuery));
}
setCtxTagColumnInfo(pFuncCtx, numOfOutput);
@ -1801,7 +1801,7 @@ static int32_t setupQueryRuntimeEnv(SQueryRuntimeEnv *pRuntimeEnv, int32_t numOf
setTableScanFilterOperatorInfo(pRuntimeEnv->pTableScanner->info, pRuntimeEnv->proot);
if (pQuery->pExpr2 != NULL) {
if (pQuery->pExpr2 != NULL && !pQuery->stableQuery) {
pRuntimeEnv->proot = createArithOperatorInfo(pRuntimeEnv, pRuntimeEnv->proot, pQuery->pExpr2, pQuery->numOfExpr2);
}
} else { // diff/add/multiply/subtract/division
@ -1916,19 +1916,19 @@ static bool isFixedOutputQuery(SQuery* pQuery) {
}
for (int32_t i = 0; i < pQuery->numOfOutput; ++i) {
SSqlFuncMsg *pExprMsg = &pQuery->pExpr1[i].base;
SSqlExpr *pExpr = &pQuery->pExpr1[i].base;
// ignore the ts_comp function
if (i == 0 && pExprMsg->functionId == TSDB_FUNC_PRJ && pExprMsg->numOfParams == 1 &&
pExprMsg->colInfo.colIndex == PRIMARYKEY_TIMESTAMP_COL_INDEX) {
if (i == 0 && pExpr->functionId == TSDB_FUNC_PRJ && pExpr->numOfParams == 1 &&
pExpr->colInfo.colIndex == PRIMARYKEY_TIMESTAMP_COL_INDEX) {
continue;
}
if (pExprMsg->functionId == TSDB_FUNC_TS || pExprMsg->functionId == TSDB_FUNC_TS_DUMMY) {
if (pExpr->functionId == TSDB_FUNC_TS || pExpr->functionId == TSDB_FUNC_TS_DUMMY) {
continue;
}
if (!IS_MULTIOUTPUT(aAggs[pExprMsg->functionId].status)) {
if (!IS_MULTIOUTPUT(aAggs[pExpr->functionId].status)) {
return true;
}
}
@ -1989,7 +1989,7 @@ static bool needReverseScan(SQuery *pQuery) {
if (functionId == TSDB_FUNC_LAST || functionId == TSDB_FUNC_LAST_DST) {
// the scan order to acquire the last result of the specified column
int32_t order = (int32_t)pQuery->pExpr1[i].base.arg->argValue.i64;
int32_t order = (int32_t)pQuery->pExpr1[i].base.param[0].i64;
if (order != pQuery->order.order) {
return true;
}
@ -2048,12 +2048,12 @@ static void setScanLimitationByResultBuffer(SQuery *pQuery) {
} else {
bool hasMultioutput = false;
for (int32_t i = 0; i < pQuery->numOfOutput; ++i) {
SSqlFuncMsg *pExprMsg = &pQuery->pExpr1[i].base;
if (pExprMsg->functionId == TSDB_FUNC_TS || pExprMsg->functionId == TSDB_FUNC_TS_DUMMY) {
SSqlExpr *pExpr = &pQuery->pExpr1[i].base;
if (pExpr->functionId == TSDB_FUNC_TS || pExpr->functionId == TSDB_FUNC_TS_DUMMY) {
continue;
}
hasMultioutput = IS_MULTIOUTPUT(aAggs[pExprMsg->functionId].status);
hasMultioutput = IS_MULTIOUTPUT(aAggs[pExpr->functionId].status);
if (!hasMultioutput) {
break;
}
@ -2554,7 +2554,7 @@ int32_t loadDataBlockOnDemand(SQueryRuntimeEnv* pRuntimeEnv, STableScanInfo* pTa
if (pQuery->stableQuery) { // todo refactor
SExprInfo* pExprInfo = &pTableScanInfo->pExpr[0];
int16_t tagId = (int16_t)pExprInfo->base.arg->argValue.i64;
int16_t tagId = (int16_t)pExprInfo->base.param[0].i64;
SColumnInfo* pColInfo = doGetTagColumnInfoById(pQuery->tagColList, pQuery->numOfTags, tagId);
// compare tag first
@ -2781,7 +2781,7 @@ void setTagValue(SOperatorInfo* pOperatorInfo, void *pTable, SQLFunctionCtx* pCt
if (pQuery->numOfOutput == 1 && pExprInfo->base.functionId == TSDB_FUNC_TS_COMP && pQuery->stableQuery) {
assert(pExprInfo->base.numOfParams == 1);
int16_t tagColId = (int16_t)pExprInfo->base.arg->argValue.i64;
int16_t tagColId = (int16_t)pExprInfo->base.param[0].i64;
SColumnInfo* pColInfo = doGetTagColumnInfoById(pQuery->tagColList, pQuery->numOfTags, tagColId);
doSetTagValueInParam(pTable, tagColId, &pCtx[0].tag, pColInfo->type, pColInfo->bytes);
@ -2800,16 +2800,16 @@ void setTagValue(SOperatorInfo* pOperatorInfo, void *pTable, SQLFunctionCtx* pCt
}
// todo use tag column index to optimize performance
doSetTagValueInParam(pTable, pLocalExprInfo->base.colInfo.colId, &pCtx[idx].tag, pLocalExprInfo->type,
pLocalExprInfo->bytes);
doSetTagValueInParam(pTable, pLocalExprInfo->base.colInfo.colId, &pCtx[idx].tag, pLocalExprInfo->base.resType,
pLocalExprInfo->base.resBytes);
if (IS_NUMERIC_TYPE(pLocalExprInfo->type) || pLocalExprInfo->type == TSDB_DATA_TYPE_BOOL) {
memcpy(pRuntimeEnv->tagVal + offset, &pCtx[idx].tag.i64, pLocalExprInfo->bytes);
if (IS_NUMERIC_TYPE(pLocalExprInfo->base.resType) || pLocalExprInfo->base.resType == TSDB_DATA_TYPE_BOOL) {
memcpy(pRuntimeEnv->tagVal + offset, &pCtx[idx].tag.i64, pLocalExprInfo->base.resBytes);
} else {
memcpy(pRuntimeEnv->tagVal + offset, pCtx[idx].tag.pz, pCtx[idx].tag.nLen);
}
offset += pLocalExprInfo->bytes;
offset += pLocalExprInfo->base.resBytes;
}
//todo : use index to avoid iterator all possible output columns
@ -2901,25 +2901,25 @@ void UNUSED_FUNC displayInterResult(tFilePage **pdata, SQueryRuntimeEnv* pRuntim
for (int32_t j = 0; j < numOfRows; ++j) {
for (int32_t i = 0; i < numOfCols; ++i) {
switch (pQuery->pExpr1[i].type) {
switch (pQuery->pExpr1[i].base.resType) {
case TSDB_DATA_TYPE_BINARY: {
int32_t type = pQuery->pExpr1[i].type;
printBinaryData(pQuery->pExpr1[i].base.functionId, pdata[i]->data + pQuery->pExpr1[i].bytes * j,
int32_t type = pQuery->pExpr1[i].base.resType;
printBinaryData(pQuery->pExpr1[i].base.functionId, pdata[i]->data + pQuery->pExpr1[i].base.resBytes * j,
type);
break;
}
case TSDB_DATA_TYPE_TIMESTAMP:
case TSDB_DATA_TYPE_BIGINT:
printf("%" PRId64 "\t", *(int64_t *)(pdata[i]->data + pQuery->pExpr1[i].bytes * j));
printf("%" PRId64 "\t", *(int64_t *)(pdata[i]->data + pQuery->pExpr1[i].base.resBytes * j));
break;
case TSDB_DATA_TYPE_INT:
printf("%d\t", *(int32_t *)(pdata[i]->data + pQuery->pExpr1[i].bytes * j));
printf("%d\t", *(int32_t *)(pdata[i]->data + pQuery->pExpr1[i].base.resBytes * j));
break;
case TSDB_DATA_TYPE_FLOAT:
printf("%f\t", *(float *)(pdata[i]->data + pQuery->pExpr1[i].bytes * j));
printf("%f\t", *(float *)(pdata[i]->data + pQuery->pExpr1[i].base.resBytes * j));
break;
case TSDB_DATA_TYPE_DOUBLE:
printf("%lf\t", *(double *)(pdata[i]->data + pQuery->pExpr1[i].bytes * j));
printf("%lf\t", *(double *)(pdata[i]->data + pQuery->pExpr1[i].base.resBytes * j));
break;
}
}
@ -3298,13 +3298,13 @@ void setResultOutputBuf(SQueryRuntimeEnv *pRuntimeEnv, SResultRow *pResult, SQLF
void setCtxTagForJoin(SQueryRuntimeEnv* pRuntimeEnv, SQLFunctionCtx* pCtx, SExprInfo* pExprInfo, void* pTable) {
SQuery* pQuery = pRuntimeEnv->pQuery;
SSqlFuncMsg* pFuncMsg = &pExprInfo->base;
SSqlExpr* pExpr = &pExprInfo->base;
if (pQuery->stableQuery && (pRuntimeEnv->pTsBuf != NULL) &&
(pFuncMsg->functionId == TSDB_FUNC_TS || pFuncMsg->functionId == TSDB_FUNC_PRJ) &&
(pFuncMsg->colInfo.colIndex == PRIMARYKEY_TIMESTAMP_COL_INDEX)) {
assert(pFuncMsg->numOfParams == 1);
(pExpr->functionId == TSDB_FUNC_TS || pExpr->functionId == TSDB_FUNC_PRJ) &&
(pExpr->colInfo.colIndex == PRIMARYKEY_TIMESTAMP_COL_INDEX)) {
assert(pExpr->numOfParams == 1);
int16_t tagColId = (int16_t)pExprInfo->base.arg->argValue.i64;
int16_t tagColId = (int16_t)pExprInfo->base.param[0].i64;
SColumnInfo* pColInfo = doGetTagColumnInfoById(pQuery->tagColList, pQuery->numOfTags, tagColId);
doSetTagValueInParam(pTable, tagColId, &pCtx->tag, pColInfo->type, pColInfo->bytes);
@ -3312,10 +3312,10 @@ void setCtxTagForJoin(SQueryRuntimeEnv* pRuntimeEnv, SQLFunctionCtx* pCtx, SExpr
int16_t tagType = pCtx[0].tag.nType;
if (tagType == TSDB_DATA_TYPE_BINARY || tagType == TSDB_DATA_TYPE_NCHAR) {
qDebug("QInfo:%p set tag value for join comparison, colId:%" PRId64 ", val:%s", pRuntimeEnv->qinfo,
pExprInfo->base.arg->argValue.i64, pCtx[0].tag.pz);
pExprInfo->base.param[0].i64, pCtx[0].tag.pz);
} else {
qDebug("QInfo:%p set tag value for join comparison, colId:%" PRId64 ", val:%" PRId64, pRuntimeEnv->qinfo,
pExprInfo->base.arg->argValue.i64, pCtx[0].tag.i64);
pExprInfo->base.param[0].i64, pCtx[0].tag.i64);
}
}
}
@ -3362,17 +3362,17 @@ int32_t setTimestampListJoinInfo(SQueryRuntimeEnv* pRuntimeEnv, tVariant* pTag,
return 0;
}
void setParamForStableStddev(SQueryRuntimeEnv* pRuntimeEnv, SQLFunctionCtx* pCtx, int32_t numOfOutput, SExprInfo* pExpr) {
void setParamForStableStddev(SQueryRuntimeEnv* pRuntimeEnv, SQLFunctionCtx* pCtx, int32_t numOfOutput, SExprInfo* pExprInfo) {
SQuery* pQuery = pRuntimeEnv->pQuery;
int32_t numOfExprs = pQuery->numOfOutput;
for(int32_t i = 0; i < numOfExprs; ++i) {
SExprInfo* pExprInfo = &(pExpr[i]);
SExprInfo* pExprInfo1 = &(pExprInfo[i]);
if (pExprInfo->base.functionId != TSDB_FUNC_STDDEV_DST) {
continue;
}
SSqlFuncMsg* pFuncMsg = &pExprInfo->base;
SSqlExpr* pExpr = &pExprInfo1->base;
pCtx[i].param[0].arr = NULL;
pCtx[i].param[0].nType = TSDB_DATA_TYPE_INT; // avoid freeing the memory by setting the type to be int
@ -3385,7 +3385,7 @@ void setParamForStableStddev(SQueryRuntimeEnv* pRuntimeEnv, SQLFunctionCtx* pCtx
int32_t numOfCols = (int32_t)taosArrayGetSize(p->pResult);
for (int32_t k = 0; k < numOfCols; ++k) {
SStddevInterResult* pres = taosArrayGet(p->pResult, k);
if (pres->colId == pFuncMsg->colInfo.colId) {
if (pres->colId == pExpr->colInfo.colId) {
pCtx[i].param[0].arr = pres->pResult;
break;
}
@ -3943,15 +3943,15 @@ static SFillColInfo* createFillColInfo(SExprInfo* pExpr, int32_t numOfOutput, in
for(int32_t i = 0; i < numOfOutput; ++i) {
SExprInfo* pExprInfo = &pExpr[i];
pFillCol[i].col.bytes = pExprInfo->bytes;
pFillCol[i].col.type = (int8_t)pExprInfo->type;
pFillCol[i].col.bytes = pExprInfo->base.resBytes;
pFillCol[i].col.type = (int8_t)pExprInfo->base.resBytes;
pFillCol[i].col.offset = offset;
pFillCol[i].tagIndex = -2;
pFillCol[i].flag = TSDB_COL_NORMAL; // always be ta normal column for table query
pFillCol[i].functionId = pExprInfo->base.functionId;
pFillCol[i].fillVal.i = fillVal[i];
offset += pExprInfo->bytes;
offset += pExprInfo->base.resBytes;
}
return pFillCol;
@ -5211,12 +5211,12 @@ static SSDataBlock* doTagScan(void* param) {
assert(pQuery->numOfOutput == 1);
SExprInfo* pExprInfo = &pOperator->pExpr[0];
int32_t rsize = pExprInfo->bytes;
int32_t rsize = pExprInfo->base.resBytes;
count = 0;
int16_t bytes = pExprInfo->bytes;
int16_t type = pExprInfo->type;
int16_t bytes = pExprInfo->base.resBytes;
int16_t type = pExprInfo->base.resType;
for(int32_t i = 0; i < pQuery->numOfTags; ++i) {
if (pQuery->tagColList[i].colId == pExprInfo->base.colInfo.colId) {
@ -5287,8 +5287,8 @@ static SSDataBlock* doTagScan(void* param) {
}
SColumnInfoData* pColInfo = taosArrayGet(pRes->pDataBlock, j);
type = pExprInfo[j].type;
bytes = pExprInfo[j].bytes;
type = pExprInfo[j].base.resType;
bytes = pExprInfo[j].base.resBytes;
if (pExprInfo[j].base.colInfo.colId == TSDB_TBNAME_COLUMN_INDEX) {
data = tsdbGetTableName(item->pTable);
@ -5296,7 +5296,7 @@ static SSDataBlock* doTagScan(void* param) {
data = tsdbGetTableTagVal(item->pTable, pExprInfo[j].base.colInfo.colId, type, bytes);
}
dst = pColInfo->pData + count * pExprInfo[j].bytes;
dst = pColInfo->pData + count * pExprInfo[j].base.resBytes;
doSetTagValueToResultBuf(dst, data, type, bytes);
}
@ -5339,30 +5339,30 @@ SOperatorInfo* createTagScanOperatorInfo(SQueryRuntimeEnv* pRuntimeEnv, SExprInf
return pOperator;
}
static int32_t getColumnIndexInSource(SQueryTableMsg *pQueryMsg, SSqlFuncMsg *pExprMsg, SColumnInfo* pTagCols) {
static int32_t getColumnIndexInSource(SQueriedTableInfo *pTableInfo, SSqlExpr *pExpr, SColumnInfo* pTagCols) {
int32_t j = 0;
if (TSDB_COL_IS_TAG(pExprMsg->colInfo.flag)) {
if (pExprMsg->colInfo.colId == TSDB_TBNAME_COLUMN_INDEX) {
if (TSDB_COL_IS_TAG(pExpr->colInfo.flag)) {
if (pExpr->colInfo.colId == TSDB_TBNAME_COLUMN_INDEX) {
return TSDB_TBNAME_COLUMN_INDEX;
} else if (pExprMsg->colInfo.colId == TSDB_BLOCK_DIST_COLUMN_INDEX) {
} else if (pExpr->colInfo.colId == TSDB_BLOCK_DIST_COLUMN_INDEX) {
return TSDB_BLOCK_DIST_COLUMN_INDEX;
}
while(j < pQueryMsg->numOfTags) {
if (pExprMsg->colInfo.colId == pTagCols[j].colId) {
while(j < pTableInfo->numOfTags) {
if (pExpr->colInfo.colId == pTagCols[j].colId) {
return j;
}
j += 1;
}
} else if (TSDB_COL_IS_UD_COL(pExprMsg->colInfo.flag)) { // user specified column data
} else if (TSDB_COL_IS_UD_COL(pExpr->colInfo.flag)) { // user specified column data
return TSDB_UD_COLUMN_INDEX;
} else {
while (j < pQueryMsg->numOfCols) {
if (pExprMsg->colInfo.colId == pQueryMsg->colList[j].colId) {
while (j < pTableInfo->numOfCols) {
if (pExpr->colInfo.colId == pTableInfo->colList[j].colId) {
return j;
}
@ -5373,8 +5373,8 @@ static int32_t getColumnIndexInSource(SQueryTableMsg *pQueryMsg, SSqlFuncMsg *pE
return INT32_MIN; // return a less than TSDB_TBNAME_COLUMN_INDEX value
}
bool validateExprColumnInfo(SQueryTableMsg *pQueryMsg, SSqlFuncMsg *pExprMsg, SColumnInfo* pTagCols) {
int32_t j = getColumnIndexInSource(pQueryMsg, pExprMsg, pTagCols);
bool validateExprColumnInfo(SQueriedTableInfo *pTableInfo, SSqlExpr *pExpr, SColumnInfo* pTagCols) {
int32_t j = getColumnIndexInSource(pTableInfo, pExpr, pTagCols);
return j != INT32_MIN;
}
@ -5413,21 +5413,21 @@ static bool validateQueryMsg(SQueryTableMsg *pQueryMsg) {
return true;
}
static bool validateQuerySourceCols(SQueryTableMsg *pQueryMsg, SSqlFuncMsg** pExprMsg, SColumnInfo* pTagCols) {
int32_t numOfTotal = pQueryMsg->numOfCols + pQueryMsg->numOfTags;
if (pQueryMsg->numOfCols < 0 || pQueryMsg->numOfTags < 0 || numOfTotal > TSDB_MAX_COLUMNS) {
qError("qmsg:%p illegal value of numOfCols %d numOfTags:%d", pQueryMsg, pQueryMsg->numOfCols, pQueryMsg->numOfTags);
static UNUSED_FUNC bool validateQueryTableCols(SQueriedTableInfo* pTableInfo, SSqlExpr** pExpr, int32_t numOfOutput,
SColumnInfo* pTagCols, void* pMsg) {
int32_t numOfTotal = pTableInfo->numOfCols + pTableInfo->numOfTags;
if (pTableInfo->numOfCols < 0 || pTableInfo->numOfTags < 0 || numOfTotal > TSDB_MAX_COLUMNS) {
qError("qmsg:%p illegal value of numOfCols %d numOfTags:%d", pMsg, pTableInfo->numOfCols, pTableInfo->numOfTags);
return false;
}
if (numOfTotal == 0) {
for(int32_t i = 0; i < pQueryMsg->numOfOutput; ++i) {
SSqlFuncMsg* pFuncMsg = pExprMsg[i];
if ((pFuncMsg->functionId == TSDB_FUNC_TAGPRJ) ||
(pFuncMsg->functionId == TSDB_FUNC_TID_TAG && pFuncMsg->colInfo.colId == TSDB_TBNAME_COLUMN_INDEX) ||
(pFuncMsg->functionId == TSDB_FUNC_COUNT && pFuncMsg->colInfo.colId == TSDB_TBNAME_COLUMN_INDEX) ||
(pFuncMsg->functionId == TSDB_FUNC_BLKINFO)) {
if (numOfTotal == 0) { // table total columns are not required.
for(int32_t i = 0; i < numOfOutput; ++i) {
SSqlExpr* p = pExpr[i];
if ((p->functionId == TSDB_FUNC_TAGPRJ) ||
(p->functionId == TSDB_FUNC_TID_TAG && p->colInfo.colId == TSDB_TBNAME_COLUMN_INDEX) ||
(p->functionId == TSDB_FUNC_COUNT && p->colInfo.colId == TSDB_TBNAME_COLUMN_INDEX) ||
(p->functionId == TSDB_FUNC_BLKINFO)) {
continue;
}
@ -5435,8 +5435,8 @@ static bool validateQuerySourceCols(SQueryTableMsg *pQueryMsg, SSqlFuncMsg** pEx
}
}
for(int32_t i = 0; i < pQueryMsg->numOfOutput; ++i) {
if (!validateExprColumnInfo(pQueryMsg, pExprMsg[i], pTagCols)) {
for(int32_t i = 0; i < numOfOutput; ++i) {
if (!validateExprColumnInfo(pTableInfo, pExpr[i], pTagCols)) {
return TSDB_CODE_QRY_INVALID_MSG;
}
}
@ -5569,38 +5569,41 @@ int32_t convertQueryMsg(SQueryTableMsg *pQueryMsg, SQueryParam* param) {
}
}
param->pExprMsg = calloc(pQueryMsg->numOfOutput, POINTER_BYTES);
if (param->pExprMsg == NULL) {
param->pExpr = calloc(pQueryMsg->numOfOutput, POINTER_BYTES);
if (param->pExpr == NULL) {
code = TSDB_CODE_QRY_OUT_OF_MEMORY;
goto _cleanup;
}
SSqlFuncMsg *pExprMsg = (SSqlFuncMsg *)pMsg;
SSqlExpr *pExprMsg = (SSqlExpr *)pMsg;
for (int32_t i = 0; i < pQueryMsg->numOfOutput; ++i) {
param->pExprMsg[i] = pExprMsg;
param->pExpr[i] = pExprMsg;
pExprMsg->colInfo.colIndex = htons(pExprMsg->colInfo.colIndex);
pExprMsg->colInfo.colId = htons(pExprMsg->colInfo.colId);
pExprMsg->colInfo.flag = htons(pExprMsg->colInfo.flag);
pExprMsg->colType = htons(pExprMsg->colType);
pExprMsg->colBytes = htons(pExprMsg->colBytes);
pExprMsg->colType = htons(pExprMsg->colType);
pExprMsg->resType = htons(pExprMsg->resType);
pExprMsg->resBytes = htons(pExprMsg->resBytes);
pExprMsg->functionId = htons(pExprMsg->functionId);
pExprMsg->numOfParams = htons(pExprMsg->numOfParams);
pExprMsg->resColId = htons(pExprMsg->resColId);
pMsg += sizeof(SSqlFuncMsg);
pMsg += sizeof(SSqlExpr);
for (int32_t j = 0; j < pExprMsg->numOfParams; ++j) {
pExprMsg->arg[j].argType = htons(pExprMsg->arg[j].argType);
pExprMsg->arg[j].argBytes = htons(pExprMsg->arg[j].argBytes);
pExprMsg->param[j].nType = htons(pExprMsg->param[j].nType);
pExprMsg->param[j].nLen = htons(pExprMsg->param[j].nLen);
if (pExprMsg->arg[j].argType == TSDB_DATA_TYPE_BINARY) {
pExprMsg->arg[j].argValue.pz = pMsg;
pMsg += pExprMsg->arg[j].argBytes; // one more for the string terminated char.
if (pExprMsg->param[j].nType == TSDB_DATA_TYPE_BINARY) {
pExprMsg->param[j].pz = pMsg;
pMsg += pExprMsg->param[j].nLen; // one more for the string terminated char.
} else {
pExprMsg->arg[j].argValue.i64 = htobe64(pExprMsg->arg[j].argValue.i64);
pExprMsg->param[j].i64 = htobe64(pExprMsg->param[j].i64);
}
}
@ -5612,36 +5615,38 @@ int32_t convertQueryMsg(SQueryTableMsg *pQueryMsg, SQueryParam* param) {
}
}
pExprMsg = (SSqlFuncMsg *)pMsg;
pExprMsg = (SSqlExpr *)pMsg;
}
if (pQueryMsg->secondStageOutput) {
pExprMsg = (SSqlFuncMsg *)pMsg;
param->pSecExprMsg = calloc(pQueryMsg->secondStageOutput, POINTER_BYTES);
pExprMsg = (SSqlExpr *)pMsg;
param->pSecExpr = calloc(pQueryMsg->secondStageOutput, POINTER_BYTES);
for (int32_t i = 0; i < pQueryMsg->secondStageOutput; ++i) {
param->pSecExprMsg[i] = pExprMsg;
param->pSecExpr[i] = pExprMsg;
pExprMsg->colInfo.colIndex = htons(pExprMsg->colInfo.colIndex);
pExprMsg->colInfo.colId = htons(pExprMsg->colInfo.colId);
pExprMsg->colInfo.flag = htons(pExprMsg->colInfo.flag);
pExprMsg->colType = htons(pExprMsg->colType);
pExprMsg->resType = htons(pExprMsg->resType);
pExprMsg->resBytes = htons(pExprMsg->resBytes);
pExprMsg->colBytes = htons(pExprMsg->colBytes);
pExprMsg->colType = htons(pExprMsg->colType);
pExprMsg->functionId = htons(pExprMsg->functionId);
pExprMsg->numOfParams = htons(pExprMsg->numOfParams);
pMsg += sizeof(SSqlFuncMsg);
pMsg += sizeof(SSqlExpr);
for (int32_t j = 0; j < pExprMsg->numOfParams; ++j) {
pExprMsg->arg[j].argType = htons(pExprMsg->arg[j].argType);
pExprMsg->arg[j].argBytes = htons(pExprMsg->arg[j].argBytes);
pExprMsg->param[j].nType = htons(pExprMsg->param[j].nType);
pExprMsg->param[j].nLen = htons(pExprMsg->param[j].nLen);
if (pExprMsg->arg[j].argType == TSDB_DATA_TYPE_BINARY) {
pExprMsg->arg[j].argValue.pz = pMsg;
pMsg += pExprMsg->arg[j].argBytes; // one more for the string terminated char.
if (pExprMsg->param[j].nType == TSDB_DATA_TYPE_BINARY) {
pExprMsg->param[j].pz = pMsg;
pMsg += pExprMsg->param[j].nLen; // one more for the string terminated char.
} else {
pExprMsg->arg[j].argValue.i64 = htobe64(pExprMsg->arg[j].argValue.i64);
pExprMsg->param[j].i64 = htobe64(pExprMsg->param[j].i64);
}
}
@ -5653,7 +5658,7 @@ int32_t convertQueryMsg(SQueryTableMsg *pQueryMsg, SQueryParam* param) {
}
}
pExprMsg = (SSqlFuncMsg *)pMsg;
pExprMsg = (SSqlExpr *)pMsg;
}
}
@ -5757,7 +5762,8 @@ int32_t convertQueryMsg(SQueryTableMsg *pQueryMsg, SQueryParam* param) {
param->sql = strndup(pMsg, pQueryMsg->sqlstrLen);
if (!validateQuerySourceCols(pQueryMsg, param->pExprMsg, param->pTagColumnInfo)) {
SQueriedTableInfo info = { .numOfTags = pQueryMsg->numOfTags, .numOfCols = pQueryMsg->numOfCols, .colList = pQueryMsg->colList};
if (!validateQueryTableCols(&info, param->pExpr, pQueryMsg->numOfOutput, param->pTagColumnInfo, pQueryMsg)) {
code = TSDB_CODE_QRY_INVALID_MSG;
goto _cleanup;
}
@ -5781,15 +5787,15 @@ static int32_t buildArithmeticExprFromMsg(SExprInfo *pArithExprInfo, void *pQuer
tExprNode* pExprNode = NULL;
TRY(TSDB_MAX_TAG_CONDITIONS) {
pExprNode = exprTreeFromBinary(pArithExprInfo->base.arg[0].argValue.pz, pArithExprInfo->base.arg[0].argBytes);
pExprNode = exprTreeFromBinary(pArithExprInfo->base.param[0].pz, pArithExprInfo->base.param[0].nLen);
} CATCH( code ) {
CLEANUP_EXECUTE();
qError("qmsg:%p failed to create arithmetic expression string from:%s, reason: %s", pQueryMsg, pArithExprInfo->base.arg[0].argValue.pz, tstrerror(code));
qError("qmsg:%p failed to create arithmetic expression string from:%s, reason: %s", pQueryMsg, pArithExprInfo->base.param[0].pz, tstrerror(code));
return code;
} END_TRY
if (pExprNode == NULL) {
qError("qmsg:%p failed to create arithmetic expression string from:%s", pQueryMsg, pArithExprInfo->base.arg[0].argValue.pz);
qError("qmsg:%p failed to create arithmetic expression string from:%s", pQueryMsg, pArithExprInfo->base.param[0].pz);
return TSDB_CODE_QRY_APP_ERROR;
}
@ -5797,18 +5803,19 @@ static int32_t buildArithmeticExprFromMsg(SExprInfo *pArithExprInfo, void *pQuer
return TSDB_CODE_SUCCESS;
}
static int32_t updateOutputBufForTopBotQuery(SQueryTableMsg* pQueryMsg, SColumnInfo* pTagCols, SExprInfo* pExprs, int32_t numOfOutput, int32_t tagLen, bool superTable) {
static int32_t updateOutputBufForTopBotQuery(SQueriedTableInfo* pTableInfo, SColumnInfo* pTagCols, SExprInfo* pExprs, int32_t numOfOutput, int32_t tagLen, bool superTable) {
for (int32_t i = 0; i < numOfOutput; ++i) {
int16_t functId = pExprs[i].base.functionId;
if (functId == TSDB_FUNC_TOP || functId == TSDB_FUNC_BOTTOM) {
int32_t j = getColumnIndexInSource(pQueryMsg, &pExprs[i].base, pTagCols);
if (j < 0 || j >= pQueryMsg->numOfCols) {
int32_t j = getColumnIndexInSource(pTableInfo, &pExprs[i].base, pTagCols);
if (j < 0 || j >= pTableInfo->numOfCols) {
return TSDB_CODE_QRY_INVALID_MSG;
} else {
SColumnInfo* pCol = &pQueryMsg->colList[j];
int32_t ret = getResultDataInfo(pCol->type, pCol->bytes, functId, (int32_t)pExprs[i].base.arg[0].argValue.i64,
&pExprs[i].type, &pExprs[i].bytes, &pExprs[i].interBytes, tagLen, superTable);
SColumnInfo* pCol = &pTableInfo->colList[j];
int32_t ret = getResultDataInfo(pCol->type, pCol->bytes, functId, (int32_t)pExprs[i].base.param[0].i64,
&pExprs[i].base.resType, &pExprs[i].base.resBytes, &pExprs[i].base.interBytes, tagLen, superTable);
assert(ret == TSDB_CODE_SUCCESS);
}
}
@ -5818,16 +5825,8 @@ static int32_t updateOutputBufForTopBotQuery(SQueryTableMsg* pQueryMsg, SColumnI
}
// TODO tag length should be passed from client
typedef struct {
int32_t numOfOutput;
int32_t numOfTags;
int32_t numOfCols;
SColumnInfo* colList;
int32_t queryTest;
} SQueriedTableMeta;
int32_t createQueryFuncExprFromMsg(SQueryTableMsg* pQueryMsg, int32_t numOfOutput, SExprInfo** pExprInfo,
SSqlFuncMsg** pExprMsg, SColumnInfo* pTagCols, bool isSuperTable) {
int32_t createQueryFunc(SQueriedTableInfo* pTableInfo, int32_t numOfOutput, SExprInfo** pExprInfo,
SSqlExpr** pExprMsg, SColumnInfo* pTagCols, int32_t queryType, void* pMsg) {
*pExprInfo = NULL;
int32_t code = TSDB_CODE_SUCCESS;
@ -5836,19 +5835,18 @@ int32_t createQueryFuncExprFromMsg(SQueryTableMsg* pQueryMsg, int32_t numOfOutpu
return TSDB_CODE_QRY_OUT_OF_MEMORY;
}
// bool isSuperTable = QUERY_IS_STABLE_QUERY(pQueryMsg->queryType);
bool isSuperTable = QUERY_IS_STABLE_QUERY(queryType);
int16_t tagLen = 0;
for (int32_t i = 0; i < numOfOutput; ++i) {
pExprs[i].base = *pExprMsg[i];
pExprs[i].bytes = 0;
int16_t type = 0;
int16_t bytes = 0;
// parse the arithmetic expression
if (pExprs[i].base.functionId == TSDB_FUNC_ARITHM) {
code = buildArithmeticExprFromMsg(&pExprs[i], pQueryMsg);
code = buildArithmeticExprFromMsg(&pExprs[i], pMsg);
if (code != TSDB_CODE_SUCCESS) {
tfree(pExprs);
@ -5865,30 +5863,29 @@ int32_t createQueryFuncExprFromMsg(SQueryTableMsg* pQueryMsg, int32_t numOfOutpu
SSchema s = tGetBlockDistColumnSchema();
type = s.type;
bytes = s.bytes;
} else if (pExprs[i].base.colInfo.colId <= TSDB_UD_COLUMN_INDEX) {
} else if (pExprs[i].base.colInfo.colId <= TSDB_UD_COLUMN_INDEX && pExprs[i].base.colInfo.colId > TSDB_RES_COL_ID) {
// it is a user-defined constant value column
assert(pExprs[i].base.functionId == TSDB_FUNC_PRJ);
type = pExprs[i].base.arg[1].argType;
bytes = pExprs[i].base.arg[1].argBytes;
type = pExprs[i].base.param[1].nType;
bytes = pExprs[i].base.param[1].nLen;
if (type == TSDB_DATA_TYPE_BINARY || type == TSDB_DATA_TYPE_NCHAR) {
bytes += VARSTR_HEADER_SIZE;
}
} else {
int32_t j = getColumnIndexInSource(pQueryMsg, &pExprs[i].base, pTagCols);
int32_t j = getColumnIndexInSource(pTableInfo, &pExprs[i].base, pTagCols);
if (TSDB_COL_IS_TAG(pExprs[i].base.colInfo.flag)) {
if (j < TSDB_BLOCK_DIST_COLUMN_INDEX || j >= pQueryMsg->numOfTags) {
if (j < TSDB_BLOCK_DIST_COLUMN_INDEX || j >= pTableInfo->numOfTags) {
return TSDB_CODE_QRY_INVALID_MSG;
}
} else {
if (j < PRIMARYKEY_TIMESTAMP_COL_INDEX || j >= pQueryMsg->numOfCols) {
if (j < PRIMARYKEY_TIMESTAMP_COL_INDEX || j >= pTableInfo->numOfCols) {
return TSDB_CODE_QRY_INVALID_MSG;
}
}
if (pExprs[i].base.colInfo.colId != TSDB_TBNAME_COLUMN_INDEX && j >= 0) {
SColumnInfo* pCol = (TSDB_COL_IS_TAG(pExprs[i].base.colInfo.flag))? &pTagCols[j]:&pQueryMsg->colList[j];
SColumnInfo* pCol = (TSDB_COL_IS_TAG(pExprs[i].base.colInfo.flag))? &pTagCols[j]:&pTableInfo->colList[j];
type = pCol->type;
bytes = pCol->bytes;
} else {
@ -5899,35 +5896,35 @@ int32_t createQueryFuncExprFromMsg(SQueryTableMsg* pQueryMsg, int32_t numOfOutpu
}
}
int32_t param = (int32_t)pExprs[i].base.arg[0].argValue.i64;
int32_t param = (int32_t)pExprs[i].base.param[0].i64;
if (pExprs[i].base.functionId != TSDB_FUNC_ARITHM &&
(type != pExprs[i].base.colType || bytes != pExprs[i].base.colBytes)) {
(type != pExprs[i].base.resType || bytes != pExprs[i].base.resBytes)) {
tfree(pExprs);
return TSDB_CODE_QRY_INVALID_MSG;
}
if (getResultDataInfo(type, bytes, pExprs[i].base.functionId, param, &pExprs[i].type, &pExprs[i].bytes,
&pExprs[i].interBytes, 0, isSuperTable) != TSDB_CODE_SUCCESS) {
if (getResultDataInfo(type, bytes, pExprs[i].base.functionId, param, &pExprs[i].base.resType, &pExprs[i].base.resBytes,
&pExprs[i].base.interBytes, 0, isSuperTable) != TSDB_CODE_SUCCESS) {
tfree(pExprs);
return TSDB_CODE_QRY_INVALID_MSG;
}
if (pExprs[i].base.functionId == TSDB_FUNC_TAG_DUMMY || pExprs[i].base.functionId == TSDB_FUNC_TS_DUMMY) {
tagLen += pExprs[i].bytes;
tagLen += pExprs[i].base.resBytes;
}
assert(isValidDataType(pExprs[i].type));
assert(isValidDataType(pExprs[i].base.resType));
}
// the tag length is affected by other tag columns, so this should be update.
updateOutputBufForTopBotQuery(pQueryMsg, pTagCols, pExprs, numOfOutput, tagLen, isSuperTable);
updateOutputBufForTopBotQuery(pTableInfo, pTagCols, pExprs, numOfOutput, tagLen, isSuperTable);
*pExprInfo = pExprs;
return TSDB_CODE_SUCCESS;
}
int32_t createIndirectQueryFuncExprFromMsg(SQueryTableMsg *pQueryMsg, int32_t numOfOutput, SExprInfo **pExprInfo,
SSqlFuncMsg **pExprMsg, SExprInfo *prevExpr) {
int32_t createIndirectQueryFuncExprFromMsg(SQueryTableMsg* pQueryMsg, int32_t numOfOutput, SExprInfo** pExprInfo,
SSqlExpr** pExpr, SExprInfo* prevExpr) {
*pExprInfo = NULL;
int32_t code = TSDB_CODE_SUCCESS;
@ -5939,8 +5936,8 @@ int32_t createIndirectQueryFuncExprFromMsg(SQueryTableMsg *pQueryMsg, int32_t nu
bool isSuperTable = QUERY_IS_STABLE_QUERY(pQueryMsg->queryType);
for (int32_t i = 0; i < numOfOutput; ++i) {
pExprs[i].base = *pExprMsg[i];
pExprs[i].bytes = 0;
pExprs[i].base = *pExpr[i];
pExprs[i].base.resType = 0;
int16_t type = 0;
int16_t bytes = 0;
@ -5960,18 +5957,18 @@ int32_t createIndirectQueryFuncExprFromMsg(SQueryTableMsg *pQueryMsg, int32_t nu
int32_t index = pExprs[i].base.colInfo.colIndex;
assert(prevExpr[index].base.resColId == pExprs[i].base.colInfo.colId);
type = prevExpr[index].type;
bytes = prevExpr[index].bytes;
type = prevExpr[index].base.resType;
bytes = prevExpr[index].base.resBytes;
}
int32_t param = (int32_t)pExprs[i].base.arg[0].argValue.i64;
if (getResultDataInfo(type, bytes, pExprs[i].base.functionId, param, &pExprs[i].type, &pExprs[i].bytes,
&pExprs[i].interBytes, 0, isSuperTable) != TSDB_CODE_SUCCESS) {
int32_t param = (int32_t)pExprs[i].base.param[0].i64;
if (getResultDataInfo(type, bytes, pExprs[i].base.functionId, param, &pExprs[i].base.resType, &pExprs[i].base.resBytes,
&pExprs[i].base.interBytes, 0, isSuperTable) != TSDB_CODE_SUCCESS) {
tfree(pExprs);
return TSDB_CODE_QRY_INVALID_MSG;
}
assert(isValidDataType(pExprs[i].type));
assert(isValidDataType(pExprs[i].base.resType));
}
*pExprInfo = pExprs;
@ -6062,7 +6059,7 @@ static void doUpdateExprColumnIndex(SQuery *pQuery) {
assert(pQuery->pExpr1 != NULL && pQuery != NULL);
for (int32_t k = 0; k < pQuery->numOfOutput; ++k) {
SSqlFuncMsg *pSqlExprMsg = &pQuery->pExpr1[k].base;
SSqlExpr *pSqlExprMsg = &pQuery->pExpr1[k].base;
if (pSqlExprMsg->functionId == TSDB_FUNC_ARITHM) {
continue;
}
@ -6179,12 +6176,12 @@ SQInfo* createQInfoImpl(SQueryTableMsg* pQueryMsg, SSqlGroupbyExpr* pGroupbyExpr
// calculate the result row size
for (int16_t col = 0; col < numOfOutput; ++col) {
assert(pExprs[col].bytes > 0);
pQuery->resultRowSize += pExprs[col].bytes;
assert(pExprs[col].base.resBytes > 0);
pQuery->resultRowSize += pExprs[col].base.resBytes;
// keep the tag length
if (TSDB_COL_IS_TAG(pExprs[col].base.colInfo.flag)) {
pQuery->tagLen += pExprs[col].bytes;
pQuery->tagLen += pExprs[col].base.resBytes;
}
}

4
src/query/src/qUtil.c
View File

@ -34,7 +34,7 @@ int32_t getOutputInterResultBufSize(SQuery* pQuery) {
int32_t size = 0;
for (int32_t i = 0; i < pQuery->numOfOutput; ++i) {
size += pQuery->pExpr1[i].interBytes;
size += pQuery->pExpr1[i].base.interBytes;
}
assert(size >= 0);
@ -139,7 +139,7 @@ void clearResultRow(SQueryRuntimeEnv *pRuntimeEnv, SResultRow *pResultRow, int16
for (int32_t i = 0; i < pRuntimeEnv->pQuery->numOfOutput; ++i) {
SResultRowCellInfo *pResultInfo = &pResultRow->pCellInfo[i];
int16_t size = pRuntimeEnv->pQuery->pExpr1[i].bytes;
int16_t size = pRuntimeEnv->pQuery->pExpr1[i].base.resType;
char * s = getPosInResultPage(pRuntimeEnv->pQuery, page, pResultRow->offset, offset);
memset(s, 0, size);

15
src/query/src/queryMain.c
View File

@ -14,7 +14,6 @@
*/
#include "os.h"
#include "qFill.h"
#include "taosmsg.h"
#include "tcache.h"
#include "tglobal.h"
@ -23,13 +22,11 @@
#include "hash.h"
#include "texpr.h"
#include "qExecutor.h"
#include "qResultbuf.h"
#include "qUtil.h"
#include "query.h"
#include "queryLog.h"
#include "tlosertree.h"
#include "ttype.h"
#include "tcompare.h"
typedef struct SQueryMgmt {
pthread_mutex_t lock;
@ -58,8 +55,8 @@ void freeParam(SQueryParam *param) {
tfree(param->tagCond);
tfree(param->tbnameCond);
tfree(param->pTableIdList);
tfree(param->pExprMsg);
tfree(param->pSecExprMsg);
// tfree(param->pExprMsg);
// tfree(param->pSecExprMsg);
tfree(param->pExprs);
tfree(param->pSecExprs);
tfree(param->pGroupColIndex);
@ -91,12 +88,14 @@ int32_t qCreateQueryInfo(void* tsdb, int32_t vgId, SQueryTableMsg* pQueryMsg, qi
goto _over;
}
if ((code = createQueryFuncExprFromMsg(pQueryMsg, pQueryMsg->numOfOutput, &param.pExprs, param.pExprMsg, param.pTagColumnInfo)) != TSDB_CODE_SUCCESS) {
SQueriedTableInfo info = { .numOfTags = pQueryMsg->numOfTags, .numOfCols = pQueryMsg->numOfCols, .colList = pQueryMsg->colList};
if ((code = createQueryFunc(&info, pQueryMsg->numOfOutput, &param.pExprs, param.pExpr, param.pTagColumnInfo,
pQueryMsg->queryType, pQueryMsg)) != TSDB_CODE_SUCCESS) {
goto _over;
}
if (param.pSecExprMsg != NULL) {
if ((code = createIndirectQueryFuncExprFromMsg(pQueryMsg, pQueryMsg->secondStageOutput, &param.pSecExprs, param.pSecExprMsg, param.pExprs)) != TSDB_CODE_SUCCESS) {
if (param.pSecExpr != NULL) {
if ((code = createIndirectQueryFuncExprFromMsg(pQueryMsg, pQueryMsg->secondStageOutput, &param.pSecExprs, param.pSecExpr, param.pExprs)) != TSDB_CODE_SUCCESS) {
goto _over;
}
}

24
src/vnode/src/vnodeRead.c
View File

@ -252,7 +252,7 @@ static int32_t vnodeProcessQueryMsg(SVnodeObj *pVnode, SVReadMsg *pRead) {
SQueryTableRsp *pRsp = (SQueryTableRsp *)rpcMallocCont(sizeof(SQueryTableRsp));
pRsp->code = code;
pRsp->qhandle = 0;
pRsp->qid = 0;
pRet->len = sizeof(SQueryTableRsp);
pRet->rsp = pRsp;
@ -270,7 +270,7 @@ static int32_t vnodeProcessQueryMsg(SVnodeObj *pVnode, SVReadMsg *pRead) {
return pRsp->code;
} else {
assert(*handle == pQInfo);
pRsp->qhandle = htobe64(qId);
pRsp->qid = htobe64(qId);
}
if (handle != NULL &&
@ -343,37 +343,37 @@ static int32_t vnodeProcessQueryMsg(SVnodeObj *pVnode, SVReadMsg *pRead) {
}
static int32_t vnodeProcessFetchMsg(SVnodeObj *pVnode, SVReadMsg *pRead) {
void * pCont = pRead->pCont;
SRspRet *pRet = &pRead->rspRet;
void *pCont = pRead->pCont;
SRspRet *pRet = &pRead->rspRet;
SRetrieveTableMsg *pRetrieve = pCont;
pRetrieve->free = htons(pRetrieve->free);
pRetrieve->qhandle = htobe64(pRetrieve->qhandle);
pRetrieve->qid = htobe64(pRetrieve->qid);
vTrace("vgId:%d, QInfo:%" PRIu64 ", retrieve msg is disposed, free:%d, conn:%p", pVnode->vgId, pRetrieve->qhandle,
vTrace("vgId:%d, qid:%" PRIu64 ", retrieve msg is disposed, free:%d, conn:%p", pVnode->vgId, pRetrieve->qid,
pRetrieve->free, pRead->rpcHandle);
memset(pRet, 0, sizeof(SRspRet));
terrno = TSDB_CODE_SUCCESS;
int32_t code = TSDB_CODE_SUCCESS;
void ** handle = qAcquireQInfo(pVnode->qMgmt, pRetrieve->qhandle);
void **handle = qAcquireQInfo(pVnode->qMgmt, pRetrieve->qid);
if (handle == NULL) {
code = terrno;
terrno = TSDB_CODE_SUCCESS;
} else if (!checkQIdEqual(*handle, pRetrieve->qhandle)) {
} else if (!checkQIdEqual(*handle, pRetrieve->qid)) {
code = TSDB_CODE_QRY_INVALID_QHANDLE;
}
if (code != TSDB_CODE_SUCCESS) {
vError("vgId:%d, invalid handle in retrieving result, code:%s, QInfo:%" PRIu64, pVnode->vgId, tstrerror(code), pRetrieve->qhandle);
vError("vgId:%d, invalid handle in retrieving result, code:%s, QInfo:%" PRIu64, pVnode->vgId, tstrerror(code), pRetrieve->qid);
vnodeBuildNoResultQueryRsp(pRet);
return code;
}
// kill current query and free corresponding resources.
if (pRetrieve->free == 1) {
vWarn("vgId:%d, QInfo:%"PRIu64 "-%p, retrieve msg received to kill query and free qhandle", pVnode->vgId, pRetrieve->qhandle, *handle);
vWarn("vgId:%d, QInfo:%"PRIu64 "-%p, retrieve msg received to kill query and free qhandle", pVnode->vgId, pRetrieve->qid, *handle);
qKillQuery(*handle);
qReleaseQInfo(pVnode->qMgmt, (void **)&handle, true);
@ -384,7 +384,7 @@ static int32_t vnodeProcessFetchMsg(SVnodeObj *pVnode, SVReadMsg *pRead) {
// register the qhandle to connect to quit query immediate if connection is broken
if (vnodeNotifyCurrentQhandle(pRead->rpcHandle, *handle, pVnode->vgId) != TSDB_CODE_SUCCESS) {
vError("vgId:%d, QInfo:%"PRIu64 "-%p, retrieve discarded since link is broken, %p", pVnode->vgId, pRetrieve->qhandle, *handle, pRead->rpcHandle);
vError("vgId:%d, QInfo:%"PRIu64 "-%p, retrieve discarded since link is broken, %p", pVnode->vgId, pRetrieve->qid, *handle, pRead->rpcHandle);
code = TSDB_CODE_RPC_NETWORK_UNAVAIL;
qKillQuery(*handle);
qReleaseQInfo(pVnode->qMgmt, (void **)&handle, true);
@ -402,7 +402,7 @@ static int32_t vnodeProcessFetchMsg(SVnodeObj *pVnode, SVReadMsg *pRead) {
memset(pRet->rsp, 0, sizeof(SRetrieveTableRsp));
freeHandle = true;
} else { // result is not ready, return immediately
// Only effects in the non-blocking model
// Only affects the non-blocking model
if (!tsRetrieveBlockingModel) {
if (!buildRes) {
assert(pRead->rpcHandle != NULL);