#3653

Committed 14 Mar 2025 08:10AM UTC coverage: 22.565% (-41.0%) from 63.596%

Build # #3653

Build Type

push

travis-ci

Committed by

web-flow

Commit Message

feat(keep): support keep on super table level. (#30097)

* Feat: support use keep while create super table.

* Test(keep): add test for create super table with keep option.

* Feat(keep): Add tmsg for create keep.

* Feat(keep): support alter table option keep.

* Fix(keep): Add baisc test for alter table option.

* Fix(keep): memory leek.

* Feat(keep): add keep to metaEntry&metaCache and fix earliestTs with stn keep.

* Test(keep): add some cases for select with stb keep.

* Fix: fix ci core while alter stb.

* Feat(keep): delete expired data in super table level.

* Feat: remove get stb keep while query.

* Fix : build error.

* Revert "Fix : build error."

This reverts commit 0ed66e4e8.

* Revert "Feat(keep): delete expired data in super table level."

This reverts commit 36330f6b4.

* Fix : build errors.

* Feat : support restart taosd.

* Fix : alter table comment problems.

* Test : add tests for super table keep.

* Fix: change sdb stb reserve size.

* Test: add more tests.

* Feat: Disable normal tables and sub tables from setting the keep parameter

* Fix: add more checks to avoid unknown address.

* Docs: Add docs for stable keep.

* Fix: some review changes.

* Fix: review errors.

Run Details

49248 of 302527 branches covered (16.28%)

Branch coverage included in aggregate %.

53 of 99 new or added lines in 12 files covered. (53.54%)

155872 existing lines in 443 files now uncovered.

87359 of 302857 relevant lines covered (28.84%)

570004.22 hits per line

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

0.0

/source/dnode/vnode/src/sma/smaTimeRange.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 "sma.h"
#include "tq.h"
#include "tsdb.h"

#define SMA_STORAGE_MINUTES_MAX  86400
#define SMA_STORAGE_MINUTES_DAY  1440
#define SMA_STORAGE_SPLIT_FACTOR 14400  // least records in tsma file

static int32_t tdProcessTSmaCreateImpl(SSma *pSma, int64_t version, const char *pMsg);
static int32_t tdProcessTSmaInsertImpl(SSma *pSma, int64_t indexUid, const char *msg);
static int32_t tdProcessTSmaGetDaysImpl(SVnodeCfg *pCfg, void *pCont, uint32_t contLen, int32_t *days);

int32_t tdProcessTSmaInsert(SSma *pSma, int64_t indexUid, const char *msg) {
  int32_t code = TSDB_CODE_SUCCESS;

  if ((code = tdProcessTSmaInsertImpl(pSma, indexUid, msg)) < 0) {
    smaError("vgId:%d, insert tsma data failed since %s", SMA_VID(pSma), tstrerror(code));
  }

  TAOS_RETURN(code);
}

int32_t tdProcessTSmaCreate(SSma *pSma, int64_t ver, const char *msg) {
#ifdef USE_TSMA
  int32_t code = tdProcessTSmaCreateImpl(pSma, ver, msg);

  TAOS_RETURN(code);
#else
  return TSDB_CODE_SUCCESS;
#endif
}

int32_t smaGetTSmaDays(SVnodeCfg *pCfg, void *pCont, uint32_t contLen, int32_t *days) {
  int32_t code = tdProcessTSmaGetDaysImpl(pCfg, pCont, contLen, days);

  TAOS_RETURN(code);
}

/**
 * @brief Judge the tsma file split days
 *
 * @param pCfg
 * @param pCont
 * @param contLen
 * @param days unit is minute
 * @return int32_t
 */
static int32_t tdProcessTSmaGetDaysImpl(SVnodeCfg *pCfg, void *pCont, uint32_t contLen, int32_t *days) {
  int32_t  code = 0;
  int32_t  lino = 0;
  SDecoder coder = {0};
  tDecoderInit(&coder, pCont, contLen);

  STSma tsma = {0};
  if (tDecodeSVCreateTSmaReq(&coder, &tsma) < 0) {
    code = TSDB_CODE_MSG_DECODE_ERROR;
    TSDB_CHECK_CODE(code, lino, _exit);
  }

  STsdbCfg *pTsdbCfg = &pCfg->tsdbCfg;
  int64_t   sInterval = -1;
  TAOS_CHECK_EXIT(convertTimeFromPrecisionToUnit(tsma.interval, pTsdbCfg->precision, TIME_UNIT_SECOND, &sInterval));
  if (0 == sInterval) {
    *days = pTsdbCfg->days;
    goto _exit;
  }
  int64_t records = pTsdbCfg->days * 60 / sInterval;
  if (records >= SMA_STORAGE_SPLIT_FACTOR) {
    *days = pTsdbCfg->days;
  } else {
    int64_t mInterval = -1;
    TAOS_CHECK_EXIT(convertTimeFromPrecisionToUnit(tsma.interval, pTsdbCfg->precision, TIME_UNIT_MINUTE, &mInterval));
    int64_t daysPerFile = mInterval * SMA_STORAGE_MINUTES_DAY * 2;

    if (daysPerFile > SMA_STORAGE_MINUTES_MAX) {
      *days = SMA_STORAGE_MINUTES_MAX;
    } else {
      *days = (int32_t)daysPerFile;
    }

    if (*days < pTsdbCfg->days) {
      *days = pTsdbCfg->days;
    }
  }
_exit:
  if (code) {
    smaWarn("vgId:%d, failed at line %d to get tsma days %d since %s", pCfg->vgId, lino, *days, tstrerror(code));
  } else {
    smaDebug("vgId:%d, succeed to get tsma days %d", pCfg->vgId, *days);
  }
  tDecoderClear(&coder);
  TAOS_RETURN(code);
}

/**
 * @brief create tsma meta and result stable
 *
 * @param pSma
 * @param version
 * @param pMsg
 * @return int32_t
 */
static int32_t tdProcessTSmaCreateImpl(SSma *pSma, int64_t ver, const char *pMsg) {
  int32_t        code = 0;
  int32_t        lino = 0;
  SSmaCfg       *pCfg = (SSmaCfg *)pMsg;
  SName          stbFullName = {0};
  SVCreateStbReq pReq = {0};

  if (TD_VID(pSma->pVnode) == pCfg->dstVgId) {
    // create tsma meta in dstVgId
    TAOS_CHECK_EXIT(metaCreateTSma(SMA_META(pSma), ver, pCfg));

    // create stable to save tsma result in dstVgId
    TAOS_CHECK_EXIT(tNameFromString(&stbFullName, pCfg->dstTbName, T_NAME_ACCT | T_NAME_DB | T_NAME_TABLE));
    pReq.name = (char *)tNameGetTableName(&stbFullName);
    pReq.suid = pCfg->dstTbUid;
    pReq.schemaRow = pCfg->schemaRow;
    pReq.schemaTag = pCfg->schemaTag;

    TAOS_CHECK_EXIT(metaCreateSuperTable(SMA_META(pSma), ver, &pReq));
  } else {
    TAOS_CHECK_EXIT(TSDB_CODE_TSMA_INVALID_STAT);
  }

_exit:
  if (code) {
    smaError("vgId:%d, failed at line %d to create sma index %s %" PRIi64 " on stb:%" PRIi64 ", dstSuid:%" PRIi64
             " dstTb:%s dstVg:%d since %s",
             SMA_VID(pSma), lino, pCfg->indexName, pCfg->indexUid, pCfg->tableUid, pCfg->dstTbUid, pReq.name,
             pCfg->dstVgId, tstrerror(code));
  } else {
    smaDebug("vgId:%d, success to create sma index %s %" PRIi64 " on stb:%" PRIi64 ", dstSuid:%" PRIi64
             " dstTb:%s dstVg:%d",
             SMA_VID(pSma), pCfg->indexName, pCfg->indexUid, pCfg->tableUid, pCfg->dstTbUid, pReq.name, pCfg->dstVgId);
  }

  TAOS_RETURN(code);
}

int32_t smaBlockToSubmit(SVnode *pVnode, const SArray *pBlocks, const STSchema *pTSchema, int64_t suid,
                         const char *stbFullName, SBatchDeleteReq *pDeleteReq, void **ppData, int32_t *pLen) {
  int32_t      code = 0;
  int32_t      lino = 0;
  void        *pBuf = NULL;
  int32_t      len = 0;
  SSubmitReq2 *pReq = NULL;
  SArray      *tagArray = NULL;
  SHashObj    *pTableIndexMap = NULL;

  int32_t numOfBlocks = taosArrayGetSize(pBlocks);

  tagArray = taosArrayInit(1, sizeof(STagVal));
  pReq = taosMemoryCalloc(1, sizeof(SSubmitReq2));

  if (!tagArray || !pReq) {
    TAOS_CHECK_EXIT(terrno);
  }

  pReq->aSubmitTbData = taosArrayInit(1, sizeof(SSubmitTbData));
  if (pReq->aSubmitTbData == NULL) {
    TAOS_CHECK_EXIT(terrno);
  }

  pTableIndexMap = taosHashInit(numOfBlocks, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BIGINT), false, HASH_NO_LOCK);
  if (pTableIndexMap == NULL) {
    TAOS_CHECK_EXIT(terrno);
  }

  // SSubmitTbData req
  for (int32_t i = 0; i < numOfBlocks; ++i) {
    SSDataBlock *pDataBlock = taosArrayGet(pBlocks, i);
    if (pDataBlock->info.type == STREAM_DELETE_RESULT) {
      pDeleteReq->suid = suid;
      pDeleteReq->deleteReqs = taosArrayInit(0, sizeof(SSingleDeleteReq));
      TAOS_CHECK_EXIT(tqBuildDeleteReq(pVnode->pTq, stbFullName, pDataBlock, pDeleteReq, "", true));
      continue;
    }

    SSubmitTbData tbData = {.suid = suid, .uid = 0, .sver = pTSchema->version, .flags = SUBMIT_REQ_AUTO_CREATE_TABLE};

    int32_t cid = taosArrayGetSize(pDataBlock->pDataBlock) + 1;

    TAOS_CHECK_EXIT(buildAutoCreateTableReq(stbFullName, suid, cid, pDataBlock, tagArray, true, &tbData.pCreateTbReq));

    {
      uint64_t groupId = pDataBlock->info.id.groupId;

      int32_t *index = taosHashGet(pTableIndexMap, &groupId, sizeof(groupId));
      if (index == NULL) {  // no data yet, append it
        code = tqSetDstTableDataPayload(suid, pTSchema, i, pDataBlock, &tbData, INT64_MIN, "");
        if (code != TSDB_CODE_SUCCESS) {
          continue;
        }

        if (taosArrayPush(pReq->aSubmitTbData, &tbData) == NULL) {
          code = terrno;
          continue;
        }

        int32_t size = (int32_t)taosArrayGetSize(pReq->aSubmitTbData) - 1;
        TAOS_CHECK_EXIT(taosHashPut(pTableIndexMap, &groupId, sizeof(groupId), &size, sizeof(size)));
      } else {
        code = tqSetDstTableDataPayload(suid, pTSchema, i, pDataBlock, &tbData, INT64_MIN, "");
        if (code != TSDB_CODE_SUCCESS) {
          continue;
        }

        SSubmitTbData *pExisted = taosArrayGet(pReq->aSubmitTbData, *index);
        code = doMergeExistedRows(pExisted, &tbData, "id");
        if (code != TSDB_CODE_SUCCESS) {
          continue;
        }
      }
    }
  }

  // encode
  tEncodeSize(tEncodeSubmitReq, pReq, len, code);
  if (TSDB_CODE_SUCCESS == code) {
    SEncoder encoder;
    len += sizeof(SSubmitReq2Msg);
    if (!(pBuf = rpcMallocCont(len))) {
      code = terrno;
      TSDB_CHECK_CODE(code, lino, _exit);
    }

    ((SSubmitReq2Msg *)pBuf)->header.vgId = TD_VID(pVnode);
    ((SSubmitReq2Msg *)pBuf)->header.contLen = htonl(len);
    ((SSubmitReq2Msg *)pBuf)->version = htobe64(1);
    tEncoderInit(&encoder, POINTER_SHIFT(pBuf, sizeof(SSubmitReq2Msg)), len - sizeof(SSubmitReq2Msg));
    if ((code = tEncodeSubmitReq(&encoder, pReq)) < 0) {
      tEncoderClear(&encoder);
      TSDB_CHECK_CODE(code, lino, _exit);
    }
    tEncoderClear(&encoder);
  }

_exit:
  taosArrayDestroy(tagArray);
  taosHashCleanup(pTableIndexMap);
  if (pReq != NULL) {
    tDestroySubmitReq(pReq, TSDB_MSG_FLG_ENCODE);
    taosMemoryFree(pReq);
  }

  if (code) {
    rpcFreeCont(pBuf);
    taosArrayDestroy(pDeleteReq->deleteReqs);
    smaWarn("vgId:%d, failed at line %d since %s", TD_VID(pVnode), lino, tstrerror(code));
  } else {
    if (ppData) *ppData = pBuf;
    if (pLen) *pLen = len;
  }
  TAOS_RETURN(code);
}

static int32_t tsmaProcessDelReq(SSma *pSma, int64_t indexUid, SBatchDeleteReq *pDelReq) {
  int32_t code = 0;
  int32_t lino = 0;

  if (taosArrayGetSize(pDelReq->deleteReqs) > 0) {
    int32_t len = 0;
    tEncodeSize(tEncodeSBatchDeleteReq, pDelReq, len, code);
    TSDB_CHECK_CODE(code, lino, _exit);

    void *pBuf = rpcMallocCont(len + sizeof(SMsgHead));
    if (!pBuf) {
      code = terrno;
      TSDB_CHECK_CODE(code, lino, _exit);
    }

    SEncoder encoder;
    tEncoderInit(&encoder, POINTER_SHIFT(pBuf, sizeof(SMsgHead)), len);
    if ((code = tEncodeSBatchDeleteReq(&encoder, pDelReq)) < 0) {
      tEncoderClear(&encoder);
      TSDB_CHECK_CODE(code, lino, _exit);
    }
    tEncoderClear(&encoder);

    ((SMsgHead *)pBuf)->vgId = TD_VID(pSma->pVnode);

    SRpcMsg delMsg = {.msgType = TDMT_VND_BATCH_DEL, .pCont = pBuf, .contLen = len + sizeof(SMsgHead)};
    code = tmsgPutToQueue(&pSma->pVnode->msgCb, WRITE_QUEUE, &delMsg);
    TSDB_CHECK_CODE(code, lino, _exit);
  }

_exit:
  taosArrayDestroy(pDelReq->deleteReqs);
  if (code) {
    smaError("vgId:%d, failed at line %d to process delete req for smaIndex %" PRIi64 " since %s", SMA_VID(pSma), lino,
             indexUid, tstrerror(code));
  }

  TAOS_RETURN(code);
}

/**
 * @brief Insert/Update Time-range-wise SMA data.
 *
 * @param pSma
 * @param msg
 * @return int32_t
 */
static int32_t tdProcessTSmaInsertImpl(SSma *pSma, int64_t indexUid, const char *msg) {
  int32_t       code = 0;
  int32_t       lino = 0;
  const SArray *pDataBlocks = (const SArray *)msg;

  if (taosArrayGetSize(pDataBlocks) <= 0) {
    code = TSDB_CODE_TSMA_INVALID_PARA;
    TSDB_CHECK_CODE(code, lino, _exit);
  }

  if (tdCheckAndInitSmaEnv(pSma, TSDB_SMA_TYPE_TIME_RANGE) != 0) {
    code = TSDB_CODE_TSMA_INIT_FAILED;
    TSDB_CHECK_CODE(code, lino, _exit);
  }

  SSmaEnv   *pEnv = SMA_TSMA_ENV(pSma);
  SSmaStat  *pStat = NULL;
  STSmaStat *pTsmaStat = NULL;

  if (!pEnv || !(pStat = SMA_ENV_STAT(pEnv))) {
    code = TSDB_CODE_TSMA_INVALID_ENV;
    TSDB_CHECK_CODE(code, lino, _exit);
  }

  pTsmaStat = SMA_STAT_TSMA(pStat);

  if (!pTsmaStat->pTSma) {
    STSma *pTSma = metaGetSmaInfoByIndex(SMA_META(pSma), indexUid);
    if (!pTSma) {
      code = TSDB_CODE_TSMA_INVALID_PTR;
      TSDB_CHECK_CODE(code, lino, _exit);
    }
    pTsmaStat->pTSma = pTSma;
    code = metaGetTbTSchemaNotNull(SMA_META(pSma), pTSma->dstTbUid, -1, 1, &pTsmaStat->pTSchema);
    TSDB_CHECK_CODE(code, lino, _exit);
  }

  if (pTsmaStat->pTSma->indexUid != indexUid) {
    code = TSDB_CODE_APP_ERROR;
    TSDB_CHECK_CODE(code, lino, _exit);
  }

  SBatchDeleteReq deleteReq = {0};
  void           *pSubmitReq = NULL;
  int32_t         contLen = 0;

  code = smaBlockToSubmit(pSma->pVnode, (const SArray *)msg, pTsmaStat->pTSchema, pTsmaStat->pTSma->dstTbUid,
                          pTsmaStat->pTSma->dstTbName, &deleteReq, &pSubmitReq, &contLen);
  TSDB_CHECK_CODE(code, lino, _exit);

  TAOS_CHECK_EXIT(tsmaProcessDelReq(pSma, indexUid, &deleteReq));

#if 0
  if (!strncasecmp("td.tsma.rst.tb", pTsmaStat->pTSma->dstTbName, 14)) {
    code = TSDB_CODE_APP_ERROR;
    smaError("vgId:%d, tsma insert for smaIndex %" PRIi64 " failed since %s, %s", SMA_VID(pSma), indexUid,
             pTsmaStat->pTSma->indexUid, tstrerror(code), pTsmaStat->pTSma->dstTbName);
    goto _err;
  }
#endif

  SRpcMsg submitReqMsg = {
      .msgType = TDMT_VND_SUBMIT,
      .pCont = pSubmitReq,
      .contLen = contLen,
  };

  code = tmsgPutToQueue(&pSma->pVnode->msgCb, WRITE_QUEUE, &submitReqMsg);
  TSDB_CHECK_CODE(code, lino, _exit);

_exit:
  if (code) {
    smaError("vgId:%d, %s failed at line %d since %s, smaIndex:%" PRIi64, SMA_VID(pSma), __func__, lino,
             tstrerror(code), indexUid);
  }
  TAOS_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 "sma.h"
17	#include "tq.h"
18	#include "tsdb.h"
19
20	#define SMA_STORAGE_MINUTES_MAX 86400
21	#define SMA_STORAGE_MINUTES_DAY 1440
22	#define SMA_STORAGE_SPLIT_FACTOR 14400 // least records in tsma file
23
24	static int32_t tdProcessTSmaCreateImpl(SSma pSma, int64_t version, const char pMsg);
25	static int32_t tdProcessTSmaInsertImpl(SSma pSma, int64_t indexUid, const char msg);
26	static int32_t tdProcessTSmaGetDaysImpl(SVnodeCfg pCfg, void pCont, uint32_t contLen, int32_t *days);
27
UNCOV 28	int32_t tdProcessTSmaInsert(SSma pSma, int64_t indexUid, const char msg) {	×
UNCOV 29	int32_t code = TSDB_CODE_SUCCESS;	×
30
UNCOV 31	if ((code = tdProcessTSmaInsertImpl(pSma, indexUid, msg)) < 0) {	×
32	smaError("vgId:%d, insert tsma data failed since %s", SMA_VID(pSma), tstrerror(code));	×
33	}
34
UNCOV 35	TAOS_RETURN(code);	×
36	}
37
UNCOV 38	int32_t tdProcessTSmaCreate(SSma pSma, int64_t ver, const char msg) {	×
39	#ifdef USE_TSMA
UNCOV 40	int32_t code = tdProcessTSmaCreateImpl(pSma, ver, msg);	×
41
UNCOV 42	TAOS_RETURN(code);	×
43	#else
44	return TSDB_CODE_SUCCESS;
45	#endif
46	}
47
UNCOV 48	int32_t smaGetTSmaDays(SVnodeCfg pCfg, void pCont, uint32_t contLen, int32_t *days) {	×
UNCOV 49	int32_t code = tdProcessTSmaGetDaysImpl(pCfg, pCont, contLen, days);	×
50
UNCOV 51	TAOS_RETURN(code);	×
52	}
53
54	/**
55	* @brief Judge the tsma file split days
56	*
57	* @param pCfg
58	* @param pCont
59	* @param contLen
60	* @param days unit is minute
61	* @return int32_t
62	*/
UNCOV 63	static int32_t tdProcessTSmaGetDaysImpl(SVnodeCfg pCfg, void pCont, uint32_t contLen, int32_t *days) {	×
UNCOV 64	int32_t code = 0;	×
UNCOV 65	int32_t lino = 0;	×
UNCOV 66	SDecoder coder = {0};	×
67	tDecoderInit(&coder, pCont, contLen);	×
68
UNCOV 69	STSma tsma = {0};	×
UNCOV 70	if (tDecodeSVCreateTSmaReq(&coder, &tsma) < 0) {	×
UNCOV 71	code = TSDB_CODE_MSG_DECODE_ERROR;	×
UNCOV 72	TSDB_CHECK_CODE(code, lino, _exit);	×
73	}
74
UNCOV 75	STsdbCfg *pTsdbCfg = &pCfg->tsdbCfg;	×
UNCOV 76	int64_t sInterval = -1;	×
UNCOV 77	TAOS_CHECK_EXIT(convertTimeFromPrecisionToUnit(tsma.interval, pTsdbCfg->precision, TIME_UNIT_SECOND, &sInterval));	×
UNCOV 78	if (0 == sInterval) {	×
UNCOV 79	*days = pTsdbCfg->days;	×
UNCOV 80	goto _exit;	×
81	}
UNCOV 82	int64_t records = pTsdbCfg->days * 60 / sInterval;	×
UNCOV 83	if (records >= SMA_STORAGE_SPLIT_FACTOR) {	×
UNCOV 84	*days = pTsdbCfg->days;	×
85	} else {
UNCOV 86	int64_t mInterval = -1;	×
87	TAOS_CHECK_EXIT(convertTimeFromPrecisionToUnit(tsma.interval, pTsdbCfg->precision, TIME_UNIT_MINUTE, &mInterval));	×
UNCOV 88	int64_t daysPerFile = mInterval * SMA_STORAGE_MINUTES_DAY * 2;	×
89
UNCOV 90	if (daysPerFile > SMA_STORAGE_MINUTES_MAX) {	×
UNCOV 91	*days = SMA_STORAGE_MINUTES_MAX;	×
92	} else {
93	*days = (int32_t)daysPerFile;	×
94	}
95
UNCOV 96	if (*days < pTsdbCfg->days) {	×
UNCOV 97	*days = pTsdbCfg->days;	×
98	}
99	}
UNCOV 100	_exit:	×
UNCOV 101	if (code) {	×
UNCOV 102	smaWarn("vgId:%d, failed at line %d to get tsma days %d since %s", pCfg->vgId, lino, *days, tstrerror(code));	×
103	} else {
UNCOV 104	smaDebug("vgId:%d, succeed to get tsma days %d", pCfg->vgId, *days);	×
105	}
UNCOV 106	tDecoderClear(&coder);	×
UNCOV 107	TAOS_RETURN(code);	×
108	}
109
110	/**
111	* @brief create tsma meta and result stable
112	*
113	* @param pSma
114	* @param version
115	* @param pMsg
116	* @return int32_t
117	*/
UNCOV 118	static int32_t tdProcessTSmaCreateImpl(SSma pSma, int64_t ver, const char pMsg) {	×
UNCOV 119	int32_t code = 0;	×
UNCOV 120	int32_t lino = 0;	×
UNCOV 121	SSmaCfg pCfg = (SSmaCfg )pMsg;	×
UNCOV 122	SName stbFullName = {0};	×
UNCOV 123	SVCreateStbReq pReq = {0};	×
124
UNCOV 125	if (TD_VID(pSma->pVnode) == pCfg->dstVgId) {	×
126	// create tsma meta in dstVgId
UNCOV 127	TAOS_CHECK_EXIT(metaCreateTSma(SMA_META(pSma), ver, pCfg));	×
128
129	// create stable to save tsma result in dstVgId
UNCOV 130	TAOS_CHECK_EXIT(tNameFromString(&stbFullName, pCfg->dstTbName, T_NAME_ACCT \| T_NAME_DB \| T_NAME_TABLE));	×
UNCOV 131	pReq.name = (char *)tNameGetTableName(&stbFullName);	×
UNCOV 132	pReq.suid = pCfg->dstTbUid;	×
UNCOV 133	pReq.schemaRow = pCfg->schemaRow;	×
134	pReq.schemaTag = pCfg->schemaTag;	×
135
UNCOV 136	TAOS_CHECK_EXIT(metaCreateSuperTable(SMA_META(pSma), ver, &pReq));	×
137	} else {
UNCOV 138	TAOS_CHECK_EXIT(TSDB_CODE_TSMA_INVALID_STAT);	×
139	}
140
UNCOV 141	_exit:	×
UNCOV 142	if (code) {	×
UNCOV 143	smaError("vgId:%d, failed at line %d to create sma index %s %" PRIi64 " on stb:%" PRIi64 ", dstSuid:%" PRIi64	×
144	" dstTb:%s dstVg:%d since %s",
145	SMA_VID(pSma), lino, pCfg->indexName, pCfg->indexUid, pCfg->tableUid, pCfg->dstTbUid, pReq.name,
146	pCfg->dstVgId, tstrerror(code));
147	} else {
UNCOV 148	smaDebug("vgId:%d, success to create sma index %s %" PRIi64 " on stb:%" PRIi64 ", dstSuid:%" PRIi64	×
149	" dstTb:%s dstVg:%d",
150	SMA_VID(pSma), pCfg->indexName, pCfg->indexUid, pCfg->tableUid, pCfg->dstTbUid, pReq.name, pCfg->dstVgId);
151	}
152
UNCOV 153	TAOS_RETURN(code);	×
154	}
155
UNCOV 156	int32_t smaBlockToSubmit(SVnode pVnode, const SArray pBlocks, const STSchema *pTSchema, int64_t suid,	×
157	const char stbFullName, SBatchDeleteReq pDeleteReq, void *ppData, int32_t pLen) {
UNCOV 158	int32_t code = 0;	×
UNCOV 159	int32_t lino = 0;	×
UNCOV 160	void *pBuf = NULL;	×
UNCOV 161	int32_t len = 0;	×
UNCOV 162	SSubmitReq2 *pReq = NULL;	×
UNCOV 163	SArray *tagArray = NULL;	×
UNCOV 164	SHashObj *pTableIndexMap = NULL;	×
165
UNCOV 166	int32_t numOfBlocks = taosArrayGetSize(pBlocks);	×
167
168	tagArray = taosArrayInit(1, sizeof(STagVal));	×
UNCOV 169	pReq = taosMemoryCalloc(1, sizeof(SSubmitReq2));	×
170
UNCOV 171	if (!tagArray \|\| !pReq) {	×
UNCOV 172	TAOS_CHECK_EXIT(terrno);	×
173	}
174
UNCOV 175	pReq->aSubmitTbData = taosArrayInit(1, sizeof(SSubmitTbData));	×
UNCOV 176	if (pReq->aSubmitTbData == NULL) {	×
UNCOV 177	TAOS_CHECK_EXIT(terrno);	×
178	}
179
UNCOV 180	pTableIndexMap = taosHashInit(numOfBlocks, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BIGINT), false, HASH_NO_LOCK);	×
UNCOV 181	if (pTableIndexMap == NULL) {	×
UNCOV 182	TAOS_CHECK_EXIT(terrno);	×
183	}
184
185	// SSubmitTbData req
186	for (int32_t i = 0; i < numOfBlocks; ++i) {	×
187	SSDataBlock *pDataBlock = taosArrayGet(pBlocks, i);	×
188	if (pDataBlock->info.type == STREAM_DELETE_RESULT) {	×
UNCOV 189	pDeleteReq->suid = suid;	×
UNCOV 190	pDeleteReq->deleteReqs = taosArrayInit(0, sizeof(SSingleDeleteReq));	×
UNCOV 191	TAOS_CHECK_EXIT(tqBuildDeleteReq(pVnode->pTq, stbFullName, pDataBlock, pDeleteReq, "", true));	×
UNCOV 192	continue;	×
193	}
194
UNCOV 195	SSubmitTbData tbData = {.suid = suid, .uid = 0, .sver = pTSchema->version, .flags = SUBMIT_REQ_AUTO_CREATE_TABLE};	×
196
UNCOV 197	int32_t cid = taosArrayGetSize(pDataBlock->pDataBlock) + 1;	×
198
UNCOV 199	TAOS_CHECK_EXIT(buildAutoCreateTableReq(stbFullName, suid, cid, pDataBlock, tagArray, true, &tbData.pCreateTbReq));	×
200
201	{
UNCOV 202	uint64_t groupId = pDataBlock->info.id.groupId;	×
203
204	int32_t *index = taosHashGet(pTableIndexMap, &groupId, sizeof(groupId));	×
UNCOV 205	if (index == NULL) { // no data yet, append it	×
UNCOV 206	code = tqSetDstTableDataPayload(suid, pTSchema, i, pDataBlock, &tbData, INT64_MIN, "");	×
UNCOV 207	if (code != TSDB_CODE_SUCCESS) {	×
208	continue;	×
209	}
210
UNCOV 211	if (taosArrayPush(pReq->aSubmitTbData, &tbData) == NULL) {	×
UNCOV 212	code = terrno;	×
UNCOV 213	continue;	×
214	}
215
216	int32_t size = (int32_t)taosArrayGetSize(pReq->aSubmitTbData) - 1;	×
217	TAOS_CHECK_EXIT(taosHashPut(pTableIndexMap, &groupId, sizeof(groupId), &size, sizeof(size)));	×
218	} else {
UNCOV 219	code = tqSetDstTableDataPayload(suid, pTSchema, i, pDataBlock, &tbData, INT64_MIN, "");	×
220	if (code != TSDB_CODE_SUCCESS) {	×
221	continue;	×
222	}
223
UNCOV 224	SSubmitTbData pExisted = taosArrayGet(pReq->aSubmitTbData, index);	×
UNCOV 225	code = doMergeExistedRows(pExisted, &tbData, "id");	×
UNCOV 226	if (code != TSDB_CODE_SUCCESS) {	×
UNCOV 227	continue;	×
228	}
229	}
230	}
231	}
232
233	// encode
UNCOV 234	tEncodeSize(tEncodeSubmitReq, pReq, len, code);	×
235	if (TSDB_CODE_SUCCESS == code) {	×
236	SEncoder encoder;
UNCOV 237	len += sizeof(SSubmitReq2Msg);	×
UNCOV 238	if (!(pBuf = rpcMallocCont(len))) {	×
UNCOV 239	code = terrno;	×
UNCOV 240	TSDB_CHECK_CODE(code, lino, _exit);	×
241	}
242
UNCOV 243	((SSubmitReq2Msg *)pBuf)->header.vgId = TD_VID(pVnode);	×
244	((SSubmitReq2Msg *)pBuf)->header.contLen = htonl(len);	×
245	((SSubmitReq2Msg *)pBuf)->version = htobe64(1);	×
UNCOV 246	tEncoderInit(&encoder, POINTER_SHIFT(pBuf, sizeof(SSubmitReq2Msg)), len - sizeof(SSubmitReq2Msg));	×
UNCOV 247	if ((code = tEncodeSubmitReq(&encoder, pReq)) < 0) {	×
UNCOV 248	tEncoderClear(&encoder);	×
UNCOV 249	TSDB_CHECK_CODE(code, lino, _exit);	×
250	}
UNCOV 251	tEncoderClear(&encoder);	×
252	}
253
UNCOV 254	_exit:	×
UNCOV 255	taosArrayDestroy(tagArray);	×
UNCOV 256	taosHashCleanup(pTableIndexMap);	×
UNCOV 257	if (pReq != NULL) {	×
UNCOV 258	tDestroySubmitReq(pReq, TSDB_MSG_FLG_ENCODE);	×
259	taosMemoryFree(pReq);	×
260	}
261
UNCOV 262	if (code) {	×
UNCOV 263	rpcFreeCont(pBuf);	×
UNCOV 264	taosArrayDestroy(pDeleteReq->deleteReqs);	×
UNCOV 265	smaWarn("vgId:%d, failed at line %d since %s", TD_VID(pVnode), lino, tstrerror(code));	×
266	} else {
UNCOV 267	if (ppData) *ppData = pBuf;	×
UNCOV 268	if (pLen) *pLen = len;	×
269	}
UNCOV 270	TAOS_RETURN(code);	×
271	}
272
UNCOV 273	static int32_t tsmaProcessDelReq(SSma pSma, int64_t indexUid, SBatchDeleteReq pDelReq) {	×
274	int32_t code = 0;	×
275	int32_t lino = 0;	×
276
UNCOV 277	if (taosArrayGetSize(pDelReq->deleteReqs) > 0) {	×
278	int32_t len = 0;	×
279	tEncodeSize(tEncodeSBatchDeleteReq, pDelReq, len, code);	×
280	TSDB_CHECK_CODE(code, lino, _exit);	×
281
UNCOV 282	void *pBuf = rpcMallocCont(len + sizeof(SMsgHead));	×
UNCOV 283	if (!pBuf) {	×
UNCOV 284	code = terrno;	×
285	TSDB_CHECK_CODE(code, lino, _exit);	×
286	}
287
288	SEncoder encoder;
UNCOV 289	tEncoderInit(&encoder, POINTER_SHIFT(pBuf, sizeof(SMsgHead)), len);	×
290	if ((code = tEncodeSBatchDeleteReq(&encoder, pDelReq)) < 0) {	×
UNCOV 291	tEncoderClear(&encoder);	×
292	TSDB_CHECK_CODE(code, lino, _exit);	×
293	}
294	tEncoderClear(&encoder);	×
295
296	((SMsgHead *)pBuf)->vgId = TD_VID(pSma->pVnode);	×
297
UNCOV 298	SRpcMsg delMsg = {.msgType = TDMT_VND_BATCH_DEL, .pCont = pBuf, .contLen = len + sizeof(SMsgHead)};	×
UNCOV 299	code = tmsgPutToQueue(&pSma->pVnode->msgCb, WRITE_QUEUE, &delMsg);	×
UNCOV 300	TSDB_CHECK_CODE(code, lino, _exit);	×
301	}
302
UNCOV 303	_exit:	×
UNCOV 304	taosArrayDestroy(pDelReq->deleteReqs);	×
UNCOV 305	if (code) {	×
UNCOV 306	smaError("vgId:%d, failed at line %d to process delete req for smaIndex %" PRIi64 " since %s", SMA_VID(pSma), lino,	×
307	indexUid, tstrerror(code));
308	}
309
UNCOV 310	TAOS_RETURN(code);	×
311	}
312
313	/**
314	* @brief Insert/Update Time-range-wise SMA data.
315	*
316	* @param pSma
317	* @param msg
318	* @return int32_t
319	*/
UNCOV 320	static int32_t tdProcessTSmaInsertImpl(SSma pSma, int64_t indexUid, const char msg) {	×
UNCOV 321	int32_t code = 0;	×
322	int32_t lino = 0;	×
323	const SArray pDataBlocks = (const SArray )msg;	×
324
UNCOV 325	if (taosArrayGetSize(pDataBlocks) <= 0) {	×
UNCOV 326	code = TSDB_CODE_TSMA_INVALID_PARA;	×
327	TSDB_CHECK_CODE(code, lino, _exit);	×
328	}
329
UNCOV 330	if (tdCheckAndInitSmaEnv(pSma, TSDB_SMA_TYPE_TIME_RANGE) != 0) {	×
UNCOV 331	code = TSDB_CODE_TSMA_INIT_FAILED;	×
UNCOV 332	TSDB_CHECK_CODE(code, lino, _exit);	×
333	}
334
UNCOV 335	SSmaEnv *pEnv = SMA_TSMA_ENV(pSma);	×
336	SSmaStat *pStat = NULL;	×
337	STSmaStat *pTsmaStat = NULL;	×
338
UNCOV 339	if (!pEnv \|\| !(pStat = SMA_ENV_STAT(pEnv))) {	×
UNCOV 340	code = TSDB_CODE_TSMA_INVALID_ENV;	×
UNCOV 341	TSDB_CHECK_CODE(code, lino, _exit);	×
342	}
343
UNCOV 344	pTsmaStat = SMA_STAT_TSMA(pStat);	×
345
346	if (!pTsmaStat->pTSma) {	×
UNCOV 347	STSma *pTSma = metaGetSmaInfoByIndex(SMA_META(pSma), indexUid);	×
UNCOV 348	if (!pTSma) {	×
UNCOV 349	code = TSDB_CODE_TSMA_INVALID_PTR;	×
UNCOV 350	TSDB_CHECK_CODE(code, lino, _exit);	×
351	}
UNCOV 352	pTsmaStat->pTSma = pTSma;	×
UNCOV 353	code = metaGetTbTSchemaNotNull(SMA_META(pSma), pTSma->dstTbUid, -1, 1, &pTsmaStat->pTSchema);	×
354	TSDB_CHECK_CODE(code, lino, _exit);	×
355	}
356
UNCOV 357	if (pTsmaStat->pTSma->indexUid != indexUid) {	×
UNCOV 358	code = TSDB_CODE_APP_ERROR;	×
UNCOV 359	TSDB_CHECK_CODE(code, lino, _exit);	×
360	}
361
UNCOV 362	SBatchDeleteReq deleteReq = {0};	×
UNCOV 363	void *pSubmitReq = NULL;	×
UNCOV 364	int32_t contLen = 0;	×
365
UNCOV 366	code = smaBlockToSubmit(pSma->pVnode, (const SArray *)msg, pTsmaStat->pTSchema, pTsmaStat->pTSma->dstTbUid,	×
UNCOV 367	pTsmaStat->pTSma->dstTbName, &deleteReq, &pSubmitReq, &contLen);	×
UNCOV 368	TSDB_CHECK_CODE(code, lino, _exit);	×
369
UNCOV 370	TAOS_CHECK_EXIT(tsmaProcessDelReq(pSma, indexUid, &deleteReq));	×
371
372	#if 0
373	if (!strncasecmp("td.tsma.rst.tb", pTsmaStat->pTSma->dstTbName, 14)) {
374	code = TSDB_CODE_APP_ERROR;
375	smaError("vgId:%d, tsma insert for smaIndex %" PRIi64 " failed since %s, %s", SMA_VID(pSma), indexUid,
376	pTsmaStat->pTSma->indexUid, tstrerror(code), pTsmaStat->pTSma->dstTbName);
377	goto _err;
378	}
379	#endif
380
UNCOV 381	SRpcMsg submitReqMsg = {	×
382	.msgType = TDMT_VND_SUBMIT,
383	.pCont = pSubmitReq,
384	.contLen = contLen,
385	};
386
UNCOV 387	code = tmsgPutToQueue(&pSma->pVnode->msgCb, WRITE_QUEUE, &submitReqMsg);	×
388	TSDB_CHECK_CODE(code, lino, _exit);	×
389
UNCOV 390	_exit:	×
UNCOV 391	if (code) {	×
UNCOV 392	smaError("vgId:%d, %s failed at line %d since %s, smaIndex:%" PRIi64, SMA_VID(pSma), __func__, lino,	×
393	tstrerror(code), indexUid);
394	}
UNCOV 395	TAOS_RETURN(code);	×
396	}

taosdata / TDengine / #3653

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