Merge remote-tracking branch 'origin/develop' into feature/wal

documentation20/webdocs/markdowndocs/connector-ch.md

@@ -6,16 +6,16 @@ TDengine提供了丰富的应用程序开发接口，其中包括C/C++、C# 、J
 
 目前TDengine的连接器可支持的平台广泛，目前包括：X64/X86/ARM64/ARM32/MIPS/Alpha等硬件平台，以及Linux/Win64/Win32等开发环境。对照矩阵如下：
 
-|                              | **CPU**   | **X64   64bit** | **X86   32bit** | **ARM64** | **ARM32** | **MIPS **  **龙芯** | **Alpha **  **申威** | **X64 **  **海光** |           |           |
+|                              | **CPU**     | **X64   64bit** | **X64   64bit** | **X64   64bit** | **X86   32bit** | **ARM64** | **ARM32** | **MIPS   龙芯** | **Alpha   申威** | **X64   海光** |
-| ---------------------------- | --------- | --------------- | --------------- | --------- | --------- | ------------------- | -------------------- | ------------------ | --------- | --------- |
+| ---------------------------- | ----------- | --------------- | --------------- | --------------- | --------------- | --------- | --------- | --------------- | ---------------- | -------------- |
-|                              | **OS**    | **Linux**       | **Win64**       | **Win32** | **Win32** | **Linux**           | **Linux**            | **Linux**          | **Linux** | **Linux** |
+|                              | **OS**      | **Linux**       | **Win64**       | **Win32**       | **Win32**       | **Linux** | **Linux** | **Linux**       | **Linux**        | **Linux**      |
-| **连**     **接**     **器** | **C/C++** | ●               | ●               | ●         | ○         | ●                   | ●                    | ●                  | ●         | ●         |
+| **连**     **接**     **器** | **C/C++**   | ●               | ●               | ●               | ○               | ●         | ●         | ●               | ●                | ●              |
-| **JDBC**                     | ●         | ●               | ●               | ○         | ●         | ●                   | ●                    | ●                  | ●         |           |
+|                              | **JDBC**    | ●               | ●               | ●               | ○               | ●         | ●         | ●               | ●                | ●              |
-| **Python**                   | ●         | ●               | ●               | ○         | ●         | ●                   | ●                    | --                 | ●         |           |
+|                              | **Python**  | ●               | ●               | ●               | ○               | ●         | ●         | ●               | --               | ●              |
-| **Go**                       | ●         | ●               | ●               | ○         | ●         | ●                   | ○                    | --                 | --        |           |
+|                              | **Go**      | ●               | ●               | ●               | ○               | ●         | ●         | ○               | --               | --             |
-| **NodeJs**                   | ●         | ●               | ○               | ○         | ●         | ●                   | ○                    | --                 | --        |           |
+|                              | **NodeJs**  | ●               | ●               | ○               | ○               | ●         | ●         | ○               | --               | --             |
-| **C#**                       | ○         | ●               | ●               | ○         | ○         | ○                   | ○                    | --                 | --        |           |
+|                              | **C#**      | ○               | ●               | ●               | ○               | ○         | ○         | ○               | --               | --             |
-| **RESTful**                  | ●         | ●               | ●               | ●         | ●         | ●                   | ●                    | ●                  | ●         |           |
+|                              | **RESTful** | ●               | ●               | ●               | ●               | ●         | ●         | ●               | ●                | ●              |
 
 注意：所有执行 SQL 语句的 API，例如 C/C++ Connector 中的 `tao_query`、`taos_query_a`、`taos_subscribe` 等，以及其它语言中与它们对应的API，每次都只能执行一条 SQL 语句，如果实际参数中包含了多条语句，它们的行为是未定义的。
 

src/client/src/tscParseInsert.c

@@ -1446,18 +1446,21 @@ static void parseFileSendDataBlock(void *param, TAOS_RES *tres, int code) {
   int32_t count = 0;
   int32_t maxRows = 0;
 
-  tscDestroyBlockArrayList(pSql->cmd.pDataBlocks);
+  tfree(pCmd->pTableMetaList);
-  pCmd->pDataBlocks = taosArrayInit(1, POINTER_BYTES);
+  pCmd->pDataBlocks = tscDestroyBlockArrayList(pCmd->pDataBlocks);
+
+  if (pCmd->pTableBlockHashList == NULL) {
+    pCmd->pTableBlockHashList = taosHashInit(16, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BIGINT), true, false);
+  }
 
   STableDataBlocks *pTableDataBlock = NULL;
-  int32_t ret = tscCreateDataBlock(TSDB_PAYLOAD_SIZE, tinfo.rowSize, sizeof(SSubmitBlk), pTableMetaInfo->name, pTableMeta, &pTableDataBlock);
+  int32_t ret = tscGetDataBlockFromList(pCmd->pTableBlockHashList, pTableMeta->id.uid, TSDB_PAYLOAD_SIZE,
+                                        sizeof(SSubmitBlk), tinfo.rowSize, pTableMetaInfo->name, pTableMeta, &pTableDataBlock, NULL);
   if (ret != TSDB_CODE_SUCCESS) {
 //    return ret;
   }
 
-  taosArrayPush(pCmd->pDataBlocks, &pTableDataBlock);
   tscAllocateMemIfNeed(pTableDataBlock, tinfo.rowSize, &maxRows);
-
   char *tokenBuf = calloc(1, 4096);
 
   while ((readLen = tgetline(&line, &n, fp)) != -1) {
@@ -1519,8 +1522,6 @@ void tscProcessMultiVnodesImportFromFile(SSqlObj *pSql) {
 
   SImportFileSupport *pSupporter = calloc(1, sizeof(SImportFileSupport));
   SSqlObj *pNew = createSubqueryObj(pSql, 0, parseFileSendDataBlock, pSupporter, TSDB_SQL_INSERT, NULL);
-
-  pNew->cmd.pDataBlocks = taosArrayInit(4, POINTER_BYTES);
   pCmd->count = 1;
 
   FILE *fp = fopen(pCmd->payload, "r");

src/client/src/tscSQLParser.c

@@ -5913,25 +5913,33 @@ int32_t doLocalQueryProcess(SSqlCmd* pCmd, SQueryInfo* pQueryInfo, SQuerySQL* pQ
   if (pExprList->nExpr != 1) {
     return invalidSqlErrMsg(tscGetErrorMsgPayload(pCmd), msg1);
   }
-
+  bool server_status = false;
   tSQLExpr* pExpr = pExprList->a[0].pNode;
   if (pExpr->operand.z == NULL) {
-    return invalidSqlErrMsg(tscGetErrorMsgPayload(pCmd), msg2);
+    //handle 'select 1'
+    if (pExpr->token.n == 1 && 0 == strncasecmp(pExpr->token.z, "1", 1)) {
+      server_status = true; 
+    } else {
+      return invalidSqlErrMsg(tscGetErrorMsgPayload(pCmd), msg2);
+    } 
   } 
-
   // TODO redefine the function
-  SDNodeDynConfOption functionsInfo[5] = {{"database()", 10},
+   SDNodeDynConfOption functionsInfo[5] = {{"database()", 10},
-                                          {"server_version()", 16},
+                                            {"server_version()", 16},
-                                          {"server_status()", 15},
+                                            {"server_status()", 15},
-                                          {"client_version()", 16},
+                                            {"client_version()", 16},
-                                          {"current_user()", 14}};
+                                            {"current_user()", 14}};
 
   int32_t index = -1;
-  for (int32_t i = 0; i < tListLen(functionsInfo); ++i) {
+  if (server_status == true) {
-    if (strncasecmp(functionsInfo[i].name, pExpr->operand.z, functionsInfo[i].len) == 0 &&
+    index = 2;
-        functionsInfo[i].len == pExpr->operand.n) {
+  } else {
-      index = i;
+    for (int32_t i = 0; i < tListLen(functionsInfo); ++i) {
-      break;
+      if (strncasecmp(functionsInfo[i].name, pExpr->operand.z, functionsInfo[i].len) == 0 &&
+          functionsInfo[i].len == pExpr->operand.n) {
+        index = i;
+        break;
+      }
     }
   }
 

src/client/src/tscServer.c

@@ -214,7 +214,7 @@ int tscSendMsgToServer(SSqlObj *pSql) {
   STscObj* pObj = pSql->pTscObj;
   SSqlCmd* pCmd = &pSql->cmd;
   
-  char *pMsg = rpcMallocCont(pCmd->payloadLen);
+  char *pMsg = rpcMallocCont(sizeof(SMsgVersion) + pCmd->payloadLen);
   if (NULL == pMsg) {
     tscError("%p msg:%s malloc failed", pSql, taosMsg[pSql->cmd.msgType]);
     return TSDB_CODE_TSC_OUT_OF_MEMORY;
@@ -225,12 +225,13 @@ int tscSendMsgToServer(SSqlObj *pSql) {
     tscDumpMgmtEpSet(pSql);
   }
 
-  memcpy(pMsg, pSql->cmd.payload, pSql->cmd.payloadLen);
+  tstrncpy(pMsg, version, sizeof(SMsgVersion));
+  memcpy(pMsg + sizeof(SMsgVersion), pSql->cmd.payload, pSql->cmd.payloadLen);
 
   SRpcMsg rpcMsg = {
       .msgType = pSql->cmd.msgType,
       .pCont   = pMsg,
-      .contLen = pSql->cmd.payloadLen,
+      .contLen = pSql->cmd.payloadLen + sizeof(SMsgVersion),
       .ahandle = (void*)pSql->self,
       .handle  = NULL,
       .code    = 0

src/client/src/tscUtil.c

@@ -2044,7 +2044,11 @@ SSqlObj* createSubqueryObj(SSqlObj* pSql, int16_t tableIndex, void (*fp)(), void
   pnCmd->numOfClause = 0;
   pnCmd->clauseIndex = 0;
   pnCmd->pDataBlocks = NULL;
+
+  pnCmd->numOfTables = 0;
   pnCmd->parseFinished = 1;
+  pnCmd->pTableMetaList = NULL;
+  pnCmd->pTableBlockHashList = NULL;
 
   if (tscAddSubqueryInfo(pnCmd) != TSDB_CODE_SUCCESS) {
     terrno = TSDB_CODE_TSC_OUT_OF_MEMORY;

src/dnode/src/dnodeMRead.c

@@ -124,8 +124,6 @@ void dnodeDispatchToMReadQueue(SRpcMsg *pMsg) {
     SMnodeMsg *pRead = mnodeCreateMsg(pMsg);
     taosWriteQitem(tsMReadQueue, TAOS_QTYPE_RPC, pRead);
   }
-
-  rpcFreeCont(pMsg->pCont);
 }
 
 static void dnodeFreeMReadMsg(SMnodeMsg *pRead) {

src/dnode/src/dnodeMWrite.c

@@ -125,8 +125,6 @@ void dnodeDispatchToMWriteQueue(SRpcMsg *pMsg) {
            taosMsg[pWrite->rpcMsg.msgType], tsMWriteQueue);
     taosWriteQitem(tsMWriteQueue, TAOS_QTYPE_RPC, pWrite);
   }
-
-  rpcFreeCont(pMsg->pCont);
 }
 
 static void dnodeFreeMWriteMsg(SMnodeMsg *pWrite) {

src/dnode/src/dnodeShell.c

@@ -127,7 +127,20 @@ static void dnodeProcessMsgFromShell(SRpcMsg *pMsg, SRpcEpSet *pEpSet) {
   } else {}
 
   if ( dnodeProcessShellMsgFp[pMsg->msgType] ) {
+    SMsgVersion *pMsgVersion = pMsg->pCont;
+    if (taosCheckVersion(pMsgVersion->clientVersion, version, 3) != TSDB_CODE_SUCCESS) {
+      rpcMsg.code = TSDB_CODE_TSC_INVALID_VERSION;
+      rpcSendResponse(&rpcMsg);
+      rpcFreeCont(pMsg->pCont);
+      return; // todo change the error code
+    }
+    pMsg->pCont += sizeof(*pMsgVersion);
+    pMsg->contLen -= sizeof(*pMsgVersion);
+
     (*dnodeProcessShellMsgFp[pMsg->msgType])(pMsg);
+
+    //pMsg->contLen += sizeof(*pMsgVersion);
+    rpcFreeCont(pMsg->pCont - sizeof(*pMsgVersion));
   } else {
     dError("RPC %p, shell msg:%s is not processed", pMsg->handle, taosMsg[pMsg->msgType]);
     rpcMsg.code = TSDB_CODE_DND_MSG_NOT_PROCESSED;

src/dnode/src/dnodeVRead.c

@@ -77,8 +77,6 @@ void dnodeDispatchToVReadQueue(SRpcMsg *pMsg) {
     SRpcMsg rpcRsp = {.handle = pMsg->handle, .code = TSDB_CODE_VND_INVALID_VGROUP_ID};
     rpcSendResponse(&rpcRsp);
   }
-
-  rpcFreeCont(pMsg->pCont);
 }
 
 void *dnodeAllocVQueryQueue(void *pVnode) {

src/dnode/src/dnodeVWrite.c

@@ -102,7 +102,6 @@ void dnodeDispatchToVWriteQueue(SRpcMsg *pRpcMsg) {
   }
 
   vnodeRelease(pVnode);
-  rpcFreeCont(pRpcMsg->pCont);
 }
 
 void *dnodeAllocVWriteQueue(void *pVnode) {

src/inc/taosmsg.h

@@ -198,6 +198,10 @@ typedef struct {
   int32_t numOfVnodes;
 } SMsgDesc;
 
+typedef struct SMsgVersion {
+  char clientVersion[TSDB_VERSION_LEN];
+} SMsgVersion;
+
 typedef struct SMsgHead {
   int32_t contLen;
   int32_t vgId;

src/rpc/src/rpcMain.c

@@ -631,15 +631,19 @@ static void rpcReleaseConn(SRpcConn *pConn) {
     // if there is an outgoing message, free it
     if (pConn->outType && pConn->pReqMsg) {
       SRpcReqContext *pContext = pConn->pContext;
-      if (pContext->pRsp) {   
+      if (pContext) {
+        if (pContext->pRsp) {   
         // for synchronous API, post semaphore to unblock app
-        pContext->pRsp->code = TSDB_CODE_RPC_APP_ERROR;
+          pContext->pRsp->code = TSDB_CODE_RPC_APP_ERROR;
-        pContext->pRsp->pCont = NULL;
+          pContext->pRsp->pCont = NULL;
-        pContext->pRsp->contLen = 0;
+          pContext->pRsp->contLen = 0;
-        tsem_post(pContext->pSem);
+          tsem_post(pContext->pSem);
+        }
+        pContext->pConn = NULL; 
+        taosRemoveRef(tsRpcRefId, pContext->rid);
+      } else {
+        assert(0); 
       }
-      pContext->pConn = NULL; 
-      taosRemoveRef(tsRpcRefId, pContext->rid);
     }
   }
 
@@ -1083,7 +1087,11 @@ static void *rpcProcessMsgFromPeer(SRecvInfo *pRecv) {
         if (code == TSDB_CODE_RPC_INVALID_TIME_STAMP || code == TSDB_CODE_RPC_AUTH_FAILURE) {
           rpcCloseConn(pConn);
         }
-        tDebug("%s %p %p, %s is sent with error code:0x%x", pRpc->label, pConn, (void *)pHead->ahandle, taosMsg[pHead->msgType+1], code);
+        if (pHead->msgType + 1 > 1 && pHead->msgType+1 < TSDB_MSG_TYPE_MAX) {
+          tDebug("%s %p %p, %s is sent with error code:0x%x", pRpc->label, pConn, (void *)pHead->ahandle, taosMsg[pHead->msgType+1], code);
+        } else {
+          tError("%s %p %p, %s is sent with error code:0x%x", pRpc->label, pConn, (void *)pHead->ahandle, taosMsg[pHead->msgType], code);
+        }     
       } 
     } else { // msg is passed to app only parsing is ok 
       rpcProcessIncomingMsg(pConn, pHead, pContext);

src/rpc/src/rpcTcp.c

@@ -242,7 +242,14 @@ static void *taosAcceptTcpConnection(void *arg) {
 
     taosKeepTcpAlive(connFd);
     struct timeval to={1, 0};
-    taosSetSockOpt(connFd, SOL_SOCKET, SO_RCVTIMEO, &to, sizeof(to));
+    int32_t ret = taosSetSockOpt(connFd, SOL_SOCKET, SO_RCVTIMEO, &to, sizeof(to));
+    if (ret != 0) {
+      taosCloseSocket(connFd);
+      tError("%s failed to set recv timeout fd(%s)for connection from:%s:%hu", pServerObj->label, strerror(errno),
+             taosInetNtoa(caddr.sin_addr), htons(caddr.sin_port));
+      continue;
+    }
+    
 
     // pick up the thread to handle this connection
     pThreadObj = pServerObj->pThreadObj[threadId];

src/rpc/test/rclient.c

@@ -188,7 +188,8 @@ int main(int argc, char *argv[]) {
   tInfo("it takes %.3f mseconds to send %d requests to server", usedTime, numOfReqs*appThreads);
   tInfo("Performance: %.3f requests per second, msgSize:%d bytes", 1000.0*numOfReqs*appThreads/usedTime, msgSize);
 
-  getchar();
+  int ch = getchar();
+  UNUSED(ch); 
 
   taosCloseLog();
 

tests/pytest/concurrent_inquiry.py

@@ -166,7 +166,7 @@ class ConcurrentInquiry:
         sel_col_list=[]
         col_rand=random.randint(0,len(col_list))
         for i,j in zip(col_list[0:col_rand],func_list):         #决定每个被查询col的函数
-            alias = 'as '+ str(i)
+            alias = ' as '+ str(i)
             pick_func = ''
             if j == 'leastsquares':
                 pick_func=j+'('+i+',1,1)'

tests/pytest/functions/function_twa_test2.py

@@ -102,7 +102,7 @@ class TDTestCase:
 
         tdSql.query("select twa(c) from t3 where ts >= '2018-09-17 08:59:00.000' and ts <= '2018-09-17 09:01:30.000'")
         tdSql.checkRows(1)
-        tdSql.checkData(-0.5)
+        tdSql.checkData(0, 0, -0.5)
 
         tdSql.query("select twa(c) from t3 where ts >= '2018-09-17 08:59:00.000' and ts <= '2018-09-17 09:01:30.000' interval(1s)")
         tdSql.checkRows(2)

tests/pytest/query/queryNullValueTest.py

@@ -50,7 +50,7 @@ class TDTestCase:
             tdSql.execute("insert into t0 values (%d, NULL)" % (self.ts))
             
             tdDnodes.stop(1)
-            tdLog.sleep(10)
+            # tdLog.sleep(10)
             tdDnodes.start(1)
             tdSql.execute("use db")
             tdSql.query("select * from t0")
@@ -62,7 +62,7 @@ class TDTestCase:
             tdSql.execute("create table t1 (ts timestamp, col %s)" % self.types[i])
             tdSql.execute("insert into t1 values (%d, NULL)" % (self.ts))
             tdDnodes.stop(1)
-            tdLog.sleep(10)
+            # tdLog.sleep(10)
             tdDnodes.start(1)
             tdSql.execute("use db")
 

tests/pytest/query/queryWithTaosdKilled.py

@@ -0,0 +1,68 @@
+###################################################################
+#           Copyright (c) 2016 by TAOS Technologies, Inc.
+#                     All rights reserved.
+#
+#  This file is proprietary and confidential to TAOS Technologies.
+#  No part of this file may be reproduced, stored, transmitted,
+#  disclosed or used in any form or by any means other than as
+#  expressly provided by the written permission from Jianhui Tao
+#
+###################################################################
+
+# -*- coding: utf-8 -*-
+
+import sys
+import os
+import taos
+from util.log import *
+from util.cases import *
+from util.sql import *
+from util.dnodes import *
+import numpy as np
+
+
+class TDTestCase:
+    def init(self, conn, logSql):
+        tdLog.debug("start to execute %s" % __file__)
+        self.conn = conn
+        tdSql.init(conn.cursor())
+
+        self.rowNum = 10
+        self.ts = 1537146000000
+
+    def createOldDir(self):        
+        path = tdDnodes.dnodes[1].getDnodeRootDir(1)
+        print(path)
+        tdLog.info("sudo mkdir -p %s/data/vnode/vnode2/wal/old" % path)
+        os.system("sudo mkdir -p %s/data/vnode/vnode2/wal/old" % path)       
+
+    def run(self):
+        # os.system("rm -rf %s/ " % tdDnodes.getDnodesRootDir())
+        tdSql.prepare()
+
+        tdSql.execute("create table st(ts timestamp, speed int)")
+        tdSql.execute("insert into st values(now, 1)")
+        tdSql.query("select count(*) from st")
+        tdSql.checkRows(1)
+
+        
+        self.createOldDir()
+        tdLog.sleep(10)
+
+        print("force kill taosd")
+        os.system("sudo kill -9 $(pgrep -x taosd)")
+        os.system("")
+        tdDnodes.start(1)
+
+        tdSql.init(self.conn.cursor())
+        tdSql.execute("use db")
+        tdSql.query("select count(*) from st")
+        tdSql.checkRows(1)
+        
+    def stop(self):
+        tdSql.close()
+        tdLog.success("%s successfully executed" % __file__)
+
+
+tdCases.addWindows(__file__, TDTestCase())
+tdCases.addLinux(__file__, TDTestCase())

tests/pytest/update/append_commit_data.py

@@ -38,38 +38,62 @@ class TDTestCase:
 
         insertRows = 200
         t0 = 1604298064000
+        sql='insert into db.t1 values '
+        temp=''
         tdLog.info("insert %d rows" % (insertRows))
         for i in range(0, insertRows):
-          ret = tdSql.execute(
+          # ret = tdSql.execute(
-              'insert into t1 values (%d , 1)' %
+          #     'insert into t1 values (%d , 1)' %
-              (t0+i))
+          #     (t0+i))
+          temp += '(%d,1)' %(t0+i)
+          if i % 100 == 0 or i == (insertRows - 1 ):
+            print(sql+temp)
+            ret = tdSql.execute(
+                sql+temp
+            )
+            temp = ''
         print("==========step2")
         print("restart to commit ")
         tdDnodes.stop(1)
         tdDnodes.start(1)
         tdSql.query("select * from db.t1")
         tdSql.checkRows(insertRows)
+        
         for k in range(0,100):
           tdLog.info("insert %d rows" % (insertRows))
+          temp=''
           for i in range (0,insertRows):
-            ret = tdSql.execute(
+            temp += '(%d,1)' %(t0+k*200+i)
-                'insert into db.t1 values(%d,1)' %
+            if i % 100 == 0 or i == (insertRows - 1 ):
-                (t0+k*200+i)
+              print(sql+temp)
-            )
+              ret = tdSql.execute(
+                  sql+temp
+              )
+              temp = ''
+            
           tdDnodes.stop(1)
           tdDnodes.start(1)
           tdSql.query("select * from db.t1")
           tdSql.checkRows(insertRows+200*k)
-        print("==========step2")
+        print("==========step3")
         print("insert into another table ")
         s = 'use db'
         tdSql.execute(s)
         ret = tdSql.execute('create table t2 (ts timestamp, a int)')
         insertRows = 20000
+        sql = 'insert into t2 values '
+        temp = ''
         for i in range(0, insertRows):
-          ret = tdSql.execute(
+          # ret = tdSql.execute(
-              'insert into t2 values (%d, 1)' %
+          #     'insert into t2 values (%d, 1)' %
-              (t0+i))
+          #     (t0+i))
+          temp += '(%d,1)' %(t0+i)
+          if i % 500 == 0 or i == (insertRows - 1 ):
+            print(sql+temp)
+            ret = tdSql.execute(
+                sql+temp
+            )
+            temp = ''
         tdDnodes.stop(1)
         tdDnodes.start(1)
         tdSql.query("select * from t2")

tests/pytest/util/dnodes.py

@@ -255,9 +255,19 @@ class TDDnode:
             tdLog.exit(cmd)
         self.running = 1
         tdLog.debug("dnode:%d is running with %s " % (self.index, cmd))
+        time.sleep(0.1)
+        key = 'from offline to online'
+        bkey = bytes(key,encoding="utf8")
+        logFile = self.logDir + "/taosdlog.0"
+        popen = subprocess.Popen('tail -f ' + logFile, stdout=subprocess.PIPE, stderr=subprocess.PIPE, shell=True)
+        while True:
+            line = popen.stdout.readline().strip()
+            if bkey in line:
+                popen.kill()
+                break
+        tdLog.debug("the dnode:%d has been started." % (self.index))
         
-        tdLog.debug("wait 5 seconds for the dnode:%d to start." % (self.index))
+        # time.sleep(5)
-        time.sleep(5)
     
     def startWithoutSleep(self):
         buildPath = self.getBuildPath()