#3660

Committed 15 Mar 2025 09:06AM UTC coverage: 62.039% (-1.3%) from 63.314%

Build # #3660

Build Type

push

travis-ci

Committed by

web-flow

Commit Message

feat(stream): support stream processing for virtual tables (#30144)

* enh: add client processing

* enh: add mnode vtables processing

* enh: add mnode vtable processing

* enh: add normal child vtable support

* fix: compile issues

* fix: compile issues

* fix: create stream issues

* fix: multi stream scan issue

* fix: remove debug info

* fix: agg task and task level issues

* fix: correct task output type

* fix: split vtablescan from agg

* fix: memory leak issues

* fix: add limitations

* Update 09-error-code.md

* Update 09-error-code.md

* fix: remove usless case

* feat(stream): extract original table data in source scan task

Implemented functionality in the source task to extract data
corresponding to the virtual table from the original table using WAL.
The extracted data is then sent to the downstream merge task for further
processing.

* feat(stream): multi-way merge using loser tree in virtual merge task

Implemented multi-way merge in the merge task using a loser tree to
combine data from multiple original table into a single virtual table.
The merged virtual table data is then pushed downstream for further
processing.  Introduced memory limit handling during the merge process
with configurable behavior when the memory limit is reached.

* fix(test): remove useless cases

---------

Co-authored-by: dapan1121 <wpan@taosdata.com>
Co-authored-by: Pan Wei <72057773+dapan1121@users.noreply.github.com>

Run Details

154078 of 317582 branches covered (48.52%)

Branch coverage included in aggregate %.

313 of 2391 new or added lines in 34 files covered. (13.09%)

26134 existing lines in 205 files now uncovered.

240261 of 318051 relevant lines covered (75.54%)

16655189.27 hits per line

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

83.69

/source/util/src/tarray.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/>.
 */

#define _DEFAULT_SOURCE
#include "tarray.h"
#include "tcoding.h"

// todo refactor API
#define BOUNDARY_SIZE         1024 * 1024 * 1024  // 1G
#define BOUNDARY_SMALL_FACTOR 1.2
#define BOUNDARY_BIG_FACTOR   2

SArray* taosArrayInit(size_t size, size_t elemSize) {
  if (elemSize == 0) {
    terrno = TSDB_CODE_INVALID_PARA;
    return NULL;
  }

  if (size < TARRAY_MIN_SIZE) {
    size = TARRAY_MIN_SIZE;
  }

  SArray* pArray = taosMemoryMalloc(sizeof(SArray));
  if (pArray == NULL) {
    return NULL;
  }

  pArray->size = 0;
  pArray->pData = taosMemoryCalloc(size, elemSize);
  if (pArray->pData == NULL) {
    taosMemoryFree(pArray);
    return NULL;
  }

  pArray->capacity = size;
  pArray->elemSize = elemSize;
  return pArray;
}

SArray* taosArrayInit_s(size_t elemSize, size_t initialSize) {
  SArray* pArray = taosMemoryMalloc(sizeof(SArray));
  if (pArray == NULL) {
    return NULL;
  }

  pArray->size = initialSize;
  pArray->pData = taosMemoryCalloc(initialSize, elemSize);
  if (pArray->pData == NULL) {
    taosMemoryFree(pArray);
    return NULL;
  }

  pArray->capacity = initialSize;
  pArray->elemSize = elemSize;
  return pArray;
}

static int32_t taosArrayResize(SArray* pArray) {
  size_t size = pArray->capacity;
  size = (size << 1u);

  void* tmp = taosMemoryRealloc(pArray->pData, size * pArray->elemSize);
  if (tmp == NULL) {  // reallocate failed, the original buffer remains
    return terrno;
  }

  pArray->pData = tmp;
  pArray->capacity = size;

  return 0;
}

int32_t taosArrayEnsureCap(SArray* pArray, size_t newCap) {
  if (newCap > pArray->capacity) {
    float factor = BOUNDARY_BIG_FACTOR;
    if (newCap * pArray->elemSize > BOUNDARY_SIZE) {
      factor = BOUNDARY_SMALL_FACTOR;
    }

    size_t tsize = (pArray->capacity * factor);
    while (newCap > tsize) {
      size_t newSize = (tsize * factor);
      tsize = (newSize == tsize) ? (tsize + 2) : newSize;
    }

    char* p = taosMemoryRealloc(pArray->pData, tsize * pArray->elemSize);
    if (p == NULL) {
      return terrno;
    }

    pArray->pData = p;
    pArray->capacity = tsize;
  }
  return 0;
}

void* taosArrayAddBatch(SArray* pArray, const void* pData, int32_t nEles) {
  if (pData == NULL) {
    terrno = TSDB_CODE_INVALID_PARA;
    return NULL;
  }

  int32_t code = taosArrayEnsureCap(pArray, pArray->size + nEles);
  if (code) {
    terrno = code;
    return NULL;
  }

  void* dst = TARRAY_GET_ELEM(pArray, pArray->size);
  memcpy(dst, pData, pArray->elemSize * nEles);

  pArray->size += nEles;
  return dst;
}

void taosArrayRemoveDuplicate(SArray* pArray, __compar_fn_t comparFn, void (*fp)(void*)) {
  size_t size = pArray->size;
  if (size <= 1) {
    return;
  }

  int32_t pos = 0;
  for (int32_t i = 1; i < size; ++i) {
    char* p1 = taosArrayGet(pArray, pos);
    char* p2 = taosArrayGet(pArray, i);

    if (comparFn(p1, p2) == 0) {
      // do nothing
    } else {
      if (pos + 1 != i) {
        void* p = taosArrayGet(pArray, pos + 1);
        if (fp != NULL) {
          fp(p);
        }

        taosArraySet(pArray, pos + 1, p2);
        memset(TARRAY_GET_ELEM(pArray, i), 0, pArray->elemSize);
        pos += 1;
      } else {
        pos += 1;
      }
    }
  }

  if (fp != NULL) {
    for (int32_t i = pos + 1; i < pArray->size; ++i) {
      void* p = taosArrayGet(pArray, i);
      fp(p);
    }
  }

  pArray->size = pos + 1;
}

void* taosArrayAddAll(SArray* pArray, const SArray* pInput) {
  if (pInput) {
    return taosArrayAddBatch(pArray, pInput->pData, (int32_t)taosArrayGetSize(pInput));
  } else {
    terrno = TSDB_CODE_INVALID_PARA;
    return NULL;
  }
}

void* taosArrayReserve(SArray* pArray, int32_t num) {
  int32_t code = taosArrayEnsureCap(pArray, pArray->size + num);
  if (code) {
    terrno = code;
    return NULL;
  }

  void* dst = TARRAY_GET_ELEM(pArray, pArray->size);
  pArray->size += num;

  memset(dst, 0, num * pArray->elemSize);

  return dst;
}

void* taosArrayPop(SArray* pArray) {
  if (pArray->size == 0) {
    return NULL;
  }

  pArray->size -= 1;
  return TARRAY_GET_ELEM(pArray, pArray->size);
}

void* taosArrayGet(const SArray* pArray, size_t index) {
  if (NULL == pArray) {
    terrno = TSDB_CODE_INVALID_PARA;
    uError("failed to return value from array of null ptr");
    return NULL;
  }

  if (index >= pArray->size) {
    uError("index is out of range, current:%" PRIzu " max:%" PRIzu, index, pArray->size);
    terrno = TSDB_CODE_OUT_OF_RANGE;
    return NULL;
  }

  return TARRAY_GET_ELEM(pArray, index);
}

void* taosArrayGetP(const SArray* pArray, size_t index) {
  void** p = taosArrayGet(pArray, index);
  if (p == NULL) {
    return NULL;
  }
  return *p;
}

void* taosArrayGetLast(const SArray* pArray) {
  if (pArray->size == 0) {
    terrno = TSDB_CODE_INVALID_PARA;
    return NULL;
  }

  return TARRAY_GET_ELEM(pArray, pArray->size - 1);
}

size_t taosArrayGetSize(const SArray* pArray) {
  if (pArray == NULL) {
    return 0;
  }
  return TARRAY_SIZE(pArray);
}

void* taosArrayInsert(SArray* pArray, size_t index, const void* pData) {
  if (pArray == NULL || pData == NULL) {
    terrno = TSDB_CODE_INVALID_PARA;
    return NULL;
  }

  if (index >= pArray->size) {
    return taosArrayPush(pArray, pData);
  }

  if (pArray->size >= pArray->capacity) {
    int32_t ret = taosArrayResize(pArray);
    if (ret < 0) {
      terrno = ret;
      return NULL;
    }
  }

  void* dst = TARRAY_GET_ELEM(pArray, index);

  int32_t remain = (int32_t)(pArray->size - index);
  memmove((char*)dst + pArray->elemSize, (char*)dst, pArray->elemSize * remain);
  memcpy(dst, pData, pArray->elemSize);

  pArray->size += 1;

  return dst;
}

void taosArraySet(SArray* pArray, size_t index, void* pData) {
  if (index < pArray->size) {
    memcpy(TARRAY_GET_ELEM(pArray, index), pData, pArray->elemSize);
  }
}

void taosArrayPopFrontBatch(SArray* pArray, size_t cnt) {
  if (cnt > pArray->size) {
    cnt = pArray->size;
  }

  pArray->size = pArray->size - cnt;
  if (pArray->size == 0 || cnt == 0) {
    return;
  }
  memmove(pArray->pData, (char*)pArray->pData + cnt * pArray->elemSize, pArray->size * pArray->elemSize);
}

void taosArrayPopTailBatch(SArray* pArray, size_t cnt) {
  if (cnt >= pArray->size) {
    cnt = pArray->size;
  }

  pArray->size = pArray->size - cnt;
}

void taosArrayRemove(SArray* pArray, size_t index) {
  if (!(index < pArray->size)) {
    return;
  }

  if (index == pArray->size - 1) {
    void* t = taosArrayPop(pArray);
    return;
  }

  size_t remain = pArray->size - index - 1;
  memmove((char*)pArray->pData + index * pArray->elemSize, (char*)pArray->pData + (index + 1) * pArray->elemSize,
          remain * pArray->elemSize);
  pArray->size -= 1;
}

void taosArrayRemoveBatch(SArray* pArray, size_t index, size_t num, FDelete fp) {
  if (index + num <= pArray->size) {
    if (fp) {
      for (int32_t i = 0; i < num; i++) {
        fp(taosArrayGet(pArray, index + i));
      }
    }

    memmove((char*)pArray->pData + index * pArray->elemSize, (char*)pArray->pData + (index + num) * pArray->elemSize,
            (pArray->size - index - num) * pArray->elemSize);
    pArray->size -= num;
  }
}

SArray* taosArrayFromList(const void* src, size_t size, size_t elemSize) {
  if (elemSize <= 0) {
    terrno = TSDB_CODE_INVALID_PARA;
    return NULL;
  }

  SArray* pDst = taosArrayInit(size, elemSize);
  if (pDst) {
    memcpy(pDst->pData, src, elemSize * size);
    pDst->size = size;
  }

  return pDst;
}

SArray* taosArrayDup(const SArray* pSrc, __array_item_dup_fn_t fn) {
  if (NULL == pSrc) {
    terrno = TSDB_CODE_INVALID_PARA;
    return NULL;
  }

  if (pSrc->size == 0) {  // empty array list
    return taosArrayInit(8, pSrc->elemSize);
  }

  SArray* dst = taosArrayInit(pSrc->size, pSrc->elemSize);

  if (dst) {
    if (fn == NULL) {
      memcpy(dst->pData, pSrc->pData, pSrc->elemSize * pSrc->size);
    } else {
      for (int32_t i = 0; i < pSrc->size; ++i) {
        void* p = fn(taosArrayGetP(pSrc, i));
        memcpy(((char*)dst->pData) + i * dst->elemSize, &p, dst->elemSize);
      }
    }

    dst->size = pSrc->size;
  }

  return dst;
}

void taosArrayClear(SArray* pArray) {
  if (pArray == NULL) return;
  pArray->size = 0;
}

void taosArrayClearEx(SArray* pArray, void (*fp)(void*)) {
  if (pArray == NULL) return;
  if (fp == NULL) {
    pArray->size = 0;
    return;
  }

  for (int32_t i = 0; i < pArray->size; ++i) {
    fp(TARRAY_GET_ELEM(pArray, i));
  }

  pArray->size = 0;
}
void taosArrayClearP(SArray* pArray, void (*fp)(void*)) {
  // if (pArray == NULL) return;
  // if (fp == NULL) {
  //   pArray->size = 0;
  //   return;
  // }

  // for (int32_t i = 0; i < pArray->size; ++i) {
  //   fp(TARRAY_GET_ELEM(pArray, i));
  // }
  if (pArray) {
    if (NULL == fp) {
      for (int32_t i = 0; i < pArray->size; i++) {
        taosMemoryFree(*(void**)TARRAY_GET_ELEM(pArray, i));
      }
    } else {
      for (int32_t i = 0; i < pArray->size; i++) {
        fp(*(void**)TARRAY_GET_ELEM(pArray, i));
      }
    }
  }
  taosArrayClear(pArray);
}

void taosArrayDestroy(SArray* pArray) {
  if (pArray) {
    taosMemoryFree(pArray->pData);
    taosMemoryFree(pArray);
  }
}

void taosArrayDestroyP(SArray* pArray, FDelete fp) {
  if (pArray) {
    if (NULL == fp) {
      for (int32_t i = 0; i < pArray->size; i++) {
        taosMemoryFree(*(void**)TARRAY_GET_ELEM(pArray, i));
      }
    } else {
      for (int32_t i = 0; i < pArray->size; i++) {
        fp(*(void**)TARRAY_GET_ELEM(pArray, i));
      }
    }
    taosArrayDestroy(pArray);
  }
}

void taosArrayDestroyEx(SArray* pArray, FDelete fp) {
  if (pArray == NULL) {
    return;
  }

  if (fp == NULL) {
    taosArrayDestroy(pArray);
    return;
  }

  for (int32_t i = 0; i < pArray->size; ++i) {
    fp(TARRAY_GET_ELEM(pArray, i));
  }

  taosArrayDestroy(pArray);
}

void taosArraySort(SArray* pArray, __compar_fn_t compar) {
  taosSort(pArray->pData, pArray->size, pArray->elemSize, compar);
}

int32_t taosArrayMSort(SArray* pArray, __compar_fn_t compar) {
  return taosMergeSort(pArray->pData, pArray->size, pArray->elemSize, compar);
}

void* taosArraySearch(const SArray* pArray, const void* key, __compar_fn_t comparFn, int32_t flags) {
  return taosbsearch(key, pArray->pData, pArray->size, pArray->elemSize, comparFn, flags);
}

int32_t taosArraySearchIdx(const SArray* pArray, const void* key, __compar_fn_t comparFn, int32_t flags) {
  void* item = taosArraySearch(pArray, key, comparFn, flags);
  return item == NULL ? -1 : (int32_t)((char*)item - (char*)pArray->pData) / pArray->elemSize;
}

#ifdef BUILD_NO_CALL
static int32_t taosArrayPartition(SArray* pArray, int32_t i, int32_t j, __ext_compar_fn_t fn, const void* userData) {
  void* key = taosArrayGetP(pArray, i);
  while (i < j) {
    while (i < j && fn(taosArrayGetP(pArray, j), key, userData) >= 0) {
      j--;
    }
    if (i < j) {
      void* a = taosArrayGetP(pArray, j);
      taosArraySet(pArray, i, &a);
    }
    while (i < j && fn(taosArrayGetP(pArray, i), key, userData) <= 0) {
      i++;
    }
    if (i < j) {
      void* a = taosArrayGetP(pArray, i);
      taosArraySet(pArray, j, &a);
    }
  }
  taosArraySet(pArray, i, &key);
  return i;
}

static void taosArrayQuicksortImpl(SArray* pArray, int32_t low, int32_t high, __ext_compar_fn_t fn, const void* param) {
  if (low < high) {
    int32_t idx = taosArrayPartition(pArray, low, high, fn, param);
    taosArrayQuicksortImpl(pArray, low, idx - 1, fn, param);
    taosArrayQuicksortImpl(pArray, idx + 1, high, fn, param);
  }
}

static void taosArrayQuickSort(SArray* pArray, __ext_compar_fn_t fn, const void* param) {
  if (pArray->size <= 1) {
    return;
  }
  taosArrayQuicksortImpl(pArray, 0, (int32_t)(taosArrayGetSize(pArray) - 1), fn, param);
}

static void taosArrayInsertSort(SArray* pArray, __ext_compar_fn_t fn, const void* param) {
  if (pArray->size <= 1) {
    return;
  }

  for (int32_t i = 1; i <= pArray->size - 1; ++i) {
    for (int32_t j = i; j > 0; --j) {
      if (fn(taosArrayGetP(pArray, j), taosArrayGetP(pArray, j - 1), param) == -1) {
        void* a = taosArrayGetP(pArray, j);
        void* b = taosArrayGetP(pArray, j - 1);
        taosArraySet(pArray, j - 1, &a);
        taosArraySet(pArray, j, &b);
      } else {
        break;
      }
    }
  }
}
#endif

int32_t taosEncodeArray(void** buf, const SArray* pArray, FEncode encode) {
  int32_t tlen = 0;
  int32_t sz = pArray->size;
  tlen += taosEncodeFixedI32(buf, sz);
  for (int32_t i = 0; i < sz; i++) {
    void* data = taosArrayGetP(pArray, i);
    tlen += encode(buf, data);
  }
  return tlen;
}

void* taosDecodeArray(const void* buf, SArray** pArray, FDecode decode, int32_t dataSz, int8_t sver) {
  int32_t sz;
  buf = taosDecodeFixedI32(buf, &sz);
  *pArray = taosArrayInit(sz, sizeof(void*));
  if (*pArray == NULL) {
    return NULL;
  }
  for (int32_t i = 0; i < sz; i++) {
    void* data = taosMemoryCalloc(1, dataSz);
    if (data == NULL) {
      return NULL;
    }

    buf = decode(buf, data, sver);

    if (taosArrayPush(*pArray, &data) == NULL) {
      taosMemoryFree(data);
      return NULL;
    }
  }
  return (void*)buf;
}

// todo remove it
// order array<type *>
int32_t taosArraySortPWithExt(SArray* pArray, __ext_compar_fn_t fn, const void* param) {
  return taosqsort(pArray->pData, pArray->size, pArray->elemSize, param, fn);
}

void taosArraySwap(SArray* a, SArray* b) {
  if (a == NULL || b == NULL) return;
  size_t t = a->size;
  a->size = b->size;
  b->size = t;

  uint32_t cap = a->capacity;
  a->capacity = b->capacity;
  b->capacity = cap;

  uint32_t elem = a->elemSize;
  a->elemSize = b->elemSize;
  b->elemSize = elem;

  void* data = a->pData;
  a->pData = b->pData;
  b->pData = data;
}

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	#define _DEFAULT_SOURCE
17	#include "tarray.h"
18	#include "tcoding.h"
19
20	// todo refactor API
21	#define BOUNDARY_SIZE 1024 * 1024 * 1024 // 1G
22	#define BOUNDARY_SMALL_FACTOR 1.2
23	#define BOUNDARY_BIG_FACTOR 2
24
25	SArray* taosArrayInit(size_t size, size_t elemSize) {	902,224,157✔
26	if (elemSize == 0) {	902,224,157✔
27	terrno = TSDB_CODE_INVALID_PARA;	1✔
28	return NULL;	1✔
29	}
30
31	if (size < TARRAY_MIN_SIZE) {	902,224,156✔
32	size = TARRAY_MIN_SIZE;	220,313,706✔
33	}
34
35	SArray* pArray = taosMemoryMalloc(sizeof(SArray));	902,224,156!
36	if (pArray == NULL) {	902,056,030!
37	return NULL;	×
38	}
39
40	pArray->size = 0;	902,056,030✔
41	pArray->pData = taosMemoryCalloc(size, elemSize);	902,056,030!
42	if (pArray->pData == NULL) {	904,255,087!
43	taosMemoryFree(pArray);	×
44	return NULL;	×
45	}
46
47	pArray->capacity = size;	904,255,087✔
48	pArray->elemSize = elemSize;	904,255,087✔
49	return pArray;	904,255,087✔
50	}
51
52	SArray* taosArrayInit_s(size_t elemSize, size_t initialSize) {	2,007,573✔
53	SArray* pArray = taosMemoryMalloc(sizeof(SArray));	2,007,573!
54	if (pArray == NULL) {	2,011,731!
55	return NULL;	×
56	}
57
58	pArray->size = initialSize;	2,011,731✔
59	pArray->pData = taosMemoryCalloc(initialSize, elemSize);	2,011,731!
60	if (pArray->pData == NULL) {	2,011,800!
61	taosMemoryFree(pArray);	×
62	return NULL;	×
63	}
64
65	pArray->capacity = initialSize;	2,011,800✔
66	pArray->elemSize = elemSize;	2,011,800✔
67	return pArray;	2,011,800✔
68	}
69
70	static int32_t taosArrayResize(SArray* pArray) {	13✔
71	size_t size = pArray->capacity;	13✔
72	size = (size << 1u);	13✔
73
74	void* tmp = taosMemoryRealloc(pArray->pData, size * pArray->elemSize);	13!
75	if (tmp == NULL) { // reallocate failed, the original buffer remains	14!
76	return terrno;	×
77	}
78
79	pArray->pData = tmp;	14✔
80	pArray->capacity = size;	14✔
81
82	return 0;	14✔
83	}
84
85	int32_t taosArrayEnsureCap(SArray* pArray, size_t newCap) {	2,147,483,647✔
86	if (newCap > pArray->capacity) {	2,147,483,647✔
87	float factor = BOUNDARY_BIG_FACTOR;	37,423,069✔
88	if (newCap * pArray->elemSize > BOUNDARY_SIZE) {	37,423,069!
89	factor = BOUNDARY_SMALL_FACTOR;	×
90	}
91
92	size_t tsize = (pArray->capacity * factor);	37,423,069✔
93	while (newCap > tsize) {	38,080,802✔
94	size_t newSize = (tsize * factor);	657,733✔
95	tsize = (newSize == tsize) ? (tsize + 2) : newSize;	657,733!
96	}
97
98	char* p = taosMemoryRealloc(pArray->pData, tsize * pArray->elemSize);	37,423,069!
99	if (p == NULL) {	37,428,073!
100	return terrno;	×
101	}
102
103	pArray->pData = p;	37,428,073✔
104	pArray->capacity = tsize;	37,428,073✔
105	}
106	return 0;	2,147,483,647✔
107	}
108
109	void* taosArrayAddBatch(SArray* pArray, const void* pData, int32_t nEles) {	2,147,483,647✔
110	if (pData == NULL) {	2,147,483,647✔
111	terrno = TSDB_CODE_INVALID_PARA;	1✔
112	return NULL;	1✔
113	}
114
115	int32_t code = taosArrayEnsureCap(pArray, pArray->size + nEles);	2,147,483,647✔
116	if (code) {	2,147,483,647!
117	terrno = code;	×
118	return NULL;	×
119	}
120
121	void* dst = TARRAY_GET_ELEM(pArray, pArray->size);	2,147,483,647✔
122	memcpy(dst, pData, pArray->elemSize * nEles);	2,147,483,647✔
123
124	pArray->size += nEles;	2,147,483,647✔
125	return dst;	2,147,483,647✔
126	}
127
128	void taosArrayRemoveDuplicate(SArray* pArray, __compar_fn_t comparFn, void (fp)(void)) {	1,081,521✔
129	size_t size = pArray->size;	1,081,521✔
130	if (size <= 1) {	1,081,521✔
131	return;	1,075,758✔
132	}
133
134	int32_t pos = 0;	5,763✔
135	for (int32_t i = 1; i < size; ++i) {	626,783✔
136	char* p1 = taosArrayGet(pArray, pos);	621,020✔
137	char* p2 = taosArrayGet(pArray, i);	621,020✔
138
139	if (comparFn(p1, p2) == 0) {	621,020✔
140	// do nothing
141	} else {
142	if (pos + 1 != i) {	524,512✔
143	void* p = taosArrayGet(pArray, pos + 1);	14,603✔
144	if (fp != NULL) {	14,603✔
145	fp(p);	4✔
146	}
147
148	taosArraySet(pArray, pos + 1, p2);	14,603✔
149	memset(TARRAY_GET_ELEM(pArray, i), 0, pArray->elemSize);	14,603✔
150	pos += 1;	14,603✔
151	} else {
152	pos += 1;	509,909✔
153	}
154	}
155	}
156
157	if (fp != NULL) {	5,763✔
158	for (int32_t i = pos + 1; i < pArray->size; ++i) {	492✔
159	void* p = taosArrayGet(pArray, i);	13✔
160	fp(p);	13✔
161	}
162	}
163
164	pArray->size = pos + 1;	6,193✔
165	}
166
167	void* taosArrayAddAll(SArray* pArray, const SArray* pInput) {	3,425,378✔
168	if (pInput) {	3,425,378✔
169	return taosArrayAddBatch(pArray, pInput->pData, (int32_t)taosArrayGetSize(pInput));	3,420,917✔
170	} else {
171	terrno = TSDB_CODE_INVALID_PARA;	4,461✔
172	return NULL;	4,610✔
173	}
174	}
175
176	void* taosArrayReserve(SArray* pArray, int32_t num) {	1,477,554,650✔
177	int32_t code = taosArrayEnsureCap(pArray, pArray->size + num);	1,477,554,650✔
178	if (code) {	1,476,949,999!
179	terrno = code;	×
180	return NULL;	×
181	}
182
183	void* dst = TARRAY_GET_ELEM(pArray, pArray->size);	1,478,774,516✔
184	pArray->size += num;	1,478,774,516✔
185
186	memset(dst, 0, num * pArray->elemSize);	1,478,774,516✔
187
188	return dst;	1,478,774,516✔
189	}
190
191	void* taosArrayPop(SArray* pArray) {	63,553,119✔
192	if (pArray->size == 0) {	63,553,119✔
193	return NULL;	891,251✔
194	}
195
196	pArray->size -= 1;	62,661,868✔
197	return TARRAY_GET_ELEM(pArray, pArray->size);	62,661,868✔
198	}
199
200	void* taosArrayGet(const SArray* pArray, size_t index) {	2,147,483,647✔
201	if (NULL == pArray) {	2,147,483,647✔
202	terrno = TSDB_CODE_INVALID_PARA;	53✔
203	uError("failed to return value from array of null ptr");	53!
204	return NULL;	53✔
205	}
206
207	if (index >= pArray->size) {	2,147,483,647✔
208	uError("index is out of range, current:%" PRIzu " max:%" PRIzu, index, pArray->size);	1,137!
209	terrno = TSDB_CODE_OUT_OF_RANGE;	1,137✔
210	return NULL;	1,137✔
211	}
212
213	return TARRAY_GET_ELEM(pArray, index);	2,147,483,647✔
214	}
215
216	void* taosArrayGetP(const SArray* pArray, size_t index) {	1,138,956,391✔
217	void** p = taosArrayGet(pArray, index);	1,138,956,391✔
218	if (p == NULL) {	1,140,929,012✔
219	return NULL;	2✔
220	}
221	return *p;	1,140,929,010✔
222	}
223
224	void* taosArrayGetLast(const SArray* pArray) {	85,806,959✔
225	if (pArray->size == 0) {	85,806,959✔
226	terrno = TSDB_CODE_INVALID_PARA;	1✔
227	return NULL;	1✔
228	}
229
230	return TARRAY_GET_ELEM(pArray, pArray->size - 1);	85,806,958✔
231	}
232
233	size_t taosArrayGetSize(const SArray* pArray) {	2,147,483,647✔
234	if (pArray == NULL) {	2,147,483,647✔
235	return 0;	169,310,683✔
236	}
237	return TARRAY_SIZE(pArray);	2,147,483,647✔
238	}
239
240	void* taosArrayInsert(SArray* pArray, size_t index, const void* pData) {	8,938,791✔
241	if (pArray == NULL \|\| pData == NULL) {	8,938,791!
242	terrno = TSDB_CODE_INVALID_PARA;	×
243	return NULL;	1✔
244	}
245
246	if (index >= pArray->size) {	8,938,898✔
247	return taosArrayPush(pArray, pData);	8,938,117✔
248	}
249
250	if (pArray->size >= pArray->capacity) {	760✔
251	int32_t ret = taosArrayResize(pArray);	14✔
252	if (ret < 0) {	14!
UNCOV 253	terrno = ret;	×
254	return NULL;	×
255	}
256	}
257
258	void* dst = TARRAY_GET_ELEM(pArray, index);	777✔
259
260	int32_t remain = (int32_t)(pArray->size - index);	777✔
261	memmove((char)dst + pArray->elemSize, (char)dst, pArray->elemSize * remain);	777✔
262	memcpy(dst, pData, pArray->elemSize);	777✔
263
264	pArray->size += 1;	777✔
265
266	return dst;	777✔
267	}
268
269	void taosArraySet(SArray* pArray, size_t index, void* pData) {	90,005,353✔
270	if (index < pArray->size) {	90,005,353!
271	memcpy(TARRAY_GET_ELEM(pArray, index), pData, pArray->elemSize);	90,009,293✔
272	}
273	}	90,005,353✔
274
275	void taosArrayPopFrontBatch(SArray* pArray, size_t cnt) {	28,496✔
276	if (cnt > pArray->size) {	28,496✔
277	cnt = pArray->size;	1✔
278	}
279
280	pArray->size = pArray->size - cnt;	28,496✔
281	if (pArray->size == 0 \|\| cnt == 0) {	28,496✔
282	return;	28,215✔
283	}
284	memmove(pArray->pData, (char)pArray->pData + cnt pArray->elemSize, pArray->size * pArray->elemSize);	281✔
285	}
286
287	void taosArrayPopTailBatch(SArray* pArray, size_t cnt) {	196,251✔
288	if (cnt >= pArray->size) {	196,251✔
289	cnt = pArray->size;	160,376✔
290	}
291
292	pArray->size = pArray->size - cnt;	196,251✔
293	}	196,251✔
294
295	void taosArrayRemove(SArray* pArray, size_t index) {	11,936,880✔
296	if (!(index < pArray->size)) {	11,936,880!
297	return;	×
298	}
299
300	if (index == pArray->size - 1) {	11,936,880✔
301	void* t = taosArrayPop(pArray);	3,506,293✔
302	return;	3,506,281✔
303	}
304
305	size_t remain = pArray->size - index - 1;	8,430,587✔
306	memmove((char)pArray->pData + index pArray->elemSize, (char)pArray->pData + (index + 1) pArray->elemSize,	8,430,587✔
307	remain * pArray->elemSize);	8,430,587✔
308	pArray->size -= 1;	8,430,587✔
309	}
310
311	void taosArrayRemoveBatch(SArray* pArray, size_t index, size_t num, FDelete fp) {	725,447✔
312	if (index + num <= pArray->size) {	725,447!
313	if (fp) {	725,452✔
314	for (int32_t i = 0; i < num; i++) {	1,447,306✔
315	fp(taosArrayGet(pArray, index + i));	723,979✔
316	}
317	}
318
319	memmove((char)pArray->pData + index pArray->elemSize, (char)pArray->pData + (index + num) pArray->elemSize,	725,457✔
320	(pArray->size - index - num) * pArray->elemSize);	725,457✔
321	pArray->size -= num;	725,457✔
322	}
323	}	725,452✔
324
325	SArray* taosArrayFromList(const void* src, size_t size, size_t elemSize) {	1✔
326	if (elemSize <= 0) {	1!
327	terrno = TSDB_CODE_INVALID_PARA;	×
328	return NULL;	×
329	}
330
331	SArray* pDst = taosArrayInit(size, elemSize);	1✔
332	if (pDst) {	1!
333	memcpy(pDst->pData, src, elemSize * size);	1✔
334	pDst->size = size;	1✔
335	}
336
337	return pDst;	1✔
338	}
339
340	SArray* taosArrayDup(const SArray* pSrc, __array_item_dup_fn_t fn) {	9,968,506✔
341	if (NULL == pSrc) {	9,968,506✔
342	terrno = TSDB_CODE_INVALID_PARA;	82,816✔
343	return NULL;	82,816✔
344	}
345
346	if (pSrc->size == 0) { // empty array list	9,885,690✔
347	return taosArrayInit(8, pSrc->elemSize);	53,075✔
348	}
349
350	SArray* dst = taosArrayInit(pSrc->size, pSrc->elemSize);	9,832,615✔
351
352	if (dst) {	9,835,264!
353	if (fn == NULL) {	9,835,342✔
354	memcpy(dst->pData, pSrc->pData, pSrc->elemSize * pSrc->size);	9,833,985✔
355	} else {
356	for (int32_t i = 0; i < pSrc->size; ++i) {	4,122✔
357	void* p = fn(taosArrayGetP(pSrc, i));	2,334✔
358	memcpy(((char)dst->pData) + i dst->elemSize, &p, dst->elemSize);	2,765✔
359	}
360	}
361
362	dst->size = pSrc->size;	9,835,773✔
363	}
364
365	return dst;	9,835,695✔
366	}
367
368	void taosArrayClear(SArray* pArray) {	392,437,082✔
369	if (pArray == NULL) return;	392,437,082✔
370	pArray->size = 0;	349,217,628✔
371	}
372
373	void taosArrayClearEx(SArray* pArray, void (fp)(void)) {	5,238,368✔
374	if (pArray == NULL) return;	5,238,368!
375	if (fp == NULL) {	5,238,368✔
376	pArray->size = 0;	1✔
377	return;	1✔
378	}
379
380	for (int32_t i = 0; i < pArray->size; ++i) {	136,169,204✔
381	fp(TARRAY_GET_ELEM(pArray, i));	130,930,788✔
382	}
383
384	pArray->size = 0;	5,238,416✔
385	}
386	void taosArrayClearP(SArray* pArray, void (fp)(void)) {	483✔
387	// if (pArray == NULL) return;
388	// if (fp == NULL) {
389	// pArray->size = 0;
390	// return;
391	// }
392
393	// for (int32_t i = 0; i < pArray->size; ++i) {
394	// fp(TARRAY_GET_ELEM(pArray, i));
395	// }
396	if (pArray) {	483!
397	if (NULL == fp) {	483!
398	for (int32_t i = 0; i < pArray->size; i++) {	497✔
399	taosMemoryFree((void*)TARRAY_GET_ELEM(pArray, i));	14!
400	}
401	} else {
402	for (int32_t i = 0; i < pArray->size; i++) {	×
403	fp((void*)TARRAY_GET_ELEM(pArray, i));	×
404	}
405	}
406	}
407	taosArrayClear(pArray);	483✔
408	}	484✔
409
410	void taosArrayDestroy(SArray* pArray) {	2,058,582,373✔
411	if (pArray) {	2,058,582,373✔
412	taosMemoryFree(pArray->pData);	908,094,328!
413	taosMemoryFree(pArray);	909,120,012!
414	}
415	}	2,058,662,715✔
416
417	void taosArrayDestroyP(SArray* pArray, FDelete fp) {	158,211,769✔
418	if (pArray) {	158,211,769✔
419	if (NULL == fp) {	140,244,730✔
420	for (int32_t i = 0; i < pArray->size; i++) {	13,660,419✔
421	taosMemoryFree((void*)TARRAY_GET_ELEM(pArray, i));	5,799,699!
422	}
423	} else {
424	for (int32_t i = 0; i < pArray->size; i++) {	263,754,471✔
425	fp((void*)TARRAY_GET_ELEM(pArray, i));	131,369,625✔
426	}
427	}
428	taosArrayDestroy(pArray);	140,245,566✔
429	}
430	}	158,212,725✔
431
432	void taosArrayDestroyEx(SArray* pArray, FDelete fp) {	183,608,521✔
433	if (pArray == NULL) {	183,608,521✔
434	return;	94,332,953✔
435	}
436
437	if (fp == NULL) {	89,275,568✔
438	taosArrayDestroy(pArray);	569,860✔
439	return;	569,887✔
440	}
441
442	for (int32_t i = 0; i < pArray->size; ++i) {	269,545,717✔
443	fp(TARRAY_GET_ELEM(pArray, i));	180,856,075✔
444	}
445
446	taosArrayDestroy(pArray);	88,689,642✔
447	}
448
449	void taosArraySort(SArray* pArray, __compar_fn_t compar) {	6,419,913✔
450	taosSort(pArray->pData, pArray->size, pArray->elemSize, compar);	6,419,913✔
451	}	6,420,060✔
452
453	int32_t taosArrayMSort(SArray* pArray, __compar_fn_t compar) {	33,663✔
454	return taosMergeSort(pArray->pData, pArray->size, pArray->elemSize, compar);	33,663✔
455	}
456
457	void* taosArraySearch(const SArray* pArray, const void* key, __compar_fn_t comparFn, int32_t flags) {	21,502,553✔
458	return taosbsearch(key, pArray->pData, pArray->size, pArray->elemSize, comparFn, flags);	21,502,553✔
459	}
460
461	int32_t taosArraySearchIdx(const SArray* pArray, const void* key, __compar_fn_t comparFn, int32_t flags) {	9,202,977✔
462	void* item = taosArraySearch(pArray, key, comparFn, flags);	9,202,977✔
463	return item == NULL ? -1 : (int32_t)((char)item - (char)pArray->pData) / pArray->elemSize;	9,203,269✔
464	}
465
466	#ifdef BUILD_NO_CALL
467	static int32_t taosArrayPartition(SArray* pArray, int32_t i, int32_t j, __ext_compar_fn_t fn, const void* userData) {
468	void* key = taosArrayGetP(pArray, i);
469	while (i < j) {
470	while (i < j && fn(taosArrayGetP(pArray, j), key, userData) >= 0) {
471	j--;
472	}
473	if (i < j) {
474	void* a = taosArrayGetP(pArray, j);
475	taosArraySet(pArray, i, &a);
476	}
477	while (i < j && fn(taosArrayGetP(pArray, i), key, userData) <= 0) {
478	i++;
479	}
480	if (i < j) {
481	void* a = taosArrayGetP(pArray, i);
482	taosArraySet(pArray, j, &a);
483	}
484	}
485	taosArraySet(pArray, i, &key);
486	return i;
487	}
488
489	static void taosArrayQuicksortImpl(SArray* pArray, int32_t low, int32_t high, __ext_compar_fn_t fn, const void* param) {
490	if (low < high) {
491	int32_t idx = taosArrayPartition(pArray, low, high, fn, param);
492	taosArrayQuicksortImpl(pArray, low, idx - 1, fn, param);
493	taosArrayQuicksortImpl(pArray, idx + 1, high, fn, param);
494	}
495	}
496
497	static void taosArrayQuickSort(SArray* pArray, __ext_compar_fn_t fn, const void* param) {
498	if (pArray->size <= 1) {
499	return;
500	}
501	taosArrayQuicksortImpl(pArray, 0, (int32_t)(taosArrayGetSize(pArray) - 1), fn, param);
502	}
503
504	static void taosArrayInsertSort(SArray* pArray, __ext_compar_fn_t fn, const void* param) {
505	if (pArray->size <= 1) {
506	return;
507	}
508
509	for (int32_t i = 1; i <= pArray->size - 1; ++i) {
510	for (int32_t j = i; j > 0; --j) {
511	if (fn(taosArrayGetP(pArray, j), taosArrayGetP(pArray, j - 1), param) == -1) {
512	void* a = taosArrayGetP(pArray, j);
513	void* b = taosArrayGetP(pArray, j - 1);
514	taosArraySet(pArray, j - 1, &a);
515	taosArraySet(pArray, j, &b);
516	} else {
517	break;
518	}
519	}
520	}
521	}
522	#endif
523
524	int32_t taosEncodeArray(void** buf, const SArray* pArray, FEncode encode) {	5,066✔
525	int32_t tlen = 0;	5,066✔
526	int32_t sz = pArray->size;	5,066✔
527	tlen += taosEncodeFixedI32(buf, sz);	5,066✔
528	for (int32_t i = 0; i < sz; i++) {	13,316✔
529	void* data = taosArrayGetP(pArray, i);	8,250✔
530	tlen += encode(buf, data);	8,250✔
531	}
532	return tlen;	5,066✔
533	}
534
535	void* taosDecodeArray(const void* buf, SArray** pArray, FDecode decode, int32_t dataSz, int8_t sver) {	1,982!
536	int32_t sz;
537	buf = taosDecodeFixedI32(buf, &sz);	1,982✔
538	pArray = taosArrayInit(sz, sizeof(void));	1,982✔
539	if (*pArray == NULL) {	1,982!
540	return NULL;	×
541	}
542	for (int32_t i = 0; i < sz; i++) {	5,538✔
543	void* data = taosMemoryCalloc(1, dataSz);	3,556!
544	if (data == NULL) {	3,556!
545	return NULL;	×
546	}
547
548	buf = decode(buf, data, sver);	3,556✔
549
550	if (taosArrayPush(*pArray, &data) == NULL) {	7,112!
551	taosMemoryFree(data);	×
552	return NULL;	×
553	}
554	}
555	return (void*)buf;	1,982✔
556	}
557
558	// todo remove it
559	// order array<type *>
560	int32_t taosArraySortPWithExt(SArray* pArray, __ext_compar_fn_t fn, const void* param) {	1✔
561	return taosqsort(pArray->pData, pArray->size, pArray->elemSize, param, fn);	1✔
562	}
563
564	void taosArraySwap(SArray* a, SArray* b) {	38✔
565	if (a == NULL \|\| b == NULL) return;	38!
566	size_t t = a->size;	39✔
567	a->size = b->size;	39✔
568	b->size = t;	39✔
569
570	uint32_t cap = a->capacity;	39✔
571	a->capacity = b->capacity;	39✔
572	b->capacity = cap;	39✔
573
574	uint32_t elem = a->elemSize;	39✔
575	a->elemSize = b->elemSize;	39✔
576	b->elemSize = elem;	39✔
577
578	void* data = a->pData;	39✔
579	a->pData = b->pData;	39✔
580	b->pData = data;	39✔
581	}

taosdata / TDengine / #3660

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