Merge pull request #12338 from taosdata/feature/data_format

fix: database options

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
 }

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)

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

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

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));
   }

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;

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);

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;

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;