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

make SDataCols buffer flexible

This commit is contained in:
Hongze Cheng 2020-07-29 17:37:49 +08:00
parent 9ca99fa73e
commit 9b0bfc8959
6 changed files with 52 additions and 18 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

7
src/common/inc/tdataformat.h
View File

@ -239,7 +239,7 @@ typedef struct {
int numOfCols; // Total number of cols
int sversion; // TODO: set sversion
void * buf;
SDataCol cols[];
SDataCol *cols;
} SDataCols;
#define keyCol(pCols) (&((pCols)->cols[0])) // Key column
@ -249,13 +249,14 @@ typedef struct {
SDataCols *tdNewDataCols(int maxRowSize, int maxCols, int maxRows);
void tdResetDataCols(SDataCols *pCols);
void tdInitDataCols(SDataCols *pCols, STSchema *pSchema);
int tdInitDataCols(SDataCols *pCols, STSchema *pSchema);
SDataCols *tdDupDataCols(SDataCols *pCols, bool keepData);
void tdFreeDataCols(SDataCols *pCols);
void tdAppendDataRowToDataCol(SDataRow row, STSchema *pSchema, SDataCols *pCols);
void tdPopDataColsPoints(SDataCols *pCols, int pointsToPop); //!!!!
int tdMergeDataCols(SDataCols *target, SDataCols *src, int rowsToMerge);
void tdMergeTwoDataCols(SDataCols *target, SDataCols *src1, int *iter1, int limit1, SDataCols *src2, int *iter2, int limit2, int tRows);
void tdMergeTwoDataCols(SDataCols *target, SDataCols *src1, int *iter1, int limit1, SDataCols *src2, int *iter2,
int limit2, int tRows);
// ----------------- K-V data row structure
/*

29
src/common/src/tdataformat.c
View File

@ -310,9 +310,15 @@ void dataColSetOffset(SDataCol *pCol, int nEle) {
}
SDataCols *tdNewDataCols(int maxRowSize, int maxCols, int maxRows) {
SDataCols *pCols = (SDataCols *)calloc(1, sizeof(SDataCols) + sizeof(SDataCol) * maxCols);
SDataCols *pCols = (SDataCols *)calloc(1, sizeof(SDataCols));
if (pCols == NULL) return NULL;
pCols->cols = (SDataCol *)calloc(maxCols, sizeof(SDataCol));
if (pCols->cols == NULL) {
tdFreeDataCols(pCols);
return NULL;
}
pCols->maxRowSize = maxRowSize;
pCols->maxCols = maxCols;
pCols->maxPoints = maxRows;
@ -320,15 +326,27 @@ SDataCols *tdNewDataCols(int maxRowSize, int maxCols, int maxRows) {
pCols->buf = malloc(pCols->bufSize);
if (pCols->buf == NULL) {
free(pCols);
tdFreeDataCols(pCols);
return NULL;
}
return pCols;
}
void tdInitDataCols(SDataCols *pCols, STSchema *pSchema) {
// assert(schemaNCols(pSchema) <= pCols->numOfCols);
int tdInitDataCols(SDataCols *pCols, STSchema *pSchema) {
if (schemaNCols(pSchema) > pCols->maxCols) {
pCols->maxCols = schemaNCols(pSchema);
pCols->cols = (SDataCol *)realloc(pCols->cols, sizeof(SDataCol) * pCols->maxCols);
if (pCols->cols == NULL) return -1;
}
if (schemaTLen(pSchema) > pCols->maxRowSize) {
pCols->maxRowSize = schemaTLen(pSchema);
pCols->bufSize = schemaTLen(pSchema) * pCols->maxPoints;
pCols->buf = realloc(pCols->buf, pCols->bufSize);
if (pCols->buf == NULL) return -1;
}
tdResetDataCols(pCols);
pCols->numOfCols = schemaNCols(pSchema);
@ -337,11 +355,14 @@ void tdInitDataCols(SDataCols *pCols, STSchema *pSchema) {
dataColInit(pCols->cols + i, schemaColAt(pSchema, i), &ptr, pCols->maxPoints);
ASSERT((char *)ptr - (char *)(pCols->buf) <= pCols->bufSize);
}
return 0;
}
void tdFreeDataCols(SDataCols *pCols) {
if (pCols) {
tfree(pCols->buf);
tfree(pCols->cols);
free(pCols);
}
}

2
src/tsdb/inc/tsdbMain.h
View File

@ -483,7 +483,7 @@ void tsdbDestroyHelper(SRWHelper* pHelper);
void tsdbResetHelper(SRWHelper* pHelper);
int tsdbSetAndOpenHelperFile(SRWHelper* pHelper, SFileGroup* pGroup);
int tsdbCloseHelperFile(SRWHelper* pHelper, bool hasError);
void tsdbSetHelperTable(SRWHelper* pHelper, STable* pTable, STsdbRepo* pRepo);
int tsdbSetHelperTable(SRWHelper* pHelper, STable* pTable, STsdbRepo* pRepo);
int tsdbCommitTableData(SRWHelper* pHelper, SCommitIter* pCommitIter, SDataCols* pDataCols, TSKEY maxKey);
int tsdbMoveLastBlockIfNeccessary(SRWHelper* pHelper);
int tsdbWriteCompInfo(SRWHelper* pHelper);

2
src/tsdb/src/tsdbMain.c
View File

@ -788,7 +788,7 @@ static int tsdbRestoreInfo(STsdbRepo *pRepo) {
for (int i = 1; i < pMeta->maxTables; i++) {
STable *pTable = pMeta->tables[i];
if (pTable == NULL) continue;
tsdbSetHelperTable(&rhelper, pTable, pRepo);
if (tsdbSetHelperTable(&rhelper, pTable, pRepo) < 0) goto _err;
SCompIdx *pIdx = &(rhelper.curCompIdx);
if (pIdx->offset > 0 && pTable->lastKey < pIdx->maxKey) pTable->lastKey = pIdx->maxKey;

7
src/tsdb/src/tsdbMemTable.c
View File

@ -610,10 +610,13 @@ static int tsdbCommitToFile(STsdbRepo *pRepo, int fid, SCommitIter *iters, SRWHe
taosRLockLatch(&(pIter->pTable->latch));
tsdbSetHelperTable(pHelper, pIter->pTable, pRepo);
if (tsdbSetHelperTable(pHelper, pIter->pTable, pRepo) < 0) goto _err;
if (pIter->pIter != NULL) {
tdInitDataCols(pDataCols, tsdbGetTableSchemaImpl(pIter->pTable, false, false, -1));
if (tdInitDataCols(pDataCols, tsdbGetTableSchemaImpl(pIter->pTable, false, false, -1)) < 0) {
terrno = TSDB_CODE_TDB_OUT_OF_MEMORY;
goto _err;
}
if (tsdbCommitTableData(pHelper, pIter, pDataCols, maxKey) < 0) {
taosRUnLockLatch(&(pIter->pTable->latch));

15
src/tsdb/src/tsdbRWHelper.c
View File

@ -262,7 +262,7 @@ int tsdbCloseHelperFile(SRWHelper *pHelper, bool hasError) {
return 0;
}
void tsdbSetHelperTable(SRWHelper *pHelper, STable *pTable, STsdbRepo *pRepo) {
int tsdbSetHelperTable(SRWHelper *pHelper, STable *pTable, STsdbRepo *pRepo) {
ASSERT(helperHasState(pHelper, TSDB_HELPER_FILE_SET_AND_OPEN | TSDB_HELPER_IDX_LOAD));
// Clear members and state used by previous table
@ -273,8 +273,15 @@ void tsdbSetHelperTable(SRWHelper *pHelper, STable *pTable, STsdbRepo *pRepo) {
pHelper->tableInfo.uid = pTable->tableId.uid;
STSchema *pSchema = tsdbGetTableSchemaImpl(pTable, false, false, -1);
tdInitDataCols(pHelper->pDataCols[0], pSchema);
tdInitDataCols(pHelper->pDataCols[1], pSchema);
if (tdInitDataCols(pHelper->pDataCols[0], pSchema) < 0) {
terrno = TSDB_CODE_TDB_OUT_OF_MEMORY;
return -1;
}
if (tdInitDataCols(pHelper->pDataCols[1], pSchema) < 0) {
terrno = TSDB_CODE_TDB_OUT_OF_MEMORY;
return -1;
}
if (pHelper->idxH.numOfIdx > 0) {
while (true) {
@ -309,6 +316,8 @@ void tsdbSetHelperTable(SRWHelper *pHelper, STable *pTable, STsdbRepo *pRepo) {
helperSetState(pHelper, TSDB_HELPER_TABLE_SET);
ASSERT(pHelper->state == ((TSDB_HELPER_TABLE_SET << 1) - 1));
return 0;
}
int tsdbCommitTableData(SRWHelper *pHelper, SCommitIter *pCommitIter, SDataCols *pDataCols, TSKEY maxKey) {