#5011

Committed 03 Apr 2026 03:59PM UTC coverage: 72.3% (+0.008%) from 72.292%

Build # #5011

Build Type

push

travis-ci

Committed by

web-flow

Commit Message

merge: from main to 3.0 branch #35067

Coverage Stats

4053 of 5985 new or added lines in 68 files covered. (67.72%)

732 existing lines in 143 files now uncovered.

257430 of 356056 relevant lines covered (72.3%)

131834103.52 hits per line

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

82.46

/source/libs/executor/src/cachescanoperator.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 "function.h"
#include "os.h"
#include "tname.h"

#include "tdatablock.h"
#include "tmsg.h"

#include "executorInt.h"
#include "functionMgt.h"
#include "operator.h"
#include "querytask.h"
#include "tcompare.h"
#include "thash.h"
#include "ttypes.h"

#include "storageapi.h"

typedef struct SCacheRowsScanInfo {
  SSDataBlock*    pRes;
  SReadHandle     readHandle;
  void*           pLastrowReader;
  SColMatchInfo   matchInfo;
  int32_t*        pSlotIds;
  int32_t*        pDstSlotIds;
  SExprSupp       pseudoExprSup;
  int32_t         retrieveType;
  int32_t         currentGroupIndex;
  SSDataBlock*    pBufferedRes;
  SArray*         pUidList;
  SArray*         pCidList;
  STableListInfo* pTableList;
  SArray*         pFuncTypeList;
  int32_t         numOfPks;
  SColumnInfo     pkCol;
  bool            gotAll;
} SCacheRowsScanInfo;

static int32_t doScanCacheNext(SOperatorInfo* pOperator, SSDataBlock** ppRes);
static void    destroyCacheScanOperator(void* param);
static int32_t extractCacheScanSlotId(const SArray* pColMatchInfo, SExecTaskInfo* pTaskInfo, int32_t** pSlotIds,
                                      int32_t** pDstSlotIds);

#define SCAN_ROW_TYPE(_t) ((_t) ? CACHESCAN_RETRIEVE_LAST : CACHESCAN_RETRIEVE_LAST_ROW)

static int32_t setColIdForCacheReadBlock(SSDataBlock* pBlock, SLastRowScanPhysiNode* pScan) {
  int32_t code = TSDB_CODE_SUCCESS;
  int32_t lino = 0;
  SNode*  pNode;
  int32_t idx = 0;
  FOREACH(pNode, pScan->pTargets) {
    if (nodeType(pNode) == QUERY_NODE_COLUMN) {
      SColumnNode*     pCol = (SColumnNode*)pNode;
      SColumnInfoData* pColInfo = taosArrayGet(pBlock->pDataBlock, idx);
      QUERY_CHECK_NULL(pColInfo, code, lino, _end, terrno);
      pColInfo->info.colId = pCol->colId;
    }
    idx++;
  }

  for (; idx < pBlock->pDataBlock->size; ++idx) {
    SColumnInfoData* pColInfo = taosArrayGet(pBlock->pDataBlock, idx);
    QUERY_CHECK_NULL(pColInfo, code, lino, _end, terrno);
    if (pScan->scan.pScanPseudoCols) {
      FOREACH(pNode, pScan->scan.pScanPseudoCols) {
        STargetNode* pTarget = (STargetNode*)pNode;
        if (pColInfo->info.slotId == pTarget->slotId) {
          pColInfo->info.colId = 0;
          break;
        }
      }
    }
  }

_end:
  if (code != TSDB_CODE_SUCCESS) {
    qError("%s failed at line %d since %s", __func__, lino, tstrerror(code));
  }
  return code;
}

int32_t createCacherowsScanOperator(SLastRowScanPhysiNode* pScanNode, SReadHandle* readHandle,
                                    STableListInfo* pTableListInfo, SExecTaskInfo* pTaskInfo,
                                    SOperatorInfo** pOptrInfo) {
  QRY_PARAM_CHECK(pOptrInfo);

  int32_t             code = TSDB_CODE_SUCCESS;
  int32_t             lino = 0;
  int32_t             numOfCols = 0;
  SNodeList*          pScanCols = pScanNode->scan.pScanCols;
  SCacheRowsScanInfo* pInfo = taosMemoryCalloc(1, sizeof(SCacheRowsScanInfo));
  SOperatorInfo*      pOperator = taosMemoryCalloc(1, sizeof(SOperatorInfo));

  if (pInfo == NULL || pOperator == NULL) {
    code = terrno;
    goto _error;
  }
  initOperatorCostInfo(pOperator);

  pInfo->pTableList = pTableListInfo;
  pInfo->readHandle = *readHandle;

  SDataBlockDescNode* pDescNode = pScanNode->scan.node.pOutputDataBlockDesc;
  pInfo->pRes = createDataBlockFromDescNode(pDescNode);
  QUERY_CHECK_NULL(pInfo->pRes, code, lino, _error, terrno);

  code = extractColMatchInfo(pScanCols, pDescNode, &numOfCols, COL_MATCH_FROM_COL_ID, &pInfo->matchInfo);
  QUERY_CHECK_CODE(code, lino, _error);

  // todo: the pk information should comes from the physical plan
  // pk info may not in pScanCols, so extract primary key from pInfo->matchInfo may failed
  SSchemaInfo* pSchemaInfo = taosArrayGet(pTaskInfo->schemaInfos, 0);
  if (pSchemaInfo != NULL) {
    if (pSchemaInfo->sw->pSchema[1].flags & COL_IS_KEY) {  // is primary key
      SSchema* pColSchema = &pSchemaInfo->sw->pSchema[1];
      pInfo->numOfPks = 1;
      pInfo->pkCol.type = pColSchema->type;
      pInfo->pkCol.bytes = pColSchema->bytes;
      pInfo->pkCol.pk = 1;
    }
  } else {
    for (int32_t i = 0; i < taosArrayGetSize(pInfo->matchInfo.pList); ++i) {
      SColMatchItem* pItem = taosArrayGet(pInfo->matchInfo.pList, i);
      QUERY_CHECK_NULL(pItem, code, lino, _error, terrno);
      if (pItem->isPk) {
        pInfo->numOfPks += 1;
        pInfo->pkCol.type = pItem->dataType.type;    // only record one primary key
        pInfo->pkCol.bytes = pItem->dataType.bytes;  // only record one primary key
        pInfo->pkCol.pk = 1;
      }
    }
  }

  SArray* pCidList = taosArrayInit(numOfCols, sizeof(int16_t));
  QUERY_CHECK_NULL(pCidList, code, lino, _error, terrno);

  pInfo->pFuncTypeList = taosArrayInit(taosArrayGetSize(pScanNode->pFuncTypes), sizeof(int32_t));
  QUERY_CHECK_NULL(pInfo->pFuncTypeList, code, lino, _error, terrno);

  void* tmp = taosArrayAddAll(pInfo->pFuncTypeList, pScanNode->pFuncTypes);
  if (!tmp && taosArrayGetSize(pScanNode->pFuncTypes) > 0) {
    QUERY_CHECK_NULL(tmp, code, lino, _error, terrno);
  }

  for (int i = 0; i < TARRAY_SIZE(pInfo->matchInfo.pList); ++i) {
    SColMatchItem* pColInfo = taosArrayGet(pInfo->matchInfo.pList, i);
    QUERY_CHECK_NULL(pColInfo, code, lino, _error, terrno);

    void*          tmp = taosArrayPush(pCidList, &pColInfo->colId);
    QUERY_CHECK_NULL(tmp, code, lino, _error, terrno);
    if (pInfo->pFuncTypeList != NULL && taosArrayGetSize(pInfo->pFuncTypeList) > i) {
      void* pFuncType = taosArrayGet(pInfo->pFuncTypeList, i);
      QUERY_CHECK_NULL(pFuncType, code, lino, _error, terrno);
      pColInfo->funcType = *(int32_t*)pFuncType;
    }
  }
  pInfo->pCidList = pCidList;

  code = extractCacheScanSlotId(pInfo->matchInfo.pList, pTaskInfo, &pInfo->pSlotIds, &pInfo->pDstSlotIds);
  QUERY_CHECK_CODE(code, lino, _error);

  int32_t totalTables = 0;
  code = tableListGetSize(pTableListInfo, &totalTables);
  QUERY_CHECK_CODE(code, lino, _error);

  int32_t capacity = 0;

  pInfo->pUidList = taosArrayInit(4, sizeof(int64_t));
  QUERY_CHECK_NULL(pInfo->pUidList, code, lino, _error, terrno);

  // partition by tbname
  if (oneTableForEachGroup(pTableListInfo) || (totalTables == 1)) {
    pInfo->retrieveType = CACHESCAN_RETRIEVE_TYPE_ALL | SCAN_ROW_TYPE(pScanNode->ignoreNull);

    STableKeyInfo* pList = tableListGetInfo(pTableListInfo, 0);
    if (totalTables) QUERY_CHECK_NULL(pList, code, lino, _error, terrno);

    uint64_t suid = tableListGetSuid(pTableListInfo);
    code = pInfo->readHandle.api.cacheFn.openReader(pInfo->readHandle.vnode, pInfo->retrieveType, pList, totalTables,
                                                    taosArrayGetSize(pInfo->matchInfo.pList), pCidList, pInfo->pSlotIds,
                                                    suid, &pInfo->pLastrowReader, pTaskInfo->id.str,
                                                    pScanNode->pFuncTypes, &pInfo->pkCol, pInfo->numOfPks);
    QUERY_CHECK_CODE(code, lino, _error);

    capacity = TMIN(totalTables, 4096);

    pInfo->pBufferedRes = NULL;
    code = createOneDataBlock(pInfo->pRes, false, &pInfo->pBufferedRes);
    QUERY_CHECK_CODE(code, lino, _error);

    code = setColIdForCacheReadBlock(pInfo->pBufferedRes, pScanNode);
    QUERY_CHECK_CODE(code, lino, _error);

    code = blockDataEnsureCapacity(pInfo->pBufferedRes, capacity);
    QUERY_CHECK_CODE(code, lino, _error);
  } else {  // by tags
    pInfo->retrieveType = CACHESCAN_RETRIEVE_TYPE_SINGLE | SCAN_ROW_TYPE(pScanNode->ignoreNull);
    capacity = 1;  // only one row output
    code = setColIdForCacheReadBlock(pInfo->pRes, pScanNode);
    QUERY_CHECK_CODE(code, lino, _error);
  }

  initResultSizeInfo(&pOperator->resultInfo, capacity);
  code = blockDataEnsureCapacity(pInfo->pRes, pOperator->resultInfo.capacity);
  QUERY_CHECK_CODE(code, lino, _error);

  if (pScanNode->scan.pScanPseudoCols != NULL) {
    SExprSupp* p = &pInfo->pseudoExprSup;
    code = createExprInfo(pScanNode->scan.pScanPseudoCols, NULL, &p->pExprInfo, &p->numOfExprs);
    TSDB_CHECK_CODE(code, lino, _error);

    p->pCtx =
        createSqlFunctionCtx(p->pExprInfo, p->numOfExprs, &p->rowEntryInfoOffset, &pTaskInfo->storageAPI.functionStore);
    QUERY_CHECK_NULL(p->pCtx, code, lino, _error, terrno);
  }

  setOperatorInfo(pOperator, "CachedRowScanOperator", QUERY_NODE_PHYSICAL_PLAN_LAST_ROW_SCAN, false, OP_NOT_OPENED,
                  pInfo, pTaskInfo);
  pOperator->exprSupp.numOfExprs = taosArrayGetSize(pInfo->pRes->pDataBlock);

  pOperator->fpSet = createOperatorFpSet(optrDummyOpenFn, doScanCacheNext, NULL, destroyCacheScanOperator, optrDefaultBufFn,
                                         NULL, optrDefaultGetNextExtFn, NULL);

  *pOptrInfo = pOperator;
  return code;

_error:
  pTaskInfo->code = code;
  if (code != TSDB_CODE_SUCCESS) {
    qError("%s failed at line %d since %s", __func__, lino, tstrerror(code));
  }
  if (pInfo != NULL) {
    pInfo->pTableList = NULL;
    destroyCacheScanOperator(pInfo);
  }
  if (pOperator != NULL) {
    pOperator->info = NULL;
    destroyOperator(pOperator);
  }
  return code;
}

static int32_t doScanCacheNext(SOperatorInfo* pOperator, SSDataBlock** ppRes) {
  int32_t code = TSDB_CODE_SUCCESS;
  int32_t lino = 0;
  if (pOperator->status == OP_EXEC_DONE) {
    (*ppRes) = NULL;
    return code;
  }

  SCacheRowsScanInfo* pInfo = pOperator->info;
  SExecTaskInfo*      pTaskInfo = pOperator->pTaskInfo;
  STableListInfo*     pTableList = pInfo->pTableList;
  SStoreCacheReader*  pReaderFn = &pInfo->readHandle.api.cacheFn;
  SSDataBlock*        pBufRes = pInfo->pBufferedRes;

  uint64_t suid = tableListGetSuid(pTableList);
  int32_t  size = 0;
  code = tableListGetSize(pTableList, &size);
  QUERY_CHECK_CODE(code, lino, _end);

  if (size == 0) {
    setOperatorCompleted(pOperator);
    (*ppRes) = NULL;
    return code;
  }

  blockDataCleanup(pInfo->pRes);

  // check if it is a group by tbname
  if ((pInfo->retrieveType & CACHESCAN_RETRIEVE_TYPE_ALL) == CACHESCAN_RETRIEVE_TYPE_ALL) {
    if (isTaskKilled(pTaskInfo)) {
      T_LONG_JMP(pTaskInfo->env, pTaskInfo->code);
    }

    if (!pInfo->gotAll) {
      blockDataCleanup(pBufRes);
      taosArrayClear(pInfo->pUidList);

      code = pReaderFn->retrieveRows(pInfo->pLastrowReader, pBufRes, pInfo->pSlotIds, pInfo->pDstSlotIds,
                                     pInfo->pUidList, &pInfo->gotAll);
      QUERY_CHECK_CODE(code, lino, _end);
    }

    SSDataBlock* pRes = pInfo->pRes;

    if (pBufRes->info.rows > 0) {
      code = copyDataBlock(pRes, pBufRes);
      QUERY_CHECK_CODE(code, lino, _end);

      void* pUid = taosArrayGet(pInfo->pUidList, 0);
      QUERY_CHECK_NULL(pUid, code, lino, _end, terrno);

      pRes->info.id.uid = *(tb_uid_t*)pUid;
      pRes->info.rows = pBufRes->info.rows;
      pRes->info.scanFlag = MAIN_SCAN;

      SExprSupp* pSup = &pInfo->pseudoExprSup;
      code = addTagPseudoColumnData(&pInfo->readHandle, pSup->pExprInfo, pSup->numOfExprs, pRes, pRes->info.rows,
                                    pTaskInfo, NULL);
      QUERY_CHECK_CODE(code, lino, _end);

      code = tableListGetTableGroupId(pTableList, pRes->info.id.uid, &pRes->info.id.groupId, &pRes->info.id.baseGId);
      QUERY_CHECK_CODE(code, lino, _end);
      (*ppRes) = pRes;
      return code;
    } else {
      setOperatorCompleted(pOperator);
      (*ppRes) = NULL;
      return code;
    }
  } else {
    size_t totalGroups = tableListGetOutputGroups(pTableList);

    while (pInfo->currentGroupIndex < totalGroups) {
      if (isTaskKilled(pTaskInfo)) {
        T_LONG_JMP(pTaskInfo->env, pTaskInfo->code);
      }

      STableKeyInfo* pList = NULL;
      int32_t        num = 0;

      code = tableListGetGroupList(pTableList, pInfo->currentGroupIndex, &pList, &num);
      QUERY_CHECK_CODE(code, lino, _end);

      if (NULL == pInfo->pLastrowReader) {
        int32_t tmpRes = pReaderFn->openReader(pInfo->readHandle.vnode, pInfo->retrieveType, pList, num,
                                               taosArrayGetSize(pInfo->matchInfo.pList), pInfo->pCidList,
                                               pInfo->pSlotIds, suid, &pInfo->pLastrowReader, pTaskInfo->id.str,
                                               pInfo->pFuncTypeList, &pInfo->pkCol, pInfo->numOfPks);

        if (tmpRes != TSDB_CODE_SUCCESS) {
          pInfo->currentGroupIndex += 1;
          taosArrayClear(pInfo->pUidList);
          continue;
        }
      } else {
        code = pReaderFn->reuseReader(pInfo->pLastrowReader, pList, num);
        QUERY_CHECK_CODE(code, lino, _end);
      }

      taosArrayClear(pInfo->pUidList);

      code = pReaderFn->retrieveRows(pInfo->pLastrowReader, pInfo->pRes, pInfo->pSlotIds, pInfo->pDstSlotIds,
                                     pInfo->pUidList, NULL);
      QUERY_CHECK_CODE(code, lino, _end);

      pInfo->currentGroupIndex += 1;

      // check for tag values
      if (pInfo->pRes->info.rows > 0) {
        SExprSupp* pSup = &pInfo->pseudoExprSup;

        STableKeyInfo* pKeyInfo = &((STableKeyInfo*)pList)[0];
        pInfo->pRes->info.id.groupId = pKeyInfo->groupId;

        if (taosArrayGetSize(pInfo->pUidList) > 0) {
          void* pUid = taosArrayGet(pInfo->pUidList, 0);
          QUERY_CHECK_NULL(pUid, code, lino, _end, terrno);
          pInfo->pRes->info.id.uid = *(tb_uid_t*)pUid;
          if (pInfo->pseudoExprSup.numOfExprs > 0) {
            code = addTagPseudoColumnData(&pInfo->readHandle, pSup->pExprInfo, pSup->numOfExprs, pInfo->pRes,
                                          pInfo->pRes->info.rows, pTaskInfo, NULL);
            QUERY_CHECK_CODE(code, lino, _end);
          }
        }

        // pInfo->pLastrowReader = tsdbCacherowsReaderClose(pInfo->pLastrowReader);
        (*ppRes) = pInfo->pRes;
        return code;
      } else {
        // pInfo->pLastrowReader = tsdbCacherowsReaderClose(pInfo->pLastrowReader);
      }
    }

    pReaderFn->closeReader(pInfo->pLastrowReader);
    pInfo->pLastrowReader = NULL;
    setOperatorCompleted(pOperator);
    (*ppRes) = NULL;
    return code;
  }

_end:
  if (code != TSDB_CODE_SUCCESS) {
    qError("%s failed at line %d since %s", __func__, lino, tstrerror(code));
    pTaskInfo->code = code;
    T_LONG_JMP(pTaskInfo->env, code);
  }
  return code;
}

void destroyCacheScanOperator(void* param) {
  SCacheRowsScanInfo* pInfo = (SCacheRowsScanInfo*)param;
  blockDataDestroy(pInfo->pRes);
  blockDataDestroy(pInfo->pBufferedRes);
  taosMemoryFreeClear(pInfo->pSlotIds);
  taosMemoryFreeClear(pInfo->pDstSlotIds);
  taosArrayDestroy(pInfo->pCidList);
  taosArrayDestroy(pInfo->pFuncTypeList);
  taosArrayDestroy(pInfo->pUidList);
  taosArrayDestroy(pInfo->matchInfo.pList);
  tableListDestroy(pInfo->pTableList);

  if (pInfo->pLastrowReader != NULL && pInfo->readHandle.api.cacheFn.closeReader != NULL) {
    pInfo->readHandle.api.cacheFn.closeReader(pInfo->pLastrowReader);
    pInfo->pLastrowReader = NULL;
  }

  cleanupExprSupp(&pInfo->pseudoExprSup);
  taosMemoryFreeClear(param);
}

int32_t extractCacheScanSlotId(const SArray* pColMatchInfo, SExecTaskInfo* pTaskInfo, int32_t** pSlotIds,
                               int32_t** pDstSlotIds) {
  size_t numOfCols = taosArrayGetSize(pColMatchInfo);

  *pSlotIds = taosMemoryMalloc(numOfCols * sizeof(int32_t));
  if (*pSlotIds == NULL) {
    return terrno;
  }

  *pDstSlotIds = taosMemoryMalloc(numOfCols * sizeof(int32_t));
  if (*pDstSlotIds == NULL) {
    taosMemoryFreeClear(*pSlotIds);
    return terrno;
  }

  SSchemaInfo*    pSchemaInfo = taosArrayGetLast(pTaskInfo->schemaInfos);
  SSchemaWrapper* pWrapper = pSchemaInfo->sw;

  for (int32_t i = 0; i < numOfCols; ++i) {
    SColMatchItem* pColMatch = taosArrayGet(pColMatchInfo, i);
    if (!pColMatch) {
      return terrno;
    }
    bool           found = false;
    for (int32_t j = 0; j < pWrapper->nCols; ++j) {
      /*      if (pColMatch->colId == pWrapper->pSchema[j].colId && pColMatch->colId == PRIMARYKEY_TIMESTAMP_COL_ID) {
        (*pSlotIds)[pColMatch->dstSlotId] = -1;
        break;
        }*/

      if (pColMatch->colId == pWrapper->pSchema[j].colId) {
        (*pSlotIds)[i] = j;
        (*pDstSlotIds)[i] = pColMatch->dstSlotId;
        found = true;
        break;
      }
    }
    if (!found) {
      (*pSlotIds)[i] = -1;
      (*pDstSlotIds)[i] = pColMatch->dstSlotId;
    }
  }

  return TSDB_CODE_SUCCESS;
}

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 "function.h"
17	#include "os.h"
18	#include "tname.h"
19
20	#include "tdatablock.h"
21	#include "tmsg.h"
22
23	#include "executorInt.h"
24	#include "functionMgt.h"
25	#include "operator.h"
26	#include "querytask.h"
27	#include "tcompare.h"
28	#include "thash.h"
29	#include "ttypes.h"
30
31	#include "storageapi.h"
32
33	typedef struct SCacheRowsScanInfo {
34	SSDataBlock* pRes;
35	SReadHandle readHandle;
36	void* pLastrowReader;
37	SColMatchInfo matchInfo;
38	int32_t* pSlotIds;
39	int32_t* pDstSlotIds;
40	SExprSupp pseudoExprSup;
41	int32_t retrieveType;
42	int32_t currentGroupIndex;
43	SSDataBlock* pBufferedRes;
44	SArray* pUidList;
45	SArray* pCidList;
46	STableListInfo* pTableList;
47	SArray* pFuncTypeList;
48	int32_t numOfPks;
49	SColumnInfo pkCol;
50	bool gotAll;
51	} SCacheRowsScanInfo;
52
53	static int32_t doScanCacheNext(SOperatorInfo* pOperator, SSDataBlock** ppRes);
54	static void destroyCacheScanOperator(void* param);
55	static int32_t extractCacheScanSlotId(const SArray* pColMatchInfo, SExecTaskInfo* pTaskInfo, int32_t** pSlotIds,
56	int32_t** pDstSlotIds);
57
58	#define SCAN_ROW_TYPE(_t) ((_t) ? CACHESCAN_RETRIEVE_LAST : CACHESCAN_RETRIEVE_LAST_ROW)
59
60	static int32_t setColIdForCacheReadBlock(SSDataBlock* pBlock, SLastRowScanPhysiNode* pScan) {	1,829,819✔
61	int32_t code = TSDB_CODE_SUCCESS;	1,829,819✔
62	int32_t lino = 0;	1,829,819✔
63	SNode* pNode;
64	int32_t idx = 0;	1,829,819✔
65	FOREACH(pNode, pScan->pTargets) {	6,346,029✔
66	if (nodeType(pNode) == QUERY_NODE_COLUMN) {	4,516,314✔
67	SColumnNode* pCol = (SColumnNode*)pNode;	4,516,314✔
68	SColumnInfoData* pColInfo = taosArrayGet(pBlock->pDataBlock, idx);	4,516,314✔
69	QUERY_CHECK_NULL(pColInfo, code, lino, _end, terrno);	4,516,314✔
70	pColInfo->info.colId = pCol->colId;	4,516,314✔
71	}
72	idx++;	4,516,210✔
73	}
74
75	for (; idx < pBlock->pDataBlock->size; ++idx) {	1,838,575✔
76	SColumnInfoData* pColInfo = taosArrayGet(pBlock->pDataBlock, idx);	9,504✔
77	QUERY_CHECK_NULL(pColInfo, code, lino, _end, terrno);	9,504✔
78	if (pScan->scan.pScanPseudoCols) {	9,504✔
79	FOREACH(pNode, pScan->scan.pScanPseudoCols) {	6,048✔
80	STargetNode* pTarget = (STargetNode*)pNode;	3,456✔
81	if (pColInfo->info.slotId == pTarget->slotId) {	3,456✔
82	pColInfo->info.colId = 0;	×
83	break;	×
84	}
85	}
86	}
87	}
88
89	_end:	1,830,463✔
90	if (code != TSDB_CODE_SUCCESS) {	1,830,463✔
91	qError("%s failed at line %d since %s", __func__, lino, tstrerror(code));	×
92	}
93	return code;	1,829,819✔
94	}
95
96	int32_t createCacherowsScanOperator(SLastRowScanPhysiNode* pScanNode, SReadHandle* readHandle,	1,829,377✔
97	STableListInfo* pTableListInfo, SExecTaskInfo* pTaskInfo,
98	SOperatorInfo** pOptrInfo) {
99	QRY_PARAM_CHECK(pOptrInfo);	1,829,377✔
100
101	int32_t code = TSDB_CODE_SUCCESS;	1,829,819✔
102	int32_t lino = 0;	1,829,819✔
103	int32_t numOfCols = 0;	1,829,819✔
104	SNodeList* pScanCols = pScanNode->scan.pScanCols;	1,828,291✔
105	SCacheRowsScanInfo* pInfo = taosMemoryCalloc(1, sizeof(SCacheRowsScanInfo));	1,828,733✔
106	SOperatorInfo* pOperator = taosMemoryCalloc(1, sizeof(SOperatorInfo));	1,828,733✔
107
108	if (pInfo == NULL \|\| pOperator == NULL) {	1,829,819✔
UNCOV 109	code = terrno;	×
110	goto _error;	×
111	}
112	initOperatorCostInfo(pOperator);	1,829,819✔
113
114	pInfo->pTableList = pTableListInfo;	1,829,377✔
115	pInfo->readHandle = *readHandle;	1,829,377✔
116
117	SDataBlockDescNode* pDescNode = pScanNode->scan.node.pOutputDataBlockDesc;	1,830,463✔
118	pInfo->pRes = createDataBlockFromDescNode(pDescNode);	1,830,021✔
119	QUERY_CHECK_NULL(pInfo->pRes, code, lino, _error, terrno);	1,829,377✔
120
121	code = extractColMatchInfo(pScanCols, pDescNode, &numOfCols, COL_MATCH_FROM_COL_ID, &pInfo->matchInfo);	1,828,733✔
122	QUERY_CHECK_CODE(code, lino, _error);	1,829,377✔
123
124	// todo: the pk information should comes from the physical plan
125	// pk info may not in pScanCols, so extract primary key from pInfo->matchInfo may failed
126	SSchemaInfo* pSchemaInfo = taosArrayGet(pTaskInfo->schemaInfos, 0);	1,829,377✔
127	if (pSchemaInfo != NULL) {	1,829,377✔
128	if (pSchemaInfo->sw->pSchema[1].flags & COL_IS_KEY) { // is primary key	1,829,377✔
129	SSchema* pColSchema = &pSchemaInfo->sw->pSchema[1];	1,142,350✔
130	pInfo->numOfPks = 1;	1,142,350✔
131	pInfo->pkCol.type = pColSchema->type;	1,142,350✔
132	pInfo->pkCol.bytes = pColSchema->bytes;	1,142,792✔
133	pInfo->pkCol.pk = 1;	1,142,350✔
134	}
135	} else {
136	for (int32_t i = 0; i < taosArrayGetSize(pInfo->matchInfo.pList); ++i) {	×
137	SColMatchItem* pItem = taosArrayGet(pInfo->matchInfo.pList, i);	×
138	QUERY_CHECK_NULL(pItem, code, lino, _error, terrno);	×
139	if (pItem->isPk) {	×
140	pInfo->numOfPks += 1;	×
141	pInfo->pkCol.type = pItem->dataType.type; // only record one primary key	×
142	pInfo->pkCol.bytes = pItem->dataType.bytes; // only record one primary key	×
143	pInfo->pkCol.pk = 1;	×
144	}
145	}
146	}
147
148	SArray* pCidList = taosArrayInit(numOfCols, sizeof(int16_t));	1,829,273✔
149	QUERY_CHECK_NULL(pCidList, code, lino, _error, terrno);	1,828,733✔
150
151	pInfo->pFuncTypeList = taosArrayInit(taosArrayGetSize(pScanNode->pFuncTypes), sizeof(int32_t));	1,828,733✔
152	QUERY_CHECK_NULL(pInfo->pFuncTypeList, code, lino, _error, terrno);	1,829,715✔
153
154	void* tmp = taosArrayAddAll(pInfo->pFuncTypeList, pScanNode->pFuncTypes);	1,828,629✔
155	if (!tmp && taosArrayGetSize(pScanNode->pFuncTypes) > 0) {	1,830,463✔
156	QUERY_CHECK_NULL(tmp, code, lino, _error, terrno);	×
157	}
158
159	for (int i = 0; i < TARRAY_SIZE(pInfo->matchInfo.pList); ++i) {	6,091,932✔
160	SColMatchItem* pColInfo = taosArrayGet(pInfo->matchInfo.pList, i);	4,262,009✔
161	QUERY_CHECK_NULL(pColInfo, code, lino, _error, terrno);	4,261,365✔
162
163	void* tmp = taosArrayPush(pCidList, &pColInfo->colId);	4,261,365✔
164	QUERY_CHECK_NULL(tmp, code, lino, _error, terrno);	4,262,757✔
165	if (pInfo->pFuncTypeList != NULL && taosArrayGetSize(pInfo->pFuncTypeList) > i) {	4,262,757✔
166	void* pFuncType = taosArrayGet(pInfo->pFuncTypeList, i);	×
167	QUERY_CHECK_NULL(pFuncType, code, lino, _error, terrno);	×
168	pColInfo->funcType = (int32_t)pFuncType;	×
169	}
170	}
171	pInfo->pCidList = pCidList;	1,829,819✔
172
173	code = extractCacheScanSlotId(pInfo->matchInfo.pList, pTaskInfo, &pInfo->pSlotIds, &pInfo->pDstSlotIds);	1,829,377✔
174	QUERY_CHECK_CODE(code, lino, _error);	1,829,377✔
175
176	int32_t totalTables = 0;	1,829,377✔
177	code = tableListGetSize(pTableListInfo, &totalTables);	1,829,377✔
178	QUERY_CHECK_CODE(code, lino, _error);	1,830,463✔
179
180	int32_t capacity = 0;	1,830,463✔
181
182	pInfo->pUidList = taosArrayInit(4, sizeof(int64_t));	1,830,463✔
183	QUERY_CHECK_NULL(pInfo->pUidList, code, lino, _error, terrno);	1,830,463✔
184
185	// partition by tbname
186	if (oneTableForEachGroup(pTableListInfo) \|\| (totalTables == 1)) {	2,837,598✔
187	pInfo->retrieveType = CACHESCAN_RETRIEVE_TYPE_ALL \| SCAN_ROW_TYPE(pScanNode->ignoreNull);	1,008,865✔
188
189	STableKeyInfo* pList = tableListGetInfo(pTableListInfo, 0);	1,008,865✔
190	if (totalTables) QUERY_CHECK_NULL(pList, code, lino, _error, terrno);	1,008,865✔
191
192	uint64_t suid = tableListGetSuid(pTableListInfo);	1,008,865✔
193	code = pInfo->readHandle.api.cacheFn.openReader(pInfo->readHandle.vnode, pInfo->retrieveType, pList, totalTables,	3,026,595✔
194	taosArrayGetSize(pInfo->matchInfo.pList), pCidList, pInfo->pSlotIds,	1,008,865✔
195	suid, &pInfo->pLastrowReader, pTaskInfo->id.str,	1,008,221✔
196	pScanNode->pFuncTypes, &pInfo->pkCol, pInfo->numOfPks);
197	QUERY_CHECK_CODE(code, lino, _error);	1,008,865✔
198
199	capacity = TMIN(totalTables, 4096);	1,008,865✔
200
201	pInfo->pBufferedRes = NULL;	1,008,865✔
202	code = createOneDataBlock(pInfo->pRes, false, &pInfo->pBufferedRes);	1,008,865✔
203	QUERY_CHECK_CODE(code, lino, _error);	1,008,865✔
204
205	code = setColIdForCacheReadBlock(pInfo->pBufferedRes, pScanNode);	1,008,865✔
206	QUERY_CHECK_CODE(code, lino, _error);	1,008,865✔
207
208	code = blockDataEnsureCapacity(pInfo->pBufferedRes, capacity);	1,008,865✔
209	QUERY_CHECK_CODE(code, lino, _error);	1,008,865✔
210	} else { // by tags
211	pInfo->retrieveType = CACHESCAN_RETRIEVE_TYPE_SINGLE \| SCAN_ROW_TYPE(pScanNode->ignoreNull);	820,954✔
212	capacity = 1; // only one row output	820,954✔
213	code = setColIdForCacheReadBlock(pInfo->pRes, pScanNode);	820,954✔
214	QUERY_CHECK_CODE(code, lino, _error);	820,954✔
215	}
216
217	initResultSizeInfo(&pOperator->resultInfo, capacity);	1,829,819✔
218	code = blockDataEnsureCapacity(pInfo->pRes, pOperator->resultInfo.capacity);	1,829,071✔
219	QUERY_CHECK_CODE(code, lino, _error);	1,829,819✔
220
221	if (pScanNode->scan.pScanPseudoCols != NULL) {	1,829,819✔
222	SExprSupp* p = &pInfo->pseudoExprSup;	188,704✔
223	code = createExprInfo(pScanNode->scan.pScanPseudoCols, NULL, &p->pExprInfo, &p->numOfExprs);	188,704✔
224	TSDB_CHECK_CODE(code, lino, _error);	187,830✔
225
226	p->pCtx =	187,398✔
227	createSqlFunctionCtx(p->pExprInfo, p->numOfExprs, &p->rowEntryInfoOffset, &pTaskInfo->storageAPI.functionStore);	187,830✔
228	QUERY_CHECK_NULL(p->pCtx, code, lino, _error, terrno);	187,398✔
229	}
230
231	setOperatorInfo(pOperator, "CachedRowScanOperator", QUERY_NODE_PHYSICAL_PLAN_LAST_ROW_SCAN, false, OP_NOT_OPENED,	1,830,031✔
232	pInfo, pTaskInfo);
233	pOperator->exprSupp.numOfExprs = taosArrayGetSize(pInfo->pRes->pDataBlock);	1,829,387✔
234
235	pOperator->fpSet = createOperatorFpSet(optrDummyOpenFn, doScanCacheNext, NULL, destroyCacheScanOperator, optrDefaultBufFn,	1,829,283✔
236	NULL, optrDefaultGetNextExtFn, NULL);
237
238	*pOptrInfo = pOperator;	1,828,197✔
239	return code;	1,828,629✔
240
241	_error:	×
242	pTaskInfo->code = code;	×
243	if (code != TSDB_CODE_SUCCESS) {	×
244	qError("%s failed at line %d since %s", __func__, lino, tstrerror(code));	×
245	}
246	if (pInfo != NULL) {	×
247	pInfo->pTableList = NULL;	×
248	destroyCacheScanOperator(pInfo);	×
249	}
250	if (pOperator != NULL) {	×
251	pOperator->info = NULL;	×
252	destroyOperator(pOperator);	×
253	}
254	return code;	×
255	}
256
257	static int32_t doScanCacheNext(SOperatorInfo* pOperator, SSDataBlock** ppRes) {	3,272,460✔
258	int32_t code = TSDB_CODE_SUCCESS;	3,272,460✔
259	int32_t lino = 0;	3,272,460✔
260	if (pOperator->status == OP_EXEC_DONE) {	3,272,460✔
261	(*ppRes) = NULL;	×
262	return code;	×
263	}
264
265	SCacheRowsScanInfo* pInfo = pOperator->info;	3,272,460✔
266	SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;	3,272,460✔
267	STableListInfo* pTableList = pInfo->pTableList;	3,272,460✔
268	SStoreCacheReader* pReaderFn = &pInfo->readHandle.api.cacheFn;	3,272,902✔
269	SSDataBlock* pBufRes = pInfo->pBufferedRes;	3,272,902✔
270
271	uint64_t suid = tableListGetSuid(pTableList);	3,272,902✔
272	int32_t size = 0;	3,272,902✔
273	code = tableListGetSize(pTableList, &size);	3,272,902✔
274	QUERY_CHECK_CODE(code, lino, _end);	3,272,902✔
275
276	if (size == 0) {	3,272,902✔
277	setOperatorCompleted(pOperator);	437,810✔
278	(*ppRes) = NULL;	437,810✔
279	return code;	437,810✔
280	}
281
282	blockDataCleanup(pInfo->pRes);	2,835,092✔
283
284	// check if it is a group by tbname
285	if ((pInfo->retrieveType & CACHESCAN_RETRIEVE_TYPE_ALL) == CACHESCAN_RETRIEVE_TYPE_ALL) {	2,835,092✔
286	if (isTaskKilled(pTaskInfo)) {	2,056,728✔
287	T_LONG_JMP(pTaskInfo->env, pTaskInfo->code);	×
288	}
289
290	if (!pInfo->gotAll) {	2,056,728✔
291	blockDataCleanup(pBufRes);	2,056,728✔
292	taosArrayClear(pInfo->pUidList);	2,056,728✔
293
294	code = pReaderFn->retrieveRows(pInfo->pLastrowReader, pBufRes, pInfo->pSlotIds, pInfo->pDstSlotIds,	2,056,728✔
295	pInfo->pUidList, &pInfo->gotAll);
296	QUERY_CHECK_CODE(code, lino, _end);	2,056,728✔
297	}
298
299	SSDataBlock* pRes = pInfo->pRes;	2,056,728✔
300
301	if (pBufRes->info.rows > 0) {	2,056,728✔
302	code = copyDataBlock(pRes, pBufRes);	1,047,863✔
303	QUERY_CHECK_CODE(code, lino, _end);	1,047,863✔
304
305	void* pUid = taosArrayGet(pInfo->pUidList, 0);	1,047,863✔
306	QUERY_CHECK_NULL(pUid, code, lino, _end, terrno);	1,047,863✔
307
308	pRes->info.id.uid = (tb_uid_t)pUid;	1,047,863✔
309	pRes->info.rows = pBufRes->info.rows;	1,047,863✔
310	pRes->info.scanFlag = MAIN_SCAN;	1,047,863✔
311
312	SExprSupp* pSup = &pInfo->pseudoExprSup;	1,047,863✔
313	code = addTagPseudoColumnData(&pInfo->readHandle, pSup->pExprInfo, pSup->numOfExprs, pRes, pRes->info.rows,	1,047,863✔
314	pTaskInfo, NULL);
315	QUERY_CHECK_CODE(code, lino, _end);	1,047,863✔
316
317	code = tableListGetTableGroupId(pTableList, pRes->info.id.uid, &pRes->info.id.groupId, &pRes->info.id.baseGId);	1,047,863✔
318	QUERY_CHECK_CODE(code, lino, _end);	1,047,863✔
319	(*ppRes) = pRes;	1,047,863✔
320	return code;	1,047,863✔
321	} else {
322	setOperatorCompleted(pOperator);	1,008,865✔
323	(*ppRes) = NULL;	1,008,865✔
324	return code;	1,008,865✔
325	}
326	} else {
327	size_t totalGroups = tableListGetOutputGroups(pTableList);	778,364✔
328
329	while (pInfo->currentGroupIndex < totalGroups) {	780,092✔
330	if (isTaskKilled(pTaskInfo)) {	396,304✔
331	T_LONG_JMP(pTaskInfo->env, pTaskInfo->code);	×
332	}
333
334	STableKeyInfo* pList = NULL;	396,304✔
335	int32_t num = 0;	396,304✔
336
337	code = tableListGetGroupList(pTableList, pInfo->currentGroupIndex, &pList, &num);	396,304✔
338	QUERY_CHECK_CODE(code, lino, _end);	396,304✔
339
340	if (NULL == pInfo->pLastrowReader) {	396,304✔
341	int32_t tmpRes = pReaderFn->openReader(pInfo->readHandle.vnode, pInfo->retrieveType, pList, num,	1,151,364✔
342	taosArrayGetSize(pInfo->matchInfo.pList), pInfo->pCidList,	383,788✔
343	pInfo->pSlotIds, suid, &pInfo->pLastrowReader, pTaskInfo->id.str,	383,788✔
344	pInfo->pFuncTypeList, &pInfo->pkCol, pInfo->numOfPks);
345
346	if (tmpRes != TSDB_CODE_SUCCESS) {	383,788✔
347	pInfo->currentGroupIndex += 1;	×
348	taosArrayClear(pInfo->pUidList);	×
349	continue;	×
350	}
351	} else {
352	code = pReaderFn->reuseReader(pInfo->pLastrowReader, pList, num);	12,516✔
353	QUERY_CHECK_CODE(code, lino, _end);	12,516✔
354	}
355
356	taosArrayClear(pInfo->pUidList);	396,304✔
357
358	code = pReaderFn->retrieveRows(pInfo->pLastrowReader, pInfo->pRes, pInfo->pSlotIds, pInfo->pDstSlotIds,	396,304✔
359	pInfo->pUidList, NULL);
360	QUERY_CHECK_CODE(code, lino, _end);	396,304✔
361
362	pInfo->currentGroupIndex += 1;	396,304✔
363
364	// check for tag values
365	if (pInfo->pRes->info.rows > 0) {	396,304✔
366	SExprSupp* pSup = &pInfo->pseudoExprSup;	394,576✔
367
368	STableKeyInfo* pKeyInfo = &((STableKeyInfo*)pList)[0];	394,576✔
369	pInfo->pRes->info.id.groupId = pKeyInfo->groupId;	394,576✔
370
371	if (taosArrayGetSize(pInfo->pUidList) > 0) {	394,576✔
372	void* pUid = taosArrayGet(pInfo->pUidList, 0);	394,576✔
373	QUERY_CHECK_NULL(pUid, code, lino, _end, terrno);	394,576✔
374	pInfo->pRes->info.id.uid = (tb_uid_t)pUid;	394,576✔
375	if (pInfo->pseudoExprSup.numOfExprs > 0) {	394,576✔
376	code = addTagPseudoColumnData(&pInfo->readHandle, pSup->pExprInfo, pSup->numOfExprs, pInfo->pRes,	39,582✔
377	pInfo->pRes->info.rows, pTaskInfo, NULL);	39,582✔
378	QUERY_CHECK_CODE(code, lino, _end);	39,582✔
379	}
380	}
381
382	// pInfo->pLastrowReader = tsdbCacherowsReaderClose(pInfo->pLastrowReader);
383	(*ppRes) = pInfo->pRes;	394,576✔
384	return code;	393,932✔
385	} else {
386	// pInfo->pLastrowReader = tsdbCacherowsReaderClose(pInfo->pLastrowReader);
387	}
388	}
389
390	pReaderFn->closeReader(pInfo->pLastrowReader);	383,788✔
391	pInfo->pLastrowReader = NULL;	383,788✔
392	setOperatorCompleted(pOperator);	383,788✔
393	(*ppRes) = NULL;	383,788✔
394	return code;	383,788✔
395	}
396
397	_end:	×
398	if (code != TSDB_CODE_SUCCESS) {	×
399	qError("%s failed at line %d since %s", __func__, lino, tstrerror(code));	×
400	pTaskInfo->code = code;	×
401	T_LONG_JMP(pTaskInfo->env, code);	×
402	}
403	return code;	×
404	}
405
406	void destroyCacheScanOperator(void* param) {	1,830,463✔
407	SCacheRowsScanInfo* pInfo = (SCacheRowsScanInfo*)param;	1,830,463✔
408	blockDataDestroy(pInfo->pRes);	1,830,463✔
409	blockDataDestroy(pInfo->pBufferedRes);	1,830,463✔
410	taosMemoryFreeClear(pInfo->pSlotIds);	1,830,463✔
411	taosMemoryFreeClear(pInfo->pDstSlotIds);	1,830,463✔
412	taosArrayDestroy(pInfo->pCidList);	1,830,463✔
413	taosArrayDestroy(pInfo->pFuncTypeList);	1,830,463✔
414	taosArrayDestroy(pInfo->pUidList);	1,830,463✔
415	taosArrayDestroy(pInfo->matchInfo.pList);	1,830,463✔
416	tableListDestroy(pInfo->pTableList);	1,830,463✔
417
418	if (pInfo->pLastrowReader != NULL && pInfo->readHandle.api.cacheFn.closeReader != NULL) {	1,830,463✔
419	pInfo->readHandle.api.cacheFn.closeReader(pInfo->pLastrowReader);	1,008,865✔
420	pInfo->pLastrowReader = NULL;	1,008,865✔
421	}
422
423	cleanupExprSupp(&pInfo->pseudoExprSup);	1,830,463✔
424	taosMemoryFreeClear(param);	1,830,463✔
425	}	1,830,463✔
426
427	int32_t extractCacheScanSlotId(const SArray* pColMatchInfo, SExecTaskInfo* pTaskInfo, int32_t** pSlotIds,	1,829,377✔
428	int32_t** pDstSlotIds) {
429	size_t numOfCols = taosArrayGetSize(pColMatchInfo);	1,829,377✔
430
431	pSlotIds = taosMemoryMalloc(numOfCols sizeof(int32_t));	1,830,463✔
432	if (*pSlotIds == NULL) {	1,830,463✔
433	return terrno;	×
434	}
435
436	pDstSlotIds = taosMemoryMalloc(numOfCols sizeof(int32_t));	1,828,629✔
437	if (*pDstSlotIds == NULL) {	1,829,819✔
438	taosMemoryFreeClear(*pSlotIds);	×
439	return terrno;	×
440	}
441
442	SSchemaInfo* pSchemaInfo = taosArrayGetLast(pTaskInfo->schemaInfos);	1,830,021✔
443	SSchemaWrapper* pWrapper = pSchemaInfo->sw;	1,829,819✔
444
445	for (int32_t i = 0; i < numOfCols; ++i) {	6,092,576✔
446	SColMatchItem* pColMatch = taosArrayGet(pColMatchInfo, i);	4,262,113✔
447	if (!pColMatch) {	4,262,757✔
448	return terrno;	×
449	}
450	bool found = false;	4,262,757✔
451	for (int32_t j = 0; j < pWrapper->nCols; ++j) {	13,186,439✔
452	/* if (pColMatch->colId == pWrapper->pSchema[j].colId && pColMatch->colId == PRIMARYKEY_TIMESTAMP_COL_ID) {
453	(*pSlotIds)[pColMatch->dstSlotId] = -1;
454	break;
455	}*/
456
457	if (pColMatch->colId == pWrapper->pSchema[j].colId) {	13,185,691✔
458	(*pSlotIds)[i] = j;	4,262,113✔
459	(*pDstSlotIds)[i] = pColMatch->dstSlotId;	4,262,113✔
460	found = true;	4,261,365✔
461	break;	4,261,365✔
462	}
463	}
464	if (!found) {	4,262,113✔
465	(*pSlotIds)[i] = -1;	×
466	(*pDstSlotIds)[i] = pColMatch->dstSlotId;	×
467	}
468	}
469
470	return TSDB_CODE_SUCCESS;	1,830,463✔
471	}

taosdata / TDengine / #5011

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