metaBDBImpl.c 16.3 KB
Newer Older
H
Hongze Cheng 已提交
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
more  
Hongze Cheng 已提交
16 17
#include "db.h"

H
Hongze Cheng 已提交
18 19
#include "metaDef.h"

H
Hongze Cheng 已提交
20
#include "tcoding.h"
H
more  
Hongze Cheng 已提交
21
#include "thash.h"
H
Hongze Cheng 已提交
22

H
Hongze Cheng 已提交
23 24 25
typedef struct {
  tb_uid_t uid;
  int32_t  sver;
H
Hongze Cheng 已提交
26
  int32_t  padding;
H
Hongze Cheng 已提交
27 28
} SSchemaKey;

H
Hongze Cheng 已提交
29
struct SMetaDB {
H
more  
Hongze Cheng 已提交
30
  // DB
H
more  
Hongze Cheng 已提交
31 32
  DB *pTbDB;
  DB *pSchemaDB;
H
more  
Hongze Cheng 已提交
33
  // IDX
H
more  
Hongze Cheng 已提交
34 35 36 37
  DB *pNameIdx;
  DB *pStbIdx;
  DB *pNtbIdx;
  DB *pCtbIdx;
H
more  
Hongze Cheng 已提交
38
  // ENV
H
Hongze Cheng 已提交
39 40 41
  DB_ENV *pEvn;
};

H
Hongze Cheng 已提交
42 43
typedef int (*bdbIdxCbPtr)(DB *, const DBT *, const DBT *, DBT *);

H
more  
Hongze Cheng 已提交
44 45
static SMetaDB *metaNewDB();
static void     metaFreeDB(SMetaDB *pDB);
H
more  
Hongze Cheng 已提交
46 47
static int      metaOpenBDBEnv(DB_ENV **ppEnv, const char *path);
static void     metaCloseBDBEnv(DB_ENV *pEnv);
H
Hongze Cheng 已提交
48
static int      metaOpenBDBDb(DB **ppDB, DB_ENV *pEnv, const char *pFName, bool isDup);
H
more  
Hongze Cheng 已提交
49
static void     metaCloseBDBDb(DB *pDB);
H
Hongze Cheng 已提交
50
static int      metaOpenBDBIdx(DB **ppIdx, DB_ENV *pEnv, const char *pFName, DB *pDB, bdbIdxCbPtr cbf, bool isDup);
H
Hongze Cheng 已提交
51
static void     metaCloseBDBIdx(DB *pIdx);
H
Hongze Cheng 已提交
52 53 54 55
static int      metaNameIdxCb(DB *pIdx, const DBT *pKey, const DBT *pValue, DBT *pSKey);
static int      metaStbIdxCb(DB *pIdx, const DBT *pKey, const DBT *pValue, DBT *pSKey);
static int      metaNtbIdxCb(DB *pIdx, const DBT *pKey, const DBT *pValue, DBT *pSKey);
static int      metaCtbIdxCb(DB *pIdx, const DBT *pKey, const DBT *pValue, DBT *pSKey);
H
Hongze Cheng 已提交
56 57 58
static int      metaEncodeTbInfo(void **buf, STbCfg *pTbCfg);
static void *   metaDecodeTbInfo(void *buf, STbCfg *pTbCfg);
static void     metaClearTbCfg(STbCfg *pTbCfg);
H
Hongze Cheng 已提交
59 60
static int      metaEncodeSchema(void **buf, SSchemaWrapper *pSW);
static void *   metaDecodeSchema(void *buf, SSchemaWrapper *pSW);
H
more  
Hongze Cheng 已提交
61

H
more  
Hongze Cheng 已提交
62
#define BDB_PERR(info, code) fprintf(stderr, info " reason: %s", db_strerror(code))
H
Hongze Cheng 已提交
63

H
Hongze Cheng 已提交
64
int metaOpenDB(SMeta *pMeta) {
H
more  
Hongze Cheng 已提交
65
  SMetaDB *pDB;
H
Hongze Cheng 已提交
66

H
more  
Hongze Cheng 已提交
67 68 69
  // Create DB object
  pDB = metaNewDB();
  if (pDB == NULL) {
H
more  
Hongze Cheng 已提交
70 71 72
    return -1;
  }

H
more  
Hongze Cheng 已提交
73
  pMeta->pDB = pDB;
H
more  
Hongze Cheng 已提交
74

H
more  
Hongze Cheng 已提交
75 76
  // Open DB Env
  if (metaOpenBDBEnv(&(pDB->pEvn), pMeta->path) < 0) {
H
more  
Hongze Cheng 已提交
77
    metaCloseDB(pMeta);
H
more  
Hongze Cheng 已提交
78 79
    return -1;
  }
H
Hongze Cheng 已提交
80

H
more  
Hongze Cheng 已提交
81
  // Open DBs
H
Hongze Cheng 已提交
82
  if (metaOpenBDBDb(&(pDB->pTbDB), pDB->pEvn, "meta.db", false) < 0) {
H
more  
Hongze Cheng 已提交
83 84
    metaCloseDB(pMeta);
    return -1;
H
more  
Hongze Cheng 已提交
85 86
  }

H
more  
Hongze Cheng 已提交
87
  if (metaOpenBDBDb(&(pDB->pSchemaDB), pDB->pEvn, "schema.db", false) < 0) {
H
more  
Hongze Cheng 已提交
88 89 90
    metaCloseDB(pMeta);
    return -1;
  }
H
Hongze Cheng 已提交
91

H
more  
Hongze Cheng 已提交
92
  // Open Indices
H
Hongze Cheng 已提交
93
  if (metaOpenBDBIdx(&(pDB->pNameIdx), pDB->pEvn, "name.index", pDB->pTbDB, &metaNameIdxCb, false) < 0) {
H
more  
Hongze Cheng 已提交
94 95 96
    metaCloseDB(pMeta);
    return -1;
  }
H
Hongze Cheng 已提交
97

H
Hongze Cheng 已提交
98
  if (metaOpenBDBIdx(&(pDB->pStbIdx), pDB->pEvn, "stb.index", pDB->pTbDB, &metaStbIdxCb, false) < 0) {
H
more  
Hongze Cheng 已提交
99 100 101
    metaCloseDB(pMeta);
    return -1;
  }
H
Hongze Cheng 已提交
102

H
Hongze Cheng 已提交
103
  if (metaOpenBDBIdx(&(pDB->pNtbIdx), pDB->pEvn, "ntb.index", pDB->pTbDB, &metaNtbIdxCb, false) < 0) {
H
more  
Hongze Cheng 已提交
104 105
    metaCloseDB(pMeta);
    return -1;
H
Hongze Cheng 已提交
106 107
  }

H
Hongze Cheng 已提交
108
  if (metaOpenBDBIdx(&(pDB->pCtbIdx), pDB->pEvn, "ctb.index", pDB->pTbDB, &metaCtbIdxCb, true) < 0) {
H
more  
Hongze Cheng 已提交
109 110
    metaCloseDB(pMeta);
    return -1;
H
more  
Hongze Cheng 已提交
111 112
  }

H
Hongze Cheng 已提交
113 114 115 116 117
  return 0;
}

void metaCloseDB(SMeta *pMeta) {
  if (pMeta->pDB) {
H
Hongze Cheng 已提交
118 119 120 121 122 123
    metaCloseBDBIdx(pMeta->pDB->pCtbIdx);
    metaCloseBDBIdx(pMeta->pDB->pNtbIdx);
    metaCloseBDBIdx(pMeta->pDB->pStbIdx);
    metaCloseBDBIdx(pMeta->pDB->pNameIdx);
    metaCloseBDBDb(pMeta->pDB->pSchemaDB);
    metaCloseBDBDb(pMeta->pDB->pTbDB);
H
more  
Hongze Cheng 已提交
124
    metaCloseBDBEnv(pMeta->pDB->pEvn);
H
more  
Hongze Cheng 已提交
125 126
    metaFreeDB(pMeta->pDB);
    pMeta->pDB = NULL;
H
Hongze Cheng 已提交
127 128 129
  }
}

H
more  
Hongze Cheng 已提交
130
int metaSaveTableToDB(SMeta *pMeta, STbCfg *pTbCfg) {
H
more  
Hongze Cheng 已提交
131 132 133 134 135
  tb_uid_t uid;
  char     buf[512];
  void *   pBuf;
  DBT      key, value;
  SSchema *pSchema = NULL;
H
Hongze Cheng 已提交
136 137 138 139 140 141 142 143 144

  if (pTbCfg->type == META_SUPER_TABLE) {
    uid = pTbCfg->stbCfg.suid;
  } else {
    uid = metaGenerateUid(pMeta);
  }

  {
    // save table info
H
Hongze Cheng 已提交
145 146 147
    pBuf = buf;
    memset(&key, 0, sizeof(key));
    memset(&value, 0, sizeof(key));
H
Hongze Cheng 已提交
148 149 150 151

    key.data = &uid;
    key.size = sizeof(uid);

H
Hongze Cheng 已提交
152
    metaEncodeTbInfo(&pBuf, pTbCfg);
H
Hongze Cheng 已提交
153 154 155

    value.data = buf;
    value.size = POINTER_DISTANCE(pBuf, buf);
H
Hongze Cheng 已提交
156
    value.app_data = pTbCfg;
H
Hongze Cheng 已提交
157 158 159 160 161

    pMeta->pDB->pTbDB->put(pMeta->pDB->pTbDB, NULL, &key, &value, 0);
  }

  // save schema
H
more  
Hongze Cheng 已提交
162
  uint32_t ncols;
H
Hongze Cheng 已提交
163
  if (pTbCfg->type == META_SUPER_TABLE) {
H
more  
Hongze Cheng 已提交
164
    ncols = pTbCfg->stbCfg.nCols;
H
Hongze Cheng 已提交
165 166
    pSchema = pTbCfg->stbCfg.pSchema;
  } else if (pTbCfg->type == META_NORMAL_TABLE) {
H
more  
Hongze Cheng 已提交
167
    ncols = pTbCfg->ntbCfg.nCols;
H
Hongze Cheng 已提交
168 169 170 171
    pSchema = pTbCfg->ntbCfg.pSchema;
  }

  if (pSchema) {
H
Hongze Cheng 已提交
172 173 174
    pBuf = buf;
    memset(&key, 0, sizeof(key));
    memset(&value, 0, sizeof(key));
H
Hongze Cheng 已提交
175
    SSchemaKey schemaKey = {uid, 0 /*TODO*/, 0};
H
Hongze Cheng 已提交
176

H
Hongze Cheng 已提交
177 178
    key.data = &schemaKey;
    key.size = sizeof(schemaKey);
H
Hongze Cheng 已提交
179

H
Hongze Cheng 已提交
180 181
    SSchemaWrapper sw = {.nCols = ncols, .pSchema = pSchema};
    metaEncodeSchema(&pBuf, &sw);
H
Hongze Cheng 已提交
182 183 184 185 186 187 188

    value.data = buf;
    value.size = POINTER_DISTANCE(pBuf, buf);

    pMeta->pDB->pSchemaDB->put(pMeta->pDB->pSchemaDB, NULL, &key, &value, 0);
  }

H
more  
Hongze Cheng 已提交
189 190 191 192 193 194 195 196 197
  return 0;
}

int metaRemoveTableFromDb(SMeta *pMeta, tb_uid_t uid) {
  // TODO
  return 0;
}

/* ------------------------ STATIC METHODS ------------------------ */
H
Hongze Cheng 已提交
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 229
static int metaEncodeSchema(void **buf, SSchemaWrapper *pSW) {
  int      tlen = 0;
  SSchema *pSchema;

  tlen += taosEncodeFixedU32(buf, pSW->nCols);
  for (int i = 0; i < pSW->nCols; i++) {
    pSchema = pSW->pSchema + i;
    tlen += taosEncodeFixedI8(buf, pSchema->type);
    tlen += taosEncodeFixedI32(buf, pSchema->colId);
    tlen += taosEncodeFixedI32(buf, pSchema->bytes);
    tlen += taosEncodeString(buf, pSchema->name);
  }

  return tlen;
}

static void *metaDecodeSchema(void *buf, SSchemaWrapper *pSW) {
  SSchema *pSchema;

  buf = taosDecodeFixedU32(buf, &pSW->nCols);
  pSW->pSchema = (SSchema *)malloc(sizeof(SSchema) * pSW->nCols);
  for (int i = 0; i < pSW->nCols; i++) {
    pSchema = pSW->pSchema + i;
    buf = taosDecodeFixedI8(buf, &pSchema->type);
    buf = taosDecodeFixedI32(buf, &pSchema->colId);
    buf = taosDecodeFixedI32(buf, &pSchema->bytes);
    buf = taosDecodeStringTo(buf, pSchema->name);
  }

  return buf;
}

H
more  
Hongze Cheng 已提交
230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257
static SMetaDB *metaNewDB() {
  SMetaDB *pDB = NULL;
  pDB = (SMetaDB *)calloc(1, sizeof(*pDB));
  if (pDB == NULL) {
    return NULL;
  }

  return pDB;
}

static void metaFreeDB(SMetaDB *pDB) {
  if (pDB) {
    free(pDB);
  }
}

static int metaOpenBDBEnv(DB_ENV **ppEnv, const char *path) {
  int     ret;
  DB_ENV *pEnv;

  if (path == NULL) return 0;

  ret = db_env_create(&pEnv, 0);
  if (ret != 0) {
    BDB_PERR("Failed to create META env", ret);
    return -1;
  }

H
Hongze Cheng 已提交
258
  ret = pEnv->open(pEnv, path, DB_CREATE | DB_INIT_CDB | DB_INIT_MPOOL, 0);
H
more  
Hongze Cheng 已提交
259 260 261 262 263 264 265 266 267 268 269 270 271 272 273 274
  if (ret != 0) {
    BDB_PERR("Failed to open META env", ret);
    return -1;
  }

  *ppEnv = pEnv;

  return 0;
}

static void metaCloseBDBEnv(DB_ENV *pEnv) {
  if (pEnv) {
    pEnv->close(pEnv, 0);
  }
}

H
Hongze Cheng 已提交
275
static int metaOpenBDBDb(DB **ppDB, DB_ENV *pEnv, const char *pFName, bool isDup) {
H
more  
Hongze Cheng 已提交
276 277 278
  int ret;
  DB *pDB;

H
Hongze Cheng 已提交
279
  ret = db_create(&(pDB), pEnv, 0);
H
more  
Hongze Cheng 已提交
280 281 282 283 284
  if (ret != 0) {
    BDB_PERR("Failed to create META DB", ret);
    return -1;
  }

H
Hongze Cheng 已提交
285 286 287 288 289 290 291 292
  if (isDup) {
    ret = pDB->set_flags(pDB, DB_DUPSORT);
    if (ret != 0) {
      BDB_PERR("Failed to set DB flags", ret);
      return -1;
    }
  }

H
more  
Hongze Cheng 已提交
293 294 295 296 297 298
  ret = pDB->open(pDB, NULL, pFName, NULL, DB_BTREE, DB_CREATE, 0);
  if (ret) {
    BDB_PERR("Failed to open META DB", ret);
    return -1;
  }

H
Hongze Cheng 已提交
299 300
  *ppDB = pDB;

H
more  
Hongze Cheng 已提交
301 302 303 304 305 306 307 308 309
  return 0;
}

static void metaCloseBDBDb(DB *pDB) {
  if (pDB) {
    pDB->close(pDB, 0);
  }
}

H
Hongze Cheng 已提交
310
static int metaOpenBDBIdx(DB **ppIdx, DB_ENV *pEnv, const char *pFName, DB *pDB, bdbIdxCbPtr cbf, bool isDup) {
H
Hongze Cheng 已提交
311 312 313
  DB *pIdx;
  int ret;

H
Hongze Cheng 已提交
314
  if (metaOpenBDBDb(ppIdx, pEnv, pFName, isDup) < 0) {
H
Hongze Cheng 已提交
315 316 317 318 319 320 321 322 323 324 325 326 327 328 329 330 331 332 333
    return -1;
  }

  pIdx = *ppIdx;
  ret = pDB->associate(pDB, NULL, pIdx, cbf, 0);
  if (ret) {
    BDB_PERR("Failed to associate META DB and Index", ret);
  }

  return 0;
}

static void metaCloseBDBIdx(DB *pIdx) {
  if (pIdx) {
    pIdx->close(pIdx, 0);
  }
}

static int metaNameIdxCb(DB *pIdx, const DBT *pKey, const DBT *pValue, DBT *pSKey) {
H
Hongze Cheng 已提交
334
  STbCfg *pTbCfg = (STbCfg *)(pValue->app_data);
H
Hongze Cheng 已提交
335

H
Hongze Cheng 已提交
336
  memset(pSKey, 0, sizeof(*pSKey));
H
Hongze Cheng 已提交
337

H
Hongze Cheng 已提交
338 339
  pSKey->data = pTbCfg->name;
  pSKey->size = strlen(pTbCfg->name);
H
Hongze Cheng 已提交
340 341 342 343

  return 0;
}

H
Hongze Cheng 已提交
344
static int metaStbIdxCb(DB *pIdx, const DBT *pKey, const DBT *pValue, DBT *pSKey) {
H
Hongze Cheng 已提交
345
  STbCfg *pTbCfg = (STbCfg *)(pValue->app_data);
H
Hongze Cheng 已提交
346

H
Hongze Cheng 已提交
347
  if (pTbCfg->type == META_SUPER_TABLE) {
H
Hongze Cheng 已提交
348 349 350
    memset(pSKey, 0, sizeof(*pSKey));
    pSKey->data = pKey->data;
    pSKey->size = pKey->size;
H
Hongze Cheng 已提交
351

H
Hongze Cheng 已提交
352
    return 0;
H
Hongze Cheng 已提交
353
  } else {
H
Hongze Cheng 已提交
354
    return DB_DONOTINDEX;
H
Hongze Cheng 已提交
355
  }
H
Hongze Cheng 已提交
356 357
}

H
Hongze Cheng 已提交
358
static int metaNtbIdxCb(DB *pIdx, const DBT *pKey, const DBT *pValue, DBT *pSKey) {
H
Hongze Cheng 已提交
359
  STbCfg *pTbCfg = (STbCfg *)(pValue->app_data);
H
Hongze Cheng 已提交
360

H
Hongze Cheng 已提交
361
  if (pTbCfg->type == META_NORMAL_TABLE) {
H
Hongze Cheng 已提交
362 363 364
    memset(pSKey, 0, sizeof(*pSKey));
    pSKey->data = pKey->data;
    pSKey->size = pKey->size;
H
Hongze Cheng 已提交
365

H
Hongze Cheng 已提交
366 367 368 369
    return 0;
  } else {
    return DB_DONOTINDEX;
  }
H
Hongze Cheng 已提交
370 371
}

H
Hongze Cheng 已提交
372
static int metaCtbIdxCb(DB *pIdx, const DBT *pKey, const DBT *pValue, DBT *pSKey) {
H
Hongze Cheng 已提交
373
  STbCfg *pTbCfg = (STbCfg *)(pValue->app_data);
H
Hongze Cheng 已提交
374
  DBT *   pDbt;
H
Hongze Cheng 已提交
375

H
Hongze Cheng 已提交
376
  if (pTbCfg->type == META_CHILD_TABLE) {
H
Hongze Cheng 已提交
377
    // pDbt = calloc(2, sizeof(DBT));
H
Hongze Cheng 已提交
378

H
Hongze Cheng 已提交
379 380 381
    // // First key is suid
    // pDbt[0].data = &(pTbCfg->ctbCfg.suid);
    // pDbt[0].size = sizeof(pTbCfg->ctbCfg.suid);
H
Hongze Cheng 已提交
382

H
Hongze Cheng 已提交
383 384 385 386
    // // Second key is the first tag
    // void *pTagVal = tdGetKVRowValOfCol(pTbCfg->ctbCfg.pTag, (kvRowColIdx(pTbCfg->ctbCfg.pTag))[0].colId);
    // pDbt[1].data = pTagVal;
    // pDbt[1].size = sizeof(int32_t);
H
Hongze Cheng 已提交
387

H
Hongze Cheng 已提交
388
    // Set index key
H
Hongze Cheng 已提交
389
    memset(pSKey, 0, sizeof(*pSKey));
H
Hongze Cheng 已提交
390
#if 0
H
Hongze Cheng 已提交
391 392 393
    pSKey->flags = DB_DBT_MULTIPLE | DB_DBT_APPMALLOC;
    pSKey->data = pDbt;
    pSKey->size = 2;
H
Hongze Cheng 已提交
394 395 396 397
#else
    pSKey->data = &(pTbCfg->ctbCfg.suid);
    pSKey->size = sizeof(pTbCfg->ctbCfg.suid);
#endif
H
Hongze Cheng 已提交
398

H
Hongze Cheng 已提交
399 400 401 402
    return 0;
  } else {
    return DB_DONOTINDEX;
  }
H
Hongze Cheng 已提交
403 404 405
}

static int metaEncodeTbInfo(void **buf, STbCfg *pTbCfg) {
H
more  
Hongze Cheng 已提交
406 407 408 409 410
  int tsize = 0;

  tsize += taosEncodeString(buf, pTbCfg->name);
  tsize += taosEncodeFixedU32(buf, pTbCfg->ttl);
  tsize += taosEncodeFixedU32(buf, pTbCfg->keep);
H
Hongze Cheng 已提交
411
  tsize += taosEncodeFixedU8(buf, pTbCfg->type);
H
more  
Hongze Cheng 已提交
412 413

  if (pTbCfg->type == META_SUPER_TABLE) {
H
Hongze Cheng 已提交
414 415
    SSchemaWrapper sw = {.nCols = pTbCfg->stbCfg.nTagCols, .pSchema = pTbCfg->stbCfg.pTagSchema};
    tsize += metaEncodeSchema(buf, &sw);
H
more  
Hongze Cheng 已提交
416 417 418 419 420 421 422 423 424
  } else if (pTbCfg->type == META_CHILD_TABLE) {
    tsize += taosEncodeFixedU64(buf, pTbCfg->ctbCfg.suid);
    tsize += tdEncodeKVRow(buf, pTbCfg->ctbCfg.pTag);
  } else if (pTbCfg->type == META_NORMAL_TABLE) {
  } else {
    ASSERT(0);
  }

  return tsize;
H
Hongze Cheng 已提交
425 426 427
}

static void *metaDecodeTbInfo(void *buf, STbCfg *pTbCfg) {
H
more  
Hongze Cheng 已提交
428 429 430
  buf = taosDecodeString(buf, &(pTbCfg->name));
  buf = taosDecodeFixedU32(buf, &(pTbCfg->ttl));
  buf = taosDecodeFixedU32(buf, &(pTbCfg->keep));
H
Hongze Cheng 已提交
431
  buf = taosDecodeFixedU8(buf, &(pTbCfg->type));
H
more  
Hongze Cheng 已提交
432 433

  if (pTbCfg->type == META_SUPER_TABLE) {
H
Hongze Cheng 已提交
434 435 436 437
    SSchemaWrapper sw;
    buf = metaDecodeSchema(buf, &sw);
    pTbCfg->stbCfg.nTagCols = sw.nCols;
    pTbCfg->stbCfg.pTagSchema = sw.pSchema;
H
more  
Hongze Cheng 已提交
438 439 440 441 442 443 444 445
  } else if (pTbCfg->type == META_CHILD_TABLE) {
    buf = taosDecodeFixedU64(buf, &(pTbCfg->ctbCfg.suid));
    buf = tdDecodeKVRow(buf, &(pTbCfg->ctbCfg.pTag));
  } else if (pTbCfg->type == META_NORMAL_TABLE) {
  } else {
    ASSERT(0);
  }
  return buf;
H
Hongze Cheng 已提交
446 447
}

H
Hongze Cheng 已提交
448 449 450 451 452 453 454
static void metaClearTbCfg(STbCfg *pTbCfg) {
  tfree(pTbCfg->name);
  if (pTbCfg->type == META_SUPER_TABLE) {
    tdFreeSchema(pTbCfg->stbCfg.pTagSchema);
  } else if (pTbCfg->type == META_CHILD_TABLE) {
    tfree(pTbCfg->ctbCfg.pTag);
  }
H
more  
Hongze Cheng 已提交
455 456 457
}

/* ------------------------ FOR QUERY ------------------------ */
H
Hongze Cheng 已提交
458
STbCfg *metaGetTbInfoByUid(SMeta *pMeta, tb_uid_t uid) {
H
Hongze Cheng 已提交
459 460 461 462 463 464 465 466 467 468 469 470 471 472 473 474 475 476 477 478 479
  STbCfg * pTbCfg = NULL;
  SMetaDB *pDB = pMeta->pDB;
  DBT      key = {0};
  DBT      value = {0};
  int      ret;

  // Set key/value
  key.data = &uid;
  key.size = sizeof(uid);

  // Query
  ret = pDB->pTbDB->get(pDB->pTbDB, NULL, &key, &value, 0);
  if (ret != 0) {
    return NULL;
  }

  // Decode
  pTbCfg = (STbCfg *)malloc(sizeof(*pTbCfg));
  if (pTbCfg == NULL) {
    return NULL;
  }
H
more  
Hongze Cheng 已提交
480

H
Hongze Cheng 已提交
481 482 483 484 485
  metaDecodeTbInfo(value.data, pTbCfg);

  return pTbCfg;
}

H
Hongze Cheng 已提交
486
STbCfg *metaGetTbInfoByName(SMeta *pMeta, char *tbname, tb_uid_t *uid) {
H
Hongze Cheng 已提交
487 488 489 490 491 492 493 494
  STbCfg * pTbCfg = NULL;
  SMetaDB *pDB = pMeta->pDB;
  DBT      key = {0};
  DBT      pkey = {0};
  DBT      pvalue = {0};
  int      ret;

  // Set key/value
H
more  
Hongze Cheng 已提交
495
  key.data = tbname;
H
Hongze Cheng 已提交
496
  key.size = strlen(tbname);
H
more  
Hongze Cheng 已提交
497

H
Hongze Cheng 已提交
498 499
  // Query
  ret = pDB->pNameIdx->pget(pDB->pNameIdx, NULL, &key, &pkey, &pvalue, 0);
H
Hongze Cheng 已提交
500
  if (ret != 0) {
H
Hongze Cheng 已提交
501
    return NULL;
H
Hongze Cheng 已提交
502 503
  }

H
Hongze Cheng 已提交
504 505 506 507 508 509 510 511 512 513
  // Decode
  *uid = *(tb_uid_t *)(pkey.data);
  pTbCfg = (STbCfg *)malloc(sizeof(*pTbCfg));
  if (pTbCfg == NULL) {
    return NULL;
  }

  metaDecodeTbInfo(pvalue.data, pTbCfg);

  return pTbCfg;
H
more  
Hongze Cheng 已提交
514 515
}

H
Hongze Cheng 已提交
516
SSchemaWrapper *metaGetTableSchema(SMeta *pMeta, tb_uid_t uid, int32_t sver, bool isinline) {
H
more  
Hongze Cheng 已提交
517 518 519 520 521 522
  uint32_t        nCols;
  SSchemaWrapper *pSW = NULL;
  SMetaDB *       pDB = pMeta->pDB;
  int             ret;
  void *          pBuf;
  SSchema *       pSchema;
H
Hongze Cheng 已提交
523
  SSchemaKey      schemaKey = {uid, sver, 0};
H
more  
Hongze Cheng 已提交
524 525 526 527 528 529 530 531 532 533
  DBT             key = {0};
  DBT             value = {0};

  // Set key/value properties
  key.data = &schemaKey;
  key.size = sizeof(schemaKey);

  // Query
  ret = pDB->pSchemaDB->get(pDB->pSchemaDB, NULL, &key, &value, 0);
  if (ret != 0) {
H
Hongze Cheng 已提交
534
    printf("failed to query schema DB since %s================\n", db_strerror(ret));
H
more  
Hongze Cheng 已提交
535 536 537 538 539
    return NULL;
  }

  // Decode the schema
  pBuf = value.data;
H
Hongze Cheng 已提交
540 541
  pSW = malloc(sizeof(*pSW));
  metaDecodeSchema(pBuf, pSW);
H
more  
Hongze Cheng 已提交
542 543

  return pSW;
H
more  
Hongze Cheng 已提交
544 545 546 547 548 549 550 551 552 553 554 555 556 557 558 559 560
}

struct SMTbCursor {
  DBC *pCur;
};

SMTbCursor *metaOpenTbCursor(SMeta *pMeta) {
  SMTbCursor *pTbCur = NULL;
  SMetaDB *   pDB = pMeta->pDB;

  pTbCur = (SMTbCursor *)calloc(1, sizeof(*pTbCur));
  if (pTbCur == NULL) {
    return NULL;
  }

  pDB->pTbDB->cursor(pDB->pTbDB, NULL, &(pTbCur->pCur), 0);

H
more  
Hongze Cheng 已提交
561 562 563 564 565 566
#if 0
    DB_BTREE_STAT *sp;
    pDB->pTbDB->stat(pDB->pTbDB, NULL, &sp, 0);
    printf("**************** %ld\n", sp->bt_nkeys);
#endif

H
more  
Hongze Cheng 已提交
567 568 569 570 571 572 573 574 575 576 577 578 579 580 581 582
  return pTbCur;
}

void metaCloseTbCursor(SMTbCursor *pTbCur) {
  if (pTbCur) {
    if (pTbCur->pCur) {
      pTbCur->pCur->close(pTbCur->pCur);
    }
    free(pTbCur);
  }
}

char *metaTbCursorNext(SMTbCursor *pTbCur) {
  DBT    key = {0};
  DBT    value = {0};
  STbCfg tbCfg;
H
Hongze Cheng 已提交
583
  void * pBuf;
H
more  
Hongze Cheng 已提交
584

H
Hongze Cheng 已提交
585 586 587 588 589
  for (;;) {
    if (pTbCur->pCur->get(pTbCur->pCur, &key, &value, DB_NEXT) == 0) {
      pBuf = value.data;
      metaDecodeTbInfo(pBuf, &tbCfg);
      if (tbCfg.type == META_SUPER_TABLE) {
H
Hongze Cheng 已提交
590
        free(tbCfg.name);
H
Hongze Cheng 已提交
591
        free(tbCfg.stbCfg.pTagSchema);
H
Hongze Cheng 已提交
592
        continue;
H
Hongze Cheng 已提交
593
      } else if (tbCfg.type == META_CHILD_TABLE) {
H
Hongze Cheng 已提交
594
        kvRowFree(tbCfg.ctbCfg.pTag);
H
Hongze Cheng 已提交
595 596 597 598 599
      }
      return tbCfg.name;
    } else {
      return NULL;
    }
H
more  
Hongze Cheng 已提交
600
  }
H
more  
Hongze Cheng 已提交
601 602 603 604 605 606 607
}

STSchema *metaGetTbTSchema(SMeta *pMeta, tb_uid_t uid, int32_t sver) {
  STSchemaBuilder sb;
  STSchema *      pTSchema = NULL;
  SSchema *       pSchema;
  SSchemaWrapper *pSW;
H
Hongze Cheng 已提交
608 609
  STbCfg *        pTbCfg;
  tb_uid_t        quid;
H
more  
Hongze Cheng 已提交
610

H
Hongze Cheng 已提交
611 612 613 614 615 616 617 618
  pTbCfg = metaGetTbInfoByUid(pMeta, uid);
  if (pTbCfg->type == META_CHILD_TABLE) {
    quid = pTbCfg->ctbCfg.suid;
  } else {
    quid = uid;
  }

  pSW = metaGetTableSchema(pMeta, quid, sver, true);
H
more  
Hongze Cheng 已提交
619 620 621 622 623 624 625 626 627 628 629 630 631 632
  if (pSW == NULL) {
    return NULL;
  }

  // Rebuild a schema
  tdInitTSchemaBuilder(&sb, 0);
  for (int32_t i = 0; i < pSW->nCols; i++) {
    pSchema = pSW->pSchema + i;
    tdAddColToSchema(&sb, pSchema->type, pSchema->colId, pSchema->bytes);
  }
  pTSchema = tdGetSchemaFromBuilder(&sb);
  tdDestroyTSchemaBuilder(&sb);

  return pTSchema;
H
more  
Hongze Cheng 已提交
633 634 635 636 637 638 639 640 641 642 643 644 645 646 647 648 649 650 651 652 653 654 655 656 657 658 659 660 661 662 663 664 665 666 667 668 669
}

struct SMCtbCursor {
  DBC *    pCur;
  tb_uid_t suid;
};

SMCtbCursor *metaOpenCtbCursor(SMeta *pMeta, tb_uid_t uid) {
  SMCtbCursor *pCtbCur = NULL;
  SMetaDB *    pDB = pMeta->pDB;
  int          ret;

  pCtbCur = (SMCtbCursor *)calloc(1, sizeof(*pCtbCur));
  if (pCtbCur == NULL) {
    return NULL;
  }

  pCtbCur->suid = uid;
  ret = pDB->pCtbIdx->cursor(pDB->pCtbIdx, NULL, &(pCtbCur->pCur), 0);
  if (ret != 0) {
    free(pCtbCur);
    return NULL;
  }

  return pCtbCur;
}

void metaCloseCtbCurosr(SMCtbCursor *pCtbCur) {
  if (pCtbCur) {
    if (pCtbCur->pCur) {
      pCtbCur->pCur->close(pCtbCur->pCur);
    }

    free(pCtbCur);
  }
}

670
tb_uid_t metaCtbCursorNext(SMCtbCursor *pCtbCur) {
H
more  
Hongze Cheng 已提交
671 672 673 674 675 676 677 678 679 680
  DBT    skey = {0};
  DBT    pkey = {0};
  DBT    pval = {0};
  void * pBuf;
  STbCfg tbCfg;

  // Set key
  skey.data = &(pCtbCur->suid);
  skey.size = sizeof(pCtbCur->suid);

681 682 683 684
  if (pCtbCur->pCur->pget(pCtbCur->pCur, &skey, &pkey, &pval, DB_NEXT) == 0) {
    tb_uid_t id = *(tb_uid_t *)pkey.data;
    assert(id != 0);
    return id;
H
Hongze Cheng 已提交
685 686
    //    metaDecodeTbInfo(pBuf, &tbCfg);
    //    return tbCfg.;
H
more  
Hongze Cheng 已提交
687
  } else {
688
    return 0;
H
more  
Hongze Cheng 已提交
689
  }
H
Hongze Cheng 已提交
690
}