smaRollup.c 45.9 KB
Newer Older
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16
/*
 * 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"
K
kailixu 已提交
17
#include "tq.h"
18

L
Liu Jicong 已提交
19 20 21 22 23
#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 已提交
24

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

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

32 33
typedef struct SRSmaQTaskInfoItem SRSmaQTaskInfoItem;

K
kailixu 已提交
34 35
extern int32_t tsdbDoRetention(STsdb *pTsdb, int64_t now);

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

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

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

C
Cary Xu 已提交
72 73 74 75 76 77 78 79 80
/**
 * @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) {
81
  if (pInfo) {
C
Cary Xu 已提交
82
    for (int32_t i = 0; i < TSDB_RETENTION_L2; ++i) {
83
      SRSmaInfoItem *pItem = &pInfo->items[i];
C
Cary Xu 已提交
84

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

91
      if (isDeepFree && pItem->pStreamState) {
dengyihao's avatar
dengyihao 已提交
92
        streamStateClose(pItem->pStreamState, false);
93 94
      }

C
Cary Xu 已提交
95
      if (isDeepFree && pInfo->taskInfo[i]) {
C
Cary Xu 已提交
96
        tdRSmaQTaskInfoFree(&pInfo->taskInfo[i], SMA_VID(pSma), i + 1);
97
      }
98
    }
C
Cary Xu 已提交
99
    if (isDeepFree) {
C
Cary Xu 已提交
100
      taosMemoryFreeClear(pInfo->pTSchema);
C
Cary Xu 已提交
101
    }
C
Cary Xu 已提交
102 103

    if (isDeepFree) {
K
kailixu 已提交
104 105 106 107 108 109 110 111
      if (pInfo->queue) {
        taosCloseQueue(pInfo->queue);
        pInfo->queue = NULL;
      }
      if (pInfo->qall) {
        taosFreeQall(pInfo->qall);
        pInfo->qall = NULL;
      }
C
Cary Xu 已提交
112 113
    }

C
Cary Xu 已提交
114 115
    taosMemoryFree(pInfo);
  }
116

117 118 119 120
  return NULL;
}

static FORCE_INLINE int32_t tdUidStoreInit(STbUidStore **pStore) {
K
kailixu 已提交
121
  *pStore = taosMemoryCalloc(1, sizeof(STbUidStore));
122
  if (*pStore == NULL) {
K
kailixu 已提交
123 124
    terrno = TSDB_CODE_OUT_OF_MEMORY;
    return TSDB_CODE_FAILED;
125
  }
K
kailixu 已提交
126

127 128 129
  return TSDB_CODE_SUCCESS;
}

130
static int32_t tdUpdateTbUidListImpl(SSma *pSma, tb_uid_t *suid, SArray *tbUids, bool isAdd) {
131 132 133 134
  SRSmaInfo *pRSmaInfo = NULL;

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

C
Cary Xu 已提交
140 141 142 143 144
  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 已提交
145
  pRSmaInfo = tdAcquireRSmaInfoBySuid(pSma, *suid);
146 147

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

C
Cary Xu 已提交
153 154
  for (int32_t i = 0; i < TSDB_RETENTION_L2; ++i) {
    if (pRSmaInfo->taskInfo[i]) {
155
      if ((terrno = qUpdateTableListForStreamScanner(pRSmaInfo->taskInfo[i], tbUids, isAdd)) < 0) {
C
Cary Xu 已提交
156
        tdReleaseRSmaInfo(pSma, pRSmaInfo);
C
Cary Xu 已提交
157 158 159 160
        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 已提交
161 162
      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 已提交
163
    }
164 165
  }

C
Cary Xu 已提交
166
  tdReleaseRSmaInfo(pSma, pRSmaInfo);
167 168 169
  return TSDB_CODE_SUCCESS;
}

170
int32_t tdUpdateTbUidList(SSma *pSma, STbUidStore *pStore, bool isAdd) {
171 172 173 174
  if (!pStore || (taosArrayGetSize(pStore->tbUids) == 0)) {
    return TSDB_CODE_SUCCESS;
  }

175
  if (tdUpdateTbUidListImpl(pSma, &pStore->suid, pStore->tbUids, isAdd) != TSDB_CODE_SUCCESS) {
176 177 178 179 180 181 182 183
    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;

184
    if (tdUpdateTbUidListImpl(pSma, pTbSuid, pTbUids, isAdd) != TSDB_CODE_SUCCESS) {
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
      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 已提交
211 212 213
  SRSmaStat *pStat = (SRSmaStat *)SMA_ENV_STAT(pEnv);
  SHashObj  *infoHash = NULL;
  if (!pStat || !(infoHash = RSMA_INFO_HASH(pStat))) {
C
Cary Xu 已提交
214
    terrno = TSDB_CODE_RSMA_INVALID_STAT;
215 216 217 218 219 220 221 222 223
    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)) {
224
    if (tdUidStoreInit(ppStore) < 0) {
225 226 227 228
      return TSDB_CODE_FAILED;
    }
  }

229
  if (tdUidStorePut(*ppStore, suid, &uid) < 0) {
230 231 232 233 234 235 236
    *ppStore = tdUidStoreFree(*ppStore);
    return TSDB_CODE_FAILED;
  }

  return TSDB_CODE_SUCCESS;
}

237
static int32_t tdSetRSmaInfoItemParams(SSma *pSma, SRSmaParam *param, SRSmaStat *pStat, SRSmaInfo *pRSmaInfo,
L
Liu Jicong 已提交
238
                                       int8_t idx) {
C
Cary Xu 已提交
239 240 241
  if ((param->qmsgLen > 0) && param->qmsg[idx]) {
    SRetention *pRetention = SMA_RETENTION(pSma);
    STsdbCfg   *pTsdbCfg = SMA_TSDB_CFG(pSma);
C
Cary Xu 已提交
242
    SVnode     *pVnode = pSma->pVnode;
243 244 245 246
    char        taskInfDir[TSDB_FILENAME_LEN] = {0};
    void       *pStreamState = NULL;

    // set the backend of stream state
K
kailixu 已提交
247 248
    tdRSmaQTaskInfoGetFullPath(pVnode, pRSmaInfo->suid, idx + 1, pVnode->pTfs, taskInfDir);

249
    if (!taosCheckExistFile(taskInfDir)) {
250
      char *s = taosStrdup(taskInfDir);
K
kailixu 已提交
251
      if (taosMulMkDir(s) != 0) {
252 253 254 255 256 257
        terrno = TAOS_SYSTEM_ERROR(errno);
        taosMemoryFree(s);
        return TSDB_CODE_FAILED;
      }
      taosMemoryFree(s);
    }
K
kailixu 已提交
258 259 260 261

    SStreamTask task = {.id.taskId = 0, .id.streamId = 0};  // TODO: assign value
    task.pMeta = pVnode->pTq->pStreamMeta;
    pStreamState = streamStateOpen(taskInfDir, &task, true, -1, -1);
262 263 264 265 266
    if (!pStreamState) {
      terrno = TSDB_CODE_RSMA_STREAM_STATE_OPEN;
      return TSDB_CODE_FAILED;
    }

H
Hongze Cheng 已提交
267
    SReadHandle handle = {.vnode = pVnode, .initTqReader = 1, .pStateBackend = pStreamState};
H
Haojun Liao 已提交
268 269
    initStorageAPI(&handle.api);

270
    pRSmaInfo->taskInfo[idx] = qCreateStreamExecTaskInfo(param->qmsg[idx], &handle, TD_VID(pVnode), 0);
C
Cary Xu 已提交
271
    if (!pRSmaInfo->taskInfo[idx]) {
C
Cary Xu 已提交
272
      terrno = TSDB_CODE_RSMA_QTASKINFO_CREATE;
C
Cary Xu 已提交
273
      return TSDB_CODE_FAILED;
C
Cary Xu 已提交
274
    }
C
Cary Xu 已提交
275
    SRSmaInfoItem *pItem = &(pRSmaInfo->items[idx]);
276
    pItem->triggerStat = TASK_TRIGGER_STAT_ACTIVE;  // fetch the data when reboot
277
    pItem->pStreamState = pStreamState;
C
Cary Xu 已提交
278
    if (param->maxdelay[idx] < TSDB_MIN_ROLLUP_MAX_DELAY) {
K
kailixu 已提交
279 280
      int64_t msInterval =
          convertTimeFromPrecisionToUnit(pRetention[idx + 1].freq, pTsdbCfg->precision, TIME_UNIT_MILLISECOND);
C
Cary Xu 已提交
281
      pItem->maxDelay = (int32_t)msInterval;
C
Cary Xu 已提交
282
    } else {
C
Cary Xu 已提交
283
      pItem->maxDelay = (int32_t)param->maxdelay[idx];
C
Cary Xu 已提交
284
    }
C
Cary Xu 已提交
285 286
    if (pItem->maxDelay > TSDB_MAX_ROLLUP_MAX_DELAY) {
      pItem->maxDelay = TSDB_MAX_ROLLUP_MAX_DELAY;
C
Cary Xu 已提交
287
    }
C
Cary Xu 已提交
288

K
kailixu 已提交
289 290
    pItem->level = idx == 0 ? TSDB_RETENTION_L1 : TSDB_RETENTION_L2;

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

294
    taosTmrReset(tdRSmaFetchTrigger, RSMA_FETCH_INTERVAL, pItem, smaMgmt.tmrHandle, &pItem->tmrId);
295 296

    smaInfo("vgId:%d, item:%p table:%" PRIi64 " level:%" PRIi8 " maxdelay:%" PRIi64 " watermark:%" PRIi64
297
            ", finally maxdelay:%" PRIi32,
298
            TD_VID(pVnode), pItem, pRSmaInfo->suid, (int8_t)(idx + 1), param->maxdelay[idx], param->watermark[idx],
299
            pItem->maxDelay);
C
Cary Xu 已提交
300 301 302 303
  }
  return TSDB_CODE_SUCCESS;
}

304
/**
305
 * @brief for rsam create or restore
306
 *
307 308 309 310
 * @param pSma
 * @param param
 * @param suid
 * @param tbName
311 312
 * @return int32_t
 */
313
int32_t tdRSmaProcessCreateImpl(SSma *pSma, SRSmaParam *param, int64_t suid, const char *tbName) {
314
  if ((param->qmsgLen[0] == 0) && (param->qmsgLen[1] == 0)) {
315
    smaDebug("vgId:%d, no qmsg1/qmsg2 for rollup table %s %" PRIi64, SMA_VID(pSma), tbName, suid);
316 317 318
    return TSDB_CODE_SUCCESS;
  }

K
kailixu 已提交
319
#if 0
C
Cary Xu 已提交
320
  if (tdCheckAndInitSmaEnv(pSma, TSDB_SMA_TYPE_ROLLUP) != TSDB_CODE_SUCCESS) {
321 322 323
    terrno = TSDB_CODE_TDB_INIT_FAILED;
    return TSDB_CODE_FAILED;
  }
K
kailixu 已提交
324
#endif
325 326

  SSmaEnv   *pEnv = SMA_RSMA_ENV(pSma);
C
Cary Xu 已提交
327
  SRSmaStat *pStat = (SRSmaStat *)SMA_ENV_STAT(pEnv);
328 329
  SRSmaInfo *pRSmaInfo = NULL;

330
  pRSmaInfo = taosHashGet(RSMA_INFO_HASH(pStat), &suid, sizeof(tb_uid_t));
331
  if (pRSmaInfo) {
332 333
    smaInfo("vgId:%d, rsma info already exists for table %s, %" PRIi64, SMA_VID(pSma), tbName, suid);
    return TSDB_CODE_SUCCESS;
334 335
  }

336
  // from write queue: single thead
337 338 339 340 341 342
  pRSmaInfo = (SRSmaInfo *)taosMemoryCalloc(1, sizeof(SRSmaInfo));
  if (!pRSmaInfo) {
    terrno = TSDB_CODE_OUT_OF_MEMORY;
    return TSDB_CODE_FAILED;
  }

343
  STSchema *pTSchema = metaGetTbTSchema(SMA_META(pSma), suid, -1, 1);
344 345 346 347
  if (!pTSchema) {
    terrno = TSDB_CODE_TDB_IVD_TB_SCHEMA_VERSION;
    goto _err;
  }
C
Cary Xu 已提交
348
  pRSmaInfo->pSma = pSma;
349
  pRSmaInfo->pTSchema = pTSchema;
C
Cary Xu 已提交
350 351
  pRSmaInfo->suid = suid;
  T_REF_INIT_VAL(pRSmaInfo, 1);
C
Cary Xu 已提交
352

K
kailixu 已提交
353 354 355
  if (!(pRSmaInfo->queue = taosOpenQueue()) || !(pRSmaInfo->qall = taosAllocateQall()) ||
      tdSetRSmaInfoItemParams(pSma, param, pStat, pRSmaInfo, 0) < 0 ||
      tdSetRSmaInfoItemParams(pSma, param, pStat, pRSmaInfo, 1) < 0) {
C
Cary Xu 已提交
356 357
    goto _err;
  }
K
kailixu 已提交
358

359
  if (taosHashPut(RSMA_INFO_HASH(pStat), &suid, sizeof(tb_uid_t), &pRSmaInfo, sizeof(pRSmaInfo)) < 0) {
360
    goto _err;
361 362
  }

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

365
  return TSDB_CODE_SUCCESS;
366
_err:
C
Cary Xu 已提交
367
  tdFreeRSmaInfo(pSma, pRSmaInfo, true);
368
  return TSDB_CODE_FAILED;
369 370
}

371
/**
C
Cary Xu 已提交
372
 * @brief Check and init qTaskInfo_t, only applicable to stable with SRSmaParam currently
373
 *
374
 * @param pSma
375 376 377
 * @param pReq
 * @return int32_t
 */
378 379
int32_t tdProcessRSmaCreate(SSma *pSma, SVCreateStbReq *pReq) {
  SVnode *pVnode = pSma->pVnode;
380
  if (!pReq->rollup) {
381 382 383 384 385 386
    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 已提交
387 388
    smaWarn("vgId:%d, not create rsma for stable %s %" PRIi64 " since vnd is not rsma", TD_VID(pVnode), pReq->name,
            pReq->suid);
389 390 391
    return TSDB_CODE_SUCCESS;
  }

392
  return tdRSmaProcessCreateImpl(pSma, &pReq->rsmaParam, pReq->suid, pReq->name);
393 394
}

395 396 397 398 399 400 401
/**
 * @brief drop cache for stb
 *
 * @param pSma
 * @param pReq
 * @return int32_t
 */
L
Liu Jicong 已提交
402
int32_t tdProcessRSmaDrop(SSma *pSma, SVDropStbReq *pReq) {
403 404
  SVnode *pVnode = pSma->pVnode;
  if (!VND_IS_RSMA(pVnode)) {
C
Cary Xu 已提交
405
    smaTrace("vgId:%d, not drop rsma for stable %s %" PRIi64 " since vnd is not rsma", TD_VID(pVnode), pReq->name,
406 407 408 409
             pReq->suid);
    return TSDB_CODE_SUCCESS;
  }

C
Cary Xu 已提交
410 411 412 413 414
  SSmaEnv *pSmaEnv = SMA_RSMA_ENV(pSma);
  if (!pSmaEnv) {
    return TSDB_CODE_SUCCESS;
  }

C
Cary Xu 已提交
415
  SRSmaStat *pRSmaStat = (SRSmaStat *)SMA_ENV_STAT(pSmaEnv);
C
Cary Xu 已提交
416

C
Cary Xu 已提交
417
  SRSmaInfo *pRSmaInfo = tdAcquireRSmaInfoBySuid(pSma, pReq->suid);
C
Cary Xu 已提交
418 419 420 421 422 423 424 425

  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
426
  atomic_store_8(&pRSmaStat->delFlag, 1);
C
Cary Xu 已提交
427 428 429
  RSMA_INFO_SET_DEL(pRSmaInfo);
  tdUnRefRSmaInfo(pSma, pRSmaInfo);

C
Cary Xu 已提交
430
  tdReleaseRSmaInfo(pSma, pRSmaInfo);
C
Cary Xu 已提交
431

K
kailixu 已提交
432 433
  // 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);
434
  return TSDB_CODE_SUCCESS;
L
Liu Jicong 已提交
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
/**
 * @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 已提交
482
          taosArrayDestroy(pUidArray);
483 484
          return TSDB_CODE_FAILED;
        }
485
        if (taosHashPut(pStore->uidHash, &suid, sizeof(suid), &pUidArray, sizeof(pUidArray)) < 0) {
486 487 488 489
          return TSDB_CODE_FAILED;
        }
      }
    } else {
490
      if (taosHashPut(pStore->uidHash, &suid, sizeof(suid), NULL, 0) < 0) {
491 492 493 494 495 496 497
        return TSDB_CODE_FAILED;
      }
    }
  }
  return TSDB_CODE_SUCCESS;
}

K
kailixu 已提交
498
static void tdUidStoreDestory(STbUidStore *pStore) {
499 500 501 502 503 504 505 506 507 508 509 510 511 512 513 514 515 516 517 518 519 520 521 522 523
  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 已提交
524 525 526 527 528 529 530 531
/**
 * @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
 */
532
static int32_t tdProcessSubmitReq(STsdb *pTsdb, int64_t version, void *pReq) {
K
kailixu 已提交
533 534 535 536 537 538
  if (pReq) {
    SSubmitReq2 *pSubmitReq = (SSubmitReq2 *)pReq;
    // spin lock for race condition during insert data
    if (tsdbInsertData(pTsdb, version, pSubmitReq, NULL) < 0) {
      return TSDB_CODE_FAILED;
    }
539 540 541 542 543
  }

  return TSDB_CODE_SUCCESS;
}

544 545 546
static int32_t tdFetchSubmitReqSuids(SSubmitReq2 *pMsg, STbUidStore *pStore) {
  SArray *pSubmitTbData = pMsg ? pMsg->aSubmitTbData : NULL;
  int32_t size = taosArrayGetSize(pSubmitTbData);
547 548 549

  terrno = TSDB_CODE_SUCCESS;

550 551
  for (int32_t i = 0; i < size; ++i) {
    SSubmitTbData *pData = TARRAY_GET_ELEM(pSubmitTbData, i);
552
    if ((terrno = tdUidStorePut(pStore, pData->suid, NULL)) < 0) {
C
Cary Xu 已提交
553 554
      return -1;
    }
555 556 557 558 559
  }

  return 0;
}

C
Cary Xu 已提交
560 561
/**
 * @brief retention of rsma1/rsma2
C
Cary Xu 已提交
562 563 564 565
 *
 * @param pSma
 * @param now
 * @return int32_t
C
Cary Xu 已提交
566
 */
567
int32_t smaDoRetention(SSma *pSma, int64_t now) {
C
Cary Xu 已提交
568
  int32_t code = TSDB_CODE_SUCCESS;
C
Cary Xu 已提交
569
  if (!VND_IS_RSMA(pSma->pVnode)) {
C
Cary Xu 已提交
570 571 572 573 574
    return code;
  }

  for (int32_t i = 0; i < TSDB_RETENTION_L2; ++i) {
    if (pSma->pRSmaTsdb[i]) {
H
Hongze Cheng 已提交
575 576
      // code = tsdbDoRetention(pSma->pRSmaTsdb[i], now);
      // if (code) goto _end;
C
Cary Xu 已提交
577 578 579 580 581 582 583
    }
  }

_end:
  return code;
}

C
Cary Xu 已提交
584
static int32_t tdRSmaExecAndSubmitResult(SSma *pSma, qTaskInfo_t taskInfo, SRSmaInfoItem *pItem, STSchema *pTSchema,
K
kailixu 已提交
585
                                         int64_t suid) {
K
kailixu 已提交
586 587 588 589
  int32_t      code = 0;
  int32_t      lino = 0;
  SSDataBlock *output = NULL;

H
Haojun Liao 已提交
590 591
  SArray *pResList = taosArrayInit(1, POINTER_BYTES);
  if (pResList == NULL) {
K
kailixu 已提交
592 593
    code = TSDB_CODE_OUT_OF_MEMORY;
    TSDB_CHECK_CODE(code, lino, _exit);
H
Haojun Liao 已提交
594
  }
595

H
Haojun Liao 已提交
596 597
  while (1) {
    uint64_t ts;
598
    bool     hasMore = false;
K
kailixu 已提交
599 600 601 602
    code = qExecTaskOpt(taskInfo, pResList, &ts, &hasMore, NULL);
    if (code == TSDB_CODE_QRY_IN_EXEC) {
      code = 0;
      break;
603
    }
K
kailixu 已提交
604
    TSDB_CHECK_CODE(code, lino, _exit);
C
Cary Xu 已提交
605

H
Haojun Liao 已提交
606 607 608
    if (taosArrayGetSize(pResList) == 0) {
      break;
    }
C
Cary Xu 已提交
609
#if 0
C
Cary Xu 已提交
610 611 612
    char flag[10] = {0};
    snprintf(flag, 10, "level %" PRIi8, pItem->level);
    blockDebugShowDataBlocks(pResList, flag);
C
Cary Xu 已提交
613
#endif
C
Cary Xu 已提交
614
    for (int32_t i = 0; i < taosArrayGetSize(pResList); ++i) {
K
kailixu 已提交
615
      output = taosArrayGetP(pResList, i);
K
kailixu 已提交
616
      smaDebug("vgId:%d, result block, uid:%" PRIu64 ", groupid:%" PRIu64 ", rows:%" PRIi64, SMA_VID(pSma),
K
kailixu 已提交
617
               output->info.id.uid, output->info.id.groupId, output->info.rows);
H
Haojun Liao 已提交
618

619
      STsdb       *sinkTsdb = (pItem->level == TSDB_RETENTION_L1 ? pSma->pRSmaTsdb[0] : pSma->pRSmaTsdb[1]);
620
      SSubmitReq2 *pReq = NULL;
C
Cary Xu 已提交
621 622

      // TODO: the schema update should be handled later(TD-17965)
K
kailixu 已提交
623
      if (buildSubmitReqFromDataBlock(&pReq, output, pTSchema, output->info.id.groupId, SMA_VID(pSma), suid) < 0) {
K
kailixu 已提交
624 625
        code = terrno ? terrno : TSDB_CODE_RSMA_RESULT;
        TSDB_CHECK_CODE(code, lino, _exit);
C
Cary Xu 已提交
626
      }
627

C
Cary Xu 已提交
628
      if (pReq && tdProcessSubmitReq(sinkTsdb, output->info.version, pReq) < 0) {
K
kailixu 已提交
629
        code = terrno ? terrno : TSDB_CODE_RSMA_RESULT;
K
kailixu 已提交
630
        tDestroySubmitReq(pReq, TSDB_MSG_FLG_ENCODE);
K
kailixu 已提交
631
        taosMemoryFree(pReq);
K
kailixu 已提交
632
        TSDB_CHECK_CODE(code, lino, _exit);
C
Cary Xu 已提交
633
      }
C
Cary Xu 已提交
634

635 636
      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);
637

K
kailixu 已提交
638
      if (pReq) {
639
        tDestroySubmitReq(pReq, TSDB_MSG_FLG_ENCODE);
K
kailixu 已提交
640 641
        taosMemoryFree(pReq);
      }
C
Cary Xu 已提交
642
    }
643
  }
K
kailixu 已提交
644 645 646 647 648 649 650 651 652
_exit:
  if (code) {
    smaError("vgId:%d, %s failed at line %d since %s, suid:%" PRIi64 ", level:%" PRIi8 ", uid:%" PRIi64
             ", ver:%" PRIi64,
             SMA_VID(pSma), __func__, lino, tstrerror(code), suid, pItem->level, output ? output->info.id.uid : -1,
             output ? output->info.version : -1);
  } else {
    smaDebug("vgId:%d, %s succeed, suid:%" PRIi64 ", level:%" PRIi8, SMA_VID(pSma), __func__, suid, pItem->level);
  }
H
Haojun Liao 已提交
653 654
  taosArrayDestroy(pResList);
  qCleanExecTaskBlockBuf(taskInfo);
K
kailixu 已提交
655
  return code;
656 657
}

C
Cary Xu 已提交
658
/**
C
Cary Xu 已提交
659
 * @brief Copy msg to rsmaQueueBuffer for batch process
C
Cary Xu 已提交
660 661
 *
 * @param pSma
K
kailixu 已提交
662
 * @param version
C
Cary Xu 已提交
663
 * @param pMsg
664
 * @param len
C
Cary Xu 已提交
665 666 667 668 669
 * @param inputType
 * @param pInfo
 * @param suid
 * @return int32_t
 */
K
kailixu 已提交
670 671 672
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 已提交
673
  void   *qItem = taosAllocateQitem(size, DEF_QITEM, 0);
C
Cary Xu 已提交
674 675 676 677 678

  if (!qItem) {
    return TSDB_CODE_FAILED;
  }

K
kailixu 已提交
679 680 681 682 683 684
  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 已提交
685 686 687

  taosWriteQitem(pInfo->queue, qItem);

C
Cary Xu 已提交
688 689
  pInfo->lastRecv = taosGetTimestampMs();

C
Cary Xu 已提交
690
  SRSmaStat *pRSmaStat = SMA_RSMA_STAT(pSma);
C
Cary Xu 已提交
691 692

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

694 695 696 697
  if (atomic_load_8(&pInfo->assigned) == 0) {
    tsem_post(&(pRSmaStat->notEmpty));
  }

C
Cary Xu 已提交
698
  // smoothing consume
C
Cary Xu 已提交
699
  int32_t n = nItems / RSMA_QTASKEXEC_SMOOTH_SIZE;
C
Cary Xu 已提交
700 701 702 703
  if (n > 1) {
    if (n > 10) {
      n = 10;
    }
C
Cary Xu 已提交
704 705
    taosMsleep(n << 3);
    if (n > 5) {
C
Cary Xu 已提交
706
      smaWarn("vgId:%d, pInfo->queue itemSize:%d, memSize:%" PRIi64 ", sleep %d ms", SMA_VID(pSma),
C
Cary Xu 已提交
707
              taosQueueItemSize(pInfo->queue), taosQueueMemorySize(pInfo->queue), n << 3);
C
Cary Xu 已提交
708 709
    }
  }
C
Cary Xu 已提交
710

C
Cary Xu 已提交
711 712 713
  return TSDB_CODE_SUCCESS;
}

714
#if 0
C
Cary Xu 已提交
715 716 717 718 719 720 721 722 723 724 725
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 已提交
726
      smaDebug("vgId:%d, numOfRows:%d, suid:%" PRIi64 ", uid:%" PRIi64 ", version:%" PRIi64 ", ts:%" PRIi64,
C
Cary Xu 已提交
727 728 729 730 731
               SMA_VID(pSma), msgIter.numOfRows, msgIter.suid, msgIter.uid, pReq->version, row->ts);
    }
  }
  return 0;
}
732
#endif
C
Cary Xu 已提交
733

C
Cary Xu 已提交
734 735 736 737 738 739 740 741
/**
 * @brief sync mode
 *
 * @param pSma
 * @param pMsg
 * @param msgSize
 * @param inputType
 * @param pInfo
C
Cary Xu 已提交
742
 * @param type
C
Cary Xu 已提交
743 744 745 746
 * @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 已提交
747
                                 ERsmaExecType type, int8_t level) {
C
Cary Xu 已提交
748
  int32_t idx = level - 1;
K
kailixu 已提交
749 750
  void   *qTaskInfo = RSMA_INFO_QTASK(pInfo, idx);

C
Cary Xu 已提交
751 752 753
  if (!qTaskInfo) {
    smaDebug("vgId:%d, no qTaskInfo to execute rsma %" PRIi8 " task for suid:%" PRIu64, SMA_VID(pSma), level,
             pInfo->suid);
754 755
    return TSDB_CODE_SUCCESS;
  }
C
Cary Xu 已提交
756
  if (!pInfo->pTSchema) {
K
kailixu 已提交
757
    terrno = TSDB_CODE_INVALID_PTR;
C
Cary Xu 已提交
758
    smaWarn("vgId:%d, no schema to execute rsma %" PRIi8 " task for suid:%" PRIu64, SMA_VID(pSma), level, pInfo->suid);
759 760
    return TSDB_CODE_FAILED;
  }
761

K
kailixu 已提交
762 763
  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);
764

C
Cary Xu 已提交
765 766 767
#if 0
  for (int32_t i = 0; i < msgSize; ++i) {
    SSubmitReq *pReq = *(SSubmitReq **)((char *)pMsg + i * sizeof(void *));
C
Cary Xu 已提交
768
    smaDebug("vgId:%d, [%d][%d] version %" PRIi64, SMA_VID(pSma), msgSize, i, pReq->version);
C
Cary Xu 已提交
769 770 771
    tdRsmaPrintSubmitReq(pSma, pReq);
  }
#endif
K
kailixu 已提交
772
  if ((terrno = qSetSMAInput(qTaskInfo, pMsg, msgSize, inputType)) < 0) {
773
    smaError("vgId:%d, rsma %" PRIi8 " qSetStreamInput failed since %s", SMA_VID(pSma), level, tstrerror(terrno));
774 775 776
    return TSDB_CODE_FAILED;
  }

C
Cary Xu 已提交
777
  SRSmaInfoItem *pItem = RSMA_INFO_ITEM(pInfo, idx);
K
kailixu 已提交
778
  tdRSmaExecAndSubmitResult(pSma, qTaskInfo, pItem, pInfo->pTSchema, pInfo->suid);
779 780 781 782

  return TSDB_CODE_SUCCESS;
}

C
Cary Xu 已提交
783 784
/**
 * @brief During async commit, the SRSmaInfo object would be COW from iRSmaInfoHash and write lock should be applied.
C
Cary Xu 已提交
785 786 787 788
 *
 * @param pSma
 * @param suid
 * @return SRSmaInfo*
C
Cary Xu 已提交
789
 */
C
Cary Xu 已提交
790
static SRSmaInfo *tdAcquireRSmaInfoBySuid(SSma *pSma, int64_t suid) {
K
kailixu 已提交
791 792
  int32_t    code = 0;
  int32_t    lino = 0;
793 794
  SSmaEnv   *pEnv = SMA_RSMA_ENV(pSma);
  SRSmaStat *pStat = NULL;
C
Cary Xu 已提交
795 796
  SRSmaInfo *pRSmaInfo = NULL;

K
kailixu 已提交
797 798
  terrno = 0;

799
  if (!pEnv) {
C
Cary Xu 已提交
800
    terrno = TSDB_CODE_RSMA_INVALID_ENV;
801
    return NULL;
802 803
  }

804 805
  pStat = (SRSmaStat *)SMA_ENV_STAT(pEnv);
  if (!pStat || !RSMA_INFO_HASH(pStat)) {
C
Cary Xu 已提交
806
    terrno = TSDB_CODE_RSMA_INVALID_STAT;
807 808
    return NULL;
  }
809

810
  taosRLockLatch(SMA_ENV_LOCK(pEnv));
C
Cary Xu 已提交
811 812
  pRSmaInfo = taosHashGet(RSMA_INFO_HASH(pStat), &suid, sizeof(tb_uid_t));
  if (pRSmaInfo && (pRSmaInfo = *(SRSmaInfo **)pRSmaInfo)) {
C
Cary Xu 已提交
813
    if (RSMA_INFO_IS_DEL(pRSmaInfo)) {
814
      taosRUnLockLatch(SMA_ENV_LOCK(pEnv));
C
Cary Xu 已提交
815 816
      return NULL;
    }
K
kailixu 已提交
817

C
Cary Xu 已提交
818
    tdRefRSmaInfo(pSma, pRSmaInfo);
819
    taosRUnLockLatch(SMA_ENV_LOCK(pEnv));
K
kailixu 已提交
820
    if (ASSERTS(pRSmaInfo->suid == suid, "suid:%" PRIi64 " != %" PRIi64, pRSmaInfo->suid, suid)) {
K
kailixu 已提交
821 822 823
      terrno = TSDB_CODE_APP_ERROR;
      return NULL;
    }
C
Cary Xu 已提交
824 825
    return pRSmaInfo;
  }
826
  taosRUnLockLatch(SMA_ENV_LOCK(pEnv));
C
Cary Xu 已提交
827

C
Cary Xu 已提交
828
  return NULL;
829 830
}

C
Cary Xu 已提交
831 832 833
static FORCE_INLINE void tdReleaseRSmaInfo(SSma *pSma, SRSmaInfo *pInfo) {
  if (pInfo) {
    tdUnRefRSmaInfo(pSma, pInfo);
C
Cary Xu 已提交
834 835 836
  }
}

C
Cary Xu 已提交
837 838 839 840
/**
 * @brief async mode
 *
 * @param pSma
K
kailixu 已提交
841
 * @param version
C
Cary Xu 已提交
842 843 844 845 846
 * @param pMsg
 * @param inputType
 * @param suid
 * @return int32_t
 */
K
kailixu 已提交
847 848
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 已提交
849 850 851 852 853 854 855
  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 已提交
856
    if (tdExecuteRSmaImplAsync(pSma, version, pMsg, len, inputType, pRSmaInfo, suid) < 0) {
C
Cary Xu 已提交
857 858 859
      tdReleaseRSmaInfo(pSma, pRSmaInfo);
      return TSDB_CODE_FAILED;
    }
C
Cary Xu 已提交
860 861 862 863 864 865 866 867 868 869
    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 已提交
870
  } else {
K
kailixu 已提交
871 872 873 874 875
    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 已提交
876 877 878 879 880 881
  }

  tdReleaseRSmaInfo(pSma, pRSmaInfo);
  return TSDB_CODE_SUCCESS;
}

K
kailixu 已提交
882
int32_t tdProcessRSmaSubmit(SSma *pSma, int64_t version, void *pReq, void *pMsg, int32_t len, int32_t inputType) {
883 884 885 886 887
  SSmaEnv *pEnv = SMA_RSMA_ENV(pSma);
  if (!pEnv) {
    // only applicable when rsma env exists
    return TSDB_CODE_SUCCESS;
  }
K
kailixu 已提交
888

C
Cary Xu 已提交
889
  STbUidStore uidStore = {0};
890

L
Liu Jicong 已提交
891
  if (inputType == STREAM_INPUT__DATA_SUBMIT) {
K
kailixu 已提交
892
    if (tdFetchSubmitReqSuids(pReq, &uidStore) < 0) {
893
      smaError("vgId:%d, failed to process rsma submit fetch suid since: %s", SMA_VID(pSma), terrstr());
C
Cary Xu 已提交
894 895
      goto _err;
    }
896 897

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

C
Cary Xu 已提交
903 904
      void *pIter = NULL;
      while ((pIter = taosHashIterate(uidStore.uidHash, pIter))) {
905
        tb_uid_t *pTbSuid = (tb_uid_t *)taosHashGetKey(pIter, NULL);
K
kailixu 已提交
906
        if (tdExecuteRSmaAsync(pSma, version, pMsg, len, inputType, *pTbSuid) < 0) {
907
          smaError("vgId:%d, failed to process rsma submit exec 2 since: %s", SMA_VID(pSma), terrstr());
C
Cary Xu 已提交
908 909
          goto _err;
        }
910 911 912
      }
    }
  }
C
Cary Xu 已提交
913
  tdUidStoreDestory(&uidStore);
914
  return TSDB_CODE_SUCCESS;
C
Cary Xu 已提交
915 916 917
_err:
  tdUidStoreDestory(&uidStore);
  return TSDB_CODE_FAILED;
918
}
C
Cary Xu 已提交
919

C
Cary Xu 已提交
920 921 922 923 924 925 926
/**
 * @brief retrieve rsma meta and init
 *
 * @param pSma
 * @param nTables number of tables of rsma
 * @return int32_t
 */
C
Cary Xu 已提交
927
static int32_t tdRSmaRestoreQTaskInfoInit(SSma *pSma, int64_t *nTables) {
K
kailixu 已提交
928 929
  int32_t     code = 0;
  int32_t     lino = 0;
C
Cary Xu 已提交
930 931 932 933
  SVnode     *pVnode = pSma->pVnode;
  SArray     *suidList = NULL;
  STbUidStore uidStore = {0};
  SMetaReader mr = {0};
K
kailixu 已提交
934
  tb_uid_t    suid = 0;
935

C
Cary Xu 已提交
936
  if (!(suidList = taosArrayInit(1, sizeof(tb_uid_t)))) {
K
kailixu 已提交
937 938
    code = TSDB_CODE_OUT_OF_MEMORY;
    TSDB_CHECK_CODE(code, lino, _exit);
C
Cary Xu 已提交
939 940 941
  }

  if (vnodeGetStbIdList(pSma->pVnode, 0, suidList) < 0) {
K
kailixu 已提交
942 943
    code = terrno;
    TSDB_CHECK_CODE(code, lino, _exit);
944 945
  }

C
Cary Xu 已提交
946 947
  int64_t arrSize = taosArrayGetSize(suidList);

C
Cary Xu 已提交
948
  if (arrSize == 0) {
C
Cary Xu 已提交
949 950 951
    if (nTables) {
      *nTables = 0;
    }
C
Cary Xu 已提交
952 953
    taosArrayDestroy(suidList);
    smaDebug("vgId:%d, no need to restore rsma env since empty stb id list", TD_VID(pVnode));
954 955 956
    return TSDB_CODE_SUCCESS;
  }

C
Cary Xu 已提交
957
  int64_t nRsmaTables = 0;
958
  metaReaderDoInit(&mr, SMA_META(pSma), 0);
C
Cary Xu 已提交
959
  if (!(uidStore.tbUids = taosArrayInit(1024, sizeof(tb_uid_t)))) {
K
kailixu 已提交
960 961
    code = TSDB_CODE_OUT_OF_MEMORY;
    TSDB_CHECK_CODE(code, lino, _exit);
C
Cary Xu 已提交
962
  }
K
kailixu 已提交
963

C
Cary Xu 已提交
964
  for (int64_t i = 0; i < arrSize; ++i) {
K
kailixu 已提交
965
    suid = *(tb_uid_t *)taosArrayGet(suidList, i);
C
Cary Xu 已提交
966
    smaDebug("vgId:%d, rsma restore, suid is %" PRIi64, TD_VID(pVnode), suid);
967
    if (metaReaderGetTableEntryByUidCache(&mr, suid) < 0) {
K
kailixu 已提交
968 969
      code = terrno;
      TSDB_CHECK_CODE(code, lino, _exit);
970
    }
971
    tDecoderClear(&mr.coder);
K
kailixu 已提交
972 973 974 975 976 977 978 979
    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);
    }
980
    if (TABLE_IS_ROLLUP(mr.me.flags)) {
C
Cary Xu 已提交
981
      ++nRsmaTables;
982
      SRSmaParam *param = &mr.me.stbEntry.rsmaParam;
C
Cary Xu 已提交
983 984 985 986
      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]);
987
      }
988
      if (tdRSmaProcessCreateImpl(pSma, &mr.me.stbEntry.rsmaParam, suid, mr.me.name) < 0) {
K
kailixu 已提交
989 990
        code = terrno;
        TSDB_CHECK_CODE(code, lino, _exit);
991
      }
C
Cary Xu 已提交
992 993 994 995

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

1000
      if (tdUpdateTbUidList(pVnode->pSma, &uidStore, true) < 0) {
K
kailixu 已提交
1001 1002
        code = terrno;
        TSDB_CHECK_CODE(code, lino, _exit);
C
Cary Xu 已提交
1003 1004 1005 1006
      }

      taosArrayClear(uidStore.tbUids);

C
Cary Xu 已提交
1007
      smaDebug("vgId:%d, rsma restore env success for %" PRIi64, TD_VID(pVnode), suid);
1008 1009 1010
    }
  }

C
Cary Xu 已提交
1011 1012 1013
  if (nTables) {
    *nTables = nRsmaTables;
  }
K
kailixu 已提交
1014 1015 1016 1017 1018
_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 已提交
1019 1020
  metaReaderClear(&mr);
  taosArrayDestroy(suidList);
C
Cary Xu 已提交
1021
  tdUidStoreDestory(&uidStore);
K
kailixu 已提交
1022
  return code;
C
Cary Xu 已提交
1023 1024 1025
}

/**
K
kailixu 已提交
1026
 * N.B. the data would be restored from the unified WAL replay procedure
C
Cary Xu 已提交
1027
 */
K
kailixu 已提交
1028
int32_t tdRSmaProcessRestoreImpl(SSma *pSma, int8_t type, int64_t qtaskFileVer, int8_t rollback) {
K
kailixu 已提交
1029 1030 1031
  int32_t code = 0;
  int64_t nTables = 0;

K
kailixu 已提交
1032 1033
  // step 1: init env
  if (tdCheckAndInitSmaEnv(pSma, TSDB_SMA_TYPE_ROLLUP) != TSDB_CODE_SUCCESS) {
K
kailixu 已提交
1034 1035
    code = TSDB_CODE_TDB_INIT_FAILED;
    goto _err;
K
kailixu 已提交
1036 1037
  }

K
kailixu 已提交
1038
  // step 2: iterate all stables to restore the rsma env
K
kailixu 已提交
1039
  if ((code = tdRSmaRestoreQTaskInfoInit(pSma, &nTables)) < 0) {
C
Cary Xu 已提交
1040 1041 1042
    goto _err;
  }

1043
_err:
K
kailixu 已提交
1044 1045 1046 1047 1048 1049 1050 1051 1052
  if (code) {
    smaError("vgId:%d, restore rsma task %" PRIi8 "from qtaskf %" PRIi64 " failed since %s", SMA_VID(pSma), type,
             qtaskFileVer, tstrerror(code));
  } else {
    smaInfo("vgId:%d, restore rsma task %" PRIi8 " from qtaskf %" PRIi64 " succeed, nTables:%" PRIi64, SMA_VID(pSma),
            type, qtaskFileVer, nTables);
  }

  return code;
1053
}
K
kailixu 已提交
1054
#if 0
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
int32_t tdRSmaPersistExecImpl(SRSmaStat *pRSmaStat, SHashObj *pInfoHash) {
  int32_t code = 0;
  int32_t lino = 0;
  SSma   *pSma = pRSmaStat->pSma;
  SVnode *pVnode = pSma->pVnode;
  SRSmaFS fs = {0};

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

  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) {
          code = TSDB_CODE_RSMA_STREAM_STATE_COMMIT;
          TSDB_CHECK_CODE(code, lino, _exit);
        }
        smaDebug("vgId:%d, rsma persist, stream state commit success, table %" PRIi64 ", level %d", TD_VID(pVnode),
                 pRSmaInfo->suid, i + 1);
      }
    }
  }

_exit:
  if (code) {
    smaError("vgId:%d, %s failed at line %d since %s", TD_VID(pVnode), __func__, lino, tstrerror(code));
  }

  terrno = code;
  return code;
}
K
kailixu 已提交
1095
#endif
C
Cary Xu 已提交
1096
/**
C
Cary Xu 已提交
1097
 * @brief trigger to get rsma result in async mode
C
Cary Xu 已提交
1098 1099 1100 1101
 *
 * @param param
 * @param tmrId
 */
C
Cary Xu 已提交
1102
static void tdRSmaFetchTrigger(void *param, void *tmrId) {
1103
  SRSmaRef      *pRSmaRef = NULL;
C
Cary Xu 已提交
1104
  SSma          *pSma = NULL;
1105 1106 1107
  SRSmaStat     *pStat = NULL;
  SRSmaInfo     *pRSmaInfo = NULL;
  SRSmaInfoItem *pItem = NULL;
C
Cary Xu 已提交
1108

1109
  if (!(pRSmaRef = taosHashGet(smaMgmt.refHash, &param, POINTER_BYTES))) {
C
Cary Xu 已提交
1110 1111
    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 已提交
1112 1113 1114
    return;
  }

1115
  if (!(pStat = (SRSmaStat *)tdAcquireSmaRef(smaMgmt.rsetId, pRSmaRef->refId))) {
H
Hongze Cheng 已提交
1116 1117
    smaWarn("rsma fetch task not start since rsma stat already destroyed, rsetId:%d refId:%" PRIi64 ")", smaMgmt.rsetId,
            pRSmaRef->refId);  // pRSmaRef freed in taosHashRemove
C
Cary Xu 已提交
1118
    taosHashRemove(smaMgmt.refHash, &param, POINTER_BYTES);
C
Cary Xu 已提交
1119
    return;
C
Cary Xu 已提交
1120 1121
  }

C
Cary Xu 已提交
1122 1123
  pSma = pStat->pSma;

1124
  if (!(pRSmaInfo = tdAcquireRSmaInfoBySuid(pSma, pRSmaRef->suid))) {
C
Cary Xu 已提交
1125
    smaDebug("rsma fetch task not start since rsma info not exist, rsetId:%d refId:%" PRIi64 ")", smaMgmt.rsetId,
1126 1127 1128 1129 1130 1131 1132
             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 已提交
1133
    smaDebug("rsma fetch task not start since rsma info already deleted, rsetId:%d refId:%" PRIi64 ")", smaMgmt.rsetId,
1134 1135 1136 1137 1138 1139 1140 1141 1142
             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 已提交
1143 1144 1145 1146 1147
  // 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 已提交
1148
      smaDebug("vgId:%d, rsma fetch task not start for level %" PRIi8 " since stat is %" PRIi8
C
Cary Xu 已提交
1149
               ", rsetId:%d refId:%" PRIi64,
1150
               SMA_VID(pSma), pItem->level, rsmaTriggerStat, smaMgmt.rsetId, pRSmaRef->refId);
C
Cary Xu 已提交
1151
      if (rsmaTriggerStat == TASK_TRIGGER_STAT_PAUSED) {
1152
        taosTmrReset(tdRSmaFetchTrigger, RSMA_FETCH_INTERVAL, pItem, smaMgmt.tmrHandle, &pItem->tmrId);
C
Cary Xu 已提交
1153
      }
1154 1155
      tdReleaseRSmaInfo(pSma, pRSmaInfo);
      tdReleaseSmaRef(smaMgmt.rsetId, pRSmaRef->refId);
C
Cary Xu 已提交
1156
      return;
C
Cary Xu 已提交
1157
    }
C
Cary Xu 已提交
1158 1159 1160 1161 1162 1163 1164 1165
    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: {
1166
      smaDebug("vgId:%d, rsma fetch task planned for level:%" PRIi8 " suid:%" PRIi64 " since stat is active",
C
Cary Xu 已提交
1167
               SMA_VID(pSma), pItem->level, pRSmaInfo->suid);
C
Cary Xu 已提交
1168
      // async process
C
Cary Xu 已提交
1169
      pItem->fetchLevel = pItem->level;
C
Cary Xu 已提交
1170
#if 0
K
kailixu 已提交
1171
      // debugging codes
C
Cary Xu 已提交
1172 1173
      SRSmaInfo     *qInfo = tdAcquireRSmaInfoBySuid(pSma, pRSmaInfo->suid);
      SRSmaInfoItem *qItem = RSMA_INFO_ITEM(qInfo, pItem->level - 1);
K
kailixu 已提交
1174 1175
      make sure(qItem->level == pItem->level);
      make sure(qItem->fetchLevel == pItem->fetchLevel);
C
Cary Xu 已提交
1176
#endif
1177 1178 1179
      if (atomic_load_8(&pRSmaInfo->assigned) == 0) {
        tsem_post(&(pStat->notEmpty));
      }
C
Cary Xu 已提交
1180 1181
    } break;
    case TASK_TRIGGER_STAT_INACTIVE: {
K
kailixu 已提交
1182
      smaDebug("vgId:%d, rsma fetch task not start for level:%" PRIi8 " suid:%" PRIi64 " since stat is inactive ",
C
Cary Xu 已提交
1183 1184 1185
               SMA_VID(pSma), pItem->level, pRSmaInfo->suid);
    } break;
    case TASK_TRIGGER_STAT_INIT: {
C
Cary Xu 已提交
1186 1187
      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 已提交
1188 1189
    } break;
    default: {
K
kailixu 已提交
1190 1191 1192
      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 已提交
1193
    } break;
C
Cary Xu 已提交
1194 1195 1196
  }

_end:
C
Cary Xu 已提交
1197
  taosTmrReset(tdRSmaFetchTrigger, pItem->maxDelay, pItem, smaMgmt.tmrHandle, &pItem->tmrId);
1198 1199
  tdReleaseRSmaInfo(pSma, pRSmaInfo);
  tdReleaseSmaRef(smaMgmt.rsetId, pRSmaRef->refId);
C
Cary Xu 已提交
1200
}
C
Cary Xu 已提交
1201

C
Cary Xu 已提交
1202 1203
static void tdFreeRSmaSubmitItems(SArray *pItems) {
  for (int32_t i = 0; i < taosArrayGetSize(pItems); ++i) {
K
kailixu 已提交
1204
    SPackedData *packData = taosArrayGet(pItems, i);
K
kailixu 已提交
1205
    taosFreeQitem(POINTER_SHIFT(packData->msgStr, -sizeof(int32_t) - sizeof(int64_t)));
C
Cary Xu 已提交
1206
  }
C
Cary Xu 已提交
1207
  taosArrayClear(pItems);
C
Cary Xu 已提交
1208 1209
}

C
Cary Xu 已提交
1210 1211 1212 1213 1214 1215 1216
/**
 * @brief fetch rsma result(consider the efficiency and functionality)
 *
 * @param pSma
 * @param pInfo
 * @return int32_t
 */
1217
static int32_t tdRSmaFetchAllResult(SSma *pSma, SRSmaInfo *pInfo) {
C
Cary Xu 已提交
1218 1219
  SSDataBlock dataBlock = {.info.type = STREAM_GET_ALL};
  for (int8_t i = 1; i <= TSDB_RETENTION_L2; ++i) {
C
Cary Xu 已提交
1220 1221 1222
    SRSmaInfoItem *pItem = RSMA_INFO_ITEM(pInfo, i - 1);
    if (pItem->fetchLevel) {
      pItem->fetchLevel = 0;
C
Cary Xu 已提交
1223 1224 1225 1226
      qTaskInfo_t taskInfo = RSMA_INFO_QTASK(pInfo, i - 1);
      if (!taskInfo) {
        continue;
      }
C
Cary Xu 已提交
1227

C
Cary Xu 已提交
1228
      if ((++pItem->nScanned * pItem->maxDelay) > RSMA_FETCH_DELAY_MAX) {
1229
        smaDebug("vgId:%d, suid:%" PRIi64 " level:%" PRIi8 " nScanned:%" PRIi16 " maxDelay:%d, fetch executed",
C
Cary Xu 已提交
1230
                 SMA_VID(pSma), pInfo->suid, i, pItem->nScanned, pItem->maxDelay);
C
Cary Xu 已提交
1231
      } else {
C
Cary Xu 已提交
1232 1233 1234 1235 1236 1237 1238 1239 1240 1241
        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 已提交
1242 1243
      }

C
Cary Xu 已提交
1244
      pItem->nScanned = 0;
C
Cary Xu 已提交
1245

L
Liu Jicong 已提交
1246
      if ((terrno = qSetSMAInput(taskInfo, &dataBlock, 1, STREAM_INPUT__DATA_BLOCK)) < 0) {
C
Cary Xu 已提交
1247 1248
        goto _err;
      }
K
kailixu 已提交
1249
      if (tdRSmaExecAndSubmitResult(pSma, taskInfo, pItem, pInfo->pTSchema, pInfo->suid) < 0) {
C
Cary Xu 已提交
1250 1251 1252
        goto _err;
      }

1253
      smaDebug("vgId:%d, suid:%" PRIi64 " level:%" PRIi8 " nScanned:%" PRIi16 " maxDelay:%d, fetch finished",
C
Cary Xu 已提交
1254
               SMA_VID(pSma), pInfo->suid, i, pItem->nScanned, pItem->maxDelay);
C
Cary Xu 已提交
1255
    } else {
1256
      smaDebug("vgId:%d, suid:%" PRIi64 " level:%" PRIi8 " nScanned:%" PRIi16
C
Cary Xu 已提交
1257
               " maxDelay:%d, fetch not executed as fetch level is %" PRIi8,
C
Cary Xu 已提交
1258
               SMA_VID(pSma), pInfo->suid, i, pItem->nScanned, pItem->maxDelay, pItem->fetchLevel);
C
Cary Xu 已提交
1259 1260 1261 1262 1263 1264 1265 1266 1267
    }
  }

_end:
  return TSDB_CODE_SUCCESS;
_err:
  return TSDB_CODE_FAILED;
}

C
Cary Xu 已提交
1268 1269 1270 1271 1272 1273
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 已提交
1274 1275 1276 1277
      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 已提交
1278
      if (!taosArrayPush(pSubmitArr, &packData)) {
C
Cary Xu 已提交
1279 1280 1281 1282 1283 1284 1285 1286 1287 1288 1289 1290 1291 1292 1293 1294 1295 1296 1297
        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 已提交
1298 1299 1300
  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 已提交
1301 1302 1303 1304 1305 1306 1307 1308 1309 1310 1311 1312
  while (1) {
    void *msg = NULL;
    taosGetQitem(qall, (void **)&msg);
    if (msg) {
      taosFreeQitem(msg);
    } else {
      break;
    }
  }
  return TSDB_CODE_FAILED;
}

C
Cary Xu 已提交
1313 1314 1315 1316
/**
 * @brief
 *
 * @param pSma
C
Cary Xu 已提交
1317
 * @param type
C
Cary Xu 已提交
1318 1319
 * @return int32_t
 */
C
Cary Xu 已提交
1320

C
Cary Xu 已提交
1321
int32_t tdRSmaProcessExecImpl(SSma *pSma, ERsmaExecType type) {
K
kailixu 已提交
1322 1323
  int32_t    code = 0;
  int32_t    lino = 0;
C
Cary Xu 已提交
1324
  SVnode    *pVnode = pSma->pVnode;
C
Cary Xu 已提交
1325 1326 1327 1328
  SSmaEnv   *pEnv = SMA_RSMA_ENV(pSma);
  SRSmaStat *pRSmaStat = (SRSmaStat *)SMA_ENV_STAT(pEnv);
  SHashObj  *infoHash = NULL;
  SArray    *pSubmitArr = NULL;
C
Cary Xu 已提交
1329
  bool       isFetchAll = false;
C
Cary Xu 已提交
1330 1331

  if (!pRSmaStat || !(infoHash = RSMA_INFO_HASH(pRSmaStat))) {
K
kailixu 已提交
1332 1333
    code = TSDB_CODE_RSMA_INVALID_STAT;
    TSDB_CHECK_CODE(code, lino, _exit);
C
Cary Xu 已提交
1334 1335
  }

C
Cary Xu 已提交
1336
  if (!(pSubmitArr =
K
kailixu 已提交
1337
            taosArrayInit(TMIN(RSMA_SUBMIT_BATCH_SIZE, atomic_load_64(&pRSmaStat->nBufItems)), sizeof(SPackedData)))) {
K
kailixu 已提交
1338 1339
    code = TSDB_CODE_OUT_OF_MEMORY;
    TSDB_CHECK_CODE(code, lino, _exit);
C
Cary Xu 已提交
1340 1341
  }

C
Cary Xu 已提交
1342 1343
  while (true) {
    // step 1: rsma exec - consume data in buffer queue for all suids
C
Cary Xu 已提交
1344
    if (type == RSMA_EXEC_OVERFLOW) {
1345 1346
      void *pIter = NULL;
      while ((pIter = taosHashIterate(infoHash, pIter))) {
C
Cary Xu 已提交
1347
        SRSmaInfo *pInfo = *(SRSmaInfo **)pIter;
1348
        if (atomic_val_compare_exchange_8(&pInfo->assigned, 0, 1) == 0) {
K
kailixu 已提交
1349
          if ((taosQueueItemSize(pInfo->queue) > 0) || RSMA_NEED_FETCH(pInfo)) {
1350 1351 1352 1353 1354
            int32_t batchCnt = -1;
            int32_t batchMax = taosHashGetSize(infoHash) / tsNumOfVnodeRsmaThreads;
            bool    occupied = (batchMax <= 1);
            if (batchMax > 1) {
              batchMax = 100 / batchMax;
C
Cary Xu 已提交
1355
              batchMax = TMAX(batchMax, 4);
C
Cary Xu 已提交
1356
            }
C
Cary Xu 已提交
1357
            while (occupied || (++batchCnt < batchMax)) {    // greedy mode
1358 1359 1360 1361
              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);
1362
                smaDebug("vgId:%d, batchSize:%d, execType:%" PRIi32, SMA_VID(pSma), qallItemSize, type);
1363 1364
              }

K
kailixu 已提交
1365
              if (RSMA_NEED_FETCH(pInfo)) {
1366 1367 1368
                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 已提交
1369
                  int32_t oldVal = atomic_fetch_add_32(&pRSmaStat->nFetchAll, 1);
K
kailixu 已提交
1370 1371

                  if (ASSERTS(oldVal >= 0, "oldVal of nFetchAll: %d < 0", oldVal)) {
K
kailixu 已提交
1372 1373 1374
                    code = TSDB_CODE_APP_ERROR;
                    TSDB_CHECK_CODE(code, lino, _exit);
                  }
K
kailixu 已提交
1375 1376 1377 1378 1379 1380 1381 1382

                  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);
                  }

1383 1384 1385
                  if (0 == atomic_sub_fetch_32(&pRSmaStat->nFetchAll, 1)) {
                    atomic_store_8(RSMA_COMMIT_STAT(pRSmaStat), 0);
                  }
C
Cary Xu 已提交
1386
                }
1387 1388 1389 1390 1391
              }

              if (qallItemSize > 0) {
                atomic_fetch_sub_64(&pRSmaStat->nBufItems, qallItemSize);
                continue;
K
kailixu 已提交
1392 1393 1394
              }
              if (RSMA_NEED_FETCH(pInfo)) {
                continue;
1395 1396 1397
              }

              break;
C
Cary Xu 已提交
1398 1399
            }
          }
1400
          atomic_val_compare_exchange_8(&pInfo->assigned, 1, 0);
C
Cary Xu 已提交
1401
        }
C
Cary Xu 已提交
1402
      }
C
Cary Xu 已提交
1403
    } else {
K
kailixu 已提交
1404
      ASSERTS(0, "unknown rsma exec type:%d", (int32_t)type);
K
kailixu 已提交
1405 1406
      code = TSDB_CODE_APP_ERROR;
      TSDB_CHECK_CODE(code, lino, _exit);
C
Cary Xu 已提交
1407 1408
    }

C
Cary Xu 已提交
1409
    if (atomic_load_64(&pRSmaStat->nBufItems) <= 0) {
C
Cary Xu 已提交
1410
      if (pEnv->flag & SMA_ENV_FLG_CLOSE) {
C
Cary Xu 已提交
1411 1412
        break;
      }
1413

1414
      tsem_wait(&pRSmaStat->notEmpty);
C
Cary Xu 已提交
1415

1416
      if ((pEnv->flag & SMA_ENV_FLG_CLOSE) && (atomic_load_64(&pRSmaStat->nBufItems) <= 0)) {
1417
        smaDebug("vgId:%d, exec task end, flag:%" PRIi8 ", nBufItems:%" PRIi64, SMA_VID(pSma), pEnv->flag,
1418
                 atomic_load_64(&pRSmaStat->nBufItems));
1419 1420
        break;
      }
C
Cary Xu 已提交
1421
    }
1422

C
Cary Xu 已提交
1423
  }  // end of while(true)
C
Cary Xu 已提交
1424

K
kailixu 已提交
1425
_exit:
C
Cary Xu 已提交
1426
  taosArrayDestroy(pSubmitArr);
K
kailixu 已提交
1427
  if (code) {
K
kailixu 已提交
1428
    smaError("vgId:%d, %s failed at line %d since %s", TD_VID(pVnode), __func__, lino, tstrerror(code));
K
kailixu 已提交
1429 1430
  }
  return code;
C
Cary Xu 已提交
1431
}