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

TD-34

This commit is contained in:
hzcheng 2020-03-22 21:06:02 +08:00
parent a9cf4bd32e
commit 987708acce
4 changed files with 26 additions and 6 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
src/common/src/dataformat.c
View File

@ -150,7 +150,7 @@ void tdFreeSchema(STSchema *pSchema) {
*/
void tdUpdateSchema(STSchema *pSchema) {
STColumn *pCol = NULL;
int32_t offset = 0;
int32_t offset = TD_DATA_ROW_HEAD_SIZE;
for (int i = 0; i < schemaNCols(pSchema); i++) {
pCol = schemaColAt(pSchema, i);
colSetOffset(pCol, offset);

20
src/vnode/tsdb/src/tsdbMain.c
View File

@ -718,7 +718,7 @@ static int32_t tsdbInsertDataToTable(tsdb_repo_t *repo, SSubmitBlk *pBlock) {
return 0;
}
static int tsdbReadRowsFromCache(SSkipListIterator *pIter, TSKEY maxKey, int maxRowsToRead, void *dst) {
static int tsdbReadRowsFromCache(SSkipListIterator *pIter, TSKEY maxKey, int maxRowsToRead, SDataCol **cols, STSchema *pSchema) {
int numOfRows = 0;
do {
SSkipListNode *node = tSkipListIterGet(pIter);
@ -727,6 +727,11 @@ static int tsdbReadRowsFromCache(SSkipListIterator *pIter, TSKEY maxKey, int max
SDataRow row = SL_GET_NODE_DATA(node);
if (dataRowKey(row) > maxKey) break;
// Convert row data to column data
// for (int i = 0; i < schemaNCols(pSchema); i++) {
// STColumn *pCol = schemaColAt(pSchema, i);
// memcpy(cols[i]->data + TYPE_BYTES[colType(pCol)] * numOfRows, dataRowAt(row, pCol->offset),
// TYPE_BYTES[colType(pCol)]);
// }
numOfRows++;
if (numOfRows > maxRowsToRead) break;
@ -754,6 +759,8 @@ static void *tsdbCommitToFile(void *arg) {
int maxCols = pMeta->maxCols;
int maxBytes = pMeta->maxRowBytes;
SDataCol **cols = (SDataCol **)malloc(sizeof(SDataCol *) * maxCols);
void *buf = malloc((maxBytes + sizeof(SDataCol)) * pCfg->maxRowsPerFileBlock);
for (int fid = sfid; fid <= efid; fid++) {
TSKEY minKey = 0, maxKey = 0;
@ -771,9 +778,16 @@ static void *tsdbCommitToFile(void *arg) {
}
}
// Init row data part
cols[0] = (SDataCol *)buf;
for (int col = 1; col < schemaNCols(pTable->schema); col++) {
cols[col] = (SDataCol *)((char *)(cols[col - 1]) + sizeof(SDataCol) + colBytes(schemaColAt(pTable->schema, col-1)) * pCfg->maxRowsPerFileBlock);
}
// Loop the iterator
int rowsRead = 0;
while ((rowsRead = tsdbReadRowsFromCache(iters[tid], maxKey, pCfg->maxRowsPerFileBlock, NULL)) > 0) {
while ((rowsRead = tsdbReadRowsFromCache(iters[tid], maxKey, pCfg->maxRowsPerFileBlock, cols, pTable->schema)) >
0) {
int k = 0;
}
}
@ -784,6 +798,8 @@ static void *tsdbCommitToFile(void *arg) {
if (iters[tid] != NULL) tSkipListDestroyIter(iters[tid]);
}
free(buf);
free(cols);
free(iters);
return NULL;

4
src/vnode/tsdb/src/tsdbMeta.c
View File

@ -236,6 +236,10 @@ int32_t tsdbCreateTableImpl(STsdbMeta *pMeta, STableCfg *pCfg) {
table->type = TSDB_NORMAL_TABLE;
table->superUid = -1;
table->schema = tdDupSchema(pCfg->schema);
if (schemaNCols(table->schema) > pMeta->maxCols) pMeta->maxCols = schemaNCols(table->schema);
tdUpdateSchema(table->schema);
int bytes = tdMaxRowBytesFromSchema(table->schema);
if (bytes > pMeta->maxRowBytes) pMeta->maxRowBytes = bytes;
}
// Register to meta

6
src/vnode/tsdb/tests/tsdbTests.cpp
View File

@ -6,7 +6,7 @@
#include "tsdbFile.h"
#include "tsdbMeta.h"
TEST(TsdbTest, tableEncodeDecode) {
TEST(TsdbTest, DISABLED_tableEncodeDecode) {
STable *pTable = (STable *)malloc(sizeof(STable));
pTable->type = TSDB_NORMAL_TABLE;
@ -106,12 +106,12 @@ TEST(TsdbTest, createRepo) {
}
TEST(TsdbTest, openRepo) {
TEST(TsdbTest, DISABLED_openRepo) {
tsdb_repo_t *pRepo = tsdbOpenRepo("/home/ubuntu/work/ttest/vnode0");
ASSERT_NE(pRepo, nullptr);
}
TEST(TsdbTest, createFileGroup) {
TEST(TsdbTest, DISABLED_createFileGroup) {
SFileGroup fGroup;
ASSERT_EQ(tsdbCreateFileGroup("/home/ubuntu/work/ttest/vnode0/data", 1820, &fGroup, 1000), 0);