fix td-735

src/client/src/tscSub.c

@@ -308,11 +308,9 @@ void tscSaveSubscriptionProgress(void* sub) {
 
   char path[256];
   sprintf(path, "%s/subscribe", tsDataDir);
-  if (access(path, 0) != 0) {
-    if (mkdir(path, 0777) != 0 && errno != EEXIST) {
+  if (tmkdir(path, 0777) != 0) {
     tscError("failed to create subscribe dir: %s", path);
   }
-  }
 
   sprintf(path, "%s/subscribe/%s", tsDataDir, pSub->topic);
   FILE* fp = fopen(path, "w+");

src/tsdb/src/tsdbMeta.c

@@ -788,6 +788,9 @@ static int tsdbAddTableToMeta(STsdbRepo *pRepo, STable *pTable, bool addIdx) {
   }
 
   if (addIdx && tsdbUnlockRepoMeta(pRepo) < 0) return -1;
+  if (TABLE_TYPE(pTable) == TSDB_STREAM_TABLE && addIdx) {
+    pTable->cqhandle = (*pRepo->appH.cqCreateFunc)(pRepo->appH.cqH, TABLE_UID(pTable), TABLE_TID(pTable), pTable->sql, tsdbGetTableSchema(pTable));
+  }
 
   tsdbTrace("vgId:%d table %s tid %d uid %" PRIu64 " is added to meta", REPO_ID(pRepo), TABLE_CHAR_NAME(pTable),
             TABLE_TID(pTable), TABLE_UID(pTable));

src/tsdb/src/tsdbRead.c

@@ -193,7 +193,7 @@ TsdbQueryHandleT* tsdbQueryTables(TSDB_REPO_T* tsdb, STsdbQueryCond* pCond, STab
       };
       
       assert(info.pTableObj != NULL && (info.pTableObj->type == TSDB_NORMAL_TABLE ||
-      info.pTableObj->type == TSDB_CHILD_TABLE));
+      info.pTableObj->type == TSDB_CHILD_TABLE || info.pTableObj->type == TSDB_STREAM_TABLE));
 
       taosArrayPush(pQueryHandle->pTableCheckInfo, &info);
     }
@@ -2236,7 +2236,7 @@ int32_t tsdbGetOneTableGroup(TSDB_REPO_T* tsdb, uint64_t uid, STableGroupInfo* p
     goto _error;
   }
 
-  assert(pTable->type == TSDB_CHILD_TABLE || pTable->type == TSDB_NORMAL_TABLE);
+  assert(pTable->type == TSDB_CHILD_TABLE || pTable->type == TSDB_NORMAL_TABLE || pTable->type == TSDB_STREAM_TABLE);
   tsdbRefTable(pTable);
   if (tsdbUnlockRepoMeta(tsdb) < 0) goto _error;