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

fix(query): enable output check

This commit is contained in:
Liu Jicong 2022-06-28 13:44:50 +08:00
parent 474b6d1ee3
commit 383e7a495b
7 changed files with 152 additions and 138 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

7
examples/c/stream_demo.c
View File

@ -98,9 +98,10 @@ int32_t create_stream() {
/*const char* sql = "select min(k), max(k), sum(k) as sum_of_k from st1";*/ /*const char* sql = "select min(k), max(k), sum(k) as sum_of_k from st1";*/
/*const char* sql = "select sum(k) from tu1 interval(10m)";*/ /*const char* sql = "select sum(k) from tu1 interval(10m)";*/
/*pRes = tmq_create_stream(pConn, "stream1", "out1", sql);*/ /*pRes = tmq_create_stream(pConn, "stream1", "out1", sql);*/
pRes = taos_query(pConn, pRes = taos_query(
"create stream stream1 trigger at_once into outstb as select _wstartts, sum(k) from st1 partition " pConn,
"by tbname interval(10s) "); "create stream stream1 trigger max_delay 10s into outstb as select _wstartts, sum(k) from st1 partition "
"by tbname session(ts, 10s) ");
if (taos_errno(pRes) != 0) { if (taos_errno(pRes) != 0) {
printf("failed to create stream stream1, reason:%s\n", taos_errstr(pRes)); printf("failed to create stream stream1, reason:%s\n", taos_errstr(pRes));
return -1; return -1;

4
examples/c/tmq.c
View File

@ -137,8 +137,8 @@ int32_t create_topic() {
} }
taos_free_result(pRes); taos_free_result(pRes);
pRes = taos_query(pConn, "create topic topic_ctb_column with meta as database abc1"); /*pRes = taos_query(pConn, "create topic topic_ctb_column with meta as database abc1");*/
/*pRes = taos_query(pConn, "create topic topic_ctb_column as select ts, c1, c2, c3 from st1");*/ pRes = taos_query(pConn, "create topic topic_ctb_column as select ts, c1, c2, c3,t1 from st1 where t1 = 2000");
if (taos_errno(pRes) != 0) { if (taos_errno(pRes) != 0) {
printf("failed to create topic topic_ctb_column, reason:%s\n", taos_errstr(pRes)); printf("failed to create topic topic_ctb_column, reason:%s\n", taos_errstr(pRes));
return -1; return -1;

5
include/libs/stream/tstream.h
View File

@ -327,9 +327,8 @@ static FORCE_INLINE int32_t streamTaskInput(SStreamTask* pTask, SStreamQueueItem
taosWriteQitem(pTask->inputQueue->queue, pItem); taosWriteQitem(pTask->inputQueue->queue, pItem);
} }
if (pItem->type != STREAM_INPUT__TRIGGER && pItem->type != STREAM_INPUT__CHECKPOINT && pTask->triggerParam != 0 && if (pItem->type != STREAM_INPUT__TRIGGER && pItem->type != STREAM_INPUT__CHECKPOINT && pTask->triggerParam != 0) {
pTask->triggerStatus == TASK_TRIGGER_STATUS__IN_ACTIVE) { atomic_val_compare_exchange_8(&pTask->triggerStatus, TASK_TRIGGER_STATUS__IN_ACTIVE, TASK_TRIGGER_STATUS__ACTIVE);
atomic_store_8(&pTask->triggerStatus, TASK_TRIGGER_STATUS__ACTIVE);
} }
// TODO: back pressure // TODO: back pressure

6
source/dnode/vnode/src/tq/tqRead.c
View File

@ -231,7 +231,7 @@ int32_t tqRetrieveDataBlock(SSDataBlock* pBlock, STqReadHandle* pHandle) {
tInitSubmitBlkIter(&pHandle->msgIter, pHandle->pBlock, &pHandle->blkIter); tInitSubmitBlkIter(&pHandle->msgIter, pHandle->pBlock, &pHandle->blkIter);
pBlock->info.groupId = 0; pBlock->info.groupId = 0;
pBlock->info.uid = pHandle->msgIter.uid; // set the uid of table for submit block pBlock->info.uid = pHandle->msgIter.uid;
pBlock->info.rows = pHandle->msgIter.numOfRows; pBlock->info.rows = pHandle->msgIter.numOfRows;
while ((row = tGetSubmitBlkNext(&pHandle->blkIter)) != NULL) { while ((row = tGetSubmitBlkNext(&pHandle->blkIter)) != NULL) {
@ -251,8 +251,8 @@ int32_t tqRetrieveDataBlock(SSDataBlock* pBlock, STqReadHandle* pHandle) {
} }
return 0; return 0;
FAIL: // todo refactor here FAIL:
// if (*ppCols) taosArrayDestroy(*ppCols); tDeleteSSDataBlock(pBlock);
return -1; return -1;
} }

128
source/libs/executor/src/executil.c
View File

@ -13,10 +13,10 @@
* along with this program. If not, see <http://www.gnu.org/licenses/>. * along with this program. If not, see <http://www.gnu.org/licenses/>.
*/ */
#include "os.h"
#include "index.h"
#include "function.h" #include "function.h"
#include "functionMgt.h" #include "functionMgt.h"
#include "index.h"
#include "os.h"
#include "tdatablock.h" #include "tdatablock.h"
#include "thash.h" #include "thash.h"
#include "tmsg.h" #include "tmsg.h"
@ -25,45 +25,41 @@
#include "executorimpl.h" #include "executorimpl.h"
#include "tcompression.h" #include "tcompression.h"
void initResultRowInfo(SResultRowInfo *pResultRowInfo) { void initResultRowInfo(SResultRowInfo* pResultRowInfo) {
pResultRowInfo->size = 0; pResultRowInfo->size = 0;
pResultRowInfo->cur.pageId = -1; pResultRowInfo->cur.pageId = -1;
} }
void cleanupResultRowInfo(SResultRowInfo *pResultRowInfo) { void cleanupResultRowInfo(SResultRowInfo* pResultRowInfo) {
if (pResultRowInfo == NULL) { if (pResultRowInfo == NULL) {
return; return;
} }
for(int32_t i = 0; i < pResultRowInfo->size; ++i) { for (int32_t i = 0; i < pResultRowInfo->size; ++i) {
// if (pResultRowInfo->pResult[i]) { // if (pResultRowInfo->pResult[i]) {
// taosMemoryFreeClear(pResultRowInfo->pResult[i]->key); // taosMemoryFreeClear(pResultRowInfo->pResult[i]->key);
// } // }
} }
} }
void closeAllResultRows(SResultRowInfo *pResultRowInfo) { void closeAllResultRows(SResultRowInfo* pResultRowInfo) {
// do nothing // do nothing
} }
bool isResultRowClosed(SResultRow* pRow) { bool isResultRowClosed(SResultRow* pRow) { return (pRow->closed == true); }
return (pRow->closed == true);
}
void closeResultRow(SResultRow* pResultRow) { void closeResultRow(SResultRow* pResultRow) { pResultRow->closed = true; }
pResultRow->closed = true;
}
// TODO refactor: use macro // TODO refactor: use macro
SResultRowEntryInfo* getResultEntryInfo(const SResultRow* pRow, int32_t index, const int32_t* offset) { SResultRowEntryInfo* getResultEntryInfo(const SResultRow* pRow, int32_t index, const int32_t* offset) {
assert(index >= 0 && offset != NULL); assert(index >= 0 && offset != NULL);
return (SResultRowEntryInfo*)((char*) pRow->pEntryInfo + offset[index]); return (SResultRowEntryInfo*)((char*)pRow->pEntryInfo + offset[index]);
} }
size_t getResultRowSize(SqlFunctionCtx* pCtx, int32_t numOfOutput) { size_t getResultRowSize(SqlFunctionCtx* pCtx, int32_t numOfOutput) {
int32_t rowSize = (numOfOutput * sizeof(SResultRowEntryInfo)) + sizeof(SResultRow); int32_t rowSize = (numOfOutput * sizeof(SResultRowEntryInfo)) + sizeof(SResultRow);
for(int32_t i = 0; i < numOfOutput; ++i) { for (int32_t i = 0; i < numOfOutput; ++i) {
rowSize += pCtx[i].resDataInfo.interBufSize; rowSize += pCtx[i].resDataInfo.interBufSize;
} }
@ -74,31 +70,29 @@ void cleanupGroupResInfo(SGroupResInfo* pGroupResInfo) {
assert(pGroupResInfo != NULL); assert(pGroupResInfo != NULL);
taosArrayDestroy(pGroupResInfo->pRows); taosArrayDestroy(pGroupResInfo->pRows);
pGroupResInfo->pRows = NULL; pGroupResInfo->pRows = NULL;
pGroupResInfo->index = 0; pGroupResInfo->index = 0;
} }
static int32_t resultrowComparAsc(const void* p1, const void* p2) { static int32_t resultrowComparAsc(const void* p1, const void* p2) {
SResKeyPos* pp1 = *(SResKeyPos**) p1; SResKeyPos* pp1 = *(SResKeyPos**)p1;
SResKeyPos* pp2 = *(SResKeyPos**) p2; SResKeyPos* pp2 = *(SResKeyPos**)p2;
if (pp1->groupId == pp2->groupId) { if (pp1->groupId == pp2->groupId) {
int64_t pts1 = *(int64_t*) pp1->key; int64_t pts1 = *(int64_t*)pp1->key;
int64_t pts2 = *(int64_t*) pp2->key; int64_t pts2 = *(int64_t*)pp2->key;
if (pts1 == pts2) { if (pts1 == pts2) {
return 0; return 0;
} else { } else {
return pts1 < pts2? -1:1; return pts1 < pts2 ? -1 : 1;
} }
} else { } else {
return pp1->groupId < pp2->groupId? -1:1; return pp1->groupId < pp2->groupId ? -1 : 1;
} }
} }
static int32_t resultrowComparDesc(const void* p1, const void* p2) { static int32_t resultrowComparDesc(const void* p1, const void* p2) { return resultrowComparAsc(p2, p1); }
return resultrowComparAsc(p2, p1);
}
void initGroupedResultInfo(SGroupResInfo* pGroupResInfo, SHashObj* pHashmap, int32_t order) { void initGroupedResultInfo(SGroupResInfo* pGroupResInfo, SHashObj* pHashmap, int32_t order) {
if (pGroupResInfo->pRows != NULL) { if (pGroupResInfo->pRows != NULL) {
@ -110,20 +104,20 @@ void initGroupedResultInfo(SGroupResInfo* pGroupResInfo, SHashObj* pHashmap, int
pGroupResInfo->pRows = taosArrayInit(10, POINTER_BYTES); pGroupResInfo->pRows = taosArrayInit(10, POINTER_BYTES);
size_t keyLen = 0; size_t keyLen = 0;
while((pData = taosHashIterate(pHashmap, pData)) != NULL) { while ((pData = taosHashIterate(pHashmap, pData)) != NULL) {
void* key = taosHashGetKey(pData, &keyLen); void* key = taosHashGetKey(pData, &keyLen);
SResKeyPos* p = taosMemoryMalloc(keyLen + sizeof(SResultRowPosition)); SResKeyPos* p = taosMemoryMalloc(keyLen + sizeof(SResultRowPosition));
p->groupId = *(uint64_t*) key; p->groupId = *(uint64_t*)key;
p->pos = *(SResultRowPosition*) pData; p->pos = *(SResultRowPosition*)pData;
memcpy(p->key, (char*)key + sizeof(uint64_t), keyLen - sizeof(uint64_t)); memcpy(p->key, (char*)key + sizeof(uint64_t), keyLen - sizeof(uint64_t));
taosArrayPush(pGroupResInfo->pRows, &p); taosArrayPush(pGroupResInfo->pRows, &p);
} }
if (order == TSDB_ORDER_ASC || order == TSDB_ORDER_DESC) { if (order == TSDB_ORDER_ASC || order == TSDB_ORDER_DESC) {
__compar_fn_t fn = (order == TSDB_ORDER_ASC)? resultrowComparAsc:resultrowComparDesc; __compar_fn_t fn = (order == TSDB_ORDER_ASC) ? resultrowComparAsc : resultrowComparDesc;
qsort(pGroupResInfo->pRows->pData, taosArrayGetSize(pGroupResInfo->pRows), POINTER_BYTES, fn); qsort(pGroupResInfo->pRows->pData, taosArrayGetSize(pGroupResInfo->pRows), POINTER_BYTES, fn);
} }
@ -155,7 +149,7 @@ int32_t getNumOfTotalRes(SGroupResInfo* pGroupResInfo) {
return 0; return 0;
} }
return (int32_t) taosArrayGetSize(pGroupResInfo->pRows); return (int32_t)taosArrayGetSize(pGroupResInfo->pRows);
} }
SArray* createSortInfo(SNodeList* pNodeList) { SArray* createSortInfo(SNodeList* pNodeList) {
@ -194,12 +188,13 @@ SSDataBlock* createResDataBlock(SDataBlockDescNode* pNode) {
pBlock->info.type = STREAM_INVALID; pBlock->info.type = STREAM_INVALID;
for (int32_t i = 0; i < numOfCols; ++i) { for (int32_t i = 0; i < numOfCols; ++i) {
SSlotDescNode* pDescNode = (SSlotDescNode*)nodesListGetNode(pNode->pSlots, i); SSlotDescNode* pDescNode = (SSlotDescNode*)nodesListGetNode(pNode->pSlots, i);
// if (!pDescNode->output) { // todo disable it temporarily if (!pDescNode->output) { // todo disable it temporarily
// continue; continue;
// } }
SColumnInfoData idata = createColumnInfoData(pDescNode->dataType.type, pDescNode->dataType.bytes, pDescNode->slotId); SColumnInfoData idata =
createColumnInfoData(pDescNode->dataType.type, pDescNode->dataType.bytes, pDescNode->slotId);
idata.info.scale = pDescNode->dataType.scale; idata.info.scale = pDescNode->dataType.scale;
idata.info.precision = pDescNode->dataType.precision; idata.info.precision = pDescNode->dataType.precision;
@ -211,10 +206,10 @@ SSDataBlock* createResDataBlock(SDataBlockDescNode* pNode) {
EDealRes doTranslateTagExpr(SNode** pNode, void* pContext) { EDealRes doTranslateTagExpr(SNode** pNode, void* pContext) {
SMetaReader* mr = (SMetaReader*)pContext; SMetaReader* mr = (SMetaReader*)pContext;
if(nodeType(*pNode) == QUERY_NODE_COLUMN){ if (nodeType(*pNode) == QUERY_NODE_COLUMN) {
SColumnNode* pSColumnNode = *(SColumnNode**)pNode; SColumnNode* pSColumnNode = *(SColumnNode**)pNode;
SValueNode *res = (SValueNode *)nodesMakeNode(QUERY_NODE_VALUE); SValueNode* res = (SValueNode*)nodesMakeNode(QUERY_NODE_VALUE);
if (NULL == res) { if (NULL == res) {
return DEAL_RES_ERROR; return DEAL_RES_ERROR;
} }
@ -227,8 +222,8 @@ EDealRes doTranslateTagExpr(SNode** pNode, void* pContext) {
const char* p = metaGetTableTagVal(&mr->me, pSColumnNode->node.resType.type, &tagVal); const char* p = metaGetTableTagVal(&mr->me, pSColumnNode->node.resType.type, &tagVal);
if (p == NULL) { if (p == NULL) {
res->node.resType.type = TSDB_DATA_TYPE_NULL; res->node.resType.type = TSDB_DATA_TYPE_NULL;
}else if (pSColumnNode->node.resType.type == TSDB_DATA_TYPE_JSON) { } else if (pSColumnNode->node.resType.type == TSDB_DATA_TYPE_JSON) {
int32_t len = ((const STag*)p) -> len; int32_t len = ((const STag*)p)->len;
res->datum.p = taosMemoryCalloc(len + 1, 1); res->datum.p = taosMemoryCalloc(len + 1, 1);
memcpy(res->datum.p, p, len); memcpy(res->datum.p, p, len);
} else if (IS_VAR_DATA_TYPE(pSColumnNode->node.resType.type)) { } else if (IS_VAR_DATA_TYPE(pSColumnNode->node.resType.type)) {
@ -240,10 +235,10 @@ EDealRes doTranslateTagExpr(SNode** pNode, void* pContext) {
} }
nodesDestroyNode(*pNode); nodesDestroyNode(*pNode);
*pNode = (SNode*)res; *pNode = (SNode*)res;
}else if (nodeType(*pNode) == QUERY_NODE_FUNCTION){ } else if (nodeType(*pNode) == QUERY_NODE_FUNCTION) {
SFunctionNode * pFuncNode = *(SFunctionNode**)pNode; SFunctionNode* pFuncNode = *(SFunctionNode**)pNode;
if(pFuncNode->funcType == FUNCTION_TYPE_TBNAME){ if (pFuncNode->funcType == FUNCTION_TYPE_TBNAME) {
SValueNode *res = (SValueNode *)nodesMakeNode(QUERY_NODE_VALUE); SValueNode* res = (SValueNode*)nodesMakeNode(QUERY_NODE_VALUE);
if (NULL == res) { if (NULL == res) {
return DEAL_RES_ERROR; return DEAL_RES_ERROR;
} }
@ -263,12 +258,12 @@ EDealRes doTranslateTagExpr(SNode** pNode, void* pContext) {
return DEAL_RES_CONTINUE; return DEAL_RES_CONTINUE;
} }
static bool isTableOk(STableKeyInfo* info, SNode *pTagCond, SMeta *metaHandle){ static bool isTableOk(STableKeyInfo* info, SNode* pTagCond, SMeta* metaHandle) {
SMetaReader mr = {0}; SMetaReader mr = {0};
metaReaderInit(&mr, metaHandle, 0); metaReaderInit(&mr, metaHandle, 0);
metaGetTableEntryByUid(&mr, info->uid); metaGetTableEntryByUid(&mr, info->uid);
SNode *pTagCondTmp = nodesCloneNode(pTagCond); SNode* pTagCondTmp = nodesCloneNode(pTagCond);
nodesRewriteExprPostOrder(&pTagCondTmp, doTranslateTagExpr, &mr); nodesRewriteExprPostOrder(&pTagCondTmp, doTranslateTagExpr, &mr);
metaReaderClear(&mr); metaReaderClear(&mr);
@ -281,7 +276,7 @@ static bool isTableOk(STableKeyInfo* info, SNode *pTagCond, SMeta *metaHandle){
} }
ASSERT(nodeType(pNew) == QUERY_NODE_VALUE); ASSERT(nodeType(pNew) == QUERY_NODE_VALUE);
SValueNode *pValue = (SValueNode *)pNew; SValueNode* pValue = (SValueNode*)pNew;
ASSERT(pValue->node.resType.type == TSDB_DATA_TYPE_BOOL); ASSERT(pValue->node.resType.type == TSDB_DATA_TYPE_BOOL);
bool result = pValue->datum.b; bool result = pValue->datum.b;
@ -292,12 +287,12 @@ static bool isTableOk(STableKeyInfo* info, SNode *pTagCond, SMeta *metaHandle){
int32_t getTableList(void* metaHandle, SScanPhysiNode* pScanNode, STableListInfo* pListInfo) { int32_t getTableList(void* metaHandle, SScanPhysiNode* pScanNode, STableListInfo* pListInfo) {
int32_t code = TSDB_CODE_SUCCESS; int32_t code = TSDB_CODE_SUCCESS;
pListInfo->pTableList = taosArrayInit(8, sizeof(STableKeyInfo)); pListInfo->pTableList = taosArrayInit(8, sizeof(STableKeyInfo));
if(pListInfo->pTableList == NULL) return TSDB_CODE_OUT_OF_MEMORY; if (pListInfo->pTableList == NULL) return TSDB_CODE_OUT_OF_MEMORY;
uint64_t tableUid = pScanNode->uid; uint64_t tableUid = pScanNode->uid;
pListInfo->suid = pScanNode->suid; pListInfo->suid = pScanNode->suid;
SNode* pTagCond = (SNode*)pListInfo->pTagCond; SNode* pTagCond = (SNode*)pListInfo->pTagCond;
SNode* pTagIndexCond = (SNode*)pListInfo->pTagIndexCond; SNode* pTagIndexCond = (SNode*)pListInfo->pTagIndexCond;
if (pScanNode->tableType == TSDB_SUPER_TABLE) { if (pScanNode->tableType == TSDB_SUPER_TABLE) {
@ -306,7 +301,7 @@ int32_t getTableList(void* metaHandle, SScanPhysiNode* pScanNode, STableListInfo
.metaEx = metaHandle, .idx = tsdbGetIdx(metaHandle), .ivtIdx = tsdbGetIvtIdx(metaHandle), .suid = tableUid}; .metaEx = metaHandle, .idx = tsdbGetIdx(metaHandle), .ivtIdx = tsdbGetIvtIdx(metaHandle), .suid = tableUid};
SArray* res = taosArrayInit(8, sizeof(uint64_t)); SArray* res = taosArrayInit(8, sizeof(uint64_t));
//code = doFilterTag(pTagIndexCond, &metaArg, res); // code = doFilterTag(pTagIndexCond, &metaArg, res);
code = TSDB_CODE_INDEX_REBUILDING; code = TSDB_CODE_INDEX_REBUILDING;
if (code == TSDB_CODE_INDEX_REBUILDING) { if (code == TSDB_CODE_INDEX_REBUILDING) {
code = tsdbGetAllTableList(metaHandle, tableUid, pListInfo->pTableList); code = tsdbGetAllTableList(metaHandle, tableUid, pListInfo->pTableList);
@ -328,33 +323,33 @@ int32_t getTableList(void* metaHandle, SScanPhysiNode* pScanNode, STableListInfo
code = tsdbGetAllTableList(metaHandle, tableUid, pListInfo->pTableList); code = tsdbGetAllTableList(metaHandle, tableUid, pListInfo->pTableList);
} }
if(pTagCond){ if (pTagCond) {
int32_t i = 0; int32_t i = 0;
while(i < taosArrayGetSize(pListInfo->pTableList)) { while (i < taosArrayGetSize(pListInfo->pTableList)) {
STableKeyInfo* info = taosArrayGet(pListInfo->pTableList, i); STableKeyInfo* info = taosArrayGet(pListInfo->pTableList, i);
bool isOk = isTableOk(info, pTagCond, metaHandle); bool isOk = isTableOk(info, pTagCond, metaHandle);
if(!isOk){ if (!isOk) {
taosArrayRemove(pListInfo->pTableList, i); taosArrayRemove(pListInfo->pTableList, i);
continue; continue;
} }
i++; i++;
} }
} }
}else { // Create one table group. } else { // Create one table group.
STableKeyInfo info = {.lastKey = 0, .uid = tableUid, .groupId = 0}; STableKeyInfo info = {.lastKey = 0, .uid = tableUid, .groupId = 0};
taosArrayPush(pListInfo->pTableList, &info); taosArrayPush(pListInfo->pTableList, &info);
} }
pListInfo->pGroupList = taosArrayInit(4, POINTER_BYTES); pListInfo->pGroupList = taosArrayInit(4, POINTER_BYTES);
if(pListInfo->pGroupList == NULL) return TSDB_CODE_OUT_OF_MEMORY; if (pListInfo->pGroupList == NULL) return TSDB_CODE_OUT_OF_MEMORY;
//put into list as default group, remove it if grouping sorting is required later // put into list as default group, remove it if grouping sorting is required later
taosArrayPush(pListInfo->pGroupList, &pListInfo->pTableList); taosArrayPush(pListInfo->pGroupList, &pListInfo->pTableList);
return code; return code;
} }
SArray* extractPartitionColInfo(SNodeList* pNodeList) { SArray* extractPartitionColInfo(SNodeList* pNodeList) {
if(!pNodeList) { if (!pNodeList) {
return NULL; return NULL;
} }
@ -383,7 +378,6 @@ SArray* extractPartitionColInfo(SNodeList* pNodeList) {
return pList; return pList;
} }
SArray* extractColMatchInfo(SNodeList* pNodeList, SDataBlockDescNode* pOutputNodeList, int32_t* numOfOutputCols, SArray* extractColMatchInfo(SNodeList* pNodeList, SDataBlockDescNode* pOutputNodeList, int32_t* numOfOutputCols,
int32_t type) { int32_t type) {
size_t numOfCols = LIST_LENGTH(pNodeList); size_t numOfCols = LIST_LENGTH(pNodeList);
@ -399,7 +393,7 @@ SArray* extractColMatchInfo(SNodeList* pNodeList, SDataBlockDescNode* pOutputNod
SColMatchInfo c = {0}; SColMatchInfo c = {0};
c.output = true; c.output = true;
c.colId = pColNode->colId; c.colId = pColNode->colId;
c.srcSlotId = pColNode->slotId; c.srcSlotId = pColNode->slotId;
c.matchType = type; c.matchType = type;
c.targetSlotId = pNode->slotId; c.targetSlotId = pNode->slotId;
@ -675,8 +669,8 @@ SqlFunctionCtx* createSqlFunctionCtx(SExprInfo* pExprInfo, int32_t numOfOutput,
} }
for (int32_t i = 1; i < numOfOutput; ++i) { for (int32_t i = 1; i < numOfOutput; ++i) {
(*rowEntryInfoOffset)[i] = (*rowEntryInfoOffset)[i] = (int32_t)((*rowEntryInfoOffset)[i - 1] + sizeof(SResultRowEntryInfo) +
(int32_t)((*rowEntryInfoOffset)[i - 1] + sizeof(SResultRowEntryInfo) + pFuncCtx[i - 1].resDataInfo.interBufSize); pFuncCtx[i - 1].resDataInfo.interBufSize);
} }
setSelectValueColumnInfo(pFuncCtx, numOfOutput); setSelectValueColumnInfo(pFuncCtx, numOfOutput);

125
source/libs/executor/src/timewindowoperator.c
View File

@ -34,7 +34,7 @@ typedef struct SWinRes {
typedef struct SPullWindowInfo { typedef struct SPullWindowInfo {
STimeWindow window; STimeWindow window;
uint64_t groupId; uint64_t groupId;
} SPullWindowInfo; } SPullWindowInfo;
static SSDataBlock* doStreamSessionAgg(SOperatorInfo* pOperator); static SSDataBlock* doStreamSessionAgg(SOperatorInfo* pOperator);
@ -695,11 +695,11 @@ static void doInterpUnclosedTimeWindow(SOperatorInfo* pOperatorInfo, int32_t num
} }
void printDataBlock(SSDataBlock* pBlock, const char* flag) { void printDataBlock(SSDataBlock* pBlock, const char* flag) {
if (pBlock == NULL){ if (pBlock == NULL) {
qDebug("======printDataBlock Block is Null"); qDebug("======printDataBlock Block is Null");
return; return;
} }
char *pBuf = NULL; char* pBuf = NULL;
qDebug("%s", dumpBlockData(pBlock, flag, &pBuf)); qDebug("%s", dumpBlockData(pBlock, flag, &pBuf));
taosMemoryFree(pBuf); taosMemoryFree(pBuf);
} }
@ -1255,10 +1255,10 @@ static void doClearWindows(SAggSupporter* pAggSup, SExprSupp* pSup1, SInterval*
TSKEY* tsCols = (TSKEY*)pTsCol->pData; TSKEY* tsCols = (TSKEY*)pTsCol->pData;
uint64_t* pGpDatas = NULL; uint64_t* pGpDatas = NULL;
if (pBlock->info.type == STREAM_RETRIEVE) { if (pBlock->info.type == STREAM_RETRIEVE) {
SColumnInfoData* pGpCol = taosArrayGet(pBlock->pDataBlock, 2); SColumnInfoData* pGpCol = taosArrayGet(pBlock->pDataBlock, 2);
pGpDatas = (uint64_t*)pGpCol->pData; pGpDatas = (uint64_t*)pGpCol->pData;
} }
int32_t step = 0; int32_t step = 0;
for (int32_t i = 0; i < pBlock->info.rows; i += step) { for (int32_t i = 0; i < pBlock->info.rows; i += step) {
SResultRowInfo dumyInfo; SResultRowInfo dumyInfo;
dumyInfo.cur.pageId = -1; dumyInfo.cur.pageId = -1;
@ -1297,8 +1297,8 @@ bool isCloseWindow(STimeWindow* pWin, STimeWindowAggSupp* pSup) {
return pSup->maxTs != INT64_MIN && pWin->ekey < pSup->maxTs - pSup->waterMark; return pSup->maxTs != INT64_MIN && pWin->ekey < pSup->maxTs - pSup->waterMark;
} }
static int32_t closeIntervalWindow(SHashObj* pHashMap, STimeWindowAggSupp* pSup, static int32_t closeIntervalWindow(SHashObj* pHashMap, STimeWindowAggSupp* pSup, SInterval* pInterval,
SInterval* pInterval, SHashObj* pPullDataMap, SArray* closeWins) { SHashObj* pPullDataMap, SArray* closeWins) {
void* pIte = NULL; void* pIte = NULL;
size_t keyLen = 0; size_t keyLen = 0;
while ((pIte = taosHashIterate(pHashMap, pIte)) != NULL) { while ((pIte = taosHashIterate(pHashMap, pIte)) != NULL) {
@ -1309,15 +1309,18 @@ static int32_t closeIntervalWindow(SHashObj* pHashMap, STimeWindowAggSupp* pSup,
SResultRowInfo dumyInfo; SResultRowInfo dumyInfo;
dumyInfo.cur.pageId = -1; dumyInfo.cur.pageId = -1;
STimeWindow win = getActiveTimeWindow(NULL, &dumyInfo, ts, pInterval, pInterval->precision, NULL); STimeWindow win = getActiveTimeWindow(NULL, &dumyInfo, ts, pInterval, pInterval->precision, NULL);
SWinRes winRe = {.ts = win.skey, .groupId = groupId,}; SWinRes winRe = {
.ts = win.skey,
.groupId = groupId,
};
void* chIds = taosHashGet(pPullDataMap, &winRe, sizeof(SWinRes)); void* chIds = taosHashGet(pPullDataMap, &winRe, sizeof(SWinRes));
if (isCloseWindow(&win, pSup)) { if (isCloseWindow(&win, pSup)) {
if (chIds && pPullDataMap) { if (chIds && pPullDataMap) {
SArray* chAy = *(SArray**) chIds; SArray* chAy = *(SArray**)chIds;
int32_t size = taosArrayGetSize(chAy); int32_t size = taosArrayGetSize(chAy);
qInfo("======window %ld wait child size:%d", win.skey ,size); qInfo("======window %ld wait child size:%d", win.skey, size);
for (int32_t i = 0; i < size; i++) { for (int32_t i = 0; i < size; i++) {
qInfo("======window %ld wait chid id:%d", win.skey ,*(int32_t*)taosArrayGet(chAy, i)); qInfo("======window %ld wait chid id:%d", win.skey, *(int32_t*)taosArrayGet(chAy, i));
} }
continue; continue;
} else if (pPullDataMap) { } else if (pPullDataMap) {
@ -1341,7 +1344,7 @@ static int32_t closeIntervalWindow(SHashObj* pHashMap, STimeWindowAggSupp* pSup,
static void closeChildIntervalWindow(SArray* pChildren, TSKEY maxTs) { static void closeChildIntervalWindow(SArray* pChildren, TSKEY maxTs) {
int32_t size = taosArrayGetSize(pChildren); int32_t size = taosArrayGetSize(pChildren);
for (int32_t i = 0; i < size; i++) { for (int32_t i = 0; i < size; i++) {
SOperatorInfo* pChildOp = taosArrayGetP(pChildren, i); SOperatorInfo* pChildOp = taosArrayGetP(pChildren, i);
SStreamFinalIntervalOperatorInfo* pChInfo = pChildOp->info; SStreamFinalIntervalOperatorInfo* pChInfo = pChildOp->info;
ASSERT(pChInfo->twAggSup.calTrigger == STREAM_TRIGGER_AT_ONCE); ASSERT(pChInfo->twAggSup.calTrigger == STREAM_TRIGGER_AT_ONCE);
pChInfo->twAggSup.maxTs = TMAX(pChInfo->twAggSup.maxTs, maxTs); pChInfo->twAggSup.maxTs = TMAX(pChInfo->twAggSup.maxTs, maxTs);
@ -1431,7 +1434,7 @@ void destroyStreamFinalIntervalOperatorInfo(void* param, int32_t numOfOutput) {
SStreamFinalIntervalOperatorInfo* pInfo = (SStreamFinalIntervalOperatorInfo*)param; SStreamFinalIntervalOperatorInfo* pInfo = (SStreamFinalIntervalOperatorInfo*)param;
cleanupBasicInfo(&pInfo->binfo); cleanupBasicInfo(&pInfo->binfo);
cleanupAggSup(&pInfo->aggSup); cleanupAggSup(&pInfo->aggSup);
//it should be empty. // it should be empty.
taosHashCleanup(pInfo->pPullDataMap); taosHashCleanup(pInfo->pPullDataMap);
taosArrayDestroy(pInfo->pPullWins); taosArrayDestroy(pInfo->pPullWins);
blockDataDestroy(pInfo->pPullDataRes); blockDataDestroy(pInfo->pPullDataRes);
@ -2236,10 +2239,10 @@ bool isDeletedWindow(STimeWindow* pWin, uint64_t groupId, SAggSupporter* pSup) {
return p1 == NULL; return p1 == NULL;
} }
int32_t getNexWindowPos(SInterval* pInterval, SDataBlockInfo* pBlockInfo, TSKEY* tsCols, int32_t getNexWindowPos(SInterval* pInterval, SDataBlockInfo* pBlockInfo, TSKEY* tsCols, int32_t startPos, TSKEY eKey,
int32_t startPos, TSKEY eKey, STimeWindow* pNextWin) { STimeWindow* pNextWin) {
int32_t forwardRows = getNumOfRowsInTimeWindow(pBlockInfo, tsCols, startPos, int32_t forwardRows =
eKey, binarySearchForKey, NULL, TSDB_ORDER_ASC); getNumOfRowsInTimeWindow(pBlockInfo, tsCols, startPos, eKey, binarySearchForKey, NULL, TSDB_ORDER_ASC);
int32_t prevEndPos = forwardRows - 1 + startPos; int32_t prevEndPos = forwardRows - 1 + startPos;
return getNextQualifiedWindow(pInterval, pNextWin, pBlockInfo, tsCols, prevEndPos, TSDB_ORDER_ASC); return getNextQualifiedWindow(pInterval, pNextWin, pBlockInfo, tsCols, prevEndPos, TSDB_ORDER_ASC);
} }
@ -2277,8 +2280,11 @@ static void doHashInterval(SOperatorInfo* pOperatorInfo, SSDataBlock* pSDataBloc
pInfo->interval.precision, NULL); pInfo->interval.precision, NULL);
while (1) { while (1) {
if (IS_FINAL_OP(pInfo) && isCloseWindow(&nextWin, &pInfo->twAggSup) && pInfo->pChildren) { if (IS_FINAL_OP(pInfo) && isCloseWindow(&nextWin, &pInfo->twAggSup) && pInfo->pChildren) {
bool ignore = true; bool ignore = true;
SWinRes winRes = {.ts = nextWin.skey, .groupId = tableGroupId,}; SWinRes winRes = {
.ts = nextWin.skey,
.groupId = tableGroupId,
};
void* chIds = taosHashGet(pInfo->pPullDataMap, &winRes, sizeof(SWinRes)); void* chIds = taosHashGet(pInfo->pPullDataMap, &winRes, sizeof(SWinRes));
if (isDeletedWindow(&nextWin, tableGroupId, &pInfo->aggSup) && !chIds) { if (isDeletedWindow(&nextWin, tableGroupId, &pInfo->aggSup) && !chIds) {
SPullWindowInfo pull = {.window = nextWin, .groupId = tableGroupId}; SPullWindowInfo pull = {.window = nextWin, .groupId = tableGroupId};
@ -2289,7 +2295,7 @@ static void doHashInterval(SOperatorInfo* pOperatorInfo, SSDataBlock* pSDataBloc
int32_t index = -1; int32_t index = -1;
SArray* chArray = NULL; SArray* chArray = NULL;
if (chIds) { if (chIds) {
chArray = *(void**) chIds; chArray = *(void**)chIds;
int32_t chId = getChildIndex(pSDataBlock); int32_t chId = getChildIndex(pSDataBlock);
index = taosArraySearchIdx(chArray, &chId, compareInt32Val, TD_EQ); index = taosArraySearchIdx(chArray, &chId, compareInt32Val, TD_EQ);
} }
@ -2300,7 +2306,7 @@ static void doHashInterval(SOperatorInfo* pOperatorInfo, SSDataBlock* pSDataBloc
taosHashRemove(pInfo->pPullDataMap, &winRes, sizeof(SWinRes)); taosHashRemove(pInfo->pPullDataMap, &winRes, sizeof(SWinRes));
} }
} }
if ( index == -1 || pSDataBlock->info.type == STREAM_PUSH_DATA) { if (index == -1 || pSDataBlock->info.type == STREAM_PUSH_DATA) {
ignore = false; ignore = false;
} }
} }
@ -2385,17 +2391,17 @@ static void doBuildPullDataBlock(SArray* array, int32_t* pIndex, SSDataBlock* pB
if (size - (*pIndex) == 0) { if (size - (*pIndex) == 0) {
return; return;
} }
blockDataEnsureCapacity(pBlock, size - (*pIndex) ); blockDataEnsureCapacity(pBlock, size - (*pIndex));
ASSERT(3 <= taosArrayGetSize(pBlock->pDataBlock)); ASSERT(3 <= taosArrayGetSize(pBlock->pDataBlock));
for (; (*pIndex) < size; (*pIndex)++) { for (; (*pIndex) < size; (*pIndex)++) {
SPullWindowInfo* pWin = taosArrayGet(array, (*pIndex) ); SPullWindowInfo* pWin = taosArrayGet(array, (*pIndex));
SColumnInfoData* pStartTs = (SColumnInfoData*) taosArrayGet(pBlock->pDataBlock, 0); SColumnInfoData* pStartTs = (SColumnInfoData*)taosArrayGet(pBlock->pDataBlock, 0);
colDataAppend(pStartTs, pBlock->info.rows, (const char*)&pWin->window.skey, false); colDataAppend(pStartTs, pBlock->info.rows, (const char*)&pWin->window.skey, false);
SColumnInfoData* pEndTs = (SColumnInfoData*) taosArrayGet(pBlock->pDataBlock, 1); SColumnInfoData* pEndTs = (SColumnInfoData*)taosArrayGet(pBlock->pDataBlock, 1);
colDataAppend(pEndTs, pBlock->info.rows, (const char*)&pWin->window.ekey, false); colDataAppend(pEndTs, pBlock->info.rows, (const char*)&pWin->window.ekey, false);
SColumnInfoData* pGroupId = (SColumnInfoData*) taosArrayGet(pBlock->pDataBlock, 2); SColumnInfoData* pGroupId = (SColumnInfoData*)taosArrayGet(pBlock->pDataBlock, 2);
colDataAppend(pGroupId, pBlock->info.rows, (const char*)&pWin->groupId, false); colDataAppend(pGroupId, pBlock->info.rows, (const char*)&pWin->groupId, false);
pBlock->info.rows++; pBlock->info.rows++;
} }
@ -2408,15 +2414,15 @@ static void doBuildPullDataBlock(SArray* array, int32_t* pIndex, SSDataBlock* pB
void processPushEmpty(SSDataBlock* pBlock, SHashObj* pMap) { void processPushEmpty(SSDataBlock* pBlock, SHashObj* pMap) {
SColumnInfoData* pStartCol = taosArrayGet(pBlock->pDataBlock, 0); SColumnInfoData* pStartCol = taosArrayGet(pBlock->pDataBlock, 0);
TSKEY* tsData = (TSKEY*)pStartCol->pData; TSKEY* tsData = (TSKEY*)pStartCol->pData;
SColumnInfoData* pGroupCol = taosArrayGet(pBlock->pDataBlock, 2); SColumnInfoData* pGroupCol = taosArrayGet(pBlock->pDataBlock, 2);
uint64_t* groupIdData = (uint64_t*)pGroupCol->pData; uint64_t* groupIdData = (uint64_t*)pGroupCol->pData;
int32_t chId = getChildIndex(pBlock); int32_t chId = getChildIndex(pBlock);
for (int32_t i = 0; i < pBlock->info.rows; i++) { for (int32_t i = 0; i < pBlock->info.rows; i++) {
SWinRes winRes = {.ts = tsData[i], .groupId = groupIdData[i]}; SWinRes winRes = {.ts = tsData[i], .groupId = groupIdData[i]};
void* chIds = taosHashGet(pMap, &winRes, sizeof(SWinRes)); void* chIds = taosHashGet(pMap, &winRes, sizeof(SWinRes));
if (chIds) { if (chIds) {
SArray* chArray = *(SArray**) chIds; SArray* chArray = *(SArray**)chIds;
int32_t index = taosArraySearchIdx(chArray, &chId, compareInt32Val, TD_EQ); int32_t index = taosArraySearchIdx(chArray, &chId, compareInt32Val, TD_EQ);
if (index != -1) { if (index != -1) {
taosArrayRemove(chArray, index); taosArrayRemove(chArray, index);
@ -2484,7 +2490,8 @@ static SSDataBlock* doStreamFinalIntervalAgg(SOperatorInfo* pOperator) {
printDataBlock(pBlock, IS_FINAL_OP(pInfo) ? "interval Final recv" : "interval Semi recv"); printDataBlock(pBlock, IS_FINAL_OP(pInfo) ? "interval Final recv" : "interval Semi recv");
maxTs = TMAX(maxTs, pBlock->info.window.ekey); maxTs = TMAX(maxTs, pBlock->info.window.ekey);
if (pBlock->info.type == STREAM_NORMAL || pBlock->info.type == STREAM_PUSH_DATA || pBlock->info.type == STREAM_INVALID) { if (pBlock->info.type == STREAM_NORMAL || pBlock->info.type == STREAM_PUSH_DATA ||
pBlock->info.type == STREAM_INVALID) {
pInfo->binfo.pRes->info.type = pBlock->info.type; pInfo->binfo.pRes->info.type = pBlock->info.type;
} else if (pBlock->info.type == STREAM_CLEAR) { } else if (pBlock->info.type == STREAM_CLEAR) {
SArray* pUpWins = taosArrayInit(8, sizeof(STimeWindow)); SArray* pUpWins = taosArrayInit(8, sizeof(STimeWindow));
@ -2513,8 +2520,7 @@ static SSDataBlock* doStreamFinalIntervalAgg(SOperatorInfo* pOperator) {
continue; continue;
} else if (pBlock->info.type == STREAM_RETRIEVE && !IS_FINAL_OP(pInfo)) { } else if (pBlock->info.type == STREAM_RETRIEVE && !IS_FINAL_OP(pInfo)) {
SArray* pUpWins = taosArrayInit(8, sizeof(STimeWindow)); SArray* pUpWins = taosArrayInit(8, sizeof(STimeWindow));
doClearWindows(&pInfo->aggSup, pSup, &pInfo->interval, 0, pOperator->exprSupp.numOfExprs, doClearWindows(&pInfo->aggSup, pSup, &pInfo->interval, 0, pOperator->exprSupp.numOfExprs, pBlock, pUpWins);
pBlock, pUpWins);
removeResults(pUpWins, pUpdated); removeResults(pUpWins, pUpdated);
taosArrayDestroy(pUpWins); taosArrayDestroy(pUpWins);
if (taosArrayGetSize(pUpdated) > 0) { if (taosArrayGetSize(pUpdated) > 0) {
@ -2552,8 +2558,8 @@ static SSDataBlock* doStreamFinalIntervalAgg(SOperatorInfo* pOperator) {
pInfo->twAggSup.maxTs = TMAX(pInfo->twAggSup.maxTs, maxTs); pInfo->twAggSup.maxTs = TMAX(pInfo->twAggSup.maxTs, maxTs);
if (IS_FINAL_OP(pInfo)) { if (IS_FINAL_OP(pInfo)) {
closeIntervalWindow(pInfo->aggSup.pResultRowHashTable, &pInfo->twAggSup, closeIntervalWindow(pInfo->aggSup.pResultRowHashTable, &pInfo->twAggSup, &pInfo->interval, pInfo->pPullDataMap,
&pInfo->interval, pInfo->pPullDataMap, pUpdated); pUpdated);
closeChildIntervalWindow(pInfo->pChildren, pInfo->twAggSup.maxTs); closeChildIntervalWindow(pInfo->pChildren, pInfo->twAggSup.maxTs);
} }
@ -2874,12 +2880,24 @@ static bool isInWindow(SResultWindowInfo* pWin, TSKEY ts, int64_t gap) {
} }
static SResultWindowInfo* insertNewSessionWindow(SArray* pWinInfos, TSKEY ts, int32_t index) { static SResultWindowInfo* insertNewSessionWindow(SArray* pWinInfos, TSKEY ts, int32_t index) {
SResultWindowInfo win = {.pos.offset = -1, .pos.pageId = -1, .win.skey = ts, .win.ekey = ts, .isOutput = false}; SResultWindowInfo win = {
.pos.offset = -1,
.pos.pageId = -1,
.win.skey = ts,
.win.ekey = ts,
.isOutput = false,
};
return taosArrayInsert(pWinInfos, index, &win); return taosArrayInsert(pWinInfos, index, &win);
} }
static SResultWindowInfo* addNewSessionWindow(SArray* pWinInfos, TSKEY ts) { static SResultWindowInfo* addNewSessionWindow(SArray* pWinInfos, TSKEY ts) {
SResultWindowInfo win = {.pos.offset = -1, .pos.pageId = -1, .win.skey = ts, .win.ekey = ts, .isOutput = false}; SResultWindowInfo win = {
.pos.offset = -1,
.pos.pageId = -1,
.win.skey = ts,
.win.ekey = ts,
.isOutput = false,
};
return taosArrayPush(pWinInfos, &win); return taosArrayPush(pWinInfos, &win);
} }
@ -3277,14 +3295,14 @@ static SSDataBlock* doStreamSessionAgg(SOperatorInfo* pOperator) {
} else if (pOperator->status == OP_RES_TO_RETURN) { } else if (pOperator->status == OP_RES_TO_RETURN) {
doBuildDeleteDataBlock(pInfo->pStDeleted, pInfo->pDelRes, &pInfo->pDelIterator); doBuildDeleteDataBlock(pInfo->pStDeleted, pInfo->pDelRes, &pInfo->pDelIterator);
if (pInfo->pDelRes->info.rows > 0) { if (pInfo->pDelRes->info.rows > 0) {
printDataBlock(pInfo->pDelRes, IS_FINAL_OP(pInfo)? "Final Session" : "Single Session"); printDataBlock(pInfo->pDelRes, IS_FINAL_OP(pInfo) ? "Final Session" : "Single Session");
return pInfo->pDelRes; return pInfo->pDelRes;
} }
doBuildResultDatablock(pOperator, pBInfo, &pInfo->groupResInfo, pInfo->streamAggSup.pResultBuf); doBuildResultDatablock(pOperator, pBInfo, &pInfo->groupResInfo, pInfo->streamAggSup.pResultBuf);
if (pBInfo->pRes->info.rows == 0 || !hasDataInGroupInfo(&pInfo->groupResInfo)) { if (pBInfo->pRes->info.rows == 0 || !hasDataInGroupInfo(&pInfo->groupResInfo)) {
doSetOperatorCompleted(pOperator); doSetOperatorCompleted(pOperator);
} }
printDataBlock(pBInfo->pRes, IS_FINAL_OP(pInfo)? "Final Session" : "Single Session"); printDataBlock(pBInfo->pRes, IS_FINAL_OP(pInfo) ? "Final Session" : "Single Session");
return pBInfo->pRes->info.rows == 0 ? NULL : pBInfo->pRes; return pBInfo->pRes->info.rows == 0 ? NULL : pBInfo->pRes;
} }
@ -3352,11 +3370,11 @@ static SSDataBlock* doStreamSessionAgg(SOperatorInfo* pOperator) {
blockDataEnsureCapacity(pInfo->binfo.pRes, pOperator->resultInfo.capacity); blockDataEnsureCapacity(pInfo->binfo.pRes, pOperator->resultInfo.capacity);
doBuildDeleteDataBlock(pInfo->pStDeleted, pInfo->pDelRes, &pInfo->pDelIterator); doBuildDeleteDataBlock(pInfo->pStDeleted, pInfo->pDelRes, &pInfo->pDelIterator);
if (pInfo->pDelRes->info.rows > 0) { if (pInfo->pDelRes->info.rows > 0) {
printDataBlock(pInfo->pDelRes, IS_FINAL_OP(pInfo)? "Final Session" : "Single Session"); printDataBlock(pInfo->pDelRes, IS_FINAL_OP(pInfo) ? "Final Session" : "Single Session");
return pInfo->pDelRes; return pInfo->pDelRes;
} }
doBuildResultDatablock(pOperator, &pInfo->binfo, &pInfo->groupResInfo, pInfo->streamAggSup.pResultBuf); doBuildResultDatablock(pOperator, &pInfo->binfo, &pInfo->groupResInfo, pInfo->streamAggSup.pResultBuf);
printDataBlock(pBInfo->pRes, IS_FINAL_OP(pInfo)? "Final Session" : "Single Session"); printDataBlock(pBInfo->pRes, IS_FINAL_OP(pInfo) ? "Final Session" : "Single Session");
return pBInfo->pRes->info.rows == 0 ? NULL : pBInfo->pRes; return pBInfo->pRes->info.rows == 0 ? NULL : pBInfo->pRes;
} }
@ -4150,16 +4168,16 @@ _error:
// merge interval operator // merge interval operator
typedef struct SMergeIntervalAggOperatorInfo { typedef struct SMergeIntervalAggOperatorInfo {
SIntervalAggOperatorInfo intervalAggOperatorInfo; SIntervalAggOperatorInfo intervalAggOperatorInfo;
SList* groupIntervals; SList* groupIntervals;
SListIter groupIntervalsIter; SListIter groupIntervalsIter;
bool hasGroupId; bool hasGroupId;
uint64_t groupId; uint64_t groupId;
SSDataBlock* prefetchedBlock; SSDataBlock* prefetchedBlock;
bool inputBlocksFinished; bool inputBlocksFinished;
} SMergeIntervalAggOperatorInfo; } SMergeIntervalAggOperatorInfo;
typedef struct SGroupTimeWindow { typedef struct SGroupTimeWindow {
uint64_t groupId; uint64_t groupId;
STimeWindow window; STimeWindow window;
} SGroupTimeWindow; } SGroupTimeWindow;
@ -4169,7 +4187,8 @@ void destroyMergeIntervalOperatorInfo(void* param, int32_t numOfOutput) {
destroyIntervalOperatorInfo(&miaInfo->intervalAggOperatorInfo, numOfOutput); destroyIntervalOperatorInfo(&miaInfo->intervalAggOperatorInfo, numOfOutput);
} }
static int32_t finalizeWindowResult(SOperatorInfo* pOperatorInfo, uint64_t tableGroupId, STimeWindow* win, SSDataBlock* pResultBlock) { static int32_t finalizeWindowResult(SOperatorInfo* pOperatorInfo, uint64_t tableGroupId, STimeWindow* win,
SSDataBlock* pResultBlock) {
SMergeIntervalAggOperatorInfo* miaInfo = pOperatorInfo->info; SMergeIntervalAggOperatorInfo* miaInfo = pOperatorInfo->info;
SIntervalAggOperatorInfo* iaInfo = &miaInfo->intervalAggOperatorInfo; SIntervalAggOperatorInfo* iaInfo = &miaInfo->intervalAggOperatorInfo;
SExecTaskInfo* pTaskInfo = pOperatorInfo->pTaskInfo; SExecTaskInfo* pTaskInfo = pOperatorInfo->pTaskInfo;
@ -4202,7 +4221,7 @@ static int32_t outputPrevIntervalResult(SOperatorInfo* pOperatorInfo, uint64_t t
SListNode* listNode = NULL; SListNode* listNode = NULL;
while ((listNode = tdListNext(&iter)) != NULL) { while ((listNode = tdListNext(&iter)) != NULL) {
SGroupTimeWindow* prevGrpWin = (SGroupTimeWindow*)listNode->data; SGroupTimeWindow* prevGrpWin = (SGroupTimeWindow*)listNode->data;
if (prevGrpWin->groupId != tableGroupId ) { if (prevGrpWin->groupId != tableGroupId) {
continue; continue;
} }
STimeWindow* prevWin = &prevGrpWin->window; STimeWindow* prevWin = &prevGrpWin->window;
@ -4445,4 +4464,4 @@ _error:
taosMemoryFreeClear(pOperator); taosMemoryFreeClear(pOperator);
pTaskInfo->code = code; pTaskInfo->code = code;
return NULL; return NULL;
} }

15
source/libs/stream/src/streamExec.c
View File

@ -46,15 +46,16 @@ static int32_t streamTaskExecImpl(SStreamTask* pTask, void* data, SArray* pRes)
} }
if (output == NULL) { if (output == NULL) {
if (pItem->type == STREAM_INPUT__DATA_RETRIEVE) { if (pItem->type == STREAM_INPUT__DATA_RETRIEVE) {
//SSDataBlock block = {0}; SSDataBlock block = {0};
//block.info.type = STREAM_PUSH_EMPTY; /*block.info.type = STREAM_PUSH_EMPTY;*/
//block.info.childId = pTask->selfChildId; // block.info.childId = pTask->selfChildId;
SStreamDataBlock* pRetrieveBlock = (SStreamDataBlock*)data; SStreamDataBlock* pRetrieveBlock = (SStreamDataBlock*)data;
ASSERT(taosArrayGetSize(pRetrieveBlock->blocks) == 1); ASSERT(taosArrayGetSize(pRetrieveBlock->blocks) == 1);
SSDataBlock* pBlock = createOneDataBlock(taosArrayGet(pRetrieveBlock->blocks, 0), true); /*SSDataBlock* pBlock = createOneDataBlock(taosArrayGet(pRetrieveBlock->blocks, 0), true);*/
pBlock->info.type = STREAM_PUSH_EMPTY; assignOneDataBlock(&block, taosArrayGet(pRetrieveBlock->blocks, 0));
pBlock->info.childId = pTask->selfChildId; block.info.type = STREAM_PUSH_EMPTY;
taosArrayPush(pRes, pBlock); block.info.childId = pTask->selfChildId;
taosArrayPush(pRes, &block);
} }
break; break;
} }