#3543

Committed 29 Nov 2024 02:58AM UTC coverage: 60.842% (+0.02%) from 60.819%

Build # #3543

Build Type

push

travis-ci

Committed by

web-flow

Commit Message

Merge pull request #28973 from taosdata/merge/mainto3.0

merge: from main to 3.0

Run Details

120460 of 253224 branches covered (47.57%)

Branch coverage included in aggregate %.

706 of 908 new or added lines in 18 files covered. (77.75%)

2401 existing lines in 137 files now uncovered.

201633 of 276172 relevant lines covered (73.01%)

19045673.23 hits per line

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

79.2

/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"

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;
  }

  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) {
  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) {
    SStreamDataBlock* pBlock = (SStreamDataBlock*)data;
    taosArrayDestroyEx(pBlock->blocks, freeItems);
    taosFreeQitem(pBlock);
  }
}

int32_t streamCreateForcewindowTrigger(SStreamTrigger** pTrigger, int32_t trigger, SInterval* pInterval, STimeWindow* pLatestWindow, const char* id) {
  QRY_PARAM_CHECK(pTrigger);
  int64_t         ts = INT64_MIN;
  SStreamTrigger* p = NULL;

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

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

  // let's calculate the previous time window
  SInterval interval = {.interval = trigger,
                        .sliding = trigger,
                        .intervalUnit = pInterval->intervalUnit,
                        .slidingUnit = pInterval->slidingUnit};

  ts = taosGetTimestampMs();

  if (pLatestWindow->skey == INT64_MIN) {
    STimeWindow window = getAlignQueryTimeWindow(&interval, ts - trigger);

    p->pBlock->info.window.skey = window.skey;
    p->pBlock->info.window.ekey = TMAX(ts, window.ekey);
  } else {
    int64_t skey = pLatestWindow->skey + trigger;
    p->pBlock->info.window.skey = skey;
    p->pBlock->info.window.ekey = TMAX(ts, skey + trigger);
  }

  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 streamCreateSinkResTrigger(SStreamTrigger** pTrigger) {
  QRY_PARAM_CHECK(pTrigger);
  SStreamTrigger* p = NULL;

  int32_t code = taosAllocateQitem(sizeof(SStreamTrigger), DEF_QITEM, 0, (void**)&p);
  if (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.type = STREAM_GET_ALL;
  *pTrigger = p;
  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
18	static int32_t streamMergedSubmitNew(SStreamMergedSubmit** pSubmit) {	22,641✔
19	*pSubmit = NULL;	22,641✔
20
21	int32_t code = taosAllocateQitem(sizeof(SStreamMergedSubmit), DEF_QITEM, 0, (void**)pSubmit);	22,641✔
22	if (code) {	22,640!
23	return TSDB_CODE_OUT_OF_MEMORY;	×
24	}
25
26	(*pSubmit)->submits = taosArrayInit(0, sizeof(SPackedData));	22,640✔
27	if ((*pSubmit)->submits == NULL) {	22,638✔
28	taosFreeQitem(*pSubmit);	3✔
29	*pSubmit = NULL;	×
30	return terrno;	×
31	}
32
33	(*pSubmit)->type = STREAM_INPUT__MERGED_SUBMIT;	22,635✔
34	return TSDB_CODE_SUCCESS;	22,635✔
35	}
36
37	static int32_t streamMergeSubmit(SStreamMergedSubmit* pMerged, SStreamDataSubmit* pSubmit) {	354,591✔
38	void* p = taosArrayPush(pMerged->submits, &pSubmit->submit);	354,591✔
39	if (p == NULL) {	354,542!
40	return terrno;	×
41	}
42
43	if (pSubmit->ver > pMerged->ver) {	354,542✔
44	pMerged->ver = pSubmit->ver;	354,541✔
45	}
46	return 0;	354,542✔
47	}
48
49	static void freeItems(void* param) {	19,222✔
50	SSDataBlock* pBlock = param;	19,222✔
51	taosArrayDestroy(pBlock->pDataBlock);	19,222✔
52	}	19,244✔
53
54	int32_t createStreamBlockFromDispatchMsg(const SStreamDispatchReq* pReq, int32_t blockType, int32_t srcVg, SStreamDataBlock** pRes) {	54,106✔
55	SStreamDataBlock* pData = NULL;	54,106✔
56	int32_t code = taosAllocateQitem(sizeof(SStreamDataBlock), DEF_QITEM, pReq->totalLen, (void**)&pData);	54,106✔
57	if (code) {	54,139!
58	return terrno = code;	×
59	}
60
61	pData->type = blockType;	54,139✔
62	pData->srcVgId = srcVg;	54,139✔
63	pData->srcTaskId = pReq->upstreamTaskId;	54,139✔
64
65	int32_t blockNum = pReq->blockNum;	54,139✔
66	SArray* pArray = taosArrayInit_s(sizeof(SSDataBlock), blockNum);	54,139✔
67	if (pArray == NULL) {	54,143✔
68	taosFreeQitem(pData);	1✔
69	return code;	×
70	}
71
72	for (int32_t i = 0; i < blockNum; i++) {	202,598✔
73	SRetrieveTableRsp* pRetrieve = (SRetrieveTableRsp*)taosArrayGetP(pReq->data, i);	148,481✔
74	SSDataBlock* pDataBlock = taosArrayGet(pArray, i);	148,513✔
75	if (pDataBlock == NULL \|\| pRetrieve == NULL) {	148,494!
76	return terrno;	×
77	}
78
79	int32_t compLen = (int32_t)pRetrieve->data;	148,497✔
80	int32_t fullLen = (int32_t)(pRetrieve->data + sizeof(int32_t));	148,497✔
81
82	char* pInput = pRetrieve->data + PAYLOAD_PREFIX_LEN;	148,497✔
83	if (pRetrieve->compressed && compLen < fullLen) {	148,497!
84	char* p = taosMemoryMalloc(fullLen);	×
85	if (p == NULL) {	×
86	return terrno;	×
87	}
88
89	int32_t len = tsDecompressString(pInput, compLen, 1, p, fullLen, ONE_STAGE_COMP, NULL, 0);	×
90	pInput = p;	×
91	}
92
93	const char* pDummy = NULL;	148,497✔
94	code = blockDecode(pDataBlock, pInput, &pDummy);	148,497✔
95	if (code) {	148,440!
96	return code;	×
97	}
98
99	if (pRetrieve->compressed && compLen < fullLen) {	148,440!
100	taosMemoryFree(pInput);	×
101	}
102
103	// TODO: refactor
104	pDataBlock->info.window.skey = be64toh(pRetrieve->skey);	148,440✔
105	pDataBlock->info.window.ekey = be64toh(pRetrieve->ekey);	148,426✔
106	pDataBlock->info.version = be64toh(pRetrieve->version);	148,429✔
107	pDataBlock->info.watermark = be64toh(pRetrieve->watermark);	148,444✔
108	memcpy(pDataBlock->info.parTbName, pRetrieve->parTbName, TSDB_TABLE_NAME_LEN);	148,456✔
109
110	pDataBlock->info.type = pRetrieve->streamBlockType;	148,456✔
111	pDataBlock->info.childId = pReq->upstreamChildId;	148,456✔
112	}
113
114	pData->blocks = pArray;	54,117✔
115	*pRes = pData;	54,117✔
116
117	return code;	54,117✔
118	}
119
120	int32_t createStreamBlockFromResults(SStreamQueueItem* pItem, SStreamTask* pTask, int64_t resultSize, SArray* pRes,	23,987✔
121	SStreamDataBlock** pBlock) {
122	int32_t code = taosAllocateQitem(sizeof(SStreamDataBlock), DEF_QITEM, resultSize, (void**)pBlock);	23,987✔
123	if (code) {	23,986!
UNCOV 124	taosArrayClearEx(pRes, (FDelete)blockDataFreeRes);	×
125	return terrno = code;	×
126	}
127
128	(*pBlock)->srcTaskId = pTask->id.taskId;	23,986✔
129	(*pBlock)->type = STREAM_INPUT__DATA_BLOCK;	23,986✔
130	(*pBlock)->blocks = pRes;	23,986✔
131
132	if (pItem == NULL) {	23,986✔
133	return code;	1,977✔
134	}
135
136	if (pItem->type == STREAM_INPUT__DATA_SUBMIT) {	22,009✔
137	SStreamDataSubmit* pSubmit = (SStreamDataSubmit*)pItem;	4,334✔
138	(*pBlock)->sourceVer = pSubmit->ver;	4,334✔
139	} else if (pItem->type == STREAM_INPUT__MERGED_SUBMIT) {	17,675✔
140	SStreamMergedSubmit* pMerged = (SStreamMergedSubmit*)pItem;	12,744✔
141	(*pBlock)->sourceVer = pMerged->ver;	12,744✔
142	}
143
144	return code;	22,009✔
145	}
146
147	void destroyStreamDataBlock(SStreamDataBlock* pBlock) {	49,600✔
148	if (pBlock == NULL) {	49,600!
149	return;	×
150	}
151
152	taosArrayDestroyEx(pBlock->blocks, (FDelete)blockDataFreeRes);	49,600✔
153	taosFreeQitem(pBlock);	49,605✔
154	}
155
156	int32_t streamRetrieveReqToData(const SStreamRetrieveReq* pReq, SStreamDataBlock* pData, const char* id) {	536✔
157	const char* pDummy = NULL;	536✔
158	SRetrieveTableRsp* pRetrieve = pReq->pRetrieve;	536✔
159	SArray* pArray = taosArrayInit(1, sizeof(SSDataBlock));	536✔
160	if (pArray == NULL) {	535!
161	stError("failed to prepare retrieve block, %s", id);	×
162	return terrno;	×
163	}
164
165	void* px = taosArrayPush(pArray, &(SSDataBlock){0});	535✔
166	if (px == NULL) {	536!
167	taosArrayDestroy(pArray);	×
168	return terrno;	×
169	}
170
171	SSDataBlock* pDataBlock = taosArrayGet(pArray, 0);	536✔
172	if (pDataBlock == NULL) {	536!
173	taosArrayDestroy(pArray);	×
174	return terrno;	×
175	}
176
177	int32_t code = blockDecode(pDataBlock, pRetrieve->data + PAYLOAD_PREFIX_LEN, &pDummy);	536✔
178	if (code) {	535!
179	taosArrayDestroy(pArray);	×
180	return code;	×
181	}
182
183	// TODO: refactor
184	pDataBlock->info.window.skey = be64toh(pRetrieve->skey);	535✔
185	pDataBlock->info.window.ekey = be64toh(pRetrieve->ekey);	535✔
186	pDataBlock->info.version = be64toh(pRetrieve->version);	535✔
187
188	pDataBlock->info.type = pRetrieve->streamBlockType;	536✔
189
190	pData->reqId = pReq->reqId;	536✔
191	pData->blocks = pArray;	536✔
192
193	return code;	536✔
194	}
195
196	int32_t streamDataSubmitNew(SPackedData* pData, int32_t type, SStreamDataSubmit** pSubmit) {	371,880✔
197	SStreamDataSubmit* pDataSubmit = NULL;	371,880✔
198	int32_t code = taosAllocateQitem(sizeof(SStreamDataSubmit), DEF_QITEM, pData->msgLen, (void**)&pDataSubmit);	371,880✔
199	if (code) {	371,898!
200	return code;	×
201	}
202
203	pDataSubmit->ver = pData->ver;	371,898✔
204	pDataSubmit->submit = *pData;	371,898✔
205	pDataSubmit->type = type;	371,898✔
206
207	*pSubmit = pDataSubmit;	371,898✔
208	return TSDB_CODE_SUCCESS;	371,898✔
209	}
210
211	void streamDataSubmitDestroy(SStreamDataSubmit* pDataSubmit) {	17,231✔
212	if (pDataSubmit != NULL && pDataSubmit->type == STREAM_INPUT__DATA_SUBMIT) {	17,231!
213	taosMemoryFree(pDataSubmit->submit.msgStr);	17,231✔
214	taosFreeQitem(pDataSubmit);	17,234✔
215	}
216	}	17,236✔
217
218	// todo handle memory error
219	int32_t streamQueueMergeQueueItem(SStreamQueueItem* dst, SStreamQueueItem* pElem, SStreamQueueItem** pRes) {	351,047✔
220	*pRes = NULL;	351,047✔
221	int32_t code = 0;	351,047✔
222
223	if (dst->type == STREAM_INPUT__DATA_BLOCK && pElem->type == STREAM_INPUT__DATA_BLOCK) {	351,047!
224	SStreamDataBlock* pBlock = (SStreamDataBlock*)dst;	18,943✔
225	SStreamDataBlock* pBlockSrc = (SStreamDataBlock*)pElem;	18,943✔
226	void* px = taosArrayAddAll(pBlock->blocks, pBlockSrc->blocks);	18,943✔
227	if (px == NULL) {	18,936!
228	return terrno;	×
229	}
230
231	taosArrayDestroy(pBlockSrc->blocks);	18,936✔
232	streamQueueItemIncSize(dst, streamQueueItemGetSize(pElem));	18,942✔
233
234	taosFreeQitem(pElem);	18,942✔
235	*pRes = dst;	18,943✔
236	return code;	18,943✔
237	} else if (dst->type == STREAM_INPUT__MERGED_SUBMIT && pElem->type == STREAM_INPUT__DATA_SUBMIT) {	332,104✔
238	SStreamMergedSubmit* pMerged = (SStreamMergedSubmit*)dst;	309,368✔
239	SStreamDataSubmit* pBlockSrc = (SStreamDataSubmit*)pElem;	309,368✔
240
241	code = streamMergeSubmit(pMerged, pBlockSrc);	309,368✔
242	streamQueueItemIncSize(dst, streamQueueItemGetSize(pElem));	309,321✔
243
244	taosFreeQitem(pElem);	309,301✔
245	*pRes = dst;	309,369✔
246	*pRes = dst;	309,369✔
247	return code;	309,369✔
248	} else if (dst->type == STREAM_INPUT__DATA_SUBMIT && pElem->type == STREAM_INPUT__DATA_SUBMIT) {	22,736✔
249	SStreamMergedSubmit* pMerged = NULL;	22,641✔
250	code = streamMergedSubmitNew(&pMerged);	22,641✔
251	if (code != 0) {	22,635!
252	return code;	×
253	}
254
255	streamQueueItemIncSize((SStreamQueueItem*)pMerged, streamQueueItemGetSize(pElem));	22,635✔
256
257	code = streamMergeSubmit(pMerged, (SStreamDataSubmit*)dst);	22,636✔
258	if (code == 0) {	22,632!
259	code = streamMergeSubmit(pMerged, (SStreamDataSubmit*)pElem);	22,632✔
260	}
261
262	taosFreeQitem(dst);	22,627✔
263	taosFreeQitem(pElem);	22,647✔
264
265	pRes = (SStreamQueueItem)pMerged;	22,642✔
266	return code;	22,642✔
267	} else {
268	code = TSDB_CODE_FAILED;	95✔
269	stDebug("block type:%s not merged with existed blocks list, type:%d", streamQueueItemGetTypeStr(pElem->type),	95✔
270	dst->type);
271	return code;	93✔
272	}
273	}
274
275	void streamFreeQitem(SStreamQueueItem* data) {	67,669✔
276	int8_t type = data->type;	67,669✔
277	if (type == STREAM_INPUT__GET_RES) {	67,669✔
278	blockDataDestroy(((SStreamTrigger*)data)->pBlock);	2,594✔
279	taosFreeQitem(data);	2,594✔
280	} else if (type == STREAM_INPUT__DATA_BLOCK \|\| type == STREAM_INPUT__DATA_RETRIEVE) {	65,075✔
281	destroyStreamDataBlock((SStreamDataBlock*)data);	3,361✔
282	} else if (type == STREAM_INPUT__DATA_SUBMIT) {	61,714✔
283	streamDataSubmitDestroy((SStreamDataSubmit*)data);	17,228✔
284	} else if (type == STREAM_INPUT__MERGED_SUBMIT) {	44,486✔
285	SStreamMergedSubmit* pMerge = (SStreamMergedSubmit*)data;	22,648✔
286
287	int32_t sz = taosArrayGetSize(pMerge->submits);	22,648✔
288	for (int32_t i = 0; i < sz; i++) {	377,312✔
289	SPackedData* pSubmit = (SPackedData*)taosArrayGet(pMerge->submits, i);	354,666✔
290	if (pSubmit == NULL) {	354,657!
291	continue;	×
292	}
293	taosMemoryFree(pSubmit->msgStr);	354,657✔
294	}
295
296	taosArrayDestroy(pMerge->submits);	22,646✔
297	taosFreeQitem(pMerge);	22,646✔
298	} else if (type == STREAM_INPUT__REF_DATA_BLOCK) {	21,838✔
299	SStreamRefDataBlock* pRefBlock = (SStreamRefDataBlock*)data;	2,603✔
300	blockDataDestroy(pRefBlock->pBlock);	2,603✔
301	taosFreeQitem(pRefBlock);	2,603✔
302	} else if (type == STREAM_INPUT__CHECKPOINT \|\| type == STREAM_INPUT__CHECKPOINT_TRIGGER \|\|	19,235!
303	type == STREAM_INPUT__TRANS_STATE) {
304	SStreamDataBlock* pBlock = (SStreamDataBlock*)data;	19,243✔
305	taosArrayDestroyEx(pBlock->blocks, freeItems);	19,243✔
306	taosFreeQitem(pBlock);	19,254✔
307	}
308	}	67,692✔
309
310	int32_t streamCreateForcewindowTrigger(SStreamTrigger** pTrigger, int32_t trigger, SInterval* pInterval, STimeWindow* pLatestWindow, const char* id) {	1,316✔
311	QRY_PARAM_CHECK(pTrigger);	1,316!
312	int64_t ts = INT64_MIN;	1,316✔
313	SStreamTrigger* p = NULL;	1,316✔
314
315	int32_t code = taosAllocateQitem(sizeof(SStreamTrigger), DEF_QITEM, 0, (void**)&p);	1,316✔
316	if (code) {	1,316!
317	return code;	×
318	}
319
320	p->type = STREAM_INPUT__GET_RES;	1,316✔
321	p->pBlock = taosMemoryCalloc(1, sizeof(SSDataBlock));	1,316✔
322	if (p->pBlock == NULL) {	1,316!
323	taosFreeQitem(p);	×
324	return terrno;	×
325	}
326
327	// let's calculate the previous time window
328	SInterval interval = {.interval = trigger,	1,316✔
329	.sliding = trigger,
330	.intervalUnit = pInterval->intervalUnit,	1,316✔
331	.slidingUnit = pInterval->slidingUnit};	1,316✔
332
333	ts = taosGetTimestampMs();	1,316✔
334
335	if (pLatestWindow->skey == INT64_MIN) {	1,316✔
336	STimeWindow window = getAlignQueryTimeWindow(&interval, ts - trigger);	67✔
337
338	p->pBlock->info.window.skey = window.skey;	67✔
339	p->pBlock->info.window.ekey = TMAX(ts, window.ekey);	67✔
340	} else {
341	int64_t skey = pLatestWindow->skey + trigger;	1,249✔
342	p->pBlock->info.window.skey = skey;	1,249✔
343	p->pBlock->info.window.ekey = TMAX(ts, skey + trigger);	1,249✔
344	}
345
346	p->pBlock->info.type = STREAM_GET_RESULT;	1,316✔
347	stDebug("s-task:%s force_window_close trigger block generated, window range:%" PRId64 "-%" PRId64, id,	1,316!
348	p->pBlock->info.window.skey, p->pBlock->info.window.ekey);
349
350	*pTrigger = p;	1,316✔
351	return code;	1,316✔
352	}
353
354	int32_t streamCreateSinkResTrigger(SStreamTrigger** pTrigger) {	1,278✔
355	QRY_PARAM_CHECK(pTrigger);	1,278!
356	SStreamTrigger* p = NULL;	1,278✔
357
358	int32_t code = taosAllocateQitem(sizeof(SStreamTrigger), DEF_QITEM, 0, (void**)&p);	1,278✔
359	if (code) {	1,278!
360	return code;	×
361	}
362
363	p->type = STREAM_INPUT__GET_RES;	1,278✔
364	p->pBlock = taosMemoryCalloc(1, sizeof(SSDataBlock));	1,278✔
365	if (p->pBlock == NULL) {	1,278!
366	taosFreeQitem(p);	×
367	return terrno;	×
368	}
369
370	p->pBlock->info.type = STREAM_GET_ALL;	1,278✔
371	*pTrigger = p;	1,278✔
372	return code;	1,278✔
373	}

taosdata / TDengine / #3543

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