xxq250
/
homework-jianmu
1
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity

more

This commit is contained in:
Hongze Cheng 2021-11-04 10:30:23 +08:00
parent b8304e818d
commit f7a8ef266c
2 changed files with 62 additions and 6 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
include/server/vnode/meta/impl/metaImpl.h
View File

@ -53,7 +53,7 @@ typedef struct {
// normal table options // normal table options
typedef struct { typedef struct {
SSchema* pSchame; STSchema* pSchame;
} SNTbOptions; } SNTbOptions;
struct STbOptions { struct STbOptions {

66
source/dnode/vnode/meta/src/metaDB.c
View File

@ -15,6 +15,12 @@
#include "metaDef.h" #include "metaDef.h"
static void metaSaveSchemaDB(SMeta *pMeta, tb_uid_t uid, STSchema *pSchema);
static void metaGetSchemaDBKey(char key[], tb_uid_t uid, int sversion);
static int metaSaveMapDB(SMeta *pMeta, tb_uid_t suid, tb_uid_t uid);
#define SCHEMA_KEY_LEN (sizeof(tb_uid_t) + sizeof(int))
#define META_OPEN_DB_IMPL(pDB, options, dir, err) \ #define META_OPEN_DB_IMPL(pDB, options, dir, err) \
do { \ do { \
pDB = rocksdb_open(options, dir, &err); \ pDB = rocksdb_open(options, dir, &err); \
@ -115,22 +121,22 @@ int metaSaveTableToDB(SMeta *pMeta, const STbOptions *pTbOptions) {
switch (pTbOptions->type) { switch (pTbOptions->type) {
case META_NORMAL_TABLE: case META_NORMAL_TABLE:
// save schemaDB // save schemaDB
rocksdb_put(pMeta->pDB->schemaDb, wopt, NULL /* TODO */, NULL /* TODO */, NULL /* TODO */, NULL /* TODO */, &err); metaSaveSchemaDB(pMeta, uid, pTbOptions->ntbOptions.pSchame);
break; break;
case META_SUPER_TABLE: case META_SUPER_TABLE:
// save schemaDB // save schemaDB
rocksdb_put(pMeta->pDB->schemaDb, wopt, NULL /* TODO */, NULL /* TODO */, NULL /* TODO */, NULL /* TODO */, &err); metaSaveSchemaDB(pMeta, uid, pTbOptions->stbOptions.pSchema);
// save mapDB (really need?) // save mapDB (really need?)
rocksdb_put(pMeta->pDB->mapDb, wopt, (char *)(&uid), sizeof(uid), "", 0, &err); rocksdb_put(pMeta->pDB->mapDb, wopt, (char *)(&uid), sizeof(uid), "", 0, &err);
break; break;
case META_CHILD_TABLE: case META_CHILD_TABLE:
// save tagDB // save tagDB
rocksdb_put(pMeta->pDB->tagDb, wopt, NULL /* TODO */, 0 /* TODO */, NULL /* TODO */, 0 /* TODO */, &err); rocksdb_put(pMeta->pDB->tagDb, wopt, (char *)(&uid), sizeof(uid), pTbOptions->ctbOptions.tags,
kvRowLen(pTbOptions->ctbOptions.tags), &err);
// save mapDB // save mapDB
rocksdb_put(pMeta->pDB->mapDb, wopt, (char *)(&(pTbOptions->ctbOptions.suid)), sizeof(tb_uid_t), NULL /* TODO */, metaSaveMapDB(pMeta, pTbOptions->ctbOptions.suid, uid);
0 /* TODO */, &err);
break; break;
default: default:
ASSERT(0); ASSERT(0);
@ -145,3 +151,53 @@ int metaRemoveTableFromDb(SMeta *pMeta, tb_uid_t uid) {
/* TODO */ /* TODO */
return 0; return 0;
} }
/* ------------------------ STATIC METHODS ------------------------ */
static void metaSaveSchemaDB(SMeta *pMeta, tb_uid_t uid, STSchema *pSchema) {
char key[64];
char pBuf[1024];
char * ppBuf = pBuf;
size_t vsize;
char * err = NULL;
rocksdb_writeoptions_t *wopt = rocksdb_writeoptions_create();
metaGetSchemaDBKey(key, uid, schemaVersion(pSchema));
vsize = tdEncodeSchema((void **)(&ppBuf), pSchema);
rocksdb_put(pMeta->pDB->schemaDb, wopt, key, SCHEMA_KEY_LEN, pBuf, vsize, &err);
rocksdb_writeoptions_destroy(wopt);
}
static void metaGetSchemaDBKey(char *key, tb_uid_t uid, int sversion) {
*(tb_uid_t *)key = uid;
*(int *)POINTER_SHIFT(key, sizeof(tb_uid_t)) = sversion;
}
static int metaSaveMapDB(SMeta *pMeta, tb_uid_t suid, tb_uid_t uid) {
size_t vlen;
char * val;
char * err = NULL;
rocksdb_readoptions_t *ropt = rocksdb_readoptions_create();
val = rocksdb_get(pMeta->pDB->mapDb, ropt, (char *)(&suid), sizeof(suid), &vlen, &err);
rocksdb_readoptions_destroy(ropt);
void *nval = malloc(vlen + sizeof(uid));
if (nval == NULL) {
return -1;
}
if (vlen) {
memcpy(nval, val, vlen);
}
memcpy(POINTER_SHIFT(nval, vlen), (void *)(&uid), sizeof(uid));
rocksdb_writeoptions_t *wopt = rocksdb_writeoptions_create();
rocksdb_put(pMeta->pDB->mapDb, wopt, (char *)(&suid), sizeof(suid), nval, vlen + sizeof(uid), &err);
rocksdb_writeoptions_destroy(wopt);
return 0;
}