Merge branch 'develop' into xiaoping/add_test_case2

documentation20/webdocs/markdowndocs/connector-ch.md

@@ -616,6 +616,43 @@ HTTP请求URL采用`sqlutc`时，返回结果集的时间戳将采用UTC时间
 - httpEnableCompress: 是否支持压缩，默认不支持，目前TDengine仅支持gzip压缩格式
 - httpDebugFlag: 日志开关，131：仅错误和报警信息，135：调试信息，143：非常详细的调试信息，默认131
 
+## CSharp Connector
+
+在Windows系统上，C#应用程序可以使用TDengine的原生C接口来执行所有数据库操作，后续版本将提供ORM（dapper）框架驱动。
+
+#### 安装TDengine客户端
+
+C#连接器需要使用`libtaos.so`和`taos.h`。因此，在使用C#连接器之前，需在程序运行的Windows环境安装TDengine的Windows客户端，以便获得相关驱动文件。
+
+安装完成后，在文件夹`C:/TDengine/examples/C#`中，将会看到两个文件
+
+- TDengineDriver.cs 调用taos.dll文件的Native C方法
+- TDengineTest.cs 参考程序示例
+
+在文件夹`C:\Windows\System32`，将会看到`taos.dll`文件
+
+#### 使用方法
+
+- 将C#接口文件TDengineDriver.cs加入到应用程序所在.NET项目中
+- 参考TDengineTest.cs来定义数据库连接参数，及执行数据插入、查询等操作的方法
+- 因为C#接口需要用到`taos.dll`文件，用户可以将`taos.dll`文件加入.NET解决方案中
+
+#### 注意事项
+
+- `taos.dll`文件使用x64平台编译，所以.NET项目在生成.exe文件时，“解决方案”/“项目”的“平台”请均选择“x64”。
+- 此.NET接口目前已经在Visual Studio 2013/2015/2017中验证过，其它VS版本尚待验证。
+
+#### 第三方驱动
+
+Maikebing.Data.Taos是一个TDengine的ADO.Net提供器，支持linux，windows。该开发包由热心贡献者`麦壳饼@@maikebing`提供，具体请参考
+
+```
+//接口下载
+https://github.com/maikebing/Maikebing.EntityFrameworkCore.Taos   
+//用法说明    
+https://www.taosdata.com/blog/2020/11/02/1901.html                    
+```
+
 
 ## Go Connector
 

src/vnode/src/vnodeMain.c

@@ -310,6 +310,8 @@ int32_t vnodeOpen(int32_t vnode, char *rootDir) {
     pVnode->version = walGetVersion(pVnode->wal);
   }
 
+  walRenew(pVnode->wal);
+
   SSyncInfo syncInfo;
   syncInfo.vgId = pVnode->vgId;
   syncInfo.version = pVnode->version;

src/wal/src/walMgmt.c

@@ -141,22 +141,11 @@ void walClose(void *handle) {
 
 static int32_t walInitObj(SWal *pWal) {
   if (taosMkDir(pWal->path, 0755) != 0) {
-    wError("vgId:%d, file:%s, failed to create directory since %s", pWal->vgId, pWal->path, strerror(errno));
+    wError("vgId:%d, path:%s, failed to create directory since %s", pWal->vgId, pWal->path, strerror(errno));
     return TAOS_SYSTEM_ERROR(errno);
   }
 
-  if (pWal->keep) {
+  wDebug("vgId:%d, object is initialized", pWal->vgId);
-    return TSDB_CODE_SUCCESS;
-  }
-
-  walRenew(pWal);
-
-  if (pWal && pWal->fd < 0) {
-    wError("vgId:%d, file:%s, failed to open file since %s", pWal->vgId, pWal->path, strerror(errno));
-    return TAOS_SYSTEM_ERROR(errno);
-  }
-
-  wDebug("vgId:%d, file is initialized", pWal->vgId);
   return TSDB_CODE_SUCCESS;
 }
 

src/wal/src/walWrite.c

@@ -21,7 +21,7 @@
 #include "twal.h"
 #include "walInt.h"
 
-static int32_t walRestoreWalFile(SWal *pWal, void *pVnode, FWalWrite writeFp, char *name);
+static int32_t walRestoreWalFile(SWal *pWal, void *pVnode, FWalWrite writeFp, char *name, int64_t fileId);
 
 int32_t walRenew(void *handle) {
   if (handle == NULL) return 0;
@@ -80,6 +80,7 @@ int32_t walWrite(void *handle, SWalHead *pHead) {
   int32_t code = 0;
 
   // no wal
+  if (pWal->fd <= 0) return 0;
   if (pWal->level == TAOS_WAL_NOLOG) return 0;
   if (pHead->version <= pWal->version) return 0;
 
@@ -93,8 +94,9 @@ int32_t walWrite(void *handle, SWalHead *pHead) {
     code = TAOS_SYSTEM_ERROR(errno);
     wError("vgId:%d, file:%s, failed to write since %s", pWal->vgId, pWal->name, strerror(errno));
   } else {
+    wTrace("vgId:%d, fileId:%" PRId64 " fd:%d, write wal ver:%" PRId64 ", head ver:%" PRIu64 ", len:%d ", pWal->vgId,
+           pWal->fileId, pWal->fd, pWal->version, pHead->version, pHead->len);
     pWal->version = pHead->version;
-    wTrace("vgId:%d, write version:%" PRId64 ", fileId:%" PRId64, pWal->vgId, pWal->version, pWal->fileId);
   }
 
   pthread_mutex_unlock(&pWal->mutex);
@@ -130,9 +132,9 @@ int32_t walRestore(void *handle, void *pVnode, int32_t (*writeFp)(void *, void *
     snprintf(walName, sizeof(pWal->name), "%s/%s%" PRId64, pWal->path, WAL_PREFIX, fileId);
 
     wDebug("vgId:%d, file:%s, will be restored", pWal->vgId, walName);
-    int32_t code = walRestoreWalFile(pWal, pVnode, writeFp, walName);
+    int32_t code = walRestoreWalFile(pWal, pVnode, writeFp, walName, fileId);
     if (code != TSDB_CODE_SUCCESS) {
-      wDebug("vgId:%d, file:%s, failed to restore since %s", pWal->vgId, walName, tstrerror(code));
+      wError("vgId:%d, file:%s, failed to restore since %s", pWal->vgId, walName, tstrerror(code));
       continue;
     }
 
@@ -187,7 +189,41 @@ int32_t walGetWalFile(void *handle, char *fileName, int64_t *fileId) {
   return code;
 }
 
-static int32_t walRestoreWalFile(SWal *pWal, void *pVnode, FWalWrite writeFp, char *name) {
+static void walFtruncate(SWal *pWal, int32_t fd, int64_t offset) {
+  taosFtruncate(fd, offset);
+  fsync(fd);
+}
+
+static int32_t walSkipCorruptedRecord(SWal *pWal, SWalHead *pHead, int32_t fd, int64_t *offset) {
+  int64_t pos = *offset;
+  while (1) {
+    pos++;
+
+    if (lseek(fd, pos, SEEK_SET) < 0) {
+      wError("vgId:%d, failed to seek from corrupted wal file since %s", pWal->vgId, strerror(errno));
+      return TSDB_CODE_WAL_FILE_CORRUPTED;
+    }
+
+    if (taosTRead(fd, pHead, sizeof(SWalHead)) <= 0) {
+      wError("vgId:%d, read to end of corrupted wal file, offset:%" PRId64, pWal->vgId, pos);
+      return TSDB_CODE_WAL_FILE_CORRUPTED;
+    }
+
+    if (pHead->signature != WAL_SIGNATURE) {
+      continue;
+    }
+
+    if (taosCheckChecksumWhole((uint8_t *)pHead, sizeof(SWalHead))) {
+      wInfo("vgId:%d, wal head cksum check passed, offset:%" PRId64, pWal->vgId, pos);
+      *offset = pos;
+      return TSDB_CODE_SUCCESS;
+    }
+  }
+
+  return TSDB_CODE_WAL_FILE_CORRUPTED;
+}
+
+static int32_t walRestoreWalFile(SWal *pWal, void *pVnode, FWalWrite writeFp, char *name, int64_t fileId) {
   int32_t size = WAL_MAX_SIZE;
   void *  buffer = tmalloc(size);
   if (buffer == NULL) {
@@ -205,7 +241,7 @@ static int32_t walRestoreWalFile(SWal *pWal, void *pVnode, FWalWrite writeFp, ch
   wDebug("vgId:%d, file:%s, start to restore", pWal->vgId, name);
 
   int32_t   code = TSDB_CODE_SUCCESS;
-  size_t    offset = 0;
+  int64_t   offset = 0;
   SWalHead *pHead = buffer;
 
   while (1) {
@@ -213,25 +249,26 @@ static int32_t walRestoreWalFile(SWal *pWal, void *pVnode, FWalWrite writeFp, ch
     if (ret == 0) break;
 
     if (ret < 0) {
-      wError("vgId:%d, file:%s, failed to read wal head part since %s", pWal->vgId, name, strerror(errno));
+      wError("vgId:%d, file:%s, failed to read wal head since %s", pWal->vgId, name, strerror(errno));
       code = TAOS_SYSTEM_ERROR(errno);
       break;
     }
 
     if (ret < sizeof(SWalHead)) {
-      wError("vgId:%d, file:%s, failed to read wal head since %s, read size:%d, skip the rest of file", pWal->vgId,
+      wError("vgId:%d, file:%s, failed to read wal head, ret is %d", pWal->vgId, name, ret);
-             name, strerror(errno), ret);
+      walFtruncate(pWal, fd, offset);
-      taosFtruncate(fd, offset);
-      fsync(fd);
       break;
     }
 
     if (!taosCheckChecksumWhole((uint8_t *)pHead, sizeof(SWalHead))) {
-      wError("vgId:%d, file:%s, wal head cksum is messed up, skip the rest of file", pWal->vgId, name);
+      wError("vgId:%d, file:%s, wal head cksum is messed up, ver:%" PRIu64 " len:%d offset:%" PRId64, pWal->vgId, name,
-      code = TSDB_CODE_WAL_FILE_CORRUPTED;
+             pHead->version, pHead->len, offset);
-      ASSERT(false);
+      code = walSkipCorruptedRecord(pWal, pHead, fd, &offset);
+      if (code != TSDB_CODE_SUCCESS) {
+        walFtruncate(pWal, fd, offset);
         break;
       }
+    }
 
     if (pHead->len > size - sizeof(SWalHead)) {
       size = sizeof(SWalHead) + pHead->len;
@@ -247,24 +284,23 @@ static int32_t walRestoreWalFile(SWal *pWal, void *pVnode, FWalWrite writeFp, ch
 
     ret = taosTRead(fd, pHead->cont, pHead->len);
     if (ret < 0) {
-      wError("vgId:%d, file:%s failed to read wal body part since %s", pWal->vgId, name, strerror(errno));
+      wError("vgId:%d, file:%s, failed to read wal body since %s", pWal->vgId, name, strerror(errno));
       code = TAOS_SYSTEM_ERROR(errno);
       break;
     }
 
     if (ret < pHead->len) {
-      wError("vgId:%d, file:%s, failed to read body since %s, read size:%d len:%d , skip the rest of file", pWal->vgId,
+      wError("vgId:%d, file:%s, failed to read wal body, ret:%d len:%d", pWal->vgId, name, ret, pHead->len);
-             name, strerror(errno), ret, pHead->len);
+      offset += sizeof(SWalHead);
-      taosFtruncate(fd, offset);
+      continue;
-      fsync(fd);
-      break;
     }
 
     offset = offset + sizeof(SWalHead) + pHead->len;
 
-    if (pWal->keep) pWal->version = pHead->version;
+    wTrace("vgId:%d, fileId:%" PRId64 ", restore wal ver:%" PRIu64 ", head ver:%" PRIu64 " len:%d", pWal->vgId, fileId,
+           pWal->version, pHead->version, pHead->len);
 
-    wTrace("vgId:%d, restore version:%" PRIu64 ", fileId:%" PRId64, pWal->vgId, pWal->version, pWal->fileId);
+    if (pWal->keep) pWal->version = pHead->version;
 
     (*writeFp)(pVnode, pHead, TAOS_QTYPE_WAL);
   }