#3842

Committed 07 Apr 2025 11:21AM UTC coverage: 62.696% (-0.3%) from 63.027%

Build # #3842

Build Type

push

travis-ci

Committed by

web-flow

Commit Message

merge: from main to 3.0 branch (#30679)

Run Details

154855 of 315075 branches covered (49.15%)

Branch coverage included in aggregate %.

6 of 8 new or added lines in 5 files covered. (75.0%)

2309 existing lines in 130 files now uncovered.

240176 of 314995 relevant lines covered (76.25%)

19119980.29 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

71.69

/source/libs/stream/src/streamData.c

/*
 * 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 "streamInt.h"
#include "ttime.h"

static int32_t streamMergedSubmitNew(SStreamMergedSubmit** pSubmit) {
  *pSubmit = NULL;

  int32_t code = taosAllocateQitem(sizeof(SStreamMergedSubmit), DEF_QITEM, 0, (void**)pSubmit);
  if (code) {
    return TSDB_CODE_OUT_OF_MEMORY;
  }

  (*pSubmit)->submits = taosArrayInit(0, sizeof(SPackedData));
  if ((*pSubmit)->submits == NULL) {
    taosFreeQitem(*pSubmit);
    *pSubmit = NULL;
    return terrno;
  }

  (*pSubmit)->type = STREAM_INPUT__MERGED_SUBMIT;
  return TSDB_CODE_SUCCESS;
}

static int32_t streamMergeSubmit(SStreamMergedSubmit* pMerged, SStreamDataSubmit* pSubmit) {
  void* p = taosArrayPush(pMerged->submits, &pSubmit->submit);
  if (p == NULL) {
    return terrno;
  }

  if (pSubmit->ver > pMerged->ver) {
    pMerged->ver = pSubmit->ver;
  }
  return 0;
}

static void freeItems(void* param) {
  SSDataBlock* pBlock = param;
  taosArrayDestroy(pBlock->pDataBlock);
}

int32_t createStreamBlockFromDispatchMsg(const SStreamDispatchReq* pReq, int32_t blockType, int32_t srcVg, SStreamDataBlock** pRes) {
  SStreamDataBlock* pData = NULL;
  int32_t code = taosAllocateQitem(sizeof(SStreamDataBlock), DEF_QITEM, pReq->totalLen, (void**)&pData);
  if (code) {
    return terrno = code;
  }

  pData->type = blockType;
  pData->srcVgId = srcVg;
  pData->srcTaskId = pReq->upstreamTaskId;

  int32_t blockNum = pReq->blockNum;
  SArray* pArray = taosArrayInit_s(sizeof(SSDataBlock), blockNum);
  if (pArray == NULL) {
    taosFreeQitem(pData);
    return code;
  }

  for (int32_t i = 0; i < blockNum; i++) {
    SRetrieveTableRsp* pRetrieve = (SRetrieveTableRsp*)taosArrayGetP(pReq->data, i);
    SSDataBlock* pDataBlock = taosArrayGet(pArray, i);
    if (pDataBlock == NULL || pRetrieve == NULL) {
      return terrno;
    }

    int32_t compLen = *(int32_t*)pRetrieve->data;
    int32_t fullLen = *(int32_t*)(pRetrieve->data + sizeof(int32_t));

    char* pInput = pRetrieve->data + PAYLOAD_PREFIX_LEN;
    if (pRetrieve->compressed && compLen < fullLen) {
      char* p = taosMemoryMalloc(fullLen);
      if (p == NULL) {
        return terrno;
      }

      int32_t len = tsDecompressString(pInput, compLen, 1, p, fullLen, ONE_STAGE_COMP, NULL, 0);
      pInput = p;
    }

    const char* pDummy = NULL;
    code = blockDecode(pDataBlock, pInput, &pDummy);
    if (code) {
      return code;
    }

    if (pRetrieve->compressed && compLen < fullLen) {
      taosMemoryFree(pInput);
    }

    // TODO: refactor
    pDataBlock->info.window.skey = be64toh(pRetrieve->skey);
    pDataBlock->info.window.ekey = be64toh(pRetrieve->ekey);
    pDataBlock->info.version = be64toh(pRetrieve->version);
    pDataBlock->info.watermark = be64toh(pRetrieve->watermark);
    memcpy(pDataBlock->info.parTbName, pRetrieve->parTbName, TSDB_TABLE_NAME_LEN);

    pDataBlock->info.type = pRetrieve->streamBlockType;
    pDataBlock->info.childId = pReq->upstreamChildId;
    pDataBlock->info.id.uid = be64toh(pRetrieve->useconds);
  }

  pData->blocks = pArray;
  *pRes = pData;

  return code;
}

int32_t createStreamBlockFromResults(SStreamQueueItem* pItem, SStreamTask* pTask, int64_t resultSize, SArray* pRes,
                                     SStreamDataBlock** pBlock) {
  int32_t code = taosAllocateQitem(sizeof(SStreamDataBlock), DEF_QITEM, resultSize, (void**)pBlock);
  if (code) {
    taosArrayClearEx(pRes, (FDelete)blockDataFreeRes);
    return terrno = code;
  }

  (*pBlock)->srcTaskId = pTask->id.taskId;
  (*pBlock)->type = STREAM_INPUT__DATA_BLOCK;
  (*pBlock)->blocks = pRes;

  if (pItem == NULL) {
    return code;
  }

  if (pItem->type == STREAM_INPUT__DATA_SUBMIT) {
    SStreamDataSubmit* pSubmit = (SStreamDataSubmit*)pItem;
    (*pBlock)->sourceVer = pSubmit->ver;
  } else if (pItem->type == STREAM_INPUT__MERGED_SUBMIT) {
    SStreamMergedSubmit* pMerged = (SStreamMergedSubmit*)pItem;
    (*pBlock)->sourceVer = pMerged->ver;
  }

  return code;
}

void destroyStreamDataBlock(SStreamDataBlock* pBlock) {
  if (pBlock == NULL) {
    return;
  }

  taosArrayDestroyEx(pBlock->blocks, (FDelete)blockDataFreeRes);
  taosFreeQitem(pBlock);
}

int32_t streamRetrieveReqToData(const SStreamRetrieveReq* pReq, SStreamDataBlock* pData, const char* id) {
  const char*        pDummy = NULL;
  SRetrieveTableRsp* pRetrieve = pReq->pRetrieve;
  SArray*            pArray = taosArrayInit(1, sizeof(SSDataBlock));
  if (pArray == NULL) {
    stError("failed to prepare retrieve block, %s", id);
    return terrno;
  }

  void* px = taosArrayPush(pArray, &(SSDataBlock){0});
  if (px == NULL) {
    taosArrayDestroy(pArray);
    return terrno;
  }

  SSDataBlock* pDataBlock = taosArrayGet(pArray, 0);
  if (pDataBlock == NULL) {
    taosArrayDestroy(pArray);
    return terrno;
  }

  int32_t code = blockDecode(pDataBlock, pRetrieve->data + PAYLOAD_PREFIX_LEN, &pDummy);
  if (code) {
    taosArrayDestroy(pArray);
    return code;
  }

  // TODO: refactor
  pDataBlock->info.window.skey = be64toh(pRetrieve->skey);
  pDataBlock->info.window.ekey = be64toh(pRetrieve->ekey);
  pDataBlock->info.version = be64toh(pRetrieve->version);

  pDataBlock->info.type = pRetrieve->streamBlockType;

  pData->reqId = pReq->reqId;
  pData->blocks = pArray;

  return code;
}

int32_t streamDataSubmitNew(SPackedData* pData, int32_t type, SStreamDataSubmit** pSubmit) {
  SStreamDataSubmit* pDataSubmit = NULL;
  int32_t code = taosAllocateQitem(sizeof(SStreamDataSubmit), DEF_QITEM, pData->msgLen, (void**)&pDataSubmit);
  if (code) {
    return code;
  }

  pDataSubmit->ver = pData->ver;
  pDataSubmit->submit = *pData;
  pDataSubmit->type = type;

  *pSubmit = pDataSubmit;
  return TSDB_CODE_SUCCESS;
}

void streamDataSubmitDestroy(SStreamDataSubmit* pDataSubmit) {
  if (pDataSubmit != NULL && pDataSubmit->type == STREAM_INPUT__DATA_SUBMIT) {
    taosMemoryFree(pDataSubmit->submit.msgStr);
    taosFreeQitem(pDataSubmit);
  }
}

// todo handle memory error
int32_t streamQueueMergeQueueItem(SStreamQueueItem* dst, SStreamQueueItem* pElem, SStreamQueueItem** pRes) {
  *pRes = NULL;
  int32_t code = 0;

  if (dst->type == STREAM_INPUT__DATA_BLOCK && pElem->type == STREAM_INPUT__DATA_BLOCK) {
    SStreamDataBlock* pBlock = (SStreamDataBlock*)dst;
    SStreamDataBlock* pBlockSrc = (SStreamDataBlock*)pElem;
    void* px = taosArrayAddAll(pBlock->blocks, pBlockSrc->blocks);
    if (px == NULL) {
      return terrno;
    }

    taosArrayDestroy(pBlockSrc->blocks);
    streamQueueItemIncSize(dst, streamQueueItemGetSize(pElem));

    taosFreeQitem(pElem);
    *pRes = dst;
    return code;
  } else if (dst->type == STREAM_INPUT__MERGED_SUBMIT && pElem->type == STREAM_INPUT__DATA_SUBMIT) {
    SStreamMergedSubmit* pMerged = (SStreamMergedSubmit*)dst;
    SStreamDataSubmit*   pBlockSrc = (SStreamDataSubmit*)pElem;

    code = streamMergeSubmit(pMerged, pBlockSrc);
    streamQueueItemIncSize(dst, streamQueueItemGetSize(pElem));

    taosFreeQitem(pElem);
    *pRes = dst;
    *pRes = dst;
    return code;
  } else if (dst->type == STREAM_INPUT__DATA_SUBMIT && pElem->type == STREAM_INPUT__DATA_SUBMIT) {
    SStreamMergedSubmit* pMerged = NULL;
    code = streamMergedSubmitNew(&pMerged);
    if (code != 0) {
      return code;
    }

    streamQueueItemIncSize((SStreamQueueItem*)pMerged, streamQueueItemGetSize(pElem));

    code = streamMergeSubmit(pMerged, (SStreamDataSubmit*)dst);
    if (code == 0) {
      code = streamMergeSubmit(pMerged, (SStreamDataSubmit*)pElem);
    }

    taosFreeQitem(dst);
    taosFreeQitem(pElem);

    *pRes = (SStreamQueueItem*)pMerged;
    return code;
  } else {
    code = TSDB_CODE_FAILED;
    stDebug("block type:%s not merged with existed blocks list, type:%d", streamQueueItemGetTypeStr(pElem->type),
            dst->type);
    return code;
  }
}

void streamFreeQitem(SStreamQueueItem* data) {
  if (data == NULL) {
    return;
  }

  int8_t type = data->type;
  if (type == STREAM_INPUT__GET_RES) {
    blockDataDestroy(((SStreamTrigger*)data)->pBlock);
    taosFreeQitem(data);
  } else if (type == STREAM_INPUT__DATA_BLOCK || type == STREAM_INPUT__DATA_RETRIEVE) {
    destroyStreamDataBlock((SStreamDataBlock*)data);
  } else if (type == STREAM_INPUT__DATA_SUBMIT) {
    streamDataSubmitDestroy((SStreamDataSubmit*)data);
  } else if (type == STREAM_INPUT__MERGED_SUBMIT) {
    SStreamMergedSubmit* pMerge = (SStreamMergedSubmit*)data;

    int32_t sz = taosArrayGetSize(pMerge->submits);
    for (int32_t i = 0; i < sz; i++) {
      SPackedData* pSubmit = (SPackedData*)taosArrayGet(pMerge->submits, i);
      if (pSubmit == NULL) {
        continue;
      }
      taosMemoryFree(pSubmit->msgStr);
    }

    taosArrayDestroy(pMerge->submits);
    taosFreeQitem(pMerge);
  } else if (type == STREAM_INPUT__REF_DATA_BLOCK) {
    SStreamRefDataBlock* pRefBlock = (SStreamRefDataBlock*)data;
    blockDataDestroy(pRefBlock->pBlock);
    taosFreeQitem(pRefBlock);
  } else if (type == STREAM_INPUT__CHECKPOINT || type == STREAM_INPUT__CHECKPOINT_TRIGGER ||
             type == STREAM_INPUT__TRANS_STATE || type == STREAM_INPUT__RECALCULATE) {
    SStreamDataBlock* pBlock = (SStreamDataBlock*)data;
    taosArrayDestroyEx(pBlock->blocks, freeItems);
    taosFreeQitem(pBlock);
  }
}

int32_t streamCreateForcewindowTrigger(SStreamTrigger** pTrigger, int32_t interval, SInterval* pInterval,
                                       STimeWindow* pLatestWindow, const char* id) {
  QRY_PARAM_CHECK(pTrigger);

  SStreamTrigger* p = NULL;
  int64_t         ts = taosGetTimestamp(pInterval->precision);
  int64_t         skey = pLatestWindow->skey + pInterval->sliding;

  int32_t code = taosAllocateQitem(sizeof(SStreamTrigger), DEF_QITEM, 0, (void**)&p);
  if (code) {
    stError("s-task:%s failed to create force_window trigger, code:%s", id, tstrerror(code));
    return code;
  }

  p->type = STREAM_INPUT__GET_RES;
  p->pBlock = taosMemoryCalloc(1, sizeof(SSDataBlock));
  if (p->pBlock == NULL) {
    taosFreeQitem(p);
    return terrno;
  }

  p->pBlock->info.window.skey = skey;
  p->pBlock->info.window.ekey = TMAX(ts, skey + pInterval->interval);
  p->pBlock->info.type = STREAM_GET_RESULT;

  stDebug("s-task:%s force_window_close trigger block generated, window range:%" PRId64 "-%" PRId64, id,
          p->pBlock->info.window.skey, p->pBlock->info.window.ekey);

  *pTrigger = p;
  return code;
}

int32_t streamCreateTriggerBlock(SStreamTrigger** pTrigger, int32_t type, int32_t blockType) {
  QRY_PARAM_CHECK(pTrigger);

  SStreamTrigger* p = NULL;
  int32_t         code = taosAllocateQitem(sizeof(SStreamTrigger), DEF_QITEM, 0, (void**)&p);
  if (code) {
    return code;
  }

  p->type = (int8_t) type;
  p->pBlock = taosMemoryCalloc(1, sizeof(SSDataBlock));
  if (p->pBlock == NULL) {
    taosFreeQitem(p);
    return terrno;
  }

  p->pBlock->info.type = blockType;
  *pTrigger = p;
  return code;
}

int32_t streamCreateRecalculateBlock(SStreamTask* pTask, SStreamDataBlock** pBlock, int32_t type) {
  int32_t           code = 0;
  SSDataBlock*      p = NULL;
  SStreamDataBlock* pRecalc = NULL;

  if (pBlock != NULL) {
    *pBlock = NULL;
  }

  code = taosAllocateQitem(sizeof(SStreamDataBlock), DEF_QITEM, sizeof(SSDataBlock), (void**)&pRecalc);
  if (code) {
    return code;
  }

  p = taosMemoryCalloc(1, sizeof(SSDataBlock));
  if (p == NULL) {
    code = terrno;
    goto _err;
  }

  pRecalc->type = STREAM_INPUT__RECALCULATE;

  p->info.type = type;
  p->info.rows = 1;
  p->info.childId = pTask->info.selfChildId;

  pRecalc->blocks = taosArrayInit(4, sizeof(SSDataBlock));  // pBlock;
  if (pRecalc->blocks == NULL) {
    code = terrno;
    goto _err;
  }

  void* px = taosArrayPush(pRecalc->blocks, p);
  if (px == NULL) {
    code = terrno;
    goto _err;
  }

  taosMemoryFree(p);
  *pBlock = pRecalc;

  return code;

_err:
  taosMemoryFree(p);
  taosFreeQitem(pRecalc);
  return code;
}

1	/*
2	* Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
3	*
4	* This program is free software: you can use, redistribute, and/or modify
5	* it under the terms of the GNU Affero General Public License, version 3
6	* or later ("AGPL"), as published by the Free Software Foundation.
7	*
8	* This program is distributed in the hope that it will be useful, but WITHOUT
9	* ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
10	* FITNESS FOR A PARTICULAR PURPOSE.
11	*
12	* You should have received a copy of the GNU Affero General Public License
13	* along with this program. If not, see <http://www.gnu.org/licenses/>.
14	*/
15
16	#include "streamInt.h"
17	#include "ttime.h"
18
19	static int32_t streamMergedSubmitNew(SStreamMergedSubmit** pSubmit) {	23,761✔
20	*pSubmit = NULL;	23,761✔
21
22	int32_t code = taosAllocateQitem(sizeof(SStreamMergedSubmit), DEF_QITEM, 0, (void**)pSubmit);	23,761✔
23	if (code) {	23,764!
24	return TSDB_CODE_OUT_OF_MEMORY;	×
25	}
26
27	(*pSubmit)->submits = taosArrayInit(0, sizeof(SPackedData));	23,764✔
28	if ((*pSubmit)->submits == NULL) {	23,765!
UNCOV 29	taosFreeQitem(*pSubmit);	×
30	*pSubmit = NULL;	×
31	return terrno;	×
32	}
33
34	(*pSubmit)->type = STREAM_INPUT__MERGED_SUBMIT;	23,766✔
35	return TSDB_CODE_SUCCESS;	23,766✔
36	}
37
38	static int32_t streamMergeSubmit(SStreamMergedSubmit* pMerged, SStreamDataSubmit* pSubmit) {	473,884✔
39	void* p = taosArrayPush(pMerged->submits, &pSubmit->submit);	473,884✔
40	if (p == NULL) {	473,846!
41	return terrno;	×
42	}
43
44	if (pSubmit->ver > pMerged->ver) {	473,846✔
45	pMerged->ver = pSubmit->ver;	473,841✔
46	}
47	return 0;	473,846✔
48	}
49
50	static void freeItems(void* param) {	19,189✔
51	SSDataBlock* pBlock = param;	19,189✔
52	taosArrayDestroy(pBlock->pDataBlock);	19,189✔
53	}	19,197✔
54
55	int32_t createStreamBlockFromDispatchMsg(const SStreamDispatchReq* pReq, int32_t blockType, int32_t srcVg, SStreamDataBlock** pRes) {	58,423✔
56	SStreamDataBlock* pData = NULL;	58,423✔
57	int32_t code = taosAllocateQitem(sizeof(SStreamDataBlock), DEF_QITEM, pReq->totalLen, (void**)&pData);	58,423✔
58	if (code) {	58,444!
59	return terrno = code;	×
60	}
61
62	pData->type = blockType;	58,444✔
63	pData->srcVgId = srcVg;	58,444✔
64	pData->srcTaskId = pReq->upstreamTaskId;	58,444✔
65
66	int32_t blockNum = pReq->blockNum;	58,444✔
67	SArray* pArray = taosArrayInit_s(sizeof(SSDataBlock), blockNum);	58,444✔
68	if (pArray == NULL) {	58,446!
69	taosFreeQitem(pData);	×
70	return code;	×
71	}
72
73	for (int32_t i = 0; i < blockNum; i++) {	195,874✔
74	SRetrieveTableRsp* pRetrieve = (SRetrieveTableRsp*)taosArrayGetP(pReq->data, i);	137,437✔
75	SSDataBlock* pDataBlock = taosArrayGet(pArray, i);	137,427✔
76	if (pDataBlock == NULL \|\| pRetrieve == NULL) {	137,362!
77	return terrno;	×
78	}
79
80	int32_t compLen = (int32_t)pRetrieve->data;	137,366✔
81	int32_t fullLen = (int32_t)(pRetrieve->data + sizeof(int32_t));	137,366✔
82
83	char* pInput = pRetrieve->data + PAYLOAD_PREFIX_LEN;	137,366✔
84	if (pRetrieve->compressed && compLen < fullLen) {	137,366!
85	char* p = taosMemoryMalloc(fullLen);	×
86	if (p == NULL) {	×
87	return terrno;	×
88	}
89
90	int32_t len = tsDecompressString(pInput, compLen, 1, p, fullLen, ONE_STAGE_COMP, NULL, 0);	×
91	pInput = p;	×
92	}
93
94	const char* pDummy = NULL;	137,366✔
95	code = blockDecode(pDataBlock, pInput, &pDummy);	137,366✔
96	if (code) {	137,324!
97	return code;	×
98	}
99
100	if (pRetrieve->compressed && compLen < fullLen) {	137,324!
101	taosMemoryFree(pInput);	×
102	}
103
104	// TODO: refactor
105	pDataBlock->info.window.skey = be64toh(pRetrieve->skey);	137,324✔
106	pDataBlock->info.window.ekey = be64toh(pRetrieve->ekey);	137,353✔
107	pDataBlock->info.version = be64toh(pRetrieve->version);	137,370✔
108	pDataBlock->info.watermark = be64toh(pRetrieve->watermark);	137,385✔
109	memcpy(pDataBlock->info.parTbName, pRetrieve->parTbName, TSDB_TABLE_NAME_LEN);	137,395✔
110
111	pDataBlock->info.type = pRetrieve->streamBlockType;	137,395✔
112	pDataBlock->info.childId = pReq->upstreamChildId;	137,395✔
113	pDataBlock->info.id.uid = be64toh(pRetrieve->useconds);	137,395✔
114	}
115
116	pData->blocks = pArray;	58,437✔
117	*pRes = pData;	58,437✔
118
119	return code;	58,437✔
120	}
121
122	int32_t createStreamBlockFromResults(SStreamQueueItem* pItem, SStreamTask* pTask, int64_t resultSize, SArray* pRes,	22,733✔
123	SStreamDataBlock** pBlock) {
124	int32_t code = taosAllocateQitem(sizeof(SStreamDataBlock), DEF_QITEM, resultSize, (void**)pBlock);	22,733✔
125	if (code) {	22,733!
126	taosArrayClearEx(pRes, (FDelete)blockDataFreeRes);	×
127	return terrno = code;	×
128	}
129
130	(*pBlock)->srcTaskId = pTask->id.taskId;	22,733✔
131	(*pBlock)->type = STREAM_INPUT__DATA_BLOCK;	22,733✔
132	(*pBlock)->blocks = pRes;	22,733✔
133
134	if (pItem == NULL) {	22,733✔
135	return code;	1,803✔
136	}
137
138	if (pItem->type == STREAM_INPUT__DATA_SUBMIT) {	20,930✔
139	SStreamDataSubmit* pSubmit = (SStreamDataSubmit*)pItem;	1,394✔
140	(*pBlock)->sourceVer = pSubmit->ver;	1,394✔
141	} else if (pItem->type == STREAM_INPUT__MERGED_SUBMIT) {	19,536✔
142	SStreamMergedSubmit* pMerged = (SStreamMergedSubmit*)pItem;	14,995✔
143	(*pBlock)->sourceVer = pMerged->ver;	14,995✔
144	}
145
146	return code;	20,930✔
147	}
148
149	void destroyStreamDataBlock(SStreamDataBlock* pBlock) {	44,120✔
150	if (pBlock == NULL) {	44,120!
151	return;	×
152	}
153
154	taosArrayDestroyEx(pBlock->blocks, (FDelete)blockDataFreeRes);	44,120✔
155	taosFreeQitem(pBlock);	44,127✔
156	}
157
158	int32_t streamRetrieveReqToData(const SStreamRetrieveReq* pReq, SStreamDataBlock* pData, const char* id) {	532✔
159	const char* pDummy = NULL;	532✔
160	SRetrieveTableRsp* pRetrieve = pReq->pRetrieve;	532✔
161	SArray* pArray = taosArrayInit(1, sizeof(SSDataBlock));	532✔
162	if (pArray == NULL) {	531!
163	stError("failed to prepare retrieve block, %s", id);	×
164	return terrno;	×
165	}
166
167	void* px = taosArrayPush(pArray, &(SSDataBlock){0});	531✔
168	if (px == NULL) {	532!
169	taosArrayDestroy(pArray);	×
170	return terrno;	×
171	}
172
173	SSDataBlock* pDataBlock = taosArrayGet(pArray, 0);	532✔
174	if (pDataBlock == NULL) {	532!
175	taosArrayDestroy(pArray);	×
176	return terrno;	×
177	}
178
179	int32_t code = blockDecode(pDataBlock, pRetrieve->data + PAYLOAD_PREFIX_LEN, &pDummy);	532✔
180	if (code) {	530!
181	taosArrayDestroy(pArray);	×
182	return code;	×
183	}
184
185	// TODO: refactor
186	pDataBlock->info.window.skey = be64toh(pRetrieve->skey);	530✔
187	pDataBlock->info.window.ekey = be64toh(pRetrieve->ekey);	530✔
188	pDataBlock->info.version = be64toh(pRetrieve->version);	530✔
189
190	pDataBlock->info.type = pRetrieve->streamBlockType;	531✔
191
192	pData->reqId = pReq->reqId;	531✔
193	pData->blocks = pArray;	531✔
194
195	return code;	531✔
196	}
197
198	int32_t streamDataSubmitNew(SPackedData* pData, int32_t type, SStreamDataSubmit** pSubmit) {	480,841✔
199	SStreamDataSubmit* pDataSubmit = NULL;	480,841✔
200	int32_t code = taosAllocateQitem(sizeof(SStreamDataSubmit), DEF_QITEM, pData->msgLen, (void**)&pDataSubmit);	480,841✔
201	if (code) {	481,026!
202	return code;	×
203	}
204
205	pDataSubmit->ver = pData->ver;	481,026✔
206	pDataSubmit->submit = *pData;	481,026✔
207	pDataSubmit->type = type;	481,026✔
208
209	*pSubmit = pDataSubmit;	481,026✔
210	return TSDB_CODE_SUCCESS;	481,026✔
211	}
212
213	void streamDataSubmitDestroy(SStreamDataSubmit* pDataSubmit) {	7,121✔
214	if (pDataSubmit != NULL && pDataSubmit->type == STREAM_INPUT__DATA_SUBMIT) {	7,121!
215	taosMemoryFree(pDataSubmit->submit.msgStr);	7,121!
216	taosFreeQitem(pDataSubmit);	7,122✔
217	}
218	}	7,122✔
219
220	// todo handle memory error
221	int32_t streamQueueMergeQueueItem(SStreamQueueItem* dst, SStreamQueueItem* pElem, SStreamQueueItem** pRes) {	477,782✔
222	*pRes = NULL;	477,782✔
223	int32_t code = 0;	477,782✔
224
225	if (dst->type == STREAM_INPUT__DATA_BLOCK && pElem->type == STREAM_INPUT__DATA_BLOCK) {	477,782!
226	SStreamDataBlock* pBlock = (SStreamDataBlock*)dst;	27,475✔
227	SStreamDataBlock* pBlockSrc = (SStreamDataBlock*)pElem;	27,475✔
228	void* px = taosArrayAddAll(pBlock->blocks, pBlockSrc->blocks);	27,475✔
229	if (px == NULL) {	27,467!
230	return terrno;	×
231	}
232
233	taosArrayDestroy(pBlockSrc->blocks);	27,467✔
234	streamQueueItemIncSize(dst, streamQueueItemGetSize(pElem));	27,474✔
235
236	taosFreeQitem(pElem);	27,472✔
237	*pRes = dst;	27,475✔
238	return code;	27,475✔
239	} else if (dst->type == STREAM_INPUT__MERGED_SUBMIT && pElem->type == STREAM_INPUT__DATA_SUBMIT) {	450,307!
240	SStreamMergedSubmit* pMerged = (SStreamMergedSubmit*)dst;	426,397✔
241	SStreamDataSubmit* pBlockSrc = (SStreamDataSubmit*)pElem;	426,397✔
242
243	code = streamMergeSubmit(pMerged, pBlockSrc);	426,397✔
244	streamQueueItemIncSize(dst, streamQueueItemGetSize(pElem));	426,348✔
245
246	taosFreeQitem(pElem);	426,317✔
247	*pRes = dst;	426,418✔
248	*pRes = dst;	426,418✔
249	return code;	426,418✔
250	} else if (dst->type == STREAM_INPUT__DATA_SUBMIT && pElem->type == STREAM_INPUT__DATA_SUBMIT) {	23,910✔
251	SStreamMergedSubmit* pMerged = NULL;	23,763✔
252	code = streamMergedSubmitNew(&pMerged);	23,763✔
253	if (code != 0) {	23,764!
254	return code;	×
255	}
256
257	streamQueueItemIncSize((SStreamQueueItem*)pMerged, streamQueueItemGetSize(pElem));	23,764✔
258
259	code = streamMergeSubmit(pMerged, (SStreamDataSubmit*)dst);	23,764✔
260	if (code == 0) {	23,764!
261	code = streamMergeSubmit(pMerged, (SStreamDataSubmit*)pElem);	23,764✔
262	}
263
264	taosFreeQitem(dst);	23,763✔
265	taosFreeQitem(pElem);	23,767✔
266
267	pRes = (SStreamQueueItem)pMerged;	23,766✔
268	return code;	23,766✔
269	} else {
270	code = TSDB_CODE_FAILED;	147✔
271	stDebug("block type:%s not merged with existed blocks list, type:%d", streamQueueItemGetTypeStr(pElem->type),	147!
272	dst->type);
273	return code;	159✔
274	}
275	}
276
277	void streamFreeQitem(SStreamQueueItem* data) {	60,597✔
278	if (data == NULL) {	60,597✔
279	return;	4✔
280	}
281
282	int8_t type = data->type;	60,593✔
283	if (type == STREAM_INPUT__GET_RES) {	60,593✔
284	blockDataDestroy(((SStreamTrigger*)data)->pBlock);	5,929✔
285	taosFreeQitem(data);	5,929✔
286	} else if (type == STREAM_INPUT__DATA_BLOCK \|\| type == STREAM_INPUT__DATA_RETRIEVE) {	54,664✔
287	destroyStreamDataBlock((SStreamDataBlock*)data);	2,236✔
288	} else if (type == STREAM_INPUT__DATA_SUBMIT) {	52,428✔
289	streamDataSubmitDestroy((SStreamDataSubmit*)data);	7,062✔
290	} else if (type == STREAM_INPUT__MERGED_SUBMIT) {	45,366✔
291	SStreamMergedSubmit* pMerge = (SStreamMergedSubmit*)data;	23,764✔
292
293	int32_t sz = taosArrayGetSize(pMerge->submits);	23,764✔
294	for (int32_t i = 0; i < sz; i++) {	497,726✔
295	SPackedData* pSubmit = (SPackedData*)taosArrayGet(pMerge->submits, i);	473,957✔
296	if (pSubmit == NULL) {	473,953!
297	continue;	×
298	}
299	taosMemoryFree(pSubmit->msgStr);	473,953!
300	}
301
302	taosArrayDestroy(pMerge->submits);	23,769✔
303	taosFreeQitem(pMerge);	23,768✔
304	} else if (type == STREAM_INPUT__REF_DATA_BLOCK) {	21,602✔
305	SStreamRefDataBlock* pRefBlock = (SStreamRefDataBlock*)data;	2,409✔
306	blockDataDestroy(pRefBlock->pBlock);	2,409✔
307	taosFreeQitem(pRefBlock);	2,409✔
308	} else if (type == STREAM_INPUT__CHECKPOINT \|\| type == STREAM_INPUT__CHECKPOINT_TRIGGER \|\|	19,193✔
309	type == STREAM_INPUT__TRANS_STATE \|\| type == STREAM_INPUT__RECALCULATE) {	14!
310	SStreamDataBlock* pBlock = (SStreamDataBlock*)data;	19,193✔
311	taosArrayDestroyEx(pBlock->blocks, freeItems);	19,193✔
312	taosFreeQitem(pBlock);	19,199✔
313	}
314	}
315
316	int32_t streamCreateForcewindowTrigger(SStreamTrigger** pTrigger, int32_t interval, SInterval* pInterval,	4,254✔
317	STimeWindow* pLatestWindow, const char* id) {
318	QRY_PARAM_CHECK(pTrigger);	4,254!
319
320	SStreamTrigger* p = NULL;	4,254✔
321	int64_t ts = taosGetTimestamp(pInterval->precision);	4,254✔
322	int64_t skey = pLatestWindow->skey + pInterval->sliding;	4,254✔
323
324	int32_t code = taosAllocateQitem(sizeof(SStreamTrigger), DEF_QITEM, 0, (void**)&p);	4,254✔
325	if (code) {	4,254!
326	stError("s-task:%s failed to create force_window trigger, code:%s", id, tstrerror(code));	×
327	return code;	×
328	}
329
330	p->type = STREAM_INPUT__GET_RES;	4,254✔
331	p->pBlock = taosMemoryCalloc(1, sizeof(SSDataBlock));	4,254!
332	if (p->pBlock == NULL) {	4,254!
333	taosFreeQitem(p);	×
334	return terrno;	×
335	}
336
337	p->pBlock->info.window.skey = skey;	4,254✔
338	p->pBlock->info.window.ekey = TMAX(ts, skey + pInterval->interval);	4,254✔
339	p->pBlock->info.type = STREAM_GET_RESULT;	4,254✔
340
341	stDebug("s-task:%s force_window_close trigger block generated, window range:%" PRId64 "-%" PRId64, id,	4,254!
342	p->pBlock->info.window.skey, p->pBlock->info.window.ekey);
343
344	*pTrigger = p;	4,254✔
345	return code;	4,254✔
346	}
347
348	int32_t streamCreateTriggerBlock(SStreamTrigger** pTrigger, int32_t type, int32_t blockType) {	1,675✔
349	QRY_PARAM_CHECK(pTrigger);	1,675!
350
351	SStreamTrigger* p = NULL;	1,675✔
352	int32_t code = taosAllocateQitem(sizeof(SStreamTrigger), DEF_QITEM, 0, (void**)&p);	1,675✔
353	if (code) {	1,675!
354	return code;	×
355	}
356
357	p->type = (int8_t) type;	1,675✔
358	p->pBlock = taosMemoryCalloc(1, sizeof(SSDataBlock));	1,675!
359	if (p->pBlock == NULL) {	1,675!
360	taosFreeQitem(p);	×
361	return terrno;	×
362	}
363
364	p->pBlock->info.type = blockType;	1,675✔
365	*pTrigger = p;	1,675✔
366	return code;	1,675✔
367	}
368
369	int32_t streamCreateRecalculateBlock(SStreamTask* pTask, SStreamDataBlock** pBlock, int32_t type) {	14✔
370	int32_t code = 0;	14✔
371	SSDataBlock* p = NULL;	14✔
372	SStreamDataBlock* pRecalc = NULL;	14✔
373
374	if (pBlock != NULL) {	14!
375	*pBlock = NULL;	14✔
376	}
377
378	code = taosAllocateQitem(sizeof(SStreamDataBlock), DEF_QITEM, sizeof(SSDataBlock), (void**)&pRecalc);	14✔
379	if (code) {	14!
380	return code;	×
381	}
382
383	p = taosMemoryCalloc(1, sizeof(SSDataBlock));	14!
384	if (p == NULL) {	14!
385	code = terrno;	×
386	goto _err;	×
387	}
388
389	pRecalc->type = STREAM_INPUT__RECALCULATE;	14✔
390
391	p->info.type = type;	14✔
392	p->info.rows = 1;	14✔
393	p->info.childId = pTask->info.selfChildId;	14✔
394
395	pRecalc->blocks = taosArrayInit(4, sizeof(SSDataBlock)); // pBlock;	14✔
396	if (pRecalc->blocks == NULL) {	14!
397	code = terrno;	×
398	goto _err;	×
399	}
400
401	void* px = taosArrayPush(pRecalc->blocks, p);	14✔
402	if (px == NULL) {	14!
403	code = terrno;	×
404	goto _err;	×
405	}
406
407	taosMemoryFree(p);	14!
408	*pBlock = pRecalc;	14✔
409
410	return code;	14✔
411
412	_err:	×
413	taosMemoryFree(p);	×
414	taosFreeQitem(pRecalc);	×
415	return code;	×
416	}

taosdata / TDengine / #3842

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous