Merge branch 'develop' into enhance/TD-2955

src/client/src/tscProfile.c

@@ -233,6 +233,7 @@ int tscBuildQueryStreamDesc(void *pMsg, STscObj *pObj) {
 
   // We extract the lock to tscBuildHeartBeatMsg function.
 
+  int64_t now = taosGetTimestampMs();
   SSqlObj *pSql = pObj->sqlList;
   while (pSql) {
     /*
@@ -247,7 +248,8 @@ int tscBuildQueryStreamDesc(void *pMsg, STscObj *pObj) {
     tstrncpy(pQdesc->sql, pSql->sqlstr, sizeof(pQdesc->sql));
     pQdesc->stime = htobe64(pSql->stime);
     pQdesc->queryId = htonl(pSql->queryId);
-    pQdesc->useconds = htobe64(pSql->res.useconds);
+    //pQdesc->useconds = htobe64(pSql->res.useconds);
+    pQdesc->useconds = htobe64(now - pSql->stime);
     pQdesc->qHandle = htobe64(pSql->res.qhandle);
 
     pHeartbeat->numOfQueries++;

src/client/src/tscSystem.c

@@ -64,7 +64,7 @@ void tscReleaseRpc(void *param)  {
     return;
   }
   pthread_mutex_lock(&rpcObjMutex);
-  taosCacheRelease(tscRpcCache, (void *)&param, true); 
+  taosCacheRelease(tscRpcCache, (void *)&param, false); 
   pthread_mutex_unlock(&rpcObjMutex);
 } 
 
@@ -101,7 +101,7 @@ int32_t tscAcquireRpc(const char *key, const char *user, const char *secretEncry
     tscError("failed to init connection to TDengine");
     return -1;
   } 
-  pRpcObj = taosCachePut(tscRpcCache, rpcObj.key, strlen(rpcObj.key), &rpcObj, sizeof(rpcObj), 1000*10);   
+  pRpcObj = taosCachePut(tscRpcCache, rpcObj.key, strlen(rpcObj.key), &rpcObj, sizeof(rpcObj), 1000*5);   
   if (pRpcObj == NULL) {
     rpcClose(rpcObj.pDnodeConn);
     pthread_mutex_unlock(&rpcObjMutex);
@@ -154,6 +154,7 @@ void taos_init_imp(void) {
   if (tscNumOfThreads < 2) {
     tscNumOfThreads = 2;
   }
+  taosTmrThreads = tscNumOfThreads;
 
   tscQhandle = taosInitScheduler(queueSize, tscNumOfThreads, "tsc");
   if (NULL == tscQhandle) {

src/mnode/src/mnodeProfile.c

@@ -356,7 +356,7 @@ static int32_t mnodeGetQueryMeta(STableMetaMsg *pMeta, SShowObj *pShow, void *pC
 
   pShow->bytes[cols] = 8;
   pSchema[cols].type = TSDB_DATA_TYPE_BIGINT;
-  strcpy(pSchema[cols].name, "time(us)");
+  strcpy(pSchema[cols].name, "time");
   pSchema[cols].bytes = htons(pShow->bytes[cols]);
   cols++;
 

src/query/src/qExecutor.c

@@ -7649,7 +7649,10 @@ int32_t qDumpRetrieveResult(qinfo_t qinfo, SRetrieveTableRsp **pRsp, int32_t *co
     *continueExec = true;
     qDebug("QInfo:%p has more results to retrieve", pQInfo);
   }
-
+  if (pQInfo->code != TSDB_CODE_SUCCESS) {
+    rpcFreeCont(*pRsp);
+    *pRsp = NULL;
+  }
   return pQInfo->code;
 }
 

src/wal/src/walWrite.c

@@ -310,6 +310,16 @@ static int32_t walRestoreWalFile(SWal *pWal, void *pVnode, FWalWrite writeFp, ch
       break;
     }
 
+    if (pHead->sver == 0 && !walValidateChecksum(pHead)) {
+      wError("vgId:%d, file:%s, wal head cksum is messed up, hver:%" PRIu64 " len:%d offset:%" PRId64, pWal->vgId, name,
+	     pHead->version, pHead->len, offset);
+      code = walSkipCorruptedRecord(pWal, pHead, tfd, &offset);
+      if (code != TSDB_CODE_SUCCESS) {
+	walFtruncate(pWal, tfd, offset);
+	break;
+      }
+    }
+
     if (pHead->len < 0 || pHead->len > size - sizeof(SWalHead)) {
       wError("vgId:%d, file:%s, wal head len out of range, hver:%" PRIu64 " len:%d offset:%" PRId64, pWal->vgId, name,
              pHead->version, pHead->len, offset);
@@ -333,7 +343,7 @@ static int32_t walRestoreWalFile(SWal *pWal, void *pVnode, FWalWrite writeFp, ch
       continue;
     }
 
-    if (!walValidateChecksum(pHead)) {
+    if (pHead->sver == 1 && !walValidateChecksum(pHead)) {
       wError("vgId:%d, file:%s, wal head cksum is messed up, hver:%" PRIu64 " len:%d offset:%" PRId64, pWal->vgId, name,
              pHead->version, pHead->len, offset);
       code = walSkipCorruptedRecord(pWal, pHead, tfd, &offset);