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

more code

This commit is contained in:
Hongze Cheng 2022-09-22 13:49:34 +08:00
parent 0e5b0e59bf
commit a3cedbf1bc
2 changed files with 77 additions and 67 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

82
source/dnode/vnode/src/inc/tsdb.h
View File

@ -38,41 +38,42 @@ extern "C" {
goto LABEL; \
}
typedef struct TSDBROW TSDBROW;
typedef struct TABLEID TABLEID;
typedef struct TSDBKEY TSDBKEY;
typedef struct SDelData SDelData;
typedef struct SDelIdx SDelIdx;
typedef struct STbData STbData;
typedef struct SMemTable SMemTable;
typedef struct STbDataIter STbDataIter;
typedef struct SMapData SMapData;
typedef struct SBlockIdx SBlockIdx;
typedef struct SDataBlk SDataBlk;
typedef struct SSttBlk SSttBlk;
typedef struct SDiskDataHdr SDiskDataHdr;
typedef struct SBlockData SBlockData;
typedef struct SDelFile SDelFile;
typedef struct SHeadFile SHeadFile;
typedef struct SDataFile SDataFile;
typedef struct SSttFile SSttFile;
typedef struct SSmaFile SSmaFile;
typedef struct SDFileSet SDFileSet;
typedef struct SDataFWriter SDataFWriter;
typedef struct SDataFReader SDataFReader;
typedef struct SDelFWriter SDelFWriter;
typedef struct SDelFReader SDelFReader;
typedef struct SRowIter SRowIter;
typedef struct STsdbFS STsdbFS;
typedef struct SRowMerger SRowMerger;
typedef struct STsdbReadSnap STsdbReadSnap;
typedef struct SBlockInfo SBlockInfo;
typedef struct SSmaInfo SSmaInfo;
typedef struct SBlockCol SBlockCol;
typedef struct SVersionRange SVersionRange;
typedef struct SLDataIter SLDataIter;
typedef struct SDiskCol SDiskCol;
typedef struct SDiskData SDiskData;
typedef struct TSDBROW TSDBROW;
typedef struct TABLEID TABLEID;
typedef struct TSDBKEY TSDBKEY;
typedef struct SDelData SDelData;
typedef struct SDelIdx SDelIdx;
typedef struct STbData STbData;
typedef struct SMemTable SMemTable;
typedef struct STbDataIter STbDataIter;
typedef struct SMapData SMapData;
typedef struct SBlockIdx SBlockIdx;
typedef struct SDataBlk SDataBlk;
typedef struct SSttBlk SSttBlk;
typedef struct SDiskDataHdr SDiskDataHdr;
typedef struct SBlockData SBlockData;
typedef struct SDelFile SDelFile;
typedef struct SHeadFile SHeadFile;
typedef struct SDataFile SDataFile;
typedef struct SSttFile SSttFile;
typedef struct SSmaFile SSmaFile;
typedef struct SDFileSet SDFileSet;
typedef struct SDataFWriter SDataFWriter;
typedef struct SDataFReader SDataFReader;
typedef struct SDelFWriter SDelFWriter;
typedef struct SDelFReader SDelFReader;
typedef struct SRowIter SRowIter;
typedef struct STsdbFS STsdbFS;
typedef struct SRowMerger SRowMerger;
typedef struct STsdbReadSnap STsdbReadSnap;
typedef struct SBlockInfo SBlockInfo;
typedef struct SSmaInfo SSmaInfo;
typedef struct SBlockCol SBlockCol;
typedef struct SVersionRange SVersionRange;
typedef struct SLDataIter SLDataIter;
typedef struct SDiskCol SDiskCol;
typedef struct SDiskData SDiskData;
typedef struct SDiskDataBuilder SDiskDataBuilder;
#define TSDB_FILE_DLMT ((uint32_t)0xF00AFA0F)
#define TSDB_MAX_SUBBLOCKS 8
@ -302,7 +303,7 @@ int32_t tsdbMerge(STsdb *pTsdb);
#define TSDB_CACHE_LAST_ROW(c) (((c).cacheLast & 1) > 0)
#define TSDB_CACHE_LAST(c) (((c).cacheLast & 2) > 0)
// tsdbCache
// tsdbCache ==============================================================================================
int32_t tsdbOpenCache(STsdb *pTsdb);
void tsdbCloseCache(STsdb *pTsdb);
int32_t tsdbCacheInsertLast(SLRUCache *pCache, tb_uid_t uid, STSRow *row, STsdb *pTsdb);
@ -320,6 +321,15 @@ size_t tsdbCacheGetCapacity(SVnode *pVnode);
int32_t tsdbCacheLastArray2Row(SArray *pLastArray, STSRow **ppRow, STSchema *pSchema);
// tsdbDiskData ==============================================================================================
int32_t tDiskDataBuilderCreate(SDiskDataBuilder **ppBuilder);
void *tDiskDataBuilderDestroy(SDiskDataBuilder *pBuilder);
int32_t tDiskDataBuilderInit(SDiskDataBuilder *pBuilder, STSchema *pTSchema, TABLEID *pId, uint8_t cmprAlg,
uint8_t calcSma);
int32_t tDiskDataBuilderAddRow(SDiskDataBuilder *pBuilder, TSDBROW *pRow, STSchema *pTSchema, TABLEID *pId);
int32_t tGnrtDiskData(SDiskDataBuilder *pBuilder, SDiskData *pDiskData);
int32_t tDiskDataDestroy(SDiskData *pDiskData);
// structs =======================
struct STsdbFS {
SDelFile *pDelFile;

62
source/dnode/vnode/src/tsdb/tsdbDiskData.c
View File

@ -15,8 +15,7 @@
#include "tsdb.h"
typedef struct SDiskDataBuilder SDiskDataBuilder;
typedef struct SDiskColBuilder SDiskColBuilder;
typedef struct SDiskColBuilder SDiskColBuilder;
struct SDiskColBuilder {
int16_t cid;
@ -45,6 +44,19 @@ struct SDiskDataBuilder {
};
// SDiskColBuilder ================================================
#define tDiskColBuilderCreate() \
(SDiskColBuilder) { 0 }
static int32_t tDiskColBuilderDestroy(SDiskColBuilder *pBuilder) {
int32_t code = 0;
tFree(pBuilder->pBitMap);
if (pBuilder->pOffC) tCompressorDestroy(pBuilder->pOffC);
if (pBuilder->pValC) tCompressorDestroy(pBuilder->pValC);
return code;
}
static int32_t tDiskColBuilderInit(SDiskColBuilder *pBuilder, int16_t cid, int8_t type, uint8_t cmprAlg) {
int32_t code = 0;
@ -74,16 +86,6 @@ static int32_t tDiskColBuilderInit(SDiskColBuilder *pBuilder, int16_t cid, int8_
return code;
}
static int32_t tDiskColBuilderClear(SDiskColBuilder *pBuilder) {
int32_t code = 0;
tFree(pBuilder->pBitMap);
if (pBuilder->pOffC) tCompressorDestroy(pBuilder->pOffC);
if (pBuilder->pValC) tCompressorDestroy(pBuilder->pValC);
return code;
}
static int32_t tGnrtDiskCol(SDiskColBuilder *pBuilder, SDiskCol *pDiskCol) {
int32_t code = 0;
@ -419,7 +421,7 @@ void *tDiskDataBuilderDestroy(SDiskDataBuilder *pBuilder) {
if (pBuilder->aBuilder) {
for (int32_t iBuilder = 0; iBuilder < taosArrayGetSize(pBuilder->aBuilder); iBuilder++) {
SDiskColBuilder *pDCBuilder = (SDiskColBuilder *)taosArrayGet(pBuilder->aBuilder, iBuilder);
tDiskColBuilderClear(pDCBuilder);
tDiskColBuilderDestroy(pDCBuilder);
}
taosArrayDestroy(pBuilder->aBuilder);
}
@ -466,7 +468,7 @@ int32_t tDiskDataBuilderInit(SDiskDataBuilder *pBuilder, STSchema *pTSchema, TAB
STColumn *pTColumn = &pTSchema->columns[iCol];
if (pBuilder->nBuilder >= taosArrayGetSize(pBuilder->aBuilder)) {
SDiskColBuilder dc = {0};
SDiskColBuilder dc = tDiskColBuilderCreate();
if (taosArrayPush(pBuilder->aBuilder, &dc) == NULL) {
code = TSDB_CODE_OUT_OF_MEMORY;
return code;
@ -491,51 +493,51 @@ int32_t tDiskDataBuilderAddRow(SDiskDataBuilder *pBuilder, TSDBROW *pRow, STSche
// uid
if (pBuilder->uid && pBuilder->uid != pId->uid) {
ASSERT(!pBuilder->calcSma);
for (int32_t iRow = 0; iRow < pBuilder->nRow; iRow++) {
code = tCompress(pBuilder->pUidC, &pBuilder->uid, sizeof(int64_t));
if (code) goto _exit;
if (code) return code;
}
pBuilder->uid = 0;
}
if (pBuilder->uid == 0) {
code = tCompress(pBuilder->pUidC, &pId->uid, sizeof(int64_t));
if (code) goto _exit;
if (code) return code;
}
// version
int64_t version = TSDBROW_VERSION(pRow);
code = tCompress(pBuilder->pVerC, &version, sizeof(int64_t));
if (code) goto _exit;
if (code) return code;
// TSKEY
TSKEY ts = TSDBROW_TS(pRow);
code = tCompress(pBuilder->pKeyC, &ts, sizeof(int64_t));
if (code) goto _exit;
if (code) return code;
SRowIter iter = {0};
tRowIterInit(&iter, pRow, pTSchema);
SColVal *pColVal = tRowIterNext(&iter);
for (int32_t iDiskColBuilder = 0; iDiskColBuilder < pBuilder->nBuilder; iDiskColBuilder++) {
SDiskColBuilder *pDiskColBuilder = (SDiskColBuilder *)taosArrayGet(pBuilder->aBuilder, iDiskColBuilder);
for (int32_t iBuilder = 0; iBuilder < pBuilder->nBuilder; iBuilder++) {
SDiskColBuilder *pDCBuilder = (SDiskColBuilder *)taosArrayGet(pBuilder->aBuilder, iBuilder);
while (pColVal && pColVal->cid < pDiskColBuilder->cid) {
while (pColVal && pColVal->cid < pDCBuilder->cid) {
pColVal = tRowIterNext(&iter);
}
if (pColVal == NULL || pColVal->cid > pDiskColBuilder->cid) {
SColVal cv = COL_VAL_NONE(pDiskColBuilder->cid, pDiskColBuilder->type);
code = tDiskColAddValImpl[pDiskColBuilder->flag](pDiskColBuilder, &cv);
if (code) goto _exit;
if (pColVal == NULL || pColVal->cid > pDCBuilder->cid) {
SColVal cv = COL_VAL_NONE(pDCBuilder->cid, pDCBuilder->type);
code = tDiskColAddValImpl[pDCBuilder->flag](pDCBuilder, &cv);
if (code) return code;
} else {
code = tDiskColAddValImpl[pDiskColBuilder->flag](pDiskColBuilder, pColVal);
if (code) goto _exit;
code = tDiskColAddValImpl[pDCBuilder->flag](pDCBuilder, pColVal);
if (code) return code;
pColVal = tRowIterNext(&iter);
}
}
pBuilder->nRow++;
_exit:
return code;
}
@ -601,9 +603,7 @@ int32_t tGnrtDiskData(SDiskDataBuilder *pBuilder, SDiskData *pDiskData) {
return code;
}
if (pDCBuilder->flag != HAS_NULL) {
pDiskData->hdr.szBlkCol += tPutBlockCol(NULL, &dCol.bCol);
}
pDiskData->hdr.szBlkCol += tPutBlockCol(NULL, &dCol.bCol);
}
return code;