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

more data

This commit is contained in:
Hongze Cheng 2022-05-13 01:19:34 +00:00
parent 71c868087d
commit 038e9558bb
2 changed files with 140 additions and 91 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

9
include/common/tdataformat.h
View File

@ -86,16 +86,17 @@ struct STSRow2 {
}; };
struct STSRowBuilder { struct STSRowBuilder {
STColumn *pTColumn;
STSchema *pTSchema; STSchema *pTSchema;
int32_t szBitMap1;
int32_t szBitMap2;
int32_t szKVBuf; int32_t szKVBuf;
uint8_t *pKVBuf; uint8_t *pKVBuf;
int32_t szTPBuf; int32_t szTPBuf;
uint8_t *pTPBuf; uint8_t *pTPBuf;
uint8_t *pBitBuf; int32_t iCol;
int32_t nCols; int32_t nCols;
int32_t kvVLen; int32_t vlenKV;
int32_t tpVLen; int32_t vlenTP;
STSRow2 row; STSRow2 row;
}; };

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

@ -25,6 +25,8 @@ struct SKVIdx {
}; };
#define TSROW_IS_KV_ROW(r) ((r)->flags & TSROW_KV_ROW) #define TSROW_IS_KV_ROW(r) ((r)->flags & TSROW_KV_ROW)
#define SET_BIT1(p, i, v)
#define SET_BIT2(p, i, v)
// STSRow2 // STSRow2
int32_t tEncodeTSRow(SEncoder *pEncoder, const STSRow2 *pRow) { int32_t tEncodeTSRow(SEncoder *pEncoder, const STSRow2 *pRow) {
@ -170,126 +172,135 @@ int32_t tTSchemaCreate(int32_t sver, SSchema *pSchema, int32_t ncols, STSchema *
return 0; return 0;
} }
void tTSchemaDestroy(STSchema *pTSchema) { taosMemoryFree(pTSchema); } void tTSchemaDestroy(STSchema *pTSchema) {
if (pTSchema) taosMemoryFree(pTSchema);
}
// STSRowBuilder // STSRowBuilder
int32_t tTSRowBuilderInit(STSRowBuilder *pBuilder, int32_t sver, SSchema *pSchema, int32_t nCols) { static int32_t tTSRowBitMapLen1(int32_t nCols) { return nCols / 8 + ((nCols % 8) == 0 ? 0 : 1); }
int32_t kvBufLen; static int32_t tTSRowBitMapLen2(int32_t nCols) { return nCols / 4 + ((nCols % 4) == 0 ? 0 : 1); }
int32_t tpBufLen;
uint8_t *p;
int32_t tTSRowBuilderInit(STSRowBuilder *pBuilder, int32_t sver, SSchema *pSchema, int32_t nCols) {
if (tTSchemaCreate(sver, pSchema, nCols, &pBuilder->pTSchema) < 0) return -1; if (tTSchemaCreate(sver, pSchema, nCols, &pBuilder->pTSchema) < 0) return -1;
kvBufLen = sizeof(SKVIdx) * nCols + pBuilder->pTSchema->flen + pBuilder->pTSchema->vlen; pBuilder->szBitMap1 = tTSRowBitMapLen1(nCols - 1);
tpBufLen = pBuilder->pTSchema->flen + pBuilder->pTSchema->vlen; pBuilder->szBitMap2 = tTSRowBitMapLen2(nCols - 1);
pBuilder->szKVBuf = sizeof(SKVIdx) * (nCols - 1) + pBuilder->pTSchema->flen + pBuilder->pTSchema->vlen;
if (pBuilder->szKVBuf < kvBufLen) { pBuilder->szTPBuf = pBuilder->szBitMap2 + pBuilder->pTSchema->flen + pBuilder->pTSchema->vlen;
p = taosMemoryRealloc(pBuilder->pKVBuf, kvBufLen); pBuilder->pKVBuf = taosMemoryMalloc(pBuilder->szKVBuf);
if (p == NULL) { if (pBuilder->pKVBuf == NULL) {
terrno = TSDB_CODE_TSC_OUT_OF_MEMORY;
return -1;
}
pBuilder->pTPBuf = taosMemoryMalloc(pBuilder->szTPBuf);
if (pBuilder->pTPBuf == NULL) {
terrno = TSDB_CODE_OUT_OF_MEMORY; terrno = TSDB_CODE_OUT_OF_MEMORY;
return -1; return -1;
} }
pBuilder->pKVBuf = p;
pBuilder->szKVBuf = kvBufLen;
}
if (pBuilder->szTPBuf < tpBufLen) {
p = taosMemoryRealloc(pBuilder->pTPBuf, tpBufLen);
if (p == NULL) {
terrno = TSDB_CODE_OUT_OF_MEMORY;
return -1;
}
pBuilder->pTPBuf = p;
pBuilder->szTPBuf = tpBufLen;
}
tTSRowBuilderReset(pBuilder);
return 0; return 0;
} }
void tTSRowBuilderClear(STSRowBuilder *pBuilder) { void tTSRowBuilderClear(STSRowBuilder *pBuilder) {
tTSchemaDestroy(pBuilder->pTSchema);
pBuilder->pTSchema = NULL;
if (pBuilder->pKVBuf) {
taosMemoryFree(pBuilder->pKVBuf); taosMemoryFree(pBuilder->pKVBuf);
pBuilder->pKVBuf = NULL;
}
if (pBuilder->pTPBuf) {
taosMemoryFree(pBuilder->pTPBuf); taosMemoryFree(pBuilder->pTPBuf);
pBuilder->pTPBuf = NULL;
}
} }
void tTSRowBuilderReset(STSRowBuilder *pBuilder) { void tTSRowBuilderReset(STSRowBuilder *pBuilder) {
for (int32_t iCol = pBuilder->pTSchema->numOfCols - 1; iCol >= 0; iCol--) { for (int32_t iCol = pBuilder->pTSchema->numOfCols - 1; iCol >= 0; iCol--) {
pBuilder->pTColumn = &pBuilder->pTSchema->columns[iCol]; STColumn *pTColumn = &pBuilder->pTSchema->columns[iCol];
COL_CLR_SET(pTColumn->flags);
pBuilder->pTColumn->flags &= 0xf;
} }
pBuilder->iCol = 0;
pBuilder->nCols = 0; pBuilder->nCols = 0;
pBuilder->kvVLen = 0; pBuilder->vlenKV = 0;
pBuilder->tpVLen = 0; pBuilder->vlenTP = 0;
pBuilder->row.flags = 0; pBuilder->row.flags = 0;
} }
int32_t tTSRowBuilderPut(STSRowBuilder *pBuilder, int32_t cid, const uint8_t *pData, uint32_t nData) { int32_t tTSRowBuilderPut(STSRowBuilder *pBuilder, int32_t cid, const uint8_t *pData, uint32_t nData) {
// search column (TODO: bsearch with interp) STColumn *pTColumn = &pBuilder->pTSchema->columns[pBuilder->iCol];
if (pBuilder->pTColumn->colId < cid) { uint8_t *p;
int32_t iCol = (pBuilder->pTColumn - pBuilder->pTSchema->columns) / sizeof(STColumn) + 1; int32_t iCol;
for (; iCol < pBuilder->pTSchema->numOfCols; iCol++) {
pBuilder->pTColumn = &pBuilder->pTSchema->columns[iCol]; // use interp search (todo)
if (pBuilder->pTColumn->colId == cid) break; if (pTColumn->colId > cid) {
for (iCol = pBuilder->iCol + 1; iCol < pBuilder->pTSchema->numOfCols; iCol++) {
pTColumn = &pBuilder->pTSchema->columns[iCol];
if (pTColumn->colId == cid) break;
} }
} else if (pBuilder->pTColumn->colId > cid) { } else if (pTColumn->colId < cid) {
int32_t iCol = (pBuilder->pTColumn - pBuilder->pTSchema->columns) / sizeof(STColumn) - 1; for (iCol = pBuilder->iCol - 1; iCol >= 0; iCol--) {
for (; iCol >= 0; iCol--) { pTColumn = &pBuilder->pTSchema->columns[iCol];
pBuilder->pTColumn = &pBuilder->pTSchema->columns[iCol]; if (pTColumn->colId == cid) break;
if (pBuilder->pTColumn->colId == cid) break;
} }
} }
// check if (pTColumn->colId != cid || COL_IS_SET(pTColumn->flags)) {
if (pBuilder->pTColumn->colId != cid || COL_IS_SET(pBuilder->pTColumn->flags)) {
return -1; return -1;
} }
pBuilder->iCol = iCol;
// set value // set value
uint8_t *p;
if (cid == 0) { if (cid == 0) {
ASSERT(pData && nData == sizeof(TSKEY)); ASSERT(pData && nData == sizeof(TSKEY) && iCol == 0);
pBuilder->row.ts = *(TSKEY *)pData; pBuilder->row.ts = *(TSKEY *)pData;
pTColumn->flags |= COL_SET_VAL;
pBuilder->pTColumn->flags |= COL_SET_VAL;
} else { } else {
if (pData) { // set val if (pData) {
// set VAL
pBuilder->row.flags |= TSROW_HAS_VAL; pBuilder->row.flags |= TSROW_HAS_VAL;
pBuilder->pTColumn->flags |= COL_SET_VAL; pTColumn->flags |= COL_SET_VAL;
// set tuple data /* KV */
p = pBuilder->pTPBuf + pBuilder->pTColumn->offset; if (1) { // avoid KV at some threshold (todo)
if (IS_VAR_DATA_TYPE(pBuilder->pTColumn->type)) {
*(int32_t *)p = pBuilder->tpVLen;
// encode the variant-length data
p = pBuilder->pTPBuf + pBuilder->pTSchema->flen + pBuilder->tpVLen;
pBuilder->tpVLen += tPutBinary(p, pData, nData);
} else {
memcpy(p, pData, nData);
}
// set kv data
p = pBuilder->pKVBuf + sizeof(SKVIdx) * pBuilder->nCols; p = pBuilder->pKVBuf + sizeof(SKVIdx) * pBuilder->nCols;
((SKVIdx *)p)->cid = cid; ((SKVIdx *)p)->cid = cid;
((SKVIdx *)p)->offset = pBuilder->kvVLen; ((SKVIdx *)p)->offset = pBuilder->vlenKV;
p = pBuilder->pKVBuf + sizeof(SKVIdx) * (pBuilder->pTSchema->numOfCols - 1) + pBuilder->kvVLen; p = pBuilder->pKVBuf + sizeof(SKVIdx) * (pBuilder->pTSchema->numOfCols - 1) + pBuilder->vlenKV;
if (IS_VAR_DATA_TYPE(pBuilder->pTColumn->type)) { if (IS_VAR_DATA_TYPE(pTColumn->type)) {
pBuilder->kvVLen += tPutBinary(p, pData, nData); ASSERT(nData <= pTColumn->bytes);
pBuilder->vlenKV += tPutBinary(p, pData, nData);
} else { } else {
ASSERT(nData == pTColumn->bytes);
memcpy(p, pData, nData); memcpy(p, pData, nData);
pBuilder->kvVLen += nData; pBuilder->vlenKV += nData;
} }
} else { // set NULL }
/* 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 += tPutBinary(p, pData, nData);
} else {
ASSERT(nData == pTColumn->bytes);
memcpy(p, pData, nData);
}
} else {
// set NULL
pBuilder->row.flags |= TSROW_HAS_NULL; pBuilder->row.flags |= TSROW_HAS_NULL;
pBuilder->pTColumn->flags |= COL_SET_NULL; pTColumn->flags |= COL_SET_NULL;
p = pBuilder->pKVBuf + sizeof(SKVIdx) * pBuilder->nCols; p = pBuilder->pKVBuf + sizeof(SKVIdx) * pBuilder->nCols;
((SKVIdx *)p)->cid = cid; ((SKVIdx *)p)->cid = cid;
((SKVIdx *)p)->offset = -1; // for TSROW_KV_ROW, offset -1 means NULL ((SKVIdx *)p)->offset = -1;
} }
pBuilder->nCols++; pBuilder->nCols++;
@ -309,7 +320,7 @@ static FORCE_INLINE int tSKVIdxCmprFn(const void *p1, const void *p2) {
return 0; return 0;
} }
int32_t tTSRowBuilderGetRow(STSRowBuilder *pBuilder, const STSRow2 **ppRow) { int32_t tTSRowBuilderGetRow(STSRowBuilder *pBuilder, const STSRow2 **ppRow) {
int32_t tpDataLen, kvDataLen; int32_t nDataTP, nDataKV;
uint32_t flags; uint32_t flags;
// error not set ts // error not set ts
@ -332,39 +343,76 @@ int32_t tTSRowBuilderGetRow(STSRowBuilder *pBuilder, const STSRow2 **ppRow) {
pBuilder->row.pData = NULL; pBuilder->row.pData = NULL;
return 0; return 0;
case TSROW_HAS_NULL | TSROW_HAS_NONE: case TSROW_HAS_NULL | TSROW_HAS_NONE:
tpDataLen = (pBuilder->pTSchema->numOfCols - 1) / 8; nDataTP = pBuilder->szBitMap1;
break; break;
case TSROW_HAS_VAL: case TSROW_HAS_VAL:
tpDataLen = pBuilder->pTSchema->flen + pBuilder->tpVLen; nDataTP = pBuilder->pTSchema->flen + pBuilder->vlenTP;
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:
tpDataLen = pBuilder->pTSchema->flen + pBuilder->tpVLen + (pBuilder->pTSchema->numOfCols - 1) / 8; nDataTP = pBuilder->szBitMap1 + pBuilder->pTSchema->flen + pBuilder->vlenTP;
break; break;
case TSROW_HAS_VAL | TSROW_HAS_NULL | TSROW_HAS_NONE: case TSROW_HAS_VAL | TSROW_HAS_NULL | TSROW_HAS_NONE:
tpDataLen = pBuilder->pTSchema->flen + pBuilder->tpVLen + (pBuilder->pTSchema->numOfCols - 1) / 4; nDataTP = pBuilder->szBitMap2 + pBuilder->pTSchema->flen + pBuilder->vlenTP;
break; break;
default: default:
// decide chose tuple or kv ASSERT(0);
kvDataLen = sizeof(SKVIdx) * pBuilder->nCols + pBuilder->kvVLen;
break;
} }
if (kvDataLen < tpDataLen) { nDataKV = sizeof(SKVIdx) * pBuilder->nCols + pBuilder->vlenKV;
ASSERT(pBuilder->row.flags & TSROW_KV_ROW == 0);
if (nDataKV < nDataTP) {
// generate KV row
pBuilder->row.flags |= TSROW_KV_ROW; pBuilder->row.flags |= TSROW_KV_ROW;
pBuilder->row.ncols = pBuilder->nCols; pBuilder->row.ncols = pBuilder->nCols;
pBuilder->row.nData = nDataKV;
pBuilder->row.nData = kvDataLen;
pBuilder->row.pData = pBuilder->pKVBuf; pBuilder->row.pData = pBuilder->pKVBuf;
qsort(pBuilder->pKVBuf, pBuilder->nCols, sizeof(SKVIdx), tSKVIdxCmprFn); qsort(pBuilder->pKVBuf, pBuilder->nCols, sizeof(SKVIdx), tSKVIdxCmprFn);
if (pBuilder->nCols < pBuilder->pTSchema->numOfCols - 1) { if (pBuilder->nCols < pBuilder->pTSchema->numOfCols - 1) {
memmove(pBuilder->pKVBuf + sizeof(SKVIdx) * pBuilder->nCols, memmove(pBuilder->pKVBuf + sizeof(SKVIdx) * pBuilder->nCols,
pBuilder->pKVBuf + sizeof(SKVIdx) * (pBuilder->pTSchema->numOfCols - 1), pBuilder->kvVLen); pBuilder->pKVBuf + sizeof(SKVIdx) * (pBuilder->pTSchema->numOfCols - 1), pBuilder->vlenKV);
} }
} else { } else {
pBuilder->row.sver = pBuilder->pTSchema->version; // generate TUPLE row
pBuilder->row.nData = tpDataLen; uint8_t *p;
STColumn *pTColumn;
pBuilder->row.sver = pBuilder->pTSchema->version;
pBuilder->row.nData = nDataTP;
if (pBuilder->row.flags & 0xf == TSROW_HAS_VAL) {
// no bitmap
pBuilder->row.pData = pBuilder->pTPBuf + pBuilder->szBitMap2;
} else if (pBuilder->row.flags & 0xf == TSROW_HAS_VAL | TSROW_HAS_NULL | TSROW_HAS_NONE) {
// bitmap2
p = pBuilder->pTPBuf;
for (int32_t iCol = 1; iCol < pBuilder->pTSchema->numOfCols; iCol++) {
pTColumn = &pBuilder->pTSchema->columns[iCol];
if (pTColumn->flags & COL_SET_VAL) {
SET_BIT2(p, iCol - 1, 0x2);
} else if (pTColumn->flags & COL_SET_VAL) {
SET_BIT2(p, iCol - 1, 0x1);
}
}
pBuilder->row.pData = p;
} else {
// bitmap1
p = pBuilder->pTPBuf + pBuilder->szBitMap2 - pBuilder->szBitMap1;
for (int32_t iCol = 1; iCol < pBuilder->pTSchema->numOfCols; iCol++) {
pTColumn = &pBuilder->pTSchema->columns[iCol];
if (1) {
SET_BIT1(p, iCol - 1, 0x1);
}
}
pBuilder->row.pData = p;
}
} }
return 0; return 0;