Merge branch 'develop' into xiaoping/add_test_case

documentation20/cn/12.taos-sql/docs.md

@@ -29,21 +29,21 @@ taos> DESCRIBE meters;
 
 ## <a class="anchor" id="data-type"></a>支持的数据类型
 
-使用TDengine，最重要的是时间戳。创建并插入记录、查询历史记录的时候，均需要指定时间戳。时间戳有如下规则：
+使用 TDengine，最重要的是时间戳。创建并插入记录、查询历史记录的时候，均需要指定时间戳。时间戳有如下规则：
 
-- 时间格式为```YYYY-MM-DD HH:mm:ss.MS```, 默认时间分辨率为毫秒。比如：```2017-08-12 18:25:58.128```
-- 内部函数now是服务器的当前时间
-- 插入记录时，如果时间戳为now，插入数据时使用服务器当前时间
-- Epoch Time: 时间戳也可以是一个长整数，表示从1970-01-01 08:00:00.000开始的毫秒数
-- 时间可以加减，比如 now-2h，表明查询时刻向前推2个小时(最近2小时)。 数字后面的时间单位可以是 a(毫秒)、s(秒)、 m(分)、h(小时)、d(天)、w(周)。 比如select * from t1 where ts > now-2w and ts <= now-1w, 表示查询两周前整整一周的数据。 在指定降频操作(down sampling)的时间窗口(interval)时，时间单位还可以使用 n(自然月) 和 y(自然年)。
+- 时间格式为 ```YYYY-MM-DD HH:mm:ss.MS```，默认时间分辨率为毫秒。比如：```2017-08-12 18:25:58.128```
+- 内部函数 now 是客户端的当前时间
+- 插入记录时，如果时间戳为 now，插入数据时使用提交这条记录的客户端的当前时间
+- Epoch Time：时间戳也可以是一个长整数，表示从 1970-01-01 08:00:00.000 开始的毫秒数
+- 时间可以加减，比如 now-2h，表明查询时刻向前推 2 个小时（最近 2 小时）。数字后面的时间单位可以是 u(微秒)、a(毫秒)、s(秒)、m(分)、h(小时)、d(天)、w(周)。 比如 `select * from t1 where ts > now-2w and ts <= now-1w`，表示查询两周前整整一周的数据。在指定降频操作（down sampling）的时间窗口（interval）时，时间单位还可以使用 n(自然月) 和 y(自然年)。
 
-TDengine缺省的时间戳是毫秒精度，但通过修改配置参数enableMicrosecond就可支持微秒。
+TDengine 缺省的时间戳是毫秒精度，但通过修改配置参数 enableMicrosecond 就可以支持微秒。
 
-在TDengine中，普通表的数据模型中可使用以下10种数据类型。 
+在TDengine中，普通表的数据模型中可使用以下 10 种数据类型。 
 
 |      |   类型    | Bytes  | 说明                                                         |
 | ---- | :-------: | ------ | ------------------------------------------------------------ |
-| 1    | TIMESTAMP | 8      | 时间戳。缺省精度毫秒，可支持微秒。从格林威治时间 1970-01-01 00:00:00.000 (UTC/GMT) 开始，计时不能早于该时间。 |
+| 1    | TIMESTAMP | 8      | 时间戳。缺省精度毫秒，可支持微秒。从格林威治时间 1970-01-01 00:00:00.000 (UTC/GMT) 开始，计时不能早于该时间。（从 2.0.18 版本开始，已经去除了这一时间范围限制） |
 | 2    |    INT    | 4      | 整型，范围 [-2^31+1,   2^31-1], -2^31 用作 NULL           |
 | 3    |  BIGINT   | 8      | 长整型，范围 [-2^63+1,   2^63-1], -2^63 用于 NULL                                 |
 | 4    |   FLOAT   | 4      | 浮点型，有效位数 6-7，范围 [-3.4E38, 3.4E38]                  |

src/client/src/tscAsync.c

@@ -301,7 +301,7 @@ static void tscAsyncResultCallback(SSchedMsg *pMsg) {
   taosReleaseRef(tscObjRef, pSql->self);
 }
 
-void tscAsyncResultOnError(SSqlObj* pSql) { 
+void tscAsyncResultOnError(SSqlObj* pSql) {
   SSchedMsg schedMsg = {0};
   schedMsg.fp = tscAsyncResultCallback;
   schedMsg.ahandle = (void *)pSql->self;
@@ -505,10 +505,7 @@ void tscTableMetaCallBack(void *param, TAOS_RES *res, int code) {
   return;
 
   _error:
-  if (code != TSDB_CODE_SUCCESS) {
-    pSql->res.code = code;
-    tscAsyncResultOnError(pSql);
-  }
-
+  pRes->code = code;
+  tscAsyncResultOnError(pSql);
   taosReleaseRef(tscObjRef, pSql->self);
 }

src/client/src/tscSQLParser.c

@@ -3293,7 +3293,8 @@ static int32_t extractColumnFilterInfo(SSqlCmd* pCmd, SQueryInfo* pQueryInfo, SC
     }
     
     if (pSchema->type == TSDB_DATA_TYPE_BOOL) {
-      if (pExpr->tokenId != TK_EQ && pExpr->tokenId != TK_NE) {
+      int32_t t = pExpr->tokenId;
+      if (t != TK_EQ && t != TK_NE && t != TK_NOTNULL && t != TK_ISNULL) {
         return invalidSqlErrMsg(tscGetErrorMsgPayload(pCmd), msg3);
       }
     }
@@ -3493,7 +3494,8 @@ static int32_t validateSQLExpr(SSqlCmd* pCmd, tSqlExpr* pExpr, SQueryInfo* pQuer
     }
 
     pList->ids[pList->num++] = index;
-  } else if (pExpr->tokenId == TK_FLOAT && (isnan(pExpr->value.dKey) || isinf(pExpr->value.dKey))) {
+  } else if ((pExpr->tokenId == TK_FLOAT && (isnan(pExpr->value.dKey) || isinf(pExpr->value.dKey))) ||
+             pExpr->tokenId == TK_NULL) {
     return TSDB_CODE_TSC_INVALID_SQL;
   } else if (pExpr->type == SQL_NODE_SQLFUNCTION) {
     if (*type == NON_ARITHMEIC_EXPR) {
@@ -3727,6 +3729,39 @@ static int32_t setExprToCond(tSqlExpr** parent, tSqlExpr* pExpr, const char* msg
   return TSDB_CODE_SUCCESS;
 }
 
+static int32_t validateNullExpr(tSqlExpr* pExpr, char* msgBuf) {
+  const char* msg = "only support is [not] null";
+
+  tSqlExpr* pRight = pExpr->pRight;
+  if (pRight->tokenId == TK_NULL && (!(pExpr->tokenId == TK_ISNULL || pExpr->tokenId == TK_NOTNULL))) {
+    return invalidSqlErrMsg(msgBuf, msg);
+  }
+
+  return TSDB_CODE_SUCCESS;
+}
+
+// check for like expression
+static int32_t validateLikeExpr(tSqlExpr* pExpr, STableMeta* pTableMeta, int32_t index, char* msgBuf) {
+  const char* msg1 = "wildcard string should be less than 20 characters";
+  const char* msg2 = "illegal column name";
+
+  tSqlExpr* pLeft  = pExpr->pLeft;
+  tSqlExpr* pRight = pExpr->pRight;
+
+  if (pExpr->tokenId == TK_LIKE) {
+    if (pRight->value.nLen > TSDB_PATTERN_STRING_MAX_LEN) {
+      return invalidSqlErrMsg(msgBuf, msg1);
+    }
+
+    SSchema* pSchema = tscGetTableSchema(pTableMeta);
+    if ((!isTablenameToken(&pLeft->colInfo)) && !IS_VAR_DATA_TYPE(pSchema[index].type)) {
+      return invalidSqlErrMsg(msgBuf, msg2);
+    }
+  }
+
+  return TSDB_CODE_SUCCESS;
+}
+
 static int32_t handleExprInQueryCond(SSqlCmd* pCmd, SQueryInfo* pQueryInfo, tSqlExpr** pExpr, SCondExpr* pCondExpr,
                                      int32_t* type, int32_t parentOptr) {
   const char* msg1 = "table query cannot use tags filter";
@@ -3736,8 +3771,7 @@ static int32_t handleExprInQueryCond(SSqlCmd* pCmd, SQueryInfo* pQueryInfo, tSql
   const char* msg5 = "not support ordinary column join";
   const char* msg6 = "only one query condition on tbname allowed";
   const char* msg7 = "only in/like allowed in filter table name";
-  const char* msg8 = "wildcard string should be less than 20 characters";
-  
+
   tSqlExpr* pLeft  = (*pExpr)->pLeft;
   tSqlExpr* pRight = (*pExpr)->pRight;
 
@@ -3753,6 +3787,18 @@ static int32_t handleExprInQueryCond(SSqlCmd* pCmd, SQueryInfo* pQueryInfo, tSql
   STableMetaInfo* pTableMetaInfo = tscGetMetaInfo(pQueryInfo, index.tableIndex);
   STableMeta*     pTableMeta = pTableMetaInfo->pTableMeta;
 
+  // validate the null expression
+  int32_t code = validateNullExpr(*pExpr, tscGetErrorMsgPayload(pCmd));
+  if (code != TSDB_CODE_SUCCESS) {
+    return code;
+  }
+
+  // validate the like expression
+  code = validateLikeExpr(*pExpr, pTableMeta, index.columnIndex, tscGetErrorMsgPayload(pCmd));
+  if (code != TSDB_CODE_SUCCESS) {
+    return code;
+  }
+
   if (index.columnIndex == PRIMARYKEY_TIMESTAMP_COL_INDEX) {  // query on time range
     if (!validateJoinExprNode(pCmd, pQueryInfo, *pExpr, &index)) {
       return TSDB_CODE_TSC_INVALID_SQL;
@@ -3774,7 +3820,6 @@ static int32_t handleExprInQueryCond(SSqlCmd* pCmd, SQueryInfo* pQueryInfo, tSql
       
       int16_t leftIdx = index.tableIndex;
 
-      SColumnIndex index = COLUMN_INDEX_INITIALIZER;
       if (getColumnIndexByName(pCmd, &pRight->colInfo, pQueryInfo, &index) != TSDB_CODE_SUCCESS) {
         return invalidSqlErrMsg(tscGetErrorMsgPayload(pCmd), msg2);
       }
@@ -3821,20 +3866,6 @@ static int32_t handleExprInQueryCond(SSqlCmd* pCmd, SQueryInfo* pQueryInfo, tSql
       return invalidSqlErrMsg(tscGetErrorMsgPayload(pCmd), msg1);
     }
 
-    // check for like expression
-    if ((*pExpr)->tokenId == TK_LIKE) {
-      if (pRight->value.nLen > TSDB_PATTERN_STRING_MAX_LEN) {
-        return invalidSqlErrMsg(tscGetErrorMsgPayload(pCmd), msg8);
-      }
-
-      SSchema* pSchema = tscGetTableSchema(pTableMetaInfo->pTableMeta);
-
-      if ((!isTablenameToken(&pLeft->colInfo)) && pSchema[index.columnIndex].type != TSDB_DATA_TYPE_BINARY &&
-          pSchema[index.columnIndex].type != TSDB_DATA_TYPE_NCHAR) {
-        return invalidSqlErrMsg(tscGetErrorMsgPayload(pCmd), msg2);
-      }
-    }
-
     // in case of in operator, keep it in a seprate attribute
     if (index.columnIndex == TSDB_TBNAME_COLUMN_INDEX) {
       if (!validTableNameOptr(*pExpr)) {

src/client/src/tscUtil.c

@@ -309,7 +309,7 @@ void tscSetResRawPtr(SSqlRes* pRes, SQueryInfo* pQueryInfo) {
 
   int32_t offset = 0;
 
-  for (int32_t i = 0; i < pRes->numOfCols; ++i) {
+  for (int32_t i = 0; i < pQueryInfo->fieldsInfo.numOfOutput; ++i) {
     SInternalField* pInfo = (SInternalField*)TARRAY_GET_ELEM(pQueryInfo->fieldsInfo.internalField, i);
 
     pRes->urow[i] = pRes->data + offset * pRes->numOfRows;

src/dnode/src/dnodeMain.c

@@ -238,9 +238,11 @@ static int32_t dnodeInitStorage() {
   }
 
   TDIR *tdir = tfsOpendir("vnode_bak/.staging");
-  if (tfsReaddir(tdir) != NULL) {
+  bool stagingNotEmpty = tfsReaddir(tdir) != NULL;
+  tfsClosedir(tdir);
+
+  if (stagingNotEmpty) {
     dError("vnode_bak/.staging dir not empty, fix it first.");
-    tfsClosedir(tdir);
     return -1;
   }
 

src/dnode/src/dnodeVWrite.c

@@ -205,7 +205,7 @@ static void *dnodeProcessVWriteQueue(void *wparam) {
              pWrite->rpcMsg.ahandle, taosMsg[pWrite->pHead.msgType], qtypeStr[qtype], pWrite->pHead.version);
 
       pWrite->code = vnodeProcessWrite(pVnode, &pWrite->pHead, qtype, pWrite);
-      if (pWrite->code <= 0) pWrite->processedCount = 1;
+      if (pWrite->code <= 0) atomic_add_fetch_32(&pWrite->processedCount, 1);
       if (pWrite->code > 0) pWrite->code = 0;
       if (pWrite->code == 0 && pWrite->pHead.msgType != TSDB_MSG_TYPE_SUBMIT) forceFsync = true;
 

src/kit/taosdemo/taosdemo.c

@@ -85,7 +85,7 @@ enum TEST_MODE {
 #define   MAX_NUM_DATATYPE 10
 
 #define   MAX_DB_COUNT           8
-#define   MAX_SUPER_TABLE_COUNT  8
+#define   MAX_SUPER_TABLE_COUNT  200
 #define   MAX_COLUMN_COUNT       1024
 #define   MAX_TAG_COUNT          128
 
@@ -506,7 +506,7 @@ static int taosRandom()
 
 #endif
 
-static int createDatabases();
+static int createDatabasesAndStables();
 static void createChildTables();
 static int queryDbExec(TAOS *taos, char *command, QUERY_TYPE type, bool quiet);
 
@@ -2416,7 +2416,7 @@ static int createSuperTable(TAOS * taos, char* dbName,
   return 0;
 }
 
-static int createDatabases() {
+static int createDatabasesAndStables() {
   TAOS * taos = NULL;
   int    ret = 0;
   taos = taos_connect(g_Dbs.host, g_Dbs.user, g_Dbs.password, NULL, g_Dbs.port);
@@ -4090,7 +4090,7 @@ static bool getInfoFromJsonFile(char* file) {
   }
 
   bool  ret = false;
-  int   maxLen = 64000;
+  int   maxLen = 6400000;
   char *content = calloc(1, maxLen + 1);
   int   len = fread(content, 1, maxLen, fp);
   if (len <= 0) {
@@ -4455,7 +4455,7 @@ static int generateDataTail(char *tableName, int32_t tableSeq,
         break;
       }
 
-      buffer += sprintf(buffer, " %s", data);
+      buffer += snprintf(buffer, retLen + 1, "%s", data);
       k++;
       len += retLen;
       remainderBufLen -= retLen;
@@ -4682,6 +4682,12 @@ static void* syncWriteInterlace(threadInfo *pThreadInfo) {
 
     for (int i = 0; i < batchPerTblTimes; i ++) {
       getTableName(tableName, pThreadInfo, tableSeq);
+      if (0 == strlen(tableName)) {
+        errorPrint("[%d] %s() LN%d, getTableName return null\n",
+            pThreadInfo->threadID, __func__, __LINE__);
+        return NULL;
+        exit(-1);
+      }
 
       int headLen;
       if (i == 0) {
@@ -4728,7 +4734,7 @@ static void* syncWriteInterlace(threadInfo *pThreadInfo) {
       remainderBufLen -= dataLen;
 
       recOfBatch += batchPerTbl;
-      startTime += batchPerTbl * superTblInfo->timeStampStep;
+//      startTime += batchPerTbl * superTblInfo->timeStampStep;
       pThreadInfo->totalInsertRows += batchPerTbl;
       verbosePrint("[%d] %s() LN%d batchPerTbl=%d recOfBatch=%d\n",
                 pThreadInfo->threadID, __func__, __LINE__,
@@ -4738,9 +4744,12 @@ static void* syncWriteInterlace(threadInfo *pThreadInfo) {
       if (insertMode == INTERLACE_INSERT_MODE) {
           if (tableSeq == pThreadInfo->start_table_from + pThreadInfo->ntables) {
             // turn to first table
-            startTime += batchPerTbl * superTblInfo->timeStampStep;
             tableSeq = pThreadInfo->start_table_from;
             generatedRecPerTbl += batchPerTbl;
+
+            startTime = pThreadInfo->start_time
+              + generatedRecPerTbl * superTblInfo->timeStampStep;
+
             flagSleep = true;
             if (generatedRecPerTbl >= insertRows)
               break;
@@ -5490,7 +5499,7 @@ static int insertTestProcess() {
   init_rand_data();
 
   // create database and super tables
-  if(createDatabases() != 0) {
+  if(createDatabasesAndStables() != 0) {
     fclose(g_fpOfInsertResult);
     return -1;
   }

src/os/inc/osArm32.h

@@ -77,6 +77,7 @@ extern "C" {
 #include <sys/resource.h>
 #include <error.h>
 #include <math.h>
+#include <poll.h>
 
 #define TAOS_OS_FUNC_LZ4
   #define BUILDIN_CLZL(val) __builtin_clzll(val)

src/os/inc/osArm64.h

@@ -78,6 +78,7 @@ extern "C" {
 #include <error.h>
 #include <linux/sysctl.h>
 #include <math.h>
+#include <poll.h>
 
 #ifdef __cplusplus
 }

src/os/inc/osLinux32.h

@@ -77,6 +77,7 @@ extern "C" {
 #include <sys/resource.h>
 #include <error.h>
 #include <math.h>
+#include <poll.h>
 
 #define TAOS_OS_FUNC_LZ4
 #define BUILDIN_CLZL(val) __builtin_clzll(val)

src/os/inc/osLinux64.h

@@ -75,6 +75,7 @@ extern "C" {
 #include <fcntl.h>
 #include <sys/utsname.h>
 #include <sys/resource.h>
+#include <poll.h>
 #ifndef _ALPINE
 #include <error.h>
 #endif

src/os/inc/osSocket.h

@@ -68,6 +68,8 @@ void    taosSetMaskSIGPIPE();
 // TAOS_OS_FUNC_SOCKET_SETSOCKETOPT
 int32_t taosSetSockOpt(SOCKET socketfd, int32_t level, int32_t optname, void *optval, int32_t optlen);
 
+int32_t taosGetSockOpt(SOCKET socketfd, int32_t level, int32_t optname, void *optval, int32_t* optlen);
+
 // TAOS_OS_FUNC_SOCKET_INET
 uint32_t    taosInetAddr(char *ipAddr);
 const char *taosInetNtoa(struct in_addr ipInt);

src/os/src/detail/osSocket.c

@@ -71,6 +71,9 @@ int32_t taosSetSockOpt(SOCKET socketfd, int32_t level, int32_t optname, void *op
   return setsockopt(socketfd, level, optname, optval, (socklen_t)optlen);
 }
 
+int32_t taosGetSockOpt(SOCKET socketfd, int32_t level, int32_t optname, void *optval, int32_t* optlen) {
+  return getsockopt(socketfd, level, optname, optval, (socklen_t *)optlen);
+} 
 #endif
 
 #ifndef TAOS_OS_FUNC_SOCKET_INET

src/query/inc/sql.y

@@ -675,6 +675,7 @@ expr(A) ::= STRING(X).           { A = tSqlExprCreateIdValue(&X, TK_STRING);}
 expr(A) ::= NOW(X).              { A = tSqlExprCreateIdValue(&X, TK_NOW); }
 expr(A) ::= VARIABLE(X).         { A = tSqlExprCreateIdValue(&X, TK_VARIABLE);}
 expr(A) ::= BOOL(X).             { A = tSqlExprCreateIdValue(&X, TK_BOOL);}
+expr(A) ::= NULL(X).             { A = tSqlExprCreateIdValue(&X, TK_NULL);}
 
 // ordinary functions: min(x), max(x), top(k, 20)
 expr(A) ::= ID(X) LP exprlist(Y) RP(E). { A = tSqlExprCreateFunction(Y, &X, &E, X.type); }

src/query/src/qSqlParser.c

@@ -127,7 +127,12 @@ tSqlExpr *tSqlExprCreateIdValue(SStrToken *pToken, int32_t optrType) {
     pSqlExpr->token = *pToken;
   }
 
-  if (optrType == TK_INTEGER || optrType == TK_STRING || optrType == TK_FLOAT || optrType == TK_BOOL) {
+  if (optrType == TK_NULL) {
+    pToken->type = TSDB_DATA_TYPE_NULL;
+    tVariantCreate(&pSqlExpr->value, pToken);
+    pSqlExpr->tokenId = optrType;
+    pSqlExpr->type    = SQL_NODE_VALUE;
+  } else if (optrType == TK_INTEGER || optrType == TK_STRING || optrType == TK_FLOAT || optrType == TK_BOOL) {
     toTSDBType(pToken->type);
 
     tVariantCreate(&pSqlExpr->value, pToken);
@@ -356,7 +361,11 @@ void tSqlExprCompact(tSqlExpr** pExpr) {
 
 bool tSqlExprIsLeaf(tSqlExpr* pExpr) {
   return (pExpr->pRight == NULL && pExpr->pLeft == NULL) &&
-         (pExpr->tokenId == 0 || pExpr->tokenId == TK_ID || (pExpr->tokenId >= TK_BOOL && pExpr->tokenId <= TK_NCHAR) || pExpr->tokenId == TK_SET);
+         (pExpr->tokenId == 0 ||
+         (pExpr->tokenId == TK_ID) ||
+         (pExpr->tokenId >= TK_BOOL && pExpr->tokenId <= TK_NCHAR) ||
+         (pExpr->tokenId == TK_NULL) ||
+         (pExpr->tokenId == TK_SET));
 }
 
 bool tSqlExprIsParentOfLeaf(tSqlExpr* pExpr) {

src/query/src/queryMain.c

@@ -372,6 +372,7 @@ int32_t qKillQuery(qinfo_t qinfo) {
     return TSDB_CODE_QRY_INVALID_QHANDLE;
   }
 
+  qDebug("QInfo:%"PRIu64" query killed", pQInfo->qId);
   setQueryKilled(pQInfo);
 
   // Wait for the query executing thread being stopped/

src/util/src/tsocket.c

@@ -22,6 +22,8 @@
   #define SIGPIPE EPIPE
 #endif
 
+#define TCP_CONN_TIMEOUT 3000  // conn timeout
+
 int32_t taosGetFqdn(char *fqdn) {
   char hostname[1024];
   hostname[1023] = '\0';
@@ -346,10 +348,47 @@ SOCKET taosOpenTcpClientSocket(uint32_t destIp, uint16_t destPort, uint32_t clie
   serverAddr.sin_addr.s_addr = destIp;
   serverAddr.sin_port = (uint16_t)htons((uint16_t)destPort);
 
-  ret = connect(sockFd, (struct sockaddr *)&serverAddr, sizeof(serverAddr));
+#ifdef _TD_LINUX 
+  taosSetNonblocking(sockFd, 1);   
+  ret = connect(sockFd, (struct sockaddr *)&serverAddr, sizeof(serverAddr)); 
+  if (ret == -1) {
+    if (errno == EHOSTUNREACH) {
+      uError("failed to connect socket, ip:0x%x, port:%hu(%s)", destIp, destPort, strerror(errno));
+      taosCloseSocket(sockFd);
+      return -1; 
+    } else if (errno == EINPROGRESS || errno == EAGAIN || errno == EWOULDBLOCK) {
+      struct pollfd wfd[1]; 
+
+      wfd[0].fd = sockFd;
+      wfd[0].events = POLLOUT;
+    
+      int res = poll(wfd, 1, TCP_CONN_TIMEOUT);
+      if (res == -1 || res == 0) {
+        uError("failed to connect socket, ip:0x%x, port:%hu(poll error/conn timeout)", destIp, destPort);
+        taosCloseSocket(sockFd); //  
+        return -1;
+      }
+      int optVal = -1, optLen = sizeof(int);  
+      if ((0 != taosGetSockOpt(sockFd, SOL_SOCKET, SO_ERROR, &optVal, &optLen)) || (optVal != 0)) {
+        uError("failed to connect socket, ip:0x%x, port:%hu(connect host error)", destIp, destPort);
+        taosCloseSocket(sockFd); //  
+        return -1;
+      }
+      ret = 0;
+    } else { // Other error
+      uError("failed to connect socket, ip:0x%x, port:%hu(target host cannot be reached)", destIp, destPort);
+      taosCloseSocket(sockFd); //  
+      return -1; 
+    } 
+  }
+  taosSetNonblocking(sockFd, 0);   
+
+#else
+  ret = connect(sockFd, (struct sockaddr *)&serverAddr, sizeof(serverAddr)); 
+#endif
 
   if (ret != 0) {
-    // uError("failed to connect socket, ip:0x%x, port:%hu(%s)", destIp, destPort, strerror(errno));
+    uError("failed to connect socket, ip:0x%x, port:%hu(%s)", destIp, destPort, strerror(errno));
     taosCloseSocket(sockFd);
     sockFd = -1;
   } else {

src/vnode/src/vnodeRead.c

@@ -364,7 +364,7 @@ static int32_t vnodeProcessFetchMsg(SVnodeObj *pVnode, SVReadMsg *pRead) {
 
   // register the qhandle to connect to quit query immediate if connection is broken
   if (vnodeNotifyCurrentQhandle(pRead->rpcHandle, pRetrieve->qId, *handle, pVnode->vgId) != TSDB_CODE_SUCCESS) {
-    vError("vgId:%d, QInfo:%"PRIu64 "-%p, retrieve discarded since link is broken, %p", pVnode->vgId, pRetrieve->qhandle, *handle, pRead->rpcHandle);
+    vError("vgId:%d, QInfo:%"PRIu64 "-%p, retrieve discarded since link is broken, conn:%p", pVnode->vgId, pRetrieve->qhandle, *handle, pRead->rpcHandle);
     code = TSDB_CODE_RPC_NETWORK_UNAVAIL;
     qKillQuery(*handle);
     qReleaseQInfo(pVnode->qMgmt, (void **)&handle, true);
@@ -409,7 +409,7 @@ static int32_t vnodeProcessFetchMsg(SVnodeObj *pVnode, SVReadMsg *pRead) {
 // client is broken, the query needs to be killed immediately.
 int32_t vnodeNotifyCurrentQhandle(void *handle, uint64_t qId, void *qhandle, int32_t vgId) {
   SRetrieveTableMsg *pMsg = rpcMallocCont(sizeof(SRetrieveTableMsg));
-  pMsg->qhandle = htobe64((uint64_t)qhandle);
+  pMsg->qId = htobe64(qId);
   pMsg->header.vgId = htonl(vgId);
   pMsg->header.contLen = htonl(sizeof(SRetrieveTableMsg));
 

src/vnode/src/vnodeWrite.c

@@ -96,6 +96,7 @@ int32_t vnodeProcessWrite(void *vparam, void *wparam, int32_t qtype, void *rpara
   // write into WAL
   code = walWrite(pVnode->wal, pHead);
   if (code < 0) {
+    if (syncCode > 0) atomic_sub_fetch_32(&pWrite->processedCount, 1);
     vError("vgId:%d, hver:%" PRIu64 " vver:%" PRIu64 " code:0x%x", pVnode->vgId, pHead->version, pVnode->version, code);
     return code;
   }
@@ -104,7 +105,10 @@ int32_t vnodeProcessWrite(void *vparam, void *wparam, int32_t qtype, void *rpara
 
   // write data locally
   code = (*vnodeProcessWriteMsgFp[pHead->msgType])(pVnode, pHead->cont, pRspRet);
-  if (code < 0) return code;
+  if (code < 0) {
+    if (syncCode > 0) atomic_sub_fetch_32(&pWrite->processedCount, 1);
+    return code;
+  }
 
   return syncCode;
 }

tests/script/general/parser/topbot.sim

@@ -325,4 +325,56 @@ if $row != 0 then
   return -1
 endi
 
+print ===============================>td-3621
+sql create table ttm2(ts timestamp, k bool);
+sql insert into ttm2 values('2021-1-1 1:1:1', true)
+sql insert into ttm2 values('2021-1-1 1:1:2', NULL)
+sql insert into ttm2 values('2021-1-1 1:1:3', false)
+sql select * from ttm2 where k is not null
+if $row !=  2 then
+  return -1
+endi
+
+if $data00 != @21-01-01 01:01:01.000@ then
+  print expect 21-01-01 01:01:01.000, actual $data00
+  return -1
+endi
+
+sql select * from ttm2 where k is null
+if $row != 1 then
+  return -1
+endi
+
+if $data00 != @21-01-01 01:01:02.000@ then
+  return -1
+endi
+
+sql select * from ttm2 where k=true
+if $row != 1 then
+  return -1
+endi
+
+if $data00 != @21-01-01 01:01:01.000@ then
+  return -1
+endi
+
+sql select * from ttm2 where k=false
+if $row != 1 then
+  return -1
+endi
+
+if $data00 != @21-01-01 01:01:03.000@ then
+  return -1
+endi
+
+sql select * from ttm2 where k<>false
+if $row != 1 then
+  return -1
+endi
+
+sql_error select * from ttm2 where k=null
+sql_error select * from ttm2 where k<>null
+sql_error select * from ttm2 where k like null
+sql_error select * from ttm2 where k<null
+
 system sh/exec.sh -n dnode1 -s stop -x SIGINT
\ No newline at end of file