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 2022-01-10 05:30:20 +00:00
parent 8d8b7b1c00
commit a511aed8d5
10 changed files with 47 additions and 9 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
include/dnode/vnode/meta/meta.h
View File

@ -58,6 +58,7 @@ int metaCommit(SMeta *pMeta);
STbCfg * metaGetTbInfoByUid(SMeta *pMeta, tb_uid_t uid);
STbCfg * metaGetTbInfoByName(SMeta *pMeta, char *tbname, tb_uid_t *uid);
SSchemaWrapper *metaGetTableSchema(SMeta *pMeta, tb_uid_t uid, int32_t sver, bool isinline);
STSchema * metaGetTbTSchema(SMeta *pMeta, tb_uid_t uid, int32_t sver);
SMTbCursor *metaOpenTbCursor(SMeta *pMeta);
void metaCloseTbCursor(SMTbCursor *pTbCur);

3
include/dnode/vnode/tsdb/tsdb.h
View File

@ -17,6 +17,7 @@
#define _TD_TSDB_H_
#include "mallocator.h"
#include "meta.h"
#ifdef __cplusplus
extern "C" {
@ -58,7 +59,7 @@ typedef struct STsdbCfg {
} STsdbCfg;
// STsdb
STsdb *tsdbOpen(const char *path, int32_t vgId, const STsdbCfg *pTsdbCfg, SMemAllocatorFactory *pMAF);
STsdb *tsdbOpen(const char *path, int32_t vgId, const STsdbCfg *pTsdbCfg, SMemAllocatorFactory *pMAF, SMeta *pMeta);
void tsdbClose(STsdb *);
void tsdbRemove(const char *path);
int tsdbInsertData(STsdb *pTsdb, SSubmitMsg *pMsg, SSubmitRsp *pRsp);

2
source/dnode/vnode/impl/src/vnodeMain.c
View File

@ -106,7 +106,7 @@ static int vnodeOpenImpl(SVnode *pVnode) {
// Open tsdb
sprintf(dir, "%s/tsdb", pVnode->path);
pVnode->pTsdb = tsdbOpen(dir, pVnode->vgId, &(pVnode->config.tsdbCfg), vBufPoolGetMAF(pVnode));
pVnode->pTsdb = tsdbOpen(dir, pVnode->vgId, &(pVnode->config.tsdbCfg), vBufPoolGetMAF(pVnode), pVnode->pMeta);
if (pVnode->pTsdb == NULL) {
// TODO: handle error
return -1;

23
source/dnode/vnode/meta/src/metaBDBImpl.c
View File

@ -590,3 +590,26 @@ char *metaTbCursorNext(SMTbCursor *pTbCur) {
}
}
}
STSchema *metaGetTbTSchema(SMeta *pMeta, tb_uid_t uid, int32_t sver) {
STSchemaBuilder sb;
STSchema * pTSchema = NULL;
SSchema * pSchema;
SSchemaWrapper *pSW;
pSW = metaGetTableSchema(pMeta, uid, sver, true);
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;
}

1
source/dnode/vnode/tsdb/CMakeLists.txt
View File

@ -29,4 +29,5 @@ target_link_libraries(
PUBLIC common
PUBLIC tkv
PUBLIC tfs
PUBLIC meta
)

2
source/dnode/vnode/tsdb/inc/tsdbDef.h
View File

@ -17,6 +17,7 @@
#define _TD_TSDB_DEF_H_
#include "mallocator.h"
#include "meta.h"
#include "tcompression.h"
#include "tglobal.h"
#include "thash.h"
@ -48,6 +49,7 @@ struct STsdb {
SRtn rtn;
SMemAllocatorFactory *pmaf;
STsdbFS * fs;
SMeta * pMeta;
};
#define REPO_ID(r) ((r)->vgId)

4
source/dnode/vnode/tsdb/inc/tsdbFS.h
View File

@ -70,7 +70,7 @@ typedef struct {
#define TSDB_FS_ITER_FORWARD TSDB_ORDER_ASC
#define TSDB_FS_ITER_BACKWARD TSDB_ORDER_DESC
STsdbFS *tsdbNewFS(STsdbCfg *pCfg);
STsdbFS *tsdbNewFS(const STsdbCfg *pCfg);
void * tsdbFreeFS(STsdbFS *pfs);
int tsdbOpenFS(STsdb *pRepo);
void tsdbCloseFS(STsdb *pRepo);
@ -79,7 +79,7 @@ int tsdbEndFSTxn(STsdb *pRepo);
int tsdbEndFSTxnWithError(STsdbFS *pfs);
void tsdbUpdateFSTxnMeta(STsdbFS *pfs, STsdbFSMeta *pMeta);
// void tsdbUpdateMFile(STsdbFS *pfs, const SMFile *pMFile);
int tsdbUpdateDFileSet(STsdbFS *pfs, const SDFileSet *pSet);
int tsdbUpdateDFileSet(STsdbFS *pfs, const SDFileSet *pSet);
void tsdbFSIterInit(SFSIter *pIter, STsdbFS *pfs, int direction);
void tsdbFSIterSeek(SFSIter *pIter, int fid);

7
source/dnode/vnode/tsdb/src/tsdbCommit.c
View File

@ -420,6 +420,13 @@ static int tsdbCreateCommitIters(SCommitH *pCommith) {
pCommitIter = pCommith->iters + i;
pCommitIter->pIter = tSkipListCreateIter(pTbData->pData);
tSkipListIterNext(pCommitIter->pIter);
#if 0
// TODO
pCommitIter->pTable = (STable *)malloc(sizeof(STable));
pCommitIter->pTable->uid = pTbData->uid;
pCommitIter->pTable->pSchema = metaGetTableSchema();
#endif
}
return 0;

2
source/dnode/vnode/tsdb/src/tsdbFS.c
View File

@ -191,7 +191,7 @@ static int tsdbAddDFileSetToStatus(SFSStatus *pStatus, const SDFileSet *pSet) {
}
// ================== STsdbFS
STsdbFS *tsdbNewFS(STsdbCfg *pCfg) {
STsdbFS *tsdbNewFS(const STsdbCfg *pCfg) {
int keep = pCfg->keep;
int days = pCfg->daysPerFile;
int maxFSet = TSDB_MAX_FSETS(keep, days);

11
source/dnode/vnode/tsdb/src/tsdbMain.c
View File

@ -15,12 +15,13 @@
#include "tsdbDef.h"
static STsdb *tsdbNew(const char *path, int32_t vgId, const STsdbCfg *pTsdbCfg, SMemAllocatorFactory *pMAF);
static STsdb *tsdbNew(const char *path, int32_t vgId, const STsdbCfg *pTsdbCfg, SMemAllocatorFactory *pMAF,
SMeta *pMeta);
static void tsdbFree(STsdb *pTsdb);
static int tsdbOpenImpl(STsdb *pTsdb);
static void tsdbCloseImpl(STsdb *pTsdb);
STsdb *tsdbOpen(const char *path, int32_t vgId, const STsdbCfg *pTsdbCfg, SMemAllocatorFactory *pMAF) {
STsdb *tsdbOpen(const char *path, int32_t vgId, const STsdbCfg *pTsdbCfg, SMemAllocatorFactory *pMAF, SMeta *pMeta) {
STsdb *pTsdb = NULL;
// Set default TSDB Options
@ -35,7 +36,7 @@ STsdb *tsdbOpen(const char *path, int32_t vgId, const STsdbCfg *pTsdbCfg, SMemAl
}
// Create the handle
pTsdb = tsdbNew(path, vgId, pTsdbCfg, pMAF);
pTsdb = tsdbNew(path, vgId, pTsdbCfg, pMAF, pMeta);
if (pTsdb == NULL) {
// TODO: handle error
return NULL;
@ -62,7 +63,8 @@ void tsdbClose(STsdb *pTsdb) {
void tsdbRemove(const char *path) { taosRemoveDir(path); }
/* ------------------------ STATIC METHODS ------------------------ */
static STsdb *tsdbNew(const char *path, int32_t vgId, const STsdbCfg *pTsdbCfg, SMemAllocatorFactory *pMAF) {
static STsdb *tsdbNew(const char *path, int32_t vgId, const STsdbCfg *pTsdbCfg, SMemAllocatorFactory *pMAF,
SMeta *pMeta) {
STsdb *pTsdb = NULL;
pTsdb = (STsdb *)calloc(1, sizeof(STsdb));
@ -75,6 +77,7 @@ static STsdb *tsdbNew(const char *path, int32_t vgId, const STsdbCfg *pTsdbCfg,
pTsdb->vgId = vgId;
tsdbOptionsCopy(&(pTsdb->config), pTsdbCfg);
pTsdb->pmaf = pMAF;
pTsdb->pMeta = pMeta;
pTsdb->fs = tsdbNewFS(pTsdbCfg);