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

Merge pull request #13421 from taosdata/feat/row_refact 

feat: vnode multi-version
This commit is contained in:
Hongze Cheng 2022-06-02 17:48:23 +08:00 committed by GitHub
parent 57ad40e32f 6b17c1cacd
commit 8bf4890737
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
6 changed files with 398 additions and 770 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

30
include/common/tdataformat.h
View File

@ -46,6 +46,7 @@ void tTSchemaDestroy(STSchema *pTSchema);
#define COL_VAL_NULL(CID) ((SColVal){.cid = (CID), .isNull = 1}) #define COL_VAL_NULL(CID) ((SColVal){.cid = (CID), .isNull = 1})
#define COL_VAL_VALUE(CID, V) ((SColVal){.cid = (CID), .value = (V)}) #define COL_VAL_VALUE(CID, V) ((SColVal){.cid = (CID), .value = (V)})
int32_t tTSRowNew(STSRowBuilder *pBuilder, SArray *pArray, STSchema *pTSchema, STSRow2 **ppRow);
int32_t tTSRowClone(const STSRow2 *pRow, STSRow2 **ppRow); int32_t tTSRowClone(const STSRow2 *pRow, STSRow2 **ppRow);
void tTSRowFree(STSRow2 *pRow); void tTSRowFree(STSRow2 *pRow);
void tTSRowGet(STSRow2 *pRow, STSchema *pTSchema, int32_t iCol, SColVal *pColVal); void tTSRowGet(STSRow2 *pRow, STSchema *pTSchema, int32_t iCol, SColVal *pColVal);
@ -54,13 +55,13 @@ int32_t tPutTSRow(uint8_t *p, STSRow2 *pRow);
int32_t tGetTSRow(uint8_t *p, STSRow2 *pRow); int32_t tGetTSRow(uint8_t *p, STSRow2 *pRow);
// STSRowBuilder // STSRowBuilder
#if 0 #define tsRowBuilderInit() ((STSRowBuilder){0})
int32_t tTSRowBuilderInit(STSRowBuilder *pBuilder, int32_t sver, int32_t nCols, SSchema *pSchema); #define tsRowBuilderClear(B) \
void tTSRowBuilderClear(STSRowBuilder *pBuilder); do { \
void tTSRowBuilderReset(STSRowBuilder *pBuilder); if ((B)->pBuf) { \
int32_t tTSRowBuilderPut(STSRowBuilder *pBuilder, int32_t cid, uint8_t *pData, uint32_t nData); taosMemoryFree((B)->pBuf); \
int32_t tTSRowBuilderGetRow(STSRowBuilder *pBuilder, const STSRow2 **ppRow); } \
#endif } while (0)
// STag // STag
int32_t tTagNew(SArray *pArray, int32_t version, int8_t isJson, STag **ppTag); int32_t tTagNew(SArray *pArray, int32_t version, int8_t isJson, STag **ppTag);
@ -106,17 +107,9 @@ struct STSRow2 {
}; };
struct STSRowBuilder { struct STSRowBuilder {
STSchema *pTSchema; STSRow2 tsRow;
int32_t szBitMap1; int32_t szBuf;
int32_t szBitMap2; uint8_t *pBuf;
int32_t szKVBuf;
uint8_t *pKVBuf;
int32_t szTPBuf;
uint8_t *pTPBuf;
int32_t iCol;
int32_t vlenKV;
int32_t vlenTP;
STSRow2 row;
}; };
struct SValue { struct SValue {
@ -422,4 +415,3 @@ int32_t tdMergeDataCols(SDataCols *target, SDataCols *source, int32_t rowsToM
#endif #endif
#endif /*_TD_COMMON_DATA_FORMAT_H_*/ #endif /*_TD_COMMON_DATA_FORMAT_H_*/

613
source/common/src/tdataformat.c
View File

@ -142,134 +142,251 @@ static FORCE_INLINE int32_t tGetValue(uint8_t *p, SValue *pValue, int8_t type) {
} }
// STSRow2 ======================================================================== // STSRow2 ========================================================================
static void tTupleTSRowNew(SArray *pArray, STSchema *pTSchema, STSRow2 *pRow) { static void setBitMap(uint8_t *pb, uint8_t v, int32_t idx, uint8_t flags) {
int32_t nColVal = taosArrayGetSize(pArray); if (pb) {
switch (flags & 0xf) {
case TSROW_HAS_NULL | TSROW_HAS_NONE:
case TSROW_HAS_VAL | TSROW_HAS_NONE:
if (v) {
SET_BIT1(pb, idx, (uint8_t)1);
} else {
SET_BIT1(pb, idx, (uint8_t)0);
}
break;
case TSROW_HAS_VAL | TSROW_HAS_NULL:
v = v - 1;
SET_BIT1(pb, idx, v);
break;
case TSROW_HAS_VAL | TSROW_HAS_NULL | TSROW_HAS_NONE:
SET_BIT2(pb, idx, v);
break;
default:
ASSERT(0);
}
}
}
#define SET_IDX(p, i, n, f) \
do { \
if ((f)&TSROW_KV_SMALL) { \
((uint8_t *)(p))[i] = (n); \
} else if ((f)&TSROW_KV_MID) { \
((uint16_t *)(p))[i] = (n); \
} else { \
((uint32_t *)(p))[i] = (n); \
} \
} while (0)
int32_t tTSRowNew(STSRowBuilder *pBuilder, SArray *pArray, STSchema *pTSchema, STSRow2 **ppRow) {
int32_t code = 0;
STColumn *pTColumn; STColumn *pTColumn;
SColVal *pColVal; SColVal *pColVal;
int32_t nColVal = taosArrayGetSize(pArray);
int32_t iColVal;
ASSERT(nColVal > 0); ASSERT(nColVal > 0);
pRow->sver = pTSchema->version; // try
// ts
pTColumn = &pTSchema->columns[0];
pColVal = (SColVal *)taosArrayGet(pArray, 0);
ASSERT(pTColumn->colId == 0 && pColVal->cid == 0);
ASSERT(pTColumn->type == TSDB_DATA_TYPE_TIMESTAMP);
pRow->ts = pColVal->value.ts;
// other fields
int32_t iColVal = 1;
int32_t bidx;
uint32_t nv = 0;
uint8_t *pb = NULL;
uint8_t *pf = NULL;
uint8_t *pv = NULL;
uint8_t flags = 0; uint8_t flags = 0;
for (int32_t iColumn = 1; iColumn < pTSchema->numOfCols; iColumn++) { uint32_t ntv = 0;
bidx = iColumn - 1; uint32_t nkv = 0;
pTColumn = &pTSchema->columns[iColumn]; int16_t nTag = 0;
uint32_t maxIdx = 0;
iColVal = 0;
for (int32_t iColumn = 0; iColumn < pTSchema->numOfCols; iColumn++) {
pTColumn = &pTSchema->columns[iColumn];
if (iColVal < nColVal) { if (iColVal < nColVal) {
pColVal = (SColVal *)taosArrayGet(pArray, iColVal); pColVal = (SColVal *)taosArrayGet(pArray, iColVal);
} else { } else {
pColVal = NULL; pColVal = NULL;
} }
if (iColumn == 0) {
ASSERT(pColVal->cid == pTColumn->colId);
ASSERT(pTColumn->type == TSDB_DATA_TYPE_TIMESTAMP);
ASSERT(pTColumn->colId == 0);
iColVal++;
} else {
if (pColVal) { if (pColVal) {
if (pColVal->cid == pTColumn->colId) { if (pColVal->cid == pTColumn->colId) {
iColVal++; iColVal++;
if (pColVal->isNone) { if (pColVal->isNone) {
goto _set_none; flags |= TSROW_HAS_NONE;
} else if (pColVal->isNull) { } else if (pColVal->isNull) {
goto _set_null; flags |= TSROW_HAS_NULL;
maxIdx = nkv;
nTag++;
nkv += tPutI16v(NULL, -pTColumn->colId);
} else { } else {
goto _set_value; flags |= TSROW_HAS_VAL;
maxIdx = nkv;
nTag++;
nkv += tPutI16v(NULL, pTColumn->colId);
nkv += tPutValue(NULL, &pColVal->value, pTColumn->type);
if (IS_VAR_DATA_TYPE(pTColumn->type)) {
ntv += tPutValue(NULL, &pColVal->value, pTColumn->type);
}
} }
} else if (pColVal->cid > pTColumn->colId) { } else if (pColVal->cid > pTColumn->colId) {
goto _set_none; flags |= TSROW_HAS_NONE;
} else { } else {
ASSERT(0); ASSERT(0);
} }
} else { } else {
goto _set_none;
}
_set_none:
flags |= TSROW_HAS_NONE; flags |= TSROW_HAS_NONE;
// SET_BIT2(pb, bidx, 0); (todo)
continue;
_set_null:
flags != TSROW_HAS_NULL;
// SET_BIT2(pb, bidx, 1); (todo)
continue;
_set_value:
flags != TSROW_HAS_VAL;
// SET_BIT2(pb, bidx, 2); (todo)
if (IS_VAR_DATA_TYPE(pTColumn->type)) {
// nv += tPutColVal(pv ? pv + nv : pv, pColVal, pTColumn->type, 1);
} else {
// tPutColVal(pf ? pf + pTColumn->offset : pf, pColVal, pTColumn->type, 1);
} }
continue; }
} }
ASSERT(flags); ASSERT(flags);
switch (flags & 0xf) {
case TSROW_HAS_NONE: // decide
case TSROW_HAS_NULL: uint32_t nData = 0;
pRow->nData = 0; uint32_t nDataT = 0;
break; uint32_t nDataK = 0;
if (flags == TSROW_HAS_NONE || flags == TSROW_HAS_NULL) {
nData = 0;
} else {
switch (flags) {
case TSROW_HAS_VAL: case TSROW_HAS_VAL:
pRow->nData = pTSchema->flen + nv; nDataT = pTSchema->flen + ntv;
break; break;
case TSROW_HAS_NULL | TSROW_HAS_NONE: case TSROW_HAS_NULL | TSROW_HAS_NONE:
pRow->nData = BIT1_SIZE(pTSchema->numOfCols - 1); nDataT = BIT1_SIZE(pTSchema->numOfCols - 1);
break; break;
case TSROW_HAS_VAL | TSROW_HAS_NONE: case TSROW_HAS_VAL | TSROW_HAS_NONE:
case TSROW_HAS_VAL | TSROW_HAS_NULL: case TSROW_HAS_VAL | TSROW_HAS_NULL:
pRow->nData = BIT1_SIZE(pTSchema->numOfCols - 1) + pTSchema->flen + nv; nDataT = BIT1_SIZE(pTSchema->numOfCols - 1) + pTSchema->flen + ntv;
break; break;
case TSROW_HAS_VAL | TSROW_HAS_NULL | TSROW_HAS_NONE: case TSROW_HAS_VAL | TSROW_HAS_NULL | TSROW_HAS_NONE:
pRow->nData = BIT2_SIZE(pTSchema->numOfCols - 1) + pTSchema->flen + nv; nDataT = BIT2_SIZE(pTSchema->numOfCols - 1) + pTSchema->flen + ntv;
break; break;
default: default:
break; ASSERT(0);
}
uint8_t tflags = 0;
if (maxIdx <= UINT8_MAX) {
nDataK = sizeof(STSKVRow) + sizeof(uint8_t) * nTag + nkv;
tflags |= TSROW_KV_SMALL;
} else if (maxIdx <= UINT16_MAX) {
nDataK = sizeof(STSKVRow) + sizeof(uint16_t) * nTag + nkv;
tflags |= TSROW_KV_MID;
} else {
nDataK = sizeof(STSKVRow) + sizeof(uint32_t) * nTag + nkv;
tflags |= TSROW_KV_BIG;
}
if (nDataT < nDataK) {
nData = nDataT;
} else {
nData = nDataK;
flags |= tflags;
} }
} }
static void tMapTSRowNew(SArray *pArray, STSchema *pTSchema, STSRow2 *pRow) { // alloc
int32_t nColVal = taosArrayGetSize(pArray); if (pBuilder) {
STColumn *pTColumn; // create from a builder
SColVal *pColVal; if (nData == 0) {
pBuilder->tsRow.nData = 0;
pBuilder->tsRow.pData = NULL;
} else {
if (pBuilder->szBuf < nData) {
uint8_t *p = taosMemoryRealloc(pBuilder->pBuf, nData);
if (p == NULL) {
code = TSDB_CODE_OUT_OF_MEMORY;
goto _exit;
}
pBuilder->pBuf = p;
pBuilder->szBuf = nData;
}
ASSERT(nColVal > 0); pBuilder->tsRow.nData = nData;
pBuilder->tsRow.pData = pBuilder->pBuf;
}
pRow->sver = pTSchema->version; *ppRow = &pBuilder->tsRow;
} else {
// create a new one
*ppRow = (STSRow2 *)taosMemoryMalloc(sizeof(STSRow2));
if (*ppRow == NULL) {
code = TSDB_CODE_OUT_OF_MEMORY;
goto _exit;
}
if (nData == 0) {
(*ppRow)->nData = 0;
(*ppRow)->pData = NULL;
} else {
(*ppRow)->nData = nData;
(*ppRow)->pData = taosMemoryMalloc(nData);
if ((*ppRow)->pData == NULL) {
taosMemoryFree(*ppRow);
code = TSDB_CODE_OUT_OF_MEMORY;
goto _exit;
}
}
}
// build
(*ppRow)->flags = flags;
(*ppRow)->sver = pTSchema->version;
// ts
pTColumn = &pTSchema->columns[0];
pColVal = (SColVal *)taosArrayGet(pArray, 0); pColVal = (SColVal *)taosArrayGet(pArray, 0);
(*ppRow)->ts = pColVal->value.ts;
ASSERT(pTColumn->colId == 0 && pColVal->cid == 0); if ((*ppRow)->pData) {
ASSERT(pTColumn->type == TSDB_DATA_TYPE_TIMESTAMP); STSKVRow *pTSKVRow = NULL;
pRow->ts = pColVal->value.ts;
// other fields
int32_t iColVal = 1;
uint32_t nv = 0;
uint8_t *pv = NULL;
uint8_t *pidx = NULL; uint8_t *pidx = NULL;
uint8_t flags = 0; uint8_t *pkv = NULL;
int16_t nCol = 0; uint8_t *pb = NULL;
uint8_t *pf = NULL;
uint8_t *ptv = NULL;
nkv = 0;
ntv = 0;
iColVal = 1;
if (flags & 0xf0 == 0) {
switch (flags & 0xf) {
case TSROW_HAS_VAL:
pf = (*ppRow)->pData;
ptv = pf + pTSchema->flen;
break;
case TSROW_HAS_NULL | TSROW_HAS_NONE:
pb = (*ppRow)->pData;
break;
case TSROW_HAS_VAL | TSROW_HAS_NONE:
case TSROW_HAS_VAL | TSROW_HAS_NULL:
pb = (*ppRow)->pData;
pf = pb + BIT1_SIZE(pTSchema->numOfCols - 1);
ptv = pf + pTSchema->flen;
break;
case TSROW_HAS_VAL | TSROW_HAS_NULL | TSROW_HAS_NONE:
pb = (*ppRow)->pData;
pf = pb + BIT2_SIZE(pTSchema->numOfCols - 1);
ptv = pf + pTSchema->flen;
break;
default:
ASSERT(0);
}
} else {
pTSKVRow = (STSKVRow *)(*ppRow)->pData;
pTSKVRow->nCols = 0;
pidx = pTSKVRow->idx;
if (flags & TSROW_KV_SMALL) {
pkv = pidx + sizeof(uint8_t) * nTag;
} else if (flags & TSROW_KV_MID) {
pkv = pidx + sizeof(uint16_t) * nTag;
} else {
pkv = pidx + sizeof(uint32_t) * nTag;
}
}
for (int32_t iColumn = 1; iColumn < pTSchema->numOfCols; iColumn++) { for (int32_t iColumn = 1; iColumn < pTSchema->numOfCols; iColumn++) {
pTColumn = &pTSchema->columns[iColumn]; pTColumn = &pTSchema->columns[iColumn];
if (iColVal < nColVal) { if (iColVal < nColVal) {
pColVal = (SColVal *)taosArrayGet(pArray, iColVal); pColVal = (SColVal *)taosArrayGet(pArray, iColVal);
} else { } else {
@ -279,6 +396,7 @@ static void tMapTSRowNew(SArray *pArray, STSchema *pTSchema, STSRow2 *pRow) {
if (pColVal) { if (pColVal) {
if (pColVal->cid == pTColumn->colId) { if (pColVal->cid == pTColumn->colId) {
iColVal++; iColVal++;
if (pColVal->isNone) { if (pColVal->isNone) {
goto _set_none; goto _set_none;
} else if (pColVal->isNull) { } else if (pColVal->isNull) {
@ -296,48 +414,42 @@ static void tMapTSRowNew(SArray *pArray, STSchema *pTSchema, STSRow2 *pRow) {
} }
_set_none: _set_none:
flags |= TSROW_HAS_NONE; if (flags & 0xf0 == 0) {
setBitMap(pb, 0, iColumn - 1, flags);
}
continue; continue;
_set_null: _set_null:
flags != TSROW_HAS_NULL; if (flags & 0xf0 == 0) {
pidx[nCol++] = nv; setBitMap(pb, 1, iColumn - 1, flags);
// nv += tPutColVal(pv ? pv + nv : pv, pColVal, pTColumn->type, 0); } else {
SET_IDX(pidx, pTSKVRow->nCols, nkv, flags);
pTSKVRow->nCols++;
nkv += tPutI16v(pkv + nkv, -pTColumn->colId);
}
continue; continue;
_set_value: _set_value:
flags != TSROW_HAS_VAL; if (flags & 0xf0 == 0) {
pidx[nCol++] = nv; setBitMap(pb, 2, iColumn - 1, flags);
// nv += tPutColVal(pv ? pv + nv : pv, pColVal, pTColumn->type, 0);
if (IS_VAR_DATA_TYPE(pTColumn->type)) {
*(VarDataOffsetT *)(pf + pTColumn->offset) = ntv;
ntv += tPutValue(ptv + ntv, &pColVal->value, pTColumn->type);
} else {
tPutValue(pf + pTColumn->offset, &pColVal->value, pTColumn->type);
}
} else {
SET_IDX(pidx, pTSKVRow->nCols, nkv, flags);
pTSKVRow->nCols++;
nkv += tPutI16v(pkv + nkv, pColVal->cid);
nkv += tPutValue(pkv + nkv, &pColVal->value, pTColumn->type);
}
continue; continue;
} }
if (nv <= UINT8_MAX) {
// small
} else if (nv <= UINT16_MAX) {
// mid
} else {
// large
}
}
// try-decide-build
int32_t tTSRowNew(SArray *pArray, STSchema *pTSchema, STSRow2 **ppRow) {
int32_t code = 0;
STSRow2 rowT = {0};
STSRow2 rowM = {0};
// try
tTupleTSRowNew(pArray, pTSchema, &rowT);
tMapTSRowNew(pArray, pTSchema, &rowM);
// decide & build
if (rowT.nData <= rowM.nData) {
tTupleTSRowNew(pArray, pTSchema, &rowT);
} else {
tMapTSRowNew(pArray, pTSchema, &rowM);
} }
_exit:
return code; return code;
} }
@ -626,266 +738,8 @@ void tTSchemaDestroy(STSchema *pTSchema) {
} }
// STSRowBuilder // STSRowBuilder
#if 0
int32_t tTSRowBuilderInit(STSRowBuilder *pBuilder, int32_t sver, int32_t nCols, SSchema *pSchema) {
if (tTSchemaCreate(sver, pSchema, nCols, &pBuilder->pTSchema) < 0) return -1;
pBuilder->szBitMap1 = BIT1_SIZE(nCols - 1);
pBuilder->szBitMap2 = BIT2_SIZE(nCols - 1);
pBuilder->szKVBuf =
sizeof(STSKVRow) + sizeof(SKVIdx) * (nCols - 1) + pBuilder->pTSchema->flen + pBuilder->pTSchema->vlen;
pBuilder->szTPBuf = pBuilder->szBitMap2 + pBuilder->pTSchema->flen + pBuilder->pTSchema->vlen;
pBuilder->pKVBuf = taosMemoryMalloc(pBuilder->szKVBuf);
if (pBuilder->pKVBuf == NULL) {
terrno = TSDB_CODE_OUT_OF_MEMORY;
tTSchemaDestroy(pBuilder->pTSchema);
return -1;
}
pBuilder->pTPBuf = taosMemoryMalloc(pBuilder->szTPBuf);
if (pBuilder->pTPBuf == NULL) {
terrno = TSDB_CODE_OUT_OF_MEMORY;
taosMemoryFree(pBuilder->pKVBuf);
tTSchemaDestroy(pBuilder->pTSchema);
return -1;
}
return 0;
}
void tTSRowBuilderClear(STSRowBuilder *pBuilder) {
if (pBuilder->pTPBuf) {
taosMemoryFree(pBuilder->pTPBuf);
pBuilder->pTPBuf = NULL;
}
if (pBuilder->pKVBuf) {
taosMemoryFree(pBuilder->pKVBuf);
pBuilder->pKVBuf = NULL;
}
tTSchemaDestroy(pBuilder->pTSchema);
pBuilder->pTSchema = NULL;
}
void tTSRowBuilderReset(STSRowBuilder *pBuilder) {
for (int32_t iCol = pBuilder->pTSchema->numOfCols - 1; iCol >= 0; iCol--) {
STColumn *pTColumn = &pBuilder->pTSchema->columns[iCol];
COL_CLR_SET(pTColumn->flags);
}
pBuilder->iCol = 0;
((STSKVRow *)pBuilder->pKVBuf)->nCols = 0;
pBuilder->vlenKV = 0;
pBuilder->vlenTP = 0;
pBuilder->row.flags = 0;
}
int32_t tTSRowBuilderPut(STSRowBuilder *pBuilder, int32_t cid, uint8_t *pData, uint32_t nData) {
STColumn *pTColumn = &pBuilder->pTSchema->columns[pBuilder->iCol];
uint8_t *p;
int32_t iCol;
STSKVRow *pTSKVRow = (STSKVRow *)pBuilder->pKVBuf;
// use interp search
if (pTColumn->colId < cid) { // right search
for (iCol = pBuilder->iCol + 1; iCol < pBuilder->pTSchema->numOfCols; iCol++) {
pTColumn = &pBuilder->pTSchema->columns[iCol];
if (pTColumn->colId >= cid) break;
}
} else if (pTColumn->colId > cid) { // left search
for (iCol = pBuilder->iCol - 1; iCol >= 0; iCol--) {
pTColumn = &pBuilder->pTSchema->columns[iCol];
if (pTColumn->colId <= cid) break;
}
}
if (pTColumn->colId != cid || COL_IS_SET(pTColumn->flags)) {
return -1;
}
pBuilder->iCol = iCol;
// set value
if (cid == 0) {
ASSERT(pData && nData == sizeof(TSKEY) && iCol == 0);
pBuilder->row.ts = *(TSKEY *)pData;
pTColumn->flags |= COL_SET_VAL;
} else {
if (pData) {
// set VAL
pBuilder->row.flags |= TSROW_HAS_VAL;
pTColumn->flags |= COL_SET_VAL;
/* KV */
if (1) { // avoid KV at some threshold (todo)
pTSKVRow->idx[pTSKVRow->nCols].cid = cid;
pTSKVRow->idx[pTSKVRow->nCols].offset = pBuilder->vlenKV;
p = pBuilder->pKVBuf + sizeof(STSKVRow) + sizeof(SKVIdx) * (pBuilder->pTSchema->numOfCols - 1) +
pBuilder->vlenKV;
if (IS_VAR_DATA_TYPE(pTColumn->type)) {
ASSERT(nData <= pTColumn->bytes);
pBuilder->vlenKV += tPutBinary(p, pData, nData);
} else {
ASSERT(nData == pTColumn->bytes);
memcpy(p, pData, nData);
pBuilder->vlenKV += nData;
}
}
/* TUPLE */
p = pBuilder->pTPBuf + pBuilder->szBitMap2 + pTColumn->offset;
if (IS_VAR_DATA_TYPE(pTColumn->type)) {
ASSERT(nData <= pTColumn->bytes);
*(int32_t *)p = pBuilder->vlenTP;
p = pBuilder->pTPBuf + pBuilder->szBitMap2 + pBuilder->pTSchema->flen + pBuilder->vlenTP;
pBuilder->vlenTP += tPutBinary(p, pData, nData);
} else {
ASSERT(nData == pTColumn->bytes);
memcpy(p, pData, nData);
}
} else {
// set NULL
pBuilder->row.flags |= TSROW_HAS_NULL;
pTColumn->flags |= COL_SET_NULL;
pTSKVRow->idx[pTSKVRow->nCols].cid = cid;
pTSKVRow->idx[pTSKVRow->nCols].offset = -1;
}
pTSKVRow->nCols++;
}
return 0;
}
static FORCE_INLINE int tSKVIdxCmprFn(const void *p1, const void *p2) {
SKVIdx *pKVIdx1 = (SKVIdx *)p1;
SKVIdx *pKVIdx2 = (SKVIdx *)p2;
if (pKVIdx1->cid > pKVIdx2->cid) {
return 1;
} else if (pKVIdx1->cid < pKVIdx2->cid) {
return -1;
}
return 0;
}
static void setBitMap(uint8_t *p, STSchema *pTSchema, uint8_t flags) {
int32_t bidx;
STColumn *pTColumn;
for (int32_t iCol = 1; iCol < pTSchema->numOfCols; iCol++) {
pTColumn = &pTSchema->columns[iCol];
bidx = iCol - 1;
switch (flags) {
case TSROW_HAS_NULL | TSROW_HAS_NONE:
if (pTColumn->flags & COL_SET_NULL) {
SET_BIT1(p, bidx, (uint8_t)1);
} else {
SET_BIT1(p, bidx, (uint8_t)0);
}
break;
case TSROW_HAS_VAL | TSROW_HAS_NULL | TSROW_HAS_NONE:
if (pTColumn->flags & COL_SET_NULL) {
SET_BIT2(p, bidx, (uint8_t)1);
} else if (pTColumn->flags & COL_SET_VAL) {
SET_BIT2(p, bidx, (uint8_t)2);
} else {
SET_BIT2(p, bidx, (uint8_t)0);
}
break;
default:
if (pTColumn->flags & COL_SET_VAL) {
SET_BIT1(p, bidx, (uint8_t)1);
} else {
SET_BIT1(p, bidx, (uint8_t)0);
}
break;
}
}
}
int32_t tTSRowBuilderGetRow(STSRowBuilder *pBuilder, const STSRow2 **ppRow) {
int32_t nDataTP, nDataKV;
STSKVRow *pTSKVRow = (STSKVRow *)pBuilder->pKVBuf;
int32_t nCols = pBuilder->pTSchema->numOfCols;
// error not set ts
if (!COL_IS_SET(pBuilder->pTSchema->columns->flags)) {
return -1;
}
ASSERT(pTSKVRow->nCols < nCols);
if (pTSKVRow->nCols < nCols - 1) {
pBuilder->row.flags |= TSROW_HAS_NONE;
}
ASSERT((pBuilder->row.flags & 0xf) != 0);
*(ppRow) = &pBuilder->row;
switch (pBuilder->row.flags & 0xf) {
case TSROW_HAS_NONE:
case TSROW_HAS_NULL:
pBuilder->row.nData = 0;
pBuilder->row.pData = NULL;
return 0;
case TSROW_HAS_NULL | TSROW_HAS_NONE:
nDataTP = pBuilder->szBitMap1;
break;
case TSROW_HAS_VAL:
nDataTP = pBuilder->pTSchema->flen + pBuilder->vlenTP;
break;
case TSROW_HAS_VAL | TSROW_HAS_NONE:
case TSROW_HAS_VAL | TSROW_HAS_NULL:
nDataTP = pBuilder->szBitMap1 + pBuilder->pTSchema->flen + pBuilder->vlenTP;
break;
case TSROW_HAS_VAL | TSROW_HAS_NULL | TSROW_HAS_NONE:
nDataTP = pBuilder->szBitMap2 + pBuilder->pTSchema->flen + pBuilder->vlenTP;
break;
default:
ASSERT(0);
}
nDataKV = sizeof(STSKVRow) + sizeof(SKVIdx) * pTSKVRow->nCols + pBuilder->vlenKV;
pBuilder->row.sver = pBuilder->pTSchema->version;
if (nDataKV < nDataTP) {
// generate KV row
ASSERT((pBuilder->row.flags & 0xf) != TSROW_HAS_VAL);
pBuilder->row.flags |= TSROW_KV_ROW;
pBuilder->row.nData = nDataKV;
pBuilder->row.pData = pBuilder->pKVBuf;
qsort(pTSKVRow->idx, pTSKVRow->nCols, sizeof(SKVIdx), tSKVIdxCmprFn);
if (pTSKVRow->nCols < nCols - 1) {
memmove(&pTSKVRow->idx[pTSKVRow->nCols], &pTSKVRow->idx[nCols - 1], pBuilder->vlenKV);
}
} else {
// generate TUPLE row
pBuilder->row.nData = nDataTP;
uint8_t *p;
uint8_t flags = (pBuilder->row.flags & 0xf);
if (flags == TSROW_HAS_VAL) {
pBuilder->row.pData = pBuilder->pTPBuf + pBuilder->szBitMap2;
} else {
if (flags == (TSROW_HAS_VAL | TSROW_HAS_NULL | TSROW_HAS_NONE)) {
pBuilder->row.pData = pBuilder->pTPBuf;
} else {
pBuilder->row.pData = pBuilder->pTPBuf + pBuilder->szBitMap2 - pBuilder->szBitMap1;
}
setBitMap(pBuilder->row.pData, pBuilder->pTSchema, flags);
}
}
return 0;
}
#endif
// STag
static int tTagValCmprFn(const void *p1, const void *p2) { static int tTagValCmprFn(const void *p1, const void *p2) {
if (((STagVal *)p1)->cid < ((STagVal *)p2)->cid) { if (((STagVal *)p1)->cid < ((STagVal *)p2)->cid) {
return -1; return -1;
@ -953,25 +807,6 @@ static void debugPrintTagVal(int8_t type, const void *val, int32_t vlen, const c
} }
} }
// if (isLarge) {
// p = (uint8_t *)&((int16_t *)pTag->idx)[pTag->nTag];
// } else {
// p = (uint8_t *)&pTag->idx[pTag->nTag];
// }
// (*ppArray) = taosArrayInit(pTag->nTag + 1, sizeof(STagVal));
// if (*ppArray == NULL) {
// code = TSDB_CODE_OUT_OF_MEMORY;
// goto _err;
// }
// for (int16_t iTag = 0; iTag < pTag->nTag; iTag++) {
// if (isLarge) {
// offset = ((int16_t *)pTag->idx)[iTag];
// } else {
// offset = pTag->idx[iTag];
// }
void debugPrintSTag(STag *pTag, const char *tag, int32_t ln) { void debugPrintSTag(STag *pTag, const char *tag, int32_t ln) {
int8_t isJson = pTag->flags & TD_TAG_JSON; int8_t isJson = pTag->flags & TD_TAG_JSON;
int8_t isLarge = pTag->flags & TD_TAG_LARGE; int8_t isLarge = pTag->flags & TD_TAG_LARGE;
@ -1215,9 +1050,7 @@ int32_t tEncodeTag(SEncoder *pEncoder, const STag *pTag) {
return tEncodeBinary(pEncoder, (const uint8_t *)pTag, pTag->len); return tEncodeBinary(pEncoder, (const uint8_t *)pTag, pTag->len);
} }
int32_t tDecodeTag(SDecoder *pDecoder, STag **ppTag) { int32_t tDecodeTag(SDecoder *pDecoder, STag **ppTag) { return tDecodeBinary(pDecoder, (uint8_t **)ppTag, NULL); }
return tDecodeBinary(pDecoder, (uint8_t **)ppTag, NULL);
}
int32_t tTagToValArray(const STag *pTag, SArray **ppArray) { int32_t tTagToValArray(const STag *pTag, SArray **ppArray) {
int32_t code = 0; int32_t code = 0;

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

@ -36,7 +36,6 @@ target_sources(
# tsdb # tsdb
"src/tsdb/tsdbCommit.c" "src/tsdb/tsdbCommit.c"
"src/tsdb/tsdbCommit2.c"
"src/tsdb/tsdbFile.c" "src/tsdb/tsdbFile.c"
"src/tsdb/tsdbFS.c" "src/tsdb/tsdbFS.c"
"src/tsdb/tsdbOpen.c" "src/tsdb/tsdbOpen.c"

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

@ -88,6 +88,18 @@ static void tsdbLoadAndMergeFromCache(STsdb *pTsdb, SDataCols *pDataCols, int *i
SDataCols *pTarget, TSKEY maxKey, int maxRows, int8_t update); SDataCols *pTarget, TSKEY maxKey, int maxRows, int8_t update);
int tsdbWriteBlockIdx(SDFile *pHeadf, SArray *pIdxA, void **ppBuf); int tsdbWriteBlockIdx(SDFile *pHeadf, SArray *pIdxA, void **ppBuf);
int tsdbBegin(STsdb *pTsdb) {
if (!pTsdb) return 0;
STsdbMemTable *pMem;
if (tsdbMemTableCreate(pTsdb, &pTsdb->mem) < 0) {
return -1;
}
return 0;
}
int tsdbApplyRtnOnFSet(STsdb *pRepo, SDFileSet *pSet, SRtn *pRtn) { int tsdbApplyRtnOnFSet(STsdb *pRepo, SDFileSet *pSet, SRtn *pRtn) {
SDiskID did; SDiskID did;
SDFileSet nSet = {0}; SDFileSet nSet = {0};

28
source/dnode/vnode/src/tsdb/tsdbCommit2.c
View File

@ -1,28 +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 "tsdb.h"
int tsdbBegin(STsdb *pTsdb) {
if (!pTsdb) return 0;
STsdbMemTable *pMem;
if (tsdbMemTableCreate(pTsdb, &pTsdb->mem) < 0) {
return -1;
}
return 0;
}

350
source/dnode/vnode/src/tsdb/tsdbMemTable2.c
View File

@ -52,20 +52,21 @@ struct SMemTable {
SArray *pArray; // SArray<SMemData> SArray *pArray; // SArray<SMemData>
}; };
#define SL_MAX_LEVEL 5
#define SL_NODE_SIZE(l) (sizeof(SMemSkipListNode) + sizeof(SMemSkipListNode *) * (l)*2) #define SL_NODE_SIZE(l) (sizeof(SMemSkipListNode) + sizeof(SMemSkipListNode *) * (l)*2)
#define SL_NODE_HALF_SIZE(l) (sizeof(SMemSkipListNode) + sizeof(SMemSkipListNode *) * (l))
#define SL_NODE_FORWARD(n, l) ((n)->forwards[l]) #define SL_NODE_FORWARD(n, l) ((n)->forwards[l])
#define SL_NODE_BACKWARD(n, l) ((n)->forwards[(n)->level + (l)]) #define SL_NODE_BACKWARD(n, l) ((n)->forwards[(n)->level + (l)])
#define SL_NODE_DATA(n) (&SL_NODE_BACKWARD(n, (n)->level)) #define SL_NODE_DATA(n) (&SL_NODE_BACKWARD(n, (n)->level))
#define SL_HEAD_FORWARD(sl, l) SL_NODE_FORWARD((sl)->pHead, l)
#define SL_TAIL_BACKWARD(sl, l) SL_NODE_FORWARD((sl)->pTail, l)
static int32_t tsdbGetOrCreateMemData(SMemTable *pMemTable, tb_uid_t suid, tb_uid_t uid, SMemData **ppMemData); static int32_t tsdbGetOrCreateMemData(SMemTable *pMemTable, tb_uid_t suid, tb_uid_t uid, SMemData **ppMemData);
static int memDataPCmprFn(const void *p1, const void *p2); static int memDataPCmprFn(const void *p1, const void *p2);
static int32_t tPutTSDBRow(uint8_t *p, TSDBROW *pRow); static int32_t tPutTSDBRow(uint8_t *p, TSDBROW *pRow);
static int32_t tGetTSDBRow(uint8_t *p, TSDBROW *pRow); static int32_t tGetTSDBRow(uint8_t *p, TSDBROW *pRow);
static int8_t tsdbMemSkipListRandLevel(SMemSkipList *pSl); static int8_t tsdbMemSkipListRandLevel(SMemSkipList *pSl);
static void memDataMovePos(SMemData *pMemData, TSDBROW *pRow, int8_t isForward, SMemSkipListNode **pos);
static int32_t memDataPutRow(SVBufPool *pPool, SMemData *pMemData, TSDBROW *pRow, int8_t isForward,
SMemSkipListNode **pos);
// SMemTable ============================================== // SMemTable ==============================================
int32_t tsdbMemTableCreate2(STsdb *pTsdb, SMemTable **ppMemTable) { int32_t tsdbMemTableCreate2(STsdb *pTsdb, SMemTable **ppMemTable) {
@ -109,6 +110,7 @@ int32_t tsdbInsertTableData2(STsdb *pTsdb, int64_t version, SVSubmitBlk *pSubmit
TSDBROW row = {.version = version}; TSDBROW row = {.version = version};
ASSERT(pMemTable); ASSERT(pMemTable);
ASSERT(pSubmitBlk->nData > 0);
{ {
// check if table exists (todo) // check if table exists (todo)
@ -122,38 +124,29 @@ int32_t tsdbInsertTableData2(STsdb *pTsdb, int64_t version, SVSubmitBlk *pSubmit
// do insert // do insert
int32_t nt; int32_t nt;
uint8_t *pt;
int32_t n = 0; int32_t n = 0;
uint8_t *p = pSubmitBlk->pData; uint8_t *p = pSubmitBlk->pData;
SVBufPool *pPool = pTsdb->pVnode->inUse; int32_t nRow = 0;
int8_t level; SMemSkipListNode *pos[SL_MAX_LEVEL] = {0};
SMemSkipListNode *pNode;
for (int8_t iLevel = 0; iLevel < SL_MAX_LEVEL; iLevel++) {
pos[iLevel] = pMemData->sl.pTail;
}
while (n < pSubmitBlk->nData) { while (n < pSubmitBlk->nData) {
nt = tGetTSRow(p + n, &row.tsRow); nt = tGetTSRow(p + n, &row.tsRow);
n += nt; n += nt;
ASSERT(n <= pSubmitBlk->nData); ASSERT(n <= pSubmitBlk->nData);
// build the node memDataMovePos(pMemData, &row, nRow ? 1 : 0, pos);
level = tsdbMemSkipListRandLevel(&pMemData->sl); code = memDataPutRow(pTsdb->pVnode->inUse, pMemData, &row, nRow ? 1 : 0, pos);
pNode = (SMemSkipListNode *)vnodeBufPoolMalloc(pPool, SL_NODE_SIZE(level) + nt + sizeof(version)); if (code) {
if (pNode == NULL) {
code = TSDB_CODE_OUT_OF_MEMORY;
goto _err; goto _err;
} }
pNode->level = level;
tPutTSDBRow((uint8_t *)SL_NODE_DATA(pNode), &row);
// put the node (todo) nRow++;
// set info
if (tsdbKeyCmprFn(&row, &pMemData->minKey) < 0) pMemData->minKey = *(TSDBKEY *)&row;
if (tsdbKeyCmprFn(&row, &pMemData->maxKey) > 0) pMemData->maxKey = *(TSDBKEY *)&row;
} }
if (tsdbKeyCmprFn(&pMemTable->minKey, &pMemData->minKey) < 0) pMemTable->minKey = pMemData->minKey;
if (tsdbKeyCmprFn(&pMemTable->maxKey, &pMemData->maxKey) > 0) pMemTable->maxKey = pMemData->maxKey;
return code; return code;
_err: _err:
@ -227,7 +220,7 @@ static int32_t tsdbGetOrCreateMemData(SMemTable *pMemTable, tb_uid_t suid, tb_ui
} }
// create // create
pMemData = vnodeBufPoolMalloc(pPool, sizeof(*pMemData) + SL_NODE_HALF_SIZE(maxLevel) * 2); pMemData = vnodeBufPoolMalloc(pPool, sizeof(*pMemData) + SL_NODE_SIZE(maxLevel) * 2);
if (pMemData == NULL) { if (pMemData == NULL) {
code = TSDB_CODE_OUT_OF_MEMORY; code = TSDB_CODE_OUT_OF_MEMORY;
goto _err; goto _err;
@ -242,11 +235,15 @@ static int32_t tsdbGetOrCreateMemData(SMemTable *pMemTable, tb_uid_t suid, tb_ui
pMemData->sl.maxLevel = maxLevel; pMemData->sl.maxLevel = maxLevel;
pMemData->sl.level = 0; pMemData->sl.level = 0;
pMemData->sl.pHead = (SMemSkipListNode *)&pMemData[1]; pMemData->sl.pHead = (SMemSkipListNode *)&pMemData[1];
pMemData->sl.pTail = (SMemSkipListNode *)POINTER_SHIFT(pMemData->sl.pHead, SL_NODE_HALF_SIZE(maxLevel)); pMemData->sl.pTail = (SMemSkipListNode *)POINTER_SHIFT(pMemData->sl.pHead, SL_NODE_SIZE(maxLevel));
pMemData->sl.pHead->level = maxLevel;
pMemData->sl.pTail->level = maxLevel;
for (int8_t iLevel = 0; iLevel < pMemData->sl.maxLevel; iLevel++) { for (int8_t iLevel = 0; iLevel < pMemData->sl.maxLevel; iLevel++) {
SL_HEAD_FORWARD(&pMemData->sl, iLevel) = pMemData->sl.pTail; SL_NODE_FORWARD(pMemData->sl.pHead, iLevel) = pMemData->sl.pTail;
SL_TAIL_BACKWARD(&pMemData->sl, iLevel) = pMemData->sl.pHead; SL_NODE_BACKWARD(pMemData->sl.pHead, iLevel) = NULL;
SL_NODE_BACKWARD(pMemData->sl.pTail, iLevel) = pMemData->sl.pHead;
SL_NODE_FORWARD(pMemData->sl.pTail, iLevel) = NULL;
} }
if (idx < 0) idx = 0; if (idx < 0) idx = 0;
@ -313,264 +310,87 @@ static FORCE_INLINE int8_t tsdbMemSkipListRandLevel(SMemSkipList *pSl) {
return level; return level;
} }
#if 0 //==================================================================================== static void memDataMovePos(SMemData *pMemData, TSDBROW *pRow, int8_t isForward, SMemSkipListNode **pos) {
TSDBKEY *pKey;
int c;
#define SL_MAX_LEVEL 5 if (isForward) {
// TODO
} else {
SMemSkipListNode *px = pMemData->sl.pTail;
struct SMemSkipListCurosr { for (int8_t iLevel = pMemData->sl.maxLevel - 1; iLevel >= 0; iLevel--) {
SMemSkipList *pSl; if (iLevel < pMemData->sl.level) {
SMemSkipListNode *pNodes[SL_MAX_LEVEL]; SMemSkipListNode *p = SL_NODE_BACKWARD(px, iLevel);
};
typedef struct { while (p != pMemData->sl.pHead) {
int64_t version; pKey = (TSDBKEY *)SL_NODE_DATA(p);
uint32_t szRow;
const STSRow *pRow;
} STsdbRow;
#define HASH_BUCKET(SUID, UID, NBUCKET) (TABS((SUID) + (UID)) % (NBUCKET)) c = tsdbKeyCmprFn(pKey, pRow);
if (c <= 0) {
#define SL_HEAD_NODE(sl) ((sl)->pHead) break;
#define SL_TAIL_NODE(sl) ((SMemSkipListNode *)&SL_NODE_FORWARD(SL_HEAD_NODE(sl), (sl)->maxLevel)) } else {
#define SL_HEAD_NODE_FORWARD(n, l) SL_NODE_FORWARD(n, l) px = p;
#define SL_TAIL_NODE_BACKWARD(n, l) SL_NODE_FORWARD(n, l) p = SL_NODE_BACKWARD(px, iLevel);
}
static int8_t tsdbMemSkipListRandLevel(SMemSkipList *pSl);
static int32_t tsdbEncodeRow(SEncoder *pEncoder, const STsdbRow *pRow);
static int32_t tsdbDecodeRow(SDecoder *pDecoder, STsdbRow *pRow);
static int32_t tsdbMemSkipListCursorCreate(int8_t maxLevel, SMemSkipListCurosr **ppSlc);
static void tsdbMemSkipListCursorDestroy(SMemSkipListCurosr *pSlc);
static void tsdbMemSkipListCursorInit(SMemSkipListCurosr *pSlc, SMemSkipList *pSl);
static void tsdbMemSkipListCursorPut(SMemSkipListCurosr *pSlc, SMemSkipListNode *pNode);
static int32_t tsdbMemSkipListCursorMoveTo(SMemSkipListCurosr *pSlc, int64_t version, TSKEY ts, int32_t flags);
static void tsdbMemSkipListCursorMoveToFirst(SMemSkipListCurosr *pSlc);
static void tsdbMemSkipListCursorMoveToLast(SMemSkipListCurosr *pSlc);
static int32_t tsdbMemSkipListCursorMoveToNext(SMemSkipListCurosr *pSlc);
static int32_t tsdbMemSkipListCursorMoveToPrev(SMemSkipListCurosr *pSlc);
static SMemSkipListNode *tsdbMemSkipListNodeCreate(SVBufPool *pPool, SMemSkipList *pSl, const STsdbRow *pTRow);
// SMemTable ========================
int32_t tsdbInsertData2(SMemTable *pMemTb, int64_t version, const SVSubmitBlk *pSubmitBlk) {
SMemData *pMemData;
STsdb *pTsdb = pMemTb->pTsdb;
SVnode *pVnode = pTsdb->pVnode;
SVBufPool *pPool = pVnode->inUse;
tb_uid_t suid = pSubmitBlk->suid;
tb_uid_t uid = pSubmitBlk->uid;
int32_t iBucket;
// search SMemData by hash
iBucket = HASH_BUCKET(suid, uid, pMemTb->nBucket);
for (pMemData = pMemTb->pBuckets[iBucket]; pMemData; pMemData = pMemData->pHashNext) {
if (pMemData->suid == suid && pMemData->uid == uid) break;
} }
// create pMemData if need pos[iLevel] = px;
if (pMemData == NULL) { }
int8_t maxLevel = pVnode->config.tsdbCfg.slLevel; }
int32_t tsize = sizeof(*pMemData) + SL_NODE_HALF_SIZE(maxLevel) * 2; }
SMemSkipListNode *pHead, *pTail;
pMemData = vnodeBufPoolMalloc(pPool, tsize);
if (pMemData == NULL) {
terrno = TSDB_CODE_OUT_OF_MEMORY;
return -1;
} }
pMemData->pHashNext = NULL; static void memMovePosFrom(SMemData *pMemData, SMemSkipListNode *pNode, TSDBROW *pRow, int8_t isForward,
pMemData->suid = suid; SMemSkipListNode **pos) {
pMemData->uid = uid; SMemSkipListNode *px = pNode;
pMemData->minKey = TSKEY_MAX; TSDBKEY *pKey;
pMemData->maxKey = TSKEY_MIN; SMemSkipListNode *p;
pMemData->minVer = -1; int c;
pMemData->maxVer = -1;
pMemData->nRows = 0; if (isForward) {
pMemData->sl.seed = taosRand(); } else {
pMemData->sl.maxLevel = maxLevel; ASSERT(pNode != pMemData->sl.pHead);
pMemData->sl.level = 0;
pMemData->sl.size = 0; for (int8_t iLevel = pMemData->sl.maxLevel - 1; iLevel >= 0; iLevel--) {
pHead = SL_HEAD_NODE(&pMemData->sl); p = SL_NODE_BACKWARD(px, iLevel);
pTail = SL_TAIL_NODE(&pMemData->sl); while (p != pMemData->sl.pHead) {
pHead->level = maxLevel; pKey = (TSDBKEY *)SL_NODE_DATA(p);
pTail->level = maxLevel;
for (int iLevel = 0; iLevel < maxLevel; iLevel++) { c = tsdbKeyCmprFn(pKey, pRow);
SL_HEAD_NODE_FORWARD(pHead, iLevel) = pTail; if (c <= 0) {
SL_TAIL_NODE_BACKWARD(pTail, iLevel) = pHead; break;
} else {
px = p;
p = SL_NODE_BACKWARD(px, iLevel);
}
} }
// add to hash pos[iLevel] = px;
if (pMemTb->nHash >= pMemTb->nBucket) { }
// rehash (todo)
} }
iBucket = HASH_BUCKET(suid, uid, pMemTb->nBucket);
pMemData->pHashNext = pMemTb->pBuckets[iBucket];
pMemTb->pBuckets[iBucket] = pMemData;
pMemTb->nHash++;
// sort organize (todo)
} }
// do insert data to SMemData static int32_t memDataPutRow(SVBufPool *pPool, SMemData *pMemData, TSDBROW *pRow, int8_t isForward,
SMemSkipListNode *forwards[SL_MAX_LEVEL]; SMemSkipListNode **pos) {
int32_t code = 0;
int8_t level;
SMemSkipListNode *pNode; SMemSkipListNode *pNode;
int32_t iRow;
STsdbRow tRow = {.version = version};
SEncoder ec = {0};
SDecoder dc = {0};
tDecoderInit(&dc, pSubmitBlk->pData, pSubmitBlk->nData); level = tsdbMemSkipListRandLevel(&pMemData->sl);
tsdbMemSkipListCursorInit(pMemTb->pSlc, &pMemData->sl); pNode = (SMemSkipListNode *)vnodeBufPoolMalloc(pPool, SL_NODE_SIZE(level) + tPutTSDBRow(NULL, pRow));
for (iRow = 0;; iRow++) {
if (tDecodeIsEnd(&dc)) break;
// decode row
if (tDecodeBinary(&dc, (uint8_t **)&tRow.pRow, &tRow.szRow) < 0) {
terrno = TSDB_CODE_INVALID_MSG;
return -1;
}
// move cursor
tsdbMemSkipListCursorMoveTo(pMemTb->pSlc, version, tRow.pRow->ts, 0);
// encode row
pNode = tsdbMemSkipListNodeCreate(pPool, &pMemData->sl, &tRow);
if (pNode == NULL) { if (pNode == NULL) {
terrno = TSDB_CODE_OUT_OF_MEMORY; code = TSDB_CODE_OUT_OF_MEMORY;
return -1; goto _exit;
} }
// put the node // do the read put
tsdbMemSkipListCursorPut(pMemTb->pSlc, pNode); if (isForward) {
// update status
if (tRow.pRow->ts < pMemData->minKey) pMemData->minKey = tRow.pRow->ts;
if (tRow.pRow->ts > pMemData->maxKey) pMemData->maxKey = tRow.pRow->ts;
}
tDecoderClear(&dc);
// update status
if (pMemData->minVer == -1) pMemData->minVer = version;
if (pMemData->maxVer == -1 || pMemData->maxVer < version) pMemData->maxVer = version;
if (pMemTb->minKey < pMemData->minKey) pMemTb->minKey = pMemData->minKey;
if (pMemTb->maxKey < pMemData->maxKey) pMemTb->maxKey = pMemData->maxKey;
if (pMemTb->minVer == -1) pMemTb->minVer = version;
if (pMemTb->maxVer == -1 || pMemTb->maxVer < version) pMemTb->maxVer = version;
return 0;
}
static FORCE_INLINE int32_t tsdbEncodeRow(SEncoder *pEncoder, const STsdbRow *pRow) {
if (tEncodeI64(pEncoder, pRow->version) < 0) return -1;
if (tEncodeBinary(pEncoder, (const uint8_t *)pRow->pRow, pRow->szRow) < 0) return -1;
return 0;
}
static FORCE_INLINE int32_t tsdbDecodeRow(SDecoder *pDecoder, STsdbRow *pRow) {
if (tDecodeI64(pDecoder, &pRow->version) < 0) return -1;
if (tDecodeBinary(pDecoder, (uint8_t **)&pRow->pRow, &pRow->szRow) < 0) return -1;
return 0;
}
static int32_t tsdbMemSkipListCursorCreate(int8_t maxLevel, SMemSkipListCurosr **ppSlc) {
*ppSlc = (SMemSkipListCurosr *)taosMemoryCalloc(1, sizeof(**ppSlc) + sizeof(SMemSkipListNode *) * maxLevel);
if (*ppSlc == NULL) {
return -1;
}
return 0;
}
static void tsdbMemSkipListCursorDestroy(SMemSkipListCurosr *pSlc) { taosMemoryFree(pSlc); }
static void tsdbMemSkipListCursorInit(SMemSkipListCurosr *pSlc, SMemSkipList *pSl) {
SMemSkipListNode *pHead = SL_HEAD_NODE(pSl);
pSlc->pSl = pSl;
// for (int8_t iLevel = 0; iLevel < pSl->maxLevel; iLevel++) {
// pSlc->forwards[iLevel] = pHead;
// }
}
static void tsdbMemSkipListCursorPut(SMemSkipListCurosr *pSlc, SMemSkipListNode *pNode) {
SMemSkipList *pSl = pSlc->pSl;
SMemSkipListNode *pNodeNext;
for (int8_t iLevel = 0; iLevel < pNode->level; iLevel++) {
// todo
ASSERT(0);
}
if (pSl->level < pNode->level) {
pSl->level = pNode->level;
}
pSl->size += 1;
}
static int32_t tsdbMemSkipListCursorMoveTo(SMemSkipListCurosr *pSlc, int64_t version, TSKEY ts, int32_t flags) {
SMemSkipListNode **pForwards = NULL;
SMemSkipList *pSl = pSlc->pSl;
int8_t maxLevel = pSl->maxLevel;
SMemSkipListNode *pHead = SL_HEAD_NODE(pSl);
SMemSkipListNode *pTail = SL_TAIL_NODE(pSl);
if (pSl->size == 0) {
for (int8_t iLevel = 0; iLevel < pSl->maxLevel; iLevel++) {
pForwards[iLevel] = pHead;
}
}
return 0;
}
static void tsdbMemSkipListCursorMoveToFirst(SMemSkipListCurosr *pSlc) {
SMemSkipList *pSl = pSlc->pSl;
SMemSkipListNode *pHead = SL_HEAD_NODE(pSl);
for (int8_t iLevel = 0; iLevel < pSl->maxLevel; iLevel++) {
pSlc->pNodes[iLevel] = pHead;
}
tsdbMemSkipListCursorMoveToNext(pSlc);
}
static void tsdbMemSkipListCursorMoveToLast(SMemSkipListCurosr *pSlc) {
SMemSkipList *pSl = pSlc->pSl;
SMemSkipListNode *pTail = SL_TAIL_NODE(pSl);
for (int8_t iLevel = 0; iLevel < pSl->maxLevel; iLevel++) {
pSlc->pNodes[iLevel] = pTail;
}
tsdbMemSkipListCursorMoveToPrev(pSlc);
}
static int32_t tsdbMemSkipListCursorMoveToNext(SMemSkipListCurosr *pSlc) {
// TODO // TODO
return 0; } else {
}
static int32_t tsdbMemSkipListCursorMoveToPrev(SMemSkipListCurosr *pSlc) {
// TODO // TODO
return 0;
} }
static SMemSkipListNode *tsdbMemSkipListNodeCreate(SVBufPool *pPool, SMemSkipList *pSl, const STsdbRow *pTRow) { _exit:
int32_t tsize; return code;
int32_t ret;
int8_t level = tsdbMemSkipListRandLevel(pSl);
SMemSkipListNode *pNode = NULL;
SEncoder ec = {0};
tEncodeSize(tsdbEncodeRow, pTRow, tsize, ret);
pNode = vnodeBufPoolMalloc(pPool, tsize + SL_NODE_SIZE(level));
if (pNode) {
pNode->level = level;
tEncoderInit(&ec, (uint8_t *)SL_NODE_DATA(pNode), tsize);
tsdbEncodeRow(&ec, pTRow);
tEncoderClear(&ec);
} }
return pNode;
}
#endif