diff --git a/include/dnode/snode/snode.h b/include/dnode/snode/snode.h
index 319708decc4e41abf60ec4358a740dc42f59ea21..c9fab140cce647f22827648404b93d507a70e4c1 100644
--- a/include/dnode/snode/snode.h
+++ b/include/dnode/snode/snode.h
@@ -16,6 +16,9 @@
 #ifndef _TD_SNODE_H_
 #define _TD_SNODE_H_
 
+#include "tmsg.h"
+#include "trpc.h"
+
 #ifdef __cplusplus
 extern "C" {
 #endif
@@ -88,4 +91,4 @@ void sndDestroy(const char *path);
 }
 #endif
 
-#endif /*_TD_SNODE_H_*/
\ No newline at end of file
+#endif /*_TD_SNODE_H_*/
diff --git a/source/dnode/mnode/impl/inc/mndDef.h b/source/dnode/mnode/impl/inc/mndDef.h
index 91d37c4eedf5946432c2d366af46f36d08df2ec1..05f7de60ff948b708ab8a34215e255aa038ac2f6 100644
--- a/source/dnode/mnode/impl/inc/mndDef.h
+++ b/source/dnode/mnode/impl/inc/mndDef.h
@@ -661,6 +661,9 @@ static FORCE_INLINE void* tDecodeSMqConsumerObj(void* buf, SMqConsumerObj* pCons
   return buf;
 }
 
+typedef struct {
+} SStreamScheduler;
+
 typedef struct SMnodeMsg {
   char    user[TSDB_USER_LEN];
   char    db[TSDB_DB_FNAME_LEN];
diff --git a/source/dnode/mnode/impl/inc/mndScheduler.h b/source/dnode/mnode/impl/inc/mndScheduler.h
new file mode 100644
index 0000000000000000000000000000000000000000..3bf6e0c33a6a07c633c38323c133099fed62e17b
--- /dev/null
+++ b/source/dnode/mnode/impl/inc/mndScheduler.h
@@ -0,0 +1,34 @@
+/*
+ * Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
+ *
+ * This program is free software: you can use, redistribute, and/or modify
+ * it under the terms of the GNU Affero General Public License, version 3
+ * or later ("AGPL"), as published by the Free Software Foundation.
+ *
+ * This program is distributed in the hope that it will be useful, but WITHOUT
+ * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
+ * FITNESS FOR A PARTICULAR PURPOSE.
+ *
+ * You should have received a copy of the GNU Affero General Public License
+ * along with this program. If not, see <http://www.gnu.org/licenses/>.
+ */
+
+#ifndef _TD_MND_SCHEDULER_H_
+#define _TD_MND_SCHEDULER_H_
+
+#include "mndInt.h"
+
+#ifdef __cplusplus
+extern "C" {
+#endif
+
+int32_t mndInitScheduler(SMnode* pMnode);
+void    mndCleanupScheduler(SMnode* pMnode);
+
+int32_t mndSchedInitSubEp(SMnode* pMnode, const SMqTopicObj* pTopic, SMqSubscribeObj* pSub);
+
+#ifdef __cplusplus
+}
+#endif
+
+#endif /*_TD_MND_SCHEDULER_H_ */
diff --git a/source/dnode/mnode/impl/src/mndDnode.c b/source/dnode/mnode/impl/src/mndDnode.c
index b872e933d5909d878bc55b62e1156e76d1413901..e231717e61de7ea717cdda2d4b686610bf1dd632 100644
--- a/source/dnode/mnode/impl/src/mndDnode.c
+++ b/source/dnode/mnode/impl/src/mndDnode.c
@@ -340,9 +340,16 @@ static int32_t mndProcessStatusReq(SMnodeMsg *pReq) {
         pVgroup->compStorage = pVload->compStorage;
         pVgroup->pointsWritten = pVload->pointsWritten;
       }
+      bool roleChanged = false;
       for (int32_t vg = 0; vg < pVgroup->replica; ++vg) {
+        if (pVgroup->vnodeGid[vg].role != pVload->role) {
+          roleChanged = true;
+        }
         pVgroup->vnodeGid[vg].role = pVload->role;
       }
+      if (roleChanged) {
+        // notify scheduler role has changed
+      }
     }
 
     mndReleaseVgroup(pMnode, pVgroup);
@@ -631,13 +638,13 @@ static int32_t mndGetConfigMeta(SMnodeMsg *pReq, SShowObj *pShow, STableMetaRsp
 
   pShow->bytes[cols] = TSDB_CONFIG_OPTION_LEN + VARSTR_HEADER_SIZE;
   pSchema[cols].type = TSDB_DATA_TYPE_BINARY;
-    strcpy(pSchema[cols].name, "name");
+  strcpy(pSchema[cols].name, "name");
   pSchema[cols].bytes = pShow->bytes[cols];
   cols++;
 
   pShow->bytes[cols] = TSDB_CONIIG_VALUE_LEN + VARSTR_HEADER_SIZE;
   pSchema[cols].type = TSDB_DATA_TYPE_BINARY;
-   strcpy(pSchema[cols].name, "value");
+  strcpy(pSchema[cols].name, "value");
   pSchema[cols].bytes = pShow->bytes[cols];
   cols++;
 
@@ -823,4 +830,4 @@ static int32_t mndRetrieveDnodes(SMnodeMsg *pReq, SShowObj *pShow, char *data, i
 static void mndCancelGetNextDnode(SMnode *pMnode, void *pIter) {
   SSdb *pSdb = pMnode->pSdb;
   sdbCancelFetch(pSdb, pIter);
-}
\ No newline at end of file
+}
diff --git a/source/dnode/mnode/impl/src/mndScheduler.c b/source/dnode/mnode/impl/src/mndScheduler.c
new file mode 100644
index 0000000000000000000000000000000000000000..9b301ad98a3908af6c82f92759722dabc08ab0aa
--- /dev/null
+++ b/source/dnode/mnode/impl/src/mndScheduler.c
@@ -0,0 +1,73 @@
+/*
+ * 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/>.
+ */
+
+#include "mndScheduler.h"
+#include "mndConsumer.h"
+#include "mndDb.h"
+#include "mndDnode.h"
+#include "mndMnode.h"
+#include "mndOffset.h"
+#include "mndShow.h"
+#include "mndStb.h"
+#include "mndSubscribe.h"
+#include "mndTopic.h"
+#include "mndTrans.h"
+#include "mndUser.h"
+#include "mndVgroup.h"
+#include "tcompare.h"
+#include "tname.h"
+
+int32_t mndSchedInitSubEp(SMnode* pMnode, const SMqTopicObj* pTopic, SMqSubscribeObj* pSub) {
+  SSdb*      pSdb = pMnode->pSdb;
+  SVgObj*    pVgroup = NULL;
+  SQueryDag* pDag = qStringToDag(pTopic->physicalPlan);
+  SArray*    pAray = NULL;
+  SArray*    unassignedVg = pSub->unassignedVg;
+
+  ASSERT(pSub->vgNum == 0);
+
+  int32_t levelNum = taosArrayGetSize(pDag->pSubplans);
+  if (levelNum != 1) {
+    return -1;
+  }
+
+  SArray*   inner = taosArrayGet(pDag->pSubplans, 0);
+  SSubplan* plan = taosArrayGetP(inner, 0);
+
+  void* pIter = NULL;
+  while (1) {
+    pIter = sdbFetch(pSdb, SDB_VGROUP, pIter, (void**)&pVgroup);
+    if (pIter == NULL) break;
+    if (pVgroup->dbUid != pTopic->dbUid) {
+      sdbRelease(pSdb, pVgroup);
+      continue;
+    }
+
+    pSub->vgNum++;
+    plan->execNode.nodeId = pVgroup->vgId;
+    plan->execNode.epset = mndGetVgroupEpset(pMnode, pVgroup);
+
+    SMqConsumerEp consumerEp = {0};
+    consumerEp.status = 0;
+    consumerEp.consumerId = -1;
+    consumerEp.epSet = plan->execNode.epset;
+    consumerEp.vgId = plan->execNode.nodeId;
+    int32_t msgLen;
+    int32_t code = qSubPlanToString(plan, &consumerEp.qmsg, &msgLen);
+    taosArrayPush(unassignedVg, &consumerEp);
+  }
+
+  return 0;
+}
diff --git a/source/dnode/mnode/impl/src/mndSubscribe.c b/source/dnode/mnode/impl/src/mndSubscribe.c
index 60533b979cc1f210eb0de67cb9dd3147df875b38..84360ab23b58bfa20c729e8f404465332e8b2d56 100644
--- a/source/dnode/mnode/impl/src/mndSubscribe.c
+++ b/source/dnode/mnode/impl/src/mndSubscribe.c
@@ -20,6 +20,7 @@
 #include "mndDnode.h"
 #include "mndMnode.h"
 #include "mndOffset.h"
+#include "mndScheduler.h"
 #include "mndShow.h"
 #include "mndStb.h"
 #include "mndTopic.h"
@@ -97,12 +98,21 @@ static SMqSubscribeObj *mndCreateSubscription(SMnode *pMnode, const SMqTopicObj
   strcpy(pSub->key, key);
   free(key);
 
+  if (mndSchedInitSubEp(pMnode, pTopic, pSub) < 0) {
+    terrno = TSDB_CODE_MND_UNSUPPORTED_TOPIC;
+    tDeleteSMqSubscribeObj(pSub);
+    free(pSub);
+    return NULL;
+  }
+
+#if 0
   if (mndInitUnassignedVg(pMnode, pTopic, pSub) < 0) {
     terrno = TSDB_CODE_OUT_OF_MEMORY;
     tDeleteSMqSubscribeObj(pSub);
     free(pSub);
     return NULL;
   }
+#endif
   // TODO: disable alter subscribed table
   return pSub;
 }
diff --git a/source/dnode/snode/src/snode.c b/source/dnode/snode/src/snode.c
index 5d21a037e7211960722ee2a341142507537920e1..01500fbc54a19a01559005c313fbcdb40970ebcb 100644
--- a/source/dnode/snode/src/snode.c
+++ b/source/dnode/snode/src/snode.c
@@ -17,6 +17,7 @@
 
 SSnode *sndOpen(const char *path, const SSnodeOpt *pOption) {
   SSnode *pSnode = calloc(1, sizeof(SSnode));
+  memcpy(&pSnode->cfg, pOption, sizeof(SSnodeOpt));
   return pSnode;
 }
 
@@ -29,4 +30,4 @@ int32_t sndProcessMsg(SSnode *pSnode, SRpcMsg *pMsg, SRpcMsg **pRsp) {
   return 0;
 }
 
-void sndDestroy(const char *path) {}
\ No newline at end of file
+void sndDestroy(const char *path) {}
diff --git a/source/dnode/vnode/src/tq/tq.c b/source/dnode/vnode/src/tq/tq.c
index ccad006657772400fe7173a8a5a53fcc44362fd4..d540b589b71d80d22d85e765960f7d35ce9b2656 100644
--- a/source/dnode/vnode/src/tq/tq.c
+++ b/source/dnode/vnode/src/tq/tq.c
@@ -376,7 +376,10 @@ int32_t tqProcessSetConnReq(STQ* pTq, char* msg) {
   for (int i = 0; i < TQ_BUFFER_SIZE; i++) {
     pTopic->buffer.output[i].status = 0;
     STqReadHandle* pReadHandle = tqInitSubmitMsgScanner(pTq->pVnodeMeta);
-    SReadHandle    handle = {.reader = pReadHandle, .meta = pTq->pVnodeMeta};
+    SReadHandle    handle = {
+           .reader = pReadHandle,
+           .meta = pTq->pVnodeMeta,
+    };
     pTopic->buffer.output[i].pReadHandle = pReadHandle;
     pTopic->buffer.output[i].task = qCreateStreamExecTaskInfo(req.qmsg, &handle);
   }
diff --git a/source/dnode/vnode/src/tq/tqRead.c b/source/dnode/vnode/src/tq/tqRead.c
index bfa811feecf7a33037152c2ed41110b2fa60f49b..2f24df03099a728afe30c1b8383e68e2b0face62 100644
--- a/source/dnode/vnode/src/tq/tqRead.c
+++ b/source/dnode/vnode/src/tq/tqRead.c
@@ -28,6 +28,7 @@ STqReadHandle* tqInitSubmitMsgScanner(SMeta* pMeta) {
   pReadHandle->sver = -1;
   pReadHandle->pSchema = NULL;
   pReadHandle->pSchemaWrapper = NULL;
+  pReadHandle->tbIdHash = NULL;
   return pReadHandle;
 }