Elgato_dark/TDengine
1
0
Fork 0
mirror of https://github.com/taosdata/TDengine synced 2026-05-24 10:09:01 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 4
TDengine/source/libs/executor/src/executil.c

5472 lines
185 KiB
C
Raw Permalink Blame History

/*
* Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
*
* This program is free software: you can use, redistribute, and/or modify
* it under the terms of the GNU Affero General Public License, version 3
* or later ("AGPL"), as published by the Free Software Foundation.
*
* This program is distributed in the hope that it will be useful, but WITHOUT
* ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
* FITNESS FOR A PARTICULAR PURPOSE.
*
* You should have received a copy of the GNU Affero General Public License
* along with this program. If not, see <http://www.gnu.org/licenses/>.
*/
#include "function.h"
#include "functionMgt.h"
#include "index.h"
#include "os.h"
#include "query.h"
#include "querynodes.h"
#include "taoserror.h"
#include "tarray.h"
#include "tcompare.h"
#include "tdatablock.h"
#include "thash.h"
#include "tmsg.h"
#include "ttime.h"
#include "executil.h"
#include "executorInt.h"
#include "querytask.h"
#include "storageapi.h"
#include "ttypes.h"
#include "tutil.h"
#include "tjson.h"
#include "trpc.h"
#include "filter.h"
#include "operator.h"
#include "tref.h"
#include "streamMsg.h"
typedef struct tagFilterAssist {
SHashObj* colHash;
int32_t index;
SArray* cInfoList;
int32_t code;
} tagFilterAssist;
typedef struct STransTagExprCtx {
int32_t code;
SMetaReader* pReader;
} STransTagExprCtx;
typedef enum {
FILTER_NO_LOGIC = 1,
FILTER_AND,
FILTER_OTHER,
} FilterCondType;
static int32_t optimizeTbnameInCondImpl(void* pVnode, SArray* list, SNode* pTagCond, SStorageAPI* pStoreAPI,
uint64_t suid);
static int64_t getLimit(const SNode* pLimit) {
return (NULL == pLimit || NULL == ((SLimitNode*)pLimit)->limit) ? -1 : ((SLimitNode*)pLimit)->limit->datum.i;
}
static int64_t getOffset(const SNode* pLimit) {
return (NULL == pLimit || NULL == ((SLimitNode*)pLimit)->offset) ? -1 : ((SLimitNode*)pLimit)->offset->datum.i;
}
static void releaseColInfoData(void* pCol);
int32_t sendFetchRemoteNodeReq(STaskSubJobCtx* ctx, int32_t subQIdx, SNode* pRes, bool reset);
void initResultRowInfo(SResultRowInfo* pResultRowInfo) {
pResultRowInfo->size = 0;
pResultRowInfo->cur.pageId = -1;
}
void closeResultRow(SResultRow* pResultRow) { pResultRow->closed = true; }
void resetResultRow(SResultRow* pResultRow, size_t entrySize) {
pResultRow->numOfRows = 0;
pResultRow->nOrigRows = 0;
pResultRow->closed = false;
pResultRow->endInterp = false;
pResultRow->startInterp = false;
if (entrySize > 0) {
memset(pResultRow->pEntryInfo, 0, entrySize);
}
}
// TODO refactor: use macro
SResultRowEntryInfo* getResultEntryInfo(const SResultRow* pRow, int32_t index, const int32_t* offset) {
return (SResultRowEntryInfo*)((char*)pRow->pEntryInfo + offset[index]);
}
size_t getResultRowSize(SqlFunctionCtx* pCtx, int32_t numOfOutput) {
int32_t rowSize = (numOfOutput * sizeof(SResultRowEntryInfo)) + sizeof(SResultRow);
for (int32_t i = 0; i < numOfOutput; ++i) {
rowSize += pCtx[i].resDataInfo.interBufSize;
}
return rowSize;
}
// Convert buf read from rocksdb to result row
int32_t getResultRowFromBuf(SExprSupp* pSup, const char* inBuf, size_t inBufSize, char** outBuf, size_t* outBufSize) {
if (inBuf == NULL || pSup == NULL) {
qError("invalid input parameters, inBuf:%p, pSup:%p", inBuf, pSup);
return TSDB_CODE_INVALID_PARA;
}
SqlFunctionCtx* pCtx = pSup->pCtx;
int32_t* offset = pSup->rowEntryInfoOffset;
SResultRow* pResultRow = NULL;
size_t processedSize = 0;
int32_t code = TSDB_CODE_SUCCESS;
// calculate the size of output buffer
*outBufSize = getResultRowSize(pCtx, pSup->numOfExprs);
*outBuf = taosMemoryMalloc(*outBufSize);
if (*outBuf == NULL) {
qError("failed to allocate memory for output buffer, size:%zu", *outBufSize);
return terrno;
}
pResultRow = (SResultRow*)*outBuf;
(void)memcpy(pResultRow, inBuf, sizeof(SResultRow));
inBuf += sizeof(SResultRow);
processedSize += sizeof(SResultRow);
for (int32_t i = 0; i < pSup->numOfExprs; ++i) {
int32_t len = *(int32_t*)inBuf;
inBuf += sizeof(int32_t);
processedSize += sizeof(int32_t);
if (pResultRow->version != FUNCTION_RESULT_INFO_VERSION && pCtx->fpSet.decode) {
code = pCtx->fpSet.decode(&pCtx[i], inBuf, getResultEntryInfo(pResultRow, i, offset), pResultRow->version);
if (code != TSDB_CODE_SUCCESS) {
qError("failed to decode result row, code:%d", code);
return code;
}
} else {
(void)memcpy(getResultEntryInfo(pResultRow, i, offset), inBuf, len);
}
inBuf += len;
processedSize += len;
}
if (processedSize < inBufSize) {
// stream stores extra data after result row
size_t leftLen = inBufSize - processedSize;
TAOS_MEMORY_REALLOC(*outBuf, *outBufSize + leftLen);
if (*outBuf == NULL) {
qError("failed to reallocate memory for output buffer, size:%zu", *outBufSize + leftLen);
return terrno;
}
(void)memcpy(*outBuf + *outBufSize, inBuf, leftLen);
inBuf += leftLen;
processedSize += leftLen;
*outBufSize += leftLen;
}
qTrace("[StreamInternal] get result inBufSize:%zu, outBufSize:%zu", inBufSize, *outBufSize);
return TSDB_CODE_SUCCESS;
}
// Convert result row to buf for rocksdb
int32_t putResultRowToBuf(SExprSupp* pSup, const char* inBuf, size_t inBufSize, char** outBuf, size_t* outBufSize) {
if (pSup == NULL || inBuf == NULL || outBuf == NULL || outBufSize == NULL) {
qError("invalid input parameters, inBuf:%p, pSup:%p, outBufSize:%p, outBuf:%p", inBuf, pSup, outBufSize, outBuf);
return TSDB_CODE_INVALID_PARA;
}
SqlFunctionCtx* pCtx = pSup->pCtx;
int32_t* offset = pSup->rowEntryInfoOffset;
SResultRow* pResultRow = (SResultRow*)inBuf;
size_t rowSize = getResultRowSize(pCtx, pSup->numOfExprs);
if (rowSize > inBufSize) {
qError("invalid input buffer size, rowSize:%zu, inBufSize:%zu", rowSize, inBufSize);
return TSDB_CODE_INVALID_PARA;
}
// calculate the size of output buffer
*outBufSize = rowSize + sizeof(int32_t) * pSup->numOfExprs;
if (rowSize < inBufSize) {
*outBufSize += inBufSize - rowSize;
}
*outBuf = taosMemoryMalloc(*outBufSize);
if (*outBuf == NULL) {
qError("failed to allocate memory for output buffer, size:%zu", *outBufSize);
return terrno;
}
char* pBuf = *outBuf;
pResultRow->version = FUNCTION_RESULT_INFO_VERSION;
(void)memcpy(pBuf, pResultRow, sizeof(SResultRow));
pBuf += sizeof(SResultRow);
for (int32_t i = 0; i < pSup->numOfExprs; ++i) {
size_t len = sizeof(SResultRowEntryInfo) + pCtx[i].resDataInfo.interBufSize;
*(int32_t*)pBuf = (int32_t)len;
pBuf += sizeof(int32_t);
(void)memcpy(pBuf, getResultEntryInfo(pResultRow, i, offset), len);
pBuf += len;
}
if (rowSize < inBufSize) {
// stream stores extra data after result row
size_t leftLen = inBufSize - rowSize;
(void)memcpy(pBuf, inBuf + rowSize, leftLen);
pBuf += leftLen;
}
qTrace("[StreamInternal] put result inBufSize:%zu, outBufSize:%zu", inBufSize, *outBufSize);
return TSDB_CODE_SUCCESS;
}
static void freeEx(void* p) { taosMemoryFree(*(void**)p); }
void cleanupGroupResInfo(SGroupResInfo* pGroupResInfo) {
taosMemoryFreeClear(pGroupResInfo->pBuf);
if (pGroupResInfo->freeItem) {
// taosArrayDestroy(pGroupResInfo->pRows);
taosArrayDestroyEx(pGroupResInfo->pRows, freeEx);
pGroupResInfo->freeItem = false;
pGroupResInfo->pRows = NULL;
} else {
taosArrayDestroy(pGroupResInfo->pRows);
pGroupResInfo->pRows = NULL;
}
pGroupResInfo->index = 0;
pGroupResInfo->delIndex = 0;
}
int32_t resultrowComparAsc(const void* p1, const void* p2) {
SResKeyPos* pp1 = *(SResKeyPos**)p1;
SResKeyPos* pp2 = *(SResKeyPos**)p2;
if (pp1->groupId == pp2->groupId) {
int64_t pts1 = *(int64_t*)pp1->key;
int64_t pts2 = *(int64_t*)pp2->key;
if (pts1 == pts2) {
return 0;
} else {
return pts1 < pts2 ? -1 : 1;
}
} else {
return pp1->groupId < pp2->groupId ? -1 : 1;
}
}
static int32_t resultrowComparDesc(const void* p1, const void* p2) { return resultrowComparAsc(p2, p1); }
int32_t initGroupedResultInfo(SGroupResInfo* pGroupResInfo, SSHashObj* pHashmap, int32_t order) {
int32_t code = TSDB_CODE_SUCCESS;
int32_t lino = 0;
if (pGroupResInfo->pRows != NULL) {
taosArrayDestroy(pGroupResInfo->pRows);
}
if (pGroupResInfo->pBuf) {
taosMemoryFree(pGroupResInfo->pBuf);
pGroupResInfo->pBuf = NULL;
}
// extract the result rows information from the hash map
int32_t size = tSimpleHashGetSize(pHashmap);
void* pData = NULL;
pGroupResInfo->pRows = taosArrayInit(size, POINTER_BYTES);
QUERY_CHECK_NULL(pGroupResInfo->pRows, code, lino, _end, terrno);
size_t keyLen = 0;
int32_t iter = 0;
int64_t bufLen = 0, offset = 0;
// todo move away and record this during create window
while ((pData = tSimpleHashIterate(pHashmap, pData, &iter)) != NULL) {
/*void* key = */ (void)tSimpleHashGetKey(pData, &keyLen);
bufLen += keyLen + sizeof(SResultRowPosition);
}
pGroupResInfo->pBuf = taosMemoryMalloc(bufLen);
QUERY_CHECK_NULL(pGroupResInfo->pBuf, code, lino, _end, terrno);
iter = 0;
while ((pData = tSimpleHashIterate(pHashmap, pData, &iter)) != NULL) {
void* key = tSimpleHashGetKey(pData, &keyLen);
SResKeyPos* p = (SResKeyPos*)(pGroupResInfo->pBuf + offset);
p->groupId = *(uint64_t*)key;
p->pos = *(SResultRowPosition*)pData;
memcpy(p->key, (char*)key + sizeof(uint64_t), keyLen - sizeof(uint64_t));
void* tmp = taosArrayPush(pGroupResInfo->pRows, &p);
QUERY_CHECK_NULL(pGroupResInfo->pBuf, code, lino, _end, terrno);
offset += keyLen + sizeof(struct SResultRowPosition);
}
if (order == TSDB_ORDER_ASC || order == TSDB_ORDER_DESC) {
__compar_fn_t fn = (order == TSDB_ORDER_ASC) ? resultrowComparAsc : resultrowComparDesc;
size = POINTER_BYTES;
taosSort(pGroupResInfo->pRows->pData, taosArrayGetSize(pGroupResInfo->pRows), size, fn);
}
pGroupResInfo->index = 0;
_end:
if (code != TSDB_CODE_SUCCESS) {
qError("%s failed at line %d since %s", __func__, lino, tstrerror(code));
}
return code;
}
void initMultiResInfoFromArrayList(SGroupResInfo* pGroupResInfo, SArray* pArrayList) {
if (pGroupResInfo->pRows != NULL) {
taosArrayDestroy(pGroupResInfo->pRows);
}
pGroupResInfo->freeItem = true;
pGroupResInfo->pRows = pArrayList;
pGroupResInfo->index = 0;
pGroupResInfo->delIndex = 0;
}
bool hasRemainResults(SGroupResInfo* pGroupResInfo) {
if (pGroupResInfo->pRows == NULL) {
return false;
}
return pGroupResInfo->index < taosArrayGetSize(pGroupResInfo->pRows);
}
int32_t getNumOfTotalRes(SGroupResInfo* pGroupResInfo) {
if (pGroupResInfo->pRows == 0) {
return 0;
}
return (int32_t)taosArrayGetSize(pGroupResInfo->pRows);
}
SArray* createSortInfo(SNodeList* pNodeList) {
size_t numOfCols = 0;
if (pNodeList != NULL) {
numOfCols = LIST_LENGTH(pNodeList);
} else {
numOfCols = 0;
}
SArray* pList = taosArrayInit(numOfCols, sizeof(SBlockOrderInfo));
if (pList == NULL) {
return pList;
}
for (int32_t i = 0; i < numOfCols; ++i) {
SOrderByExprNode* pSortKey = (SOrderByExprNode*)nodesListGetNode(pNodeList, i);
if (!pSortKey) {
qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(terrno));
taosArrayDestroy(pList);
pList = NULL;
terrno = TSDB_CODE_QRY_EXECUTOR_INTERNAL_ERROR;
break;
}
SBlockOrderInfo bi = {0};
bi.order = (pSortKey->order == ORDER_ASC) ? TSDB_ORDER_ASC : TSDB_ORDER_DESC;
bi.nullFirst = (pSortKey->nullOrder == NULL_ORDER_FIRST);
if (nodeType(pSortKey->pExpr) != QUERY_NODE_COLUMN) {
qError("invalid order by expr type:%d", nodeType(pSortKey->pExpr));
taosArrayDestroy(pList);
pList = NULL;
terrno = TSDB_CODE_QRY_EXECUTOR_INTERNAL_ERROR;
break;
}
SColumnNode* pColNode = (SColumnNode*)pSortKey->pExpr;
bi.slotId = pColNode->slotId;
void* tmp = taosArrayPush(pList, &bi);
if (!tmp) {
taosArrayDestroy(pList);
pList = NULL;
break;
}
}
return pList;
}
SSDataBlock* createDataBlockFromDescNode(void* p) {
SDataBlockDescNode* pNode = (SDataBlockDescNode*)p;
int32_t numOfCols = LIST_LENGTH(pNode->pSlots);
SSDataBlock* pBlock = NULL;
int32_t code = TSDB_CODE_SUCCESS;
int32_t lino = 0;
code = createDataBlock(&pBlock);
QUERY_CHECK_CODE(code, lino, _return);
pBlock->info.id.blockId = pNode->dataBlockId;
pBlock->info.calWin = (STimeWindow){.skey = INT64_MIN, .ekey = INT64_MAX};
pBlock->info.watermark = INT64_MIN;
int32_t i = 0;
SNode* node = NULL;
FOREACH(node, pNode->pSlots) {
SSlotDescNode* pDescNode = (SSlotDescNode*)node;
SColumnInfoData idata =
createColumnInfoData(pDescNode->dataType.type, pDescNode->dataType.bytes, pDescNode->slotId);
idata.info.scale = pDescNode->dataType.scale;
idata.info.precision = pDescNode->dataType.precision;
idata.info.noData = pDescNode->reserve;
code = blockDataAppendColInfo(pBlock, &idata);
QUERY_CHECK_CODE(code, lino, _return);
++i;
}
if (pBlock != NULL && i != numOfCols) {
code = TSDB_CODE_INVALID_PARA;
QUERY_CHECK_CODE(code, lino, _return);
}
return pBlock;
_return:
qError("%s failed at line %d since %s", __func__, lino, tstrerror(code));
blockDataDestroy(pBlock);
terrno = code;
return NULL;
}
int32_t prepareDataBlockBuf(SSDataBlock* pDataBlock, SColMatchInfo* pMatchInfo) {
SDataBlockInfo* pBlockInfo = &pDataBlock->info;
for (int32_t i = 0; i < taosArrayGetSize(pMatchInfo->pList); ++i) {
SColMatchItem* pItem = taosArrayGet(pMatchInfo->pList, i);
if (!pItem) {
qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(terrno));
return terrno;
}
if (pItem->isPk) {
SColumnInfoData* pInfoData = taosArrayGet(pDataBlock->pDataBlock, pItem->dstSlotId);
if (!pInfoData) {
qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(terrno));
return terrno;
}
pBlockInfo->pks[0].type = pInfoData->info.type;
pBlockInfo->pks[1].type = pInfoData->info.type;
// allocate enough buffer size, which is pInfoData->info.bytes
if (IS_VAR_DATA_TYPE(pItem->dataType.type)) {
pBlockInfo->pks[0].pData = taosMemoryCalloc(1, pInfoData->info.bytes);
if (pBlockInfo->pks[0].pData == NULL) {
return terrno;
}
pBlockInfo->pks[1].pData = taosMemoryCalloc(1, pInfoData->info.bytes);
if (pBlockInfo->pks[1].pData == NULL) {
taosMemoryFreeClear(pBlockInfo->pks[0].pData);
return terrno;
}
pBlockInfo->pks[0].nData = pInfoData->info.bytes;
pBlockInfo->pks[1].nData = pInfoData->info.bytes;
}
break;
}
}
return TSDB_CODE_SUCCESS;
}
EDealRes doTranslateTagExpr(SNode** pNode, void* pContext) {
STransTagExprCtx* pCtx = pContext;
SMetaReader* mr = pCtx->pReader;
bool isTagCol = false, isTbname = false;
if (nodeType(*pNode) == QUERY_NODE_COLUMN) {
SColumnNode* pCol = (SColumnNode*)*pNode;
if (pCol->colType == COLUMN_TYPE_TBNAME)
isTbname = true;
else
isTagCol = true;
} else if (nodeType(*pNode) == QUERY_NODE_FUNCTION) {
SFunctionNode* pFunc = (SFunctionNode*)*pNode;
if (pFunc->funcType == FUNCTION_TYPE_TBNAME) isTbname = true;
}
if (isTagCol) {
SColumnNode* pSColumnNode = *(SColumnNode**)pNode;
SValueNode* res = NULL;
pCtx->code = nodesMakeNode(QUERY_NODE_VALUE, (SNode**)&res);
if (NULL == res) {
return DEAL_RES_ERROR;
}
res->translate = true;
res->node.resType = pSColumnNode->node.resType;
STagVal tagVal = {0};
tagVal.cid = pSColumnNode->colId;
const char* p = mr->pAPI->extractTagVal(mr->me.ctbEntry.pTags, pSColumnNode->node.resType.type, &tagVal);
if (p == NULL) {
res->node.resType.type = TSDB_DATA_TYPE_NULL;
} else if (pSColumnNode->node.resType.type == TSDB_DATA_TYPE_JSON) {
int32_t len = ((const STag*)p)->len;
res->datum.p = taosMemoryCalloc(len + 1, 1);
if (NULL == res->datum.p) {
return DEAL_RES_ERROR;
}
memcpy(res->datum.p, p, len);
} else if (IS_VAR_DATA_TYPE(pSColumnNode->node.resType.type)) {
if (IS_STR_DATA_BLOB(pSColumnNode->node.resType.type)) {
return TSDB_CODE_BLOB_NOT_SUPPORT_TAG;
}
res->datum.p = taosMemoryCalloc(tagVal.nData + VARSTR_HEADER_SIZE + 1, 1);
if (NULL == res->datum.p) {
return DEAL_RES_ERROR;
}
if (IS_STR_DATA_BLOB(pSColumnNode->node.resType.type)) {
memcpy(blobDataVal(res->datum.p), tagVal.pData, tagVal.nData);
blobDataSetLen(res->datum.p, tagVal.nData);
} else {
memcpy(varDataVal(res->datum.p), tagVal.pData, tagVal.nData);
varDataSetLen(res->datum.p, tagVal.nData);
}
} else {
int32_t code = nodesSetValueNodeValue(res, &(tagVal.i64));
if (code != TSDB_CODE_SUCCESS) {
return DEAL_RES_ERROR;
}
}
nodesDestroyNode(*pNode);
*pNode = (SNode*)res;
} else if (isTbname) {
SValueNode* res = NULL;
pCtx->code = nodesMakeNode(QUERY_NODE_VALUE, (SNode**)&res);
if (NULL == res) {
return DEAL_RES_ERROR;
}
res->translate = true;
res->node.resType = ((SExprNode*)(*pNode))->resType;
int32_t len = strlen(mr->me.name);
res->datum.p = taosMemoryCalloc(len + VARSTR_HEADER_SIZE + 1, 1);
if (NULL == res->datum.p) {
return DEAL_RES_ERROR;
}
memcpy(varDataVal(res->datum.p), mr->me.name, len);
varDataSetLen(res->datum.p, len);
nodesDestroyNode(*pNode);
*pNode = (SNode*)res;
}
return DEAL_RES_CONTINUE;
}
int32_t isQualifiedTable(int64_t uid, SNode* pTagCond, void* vnode, bool* pQualified, SStorageAPI* pAPI) {
int32_t code = TSDB_CODE_SUCCESS;
SMetaReader mr = {0};
pAPI->metaReaderFn.initReader(&mr, vnode, META_READER_LOCK, &pAPI->metaFn);
code = pAPI->metaReaderFn.getEntryGetUidCache(&mr, uid);
if (TSDB_CODE_SUCCESS != code) {
pAPI->metaReaderFn.clearReader(&mr);
*pQualified = false;
return TSDB_CODE_SUCCESS;
}
SNode* pTagCondTmp = NULL;
code = nodesCloneNode(pTagCond, &pTagCondTmp);
if (TSDB_CODE_SUCCESS != code) {
*pQualified = false;
pAPI->metaReaderFn.clearReader(&mr);
return code;
}
STransTagExprCtx ctx = {.code = 0, .pReader = &mr};
nodesRewriteExprPostOrder(&pTagCondTmp, doTranslateTagExpr, &ctx);
pAPI->metaReaderFn.clearReader(&mr);
if (TSDB_CODE_SUCCESS != ctx.code) {
*pQualified = false;
nodesDestroyNode(pTagCondTmp);
terrno = code;
return code;
}
SNode* pNew = NULL;
code = scalarCalculateConstants(pTagCondTmp, &pNew);
if (TSDB_CODE_SUCCESS != code) {
terrno = code;
nodesDestroyNode(pTagCondTmp);
*pQualified = false;
return code;
}
SValueNode* pValue = (SValueNode*)pNew;
*pQualified = pValue->datum.b;
nodesDestroyNode(pNew);
return TSDB_CODE_SUCCESS;
}
static EDealRes getColumn(SNode** pNode, void* pContext) {
tagFilterAssist* pData = (tagFilterAssist*)pContext;
SColumnNode* pSColumnNode = NULL;
if (QUERY_NODE_COLUMN == nodeType((*pNode))) {
pSColumnNode = *(SColumnNode**)pNode;
} else if (QUERY_NODE_FUNCTION == nodeType((*pNode))) {
SFunctionNode* pFuncNode = *(SFunctionNode**)(pNode);
if (pFuncNode->funcType == FUNCTION_TYPE_TBNAME) {
pData->code = nodesMakeNode(QUERY_NODE_COLUMN, (SNode**)&pSColumnNode);
if (NULL == pSColumnNode) {
return DEAL_RES_ERROR;
}
pSColumnNode->colId = -1;
pSColumnNode->colType = COLUMN_TYPE_TBNAME;
pSColumnNode->node.resType.type = TSDB_DATA_TYPE_VARCHAR;
pSColumnNode->node.resType.bytes = TSDB_TABLE_FNAME_LEN - 1 + VARSTR_HEADER_SIZE;
nodesDestroyNode(*pNode);
*pNode = (SNode*)pSColumnNode;
} else {
return DEAL_RES_CONTINUE;
}
} else {
return DEAL_RES_CONTINUE;
}
void* data = taosHashGet(pData->colHash, &pSColumnNode->colId, sizeof(pSColumnNode->colId));
if (!data) {
int32_t tempRes =
taosHashPut(pData->colHash, &pSColumnNode->colId, sizeof(pSColumnNode->colId), pNode, sizeof((*pNode)));
if (tempRes != TSDB_CODE_SUCCESS && tempRes != TSDB_CODE_DUP_KEY) {
return DEAL_RES_ERROR;
}
pSColumnNode->slotId = pData->index++;
SColumnInfo cInfo = {.colId = pSColumnNode->colId,
.type = pSColumnNode->node.resType.type,
.bytes = pSColumnNode->node.resType.bytes,
.pk = pSColumnNode->isPk};
#if TAG_FILTER_DEBUG
qDebug("tagfilter build column info, slotId:%d, colId:%d, type:%d", pSColumnNode->slotId, cInfo.colId, cInfo.type);
#endif
void* tmp = taosArrayPush(pData->cInfoList, &cInfo);
if (!tmp) {
return DEAL_RES_ERROR;
}
} else {
SColumnNode* col = *(SColumnNode**)data;
pSColumnNode->slotId = col->slotId;
}
return DEAL_RES_CONTINUE;
}
static int32_t createResultData(SDataType* pType, int32_t numOfRows, SScalarParam* pParam) {
SColumnInfoData* pColumnData = taosMemoryCalloc(1, sizeof(SColumnInfoData));
if (pColumnData == NULL) {
return terrno;
}
pColumnData->info.type = pType->type;
pColumnData->info.bytes = pType->bytes;
pColumnData->info.scale = pType->scale;
pColumnData->info.precision = pType->precision;
int32_t code = colInfoDataEnsureCapacity(pColumnData, numOfRows, true);
if (code != TSDB_CODE_SUCCESS) {
terrno = code;
releaseColInfoData(pColumnData);
return terrno;
}
pParam->columnData = pColumnData;
pParam->colAlloced = true;
return TSDB_CODE_SUCCESS;
}
static void releaseColInfoData(void* pCol) {
if (pCol) {
SColumnInfoData* col = (SColumnInfoData*)pCol;
colDataDestroy(col);
taosMemoryFree(col);
}
}
void freeItem(void* p) {
STUidTagInfo* pInfo = p;
if (pInfo->pTagVal != NULL) {
taosMemoryFree(pInfo->pTagVal);
}
}
typedef struct {
col_id_t colId;
SNode* pValueNode;
int32_t bytes; // length defined in schema
} STagDataEntry;
static int compareTagDataEntry(const void* a, const void* b) {
STagDataEntry* p1 = (STagDataEntry*)a;
STagDataEntry* p2 = (STagDataEntry*)b;
return compareInt16Val(&p1->colId, &p2->colId);
}
static int32_t buildTagDataEntryKey(SArray* pIdWithValue, char** keyBuf, int32_t keyLen) {
*keyBuf = (char*)taosMemoryCalloc(1, keyLen);
if (NULL == *keyBuf) {
qError(
"failed to allocate memory for tag filter optimization key, size:%d",
keyLen);
return terrno;
}
char* pStart = *keyBuf;
for (int32_t i = 0; i < taosArrayGetSize(pIdWithValue); ++i) {
STagDataEntry* entry = (STagDataEntry*)taosArrayGet(pIdWithValue, i);
SValueNode* pValueNode = (SValueNode*)entry->pValueNode;
// num type may have different bytes length, use the smaller one
int32_t bytes = TMIN(entry->bytes, pValueNode->node.resType.bytes);
(void)memcpy(pStart, &entry->colId, sizeof(col_id_t));
pStart += sizeof(col_id_t);
if (!pValueNode->isNull) {
switch (pValueNode->node.resType.type) {
case TSDB_DATA_TYPE_BOOL:
(void)memcpy(
pStart, &pValueNode->datum.b, bytes);
pStart += bytes;
break;
case TSDB_DATA_TYPE_TINYINT:
case TSDB_DATA_TYPE_SMALLINT:
case TSDB_DATA_TYPE_INT:
case TSDB_DATA_TYPE_BIGINT:
case TSDB_DATA_TYPE_TIMESTAMP:
(void)memcpy(
pStart, &pValueNode->datum.i, bytes);
pStart += bytes;
break;
case TSDB_DATA_TYPE_UTINYINT:
case TSDB_DATA_TYPE_USMALLINT:
case TSDB_DATA_TYPE_UINT:
case TSDB_DATA_TYPE_UBIGINT:
(void)memcpy(
pStart, &pValueNode->datum.u, bytes);
pStart += bytes;
break;
case TSDB_DATA_TYPE_FLOAT:
case TSDB_DATA_TYPE_DOUBLE:
(void)memcpy(
pStart, &pValueNode->datum.d, bytes);
pStart += bytes;
break;
case TSDB_DATA_TYPE_VARCHAR:
case TSDB_DATA_TYPE_VARBINARY:
case TSDB_DATA_TYPE_NCHAR:
(void)memcpy(pStart,
varDataVal(pValueNode->datum.p), varDataLen(pValueNode->datum.p));
pStart += varDataLen(pValueNode->datum.p);
break;
default:
qError("unsupported tag data type %d in tag filter optimization",
pValueNode->node.resType.type);
return TSDB_CODE_STREAM_INTERNAL_ERROR;
}
}
}
return TSDB_CODE_SUCCESS;
}
static void extractTagDataEntry(
SOperatorNode* pOpNode, SArray* pIdWithValue) {
SNode* pLeft = pOpNode->pLeft;
SNode* pRight = pOpNode->pRight;
SColumnNode* pColNode = nodeType(pLeft) == QUERY_NODE_COLUMN ?
(SColumnNode*)pLeft : (SColumnNode*)pRight;
SValueNode* pValueNode = nodeType(pLeft) == QUERY_NODE_VALUE ?
(SValueNode*)pLeft : (SValueNode*)pRight;
STagDataEntry entry = {0};
entry.colId = pColNode->colId;
entry.pValueNode = (SNode*)pValueNode;
entry.bytes = pColNode->node.resType.bytes;
void* _tmp = taosArrayPush(pIdWithValue, &entry);
}
static int32_t extractTagFilterTagDataEntries(
const SNode* pTagCond, SArray* pIdWithVal) {
if (NULL == pTagCond || NULL == pIdWithVal ||
(nodeType(pTagCond) != QUERY_NODE_OPERATOR &&
nodeType(pTagCond) != QUERY_NODE_LOGIC_CONDITION)) {
qError("invalid parameter to extract tag filter symbol");
return TSDB_CODE_STREAM_INTERNAL_ERROR;
}
if (nodeType(pTagCond) == QUERY_NODE_OPERATOR) {
extractTagDataEntry((SOperatorNode*)pTagCond, pIdWithVal);
} else if (nodeType(pTagCond) == QUERY_NODE_LOGIC_CONDITION) {
SNode* pChild = NULL;
FOREACH(pChild, ((SLogicConditionNode*)pTagCond)->pParameterList) {
extractTagDataEntry((SOperatorNode*)pChild, pIdWithVal);
}
}
taosArraySort(pIdWithVal, compareTagDataEntry);
return TSDB_CODE_SUCCESS;
}
static int32_t genStableTagFilterDigest(const SNode* pTagCond, T_MD5_CTX* pContext) {
if (pTagCond == NULL) {
return TSDB_CODE_SUCCESS;
}
char* payload = NULL;
int32_t len = 0;
int32_t code = TSDB_CODE_SUCCESS;
int32_t lino = 0;
SArray* pIdWithVal = taosArrayInit(TARRAY_MIN_SIZE, sizeof(STagDataEntry));
code = extractTagFilterTagDataEntries(pTagCond, pIdWithVal);
QUERY_CHECK_CODE(code, lino, _end);
for (int32_t i = 0; i < taosArrayGetSize(pIdWithVal); ++i) {
STagDataEntry* pEntry = taosArrayGet(pIdWithVal, i);
len += sizeof(col_id_t) + pEntry->bytes;
}
code = buildTagDataEntryKey(pIdWithVal, &payload, len);
QUERY_CHECK_CODE(code, lino, _end);
tMD5Init(pContext);
tMD5Update(pContext, (uint8_t*)payload, (uint32_t)len);
tMD5Final(pContext);
_end:
if (TSDB_CODE_SUCCESS != code) {
qError("%s failed at line %d since %s",
__func__, __LINE__, tstrerror(code));
}
taosArrayDestroy(pIdWithVal);
taosMemoryFree(payload);
return code;
}
static int32_t genTagFilterDigest(const SNode* pTagCond, T_MD5_CTX* pContext) {
if (pTagCond == NULL) {
return TSDB_CODE_SUCCESS;
}
char* payload = NULL;
int32_t len = 0;
int32_t code = nodesNodeToMsg(pTagCond, &payload, &len);
if (code != TSDB_CODE_SUCCESS) {
qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(code));
return code;
}
tMD5Init(pContext);
tMD5Update(pContext, (uint8_t*)payload, (uint32_t)len);
tMD5Final(pContext);
// void* tmp = NULL;
// uint32_t size = 0;
// (void)taosAscii2Hex((const char*)pContext->digest, 16, &tmp, &size);
// qInfo("tag filter digest payload: %s", tmp);
// taosMemoryFree(tmp);
taosMemoryFree(payload);
return TSDB_CODE_SUCCESS;
}
static int32_t genTbGroupDigest(const SNode* pGroup, uint8_t* filterDigest, T_MD5_CTX* pContext) {
int32_t code = TSDB_CODE_SUCCESS;
int32_t lino = 0;
char* payload = NULL;
int32_t len = 0;
code = nodesNodeToMsg(pGroup, &payload, &len);
QUERY_CHECK_CODE(code, lino, _end);
if (filterDigest[0]) {
payload = taosMemoryRealloc(payload, len + tListLen(pContext->digest));
QUERY_CHECK_NULL(payload, code, lino, _end, terrno);
memcpy(payload + len, filterDigest + 1, tListLen(pContext->digest));
len += tListLen(pContext->digest);
}
tMD5Init(pContext);
tMD5Update(pContext, (uint8_t*)payload, (uint32_t)len);
tMD5Final(pContext);
_end:
taosMemoryFree(payload);
if (code != TSDB_CODE_SUCCESS) {
qError("%s failed at line %d since %s", __func__, lino, tstrerror(code));
}
return code;
}
int32_t qGetColumnsFromNodeList(void* data, bool isList, SArray** pColList) {
int32_t code = TSDB_CODE_SUCCESS;
tagFilterAssist ctx = {0};
ctx.colHash = taosHashInit(4, taosGetDefaultHashFunction(TSDB_DATA_TYPE_SMALLINT), false, HASH_NO_LOCK);
if (ctx.colHash == NULL) {
code = terrno;
goto end;
}
ctx.index = 0;
ctx.cInfoList = taosArrayInit(4, sizeof(SColumnInfo));
if (ctx.cInfoList == NULL) {
code = terrno;
goto end;
}
if (isList) {
SNode* pNode = NULL;
FOREACH(pNode, (SNodeList*)data) {
nodesRewriteExprPostOrder(&pNode, getColumn, (void*)&ctx);
if (TSDB_CODE_SUCCESS != ctx.code) {
code = ctx.code;
goto end;
}
REPLACE_NODE(pNode);
}
} else {
SNode* pNode = (SNode*)data;
nodesRewriteExprPostOrder(&pNode, getColumn, (void*)&ctx);
if (TSDB_CODE_SUCCESS != ctx.code) {
code = ctx.code;
goto end;
}
}
if (pColList != NULL) *pColList = ctx.cInfoList;
ctx.cInfoList = NULL;
end:
taosHashCleanup(ctx.colHash);
taosArrayDestroy(ctx.cInfoList);
return code;
}
static int32_t buildGroupInfo(SColumnInfoData* pValue, int32_t i, SArray* gInfo) {
int32_t code = TSDB_CODE_SUCCESS;
SStreamGroupValue* v = taosArrayReserve(gInfo, 1);
if (v == NULL) {
code = terrno;
goto end;
}
if (colDataIsNull_s(pValue, i)) {
v->isNull = true;
} else {
v->isNull = false;
char* data = colDataGetData(pValue, i);
if (pValue->info.type == TSDB_DATA_TYPE_JSON) {
if (tTagIsJson(data)) {
code = TSDB_CODE_QRY_JSON_IN_GROUP_ERROR;
goto end;
}
if (tTagIsJsonNull(data)) {
v->isNull = true;
goto end;
}
int32_t len = getJsonValueLen(data);
v->data.type = pValue->info.type;
v->data.nData = len;
v->data.pData = taosMemoryCalloc(1, len + 1);
if (v->data.pData == NULL) {
code = terrno;
goto end;
}
memcpy(v->data.pData, data, len);
qDebug("buildGroupInfo:%d add json data len:%d, data:%s", i, len, (char*)v->data.pData);
} else if (IS_VAR_DATA_TYPE(pValue->info.type)) {
if (varDataTLen(data) > pValue->info.bytes) {
code = TSDB_CODE_TDB_INVALID_TABLE_SCHEMA_VER;
goto end;
}
v->data.type = pValue->info.type;
v->data.nData = varDataLen(data);
v->data.pData = taosMemoryCalloc(1, varDataLen(data) + 1);
if (v->data.pData == NULL) {
code = terrno;
goto end;
}
memcpy(v->data.pData, varDataVal(data), varDataLen(data));
qDebug("buildGroupInfo:%d add var data type:%d, len:%d, data:%s", i, pValue->info.type, varDataLen(data), (char*)v->data.pData);
} else if (pValue->info.type == TSDB_DATA_TYPE_DECIMAL) { // reader todo decimal
v->data.type = pValue->info.type;
v->data.nData = pValue->info.bytes;
v->data.pData = taosMemoryCalloc(1, pValue->info.bytes);
if (v->data.pData == NULL) {
code = terrno;
goto end;
}
memcpy(&v->data.pData, data, pValue->info.bytes);
qDebug("buildGroupInfo:%d add data type:%d, data:%"PRId64, i, pValue->info.type, v->data.val);
} else { // reader todo decimal
v->data.type = pValue->info.type;
memcpy(&v->data.val, data, pValue->info.bytes);
qDebug("buildGroupInfo:%d add data type:%d, data:%"PRId64, i, pValue->info.type, v->data.val);
}
}
end:
if (code != TSDB_CODE_SUCCESS) {
qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(code));
v->isNull = true;
}
return code;
}
static void getColInfoResultForGroupbyForStream(void* pVnode, SNodeList* group, STableListInfo* pTableListInfo,
SStorageAPI* pAPI, SHashObj* groupIdMap) {
int32_t code = TSDB_CODE_SUCCESS;
int32_t lino = 0;
SArray* pBlockList = NULL;
SSDataBlock* pResBlock = NULL;
SArray* groupData = NULL;
SArray* pUidTagList = NULL;
SArray* gInfo = NULL;
int32_t tbNameIndex = 0;
int32_t rows = taosArrayGetSize(pTableListInfo->pTableList);
if (rows == 0) {
return;
}
pUidTagList = taosArrayInit(8, sizeof(STUidTagInfo));
QUERY_CHECK_NULL(pUidTagList, code, lino, end, terrno);
for (int32_t i = 0; i < rows; ++i) {
STableKeyInfo* pkeyInfo = taosArrayGet(pTableListInfo->pTableList, i);
QUERY_CHECK_NULL(pkeyInfo, code, lino, end, terrno);
STUidTagInfo info = {.uid = pkeyInfo->uid};
void* tmp = taosArrayPush(pUidTagList, &info);
QUERY_CHECK_NULL(tmp, code, lino, end, terrno);
}
if (taosArrayGetSize(pUidTagList) > 0) {
code = pAPI->metaFn.getTableTagsByUid(pVnode, pTableListInfo->idInfo.suid, pUidTagList);
} else {
code = pAPI->metaFn.getTableTags(pVnode, pTableListInfo->idInfo.suid, pUidTagList);
}
if (code != TSDB_CODE_SUCCESS) {
goto end;
}
SArray* pColList = NULL;
code = qGetColumnsFromNodeList(group, true, &pColList);
if (code != TSDB_CODE_SUCCESS) {
goto end;
}
for (int32_t i = 0; i < taosArrayGetSize(pColList); ++i) {
SColumnInfo* tmp = (SColumnInfo*)taosArrayGet(pColList, i);
if (tmp != NULL && tmp->colId == -1) {
tbNameIndex = i;
}
}
int32_t numOfTables = taosArrayGetSize(pUidTagList);
pResBlock = createTagValBlockForFilter(pColList, numOfTables, pUidTagList, pVnode, pAPI);
taosArrayDestroy(pColList);
if (pResBlock == NULL) {
code = terrno;
goto end;
}
pBlockList = taosArrayInit(2, POINTER_BYTES);
QUERY_CHECK_NULL(pBlockList, code, lino, end, terrno);
void* tmp = taosArrayPush(pBlockList, &pResBlock);
QUERY_CHECK_NULL(tmp, code, lino, end, terrno);
groupData = taosArrayInit(2, POINTER_BYTES);
QUERY_CHECK_NULL(groupData, code, lino, end, terrno);
SNode* pNode = NULL;
FOREACH(pNode, group) {
SScalarParam output = {0};
switch (nodeType(pNode)) {
case QUERY_NODE_VALUE:
break;
case QUERY_NODE_COLUMN:
case QUERY_NODE_OPERATOR:
case QUERY_NODE_FUNCTION: {
SExprNode* expNode = (SExprNode*)pNode;
code = createResultData(&expNode->resType, rows, &output);
if (code != TSDB_CODE_SUCCESS) {
goto end;
}
break;
}
default:
code = TSDB_CODE_OPS_NOT_SUPPORT;
goto end;
}
if (nodeType(pNode) == QUERY_NODE_COLUMN) {
SColumnNode* pSColumnNode = (SColumnNode*)pNode;
SColumnInfoData* pColInfo = (SColumnInfoData*)taosArrayGet(pResBlock->pDataBlock, pSColumnNode->slotId);
QUERY_CHECK_NULL(pColInfo, code, lino, end, terrno);
code = colDataAssign(output.columnData, pColInfo, rows, NULL);
} else if (nodeType(pNode) == QUERY_NODE_VALUE) {
continue;
} else {
gTaskScalarExtra.pStreamInfo = NULL;
gTaskScalarExtra.pStreamRange = NULL;
code = scalarCalculate(pNode, pBlockList, &output, &gTaskScalarExtra);
}
if (code != TSDB_CODE_SUCCESS) {
releaseColInfoData(output.columnData);
goto end;
}
void* tmp = taosArrayPush(groupData, &output.columnData);
QUERY_CHECK_NULL(tmp, code, lino, end, terrno);
}
for (int i = 0; i < rows; i++) {
gInfo = taosArrayInit(taosArrayGetSize(groupData), sizeof(SStreamGroupValue));
QUERY_CHECK_NULL(gInfo, code, lino, end, terrno);
STableKeyInfo* info = taosArrayGet(pTableListInfo->pTableList, i);
QUERY_CHECK_NULL(info, code, lino, end, terrno);
for (int j = 0; j < taosArrayGetSize(groupData); j++) {
SColumnInfoData* pValue = (SColumnInfoData*)taosArrayGetP(groupData, j);
int32_t ret = buildGroupInfo(pValue, i, gInfo);
if (ret != TSDB_CODE_SUCCESS) {
qError("buildGroupInfo failed at line %d since %s", __LINE__, tstrerror(ret));
goto end;
}
if (j == tbNameIndex) {
SStreamGroupValue* v = taosArrayGetLast(gInfo);
if (v != NULL){
v->isTbname = true;
v->uid = info->uid;
}
}
}
int32_t ret = taosHashPut(groupIdMap, &info->uid, sizeof(info->uid), &gInfo, POINTER_BYTES);
if (ret != TSDB_CODE_SUCCESS) {
qError("put groupid to map failed at line %d since %s", __LINE__, tstrerror(ret));
goto end;
}
qDebug("put groupid to map gid:%" PRIu64, info->uid);
gInfo = NULL;
}
end:
blockDataDestroy(pResBlock);
taosArrayDestroy(pBlockList);
taosArrayDestroyEx(pUidTagList, freeItem);
taosArrayDestroyP(groupData, releaseColInfoData);
taosArrayDestroyEx(gInfo, tDestroySStreamGroupValue);
if (code != TSDB_CODE_SUCCESS) {
qError("%s failed at line %d since %s", __func__, lino, tstrerror(code));
}
}
int32_t getColInfoResultForGroupby(SReadHandle* pHandle, SNodeList* group, STableListInfo* pTableListInfo, uint8_t* digest,
SStorageAPI* pAPI, bool initRemainGroups, SHashObj* groupIdMap, bool gIdFromBaseId) {
int32_t code = TSDB_CODE_SUCCESS;
int32_t lino = 0;
SArray* pBlockList = NULL;
SSDataBlock* pResBlock = NULL;
void* keyBuf = NULL;
SArray* groupData = NULL;
SArray* pUidTagList = NULL;
SArray* tableList = NULL;
SArray* gInfo = NULL;
void* pVnode = pHandle->vnode;
int32_t rows = taosArrayGetSize(pTableListInfo->pTableList);
if (rows == 0) {
return TSDB_CODE_SUCCESS;
}
T_MD5_CTX context = {0};
if (tsTagFilterCache && groupIdMap == NULL && digest) {
SNodeListNode* listNode = NULL;
code = nodesMakeNode(QUERY_NODE_NODE_LIST, (SNode**)&listNode);
if (TSDB_CODE_SUCCESS != code) {
goto end;
}
listNode->pNodeList = group;
code = genTbGroupDigest((SNode*)listNode, digest, &context);
QUERY_CHECK_CODE(code, lino, end);
nodesFree(listNode);
code = pAPI->metaFn.metaGetCachedTbGroup(pVnode, pTableListInfo->idInfo.suid, context.digest,
tListLen(context.digest), &tableList);
QUERY_CHECK_CODE(code, lino, end);
if (tableList) {
taosArrayDestroy(pTableListInfo->pTableList);
pTableListInfo->pTableList = tableList;
qDebug("retrieve tb group list from cache, numOfTables:%d",
(int32_t)taosArrayGetSize(pTableListInfo->pTableList));
goto end;
}
}
pUidTagList = taosArrayInit(8, sizeof(STUidTagInfo));
QUERY_CHECK_NULL(pUidTagList, code, lino, end, terrno);
for (int32_t i = 0; i < rows; ++i) {
STableKeyInfo* pkeyInfo = taosArrayGet(pTableListInfo->pTableList, i);
QUERY_CHECK_NULL(pkeyInfo, code, lino, end, terrno);
STUidTagInfo info = {.uid = pkeyInfo->uid};
void* tmp = taosArrayPush(pUidTagList, &info);
QUERY_CHECK_NULL(tmp, code, lino, end, terrno);
}
if (taosArrayGetSize(pUidTagList) > 0) {
code = pAPI->metaFn.getTableTagsByUid(pVnode, pTableListInfo->idInfo.suid, pUidTagList);
} else {
code = pAPI->metaFn.getTableTags(pVnode, pTableListInfo->idInfo.suid, pUidTagList);
}
if (code != TSDB_CODE_SUCCESS) {
goto end;
}
SArray* pColList = NULL;
code = qGetColumnsFromNodeList(group, true, &pColList);
if (code != TSDB_CODE_SUCCESS) {
goto end;
}
int32_t numOfTables = taosArrayGetSize(pUidTagList);
pResBlock = createTagValBlockForFilter(pColList, numOfTables, pUidTagList, pVnode, pAPI);
taosArrayDestroy(pColList);
if (pResBlock == NULL) {
code = terrno;
goto end;
}
// int64_t st1 = taosGetTimestampUs();
// qDebug("generate tag block rows:%d, cost:%ld us", rows, st1-st);
pBlockList = taosArrayInit(2, POINTER_BYTES);
QUERY_CHECK_NULL(pBlockList, code, lino, end, terrno);
void* tmp = taosArrayPush(pBlockList, &pResBlock);
QUERY_CHECK_NULL(tmp, code, lino, end, terrno);
groupData = taosArrayInit(2, POINTER_BYTES);
QUERY_CHECK_NULL(groupData, code, lino, end, terrno);
SNode* pNode = NULL;
FOREACH(pNode, group) {
SScalarParam output = {0};
switch (nodeType(pNode)) {
case QUERY_NODE_VALUE:
break;
case QUERY_NODE_COLUMN:
case QUERY_NODE_OPERATOR:
case QUERY_NODE_FUNCTION: {
SExprNode* expNode = (SExprNode*)pNode;
code = createResultData(&expNode->resType, rows, &output);
if (code != TSDB_CODE_SUCCESS) {
goto end;
}
break;
}
case QUERY_NODE_REMOTE_VALUE: {
SRemoteValueNode* pRemote = (SRemoteValueNode*)pNode;
code = qFetchRemoteNode(gTaskScalarExtra.pSubJobCtx, pRemote->subQIdx, pNode);
QUERY_CHECK_CODE(code, lino, end);
break;
}
default:
code = TSDB_CODE_OPS_NOT_SUPPORT;
goto end;
}
if (nodeType(pNode) == QUERY_NODE_COLUMN) {
SColumnNode* pSColumnNode = (SColumnNode*)pNode;
SColumnInfoData* pColInfo = (SColumnInfoData*)taosArrayGet(pResBlock->pDataBlock, pSColumnNode->slotId);
QUERY_CHECK_NULL(pColInfo, code, lino, end, terrno);
code = colDataAssign(output.columnData, pColInfo, rows, NULL);
} else if (nodeType(pNode) == QUERY_NODE_VALUE) {
continue;
} else {
gTaskScalarExtra.pStreamInfo = NULL;
gTaskScalarExtra.pStreamRange = NULL;
code = scalarCalculate(pNode, pBlockList, &output, &gTaskScalarExtra);
}
if (code != TSDB_CODE_SUCCESS) {
releaseColInfoData(output.columnData);
goto end;
}
void* tmp = taosArrayPush(groupData, &output.columnData);
QUERY_CHECK_NULL(tmp, code, lino, end, terrno);
}
int32_t baseIdLen = gIdFromBaseId ? sizeof(int64_t) : 0;
int32_t keyLen = baseIdLen;
SNode* node;
FOREACH(node, group) {
SExprNode* pExpr = (SExprNode*)node;
keyLen += pExpr->resType.bytes;
}
int32_t nullFlagSize = sizeof(int8_t) * LIST_LENGTH(group);
keyLen += nullFlagSize;
keyBuf = taosMemoryCalloc(1, keyLen);
if (keyBuf == NULL) {
code = terrno;
goto end;
}
if (initRemainGroups) {
pTableListInfo->remainGroups =
taosHashInit(rows, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BIGINT), false, HASH_NO_LOCK);
if (pTableListInfo->remainGroups == NULL) {
code = terrno;
goto end;
}
}
for (int i = 0; i < rows; i++) {
STableKeyInfo* info = taosArrayGet(pTableListInfo->pTableList, i);
QUERY_CHECK_NULL(info, code, lino, end, terrno);
if (groupIdMap != NULL){
gInfo = taosArrayInit(taosArrayGetSize(groupData), sizeof(SStreamGroupValue));
}
if (baseIdLen > 0) {
*(uint64_t*)keyBuf = info->baseGId;
}
char* isNull = (char*)keyBuf + baseIdLen;
char* pStart = (char*)keyBuf + baseIdLen + sizeof(int8_t) * LIST_LENGTH(group);
for (int j = 0; j < taosArrayGetSize(groupData); j++) {
SColumnInfoData* pValue = (SColumnInfoData*)taosArrayGetP(groupData, j);
if (groupIdMap != NULL && gInfo != NULL) {
int32_t ret = buildGroupInfo(pValue, i, gInfo);
if (ret != TSDB_CODE_SUCCESS) {
qError("buildGroupInfo failed at line %d since %s", __LINE__, tstrerror(ret));
taosArrayDestroyEx(gInfo, tDestroySStreamGroupValue);
gInfo = NULL;
}
}
if (colDataIsNull_s(pValue, i)) {
isNull[j] = 1;
} else {
isNull[j] = 0;
char* data = colDataGetData(pValue, i);
if (pValue->info.type == TSDB_DATA_TYPE_JSON) {
// if (tTagIsJson(data)) {
// code = TSDB_CODE_QRY_JSON_IN_GROUP_ERROR;
// goto end;
// }
if (tTagIsJsonNull(data)) {
isNull[j] = 1;
continue;
}
int32_t len = getJsonValueLen(data);
memcpy(pStart, data, len);
pStart += len;
} else if (IS_VAR_DATA_TYPE(pValue->info.type)) {
if (IS_STR_DATA_BLOB(pValue->info.type)) {
if (blobDataTLen(data) > TSDB_MAX_BLOB_LEN) {
code = TSDB_CODE_TDB_INVALID_TABLE_SCHEMA_VER;
goto end;
}
memcpy(pStart, data, blobDataTLen(data));
pStart += blobDataTLen(data);
} else {
if (varDataTLen(data) > pValue->info.bytes) {
code = TSDB_CODE_TDB_INVALID_TABLE_SCHEMA_VER;
goto end;
}
memcpy(pStart, data, varDataTLen(data));
pStart += varDataTLen(data);
}
} else {
memcpy(pStart, data, pValue->info.bytes);
pStart += pValue->info.bytes;
}
}
}
int32_t len = (int32_t)(pStart - (char*)keyBuf);
info->groupId = calcGroupId(keyBuf, len);
if (groupIdMap != NULL && gInfo != NULL) {
int32_t ret = taosHashPut(groupIdMap, &info->groupId, sizeof(info->groupId), &gInfo, POINTER_BYTES);
if (ret != TSDB_CODE_SUCCESS) {
qError("put groupid to map failed at line %d since %s", __LINE__, tstrerror(ret));
taosArrayDestroyEx(gInfo, tDestroySStreamGroupValue);
}
qDebug("put groupid to map gid:%" PRIu64, info->groupId);
gInfo = NULL;
}
if (initRemainGroups) {
// groupId ~ table uid
code = taosHashPut(pTableListInfo->remainGroups, &(info->groupId), sizeof(info->groupId), &(info->uid),
sizeof(info->uid));
if (code == TSDB_CODE_DUP_KEY) {
code = TSDB_CODE_SUCCESS;
}
QUERY_CHECK_CODE(code, lino, end);
}
}
if (tsTagFilterCache && groupIdMap == NULL && digest) {
tableList = taosArrayDup(pTableListInfo->pTableList, NULL);
QUERY_CHECK_NULL(tableList, code, lino, end, terrno);
code = pAPI->metaFn.metaPutTbGroupToCache(pVnode, pTableListInfo->idInfo.suid, context.digest,
tListLen(context.digest), tableList,
taosArrayGetSize(tableList) * sizeof(STableKeyInfo));
QUERY_CHECK_CODE(code, lino, end);
}
// int64_t st2 = taosGetTimestampUs();
// qDebug("calculate tag block rows:%d, cost:%ld us", rows, st2-st1);
end:
taosMemoryFreeClear(keyBuf);
blockDataDestroy(pResBlock);
taosArrayDestroy(pBlockList);
taosArrayDestroyEx(pUidTagList, freeItem);
taosArrayDestroyP(groupData, releaseColInfoData);
taosArrayDestroyEx(gInfo, tDestroySStreamGroupValue);
if (code != TSDB_CODE_SUCCESS) {
qError("%s failed at line %d since %s", __func__, lino, tstrerror(code));
}
return code;
}
static int32_t nameComparFn(const void* p1, const void* p2) {
const char* pName1 = *(const char**)p1;
const char* pName2 = *(const char**)p2;
int32_t ret = strcmp(pName1, pName2);
if (ret == 0) {
return 0;
} else {
return (ret > 0) ? 1 : -1;
}
}
static SArray* getTableNameList(const SNodeListNode* pList) {
int32_t code = TSDB_CODE_SUCCESS;
int32_t lino = 0;
int32_t len = LIST_LENGTH(pList->pNodeList);
SListCell* cell = pList->pNodeList->pHead;
SArray* pTbList = taosArrayInit(len, POINTER_BYTES);
QUERY_CHECK_NULL(pTbList, code, lino, _end, terrno);
for (int i = 0; i < pList->pNodeList->length; i++) {
SValueNode* valueNode = (SValueNode*)cell->pNode;
if (!IS_VAR_DATA_TYPE(valueNode->node.resType.type)) {
terrno = TSDB_CODE_INVALID_PARA;
taosArrayDestroy(pTbList);
return NULL;
}
char* name = varDataVal(valueNode->datum.p);
void* tmp = taosArrayPush(pTbList, &name);
QUERY_CHECK_NULL(tmp, code, lino, _end, terrno);
cell = cell->pNext;
}
size_t numOfTables = taosArrayGetSize(pTbList);
// order the name
taosArraySort(pTbList, nameComparFn);
// remove the duplicates
SArray* pNewList = taosArrayInit(taosArrayGetSize(pTbList), sizeof(void*));
QUERY_CHECK_NULL(pNewList, code, lino, _end, terrno);
void* tmpTbl = taosArrayGet(pTbList, 0);
QUERY_CHECK_NULL(tmpTbl, code, lino, _end, terrno);
void* tmp = taosArrayPush(pNewList, tmpTbl);
QUERY_CHECK_NULL(tmp, code, lino, _end, terrno);
for (int32_t i = 1; i < numOfTables; ++i) {
char** name = taosArrayGetLast(pNewList);
char** nameInOldList = taosArrayGet(pTbList, i);
QUERY_CHECK_NULL(nameInOldList, code, lino, _end, terrno);
if (strcmp(*name, *nameInOldList) == 0) {
continue;
}
tmp = taosArrayPush(pNewList, nameInOldList);
QUERY_CHECK_NULL(tmp, code, lino, _end, terrno);
}
_end:
taosArrayDestroy(pTbList);
if (code != TSDB_CODE_SUCCESS) {
qError("%s failed at line %d since %s", __func__, lino, tstrerror(code));
return NULL;
}
return pNewList;
}
static int tableUidCompare(const void* a, const void* b) {
uint64_t u1 = *(uint64_t*)a;
uint64_t u2 = *(uint64_t*)b;
if (u1 == u2) {
return 0;
}
return u1 < u2 ? -1 : 1;
}
static int32_t filterTableInfoCompare(const void* a, const void* b) {
STUidTagInfo* p1 = (STUidTagInfo*)a;
STUidTagInfo* p2 = (STUidTagInfo*)b;
if (p1->uid == p2->uid) {
return 0;
}
return p1->uid < p2->uid ? -1 : 1;
}
static FilterCondType checkTagCond(SNode* cond) {
if (nodeType(cond) == QUERY_NODE_OPERATOR) {
return FILTER_NO_LOGIC;
}
if (nodeType(cond) == QUERY_NODE_LOGIC_CONDITION && ((SLogicConditionNode*)cond)->condType == LOGIC_COND_TYPE_AND) {
return FILTER_AND;
}
return FILTER_OTHER;
}
static int32_t doInWithAnd(SNode* cond, void* pVnode, SArray* list, SStorageAPI* pAPI, uint64_t suid) {
if (nodeType(cond) != QUERY_NODE_LOGIC_CONDITION || ((SLogicConditionNode*)cond)->condType != LOGIC_COND_TYPE_AND) {
return -1;
}
SLogicConditionNode* pNode = (SLogicConditionNode*)cond;
SNodeList* pList = (SNodeList*)pNode->pParameterList;
int32_t len = LIST_LENGTH(pList);
if (len <= 0) {
return -1;
}
SListCell* cell = pList->pHead;
for (int i = 0; i < len; i++) {
if (cell == NULL) break;
if (nodeType(cell->pNode) == QUERY_NODE_OPERATOR && optimizeTbnameInCondImpl(pVnode, list, cell->pNode, pAPI, suid) == 0) {
return 0;
}
cell = cell->pNext;
}
return -1;
}
static int32_t optimizeTbnameInCond(void* pVnode, int64_t suid, SArray* list, SNode* cond, SStorageAPI* pAPI) {
int32_t code = 0;
int32_t lino = 0;
if (nodeType(cond) == QUERY_NODE_OPERATOR) {
code = optimizeTbnameInCondImpl(pVnode, list, cond, pAPI, suid);
} else {
code = doInWithAnd(cond, pVnode, list, pAPI, suid);
QUERY_CHECK_CODE(code, lino, end);
code = pAPI->metaFn.getTableTagsByUid(pVnode, suid, list);
}
end:
return code;
}
static int32_t getTableListInInOperator(void* pVnode, SArray* pExistedUidList, SNodeListNode* pList, SStorageAPI* pStoreAPI,
uint64_t suid) {
int32_t code = 0;
SArray* pTbList = getTableNameList(pList);
int32_t numOfTables = taosArrayGetSize(pTbList);
SHashObj* uHash = NULL;
size_t numOfExisted = taosArrayGetSize(pExistedUidList); // len > 0 means there already have uids
if (numOfExisted > 0) {
uHash = taosHashInit(numOfExisted / 0.7, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BIGINT), false, HASH_NO_LOCK);
if (!uHash) {
qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(terrno));
code = terrno;
goto end;
}
for (int i = 0; i < numOfExisted; i++) {
STUidTagInfo* pTInfo = taosArrayGet(pExistedUidList, i);
if (!pTInfo) {
qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(terrno));
code = terrno;
goto end;
}
code = taosHashPut(uHash, &pTInfo->uid, sizeof(uint64_t), &i, sizeof(i));
if (code != TSDB_CODE_SUCCESS && code != TSDB_CODE_DUP_KEY) {
qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(code));
goto end;
}
}
}
taosArrayClear(pExistedUidList);
for (int i = 0; i < numOfTables; i++) {
char* name = taosArrayGetP(pTbList, i);
uint64_t uid = 0, csuid = 0;
if (pStoreAPI->metaFn.getTableUidByName(pVnode, name, &uid) != 0) {
continue;
}
ETableType tbType = TSDB_TABLE_MAX;
if (pStoreAPI->metaFn.getTableTypeSuidByName(pVnode, name, &tbType, &csuid) == 0 &&
(tbType == TSDB_CHILD_TABLE || tbType == TSDB_VIRTUAL_CHILD_TABLE) &&
csuid == suid) {
if (NULL == uHash || taosHashGet(uHash, &uid, sizeof(uid)) != NULL) {
STUidTagInfo s = {.uid = uid, .name = name, .pTagVal = NULL};
void* tmp = taosArrayPush(pExistedUidList, &s);
if (!tmp) {
code = terrno;
goto end;
}
}
}
}
taosArraySort(pExistedUidList, filterTableInfoCompare);
taosArrayRemoveDuplicate(pExistedUidList, filterTableInfoCompare, NULL);
end:
taosArrayDestroy(pTbList);
taosHashCleanup(uHash);
return code;
}
// only return uid that does not contained in pExistedUidList
static int32_t optimizeTbnameInCondImpl(void* pVnode, SArray* pExistedUidList, SNode* pTagCond, SStorageAPI* pStoreAPI,
uint64_t suid) {
SOperatorNode* pNode = (SOperatorNode*)pTagCond;
if (pNode->opType != OP_TYPE_IN) {
return -1;
}
if ((pNode->pLeft != NULL && ((nodeType(pNode->pLeft) == QUERY_NODE_FUNCTION &&
((SFunctionNode*)pNode->pLeft)->funcType == FUNCTION_TYPE_TBNAME)) ||
(nodeType(pNode->pLeft) == QUERY_NODE_COLUMN && ((SColumnNode*)pNode->pLeft)->colType == COLUMN_TYPE_TBNAME)) &&
(pNode->pRight != NULL && nodeType(pNode->pRight) == QUERY_NODE_NODE_LIST)) {
SNodeListNode* pList = (SNodeListNode*)pNode->pRight;
int32_t len = LIST_LENGTH(pList->pNodeList);
if (len <= 0) {
return -1;
}
return getTableListInInOperator(pVnode, pExistedUidList, pList, pStoreAPI, suid);
}
return -1;
}
SSDataBlock* createTagValBlockForFilter(SArray* pColList, int32_t numOfTables, SArray* pUidTagList, void* pVnode,
SStorageAPI* pStorageAPI) {
int32_t code = TSDB_CODE_SUCCESS;
int32_t lino = 0;
SSDataBlock* pResBlock = NULL;
code = createDataBlock(&pResBlock);
QUERY_CHECK_CODE(code, lino, _end);
for (int32_t i = 0; i < taosArrayGetSize(pColList); ++i) {
SColumnInfoData colInfo = {0};
void* tmp = taosArrayGet(pColList, i);
QUERY_CHECK_NULL(tmp, code, lino, _end, terrno);
colInfo.info = *(SColumnInfo*)tmp;
code = blockDataAppendColInfo(pResBlock, &colInfo);
QUERY_CHECK_CODE(code, lino, _end);
}
code = blockDataEnsureCapacity(pResBlock, numOfTables);
if (code != TSDB_CODE_SUCCESS) {
terrno = code;
blockDataDestroy(pResBlock);
return NULL;
}
pResBlock->info.rows = numOfTables;
int32_t numOfCols = taosArrayGetSize(pResBlock->pDataBlock);
for (int32_t i = 0; i < numOfTables; i++) {
STUidTagInfo* p1 = taosArrayGet(pUidTagList, i);
QUERY_CHECK_NULL(p1, code, lino, _end, terrno);
for (int32_t j = 0; j < numOfCols; j++) {
SColumnInfoData* pColInfo = (SColumnInfoData*)taosArrayGet(pResBlock->pDataBlock, j);
QUERY_CHECK_NULL(pColInfo, code, lino, _end, terrno);
if (pColInfo->info.colId == -1) { // tbname
char str[TSDB_TABLE_FNAME_LEN + VARSTR_HEADER_SIZE] = {0};
if (p1->name != NULL) {
STR_TO_VARSTR(str, p1->name);
} else { // name is not retrieved during filter
code = pStorageAPI->metaFn.getTableNameByUid(pVnode, p1->uid, str);
QUERY_CHECK_CODE(code, lino, _end);
}
code = colDataSetVal(pColInfo, i, str, false);
QUERY_CHECK_CODE(code, lino, _end);
#if TAG_FILTER_DEBUG
qDebug("tagfilter uid:%ld, tbname:%s", *uid, str + 2);
#endif
} else {
STagVal tagVal = {0};
tagVal.cid = pColInfo->info.colId;
if (p1->pTagVal == NULL) {
colDataSetNULL(pColInfo, i);
} else {
const char* p = pStorageAPI->metaFn.extractTagVal(p1->pTagVal, pColInfo->info.type, &tagVal);
if (p == NULL || (pColInfo->info.type == TSDB_DATA_TYPE_JSON && ((STag*)p)->nTag == 0)) {
colDataSetNULL(pColInfo, i);
} else if (pColInfo->info.type == TSDB_DATA_TYPE_JSON) {
code = colDataSetVal(pColInfo, i, p, false);
QUERY_CHECK_CODE(code, lino, _end);
} else if (IS_VAR_DATA_TYPE(pColInfo->info.type)) {
if (IS_STR_DATA_BLOB(pColInfo->info.type)) {
QUERY_CHECK_CODE(code = TSDB_CODE_BLOB_NOT_SUPPORT_TAG, lino, _end);
}
char* tmp = taosMemoryMalloc(tagVal.nData + VARSTR_HEADER_SIZE + 1);
QUERY_CHECK_NULL(tmp, code, lino, _end, terrno);
varDataSetLen(tmp, tagVal.nData);
memcpy(tmp + VARSTR_HEADER_SIZE, tagVal.pData, tagVal.nData);
code = colDataSetVal(pColInfo, i, tmp, false);
#if TAG_FILTER_DEBUG
qDebug("tagfilter varch:%s", tmp + 2);
#endif
taosMemoryFree(tmp);
QUERY_CHECK_CODE(code, lino, _end);
} else {
code = colDataSetVal(pColInfo, i, (const char*)&tagVal.i64, false);
QUERY_CHECK_CODE(code, lino, _end);
#if TAG_FILTER_DEBUG
if (pColInfo->info.type == TSDB_DATA_TYPE_INT) {
qDebug("tagfilter int:%d", *(int*)(&tagVal.i64));
} else if (pColInfo->info.type == TSDB_DATA_TYPE_DOUBLE) {
qDebug("tagfilter double:%f", *(double*)(&tagVal.i64));
}
#endif
}
}
}
}
}
_end:
if (code != TSDB_CODE_SUCCESS) {
blockDataDestroy(pResBlock);
qError("%s failed at line %d since %s", __func__, lino, tstrerror(code));
terrno = code;
return NULL;
}
return pResBlock;
}
static int32_t doSetQualifiedUid(SArray* pUidList, const SArray* pUidTagList, SColumnInfoData* colData) {
taosArrayClear(pUidList);
int32_t numOfTables = taosArrayGetSize(pUidTagList);
bool* pResultList = colData == NULL ? NULL : (bool*)colData->pData;
for (int32_t i = 0; i < numOfTables; ++i) {
if (pResultList != NULL && !pResultList[i]) {
continue;
}
STUidTagInfo* tmpTag = (STUidTagInfo*)taosArrayGet(pUidTagList, i);
if (!tmpTag) {
qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(terrno));
return terrno;
}
uint64_t uid = tmpTag->uid;
qDebug("tagfilter get uid:%" PRId64, uid);
void* tmp = taosArrayPush(pUidList, &uid);
if (tmp == NULL) {
return terrno;
}
}
return TSDB_CODE_SUCCESS;
}
static int32_t copyExistedUids(SArray* pUidTagList, const SArray* pUidList) {
int32_t code = TSDB_CODE_SUCCESS;
int32_t numOfExisted = taosArrayGetSize(pUidList);
if (numOfExisted == 0) {
return code;
}
for (int32_t i = 0; i < numOfExisted; ++i) {
uint64_t* uid = taosArrayGet(pUidList, i);
if (!uid) {
qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(terrno));
return terrno;
}
STUidTagInfo info = {.uid = *uid};
void* tmp = taosArrayPush(pUidTagList, &info);
if (!tmp) {
qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(terrno));
return code;
}
}
return code;
}
static EDealRes getCids(SNode** pNode, void* pContext) {
SHashObj* colHash = (SHashObj*)pContext;
col_id_t colId = 0;
if (QUERY_NODE_COLUMN == nodeType((*pNode))) {
colId = (*(SColumnNode**)pNode)->colId;
} else {
return DEAL_RES_CONTINUE;
}
int32_t tempRes = taosHashPut(colHash, &colId, sizeof(colId), pNode, sizeof((*pNode)));
if (tempRes != TSDB_CODE_SUCCESS ) {
return DEAL_RES_ERROR;
}
return DEAL_RES_CONTINUE;
}
SNode* getTagCondNodeForQueryTmq(void* tinfo) {
SExecTaskInfo* pTaskInfo = (SExecTaskInfo*)tinfo;
const char* id = GET_TASKID(pTaskInfo);
int32_t code = 0;
// traverse to the stream scanner node to add this table id
SOperatorInfo* pInfo = NULL;
code = extractOperatorInTree(pTaskInfo->pRoot, QUERY_NODE_PHYSICAL_PLAN_STREAM_SCAN, id, &pInfo);
if (code != 0 || pInfo == NULL) {
return NULL;
}
STmqQueryScanInfo* pScanInfo = pInfo->info;
return pScanInfo->pTagCond;
}
SNode* getTagCondNodeForStableTmq(void* node) {
return node == NULL ? NULL : ((SSubplan*)node)->pTagCond;
}
bool checkCidInTagCondition(SNode* pTagCond, SArray* cidList) {
if (pTagCond == NULL) {
return false;
}
bool result = true;
SHashObj *colHash = taosHashInit(4, taosGetDefaultHashFunction(TSDB_DATA_TYPE_SMALLINT), true, HASH_NO_LOCK);
if (colHash == NULL) {
goto end;
}
nodesRewriteExprPostOrder(&pTagCond, getCids, (void*)colHash);
if (cidList != NULL) {
int32_t numOfCols = taosArrayGetSize(cidList);
int32_t i = 0;
for (; i < numOfCols; ++i) {
col_id_t* pCid = taosArrayGet(cidList, i);
void* data = taosHashGet(colHash, pCid, sizeof(*pCid));
if (data != NULL) break;
}
if (i == numOfCols) { // no tag column involved in condition
result = false;
goto end;
}
}
end:
taosHashCleanup(colHash);
return result;
}
int32_t doFilterByTagCond(int64_t suid, SArray* pUidList, SNode* pTagCond, void* pVnode,
SIdxFltStatus status, SStorageAPI* pAPI, void* pStreamInfo) {
if (pTagCond == NULL) {
return TSDB_CODE_SUCCESS;
}
terrno = TSDB_CODE_SUCCESS;
int32_t lino = 0;
int32_t code = TSDB_CODE_SUCCESS;
SArray* pBlockList = NULL;
SSDataBlock* pResBlock = NULL;
SScalarParam output = {0};
SArray* pUidTagList = NULL;
SDataType type = {.type = TSDB_DATA_TYPE_BOOL, .bytes = sizeof(bool)};
pUidTagList = taosArrayInit(10, sizeof(STUidTagInfo));
QUERY_CHECK_NULL(pUidTagList, code, lino, end, terrno);
// pUidList size is 0 if tagIndex is NULL and in query condition
code = copyExistedUids(pUidTagList, pUidList);
QUERY_CHECK_CODE(code, lino, end);
// Narrow down the scope of the tablelist set if there is tbname in condition and And Logical operator
code = optimizeTbnameInCond(pVnode, suid, pUidTagList, pTagCond, pAPI);
if (code == 0) {
if (nodeType(pTagCond) == QUERY_NODE_OPERATOR) {
goto end;
}
} else {
qDebug("pUidTagList size:%d", (int32_t)taosArrayGetSize(pUidTagList));
FilterCondType condType = checkTagCond(pTagCond);
if (((condType == FILTER_NO_LOGIC || condType == FILTER_AND) && status != SFLT_NOT_INDEX) || // (super table) use tagIndex and operator is and
(status == SFLT_NOT_INDEX && taosArrayGetSize(pUidTagList) > 0)) { // (child table with tagCond)
code = pAPI->metaFn.getTableTagsByUid(pVnode, suid, pUidTagList);
} else {
taosArrayClearEx(pUidTagList, freeItem); // clear tablelist if using tagIndex and or condition
code = pAPI->metaFn.getTableTags(pVnode, suid, pUidTagList);
}
if (code != TSDB_CODE_SUCCESS) {
qError("failed to get table tags from meta, reason:%s, suid:%" PRIu64, tstrerror(code),suid);
terrno = code;
QUERY_CHECK_CODE(code, lino, end);
}
}
qDebug("final pUidTagList size:%d", (int32_t)taosArrayGetSize(pUidTagList));
int32_t numOfTables = taosArrayGetSize(pUidTagList);
if (numOfTables == 0) {
goto end;
}
SArray* pColList = NULL;
code = qGetColumnsFromNodeList(pTagCond, false, &pColList);
if (code != TSDB_CODE_SUCCESS) {
goto end;
}
pResBlock = createTagValBlockForFilter(pColList, numOfTables, pUidTagList, pVnode, pAPI);
taosArrayDestroy(pColList);
if (pResBlock == NULL) {
code = terrno;
QUERY_CHECK_CODE(code, lino, end);
}
//fprintDataBlock(pResBlock, "tagFilter", "", 0);
// int64_t st1 = taosGetTimestampUs();
// qDebug("generate tag block rows:%d, cost:%ld us", rows, st1-st);
pBlockList = taosArrayInit(2, POINTER_BYTES);
QUERY_CHECK_NULL(pBlockList, code, lino, end, terrno);
void* tmp = taosArrayPush(pBlockList, &pResBlock);
QUERY_CHECK_NULL(tmp, code, lino, end, terrno);
code = createResultData(&type, numOfTables, &output);
if (code != TSDB_CODE_SUCCESS) {
terrno = code;
QUERY_CHECK_CODE(code, lino, end);
}
gTaskScalarExtra.pStreamInfo = pStreamInfo;
gTaskScalarExtra.pStreamRange = NULL;
code = scalarCalculate(pTagCond, pBlockList, &output, &gTaskScalarExtra);
if (code != TSDB_CODE_SUCCESS) {
qError("failed to calculate scalar, reason:%s", tstrerror(code));
terrno = code;
QUERY_CHECK_CODE(code, lino, end);
}
end:
if (code == 0) {
code = doSetQualifiedUid(pUidList, pUidTagList, output.columnData);
lino = __LINE__;
}
if (code != 0) {
qError("%s failed at line %d since %s", __func__, lino, tstrerror(code));
}
blockDataDestroy(pResBlock);
taosArrayDestroy(pBlockList);
taosArrayDestroyEx(pUidTagList, freeItem);
colDataDestroy(output.columnData);
taosMemoryFreeClear(output.columnData);
return code;
}
typedef struct {
int32_t code;
SStreamRuntimeFuncInfo* pStreamRuntimeInfo;
} PlaceHolderContext;
static EDealRes replacePlaceHolderColumn(SNode** pNode, void* pContext) {
PlaceHolderContext* pData = (PlaceHolderContext*)pContext;
if (QUERY_NODE_FUNCTION != nodeType((*pNode))) {
return DEAL_RES_CONTINUE;
}
SFunctionNode* pFuncNode = *(SFunctionNode**)(pNode);
if (!fmIsStreamPesudoColVal(pFuncNode->funcId)) {
return DEAL_RES_CONTINUE;
}
pData->code = fmSetStreamPseudoFuncParamVal(pFuncNode->funcId, pFuncNode->pParameterList, pData->pStreamRuntimeInfo);
if (pData->code != TSDB_CODE_SUCCESS) {
return DEAL_RES_ERROR;
}
SNode* pFirstParam = nodesListGetNode(pFuncNode->pParameterList, 0);
((SValueNode*)pFirstParam)->translate = true;
SValueNode* res = NULL;
pData->code = nodesCloneNode(pFirstParam, (SNode**)&res);
if (NULL == res) {
return DEAL_RES_ERROR;
}
nodesDestroyNode(*pNode);
*pNode = (SNode*)res;
return DEAL_RES_CONTINUE;
}
static void extractTagColId(SOperatorNode* pOpNode, SArray* pColIdArray) {
SNode* pLeft = pOpNode->pLeft;
SNode* pRight = pOpNode->pRight;
SColumnNode* pColNode = nodeType(pLeft) == QUERY_NODE_COLUMN ?
(SColumnNode*)pLeft : (SColumnNode*)pRight;
col_id_t colId = pColNode->colId;
void* _tmp = taosArrayPush(pColIdArray, &colId);
}
static int32_t buildTagCondKey(
const SNode* pTagCond, char** pTagCondKey,
int32_t* tagCondKeyLen, SArray** pTagColIds) {
if (NULL == pTagCond ||
(nodeType(pTagCond) != QUERY_NODE_OPERATOR &&
nodeType(pTagCond) != QUERY_NODE_LOGIC_CONDITION)) {
qError("invalid parameter to extract tag filter symbol");
return TSDB_CODE_INTERNAL_ERROR;
}
int32_t code = TSDB_CODE_SUCCESS;
int32_t lino = 0;
*pTagColIds = taosArrayInit(4, sizeof(col_id_t));
if (nodeType(pTagCond) == QUERY_NODE_OPERATOR) {
extractTagColId((SOperatorNode*)pTagCond, *pTagColIds);
} else if (nodeType(pTagCond) == QUERY_NODE_LOGIC_CONDITION) {
SNode* pChild = NULL;
FOREACH(pChild, ((SLogicConditionNode*)pTagCond)->pParameterList) {
extractTagColId((SOperatorNode*)pChild, *pTagColIds);
}
}
taosArraySort(*pTagColIds, compareUint16Val);
// encode ordered colIds into key string, separated by ','
*tagCondKeyLen =
(int32_t)(taosArrayGetSize(*pTagColIds) * (sizeof(col_id_t) + 1) - 1);
*pTagCondKey = (char*)taosMemoryCalloc(1, *tagCondKeyLen);
TSDB_CHECK_NULL(*pTagCondKey, code, lino, _end, terrno);
char* pStart = *pTagCondKey;
for (int32_t i = 0; i < taosArrayGetSize(*pTagColIds); ++i) {
col_id_t* pColId = (col_id_t*)taosArrayGet(*pTagColIds, i);
TSDB_CHECK_NULL(pColId, code, lino, _end, terrno);
memcpy(pStart, pColId, sizeof(col_id_t));
pStart += sizeof(col_id_t);
if (i != taosArrayGetSize(*pTagColIds) - 1) {
*pStart = ',';
pStart += 1;
}
}
_end:
if (TSDB_CODE_SUCCESS != code) {
qError("%s failed at line %d since %s", __func__, lino, tstrerror(code));
terrno = code;
}
return code;
}
static EDealRes canOptimizeTagCondFilter(SNode* pTagCond, void* pContext) {
if (NULL == pTagCond) {
*(bool*)pContext = false;
return DEAL_RES_END;
}
if (nodeType(pTagCond) == QUERY_NODE_VALUE ||
nodeType(pTagCond) == QUERY_NODE_COLUMN) {
return DEAL_RES_CONTINUE;
}
if (nodeType(pTagCond) == QUERY_NODE_OPERATOR &&
((SOperatorNode*)pTagCond)->opType == OP_TYPE_EQUAL) {
return DEAL_RES_CONTINUE;
}
if (nodeType(pTagCond) == QUERY_NODE_LOGIC_CONDITION &&
((SLogicConditionNode*)pTagCond)->condType == LOGIC_COND_TYPE_AND) {
return DEAL_RES_CONTINUE;
}
if (nodeType(pTagCond) == QUERY_NODE_FUNCTION &&
fmIsStreamPesudoColVal(((SFunctionNode*)pTagCond)->funcId)) {
return DEAL_RES_CONTINUE;
}
*(bool*)pContext = false;
return DEAL_RES_END;
}
int32_t getTableList(void* pVnode, SScanPhysiNode* pScanNode, SNode* pTagCond, SNode* pTagIndexCond,
STableListInfo* pListInfo, uint8_t* digest, const char* idstr, SStorageAPI* pStorageAPI, void* pStreamInfo) {
int32_t code = TSDB_CODE_SUCCESS;
int32_t lino = 0;
size_t numOfTables = 0;
pListInfo->idInfo.suid = pScanNode->suid;
pListInfo->idInfo.tableType = pScanNode->tableType;
SArray* pUidList = taosArrayInit(8, sizeof(uint64_t));
QUERY_CHECK_NULL(pUidList, code, lino, end, terrno);
char* pTagCondKey = NULL;
int32_t tagCondKeyLen;
SArray* pTagColIds = NULL;
char* pPayload = NULL;
SIdxFltStatus status = SFLT_NOT_INDEX;
qTrace("getTableList called, suid:%" PRIu64
", tagCond:%p, tagIndexCond:%p, %d %d", pScanNode->suid, pTagCond,
pTagIndexCond, pScanNode->tableType, pScanNode->virtualStableScan);
if (pScanNode->tableType != TSDB_SUPER_TABLE && !pScanNode->virtualStableScan) {
pListInfo->idInfo.uid = pScanNode->uid;
if (pStorageAPI->metaFn.isTableExisted(pVnode, pScanNode->uid)) {
void* tmp = taosArrayPush(pUidList, &pScanNode->uid);
QUERY_CHECK_NULL(tmp, code, lino, end, terrno);
}
code = doFilterByTagCond(pScanNode->suid, pUidList, pTagCond, pVnode, status, pStorageAPI, pStreamInfo);
QUERY_CHECK_CODE(code, lino, end);
} else {
bool isStream = (pStreamInfo != NULL);
bool hasTagCond = (pTagCond != NULL);
bool canCacheTagEqCondFilter = false;
T_MD5_CTX context = {0};
qTrace("start to get table list by tag filter, suid:%" PRIu64
",tsStableTagFilterCache:%d, tsTagFilterCache:%d",
pScanNode->suid, tsStableTagFilterCache, tsTagFilterCache);
bool acquired = false;
// first, check whether we can use stable tag filter cache
if (tsStableTagFilterCache && isStream && hasTagCond) {
canCacheTagEqCondFilter = true;
nodesWalkExpr(pTagCond, canOptimizeTagCondFilter,
(void*)&canCacheTagEqCondFilter);
}
if (canCacheTagEqCondFilter) {
qDebug("%s, stable tag filter condition can be optimized", idstr);
if (((SStreamRuntimeFuncInfo*)pStreamInfo)->hasPlaceHolder) {
SNode* tmp = NULL;
code = nodesCloneNode((SNode*)pTagCond, &tmp);
QUERY_CHECK_CODE(code, lino, end);
PlaceHolderContext ctx = {.code = TSDB_CODE_SUCCESS, .pStreamRuntimeInfo = (SStreamRuntimeFuncInfo*)pStreamInfo};
nodesRewriteExpr(&tmp, replacePlaceHolderColumn, (void*)&ctx);
if (TSDB_CODE_SUCCESS != ctx.code) {
nodesDestroyNode(tmp);
code = ctx.code;
goto end;
}
code = genStableTagFilterDigest(tmp, &context);
nodesDestroyNode(tmp);
} else {
code = genStableTagFilterDigest(pTagCond, &context);
}
QUERY_CHECK_CODE(code, lino, end);
code = buildTagCondKey(
pTagCond, &pTagCondKey, &tagCondKeyLen, &pTagColIds);
QUERY_CHECK_CODE(code, lino, end);
code = pStorageAPI->metaFn.getStableCachedTableList(
pVnode, pScanNode->suid, pTagCondKey, tagCondKeyLen,
context.digest, tListLen(context.digest), pUidList, &acquired);
QUERY_CHECK_CODE(code, lino, end);
} else if (tsTagFilterCache) {
// second, try to use normal tag filter cache
qDebug("%s using normal tag filter cache", idstr);
if (pStreamInfo != NULL && ((SStreamRuntimeFuncInfo*)pStreamInfo)->hasPlaceHolder) {
SNode* tmp = NULL;
code = nodesCloneNode((SNode*)pTagCond, &tmp);
QUERY_CHECK_CODE(code, lino, end);
PlaceHolderContext ctx = {.code = TSDB_CODE_SUCCESS, .pStreamRuntimeInfo = (SStreamRuntimeFuncInfo*)pStreamInfo};
nodesRewriteExpr(&tmp, replacePlaceHolderColumn, (void*)&ctx);
if (TSDB_CODE_SUCCESS != ctx.code) {
nodesDestroyNode(tmp);
code = ctx.code;
goto end;
}
code = genTagFilterDigest(tmp, &context);
nodesDestroyNode(tmp);
} else {
code = genTagFilterDigest(pTagCond, &context);
}
// try to retrieve the result from meta cache
QUERY_CHECK_CODE(code, lino, end);
code = pStorageAPI->metaFn.getCachedTableList(
pVnode, pScanNode->suid, context.digest,
tListLen(context.digest), pUidList, &acquired);
QUERY_CHECK_CODE(code, lino, end);
}
if (acquired) {
taosArrayDestroy(pTagColIds);
pTagColIds = NULL;
if (digest != NULL) {
digest[0] = 1;
memcpy(digest + 1, context.digest, tListLen(context.digest));
}
qDebug("suid:%" PRIu64 ", %s retrieve table uid list from cache, numOfTables:%d",
pScanNode->suid, idstr, (int32_t)taosArrayGetSize(pUidList));
goto end;
} else {
qDebug("suid:%" PRIu64
", failed to get table uid list from cache", pScanNode->suid);
}
if (!pTagCond) { // no tag filter condition exists, let's fetch all tables of this super table
code = pStorageAPI->metaFn.getChildTableList(pVnode, pScanNode->suid, pUidList);
QUERY_CHECK_CODE(code, lino, end);
qTrace("no tag filter, get all child tables, numOfTables:%d", (int32_t)taosArrayGetSize(pUidList));
} else {
SIdxFltStatus status = SFLT_NOT_INDEX;
if (pTagIndexCond) {
void* pIndex = pStorageAPI->metaFn.getInvertIndex(pVnode);
SIndexMetaArg metaArg = {.metaEx = pVnode,
.idx = pStorageAPI->metaFn.storeGetIndexInfo(pVnode),
.ivtIdx = pIndex,
.suid = pScanNode->uid};
code = doFilterTag(pTagIndexCond, &metaArg, pUidList, &status, &pStorageAPI->metaFilter);
if (code != 0 || status == SFLT_NOT_INDEX) { // temporarily disable it for performance sake
qDebug("failed to get tableIds from index, suid:%" PRIu64 ", uidListSize:%d", pScanNode->uid, (int32_t)taosArrayGetSize(pUidList));
} else {
qDebug("succ to get filter result, table num: %d", (int)taosArrayGetSize(pUidList));
}
}
}
qTrace("after index filter, pTagCond:%p uidListSize:%d", pTagCond, (int32_t)taosArrayGetSize(pUidList));
code = doFilterByTagCond(pScanNode->suid, pUidList, pTagCond, pVnode, status, pStorageAPI, pStreamInfo);
QUERY_CHECK_CODE(code, lino, end);
// let's add the filter results into meta-cache
numOfTables = taosArrayGetSize(pUidList);
if (canCacheTagEqCondFilter) {
qInfo("%s, suid:%" PRIu64 ", add uid list to stable tag filter cache, "
"uidListSize:%d, origin key:%" PRIu64 ",%" PRIu64,
idstr, pScanNode->suid, (int32_t)numOfTables,
*(uint64_t*)context.digest, *(uint64_t*)(context.digest + 8));
code = pStorageAPI->metaFn.putStableCachedTableList(
pVnode, pScanNode->suid, pTagCondKey, tagCondKeyLen,
context.digest, tListLen(context.digest),
pUidList, &pTagColIds);
QUERY_CHECK_CODE(code, lino, end);
if (digest != NULL) {
digest[0] = 1;
memcpy(digest + 1, context.digest, tListLen(context.digest));
}
} else if (tsTagFilterCache) {
qInfo("%s, suid:%" PRIu64 ", add uid list to normal tag filter cache, "
"uidListSize:%d, origin key:%" PRIu64 ",%" PRIu64,
idstr, pScanNode->suid, (int32_t)numOfTables,
*(uint64_t*)context.digest, *(uint64_t*)(context.digest + 8));
size_t size = numOfTables * sizeof(uint64_t) + sizeof(int32_t);
pPayload = taosMemoryMalloc(size);
QUERY_CHECK_NULL(pPayload, code, lino, end, terrno);
*(int32_t*)pPayload = (int32_t)numOfTables;
if (numOfTables > 0) {
void* tmp = taosArrayGet(pUidList, 0);
QUERY_CHECK_NULL(tmp, code, lino, end, terrno);
memcpy(pPayload + sizeof(int32_t), tmp, numOfTables * sizeof(uint64_t));
}
code = pStorageAPI->metaFn.putCachedTableList(pVnode, pScanNode->suid,
context.digest,
tListLen(context.digest),
pPayload, size, 1);
if (TSDB_CODE_SUCCESS == code) {
/*
data referenced by pPayload is used in lru cache,
reset pPayload to NULL to avoid being freed in _error block
*/
pPayload = NULL;
} else {
if (TSDB_CODE_DUP_KEY == code) {
/*
another thread has already put the same key into cache,
we can just ignore this error
*/
code = TSDB_CODE_SUCCESS;
}
QUERY_CHECK_CODE(code, lino, end);
}
if (digest != NULL) {
digest[0] = 1;
memcpy(digest + 1, context.digest, tListLen(context.digest));
}
}
}
qDebug("%s, table list with %d uids built", idstr, (int32_t)numOfTables);
end:
if (code == 0) {
STableKeyInfo info = {.uid = 0, .groupId = 0};
for (int32_t i = 0; i < taosArrayGetSize(pUidList); ++i) {
int64_t* uid = (int64_t*)taosArrayGet(pUidList, i);
QUERY_CHECK_NULL(uid, code, lino, end, terrno);
info.uid = *uid;
//qInfo("doSetQualifiedUid row:%d added to pTableList", i);
void* p = taosArrayPush(pListInfo->pTableList, &info);
QUERY_CHECK_NULL(p, code, lino, end, terrno);
}
} else {
qError("%s failed at line %d since %s", __func__, lino, tstrerror(code));
}
taosArrayDestroy(pUidList);
taosArrayDestroy(pTagColIds);
taosMemFreeClear(pTagCondKey);
taosMemFreeClear(pPayload);
return code;
}
int32_t qGetTableList(int64_t suid, void* pVnode, void* node, SArray** tableList, void* pTaskInfo) {
int32_t code = TSDB_CODE_SUCCESS;
int32_t lino = 0;
SSubplan* pSubplan = (SSubplan*)node;
SScanPhysiNode pNode = {0};
pNode.suid = suid;
pNode.uid = suid;
pNode.tableType = TSDB_SUPER_TABLE;
STableListInfo* pTableListInfo = tableListCreate();
QUERY_CHECK_NULL(pTableListInfo, code, lino, _end, terrno);
code = getTableList(pVnode, &pNode, pSubplan ? pSubplan->pTagCond : NULL, pSubplan ? pSubplan->pTagIndexCond : NULL,
pTableListInfo, NULL, "qGetTableList", &((SExecTaskInfo*)pTaskInfo)->storageAPI, NULL);
QUERY_CHECK_CODE(code, lino, _end);
*tableList = pTableListInfo->pTableList;
pTableListInfo->pTableList = NULL;
_end:
tableListDestroy(pTableListInfo);
if (code != TSDB_CODE_SUCCESS) {
qError("%s failed at line %d since %s", __func__, lino, tstrerror(code));
}
return code;
}
size_t getTableTagsBufLen(const SNodeList* pGroups) {
size_t keyLen = 0;
SNode* node;
FOREACH(node, pGroups) {
SExprNode* pExpr = (SExprNode*)node;
keyLen += pExpr->resType.bytes;
}
keyLen += sizeof(int8_t) * LIST_LENGTH(pGroups);
return keyLen;
}
int32_t getGroupIdFromTagsVal(void* pVnode, uint64_t uid, SNodeList* pGroupNode, char* keyBuf, uint64_t* pGroupId,
SStorageAPI* pAPI) {
SMetaReader mr = {0};
pAPI->metaReaderFn.initReader(&mr, pVnode, META_READER_LOCK, &pAPI->metaFn);
if (pAPI->metaReaderFn.getEntryGetUidCache(&mr, uid) != 0) { // table not exist
pAPI->metaReaderFn.clearReader(&mr);
return TSDB_CODE_PAR_TABLE_NOT_EXIST;
}
SNodeList* groupNew = NULL;
int32_t code = nodesCloneList(pGroupNode, &groupNew);
if (TSDB_CODE_SUCCESS != code) {
pAPI->metaReaderFn.clearReader(&mr);
return code;
}
STransTagExprCtx ctx = {.code = 0, .pReader = &mr};
nodesRewriteExprsPostOrder(groupNew, doTranslateTagExpr, &ctx);
if (TSDB_CODE_SUCCESS != ctx.code) {
nodesDestroyList(groupNew);
pAPI->metaReaderFn.clearReader(&mr);
return code;
}
char* isNull = (char*)keyBuf;
char* pStart = (char*)keyBuf + sizeof(int8_t) * LIST_LENGTH(pGroupNode);
SNode* pNode;
int32_t index = 0;
FOREACH(pNode, groupNew) {
SNode* pNew = NULL;
int32_t code = scalarCalculateConstants(pNode, &pNew);
if (TSDB_CODE_SUCCESS == code) {
REPLACE_NODE(pNew);
} else {
nodesDestroyList(groupNew);
pAPI->metaReaderFn.clearReader(&mr);
return code;
}
if (nodeType(pNew) != QUERY_NODE_VALUE) {
nodesDestroyList(groupNew);
pAPI->metaReaderFn.clearReader(&mr);
qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(TSDB_CODE_QRY_EXECUTOR_INTERNAL_ERROR));
return TSDB_CODE_QRY_EXECUTOR_INTERNAL_ERROR;
}
SValueNode* pValue = (SValueNode*)pNew;
if (pValue->node.resType.type == TSDB_DATA_TYPE_NULL || pValue->isNull) {
isNull[index++] = 1;
continue;
} else {
isNull[index++] = 0;
char* data = nodesGetValueFromNode(pValue);
if (pValue->node.resType.type == TSDB_DATA_TYPE_JSON) {
if (tTagIsJson(data)) {
terrno = TSDB_CODE_QRY_JSON_IN_GROUP_ERROR;
nodesDestroyList(groupNew);
pAPI->metaReaderFn.clearReader(&mr);
return terrno;
}
int32_t len = getJsonValueLen(data);
memcpy(pStart, data, len);
pStart += len;
} else if (IS_VAR_DATA_TYPE(pValue->node.resType.type)) {
if (IS_STR_DATA_BLOB(pValue->node.resType.type)) {
return TSDB_CODE_BLOB_NOT_SUPPORT_TAG;
}
memcpy(pStart, data, varDataTLen(data));
pStart += varDataTLen(data);
} else {
memcpy(pStart, data, pValue->node.resType.bytes);
pStart += pValue->node.resType.bytes;
}
}
}
int32_t len = (int32_t)(pStart - (char*)keyBuf);
*pGroupId = calcGroupId(keyBuf, len);
nodesDestroyList(groupNew);
pAPI->metaReaderFn.clearReader(&mr);
return TSDB_CODE_SUCCESS;
}
SArray* makeColumnArrayFromList(SNodeList* pNodeList) {
if (!pNodeList) {
return NULL;
}
size_t numOfCols = LIST_LENGTH(pNodeList);
SArray* pList = taosArrayInit(numOfCols, sizeof(SColumn));
if (pList == NULL) {
return NULL;
}
for (int32_t i = 0; i < numOfCols; ++i) {
SColumnNode* pColNode = (SColumnNode*)nodesListGetNode(pNodeList, i);
if (!pColNode) {
taosArrayDestroy(pList);
qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(TSDB_CODE_QRY_EXECUTOR_INTERNAL_ERROR));
return NULL;
}
// todo extract method
SColumn c = {0};
c.slotId = pColNode->slotId;
c.colId = pColNode->colId;
c.type = pColNode->node.resType.type;
c.bytes = pColNode->node.resType.bytes;
c.precision = pColNode->node.resType.precision;
c.scale = pColNode->node.resType.scale;
void* tmp = taosArrayPush(pList, &c);
if (!tmp) {
taosArrayDestroy(pList);
qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(terrno));
return NULL;
}
}
return pList;
}
int32_t extractColMatchInfo(SNodeList* pNodeList, SDataBlockDescNode* pOutputNodeList, int32_t* numOfOutputCols,
int32_t type, SColMatchInfo* pMatchInfo) {
size_t numOfCols = LIST_LENGTH(pNodeList);
int32_t code = TSDB_CODE_SUCCESS;
int32_t lino = 0;
bool colIdOrdered = true;
bool hasPrevColId = false;
int32_t prevColId = 0;
pMatchInfo->matchType = type;
SArray* pList = taosArrayInit(numOfCols, sizeof(SColMatchItem));
if (pList == NULL) {
code = terrno;
return code;
}
SColMatchItem** infoBySlot = NULL;
if (numOfCols > 0) {
infoBySlot = taosMemoryCalloc(numOfCols, sizeof(*infoBySlot));
if (infoBySlot == NULL) {
code = terrno;
goto _end;
}
}
SNode* node = NULL;
FOREACH(node, pNodeList) {
STargetNode* pNode = (STargetNode*)node;
QUERY_CHECK_NULL(pNode, code, lino, _end, terrno);
if (nodeType(pNode->pExpr) == QUERY_NODE_COLUMN) {
SColumnNode* pColNode = (SColumnNode*)pNode->pExpr;
SColMatchItem c = {.needOutput = true};
c.colId = pColNode->colId;
c.srcSlotId = pColNode->slotId;
c.dstSlotId = pNode->slotId;
c.isPk = pColNode->isPk;
c.dataType = pColNode->node.resType;
void* tmp = taosArrayPush(pList, &c);
QUERY_CHECK_NULL(tmp, code, lino, _end, terrno);
if (hasPrevColId) {
if (c.colId < prevColId) {
colIdOrdered = false;
}
} else {
hasPrevColId = true;
}
prevColId = c.colId;
if (pNode->slotId >= 0 && pNode->slotId < numOfCols) {
infoBySlot[pNode->slotId] = (SColMatchItem*)taosArrayGet(pList, taosArrayGetSize(pList) - 1);
}
}
}
// set the output flag for each column in SColMatchInfo, according to the
*numOfOutputCols = 0;
SNode* slotNode = NULL;
FOREACH(slotNode, pOutputNodeList->pSlots) {
SSlotDescNode* pNode = (SSlotDescNode*)slotNode;
QUERY_CHECK_NULL(pNode, code, lino, _end, terrno);
// todo: add reserve flag check
// it is a column reserved for the arithmetic expression calculation
if (pNode->slotId >= numOfCols) {
(*numOfOutputCols) += 1;
continue;
}
SColMatchItem* info = infoBySlot ? infoBySlot[pNode->slotId] : NULL;
if (pNode->output) {
(*numOfOutputCols) += 1;
} else if (info != NULL) {
// select distinct tbname from stb where tbname='abc';
info->needOutput = false;
}
}
pMatchInfo->pList = pList;
pMatchInfo->colIdOrdered = colIdOrdered;
_end:
taosMemoryFree(infoBySlot);
if (code != TSDB_CODE_SUCCESS) {
qError("%s failed at line %d since %s", __func__, lino, tstrerror(code));
}
return code;
}
static SResSchema createResSchema(int32_t type, int32_t bytes, int32_t slotId, int32_t scale, int32_t precision,
const char* name) {
SResSchema s = {0};
s.scale = scale;
s.type = type;
s.bytes = bytes;
s.slotId = slotId;
s.precision = precision;
tstrncpy(s.name, name, tListLen(s.name));
return s;
}
static SColumn* createColumn(int32_t blockId, int32_t slotId, int32_t colId, SDataType* pType, EColumnType colType) {
SColumn* pCol = taosMemoryCalloc(1, sizeof(SColumn));
if (pCol == NULL) {
return NULL;
}
pCol->slotId = slotId;
pCol->colId = colId;
pCol->bytes = pType->bytes;
pCol->type = pType->type;
pCol->scale = pType->scale;
pCol->precision = pType->precision;
pCol->dataBlockId = blockId;
pCol->colType = colType;
return pCol;
}
int32_t createExprFromOneNode(SExprInfo* pExp, SNode* pNode, int16_t slotId) {
int32_t code = TSDB_CODE_SUCCESS;
int32_t lino = 0;
pExp->base.numOfParams = 0;
pExp->base.pParam = NULL;
pExp->pExpr = taosMemoryCalloc(1, sizeof(tExprNode));
QUERY_CHECK_NULL(pExp->pExpr, code, lino, _end, terrno);
pExp->pExpr->_function.num = 1;
pExp->pExpr->_function.functionId = -1;
int32_t type = nodeType(pNode);
// it is a project query, or group by column
if (type == QUERY_NODE_COLUMN) {
pExp->pExpr->nodeType = QUERY_NODE_COLUMN;
SColumnNode* pColNode = (SColumnNode*)pNode;
pExp->base.pParam = taosMemoryCalloc(1, sizeof(SFunctParam));
QUERY_CHECK_NULL(pExp->base.pParam, code, lino, _end, terrno);
pExp->base.numOfParams = 1;
SDataType* pType = &pColNode->node.resType;
pExp->base.resSchema =
createResSchema(pType->type, pType->bytes, slotId, pType->scale, pType->precision, pColNode->colName);
pExp->base.pParam[0].pCol =
createColumn(pColNode->dataBlockId, pColNode->slotId, pColNode->colId, pType, pColNode->colType);
QUERY_CHECK_NULL(pExp->base.pParam[0].pCol, code, lino, _end, terrno);
pExp->base.pParam[0].type = FUNC_PARAM_TYPE_COLUMN;
} else if (type == QUERY_NODE_VALUE) {
pExp->pExpr->nodeType = QUERY_NODE_VALUE;
SValueNode* pValNode = (SValueNode*)pNode;
pExp->base.pParam = taosMemoryCalloc(1, sizeof(SFunctParam));
QUERY_CHECK_NULL(pExp->base.pParam, code, lino, _end, terrno);
pExp->base.numOfParams = 1;
SDataType* pType = &pValNode->node.resType;
pExp->base.resSchema =
createResSchema(pType->type, pType->bytes, slotId, pType->scale, pType->precision, pValNode->node.aliasName);
pExp->base.pParam[0].type = FUNC_PARAM_TYPE_VALUE;
code = nodesValueNodeToVariant(pValNode, &pExp->base.pParam[0].param);
QUERY_CHECK_CODE(code, lino, _end);
} else if (type == QUERY_NODE_REMOTE_VALUE) {
SRemoteValueNode* pRemote = (SRemoteValueNode*)pNode;
code = qFetchRemoteNode(gTaskScalarExtra.pSubJobCtx, pRemote->subQIdx, pNode);
QUERY_CHECK_CODE(code, lino, _end);
pExp->pExpr->nodeType = QUERY_NODE_VALUE;
SValueNode* pValNode = (SValueNode*)pNode;
pExp->base.pParam = taosMemoryCalloc(1, sizeof(SFunctParam));
QUERY_CHECK_NULL(pExp->base.pParam, code, lino, _end, terrno);
pExp->base.numOfParams = 1;
SDataType* pType = &pValNode->node.resType;
pExp->base.resSchema =
createResSchema(pType->type, pType->bytes, slotId, pType->scale, pType->precision, pValNode->node.aliasName);
pExp->base.pParam[0].type = FUNC_PARAM_TYPE_VALUE;
code = nodesValueNodeToVariant(pValNode, &pExp->base.pParam[0].param);
QUERY_CHECK_CODE(code, lino, _end);
} else if (type == QUERY_NODE_FUNCTION) {
pExp->pExpr->nodeType = QUERY_NODE_FUNCTION;
SFunctionNode* pFuncNode = (SFunctionNode*)pNode;
SDataType* pType = &pFuncNode->node.resType;
pExp->base.resSchema =
createResSchema(pType->type, pType->bytes, slotId, pType->scale, pType->precision, pFuncNode->node.aliasName);
tExprNode* pExprNode = pExp->pExpr;
pExprNode->_function.functionId = pFuncNode->funcId;
pExprNode->_function.pFunctNode = pFuncNode;
pExprNode->_function.functionType = pFuncNode->funcType;
tstrncpy(pExprNode->_function.functionName, pFuncNode->functionName, tListLen(pExprNode->_function.functionName));
pExp->base.pParamList = pFuncNode->pParameterList;
#if 1
// todo refactor: add the parameter for tbname function
const char* name = "tbname";
int32_t len = strlen(name);
if (!pFuncNode->pParameterList && (memcmp(pExprNode->_function.functionName, name, len) == 0) &&
pExprNode->_function.functionName[len] == 0) {
pFuncNode->pParameterList = NULL;
int32_t code = nodesMakeList(&pFuncNode->pParameterList);
SValueNode* res = NULL;
if (TSDB_CODE_SUCCESS == code) {
code = nodesMakeNode(QUERY_NODE_VALUE, (SNode**)&res);
}
QUERY_CHECK_CODE(code, lino, _end);
res->node.resType = (SDataType){.bytes = sizeof(int64_t), .type = TSDB_DATA_TYPE_BIGINT};
code = nodesListAppend(pFuncNode->pParameterList, (SNode*)res);
if (code != TSDB_CODE_SUCCESS) {
nodesDestroyNode((SNode*)res);
res = NULL;
}
QUERY_CHECK_CODE(code, lino, _end);
}
#endif
int32_t numOfParam = LIST_LENGTH(pFuncNode->pParameterList);
pExp->base.pParam = taosMemoryCalloc(numOfParam, sizeof(SFunctParam));
QUERY_CHECK_NULL(pExp->base.pParam, code, lino, _end, terrno);
pExp->base.numOfParams = numOfParam;
for (int32_t j = 0; j < numOfParam && TSDB_CODE_SUCCESS == code; ++j) {
SNode* p1 = nodesListGetNode(pFuncNode->pParameterList, j);
QUERY_CHECK_NULL(p1, code, lino, _end, terrno);
if (p1->type == QUERY_NODE_COLUMN) {
SColumnNode* pcn = (SColumnNode*)p1;
pExp->base.pParam[j].type = FUNC_PARAM_TYPE_COLUMN;
pExp->base.pParam[j].pCol =
createColumn(pcn->dataBlockId, pcn->slotId, pcn->colId, &pcn->node.resType, pcn->colType);
QUERY_CHECK_NULL(pExp->base.pParam[j].pCol, code, lino, _end, terrno);
} else if (p1->type == QUERY_NODE_VALUE) {
SValueNode* pvn = (SValueNode*)p1;
pExp->base.pParam[j].type = FUNC_PARAM_TYPE_VALUE;
code = nodesValueNodeToVariant(pvn, &pExp->base.pParam[j].param);
QUERY_CHECK_CODE(code, lino, _end);
} else if (p1->type == QUERY_NODE_REMOTE_VALUE) {
SRemoteValueNode* pRemote = (SRemoteValueNode*)p1;
code = qFetchRemoteNode(gTaskScalarExtra.pSubJobCtx, pRemote->subQIdx, p1);
QUERY_CHECK_CODE(code, lino, _end);
SValueNode* pvn = (SValueNode*)pRemote;
pExp->base.pParam[j].type = FUNC_PARAM_TYPE_VALUE;
code = nodesValueNodeToVariant(pvn, &pExp->base.pParam[j].param);
QUERY_CHECK_CODE(code, lino, _end);
}
}
pExp->pExpr->_function.bindExprID = ((SExprNode*)pNode)->bindExprID;
} else if (type == QUERY_NODE_OPERATOR) {
pExp->pExpr->nodeType = QUERY_NODE_OPERATOR;
SOperatorNode* pOpNode = (SOperatorNode*)pNode;
pExp->base.pParam = taosMemoryCalloc(1, sizeof(SFunctParam));
QUERY_CHECK_NULL(pExp->base.pParam, code, lino, _end, terrno);
pExp->base.numOfParams = 1;
SDataType* pType = &pOpNode->node.resType;
pExp->base.resSchema =
createResSchema(pType->type, pType->bytes, slotId, pType->scale, pType->precision, pOpNode->node.aliasName);
pExp->pExpr->_optrRoot.pRootNode = pNode;
} else if (type == QUERY_NODE_CASE_WHEN) {
pExp->pExpr->nodeType = QUERY_NODE_OPERATOR;
SCaseWhenNode* pCaseNode = (SCaseWhenNode*)pNode;
pExp->base.pParam = taosMemoryCalloc(1, sizeof(SFunctParam));
QUERY_CHECK_NULL(pExp->base.pParam, code, lino, _end, terrno);
pExp->base.numOfParams = 1;
SDataType* pType = &pCaseNode->node.resType;
pExp->base.resSchema =
createResSchema(pType->type, pType->bytes, slotId, pType->scale, pType->precision, pCaseNode->node.aliasName);
pExp->pExpr->_optrRoot.pRootNode = pNode;
} else if (type == QUERY_NODE_LOGIC_CONDITION) {
pExp->pExpr->nodeType = QUERY_NODE_OPERATOR;
SLogicConditionNode* pCond = (SLogicConditionNode*)pNode;
pExp->base.pParam = taosMemoryCalloc(1, sizeof(SFunctParam));
QUERY_CHECK_NULL(pExp->base.pParam, code, lino, _end, terrno);
pExp->base.numOfParams = 1;
SDataType* pType = &pCond->node.resType;
pExp->base.resSchema =
createResSchema(pType->type, pType->bytes, slotId, pType->scale, pType->precision, pCond->node.aliasName);
pExp->pExpr->_optrRoot.pRootNode = pNode;
} else {
code = TSDB_CODE_QRY_EXECUTOR_INTERNAL_ERROR;
QUERY_CHECK_CODE(code, lino, _end);
}
pExp->pExpr->relatedTo = ((SExprNode*)pNode)->relatedTo;
_end:
if (code != TSDB_CODE_SUCCESS) {
qError("%s failed at line %d since %s", __func__, lino, tstrerror(code));
}
return code;
}
int32_t createExprFromTargetNode(SExprInfo* pExp, STargetNode* pTargetNode) {
return createExprFromOneNode(pExp, pTargetNode->pExpr, pTargetNode->slotId);
}
SExprInfo* createExpr(SNodeList* pNodeList, int32_t* numOfExprs) {
*numOfExprs = LIST_LENGTH(pNodeList);
SExprInfo* pExprs = taosMemoryCalloc(*numOfExprs, sizeof(SExprInfo));
if (!pExprs) {
return NULL;
}
for (int32_t i = 0; i < (*numOfExprs); ++i) {
SExprInfo* pExp = &pExprs[i];
int32_t code = createExprFromOneNode(pExp, nodesListGetNode(pNodeList, i), i + UD_TAG_COLUMN_INDEX);
if (code != TSDB_CODE_SUCCESS) {
taosMemoryFreeClear(pExprs);
terrno = code;
qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(code));
return NULL;
}
}
return pExprs;
}
int32_t createExprInfo(SNodeList* pNodeList, SNodeList* pGroupKeys, SExprInfo** pExprInfo, int32_t* numOfExprs) {
QRY_PARAM_CHECK(pExprInfo);
int32_t code = 0;
int32_t lino = 0;
int32_t numOfFuncs = LIST_LENGTH(pNodeList);
int32_t numOfGroupKeys = 0;
if (pGroupKeys != NULL) {
numOfGroupKeys = LIST_LENGTH(pGroupKeys);
}
*numOfExprs = numOfFuncs + numOfGroupKeys;
if (*numOfExprs == 0) {
return code;
}
SExprInfo* pExprs = taosMemoryCalloc(*numOfExprs, sizeof(SExprInfo));
QUERY_CHECK_NULL(pExprs, code, lino, _return, terrno);
int32_t i = 0;
if (pNodeList != NULL) {
SNode* node = NULL;
FOREACH(node, pNodeList) {
STargetNode* pTargetNode = (STargetNode*)node;
SExprInfo* pExp = &pExprs[i++];
code = createExprFromTargetNode(pExp, pTargetNode);
QUERY_CHECK_CODE(code, lino, _return);
}
}
if (pGroupKeys != NULL) {
SNode* node = NULL;
FOREACH(node, pGroupKeys) {
STargetNode* pTargetNode = (STargetNode*)node;
SExprInfo* pExp = &pExprs[i++];
code = createExprFromTargetNode(pExp, pTargetNode);
QUERY_CHECK_CODE(code, lino, _return);
}
}
*pExprInfo = pExprs;
return code;
_return:
qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(code));
destroyExprInfo(pExprs, *numOfExprs);
taosMemoryFreeClear(pExprs);
return code;
}
static void deleteSubsidiareCtx(void* pData) {
SSubsidiaryResInfo* pCtx = (SSubsidiaryResInfo*)pData;
if (pCtx->pCtx) {
taosMemoryFreeClear(pCtx->pCtx);
}
}
// set the output buffer for the selectivity + tag query
static int32_t setSelectValueColumnInfo(SqlFunctionCtx* pCtx, int32_t numOfOutput) {
int32_t num = 0;
int32_t code = TSDB_CODE_SUCCESS;
int32_t lino = 0;
SArray* pValCtxArray = NULL;
for (int32_t i = numOfOutput - 1; i > 0; --i) { // select Func is at the end of the list
int32_t funcIdx = pCtx[i].pExpr->pExpr->_function.bindExprID;
if (funcIdx > 0) {
if (pValCtxArray == NULL) {
// the end of the list is the select function of biggest index
pValCtxArray = taosArrayInit_s(sizeof(SSubsidiaryResInfo*), funcIdx);
if (pValCtxArray == NULL) {
return terrno;
}
}
if (funcIdx > pValCtxArray->size) {
qError("funcIdx:%d is out of range", funcIdx);
taosArrayDestroyP(pValCtxArray, deleteSubsidiareCtx);
return TSDB_CODE_QRY_EXECUTOR_INTERNAL_ERROR;
}
SSubsidiaryResInfo* pSubsidiary = &pCtx[i].subsidiaries;
pSubsidiary->pCtx = taosMemoryCalloc(numOfOutput, POINTER_BYTES);
if (pSubsidiary->pCtx == NULL) {
taosArrayDestroyP(pValCtxArray, deleteSubsidiareCtx);
return terrno;
}
pSubsidiary->num = 0;
taosArraySet(pValCtxArray, funcIdx - 1, &pSubsidiary);
}
}
SqlFunctionCtx* p = NULL;
SqlFunctionCtx** pValCtx = NULL;
if (pValCtxArray == NULL) {
pValCtx = taosMemoryCalloc(numOfOutput, POINTER_BYTES);
if (pValCtx == NULL) {
QUERY_CHECK_CODE(terrno, lino, _end);
}
}
for (int32_t i = 0; i < numOfOutput; ++i) {
const char* pName = pCtx[i].pExpr->pExpr->_function.functionName;
if ((strcmp(pName, "_select_value") == 0)) {
if (pValCtxArray == NULL) {
pValCtx[num++] = &pCtx[i];
} else {
int32_t bindFuncIndex = pCtx[i].pExpr->pExpr->relatedTo; // start from index 1;
if (bindFuncIndex > 0) { // 0 is default index related to the select function
bindFuncIndex -= 1;
}
SSubsidiaryResInfo** pSubsidiary = taosArrayGet(pValCtxArray, bindFuncIndex);
if (pSubsidiary == NULL) {
QUERY_CHECK_CODE(TSDB_CODE_QRY_EXECUTOR_INTERNAL_ERROR, lino, _end);
}
(*pSubsidiary)->pCtx[(*pSubsidiary)->num] = &pCtx[i];
(*pSubsidiary)->num++;
}
} else if (fmIsSelectFunc(pCtx[i].functionId)) {
if (pValCtxArray == NULL) {
p = &pCtx[i];
}
}
}
if (p != NULL) {
p->subsidiaries.pCtx = pValCtx;
p->subsidiaries.num = num;
} else {
taosMemoryFreeClear(pValCtx);
}
_end:
if (code != TSDB_CODE_SUCCESS) {
taosArrayDestroyP(pValCtxArray, deleteSubsidiareCtx);
taosMemoryFreeClear(pValCtx);
qError("%s failed at line %d since %s", __func__, lino, tstrerror(code));
} else {
taosArrayDestroy(pValCtxArray);
}
return code;
}
SqlFunctionCtx* createSqlFunctionCtx(SExprInfo* pExprInfo, int32_t numOfOutput, int32_t** rowEntryInfoOffset,
SFunctionStateStore* pStore) {
int32_t code = TSDB_CODE_SUCCESS;
int32_t lino = 0;
SqlFunctionCtx* pFuncCtx = (SqlFunctionCtx*)taosMemoryCalloc(numOfOutput, sizeof(SqlFunctionCtx));
if (pFuncCtx == NULL) {
return NULL;
}
*rowEntryInfoOffset = taosMemoryCalloc(numOfOutput, sizeof(int32_t));
if (*rowEntryInfoOffset == 0) {
taosMemoryFreeClear(pFuncCtx);
return NULL;
}
for (int32_t i = 0; i < numOfOutput; ++i) {
SExprInfo* pExpr = &pExprInfo[i];
SExprBasicInfo* pFunct = &pExpr->base;
SqlFunctionCtx* pCtx = &pFuncCtx[i];
pCtx->functionId = -1;
pCtx->pExpr = pExpr;
if (pExpr->pExpr->nodeType == QUERY_NODE_FUNCTION) {
SFuncExecEnv env = {0};
pCtx->functionId = pExpr->pExpr->_function.pFunctNode->funcId;
pCtx->isPseudoFunc = fmIsWindowPseudoColumnFunc(pCtx->functionId) || fmIsPlaceHolderFunc(pCtx->functionId);
pCtx->isNotNullFunc = fmIsNotNullOutputFunc(pCtx->functionId);
bool isUdaf = fmIsUserDefinedFunc(pCtx->functionId);
if (fmIsAggFunc(pCtx->functionId) || fmIsIndefiniteRowsFunc(pCtx->functionId) ||
fmIsProcessByRowFunc(pCtx->functionId)) {
if (!isUdaf) {
code = fmGetFuncExecFuncs(pCtx->functionId, &pCtx->fpSet);
QUERY_CHECK_CODE(code, lino, _end);
} else {
char* udfName = pExpr->pExpr->_function.pFunctNode->functionName;
pCtx->udfName = taosStrdup(udfName);
QUERY_CHECK_NULL(pCtx->udfName, code, lino, _end, terrno);
code = fmGetUdafExecFuncs(pCtx->functionId, &pCtx->fpSet);
QUERY_CHECK_CODE(code, lino, _end);
}
bool tmp = pCtx->fpSet.getEnv(pExpr->pExpr->_function.pFunctNode, &env);
if (!tmp) {
code = terrno;
QUERY_CHECK_CODE(code, lino, _end);
}
} else {
if (fmIsPlaceHolderFunc(pCtx->functionId)) {
code = fmGetStreamPesudoFuncEnv(pCtx->functionId, pExpr->base.pParamList, &env);
QUERY_CHECK_CODE(code, lino, _end);
}
code = fmGetScalarFuncExecFuncs(pCtx->functionId, &pCtx->sfp);
if (code != TSDB_CODE_SUCCESS && isUdaf) {
code = TSDB_CODE_SUCCESS;
}
QUERY_CHECK_CODE(code, lino, _end);
if (pCtx->sfp.getEnv != NULL) {
bool tmp = pCtx->sfp.getEnv(pExpr->pExpr->_function.pFunctNode, &env);
if (!tmp) {
code = terrno;
QUERY_CHECK_CODE(code, lino, _end);
}
}
}
pCtx->resDataInfo.interBufSize = env.calcMemSize;
} else if (pExpr->pExpr->nodeType == QUERY_NODE_COLUMN || pExpr->pExpr->nodeType == QUERY_NODE_OPERATOR ||
pExpr->pExpr->nodeType == QUERY_NODE_VALUE) {
// for simple column, the result buffer needs to hold at least one element.
pCtx->resDataInfo.interBufSize = pFunct->resSchema.bytes;
}
pCtx->input.numOfInputCols = pFunct->numOfParams;
pCtx->input.pData = taosMemoryCalloc(pFunct->numOfParams, POINTER_BYTES);
QUERY_CHECK_NULL(pCtx->input.pData, code, lino, _end, terrno);
pCtx->input.pColumnDataAgg = taosMemoryCalloc(pFunct->numOfParams, POINTER_BYTES);
QUERY_CHECK_NULL(pCtx->input.pColumnDataAgg, code, lino, _end, terrno);
pCtx->pTsOutput = NULL;
pCtx->resDataInfo.bytes = pFunct->resSchema.bytes;
pCtx->resDataInfo.type = pFunct->resSchema.type;
pCtx->order = TSDB_ORDER_ASC;
pCtx->start.key = INT64_MIN;
pCtx->end.key = INT64_MIN;
pCtx->numOfParams = pExpr->base.numOfParams;
pCtx->param = pFunct->pParam;
pCtx->saveHandle.currentPage = -1;
pCtx->pStore = pStore;
pCtx->hasWindowOrGroup = false;
pCtx->needCleanup = false;
pCtx->skipDynDataCheck = false;
}
for (int32_t i = 1; i < numOfOutput; ++i) {
(*rowEntryInfoOffset)[i] = (int32_t)((*rowEntryInfoOffset)[i - 1] + sizeof(SResultRowEntryInfo) +
pFuncCtx[i - 1].resDataInfo.interBufSize);
}
code = setSelectValueColumnInfo(pFuncCtx, numOfOutput);
QUERY_CHECK_CODE(code, lino, _end);
_end:
if (code != TSDB_CODE_SUCCESS) {
qError("%s failed at line %d since %s", __func__, lino, tstrerror(code));
for (int32_t i = 0; i < numOfOutput; ++i) {
taosMemoryFree(pFuncCtx[i].input.pData);
taosMemoryFree(pFuncCtx[i].input.pColumnDataAgg);
}
taosMemoryFreeClear(*rowEntryInfoOffset);
taosMemoryFreeClear(pFuncCtx);
terrno = code;
return NULL;
}
return pFuncCtx;
}
// NOTE: sources columns are more than the destination SSDatablock columns.
// doFilter in table scan needs every column even its output is false
int32_t relocateColumnData(SSDataBlock* pBlock, const SArray* pColMatchInfo, SArray* pCols, bool outputEveryColumn) {
int32_t code = TSDB_CODE_SUCCESS;
size_t numOfSrcCols = taosArrayGetSize(pCols);
int32_t i = 0, j = 0;
while (i < numOfSrcCols && j < taosArrayGetSize(pColMatchInfo)) {
SColumnInfoData* p = taosArrayGet(pCols, i);
if (!p) {
qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(terrno));
return terrno;
}
SColMatchItem* pmInfo = taosArrayGet(pColMatchInfo, j);
if (!pmInfo) {
return terrno;
}
if (p->info.colId == pmInfo->colId) {
SColumnInfoData* pDst = taosArrayGet(pBlock->pDataBlock, pmInfo->dstSlotId);
if (!pDst) {
return terrno;
}
code = colDataAssign(pDst, p, pBlock->info.rows, &pBlock->info);
if (code != TSDB_CODE_SUCCESS) {
qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(code));
return code;
}
i++;
j++;
} else if (p->info.colId < pmInfo->colId) {
i++;
} else {
qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(TSDB_CODE_QRY_EXECUTOR_INTERNAL_ERROR));
return TSDB_CODE_QRY_EXECUTOR_INTERNAL_ERROR;
}
}
return code;
}
SInterval extractIntervalInfo(const STableScanPhysiNode* pTableScanNode) {
SInterval interval = {
.interval = pTableScanNode->interval,
.sliding = pTableScanNode->sliding,
.intervalUnit = pTableScanNode->intervalUnit,
.slidingUnit = pTableScanNode->slidingUnit,
.offset = pTableScanNode->offset,
.precision = pTableScanNode->scan.node.pOutputDataBlockDesc->precision,
.timeRange = pTableScanNode->scanRange,
};
calcIntervalAutoOffset(&interval);
return interval;
}
SColumn extractColumnFromColumnNode(SColumnNode* pColNode) {
SColumn c = {0};
c.slotId = pColNode->slotId;
c.colId = pColNode->colId;
c.type = pColNode->node.resType.type;
c.bytes = pColNode->node.resType.bytes;
c.scale = pColNode->node.resType.scale;
c.precision = pColNode->node.resType.precision;
return c;
}
/**
* @brief Determine the actual time range for reading data based on the RANGE clause and the WHERE conditions.
* @param[in] cond The range specified by WHERE condition.
* @param[in] range The range specified by RANGE clause.
* @param[out] twindow The range to be read in DESC order, and only one record is needed.
* @param[out] extTwindow The external range to read for only one record, which is used for FILL clause.
* @note `cond` and `twindow` may be the same address.
*/
static int32_t getQueryExtWindow(const STimeWindow* cond, const STimeWindow* range, STimeWindow* twindow,
STimeWindow* extTwindows) {
int32_t code = TSDB_CODE_SUCCESS;
int32_t lino = 0;
STimeWindow tempWindow;
if (cond->skey > cond->ekey || range->skey > range->ekey) {
*twindow = extTwindows[0] = extTwindows[1] = TSWINDOW_DESC_INITIALIZER;
return code;
}
if (range->ekey < cond->skey) {
extTwindows[1] = *cond;
*twindow = extTwindows[0] = TSWINDOW_DESC_INITIALIZER;
return code;
}
if (cond->ekey < range->skey) {
extTwindows[0] = *cond;
*twindow = extTwindows[1] = TSWINDOW_DESC_INITIALIZER;
return code;
}
// Only scan data in the time range intersecion.
extTwindows[0] = extTwindows[1] = *cond;
twindow->skey = TMAX(cond->skey, range->skey);
twindow->ekey = TMIN(cond->ekey, range->ekey);
extTwindows[0].ekey = twindow->skey - 1;
extTwindows[1].skey = twindow->ekey + 1;
return code;
}
static EDealRes condHasRemoteValueWalker(SNode* pNode, void* pContext) {
if (nodeType(pNode) == QUERY_NODE_REMOTE_VALUE ||
nodeType(pNode) == QUERY_NODE_REMOTE_VALUE_LIST ||
nodeType(pNode) == QUERY_NODE_REMOTE_ROW ||
nodeType(pNode) == QUERY_NODE_REMOTE_TABLE ||
nodeType(pNode) == QUERY_NODE_REMOTE_ZERO_ROWS) {
*(bool*)pContext = true;
return DEAL_RES_END;
}
return DEAL_RES_CONTINUE;
}
static bool condHasRemoteValue(SNode* pNode) {
bool found = false;
nodesWalkExpr(pNode, condHasRemoteValueWalker, &found);
return found;
}
// Extract static (non-remote) primary-key time bounds from an AND condition
// that may contain remote subquery references. Remote-bearing children are
// stripped; the remaining literal comparisons are passed to filterGetTimeRange.
// Returns TSDB_CODE_SUCCESS with extracted range, or error on clone/filter failure.
// If no static bound can be extracted, *pExtracted remains at TSWINDOW_INITIALIZER.
static int32_t extractStaticTimeRangeFromAndCond(SNode* pCond, STimeWindow* pExtracted, bool* pStrict) {
int32_t code = TSDB_CODE_SUCCESS;
SNode* pProbe = NULL;
code = nodesCloneNode(pCond, &pProbe);
if (code != TSDB_CODE_SUCCESS || pProbe == NULL) {
nodesDestroyNode(pProbe);
return code;
}
if (nodeType(pProbe) == QUERY_NODE_LOGIC_CONDITION &&
((SLogicConditionNode*)pProbe)->condType == LOGIC_COND_TYPE_AND) {
SLogicConditionNode* pAnd = (SLogicConditionNode*)pProbe;
SListCell* pCell = pAnd->pParameterList ? pAnd->pParameterList->pHead : NULL;
while (pCell != NULL) {
if (condHasRemoteValue(pCell->pNode)) {
pCell = nodesListErase(pAnd->pParameterList, pCell);
} else {
pCell = pCell->pNext;
}
}
int32_t remaining = pAnd->pParameterList ? (int32_t)pAnd->pParameterList->length : 0;
if (remaining == 0) {
nodesDestroyNode(pProbe);
return TSDB_CODE_SUCCESS;
}
if (remaining == 1) {
SNode* pOnly = pAnd->pParameterList->pHead->pNode;
pAnd->pParameterList->pHead->pNode = NULL;
nodesDestroyNode(pProbe);
pProbe = pOnly;
}
} else if (condHasRemoteValue(pProbe)) {
// Non-AND node that is itself remote — no static bound extractable.
nodesDestroyNode(pProbe);
return TSDB_CODE_SUCCESS;
}
code = filterGetTimeRange(pProbe, pExtracted, pStrict, NULL);
nodesDestroyNode(pProbe);
return code;
}
static int32_t getPrimaryTimeRange(SNode** pPrimaryKeyCond, STimeWindow* pTimeRange, bool* isStrict, bool isStream) {
// In stream mode, a WHERE primary-key cond that references a remote
// subquery (e.g. WHERE ts >= (SELECT last_row(ts) FROM ref)) must not be
// folded to a literal time range here; doing so freezes the bound at
// task-build time and the same range is reused for every trigger event.
// Keep the original cond intact so the caller merges it into pConditions
// and the runtime filter re-evaluates the RemoteValueNode per fetch.
if (isStream && condHasRemoteValue(*pPrimaryKeyCond)) {
*isStrict = false;
STimeWindow incoming = *pTimeRange;
STimeWindow extracted = TSWINDOW_INITIALIZER;
bool probeStrict = true;
int32_t code = extractStaticTimeRangeFromAndCond(*pPrimaryKeyCond, &extracted, &probeStrict);
if (TSDB_CODE_SUCCESS == code && extracted.skey != INT64_MIN) {
// Intersect the extracted literal range with the incoming window.
pTimeRange->skey = TMAX(incoming.skey, extracted.skey);
pTimeRange->ekey = TMIN(incoming.ekey, extracted.ekey);
} else if (code != TSDB_CODE_SUCCESS) {
*pTimeRange = incoming;
}
*isStrict = false;
return code;
}
SNode* pNew = NULL;
int32_t code = scalarCalculateRemoteConstants(*pPrimaryKeyCond, &pNew);
if (TSDB_CODE_SUCCESS == code) {
*pPrimaryKeyCond = pNew;
if (nodeType(pNew) != QUERY_NODE_VALUE) {
code = filterGetTimeRange(*pPrimaryKeyCond, pTimeRange, isStrict, NULL);
}
}
return code;
}
int32_t initQueryTableDataCond(SQueryTableDataCond* pCond, STableScanPhysiNode* pTableScanNode,
const SReadHandle* readHandle, bool applyExtWin) {
int32_t code = 0;
pCond->order = pTableScanNode->scanSeq[0] > 0 ? TSDB_ORDER_ASC : TSDB_ORDER_DESC;
pCond->numOfCols = LIST_LENGTH(pTableScanNode->scan.pScanCols);
pCond->colList = taosMemoryCalloc(pCond->numOfCols, sizeof(SColumnInfo));
if (!pCond->colList) {
return terrno;
}
pCond->pSlotList = taosMemoryMalloc(sizeof(int32_t) * pCond->numOfCols);
if (pCond->pSlotList == NULL) {
taosMemoryFreeClear(pCond->colList);
return terrno;
}
// TODO: get it from stable scan node
pCond->twindows = pTableScanNode->scanRange;
pCond->suid = pTableScanNode->scan.suid;
pCond->type = TIMEWINDOW_RANGE_CONTAINED;
pCond->startVersion = -1;
pCond->endVersion = -1;
pCond->skipRollup = readHandle->skipRollup;
if (readHandle->winRangeValid) {
pCond->twindows = readHandle->winRange;
}
pCond->cacheSttStatis = readHandle->cacheSttStatis;
// allowed read stt file optimization mode
pCond->notLoadData = (pTableScanNode->dataRequired == FUNC_DATA_REQUIRED_NOT_LOAD) &&
(pTableScanNode->scan.node.pConditions == NULL) && (pTableScanNode->interval == 0);
int32_t j = 0;
SNode* node = NULL;
FOREACH(node, pTableScanNode->scan.pScanCols) {
STargetNode* pNode = (STargetNode*)node;
if (!pNode) {
qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(terrno));
return terrno;
}
SColumnNode* pColNode = (SColumnNode*)pNode->pExpr;
if (pColNode->colType == COLUMN_TYPE_TAG) {
continue;
}
pCond->colList[j].type = pColNode->node.resType.type;
pCond->colList[j].bytes = pColNode->node.resType.bytes;
pCond->colList[j].colId = pColNode->colId;
pCond->colList[j].pk = pColNode->isPk;
pCond->pSlotList[j] = pNode->slotId;
j += 1;
}
pCond->numOfCols = j;
if (applyExtWin) {
if (NULL != pTableScanNode->pExtScanRange) {
pCond->type = TIMEWINDOW_RANGE_EXTERNAL;
code = getQueryExtWindow(&pCond->twindows, pTableScanNode->pExtScanRange, &pCond->twindows, pCond->extTwindows);
} else if (readHandle->extWinRangeValid) {
pCond->type = TIMEWINDOW_RANGE_EXTERNAL;
code = getQueryExtWindow(&pCond->twindows, &readHandle->extWinRange, &pCond->twindows, pCond->extTwindows);
}
}
if (pTableScanNode->pPrimaryCond) {
bool isStrict = false;
bool isStream = (readHandle != NULL && readHandle->streamRtInfo != NULL);
code = getPrimaryTimeRange((SNode**)&pTableScanNode->pPrimaryCond, &pCond->twindows, &isStrict, isStream);
if (code || !isStrict) {
code = nodesMergeNode((SNode**)&pTableScanNode->scan.node.pConditions, &pTableScanNode->pPrimaryCond);
}
}
return code;
}
int32_t initQueryTableDataCondWithColArray(SQueryTableDataCond* pCond, SQueryTableDataCond* pOrgCond,
const SReadHandle* readHandle, SArray* colArray) {
int32_t code = TSDB_CODE_SUCCESS;
int32_t lino = 0;
pCond->order = TSDB_ORDER_ASC;
pCond->numOfCols = (int32_t)taosArrayGetSize(colArray);
pCond->colList = taosMemoryCalloc(pCond->numOfCols, sizeof(SColumnInfo));
QUERY_CHECK_NULL(pCond->colList, code, lino, _return, terrno);
pCond->pSlotList = taosMemoryMalloc(sizeof(int32_t) * pCond->numOfCols);
QUERY_CHECK_NULL(pCond->pSlotList, code, lino, _return, terrno);
pCond->twindows = pOrgCond->twindows;
pCond->order = pOrgCond->order;
pCond->type = pOrgCond->type;
pCond->startVersion = -1;
pCond->endVersion = -1;
pCond->skipRollup = true;
pCond->notLoadData = false;
for (int32_t i = 0; i < pCond->numOfCols; ++i) {
SColIdPair* pColPair = taosArrayGet(colArray, i);
QUERY_CHECK_NULL(pColPair, code, lino, _return, terrno);
bool find = false;
for (int32_t j = 0; j < pOrgCond->numOfCols; ++j) {
if (pOrgCond->colList[j].colId == pColPair->vtbColId) {
pCond->colList[i].type = pOrgCond->colList[j].type;
// For variable-length types (nchar/binary/varchar/varbinary), use the source table's bytes
// to avoid tsdb reader doCopyColVal length check failure when source data is longer
// than the virtual table's defined length.
if (IS_VAR_DATA_TYPE(pOrgCond->colList[j].type) && pColPair->type.bytes > 0) {
pCond->colList[i].bytes = TMAX(pOrgCond->colList[j].bytes, pColPair->type.bytes);
} else {
pCond->colList[i].bytes = pOrgCond->colList[j].bytes;
}
pCond->colList[i].colId = pColPair->orgColId;
pCond->colList[i].pk = pOrgCond->colList[j].pk;
pCond->pSlotList[i] = i;
find = true;
qDebug("%s mapped vtb colId:%d to org colId:%d, type:%d, bytes:%d", __func__, pColPair->vtbColId,
pColPair->orgColId, pCond->colList[i].type, pCond->colList[i].bytes);
break;
}
}
QUERY_CHECK_CONDITION(find, code, lino, _return, TSDB_CODE_NOT_FOUND);
}
return code;
_return:
qError("%s failed at line %d since %s", __func__, lino, tstrerror(terrno));
taosMemoryFreeClear(pCond->colList);
taosMemoryFreeClear(pCond->pSlotList);
return code;
}
void cleanupQueryTableDataCond(SQueryTableDataCond* pCond) {
taosMemoryFreeClear(pCond->colList);
taosMemoryFreeClear(pCond->pSlotList);
}
int32_t convertFillType(int32_t mode) {
int32_t type = TSDB_FILL_NONE;
switch (mode) {
case FILL_MODE_PREV:
type = TSDB_FILL_PREV;
break;
case FILL_MODE_NONE:
type = TSDB_FILL_NONE;
break;
case FILL_MODE_NULL:
type = TSDB_FILL_NULL;
break;
case FILL_MODE_NULL_F:
type = TSDB_FILL_NULL_F;
break;
case FILL_MODE_NEXT:
type = TSDB_FILL_NEXT;
break;
case FILL_MODE_VALUE:
type = TSDB_FILL_SET_VALUE;
break;
case FILL_MODE_VALUE_F:
type = TSDB_FILL_SET_VALUE_F;
break;
case FILL_MODE_LINEAR:
type = TSDB_FILL_LINEAR;
break;
case FILL_MODE_NEAR:
type = TSDB_FILL_NEAR;
break;
default:
type = TSDB_FILL_NONE;
}
return type;
}
void getInitialStartTimeWindow(SInterval* pInterval, TSKEY ts, STimeWindow* w, bool ascQuery) {
if (ascQuery) {
*w = getAlignQueryTimeWindow(pInterval, ts);
} else {
// the start position of the first time window in the endpoint that spreads beyond the queried last timestamp
*w = getAlignQueryTimeWindow(pInterval, ts);
int64_t key = w->skey;
while (key < ts) { // moving towards end
key = getNextTimeWindowStart(pInterval, key, TSDB_ORDER_ASC);
if (key > ts) {
break;
}
w->skey = key;
}
w->ekey = taosTimeAdd(w->skey, pInterval->interval, pInterval->intervalUnit, pInterval->precision, NULL) - 1;
}
}
static STimeWindow doCalculateTimeWindow(int64_t ts, SInterval* pInterval) {
STimeWindow w = {0};
w.skey = taosTimeTruncate(ts, pInterval);
w.ekey = taosTimeGetIntervalEnd(w.skey, pInterval);
return w;
}
STimeWindow getFirstQualifiedTimeWindow(int64_t ts, STimeWindow* pWindow, SInterval* pInterval, int32_t order) {
STimeWindow win = *pWindow;
STimeWindow save = win;
while (win.skey <= ts && win.ekey >= ts) {
save = win;
// get previous time window
getNextTimeWindow(pInterval, &win, order == TSDB_ORDER_DESC ? TSDB_ORDER_ASC : TSDB_ORDER_DESC);
}
return save;
}
// get the correct time window according to the handled timestamp
// todo refactor
STimeWindow getActiveTimeWindow(SDiskbasedBuf* pBuf, SResultRowInfo* pResultRowInfo, int64_t ts, SInterval* pInterval,
int32_t order) {
STimeWindow w = {0};
if (pResultRowInfo->cur.pageId == -1) { // the first window, from the previous stored value
getInitialStartTimeWindow(pInterval, ts, &w, (order != TSDB_ORDER_DESC));
return w;
}
SResultRow* pRow = getResultRowByPos(pBuf, &pResultRowInfo->cur, false);
if (pRow) {
TAOS_SET_OBJ_ALIGNED(&w, pRow->win);
}
// in case of typical time window, we can calculate time window directly.
if (w.skey > ts || w.ekey < ts) {
w = doCalculateTimeWindow(ts, pInterval);
}
if (pInterval->interval != pInterval->sliding) {
// it is an sliding window query, in which sliding value is not equalled to
// interval value, and we need to find the first qualified time window.
w = getFirstQualifiedTimeWindow(ts, &w, pInterval, order);
}
return w;
}
void getNextTimeWindow(const SInterval* pInterval, STimeWindow* tw, int32_t order) {
tw->skey = getNextTimeWindowStart(pInterval, tw->skey, order);
tw->ekey = taosTimeGetIntervalEnd(tw->skey, pInterval);
}
bool hasLimitOffsetInfo(SLimitInfo* pLimitInfo) {
return (pLimitInfo->limit.limit != -1 || pLimitInfo->limit.offset != -1 || pLimitInfo->slimit.limit != -1 ||
pLimitInfo->slimit.offset != -1);
}
bool hasSlimitOffsetInfo(SLimitInfo* pLimitInfo) {
return (pLimitInfo->slimit.limit != -1 || pLimitInfo->slimit.offset != -1);
}
void initLimitInfo(const SNode* pLimit, const SNode* pSLimit, SLimitInfo* pLimitInfo) {
SLimit limit = {.limit = getLimit(pLimit), .offset = getOffset(pLimit)};
SLimit slimit = {.limit = getLimit(pSLimit), .offset = getOffset(pSLimit)};
pLimitInfo->limit = limit;
pLimitInfo->slimit = slimit;
pLimitInfo->remainOffset = limit.offset;
pLimitInfo->remainGroupOffset = slimit.offset;
pLimitInfo->numOfOutputRows = 0;
pLimitInfo->numOfOutputGroups = 0;
pLimitInfo->currentGroupId = 0;
}
void resetLimitInfoForNextGroup(SLimitInfo* pLimitInfo) {
pLimitInfo->numOfOutputRows = 0;
pLimitInfo->remainOffset = pLimitInfo->limit.offset;
}
int32_t tableListGetSize(const STableListInfo* pTableList, int32_t* pRes) {
if (taosArrayGetSize(pTableList->pTableList) != taosHashGetSize(pTableList->map)) {
qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(TSDB_CODE_QRY_EXECUTOR_INTERNAL_ERROR));
return TSDB_CODE_QRY_EXECUTOR_INTERNAL_ERROR;
}
(*pRes) = taosArrayGetSize(pTableList->pTableList);
return TSDB_CODE_SUCCESS;
}
uint64_t tableListGetSuid(const STableListInfo* pTableList) { return pTableList->idInfo.suid; }
STableKeyInfo* tableListGetInfo(const STableListInfo* pTableList, int32_t index) {
if (taosArrayGetSize(pTableList->pTableList) == 0) {
return NULL;
}
return taosArrayGet(pTableList->pTableList, index);
}
int32_t tableListFind(const STableListInfo* pTableList, uint64_t uid, int32_t startIndex) {
int32_t numOfTables = taosArrayGetSize(pTableList->pTableList);
if (startIndex >= numOfTables) {
return -1;
}
for (int32_t i = startIndex; i < numOfTables; ++i) {
STableKeyInfo* p = taosArrayGet(pTableList->pTableList, i);
if (!p) {
qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(terrno));
return -1;
}
if (p->uid == uid) {
return i;
}
}
return -1;
}
void tableListGetSourceTableInfo(const STableListInfo* pTableList, uint64_t* psuid, uint64_t* uid, int32_t* type) {
*psuid = pTableList->idInfo.suid;
*uid = pTableList->idInfo.uid;
*type = pTableList->idInfo.tableType;
}
int32_t tableListGetTableGroupId(const STableListInfo* pTableList, uint64_t tableUid, uint64_t* gid, uint64_t* baseGid) {
int32_t* slot = taosHashGet(pTableList->map, &tableUid, sizeof(tableUid));
if (slot == NULL) {
qDebug("table:%" PRIu64 " not found in table list", tableUid);
return TSDB_CODE_PAR_TABLE_NOT_EXIST;
}
STableKeyInfo* pKeyInfo = taosArrayGet(pTableList->pTableList, *slot);
if (pKeyInfo == NULL) {
qDebug("table:%" PRIu64 " not found in table list", tableUid);
return TSDB_CODE_PAR_TABLE_NOT_EXIST;
}
*gid = pKeyInfo->groupId;
*baseGid = pKeyInfo->baseGId;
return 0;
}
// TODO handle the group offset info, fix it, the rule of group output will be broken by this function
// int32_t tableListRemoveTableInfo(STableListInfo* pTableList, uint64_t uid) {
// int32_t code = TSDB_CODE_SUCCESS;
// int32_t lino = 0;
// int32_t* slot = taosHashGet(pTableList->map, &uid, sizeof(uid));
// if (slot == NULL) {
// qDebug("table:%" PRIu64 " not found in table list", uid);
// return 0;
// }
// taosArrayRemove(pTableList->pTableList, *slot);
// code = taosHashRemove(pTableList->map, &uid, sizeof(uid));
// _end:
// if (code != TSDB_CODE_SUCCESS) {
// qError("%s failed at line %d since %s", __func__, lino, tstrerror(code));
// } else {
// qDebug("uid:%" PRIu64 ", remove from table list", uid);
// }
// return code;
// }
int32_t tableListAddTableInfo(STableListInfo* pTableList, uint64_t uid, uint64_t gid) {
int32_t code = TSDB_CODE_SUCCESS;
int32_t lino = 0;
if (pTableList->map == NULL) {
pTableList->map = taosHashInit(32, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY), false, HASH_ENTRY_LOCK);
QUERY_CHECK_NULL(pTableList->map, code, lino, _end, terrno);
}
STableKeyInfo keyInfo = {.uid = uid, .groupId = gid};
void* p = taosHashGet(pTableList->map, &uid, sizeof(uid));
if (p != NULL) {
qInfo("table:%" PRId64 " already in tableIdList, ignore it", uid);
goto _end;
}
void* tmp = taosArrayPush(pTableList->pTableList, &keyInfo);
QUERY_CHECK_NULL(tmp, code, lino, _end, terrno);
int32_t slot = (int32_t)taosArrayGetSize(pTableList->pTableList) - 1;
code = taosHashPut(pTableList->map, &uid, sizeof(uid), &slot, sizeof(slot));
if (code != TSDB_CODE_SUCCESS) {
// we have checked the existence of uid in hash map above
QUERY_CHECK_CONDITION((code != TSDB_CODE_DUP_KEY), code, lino, _end, TSDB_CODE_QRY_EXECUTOR_INTERNAL_ERROR);
taosArrayPopTailBatch(pTableList->pTableList, 1); // let's pop the last element in the array list
}
_end:
if (code != TSDB_CODE_SUCCESS) {
qError("%s failed at line %d since %s", __func__, lino, tstrerror(code));
} else {
qDebug("uid:%" PRIu64 ", groupId:%" PRIu64 " added into table list, slot:%d, total:%d", uid, gid, slot, slot + 1);
}
return code;
}
int32_t tableListGetGroupList(const STableListInfo* pTableList, int32_t ordinalGroupIndex, STableKeyInfo** pKeyInfo,
int32_t* size) {
int32_t totalGroups = tableListGetOutputGroups(pTableList);
int32_t numOfTables = 0;
int32_t code = tableListGetSize(pTableList, &numOfTables);
if (code != TSDB_CODE_SUCCESS) {
qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(code));
return code;
}
if (ordinalGroupIndex < 0 || ordinalGroupIndex >= totalGroups) {
return TSDB_CODE_INVALID_PARA;
}
// here handle two special cases:
// 1. only one group exists, and 2. one table exists for each group.
if (totalGroups == 1) {
*size = numOfTables;
*pKeyInfo = (*size == 0) ? NULL : taosArrayGet(pTableList->pTableList, 0);
return TSDB_CODE_SUCCESS;
} else if (totalGroups == numOfTables) {
*size = 1;
*pKeyInfo = taosArrayGet(pTableList->pTableList, ordinalGroupIndex);
return TSDB_CODE_SUCCESS;
}
int32_t offset = pTableList->groupOffset[ordinalGroupIndex];
if (ordinalGroupIndex < totalGroups - 1) {
*size = pTableList->groupOffset[ordinalGroupIndex + 1] - offset;
} else {
*size = numOfTables - offset;
}
*pKeyInfo = taosArrayGet(pTableList->pTableList, offset);
return TSDB_CODE_SUCCESS;
}
int32_t tableListGetOutputGroups(const STableListInfo* pTableList) { return pTableList->numOfOuputGroups; }
bool oneTableForEachGroup(const STableListInfo* pTableList) { return pTableList->oneTableForEachGroup; }
STableListInfo* tableListCreate() {
STableListInfo* pListInfo = taosMemoryCalloc(1, sizeof(STableListInfo));
if (pListInfo == NULL) {
return NULL;
}
pListInfo->remainGroups = NULL;
pListInfo->pTableList = taosArrayInit(4, sizeof(STableKeyInfo));
if (pListInfo->pTableList == NULL) {
goto _error;
}
pListInfo->map = taosHashInit(1024, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BIGINT), false, HASH_ENTRY_LOCK);
if (pListInfo->map == NULL) {
goto _error;
}
pListInfo->numOfOuputGroups = 1;
return pListInfo;
_error:
tableListDestroy(pListInfo);
return NULL;
}
void tableListDestroy(STableListInfo* pTableListInfo) {
if (pTableListInfo == NULL) {
return;
}
taosArrayDestroy(pTableListInfo->pTableList);
taosMemoryFreeClear(pTableListInfo->groupOffset);
taosHashCleanup(pTableListInfo->map);
taosHashCleanup(pTableListInfo->remainGroups);
pTableListInfo->pTableList = NULL;
pTableListInfo->map = NULL;
taosMemoryFree(pTableListInfo);
}
void tableListClear(STableListInfo* pTableListInfo) {
if (pTableListInfo == NULL) {
return;
}
taosArrayClear(pTableListInfo->pTableList);
taosHashClear(pTableListInfo->map);
taosHashClear(pTableListInfo->remainGroups);
taosMemoryFree(pTableListInfo->groupOffset);
pTableListInfo->numOfOuputGroups = 1;
pTableListInfo->oneTableForEachGroup = false;
}
static int32_t orderbyGroupIdComparFn(const void* p1, const void* p2) {
STableKeyInfo* pInfo1 = (STableKeyInfo*)p1;
STableKeyInfo* pInfo2 = (STableKeyInfo*)p2;
if (pInfo1->groupId == pInfo2->groupId) {
return 0;
} else {
return pInfo1->groupId < pInfo2->groupId ? -1 : 1;
}
}
int32_t sortTableGroup(STableListInfo* pTableListInfo) {
int32_t code = TSDB_CODE_SUCCESS;
taosArraySort(pTableListInfo->pTableList, orderbyGroupIdComparFn);
int32_t size = taosArrayGetSize(pTableListInfo->pTableList);
if (size == 0) {
pTableListInfo->numOfOuputGroups = 0;
return code;
}
SArray* pList = taosArrayInit(4, sizeof(int32_t));
if (!pList) {
code = terrno;
goto end;
}
STableKeyInfo* pInfo = taosArrayGet(pTableListInfo->pTableList, 0);
if (pInfo == NULL) {
qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(terrno));
code = terrno;
goto end;
}
uint64_t gid = pInfo->groupId;
int32_t start = 0;
void* tmp = taosArrayPush(pList, &start);
if (tmp == NULL) {
qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(terrno));
code = terrno;
goto end;
}
for (int32_t i = 1; i < size; ++i) {
pInfo = taosArrayGet(pTableListInfo->pTableList, i);
if (pInfo == NULL) {
qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(terrno));
code = terrno;
goto end;
}
if (pInfo->groupId != gid) {
tmp = taosArrayPush(pList, &i);
if (tmp == NULL) {
qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(terrno));
code = terrno;
goto end;
}
gid = pInfo->groupId;
}
}
pTableListInfo->numOfOuputGroups = taosArrayGetSize(pList);
pTableListInfo->groupOffset = taosMemoryMalloc(sizeof(int32_t) * pTableListInfo->numOfOuputGroups);
if (pTableListInfo->groupOffset == NULL) {
code = terrno;
goto end;
}
memcpy(pTableListInfo->groupOffset, taosArrayGet(pList, 0), sizeof(int32_t) * pTableListInfo->numOfOuputGroups);
end:
taosArrayDestroy(pList);
return code;
}
int32_t buildGroupIdMapForAllTables(STableListInfo* pTableListInfo, SReadHandle* pHandle, SScanPhysiNode* pScanNode,
SNodeList* group, bool groupSort, uint8_t* digest, SStorageAPI* pAPI, SHashObj* groupIdMap, bool gIdFromBaseId) {
int32_t code = TSDB_CODE_SUCCESS;
bool groupByTbname = groupbyTbname(group);
size_t numOfTables = taosArrayGetSize(pTableListInfo->pTableList);
if (!numOfTables) {
return code;
}
qDebug("numOfTables:%zu, groupByTbname:%d, group:%p", numOfTables, groupByTbname, group);
if (group == NULL || groupByTbname) {
if (tsCountAlwaysReturnValue && QUERY_NODE_PHYSICAL_PLAN_TABLE_SCAN == nodeType(pScanNode) &&
((STableScanPhysiNode*)pScanNode)->needCountEmptyTable) {
pTableListInfo->remainGroups =
taosHashInit(numOfTables, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BIGINT), false, HASH_NO_LOCK);
if (pTableListInfo->remainGroups == NULL) {
return terrno;
}
for (int i = 0; i < numOfTables; i++) {
STableKeyInfo* info = taosArrayGet(pTableListInfo->pTableList, i);
if (!info) {
qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(terrno));
return terrno;
}
info->groupId = groupByTbname ? info->uid : 0;
int32_t tempRes = taosHashPut(pTableListInfo->remainGroups, &(info->groupId), sizeof(info->groupId),
&(info->uid), sizeof(info->uid));
if (tempRes != TSDB_CODE_SUCCESS && tempRes != TSDB_CODE_DUP_KEY) {
qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(tempRes));
return tempRes;
}
}
} else {
for (int32_t i = 0; i < numOfTables; i++) {
STableKeyInfo* info = taosArrayGet(pTableListInfo->pTableList, i);
if (!info) {
qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(terrno));
return terrno;
}
info->groupId = groupByTbname ? info->uid : 0;
}
}
if (groupIdMap && group != NULL){
getColInfoResultForGroupbyForStream(pHandle->vnode, group, pTableListInfo, pAPI, groupIdMap);
}
pTableListInfo->oneTableForEachGroup = groupByTbname;
if (numOfTables == 1 && pTableListInfo->idInfo.tableType == TSDB_CHILD_TABLE) {
pTableListInfo->oneTableForEachGroup = true;
}
if (groupSort && groupByTbname) {
taosArraySort(pTableListInfo->pTableList, orderbyGroupIdComparFn);
pTableListInfo->numOfOuputGroups = numOfTables;
} else if (groupByTbname && pScanNode->groupOrderScan) {
pTableListInfo->numOfOuputGroups = numOfTables;
} else {
pTableListInfo->numOfOuputGroups = 1;
}
if (groupSort || pScanNode->groupOrderScan) {
code = sortTableGroup(pTableListInfo);
}
} else {
bool initRemainGroups = false;
if (QUERY_NODE_PHYSICAL_PLAN_TABLE_SCAN == nodeType(pScanNode)) {
STableScanPhysiNode* pTableScanNode = (STableScanPhysiNode*)pScanNode;
if (tsCountAlwaysReturnValue && pTableScanNode->needCountEmptyTable &&
!(groupSort || pScanNode->groupOrderScan)) {
initRemainGroups = true;
}
}
code = getColInfoResultForGroupby(pHandle, group, pTableListInfo, digest, pAPI, initRemainGroups, groupIdMap, gIdFromBaseId);
if (code != TSDB_CODE_SUCCESS) {
return code;
}
if (pScanNode->groupOrderScan) pTableListInfo->numOfOuputGroups = taosArrayGetSize(pTableListInfo->pTableList);
if (groupSort || pScanNode->groupOrderScan) {
code = sortTableGroup(pTableListInfo);
}
}
qTrace("EXEC_GROUP_TRACE build_group_map done scanType:%d group:%p groupByTbname:%d groupSort:%d groupOrderScan:%d "
"oneTableEachGroup:%d numTables:%zu numOutputGroups:%d code:%d",
nodeType((SNode*)pScanNode), group, groupByTbname, groupSort, pScanNode->groupOrderScan,
pTableListInfo->oneTableForEachGroup, taosArrayGetSize(pTableListInfo->pTableList),
pTableListInfo->numOfOuputGroups, code);
// add all table entry in the hash map
size_t size = taosArrayGetSize(pTableListInfo->pTableList);
for (int32_t i = 0; i < size; ++i) {
STableKeyInfo* p = taosArrayGet(pTableListInfo->pTableList, i);
if (!p) {
qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(terrno));
return terrno;
}
int32_t tempRes = taosHashPut(pTableListInfo->map, &p->uid, sizeof(uint64_t), &i, sizeof(int32_t));
if (tempRes != TSDB_CODE_SUCCESS && tempRes != TSDB_CODE_DUP_KEY) {
qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(tempRes));
return tempRes;
}
}
return code;
}
int32_t createNonStreamScanTableListInfo(SScanPhysiNode* pScanNode, SNodeList* pGroupTags, bool groupSort, SReadHandle* pHandle,
STableListInfo* pTableListInfo, SNode* pTagCond, SNode* pTagIndexCond,
SExecTaskInfo* pTaskInfo, SHashObj* groupIdMap) {
int64_t st = taosGetTimestampUs();
const char* idStr = GET_TASKID(pTaskInfo);
if (pHandle == NULL) {
qError("invalid handle, in creating operator tree, %s", idStr);
return TSDB_CODE_INVALID_PARA;
}
if (pHandle->uid != 0) {
pScanNode->uid = pHandle->uid;
pScanNode->tableType = TSDB_CHILD_TABLE;
}
uint8_t digest[17] = {0};
int32_t code = getTableList(pHandle->vnode, pScanNode, pTagCond, pTagIndexCond, pTableListInfo, digest, idStr,
&pTaskInfo->storageAPI, pTaskInfo->pStreamRuntimeInfo);
if (code != TSDB_CODE_SUCCESS) {
qError("failed to getTableList, code:%s", tstrerror(code));
return code;
}
int32_t numOfTables = taosArrayGetSize(pTableListInfo->pTableList);
int64_t st1 = taosGetTimestampUs();
pTaskInfo->cost.extractListTime = (st1 - st) / 1000.0;
qDebug("extract queried table list completed, %d tables, elapsed time:%.2f ms %s", numOfTables,
pTaskInfo->cost.extractListTime, idStr);
if (numOfTables == 0) {
qDebug("no table qualified for query, %s", idStr);
return TSDB_CODE_SUCCESS;
}
code = buildGroupIdMapForAllTables(pTableListInfo, pHandle, pScanNode, pGroupTags, groupSort, digest, &pTaskInfo->storageAPI, groupIdMap, false);
if (code != TSDB_CODE_SUCCESS) {
return code;
}
pTaskInfo->cost.groupIdMapTime = (taosGetTimestampUs() - st1) / 1000.0;
qDebug("generate group id map completed, elapsed time:%.2f ms %s", pTaskInfo->cost.groupIdMapTime, idStr);
return TSDB_CODE_SUCCESS;
}
int32_t createStreamGrpTableListFromCond(SScanPhysiNode* pScanNode, SReadHandle* pHandle, SExecTaskInfo* pTaskInfo, SStreamRuntimeFuncInfo* pStream, STableListInfo* pTableListInfo, SNode* pTagCond, SNode* pTagIndexCond, int32_t grpNum, bool hasCalcGrp) {
int32_t code = 0, lino = 0;
for (int32_t i = 0; i < grpNum; ++i) {
SSTriggerGroupReadInfo* pGrpRead = taosArrayGet(pStream->curGrpRead, i);
SSTriggerGroupCalcInfo* pGrpCalc = tSimpleHashGet(pStream->pGroupCalcInfos, &pGrpRead->gid, sizeof(pGrpRead->gid));
pStream->pStreamPartColVals = pGrpCalc->pGroupColVals;
int32_t oldSize = taosArrayGetSize(pTableListInfo->pTableList);
TAOS_CHECK_EXIT(getTableList(pHandle->vnode, pScanNode, pTagCond, pTagIndexCond, pTableListInfo, NULL, GET_TASKID(pTaskInfo),
&pTaskInfo->storageAPI, pTaskInfo->pStreamRuntimeInfo));
int32_t newSize = taosArrayGetSize(pTableListInfo->pTableList);
for (int32_t i = oldSize; i < newSize; ++i) {
STableKeyInfo* pKey = taosArrayGet(pTableListInfo->pTableList, i);
pKey->baseGId = pGrpRead->gid;
if (!hasCalcGrp) {
pKey->groupId = pGrpRead->gid;
}
}
qDebug("%s get %d tables for tgrp %" PRIu64, GET_TASKID(pTaskInfo), newSize - oldSize, pGrpRead->gid);
}
qDebug("%s get total %d tables in table list from cond", GET_TASKID(pTaskInfo), (int32_t)taosArrayGetSize(pTableListInfo->pTableList));
_exit:
if (code) {
qError("%s %s failed at line %d since %s", GET_TASKID(pTaskInfo), __func__, __LINE__, tstrerror(code));
}
return code;}
int32_t createStreamGrpTableListFromTrig(SExecTaskInfo* pTaskInfo, SStreamRuntimeFuncInfo* pStream, STableListInfo* pTableListInfo, int32_t grpNum, bool hasCalcGrp) {
STableKeyInfo tableKey = {0};
int32_t tblNum = 0, idx = 0;
int32_t code = 0, lino = 0;
if (!hasCalcGrp) {
pTableListInfo->numOfOuputGroups = grpNum;
}
for (int32_t i = 0; i < grpNum; ++i) {
SSTriggerGroupReadInfo* pGrp = taosArrayGet(pStream->curGrpRead, i);
tableKey.baseGId = pGrp->gid;
tblNum = taosArrayGetSize(pGrp->pTables);
if (pStream->stbPartByTbname) {
tableKey.uid = pGrp->gid;
if (!hasCalcGrp) {
tableKey.groupId = pGrp->gid;
pTableListInfo->oneTableForEachGroup = true;
}
TSDB_CHECK_NULL(taosArrayPush(pTableListInfo->pTableList, &tableKey), code, lino, _exit, terrno);
if (!hasCalcGrp) {
int32_t tempRes = taosHashPut(pTableListInfo->map, &tableKey.uid, sizeof(uint64_t), &idx, sizeof(int32_t));
if (tempRes != TSDB_CODE_SUCCESS && tempRes != TSDB_CODE_DUP_KEY) {
TAOS_CHECK_EXIT(tempRes);
}
}
qDebug("%s gid:%" PRIu64 " table:%" PRIu64 " added to stream table list, partByTbname", GET_TASKID(pTaskInfo), tableKey.groupId, tableKey.uid);
idx++;
continue;
}
for (int32_t n = 0; n < tblNum; ++n, ++idx) {
uint64_t* pUid = taosArrayGet(pGrp->pTables, n);
tableKey.uid = *pUid;
if (!hasCalcGrp) {
tableKey.groupId = pGrp->gid;
}
TSDB_CHECK_NULL(taosArrayPush(pTableListInfo->pTableList, &tableKey), code, lino, _exit, terrno);
if (!hasCalcGrp) {
int32_t tempRes = taosHashPut(pTableListInfo->map, &tableKey.uid, sizeof(uint64_t), &idx, sizeof(int32_t));
if (tempRes != TSDB_CODE_SUCCESS && tempRes != TSDB_CODE_DUP_KEY) {
TAOS_CHECK_EXIT(tempRes);
}
}
qDebug("%s gid:%" PRIu64 " table:%" PRIu64 " added to stream table list", GET_TASKID(pTaskInfo), tableKey.groupId, tableKey.uid);
}
}
qDebug("%s get total %d tables in table list from trig", GET_TASKID(pTaskInfo), (int32_t)taosArrayGetSize(pTableListInfo->pTableList));
_exit:
if (code) {
qError("%s %s failed at line %d since %s", GET_TASKID(pTaskInfo), __func__, __LINE__, tstrerror(code));
}
return code;
}
int32_t createStreamMultiGrpTableListInfo(SScanPhysiNode* pScanNode, SNodeList* pGroupTags, bool groupSort, SReadHandle* pHandle,
STableListInfo* pTableListInfo, SNode* pTagCond, SNode* pTagIndexCond,
SExecTaskInfo* pTaskInfo, SHashObj* groupIdMap) {
int32_t code = 0, lino = 0;
const char* idStr = GET_TASKID(pTaskInfo);
SStreamRuntimeFuncInfo* pStream = &pTaskInfo->pStreamRuntimeInfo->funcInfo;
bool hasCalcGrp = (pGroupTags && pGroupTags->length > 0);
int32_t grpNum = taosArrayGetSize(pStream->curGrpRead);
SSTriggerGroupReadInfo* pGrp = taosArrayGet(pStream->curGrpRead, 0);
int64_t st = taosGetTimestampUs();
qDebug("%s start to create table list, hasCalcGrp:%d, tgrpNum:%d", idStr, hasCalcGrp, grpNum);
if (taosArrayGetSize(pGrp->pTables) <= 0) {
TAOS_CHECK_EXIT(createStreamGrpTableListFromCond(pScanNode, pHandle, pTaskInfo, pStream, pTableListInfo, pTagCond, pTagIndexCond, grpNum, hasCalcGrp));
} else {
TAOS_CHECK_EXIT(createStreamGrpTableListFromTrig(pTaskInfo, pStream, pTableListInfo, grpNum, hasCalcGrp));
}
pTableListInfo->idInfo.suid = pScanNode->suid;
pTableListInfo->idInfo.tableType = pScanNode->tableType;
int32_t numOfTables = taosArrayGetSize(pTableListInfo->pTableList);
int64_t st1 = taosGetTimestampUs();
qDebug("%s build multi tgrp table list completed, elapsed time:%.2f ms", idStr, (st1 - st) / 1000.0);
if (numOfTables == 0) {
qDebug("%s no table qualified for stream multi tgrp", idStr);
return TSDB_CODE_SUCCESS;
}
code = buildGroupIdMapForAllTables(pTableListInfo, pHandle, pScanNode, pGroupTags, groupSort, NULL, &pTaskInfo->storageAPI, groupIdMap, true);
if (code != TSDB_CODE_SUCCESS) {
return code;
}
pTaskInfo->cost.groupIdMapTime = (taosGetTimestampUs() - st1) / 1000.0;
qDebug("%s generate stream group id map completed, elapsed time:%.2f ms", idStr, pTaskInfo->cost.groupIdMapTime);
_exit:
if (code) {
qError("%s %s failed at line %d since %s", GET_TASKID(pTaskInfo), __func__, lino, tstrerror(code));
}
return code;
}
int32_t createScanTableListInfo(SScanPhysiNode* pScanNode, SNodeList* pGroupTags, bool groupSort, SReadHandle* pHandle,
STableListInfo* pTableListInfo, SNode* pTagCond, SNode* pTagIndexCond,
SExecTaskInfo* pTaskInfo, SHashObj* groupIdMap) {
int32_t code = 0;
// Ensure curGrpRead is prepared from pre-initialized group calc infos for non-stream external-window initialization.
if (pTaskInfo->pStreamRuntimeInfo) {
SStreamRuntimeFuncInfo* pStream = &pTaskInfo->pStreamRuntimeInfo->funcInfo;
if (pStream->isMultiGroupCalc && pStream->curGrpRead == NULL && pStream->pGroupCalcInfos != NULL) {
int32_t n = tSimpleHashGetSize(pStream->pGroupCalcInfos);
if (n < 0) n = 0;
pStream->curGrpRead = taosArrayInit_s(sizeof(SSTriggerGroupReadInfo), n > 0 ? n : 4);
if (pStream->curGrpRead == NULL) {
return terrno;
}
int32_t iter = 0;
SSTriggerGroupCalcInfo* pGrp = tSimpleHashIterate(pStream->pGroupCalcInfos, NULL, &iter);
while (pGrp != NULL) {
uint64_t gid = *(uint64_t*)tSimpleHashGetKey(pGrp, NULL);
SSTriggerGroupReadInfo readInfo = {0};
readInfo.gid = gid;
readInfo.pTables = NULL; // tables decided by getTableList when needed
if (taosArrayPush(pStream->curGrpRead, &readInfo) == NULL) {
return terrno;
}
pGrp = tSimpleHashIterate(pStream->pGroupCalcInfos, pGrp, &iter);
}
}
}
if (pTaskInfo->pStreamRuntimeInfo && pTaskInfo->pStreamRuntimeInfo->funcInfo.isMultiGroupCalc /*&&got table list*/) {
code = createStreamMultiGrpTableListInfo(pScanNode, pGroupTags, groupSort,
pHandle, pTableListInfo, pTagCond, pTagIndexCond, pTaskInfo, groupIdMap);
if (code) {
qError("%s failed to createStreamMultiGrpTableListInfo, code:%s", __func__, tstrerror(code));
return code;
}
} else if (!pScanNode->node.dynamicOp) {
code = createNonStreamScanTableListInfo(pScanNode, pGroupTags, groupSort,
pHandle, pTableListInfo, pTagCond, pTagIndexCond, pTaskInfo, groupIdMap);
if (code) {
qError("%s failed to createScanTableListInfo, code:%s", __func__, tstrerror(code));
return code;
}
}
return TSDB_CODE_SUCCESS;
}
char* getStreamOpName(uint16_t opType) {
switch (opType) {
case QUERY_NODE_PHYSICAL_PLAN_STREAM_SCAN:
return "stream scan";
case QUERY_NODE_PHYSICAL_PLAN_PROJECT:
return "project";
case QUERY_NODE_PHYSICAL_PLAN_EXTERNAL_WINDOW:
return "external window";
}
return "error name";
}
void printDataBlock(SSDataBlock* pBlock, const char* flag, const char* taskIdStr, int64_t qId) {
if (qDebugFlag & DEBUG_TRACE) {
if (!pBlock) {
qDebug("%" PRIx64 " %s %s %s: Block is Null", qId, taskIdStr, flag, __func__);
return;
} else if (pBlock->info.rows == 0) {
qDebug("%" PRIx64 " %s %s %s: Block is Empty.", qId, taskIdStr, flag, __func__);
return;
}
char* pBuf = NULL;
int32_t code = dumpBlockData(pBlock, flag, &pBuf, taskIdStr, qId);
if (code == 0) {
qDebugL("%" PRIx64 " %s %s", qId, __func__, pBuf);
taosMemoryFree(pBuf);
}
}
}
void printSpecDataBlock(SSDataBlock* pBlock, const char* flag, const char* opStr, const char* taskIdStr) {
if (!pBlock) {
qDebug("%s===stream===%s %s: Block is Null", taskIdStr, flag, opStr);
return;
} else if (pBlock->info.rows == 0) {
qDebug("%s===stream===%s %s: Block is Empty.skey:%" PRId64 ",ekey:%" PRId64 ",version%" PRId64,
taskIdStr, flag, opStr, pBlock->info.window.skey, pBlock->info.window.ekey,
pBlock->info.version);
return;
}
if (qDebugFlag & DEBUG_TRACE) {
char* pBuf = NULL;
char flagBuf[64];
snprintf(flagBuf, sizeof(flagBuf), "%s %s", flag, opStr);
int32_t code = dumpBlockData(pBlock, flagBuf, &pBuf, taskIdStr, 0);
if (code == 0) {
qDebug("%s", pBuf);
taosMemoryFree(pBuf);
}
}
}
TSKEY getStartTsKey(STimeWindow* win, const TSKEY* tsCols) { return tsCols == NULL ? win->skey : tsCols[0]; }
void updateTimeWindowInfo(SColumnInfoData* pColData, const STimeWindow* pWin, int64_t delta) {
int64_t* ts = (int64_t*)pColData->pData;
int64_t duration = pWin->ekey > pWin->skey ? pWin->ekey - pWin->skey + delta : pWin->skey - pWin->ekey + delta;
ts[2] = duration; // set the duration
ts[3] = pWin->skey; // window start key
ts[4] = pWin->ekey + delta; // window end key
}
int32_t compKeys(const SArray* pSortGroupCols, const char* oldkeyBuf, int32_t oldKeysLen, const SSDataBlock* pBlock,
int32_t rowIndex) {
SColumnDataAgg* pColAgg = NULL;
const char* isNull = oldkeyBuf;
const char* p = oldkeyBuf + sizeof(int8_t) * pSortGroupCols->size;
for (int32_t i = 0; i < pSortGroupCols->size; ++i) {
const SColumn* pCol = (SColumn*)TARRAY_GET_ELEM(pSortGroupCols, i);
const SColumnInfoData* pColInfoData = TARRAY_GET_ELEM(pBlock->pDataBlock, pCol->slotId);
if (pBlock->pBlockAgg) pColAgg = &pBlock->pBlockAgg[pCol->slotId];
if (colDataIsNull(pColInfoData, pBlock->info.rows, rowIndex, pColAgg)) {
if (isNull[i] != 1) return 1;
} else {
if (isNull[i] != 0) return 1;
const char* val = colDataGetData(pColInfoData, rowIndex);
if (pCol->type == TSDB_DATA_TYPE_JSON) {
int32_t len = getJsonValueLen(val);
if (memcmp(p, val, len) != 0) return 1;
p += len;
} else if (IS_VAR_DATA_TYPE(pCol->type)) {
if (IS_STR_DATA_BLOB(pCol->type)) {
if (memcmp(p, val, blobDataTLen(val)) != 0) return 1;
p += blobDataTLen(val);
} else {
if (memcmp(p, val, varDataTLen(val)) != 0) return 1;
p += varDataTLen(val);
}
} else {
if (0 != memcmp(p, val, pCol->bytes)) return 1;
p += pCol->bytes;
}
}
}
if ((int32_t)(p - oldkeyBuf) != oldKeysLen) return 1;
return 0;
}
int32_t buildKeys(char* keyBuf, const SArray* pSortGroupCols, const SSDataBlock* pBlock, int32_t rowIndex) {
uint32_t colNum = pSortGroupCols->size;
SColumnDataAgg* pColAgg = NULL;
char* isNull = keyBuf;
char* p = keyBuf + sizeof(int8_t) * colNum;
for (int32_t i = 0; i < colNum; ++i) {
const SColumn* pCol = (SColumn*)TARRAY_GET_ELEM(pSortGroupCols, i);
const SColumnInfoData* pColInfoData = TARRAY_GET_ELEM(pBlock->pDataBlock, pCol->slotId);
if (pCol->slotId > pBlock->pDataBlock->size) continue;
if (pBlock->pBlockAgg) pColAgg = &pBlock->pBlockAgg[pCol->slotId];
if (colDataIsNull(pColInfoData, pBlock->info.rows, rowIndex, pColAgg)) {
isNull[i] = 1;
} else {
isNull[i] = 0;
const char* val = colDataGetData(pColInfoData, rowIndex);
if (pCol->type == TSDB_DATA_TYPE_JSON) {
int32_t len = getJsonValueLen(val);
memcpy(p, val, len);
p += len;
} else if (IS_VAR_DATA_TYPE(pCol->type)) {
if (IS_STR_DATA_BLOB(pCol->type)) {
blobDataCopy(p, val);
p += blobDataTLen(val);
} else {
varDataCopy(p, val);
p += varDataTLen(val);
}
} else {
memcpy(p, val, pCol->bytes);
p += pCol->bytes;
}
}
}
return (int32_t)(p - keyBuf);
}
uint64_t calcGroupId(char* pData, int32_t len) {
T_MD5_CTX context;
tMD5Init(&context);
tMD5Update(&context, (uint8_t*)pData, len);
tMD5Final(&context);
// NOTE: only extract the initial 8 bytes of the final MD5 digest
uint64_t id = 0;
memcpy(&id, context.digest, sizeof(uint64_t));
if (0 == id) memcpy(&id, context.digest + 8, sizeof(uint64_t));
return id;
}
SNodeList* makeColsNodeArrFromSortKeys(SNodeList* pSortKeys) {
SNode* node;
SNodeList* ret = NULL;
FOREACH(node, pSortKeys) {
SOrderByExprNode* pSortKey = (SOrderByExprNode*)node;
int32_t code = nodesListMakeAppend(&ret, pSortKey->pExpr);
if (code != TSDB_CODE_SUCCESS) {
qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(code));
terrno = code;
return NULL;
}
}
return ret;
}
int32_t extractKeysLen(const SArray* keys, int32_t* pLen) {
int32_t code = TSDB_CODE_SUCCESS;
int32_t lino = 0;
int32_t len = 0;
int32_t keyNum = taosArrayGetSize(keys);
for (int32_t i = 0; i < keyNum; ++i) {
SColumn* pCol = (SColumn*)taosArrayGet(keys, i);
QUERY_CHECK_NULL(pCol, code, lino, _end, terrno);
len += pCol->bytes;
}
len += sizeof(int8_t) * keyNum; // null flag
*pLen = len;
_end:
if (code != TSDB_CODE_SUCCESS) {
qError("%s failed at line %d since %s", __func__, lino, tstrerror(code));
}
return code;
}
int32_t parseErrorMsgFromAnalyticServer(SJson* pJson, const char* typeStr, const char* pId) {
int32_t code = TSDB_CODE_ANA_ANODE_RETURN_ERROR;
char* pMsg = NULL;
if (pJson == NULL) {
return code;
}
int32_t ret = tjsonDupStringValue(pJson, "msg", &pMsg);
if (ret == 0 && pMsg != NULL) {
qError("%s failed to exec %s operation, msg:%s", pId, typeStr, pMsg);
if (strstr(pMsg, "white noise") != NULL) {
code = TSDB_CODE_ANA_WN_DATA;
} else if (strstr(pMsg, "white-noise") != NULL) {
code = TSDB_CODE_ANA_WN_DATA;
} else if (strstr(pMsg, "[Errno 111] Connection refused") != NULL) {
code = TSDB_CODE_ANA_ALGO_NOT_LOAD;
} else if (strstr(pMsg, "failed to load model") != NULL) {
code = TSDB_CODE_ANA_ALGO_NOT_LOAD;
}
} else {
qError("%s failed to extract msg from server, unknown error", pId);
}
taosMemoryFreeClear(pMsg);
return code;
}
int32_t createExprSubQResBlock(SSDataBlock** ppBlock, SDataType* pResType, int32_t cols) {
int32_t code = 0;
SSDataBlock* pBlock = taosMemoryCalloc(1, sizeof(SSDataBlock));
if (pBlock == NULL) {
return terrno;
}
pBlock->pDataBlock = taosArrayInit(cols, sizeof(SColumnInfoData));
if (pBlock->pDataBlock == NULL) {
code = terrno;
taosMemoryFree(pBlock);
return code;
}
for (int32_t i = 0; i < cols; ++i) {
SColumnInfoData idata = {0};
if (pResType) {
idata = createColumnInfoData(pResType->type, pResType->bytes, 0);
idata.info.scale = pResType->scale;
idata.info.precision = pResType->precision;
}
code = blockDataAppendColInfo(pBlock, &idata);
if (code != TSDB_CODE_SUCCESS) {
qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(code));
blockDataDestroy(pBlock);
*ppBlock = NULL;
return code;
}
}
*ppBlock = pBlock;
return code;
}
int32_t extractSingleRspBlock(SRetrieveTableRsp* pRetrieveRsp, SSDataBlock* pb) {
int32_t code = TSDB_CODE_SUCCESS;
int32_t lino = 0;
void* decompBuf = NULL;
char* pNextStart = pRetrieveRsp->data;
char* pStart = pNextStart;
int32_t index = 0;
if (pRetrieveRsp->compressed) { // decompress the data
decompBuf = taosMemoryMalloc(pRetrieveRsp->payloadLen);
QUERY_CHECK_NULL(decompBuf, code, lino, _end, terrno);
}
int32_t compLen = *(int32_t*)pStart;
pStart += sizeof(int32_t);
int32_t rawLen = *(int32_t*)pStart;
pStart += sizeof(int32_t);
QUERY_CHECK_CONDITION((compLen <= rawLen && compLen != 0), code, lino, _end, TSDB_CODE_QRY_EXECUTOR_INTERNAL_ERROR);
pNextStart = pStart + compLen;
if (pRetrieveRsp->compressed && (compLen < rawLen)) {
int32_t t = tsDecompressString(pStart, compLen, 1, decompBuf, rawLen, ONE_STAGE_COMP, NULL, 0);
QUERY_CHECK_CONDITION((t == rawLen), code, lino, _end, TSDB_CODE_QRY_EXECUTOR_INTERNAL_ERROR);
pStart = decompBuf;
}
code = blockDecodeInternal(pb, pStart, (const char**)&pStart);
if (code != 0) {
taosMemoryFreeClear(pRetrieveRsp);
goto _end;
}
_end:
if (code != TSDB_CODE_SUCCESS) {
blockDataDestroy(pb);
qError("%s failed at line %d since %s", __func__, lino, tstrerror(code));
}
return code;
}
int32_t setValueFromResBlock(STaskSubJobCtx* ctx, SValueNode* pRes, SSDataBlock* pBlock) {
int32_t code = 0;
bool needFree = true;
int32_t colNum = taosArrayGetSize(pBlock->pDataBlock);
if (NULL == pBlock->pDataBlock || 1 > colNum || pBlock->info.rows > 1) {
qError("%s invalid scl fetch res block, pDataBlock:%p, colNum:%d, rows:%" PRId64,
ctx->idStr, pBlock->pDataBlock, colNum, pBlock->info.rows);
return TSDB_CODE_PAR_INVALID_SCALAR_SUBQ_RES_ROWS;
}
pRes->flag &= (~VALUE_FLAG_VAL_UNSET);
pRes->translate = true;
// Reset isNull so a previously-NULL placeholder (e.g. from an earlier
// stream event whose subquery returned no rows) isn't re-emitted as
// NULL when this fetch produced a real value.
pRes->isNull = false;
// For variable-length / DECIMAL types, datum.p owns a heap buffer that
// was transferred from the response block on the previous fetch. In
// stream mode this function is called per trigger event, so free the
// prior buffer before nodesSetValueNodeValueExt overwrites datum.p,
// otherwise we leak one buffer per event.
if (IS_VAR_DATA_TYPE(pRes->node.resType.type) ||
pRes->node.resType.type == TSDB_DATA_TYPE_DECIMAL) {
taosMemoryFreeClear(pRes->datum.p);
}
SColumnInfoData* pCol = taosArrayGet(pBlock->pDataBlock, 0);
if (colDataIsNull_s(pCol, 0)) {
pRes->isNull = true;
} else {
code = nodesSetValueNodeValueExt(pRes, colDataGetData(pCol, 0), &needFree);
}
if (!needFree) {
pCol->pData = NULL;
}
return code;
}
void handleRemoteValueRes(SScalarFetchParam* pParam, STaskSubJobCtx* ctx, SRetrieveTableRsp* pRsp, bool* fetchDone) {
SSDataBlock* pResBlock = NULL;
SExecTaskInfo* pTaskInfo = ctx->pTaskInfo;
SRemoteValueNode* pRemote = (SRemoteValueNode*)pParam->pRes;
qDebug("%s scl fetch value rsp received, subQIdx:%d, rows:%" PRId64 , ctx->idStr, pParam->subQIdx, pRsp->numOfRows);
if (IS_STREAM_MODE(pTaskInfo)) {
SNode** ppRes = taosArrayGet(ctx->subResNodes, pParam->subQIdx);
if (NULL == *ppRes && 0 == pRsp->numOfRows) {
// First-call empty result: keep node type as REMOTE_VALUE so the
// scalar walker re-dispatches sclWalkRemoteValue per evaluation.
pRemote->val.isNull = true;
pRemote->val.translate = true;
pRemote->val.flag &= (~VALUE_FLAG_VAL_UNSET);
taosArraySet(ctx->subResNodes, pParam->subQIdx, &pParam->pRes);
} else if (pRsp->numOfRows > 0 || pRsp->numOfBlocks > 0) {
ctx->code = createExprSubQResBlock(&pResBlock, &((SRemoteValueNode*)pParam->pRes)->val.node.resType, 1);
if (TSDB_CODE_SUCCESS == ctx->code) {
ctx->code = blockDataEnsureCapacity(pResBlock, 1);
}
if (TSDB_CODE_SUCCESS == ctx->code) {
ctx->code = extractSingleRspBlock(pRsp, pResBlock);
}
if (TSDB_CODE_SUCCESS == ctx->code) {
ctx->code = setValueFromResBlock(ctx, &pRemote->val, pResBlock);
pRemote->val.node.type = QUERY_NODE_REMOTE_VALUE;
}
if (TSDB_CODE_SUCCESS == ctx->code) {
taosArraySet(ctx->subResNodes, pParam->subQIdx, &pParam->pRes);
}
blockDataDestroy(pResBlock);
} else if (NULL != *ppRes && 0 == pRsp->numOfRows) {
pRsp->completed = true;
}
*fetchDone = (TSDB_CODE_SUCCESS != ctx->code || pRsp->completed) ? true : false;
if (!(*fetchDone)) {
int32_t code = sendFetchRemoteNodeReq(ctx, pParam->subQIdx, pParam->pRes, false);
if (TSDB_CODE_SUCCESS != code) {
ctx->code = code;
*fetchDone = true;
}
}
return;
}
*fetchDone = true;
if (pRsp->numOfRows > 1 || pRsp->numOfBlocks > 1 || !pRsp->completed) {
qError("%s invalid scl value fetch rsp received, subQIdx:%d, rows:%" PRId64 ", blocks:%d, completed:%d",
ctx->idStr, pParam->subQIdx, pRsp->numOfRows, pRsp->numOfBlocks, pRsp->completed);
ctx->code = TSDB_CODE_PAR_INVALID_SCALAR_SUBQ_RES_ROWS;
return;
}
if (1 != pRsp->numOfCols && pRsp->numOfRows > 0) {
qError("%s invalid scl value fetch rsp received, subQIdx:%d, cols:%" PRId64, ctx->idStr, pParam->subQIdx, pRsp->numOfCols);
ctx->code = TSDB_CODE_PAR_INVALID_SCALAR_SUBQ_RES_COLS;
return;
}
if (0 == pRsp->numOfRows) {
pRemote->val.node.type = QUERY_NODE_VALUE;
pRemote->val.isNull = true;
pRemote->val.translate = true;
pRemote->val.flag &= (~VALUE_FLAG_VAL_UNSET);
taosArraySet(ctx->subResNodes, pParam->subQIdx, &pParam->pRes);
return;
}
ctx->code = createExprSubQResBlock(&pResBlock, &((SRemoteValueNode*)pParam->pRes)->val.node.resType, 1);
if (TSDB_CODE_SUCCESS == ctx->code) {
ctx->code = blockDataEnsureCapacity(pResBlock, 1);
}
if (TSDB_CODE_SUCCESS == ctx->code) {
ctx->code = extractSingleRspBlock(pRsp, pResBlock);
}
if (TSDB_CODE_SUCCESS == ctx->code) {
ctx->code = setValueFromResBlock(ctx, &pRemote->val, pResBlock);
}
if (TSDB_CODE_SUCCESS == ctx->code) {
pRemote->val.node.type = QUERY_NODE_VALUE;
taosArraySet(ctx->subResNodes, pParam->subQIdx, &pParam->pRes);
}
blockDataDestroy(pResBlock);
}
int32_t setValueListFromResBlock(STaskSubJobCtx* ctx, SRemoteValueListNode* pRes, SSDataBlock* pBlock) {
int32_t code = 0, lino = 0;
int32_t colNum = taosArrayGetSize(pBlock->pDataBlock);
if (NULL == pBlock->pDataBlock || 1 != colNum) {
qError("%s invalid scl fetch res block, pDataBlock:%p, colNum:%d", ctx->idStr, pBlock->pDataBlock, colNum);
return TSDB_CODE_QRY_EXECUTOR_INTERNAL_ERROR;
}
pRes->hasValue = true;
SColumnInfoData* pCol = taosArrayGet(pBlock->pDataBlock, 0);
TAOS_CHECK_EXIT(scalarBuildRemoteListHash(ctx->idStr, pRes, pCol, pBlock->info.rows));
_exit:
if (code) {
qError("%s %s failed with error: %s", ctx->idStr, __func__, tstrerror(code));
}
return code;
}
void handleRemoteValueListRes(SScalarFetchParam* pParam, STaskSubJobCtx* ctx, SRetrieveTableRsp* pRsp, bool* fetchDone) {
SSDataBlock* pResBlock = NULL;
SRemoteValueListNode* pRemote = (SRemoteValueListNode*)pParam->pRes;
SExecTaskInfo* pTaskInfo = ctx->pTaskInfo;
qDebug("%s scl fetch valueList rsp received, subQIdx:%d, rows:%" PRId64 , ctx->idStr, pParam->subQIdx, pRsp->numOfRows);
if (pRsp->numOfRows > 0) {
if (1 != pRsp->numOfCols) {
qError("%s invalid scl valueList fetch rsp received, subQIdx:%d, cols:%" PRId64, ctx->idStr, pParam->subQIdx, pRsp->numOfCols);
ctx->code = TSDB_CODE_PAR_INVALID_SCALAR_SUBQ_RES_COLS;
*fetchDone = true;
return;
}
ctx->code = createExprSubQResBlock(&pResBlock, &((SExprNode*)pParam->pRes)->resType, 1);
if (TSDB_CODE_SUCCESS == ctx->code) {
ctx->code = blockDataEnsureCapacity(pResBlock, pRsp->numOfRows);
}
if (TSDB_CODE_SUCCESS == ctx->code) {
ctx->code = extractSingleRspBlock(pRsp, pResBlock);
}
if (TSDB_CODE_SUCCESS == ctx->code) {
ctx->code = setValueListFromResBlock(ctx, pRemote, pResBlock);
}
if (TSDB_CODE_SUCCESS == ctx->code && pRsp->completed) {
pRemote->flag &= (~VALUELIST_FLAG_VAL_UNSET);
taosArraySet(ctx->subResNodes, pParam->subQIdx, &pParam->pRes);
}
blockDataDestroy(pResBlock);
} else if (0 == pRsp->numOfRows && pRsp->completed) {
if (!pRemote->hasValue) {
ctx->code = scalarBuildRemoteListHash(ctx->idStr, pRemote, NULL, 0);
}
if (TSDB_CODE_SUCCESS == ctx->code) {
pRemote->flag &= (~VALUELIST_FLAG_VAL_UNSET);
taosArraySet(ctx->subResNodes, pParam->subQIdx, &pParam->pRes);
}
}
if (IS_STREAM_MODE(pTaskInfo) && 0 == pRsp->numOfRows) {
if (!pRemote->hasValue) {
ctx->code = scalarBuildRemoteListHash(ctx->idStr, pRemote, NULL, 0);
}
if (TSDB_CODE_SUCCESS == ctx->code) {
pRemote->flag &= (~VALUELIST_FLAG_VAL_UNSET);
taosArraySet(ctx->subResNodes, pParam->subQIdx, &pParam->pRes);
}
pRsp->completed = true;
}
*fetchDone = (TSDB_CODE_SUCCESS != ctx->code || pRsp->completed) ? true : false;
if (!(*fetchDone)) {
int32_t code = sendFetchRemoteNodeReq(ctx, pParam->subQIdx, pParam->pRes, false);
if (TSDB_CODE_SUCCESS != code) {
ctx->code = code;
*fetchDone = true;
}
}
}
int32_t setRowHasNullFromResBlock(STaskSubJobCtx* ctx, bool* hasNull, SSDataBlock* pBlock) {
int32_t code = 0;
int32_t colNum = taosArrayGetSize(pBlock->pDataBlock);
if (2 != colNum) {
qError("%s invalid scl fetch res block, colNum:%d", ctx->idStr, colNum);
return TSDB_CODE_QRY_EXECUTOR_INTERNAL_ERROR;
}
SColumnInfoData* pCol = taosArrayGet(pBlock->pDataBlock, 1);
if (colDataIsNull_s(pCol, 0)) {
qError("%s invalid has_null res since it's null", ctx->idStr);
return TSDB_CODE_QRY_EXECUTOR_INTERNAL_ERROR;
} else {
*hasNull = *(bool*)colDataGetData(pCol, 0);
}
return code;
}
void handleRemoteRowRes(SScalarFetchParam* pParam, STaskSubJobCtx* ctx, SRetrieveTableRsp* pRsp, bool* fetchDone) {
SSDataBlock* pResBlock = NULL;
SExecTaskInfo* pTaskInfo = ctx->pTaskInfo;
qDebug("%s scl fetch row rsp received, subQIdx:%d, rows:%" PRId64 , ctx->idStr, pParam->subQIdx, pRsp->numOfRows);
SRemoteRowNode* pRemote = (SRemoteRowNode*)pParam->pRes;
if (IS_STREAM_MODE(pTaskInfo)) {
SNode** ppRes = taosArrayGet(ctx->subResNodes, pParam->subQIdx);
if (NULL == *ppRes && 0 == pRsp->numOfRows) {
pRemote->valSet = true;
pRemote->hasValue = false;
pRemote->hasNull = false;
pRemote->val.isNull = true;
pRemote->val.translate = true;
pRemote->val.flag &= (~VALUE_FLAG_VAL_UNSET);
taosArraySet(ctx->subResNodes, pParam->subQIdx, &pParam->pRes);
} else if (pRsp->numOfRows > 0 || pRsp->numOfBlocks > 0) {
if (2 != pRsp->numOfCols) {
qError("%s invalid scl fetch row rsp received, subQIdx:%d, cols:%" PRId64, ctx->idStr, pParam->subQIdx, pRsp->numOfCols);
ctx->code = TSDB_CODE_QRY_EXECUTOR_INTERNAL_ERROR;
*fetchDone = true;
return;
}
ctx->code = createExprSubQResBlock(&pResBlock, &pRemote->val.node.resType, 1);
if (TSDB_CODE_SUCCESS == ctx->code) {
SColumnInfoData idata = createColumnInfoData(TSDB_DATA_TYPE_BOOL, tDataTypes[TSDB_DATA_TYPE_BOOL].bytes, 0);
ctx->code = blockDataAppendColInfo(pResBlock, &idata);
}
if (TSDB_CODE_SUCCESS == ctx->code) {
ctx->code = blockDataEnsureCapacity(pResBlock, 1);
}
if (TSDB_CODE_SUCCESS == ctx->code) {
ctx->code = extractSingleRspBlock(pRsp, pResBlock);
}
if (TSDB_CODE_SUCCESS == ctx->code) {
ctx->code = setValueFromResBlock(ctx, &pRemote->val, pResBlock);
}
if (TSDB_CODE_SUCCESS == ctx->code) {
ctx->code = setRowHasNullFromResBlock(ctx, &pRemote->hasNull, pResBlock);
}
if (TSDB_CODE_SUCCESS == ctx->code) {
pRemote->valSet = true;
pRemote->hasValue = true;
pRemote->val.node.type = QUERY_NODE_REMOTE_ROW;
taosArraySet(ctx->subResNodes, pParam->subQIdx, &pParam->pRes);
}
blockDataDestroy(pResBlock);
} else if (NULL != *ppRes && 0 == pRsp->numOfRows) {
// EOF after data in stream mode: keep the node typed as REMOTE_ROW
// (do NOT rewrite to QUERY_NODE_VALUE) so the next per-event walker
// pass re-dispatches the case in sclInitParam and re-fires the
// fetch; otherwise > ANY / row subqueries would permanently replay
// the previous event's value.
pRsp->completed = true;
}
*fetchDone = (TSDB_CODE_SUCCESS != ctx->code || pRsp->completed) ? true : false;
if (!(*fetchDone)) {
ctx->code = sendFetchRemoteNodeReq(ctx, pParam->subQIdx, pParam->pRes, false);
if (TSDB_CODE_SUCCESS != ctx->code) {
*fetchDone = true;
}
}
return;
}
*fetchDone = true;
if (pRsp->numOfRows > 1 || pRsp->numOfBlocks > 1 || !pRsp->completed) {
qError("%s invalid scl fetch row rsp received, subQIdx:%d, rows:%" PRId64 ", blocks:%d, completed:%d",
ctx->idStr, pParam->subQIdx, pRsp->numOfRows, pRsp->numOfBlocks, pRsp->completed);
ctx->code = TSDB_CODE_PAR_INVALID_SCALAR_SUBQ_RES_ROWS;
return;
}
if (0 == pRsp->numOfRows) {
pRemote->valSet = true;
pRemote->hasValue = false;
pRemote->hasNull = false;
pRemote->val.isNull = true;
pRemote->val.translate = true;
pRemote->val.flag &= (~VALUE_FLAG_VAL_UNSET);
taosArraySet(ctx->subResNodes, pParam->subQIdx, &pParam->pRes);
return;
}
if (2 != pRsp->numOfCols) {
qError("%s invalid scl fetch row rsp received, subQIdx:%d, cols:%" PRId64, ctx->idStr, pParam->subQIdx, pRsp->numOfCols);
ctx->code = TSDB_CODE_QRY_EXECUTOR_INTERNAL_ERROR;
return;
}
ctx->code = createExprSubQResBlock(&pResBlock, &pRemote->val.node.resType, 1);
if (TSDB_CODE_SUCCESS == ctx->code) {
SColumnInfoData idata = createColumnInfoData(TSDB_DATA_TYPE_BOOL, tDataTypes[TSDB_DATA_TYPE_BOOL].bytes, 0);
ctx->code = blockDataAppendColInfo(pResBlock, &idata);
}
if (TSDB_CODE_SUCCESS == ctx->code) {
ctx->code = blockDataEnsureCapacity(pResBlock, 1);
}
if (TSDB_CODE_SUCCESS == ctx->code) {
ctx->code = extractSingleRspBlock(pRsp, pResBlock);
}
if (TSDB_CODE_SUCCESS == ctx->code) {
ctx->code = setValueFromResBlock(ctx, &pRemote->val, pResBlock);
}
if (TSDB_CODE_SUCCESS == ctx->code) {
ctx->code = setRowHasNullFromResBlock(ctx, &pRemote->hasNull, pResBlock);
}
if (TSDB_CODE_SUCCESS == ctx->code) {
taosArraySet(ctx->subResNodes, pParam->subQIdx, &pParam->pRes);
}
if (TSDB_CODE_SUCCESS == ctx->code) {
pRemote->valSet = true;
pRemote->hasValue = true;
}
blockDataDestroy(pResBlock);
}
int32_t setZeroRowsResValue(STaskSubJobCtx* ctx, SValueNode* pRes, int32_t rows) {
pRes->node.type = QUERY_NODE_VALUE;
pRes->flag &= (~VALUE_FLAG_VAL_UNSET);
pRes->translate = true;
return nodesSetValueNodeValue(pRes, &rows);
}
void handleRemoteZeroRowsRes(SScalarFetchParam* pParam, STaskSubJobCtx* ctx, SRetrieveTableRsp* pRsp, bool* fetchDone) {
SRemoteZeroRowsNode* pRemote = (SRemoteZeroRowsNode*)pParam->pRes;
SExecTaskInfo* pTaskInfo = ctx->pTaskInfo;
qDebug("%s scl fetch zeroRows rsp received, subQIdx:%d, rows:%" PRId64 , ctx->idStr, pParam->subQIdx, pRsp->numOfRows);
int32_t resRows = (pRsp->numOfRows > 0) ? 1 : 0;
if (resRows > 0 || pRsp->completed) {
ctx->code = setZeroRowsResValue(ctx, &pRemote->val, resRows);
if (TSDB_CODE_SUCCESS == ctx->code) {
taosArraySet(ctx->subResNodes, pParam->subQIdx, &pParam->pRes);
}
*fetchDone = true;
} else {
*fetchDone = false;
}
if (IS_STREAM_MODE(pTaskInfo) && 0 == pRsp->numOfRows) {
ctx->code = setZeroRowsResValue(ctx, &pRemote->val, 0);
if (TSDB_CODE_SUCCESS == ctx->code) {
taosArraySet(ctx->subResNodes, pParam->subQIdx, &pParam->pRes);
}
pRsp->completed = true;
*fetchDone = true;
}
if (!(*fetchDone)) {
int32_t code = sendFetchRemoteNodeReq(ctx, pParam->subQIdx, pParam->pRes, false);
if (TSDB_CODE_SUCCESS != code) {
ctx->code = code;
*fetchDone = true;
}
}
}
int32_t setTableFromResBlock(STaskSubJobCtx* ctx, SRemoteTableNode* pRes, SSDataBlock** ppBlock) {
int32_t code = 0, lino = 0;
if (NULL == pRes->pResBlks) {
pRes->pResBlks = taosArrayInit(10, POINTER_BYTES);
TSDB_CHECK_NULL(pRes->pResBlks, code, lino, _exit, terrno);
}
TSDB_CHECK_NULL(taosArrayPush(pRes->pResBlks, ppBlock), code, lino, _exit, terrno);
_exit:
if (code) {
blockDataDestroy(*ppBlock);
qError("%s failed at line %d since %s", __func__, lino, tstrerror(code));
}
*ppBlock = NULL;
return code;
}
void handleRemoteTableRes(SScalarFetchParam* pParam, STaskSubJobCtx* ctx, SRetrieveTableRsp* pRsp, bool* fetchDone) {
SSDataBlock* pResBlock = NULL;
SRemoteTableNode* pRemote = (SRemoteTableNode*)pParam->pRes;
qDebug("%s scl fetch remoteTable rsp received, subQIdx:%d, rows:%" PRId64 ", cols:%" PRId64, ctx->idStr, pParam->subQIdx, pRsp->numOfRows, pRsp->numOfCols);
if (pRsp->numOfRows > 0) {
ctx->code = createExprSubQResBlock(&pResBlock, NULL, pRemote->resCols);
if (TSDB_CODE_SUCCESS == ctx->code) {
ctx->code = extractSingleRspBlock(pRsp, pResBlock);
}
if (TSDB_CODE_SUCCESS == ctx->code) {
ctx->code = setTableFromResBlock(ctx, pRemote, &pResBlock);
}
if (TSDB_CODE_SUCCESS == ctx->code && pRsp->completed) {
pRemote->flag &= (~REMOTE_TABLE_FLAG_VAL_UNSET);
taosArraySet(ctx->subResNodes, pParam->subQIdx, &pParam->pRes);
}
blockDataDestroy(pResBlock);
} else if (0 == pRsp->numOfRows && pRsp->completed) {
pRemote->flag &= (~REMOTE_TABLE_FLAG_VAL_UNSET);
taosArraySet(ctx->subResNodes, pParam->subQIdx, &pParam->pRes);
}
*fetchDone = (TSDB_CODE_SUCCESS != ctx->code || pRsp->completed) ? true : false;
if (!(*fetchDone)) {
int32_t code = sendFetchRemoteNodeReq(ctx, pParam->subQIdx, pParam->pRes, false);
if (TSDB_CODE_SUCCESS != code) {
ctx->code = code;
*fetchDone = true;
}
}
}
int32_t remoteFetchCallBack(void* param, SDataBuf* pMsg, int32_t code) {
taosMemoryFreeClear(pMsg->pEpSet);
SScalarFetchParam* pParam = (SScalarFetchParam*)param;
STaskSubJobCtx* ctx = taosAcquireRef(fetchObjRefPool, pParam->subJobRefId);
if (ctx == NULL) {
qWarn("failed to acquire subJobCtx, since it may have been released, refId:%" PRIu64, pParam->subJobRefId);
taosMemoryFree(pMsg->pData);
return TSDB_CODE_SUCCESS;
}
// Copy idStr to a local buffer immediately. ctx->idStr points into the task
// memory pool which may be freed by doDestroyTask on a concurrent thread
// after the ref is acquired but before the log calls below execute.
char idStr[64];
tstrncpy(idStr, ctx->idStr, sizeof(idStr));
qDebug("%s subQIdx %d got rsp, blockIdx:%" PRId64 ", code:%d, rsp:%p", idStr, pParam->subQIdx, ctx->blockIdx, code, pMsg->pData);
if (ctx->transporterId > 0) {
int32_t ret = asyncFreeConnById(ctx->rpcHandle, ctx->transporterId);
if (ret != 0) {
qDebug("%s failed to free subQ rpc handle, code:%s, subQIdx:%d", idStr, tstrerror(ret), pParam->subQIdx);
}
ctx->transporterId = -1;
}
if (0 == code && NULL == pMsg->pData) {
qError("%s invalid rsp msg, msgType:%d, len:%d", idStr, pMsg->msgType, pMsg->len);
code = TSDB_CODE_QRY_INVALID_MSG;
}
if (code == TSDB_CODE_SUCCESS) {
SRetrieveTableRsp* pRsp = pMsg->pData;
pRsp->numOfRows = htobe64(pRsp->numOfRows);
pRsp->compLen = htonl(pRsp->compLen);
pRsp->payloadLen = htonl(pRsp->payloadLen);
pRsp->numOfCols = htonl(pRsp->numOfCols);
pRsp->useconds = htobe64(pRsp->useconds);
pRsp->numOfBlocks = htonl(pRsp->numOfBlocks);
qDebug("%s subQIdx %d blockIdx:%" PRIu64 " rsp detail, numOfBlocks:%d, numOfRows:%" PRId64 ", numOfCols:%" PRId64 ", completed:%d",
idStr, pParam->subQIdx, ctx->blockIdx, pRsp->numOfBlocks, pRsp->numOfRows, pRsp->numOfCols, pRsp->completed);
ctx->blockIdx++;
switch (nodeType(pParam->pRes)) {
case QUERY_NODE_REMOTE_VALUE: {
bool fetchDone = false;
handleRemoteValueRes(pParam, ctx, pRsp, &fetchDone);
qDebug("%s subQIdx %d handle remote value, fetchDone:%d", idStr, pParam->subQIdx, fetchDone);
if (!fetchDone) {
goto _exit;
}
break;
}
case QUERY_NODE_REMOTE_VALUE_LIST: {
bool fetchDone = false;
handleRemoteValueListRes(pParam, ctx, pRsp, &fetchDone);
qDebug("%s subQIdx %d handle remote value list finished, fetchDone:%d", idStr, pParam->subQIdx, fetchDone);
if (!fetchDone) {
goto _exit;
}
break;
}
case QUERY_NODE_REMOTE_ROW: {
bool fetchDone = false;
handleRemoteRowRes(pParam, ctx, pRsp, &fetchDone);
qDebug("%s subQIdx %d handle remote row finished, fetchDone:%d", idStr, pParam->subQIdx, fetchDone);
if (!fetchDone) {
goto _exit;
}
break;
}
case QUERY_NODE_REMOTE_ZERO_ROWS: {
bool fetchDone = false;
handleRemoteZeroRowsRes(pParam, ctx, pRsp, &fetchDone);
qDebug("%s subQIdx %d handle remote zeroRows finished, fetchDone:%d", idStr, pParam->subQIdx, fetchDone);
if (!fetchDone) {
goto _exit;
}
break;
}
case QUERY_NODE_REMOTE_TABLE: {
bool fetchDone = false;
handleRemoteTableRes(pParam, ctx, pRsp, &fetchDone);
qDebug("%s subQIdx %d handle remote table finished, fetchDone:%d", idStr, pParam->subQIdx, fetchDone);
if (!fetchDone) {
goto _exit;
}
break;
}
default:
qError("%s invalid scl fetch res node %d, subQIdx:%d", idStr, nodeType(pParam->pRes), pParam->subQIdx);
ctx->code = TSDB_CODE_QRY_EXECUTOR_INTERNAL_ERROR;
break;
}
} else {
ctx->code = rpcCvtErrCode(code);
if (ctx->code != code) {
qError("%s scl fetch rsp received, subQIdx:%d, error:%s, cvted error: %s", idStr, pParam->subQIdx,
tstrerror(code), tstrerror(ctx->code));
} else {
qError("%s scl fetch rsp received, subQIdx:%d, error:%s", idStr, pParam->subQIdx, tstrerror(code));
}
}
qDebug("%s subQIdx %d sem_post subQ ready", idStr, pParam->subQIdx);
code = tsem_post(&ctx->ready);
if (code != TSDB_CODE_SUCCESS) {
qError("failed to invoke post when scl fetch rsp is ready, code:%s", tstrerror(code));
}
_exit:
code = taosReleaseRef(fetchObjRefPool, pParam->subJobRefId);
if (code != TSDB_CODE_SUCCESS) {
qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(code));
}
taosMemoryFree(pMsg->pData);
return code;
}
#define STREAM_TASK_REPLICA_NUM 5
#define STREAM_TASK_DEPLOY_NUM 3
int32_t sendFetchRemoteNodeReq(STaskSubJobCtx* ctx, int32_t subQIdx, SNode* pRes, bool reset) {
int32_t code = TSDB_CODE_SUCCESS;
int32_t lino = 0;
int32_t innerIdx = 0;
SExecTaskInfo* pTaskInfo = ctx->pTaskInfo;
bool needStreamPesudoFuncVals = false;
if (IS_STREAM_MODE(pTaskInfo)) {
innerIdx = pTaskInfo->pStreamRuntimeInfo->execId / STREAM_TASK_REPLICA_NUM;
subQIdx = subQIdx * STREAM_TASK_DEPLOY_NUM + innerIdx;
}
SDownstreamSourceNode* pSource = (SDownstreamSourceNode*)taosArrayGetP(ctx->subEndPoints, subQIdx);
SResFetchReq req = {0};
req.header.vgId = pSource->addr.nodeId;
req.sId = pSource->sId;
req.clientId = pSource->clientId;
req.taskId = pSource->taskId;
req.srcTaskId = ctx->taskId;
req.blockIdx = ctx->blockIdx;
req.queryId = ctx->queryId;
req.execId = pSource->execId;
if (IS_STREAM_MODE(pTaskInfo)) {
req.queryId = pSource->clientId;
// Use deterministic execId from stream runtime (unique per reader task execution)
req.execId = pTaskInfo->pStreamRuntimeInfo->execId;
req.pStRtFuncInfo = &pTaskInfo->pStreamRuntimeInfo->funcInfo;
req.reset = reset;
needStreamPesudoFuncVals = true;
subQIdx = (subQIdx - innerIdx) / STREAM_TASK_DEPLOY_NUM;
}
int32_t msgSize = tSerializeSResFetchReq(NULL, 0, &req, needStreamPesudoFuncVals, false);
if (msgSize < 0) {
return msgSize;
}
void* msg = taosMemoryCalloc(1, msgSize);
if (NULL == msg) {
return terrno;
}
msgSize = tSerializeSResFetchReq(msg, msgSize, &req, needStreamPesudoFuncVals, false);
if (msgSize < 0) {
taosMemoryFree(msg);
return msgSize;
}
qDebug("%s scl build fetch msg and send to nodeId:%d, ep:%s, clientId:0x%" PRIx64 " taskId:0x%" PRIx64
", execId:%d, blockIdx:%" PRId64 ", reset:%d",
ctx->idStr, pSource->addr.nodeId, pSource->addr.epSet.eps[0].fqdn, pSource->clientId,
pSource->taskId, req.execId, req.blockIdx, req.reset);
// send the fetch remote task result reques
SMsgSendInfo* pMsgSendInfo = taosMemoryCalloc(1, sizeof(SMsgSendInfo));
if (NULL == pMsgSendInfo) {
taosMemoryFreeClear(msg);
qError("%s prepare message %d failed", ctx->idStr, (int32_t)sizeof(SMsgSendInfo));
return terrno;
}
SScalarFetchParam* param = taosMemoryCalloc(1, sizeof(SScalarFetchParam));
if (NULL == param) {
taosMemoryFreeClear(msg);
taosMemoryFreeClear(pMsgSendInfo);
qError("%s prepare param %d failed", ctx->idStr, (int32_t)sizeof(SScalarFetchParam));
return terrno;
}
if (ctx->code) {
qError("task has been killed, error:%s", tstrerror(ctx->code));
taosMemoryFree(param);
taosMemoryFreeClear(msg);
taosMemoryFreeClear(pMsgSendInfo);
code = ctx->code;
goto _end;
}
param->subQIdx = subQIdx;
param->pRes = pRes;
param->subJobRefId = ctx->subJobRefId;
pMsgSendInfo->param = param;
pMsgSendInfo->paramFreeFp = taosAutoMemoryFree;
pMsgSendInfo->msgInfo.pData = msg;
pMsgSendInfo->msgInfo.len = msgSize;
pMsgSendInfo->msgType = pSource->fetchMsgType;
pMsgSendInfo->fp = remoteFetchCallBack;
pMsgSendInfo->requestId = ctx->queryId;
code = asyncSendMsgToServer(ctx->rpcHandle, &pSource->addr.epSet, &ctx->transporterId, pMsgSendInfo);
QUERY_CHECK_CODE(code, lino, _end);
_end:
if (code != TSDB_CODE_SUCCESS) {
qError("%s %s failed at line %d since %s", ctx->idStr, __func__, lino, tstrerror(code));
}
return code;
}
int32_t fetchRemoteNodeImpl(STaskSubJobCtx* ctx, int32_t subQIdx, SNode* pRes) {
int32_t code = TSDB_CODE_SUCCESS;
int32_t lino = 0;
int32_t retryCount = 0;
const int32_t maxRetries = 100;
SExecTaskInfo* pTask = (SExecTaskInfo*)ctx->pTaskInfo;
bool blocking = false;
_retry:
ctx->blockIdx = 0;
ctx->code = TSDB_CODE_SUCCESS;
code = sendFetchRemoteNodeReq(ctx, subQIdx, pRes, true);
QUERY_CHECK_CODE(code, lino, _end);
if (!blocking && pTask->pWorkerCb) {
code = pTask->pWorkerCb->beforeBlocking(pTask->pWorkerCb->pPool);
if (code != TSDB_CODE_SUCCESS) {
pTask->code = code;
lino = __LINE__;
goto _end;
}
blocking = true;
}
int32_t waitCount = 0;
while (true) {
int32_t ret = tsem_timewait(&ctx->ready, 1000); // 1s timeout
if (ret == 0) {
break; // semaphore posted — got response
}
if (isTaskKilled(ctx->pTaskInfo)) {
qWarn("%s task killed while waiting for remote fetch response", ctx->idStr);
code = getTaskCode(ctx->pTaskInfo);
goto _after_wait;
}
waitCount++;
if (waitCount >= 100) {
qError("%s remote fetch response not received after %d seconds", ctx->idStr, waitCount);
code = TSDB_CODE_TIMEOUT_ERROR;
goto _after_wait;
}
}
_after_wait:
if (code == TSDB_CODE_SUCCESS) {
if (isTaskKilled(ctx->pTaskInfo)) {
code = getTaskCode(ctx->pTaskInfo);
} else {
code = ctx->code;
}
}
// If the runner slot is busy (another vnode's request is using it), retry
// after a short delay until the slot becomes free.
if (code == TSDB_CODE_STREAM_TASK_IVLD_STATUS && retryCount < maxRetries) {
retryCount++;
if (retryCount % 10 == 1) {
qDebug("%s subQIdx %d runner slot busy, retry %d", ctx->idStr, subQIdx, retryCount);
}
if (isTaskKilled(ctx->pTaskInfo)) {
code = getTaskCode(ctx->pTaskInfo);
goto _end;
}
taosMsleep(300 + taosRand() % 201);
code = TSDB_CODE_SUCCESS;
goto _retry;
}
_end:
if (blocking && pTask->pWorkerCb) {
int32_t ret = pTask->pWorkerCb->afterRecoverFromBlocking(pTask->pWorkerCb->pPool);
if (ret != TSDB_CODE_SUCCESS && code == TSDB_CODE_SUCCESS) {
code = ret;
pTask->code = code;
}
}
if (code != TSDB_CODE_SUCCESS) {
qError("%s %s failed at line %d since %s", ctx->idStr, __func__, lino, tstrerror(code));
}
return code;
}
int32_t remoteNodeCopy(SNode* pSrc, SNode* pDst) {
int32_t code = 0, lino = 0;
switch (nodeType(pSrc)) {
case QUERY_NODE_VALUE:
TAOS_CHECK_EXIT(valueNodeCopy((SValueNode*)pSrc, &((SRemoteValueNode*)pDst)->val));
((SRemoteValueNode*)pDst)->val.node.type = QUERY_NODE_VALUE;
break;
case QUERY_NODE_REMOTE_VALUE_LIST: {
SRemoteValueListNode* pDstNode = (SRemoteValueListNode*)pDst;
memcpy(pDst, pSrc, sizeof(SRemoteValueListNode));
pDstNode->hashAllocated = false;
break;
}
case QUERY_NODE_REMOTE_ROW: {
SRemoteRowNode* pRemote = (SRemoteRowNode*)pDst;
TAOS_CHECK_EXIT(valueNodeCopy((SValueNode*)pSrc, &pRemote->val));
pRemote->valSet = true;
pRemote->hasValue = ((SRemoteRowNode*)pSrc)->hasValue;
pRemote->hasNull = ((SRemoteRowNode*)pSrc)->hasNull;
break;
}
case QUERY_NODE_REMOTE_TABLE: {
SRemoteTableNode* pDstNode = (SRemoteTableNode*)pDst;
SRemoteTableNode* pSrcNode = (SRemoteTableNode*)pSrc;
pDstNode->pResBlks = pSrcNode->pResBlks;
pDstNode->flag = 0;
break;
}
case QUERY_NODE_REMOTE_ZERO_ROWS: {
SRemoteZeroRowsNode* pRemote = (SRemoteZeroRowsNode*)pDst;
TAOS_CHECK_EXIT(valueNodeCopy((SValueNode*)pSrc, &pRemote->val));
pRemote->val.node.type = QUERY_NODE_VALUE;
break;
}
default:
break;
}
_exit:
if (code) {
qError("%s failed at line %d since %s", __func__, lino, tstrerror(code));
}
return code;
}
int32_t qFetchRemoteNode(void* pCtx, int32_t subQIdx, SNode* pRes) {
STaskSubJobCtx* ctx = (STaskSubJobCtx*)pCtx;
int32_t code = 0, lino = 0;
int32_t subEndPoinsNum = taosArrayGetSize(ctx->subEndPoints);
if (subQIdx >= subEndPoinsNum) {
qError("%s invalid subQIdx %d, subEndPointsNum:%d", ctx->idStr, subQIdx, subEndPoinsNum);
return TSDB_CODE_QRY_SUBQ_NOT_FOUND;
}
// In stream mode the subquery must be re-evaluated for every trigger event;
// a cached SNode from an earlier event would replay a stale value and turn
// dynamic subqueries (e.g. WHERE ts >= (SELECT last_row(ts) FROM ref))
// into constants. Always go to the wire so req.reset=true reaches the
// vnode reader and rebuilds its task with the current ranges.
//
// Note: pRes here is a pointer into the operator's AST (the SRemoteValueNode
// placeholder), not a heap-owned node. In non-stream mode we still cache
// that borrowed pointer in subResNodes[] so subsequent calls memcpy from it.
// In stream mode the slot must not retain a stale pRes across trigger
// events or be treated as owning pRes: a stale entry would create a
// dangling alias once the AST is rebuilt, and freeing through that slot
// would double-free the AST node. Stream response handlers (e.g.
// handleRemoteValueRes) may still use subResNodes[] transiently during a
// single fetch to distinguish data from the completion sentinel; the
// requirement here is to clear the slot to NULL before each refetch so
// that only stale per-event state is invalidated, not to take ownership.
SNode** ppRes = taosArrayGet(ctx->subResNodes, subQIdx);
if (ctx->isStream) {
// Stream mode: re-evaluate subquery for every trigger event.
// Save previous slot value so we can restore it if the fetch fails.
// This prevents a failed fetch from leaving NULL in the slot, which
// would cause the next event's callback to misinterpret state as
// "first-call empty" if the error is swallowed upstream.
SNode* prevRes = ppRes ? *ppRes : NULL;
if (ppRes) {
*ppRes = NULL;
}
code = fetchRemoteNodeImpl(ctx, subQIdx, pRes);
if (code != TSDB_CODE_SUCCESS && ppRes) {
*ppRes = prevRes;
}
TAOS_CHECK_EXIT(code);
} else if (NULL == *ppRes) {
TAOS_CHECK_EXIT(fetchRemoteNodeImpl(ctx, subQIdx, pRes));
*ppRes = pRes;
} else {
TAOS_CHECK_EXIT(remoteNodeCopy(*ppRes, pRes));
}
_exit:
if (code) {
qError("%s %s failed at line %d since %s", ctx->idStr, __func__, lino, tstrerror(code));
} else {
qDebug("%s %s subQIdx %d succeed", ctx->idStr, __func__, subQIdx);
}
return code;
}
int32_t findDataBlockColIndexBySlotId(const SSDataBlock* pBlock, int32_t slotId) {
if (NULL == pBlock || NULL == pBlock->pDataBlock) {
return -1;
}
int32_t numOfCols = (int32_t)taosArrayGetSize(pBlock->pDataBlock);
if (slotId >= 0 && slotId < numOfCols) {
SColumnInfoData* pCol = taosArrayGet(pBlock->pDataBlock, slotId);
if (NULL != pCol && pCol->info.slotId == slotId) {
return slotId;
}
}
for (int32_t i = 0; i < numOfCols; ++i) {
SColumnInfoData* pCol = taosArrayGet(pBlock->pDataBlock, i);
if (NULL != pCol && pCol->info.slotId == slotId) {
return i;
}
}
return -1;
}
SColumnInfoData* getDataBlockColBySlotId(const SSDataBlock* pBlock, int32_t slotId, int32_t* pIndex) {
int32_t index = findDataBlockColIndexBySlotId(pBlock, slotId);
if (NULL != pIndex) {
*pIndex = index;
}
return (index >= 0) ? taosArrayGet(pBlock->pDataBlock, index) : NULL;
}
Reference in a new issue View git blame Copy permalink