Merge branch 'develop' into test/TD-4475

src/client/inc/tsclient.h

@@ -283,6 +283,7 @@ typedef struct SSqlStream {
   int64_t ctime;     // stream created time
   int64_t stime;     // stream next executed time
   int64_t etime;     // stream end query time, when time is larger then etime, the stream will be closed
+  int64_t ltime;     // stream last row time in stream table
   SInterval interval;
   void *  pTimer;
 

src/client/src/tscStream.c

@@ -24,6 +24,7 @@
 #include "tutil.h"
 
 #include "tscProfile.h"
+#include "tscSubquery.h"
 
 static void tscProcessStreamQueryCallback(void *param, TAOS_RES *tres, int numOfRows);
 static void tscProcessStreamRetrieveResult(void *param, TAOS_RES *res, int numOfRows);
@@ -47,8 +48,8 @@ static bool isProjectStream(SQueryInfo* pQueryInfo) {
 
 static int64_t tscGetRetryDelayTime(SSqlStream* pStream, int64_t slidingTime, int16_t prec) {
   float retryRangeFactor = 0.3f;
-  int64_t retryDelta = (int64_t)(tsStreamCompRetryDelay * retryRangeFactor);
-  retryDelta = ((rand() % retryDelta) + tsStreamCompRetryDelay) * 1000L;
+  int64_t retryDelta = (int64_t)(tsRetryStreamCompDelay * retryRangeFactor);
+  retryDelta = ((rand() % retryDelta) + tsRetryStreamCompDelay) * 1000L;
 
   if (pStream->interval.intervalUnit != 'n' && pStream->interval.intervalUnit != 'y') {
     // change to ms
@@ -575,6 +576,14 @@ static void tscCreateStream(void *param, TAOS_RES *res, int code) {
 
   pStream->stime = tscGetStreamStartTimestamp(pSql, pStream, pStream->stime);
 
+  // set stime with ltime if ltime > stime
+  const char* dstTable = pStream->dstTable? pStream->dstTable: "";
+  tscDebug(" CQ table=%s ltime is %"PRId64, dstTable, pStream->ltime);
+  if(pStream->ltime != INT64_MIN && pStream->ltime > pStream->stime) {
+    tscWarn(" CQ set stream %s stime=%"PRId64" replace with ltime=%"PRId64" if ltime>0  ", dstTable, pStream->stime, pStream->ltime);
+    pStream->stime = pStream->ltime;
+  }
+
   int64_t starttime = tscGetLaunchTimestamp(pStream);
   pCmd->command = TSDB_SQL_SELECT;
 
@@ -590,7 +599,66 @@ void tscSetStreamDestTable(SSqlStream* pStream, const char* dstTable) {
   pStream->dstTable = dstTable;
 }
 
-TAOS_STREAM *taos_open_stream(TAOS *taos, const char *sqlstr, void (*fp)(void *param, TAOS_RES *, TAOS_ROW row),
+// fetchFp call back
+void fetchFpStreamLastRow(void* param ,TAOS_RES* res, int num) {
+  SSqlStream* pStream = (SSqlStream*)param;
+  SSqlObj* pSql = res;
+  
+  // get row data set to ltime
+  tscSetSqlOwner(pSql);
+  TAOS_ROW row = doSetResultRowData(pSql);
+  if( row && row[0] ) {
+    pStream->ltime = *((int64_t*)row[0]);
+    const char* dstTable = pStream->dstTable? pStream->dstTable: "";
+    tscDebug(" CQ stream table=%s last row time=%"PRId64" .", dstTable, pStream->ltime);
+  }
+  tscClearSqlOwner(pSql);
+
+  // no condition call 
+  tscCreateStream(param, pStream->pSql, TSDB_CODE_SUCCESS);
+  taos_free_result(res);
+}
+
+//  fp callback 
+void fpStreamLastRow(void* param ,TAOS_RES* res, int code) {
+  // check result successful
+  if (code != TSDB_CODE_SUCCESS) {
+    tscCreateStream(param, res, TSDB_CODE_SUCCESS);
+    taos_free_result(res);
+    return ;
+  }
+
+  // asynchronous fetch last row data
+  taos_fetch_rows_a(res, fetchFpStreamLastRow, param);
+}
+
+void cbParseSql(void* param, TAOS_RES* res, int code) {
+  // check result successful
+  SSqlStream* pStream = (SSqlStream*)param;
+  SSqlObj* pSql = pStream->pSql;
+  SSqlCmd* pCmd = &pSql->cmd;
+  if (code != TSDB_CODE_SUCCESS) {
+    pSql->res.code = code;
+    tscDebug("0x%"PRIx64" open stream parse sql failed, sql:%s, reason:%s, code:%s", pSql->self, pSql->sqlstr, pCmd->payload, tstrerror(code));
+    pStream->fp(pStream->param, NULL, NULL);
+    return;
+  }
+
+  // check dstTable valid
+  if(pStream->dstTable == NULL || strlen(pStream->dstTable) == 0) {
+    tscDebug(" cbParseSql dstTable is empty.");
+    tscCreateStream(param, res, code);
+    return ;
+  }
+
+  // query stream last row time async
+  char sql[128] = "";
+  sprintf(sql, "select last_row(*) from %s;", pStream->dstTable); 
+  taos_query_a(pSql->pTscObj, sql, fpStreamLastRow, param);
+  return ;
+}
+
+TAOS_STREAM *taos_open_stream_withname(TAOS *taos, const char* dstTable, const char *sqlstr, void (*fp)(void *param, TAOS_RES *, TAOS_ROW row),
                               int64_t stime, void *param, void (*callback)(void *)) {
   STscObj *pObj = (STscObj *)taos;
   if (pObj == NULL || pObj->signature != pObj) return NULL;
@@ -613,11 +681,16 @@ TAOS_STREAM *taos_open_stream(TAOS *taos, const char *sqlstr, void (*fp)(void *p
     return NULL;
   }
 
-  pStream->stime    = stime;
-  pStream->fp       = fp;
+  pStream->ltime = INT64_MIN;
+  pStream->stime = stime;
+  pStream->fp = fp;
   pStream->callback = callback;
-  pStream->param    = param;
-  pStream->pSql     = pSql;
+  pStream->param = param;
+  pStream->pSql = pSql;
+  pSql->pStream = pStream;
+  pSql->param = pStream;
+  pSql->maxRetry = TSDB_MAX_REPLICA;
+  tscSetStreamDestTable(pStream, dstTable);
 
   pSql->pStream  = pStream;
   pSql->param    = pStream;
@@ -640,10 +713,17 @@ TAOS_STREAM *taos_open_stream(TAOS *taos, const char *sqlstr, void (*fp)(void *p
 
   tscDebugL("0x%"PRIx64" SQL: %s", pSql->self, pSql->sqlstr);
 
+  pSql->fp      = cbParseSql;
+  pSql->fetchFp = cbParseSql;
+
+  registerSqlObj(pSql);
+
   int32_t code = tsParseSql(pSql, true);
   if (code == TSDB_CODE_SUCCESS) {
-    tscCreateStream(pStream, pSql, code);
-  } else if (code != TSDB_CODE_TSC_ACTION_IN_PROGRESS) {
+    cbParseSql(pStream, pSql, code);
+  } else if (code == TSDB_CODE_TSC_ACTION_IN_PROGRESS) {
+     tscDebug(" CQ taso_open_stream IN Process. sql=%s", sqlstr);
+  } else {
     tscError("0x%"PRIx64" open stream failed, sql:%s, code:%s", pSql->self, sqlstr, tstrerror(code));
     taosReleaseRef(tscObjRef, pSql->self);
     free(pStream);
@@ -653,6 +733,11 @@ TAOS_STREAM *taos_open_stream(TAOS *taos, const char *sqlstr, void (*fp)(void *p
   return pStream;
 }
 
+TAOS_STREAM *taos_open_stream(TAOS *taos, const char *sqlstr, void (*fp)(void *param, TAOS_RES *, TAOS_ROW row),
+                              int64_t stime, void *param, void (*callback)(void *)) {  
+  return taos_open_stream_withname(taos, "", sqlstr, fp, stime, param, callback);
+}
+
 void taos_close_stream(TAOS_STREAM *handle) {
   SSqlStream *pStream = (SSqlStream *)handle;
 

src/common/inc/tdataformat.h

@@ -319,7 +319,7 @@ int        tdInitDataCols(SDataCols *pCols, STSchema *pSchema);
 SDataCols *tdDupDataCols(SDataCols *pCols, bool keepData);
 SDataCols *tdFreeDataCols(SDataCols *pCols);
 void       tdAppendDataRowToDataCol(SDataRow row, STSchema *pSchema, SDataCols *pCols);
-int        tdMergeDataCols(SDataCols *target, SDataCols *src, int rowsToMerge);
+int        tdMergeDataCols(SDataCols *target, SDataCols *source, int rowsToMerge, int *pOffset);
 
 // ----------------- K-V data row structure
 /*

src/common/inc/tglobal.h

@@ -39,6 +39,7 @@ extern int8_t   tsEnableTelemetryReporting;
 extern char     tsEmail[];
 extern char     tsArbitrator[];
 extern int8_t   tsArbOnline;
+extern int64_t  tsArbOnlineTimestamp;
 extern int32_t  tsDnodeId;
 
 // common
@@ -75,7 +76,7 @@ extern int32_t tsMinSlidingTime;
 extern int32_t tsMinIntervalTime;
 extern int32_t tsMaxStreamComputDelay;
 extern int32_t tsStreamCompStartDelay;
-extern int32_t tsStreamCompRetryDelay;
+extern int32_t tsRetryStreamCompDelay;
 extern float   tsStreamComputDelayRatio;  // the delayed computing ration of the whole time window
 extern int32_t tsProjectExecInterval;
 extern int64_t tsMaxRetentWindow;

src/common/src/tdataformat.c

@@ -441,30 +441,35 @@ void tdAppendDataRowToDataCol(SDataRow row, STSchema *pSchema, SDataCols *pCols)
   pCols->numOfRows++;
 }
 
-int tdMergeDataCols(SDataCols *target, SDataCols *source, int rowsToMerge) {
+int tdMergeDataCols(SDataCols *target, SDataCols *source, int rowsToMerge, int *pOffset) {
   ASSERT(rowsToMerge > 0 && rowsToMerge <= source->numOfRows);
   ASSERT(target->numOfCols == source->numOfCols);
+  int offset = 0;
+
+  if (pOffset == NULL) {
+    pOffset = &offset;
+  }
 
   SDataCols *pTarget = NULL;
 
-  if (dataColsKeyLast(target) < dataColsKeyFirst(source)) {  // No overlap
+  if ((target->numOfRows == 0) || (dataColsKeyLast(target) < dataColsKeyFirst(source))) {  // No overlap
     ASSERT(target->numOfRows + rowsToMerge <= target->maxPoints);
     for (int i = 0; i < rowsToMerge; i++) {
       for (int j = 0; j < source->numOfCols; j++) {
         if (source->cols[j].len > 0) {
-          dataColAppendVal(target->cols + j, tdGetColDataOfRow(source->cols + j, i), target->numOfRows,
+          dataColAppendVal(target->cols + j, tdGetColDataOfRow(source->cols + j, i + (*pOffset)), target->numOfRows,
                            target->maxPoints);
         }
       }
       target->numOfRows++;
     }
+    (*pOffset) += rowsToMerge;
   } else {
     pTarget = tdDupDataCols(target, true);
     if (pTarget == NULL) goto _err;
 
     int iter1 = 0;
-    int iter2 = 0;
-    tdMergeTwoDataCols(target, pTarget, &iter1, pTarget->numOfRows, source, &iter2, source->numOfRows,
+    tdMergeTwoDataCols(target, pTarget, &iter1, pTarget->numOfRows, source, pOffset, source->numOfRows,
                        pTarget->numOfRows + rowsToMerge);
   }
 

src/common/src/tglobal.c

@@ -42,6 +42,7 @@ int32_t  tsNumOfMnodes = 3;
 int8_t   tsEnableVnodeBak = 1;
 int8_t   tsEnableTelemetryReporting = 1;
 int8_t   tsArbOnline = 0;
+int64_t  tsArbOnlineTimestamp = TSDB_ARB_DUMMY_TIME;
 char     tsEmail[TSDB_FQDN_LEN] = {0};
 int32_t  tsDnodeId = 0;
 
@@ -93,7 +94,7 @@ int32_t tsMaxStreamComputDelay = 20000;
 int32_t tsStreamCompStartDelay = 10000;
 
 // the stream computing delay time after executing failed, change accordingly
-int32_t tsStreamCompRetryDelay = 10;
+int32_t tsRetryStreamCompDelay = 10*1000;
 
 // The delayed computing ration. 10% of the whole computing time window by default.
 float tsStreamComputDelayRatio = 0.1f;
@@ -710,7 +711,7 @@ static void doInitGlobalConfig(void) {
   taosInitConfigOption(cfg);
 
   cfg.option = "retryStreamCompDelay";
-  cfg.ptr = &tsStreamCompRetryDelay;
+  cfg.ptr = &tsRetryStreamCompDelay;
   cfg.valType = TAOS_CFG_VTYPE_INT32;
   cfg.cfgType = TSDB_CFG_CTYPE_B_CONFIG | TSDB_CFG_CTYPE_B_SHOW;
   cfg.minValue = 10;

src/cq/src/cqMain.c

@@ -437,6 +437,10 @@ static void cqProcessCreateTimer(void *param, void *tmrId) {
   taosReleaseRef(cqObjRef, (int64_t)param);
 }
 
+// inner implement in tscStream.c
+TAOS_STREAM *taos_open_stream_withname(TAOS *taos, const char* desName, const char *sqlstr, void (*fp)(void *param, TAOS_RES *, TAOS_ROW row),
+                              int64_t stime, void *param, void (*callback)(void *));
+
 static void cqCreateStream(SCqContext *pContext, SCqObj *pObj) {
   pObj->pContext = pContext;
 
@@ -449,11 +453,10 @@ static void cqCreateStream(SCqContext *pContext, SCqObj *pObj) {
   pObj->tmrId = 0;
 
   if (pObj->pStream == NULL) {
-    pObj->pStream = taos_open_stream(pContext->dbConn, pObj->sqlStr, cqProcessStreamRes, INT64_MIN, (void *)pObj->rid, NULL);
+    pObj->pStream = taos_open_stream_withname(pContext->dbConn, pObj->dstTable, pObj->sqlStr, cqProcessStreamRes, INT64_MIN, (void *)pObj->rid, NULL);
 
     // TODO the pObj->pStream may be released if error happens
     if (pObj->pStream) {
-      tscSetStreamDestTable(pObj->pStream, pObj->dstTable);
       pContext->num++;
       cDebug("vgId:%d, id:%d CQ:%s is opened", pContext->vgId, pObj->tid, pObj->sqlStr);
     } else {

src/inc/taosdef.h

@@ -375,6 +375,8 @@ do { \
 
 #define TSDB_MAX_WAL_SIZE    (1024*1024*3)
 
+#define TSDB_ARB_DUMMY_TIME                    4765104000000 // 2121-01-01 00:00:00.000, :P
+
 typedef enum {
   TAOS_QTYPE_RPC   = 0,
   TAOS_QTYPE_FWD   = 1,

src/inc/tfs.h

@@ -31,6 +31,8 @@ typedef struct {
 #define TFS_UNDECIDED_ID -1
 #define TFS_PRIMARY_LEVEL 0
 #define TFS_PRIMARY_ID 0
+#define TFS_MIN_LEVEL 0
+#define TFS_MAX_LEVEL (TSDB_MAX_TIERS - 1)
 
 // FS APIs ====================================
 typedef struct {

src/inc/tsdb.h

@@ -409,6 +409,9 @@ void tsdbDecCommitRef(int vgId);
 int tsdbSyncSend(void *pRepo, SOCKET socketFd);
 int tsdbSyncRecv(void *pRepo, SOCKET socketFd);
 
+// For TSDB Compact
+int tsdbCompact(STsdbRepo *pRepo);
+
 #ifdef __cplusplus
 }
 #endif

src/mnode/src/mnodeDnode.c

@@ -941,7 +941,7 @@ static int32_t mnodeRetrieveDnodes(SShowObj *pShow, char *data, int32_t rows, vo
     cols++;
 
     pWrite = data + pShow->offset[cols] * rows + pShow->bytes[cols] * numOfRows;
-    *(int64_t *)pWrite = 0;
+    *(int64_t *)pWrite = tsArbOnlineTimestamp;
     cols++;
 
     pWrite = data + pShow->offset[cols] * rows + pShow->bytes[cols] * numOfRows;

src/sync/src/syncMain.c

@@ -1150,7 +1150,12 @@ static void syncSetupPeerConnection(SSyncPeer *pPeer) {
     pPeer->peerFd = connFd;
     pPeer->role = TAOS_SYNC_ROLE_UNSYNCED;
     pPeer->pConn = syncAllocateTcpConn(tsTcpPool, pPeer->rid, connFd);
-    if (pPeer->isArb) tsArbOnline = 1;
+    if (pPeer->isArb) {
+      tsArbOnline = 1;
+      if (tsArbOnlineTimestamp == TSDB_ARB_DUMMY_TIME) {
+        tsArbOnlineTimestamp = taosGetTimestampMs();
+      }
+    }
   } else {
     sDebug("%s, failed to setup peer connection to server since %s, try later", pPeer->id, strerror(errno));
     taosCloseSocket(connFd);

src/tsdb/CMakeLists.txt

@@ -6,6 +6,10 @@ AUX_SOURCE_DIRECTORY(src SRC)
 ADD_LIBRARY(tsdb ${SRC})
 TARGET_LINK_LIBRARIES(tsdb tfs common tutil)
 
+IF (TD_TSDB_PLUGINS)
+  TARGET_LINK_LIBRARIES(tsdb tsdbPlugins)
+ENDIF ()
+
 IF (TD_LINUX)
   # Someone has no gtest directory, so comment it
   # ADD_SUBDIRECTORY(tests)

src/tsdb/inc/tsdbCommit.h

@@ -29,10 +29,17 @@ typedef struct {
   int64_t  size;
 } SKVRecord;
 
+#define TSDB_DEFAULT_BLOCK_ROWS(maxRows) ((maxRows)*4 / 5)
+
 void  tsdbGetRtnSnap(STsdbRepo *pRepo, SRtn *pRtn);
 int   tsdbEncodeKVRecord(void **buf, SKVRecord *pRecord);
 void *tsdbDecodeKVRecord(void *buf, SKVRecord *pRecord);
 void *tsdbCommitData(STsdbRepo *pRepo);
+int   tsdbApplyRtnOnFSet(STsdbRepo *pRepo, SDFileSet *pSet, SRtn *pRtn);
+int tsdbWriteBlockInfoImpl(SDFile *pHeadf, STable *pTable, SArray *pSupA, SArray *pSubA, void **ppBuf, SBlockIdx *pIdx);
+int tsdbWriteBlockIdx(SDFile *pHeadf, SArray *pIdxA, void **ppBuf);
+int tsdbWriteBlockImpl(STsdbRepo *pRepo, STable *pTable, SDFile *pDFile, SDataCols *pDataCols, SBlock *pBlock,
+                       bool isLast, bool isSuper, void **ppBuf, void **ppCBuf);
 int   tsdbApplyRtn(STsdbRepo *pRepo);
 
 static FORCE_INLINE int tsdbGetFidLevel(int fid, SRtn *pRtn) {

src/tsdb/inc/tsdbCommitQueue.h

@@ -16,6 +16,8 @@
 #ifndef _TD_TSDB_COMMIT_QUEUE_H_
 #define _TD_TSDB_COMMIT_QUEUE_H_
 
-int tsdbScheduleCommit(STsdbRepo *pRepo);
+typedef enum { COMMIT_REQ, COMPACT_REQ } TSDB_REQ_T;
+
+int tsdbScheduleCommit(STsdbRepo *pRepo, TSDB_REQ_T req);
 
 #endif /* _TD_TSDB_COMMIT_QUEUE_H_ */
\ No newline at end of file

src/tsdb/inc/tsdbCompact.h

+/*
+ * Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
+ *
+ * This program is free software: you can use, redistribute, and/or modify
+ * it under the terms of the GNU Affero General Public License, version 3
+ * or later ("AGPL"), as published by the Free Software Foundation.
+ *
+ * This program is distributed in the hope that it will be useful, but WITHOUT
+ * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
+ * FITNESS FOR A PARTICULAR PURPOSE.
+ *
+ * You should have received a copy of the GNU Affero General Public License
+ * along with this program. If not, see <http://www.gnu.org/licenses/>.
+ */
+#ifndef _TD_TSDB_COMPACT_H_
+#define _TD_TSDB_COMPACT_H_
+
+#ifdef __cplusplus
+extern "C" {
+#endif
+
+void *tsdbCompactImpl(STsdbRepo *pRepo);
+
+#ifdef __cplusplus
+}
+#endif
+
+#endif /* _TD_TSDB_COMPACT_H_ */
\ No newline at end of file

src/tsdb/inc/tsdbint.h

@@ -64,6 +64,8 @@ extern "C" {
 #include "tsdbReadImpl.h"
 // Commit
 #include "tsdbCommit.h"
+// Compact
+#include "tsdbCompact.h"
 // Commit Queue
 #include "tsdbCommitQueue.h"
 // Main definitions

src/tsdb/src/tsdbCommit.c

@@ -51,7 +51,7 @@ typedef struct {
 #define TSDB_COMMIT_LAST_FILE(ch) TSDB_DFILE_IN_SET(TSDB_COMMIT_WRITE_FSET(ch), TSDB_FILE_LAST)
 #define TSDB_COMMIT_BUF(ch) TSDB_READ_BUF(&((ch)->readh))
 #define TSDB_COMMIT_COMP_BUF(ch) TSDB_READ_COMP_BUF(&((ch)->readh))
-#define TSDB_COMMIT_DEFAULT_ROWS(ch) (TSDB_COMMIT_REPO(ch)->config.maxRowsPerFileBlock * 4 / 5)
+#define TSDB_COMMIT_DEFAULT_ROWS(ch) TSDB_DEFAULT_BLOCK_ROWS(TSDB_COMMIT_REPO(ch)->config.maxRowsPerFileBlock)
 #define TSDB_COMMIT_TXN_VERSION(ch) FS_TXN_VERSION(REPO_FS(TSDB_COMMIT_REPO(ch)))
 
 static int  tsdbCommitMeta(STsdbRepo *pRepo);
@@ -72,7 +72,6 @@ static int  tsdbCommitToTable(SCommitH *pCommith, int tid);
 static int  tsdbSetCommitTable(SCommitH *pCommith, STable *pTable);
 static int  tsdbComparKeyBlock(const void *arg1, const void *arg2);
 static int  tsdbWriteBlockInfo(SCommitH *pCommih);
-static int  tsdbWriteBlockIdx(SCommitH *pCommih);
 static int  tsdbCommitMemData(SCommitH *pCommith, SCommitIter *pIter, TSKEY keyLimit, bool toData);
 static int  tsdbMergeMemData(SCommitH *pCommith, SCommitIter *pIter, int bidx);
 static int  tsdbMoveBlock(SCommitH *pCommith, int bidx);
@@ -86,7 +85,6 @@ static void tsdbCloseCommitFile(SCommitH *pCommith, bool hasError);
 static bool tsdbCanAddSubBlock(SCommitH *pCommith, SBlock *pBlock, SMergeInfo *pInfo);
 static void tsdbLoadAndMergeFromCache(SDataCols *pDataCols, int *iter, SCommitIter *pCommitIter, SDataCols *pTarget,
                                       TSKEY maxKey, int maxRows, int8_t update);
-static int  tsdbApplyRtnOnFSet(STsdbRepo *pRepo, SDFileSet *pSet, SRtn *pRtn);
 
 void *tsdbCommitData(STsdbRepo *pRepo) {
   if (pRepo->imem == NULL) {
@@ -117,6 +115,151 @@ _err:
   return NULL;
 }
 
+int tsdbApplyRtnOnFSet(STsdbRepo *pRepo, SDFileSet *pSet, SRtn *pRtn) {
+  SDiskID   did;
+  SDFileSet nSet;
+  STsdbFS * pfs = REPO_FS(pRepo);
+  int       level;
+
+  ASSERT(pSet->fid >= pRtn->minFid);
+
+  level = tsdbGetFidLevel(pSet->fid, pRtn);
+
+  tfsAllocDisk(level, &(did.level), &(did.id));
+  if (did.level == TFS_UNDECIDED_LEVEL) {
+    terrno = TSDB_CODE_TDB_NO_AVAIL_DISK;
+    return -1;
+  }
+
+  if (did.level > TSDB_FSET_LEVEL(pSet)) {
+    // Need to move the FSET to higher level
+    tsdbInitDFileSet(&nSet, did, REPO_ID(pRepo), pSet->fid, FS_TXN_VERSION(pfs));
+
+    if (tsdbCopyDFileSet(pSet, &nSet) < 0) {
+      tsdbError("vgId:%d failed to copy FSET %d from level %d to level %d since %s", REPO_ID(pRepo), pSet->fid,
+                TSDB_FSET_LEVEL(pSet), did.level, tstrerror(terrno));
+      return -1;
+    }
+
+    if (tsdbUpdateDFileSet(pfs, &nSet) < 0) {
+      return -1;
+    }
+
+    tsdbInfo("vgId:%d FSET %d is copied from level %d disk id %d to level %d disk id %d", REPO_ID(pRepo), pSet->fid,
+             TSDB_FSET_LEVEL(pSet), TSDB_FSET_ID(pSet), did.level, did.id);
+  } else {
+    // On a correct level
+    if (tsdbUpdateDFileSet(pfs, pSet) < 0) {
+      return -1;
+    }
+  }
+
+  return 0;
+}
+
+int tsdbWriteBlockInfoImpl(SDFile *pHeadf, STable *pTable, SArray *pSupA, SArray *pSubA, void **ppBuf,
+                           SBlockIdx *pIdx) {
+  size_t      nSupBlocks;
+  size_t      nSubBlocks;
+  uint32_t    tlen;
+  SBlockInfo *pBlkInfo;
+  int64_t     offset;
+  SBlock *    pBlock;
+
+  memset(pIdx, 0, sizeof(*pIdx));
+
+  nSupBlocks = taosArrayGetSize(pSupA);
+  nSubBlocks = (pSubA == NULL) ? 0 : taosArrayGetSize(pSubA);
+
+  if (nSupBlocks <= 0) {
+    // No data (data all deleted)
+    return 0;
+  }
+
+  tlen = (uint32_t)(sizeof(SBlockInfo) + sizeof(SBlock) * (nSupBlocks + nSubBlocks) + sizeof(TSCKSUM));
+  if (tsdbMakeRoom(ppBuf, tlen) < 0) return -1;
+  pBlkInfo = *ppBuf;
+
+  pBlkInfo->delimiter = TSDB_FILE_DELIMITER;
+  pBlkInfo->tid = TABLE_TID(pTable);
+  pBlkInfo->uid = TABLE_UID(pTable);
+
+  memcpy((void *)(pBlkInfo->blocks), taosArrayGet(pSupA, 0), nSupBlocks * sizeof(SBlock));
+  if (nSubBlocks > 0) {
+    memcpy((void *)(pBlkInfo->blocks + nSupBlocks), taosArrayGet(pSubA, 0), nSubBlocks * sizeof(SBlock));
+
+    for (int i = 0; i < nSupBlocks; i++) {
+      pBlock = pBlkInfo->blocks + i;
+
+      if (pBlock->numOfSubBlocks > 1) {
+        pBlock->offset += (sizeof(SBlockInfo) + sizeof(SBlock) * nSupBlocks);
+      }
+    }
+  }
+
+  taosCalcChecksumAppend(0, (uint8_t *)pBlkInfo, tlen);
+
+  if (tsdbAppendDFile(pHeadf, (void *)pBlkInfo, tlen, &offset) < 0) {
+    return -1;
+  }
+
+  tsdbUpdateDFileMagic(pHeadf, POINTER_SHIFT(pBlkInfo, tlen - sizeof(TSCKSUM)));
+
+  // Set pIdx
+  pBlock = taosArrayGetLast(pSupA);
+
+  pIdx->tid = TABLE_TID(pTable);
+  pIdx->uid = TABLE_UID(pTable);
+  pIdx->hasLast = pBlock->last ? 1 : 0;
+  pIdx->maxKey = pBlock->keyLast;
+  pIdx->numOfBlocks = (uint32_t)nSupBlocks;
+  pIdx->len = tlen;
+  pIdx->offset = (uint32_t)offset;
+
+  return 0;
+}
+
+int tsdbWriteBlockIdx(SDFile *pHeadf, SArray *pIdxA, void **ppBuf) {
+  SBlockIdx *pBlkIdx;
+  size_t     nidx = taosArrayGetSize(pIdxA);
+  int        tlen = 0, size;
+  int64_t    offset;
+
+  if (nidx <= 0) {
+    // All data are deleted
+    pHeadf->info.offset = 0;
+    pHeadf->info.len = 0;
+    return 0;
+  }
+
+  for (size_t i = 0; i < nidx; i++) {
+    pBlkIdx = (SBlockIdx *)taosArrayGet(pIdxA, i);
+
+    size = tsdbEncodeSBlockIdx(NULL, pBlkIdx);
+    if (tsdbMakeRoom(ppBuf, tlen + size) < 0) return -1;
+
+    void *ptr = POINTER_SHIFT(*ppBuf, tlen);
+    tsdbEncodeSBlockIdx(&ptr, pBlkIdx);
+
+    tlen += size;
+  }
+
+  tlen += sizeof(TSCKSUM);
+  if (tsdbMakeRoom(ppBuf, tlen) < 0) return -1;
+  taosCalcChecksumAppend(0, (uint8_t *)(*ppBuf), tlen);
+
+  if (tsdbAppendDFile(pHeadf, *ppBuf, tlen, &offset) < tlen) {
+    return -1;
+  }
+
+  tsdbUpdateDFileMagic(pHeadf, POINTER_SHIFT(*ppBuf, tlen - sizeof(TSCKSUM)));
+  pHeadf->info.offset = (uint32_t)offset;
+  pHeadf->info.len = tlen;
+
+  return 0;
+}
+
+
 // =================== Commit Meta Data
 static int tsdbCommitMeta(STsdbRepo *pRepo) {
   STsdbFS *  pfs = REPO_FS(pRepo);
@@ -446,7 +589,8 @@ static int tsdbCommitToFile(SCommitH *pCommith, SDFileSet *pSet, int fid) {
     }
   }
 
-  if (tsdbWriteBlockIdx(pCommith) < 0) {
+  if (tsdbWriteBlockIdx(TSDB_COMMIT_HEAD_FILE(pCommith), pCommith->aBlkIdx, (void **)(&(TSDB_COMMIT_BUF(pCommith)))) <
+      0) {
     tsdbError("vgId:%d failed to write SBlockIdx part to FSET %d since %s", REPO_ID(pRepo), fid, tstrerror(terrno));
     tsdbCloseCommitFile(pCommith, true);
     // revert the file change
@@ -754,23 +898,21 @@ static int tsdbComparKeyBlock(const void *arg1, const void *arg2) {
   }
 }
 
-static int tsdbWriteBlock(SCommitH *pCommith, SDFile *pDFile, SDataCols *pDataCols, SBlock *pBlock, bool isLast,
-                          bool isSuper) {
-  STsdbRepo * pRepo = TSDB_COMMIT_REPO(pCommith);
+int tsdbWriteBlockImpl(STsdbRepo *pRepo, STable *pTable, SDFile *pDFile, SDataCols *pDataCols, SBlock *pBlock,
+                       bool isLast, bool isSuper, void **ppBuf, void **ppCBuf) {
   STsdbCfg *  pCfg = REPO_CFG(pRepo);
   SBlockData *pBlockData;
   int64_t     offset = 0;
-  STable *    pTable = TSDB_COMMIT_TABLE(pCommith);
   int         rowsToWrite = pDataCols->numOfRows;
 
   ASSERT(rowsToWrite > 0 && rowsToWrite <= pCfg->maxRowsPerFileBlock);
   ASSERT((!isLast) || rowsToWrite < pCfg->minRowsPerFileBlock);
 
   // Make buffer space
-  if (tsdbMakeRoom((void **)(&TSDB_COMMIT_BUF(pCommith)), TSDB_BLOCK_STATIS_SIZE(pDataCols->numOfCols)) < 0) {
+  if (tsdbMakeRoom(ppBuf, TSDB_BLOCK_STATIS_SIZE(pDataCols->numOfCols)) < 0) {
     return -1;
   }
-  pBlockData = (SBlockData *)TSDB_COMMIT_BUF(pCommith);
+  pBlockData = (SBlockData *)(*ppBuf);
 
   // Get # of cols not all NULL(not including key column)
   int nColsNotAllNull = 0;
@@ -816,23 +958,23 @@ static int tsdbWriteBlock(SCommitH *pCommith, SDFile *pDFile, SDataCols *pDataCo
     void *  tptr;
 
     // Make room
-    if (tsdbMakeRoom((void **)(&TSDB_COMMIT_BUF(pCommith)), lsize + tlen + COMP_OVERFLOW_BYTES + sizeof(TSCKSUM)) < 0) {
+    if (tsdbMakeRoom(ppBuf, lsize + tlen + COMP_OVERFLOW_BYTES + sizeof(TSCKSUM)) < 0) {
       return -1;
     }
-    pBlockData = (SBlockData *)TSDB_COMMIT_BUF(pCommith);
+    pBlockData = (SBlockData *)(*ppBuf);
     pBlockCol = pBlockData->cols + tcol;
     tptr = POINTER_SHIFT(pBlockData, lsize);
 
     if (pCfg->compression == TWO_STAGE_COMP &&
-        tsdbMakeRoom((void **)(&TSDB_COMMIT_COMP_BUF(pCommith)), tlen + COMP_OVERFLOW_BYTES) < 0) {
+        tsdbMakeRoom(ppCBuf, tlen + COMP_OVERFLOW_BYTES) < 0) {
       return -1;
     }
 
     // Compress or just copy
     if (pCfg->compression) {
       flen = (*(tDataTypes[pDataCol->type].compFunc))((char *)pDataCol->pData, tlen, rowsToWrite, tptr,
-                                                      tlen + COMP_OVERFLOW_BYTES, pCfg->compression,
-                                                      TSDB_COMMIT_COMP_BUF(pCommith), tlen + COMP_OVERFLOW_BYTES);
+                                                      tlen + COMP_OVERFLOW_BYTES, pCfg->compression, *ppCBuf,
+                                                      tlen + COMP_OVERFLOW_BYTES);
     } else {
       flen = tlen;
       memcpy(tptr, pDataCol->pData, flen);
@@ -888,117 +1030,33 @@ static int tsdbWriteBlock(SCommitH *pCommith, SDFile *pDFile, SDataCols *pDataCo
   return 0;
 }
 
-static int tsdbWriteBlockInfo(SCommitH *pCommih) {
-  SDFile *    pHeadf = TSDB_COMMIT_HEAD_FILE(pCommih);
-  SBlockIdx   blkIdx;
-  STable *    pTable = TSDB_COMMIT_TABLE(pCommih);
-  SBlock *    pBlock;
-  size_t      nSupBlocks;
-  size_t      nSubBlocks;
-  uint32_t    tlen;
-  SBlockInfo *pBlkInfo;
-  int64_t     offset;
-
-  nSupBlocks = taosArrayGetSize(pCommih->aSupBlk);
-  nSubBlocks = taosArrayGetSize(pCommih->aSubBlk);
-
-  if (nSupBlocks <= 0) {
-    // No data (data all deleted)
-    return 0;
-  }
-
-  tlen = (uint32_t)(sizeof(SBlockInfo) + sizeof(SBlock) * (nSupBlocks + nSubBlocks) + sizeof(TSCKSUM));
-
-  // Write SBlockInfo part
-  if (tsdbMakeRoom((void **)(&(TSDB_COMMIT_BUF(pCommih))), tlen) < 0) return -1;
-  pBlkInfo = TSDB_COMMIT_BUF(pCommih);
-
-  pBlkInfo->delimiter = TSDB_FILE_DELIMITER;
-  pBlkInfo->tid = TABLE_TID(pTable);
-  pBlkInfo->uid = TABLE_UID(pTable);
-
-  memcpy((void *)(pBlkInfo->blocks), taosArrayGet(pCommih->aSupBlk, 0), nSupBlocks * sizeof(SBlock));
-  if (nSubBlocks > 0) {
-    memcpy((void *)(pBlkInfo->blocks + nSupBlocks), taosArrayGet(pCommih->aSubBlk, 0), nSubBlocks * sizeof(SBlock));
-
-    for (int i = 0; i < nSupBlocks; i++) {
-      pBlock = pBlkInfo->blocks + i;
-
-      if (pBlock->numOfSubBlocks > 1) {
-        pBlock->offset += (sizeof(SBlockInfo) + sizeof(SBlock) * nSupBlocks);
-      }
-    }
-  }
-
-  taosCalcChecksumAppend(0, (uint8_t *)pBlkInfo, tlen);
-
-  if (tsdbAppendDFile(pHeadf, TSDB_COMMIT_BUF(pCommih), tlen, &offset) < 0) {
-    return -1;
-  }
-
-  tsdbUpdateDFileMagic(pHeadf, POINTER_SHIFT(pBlkInfo, tlen - sizeof(TSCKSUM)));
-
-  // Set blkIdx
-  pBlock = taosArrayGet(pCommih->aSupBlk, nSupBlocks - 1);
+static int tsdbWriteBlock(SCommitH *pCommith, SDFile *pDFile, SDataCols *pDataCols, SBlock *pBlock, bool isLast,
+                          bool isSuper) {
+  return tsdbWriteBlockImpl(TSDB_COMMIT_REPO(pCommith), TSDB_COMMIT_TABLE(pCommith), pDFile, pDataCols, pBlock, isLast,
+                            isSuper, (void **)(&(TSDB_COMMIT_BUF(pCommith))),
+                            (void **)(&(TSDB_COMMIT_COMP_BUF(pCommith))));
+}
 
-  blkIdx.tid = TABLE_TID(pTable);
-  blkIdx.uid = TABLE_UID(pTable);
-  blkIdx.hasLast = pBlock->last ? 1 : 0;
-  blkIdx.maxKey = pBlock->keyLast;
-  blkIdx.numOfBlocks = (uint32_t)nSupBlocks;
-  blkIdx.len = tlen;
-  blkIdx.offset = (uint32_t)offset;
 
-  ASSERT(blkIdx.numOfBlocks > 0);
+static int tsdbWriteBlockInfo(SCommitH *pCommih) {
+  SDFile *  pHeadf = TSDB_COMMIT_HEAD_FILE(pCommih);
+  SBlockIdx blkIdx;
+  STable *  pTable = TSDB_COMMIT_TABLE(pCommih);
 
-  if (taosArrayPush(pCommih->aBlkIdx, (void *)(&blkIdx)) == NULL) {
-    terrno = TSDB_CODE_TDB_OUT_OF_MEMORY;
+  if (tsdbWriteBlockInfoImpl(pHeadf, pTable, pCommih->aSupBlk, pCommih->aSubBlk, (void **)(&(TSDB_COMMIT_BUF(pCommih))),
+                             &blkIdx) < 0) {
     return -1;
   }
 
-  return 0;
-}
-
-static int tsdbWriteBlockIdx(SCommitH *pCommih) {
-  SBlockIdx *pBlkIdx = NULL;
-  SDFile *   pHeadf = TSDB_COMMIT_HEAD_FILE(pCommih);
-  size_t     nidx = taosArrayGetSize(pCommih->aBlkIdx);
-  int        tlen = 0, size = 0;
-  int64_t    offset = 0;
-
-  if (nidx <= 0) {
-    // All data are deleted
-    pHeadf->info.offset = 0;
-    pHeadf->info.len = 0;
+  if (blkIdx.numOfBlocks == 0) {
     return 0;
   }
 
-  for (size_t i = 0; i < nidx; i++) {
-    pBlkIdx = (SBlockIdx *)taosArrayGet(pCommih->aBlkIdx, i);
-
-    size = tsdbEncodeSBlockIdx(NULL, pBlkIdx);
-    if (tsdbMakeRoom((void **)(&TSDB_COMMIT_BUF(pCommih)), tlen + size) < 0) return -1;
-
-    void *ptr = POINTER_SHIFT(TSDB_COMMIT_BUF(pCommih), tlen);
-    tsdbEncodeSBlockIdx(&ptr, pBlkIdx);
-
-    tlen += size;
-  }
-
-  tlen += sizeof(TSCKSUM);
-  if (tsdbMakeRoom((void **)(&TSDB_COMMIT_BUF(pCommih)), tlen) < 0) return -1;
-  taosCalcChecksumAppend(0, (uint8_t *)TSDB_COMMIT_BUF(pCommih), tlen);
-
-  if (tsdbAppendDFile(pHeadf, TSDB_COMMIT_BUF(pCommih), tlen, &offset) < tlen) {
-    tsdbError("vgId:%d failed to write block index part to file %s since %s", TSDB_COMMIT_REPO_ID(pCommih),
-              TSDB_FILE_FULL_NAME(pHeadf), tstrerror(terrno));
+  if (taosArrayPush(pCommih->aBlkIdx, (void *)(&blkIdx)) == NULL) {
+    terrno = TSDB_CODE_TDB_OUT_OF_MEMORY;
     return -1;
   }
 
-  tsdbUpdateDFileMagic(pHeadf, POINTER_SHIFT(TSDB_COMMIT_BUF(pCommih), tlen - sizeof(TSCKSUM)));
-  pHeadf->info.offset = (uint32_t)offset;
-  pHeadf->info.len = tlen;
-
   return 0;
 }
 
@@ -1454,45 +1512,3 @@ int tsdbApplyRtn(STsdbRepo *pRepo) {
 
   return 0;
 }
-
-static int tsdbApplyRtnOnFSet(STsdbRepo *pRepo, SDFileSet *pSet, SRtn *pRtn) {
-  SDiskID   did;
-  SDFileSet nSet;
-  STsdbFS * pfs = REPO_FS(pRepo);
-  int       level;
-
-  ASSERT(pSet->fid >= pRtn->minFid);
-
-  level = tsdbGetFidLevel(pSet->fid, pRtn);
-
-  tfsAllocDisk(level, &(did.level), &(did.id));
-  if (did.level == TFS_UNDECIDED_LEVEL) {
-    terrno = TSDB_CODE_TDB_NO_AVAIL_DISK;
-    return -1;
-  }
-
-  if (did.level > TSDB_FSET_LEVEL(pSet)) {
-    // Need to move the FSET to higher level
-    tsdbInitDFileSet(&nSet, did, REPO_ID(pRepo), pSet->fid, FS_TXN_VERSION(pfs));
-
-    if (tsdbCopyDFileSet(pSet, &nSet) < 0) {
-      tsdbError("vgId:%d failed to copy FSET %d from level %d to level %d since %s", REPO_ID(pRepo), pSet->fid,
-                TSDB_FSET_LEVEL(pSet), did.level, tstrerror(terrno));
-      return -1;
-    }
-
-    if (tsdbUpdateDFileSet(pfs, &nSet) < 0) {
-      return -1;
-    }
-
-    tsdbInfo("vgId:%d FSET %d is copied from level %d disk id %d to level %d disk id %d", REPO_ID(pRepo), pSet->fid,
-             TSDB_FSET_LEVEL(pSet), TSDB_FSET_ID(pSet), did.level, did.id);
-  } else {
-    // On a correct level
-    if (tsdbUpdateDFileSet(pfs, pSet) < 0) {
-      return -1;
-    }
-  }
-
-  return 0;
-}
\ No newline at end of file

src/tsdb/src/tsdbCommitQueue.c

@@ -26,8 +26,9 @@ typedef struct {
 } SCommitQueue;
 
 typedef struct {
+  TSDB_REQ_T req;
   STsdbRepo *pRepo;
-} SCommitReq;
+} SReq;
 
 static void *tsdbLoopCommit(void *arg);
 
@@ -90,16 +91,17 @@ void tsdbDestroyCommitQueue() {
   pthread_mutex_destroy(&(pQueue->lock));
 }
 
-int tsdbScheduleCommit(STsdbRepo *pRepo) {
+int tsdbScheduleCommit(STsdbRepo *pRepo, TSDB_REQ_T req) {
   SCommitQueue *pQueue = &tsCommitQueue;
 
-  SListNode *pNode = (SListNode *)calloc(1, sizeof(SListNode) + sizeof(SCommitReq));
+  SListNode *pNode = (SListNode *)calloc(1, sizeof(SListNode) + sizeof(SReq));
   if (pNode == NULL) {
     terrno = TSDB_CODE_TDB_OUT_OF_MEMORY;
     return -1;
   }
 
-  ((SCommitReq *)pNode->data)->pRepo = pRepo;
+  ((SReq *)pNode->data)->req = req;
+  ((SReq *)pNode->data)->pRepo = pRepo;
 
   pthread_mutex_lock(&(pQueue->lock));
 
@@ -154,6 +156,7 @@ static void *tsdbLoopCommit(void *arg) {
   SCommitQueue *pQueue = &tsCommitQueue;
   SListNode *   pNode = NULL;
   STsdbRepo *   pRepo = NULL;
+  TSDB_REQ_T    req;
 
   while (true) {
     pthread_mutex_lock(&(pQueue->lock));
@@ -174,14 +177,22 @@ static void *tsdbLoopCommit(void *arg) {
 
     pthread_mutex_unlock(&(pQueue->lock));
 
-    pRepo = ((SCommitReq *)pNode->data)->pRepo;
+    req = ((SReq *)pNode->data)->req;
+    pRepo = ((SReq *)pNode->data)->pRepo;
 
     // check if need to apply new config
-    if (pRepo->config_changed) {             
+    if (pRepo->config_changed) {
       tsdbApplyRepoConfig(pRepo);
     }
 
-    tsdbCommitData(pRepo);
+    if (req == COMMIT_REQ) {
+      tsdbCommitData(pRepo);
+    } else if (req == COMPACT_REQ) {
+      tsdbCompactImpl(pRepo);
+    } else {
+      ASSERT(0);
+    }
+
     listNodeFree(pNode);
   }
 

src/tsdb/src/tsdbCompact.c

@@ -11,4 +11,12 @@
  *
  * You should have received a copy of the GNU Affero General Public License
  * along with this program. If not, see <http://www.gnu.org/licenses/>.
- */
\ No newline at end of file
+ */
+#include "tsdb.h"
+
+#ifndef _TSDB_PLUGINS
+
+int   tsdbCompact(STsdbRepo *pRepo) { return 0; }
+void *tsdbCompactImpl(STsdbRepo *pRepo) { return NULL; }
+
+#endif
\ No newline at end of file

src/tsdb/src/tsdbMemTable.c

@@ -288,7 +288,7 @@ int tsdbAsyncCommit(STsdbRepo *pRepo) {
   if (tsdbLockRepo(pRepo) < 0) return -1;
   pRepo->imem = pRepo->mem;
   pRepo->mem = NULL;
-  tsdbScheduleCommit(pRepo);
+  tsdbScheduleCommit(pRepo, COMMIT_REQ);
   if (tsdbUnlockRepo(pRepo) < 0) return -1;
 
   return 0;

src/tsdb/src/tsdbReadImpl.c

@@ -258,7 +258,7 @@ int tsdbLoadBlockData(SReadH *pReadh, SBlock *pBlock, SBlockInfo *pBlkInfo) {
   for (int i = 1; i < pBlock->numOfSubBlocks; i++) {
     iBlock++;
     if (tsdbLoadBlockDataImpl(pReadh, iBlock, pReadh->pDCols[1]) < 0) return -1;
-    if (tdMergeDataCols(pReadh->pDCols[0], pReadh->pDCols[1], pReadh->pDCols[1]->numOfRows) < 0) return -1;
+    if (tdMergeDataCols(pReadh->pDCols[0], pReadh->pDCols[1], pReadh->pDCols[1]->numOfRows, NULL) < 0) return -1;
   }
 
   ASSERT(pReadh->pDCols[0]->numOfRows == pBlock->numOfRows);
@@ -284,7 +284,7 @@ int tsdbLoadBlockDataCols(SReadH *pReadh, SBlock *pBlock, SBlockInfo *pBlkInfo,
   for (int i = 1; i < pBlock->numOfSubBlocks; i++) {
     iBlock++;
     if (tsdbLoadBlockDataColsImpl(pReadh, iBlock, pReadh->pDCols[1], colIds, numOfColsIds) < 0) return -1;
-    if (tdMergeDataCols(pReadh->pDCols[0], pReadh->pDCols[1], pReadh->pDCols[1]->numOfRows) < 0) return -1;
+    if (tdMergeDataCols(pReadh->pDCols[0], pReadh->pDCols[1], pReadh->pDCols[1]->numOfRows, NULL) < 0) return -1;
   }
 
   ASSERT(pReadh->pDCols[0]->numOfRows == pBlock->numOfRows);

tests/pytest/util/dnodes.py

@@ -432,7 +432,7 @@ class TDDnodes:
         self.simDeployed = False
 
     def init(self, path):
-        psCmd = "ps -ef|grep -w taosd| grep -v grep | awk '{print $2}'"
+        psCmd = "ps -ef|grep -w taosd| grep -v grep| grep -v defunct | awk '{print $2}'"
         processID = subprocess.check_output(psCmd, shell=True).decode("utf-8")
         while(processID):
             killCmd = "kill -TERM %s > /dev/null 2>&1" % processID
@@ -545,14 +545,14 @@ class TDDnodes:
         for i in range(len(self.dnodes)):
             self.dnodes[i].stop()
 
-        psCmd = "ps -ef | grep -w taosd | grep 'root' | grep -v grep | awk '{print $2}'"
+        psCmd = "ps -ef | grep -w taosd | grep 'root' | grep -v grep| grep -v defunct | awk '{print $2}'"
         processID = subprocess.check_output(psCmd, shell=True).decode("utf-8")
         if processID:
             cmd = "sudo systemctl stop taosd"
             os.system(cmd)
         # if os.system(cmd) != 0 :
         # tdLog.exit(cmd)
-        psCmd = "ps -ef|grep -w taosd| grep -v grep | awk '{print $2}'"
+        psCmd = "ps -ef|grep -w taosd| grep -v grep| grep -v defunct | awk '{print $2}'"
         processID = subprocess.check_output(psCmd, shell=True).decode("utf-8")
         while(processID):
             killCmd = "kill -TERM %s > /dev/null 2>&1" % processID