Skip to content

T
TDengine

taosdata / TDengine
12 个月 前同步成功

通知 1181
Star 22014
Fork 4786
T
TDengine

体验新版 GitCode，发现更多精彩内容 >>

未验证 提交 e78e3031 编写于 作者: M Minglei Jin 提交者： GitHub
浏览文件
操作

Merge pull request #9769 from taosdata/fix/TS-1029-D 

[TS-1029]<fix>(query): added some memory allocation checks
上级 c68b6987 035d5157
隐藏空白更改
内联 并排
Showing with 584 addition and 40 deletion
src/client/src/tscServer.c
浏览文件 @ e78e3031
......@@ -1882,6 +1882,13 @@ int tscProcessRetrieveGlobalMergeRsp(SSqlObj *pSql) {
tscDebug("0x%"PRIx64" create QInfo 0x%"PRIx64" to execute query processing", pSql->self, pSql->self);
pQueryInfo->pQInfo = createQInfoFromQueryNode(pQueryInfo, &tableGroupInfo, NULL, NULL, pRes->pMerger, MERGE_STAGE, pSql->self);
if (pQueryInfo->pQInfo == NULL) {
taosHashCleanup(tableGroupInfo.map);
taosArrayDestroy(&group);
tscAsyncResultOnError(pSql);
pRes->code = TSDB_CODE_QRY_OUT_OF_MEMORY;
return pRes->code;
}
}
uint64_t localQueryId = pSql->self;
......
src/client/src/tscSubquery.c
浏览文件 @ e78e3031
......@@ -3902,8 +3902,11 @@ void* createQInfoFromQueryNode(SQueryInfo* pQueryInfo, STableGroupInfo* pTableGr
STsBufInfo bufInfo = {0};
SQueryParam param = {.pOperator = pa};
/*int32_t code = */initQInfo(&bufInfo, NULL, pSourceOperator, pQInfo, &param, NULL, 0, merger);
int32_t code = initQInfo(&bufInfo, NULL, pSourceOperator, pQInfo, &param, NULL, 0, merger);
taosArrayDestroy(&pa);
if (code != TSDB_CODE_SUCCESS) {
goto _cleanup;
}
return pQInfo;
......
src/query/src/qExecutor.c
浏览文件 @ e78e3031
......@@ -340,9 +340,17 @@ SSDataBlock* createOutputBuf(SExprInfo* pExpr, int32_t numOfOutput, int32_t numO
const static int32_t minSize = 8;
SSDataBlock *res = calloc(1, sizeof(SSDataBlock));
res->info.numOfCols = numOfOutput;
if (res == NULL) {
qError("failed to allocate for output buffer");
goto _clean;
}
res->pDataBlock = taosArrayInit(numOfOutput, sizeof(SColumnInfoData));
if (res->pDataBlock == NULL) {
qError("failed to init arrary for data block of output buffer");
goto _clean;
}
for (int32_t i = 0; i < numOfOutput; ++i) {
SColumnInfoData idata = {{0}};
idata.info.type = pExpr[i].base.resType;
......@@ -351,10 +359,20 @@ SSDataBlock* createOutputBuf(SExprInfo* pExpr, int32_t numOfOutput, int32_t numO
int32_t size = MAX(idata.info.bytes * numOfRows, minSize);
idata.pData = calloc(1, size); // at least to hold a pointer on x64 platform
if (idata.pData == NULL) {
qError("failed to allocate column buffer for output buffer");
goto _clean;
}
taosArrayPush(res->pDataBlock, &idata);
res->info.numOfCols++;
}
return res;
_clean:
destroyOutputBuf(res);
return NULL;
}
void* destroyOutputBuf(SSDataBlock* pBlock) {
......@@ -1810,15 +1828,15 @@ static bool functionNeedToExecute(SQueryRuntimeEnv *pRuntimeEnv, SQLFunctionCtx
if (functionId == TSDB_FUNC_FIRST_DST || functionId == TSDB_FUNC_FIRST) {
// if param[2] is set value, input data come from client, order is no relation with pQueryAttr->order, so always return true
if(pCtx->param[2].nType == TSDB_DATA_TYPE_INT)
return true;
return true;
return QUERY_IS_ASC_QUERY(pQueryAttr);
}
// denote the order type
if ((functionId == TSDB_FUNC_LAST_DST || functionId == TSDB_FUNC_LAST)) {
// if param[2] is set value, input data come from client, order is no relation with pQueryAttr->order , so always return true
// if param[2] is set value, input data come from client, order is no relation with pQueryAttr->order, so always return true
if(pCtx->param[2].nType == TSDB_DATA_TYPE_INT)
return true;
return true;
return pCtx->param[0].i64 == pQueryAttr->order.order;
}
......@@ -2072,17 +2090,26 @@ static int32_t setupQueryRuntimeEnv(SQueryRuntimeEnv *pRuntimeEnv, int32_t numOf
switch (*op) {
case OP_TagScan: {
pRuntimeEnv->proot = createTagScanOperatorInfo(pRuntimeEnv, pQueryAttr->pExpr1, pQueryAttr->numOfOutput);
if (pRuntimeEnv->proot == NULL) {
goto _clean;
}
break;
}
case OP_MultiTableTimeInterval: {
pRuntimeEnv->proot =
createMultiTableTimeIntervalOperatorInfo(pRuntimeEnv, pRuntimeEnv->proot, pQueryAttr->pExpr1, pQueryAttr->numOfOutput);
if (pRuntimeEnv->proot == NULL) {
goto _clean;
}
setTableScanFilterOperatorInfo(pRuntimeEnv->proot->upstream[0]->info, pRuntimeEnv->proot);
break;
}
case OP_TimeWindow: {
pRuntimeEnv->proot =
createTimeIntervalOperatorInfo(pRuntimeEnv, pRuntimeEnv->proot, pQueryAttr->pExpr1, pQueryAttr->numOfOutput);
if (pRuntimeEnv->proot == NULL) {
goto _clean;
}
int32_t opType = pRuntimeEnv->proot->upstream[0]->operatorType;
if (opType != OP_DummyInput && opType != OP_Join) {
setTableScanFilterOperatorInfo(pRuntimeEnv->proot->upstream[0]->info, pRuntimeEnv->proot);
......@@ -2092,6 +2119,9 @@ static int32_t setupQueryRuntimeEnv(SQueryRuntimeEnv *pRuntimeEnv, int32_t numOf
case OP_TimeEvery: {
pRuntimeEnv->proot =
createTimeEveryOperatorInfo(pRuntimeEnv, pRuntimeEnv->proot, pQueryAttr->pExpr1, pQueryAttr->numOfOutput);
if (pRuntimeEnv->proot == NULL) {
goto _clean;
}
int32_t opType = pRuntimeEnv->proot->upstream[0]->operatorType;
if (opType != OP_DummyInput && opType != OP_Join) {
setTableScanFilterOperatorInfo(pRuntimeEnv->proot->upstream[0]->info, pRuntimeEnv->proot);
......@@ -2101,7 +2131,9 @@ static int32_t setupQueryRuntimeEnv(SQueryRuntimeEnv *pRuntimeEnv, int32_t numOf
case OP_Groupby: {
pRuntimeEnv->proot =
createGroupbyOperatorInfo(pRuntimeEnv, pRuntimeEnv->proot, pQueryAttr->pExpr1, pQueryAttr->numOfOutput);
if (pRuntimeEnv->proot == NULL) {
goto _clean;
}
int32_t opType = pRuntimeEnv->proot->upstream[0]->operatorType;
if (opType != OP_DummyInput) {
setTableScanFilterOperatorInfo(pRuntimeEnv->proot->upstream[0]->info, pRuntimeEnv->proot);
......@@ -2111,6 +2143,9 @@ static int32_t setupQueryRuntimeEnv(SQueryRuntimeEnv *pRuntimeEnv, int32_t numOf
case OP_SessionWindow: {
pRuntimeEnv->proot =
createSWindowOperatorInfo(pRuntimeEnv, pRuntimeEnv->proot, pQueryAttr->pExpr1, pQueryAttr->numOfOutput);
if (pRuntimeEnv->proot == NULL) {
goto _clean;
}
int32_t opType = pRuntimeEnv->proot->upstream[0]->operatorType;
if (opType != OP_DummyInput) {
setTableScanFilterOperatorInfo(pRuntimeEnv->proot->upstream[0]->info, pRuntimeEnv->proot);
......@@ -2120,13 +2155,18 @@ static int32_t setupQueryRuntimeEnv(SQueryRuntimeEnv *pRuntimeEnv, int32_t numOf
case OP_MultiTableAggregate: {
pRuntimeEnv->proot =
createMultiTableAggOperatorInfo(pRuntimeEnv, pRuntimeEnv->proot, pQueryAttr->pExpr1, pQueryAttr->numOfOutput);
if (pRuntimeEnv->proot == NULL) {
goto _clean;
}
setTableScanFilterOperatorInfo(pRuntimeEnv->proot->upstream[0]->info, pRuntimeEnv->proot);
break;
}
case OP_Aggregate: {
pRuntimeEnv->proot =
createAggregateOperatorInfo(pRuntimeEnv, pRuntimeEnv->proot, pQueryAttr->pExpr1, pQueryAttr->numOfOutput);
if (pRuntimeEnv->proot == NULL) {
goto _clean;
}
int32_t opType = pRuntimeEnv->proot->upstream[0]->operatorType;
if (opType != OP_DummyInput && opType != OP_Join) {
setTableScanFilterOperatorInfo(pRuntimeEnv->proot->upstream[0]->info, pRuntimeEnv->proot);
......@@ -2146,11 +2186,18 @@ static int32_t setupQueryRuntimeEnv(SQueryRuntimeEnv *pRuntimeEnv, int32_t numOf
assert(pQueryAttr->pExpr2 != NULL);
pRuntimeEnv->proot = createProjectOperatorInfo(pRuntimeEnv, prev, pQueryAttr->pExpr2, pQueryAttr->numOfExpr2);
}
if (pRuntimeEnv->proot == NULL) {
goto _clean;
}
break;
}
case OP_StateWindow: {
pRuntimeEnv->proot = createStatewindowOperatorInfo(pRuntimeEnv, pRuntimeEnv->proot, pQueryAttr->pExpr1, pQueryAttr->numOfOutput);
pRuntimeEnv->proot = createStatewindowOperatorInfo(pRuntimeEnv, pRuntimeEnv->proot, pQueryAttr->pExpr1, pQueryAttr->numOfOutput);
if (pRuntimeEnv->proot == NULL) {
goto _clean;
}
int32_t opType = pRuntimeEnv->proot->upstream[0]->operatorType;
if (opType != OP_DummyInput) {
setTableScanFilterOperatorInfo(pRuntimeEnv->proot->upstream[0]->info, pRuntimeEnv->proot);
......@@ -2160,6 +2207,9 @@ static int32_t setupQueryRuntimeEnv(SQueryRuntimeEnv *pRuntimeEnv, int32_t numOf
case OP_Limit: {
pRuntimeEnv->proot = createLimitOperatorInfo(pRuntimeEnv, pRuntimeEnv->proot);
if (pRuntimeEnv->proot == NULL) {
goto _clean;
}
break;
}
......@@ -2171,12 +2221,18 @@ static int32_t setupQueryRuntimeEnv(SQueryRuntimeEnv *pRuntimeEnv, int32_t numOf
pRuntimeEnv->proot = createFilterOperatorInfo(pRuntimeEnv, pRuntimeEnv->proot, pQueryAttr->pExpr3,
pQueryAttr->numOfExpr3, pColInfo, numOfFilterCols);
freeColumnInfo(pColInfo, pQueryAttr->numOfExpr3);
if (pRuntimeEnv->proot == NULL) {
goto _clean;
}
} else {
SColumnInfo* pColInfo =
extractColumnFilterInfo(pQueryAttr->pExpr1, pQueryAttr->numOfOutput, &numOfFilterCols);
pRuntimeEnv->proot = createFilterOperatorInfo(pRuntimeEnv, pRuntimeEnv->proot, pQueryAttr->pExpr1,
pQueryAttr->numOfOutput, pColInfo, numOfFilterCols);
freeColumnInfo(pColInfo, pQueryAttr->numOfOutput);
if (pRuntimeEnv->proot == NULL) {
goto _clean;
}
}
break;
......@@ -2185,11 +2241,17 @@ static int32_t setupQueryRuntimeEnv(SQueryRuntimeEnv *pRuntimeEnv, int32_t numOf
case OP_Fill: {
SOperatorInfo* pInfo = pRuntimeEnv->proot;
pRuntimeEnv->proot = createFillOperatorInfo(pRuntimeEnv, pInfo, pInfo->pExpr, pInfo->numOfOutput, pQueryAttr->multigroupResult);
if (pRuntimeEnv->proot == NULL) {
goto _clean;
}
break;
}
case OP_MultiwayMergeSort: {
pRuntimeEnv->proot = createMultiwaySortOperatorInfo(pRuntimeEnv, pQueryAttr->pExpr1, pQueryAttr->numOfOutput, 200, merger); // TD-10899
if (pRuntimeEnv->proot == NULL) {
goto _clean;
}
break;
}
......@@ -2201,6 +2263,9 @@ static int32_t setupQueryRuntimeEnv(SQueryRuntimeEnv *pRuntimeEnv, int32_t numOf
pRuntimeEnv->proot = createGlobalAggregateOperatorInfo(pRuntimeEnv, pRuntimeEnv->proot, pQueryAttr->pExpr3,
pQueryAttr->numOfExpr3, merger, pQueryAttr->pUdfInfo, multigroupResult);
if (pRuntimeEnv->proot == NULL) {
goto _clean;
}
break;
}
......@@ -2208,11 +2273,17 @@ static int32_t setupQueryRuntimeEnv(SQueryRuntimeEnv *pRuntimeEnv, int32_t numOf
int32_t num = pRuntimeEnv->proot->numOfOutput;
SExprInfo* pExpr = pRuntimeEnv->proot->pExpr;
pRuntimeEnv->proot = createSLimitOperatorInfo(pRuntimeEnv, pRuntimeEnv->proot, pExpr, num, merger, pQueryAttr->multigroupResult);
if (pRuntimeEnv->proot == NULL) {
goto _clean;
}
break;
}
case OP_Distinct: {
pRuntimeEnv->proot = createDistinctOperatorInfo(pRuntimeEnv, pRuntimeEnv->proot, pQueryAttr->pExpr1, pQueryAttr->numOfOutput);
if (pRuntimeEnv->proot == NULL) {
goto _clean;
}
break;
}
......@@ -2224,6 +2295,9 @@ static int32_t setupQueryRuntimeEnv(SQueryRuntimeEnv *pRuntimeEnv, int32_t numOf
pRuntimeEnv->proot = createOrderOperatorInfo(pRuntimeEnv, pRuntimeEnv->proot, pQueryAttr->pExpr1,
pQueryAttr->numOfOutput, &pQueryAttr->order);
}
if (pRuntimeEnv->proot == NULL) {
goto _clean;
}
break;
}
......@@ -4824,18 +4898,30 @@ int32_t doInitQInfo(SQInfo* pQInfo, STSBuf* pTsBuf, void* tsdb, void* sourceOptr
switch(tbScanner) {
case OP_TableBlockInfoScan: {
pRuntimeEnv->proot = createTableBlockInfoScanOperator(pRuntimeEnv->pQueryHandle, pRuntimeEnv);
if (pRuntimeEnv->proot == NULL) {
return TSDB_CODE_QRY_OUT_OF_MEMORY;
}
break;
}
case OP_TableSeqScan: {
pRuntimeEnv->proot = createTableSeqScanOperator(pRuntimeEnv->pQueryHandle, pRuntimeEnv);
if (pRuntimeEnv->proot == NULL) {
return TSDB_CODE_QRY_OUT_OF_MEMORY;
}
break;
}
case OP_DataBlocksOptScan: {
pRuntimeEnv->proot = createDataBlocksOptScanInfo(pRuntimeEnv->pQueryHandle, pRuntimeEnv, getNumOfScanTimes(pQueryAttr), pQueryAttr->needReverseScan? 1:0);
if (pRuntimeEnv->proot == NULL) {
return TSDB_CODE_QRY_OUT_OF_MEMORY;
}
break;
}
case OP_TableScan: {
pRuntimeEnv->proot = createTableScanOperator(pRuntimeEnv->pQueryHandle, pRuntimeEnv, getNumOfScanTimes(pQueryAttr));
if (pRuntimeEnv->proot == NULL) {
return TSDB_CODE_QRY_OUT_OF_MEMORY;
}
break;
}
default: { // do nothing
......@@ -5152,6 +5238,10 @@ SOperatorInfo* createTableScanOperator(void* pTsdbQueryHandle, SQueryRuntimeEnv*
assert(repeatTime > 0);
STableScanInfo* pInfo = calloc(1, sizeof(STableScanInfo));
if (pInfo == NULL) {
return NULL;
}
pInfo->pQueryHandle = pTsdbQueryHandle;
pInfo->times = repeatTime;
pInfo->reverseTimes = 0;
......@@ -5159,6 +5249,11 @@ SOperatorInfo* createTableScanOperator(void* pTsdbQueryHandle, SQueryRuntimeEnv*
pInfo->current = 0;
SOperatorInfo* pOperator = calloc(1, sizeof(SOperatorInfo));
if (pOperator == NULL) {
tfree(pInfo);
return NULL;
}
pOperator->name = "TableScanOperator";
pOperator->operatorType = OP_TableScan;
pOperator->blockingOptr = false;
......@@ -5173,6 +5268,9 @@ SOperatorInfo* createTableScanOperator(void* pTsdbQueryHandle, SQueryRuntimeEnv*
SOperatorInfo* createTableSeqScanOperator(void* pTsdbQueryHandle, SQueryRuntimeEnv* pRuntimeEnv) {
STableScanInfo* pInfo = calloc(1, sizeof(STableScanInfo));
if (pInfo == NULL) {
return NULL;
}
pInfo->pQueryHandle = pTsdbQueryHandle;
pInfo->times = 1;
......@@ -5183,6 +5281,11 @@ SOperatorInfo* createTableSeqScanOperator(void* pTsdbQueryHandle, SQueryRuntimeE
pRuntimeEnv->enableGroupData = true;
SOperatorInfo* pOperator = calloc(1, sizeof(SOperatorInfo));
if (pOperator == NULL) {
tfree(pInfo);
return NULL;
}
pOperator->name = "TableSeqScanOperator";
pOperator->operatorType = OP_TableSeqScan;
pOperator->blockingOptr = false;
......@@ -5197,9 +5300,15 @@ SOperatorInfo* createTableSeqScanOperator(void* pTsdbQueryHandle, SQueryRuntimeE
SOperatorInfo* createTableBlockInfoScanOperator(void* pTsdbQueryHandle, SQueryRuntimeEnv* pRuntimeEnv) {
STableScanInfo* pInfo = calloc(1, sizeof(STableScanInfo));
if (pInfo == NULL) {
return NULL;
}
pInfo->pQueryHandle = pTsdbQueryHandle;
pInfo->block.pDataBlock = taosArrayInit(1, sizeof(SColumnInfoData));
if (pInfo->block.pDataBlock == NULL) {
goto _clean;
}
SColumnInfoData infoData = {{0}};
infoData.info.type = TSDB_DATA_TYPE_BINARY;
......@@ -5208,6 +5317,11 @@ SOperatorInfo* createTableBlockInfoScanOperator(void* pTsdbQueryHandle, SQueryRu
taosArrayPush(pInfo->block.pDataBlock, &infoData);
SOperatorInfo* pOperator = calloc(1, sizeof(SOperatorInfo));
if (pOperator == NULL) {
taosArrayDestroy(&pInfo->block.pDataBlock);
goto _clean;
}
pOperator->name = "TableBlockInfoScanOperator";
pOperator->operatorType = OP_TableBlockInfoScan;
pOperator->blockingOptr = false;
......@@ -5218,6 +5332,11 @@ SOperatorInfo* createTableBlockInfoScanOperator(void* pTsdbQueryHandle, SQueryRu
pOperator->exec = doBlockInfoScan;
return pOperator;
_clean:
tfree(pInfo);
return NULL;
}
void setTableScanFilterOperatorInfo(STableScanInfo* pTableScanInfo, SOperatorInfo* pDownstream) {
......@@ -5285,6 +5404,10 @@ SOperatorInfo* createDataBlocksOptScanInfo(void* pTsdbQueryHandle, SQueryRuntime
assert(repeatTime > 0);
STableScanInfo* pInfo = calloc(1, sizeof(STableScanInfo));
if (pInfo == NULL) {
return NULL;
}
pInfo->pQueryHandle = pTsdbQueryHandle;
pInfo->times = repeatTime;
pInfo->reverseTimes = reverseTime;
......@@ -5296,6 +5419,11 @@ SOperatorInfo* createDataBlocksOptScanInfo(void* pTsdbQueryHandle, SQueryRuntime
}
SOperatorInfo* pOptr = calloc(1, sizeof(SOperatorInfo));
if (pOptr == NULL) {
tfree(pInfo);
return NULL;
}
pOptr->name = "DataBlocksOptimizedScanOperator";
pOptr->operatorType = OP_DataBlocksOptScan;
pOptr->pRuntimeEnv = pRuntimeEnv;
......@@ -5317,6 +5445,10 @@ SArray* getOrderCheckColumns(SQueryAttr* pQuery) {
pOrderColumns = taosArrayInit(4, sizeof(SColIndex));
}
if (pOrderColumns == NULL) {
return NULL;
}
if (pQuery->interval.interval > 0) {
if (pOrderColumns == NULL) {
pOrderColumns = taosArrayInit(1, sizeof(SColIndex));
......@@ -5356,7 +5488,11 @@ SArray* getResultGroupCheckColumns(SQueryAttr* pQuery) {
pOrderColumns = taosArrayInit(4, sizeof(SColIndex));
}
for(int32_t i = 0; i < numOfCols; ++i) {
if (pOrderColumns == NULL) {
return NULL;
}
for (int32_t i = 0; i < numOfCols; ++i) {
SColIndex* index = taosArrayGet(pOrderColumns, i);
bool found = false;
......@@ -5384,21 +5520,45 @@ static void destroyGlobalAggOperatorInfo(void* param, int32_t numOfOutput) {
SMultiwayMergeInfo *pInfo = (SMultiwayMergeInfo*) param;
destroyBasicOperatorInfo(&pInfo->binfo, numOfOutput);
taosArrayDestroy(&pInfo->orderColumnList);
taosArrayDestroy(&pInfo->groupColumnList);
tfree(pInfo->prevRow);
tfree(pInfo->currentGroupColData);
if (pInfo->orderColumnList) {
taosArrayDestroy(&pInfo->orderColumnList);
}
if (pInfo->groupColumnList) {
taosArrayDestroy(&pInfo->groupColumnList);
}
if (pInfo->prevRow) {
tfree(pInfo->prevRow);
}
if (pInfo->currentGroupColData) {
tfree(pInfo->currentGroupColData);
}
}
static void destroySlimitOperatorInfo(void* param, int32_t numOfOutput) {
SSLimitOperatorInfo *pInfo = (SSLimitOperatorInfo*) param;
taosArrayDestroy(&pInfo->orderColumnList);
pInfo->pRes = destroyOutputBuf(pInfo->pRes);
tfree(pInfo->prevRow);
if (pInfo->orderColumnList) {
taosArrayDestroy(&pInfo->orderColumnList);
}
if (pInfo->pRes) {
pInfo->pRes = destroyOutputBuf(pInfo->pRes);
}
if (pInfo->prevRow) {
tfree(pInfo->prevRow);
}
}
SOperatorInfo* createGlobalAggregateOperatorInfo(SQueryRuntimeEnv* pRuntimeEnv, SOperatorInfo* upstream,
SExprInfo* pExpr, int32_t numOfOutput, void* param, SArray* pUdfInfo, bool groupResultMixedUp) {
SMultiwayMergeInfo* pInfo = calloc(1, sizeof(SMultiwayMergeInfo));
if (pInfo == NULL) {
return NULL;
}
pInfo->resultRowFactor =
(int32_t)(getRowNumForMultioutput(pRuntimeEnv->pQueryAttr, pRuntimeEnv->pQueryAttr->topBotQuery, false));
......@@ -5414,6 +5574,10 @@ SOperatorInfo* createGlobalAggregateOperatorInfo(SQueryRuntimeEnv* pRuntimeEnv,
pInfo->orderColumnList = getOrderCheckColumns(pRuntimeEnv->pQueryAttr);
pInfo->groupColumnList = getResultGroupCheckColumns(pRuntimeEnv->pQueryAttr);
if (pInfo->binfo.pRes == NULL || pInfo->binfo.pCtx == NULL || pInfo->orderColumnList == NULL || pInfo->groupColumnList == NULL) {
goto _clean;
}
// TODO refactor
int32_t len = 0;
for(int32_t i = 0; i < numOfOutput; ++i) {
......@@ -5433,6 +5597,10 @@ SOperatorInfo* createGlobalAggregateOperatorInfo(SQueryRuntimeEnv* pRuntimeEnv,
numOfCols = (pInfo->groupColumnList != NULL)? (int32_t)taosArrayGetSize(pInfo->groupColumnList):0;
pInfo->currentGroupColData = calloc(1, (POINTER_BYTES * numOfCols + len));
if (pInfo->currentGroupColData == NULL) {
goto _clean;
}
offset = POINTER_BYTES * numOfCols;
for(int32_t i = 0; i < numOfCols; ++i) {
......@@ -5443,11 +5611,18 @@ SOperatorInfo* createGlobalAggregateOperatorInfo(SQueryRuntimeEnv* pRuntimeEnv,
}
initResultRowInfo(&pInfo->binfo.resultRowInfo, 8, TSDB_DATA_TYPE_INT);
if (pInfo->binfo.resultRowInfo.pResult == NULL) {
goto _clean;
}
pInfo->seed = rand();
setDefaultOutputBuf(pRuntimeEnv, &pInfo->binfo, pInfo->seed, MERGE_STAGE);
SOperatorInfo* pOperator = calloc(1, sizeof(SOperatorInfo));
if (pOperator == NULL) {
return NULL;
}
pOperator->name = "GlobalAggregate";
pOperator->operatorType = OP_GlobalAggregate;
pOperator->blockingOptr = true;
......@@ -5462,17 +5637,30 @@ SOperatorInfo* createGlobalAggregateOperatorInfo(SQueryRuntimeEnv* pRuntimeEnv,
appendUpstream(pOperator, upstream);
return pOperator;
_clean:
destroyGlobalAggOperatorInfo((void *) pInfo, numOfOutput);
tfree(pInfo);
return NULL;
}
SOperatorInfo *createMultiwaySortOperatorInfo(SQueryRuntimeEnv *pRuntimeEnv, SExprInfo *pExpr, int32_t numOfOutput,
int32_t numOfRows, void *merger) {
SMultiwayMergeInfo* pInfo = calloc(1, sizeof(SMultiwayMergeInfo));
if (pInfo == NULL) {
return NULL;
}
pInfo->pMerge = merger;
pInfo->bufCapacity = numOfRows;
pInfo->orderColumnList = getResultGroupCheckColumns(pRuntimeEnv->pQueryAttr);
pInfo->binfo.pRes = createOutputBuf(pExpr, numOfOutput, numOfRows);
if (pInfo->orderColumnList == NULL || pInfo->binfo.pRes == NULL) {
goto _clean;
}
{ // todo extract method to create prev compare buffer
int32_t len = 0;
for(int32_t i = 0; i < numOfOutput; ++i) {
......@@ -5492,6 +5680,10 @@ SOperatorInfo *createMultiwaySortOperatorInfo(SQueryRuntimeEnv *pRuntimeEnv, SEx
}
SOperatorInfo* pOperator = calloc(1, sizeof(SOperatorInfo));
if (pOperator == NULL) {
goto _clean;
}
pOperator->name = "MultiwaySortOperator";
pOperator->operatorType = OP_MultiwayMergeSort;
pOperator->blockingOptr = false;
......@@ -5503,6 +5695,12 @@ SOperatorInfo *createMultiwaySortOperatorInfo(SQueryRuntimeEnv *pRuntimeEnv, SEx
pOperator->exec = doMultiwayMergeSort;
pOperator->cleanup = destroyGlobalAggOperatorInfo;
return pOperator;
_clean:
destroyGlobalAggOperatorInfo((void *)pInfo, numOfOutput);
tfree(pInfo);
return NULL;
}
static int32_t doMergeSDatablock(SSDataBlock* pDest, SSDataBlock* pSrc) {
......@@ -5579,11 +5777,22 @@ static SSDataBlock* doSort(void* param, bool* newgroup) {
SOperatorInfo *createOrderOperatorInfo(SQueryRuntimeEnv* pRuntimeEnv, SOperatorInfo* upstream, SExprInfo* pExpr, int32_t numOfOutput, SOrderVal* pOrderVal) {
SOrderOperatorInfo* pInfo = calloc(1, sizeof(SOrderOperatorInfo));
if (pInfo == NULL) {
return NULL;
}
{
SSDataBlock* pDataBlock = calloc(1, sizeof(SSDataBlock));
if (pDataBlock == NULL) {
goto _clean;
}
pDataBlock->pDataBlock = taosArrayInit(numOfOutput, sizeof(SColumnInfoData));
for(int32_t i = 0; i < numOfOutput; ++i) {
if (pDataBlock->pDataBlock == NULL) {
goto _clean;
}
for (int32_t i = 0; i < numOfOutput; ++i) {
SColumnInfoData col = {{0}};
col.info.colId = pExpr[i].base.colInfo.colId;
col.info.bytes = pExpr[i].base.resBytes;
......@@ -5601,6 +5810,10 @@ SOperatorInfo *createOrderOperatorInfo(SQueryRuntimeEnv* pRuntimeEnv, SOperatorI
}
SOperatorInfo* pOperator = calloc(1, sizeof(SOperatorInfo));
if (pOperator == NULL) {
goto _clean;
}
pOperator->name = "InMemoryOrder";
pOperator->operatorType = OP_Order;
pOperator->blockingOptr = true;
......@@ -5612,6 +5825,12 @@ SOperatorInfo *createOrderOperatorInfo(SQueryRuntimeEnv* pRuntimeEnv, SOperatorI
appendUpstream(pOperator, upstream);
return pOperator;
_clean:
destroyOrderOperatorInfo((void *)pInfo, numOfOutput);
tfree(pInfo);
return NULL;
}
static int32_t getTableScanOrder(STableScanInfo* pTableScanInfo) {
......@@ -5891,8 +6110,7 @@ static SSDataBlock* doLimit(void* param, bool* newgroup) {
if (pRuntimeEnv->currentOffset == 0) {
break;
}
else if(srows > 0) {
} else if(srows > 0) {
if(pRuntimeEnv->currentOffset - srows >= pBlock->info.rows) {
pRuntimeEnv->currentOffset -= pBlock->info.rows;
} else {
......@@ -6982,6 +7200,9 @@ static void destroyOperatorInfo(SOperatorInfo* pOperator) {
SOperatorInfo* createAggregateOperatorInfo(SQueryRuntimeEnv* pRuntimeEnv, SOperatorInfo* upstream, SExprInfo* pExpr, int32_t numOfOutput) {
SAggOperatorInfo* pInfo = calloc(1, sizeof(SAggOperatorInfo));
if (pInfo == NULL) {
return NULL;
}
SQueryAttr* pQueryAttr = pRuntimeEnv->pQueryAttr;
int32_t numOfRows = (int32_t)(getRowNumForMultioutput(pQueryAttr, pQueryAttr->topBotQuery, pQueryAttr->stableQuery));
......@@ -6991,10 +7212,18 @@ SOperatorInfo* createAggregateOperatorInfo(SQueryRuntimeEnv* pRuntimeEnv, SOpera
initResultRowInfo(&pInfo->binfo.resultRowInfo, 8, TSDB_DATA_TYPE_INT);
if (pInfo->binfo.pRes == NULL || pInfo->binfo.pCtx == NULL || pInfo->binfo.resultRowInfo.pResult == NULL) {
goto _clean;
}
pInfo->seed = rand();
setDefaultOutputBuf(pRuntimeEnv, &pInfo->binfo, pInfo->seed, MASTER_SCAN);
SOperatorInfo* pOperator = calloc(1, sizeof(SOperatorInfo));
if (pOperator == NULL) {
goto _clean;
}
pOperator->name = "TableAggregate";
pOperator->operatorType = OP_Aggregate;
pOperator->blockingOptr = true;
......@@ -7009,31 +7238,53 @@ SOperatorInfo* createAggregateOperatorInfo(SQueryRuntimeEnv* pRuntimeEnv, SOpera
appendUpstream(pOperator, upstream);
return pOperator;
_clean:
destroyAggOperatorInfo((void *)pInfo, numOfOutput);
tfree(pInfo);
return NULL;
}
static void doDestroyBasicInfo(SOptrBasicInfo* pInfo, int32_t numOfOutput) {
assert(pInfo != NULL);
destroySQLFunctionCtx(pInfo->pCtx, numOfOutput);
tfree(pInfo->rowCellInfoOffset);
if (pInfo->pCtx) {
destroySQLFunctionCtx(pInfo->pCtx, numOfOutput);
}
if (pInfo->rowCellInfoOffset) {
tfree(pInfo->rowCellInfoOffset);
}
if (pInfo->resultRowInfo.pResult) {
cleanupResultRowInfo(&pInfo->resultRowInfo);
}
cleanupResultRowInfo(&pInfo->resultRowInfo);
pInfo->pRes = destroyOutputBuf(pInfo->pRes);
if (pInfo->pRes) {
pInfo->pRes = destroyOutputBuf(pInfo->pRes);
}
}
static void destroyBasicOperatorInfo(void* param, int32_t numOfOutput) {
SOptrBasicInfo* pInfo = (SOptrBasicInfo*) param;
doDestroyBasicInfo(pInfo, numOfOutput);
}
static void destroyStateWindowOperatorInfo(void* param, int32_t numOfOutput) {
SStateWindowOperatorInfo* pInfo = (SStateWindowOperatorInfo*) param;
doDestroyBasicInfo(&pInfo->binfo, numOfOutput);
tfree(pInfo->prevData);
if (pInfo->prevData) {
tfree(pInfo->prevData);
}
}
static void destroyAggOperatorInfo(void* param, int32_t numOfOutput) {
SAggOperatorInfo* pInfo = (SAggOperatorInfo*) param;
doDestroyBasicInfo(&pInfo->binfo, numOfOutput);
}
static void destroySWindowOperatorInfo(void* param, int32_t numOfOutput) {
SSWindowOperatorInfo* pInfo = (SSWindowOperatorInfo*) param;
doDestroyBasicInfo(&pInfo->binfo, numOfOutput);
......@@ -7041,15 +7292,27 @@ static void destroySWindowOperatorInfo(void* param, int32_t numOfOutput) {
static void destroySFillOperatorInfo(void* param, int32_t numOfOutput) {
SFillOperatorInfo* pInfo = (SFillOperatorInfo*) param;
pInfo->pFillInfo = taosDestroyFillInfo(pInfo->pFillInfo);
pInfo->pRes = destroyOutputBuf(pInfo->pRes);
tfree(pInfo->p);
if (pInfo->pFillInfo) {
pInfo->pFillInfo = taosDestroyFillInfo(pInfo->pFillInfo);
}
if (pInfo->pRes) {
pInfo->pRes = destroyOutputBuf(pInfo->pRes);
}
if (pInfo->p) {
tfree(pInfo->p);
}
}
static void destroyGroupbyOperatorInfo(void* param, int32_t numOfOutput) {
SGroupbyOperatorInfo* pInfo = (SGroupbyOperatorInfo*) param;
doDestroyBasicInfo(&pInfo->binfo, numOfOutput);
tfree(pInfo->prevData);
if (pInfo->prevData) {
tfree(pInfo->prevData);
}
}
static void destroyProjectOperatorInfo(void* param, int32_t numOfOutput) {
......@@ -7060,18 +7323,27 @@ static void destroyProjectOperatorInfo(void* param, int32_t numOfOutput) {
static void destroyTimeEveryOperatorInfo(void* param, int32_t numOfOutput) {
STimeEveryOperatorInfo* pInfo = (STimeEveryOperatorInfo*) param;
doDestroyBasicInfo(&pInfo->binfo, numOfOutput);
taosHashCleanup(pInfo->rangeStart);
if (pInfo->rangeStart) {
taosHashCleanup(pInfo->rangeStart);
}
}
static void destroyTagScanOperatorInfo(void* param, int32_t numOfOutput) {
STagScanInfo* pInfo = (STagScanInfo*) param;
pInfo->pRes = destroyOutputBuf(pInfo->pRes);
if (pInfo->pRes) {
pInfo->pRes = destroyOutputBuf(pInfo->pRes);
}
}
static void destroyOrderOperatorInfo(void* param, int32_t numOfOutput) {
SOrderOperatorInfo* pInfo = (SOrderOperatorInfo*) param;
pInfo->pDataBlock = destroyOutputBuf(pInfo->pDataBlock);
if (pInfo->pDataBlock) {
pInfo->pDataBlock = destroyOutputBuf(pInfo->pDataBlock);
}
}
static void destroyConditionOperatorInfo(void* param, int32_t numOfOutput) {
......@@ -7081,14 +7353,29 @@ static void destroyConditionOperatorInfo(void* param, int32_t numOfOutput) {
static void destroyDistinctOperatorInfo(void* param, int32_t numOfOutput) {
SDistinctOperatorInfo* pInfo = (SDistinctOperatorInfo*) param;
taosHashCleanup(pInfo->pSet);
tfree(pInfo->buf);
taosArrayDestroy(&pInfo->pDistinctDataInfo);
pInfo->pRes = destroyOutputBuf(pInfo->pRes);
if (pInfo->pSet) {
taosHashCleanup(pInfo->pSet);
}
if (pInfo->buf) {
tfree(pInfo->buf);
}
if (pInfo->pDistinctDataInfo) {
taosArrayDestroy(&pInfo->pDistinctDataInfo);
}
if (pInfo->pRes) {
pInfo->pRes = destroyOutputBuf(pInfo->pRes);
}
}
SOperatorInfo* createMultiTableAggOperatorInfo(SQueryRuntimeEnv* pRuntimeEnv, SOperatorInfo* upstream, SExprInfo* pExpr, int32_t numOfOutput) {
SAggOperatorInfo* pInfo = calloc(1, sizeof(SAggOperatorInfo));
if (pInfo == NULL) {
return NULL;
}
size_t tableGroup = GET_NUM_OF_TABLEGROUP(pRuntimeEnv);
......@@ -7096,7 +7383,15 @@ SOperatorInfo* createMultiTableAggOperatorInfo(SQueryRuntimeEnv* pRuntimeEnv, SO
pInfo->binfo.pCtx = createSQLFunctionCtx(pRuntimeEnv, pExpr, numOfOutput, &pInfo->binfo.rowCellInfoOffset);
initResultRowInfo(&pInfo->binfo.resultRowInfo, (int32_t)tableGroup, TSDB_DATA_TYPE_INT);
if (pInfo->binfo.pRes == NULL || pInfo->binfo.pCtx == NULL || pInfo->binfo.resultRowInfo.pResult == NULL) {
goto _clean;
}
SOperatorInfo* pOperator = calloc(1, sizeof(SOperatorInfo));
if (pOperator == NULL) {
goto _clean;
}
pOperator->name = "MultiTableAggregate";
pOperator->operatorType = OP_MultiTableAggregate;
pOperator->blockingOptr = true;
......@@ -7111,10 +7406,19 @@ SOperatorInfo* createMultiTableAggOperatorInfo(SQueryRuntimeEnv* pRuntimeEnv, SO
appendUpstream(pOperator, upstream);
return pOperator;
_clean:
destroyAggOperatorInfo((void *)pInfo, numOfOutput);
tfree(pInfo);
return NULL;
}
SOperatorInfo* createProjectOperatorInfo(SQueryRuntimeEnv* pRuntimeEnv, SOperatorInfo* upstream, SExprInfo* pExpr, int32_t numOfOutput) {
SProjectOperatorInfo* pInfo = calloc(1, sizeof(SProjectOperatorInfo));
if (pInfo == NULL) {
return NULL;
}
pInfo->seed = rand();
pInfo->bufCapacity = pRuntimeEnv->resultInfo.capacity;
......@@ -7124,9 +7428,18 @@ SOperatorInfo* createProjectOperatorInfo(SQueryRuntimeEnv* pRuntimeEnv, SOperato
pBInfo->pCtx = createSQLFunctionCtx(pRuntimeEnv, pExpr, numOfOutput, &pBInfo->rowCellInfoOffset);
initResultRowInfo(&pBInfo->resultRowInfo, 8, TSDB_DATA_TYPE_INT);
if (pInfo->binfo.pRes == NULL || pInfo->binfo.pCtx == NULL || pInfo->binfo.resultRowInfo.pResult == NULL) {
goto _clean;
}
setDefaultOutputBuf(pRuntimeEnv, pBInfo, pInfo->seed, MASTER_SCAN);
SOperatorInfo* pOperator = calloc(1, sizeof(SOperatorInfo));
if (pOperator == NULL) {
goto _clean;
}
pOperator->name = "ProjectOperator";
pOperator->operatorType = OP_Project;
pOperator->blockingOptr = false;
......@@ -7141,6 +7454,12 @@ SOperatorInfo* createProjectOperatorInfo(SQueryRuntimeEnv* pRuntimeEnv, SOperato
appendUpstream(pOperator, upstream);
return pOperator;
_clean:
destroyProjectOperatorInfo((void *)pInfo, numOfOutput);
tfree(pInfo);
return NULL;
}
SColumnInfo* extractColumnFilterInfo(SExprInfo* pExpr, int32_t numOfOutput, int32_t* numOfFilterCols) {
......@@ -7175,12 +7494,18 @@ SColumnInfo* extractColumnFilterInfo(SExprInfo* pExpr, int32_t numOfOutput, int3
SOperatorInfo* createFilterOperatorInfo(SQueryRuntimeEnv* pRuntimeEnv, SOperatorInfo* upstream, SExprInfo* pExpr,
int32_t numOfOutput, SColumnInfo* pCols, int32_t numOfFilter) {
SFilterOperatorInfo* pInfo = calloc(1, sizeof(SFilterOperatorInfo));
if (pInfo == NULL) {
return NULL;
}
assert(numOfFilter > 0 && pCols != NULL);
doCreateFilterInfo(pCols, numOfOutput, numOfFilter, &pInfo->pFilterInfo, 0);
pInfo->numOfFilterCols = numOfFilter;
SOperatorInfo* pOperator = calloc(1, sizeof(SOperatorInfo));
if (pOperator == NULL) {
goto _clean;
}
pOperator->name = "FilterOperator";
pOperator->operatorType = OP_Filter;
......@@ -7195,13 +7520,27 @@ SOperatorInfo* createFilterOperatorInfo(SQueryRuntimeEnv* pRuntimeEnv, SOperator
appendUpstream(pOperator, upstream);
return pOperator;
_clean:
destroyConditionOperatorInfo((void *)pInfo, numOfOutput);
tfree(pInfo);
return NULL;
}
SOperatorInfo* createLimitOperatorInfo(SQueryRuntimeEnv* pRuntimeEnv, SOperatorInfo* upstream) {
SLimitOperatorInfo* pInfo = calloc(1, sizeof(SLimitOperatorInfo));
if (pInfo == NULL) {
return NULL;
}
pInfo->limit = pRuntimeEnv->pQueryAttr->limit.limit;
SOperatorInfo* pOperator = calloc(1, sizeof(SOperatorInfo));
if (pOperator == NULL) {
tfree(pInfo);
return NULL;
}
pOperator->name = "LimitOperator";
pOperator->operatorType = OP_Limit;
......@@ -7217,12 +7556,22 @@ SOperatorInfo* createLimitOperatorInfo(SQueryRuntimeEnv* pRuntimeEnv, SOperatorI
SOperatorInfo* createTimeIntervalOperatorInfo(SQueryRuntimeEnv* pRuntimeEnv, SOperatorInfo* upstream, SExprInfo* pExpr, int32_t numOfOutput) {
STableIntervalOperatorInfo* pInfo = calloc(1, sizeof(STableIntervalOperatorInfo));
if (pInfo == NULL) {
return NULL;
}
pInfo->pCtx = createSQLFunctionCtx(pRuntimeEnv, pExpr, numOfOutput, &pInfo->rowCellInfoOffset);
pInfo->pRes = createOutputBuf(pExpr, numOfOutput, pRuntimeEnv->resultInfo.capacity);
initResultRowInfo(&pInfo->resultRowInfo, 8, TSDB_DATA_TYPE_INT);
if (pInfo->pRes == NULL || pInfo->pCtx == NULL || pInfo->resultRowInfo.pResult == NULL) {
goto _clean;
}
SOperatorInfo* pOperator = calloc(1, sizeof(SOperatorInfo));
if (pOperator == NULL) {
goto _clean;
}
pOperator->name = "TimeIntervalAggOperator";
pOperator->operatorType = OP_TimeWindow;
......@@ -7237,12 +7586,22 @@ SOperatorInfo* createTimeIntervalOperatorInfo(SQueryRuntimeEnv* pRuntimeEnv, SOp
appendUpstream(pOperator, upstream);
return pOperator;
_clean:
destroyBasicOperatorInfo((void *)pInfo, numOfOutput);
tfree(pInfo);
return NULL;
}
SOperatorInfo* createTimeEveryOperatorInfo(SQueryRuntimeEnv* pRuntimeEnv, SOperatorInfo* upstream, SExprInfo* pExpr, int32_t numOfOutput) {
STimeEveryOperatorInfo* pInfo = calloc(1, sizeof(STimeEveryOperatorInfo));
SQueryAttr* pQueryAttr = pRuntimeEnv->pQueryAttr;
if (pInfo == NULL) {
return NULL;
}
SQueryAttr *pQueryAttr = pRuntimeEnv->pQueryAttr;
pInfo->seed = rand();
pInfo->bufCapacity = pRuntimeEnv->resultInfo.capacity;
......@@ -7258,9 +7617,20 @@ SOperatorInfo* createTimeEveryOperatorInfo(SQueryRuntimeEnv* pRuntimeEnv, SOpera
}
initResultRowInfo(&pBInfo->resultRowInfo, 8, TSDB_DATA_TYPE_INT);
if (pBInfo->pRes == NULL || pBInfo->pCtx == NULL || pBInfo->resultRowInfo.pResult == NULL ||
(pQueryAttr->needReverseScan && pInfo->rangeStart == NULL))
{
goto _clean;
}
setDefaultOutputBuf(pRuntimeEnv, pBInfo, pInfo->seed, MASTER_SCAN);
SOperatorInfo* pOperator = calloc(1, sizeof(SOperatorInfo));
if (pOperator == NULL) {
goto _clean;
}
pOperator->name = "TimeEveryOperator";
pOperator->operatorType = OP_TimeEvery;
pOperator->blockingOptr = false;
......@@ -7275,18 +7645,36 @@ SOperatorInfo* createTimeEveryOperatorInfo(SQueryRuntimeEnv* pRuntimeEnv, SOpera
appendUpstream(pOperator, upstream);
return pOperator;
_clean:
destroyTimeEveryOperatorInfo((void *)pInfo, numOfOutput);
tfree(pInfo);
return NULL;
}
SOperatorInfo* createStatewindowOperatorInfo(SQueryRuntimeEnv* pRuntimeEnv, SOperatorInfo* upstream, SExprInfo* pExpr, int32_t numOfOutput) {
SStateWindowOperatorInfo* pInfo = calloc(1, sizeof(SStateWindowOperatorInfo));
if (pInfo == NULL) {
return NULL;
}
pInfo->colIndex = -1;
pInfo->reptScan = false;
pInfo->binfo.pCtx = createSQLFunctionCtx(pRuntimeEnv, pExpr, numOfOutput, &pInfo->binfo.rowCellInfoOffset);
pInfo->binfo.pRes = createOutputBuf(pExpr, numOfOutput, pRuntimeEnv->resultInfo.capacity);
initResultRowInfo(&pInfo->binfo.resultRowInfo, 8, TSDB_DATA_TYPE_INT);
if (pInfo->binfo.pCtx == NULL || pInfo->binfo.pRes == NULL || pInfo->binfo.resultRowInfo.pResult == NULL) {
goto _clean;
}
SOperatorInfo* pOperator = calloc(1, sizeof(SOperatorInfo));
if (pOperator == NULL) {
goto _clean;
}
pOperator->name = "StateWindowOperator";
pOperator->operatorType = OP_StateWindow;
pOperator->blockingOptr = true;
......@@ -7300,17 +7688,34 @@ SOperatorInfo* createStatewindowOperatorInfo(SQueryRuntimeEnv* pRuntimeEnv, SOpe
appendUpstream(pOperator, upstream);
return pOperator;
_clean:
destroyStateWindowOperatorInfo((void *)pInfo, numOfOutput);
tfree(pInfo);
return NULL;
}
SOperatorInfo* createSWindowOperatorInfo(SQueryRuntimeEnv* pRuntimeEnv, SOperatorInfo* upstream, SExprInfo* pExpr, int32_t numOfOutput) {
SSWindowOperatorInfo* pInfo = calloc(1, sizeof(SSWindowOperatorInfo));
if (pInfo == NULL) {
return NULL;
}
pInfo->binfo.pCtx = createSQLFunctionCtx(pRuntimeEnv, pExpr, numOfOutput, &pInfo->binfo.rowCellInfoOffset);
pInfo->binfo.pRes = createOutputBuf(pExpr, numOfOutput, pRuntimeEnv->resultInfo.capacity);
initResultRowInfo(&pInfo->binfo.resultRowInfo, 8, TSDB_DATA_TYPE_INT);
if (pInfo->binfo.pCtx == NULL || pInfo->binfo.pRes == NULL || pInfo->binfo.resultRowInfo.pResult == NULL) {
goto _clean;
}
pInfo->prevTs = INT64_MIN;
pInfo->reptScan = false;
SOperatorInfo* pOperator = calloc(1, sizeof(SOperatorInfo));
if (pOperator == NULL) {
goto _clean;
}
pOperator->name = "SessionWindowAggOperator";
pOperator->operatorType = OP_SessionWindow;
......@@ -7325,16 +7730,33 @@ SOperatorInfo* createSWindowOperatorInfo(SQueryRuntimeEnv* pRuntimeEnv, SOperato
appendUpstream(pOperator, upstream);
return pOperator;
_clean:
destroyStateWindowOperatorInfo((void *)pInfo, numOfOutput);
tfree(pInfo);
return NULL;
}
SOperatorInfo* createMultiTableTimeIntervalOperatorInfo(SQueryRuntimeEnv* pRuntimeEnv, SOperatorInfo* upstream, SExprInfo* pExpr, int32_t numOfOutput) {
STableIntervalOperatorInfo* pInfo = calloc(1, sizeof(STableIntervalOperatorInfo));
if (pInfo == NULL) {
return NULL;
}
pInfo->pCtx = createSQLFunctionCtx(pRuntimeEnv, pExpr, numOfOutput, &pInfo->rowCellInfoOffset);
pInfo->pRes = createOutputBuf(pExpr, numOfOutput, pRuntimeEnv->resultInfo.capacity);
initResultRowInfo(&pInfo->resultRowInfo, 8, TSDB_DATA_TYPE_INT);
if (pInfo->pCtx == NULL || pInfo->pRes == NULL || pInfo->resultRowInfo.pResult == NULL) {
goto _clean;
}
SOperatorInfo* pOperator = calloc(1, sizeof(SOperatorInfo));
if (pOperator == NULL) {
goto _clean;
}
pOperator->name = "MultiTableTimeIntervalOperator";
pOperator->operatorType = OP_MultiTableTimeInterval;
pOperator->blockingOptr = true;
......@@ -7349,14 +7771,22 @@ SOperatorInfo* createMultiTableTimeIntervalOperatorInfo(SQueryRuntimeEnv* pRunti
appendUpstream(pOperator, upstream);
return pOperator;
_clean:
destroyBasicOperatorInfo((void *)pInfo, numOfOutput);
tfree(pInfo);
return NULL;
}
SOperatorInfo* createGroupbyOperatorInfo(SQueryRuntimeEnv* pRuntimeEnv, SOperatorInfo* upstream, SExprInfo* pExpr, int32_t numOfOutput) {
SGroupbyOperatorInfo* pInfo = calloc(1, sizeof(SGroupbyOperatorInfo));
pInfo->colIndex = -1; // group by column index
if (pInfo == NULL) {
return NULL;
}
pInfo->colIndex = -1; // group by column index
pInfo->binfo.pCtx = createSQLFunctionCtx(pRuntimeEnv, pExpr, numOfOutput, &pInfo->binfo.rowCellInfoOffset);
SQueryAttr *pQueryAttr = pRuntimeEnv->pQueryAttr;
......@@ -7367,7 +7797,15 @@ SOperatorInfo* createGroupbyOperatorInfo(SQueryRuntimeEnv* pRuntimeEnv, SOperato
pInfo->binfo.pRes = createOutputBuf(pExpr, numOfOutput, pRuntimeEnv->resultInfo.capacity);
initResultRowInfo(&pInfo->binfo.resultRowInfo, 8, TSDB_DATA_TYPE_INT);
if (pInfo->binfo.pCtx == NULL || pInfo->binfo.pRes == NULL || pInfo->binfo.resultRowInfo.pResult == NULL) {
goto _clean;
}
SOperatorInfo* pOperator = calloc(1, sizeof(SOperatorInfo));
if (pOperator == NULL) {
goto _clean;
}
pOperator->name = "GroupbyAggOperator";
pOperator->blockingOptr = true;
pOperator->status = OP_IN_EXECUTING;
......@@ -7381,16 +7819,34 @@ SOperatorInfo* createGroupbyOperatorInfo(SQueryRuntimeEnv* pRuntimeEnv, SOperato
appendUpstream(pOperator, upstream);
return pOperator;
_clean:
destroyGroupbyOperatorInfo((void *)pInfo, numOfOutput);
tfree(pInfo);
return NULL;
}
SOperatorInfo* createFillOperatorInfo(SQueryRuntimeEnv* pRuntimeEnv, SOperatorInfo* upstream, SExprInfo* pExpr, int32_t numOfOutput, bool multigroupResult) {
SFillOperatorInfo* pInfo = calloc(1, sizeof(SFillOperatorInfo));
if (pInfo == NULL) {
return NULL;
}
pInfo->pRes = createOutputBuf(pExpr, numOfOutput, pRuntimeEnv->resultInfo.capacity);
if (pInfo->pRes == NULL) {
goto _clean;
}
pInfo->multigroupResult = multigroupResult;
{
SQueryAttr* pQueryAttr = pRuntimeEnv->pQueryAttr;
SFillColInfo* pColInfo = createFillColInfo(pExpr, numOfOutput, pQueryAttr->fillVal);
if (pColInfo == NULL) {
goto _clean;
}
STimeWindow w = TSWINDOW_INITIALIZER;
TSKEY sk = MIN(pQueryAttr->window.skey, pQueryAttr->window.ekey);
......@@ -7401,11 +7857,20 @@ SOperatorInfo* createFillOperatorInfo(SQueryRuntimeEnv* pRuntimeEnv, SOperatorIn
taosCreateFillInfo(pQueryAttr->order.order, w.skey, 0, (int32_t)pRuntimeEnv->resultInfo.capacity, numOfOutput,
pQueryAttr->interval.sliding, pQueryAttr->interval.slidingUnit,
(int8_t)pQueryAttr->precision, pQueryAttr->fillType, pColInfo, pRuntimeEnv->qinfo);
if (pInfo->pFillInfo == NULL) {
goto _clean;
}
pInfo->p = calloc(pInfo->pFillInfo->numOfCols, POINTER_BYTES);
if (pInfo->p == NULL) {
goto _clean;
}
}
SOperatorInfo* pOperator = calloc(1, sizeof(SOperatorInfo));
if (pOperator == NULL) {
goto _clean;
}
pOperator->name = "FillOperator";
pOperator->blockingOptr = false;
......@@ -7420,14 +7885,27 @@ SOperatorInfo* createFillOperatorInfo(SQueryRuntimeEnv* pRuntimeEnv, SOperatorIn
appendUpstream(pOperator, upstream);
return pOperator;
_clean:
destroySFillOperatorInfo((void *)pInfo, numOfOutput);
tfree(pInfo);
return NULL;
}
SOperatorInfo* createSLimitOperatorInfo(SQueryRuntimeEnv* pRuntimeEnv, SOperatorInfo* upstream, SExprInfo* pExpr, int32_t numOfOutput, void* pMerger, bool multigroupResult) {
SSLimitOperatorInfo* pInfo = calloc(1, sizeof(SSLimitOperatorInfo));
if (pInfo == NULL) {
return NULL;
}
SQueryAttr* pQueryAttr = pRuntimeEnv->pQueryAttr;
pInfo->orderColumnList = getResultGroupCheckColumns(pQueryAttr);
if (pInfo->orderColumnList == NULL) {
goto _clean;
}
pInfo->slimit = pQueryAttr->slimit;
pInfo->limit = pQueryAttr->limit;
pInfo->capacity = pRuntimeEnv->resultInfo.capacity;
......@@ -7444,6 +7922,9 @@ SOperatorInfo* createSLimitOperatorInfo(SQueryRuntimeEnv* pRuntimeEnv, SOperator
int32_t numOfCols = (pInfo->orderColumnList != NULL)? (int32_t) taosArrayGetSize(pInfo->orderColumnList):0;
pInfo->prevRow = calloc(1, (POINTER_BYTES * numOfCols + len));
if (pInfo->prevRow == NULL) {
goto _clean;
}
int32_t offset = POINTER_BYTES * numOfCols;
for(int32_t i = 0; i < numOfCols; ++i) {
......@@ -7457,6 +7938,10 @@ SOperatorInfo* createSLimitOperatorInfo(SQueryRuntimeEnv* pRuntimeEnv, SOperator
SOperatorInfo* pOperator = calloc(1, sizeof(SOperatorInfo));
if (pInfo->pRes == NULL || pOperator == NULL) {
goto _clean;
}
pOperator->name = "SLimitOperator";
pOperator->operatorType = OP_SLimit;
pOperator->blockingOptr = false;
......@@ -7468,6 +7953,12 @@ SOperatorInfo* createSLimitOperatorInfo(SQueryRuntimeEnv* pRuntimeEnv, SOperator
appendUpstream(pOperator, upstream);
return pOperator;
_clean:
destroySlimitOperatorInfo((void *)pInfo, numOfOutput);
tfree(pInfo);
return NULL;
}
static SSDataBlock* doTagScan(void* param, bool* newgroup) {
......@@ -7618,7 +8109,14 @@ static SSDataBlock* doTagScan(void* param, bool* newgroup) {
SOperatorInfo* createTagScanOperatorInfo(SQueryRuntimeEnv* pRuntimeEnv, SExprInfo* pExpr, int32_t numOfOutput) {
STagScanInfo* pInfo = calloc(1, sizeof(STagScanInfo));
if (pInfo == NULL) {
return NULL;
}
pInfo->pRes = createOutputBuf(pExpr, numOfOutput, pRuntimeEnv->resultInfo.capacity);
if (pInfo->pRes == NULL) {
goto _clean;
}
size_t numOfGroup = GET_NUM_OF_TABLEGROUP(pRuntimeEnv);
assert(numOfGroup == 0 || numOfGroup == 1);
......@@ -7627,6 +8125,10 @@ SOperatorInfo* createTagScanOperatorInfo(SQueryRuntimeEnv* pRuntimeEnv, SExprInf
pInfo->curPos = 0;
SOperatorInfo* pOperator = calloc(1, sizeof(SOperatorInfo));
if (pOperator == NULL) {
goto _clean;
}
pOperator->name = "SeqTableTagScan";
pOperator->operatorType = OP_TagScan;
pOperator->blockingOptr = false;
......@@ -7639,7 +8141,14 @@ SOperatorInfo* createTagScanOperatorInfo(SQueryRuntimeEnv* pRuntimeEnv, SExprInf
pOperator->cleanup = destroyTagScanOperatorInfo;
return pOperator;
_clean:
destroyTagScanOperatorInfo((void *)pInfo, numOfOutput);
tfree(pInfo);
return NULL;
}
static bool initMultiDistinctInfo(SDistinctOperatorInfo *pInfo, SOperatorInfo* pOperator, SSDataBlock *pBlock) {
if (taosArrayGetSize(pInfo->pDistinctDataInfo) == pOperator->numOfOutput) {
// distinct info already inited
......@@ -7756,6 +8265,10 @@ static SSDataBlock* hashDistinct(void* param, bool* newgroup) {
SOperatorInfo* createDistinctOperatorInfo(SQueryRuntimeEnv* pRuntimeEnv, SOperatorInfo* upstream, SExprInfo* pExpr, int32_t numOfOutput) {
SDistinctOperatorInfo* pInfo = calloc(1, sizeof(SDistinctOperatorInfo));
if (pInfo == NULL) {
return NULL;
}
pInfo->totalBytes = 0;
pInfo->buf = NULL;
pInfo->threshold = tsMaxNumOfDistinctResults; // distinct result threshold
......@@ -7764,8 +8277,15 @@ SOperatorInfo* createDistinctOperatorInfo(SQueryRuntimeEnv* pRuntimeEnv, SOperat
pInfo->pSet = taosHashInit(64, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY), false, HASH_NO_LOCK);
pInfo->pRes = createOutputBuf(pExpr, numOfOutput, (int32_t) pInfo->outputCapacity);
if (pInfo->pDistinctDataInfo == NULL || pInfo->pSet == NULL || pInfo->pRes == NULL) {
goto _clean;
}
SOperatorInfo* pOperator = calloc(1, sizeof(SOperatorInfo));
if (pOperator == NULL) {
goto _clean;
}
pOperator->name = "DistinctOperator";
pOperator->blockingOptr = false;
pOperator->status = OP_IN_EXECUTING;
......@@ -7780,6 +8300,12 @@ SOperatorInfo* createDistinctOperatorInfo(SQueryRuntimeEnv* pRuntimeEnv, SOperat
appendUpstream(pOperator, upstream);
return pOperator;
_clean:
destroyDistinctOperatorInfo((void *)pInfo, numOfOutput);
tfree(pInfo);
return NULL;
}
static int32_t getColumnIndexInSource(SQueriedTableInfo *pTableInfo, SSqlExpr *pExpr, SColumnInfo* pTagCols) {
......@@ -8991,9 +9517,9 @@ SQInfo* createQInfoImpl(SQueryTableMsg* pQueryMsg, SGroupbyExpr* pGroupbyExpr, S
doUpdateExprColumnIndex(pQueryAttr);
// calc skipOffset
// calc skipOffset
if(pQueryMsg->offset > 0 && TSDB_QUERY_HAS_TYPE(pQueryMsg->queryType, TSDB_QUERY_TYPE_PROJECTION_QUERY)) {
if(pQueryAttr->stableQuery)
if(pQueryAttr->stableQuery)
pQueryAttr->skipOffset = false;
else
pQueryAttr->skipOffset = pQueryAttr->pFilters == NULL;
......
src/query/src/qFill.c
浏览文件 @ e78e3031
......@@ -354,6 +354,10 @@ SFillInfo* taosCreateFillInfo(int32_t order, TSKEY skey, int32_t numOfTags, int3
}
SFillInfo* pFillInfo = calloc(1, sizeof(SFillInfo));
if (pFillInfo == NULL) {
return NULL;
}
taosResetFillInfo(pFillInfo, skey);
pFillInfo->order = order;
......@@ -371,6 +375,10 @@ SFillInfo* taosCreateFillInfo(int32_t order, TSKEY skey, int32_t numOfTags, int3
pFillInfo->interval.slidingUnit = slidingUnit;
pFillInfo->pData = malloc(POINTER_BYTES * numOfCols);
if (pFillInfo->pData == NULL) {
tfree(pFillInfo);
return NULL;
}
// if (numOfTags > 0) {
pFillInfo->pTags = calloc(numOfCols, sizeof(SFillTagColInfo));
......
Markdown is supported
0% 或 .
You are about to add 0 people to the discussion. Proceed with caution.
先完成此消息的编辑！
想要评论请 注册 或