Merge pull request #9648 from taosdata/feature/3.0_liaohj

Feature/3.0 liaohj

source/client/test/clientTests.cpp

@@ -152,7 +152,7 @@ TEST(testCase, create_db_Test) {
   TAOS* pConn = taos_connect("localhost", "root", "taosdata", NULL, 0);
   assert(pConn != NULL);
 
-  TAOS_RES* pRes = taos_query(pConn, "create database abc1");
+  TAOS_RES* pRes = taos_query(pConn, "create database abc1 vgroups 2");
   if (taos_errno(pRes) != 0) {
     printf("error in create db, reason:%s\n", taos_errstr(pRes));
   }
@@ -254,7 +254,7 @@ TEST(testCase, use_db_test) {
   TAOS* pConn = taos_connect("localhost", "root", "taosdata", NULL, 0);
   assert(pConn != NULL);
 
-  TAOS_RES* pRes = taos_query(pConn, "create database abc1");
+  TAOS_RES* pRes = taos_query(pConn, "create database abc1 vgroups 2");
   if (taos_errno(pRes) != 0) {
     printf("error in create db, reason:%s\n", taos_errstr(pRes));
   }
@@ -505,15 +505,17 @@ TEST(testCase, create_multiple_tables) {
 
   taos_free_result(pRes);
 
-//  for(int32_t i = 0; i < 1000; ++i) {
-//    char sql[512] = {0};
-//    snprintf(sql, tListLen(sql), "create table t_x_%d using st1 tags(2)", i);
-//    TAOS_RES* pres = taos_query(pConn, sql);
-//    if (taos_errno(pres) != 0) {
-//      printf("failed to create table %d\n, reason:%s", i, taos_errstr(pres));
-//    }
-//    taos_free_result(pres);
-//  }
+  for(int32_t i = 0; i < 200000; ++i) {
+    char sql[512] = {0};
+    snprintf(sql, tListLen(sql), "create table t_x_%d using st1 tags(2)", i);
+    TAOS_RES* pres = taos_query(pConn, sql);
+    if (taos_errno(pres) != 0) {
+      printf("failed to create table %d\n, reason:%s", i, taos_errstr(pres));
+    }
+
+    printf("%d\n", i);
+    taos_free_result(pres);
+  }
 
   taos_close(pConn);
 }

source/libs/catalog/src/catalog.c

@@ -310,7 +310,7 @@ int32_t ctgGetVgInfoFromHashValue(SDBVgroupInfo *dbInfo, const SName *pTableName
   }
 
   if (NULL == vgInfo) {
-    ctgError("no hash range found for hashvalue [%u], numOfVgId:%d", hashValue, taosHashGetSize(dbInfo->vgInfo));
+    ctgError("no hash range found for hash value [%u], numOfVgId:%d", hashValue, taosHashGetSize(dbInfo->vgInfo));
 
     void *pIter1 = taosHashIterate(dbInfo->vgInfo, NULL);
     while (pIter1) {

source/libs/scheduler/src/scheduler.c

@@ -273,19 +273,20 @@ int32_t schSetTaskCandidateAddrs(SSchJob *job, SSchTask *task) {
   return TSDB_CODE_SUCCESS;
 }
 
-
-int32_t schPushTaskToExecList(SSchJob *job, SSchTask *task) {
-  if (0 != taosHashPut(job->execTasks, &task->taskId, sizeof(task->taskId), &task, POINTER_BYTES)) {
-    qError("taosHashPut failed");
+int32_t schPushTaskToExecList(SSchJob *pJob, SSchTask *pTask) {
+  if (0 != taosHashPut(pJob->execTasks, &pTask->taskId, sizeof(pTask->taskId), &pTask, POINTER_BYTES)) {
+    qError("failed to add new task, taskId:0x%"PRIx64", reqId:0x"PRIx64", out of memory", pJob->queryId);
     SCH_ERR_RET(TSDB_CODE_QRY_OUT_OF_MEMORY);
   }
 
+  qDebug("add one task, taskId:0x%"PRIx64", numOfTasks:%d, reqId:0x%"PRIx64, pTask->taskId, taosHashGetSize(pJob->execTasks),
+         pJob->queryId);
   return TSDB_CODE_SUCCESS;
 }
 
 int32_t schMoveTaskToSuccList(SSchJob *job, SSchTask *task, bool *moved) {
   if (0 != taosHashRemove(job->execTasks, &task->taskId, sizeof(task->taskId))) {
-    qWarn("remove task[%"PRIx64"] from execTasks failed", task->taskId);
+    qError("remove task taskId:0x%"PRIx64" from execTasks failed, reqId:0x%"PRIx64, task->taskId, job->queryId);
     return TSDB_CODE_SUCCESS;
   }
 
@@ -583,9 +584,12 @@ int32_t schHandleCallback(void* param, const SDataBuf* pMsg, int32_t msgType, in
     SCH_ERR_JRET(TSDB_CODE_SCH_INTERNAL_ERROR);
   }
 
+  int32_t s = taosHashGetSize((*job)->execTasks);
+  assert(s != 0);
+
   SSchTask **task = taosHashGet((*job)->execTasks, &pParam->taskId, sizeof(pParam->taskId));
   if (NULL == task || NULL == (*task)) {
-    qError("taosHashGet taskId:%"PRIx64" not exist", pParam->taskId);
+    qError("failed to get task, taskId:%"PRIx64" not exist, reqId:0x%"PRIx64, pParam->taskId, (*job)->queryId);
     SCH_ERR_JRET(TSDB_CODE_SCH_INTERNAL_ERROR);
   }
   
@@ -819,12 +823,13 @@ int32_t schLaunchTask(SSchJob *job, SSchTask *task) {
   SCH_ERR_RET(schSetTaskCandidateAddrs(job, task));
 
   if (NULL == task->candidateAddrs || taosArrayGetSize(task->candidateAddrs) <= 0) {
-    SCH_TASK_ERR_LOG("no valid condidate node for task:%"PRIx64, task->taskId);
+    SCH_TASK_ERR_LOG("no valid candidate node for task:%"PRIx64, task->taskId);
     SCH_ERR_RET(TSDB_CODE_SCH_INTERNAL_ERROR);
   }
 
-  SCH_ERR_RET(schBuildAndSendMsg(job, task, plan->msgType));
+  // NOTE: race condition: the task should be put into the hash table before send msg to server
   SCH_ERR_RET(schPushTaskToExecList(job, task));
+  SCH_ERR_RET(schBuildAndSendMsg(job, task, plan->msgType));
 
   task->status = JOB_TASK_STATUS_EXECUTING;
   return TSDB_CODE_SUCCESS;
@@ -975,7 +980,7 @@ _return:
 }
 
 int32_t scheduleExecJob(void *transport, SArray *nodeList, SQueryDag* pDag, void** pJob, SQueryResult *pRes) {
-  if (NULL == transport || /* NULL == nodeList || */ NULL == pDag || NULL == pDag->pSubplans || NULL == pJob || NULL == pRes) {
+  if (NULL == transport || NULL == pDag || NULL == pDag->pSubplans || NULL == pJob || NULL == pRes) {
     SCH_ERR_RET(TSDB_CODE_QRY_INVALID_INPUT);
   }