smaRollup.c 39.2 KB
Newer Older
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17
/*
 * 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"

C
Cary Xu 已提交
18
#define RSMA_QTASKINFO_PERSIST_MS 7200000
19
#define RSMA_QTASKINFO_BUFSIZE    32768
C
Cary Xu 已提交
20
#define RSMA_QTASKINFO_HEAD_LEN   (sizeof(int32_t) + sizeof(int8_t) + sizeof(int64_t))  // len + type + suid
C
Cary Xu 已提交
21 22 23
typedef enum { TD_QTASK_TMP_FILE = 0, TD_QTASK_CUR_FILE } TD_QTASK_FILE_T;
static const char *tdQTaskInfoFname[] = {"qtaskinfo.t", "qtaskinfo"};

24
typedef struct SRSmaQTaskInfoItem SRSmaQTaskInfoItem;
C
Cary Xu 已提交
25
typedef struct SRSmaQTaskInfoIter SRSmaQTaskInfoIter;
26

C
Cary Xu 已提交
27 28 29 30 31 32
static int32_t tdUidStorePut(STbUidStore *pStore, tb_uid_t suid, tb_uid_t *uid);
static int32_t tdUpdateTbUidListImpl(SSma *pSma, tb_uid_t *suid, SArray *tbUids);
static int32_t tdSetRSmaInfoItemParams(SSma *pSma, SRSmaParam *param, SRSmaInfo *pRSmaInfo, SReadHandle *handle,
                                       int8_t idx);
static int32_t tdExecuteRSmaImpl(SSma *pSma, const void *pMsg, int32_t inputType, SRSmaInfoItem *rsmaItem,
                                 tb_uid_t suid, int8_t level);
C
Cary Xu 已提交
33 34
static void    tdRSmaFetchTrigger(void *param, void *tmrId);
static void    tdRSmaPersistTrigger(void *param, void *tmrId);
35
static void   *tdRSmaPersistExec(void *param);
C
Cary Xu 已提交
36
static void    tdRSmaQTaskInfoGetFName(int32_t vid, int8_t ftype, char *outputName);
37

C
Cary Xu 已提交
38 39 40
static int32_t tdRSmaQTaskInfoIterInit(SRSmaQTaskInfoIter *pIter, STFile *pTFile);
static int32_t tdRSmaQTaskInfoIterNextBlock(SRSmaQTaskInfoIter *pIter, bool *isFinish);
static int32_t tdRSmaQTaskInfoRestore(SSma *pSma, SRSmaQTaskInfoIter *pIter);
41
static int32_t tdRSmaQTaskInfoItemRestore(SSma *pSma, const SRSmaQTaskInfoItem *infoItem);
42

C
Cary Xu 已提交
43 44 45 46
static int32_t tdRSmaRestoreQTaskInfoInit(SSma *pSma);
static int32_t tdRSmaRestoreQTaskInfoReload(SSma *pSma);
static int32_t tdRSmaRestoreTSDataReload(SSma *pSma);

47 48 49 50 51 52
struct SRSmaInfoItem {
  SRSmaInfo *pRsmaInfo;
  void      *taskInfo;  // qTaskInfo_t
  tmr_h      tmrId;
  int8_t     level;
  int8_t     tmrInitFlag;
C
Cary Xu 已提交
53
  int8_t     triggerStat;
54 55
  int32_t    maxDelay;
};
C
Cary Xu 已提交
56

57
struct SRSmaInfo {
58 59 60 61
  STSchema     *pTSchema;
  SSma         *pSma;
  int64_t       suid;
  SRSmaInfoItem items[TSDB_RETENTION_L2];
62 63
};

64 65 66 67 68 69 70
struct SRSmaQTaskInfoItem {
  int32_t len;
  int8_t  type;
  int64_t suid;
  void   *qTaskInfo;
};

C
Cary Xu 已提交
71
struct SRSmaQTaskInfoIter {
72 73 74 75 76
  STFile *pTFile;
  int64_t offset;
  int64_t fsize;
  int32_t nBytes;
  int32_t nAlloc;
C
Cary Xu 已提交
77
  char   *pBuf;
78
  // ------------
C
Cary Xu 已提交
79
  char   *qBuf;  // for iterator
80 81 82
  int32_t nBufPos;
};

C
Cary Xu 已提交
83
static FORCE_INLINE int32_t tdRSmaQTaskInfoContLen(int32_t lenWithHead) {
C
Cary Xu 已提交
84
  return lenWithHead - RSMA_QTASKINFO_HEAD_LEN;
C
Cary Xu 已提交
85 86
}

C
Cary Xu 已提交
87
static FORCE_INLINE void tdRSmaQTaskInfoIterDestroy(SRSmaQTaskInfoIter *pIter) { taosMemoryFreeClear(pIter->pBuf); }
C
Cary Xu 已提交
88

C
Cary Xu 已提交
89
static FORCE_INLINE void tdFreeTaskHandle(qTaskInfo_t *taskHandle, int32_t vgId, int32_t level) {
90 91 92
  // Note: free/kill may in RC
  qTaskInfo_t otaskHandle = atomic_load_ptr(taskHandle);
  if (otaskHandle && atomic_val_compare_exchange_ptr(taskHandle, otaskHandle, NULL)) {
C
Cary Xu 已提交
93
    smaDebug("vgId:%d, free qTaskInfo_t %p of level %d", vgId, otaskHandle, level);
94
    qDestroyTask(otaskHandle);
C
Cary Xu 已提交
95
  } else {
C
Cary Xu 已提交
96
    smaDebug("vgId:%d, not free qTaskInfo_t %p of level %d", vgId, otaskHandle, level);
97 98 99 100
  }
}

void *tdFreeRSmaInfo(SRSmaInfo *pInfo) {
101
  if (pInfo) {
C
Cary Xu 已提交
102
    for (int32_t i = 0; i < TSDB_RETENTION_L2; ++i) {
103 104
      SRSmaInfoItem *pItem = &pInfo->items[i];
      if (pItem->taskInfo) {
C
Cary Xu 已提交
105 106 107 108 109 110 111
        smaDebug("vgId:%d, stb %" PRIi64 " stop fetch-timer %p level %d", SMA_VID(pInfo->pSma), pInfo->suid,
                 pItem->tmrId, i + 1);
        taosTmrStopA(&pItem->tmrId);
        tdFreeTaskHandle(&pItem->taskInfo, SMA_VID(pInfo->pSma), i + 1);
      } else {
        smaDebug("vgId:%d, stb %" PRIi64 " no need to destroy rsma info level %d since empty taskInfo",
                 SMA_VID(pInfo->pSma), pInfo->suid, i + 1);
112
      }
113
    }
114 115
    taosMemoryFree(pInfo->pTSchema);
    taosMemoryFree(pInfo);
C
Cary Xu 已提交
116 117
  } else {
    smaDebug("vgId:%d, stb %" PRIi64 " no need to destroy rsma info since empty", SMA_VID(pInfo->pSma), pInfo->suid);
118
  }
119

120 121 122 123 124 125 126 127 128 129 130 131 132
  return NULL;
}

static FORCE_INLINE int32_t tdUidStoreInit(STbUidStore **pStore) {
  ASSERT(*pStore == NULL);
  *pStore = taosMemoryCalloc(1, sizeof(STbUidStore));
  if (*pStore == NULL) {
    terrno = TSDB_CODE_OUT_OF_MEMORY;
    return TSDB_CODE_FAILED;
  }
  return TSDB_CODE_SUCCESS;
}

C
Cary Xu 已提交
133
static int32_t tdUpdateTbUidListImpl(SSma *pSma, tb_uid_t *suid, SArray *tbUids) {
134
  SSmaEnv   *pEnv = SMA_RSMA_ENV(pSma);
C
Cary Xu 已提交
135
  SRSmaStat *pStat = (SRSmaStat *)SMA_ENV_STAT(pEnv);
136 137 138 139
  SRSmaInfo *pRSmaInfo = NULL;

  if (!suid || !tbUids) {
    terrno = TSDB_CODE_INVALID_PTR;
S
Shengliang Guan 已提交
140
    smaError("vgId:%d, failed to get rsma info for uid:%" PRIi64 " since %s", SMA_VID(pSma), *suid, terrstr(terrno));
141 142 143
    return TSDB_CODE_FAILED;
  }

C
Cary Xu 已提交
144
  pRSmaInfo = taosHashGet(RSMA_INFO_HASH(pStat), suid, sizeof(tb_uid_t));
145
  if (!pRSmaInfo || !(pRSmaInfo = *(SRSmaInfo **)pRSmaInfo)) {
S
Shengliang Guan 已提交
146
    smaError("vgId:%d, failed to get rsma info for uid:%" PRIi64, SMA_VID(pSma), *suid);
C
Cary Xu 已提交
147
    terrno = TSDB_CODE_RSMA_INVALID_STAT;
148 149 150
    return TSDB_CODE_FAILED;
  }

C
Cary Xu 已提交
151 152 153 154 155 156 157 158
  if (pRSmaInfo->items[0].taskInfo) {
    if ((qUpdateQualifiedTableId(pRSmaInfo->items[0].taskInfo, tbUids, true) < 0)) {
      smaError("vgId:%d, update tbUidList failed for uid:%" PRIi64 " since %s", SMA_VID(pSma), *suid, terrstr(terrno));
      return TSDB_CODE_FAILED;
    } else {
      smaDebug("vgId:%d, update tbUidList succeed for qTaskInfo:%p with suid:%" PRIi64 ", uid:%" PRIi64, SMA_VID(pSma),
               pRSmaInfo->items[0].taskInfo, *suid, *(int64_t *)taosArrayGet(tbUids, 0));
    }
159 160
  }

C
Cary Xu 已提交
161 162 163 164 165 166 167 168
  if (pRSmaInfo->items[1].taskInfo) {
    if ((qUpdateQualifiedTableId(pRSmaInfo->items[1].taskInfo, tbUids, true) < 0)) {
      smaError("vgId:%d, update tbUidList failed for uid:%" PRIi64 " since %s", SMA_VID(pSma), *suid, terrstr(terrno));
      return TSDB_CODE_FAILED;
    } else {
      smaDebug("vgId:%d, update tbUidList succeed for qTaskInfo:%p with suid:%" PRIi64 ", uid:%" PRIi64, SMA_VID(pSma),
               pRSmaInfo->items[1].taskInfo, *suid, *(int64_t *)taosArrayGet(tbUids, 0));
    }
169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214
  }

  return TSDB_CODE_SUCCESS;
}

int32_t tdUpdateTbUidList(SSma *pSma, STbUidStore *pStore) {
  if (!pStore || (taosArrayGetSize(pStore->tbUids) == 0)) {
    return TSDB_CODE_SUCCESS;
  }

  if (tdUpdateTbUidListImpl(pSma, &pStore->suid, pStore->tbUids) != TSDB_CODE_SUCCESS) {
    return TSDB_CODE_FAILED;
  }

  void *pIter = taosHashIterate(pStore->uidHash, NULL);
  while (pIter) {
    tb_uid_t *pTbSuid = (tb_uid_t *)taosHashGetKey(pIter, NULL);
    SArray   *pTbUids = *(SArray **)pIter;

    if (tdUpdateTbUidListImpl(pSma, pTbSuid, pTbUids) != TSDB_CODE_SUCCESS) {
      taosHashCancelIterate(pStore->uidHash, pIter);
      return TSDB_CODE_FAILED;
    }

    pIter = taosHashIterate(pStore->uidHash, pIter);
  }
  return TSDB_CODE_SUCCESS;
}

/**
 * @brief fetch suid/uids when create child tables of rollup SMA
 *
 * @param pTsdb
 * @param ppStore
 * @param suid
 * @param uid
 * @return int32_t
 */
int32_t tdFetchTbUidList(SSma *pSma, STbUidStore **ppStore, tb_uid_t suid, tb_uid_t uid) {
  SSmaEnv *pEnv = SMA_RSMA_ENV(pSma);

  // only applicable to rollup SMA ctables
  if (!pEnv) {
    return TSDB_CODE_SUCCESS;
  }

C
Cary Xu 已提交
215 216 217
  SRSmaStat *pStat = (SRSmaStat *)SMA_ENV_STAT(pEnv);
  SHashObj  *infoHash = NULL;
  if (!pStat || !(infoHash = RSMA_INFO_HASH(pStat))) {
C
Cary Xu 已提交
218
    terrno = TSDB_CODE_RSMA_INVALID_STAT;
219 220 221 222 223 224 225 226 227 228 229
    return TSDB_CODE_FAILED;
  }

  // info cached when create rsma stable and return directly for non-rsma ctables
  if (!taosHashGet(infoHash, &suid, sizeof(tb_uid_t))) {
    return TSDB_CODE_SUCCESS;
  }

  ASSERT(ppStore != NULL);

  if (!(*ppStore)) {
230
    if (tdUidStoreInit(ppStore) < 0) {
231 232 233 234
      return TSDB_CODE_FAILED;
    }
  }

235
  if (tdUidStorePut(*ppStore, suid, &uid) < 0) {
236 237 238 239 240 241 242
    *ppStore = tdUidStoreFree(*ppStore);
    return TSDB_CODE_FAILED;
  }

  return TSDB_CODE_SUCCESS;
}

C
Cary Xu 已提交
243 244 245 246 247 248
static int32_t tdSetRSmaInfoItemParams(SSma *pSma, SRSmaParam *param, SRSmaInfo *pRSmaInfo, SReadHandle *pReadHandle,
                                       int8_t idx) {
  SRetention *pRetention = SMA_RETENTION(pSma);
  STsdbCfg   *pTsdbCfg = SMA_TSDB_CFG(pSma);

  if (param->qmsg[idx]) {
C
Cary Xu 已提交
249 250
    SRSmaInfoItem *pItem = &(pRSmaInfo->items[idx]);
    pItem->pRsmaInfo = pRSmaInfo;
C
Cary Xu 已提交
251
    pItem->taskInfo = qCreateStreamExecTaskInfo(param->qmsg[idx], pReadHandle);
C
Cary Xu 已提交
252
    if (!pItem->taskInfo) {
C
Cary Xu 已提交
253 254
      goto _err;
    }
C
Cary Xu 已提交
255
    pItem->triggerStat = TASK_TRIGGER_STAT_INACTIVE;
C
Cary Xu 已提交
256 257 258
    if (param->maxdelay[idx] < TSDB_MIN_ROLLUP_MAX_DELAY) {
      int64_t msInterval =
          convertTimeFromPrecisionToUnit(pRetention[idx + 1].freq, pTsdbCfg->precision, TIME_UNIT_MILLISECOND);
C
Cary Xu 已提交
259
      pItem->maxDelay = (int32_t)msInterval;
C
Cary Xu 已提交
260
    } else {
C
Cary Xu 已提交
261
      pItem->maxDelay = (int32_t)param->maxdelay[idx];
C
Cary Xu 已提交
262
    }
C
Cary Xu 已提交
263 264
    if (pItem->maxDelay > TSDB_MAX_ROLLUP_MAX_DELAY) {
      pItem->maxDelay = TSDB_MAX_ROLLUP_MAX_DELAY;
C
Cary Xu 已提交
265
    }
C
Cary Xu 已提交
266
    pItem->level = (idx == 0 ? TSDB_RETENTION_L1 : TSDB_RETENTION_L2);
C
Cary Xu 已提交
267 268 269 270 271 272
  }
  return TSDB_CODE_SUCCESS;
_err:
  return TSDB_CODE_FAILED;
}

273
/**
274
 * @brief for rsam create or restore
275
 *
276 277 278 279
 * @param pSma
 * @param param
 * @param suid
 * @param tbName
280 281
 * @return int32_t
 */
282 283 284 285
int32_t tdProcessRSmaCreateImpl(SSma *pSma, SRSmaParam *param, int64_t suid, const char *tbName) {
  SVnode *pVnode = pSma->pVnode;
  SMeta  *pMeta = pVnode->pMeta;
  SMsgCb *pMsgCb = &pVnode->msgCb;
286

287
  if ((param->qmsgLen[0] == 0) && (param->qmsgLen[1] == 0)) {
288
    smaDebug("vgId:%d, no qmsg1/qmsg2 for rollup table %s %" PRIi64, SMA_VID(pSma), tbName, suid);
289 290 291
    return TSDB_CODE_SUCCESS;
  }

C
Cary Xu 已提交
292
  if (tdCheckAndInitSmaEnv(pSma, TSDB_SMA_TYPE_ROLLUP) != TSDB_CODE_SUCCESS) {
293 294 295 296 297
    terrno = TSDB_CODE_TDB_INIT_FAILED;
    return TSDB_CODE_FAILED;
  }

  SSmaEnv   *pEnv = SMA_RSMA_ENV(pSma);
C
Cary Xu 已提交
298
  SRSmaStat *pStat = (SRSmaStat *)SMA_ENV_STAT(pEnv);
299 300
  SRSmaInfo *pRSmaInfo = NULL;

301
  pRSmaInfo = taosHashGet(RSMA_INFO_HASH(pStat), &suid, sizeof(tb_uid_t));
302
  if (pRSmaInfo) {
C
Cary Xu 已提交
303
    ASSERT(0);  // TODO: free original pRSmaInfo if exists abnormally
304
    smaDebug("vgId:%d, rsma info already exists for table %s, %" PRIi64, SMA_VID(pSma), tbName, suid);
305 306 307
    return TSDB_CODE_SUCCESS;
  }

308
  // from write queue: single thead
309 310 311 312 313 314 315 316 317
  pRSmaInfo = (SRSmaInfo *)taosMemoryCalloc(1, sizeof(SRSmaInfo));
  if (!pRSmaInfo) {
    terrno = TSDB_CODE_OUT_OF_MEMORY;
    return TSDB_CODE_FAILED;
  }

  STqReadHandle *pReadHandle = tqInitSubmitMsgScanner(pMeta);
  if (!pReadHandle) {
    terrno = TSDB_CODE_OUT_OF_MEMORY;
318
    goto _err;
319 320 321 322 323 324
  }

  SReadHandle handle = {
      .reader = pReadHandle,
      .meta = pMeta,
      .pMsgCb = pMsgCb,
5
54liuyao 已提交
325
      .vnode = pVnode,
326 327
  };

328
  STSchema *pTSchema = metaGetTbTSchema(SMA_META(pSma), suid, -1);
329 330 331 332 333 334
  if (!pTSchema) {
    terrno = TSDB_CODE_TDB_IVD_TB_SCHEMA_VERSION;
    goto _err;
  }
  pRSmaInfo->pTSchema = pTSchema;
  pRSmaInfo->pSma = pSma;
335
  pRSmaInfo->suid = suid;
336

C
Cary Xu 已提交
337 338 339
  if (tdSetRSmaInfoItemParams(pSma, param, pRSmaInfo, &handle, 0) < 0) {
    goto _err;
  }
C
Cary Xu 已提交
340

C
Cary Xu 已提交
341 342 343
  if (tdSetRSmaInfoItemParams(pSma, param, pRSmaInfo, &handle, 1) < 0) {
    goto _err;
  }
344

345
  if (taosHashPut(RSMA_INFO_HASH(pStat), &suid, sizeof(tb_uid_t), &pRSmaInfo, sizeof(pRSmaInfo)) < 0) {
346
    goto _err;
347 348
  }

C
Cary Xu 已提交
349 350
  smaDebug("vgId:%d, register rsma info succeed for suid:%" PRIi64, SMA_VID(pSma), suid);

C
Cary Xu 已提交
351 352 353
  // start the persist timer
  if (TASK_TRIGGER_STAT_INIT ==
      atomic_val_compare_exchange_8(RSMA_TRIGGER_STAT(pStat), TASK_TRIGGER_STAT_INIT, TASK_TRIGGER_STAT_ACTIVE)) {
354
    taosTmrStart(tdRSmaPersistTrigger, RSMA_QTASKINFO_PERSIST_MS, pStat, RSMA_TMR_HANDLE(pStat));
C
Cary Xu 已提交
355 356
  }

357
  return TSDB_CODE_SUCCESS;
358 359 360 361
_err:
  tdFreeRSmaInfo(pRSmaInfo);
  taosMemoryFree(pReadHandle);
  return TSDB_CODE_FAILED;
362 363
}

364
/**
C
Cary Xu 已提交
365
 * @brief Check and init qTaskInfo_t, only applicable to stable with SRSmaParam currently
366
 *
C
Cary Xu 已提交
367
 * @param pVnode
368 369 370 371 372 373 374 375 376 377 378 379 380
 * @param pReq
 * @return int32_t
 */
int32_t tdProcessRSmaCreate(SVnode *pVnode, SVCreateStbReq *pReq) {
  SSma *pSma = pVnode->pSma;
  if (!pReq->rollup) {
    smaTrace("vgId:%d, return directly since no rollup for stable %s %" PRIi64, SMA_VID(pSma), pReq->name, pReq->suid);
    return TSDB_CODE_SUCCESS;
  }

  return tdProcessRSmaCreateImpl(pSma, &pReq->rsmaParam, pReq->suid, pReq->name);
}

381 382 383 384 385 386 387 388 389 390 391 392 393 394 395 396 397 398 399 400 401 402 403 404 405 406 407 408 409 410 411 412 413 414 415 416 417 418 419 420 421 422 423 424 425 426 427
/**
 * @brief store suid/[uids], prefer to use array and then hash
 *
 * @param pStore
 * @param suid
 * @param uid
 * @return int32_t
 */
static int32_t tdUidStorePut(STbUidStore *pStore, tb_uid_t suid, tb_uid_t *uid) {
  // prefer to store suid/uids in array
  if ((suid == pStore->suid) || (pStore->suid == 0)) {
    if (pStore->suid == 0) {
      pStore->suid = suid;
    }
    if (uid) {
      if (!pStore->tbUids) {
        if (!(pStore->tbUids = taosArrayInit(1, sizeof(tb_uid_t)))) {
          terrno = TSDB_CODE_OUT_OF_MEMORY;
          return TSDB_CODE_FAILED;
        }
      }
      if (!taosArrayPush(pStore->tbUids, uid)) {
        return TSDB_CODE_FAILED;
      }
    }
  } else {
    // store other suid/uids in hash when multiple stable/table included in 1 batch of request
    if (!pStore->uidHash) {
      pStore->uidHash = taosHashInit(4, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BIGINT), false, HASH_ENTRY_LOCK);
      if (!pStore->uidHash) {
        return TSDB_CODE_FAILED;
      }
    }
    if (uid) {
      SArray *uidArray = taosHashGet(pStore->uidHash, &suid, sizeof(tb_uid_t));
      if (uidArray && ((uidArray = *(SArray **)uidArray))) {
        taosArrayPush(uidArray, uid);
      } else {
        SArray *pUidArray = taosArrayInit(1, sizeof(tb_uid_t));
        if (!pUidArray) {
          terrno = TSDB_CODE_OUT_OF_MEMORY;
          return TSDB_CODE_FAILED;
        }
        if (!taosArrayPush(pUidArray, uid)) {
          terrno = TSDB_CODE_OUT_OF_MEMORY;
          return TSDB_CODE_FAILED;
        }
428
        if (taosHashPut(pStore->uidHash, &suid, sizeof(suid), &pUidArray, sizeof(pUidArray)) < 0) {
429 430 431 432
          return TSDB_CODE_FAILED;
        }
      }
    } else {
433
      if (taosHashPut(pStore->uidHash, &suid, sizeof(suid), NULL, 0) < 0) {
434 435 436 437 438 439 440 441 442 443 444 445 446 447 448 449 450 451 452 453 454 455 456 457 458 459 460 461 462 463 464 465 466 467 468 469 470 471 472 473 474 475 476 477 478 479 480 481 482 483 484 485 486 487 488 489 490 491 492 493 494 495 496 497 498 499 500 501 502
        return TSDB_CODE_FAILED;
      }
    }
  }
  return TSDB_CODE_SUCCESS;
}

void tdUidStoreDestory(STbUidStore *pStore) {
  if (pStore) {
    if (pStore->uidHash) {
      if (pStore->tbUids) {
        // When pStore->tbUids not NULL, the pStore->uidHash has k/v; otherwise pStore->uidHash only has keys.
        void *pIter = taosHashIterate(pStore->uidHash, NULL);
        while (pIter) {
          SArray *arr = *(SArray **)pIter;
          taosArrayDestroy(arr);
          pIter = taosHashIterate(pStore->uidHash, pIter);
        }
      }
      taosHashCleanup(pStore->uidHash);
    }
    taosArrayDestroy(pStore->tbUids);
  }
}

void *tdUidStoreFree(STbUidStore *pStore) {
  if (pStore) {
    tdUidStoreDestory(pStore);
    taosMemoryFree(pStore);
  }
  return NULL;
}

static int32_t tdProcessSubmitReq(STsdb *pTsdb, int64_t version, void *pReq) {
  if (!pReq) {
    terrno = TSDB_CODE_INVALID_PTR;
    return TSDB_CODE_FAILED;
  }

  SSubmitReq *pSubmitReq = (SSubmitReq *)pReq;

  if (tsdbInsertData(pTsdb, version, pSubmitReq, NULL) < 0) {
    return TSDB_CODE_FAILED;
  }

  return TSDB_CODE_SUCCESS;
}

static int32_t tdFetchSubmitReqSuids(SSubmitReq *pMsg, STbUidStore *pStore) {
  ASSERT(pMsg != NULL);
  SSubmitMsgIter msgIter = {0};
  SSubmitBlk    *pBlock = NULL;
  SSubmitBlkIter blkIter = {0};
  STSRow        *row = NULL;

  terrno = TSDB_CODE_SUCCESS;

  if (tInitSubmitMsgIter(pMsg, &msgIter) < 0) return -1;
  while (true) {
    if (tGetSubmitMsgNext(&msgIter, &pBlock) < 0) return -1;

    if (!pBlock) break;
    tdUidStorePut(pStore, msgIter.suid, NULL);
  }

  if (terrno != TSDB_CODE_SUCCESS) return -1;
  return 0;
}

C
Cary Xu 已提交
503 504 505 506 507 508 509 510 511 512
static void tdDestroySDataBlockArray(SArray *pArray) {
#if 0
  for (int32_t i = 0; i < taosArrayGetSize(pArray); ++i) {
    SSDataBlock *pDataBlock = taosArrayGet(pArray, i);
    blockDestroyInner(pDataBlock);
  }
#endif
  taosArrayDestroy(pArray);
}

513 514 515 516
static int32_t tdFetchAndSubmitRSmaResult(SRSmaInfoItem *pItem, int8_t blkType) {
  SArray    *pResult = NULL;
  SRSmaInfo *pRSmaInfo = pItem->pRsmaInfo;
  SSma      *pSma = pRSmaInfo->pSma;
517 518 519 520

  while (1) {
    SSDataBlock *output = NULL;
    uint64_t     ts;
521
    if (qExecTask(pItem->taskInfo, &output, &ts) < 0) {
522 523 524 525 526
      ASSERT(false);
    }
    if (!output) {
      break;
    }
C
Cary Xu 已提交
527

528
    if (!pResult) {
C
Cary Xu 已提交
529
      pResult = taosArrayInit(1, sizeof(SSDataBlock));
530 531 532 533 534 535 536 537 538 539
      if (!pResult) {
        terrno = TSDB_CODE_OUT_OF_MEMORY;
        return TSDB_CODE_FAILED;
      }
    }

    taosArrayPush(pResult, output);
  }

  if (taosArrayGetSize(pResult) > 0) {
C
Cary Xu 已提交
540
#if 1
C
Cary Xu 已提交
541
    char flag[10] = {0};
542
    snprintf(flag, 10, "level %" PRIi8, pItem->level);
S
slzhou 已提交
543
    blockDebugShowDataBlocks(pResult, flag);
C
Cary Xu 已提交
544
#endif
545
    STsdb      *sinkTsdb = (pItem->level == TSDB_RETENTION_L1 ? pSma->pRSmaTsdb1 : pSma->pRSmaTsdb2);
546
    SSubmitReq *pReq = NULL;
C
Cary Xu 已提交
547
    // TODO: the schema update should be handled
548
    if (buildSubmitReqFromDataBlock(&pReq, pResult, pRSmaInfo->pTSchema, SMA_VID(pSma), pRSmaInfo->suid) < 0) {
C
Cary Xu 已提交
549
      goto _err;
550
    }
551

552
    if (pReq && tdProcessSubmitReq(sinkTsdb, INT64_MAX, pReq) < 0) {
553
      taosMemoryFreeClear(pReq);
C
Cary Xu 已提交
554
      goto _err;
555
    }
556

557
    taosMemoryFreeClear(pReq);
C
Cary Xu 已提交
558 559
  } else if (terrno == 0) {
    smaDebug("vgId:%d, no rsma %" PRIi8 " data fetched yet", SMA_VID(pSma), pItem->level);
560
  } else {
C
Cary Xu 已提交
561
    smaDebug("vgId:%d, no rsma %" PRIi8 " data fetched since %s", SMA_VID(pSma), pItem->level, tstrerror(terrno));
562 563
  }

C
Cary Xu 已提交
564
  tdDestroySDataBlockArray(pResult);
C
Cary Xu 已提交
565 566
  return TSDB_CODE_SUCCESS;
_err:
C
Cary Xu 已提交
567
  tdDestroySDataBlockArray(pResult);
C
Cary Xu 已提交
568
  return TSDB_CODE_FAILED;
569 570 571 572 573 574 575 576
}

/**
 * @brief trigger to get rsma result
 *
 * @param param
 * @param tmrId
 */
C
Cary Xu 已提交
577
static void tdRSmaFetchTrigger(void *param, void *tmrId) {
578
  SRSmaInfoItem *pItem = param;
C
Cary Xu 已提交
579 580
  SSma          *pSma = pItem->pRsmaInfo->pSma;
  SRSmaStat     *pStat = (SRSmaStat *)SMA_ENV_STAT((SSmaEnv *)pSma->pRSmaEnv);
581

C
Cary Xu 已提交
582 583
  int8_t rsmaTriggerStat = atomic_load_8(RSMA_TRIGGER_STAT(pStat));
  if (rsmaTriggerStat == TASK_TRIGGER_STAT_CANCELLED || rsmaTriggerStat == TASK_TRIGGER_STAT_FINISHED) {
C
Cary Xu 已提交
584 585
    smaDebug("vgId:%d, level %" PRIi8 " not fetch since stat is cancelled for table suid:%" PRIi64, SMA_VID(pSma),
             pItem->level, pItem->pRsmaInfo->suid);
C
Cary Xu 已提交
586 587
    return;
  }
588

C
Cary Xu 已提交
589 590 591
  int8_t fetchTriggerStat =
      atomic_val_compare_exchange_8(&pItem->triggerStat, TASK_TRIGGER_STAT_ACTIVE, TASK_TRIGGER_STAT_INACTIVE);
  if (fetchTriggerStat == TASK_TRIGGER_STAT_ACTIVE) {
C
Cary Xu 已提交
592 593
    smaDebug("vgId:%d, level %" PRIi8 " stat is active for table suid:%" PRIi64, SMA_VID(pSma), pItem->level,
             pItem->pRsmaInfo->suid);
C
Cary Xu 已提交
594 595

    tdRefSmaStat(pSma, (SSmaStat *)pStat);
596

C
Cary Xu 已提交
597 598
    SSDataBlock dataBlock = {.info.type = STREAM_GET_ALL};
    qSetStreamInput(pItem->taskInfo, &dataBlock, STREAM_DATA_TYPE_SSDATA_BLOCK, false);
599
    tdFetchAndSubmitRSmaResult(pItem, STREAM_DATA_TYPE_SSDATA_BLOCK);
C
Cary Xu 已提交
600 601 602

    tdUnRefSmaStat(pSma, (SSmaStat *)pStat);

603
  } else {
C
Cary Xu 已提交
604 605
    smaDebug("vgId:%d, level %" PRIi8 " stat is inactive for table suid:%" PRIi64, SMA_VID(pSma), pItem->level,
             pItem->pRsmaInfo->suid);
606 607 608
  }
}

C
Cary Xu 已提交
609 610
static int32_t tdExecuteRSmaImpl(SSma *pSma, const void *pMsg, int32_t inputType, SRSmaInfoItem *pItem, tb_uid_t suid,
                                 int8_t level) {
611 612 613 614 615 616 617 618
  if (!pItem || !pItem->taskInfo) {
    smaDebug("vgId:%d, no qTaskInfo to execute rsma %" PRIi8 " task for suid:%" PRIu64, SMA_VID(pSma), level, suid);
    return TSDB_CODE_SUCCESS;
  }

  smaDebug("vgId:%d, execute rsma %" PRIi8 " task for qTaskInfo:%p suid:%" PRIu64, SMA_VID(pSma), level,
           pItem->taskInfo, suid);

C
Cary Xu 已提交
619
  if (qSetStreamInput(pItem->taskInfo, pMsg, inputType, true) < 0) {  // STREAM_DATA_TYPE_SUBMIT_BLOCK
620 621 622 623 624
    smaError("vgId:%d, rsma % " PRIi8 " qSetStreamInput failed since %s", SMA_VID(pSma), level, tstrerror(terrno));
    return TSDB_CODE_FAILED;
  }

  tdFetchAndSubmitRSmaResult(pItem, STREAM_DATA_TYPE_SUBMIT_BLOCK);
C
Cary Xu 已提交
625
  atomic_store_8(&pItem->triggerStat, TASK_TRIGGER_STAT_ACTIVE);
C
Cary Xu 已提交
626
  smaDebug("vgId:%d, process rsma insert", SMA_VID(pSma));
C
Cary Xu 已提交
627 628 629 630

  SSmaEnv   *pEnv = SMA_RSMA_ENV(pSma);
  SRSmaStat *pStat = SMA_RSMA_STAT(pEnv->pStat);

C
Cary Xu 已提交
631 632 633 634 635
  if (pStat->tmrHandle) {
    taosTmrReset(tdRSmaFetchTrigger, pItem->maxDelay, pItem, pStat->tmrHandle, &pItem->tmrId);
  } else {
    ASSERT(0);
  }
636 637 638 639

  return TSDB_CODE_SUCCESS;
}

640
static int32_t tdExecuteRSma(SSma *pSma, const void *pMsg, int32_t inputType, tb_uid_t suid) {
641 642 643 644 645 646
  SSmaEnv *pEnv = SMA_RSMA_ENV(pSma);
  if (!pEnv) {
    // only applicable when rsma env exists
    return TSDB_CODE_SUCCESS;
  }

C
Cary Xu 已提交
647
  SRSmaStat *pStat = (SRSmaStat *)SMA_ENV_STAT(pEnv);
648 649
  SRSmaInfo *pRSmaInfo = NULL;

C
Cary Xu 已提交
650
  pRSmaInfo = taosHashGet(RSMA_INFO_HASH(pStat), &suid, sizeof(tb_uid_t));
651 652

  if (!pRSmaInfo || !(pRSmaInfo = *(SRSmaInfo **)pRSmaInfo)) {
653
    smaDebug("vgId:%d, return as no rsma info for suid:%" PRIu64, SMA_VID(pSma), suid);
654 655
    return TSDB_CODE_SUCCESS;
  }
656 657 658

  if (!pRSmaInfo->items[0].taskInfo) {
    smaDebug("vgId:%d, return as no rsma qTaskInfo for suid:%" PRIu64, SMA_VID(pSma), suid);
659 660 661 662
    return TSDB_CODE_SUCCESS;
  }

  if (inputType == STREAM_DATA_TYPE_SUBMIT_BLOCK) {
663 664
    tdExecuteRSmaImpl(pSma, pMsg, inputType, &pRSmaInfo->items[0], suid, TSDB_RETENTION_L1);
    tdExecuteRSmaImpl(pSma, pMsg, inputType, &pRSmaInfo->items[1], suid, TSDB_RETENTION_L2);
665 666 667 668 669 670 671 672 673 674 675 676
  }

  return TSDB_CODE_SUCCESS;
}

int32_t tdProcessRSmaSubmit(SSma *pSma, void *pMsg, int32_t inputType) {
  SSmaEnv *pEnv = SMA_RSMA_ENV(pSma);
  if (!pEnv) {
    // only applicable when rsma env exists
    return TSDB_CODE_SUCCESS;
  }

677 678 679 680 681 682
  SRetention *pRetention = SMA_RETENTION(pSma);
  if (!RETENTION_VALID(pRetention + 1)) {
    // return directly if retention level 1 is invalid
    return TSDB_CODE_SUCCESS;
  }

683 684 685 686 687
  if (inputType == STREAM_DATA_TYPE_SUBMIT_BLOCK) {
    STbUidStore uidStore = {0};
    tdFetchSubmitReqSuids(pMsg, &uidStore);

    if (uidStore.suid != 0) {
688
      tdExecuteRSma(pSma, pMsg, inputType, uidStore.suid);
689 690 691 692

      void *pIter = taosHashIterate(uidStore.uidHash, NULL);
      while (pIter) {
        tb_uid_t *pTbSuid = (tb_uid_t *)taosHashGetKey(pIter, NULL);
693
        tdExecuteRSma(pSma, pMsg, inputType, *pTbSuid);
694 695 696 697 698 699 700 701
        pIter = taosHashIterate(uidStore.uidHash, pIter);
      }

      tdUidStoreDestory(&uidStore);
    }
  }
  return TSDB_CODE_SUCCESS;
}
C
Cary Xu 已提交
702

C
Cary Xu 已提交
703
static int32_t tdRSmaRestoreQTaskInfoInit(SSma *pSma) {
704 705 706 707
  SVnode *pVnode = pSma->pVnode;

  SArray *suidList = taosArrayInit(1, sizeof(tb_uid_t));
  if (tsdbGetStbIdList(SMA_META(pSma), 0, suidList) < 0) {
C
Cary Xu 已提交
708 709
    taosArrayDestroy(suidList);
    smaError("vgId:%d, failed to restore rsma env since get stb id list error: %s", TD_VID(pVnode), terrstr());
710 711 712
    return TSDB_CODE_FAILED;
  }

C
Cary Xu 已提交
713 714 715 716
  int32_t arrSize = taosArrayGetSize(suidList);
  if (arrSize == 0) {
    taosArrayDestroy(suidList);
    smaDebug("vgId:%d, no need to restore rsma env since empty stb id list", TD_VID(pVnode));
717 718 719 720 721
    return TSDB_CODE_SUCCESS;
  }

  SMetaReader mr = {0};
  metaReaderInit(&mr, SMA_META(pSma), 0);
C
Cary Xu 已提交
722
  for (int32_t i = 0; i < arrSize; ++i) {
723
    tb_uid_t suid = *(tb_uid_t *)taosArrayGet(suidList, i);
C
Cary Xu 已提交
724
    smaDebug("vgId:%d, rsma restore, suid[%d] is %" PRIi64, TD_VID(pVnode), i, suid);
725
    if (metaGetTableEntryByUid(&mr, suid) < 0) {
C
Cary Xu 已提交
726 727
      smaError("vgId:%d, rsma restore, failed to get table meta for %" PRIi64 " since %s", TD_VID(pVnode), suid,
               terrstr());
728 729 730 731 732 733
      goto _err;
    }
    ASSERT(mr.me.type == TSDB_SUPER_TABLE);
    ASSERT(mr.me.uid == suid);
    if (TABLE_IS_ROLLUP(mr.me.flags)) {
      SRSmaParam *param = &mr.me.stbEntry.rsmaParam;
C
Cary Xu 已提交
734 735 736 737
      for (int i = 0; i < TSDB_RETENTION_L2; ++i) {
        smaDebug("vgId:%d, rsma restore, table:%" PRIi64 " level:%d, maxdelay:%" PRIi64 " watermark:%" PRIi64
                 " qmsgLen:%" PRIi32,
                 TD_VID(pVnode), suid, i, param->maxdelay[i], param->watermark[i], param->qmsgLen[i]);
738 739
      }
      if (tdProcessRSmaCreateImpl(pSma, &mr.me.stbEntry.rsmaParam, suid, mr.me.name) < 0) {
C
Cary Xu 已提交
740
        smaError("vgId:%d, rsma restore env failed for %" PRIi64 " since %s", TD_VID(pVnode), suid, terrstr());
741 742
        goto _err;
      }
C
Cary Xu 已提交
743
      smaDebug("vgId:%d, rsma restore env success for %" PRIi64, TD_VID(pVnode), suid);
744 745 746
    }
  }

C
Cary Xu 已提交
747 748 749 750 751 752 753 754 755 756 757 758 759
  metaReaderClear(&mr);
  taosArrayDestroy(suidList);

  return TSDB_CODE_SUCCESS;
_err:
  metaReaderClear(&mr);
  taosArrayDestroy(suidList);

  return TSDB_CODE_FAILED;
}

static int32_t tdRSmaRestoreQTaskInfoReload(SSma *pSma) {
  SVnode *pVnode = pSma->pVnode;
C
Cary Xu 已提交
760 761
  STFile  tFile = {0};
  char    qTaskInfoFName[TSDB_FILENAME_LEN];
762

C
Cary Xu 已提交
763
  tdRSmaQTaskInfoGetFName(TD_VID(pVnode), TD_QTASK_CUR_FILE, qTaskInfoFName);
764 765 766
  if (tdInitTFile(&tFile, pVnode->pTfs, qTaskInfoFName) < 0) {
    goto _err;
  }
C
Cary Xu 已提交
767 768

  if (!taosCheckExistFile(TD_TFILE_FULL_NAME(&tFile))) {
769 770 771
    return TSDB_CODE_SUCCESS;
  }

772 773 774
  if (tdOpenTFile(&tFile, TD_FILE_READ) < 0) {
    goto _err;
  }
C
Cary Xu 已提交
775

C
Cary Xu 已提交
776
  SRSmaQTaskInfoIter fIter = {0};
777
  if (tdRSmaQTaskInfoIterInit(&fIter, &tFile) < 0) {
C
Cary Xu 已提交
778 779
    tdRSmaQTaskInfoIterDestroy(&fIter);
    tdCloseTFile(&tFile);
780 781
    goto _err;
  }
C
Cary Xu 已提交
782

C
Cary Xu 已提交
783
  if (tdRSmaQTaskInfoRestore(pSma, &fIter) < 0) {
C
Cary Xu 已提交
784 785
    tdRSmaQTaskInfoIterDestroy(&fIter);
    tdCloseTFile(&tFile);
C
Cary Xu 已提交
786 787
    goto _err;
  }
788

C
Cary Xu 已提交
789
  tdRSmaQTaskInfoIterDestroy(&fIter);
C
Cary Xu 已提交
790 791 792
  tdCloseTFile(&tFile);
  return TSDB_CODE_SUCCESS;
_err:
C
Cary Xu 已提交
793 794 795 796 797 798 799 800 801 802 803 804 805 806 807 808 809 810 811 812 813 814 815 816 817 818 819 820 821 822 823 824 825 826 827
  smaError("failed to restore rsma task since %s", terrstr());
  return TSDB_CODE_FAILED;
}

/**
 * @brief reload ts data from checkpoint
 * 
 * @param pSma 
 * @return int32_t 
 */
static int32_t tdRSmaRestoreTSDataReload(SSma *pSma) {
  // TODO
  return TSDB_CODE_SUCCESS;
_err:
  return TSDB_CODE_FAILED;
}

int32_t tdProcessRSmaRestoreImpl(SSma *pSma) {
  SVnode *pVnode = pSma->pVnode;

  // step 1: iterate all stables to restore the rsma env
  if (tdRSmaRestoreQTaskInfoInit(pSma) < 0) {
    goto _err;
  }

  // step 2: retrieve qtaskinfo items from the persistence file(rsma/qtaskinfo) and restore
  if (tdRSmaRestoreQTaskInfoReload(pSma) < 0) {
    goto _err;
  }

  // step 3: reload ts data from checkpoint
  if (tdRSmaRestoreTSDataReload(pSma) < 0) {
    goto _err;
  }

828 829
  return TSDB_CODE_SUCCESS;
_err:
C
Cary Xu 已提交
830
  smaError("failed to restore rsma task since %s", terrstr());
831 832 833
  return TSDB_CODE_FAILED;
}

C
Cary Xu 已提交
834
static int32_t tdRSmaQTaskInfoItemRestore(SSma *pSma, const SRSmaQTaskInfoItem *pItem) {
835 836 837 838
  SRSmaStat *pStat = (SRSmaStat *)SMA_ENV_STAT((SSmaEnv *)pSma->pRSmaEnv);
  SRSmaInfo *pRSmaInfo = NULL;
  void      *qTaskInfo = NULL;

C
Cary Xu 已提交
839
  pRSmaInfo = taosHashGet(RSMA_INFO_HASH(pStat), &pItem->suid, sizeof(pItem->suid));
840 841

  if (!pRSmaInfo || !(pRSmaInfo = *(SRSmaInfo **)pRSmaInfo)) {
C
Cary Xu 已提交
842
    smaDebug("vgId:%d, no restore as no rsma info for table:%" PRIu64, SMA_VID(pSma), pItem->suid);
843 844 845
    return TSDB_CODE_SUCCESS;
  }

C
Cary Xu 已提交
846
  if (pItem->type == 1) {
847
    qTaskInfo = pRSmaInfo->items[0].taskInfo;
C
Cary Xu 已提交
848
  } else if (pItem->type == 2) {
849 850 851 852 853 854
    qTaskInfo = pRSmaInfo->items[1].taskInfo;
  } else {
    ASSERT(0);
  }

  if (!qTaskInfo) {
C
Cary Xu 已提交
855
    smaDebug("vgId:%d, no restore as NULL rsma qTaskInfo for table:%" PRIu64, SMA_VID(pSma), pItem->suid);
856 857 858
    return TSDB_CODE_SUCCESS;
  }

C
Cary Xu 已提交
859 860 861
  if (qDeserializeTaskStatus(qTaskInfo, pItem->qTaskInfo, pItem->len) < 0) {
    smaError("vgId:%d, restore rsma task failed for table:%" PRIi64 " level %d since %s", SMA_VID(pSma), pItem->suid,
             pItem->type, terrstr(terrno));
862 863
    return TSDB_CODE_FAILED;
  }
C
Cary Xu 已提交
864
  smaDebug("vgId:%d, restore rsma task success for table:%" PRIi64 " level %d", SMA_VID(pSma), pItem->suid, pItem->type);
865 866 867 868

  return TSDB_CODE_SUCCESS;
}

C
Cary Xu 已提交
869
static int32_t tdRSmaQTaskInfoIterInit(SRSmaQTaskInfoIter *pIter, STFile *pTFile) {
870 871 872 873 874 875 876 877 878 879 880 881 882 883 884 885 886 887
  memset(pIter, 0, sizeof(*pIter));
  pIter->pTFile = pTFile;
  pIter->offset = TD_FILE_HEAD_SIZE;

  if (tdGetTFileSize(pTFile, &pIter->fsize) < 0) {
    return TSDB_CODE_FAILED;
  }

  if ((pIter->fsize - TD_FILE_HEAD_SIZE) < RSMA_QTASKINFO_BUFSIZE) {
    pIter->nAlloc = pIter->fsize - TD_FILE_HEAD_SIZE;
  } else {
    pIter->nAlloc = RSMA_QTASKINFO_BUFSIZE;
  }

  if (pIter->nAlloc < TD_FILE_HEAD_SIZE) {
    pIter->nAlloc = TD_FILE_HEAD_SIZE;
  }

C
Cary Xu 已提交
888 889
  pIter->pBuf = taosMemoryMalloc(pIter->nAlloc);
  if (!pIter->pBuf) {
890 891 892
    terrno = TSDB_CODE_OUT_OF_MEMORY;
    return TSDB_CODE_FAILED;
  }
C
Cary Xu 已提交
893
  pIter->qBuf = pIter->pBuf;
894 895 896 897

  return TSDB_CODE_SUCCESS;
}

C
Cary Xu 已提交
898
static int32_t tdRSmaQTaskInfoIterNextBlock(SRSmaQTaskInfoIter *pIter, bool *isFinish) {
899 900 901 902 903 904 905 906 907 908 909 910 911 912 913 914 915
  STFile *pTFile = pIter->pTFile;
  int64_t nBytes = RSMA_QTASKINFO_BUFSIZE;

  if (pIter->offset >= pIter->fsize) {
    *isFinish = true;
    return TSDB_CODE_SUCCESS;
  }

  if ((pIter->fsize - pIter->offset) < RSMA_QTASKINFO_BUFSIZE) {
    nBytes = pIter->fsize - pIter->offset;
  }

  if (tdSeekTFile(pTFile, pIter->offset, SEEK_SET) < 0) {
    ASSERT(0);
    return TSDB_CODE_FAILED;
  }

C
Cary Xu 已提交
916
  if (tdReadTFile(pTFile, pIter->qBuf, nBytes) != nBytes) {
917 918 919 920 921
    ASSERT(0);
    return TSDB_CODE_FAILED;
  }

  int32_t infoLen = 0;
C
Cary Xu 已提交
922
  taosDecodeFixedI32(pIter->qBuf, &infoLen);
923 924 925
  if (infoLen > nBytes) {
    ASSERT(infoLen > RSMA_QTASKINFO_BUFSIZE);
    pIter->nAlloc = infoLen;
C
Cary Xu 已提交
926
    void *pBuf = taosMemoryRealloc(pIter->pBuf, infoLen);
927 928 929 930
    if (!pBuf) {
      terrno = TSDB_CODE_OUT_OF_MEMORY;
      return TSDB_CODE_FAILED;
    }
C
Cary Xu 已提交
931 932
    pIter->pBuf = pBuf;
    pIter->qBuf = pIter->pBuf;
933 934 935 936 937 938 939
    nBytes = infoLen;

    if (tdSeekTFile(pTFile, pIter->offset, SEEK_SET)) {
      ASSERT(0);
      return TSDB_CODE_FAILED;
    }

C
Cary Xu 已提交
940
    if (tdReadTFile(pTFile, pIter->pBuf, nBytes) != nBytes) {
941 942 943 944 945 946 947 948 949 950 951 952
      ASSERT(0);
      return TSDB_CODE_FAILED;
    }
  }

  pIter->offset += nBytes;
  pIter->nBytes = nBytes;
  pIter->nBufPos = 0;

  return TSDB_CODE_SUCCESS;
}

C
Cary Xu 已提交
953
static int32_t tdRSmaQTaskInfoRestore(SSma *pSma, SRSmaQTaskInfoIter *pIter) {
954 955 956 957 958 959 960 961 962 963 964 965 966
  while (1) {
    // block iter
    bool isFinish = false;
    if (tdRSmaQTaskInfoIterNextBlock(pIter, &isFinish) < 0) {
      ASSERT(0);
      return TSDB_CODE_FAILED;
    }
    if (isFinish) {
      return TSDB_CODE_SUCCESS;
    }

    // consume the block
    int32_t qTaskInfoLenWithHead = 0;
C
Cary Xu 已提交
967 968
    pIter->qBuf = taosDecodeFixedI32(pIter->qBuf, &qTaskInfoLenWithHead);
    if (qTaskInfoLenWithHead < RSMA_QTASKINFO_HEAD_LEN) {
969 970 971
      terrno = TSDB_CODE_TDB_FILE_CORRUPTED;
      return TSDB_CODE_FAILED;
    }
C
Cary Xu 已提交
972

973 974
    while (1) {
      if ((pIter->nBufPos + qTaskInfoLenWithHead) <= pIter->nBytes) {
C
Cary Xu 已提交
975 976 977 978 979
        SRSmaQTaskInfoItem infoItem = {0};
        pIter->qBuf = taosDecodeFixedI8(pIter->qBuf, &infoItem.type);
        pIter->qBuf = taosDecodeFixedI64(pIter->qBuf, &infoItem.suid);
        infoItem.qTaskInfo = pIter->qBuf;
        infoItem.len = tdRSmaQTaskInfoContLen(qTaskInfoLenWithHead);
980
        // do the restore job
C
Cary Xu 已提交
981 982 983
        smaDebug("vgId:%d, restore the qtask info %s offset:%" PRIi64 "\n", SMA_VID(pSma),
                 TD_TFILE_FULL_NAME(pIter->pTFile), pIter->offset - pIter->nBytes + pIter->nBufPos);
        tdRSmaQTaskInfoItemRestore(pSma, &infoItem);
984

C
Cary Xu 已提交
985
        pIter->qBuf = POINTER_SHIFT(pIter->qBuf, infoItem.len);
986 987
        pIter->nBufPos += qTaskInfoLenWithHead;

C
Cary Xu 已提交
988 989 990 991 992 993 994
        if ((pIter->nBufPos + RSMA_QTASKINFO_HEAD_LEN) >= pIter->nBytes) {
          // prepare and load next block in the file
          pIter->offset -= (pIter->nBytes - pIter->nBufPos);
          break;
        }

        pIter->qBuf = taosDecodeFixedI32(pIter->qBuf, &qTaskInfoLenWithHead);
995 996 997 998 999 1000 1001 1002 1003 1004
        continue;
      }
      // prepare and load next block in the file
      pIter->offset -= (pIter->nBytes - pIter->nBufPos);
      break;
    }
  }

  return TSDB_CODE_SUCCESS;
}
C
Cary Xu 已提交
1005 1006 1007 1008 1009 1010 1011 1012 1013 1014 1015 1016 1017 1018 1019 1020 1021 1022 1023 1024 1025 1026 1027 1028 1029 1030 1031 1032 1033 1034 1035 1036 1037 1038 1039 1040 1041 1042 1043 1044 1045 1046 1047 1048 1049 1050 1051 1052 1053 1054 1055 1056 1057 1058 1059 1060 1061 1062 1063 1064 1065 1066 1067 1068 1069 1070 1071 1072 1073 1074 1075 1076 1077 1078 1079 1080 1081 1082 1083 1084 1085 1086 1087 1088 1089 1090 1091 1092 1093 1094 1095 1096 1097 1098 1099 1100 1101 1102 1103 1104 1105 1106 1107 1108 1109 1110 1111 1112 1113 1114 1115 1116 1117 1118 1119 1120 1121 1122 1123 1124 1125 1126 1127 1128 1129 1130 1131 1132 1133 1134 1135 1136 1137 1138 1139 1140 1141 1142 1143 1144 1145 1146 1147 1148 1149 1150 1151 1152 1153 1154 1155 1156 1157 1158 1159 1160 1161 1162 1163 1164 1165 1166 1167 1168 1169 1170 1171 1172 1173 1174 1175 1176 1177 1178 1179 1180 1181 1182 1183 1184 1185 1186 1187 1188 1189 1190 1191 1192 1193 1194 1195 1196 1197 1198 1199 1200 1201 1202 1203 1204 1205 1206 1207 1208 1209

static void tdRSmaQTaskInfoGetFName(int32_t vid, int8_t ftype, char *outputName) {
  tdGetVndFileName(vid, VNODE_RSMA_DIR, tdQTaskInfoFname[ftype], outputName);
}

static void *tdRSmaPersistExec(void *param) {
  setThreadName("rsma-task-persist");
  SRSmaStat *pRSmaStat = param;
  SSma      *pSma = pRSmaStat->pSma;
  STfs      *pTfs = pSma->pVnode->pTfs;
  int32_t    vid = SMA_VID(pSma);
  int64_t    toffset = 0;
  bool       isFileCreated = false;

  if (TASK_TRIGGER_STAT_CANCELLED == atomic_load_8(RSMA_TRIGGER_STAT(pRSmaStat))) {
    goto _end;
  }

  void *infoHash = taosHashIterate(RSMA_INFO_HASH(pRSmaStat), NULL);
  if (!infoHash) {
    goto _end;
  }

  STFile tFile = {0};
  while (infoHash) {
    SRSmaInfo *pRSmaInfo = *(SRSmaInfo **)infoHash;
    for (int32_t i = 0; i < TSDB_RETENTION_L2; ++i) {
      qTaskInfo_t taskInfo = pRSmaInfo->items[i].taskInfo;
      if (!taskInfo) {
        smaDebug("vgId:%d, table %" PRIi64 " level %d qTaskInfo is NULL", vid, pRSmaInfo->suid, i + 1);
        continue;
      }

      char   *pOutput = NULL;
      int32_t len = 0;
      int8_t  type = (int8_t)(i + 1);
      if (qSerializeTaskStatus(taskInfo, &pOutput, &len) < 0) {
        smaError("vgId:%d, table %" PRIi64 " level %d serialize rsma task failed since %s", vid, pRSmaInfo->suid, i + 1,
                 terrstr(terrno));
        goto _err;
      }
      if (!pOutput || len <= 0) {
        smaDebug("vgId:%d, table %" PRIi64 " level %d serialize rsma task success but no output(len %d), not persist",
                 vid, pRSmaInfo->suid, i + 1, len);
        taosMemoryFreeClear(pOutput);
        continue;
      }

      smaDebug("vgId:%d, table %" PRIi64 " level %d serialize rsma task success with len %d, need persist", vid,
               pRSmaInfo->suid, i + 1, len);
#if 0
      if (qDeserializeTaskStatus(taskInfo, pOutput, len) < 0) {
        smaError("vgId:%d, table %" PRIi64 "level %d  deserialize rsma task failed since %s", vid, pRSmaInfo->suid,
                 i + 1, terrstr(terrno));
      } else {
        smaDebug("vgId:%d, table %" PRIi64 " level %d deserialize rsma task success", vid, pRSmaInfo->suid, i + 1);
      }
#endif

      if (!isFileCreated) {
        char qTaskInfoFName[TSDB_FILENAME_LEN];
        tdRSmaQTaskInfoGetFName(vid, TD_QTASK_TMP_FILE, qTaskInfoFName);
        tdInitTFile(&tFile, pTfs, qTaskInfoFName);
        tdCreateTFile(&tFile, pTfs, true, -1);

        isFileCreated = true;
      }

      char    tmpBuf[RSMA_QTASKINFO_HEAD_LEN] = {0};
      void   *pTmpBuf = &tmpBuf;
      int32_t headLen = 0;
      headLen += taosEncodeFixedI32(&pTmpBuf, len + RSMA_QTASKINFO_HEAD_LEN);
      headLen += taosEncodeFixedI8(&pTmpBuf, type);
      headLen += taosEncodeFixedI64(&pTmpBuf, pRSmaInfo->suid);

      ASSERT(headLen <= RSMA_QTASKINFO_HEAD_LEN);
      tdAppendTFile(&tFile, (void *)&tmpBuf, headLen, &toffset);
      smaDebug("vgId:%d, table %" PRIi64 " level %d head part len:%d appended to offset:%" PRIi64, vid, pRSmaInfo->suid,
               i + 1, headLen, toffset);
      tdAppendTFile(&tFile, pOutput, len, &toffset);
      smaDebug("vgId:%d, table %" PRIi64 " level %d body part len:%d appended to offset:%" PRIi64, vid, pRSmaInfo->suid,
               i + 1, len, toffset);

      taosMemoryFree(pOutput);
    }

    infoHash = taosHashIterate(RSMA_INFO_HASH(pRSmaStat), infoHash);
  }
_normal:
  if (isFileCreated) {
    if (tdUpdateTFileHeader(&tFile) < 0) {
      smaError("vgId:%d, failed to update tfile %s header since %s", vid, TD_TFILE_FULL_NAME(&tFile),
               tstrerror(terrno));
      tdCloseTFile(&tFile);
      tdRemoveTFile(&tFile);
      goto _err;
    } else {
      smaDebug("vgId:%d, succeed to update tfile %s header", vid, TD_TFILE_FULL_NAME(&tFile));
    }

    tdCloseTFile(&tFile);

    char newFName[TSDB_FILENAME_LEN];
    strncpy(newFName, TD_TFILE_FULL_NAME(&tFile), TSDB_FILENAME_LEN);
    char *pos = strstr(newFName, tdQTaskInfoFname[TD_QTASK_TMP_FILE]);
    strncpy(pos, tdQTaskInfoFname[TD_QTASK_CUR_FILE], TSDB_FILENAME_LEN - POINTER_DISTANCE(pos, newFName));
    if (taosRenameFile(TD_TFILE_FULL_NAME(&tFile), newFName) != 0) {
      smaError("vgId:%d, failed to rename %s to %s", vid, TD_TFILE_FULL_NAME(&tFile), newFName);
      goto _err;
    } else {
      smaDebug("vgId:%d, succeed to rename %s to %s", vid, TD_TFILE_FULL_NAME(&tFile), newFName);
    }
  }
  goto _end;
_err:
  if (isFileCreated) {
    tdRemoveTFile(&tFile);
  }
_end:
  if (TASK_TRIGGER_STAT_INACTIVE == atomic_val_compare_exchange_8(RSMA_TRIGGER_STAT(pRSmaStat),
                                                                  TASK_TRIGGER_STAT_INACTIVE,
                                                                  TASK_TRIGGER_STAT_ACTIVE)) {
    smaDebug("vgId:%d, persist task is active again", vid);
  } else if (TASK_TRIGGER_STAT_CANCELLED == atomic_val_compare_exchange_8(RSMA_TRIGGER_STAT(pRSmaStat),
                                                                          TASK_TRIGGER_STAT_CANCELLED,
                                                                          TASK_TRIGGER_STAT_FINISHED)) {
    smaDebug("vgId:%d, persist task is cancelled", vid);
  } else {
    smaWarn("vgId:%d, persist task in abnormal stat %" PRIi8, vid, atomic_load_8(RSMA_TRIGGER_STAT(pRSmaStat)));
    ASSERT(0);
  }
  atomic_store_8(RSMA_RUNNING_STAT(pRSmaStat), 0);
  taosThreadExit(NULL);
  return NULL;
}

static void tdRSmaPersistTask(SRSmaStat *pRSmaStat) {
  TdThreadAttr thAttr;
  taosThreadAttrInit(&thAttr);
  taosThreadAttrSetDetachState(&thAttr, PTHREAD_CREATE_DETACHED);
  TdThread tid;

  if (taosThreadCreate(&tid, &thAttr, tdRSmaPersistExec, pRSmaStat) != 0) {
    if (TASK_TRIGGER_STAT_INACTIVE == atomic_val_compare_exchange_8(RSMA_TRIGGER_STAT(pRSmaStat),
                                                                    TASK_TRIGGER_STAT_INACTIVE,
                                                                    TASK_TRIGGER_STAT_ACTIVE)) {
      smaDebug("vgId:%d, persist task is active again", SMA_VID(pRSmaStat->pSma));
    } else if (TASK_TRIGGER_STAT_CANCELLED == atomic_val_compare_exchange_8(RSMA_TRIGGER_STAT(pRSmaStat),
                                                                            TASK_TRIGGER_STAT_CANCELLED,
                                                                            TASK_TRIGGER_STAT_FINISHED)) {
      smaDebug("vgId:%d, persist task is cancelled and set finished", SMA_VID(pRSmaStat->pSma));
    } else {
      smaWarn("vgId:%d, persist task in abnormal stat %" PRIi8, atomic_load_8(RSMA_TRIGGER_STAT(pRSmaStat)),
              SMA_VID(pRSmaStat->pSma));
      ASSERT(0);
    }
    atomic_store_8(RSMA_RUNNING_STAT(pRSmaStat), 0);
  }

  taosThreadAttrDestroy(&thAttr);
}

/**
 * @brief trigger to persist rsma qTaskInfo
 *
 * @param param
 * @param tmrId
 */
static void tdRSmaPersistTrigger(void *param, void *tmrId) {
  SRSmaStat *pRSmaStat = param;

  int8_t tmrStat =
      atomic_val_compare_exchange_8(RSMA_TRIGGER_STAT(pRSmaStat), TASK_TRIGGER_STAT_ACTIVE, TASK_TRIGGER_STAT_INACTIVE);
  switch (tmrStat) {
    case TASK_TRIGGER_STAT_ACTIVE: {
      atomic_store_8(RSMA_RUNNING_STAT(pRSmaStat), 1);
      if (TASK_TRIGGER_STAT_CANCELLED != atomic_val_compare_exchange_8(RSMA_TRIGGER_STAT(pRSmaStat),
                                                                       TASK_TRIGGER_STAT_CANCELLED,
                                                                       TASK_TRIGGER_STAT_FINISHED)) {
        smaDebug("vgId:%d, rsma persistence start since active", SMA_VID(pRSmaStat->pSma));

        // start persist task
        tdRSmaPersistTask(pRSmaStat);

        taosTmrReset(tdRSmaPersistTrigger, RSMA_QTASKINFO_PERSIST_MS, pRSmaStat, pRSmaStat->tmrHandle,
                     &pRSmaStat->tmrId);
      } else {
        atomic_store_8(RSMA_RUNNING_STAT(pRSmaStat), 0);
      }
    } break;
    case TASK_TRIGGER_STAT_CANCELLED: {
      atomic_store_8(RSMA_TRIGGER_STAT(pRSmaStat), TASK_TRIGGER_STAT_FINISHED);
      smaDebug("rsma persistence not start since cancelled and finished");
    } break;
    case TASK_TRIGGER_STAT_INACTIVE: {
      smaDebug("rsma persistence not start since inactive");
    } break;
    case TASK_TRIGGER_STAT_INIT: {
      smaDebug("rsma persistence not start since init");
    } break;
    default: {
      smaWarn("rsma persistence not start since unknown stat %" PRIi8, tmrStat);
    } break;
  }
}