add wal interface

src/common/src/dataformat.c

@@ -346,7 +346,6 @@ void tdResetDataCols(SDataCols *pCols) {
 }
 
 void tdAppendDataRowToDataCol(SDataRow row, SDataCols *pCols) {
-  TSKEY key = dataRowKey(row);
   for (int i = 0; i < pCols->numOfCols; i++) {
     SDataCol *pCol = pCols->cols + i;
     memcpy((void *)((char *)(pCol->pData) + pCol->len), dataRowAt(row, pCol->offset), pCol->bytes);
@@ -384,5 +383,5 @@ static int tdFLenFromSchema(STSchema *pSchema) {
 }
 
 int tdMergeDataCols(SDataCols *target, SDataCols *source) {
-
+  return 0;
 }

src/vnode/main/src/vnodeMain.c

@@ -32,6 +32,7 @@
 static void *tsDnodeVnodesHash;
 static void  vnodeCleanUp(SVnodeObj *pVnode);
 static void  vnodeBuildVloadMsg(char *pNode, void * param);
+static int   vnodeWALCallback(void *arg);
 
 static int   tsOpennedVnodes;
 static pthread_once_t  vnodeModuleInit = PTHREAD_ONCE_INIT;
@@ -120,23 +121,28 @@ int32_t vnodeOpen(int32_t vnode, char *rootDir) {
   vnodeObj.version  = 0;  
   SVnodeObj *pVnode = (SVnodeObj *)taosAddIntHash(tsDnodeVnodesHash, vnodeObj.vgId, (char *)(&vnodeObj));
 
+  pVnode->wqueue = dnodeAllocateWqueue(pVnode);
+  pVnode->rqueue = dnodeAllocateRqueue(pVnode);
+
+  sprintf(temp, "%s/wal", rootDir);
+  pVnode->wal      = walOpen(temp, 3, tsCommitLog);
+  pVnode->sync     = NULL;
+  pVnode->events   = NULL;
+  pVnode->cq       = NULL;
+
+  STsdbAppH appH = {0};
+  appH.appH = (void *)pVnode;
+  appH.walCallBack = vnodeWALCallback;
+
   sprintf(temp, "%s/tsdb", rootDir);
-  void *pTsdb = tsdbOpenRepo(temp);
+  void *pTsdb = tsdbOpenRepo(temp, &appH);
   if (pTsdb == NULL) {
     dError("pVnode:%p vgId:%d, failed to open tsdb at %s(%s)", pVnode, pVnode->vgId, temp, tstrerror(terrno));
     taosDeleteIntHash(tsDnodeVnodesHash, pVnode->vgId);
     return terrno;
   }
 
-  pVnode->wqueue = dnodeAllocateWqueue(pVnode);
-  pVnode->rqueue = dnodeAllocateRqueue(pVnode);
-
-  sprintf(temp, "%s/wal", rootDir);
-  pVnode->wal      = walOpen(temp, 3, tsCommitLog);
-  pVnode->tsdb     = pTsdb;
-  pVnode->sync     = NULL;
-  pVnode->events   = NULL;
-  pVnode->cq       = NULL;
+  pVnode->tsdb = pTsdb;
 
   walRestore(pVnode->wal, pVnode, vnodeWriteToQueue);
 
@@ -249,3 +255,8 @@ static void vnodeCleanUp(SVnodeObj *pVnode) {
 
   vnodeRelease(pVnode);
 }
+
+static int vnodeWALCallback(void *arg) {
+  SVnodeObj *pVnode = arg;
+  return walRenew(pVnode->wal);
+}

src/vnode/tsdb/inc/tsdb.h

@@ -34,6 +34,15 @@ extern "C" {
 
 #define TSDB_INVALID_SUPER_TABLE_ID -1
 
+// --------- TSDB APPLICATION HANDLE DEFINITION
+typedef struct {
+  // WAL handle
+  void *appH;
+  int (*walCallBack)(void *);
+  int (*eventCallBack)(void *);
+  int (*cqueryCallBack)(void *);
+} STsdbAppH;
+
 // --------- TSDB REPOSITORY CONFIGURATION DEFINITION
 typedef struct {
   int8_t  precision;
@@ -55,7 +64,7 @@ typedef void tsdb_repo_t;  // use void to hide implementation details from outsi
 
 int            tsdbCreateRepo(char *rootDir, STsdbCfg *pCfg, void *limiter);
 int32_t        tsdbDropRepo(tsdb_repo_t *repo);
-tsdb_repo_t *  tsdbOpenRepo(char *tsdbDir);
+tsdb_repo_t *  tsdbOpenRepo(char *tsdbDir, STsdbAppH *pAppH);
 int32_t        tsdbCloseRepo(tsdb_repo_t *repo);
 int32_t        tsdbConfigRepo(tsdb_repo_t *repo, STsdbCfg *pCfg);
 int32_t        tsdbTriggerCommit(tsdb_repo_t *repo);

src/vnode/tsdb/inc/tsdbMain.h

@@ -322,6 +322,8 @@ typedef struct _tsdb_repo {
   // TSDB configuration
   STsdbCfg config;
 
+  STsdbAppH appH;
+
   // The meter meta handle of this TSDB repository
   STsdbMeta *tsdbMeta;
 

src/vnode/tsdb/src/tsdbMain.c

@@ -177,7 +177,7 @@ int32_t tsdbDropRepo(tsdb_repo_t *repo) {
  *
  * @return a TSDB repository handle on success, NULL for failure and the error number is set
  */
-tsdb_repo_t *tsdbOpenRepo(char *tsdbDir) {
+tsdb_repo_t *tsdbOpenRepo(char *tsdbDir, STsdbAppH *pAppH) {
   char dataDir[128] = "\0";
   if (access(tsdbDir, F_OK | W_OK | R_OK) < 0) {
     return NULL;
@@ -191,6 +191,7 @@ tsdb_repo_t *tsdbOpenRepo(char *tsdbDir) {
   pRepo->rootDir = strdup(tsdbDir);
 
   tsdbRestoreCfg(pRepo, &(pRepo->config));
+  if (pAppH) pRepo->appH = *pAppH;
 
   pRepo->tsdbMeta = tsdbInitMeta(tsdbDir, pRepo->config.maxTables);
   if (pRepo->tsdbMeta == NULL) {

src/vnode/tsdb/tests/tsdbTests.cpp

@@ -140,7 +140,7 @@ TEST(TsdbTest, createRepo) {
 
 // TEST(TsdbTest, DISABLED_openRepo) {
 TEST(TsdbTest, openRepo) {
-  tsdb_repo_t *repo = tsdbOpenRepo("/home/ubuntu/work/ttest/vnode0");
+  tsdb_repo_t *repo = tsdbOpenRepo("/home/ubuntu/work/ttest/vnode0", NULL);
   ASSERT_NE(repo, nullptr);
 
   STsdbRepo *pRepo = (STsdbRepo *)repo;