#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

55.32

/source/dnode/mnode/impl/src/mndPerfSchema.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 "mndInt.h"
#include "systable.h"

// connection/application/
int32_t mndInitPerfsTableSchema(const SSysDbTableSchema *pSrc, int32_t colNum, SSchema **pDst) {
  int32_t  code = 0;
  SSchema *schema = taosMemoryCalloc(colNum, sizeof(SSchema));
  if (NULL == schema) {
    code = terrno;
    TAOS_RETURN(code);
  }

  for (int32_t i = 0; i < colNum; ++i) {
    tstrncpy(schema[i].name, pSrc[i].name, sizeof(schema[i].name));

    schema[i].type = pSrc[i].type;
    schema[i].colId = i + 1;
    schema[i].bytes = pSrc[i].bytes;
  }

  *pDst = schema;
  TAOS_RETURN(code);
}

int32_t mndPerfsInitMeta(SHashObj *hash) {
  int32_t       code = 0;
  STableMetaRsp meta = {0};

  tstrncpy(meta.dbFName, TSDB_PERFORMANCE_SCHEMA_DB, sizeof(meta.dbFName));
  meta.tableType = TSDB_SYSTEM_TABLE;
  meta.sversion = 1;
  meta.tversion = 1;
  meta.virtualStb = false;

  size_t               size = 0;
  const SSysTableMeta *pSysDbTableMeta = NULL;
  getPerfDbMeta(&pSysDbTableMeta, &size);

  for (int32_t i = 0; i < size; ++i) {
    tstrncpy(meta.tbName, pSysDbTableMeta[i].name, sizeof(meta.tbName));
    meta.numOfColumns = pSysDbTableMeta[i].colNum;

    TAOS_CHECK_RETURN(mndInitPerfsTableSchema(pSysDbTableMeta[i].schema, pSysDbTableMeta[i].colNum, &meta.pSchemas));

    if (taosHashPut(hash, meta.tbName, strlen(meta.tbName), &meta, sizeof(meta))) {
      code = terrno;
      TAOS_RETURN(code);
    }
  }

  TAOS_RETURN(code);
}

int32_t mndBuildPerfsTableSchema(SMnode *pMnode, const char *dbFName, const char *tbName, STableMetaRsp *pRsp) {
  int32_t code = 0;
  if (NULL == pMnode->perfsMeta) {
    code = TSDB_CODE_APP_ERROR;
    TAOS_RETURN(code);
  }

  STableMetaRsp *meta = (STableMetaRsp *)taosHashGet(pMnode->perfsMeta, tbName, strlen(tbName));
  if (NULL == meta) {
    mError("invalid performance schema table name:%s", tbName);
    code = TSDB_CODE_PAR_TABLE_NOT_EXIST;
    TAOS_RETURN(code);
  }

  *pRsp = *meta;

  pRsp->pSchemas = taosMemoryCalloc(meta->numOfColumns, sizeof(SSchema));
  if (pRsp->pSchemas == NULL) {
    code = terrno;
    pRsp->pSchemas = NULL;
    TAOS_RETURN(code);
  }

  memcpy(pRsp->pSchemas, meta->pSchemas, meta->numOfColumns * sizeof(SSchema));
  TAOS_RETURN(code);
}

int32_t mndBuildPerfsTableCfg(SMnode *pMnode, const char *dbFName, const char *tbName, STableCfgRsp *pRsp) {
  int32_t code = 0;
  if (NULL == pMnode->perfsMeta) {
    code = TSDB_CODE_APP_ERROR;
    TAOS_RETURN(code);
  }

  STableMetaRsp *pMeta = taosHashGet(pMnode->perfsMeta, tbName, strlen(tbName));
  if (NULL == pMeta) {
    mError("invalid performance schema table name:%s", tbName);
    code = TSDB_CODE_PAR_TABLE_NOT_EXIST;
    TAOS_RETURN(code);
  }

  tstrncpy(pRsp->tbName, pMeta->tbName, sizeof(pRsp->tbName));
  tstrncpy(pRsp->stbName, pMeta->stbName, sizeof(pRsp->stbName));
  tstrncpy(pRsp->dbFName, pMeta->dbFName, sizeof(pRsp->dbFName));
  pRsp->numOfTags = pMeta->numOfTags;
  pRsp->numOfColumns = pMeta->numOfColumns;
  pRsp->tableType = pMeta->tableType;
  pRsp->virtualStb = pMeta->virtualStb;

  pRsp->pSchemas = taosMemoryCalloc(pMeta->numOfColumns, sizeof(SSchema));
  if (pRsp->pSchemas == NULL) {
    code = terrno;
    pRsp->pSchemas = NULL;
    TAOS_RETURN(code);
  }

  memcpy(pRsp->pSchemas, pMeta->pSchemas, pMeta->numOfColumns * sizeof(SSchema));
  TAOS_RETURN(code);
}

int32_t mndInitPerfs(SMnode *pMnode) {
  int32_t code = 0;
  pMnode->perfsMeta = taosHashInit(20, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY), false, HASH_NO_LOCK);
  if (pMnode->perfsMeta == NULL) {
    code = terrno;
    TAOS_RETURN(code);
  }

  return mndPerfsInitMeta(pMnode->perfsMeta);
}

void mndCleanupPerfs(SMnode *pMnode) {
  if (NULL == pMnode->perfsMeta) {
    return;
  }

  void *pIter = taosHashIterate(pMnode->perfsMeta, NULL);
  while (pIter) {
    STableMetaRsp *meta = (STableMetaRsp *)pIter;

    taosMemoryFreeClear(meta->pSchemas);

    pIter = taosHashIterate(pMnode->perfsMeta, pIter);
  }

  taosHashCleanup(pMnode->perfsMeta);
  pMnode->perfsMeta = NULL;
}

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 "mndInt.h"
18	#include "systable.h"
19
20	// connection/application/
21	int32_t mndInitPerfsTableSchema(const SSysDbTableSchema pSrc, int32_t colNum, SSchema *pDst) {	8,740✔
22	int32_t code = 0;	8,740✔
23	SSchema *schema = taosMemoryCalloc(colNum, sizeof(SSchema));	8,740!
24	if (NULL == schema) {	8,740!
25	code = terrno;	×
26	TAOS_RETURN(code);	×
27	}
28
29	for (int32_t i = 0; i < colNum; ++i) {	115,368✔
30	tstrncpy(schema[i].name, pSrc[i].name, sizeof(schema[i].name));	106,628✔
31
32	schema[i].type = pSrc[i].type;	106,628✔
33	schema[i].colId = i + 1;	106,628✔
34	schema[i].bytes = pSrc[i].bytes;	106,628✔
35	}
36
37	*pDst = schema;	8,740✔
38	TAOS_RETURN(code);	8,740✔
39	}
40
41	int32_t mndPerfsInitMeta(SHashObj *hash) {	1,748✔
42	int32_t code = 0;	1,748✔
43	STableMetaRsp meta = {0};	1,748✔
44
45	tstrncpy(meta.dbFName, TSDB_PERFORMANCE_SCHEMA_DB, sizeof(meta.dbFName));	1,748✔
46	meta.tableType = TSDB_SYSTEM_TABLE;	1,748✔
47	meta.sversion = 1;	1,748✔
48	meta.tversion = 1;	1,748✔
49	meta.virtualStb = false;	1,748✔
50
51	size_t size = 0;	1,748✔
52	const SSysTableMeta *pSysDbTableMeta = NULL;	1,748✔
53	getPerfDbMeta(&pSysDbTableMeta, &size);	1,748✔
54
55	for (int32_t i = 0; i < size; ++i) {	10,488✔
56	tstrncpy(meta.tbName, pSysDbTableMeta[i].name, sizeof(meta.tbName));	8,740✔
57	meta.numOfColumns = pSysDbTableMeta[i].colNum;	8,740✔
58
59	TAOS_CHECK_RETURN(mndInitPerfsTableSchema(pSysDbTableMeta[i].schema, pSysDbTableMeta[i].colNum, &meta.pSchemas));	8,740!
60
61	if (taosHashPut(hash, meta.tbName, strlen(meta.tbName), &meta, sizeof(meta))) {	8,740!
62	code = terrno;	×
UNCOV 63	TAOS_RETURN(code);	×
64	}
65	}
66
67	TAOS_RETURN(code);	1,748✔
68	}
69
70	int32_t mndBuildPerfsTableSchema(SMnode pMnode, const char dbFName, const char tbName, STableMetaRsp pRsp) {	27,024✔
71	int32_t code = 0;	27,024✔
72	if (NULL == pMnode->perfsMeta) {	27,024!
73	code = TSDB_CODE_APP_ERROR;	×
UNCOV 74	TAOS_RETURN(code);	×
75	}
76
77	STableMetaRsp meta = (STableMetaRsp )taosHashGet(pMnode->perfsMeta, tbName, strlen(tbName));	27,024✔
78	if (NULL == meta) {	27,024✔
79	mError("invalid performance schema table name:%s", tbName);	117!
80	code = TSDB_CODE_PAR_TABLE_NOT_EXIST;	117✔
81	TAOS_RETURN(code);	117✔
82	}
83
84	pRsp = meta;	26,907✔
85
86	pRsp->pSchemas = taosMemoryCalloc(meta->numOfColumns, sizeof(SSchema));	26,907!
87	if (pRsp->pSchemas == NULL) {	26,907!
88	code = terrno;	×
89	pRsp->pSchemas = NULL;	×
UNCOV 90	TAOS_RETURN(code);	×
91	}
92
93	memcpy(pRsp->pSchemas, meta->pSchemas, meta->numOfColumns * sizeof(SSchema));	26,907✔
94	TAOS_RETURN(code);	26,907✔
95	}
96
97	int32_t mndBuildPerfsTableCfg(SMnode pMnode, const char dbFName, const char tbName, STableCfgRsp pRsp) {	×
98	int32_t code = 0;	×
99	if (NULL == pMnode->perfsMeta) {	×
100	code = TSDB_CODE_APP_ERROR;	×
UNCOV 101	TAOS_RETURN(code);	×
102	}
103
104	STableMetaRsp *pMeta = taosHashGet(pMnode->perfsMeta, tbName, strlen(tbName));	×
105	if (NULL == pMeta) {	×
106	mError("invalid performance schema table name:%s", tbName);	×
107	code = TSDB_CODE_PAR_TABLE_NOT_EXIST;	×
UNCOV 108	TAOS_RETURN(code);	×
109	}
110
111	tstrncpy(pRsp->tbName, pMeta->tbName, sizeof(pRsp->tbName));	×
112	tstrncpy(pRsp->stbName, pMeta->stbName, sizeof(pRsp->stbName));	×
113	tstrncpy(pRsp->dbFName, pMeta->dbFName, sizeof(pRsp->dbFName));	×
114	pRsp->numOfTags = pMeta->numOfTags;	×
115	pRsp->numOfColumns = pMeta->numOfColumns;	×
UNCOV 116	pRsp->tableType = pMeta->tableType;	×
117	pRsp->virtualStb = pMeta->virtualStb;	×
118
119	pRsp->pSchemas = taosMemoryCalloc(pMeta->numOfColumns, sizeof(SSchema));	×
120	if (pRsp->pSchemas == NULL) {	×
121	code = terrno;	×
UNCOV 122	pRsp->pSchemas = NULL;	×
UNCOV 123	TAOS_RETURN(code);	×
124	}
125
UNCOV 126	memcpy(pRsp->pSchemas, pMeta->pSchemas, pMeta->numOfColumns * sizeof(SSchema));	×
UNCOV 127	TAOS_RETURN(code);	×
128	}
129
130	int32_t mndInitPerfs(SMnode *pMnode) {	1,748✔
131	int32_t code = 0;	1,748✔
132	pMnode->perfsMeta = taosHashInit(20, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY), false, HASH_NO_LOCK);	1,748✔
133	if (pMnode->perfsMeta == NULL) {	1,748!
UNCOV 134	code = terrno;	×
UNCOV 135	TAOS_RETURN(code);	×
136	}
137
138	return mndPerfsInitMeta(pMnode->perfsMeta);	1,748✔
139	}
140
141	void mndCleanupPerfs(SMnode *pMnode) {	1,747✔
142	if (NULL == pMnode->perfsMeta) {	1,747!
UNCOV 143	return;	×
144	}
145
146	void *pIter = taosHashIterate(pMnode->perfsMeta, NULL);	1,747✔
147	while (pIter) {	10,482✔
148	STableMetaRsp meta = (STableMetaRsp )pIter;	8,735✔
149
150	taosMemoryFreeClear(meta->pSchemas);	8,735!
151
152	pIter = taosHashIterate(pMnode->perfsMeta, pIter);	8,735✔
153	}
154
155	taosHashCleanup(pMnode->perfsMeta);	1,747✔
156	pMnode->perfsMeta = NULL;	1,747✔
157	}

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