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

Merge pull request #11115 from taosdata/feature/meta 

Feature/meta
This commit is contained in:
Hongze Cheng 2022-03-30 11:54:11 +08:00 committed by GitHub
parent 2a422c3abb abfb80acac
commit 5f27540e35
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
12 changed files with 725 additions and 181 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

9
source/dnode/vnode/CMakeLists.txt
View File

@ -46,14 +46,11 @@ target_link_libraries(
PUBLIC stream
PUBLIC qworker
PUBLIC sync
# TODO: get rid of BDB
PUBLIC bdb
PUBLIC tdb
)
if(${META_DB_IMPL} STREQUAL "BDB")
target_link_libraries(vnode PUBLIC bdb)
elseif(${META_DB_IMPL} STREQUAL "TDB")
target_link_libraries(vnode PUBLIC tdb)
endif()
if(${BUILD_TEST})
add_subdirectory(test)
endif(${BUILD_TEST})

638
source/dnode/vnode/src/meta/metaTDBImpl.c
View File

@ -15,131 +15,637 @@
#include "metaDef.h"
#include "tdb.h"
#include "tdbInt.h"
struct SMetaDB {
TENV *pEnv;
TDB * pTbDB;
TDB * pSchemaDB;
TDB * pNameIdx;
TDB * pStbIdx;
TDB * pNtbIdx;
TDB * pCtbIdx;
// tag index hash table
// suid+colid --> TDB *
struct {
} tagIdxHt;
TDB *pTbDB;
TDB *pSchemaDB;
TDB *pNameIdx;
TDB *pStbIdx;
TDB *pNtbIdx;
TDB *pCtbIdx;
};
#define A(op, flag) \
do { \
if ((ret = op) != 0) goto flag; \
} while (0)
typedef struct __attribute__((__packed__)) {
tb_uid_t uid;
int32_t sver;
} SSchemaDbKey;
typedef struct {
char *name;
tb_uid_t uid;
} SNameIdxKey;
typedef struct {
tb_uid_t suid;
tb_uid_t uid;
} SCtbIdxKey;
static int metaEncodeTbInfo(void **buf, STbCfg *pTbCfg);
static void *metaDecodeTbInfo(void *buf, STbCfg *pTbCfg);
static int metaEncodeSchema(void **buf, SSchemaWrapper *pSW);
static void *metaDecodeSchema(void *buf, SSchemaWrapper *pSW);
static inline int metaUidCmpr(const void *arg1, int len1, const void *arg2, int len2) {
tb_uid_t uid1, uid2;
ASSERT(len1 == sizeof(tb_uid_t));
ASSERT(len2 == sizeof(tb_uid_t));
uid1 = ((tb_uid_t *)arg1)[0];
uid2 = ((tb_uid_t *)arg2)[0];
if (uid1 < uid2) {
return -1;
}
if (uid1 == uid2) {
return 0;
} else {
return 1;
}
}
static inline int metaSchemaKeyCmpr(const void *arg1, int len1, const void *arg2, int len2) {
int c;
SSchemaDbKey *pKey1 = (SSchemaDbKey *)arg1;
SSchemaDbKey *pKey2 = (SSchemaDbKey *)arg2;
c = metaUidCmpr(arg1, sizeof(tb_uid_t), arg2, sizeof(tb_uid_t));
if (c) return c;
if (pKey1->sver > pKey2->sver) {
return 1;
} else if (pKey1->sver == pKey2->sver) {
return 0;
} else {
return -1;
}
}
static inline int metaNameIdxCmpr(const void *arg1, int len1, const void *arg2, int len2) {
return strcmp((char *)arg1, (char *)arg2);
}
static inline int metaCtbIdxCmpr(const void *arg1, int len1, const void *arg2, int len2) {
int c;
SCtbIdxKey *pKey1 = (SCtbIdxKey *)arg1;
SCtbIdxKey *pKey2 = (SCtbIdxKey *)arg2;
c = metaUidCmpr(arg1, sizeof(tb_uid_t), arg2, sizeof(tb_uid_t));
if (c) return c;
return metaUidCmpr(&pKey1->uid, sizeof(tb_uid_t), &pKey2->uid, sizeof(tb_uid_t));
}
int metaOpenDB(SMeta *pMeta) {
SMetaDB *pDb;
TENV * pEnv;
TDB * pTbDB;
TDB * pSchemaDB;
TDB * pNameIdx;
TDB * pStbIdx;
TDB * pNtbIdx;
TDB * pCtbIdx;
SMetaDB *pMetaDb;
int ret;
pDb = (SMetaDB *)taosMemoryCalloc(1, sizeof(*pDb));
if (pDb == NULL) {
// allocate DB handle
pMetaDb = taosMemoryCalloc(1, sizeof(*pMetaDb));
if (pMetaDb == NULL) {
// TODO
ASSERT(0);
return -1;
}
// Create and open the ENV
A((tdbEnvCreate(&pEnv)), _err);
#if 0
// Set options of the environment
A(tdbEnvSetPageSize(pEnv, 8192), _err);
A(tdbEnvSetCacheSize(pEnv, 16 * 1024 * 1024), _err);
#endif
A((tdbEnvOpen(&pEnv)), _err);
// open the ENV
ret = tdbEnvOpen(pMeta->path, 4096, 256, &(pMetaDb->pEnv));
if (ret < 0) {
// TODO
ASSERT(0);
return -1;
}
// Create and open each DB
A(tdbCreate(&pTbDB), _err);
A(tdbOpen(&pTbDB, "table.db", NULL, pEnv), _err);
// open table DB
ret = tdbDbOpen("table.db", sizeof(tb_uid_t), TDB_VARIANT_LEN, metaUidCmpr, pMetaDb->pEnv, &(pMetaDb->pTbDB));
if (ret < 0) {
// TODO
ASSERT(0);
return -1;
}
A(tdbCreate(&pSchemaDB), _err);
A(tdbOpen(&pSchemaDB, "schema.db", NULL, pEnv), _err);
// open schema DB
ret = tdbDbOpen("schema.db", sizeof(SSchemaDbKey), TDB_VARIANT_LEN, metaSchemaKeyCmpr, pMetaDb->pEnv,
&(pMetaDb->pSchemaDB));
if (ret < 0) {
// TODO
ASSERT(0);
return -1;
}
A(tdbCreate(&pNameIdx), _err);
A(tdbOpen(&pNameIdx, "name.db", NULL, pEnv), _err);
// tdbAssociate();
ret = tdbDbOpen("name.idx", TDB_VARIANT_LEN, 0, metaNameIdxCmpr, pMetaDb->pEnv, &(pMetaDb->pNameIdx));
if (ret < 0) {
// TODO
ASSERT(0);
return -1;
}
pDb->pEnv = pEnv;
pDb->pTbDB = pTbDB;
pDb->pSchemaDB = pSchemaDB;
pMeta->pDB = pDb;
ret = tdbDbOpen("stb.idx", sizeof(tb_uid_t), 0, metaUidCmpr, pMetaDb->pEnv, &(pMetaDb->pStbIdx));
if (ret < 0) {
// TODO
ASSERT(0);
return -1;
}
ret = tdbDbOpen("ntb.idx", sizeof(tb_uid_t), 0, metaUidCmpr, pMetaDb->pEnv, &(pMetaDb->pNtbIdx));
if (ret < 0) {
// TODO
ASSERT(0);
return -1;
}
ret = tdbDbOpen("ctb.idx", sizeof(SCtbIdxKey), 0, metaCtbIdxCmpr, pMetaDb->pEnv, &(pMetaDb->pCtbIdx));
if (ret < 0) {
// TODO
ASSERT(0);
return -1;
}
pMeta->pDB = pMetaDb;
return 0;
_err:
return -1;
}
void metaCloseDB(SMeta *pMeta) {
// TODO
if (pMeta->pDB) {
tdbDbClose(pMeta->pDB->pCtbIdx);
tdbDbClose(pMeta->pDB->pNtbIdx);
tdbDbClose(pMeta->pDB->pStbIdx);
tdbDbClose(pMeta->pDB->pNameIdx);
tdbDbClose(pMeta->pDB->pSchemaDB);
tdbDbClose(pMeta->pDB->pTbDB);
taosMemoryFree(pMeta->pDB);
}
}
int metaSaveTableToDB(SMeta *pMeta, STbCfg *pTbCfg) {
// TODO
tb_uid_t uid;
SMetaDB *pMetaDb;
void *pKey;
void *pVal;
int kLen;
int vLen;
int ret;
char buf[512];
void *pBuf;
SCtbIdxKey ctbIdxKey;
SSchemaDbKey schemaDbKey;
SSchemaWrapper schemaWrapper;
pMetaDb = pMeta->pDB;
// TODO: make this operation pre-process
if (pTbCfg->type == META_SUPER_TABLE) {
uid = pTbCfg->stbCfg.suid;
} else {
uid = metaGenerateUid(pMeta);
}
// save to table.db
pKey = &uid;
kLen = sizeof(uid);
pVal = pBuf = buf;
metaEncodeTbInfo(&pBuf, pTbCfg);
vLen = POINTER_DISTANCE(pBuf, buf);
ret = tdbDbInsert(pMetaDb->pTbDB, pKey, kLen, pVal, vLen);
if (ret < 0) {
return -1;
}
// save to schema.db for META_SUPER_TABLE and META_NORMAL_TABLE
if (pTbCfg->type != META_CHILD_TABLE) {
schemaDbKey.uid = uid;
schemaDbKey.sver = 0; // TODO
pKey = &schemaDbKey;
kLen = sizeof(schemaDbKey);
if (pTbCfg->type == META_SUPER_TABLE) {
schemaWrapper.nCols = pTbCfg->stbCfg.nCols;
schemaWrapper.pSchema = pTbCfg->stbCfg.pSchema;
} else {
schemaWrapper.nCols = pTbCfg->ntbCfg.nCols;
schemaWrapper.pSchema = pTbCfg->ntbCfg.pSchema;
}
pVal = pBuf = buf;
metaEncodeSchema(&pBuf, &schemaWrapper);
vLen = POINTER_DISTANCE(pBuf, buf);
ret = tdbDbInsert(pMetaDb->pSchemaDB, pKey, kLen, pVal, vLen);
if (ret < 0) {
return -1;
}
}
// update name.idx
int nameLen = strlen(pTbCfg->name);
memcpy(buf, pTbCfg->name, nameLen + 1);
((tb_uid_t *)(buf + nameLen + 1))[0] = uid;
pKey = buf;
kLen = nameLen + 1 + sizeof(uid);
pVal = NULL;
vLen = 0;
ret = tdbDbInsert(pMetaDb->pNameIdx, pKey, kLen, pVal, vLen);
if (ret < 0) {
return -1;
}
// update other index
if (pTbCfg->type == META_SUPER_TABLE) {
pKey = &uid;
kLen = sizeof(uid);
pVal = NULL;
vLen = 0;
ret = tdbDbInsert(pMetaDb->pStbIdx, pKey, kLen, pVal, vLen);
if (ret < 0) {
return -1;
}
} else if (pTbCfg->type == META_CHILD_TABLE) {
ctbIdxKey.suid = pTbCfg->ctbCfg.suid;
ctbIdxKey.uid = uid;
pKey = &ctbIdxKey;
kLen = sizeof(ctbIdxKey);
pVal = NULL;
vLen = 0;
ret = tdbDbInsert(pMetaDb->pCtbIdx, pKey, kLen, pVal, vLen);
if (ret < 0) {
return -1;
}
} else if (pTbCfg->type == META_NORMAL_TABLE) {
pKey = &uid;
kLen = sizeof(uid);
pVal = NULL;
vLen = 0;
ret = tdbDbInsert(pMetaDb->pNtbIdx, pKey, kLen, pVal, vLen);
if (ret < 0) {
return -1;
}
}
return 0;
}
int metaRemoveTableFromDb(SMeta *pMeta, tb_uid_t uid) {
// TODO
ASSERT(0);
return 0;
}
STbCfg *metaGetTbInfoByUid(SMeta *pMeta, tb_uid_t uid) {
// TODO
return NULL;
int ret;
SMetaDB *pMetaDb = pMeta->pDB;
void *pKey;
void *pVal;
int kLen;
int vLen;
STbCfg *pTbCfg;
// Fetch
pKey = &uid;
kLen = sizeof(uid);
pVal = NULL;
ret = tdbDbGet(pMetaDb->pTbDB, pKey, kLen, &pVal, &vLen);
if (ret < 0) {
return NULL;
}
// Decode
pTbCfg = taosMemoryMalloc(sizeof(*pTbCfg));
metaDecodeTbInfo(pVal, pTbCfg);
TDB_FREE(pVal);
return pTbCfg;
}
STbCfg *metaGetTbInfoByName(SMeta *pMeta, char *tbname, tb_uid_t *uid) {
// TODO
return NULL;
void *pKey;
void *pVal;
void *ppKey;
int pkLen;
int kLen;
int vLen;
int ret;
pKey = tbname;
kLen = strlen(tbname) + 1;
pVal = NULL;
ppKey = NULL;
ret = tdbDbPGet(pMeta->pDB->pNameIdx, pKey, kLen, &ppKey, &pkLen, &pVal, &vLen);
if (ret < 0) {
return NULL;
}
ASSERT(pkLen == kLen + sizeof(uid));
*uid = *(tb_uid_t *)POINTER_SHIFT(ppKey, kLen);
TDB_FREE(ppKey);
TDB_FREE(pVal);
return metaGetTbInfoByUid(pMeta, *uid);
}
SSchemaWrapper *metaGetTableSchema(SMeta *pMeta, tb_uid_t uid, int32_t sver, bool isinline) {
// TODO
return NULL;
void *pKey;
void *pVal;
int kLen;
int vLen;
int ret;
SSchemaDbKey schemaDbKey;
SSchemaWrapper *pSchemaWrapper;
void *pBuf;
// fetch
schemaDbKey.uid = uid;
schemaDbKey.sver = sver;
pKey = &schemaDbKey;
kLen = sizeof(schemaDbKey);
pVal = NULL;
ret = tdbDbGet(pMeta->pDB->pSchemaDB, pKey, kLen, &pVal, &vLen);
if (ret < 0) {
return NULL;
}
// decode
pBuf = pVal;
pSchemaWrapper = taosMemoryMalloc(sizeof(*pSchemaWrapper));
metaDecodeSchema(pBuf, pSchemaWrapper);
TDB_FREE(pVal);
return pSchemaWrapper;
}
STSchema *metaGetTbTSchema(SMeta *pMeta, tb_uid_t uid, int32_t sver) {
// TODO
return NULL;
tb_uid_t quid;
SSchemaWrapper *pSW;
STSchemaBuilder sb;
SSchema *pSchema;
STSchema *pTSchema;
STbCfg *pTbCfg;
pTbCfg = metaGetTbInfoByUid(pMeta, uid);
if (pTbCfg->type == META_CHILD_TABLE) {
quid = pTbCfg->ctbCfg.suid;
} else {
quid = uid;
}
pSW = metaGetTableSchema(pMeta, quid, sver, true);
if (pSW == NULL) {
return NULL;
}
tdInitTSchemaBuilder(&sb, 0);
for (int 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;
}
struct SMTbCursor {
TDBC *pDbc;
};
SMTbCursor *metaOpenTbCursor(SMeta *pMeta) {
// TODO
return NULL;
SMTbCursor *pTbCur = NULL;
SMetaDB *pDB = pMeta->pDB;
pTbCur = (SMTbCursor *)taosMemoryCalloc(1, sizeof(*pTbCur));
if (pTbCur == NULL) {
return NULL;
}
tdbDbcOpen(pDB->pTbDB, &pTbCur->pDbc);
return pTbCur;
}
void metaCloseTbCursor(SMTbCursor *pTbCur) {
// TODO
if (pTbCur) {
if (pTbCur->pDbc) {
tdbDbcClose(pTbCur->pDbc);
}
taosMemoryFree(pTbCur);
}
}
char *metaTbCursorNext(SMTbCursor *pTbCur) {
// TODO
void *pKey = NULL;
void *pVal = NULL;
int kLen;
int vLen;
int ret;
void *pBuf;
STbCfg tbCfg;
for (;;) {
ret = tdbDbNext(pTbCur->pDbc, &pKey, &kLen, &pVal, &vLen);
if (ret < 0) break;
pBuf = pVal;
metaDecodeTbInfo(pBuf, &tbCfg);
if (tbCfg.type == META_SUPER_TABLE) {
taosMemoryFree(tbCfg.name);
taosMemoryFree(tbCfg.stbCfg.pTagSchema);
continue;
;
} else if (tbCfg.type == META_CHILD_TABLE) {
kvRowFree(tbCfg.ctbCfg.pTag);
}
return tbCfg.name;
}
return NULL;
}
struct SMCtbCursor {
TDBC *pCur;
tb_uid_t suid;
void *pKey;
void *pVal;
int kLen;
int vLen;
};
SMCtbCursor *metaOpenCtbCursor(SMeta *pMeta, tb_uid_t uid) {
// TODO
return NULL;
SMCtbCursor *pCtbCur = NULL;
SMetaDB *pDB = pMeta->pDB;
int ret;
pCtbCur = (SMCtbCursor *)taosMemoryCalloc(1, sizeof(*pCtbCur));
if (pCtbCur == NULL) {
return NULL;
}
pCtbCur->suid = uid;
ret = tdbDbcOpen(pDB->pCtbIdx, &pCtbCur->pCur);
if (ret < 0) {
taosMemoryFree(pCtbCur);
return NULL;
}
// TODO: move the cursor to the suid there
return pCtbCur;
}
void metaCloseCtbCurosr(SMCtbCursor *pCtbCur) {
// TODO
if (pCtbCur) {
if (pCtbCur->pCur) {
tdbDbcClose(pCtbCur->pCur);
TDB_FREE(pCtbCur->pKey);
TDB_FREE(pCtbCur->pVal);
}
taosMemoryFree(pCtbCur);
}
}
tb_uid_t metaCtbCursorNext(SMCtbCursor *pCtbCur) {
int ret;
SCtbIdxKey *pCtbIdxKey;
ret = tdbDbNext(pCtbCur->pCur, &pCtbCur->pKey, &pCtbCur->kLen, &pCtbCur->pVal, &pCtbCur->vLen);
if (ret < 0) {
return 0;
}
pCtbIdxKey = pCtbCur->pVal;
return pCtbIdxKey->uid;
}
int metaGetTbNum(SMeta *pMeta) {
// TODO
// ASSERT(0);
return 0;
}
}
STSmaWrapper *metaGetSmaInfoByTable(SMeta *pMeta, tb_uid_t uid) {
// TODO
ASSERT(0);
return NULL;
}
int metaRemoveSmaFromDb(SMeta *pMeta, int64_t indexUid) {
// TODO
ASSERT(0);
return 0;
}
int metaSaveSmaToDB(SMeta *pMeta, STSma *pSmaCfg) {
// TODO
ASSERT(0);
return 0;
}
STSma *metaGetSmaInfoByIndex(SMeta *pMeta, int64_t indexUid) {
// TODO
ASSERT(0);
return NULL;
}
const char *metaSmaCursorNext(SMSmaCursor *pCur) {
// TODO
ASSERT(0);
return NULL;
}
void metaCloseSmaCurosr(SMSmaCursor *pCur) {
// TODO
ASSERT(0);
}
SArray *metaGetSmaTbUids(SMeta *pMeta, bool isDup) {
// TODO
ASSERT(0);
return NULL;
}
SMSmaCursor *metaOpenSmaCursor(SMeta *pMeta, tb_uid_t uid) {
// TODO
ASSERT(0);
return NULL;
}
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 += taosEncodeFixedI16(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 *)taosMemoryMalloc(sizeof(SSchema) * pSW->nCols);
for (int i = 0; i < pSW->nCols; i++) {
pSchema = pSW->pSchema + i;
buf = taosDecodeFixedI8(buf, &pSchema->type);
buf = taosDecodeFixedI16(buf, &pSchema->colId);
buf = taosDecodeFixedI32(buf, &pSchema->bytes);
buf = taosDecodeStringTo(buf, pSchema->name);
}
return buf;
}
static int metaEncodeTbInfo(void **buf, STbCfg *pTbCfg) {
int tsize = 0;
tsize += taosEncodeString(buf, pTbCfg->name);
tsize += taosEncodeFixedU32(buf, pTbCfg->ttl);
tsize += taosEncodeFixedU32(buf, pTbCfg->keep);
tsize += taosEncodeFixedU8(buf, pTbCfg->info);
if (pTbCfg->type == META_SUPER_TABLE) {
SSchemaWrapper sw = {.nCols = pTbCfg->stbCfg.nTagCols, .pSchema = pTbCfg->stbCfg.pTagSchema};
tsize += metaEncodeSchema(buf, &sw);
} 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) {
// TODO
} else {
ASSERT(0);
}
return tsize;
}
static void *metaDecodeTbInfo(void *buf, STbCfg *pTbCfg) {
buf = taosDecodeString(buf, &(pTbCfg->name));
buf = taosDecodeFixedU32(buf, &(pTbCfg->ttl));
buf = taosDecodeFixedU32(buf, &(pTbCfg->keep));
buf = taosDecodeFixedU8(buf, &(pTbCfg->info));
if (pTbCfg->type == META_SUPER_TABLE) {
SSchemaWrapper sw;
buf = metaDecodeSchema(buf, &sw);
pTbCfg->stbCfg.nTagCols = sw.nCols;
pTbCfg->stbCfg.pTagSchema = sw.pSchema;
} 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) {
// TODO
} else {
ASSERT(0);
}
return buf;
}

3
source/libs/tdb/CMakeLists.txt
View File

@ -9,9 +9,8 @@ target_sources(tdb
"src/db/tdbDb.c"
"src/db/tdbEnv.c"
"src/db/tdbTxn.c"
"src/db/tdbPage.c"
"src/db/tdbOs.c"
"src/page/tdbPage.c"
"src/page/tdbPageL.c"
)
target_include_directories(

56
source/libs/tdb/src/db/tdbBtree.c
View File

@ -30,7 +30,6 @@ struct SBTree {
int valLen;
SPager *pPager;
FKeyComparator kcmpr;
u8 fanout;
int pageSize;
int maxLocal;
int minLocal;
@ -85,6 +84,8 @@ int tdbBtreeOpen(int keyLen, int valLen, SPager *pPager, FKeyComparator kcmpr, S
SBTree *pBt;
int ret;
ASSERT(keyLen != 0);
*ppBt = NULL;
pBt = (SBTree *)tdbOsCalloc(1, sizeof(*pBt));
@ -100,21 +101,14 @@ int tdbBtreeOpen(int keyLen, int valLen, SPager *pPager, FKeyComparator kcmpr, S
pBt->pPager = pPager;
// pBt->kcmpr
pBt->kcmpr = kcmpr ? kcmpr : tdbDefaultKeyCmprFn;
// pBt->fanout
if (keyLen == TDB_VARIANT_LEN) {
pBt->fanout = TDB_DEFAULT_FANOUT;
} else {
ASSERT(0);
// TODO: pBt->fanout = 0;
}
// pBt->pageSize
pBt->pageSize = tdbPagerGetPageSize(pPager);
// pBt->maxLocal
pBt->maxLocal = (pBt->pageSize - 14) / pBt->fanout;
pBt->maxLocal = tdbPageCapacity(pBt->pageSize, sizeof(SIntHdr)) / 4;
// pBt->minLocal: Should not be allowed smaller than 15, which is [nPayload][nKey][nData]
pBt->minLocal = (pBt->pageSize - 14) / pBt->fanout / 2;
pBt->minLocal = pBt->maxLocal / 2;
// pBt->maxLeaf
pBt->maxLeaf = pBt->pageSize - 14;
pBt->maxLeaf = tdbPageCapacity(pBt->pageSize, sizeof(SLeafHdr));
// pBt->minLeaf
pBt->minLeaf = pBt->minLocal;
@ -226,6 +220,43 @@ int tdbBtreeGet(SBTree *pBt, const void *pKey, int kLen, void **ppVal, int *vLen
return 0;
}
int tdbBtreePGet(SBTree *pBt, const void *pKey, int kLen, void **ppKey, int *pkLen, void **ppVal, int *vLen) {
SBTC btc;
SCell *pCell;
int cret;
void *pTKey;
void *pTVal;
SCellDecoder cd;
tdbBtcOpen(&btc, pBt);
tdbBtCursorMoveTo(&btc, pKey, kLen, &cret);
if (cret) {
return cret;
}
pCell = tdbPageGetCell(btc.pPage, btc.idx);
tdbBtreeDecodeCell(btc.pPage, pCell, &cd);
pTKey = TDB_REALLOC(*ppKey, cd.kLen);
pTVal = TDB_REALLOC(*ppVal, cd.vLen);
if (pTKey == NULL || pTVal == NULL) {
TDB_FREE(pTKey);
TDB_FREE(pTVal);
}
*ppKey = pTKey;
*ppVal = pTVal;
*pkLen = cd.kLen;
*vLen = cd.vLen;
memcpy(*ppKey, cd.pKey, cd.kLen);
memcpy(*ppVal, cd.pVal, cd.vLen);
return 0;
}
static int tdbBtCursorMoveTo(SBTC *pBtc, const void *pKey, int kLen, int *pCRst) {
int ret;
SBTree *pBt;
@ -929,7 +960,7 @@ static int tdbBtreeEncodeCell(SPage *pPage, const void *pKey, int kLen, const vo
}
// 2. Encode payload part
if (leaf) {
if (leaf && vLen > 0) {
ret = tdbBtreeEncodePayload(pPage, pCell + nHeader, pKey, kLen, pVal, vLen, &nPayload);
} else {
ret = tdbBtreeEncodePayload(pPage, pCell + nHeader, pKey, kLen, NULL, 0, &nPayload);
@ -968,6 +999,7 @@ static int tdbBtreeDecodePayload(SPage *pPage, const u8 *pPayload, SCellDecoder
return 0;
}
// TODO: here has problem
static int tdbBtreeDecodeCell(SPage *pPage, const SCell *pCell, SCellDecoder *pDecoder) {
u8 flags;
u8 leaf;

10
source/libs/tdb/src/db/tdbDb.c
View File

@ -96,6 +96,10 @@ int tdbDbGet(TDB *pDb, const void *pKey, int kLen, void **ppVal, int *vLen) {
return tdbBtreeGet(pDb->pBt, pKey, kLen, ppVal, vLen);
}
int tdbDbPGet(TDB *pDb, const void *pKey, int kLen, void **ppKey, int *pkLen, void **ppVal, int *vLen) {
return tdbBtreePGet(pDb->pBt, pKey, kLen, ppKey, pkLen, ppVal, vLen);
}
int tdbDbcOpen(TDB *pDb, TDBC **ppDbc) {
int ret;
TDBC *pDbc = NULL;
@ -129,5 +133,11 @@ int tdbDbcClose(TDBC *pDbc) {
tdbOsFree(pDbc);
}
return 0;
}
int tdbDbcInsert(TDBC *pDbc, const void *pKey, int keyLen, const void *pVal, int valLen) {
// TODO
ASSERT(0);
return 0;
}

2
source/libs/tdb/src/db/tdbPCache.c
View File

@ -122,7 +122,7 @@ static void tdbPCacheClearLock(SPCache *pCache) { tdbMutexDestroy(&(pCache->mute
static void tdbPCacheLock(SPCache *pCache) { tdbMutexLock(&(pCache->mutex)); }
static void tdbPCacheUnlock(SPCache *pCache) { tdbMutexDestroy(&(pCache->mutex)); }
static void tdbPCacheUnlock(SPCache *pCache) { tdbMutexUnlock(&(pCache->mutex)); }
static bool tdbPCacheLocked(SPCache *pCache) {
assert(0);

89
source/libs/tdb/src/page/tdbPage.c → source/libs/tdb/src/db/tdbPage.c
View File

@ -242,6 +242,18 @@ void tdbPageCopy(SPage *pFromPage, SPage *pToPage) {
pToPage->nOverflow = pFromPage->nOverflow;
}
int tdbPageCapacity(int pageSize, int amHdrSize) {
int szPageHdr;
if (pageSize < 65536) {
szPageHdr = pageMethods.szPageHdr;
} else {
szPageHdr = pageLargeMethods.szPageHdr;
}
return pageSize - szPageHdr - amHdrSize;
}
static int tdbPageAllocate(SPage *pPage, int szCell, SCell **ppCell) {
SCell *pFreeCell;
u8 *pOffset;
@ -503,4 +515,81 @@ SPageMethods pageMethods = {
setPageCellOffset, // setCellOffset
getPageFreeCellInfo, // getFreeCellInfo
setPageFreeCellInfo // setFreeCellInfo
};
typedef struct __attribute__((__packed__)) {
u8 cellNum[3];
u8 cellBody[3];
u8 cellFree[3];
u8 nFree[3];
} SPageHdrL;
typedef struct __attribute__((__packed__)) {
u8 szCell[3];
u8 nxOffset[3];
} SFreeCellL;
// cellNum
static inline int getLPageCellNum(SPage *pPage) { return TDB_GET_U24(((SPageHdrL *)(pPage->pPageHdr))[0].cellNum); }
static inline void setLPageCellNum(SPage *pPage, int cellNum) {
TDB_PUT_U24(((SPageHdrL *)(pPage->pPageHdr))[0].cellNum, cellNum);
}
// cellBody
static inline int getLPageCellBody(SPage *pPage) { return TDB_GET_U24(((SPageHdrL *)(pPage->pPageHdr))[0].cellBody); }
static inline void setLPageCellBody(SPage *pPage, int cellBody) {
TDB_PUT_U24(((SPageHdrL *)(pPage->pPageHdr))[0].cellBody, cellBody);
}
// cellFree
static inline int getLPageCellFree(SPage *pPage) { return TDB_GET_U24(((SPageHdrL *)(pPage->pPageHdr))[0].cellFree); }
static inline void setLPageCellFree(SPage *pPage, int cellFree) {
TDB_PUT_U24(((SPageHdrL *)(pPage->pPageHdr))[0].cellFree, cellFree);
}
// nFree
static inline int getLPageNFree(SPage *pPage) { return TDB_GET_U24(((SPageHdrL *)(pPage->pPageHdr))[0].nFree); }
static inline void setLPageNFree(SPage *pPage, int nFree) {
TDB_PUT_U24(((SPageHdrL *)(pPage->pPageHdr))[0].nFree, nFree);
}
// cell offset
static inline int getLPageCellOffset(SPage *pPage, int idx) {
ASSERT(idx >= 0 && idx < getPageCellNum(pPage));
return TDB_GET_U24(pPage->pCellIdx + 3 * idx);
}
static inline void setLPageCellOffset(SPage *pPage, int idx, int offset) {
TDB_PUT_U24(pPage->pCellIdx + 3 * idx, offset);
}
// free cell info
static inline void getLPageFreeCellInfo(SCell *pCell, int *szCell, int *nxOffset) {
SFreeCellL *pFreeCell = (SFreeCellL *)pCell;
*szCell = TDB_GET_U24(pFreeCell->szCell);
*nxOffset = TDB_GET_U24(pFreeCell->nxOffset);
}
static inline void setLPageFreeCellInfo(SCell *pCell, int szCell, int nxOffset) {
SFreeCellL *pFreeCell = (SFreeCellL *)pCell;
TDB_PUT_U24(pFreeCell->szCell, szCell);
TDB_PUT_U24(pFreeCell->nxOffset, nxOffset);
}
SPageMethods pageLargeMethods = {
3, // szOffset
sizeof(SPageHdrL), // szPageHdr
sizeof(SFreeCellL), // szFreeCell
getLPageCellNum, // getCellNum
setLPageCellNum, // setCellNum
getLPageCellBody, // getCellBody
setLPageCellBody, // setCellBody
getLPageCellFree, // getCellFree
setLPageCellFree, // setCellFree
getLPageNFree, // getFreeBytes
setLPageNFree, // setFreeBytes
getLPageCellOffset, // getCellOffset
setLPageCellOffset, // setCellOffset
getLPageFreeCellInfo, // getFreeCellInfo
setLPageFreeCellInfo // setFreeCellInfo
};

1
source/libs/tdb/src/inc/tdbBtree.h
View File

@ -43,6 +43,7 @@ int tdbBtreeOpen(int keyLen, int valLen, SPager *pFile, FKeyComparator kcmpr, SB
int tdbBtreeClose(SBTree *pBt);
int tdbBtCursorInsert(SBTC *pCur, const void *pKey, int kLen, const void *pVal, int vLen);
int tdbBtreeGet(SBTree *pBt, const void *pKey, int kLen, void **ppVal, int *vLen);
int tdbBtreePGet(SBTree *pBt, const void *pKey, int kLen, void **ppKey, int *pkLen, void **ppVal, int *vLen);
// SBTC
int tdbBtcOpen(SBTC *pCur, SBTree *pBt);

2
source/libs/tdb/src/inc/tdbDb.h
View File

@ -29,11 +29,13 @@ int tdbDbClose(TDB *pDb);
int tdbDbDrop(TDB *pDb);
int tdbDbInsert(TDB *pDb, const void *pKey, int keyLen, const void *pVal, int valLen);
int tdbDbGet(TDB *pDb, const void *pKey, int kLen, void **ppVal, int *vLen);
int tdbDbPGet(TDB *pDb, const void *pKey, int kLen, void **ppKey, int *pkLen, void **ppVal, int *vLen);
// TDBC
int tdbDbcOpen(TDB *pDb, TDBC **ppDbc);
int tdbDbNext(TDBC *pDbc, void **ppKey, int *kLen, void **ppVal, int *vLen);
int tdbDbcClose(TDBC *pDbc);
int tdbDbcInsert(TDBC *pDbc, const void *pKey, int keyLen, const void *pVal, int valLen);
#ifdef __cplusplus
}

1
source/libs/tdb/src/inc/tdbPage.h
View File

@ -111,6 +111,7 @@ void tdbPageInit(SPage *pPage, u8 szAmHdr, int (*xCellSize)(const SPage *, SCell
int tdbPageInsertCell(SPage *pPage, int idx, SCell *pCell, int szCell, u8 asOvfl);
int tdbPageDropCell(SPage *pPage, int idx);
void tdbPageCopy(SPage *pFromPage, SPage *pToPage);
int tdbPageCapacity(int pageSize, int amHdrSize);
static inline SCell *tdbPageGetCell(SPage *pPage, int idx) {
SCell *pCell;

2
source/libs/tdb/src/inc/tdbTxn.h
View File

@ -20,7 +20,7 @@
extern "C" {
#endif
typedef struct STxn STXN;
typedef struct STxn TXN;
struct STxn {
u64 txnId;

93
source/libs/tdb/src/page/tdbPageL.c
View File

@ -1,93 +0,0 @@
/*
* 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/>.
*/
#include "tdbInt.h"
typedef struct __attribute__((__packed__)) {
u8 cellNum[3];
u8 cellBody[3];
u8 cellFree[3];
u8 nFree[3];
} SPageHdrL;
typedef struct __attribute__((__packed__)) {
u8 szCell[3];
u8 nxOffset[3];
} SFreeCellL;
// cellNum
static inline int getPageCellNum(SPage *pPage) { return TDB_GET_U24(((SPageHdrL *)(pPage->pPageHdr))[0].cellNum); }
static inline void setPageCellNum(SPage *pPage, int cellNum) {
TDB_PUT_U24(((SPageHdrL *)(pPage->pPageHdr))[0].cellNum, cellNum);
}
// cellBody
static inline int getPageCellBody(SPage *pPage) { return TDB_GET_U24(((SPageHdrL *)(pPage->pPageHdr))[0].cellBody); }
static inline void setPageCellBody(SPage *pPage, int cellBody) {
TDB_PUT_U24(((SPageHdrL *)(pPage->pPageHdr))[0].cellBody, cellBody);
}
// cellFree
static inline int getPageCellFree(SPage *pPage) { return TDB_GET_U24(((SPageHdrL *)(pPage->pPageHdr))[0].cellFree); }
static inline void setPageCellFree(SPage *pPage, int cellFree) {
TDB_PUT_U24(((SPageHdrL *)(pPage->pPageHdr))[0].cellFree, cellFree);
}
// nFree
static inline int getPageNFree(SPage *pPage) { return TDB_GET_U24(((SPageHdrL *)(pPage->pPageHdr))[0].nFree); }
static inline void setPageNFree(SPage *pPage, int nFree) {
TDB_PUT_U24(((SPageHdrL *)(pPage->pPageHdr))[0].nFree, nFree);
}
// cell offset
static inline int getPageCellOffset(SPage *pPage, int idx) {
ASSERT(idx >= 0 && idx < getPageCellNum(pPage));
return TDB_GET_U24(pPage->pCellIdx + 3 * idx);
}
static inline void setPageCellOffset(SPage *pPage, int idx, int offset) {
TDB_PUT_U24(pPage->pCellIdx + 3 * idx, offset);
}
// free cell info
static inline void getPageFreeCellInfo(SCell *pCell, int *szCell, int *nxOffset) {
SFreeCellL *pFreeCell = (SFreeCellL *)pCell;
*szCell = TDB_GET_U24(pFreeCell->szCell);
*nxOffset = TDB_GET_U24(pFreeCell->nxOffset);
}
static inline void setPageFreeCellInfo(SCell *pCell, int szCell, int nxOffset) {
SFreeCellL *pFreeCell = (SFreeCellL *)pCell;
TDB_PUT_U24(pFreeCell->szCell, szCell);
TDB_PUT_U24(pFreeCell->nxOffset, nxOffset);
}
SPageMethods pageLargeMethods = {
3, // szOffset
sizeof(SPageHdrL), // szPageHdr
sizeof(SFreeCellL), // szFreeCell
getPageCellNum, // getCellNum
setPageCellNum, // setCellNum
getPageCellBody, // getCellBody
setPageCellBody, // setCellBody
getPageCellFree, // getCellFree
setPageCellFree, // setCellFree
getPageNFree, // getFreeBytes
setPageNFree, // setFreeBytes
getPageCellOffset, // getCellOffset
setPageCellOffset, // setCellOffset
getPageFreeCellInfo, // getFreeCellInfo
setPageFreeCellInfo // setFreeCellInfo
};