tsdbMain.h 18.2 KB
Newer Older
H
more  
hzcheng 已提交
1 2 3 4 5 6 7 8 9 10 11 12 13 14
/*
 * 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
TD-34  
hzcheng 已提交
15 16
#ifndef _TD_TSDB_MAIN_H_
#define _TD_TSDB_MAIN_H_
H
more  
Hongze Cheng 已提交
17

S
TD-1057  
Shengliang Guan 已提交
18
#include "os.h"
H
TD-353  
Hongze Cheng 已提交
19 20
#include "hash.h"
#include "tcoding.h"
S
slguan 已提交
21
#include "tglobal.h"
H
TD-353  
Hongze Cheng 已提交
22
#include "tkvstore.h"
H
hzcheng 已提交
23
#include "tlist.h"
H
TD-353  
Hongze Cheng 已提交
24
#include "tlog.h"
B
Bomin Zhang 已提交
25
#include "tlockfree.h"
H
hzcheng 已提交
26
#include "tsdb.h"
H
TD-34  
hzcheng 已提交
27 28
#include "tskiplist.h"
#include "tutil.h"
H
more  
Hongze Cheng 已提交
29

H
more  
hzcheng 已提交
30 31 32 33
#ifdef __cplusplus
extern "C" {
#endif

H
hzcheng 已提交
34 35
extern int tsdbDebugFlag;

S
Shengliang Guan 已提交
36 37
#define tsdbFatal(...) { if (tsdbDebugFlag & DEBUG_FATAL) { taosPrintLog("TDB FATAL ", 255, __VA_ARGS__); }}
#define tsdbError(...) { if (tsdbDebugFlag & DEBUG_ERROR) { taosPrintLog("TDB ERROR ", 255, __VA_ARGS__); }}
S
Shengliang Guan 已提交
38 39 40 41
#define tsdbWarn(...)  { if (tsdbDebugFlag & DEBUG_WARN)  { taosPrintLog("TDB WARN ", 255, __VA_ARGS__); }}
#define tsdbInfo(...)  { if (tsdbDebugFlag & DEBUG_INFO)  { taosPrintLog("TDB ", 255, __VA_ARGS__); }}
#define tsdbDebug(...) { if (tsdbDebugFlag & DEBUG_DEBUG) { taosPrintLog("TDB ", tsdbDebugFlag, __VA_ARGS__); }}
#define tsdbTrace(...) { if (tsdbDebugFlag & DEBUG_TRACE) { taosPrintLog("TDB ", tsdbDebugFlag, __VA_ARGS__); }}
H
hzcheng 已提交
42

H
TD-90  
Hongze Cheng 已提交
43
#define TSDB_MAX_TABLE_SCHEMAS 16
H
TD-353  
Hongze Cheng 已提交
44 45
#define TSDB_FILE_HEAD_SIZE 512
#define TSDB_FILE_DELIMITER 0xF00AFA0F
46
#define TSDB_FILE_INIT_MAGIC 0xFFFFFFFF
H
TD-353  
Hongze Cheng 已提交
47

H
Hongze Cheng 已提交
48 49
#define TAOS_IN_RANGE(key, keyMin, keyLast) (((key) >= (keyMin)) && ((key) <= (keyMax)))

50 51 52 53
// NOTE: Any file format change must increase this version number by 1
//       Also, implement the convert function
#define TSDB_FILE_VERSION ((uint32_t)0)

H
TD-353  
Hongze Cheng 已提交
54 55
// Definitions
// ------------------ tsdbMeta.c
H
TD-34  
hzcheng 已提交
56
typedef struct STable {
H
Haojun Liao 已提交
57
  STableId       tableId;
H
TD-353  
Hongze Cheng 已提交
58 59 60 61 62
  ETableType     type;
  tstr*          name;  // NOTE: there a flexible string here
  uint64_t       suid;
  struct STable* pSuper;  // super table pointer
  uint8_t        numOfSchemas;
H
TD-353  
Hongze Cheng 已提交
63
  STSchema*      schema[TSDB_MAX_TABLE_SCHEMAS];
H
TD-353  
Hongze Cheng 已提交
64 65
  STSchema*      tagSchema;
  SKVRow         tagVal;
H
TD-353  
Hongze Cheng 已提交
66
  SSkipList*     pIndex;         // For TSDB_SUPER_TABLE, it is the skiplist index
H
TD-353  
Hongze Cheng 已提交
67 68 69 70 71
  void*          eventHandler;   // TODO
  void*          streamHandler;  // TODO
  TSKEY          lastKey;        // lastkey inserted in this table, initialized as 0, TODO: make a structure
  char*          sql;
  void*          cqhandle;
H
Hongze Cheng 已提交
72
  SRWLatch       latch;  // TODO: implementa latch functions
S
TD-1057  
Shengliang Guan 已提交
73
  T_REF_DECLARE()
H
TD-34  
hzcheng 已提交
74 75 76
} STable;

typedef struct {
H
TD-353  
Hongze Cheng 已提交
77 78 79
  pthread_rwlock_t rwLock;

  int32_t   nTables;
H
TD-987  
Hongze Cheng 已提交
80
  int32_t   maxTables;
H
TD-353  
Hongze Cheng 已提交
81 82 83 84
  STable**  tables;
  SList*    superList;
  SHashObj* uidMap;
  SKVStore* pStore;
H
TD-353  
Hongze Cheng 已提交
85 86
  int       maxRowBytes;
  int       maxCols;
H
TD-34  
hzcheng 已提交
87 88
} STsdbMeta;

H
TD-353  
Hongze Cheng 已提交
89
// ------------------ tsdbBuffer.c
H
TD-34  
hzcheng 已提交
90
typedef struct {
H
TD-353  
Hongze Cheng 已提交
91 92 93 94 95
  int64_t blockId;
  int     offset;
  int     remain;
  char    data[];
} STsdbBufBlock;
H
TD-34  
hzcheng 已提交
96 97

typedef struct {
H
TD-353  
Hongze Cheng 已提交
98 99 100 101 102 103 104 105 106
  pthread_cond_t poolNotEmpty;
  int            bufBlockSize;
  int            tBufBlocks;
  int            nBufBlocks;
  int64_t        index;
  SList*         bufBlockList;
} STsdbBufPool;

// ------------------ tsdbMemTable.c
H
Hongze Cheng 已提交
107 108 109 110 111
typedef struct {
  STable *           pTable;
  SSkipListIterator *pIter;
} SCommitIter;

H
TD-34  
hzcheng 已提交
112
typedef struct {
H
TD-353  
Hongze Cheng 已提交
113 114 115 116 117 118
  uint64_t   uid;
  TSKEY      keyFirst;
  TSKEY      keyLast;
  int64_t    numOfRows;
  SSkipList* pData;
} STableData;
H
TD-34  
hzcheng 已提交
119 120

typedef struct {
S
TD-1057  
Shengliang Guan 已提交
121
  T_REF_DECLARE()
H
TD-987  
Hongze Cheng 已提交
122
  SRWLatch     latch;
H
TD-353  
Hongze Cheng 已提交
123 124 125
  TSKEY        keyFirst;
  TSKEY        keyLast;
  int64_t      numOfRows;
H
TD-987  
Hongze Cheng 已提交
126
  int32_t      maxTables;
H
TD-353  
Hongze Cheng 已提交
127 128
  STableData** tData;
  SList*       actList;
H
Hongze Cheng 已提交
129
  SList*       extraBuffList;
H
TD-353  
Hongze Cheng 已提交
130 131
  SList*       bufBlockList;
} SMemTable;
H
TD-34  
hzcheng 已提交
132

H
TD-353  
Hongze Cheng 已提交
133
enum { TSDB_UPDATE_META, TSDB_DROP_META };
S
TD-1057  
Shengliang Guan 已提交
134 135 136 137 138

#ifdef WINDOWS
#pragma pack(push ,1) 
typedef struct {
#else
H
TD-353  
Hongze Cheng 已提交
139
typedef struct __attribute__((packed)){
S
TD-1057  
Shengliang Guan 已提交
140
#endif
H
TD-353  
Hongze Cheng 已提交
141 142 143
  char     act;
  uint64_t uid;
} SActObj;
S
TD-1057  
Shengliang Guan 已提交
144 145 146
#ifdef WINDOWS
#pragma pack(pop) 
#endif
H
TD-353  
Hongze Cheng 已提交
147 148 149 150 151 152

typedef struct {
  int  len;
  char cont[];
} SActCont;

H
TD-353  
Hongze Cheng 已提交
153
// ------------------ tsdbFile.c
H
TD-353  
Hongze Cheng 已提交
154
extern const char* tsdbFileSuffix[];
H
TD-353  
Hongze Cheng 已提交
155
typedef enum {
H
Hongze Cheng 已提交
156
  TSDB_FILE_TYPE_HEAD = 0,
H
TD-353  
Hongze Cheng 已提交
157 158
  TSDB_FILE_TYPE_DATA,
  TSDB_FILE_TYPE_LAST,
H
Hongze Cheng 已提交
159
  TSDB_FILE_TYPE_STAT,
H
TD-353  
Hongze Cheng 已提交
160
  TSDB_FILE_TYPE_NHEAD,
H
Hongze Cheng 已提交
161 162 163
  TSDB_FILE_TYPE_NDATA,
  TSDB_FILE_TYPE_NLAST,
  TSDB_FILE_TYPE_NSTAT
H
TD-353  
Hongze Cheng 已提交
164
} TSDB_FILE_TYPE;
H
Hongze Cheng 已提交
165

H
Hongze Cheng 已提交
166 167 168 169 170 171
#ifndef TDINTERNAL
#define TSDB_FILE_TYPE_MAX (TSDB_FILE_TYPE_LAST+1)
#else
#define TSDB_FILE_TYPE_MAX (TSDB_FILE_TYPE_STAT+1)
#endif

H
more  
Hongze Cheng 已提交
172
typedef struct {
H
Hongze Cheng 已提交
173
  uint32_t magic;
H
TD-353  
Hongze Cheng 已提交
174 175 176
  uint32_t len;
  uint32_t totalBlocks;
  uint32_t totalSubBlocks;
H
Hongze Cheng 已提交
177
  uint32_t offset;
H
Hongze Cheng 已提交
178 179
  uint64_t size;      // total size of the file
  uint64_t tombSize;  // unused file size
180
} STsdbFileInfo;
H
TD-34  
hzcheng 已提交
181 182

typedef struct {
H
TD-353  
Hongze Cheng 已提交
183
  char  fname[TSDB_FILENAME_LEN];
H
TD-353  
Hongze Cheng 已提交
184 185
  int   fd;

H
TD-353  
Hongze Cheng 已提交
186
  STsdbFileInfo info;
H
hzcheng 已提交
187
} SFile;
H
more  
Hongze Cheng 已提交
188

H
hzcheng 已提交
189
typedef struct {
H
TD-353  
Hongze Cheng 已提交
190
  int   fileId;
H
TD-1102  
Hongze Cheng 已提交
191
  int   state; // 0 for health, 1 for problem
H
TD-353  
Hongze Cheng 已提交
192
  SFile files[TSDB_FILE_TYPE_MAX];
H
hzcheng 已提交
193 194 195
} SFileGroup;

typedef struct {
H
TD-353  
Hongze Cheng 已提交
196 197
  pthread_rwlock_t fhlock;

H
TD-353  
Hongze Cheng 已提交
198 199 200
  int         maxFGroups;
  int         nFGroups;
  SFileGroup* pFGroup;
H
hzcheng 已提交
201
} STsdbFileH;
H
more  
Hongze Cheng 已提交
202

H
TD-34  
hzcheng 已提交
203 204
typedef struct {
  int         direction;
205 206 207
  STsdbFileH* pFileH;
  int         fileId;
  int         index;
H
TD-34  
hzcheng 已提交
208 209
} SFileGroupIter;

H
TD-353  
Hongze Cheng 已提交
210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228
// ------------------ tsdbMain.c
typedef struct {
  int8_t state;

  char*           rootDir;
  STsdbCfg        config;
  STsdbAppH       appH;
  STsdbStat       stat;
  STsdbMeta*      tsdbMeta;
  STsdbBufPool*   pPool;
  SMemTable*      mem;
  SMemTable*      imem;
  STsdbFileH*     tsdbFileH;
  int             commit;
  pthread_t       commitThread;
  pthread_mutex_t mutex;
  bool            repoLocked;
} STsdbRepo;

H
TD-353  
Hongze Cheng 已提交
229
// ------------------ tsdbRWHelper.c
H
TD-34  
hzcheng 已提交
230
typedef struct {
H
Hongze Cheng 已提交
231
  int32_t  tid;
H
TD-185  
Hongze Cheng 已提交
232 233 234 235
  uint32_t len;
  uint32_t offset;
  uint32_t hasLast : 2;
  uint32_t numOfBlocks : 30;
236 237
  uint64_t uid;
  TSKEY    maxKey;
H
TD-353  
Hongze Cheng 已提交
238 239
} SCompIdx;

240
typedef struct {
H
TD-353  
Hongze Cheng 已提交
241 242 243 244 245
  int64_t last : 1;
  int64_t offset : 63;
  int32_t algorithm : 8;
  int32_t numOfRows : 24;
  int32_t len;
H
Hongze Cheng 已提交
246
  int32_t keyLen;     // key column length, keyOffset = offset+sizeof(SCompData)+sizeof(SCompCol)*numOfCols
H
TD-353  
Hongze Cheng 已提交
247
  int16_t numOfSubBlocks;
H
Hongze Cheng 已提交
248
  int16_t numOfCols; // not including timestamp column
249 250 251 252
  TSKEY   keyFirst;
  TSKEY   keyLast;
} SCompBlock;

H
TD-34  
hzcheng 已提交
253 254
typedef struct {
  int32_t    delimiter;  // For recovery usage
255
  int32_t    tid;
256
  uint64_t   uid;
H
TD-34  
hzcheng 已提交
257 258
  SCompBlock blocks[];
} SCompInfo;
H
hzcheng 已提交
259

H
TD-34  
hzcheng 已提交
260
typedef struct {
H
TD-353  
Hongze Cheng 已提交
261
  int16_t colId;
H
TD-541  
Hongze Cheng 已提交
262
  int32_t len;
H
TD-34  
hzcheng 已提交
263 264
  int32_t type : 8;
  int32_t offset : 24;
H
TD-321  
Hongze Cheng 已提交
265 266 267 268 269 270 271
  int64_t sum;
  int64_t max;
  int64_t min;
  int16_t maxIndex;
  int16_t minIndex;
  int16_t numOfNull;
  char    padding[2];
H
TD-34  
hzcheng 已提交
272 273 274 275 276
} SCompCol;

typedef struct {
  int32_t  delimiter;  // For recovery usage
  int32_t  numOfCols;  // For recovery usage
277
  uint64_t uid;        // For recovery usage
H
TD-34  
hzcheng 已提交
278 279
  SCompCol cols[];
} SCompData;
H
TD-34  
hzcheng 已提交
280

H
hzcheng 已提交
281 282 283
typedef enum { TSDB_WRITE_HELPER, TSDB_READ_HELPER } tsdb_rw_helper_t;

typedef struct {
H
Hongze Cheng 已提交
284 285 286 287 288
  TSKEY      minKey;
  TSKEY      maxKey;
  SFileGroup fGroup;
  SFile      nHeadF;
  SFile      nLastF;
H
hzcheng 已提交
289 290 291
} SHelperFile;

typedef struct {
292 293
  uint64_t uid;
  int32_t  tid;
H
hzcheng 已提交
294 295
} SHelperTable;

H
Hongze Cheng 已提交
296 297 298 299 300 301
typedef struct {
  SCompIdx* pIdxArray;
  int       numOfIdx;
  int       curIdx;
} SIdxH;

H
hzcheng 已提交
302
typedef struct {
H
TD-353  
Hongze Cheng 已提交
303
  tsdb_rw_helper_t type;
H
TD-100  
hzcheng 已提交
304

H
TD-353  
Hongze Cheng 已提交
305 306
  STsdbRepo* pRepo;
  int8_t     state;
H
TD-100  
hzcheng 已提交
307
  // For file set usage
H
hzcheng 已提交
308
  SHelperFile files;
H
Hongze Cheng 已提交
309 310 311
  SIdxH       idxH;
  SCompIdx    curCompIdx;
  void*       pWIdx;
H
TD-100  
hzcheng 已提交
312
  // For table set usage
H
hzcheng 已提交
313
  SHelperTable tableInfo;
H
TD-353  
Hongze Cheng 已提交
314
  SCompInfo*   pCompInfo;
H
TD-100  
hzcheng 已提交
315 316
  bool         hasOldLastBlock;
  // For block set usage
H
TD-353  
Hongze Cheng 已提交
317
  SCompData* pCompData;
H
TD-353  
Hongze Cheng 已提交
318 319 320
  SDataCols* pDataCols[2];
  void*      pBuffer;     // Buffer to hold the whole data block
  void*      compBuffer;  // Buffer for temperary compress/decompress purpose
H
hzcheng 已提交
321 322
} SRWHelper;

H
TD-1027  
Hongze Cheng 已提交
323 324 325 326 327 328 329 330 331 332
// ------------------ tsdbScan.c
typedef struct {
  SFileGroup fGroup;
  int        numOfIdx;
  SCompIdx*  pCompIdx;
  SCompInfo* pCompInfo;
  void*      pBuf;
  FILE*      tLogStream;
} STsdbScanHandle;

H
TD-353  
Hongze Cheng 已提交
333 334
// Operations
// ------------------ tsdbMeta.c
H
TD-987  
Hongze Cheng 已提交
335
#define TSDB_INIT_NTABLES 1024
H
TD-353  
Hongze Cheng 已提交
336 337 338
#define TABLE_TYPE(t) (t)->type
#define TABLE_NAME(t) (t)->name
#define TABLE_CHAR_NAME(t) TABLE_NAME(t)->data
H
TD-353  
Hongze Cheng 已提交
339
#define TABLE_UID(t) (t)->tableId.uid
H
TD-353  
Hongze Cheng 已提交
340
#define TABLE_TID(t) (t)->tableId.tid
H
TD-353  
Hongze Cheng 已提交
341
#define TABLE_SUID(t) (t)->suid
H
TD-353  
Hongze Cheng 已提交
342
#define TABLE_LASTKEY(t) (t)->lastKey
343
#define TSDB_META_FILE_MAGIC(m) KVSTORE_MAGIC((m)->pStore)
H
TD-353  
Hongze Cheng 已提交
344 345 346

STsdbMeta* tsdbNewMeta(STsdbCfg* pCfg);
void       tsdbFreeMeta(STsdbMeta* pMeta);
H
TD-353  
Hongze Cheng 已提交
347 348 349 350 351 352 353
int        tsdbOpenMeta(STsdbRepo* pRepo);
int        tsdbCloseMeta(STsdbRepo* pRepo);
STable*    tsdbGetTableByUid(STsdbMeta* pMeta, uint64_t uid);
STSchema*  tsdbGetTableSchemaByVersion(STable* pTable, int16_t version);
int        tsdbWLockRepoMeta(STsdbRepo* pRepo);
int        tsdbRLockRepoMeta(STsdbRepo* pRepo);
int        tsdbUnlockRepoMeta(STsdbRepo* pRepo);
H
TD-353  
Hongze Cheng 已提交
354 355
void       tsdbRefTable(STable* pTable);
void       tsdbUnRefTable(STable* pTable);
H
Hongze Cheng 已提交
356 357 358 359 360 361 362 363 364 365 366 367 368 369 370 371 372 373 374 375 376 377 378 379 380 381 382 383 384 385 386 387 388 389 390 391 392 393 394 395 396 397
void       tsdbUpdateTableSchema(STsdbRepo* pRepo, STable* pTable, STSchema* pSchema, bool insertAct);

static FORCE_INLINE int tsdbCompareSchemaVersion(const void *key1, const void *key2) {
  if (*(int16_t *)key1 < schemaVersion(*(STSchema **)key2)) {
    return -1;
  } else if (*(int16_t *)key1 > schemaVersion(*(STSchema **)key2)) {
    return 1;
  } else {
    return 0;
  }
}

static FORCE_INLINE STSchema* tsdbGetTableSchemaImpl(STable* pTable, bool lock, bool copy, int16_t version) {
  STable*   pDTable = (TABLE_TYPE(pTable) == TSDB_CHILD_TABLE) ? pTable->pSuper : pTable;
  STSchema* pSchema = NULL;
  STSchema* pTSchema = NULL;

  if (lock) taosRLockLatch(&(pDTable->latch));
  if (version < 0) {  // get the latest version of schema
    pTSchema = pDTable->schema[pDTable->numOfSchemas - 1];
  } else {  // get the schema with version
    void* ptr = taosbsearch(&version, pDTable->schema, pDTable->numOfSchemas, sizeof(STSchema*),
                            tsdbCompareSchemaVersion, TD_EQ);
    if (ptr == NULL) {
      terrno = TSDB_CODE_TDB_IVD_TB_SCHEMA_VERSION;
      goto _exit;
    }
    pTSchema = *(STSchema**)ptr;
  }

  ASSERT(pTSchema != NULL);

  if (copy) {
    if ((pSchema = tdDupSchema(pTSchema)) == NULL) terrno = TSDB_CODE_TDB_OUT_OF_MEMORY;
  } else {
    pSchema = pTSchema;
  }

_exit:
  if (lock) taosRUnLockLatch(&(pDTable->latch));
  return pSchema;
}
H
TD-353  
Hongze Cheng 已提交
398

H
Hongze Cheng 已提交
399 400 401 402 403 404 405 406 407 408 409 410 411 412 413 414
static FORCE_INLINE STSchema* tsdbGetTableSchema(STable* pTable) {
  return tsdbGetTableSchemaImpl(pTable, false, false, -1);
}

static FORCE_INLINE STSchema *tsdbGetTableTagSchema(STable *pTable) {
  if (pTable->type == TSDB_CHILD_TABLE) {  // check child table first
    STable *pSuper = pTable->pSuper;
    if (pSuper == NULL) return NULL;
    return pSuper->tagSchema;
  } else if (pTable->type == TSDB_SUPER_TABLE) {
    return pTable->tagSchema;
  } else {
    return NULL;
  }
}

H
TD-353  
Hongze Cheng 已提交
415
// ------------------ tsdbBuffer.c
H
Hongze Cheng 已提交
416 417
#define TSDB_BUFFER_RESERVE 1024  // Reseve 1K as commit threshold

H
TD-353  
Hongze Cheng 已提交
418 419 420
STsdbBufPool* tsdbNewBufPool();
void          tsdbFreeBufPool(STsdbBufPool* pBufPool);
int           tsdbOpenBufPool(STsdbRepo* pRepo);
H
TD-353  
Hongze Cheng 已提交
421
void          tsdbCloseBufPool(STsdbRepo* pRepo);
H
TD-353  
Hongze Cheng 已提交
422 423 424
SListNode*    tsdbAllocBufBlockFromPool(STsdbRepo* pRepo);

// ------------------ tsdbMemTable.c
H
TD-353  
Hongze Cheng 已提交
425 426 427 428
int   tsdbInsertRowToMem(STsdbRepo* pRepo, SDataRow row, STable* pTable);
int   tsdbRefMemTable(STsdbRepo* pRepo, SMemTable* pMemTable);
int   tsdbUnRefMemTable(STsdbRepo* pRepo, SMemTable* pMemTable);
int   tsdbTakeMemSnapshot(STsdbRepo* pRepo, SMemTable** pMem, SMemTable** pIMem);
H
TD-987  
Hongze Cheng 已提交
429
void  tsdbUnTakeMemSnapShot(STsdbRepo* pRepo, SMemTable* pMem, SMemTable* pIMem);
H
TD-353  
Hongze Cheng 已提交
430
void* tsdbAllocBytes(STsdbRepo* pRepo, int bytes);
H
TD-353  
Hongze Cheng 已提交
431
int   tsdbAsyncCommit(STsdbRepo* pRepo);
H
Hongze Cheng 已提交
432
int   tsdbLoadDataFromCache(STable* pTable, SSkipListIterator* pIter, TSKEY maxKey, int maxRowsToRead, SDataCols* pCols,
H
TD-1438  
Hongze Cheng 已提交
433
                            TSKEY* filterKeys, int nFilterKeys, bool keepDup);
H
Hongze Cheng 已提交
434 435 436 437 438 439 440

static FORCE_INLINE SDataRow tsdbNextIterRow(SSkipListIterator* pIter) {
  if (pIter == NULL) return NULL;

  SSkipListNode* node = tSkipListIterGet(pIter);
  if (node == NULL) return NULL;

H
Hongze Cheng 已提交
441
  return *(SDataRow *)SL_GET_NODE_DATA(node);
H
Hongze Cheng 已提交
442 443 444 445 446 447 448 449
}

static FORCE_INLINE TSKEY tsdbNextIterKey(SSkipListIterator* pIter) {
  SDataRow row = tsdbNextIterRow(pIter);
  if (row == NULL) return -1;

  return dataRowKey(row);
}
H
TD-353  
Hongze Cheng 已提交
450

H
Hongze Cheng 已提交
451 452 453 454 455 456 457 458 459 460 461 462 463
static FORCE_INLINE STsdbBufBlock* tsdbGetCurrBufBlock(STsdbRepo* pRepo) {
  ASSERT(pRepo != NULL);
  if (pRepo->mem == NULL) return NULL;

  SListNode* pNode = listTail(pRepo->mem->bufBlockList);
  if (pNode == NULL) return NULL;

  STsdbBufBlock* pBufBlock = NULL;
  tdListNodeGetData(pRepo->mem->bufBlockList, pNode, (void*)(&pBufBlock));

  return pBufBlock;
}

H
TD-353  
Hongze Cheng 已提交
464 465 466 467 468
// ------------------ tsdbFile.c
#define TSDB_KEY_FILEID(key, daysPerFile, precision) ((key) / tsMsPerDay[(precision)] / (daysPerFile))
#define TSDB_MAX_FILE(keep, daysPerFile) ((keep) / (daysPerFile) + 3)
#define TSDB_MIN_FILE_ID(fh) (fh)->pFGroup[0].fileId
#define TSDB_MAX_FILE_ID(fh) (fh)->pFGroup[(fh)->nFGroups - 1].fileId
H
TD-353  
Hongze Cheng 已提交
469
#define TSDB_IS_FILE_OPENED(f) ((f)->fd > 0)
H
TD-353  
Hongze Cheng 已提交
470 471 472
#define TSDB_FGROUP_ITER_FORWARD TSDB_ORDER_ASC
#define TSDB_FGROUP_ITER_BACKWARD TSDB_ORDER_DESC

H
TD-353  
Hongze Cheng 已提交
473 474
STsdbFileH* tsdbNewFileH(STsdbCfg* pCfg);
void        tsdbFreeFileH(STsdbFileH* pFileH);
H
TD-353  
Hongze Cheng 已提交
475
int         tsdbOpenFileH(STsdbRepo* pRepo);
H
TD-353  
Hongze Cheng 已提交
476
void        tsdbCloseFileH(STsdbRepo* pRepo);
H
TD-987  
Hongze Cheng 已提交
477
SFileGroup* tsdbCreateFGroupIfNeed(STsdbRepo* pRepo, char* dataDir, int fid);
H
TD-353  
Hongze Cheng 已提交
478 479 480 481 482 483 484 485
void        tsdbInitFileGroupIter(STsdbFileH* pFileH, SFileGroupIter* pIter, int direction);
void        tsdbSeekFileGroupIter(SFileGroupIter* pIter, int fid);
SFileGroup* tsdbGetFileGroupNext(SFileGroupIter* pIter);
int         tsdbOpenFile(SFile* pFile, int oflag);
void        tsdbCloseFile(SFile* pFile);
int         tsdbCreateFile(SFile* pFile, STsdbRepo* pRepo, int fid, int type);
SFileGroup* tsdbSearchFGroup(STsdbFileH* pFileH, int fid, int flags);
void        tsdbFitRetention(STsdbRepo* pRepo);
486
int         tsdbUpdateFileHeader(SFile* pFile);
H
TD-353  
Hongze Cheng 已提交
487
int         tsdbEncodeSFileInfo(void** buf, const STsdbFileInfo* pInfo);
H
TD-353  
Hongze Cheng 已提交
488
void*       tsdbDecodeSFileInfo(void* buf, STsdbFileInfo* pInfo);
H
TD-353  
Hongze Cheng 已提交
489
void        tsdbRemoveFileGroup(STsdbRepo* pRepo, SFileGroup* pFGroup);
H
Hongze Cheng 已提交
490
int         tsdbLoadFileHeader(SFile* pFile, uint32_t* version);
491
void        tsdbGetFileInfoImpl(char* fname, uint32_t* magic, int64_t* size);
492
void        tsdbGetFidKeyRange(int daysPerFile, int8_t precision, int fileId, TSKEY *minKey, TSKEY *maxKey);
H
TD-353  
Hongze Cheng 已提交
493 494

// ------------------ tsdbRWHelper.c
H
TD-353  
Hongze Cheng 已提交
495 496 497 498 499 500
#define TSDB_HELPER_CLEAR_STATE 0x0        // Clear state
#define TSDB_HELPER_FILE_SET_AND_OPEN 0x1  // File is set
#define TSDB_HELPER_IDX_LOAD 0x2           // SCompIdx part is loaded
#define TSDB_HELPER_TABLE_SET 0x4          // Table is set
#define TSDB_HELPER_INFO_LOAD 0x8          // SCompInfo part is loaded
#define TSDB_HELPER_FILE_DATA_LOAD 0x10    // SCompData part is loaded
H
TD-353  
Hongze Cheng 已提交
501 502 503 504
#define helperSetState(h, s) (((h)->state) |= (s))
#define helperClearState(h, s) ((h)->state &= (~(s)))
#define helperHasState(h, s) ((((h)->state) & (s)) == (s))
#define blockAtIdx(h, idx) ((h)->pCompInfo->blocks + idx)
H
TD-353  
Hongze Cheng 已提交
505 506
#define TSDB_MAX_SUBBLOCKS 8
#define IS_SUB_BLOCK(pBlock) ((pBlock)->numOfSubBlocks == 0)
H
TD-353  
Hongze Cheng 已提交
507 508 509
#define helperType(h) (h)->type
#define helperRepo(h) (h)->pRepo
#define helperState(h) (h)->state
H
Hongze Cheng 已提交
510
#define TSDB_NLAST_FILE_OPENED(h) ((h)->files.nLastF.fd > 0)
H
Hongze Cheng 已提交
511 512 513 514 515 516
#define helperFileId(h) ((h)->files.fGroup.fileId)
#define helperHeadF(h) (&((h)->files.fGroup.files[TSDB_FILE_TYPE_HEAD]))
#define helperDataF(h) (&((h)->files.fGroup.files[TSDB_FILE_TYPE_DATA]))
#define helperLastF(h) (&((h)->files.fGroup.files[TSDB_FILE_TYPE_LAST]))
#define helperNewHeadF(h) (&((h)->files.nHeadF))
#define helperNewLastF(h) (&((h)->files.nLastF))
H
Hongze Cheng 已提交
517 518 519 520 521 522

int  tsdbInitReadHelper(SRWHelper* pHelper, STsdbRepo* pRepo);
int  tsdbInitWriteHelper(SRWHelper* pHelper, STsdbRepo* pRepo);
void tsdbDestroyHelper(SRWHelper* pHelper);
void tsdbResetHelper(SRWHelper* pHelper);
int  tsdbSetAndOpenHelperFile(SRWHelper* pHelper, SFileGroup* pGroup);
H
Hongze Cheng 已提交
523
int  tsdbCloseHelperFile(SRWHelper* pHelper, bool hasError, SFileGroup* pGroup);
H
Hongze Cheng 已提交
524
int  tsdbSetHelperTable(SRWHelper* pHelper, STable* pTable, STsdbRepo* pRepo);
H
Hongze Cheng 已提交
525 526 527 528
int  tsdbCommitTableData(SRWHelper* pHelper, SCommitIter* pCommitIter, SDataCols* pDataCols, TSKEY maxKey);
int  tsdbMoveLastBlockIfNeccessary(SRWHelper* pHelper);
int  tsdbWriteCompInfo(SRWHelper* pHelper);
int  tsdbWriteCompIdx(SRWHelper* pHelper);
H
Hongze Cheng 已提交
529 530
int  tsdbLoadCompIdxImpl(SFile* pFile, uint32_t offset, uint32_t len, void* buffer);
int  tsdbDecodeSCompIdxImpl(void* buffer, uint32_t len, SCompIdx** ppCompIdx, int* numOfIdx);
H
Hongze Cheng 已提交
531
int  tsdbLoadCompIdx(SRWHelper* pHelper, void* target);
H
Hongze Cheng 已提交
532
int  tsdbLoadCompInfoImpl(SFile* pFile, SCompIdx* pIdx, SCompInfo** ppCompInfo);
H
Hongze Cheng 已提交
533 534 535 536 537 538 539 540 541 542 543 544 545 546 547 548
int  tsdbLoadCompInfo(SRWHelper* pHelper, void* target);
int  tsdbLoadCompData(SRWHelper* phelper, SCompBlock* pcompblock, void* target);
void tsdbGetDataStatis(SRWHelper* pHelper, SDataStatis* pStatis, int numOfCols);
int  tsdbLoadBlockDataCols(SRWHelper* pHelper, SCompBlock* pCompBlock, SCompInfo* pCompInfo, int16_t* colIds,
                           int numOfColIds);
int  tsdbLoadBlockData(SRWHelper* pHelper, SCompBlock* pCompBlock, SCompInfo* pCompInfo);

static FORCE_INLINE int compTSKEY(const void* key1, const void* key2) {
  if (*(TSKEY*)key1 > *(TSKEY*)key2) {
    return 1;
  } else if (*(TSKEY*)key1 == *(TSKEY*)key2) {
    return 0;
  } else {
    return -1;
  }
}
H
TD-353  
Hongze Cheng 已提交
549

H
TD-353  
Hongze Cheng 已提交
550 551
// ------------------ tsdbMain.c
#define REPO_ID(r) (r)->config.tsdbId
H
TD-353  
Hongze Cheng 已提交
552
#define IS_REPO_LOCKED(r) (r)->repoLocked
H
TD-353  
Hongze Cheng 已提交
553
#define TSDB_SUBMIT_MSG_HEAD_SIZE sizeof(SSubmitMsg)
H
TD-353  
Hongze Cheng 已提交
554

H
TD-353  
Hongze Cheng 已提交
555
char*       tsdbGetMetaFileName(char* rootDir);
H
Hongze Cheng 已提交
556
void        tsdbGetDataFileName(char* rootDir, int vid, int fid, int type, char* fname);
H
TD-353  
Hongze Cheng 已提交
557 558 559
int         tsdbLockRepo(STsdbRepo* pRepo);
int         tsdbUnlockRepo(STsdbRepo* pRepo);
char*       tsdbGetDataDirName(char* rootDir);
H
TD-987  
Hongze Cheng 已提交
560
int         tsdbGetNextMaxTables(int tid);
H
TD-353  
Hongze Cheng 已提交
561 562
STsdbMeta*  tsdbGetMeta(TSDB_REPO_T* pRepo);
STsdbFileH* tsdbGetFile(TSDB_REPO_T* pRepo);
H
Hongze Cheng 已提交
563
int         tsdbCheckCommit(STsdbRepo* pRepo);
H
TD-353  
Hongze Cheng 已提交
564

H
TD-1027  
Hongze Cheng 已提交
565 566 567 568 569 570 571 572 573 574
// ------------------ tsdbScan.c
int              tsdbScanFGroup(STsdbScanHandle* pScanHandle, char* rootDir, int fid);
STsdbScanHandle* tsdbNewScanHandle();
void             tsdbSetScanLogStream(STsdbScanHandle* pScanHandle, FILE* fLogStream);
int              tsdbSetAndOpenScanFile(STsdbScanHandle* pScanHandle, char* rootDir, int fid);
int              tsdbScanSCompIdx(STsdbScanHandle* pScanHandle);
int              tsdbScanSCompBlock(STsdbScanHandle* pScanHandle, int idx);
int              tsdbCloseScanFile(STsdbScanHandle* pScanHandle);
void             tsdbFreeScanHandle(STsdbScanHandle* pScanHandle);

H
more  
hzcheng 已提交
575 576 577 578
#ifdef __cplusplus
}
#endif

H
TD-34  
hzcheng 已提交
579
#endif