fix crash

source/dnode/mnode/impl/src/mndScheduler.c

@@ -47,7 +47,7 @@ int32_t mndSchedInitSubEp(SMnode* pMnode, const SMqTopicObj* pTopic, SMqSubscrib
     return -1;
   }
 
-  SArray* plans = taosArrayGet(pDag->pSubplans, 0);
+  SArray* plans = taosArrayGetP(pDag->pSubplans, 0);
 
   int32_t opNum = taosArrayGetSize(plans);
   if (opNum != 1) {
@@ -85,7 +85,7 @@ int32_t mndSchedInitSubEp(SMnode* pMnode, const SMqTopicObj* pTopic, SMqSubscrib
     taosArrayPush(pSub->unassignedVg, &consumerEp);
   }
 
-  /*qDestroyQueryDag(pDag);*/
+  qDestroyQueryDag(pDag);
 
   return 0;
 }

source/libs/planner/src/physicalPlanJson.c

@@ -13,9 +13,9 @@
  * along with this program. If not, see <http://www.gnu.org/licenses/>.
  */
 
-#include "plannerInt.h"
-#include "parser.h"
 #include "cJSON.h"
+#include "parser.h"
+#include "plannerInt.h"
 
 typedef bool (*FToJson)(const void* obj, cJSON* json);
 typedef bool (*FFromJson)(const cJSON* json, void* obj);
@@ -39,7 +39,7 @@ static uint64_t getBigintFromString(const cJSON* json, const char* name) {
 
 static int64_t getNumber(const cJSON* json, const char* name) {
   double d = cJSON_GetNumberValue(cJSON_GetObjectItem(json, name));
-  return (int64_t) d;
+  return (int64_t)d;
 }
 
 static bool addObject(cJSON* json, const char* name, FToJson func, const void* obj) {
@@ -72,7 +72,8 @@ static bool fromObject(const cJSON* json, const char* name, FFromJson func, void
   return func(jObj, obj);
 }
 
-static bool fromObjectWithAlloc(const cJSON* json, const char* name, FFromJson func, void** obj, int32_t size, bool required) {
+static bool fromObjectWithAlloc(const cJSON* json, const char* name, FFromJson func, void** obj, int32_t size,
+                                bool required) {
   cJSON* jObj = cJSON_GetObjectItem(json, name);
   if (NULL == jObj) {
     return !required;
@@ -161,7 +162,8 @@ static bool addArray(cJSON* json, const char* name, FToJson func, const SArray*
   return addTarray(json, name, func, array, true);
 }
 
-static bool fromTarray(const cJSON* json, const char* name, FFromJson func, SArray** array, int32_t itemSize, bool isPoint) {
+static bool fromTarray(const cJSON* json, const char* name, FFromJson func, SArray** array, int32_t itemSize,
+                       bool isPoint) {
   const cJSON* jArray = cJSON_GetObjectItem(json, name);
   int32_t      size = (NULL == jArray ? 0 : cJSON_GetArraySize(jArray));
   if (size > 0) {
@@ -188,7 +190,8 @@ static bool fromArray(const cJSON* json, const char* name, FFromJson func, SArra
   return fromTarray(json, name, func, array, itemSize, true);
 }
 
-static bool addRawArray(cJSON* json, const char* name, FToJson func, const void* array, int32_t itemSize, int32_t size) {
+static bool addRawArray(cJSON* json, const char* name, FToJson func, const void* array, int32_t itemSize,
+                        int32_t size) {
   if (size > 0) {
     cJSON* jArray = cJSON_AddArrayToObject(json, name);
     if (NULL == jArray) {
@@ -218,7 +221,8 @@ static bool fromItem(const cJSON* jArray, FFromJson func, void* array, int32_t i
   return true;
 }
 
-static bool fromRawArrayWithAlloc(const cJSON* json, const char* name, FFromJson func, void** array, int32_t itemSize, int32_t* size) {
+static bool fromRawArrayWithAlloc(const cJSON* json, const char* name, FFromJson func, void** array, int32_t itemSize,
+                                  int32_t* size) {
   const cJSON* jArray = getArray(json, name, size);
   if (*size > 0) {
     *array = calloc(1, itemSize * (*size));
@@ -229,7 +233,8 @@ static bool fromRawArrayWithAlloc(const cJSON* json, const char* name, FFromJson
   return fromItem(jArray, func, *array, itemSize, *size);
 }
 
-static bool fromRawArray(const cJSON* json, const char* name, FFromJson func, void* array, int32_t itemSize, int32_t* size) {
+static bool fromRawArray(const cJSON* json, const char* name, FFromJson func, void* array, int32_t itemSize,
+                         int32_t* size) {
   const cJSON* jArray = getArray(json, name, size);
   return fromItem(jArray, func, array, itemSize, *size);
 }
@@ -264,7 +269,8 @@ static const char* jkDataBlockSchemaPrecision = "Precision";
 
 static bool dataBlockSchemaToJson(const void* obj, cJSON* json) {
   const SDataBlockSchema* schema = (const SDataBlockSchema*)obj;
-  bool res = addRawArray(json, jkDataBlockSchemaSlotSchema, schemaToJson, schema->pSchema, sizeof(SSlotSchema), schema->numOfCols);
+  bool res = addRawArray(json, jkDataBlockSchemaSlotSchema, schemaToJson, schema->pSchema, sizeof(SSlotSchema),
+                         schema->numOfCols);
   if (res) {
     res = cJSON_AddNumberToObject(json, jkDataBlockSchemaResultRowSize, schema->resultRowSize);
   }
@@ -279,7 +285,8 @@ static bool dataBlockSchemaFromJson(const cJSON* json, void* obj) {
   schema->resultRowSize = getNumber(json, jkDataBlockSchemaResultRowSize);
   schema->precision = getNumber(json, jkDataBlockSchemaPrecision);
 
-  return fromRawArrayWithAlloc(json, jkDataBlockSchemaSlotSchema, schemaFromJson, (void**)&(schema->pSchema), sizeof(SSlotSchema), &schema->numOfCols);
+  return fromRawArrayWithAlloc(json, jkDataBlockSchemaSlotSchema, schemaFromJson, (void**)&(schema->pSchema),
+                               sizeof(SSlotSchema), &schema->numOfCols);
 }
 
 static const char* jkColumnFilterInfoLowerRelOptr = "LowerRelOptr";
@@ -332,7 +339,8 @@ static bool columnInfoToJson(const void* obj, cJSON* jCol) {
   }
 
   if (res) {  // TODO: temporarily disable it
-//    res = addRawArray(jCol, jkColumnInfoFilterList, columnFilterInfoToJson, col->flist.filterInfo, sizeof(SColumnFilterInfo), col->flist.numOfFilters);
+              //    res = addRawArray(jCol, jkColumnInfoFilterList, columnFilterInfoToJson, col->flist.filterInfo,
+              //    sizeof(SColumnFilterInfo), col->flist.numOfFilters);
   }
 
   return res;
@@ -344,7 +352,8 @@ static bool columnInfoFromJson(const cJSON* json, void* obj) {
   col->type = getNumber(json, jkColumnInfoType);
   col->bytes = getNumber(json, jkColumnInfoBytes);
   int32_t size = 0;
-  bool res = fromRawArrayWithAlloc(json, jkColumnInfoFilterList, columnFilterInfoFromJson, (void**)&col->flist.filterInfo, sizeof(SColumnFilterInfo), &size);
+  bool    res = fromRawArrayWithAlloc(json, jkColumnInfoFilterList, columnFilterInfoFromJson,
+                                      (void**)&col->flist.filterInfo, sizeof(SColumnFilterInfo), &size);
   col->flist.numOfFilters = size;
   return res;
 }
@@ -408,7 +417,8 @@ static bool functionToJson(const void* obj, cJSON* jFunc) {
   const tExprNode* exprInfo = (const tExprNode*)obj;
   bool             res = cJSON_AddStringToObject(jFunc, jkFunctionName, exprInfo->_function.functionName);
   if (res && NULL != exprInfo->_function.pChild) {
-      res = addRawArray(jFunc, jkFunctionChild, exprNodeToJson, exprInfo->_function.pChild, sizeof(tExprNode*), exprInfo->_function.num);
+    res = addRawArray(jFunc, jkFunctionChild, exprNodeToJson, exprInfo->_function.pChild, sizeof(tExprNode*),
+                      exprInfo->_function.num);
   }
   return res;
 }
@@ -420,7 +430,8 @@ static bool functionFromJson(const cJSON* json, void* obj) {
   if (NULL == exprInfo->_function.pChild) {
     return false;
   }
-  return fromRawArrayWithAlloc(json, jkFunctionChild, exprNodeFromJson, (void**)exprInfo->_function.pChild, sizeof(tExprNode*), &exprInfo->_function.num);
+  return fromRawArrayWithAlloc(json, jkFunctionChild, exprNodeFromJson, (void**)exprInfo->_function.pChild,
+                               sizeof(tExprNode*), &exprInfo->_function.num);
 }
 
 static const char* jkVariantType = "Type";
@@ -435,7 +446,7 @@ static bool variantToJson(const void* obj, cJSON* jVar) {
     res = cJSON_AddNumberToObject(jVar, jkVariantLen, var->nLen);
   }
   if (res) {
-    if (0/* in */) {
+    if (0 /* in */) {
       res = addArray(jVar, jkVariantvalues, variantToJson, var->arr);
     } else if (IS_NUMERIC_TYPE(var->nType)) {
       res = cJSON_AddNumberToObject(jVar, jkVariantValue, var->d);
@@ -450,7 +461,7 @@ static bool variantFromJson(const cJSON* json, void* obj) {
   SVariant* var = (SVariant*)obj;
   var->nType = getNumber(json, jkVariantType);
   var->nLen = getNumber(json, jkVariantLen);
-  if (0/* in */) {
+  if (0 /* in */) {
     return fromArray(json, jkVariantvalues, variantFromJson, &var->arr, sizeof(SVariant));
   } else if (IS_NUMERIC_TYPE(var->nType)) {
     var->d = getNumber(json, jkVariantValue);
@@ -502,7 +513,8 @@ static bool exprNodeFromJson(const cJSON* json, void* obj) {
     case TEXPR_FUNCTION_NODE:
       return fromObject(json, jkExprNodeFunction, functionFromJson, exprInfo, false);
     case TEXPR_COL_NODE:
-      return fromObjectWithAlloc(json, jkExprNodeColumn, schemaFromJson, (void**)&exprInfo->pSchema, sizeof(SSchema), false);
+      return fromObjectWithAlloc(json, jkExprNodeColumn, schemaFromJson, (void**)&exprInfo->pSchema, sizeof(SSchema),
+                                 false);
     case TEXPR_VALUE_NODE:
       return fromObject(json, jkExprNodeValue, variantFromJson, exprInfo->pVal, false);
     default:
@@ -535,7 +547,8 @@ static bool sqlExprFromJson(const cJSON* json, void* obj) {
   SSqlExpr* expr = (SSqlExpr*)obj;
   bool      res = fromObject(json, jkSqlExprSchema, schemaFromJson, &expr->resSchema, false);
   if (res) {
-    res = fromRawArrayWithAlloc(json, jkSqlExprColumns, columnFromJson, (void**)&expr->pColumns, sizeof(SColumn), &expr->numOfCols);
+    res = fromRawArrayWithAlloc(json, jkSqlExprColumns, columnFromJson, (void**)&expr->pColumns, sizeof(SColumn),
+                                &expr->numOfCols);
   }
   if (res) {
     expr->interBytes = getNumber(json, jkSqlExprInterBytes);
@@ -564,7 +577,8 @@ static bool exprInfoFromJson(const cJSON* json, void* obj) {
   SExprInfo* exprInfo = (SExprInfo*)obj;
   bool       res = fromObject(json, jkExprInfoBase, sqlExprFromJson, &exprInfo->base, true);
   if (res) {
-    res = fromObjectWithAlloc(json, jkExprInfoExpr, exprNodeFromJson, (void**)&exprInfo->pExpr, sizeof(tExprNode), true);
+    res =
+        fromObjectWithAlloc(json, jkExprInfoExpr, exprNodeFromJson, (void**)&exprInfo->pExpr, sizeof(tExprNode), true);
   }
   return res;
 }
@@ -576,12 +590,12 @@ static bool timeWindowToJson(const void* obj, cJSON* json) {
   const STimeWindow* win = (const STimeWindow*)obj;
 
   char tmp[40] = {0};
-  snprintf(tmp, tListLen(tmp),"%"PRId64, win->skey);
+  snprintf(tmp, tListLen(tmp), "%" PRId64, win->skey);
 
   bool res = cJSON_AddStringToObject(json, jkTimeWindowStartKey, tmp);
   if (res) {
     memset(tmp, 0, tListLen(tmp));
-    snprintf(tmp, tListLen(tmp),"%"PRId64, win->ekey);
+    snprintf(tmp, tListLen(tmp), "%" PRId64, win->ekey);
     res = cJSON_AddStringToObject(json, jkTimeWindowEndKey, tmp);
   }
   return res;
@@ -604,7 +618,7 @@ static bool scanNodeToJson(const void* obj, cJSON* json) {
   const SScanPhyNode* pNode = (const SScanPhyNode*)obj;
 
   char uid[40] = {0};
-  snprintf(uid, tListLen(uid), "%"PRIu64, pNode->uid);
+  snprintf(uid, tListLen(uid), "%" PRIu64, pNode->uid);
   bool res = cJSON_AddStringToObject(json, jkScanNodeTableId, uid);
 
   if (res) {
@@ -758,7 +772,7 @@ static const char* jkNodeTaskId      = "TaskId";
 static const char* jkNodeTaskSchedId = "SchedId";
 
 static bool queryNodeAddrToJson(const void* obj, cJSON* json) {
-  const SQueryNodeAddr* pAddr = (const SQueryNodeAddr*) obj;
+  const SQueryNodeAddr* pAddr = (const SQueryNodeAddr*)obj;
   bool                  res = cJSON_AddNumberToObject(json, jkNodeAddrId, pAddr->nodeId);
 
   if (res) {
@@ -772,7 +786,7 @@ static bool queryNodeAddrToJson(const void* obj, cJSON* json) {
 }
 
 static bool queryNodeAddrFromJson(const cJSON* json, void* obj) {
-  SQueryNodeAddr* pAddr = (SQueryNodeAddr*) obj;
+  SQueryNodeAddr* pAddr = (SQueryNodeAddr*)obj;
 
   pAddr->nodeId = getNumber(json, jkNodeAddrId);
   pAddr->epSet.inUse = getNumber(json, jkNodeAddrInUse);
@@ -784,12 +798,12 @@ static bool queryNodeAddrFromJson(const cJSON* json, void* obj) {
 }
 
 static bool nodeAddrToJson(const void* obj, cJSON* json) {
-  const SDownstreamSource* pSource = (const SDownstreamSource*) obj;
+  const SDownstreamSource* pSource = (const SDownstreamSource*)obj;
   bool                     res = cJSON_AddNumberToObject(json, jkNodeTaskId, pSource->taskId);
 
   if (res) {
     char t[30] = {0};
-    snprintf(t, tListLen(t), "%"PRIu64, pSource->schedId);
+    snprintf(t, tListLen(t), "%" PRIu64, pSource->schedId);
     res = cJSON_AddStringToObject(json, jkNodeTaskSchedId, t);
   }
 
@@ -815,7 +829,8 @@ static bool exchangeNodeToJson(const void* obj, cJSON* json) {
   const SExchangePhyNode* exchange = (const SExchangePhyNode*)obj;
   bool                    res = cJSON_AddNumberToObject(json, jkExchangeNodeSrcTemplateId, exchange->srcTemplateId);
   if (res) {
-    res = addRawArray(json, jkExchangeNodeSrcEndPoints, nodeAddrToJson, exchange->pSrcEndPoints->pData, sizeof(SDownstreamSource), taosArrayGetSize(exchange->pSrcEndPoints));
+    res = addRawArray(json, jkExchangeNodeSrcEndPoints, nodeAddrToJson, exchange->pSrcEndPoints->pData,
+                      sizeof(SDownstreamSource), taosArrayGetSize(exchange->pSrcEndPoints));
   }
   return res;
 }
@@ -823,7 +838,8 @@ static bool exchangeNodeToJson(const void* obj, cJSON* json) {
 static bool exchangeNodeFromJson(const cJSON* json, void* obj) {
   SExchangePhyNode* exchange = (SExchangePhyNode*)obj;
   exchange->srcTemplateId = getNumber(json, jkExchangeNodeSrcTemplateId);
-  return fromInlineArray(json, jkExchangeNodeSrcEndPoints, nodeAddrFromJson, &exchange->pSrcEndPoints, sizeof(SDownstreamSource));
+  return fromInlineArray(json, jkExchangeNodeSrcEndPoints, nodeAddrFromJson, &exchange->pSrcEndPoints,
+                         sizeof(SDownstreamSource));
 }
 
 static bool specificPhyNodeToJson(const void* obj, cJSON* json) {
@@ -933,7 +949,7 @@ static bool phyNodeToJson(const void* obj, cJSON* jNode) {
 }
 
 static bool phyNodeFromJson(const cJSON* json, void* obj) {
-  SPhyNode* node = (SPhyNode*) obj;
+  SPhyNode* node = (SPhyNode*)obj;
 
   node->info.type = getNumber(json, jkPnodeType);
   node->info.name = opTypeToOpName(node->info.type);
@@ -1034,7 +1050,7 @@ static bool subplanIdToJson(const void* obj, cJSON* jId) {
   const SSubplanId* id = (const SSubplanId*)obj;
 
   char ids[40] = {0};
-  snprintf(ids, tListLen(ids), "%"PRIu64, id->queryId);
+  snprintf(ids, tListLen(ids), "%" PRIu64, id->queryId);
 
   bool res = cJSON_AddStringToObject(jId, jkIdQueryId, ids);
   if (res) {
@@ -1094,7 +1110,8 @@ static SSubplan* subplanFromJson(const cJSON* json) {
   }
 
   if (res) {
-    res = fromObjectWithAlloc(json, jkSubplanDataSink, dataSinkFromJson, (void**)&subplan->pDataSink, sizeof(SDataSink), false);
+    res = fromObjectWithAlloc(json, jkSubplanDataSink, dataSinkFromJson, (void**)&subplan->pDataSink, sizeof(SDataSink),
+                              false);
   }
 
   if (!res) {
@@ -1137,15 +1154,15 @@ int32_t stringToSubplan(const char* str, SSubplan** subplan) {
 
 cJSON* qDagToJson(const SQueryDag* pDag) {
   cJSON* pRoot = cJSON_CreateObject();
-  if(pRoot == NULL) {
+  if (pRoot == NULL) {
     return NULL;
   }
 
   cJSON_AddNumberToObject(pRoot, "Number", pDag->numOfSubplans);
   cJSON_AddNumberToObject(pRoot, "QueryId", pDag->queryId);
 
-  cJSON *pLevels = cJSON_CreateArray();
-  if(pLevels == NULL) {
+  cJSON* pLevels = cJSON_CreateArray();
+  if (pLevels == NULL) {
     cJSON_Delete(pRoot);
     return NULL;
   }
@@ -1153,19 +1170,19 @@ cJSON* qDagToJson(const SQueryDag* pDag) {
   cJSON_AddItemToObject(pRoot, "Subplans", pLevels);
 
   size_t level = taosArrayGetSize(pDag->pSubplans);
-  for(size_t i = 0; i < level; i++) {
+  for (size_t i = 0; i < level; i++) {
     const SArray* pSubplans = (const SArray*)taosArrayGetP(pDag->pSubplans, i);
     size_t        num = taosArrayGetSize(pSubplans);
     cJSON*        plansOneLevel = cJSON_CreateArray();
-    if(plansOneLevel == NULL) {
+    if (plansOneLevel == NULL) {
       cJSON_Delete(pRoot);
       return NULL;
     }
 
     cJSON_AddItemToArray(pLevels, plansOneLevel);
-    for(size_t j = 0; j < num; j++) {
+    for (size_t j = 0; j < num; j++) {
       cJSON* pSubplan = subplanToJson((const SSubplan*)taosArrayGetP(pSubplans, j));
-      if(pSubplan == NULL) {
+      if (pSubplan == NULL) {
         cJSON_Delete(pRoot);
         return NULL;
       }
@@ -1183,22 +1200,22 @@ char* qDagToString(const SQueryDag* pDag) {
 
 SQueryDag* qJsonToDag(const cJSON* pRoot) {
   SQueryDag* pDag = malloc(sizeof(SQueryDag));
-  if(pDag == NULL) {
+  if (pDag == NULL) {
     return NULL;
   }
   pDag->numOfSubplans = cJSON_GetNumberValue(cJSON_GetObjectItem(pRoot, "Number"));
   pDag->queryId = cJSON_GetNumberValue(cJSON_GetObjectItem(pRoot, "QueryId"));
-  pDag->pSubplans = taosArrayInit(0, sizeof(SArray));
+  pDag->pSubplans = taosArrayInit(0, sizeof(void*));
   if (pDag->pSubplans == NULL) {
     free(pDag);
     return NULL;
   }
   cJSON* pLevels = cJSON_GetObjectItem(pRoot, "Subplans");
   int    level = cJSON_GetArraySize(pLevels);
-  for(int i = 0; i < level; i++) {
+  for (int i = 0; i < level; i++) {
     SArray* plansOneLevel = taosArrayInit(0, sizeof(void*));
-    if(plansOneLevel == NULL) {
-      for(int j = 0; j < i; j++) {
+    if (plansOneLevel == NULL) {
+      for (int j = 0; j < i; j++) {
         taosArrayDestroy(taosArrayGetP(pDag->pSubplans, j));
       }
       taosArrayDestroy(pDag->pSubplans);
@@ -1207,12 +1224,12 @@ SQueryDag* qJsonToDag(const cJSON* pRoot) {
     }
     cJSON* pItem = cJSON_GetArrayItem(pLevels, i);
     int    sz = cJSON_GetArraySize(pItem);
-    for(int j = 0; j < sz; j++) {
+    for (int j = 0; j < sz; j++) {
       cJSON*    pSubplanJson = cJSON_GetArrayItem(pItem, j);
       SSubplan* pSubplan = subplanFromJson(pSubplanJson);
       taosArrayPush(plansOneLevel, &pSubplan);
     }
-    taosArrayPush(pDag->pSubplans, plansOneLevel);
+    taosArrayPush(pDag->pSubplans, &plansOneLevel);
   }
   return pDag;
 }