diff --git a/include/common/tdataformat.h b/include/common/tdataformat.h index 1f3b787538..2e8a214a9d 100644 --- a/include/common/tdataformat.h +++ b/include/common/tdataformat.h @@ -18,6 +18,7 @@ #include "os.h" #include "talgo.h" +#include "tencode.h" #include "ttypes.h" #include "tutil.h" @@ -25,6 +26,73 @@ extern "C" { #endif +typedef struct SSchema SSchema; +typedef struct STColumn STColumn; +typedef struct STSchema STSchema; +typedef struct STSRow2 STSRow2; +typedef struct STSRowBuilder STSRowBuilder; +typedef struct SKVIdx SKVIdx; + +// STSchema + +// STSRow2 +int32_t tEncodeTSRow(SEncoder *pEncoder, const STSRow2 *pRow); +int32_t tDecodeTSRow(SDecoder *pDecoder, STSRow2 *pRow); + +// STSchema +int32_t tTSchemaCreate(int32_t sver, SSchema *pSchema, int32_t nCols, STSchema **ppTSchema); +void tTSchemaDestroy(STSchema *pTSchema); + +// STSRowBuilder +int32_t tTSRowBuilderInit(STSRowBuilder *pBuilder, int32_t sver, SSchema *pSchema, int32_t nCols); +void tTSRowBuilderClear(STSRowBuilder *pBuilder); +void tTSRowBuilderReset(STSRowBuilder *pBuilder); +int32_t tTSRowBuilderPut(STSRowBuilder *pBuilder, int32_t cid, const uint8_t *pData, uint32_t nData); +int32_t tTSRowBuilderGetRow(STSRowBuilder *pBuilder, const STSRow2 **ppRow); + +// STRUCT ================= +struct STColumn { + col_id_t colId; + int8_t type; + int8_t flags; + int32_t bytes; + int32_t offset; +}; + +struct STSchema { + int32_t numOfCols; + int32_t version; + int32_t flen; + int32_t vlen; + int32_t tlen; + STColumn columns[]; +}; + +struct STSRow2 { + TSKEY ts; + uint32_t flags; + union { + int32_t sver; + int32_t ncols; + }; + uint32_t nData; + const uint8_t *pData; +}; + +struct STSRowBuilder { + STColumn *pTColumn; + STSchema *pTSchema; + int32_t szKVBuf; + uint8_t *pKVBuf; + int32_t szTPBuf; + uint8_t *pTPBuf; + int32_t nCols; + int32_t kvVLen; + int32_t tpVLen; + STSRow2 row; +}; + +#if 1 //==================================== // Imported since 3.0 and use bitmap to demonstrate None/Null/Norm, while use Null/Norm below 3.0 without of bitmap. #define TD_SUPPORT_BITMAP #define TD_SUPPORT_READ2 @@ -59,15 +127,6 @@ extern "C" { } while (0); // ----------------- TSDB COLUMN DEFINITION -#pragma pack(push, 1) -typedef struct { - col_id_t colId; // column ID(start from PRIMARYKEY_TIMESTAMP_COL_ID(1)) - int8_t type; // column type - int8_t flags; // flags: 0 no index, 1 SCHEMA_SMA_ON, 2 SCHEMA_IDX_ON - int32_t bytes; // column bytes (0~16M) - int32_t offset; // point offset in STpRow after the header part. -} STColumn; -#pragma pack(pop) #define colType(col) ((col)->type) #define colFlags(col) ((col)->flags) @@ -82,15 +141,6 @@ typedef struct { #define colSetOffset(col, o) (colOffset(col) = (o)) // ----------------- TSDB SCHEMA DEFINITION -typedef struct { - int32_t numOfCols; // Number of columns appended - schema_ver_t version; // schema version - uint16_t flen; // First part length in a STpRow after the header part - int32_t vlen; // pure value part length, excluded the overhead (bytes only) - int32_t tlen; // maximum length of a STpRow without the header part - // (sizeof(VarDataOffsetT) + sizeof(VarDataLenT) + (bytes)) - STColumn columns[]; -} STSchema; #define schemaNCols(s) ((s)->numOfCols) #define schemaVersion(s) ((s)->version) @@ -386,6 +436,7 @@ static FORCE_INLINE int32_t tdAddColToKVRow(SKVRowBuilder *pBuilder, col_id_t co return 0; } +#endif #ifdef __cplusplus } diff --git a/include/common/tmsg.h b/include/common/tmsg.h index b50367af03..d946fe6a98 100644 --- a/include/common/tmsg.h +++ b/include/common/tmsg.h @@ -267,8 +267,9 @@ typedef struct { SSubmitRspBlock failedBlocks[]; } SSubmitRsp; -#define SCHEMA_SMA_ON 0x1 -#define SCHEMA_IDX_ON 0x2 +#define COL_SMA_ON ((int8_t)0x1) +#define COL_IDX_ON ((int8_t)0x2) +#define COL_VAL_SET ((int8_t)0x4) typedef struct SSchema { int8_t type; int8_t flags; @@ -277,7 +278,7 @@ typedef struct SSchema { char name[TSDB_COL_NAME_LEN]; } SSchema; -#define IS_BSMA_ON(s) (((s)->flags & 0x01) == SCHEMA_SMA_ON) +#define IS_BSMA_ON(s) (((s)->flags & 0x01) == COL_SMA_ON) #define SSCHMEA_TYPE(s) ((s)->type) #define SSCHMEA_FLAGS(s) ((s)->flags) diff --git a/include/util/tencode.h b/include/util/tencode.h index 8504aec5ea..2a43d7934f 100644 --- a/include/util/tencode.h +++ b/include/util/tencode.h @@ -457,6 +457,55 @@ static FORCE_INLINE void* tDecoderMalloc(SDecoder* pCoder, int32_t size) { return p; } +static FORCE_INLINE int32_t tPutBinary(uint8_t* p, const uint8_t* pData, uint32_t nData) { + int n = 0; + uint32_t v = nData; + + for (;;) { + if (v <= 0x7f) { + if (p) p[n] = v; + n++; + break; + } + + if (p) p[n] = (v & 0x7f) | 0x80; + n++; + v >>= 7; + } + + if (p) { + memcpy(p + n, pData, nData); + } + n += nData; + + return n; +} + +static FORCE_INLINE int32_t tGetBinary(const uint8_t* p, const uint8_t** ppData, uint32_t* nData) { + int32_t n = 0; + uint32_t tv = 0; + uint32_t t; + + for (;;) { + if (p[n] <= 0x7f) { + t = p[n]; + tv |= (t << (7 * n)); + n++; + break; + } + + t = p[n] & 0x7f; + tv |= (t << (7 * n)); + n++; + } + + if (nData) *nData = n; + if (ppData) *ppData = p + n; + + n += tv; + return n; +} + #ifdef __cplusplus } #endif diff --git a/source/common/src/tdataformat.c b/source/common/src/tdataformat.c index 5d893fe398..ad020fe7d9 100644 --- a/source/common/src/tdataformat.c +++ b/source/common/src/tdataformat.c @@ -19,12 +19,238 @@ #include "tdatablock.h" #include "tlog.h" +#define TD_KV_ROW 0x1U + +struct SKVIdx { + int32_t cid; + int32_t offset; +}; + +int32_t tEncodeTSRow(SEncoder *pEncoder, const STSRow2 *pRow) { + if (tEncodeI64(pEncoder, pRow->ts) < 0) return -1; + if (tEncodeU32v(pEncoder, pRow->flags) < 0) return -1; + if (pRow->flags & TD_KV_ROW) { + if (tEncodeI32v(pEncoder, pRow->ncols) < 0) return -1; + } else { + if (tEncodeI32v(pEncoder, pRow->sver) < 0) return -1; + } + if (tEncodeBinary(pEncoder, pRow->pData, pRow->nData) < 0) return -1; + return 0; +} + +int32_t tDecodeTSRow(SDecoder *pDecoder, STSRow2 *pRow) { + if (tDecodeI64(pDecoder, &pRow->ts) < 0) return -1; + if (tDecodeU32v(pDecoder, &pRow->flags) < 0) return -1; + if (pRow->flags & TD_KV_ROW) { + if (tDecodeI32v(pDecoder, &pRow->ncols) < 0) return -1; + } else { + if (tDecodeI32v(pDecoder, &pRow->sver) < 0) return -1; + } + if (tDecodeBinary(pDecoder, &pRow->pData, &pRow->nData) < 0) return -1; + return 0; +} + +int32_t tTSchemaCreate(int32_t sver, SSchema *pSchema, int32_t ncols, STSchema **ppTSchema) { + *ppTSchema = (STSchema *)taosMemoryMalloc(sizeof(STSchema) + sizeof(STColumn) * ncols); + if (*ppTSchema == NULL) { + terrno = TSDB_CODE_OUT_OF_MEMORY; + return -1; + } + + (*ppTSchema)->numOfCols = ncols; + (*ppTSchema)->version = sver; + (*ppTSchema)->flen = 0; + (*ppTSchema)->vlen = 0; + (*ppTSchema)->tlen = 0; + + for (int32_t iCol = 0; iCol < ncols; iCol++) { + SSchema *pColumn = &pSchema[iCol]; + STColumn *pTColumn = &((*ppTSchema)->columns[iCol]); + + pTColumn->colId = pColumn->colId; + pTColumn->type = pColumn->type; + pTColumn->flags = pColumn->flags; + pTColumn->bytes = pColumn->bytes; + pTColumn->offset = (*ppTSchema)->flen; + + // skip first column + if (iCol) { + (*ppTSchema)->flen += TYPE_BYTES[pColumn->type]; + if (IS_VAR_DATA_TYPE(pColumn->type)) { + (*ppTSchema)->vlen += (pColumn->bytes + 5); + } + } + } + + return 0; +} + +void tTSchemaDestroy(STSchema *pTSchema) { taosMemoryFree(pTSchema); } + +int32_t tTSRowBuilderInit(STSRowBuilder *pBuilder, int32_t sver, SSchema *pSchema, int32_t nCols) { + int32_t kvBufLen; + int32_t tpBufLen; + uint8_t *p; + + if (tTSchemaCreate(sver, pSchema, nCols, &pBuilder->pTSchema) < 0) return -1; + + kvBufLen = sizeof(SKVIdx) * nCols + pBuilder->pTSchema->flen + pBuilder->pTSchema->vlen; + tpBufLen = pBuilder->pTSchema->flen + pBuilder->pTSchema->vlen; + + if (pBuilder->szKVBuf < kvBufLen) { + p = taosMemoryRealloc(pBuilder->pKVBuf, kvBufLen); + if (p == NULL) { + terrno = TSDB_CODE_OUT_OF_MEMORY; + 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; +} + +void tTSRowBuilderClear(STSRowBuilder *pBuilder) { + taosMemoryFree(pBuilder->pKVBuf); + taosMemoryFree(pBuilder->pTPBuf); +} + +void tTSRowBuilderReset(STSRowBuilder *pBuilder) { + for (int32_t iCol = pBuilder->pTSchema->numOfCols - 1; iCol >= 0; iCol--) { + pBuilder->pTColumn = &pBuilder->pTSchema->columns[iCol]; + + pBuilder->pTColumn->flags &= (~COL_VAL_SET); + } + + pBuilder->nCols = 0; + pBuilder->kvVLen = 0; + pBuilder->tpVLen = 0; + pBuilder->row.flags = 0; +} + +int32_t tTSRowBuilderPut(STSRowBuilder *pBuilder, int32_t cid, const uint8_t *pData, uint32_t nData) { + int32_t iCol; + uint8_t *p; + + // search column + if (pBuilder->pTColumn->colId < cid) { + iCol = (pBuilder->pTColumn - pBuilder->pTSchema->columns) / sizeof(STColumn) + 1; + for (; iCol < pBuilder->pTSchema->numOfCols; iCol++) { + pBuilder->pTColumn = &pBuilder->pTSchema->columns[iCol]; + if (pBuilder->pTColumn->colId == cid) break; + } + } else if (pBuilder->pTColumn->colId > cid) { + iCol = (pBuilder->pTColumn - pBuilder->pTSchema->columns) / sizeof(STColumn) - 1; + for (; iCol >= 0; iCol--) { + pBuilder->pTColumn = &pBuilder->pTSchema->columns[iCol]; + if (pBuilder->pTColumn->colId == cid) break; + } + } + + // check + if (pBuilder->pTColumn->colId != cid || pBuilder->pTColumn->flags & COL_VAL_SET) { + return -1; + } + + // set value + if (cid == 0) { + ASSERT(pData && nData == sizeof(TSKEY)); + pBuilder->row.ts = *(TSKEY *)pData; + } else { + if (pData) { + // ASSERT(!IS_NULL(pData)); + + // set tuple data + p = pBuilder->pTPBuf + pBuilder->pTColumn->offset; + 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; + ((SKVIdx *)p)->cid = cid; + ((SKVIdx *)p)->offset = pBuilder->kvVLen; + + p = pBuilder->pKVBuf + sizeof(SKVIdx) * pBuilder->pTSchema->numOfCols + pBuilder->kvVLen; + if (IS_VAR_DATA_TYPE(pBuilder->pTColumn->type)) { + pBuilder->kvVLen += tPutBinary(p, pData, nData); + } else { + memcpy(p, pData, nData); + pBuilder->kvVLen += nData; + } + } else { + // set NULL val + } + } + + pBuilder->pTColumn->flags |= COL_VAL_SET; + pBuilder->nCols++; + return 0; +} + +int32_t tTSRowBuilderGetRow(STSRowBuilder *pBuilder, const STSRow2 **ppRow) { + if ((pBuilder->pTSchema->columns[0].flags & COL_VAL_SET) == 0) { + return -1; + } + + if (pBuilder->nCols * sizeof(SKVIdx) + pBuilder->kvVLen < pBuilder->pTSchema->flen + pBuilder->tpVLen) { + // encode as TD_KV_ROW + pBuilder->row.flags |= TD_KV_ROW; + pBuilder->row.ncols = pBuilder->nCols; + pBuilder->row.nData = pBuilder->nCols * sizeof(SKVIdx) + pBuilder->kvVLen; + pBuilder->row.pData = pBuilder->pKVBuf; + + if (pBuilder->nCols < pBuilder->pTSchema->numOfCols) { + memmove(pBuilder->pKVBuf + sizeof(SKVIdx) * pBuilder->nCols, + pBuilder->pKVBuf + sizeof(SKVIdx) * pBuilder->pTSchema->numOfCols, pBuilder->kvVLen); + } + } else { + // encode as TD_TUPLE_ROW + pBuilder->row.flags &= (~TD_KV_ROW); + pBuilder->row.sver = pBuilder->pTSchema->version; + pBuilder->row.nData = pBuilder->pTSchema->flen + pBuilder->tpVLen; + pBuilder->row.pData = pBuilder->pTPBuf; + + if (pBuilder->nCols < pBuilder->pTSchema->numOfCols) { + // set non-set cols as None + for (int32_t iCol = 1; iCol < pBuilder->pTSchema->numOfCols; iCol++) { + pBuilder->pTColumn = &pBuilder->pTSchema->columns[iCol]; + if (pBuilder->pTColumn->flags & COL_VAL_SET) continue; + + { + // set None (todo) + } + + pBuilder->pTColumn->flags |= COL_VAL_SET; + } + } + } + + *ppRow = &pBuilder->row; + return 0; +} + +#if 1 // ==================== static void dataColSetNEleNull(SDataCol *pCol, int nEle); -#if 0 -static void tdMergeTwoDataCols(SDataCols *target, SDataCols *src1, int *iter1, int limit1, SDataCols *src2, int *iter2, - int limit2, int tRows, bool forceSetNull); -#endif -int tdAllocMemForCol(SDataCol *pCol, int maxPoints) { +int tdAllocMemForCol(SDataCol *pCol, int maxPoints) { int spaceNeeded = pCol->bytes * maxPoints; if (IS_VAR_DATA_TYPE(pCol->type)) { spaceNeeded += sizeof(VarDataOffsetT) * maxPoints; @@ -504,3 +730,4 @@ SKVRow tdGetKVRowFromBuilder(SKVRowBuilder *pBuilder) { return row; } +#endif \ No newline at end of file diff --git a/source/dnode/mgmt/mgmt_vnode/src/vmHandle.c b/source/dnode/mgmt/mgmt_vnode/src/vmHandle.c index c5919e06b6..6a0f4984b1 100644 --- a/source/dnode/mgmt/mgmt_vnode/src/vmHandle.c +++ b/source/dnode/mgmt/mgmt_vnode/src/vmHandle.c @@ -140,11 +140,17 @@ static void vmGenerateVnodeCfg(SCreateVnodeReq *pCreate, SVnodeCfg *pCfg) { pCfg->vgId = pCreate->vgId; tstrncpy(pCfg->dbname, pCreate->db, sizeof(pCfg->dbname)); pCfg->dbId = pCreate->dbUid; + pCfg->szPage = pCreate->pageSize * 1024; + pCfg->szCache = pCreate->pages; + pCfg->szBuf = pCreate->buffer * 1024 * 1024; pCfg->isWeak = true; + pCfg->tsdbCfg.precision = pCreate->precision; pCfg->tsdbCfg.days = 10; pCfg->tsdbCfg.keep0 = 3650; pCfg->tsdbCfg.keep1 = 3650; pCfg->tsdbCfg.keep2 = 3650; + pCfg->tsdbCfg.minRows = pCreate->minRows; + pCfg->tsdbCfg.maxRows = pCreate->maxRows; for (size_t i = 0; i < taosArrayGetSize(pCreate->pRetensions); ++i) { memcpy(&pCfg->tsdbCfg.retentions[i], taosArrayGet(pCreate->pRetensions, i), sizeof(SRetention)); } diff --git a/source/dnode/mnode/impl/src/mndDb.c b/source/dnode/mnode/impl/src/mndDb.c index 645f2ff0e7..c84cc10050 100644 --- a/source/dnode/mnode/impl/src/mndDb.c +++ b/source/dnode/mnode/impl/src/mndDb.c @@ -412,8 +412,8 @@ static void mndSetDefaultDbCfg(SDbCfg *pCfg) { if (pCfg->numOfVgroups < 0) pCfg->numOfVgroups = TSDB_DEFAULT_VN_PER_DB; if (pCfg->numOfStables < 0) pCfg->numOfStables = TSDB_DEFAULT_DB_SINGLE_STABLE; if (pCfg->buffer < 0) pCfg->buffer = TSDB_DEFAULT_BUFFER_PER_VNODE; - if (pCfg->pageSize < 0) pCfg->pageSize = TSDB_DEFAULT_PAGES_PER_VNODE; - if (pCfg->pages < 0) pCfg->pages = TSDB_MAX_PAGESIZE_PER_VNODE; + if (pCfg->pageSize < 0) pCfg->pageSize = TSDB_DEFAULT_PAGESIZE_PER_VNODE; + if (pCfg->pages < 0) pCfg->pages = TSDB_DEFAULT_PAGES_PER_VNODE; if (pCfg->daysPerFile < 0) pCfg->daysPerFile = TSDB_DEFAULT_DURATION_PER_FILE; if (pCfg->daysToKeep0 < 0) pCfg->daysToKeep0 = TSDB_DEFAULT_KEEP; if (pCfg->daysToKeep1 < 0) pCfg->daysToKeep1 = pCfg->daysToKeep0; diff --git a/source/dnode/vnode/src/tq/tq.c b/source/dnode/vnode/src/tq/tq.c index 4b9551b250..b49c148ac5 100644 --- a/source/dnode/vnode/src/tq/tq.c +++ b/source/dnode/vnode/src/tq/tq.c @@ -66,9 +66,9 @@ static void tdSRowDemo() { SRowBuilder rb = {0}; SSchema schema[DEMO_N_COLS] = { - {.type = TSDB_DATA_TYPE_TIMESTAMP, .colId = 1, .name = "ts", .bytes = 8, .flags = SCHEMA_SMA_ON}, - {.type = TSDB_DATA_TYPE_INT, .colId = 2, .name = "c1", .bytes = 4, .flags = SCHEMA_SMA_ON}, - {.type = TSDB_DATA_TYPE_INT, .colId = 3, .name = "c2", .bytes = 4, .flags = SCHEMA_SMA_ON}}; + {.type = TSDB_DATA_TYPE_TIMESTAMP, .colId = 1, .name = "ts", .bytes = 8, .flags = COL_SMA_ON}, + {.type = TSDB_DATA_TYPE_INT, .colId = 2, .name = "c1", .bytes = 4, .flags = COL_SMA_ON}, + {.type = TSDB_DATA_TYPE_INT, .colId = 3, .name = "c2", .bytes = 4, .flags = COL_SMA_ON}}; SSchema* pSchema = schema; STSchema* pTSChema = tdGetSTSChemaFromSSChema(&pSchema, numOfCols); diff --git a/source/libs/parser/src/parTranslater.c b/source/libs/parser/src/parTranslater.c index cdcb2592a7..5b984d766c 100644 --- a/source/libs/parser/src/parTranslater.c +++ b/source/libs/parser/src/parTranslater.c @@ -2138,7 +2138,7 @@ static int32_t columnDefNodeToField(SNodeList* pList, SArray** pArray) { SField field = {.type = pCol->dataType.type, .bytes = calcTypeBytes(pCol->dataType)}; strcpy(field.name, pCol->colName); if (pCol->sma) { - field.flags |= SCHEMA_SMA_ON; + field.flags |= COL_SMA_ON; } taosArrayPush(*pArray, &field); } @@ -2321,7 +2321,7 @@ static int32_t checkCreateTable(STranslateContext* pCxt, SCreateTableStmt* pStmt static void toSchema(const SColumnDefNode* pCol, col_id_t colId, SSchema* pSchema) { int8_t flags = 0; if (pCol->sma) { - flags |= SCHEMA_SMA_ON; + flags |= COL_SMA_ON; } pSchema->colId = colId; pSchema->type = pCol->dataType.type; diff --git a/source/libs/parser/test/parInitialCTest.cpp b/source/libs/parser/test/parInitialCTest.cpp index cf364aba5c..80921f59a9 100644 --- a/source/libs/parser/test/parInitialCTest.cpp +++ b/source/libs/parser/test/parInitialCTest.cpp @@ -117,7 +117,7 @@ TEST_F(ParserInitialCTest, createStable) { }; auto addFieldToCreateStbReqFunc = [&](bool col, const char* pFieldName, uint8_t type, int32_t bytes = 0, - int8_t flags = SCHEMA_SMA_ON) { + int8_t flags = COL_SMA_ON) { SField field = {0}; strcpy(field.name, pFieldName); field.type = type;