feat: vnode snapshot 1

include/common/tdataformat.h

@@ -64,7 +64,7 @@ void    tTagFree(STag *pTag);
 int32_t tTagSet(STag *pTag, SSchema *pSchema, int32_t nCols, int iCol, uint8_t *pData, uint32_t nData, STag **ppTag);
 void    tTagGet(STag *pTag, int16_t cid, int8_t type, uint8_t **ppData, uint32_t *nData);
 int32_t tEncodeTag(SEncoder *pEncoder, const STag *pTag);
-int32_t tDecodeTag(SDecoder *pDecoder, const STag **ppTag);
+int32_t tDecodeTag(SDecoder *pDecoder, STag **ppTag);
 
 // STRUCT =================
 struct STColumn {

include/util/taoserror.h

@@ -313,6 +313,7 @@ int32_t* taosGetErrno();
 #define TSDB_CODE_VND_INVALID_TABLE_ACTION      TAOS_DEF_ERROR_CODE(0, 0x0519)
 #define TSDB_CODE_VND_COL_ALREADY_EXISTS        TAOS_DEF_ERROR_CODE(0, 0x051a)
 #define TSDB_CODE_VND_TABLE_COL_NOT_EXISTS      TAOS_DEF_ERROR_CODE(0, 0x051b)
+#define TSDB_CODE_VND_READ_END                  TAOS_DEF_ERROR_CODE(0, 0x051c)
 
 // tsdb
 #define TSDB_CODE_TDB_INVALID_TABLE_ID          TAOS_DEF_ERROR_CODE(0, 0x0600)

source/common/src/tdataformat.c

@@ -646,9 +646,7 @@ int32_t tEncodeTag(SEncoder *pEncoder, const STag *pTag) {
   return tEncodeBinary(pEncoder, (const uint8_t *)pTag, pTag->len);
 }
 
-int32_t tDecodeTag(SDecoder *pDecoder, const STag **ppTag) {
-  return tDecodeBinary(pDecoder, (const uint8_t **)ppTag, NULL);
-}
+int32_t tDecodeTag(SDecoder *pDecoder, STag **ppTag) { return tDecodeBinary(pDecoder, (uint8_t **)ppTag, NULL); }
 
 #if 1  // ===================================================================================================================
 static void dataColSetNEleNull(SDataCol *pCol, int nEle);
@@ -1127,7 +1125,7 @@ SKVRow tdGetKVRowFromBuilder(SKVRowBuilder *pBuilder) {
   kvRowSetNCols(row, pBuilder->nCols);
   kvRowSetLen(row, tlen);
 
-  if(pBuilder->nCols > 0){
+  if (pBuilder->nCols > 0) {
     memcpy(kvRowColIdx(row), pBuilder->pColIdx, sizeof(SColIdx) * pBuilder->nCols);
     memcpy(kvRowValues(row), pBuilder->buf, pBuilder->size);
   }

source/dnode/vnode/CMakeLists.txt

@@ -13,6 +13,7 @@ target_sources(
     "src/vnd/vnodeModule.c"
     "src/vnd/vnodeSvr.c"
     "src/vnd/vnodeSync.c"
+    "src/vnd/vnodeSnapshot.c"
 
     # meta
     "src/meta/metaOpen.c"
@@ -22,6 +23,7 @@ target_sources(
     "src/meta/metaQuery.c"
     "src/meta/metaCommit.c"
     "src/meta/metaEntry.c"
+    "src/meta/metaSnapshot.c"
 
     # sma
     "src/sma/sma.c"
@@ -44,6 +46,7 @@ target_sources(
     "src/tsdb/tsdbReadImpl.c"
     # "src/tsdb/tsdbSma.c"
     "src/tsdb/tsdbWrite.c"
+    "src/tsdb/tsdbSnapshot.c"
 
     # tq
     "src/tq/tq.c"

source/dnode/vnode/inc/vnode.h

@@ -39,9 +39,10 @@ extern "C" {
 #endif
 
 // vnode
-typedef struct SVnode    SVnode;
-typedef struct STsdbCfg  STsdbCfg;  // todo: remove
-typedef struct SVnodeCfg SVnodeCfg;
+typedef struct SVnode           SVnode;
+typedef struct STsdbCfg         STsdbCfg;  // todo: remove
+typedef struct SVnodeCfg        SVnodeCfg;
+typedef struct SVSnapshotReader SVSnapshotReader;
 
 extern const SVnodeCfg vnodeCfgDefault;
 
@@ -59,13 +60,14 @@ int32_t vnodeProcessQueryMsg(SVnode *pVnode, SRpcMsg *pMsg);
 int32_t vnodeProcessFetchMsg(SVnode *pVnode, SRpcMsg *pMsg, SQueueInfo *pInfo);
 int32_t vnodeGetLoad(SVnode *pVnode, SVnodeLoad *pLoad);
 int32_t vnodeValidateTableHash(SVnode *pVnode, char *tableFName);
-
 int32_t vnodeStart(SVnode *pVnode);
 void    vnodeStop(SVnode *pVnode);
-
 int64_t vnodeGetSyncHandle(SVnode *pVnode);
 void    vnodeGetSnapshot(SVnode *pVnode, SSnapshot *pSnapshot);
 void    vnodeGetInfo(SVnode *pVnode, const char **dbname, int32_t *vgId);
+int32_t vnodeSnapshotReaderOpen(SVnode *pVnode, SVSnapshotReader **ppReader, int64_t sver, int64_t ever);
+int32_t vnodeSnapshotReaderClose(SVSnapshotReader *pReader);
+int32_t vnodeSnapshotRead(SVSnapshotReader *pReader, void **ppData, uint32_t *nData);
 
 // meta
 typedef struct SMeta       SMeta;  // todo: remove

source/dnode/vnode/src/inc/vnodeInt.h

@@ -47,15 +47,17 @@
 extern "C" {
 #endif
 
-typedef struct SVnodeInfo   SVnodeInfo;
-typedef struct SMeta        SMeta;
-typedef struct SSma         SSma;
-typedef struct STsdb        STsdb;
-typedef struct STQ          STQ;
-typedef struct SVState      SVState;
-typedef struct SVBufPool    SVBufPool;
-typedef struct SQWorker     SQHandle;
-typedef struct STsdbKeepCfg STsdbKeepCfg;
+typedef struct SVnodeInfo          SVnodeInfo;
+typedef struct SMeta               SMeta;
+typedef struct SSma                SSma;
+typedef struct STsdb               STsdb;
+typedef struct STQ                 STQ;
+typedef struct SVState             SVState;
+typedef struct SVBufPool           SVBufPool;
+typedef struct SQWorker            SQHandle;
+typedef struct STsdbKeepCfg        STsdbKeepCfg;
+typedef struct SMetaSnapshotReader SMetaSnapshotReader;
+typedef struct STsdbSnapshotReader STsdbSnapshotReader;
 
 #define VNODE_META_DIR  "meta"
 #define VNODE_TSDB_DIR  "tsdb"
@@ -95,6 +97,9 @@ STSma*          metaGetSmaInfoByIndex(SMeta* pMeta, int64_t indexUid);
 STSmaWrapper*   metaGetSmaInfoByTable(SMeta* pMeta, tb_uid_t uid, bool deepCopy);
 SArray*         metaGetSmaIdsByTable(SMeta* pMeta, tb_uid_t uid);
 SArray*         metaGetSmaTbUids(SMeta* pMeta);
+int32_t         metaSnapshotReaderOpen(SMeta* pMeta, SMetaSnapshotReader** ppReader, int64_t sver, int64_t ever);
+int32_t         metaSnapshotReaderClose(SMetaSnapshotReader* pReader);
+int32_t         metaSnapshotRead(SMetaSnapshotReader* pReader, void** ppData, uint32_t* nData);
 
 int32_t metaCreateTSma(SMeta* pMeta, int64_t version, SSmaCfg* pCfg);
 int32_t metaDropTSma(SMeta* pMeta, int64_t indexUid);
@@ -112,6 +117,9 @@ tsdbReaderT* tsdbQueryTables(SVnode* pVnode, SQueryTableDataCond* pCond, STableG
 tsdbReaderT  tsdbQueryCacheLastT(STsdb* tsdb, SQueryTableDataCond* pCond, STableGroupInfo* groupList, uint64_t qId,
                                  void* pMemRef);
 int32_t      tsdbGetTableGroupFromIdListT(STsdb* tsdb, SArray* pTableIdList, STableGroupInfo* pGroupInfo);
+int32_t      tsdbSnapshotReaderOpen(STsdb* pTsdb, STsdbSnapshotReader** ppReader, int64_t sver, int64_t ever);
+int32_t      tsdbSnapshotReaderClose(STsdbSnapshotReader* pReader);
+int32_t      tsdbSnapshotRead(STsdbSnapshotReader* pReader, void** ppData, uint32_t* nData);
 
 // tq
 STQ*    tqOpen(const char* path, SVnode* pVnode, SWal* pWal);

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

+/*
+ * 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 "meta.h"
+
+struct SMetaSnapshotReader {
+  // TODO
+};
+
+int32_t metaSnapshotReaderOpen(SMeta* pMeta, SMetaSnapshotReader** ppReader, int64_t sver, int64_t ever) {
+  // TODO
+  return 0;
+}
+
+int32_t metaSnapshotReaderClose(SMetaSnapshotReader* pReader) {
+  // TODO
+  return 0;
+}
+
+int32_t metaSnapshotRead(SMetaSnapshotReader* pReader, void** ppData, uint32_t* nData) {
+  // TODO
+  return 0;
+}
\ No newline at end of file

source/dnode/vnode/src/tsdb/tsdbSnapshot.c

+/*
+ * 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"
+
+struct STsdbSnapshotReader {
+  // TODO
+};
+
+int32_t tsdbSnapshotReaderOpen(STsdb* pTsdb, STsdbSnapshotReader** ppReader, int64_t sver, int64_t ever) {
+  // TODO
+  return 0;
+}
+
+int32_t tsdbSnapshotReaderClose(STsdbSnapshotReader* pReader) {
+  // TODO
+  return 0;
+}
+
+int32_t tsdbSnapshotRead(STsdbSnapshotReader* pReader, void** ppData, uint32_t* nData) {
+  // TODO
+  return 0;
+}

source/dnode/vnode/src/vnd/vnodeSnapshot.c

+/*
+ * 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 "vnodeInt.h"
+
+struct SVSnapshotReader {
+  SVnode              *pVnode;
+  int64_t              sver;
+  int64_t              ever;
+  int8_t               isMetaEnd;
+  int8_t               isTsdbEnd;
+  SMetaSnapshotReader *pMetaReader;
+  STsdbSnapshotReader *pTsdbReader;
+};
+
+int32_t vnodeSnapshotReaderOpen(SVnode *pVnode, SVSnapshotReader **ppReader, int64_t sver, int64_t ever) {
+  SVSnapshotReader *pReader = NULL;
+
+  pReader = (SVSnapshotReader *)taosMemoryCalloc(1, sizeof(*pReader));
+  if (pReader == NULL) {
+    terrno = TSDB_CODE_OUT_OF_MEMORY;
+    goto _err;
+  }
+  pReader->pVnode = pVnode;
+  pReader->sver = sver;
+  pReader->ever = ever;
+  pReader->isMetaEnd = 0;
+  pReader->isTsdbEnd = 0;
+
+  if (metaSnapshotReaderOpen(pVnode->pMeta, &pReader->pMetaReader, sver, ever) < 0) {
+    taosMemoryFree(pReader);
+    goto _err;
+  }
+
+  if (tsdbSnapshotReaderOpen(pVnode->pTsdb, &pReader->pTsdbReader, sver, ever) < 0) {
+    metaSnapshotReaderClose(pReader->pMetaReader);
+    taosMemoryFree(pReader);
+    goto _err;
+  }
+
+_exit:
+  *ppReader = pReader;
+  return 0;
+
+_err:
+  *ppReader = NULL;
+  return -1;
+}
+
+int32_t vnodeSnapshotReaderClose(SVSnapshotReader *pReader) {
+  if (pReader) {
+    tsdbSnapshotReaderClose(pReader->pTsdbReader);
+    metaSnapshotReaderClose(pReader->pMetaReader);
+    taosMemoryFree(pReader);
+  }
+  return 0;
+}
+
+int32_t vnodeSnapshotRead(SVSnapshotReader *pReader, void **ppData, uint32_t *nData) {
+  int32_t code = 0;
+
+  if (!pReader->isMetaEnd) {
+    code = metaSnapshotRead(pReader->pMetaReader, ppData, nData);
+    if (code) {
+      if (code == TSDB_CODE_VND_READ_END) {
+        pReader->isMetaEnd = 1;
+      } else {
+        return code;
+      }
+    } else {
+      return code;
+    }
+  }
+
+  if (!pReader->isTsdbEnd) {
+    code = tsdbSnapshotRead(pReader->pTsdbReader, ppData, nData);
+    if (code) {
+    } else {
+      return code;
+    }
+  }
+
+  code = TSDB_CODE_VND_READ_END;
+  return code;
+}
\ No newline at end of file

source/util/src/terror.c

@@ -311,6 +311,7 @@ TAOS_DEFINE_ERROR(TSDB_CODE_VND_TABLE_NOT_EXIST,          "Table does not exists
 TAOS_DEFINE_ERROR(TSDB_CODE_VND_INVALID_TABLE_ACTION,     "Invalid table action")
 TAOS_DEFINE_ERROR(TSDB_CODE_VND_COL_ALREADY_EXISTS,       "Table column already exists")
 TAOS_DEFINE_ERROR(TSDB_CODE_VND_TABLE_COL_NOT_EXISTS,     "Table column not exists")
+TAOS_DEFINE_ERROR(TSDB_CODE_VND_READ_END,                 "Read end")
 
 
 // tsdb