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 5
TDengine/source/libs/executor/src/streamNotify.c

913 lines
34 KiB
C
Raw 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 "streamexecutorInt.h"
#include "executorInt.h"
#include "operator.h"
#include "cmdnodes.h"
#include "tdatablock.h"
#define UPDATE_OPERATOR_INFO BIT_FLAG_MASK(0)
#define FILL_HISTORY_OPERATOR BIT_FLAG_MASK(1)
#define RECALCULATE_OPERATOR BIT_FLAG_MASK(2)
#define SEMI_OPERATOR BIT_FLAG_MASK(3)
#define FINAL_OPERATOR BIT_FLAG_MASK(4)
#define SINGLE_OPERATOR BIT_FLAG_MASK(5)
#define NOTIFY_EVENT_NAME_CACHE_LIMIT_MB 16
typedef struct SStreamNotifyEvent {
uint64_t gid;
int64_t eventType;
STimeWindow win;
cJSON* pJson;
} SStreamNotifyEvent;
#define NOTIFY_EVENT_KEY_SIZE \
((sizeof(((struct SStreamNotifyEvent*)0)->gid) + sizeof(((struct SStreamNotifyEvent*)0)->eventType)) + \
sizeof(((struct SStreamNotifyEvent*)0)->win.skey))
static void destroyStreamWindowEvent(void* ptr) {
SStreamNotifyEvent* pEvent = (SStreamNotifyEvent*)ptr;
if (pEvent) {
if (pEvent->pJson) {
cJSON_Delete(pEvent->pJson);
}
*pEvent = (SStreamNotifyEvent){0};
}
}
static void destroyStreamNotifyEventSupp(SStreamNotifyEventSupp* sup) {
if (sup == NULL) return;
taosHashCleanup(sup->pWindowEventHashMap);
taosHashCleanup(sup->pTableNameHashMap);
blockDataDestroy(sup->pEventBlock);
taosArrayDestroy(sup->pSessionKeys);
*sup = (SStreamNotifyEventSupp){0};
}
static int32_t initStreamNotifyEventSupp(SStreamNotifyEventSupp* sup, const char* windowType, int32_t resCapacity) {
int32_t code = TSDB_CODE_SUCCESS;
int32_t lino = 0;
SSDataBlock* pBlock = NULL;
SColumnInfoData infoData = {0};
if (sup == NULL || sup->pWindowEventHashMap != NULL) {
goto _end;
}
code = createDataBlock(&pBlock);
QUERY_CHECK_CODE(code, lino, _end);
pBlock->info.type = STREAM_NOTIFY_EVENT;
pBlock->info.watermark = INT64_MIN;
infoData.info.type = TSDB_DATA_TYPE_VARCHAR;
infoData.info.bytes = tDataTypes[infoData.info.type].bytes;
code = blockDataAppendColInfo(pBlock, &infoData);
QUERY_CHECK_CODE(code, lino, _end);
sup->pWindowEventHashMap = taosHashInit(4096, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY), true, HASH_NO_LOCK);
QUERY_CHECK_NULL(sup->pWindowEventHashMap, code, lino, _end, terrno);
taosHashSetFreeFp(sup->pWindowEventHashMap, destroyStreamWindowEvent);
sup->pTableNameHashMap = taosHashInit(1024, taosGetDefaultHashFunction(TSDB_DATA_TYPE_UBIGINT), false, HASH_NO_LOCK);
QUERY_CHECK_NULL(sup->pTableNameHashMap, code, lino, _end, terrno);
sup->pEventBlock = pBlock;
pBlock = NULL;
code = blockDataEnsureCapacity(sup->pEventBlock, resCapacity);
QUERY_CHECK_CODE(code, lino, _end);
sup->windowType = windowType;
sup->pSessionKeys = taosArrayInit(resCapacity, sizeof(SSessionKey));
QUERY_CHECK_NULL(sup->pSessionKeys, code, lino, _end, terrno);
_end:
if (code != TSDB_CODE_SUCCESS) {
qError("%s failed at line %d since %s", __func__, lino, tstrerror(code));
if (sup) {
destroyStreamNotifyEventSupp(sup);
}
}
if (pBlock != NULL) {
blockDataDestroy(pBlock);
}
return code;
}
int32_t initStreamBasicInfo(SSteamOpBasicInfo* pBasicInfo, const struct SOperatorInfo* pOperator) {
int32_t code = TSDB_CODE_SUCCESS;
int32_t lino = 0;
pBasicInfo->primaryPkIndex = -1;
pBasicInfo->operatorFlag = 0;
code = createSpecialDataBlock(STREAM_DELETE_RESULT, &pBasicInfo->pDelRes);
QUERY_CHECK_CODE(code, lino, _end);
pBasicInfo->pUpdated = taosArrayInit(1024, sizeof(SResultWindowInfo));
QUERY_CHECK_NULL(pBasicInfo->pUpdated, code, lino, _end, terrno);
_hash_fn_t hashFn = taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY);
pBasicInfo->pSeDeleted = tSimpleHashInit(32, hashFn);
const char* windowType = NULL;
// if (IS_NORMAL_INTERVAL_OP(pOperator)) {
// windowType = "Time";
// } else if (IS_NORMAL_SESSION_OP(pOperator)) {
// windowType = "Session";
// } else if (IS_NORMAL_STATE_OP(pOperator)) {
// windowType = "State";
// } else if (IS_NORMAL_EVENT_OP(pOperator)) {
// windowType = "Event";
// } else if (IS_NORMAL_COUNT_OP(pOperator)) {
// windowType = "Count";
// } else {
// return TSDB_CODE_SUCCESS;
// }
code = initStreamNotifyEventSupp(&pBasicInfo->notifyEventSup, windowType, pOperator->resultInfo.capacity);
QUERY_CHECK_CODE(code, lino, _end);
_end:
if (code != TSDB_CODE_SUCCESS) {
qError("%s failed at line %d since %s", __func__, lino, tstrerror(code));
}
return code;
}
void destroyStreamBasicInfo(SSteamOpBasicInfo* pBasicInfo) {
blockDataDestroy(pBasicInfo->pCheckpointRes);
pBasicInfo->pCheckpointRes = NULL;
tSimpleHashCleanup(pBasicInfo->pSeDeleted);
pBasicInfo->pSeDeleted = NULL;
blockDataDestroy(pBasicInfo->pDelRes);
pBasicInfo->pDelRes = NULL;
pBasicInfo->pUpdated = NULL;
pBasicInfo->pTsDataState = NULL;
destroyStreamNotifyEventSupp(&pBasicInfo->notifyEventSup);
}
static int32_t encodeStreamNotifyEventSupp(void** buf, const SStreamNotifyEventSupp* sup) {
int32_t tlen = 0;
void* pIter = NULL;
char* str = NULL;
if (sup == NULL) {
return tlen;
}
tlen += taosEncodeFixedI32(buf, taosHashGetSize(sup->pWindowEventHashMap));
pIter = taosHashIterate(sup->pWindowEventHashMap, NULL);
while (pIter) {
const SStreamNotifyEvent* pEvent = (const SStreamNotifyEvent*)pIter;
str = cJSON_PrintUnformatted(pEvent->pJson);
tlen += taosEncodeFixedU64(buf, pEvent->gid);
tlen += taosEncodeFixedI64(buf, pEvent->eventType);
tlen += taosEncodeFixedI64(buf, pEvent->win.skey);
tlen += taosEncodeFixedI64(buf, pEvent->win.ekey);
tlen += taosEncodeString(buf, str);
cJSON_free(str);
pIter = taosHashIterate(sup->pWindowEventHashMap, pIter);
}
return tlen;
}
static int32_t decodeStreamNotifyEventSupp(void** buf, SStreamNotifyEventSupp* sup) {
int32_t code = TSDB_CODE_SUCCESS;
int32_t lino = 0;
void* p = *buf;
int32_t size = 0;
uint64_t len = 0;
SStreamNotifyEvent item = {0};
p = taosDecodeFixedI32(p, &size);
for (int32_t i = 0; i < size; i++) {
p = taosDecodeFixedU64(p, &item.gid);
p = taosDecodeFixedI64(p, &item.eventType);
p = taosDecodeFixedI64(p, &item.win.skey);
p = taosDecodeFixedI64(p, &item.win.ekey);
p = taosDecodeVariantU64(p, &len);
item.pJson = cJSON_Parse(p);
if (item.pJson == NULL) {
qWarn("failed to parse the json content since %s", cJSON_GetErrorPtr());
}
QUERY_CHECK_NULL(item.pJson, code, lino, _end, TSDB_CODE_INTERNAL_ERROR);
p = POINTER_SHIFT(p, len);
code = taosHashPut(sup->pWindowEventHashMap, &item, NOTIFY_EVENT_KEY_SIZE, &item, sizeof(SStreamNotifyEvent));
QUERY_CHECK_CODE(code, lino, _end);
item.pJson = NULL;
}
*buf = p;
_end:
if (code != TSDB_CODE_SUCCESS) {
qError("%s failed at line %d since %s", __func__, lino, tstrerror(code));
}
destroyStreamWindowEvent(&item);
return code;
}
int32_t encodeStreamBasicInfo(void** buf, const SSteamOpBasicInfo* pBasicInfo) {
return encodeStreamNotifyEventSupp(buf, &pBasicInfo->notifyEventSup);
}
int32_t decodeStreamBasicInfo(void** buf, SSteamOpBasicInfo* pBasicInfo) {
return decodeStreamNotifyEventSupp(buf, &pBasicInfo->notifyEventSup);
}
static void streamNotifyGetEventWindowId(const SSessionKey* pSessionKey, char* buf) {
uint64_t hash = 0;
uint64_t ar[2];
ar[0] = pSessionKey->groupId;
ar[1] = pSessionKey->win.skey;
hash = MurmurHash3_64((char*)ar, sizeof(ar));
buf = u64toaFastLut(hash, buf);
}
#define JSON_CHECK_ADD_ITEM(obj, str, item) \
QUERY_CHECK_CONDITION(cJSON_AddItemToObjectCS(obj, str, item), code, lino, _end, TSDB_CODE_OUT_OF_MEMORY)
static int32_t jsonAddColumnField(const char* colName, int16_t type, bool isNull, const char* pData, cJSON* obj) {
int32_t code = TSDB_CODE_SUCCESS;
int32_t lino = 0;
char* temp = NULL;
QUERY_CHECK_NULL(colName, code, lino, _end, TSDB_CODE_INVALID_PARA);
QUERY_CHECK_CONDITION(isNull || (pData != NULL), code, lino, _end, TSDB_CODE_INVALID_PARA);
QUERY_CHECK_NULL(obj, code, lino, _end, TSDB_CODE_INVALID_PARA);
if (isNull) {
JSON_CHECK_ADD_ITEM(obj, colName, cJSON_CreateNull());
goto _end;
}
switch (type) {
case TSDB_DATA_TYPE_BOOL: {
bool val = *(const bool*)pData;
JSON_CHECK_ADD_ITEM(obj, colName, cJSON_CreateBool(val));
break;
}
case TSDB_DATA_TYPE_TINYINT: {
int8_t val = *(const int8_t*)pData;
JSON_CHECK_ADD_ITEM(obj, colName, cJSON_CreateNumber(val));
break;
}
case TSDB_DATA_TYPE_SMALLINT: {
int16_t val = *(const int16_t*)pData;
JSON_CHECK_ADD_ITEM(obj, colName, cJSON_CreateNumber(val));
break;
}
case TSDB_DATA_TYPE_INT: {
int32_t val = *(const int32_t*)pData;
JSON_CHECK_ADD_ITEM(obj, colName, cJSON_CreateNumber(val));
break;
}
case TSDB_DATA_TYPE_BIGINT:
case TSDB_DATA_TYPE_TIMESTAMP: {
int64_t val = *(const int64_t*)pData;
JSON_CHECK_ADD_ITEM(obj, colName, cJSON_CreateNumber(val));
break;
}
case TSDB_DATA_TYPE_FLOAT: {
float val = *(const float*)pData;
JSON_CHECK_ADD_ITEM(obj, colName, cJSON_CreateNumber(val));
break;
}
case TSDB_DATA_TYPE_DOUBLE: {
double val = *(const double*)pData;
JSON_CHECK_ADD_ITEM(obj, colName, cJSON_CreateNumber(val));
break;
}
case TSDB_DATA_TYPE_VARCHAR:
case TSDB_DATA_TYPE_NCHAR: {
// cJSON requires null-terminated strings, but this data is not null-terminated,
// so we need to manually copy the string and add null termination.
const char* src = varDataVal(pData);
int32_t len = varDataLen(pData);
temp = cJSON_malloc(len + 1);
QUERY_CHECK_NULL(temp, code, lino, _end, TSDB_CODE_OUT_OF_MEMORY);
memcpy(temp, src, len);
temp[len] = '\0';
cJSON* item = cJSON_CreateStringReference(temp);
JSON_CHECK_ADD_ITEM(obj, colName, item);
// let the cjson object to free memory later
item->type &= ~cJSON_IsReference;
temp = NULL;
break;
}
case TSDB_DATA_TYPE_UTINYINT: {
uint8_t val = *(const uint8_t*)pData;
JSON_CHECK_ADD_ITEM(obj, colName, cJSON_CreateNumber(val));
break;
}
case TSDB_DATA_TYPE_USMALLINT: {
uint16_t val = *(const uint16_t*)pData;
JSON_CHECK_ADD_ITEM(obj, colName, cJSON_CreateNumber(val));
break;
}
case TSDB_DATA_TYPE_UINT: {
uint32_t val = *(const uint32_t*)pData;
JSON_CHECK_ADD_ITEM(obj, colName, cJSON_CreateNumber(val));
break;
}
case TSDB_DATA_TYPE_UBIGINT: {
uint64_t val = *(const uint64_t*)pData;
JSON_CHECK_ADD_ITEM(obj, colName, cJSON_CreateNumber(val));
break;
}
default: {
JSON_CHECK_ADD_ITEM(obj, colName, cJSON_CreateStringReference("<Unable to display this data type>"));
break;
}
}
_end:
if (code != TSDB_CODE_SUCCESS) {
qError("%s failed at line %d since %s", __func__, lino, tstrerror(code));
}
if (temp) {
cJSON_free(temp);
}
return code;
}
static cJSON* createBasicAggNotifyEvent(const char* windowType, EStreamNotifyEventType eventType,
const SSessionKey* pSessionKey) {
int32_t code = TSDB_CODE_SUCCESS;
int32_t lino = 0;
const char* eventTypeStr = NULL;
cJSON* event = NULL;
char windowId[32];
char groupId[32];
QUERY_CHECK_NULL(windowType, code, lino, _end, TSDB_CODE_INVALID_PARA);
QUERY_CHECK_NULL(pSessionKey, code, lino, _end, TSDB_CODE_INVALID_PARA);
if (eventType == SNOTIFY_EVENT_WINDOW_OPEN) {
eventTypeStr = "WINDOW_OPEN";
} else if (eventType == SNOTIFY_EVENT_WINDOW_CLOSE) {
eventTypeStr = "WINDOW_CLOSE";
} else if (eventType == SNOTIFY_EVENT_WINDOW_INVALIDATION) {
eventTypeStr = "WINDOW_INVALIDATION";
} else {
QUERY_CHECK_CONDITION(false, code, lino, _end, TSDB_CODE_INVALID_PARA);
}
qDebug("add stream notify event from %s Window, type: %s, start: %" PRId64 ", end: %" PRId64, windowType,
eventTypeStr, pSessionKey->win.skey, pSessionKey->win.ekey);
event = cJSON_CreateObject();
QUERY_CHECK_NULL(event, code, lino, _end, TSDB_CODE_OUT_OF_MEMORY);
// add basic info
streamNotifyGetEventWindowId(pSessionKey, windowId);
JSON_CHECK_ADD_ITEM(event, "eventType", cJSON_CreateStringReference(eventTypeStr));
JSON_CHECK_ADD_ITEM(event, "eventTime", cJSON_CreateNumber(taosGetTimestampMs()));
JSON_CHECK_ADD_ITEM(event, "windowId", cJSON_CreateString(windowId));
JSON_CHECK_ADD_ITEM(event, "windowType", cJSON_CreateStringReference(windowType));
char* p = u64toaFastLut(pSessionKey->groupId, groupId);
JSON_CHECK_ADD_ITEM(event, "groupId", cJSON_CreateString(groupId));
JSON_CHECK_ADD_ITEM(event, "windowStart", cJSON_CreateNumber(pSessionKey->win.skey));
if (eventType != SNOTIFY_EVENT_WINDOW_OPEN) {
if (strcmp(windowType, "Time") == 0) {
JSON_CHECK_ADD_ITEM(event, "windowEnd", cJSON_CreateNumber(pSessionKey->win.ekey + 1));
} else {
JSON_CHECK_ADD_ITEM(event, "windowEnd", cJSON_CreateNumber(pSessionKey->win.ekey));
}
}
_end:
if (code != TSDB_CODE_SUCCESS) {
qError("%s failed at line %d since %s", __func__, lino, tstrerror(code));
terrno = code;
cJSON_Delete(event);
event = NULL;
}
return event;
}
int32_t addEventAggNotifyEvent(EStreamNotifyEventType eventType, const SSessionKey* pSessionKey,
const SSDataBlock* pInputBlock, const SNodeList* pCondCols, int32_t ri,
SStreamNotifyEventSupp* sup, STaskNotifyEventStat* pNotifyEventStat) {
int32_t code = TSDB_CODE_SUCCESS;
int32_t lino = 0;
cJSON* event = NULL;
cJSON* fields = NULL;
cJSON* cond = NULL;
const SNode* pNode = NULL;
int32_t origSize = 0;
int64_t startTime = 0;
int64_t endTime = 0;
SStreamNotifyEvent item = {0};
QUERY_CHECK_NULL(pInputBlock, code, lino, _end, TSDB_CODE_INVALID_PARA);
QUERY_CHECK_NULL(pInputBlock->pDataBlock, code, lino, _end, TSDB_CODE_INVALID_PARA);
QUERY_CHECK_NULL(pCondCols, code, lino, _end, TSDB_CODE_INVALID_PARA);
QUERY_CHECK_NULL(sup, code, lino, _end, TSDB_CODE_INVALID_PARA);
QUERY_CHECK_NULL(sup->windowType, code, lino, _end, TSDB_CODE_INVALID_PARA);
QUERY_CHECK_NULL(pNotifyEventStat, code, lino, _end, TSDB_CODE_INVALID_PARA);
startTime = taosGetMonoTimestampMs();
event = createBasicAggNotifyEvent(sup->windowType, eventType, pSessionKey);
QUERY_CHECK_NULL(event, code, lino, _end, terrno);
// create fields object to store matched column values
fields = cJSON_CreateObject();
QUERY_CHECK_NULL(fields, code, lino, _end, TSDB_CODE_OUT_OF_MEMORY);
FOREACH(pNode, pCondCols) {
const SColumnNode* pColDef = (const SColumnNode*)pNode;
const SColumnInfoData* pColData = taosArrayGet(pInputBlock->pDataBlock, pColDef->slotId);
code = jsonAddColumnField(pColDef->colName, pColData->info.type, colDataIsNull_s(pColData, ri),
colDataGetData(pColData, ri), fields);
QUERY_CHECK_CODE(code, lino, _end);
}
// add trigger condition
cond = cJSON_CreateObject();
QUERY_CHECK_NULL(cond, code, lino, _end, TSDB_CODE_OUT_OF_MEMORY);
JSON_CHECK_ADD_ITEM(cond, "conditionIndex", cJSON_CreateNumber(0));
JSON_CHECK_ADD_ITEM(cond, "fieldValues", fields);
fields = NULL;
JSON_CHECK_ADD_ITEM(event, "triggerCondition", cond);
cond = NULL;
item.gid = pSessionKey->groupId;
item.win = pSessionKey->win;
item.eventType = eventType;
item.pJson = event;
event = NULL;
origSize = taosHashGetSize(sup->pWindowEventHashMap);
code = taosHashPut(sup->pWindowEventHashMap, &item, NOTIFY_EVENT_KEY_SIZE, &item, sizeof(SStreamNotifyEvent));
QUERY_CHECK_CODE(code, lino, _end);
item.pJson = NULL;
endTime = taosGetMonoTimestampMs();
pNotifyEventStat->notifyEventAddTimes++;
pNotifyEventStat->notifyEventAddElems += taosHashGetSize(sup->pWindowEventHashMap) - origSize;
pNotifyEventStat->notifyEventAddCostSec += (endTime - startTime) / 1000.0;
pNotifyEventStat->notifyEventHoldElems = taosHashGetSize(sup->pWindowEventHashMap);
_end:
if (code != TSDB_CODE_SUCCESS) {
qError("%s failed at line %d since %s", __func__, lino, tstrerror(code));
}
destroyStreamWindowEvent(&item);
if (cond != NULL) {
cJSON_Delete(cond);
}
if (fields != NULL) {
cJSON_Delete(fields);
}
if (event != NULL) {
cJSON_Delete(event);
}
return code;
}
int32_t addStateAggNotifyEvent(EStreamNotifyEventType eventType, const SSessionKey* pSessionKey,
const SStateKeys* pCurState, const SStateKeys* pAnotherState, bool onlyUpdate,
SStreamNotifyEventSupp* sup, STaskNotifyEventStat* pNotifyEventStat) {
int32_t code = TSDB_CODE_SUCCESS;
int32_t lino = 0;
cJSON* event = NULL;
int32_t origSize = 0;
int64_t startTime = 0;
int64_t endTime = 0;
SStreamNotifyEvent item = {0};
QUERY_CHECK_NULL(pCurState, code, lino, _end, TSDB_CODE_INVALID_PARA);
QUERY_CHECK_NULL(sup, code, lino, _end, TSDB_CODE_INVALID_PARA);
QUERY_CHECK_NULL(sup->windowType, code, lino, _end, TSDB_CODE_INVALID_PARA);
QUERY_CHECK_NULL(pNotifyEventStat, code, lino, _end, TSDB_CODE_INVALID_PARA);
item.gid = pSessionKey->groupId;
item.win = pSessionKey->win;
item.eventType = eventType;
// Check if the notify event exists for update
if (onlyUpdate && taosHashGet(sup->pWindowEventHashMap, &item, NOTIFY_EVENT_KEY_SIZE) == NULL) {
goto _end;
}
startTime = taosGetMonoTimestampMs();
event = createBasicAggNotifyEvent(sup->windowType, eventType, pSessionKey);
QUERY_CHECK_NULL(event, code, lino, _end, terrno);
// add state value
if (eventType == SNOTIFY_EVENT_WINDOW_OPEN) {
if (pAnotherState) {
code = jsonAddColumnField("prevState", pAnotherState->type, pAnotherState->isNull, pAnotherState->pData, event);
QUERY_CHECK_CODE(code, lino, _end);
} else {
code = jsonAddColumnField("prevState", pCurState->type, true, NULL, event);
QUERY_CHECK_CODE(code, lino, _end);
}
}
code = jsonAddColumnField("curState", pCurState->type, pCurState->isNull, pCurState->pData, event);
QUERY_CHECK_CODE(code, lino, _end);
if (eventType == SNOTIFY_EVENT_WINDOW_CLOSE) {
if (pAnotherState) {
code = jsonAddColumnField("nextState", pAnotherState->type, pAnotherState->isNull, pAnotherState->pData, event);
QUERY_CHECK_CODE(code, lino, _end);
} else {
code = jsonAddColumnField("nextState", pCurState->type, true, NULL, event);
QUERY_CHECK_CODE(code, lino, _end);
}
}
item.pJson = event;
event = NULL;
origSize = taosHashGetSize(sup->pWindowEventHashMap);
code = taosHashPut(sup->pWindowEventHashMap, &item, NOTIFY_EVENT_KEY_SIZE, &item, sizeof(SStreamNotifyEvent));
QUERY_CHECK_CODE(code, lino, _end);
item.pJson = NULL;
endTime = taosGetMonoTimestampMs();
pNotifyEventStat->notifyEventAddTimes++;
pNotifyEventStat->notifyEventAddElems += taosHashGetSize(sup->pWindowEventHashMap) - origSize;
pNotifyEventStat->notifyEventAddCostSec += (endTime - startTime) / 1000.0;
pNotifyEventStat->notifyEventHoldElems = taosHashGetSize(sup->pWindowEventHashMap);
_end:
if (code != TSDB_CODE_SUCCESS) {
qError("%s failed at line %d since %s", __func__, lino, tstrerror(code));
}
destroyStreamWindowEvent(&item);
if (event != NULL) {
cJSON_Delete(event);
}
return code;
}
static int32_t addNormalAggNotifyEvent(EStreamNotifyEventType eventType, const SSessionKey* pSessionKey,
SStreamNotifyEventSupp* sup, STaskNotifyEventStat* pNotifyEventStat) {
int32_t code = TSDB_CODE_SUCCESS;
int32_t lino = 0;
cJSON* event = NULL;
int32_t origSize = 0;
int64_t startTime = 0;
int64_t endTime = 0;
SStreamNotifyEvent item = {0};
QUERY_CHECK_NULL(pSessionKey, code, lino, _end, TSDB_CODE_INVALID_PARA);
QUERY_CHECK_NULL(sup, code, lino, _end, TSDB_CODE_INVALID_PARA);
QUERY_CHECK_NULL(sup->windowType, code, lino, _end, TSDB_CODE_INVALID_PARA);
QUERY_CHECK_NULL(pNotifyEventStat, code, lino, _end, TSDB_CODE_INVALID_PARA);
startTime = taosGetMonoTimestampMs();
event = createBasicAggNotifyEvent(sup->windowType, eventType, pSessionKey);
QUERY_CHECK_NULL(event, code, lino, _end, terrno);
item.gid = pSessionKey->groupId;
item.win = pSessionKey->win;
item.eventType = eventType;
item.pJson = event;
event = NULL;
origSize = taosHashGetSize(sup->pWindowEventHashMap);
code = taosHashPut(sup->pWindowEventHashMap, &item, NOTIFY_EVENT_KEY_SIZE, &item, sizeof(SStreamNotifyEvent));
QUERY_CHECK_CODE(code, lino, _end);
item.pJson = NULL;
endTime = taosGetMonoTimestampMs();
pNotifyEventStat->notifyEventAddTimes++;
pNotifyEventStat->notifyEventAddElems += taosHashGetSize(sup->pWindowEventHashMap) - origSize;
pNotifyEventStat->notifyEventAddCostSec += (endTime - startTime) / 1000.0;
pNotifyEventStat->notifyEventHoldElems = taosHashGetSize(sup->pWindowEventHashMap);
_end:
if (code != TSDB_CODE_SUCCESS) {
qError("%s failed at line %d since %s", __func__, lino, tstrerror(code));
}
destroyStreamWindowEvent(&item);
if (event != NULL) {
cJSON_Delete(event);
}
return code;
}
int32_t addIntervalAggNotifyEvent(EStreamNotifyEventType eventType, const SSessionKey* pSessionKey,
SStreamNotifyEventSupp* sup, STaskNotifyEventStat* pNotifyEventStat) {
return addNormalAggNotifyEvent(eventType, pSessionKey, sup, pNotifyEventStat);
}
int32_t addSessionAggNotifyEvent(EStreamNotifyEventType eventType, const SSessionKey* pSessionKey,
SStreamNotifyEventSupp* sup, STaskNotifyEventStat* pNotifyEventStat) {
return addNormalAggNotifyEvent(eventType, pSessionKey, sup, pNotifyEventStat);
}
int32_t addCountAggNotifyEvent(EStreamNotifyEventType eventType, const SSessionKey* pSessionKey,
SStreamNotifyEventSupp* sup, STaskNotifyEventStat* pNotifyEventStat) {
return addNormalAggNotifyEvent(eventType, pSessionKey, sup, pNotifyEventStat);
}
int32_t addAggResultNotifyEvent(const SSDataBlock* pResultBlock, const SArray* pSessionKeys,
const SSchemaWrapper* pSchemaWrapper, SStreamNotifyEventSupp* sup,
STaskNotifyEventStat* pNotifyEventStat) {
int32_t code = TSDB_CODE_SUCCESS;
int32_t lino = 0;
cJSON* result = NULL;
int32_t origSize = 0;
int64_t startTime = 0;
int64_t endTime = 0;
SStreamNotifyEvent item = {0};
QUERY_CHECK_NULL(pResultBlock, code, lino, _end, TSDB_CODE_INVALID_PARA);
QUERY_CHECK_NULL(pSessionKeys, code, lino, _end, TSDB_CODE_INVALID_PARA);
QUERY_CHECK_NULL(pSchemaWrapper, code, lino, _end, TSDB_CODE_INVALID_PARA);
QUERY_CHECK_NULL(sup, code, lino, _end, TSDB_CODE_INVALID_PARA);
QUERY_CHECK_NULL(pNotifyEventStat, code, lino, _end, TSDB_CODE_INVALID_PARA);
qDebug("add %" PRId64 " stream notify results from window agg", pResultBlock->info.rows);
startTime = taosGetMonoTimestampMs();
origSize = taosHashGetSize(sup->pWindowEventHashMap);
for (int32_t i = 0; i < pResultBlock->info.rows; ++i) {
const SSessionKey* pSessionKey = taosArrayGet(pSessionKeys, i);
item.gid = pSessionKey->groupId;
item.win = pSessionKey->win;
item.eventType = SNOTIFY_EVENT_WINDOW_CLOSE;
SStreamNotifyEvent* pItem = taosHashGet(sup->pWindowEventHashMap, &item, NOTIFY_EVENT_KEY_SIZE);
if (pItem == NULL) {
item.pJson = createBasicAggNotifyEvent(sup->windowType, SNOTIFY_EVENT_WINDOW_CLOSE, pSessionKey);
QUERY_CHECK_NULL(item.pJson, code, lino, _end, terrno);
if (strcmp(sup->windowType, "Event") == 0) {
JSON_CHECK_ADD_ITEM(item.pJson, "triggerCondition", cJSON_CreateNull());
} else if (strcmp(sup->windowType, "State") == 0) {
JSON_CHECK_ADD_ITEM(item.pJson, "curState", cJSON_CreateNull());
JSON_CHECK_ADD_ITEM(item.pJson, "nextState", cJSON_CreateNull());
}
code = taosHashPut(sup->pWindowEventHashMap, &item, NOTIFY_EVENT_KEY_SIZE, &item, sizeof(SStreamNotifyEvent));
QUERY_CHECK_CODE(code, lino, _end);
item.pJson = NULL;
pItem = taosHashGet(sup->pWindowEventHashMap, &item, NOTIFY_EVENT_KEY_SIZE);
QUERY_CHECK_NULL(pItem, code, lino, _end, TSDB_CODE_INTERNAL_ERROR);
}
// convert the result row into json
result = cJSON_CreateObject();
QUERY_CHECK_NULL(result, code, lino, _end, TSDB_CODE_OUT_OF_MEMORY);
for (int32_t j = 0; j < pSchemaWrapper->nCols; ++j) {
const SSchema* pCol = pSchemaWrapper->pSchema + j;
if (pCol->colId - 1 < taosArrayGetSize(pResultBlock->pDataBlock)) {
const SColumnInfoData* pColData = taosArrayGet(pResultBlock->pDataBlock, pCol->colId - 1);
code = jsonAddColumnField(pCol->name, pColData->info.type, colDataIsNull_s(pColData, i),
colDataGetData(pColData, i), result);
QUERY_CHECK_CODE(code, lino, _end);
}
}
JSON_CHECK_ADD_ITEM(pItem->pJson, "result", result);
result = NULL;
}
endTime = taosGetMonoTimestampMs();
pNotifyEventStat->notifyEventAddTimes++;
pNotifyEventStat->notifyEventAddElems += taosHashGetSize(sup->pWindowEventHashMap) - origSize;
pNotifyEventStat->notifyEventAddCostSec += (endTime - startTime) / 1000.0;
pNotifyEventStat->notifyEventHoldElems = taosHashGetSize(sup->pWindowEventHashMap);
_end:
if (code != TSDB_CODE_SUCCESS) {
qError("%s failed at line %d since %s", __func__, lino, tstrerror(code));
}
destroyStreamWindowEvent(&item);
if (result != NULL) {
cJSON_Delete(result);
}
return code;
}
int32_t addAggDeleteNotifyEvent(const SSDataBlock* pDeleteBlock, SStreamNotifyEventSupp* sup,
STaskNotifyEventStat* pNotifyEventStat) {
int32_t code = TSDB_CODE_SUCCESS;
int32_t lino = 0;
SSessionKey sessionKey = {0};
SColumnInfoData* pWstartCol = NULL;
SColumnInfoData* pWendCol = NULL;
SColumnInfoData* pGroupIdCol = NULL;
QUERY_CHECK_NULL(pDeleteBlock, code, lino, _end, TSDB_CODE_INVALID_PARA);
QUERY_CHECK_NULL(sup, code, lino, _end, TSDB_CODE_INVALID_PARA);
QUERY_CHECK_NULL(pNotifyEventStat, code, lino, _end, TSDB_CODE_INVALID_PARA);
qDebug("add %" PRId64 " stream notify delete events from window agg", pDeleteBlock->info.rows);
pWstartCol = taosArrayGet(pDeleteBlock->pDataBlock, START_TS_COLUMN_INDEX);
pWendCol = taosArrayGet(pDeleteBlock->pDataBlock, END_TS_COLUMN_INDEX);
pGroupIdCol = taosArrayGet(pDeleteBlock->pDataBlock, GROUPID_COLUMN_INDEX);
for (int32_t i = 0; i < pDeleteBlock->info.rows; ++i) {
sessionKey.win.skey = *(int64_t*)colDataGetNumData(pWstartCol, i);
sessionKey.win.ekey = *(int64_t*)colDataGetNumData(pWendCol, i);
sessionKey.groupId = *(uint64_t*)colDataGetNumData(pGroupIdCol, i);
code = addNormalAggNotifyEvent(SNOTIFY_EVENT_WINDOW_INVALIDATION, &sessionKey, sup, pNotifyEventStat);
QUERY_CHECK_CODE(code, lino, _end);
}
_end:
if (code != TSDB_CODE_SUCCESS) {
qError("%s failed at line %d since %s", __func__, lino, tstrerror(code));
}
return code;
}
static int32_t streamNotifyGetDestTableName(const SExecTaskInfo* pTaskInfo, uint64_t gid, char** pTableName) {
int32_t code = TSDB_CODE_SUCCESS;
int32_t lino = 0;
const SStorageAPI* pAPI = NULL;
void* tbname = NULL;
int32_t winCode = TSDB_CODE_SUCCESS;
char parTbName[TSDB_TABLE_NAME_LEN];
const SStreamTaskInfo* pStreamInfo = NULL;
QUERY_CHECK_NULL(pTaskInfo, code, lino, _end, TSDB_CODE_INVALID_PARA);
QUERY_CHECK_NULL(pTableName, code, lino, _end, TSDB_CODE_INVALID_PARA);
*pTableName = NULL;
pAPI = &pTaskInfo->storageAPI;
code = pAPI->stateStore.streamStateGetParName((void*)pTaskInfo->streamInfo.pState, gid, &tbname, false, &winCode);
QUERY_CHECK_CODE(code, lino, _end);
if (winCode != TSDB_CODE_SUCCESS) {
parTbName[0] = '\0';
} else {
tstrncpy(parTbName, tbname, sizeof(parTbName));
}
pAPI->stateStore.streamStateFreeVal(tbname);
pStreamInfo = &pTaskInfo->streamInfo;
code = buildSinkDestTableName(parTbName, pStreamInfo->stbFullName, gid, pStreamInfo->newSubTableRule, pTableName);
QUERY_CHECK_CODE(code, lino, _end);
_end:
if (code != TSDB_CODE_SUCCESS) {
qError("%s failed at line %d since %s", __func__, lino, tstrerror(code));
}
return code;
}
int32_t buildNotifyEventBlock(const SExecTaskInfo* pTaskInfo, SStreamNotifyEventSupp* sup,
STaskNotifyEventStat* pNotifyEventStat) {
int32_t code = TSDB_CODE_SUCCESS;
int32_t lino = 0;
int32_t nWindowEvents = 0;
SColumnInfoData* pEventStrCol = NULL;
int64_t startTime = 0;
int64_t endTime = 0;
void* pIter = NULL;
if (pTaskInfo == NULL || sup == NULL || sup->pEventBlock == NULL || pNotifyEventStat == NULL) {
goto _end;
}
QUERY_CHECK_NULL(sup->pEventBlock, code, lino, _end, TSDB_CODE_INVALID_PARA);
startTime = taosGetMonoTimestampMs();
blockDataCleanup(sup->pEventBlock);
nWindowEvents = taosHashGetSize(sup->pWindowEventHashMap);
qDebug("start to build stream notify event block, nWindowEvents: %d", nWindowEvents);
pEventStrCol = taosArrayGet(sup->pEventBlock->pDataBlock, NOTIFY_EVENT_STR_COLUMN_INDEX);
QUERY_CHECK_NULL(pEventStrCol, code, lino, _end, terrno);
// Append all events content into data block.
pIter = taosHashIterate(sup->pWindowEventHashMap, NULL);
while (pIter) {
const SStreamNotifyEvent* pEvent = (const SStreamNotifyEvent*)pIter;
pIter = taosHashIterate(sup->pWindowEventHashMap, pIter);
if (pEvent->eventType == SNOTIFY_EVENT_WINDOW_CLOSE && !cJSON_HasObjectItem(pEvent->pJson, "result")) {
// current WINDOW_CLOSE event cannot be pushed yet due to watermark
continue;
}
// get name of the dest child table
char* tableName = taosHashGet(sup->pTableNameHashMap, &pEvent->gid, sizeof(&pEvent->gid));
if (tableName == NULL) {
code = streamNotifyGetDestTableName(pTaskInfo, pEvent->gid, &tableName);
QUERY_CHECK_CODE(code, lino, _end);
code = taosHashPut(sup->pTableNameHashMap, &pEvent->gid, sizeof(pEvent->gid), tableName, strlen(tableName) + 1);
taosMemoryFreeClear(tableName);
QUERY_CHECK_CODE(code, lino, _end);
tableName = taosHashGet(sup->pTableNameHashMap, &pEvent->gid, sizeof(pEvent->gid));
QUERY_CHECK_NULL(tableName, code, lino, _end, TSDB_CODE_INTERNAL_ERROR);
}
JSON_CHECK_ADD_ITEM(pEvent->pJson, "tableName", cJSON_CreateStringReference(tableName));
// convert the json object into string and append it into the block
char* str = cJSON_PrintUnformatted(pEvent->pJson);
QUERY_CHECK_NULL(str, code, lino, _end, TSDB_CODE_OUT_OF_MEMORY);
int32_t len = strlen(str);
code = varColSetVarData(pEventStrCol, sup->pEventBlock->info.rows, str, len, false);
cJSON_free(str);
QUERY_CHECK_CODE(code, lino, _end);
sup->pEventBlock->info.rows++;
code = taosHashRemove(sup->pWindowEventHashMap, pEvent, NOTIFY_EVENT_KEY_SIZE);
if (code == TSDB_CODE_NOT_FOUND) {
code = TSDB_CODE_SUCCESS;
}
QUERY_CHECK_CODE(code, lino, _end);
if (sup->pEventBlock->info.rows >= sup->pEventBlock->info.capacity) {
break;
}
}
if (taosHashGetMemSize(sup->pTableNameHashMap) >= NOTIFY_EVENT_NAME_CACHE_LIMIT_MB * 1024 * 1024) {
taosHashClear(sup->pTableNameHashMap);
}
endTime = taosGetMonoTimestampMs();
if (sup->pEventBlock->info.rows > 0) {
pNotifyEventStat->notifyEventPushTimes++;
pNotifyEventStat->notifyEventPushElems += sup->pEventBlock->info.rows;
pNotifyEventStat->notifyEventPushCostSec += (endTime - startTime) / 1000.0;
}
pNotifyEventStat->notifyEventHoldElems = taosHashGetSize(sup->pWindowEventHashMap);
_end:
if (code != TSDB_CODE_SUCCESS) {
qError("%s failed at line %d since %s", __func__, lino, tstrerror(code));
}
if (pIter) {
taosHashCancelIterate(sup->pWindowEventHashMap, pIter);
}
return code;
}
int32_t removeOutdatedNotifyEvents(STimeWindowAggSupp* pTwSup, SStreamNotifyEventSupp* sup,
STaskNotifyEventStat* pNotifyEventStat) {
int32_t code = TSDB_CODE_SUCCESS;
int32_t lino = 0;
void* pIter = NULL;
if (pTwSup || sup == NULL || pNotifyEventStat == NULL) {
goto _end;
}
pIter = taosHashIterate(sup->pWindowEventHashMap, NULL);
while (pIter) {
const SStreamNotifyEvent* pEvent = (const SStreamNotifyEvent*)pIter;
pIter = taosHashIterate(sup->pWindowEventHashMap, pIter);
}
pNotifyEventStat->notifyEventHoldElems = taosHashGetSize(sup->pWindowEventHashMap);
_end:
if (code != TSDB_CODE_SUCCESS) {
qError("%s failed at line %d since %s", __func__, lino, tstrerror(code));
}
return code;
}
void setFinalOperatorFlag(SSteamOpBasicInfo* pBasicInfo) {
BIT_FLAG_SET_MASK(pBasicInfo->operatorFlag, FINAL_OPERATOR);
}
bool isFinalOperator(SSteamOpBasicInfo* pBasicInfo) {
return BIT_FLAG_TEST_MASK(pBasicInfo->operatorFlag, FINAL_OPERATOR);
}
void setRecalculateOperatorFlag(SSteamOpBasicInfo* pBasicInfo) {
BIT_FLAG_SET_MASK(pBasicInfo->operatorFlag, RECALCULATE_OPERATOR);
}
void unsetRecalculateOperatorFlag(SSteamOpBasicInfo* pBasicInfo) {
BIT_FLAG_UNSET_MASK(pBasicInfo->operatorFlag, RECALCULATE_OPERATOR);
}
bool isRecalculateOperator(SSteamOpBasicInfo* pBasicInfo) {
return BIT_FLAG_TEST_MASK(pBasicInfo->operatorFlag, RECALCULATE_OPERATOR);
}
void setSingleOperatorFlag(SSteamOpBasicInfo* pBasicInfo) {
BIT_FLAG_SET_MASK(pBasicInfo->operatorFlag, SINGLE_OPERATOR);
}
bool isSingleOperator(SSteamOpBasicInfo* pBasicInfo) {
return BIT_FLAG_TEST_MASK(pBasicInfo->operatorFlag, SINGLE_OPERATOR);
}
Reference in a new issue View git blame Copy permalink