tsdbSma.c 70.7 KB
Newer Older
C
Cary Xu 已提交
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15
/*
 * 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/>.
 */

H
Hongze Cheng 已提交
16 17
#include "tsdbSma.h"
#include "tsdb.h"
C
Cary Xu 已提交
18

C
Cary Xu 已提交
19 20 21 22 23
static const char *TSDB_SMA_DNAME[] = {
    "",      // TSDB_SMA_TYPE_BLOCK
    "tsma",  // TSDB_SMA_TYPE_TIME_RANGE
    "rsma",  // TSDB_SMA_TYPE_ROLLUP
};
C
Cary Xu 已提交
24

C
Cary Xu 已提交
25
#undef _TEST_SMA_PRINT_DEBUG_LOG_
C
Cary Xu 已提交
26
#define SMA_STORAGE_TSDB_DAYS   30
C
Cary Xu 已提交
27
#define SMA_STORAGE_TSDB_TIMES  10
C
Cary Xu 已提交
28
#define SMA_STORAGE_SPLIT_HOURS 24
C
Cary Xu 已提交
29
#define SMA_KEY_LEN             16  // TSKEY+groupId 8+8
C
Cary Xu 已提交
30
#define SMA_DROP_EXPIRED_TIME   10  // default is 10 seconds
C
Cary Xu 已提交
31

32 33 34 35
#define SMA_STATE_HASH_SLOT      4
#define SMA_STATE_ITEM_HASH_SLOT 32

#define SMA_TEST_INDEX_NAME "smaTestIndexName"  // TODO: just for test
36
#define SMA_TEST_INDEX_UID  2000000001          // TODO: just for test
C
Cary Xu 已提交
37 38

typedef struct SRSmaInfo SRSmaInfo;
C
Cary Xu 已提交
39
typedef enum {
C
Cary Xu 已提交
40 41
  SMA_STORAGE_LEVEL_TSDB = 0,     // use days of self-defined  e.g. vnode${N}/tsdb/tsma/sma_index_uid/v2f200.tsma
  SMA_STORAGE_LEVEL_DFILESET = 1  // use days of TS data       e.g. vnode${N}/tsdb/tsma/sma_index_uid/v2f1906.tsma
C
Cary Xu 已提交
42 43
} ESmaStorageLevel;

C
Cary Xu 已提交
44 45 46 47 48 49 50 51
typedef struct SPoolMem {
  int64_t          size;
  struct SPoolMem *prev;
  struct SPoolMem *next;
} SPoolMem;

struct SSmaEnv {
  TdThreadRwlock lock;
C
Cary Xu 已提交
52
  int8_t         type;
C
Cary Xu 已提交
53 54 55 56 57 58 59 60 61
  TXN            txn;
  SPoolMem      *pPool;
  SDiskID        did;
  TENV          *dbEnv;  // TODO: If it's better to put it in smaIndex level?
  char          *path;   // relative path
  SSmaStat      *pStat;
};

#define SMA_ENV_LOCK(env)       ((env)->lock)
C
Cary Xu 已提交
62
#define SMA_ENV_TYPE(env)       ((env)->type)
C
Cary Xu 已提交
63 64 65 66 67 68
#define SMA_ENV_DID(env)        ((env)->did)
#define SMA_ENV_ENV(env)        ((env)->dbEnv)
#define SMA_ENV_PATH(env)       ((env)->path)
#define SMA_ENV_STAT(env)       ((env)->pStat)
#define SMA_ENV_STAT_ITEMS(env) ((env)->pStat->smaStatItems)

C
Cary Xu 已提交
69
typedef struct {
C
Cary Xu 已提交
70 71 72 73
  STsdb        *pTsdb;
  SDBFile       dFile;
  const SArray *pDataBlocks;  // sma data
  int32_t       interval;     // interval with the precision of DB
C
Cary Xu 已提交
74 75 76 77
} STSmaWriteH;

typedef struct {
  int32_t iter;
C
Cary Xu 已提交
78
  int32_t fid;
C
Cary Xu 已提交
79
} SmaFsIter;
C
Cary Xu 已提交
80

C
Cary Xu 已提交
81
typedef struct {
C
Cary Xu 已提交
82
  STsdb    *pTsdb;
C
Cary Xu 已提交
83
  SDBFile   dFile;
C
Cary Xu 已提交
84 85 86 87 88 89 90
  int32_t   interval;   // interval with the precision of DB
  int32_t   blockSize;  // size of SMA block item
  int8_t    storageLevel;
  int8_t    days;
  SmaFsIter smaFsIter;
} STSmaReadH;

91 92 93 94 95
typedef struct {
  /**
   * @brief The field 'state' is here to demonstrate if one smaIndex is ready to provide service.
   *    - TSDB_SMA_STAT_OK: 1) The sma calculation of history data is finished; 2) Or recevied information from
   * Streaming Module or TSDB local persistence.
C
Cary Xu 已提交
96 97 98
   *    - TSDB_SMA_STAT_EXPIRED: 1) If sma calculation of history TS data is not finished; 2) Or if the TSDB is open,
   * without information about its previous state.
   *    - TSDB_SMA_STAT_DROPPED: 1)sma dropped
C
Cary Xu 已提交
99
   * N.B. only applicable to tsma
100 101 102
   */
  int8_t    state;           // ETsdbSmaStat
  SHashObj *expiredWindows;  // key: skey of time window, value: N/A
C
Cary Xu 已提交
103
  STSma    *pSma;            // cache schema
104 105
} SSmaStatItem;

C
Cary Xu 已提交
106 107
#define RSMA_TASK_INFO_HASH_SLOT 8
struct SRSmaInfo {
C
Cary Xu 已提交
108
  void *taskInfo[TSDB_RETENTION_L2];  // qTaskInfo_t
C
Cary Xu 已提交
109 110
};

111
struct SSmaStat {
C
Cary Xu 已提交
112 113 114 115
  union {
    SHashObj *smaStatItems;  // key: indexUid, value: SSmaStatItem for tsma
    SHashObj *rsmaInfoHash;  // key: stbUid, value: SRSmaInfo;
  };
116
  T_REF_DECLARE()
117
};
C
Cary Xu 已提交
118 119
#define SMA_STAT_ITEMS(s)     ((s)->smaStatItems)
#define SMA_STAT_INFO_HASH(s) ((s)->rsmaInfoHash)
C
Cary Xu 已提交
120 121 122 123 124 125 126 127 128 129

static FORCE_INLINE void tsdbFreeTaskHandle(qTaskInfo_t *taskHandle) {
  // Note: free/kill may in RC
  qTaskInfo_t otaskHandle = atomic_load_ptr(taskHandle);
  if (otaskHandle && atomic_val_compare_exchange_ptr(taskHandle, otaskHandle, NULL)) {
    qDestroyTask(otaskHandle);
  }
}

static FORCE_INLINE void *tsdbFreeRSmaInfo(SRSmaInfo *pInfo) {
C
Cary Xu 已提交
130
  for (int32_t i = 0; i < TSDB_RETENTION_MAX; ++i) {
C
Cary Xu 已提交
131 132 133 134 135 136
    if (pInfo->taskInfo[i]) {
      tsdbFreeTaskHandle(pInfo->taskInfo[i]);
    }
  }
  return NULL;
}
137

C
Cary Xu 已提交
138
// declaration of static functions
C
Cary Xu 已提交
139

140
// expired window
141
static int32_t  tsdbUpdateExpiredWindowImpl(STsdb *pTsdb, SSubmitReq *pMsg, int64_t version);
H
refact  
Hongze Cheng 已提交
142 143
static int32_t  tsdbSetExpiredWindow(STsdb *pTsdb, SHashObj *pItemsHash, int64_t indexUid, int64_t winSKey,
                                     int64_t version);
C
Cary Xu 已提交
144
static int32_t  tsdbInitSmaStat(SSmaStat **pSmaStat, int8_t smaType);
C
Cary Xu 已提交
145
static void    *tsdbFreeSmaStatItem(SSmaStatItem *pSmaStatItem);
C
Cary Xu 已提交
146
static int32_t  tsdbDestroySmaState(SSmaStat *pSmaStat, int8_t smaType);
C
Cary Xu 已提交
147 148
static SSmaEnv *tsdbNewSmaEnv(const STsdb *pTsdb, int8_t smaType, const char *path, SDiskID did);
static int32_t  tsdbInitSmaEnv(STsdb *pTsdb, int8_t smaType, const char *path, SDiskID did, SSmaEnv **pEnv);
149 150 151 152 153 154
static int32_t  tsdbResetExpiredWindow(STsdb *pTsdb, SSmaStat *pStat, int64_t indexUid, TSKEY skey);
static int32_t  tsdbRefSmaStat(STsdb *pTsdb, SSmaStat *pStat);
static int32_t  tsdbUnRefSmaStat(STsdb *pTsdb, SSmaStat *pStat);

// read data
// TODO: This is the basic params, and should wrap the params to a queryHandle.
C
Cary Xu 已提交
155
static int32_t tsdbGetTSmaDataImpl(STsdb *pTsdb, char *pData, int64_t indexUid, TSKEY querySKey, int32_t nMaxResult);
C
Cary Xu 已提交
156

157
// insert data
C
Cary Xu 已提交
158
static int32_t tsdbInitTSmaWriteH(STSmaWriteH *pSmaH, STsdb *pTsdb, const SArray *pDataBlocks, int64_t interval,
C
Cary Xu 已提交
159
                                  int8_t intervalUnit);
160 161 162
static void    tsdbDestroyTSmaWriteH(STSmaWriteH *pSmaH);
static int32_t tsdbInitTSmaReadH(STSmaReadH *pSmaH, STsdb *pTsdb, int64_t interval, int8_t intervalUnit);
static int32_t tsdbGetSmaStorageLevel(int64_t interval, int8_t intervalUnit);
C
Cary Xu 已提交
163
static int32_t tsdbSetRSmaDataFile(STSmaWriteH *pSmaH, int32_t fid);
C
Cary Xu 已提交
164 165
static int32_t tsdbInsertTSmaBlocks(STSmaWriteH *pSmaH, void *smaKey, int32_t keyLen, void *pData, int32_t dataLen,
                                    TXN *txn);
C
Cary Xu 已提交
166
static int64_t tsdbGetIntervalByPrecision(int64_t interval, uint8_t intervalUnit, int8_t precision, bool adjusted);
C
Cary Xu 已提交
167
static int32_t tsdbGetTSmaDays(STsdb *pTsdb, int64_t interval, int32_t storageLevel);
C
Cary Xu 已提交
168
static int32_t tsdbSetTSmaDataFile(STSmaWriteH *pSmaH, int64_t indexUid, int32_t fid);
169
static int32_t tsdbInitTSmaFile(STSmaReadH *pSmaH, int64_t indexUid, TSKEY skey);
C
Cary Xu 已提交
170
static bool    tsdbSetAndOpenTSmaFile(STSmaReadH *pReadH, TSKEY *queryKey);
C
Cary Xu 已提交
171
static void    tsdbGetSmaDir(int32_t vgId, ETsdbSmaType smaType, char dirName[]);
C
Cary Xu 已提交
172 173
static int32_t tsdbInsertTSmaDataImpl(STsdb *pTsdb, int64_t indexUid, const char *msg);
static int32_t tsdbInsertRSmaDataImpl(STsdb *pTsdb, const char *msg);
C
Cary Xu 已提交
174

C
Cary Xu 已提交
175
static FORCE_INLINE int32_t tsdbUidStorePut(STbUidStore *pStore, tb_uid_t suid, tb_uid_t *uid);
C
Cary Xu 已提交
176
static FORCE_INLINE int32_t tsdbUpdateTbUidListImpl(STsdb *pTsdb, tb_uid_t *suid, SArray *tbUids);
C
Cary Xu 已提交
177
static FORCE_INLINE int32_t tsdbExecuteRSmaImpl(STsdb *pTsdb, const void *pMsg, int32_t inputType,
C
Cary Xu 已提交
178 179
                                                qTaskInfo_t *taskInfo, STSchema *pTSchema, tb_uid_t suid, tb_uid_t uid,
                                                int8_t level);
C
Cary Xu 已提交
180 181 182
// mgmt interface
static int32_t tsdbDropTSmaDataImpl(STsdb *pTsdb, int64_t indexUid);

C
Cary Xu 已提交
183 184 185 186 187 188 189 190 191 192
// Pool Memory
static SPoolMem *openPool();
static void      clearPool(SPoolMem *pPool);
static void      closePool(SPoolMem *pPool);
static void     *poolMalloc(void *arg, size_t size);
static void      poolFree(void *arg, void *ptr);

static int tsdbSmaBeginCommit(SSmaEnv *pEnv);
static int tsdbSmaEndCommit(SSmaEnv *pEnv);

193
// implementation
C
Cary Xu 已提交
194 195 196 197 198 199 200 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 227 228
static FORCE_INLINE int16_t tsdbTSmaAdd(STsdb *pTsdb, int16_t n) {
  return atomic_add_fetch_16(&REPO_TSMA_NUM(pTsdb), n);
}
static FORCE_INLINE int16_t tsdbTSmaSub(STsdb *pTsdb, int16_t n) {
  return atomic_sub_fetch_16(&REPO_TSMA_NUM(pTsdb), n);
}

static FORCE_INLINE int32_t tsdbRLockSma(SSmaEnv *pEnv) {
  int code = taosThreadRwlockRdlock(&(pEnv->lock));
  if (code != 0) {
    terrno = TAOS_SYSTEM_ERROR(code);
    return -1;
  }
  return 0;
}

static FORCE_INLINE int32_t tsdbWLockSma(SSmaEnv *pEnv) {
  int code = taosThreadRwlockWrlock(&(pEnv->lock));
  if (code != 0) {
    terrno = TAOS_SYSTEM_ERROR(code);
    return -1;
  }
  return 0;
}

static FORCE_INLINE int32_t tsdbUnLockSma(SSmaEnv *pEnv) {
  int code = taosThreadRwlockUnlock(&(pEnv->lock));
  if (code != 0) {
    terrno = TAOS_SYSTEM_ERROR(code);
    return -1;
  }
  return 0;
}

static SPoolMem *openPool() {
H
Hongze Cheng 已提交
229
  SPoolMem *pPool = (SPoolMem *)taosMemoryMalloc(sizeof(*pPool));
C
Cary Xu 已提交
230 231 232 233 234 235 236 237 238

  pPool->prev = pPool->next = pPool;
  pPool->size = 0;

  return pPool;
}

static void clearPool(SPoolMem *pPool) {
  if (!pPool) return;
H
refact  
Hongze Cheng 已提交
239

C
Cary Xu 已提交
240 241 242 243 244 245 246 247 248 249 250
  SPoolMem *pMem;

  do {
    pMem = pPool->next;

    if (pMem == pPool) break;

    pMem->next->prev = pMem->prev;
    pMem->prev->next = pMem->next;
    pPool->size -= pMem->size;

H
Hongze Cheng 已提交
251
    taosMemoryFree(pMem);
C
Cary Xu 已提交
252 253 254 255 256 257 258 259
  } while (1);

  assert(pPool->size == 0);
}

static void closePool(SPoolMem *pPool) {
  if (pPool) {
    clearPool(pPool);
H
Hongze Cheng 已提交
260
    taosMemoryFree(pPool);
C
Cary Xu 已提交
261 262 263 264 265 266 267 268
  }
}

static void *poolMalloc(void *arg, size_t size) {
  void     *ptr = NULL;
  SPoolMem *pPool = (SPoolMem *)arg;
  SPoolMem *pMem;

H
Hongze Cheng 已提交
269
  pMem = (SPoolMem *)taosMemoryMalloc(sizeof(*pMem) + size);
C
Cary Xu 已提交
270
  if (!pMem) {
C
Cary Xu 已提交
271 272 273 274 275 276 277 278 279 280 281 282 283 284 285 286 287 288 289 290 291 292 293 294 295
    assert(0);
  }

  pMem->size = sizeof(*pMem) + size;
  pMem->next = pPool->next;
  pMem->prev = pPool;

  pPool->next->prev = pMem;
  pPool->next = pMem;
  pPool->size += pMem->size;

  ptr = (void *)(&pMem[1]);
  return ptr;
}

static void poolFree(void *arg, void *ptr) {
  SPoolMem *pPool = (SPoolMem *)arg;
  SPoolMem *pMem;

  pMem = &(((SPoolMem *)ptr)[-1]);

  pMem->next->prev = pMem->prev;
  pMem->prev->next = pMem->next;
  pPool->size -= pMem->size;

H
Hongze Cheng 已提交
296
  taosMemoryFree(pMem);
C
Cary Xu 已提交
297
}
C
Cary Xu 已提交
298 299 300

int32_t tsdbInitSma(STsdb *pTsdb) {
  // tSma
H
Hongze Cheng 已提交
301
  int32_t numOfTSma = taosArrayGetSize(metaGetSmaTbUids(REPO_META(pTsdb), false));
C
Cary Xu 已提交
302 303 304 305 306 307 308
  if (numOfTSma > 0) {
    atomic_store_16(&REPO_TSMA_NUM(pTsdb), (int16_t)numOfTSma);
  }
  // TODO: rSma
  return TSDB_CODE_SUCCESS;
}

C
Cary Xu 已提交
309 310 311 312 313 314 315 316
static FORCE_INLINE int8_t tsdbSmaStat(SSmaStatItem *pStatItem) {
  if (pStatItem) {
    return atomic_load_8(&pStatItem->state);
  }
  return TSDB_SMA_STAT_UNKNOWN;
}

static FORCE_INLINE bool tsdbSmaStatIsOK(SSmaStatItem *pStatItem, int8_t *state) {
C
Cary Xu 已提交
317
  if (!pStatItem) {
C
Cary Xu 已提交
318 319 320 321 322 323 324 325 326 327 328 329 330 331 332 333 334 335 336 337 338 339 340 341 342 343 344 345 346 347 348 349 350 351 352 353
    return false;
  }

  if (state) {
    *state = atomic_load_8(&pStatItem->state);
    return *state == TSDB_SMA_STAT_OK;
  }
  return atomic_load_8(&pStatItem->state) == TSDB_SMA_STAT_OK;
}

static FORCE_INLINE bool tsdbSmaStatIsExpired(SSmaStatItem *pStatItem) {
  return pStatItem ? (atomic_load_8(&pStatItem->state) & TSDB_SMA_STAT_EXPIRED) : true;
}

static FORCE_INLINE bool tsdbSmaStatIsDropped(SSmaStatItem *pStatItem) {
  return pStatItem ? (atomic_load_8(&pStatItem->state) & TSDB_SMA_STAT_DROPPED) : true;
}

static FORCE_INLINE void tsdbSmaStatSetOK(SSmaStatItem *pStatItem) {
  if (pStatItem) {
    atomic_store_8(&pStatItem->state, TSDB_SMA_STAT_OK);
  }
}

static FORCE_INLINE void tsdbSmaStatSetExpired(SSmaStatItem *pStatItem) {
  if (pStatItem) {
    atomic_or_fetch_8(&pStatItem->state, TSDB_SMA_STAT_EXPIRED);
  }
}

static FORCE_INLINE void tsdbSmaStatSetDropped(SSmaStatItem *pStatItem) {
  if (pStatItem) {
    atomic_or_fetch_8(&pStatItem->state, TSDB_SMA_STAT_DROPPED);
  }
}

C
Cary Xu 已提交
354
static void tsdbGetSmaDir(int32_t vgId, ETsdbSmaType smaType, char dirName[]) {
C
Cary Xu 已提交
355
  snprintf(dirName, TSDB_FILENAME_LEN, "vnode%svnode%d%s%s", TD_DIRSEP, vgId, TD_DIRSEP, TSDB_SMA_DNAME[smaType]);
C
Cary Xu 已提交
356
}
C
Cary Xu 已提交
357

C
Cary Xu 已提交
358
static SSmaEnv *tsdbNewSmaEnv(const STsdb *pTsdb, int8_t smaType, const char *path, SDiskID did) {
C
Cary Xu 已提交
359 360
  SSmaEnv *pEnv = NULL;

wafwerar's avatar
wafwerar 已提交
361
  pEnv = (SSmaEnv *)taosMemoryCalloc(1, sizeof(SSmaEnv));
C
Cary Xu 已提交
362
  if (!pEnv) {
C
Cary Xu 已提交
363 364 365 366
    terrno = TSDB_CODE_OUT_OF_MEMORY;
    return NULL;
  }

C
Cary Xu 已提交
367 368
  SMA_ENV_TYPE(pEnv) = smaType;

wafwerar's avatar
wafwerar 已提交
369
  int code = taosThreadRwlockInit(&(pEnv->lock), NULL);
C
Cary Xu 已提交
370 371
  if (code) {
    terrno = TAOS_SYSTEM_ERROR(code);
wafwerar's avatar
wafwerar 已提交
372
    taosMemoryFree(pEnv);
C
Cary Xu 已提交
373 374 375 376
    return NULL;
  }

  ASSERT(path && (strlen(path) > 0));
C
Cary Xu 已提交
377
  SMA_ENV_PATH(pEnv) = strdup(path);
C
Cary Xu 已提交
378
  if (!SMA_ENV_PATH(pEnv)) {
C
Cary Xu 已提交
379 380 381 382
    tsdbFreeSmaEnv(pEnv);
    return NULL;
  }

C
Cary Xu 已提交
383
  SMA_ENV_DID(pEnv) = did;
384

C
Cary Xu 已提交
385
  if (tsdbInitSmaStat(&SMA_ENV_STAT(pEnv), smaType) != TSDB_CODE_SUCCESS) {
C
Cary Xu 已提交
386 387 388 389
    tsdbFreeSmaEnv(pEnv);
    return NULL;
  }

390
  char aname[TSDB_FILENAME_LEN] = {0};
H
Hongze Cheng 已提交
391
  tfsAbsoluteName(REPO_TFS(pTsdb), did, path, aname);
C
Cary Xu 已提交
392 393 394 395 396
  if (tsdbOpenDBEnv(&pEnv->dbEnv, aname) != TSDB_CODE_SUCCESS) {
    tsdbFreeSmaEnv(pEnv);
    return NULL;
  }

C
Cary Xu 已提交
397
  if (!(pEnv->pPool = openPool())) {
398 399 400 401
    tsdbFreeSmaEnv(pEnv);
    return NULL;
  }

C
Cary Xu 已提交
402 403 404
  return pEnv;
}

C
Cary Xu 已提交
405
static int32_t tsdbInitSmaEnv(STsdb *pTsdb, int8_t smaType, const char *path, SDiskID did, SSmaEnv **pEnv) {
C
Cary Xu 已提交
406 407 408 409 410
  if (!pEnv) {
    terrno = TSDB_CODE_INVALID_PTR;
    return TSDB_CODE_FAILED;
  }

C
Cary Xu 已提交
411 412
  if (!(*pEnv)) {
    if (!(*pEnv = tsdbNewSmaEnv(pTsdb, smaType, path, did))) {
C
Cary Xu 已提交
413 414 415 416 417 418 419 420 421 422 423 424 425 426 427
      return TSDB_CODE_FAILED;
    }
  }

  return TSDB_CODE_SUCCESS;
}

/**
 * @brief Release resources allocated for its member fields, not including itself.
 *
 * @param pSmaEnv
 * @return int32_t
 */
void tsdbDestroySmaEnv(SSmaEnv *pSmaEnv) {
  if (pSmaEnv) {
C
Cary Xu 已提交
428
    tsdbDestroySmaState(pSmaEnv->pStat, SMA_ENV_TYPE(pSmaEnv));
wafwerar's avatar
wafwerar 已提交
429 430
    taosMemoryFreeClear(pSmaEnv->pStat);
    taosMemoryFreeClear(pSmaEnv->path);
wafwerar's avatar
wafwerar 已提交
431
    taosThreadRwlockDestroy(&(pSmaEnv->lock));
C
Cary Xu 已提交
432 433
    tsdbCloseDBEnv(pSmaEnv->dbEnv);
    closePool(pSmaEnv->pPool);
C
Cary Xu 已提交
434 435 436 437 438
  }
}

void *tsdbFreeSmaEnv(SSmaEnv *pSmaEnv) {
  tsdbDestroySmaEnv(pSmaEnv);
wafwerar's avatar
wafwerar 已提交
439
  taosMemoryFreeClear(pSmaEnv);
C
Cary Xu 已提交
440 441 442
  return NULL;
}

443
static int32_t tsdbRefSmaStat(STsdb *pTsdb, SSmaStat *pStat) {
C
Cary Xu 已提交
444
  if (!pStat) return 0;
C
Cary Xu 已提交
445

446
  int ref = T_REF_INC(pStat);
C
Cary Xu 已提交
447
  tsdbDebug("vgId:%d ref sma stat:%p, val:%d", REPO_ID(pTsdb), pStat, ref);
448 449 450 451
  return 0;
}

static int32_t tsdbUnRefSmaStat(STsdb *pTsdb, SSmaStat *pStat) {
C
Cary Xu 已提交
452
  if (!pStat) return 0;
453 454

  int ref = T_REF_DEC(pStat);
C
Cary Xu 已提交
455
  tsdbDebug("vgId:%d unref sma stat:%p, val:%d", REPO_ID(pTsdb), pStat, ref);
456 457 458
  return 0;
}

C
Cary Xu 已提交
459
static int32_t tsdbInitSmaStat(SSmaStat **pSmaStat, int8_t smaType) {
460
  ASSERT(pSmaStat != NULL);
C
Cary Xu 已提交
461

C
Cary Xu 已提交
462
  if (*pSmaStat) {  // no lock
C
Cary Xu 已提交
463 464 465
    return TSDB_CODE_SUCCESS;
  }

466 467 468 469 470
  /**
   *  1. Lazy mode utilized when init SSmaStat to update expired window(or hungry mode when tsdbNew).
   *  2. Currently, there is mutex lock when init SSmaEnv, thus no need add lock on SSmaStat, and please add lock if
   * tsdbInitSmaStat invoked in other multithread environment later.
   */
C
Cary Xu 已提交
471
  if (!(*pSmaStat)) {
wafwerar's avatar
wafwerar 已提交
472
    *pSmaStat = (SSmaStat *)taosMemoryCalloc(1, sizeof(SSmaStat));
C
Cary Xu 已提交
473
    if (!(*pSmaStat)) {
474 475 476 477
      terrno = TSDB_CODE_OUT_OF_MEMORY;
      return TSDB_CODE_FAILED;
    }

C
Cary Xu 已提交
478
    if (smaType == TSDB_SMA_TYPE_ROLLUP) {
C
Cary Xu 已提交
479
      SMA_STAT_INFO_HASH(*pSmaStat) = taosHashInit(
C
Cary Xu 已提交
480
          RSMA_TASK_INFO_HASH_SLOT, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BIGINT), true, HASH_ENTRY_LOCK);
481

C
Cary Xu 已提交
482
      if (!SMA_STAT_INFO_HASH(*pSmaStat)) {
C
Cary Xu 已提交
483 484 485 486
        taosMemoryFreeClear(*pSmaStat);
        return TSDB_CODE_FAILED;
      }
    } else if (smaType == TSDB_SMA_TYPE_TIME_RANGE) {
C
Cary Xu 已提交
487
      SMA_STAT_ITEMS(*pSmaStat) =
C
Cary Xu 已提交
488 489
          taosHashInit(SMA_STATE_HASH_SLOT, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY), true, HASH_ENTRY_LOCK);

C
Cary Xu 已提交
490
      if (!SMA_STAT_ITEMS(*pSmaStat)) {
C
Cary Xu 已提交
491 492 493 494 495
        taosMemoryFreeClear(*pSmaStat);
        return TSDB_CODE_FAILED;
      }
    } else {
      ASSERT(0);
496 497 498 499 500 501 502 503
    }
  }
  return TSDB_CODE_SUCCESS;
}

static SSmaStatItem *tsdbNewSmaStatItem(int8_t state) {
  SSmaStatItem *pItem = NULL;

wafwerar's avatar
wafwerar 已提交
504
  pItem = (SSmaStatItem *)taosMemoryCalloc(1, sizeof(SSmaStatItem));
505 506 507 508 509
  if (pItem) {
    pItem->state = state;
    pItem->expiredWindows = taosHashInit(SMA_STATE_ITEM_HASH_SLOT, taosGetDefaultHashFunction(TSDB_DATA_TYPE_TIMESTAMP),
                                         true, HASH_ENTRY_LOCK);
    if (!pItem->expiredWindows) {
wafwerar's avatar
wafwerar 已提交
510
      taosMemoryFreeClear(pItem);
511 512 513 514 515
    }
  }
  return pItem;
}

C
Cary Xu 已提交
516
static void *tsdbFreeSmaStatItem(SSmaStatItem *pSmaStatItem) {
C
Cary Xu 已提交
517
  if (pSmaStatItem) {
C
Cary Xu 已提交
518
    tdDestroyTSma(pSmaStatItem->pSma);
wafwerar's avatar
wafwerar 已提交
519
    taosMemoryFreeClear(pSmaStatItem->pSma);
C
Cary Xu 已提交
520
    taosHashCleanup(pSmaStatItem->expiredWindows);
wafwerar's avatar
wafwerar 已提交
521
    taosMemoryFreeClear(pSmaStatItem);
C
Cary Xu 已提交
522 523 524 525
  }
  return NULL;
}

C
Cary Xu 已提交
526 527
/**
 * @brief Release resources allocated for its member fields, not including itself.
528 529 530
 *
 * @param pSmaStat
 * @return int32_t
C
Cary Xu 已提交
531
 */
C
Cary Xu 已提交
532
int32_t tsdbDestroySmaState(SSmaStat *pSmaStat, int8_t smaType) {
533 534
  if (pSmaStat) {
    // TODO: use taosHashSetFreeFp when taosHashSetFreeFp is ready.
C
Cary Xu 已提交
535
    if (smaType == TSDB_SMA_TYPE_TIME_RANGE) {
C
Cary Xu 已提交
536
      void *item = taosHashIterate(SMA_STAT_ITEMS(pSmaStat), NULL);
C
Cary Xu 已提交
537
      while (item) {
C
Cary Xu 已提交
538 539
        SSmaStatItem *pItem = *(SSmaStatItem **)item;
        tsdbFreeSmaStatItem(pItem);
C
Cary Xu 已提交
540
        item = taosHashIterate(SMA_STAT_ITEMS(pSmaStat), item);
C
Cary Xu 已提交
541
      }
C
Cary Xu 已提交
542
      taosHashCleanup(SMA_STAT_ITEMS(pSmaStat));
C
Cary Xu 已提交
543
    } else if (smaType == TSDB_SMA_TYPE_ROLLUP) {
C
Cary Xu 已提交
544
      void *infoHash = taosHashIterate(SMA_STAT_INFO_HASH(pSmaStat), NULL);
C
Cary Xu 已提交
545
      while (infoHash) {
C
Cary Xu 已提交
546 547
        SRSmaInfo *pInfoHash = *(SRSmaInfo **)infoHash;
        tsdbFreeRSmaInfo(pInfoHash);
C
Cary Xu 已提交
548
        infoHash = taosHashIterate(SMA_STAT_INFO_HASH(pSmaStat), infoHash);
C
Cary Xu 已提交
549
      }
C
Cary Xu 已提交
550
      taosHashCleanup(SMA_STAT_INFO_HASH(pSmaStat));
C
Cary Xu 已提交
551 552
    } else {
      ASSERT(0);
553 554
    }
  }
555
  return TSDB_CODE_SUCCESS;
556 557
}

C
Cary Xu 已提交
558
static int32_t tsdbCheckAndInitSmaEnv(STsdb *pTsdb, int8_t smaType) {
559 560
  SSmaEnv *pEnv = NULL;

C
Cary Xu 已提交
561
  // return if already init
C
Cary Xu 已提交
562 563
  switch (smaType) {
    case TSDB_SMA_TYPE_TIME_RANGE:
C
Cary Xu 已提交
564
      if ((pEnv = (SSmaEnv *)atomic_load_ptr(&REPO_TSMA_ENV(pTsdb)))) {
C
Cary Xu 已提交
565 566 567 568
        return TSDB_CODE_SUCCESS;
      }
      break;
    case TSDB_SMA_TYPE_ROLLUP:
C
Cary Xu 已提交
569
      if ((pEnv = (SSmaEnv *)atomic_load_ptr(&REPO_RSMA_ENV(pTsdb)))) {
C
Cary Xu 已提交
570 571 572 573 574 575 576 577
        return TSDB_CODE_SUCCESS;
      }
      break;
    default:
      terrno = TSDB_CODE_INVALID_PARA;
      return TSDB_CODE_FAILED;
  }

C
Cary Xu 已提交
578 579
  // init sma env
  tsdbLockRepo(pTsdb);
C
Cary Xu 已提交
580 581
  pEnv = (smaType == TSDB_SMA_TYPE_TIME_RANGE) ? atomic_load_ptr(&REPO_TSMA_ENV(pTsdb))
                                               : atomic_load_ptr(&REPO_RSMA_ENV(pTsdb));
C
Cary Xu 已提交
582
  if (!pEnv) {
C
Cary Xu 已提交
583
    char rname[TSDB_FILENAME_LEN] = {0};
C
Cary Xu 已提交
584

C
Cary Xu 已提交
585
    SDiskID did = {0};
H
Hongze Cheng 已提交
586
    tfsAllocDisk(REPO_TFS(pTsdb), TFS_PRIMARY_LEVEL, &did);
C
Cary Xu 已提交
587 588 589 590 591 592
    if (did.level < 0 || did.id < 0) {
      tsdbUnlockRepo(pTsdb);
      return TSDB_CODE_FAILED;
    }
    tsdbGetSmaDir(REPO_ID(pTsdb), smaType, rname);

H
Hongze Cheng 已提交
593
    if (tfsMkdirRecurAt(REPO_TFS(pTsdb), rname, did) != TSDB_CODE_SUCCESS) {
C
Cary Xu 已提交
594 595 596 597
      tsdbUnlockRepo(pTsdb);
      return TSDB_CODE_FAILED;
    }

C
Cary Xu 已提交
598
    if (tsdbInitSmaEnv(pTsdb, smaType, rname, did, &pEnv) != TSDB_CODE_SUCCESS) {
C
Cary Xu 已提交
599 600 601 602
      tsdbUnlockRepo(pTsdb);
      return TSDB_CODE_FAILED;
    }

C
Cary Xu 已提交
603 604
    (smaType == TSDB_SMA_TYPE_TIME_RANGE) ? atomic_store_ptr(&REPO_TSMA_ENV(pTsdb), pEnv)
                                          : atomic_store_ptr(&REPO_RSMA_ENV(pTsdb), pEnv);
C
Cary Xu 已提交
605
  }
C
Cary Xu 已提交
606
  tsdbUnlockRepo(pTsdb);
C
Cary Xu 已提交
607 608 609 610

  return TSDB_CODE_SUCCESS;
};

H
refact  
Hongze Cheng 已提交
611 612
static int32_t tsdbSetExpiredWindow(STsdb *pTsdb, SHashObj *pItemsHash, int64_t indexUid, int64_t winSKey,
                                    int64_t version) {
613
  SSmaStatItem *pItem = taosHashGet(pItemsHash, &indexUid, sizeof(indexUid));
C
Cary Xu 已提交
614
  if (!pItem) {
C
Cary Xu 已提交
615 616
    // TODO: use TSDB_SMA_STAT_EXPIRED and update by stream computing later
    pItem = tsdbNewSmaStatItem(TSDB_SMA_STAT_OK);  // TODO use the real state
C
Cary Xu 已提交
617
    if (!pItem) {
618
      // Response to stream computing: OOM
619
      // For query, if the indexUid not found, the TSDB should tell query module to query raw TS data.
620 621 622
      return TSDB_CODE_FAILED;
    }

C
Cary Xu 已提交
623
    // cache smaMeta
H
Hongze Cheng 已提交
624
    STSma *pSma = metaGetSmaInfoByIndex(REPO_META(pTsdb), indexUid, true);
C
Cary Xu 已提交
625
    if (!pSma) {
626
      terrno = TSDB_CODE_TDB_NO_SMA_INDEX_IN_META;
C
Cary Xu 已提交
627
      taosHashCleanup(pItem->expiredWindows);
wafwerar's avatar
wafwerar 已提交
628
      taosMemoryFree(pItem);
629 630
      tsdbWarn("vgId:%d update expired window failed for smaIndex %" PRIi64 " since %s", REPO_ID(pTsdb), indexUid,
               tstrerror(terrno));
C
Cary Xu 已提交
631 632 633 634
      return TSDB_CODE_FAILED;
    }
    pItem->pSma = pSma;

C
Cary Xu 已提交
635
    if (taosHashPut(pItemsHash, &indexUid, sizeof(indexUid), &pItem, sizeof(pItem)) != 0) {
636 637
      // If error occurs during put smaStatItem, free the resources of pItem
      taosHashCleanup(pItem->expiredWindows);
wafwerar's avatar
wafwerar 已提交
638
      taosMemoryFree(pItem);
639 640
      return TSDB_CODE_FAILED;
    }
C
Cary Xu 已提交
641
  } else if (!(pItem = *(SSmaStatItem **)pItem)) {
C
Cary Xu 已提交
642 643
    terrno = TSDB_CODE_INVALID_PTR;
    return TSDB_CODE_FAILED;
644 645
  }

646
  if (taosHashPut(pItem->expiredWindows, &winSKey, sizeof(TSKEY), &version, sizeof(version)) != 0) {
C
Cary Xu 已提交
647 648 649 650 651 652 653
    // If error occurs during taosHashPut expired windows, remove the smaIndex from pTsdb->pSmaStat, thus TSDB would
    // tell query module to query raw TS data.
    // N.B.
    //  1) It is assumed to be extemely little probability event of fail to taosHashPut.
    //  2) This would solve the inconsistency to some extent, but not completely, unless we record all expired
    // windows failed to put into hash table.
    taosHashCleanup(pItem->expiredWindows);
wafwerar's avatar
wafwerar 已提交
654
    taosMemoryFreeClear(pItem->pSma);
C
Cary Xu 已提交
655
    taosHashRemove(pItemsHash, &indexUid, sizeof(indexUid));
C
Cary Xu 已提交
656 657
    tsdbWarn("vgId:%d smaIndex %" PRIi64 ", put skey %" PRIi64 " to expire window fail", REPO_ID(pTsdb), indexUid,
             winSKey);
C
Cary Xu 已提交
658 659
    return TSDB_CODE_FAILED;
  }
wmmhello's avatar
wmmhello 已提交
660

C
Cary Xu 已提交
661 662
  tsdbDebug("vgId:%d smaIndex %" PRIi64 ", put skey %" PRIi64 " to expire window succeed", REPO_ID(pTsdb), indexUid,
            winSKey);
663
  return TSDB_CODE_SUCCESS;
C
Cary Xu 已提交
664 665 666 667 668 669 670 671 672
}

/**
 * @brief Update expired window according to msg from stream computing module.
 *
 * @param pTsdb
 * @param msg SSubmitReq
 * @return int32_t
 */
673 674
int32_t tsdbUpdateExpiredWindowImpl(STsdb *pTsdb, SSubmitReq *pMsg, int64_t version) {
  // no time-range-sma, just return success
C
Cary Xu 已提交
675 676 677 678 679
  if (atomic_load_16(&REPO_TSMA_NUM(pTsdb)) <= 0) {
    tsdbTrace("vgId:%d not update expire window since no tSma", REPO_ID(pTsdb));
    return TSDB_CODE_SUCCESS;
  }

H
Hongze Cheng 已提交
680
  if (!REPO_META(pTsdb)) {
C
Cary Xu 已提交
681 682 683 684 685 686 687 688 689 690 691 692
    terrno = TSDB_CODE_INVALID_PTR;
    return TSDB_CODE_FAILED;
  }

  if (tsdbCheckAndInitSmaEnv(pTsdb, TSDB_SMA_TYPE_TIME_RANGE) != TSDB_CODE_SUCCESS) {
    terrno = TSDB_CODE_TDB_INIT_FAILED;
    return TSDB_CODE_FAILED;
  }

  // Firstly, assume that tSma can only be created on super table/normal table.
  // getActiveTimeWindow

C
Cary Xu 已提交
693
  SSmaEnv  *pEnv = REPO_TSMA_ENV(pTsdb);
C
Cary Xu 已提交
694 695 696
  SSmaStat *pStat = SMA_ENV_STAT(pEnv);
  SHashObj *pItemsHash = SMA_ENV_STAT_ITEMS(pEnv);

C
Cary Xu 已提交
697
  TASSERT(pEnv && pStat && pItemsHash);
C
Cary Xu 已提交
698

C
Cary Xu 已提交
699 700 701
  // basic procedure
  // TODO: optimization
  tsdbRefSmaStat(pTsdb, pStat);
C
Cary Xu 已提交
702 703 704 705

  SSubmitMsgIter msgIter = {0};
  SSubmitBlk    *pBlock = NULL;
  SInterval      interval = {0};
C
Cary Xu 已提交
706
  TSKEY          lastWinSKey = INT64_MIN;
C
Cary Xu 已提交
707

C
Cary Xu 已提交
708
  if (tInitSubmitMsgIter(pMsg, &msgIter) != TSDB_CODE_SUCCESS) {
C
Cary Xu 已提交
709 710 711 712
    return TSDB_CODE_FAILED;
  }

  while (true) {
C
Cary Xu 已提交
713
    tGetSubmitMsgNext(&msgIter, &pBlock);
C
Cary Xu 已提交
714
    if (!pBlock) break;
C
Cary Xu 已提交
715 716 717 718

    STSmaWrapper *pSW = NULL;
    STSma        *pTSma = NULL;

C
Cary Xu 已提交
719
    SSubmitBlkIter blkIter = {0};
C
Cary Xu 已提交
720
    if (tInitSubmitBlkIter(&msgIter, pBlock, &blkIter) != TSDB_CODE_SUCCESS) {
C
Cary Xu 已提交
721
      pSW = tdFreeTSmaWrapper(pSW);
C
Cary Xu 已提交
722 723 724
      break;
    }

C
Cary Xu 已提交
725
    while (true) {
C
Cary Xu 已提交
726
      STSRow *row = tGetSubmitBlkNext(&blkIter);
C
Cary Xu 已提交
727
      if (!row) {
C
Cary Xu 已提交
728 729 730
        tdFreeTSmaWrapper(pSW);
        break;
      }
C
Cary Xu 已提交
731 732 733 734
      if (!pSW || (pTSma->tableUid != pBlock->suid)) {
        if (pSW) {
          pSW = tdFreeTSmaWrapper(pSW);
        }
C
Cary Xu 已提交
735
        if (!(pSW = metaGetSmaInfoByTable(REPO_META(pTsdb), pBlock->suid))) {
C
Cary Xu 已提交
736 737
          break;
        }
C
Cary Xu 已提交
738
        if ((pSW->number) <= 0 || !pSW->tSma) {
C
Cary Xu 已提交
739
          pSW = tdFreeTSmaWrapper(pSW);
C
Cary Xu 已提交
740 741
          break;
        }
C
Cary Xu 已提交
742

C
Cary Xu 已提交
743 744
        pTSma = pSW->tSma;

C
Cary Xu 已提交
745 746 747 748 749 750 751
        interval.interval = pTSma->interval;
        interval.intervalUnit = pTSma->intervalUnit;
        interval.offset = pTSma->offset;
        interval.precision = REPO_CFG(pTsdb)->precision;
        interval.sliding = pTSma->sliding;
        interval.slidingUnit = pTSma->slidingUnit;
      }
C
Cary Xu 已提交
752

C
update  
Cary Xu 已提交
753 754
      TSKEY winSKey = taosTimeTruncate(TD_ROW_KEY(row), &interval, interval.precision);

C
Cary Xu 已提交
755 756 757 758 759 760 761
      if (lastWinSKey != winSKey) {
        lastWinSKey = winSKey;
        tsdbSetExpiredWindow(pTsdb, pItemsHash, pTSma->indexUid, winSKey, version);
      } else {
        tsdbDebug("vgId:%d smaIndex %" PRIi64 ", put skey %" PRIi64 " to expire window ignore as duplicated",
                  REPO_ID(pTsdb), pTSma->indexUid, winSKey);
      }
762 763 764
    }
  }

765
  tsdbUnRefSmaStat(pTsdb, pStat);
C
Cary Xu 已提交
766

767 768 769
  return TSDB_CODE_SUCCESS;
}

C
Cary Xu 已提交
770 771 772 773 774 775 776 777 778
/**
 * @brief When sma data received from stream computing, make the relative expired window valid.
 *
 * @param pTsdb
 * @param pStat
 * @param indexUid
 * @param skey
 * @return int32_t
 */
779
static int32_t tsdbResetExpiredWindow(STsdb *pTsdb, SSmaStat *pStat, int64_t indexUid, TSKEY skey) {
C
Cary Xu 已提交
780 781
  SSmaStatItem *pItem = NULL;

782 783
  tsdbRefSmaStat(pTsdb, pStat);

C
Cary Xu 已提交
784 785
  if (pStat && SMA_STAT_ITEMS(pStat)) {
    pItem = taosHashGet(SMA_STAT_ITEMS(pStat), &indexUid, sizeof(indexUid));
C
Cary Xu 已提交
786
  }
C
Cary Xu 已提交
787
  if ((pItem) && ((pItem = *(SSmaStatItem **)pItem))) {
788 789
    // pItem resides in hash buffer all the time unless drop sma index
    // TODO: multithread protect
C
Cary Xu 已提交
790 791
    if (taosHashRemove(pItem->expiredWindows, &skey, sizeof(TSKEY)) != 0) {
      // error handling
792
      tsdbUnRefSmaStat(pTsdb, pStat);
C
Cary Xu 已提交
793
      tsdbWarn("vgId:%d remove skey %" PRIi64 " from expired window for sma index %" PRIi64 " fail", REPO_ID(pTsdb),
794 795
               skey, indexUid);
      return TSDB_CODE_FAILED;
C
Cary Xu 已提交
796
    }
C
Cary Xu 已提交
797 798
    tsdbDebug("vgId:%d remove skey %" PRIi64 " from expired window for sma index %" PRIi64 " succeed", REPO_ID(pTsdb),
              skey, indexUid);
C
Cary Xu 已提交
799 800 801 802 803 804 805 806 807
    // TODO: use a standalone interface to received state upate notification from stream computing module.
    /**
     * @brief state
     *  - When SMA env init in TSDB, its status is TSDB_SMA_STAT_OK.
     *  - In startup phase of stream computing module, it should notify the SMA env in TSDB to expired if needed(e.g.
     * when batch data caculation not finised)
     *  - When TSDB_SMA_STAT_OK, the stream computing module should also notify that to the SMA env in TSDB.
     */
    pItem->state = TSDB_SMA_STAT_OK;
C
Cary Xu 已提交
808 809
  } else {
    // error handling
810 811 812
    tsdbUnRefSmaStat(pTsdb, pStat);
    tsdbWarn("vgId:%d expired window %" PRIi64 " not exists for sma index %" PRIi64, REPO_ID(pTsdb), skey, indexUid);
    return TSDB_CODE_FAILED;
C
Cary Xu 已提交
813
  }
814 815

  tsdbUnRefSmaStat(pTsdb, pStat);
C
Cary Xu 已提交
816 817 818
  return TSDB_CODE_SUCCESS;
}

C
Cary Xu 已提交
819 820 821 822 823 824 825
/**
 * @brief Judge the tSma storage level
 *
 * @param interval
 * @param intervalUnit
 * @return int32_t
 */
C
Cary Xu 已提交
826
static int32_t tsdbGetSmaStorageLevel(int64_t interval, int8_t intervalUnit) {
C
Cary Xu 已提交
827 828
  // TODO: configurable for SMA_STORAGE_SPLIT_HOURS?
  switch (intervalUnit) {
C
Cary Xu 已提交
829
    case TIME_UNIT_HOUR:
C
Cary Xu 已提交
830 831 832 833
      if (interval < SMA_STORAGE_SPLIT_HOURS) {
        return SMA_STORAGE_LEVEL_DFILESET;
      }
      break;
C
Cary Xu 已提交
834
    case TIME_UNIT_MINUTE:
C
Cary Xu 已提交
835 836 837 838
      if (interval < 60 * SMA_STORAGE_SPLIT_HOURS) {
        return SMA_STORAGE_LEVEL_DFILESET;
      }
      break;
C
Cary Xu 已提交
839
    case TIME_UNIT_SECOND:
C
Cary Xu 已提交
840 841 842 843
      if (interval < 3600 * SMA_STORAGE_SPLIT_HOURS) {
        return SMA_STORAGE_LEVEL_DFILESET;
      }
      break;
C
Cary Xu 已提交
844
    case TIME_UNIT_MILLISECOND:
C
Cary Xu 已提交
845 846 847 848
      if (interval < 3600 * 1e3 * SMA_STORAGE_SPLIT_HOURS) {
        return SMA_STORAGE_LEVEL_DFILESET;
      }
      break;
C
Cary Xu 已提交
849
    case TIME_UNIT_MICROSECOND:
C
Cary Xu 已提交
850 851 852 853
      if (interval < 3600 * 1e6 * SMA_STORAGE_SPLIT_HOURS) {
        return SMA_STORAGE_LEVEL_DFILESET;
      }
      break;
C
Cary Xu 已提交
854
    case TIME_UNIT_NANOSECOND:
C
Cary Xu 已提交
855 856 857 858 859 860 861 862 863 864 865
      if (interval < 3600 * 1e9 * SMA_STORAGE_SPLIT_HOURS) {
        return SMA_STORAGE_LEVEL_DFILESET;
      }
      break;
    default:
      break;
  }
  return SMA_STORAGE_LEVEL_TSDB;
}

/**
C
Cary Xu 已提交
866
 * @brief Insert TSma data blocks to DB File build by B+Tree
C
Cary Xu 已提交
867
 *
C
Cary Xu 已提交
868
 * @param pSmaH
869
 * @param smaKey  tableUid-colId-skeyOfWindow(8-2-8)
C
Cary Xu 已提交
870
 * @param keyLen
C
Cary Xu 已提交
871 872 873 874
 * @param pData
 * @param dataLen
 * @return int32_t
 */
C
Cary Xu 已提交
875 876
static int32_t tsdbInsertTSmaBlocks(STSmaWriteH *pSmaH, void *smaKey, int32_t keyLen, void *pData, int32_t dataLen,
                                    TXN *txn) {
C
Cary Xu 已提交
877
  SDBFile *pDBFile = &pSmaH->dFile;
C
Cary Xu 已提交
878

879
  // TODO: insert sma data blocks into B+Tree(TDB)
C
Cary Xu 已提交
880
  if (tsdbSaveSmaToDB(pDBFile, smaKey, keyLen, pData, dataLen, txn) != 0) {
C
Cary Xu 已提交
881 882
    tsdbWarn("vgId:%d insert sma data blocks into %s: smaKey %" PRIx64 "-%" PRIx64 ", dataLen %" PRIu32 " fail",
             REPO_ID(pSmaH->pTsdb), pDBFile->path, *(int64_t *)smaKey, *(int64_t *)POINTER_SHIFT(smaKey, 8), dataLen);
883 884
    return TSDB_CODE_FAILED;
  }
C
Cary Xu 已提交
885 886
  tsdbDebug("vgId:%d insert sma data blocks into %s: smaKey %" PRIx64 "-%" PRIx64 ", dataLen %" PRIu32 " succeed",
            REPO_ID(pSmaH->pTsdb), pDBFile->path, *(int64_t *)smaKey, *(int64_t *)POINTER_SHIFT(smaKey, 8), dataLen);
887

C
Cary Xu 已提交
888
#ifdef _TEST_SMA_PRINT_DEBUG_LOG_
C
Cary Xu 已提交
889
  uint32_t valueSize = 0;
C
Cary Xu 已提交
890
  void    *data = tsdbGetSmaDataByKey(pDBFile, smaKey, keyLen, &valueSize);
C
Cary Xu 已提交
891 892
  ASSERT(data != NULL);
  for (uint32_t v = 0; v < valueSize; v += 8) {
C
Cary Xu 已提交
893
    tsdbWarn("vgId:%d insert sma data val[%d] %" PRIi64, REPO_ID(pSmaH->pTsdb), v, *(int64_t *)POINTER_SHIFT(data, v));
C
Cary Xu 已提交
894 895
  }
#endif
C
Cary Xu 已提交
896 897 898
  return TSDB_CODE_SUCCESS;
}

C
Cary Xu 已提交
899 900 901 902 903 904
/**
 * @brief Approximate value for week/month/year.
 *
 * @param interval
 * @param intervalUnit
 * @param precision
C
Cary Xu 已提交
905
 * @param adjusted Interval already adjusted according to DB precision
C
Cary Xu 已提交
906 907
 * @return int64_t
 */
C
Cary Xu 已提交
908 909 910 911 912
static int64_t tsdbGetIntervalByPrecision(int64_t interval, uint8_t intervalUnit, int8_t precision, bool adjusted) {
  if (adjusted) {
    return interval;
  }

C
Cary Xu 已提交
913 914 915 916 917 918 919 920 921 922 923 924 925 926 927 928 929 930 931 932 933 934 935 936
  switch (intervalUnit) {
    case TIME_UNIT_YEAR:  // approximate value
      interval *= 365 * 86400 * 1e3;
      break;
    case TIME_UNIT_MONTH:  // approximate value
      interval *= 30 * 86400 * 1e3;
      break;
    case TIME_UNIT_WEEK:  // approximate value
      interval *= 7 * 86400 * 1e3;
      break;
    case TIME_UNIT_DAY:  // the interval for tSma calculation must <= day
      interval *= 86400 * 1e3;
      break;
    case TIME_UNIT_HOUR:
      interval *= 3600 * 1e3;
      break;
    case TIME_UNIT_MINUTE:
      interval *= 60 * 1e3;
      break;
    case TIME_UNIT_SECOND:
      interval *= 1e3;
      break;
    default:
      break;
C
Cary Xu 已提交
937 938
  }

C
Cary Xu 已提交
939 940
  switch (precision) {
    case TSDB_TIME_PRECISION_MILLI:
C
Cary Xu 已提交
941
      if (TIME_UNIT_MICROSECOND == intervalUnit) {  // us
C
Cary Xu 已提交
942
        return interval / 1e3;
C
Cary Xu 已提交
943
      } else if (TIME_UNIT_NANOSECOND == intervalUnit) {  //  nano second
C
Cary Xu 已提交
944
        return interval / 1e6;
945
      } else {  // ms
C
Cary Xu 已提交
946 947 948
        return interval;
      }
      break;
C
Cary Xu 已提交
949
    case TSDB_TIME_PRECISION_MICRO:
C
Cary Xu 已提交
950
      if (TIME_UNIT_MICROSECOND == intervalUnit) {  // us
C
Cary Xu 已提交
951
        return interval;
952
      } else if (TIME_UNIT_NANOSECOND == intervalUnit) {  //  ns
C
Cary Xu 已提交
953
        return interval / 1e3;
954
      } else {  // ms
C
Cary Xu 已提交
955 956 957
        return interval * 1e3;
      }
      break;
C
Cary Xu 已提交
958
    case TSDB_TIME_PRECISION_NANO:
959
      if (TIME_UNIT_MICROSECOND == intervalUnit) {  // us
C
Cary Xu 已提交
960
        return interval * 1e3;
961
      } else if (TIME_UNIT_NANOSECOND == intervalUnit) {  // ns
C
Cary Xu 已提交
962
        return interval;
963
      } else {  // ms
C
Cary Xu 已提交
964
        return interval * 1e6;
C
Cary Xu 已提交
965 966
      }
      break;
C
Cary Xu 已提交
967
    default:                                        // ms
C
Cary Xu 已提交
968
      if (TIME_UNIT_MICROSECOND == intervalUnit) {  // us
C
Cary Xu 已提交
969
        return interval / 1e3;
970
      } else if (TIME_UNIT_NANOSECOND == intervalUnit) {  //  ns
C
Cary Xu 已提交
971
        return interval / 1e6;
972
      } else {  // ms
C
Cary Xu 已提交
973
        return interval;
C
Cary Xu 已提交
974 975 976 977 978 979
      }
      break;
  }
  return interval;
}

C
Cary Xu 已提交
980 981
static int32_t tsdbInitTSmaWriteH(STSmaWriteH *pSmaH, STsdb *pTsdb, const SArray *pDataBlocks, int64_t interval,
                                  int8_t intervalUnit) {
C
Cary Xu 已提交
982
  pSmaH->pTsdb = pTsdb;
C
Cary Xu 已提交
983 984 985
  pSmaH->interval = tsdbGetIntervalByPrecision(interval, intervalUnit, REPO_CFG(pTsdb)->precision, true);
  pSmaH->pDataBlocks = pDataBlocks;
  pSmaH->dFile.fid = TSDB_IVLD_FID;
986 987 988 989 990
  return TSDB_CODE_SUCCESS;
}

static void tsdbDestroyTSmaWriteH(STSmaWriteH *pSmaH) {
  if (pSmaH) {
C
Cary Xu 已提交
991
    tsdbCloseDBF(&pSmaH->dFile);
992
  }
C
Cary Xu 已提交
993 994
}

C
Cary Xu 已提交
995
static int32_t tsdbSetTSmaDataFile(STSmaWriteH *pSmaH, int64_t indexUid, int32_t fid) {
C
Cary Xu 已提交
996
  STsdb *pTsdb = pSmaH->pTsdb;
C
Cary Xu 已提交
997
  ASSERT(!pSmaH->dFile.path && !pSmaH->dFile.pDB);
998 999

  pSmaH->dFile.fid = fid;
1000
  char tSmaFile[TSDB_FILENAME_LEN] = {0};
1001
  snprintf(tSmaFile, TSDB_FILENAME_LEN, "%" PRIi64 "%sv%df%d.tsma", indexUid, TD_DIRSEP, REPO_ID(pTsdb), fid);
C
Cary Xu 已提交
1002
  pSmaH->dFile.path = strdup(tSmaFile);
1003

C
Cary Xu 已提交
1004
  return TSDB_CODE_SUCCESS;
C
Cary Xu 已提交
1005
}
C
Cary Xu 已提交
1006

C
Cary Xu 已提交
1007 1008 1009 1010 1011 1012 1013 1014 1015
/**
 * @brief
 *
 * @param pTsdb
 * @param interval Interval calculated by DB's precision
 * @param storageLevel
 * @return int32_t
 */
static int32_t tsdbGetTSmaDays(STsdb *pTsdb, int64_t interval, int32_t storageLevel) {
C
Cary Xu 已提交
1016 1017
  STsdbKeepCfg *pCfg = REPO_KEEP_CFG(pTsdb);
  int32_t       daysPerFile = pCfg->days;
C
Cary Xu 已提交
1018 1019

  if (storageLevel == SMA_STORAGE_LEVEL_TSDB) {
1020
    int32_t days = SMA_STORAGE_TSDB_TIMES * (interval / tsTickPerMin[pCfg->precision]);
C
Cary Xu 已提交
1021 1022 1023 1024 1025
    daysPerFile = days > SMA_STORAGE_TSDB_DAYS ? days : SMA_STORAGE_TSDB_DAYS;
  }

  return daysPerFile;
}
C
Cary Xu 已提交
1026

C
Cary Xu 已提交
1027 1028 1029 1030 1031
static int tsdbSmaBeginCommit(SSmaEnv *pEnv) {
  TXN *pTxn = &pEnv->txn;
  // start a new txn
  tdbTxnOpen(pTxn, 0, poolMalloc, poolFree, pEnv->pPool, TDB_TXN_WRITE | TDB_TXN_READ_UNCOMMITTED);
  if (tdbBegin(pEnv->dbEnv, pTxn) != 0) {
1032
    tsdbWarn("tsdbSma tdb begin commit fail");
C
Cary Xu 已提交
1033 1034 1035 1036 1037 1038 1039 1040 1041 1042
    return -1;
  }
  return 0;
}

static int tsdbSmaEndCommit(SSmaEnv *pEnv) {
  TXN *pTxn = &pEnv->txn;

  // Commit current txn
  if (tdbCommit(pEnv->dbEnv, pTxn) != 0) {
1043
    tsdbWarn("tsdbSma tdb end commit fail");
C
Cary Xu 已提交
1044 1045 1046 1047 1048 1049 1050
    return -1;
  }
  tdbTxnClose(pTxn);
  clearPool(pEnv->pPool);
  return 0;
}

C
Cary Xu 已提交
1051 1052 1053 1054 1055 1056 1057 1058 1059
/**
 * @brief Insert/Update Time-range-wise SMA data.
 *  - If interval < SMA_STORAGE_SPLIT_HOURS(e.g. 24), save the SMA data as a part of DFileSet to e.g.
 * v3f1900.tsma.${sma_index_name}. The days is the same with that for TS data files.
 *  - If interval >= SMA_STORAGE_SPLIT_HOURS, save the SMA data to e.g. vnode3/tsma/v3f632.tsma.${sma_index_name}. The
 * days is 30 times of the interval, and the minimum days is SMA_STORAGE_TSDB_DAYS(30d).
 *  - The destination file of one data block for some interval is determined by its start TS key.
 *
 * @param pTsdb
C
Cary Xu 已提交
1060
 * @param msg
C
Cary Xu 已提交
1061 1062
 * @return int32_t
 */
C
Cary Xu 已提交
1063 1064 1065
static int32_t tsdbInsertTSmaDataImpl(STsdb *pTsdb, int64_t indexUid, const char *msg) {
  STsdbCfg     *pCfg = REPO_CFG(pTsdb);
  const SArray *pDataBlocks = (const SArray *)msg;
C
Cary Xu 已提交
1066

C
Cary Xu 已提交
1067 1068
  // TODO: destroy SSDataBlocks(msg)

1069 1070 1071 1072 1073
  // For super table aggregation, the sma data is stored in vgroup calculated from the hash value of stable name. Thus
  // the sma data would arrive ahead of the update-expired-window msg.
  if (tsdbCheckAndInitSmaEnv(pTsdb, TSDB_SMA_TYPE_TIME_RANGE) != TSDB_CODE_SUCCESS) {
    terrno = TSDB_CODE_TDB_INIT_FAILED;
    return TSDB_CODE_FAILED;
1074
  }
C
Cary Xu 已提交
1075

C
Cary Xu 已提交
1076
  if (!pDataBlocks) {
C
Cary Xu 已提交
1077
    terrno = TSDB_CODE_INVALID_PTR;
C
Cary Xu 已提交
1078
    tsdbWarn("vgId:%d insert tSma data failed since pDataBlocks is NULL", REPO_ID(pTsdb));
C
Cary Xu 已提交
1079
    return terrno;
C
Cary Xu 已提交
1080 1081
  }

C
Cary Xu 已提交
1082
  if (taosArrayGetSize(pDataBlocks) <= 0) {
C
Cary Xu 已提交
1083
    terrno = TSDB_CODE_INVALID_PARA;
C
Cary Xu 已提交
1084
    tsdbWarn("vgId:%d insert tSma data failed since pDataBlocks is empty", REPO_ID(pTsdb));
1085
    return TSDB_CODE_FAILED;
C
Cary Xu 已提交
1086
  }
C
Cary Xu 已提交
1087

1088
  SSmaEnv      *pEnv = REPO_TSMA_ENV(pTsdb);
C
Cary Xu 已提交
1089
  SSmaStat     *pStat = SMA_ENV_STAT(pEnv);
C
Cary Xu 已提交
1090 1091 1092 1093
  SSmaStatItem *pItem = NULL;

  tsdbRefSmaStat(pTsdb, pStat);

C
Cary Xu 已提交
1094 1095
  if (pStat && SMA_STAT_ITEMS(pStat)) {
    pItem = taosHashGet(SMA_STAT_ITEMS(pStat), &indexUid, sizeof(indexUid));
C
Cary Xu 已提交
1096
  }
C
Cary Xu 已提交
1097

C
Cary Xu 已提交
1098
  if (!pItem || !(pItem = *(SSmaStatItem **)pItem) || tsdbSmaStatIsDropped(pItem)) {
C
Cary Xu 已提交
1099 1100 1101 1102 1103
    terrno = TSDB_CODE_TDB_INVALID_SMA_STAT;
    tsdbUnRefSmaStat(pTsdb, pStat);
    return TSDB_CODE_FAILED;
  }

C
Cary Xu 已提交
1104
  STSma      *pSma = pItem->pSma;
C
Cary Xu 已提交
1105 1106
  STSmaWriteH tSmaH = {0};

C
Cary Xu 已提交
1107
  if (tsdbInitTSmaWriteH(&tSmaH, pTsdb, pDataBlocks, pSma->interval, pSma->intervalUnit) != 0) {
C
Cary Xu 已提交
1108 1109 1110
    return TSDB_CODE_FAILED;
  }

C
Cary Xu 已提交
1111 1112
  char rPath[TSDB_FILENAME_LEN] = {0};
  char aPath[TSDB_FILENAME_LEN] = {0};
1113 1114 1115 1116
  snprintf(rPath, TSDB_FILENAME_LEN, "%s%s%" PRIi64, SMA_ENV_PATH(pEnv), TD_DIRSEP, indexUid);
  tfsAbsoluteName(REPO_TFS(pTsdb), SMA_ENV_DID(pEnv), rPath, aPath);
  if (!taosCheckExistFile(aPath)) {
    if (tfsMkdirRecurAt(REPO_TFS(pTsdb), rPath, SMA_ENV_DID(pEnv)) != TSDB_CODE_SUCCESS) {
C
Cary Xu 已提交
1117
      tsdbUnRefSmaStat(pTsdb, pStat);
1118 1119 1120 1121
      return TSDB_CODE_FAILED;
    }
  }

C
Cary Xu 已提交
1122
  // Step 1: Judge the storage level and days
C
Cary Xu 已提交
1123
  int32_t storageLevel = tsdbGetSmaStorageLevel(pSma->interval, pSma->intervalUnit);
C
Cary Xu 已提交
1124
  int32_t daysPerFile = tsdbGetTSmaDays(pTsdb, tSmaH.interval, storageLevel);
C
Cary Xu 已提交
1125

C
Cary Xu 已提交
1126 1127
  char    smaKey[SMA_KEY_LEN] = {0};  // key: skey + groupId
  char    dataBuf[512] = {0};         // val: aggr data // TODO: handle 512 buffer?
C
Cary Xu 已提交
1128
  void   *pDataBuf = NULL;
C
Cary Xu 已提交
1129 1130
  int32_t sz = taosArrayGetSize(pDataBlocks);
  for (int32_t i = 0; i < sz; ++i) {
C
Cary Xu 已提交
1131
    SSDataBlock *pDataBlock = taosArrayGet(pDataBlocks, i);
C
Cary Xu 已提交
1132 1133 1134 1135 1136 1137
    int32_t      colNum = pDataBlock->info.numOfCols;
    int32_t      rows = pDataBlock->info.rows;
    int32_t      rowSize = pDataBlock->info.rowSize;
    int64_t      groupId = pDataBlock->info.groupId;
    for (int32_t j = 0; j < rows; ++j) {
      printf("|");
C
Cary Xu 已提交
1138
      TSKEY skey = TSKEY_INITIAL_VAL;  //  the start key of TS window by interval
C
Cary Xu 已提交
1139 1140
      void *pSmaKey = &smaKey;
      bool  isStartKey = false;
C
Cary Xu 已提交
1141

C
Cary Xu 已提交
1142 1143
      int32_t tlen = 0;     // reset the len
      pDataBuf = &dataBuf;  // reset the buf
C
Cary Xu 已提交
1144
      for (int32_t k = 0; k < colNum; ++k) {
C
Cary Xu 已提交
1145
        SColumnInfoData *pColInfoData = taosArrayGet(pDataBlock->pDataBlock, k);
C
Cary Xu 已提交
1146 1147 1148
        void            *var = POINTER_SHIFT(pColInfoData->pData, j * pColInfoData->info.bytes);
        switch (pColInfoData->info.type) {
          case TSDB_DATA_TYPE_TIMESTAMP:
C
Cary Xu 已提交
1149 1150 1151
            if (!isStartKey) {
              isStartKey = true;
              skey = *(TSKEY *)var;
C
Cary Xu 已提交
1152
              printf("= skey %" PRIi64 " groupId = %" PRIi64 "|", skey, groupId);
C
Cary Xu 已提交
1153 1154 1155 1156 1157 1158
              tsdbEncodeTSmaKey(groupId, skey, &pSmaKey);
            } else {
              printf(" %" PRIi64 " |", *(int64_t *)var);
              tlen += taosEncodeFixedI64(&pDataBuf, *(int64_t *)var);
              break;
            }
C
Cary Xu 已提交
1159 1160 1161 1162 1163 1164 1165 1166 1167 1168 1169 1170 1171 1172 1173 1174 1175 1176 1177 1178 1179 1180 1181
            break;
          case TSDB_DATA_TYPE_BOOL:
          case TSDB_DATA_TYPE_UTINYINT:
            printf(" %15d |", *(uint8_t *)var);
            tlen += taosEncodeFixedU8(&pDataBuf, *(uint8_t *)var);
            break;
          case TSDB_DATA_TYPE_TINYINT:
            printf(" %15d |", *(int8_t *)var);
            tlen += taosEncodeFixedI8(&pDataBuf, *(int8_t *)var);
            break;
          case TSDB_DATA_TYPE_SMALLINT:
            printf(" %15d |", *(int16_t *)var);
            tlen += taosEncodeFixedI16(&pDataBuf, *(int16_t *)var);
            break;
          case TSDB_DATA_TYPE_USMALLINT:
            printf(" %15d |", *(uint16_t *)var);
            tlen += taosEncodeFixedU16(&pDataBuf, *(uint16_t *)var);
            break;
          case TSDB_DATA_TYPE_INT:
            printf(" %15d |", *(int32_t *)var);
            tlen += taosEncodeFixedI32(&pDataBuf, *(int32_t *)var);
            break;
          case TSDB_DATA_TYPE_FLOAT:
C
Cary Xu 已提交
1182 1183 1184
            printf(" %15f |", *(float *)var);
            tlen += taosEncodeBinary(&pDataBuf, var, sizeof(float));
            break;
C
Cary Xu 已提交
1185 1186 1187 1188 1189 1190 1191 1192 1193
          case TSDB_DATA_TYPE_UINT:
            printf(" %15u |", *(uint32_t *)var);
            tlen += taosEncodeFixedU32(&pDataBuf, *(uint32_t *)var);
            break;
          case TSDB_DATA_TYPE_BIGINT:
            printf(" %15ld |", *(int64_t *)var);
            tlen += taosEncodeFixedI64(&pDataBuf, *(int64_t *)var);
            break;
          case TSDB_DATA_TYPE_DOUBLE:
C
Cary Xu 已提交
1194 1195
            printf(" %15lf |", *(double *)var);
            tlen += taosEncodeBinary(&pDataBuf, var, sizeof(double));
C
Cary Xu 已提交
1196 1197 1198 1199 1200 1201 1202 1203 1204 1205 1206
          case TSDB_DATA_TYPE_UBIGINT:
            printf(" %15lu |", *(uint64_t *)var);
            tlen += taosEncodeFixedU64(&pDataBuf, *(uint64_t *)var);
            break;
          case TSDB_DATA_TYPE_NCHAR: {
            char tmpChar[100] = {0};
            strncpy(tmpChar, varDataVal(var), varDataLen(var));
            printf(" %s |", tmpChar);
            tlen += taosEncodeBinary(&pDataBuf, varDataVal(var), varDataLen(var));
            break;
          }
C
Cary Xu 已提交
1207 1208
          case TSDB_DATA_TYPE_VARCHAR: {  // TSDB_DATA_TYPE_BINARY
            char tmpChar[100] = {0};
C
Cary Xu 已提交
1209 1210 1211 1212 1213 1214 1215 1216 1217 1218 1219 1220 1221 1222
            strncpy(tmpChar, varDataVal(var), varDataLen(var));
            printf(" %s |", tmpChar);
            tlen += taosEncodeBinary(&pDataBuf, varDataVal(var), varDataLen(var));
            break;
          }
          case TSDB_DATA_TYPE_VARBINARY:
            // TODO: add binary/varbinary
            TASSERT(0);
          default:
            printf("the column type %" PRIi16 " is undefined\n", pColInfoData->info.type);
            TASSERT(0);
            break;
        }
      }
C
Cary Xu 已提交
1223 1224
      // if ((tlen > 0) && (skey != TSKEY_INITIAL_VAL)) {
      if (tlen > 0) {
C
Cary Xu 已提交
1225 1226 1227 1228 1229 1230 1231 1232
        int32_t fid = (int32_t)(TSDB_KEY_FID(skey, daysPerFile, pCfg->precision));

        // Step 2: Set the DFile for storage of SMA index, and iterate/split the TSma data and store to B+Tree index
        // file
        //         - Set and open the DFile or the B+Tree file
        // TODO: tsdbStartTSmaCommit();
        if (fid != tSmaH.dFile.fid) {
          if (tSmaH.dFile.fid != TSDB_IVLD_FID) {
C
Cary Xu 已提交
1233
            tsdbSmaEndCommit(pEnv);
C
Cary Xu 已提交
1234 1235 1236
            tsdbCloseDBF(&tSmaH.dFile);
          }
          tsdbSetTSmaDataFile(&tSmaH, indexUid, fid);
C
Cary Xu 已提交
1237
          if (tsdbOpenDBF(pEnv->dbEnv, &tSmaH.dFile) != 0) {
C
Cary Xu 已提交
1238 1239 1240 1241 1242 1243
            tsdbWarn("vgId:%d open DB file %s failed since %s", REPO_ID(pTsdb),
                     tSmaH.dFile.path ? tSmaH.dFile.path : "path is NULL", tstrerror(terrno));
            tsdbDestroyTSmaWriteH(&tSmaH);
            tsdbUnRefSmaStat(pTsdb, pStat);
            return TSDB_CODE_FAILED;
          }
C
Cary Xu 已提交
1244
          tsdbSmaBeginCommit(pEnv);
C
Cary Xu 已提交
1245
        }
C
Cary Xu 已提交
1246

C
Cary Xu 已提交
1247
        if (tsdbInsertTSmaBlocks(&tSmaH, &smaKey, SMA_KEY_LEN, dataBuf, tlen, &pEnv->txn) != 0) {
C
Cary Xu 已提交
1248
          tsdbWarn("vgId:%d insert tSma data blocks fail for index %" PRIi64 ", skey %" PRIi64 ", groupId %" PRIi64
C
Cary Xu 已提交
1249 1250
                   " since %s",
                   REPO_ID(pTsdb), indexUid, skey, groupId, tstrerror(terrno));
C
Cary Xu 已提交
1251
          tsdbSmaEndCommit(pEnv);
C
Cary Xu 已提交
1252 1253 1254 1255
          tsdbDestroyTSmaWriteH(&tSmaH);
          tsdbUnRefSmaStat(pTsdb, pStat);
          return TSDB_CODE_FAILED;
        }
C
Cary Xu 已提交
1256 1257
        tsdbDebug("vgId:%d insert tSma data blocks success for index %" PRIi64 ", skey %" PRIi64 ", groupId %" PRIi64,
                  REPO_ID(pTsdb), indexUid, skey, groupId);
C
Cary Xu 已提交
1258
        // TODO:tsdbEndTSmaCommit();
C
Cary Xu 已提交
1259

C
Cary Xu 已提交
1260
        // Step 3: reset the SSmaStat
C
Cary Xu 已提交
1261
        tsdbResetExpiredWindow(pTsdb, pStat, indexUid, skey);
C
Cary Xu 已提交
1262 1263 1264 1265
      } else {
        tsdbWarn("vgId:%d invalid data skey:%" PRIi64 ", tlen %" PRIi32 " during insert tSma data for %" PRIi64,
                 REPO_ID(pTsdb), skey, tlen, indexUid);
      }
C
Cary Xu 已提交
1266

C
Cary Xu 已提交
1267 1268
      printf("\n");
    }
1269
  }
C
Cary Xu 已提交
1270
  tsdbSmaEndCommit(pEnv);  // TODO: not commit for every insert
1271
  tsdbDestroyTSmaWriteH(&tSmaH);
C
Cary Xu 已提交
1272
  tsdbUnRefSmaStat(pTsdb, pStat);
C
Cary Xu 已提交
1273

C
Cary Xu 已提交
1274 1275 1276
  return TSDB_CODE_SUCCESS;
}

C
Cary Xu 已提交
1277 1278 1279
/**
 * @brief Drop tSma data and local cache
 *        - insert/query reference
C
Cary Xu 已提交
1280 1281 1282
 * @param pTsdb
 * @param msg
 * @return int32_t
C
Cary Xu 已提交
1283 1284
 */
static int32_t tsdbDropTSmaDataImpl(STsdb *pTsdb, int64_t indexUid) {
C
Cary Xu 已提交
1285
  SSmaEnv *pEnv = atomic_load_ptr(&REPO_TSMA_ENV(pTsdb));
C
Cary Xu 已提交
1286 1287 1288 1289 1290 1291

  // clear local cache
  if (pEnv) {
    tsdbDebug("vgId:%d drop tSma local cache for %" PRIi64, REPO_ID(pTsdb), indexUid);

    SSmaStatItem *pItem = taosHashGet(SMA_ENV_STAT_ITEMS(pEnv), &indexUid, sizeof(indexUid));
C
Cary Xu 已提交
1292
    if ((pItem) || ((pItem = *(SSmaStatItem **)pItem))) {
C
Cary Xu 已提交
1293 1294
      if (tsdbSmaStatIsDropped(pItem)) {
        tsdbDebug("vgId:%d tSma stat is already dropped for %" PRIi64, REPO_ID(pTsdb), indexUid);
C
Cary Xu 已提交
1295
        return TSDB_CODE_TDB_INVALID_ACTION;  // TODO: duplicate drop msg would be intercepted by mnode
C
Cary Xu 已提交
1296 1297 1298 1299 1300 1301
      }

      tsdbWLockSma(pEnv);
      if (tsdbSmaStatIsDropped(pItem)) {
        tsdbUnLockSma(pEnv);
        tsdbDebug("vgId:%d tSma stat is already dropped for %" PRIi64, REPO_ID(pTsdb), indexUid);
C
Cary Xu 已提交
1302
        return TSDB_CODE_TDB_INVALID_ACTION;  // TODO: duplicate drop msg would be intercepted by mnode
C
Cary Xu 已提交
1303 1304 1305 1306 1307
      }
      tsdbSmaStatSetDropped(pItem);
      tsdbUnLockSma(pEnv);

      int32_t nSleep = 0;
C
Cary Xu 已提交
1308
      int32_t refVal = INT32_MAX;
C
Cary Xu 已提交
1309
      while (true) {
C
Cary Xu 已提交
1310 1311
        if ((refVal = T_REF_VAL_GET(SMA_ENV_STAT(pEnv))) <= 0) {
          tsdbDebug("vgId:%d drop index %" PRIi64 " since refVal=%d", REPO_ID(pTsdb), indexUid, refVal);
C
Cary Xu 已提交
1312 1313
          break;
        }
C
Cary Xu 已提交
1314
        tsdbDebug("vgId:%d wait 1s to drop index %" PRIi64 " since refVal=%d", REPO_ID(pTsdb), indexUid, refVal);
C
Cary Xu 已提交
1315 1316
        taosSsleep(1);
        if (++nSleep > SMA_DROP_EXPIRED_TIME) {
C
Cary Xu 已提交
1317 1318
          tsdbDebug("vgId:%d drop index %" PRIi64 " after wait %d (refVal=%d)", REPO_ID(pTsdb), indexUid, nSleep,
                    refVal);
C
Cary Xu 已提交
1319 1320 1321 1322 1323 1324 1325 1326 1327
          break;
        };
      }

      tsdbFreeSmaStatItem(pItem);
      tsdbDebug("vgId:%d getTSmaDataImpl failed since no index %" PRIi64 " in local cache", REPO_ID(pTsdb), indexUid);
    }
  }
  // clear sma data files
C
Cary Xu 已提交
1328
  // TODO:
1329
  return TSDB_CODE_SUCCESS;
C
Cary Xu 已提交
1330 1331
}

C
Cary Xu 已提交
1332
static int32_t tsdbSetRSmaDataFile(STSmaWriteH *pSmaH, int32_t fid) {
1333 1334 1335 1336
  STsdb *pTsdb = pSmaH->pTsdb;

  char tSmaFile[TSDB_FILENAME_LEN] = {0};
  snprintf(tSmaFile, TSDB_FILENAME_LEN, "v%df%d.rsma", REPO_ID(pTsdb), fid);
C
Cary Xu 已提交
1337
  pSmaH->dFile.path = strdup(tSmaFile);
C
Cary Xu 已提交
1338 1339 1340 1341

  return TSDB_CODE_SUCCESS;
}

C
Cary Xu 已提交
1342 1343 1344
static int32_t tsdbInsertRSmaDataImpl(STsdb *pTsdb, const char *msg) {
  STsdbCfg     *pCfg = REPO_CFG(pTsdb);
  const SArray *pDataBlocks = (const SArray *)msg;
C
Cary Xu 已提交
1345
  SSmaEnv      *pEnv = atomic_load_ptr(&REPO_RSMA_ENV(pTsdb));
C
Cary Xu 已提交
1346
  int64_t       indexUid = SMA_TEST_INDEX_UID;
C
Cary Xu 已提交
1347

C
Cary Xu 已提交
1348
  if (!pEnv) {
C
Cary Xu 已提交
1349
    terrno = TSDB_CODE_INVALID_PTR;
C
Cary Xu 已提交
1350 1351 1352 1353
    tsdbWarn("vgId:%d insert rSma data failed since pTSmaEnv is NULL", REPO_ID(pTsdb));
    return terrno;
  }

C
Cary Xu 已提交
1354
  if (!pDataBlocks) {
C
Cary Xu 已提交
1355
    terrno = TSDB_CODE_INVALID_PTR;
C
Cary Xu 已提交
1356
    tsdbWarn("vgId:%d insert rSma data failed since pDataBlocks is NULL", REPO_ID(pTsdb));
C
Cary Xu 已提交
1357 1358 1359
    return terrno;
  }

C
Cary Xu 已提交
1360
  if (taosArrayGetSize(pDataBlocks) <= 0) {
C
Cary Xu 已提交
1361
    terrno = TSDB_CODE_INVALID_PARA;
C
Cary Xu 已提交
1362
    tsdbWarn("vgId:%d insert rSma data failed since pDataBlocks is empty", REPO_ID(pTsdb));
C
Cary Xu 已提交
1363 1364 1365
    return TSDB_CODE_FAILED;
  }

C
Cary Xu 已提交
1366
  SSmaStat     *pStat = SMA_ENV_STAT(pEnv);
C
Cary Xu 已提交
1367 1368 1369 1370
  SSmaStatItem *pItem = NULL;

  tsdbRefSmaStat(pTsdb, pStat);

C
Cary Xu 已提交
1371 1372
  if (pStat && SMA_STAT_ITEMS(pStat)) {
    pItem = taosHashGet(SMA_STAT_ITEMS(pStat), &indexUid, sizeof(indexUid));
C
Cary Xu 已提交
1373 1374
  }

C
Cary Xu 已提交
1375
  if (!pItem || !(pItem = *(SSmaStatItem **)pItem) || tsdbSmaStatIsDropped(pItem)) {
C
Cary Xu 已提交
1376 1377
    terrno = TSDB_CODE_TDB_INVALID_SMA_STAT;
    tsdbUnRefSmaStat(pTsdb, pStat);
C
Cary Xu 已提交
1378
    return TSDB_CODE_FAILED;
C
Cary Xu 已提交
1379
  }
C
Cary Xu 已提交
1380

C
Cary Xu 已提交
1381 1382
  STSma *pSma = pItem->pSma;

C
Cary Xu 已提交
1383
  STSmaWriteH tSmaH = {0};
C
Cary Xu 已提交
1384

C
Cary Xu 已提交
1385
  if (tsdbInitTSmaWriteH(&tSmaH, pTsdb, pDataBlocks, pSma->interval, pSma->intervalUnit) != 0) {
C
Cary Xu 已提交
1386 1387
    return TSDB_CODE_FAILED;
  }
C
Cary Xu 已提交
1388

C
Cary Xu 已提交
1389 1390
  char rPath[TSDB_FILENAME_LEN] = {0};
  char aPath[TSDB_FILENAME_LEN] = {0};
C
Cary Xu 已提交
1391 1392 1393 1394 1395 1396 1397 1398 1399
  snprintf(rPath, TSDB_FILENAME_LEN, "%s%s%" PRIi64, SMA_ENV_PATH(pEnv), TD_DIRSEP, indexUid);
  tfsAbsoluteName(REPO_TFS(pTsdb), SMA_ENV_DID(pEnv), rPath, aPath);
  if (!taosCheckExistFile(aPath)) {
    if (tfsMkdirRecurAt(REPO_TFS(pTsdb), rPath, SMA_ENV_DID(pEnv)) != TSDB_CODE_SUCCESS) {
      return TSDB_CODE_FAILED;
    }
  }

  // Step 1: Judge the storage level and days
C
Cary Xu 已提交
1400
  int32_t storageLevel = tsdbGetSmaStorageLevel(pSma->interval, pSma->intervalUnit);
C
Cary Xu 已提交
1401
  int32_t daysPerFile = tsdbGetTSmaDays(pTsdb, tSmaH.interval, storageLevel);
C
Cary Xu 已提交
1402
#if 0
C
Cary Xu 已提交
1403 1404
  int32_t fid = (int32_t)(TSDB_KEY_FID(pData->skey, daysPerFile, pCfg->precision));

C
Cary Xu 已提交
1405 1406
  // Step 2: Set the DFile for storage of SMA index, and iterate/split the TSma data and store to B+Tree index file
  //         - Set and open the DFile or the B+Tree file
C
Cary Xu 已提交
1407
  // TODO: tsdbStartTSmaCommit();
C
Cary Xu 已提交
1408 1409 1410 1411 1412 1413 1414
  tsdbSetTSmaDataFile(&tSmaH, pData, indexUid, fid);
  if (tsdbOpenDBF(pTsdb->pTSmaEnv->dbEnv, &tSmaH.dFile) != 0) {
    tsdbWarn("vgId:%d open DB file %s failed since %s", REPO_ID(pTsdb),
             tSmaH.dFile.path ? tSmaH.dFile.path : "path is NULL", tstrerror(terrno));
    tsdbDestroyTSmaWriteH(&tSmaH);
    return TSDB_CODE_FAILED;
  }
1415

C
Cary Xu 已提交
1416 1417 1418 1419 1420
  if (tsdbInsertTSmaDataSection(&tSmaH, pData) != 0) {
    tsdbWarn("vgId:%d insert tSma data section failed since %s", REPO_ID(pTsdb), tstrerror(terrno));
    tsdbDestroyTSmaWriteH(&tSmaH);
    return TSDB_CODE_FAILED;
  }
C
Cary Xu 已提交
1421 1422
  // TODO:tsdbEndTSmaCommit();

C
Cary Xu 已提交
1423 1424
  // Step 3: reset the SSmaStat
  tsdbResetExpiredWindow(pTsdb, SMA_ENV_STAT(pTsdb->pTSmaEnv), pData->indexUid, pData->skey);
C
Cary Xu 已提交
1425
#endif
C
Cary Xu 已提交
1426

C
Cary Xu 已提交
1427
  tsdbDestroyTSmaWriteH(&tSmaH);
C
Cary Xu 已提交
1428
  tsdbUnRefSmaStat(pTsdb, pStat);
C
Cary Xu 已提交
1429 1430 1431 1432
  return TSDB_CODE_SUCCESS;
}

/**
C
Cary Xu 已提交
1433
 * @brief
C
Cary Xu 已提交
1434 1435 1436
 *
 * @param pSmaH
 * @param pTsdb
C
Cary Xu 已提交
1437 1438
 * @param interval
 * @param intervalUnit
C
Cary Xu 已提交
1439 1440
 * @return int32_t
 */
C
Cary Xu 已提交
1441
static int32_t tsdbInitTSmaReadH(STSmaReadH *pSmaH, STsdb *pTsdb, int64_t interval, int8_t intervalUnit) {
C
Cary Xu 已提交
1442
  pSmaH->pTsdb = pTsdb;
C
Cary Xu 已提交
1443
  pSmaH->interval = tsdbGetIntervalByPrecision(interval, intervalUnit, REPO_CFG(pTsdb)->precision, true);
C
Cary Xu 已提交
1444 1445
  pSmaH->storageLevel = tsdbGetSmaStorageLevel(interval, intervalUnit);
  pSmaH->days = tsdbGetTSmaDays(pTsdb, pSmaH->interval, pSmaH->storageLevel);
1446
  return TSDB_CODE_SUCCESS;
C
Cary Xu 已提交
1447 1448 1449 1450 1451 1452
}

/**
 * @brief Init of tSma FS
 *
 * @param pReadH
1453
 * @param indexUid
C
Cary Xu 已提交
1454
 * @param skey
C
Cary Xu 已提交
1455 1456
 * @return int32_t
 */
1457 1458 1459 1460
static int32_t tsdbInitTSmaFile(STSmaReadH *pSmaH, int64_t indexUid, TSKEY skey) {
  STsdb *pTsdb = pSmaH->pTsdb;

  int32_t fid = (int32_t)(TSDB_KEY_FID(skey, pSmaH->days, REPO_CFG(pTsdb)->precision));
C
Cary Xu 已提交
1461
  char    tSmaFile[TSDB_FILENAME_LEN] = {0};
1462
  snprintf(tSmaFile, TSDB_FILENAME_LEN, "%" PRIi64 "%sv%df%d.tsma", indexUid, TD_DIRSEP, REPO_ID(pTsdb), fid);
C
Cary Xu 已提交
1463 1464 1465
  pSmaH->dFile.path = strdup(tSmaFile);
  pSmaH->smaFsIter.iter = 0;
  pSmaH->smaFsIter.fid = fid;
1466
  return TSDB_CODE_SUCCESS;
C
Cary Xu 已提交
1467 1468 1469 1470 1471 1472 1473 1474 1475 1476 1477
}

/**
 * @brief Set and open tSma file if it has key locates in queryWin.
 *
 * @param pReadH
 * @param param
 * @param queryWin
 * @return true
 * @return false
 */
C
Cary Xu 已提交
1478
static bool tsdbSetAndOpenTSmaFile(STSmaReadH *pReadH, TSKEY *queryKey) {
C
Cary Xu 已提交
1479
  SArray *smaFs = pReadH->pTsdb->fs->cstatus->sf;
C
Cary Xu 已提交
1480 1481
  int32_t nSmaFs = taosArrayGetSize(smaFs);

C
Cary Xu 已提交
1482
  tsdbCloseDBF(&pReadH->dFile);
C
Cary Xu 已提交
1483

C
Cary Xu 已提交
1484
#if 0
C
Cary Xu 已提交
1485 1486 1487 1488
  while (pReadH->smaFsIter.iter < nSmaFs) {
    void *pSmaFile = taosArrayGet(smaFs, pReadH->smaFsIter.iter);
    if (pSmaFile) {  // match(indexName, queryWindow)
      // TODO: select the file by index_name ...
C
Cary Xu 已提交
1489
      pReadH->dFile = pSmaFile;
C
Cary Xu 已提交
1490 1491 1492 1493 1494 1495
      ++pReadH->smaFsIter.iter;
      break;
    }
    ++pReadH->smaFsIter.iter;
  }

C
Cary Xu 已提交
1496
  if (pReadH->pDFile) {
C
Cary Xu 已提交
1497 1498 1499
    tsdbDebug("vg%d: smaFile %s matched", REPO_ID(pReadH->pTsdb), "[pSmaFile dir]");
    return true;
  }
C
Cary Xu 已提交
1500
#endif
C
Cary Xu 已提交
1501 1502 1503 1504 1505

  return false;
}

/**
C
Cary Xu 已提交
1506
 * @brief
C
Cary Xu 已提交
1507
 *
C
Cary Xu 已提交
1508
 * @param pTsdb Return the data between queryWin and fill the pData.
C
Cary Xu 已提交
1509
 * @param pData
C
Cary Xu 已提交
1510 1511
 * @param indexUid
 * @param pQuerySKey
C
Cary Xu 已提交
1512 1513 1514
 * @param nMaxResult The query invoker should control the nMaxResult need to return to avoid OOM.
 * @return int32_t
 */
C
Cary Xu 已提交
1515
static int32_t tsdbGetTSmaDataImpl(STsdb *pTsdb, char *pData, int64_t indexUid, TSKEY querySKey, int32_t nMaxResult) {
C
Cary Xu 已提交
1516 1517
  SSmaEnv  *pEnv = atomic_load_ptr(&REPO_TSMA_ENV(pTsdb));
  SSmaStat *pStat = NULL;
1518 1519

  if (!pEnv) {
C
Cary Xu 已提交
1520 1521 1522 1523 1524
    terrno = TSDB_CODE_INVALID_PTR;
    tsdbWarn("vgId:%d getTSmaDataImpl failed since pTSmaEnv is NULL", REPO_ID(pTsdb));
    return TSDB_CODE_FAILED;
  }

C
Cary Xu 已提交
1525 1526 1527
  pStat = SMA_ENV_STAT(pEnv);

  tsdbRefSmaStat(pTsdb, pStat);
1528
  SSmaStatItem *pItem = taosHashGet(SMA_ENV_STAT_ITEMS(pEnv), &indexUid, sizeof(indexUid));
C
Cary Xu 已提交
1529
  if (!pItem || !(pItem = *(SSmaStatItem **)pItem)) {
C
Cary Xu 已提交
1530 1531
    // Normally pItem should not be NULL, mark all windows as expired and notify query module to fetch raw TS data if
    // it's NULL.
C
Cary Xu 已提交
1532
    tsdbUnRefSmaStat(pTsdb, pStat);
C
Cary Xu 已提交
1533
    terrno = TSDB_CODE_TDB_INVALID_ACTION;
1534
    tsdbDebug("vgId:%d getTSmaDataImpl failed since no index %" PRIi64, REPO_ID(pTsdb), indexUid);
C
Cary Xu 已提交
1535
    return TSDB_CODE_FAILED;
C
Cary Xu 已提交
1536 1537
  }

C
Cary Xu 已提交
1538 1539
#if 0
  int32_t nQueryWin = taosArrayGetSize(pQuerySKey);
C
Cary Xu 已提交
1540
  for (int32_t n = 0; n < nQueryWin; ++n) {
C
Cary Xu 已提交
1541
    TSKEY skey = taosArrayGet(pQuerySKey, n);
C
Cary Xu 已提交
1542
    if (taosHashGet(pItem->expiredWindows, &skey, sizeof(TSKEY))) {
C
Cary Xu 已提交
1543 1544 1545
      // TODO: mark this window as expired.
    }
  }
C
Cary Xu 已提交
1546
#endif
C
Cary Xu 已提交
1547

1548
#if 1
C
Cary Xu 已提交
1549 1550
  int8_t smaStat = 0;
  if (!tsdbSmaStatIsOK(pItem, &smaStat)) {  // TODO: multiple check for large scale sma query
C
Cary Xu 已提交
1551
    tsdbUnRefSmaStat(pTsdb, pStat);
C
Cary Xu 已提交
1552
    terrno = TSDB_CODE_TDB_INVALID_SMA_STAT;
C
Cary Xu 已提交
1553 1554
    tsdbWarn("vgId:%d getTSmaDataImpl failed from index %" PRIi64 " since %s %" PRIi8, REPO_ID(pTsdb), indexUid,
             tstrerror(terrno), smaStat);
C
Cary Xu 已提交
1555 1556 1557
    return TSDB_CODE_FAILED;
  }

C
Cary Xu 已提交
1558
  if (taosHashGet(pItem->expiredWindows, &querySKey, sizeof(TSKEY))) {
C
Cary Xu 已提交
1559
    // TODO: mark this window as expired.
1560 1561 1562 1563 1564
    tsdbDebug("vgId:%d skey %" PRIi64 " of window exists in expired window for index %" PRIi64, REPO_ID(pTsdb),
              querySKey, indexUid);
  } else {
    tsdbDebug("vgId:%d skey %" PRIi64 " of window not in expired window for index %" PRIi64, REPO_ID(pTsdb), querySKey,
              indexUid);
C
Cary Xu 已提交
1565
  }
C
Cary Xu 已提交
1566 1567

  STSma *pTSma = pItem->pSma;
C
Cary Xu 已提交
1568
#endif
1569

C
Cary Xu 已提交
1570
  STSmaReadH tReadH = {0};
C
Cary Xu 已提交
1571
  tsdbInitTSmaReadH(&tReadH, pTsdb, pTSma->interval, pTSma->intervalUnit);
C
Cary Xu 已提交
1572
  tsdbCloseDBF(&tReadH.dFile);
C
Cary Xu 已提交
1573 1574

  tsdbUnRefSmaStat(pTsdb, pStat);
C
Cary Xu 已提交
1575

1576
  tsdbInitTSmaFile(&tReadH, indexUid, querySKey);
C
Cary Xu 已提交
1577
  if (tsdbOpenDBF(pEnv->dbEnv, &tReadH.dFile) != 0) {
C
Cary Xu 已提交
1578 1579 1580 1581
    tsdbWarn("vgId:%d open DBF %s failed since %s", REPO_ID(pTsdb), tReadH.dFile.path, tstrerror(terrno));
    return TSDB_CODE_FAILED;
  }

C
Cary Xu 已提交
1582 1583
  char    smaKey[SMA_KEY_LEN] = {0};
  void   *pSmaKey = &smaKey;
C
Cary Xu 已提交
1584 1585
  int64_t queryGroupId = 1;
  tsdbEncodeTSmaKey(queryGroupId, querySKey, (void **)&pSmaKey);
C
Cary Xu 已提交
1586

C
Cary Xu 已提交
1587 1588
  tsdbDebug("vgId:%d get sma data from %s: smaKey %" PRIx64 "-%" PRIx64 ", keyLen %d", REPO_ID(pTsdb),
            tReadH.dFile.path, *(int64_t *)smaKey, *(int64_t *)POINTER_SHIFT(smaKey, 8), SMA_KEY_LEN);
C
Cary Xu 已提交
1589

C
Cary Xu 已提交
1590 1591
  void   *result = NULL;
  int32_t valueSize = 0;
C
Cary Xu 已提交
1592
  if (!(result = tsdbGetSmaDataByKey(&tReadH.dFile, smaKey, SMA_KEY_LEN, &valueSize))) {
C
Cary Xu 已提交
1593 1594
    tsdbWarn("vgId:%d get sma data failed from smaIndex %" PRIi64 ", smaKey %" PRIx64 "-%" PRIx64 " since %s",
             REPO_ID(pTsdb), indexUid, *(int64_t *)smaKey, *(int64_t *)POINTER_SHIFT(smaKey, 8), tstrerror(terrno));
C
Cary Xu 已提交
1595 1596 1597
    tsdbCloseDBF(&tReadH.dFile);
    return TSDB_CODE_FAILED;
  }
C
Cary Xu 已提交
1598 1599

#ifdef _TEST_SMA_PRINT_DEBUG_LOG_
C
Cary Xu 已提交
1600
  for (uint32_t v = 0; v < valueSize; v += 8) {
C
Cary Xu 已提交
1601
    tsdbWarn("vgId:%d get sma data v[%d]=%" PRIi64, REPO_ID(pTsdb), v, *(int64_t *)POINTER_SHIFT(result, v));
C
Cary Xu 已提交
1602 1603
  }
#endif
wafwerar's avatar
wafwerar 已提交
1604
  taosMemoryFreeClear(result);  // TODO: fill the result to output
C
Cary Xu 已提交
1605

C
Cary Xu 已提交
1606
#if 0
C
Cary Xu 已提交
1607 1608 1609 1610 1611 1612 1613 1614 1615
  int32_t nResult = 0;
  int64_t lastKey = 0;

  while (true) {
    if (nResult >= nMaxResult) {
      break;
    }

    // set and open the file according to the STSma param
C
Cary Xu 已提交
1616
    if (tsdbSetAndOpenTSmaFile(&tReadH, queryWin)) {
C
Cary Xu 已提交
1617 1618 1619 1620 1621 1622 1623 1624 1625 1626 1627
      char bTree[100] = "\0";
      while (strncmp(bTree, "has more nodes", 100) == 0) {
        if (nResult >= nMaxResult) {
          break;
        }
        // tsdbGetDataFromBTree(bTree, queryWin, lastKey)
        // fill the pData
        ++nResult;
      }
    }
  }
C
Cary Xu 已提交
1628
#endif
C
Cary Xu 已提交
1629
  // read data from file and fill the result
C
Cary Xu 已提交
1630
  tsdbCloseDBF(&tReadH.dFile);
C
Cary Xu 已提交
1631 1632 1633
  return TSDB_CODE_SUCCESS;
}

C
Cary Xu 已提交
1634 1635
int32_t tsdbCreateTSma(STsdb *pTsdb, char *pMsg) {
  SSmaCfg vCreateSmaReq = {0};
C
Cary Xu 已提交
1636
  if (!tDeserializeSVCreateTSmaReq(pMsg, &vCreateSmaReq)) {
C
Cary Xu 已提交
1637
    terrno = TSDB_CODE_OUT_OF_MEMORY;
C
Cary Xu 已提交
1638
    tsdbWarn("vgId:%d tsma create msg received but deserialize failed since %s", REPO_ID(pTsdb), terrstr(terrno));
C
Cary Xu 已提交
1639 1640
    return -1;
  }
C
Cary Xu 已提交
1641

C
Cary Xu 已提交
1642 1643
  tsdbDebug("vgId:%d tsma create msg %s:%" PRIi64 " for table %" PRIi64 " received", REPO_ID(pTsdb),
            vCreateSmaReq.tSma.indexName, vCreateSmaReq.tSma.indexUid, vCreateSmaReq.tSma.tableUid);
C
Cary Xu 已提交
1644 1645 1646 1647

  // record current timezone of server side
  vCreateSmaReq.tSma.timezoneInt = tsTimezone;

H
Hongze Cheng 已提交
1648
  if (metaCreateTSma(REPO_META(pTsdb), &vCreateSmaReq) < 0) {
C
Cary Xu 已提交
1649
    // TODO: handle error
C
Cary Xu 已提交
1650 1651
    tsdbWarn("vgId:%d tsma %s:%" PRIi64 " create failed for table %" PRIi64 " since %s", REPO_ID(pTsdb),
             vCreateSmaReq.tSma.indexName, vCreateSmaReq.tSma.indexUid, vCreateSmaReq.tSma.tableUid, terrstr(terrno));
C
Cary Xu 已提交
1652 1653 1654 1655 1656 1657 1658 1659
    tdDestroyTSma(&vCreateSmaReq.tSma);
    return -1;
  }

  tsdbTSmaAdd(pTsdb, 1);

  tdDestroyTSma(&vCreateSmaReq.tSma);
  // TODO: return directly or go on follow steps?
1660
  return TSDB_CODE_SUCCESS;
C
Cary Xu 已提交
1661 1662 1663 1664
}

int32_t tsdbDropTSma(STsdb *pTsdb, char *pMsg) {
  SVDropTSmaReq vDropSmaReq = {0};
C
Cary Xu 已提交
1665
  if (!tDeserializeSVDropTSmaReq(pMsg, &vDropSmaReq)) {
C
Cary Xu 已提交
1666 1667 1668 1669 1670 1671 1672 1673 1674 1675 1676
    terrno = TSDB_CODE_OUT_OF_MEMORY;
    return -1;
  }

  // TODO: send msg to stream computing to drop tSma
  // if ((send msg to stream computing) < 0) {
  //   tdDestroyTSma(&vCreateSmaReq);
  //   return -1;
  // }
  //

H
Hongze Cheng 已提交
1677
  if (metaDropTSma(REPO_META(pTsdb), vDropSmaReq.indexUid) < 0) {
C
Cary Xu 已提交
1678 1679 1680 1681 1682 1683 1684 1685 1686
    // TODO: handle error
    return -1;
  }

  if (tsdbDropTSmaData(pTsdb, vDropSmaReq.indexUid) < 0) {
    // TODO: handle error
    return -1;
  }

C
Cary Xu 已提交
1687
  tsdbTSmaSub(pTsdb, 1);
C
Cary Xu 已提交
1688 1689

  // TODO: return directly or go on follow steps?
1690
  return TSDB_CODE_SUCCESS;
C
Cary Xu 已提交
1691 1692
}

C
Cary Xu 已提交
1693
/**
C
Cary Xu 已提交
1694
 * @brief Check and init qTaskInfo_t, only applicable to stable with SRSmaParam.
C
Cary Xu 已提交
1695 1696 1697 1698 1699 1700
 *
 * @param pTsdb
 * @param pMeta
 * @param pReq
 * @return int32_t
 */
1701
int32_t tsdbRegisterRSma(STsdb *pTsdb, SMeta *pMeta, SVCreateStbReq *pReq, SMsgCb *pMsgCb) {
C
Cary Xu 已提交
1702 1703
  if (!pReq->rollup) {
    tsdbDebug("vgId:%d return directly since no rollup for stable %s %" PRIi64, REPO_ID(pTsdb), pReq->name, pReq->suid);
C
Cary Xu 已提交
1704 1705 1706
    return TSDB_CODE_SUCCESS;
  }

C
Cary Xu 已提交
1707 1708
  SRSmaParam *param = &pReq->pRSmaParam;

C
Cary Xu 已提交
1709
  if ((param->qmsg1Len == 0) && (param->qmsg2Len == 0)) {
C
Cary Xu 已提交
1710
    tsdbWarn("vgId:%d no qmsg1/qmsg2 for rollup stable %s %" PRIi64, REPO_ID(pTsdb), pReq->name, pReq->suid);
C
Cary Xu 已提交
1711 1712 1713 1714 1715 1716 1717 1718 1719 1720 1721 1722
    return TSDB_CODE_SUCCESS;
  }

  if (tsdbCheckAndInitSmaEnv(pTsdb, TSDB_SMA_TYPE_ROLLUP) != TSDB_CODE_SUCCESS) {
    terrno = TSDB_CODE_TDB_INIT_FAILED;
    return TSDB_CODE_FAILED;
  }

  SSmaEnv   *pEnv = REPO_RSMA_ENV(pTsdb);
  SSmaStat  *pStat = SMA_ENV_STAT(pEnv);
  SRSmaInfo *pRSmaInfo = NULL;

C
Cary Xu 已提交
1723
  pRSmaInfo = taosHashGet(SMA_STAT_INFO_HASH(pStat), &pReq->suid, sizeof(tb_uid_t));
C
Cary Xu 已提交
1724
  if (pRSmaInfo) {
C
Cary Xu 已提交
1725
    tsdbWarn("vgId:%d rsma info already exists for stb: %s, %" PRIi64, REPO_ID(pTsdb), pReq->name, pReq->suid);
C
Cary Xu 已提交
1726
    return TSDB_CODE_SUCCESS;
C
Cary Xu 已提交
1727 1728
  }

C
Cary Xu 已提交
1729
  pRSmaInfo = (SRSmaInfo *)taosMemoryCalloc(1, sizeof(SRSmaInfo));
C
Cary Xu 已提交
1730
  if (!pRSmaInfo) {
C
Cary Xu 已提交
1731 1732 1733 1734
    terrno = TSDB_CODE_OUT_OF_MEMORY;
    return TSDB_CODE_FAILED;
  }

C
Cary Xu 已提交
1735
  STqReadHandle *pReadHandle = tqInitSubmitMsgScanner(pMeta);
C
Cary Xu 已提交
1736
  if (!pReadHandle) {
C
Cary Xu 已提交
1737
    taosMemoryFree(pRSmaInfo);
C
Cary Xu 已提交
1738 1739 1740 1741 1742 1743 1744
    terrno = TSDB_CODE_OUT_OF_MEMORY;
    return TSDB_CODE_FAILED;
  }

  SReadHandle handle = {
      .reader = pReadHandle,
      .meta = pMeta,
1745
      .pMsgCb = pMsgCb,
C
Cary Xu 已提交
1746 1747 1748 1749
  };

  if (param->qmsg1) {
    pRSmaInfo->taskInfo[0] = qCreateStreamExecTaskInfo(param->qmsg1, &handle);
C
Cary Xu 已提交
1750
    if (!pRSmaInfo->taskInfo[0]) {
C
Cary Xu 已提交
1751 1752 1753 1754 1755 1756 1757 1758
      taosMemoryFree(pRSmaInfo);
      taosMemoryFree(pReadHandle);
      return TSDB_CODE_FAILED;
    }
  }

  if (param->qmsg2) {
    pRSmaInfo->taskInfo[1] = qCreateStreamExecTaskInfo(param->qmsg2, &handle);
C
Cary Xu 已提交
1759
    if (!pRSmaInfo->taskInfo[1]) {
C
Cary Xu 已提交
1760 1761 1762 1763 1764 1765
      taosMemoryFree(pRSmaInfo);
      taosMemoryFree(pReadHandle);
      return TSDB_CODE_FAILED;
    }
  }

C
Cary Xu 已提交
1766
  if (taosHashPut(SMA_STAT_INFO_HASH(pStat), &pReq->suid, sizeof(tb_uid_t), &pRSmaInfo, sizeof(pRSmaInfo)) !=
C
Cary Xu 已提交
1767
      TSDB_CODE_SUCCESS) {
C
Cary Xu 已提交
1768
    return TSDB_CODE_FAILED;
C
Cary Xu 已提交
1769
  } else {
C
Cary Xu 已提交
1770
    tsdbDebug("vgId:%d register rsma info succeed for suid:%" PRIi64, REPO_ID(pTsdb), pReq->suid);
C
Cary Xu 已提交
1771 1772 1773 1774 1775
  }

  return TSDB_CODE_SUCCESS;
}

C
Cary Xu 已提交
1776 1777 1778 1779 1780 1781 1782 1783
/**
 * @brief store suid/[uids], prefer to use array and then hash
 *
 * @param pStore
 * @param suid
 * @param uid
 * @return int32_t
 */
C
Cary Xu 已提交
1784
static int32_t tsdbUidStorePut(STbUidStore *pStore, tb_uid_t suid, tb_uid_t *uid) {
C
Cary Xu 已提交
1785
  // prefer to store suid/uids in array
C
Cary Xu 已提交
1786
  if ((suid == pStore->suid) || (pStore->suid == 0)) {
C
Cary Xu 已提交
1787
    if (pStore->suid == 0) {
C
Cary Xu 已提交
1788 1789
      pStore->suid = suid;
    }
C
Cary Xu 已提交
1790 1791 1792 1793 1794 1795 1796
    if (uid) {
      if (!pStore->tbUids) {
        if (!(pStore->tbUids = taosArrayInit(1, sizeof(tb_uid_t)))) {
          terrno = TSDB_CODE_OUT_OF_MEMORY;
          return TSDB_CODE_FAILED;
        }
      }
C
Cary Xu 已提交
1797
      if (!taosArrayPush(pStore->tbUids, uid)) {
C
Cary Xu 已提交
1798 1799
        return TSDB_CODE_FAILED;
      }
C
Cary Xu 已提交
1800 1801
    }
  } else {
C
Cary Xu 已提交
1802 1803
    // store other suid/uids in hash when multiple stable/table included in 1 batch of request
    if (!pStore->uidHash) {
C
Cary Xu 已提交
1804
      pStore->uidHash = taosHashInit(4, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BIGINT), false, HASH_ENTRY_LOCK);
C
Cary Xu 已提交
1805
      if (!pStore->uidHash) {
C
Cary Xu 已提交
1806 1807 1808
        return TSDB_CODE_FAILED;
      }
    }
C
Cary Xu 已提交
1809 1810 1811
    if (uid) {
      SArray *uidArray = taosHashGet(pStore->uidHash, &suid, sizeof(tb_uid_t));
      if (uidArray && ((uidArray = *(SArray **)uidArray))) {
C
Cary Xu 已提交
1812
        taosArrayPush(uidArray, uid);
C
Cary Xu 已提交
1813 1814 1815 1816 1817 1818
      } else {
        SArray *pUidArray = taosArrayInit(1, sizeof(tb_uid_t));
        if (!pUidArray) {
          terrno = TSDB_CODE_OUT_OF_MEMORY;
          return TSDB_CODE_FAILED;
        }
C
Cary Xu 已提交
1819
        if (!taosArrayPush(pUidArray, uid)) {
C
Cary Xu 已提交
1820 1821 1822 1823 1824 1825
          terrno = TSDB_CODE_OUT_OF_MEMORY;
          return TSDB_CODE_FAILED;
        }
        if (taosHashPut(pStore->uidHash, &suid, sizeof(suid), &pUidArray, sizeof(pUidArray)) != 0) {
          return TSDB_CODE_FAILED;
        }
C
Cary Xu 已提交
1826
      }
C
Cary Xu 已提交
1827 1828
    } else {
      if (taosHashPut(pStore->uidHash, &suid, sizeof(suid), NULL, 0) != 0) {
C
Cary Xu 已提交
1829 1830 1831 1832 1833 1834 1835
        return TSDB_CODE_FAILED;
      }
    }
  }
  return TSDB_CODE_SUCCESS;
}

C
Cary Xu 已提交
1836 1837 1838 1839
void tsdbUidStoreDestory(STbUidStore *pStore) {
  if (pStore) {
    if (pStore->uidHash) {
      if (pStore->tbUids) {
C
Cary Xu 已提交
1840
        // When pStore->tbUids not NULL, the pStore->uidHash has k/v; otherwise pStore->uidHash only has keys.
C
Cary Xu 已提交
1841 1842 1843 1844 1845 1846 1847 1848 1849 1850 1851 1852 1853 1854
        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 *tsdbUidStoreFree(STbUidStore *pStore) {
C
Cary Xu 已提交
1855 1856 1857 1858
  if (pStore) {
    tsdbUidStoreDestory(pStore);
    taosMemoryFree(pStore);
  }
C
Cary Xu 已提交
1859 1860 1861 1862 1863 1864 1865 1866 1867 1868 1869 1870
  return NULL;
}

/**
 * @brief fetch suid/uids when create child tables of rollup SMA
 *
 * @param pTsdb
 * @param ppStore
 * @param suid
 * @param uid
 * @return int32_t
 */
C
Cary Xu 已提交
1871
int32_t tsdbFetchTbUidList(STsdb *pTsdb, STbUidStore **ppStore, tb_uid_t suid, tb_uid_t uid) {
C
Cary Xu 已提交
1872
  SSmaEnv *pEnv = REPO_RSMA_ENV((STsdb *)pTsdb);
C
Cary Xu 已提交
1873 1874

  // only applicable to rollup SMA ctables
C
Cary Xu 已提交
1875 1876 1877 1878 1879 1880 1881 1882 1883 1884 1885
  if (!pEnv) {
    return TSDB_CODE_SUCCESS;
  }

  SSmaStat *pStat = SMA_ENV_STAT(pEnv);
  SHashObj *infoHash = NULL;
  if (!pStat || !(infoHash = SMA_STAT_INFO_HASH(pStat))) {
    terrno = TSDB_CODE_TDB_INVALID_SMA_STAT;
    return TSDB_CODE_FAILED;
  }

C
Cary Xu 已提交
1886
  // info cached when create rsma stable and return directly for non-rsma ctables
C
Cary Xu 已提交
1887
  if (!taosHashGet(infoHash, &suid, sizeof(tb_uid_t))) {
C
Cary Xu 已提交
1888 1889 1890
    return TSDB_CODE_SUCCESS;
  }

C
Cary Xu 已提交
1891 1892
  ASSERT(ppStore != NULL);

C
Cary Xu 已提交
1893
  if (!(*ppStore)) {
C
Cary Xu 已提交
1894
    if (tsdbUidStoreInit(ppStore) != 0) {
C
Cary Xu 已提交
1895 1896 1897
      return TSDB_CODE_FAILED;
    }
  }
C
Cary Xu 已提交
1898

C
Cary Xu 已提交
1899
  if (tsdbUidStorePut(*ppStore, suid, &uid) != 0) {
C
Cary Xu 已提交
1900 1901 1902
    *ppStore = tsdbUidStoreFree(*ppStore);
    return TSDB_CODE_FAILED;
  }
C
Cary Xu 已提交
1903

C
Cary Xu 已提交
1904 1905 1906
  return TSDB_CODE_SUCCESS;
}

C
Cary Xu 已提交
1907 1908 1909 1910 1911 1912 1913 1914 1915
static FORCE_INLINE int32_t tsdbUpdateTbUidListImpl(STsdb *pTsdb, tb_uid_t *suid, SArray *tbUids) {
  SSmaEnv   *pEnv = REPO_RSMA_ENV(pTsdb);
  SSmaStat  *pStat = SMA_ENV_STAT(pEnv);
  SRSmaInfo *pRSmaInfo = NULL;

  if (!suid || !tbUids) {
    terrno = TSDB_CODE_INVALID_PTR;
    tsdbError("vgId:%d failed to get rsma info for uid:%" PRIi64 " since %s", REPO_ID(pTsdb), *suid, terrstr(terrno));
    return TSDB_CODE_FAILED;
C
Cary Xu 已提交
1916
  }
C
Cary Xu 已提交
1917 1918 1919 1920 1921 1922 1923 1924 1925 1926 1927

  pRSmaInfo = taosHashGet(SMA_STAT_INFO_HASH(pStat), suid, sizeof(tb_uid_t));
  if (!pRSmaInfo || !(pRSmaInfo = *(SRSmaInfo **)pRSmaInfo)) {
    tsdbError("vgId:%d failed to get rsma info for uid:%" PRIi64, REPO_ID(pTsdb), *suid);
    terrno = TSDB_CODE_TDB_INVALID_SMA_STAT;
    return TSDB_CODE_FAILED;
  }

  if (pRSmaInfo->taskInfo[0] && (qUpdateQualifiedTableId(pRSmaInfo->taskInfo[0], tbUids, true) != 0)) {
    tsdbError("vgId:%d update tbUidList failed for uid:%" PRIi64 " since %s", REPO_ID(pTsdb), *suid, terrstr(terrno));
    return TSDB_CODE_FAILED;
C
Cary Xu 已提交
1928 1929 1930
  } else {
    tsdbDebug("vgId:%d update tbUidList succeed for qTaskInfo:%p with suid:%" PRIi64 ", uid:%" PRIi64, REPO_ID(pTsdb),
              pRSmaInfo->taskInfo[0], *suid, *(int64_t *)taosArrayGet(tbUids, 0));
C
Cary Xu 已提交
1931
  }
C
Cary Xu 已提交
1932

C
Cary Xu 已提交
1933 1934 1935
  if (pRSmaInfo->taskInfo[1] && (qUpdateQualifiedTableId(pRSmaInfo->taskInfo[1], tbUids, true) != 0)) {
    tsdbError("vgId:%d update tbUidList failed for uid:%" PRIi64 " since %s", REPO_ID(pTsdb), *suid, terrstr(terrno));
    return TSDB_CODE_FAILED;
C
Cary Xu 已提交
1936 1937 1938
  } else {
    tsdbDebug("vgId:%d update tbUidList succeed for qTaskInfo:%p with suid:%" PRIi64 ", uid:%" PRIi64, REPO_ID(pTsdb),
              pRSmaInfo->taskInfo[1], *suid, *(int64_t *)taosArrayGet(tbUids, 0));
C
Cary Xu 已提交
1939 1940 1941
  }

  return TSDB_CODE_SUCCESS;
C
Cary Xu 已提交
1942 1943
}

C
Cary Xu 已提交
1944 1945
int32_t tsdbUpdateTbUidList(STsdb *pTsdb, STbUidStore *pStore) {
  if (!pStore || (taosArrayGetSize(pStore->tbUids) == 0)) {
C
Cary Xu 已提交
1946 1947 1948
    return TSDB_CODE_SUCCESS;
  }

C
Cary Xu 已提交
1949
  if (tsdbUpdateTbUidListImpl(pTsdb, &pStore->suid, pStore->tbUids) != TSDB_CODE_SUCCESS) {
C
Cary Xu 已提交
1950 1951 1952
    return TSDB_CODE_FAILED;
  }

C
Cary Xu 已提交
1953 1954 1955 1956 1957 1958 1959 1960 1961 1962 1963 1964
  void *pIter = taosHashIterate(pStore->uidHash, NULL);
  while (pIter) {
    tb_uid_t *pTbSuid = (tb_uid_t *)taosHashGetKey(pIter, NULL);
    SArray   *pTbUids = *(SArray **)pIter;

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

    pIter = taosHashIterate(pStore->uidHash, pIter);
  }
C
Cary Xu 已提交
1965 1966 1967
  return TSDB_CODE_SUCCESS;
}

C
Cary Xu 已提交
1968 1969 1970 1971 1972 1973 1974 1975 1976 1977 1978 1979 1980 1981 1982
static int32_t tsdbProcessSubmitReq(STsdb *pTsdb, int64_t version, void *pReq) {
  if (!pReq) {
    terrno = TSDB_CODE_INVALID_PTR;
    return TSDB_CODE_FAILED;
  }

  SSubmitReq *pSubmitReq = (SSubmitReq *)pReq;

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

  return TSDB_CODE_SUCCESS;
}

C
Cary Xu 已提交
1983
static int32_t tsdbFetchSubmitReqSuids(SSubmitReq *pMsg, STbUidStore *pStore) {
C
Cary Xu 已提交
1984 1985 1986 1987 1988 1989 1990 1991
  ASSERT(pMsg != NULL);
  SSubmitMsgIter msgIter = {0};
  SSubmitBlk    *pBlock = NULL;
  SSubmitBlkIter blkIter = {0};
  STSRow        *row = NULL;

  terrno = TSDB_CODE_SUCCESS;

C
Cary Xu 已提交
1992
  if (tInitSubmitMsgIter(pMsg, &msgIter) < 0) return -1;
C
Cary Xu 已提交
1993
  while (true) {
C
Cary Xu 已提交
1994
    if (tGetSubmitMsgNext(&msgIter, &pBlock) < 0) return -1;
C
Cary Xu 已提交
1995

C
Cary Xu 已提交
1996
    if (!pBlock) break;
C
Cary Xu 已提交
1997
    tsdbUidStorePut(pStore, msgIter.suid, NULL);
C
Cary Xu 已提交
1998
    pStore->uid = msgIter.uid;  // TODO: remove, just for debugging
C
Cary Xu 已提交
1999 2000 2001 2002 2003 2004
  }

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

C
Cary Xu 已提交
2005
static FORCE_INLINE int32_t tsdbExecuteRSmaImpl(STsdb *pTsdb, const void *pMsg, int32_t inputType,
C
Cary Xu 已提交
2006 2007
                                                qTaskInfo_t *taskInfo, STSchema *pTSchema, tb_uid_t suid, tb_uid_t uid,
                                                int8_t level) {
C
Cary Xu 已提交
2008
  SArray *pResult = NULL;
C
Cary Xu 已提交
2009 2010 2011 2012 2013 2014

  if (!taskInfo) {
    tsdbDebug("vgId:%d no qTaskInfo to execute rsma %" PRIi8 " task for suid:%" PRIu64, REPO_ID(pTsdb), level, suid);
    return TSDB_CODE_SUCCESS;
  }

C
Cary Xu 已提交
2015
  tsdbDebug("vgId:%d execute rsma %" PRIi8 " task for qTaskInfo:%p suid:%" PRIu64, REPO_ID(pTsdb), level, taskInfo,
C
Cary Xu 已提交
2016
            suid);
C
Cary Xu 已提交
2017

C
Cary Xu 已提交
2018 2019
  qSetStreamInput(taskInfo, pMsg, inputType);
  while (1) {
C
Cary Xu 已提交
2020
    SSDataBlock *output = NULL;
C
Cary Xu 已提交
2021 2022 2023 2024 2025 2026 2027 2028 2029 2030 2031 2032 2033 2034 2035 2036 2037 2038 2039 2040
    uint64_t     ts;
    if (qExecTask(taskInfo, &output, &ts) < 0) {
      ASSERT(false);
    }
    if (!output) {
      break;
    }
    if (!pResult) {
      pResult = taosArrayInit(0, sizeof(SSDataBlock));
      if (!pResult) {
        terrno = TSDB_CODE_OUT_OF_MEMORY;
        return TSDB_CODE_FAILED;
      }
    }

    taosArrayPush(pResult, output);
  }

  if (taosArrayGetSize(pResult) > 0) {
    blockDebugShowData(pResult);
C
Cary Xu 已提交
2041
    STsdb      *sinkTsdb = (level == TSDB_RETENTION_L1 ? pTsdb->pVnode->pRSma1 : pTsdb->pVnode->pRSma2);
C
Cary Xu 已提交
2042
    SSubmitReq *pReq = NULL;
C
Cary Xu 已提交
2043 2044 2045 2046 2047 2048 2049 2050 2051 2052
    if (buildSubmitReqFromDataBlock(&pReq, pResult, pTSchema, TD_VID(pTsdb->pVnode), uid, suid) != 0) {
      taosArrayDestroy(pResult);
      return TSDB_CODE_FAILED;
    }
    if (tsdbProcessSubmitReq(sinkTsdb, INT64_MAX, pReq) != 0) {
      taosArrayDestroy(pResult);
      taosMemoryFreeClear(pReq);
      return TSDB_CODE_FAILED;
    }
    taosMemoryFreeClear(pReq);
C
Cary Xu 已提交
2053
  } else {
C
Cary Xu 已提交
2054
    tsdbWarn("vgId:%d no rsma % " PRIi8 " data generated since %s", REPO_ID(pTsdb), level, tstrerror(terrno));
C
Cary Xu 已提交
2055
  }
C
Cary Xu 已提交
2056

C
Cary Xu 已提交
2057 2058 2059 2060 2061
  taosArrayDestroy(pResult);

  return TSDB_CODE_SUCCESS;
}

C
Cary Xu 已提交
2062
static int32_t tsdbExecuteRSma(STsdb *pTsdb, const void *pMsg, int32_t inputType, tb_uid_t suid, tb_uid_t uid) {
C
Cary Xu 已提交
2063 2064
  SSmaEnv *pEnv = REPO_RSMA_ENV(pTsdb);
  if (!pEnv) {
C
Cary Xu 已提交
2065
    // only applicable when rsma env exists
C
Cary Xu 已提交
2066 2067 2068
    return TSDB_CODE_SUCCESS;
  }

C
Cary Xu 已提交
2069
  ASSERT(uid != 0);  // TODO: remove later
C
Cary Xu 已提交
2070

C
Cary Xu 已提交
2071 2072 2073
  SSmaStat  *pStat = SMA_ENV_STAT(pEnv);
  SRSmaInfo *pRSmaInfo = NULL;

C
Cary Xu 已提交
2074
  pRSmaInfo = taosHashGet(SMA_STAT_INFO_HASH(pStat), &suid, sizeof(tb_uid_t));
C
Cary Xu 已提交
2075 2076

  if (!pRSmaInfo || !(pRSmaInfo = *(SRSmaInfo **)pRSmaInfo)) {
C
Cary Xu 已提交
2077
    tsdbDebug("vgId:%d no rsma info for suid:%" PRIu64, REPO_ID(pTsdb), suid);
C
Cary Xu 已提交
2078 2079
    return TSDB_CODE_SUCCESS;
  }
C
Cary Xu 已提交
2080 2081 2082 2083
  if (!pRSmaInfo->taskInfo[0]) {
    tsdbDebug("vgId:%d no rsma qTaskInfo for suid:%" PRIu64, REPO_ID(pTsdb), suid);
    return TSDB_CODE_SUCCESS;
  }
C
Cary Xu 已提交
2084

C
Cary Xu 已提交
2085
  if (inputType == STREAM_DATA_TYPE_SUBMIT_BLOCK) {
C
Cary Xu 已提交
2086 2087
    // TODO: use the proper schema instead of 0, and cache STSchema in cache
    STSchema *pTSchema = metaGetTbTSchema(pTsdb->pVnode->pMeta, suid, 0);
C
Cary Xu 已提交
2088 2089 2090 2091
    if (!pTSchema) {
      terrno = TSDB_CODE_TDB_IVD_TB_SCHEMA_VERSION;
      return TSDB_CODE_FAILED;
    }
C
Cary Xu 已提交
2092 2093 2094
    tsdbExecuteRSmaImpl(pTsdb, pMsg, inputType, pRSmaInfo->taskInfo[0], pTSchema, suid, uid, TSDB_RETENTION_L1);
    tsdbExecuteRSmaImpl(pTsdb, pMsg, inputType, pRSmaInfo->taskInfo[1], pTSchema, suid, uid, TSDB_RETENTION_L2);
    taosMemoryFree(pTSchema);
C
Cary Xu 已提交
2095
  }
C
Cary Xu 已提交
2096

C
Cary Xu 已提交
2097 2098
  return TSDB_CODE_SUCCESS;
}
C
Cary Xu 已提交
2099

C
Cary Xu 已提交
2100
int32_t tsdbTriggerRSma(STsdb *pTsdb, void *pMsg, int32_t inputType) {
C
Cary Xu 已提交
2101 2102 2103 2104 2105 2106 2107 2108 2109 2110 2111
  SSmaEnv *pEnv = REPO_RSMA_ENV(pTsdb);
  if (!pEnv) {
    // only applicable when rsma env exists
    return TSDB_CODE_SUCCESS;
  }

  if (inputType == STREAM_DATA_TYPE_SUBMIT_BLOCK) {
    STbUidStore uidStore = {0};
    tsdbFetchSubmitReqSuids(pMsg, &uidStore);

    if (uidStore.suid != 0) {
C
Cary Xu 已提交
2112
      tsdbExecuteRSma(pTsdb, pMsg, inputType, uidStore.suid, uidStore.uid);
C
Cary Xu 已提交
2113 2114 2115 2116

      void *pIter = taosHashIterate(uidStore.uidHash, NULL);
      while (pIter) {
        tb_uid_t *pTbSuid = (tb_uid_t *)taosHashGetKey(pIter, NULL);
C
Cary Xu 已提交
2117
        tsdbExecuteRSma(pTsdb, pMsg, inputType, *pTbSuid, 0);
C
Cary Xu 已提交
2118
        pIter = taosHashIterate(uidStore.uidHash, pIter);
C
Cary Xu 已提交
2119
      }
C
Cary Xu 已提交
2120 2121

      tsdbUidStoreDestory(&uidStore);
C
Cary Xu 已提交
2122 2123 2124 2125 2126
    }
  }
  return TSDB_CODE_SUCCESS;
}

C
Cary Xu 已提交
2127
#if 0
C
Cary Xu 已提交
2128 2129 2130 2131 2132 2133 2134 2135 2136 2137
/**
 * @brief Get the start TS key of the last data block of one interval/sliding.
 *
 * @param pTsdb
 * @param param
 * @param result
 * @return int32_t
 *         1) Return 0 and fill the result if the check procedure is normal;
 *         2) Return -1 if error occurs during the check procedure.
 */
C
Cary Xu 已提交
2138
int32_t tsdbGetTSmaStatus(STsdb *pTsdb, void *smaIndex, void *result) {
C
Cary Xu 已提交
2139 2140 2141 2142 2143 2144 2145 2146 2147 2148 2149 2150 2151 2152 2153 2154
  const char *procedure = "";
  if (strncmp(procedure, "get the start TS key of the last data block", 100) != 0) {
    return -1;
  }
  // fill the result
  return TSDB_CODE_SUCCESS;
}

/**
 * @brief Remove the tSma data files related to param between pWin.
 *
 * @param pTsdb
 * @param param
 * @param pWin
 * @return int32_t
 */
C
Cary Xu 已提交
2155
int32_t tsdbRemoveTSmaData(STsdb *pTsdb, void *smaIndex, STimeWindow *pWin) {
C
Cary Xu 已提交
2156 2157 2158 2159
  // for ("tSmaFiles of param-interval-sliding between pWin") {
  //   // remove the tSmaFile
  // }
  return TSDB_CODE_SUCCESS;
C
Cary Xu 已提交
2160
}
C
Cary Xu 已提交
2161 2162
#endif

C
Cary Xu 已提交
2163
// TODO: Who is responsible for resource allocate and release?
C
Cary Xu 已提交
2164
int32_t tsdbInsertTSmaData(STsdb *pTsdb, int64_t indexUid, const char *msg) {
C
Cary Xu 已提交
2165
  int32_t code = TSDB_CODE_SUCCESS;
C
Cary Xu 已提交
2166
  if ((code = tsdbInsertTSmaDataImpl(pTsdb, indexUid, msg)) < 0) {
C
Cary Xu 已提交
2167 2168
    tsdbWarn("vgId:%d insert tSma data failed since %s", REPO_ID(pTsdb), tstrerror(terrno));
  }
C
Cary Xu 已提交
2169
  // TODO: destroy SSDataBlocks(msg)
C
Cary Xu 已提交
2170 2171 2172
  return code;
}

2173
int32_t tsdbUpdateSmaWindow(STsdb *pTsdb, SSubmitReq *pMsg, int64_t version) {
C
Cary Xu 已提交
2174
  int32_t code = TSDB_CODE_SUCCESS;
2175
  if ((code = tsdbUpdateExpiredWindowImpl(pTsdb, pMsg, version)) < 0) {
C
Cary Xu 已提交
2176 2177 2178 2179 2180 2181 2182 2183 2184 2185 2186 2187 2188
    tsdbWarn("vgId:%d update expired sma window failed since %s", REPO_ID(pTsdb), tstrerror(terrno));
  }
  return code;
}

int32_t tsdbInsertRSmaData(STsdb *pTsdb, char *msg) {
  int32_t code = TSDB_CODE_SUCCESS;
  if ((code = tsdbInsertRSmaDataImpl(pTsdb, msg)) < 0) {
    tsdbWarn("vgId:%d insert rSma data failed since %s", REPO_ID(pTsdb), tstrerror(terrno));
  }
  return code;
}

C
Cary Xu 已提交
2189
int32_t tsdbGetTSmaData(STsdb *pTsdb, char *pData, int64_t indexUid, TSKEY querySKey, int32_t nMaxResult) {
C
Cary Xu 已提交
2190
  int32_t code = TSDB_CODE_SUCCESS;
C
Cary Xu 已提交
2191
  if ((code = tsdbGetTSmaDataImpl(pTsdb, pData, indexUid, querySKey, nMaxResult)) < 0) {
C
Cary Xu 已提交
2192 2193 2194
    tsdbWarn("vgId:%d get tSma data failed since %s", REPO_ID(pTsdb), tstrerror(terrno));
  }
  return code;
C
Cary Xu 已提交
2195 2196 2197 2198 2199 2200 2201 2202
}

int32_t tsdbDropTSmaData(STsdb *pTsdb, int64_t indexUid) {
  int32_t code = TSDB_CODE_SUCCESS;
  if ((code = tsdbDropTSmaDataImpl(pTsdb, indexUid)) < 0) {
    tsdbWarn("vgId:%d drop tSma data failed since %s", REPO_ID(pTsdb), tstrerror(terrno));
  }
  return code;
C
Cary Xu 已提交
2203
}