more

source/dnode/vnode/inc/vnode.h

@@ -43,6 +43,7 @@ typedef struct SVnode           SVnode;
 typedef struct STsdbCfg         STsdbCfg;  // todo: remove
 typedef struct SVnodeCfg        SVnodeCfg;
 typedef struct SVSnapshotReader SVSnapshotReader;
+typedef struct SVSnapshotWriter SVSnapshotWriter;
 
 extern const SVnodeCfg vnodeCfgDefault;
 
@@ -66,9 +67,6 @@ 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, const void **ppData, uint32_t *nData);
 int32_t vnodeProcessCreateTSma(SVnode *pVnode, void *pCont, uint32_t contLen);
 int32_t vnodeGetAllTableList(SVnode *pVnode, uint64_t uid, SArray *list);
 int32_t vnodeGetCtbIdList(SVnode *pVnode, int64_t suid, SArray *list);
@@ -159,7 +157,14 @@ int32_t tqRetrieveDataBlock(SSDataBlock *pBlock, SStreamReader *pHandle);
 // sma
 int32_t smaGetTSmaDays(SVnodeCfg *pCfg, void *pCont, uint32_t contLen, int32_t *days);
 
-// need to reposition
+// SVSnapshotReader
+int32_t vnodeSnapshotReaderOpen(SVnode *pVnode, SVSnapshotReader **ppReader, int64_t sver, int64_t ever);
+int32_t vnodeSnapshotReaderClose(SVSnapshotReader *pReader);
+int32_t vnodeSnapshotRead(SVSnapshotReader *pReader, const void **ppData, uint32_t *nData);
+// SVSnapshotWriter;
+int32_t vnodeSnapshotWriterOpen(SVnode *pVnode, int64_t sver, int64_t ever, SVSnapshotWriter **ppWriter);
+int32_t vnodeSnapshotWrite(SVSnapshotWriter *pWriter, uint8_t *pData, uint32_t nData);
+int32_t vnodeSnapshotWriterClose(SVSnapshotWriter *pWriter, int8_t rollback);
 
 // structs
 struct STsdbCfg {

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

@@ -16,21 +16,60 @@
 #include "tsdb.h"
 
 struct STsdbSnapshotReader {
-  STsdb* pTsdb;
+  STsdb*  pTsdb;
-  // TODO
+  int64_t sver;
+  int64_t ever;
+  // for data file
+  SDataFReader* pDataFReader;
+  // for del file
+  SDelFReader* pDelFReader;
 };
 
-int32_t tsdbSnapshotReaderOpen(STsdb* pTsdb, STsdbSnapshotReader** ppReader, int64_t sver, int64_t ever) {
+typedef struct STsdbSnapshotWriter {
-  // TODO
+  STsdb*  pTsdb;
-  return 0;
+  int64_t sver;
-}
+  int64_t ever;
+  // for data file
+  SDataFWriter* pDataFWriter;
+  // for del file
+  SDelFWriter* pDelFWriter;
+} STsdbSnapshotWriter;
 
-int32_t tsdbSnapshotReaderClose(STsdbSnapshotReader* pReader) {
+int32_t tsdbSnapshotReaderOpen(STsdb* pTsdb, STsdbSnapshotReader** ppReader, int64_t sver, int64_t ever) {
-  // TODO
+  int32_t              code = 0;
-  return 0;
+  STsdbSnapshotReader* pReader = NULL;
+
+  // alloc
+  pReader = (STsdbSnapshotReader*)taosMemoryCalloc(1, sizeof(*pReader));
+  if (pReader == NULL) {
+    code = TSDB_CODE_OUT_OF_MEMORY;
+    goto _err;
+  }
+  pReader->pTsdb = pTsdb;
+  pReader->sver = sver;
+  pReader->ever = ever;
+
+  *ppReader = pReader;
+  return code;
+
+_err:
+  tsdbError("vgId:%d snapshot reader open failed since %s", TD_VID(pTsdb->pVnode), tstrerror(code));
+  *ppReader = NULL;
+  return code;
 }
 
 int32_t tsdbSnapshotRead(STsdbSnapshotReader* pReader, void** ppData, uint32_t* nData) {
+  int32_t code = 0;
   // TODO
-  return 0;
+  return code;
+
+_err:
+  tsdbError("vgId:%d snapshot read failed since %s", TD_VID(pReader->pTsdb->pVnode), tstrerror(code));
+  return code;
+}
+
+int32_t tsdbSnapshotReaderClose(STsdbSnapshotReader* pReader) {
+  int32_t code = 0;
+  taosMemoryFree(pReader);
+  return code;
 }

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

@@ -27,6 +27,13 @@ struct SVSnapshotReader {
   int32_t              nData;
 };
 
+struct SVSnapshotWriter {
+  SVnode *pVnode;
+  int64_t sver;
+  int64_t ever;
+};
+
+// SVSnapshotReader ========================================================
 int32_t vnodeSnapshotReaderOpen(SVnode *pVnode, SVSnapshotReader **ppReader, int64_t sver, int64_t ever) {
   SVSnapshotReader *pReader = NULL;
 
@@ -106,4 +113,44 @@ int32_t vnodeSnapshotRead(SVSnapshotReader *pReader, const void **ppData, uint32
 
   code = TSDB_CODE_VND_READ_END;
   return code;
+}
+
+// SVSnapshotWriter ========================================================
+int32_t vnodeSnapshotWriterOpen(SVnode *pVnode, int64_t sver, int64_t ever, SVSnapshotWriter **ppWriter) {
+  int32_t           code = 0;
+  SVSnapshotWriter *pWriter = NULL;
+
+  // alloc
+  pWriter = (SVSnapshotWriter *)taosMemoryCalloc(1, sizeof(*pWriter));
+  if (pWriter == NULL) {
+    code = TSDB_CODE_OUT_OF_MEMORY;
+    goto _err;
+  }
+  pWriter->pVnode = pVnode;
+  pWriter->sver = sver;
+  pWriter->ever = ever;
+
+  return code;
+
+_err:
+  return code;
+}
+
+int32_t vnodeSnapshotWrite(SVSnapshotWriter *pWriter, uint8_t *pData, uint32_t nData) {
+  int32_t code = 0;
+  // TODO
+  return code;
+}
+
+int32_t vnodeSnapshotWriterClose(SVSnapshotWriter *pWriter, int8_t rollback) {
+  int32_t code = 0;
+
+  if (!rollback) {
+    // apply the change
+  } else {
+    // rollback the change
+  }
+
+  taosMemoryFree(pWriter);
+  return code;
 }