smaRollup.c 53.7 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"

L
Liu Jicong 已提交
18 19 20 21 22
#define RSMA_QTASKEXEC_SMOOTH_SIZE (100)     // cnt
#define RSMA_SUBMIT_BATCH_SIZE     (1024)    // cnt
#define RSMA_FETCH_DELAY_MAX       (120000)  // ms
#define RSMA_FETCH_ACTIVE_MAX      (1000)    // ms
#define RSMA_FETCH_INTERVAL        (5000)    // ms
C
Cary Xu 已提交
23

K
kailixu 已提交
24 25
#define RSMA_NEED_FETCH(r) (RSMA_INFO_ITEM((r), 0)->fetchLevel || RSMA_INFO_ITEM((r), 1)->fetchLevel)

26
SSmaMgmt smaMgmt = {
C
Cary Xu 已提交
27 28
    .inited = 0,
    .rsetId = -1,
29 30
};

31 32
typedef struct SRSmaQTaskInfoItem SRSmaQTaskInfoItem;

33
static int32_t    tdUidStorePut(STbUidStore *pStore, tb_uid_t suid, tb_uid_t *uid);
K
kailixu 已提交
34
static void       tdUidStoreDestory(STbUidStore *pStore);
35
static int32_t    tdUpdateTbUidListImpl(SSma *pSma, tb_uid_t *suid, SArray *tbUids, bool isAdd);
36
static int32_t    tdSetRSmaInfoItemParams(SSma *pSma, SRSmaParam *param, SRSmaStat *pStat, SRSmaInfo *pRSmaInfo,
L
Liu Jicong 已提交
37
                                          int8_t idx);
C
Cary Xu 已提交
38
static int32_t    tdExecuteRSmaImpl(SSma *pSma, const void *pMsg, int32_t msgSize, int32_t inputType, SRSmaInfo *pInfo,
C
Cary Xu 已提交
39
                                    ERsmaExecType type, int8_t level);
C
Cary Xu 已提交
40 41
static SRSmaInfo *tdAcquireRSmaInfoBySuid(SSma *pSma, int64_t suid);
static void       tdReleaseRSmaInfo(SSma *pSma, SRSmaInfo *pInfo);
C
Cary Xu 已提交
42
static void       tdFreeRSmaSubmitItems(SArray *pItems);
43
static int32_t    tdRSmaFetchAllResult(SSma *pSma, SRSmaInfo *pInfo);
C
Cary Xu 已提交
44 45
static int32_t    tdRSmaExecAndSubmitResult(SSma *pSma, qTaskInfo_t taskInfo, SRSmaInfoItem *pItem, STSchema *pTSchema,
                                            int64_t suid);
C
Cary Xu 已提交
46
static void       tdRSmaFetchTrigger(void *param, void *tmrId);
47 48
static int32_t    tdRSmaInfoClone(SSma *pSma, SRSmaInfo *pInfo);
static void       tdRSmaQTaskInfoFree(qTaskInfo_t *taskHandle, int32_t vgId, int32_t level);
C
Cary Xu 已提交
49 50 51
static int32_t    tdRSmaRestoreQTaskInfoInit(SSma *pSma, int64_t *nTables);
static int32_t    tdRSmaRestoreQTaskInfoReload(SSma *pSma, int8_t type, int64_t qTaskFileVer);
static int32_t    tdRSmaRestoreTSDataReload(SSma *pSma);
C
Cary Xu 已提交
52

53 54 55 56 57 58 59
struct SRSmaQTaskInfoItem {
  int32_t len;
  int8_t  type;
  int64_t suid;
  void   *qTaskInfo;
};

60
static void tdRSmaQTaskInfoFree(qTaskInfo_t *taskHandle, int32_t vgId, int32_t level) {
61
  // Note: free/kill may in RC
C
Cary Xu 已提交
62
  if (!taskHandle || !(*taskHandle)) return;
63 64
  qTaskInfo_t otaskHandle = atomic_load_ptr(taskHandle);
  if (otaskHandle && atomic_val_compare_exchange_ptr(taskHandle, otaskHandle, NULL)) {
C
Cary Xu 已提交
65
    smaDebug("vgId:%d, free qTaskInfo_t %p of level %d", vgId, otaskHandle, level);
66
    qDestroyTask(otaskHandle);
C
Cary Xu 已提交
67
  } else {
C
Cary Xu 已提交
68
    smaDebug("vgId:%d, not free qTaskInfo_t %p of level %d", vgId, otaskHandle, level);
69
  }
C
Cary Xu 已提交
70
  // TODO: clear files related to qTaskInfo?
71 72
}

C
Cary Xu 已提交
73 74 75 76 77 78 79 80 81
/**
 * @brief general function to free rsmaInfo
 *
 * @param pSma
 * @param pInfo
 * @param isDeepFree Only stop tmrId and free pTSchema for deep free
 * @return void*
 */
void *tdFreeRSmaInfo(SSma *pSma, SRSmaInfo *pInfo, bool isDeepFree) {
82
  if (pInfo) {
C
Cary Xu 已提交
83
    for (int32_t i = 0; i < TSDB_RETENTION_L2; ++i) {
84
      SRSmaInfoItem *pItem = &pInfo->items[i];
C
Cary Xu 已提交
85

C
Cary Xu 已提交
86
      if (isDeepFree && pItem->tmrId) {
C
Cary Xu 已提交
87 88
        smaDebug("vgId:%d, stop fetch timer %p for table %" PRIi64 " level %d", SMA_VID(pSma), pItem->tmrId,
                 pInfo->suid, i + 1);
C
Cary Xu 已提交
89 90 91
        taosTmrStopA(&pItem->tmrId);
      }

92 93 94 95
      if (isDeepFree && pItem->pStreamState) {
        streamStateClose(pItem->pStreamState);
      }

C
Cary Xu 已提交
96
      if (isDeepFree && pInfo->taskInfo[i]) {
C
Cary Xu 已提交
97
        tdRSmaQTaskInfoFree(&pInfo->taskInfo[i], SMA_VID(pSma), i + 1);
C
Cary Xu 已提交
98
      } else {
C
Cary Xu 已提交
99
        smaDebug("vgId:%d, table %" PRIi64 " no need to destroy rsma info level %d since empty taskInfo", SMA_VID(pSma),
100
                 pInfo->suid, i + 1);
101
      }
C
Cary Xu 已提交
102 103

      if (pInfo->iTaskInfo[i]) {
C
Cary Xu 已提交
104
        tdRSmaQTaskInfoFree(&pInfo->iTaskInfo[i], SMA_VID(pSma), i + 1);
C
Cary Xu 已提交
105
      } else {
C
Cary Xu 已提交
106 107
        smaDebug("vgId:%d, table %" PRIi64 " no need to destroy rsma info level %d since empty iTaskInfo",
                 SMA_VID(pSma), pInfo->suid, i + 1);
C
Cary Xu 已提交
108
      }
109
    }
C
Cary Xu 已提交
110
    if (isDeepFree) {
C
Cary Xu 已提交
111
      taosMemoryFreeClear(pInfo->pTSchema);
C
Cary Xu 已提交
112
    }
C
Cary Xu 已提交
113 114 115 116

    if (isDeepFree) {
      if (pInfo->queue) taosCloseQueue(pInfo->queue);
      if (pInfo->qall) taosFreeQall(pInfo->qall);
C
Cary Xu 已提交
117 118
      if (pInfo->iQueue) taosCloseQueue(pInfo->iQueue);
      if (pInfo->iQall) taosFreeQall(pInfo->iQall);
C
Cary Xu 已提交
119 120
      pInfo->queue = NULL;
      pInfo->qall = NULL;
C
Cary Xu 已提交
121 122
      pInfo->iQueue = NULL;
      pInfo->iQall = NULL;
C
Cary Xu 已提交
123 124
    }

C
Cary Xu 已提交
125 126
    taosMemoryFree(pInfo);
  }
127

128 129 130 131
  return NULL;
}

static FORCE_INLINE int32_t tdUidStoreInit(STbUidStore **pStore) {
K
kailixu 已提交
132 133 134 135 136 137
  if (ASSERTS(*pStore == NULL, "*pStore:%p != NULL", *pStore)) {
    terrno = TSDB_CODE_APP_ERROR;
    return TSDB_CODE_FAILED;
  }

  *pStore = taosMemoryCalloc(1, sizeof(STbUidStore));
138
  if (*pStore == NULL) {
K
kailixu 已提交
139 140
    terrno = TSDB_CODE_OUT_OF_MEMORY;
    return TSDB_CODE_FAILED;
141
  }
K
kailixu 已提交
142

143 144 145
  return TSDB_CODE_SUCCESS;
}

146
static int32_t tdUpdateTbUidListImpl(SSma *pSma, tb_uid_t *suid, SArray *tbUids, bool isAdd) {
147 148 149 150
  SRSmaInfo *pRSmaInfo = NULL;

  if (!suid || !tbUids) {
    terrno = TSDB_CODE_INVALID_PTR;
C
Cary Xu 已提交
151 152
    smaError("vgId:%d, failed to get rsma info for uid:%" PRIi64 " since %s", SMA_VID(pSma), suid ? *suid : -1,
             terrstr());
153 154 155
    return TSDB_CODE_FAILED;
  }

C
Cary Xu 已提交
156 157 158 159 160
  if (!taosArrayGetSize(tbUids)) {
    smaDebug("vgId:%d, no need to update tbUidList for suid:%" PRIi64 " since Empty tbUids", SMA_VID(pSma), *suid);
    return TSDB_CODE_SUCCESS;
  }

C
Cary Xu 已提交
161
  pRSmaInfo = tdAcquireRSmaInfoBySuid(pSma, *suid);
162 163

  if (!pRSmaInfo) {
S
Shengliang Guan 已提交
164
    smaError("vgId:%d, failed to get rsma info for uid:%" PRIi64, SMA_VID(pSma), *suid);
C
Cary Xu 已提交
165
    terrno = TSDB_CODE_RSMA_INVALID_STAT;
166 167 168
    return TSDB_CODE_FAILED;
  }

C
Cary Xu 已提交
169 170
  for (int32_t i = 0; i < TSDB_RETENTION_L2; ++i) {
    if (pRSmaInfo->taskInfo[i]) {
K
kailixu 已提交
171
      if ((terrno = qUpdateQualifiedTableId(pRSmaInfo->taskInfo[i], tbUids, isAdd)) < 0) {
C
Cary Xu 已提交
172
        tdReleaseRSmaInfo(pSma, pRSmaInfo);
C
Cary Xu 已提交
173 174 175 176
        smaError("vgId:%d, update tbUidList failed for uid:%" PRIi64 " level %d since %s", SMA_VID(pSma), *suid, i,
                 terrstr());
        return TSDB_CODE_FAILED;
      }
K
kailixu 已提交
177 178
      smaDebug("vgId:%d, update tbUidList succeed for qTaskInfo:%p with suid:%" PRIi64 " uid:%" PRIi64 " level %d",
               SMA_VID(pSma), pRSmaInfo->taskInfo[i], *suid, *(int64_t *)taosArrayGet(tbUids, 0), i);
C
Cary Xu 已提交
179
    }
180 181
  }

C
Cary Xu 已提交
182
  tdReleaseRSmaInfo(pSma, pRSmaInfo);
183 184 185
  return TSDB_CODE_SUCCESS;
}

186
int32_t tdUpdateTbUidList(SSma *pSma, STbUidStore *pStore, bool isAdd) {
187 188 189 190
  if (!pStore || (taosArrayGetSize(pStore->tbUids) == 0)) {
    return TSDB_CODE_SUCCESS;
  }

191
  if (tdUpdateTbUidListImpl(pSma, &pStore->suid, pStore->tbUids, isAdd) != TSDB_CODE_SUCCESS) {
192 193 194 195 196 197 198 199
    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;

200
    if (tdUpdateTbUidListImpl(pSma, pTbSuid, pTbUids, isAdd) != TSDB_CODE_SUCCESS) {
201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226
      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 已提交
227 228 229
  SRSmaStat *pStat = (SRSmaStat *)SMA_ENV_STAT(pEnv);
  SHashObj  *infoHash = NULL;
  if (!pStat || !(infoHash = RSMA_INFO_HASH(pStat))) {
C
Cary Xu 已提交
230
    terrno = TSDB_CODE_RSMA_INVALID_STAT;
231 232 233 234 235 236 237 238 239
    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;
  }

  if (!(*ppStore)) {
240
    if (tdUidStoreInit(ppStore) < 0) {
241 242 243 244
      return TSDB_CODE_FAILED;
    }
  }

245
  if (tdUidStorePut(*ppStore, suid, &uid) < 0) {
246 247 248 249 250 251 252
    *ppStore = tdUidStoreFree(*ppStore);
    return TSDB_CODE_FAILED;
  }

  return TSDB_CODE_SUCCESS;
}

253
static int32_t tdSetRSmaInfoItemParams(SSma *pSma, SRSmaParam *param, SRSmaStat *pStat, SRSmaInfo *pRSmaInfo,
L
Liu Jicong 已提交
254
                                       int8_t idx) {
C
Cary Xu 已提交
255 256 257
  if ((param->qmsgLen > 0) && param->qmsg[idx]) {
    SRetention *pRetention = SMA_RETENTION(pSma);
    STsdbCfg   *pTsdbCfg = SMA_TSDB_CFG(pSma);
C
Cary Xu 已提交
258
    SVnode     *pVnode = pSma->pVnode;
259 260 261 262 263 264
    char        taskInfDir[TSDB_FILENAME_LEN] = {0};
    void       *pStreamState = NULL;

    // set the backend of stream state
    tdRSmaQTaskInfoGetFullPathEx(TD_VID(pVnode), pRSmaInfo->suid, idx + 1, tfsGetPrimaryPath(pVnode->pTfs), taskInfDir);
    if (!taosCheckExistFile(taskInfDir)) {
265
      char *s = taosStrdup(taskInfDir);
266 267 268 269 270 271 272
      if (taosMulMkDir(taosDirName(s)) != 0) {
        terrno = TAOS_SYSTEM_ERROR(errno);
        taosMemoryFree(s);
        return TSDB_CODE_FAILED;
      }
      taosMemoryFree(s);
    }
273
    pStreamState = streamStateOpen(taskInfDir, NULL, true, -1, -1);
274 275 276 277 278
    if (!pStreamState) {
      terrno = TSDB_CODE_RSMA_STREAM_STATE_OPEN;
      return TSDB_CODE_FAILED;
    }

C
Cary Xu 已提交
279
    SReadHandle handle = {
C
Cary Xu 已提交
280 281
        .meta = pVnode->pMeta,
        .vnode = pVnode,
C
Cary Xu 已提交
282
        .initTqReader = 1,
283
        .pStateBackend = pStreamState,
C
Cary Xu 已提交
284
    };
D
dapan1121 已提交
285
    pRSmaInfo->taskInfo[idx] = qCreateStreamExecTaskInfo(param->qmsg[idx], &handle, TD_VID(pVnode));
C
Cary Xu 已提交
286
    if (!pRSmaInfo->taskInfo[idx]) {
C
Cary Xu 已提交
287
      terrno = TSDB_CODE_RSMA_QTASKINFO_CREATE;
C
Cary Xu 已提交
288
      return TSDB_CODE_FAILED;
C
Cary Xu 已提交
289
    }
C
Cary Xu 已提交
290
    SRSmaInfoItem *pItem = &(pRSmaInfo->items[idx]);
291
    pItem->triggerStat = TASK_TRIGGER_STAT_ACTIVE;  // fetch the data when reboot
292
    pItem->pStreamState = pStreamState;
C
Cary Xu 已提交
293
    if (param->maxdelay[idx] < TSDB_MIN_ROLLUP_MAX_DELAY) {
K
kailixu 已提交
294 295
      int64_t msInterval =
          convertTimeFromPrecisionToUnit(pRetention[idx + 1].freq, pTsdbCfg->precision, TIME_UNIT_MILLISECOND);
C
Cary Xu 已提交
296
      pItem->maxDelay = (int32_t)msInterval;
C
Cary Xu 已提交
297
    } else {
C
Cary Xu 已提交
298
      pItem->maxDelay = (int32_t)param->maxdelay[idx];
C
Cary Xu 已提交
299
    }
C
Cary Xu 已提交
300 301
    if (pItem->maxDelay > TSDB_MAX_ROLLUP_MAX_DELAY) {
      pItem->maxDelay = TSDB_MAX_ROLLUP_MAX_DELAY;
C
Cary Xu 已提交
302
    }
C
Cary Xu 已提交
303

K
kailixu 已提交
304 305 306 307 308 309
    pItem->level = idx == 0 ? TSDB_RETENTION_L1 : TSDB_RETENTION_L2;

    if (ASSERTS(pItem->level > 0, "pItem level:%" PRIi8 " should > 0", pItem->level)) {
      terrno = TSDB_CODE_APP_ERROR;
      return TSDB_CODE_FAILED;
    }
310 311 312 313

    SRSmaRef rsmaRef = {.refId = pStat->refId, .suid = pRSmaInfo->suid};
    taosHashPut(smaMgmt.refHash, &pItem, POINTER_BYTES, &rsmaRef, sizeof(rsmaRef));

314
    taosTmrReset(tdRSmaFetchTrigger, RSMA_FETCH_INTERVAL, pItem, smaMgmt.tmrHandle, &pItem->tmrId);
315 316

    smaInfo("vgId:%d, item:%p table:%" PRIi64 " level:%" PRIi8 " maxdelay:%" PRIi64 " watermark:%" PRIi64
317
            ", finally maxdelay:%" PRIi32,
318
            TD_VID(pVnode), pItem, pRSmaInfo->suid, (int8_t)(idx + 1), param->maxdelay[idx], param->watermark[idx],
319
            pItem->maxDelay);
C
Cary Xu 已提交
320 321 322 323
  }
  return TSDB_CODE_SUCCESS;
}

324
/**
325
 * @brief for rsam create or restore
326
 *
327 328 329 330
 * @param pSma
 * @param param
 * @param suid
 * @param tbName
331 332
 * @return int32_t
 */
333
int32_t tdRSmaProcessCreateImpl(SSma *pSma, SRSmaParam *param, int64_t suid, const char *tbName) {
334
  if ((param->qmsgLen[0] == 0) && (param->qmsgLen[1] == 0)) {
335
    smaDebug("vgId:%d, no qmsg1/qmsg2 for rollup table %s %" PRIi64, SMA_VID(pSma), tbName, suid);
336 337 338
    return TSDB_CODE_SUCCESS;
  }

K
kailixu 已提交
339
#if 0
C
Cary Xu 已提交
340
  if (tdCheckAndInitSmaEnv(pSma, TSDB_SMA_TYPE_ROLLUP) != TSDB_CODE_SUCCESS) {
341 342 343
    terrno = TSDB_CODE_TDB_INIT_FAILED;
    return TSDB_CODE_FAILED;
  }
K
kailixu 已提交
344
#endif
345 346

  SSmaEnv   *pEnv = SMA_RSMA_ENV(pSma);
C
Cary Xu 已提交
347
  SRSmaStat *pStat = (SRSmaStat *)SMA_ENV_STAT(pEnv);
348 349
  SRSmaInfo *pRSmaInfo = NULL;

350
  pRSmaInfo = taosHashGet(RSMA_INFO_HASH(pStat), &suid, sizeof(tb_uid_t));
351
  if (pRSmaInfo) {
352 353
    smaInfo("vgId:%d, rsma info already exists for table %s, %" PRIi64, SMA_VID(pSma), tbName, suid);
    return TSDB_CODE_SUCCESS;
354 355
  }

356
  // from write queue: single thead
357 358 359 360 361 362
  pRSmaInfo = (SRSmaInfo *)taosMemoryCalloc(1, sizeof(SRSmaInfo));
  if (!pRSmaInfo) {
    terrno = TSDB_CODE_OUT_OF_MEMORY;
    return TSDB_CODE_FAILED;
  }

363
  STSchema *pTSchema = metaGetTbTSchema(SMA_META(pSma), suid, -1, 1);
364 365 366 367
  if (!pTSchema) {
    terrno = TSDB_CODE_TDB_IVD_TB_SCHEMA_VERSION;
    goto _err;
  }
C
Cary Xu 已提交
368
  pRSmaInfo->pSma = pSma;
369
  pRSmaInfo->pTSchema = pTSchema;
C
Cary Xu 已提交
370 371
  pRSmaInfo->suid = suid;
  T_REF_INIT_VAL(pRSmaInfo, 1);
C
Cary Xu 已提交
372 373 374
  if (!(pRSmaInfo->queue = taosOpenQueue())) {
    goto _err;
  }
C
Cary Xu 已提交
375

C
Cary Xu 已提交
376 377 378
  if (!(pRSmaInfo->qall = taosAllocateQall())) {
    goto _err;
  }
C
Cary Xu 已提交
379 380 381 382 383 384
  if (!(pRSmaInfo->iQueue = taosOpenQueue())) {
    goto _err;
  }
  if (!(pRSmaInfo->iQall = taosAllocateQall())) {
    goto _err;
  }
385

L
Liu Jicong 已提交
386
  if (tdSetRSmaInfoItemParams(pSma, param, pStat, pRSmaInfo, 0) < 0) {
C
Cary Xu 已提交
387 388
    goto _err;
  }
C
Cary Xu 已提交
389

L
Liu Jicong 已提交
390
  if (tdSetRSmaInfoItemParams(pSma, param, pStat, pRSmaInfo, 1) < 0) {
C
Cary Xu 已提交
391 392
    goto _err;
  }
393

394
  if (taosHashPut(RSMA_INFO_HASH(pStat), &suid, sizeof(tb_uid_t), &pRSmaInfo, sizeof(pRSmaInfo)) < 0) {
395
    goto _err;
396 397
  }

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

400
  return TSDB_CODE_SUCCESS;
401
_err:
C
Cary Xu 已提交
402
  tdFreeRSmaInfo(pSma, pRSmaInfo, true);
403
  return TSDB_CODE_FAILED;
404 405
}

406
/**
C
Cary Xu 已提交
407
 * @brief Check and init qTaskInfo_t, only applicable to stable with SRSmaParam currently
408
 *
409
 * @param pSma
410 411 412
 * @param pReq
 * @return int32_t
 */
413 414
int32_t tdProcessRSmaCreate(SSma *pSma, SVCreateStbReq *pReq) {
  SVnode *pVnode = pSma->pVnode;
415
  if (!pReq->rollup) {
416 417 418 419 420 421
    smaTrace("vgId:%d, not create rsma for stable %s %" PRIi64 " since no rollup in req", TD_VID(pVnode), pReq->name,
             pReq->suid);
    return TSDB_CODE_SUCCESS;
  }

  if (!VND_IS_RSMA(pVnode)) {
K
kailixu 已提交
422 423
    smaWarn("vgId:%d, not create rsma for stable %s %" PRIi64 " since vnd is not rsma", TD_VID(pVnode), pReq->name,
            pReq->suid);
424 425 426
    return TSDB_CODE_SUCCESS;
  }

427
  return tdRSmaProcessCreateImpl(pSma, &pReq->rsmaParam, pReq->suid, pReq->name);
428 429
}

430 431 432 433 434 435 436
/**
 * @brief drop cache for stb
 *
 * @param pSma
 * @param pReq
 * @return int32_t
 */
L
Liu Jicong 已提交
437
int32_t tdProcessRSmaDrop(SSma *pSma, SVDropStbReq *pReq) {
438 439
  SVnode *pVnode = pSma->pVnode;
  if (!VND_IS_RSMA(pVnode)) {
C
Cary Xu 已提交
440
    smaTrace("vgId:%d, not drop rsma for stable %s %" PRIi64 " since vnd is not rsma", TD_VID(pVnode), pReq->name,
441 442 443 444
             pReq->suid);
    return TSDB_CODE_SUCCESS;
  }

C
Cary Xu 已提交
445 446 447 448 449
  SSmaEnv *pSmaEnv = SMA_RSMA_ENV(pSma);
  if (!pSmaEnv) {
    return TSDB_CODE_SUCCESS;
  }

C
Cary Xu 已提交
450
  SRSmaStat *pRSmaStat = (SRSmaStat *)SMA_ENV_STAT(pSmaEnv);
C
Cary Xu 已提交
451

C
Cary Xu 已提交
452
  SRSmaInfo *pRSmaInfo = tdAcquireRSmaInfoBySuid(pSma, pReq->suid);
C
Cary Xu 已提交
453 454 455 456 457 458 459 460

  if (!pRSmaInfo) {
    smaWarn("vgId:%d, drop rsma for stable %s %" PRIi64 " failed no rsma in hash", TD_VID(pVnode), pReq->name,
            pReq->suid);
    return TSDB_CODE_SUCCESS;
  }

  // set del flag for data in mem
461
  atomic_store_8(&pRSmaStat->delFlag, 1);
C
Cary Xu 已提交
462 463 464
  RSMA_INFO_SET_DEL(pRSmaInfo);
  tdUnRefRSmaInfo(pSma, pRSmaInfo);

C
Cary Xu 已提交
465
  tdReleaseRSmaInfo(pSma, pRSmaInfo);
C
Cary Xu 已提交
466

K
kailixu 已提交
467 468
  // no need to save to file as triggered by dropping stable
  smaDebug("vgId:%d, drop rsma for stable %" PRIi64 " succeed", TD_VID(pVnode), pReq->suid);
469
  return TSDB_CODE_SUCCESS;
L
Liu Jicong 已提交
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 503 504 505 506 507 508 509 510 511 512 513 514 515 516
/**
 * @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;
C
Cary Xu 已提交
517
          taosArrayDestroy(pUidArray);
518 519
          return TSDB_CODE_FAILED;
        }
520
        if (taosHashPut(pStore->uidHash, &suid, sizeof(suid), &pUidArray, sizeof(pUidArray)) < 0) {
521 522 523 524
          return TSDB_CODE_FAILED;
        }
      }
    } else {
525
      if (taosHashPut(pStore->uidHash, &suid, sizeof(suid), NULL, 0) < 0) {
526 527 528 529 530 531 532
        return TSDB_CODE_FAILED;
      }
    }
  }
  return TSDB_CODE_SUCCESS;
}

K
kailixu 已提交
533
static void tdUidStoreDestory(STbUidStore *pStore) {
534 535 536 537 538 539 540 541 542 543 544 545 546 547 548 549 550 551 552 553 554 555 556 557 558
  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;
}

C
Cary Xu 已提交
559 560 561 562 563 564 565 566
/**
 * @brief The SubmitReq for rsma L2/L3 is inserted by tsdbInsertData method directly while not by WriteQ, as the queue
 * would be freed when close Vnode, thus lock should be used if with race condition.
 * @param pTsdb
 * @param version
 * @param pReq
 * @return int32_t
 */
567 568 569 570 571 572
static int32_t tdProcessSubmitReq(STsdb *pTsdb, int64_t version, void *pReq) {
  if (!pReq) {
    terrno = TSDB_CODE_INVALID_PTR;
    return TSDB_CODE_FAILED;
  }

573 574
  SSubmitReq2 *pSubmitReq = (SSubmitReq2 *)pReq;
  // spin lock for race condition during insert data
575 576 577 578 579 580 581
  if (tsdbInsertData(pTsdb, version, pSubmitReq, NULL) < 0) {
    return TSDB_CODE_FAILED;
  }

  return TSDB_CODE_SUCCESS;
}

582 583 584
static int32_t tdFetchSubmitReqSuids(SSubmitReq2 *pMsg, STbUidStore *pStore) {
  SArray *pSubmitTbData = pMsg ? pMsg->aSubmitTbData : NULL;
  int32_t size = taosArrayGetSize(pSubmitTbData);
585 586 587

  terrno = TSDB_CODE_SUCCESS;

588 589
  for (int32_t i = 0; i < size; ++i) {
    SSubmitTbData *pData = TARRAY_GET_ELEM(pSubmitTbData, i);
590
    if ((terrno = tdUidStorePut(pStore, pData->suid, NULL)) < 0) {
C
Cary Xu 已提交
591 592
      return -1;
    }
593 594 595 596 597
  }

  return 0;
}

H
Hongze Cheng 已提交
598
#if 0
C
Cary Xu 已提交
599 600
/**
 * @brief retention of rsma1/rsma2
C
Cary Xu 已提交
601 602 603 604
 *
 * @param pSma
 * @param now
 * @return int32_t
C
Cary Xu 已提交
605
 */
606
int32_t smaDoRetention(SSma *pSma, int64_t now) {
C
Cary Xu 已提交
607
  int32_t code = TSDB_CODE_SUCCESS;
C
Cary Xu 已提交
608
  if (!VND_IS_RSMA(pSma->pVnode)) {
C
Cary Xu 已提交
609 610 611 612 613
    return code;
  }

  for (int32_t i = 0; i < TSDB_RETENTION_L2; ++i) {
    if (pSma->pRSmaTsdb[i]) {
614
      code = tsdbDoRetention(pSma->pRSmaTsdb[i], now);
C
Cary Xu 已提交
615 616 617 618 619 620 621
      if (code) goto _end;
    }
  }

_end:
  return code;
}
H
Hongze Cheng 已提交
622
#endif
C
Cary Xu 已提交
623

K
kailixu 已提交
624 625 626 627 628 629 630
static void tdBlockDataDestroy(SArray *pBlockArr) {
  for (int32_t i = 0; i < taosArrayGetSize(pBlockArr); ++i) {
    blockDataDestroy(taosArrayGetP(pBlockArr, i));
  }
  taosArrayDestroy(pBlockArr);
}

C
Cary Xu 已提交
631 632
static int32_t tdRSmaExecAndSubmitResult(SSma *pSma, qTaskInfo_t taskInfo, SRSmaInfoItem *pItem, STSchema *pTSchema,
                                         int64_t suid) {
H
Haojun Liao 已提交
633 634 635 636 637
  SArray *pResList = taosArrayInit(1, POINTER_BYTES);
  if (pResList == NULL) {
    terrno = TSDB_CODE_OUT_OF_MEMORY;
    goto _err;
  }
638

H
Haojun Liao 已提交
639 640
  while (1) {
    uint64_t ts;
641
    bool     hasMore = false;
642
    int32_t  code = qExecTaskOpt(taskInfo, pResList, &ts, &hasMore, NULL);
C
Cary Xu 已提交
643
    if (code < 0) {
C
Cary Xu 已提交
644 645 646 647 648 649 650
      if (code == TSDB_CODE_QRY_IN_EXEC) {
        break;
      } else {
        smaError("vgId:%d, qExecTask for rsma table %" PRIi64 " level %" PRIi8 " failed since %s", SMA_VID(pSma), suid,
                 pItem->level, terrstr(code));
        goto _err;
      }
651
    }
C
Cary Xu 已提交
652

H
Haojun Liao 已提交
653 654
    if (taosArrayGetSize(pResList) == 0) {
      if (terrno == 0) {
655
        // smaDebug("vgId:%d, no rsma level %" PRIi8 " data fetched yet", SMA_VID(pSma), pItem->level);
H
Haojun Liao 已提交
656
      } else {
657
        smaDebug("vgId:%d, no rsma level %" PRIi8 " data fetched since %s", SMA_VID(pSma), pItem->level, terrstr());
H
Haojun Liao 已提交
658 659 660 661
        goto _err;
      }

      break;
C
Cary Xu 已提交
662
    } else {
663
      smaDebug("vgId:%d, rsma level %" PRIi8 " data fetched", SMA_VID(pSma), pItem->level);
H
Haojun Liao 已提交
664
    }
C
Cary Xu 已提交
665
#if 0
C
Cary Xu 已提交
666 667 668
    char flag[10] = {0};
    snprintf(flag, 10, "level %" PRIi8, pItem->level);
    blockDebugShowDataBlocks(pResList, flag);
C
Cary Xu 已提交
669
#endif
C
Cary Xu 已提交
670 671
    for (int32_t i = 0; i < taosArrayGetSize(pResList); ++i) {
      SSDataBlock *output = taosArrayGetP(pResList, i);
K
kailixu 已提交
672 673
      smaDebug("result block, uid:%" PRIu64 ", groupid:%" PRIu64 ", rows:%d", output->info.id.uid,
               output->info.id.groupId, output->info.rows);
H
Haojun Liao 已提交
674

675
      STsdb       *sinkTsdb = (pItem->level == TSDB_RETENTION_L1 ? pSma->pRSmaTsdb[0] : pSma->pRSmaTsdb[1]);
676
      SSubmitReq2 *pReq = NULL;
C
Cary Xu 已提交
677 678

      // TODO: the schema update should be handled later(TD-17965)
679
      if (buildSubmitReqFromDataBlock(&pReq, output, pTSchema, output->info.id.groupId, SMA_VID(pSma), suid) < 0) {
K
kailixu 已提交
680 681
        smaError("vgId:%d, build submit req for rsma table suid:%" PRIu64 ", uid:%" PRIu64 ", level %" PRIi8
                 " failed since %s",
H
Haojun Liao 已提交
682
                 SMA_VID(pSma), suid, output->info.id.groupId, pItem->level, terrstr());
C
Cary Xu 已提交
683 684
        goto _err;
      }
685

C
Cary Xu 已提交
686
      if (pReq && tdProcessSubmitReq(sinkTsdb, output->info.version, pReq) < 0) {
687
        tDestroySSubmitReq2(pReq, TSDB_MSG_FLG_ENCODE);
K
kailixu 已提交
688
        taosMemoryFree(pReq);
K
kailixu 已提交
689 690
        smaError("vgId:%d, process submit req for rsma suid:%" PRIu64 ", uid:%" PRIu64 " level %" PRIi8
                 " failed since %s",
H
Haojun Liao 已提交
691
                 SMA_VID(pSma), suid, output->info.id.groupId, pItem->level, terrstr());
C
Cary Xu 已提交
692 693
        goto _err;
      }
C
Cary Xu 已提交
694

695 696
      smaDebug("vgId:%d, process submit req for rsma suid:%" PRIu64 ",uid:%" PRIu64 ", level %" PRIi8 " ver %" PRIi64,
               SMA_VID(pSma), suid, output->info.id.groupId, pItem->level, output->info.version);
697

K
kailixu 已提交
698 699 700 701
      if (pReq) {
        tDestroySSubmitReq2(pReq, TSDB_MSG_FLG_ENCODE);
        taosMemoryFree(pReq);
      }
C
Cary Xu 已提交
702
    }
703 704
  }

H
Haojun Liao 已提交
705 706
  taosArrayDestroy(pResList);
  qCleanExecTaskBlockBuf(taskInfo);
C
Cary Xu 已提交
707
  return TSDB_CODE_SUCCESS;
H
Haojun Liao 已提交
708

C
Cary Xu 已提交
709
_err:
H
Haojun Liao 已提交
710 711
  taosArrayDestroy(pResList);
  qCleanExecTaskBlockBuf(taskInfo);
C
Cary Xu 已提交
712
  return TSDB_CODE_FAILED;
713 714
}

C
Cary Xu 已提交
715
/**
C
Cary Xu 已提交
716
 * @brief Copy msg to rsmaQueueBuffer for batch process
C
Cary Xu 已提交
717 718
 *
 * @param pSma
K
kailixu 已提交
719
 * @param version
C
Cary Xu 已提交
720
 * @param pMsg
721
 * @param len
C
Cary Xu 已提交
722 723 724 725 726
 * @param inputType
 * @param pInfo
 * @param suid
 * @return int32_t
 */
K
kailixu 已提交
727 728 729
static int32_t tdExecuteRSmaImplAsync(SSma *pSma, int64_t version, const void *pMsg, int32_t len, int32_t inputType,
                                      SRSmaInfo *pInfo, tb_uid_t suid) {
  int32_t size = sizeof(int32_t) + sizeof(int64_t) + len;
K
kailixu 已提交
730
  void   *qItem = taosAllocateQitem(size, DEF_QITEM, 0);
C
Cary Xu 已提交
731 732 733 734 735

  if (!qItem) {
    return TSDB_CODE_FAILED;
  }

K
kailixu 已提交
736 737 738 739 740 741
  void *pItem = qItem;

  *(int32_t *)pItem = len;
  pItem = POINTER_SHIFT(pItem, sizeof(int32_t));
  *(int64_t *)pItem = version;
  memcpy(POINTER_SHIFT(pItem, sizeof(int64_t)), pMsg, len);
C
Cary Xu 已提交
742 743 744

  taosWriteQitem(pInfo->queue, qItem);

C
Cary Xu 已提交
745 746
  pInfo->lastRecv = taosGetTimestampMs();

C
Cary Xu 已提交
747
  SRSmaStat *pRSmaStat = SMA_RSMA_STAT(pSma);
C
Cary Xu 已提交
748 749

  int64_t nItems = atomic_fetch_add_64(&pRSmaStat->nBufItems, 1);
C
Cary Xu 已提交
750

751 752 753 754
  if (atomic_load_8(&pInfo->assigned) == 0) {
    tsem_post(&(pRSmaStat->notEmpty));
  }

C
Cary Xu 已提交
755
  // smoothing consume
C
Cary Xu 已提交
756
  int32_t n = nItems / RSMA_QTASKEXEC_SMOOTH_SIZE;
C
Cary Xu 已提交
757 758 759 760
  if (n > 1) {
    if (n > 10) {
      n = 10;
    }
C
Cary Xu 已提交
761 762
    taosMsleep(n << 3);
    if (n > 5) {
C
Cary Xu 已提交
763
      smaWarn("vgId:%d, pInfo->queue itemSize:%d, memSize:%" PRIi64 ", sleep %d ms", SMA_VID(pSma),
C
Cary Xu 已提交
764
              taosQueueItemSize(pInfo->queue), taosQueueMemorySize(pInfo->queue), n << 3);
C
Cary Xu 已提交
765 766
    }
  }
C
Cary Xu 已提交
767

C
Cary Xu 已提交
768 769 770
  return TSDB_CODE_SUCCESS;
}

771
#if 0
C
Cary Xu 已提交
772 773 774 775 776 777 778 779 780 781 782
static int32_t tdRsmaPrintSubmitReq(SSma *pSma, SSubmitReq *pReq) {
  SSubmitMsgIter msgIter = {0};
  SSubmitBlkIter blkIter = {0};
  STSRow        *row = NULL;
  if (tInitSubmitMsgIter(pReq, &msgIter) < 0) return -1;
  while (true) {
    SSubmitBlk *pBlock = NULL;
    if (tGetSubmitMsgNext(&msgIter, &pBlock) < 0) return -1;
    if (pBlock == NULL) break;
    tInitSubmitBlkIter(&msgIter, pBlock, &blkIter);
    while ((row = tGetSubmitBlkNext(&blkIter)) != NULL) {
C
Cary Xu 已提交
783
      smaDebug("vgId:%d, numOfRows:%d, suid:%" PRIi64 ", uid:%" PRIi64 ", version:%" PRIi64 ", ts:%" PRIi64,
C
Cary Xu 已提交
784 785 786 787 788
               SMA_VID(pSma), msgIter.numOfRows, msgIter.suid, msgIter.uid, pReq->version, row->ts);
    }
  }
  return 0;
}
789
#endif
C
Cary Xu 已提交
790

C
Cary Xu 已提交
791 792 793 794 795 796 797 798
/**
 * @brief sync mode
 *
 * @param pSma
 * @param pMsg
 * @param msgSize
 * @param inputType
 * @param pInfo
C
Cary Xu 已提交
799
 * @param type
C
Cary Xu 已提交
800 801 802 803
 * @param level
 * @return int32_t
 */
static int32_t tdExecuteRSmaImpl(SSma *pSma, const void *pMsg, int32_t msgSize, int32_t inputType, SRSmaInfo *pInfo,
C
Cary Xu 已提交
804
                                 ERsmaExecType type, int8_t level) {
C
Cary Xu 已提交
805
  int32_t idx = level - 1;
K
kailixu 已提交
806
  void   *qTaskInfo = (type == RSMA_EXEC_COMMIT) ? RSMA_INFO_IQTASK(pInfo, idx) : RSMA_INFO_QTASK(pInfo, idx);
C
Cary Xu 已提交
807 808 809
  if (!qTaskInfo) {
    smaDebug("vgId:%d, no qTaskInfo to execute rsma %" PRIi8 " task for suid:%" PRIu64, SMA_VID(pSma), level,
             pInfo->suid);
810 811
    return TSDB_CODE_SUCCESS;
  }
C
Cary Xu 已提交
812
  if (!pInfo->pTSchema) {
K
kailixu 已提交
813
    terrno = TSDB_CODE_INVALID_PTR;
C
Cary Xu 已提交
814
    smaWarn("vgId:%d, no schema to execute rsma %" PRIi8 " task for suid:%" PRIu64, SMA_VID(pSma), level, pInfo->suid);
815 816
    return TSDB_CODE_FAILED;
  }
817

K
kailixu 已提交
818 819
  smaDebug("vgId:%d, execute rsma %" PRIi8 " task for qTaskInfo:%p suid:%" PRIu64 " nMsg:%d", SMA_VID(pSma), level,
           RSMA_INFO_QTASK(pInfo, idx), pInfo->suid, msgSize);
820

C
Cary Xu 已提交
821 822 823
#if 0
  for (int32_t i = 0; i < msgSize; ++i) {
    SSubmitReq *pReq = *(SSubmitReq **)((char *)pMsg + i * sizeof(void *));
C
Cary Xu 已提交
824
    smaDebug("vgId:%d, [%d][%d] version %" PRIi64, SMA_VID(pSma), msgSize, i, pReq->version);
C
Cary Xu 已提交
825 826 827
    tdRsmaPrintSubmitReq(pSma, pReq);
  }
#endif
K
kailixu 已提交
828
  if ((terrno = qSetSMAInput(qTaskInfo, pMsg, msgSize, inputType)) < 0) {
829
    smaError("vgId:%d, rsma %" PRIi8 " qSetStreamInput failed since %s", SMA_VID(pSma), level, tstrerror(terrno));
830 831 832
    return TSDB_CODE_FAILED;
  }

C
Cary Xu 已提交
833
  SRSmaInfoItem *pItem = RSMA_INFO_ITEM(pInfo, idx);
C
Cary Xu 已提交
834
  tdRSmaExecAndSubmitResult(pSma, qTaskInfo, pItem, pInfo->pTSchema, pInfo->suid);
835 836 837 838

  return TSDB_CODE_SUCCESS;
}

839 840
static int32_t tdCloneQTaskInfo(SSma *pSma, qTaskInfo_t dstTaskInfo, qTaskInfo_t srcTaskInfo, SRSmaParam *param,
                                tb_uid_t suid, int8_t idx) {
K
kailixu 已提交
841 842
  int32_t code = 0;
  int32_t lino = 0;
843 844 845 846
  SVnode *pVnode = pSma->pVnode;
  char   *pOutput = NULL;
  int32_t len = 0;

K
kailixu 已提交
847
  if (!srcTaskInfo) {
K
kailixu 已提交
848
    code = TSDB_CODE_INVALID_PTR;
K
kailixu 已提交
849
    smaWarn("vgId:%d, rsma clone, table %" PRIi64 ", no need since srcTaskInfo is NULL", TD_VID(pVnode), suid);
K
kailixu 已提交
850
    TSDB_CHECK_CODE(code, lino, _exit);
K
kailixu 已提交
851 852
  }

K
kailixu 已提交
853 854
  code = qSerializeTaskStatus(srcTaskInfo, &pOutput, &len);
  TSDB_CHECK_CODE(code, lino, _exit);
855 856 857 858 859 860

  SReadHandle handle = {
      .meta = pVnode->pMeta,
      .vnode = pVnode,
      .initTqReader = 1,
  };
K
kailixu 已提交
861

K
kailixu 已提交
862
  if (ASSERTS(!dstTaskInfo, "dstTaskInfo:%p is not NULL", dstTaskInfo)) {
K
kailixu 已提交
863 864 865 866
    code = TSDB_CODE_APP_ERROR;
    TSDB_CHECK_CODE(code, lino, _exit);
  }

D
dapan1121 已提交
867
  dstTaskInfo = qCreateStreamExecTaskInfo(param->qmsg[idx], &handle, TD_VID(pVnode));
868
  if (!dstTaskInfo) {
K
kailixu 已提交
869 870
    code = TSDB_CODE_RSMA_QTASKINFO_CREATE;
    TSDB_CHECK_CODE(code, lino, _exit);
871 872
  }

K
kailixu 已提交
873 874
  code = qDeserializeTaskStatus(dstTaskInfo, pOutput, len);
  TSDB_CHECK_CODE(code, lino, _exit);
875 876 877

  smaDebug("vgId:%d, rsma clone, restore rsma task for table:%" PRIi64 " succeed", TD_VID(pVnode), suid);

K
kailixu 已提交
878
_exit:
879
  taosMemoryFreeClear(pOutput);
K
kailixu 已提交
880 881 882 883 884 885
  if (code) {
    tdRSmaQTaskInfoFree(dstTaskInfo, TD_VID(pVnode), idx + 1);
    smaError("vgId:%d, rsma clone, restore rsma task for table:%" PRIi64 " failed since %s", TD_VID(pVnode), suid,
             terrstr());
  }
  return code;
886 887 888 889 890 891 892 893 894 895
}

/**
 * @brief Clone qTaskInfo of SRSmaInfo
 *
 * @param pSma
 * @param pInfo
 * @return int32_t
 */
static int32_t tdRSmaInfoClone(SSma *pSma, SRSmaInfo *pInfo) {
K
kailixu 已提交
896 897
  int32_t     code = 0;
  int32_t     lino = 0;
898
  SRSmaParam *param = NULL;
K
kailixu 已提交
899 900
  SMetaReader mr = {0};

901 902 903 904 905 906
  if (!pInfo) {
    return TSDB_CODE_SUCCESS;
  }

  metaReaderInit(&mr, SMA_META(pSma), 0);
  smaDebug("vgId:%d, rsma clone qTaskInfo for suid:%" PRIi64, SMA_VID(pSma), pInfo->suid);
H
Haojun Liao 已提交
907
  if (metaGetTableEntryByUidCache(&mr, pInfo->suid) < 0) {
K
kailixu 已提交
908 909 910
    code = terrno;
    TSDB_CHECK_CODE(code, lino, _exit);
  }
K
kailixu 已提交
911

K
kailixu 已提交
912 913 914 915 916 917 918
  if (mr.me.type != TSDB_SUPER_TABLE) {
    code = TSDB_CODE_RSMA_INVALID_SCHEMA;
    TSDB_CHECK_CODE(code, lino, _exit);
  }
  if (mr.me.uid != pInfo->suid) {
    code = TSDB_CODE_RSMA_INVALID_SCHEMA;
    TSDB_CHECK_CODE(code, lino, _exit);
919
  }
K
kailixu 已提交
920

921 922 923 924 925 926
  if (TABLE_IS_ROLLUP(mr.me.flags)) {
    param = &mr.me.stbEntry.rsmaParam;
    for (int32_t i = 0; i < TSDB_RETENTION_L2; ++i) {
      if (!pInfo->iTaskInfo[i]) {
        continue;
      }
K
kailixu 已提交
927 928
      code = tdCloneQTaskInfo(pSma, pInfo->taskInfo[i], pInfo->iTaskInfo[i], param, pInfo->suid, i);
      TSDB_CHECK_CODE(code, lino, _exit);
929 930 931
    }
    smaDebug("vgId:%d, rsma clone env success for %" PRIi64, SMA_VID(pSma), pInfo->suid);
  } else {
K
kailixu 已提交
932 933
    code = TSDB_CODE_RSMA_INVALID_SCHEMA;
    TSDB_CHECK_CODE(code, lino, _exit);
934 935
  }

K
kailixu 已提交
936 937
_exit:
  if (code) {
K
kailixu 已提交
938 939
    smaError("vgId:%d, %s failed at line %d since %s, suid:%" PRIi64 ", flags:%" PRIi8 ",type:%" PRIi8 ", uid:%" PRIi64,
             SMA_VID(pSma), __func__, lino, tstrerror(code), pInfo->suid, mr.me.flags, mr.me.type, mr.me.uid);
K
kailixu 已提交
940
  }
941
  metaReaderClear(&mr);
K
kailixu 已提交
942
  return code;
943 944
}

C
Cary Xu 已提交
945 946
/**
 * @brief During async commit, the SRSmaInfo object would be COW from iRSmaInfoHash and write lock should be applied.
C
Cary Xu 已提交
947 948 949 950
 *
 * @param pSma
 * @param suid
 * @return SRSmaInfo*
C
Cary Xu 已提交
951
 */
C
Cary Xu 已提交
952
static SRSmaInfo *tdAcquireRSmaInfoBySuid(SSma *pSma, int64_t suid) {
K
kailixu 已提交
953 954
  int32_t    code = 0;
  int32_t    lino = 0;
955 956
  SSmaEnv   *pEnv = SMA_RSMA_ENV(pSma);
  SRSmaStat *pStat = NULL;
C
Cary Xu 已提交
957 958
  SRSmaInfo *pRSmaInfo = NULL;

K
kailixu 已提交
959 960
  terrno = 0;

961
  if (!pEnv) {
C
Cary Xu 已提交
962
    terrno = TSDB_CODE_RSMA_INVALID_ENV;
963
    return NULL;
964 965
  }

966 967
  pStat = (SRSmaStat *)SMA_ENV_STAT(pEnv);
  if (!pStat || !RSMA_INFO_HASH(pStat)) {
C
Cary Xu 已提交
968
    terrno = TSDB_CODE_RSMA_INVALID_STAT;
969 970
    return NULL;
  }
971

972
  taosRLockLatch(SMA_ENV_LOCK(pEnv));
C
Cary Xu 已提交
973 974
  pRSmaInfo = taosHashGet(RSMA_INFO_HASH(pStat), &suid, sizeof(tb_uid_t));
  if (pRSmaInfo && (pRSmaInfo = *(SRSmaInfo **)pRSmaInfo)) {
C
Cary Xu 已提交
975
    if (RSMA_INFO_IS_DEL(pRSmaInfo)) {
976
      taosRUnLockLatch(SMA_ENV_LOCK(pEnv));
C
Cary Xu 已提交
977 978
      return NULL;
    }
C
Cary Xu 已提交
979
    if (!pRSmaInfo->taskInfo[0]) {
K
kailixu 已提交
980
      if ((terrno = tdRSmaInfoClone(pSma, pRSmaInfo)) < 0) {
981
        taosRUnLockLatch(SMA_ENV_LOCK(pEnv));
C
Cary Xu 已提交
982 983 984
        return NULL;
      }
    }
C
Cary Xu 已提交
985
    tdRefRSmaInfo(pSma, pRSmaInfo);
986
    taosRUnLockLatch(SMA_ENV_LOCK(pEnv));
K
kailixu 已提交
987
    if (ASSERTS(pRSmaInfo->suid == suid, "suid:%" PRIi64 " != %" PRIi64, pRSmaInfo->suid, suid)) {
K
kailixu 已提交
988 989 990
      terrno = TSDB_CODE_APP_ERROR;
      return NULL;
    }
C
Cary Xu 已提交
991 992
    return pRSmaInfo;
  }
993
  taosRUnLockLatch(SMA_ENV_LOCK(pEnv));
C
Cary Xu 已提交
994

C
Cary Xu 已提交
995
  return NULL;
996 997
}

C
Cary Xu 已提交
998 999 1000
static FORCE_INLINE void tdReleaseRSmaInfo(SSma *pSma, SRSmaInfo *pInfo) {
  if (pInfo) {
    tdUnRefRSmaInfo(pSma, pInfo);
C
Cary Xu 已提交
1001 1002 1003
  }
}

C
Cary Xu 已提交
1004 1005 1006 1007
/**
 * @brief async mode
 *
 * @param pSma
K
kailixu 已提交
1008
 * @param version
C
Cary Xu 已提交
1009 1010 1011 1012 1013
 * @param pMsg
 * @param inputType
 * @param suid
 * @return int32_t
 */
K
kailixu 已提交
1014 1015
static int32_t tdExecuteRSmaAsync(SSma *pSma, int64_t version, const void *pMsg, int32_t len, int32_t inputType,
                                  tb_uid_t suid) {
C
Cary Xu 已提交
1016 1017 1018 1019 1020 1021 1022
  SRSmaInfo *pRSmaInfo = tdAcquireRSmaInfoBySuid(pSma, suid);
  if (!pRSmaInfo) {
    smaDebug("vgId:%d, execute rsma, no rsma info for suid:%" PRIu64, SMA_VID(pSma), suid);
    return TSDB_CODE_SUCCESS;
  }

  if (inputType == STREAM_INPUT__DATA_SUBMIT) {
K
kailixu 已提交
1023
    if (tdExecuteRSmaImplAsync(pSma, version, pMsg, len, inputType, pRSmaInfo, suid) < 0) {
C
Cary Xu 已提交
1024 1025 1026
      tdReleaseRSmaInfo(pSma, pRSmaInfo);
      return TSDB_CODE_FAILED;
    }
C
Cary Xu 已提交
1027 1028 1029 1030 1031 1032 1033 1034 1035 1036
    if (smaMgmt.tmrHandle) {
      SRSmaInfoItem *pItem = RSMA_INFO_ITEM(pRSmaInfo, 0);
      if (pItem->level > 0) {
        atomic_store_8(&pItem->triggerStat, TASK_TRIGGER_STAT_ACTIVE);
      }
      pItem = RSMA_INFO_ITEM(pRSmaInfo, 1);
      if (pItem->level > 0) {
        atomic_store_8(&pItem->triggerStat, TASK_TRIGGER_STAT_ACTIVE);
      }
    }
C
Cary Xu 已提交
1037
  } else {
K
kailixu 已提交
1038 1039 1040 1041 1042
    terrno = TSDB_CODE_APP_ERROR;
    tdReleaseRSmaInfo(pSma, pRSmaInfo);
    smaError("vgId:%d, execute rsma, failed for suid:%" PRIu64 " since %s, type:%d", SMA_VID(pSma), suid,
             tstrerror(terrno), inputType);
    return TSDB_CODE_FAILED;
C
Cary Xu 已提交
1043 1044 1045 1046 1047 1048
  }

  tdReleaseRSmaInfo(pSma, pRSmaInfo);
  return TSDB_CODE_SUCCESS;
}

K
kailixu 已提交
1049
int32_t tdProcessRSmaSubmit(SSma *pSma, int64_t version, void *pReq, void *pMsg, int32_t len, int32_t inputType) {
1050 1051 1052 1053 1054
  SSmaEnv *pEnv = SMA_RSMA_ENV(pSma);
  if (!pEnv) {
    // only applicable when rsma env exists
    return TSDB_CODE_SUCCESS;
  }
K
kailixu 已提交
1055

C
Cary Xu 已提交
1056
  STbUidStore uidStore = {0};
1057

L
Liu Jicong 已提交
1058
  if (inputType == STREAM_INPUT__DATA_SUBMIT) {
K
kailixu 已提交
1059
    if (tdFetchSubmitReqSuids(pReq, &uidStore) < 0) {
1060
      smaError("vgId:%d, failed to process rsma submit fetch suid since: %s", SMA_VID(pSma), terrstr());
C
Cary Xu 已提交
1061 1062
      goto _err;
    }
1063 1064

    if (uidStore.suid != 0) {
K
kailixu 已提交
1065
      if (tdExecuteRSmaAsync(pSma, version, pMsg, len, inputType, uidStore.suid) < 0) {
1066
        smaError("vgId:%d, failed to process rsma submit exec 1 since: %s", SMA_VID(pSma), terrstr());
C
Cary Xu 已提交
1067 1068
        goto _err;
      }
1069

C
Cary Xu 已提交
1070 1071
      void *pIter = NULL;
      while ((pIter = taosHashIterate(uidStore.uidHash, pIter))) {
1072
        tb_uid_t *pTbSuid = (tb_uid_t *)taosHashGetKey(pIter, NULL);
K
kailixu 已提交
1073
        if (tdExecuteRSmaAsync(pSma, version, pMsg, len, inputType, *pTbSuid) < 0) {
1074
          smaError("vgId:%d, failed to process rsma submit exec 2 since: %s", SMA_VID(pSma), terrstr());
C
Cary Xu 已提交
1075 1076
          goto _err;
        }
1077 1078 1079
      }
    }
  }
C
Cary Xu 已提交
1080
  tdUidStoreDestory(&uidStore);
1081
  return TSDB_CODE_SUCCESS;
C
Cary Xu 已提交
1082 1083 1084
_err:
  tdUidStoreDestory(&uidStore);
  return TSDB_CODE_FAILED;
1085
}
C
Cary Xu 已提交
1086

C
Cary Xu 已提交
1087 1088 1089 1090 1091 1092 1093
/**
 * @brief retrieve rsma meta and init
 *
 * @param pSma
 * @param nTables number of tables of rsma
 * @return int32_t
 */
C
Cary Xu 已提交
1094
static int32_t tdRSmaRestoreQTaskInfoInit(SSma *pSma, int64_t *nTables) {
K
kailixu 已提交
1095 1096
  int32_t     code = 0;
  int32_t     lino = 0;
C
Cary Xu 已提交
1097 1098 1099 1100
  SVnode     *pVnode = pSma->pVnode;
  SArray     *suidList = NULL;
  STbUidStore uidStore = {0};
  SMetaReader mr = {0};
K
kailixu 已提交
1101
  tb_uid_t    suid = 0;
1102

C
Cary Xu 已提交
1103
  if (!(suidList = taosArrayInit(1, sizeof(tb_uid_t)))) {
K
kailixu 已提交
1104 1105
    code = TSDB_CODE_OUT_OF_MEMORY;
    TSDB_CHECK_CODE(code, lino, _exit);
C
Cary Xu 已提交
1106 1107 1108
  }

  if (vnodeGetStbIdList(pSma->pVnode, 0, suidList) < 0) {
K
kailixu 已提交
1109 1110
    code = terrno;
    TSDB_CHECK_CODE(code, lino, _exit);
1111 1112
  }

C
Cary Xu 已提交
1113 1114
  int64_t arrSize = taosArrayGetSize(suidList);

C
Cary Xu 已提交
1115
  if (arrSize == 0) {
C
Cary Xu 已提交
1116 1117 1118
    if (nTables) {
      *nTables = 0;
    }
C
Cary Xu 已提交
1119 1120
    taosArrayDestroy(suidList);
    smaDebug("vgId:%d, no need to restore rsma env since empty stb id list", TD_VID(pVnode));
1121 1122 1123
    return TSDB_CODE_SUCCESS;
  }

C
Cary Xu 已提交
1124
  int64_t nRsmaTables = 0;
1125
  metaReaderInit(&mr, SMA_META(pSma), 0);
C
Cary Xu 已提交
1126
  if (!(uidStore.tbUids = taosArrayInit(1024, sizeof(tb_uid_t)))) {
K
kailixu 已提交
1127 1128
    code = TSDB_CODE_OUT_OF_MEMORY;
    TSDB_CHECK_CODE(code, lino, _exit);
C
Cary Xu 已提交
1129
  }
K
kailixu 已提交
1130

C
Cary Xu 已提交
1131
  for (int64_t i = 0; i < arrSize; ++i) {
K
kailixu 已提交
1132
    suid = *(tb_uid_t *)taosArrayGet(suidList, i);
C
Cary Xu 已提交
1133
    smaDebug("vgId:%d, rsma restore, suid is %" PRIi64, TD_VID(pVnode), suid);
H
Haojun Liao 已提交
1134
    if (metaGetTableEntryByUidCache(&mr, suid) < 0) {
K
kailixu 已提交
1135 1136
      code = terrno;
      TSDB_CHECK_CODE(code, lino, _exit);
1137
    }
1138
    tDecoderClear(&mr.coder);
K
kailixu 已提交
1139 1140 1141 1142 1143 1144 1145 1146
    if (mr.me.type != TSDB_SUPER_TABLE) {
      code = TSDB_CODE_RSMA_INVALID_SCHEMA;
      TSDB_CHECK_CODE(code, lino, _exit);
    }
    if (mr.me.uid != suid) {
      code = TSDB_CODE_RSMA_INVALID_SCHEMA;
      TSDB_CHECK_CODE(code, lino, _exit);
    }
1147
    if (TABLE_IS_ROLLUP(mr.me.flags)) {
C
Cary Xu 已提交
1148
      ++nRsmaTables;
1149
      SRSmaParam *param = &mr.me.stbEntry.rsmaParam;
C
Cary Xu 已提交
1150 1151 1152 1153
      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]);
1154
      }
1155
      if (tdRSmaProcessCreateImpl(pSma, &mr.me.stbEntry.rsmaParam, suid, mr.me.name) < 0) {
K
kailixu 已提交
1156 1157
        code = terrno;
        TSDB_CHECK_CODE(code, lino, _exit);
1158
      }
C
Cary Xu 已提交
1159 1160 1161 1162

      // reload all ctbUids for suid
      uidStore.suid = suid;
      if (vnodeGetCtbIdList(pVnode, suid, uidStore.tbUids) < 0) {
K
kailixu 已提交
1163 1164
        code = terrno;
        TSDB_CHECK_CODE(code, lino, _exit);
C
Cary Xu 已提交
1165 1166
      }

1167
      if (tdUpdateTbUidList(pVnode->pSma, &uidStore, true) < 0) {
K
kailixu 已提交
1168 1169
        code = terrno;
        TSDB_CHECK_CODE(code, lino, _exit);
C
Cary Xu 已提交
1170 1171 1172 1173
      }

      taosArrayClear(uidStore.tbUids);

C
Cary Xu 已提交
1174
      smaDebug("vgId:%d, rsma restore env success for %" PRIi64, TD_VID(pVnode), suid);
1175 1176 1177
    }
  }

C
Cary Xu 已提交
1178 1179 1180
  if (nTables) {
    *nTables = nRsmaTables;
  }
K
kailixu 已提交
1181 1182 1183 1184 1185
_exit:
  if (code) {
    smaError("vgId:%d, %s failed at line %d since %s, suid:%" PRIi64 ", type:%" PRIi8 ", uid:%" PRIi64, TD_VID(pVnode),
             __func__, lino, tstrerror(code), suid, mr.me.type, mr.me.uid);
  }
C
Cary Xu 已提交
1186 1187
  metaReaderClear(&mr);
  taosArrayDestroy(suidList);
C
Cary Xu 已提交
1188
  tdUidStoreDestory(&uidStore);
K
kailixu 已提交
1189
  return code;
C
Cary Xu 已提交
1190 1191 1192
}

/**
K
kailixu 已提交
1193
 * N.B. the data would be restored from the unified WAL replay procedure
C
Cary Xu 已提交
1194
 */
K
kailixu 已提交
1195
int32_t tdRSmaProcessRestoreImpl(SSma *pSma, int8_t type, int64_t qtaskFileVer, int8_t rollback) {
K
kailixu 已提交
1196 1197 1198 1199 1200 1201 1202 1203 1204 1205 1206 1207
  // step 1: init env
  if (tdCheckAndInitSmaEnv(pSma, TSDB_SMA_TYPE_ROLLUP) != TSDB_CODE_SUCCESS) {
    terrno = TSDB_CODE_TDB_INIT_FAILED;
    return TSDB_CODE_FAILED;
  }

  // step 2: open SRSmaFS for qTaskFiles
  if (tdRSmaFSOpen(pSma, qtaskFileVer, rollback) < 0) {
    goto _err;
  }

  // step 3: iterate all stables to restore the rsma env
C
Cary Xu 已提交
1208
  int64_t nTables = 0;
C
Cary Xu 已提交
1209
  if (tdRSmaRestoreQTaskInfoInit(pSma, &nTables) < 0) {
C
Cary Xu 已提交
1210 1211
    goto _err;
  }
C
Cary Xu 已提交
1212
  if (nTables <= 0) {
1213
    smaDebug("vgId:%d, no need to restore rsma task %" PRIi8 " since no tables", SMA_VID(pSma), type);
C
Cary Xu 已提交
1214 1215
    return TSDB_CODE_SUCCESS;
  }
C
Cary Xu 已提交
1216

1217
  smaInfo("vgId:%d, restore rsma task %" PRIi8 " from qtaskf %" PRIi64 " succeed", SMA_VID(pSma), type, qtaskFileVer);
1218 1219
  return TSDB_CODE_SUCCESS;
_err:
1220 1221
  smaError("vgId:%d, restore rsma task %" PRIi8 "from qtaskf %" PRIi64 " failed since %s", SMA_VID(pSma), type,
           qtaskFileVer, terrstr());
1222 1223 1224
  return TSDB_CODE_FAILED;
}

1225
int32_t tdRSmaPersistExecImpl(SRSmaStat *pRSmaStat, SHashObj *pInfoHash) {
K
kailixu 已提交
1226 1227 1228 1229 1230 1231 1232 1233 1234 1235 1236
  int32_t   code = 0;
  int32_t   lino = 0;
  SSma     *pSma = pRSmaStat->pSma;
  SVnode   *pVnode = pSma->pVnode;
  SArray   *qTaskFArray = NULL;
  int64_t   version = pRSmaStat->commitAppliedVer;
  TdFilePtr pOutFD = NULL;
  TdFilePtr pInFD = NULL;
  char      fname[TSDB_FILENAME_LEN];
  char      fnameVer[TSDB_FILENAME_LEN];
  SRSmaFS   fs = {0};
1237 1238 1239 1240 1241

  if (taosHashGetSize(pInfoHash) <= 0) {
    return TSDB_CODE_SUCCESS;
  }

K
kailixu 已提交
1242 1243 1244 1245
  qTaskFArray = taosArrayInit(taosHashGetSize(pInfoHash) << 1, sizeof(SQTaskFile));
  if (!qTaskFArray) {
    code = TSDB_CODE_OUT_OF_MEMORY;
    TSDB_CHECK_CODE(code, lino, _exit);
1246 1247 1248 1249 1250 1251 1252 1253 1254 1255 1256 1257 1258 1259
  }

  void *infoHash = NULL;
  while ((infoHash = taosHashIterate(pInfoHash, infoHash))) {
    SRSmaInfo *pRSmaInfo = *(SRSmaInfo **)infoHash;

    if (RSMA_INFO_IS_DEL(pRSmaInfo)) {
      continue;
    }

    for (int32_t i = 0; i < TSDB_RETENTION_L2; ++i) {
      SRSmaInfoItem *pItem = RSMA_INFO_ITEM(pRSmaInfo, i);
      if (pItem && pItem->pStreamState) {
        if (streamStateCommit(pItem->pStreamState) < 0) {
K
kailixu 已提交
1260 1261 1262
          code = TSDB_CODE_RSMA_STREAM_STATE_COMMIT;
          TSDB_CHECK_CODE(code, lino, _exit);
        }
1263
        smaDebug("vgId:%d, rsma persist, stream state commit success, table %" PRIi64 ", level %d", TD_VID(pVnode),
K
kailixu 已提交
1264 1265 1266 1267 1268 1269 1270 1271 1272 1273 1274 1275 1276 1277 1278 1279 1280 1281 1282 1283 1284
                 pRSmaInfo->suid, i + 1);

        // qTaskInfo file
        tdRSmaQTaskInfoGetFullName(TD_VID(pVnode), pRSmaInfo->suid, i + 1, -1, tfsGetPrimaryPath(pVnode->pTfs), fname);
        tdRSmaQTaskInfoGetFullName(TD_VID(pVnode), pRSmaInfo->suid, i + 1, version, tfsGetPrimaryPath(pVnode->pTfs),
                                   fnameVer);
        if (taosCheckExistFile(fnameVer)) {
          smaWarn("vgId:%d, rsma persist, duplicate file %s exist", TD_VID(pVnode), fnameVer);
        }

        pOutFD = taosCreateFile(fnameVer, TD_FILE_WRITE | TD_FILE_CREATE | TD_FILE_TRUNC);
        if (pOutFD == NULL) {
          code = TAOS_SYSTEM_ERROR(errno);
          TSDB_CHECK_CODE(code, lino, _exit);
        }
        pInFD = taosOpenFile(fname, TD_FILE_READ);
        if (pInFD == NULL) {
          code = TAOS_SYSTEM_ERROR(errno);
          TSDB_CHECK_CODE(code, lino, _exit);
        }

1285 1286 1287
        int64_t  size = 0;
        uint32_t mtime = 0;
        if (taosFStatFile(pInFD, &size, &mtime) < 0) {
K
kailixu 已提交
1288 1289
          code = TAOS_SYSTEM_ERROR(errno);
          TSDB_CHECK_CODE(code, lino, _exit);
1290
        }
K
kailixu 已提交
1291 1292 1293 1294

        int64_t offset = 0;
        if (taosFSendFile(pOutFD, pInFD, &offset, size) < 0) {
          code = TAOS_SYSTEM_ERROR(errno);
1295 1296
          smaError("vgId:%d, rsma persist, send qtaskinfo file %s to %s failed since %s", TD_VID(pVnode), fname,
                   fnameVer, tstrerror(code));
K
kailixu 已提交
1297 1298 1299 1300 1301
          TSDB_CHECK_CODE(code, lino, _exit);
        }
        taosCloseFile(&pOutFD);
        taosCloseFile(&pInFD);

1302 1303 1304
        SQTaskFile qTaskF = {
            .nRef = 1, .level = i + 1, .suid = pRSmaInfo->suid, .version = version, .size = size, .mtime = mtime};

K
kailixu 已提交
1305
        taosArrayPush(qTaskFArray, &qTaskF);
1306 1307 1308 1309
      }
    }
  }

K
kailixu 已提交
1310
  // prepare
K
kailixu 已提交
1311
  code = tdRSmaFSCopy(pSma, &fs);
K
kailixu 已提交
1312
  TSDB_CHECK_CODE(code, lino, _exit);
1313

K
kailixu 已提交
1314
  code = tdRSmaFSUpsertQTaskFile(pSma, &fs, qTaskFArray->pData, taosArrayGetSize(qTaskFArray));
K
kailixu 已提交
1315
  TSDB_CHECK_CODE(code, lino, _exit);
1316

K
kailixu 已提交
1317 1318 1319 1320 1321 1322 1323
  code = tdRSmaFSPrepareCommit(pSma, &fs);
  TSDB_CHECK_CODE(code, lino, _exit);

_exit:

  taosArrayDestroy(fs.aQTaskInf);
  taosArrayDestroy(qTaskFArray);
1324

K
kailixu 已提交
1325 1326 1327 1328 1329
  if (code) {
    if (pOutFD) taosCloseFile(&pOutFD);
    if (pInFD) taosCloseFile(&pInFD);
    smaError("vgId:%d, %s failed at line %d since %s", TD_VID(pVnode), __func__, lino, tstrerror(code));
  }
1330

K
kailixu 已提交
1331 1332
  terrno = code;
  return code;
1333 1334
}

C
Cary Xu 已提交
1335
/**
C
Cary Xu 已提交
1336
 * @brief trigger to get rsma result in async mode
C
Cary Xu 已提交
1337 1338 1339 1340
 *
 * @param param
 * @param tmrId
 */
C
Cary Xu 已提交
1341
static void tdRSmaFetchTrigger(void *param, void *tmrId) {
1342
  SRSmaRef      *pRSmaRef = NULL;
C
Cary Xu 已提交
1343
  SSma          *pSma = NULL;
1344 1345 1346
  SRSmaStat     *pStat = NULL;
  SRSmaInfo     *pRSmaInfo = NULL;
  SRSmaInfoItem *pItem = NULL;
C
Cary Xu 已提交
1347

1348
  if (!(pRSmaRef = taosHashGet(smaMgmt.refHash, &param, POINTER_BYTES))) {
C
Cary Xu 已提交
1349 1350
    smaDebug("rsma fetch task not start since rsma info item:%p not exist in refHash:%p, rsetId:%d", param,
             smaMgmt.refHash, smaMgmt.rsetId);
C
Cary Xu 已提交
1351 1352 1353
    return;
  }

1354
  if (!(pStat = (SRSmaStat *)tdAcquireSmaRef(smaMgmt.rsetId, pRSmaRef->refId))) {
C
Cary Xu 已提交
1355
    smaDebug("rsma fetch task not start since rsma stat already destroyed, rsetId:%d refId:%" PRIi64 ")",
C
Cary Xu 已提交
1356 1357
             smaMgmt.rsetId, pRSmaRef->refId);  // pRSmaRef freed in taosHashRemove
    taosHashRemove(smaMgmt.refHash, &param, POINTER_BYTES);
C
Cary Xu 已提交
1358
    return;
C
Cary Xu 已提交
1359 1360
  }

C
Cary Xu 已提交
1361 1362
  pSma = pStat->pSma;

1363
  if (!(pRSmaInfo = tdAcquireRSmaInfoBySuid(pSma, pRSmaRef->suid))) {
C
Cary Xu 已提交
1364
    smaDebug("rsma fetch task not start since rsma info not exist, rsetId:%d refId:%" PRIi64 ")", smaMgmt.rsetId,
1365 1366 1367 1368 1369 1370 1371
             pRSmaRef->refId);  // pRSmaRef freed in taosHashRemove
    tdReleaseSmaRef(smaMgmt.rsetId, pRSmaRef->refId);
    taosHashRemove(smaMgmt.refHash, &param, POINTER_BYTES);
    return;
  }

  if (RSMA_INFO_IS_DEL(pRSmaInfo)) {
C
Cary Xu 已提交
1372
    smaDebug("rsma fetch task not start since rsma info already deleted, rsetId:%d refId:%" PRIi64 ")", smaMgmt.rsetId,
1373 1374 1375 1376 1377 1378 1379 1380 1381
             pRSmaRef->refId);  // pRSmaRef freed in taosHashRemove
    tdReleaseRSmaInfo(pSma, pRSmaInfo);
    tdReleaseSmaRef(smaMgmt.rsetId, pRSmaRef->refId);
    taosHashRemove(smaMgmt.refHash, &param, POINTER_BYTES);
    return;
  }

  pItem = *(SRSmaInfoItem **)&param;

C
Cary Xu 已提交
1382 1383 1384 1385 1386
  // if rsma trigger stat in paused, cancelled or finished, not start fetch task
  int8_t rsmaTriggerStat = atomic_load_8(RSMA_TRIGGER_STAT(pStat));
  switch (rsmaTriggerStat) {
    case TASK_TRIGGER_STAT_PAUSED:
    case TASK_TRIGGER_STAT_CANCELLED: {
C
Cary Xu 已提交
1387
      smaDebug("vgId:%d, rsma fetch task not start for level %" PRIi8 " since stat is %" PRIi8
C
Cary Xu 已提交
1388
               ", rsetId:%d refId:%" PRIi64,
1389
               SMA_VID(pSma), pItem->level, rsmaTriggerStat, smaMgmt.rsetId, pRSmaRef->refId);
C
Cary Xu 已提交
1390
      if (rsmaTriggerStat == TASK_TRIGGER_STAT_PAUSED) {
1391
        taosTmrReset(tdRSmaFetchTrigger, RSMA_FETCH_INTERVAL, pItem, smaMgmt.tmrHandle, &pItem->tmrId);
C
Cary Xu 已提交
1392
      }
1393 1394
      tdReleaseRSmaInfo(pSma, pRSmaInfo);
      tdReleaseSmaRef(smaMgmt.rsetId, pRSmaRef->refId);
C
Cary Xu 已提交
1395
      return;
C
Cary Xu 已提交
1396
    }
C
Cary Xu 已提交
1397 1398 1399 1400 1401 1402 1403 1404
    default:
      break;
  }

  int8_t fetchTriggerStat =
      atomic_val_compare_exchange_8(&pItem->triggerStat, TASK_TRIGGER_STAT_ACTIVE, TASK_TRIGGER_STAT_INACTIVE);
  switch (fetchTriggerStat) {
    case TASK_TRIGGER_STAT_ACTIVE: {
1405
      smaDebug("vgId:%d, rsma fetch task planned for level:%" PRIi8 " suid:%" PRIi64 " since stat is active",
C
Cary Xu 已提交
1406
               SMA_VID(pSma), pItem->level, pRSmaInfo->suid);
C
Cary Xu 已提交
1407
      // async process
C
Cary Xu 已提交
1408
      pItem->fetchLevel = pItem->level;
C
Cary Xu 已提交
1409
#if 0
K
kailixu 已提交
1410
      // debugging codes
C
Cary Xu 已提交
1411 1412
      SRSmaInfo     *qInfo = tdAcquireRSmaInfoBySuid(pSma, pRSmaInfo->suid);
      SRSmaInfoItem *qItem = RSMA_INFO_ITEM(qInfo, pItem->level - 1);
K
kailixu 已提交
1413 1414
      make sure(qItem->level == pItem->level);
      make sure(qItem->fetchLevel == pItem->fetchLevel);
C
Cary Xu 已提交
1415
#endif
1416 1417 1418
      if (atomic_load_8(&pRSmaInfo->assigned) == 0) {
        tsem_post(&(pStat->notEmpty));
      }
C
Cary Xu 已提交
1419 1420
    } break;
    case TASK_TRIGGER_STAT_INACTIVE: {
K
kailixu 已提交
1421
      smaDebug("vgId:%d, rsma fetch task not start for level:%" PRIi8 " suid:%" PRIi64 " since stat is inactive ",
C
Cary Xu 已提交
1422 1423 1424
               SMA_VID(pSma), pItem->level, pRSmaInfo->suid);
    } break;
    case TASK_TRIGGER_STAT_INIT: {
C
Cary Xu 已提交
1425 1426
      smaDebug("vgId:%d, rsma fetch task not start for level:%" PRIi8 " suid::%" PRIi64 " since stat is init",
               SMA_VID(pSma), pItem->level, pRSmaInfo->suid);
C
Cary Xu 已提交
1427 1428
    } break;
    default: {
K
kailixu 已提交
1429 1430 1431
      smaDebug("vgId:%d, rsma fetch task not start for level:%" PRIi8 " suid:%" PRIi64 " since stat:%" PRIi8
               " is unknown",
               SMA_VID(pSma), pItem->level, pRSmaInfo->suid, fetchTriggerStat);
C
Cary Xu 已提交
1432
    } break;
C
Cary Xu 已提交
1433 1434 1435
  }

_end:
C
Cary Xu 已提交
1436
  taosTmrReset(tdRSmaFetchTrigger, pItem->maxDelay, pItem, smaMgmt.tmrHandle, &pItem->tmrId);
1437 1438
  tdReleaseRSmaInfo(pSma, pRSmaInfo);
  tdReleaseSmaRef(smaMgmt.rsetId, pRSmaRef->refId);
C
Cary Xu 已提交
1439
}
C
Cary Xu 已提交
1440

C
Cary Xu 已提交
1441 1442
static void tdFreeRSmaSubmitItems(SArray *pItems) {
  for (int32_t i = 0; i < taosArrayGetSize(pItems); ++i) {
K
kailixu 已提交
1443
    SPackedData *packData = taosArrayGet(pItems, i);
K
kailixu 已提交
1444
    taosFreeQitem(POINTER_SHIFT(packData->msgStr, -sizeof(int32_t) - sizeof(int64_t)));
C
Cary Xu 已提交
1445
  }
C
Cary Xu 已提交
1446
  taosArrayClear(pItems);
C
Cary Xu 已提交
1447 1448
}

C
Cary Xu 已提交
1449 1450 1451 1452 1453 1454 1455
/**
 * @brief fetch rsma result(consider the efficiency and functionality)
 *
 * @param pSma
 * @param pInfo
 * @return int32_t
 */
1456
static int32_t tdRSmaFetchAllResult(SSma *pSma, SRSmaInfo *pInfo) {
C
Cary Xu 已提交
1457 1458
  SSDataBlock dataBlock = {.info.type = STREAM_GET_ALL};
  for (int8_t i = 1; i <= TSDB_RETENTION_L2; ++i) {
C
Cary Xu 已提交
1459 1460 1461
    SRSmaInfoItem *pItem = RSMA_INFO_ITEM(pInfo, i - 1);
    if (pItem->fetchLevel) {
      pItem->fetchLevel = 0;
C
Cary Xu 已提交
1462 1463 1464 1465
      qTaskInfo_t taskInfo = RSMA_INFO_QTASK(pInfo, i - 1);
      if (!taskInfo) {
        continue;
      }
C
Cary Xu 已提交
1466

C
Cary Xu 已提交
1467
      if ((++pItem->nScanned * pItem->maxDelay) > RSMA_FETCH_DELAY_MAX) {
1468
        smaDebug("vgId:%d, suid:%" PRIi64 " level:%" PRIi8 " nScanned:%" PRIi16 " maxDelay:%d, fetch executed",
C
Cary Xu 已提交
1469
                 SMA_VID(pSma), pInfo->suid, i, pItem->nScanned, pItem->maxDelay);
C
Cary Xu 已提交
1470
      } else {
C
Cary Xu 已提交
1471 1472 1473 1474 1475 1476 1477 1478 1479 1480
        int64_t curMs = taosGetTimestampMs();
        if ((curMs - pInfo->lastRecv) < RSMA_FETCH_ACTIVE_MAX) {
          smaTrace("vgId:%d, suid:%" PRIi64 " level:%" PRIi8 " curMs:%" PRIi64 " lastRecv:%" PRIi64 ", fetch skipped ",
                   SMA_VID(pSma), pInfo->suid, i, curMs, pInfo->lastRecv);
          atomic_store_8(&pItem->triggerStat, TASK_TRIGGER_STAT_ACTIVE);  // restore the active stat
          continue;
        } else {
          smaDebug("vgId:%d, suid:%" PRIi64 " level:%" PRIi8 " curMs:%" PRIi64 " lastRecv:%" PRIi64 ", fetch executed ",
                   SMA_VID(pSma), pInfo->suid, i, curMs, pInfo->lastRecv);
        }
C
Cary Xu 已提交
1481 1482
      }

C
Cary Xu 已提交
1483
      pItem->nScanned = 0;
C
Cary Xu 已提交
1484

L
Liu Jicong 已提交
1485
      if ((terrno = qSetSMAInput(taskInfo, &dataBlock, 1, STREAM_INPUT__DATA_BLOCK)) < 0) {
C
Cary Xu 已提交
1486 1487
        goto _err;
      }
C
Cary Xu 已提交
1488
      if (tdRSmaExecAndSubmitResult(pSma, taskInfo, pItem, pInfo->pTSchema, pInfo->suid) < 0) {
C
Cary Xu 已提交
1489 1490 1491
        goto _err;
      }

1492
      smaDebug("vgId:%d, suid:%" PRIi64 " level:%" PRIi8 " nScanned:%" PRIi16 " maxDelay:%d, fetch finished",
C
Cary Xu 已提交
1493
               SMA_VID(pSma), pInfo->suid, i, pItem->nScanned, pItem->maxDelay);
C
Cary Xu 已提交
1494
    } else {
1495
      smaDebug("vgId:%d, suid:%" PRIi64 " level:%" PRIi8 " nScanned:%" PRIi16
C
Cary Xu 已提交
1496
               " maxDelay:%d, fetch not executed as fetch level is %" PRIi8,
C
Cary Xu 已提交
1497
               SMA_VID(pSma), pInfo->suid, i, pItem->nScanned, pItem->maxDelay, pItem->fetchLevel);
C
Cary Xu 已提交
1498 1499 1500 1501 1502 1503 1504 1505 1506
    }
  }

_end:
  return TSDB_CODE_SUCCESS;
_err:
  return TSDB_CODE_FAILED;
}

C
Cary Xu 已提交
1507 1508 1509 1510 1511 1512
static int32_t tdRSmaBatchExec(SSma *pSma, SRSmaInfo *pInfo, STaosQall *qall, SArray *pSubmitArr, ERsmaExecType type) {
  taosArrayClear(pSubmitArr);
  while (1) {
    void *msg = NULL;
    taosGetQitem(qall, (void **)&msg);
    if (msg) {
K
kailixu 已提交
1513 1514 1515 1516
      SPackedData packData = {.msgLen = *(int32_t *)msg,
                              .ver = *(int64_t *)POINTER_SHIFT(msg, sizeof(int32_t)),
                              .msgStr = POINTER_SHIFT(msg, sizeof(int32_t) + sizeof(int64_t))};

K
kailixu 已提交
1517
      if (!taosArrayPush(pSubmitArr, &packData)) {
C
Cary Xu 已提交
1518 1519 1520 1521 1522 1523 1524 1525 1526 1527 1528 1529 1530 1531 1532 1533 1534 1535 1536
        tdFreeRSmaSubmitItems(pSubmitArr);
        goto _err;
      }
    } else {
      break;
    }
  }

  int32_t size = taosArrayGetSize(pSubmitArr);
  if (size > 0) {
    for (int32_t i = 1; i <= TSDB_RETENTION_L2; ++i) {
      if (tdExecuteRSmaImpl(pSma, pSubmitArr->pData, size, STREAM_INPUT__MERGED_SUBMIT, pInfo, type, i) < 0) {
        goto _err;
      }
    }
    tdFreeRSmaSubmitItems(pSubmitArr);
  }
  return TSDB_CODE_SUCCESS;
_err:
K
kailixu 已提交
1537 1538 1539
  smaError("vgId:%d, batch exec for suid:%" PRIi64 " execType:%d size:%d failed since %s", SMA_VID(pSma), pInfo->suid,
           type, (int32_t)taosArrayGetSize(pSubmitArr), terrstr());
  tdFreeRSmaSubmitItems(pSubmitArr);
C
Cary Xu 已提交
1540 1541 1542 1543 1544 1545 1546 1547 1548 1549 1550 1551
  while (1) {
    void *msg = NULL;
    taosGetQitem(qall, (void **)&msg);
    if (msg) {
      taosFreeQitem(msg);
    } else {
      break;
    }
  }
  return TSDB_CODE_FAILED;
}

C
Cary Xu 已提交
1552 1553 1554 1555
/**
 * @brief
 *
 * @param pSma
C
Cary Xu 已提交
1556
 * @param type
C
Cary Xu 已提交
1557 1558
 * @return int32_t
 */
C
Cary Xu 已提交
1559

C
Cary Xu 已提交
1560
int32_t tdRSmaProcessExecImpl(SSma *pSma, ERsmaExecType type) {
K
kailixu 已提交
1561 1562
  int32_t    code = 0;
  int32_t    lino = 0;
C
Cary Xu 已提交
1563
  SVnode    *pVnode = pSma->pVnode;
C
Cary Xu 已提交
1564 1565 1566 1567
  SSmaEnv   *pEnv = SMA_RSMA_ENV(pSma);
  SRSmaStat *pRSmaStat = (SRSmaStat *)SMA_ENV_STAT(pEnv);
  SHashObj  *infoHash = NULL;
  SArray    *pSubmitArr = NULL;
C
Cary Xu 已提交
1568
  bool       isFetchAll = false;
C
Cary Xu 已提交
1569 1570

  if (!pRSmaStat || !(infoHash = RSMA_INFO_HASH(pRSmaStat))) {
K
kailixu 已提交
1571 1572
    code = TSDB_CODE_RSMA_INVALID_STAT;
    TSDB_CHECK_CODE(code, lino, _exit);
C
Cary Xu 已提交
1573 1574
  }

C
Cary Xu 已提交
1575
  if (!(pSubmitArr =
K
kailixu 已提交
1576
            taosArrayInit(TMIN(RSMA_SUBMIT_BATCH_SIZE, atomic_load_64(&pRSmaStat->nBufItems)), sizeof(SPackedData)))) {
K
kailixu 已提交
1577 1578
    code = TSDB_CODE_OUT_OF_MEMORY;
    TSDB_CHECK_CODE(code, lino, _exit);
C
Cary Xu 已提交
1579 1580
  }

C
Cary Xu 已提交
1581 1582
  while (true) {
    // step 1: rsma exec - consume data in buffer queue for all suids
C
Cary Xu 已提交
1583
    if (type == RSMA_EXEC_OVERFLOW) {
1584 1585
      void *pIter = NULL;
      while ((pIter = taosHashIterate(infoHash, pIter))) {
C
Cary Xu 已提交
1586
        SRSmaInfo *pInfo = *(SRSmaInfo **)pIter;
1587
        if (atomic_val_compare_exchange_8(&pInfo->assigned, 0, 1) == 0) {
K
kailixu 已提交
1588
          if ((taosQueueItemSize(pInfo->queue) > 0) || RSMA_NEED_FETCH(pInfo)) {
1589 1590 1591 1592 1593
            int32_t batchCnt = -1;
            int32_t batchMax = taosHashGetSize(infoHash) / tsNumOfVnodeRsmaThreads;
            bool    occupied = (batchMax <= 1);
            if (batchMax > 1) {
              batchMax = 100 / batchMax;
C
Cary Xu 已提交
1594
              batchMax = TMAX(batchMax, 4);
C
Cary Xu 已提交
1595
            }
C
Cary Xu 已提交
1596
            while (occupied || (++batchCnt < batchMax)) {    // greedy mode
1597 1598 1599 1600
              taosReadAllQitems(pInfo->queue, pInfo->qall);  // queue has mutex lock
              int32_t qallItemSize = taosQallItemSize(pInfo->qall);
              if (qallItemSize > 0) {
                tdRSmaBatchExec(pSma, pInfo, pInfo->qall, pSubmitArr, type);
1601
                smaDebug("vgId:%d, batchSize:%d, execType:%" PRIi32, SMA_VID(pSma), qallItemSize, type);
1602 1603
              }

K
kailixu 已提交
1604
              if (RSMA_NEED_FETCH(pInfo)) {
1605 1606 1607
                int8_t oldStat = atomic_val_compare_exchange_8(RSMA_COMMIT_STAT(pRSmaStat), 0, 2);
                if (oldStat == 0 ||
                    ((oldStat == 2) && atomic_load_8(RSMA_TRIGGER_STAT(pRSmaStat)) < TASK_TRIGGER_STAT_PAUSED)) {
C
Cary Xu 已提交
1608
                  int32_t oldVal = atomic_fetch_add_32(&pRSmaStat->nFetchAll, 1);
K
kailixu 已提交
1609 1610

                  if (ASSERTS(oldVal >= 0, "oldVal of nFetchAll: %d < 0", oldVal)) {
K
kailixu 已提交
1611 1612 1613
                    code = TSDB_CODE_APP_ERROR;
                    TSDB_CHECK_CODE(code, lino, _exit);
                  }
K
kailixu 已提交
1614 1615 1616 1617 1618 1619 1620 1621

                  int8_t curStat = atomic_load_8(RSMA_COMMIT_STAT(pRSmaStat));
                  if (curStat == 1) {
                    smaDebug("vgId:%d, fetch all not exec as commit stat is %" PRIi8, SMA_VID(pSma), curStat);
                  } else {
                    tdRSmaFetchAllResult(pSma, pInfo);
                  }

1622 1623 1624
                  if (0 == atomic_sub_fetch_32(&pRSmaStat->nFetchAll, 1)) {
                    atomic_store_8(RSMA_COMMIT_STAT(pRSmaStat), 0);
                  }
C
Cary Xu 已提交
1625
                }
1626 1627 1628 1629 1630
              }

              if (qallItemSize > 0) {
                atomic_fetch_sub_64(&pRSmaStat->nBufItems, qallItemSize);
                continue;
K
kailixu 已提交
1631 1632 1633
              }
              if (RSMA_NEED_FETCH(pInfo)) {
                continue;
1634 1635 1636
              }

              break;
C
Cary Xu 已提交
1637 1638
            }
          }
1639
          atomic_val_compare_exchange_8(&pInfo->assigned, 1, 0);
C
Cary Xu 已提交
1640
        }
C
Cary Xu 已提交
1641
      }
C
Cary Xu 已提交
1642
    } else {
K
kailixu 已提交
1643
      ASSERTS(0, "unknown rsma exec type:%d", (int32_t)type);
K
kailixu 已提交
1644 1645
      code = TSDB_CODE_APP_ERROR;
      TSDB_CHECK_CODE(code, lino, _exit);
C
Cary Xu 已提交
1646 1647
    }

C
Cary Xu 已提交
1648
    if (atomic_load_64(&pRSmaStat->nBufItems) <= 0) {
C
Cary Xu 已提交
1649
      if (pEnv->flag & SMA_ENV_FLG_CLOSE) {
C
Cary Xu 已提交
1650 1651
        break;
      }
1652

1653
      tsem_wait(&pRSmaStat->notEmpty);
C
Cary Xu 已提交
1654

1655
      if ((pEnv->flag & SMA_ENV_FLG_CLOSE) && (atomic_load_64(&pRSmaStat->nBufItems) <= 0)) {
1656
        smaDebug("vgId:%d, exec task end, flag:%" PRIi8 ", nBufItems:%" PRIi64, SMA_VID(pSma), pEnv->flag,
1657
                 atomic_load_64(&pRSmaStat->nBufItems));
1658 1659
        break;
      }
C
Cary Xu 已提交
1660
    }
1661

C
Cary Xu 已提交
1662
  }  // end of while(true)
C
Cary Xu 已提交
1663

K
kailixu 已提交
1664
_exit:
C
Cary Xu 已提交
1665
  taosArrayDestroy(pSubmitArr);
K
kailixu 已提交
1666
  if (code) {
K
kailixu 已提交
1667
    smaError("vgId:%d, %s failed at line %d since %s", TD_VID(pVnode), __func__, lino, tstrerror(code));
K
kailixu 已提交
1668 1669
  }
  return code;
C
Cary Xu 已提交
1670
}