#4113

Committed 17 May 2025 06:43AM UTC coverage: 62.054% (-0.8%) from 62.857%

Build # #4113

Build Type

push

travis-ci

Committed by

web-flow

Commit Message

Merge pull request #31115 from taosdata/merge/mainto3.0

merge: from main to 3.0 branch

Run Details

154737 of 318088 branches covered (48.65%)

Branch coverage included in aggregate %.

175 of 225 new or added lines in 20 files covered. (77.78%)

5853 existing lines in 216 files now uncovered.

239453 of 317147 relevant lines covered (75.5%)

15121865.73 hits per line

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

74.11

/source/libs/parser/src/parInsertSml.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 "parInsertUtil.h"
#include "parInt.h"
#include "parToken.h"
#include "ttime.h"

int32_t qCreateSName(SName* pName, const char* pTableName, int32_t acctId, char* dbName, char* msgBuf,
                     int32_t msgBufLen) {
  SMsgBuf    msg = {.buf = msgBuf, .len = msgBufLen};
  SToken  sToken = {0};
  int      code  = TSDB_CODE_SUCCESS;
  int32_t  lino  = 0;

  NEXT_TOKEN(pTableName, sToken);
  TSDB_CHECK_CONDITION(sToken.n != 0, code, lino, end, TSDB_CODE_TSC_INVALID_OPERATION);
  code = insCreateSName(pName, &sToken, acctId, dbName, &msg);
  TSDB_CHECK_CODE(code, lino, end);
  NEXT_TOKEN(pTableName, sToken);
  TSDB_CHECK_CONDITION(sToken.n <= 0, code, lino, end, TSDB_CODE_TSC_INVALID_OPERATION);

end:
  if (code != 0){
    uError("%s failed at %d since %s", __func__, lino, tstrerror(code));
  }
  return TSDB_CODE_SUCCESS;
}

static int32_t smlBoundColumnData(SArray* cols, SBoundColInfo* pBoundInfo, SSchema* pSchema, bool isTag) {
  int      code = TSDB_CODE_SUCCESS;
  int32_t  lino = 0;
  bool* pUseCols = taosMemoryCalloc(pBoundInfo->numOfCols, sizeof(bool));
  TSDB_CHECK_NULL(pUseCols, code, lino, end, terrno);
  pBoundInfo->numOfBound = 0;
  int16_t lastColIdx = -1;  // last column found

  for (int i = 0; i < taosArrayGetSize(cols); ++i) {
    SSmlKv*  kv = taosArrayGet(cols, i);
    SToken   sToken = {.n = kv->keyLen, .z = (char*)kv->key};
    col_id_t t = lastColIdx + 1;
    col_id_t index = ((t == 0 && !isTag) ? 0 : insFindCol(&sToken, t, pBoundInfo->numOfCols, pSchema));
    uTrace("SML, index:%d, t:%d, ncols:%d", index, t, pBoundInfo->numOfCols);
    if (index < 0 && t > 0) {
      index = insFindCol(&sToken, 0, t, pSchema);
    }

    TSDB_CHECK_CONDITION(index >= 0, code, lino, end, TSDB_CODE_SML_INVALID_DATA);
    TSDB_CHECK_CONDITION(!pUseCols[index], code, lino, end, TSDB_CODE_SML_INVALID_DATA);

    lastColIdx = index;
    pUseCols[index] = true;
    pBoundInfo->pColIndex[pBoundInfo->numOfBound] = index;
    ++pBoundInfo->numOfBound;
  }

end:
  if (code != 0){
    uError("%s failed at %d since %s", __func__, lino, tstrerror(code));
  }
  taosMemoryFree(pUseCols);
  return code;
}

static int32_t smlMbsToUcs4(const char* mbs, size_t mbsLen, void** result, int32_t* resultLen, int32_t maxLen, void* charsetCxt){
  int      code = TSDB_CODE_SUCCESS;
  void*    pUcs4 = NULL;
  int32_t  lino = 0;
  pUcs4 = taosMemoryCalloc(1, maxLen);
  TSDB_CHECK_NULL(pUcs4, code, lino, end, terrno);
  TSDB_CHECK_CONDITION(taosMbsToUcs4(mbs, mbsLen, (TdUcs4*)pUcs4, maxLen, resultLen, charsetCxt), code, lino, end, terrno);
  *result = pUcs4;
  pUcs4 = NULL;

end:
  if (code != 0){
    uError("%s failed at %d since %s", __func__, lino, tstrerror(code));
  }
  taosMemoryFree(pUcs4);
  return code;
}
/**
 * @brief No json tag for schemaless
 *
 * @param cols
 * @param tags
 * @param pSchema
 * @param ppTag
 * @param msg
 * @return int32_t
 */
static int32_t smlBuildTagRow(SArray* cols, SBoundColInfo* tags, SSchema* pSchema, STag** ppTag, SArray** tagName,
                              SMsgBuf* msg, void* charsetCxt) {
  int      code = TSDB_CODE_SUCCESS;
  int32_t  lino = 0;
  SArray*  pTagArray = taosArrayInit(tags->numOfBound, sizeof(STagVal));
  TSDB_CHECK_NULL(pTagArray, code, lino, end, terrno);
  *tagName = taosArrayInit(8, TSDB_COL_NAME_LEN);
  TSDB_CHECK_NULL(*tagName, code, lino, end, terrno);

  for (int i = 0; i < tags->numOfBound; ++i) {
    SSchema* pTagSchema = &pSchema[tags->pColIndex[i]];
    SSmlKv*  kv = taosArrayGet(cols, i);
    TSDB_CHECK_NULL(kv, code, lino, end, terrno);
    bool cond = (kv->keyLen == strlen(pTagSchema->name) && memcmp(kv->key, pTagSchema->name, kv->keyLen) == 0 && kv->type == pTagSchema->type);
    TSDB_CHECK_CONDITION(cond, code, lino, end, TSDB_CODE_SML_INVALID_DATA);
    TSDB_CHECK_NULL(taosArrayPush(*tagName, pTagSchema->name), code, lino, end, terrno);
    STagVal val = {.cid = pTagSchema->colId, .type = pTagSchema->type};
    if (pTagSchema->type == TSDB_DATA_TYPE_BINARY || pTagSchema->type == TSDB_DATA_TYPE_VARBINARY ||
        pTagSchema->type == TSDB_DATA_TYPE_GEOMETRY) {
      val.pData = (uint8_t*)kv->value;
      val.nData = kv->length;
    } else if (pTagSchema->type == TSDB_DATA_TYPE_NCHAR) {
      code = smlMbsToUcs4(kv->value, kv->length, (void**)&val.pData, (int32_t*)&val.nData, kv->length * TSDB_NCHAR_SIZE, charsetCxt);
      TSDB_CHECK_CODE(code, lino, end);
    } else {
      (void)memcpy(&val.i64, &(kv->value), kv->length);
    }
    TSDB_CHECK_NULL(taosArrayPush(pTagArray, &val), code, lino, end, terrno);
  }
  code = tTagNew(pTagArray, 1, false, ppTag);

end:
  if (code != 0){
    uError("%s failed at %d since %s", __func__, lino, tstrerror(code));
  }
  for (int i = 0; i < taosArrayGetSize(pTagArray); ++i) {
    STagVal* p = (STagVal*)taosArrayGet(pTagArray, i);
    if (p->type == TSDB_DATA_TYPE_NCHAR) {
      taosMemoryFree(p->pData);
    }
  }
  taosArrayDestroy(pTagArray);
  return code;
}

int32_t smlInitTableDataCtx(SQuery* query, STableMeta* pTableMeta, STableDataCxt** cxt) {
  int      ret = TSDB_CODE_SUCCESS;
  int32_t  lino = 0;
  SVCreateTbReq* pCreateTbReq = NULL;
  ret = insGetTableDataCxt(((SVnodeModifyOpStmt*)(query->pRoot))->pTableBlockHashObj, &pTableMeta->uid,
                                          sizeof(pTableMeta->uid), pTableMeta, &pCreateTbReq, cxt, false, false);
  TSDB_CHECK_CODE(ret, lino, end);
  ret = initTableColSubmitData(*cxt);
  TSDB_CHECK_CODE(ret, lino, end);

end:
  if (ret != 0){
    uError("%s failed at %d since %s", __func__, lino, tstrerror(ret));
  }
  return ret;
}

void clearColValArraySml(SArray* pCols) {
  int32_t num = taosArrayGetSize(pCols);
  for (int32_t i = 0; i < num; ++i) {
    SColVal* pCol = taosArrayGet(pCols, i);
    if (TSDB_DATA_TYPE_NCHAR == pCol->value.type || TSDB_DATA_TYPE_GEOMETRY == pCol->value.type ||
        TSDB_DATA_TYPE_VARBINARY == pCol->value.type) {
      taosMemoryFreeClear(pCol->value.pData);
    }
    pCol->flag = CV_FLAG_NONE;
    pCol->value.val = 0;
  }
}

int32_t smlBuildRow(STableDataCxt* pTableCxt) {
  int      ret = TSDB_CODE_SUCCESS;
  int32_t  lino = 0;
  SRow**   pRow = taosArrayReserve(pTableCxt->pData->aRowP, 1);
  TSDB_CHECK_NULL(pRow, ret, lino, end, terrno);
  ret = tRowBuild(pTableCxt->pValues, pTableCxt->pSchema, pRow);
  TSDB_CHECK_CODE(ret, lino, end);
  SRowKey key;
  tRowGetKey(*pRow, &key);
  insCheckTableDataOrder(pTableCxt, &key);
end:
  if (ret != 0){
    uError("%s failed at %d since %s", __func__, lino, tstrerror(ret));
  }
  return ret;
}

int32_t smlBuildCol(STableDataCxt* pTableCxt, SSchema* schema, void* data, int32_t index, void* charsetCxt) {
  int      ret = TSDB_CODE_SUCCESS;
  int32_t  lino = 0;
  SSchema* pColSchema = schema + index;
  SColVal* pVal = taosArrayGet(pTableCxt->pValues, index);
  TSDB_CHECK_NULL(pVal, ret, lino, end, TSDB_CODE_SUCCESS);
  SSmlKv*  kv = (SSmlKv*)data;
  if (kv->keyLen != strlen(pColSchema->name) || memcmp(kv->key, pColSchema->name, kv->keyLen) != 0 ||
      kv->type != pColSchema->type) {
    ret = TSDB_CODE_SML_INVALID_DATA;
    char* tmp = taosMemoryCalloc(kv->keyLen + 1, 1);
    TSDB_CHECK_NULL(tmp, ret, lino, end, terrno);
    (void)memcpy(tmp, kv->key, kv->keyLen);
    uInfo("SML data(name:%s type:%s) is not same like the db data(name:%s type:%s)", tmp, tDataTypes[kv->type].name,
          pColSchema->name, tDataTypes[pColSchema->type].name);
    taosMemoryFree(tmp);
    goto end;
  }
  if (kv->type == TSDB_DATA_TYPE_NCHAR) {
    ret = smlMbsToUcs4(kv->value, kv->length, (void**)&pVal->value.pData, &pVal->value.nData, pColSchema->bytes - VARSTR_HEADER_SIZE, charsetCxt);
    TSDB_CHECK_CODE(ret, lino, end);
  } else if (kv->type == TSDB_DATA_TYPE_BINARY) {
    pVal->value.nData = kv->length;
    pVal->value.pData = (uint8_t*)kv->value;
  } else if (kv->type == TSDB_DATA_TYPE_GEOMETRY || kv->type == TSDB_DATA_TYPE_VARBINARY) {
    pVal->value.nData = kv->length;
    pVal->value.pData = taosMemoryMalloc(kv->length);
    TSDB_CHECK_NULL(pVal->value.pData, ret, lino, end, terrno);

    (void)memcpy(pVal->value.pData, (uint8_t*)kv->value, kv->length);
  } else {
    valueSetDatum(&pVal->value, kv->type, &(kv->value), kv->length);
  }
  pVal->flag = CV_FLAG_VALUE;

end:
  if (ret != 0){
    uError("%s failed at %d since %s", __func__, lino, tstrerror(ret));
  }
  return ret;
}

int32_t smlBindData(SQuery* query, bool dataFormat, SArray* tags, SArray* colsSchema, SArray* cols,
                    STableMeta* pTableMeta, char* tableName, const char* sTableName, int32_t sTableNameLen, int32_t ttl,
                    char* msgBuf, int32_t msgBufLen, void* charsetCxt) {
  int32_t lino = 0;
  int32_t ret = 0;
  SMsgBuf pBuf = {.buf = msgBuf, .len = msgBufLen};

  SSchema*       pTagsSchema = getTableTagSchema(pTableMeta);
  SBoundColInfo  bindTags = {0};
  SVCreateTbReq* pCreateTblReq = NULL;
  SArray*        tagName = NULL;

  ret = insInitBoundColsInfo(getNumOfTags(pTableMeta), &bindTags);
  TSDB_CHECK_CODE(ret, lino, end);

  ret = smlBoundColumnData(tags, &bindTags, pTagsSchema, true);
  TSDB_CHECK_CODE(ret, lino, end);

  STag* pTag = NULL;
  ret = smlBuildTagRow(tags, &bindTags, pTagsSchema, &pTag, &tagName, &pBuf, charsetCxt);
  TSDB_CHECK_CODE(ret, lino, end);

  pCreateTblReq = taosMemoryCalloc(1, sizeof(SVCreateTbReq));
  TSDB_CHECK_NULL(pCreateTblReq, ret, lino, end, terrno);

  ret = insBuildCreateTbReq(pCreateTblReq, tableName, pTag, pTableMeta->suid, NULL, tagName,
                            pTableMeta->tableInfo.numOfTags, ttl);
  TSDB_CHECK_CODE(ret, lino, end);

  pCreateTblReq->ctb.stbName = taosMemoryCalloc(1, sTableNameLen + 1);
  TSDB_CHECK_NULL(pCreateTblReq->ctb.stbName, ret, lino, end, terrno);

  (void)memcpy(pCreateTblReq->ctb.stbName, sTableName, sTableNameLen);

  if (dataFormat) {
    STableDataCxt** pTableCxt = (STableDataCxt**)taosHashGet(((SVnodeModifyOpStmt*)(query->pRoot))->pTableBlockHashObj,
                                                             &pTableMeta->uid, sizeof(pTableMeta->uid));
    TSDB_CHECK_NULL(pTableCxt, ret, lino, end, TSDB_CODE_TSC_INVALID_OPERATION);
    (*pTableCxt)->pData->flags |= SUBMIT_REQ_AUTO_CREATE_TABLE;
    (*pTableCxt)->pData->pCreateTbReq = pCreateTblReq;
    (*pTableCxt)->pMeta->uid = pTableMeta->uid;
    (*pTableCxt)->pMeta->vgId = pTableMeta->vgId;
    pCreateTblReq = NULL;
    goto end;
  }

  STableDataCxt* pTableCxt = NULL;
  ret = insGetTableDataCxt(((SVnodeModifyOpStmt*)(query->pRoot))->pTableBlockHashObj, &pTableMeta->uid,
                           sizeof(pTableMeta->uid), pTableMeta, &pCreateTblReq, &pTableCxt, false, false);
  TSDB_CHECK_CODE(ret, lino, end);

  SSchema* pSchema = getTableColumnSchema(pTableMeta);
  ret = smlBoundColumnData(colsSchema, &pTableCxt->boundColsInfo, pSchema, false);
  TSDB_CHECK_CODE(ret, lino, end);

  ret = initTableColSubmitData(pTableCxt);
  TSDB_CHECK_CODE(ret, lino, end);

  int32_t rowNum = taosArrayGetSize(cols);
  TSDB_CHECK_CONDITION(rowNum > 0, ret, lino, end, TSDB_CODE_TSC_INVALID_OPERATION);

  for (int32_t r = 0; r < rowNum; ++r) {
    void* rowData = taosArrayGetP(cols, r);
    TSDB_CHECK_NULL(rowData, ret, lino, end, terrno);

    // 1. set the parsed value from sql string
    for (int c = 0; c < pTableCxt->boundColsInfo.numOfBound; ++c) {
      SSchema* pColSchema = &pSchema[pTableCxt->boundColsInfo.pColIndex[c]];
      SColVal* pVal = taosArrayGet(pTableCxt->pValues, pTableCxt->boundColsInfo.pColIndex[c]);
      TSDB_CHECK_NULL(pVal, ret, lino, end, terrno);
      void**   p = taosHashGet(rowData, pColSchema->name, strlen(pColSchema->name));
      if (p == NULL) {
        continue;
      }
      SSmlKv* kv = *(SSmlKv**)p;
      TSDB_CHECK_CONDITION(kv->type == pColSchema->type, ret, lino, end, TSDB_CODE_TSC_INVALID_OPERATION);

      if (pColSchema->type == TSDB_DATA_TYPE_TIMESTAMP) {
        kv->i = convertTimePrecision(kv->i, TSDB_TIME_PRECISION_NANO, pTableMeta->tableInfo.precision);
      }
      if (kv->type == TSDB_DATA_TYPE_NCHAR) {
        ret = smlMbsToUcs4(kv->value, kv->length, (void**)&pVal->value.pData, (int32_t*)&pVal->value.nData, pColSchema->bytes - VARSTR_HEADER_SIZE, charsetCxt);
        TSDB_CHECK_CODE(ret, lino, end);
      } else if (kv->type == TSDB_DATA_TYPE_BINARY) {
        pVal->value.nData = kv->length;
        pVal->value.pData = (uint8_t*)kv->value;
      } else if (kv->type == TSDB_DATA_TYPE_GEOMETRY || kv->type == TSDB_DATA_TYPE_VARBINARY) {
        pVal->value.nData = kv->length;
        pVal->value.pData = taosMemoryMalloc(kv->length);
        TSDB_CHECK_NULL(pVal->value.pData, ret, lino, end, terrno);
        (void)memcpy(pVal->value.pData, (uint8_t*)kv->value, kv->length);
      } else {
        valueSetDatum(&pVal->value, kv->type, &(kv->value), kv->length);
      }
      pVal->flag = CV_FLAG_VALUE;
    }

    SRow** pRow = taosArrayReserve(pTableCxt->pData->aRowP, 1);
    TSDB_CHECK_NULL(pRow, ret, lino, end, terrno);
    ret = tRowBuild(pTableCxt->pValues, pTableCxt->pSchema, pRow);
    TSDB_CHECK_CODE(ret, lino, end);
    SRowKey key = {0};
    tRowGetKey(*pRow, &key);
    insCheckTableDataOrder(pTableCxt, &key);
    clearColValArraySml(pTableCxt->pValues);
  }

end:
  if (ret != 0){
    uError("%s failed at %d since %s", __func__, lino, tstrerror(ret));
    ret = buildInvalidOperationMsg(&pBuf, tstrerror(ret));
  }
  insDestroyBoundColInfo(&bindTags);
  tdDestroySVCreateTbReq(pCreateTblReq);
  taosMemoryFree(pCreateTblReq);
  taosArrayDestroy(tagName);
  return ret;
}

int32_t smlInitHandle(SQuery** query) {
  int32_t lino = 0;
  int32_t code = 0;
  SQuery* pQuery = NULL;
  SVnodeModifyOpStmt* stmt = NULL;
  TSDB_CHECK_NULL(query, code, lino, end, TSDB_CODE_INVALID_PARA);

  *query = NULL;
  code = nodesMakeNode(QUERY_NODE_QUERY, (SNode**)&pQuery);
  TSDB_CHECK_CODE(code, lino, end);
  pQuery->execMode = QUERY_EXEC_MODE_SCHEDULE;
  pQuery->haveResultSet = false;
  pQuery->msgType = TDMT_VND_SUBMIT;
  code = nodesMakeNode(QUERY_NODE_VNODE_MODIFY_STMT, (SNode**)&stmt);
  TSDB_CHECK_CODE(code, lino, end);
  stmt->pTableBlockHashObj = taosHashInit(16, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BIGINT), true, HASH_NO_LOCK);
  TSDB_CHECK_NULL(stmt->pTableBlockHashObj, code, lino, end, terrno);
  stmt->freeHashFunc = insDestroyTableDataCxtHashMap;
  stmt->freeArrayFunc = insDestroyVgroupDataCxtList;

  pQuery->pRoot = (SNode*)stmt;
  *query = pQuery;
  return code;

end:
  if (code != 0) {
    uError("%s failed at %d since %s", __func__, lino, tstrerror(code));
  }
  nodesDestroyNode((SNode*)stmt);
  qDestroyQuery(pQuery);
  return code;
}

int32_t smlBuildOutput(SQuery* handle, SHashObj* pVgHash) {
  int32_t lino = 0;
  int32_t code = 0;

  SVnodeModifyOpStmt* pStmt = (SVnodeModifyOpStmt*)(handle)->pRoot;
  code = insMergeTableDataCxt(pStmt->pTableBlockHashObj, &pStmt->pVgDataBlocks, true);
  TSDB_CHECK_CODE(code, lino, end);
  code = insBuildVgDataBlocks(pVgHash, pStmt->pVgDataBlocks, &pStmt->pDataBlocks, false);
  TSDB_CHECK_CODE(code, lino, end);

end:
  if (code != 0) {
    uError("%s failed at %d since %s", __func__, lino, tstrerror(code));
  }
  return code;
}

int32_t smlBuildOutputRaw(SQuery* handle, SHashObj* pVgHash) {
  int32_t lino = 0;
  int32_t code = 0;

  SVnodeModifyOpStmt* pStmt = (SVnodeModifyOpStmt*)(handle)->pRoot;
  code = insBuildVgDataBlocks(pVgHash, pStmt->pVgDataBlocks, &pStmt->pDataBlocks, false);
  TSDB_CHECK_CODE(code, lino, end);

  end:
  if (code != 0) {
    uError("%s failed at %d since %s", __func__, lino, tstrerror(code));
  }
  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 "parInsertUtil.h"
17	#include "parInt.h"
18	#include "parToken.h"
19	#include "ttime.h"
20
21	int32_t qCreateSName(SName* pName, const char* pTableName, int32_t acctId, char* dbName, char* msgBuf,	52,389✔
22	int32_t msgBufLen) {
23	SMsgBuf msg = {.buf = msgBuf, .len = msgBufLen};	52,389✔
24	SToken sToken = {0};	52,389✔
25	int code = TSDB_CODE_SUCCESS;	52,389✔
26	int32_t lino = 0;	52,389✔
27
28	NEXT_TOKEN(pTableName, sToken);	52,389✔
29	TSDB_CHECK_CONDITION(sToken.n != 0, code, lino, end, TSDB_CODE_TSC_INVALID_OPERATION);	52,458!
30	code = insCreateSName(pName, &sToken, acctId, dbName, &msg);	52,458✔
31	TSDB_CHECK_CODE(code, lino, end);	52,434!
32	NEXT_TOKEN(pTableName, sToken);	52,434✔
33	TSDB_CHECK_CONDITION(sToken.n <= 0, code, lino, end, TSDB_CODE_TSC_INVALID_OPERATION);	52,450!
34
35	end:	52,450✔
36	if (code != 0){	52,450!
UNCOV 37	uError("%s failed at %d since %s", __func__, lino, tstrerror(code));	×
38	}
39	return TSDB_CODE_SUCCESS;	52,413✔
40	}
41
42	static int32_t smlBoundColumnData(SArray* cols, SBoundColInfo* pBoundInfo, SSchema* pSchema, bool isTag) {	3,360✔
43	int code = TSDB_CODE_SUCCESS;	3,360✔
44	int32_t lino = 0;	3,360✔
45	bool* pUseCols = taosMemoryCalloc(pBoundInfo->numOfCols, sizeof(bool));	3,360!
46	TSDB_CHECK_NULL(pUseCols, code, lino, end, terrno);	3,361!
47	pBoundInfo->numOfBound = 0;	3,361✔
48	int16_t lastColIdx = -1; // last column found	3,361✔
49
50	for (int i = 0; i < taosArrayGetSize(cols); ++i) {	28,496✔
51	SSmlKv* kv = taosArrayGet(cols, i);	25,133✔
52	SToken sToken = {.n = kv->keyLen, .z = (char*)kv->key};	25,133✔
53	col_id_t t = lastColIdx + 1;	25,133✔
54	col_id_t index = ((t == 0 && !isTag) ? 0 : insFindCol(&sToken, t, pBoundInfo->numOfCols, pSchema));	25,133✔
55	uTrace("SML, index:%d, t:%d, ncols:%d", index, t, pBoundInfo->numOfCols);	25,134!
56	if (index < 0 && t > 0) {	25,134!
57	index = insFindCol(&sToken, 0, t, pSchema);	68✔
58	}
59
60	TSDB_CHECK_CONDITION(index >= 0, code, lino, end, TSDB_CODE_SML_INVALID_DATA);	25,135!
61	TSDB_CHECK_CONDITION(!pUseCols[index], code, lino, end, TSDB_CODE_SML_INVALID_DATA);	25,135!
62
63	lastColIdx = index;	25,135✔
64	pUseCols[index] = true;	25,135✔
65	pBoundInfo->pColIndex[pBoundInfo->numOfBound] = index;	25,135✔
66	++pBoundInfo->numOfBound;	25,135✔
67	}
68
69	end:	3,360✔
70	if (code != 0){	3,360!
71	uError("%s failed at %d since %s", __func__, lino, tstrerror(code));	×
72	}
73	taosMemoryFree(pUseCols);	3,360!
74	return code;	3,361✔
75	}
76
77	static int32_t smlMbsToUcs4(const char* mbs, size_t mbsLen, void** result, int32_t* resultLen, int32_t maxLen, void* charsetCxt){	10,972✔
78	int code = TSDB_CODE_SUCCESS;	10,972✔
79	void* pUcs4 = NULL;	10,972✔
80	int32_t lino = 0;	10,972✔
81	pUcs4 = taosMemoryCalloc(1, maxLen);	10,972!
82	TSDB_CHECK_NULL(pUcs4, code, lino, end, terrno);	10,974!
83	TSDB_CHECK_CONDITION(taosMbsToUcs4(mbs, mbsLen, (TdUcs4*)pUcs4, maxLen, resultLen, charsetCxt), code, lino, end, terrno);	10,974!
84	*result = pUcs4;	10,975✔
85	pUcs4 = NULL;	10,975✔
86
87	end:	10,975✔
88	if (code != 0){	10,975!
89	uError("%s failed at %d since %s", __func__, lino, tstrerror(code));	×
90	}
91	taosMemoryFree(pUcs4);	10,975!
92	return code;	10,975✔
93	}
94	/**
95	* @brief No json tag for schemaless
96	*
97	* @param cols
98	* @param tags
99	* @param pSchema
100	* @param ppTag
101	* @param msg
102	* @return int32_t
103	*/
104	static int32_t smlBuildTagRow(SArray* cols, SBoundColInfo* tags, SSchema* pSchema, STag ppTag, SArray tagName,	1,983✔
105	SMsgBuf* msg, void* charsetCxt) {
106	int code = TSDB_CODE_SUCCESS;	1,983✔
107	int32_t lino = 0;	1,983✔
108	SArray* pTagArray = taosArrayInit(tags->numOfBound, sizeof(STagVal));	1,983✔
109	TSDB_CHECK_NULL(pTagArray, code, lino, end, terrno);	1,982!
110	*tagName = taosArrayInit(8, TSDB_COL_NAME_LEN);	1,982✔
111	TSDB_CHECK_NULL(*tagName, code, lino, end, terrno);	1,983!
112
113	for (int i = 0; i < tags->numOfBound; ++i) {	16,359✔
114	SSchema* pTagSchema = &pSchema[tags->pColIndex[i]];	14,374✔
115	SSmlKv* kv = taosArrayGet(cols, i);	14,374✔
116	TSDB_CHECK_NULL(kv, code, lino, end, terrno);	14,374!
117	bool cond = (kv->keyLen == strlen(pTagSchema->name) && memcmp(kv->key, pTagSchema->name, kv->keyLen) == 0 && kv->type == pTagSchema->type);	14,375!
118	TSDB_CHECK_CONDITION(cond, code, lino, end, TSDB_CODE_SML_INVALID_DATA);	14,375!
119	TSDB_CHECK_NULL(taosArrayPush(*tagName, pTagSchema->name), code, lino, end, terrno);	28,746!
120	STagVal val = {.cid = pTagSchema->colId, .type = pTagSchema->type};	14,371✔
121	if (pTagSchema->type == TSDB_DATA_TYPE_BINARY \|\| pTagSchema->type == TSDB_DATA_TYPE_VARBINARY \|\|	14,371!
122	pTagSchema->type == TSDB_DATA_TYPE_GEOMETRY) {	13,202!
123	val.pData = (uint8_t*)kv->value;	1,169✔
124	val.nData = kv->length;	1,169✔
125	} else if (pTagSchema->type == TSDB_DATA_TYPE_NCHAR) {	13,202✔
126	code = smlMbsToUcs4(kv->value, kv->length, (void*)&val.pData, (int32_t)&val.nData, kv->length * TSDB_NCHAR_SIZE, charsetCxt);	9,436✔
127	TSDB_CHECK_CODE(code, lino, end);	9,439!
128	} else {
129	(void)memcpy(&val.i64, &(kv->value), kv->length);	3,766✔
130	}
131	TSDB_CHECK_NULL(taosArrayPush(pTagArray, &val), code, lino, end, terrno);	14,376!
132	}
133	code = tTagNew(pTagArray, 1, false, ppTag);	1,985✔
134
135	end:	1,984✔
136	if (code != 0){	1,984!
137	uError("%s failed at %d since %s", __func__, lino, tstrerror(code));	×
138	}
139	for (int i = 0; i < taosArrayGetSize(pTagArray); ++i) {	16,350✔
140	STagVal* p = (STagVal*)taosArrayGet(pTagArray, i);	14,372✔
141	if (p->type == TSDB_DATA_TYPE_NCHAR) {	14,370✔
142	taosMemoryFree(p->pData);	9,433!
143	}
144	}
145	taosArrayDestroy(pTagArray);	1,980✔
146	return code;	1,984✔
147	}
148
149	int32_t smlInitTableDataCtx(SQuery* query, STableMeta* pTableMeta, STableDataCxt** cxt) {	727✔
150	int ret = TSDB_CODE_SUCCESS;	727✔
151	int32_t lino = 0;	727✔
152	SVCreateTbReq* pCreateTbReq = NULL;	727✔
153	ret = insGetTableDataCxt(((SVnodeModifyOpStmt*)(query->pRoot))->pTableBlockHashObj, &pTableMeta->uid,	727✔
154	sizeof(pTableMeta->uid), pTableMeta, &pCreateTbReq, cxt, false, false);
155	TSDB_CHECK_CODE(ret, lino, end);	729!
156	ret = initTableColSubmitData(*cxt);	729✔
157	TSDB_CHECK_CODE(ret, lino, end);	729!
158
159	end:	729✔
160	if (ret != 0){	729!
161	uError("%s failed at %d since %s", __func__, lino, tstrerror(ret));	×
162	}
163	return ret;	729✔
164	}
165
166	void clearColValArraySml(SArray* pCols) {	96,427✔
167	int32_t num = taosArrayGetSize(pCols);	96,427✔
168	for (int32_t i = 0; i < num; ++i) {	422,181✔
169	SColVal* pCol = taosArrayGet(pCols, i);	330,276✔
170	if (TSDB_DATA_TYPE_NCHAR == pCol->value.type \|\| TSDB_DATA_TYPE_GEOMETRY == pCol->value.type \|\|	326,637!
171	TSDB_DATA_TYPE_VARBINARY == pCol->value.type) {	325,119✔
172	taosMemoryFreeClear(pCol->value.pData);	1,536!
173	}
174	pCol->flag = CV_FLAG_NONE;	326,638✔
175	pCol->value.val = 0;	326,638✔
176	}
177	}	91,905✔
178
179	int32_t smlBuildRow(STableDataCxt* pTableCxt) {	5,077✔
180	int ret = TSDB_CODE_SUCCESS;	5,077✔
181	int32_t lino = 0;	5,077✔
182	SRow** pRow = taosArrayReserve(pTableCxt->pData->aRowP, 1);	5,077✔
183	TSDB_CHECK_NULL(pRow, ret, lino, end, terrno);	5,076!
184	ret = tRowBuild(pTableCxt->pValues, pTableCxt->pSchema, pRow);	5,076✔
185	TSDB_CHECK_CODE(ret, lino, end);	5,078!
186	SRowKey key;
187	tRowGetKey(*pRow, &key);	10,156!
188	insCheckTableDataOrder(pTableCxt, &key);	5,078✔
189	end:	5,074✔
190	if (ret != 0){	5,074!
191	uError("%s failed at %d since %s", __func__, lino, tstrerror(ret));	×
192	}
193	return ret;	5,075✔
194	}
195
196	int32_t smlBuildCol(STableDataCxt* pTableCxt, SSchema* schema, void* data, int32_t index, void* charsetCxt) {	10,240✔
197	int ret = TSDB_CODE_SUCCESS;	10,240✔
198	int32_t lino = 0;	10,240✔
199	SSchema* pColSchema = schema + index;	10,240✔
200	SColVal* pVal = taosArrayGet(pTableCxt->pValues, index);	10,240✔
201	TSDB_CHECK_NULL(pVal, ret, lino, end, TSDB_CODE_SUCCESS);	10,242!
202	SSmlKv* kv = (SSmlKv*)data;	10,242✔
203	if (kv->keyLen != strlen(pColSchema->name) \|\| memcmp(kv->key, pColSchema->name, kv->keyLen) != 0 \|\|	10,242✔
204	kv->type != pColSchema->type) {	10,230!
205	ret = TSDB_CODE_SML_INVALID_DATA;	10✔
206	char* tmp = taosMemoryCalloc(kv->keyLen + 1, 1);	10!
207	TSDB_CHECK_NULL(tmp, ret, lino, end, terrno);	12!
208	(void)memcpy(tmp, kv->key, kv->keyLen);	12✔
209	uInfo("SML data(name:%s type:%s) is not same like the db data(name:%s type:%s)", tmp, tDataTypes[kv->type].name,	12!
210	pColSchema->name, tDataTypes[pColSchema->type].name);
211	taosMemoryFree(tmp);	12!
212	goto end;	12✔
213	}
214	if (kv->type == TSDB_DATA_TYPE_NCHAR) {	10,232✔
215	ret = smlMbsToUcs4(kv->value, kv->length, (void**)&pVal->value.pData, &pVal->value.nData, pColSchema->bytes - VARSTR_HEADER_SIZE, charsetCxt);	240✔
216	TSDB_CHECK_CODE(ret, lino, end);	240!
217	} else if (kv->type == TSDB_DATA_TYPE_BINARY) {	9,992✔
218	pVal->value.nData = kv->length;	729✔
219	pVal->value.pData = (uint8_t*)kv->value;	729✔
220	} else if (kv->type == TSDB_DATA_TYPE_GEOMETRY \|\| kv->type == TSDB_DATA_TYPE_VARBINARY) {	9,263!
221	pVal->value.nData = kv->length;	×
222	pVal->value.pData = taosMemoryMalloc(kv->length);	×
223	TSDB_CHECK_NULL(pVal->value.pData, ret, lino, end, terrno);	×
224
225	(void)memcpy(pVal->value.pData, (uint8_t*)kv->value, kv->length);	×
226	} else {
227	valueSetDatum(&pVal->value, kv->type, &(kv->value), kv->length);	9,267✔
228	}
229	pVal->flag = CV_FLAG_VALUE;	10,234✔
230
231	end:	10,246✔
232	if (ret != 0){	10,246✔
233	uError("%s failed at %d since %s", __func__, lino, tstrerror(ret));	12!
234	}
235	return ret;	10,246✔
236	}
237
238	int32_t smlBindData(SQuery* query, bool dataFormat, SArray* tags, SArray* colsSchema, SArray* cols,	1,982✔
239	STableMeta* pTableMeta, char* tableName, const char* sTableName, int32_t sTableNameLen, int32_t ttl,
240	char* msgBuf, int32_t msgBufLen, void* charsetCxt) {
241	int32_t lino = 0;	1,982✔
242	int32_t ret = 0;	1,982✔
243	SMsgBuf pBuf = {.buf = msgBuf, .len = msgBufLen};	1,982✔
244
245	SSchema* pTagsSchema = getTableTagSchema(pTableMeta);	1,982✔
246	SBoundColInfo bindTags = {0};	1,983✔
247	SVCreateTbReq* pCreateTblReq = NULL;	1,983✔
248	SArray* tagName = NULL;	1,983✔
249
250	ret = insInitBoundColsInfo(getNumOfTags(pTableMeta), &bindTags);	1,983✔
251	TSDB_CHECK_CODE(ret, lino, end);	1,984!
252
253	ret = smlBoundColumnData(tags, &bindTags, pTagsSchema, true);	1,984✔
254	TSDB_CHECK_CODE(ret, lino, end);	1,983!
255
256	STag* pTag = NULL;	1,983✔
257	ret = smlBuildTagRow(tags, &bindTags, pTagsSchema, &pTag, &tagName, &pBuf, charsetCxt);	1,983✔
258	TSDB_CHECK_CODE(ret, lino, end);	1,984!
259
260	pCreateTblReq = taosMemoryCalloc(1, sizeof(SVCreateTbReq));	1,984!
261	TSDB_CHECK_NULL(pCreateTblReq, ret, lino, end, terrno);	1,984!
262
263	ret = insBuildCreateTbReq(pCreateTblReq, tableName, pTag, pTableMeta->suid, NULL, tagName,	1,984✔
264	pTableMeta->tableInfo.numOfTags, ttl);	1,984✔
265	TSDB_CHECK_CODE(ret, lino, end);	1,984!
266
267	pCreateTblReq->ctb.stbName = taosMemoryCalloc(1, sTableNameLen + 1);	1,984!
268	TSDB_CHECK_NULL(pCreateTblReq->ctb.stbName, ret, lino, end, terrno);	1,984!
269
270	(void)memcpy(pCreateTblReq->ctb.stbName, sTableName, sTableNameLen);	1,984✔
271
272	if (dataFormat) {	1,984✔
273	STableDataCxt pTableCxt = (STableDataCxt)taosHashGet(((SVnodeModifyOpStmt*)(query->pRoot))->pTableBlockHashObj,	606✔
274	&pTableMeta->uid, sizeof(pTableMeta->uid));	606✔
275	TSDB_CHECK_NULL(pTableCxt, ret, lino, end, TSDB_CODE_TSC_INVALID_OPERATION);	606!
276	(*pTableCxt)->pData->flags \|= SUBMIT_REQ_AUTO_CREATE_TABLE;	606✔
277	(*pTableCxt)->pData->pCreateTbReq = pCreateTblReq;	606✔
278	(*pTableCxt)->pMeta->uid = pTableMeta->uid;	606✔
279	(*pTableCxt)->pMeta->vgId = pTableMeta->vgId;	606✔
280	pCreateTblReq = NULL;	606✔
281	goto end;	606✔
282	}
283
284	STableDataCxt* pTableCxt = NULL;	1,378✔
285	ret = insGetTableDataCxt(((SVnodeModifyOpStmt*)(query->pRoot))->pTableBlockHashObj, &pTableMeta->uid,	1,378✔
286	sizeof(pTableMeta->uid), pTableMeta, &pCreateTblReq, &pTableCxt, false, false);
287	TSDB_CHECK_CODE(ret, lino, end);	1,378!
288
289	SSchema* pSchema = getTableColumnSchema(pTableMeta);	1,378✔
290	ret = smlBoundColumnData(colsSchema, &pTableCxt->boundColsInfo, pSchema, false);	1,378✔
291	TSDB_CHECK_CODE(ret, lino, end);	1,378!
292
293	ret = initTableColSubmitData(pTableCxt);	1,378✔
294	TSDB_CHECK_CODE(ret, lino, end);	1,377!
295
296	int32_t rowNum = taosArrayGetSize(cols);	1,377✔
297	TSDB_CHECK_CONDITION(rowNum > 0, ret, lino, end, TSDB_CODE_TSC_INVALID_OPERATION);	1,378!
298
299	for (int32_t r = 0; r < rowNum; ++r) {	80,006✔
300	void* rowData = taosArrayGetP(cols, r);	78,628✔
301	TSDB_CHECK_NULL(rowData, ret, lino, end, terrno);	78,343!
302
303	// 1. set the parsed value from sql string
304	for (int c = 0; c < pTableCxt->boundColsInfo.numOfBound; ++c) {	352,432✔
305	SSchema* pColSchema = &pSchema[pTableCxt->boundColsInfo.pColIndex[c]];	270,645✔
306	SColVal* pVal = taosArrayGet(pTableCxt->pValues, pTableCxt->boundColsInfo.pColIndex[c]);	270,645✔
307	TSDB_CHECK_NULL(pVal, ret, lino, end, terrno);	261,033!
308	void** p = taosHashGet(rowData, pColSchema->name, strlen(pColSchema->name));	261,033✔
309	if (p == NULL) {	280,685✔
310	continue;	276✔
311	}
312	SSmlKv* kv = (SSmlKv*)p;	280,409✔
313	TSDB_CHECK_CONDITION(kv->type == pColSchema->type, ret, lino, end, TSDB_CODE_TSC_INVALID_OPERATION);	280,409!
314
315	if (pColSchema->type == TSDB_DATA_TYPE_TIMESTAMP) {	280,409✔
316	kv->i = convertTimePrecision(kv->i, TSDB_TIME_PRECISION_NANO, pTableMeta->tableInfo.precision);	83,751✔
317	}
318	if (kv->type == TSDB_DATA_TYPE_NCHAR) {	280,613✔
319	ret = smlMbsToUcs4(kv->value, kv->length, (void*)&pVal->value.pData, (int32_t)&pVal->value.nData, pColSchema->bytes - VARSTR_HEADER_SIZE, charsetCxt);	1,296✔
320	TSDB_CHECK_CODE(ret, lino, end);	1,296!
321	} else if (kv->type == TSDB_DATA_TYPE_BINARY) {	279,317✔
322	pVal->value.nData = kv->length;	1,614✔
323	pVal->value.pData = (uint8_t*)kv->value;	1,614✔
324	} else if (kv->type == TSDB_DATA_TYPE_GEOMETRY \|\| kv->type == TSDB_DATA_TYPE_VARBINARY) {	277,703!
325	pVal->value.nData = kv->length;	9,827✔
326	pVal->value.pData = taosMemoryMalloc(kv->length);	9,827!
327	TSDB_CHECK_NULL(pVal->value.pData, ret, lino, end, terrno);	48!
328	(void)memcpy(pVal->value.pData, (uint8_t*)kv->value, kv->length);	48✔
329	} else {
330	valueSetDatum(&pVal->value, kv->type, &(kv->value), kv->length);	267,876✔
331	}
332	pVal->flag = CV_FLAG_VALUE;	273,789✔
333	}
334
335	SRow** pRow = taosArrayReserve(pTableCxt->pData->aRowP, 1);	81,787✔
336	TSDB_CHECK_NULL(pRow, ret, lino, end, terrno);	83,439!
337	ret = tRowBuild(pTableCxt->pValues, pTableCxt->pSchema, pRow);	83,439✔
338	TSDB_CHECK_CODE(ret, lino, end);	92,133!
339	SRowKey key = {0};	92,133✔
340	tRowGetKey(*pRow, &key);	184,266!
341	insCheckTableDataOrder(pTableCxt, &key);	92,133✔
342	clearColValArraySml(pTableCxt->pValues);	93,103✔
343	}
344
345	end:	1,378✔
346	if (ret != 0){	1,984!
347	uError("%s failed at %d since %s", __func__, lino, tstrerror(ret));	×
348	ret = buildInvalidOperationMsg(&pBuf, tstrerror(ret));	×
349	}
350	insDestroyBoundColInfo(&bindTags);	1,984✔
351	tdDestroySVCreateTbReq(pCreateTblReq);	1,983✔
352	taosMemoryFree(pCreateTblReq);	1,983!
353	taosArrayDestroy(tagName);	1,983✔
354	return ret;	1,983✔
355	}
356
357	int32_t smlInitHandle(SQuery** query) {	1,273✔
358	int32_t lino = 0;	1,273✔
359	int32_t code = 0;	1,273✔
360	SQuery* pQuery = NULL;	1,273✔
361	SVnodeModifyOpStmt* stmt = NULL;	1,273✔
362	TSDB_CHECK_NULL(query, code, lino, end, TSDB_CODE_INVALID_PARA);	1,273✔
363
364	*query = NULL;	1,272✔
365	code = nodesMakeNode(QUERY_NODE_QUERY, (SNode**)&pQuery);	1,272✔
366	TSDB_CHECK_CODE(code, lino, end);	1,292!
367	pQuery->execMode = QUERY_EXEC_MODE_SCHEDULE;	1,292✔
368	pQuery->haveResultSet = false;	1,292✔
369	pQuery->msgType = TDMT_VND_SUBMIT;	1,292✔
370	code = nodesMakeNode(QUERY_NODE_VNODE_MODIFY_STMT, (SNode**)&stmt);	1,292✔
371	TSDB_CHECK_CODE(code, lino, end);	1,291!
372	stmt->pTableBlockHashObj = taosHashInit(16, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BIGINT), true, HASH_NO_LOCK);	1,291✔
373	TSDB_CHECK_NULL(stmt->pTableBlockHashObj, code, lino, end, terrno);	1,291!
374	stmt->freeHashFunc = insDestroyTableDataCxtHashMap;	1,291✔
375	stmt->freeArrayFunc = insDestroyVgroupDataCxtList;	1,291✔
376
377	pQuery->pRoot = (SNode*)stmt;	1,291✔
378	*query = pQuery;	1,291✔
379	return code;	1,291✔
380
381	end:	1✔
382	if (code != 0) {	1!
383	uError("%s failed at %d since %s", __func__, lino, tstrerror(code));	1!
384	}
385	nodesDestroyNode((SNode*)stmt);	1✔
386	qDestroyQuery(pQuery);	1✔
387	return code;	1✔
388	}
389
390	int32_t smlBuildOutput(SQuery* handle, SHashObj* pVgHash) {	1,134✔
391	int32_t lino = 0;	1,134✔
392	int32_t code = 0;	1,134✔
393
394	SVnodeModifyOpStmt* pStmt = (SVnodeModifyOpStmt*)(handle)->pRoot;	1,134✔
395	code = insMergeTableDataCxt(pStmt->pTableBlockHashObj, &pStmt->pVgDataBlocks, true);	1,134✔
396	TSDB_CHECK_CODE(code, lino, end);	1,135!
397	code = insBuildVgDataBlocks(pVgHash, pStmt->pVgDataBlocks, &pStmt->pDataBlocks, false);	1,135✔
398	TSDB_CHECK_CODE(code, lino, end);	1,135!
399
400	end:	1,135✔
401	if (code != 0) {	1,135!
402	uError("%s failed at %d since %s", __func__, lino, tstrerror(code));	×
403	}
404	return code;	1,135✔
405	}
406
407	int32_t smlBuildOutputRaw(SQuery* handle, SHashObj* pVgHash) {	17✔
408	int32_t lino = 0;	17✔
409	int32_t code = 0;	17✔
410
411	SVnodeModifyOpStmt* pStmt = (SVnodeModifyOpStmt*)(handle)->pRoot;	17✔
412	code = insBuildVgDataBlocks(pVgHash, pStmt->pVgDataBlocks, &pStmt->pDataBlocks, false);	17✔
413	TSDB_CHECK_CODE(code, lino, end);	17!
414
415	end:	17✔
416	if (code != 0) {	17!
417	uError("%s failed at %d since %s", __func__, lino, tstrerror(code));	×
418	}
419	return code;	17✔
420	}

taosdata / TDengine / #4113

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