Revert "Hotfix/sangshuduo/td 4240 taosdemo subscribe morethan 100 (#6155)"

This reverts commit e8d2162c.

src/client/inc/tscUtil.h

@@ -307,7 +307,7 @@ STableMeta* createSuperTableMeta(STableMetaMsg* pChild);
 uint32_t tscGetTableMetaSize(STableMeta* pTableMeta);
 CChildTableMeta* tscCreateChildMeta(STableMeta* pTableMeta);
 uint32_t tscGetTableMetaMaxSize();
-int32_t tscCreateTableMetaFromSTableMeta(STableMeta* pChild, const char* name, void* buf);
+int32_t tscCreateTableMetaFromCChildMeta(STableMeta* pChild, const char* name, void* buf);
 STableMeta* tscTableMetaDup(STableMeta* pTableMeta);
 int32_t tscCreateQueryFromQueryInfo(SQueryInfo* pQueryInfo, SQueryAttr* pQueryAttr, void* addr);
 

src/client/inc/tsclient.h

@@ -68,16 +68,14 @@ typedef struct CChildTableMeta {
   int32_t        vgId;
   STableId       id;
   uint8_t        tableType;
-  char           sTableName[TSDB_TABLE_FNAME_LEN];  // TODO: refactor super table name, not full name
-  uint64_t       suid;                              // super table id
+  char           sTableName[TSDB_TABLE_FNAME_LEN];  //super table name, not full name
 } CChildTableMeta;
 
 typedef struct STableMeta {
   int32_t        vgId;
   STableId       id;
   uint8_t        tableType;
-  char           sTableName[TSDB_TABLE_FNAME_LEN];  // super table name
-  uint64_t       suid;       // super table id
+  char           sTableName[TSDB_TABLE_FNAME_LEN];
   int16_t        sversion;
   int16_t        tversion;
   STableComInfo  tableInfo;

src/client/src/tscSchemaUtil.c

@@ -94,7 +94,6 @@ STableMeta* tscCreateTableMetaFromMsg(STableMetaMsg* pTableMetaMsg) {
 
   pTableMeta->tableType = pTableMetaMsg->tableType;
   pTableMeta->vgId      = pTableMetaMsg->vgroup.vgId;
-  pTableMeta->suid      = pTableMetaMsg->suid;
 
   pTableMeta->tableInfo = (STableComInfo) {
     .numOfTags    = pTableMetaMsg->numOfTags,

src/client/src/tscServer.c

@@ -1832,8 +1832,8 @@ int tscProcessTableMetaRsp(SSqlObj *pSql) {
   pMetaMsg->sversion = htons(pMetaMsg->sversion);
   pMetaMsg->tversion = htons(pMetaMsg->tversion);
   pMetaMsg->vgroup.vgId = htonl(pMetaMsg->vgroup.vgId);
+  
   pMetaMsg->uid = htobe64(pMetaMsg->uid);
-  pMetaMsg->suid        = pMetaMsg->suid;
   pMetaMsg->contLen = htons(pMetaMsg->contLen);
   pMetaMsg->numOfColumns = htons(pMetaMsg->numOfColumns);
   
@@ -2453,10 +2453,13 @@ int32_t tscGetTableMeta(SSqlObj *pSql, STableMetaInfo *pTableMetaInfo) {
       return TSDB_CODE_TSC_OUT_OF_MEMORY;
     }
     pTableMetaInfo->pTableMeta = (STableMeta *)tmp;
-  }
-
     memset(pTableMetaInfo->pTableMeta, 0, size);
     pTableMetaInfo->tableMetaSize = size;
+  } else {
+    //uint32_t s = tscGetTableMetaSize(pTableMetaInfo->pTableMeta);
+    memset(pTableMetaInfo->pTableMeta, 0, size);
+    pTableMetaInfo->tableMetaSize = size;
+  }
 
   pTableMetaInfo->pTableMeta->tableType = -1;
   pTableMetaInfo->pTableMeta->tableInfo.numOfColumns  = -1;
@@ -2473,9 +2476,8 @@ int32_t tscGetTableMeta(SSqlObj *pSql, STableMetaInfo *pTableMetaInfo) {
 
   STableMeta* pMeta = pTableMetaInfo->pTableMeta;
   if (pMeta->id.uid > 0) {
-    // in case of child table, here only get the
     if (pMeta->tableType == TSDB_CHILD_TABLE) {
-      int32_t code = tscCreateTableMetaFromSTableMeta(pTableMetaInfo->pTableMeta, name, buf);
+      int32_t code = tscCreateTableMetaFromCChildMeta(pTableMetaInfo->pTableMeta, name, buf);
       if (code != TSDB_CODE_SUCCESS) {
         return getTableMetaFromMnode(pSql, pTableMetaInfo);
       }

src/client/src/tscUtil.c

@@ -3370,25 +3370,22 @@ CChildTableMeta* tscCreateChildMeta(STableMeta* pTableMeta) {
   assert(pTableMeta != NULL);
 
   CChildTableMeta* cMeta = calloc(1, sizeof(CChildTableMeta));
-
   cMeta->tableType = TSDB_CHILD_TABLE;
   cMeta->vgId = pTableMeta->vgId;
   cMeta->id   = pTableMeta->id;
-  cMeta->suid      = pTableMeta->suid;
   tstrncpy(cMeta->sTableName, pTableMeta->sTableName, TSDB_TABLE_FNAME_LEN);
 
   return cMeta;
 }
 
-int32_t tscCreateTableMetaFromSTableMeta(STableMeta* pChild, const char* name, void* buf) {
+int32_t tscCreateTableMetaFromCChildMeta(STableMeta* pChild, const char* name, void* buf) {
   assert(pChild != NULL && buf != NULL);
 
-  STableMeta* p = buf;
-  taosHashGetClone(tscTableMetaInfo, pChild->sTableName, strnlen(pChild->sTableName, TSDB_TABLE_FNAME_LEN), NULL, p, -1);
+//  uint32_t size = tscGetTableMetaMaxSize();
+  STableMeta* p = buf;//calloc(1, size);
 
-  // tableMeta exists, build child table meta according to the super table meta
-  // the uid need to be checked in addition to the general name of the super table.
-  if (p->id.uid > 0 && pChild->suid == p->id.uid) {
+  taosHashGetClone(tscTableMetaInfo, pChild->sTableName, strnlen(pChild->sTableName, TSDB_TABLE_FNAME_LEN), NULL, p, -1);
+  if (p->id.uid > 0) { // tableMeta exists, build child table meta and return
     pChild->sversion = p->sversion;
     pChild->tversion = p->tversion;
 
@@ -3396,9 +3393,13 @@ int32_t tscCreateTableMetaFromSTableMeta(STableMeta* pChild, const char* name, v
     int32_t total = pChild->tableInfo.numOfColumns + pChild->tableInfo.numOfTags;
 
     memcpy(pChild->schema, p->schema, sizeof(SSchema) *total);
+
+//    tfree(p);
     return TSDB_CODE_SUCCESS;
   } else { // super table has been removed, current tableMeta is also expired. remove it here
     taosHashRemove(tscTableMetaInfo, name, strnlen(name, TSDB_TABLE_FNAME_LEN));
+
+//    tfree(p);
     return -1;
   }
 }

src/kit/taosdemo/taosdemo.c

@@ -526,8 +526,6 @@ static int taosRandom()
 
 #endif // ifdef Windows
 
-static void prompt();
-static void prompt2();
 static int createDatabasesAndStables();
 static void createChildTables();
 static int queryDbExec(TAOS *taos, char *command, QUERY_TYPE type, bool quiet);
@@ -1064,8 +1062,10 @@ static void parse_args(int argc, char *argv[], SArguments *arguments) {
     printf("# Print debug info:                  %d\n", arguments->debug_print);
     printf("# Print verbose info:                %d\n", arguments->verbose_print);
     printf("###################################################################\n");
-
-    prompt();
+    if (!arguments->answer_yes) {
+        printf("Press enter key to continue\n\n");
+        (void) getchar();
+    }
   }
 }
 
@@ -3446,7 +3446,10 @@ static bool getMetaFromInsertJsonFile(cJSON* root) {
               g_args.interlace_rows, g_args.num_of_RPR);
       printf("        interlace rows value will be set to num_of_records_per_req %"PRIu64"\n\n",
               g_args.num_of_RPR);
-      prompt2();
+      if (!g_args.answer_yes) {
+        printf("        press Enter key to continue or Ctrl-C to stop.");
+        (void)getchar();
+      }
       g_args.interlace_rows = g_args.num_of_RPR;
     }
   } else if (!interlaceRows) {
@@ -3503,9 +3506,9 @@ static bool getMetaFromInsertJsonFile(cJSON* root) {
       g_args.answer_yes = false;
     }
   } else if (!answerPrompt) {
-    g_args.answer_yes = true;   // default is no, mean answer_yes.
+    g_args.answer_yes = false;
   } else {
-    errorPrint("%s", "failed to read json, confirm_parameter_prompt input mistake\n");
+    printf("ERROR: failed to read json, confirm_parameter_prompt not found\n");
     goto PARSE_OVER;
   }
 
@@ -5392,7 +5395,6 @@ static void* syncWriteProgressive(threadInfo *pThreadInfo) {
     int64_t start_time = pThreadInfo->start_time;
 
     int64_t insertRows = (superTblInfo)?superTblInfo->insertRows:g_args.num_of_DPT;
-
     verbosePrint("%s() LN%d insertRows=%"PRId64"\n", __func__, __LINE__, insertRows);
 
     for (uint64_t i = 0; i < insertRows;) {
@@ -6093,21 +6095,6 @@ static void *readMetric(void *sarg) {
   return NULL;
 }
 
-static void prompt()
-{
-  if (!g_args.answer_yes) {
-    printf("Press enter key to continue\n\n");
-    (void)getchar();
-  }
-}
-
-static void prompt2()
-{
-    if (!g_args.answer_yes) {
-        printf("        press Enter key to continue or Ctrl-C to stop.");
-        (void)getchar();
-    }
-}
 
 static int insertTestProcess() {
 
@@ -6128,7 +6115,10 @@ static int insertTestProcess() {
   if (g_fpOfInsertResult)
     printfInsertMetaToFile(g_fpOfInsertResult);
 
-  prompt();
+  if (!g_args.answer_yes) {
+    printf("Press enter key to continue\n\n");
+    (void)getchar();
+  }
 
   init_rand_data();
 
@@ -6400,7 +6390,10 @@ static int queryTestProcess() {
             &g_queryInfo.superQueryInfo.childTblCount);
   }
 
-  prompt();
+  if (!g_args.answer_yes) {
+    printf("Press enter key to continue\n\n");
+    (void)getchar();
+  }
 
   if (g_args.debug_print || g_args.verbose_print) {
     printfQuerySystemInfo(taos);
@@ -6583,15 +6576,6 @@ static void *superSubscribe(void *sarg) {
   if (g_queryInfo.superQueryInfo.sqlCount == 0)
     return NULL;
 
-  if (g_queryInfo.superQueryInfo.sqlCount * pThreadInfo->ntables > MAX_QUERY_SQL_COUNT) {
-      errorPrint("The number %"PRId64" of sql count(%"PRIu64") multiple the table number(%"PRId64") of the thread is more than max query sql count: %d\n",
-              g_queryInfo.superQueryInfo.sqlCount * pThreadInfo->ntables,
-              g_queryInfo.superQueryInfo.sqlCount,
-              pThreadInfo->ntables,
-              MAX_QUERY_SQL_COUNT);
-      exit(-1);
-  }
-
   if (pThreadInfo->taos == NULL) {
     TAOS * taos = NULL;
     taos = taos_connect(g_queryInfo.host,
@@ -6762,7 +6746,10 @@ static int subscribeTestProcess() {
   printfQueryMeta();
   resetAfterAnsiEscape();
 
-  prompt();
+  if (!g_args.answer_yes) {
+    printf("Press enter key to continue\n\n");
+    (void) getchar();
+  }
 
   TAOS * taos = NULL;
   taos = taos_connect(g_queryInfo.host,

src/mnode/src/mnodeDb.c

@@ -389,7 +389,7 @@ static void mnodeSetDefaultDbCfg(SDbCfg *pCfg) {
   if (pCfg->compression < 0) pCfg->compression = tsCompression;
   if (pCfg->walLevel < 0) pCfg->walLevel = tsWAL;
   if (pCfg->replications < 0) pCfg->replications = tsReplications;
-  if (pCfg->quorum < 0) pCfg->quorum = MIN(tsQuorum, pCfg->replications);
+  if (pCfg->quorum < 0) pCfg->quorum = tsQuorum;
   if (pCfg->update < 0) pCfg->update = tsUpdate;
   if (pCfg->cacheLastRow < 0) pCfg->cacheLastRow = tsCacheLastRow;
   if (pCfg->dbType < 0) pCfg->dbType = 0;

src/vnode/src/vnodeMain.c

@@ -120,14 +120,12 @@ int32_t vnodeDrop(int32_t vgId) {
     vDebug("vgId:%d, failed to drop, vnode not find", vgId);
     return TSDB_CODE_VND_INVALID_VGROUP_ID;
   }
-  if (pVnode->dropped) {
-    vnodeRelease(pVnode);
-    return TSDB_CODE_SUCCESS;
-  }
 
   vInfo("vgId:%d, vnode will be dropped, refCount:%d pVnode:%p", pVnode->vgId, pVnode->refCount, pVnode);
   pVnode->dropped = 1;
 
+  // remove from hash, so new messages wont be consumed
+  vnodeRemoveFromHash(pVnode);
   vnodeRelease(pVnode);
   vnodeCleanupInMWorker(pVnode);
 
@@ -427,10 +425,6 @@ int32_t vnodeOpen(int32_t vgId) {
 int32_t vnodeClose(int32_t vgId) {
   SVnodeObj *pVnode = vnodeAcquire(vgId);
   if (pVnode == NULL) return 0;
-  if (pVnode->dropped) {
-    vnodeRelease(pVnode);
-    return 0;
-  }
 
   vDebug("vgId:%d, vnode will be closed, pVnode:%p", pVnode->vgId, pVnode);
   vnodeRemoveFromHash(pVnode);
@@ -516,8 +510,6 @@ void vnodeCleanUp(SVnodeObj *pVnode) {
 
   vnodeSetClosingStatus(pVnode);
 
-  vnodeRemoveFromHash(pVnode);
-
   // stop replication module
   if (pVnode->sync > 0) {
     int64_t sync = pVnode->sync;

src/vnode/src/vnodeRead.c

@@ -117,17 +117,14 @@ static SVReadMsg *vnodeBuildVReadMsg(SVnodeObj *pVnode, void *pCont, int32_t con
 }
 
 int32_t vnodeWriteToRQueue(void *vparam, void *pCont, int32_t contLen, int8_t qtype, void *rparam) {
-  SVnodeObj *pVnode = vparam;
-  if (pVnode->dropped) {
-    return TSDB_CODE_APP_NOT_READY;
-  }
-
   SVReadMsg *pRead = vnodeBuildVReadMsg(vparam, pCont, contLen, qtype, rparam);
   if (pRead == NULL) {
     assert(terrno != 0);
     return terrno;
   }
 
+  SVnodeObj *pVnode = vparam;
+
   int32_t code = vnodeCheckRead(pVnode);
   if (code != TSDB_CODE_SUCCESS) {
     taosFreeQitem(pRead);

src/vnode/src/vnodeStatus.c

@@ -66,9 +66,6 @@ static bool vnodeSetClosingStatusImp(SVnodeObj* pVnode) {
 }
 
 bool vnodeSetClosingStatus(SVnodeObj* pVnode) {
-  if (pVnode->status == TAOS_VN_STATUS_CLOSING)
-    return true;
-
   while (!vnodeSetClosingStatusImp(pVnode)) {
     taosMsleep(1);
   }

src/vnode/src/vnodeSync.c

@@ -55,11 +55,6 @@ void vnodeNotifyRole(int32_t vgId, int8_t role) {
     vTrace("vgId:%d, vnode not found while notify role", vgId);
     return;
   }
-  if (pVnode->dropped) {
-    vTrace("vgId:%d, vnode dropped while notify role", vgId);
-    vnodeRelease(pVnode);
-    return;
-  }
 
   vInfo("vgId:%d, sync role changed from %s to %s", pVnode->vgId, syncRole[pVnode->role], syncRole[role]);
   pVnode->role = role;
@@ -80,11 +75,6 @@ void vnodeCtrlFlow(int32_t vgId, int32_t level) {
     vTrace("vgId:%d, vnode not found while flow ctrl", vgId);
     return;
   }
-  if (pVnode->dropped) {
-    vTrace("vgId:%d, vnode dropped while flow ctrl", vgId);
-    vnodeRelease(pVnode);
-    return;
-  }
 
   if (pVnode->flowctrlLevel != level) {
     vDebug("vgId:%d, set flowctrl level from %d to %d", pVnode->vgId, pVnode->flowctrlLevel, level);
@@ -139,7 +129,6 @@ int32_t vnodeWriteToCache(int32_t vgId, void *wparam, int32_t qtype, void *rpara
   SVnodeObj *pVnode = vnodeAcquire(vgId);
   if (pVnode == NULL) {
     vError("vgId:%d, vnode not found while write to cache", vgId);
-    vnodeRelease(pVnode);
     return TSDB_CODE_VND_INVALID_VGROUP_ID;
   }
 

src/vnode/src/vnodeWrite.c

@@ -386,6 +386,4 @@ void vnodeWaitWriteCompleted(SVnodeObj *pVnode) {
     vTrace("vgId:%d, queued wmsg num:%d", pVnode->vgId, pVnode->queuedWMsg);
     taosMsleep(10);
   }
-
-  taosMsleep(900);
 }

tests/pytest/tools/taosdemoTestWithJson.py

@@ -27,28 +27,28 @@ class TDTestCase:
     def getBuildPath(self):
         selfPath = os.path.dirname(os.path.realpath(__file__))
 
-        if "community" in selfPath:
-            projPath = selfPath[: selfPath.find("community")]
+        if ("community" in selfPath):
+            projPath = selfPath[:selfPath.find("community")]
         else:
-            projPath = selfPath[: selfPath.find("tests")]
+            projPath = selfPath[:selfPath.find("tests")]
 
         for root, dirs, files in os.walk(projPath):
-            if "taosd" in files:
+            if ("taosd" in files):
                 rootRealPath = os.path.dirname(os.path.realpath(root))
-                if "packaging" not in rootRealPath:
-                    buildPath = root[: len(root) - len("/build/bin")]
+                if ("packaging" not in rootRealPath):
+                    buildPath = root[:len(root)-len("/build/bin")]
                     break
         return buildPath
         
     def run(self):
         tdSql.prepare()
         buildPath = self.getBuildPath()
-        if buildPath == "":
+        if (buildPath == ""):
             tdLog.exit("taosd not found!")
         else:
             tdLog.info("taosd found in %s" % buildPath)
-        binPath = buildPath + "/build/bin/"
-        os.system("%staosdemo -f tools/insert.json -y" % binPath)
+        binPath = buildPath+ "/build/bin/"
+        os.system("yes | %staosdemo -f tools/insert.json" % binPath)
 
         tdSql.execute("use db01")
         tdSql.query("select count(*) from stb01")

tests/script/general/parser/commit.sim

@@ -68,7 +68,7 @@ while $loop <= $loops
   while $i < 10
     sql select count(*) from $stb where t1 = $i 
     if $data00 != $rowNum then
-      print expect $rowNum , actual: $data00
+      print expect $rowNum, actual: $data00
       return -1
     endi
     $i = $i + 1