#4791

Committed 13 Oct 2025 06:50AM UTC coverage: 57.628% (-0.8%) from 58.476%

Build # #4791

Build Type

push

travis-ci

Committed by

web-flow

Commit Message

Merge pull request #33213 from taosdata/fix/huoh/timemoe_model_directory

fix: fix tdgpt timemoe model directory

Coverage Stats

136628 of 303332 branches covered (45.04%)

Branch coverage included in aggregate %.

208121 of 294900 relevant lines covered (70.57%)

4250784.02 hits per line

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

68.81

/source/libs/executor/src/rollupoperator.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 "executorInt.h"
#include "filter.h"
#include "function.h"
#include "index.h"
#include "nodes.h"
#include "operator.h"
#include "os.h"
#include "query.h"
#include "querynodes.h"
#include "querytask.h"
#include "tcompare.h"
#include "tdatablock.h"
#include "tfill.h"
#include "tglobal.h"
#include "thash.h"
#include "tname.h"
#include "ttypes.h"

static int32_t createDataBlockForSchema(SRSchema *pRSchema, SSDataBlock **ppDataBlock, int32_t maxBufRows) {
  int32_t      code = 0, lino = 0;
  STSchema    *pTSchema = pRSchema->tSchema;
  SExtSchema  *pExtSchema = (SExtSchema *)pRSchema->extSchema;
  int32_t      numOfCols = pTSchema->numOfCols;
  SSDataBlock *pBlock = NULL;

  if (numOfCols < 1) {
    TAOS_CHECK_EXIT(TSDB_CODE_RSMA_INVALID_SCHEMA);
  }
  TAOS_CHECK_EXIT(createDataBlock(&pBlock));

  // the primary timestamp column is included since it's needed by funcs like first/last.
  for (int32_t i = 0; i < numOfCols; ++i) {
    STColumn       *pCol = pTSchema->columns + i;
    SColumnInfoData colInfoData = createColumnInfoData(pCol->type, pCol->bytes, pCol->colId);
    if (pExtSchema && IS_DECIMAL_TYPE(pCol->type)) {
      decimalFromTypeMod(pExtSchema[i].typeMod, &colInfoData.info.precision, &colInfoData.info.scale);
    }
    TAOS_CHECK_EXIT(blockDataAppendColInfo(pBlock, &colInfoData));
  }
  TAOS_CHECK_EXIT(blockDataEnsureCapacity(pBlock, maxBufRows));

_exit:
  if (code != TSDB_CODE_SUCCESS) {
    blockDataDestroy(pBlock);
    pBlock = NULL;
  }
  *ppDataBlock = pBlock;
  return code;
}

static int32_t createDataBlockForTargets(SRSchema *pRSchema, SNodeList *pTargets, SSDataBlock **ppDataBlock,
                                         int32_t maxBufRows) {
  int32_t      code = 0, lino = 0;
  int32_t      numOfCols = LIST_LENGTH(pTargets);
  SExtSchema  *pExtSchema = (SExtSchema *)pRSchema->extSchema;
  SSDataBlock *pBlock = NULL;

  if (numOfCols < 1) {
    TAOS_CHECK_EXIT(TSDB_CODE_APP_ERROR);
  }
  TAOS_CHECK_EXIT(createDataBlock(&pBlock));
  if (pExtSchema) ++pExtSchema;
  for (int32_t i = 0; i < numOfCols; ++i) {  // the first timestamp column is not included in targets
    STargetNode *pTargetNode = (STargetNode *)nodesListGetNode(pTargets, i);
    if (!pTargetNode) {
      TAOS_CHECK_EXIT(TSDB_CODE_APP_ERROR);
    }
    SFunctionNode *pFuncNode = (SFunctionNode *)pTargetNode->pExpr;
    if (!pFuncNode || pFuncNode->node.type != QUERY_NODE_FUNCTION) {
      TAOS_CHECK_EXIT(TSDB_CODE_APP_ERROR);
    }
    if (LIST_LENGTH(pFuncNode->pParameterList) < 1 || LIST_LENGTH(pFuncNode->pParameterList) > 3) { // col[,pk[,composite key]]
      TAOS_CHECK_EXIT(TSDB_CODE_RSMA_INVALID_FUNC_PARAM);
    }
    SColumnNode *pColNode = (SColumnNode *)nodesListGetNode(pFuncNode->pParameterList, 0);
    if (!pColNode || pColNode->node.type != QUERY_NODE_COLUMN) {
      TAOS_CHECK_EXIT(TSDB_CODE_RSMA_INVALID_FUNC_PARAM);
    }

    SColumnInfoData colInfoData =
        createColumnInfoData(pFuncNode->node.resType.type, pFuncNode->node.resType.bytes, pColNode->colId);
    if (pExtSchema && IS_DECIMAL_TYPE(pColNode->node.resType.type)) {
      decimalFromTypeMod(pExtSchema[i].typeMod, &colInfoData.info.precision, &colInfoData.info.scale);
    }
    TAOS_CHECK_EXIT(blockDataAppendColInfo(pBlock, &colInfoData));
  }
  TAOS_CHECK_EXIT(blockDataEnsureCapacity(pBlock, maxBufRows));

_exit:
  if (code != TSDB_CODE_SUCCESS) {
    qError("%s failed at line %d since %s", __func__, lino, tstrerror(code));
    blockDataDestroy(pBlock);
    pBlock = NULL;
  }
  *ppDataBlock = pBlock;
  return code;
}

int32_t tdRollupCtxInit(SRollupCtx *pCtx, SRSchema *pRSchema, int8_t precision, const char *dbName) {
  int32_t         code = 0, lino = 0;
  STSchema       *pTSchema = pRSchema->tSchema;
  SExtSchema     *pExtSchema = pRSchema->extSchema;
  SSDataBlock    *pInputBlock = NULL;
  SSDataBlock    *pResBlock = NULL;
  SExprSupp      *pExprSup = NULL;
  SAggSupporter  *pAggSup = NULL;
  SGroupResInfo  *pGroupResInfo = NULL;
  SResultRowInfo *pResultRowInfo = NULL;
  SExecTaskInfo  *pTaskInfo = NULL;
  SResultRow     *pResultRow = NULL;
  STargetNode    *pTargetNode = NULL;
  SFunctionNode  *pFuncNode = NULL;
  SColumnNode    *pColNode = NULL;
  SColumnNode    *pPrimaryKeyColNode = NULL;
  SColumnNode    *pCompositeKeyColNode = NULL;
  int32_t         nCols = pTSchema->numOfCols;
  int32_t         exprNum = 0;
  bool            hasPk = false;
  int32_t         pkBytes = 0;
  SExprInfo      *pExprInfo = NULL;
  char            buf[512] = "\0";

  if (nCols < 2) {
    TAOS_CHECK_EXIT(TSDB_CODE_APP_ERROR);
  }

  if (!(pExprSup = pCtx->exprSup)) {
    if (!(pExprSup = taosMemoryCalloc(1, sizeof(SExprSupp)))) {
      TAOS_CHECK_EXIT(terrno);
    }
    pCtx->exprSup = pExprSup;
  }

  if (!(pAggSup = pCtx->aggSup)) {
    if (!(pAggSup = taosMemoryCalloc(1, sizeof(SAggSupporter)))) {
      TAOS_CHECK_EXIT(terrno);
    }
    pCtx->aggSup = pAggSup;
  }

  if (!(pResultRowInfo = pCtx->resultRowInfo)) {
    if (!(pResultRowInfo = taosMemoryCalloc(1, sizeof(SResultRowInfo)))) {
      TAOS_CHECK_EXIT(terrno);
    }
    pCtx->resultRowInfo = pResultRowInfo;
  }

  if (!(pGroupResInfo = pCtx->pGroupResInfo)) {
    if (!(pGroupResInfo = taosMemoryCalloc(1, sizeof(SGroupResInfo)))) {
      TAOS_CHECK_EXIT(terrno);
    }
    pCtx->pGroupResInfo = pGroupResInfo;
  }

  if (!(pTaskInfo = pCtx->pTaskInfo)) {
    if (!(pTaskInfo = taosMemoryCalloc(1, sizeof(SExecTaskInfo)))) {
      TAOS_CHECK_EXIT(terrno);
    }
    pCtx->pTaskInfo = pTaskInfo;
    pTaskInfo->id.str = taosStrdup("rollup");
    if (!pTaskInfo->id.str) {
      TAOS_CHECK_EXIT(terrno);
    }
  }

  if (!pCtx->pColValArr && !(pCtx->pColValArr = taosArrayInit(nCols, sizeof(SColVal)))) {
    TAOS_CHECK_EXIT(terrno);
  }

  if (!pCtx->pBuf && !(pCtx->pBuf = taosMemoryMalloc(TSDB_MAX_BYTES_PER_ROW + VARSTR_HEADER_SIZE))) {
    TAOS_CHECK_EXIT(terrno);
  }

  int32_t inputRowSize = sizeof(int64_t);  // for the timestamp column
  if (pTSchema->columns[1].flags & COL_IS_KEY) {
    hasPk = true;
    pkBytes = pTSchema->columns[1].bytes;
  }
  for (int32_t i = 1; i < nCols; i++) {    // skip the first timestamp column
    STColumn *pCol = pTSchema->columns + i;
    TAOS_CHECK_EXIT(nodesMakeNode(QUERY_NODE_COLUMN, (SNode **)&pColNode));
    TAOS_CHECK_EXIT(nodesMakeNode(QUERY_NODE_FUNCTION, (SNode **)&pFuncNode));
    TAOS_CHECK_EXIT(nodesMakeNode(QUERY_NODE_TARGET, (SNode **)&pTargetNode));

    // build the column node
    pColNode->node.resType.type = pCol->type;
    pColNode->node.resType.bytes = pCol->bytes;
    pColNode->node.resType.precision = 0;
    pColNode->node.resType.scale = 0;
    if (pExtSchema && IS_DECIMAL_TYPE(pCol->type)) {
      decimalFromTypeMod(pExtSchema[i].typeMod, &pColNode->node.resType.precision, &pColNode->node.resType.scale);
    }

    pColNode->colId = pCol->colId;
    pColNode->colType = COLUMN_TYPE_COLUMN;

    pColNode->slotId = i;

    pColNode->tableId = pRSchema->tbUid;
    pColNode->tableType = pRSchema->tbType;
    // (void)snprintf(pColNode->dbName, TSDB_DB_NAME_LEN, "%s", dbName);
    // (void)snprintf(pColNode->tableName, TSDB_TABLE_NAME_LEN, "%s", pRSchema->tbName);
    // (void)snprintf(pColNode->tableAlias, TSDB_TABLE_NAME_LEN, "%s", pRSchema->tbName);
    // snprintf(pColNode->colName, TSDB_COL_NAME_LEN, "c%d", i); // TODO: fill if necessary

    // build the function node
    // pFuncNode->node.resType = pColNode->node.resType;
    // pFuncNode->funcId = pRSchema->funcIds[i];
    // pFuncNode->funcType = fmGetFuncTypeById(pFuncNode->funcId);
    (void)snprintf(pFuncNode->functionName, TSDB_FUNC_NAME_LEN, "%s", fmGetFuncName(pRSchema->funcIds[i]));
    TAOS_CHECK_EXIT(nodesListMakeAppend(&pFuncNode->pParameterList, (SNode *)pColNode));
    pColNode = NULL;
    TAOS_CHECK_EXIT(fmGetFuncInfo(pFuncNode, buf, sizeof(buf)));
    pFuncNode->hasPk = hasPk;
    pFuncNode->pkBytes = pkBytes;

    if (fmIsImplicitTsFunc(pFuncNode->funcId)) {
      TAOS_CHECK_EXIT(nodesMakeNode(QUERY_NODE_COLUMN, (SNode **)&pPrimaryKeyColNode));
      pPrimaryKeyColNode->node.resType.type = TSDB_DATA_TYPE_TIMESTAMP;
      pPrimaryKeyColNode->node.resType.bytes = sizeof(int64_t);
      pPrimaryKeyColNode->colId = PRIMARYKEY_TIMESTAMP_COL_ID;
      pPrimaryKeyColNode->colType = COLUMN_TYPE_COLUMN;
      pPrimaryKeyColNode->slotId = 0;  // put the timestamp column at the 1st position
      // (void)snprintf(pPrimaryKeyColNode->tableName, TSDB_TABLE_NAME_LEN, "%s", pRSchema->tbName);
      (void)snprintf(pFuncNode->functionName, TSDB_FUNC_NAME_LEN, "%s", pFuncNode->functionName);
      TAOS_CHECK_EXIT(nodesListMakeAppend(&pFuncNode->pParameterList, (SNode *)pPrimaryKeyColNode));
      pPrimaryKeyColNode = NULL;
      if (hasPk) {
        TAOS_CHECK_EXIT(nodesMakeNode(QUERY_NODE_COLUMN, (SNode **)&pCompositeKeyColNode));
        pCompositeKeyColNode->node.resType.type = pTSchema->columns[1].type;
        pCompositeKeyColNode->node.resType.bytes = pTSchema->columns[1].bytes;
        pCompositeKeyColNode->colId = pTSchema->columns[1].colId;
        pCompositeKeyColNode->colType = COLUMN_TYPE_COLUMN;
        pCompositeKeyColNode->slotId = 1;
        TAOS_CHECK_EXIT(nodesListMakeAppend(&pFuncNode->pParameterList, (SNode *)pCompositeKeyColNode));
        pCompositeKeyColNode = NULL;
      }
    }

    // build the target node
    pTargetNode->slotId = (int16_t)(i - 1);
    pTargetNode->dataBlockId = 0;  // only one data block for rollup
    pTargetNode->pExpr = (SNode *)pFuncNode;
    pFuncNode = NULL;

    TAOS_CHECK_EXIT(nodesListMakeAppend((SNodeList **)&pCtx->pTargets, (SNode *)pTargetNode));
    pTargetNode = NULL;

    inputRowSize += pCol->bytes;
  }

  pCtx->rowSize = inputRowSize;
  pCtx->maxBufRows = (64LL * 1024 * 1024) / inputRowSize;
  if (pCtx->maxBufRows > 4096) {
    pCtx->maxBufRows = 4096;
  } else if (pCtx->maxBufRows < 1024) {
    pCtx->maxBufRows = 1024;
  }

  TAOS_CHECK_EXIT(createDataBlockForSchema(pRSchema, &pCtx->pInputBlock, pCtx->maxBufRows));
  TAOS_CHECK_EXIT(createDataBlockForTargets(pRSchema, pCtx->pTargets, &pCtx->pResBlock, pCtx->maxBufRows));

  initResultRowInfo(pResultRowInfo);

  TAOS_CHECK_EXIT(createExprInfo((SNodeList *)pCtx->pTargets, NULL, &pExprInfo, &exprNum));
  size_t keyBufSize = sizeof(int64_t) + sizeof(int64_t) + POINTER_BYTES;
  TAOS_CHECK_EXIT(initAggSup(pExprSup, pAggSup, pExprInfo, exprNum, keyBufSize, "rollup", NULL, NULL));

  int32_t groupId = 0;
  pResultRow = doSetResultOutBufByKey(pAggSup->pResultBuf, pResultRowInfo, (char *)&groupId, sizeof(groupId), true,
                                      groupId, pTaskInfo, false, pAggSup, true);
  if (pResultRow == NULL || pTaskInfo->code != 0) {
    TAOS_CHECK_EXIT(pTaskInfo->code);
  }
  pCtx->resultRow = pResultRow;
  int32_t rowSize = pAggSup->resultRowSize;
  if (pResultRow->pageId == -1) {
    TAOS_CHECK_EXIT(addNewResultRowBuf(pResultRow, pAggSup->pResultBuf, rowSize));
  }

  TAOS_CHECK_EXIT(setResultRowInitCtx(pResultRow, pExprSup->pCtx, exprNum, pExprSup->rowEntryInfoOffset));

  TAOS_CHECK_EXIT(initGroupedResultInfo(pCtx->pGroupResInfo, pCtx->aggSup->pResultRowHashTable, 0));

_exit:
  nodesDestroyNode((SNode *)pCompositeKeyColNode);
  nodesDestroyNode((SNode *)pPrimaryKeyColNode);
  nodesDestroyNode((SNode *)pColNode);
  nodesDestroyNode((SNode *)pFuncNode);
  nodesDestroyNode((SNode *)pTargetNode);
  if (code != 0) {
    if (buf[0] != 0) {
      qError("%s failed at line %d since %s(%s)", __func__, lino, tstrerror(code), buf);
    } else {
      qError("%s failed at line %d since %s", __func__, lino, tstrerror(code));
    }
  }

  return code;
}

int32_t tdRollupDoAggregate(SRollupCtx *pCtx) {
  int32_t         code = 0, lino = 0;
  SqlFunctionCtx *pFuncCtx = pCtx->exprSup->pCtx;
  SExprSupp      *pExprSup = pCtx->exprSup;

  if ((pExprSup->numOfExprs > 0 && pFuncCtx == NULL)) {
    TAOS_CHECK_EXIT(TSDB_CODE_QRY_EXECUTOR_INTERNAL_ERROR);
  }

  TAOS_CHECK_EXIT(
      setInputDataBlock(pCtx->exprSup, pCtx->pInputBlock, ORDER_ASC, pCtx->pInputBlock->info.scanFlag, true));

  for (int32_t k = 0; k < pExprSup->numOfExprs; ++k) {
    if (functionNeedToExecute(&pFuncCtx[k])) {
      // todo add a dummy function to avoid process check
      if (pFuncCtx[k].fpSet.process == NULL) {
        continue;
      }

      if ((&pFuncCtx[k])->input.pData[0] == NULL) {
        code = TSDB_CODE_QRY_EXECUTOR_INTERNAL_ERROR;
      } else {
        code = pFuncCtx[k].fpSet.process(&pFuncCtx[k]);
      }

      if (code != TSDB_CODE_SUCCESS) {
        if (pFuncCtx[k].fpSet.cleanup != NULL) {
          pFuncCtx[k].fpSet.cleanup(&pFuncCtx[k]);
        }
        TAOS_CHECK_EXIT(code);
      }
    }
  }
_exit:
  if (code != 0) {
    qError("%s failed at line %d since %s", __func__, lino, tstrerror(code));
  }
  return code;
}

int32_t tdRollupFinalize(SRollupCtx *pCtx) {
  int32_t      code = 0, lino = 0;
  SSDataBlock *pResBlock = pCtx->pResBlock;

  blockDataCleanup(pResBlock);
  // pResBlock->info.id.groupId = 0;
  doCopyToSDataBlock(pCtx->pTaskInfo, pResBlock, pCtx->exprSup, pCtx->aggSup->pResultBuf, pCtx->pGroupResInfo,
                     pCtx->maxBufRows, true, 0);
_exit:
  return code;
}

void tdRollupCtxCleanup(SRollupCtx *pCtx, bool deep) {
  if (pCtx) {
    if (pCtx->pTargets) {
      nodesDestroyList((SNodeList *)pCtx->pTargets);
      pCtx->pTargets = NULL;
    }
    if (pCtx->pBuf && deep) {
      taosMemFreeClear(pCtx->pBuf);
    }
    if (pCtx->exprSup) {
      cleanupExprSupp(pCtx->exprSup);
      if (deep) taosMemFreeClear(pCtx->exprSup);
    }
    if (pCtx->aggSup) {
      cleanupAggSup(pCtx->aggSup);
      if (deep) taosMemFreeClear(pCtx->aggSup);
    }
    if (pCtx->resultRowInfo) {
      pCtx->resultRowInfo->openWindow = tdListFree(pCtx->resultRowInfo->openWindow);
      if (deep) taosMemFreeClear(pCtx->resultRowInfo);
    }
    if (pCtx->pGroupResInfo) {
      cleanupGroupResInfo(pCtx->pGroupResInfo);
      if (deep) taosMemFreeClear(pCtx->pGroupResInfo);
    }
    if (pCtx->pTaskInfo) {
      if (deep) {
        taosMemFreeClear(pCtx->pTaskInfo->id.str);
        taosMemFreeClear(pCtx->pTaskInfo);
      }
    }
    if (pCtx->pColValArr) {
      if (deep) {
        taosArrayDestroy(pCtx->pColValArr);
        pCtx->pColValArr = NULL;
      } else {
        taosArrayClear(pCtx->pColValArr);
      }
    }
    blockDataDestroy(pCtx->pInputBlock);
    blockDataDestroy(pCtx->pResBlock);
  }
}

int32_t tdRollupCtxReset(SRollupCtx *pCtx) {
  int32_t code = 0, lino = 0;
  pCtx->winTotalRows = 0;
  pCtx->pGroupResInfo->index = 0;
  pCtx->pGroupResInfo->delIndex = 0;
  clearResultRowInitFlag(pCtx->exprSup->pCtx, pCtx->exprSup->numOfExprs);
  TAOS_CHECK_EXIT(setResultRowInitCtx(pCtx->resultRow, pCtx->exprSup->pCtx, pCtx->exprSup->numOfExprs,
                                      pCtx->exprSup->rowEntryInfoOffset));
_exit:
  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 "executorInt.h"
17	#include "filter.h"
18	#include "function.h"
19	#include "index.h"
20	#include "nodes.h"
21	#include "operator.h"
22	#include "os.h"
23	#include "query.h"
24	#include "querynodes.h"
25	#include "querytask.h"
26	#include "tcompare.h"
27	#include "tdatablock.h"
28	#include "tfill.h"
29	#include "tglobal.h"
30	#include "thash.h"
31	#include "tname.h"
32	#include "ttypes.h"
33
34	static int32_t createDataBlockForSchema(SRSchema pRSchema, SSDataBlock *ppDataBlock, int32_t maxBufRows) {	34✔
35	int32_t code = 0, lino = 0;	34✔
36	STSchema *pTSchema = pRSchema->tSchema;	34✔
37	SExtSchema pExtSchema = (SExtSchema )pRSchema->extSchema;	34✔
38	int32_t numOfCols = pTSchema->numOfCols;	34✔
39	SSDataBlock *pBlock = NULL;	34✔
40
41	if (numOfCols < 1) {	34!
42	TAOS_CHECK_EXIT(TSDB_CODE_RSMA_INVALID_SCHEMA);	×
43	}
44	TAOS_CHECK_EXIT(createDataBlock(&pBlock));	34!
45
46	// the primary timestamp column is included since it's needed by funcs like first/last.
47	for (int32_t i = 0; i < numOfCols; ++i) {	280✔
48	STColumn *pCol = pTSchema->columns + i;	246✔
49	SColumnInfoData colInfoData = createColumnInfoData(pCol->type, pCol->bytes, pCol->colId);	246✔
50	if (pExtSchema && IS_DECIMAL_TYPE(pCol->type)) {	246✔
51	decimalFromTypeMod(pExtSchema[i].typeMod, &colInfoData.info.precision, &colInfoData.info.scale);	19✔
52	}
53	TAOS_CHECK_EXIT(blockDataAppendColInfo(pBlock, &colInfoData));	246!
54	}
55	TAOS_CHECK_EXIT(blockDataEnsureCapacity(pBlock, maxBufRows));	34!
56
57	_exit:	34✔
58	if (code != TSDB_CODE_SUCCESS) {	34!
59	blockDataDestroy(pBlock);	×
60	pBlock = NULL;	×
61	}
62	*ppDataBlock = pBlock;	34✔
63	return code;	34✔
64	}
65
66	static int32_t createDataBlockForTargets(SRSchema pRSchema, SNodeList pTargets, SSDataBlock **ppDataBlock,	34✔
67	int32_t maxBufRows) {
68	int32_t code = 0, lino = 0;	34✔
69	int32_t numOfCols = LIST_LENGTH(pTargets);	34!
70	SExtSchema pExtSchema = (SExtSchema )pRSchema->extSchema;	34✔
71	SSDataBlock *pBlock = NULL;	34✔
72
73	if (numOfCols < 1) {	34!
74	TAOS_CHECK_EXIT(TSDB_CODE_APP_ERROR);	×
75	}
76	TAOS_CHECK_EXIT(createDataBlock(&pBlock));	34!
77	if (pExtSchema) ++pExtSchema;	34✔
78	for (int32_t i = 0; i < numOfCols; ++i) { // the first timestamp column is not included in targets	246✔
79	STargetNode pTargetNode = (STargetNode )nodesListGetNode(pTargets, i);	212✔
80	if (!pTargetNode) {	212!
81	TAOS_CHECK_EXIT(TSDB_CODE_APP_ERROR);	×
82	}
83	SFunctionNode pFuncNode = (SFunctionNode )pTargetNode->pExpr;	212✔
84	if (!pFuncNode \|\| pFuncNode->node.type != QUERY_NODE_FUNCTION) {	212!
85	TAOS_CHECK_EXIT(TSDB_CODE_APP_ERROR);	×
86	}
87	if (LIST_LENGTH(pFuncNode->pParameterList) < 1 \|\| LIST_LENGTH(pFuncNode->pParameterList) > 3) { // col[,pk[,composite key]]	212!
88	TAOS_CHECK_EXIT(TSDB_CODE_RSMA_INVALID_FUNC_PARAM);	×
89	}
90	SColumnNode pColNode = (SColumnNode )nodesListGetNode(pFuncNode->pParameterList, 0);	212✔
91	if (!pColNode \|\| pColNode->node.type != QUERY_NODE_COLUMN) {	212!
92	TAOS_CHECK_EXIT(TSDB_CODE_RSMA_INVALID_FUNC_PARAM);	×
93	}
94
95	SColumnInfoData colInfoData =
96	createColumnInfoData(pFuncNode->node.resType.type, pFuncNode->node.resType.bytes, pColNode->colId);	212✔
97	if (pExtSchema && IS_DECIMAL_TYPE(pColNode->node.resType.type)) {	212✔
98	decimalFromTypeMod(pExtSchema[i].typeMod, &colInfoData.info.precision, &colInfoData.info.scale);	19✔
99	}
100	TAOS_CHECK_EXIT(blockDataAppendColInfo(pBlock, &colInfoData));	212!
101	}
102	TAOS_CHECK_EXIT(blockDataEnsureCapacity(pBlock, maxBufRows));	34!
103
104	_exit:	34✔
105	if (code != TSDB_CODE_SUCCESS) {	34!
106	qError("%s failed at line %d since %s", __func__, lino, tstrerror(code));	×
107	blockDataDestroy(pBlock);	×
108	pBlock = NULL;	×
109	}
110	*ppDataBlock = pBlock;	34✔
111	return code;	34✔
112	}
113
114	int32_t tdRollupCtxInit(SRollupCtx pCtx, SRSchema pRSchema, int8_t precision, const char *dbName) {	34✔
115	int32_t code = 0, lino = 0;	34✔
116	STSchema *pTSchema = pRSchema->tSchema;	34✔
117	SExtSchema *pExtSchema = pRSchema->extSchema;	34✔
118	SSDataBlock *pInputBlock = NULL;	34✔
119	SSDataBlock *pResBlock = NULL;	34✔
120	SExprSupp *pExprSup = NULL;	34✔
121	SAggSupporter *pAggSup = NULL;	34✔
122	SGroupResInfo *pGroupResInfo = NULL;	34✔
123	SResultRowInfo *pResultRowInfo = NULL;	34✔
124	SExecTaskInfo *pTaskInfo = NULL;	34✔
125	SResultRow *pResultRow = NULL;	34✔
126	STargetNode *pTargetNode = NULL;	34✔
127	SFunctionNode *pFuncNode = NULL;	34✔
128	SColumnNode *pColNode = NULL;	34✔
129	SColumnNode *pPrimaryKeyColNode = NULL;	34✔
130	SColumnNode *pCompositeKeyColNode = NULL;	34✔
131	int32_t nCols = pTSchema->numOfCols;	34✔
132	int32_t exprNum = 0;	34✔
133	bool hasPk = false;	34✔
134	int32_t pkBytes = 0;	34✔
135	SExprInfo *pExprInfo = NULL;	34✔
136	char buf[512] = "\0";	34✔
137
138	if (nCols < 2) {	34!
139	TAOS_CHECK_EXIT(TSDB_CODE_APP_ERROR);	×
140	}
141
142	if (!(pExprSup = pCtx->exprSup)) {	34✔
143	if (!(pExprSup = taosMemoryCalloc(1, sizeof(SExprSupp)))) {	21!
144	TAOS_CHECK_EXIT(terrno);	×
145	}
146	pCtx->exprSup = pExprSup;	21✔
147	}
148
149	if (!(pAggSup = pCtx->aggSup)) {	34✔
150	if (!(pAggSup = taosMemoryCalloc(1, sizeof(SAggSupporter)))) {	21!
151	TAOS_CHECK_EXIT(terrno);	×
152	}
153	pCtx->aggSup = pAggSup;	21✔
154	}
155
156	if (!(pResultRowInfo = pCtx->resultRowInfo)) {	34✔
157	if (!(pResultRowInfo = taosMemoryCalloc(1, sizeof(SResultRowInfo)))) {	21!
158	TAOS_CHECK_EXIT(terrno);	×
159	}
160	pCtx->resultRowInfo = pResultRowInfo;	21✔
161	}
162
163	if (!(pGroupResInfo = pCtx->pGroupResInfo)) {	34✔
164	if (!(pGroupResInfo = taosMemoryCalloc(1, sizeof(SGroupResInfo)))) {	21!
165	TAOS_CHECK_EXIT(terrno);	×
166	}
167	pCtx->pGroupResInfo = pGroupResInfo;	21✔
168	}
169
170	if (!(pTaskInfo = pCtx->pTaskInfo)) {	34✔
171	if (!(pTaskInfo = taosMemoryCalloc(1, sizeof(SExecTaskInfo)))) {	21!
172	TAOS_CHECK_EXIT(terrno);	×
173	}
174	pCtx->pTaskInfo = pTaskInfo;	21✔
175	pTaskInfo->id.str = taosStrdup("rollup");	21!
176	if (!pTaskInfo->id.str) {	21!
177	TAOS_CHECK_EXIT(terrno);	×
178	}
179	}
180
181	if (!pCtx->pColValArr && !(pCtx->pColValArr = taosArrayInit(nCols, sizeof(SColVal)))) {	34!
182	TAOS_CHECK_EXIT(terrno);	×
183	}
184
185	if (!pCtx->pBuf && !(pCtx->pBuf = taosMemoryMalloc(TSDB_MAX_BYTES_PER_ROW + VARSTR_HEADER_SIZE))) {	34!
186	TAOS_CHECK_EXIT(terrno);	×
187	}
188
189	int32_t inputRowSize = sizeof(int64_t); // for the timestamp column	34✔
190	if (pTSchema->columns[1].flags & COL_IS_KEY) {	34✔
191	hasPk = true;	12✔
192	pkBytes = pTSchema->columns[1].bytes;	12✔
193	}
194	for (int32_t i = 1; i < nCols; i++) { // skip the first timestamp column	246✔
195	STColumn *pCol = pTSchema->columns + i;	212✔
196	TAOS_CHECK_EXIT(nodesMakeNode(QUERY_NODE_COLUMN, (SNode **)&pColNode));	212!
197	TAOS_CHECK_EXIT(nodesMakeNode(QUERY_NODE_FUNCTION, (SNode **)&pFuncNode));	212!
198	TAOS_CHECK_EXIT(nodesMakeNode(QUERY_NODE_TARGET, (SNode **)&pTargetNode));	212!
199
200	// build the column node
201	pColNode->node.resType.type = pCol->type;	212✔
202	pColNode->node.resType.bytes = pCol->bytes;	212✔
203	pColNode->node.resType.precision = 0;	212✔
204	pColNode->node.resType.scale = 0;	212✔
205	if (pExtSchema && IS_DECIMAL_TYPE(pCol->type)) {	212✔
206	decimalFromTypeMod(pExtSchema[i].typeMod, &pColNode->node.resType.precision, &pColNode->node.resType.scale);	19✔
207	}
208
209	pColNode->colId = pCol->colId;	212✔
210	pColNode->colType = COLUMN_TYPE_COLUMN;	212✔
211
212	pColNode->slotId = i;	212✔
213
214	pColNode->tableId = pRSchema->tbUid;	212✔
215	pColNode->tableType = pRSchema->tbType;	212✔
216	// (void)snprintf(pColNode->dbName, TSDB_DB_NAME_LEN, "%s", dbName);
217	// (void)snprintf(pColNode->tableName, TSDB_TABLE_NAME_LEN, "%s", pRSchema->tbName);
218	// (void)snprintf(pColNode->tableAlias, TSDB_TABLE_NAME_LEN, "%s", pRSchema->tbName);
219	// snprintf(pColNode->colName, TSDB_COL_NAME_LEN, "c%d", i); // TODO: fill if necessary
220
221	// build the function node
222	// pFuncNode->node.resType = pColNode->node.resType;
223	// pFuncNode->funcId = pRSchema->funcIds[i];
224	// pFuncNode->funcType = fmGetFuncTypeById(pFuncNode->funcId);
225	(void)snprintf(pFuncNode->functionName, TSDB_FUNC_NAME_LEN, "%s", fmGetFuncName(pRSchema->funcIds[i]));	212✔
226	TAOS_CHECK_EXIT(nodesListMakeAppend(&pFuncNode->pParameterList, (SNode *)pColNode));	212!
227	pColNode = NULL;	212✔
228	TAOS_CHECK_EXIT(fmGetFuncInfo(pFuncNode, buf, sizeof(buf)));	212!
229	pFuncNode->hasPk = hasPk;	212✔
230	pFuncNode->pkBytes = pkBytes;	212✔
231
232	if (fmIsImplicitTsFunc(pFuncNode->funcId)) {	212✔
233	TAOS_CHECK_EXIT(nodesMakeNode(QUERY_NODE_COLUMN, (SNode **)&pPrimaryKeyColNode));	98!
234	pPrimaryKeyColNode->node.resType.type = TSDB_DATA_TYPE_TIMESTAMP;	98✔
235	pPrimaryKeyColNode->node.resType.bytes = sizeof(int64_t);	98✔
236	pPrimaryKeyColNode->colId = PRIMARYKEY_TIMESTAMP_COL_ID;	98✔
237	pPrimaryKeyColNode->colType = COLUMN_TYPE_COLUMN;	98✔
238	pPrimaryKeyColNode->slotId = 0; // put the timestamp column at the 1st position	98✔
239	// (void)snprintf(pPrimaryKeyColNode->tableName, TSDB_TABLE_NAME_LEN, "%s", pRSchema->tbName);
240	(void)snprintf(pFuncNode->functionName, TSDB_FUNC_NAME_LEN, "%s", pFuncNode->functionName);	98✔
241	TAOS_CHECK_EXIT(nodesListMakeAppend(&pFuncNode->pParameterList, (SNode *)pPrimaryKeyColNode));	98!
242	pPrimaryKeyColNode = NULL;	98✔
243	if (hasPk) {	98✔
244	TAOS_CHECK_EXIT(nodesMakeNode(QUERY_NODE_COLUMN, (SNode **)&pCompositeKeyColNode));	24!
245	pCompositeKeyColNode->node.resType.type = pTSchema->columns[1].type;	24✔
246	pCompositeKeyColNode->node.resType.bytes = pTSchema->columns[1].bytes;	24✔
247	pCompositeKeyColNode->colId = pTSchema->columns[1].colId;	24✔
248	pCompositeKeyColNode->colType = COLUMN_TYPE_COLUMN;	24✔
249	pCompositeKeyColNode->slotId = 1;	24✔
250	TAOS_CHECK_EXIT(nodesListMakeAppend(&pFuncNode->pParameterList, (SNode *)pCompositeKeyColNode));	24!
251	pCompositeKeyColNode = NULL;	24✔
252	}
253	}
254
255	// build the target node
256	pTargetNode->slotId = (int16_t)(i - 1);	212✔
257	pTargetNode->dataBlockId = 0; // only one data block for rollup	212✔
258	pTargetNode->pExpr = (SNode *)pFuncNode;	212✔
259	pFuncNode = NULL;	212✔
260
261	TAOS_CHECK_EXIT(nodesListMakeAppend((SNodeList *)&pCtx->pTargets, (SNode )pTargetNode));	212!
262	pTargetNode = NULL;	212✔
263
264	inputRowSize += pCol->bytes;	212✔
265	}
266
267	pCtx->rowSize = inputRowSize;	34✔
268	pCtx->maxBufRows = (64LL * 1024 * 1024) / inputRowSize;	34✔
269	if (pCtx->maxBufRows > 4096) {	34!
270	pCtx->maxBufRows = 4096;	34✔
271	} else if (pCtx->maxBufRows < 1024) {	×
272	pCtx->maxBufRows = 1024;	×
273	}
274
275	TAOS_CHECK_EXIT(createDataBlockForSchema(pRSchema, &pCtx->pInputBlock, pCtx->maxBufRows));	34!
276	TAOS_CHECK_EXIT(createDataBlockForTargets(pRSchema, pCtx->pTargets, &pCtx->pResBlock, pCtx->maxBufRows));	34!
277
278	initResultRowInfo(pResultRowInfo);	34✔
279
280	TAOS_CHECK_EXIT(createExprInfo((SNodeList *)pCtx->pTargets, NULL, &pExprInfo, &exprNum));	34!
281	size_t keyBufSize = sizeof(int64_t) + sizeof(int64_t) + POINTER_BYTES;	34✔
282	TAOS_CHECK_EXIT(initAggSup(pExprSup, pAggSup, pExprInfo, exprNum, keyBufSize, "rollup", NULL, NULL));	34!
283
284	int32_t groupId = 0;	34✔
285	pResultRow = doSetResultOutBufByKey(pAggSup->pResultBuf, pResultRowInfo, (char *)&groupId, sizeof(groupId), true,	34✔
286	groupId, pTaskInfo, false, pAggSup, true);
287	if (pResultRow == NULL \|\| pTaskInfo->code != 0) {	34!
288	TAOS_CHECK_EXIT(pTaskInfo->code);	×
289	}
290	pCtx->resultRow = pResultRow;	34✔
291	int32_t rowSize = pAggSup->resultRowSize;	34✔
292	if (pResultRow->pageId == -1) {	34!
293	TAOS_CHECK_EXIT(addNewResultRowBuf(pResultRow, pAggSup->pResultBuf, rowSize));	×
294	}
295
296	TAOS_CHECK_EXIT(setResultRowInitCtx(pResultRow, pExprSup->pCtx, exprNum, pExprSup->rowEntryInfoOffset));	34!
297
298	TAOS_CHECK_EXIT(initGroupedResultInfo(pCtx->pGroupResInfo, pCtx->aggSup->pResultRowHashTable, 0));	34!
299
300	_exit:	34✔
301	nodesDestroyNode((SNode *)pCompositeKeyColNode);	34✔
302	nodesDestroyNode((SNode *)pPrimaryKeyColNode);	34✔
303	nodesDestroyNode((SNode *)pColNode);	34✔
304	nodesDestroyNode((SNode *)pFuncNode);	34✔
305	nodesDestroyNode((SNode *)pTargetNode);	34✔
306	if (code != 0) {	34!
307	if (buf[0] != 0) {	×
308	qError("%s failed at line %d since %s(%s)", __func__, lino, tstrerror(code), buf);	×
309	} else {
310	qError("%s failed at line %d since %s", __func__, lino, tstrerror(code));	×
311	}
312	}
313
314	return code;	34✔
315	}
316
317	int32_t tdRollupDoAggregate(SRollupCtx *pCtx) {	66✔
318	int32_t code = 0, lino = 0;	66✔
319	SqlFunctionCtx *pFuncCtx = pCtx->exprSup->pCtx;	66✔
320	SExprSupp *pExprSup = pCtx->exprSup;	66✔
321
322	if ((pExprSup->numOfExprs > 0 && pFuncCtx == NULL)) {	66!
323	TAOS_CHECK_EXIT(TSDB_CODE_QRY_EXECUTOR_INTERNAL_ERROR);	×
324	}
325
326	TAOS_CHECK_EXIT(	66!
327	setInputDataBlock(pCtx->exprSup, pCtx->pInputBlock, ORDER_ASC, pCtx->pInputBlock->info.scanFlag, true));
328
329	for (int32_t k = 0; k < pExprSup->numOfExprs; ++k) {	496✔
330	if (functionNeedToExecute(&pFuncCtx[k])) {	430!
331	// todo add a dummy function to avoid process check
332	if (pFuncCtx[k].fpSet.process == NULL) {	430!
333	continue;	×
334	}
335
336	if ((&pFuncCtx[k])->input.pData[0] == NULL) {	430!
337	code = TSDB_CODE_QRY_EXECUTOR_INTERNAL_ERROR;	×
338	} else {
339	code = pFuncCtx[k].fpSet.process(&pFuncCtx[k]);	430✔
340	}
341
342	if (code != TSDB_CODE_SUCCESS) {	430!
343	if (pFuncCtx[k].fpSet.cleanup != NULL) {	×
344	pFuncCtx[k].fpSet.cleanup(&pFuncCtx[k]);	×
345	}
346	TAOS_CHECK_EXIT(code);	×
347	}
348	}
349	}
350	_exit:	66✔
351	if (code != 0) {	66!
352	qError("%s failed at line %d since %s", __func__, lino, tstrerror(code));	×
353	}
354	return code;	66✔
355	}
356
357	int32_t tdRollupFinalize(SRollupCtx *pCtx) {	66✔
358	int32_t code = 0, lino = 0;	66✔
359	SSDataBlock *pResBlock = pCtx->pResBlock;	66✔
360
361	blockDataCleanup(pResBlock);	66✔
362	// pResBlock->info.id.groupId = 0;
363	doCopyToSDataBlock(pCtx->pTaskInfo, pResBlock, pCtx->exprSup, pCtx->aggSup->pResultBuf, pCtx->pGroupResInfo,	66✔
364	pCtx->maxBufRows, true, 0);
365	_exit:	66✔
366	return code;	66✔
367	}
368
369	void tdRollupCtxCleanup(SRollupCtx *pCtx, bool deep) {	55✔
370	if (pCtx) {	55!
371	if (pCtx->pTargets) {	55✔
372	nodesDestroyList((SNodeList *)pCtx->pTargets);	34✔
373	pCtx->pTargets = NULL;	34✔
374	}
375	if (pCtx->pBuf && deep) {	55✔
376	taosMemFreeClear(pCtx->pBuf);	21!
377	}
378	if (pCtx->exprSup) {	55✔
379	cleanupExprSupp(pCtx->exprSup);	34✔
380	if (deep) taosMemFreeClear(pCtx->exprSup);	34!
381	}
382	if (pCtx->aggSup) {	55✔
383	cleanupAggSup(pCtx->aggSup);	34✔
384	if (deep) taosMemFreeClear(pCtx->aggSup);	34!
385	}
386	if (pCtx->resultRowInfo) {	55✔
387	pCtx->resultRowInfo->openWindow = tdListFree(pCtx->resultRowInfo->openWindow);	34✔
388	if (deep) taosMemFreeClear(pCtx->resultRowInfo);	34!
389	}
390	if (pCtx->pGroupResInfo) {	55✔
391	cleanupGroupResInfo(pCtx->pGroupResInfo);	34✔
392	if (deep) taosMemFreeClear(pCtx->pGroupResInfo);	34!
393	}
394	if (pCtx->pTaskInfo) {	55✔
395	if (deep) {	34✔
396	taosMemFreeClear(pCtx->pTaskInfo->id.str);	21!
397	taosMemFreeClear(pCtx->pTaskInfo);	21!
398	}
399	}
400	if (pCtx->pColValArr) {	55✔
401	if (deep) {	34✔
402	taosArrayDestroy(pCtx->pColValArr);	21✔
403	pCtx->pColValArr = NULL;	21✔
404	} else {
405	taosArrayClear(pCtx->pColValArr);	13✔
406	}
407	}
408	blockDataDestroy(pCtx->pInputBlock);	55✔
409	blockDataDestroy(pCtx->pResBlock);	55✔
410	}
411	}	55✔
412
413	int32_t tdRollupCtxReset(SRollupCtx *pCtx) {	66✔
414	int32_t code = 0, lino = 0;	66✔
415	pCtx->winTotalRows = 0;	66✔
416	pCtx->pGroupResInfo->index = 0;	66✔
417	pCtx->pGroupResInfo->delIndex = 0;	66✔
418	clearResultRowInitFlag(pCtx->exprSup->pCtx, pCtx->exprSup->numOfExprs);	66✔
419	TAOS_CHECK_EXIT(setResultRowInitCtx(pCtx->resultRow, pCtx->exprSup->pCtx, pCtx->exprSup->numOfExprs,	66!
420	pCtx->exprSup->rowEntryInfoOffset));
421	_exit:	66✔
422	return code;	66✔
423	}

taosdata / TDengine / #4791

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