Merge branch 'develop' into feature/TD-3188

CMakeLists.txt

@@ -17,6 +17,7 @@ SET(TD_MQTT FALSE)
 SET(TD_TSDB_PLUGINS FALSE)
 SET(TD_STORAGE FALSE)
 SET(TD_TOPIC FALSE)
+SET(TD_MODULE FALSE)
 
 SET(TD_COVER FALSE)
 SET(TD_MEM_CHECK FALSE)

Jenkinsfile

@@ -6,6 +6,7 @@ node {
 }
 
 def skipstage=0
+
 def abortPreviousBuilds() {
   def currentJobName = env.JOB_NAME
   def currentBuildNumber = env.BUILD_NUMBER.toInteger()
@@ -24,7 +25,7 @@ def abortPreviousBuilds() {
     build.doKill()    //doTerm(),doKill(),doTerm()
   }
 }
-//abort previous build
+//  abort previous build
 abortPreviousBuilds()
 def abort_previous(){
   def buildNumber = env.BUILD_NUMBER as int
@@ -33,18 +34,19 @@ def abort_previous(){
 }
 def pre_test(){
     
+    
     sh '''
     sudo rmtaos || echo "taosd has not installed"
     '''
     sh '''
-    
+    killall -9 taosd ||echo "no taosd running"
+    killall -9 gdb || echo "no gdb running"
     cd ${WKC}
     git checkout develop
     git reset --hard HEAD~10 >/dev/null 
     git pull >/dev/null
     git fetch origin +refs/pull/${CHANGE_ID}/merge
     git checkout -qf FETCH_HEAD
-    git --no-pager diff --name-only FETCH_HEAD $(git merge-base FETCH_HEAD develop)|grep -v -E '.*md|//src//connector|Jenkinsfile'
     find ${WKC}/tests/pytest -name \'*\'.sql -exec rm -rf {} \\;
     cd ${WK}
     git reset --hard HEAD~10
@@ -79,6 +81,10 @@ pipeline {
               changeRequest()
           }
           steps {
+            script{
+              abort_previous()
+              abortPreviousBuilds()
+            }
           sh'''
           cp -r ${WORKSPACE} ${WORKSPACE}.tes
           cd ${WORKSPACE}.tes
@@ -179,6 +185,14 @@ pipeline {
             rm -rf /var/log/taos/*
             ./handle_crash_gen_val_log.sh
             '''
+            catchError(buildResult: 'SUCCESS', stageResult: 'FAILURE') {
+                sh '''
+                cd ${WKC}/tests/pytest
+                rm -rf /var/lib/taos/*
+                rm -rf /var/log/taos/*
+                ./handle_taosd_val_log.sh
+                '''
+            }
             timeout(time: 45, unit: 'MINUTES'){
                 sh '''
                 date
@@ -209,6 +223,11 @@ pipeline {
               cd ${WKC}/tests
               ./test-all.sh b3fq
               date'''
+              sh '''
+              date
+              cd ${WKC}/tests
+              ./test-all.sh full example
+              date'''
             }
           }
         }

README-CN.md

@@ -258,10 +258,16 @@ TDengine 社区生态中也有一些非常友好的第三方连接器，可以
 
 TDengine 的测试框架和所有测试例全部开源。
 
-点击[这里](tests/How-To-Run-Test-And-How-To-Add-New-Test-Case.md)，了解如何运行测试例和添加新的测试例。
+点击 [这里](tests/How-To-Run-Test-And-How-To-Add-New-Test-Case.md)，了解如何运行测试例和添加新的测试例。
 
 # 成为社区贡献者
-点击[这里](https://www.taosdata.com/cn/contributor/)，了解如何成为 TDengine 的贡献者。
 
-#加入技术交流群
-TDengine官方社群「物联网大数据群」对外开放，欢迎您加入讨论。搜索微信号 "tdengine"，加小T为好友，即可入群。
+点击 [这里](https://www.taosdata.com/cn/contributor/)，了解如何成为 TDengine 的贡献者。
+
+# 加入技术交流群
+
+TDengine 官方社群「物联网大数据群」对外开放，欢迎您加入讨论。搜索微信号 "tdengine"，加小T为好友，即可入群。
+
+# [谁在使用TDengine](https://github.com/taosdata/TDengine/issues/2432)
+
+欢迎所有 TDengine 用户及贡献者在 [这里](https://github.com/taosdata/TDengine/issues/2432) 分享您在当前工作中开发/使用 TDengine 的故事。 

README.md

@@ -250,3 +250,6 @@ Please follow the [contribution guidelines](CONTRIBUTING.md) to contribute to th
 
 Add WeChat “tdengine” to join the group，you can communicate with other users.
 
+# [User List](https://github.com/taosdata/TDengine/issues/2432)
+
+If you are using TDengine and feel it helps or you'd like to do some contributions, please add your company to [user list](https://github.com/taosdata/TDengine/issues/2432) and let us know your needs.

cmake/define.inc

@@ -29,6 +29,10 @@ IF (TD_TOPIC)
   ADD_DEFINITIONS(-D_TOPIC)
 ENDIF ()
 
+IF (TD_MODULE)
+  ADD_DEFINITIONS(-D_MODULE)
+ENDIF ()
+
 IF (TD_GODLL)
   ADD_DEFINITIONS(-D_TD_GO_DLL_)
 ENDIF ()

cmake/input.inc

@@ -17,6 +17,14 @@ ELSEIF (${TOPIC} MATCHES "false")
   MESSAGE(STATUS "Build without topic plugins")
 ENDIF ()
 
+IF (${TD_MODULE} MATCHES "true")
+  SET(TD_MODULE TRUE)
+  MESSAGE(STATUS "Build with module plugins")
+ELSEIF (${TOPIC} MATCHES "false")
+  SET(TD_MODULE FALSE)
+  MESSAGE(STATUS "Build without module plugins")
+ENDIF ()
+
 IF (${COVER} MATCHES "true")
   SET(TD_COVER TRUE)
   MESSAGE(STATUS "Build with test coverage")

documentation20/cn/00.index/docs.md

@@ -120,7 +120,7 @@ TDengine是一个高效的存储、查询、分析时序大数据的平台，专
 * [TDengine性能对比测试工具](https://www.taosdata.com/blog/2020/01/18/1166.html)
 * [IDEA数据库管理工具可视化使用TDengine](https://www.taosdata.com/blog/2020/08/27/1767.html)
 * [基于eletron开发的跨平台TDengine图形化管理工具](https://github.com/skye0207/TDengineGUI)
-* [DataX，支持TDengine的离线数据采集/同步工具](https://github.com/alibaba/DataX)
+* [DataX，支持TDengine的离线数据采集/同步工具](https://github.com/wgzhao/DataX)（文档：[读取插件](https://github.com/wgzhao/DataX/blob/master/docs/src/main/sphinx/reader/tdenginereader.md)、[写入插件](https://github.com/wgzhao/DataX/blob/master/docs/src/main/sphinx/writer/tdenginewriter.md)）
 
 ## TDengine与其他数据库的对比测试
 

documentation20/cn/02.getting-started/docs.md

@@ -179,18 +179,18 @@ taos> select avg(f1), max(f2), min(f3) from test.t10 interval(10s);
 
 ### TDengine服务器支持的平台列表
 
-|                | **CentOS**  **6/7/8** | **Ubuntu**  **16/18/20** | **Other Linux** | **统信****UOS** | **银河****/****中标麒麟** | **凝思**  **V60/V80** |
+|                | **CentOS 6/7/8** | **Ubuntu 16/18/20** | **Other Linux** | **统信 UOS** | **银河/中标麒麟** | **凝思 V60/V80** |
 | -------------- | --------------------- | ------------------------ | --------------- | --------------- | ------------------------- | --------------------- |
 | X64            | ●                     | ●                        |                 | ○               | ●                         | ●                     |
-| 树莓派ARM32    |                       | ●                        | ●               |                 |                           |                       |
-| 龙芯MIPS64     |                       |                          | ●               |                 |                           |                       |
+| 树莓派 ARM32    |                       | ●                        | ●               |                 |                           |                       |
+| 龙芯 MIPS64     |                       |                          | ●               |                 |                           |                       |
 | 鲲鹏 ARM64    |                       | ○                        | ○               |                 | ●                         |                       |
 | 申威 Alpha64  |                       |                          | ○               | ●               |                           |                       |
-| 飞腾ARM64      |                       | ○优麒麟                  |                 |                 |                           |                       |
-| 海光X64        | ●                     | ●                        | ●               | ○               | ●                         | ●                     |
-| 瑞芯微ARM64/32 |                       |                          | ○               |                 |                           |                       |
-| 全志ARM64/32   |                       |                          | ○               |                 |                           |                       |
-| 炬力ARM64/32   |                       |                          | ○               |                 |                           |                       |
+| 飞腾 ARM64      |                       | ○ 优麒麟                  |                 |                 |                           |                       |
+| 海光 X64        | ●                     | ●                        | ●               | ○               | ●                         | ●                     |
+| 瑞芯微 ARM64/32 |                       |                          | ○               |                 |                           |                       |
+| 全志 ARM64/32   |                       |                          | ○               |                 |                           |                       |
+| 炬力 ARM64/32   |                       |                          | ○               |                 |                           |                       |
 | TI ARM32       |                       |                          | ○               |                 |                           |                       |
 
 注： ● 表示经过官方测试验证， ○ 表示非官方测试验证。
@@ -203,7 +203,7 @@ taos> select avg(f1), max(f2), min(f3) from test.t10 interval(10s);
 
 对照矩阵如下：
 
-| **CPU**     | **X64   64bit** |           |           | **X86   32bit** | **ARM64** | **ARM32** | **MIPS **  **龙芯** | **Alpha **  **申威** | **X64 **  **海光** |
+| **CPU**     | **X64 64bit** |           |           | **X86 32bit** | **ARM64** | **ARM32** | **MIPS 龙芯** | **Alpha 申威** | **X64 海光** |
 | ----------- | --------------- | --------- | --------- | --------------- | --------- | --------- | ------------------- | -------------------- | ------------------ |
 | **OS**      | **Linux**       | **Win64** | **Win32** | **Win32**       | **Linux** | **Linux** | **Linux**           | **Linux**            | **Linux**          |
 | **C/C++**   | ●               | ●         | ●         | ○               | ●         | ●         | ●                   | ●                    | ●                  |

documentation20/cn/08.connector/01.java/docs.md

@@ -471,9 +471,11 @@ TDengine 目前支持时间戳、数字、字符、布尔类型，与 Java 对
 | BIGINT            | java.lang.Long     |
 | FLOAT             | java.lang.Float    |
 | DOUBLE            | java.lang.Double   |
-| SMALLINT, TINYINT | java.lang.Short    |
+| SMALLINT	    | java.lang.Short    |
+| TINYINT           | java.lang.Byte     |
 | BOOL              | java.lang.Boolean  |
-| BINARY, NCHAR     | java.lang.String   |
+| BINARY            | byte array         |
+| NCHAR             | java.lang.String   |
 
 
 

documentation20/cn/10.cluster/docs.md

@@ -111,9 +111,10 @@ taos>
 
 **提示：**
 
-- 任何已经加入集群在线的数据节点，都可以作为后续待加入节点的firstEP。
-- firstEp这个参数仅仅在该数据节点首次加入集群时有作用，加入集群后，该数据节点会保存最新的mnode的End Point列表，不再依赖这个参数。
-- 两个没有配置firstEp参数的数据节点dnode启动后，会独立运行起来。这个时候，无法将其中一个数据节点加入到另外一个数据节点，形成集群。**无法将两个独立的集群合并成为新的集群**。
+- 任何已经加入集群在线的数据节点，都可以作为后续待加入节点的 firstEP。
+- firstEp 这个参数仅仅在该数据节点首次加入集群时有作用，加入集群后，该数据节点会保存最新的 mnode 的 End Point 列表，不再依赖这个参数。
+  - 接下来，配置文件中的 firstEp 参数就主要在客户端连接的时候使用了，例如 taos shell 如果不加参数，会默认连接由 firstEp 指定的节点。
+- 两个没有配置 firstEp 参数的数据节点 dnode 启动后，会独立运行起来。这个时候，无法将其中一个数据节点加入到另外一个数据节点，形成集群。**无法将两个独立的集群合并成为新的集群**。
 
 ## <a class="anchor" id="management"></a>数据节点管理
 

documentation20/cn/11.administrator/docs.md

@@ -6,19 +6,27 @@
 
 ### 内存需求
 
-每个 DB 可以创建固定数目的 vgroup，默认与 CPU 核数相同，可通过 maxVgroupsPerDb 配置；vgroup 中的每个副本会是一个 vnode；每个 vnode 会占用固定大小的内存（大小与数据库的配置参数 blocks 和 cache 有关)；每个 Table 会占用与标签总长度有关的内存；此外，系统会有一些固定的内存开销。因此，每个 DB 需要的系统内存可通过如下公式计算：
+每个 Database 可以创建固定数目的 vgroup，默认与 CPU 核数相同，可通过 maxVgroupsPerDb 配置；vgroup 中的每个副本会是一个 vnode；每个 vnode 会占用固定大小的内存（大小与数据库的配置参数 blocks 和 cache 有关)；每个 Table 会占用与标签总长度有关的内存；此外，系统会有一些固定的内存开销。因此，每个 DB 需要的系统内存可通过如下公式计算：
 
 ```
-Memory Size = maxVgroupsPerDb * (blocks * cache + 10MB) + numOfTables * (tagSizePerTable + 0.5KB)
+Database Memory Size = maxVgroupsPerDb * (blocks * cache + 10MB) + numOfTables * (tagSizePerTable + 0.5KB)
 ```
 
-示例：假设是 4 核机器，cache 是缺省大小 16M, blocks 是缺省值 6，假设有 10 万张表，标签总长度是 256 字节，则总的内存需求为：4 \* (16 \* 6 + 10) + 100000 \* (0.25 + 0.5) / 1000 = 499M。
+示例：假设是 4 核机器，cache 是缺省大小 16M, blocks 是缺省值 6，并且一个 DB 中有 10 万张表，标签总长度是 256 字节，则这个 DB 总的内存需求为：4 \* (16 \* 6 + 10) + 100000 \* (0.25 + 0.5) / 1000 = 499M。
 
-注意：从这个公式计算得到的内存容量，应理解为系统的“必要需求”，而不是“内存总数”。在实际运行的生产系统中，由于操作系统缓存、资源管理调度等方面的需要，内存规划应当在计算结果的基础上保留一定冗余，以维持系统状态和系统性能的稳定性。
+在实际的系统运维中，我们通常会更关心 TDengine 服务进程（taosd）会占用的内存量。
+```
+taosd 内存总量 = vnode 内存 + mnode 内存 + 查询内存
+```
+
+其中：
+1. “vnode 内存”指的是集群中所有的 Database 存储分摊到当前 taosd 节点上所占用的内存资源。可以按上文“Database Memory Size”计算公式估算每个 DB 的内存占用量进行加总，再按集群中总共的 TDengine 节点数做平均（如果设置为多副本，则还需要乘以对应的副本倍数）。
+2. “mnode 内存”指的是集群中管理节点所占用的资源。如果一个 taosd 节点上分布有 mnode 管理节点，则内存消耗还需要增加“0.2KB * 集群中数据表总数”。
+3. “查询内存”指的是服务端处理查询请求时所需要占用的内存。单条查询语句至少会占用“0.2KB * 查询涉及的数据表总数”的内存量。
 
-实际运行的系统往往会根据数据特点的不同，将数据存放在不同的 DB 里。因此做规划时，也需要考虑。
+注意：以上内存估算方法，主要讲解了系统的“必须内存需求”，而不是“内存总数上限”。在实际运行的生产环境中，由于操作系统缓存、资源管理调度等方面的原因，内存规划应当在估算结果的基础上保留一定冗余，以维持系统状态和系统性能的稳定性。并且，生产环境通常会配置系统资源的监控工具，以便及时发现硬件资源的紧缺情况。
 
-如果内存充裕，可以加大 Blocks 的配置，这样更多数据将保存在内存里，提高查询速度。
+最后，如果内存充裕，可以考虑加大 Blocks 的配置，这样更多数据将保存在内存里，提高查询速度。
 
 ### CPU 需求
 

src/client/inc/tscUtil.h

@@ -83,6 +83,22 @@ typedef struct SJoinSupporter {
   SArray*         pVgroupTables;
 } SJoinSupporter;
 
+
+typedef struct SMergeCtx {
+  SJoinSupporter* p;
+  int32_t         idx;
+  SArray*         res;
+  int8_t          compared;
+}SMergeCtx;
+
+typedef struct SMergeTsCtx {
+  SJoinSupporter* p;
+  STSBuf*         res;
+  int64_t         numOfInput;
+  int8_t          compared;
+}SMergeTsCtx;
+
+
 typedef struct SVgroupTableInfo {
   SVgroupInfo vgInfo;
   SArray*     itemList;   //SArray<STableIdInfo>
@@ -183,6 +199,7 @@ int32_t   tscSqlExprCopy(SArray* dst, const SArray* src, uint64_t uid, bool deep
 void      tscSqlExprInfoDestroy(SArray* pExprInfo);
 
 SColumn* tscColumnClone(const SColumn* src);
+bool tscColumnExists(SArray* pColumnList, SColumnIndex* pColIndex);
 SColumn* tscColumnListInsert(SArray* pColList, SColumnIndex* colIndex);
 SArray* tscColumnListClone(const SArray* src, int16_t tableIndex);
 void tscColumnListDestroy(SArray* pColList);

src/client/inc/tsclient.h

@@ -142,15 +142,15 @@ typedef struct SCond {
 } SCond;
 
 typedef struct SJoinNode {
-  char     tableName[TSDB_TABLE_FNAME_LEN];
   uint64_t uid;
   int16_t  tagColId;
+  SArray*  tsJoin;
+  SArray*  tagJoin;
 } SJoinNode;
 
 typedef struct SJoinInfo {
   bool      hasJoin;  
-  SJoinNode left;
-  SJoinNode right;
+  SJoinNode*  joinTables[TSDB_MAX_JOIN_TABLE_NUM];
 } SJoinInfo;
 
 typedef struct STagCond {
@@ -285,7 +285,7 @@ typedef struct {
   char *         pRsp;
   int32_t        rspType;
   int32_t        rspLen;
-  uint64_t       qhandle;
+  uint64_t       qId;
   int64_t        useconds;
   int64_t        offset;  // offset value from vnode during projection query of stable
   int32_t        row;
@@ -368,7 +368,7 @@ typedef struct SSqlObj {
   int64_t          svgroupRid;
 
   int64_t          squeryLock;
-
+  int32_t          retryReason;  // previous error code
   struct SSqlObj  *prev, *next;
   int64_t          self;
 } SSqlObj;

src/client/src/tscAsync.c

@@ -160,8 +160,8 @@ static void tscProcessAsyncRetrieveImpl(void *param, TAOS_RES *tres, int numOfRo
   SSqlCmd *pCmd = &pSql->cmd;
   SSqlRes *pRes = &pSql->res;
 
-  if ((pRes->qhandle == 0 || numOfRows != 0) && pCmd->command < TSDB_SQL_LOCAL) {
-    if (pRes->qhandle == 0 && numOfRows != 0) {
+  if ((pRes->qId == 0 || numOfRows != 0) && pCmd->command < TSDB_SQL_LOCAL) {
+    if (pRes->qId == 0 && numOfRows != 0) {
       tscError("qhandle is NULL");
     } else {
       pRes->code = numOfRows;
@@ -208,7 +208,7 @@ void taos_fetch_rows_a(TAOS_RES *taosa, __async_cb_func_t fp, void *param) {
   pSql->fetchFp = fp;
   pSql->fp = tscAsyncFetchRowsProxy;
 
-  if (pRes->qhandle == 0) {
+  if (pRes->qId == 0) {
     tscError("qhandle is NULL");
     pRes->code = TSDB_CODE_TSC_INVALID_QHANDLE;
     pSql->param = param;
@@ -310,9 +310,50 @@ void tscAsyncResultOnError(SSqlObj* pSql) {
   taosScheduleTask(tscQhandle, &schedMsg);
 }
 
+int tscSendMsgToServer(SSqlObj *pSql);
 
+static int32_t updateMetaBeforeRetryQuery(SSqlObj* pSql, STableMetaInfo* pTableMetaInfo, SQueryInfo* pQueryInfo) {
+  // handle the invalid table error code for super table.
+  // update the pExpr info, colList info, number of table columns
+  // TODO Re-parse this sql and issue the corresponding subquery as an alternative for this case.
+  if (pSql->retryReason == TSDB_CODE_TDB_INVALID_TABLE_ID) {
+    int32_t numOfExprs = (int32_t) tscSqlExprNumOfExprs(pQueryInfo);
+    int32_t numOfCols = tscGetNumOfColumns(pTableMetaInfo->pTableMeta);
+    int32_t numOfTags = tscGetNumOfTags(pTableMetaInfo->pTableMeta);
 
-int tscSendMsgToServer(SSqlObj *pSql);
+    SSchema *pSchema = tscGetTableSchema(pTableMetaInfo->pTableMeta);
+    for (int32_t i = 0; i < numOfExprs; ++i) {
+      SSqlExpr *pExpr = tscSqlExprGet(pQueryInfo, i);
+      pExpr->uid = pTableMetaInfo->pTableMeta->id.uid;
+
+      if (pExpr->colInfo.colIndex >= 0) {
+        int32_t index = pExpr->colInfo.colIndex;
+
+        if ((TSDB_COL_IS_NORMAL_COL(pExpr->colInfo.flag) && index >= numOfCols) ||
+            (TSDB_COL_IS_TAG(pExpr->colInfo.flag) && (index < numOfCols || index >= (numOfCols + numOfTags)))) {
+          return pSql->retryReason;
+        }
+
+        if ((pSchema[pExpr->colInfo.colIndex].colId != pExpr->colInfo.colId) &&
+            strcasecmp(pExpr->colInfo.name, pSchema[pExpr->colInfo.colIndex].name) != 0) {
+          return pSql->retryReason;
+        }
+      }
+    }
+
+    // validate the table columns information
+    for (int32_t i = 0; i < taosArrayGetSize(pQueryInfo->colList); ++i) {
+      SColumn *pCol = taosArrayGetP(pQueryInfo->colList, i);
+      if (pCol->colIndex.columnIndex >= numOfCols) {
+        return pSql->retryReason;
+      }
+    }
+  } else {
+    // do nothing
+  }
+
+  return TSDB_CODE_SUCCESS;
+}
 
 void tscTableMetaCallBack(void *param, TAOS_RES *res, int code) {
   SSqlObj* pSql = (SSqlObj*)taosAcquireRef(tscObjRef, (int64_t)param);
@@ -339,7 +380,8 @@ void tscTableMetaCallBack(void *param, TAOS_RES *res, int code) {
     if (TSDB_QUERY_HAS_TYPE(pQueryInfo->type, (TSDB_QUERY_TYPE_STABLE_SUBQUERY|TSDB_QUERY_TYPE_SUBQUERY|TSDB_QUERY_TYPE_TAG_FILTER_QUERY))) {
       tscDebug("%p update local table meta, continue to process sql and send the corresponding query", pSql);
 
-      STableMetaInfo* pTableMetaInfo = tscGetMetaInfo(pQueryInfo, 0);
+      STableMetaInfo *pTableMetaInfo = tscGetMetaInfo(pQueryInfo, 0);
+
       code = tscGetTableMeta(pSql, pTableMetaInfo);
       assert(code == TSDB_CODE_TSC_ACTION_IN_PROGRESS || code == TSDB_CODE_SUCCESS);
 
@@ -349,6 +391,10 @@ void tscTableMetaCallBack(void *param, TAOS_RES *res, int code) {
       }
 
       assert((tscGetNumOfTags(pTableMetaInfo->pTableMeta) != 0));
+      code = updateMetaBeforeRetryQuery(pSql, pTableMetaInfo, pQueryInfo);
+      if (code != TSDB_CODE_SUCCESS) {
+        goto _error;
+      }
 
       // tscProcessSql can add error into async res
       tscProcessSql(pSql);

src/client/src/tscLocal.c

@@ -309,7 +309,7 @@ TAOS_ROW tscFetchRow(void *param) {
   SSqlCmd *pCmd = &pSql->cmd;
   SSqlRes *pRes = &pSql->res;
   
-  if (pRes->qhandle == 0 ||
+  if (pRes->qId == 0 ||
       pCmd->command == TSDB_SQL_RETRIEVE_EMPTY_RESULT ||
       pCmd->command == TSDB_SQL_INSERT) {
     return NULL;
@@ -905,7 +905,7 @@ int tscProcessLocalCmd(SSqlObj *pSql) {
      * set the qhandle to be 1 in order to pass the qhandle check, and to call partial release function to
      * free allocated resources and remove the SqlObj from sql query linked list
      */
-    pRes->qhandle = 0x1;
+    pRes->qId = 0x1;
     pRes->numOfRows = 0;
   } else if (pCmd->command == TSDB_SQL_SHOW_CREATE_TABLE) {
     pRes->code = tscProcessShowCreateTable(pSql); 

src/client/src/tscLocalMerge.c

@@ -1606,7 +1606,7 @@ void tscInitResObjForLocalQuery(SSqlObj *pObj, int32_t numOfRes, int32_t rowLen)
     tscDestroyLocalMerger(pObj);
   }
 
-  pRes->qhandle = 1;  // hack to pass the safety check in fetch_row function
+  pRes->qId = 1;  // hack to pass the safety check in fetch_row function
   pRes->numOfRows = 0;
   pRes->row = 0;
 

src/client/src/tscPrepare.c

@@ -903,7 +903,7 @@ int taos_stmt_prepare(TAOS_STMT* stmt, const char* sql, unsigned long length) {
     return TSDB_CODE_TSC_OUT_OF_MEMORY;
   }
 
-  pRes->qhandle = 0;
+  pRes->qId = 0;
   pRes->numOfRows = 1;
 
   strtolower(pSql->sqlstr, sql);

src/client/src/tscProfile.c

@@ -249,8 +249,8 @@ int tscBuildQueryStreamDesc(void *pMsg, STscObj *pObj) {
     pQdesc->stime = htobe64(pSql->stime);
     pQdesc->queryId = htonl(pSql->queryId);
     //pQdesc->useconds = htobe64(pSql->res.useconds);
-    pQdesc->useconds = htobe64(now - pSql->stime);
-    pQdesc->qHandle = htobe64(pSql->res.qhandle);
+    pQdesc->useconds = htobe64(now - pSql->stime);  // use local time instead of sever rsp elapsed time
+    pQdesc->qHandle = htobe64(pSql->res.qId);
 
     pHeartbeat->numOfQueries++;
     pQdesc++;

src/client/src/tscSQLParser.c

@@ -807,7 +807,18 @@ int32_t tscToSQLCmd(SSqlObj* pSql, struct SSqlInfo* pInfo) {
       if ((code = setKillInfo(pSql, pInfo, pInfo->type)) != TSDB_CODE_SUCCESS) {
         return code;
       }
+      break;
+    }
+
+    case TSDB_SQL_SYNC_DB_REPLICA: {
+      const char* msg1 = "invalid db name";
+      SStrToken* pzName = taosArrayGet(pInfo->pMiscInfo->a, 0);
 
+      assert(taosArrayGetSize(pInfo->pMiscInfo->a) == 1);
+      code = tNameSetDbName(&pTableMetaInfo->name, getAccountId(pSql), pzName);
+      if (code != TSDB_CODE_SUCCESS) {
+        return invalidSqlErrMsg(tscGetErrorMsgPayload(pCmd), msg1);
+      }
       break;
     }
 
@@ -1103,12 +1114,18 @@ int32_t parseSlidingClause(SSqlCmd* pCmd, SQueryInfo* pQueryInfo, SStrToken* pSl
 int32_t tscSetTableFullName(STableMetaInfo* pTableMetaInfo, SStrToken* pTableName, SSqlObj* pSql) {
   const char* msg1 = "name too long";
   const char* msg2 = "acctId too long";
+  const char* msg3 = "no acctId";
 
   SSqlCmd* pCmd = &pSql->cmd;
   int32_t  code = TSDB_CODE_SUCCESS;
 
   if (hasSpecifyDB(pTableName)) { // db has been specified in sql string so we ignore current db path
-    code = tNameSetAcctId(&pTableMetaInfo->name, getAccountId(pSql));
+    char* acctId = getAccountId(pSql);
+    if (acctId == NULL || strlen(acctId) <= 0) {
+      return invalidSqlErrMsg(tscGetErrorMsgPayload(pCmd), msg3);
+    }
+    
+    code = tNameSetAcctId(&pTableMetaInfo->name, acctId);
     if (code != 0) {
       return invalidSqlErrMsg(tscGetErrorMsgPayload(pCmd), msg2);
     }
@@ -3613,23 +3630,25 @@ static int32_t getColumnQueryCondInfo(SSqlCmd* pCmd, SQueryInfo* pQueryInfo, tSq
   }
 }
 
-static int32_t getJoinCondInfo(SSqlCmd* pCmd, SQueryInfo* pQueryInfo, tSqlExpr* pExpr) {
-  const char* msg1 = "invalid join query condition";
-  const char* msg2 = "invalid table name in join query";
+static int32_t checkAndSetJoinCondInfo(SSqlCmd* pCmd, SQueryInfo* pQueryInfo, tSqlExpr* pExpr) {
+  int32_t code = 0;
+  const char* msg1 = "timestamp required for join tables";
   const char* msg3 = "type of join columns must be identical";
   const char* msg4 = "invalid column name in join condition";
+  const char* msg5 = "only support one join tag for each table";
 
   if (pExpr == NULL) {
     return TSDB_CODE_SUCCESS;
   }
 
   if (!tSqlExprIsParentOfLeaf(pExpr)) {
-    return invalidSqlErrMsg(tscGetErrorMsgPayload(pCmd), msg1);
+    code = checkAndSetJoinCondInfo(pCmd, pQueryInfo, pExpr->pLeft);
+    if (code) {
+      return code;
     }
     
-  STagCond*  pTagCond = &pQueryInfo->tagCond;
-  SJoinNode* pLeft = &pTagCond->joinInfo.left;
-  SJoinNode* pRight = &pTagCond->joinInfo.right;
+    return checkAndSetJoinCondInfo(pCmd, pQueryInfo, pExpr->pRight);
+  }
 
   SColumnIndex index = COLUMN_INDEX_INITIALIZER;
   if (getColumnIndexByName(pCmd, &pExpr->pLeft->colInfo, pQueryInfo, &index) != TSDB_CODE_SUCCESS) {
@@ -3639,14 +3658,29 @@ static int32_t getJoinCondInfo(SSqlCmd* pCmd, SQueryInfo* pQueryInfo, tSqlExpr*
   STableMetaInfo* pTableMetaInfo = tscGetMetaInfo(pQueryInfo, index.tableIndex);
   SSchema* pTagSchema1 = tscGetTableColumnSchema(pTableMetaInfo->pTableMeta, index.columnIndex);
 
-  pLeft->uid = pTableMetaInfo->pTableMeta->id.uid;
-  pLeft->tagColId = pTagSchema1->colId;
+  assert(index.tableIndex >= 0 && index.tableIndex < TSDB_MAX_JOIN_TABLE_NUM);
 
-  int32_t code = tNameExtractFullName(&pTableMetaInfo->name, pLeft->tableName);
-  if (code != TSDB_CODE_SUCCESS) {
-    return invalidSqlErrMsg(tscGetErrorMsgPayload(pCmd), msg2);
+  SJoinNode **leftNode = &pQueryInfo->tagCond.joinInfo.joinTables[index.tableIndex];
+  if (*leftNode == NULL) {
+    return invalidSqlErrMsg(tscGetErrorMsgPayload(pCmd), msg1);
   }
   
+  (*leftNode)->uid = pTableMetaInfo->pTableMeta->id.uid;
+  (*leftNode)->tagColId = pTagSchema1->colId;
+
+  if (UTIL_TABLE_IS_SUPER_TABLE(pTableMetaInfo)) {
+    index.columnIndex = index.columnIndex - tscGetNumOfColumns(pTableMetaInfo->pTableMeta);
+    if (!tscColumnExists(pTableMetaInfo->tagColList, &index)) {
+      tscColumnListInsert(pTableMetaInfo->tagColList, &index);
+      if (taosArrayGetSize(pTableMetaInfo->tagColList) > 1) {
+        return invalidSqlErrMsg(tscGetErrorMsgPayload(pCmd), msg5);
+      }
+    }
+  }
+
+  int16_t leftIdx = index.tableIndex;
+
+
   index = (SColumnIndex)COLUMN_INDEX_INITIALIZER;
   if (getColumnIndexByName(pCmd, &pExpr->pRight->colInfo, pQueryInfo, &index) != TSDB_CODE_SUCCESS) {
     return invalidSqlErrMsg(tscGetErrorMsgPayload(pCmd), msg4);
@@ -3655,20 +3689,55 @@ static int32_t getJoinCondInfo(SSqlCmd* pCmd, SQueryInfo* pQueryInfo, tSqlExpr*
   pTableMetaInfo = tscGetMetaInfo(pQueryInfo, index.tableIndex);
   SSchema* pTagSchema2 = tscGetTableColumnSchema(pTableMetaInfo->pTableMeta, index.columnIndex);
 
-  pRight->uid = pTableMetaInfo->pTableMeta->id.uid;
-  pRight->tagColId = pTagSchema2->colId;
+  assert(index.tableIndex >= 0 && index.tableIndex < TSDB_MAX_JOIN_TABLE_NUM);
 
-  code = tNameExtractFullName(&pTableMetaInfo->name, pRight->tableName);
-  if (code != TSDB_CODE_SUCCESS) {
-    return invalidSqlErrMsg(tscGetErrorMsgPayload(pCmd), msg2);
+  SJoinNode **rightNode = &pQueryInfo->tagCond.joinInfo.joinTables[index.tableIndex];
+  if (*rightNode == NULL) {
+    return invalidSqlErrMsg(tscGetErrorMsgPayload(pCmd), msg1);
   }
 
+  (*rightNode)->uid = pTableMetaInfo->pTableMeta->id.uid;
+  (*rightNode)->tagColId = pTagSchema2->colId;
+
+  if (UTIL_TABLE_IS_SUPER_TABLE(pTableMetaInfo)) {
+    index.columnIndex = index.columnIndex - tscGetNumOfColumns(pTableMetaInfo->pTableMeta);
+    if (!tscColumnExists(pTableMetaInfo->tagColList, &index)) {
+      tscColumnListInsert(pTableMetaInfo->tagColList, &index);
+      if (taosArrayGetSize(pTableMetaInfo->tagColList) > 1) {
+        return invalidSqlErrMsg(tscGetErrorMsgPayload(pCmd), msg5);
+      }
+    }
+  }
+
+  int16_t rightIdx = index.tableIndex;
+
   if (pTagSchema1->type != pTagSchema2->type) {
     return invalidSqlErrMsg(tscGetErrorMsgPayload(pCmd), msg3);
   }
 
-  pTagCond->joinInfo.hasJoin = true;
+  if ((*leftNode)->tagJoin == NULL) {
+    (*leftNode)->tagJoin = taosArrayInit(2, sizeof(int16_t));
+  }
+  
+  if ((*rightNode)->tagJoin == NULL) {
+    (*rightNode)->tagJoin = taosArrayInit(2, sizeof(int16_t));
+  }      
+  
+  taosArrayPush((*leftNode)->tagJoin, &rightIdx);
+  taosArrayPush((*rightNode)->tagJoin, &leftIdx);
+  
+  pQueryInfo->tagCond.joinInfo.hasJoin = true;
+  
   return TSDB_CODE_SUCCESS;
+
+}
+
+static int32_t getJoinCondInfo(SSqlCmd* pCmd, SQueryInfo* pQueryInfo, tSqlExpr* pExpr) {
+  if (pExpr == NULL) {
+    return TSDB_CODE_SUCCESS;
+  }
+
+  return checkAndSetJoinCondInfo(pCmd, pQueryInfo, pExpr);
 }
 
 static int32_t validateSQLExpr(SSqlCmd* pCmd, tSqlExpr* pExpr, SQueryInfo* pQueryInfo, SColumnList* pList,
@@ -3932,7 +4001,7 @@ static int32_t handleExprInQueryCond(SSqlCmd* pCmd, SQueryInfo* pQueryInfo, tSql
   const char* msg1 = "table query cannot use tags filter";
   const char* msg2 = "illegal column name";
   const char* msg3 = "only one query time range allowed";
-  const char* msg4 = "only one join condition allowed";
+  const char* msg4 = "too many join tables";
   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";
@@ -3963,6 +4032,47 @@ static int32_t handleExprInQueryCond(SSqlCmd* pCmd, SQueryInfo* pQueryInfo, tSql
       TSDB_QUERY_SET_TYPE(pQueryInfo->type, TSDB_QUERY_TYPE_JOIN_QUERY);
       pCondExpr->tsJoin = true;
 
+      assert(index.tableIndex >= 0 && index.tableIndex < TSDB_MAX_JOIN_TABLE_NUM);
+      SJoinNode **leftNode = &pQueryInfo->tagCond.joinInfo.joinTables[index.tableIndex];
+      if (*leftNode == NULL) {
+        *leftNode = calloc(1, sizeof(SJoinNode));
+        if (*leftNode == NULL) {
+          return TSDB_CODE_TSC_OUT_OF_MEMORY;
+        }
+      }
+      
+      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);
+      }
+
+      if (index.tableIndex < 0 || index.tableIndex >= TSDB_MAX_JOIN_TABLE_NUM) {
+        return invalidSqlErrMsg(tscGetErrorMsgPayload(pCmd), msg4);
+      }
+
+      SJoinNode **rightNode = &pQueryInfo->tagCond.joinInfo.joinTables[index.tableIndex];
+      if (*rightNode == NULL) {
+        *rightNode = calloc(1, sizeof(SJoinNode));
+        if (*rightNode == NULL) {
+          return TSDB_CODE_TSC_OUT_OF_MEMORY;
+        }
+      }
+      
+      int16_t rightIdx = index.tableIndex;
+
+      if ((*leftNode)->tsJoin == NULL) {
+        (*leftNode)->tsJoin = taosArrayInit(2, sizeof(int16_t));
+      }
+      
+      if ((*rightNode)->tsJoin == NULL) {
+        (*rightNode)->tsJoin = taosArrayInit(2, sizeof(int16_t));
+      }      
+      
+      taosArrayPush((*leftNode)->tsJoin, &rightIdx);
+      taosArrayPush((*rightNode)->tsJoin, &leftIdx);
+      
       /*
        * to release expression, e.g., m1.ts = m2.ts,
        * since this expression is used to set the join query type
@@ -4020,10 +4130,6 @@ static int32_t handleExprInQueryCond(SSqlCmd* pCmd, SQueryInfo* pQueryInfo, tSql
           return TSDB_CODE_TSC_INVALID_SQL;
         }
 
-        if (pCondExpr->pJoinExpr != NULL) {
-          return invalidSqlErrMsg(tscGetErrorMsgPayload(pCmd), msg4);
-        }
-
         pQueryInfo->type |= TSDB_QUERY_TYPE_JOIN_QUERY;
         ret = setExprToCond(&pCondExpr->pJoinExpr, *pExpr, NULL, parentOptr, pQueryInfo->msg);
         *pExpr = NULL;
@@ -4251,6 +4357,7 @@ static bool validateFilterExpr(SQueryInfo* pQueryInfo) {
 static int32_t getTimeRangeFromExpr(SSqlCmd* pCmd, SQueryInfo* pQueryInfo, tSqlExpr* pExpr) {
   const char* msg0 = "invalid timestamp";
   const char* msg1 = "only one time stamp window allowed";
+  int32_t code = 0;
   
   if (pExpr == NULL) {
     return TSDB_CODE_SUCCESS;
@@ -4261,7 +4368,10 @@ static int32_t getTimeRangeFromExpr(SSqlCmd* pCmd, SQueryInfo* pQueryInfo, tSqlE
       return invalidSqlErrMsg(tscGetErrorMsgPayload(pCmd), msg1);
     }
 
-    getTimeRangeFromExpr(pCmd, pQueryInfo, pExpr->pLeft);
+    code = getTimeRangeFromExpr(pCmd, pQueryInfo, pExpr->pLeft);
+    if (code) {
+      return code;
+    }
     
     return getTimeRangeFromExpr(pCmd, pQueryInfo, pExpr->pRight);
   } else {
@@ -4343,6 +4453,7 @@ static void cleanQueryExpr(SCondExpr* pCondExpr) {
   }
 }
 
+/*
 static void doAddJoinTagsColumnsIntoTagList(SSqlCmd* pCmd, SQueryInfo* pQueryInfo, SCondExpr* pCondExpr) {
   STableMetaInfo* pTableMetaInfo = tscGetMetaInfo(pQueryInfo, 0);
   if (QUERY_IS_JOIN_QUERY(pQueryInfo->type) && UTIL_TABLE_IS_SUPER_TABLE(pTableMetaInfo)) {
@@ -4365,6 +4476,7 @@ static void doAddJoinTagsColumnsIntoTagList(SSqlCmd* pCmd, SQueryInfo* pQueryInf
     tscColumnListInsert(pTableMetaInfo->tagColList, &index);
   }
 }
+*/
 
 static int32_t validateTagCondExpr(SSqlCmd* pCmd, tExprNode *p) {
   const char *msg1 = "invalid tag operator";
@@ -4508,6 +4620,102 @@ static int32_t getTagQueryCondExpr(SSqlCmd* pCmd, SQueryInfo* pQueryInfo, SCondE
   return ret;
 }
 
+int32_t validateJoinNodes(SQueryInfo* pQueryInfo, SSqlObj* pSql) {
+  const char* msg1 = "timestamp required for join tables";
+  const char* msg2 = "tag required for join stables";
+  
+  for (int32_t i = 0; i < pQueryInfo->numOfTables; ++i) {
+    SJoinNode *node = pQueryInfo->tagCond.joinInfo.joinTables[i];  
+    
+    if (node == NULL || node->tsJoin == NULL || taosArrayGetSize(node->tsJoin) <= 0) {
+      return invalidSqlErrMsg(tscGetErrorMsgPayload(&pSql->cmd), msg1);
+    }
+  }
+
+  STableMetaInfo* pTableMetaInfo = tscGetMetaInfo(pQueryInfo, 0);
+  if (UTIL_TABLE_IS_SUPER_TABLE(pTableMetaInfo)) {
+    for (int32_t i = 0; i < pQueryInfo->numOfTables; ++i) {
+      SJoinNode *node = pQueryInfo->tagCond.joinInfo.joinTables[i];  
+      
+      if (node == NULL || node->tagJoin == NULL || taosArrayGetSize(node->tagJoin) <= 0) {
+        return invalidSqlErrMsg(tscGetErrorMsgPayload(&pSql->cmd), msg2);
+      }
+    }
+  }
+
+  return TSDB_CODE_SUCCESS;
+}
+
+
+void mergeJoinNodesImpl(int8_t* r, int8_t* p, int16_t* tidx, SJoinNode** nodes, int32_t type) {
+  SJoinNode *node = nodes[*tidx];  
+  SArray* arr = (type == 0) ? node->tsJoin : node->tagJoin;
+  size_t size = taosArrayGetSize(arr);
+
+  p[*tidx] = 1;
+  
+  for (int32_t j = 0; j < size; j++) {
+    int16_t* idx = taosArrayGet(arr, j);
+    r[*idx] = 1;
+    if (p[*idx] == 0) {
+      mergeJoinNodesImpl(r, p, idx, nodes, type);
+    }
+  }
+}
+
+int32_t mergeJoinNodes(SQueryInfo* pQueryInfo, SSqlObj* pSql) {
+  const char* msg1 = "not all join tables have same timestamp";
+  const char* msg2 = "not all join tables have same tag";
+  
+  int8_t r[TSDB_MAX_JOIN_TABLE_NUM] = {0};
+  int8_t p[TSDB_MAX_JOIN_TABLE_NUM] = {0};
+  
+  for (int16_t i = 0; i < pQueryInfo->numOfTables; ++i) {
+    mergeJoinNodesImpl(r, p, &i, pQueryInfo->tagCond.joinInfo.joinTables, 0);
+    
+    taosArrayClear(pQueryInfo->tagCond.joinInfo.joinTables[i]->tsJoin);
+  
+    for (int32_t j = 0; j < TSDB_MAX_JOIN_TABLE_NUM; ++j) {
+      if (r[j]) {
+        taosArrayPush(pQueryInfo->tagCond.joinInfo.joinTables[i]->tsJoin, &j);
+      }
+    }
+    
+    memset(r, 0, sizeof(r));
+    memset(p, 0, sizeof(p));
+  }
+
+  if (taosArrayGetSize(pQueryInfo->tagCond.joinInfo.joinTables[0]->tsJoin) != pQueryInfo->numOfTables) {
+    return invalidSqlErrMsg(tscGetErrorMsgPayload(&pSql->cmd), msg1);
+  }
+  
+  STableMetaInfo* pTableMetaInfo = tscGetMetaInfo(pQueryInfo, 0);
+  if (UTIL_TABLE_IS_SUPER_TABLE(pTableMetaInfo)) {
+    for (int16_t i = 0; i < pQueryInfo->numOfTables; ++i) {
+      mergeJoinNodesImpl(r, p, &i, pQueryInfo->tagCond.joinInfo.joinTables, 1);
+      
+      taosArrayClear(pQueryInfo->tagCond.joinInfo.joinTables[i]->tagJoin);
+    
+      for (int32_t j = 0; j < TSDB_MAX_JOIN_TABLE_NUM; ++j) {
+        if (r[j]) {
+          taosArrayPush(pQueryInfo->tagCond.joinInfo.joinTables[i]->tagJoin, &j);
+        }
+      }
+      
+      memset(r, 0, sizeof(r));
+      memset(p, 0, sizeof(p));
+    }
+
+    if (taosArrayGetSize(pQueryInfo->tagCond.joinInfo.joinTables[0]->tagJoin) != pQueryInfo->numOfTables) {
+      return invalidSqlErrMsg(tscGetErrorMsgPayload(&pSql->cmd), msg2);
+    }
+
+  }
+
+  return TSDB_CODE_SUCCESS;
+}
+
+
 int32_t parseWhereClause(SQueryInfo* pQueryInfo, tSqlExpr** pExpr, SSqlObj* pSql) {
   if (pExpr == NULL) {
     return TSDB_CODE_SUCCESS;
@@ -4553,17 +4761,17 @@ int32_t parseWhereClause(SQueryInfo* pQueryInfo, tSqlExpr** pExpr, SSqlObj* pSql
 
   // 4. get the table name query condition
   if ((ret = getTablenameCond(&pSql->cmd, pQueryInfo, condExpr.pTableCond, &sb)) != TSDB_CODE_SUCCESS) {
-    return ret;
+    goto PARSE_WHERE_EXIT;
   }
 
   // 5. other column query condition
   if ((ret = getColumnQueryCondInfo(&pSql->cmd, pQueryInfo, condExpr.pColumnCond, TK_AND)) != TSDB_CODE_SUCCESS) {
-    return ret;
+    goto PARSE_WHERE_EXIT;
   }
 
   // 6. join condition
   if ((ret = getJoinCondInfo(&pSql->cmd, pQueryInfo, condExpr.pJoinExpr)) != TSDB_CODE_SUCCESS) {
-    return ret;
+    goto PARSE_WHERE_EXIT;
   }
 
   // 7. query condition for table name
@@ -4571,12 +4779,29 @@ int32_t parseWhereClause(SQueryInfo* pQueryInfo, tSqlExpr** pExpr, SSqlObj* pSql
 
   ret = setTableCondForSTableQuery(&pSql->cmd, pQueryInfo, getAccountId(pSql), condExpr.pTableCond, condExpr.tableCondIndex, &sb);
   taosStringBuilderDestroy(&sb);
+  if (ret) {
+    goto PARSE_WHERE_EXIT;
+  }
 
   if (!validateFilterExpr(pQueryInfo)) {
-    return invalidSqlErrMsg(tscGetErrorMsgPayload(&pSql->cmd), msg2);
+    ret = invalidSqlErrMsg(tscGetErrorMsgPayload(&pSql->cmd), msg2);
+    goto PARSE_WHERE_EXIT;
+  }
+
+  //doAddJoinTagsColumnsIntoTagList(&pSql->cmd, pQueryInfo, &condExpr);
+  if (condExpr.tsJoin) {
+    ret = validateJoinNodes(pQueryInfo, pSql);
+    if (ret) {
+      goto PARSE_WHERE_EXIT;
+    }
+
+    ret = mergeJoinNodes(pQueryInfo, pSql);
+    if (ret) {
+      goto PARSE_WHERE_EXIT;
+    }    
   }
 
-  doAddJoinTagsColumnsIntoTagList(&pSql->cmd, pQueryInfo, &condExpr);
+PARSE_WHERE_EXIT:
 
   cleanQueryExpr(&condExpr);
   return ret;
@@ -6818,7 +7043,6 @@ int32_t doValidateSqlNode(SSqlObj* pSql, SQuerySqlNode* pQuerySqlNode, int32_t i
   const char* msg1 = "point interpolation query needs timestamp";
   const char* msg2 = "fill only available for interval query";
   const char* msg3 = "start(end) time of query range required or time range too large";
-  const char* msg4 = "illegal number of tables in from clause";
   const char* msg5 = "too many columns in selection clause";
   const char* msg6 = "too many tables in from clause";
   const char* msg7 = "invalid table alias name";
@@ -6855,14 +7079,11 @@ int32_t doValidateSqlNode(SSqlObj* pSql, SQuerySqlNode* pQuerySqlNode, int32_t i
   }
 
   size_t fromSize = taosArrayGetSize(pQuerySqlNode->from->tableList);
-  if (fromSize > TSDB_MAX_JOIN_TABLE_NUM) {
-    return invalidSqlErrMsg(tscGetErrorMsgPayload(pCmd), msg4);
+  if (fromSize > TSDB_MAX_JOIN_TABLE_NUM * 2) {
+    return invalidSqlErrMsg(tscGetErrorMsgPayload(pCmd), msg6);
   }
 
   pQueryInfo->command = TSDB_SQL_SELECT;
-  if (fromSize > 2) {
-    return invalidSqlErrMsg(tscGetErrorMsgPayload(pCmd), msg6);
-  }
 
   // set all query tables, which are maybe more than one.
   for (int32_t i = 0; i < fromSize; ++i) {

src/client/src/tscServer.c

@@ -350,8 +350,8 @@ void tscProcessMsgFromServer(SRpcMsg *rpcMsg, SRpcEpSet *pEpSet) {
         taosMsleep(duration);
       }
 
+      pSql->retryReason = rpcMsg->code;
       rpcMsg->code = tscRenewTableMeta(pSql, 0);
-
       // if there is an error occurring, proceed to the following error handling procedure.
       if (rpcMsg->code == TSDB_CODE_TSC_ACTION_IN_PROGRESS) {
         taosReleaseRef(tscObjRef, handle);
@@ -509,7 +509,7 @@ int tscBuildFetchMsg(SSqlObj *pSql, SSqlInfo *pInfo) {
 
   SQueryInfo *pQueryInfo = tscGetQueryInfoDetail(&pSql->cmd, pSql->cmd.clauseIndex);
   pRetrieveMsg->free    = htons(pQueryInfo->type);
-  pRetrieveMsg->qhandle = htobe64(pSql->res.qhandle);
+  pRetrieveMsg->qId     = htobe64(pSql->res.qId);
 
   // todo valid the vgroupId at the client side
   STableMetaInfo* pTableMetaInfo = tscGetMetaInfo(pQueryInfo, 0);
@@ -521,7 +521,7 @@ int tscBuildFetchMsg(SSqlObj *pSql, SSqlInfo *pInfo) {
       assert(pVgroupInfo->vgroups[vgIndex].vgId > 0 && vgIndex < pTableMetaInfo->vgroupList->numOfVgroups);
 
       pRetrieveMsg->header.vgId = htonl(pVgroupInfo->vgroups[vgIndex].vgId);
-      tscDebug("%p build fetch msg from vgId:%d, vgIndex:%d, qhandle:%" PRIX64, pSql, pVgroupInfo->vgroups[vgIndex].vgId, vgIndex, pSql->res.qhandle);
+      tscDebug("%p build fetch msg from vgId:%d, vgIndex:%d, qhandle:%" PRIX64, pSql, pVgroupInfo->vgroups[vgIndex].vgId, vgIndex, pSql->res.qId);
     } else {
       int32_t numOfVgroups = (int32_t)taosArrayGetSize(pTableMetaInfo->pVgroupTables);
       assert(vgIndex >= 0 && vgIndex < numOfVgroups);
@@ -529,12 +529,12 @@ int tscBuildFetchMsg(SSqlObj *pSql, SSqlInfo *pInfo) {
       SVgroupTableInfo* pTableIdList = taosArrayGet(pTableMetaInfo->pVgroupTables, vgIndex);
 
       pRetrieveMsg->header.vgId = htonl(pTableIdList->vgInfo.vgId);
-      tscDebug("%p build fetch msg from vgId:%d, vgIndex:%d, qhandle:%" PRIX64, pSql, pTableIdList->vgInfo.vgId, vgIndex, pSql->res.qhandle);
+      tscDebug("%p build fetch msg from vgId:%d, vgIndex:%d, qId:%" PRIu64, pSql, pTableIdList->vgInfo.vgId, vgIndex, pSql->res.qId);
     }
   } else {
     STableMeta* pTableMeta = pTableMetaInfo->pTableMeta;
     pRetrieveMsg->header.vgId = htonl(pTableMeta->vgId);
-    tscDebug("%p build fetch msg from only one vgroup, vgId:%d, qhandle:%" PRIX64, pSql, pTableMeta->vgId, pSql->res.qhandle);
+    tscDebug("%p build fetch msg from only one vgroup, vgId:%d, qId:%" PRIu64, pSql, pTableMeta->vgId, pSql->res.qId);
   }
 
   pSql->cmd.payloadLen = sizeof(SRetrieveTableMsg);
@@ -614,7 +614,7 @@ static int32_t tscEstimateQueryMsgSize(SSqlObj *pSql, int32_t clauseIndex) {
          tableSerialize + sqlLen + 4096 + pQueryInfo->bufLen;
 }
 
-static char *doSerializeTableInfo(SQueryTableMsg* pQueryMsg, SSqlObj *pSql, char *pMsg) {
+static char *doSerializeTableInfo(SQueryTableMsg* pQueryMsg, SSqlObj *pSql, char *pMsg, int32_t *succeed) {
   STableMetaInfo *pTableMetaInfo = tscGetTableMetaInfoFromCmd(&pSql->cmd, pSql->cmd.clauseIndex, 0);
   TSKEY dfltKey = htobe64(pQueryMsg->window.skey);
 
@@ -627,9 +627,14 @@ static char *doSerializeTableInfo(SQueryTableMsg* pQueryMsg, SSqlObj *pSql, char
       assert(index >= 0);
 
       SVgroupInfo* pVgroupInfo = NULL;
-      if (pTableMetaInfo->vgroupList->numOfVgroups > 0) {
+      if (pTableMetaInfo->vgroupList && pTableMetaInfo->vgroupList->numOfVgroups > 0) {
         assert(index < pTableMetaInfo->vgroupList->numOfVgroups);
         pVgroupInfo = &pTableMetaInfo->vgroupList->vgroups[index];
+      } else {
+        tscError("%p No vgroup info found", pSql);
+        
+        *succeed = 0;
+        return pMsg;
       }
 
       vgId = pVgroupInfo->vgId;
@@ -955,8 +960,13 @@ int tscBuildQueryMsg(SSqlObj *pSql, SSqlInfo *pInfo) {
     pQueryMsg->secondStageOutput = 0;
   }
 
+  int32_t succeed = 1;
+  
   // serialize the table info (sid, uid, tags)
-  pMsg = doSerializeTableInfo(pQueryMsg, pSql, pMsg);
+  pMsg = doSerializeTableInfo(pQueryMsg, pSql, pMsg, &succeed);
+  if (succeed == 0) {
+    return TSDB_CODE_TSC_APP_ERROR;
+  }
   
   SSqlGroupbyExpr *pGroupbyExpr = &pQueryInfo->groupbyExpr;
   if (pGroupbyExpr->numOfGroupCols > 0) {
@@ -1339,6 +1349,23 @@ int32_t tscBuildUseDbMsg(SSqlObj *pSql, SSqlInfo *pInfo) {
   return TSDB_CODE_SUCCESS;
 }
 
+int32_t tscBuildSyncDbReplicaMsg(SSqlObj* pSql, SSqlInfo *pInfo) {
+  SSqlCmd *pCmd = &pSql->cmd;
+  pCmd->payloadLen = sizeof(SSyncDbMsg);
+
+  if (TSDB_CODE_SUCCESS != tscAllocPayload(pCmd, pCmd->payloadLen)) {
+    tscError("%p failed to malloc for query msg", pSql);
+    return TSDB_CODE_TSC_OUT_OF_MEMORY;
+  }
+
+  SSyncDbMsg *pSyncMsg = (SSyncDbMsg *)pCmd->payload;
+  STableMetaInfo *pTableMetaInfo = tscGetTableMetaInfoFromCmd(pCmd, pCmd->clauseIndex, 0);
+  tNameExtractFullName(&pTableMetaInfo->name, pSyncMsg->db);
+  pCmd->msgType = TSDB_MSG_TYPE_CM_SYNC_DB;
+
+  return TSDB_CODE_SUCCESS;
+}
+
 int32_t tscBuildShowMsg(SSqlObj *pSql, SSqlInfo *pInfo) {
   STscObj *pObj = pSql->pTscObj;
   SSqlCmd *pCmd = &pSql->cmd;
@@ -1619,7 +1646,7 @@ int tscBuildRetrieveFromMgmtMsg(SSqlObj *pSql, SSqlInfo *pInfo) {
 
   SQueryInfo *pQueryInfo = tscGetQueryInfoDetail(pCmd, 0);
   SRetrieveTableMsg *pRetrieveMsg = (SRetrieveTableMsg*)pCmd->payload;
-  pRetrieveMsg->qhandle = htobe64(pSql->res.qhandle);
+  pRetrieveMsg->qId  = htobe64(pSql->res.qId);
   pRetrieveMsg->free = htons(pQueryInfo->type);
 
   return TSDB_CODE_SUCCESS;
@@ -2195,6 +2222,10 @@ int tscProcessSTableVgroupRsp(SSqlObj *pSql) {
     assert(pInfo->vgroupList != NULL);
 
     pInfo->vgroupList->numOfVgroups = pVgroupMsg->numOfVgroups;
+    if (pInfo->vgroupList->numOfVgroups <= 0) {
+      //tfree(pInfo->vgroupList);
+      tscError("%p empty vgroup info", pSql);
+    } else {
       for (int32_t j = 0; j < pInfo->vgroupList->numOfVgroups; ++j) {
         //just init, no need to lock
         SVgroupInfo *pVgroups = &pInfo->vgroupList->vgroups[j];
@@ -2210,6 +2241,7 @@ int tscProcessSTableVgroupRsp(SSqlObj *pSql) {
           pVgroups->epAddr[k].fqdn = strndup(vmsg->epAddr[k].fqdn, tListLen(vmsg->epAddr[k].fqdn));
         }
       }
+    }
 
     pMsg += size;
   }
@@ -2233,7 +2265,7 @@ int tscProcessShowRsp(SSqlObj *pSql) {
 
   pShow = (SShowRsp *)pRes->pRsp;
   pShow->qhandle = htobe64(pShow->qhandle);
-  pRes->qhandle = pShow->qhandle;
+  pRes->qId = pShow->qhandle;
 
   tscResetForNextRetrieve(pRes);
   pMetaMsg = &(pShow->tableMeta);
@@ -2415,11 +2447,12 @@ int tscProcessQueryRsp(SSqlObj *pSql) {
   SSqlRes *pRes = &pSql->res;
 
   SQueryTableRsp *pQuery = (SQueryTableRsp *)pRes->pRsp;
-  pQuery->qhandle = htobe64(pQuery->qhandle);
-  pRes->qhandle = pQuery->qhandle;
+  pQuery->qId = htobe64(pQuery->qId);
+  pRes->qId   = pQuery->qId;
 
   pRes->data = NULL;
   tscResetForNextRetrieve(pRes);
+  tscDebug("%p query rsp received, qId:%"PRIu64, pSql, pRes->qId);
   return 0;
 }
 
@@ -2477,7 +2510,8 @@ int tscProcessRetrieveRspFromNode(SSqlObj *pSql) {
   }
 
   pRes->row = 0;
-  tscDebug("%p numOfRows:%d, offset:%" PRId64 ", complete:%d", pSql, pRes->numOfRows, pRes->offset, pRes->completed);
+  tscDebug("%p numOfRows:%d, offset:%" PRId64 ", complete:%d, qId:%"PRIu64, pSql, pRes->numOfRows, pRes->offset,
+      pRes->completed, pRes->qId);
 
   return 0;
 }
@@ -2736,6 +2770,7 @@ void tscInitMsgsFp() {
   tscBuildMsg[TSDB_SQL_DROP_ACCT] = tscBuildDropUserAcctMsg;
   tscBuildMsg[TSDB_SQL_DROP_DB] = tscBuildDropDbMsg;
   tscBuildMsg[TSDB_SQL_DROP_FUNCTION] = tscBuildDropFuncMsg;
+  tscBuildMsg[TSDB_SQL_SYNC_DB_REPLICA] = tscBuildSyncDbReplicaMsg;
   tscBuildMsg[TSDB_SQL_DROP_TABLE] = tscBuildDropTableMsg;
   tscBuildMsg[TSDB_SQL_ALTER_USER] = tscBuildUserMsg;
   tscBuildMsg[TSDB_SQL_CREATE_DNODE] = tscBuildCreateDnodeMsg;

src/client/src/tscSql.c

@@ -476,7 +476,7 @@ TAOS_ROW taos_fetch_row(TAOS_RES *res) {
   SSqlCmd *pCmd = &pSql->cmd;
   SSqlRes *pRes = &pSql->res;
   
-  if (pRes->qhandle == 0 ||
+  if (pRes->qId == 0 ||
       pRes->code == TSDB_CODE_TSC_QUERY_CANCELLED ||
       pCmd->command == TSDB_SQL_RETRIEVE_EMPTY_RESULT ||
       pCmd->command == TSDB_SQL_INSERT) {
@@ -508,7 +508,7 @@ int taos_fetch_block(TAOS_RES *res, TAOS_ROW *rows) {
   SSqlCmd *pCmd = &pSql->cmd;
   SSqlRes *pRes = &pSql->res;
 
-  if (pRes->qhandle == 0 ||
+  if (pRes->qId == 0 ||
       pRes->code == TSDB_CODE_TSC_QUERY_CANCELLED ||
       pCmd->command == TSDB_SQL_RETRIEVE_EMPTY_RESULT ||
       pCmd->command == TSDB_SQL_INSERT) {
@@ -554,7 +554,7 @@ static bool tscKillQueryInDnode(SSqlObj* pSql) {
   SSqlCmd* pCmd = &pSql->cmd;
   SSqlRes* pRes = &pSql->res;
 
-  if (pRes == NULL || pRes->qhandle == 0) {
+  if (pRes == NULL || pRes->qId == 0) {
     return true;
   }
 
@@ -1050,7 +1050,7 @@ int taos_load_table_info(TAOS *taos, const char *tableNameList) {
    * If qhandle is NOT set 0, the function of taos_free_result() will send message to server by calling tscProcessSql()
    * to free connection, which may cause segment fault, when the parse phrase is not even successfully executed.
    */
-  pRes->qhandle = 0;
+  pRes->qId = 0;
   free(str);
 
   if (code != TSDB_CODE_SUCCESS) {

src/client/src/tscSub.c

@@ -149,7 +149,7 @@ static SSub* tscCreateSubscription(STscObj* pObj, const char* topic, const char*
   }
 
   strtolower(pSql->sqlstr, pSql->sqlstr);
-  pRes->qhandle = 0;
+  pRes->qId = 0;
   pRes->numOfRows = 1;
 
   code = tscAllocPayload(pCmd, TSDB_DEFAULT_PAYLOAD_SIZE);
@@ -448,7 +448,7 @@ SSqlObj* recreateSqlObj(SSub* pSub) {
     return NULL;
   }
 
-  pRes->qhandle = 0;
+  pRes->qId = 0;
   pRes->numOfRows = 1;
 
   int code = tscAllocPayload(pCmd, TSDB_DEFAULT_PAYLOAD_SIZE);
@@ -546,7 +546,7 @@ TAOS_RES *taos_consume(TAOS_SUB *tsub) {
     uint32_t type = pQueryInfo->type;
     tscFreeSqlResult(pSql);
     pRes->numOfRows = 1;
-    pRes->qhandle = 0;
+    pRes->qId = 0;
     pSql->cmd.command = TSDB_SQL_SELECT;
     pQueryInfo->type = type;
 

src/client/src/tscUtil.c

@@ -1279,6 +1279,34 @@ int32_t tscSqlExprCopy(SArray* dst, const SArray* src, uint64_t uid, bool deepco
   return 0;
 }
 
+bool tscColumnExists(SArray* pColumnList, SColumnIndex* pColIndex) {
+  // ignore the tbname columnIndex to be inserted into source list
+  if (pColIndex->columnIndex < 0) {
+    return false;
+  }
+  
+  size_t numOfCols = taosArrayGetSize(pColumnList);
+  int16_t col = pColIndex->columnIndex;
+
+  int32_t i = 0;
+  while (i < numOfCols) {
+    SColumn* pCol = taosArrayGetP(pColumnList, i);
+    if ((pCol->colIndex.columnIndex != col) || (pCol->colIndex.tableIndex != pColIndex->tableIndex)) {
+      ++i;
+      continue;
+    } else {
+      break;
+    }
+  }
+
+  if (i >= numOfCols || numOfCols == 0) {
+    return false;
+  }
+
+  return true;
+}
+
+
 SColumn* tscColumnListInsert(SArray* pColumnList, SColumnIndex* pColIndex) {
   // ignore the tbname columnIndex to be inserted into source list
   if (pColIndex->columnIndex < 0) {
@@ -1583,7 +1611,25 @@ int32_t tscTagCondCopy(STagCond* dest, const STagCond* src) {
   dest->tbnameCond.uid = src->tbnameCond.uid;
   dest->tbnameCond.len = src->tbnameCond.len;
 
-  memcpy(&dest->joinInfo, &src->joinInfo, sizeof(SJoinInfo));
+  dest->joinInfo.hasJoin = src->joinInfo.hasJoin;
+  
+  for (int32_t i = 0; i < TSDB_MAX_JOIN_TABLE_NUM; ++i) {
+    if (src->joinInfo.joinTables[i]) {
+      dest->joinInfo.joinTables[i] = calloc(1, sizeof(SJoinNode));
+
+      memcpy(dest->joinInfo.joinTables[i], src->joinInfo.joinTables[i], sizeof(SJoinNode));
+
+      if (src->joinInfo.joinTables[i]->tsJoin) {
+        dest->joinInfo.joinTables[i]->tsJoin = taosArrayDup(src->joinInfo.joinTables[i]->tsJoin);
+      }
+
+      if (src->joinInfo.joinTables[i]->tagJoin) {
+        dest->joinInfo.joinTables[i]->tagJoin = taosArrayDup(src->joinInfo.joinTables[i]->tagJoin);
+      }
+    }
+  }
+
+  
   dest->relType = src->relType;
   
   if (src->pCond == NULL) {
@@ -1629,6 +1675,23 @@ void tscTagCondRelease(STagCond* pTagCond) {
     taosArrayDestroy(pTagCond->pCond);
   }
 
+  for (int32_t i = 0; i < TSDB_MAX_JOIN_TABLE_NUM; ++i) {
+    SJoinNode *node = pTagCond->joinInfo.joinTables[i];
+    if (node == NULL) {
+      continue;
+    }
+
+    if (node->tsJoin != NULL) {
+      taosArrayDestroy(node->tsJoin);
+    }
+
+    if (node->tagJoin != NULL) {
+      taosArrayDestroy(node->tagJoin);
+    }
+
+    tfree(node);
+  }
+
   memset(pTagCond, 0, sizeof(STagCond));
 }
 
@@ -2318,16 +2381,21 @@ void tscDoQuery(SSqlObj* pSql) {
 }
 
 int16_t tscGetJoinTagColIdByUid(STagCond* pTagCond, uint64_t uid) {
-  if (pTagCond->joinInfo.left.uid == uid) {
-    return pTagCond->joinInfo.left.tagColId;
-  } else if (pTagCond->joinInfo.right.uid == uid) {
-    return pTagCond->joinInfo.right.tagColId;
-  } else {
+  int32_t i = 0;
+  while (i < TSDB_MAX_JOIN_TABLE_NUM) {
+    SJoinNode* node = pTagCond->joinInfo.joinTables[i];
+    if (node && node->uid == uid) {
+      return node->tagColId;
+    }
+
+    i++;
+  }
+
   assert(0);
   return -1;
-  }
 }
 
+
 int16_t tscGetTagColIndexById(STableMeta* pTableMeta, int16_t colId) {
   int32_t numOfTags = tscGetNumOfTags(pTableMeta);
 

src/common/inc/tcmdtype.h

@@ -53,6 +53,7 @@ enum {
   TSDB_DEFINE_SQL_TYPE( TSDB_SQL_ALTER_ACCT, "alter-acct" )
   TSDB_DEFINE_SQL_TYPE( TSDB_SQL_ALTER_TABLE, "alter-table" )
   TSDB_DEFINE_SQL_TYPE( TSDB_SQL_ALTER_DB, "alter-db" )
+  TSDB_DEFINE_SQL_TYPE(TSDB_SQL_SYNC_DB_REPLICA, "sync db-replica")
   TSDB_DEFINE_SQL_TYPE( TSDB_SQL_CREATE_MNODE, "create-mnode" )
   TSDB_DEFINE_SQL_TYPE( TSDB_SQL_DROP_MNODE, "drop-mnode" )
   TSDB_DEFINE_SQL_TYPE( TSDB_SQL_CREATE_DNODE, "create-dnode" )

src/common/src/tglobal.c

@@ -71,7 +71,7 @@ int32_t tsMaxBinaryDisplayWidth = 30;
 int32_t tsCompressMsgSize = -1;
 
 // client
-int32_t tsMaxSQLStringLen = TSDB_MAX_SQL_LEN;
+int32_t tsMaxSQLStringLen = TSDB_MAX_ALLOWED_SQL_LEN;
 int8_t  tsTscEnableRecordSql = 0;
 
 // the maximum number of results for projection query on super table that are returned from

src/dnode/CMakeLists.txt

@@ -35,6 +35,10 @@ IF (TD_TOPIC)
   TARGET_LINK_LIBRARIES(taosd topic)
 ENDIF ()
 
+IF (TD_MODULE AND TD_LINUX)
+  TARGET_LINK_LIBRARIES(taosd module dl)
+ENDIF ()
+
 SET(PREPARE_ENV_CMD "prepare_env_cmd")
 SET(PREPARE_ENV_TARGET "prepare_env_target")
 ADD_CUSTOM_COMMAND(OUTPUT ${PREPARE_ENV_CMD}

src/dnode/src/dnodeMain.c

@@ -39,6 +39,13 @@
 #include "dnodeMPeer.h"
 #include "dnodeShell.h"
 #include "dnodeTelemetry.h"
+#include "module.h"
+
+#if !defined(_MODULE) || !defined(_TD_LINUX)
+int32_t moduleStart() { return 0; }
+void    moduleStop() {}
+#endif
+
 
 void *tsDnodeTmr = NULL;
 static SRunStatus tsRunStatus = TSDB_RUN_STATUS_STOPPED;
@@ -146,6 +153,7 @@ int32_t dnodeInitSystem() {
   }
 
   dnodeSetRunStatus(TSDB_RUN_STATUS_RUNING);
+  moduleStart();
 
   dnodeReportStep("TDengine", "initialized successfully", 1);
   dInfo("TDengine is initialized successfully");
@@ -155,6 +163,7 @@ int32_t dnodeInitSystem() {
 
 void dnodeCleanUpSystem() {
   if (dnodeGetRunStatus() != TSDB_RUN_STATUS_STOPPED) {
+    moduleStop();
     dnodeSetRunStatus(TSDB_RUN_STATUS_STOPPED);
     dnodeCleanupTmr();
     dnodeCleanupComponents();

src/dnode/src/dnodePeer.c

@@ -43,6 +43,7 @@ int32_t dnodeInitServer() {
 
   dnodeProcessReqMsgFp[TSDB_MSG_TYPE_MD_CREATE_VNODE] = dnodeDispatchToVMgmtQueue; 
   dnodeProcessReqMsgFp[TSDB_MSG_TYPE_MD_ALTER_VNODE]  = dnodeDispatchToVMgmtQueue; 
+  dnodeProcessReqMsgFp[TSDB_MSG_TYPE_MD_SYNC_VNODE]  = dnodeDispatchToVMgmtQueue;
   dnodeProcessReqMsgFp[TSDB_MSG_TYPE_MD_DROP_VNODE]   = dnodeDispatchToVMgmtQueue;
   dnodeProcessReqMsgFp[TSDB_MSG_TYPE_MD_ALTER_STREAM] = dnodeDispatchToVMgmtQueue;
   dnodeProcessReqMsgFp[TSDB_MSG_TYPE_MD_CONFIG_DNODE] = dnodeDispatchToVMgmtQueue;

src/dnode/src/dnodeShell.c

@@ -50,6 +50,7 @@ int32_t dnodeInitShell() {
   dnodeProcessShellMsgFp[TSDB_MSG_TYPE_CM_CREATE_TP]   = dnodeDispatchToMWriteQueue;
   dnodeProcessShellMsgFp[TSDB_MSG_TYPE_CM_CREATE_FUNCTION] = dnodeDispatchToMWriteQueue;
   dnodeProcessShellMsgFp[TSDB_MSG_TYPE_CM_DROP_DB]     = dnodeDispatchToMWriteQueue;
+  dnodeProcessShellMsgFp[TSDB_MSG_TYPE_CM_SYNC_DB]     = dnodeDispatchToMWriteQueue;
   dnodeProcessShellMsgFp[TSDB_MSG_TYPE_CM_DROP_TP]     = dnodeDispatchToMWriteQueue;
   dnodeProcessShellMsgFp[TSDB_MSG_TYPE_CM_DROP_FUNCTION] = dnodeDispatchToMWriteQueue;
   dnodeProcessShellMsgFp[TSDB_MSG_TYPE_CM_ALTER_DB]    = dnodeDispatchToMWriteQueue;

src/dnode/src/dnodeVMgmt.c

@@ -30,6 +30,7 @@ static taos_queue  tsVMgmtQueue = NULL;
 static void *  dnodeProcessMgmtQueue(void *param);
 static int32_t dnodeProcessCreateVnodeMsg(SRpcMsg *pMsg);
 static int32_t dnodeProcessAlterVnodeMsg(SRpcMsg *pMsg);
+static int32_t dnodeProcessSyncVnodeMsg(SRpcMsg *pMsg);
 static int32_t dnodeProcessDropVnodeMsg(SRpcMsg *pMsg);
 static int32_t dnodeProcessAlterStreamMsg(SRpcMsg *pMsg);
 static int32_t dnodeProcessConfigDnodeMsg(SRpcMsg *pMsg);
@@ -39,6 +40,7 @@ static int32_t (*dnodeProcessMgmtMsgFp[TSDB_MSG_TYPE_MAX])(SRpcMsg *pMsg);
 int32_t dnodeInitVMgmt() {
   dnodeProcessMgmtMsgFp[TSDB_MSG_TYPE_MD_CREATE_VNODE] = dnodeProcessCreateVnodeMsg;
   dnodeProcessMgmtMsgFp[TSDB_MSG_TYPE_MD_ALTER_VNODE]  = dnodeProcessAlterVnodeMsg;
+  dnodeProcessMgmtMsgFp[TSDB_MSG_TYPE_MD_SYNC_VNODE]  = dnodeProcessSyncVnodeMsg;
   dnodeProcessMgmtMsgFp[TSDB_MSG_TYPE_MD_DROP_VNODE]   = dnodeProcessDropVnodeMsg;
   dnodeProcessMgmtMsgFp[TSDB_MSG_TYPE_MD_ALTER_STREAM] = dnodeProcessAlterStreamMsg;
   dnodeProcessMgmtMsgFp[TSDB_MSG_TYPE_MD_CONFIG_DNODE] = dnodeProcessConfigDnodeMsg;
@@ -179,6 +181,13 @@ static int32_t dnodeProcessAlterVnodeMsg(SRpcMsg *rpcMsg) {
   }
 }
 
+static int32_t dnodeProcessSyncVnodeMsg(SRpcMsg *rpcMsg) {
+  SSyncVnodeMsg *pSyncVnode = rpcMsg->pCont;
+  pSyncVnode->vgId = htonl(pSyncVnode->vgId);
+
+  return vnodeSync(pSyncVnode->vgId);
+}
+
 static int32_t dnodeProcessDropVnodeMsg(SRpcMsg *rpcMsg) {
   SDropVnodeMsg *pDrop = rpcMsg->pCont;
   pDrop->vgId = htonl(pDrop->vgId);

src/dnode/src/dnodeVnodes.c

@@ -206,6 +206,7 @@ static void dnodeProcessStatusRsp(SRpcMsg *pMsg) {
         exit(EXIT_FAILURE);
       }
     }
+
     taosTmrReset(dnodeSendStatusMsg, tsStatusInterval * 1000, NULL, tsDnodeTmr, &tsStatusTimer);
     return;
   }

src/inc/module.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 TDENGINE_MODULE
+#define TDENGINE_MODULE
+
+#ifdef __cplusplus
+extern "C" {
+#endif
+
+int32_t moduleStart();
+void    moduleStop();
+
+#ifdef __cplusplus
+}
+#endif
+
+#endif

src/inc/query.h

@@ -38,7 +38,7 @@ int32_t qCreateQueryInfo(void* tsdb, int32_t vgId, SQueryTableMsg* pQueryTableMs
  * @param qinfo
  * @return
  */
-bool qTableQuery(qinfo_t qinfo);
+bool qTableQuery(qinfo_t qinfo, uint64_t *qId);
 
 /**
  * Retrieve the produced results information, if current query is not paused or completed,

src/inc/taosdef.h

@@ -262,7 +262,7 @@ do { \
 #define TSDB_MIN_TABLES                 4
 #define TSDB_MAX_TABLES                 10000000
 #define TSDB_DEFAULT_TABLES             1000000
-#define TSDB_TABLES_STEP                1000
+#define TSDB_TABLES_STEP                100
 
 #define TSDB_MIN_DAYS_PER_FILE          1
 #define TSDB_MAX_DAYS_PER_FILE          3650 
@@ -320,7 +320,7 @@ do { \
 #define TSDB_MAX_DB_QUORUM_OPTION       2
 #define TSDB_DEFAULT_DB_QUORUM_OPTION   1
 
-#define TSDB_MAX_JOIN_TABLE_NUM         5
+#define TSDB_MAX_JOIN_TABLE_NUM         10
 #define TSDB_MAX_UNION_CLAUSE           5
 
 #define TSDB_MAX_BINARY_LEN            (TSDB_MAX_BYTES_PER_ROW-TSDB_KEYSIZE)

src/inc/taosmsg.h

@@ -59,6 +59,7 @@ TAOS_DEFINE_MESSAGE_TYPE( TSDB_MSG_TYPE_MD_DROP_STABLE, "drop-stable" )
 TAOS_DEFINE_MESSAGE_TYPE( TSDB_MSG_TYPE_MD_ALTER_STREAM, "alter-stream" )
 TAOS_DEFINE_MESSAGE_TYPE( TSDB_MSG_TYPE_MD_CONFIG_DNODE, "config-dnode" )
 TAOS_DEFINE_MESSAGE_TYPE( TSDB_MSG_TYPE_MD_ALTER_VNODE, "alter-vnode" )
+TAOS_DEFINE_MESSAGE_TYPE( TSDB_MSG_TYPE_MD_SYNC_VNODE, "sync-vnode" )
 TAOS_DEFINE_MESSAGE_TYPE( TSDB_MSG_TYPE_MD_CREATE_MNODE, "create-mnode" )
 TAOS_DEFINE_MESSAGE_TYPE( TSDB_MSG_TYPE_DUMMY6, "dummy6" )
 TAOS_DEFINE_MESSAGE_TYPE( TSDB_MSG_TYPE_DUMMY7, "dummy7" )
@@ -79,6 +80,7 @@ TAOS_DEFINE_MESSAGE_TYPE( TSDB_MSG_TYPE_CM_DROP_DB, "drop-db" )
 TAOS_DEFINE_MESSAGE_TYPE( TSDB_MSG_TYPE_CM_DROP_FUNCTION, "drop-function" )   
 TAOS_DEFINE_MESSAGE_TYPE( TSDB_MSG_TYPE_CM_USE_DB, "use-db" )	 
 TAOS_DEFINE_MESSAGE_TYPE( TSDB_MSG_TYPE_CM_ALTER_DB, "alter-db" )
+TAOS_DEFINE_MESSAGE_TYPE( TSDB_MSG_TYPE_CM_SYNC_DB, "sync-db-replica" )
 TAOS_DEFINE_MESSAGE_TYPE( TSDB_MSG_TYPE_CM_CREATE_TABLE, "create-table" )
 TAOS_DEFINE_MESSAGE_TYPE( TSDB_MSG_TYPE_CM_DROP_TABLE, "drop-table" )	
 TAOS_DEFINE_MESSAGE_TYPE( TSDB_MSG_TYPE_CM_ALTER_TABLE, "alter-table" )
@@ -391,7 +393,7 @@ typedef struct {
 
 typedef struct {
   int32_t vgId;
-} SDropVnodeMsg;
+} SDropVnodeMsg, SSyncVnodeMsg;
 
 typedef struct SColIndex {
   int16_t  colId;      // column id
@@ -519,12 +521,13 @@ typedef struct {
 
 typedef struct {
   int32_t  code;
-  uint64_t qhandle; // query handle
+  union{uint64_t qhandle; uint64_t qId;}; // query handle
 } SQueryTableRsp;
 
+// todo: the show handle should be replaced with id
 typedef struct {
   SMsgHead header;
-  uint64_t qhandle;
+  union{uint64_t qhandle; uint64_t qId;}; // query handle
   uint16_t free;
 } SRetrieveTableMsg;
 
@@ -613,7 +616,7 @@ typedef struct {
 typedef struct {
   char    db[TSDB_TABLE_FNAME_LEN];
   uint8_t ignoreNotExists;
-} SDropDbMsg, SUseDbMsg;
+} SDropDbMsg, SUseDbMsg, SSyncDbMsg;
 
 // IMPORTANT: sizeof(SVnodeStatisticInfo) should not exceed
 // TSDB_FILE_HEADER_LEN/4 - TSDB_FILE_HEADER_VERSION_SIZE

src/inc/tsdb.h

@@ -245,7 +245,7 @@ typedef struct {
  * @param qinfo      query info handle from query processor
  * @return
  */
-TsdbQueryHandleT *tsdbQueryTables(STsdbRepo *tsdb, STsdbQueryCond *pCond, STableGroupInfo *tableInfoGroup, void *qinfo,
+TsdbQueryHandleT *tsdbQueryTables(STsdbRepo *tsdb, STsdbQueryCond *pCond, STableGroupInfo *tableInfoGroup, uint64_t qId,
                                   SMemRef *pRef);
 
 /**
@@ -258,7 +258,7 @@ TsdbQueryHandleT *tsdbQueryTables(STsdbRepo *tsdb, STsdbQueryCond *pCond, STable
  * @param tableInfo  table list.
  * @return
  */
-TsdbQueryHandleT tsdbQueryLastRow(STsdbRepo *tsdb, STsdbQueryCond *pCond, STableGroupInfo *tableInfo, void *qinfo,
+TsdbQueryHandleT tsdbQueryLastRow(STsdbRepo *tsdb, STsdbQueryCond *pCond, STableGroupInfo *tableInfo, uint64_t qId,
                                   SMemRef *pRef);
 
 /**
@@ -277,7 +277,7 @@ SArray *tsdbGetQueriedTableList(TsdbQueryHandleT *pHandle);
  * @return
  */
 TsdbQueryHandleT tsdbQueryRowsInExternalWindow(STsdbRepo *tsdb, STsdbQueryCond *pCond, STableGroupInfo *groupList,
-                                               void *qinfo, SMemRef *pRef);
+                                               uint64_t qId, SMemRef *pRef);
 
 
 /**

src/inc/ttokendef.h

@@ -156,56 +156,58 @@
 #define TK_NOW                            137
 #define TK_RESET                          138
 #define TK_QUERY                          139
-#define TK_ADD                            140
-#define TK_COLUMN                         141
-#define TK_TAG                            142
-#define TK_CHANGE                         143
-#define TK_SET                            144
-#define TK_KILL                           145
-#define TK_CONNECTION                     146
-#define TK_STREAM                         147
-#define TK_COLON                          148
-#define TK_ABORT                          149
-#define TK_AFTER                          150
-#define TK_ATTACH                         151
-#define TK_BEFORE                         152
-#define TK_BEGIN                          153
-#define TK_CASCADE                        154
-#define TK_CLUSTER                        155
-#define TK_CONFLICT                       156
-#define TK_COPY                           157
-#define TK_DEFERRED                       158
-#define TK_DELIMITERS                     159
-#define TK_DETACH                         160
-#define TK_EACH                           161
-#define TK_END                            162
-#define TK_EXPLAIN                        163
-#define TK_FAIL                           164
-#define TK_FOR                            165
-#define TK_IGNORE                         166
-#define TK_IMMEDIATE                      167
-#define TK_INITIALLY                      168
-#define TK_INSTEAD                        169
-#define TK_MATCH                          170
-#define TK_KEY                            171
-#define TK_OF                             172
-#define TK_RAISE                          173
-#define TK_REPLACE                        174
-#define TK_RESTRICT                       175
-#define TK_ROW                            176
-#define TK_STATEMENT                      177
-#define TK_TRIGGER                        178
-#define TK_VIEW                           179
-#define TK_SEMI                           180
-#define TK_NONE                           181
-#define TK_PREV                           182
-#define TK_LINEAR                         183
-#define TK_IMPORT                         184
-#define TK_TBNAME                         185
-#define TK_JOIN                           186
-#define TK_INSERT                         187
-#define TK_INTO                           188
-#define TK_VALUES                         189
+#define TK_SYNCDB                         140
+#define TK_ADD                            141
+#define TK_COLUMN                         142
+#define TK_TAG                            143
+#define TK_CHANGE                         144
+#define TK_SET                            145
+#define TK_KILL                           146
+#define TK_CONNECTION                     147
+#define TK_STREAM                         148
+#define TK_COLON                          149
+#define TK_ABORT                          150
+#define TK_AFTER                          151
+#define TK_ATTACH                         152
+#define TK_BEFORE                         153
+#define TK_BEGIN                          154
+#define TK_CASCADE                        155
+#define TK_CLUSTER                        156
+#define TK_CONFLICT                       157
+#define TK_COPY                           158
+#define TK_DEFERRED                       159
+#define TK_DELIMITERS                     160
+#define TK_DETACH                         161
+#define TK_EACH                           162
+#define TK_END                            163
+#define TK_EXPLAIN                        164
+#define TK_FAIL                           165
+#define TK_FOR                            166
+#define TK_IGNORE                         167
+#define TK_IMMEDIATE                      168
+#define TK_INITIALLY                      169
+#define TK_INSTEAD                        170
+#define TK_MATCH                          171
+#define TK_KEY                            172
+#define TK_OF                             173
+#define TK_RAISE                          174
+#define TK_REPLACE                        175
+#define TK_RESTRICT                       176
+#define TK_ROW                            177
+#define TK_STATEMENT                      178
+#define TK_TRIGGER                        179
+#define TK_VIEW                           180
+#define TK_SEMI                           181
+#define TK_NONE                           182
+#define TK_PREV                           183
+#define TK_LINEAR                         184
+#define TK_IMPORT                         185
+#define TK_TBNAME                         186
+#define TK_JOIN                           187
+#define TK_INSERT                         188
+#define TK_INTO                           189
+#define TK_VALUES                         190
+
 
 
 

src/inc/vnode.h

@@ -60,6 +60,7 @@ int32_t vnodeCreate(SCreateVnodeMsg *pVnodeCfg);
 int32_t vnodeDrop(int32_t vgId);
 int32_t vnodeOpen(int32_t vgId);
 int32_t vnodeAlter(void *pVnode, SCreateVnodeMsg *pVnodeCfg);
+int32_t vnodeSync(int32_t vgId);
 int32_t vnodeClose(int32_t vgId);
 
 // vnodeMgmt

src/kit/shell/inc/shell.h

@@ -74,6 +74,7 @@ void source_file(TAOS* con, char* fptr);
 void source_dir(TAOS* con, SShellArguments* args);
 void shellCheck(TAOS* con, SShellArguments* args);
 void get_history_path(char* history);
+void shellCheck(TAOS* con, SShellArguments* args);
 void cleanup_handler(void* arg);
 void exitShell();
 int shellDumpResult(TAOS_RES* con, char* fname, int* error_no, bool printMode);

src/kit/shell/src/shellEngine.c

@@ -132,7 +132,6 @@ TAOS *shellInit(SShellArguments *args) {
   return con;
 }
 
-
 static bool isEmptyCommand(const char* cmd) {
   for (char c = *cmd++; c != 0; c = *cmd++) {
     if (c != ' ' && c != '\t' && c != ';') {

src/kit/taosdemo/CMakeLists.txt

@@ -3,6 +3,55 @@ PROJECT(TDengine)
 
 INCLUDE_DIRECTORIES(${TD_COMMUNITY_DIR}/src/client/inc)
 
+FIND_PACKAGE(Git)
+IF (GIT_FOUND)
+    MESSAGE("Git found")
+    EXECUTE_PROCESS(
+        COMMAND ${GIT_EXECUTABLE} log --pretty=oneline -n 1 ${CMAKE_CURRENT_LIST_DIR}/taosdemo.c
+        RESULT_VARIABLE RESULT
+        OUTPUT_VARIABLE TAOSDEMO_COMMIT)
+    EXECUTE_PROCESS(
+        COMMAND bash "-c" "echo '${TAOSDEMO_COMMIT}' | awk '{print $1}' | cut -c -9"
+        RESULT_VARIABLE RESULT
+        OUTPUT_VARIABLE TAOSDEMO_COMMIT_SHA1)
+    EXECUTE_PROCESS(
+        COMMAND ${GIT_EXECUTABLE} status -z -s ${CMAKE_CURRENT_LIST_DIR}/taosdemo.c
+        RESULT_VARIABLE RESULT
+        OUTPUT_VARIABLE TAOSDEMO_STATUS)
+    EXECUTE_PROCESS(
+        COMMAND bash "-c" "echo '${TAOSDEMO_STATUS}' | awk '{print $1}'"
+        RESULT_VARIABLE RESULT
+        OUTPUT_VARIABLE TAOSDEMO_STATUS)
+    MESSAGE("taosdemo.c status: " ${TAOSDEMO_STATUS})
+ELSE()
+    MESSAGE("Git not found")
+    SET(TAOSDEMO_COMMIT_SHA1 "unknown")
+    SET(TAOSDEMO_STATUS "unknown")
+ENDIF (GIT_FOUND)
+
+STRING(STRIP ${TAOSDEMO_COMMIT_SHA1} TAOSDEMO_COMMIT_SHA1)
+MESSAGE("taosdemo's latest commit in short is:" ${TAOSDEMO_COMMIT_SHA1})
+STRING(STRIP ${TAOSDEMO_STATUS} TAOSDEMO_STATUS)
+
+IF (TAOSDEMO_STATUS MATCHES "M")
+    SET(TAOSDEMO_STATUS "modified")
+ELSE()
+    SET(TAOSDEMO_STATUS "")
+ENDIF ()
+MESSAGE("taosdemo's status is:" ${TAOSDEMO_STATUS})
+
+ADD_DEFINITIONS(-DTAOSDEMO_COMMIT_SHA1="${TAOSDEMO_COMMIT_SHA1}")
+ADD_DEFINITIONS(-DTAOSDEMO_STATUS="${TAOSDEMO_STATUS}")
+
+MESSAGE("VERNUMBER is:" ${VERNUMBER})
+IF ("${VERNUMBER}" STREQUAL "")
+    SET(TD_VERSION_NUMBER "TDengine-version-unknown")
+ELSE()
+    SET(TD_VERSION_NUMBER ${VERNUMBER})
+ENDIF ()
+MESSAGE("TD_VERSION_NUMBER is:" ${TD_VERSION_NUMBER})
+ADD_DEFINITIONS(-DTD_VERNUMBER="${TD_VERSION_NUMBER}")
+
 IF (TD_LINUX)
   AUX_SOURCE_DIRECTORY(. SRC)
   ADD_EXECUTABLE(taosdemo ${SRC})

src/kit/taosdemo/insert-interlace.json

@@ -41,7 +41,7 @@
             "insert_mode": "taosc",
             "insert_rows": 1000,
             "multi_thread_write_one_tbl": "no",
-            "rows_per_tbl": 20,
+            "interlace_rows": 20,
             "max_sql_len": 1024000,
             "disorder_ratio": 0,
             "disorder_range": 1000,

src/kit/taosdemo/insert.json

@@ -41,7 +41,7 @@
             "insert_mode": "taosc",
             "insert_rows": 100000,
             "multi_thread_write_one_tbl": "no",
-            "rows_per_tbl": 0,
+            "interlace_rows": 0,
             "max_sql_len": 1024000,
             "disorder_ratio": 0,
             "disorder_range": 1000,

src/mnode/inc/mnodeVgroup.h

@@ -49,6 +49,7 @@ void    mnodeRemoveTableFromVgroup(SVgObj *pVgroup, SCTableObj *pTable);
 void    mnodeSendDropVnodeMsg(int32_t vgId, SRpcEpSet *epSet, void *ahandle);
 void    mnodeSendCreateVgroupMsg(SVgObj *pVgroup, void *ahandle);
 void    mnodeSendAlterVgroupMsg(SVgObj *pVgroup);
+void    mnodeSendSyncVgroupMsg(SVgObj *pVgroup);
 
 SRpcEpSet mnodeGetEpSetFromVgroup(SVgObj *pVgroup);
 SRpcEpSet mnodeGetEpSetFromIp(char *ep);

src/mnode/src/mnodeDb.c

@@ -50,7 +50,7 @@ static int32_t mnodeGetDbMeta(STableMetaMsg *pMeta, SShowObj *pShow, void *pConn
 static int32_t mnodeRetrieveDbs(SShowObj *pShow, char *data, int32_t rows, void *pConn);
 static int32_t mnodeProcessCreateDbMsg(SMnodeMsg *pMsg);
 static int32_t mnodeProcessDropDbMsg(SMnodeMsg *pMsg);
-
+static int32_t mnodeProcessSyncDbMsg(SMnodeMsg *pMsg);
 int32_t mnodeProcessAlterDbMsg(SMnodeMsg *pMsg);
 
 #ifndef _TOPIC
@@ -179,6 +179,7 @@ int32_t mnodeInitDbs() {
   mnodeAddWriteMsgHandle(TSDB_MSG_TYPE_CM_CREATE_DB, mnodeProcessCreateDbMsg);
   mnodeAddWriteMsgHandle(TSDB_MSG_TYPE_CM_ALTER_DB, mnodeProcessAlterDbMsg);
   mnodeAddWriteMsgHandle(TSDB_MSG_TYPE_CM_DROP_DB, mnodeProcessDropDbMsg);
+  mnodeAddWriteMsgHandle(TSDB_MSG_TYPE_CM_SYNC_DB, mnodeProcessSyncDbMsg);
   mnodeAddShowMetaHandle(TSDB_MGMT_TABLE_DB, mnodeGetDbMeta);
   mnodeAddShowRetrieveHandle(TSDB_MGMT_TABLE_DB, mnodeRetrieveDbs);
   mnodeAddShowFreeIterHandle(TSDB_MGMT_TABLE_DB, mnodeCancelGetNextDb);
@@ -1188,6 +1189,46 @@ static int32_t mnodeProcessDropDbMsg(SMnodeMsg *pMsg) {
   return mnodeDropDb(pMsg);
 }
 
+static int32_t mnodeSyncDb(SDbObj *pDb, SMnodeMsg *pMsg) {
+  void *pIter = NULL;
+  SVgObj *pVgroup = NULL;
+    while (1) {
+    pIter = mnodeGetNextVgroup(pIter, &pVgroup);
+    if (pVgroup == NULL) break;
+    if (pVgroup->pDb == pDb) {
+      mnodeSendSyncVgroupMsg(pVgroup);
+    }
+    mnodeDecVgroupRef(pVgroup);
+  }
+
+  mLInfo("db:%s, is synced by %s", pDb->name, mnodeGetUserFromMsg(pMsg));
+
+  return TSDB_CODE_SUCCESS;
+}
+
+static int32_t mnodeProcessSyncDbMsg(SMnodeMsg *pMsg) {
+  SSyncDbMsg *pSyncDb = pMsg->rpcMsg.pCont;
+  mDebug("db:%s, syncdb is received from thandle:%p, ignore:%d", pSyncDb->db, pMsg->rpcMsg.handle, pSyncDb->ignoreNotExists);
+
+  if (pMsg->pDb == NULL) pMsg->pDb = mnodeGetDb(pSyncDb->db);
+  if (pMsg->pDb == NULL) {
+        if (pSyncDb->ignoreNotExists) {
+          mDebug("db:%s, db is not exist, treat as success", pSyncDb->db);
+          return TSDB_CODE_SUCCESS;
+        } else {
+          mError("db:%s, failed to sync, invalid db", pSyncDb->db);
+          return TSDB_CODE_MND_INVALID_DB;
+        }
+  }
+
+  if (pMsg->pDb->status != TSDB_DB_STATUS_READY) {
+    mError("db:%s, status:%d, in dropping", pSyncDb->db, pMsg->pDb->status);
+    return TSDB_CODE_MND_DB_IN_DROPPING;
+  }
+
+  return mnodeSyncDb(pMsg->pDb, pMsg);
+}
+
 void  mnodeDropAllDbs(SAcctObj *pAcct)  {
   int32_t numOfDbs = 0;
   SDbObj *pDb = NULL;

src/mnode/src/mnodeVgroup.c

@@ -60,6 +60,7 @@ static int32_t mnodeGetVgroupMeta(STableMetaMsg *pMeta, SShowObj *pShow, void *p
 static int32_t mnodeRetrieveVgroups(SShowObj *pShow, char *data, int32_t rows, void *pConn);
 static void    mnodeProcessCreateVnodeRsp(SRpcMsg *rpcMsg);
 static void    mnodeProcessAlterVnodeRsp(SRpcMsg *rpcMsg);
+static void    mnodeProcessSyncVnodeRsp(SRpcMsg *rpcMsg);
 static void    mnodeProcessDropVnodeRsp(SRpcMsg *rpcMsg);
 static int32_t mnodeProcessVnodeCfgMsg(SMnodeMsg *pMsg) ;
 static void    mnodeSendDropVgroupMsg(SVgObj *pVgroup, void *ahandle);
@@ -236,6 +237,7 @@ int32_t mnodeInitVgroups() {
   mnodeAddShowFreeIterHandle(TSDB_MGMT_TABLE_VGROUP, mnodeCancelGetNextVgroup);
   mnodeAddPeerRspHandle(TSDB_MSG_TYPE_MD_CREATE_VNODE_RSP, mnodeProcessCreateVnodeRsp);
   mnodeAddPeerRspHandle(TSDB_MSG_TYPE_MD_ALTER_VNODE_RSP, mnodeProcessAlterVnodeRsp);
+  mnodeAddPeerRspHandle(TSDB_MSG_TYPE_MD_ALTER_VNODE_RSP, mnodeProcessSyncVnodeRsp);
   mnodeAddPeerRspHandle(TSDB_MSG_TYPE_MD_DROP_VNODE_RSP, mnodeProcessDropVnodeRsp);
   mnodeAddPeerMsgHandle(TSDB_MSG_TYPE_DM_CONFIG_VNODE, mnodeProcessVnodeCfgMsg);
 
@@ -967,6 +969,38 @@ void mnodeSendAlterVgroupMsg(SVgObj *pVgroup) {
   }
 }
 
+static SSyncVnodeMsg *mnodeBuildSyncVnodeMsg(int32_t vgId) {
+  SSyncVnodeMsg *pSyncVnode = rpcMallocCont(sizeof(SSyncVnodeMsg));
+  if (pSyncVnode == NULL) return NULL;
+
+  pSyncVnode->vgId = htonl(vgId);
+  return pSyncVnode;
+}
+
+static void mnodeSendSyncVnodeMsg(SVgObj *pVgroup, SRpcEpSet *epSet) {
+  SSyncVnodeMsg *pSyncVnode = mnodeBuildSyncVnodeMsg(pVgroup->vgId);
+  SRpcMsg rpcMsg = {
+    .ahandle = NULL,
+    .pCont   = pSyncVnode,
+    .contLen = pSyncVnode ? sizeof(SSyncVnodeMsg) : 0,
+    .code    = 0,
+    .msgType = TSDB_MSG_TYPE_MD_SYNC_VNODE
+  };
+
+  dnodeSendMsgToDnode(epSet, &rpcMsg);
+}
+
+void mnodeSendSyncVgroupMsg(SVgObj *pVgroup) {
+  mDebug("vgId:%d, send sync all vnodes msg, numOfVnodes:%d db:%s", pVgroup->vgId, pVgroup->numOfVnodes,
+         pVgroup->dbName);
+  for (int32_t i = 0; i < pVgroup->numOfVnodes; ++i) {
+    SRpcEpSet epSet = mnodeGetEpSetFromIp(pVgroup->vnodeGid[i].pDnode->dnodeEp);
+    mDebug("vgId:%d, index:%d, send sync vnode msg to dnode %s", pVgroup->vgId, i,
+           pVgroup->vnodeGid[i].pDnode->dnodeEp);
+    mnodeSendSyncVnodeMsg(pVgroup, &epSet);
+  }
+}
+
 static void mnodeSendCreateVnodeMsg(SVgObj *pVgroup, SRpcEpSet *epSet, void *ahandle) {
   SCreateVnodeMsg *pCreate = mnodeBuildVnodeMsg(pVgroup);
   SRpcMsg rpcMsg = {
@@ -994,6 +1028,10 @@ static void mnodeProcessAlterVnodeRsp(SRpcMsg *rpcMsg) {
   mDebug("alter vnode rsp received");
 }
 
+static void mnodeProcessSyncVnodeRsp(SRpcMsg *rpcMsg) {
+  mDebug("sync vnode rsp received");
+}
+
 static void mnodeProcessCreateVnodeRsp(SRpcMsg *rpcMsg) {
   if (rpcMsg->ahandle == NULL) return;
 

src/query/inc/qResultbuf.h

@@ -72,7 +72,7 @@ typedef struct SDiskbasedResultBuf {
   bool      comp;                // compressed before flushed to disk
   int32_t   nextPos;             // next page flush position
 
-  const void*      handle;       // for debug purpose
+  uint64_t  qId;                 // for debug purpose
   SResultBufStatis statis;
 } SDiskbasedResultBuf;
 
@@ -88,7 +88,7 @@ typedef struct SDiskbasedResultBuf {
  * @param handle
  * @return
  */
-int32_t createDiskbasedResultBuffer(SDiskbasedResultBuf** pResultBuf, int32_t pagesize, int32_t inMemBufSize, const void* handle);
+int32_t createDiskbasedResultBuffer(SDiskbasedResultBuf** pResultBuf, int32_t pagesize, int32_t inMemBufSize, uint64_t qId);
 
 /**
  *

src/query/inc/qUtil.h

@@ -25,6 +25,7 @@
   } while (0)
 
 #define GET_RES_WINDOW_KEY_LEN(_l) ((_l) + sizeof(uint64_t))
+#define GET_QID(_r)  (((SQInfo*)((_r)->qinfo))->qId)
 
 #define curTimeWindowIndex(_winres)        ((_winres)->curIndex)
 #define GET_ROW_PARAM_FOR_MULTIOUTPUT(_q, tbq, sq) (((tbq) && (!(sq)))? (_q)->pExpr1[1].base.arg->argValue.i64:1)

src/query/inc/sql.y

@@ -730,6 +730,9 @@ expritem(A) ::= .                              {A = 0;}
 ///////////////////////////////////reset query cache//////////////////////////////////////
 cmd ::= RESET QUERY CACHE.  { setDCLSqlElems(pInfo, TSDB_SQL_RESET_CACHE, 0);}
 
+///////////////////////////////////sync replica database//////////////////////////////////
+cmd ::= SYNCDB ids(X) REPLICA.{ setDCLSqlElems(pInfo, TSDB_SQL_SYNC_DB_REPLICA, 1, &X);}
+
 ///////////////////////////////////ALTER TABLE statement//////////////////////////////////
 cmd ::= ALTER TABLE ids(X) cpxName(F) ADD COLUMN columnlist(A).     {
     X.n += F.n;

src/query/src/qPercentile.c

@@ -254,7 +254,7 @@ tMemBucket *tMemBucketCreate(int16_t nElemSize, int16_t dataType, double minval,
 
   resetSlotInfo(pBucket);
 
-  int32_t ret = createDiskbasedResultBuffer(&pBucket->pBuffer, pBucket->bufPageSize, pBucket->bufPageSize * 512, NULL);
+  int32_t ret = createDiskbasedResultBuffer(&pBucket->pBuffer, pBucket->bufPageSize, pBucket->bufPageSize * 512, 1);
   if (ret != TSDB_CODE_SUCCESS) {
     tMemBucketDestroy(pBucket);
     return NULL;

src/query/src/qResultbuf.c

@@ -9,7 +9,7 @@
 #define GET_DATA_PAYLOAD(_p) ((char *)(_p)->pData + POINTER_BYTES)
 #define NO_IN_MEM_AVAILABLE_PAGES(_b) (listNEles((_b)->lruList) >= (_b)->inMemPages)
 
-int32_t createDiskbasedResultBuffer(SDiskbasedResultBuf** pResultBuf, int32_t pagesize, int32_t inMemBufSize, const void* handle) {
+int32_t createDiskbasedResultBuffer(SDiskbasedResultBuf** pResultBuf, int32_t pagesize, int32_t inMemBufSize, uint64_t qId) {
   *pResultBuf = calloc(1, sizeof(SDiskbasedResultBuf));
 
   SDiskbasedResultBuf* pResBuf = *pResultBuf;
@@ -24,7 +24,7 @@ int32_t createDiskbasedResultBuffer(SDiskbasedResultBuf** pResultBuf, int32_t pa
   pResBuf->allocateId   = -1;
   pResBuf->comp         = true;
   pResBuf->file         = NULL;
-  pResBuf->handle       = handle;
+  pResBuf->qId          = qId;
   pResBuf->fileSize     = 0;
 
   // at least more than 2 pages must be in memory
@@ -43,7 +43,7 @@ int32_t createDiskbasedResultBuffer(SDiskbasedResultBuf** pResultBuf, int32_t pa
 
   pResBuf->emptyDummyIdList = taosArrayInit(1, sizeof(int32_t));
 
-  qDebug("QInfo:%p create resBuf for output, page size:%d, inmem buf pages:%d, file:%s", handle, pResBuf->pageSize,
+  qDebug("QInfo:%"PRIu64" create resBuf for output, page size:%d, inmem buf pages:%d, file:%s", qId, pResBuf->pageSize,
          pResBuf->inMemPages, pResBuf->path);
 
   return TSDB_CODE_SUCCESS;
@@ -406,13 +406,13 @@ void destroyResultBuf(SDiskbasedResultBuf* pResultBuf) {
   }
 
   if (pResultBuf->file != NULL) {
-    qDebug("QInfo:%p res output buffer closed, total:%.2f Kb, inmem size:%.2f Kb, file size:%.2f Kb",
-        pResultBuf->handle, pResultBuf->totalBufSize/1024.0, listNEles(pResultBuf->lruList) * pResultBuf->pageSize / 1024.0,
+    qDebug("QInfo:%"PRIu64" res output buffer closed, total:%.2f Kb, inmem size:%.2f Kb, file size:%.2f Kb",
+        pResultBuf->qId, pResultBuf->totalBufSize/1024.0, listNEles(pResultBuf->lruList) * pResultBuf->pageSize / 1024.0,
         pResultBuf->fileSize/1024.0);
 
     fclose(pResultBuf->file);
   } else {
-    qDebug("QInfo:%p res output buffer closed, total:%.2f Kb, no file created", pResultBuf->handle,
+    qDebug("QInfo:%"PRIu64" res output buffer closed, total:%.2f Kb, no file created", pResultBuf->qId,
            pResultBuf->totalBufSize/1024.0);
   }
 

src/query/src/qSqlParser.c

@@ -911,6 +911,7 @@ void setDCLSqlElems(SSqlInfo *pInfo, int32_t type, int32_t nParam, ...) {
     SStrToken *pToken = va_arg(va, SStrToken *);
     taosArrayPush(pInfo->pMiscInfo->a, pToken);
   }
+
   va_end(va);
 }
 

src/query/src/qTokenizer.c

@@ -100,6 +100,7 @@ static SKeyword keywordTable[] = {
     {"ACCOUNT",      TK_ACCOUNT},
     {"USE",          TK_USE},
     {"DESCRIBE",     TK_DESCRIBE},
+    {"SYNCDB",       TK_SYNCDB},
     {"ALTER",        TK_ALTER},
     {"PASS",         TK_PASS},
     {"PRIVILEGE",    TK_PRIVILEGE},

src/query/src/qUtil.c

@@ -468,7 +468,7 @@ static int32_t mergeIntoGroupResultImpl(SQueryRuntimeEnv *pRuntimeEnv, SGroupRes
   pTableQueryInfoList = malloc(POINTER_BYTES * size);
 
   if (pTableQueryInfoList == NULL || posList == NULL || pGroupResInfo->pRows == NULL || pGroupResInfo->pRows == NULL) {
-    qError("QInfo:%p failed alloc memory", pRuntimeEnv->qinfo);
+    qError("QInfo:%"PRIu64" failed alloc memory", GET_QID(pRuntimeEnv));
     code = TSDB_CODE_QRY_OUT_OF_MEMORY;
     goto _end;
   }
@@ -540,7 +540,7 @@ static int32_t mergeIntoGroupResultImpl(SQueryRuntimeEnv *pRuntimeEnv, SGroupRes
 
   int64_t endt = taosGetTimestampMs();
 
-  qDebug("QInfo:%p result merge completed for group:%d, elapsed time:%" PRId64 " ms", pRuntimeEnv->qinfo,
+  qDebug("QInfo:%"PRIu64" result merge completed for group:%d, elapsed time:%" PRId64 " ms", GET_QID(pRuntimeEnv),
          pGroupResInfo->currentGroup, endt - startt);
 
   _end:
@@ -567,13 +567,13 @@ int32_t mergeIntoGroupResult(SGroupResInfo* pGroupResInfo, SQueryRuntimeEnv* pRu
       break;
     }
 
-    qDebug("QInfo:%p no result in group %d, continue", pRuntimeEnv->qinfo, pGroupResInfo->currentGroup);
+    qDebug("QInfo:%"PRIu64" no result in group %d, continue", GET_QID(pRuntimeEnv), pGroupResInfo->currentGroup);
     cleanupGroupResInfo(pGroupResInfo);
     incNextGroup(pGroupResInfo);
   }
 
   int64_t elapsedTime = taosGetTimestampUs() - st;
-  qDebug("QInfo:%p merge res data into group, index:%d, total group:%d, elapsed time:%" PRId64 "us", pRuntimeEnv->qinfo,
+  qDebug("QInfo:%"PRIu64" merge res data into group, index:%d, total group:%d, elapsed time:%" PRId64 "us", GET_QID(pRuntimeEnv),
          pGroupResInfo->currentGroup, pGroupResInfo->totalGroup, elapsedTime);
 
 //  pQInfo->summary.firstStageMergeTime += elapsedTime;

src/query/src/queryMain.c

@@ -200,29 +200,30 @@ int32_t qCreateQueryInfo(void* tsdb, int32_t vgId, SQueryTableMsg* pQueryMsg, qi
   return code;
 }
 
-bool qTableQuery(qinfo_t qinfo) {
+bool qTableQuery(qinfo_t qinfo, uint64_t *qId) {
   SQInfo *pQInfo = (SQInfo *)qinfo;
   assert(pQInfo && pQInfo->signature == pQInfo);
   int64_t threadId = taosGetSelfPthreadId();
 
   int64_t curOwner = 0;
   if ((curOwner = atomic_val_compare_exchange_64(&pQInfo->owner, 0, threadId)) != 0) {
-    qError("QInfo:%p qhandle is now executed by thread:%p", pQInfo, (void*) curOwner);
+    qError("QInfo:%"PRIu64"-%p qhandle is now executed by thread:%p", pQInfo->qId, pQInfo, (void*) curOwner);
     pQInfo->code = TSDB_CODE_QRY_IN_EXEC;
     return false;
   }
 
 
+  *qId = pQInfo->qId;
   pQInfo->startExecTs = taosGetTimestampSec();
 
   if (isQueryKilled(pQInfo)) {
-    qDebug("QInfo:%p it is already killed, abort", pQInfo);
+    qDebug("QInfo:%"PRIu64" it is already killed, abort", pQInfo->qId);
     return doBuildResCheck(pQInfo);
   }
 
   SQueryRuntimeEnv* pRuntimeEnv = &pQInfo->runtimeEnv;
   if (pRuntimeEnv->tableqinfoGroupInfo.numOfTables == 0) {
-    qDebug("QInfo:%p no table exists for query, abort", pQInfo);
+    qDebug("QInfo:%"PRIu64" no table exists for query, abort", pQInfo->qId);
     setQueryStatus(pRuntimeEnv, QUERY_COMPLETED);
     return doBuildResCheck(pQInfo);
   }
@@ -231,22 +232,22 @@ bool qTableQuery(qinfo_t qinfo) {
   int32_t ret = setjmp(pQInfo->runtimeEnv.env);
   if (ret != TSDB_CODE_SUCCESS) {
     pQInfo->code = ret;
-    qDebug("QInfo:%p query abort due to error/cancel occurs, code:%s", pQInfo, tstrerror(pQInfo->code));
+    qDebug("QInfo:%"PRIu64" query abort due to error/cancel occurs, code:%s", pQInfo->qId, tstrerror(pQInfo->code));
     return doBuildResCheck(pQInfo);
   }
 
-  qDebug("QInfo:%p query task is launched", pQInfo);
+  qDebug("QInfo:%"PRIu64" query task is launched", pQInfo->qId);
 
   pRuntimeEnv->outputBuf = pRuntimeEnv->proot->exec(pRuntimeEnv->proot);
 
   if (isQueryKilled(pQInfo)) {
-    qDebug("QInfo:%p query is killed", pQInfo);
+    qDebug("QInfo:%"PRIu64" query is killed", pQInfo->qId);
   } else if (GET_NUM_OF_RESULTS(pRuntimeEnv) == 0) {
-    qDebug("QInfo:%p over, %u tables queried, %"PRId64" rows are returned", pQInfo, pRuntimeEnv->tableqinfoGroupInfo.numOfTables,
+    qDebug("QInfo:%"PRIu64" over, %u tables queried, %"PRId64" rows are returned", pQInfo->qId, pRuntimeEnv->tableqinfoGroupInfo.numOfTables,
            pRuntimeEnv->resultInfo.total);
   } else {
-    qDebug("QInfo:%p query paused, %d rows returned, numOfTotal:%" PRId64 " rows",
-           pQInfo, GET_NUM_OF_RESULTS(pRuntimeEnv), pRuntimeEnv->resultInfo.total + GET_NUM_OF_RESULTS(pRuntimeEnv));
+    qDebug("QInfo:%"PRIu64" query paused, %d rows returned, numOfTotal:%" PRId64 " rows",
+           pQInfo->qId, GET_NUM_OF_RESULTS(pRuntimeEnv), pRuntimeEnv->resultInfo.total + GET_NUM_OF_RESULTS(pRuntimeEnv));
   }
 
   return doBuildResCheck(pQInfo);
@@ -256,13 +257,13 @@ int32_t qRetrieveQueryResultInfo(qinfo_t qinfo, bool* buildRes, void* pRspContex
   SQInfo *pQInfo = (SQInfo *)qinfo;
 
   if (pQInfo == NULL || !isValidQInfo(pQInfo)) {
-    qError("QInfo:%p invalid qhandle", pQInfo);
+    qError("QInfo:%"PRIu64" invalid qhandle", pQInfo->qId);
     return TSDB_CODE_QRY_INVALID_QHANDLE;
   }
 
   *buildRes = false;
   if (IS_QUERY_KILLED(pQInfo)) {
-    qDebug("QInfo:%p query is killed, code:0x%08x", pQInfo, pQInfo->code);
+    qDebug("QInfo:%"PRIu64" query is killed, code:0x%08x", pQInfo->qId, pQInfo->code);
     return pQInfo->code;
   }
 
@@ -282,11 +283,11 @@ int32_t qRetrieveQueryResultInfo(qinfo_t qinfo, bool* buildRes, void* pRspContex
     assert(pQInfo->rspContext == NULL);
     if (pQInfo->dataReady == QUERY_RESULT_READY) {
       *buildRes = true;
-      qDebug("QInfo:%p retrieve result info, rowsize:%d, rows:%d, code:%s", pQInfo, pQuery->resultRowSize,
+      qDebug("QInfo:%"PRIu64" retrieve result info, rowsize:%d, rows:%d, code:%s", pQInfo->qId, pQuery->resultRowSize,
              GET_NUM_OF_RESULTS(pRuntimeEnv), tstrerror(pQInfo->code));
     } else {
       *buildRes = false;
-      qDebug("QInfo:%p retrieve req set query return result after paused", pQInfo);
+      qDebug("QInfo:%"PRIu64" retrieve req set query return result after paused", pQInfo->qId);
       pQInfo->rspContext = pRspContext;
       assert(pQInfo->rspContext != NULL);
     }
@@ -345,9 +346,10 @@ int32_t qDumpRetrieveResult(qinfo_t qinfo, SRetrieveTableRsp **pRsp, int32_t *co
     // here current thread hold the refcount, so it is safe to free tsdbQueryHandle.
     *continueExec = false;
     (*pRsp)->completed = 1;  // notify no more result to client
+    qDebug("QInfo:%"PRIu64" no more results to retrieve", pQInfo->qId);
   } else {
     *continueExec = true;
-    qDebug("QInfo:%p has more results to retrieve", pQInfo);
+    qDebug("QInfo:%"PRIu64" has more results to retrieve", pQInfo->qId);
   }
 
   // the memory should be freed if the code of pQInfo is not TSDB_CODE_SUCCESS
@@ -400,7 +402,7 @@ void qDestroyQueryInfo(qinfo_t qHandle) {
     return;
   }
 
-  qDebug("QInfo:%p query completed", pQInfo);
+  qDebug("QInfo:%"PRIu64" query completed", pQInfo->qId);
   queryCostStatis(pQInfo);   // print the query cost summary
   freeQInfo(pQInfo);
 }
@@ -483,7 +485,7 @@ void** qRegisterQInfo(void* pMgmt, uint64_t qId, uint64_t qInfo) {
 
   SQueryMgmt *pQueryMgmt = pMgmt;
   if (pQueryMgmt->qinfoPool == NULL) {
-    qError("QInfo:%p failed to add qhandle into qMgmt, since qMgmt is closed", (void *)qInfo);
+    qError("QInfo:%"PRIu64"-%p failed to add qhandle into qMgmt, since qMgmt is closed", qId, (void*)qInfo);
     terrno = TSDB_CODE_VND_INVALID_VGROUP_ID;
     return NULL;
   }
@@ -491,7 +493,7 @@ void** qRegisterQInfo(void* pMgmt, uint64_t qId, uint64_t qInfo) {
   pthread_mutex_lock(&pQueryMgmt->lock);
   if (pQueryMgmt->closed) {
     pthread_mutex_unlock(&pQueryMgmt->lock);
-    qError("QInfo:%p failed to add qhandle into cache, since qMgmt is colsing", (void *)qInfo);
+    qError("QInfo:%"PRIu64"-%p failed to add qhandle into cache, since qMgmt is colsing", qId, (void*)qInfo);
     terrno = TSDB_CODE_VND_INVALID_VGROUP_ID;
     return NULL;
   } else {

src/rpc/src/rpcMain.c

@@ -1017,6 +1017,13 @@ static SRpcConn *rpcProcessMsgHead(SRpcInfo *pRpc, SRecvInfo *pRecv, SRpcReqCont
   return pConn;
 }
 
+static void doRpcReportBrokenLinkToServer(void *param, void *id) {
+   SRpcMsg *pRpcMsg = (SRpcMsg *)(param); 
+   SRpcConn *pConn  = (SRpcConn *)(pRpcMsg->handle);
+   SRpcInfo *pRpc   = pConn->pRpc; 
+   (*(pRpc->cfp))(pRpcMsg, NULL);
+   free(pRpcMsg);
+}
 static void rpcReportBrokenLinkToServer(SRpcConn *pConn) {
   SRpcInfo *pRpc = pConn->pRpc;
   if (pConn->pReqMsg == NULL) return;
@@ -1025,16 +1032,20 @@ static void rpcReportBrokenLinkToServer(SRpcConn *pConn) {
   rpcAddRef(pRpc);
   tDebug("%s, notify the server app, connection is gone", pConn->info);
 
-  SRpcMsg rpcMsg;
-  rpcMsg.pCont = pConn->pReqMsg;     // pReqMsg is re-used to store the APP context from server
-  rpcMsg.contLen = pConn->reqMsgLen; // reqMsgLen is re-used to store the APP context length
-  rpcMsg.ahandle = pConn->ahandle;
-  rpcMsg.handle = pConn;
-  rpcMsg.msgType = pConn->inType;
-  rpcMsg.code = TSDB_CODE_RPC_NETWORK_UNAVAIL; 
+  SRpcMsg *rpcMsg = malloc(sizeof(SRpcMsg));
+  rpcMsg->pCont = pConn->pReqMsg;     // pReqMsg is re-used to store the APP context from server
+  rpcMsg->contLen = pConn->reqMsgLen; // reqMsgLen is re-used to store the APP context length
+  rpcMsg->ahandle = pConn->ahandle;
+  rpcMsg->handle = pConn;
+  rpcMsg->msgType = pConn->inType;
+  rpcMsg->code = TSDB_CODE_RPC_NETWORK_UNAVAIL; 
   pConn->pReqMsg = NULL;
   pConn->reqMsgLen = 0;
-  if (pRpc->cfp) (*(pRpc->cfp))(&rpcMsg, NULL);
+  if (pRpc->cfp) {
+    taosTmrStart(doRpcReportBrokenLinkToServer, 0, rpcMsg, pRpc->tmrCtrl);
+  } else {
+    free(rpcMsg);
+  }
 }
 
 static void rpcProcessBrokenLink(SRpcConn *pConn) {

src/sync/src/syncMain.c

@@ -410,7 +410,7 @@ void syncConfirmForward(int64_t rid, uint64_t version, int32_t code, bool force)
   syncReleaseNode(pNode);
 }
 
-#if 0
+#if 1
 void syncRecover(int64_t rid) {
   SSyncPeer *pPeer;
 

src/tsdb/src/tsdbCommit.c

@@ -17,7 +17,7 @@
 #define TSDB_MAX_SUBBLOCKS 8
 static FORCE_INLINE int TSDB_KEY_FID(TSKEY key, int32_t days, int8_t precision) {
   if (key < 0) {
-    return (int)(-((-key) / tsMsPerDay[precision] / days + 1));
+    return (int)((key + 1) / tsMsPerDay[precision] / days + 1);
   } else {
     return (int)((key / tsMsPerDay[precision] / days));
   }
@@ -452,6 +452,14 @@ static int tsdbCommitToFile(SCommitH *pCommith, SDFileSet *pSet, int fid) {
     return -1;
   }
 
+  if (tsdbUpdateDFileSetHeader(&(pCommith->wSet)) < 0) {
+    tsdbError("vgId:%d failed to update FSET %d header since %s", REPO_ID(pRepo), fid, tstrerror(terrno));
+    tsdbCloseCommitFile(pCommith, true);
+    // revert the file change
+    tsdbApplyDFileSetChange(TSDB_COMMIT_WRITE_FSET(pCommith), pSet);
+    return -1;
+  }
+
   // Close commit file
   tsdbCloseCommitFile(pCommith, false);
 

src/tsdb/src/tsdbFile.c

@@ -134,14 +134,14 @@ int tsdbCreateMFile(SMFile *pMFile, bool updateHeader) {
     return 0;
   }
 
+  pMFile->info.size += TSDB_FILE_HEAD_SIZE;
+
   if (tsdbUpdateMFileHeader(pMFile) < 0) {
     tsdbCloseMFile(pMFile);
     tsdbRemoveMFile(pMFile);
     return -1;
   }
 
-  pMFile->info.size += TSDB_FILE_HEAD_SIZE;
-
   return 0;
 }
 
@@ -378,14 +378,14 @@ int tsdbCreateDFile(SDFile *pDFile, bool updateHeader) {
     return 0;
   }
 
+  pDFile->info.size += TSDB_FILE_HEAD_SIZE;
+
   if (tsdbUpdateDFileHeader(pDFile) < 0) {
     tsdbCloseDFile(pDFile);
     tsdbRemoveDFile(pDFile);
     return -1;
   }
 
-  pDFile->info.size += TSDB_FILE_HEAD_SIZE;
-
   return 0;
 }
 

src/vnode/inc/vnodeMain.h

@@ -25,6 +25,7 @@ int32_t vnodeCreate(SCreateVnodeMsg *pVnodeCfg);
 int32_t vnodeDrop(int32_t vgId);
 int32_t vnodeOpen(int32_t vgId);
 int32_t vnodeAlter(void *pVnode, SCreateVnodeMsg *pVnodeCfg);
+int32_t vnodeSync(int32_t vgId);
 int32_t vnodeClose(int32_t vgId);
 void    vnodeCleanUp(SVnodeObj *pVnode);
 void    vnodeDestroy(SVnodeObj *pVnode);

src/vnode/src/vnodeMain.c

@@ -91,6 +91,23 @@ int32_t vnodeCreate(SCreateVnodeMsg *pVnodeCfg) {
   return code;
 }
 
+int32_t vnodeSync(int32_t vgId) {
+  SVnodeObj *pVnode = vnodeAcquire(vgId);
+  if (pVnode == NULL) {
+    vDebug("vgId:%d, failed to sync, vnode not find", vgId);
+    return TSDB_CODE_VND_INVALID_VGROUP_ID;
+  }
+
+  if (pVnode->role != TAOS_SYNC_ROLE_MASTER) {
+    vInfo("vgId:%d, vnode will sync, refCount:%d pVnode:%p", pVnode->vgId, pVnode->refCount, pVnode);
+    syncRecover(pVnode->sync);
+  }
+
+  vnodeRelease(pVnode);
+
+  return TSDB_CODE_SUCCESS;
+}
+
 int32_t vnodeDrop(int32_t vgId) {
   SVnodeObj *pVnode = vnodeAcquire(vgId);
   if (pVnode == NULL) {

tests/Jenkinsfile

 def pre_test(){
-    catchError(buildResult: 'SUCCESS', stageResult: 'FAILURE') {
+   
     sh '''
-                sudo rmtaos
+    sudo rmtaos||echo 'no taosd installed'
     '''
-    }
     sh '''
     cd ${WKC}
     git reset --hard
@@ -56,14 +55,8 @@ pipeline {
             cd ${WKC}/tests
             ./test-all.sh b1
             date'''
-            sh '''
-            cd ${WKC}/tests
-            ./test-all.sh full jdbc
-            date'''
-            sh '''
-            cd ${WKC}/tests
-            ./test-all.sh full unit
-            date'''
+            
+            
           }
         }
 
@@ -83,9 +76,20 @@ pipeline {
             catchError(buildResult: 'SUCCESS', stageResult: 'FAILURE') {
                 sh '''
                 cd ${WKC}/tests/pytest
+                rm -rf /var/lib/taos/*
+                rm -rf /var/log/taos/*
                 ./handle_crash_gen_val_log.sh
                 '''
             }
+            catchError(buildResult: 'SUCCESS', stageResult: 'FAILURE') {
+                sh '''
+                cd ${WKC}/tests/pytest
+                rm -rf /var/lib/taos/*
+                rm -rf /var/log/taos/*
+                ./handle_taosd_val_log.sh
+                '''
+            }
+            
             sh'''
             systemctl start taosd 
             sleep 10
@@ -136,6 +140,10 @@ pipeline {
             ./test-all.sh b2
             date
             '''
+            sh '''
+            cd ${WKC}/tests
+            ./test-all.sh full unit
+            date'''
           }
         }
 
@@ -154,6 +162,10 @@ pipeline {
                 '''
             }
             sh '''
+            cd ${WKC}/tests
+            ./test-all.sh full jdbc
+            date'''
+            sh '''
             cd ${WKC}/tests/pytest
             ./valgrind-test.sh 2>&1 > mem-error-out.log
             ./handle_val_log.sh
@@ -162,6 +174,11 @@ pipeline {
             cd ${WKC}/tests
             ./test-all.sh b3
             date'''
+            sh '''
+            date
+            cd ${WKC}/tests
+            ./test-all.sh full example
+            date'''
           }
         }
        

tests/examples/JDBC/taosdemo/src/main/resources/insert.json

@@ -38,7 +38,7 @@
           "insert_rows": 100,
           "multi_thread_write_one_tbl": "no",
           "number_of_tbl_in_one_sql": 0,
-          "rows_per_tbl": 3,
+          "interlace_rows": 3,
           "max_sql_len": 1024,
           "disorder_ratio": 0,
           "disorder_range": 1000,