metaBDBImpl.c 22.2 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/>.
 */

16
#define ALLOW_FORBID_FUNC
H
more  
Hongze Cheng 已提交
17 18
#include "db.h"

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

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

H
Hongze Cheng 已提交
24 25 26 27
#define IMPL_WITH_LOCK 1
// #if IMPL_WITH_LOCK
// #endif

H
Hongze Cheng 已提交
28 29 30
typedef struct {
  tb_uid_t uid;
  int32_t  sver;
H
Hongze Cheng 已提交
31
  int32_t  padding;
H
Hongze Cheng 已提交
32 33
} SSchemaKey;

H
Hongze Cheng 已提交
34
struct SMetaDB {
H
Hongze Cheng 已提交
35 36 37
#if IMPL_WITH_LOCK
  pthread_rwlock_t rwlock;
#endif
H
more  
Hongze Cheng 已提交
38
  // DB
H
more  
Hongze Cheng 已提交
39 40
  DB *pTbDB;
  DB *pSchemaDB;
C
Cary Xu 已提交
41
  DB *pSmaDB;
C
Cary Xu 已提交
42

H
more  
Hongze Cheng 已提交
43
  // IDX
H
more  
Hongze Cheng 已提交
44 45 46 47
  DB *pNameIdx;
  DB *pStbIdx;
  DB *pNtbIdx;
  DB *pCtbIdx;
C
Cary Xu 已提交
48
  DB *pSmaIdx;
H
more  
Hongze Cheng 已提交
49
  // ENV
H
Hongze Cheng 已提交
50 51 52
  DB_ENV *pEvn;
};

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

H
more  
Hongze Cheng 已提交
55 56
static SMetaDB *metaNewDB();
static void     metaFreeDB(SMetaDB *pDB);
H
more  
Hongze Cheng 已提交
57 58
static int      metaOpenBDBEnv(DB_ENV **ppEnv, const char *path);
static void     metaCloseBDBEnv(DB_ENV *pEnv);
H
Hongze Cheng 已提交
59
static int      metaOpenBDBDb(DB **ppDB, DB_ENV *pEnv, const char *pFName, bool isDup);
H
more  
Hongze Cheng 已提交
60
static void     metaCloseBDBDb(DB *pDB);
H
Hongze Cheng 已提交
61
static int      metaOpenBDBIdx(DB **ppIdx, DB_ENV *pEnv, const char *pFName, DB *pDB, bdbIdxCbPtr cbf, bool isDup);
H
Hongze Cheng 已提交
62
static void     metaCloseBDBIdx(DB *pIdx);
H
Hongze Cheng 已提交
63 64 65 66
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);
C
Cary Xu 已提交
67
static int      metaSmaIdxCb(DB *pIdx, const DBT *pKey, const DBT *pValue, DBT *pSKey);
H
Hongze Cheng 已提交
68
static int      metaEncodeTbInfo(void **buf, STbCfg *pTbCfg);
L
fix  
Liu Jicong 已提交
69
static void    *metaDecodeTbInfo(void *buf, STbCfg *pTbCfg);
H
Hongze Cheng 已提交
70
static void     metaClearTbCfg(STbCfg *pTbCfg);
H
Hongze Cheng 已提交
71
static int      metaEncodeSchema(void **buf, SSchemaWrapper *pSW);
L
fix  
Liu Jicong 已提交
72
static void    *metaDecodeSchema(void *buf, SSchemaWrapper *pSW);
H
Hongze Cheng 已提交
73 74 75
static void     metaDBWLock(SMetaDB *pDB);
static void     metaDBRLock(SMetaDB *pDB);
static void     metaDBULock(SMetaDB *pDB);
H
more  
Hongze Cheng 已提交
76

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

H
Hongze Cheng 已提交
79
int metaOpenDB(SMeta *pMeta) {
H
more  
Hongze Cheng 已提交
80
  SMetaDB *pDB;
H
Hongze Cheng 已提交
81

H
more  
Hongze Cheng 已提交
82 83 84
  // Create DB object
  pDB = metaNewDB();
  if (pDB == NULL) {
H
more  
Hongze Cheng 已提交
85 86 87
    return -1;
  }

H
more  
Hongze Cheng 已提交
88
  pMeta->pDB = pDB;
H
more  
Hongze Cheng 已提交
89

H
more  
Hongze Cheng 已提交
90 91
  // Open DB Env
  if (metaOpenBDBEnv(&(pDB->pEvn), pMeta->path) < 0) {
H
more  
Hongze Cheng 已提交
92
    metaCloseDB(pMeta);
H
more  
Hongze Cheng 已提交
93 94
    return -1;
  }
H
Hongze Cheng 已提交
95

H
more  
Hongze Cheng 已提交
96
  // Open DBs
H
Hongze Cheng 已提交
97
  if (metaOpenBDBDb(&(pDB->pTbDB), pDB->pEvn, "meta.db", false) < 0) {
H
more  
Hongze Cheng 已提交
98 99
    metaCloseDB(pMeta);
    return -1;
H
more  
Hongze Cheng 已提交
100 101
  }

H
more  
Hongze Cheng 已提交
102
  if (metaOpenBDBDb(&(pDB->pSchemaDB), pDB->pEvn, "schema.db", false) < 0) {
H
more  
Hongze Cheng 已提交
103 104 105
    metaCloseDB(pMeta);
    return -1;
  }
H
Hongze Cheng 已提交
106

C
Cary Xu 已提交
107 108 109 110 111
  if (metaOpenBDBDb(&(pDB->pSmaDB), pDB->pEvn, "sma.db", false) < 0) {
    metaCloseDB(pMeta);
    return -1;
  }

H
more  
Hongze Cheng 已提交
112
  // Open Indices
H
Hongze Cheng 已提交
113
  if (metaOpenBDBIdx(&(pDB->pNameIdx), pDB->pEvn, "name.index", pDB->pTbDB, &metaNameIdxCb, false) < 0) {
H
more  
Hongze Cheng 已提交
114 115 116
    metaCloseDB(pMeta);
    return -1;
  }
H
Hongze Cheng 已提交
117

H
Hongze Cheng 已提交
118
  if (metaOpenBDBIdx(&(pDB->pStbIdx), pDB->pEvn, "stb.index", pDB->pTbDB, &metaStbIdxCb, false) < 0) {
H
more  
Hongze Cheng 已提交
119 120 121
    metaCloseDB(pMeta);
    return -1;
  }
H
Hongze Cheng 已提交
122

H
Hongze Cheng 已提交
123
  if (metaOpenBDBIdx(&(pDB->pNtbIdx), pDB->pEvn, "ntb.index", pDB->pTbDB, &metaNtbIdxCb, false) < 0) {
H
more  
Hongze Cheng 已提交
124 125
    metaCloseDB(pMeta);
    return -1;
H
Hongze Cheng 已提交
126 127
  }

H
Hongze Cheng 已提交
128
  if (metaOpenBDBIdx(&(pDB->pCtbIdx), pDB->pEvn, "ctb.index", pDB->pTbDB, &metaCtbIdxCb, true) < 0) {
H
more  
Hongze Cheng 已提交
129 130
    metaCloseDB(pMeta);
    return -1;
H
more  
Hongze Cheng 已提交
131 132
  }

C
Cary Xu 已提交
133 134 135 136 137
  if (metaOpenBDBIdx(&(pDB->pSmaIdx), pDB->pEvn, "sma.index", pDB->pSmaDB, &metaSmaIdxCb, true) < 0) {
    metaCloseDB(pMeta);
    return -1;
  }

H
Hongze Cheng 已提交
138 139 140 141 142
  return 0;
}

void metaCloseDB(SMeta *pMeta) {
  if (pMeta->pDB) {
C
Cary Xu 已提交
143
    metaCloseBDBIdx(pMeta->pDB->pSmaIdx);
H
Hongze Cheng 已提交
144 145 146 147
    metaCloseBDBIdx(pMeta->pDB->pCtbIdx);
    metaCloseBDBIdx(pMeta->pDB->pNtbIdx);
    metaCloseBDBIdx(pMeta->pDB->pStbIdx);
    metaCloseBDBIdx(pMeta->pDB->pNameIdx);
C
Cary Xu 已提交
148
    metaCloseBDBDb(pMeta->pDB->pSmaDB);
H
Hongze Cheng 已提交
149 150
    metaCloseBDBDb(pMeta->pDB->pSchemaDB);
    metaCloseBDBDb(pMeta->pDB->pTbDB);
H
more  
Hongze Cheng 已提交
151
    metaCloseBDBEnv(pMeta->pDB->pEvn);
H
more  
Hongze Cheng 已提交
152 153
    metaFreeDB(pMeta->pDB);
    pMeta->pDB = NULL;
H
Hongze Cheng 已提交
154 155 156
  }
}

H
more  
Hongze Cheng 已提交
157
int metaSaveTableToDB(SMeta *pMeta, STbCfg *pTbCfg) {
H
more  
Hongze Cheng 已提交
158 159
  tb_uid_t uid;
  char     buf[512];
H
fix  
Hongze Cheng 已提交
160
  char     buf1[512];
L
fix  
Liu Jicong 已提交
161
  void    *pBuf;
H
Hongze Cheng 已提交
162 163
  DBT      key1, value1;
  DBT      key2, value2;
H
more  
Hongze Cheng 已提交
164
  SSchema *pSchema = NULL;
H
Hongze Cheng 已提交
165 166 167 168 169 170 171 172 173

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

  {
    // save table info
H
Hongze Cheng 已提交
174
    pBuf = buf;
H
Hongze Cheng 已提交
175 176
    memset(&key1, 0, sizeof(key1));
    memset(&value1, 0, sizeof(key1));
H
Hongze Cheng 已提交
177

H
Hongze Cheng 已提交
178 179
    key1.data = &uid;
    key1.size = sizeof(uid);
H
Hongze Cheng 已提交
180

H
Hongze Cheng 已提交
181
    metaEncodeTbInfo(&pBuf, pTbCfg);
H
Hongze Cheng 已提交
182

H
Hongze Cheng 已提交
183 184 185
    value1.data = buf;
    value1.size = POINTER_DISTANCE(pBuf, buf);
    value1.app_data = pTbCfg;
H
Hongze Cheng 已提交
186 187 188
  }

  // save schema
H
more  
Hongze Cheng 已提交
189
  uint32_t ncols;
H
Hongze Cheng 已提交
190
  if (pTbCfg->type == META_SUPER_TABLE) {
H
more  
Hongze Cheng 已提交
191
    ncols = pTbCfg->stbCfg.nCols;
H
Hongze Cheng 已提交
192 193
    pSchema = pTbCfg->stbCfg.pSchema;
  } else if (pTbCfg->type == META_NORMAL_TABLE) {
H
more  
Hongze Cheng 已提交
194
    ncols = pTbCfg->ntbCfg.nCols;
H
Hongze Cheng 已提交
195 196 197 198
    pSchema = pTbCfg->ntbCfg.pSchema;
  }

  if (pSchema) {
H
fix  
Hongze Cheng 已提交
199
    pBuf = buf1;
H
Hongze Cheng 已提交
200 201
    memset(&key2, 0, sizeof(key2));
    memset(&value2, 0, sizeof(key2));
H
Hongze Cheng 已提交
202
    SSchemaKey schemaKey = {uid, 0 /*TODO*/, 0};
H
Hongze Cheng 已提交
203

H
Hongze Cheng 已提交
204 205
    key2.data = &schemaKey;
    key2.size = sizeof(schemaKey);
H
Hongze Cheng 已提交
206

H
Hongze Cheng 已提交
207 208
    SSchemaWrapper sw = {.nCols = ncols, .pSchema = pSchema};
    metaEncodeSchema(&pBuf, &sw);
H
Hongze Cheng 已提交
209

H
fix  
Hongze Cheng 已提交
210 211
    value2.data = buf1;
    value2.size = POINTER_DISTANCE(pBuf, buf1);
H
Hongze Cheng 已提交
212 213
  }

H
Hongze Cheng 已提交
214 215
  metaDBWLock(pMeta->pDB);
  pMeta->pDB->pTbDB->put(pMeta->pDB->pTbDB, NULL, &key1, &value1, 0);
H
fix  
Hongze Cheng 已提交
216 217 218
  if (pSchema) {
    pMeta->pDB->pSchemaDB->put(pMeta->pDB->pSchemaDB, NULL, &key2, &value2, 0);
  }
H
Hongze Cheng 已提交
219 220
  metaDBULock(pMeta->pDB);

H
more  
Hongze Cheng 已提交
221 222 223 224 225 226 227 228
  return 0;
}

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

C
Cary Xu 已提交
229
int metaSaveSmaToDB(SMeta *pMeta, STSma *pSmaCfg) {
C
Cary Xu 已提交
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 258 259 260 261 262 263 264 265 266 267 268 269 270 271
  char  buf[512] = {0};  // TODO: may overflow
  void *pBuf = NULL;
  DBT   key1 = {0}, value1 = {0};

  {
    // save sma info
    pBuf = buf;

    key1.data = pSmaCfg->indexName;
    key1.size = strlen(key1.data);

    tEncodeTSma(&pBuf, pSmaCfg);

    value1.data = buf;
    value1.size = POINTER_DISTANCE(pBuf, buf);
    value1.app_data = pSmaCfg;
  }

  metaDBWLock(pMeta->pDB);
  pMeta->pDB->pSmaDB->put(pMeta->pDB->pSmaDB, NULL, &key1, &value1, 0);
  metaDBULock(pMeta->pDB);

  return 0;
}

int metaRemoveSmaFromDb(SMeta *pMeta, const char *indexName) {
  // TODO
#if 0
  DBT key = {0};

  key.data = (void *)indexName;
  key.size = strlen(indexName);

  metaDBWLock(pMeta->pDB);
  // TODO: No guarantee of consistence.
  // Use transaction or DB->sync() for some guarantee.
  pMeta->pDB->pSmaDB->del(pMeta->pDB->pSmaDB, NULL, &key, 0);
  metaDBULock(pMeta->pDB);
#endif
  return 0;
}

H
more  
Hongze Cheng 已提交
272
/* ------------------------ STATIC METHODS ------------------------ */
H
Hongze Cheng 已提交
273 274 275 276 277 278 279 280 281 282 283 284 285 286 287 288 289 290 291 292 293 294 295 296 297 298 299 300 301 302 303 304
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 已提交
305 306 307 308 309 310 311
static SMetaDB *metaNewDB() {
  SMetaDB *pDB = NULL;
  pDB = (SMetaDB *)calloc(1, sizeof(*pDB));
  if (pDB == NULL) {
    return NULL;
  }

H
Hongze Cheng 已提交
312 313 314 315
#if IMPL_WITH_LOCK
  pthread_rwlock_init(&pDB->rwlock, NULL);
#endif

H
more  
Hongze Cheng 已提交
316 317 318 319 320
  return pDB;
}

static void metaFreeDB(SMetaDB *pDB) {
  if (pDB) {
H
Hongze Cheng 已提交
321 322 323
#if IMPL_WITH_LOCK
    pthread_rwlock_destroy(&pDB->rwlock);
#endif
H
more  
Hongze Cheng 已提交
324 325 326 327 328 329 330 331 332 333 334 335 336 337 338 339
    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 已提交
340
  ret = pEnv->open(pEnv, path, DB_CREATE | DB_INIT_CDB | DB_INIT_MPOOL, 0);
H
more  
Hongze Cheng 已提交
341 342 343 344 345 346 347 348 349 350 351 352 353 354 355 356
  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 已提交
357
static int metaOpenBDBDb(DB **ppDB, DB_ENV *pEnv, const char *pFName, bool isDup) {
H
more  
Hongze Cheng 已提交
358 359 360
  int ret;
  DB *pDB;

H
Hongze Cheng 已提交
361
  ret = db_create(&(pDB), pEnv, 0);
H
more  
Hongze Cheng 已提交
362 363 364 365 366
  if (ret != 0) {
    BDB_PERR("Failed to create META DB", ret);
    return -1;
  }

H
Hongze Cheng 已提交
367 368 369 370 371 372 373 374
  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 已提交
375 376 377 378 379 380
  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 已提交
381 382
  *ppDB = pDB;

H
more  
Hongze Cheng 已提交
383 384 385 386 387 388 389 390 391
  return 0;
}

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

H
Hongze Cheng 已提交
392
static int metaOpenBDBIdx(DB **ppIdx, DB_ENV *pEnv, const char *pFName, DB *pDB, bdbIdxCbPtr cbf, bool isDup) {
H
Hongze Cheng 已提交
393 394 395
  DB *pIdx;
  int ret;

H
Hongze Cheng 已提交
396
  if (metaOpenBDBDb(ppIdx, pEnv, pFName, isDup) < 0) {
H
Hongze Cheng 已提交
397 398 399 400 401 402 403 404 405 406 407 408 409 410 411 412 413 414 415
    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 已提交
416
  STbCfg *pTbCfg = (STbCfg *)(pValue->app_data);
H
Hongze Cheng 已提交
417

H
Hongze Cheng 已提交
418
  memset(pSKey, 0, sizeof(*pSKey));
H
Hongze Cheng 已提交
419

H
Hongze Cheng 已提交
420 421
  pSKey->data = pTbCfg->name;
  pSKey->size = strlen(pTbCfg->name);
H
Hongze Cheng 已提交
422 423 424 425

  return 0;
}

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

H
Hongze Cheng 已提交
429
  if (pTbCfg->type == META_SUPER_TABLE) {
H
Hongze Cheng 已提交
430 431 432
    memset(pSKey, 0, sizeof(*pSKey));
    pSKey->data = pKey->data;
    pSKey->size = pKey->size;
H
Hongze Cheng 已提交
433

H
Hongze Cheng 已提交
434
    return 0;
H
Hongze Cheng 已提交
435
  } else {
H
Hongze Cheng 已提交
436
    return DB_DONOTINDEX;
H
Hongze Cheng 已提交
437
  }
H
Hongze Cheng 已提交
438 439
}

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

H
Hongze Cheng 已提交
443
  if (pTbCfg->type == META_NORMAL_TABLE) {
H
Hongze Cheng 已提交
444 445 446
    memset(pSKey, 0, sizeof(*pSKey));
    pSKey->data = pKey->data;
    pSKey->size = pKey->size;
H
Hongze Cheng 已提交
447

H
Hongze Cheng 已提交
448 449 450 451
    return 0;
  } else {
    return DB_DONOTINDEX;
  }
H
Hongze Cheng 已提交
452 453
}

H
Hongze Cheng 已提交
454
static int metaCtbIdxCb(DB *pIdx, const DBT *pKey, const DBT *pValue, DBT *pSKey) {
H
Hongze Cheng 已提交
455
  STbCfg *pTbCfg = (STbCfg *)(pValue->app_data);
L
fix  
Liu Jicong 已提交
456
  DBT    *pDbt;
H
Hongze Cheng 已提交
457

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

H
Hongze Cheng 已提交
461 462 463
    // // First key is suid
    // pDbt[0].data = &(pTbCfg->ctbCfg.suid);
    // pDbt[0].size = sizeof(pTbCfg->ctbCfg.suid);
H
Hongze Cheng 已提交
464

H
Hongze Cheng 已提交
465 466 467 468
    // // 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 已提交
469

H
Hongze Cheng 已提交
470
    // Set index key
H
Hongze Cheng 已提交
471
    memset(pSKey, 0, sizeof(*pSKey));
H
Hongze Cheng 已提交
472
#if 0
H
Hongze Cheng 已提交
473 474 475
    pSKey->flags = DB_DBT_MULTIPLE | DB_DBT_APPMALLOC;
    pSKey->data = pDbt;
    pSKey->size = 2;
H
Hongze Cheng 已提交
476 477 478 479
#else
    pSKey->data = &(pTbCfg->ctbCfg.suid);
    pSKey->size = sizeof(pTbCfg->ctbCfg.suid);
#endif
H
Hongze Cheng 已提交
480

H
Hongze Cheng 已提交
481 482 483 484
    return 0;
  } else {
    return DB_DONOTINDEX;
  }
H
Hongze Cheng 已提交
485 486
}

C
Cary Xu 已提交
487
static int metaSmaIdxCb(DB *pIdx, const DBT *pKey, const DBT *pValue, DBT *pSKey) {
C
Cary Xu 已提交
488
  STSma *pSmaCfg = (STSma *)(pValue->app_data);
C
Cary Xu 已提交
489 490 491 492 493 494 495 496

  memset(pSKey, 0, sizeof(*pSKey));
  pSKey->data = &(pSmaCfg->tableUid);
  pSKey->size = sizeof(pSmaCfg->tableUid);

  return 0;
}

H
Hongze Cheng 已提交
497
static int metaEncodeTbInfo(void **buf, STbCfg *pTbCfg) {
H
more  
Hongze Cheng 已提交
498 499 500 501 502
  int tsize = 0;

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

  if (pTbCfg->type == META_SUPER_TABLE) {
H
Hongze Cheng 已提交
506 507
    SSchemaWrapper sw = {.nCols = pTbCfg->stbCfg.nTagCols, .pSchema = pTbCfg->stbCfg.pTagSchema};
    tsize += metaEncodeSchema(buf, &sw);
H
more  
Hongze Cheng 已提交
508 509 510 511 512 513 514 515 516
  } 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 已提交
517 518 519
}

static void *metaDecodeTbInfo(void *buf, STbCfg *pTbCfg) {
H
more  
Hongze Cheng 已提交
520 521 522
  buf = taosDecodeString(buf, &(pTbCfg->name));
  buf = taosDecodeFixedU32(buf, &(pTbCfg->ttl));
  buf = taosDecodeFixedU32(buf, &(pTbCfg->keep));
H
Hongze Cheng 已提交
523
  buf = taosDecodeFixedU8(buf, &(pTbCfg->type));
H
more  
Hongze Cheng 已提交
524 525

  if (pTbCfg->type == META_SUPER_TABLE) {
H
Hongze Cheng 已提交
526 527 528 529
    SSchemaWrapper sw;
    buf = metaDecodeSchema(buf, &sw);
    pTbCfg->stbCfg.nTagCols = sw.nCols;
    pTbCfg->stbCfg.pTagSchema = sw.pSchema;
H
more  
Hongze Cheng 已提交
530 531 532 533 534 535 536 537
  } 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 已提交
538 539
}

H
Hongze Cheng 已提交
540 541 542 543 544 545 546
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 已提交
547 548 549
}

/* ------------------------ FOR QUERY ------------------------ */
H
Hongze Cheng 已提交
550
STbCfg *metaGetTbInfoByUid(SMeta *pMeta, tb_uid_t uid) {
L
fix  
Liu Jicong 已提交
551
  STbCfg  *pTbCfg = NULL;
H
Hongze Cheng 已提交
552 553 554 555 556 557 558 559 560 561
  SMetaDB *pDB = pMeta->pDB;
  DBT      key = {0};
  DBT      value = {0};
  int      ret;

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

  // Query
H
Hongze Cheng 已提交
562
  metaDBRLock(pDB);
H
Hongze Cheng 已提交
563
  ret = pDB->pTbDB->get(pDB->pTbDB, NULL, &key, &value, 0);
H
Hongze Cheng 已提交
564
  metaDBULock(pDB);
H
Hongze Cheng 已提交
565 566 567 568 569 570 571 572 573
  if (ret != 0) {
    return NULL;
  }

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

H
Hongze Cheng 已提交
575 576 577 578 579
  metaDecodeTbInfo(value.data, pTbCfg);

  return pTbCfg;
}

H
Hongze Cheng 已提交
580
STbCfg *metaGetTbInfoByName(SMeta *pMeta, char *tbname, tb_uid_t *uid) {
L
fix  
Liu Jicong 已提交
581
  STbCfg  *pTbCfg = NULL;
H
Hongze Cheng 已提交
582 583 584 585 586 587 588
  SMetaDB *pDB = pMeta->pDB;
  DBT      key = {0};
  DBT      pkey = {0};
  DBT      pvalue = {0};
  int      ret;

  // Set key/value
H
more  
Hongze Cheng 已提交
589
  key.data = tbname;
H
Hongze Cheng 已提交
590
  key.size = strlen(tbname);
H
more  
Hongze Cheng 已提交
591

H
Hongze Cheng 已提交
592
  // Query
H
Hongze Cheng 已提交
593
  metaDBRLock(pDB);
H
Hongze Cheng 已提交
594
  ret = pDB->pNameIdx->pget(pDB->pNameIdx, NULL, &key, &pkey, &pvalue, 0);
H
Hongze Cheng 已提交
595
  metaDBULock(pDB);
H
Hongze Cheng 已提交
596
  if (ret != 0) {
H
Hongze Cheng 已提交
597
    return NULL;
H
Hongze Cheng 已提交
598 599
  }

H
Hongze Cheng 已提交
600 601 602 603 604 605 606 607 608 609
  // 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 已提交
610 611
}

C
Cary Xu 已提交
612 613
STSma *metaGetSmaInfoByName(SMeta *pMeta, const char *indexName) {
  STSma *  pCfg = NULL;
C
Cary Xu 已提交
614 615 616 617 618 619 620 621 622 623 624 625 626 627 628 629 630 631
  SMetaDB *pDB = pMeta->pDB;
  DBT      key = {0};
  DBT      value = {0};
  int      ret;

  // Set key/value
  key.data = (void *)indexName;
  key.size = strlen(indexName);

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

  // Decode
C
Cary Xu 已提交
632
  pCfg = (STSma *)malloc(sizeof(STSma));
C
Cary Xu 已提交
633 634 635 636 637 638 639 640 641
  if (pCfg == NULL) {
    return NULL;
  }

  tDecodeTSma(value.data, pCfg);

  return pCfg;
}

H
Hongze Cheng 已提交
642
SSchemaWrapper *metaGetTableSchema(SMeta *pMeta, tb_uid_t uid, int32_t sver, bool isinline) {
H
more  
Hongze Cheng 已提交
643 644
  uint32_t        nCols;
  SSchemaWrapper *pSW = NULL;
L
fix  
Liu Jicong 已提交
645
  SMetaDB        *pDB = pMeta->pDB;
H
more  
Hongze Cheng 已提交
646
  int             ret;
L
fix  
Liu Jicong 已提交
647 648
  void           *pBuf;
  SSchema        *pSchema;
H
Hongze Cheng 已提交
649
  SSchemaKey      schemaKey = {uid, sver, 0};
H
more  
Hongze Cheng 已提交
650 651 652 653 654 655 656 657
  DBT             key = {0};
  DBT             value = {0};

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

  // Query
H
Hongze Cheng 已提交
658
  metaDBRLock(pDB);
H
more  
Hongze Cheng 已提交
659
  ret = pDB->pSchemaDB->get(pDB->pSchemaDB, NULL, &key, &value, 0);
H
Hongze Cheng 已提交
660
  metaDBULock(pDB);
H
more  
Hongze Cheng 已提交
661
  if (ret != 0) {
H
Hongze Cheng 已提交
662
    printf("failed to query schema DB since %s================\n", db_strerror(ret));
H
more  
Hongze Cheng 已提交
663 664 665 666 667
    return NULL;
  }

  // Decode the schema
  pBuf = value.data;
H
Hongze Cheng 已提交
668 669
  pSW = malloc(sizeof(*pSW));
  metaDecodeSchema(pBuf, pSW);
H
more  
Hongze Cheng 已提交
670 671

  return pSW;
H
more  
Hongze Cheng 已提交
672 673 674 675 676 677 678 679
}

struct SMTbCursor {
  DBC *pCur;
};

SMTbCursor *metaOpenTbCursor(SMeta *pMeta) {
  SMTbCursor *pTbCur = NULL;
L
fix  
Liu Jicong 已提交
680
  SMetaDB    *pDB = pMeta->pDB;
H
more  
Hongze Cheng 已提交
681 682 683 684 685 686 687 688

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

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

H
more  
Hongze Cheng 已提交
689 690 691 692 693 694
#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 已提交
695 696 697 698 699 700 701 702 703 704 705 706 707 708 709 710
  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;
L
fix  
Liu Jicong 已提交
711
  void  *pBuf;
H
more  
Hongze Cheng 已提交
712

H
Hongze Cheng 已提交
713 714 715 716 717
  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 已提交
718
        free(tbCfg.name);
H
Hongze Cheng 已提交
719
        free(tbCfg.stbCfg.pTagSchema);
H
Hongze Cheng 已提交
720
        continue;
H
Hongze Cheng 已提交
721
      } else if (tbCfg.type == META_CHILD_TABLE) {
H
Hongze Cheng 已提交
722
        kvRowFree(tbCfg.ctbCfg.pTag);
H
Hongze Cheng 已提交
723 724 725 726 727
      }
      return tbCfg.name;
    } else {
      return NULL;
    }
H
more  
Hongze Cheng 已提交
728
  }
H
more  
Hongze Cheng 已提交
729 730 731 732
}

STSchema *metaGetTbTSchema(SMeta *pMeta, tb_uid_t uid, int32_t sver) {
  STSchemaBuilder sb;
L
fix  
Liu Jicong 已提交
733 734
  STSchema       *pTSchema = NULL;
  SSchema        *pSchema;
H
more  
Hongze Cheng 已提交
735
  SSchemaWrapper *pSW;
L
fix  
Liu Jicong 已提交
736
  STbCfg         *pTbCfg;
H
Hongze Cheng 已提交
737
  tb_uid_t        quid;
H
more  
Hongze Cheng 已提交
738

H
Hongze Cheng 已提交
739 740 741 742 743 744 745 746
  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 已提交
747 748 749 750 751 752 753 754 755 756 757 758 759 760
  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 已提交
761 762 763
}

struct SMCtbCursor {
L
fix  
Liu Jicong 已提交
764
  DBC     *pCur;
H
more  
Hongze Cheng 已提交
765 766 767 768 769
  tb_uid_t suid;
};

SMCtbCursor *metaOpenCtbCursor(SMeta *pMeta, tb_uid_t uid) {
  SMCtbCursor *pCtbCur = NULL;
L
fix  
Liu Jicong 已提交
770
  SMetaDB     *pDB = pMeta->pDB;
H
more  
Hongze Cheng 已提交
771 772 773 774 775 776 777 778 779 780 781 782 783 784 785 786 787 788 789 790 791 792 793 794 795 796 797
  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);
  }
}

798
tb_uid_t metaCtbCursorNext(SMCtbCursor *pCtbCur) {
H
more  
Hongze Cheng 已提交
799 800 801
  DBT    skey = {0};
  DBT    pkey = {0};
  DBT    pval = {0};
L
fix  
Liu Jicong 已提交
802
  void  *pBuf;
H
more  
Hongze Cheng 已提交
803 804 805 806 807 808
  STbCfg tbCfg;

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

809 810 811 812
  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 已提交
813 814
    //    metaDecodeTbInfo(pBuf, &tbCfg);
    //    return tbCfg.;
H
more  
Hongze Cheng 已提交
815
  } else {
816
    return 0;
H
more  
Hongze Cheng 已提交
817
  }
H
Hongze Cheng 已提交
818 819
}

C
Cary Xu 已提交
820 821 822 823 824 825 826 827 828 829 830 831 832 833 834 835
struct SMSmaCursor {
  DBC     *pCur;
  tb_uid_t uid;
};

SMSmaCursor *metaOpenSmaCursor(SMeta *pMeta, tb_uid_t uid) {
  SMSmaCursor *pCur = NULL;
  SMetaDB     *pDB = pMeta->pDB;
  int          ret;

  pCur = (SMSmaCursor *)calloc(1, sizeof(*pCur));
  if (pCur == NULL) {
    return NULL;
  }

  pCur->uid = uid;
C
Cary Xu 已提交
836
  // TODO: lock?
C
Cary Xu 已提交
837 838 839 840 841 842 843 844 845 846 847 848 849 850 851 852 853 854 855
  ret = pDB->pCtbIdx->cursor(pDB->pSmaIdx, NULL, &(pCur->pCur), 0);
  if (ret != 0) {
    free(pCur);
    return NULL;
  }

  return pCur;
}

void metaCloseSmaCurosr(SMSmaCursor *pCur) {
  if (pCur) {
    if (pCur->pCur) {
      pCur->pCur->close(pCur->pCur);
    }

    free(pCur);
  }
}

C
Cary Xu 已提交
856 857 858 859
const char *metaSmaCursorNext(SMSmaCursor *pCur) {
  DBT skey = {0};
  DBT pkey = {0};
  DBT pval = {0};
C
Cary Xu 已提交
860 861 862 863

  // Set key
  skey.data = &(pCur->uid);
  skey.size = sizeof(pCur->uid);
C
Cary Xu 已提交
864
  // TODO: lock?
C
Cary Xu 已提交
865
  if (pCur->pCur->pget(pCur->pCur, &skey, &pkey, &pval, DB_NEXT) == 0) {
C
Cary Xu 已提交
866
    const char *indexName = (const char *)pkey.data;
C
Cary Xu 已提交
867 868 869
    assert(indexName != NULL);
    return indexName;
  } else {
C
Cary Xu 已提交
870
    return NULL;
C
Cary Xu 已提交
871 872 873
  }
}

C
Cary Xu 已提交
874 875 876 877 878 879 880 881 882 883 884 885 886 887
STSmaWrapper *metaGetSmaInfoByUid(SMeta *pMeta, tb_uid_t uid) {
  STSmaWrapper *pSW = NULL;

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

  SMSmaCursor *pCur = metaOpenSmaCursor(pMeta, uid);
  if (pCur == NULL) {
    free(pSW);
    return NULL;
  }

C
Cary Xu 已提交
888 889
  DBT   skey = {.data = &(pCur->uid), .size = sizeof(pCur->uid)};
  DBT   pval = {0};
C
Cary Xu 已提交
890 891 892 893 894 895 896 897 898
  void *pBuf = NULL;

  while (true) {
    // TODO: lock?
    if (pCur->pCur->pget(pCur->pCur, &skey, NULL, &pval, DB_NEXT) == 0) {
      ++pSW->number;
      STSma *tptr = (STSma *)realloc(pSW->tSma, pSW->number * sizeof(STSma));
      if (tptr == NULL) {
        metaCloseSmaCurosr(pCur);
C
Cary Xu 已提交
899 900
        tdDestroyTSmaWrapper(pSW);
        tfree(pSW);
C
Cary Xu 已提交
901 902 903 904 905 906
        return NULL;
      }
      pSW->tSma = tptr;
      pBuf = pval.data;
      if (tDecodeTSma(pBuf, pSW->tSma + pSW->number - 1) == NULL) {
        metaCloseSmaCurosr(pCur);
C
Cary Xu 已提交
907 908
        tdDestroyTSmaWrapper(pSW);
        tfree(pSW);
C
Cary Xu 已提交
909 910 911 912 913 914 915 916
        return NULL;
      }
      continue;
    }
    break;
  }

  metaCloseSmaCurosr(pCur);
C
Cary Xu 已提交
917
  
C
Cary Xu 已提交
918 919
  return pSW;
}
C
Cary Xu 已提交
920

C
Cary Xu 已提交
921 922 923 924 925 926 927 928 929 930 931 932 933 934 935 936 937 938 939 940 941 942 943 944 945 946 947 948 949 950 951 952 953 954 955 956 957 958 959
SArray *metaGetSmaTbUids(SMeta *pMeta, bool isDup) {
  SArray * pUids = NULL;
  SMetaDB *pDB = pMeta->pDB;
  DBC *    pCur = NULL;
  DBT      pkey = {0}, pval = {0};
  int      ret;

  pUids = taosArrayInit(16, sizeof(tb_uid_t));

  if (!pUids) {
    return NULL;
  }

  // TODO: lock?
  ret = pDB->pCtbIdx->cursor(pDB->pSmaIdx, NULL, &pCur, 0);
  if (ret != 0) {
    taosArrayDestroy(pUids);
    return NULL;
  }

  void *pBuf = NULL;

  // TODO: lock?
  while (true) {
    ret = pCur->get(pCur, &pkey, &pval, isDup ? DB_NEXT_DUP : DB_NEXT_NODUP);
    if(ret == 0) {
      taosArrayPush(pUids, pkey.data);
      continue;
    }
    break;
  }

  if (pCur) {
    pCur->close(pCur);
  }

  return pUids;
}

H
Hongze Cheng 已提交
960 961 962 963 964 965 966 967 968 969 970 971 972 973 974 975 976
static void metaDBWLock(SMetaDB *pDB) {
#if IMPL_WITH_LOCK
  pthread_rwlock_wrlock(&(pDB->rwlock));
#endif
}

static void metaDBRLock(SMetaDB *pDB) {
#if IMPL_WITH_LOCK
  pthread_rwlock_rdlock(&(pDB->rwlock));
#endif
}

static void metaDBULock(SMetaDB *pDB) {
#if IMPL_WITH_LOCK
  pthread_rwlock_unlock(&(pDB->rwlock));
#endif
}