Merge pull request #9493 from taosdata/feature/vnode

Feature/vnode

.devcontainer/devcontainer.json

@@ -18,7 +18,8 @@
 		"ms-vscode.cpptools",
 		"ms-vscode.cmake-tools",
 		"austin.code-gnu-global",
-		"visualstudioexptteam.vscodeintel"
+		"visualstudioexptteam.vscodeintel",
+		"eamodio.gitlens"
 	],
 
 	// Use 'forwardPorts' to make a list of ports inside the container available locally.

include/common/tmsg.h

@@ -25,7 +25,9 @@ extern "C" {
 #include "taoserror.h"
 #include "tcoding.h"
 #include "tdataformat.h"
+#include "tlist.h"
 
+/* ------------------------ MESSAGE DEFINITIONS ------------------------ */
 #define TD_MSG_NUMBER_
 #undef TD_MSG_DICT_
 #undef TD_MSG_INFO_
@@ -54,6 +56,47 @@ extern int   tMsgDict[];
 
 typedef uint16_t tmsg_t;
 
+/* ------------------------ ENCODE/DECODE FUNCTIONS AND MACROS ------------------------ */
+struct SMEListNode {
+  TD_SLIST_NODE(SMEListNode);
+  SEncoder coder;
+};
+
+typedef struct SMsgEncoder {
+  SEncoder coder;
+  TD_SLIST(SMEListNode) eStack;  // encode stack
+} SMsgEncoder;
+
+struct SMDFreeListNode {
+  TD_SLIST_NODE(SMDFreeListNode);
+  char payload[];
+};
+
+struct SMDListNode {
+  TD_SLIST_NODE(SMDListNode);
+  SDecoder coder;
+};
+
+typedef struct SMsgDecoder {
+  SDecoder coder;
+  TD_SLIST(SMDListNode) dStack;
+  TD_SLIST(SMDFreeListNode) freeList;
+} SMsgDecoder;
+
+#define TMSG_MALLOC(SIZE, DECODER)                                         \
+  ({                                                                       \
+    void* ptr = malloc((SIZE) + sizeof(struct SMDFreeListNode));           \
+    if (ptr) {                                                             \
+      TD_SLIST_PUSH(&((DECODER)->freeList), (struct SMDFreeListNode*)ptr); \
+      ptr = POINTER_SHIFT(ptr, sizeof(struct SMDFreeListNode*));           \
+    }                                                                      \
+    ptr;                                                                   \
+  })
+
+void tmsgInitMsgDecoder(SMsgDecoder* pMD, td_endian_t endian, uint8_t* data, int64_t size);
+void tmsgClearMsgDecoder(SMsgDecoder* pMD);
+
+/* ------------------------ OTHER DEFINITIONS ------------------------ */
 // IE type
 #define TSDB_IE_TYPE_SEC 1
 #define TSDB_IE_TYPE_META 2
@@ -1228,100 +1271,11 @@ typedef struct SVCreateTbReq {
   };
 } SVCreateTbReq;
 
-static FORCE_INLINE int tSerializeSVCreateTbReq(void** buf, const SVCreateTbReq* pReq) {
-  int tlen = 0;
-
-  tlen += taosEncodeFixedU64(buf, pReq->ver);
-  tlen += taosEncodeString(buf, pReq->name);
-  tlen += taosEncodeFixedU32(buf, pReq->ttl);
-  tlen += taosEncodeFixedU32(buf, pReq->keep);
-  tlen += taosEncodeFixedU8(buf, pReq->type);
-
-  switch (pReq->type) {
-    case TD_SUPER_TABLE:
-      tlen += taosEncodeFixedU64(buf, pReq->stbCfg.suid);
-      tlen += taosEncodeFixedU32(buf, pReq->stbCfg.nCols);
-      for (uint32_t i = 0; i < pReq->stbCfg.nCols; i++) {
-        tlen += taosEncodeFixedI8(buf, pReq->stbCfg.pSchema[i].type);
-        tlen += taosEncodeFixedI32(buf, pReq->stbCfg.pSchema[i].colId);
-        tlen += taosEncodeFixedI32(buf, pReq->stbCfg.pSchema[i].bytes);
-        tlen += taosEncodeString(buf, pReq->stbCfg.pSchema[i].name);
-      }
-      tlen += taosEncodeFixedU32(buf, pReq->stbCfg.nTagCols);
-      for (uint32_t i = 0; i < pReq->stbCfg.nTagCols; i++) {
-        tlen += taosEncodeFixedI8(buf, pReq->stbCfg.pTagSchema[i].type);
-        tlen += taosEncodeFixedI32(buf, pReq->stbCfg.pTagSchema[i].colId);
-        tlen += taosEncodeFixedI32(buf, pReq->stbCfg.pTagSchema[i].bytes);
-        tlen += taosEncodeString(buf, pReq->stbCfg.pTagSchema[i].name);
-      }
-      break;
-    case TD_CHILD_TABLE:
-      tlen += taosEncodeFixedU64(buf, pReq->ctbCfg.suid);
-      tlen += tdEncodeKVRow(buf, pReq->ctbCfg.pTag);
-      break;
-    case TD_NORMAL_TABLE:
-      tlen += taosEncodeFixedU32(buf, pReq->ntbCfg.nCols);
-      for (uint32_t i = 0; i < pReq->ntbCfg.nCols; i++) {
-        tlen += taosEncodeFixedI8(buf, pReq->ntbCfg.pSchema[i].type);
-        tlen += taosEncodeFixedI32(buf, pReq->ntbCfg.pSchema[i].colId);
-        tlen += taosEncodeFixedI32(buf, pReq->ntbCfg.pSchema[i].bytes);
-        tlen += taosEncodeString(buf, pReq->ntbCfg.pSchema[i].name);
-      }
-      break;
-    default:
-      ASSERT(0);
-  }
-
-  return tlen;
-}
-
-static FORCE_INLINE void* tDeserializeSVCreateTbReq(void* buf, SVCreateTbReq* pReq) {
-  buf = taosDecodeFixedU64(buf, &(pReq->ver));
-  buf = taosDecodeString(buf, &(pReq->name));
-  buf = taosDecodeFixedU32(buf, &(pReq->ttl));
-  buf = taosDecodeFixedU32(buf, &(pReq->keep));
-  buf = taosDecodeFixedU8(buf, &(pReq->type));
-
-  switch (pReq->type) {
-    case TD_SUPER_TABLE:
-      buf = taosDecodeFixedU64(buf, &(pReq->stbCfg.suid));
-      buf = taosDecodeFixedU32(buf, &(pReq->stbCfg.nCols));
-      pReq->stbCfg.pSchema = (SSchema*)malloc(pReq->stbCfg.nCols * sizeof(SSchema));
-      for (uint32_t i = 0; i < pReq->stbCfg.nCols; i++) {
-        buf = taosDecodeFixedI8(buf, &(pReq->stbCfg.pSchema[i].type));
-        buf = taosDecodeFixedI32(buf, &(pReq->stbCfg.pSchema[i].colId));
-        buf = taosDecodeFixedI32(buf, &(pReq->stbCfg.pSchema[i].bytes));
-        buf = taosDecodeStringTo(buf, pReq->stbCfg.pSchema[i].name);
-      }
-      buf = taosDecodeFixedU32(buf, &pReq->stbCfg.nTagCols);
-      pReq->stbCfg.pTagSchema = (SSchema*)malloc(pReq->stbCfg.nTagCols * sizeof(SSchema));
-      for (uint32_t i = 0; i < pReq->stbCfg.nTagCols; i++) {
-        buf = taosDecodeFixedI8(buf, &(pReq->stbCfg.pTagSchema[i].type));
-        buf = taosDecodeFixedI32(buf, &pReq->stbCfg.pTagSchema[i].colId);
-        buf = taosDecodeFixedI32(buf, &pReq->stbCfg.pTagSchema[i].bytes);
-        buf = taosDecodeStringTo(buf, pReq->stbCfg.pTagSchema[i].name);
-      }
-      break;
-    case TD_CHILD_TABLE:
-      buf = taosDecodeFixedU64(buf, &pReq->ctbCfg.suid);
-      buf = tdDecodeKVRow(buf, &pReq->ctbCfg.pTag);
-      break;
-    case TD_NORMAL_TABLE:
-      buf = taosDecodeFixedU32(buf, &pReq->ntbCfg.nCols);
-      pReq->ntbCfg.pSchema = (SSchema*)malloc(pReq->ntbCfg.nCols * sizeof(SSchema));
-      for (uint32_t i = 0; i < pReq->ntbCfg.nCols; i++) {
-        buf = taosDecodeFixedI8(buf, &pReq->ntbCfg.pSchema[i].type);
-        buf = taosDecodeFixedI32(buf, &pReq->ntbCfg.pSchema[i].colId);
-        buf = taosDecodeFixedI32(buf, &pReq->ntbCfg.pSchema[i].bytes);
-        buf = taosDecodeStringTo(buf, pReq->ntbCfg.pSchema[i].name);
-      }
-      break;
-    default:
-      ASSERT(0);
-  }
+int   tmsgSVCreateTbReqEncode(SMsgEncoder* pCoder, SVCreateTbReq* pReq);
+int   tmsgSVCreateTbReqDecode(SMsgDecoder* pCoder, SVCreateTbReq* pReq);
+int   tSerializeSVCreateTbReq(void** buf, const SVCreateTbReq* pReq);
+void* tDeserializeSVCreateTbReq(void* buf, SVCreateTbReq* pReq);
 
-  return buf;
-}
 typedef struct SVCreateTbRsp {
 } SVCreateTbRsp;
 

include/util/encode.h

@@ -26,8 +26,8 @@ extern "C" {
 typedef struct {
   td_endian_t endian;
   uint8_t*    data;
-  int64_t     size;
-  int64_t     pos;
+  int32_t     size;
+  int32_t     pos;
 } SEncoder, SDecoder;
 
 #define tPut(TYPE, BUF, VAL) ((TYPE*)(BUF))[0] = (VAL)
@@ -62,7 +62,7 @@ typedef struct {
 #define TD_CHECK_CODER_CAPACITY_FAILED(CODER, EXPSIZE) (((CODER)->size - (CODER)->pos) < (EXPSIZE))
 
 /* ------------------------ FOR ENCODER ------------------------ */
-static FORCE_INLINE void tInitEncoder(SEncoder* pEncoder, td_endian_t endian, uint8_t* data, int64_t size) {
+static FORCE_INLINE void tInitEncoder(SEncoder* pEncoder, td_endian_t endian, uint8_t* data, int32_t size) {
   pEncoder->endian = endian;
   pEncoder->data = data;
   pEncoder->size = (data) ? size : 0;
@@ -266,7 +266,7 @@ static FORCE_INLINE int tEncodeCStr(SEncoder* pEncoder, const char* val) {
 }
 
 /* ------------------------ FOR DECODER ------------------------ */
-static FORCE_INLINE void tInitDecoder(SDecoder* pDecoder, td_endian_t endian, uint8_t* data, int64_t size) {
+static FORCE_INLINE void tInitDecoder(SDecoder* pDecoder, td_endian_t endian, uint8_t* data, int32_t size) {
   ASSERT(!TD_IS_NULL(data));
   pDecoder->endian = endian;
   pDecoder->data = data;

source/common/src/tmsg.c

@@ -25,4 +25,230 @@
 #undef TD_MSG_INFO_
 #define TD_MSG_DICT_
 #undef TD_MSG_SEG_CODE_
-#include "tmsgdef.h"
\ No newline at end of file
+#include "tmsgdef.h"
+
+static int  tmsgStartEncode(SMsgEncoder *pME);
+static void tmsgEndEncode(SMsgEncoder *pME);
+static int  tmsgStartDecode(SMsgDecoder *pMD);
+static void tmsgEndDecode(SMsgDecoder *pMD);
+
+/* ------------------------ ENCODE/DECODE FUNCTIONS ------------------------ */
+void tmsgInitMsgEncoder(SMsgEncoder *pME, td_endian_t endian, uint8_t *data, int64_t size) {
+  tInitEncoder(&(pME->coder), endian, data, size);
+  TD_SLIST_INIT(&(pME->eStack));
+}
+
+void tmsgClearMsgEncoder(SMsgEncoder *pME) {
+  struct SMEListNode *pNode;
+  for (;;) {
+    pNode = TD_SLIST_HEAD(&(pME->eStack));
+    if (TD_IS_NULL(pNode)) break;
+    TD_SLIST_POP(&(pME->eStack));
+    free(pNode);
+  }
+}
+
+void tmsgInitMsgDecoder(SMsgDecoder *pMD, td_endian_t endian, uint8_t *data, int64_t size) {
+  tInitDecoder(&pMD->coder, endian, data, size);
+  TD_SLIST_INIT(&(pMD->dStack));
+  TD_SLIST_INIT(&(pMD->freeList));
+}
+
+void tmsgClearMsgDecoder(SMsgDecoder *pMD) {
+  {
+    struct SMDFreeListNode *pNode;
+    for (;;) {
+      pNode = TD_SLIST_HEAD(&(pMD->freeList));
+      if (TD_IS_NULL(pNode)) break;
+      TD_SLIST_POP(&(pMD->freeList));
+      free(pNode);
+    }
+  }
+  {
+    struct SMDListNode *pNode;
+    for (;;) {
+      pNode = TD_SLIST_HEAD(&(pMD->dStack));
+      if (TD_IS_NULL(pNode)) break;
+      TD_SLIST_POP(&(pMD->dStack));
+      free(pNode);
+    }
+  }
+}
+
+/* ------------------------ MESSAGE ENCODE/DECODE ------------------------ */
+int tmsgSVCreateTbReqEncode(SMsgEncoder *pCoder, SVCreateTbReq *pReq) {
+  tmsgStartEncode(pCoder);
+  // TODO
+
+  tmsgEndEncode(pCoder);
+  return 0;
+}
+
+int tmsgSVCreateTbReqDecode(SMsgDecoder *pCoder, SVCreateTbReq *pReq) {
+  tmsgStartDecode(pCoder);
+
+  // TODO: decode
+
+  // Decode is not end
+  if (pCoder->coder.pos != pCoder->coder.size) {
+    // Continue decode
+  }
+
+  tmsgEndDecode(pCoder);
+  return 0;
+}
+
+int tSerializeSVCreateTbReq(void **buf, const SVCreateTbReq *pReq) {
+  int tlen = 0;
+
+  tlen += taosEncodeFixedU64(buf, pReq->ver);
+  tlen += taosEncodeString(buf, pReq->name);
+  tlen += taosEncodeFixedU32(buf, pReq->ttl);
+  tlen += taosEncodeFixedU32(buf, pReq->keep);
+  tlen += taosEncodeFixedU8(buf, pReq->type);
+
+  switch (pReq->type) {
+    case TD_SUPER_TABLE:
+      tlen += taosEncodeFixedU64(buf, pReq->stbCfg.suid);
+      tlen += taosEncodeFixedU32(buf, pReq->stbCfg.nCols);
+      for (uint32_t i = 0; i < pReq->stbCfg.nCols; i++) {
+        tlen += taosEncodeFixedI8(buf, pReq->stbCfg.pSchema[i].type);
+        tlen += taosEncodeFixedI32(buf, pReq->stbCfg.pSchema[i].colId);
+        tlen += taosEncodeFixedI32(buf, pReq->stbCfg.pSchema[i].bytes);
+        tlen += taosEncodeString(buf, pReq->stbCfg.pSchema[i].name);
+      }
+      tlen += taosEncodeFixedU32(buf, pReq->stbCfg.nTagCols);
+      for (uint32_t i = 0; i < pReq->stbCfg.nTagCols; i++) {
+        tlen += taosEncodeFixedI8(buf, pReq->stbCfg.pTagSchema[i].type);
+        tlen += taosEncodeFixedI32(buf, pReq->stbCfg.pTagSchema[i].colId);
+        tlen += taosEncodeFixedI32(buf, pReq->stbCfg.pTagSchema[i].bytes);
+        tlen += taosEncodeString(buf, pReq->stbCfg.pTagSchema[i].name);
+      }
+      break;
+    case TD_CHILD_TABLE:
+      tlen += taosEncodeFixedU64(buf, pReq->ctbCfg.suid);
+      tlen += tdEncodeKVRow(buf, pReq->ctbCfg.pTag);
+      break;
+    case TD_NORMAL_TABLE:
+      tlen += taosEncodeFixedU32(buf, pReq->ntbCfg.nCols);
+      for (uint32_t i = 0; i < pReq->ntbCfg.nCols; i++) {
+        tlen += taosEncodeFixedI8(buf, pReq->ntbCfg.pSchema[i].type);
+        tlen += taosEncodeFixedI32(buf, pReq->ntbCfg.pSchema[i].colId);
+        tlen += taosEncodeFixedI32(buf, pReq->ntbCfg.pSchema[i].bytes);
+        tlen += taosEncodeString(buf, pReq->ntbCfg.pSchema[i].name);
+      }
+      break;
+    default:
+      ASSERT(0);
+  }
+
+  return tlen;
+}
+
+void *tDeserializeSVCreateTbReq(void *buf, SVCreateTbReq *pReq) {
+  buf = taosDecodeFixedU64(buf, &(pReq->ver));
+  buf = taosDecodeString(buf, &(pReq->name));
+  buf = taosDecodeFixedU32(buf, &(pReq->ttl));
+  buf = taosDecodeFixedU32(buf, &(pReq->keep));
+  buf = taosDecodeFixedU8(buf, &(pReq->type));
+
+  switch (pReq->type) {
+    case TD_SUPER_TABLE:
+      buf = taosDecodeFixedU64(buf, &(pReq->stbCfg.suid));
+      buf = taosDecodeFixedU32(buf, &(pReq->stbCfg.nCols));
+      pReq->stbCfg.pSchema = (SSchema *)malloc(pReq->stbCfg.nCols * sizeof(SSchema));
+      for (uint32_t i = 0; i < pReq->stbCfg.nCols; i++) {
+        buf = taosDecodeFixedI8(buf, &(pReq->stbCfg.pSchema[i].type));
+        buf = taosDecodeFixedI32(buf, &(pReq->stbCfg.pSchema[i].colId));
+        buf = taosDecodeFixedI32(buf, &(pReq->stbCfg.pSchema[i].bytes));
+        buf = taosDecodeStringTo(buf, pReq->stbCfg.pSchema[i].name);
+      }
+      buf = taosDecodeFixedU32(buf, &pReq->stbCfg.nTagCols);
+      pReq->stbCfg.pTagSchema = (SSchema *)malloc(pReq->stbCfg.nTagCols * sizeof(SSchema));
+      for (uint32_t i = 0; i < pReq->stbCfg.nTagCols; i++) {
+        buf = taosDecodeFixedI8(buf, &(pReq->stbCfg.pTagSchema[i].type));
+        buf = taosDecodeFixedI32(buf, &pReq->stbCfg.pTagSchema[i].colId);
+        buf = taosDecodeFixedI32(buf, &pReq->stbCfg.pTagSchema[i].bytes);
+        buf = taosDecodeStringTo(buf, pReq->stbCfg.pTagSchema[i].name);
+      }
+      break;
+    case TD_CHILD_TABLE:
+      buf = taosDecodeFixedU64(buf, &pReq->ctbCfg.suid);
+      buf = tdDecodeKVRow(buf, &pReq->ctbCfg.pTag);
+      break;
+    case TD_NORMAL_TABLE:
+      buf = taosDecodeFixedU32(buf, &pReq->ntbCfg.nCols);
+      pReq->ntbCfg.pSchema = (SSchema *)malloc(pReq->ntbCfg.nCols * sizeof(SSchema));
+      for (uint32_t i = 0; i < pReq->ntbCfg.nCols; i++) {
+        buf = taosDecodeFixedI8(buf, &pReq->ntbCfg.pSchema[i].type);
+        buf = taosDecodeFixedI32(buf, &pReq->ntbCfg.pSchema[i].colId);
+        buf = taosDecodeFixedI32(buf, &pReq->ntbCfg.pSchema[i].bytes);
+        buf = taosDecodeStringTo(buf, pReq->ntbCfg.pSchema[i].name);
+      }
+      break;
+    default:
+      ASSERT(0);
+  }
+
+  return buf;
+}
+
+/* ------------------------ STATIC METHODS ------------------------ */
+static int tmsgStartEncode(SMsgEncoder *pME) {
+  struct SMEListNode *pNode = (struct SMEListNode *)malloc(sizeof(*pNode));
+  if (TD_IS_NULL(pNode)) return -1;
+
+  pNode->coder = pME->coder;
+  TD_SLIST_PUSH(&(pME->eStack), pNode);
+  TD_CODER_MOVE_POS(&(pME->coder), sizeof(int32_t));
+
+  return 0;
+}
+
+static void tmsgEndEncode(SMsgEncoder *pME) {
+  int32_t             size;
+  struct SMEListNode *pNode;
+
+  pNode = TD_SLIST_HEAD(&(pME->eStack));
+  ASSERT(pNode);
+  TD_SLIST_POP(&(pME->eStack));
+
+  size = pME->coder.pos - pNode->coder.pos;
+  tEncodeI32(&(pNode->coder), size);
+
+  free(pNode);
+}
+
+static int tmsgStartDecode(SMsgDecoder *pMD) {
+  struct SMDListNode *pNode;
+  int32_t             size;
+
+  pNode = (struct SMDListNode *)malloc(sizeof(*pNode));
+  if (pNode == NULL) return -1;
+
+  tDecodeI32(&(pMD->coder), &size);
+
+  pNode->coder = pMD->coder;
+  TD_SLIST_PUSH(&(pMD->dStack), pNode);
+
+  pMD->coder.pos = 0;
+  pMD->coder.size = size - sizeof(int32_t);
+  pMD->coder.data = TD_CODER_CURRENT(&(pNode->coder));
+
+  return 0;
+}
+
+static void tmsgEndDecode(SMsgDecoder *pMD) {
+  ASSERT(pMD->coder.pos == pMD->coder.size);
+  struct SMDListNode *pNode;
+
+  pNode = TD_SLIST_HEAD(&(pMD->dStack));
+  ASSERT(pNode);
+  TD_SLIST_POP(&(pMD->dStack));
+
+  pNode->coder.pos += pMD->coder.size;
+
+  pMD->coder = pNode->coder;
+
+  free(pNode);
+}
\ No newline at end of file

source/common/test/CMakeLists.txt

@@ -18,11 +18,11 @@ TARGET_INCLUDE_DIRECTORIES(
 )
 
 # tmsg test
-add_executable(tmsgTest "")
-target_sources(tmsgTest 
-        PRIVATE
-        "tmsgTest.cpp"
-        "../src/tmsg.c"
-)
-target_include_directories(tmsgTest PUBLIC "${CMAKE_SOURCE_DIR}/include/common/")
-target_link_libraries(tmsgTest PUBLIC os util gtest gtest_main)
\ No newline at end of file
+# add_executable(tmsgTest "")
+# target_sources(tmsgTest 
+#         PRIVATE
+#         "tmsgTest.cpp"
+#         "../src/tmsg.c"
+# )
+# target_include_directories(tmsgTest PUBLIC "${CMAKE_SOURCE_DIR}/include/common/")
+# target_link_libraries(tmsgTest PUBLIC os util gtest gtest_main)
\ No newline at end of file

source/dnode/vnode/meta/src/metaBDBImpl.c

@@ -444,6 +444,7 @@ int metaGetTableInfo(SMeta *pMeta, char *tbname, STableMetaMsg **ppMsg) {
   void *         pBuf;
   int            tlen;
   STableMetaMsg *pMsg;
+  SSchema *      pSchema;
 
   key.data = tbname;
   key.size = strlen(tbname) + 1;
@@ -487,10 +488,9 @@ int metaGetTableInfo(SMeta *pMeta, char *tbname, STableMetaMsg **ppMsg) {
       pMsg->tversion = 0;
       pMsg->suid = tbCfg.stbCfg.suid;
       pMsg->tuid = tbCfg.stbCfg.suid;
-      for (size_t i = 0; i < tbCfg.stbCfg.nTagCols; i++) {
-
-      }
-
+      memcpy(pMsg->pSchema, tbCfg.stbCfg.pSchema, sizeof(SSchema) * tbCfg.stbCfg.nCols);
+      memcpy(POINTER_SHIFT(pMsg->pSchema, sizeof(SSchema) * tbCfg.stbCfg.nCols), tbCfg.stbCfg.pTagSchema,
+              sizeof(SSchema) * tbCfg.stbCfg.nTagCols);
       break;
     case META_CHILD_TABLE:
       ASSERT(0);