#3814

Committed 01 Apr 2025 05:36AM UTC coverage: 30.169% (-3.6%) from 33.798%

Build # #3814

Build Type

push

travis-ci

Committed by

happyguoxy

Commit Message

test:add build cmake

Run Details

129680 of 592945 branches covered (21.87%)

Branch coverage included in aggregate %.

196213 of 487270 relevant lines covered (40.27%)

555639.42 hits per line

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

0.0

/source/libs/executor/src/dataInserter.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 "dataSinkInt.h"
#include "dataSinkMgt.h"
#include "executorInt.h"
#include "planner.h"
#include "storageapi.h"
#include "tcompression.h"
#include "tdatablock.h"
#include "tglobal.h"
#include "tqueue.h"

extern SDataSinkStat gDataSinkStat;

typedef struct SSubmitRes {
  int64_t      affectedRows;
  int32_t      code;
  SSubmitRsp2* pRsp;
} SSubmitRes;

typedef struct SDataInserterHandle {
  SDataSinkHandle     sink;
  SDataSinkManager*   pManager;
  STSchema*           pSchema;
  SQueryInserterNode* pNode;
  SSubmitRes          submitRes;
  SInserterParam*     pParam;
  SArray*             pDataBlocks;
  SHashObj*           pCols;
  int32_t             status;
  bool                queryEnd;
  bool                fullOrderColList;
  uint64_t            useconds;
  uint64_t            cachedSize;
  uint64_t            flags;
  TdThreadMutex       mutex;
  tsem_t              ready;
  bool                explain;
} SDataInserterHandle;

typedef struct SSubmitRspParam {
  SDataInserterHandle* pInserter;
} SSubmitRspParam;

int32_t inserterCallback(void* param, SDataBuf* pMsg, int32_t code) {
  SSubmitRspParam*     pParam = (SSubmitRspParam*)param;
  SDataInserterHandle* pInserter = pParam->pInserter;
  int32_t code2 = 0;

  if (code) {
    pInserter->submitRes.code = code;
  }

  if (code == TSDB_CODE_SUCCESS) {
    pInserter->submitRes.pRsp = taosMemoryCalloc(1, sizeof(SSubmitRsp2));
    if (NULL == pInserter->submitRes.pRsp) {
      pInserter->submitRes.code = terrno;
      goto _return;
    }

    SDecoder coder = {0};
    tDecoderInit(&coder, pMsg->pData, pMsg->len);
    code = tDecodeSSubmitRsp2(&coder, pInserter->submitRes.pRsp);
    if (code) {
      taosMemoryFree(pInserter->submitRes.pRsp);
      pInserter->submitRes.code = code;
      goto _return;
    }

    if (pInserter->submitRes.pRsp->affectedRows > 0) {
      SArray* pCreateTbList = pInserter->submitRes.pRsp->aCreateTbRsp;
      int32_t numOfTables = taosArrayGetSize(pCreateTbList);

      for (int32_t i = 0; i < numOfTables; ++i) {
        SVCreateTbRsp* pRsp = taosArrayGet(pCreateTbList, i);
        if (NULL == pRsp) {
          pInserter->submitRes.code = TSDB_CODE_QRY_EXECUTOR_INTERNAL_ERROR;
          goto _return;
        }
        if (TSDB_CODE_SUCCESS != pRsp->code) {
          code = pRsp->code;
          taosMemoryFree(pInserter->submitRes.pRsp);
          pInserter->submitRes.code = code;
          goto _return;
        }
      }
    }

    pInserter->submitRes.affectedRows += pInserter->submitRes.pRsp->affectedRows;
    qDebug("submit rsp received, affectedRows:%d, total:%" PRId64, pInserter->submitRes.pRsp->affectedRows,
           pInserter->submitRes.affectedRows);
    tDecoderClear(&coder);
    taosMemoryFree(pInserter->submitRes.pRsp);
  }

_return:

  code2 = tsem_post(&pInserter->ready);
  if (code2 < 0) {
    qError("tsem_post inserter ready failed, error:%s", tstrerror(code2));
    if (TSDB_CODE_SUCCESS == code) {
      pInserter->submitRes.code = code2;
    }
  }
  
  taosMemoryFree(pMsg->pData);

  return TSDB_CODE_SUCCESS;
}

static int32_t sendSubmitRequest(SDataInserterHandle* pInserter, void* pMsg, int32_t msgLen, void* pTransporter,
                                 SEpSet* pEpset) {
  // send the fetch remote task result reques
  SMsgSendInfo* pMsgSendInfo = taosMemoryCalloc(1, sizeof(SMsgSendInfo));
  if (NULL == pMsgSendInfo) {
    taosMemoryFreeClear(pMsg);
    return terrno;
  }

  SSubmitRspParam* pParam = taosMemoryCalloc(1, sizeof(SSubmitRspParam));
  if (NULL == pParam) {
    taosMemoryFreeClear(pMsg);
    taosMemoryFreeClear(pMsgSendInfo);
    return terrno;
  }
  pParam->pInserter = pInserter;

  pMsgSendInfo->param = pParam;
  pMsgSendInfo->paramFreeFp = taosAutoMemoryFree;
  pMsgSendInfo->msgInfo.pData = pMsg;
  pMsgSendInfo->msgInfo.len = msgLen;
  pMsgSendInfo->msgType = TDMT_VND_SUBMIT;
  pMsgSendInfo->fp = inserterCallback;

  return asyncSendMsgToServer(pTransporter, pEpset, NULL, pMsgSendInfo);
}

static int32_t submitReqToMsg(int32_t vgId, SSubmitReq2* pReq, void** pData, int32_t* pLen) {
  int32_t code = TSDB_CODE_SUCCESS;
  int32_t len = 0;
  void*   pBuf = NULL;
  tEncodeSize(tEncodeSubmitReq, pReq, len, code);
  if (TSDB_CODE_SUCCESS == code) {
    SEncoder encoder;
    len += sizeof(SSubmitReq2Msg);
    pBuf = taosMemoryMalloc(len);
    if (NULL == pBuf) {
      return terrno;
    }
    ((SSubmitReq2Msg*)pBuf)->header.vgId = htonl(vgId);
    ((SSubmitReq2Msg*)pBuf)->header.contLen = htonl(len);
    ((SSubmitReq2Msg*)pBuf)->version = htobe64(1);
    tEncoderInit(&encoder, POINTER_SHIFT(pBuf, sizeof(SSubmitReq2Msg)), len - sizeof(SSubmitReq2Msg));
    code = tEncodeSubmitReq(&encoder, pReq);
    tEncoderClear(&encoder);
  }

  if (TSDB_CODE_SUCCESS == code) {
    *pData = pBuf;
    *pLen = len;
  } else {
    taosMemoryFree(pBuf);
  }

  return code;
}

int32_t buildSubmitReqFromBlock(SDataInserterHandle* pInserter, SSubmitReq2** ppReq, const SSDataBlock* pDataBlock,
                                const STSchema* pTSchema, int64_t uid, int32_t vgId, tb_uid_t suid) {
  SSubmitReq2* pReq = *ppReq;
  SArray*      pVals = NULL;
  int32_t      numOfBlks = 0;

  terrno = TSDB_CODE_SUCCESS;

  if (NULL == pReq) {
    if (!(pReq = taosMemoryCalloc(1, sizeof(SSubmitReq2)))) {
      goto _end;
    }

    if (!(pReq->aSubmitTbData = taosArrayInit(1, sizeof(SSubmitTbData)))) {
      goto _end;
    }
  }

  int32_t colNum = taosArrayGetSize(pDataBlock->pDataBlock);
  int32_t rows = pDataBlock->info.rows;

  SSubmitTbData tbData = {0};
  if (!(tbData.aRowP = taosArrayInit(rows, sizeof(SRow*)))) {
    goto _end;
  }
  tbData.suid = suid;
  tbData.uid = uid;
  tbData.sver = pTSchema->version;

  if (!pVals && !(pVals = taosArrayInit(colNum, sizeof(SColVal)))) {
    taosArrayDestroy(tbData.aRowP);
    goto _end;
  }

  int64_t lastTs = TSKEY_MIN;
  bool    needSortMerge = false;

  for (int32_t j = 0; j < rows; ++j) {  // iterate by row
    taosArrayClear(pVals);

    int32_t offset = 0;
    for (int32_t k = 0; k < pTSchema->numOfCols; ++k) {  // iterate by column
      int16_t         colIdx = k;
      const STColumn* pCol = &pTSchema->columns[k];
      if (!pInserter->fullOrderColList) {
        int16_t* slotId = taosHashGet(pInserter->pCols, &pCol->colId, sizeof(pCol->colId));
        if (NULL == slotId) {
          continue;
        }

        colIdx = *slotId;
      }

      SColumnInfoData* pColInfoData = taosArrayGet(pDataBlock->pDataBlock, colIdx);
      if (NULL == pColInfoData) {
        terrno = TSDB_CODE_QRY_EXECUTOR_INTERNAL_ERROR;
        goto _end;
      }
      void* var = POINTER_SHIFT(pColInfoData->pData, j * pColInfoData->info.bytes);

      switch (pColInfoData->info.type) {
        case TSDB_DATA_TYPE_NCHAR:
        case TSDB_DATA_TYPE_VARBINARY:
        case TSDB_DATA_TYPE_VARCHAR: {  // TSDB_DATA_TYPE_BINARY
          if (pColInfoData->info.type != pCol->type) {
            qError("column:%d type:%d in block dismatch with schema col:%d type:%d", colIdx, pColInfoData->info.type, k,
                   pCol->type);
            terrno = TSDB_CODE_QRY_EXECUTOR_INTERNAL_ERROR;
            goto _end;
          }
          if (colDataIsNull_s(pColInfoData, j)) {
            SColVal cv = COL_VAL_NULL(pCol->colId, pCol->type);
            if (NULL == taosArrayPush(pVals, &cv)) {
              goto _end;
            }
          } else {
            void*  data = colDataGetVarData(pColInfoData, j);
            SValue sv = (SValue){
                .type = pCol->type, .nData = varDataLen(data), .pData = varDataVal(data)};  // address copy, no value
            SColVal cv = COL_VAL_VALUE(pCol->colId, sv);
            if (NULL == taosArrayPush(pVals, &cv)) {
              goto _end;
            }
          }
          break;
        }
        case TSDB_DATA_TYPE_BLOB:
        case TSDB_DATA_TYPE_JSON:
        case TSDB_DATA_TYPE_MEDIUMBLOB:
          qError("the column type %" PRIi16 " is defined but not implemented yet", pColInfoData->info.type);
          terrno = TSDB_CODE_APP_ERROR;
          goto _end;
          break;
        default:
          if (pColInfoData->info.type < TSDB_DATA_TYPE_MAX && pColInfoData->info.type > TSDB_DATA_TYPE_NULL) {
            if (colDataIsNull_s(pColInfoData, j)) {
              if (PRIMARYKEY_TIMESTAMP_COL_ID == pCol->colId) {
                qError("Primary timestamp column should not be null");
                terrno = TSDB_CODE_PAR_INCORRECT_TIMESTAMP_VAL;
                goto _end;
              }

              SColVal cv = COL_VAL_NULL(pCol->colId, pCol->type);  // should use pCol->type
              if (NULL == taosArrayPush(pVals, &cv)) {
                goto _end;
              }
            } else {
              if (PRIMARYKEY_TIMESTAMP_COL_ID == pCol->colId && !needSortMerge) {
                if (*(int64_t*)var <= lastTs) {
                  needSortMerge = true;
                } else {
                  lastTs = *(int64_t*)var;
                }
              }

              SValue sv = {.type = pCol->type};
              valueSetDatum(&sv, sv.type, var, tDataTypes[pCol->type].bytes);
              SColVal cv = COL_VAL_VALUE(pCol->colId, sv);
              if (NULL == taosArrayPush(pVals, &cv)) {
                goto _end;
              }
            }
          } else {
            uError("the column type %" PRIi16 " is undefined\n", pColInfoData->info.type);
            terrno = TSDB_CODE_APP_ERROR;
            goto _end;
          }
          break;
      }
    }

    SRow* pRow = NULL;
    if ((terrno = tRowBuild(pVals, pTSchema, &pRow)) < 0) {
      tDestroySubmitTbData(&tbData, TSDB_MSG_FLG_ENCODE);
      goto _end;
    }
    if (NULL == taosArrayPush(tbData.aRowP, &pRow)) {
      goto _end;
    }
  }

  if (needSortMerge) {
    if ((tRowSort(tbData.aRowP) != TSDB_CODE_SUCCESS) ||
        (terrno = tRowMerge(tbData.aRowP, (STSchema*)pTSchema, 0)) != 0) {
      goto _end;
    }
  }

  if (NULL == taosArrayPush(pReq->aSubmitTbData, &tbData)) {
    goto _end;
  }

_end:

  taosArrayDestroy(pVals);
  if (terrno != 0) {
    *ppReq = NULL;
    if (pReq) {
      tDestroySubmitReq(pReq, TSDB_MSG_FLG_ENCODE);
      taosMemoryFree(pReq);
    }

    return terrno;
  }
  *ppReq = pReq;

  return TSDB_CODE_SUCCESS;
}

int32_t dataBlocksToSubmitReq(SDataInserterHandle* pInserter, void** pMsg, int32_t* msgLen) {
  const SArray*   pBlocks = pInserter->pDataBlocks;
  const STSchema* pTSchema = pInserter->pSchema;
  int64_t         uid = pInserter->pNode->tableId;
  int64_t         suid = pInserter->pNode->stableId;
  int32_t         vgId = pInserter->pNode->vgId;
  int32_t         sz = taosArrayGetSize(pBlocks);
  int32_t         code = 0;
  SSubmitReq2*    pReq = NULL;

  for (int32_t i = 0; i < sz; i++) {
    SSDataBlock* pDataBlock = taosArrayGetP(pBlocks, i);
    if (NULL == pDataBlock) {
      return TSDB_CODE_QRY_EXECUTOR_INTERNAL_ERROR;
    }
    code = buildSubmitReqFromBlock(pInserter, &pReq, pDataBlock, pTSchema, uid, vgId, suid);
    if (code) {
      if (pReq) {
        tDestroySubmitReq(pReq, TSDB_MSG_FLG_ENCODE);
        taosMemoryFree(pReq);
      }

      return code;
    }
  }

  code = submitReqToMsg(vgId, pReq, pMsg, msgLen);
  tDestroySubmitReq(pReq, TSDB_MSG_FLG_ENCODE);
  taosMemoryFree(pReq);

  return code;
}

static int32_t putDataBlock(SDataSinkHandle* pHandle, const SInputData* pInput, bool* pContinue) {
  SDataInserterHandle* pInserter = (SDataInserterHandle*)pHandle;
  if (!pInserter->explain) {
    if (NULL == taosArrayPush(pInserter->pDataBlocks, &pInput->pData)) {
      return terrno;
    }
    void*   pMsg = NULL;
    int32_t msgLen = 0;
    int32_t code = dataBlocksToSubmitReq(pInserter, &pMsg, &msgLen);
    if (code) {
      return code;
    }

    taosArrayClear(pInserter->pDataBlocks);

    code = sendSubmitRequest(pInserter, pMsg, msgLen, pInserter->pParam->readHandle->pMsgCb->clientRpc,
                             &pInserter->pNode->epSet);
    if (code) {
      return code;
    }

    QRY_ERR_RET(tsem_wait(&pInserter->ready));

    if (pInserter->submitRes.code) {
      return pInserter->submitRes.code;
    }
  }

  *pContinue = true;

  return TSDB_CODE_SUCCESS;
}

static void endPut(struct SDataSinkHandle* pHandle, uint64_t useconds) {
  SDataInserterHandle* pInserter = (SDataInserterHandle*)pHandle;
  (void)taosThreadMutexLock(&pInserter->mutex);
  pInserter->queryEnd = true;
  pInserter->useconds = useconds;
  (void)taosThreadMutexUnlock(&pInserter->mutex);
}

static void getDataLength(SDataSinkHandle* pHandle, int64_t* pLen, int64_t* pRawLen, bool* pQueryEnd) {
  SDataInserterHandle* pDispatcher = (SDataInserterHandle*)pHandle;
  *pLen = pDispatcher->submitRes.affectedRows;
  qDebug("got total affectedRows %" PRId64, *pLen);
}

static int32_t destroyDataSinker(SDataSinkHandle* pHandle) {
  SDataInserterHandle* pInserter = (SDataInserterHandle*)pHandle;
  (void)atomic_sub_fetch_64(&gDataSinkStat.cachedSize, pInserter->cachedSize);
  taosArrayDestroy(pInserter->pDataBlocks);
  taosMemoryFree(pInserter->pSchema);
  taosMemoryFree(pInserter->pParam);
  taosHashCleanup(pInserter->pCols);
  nodesDestroyNode((SNode *)pInserter->pNode);
  pInserter->pNode = NULL;
  
  (void)taosThreadMutexDestroy(&pInserter->mutex);

  taosMemoryFree(pInserter->pManager);
  return TSDB_CODE_SUCCESS;
}

static int32_t getCacheSize(struct SDataSinkHandle* pHandle, uint64_t* size) {
  SDataInserterHandle* pDispatcher = (SDataInserterHandle*)pHandle;

  *size = atomic_load_64(&pDispatcher->cachedSize);
  return TSDB_CODE_SUCCESS;
}

static int32_t getSinkFlags(struct SDataSinkHandle* pHandle, uint64_t* pFlags) {
  SDataInserterHandle* pDispatcher = (SDataInserterHandle*)pHandle;

  *pFlags = atomic_load_64(&pDispatcher->flags);
  return TSDB_CODE_SUCCESS;
}

int32_t createDataInserter(SDataSinkManager* pManager, SDataSinkNode** ppDataSink, DataSinkHandle* pHandle,
                           void* pParam) {
  SDataSinkNode* pDataSink = *ppDataSink;
  SDataInserterHandle* inserter = taosMemoryCalloc(1, sizeof(SDataInserterHandle));
  if (NULL == inserter) {
    taosMemoryFree(pParam);
    goto _return;
  }

  SQueryInserterNode* pInserterNode = (SQueryInserterNode*)pDataSink;
  inserter->sink.fPut = putDataBlock;
  inserter->sink.fEndPut = endPut;
  inserter->sink.fGetLen = getDataLength;
  inserter->sink.fGetData = NULL;
  inserter->sink.fDestroy = destroyDataSinker;
  inserter->sink.fGetCacheSize = getCacheSize;
  inserter->sink.fGetFlags = getSinkFlags;
  inserter->pManager = pManager;
  inserter->pNode = pInserterNode;
  inserter->pParam = pParam;
  inserter->status = DS_BUF_EMPTY;
  inserter->queryEnd = false;
  inserter->explain = pInserterNode->explain;
  *ppDataSink = NULL;

  int64_t suid = 0;
  int32_t code = pManager->pAPI->metaFn.getTableSchema(inserter->pParam->readHandle->vnode, pInserterNode->tableId,
                                                       &inserter->pSchema, &suid);
  if (code) {
    terrno = code;
    goto _return;
  }

  if (pInserterNode->stableId != suid) {
    terrno = TSDB_CODE_TDB_INVALID_TABLE_ID;
    goto _return;
  }

  inserter->pDataBlocks = taosArrayInit(1, POINTER_BYTES);
  if (NULL == inserter->pDataBlocks) {
    goto _return;
  }
  QRY_ERR_JRET(taosThreadMutexInit(&inserter->mutex, NULL));

  inserter->fullOrderColList = pInserterNode->pCols->length == inserter->pSchema->numOfCols;

  inserter->pCols = taosHashInit(pInserterNode->pCols->length, taosGetDefaultHashFunction(TSDB_DATA_TYPE_SMALLINT),
                                 false, HASH_NO_LOCK);
  if (NULL == inserter->pCols) {
    goto _return;
  }

  SNode*  pNode = NULL;
  int32_t i = 0;
  FOREACH(pNode, pInserterNode->pCols) {
    SColumnNode* pCol = (SColumnNode*)pNode;
    QRY_ERR_JRET(taosHashPut(inserter->pCols, &pCol->colId, sizeof(pCol->colId), &pCol->slotId, sizeof(pCol->slotId)));
    if (inserter->fullOrderColList && pCol->colId != inserter->pSchema->columns[i].colId) {
      inserter->fullOrderColList = false;
    }
    ++i;
  }

  QRY_ERR_JRET(tsem_init(&inserter->ready, 0, 0));

  *pHandle = inserter;
  return TSDB_CODE_SUCCESS;

_return:

  if (inserter) {
    (void)destroyDataSinker((SDataSinkHandle*)inserter);
    taosMemoryFree(inserter);
  } else {
    taosMemoryFree(pManager);
  }

  nodesDestroyNode((SNode *)*ppDataSink);
  *ppDataSink = NULL;

  return terrno;
}

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 "dataSinkInt.h"
17	#include "dataSinkMgt.h"
18	#include "executorInt.h"
19	#include "planner.h"
20	#include "storageapi.h"
21	#include "tcompression.h"
22	#include "tdatablock.h"
23	#include "tglobal.h"
24	#include "tqueue.h"
25
26	extern SDataSinkStat gDataSinkStat;
27
28	typedef struct SSubmitRes {
29	int64_t affectedRows;
30	int32_t code;
31	SSubmitRsp2* pRsp;
32	} SSubmitRes;
33
34	typedef struct SDataInserterHandle {
35	SDataSinkHandle sink;
36	SDataSinkManager* pManager;
37	STSchema* pSchema;
38	SQueryInserterNode* pNode;
39	SSubmitRes submitRes;
40	SInserterParam* pParam;
41	SArray* pDataBlocks;
42	SHashObj* pCols;
43	int32_t status;
44	bool queryEnd;
45	bool fullOrderColList;
46	uint64_t useconds;
47	uint64_t cachedSize;
48	uint64_t flags;
49	TdThreadMutex mutex;
50	tsem_t ready;
51	bool explain;
52	} SDataInserterHandle;
53
54	typedef struct SSubmitRspParam {
55	SDataInserterHandle* pInserter;
56	} SSubmitRspParam;
57
58	int32_t inserterCallback(void* param, SDataBuf* pMsg, int32_t code) {	×
59	SSubmitRspParam* pParam = (SSubmitRspParam*)param;	×
60	SDataInserterHandle* pInserter = pParam->pInserter;	×
61	int32_t code2 = 0;	×
62
63	if (code) {	×
64	pInserter->submitRes.code = code;	×
65	}
66
67	if (code == TSDB_CODE_SUCCESS) {	×
68	pInserter->submitRes.pRsp = taosMemoryCalloc(1, sizeof(SSubmitRsp2));	×
69	if (NULL == pInserter->submitRes.pRsp) {	×
70	pInserter->submitRes.code = terrno;	×
71	goto _return;	×
72	}
73
74	SDecoder coder = {0};	×
75	tDecoderInit(&coder, pMsg->pData, pMsg->len);	×
76	code = tDecodeSSubmitRsp2(&coder, pInserter->submitRes.pRsp);	×
77	if (code) {	×
78	taosMemoryFree(pInserter->submitRes.pRsp);	×
79	pInserter->submitRes.code = code;	×
80	goto _return;	×
81	}
82
83	if (pInserter->submitRes.pRsp->affectedRows > 0) {	×
84	SArray* pCreateTbList = pInserter->submitRes.pRsp->aCreateTbRsp;	×
85	int32_t numOfTables = taosArrayGetSize(pCreateTbList);	×
86
87	for (int32_t i = 0; i < numOfTables; ++i) {	×
88	SVCreateTbRsp* pRsp = taosArrayGet(pCreateTbList, i);	×
89	if (NULL == pRsp) {	×
90	pInserter->submitRes.code = TSDB_CODE_QRY_EXECUTOR_INTERNAL_ERROR;	×
91	goto _return;	×
92	}
93	if (TSDB_CODE_SUCCESS != pRsp->code) {	×
94	code = pRsp->code;	×
95	taosMemoryFree(pInserter->submitRes.pRsp);	×
96	pInserter->submitRes.code = code;	×
97	goto _return;	×
98	}
99	}
100	}
101
102	pInserter->submitRes.affectedRows += pInserter->submitRes.pRsp->affectedRows;	×
103	qDebug("submit rsp received, affectedRows:%d, total:%" PRId64, pInserter->submitRes.pRsp->affectedRows,	×
104	pInserter->submitRes.affectedRows);
105	tDecoderClear(&coder);	×
106	taosMemoryFree(pInserter->submitRes.pRsp);	×
107	}
108
109	_return:	×
110
111	code2 = tsem_post(&pInserter->ready);	×
112	if (code2 < 0) {	×
113	qError("tsem_post inserter ready failed, error:%s", tstrerror(code2));	×
114	if (TSDB_CODE_SUCCESS == code) {	×
115	pInserter->submitRes.code = code2;	×
116	}
117	}
118
119	taosMemoryFree(pMsg->pData);	×
120
121	return TSDB_CODE_SUCCESS;	×
122	}
123
124	static int32_t sendSubmitRequest(SDataInserterHandle* pInserter, void* pMsg, int32_t msgLen, void* pTransporter,	×
125	SEpSet* pEpset) {
126	// send the fetch remote task result reques
127	SMsgSendInfo* pMsgSendInfo = taosMemoryCalloc(1, sizeof(SMsgSendInfo));	×
128	if (NULL == pMsgSendInfo) {	×
129	taosMemoryFreeClear(pMsg);	×
130	return terrno;	×
131	}
132
133	SSubmitRspParam* pParam = taosMemoryCalloc(1, sizeof(SSubmitRspParam));	×
134	if (NULL == pParam) {	×
135	taosMemoryFreeClear(pMsg);	×
136	taosMemoryFreeClear(pMsgSendInfo);	×
137	return terrno;	×
138	}
139	pParam->pInserter = pInserter;	×
140
141	pMsgSendInfo->param = pParam;	×
142	pMsgSendInfo->paramFreeFp = taosAutoMemoryFree;	×
143	pMsgSendInfo->msgInfo.pData = pMsg;	×
144	pMsgSendInfo->msgInfo.len = msgLen;	×
145	pMsgSendInfo->msgType = TDMT_VND_SUBMIT;	×
146	pMsgSendInfo->fp = inserterCallback;	×
147
148	return asyncSendMsgToServer(pTransporter, pEpset, NULL, pMsgSendInfo);	×
149	}
150
151	static int32_t submitReqToMsg(int32_t vgId, SSubmitReq2* pReq, void** pData, int32_t* pLen) {	×
152	int32_t code = TSDB_CODE_SUCCESS;	×
153	int32_t len = 0;	×
154	void* pBuf = NULL;	×
155	tEncodeSize(tEncodeSubmitReq, pReq, len, code);	×
156	if (TSDB_CODE_SUCCESS == code) {	×
157	SEncoder encoder;
158	len += sizeof(SSubmitReq2Msg);	×
159	pBuf = taosMemoryMalloc(len);	×
160	if (NULL == pBuf) {	×
161	return terrno;	×
162	}
163	((SSubmitReq2Msg*)pBuf)->header.vgId = htonl(vgId);	×
164	((SSubmitReq2Msg*)pBuf)->header.contLen = htonl(len);	×
165	((SSubmitReq2Msg*)pBuf)->version = htobe64(1);	×
166	tEncoderInit(&encoder, POINTER_SHIFT(pBuf, sizeof(SSubmitReq2Msg)), len - sizeof(SSubmitReq2Msg));	×
167	code = tEncodeSubmitReq(&encoder, pReq);	×
168	tEncoderClear(&encoder);	×
169	}
170
171	if (TSDB_CODE_SUCCESS == code) {	×
172	*pData = pBuf;	×
173	*pLen = len;	×
174	} else {
175	taosMemoryFree(pBuf);	×
176	}
177
178	return code;	×
179	}
180
181	int32_t buildSubmitReqFromBlock(SDataInserterHandle* pInserter, SSubmitReq2** ppReq, const SSDataBlock* pDataBlock,	×
182	const STSchema* pTSchema, int64_t uid, int32_t vgId, tb_uid_t suid) {
183	SSubmitReq2* pReq = *ppReq;	×
184	SArray* pVals = NULL;	×
185	int32_t numOfBlks = 0;	×
186
187	terrno = TSDB_CODE_SUCCESS;	×
188
189	if (NULL == pReq) {	×
190	if (!(pReq = taosMemoryCalloc(1, sizeof(SSubmitReq2)))) {	×
191	goto _end;	×
192	}
193
194	if (!(pReq->aSubmitTbData = taosArrayInit(1, sizeof(SSubmitTbData)))) {	×
195	goto _end;	×
196	}
197	}
198
199	int32_t colNum = taosArrayGetSize(pDataBlock->pDataBlock);	×
200	int32_t rows = pDataBlock->info.rows;	×
201
202	SSubmitTbData tbData = {0};	×
203	if (!(tbData.aRowP = taosArrayInit(rows, sizeof(SRow*)))) {	×
204	goto _end;	×
205	}
206	tbData.suid = suid;	×
207	tbData.uid = uid;	×
208	tbData.sver = pTSchema->version;	×
209
210	if (!pVals && !(pVals = taosArrayInit(colNum, sizeof(SColVal)))) {	×
211	taosArrayDestroy(tbData.aRowP);	×
212	goto _end;	×
213	}
214
215	int64_t lastTs = TSKEY_MIN;	×
216	bool needSortMerge = false;	×
217
218	for (int32_t j = 0; j < rows; ++j) { // iterate by row	×
219	taosArrayClear(pVals);	×
220
221	int32_t offset = 0;	×
222	for (int32_t k = 0; k < pTSchema->numOfCols; ++k) { // iterate by column	×
223	int16_t colIdx = k;	×
224	const STColumn* pCol = &pTSchema->columns[k];	×
225	if (!pInserter->fullOrderColList) {	×
226	int16_t* slotId = taosHashGet(pInserter->pCols, &pCol->colId, sizeof(pCol->colId));	×
227	if (NULL == slotId) {	×
228	continue;	×
229	}
230
231	colIdx = *slotId;	×
232	}
233
234	SColumnInfoData* pColInfoData = taosArrayGet(pDataBlock->pDataBlock, colIdx);	×
235	if (NULL == pColInfoData) {	×
236	terrno = TSDB_CODE_QRY_EXECUTOR_INTERNAL_ERROR;	×
237	goto _end;	×
238	}
239	void* var = POINTER_SHIFT(pColInfoData->pData, j * pColInfoData->info.bytes);	×
240
241	switch (pColInfoData->info.type) {	×
242	case TSDB_DATA_TYPE_NCHAR:	×
243	case TSDB_DATA_TYPE_VARBINARY:
244	case TSDB_DATA_TYPE_VARCHAR: { // TSDB_DATA_TYPE_BINARY
245	if (pColInfoData->info.type != pCol->type) {	×
246	qError("column:%d type:%d in block dismatch with schema col:%d type:%d", colIdx, pColInfoData->info.type, k,	×
247	pCol->type);
248	terrno = TSDB_CODE_QRY_EXECUTOR_INTERNAL_ERROR;	×
249	goto _end;	×
250	}
251	if (colDataIsNull_s(pColInfoData, j)) {	×
252	SColVal cv = COL_VAL_NULL(pCol->colId, pCol->type);	×
253	if (NULL == taosArrayPush(pVals, &cv)) {	×
254	goto _end;	×
255	}
256	} else {
257	void* data = colDataGetVarData(pColInfoData, j);	×
258	SValue sv = (SValue){	×
259	.type = pCol->type, .nData = varDataLen(data), .pData = varDataVal(data)}; // address copy, no value	×
260	SColVal cv = COL_VAL_VALUE(pCol->colId, sv);	×
261	if (NULL == taosArrayPush(pVals, &cv)) {	×
262	goto _end;	×
263	}
264	}
265	break;	×
266	}
267	case TSDB_DATA_TYPE_BLOB:	×
268	case TSDB_DATA_TYPE_JSON:
269	case TSDB_DATA_TYPE_MEDIUMBLOB:
270	qError("the column type %" PRIi16 " is defined but not implemented yet", pColInfoData->info.type);	×
271	terrno = TSDB_CODE_APP_ERROR;	×
272	goto _end;	×
273	break;
274	default:	×
275	if (pColInfoData->info.type < TSDB_DATA_TYPE_MAX && pColInfoData->info.type > TSDB_DATA_TYPE_NULL) {	×
276	if (colDataIsNull_s(pColInfoData, j)) {	×
277	if (PRIMARYKEY_TIMESTAMP_COL_ID == pCol->colId) {	×
278	qError("Primary timestamp column should not be null");	×
279	terrno = TSDB_CODE_PAR_INCORRECT_TIMESTAMP_VAL;	×
280	goto _end;	×
281	}
282
283	SColVal cv = COL_VAL_NULL(pCol->colId, pCol->type); // should use pCol->type	×
284	if (NULL == taosArrayPush(pVals, &cv)) {	×
285	goto _end;	×
286	}
287	} else {
288	if (PRIMARYKEY_TIMESTAMP_COL_ID == pCol->colId && !needSortMerge) {	×
289	if ((int64_t)var <= lastTs) {	×
290	needSortMerge = true;	×
291	} else {
292	lastTs = (int64_t)var;	×
293	}
294	}
295
296	SValue sv = {.type = pCol->type};	×
297	valueSetDatum(&sv, sv.type, var, tDataTypes[pCol->type].bytes);	×
298	SColVal cv = COL_VAL_VALUE(pCol->colId, sv);	×
299	if (NULL == taosArrayPush(pVals, &cv)) {	×
300	goto _end;	×
301	}
302	}
303	} else {
304	uError("the column type %" PRIi16 " is undefined\n", pColInfoData->info.type);	×
305	terrno = TSDB_CODE_APP_ERROR;	×
306	goto _end;	×
307	}
308	break;	×
309	}
310	}
311
312	SRow* pRow = NULL;	×
313	if ((terrno = tRowBuild(pVals, pTSchema, &pRow)) < 0) {	×
314	tDestroySubmitTbData(&tbData, TSDB_MSG_FLG_ENCODE);	×
315	goto _end;	×
316	}
317	if (NULL == taosArrayPush(tbData.aRowP, &pRow)) {	×
318	goto _end;	×
319	}
320	}
321
322	if (needSortMerge) {	×
323	if ((tRowSort(tbData.aRowP) != TSDB_CODE_SUCCESS) \|\|	×
324	(terrno = tRowMerge(tbData.aRowP, (STSchema*)pTSchema, 0)) != 0) {	×
325	goto _end;	×
326	}
327	}
328
329	if (NULL == taosArrayPush(pReq->aSubmitTbData, &tbData)) {	×
330	goto _end;	×
331	}
332
333	_end:	×
334
335	taosArrayDestroy(pVals);	×
336	if (terrno != 0) {	×
337	*ppReq = NULL;	×
338	if (pReq) {	×
339	tDestroySubmitReq(pReq, TSDB_MSG_FLG_ENCODE);	×
340	taosMemoryFree(pReq);	×
341	}
342
343	return terrno;	×
344	}
345	*ppReq = pReq;	×
346
347	return TSDB_CODE_SUCCESS;	×
348	}
349
350	int32_t dataBlocksToSubmitReq(SDataInserterHandle* pInserter, void** pMsg, int32_t* msgLen) {	×
351	const SArray* pBlocks = pInserter->pDataBlocks;	×
352	const STSchema* pTSchema = pInserter->pSchema;	×
353	int64_t uid = pInserter->pNode->tableId;	×
354	int64_t suid = pInserter->pNode->stableId;	×
355	int32_t vgId = pInserter->pNode->vgId;	×
356	int32_t sz = taosArrayGetSize(pBlocks);	×
357	int32_t code = 0;	×
358	SSubmitReq2* pReq = NULL;	×
359
360	for (int32_t i = 0; i < sz; i++) {	×
361	SSDataBlock* pDataBlock = taosArrayGetP(pBlocks, i);	×
362	if (NULL == pDataBlock) {	×
363	return TSDB_CODE_QRY_EXECUTOR_INTERNAL_ERROR;	×
364	}
365	code = buildSubmitReqFromBlock(pInserter, &pReq, pDataBlock, pTSchema, uid, vgId, suid);	×
366	if (code) {	×
367	if (pReq) {	×
368	tDestroySubmitReq(pReq, TSDB_MSG_FLG_ENCODE);	×
369	taosMemoryFree(pReq);	×
370	}
371
372	return code;	×
373	}
374	}
375
376	code = submitReqToMsg(vgId, pReq, pMsg, msgLen);	×
377	tDestroySubmitReq(pReq, TSDB_MSG_FLG_ENCODE);	×
378	taosMemoryFree(pReq);	×
379
380	return code;	×
381	}
382
383	static int32_t putDataBlock(SDataSinkHandle* pHandle, const SInputData* pInput, bool* pContinue) {	×
384	SDataInserterHandle* pInserter = (SDataInserterHandle*)pHandle;	×
385	if (!pInserter->explain) {	×
386	if (NULL == taosArrayPush(pInserter->pDataBlocks, &pInput->pData)) {	×
387	return terrno;	×
388	}
389	void* pMsg = NULL;	×
390	int32_t msgLen = 0;	×
391	int32_t code = dataBlocksToSubmitReq(pInserter, &pMsg, &msgLen);	×
392	if (code) {	×
393	return code;	×
394	}
395
396	taosArrayClear(pInserter->pDataBlocks);	×
397
398	code = sendSubmitRequest(pInserter, pMsg, msgLen, pInserter->pParam->readHandle->pMsgCb->clientRpc,	×
399	&pInserter->pNode->epSet);	×
400	if (code) {	×
401	return code;	×
402	}
403
404	QRY_ERR_RET(tsem_wait(&pInserter->ready));	×
405
406	if (pInserter->submitRes.code) {	×
407	return pInserter->submitRes.code;	×
408	}
409	}
410
411	*pContinue = true;	×
412
413	return TSDB_CODE_SUCCESS;	×
414	}
415
416	static void endPut(struct SDataSinkHandle* pHandle, uint64_t useconds) {	×
417	SDataInserterHandle* pInserter = (SDataInserterHandle*)pHandle;	×
418	(void)taosThreadMutexLock(&pInserter->mutex);	×
419	pInserter->queryEnd = true;	×
420	pInserter->useconds = useconds;	×
421	(void)taosThreadMutexUnlock(&pInserter->mutex);	×
422	}	×
423
424	static void getDataLength(SDataSinkHandle* pHandle, int64_t* pLen, int64_t* pRawLen, bool* pQueryEnd) {	×
425	SDataInserterHandle* pDispatcher = (SDataInserterHandle*)pHandle;	×
426	*pLen = pDispatcher->submitRes.affectedRows;	×
427	qDebug("got total affectedRows %" PRId64, *pLen);	×
428	}	×
429
430	static int32_t destroyDataSinker(SDataSinkHandle* pHandle) {	×
431	SDataInserterHandle* pInserter = (SDataInserterHandle*)pHandle;	×
432	(void)atomic_sub_fetch_64(&gDataSinkStat.cachedSize, pInserter->cachedSize);	×
433	taosArrayDestroy(pInserter->pDataBlocks);	×
434	taosMemoryFree(pInserter->pSchema);	×
435	taosMemoryFree(pInserter->pParam);	×
436	taosHashCleanup(pInserter->pCols);	×
437	nodesDestroyNode((SNode *)pInserter->pNode);	×
438	pInserter->pNode = NULL;	×
439
440	(void)taosThreadMutexDestroy(&pInserter->mutex);	×
441
442	taosMemoryFree(pInserter->pManager);	×
443	return TSDB_CODE_SUCCESS;	×
444	}
445
446	static int32_t getCacheSize(struct SDataSinkHandle* pHandle, uint64_t* size) {	×
447	SDataInserterHandle* pDispatcher = (SDataInserterHandle*)pHandle;	×
448
449	*size = atomic_load_64(&pDispatcher->cachedSize);	×
450	return TSDB_CODE_SUCCESS;	×
451	}
452
453	static int32_t getSinkFlags(struct SDataSinkHandle* pHandle, uint64_t* pFlags) {	×
454	SDataInserterHandle* pDispatcher = (SDataInserterHandle*)pHandle;	×
455
456	*pFlags = atomic_load_64(&pDispatcher->flags);	×
457	return TSDB_CODE_SUCCESS;	×
458	}
459
460	int32_t createDataInserter(SDataSinkManager* pManager, SDataSinkNode** ppDataSink, DataSinkHandle* pHandle,	×
461	void* pParam) {
462	SDataSinkNode* pDataSink = *ppDataSink;	×
463	SDataInserterHandle* inserter = taosMemoryCalloc(1, sizeof(SDataInserterHandle));	×
464	if (NULL == inserter) {	×
465	taosMemoryFree(pParam);	×
466	goto _return;	×
467	}
468
469	SQueryInserterNode* pInserterNode = (SQueryInserterNode*)pDataSink;	×
470	inserter->sink.fPut = putDataBlock;	×
471	inserter->sink.fEndPut = endPut;	×
472	inserter->sink.fGetLen = getDataLength;	×
473	inserter->sink.fGetData = NULL;	×
474	inserter->sink.fDestroy = destroyDataSinker;	×
475	inserter->sink.fGetCacheSize = getCacheSize;	×
476	inserter->sink.fGetFlags = getSinkFlags;	×
477	inserter->pManager = pManager;	×
478	inserter->pNode = pInserterNode;	×
479	inserter->pParam = pParam;	×
480	inserter->status = DS_BUF_EMPTY;	×
481	inserter->queryEnd = false;	×
482	inserter->explain = pInserterNode->explain;	×
483	*ppDataSink = NULL;	×
484
485	int64_t suid = 0;	×
486	int32_t code = pManager->pAPI->metaFn.getTableSchema(inserter->pParam->readHandle->vnode, pInserterNode->tableId,	×
487	&inserter->pSchema, &suid);
488	if (code) {	×
489	terrno = code;	×
490	goto _return;	×
491	}
492
493	if (pInserterNode->stableId != suid) {	×
494	terrno = TSDB_CODE_TDB_INVALID_TABLE_ID;	×
495	goto _return;	×
496	}
497
498	inserter->pDataBlocks = taosArrayInit(1, POINTER_BYTES);	×
499	if (NULL == inserter->pDataBlocks) {	×
500	goto _return;	×
501	}
502	QRY_ERR_JRET(taosThreadMutexInit(&inserter->mutex, NULL));	×
503
504	inserter->fullOrderColList = pInserterNode->pCols->length == inserter->pSchema->numOfCols;	×
505
506	inserter->pCols = taosHashInit(pInserterNode->pCols->length, taosGetDefaultHashFunction(TSDB_DATA_TYPE_SMALLINT),	×
507	false, HASH_NO_LOCK);
508	if (NULL == inserter->pCols) {	×
509	goto _return;	×
510	}
511
512	SNode* pNode = NULL;	×
513	int32_t i = 0;	×
514	FOREACH(pNode, pInserterNode->pCols) {	×
515	SColumnNode* pCol = (SColumnNode*)pNode;	×
516	QRY_ERR_JRET(taosHashPut(inserter->pCols, &pCol->colId, sizeof(pCol->colId), &pCol->slotId, sizeof(pCol->slotId)));	×
517	if (inserter->fullOrderColList && pCol->colId != inserter->pSchema->columns[i].colId) {	×
518	inserter->fullOrderColList = false;	×
519	}
520	++i;	×
521	}
522
523	QRY_ERR_JRET(tsem_init(&inserter->ready, 0, 0));	×
524
525	*pHandle = inserter;	×
526	return TSDB_CODE_SUCCESS;	×
527
528	_return:	×
529
530	if (inserter) {	×
531	(void)destroyDataSinker((SDataSinkHandle*)inserter);	×
532	taosMemoryFree(inserter);	×
533	} else {
534	taosMemoryFree(pManager);	×
535	}
536
537	nodesDestroyNode((SNode )ppDataSink);	×
538	*ppDataSink = NULL;	×
539
540	return terrno;	×
541	}

taosdata / TDengine / #3814

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